openvibers/openvibe.media Commits
Never delete timeline bytes another object still names; re-key every naming row when packing (#35)
Changes
@@@ -136,6 +136,20 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}chunk's delete fails). All of it behind `MEDIA_HLS_ENABLED`; no schema change (F3.1's columns carry it). A finished`object.cmaf` queues it by itself (`objects/timeline-queue.js`; `dedupeActive`, so a rerun while a pack is active joinsit and a later cut queues a new one), so packing follows the cut without an operator.++- **F3.4, shipped: shared locations and reference counts (the foundation of materialized clips, §4).** A location names++ bytes, not an owner: any number of `media_timeline` rows — of any number of objects (a source and the clips over it) —++ may name one location, because a location is content-addressed and names the same bytes. **A location's bytes are++ deleted exactly when no row names it.** The reference count is a query over the rows (`timeline.namedElsewhere`: the++ rows of another object naming a durable `(durable_provider, key)` or a `local_path`), never a counter table, so a count++ cannot drift from the rows; migration `0003_media_timeline_locations.sql` indexes both so the query is an index++ lookup. `timeline.removeObject(objectId)`/`deleteBytes` keep a location another object's rows still name: deleting a++ clip never deletes its source's bytes, deleting a source a clip still names keeps them (its rows go; the bytes are++ deleted when the last naming object is removed), and only a delete that fails keeps rows for a retry, as before.++ `object.pack` re-keys **every** row that names a packed segment's old location with the same sha256 — not only the++ source's — to the chunk in the same transaction, so a clip follows its source into the chunk and the per-segment++ location is then freed only once nothing names it; a row naming the old location with a different sha256 (which should++ not happen) is left alone and keeps the location named. `object.cmaf` deletes an older cut's locations under the same++ rule. No schema besides the two indexes; additive, an older release never reads it.- **F3.c, shipped: the timeline queued by itself, signed playlists, the MP4 fallback.** A recording's finalize queues`object.cmaf` for its object once it is ready (`server/objects/timeline-queue.js`; `dedupeActive` and the idempotencykey `object.cmaf:<id>`, so a second finalize joins the same job; a queue error is logged, never fails the finalize),@@@ -157,9 +171,9 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}and metadata are the seek path's, so the player contract is unchanged. A source without a timeline, or with`MEDIA_HLS_ENABLED` off, keeps the one-fast-seek-per-frame path; captions stay with AI's `media.analyze`.**Still open in F3:** the growing live/DVR playlist written as OpenRe segments (F3.2), write-behind durability for-- live segments and its upload-lag metric, a materialized clip that references the source's chunks (chunk reference-- counts, boundary-only re-encode), captions on the timeline, segment-bucket demand, and Live's player-- moving to HLS. Renditions are F4.++ live segments and its upload-lag metric, a materialized clip that references the source's chunks (reference counts++ shipped, F3.4; its own persisted rows and boundary-only re-encode are open), captions on the timeline,++ segment-bucket demand, and Live's player moving to HLS. Renditions are F4.## 4. Clips reuse the source@@@ -190,8 +204,13 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}through `clip_of`.- **Materialized clip** (shared externally, downloaded, edited, popular): middle GOPs are **referenced**, not copied(the new manifest points at the source's packed chunks); only the two boundary segments are re-encoded for-- frame-accurate cuts. Segments and chunks are content-addressed (sha256) with reference counts, so deleting a VOD-- that a clip still references keeps the referenced chunks (and holds still freeze everything).++ frame-accurate cuts. Segments and chunks are content-addressed (sha256), so the reference count of a location is just++ the rows that name it (F3.4, shipped: `timeline.namedElsewhere`, no counter table; migration 0003 indexes it) and++ deleting a VOD that a clip still references keeps the referenced chunks — they go when the last naming object is++ removed, and holds still freeze everything. The clip's own persisted rows over the source's locations are the next++ step (§3, F3.4 carries the deletion rule). Inserting a clip's rows over a source's locations must check, in the same++ transaction, that a source row still names each location (and take a per-location lock with the delete check), or a++ concurrent delete can drop bytes the new clip names.## 5. Derivatives are cost-managed computed cachesnnew file mode 100644@@@ -0,0 +1,10 @@++-- phase: expand++-- Reference counts for shared segment locations (docs/media-fabric.md §3/§4; the foundation of materialized clips). A++-- row names a location by the durable pair (durable_provider, key) and/or its local copy (local_path); a location's++-- bytes are shared whenever several objects' rows name it (a source and the clips over it). "How many objects name this++-- location?" (timeline.namedElsewhere) is answered by a query over media_timeline — there is no counter table, so a count++-- can never drift from the rows — and these two indexes make those reference queries index lookups.++-- Additive only: an older release never reads it (it just keeps deleting its own locations, as before). Never edited++-- after it runs; a further change is a new migration.++CREATE INDEX IF NOT EXISTS media_timeline_location_durable ON media_timeline (durable_provider, key);++CREATE INDEX IF NOT EXISTS media_timeline_location_local ON media_timeline (local_path);@@@ -186,8 +186,12 @@ async function run(job, ctx) {// A deletion that fails leaves an orphan the storage report names, never bytes a live row still points at.const named = new Set(rows.map((r) => `${r.key}\n${r.local_path || ''}`));const stale = [...prev.values()].filter((old) => !named.has(`${old.key}\n${old.local_path || ''}`));-- const { failed: stuck } = await timeline.deleteBytes(stale);++ // A location any row still names — another object's, or the source's own (another rendition/seq, or a++ // different local_path with the same key) — is kept: the old cut's rows go, the shared bytes stay, and the++ // last naming object's removal deletes them. `isNamed` counts every object, no exception.++ const { failed: stuck, kept } = await timeline.deleteBytes(stale);if (stuck.length) console.warn(`[Cmaf] ${job.id}: ${stuck.length} old segment(s) left behind (the storage orphan report names them), first ${stuck[0].provider}:${stuck[0].key}`);++ if (kept.length) console.warn(`[Cmaf] ${job.id}: ${kept.length} old segment location(s) kept — another object's rows name them`);if (failed.length) {throw new JobError('upload_failed', `${failed.length} segment(s) not durable yet (${failed[0].slice(0, 200)}); a retry uploads only those`, { retryAfterS: 600 });}@@@ -170,7 +170,7 @@ async function run(job, ctx) {if (!back || back.sha256 !== sha) throw new JobError('pack_unverified', `${provider}:${key} does not read back as ${sha.slice(0, 12)}`, { retryAfterS: 600 });if (localPath) { fs.mkdirSync(path.dirname(localPath), { recursive: true }); fs.renameSync(tmp, localPath); }aborted(ctx);-- committed = await timeline.markPacked(src.id, rendition, placed.map((p) => ({++ committed = await timeline.markPacked(src.id, placed.map((p) => ({row: p.row, key, local_path: localPath, durable_provider: provider, packed_object_id: sha, byte_offset: p.byte_offset,})), { jobId: job.id });if (!committed) throw new JobError('media.timeline.changed', 'The timeline changed while packing (a re-cut); a retry packs the new segments', { retryAfterS: 60 });@@@ -182,9 +182,12 @@ async function run(job, ctx) {try { fs.unlinkSync(tmp); } catch { /* moved or gone */ }}}-- // Phase two, after the commit: the per-segment copies the chunk now replaces.-- const { failed } = await timeline.deleteBytes(chunk);++ // Phase two, after the commit: the per-segment copies the chunk now replaces. A location any row still++ // names — another object's, or the source's own if it somehow did not follow the sha — is kept, not++ // deleted: `isNamed` counts every object, no exception.++ const { failed, kept } = await timeline.deleteBytes(chunk);if (failed.length) console.warn(`[Pack] ${job.id}: ${failed.length} packed segment key(s) left behind (the storage orphan report names them), first ${failed[0].provider}:${failed[0].key}`);++ if (kept.length) console.warn(`[Pack] ${job.id}: ${kept.length} packed segment location(s) kept — another object's rows name them`);out.left_behind += failed.length;out.removed_segments += chunk.length - failed.length;out.packs++;@@@ -16,6 +16,15 @@* row's key/local_path/durable_provider then name the chunk, packed_object_id its sha256 (content address) and* byte_offset/byte_length the segment's bytes inside it; its name, times and sha256 stay the segment's. Several rows* share one chunk, so the bytes are deleted once per location and kept (for a retry) while any delete of them fails.++ *++ * Locations are shared (§3/§4): any number of rows — of any number of objects (a source and the clips over it, F3.5/F4) —++ * may name one location, because a location is content-addressed and names bytes, not an owner. A location's bytes are++ * deleted only when no row names it. "How many objects name this location?" is a query over these rows (namedElsewhere),++ * never a counter table, so a count cannot drift from them; migration 0003 indexes (durable_provider, key) and++ * (local_path) for it. removeObject and deleteBytes therefore keep a location another object's rows still name, and++ * object.pack re-keys every row that names a packed segment's old location (same sha256) to the chunk in the same++ * transaction, so a clip follows its source into the chunk and the per-segment location is freed only once nothing names++ * it. A source's own purge deletes its rows; the shared bytes go when the last naming object does.*/'use strict';@@@ -109,62 +118,98 @@ async function replace(objectId, rendition, rows, { jobId = null, expect = null/*** Delete the bytes a set of rows names: this node's copy (only inside the timeline root) and the durable copy. A remote-- * delete that fails is returned, never swallowed: the caller keeps those rows, and their keys, so a later pass retries.-- * → { failed: [{ row, provider, key, error }] }.++ * delete that fails is returned, never swallowed: the caller keeps (or restores) those rows, and their keys, so a later++ * pass retries. A location any row still names (`isNamed`, across EVERY object, the caller's own included) is kept,++ * never deleted: it is shared (a source and the clips over it). Callers therefore delete the rows that named a location++ * first, in one transaction, so the check sees the other objects' rows: concurrent removals of objects sharing a++ * location converge on the last committer deleting it, and a double delete is idempotent. → { failed: [{ row, provider,++ * key, error }], kept: [{ row, provider, key }] } (kept = intentionally not deleted; for a local location provider is 'local').*/async function deleteBytes(rows) {const root = localRoot() + path.sep;const vodStorage = require('../vod/vod-storage');const dirs = new Set();const failed = [];-- const done = new Set(); // a packed chunk is named by every row it holds: each location is deleted once++ const kept = [];++ const done = new Set(); // a packed chunk is named by every row it holds: each location is handled oncefor (const r of rows) {if (r.local_path && path.resolve(r.local_path).startsWith(root) && !done.has(`local\n${r.local_path}`)) {done.add(`local\n${r.local_path}`);-- try { fs.unlinkSync(r.local_path); } catch { /* already gone */ }-- dirs.add(path.dirname(r.local_path));++ if (await isNamed({ localPath: r.local_path })) kept.push({ row: r, provider: 'local', key: r.local_path });++ else { try { fs.unlinkSync(r.local_path); } catch { /* already gone */ } dirs.add(path.dirname(r.local_path)); }}if (r.durable_provider && vodStorage.providerConfigured(r.durable_provider) && !done.has(`${r.durable_provider}\n${r.key}`)) {done.add(`${r.durable_provider}\n${r.key}`);-- const ok = await vodStorage.deleteObject(r.durable_provider, r.key);-- if (!ok) failed.push({ row: r, provider: r.durable_provider, key: r.key, error: 'delete failed' });++ if (await isNamed({ provider: r.durable_provider, key: r.key })) kept.push({ row: r, provider: r.durable_provider, key: r.key });++ else {++ const ok = await vodStorage.deleteObject(r.durable_provider, r.key);++ if (!ok) failed.push({ row: r, provider: r.durable_provider, key: r.key, error: 'delete failed' });++ }}}for (const d of dirs) {for (const p of [d, path.dirname(d)]) { try { fs.rmdirSync(p); } catch { /* not empty or gone */ } }}-- return { failed };++ return { failed, kept };}/**-- * Whether any timeline row still names this location (a durable `provider` + `key`, or a local `localPath`). Keys and-- * paths are content-addressed, so a run that lost a race may have built exactly the bytes a winner committed: it deletes-- * what it staged only when no row names it.++ * Whether any row of ANOTHER object names this location (a durable `provider` + `key`, or a local `localPath`), i.e.++ * whether its bytes are shared with an object other than `exceptObjectId`. That is the reference count's only question:++ * a location's bytes may be deleted exactly when this is false. Keys and paths are content-addressed, so two objects'++ * rows can name one location (a source and the clips over it). `exceptObjectId`, when given, is an object whose own rows++ * to ignore (a caller that has not deleted them yet); null counts every object (isNamed).*/--async function isNamed({ provider = null, key = null, localPath = null }) {-- if (provider && key && await db.get('SELECT 1 AS x FROM media_timeline WHERE durable_provider = ? AND key = ? LIMIT 1', [provider, key])) return true;-- if (localPath && await db.get('SELECT 1 AS x FROM media_timeline WHERE local_path = ? LIMIT 1', [localPath])) return true;++async function namedElsewhere({ provider = null, key = null, localPath = null }, exceptObjectId = null) {++ const skip = exceptObjectId ? ' AND object_id <> ?' : '';++ const rest = exceptObjectId ? [exceptObjectId] : [];++ if (provider && key && await db.get(`SELECT 1 AS x FROM media_timeline WHERE durable_provider = ? AND key = ?${skip} LIMIT 1`, [provider, key, ...rest])) return true;++ if (localPath && await db.get(`SELECT 1 AS x FROM media_timeline WHERE local_path = ?${skip} LIMIT 1`, [localPath, ...rest])) return true;return false;}/**-- * The object's whole timeline is gone with its bytes (a purge, or a vod/clip deleted for good): local files, then the-- * durable copies. A row whose durable delete failed is kept — with its key — so a later pass can retry; dropping it-- * would lose the only record of the bytes still in B2/R2. Callers have already refused a held object.-- * → { removed, pending } (pending = rows kept for a retry).++ * Whether any timeline row still names this location. Keys and paths are content-addressed, so a run that lost a race++ * may have built exactly the bytes a winner committed: it deletes what it staged only when no row names it.++ */++async function isNamed(loc) { return await namedElsewhere(loc, null); }++++/** Put a row back exactly as it was read, so a later pass retries a delete that failed (a packed chunk: every row). */++async function reinsertRow(r) {++ await db.run(`INSERT INTO media_timeline (object_id, rendition, seq, ${FIELDS.join(', ')}, job_id)++ VALUES (?, ?, ?, ${FIELDS.map(() => '?').join(', ')}, ?)++ ON CONFLICT (object_id, rendition, seq) DO UPDATE SET ${FIELDS.map((f) => `${f} = excluded.${f}`).join(', ')},++ job_id = excluded.job_id, updated_at = ov_now()`,++ [r.object_id, r.rendition, r.seq, ...FIELDS.map((f) => r[f] ?? null), r.job_id ?? null]);++}++++/**++ * The object's whole timeline is gone with its bytes (a purge, or a vod/clip deleted for good): row, then local files,++ * then the durable copies. The rows go first, in one transaction, and the bytes are deleted only once `isNamed` — a++ * check across ALL objects — sees no row naming a location; a location another object's rows still name is kept (its++ * rows are already gone, the shared bytes stay, and they are deleted when the last naming object is removed). That is++ * what makes concurrent removals converge: each remover deletes its rows before checking, so the last one to commit++ * sees no naming row and deletes, and the loser's double delete is idempotent. A durable delete that failed gets its++ * rows back — with their keys — so a later pass retries; dropping them would lose the only record of the bytes still in++ * B2/R2. Callers have already refused a held object. → { removed, pending } (pending = rows restored for a retry).*/async function removeObject(objectId) {if (!objectId) return { removed: 0, pending: 0 };const rows = await db.all('SELECT * FROM media_timeline WHERE object_id = ?', [objectId]);if (!rows.length) return { removed: 0, pending: 0 };++ // (a) The rows go first, in one transaction, before their bytes; the delete is not conditioned on what another++ // object still names, so two concurrent removals of a shared location cannot each keep it and both drop their rows.++ let removed = 0;++ await db.getDb().tx(async () => {++ removed = (await db.run('DELETE FROM media_timeline WHERE object_id = ?', [objectId])).changes || 0;++ });++ // (b) After the commit, each location the rows named, deleted only when no row of ANY object names it.const { failed } = await deleteBytes(rows);-- // Every row naming a location whose delete failed is kept: for a packed chunk that is all the rows it holds.++ // (c) Every row naming a location whose delete failed is restored: for a packed chunk that is all the rows it holds.const stuck = new Set(failed.map((f) => `${f.provider}\n${f.key}`));-- let removed = 0;let pending = 0;for (const r of rows) {-- if (r.durable_provider && stuck.has(`${r.durable_provider}\n${r.key}`)) { pending++; continue; }-- removed += (await db.run('DELETE FROM media_timeline WHERE object_id = ? AND rendition = ? AND seq = ?', [objectId, r.rendition, r.seq])).changes || 0;++ if (r.durable_provider && stuck.has(`${r.durable_provider}\n${r.key}`)) { await reinsertRow(r); pending++; }}if (pending) console.warn(`[Timeline] ${objectId}: ${pending} segment row(s) kept — their durable copy could not be deleted`);return { removed, pending };@@@ -172,21 +217,38 @@ async function removeObject(objectId) {/*** Point rows at the chunk they were packed into, in one transaction. `updates` = [{ row, key, local_path,-- * durable_provider, packed_object_id, byte_offset }]; each row changes only if it is still exactly the unpacked segment-- * the packer read (same key and sha256, not packed): a re-cut in between makes the whole commit roll back and false is-- * returned, so the caller drops its chunk. Never deletes a row.++ * durable_provider, packed_object_id, byte_offset }]; the location is the segment's old key (content-addressed, versioned++ * by its sha), and EVERY row of any object that names it with the same sha256 — the source's, and a clip's over it —++ * moves to the chunk in this same transaction, so a clip follows its source into the chunk. Each group changes only if++ * it still names the unpacked segment the packer read (same key and sha256, not packed) AND at least one changed row is++ * `objectId`'s own: a re-cut in between makes the whole commit roll back and false is returned, so the caller drops its++ * chunk. Without that second condition a re-cut source row (new key/sha) would leave only a clip's same-sha row matching++ * and the pack would commit while the source is not in the chunk. A row naming the old location with a different sha256++ * (which should not happen) does not match, is left alone, and so keeps the old location named (its bytes are not++ * deleted). Never deletes a row.*/--async function markPacked(objectId, rendition, updates, { jobId = null } = {}) {++async function markPacked(objectId, updates, { jobId = null } = {}) {const CONFLICT = new Error('timeline changed');try {await db.getDb().tx(async () => {await lockObject(objectId);for (const u of updates) {++ // The old location: the durable pair when the row was durable (packed rows always are), the local copy++ // too. The (durable_provider, key) index serves the first, (local_path) the second.++ const loc = u.row.durable_provider ? '(durable_provider = ? AND key = ?)' : 'key = ?';++ const at = u.row.durable_provider ? [u.row.durable_provider, u.row.key] : [u.row.key];++ const names = u.row.local_path ? `(${loc} OR local_path = ?)` : loc;const r = await db.run(`UPDATE media_timeline SET key = ?, local_path = ?, durable_provider = ?, durability = 'durable',-- packed_object_id = ?, byte_offset = ?, job_id = ?, updated_at = ov_now()-- WHERE object_id = ? AND rendition = ? AND seq = ? AND key = ? AND sha256 = ? AND packed_object_id IS NULL`,-- [u.key, u.local_path, u.durable_provider, u.packed_object_id, u.byte_offset, jobId, objectId, rendition, u.row.seq, u.row.key, u.row.sha256]);-- if (!r.changes) throw CONFLICT;++ packed_object_id = ?, byte_offset = ?, byte_length = ?, job_id = ?, updated_at = ov_now()++ WHERE sha256 = ? AND packed_object_id IS NULL AND ${names}++ RETURNING object_id`,++ [u.key, u.local_path, u.durable_provider, u.packed_object_id, u.byte_offset, u.row.byte_length, jobId,++ u.row.sha256, ...at, ...(u.row.local_path ? [u.row.local_path] : [])]);++ // The pack changes `objectId`'s OWN timeline: at least one re-keyed row must be its. A re-cut between++ // the read and this commit leaves the source's row with a new key/sha; the same-sha match is then only++ // some other object's row (a clip's), and committing would leave the source out of the chunk it names.++ // Treat that as a conflict — nothing is written and the caller drops its chunk and retries.++ if (!r.rows.some((row) => String(row.object_id) === String(objectId))) throw CONFLICT;}});return true;@@@ -275,6 +337,6 @@ function clipRows(rows, startMs, endMs) {module.exports = {SOURCE, INIT_NAME, FIELDS, localRoot, localPathFor, keyFor, segmentName,-- list, segments, byName, has, segmentAt, replace, TimelineChanged, isNamed, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,++ list, segments, byName, has, segmentAt, replace, TimelineChanged, isNamed, namedElsewhere, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,mediaPlaylist, masterPlaylist, clipRows,};@@@ -165,6 +165,10 @@ const { spawn, spawnSync } = require('child_process');const sto = await model.getObject(st.result.object_id);assert.strictEqual(model.parseJson(sto.metadata, {}).source_duration_seconds, 7, 'the seek metadata carries the real duration');assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');++ // shorty's rows name pub's segment files (two objects, one location): dropping them keeps the bytes, which pub still names,++ // and lets the purge below delete them, since after that no row of any object names them++ await timeline.removeObject(shorty);++ assert.ok(rows.every((r) => fs.existsSync(r.local_path)), 'pub still names its files after a sharing object is removed');// ── The finished cut queues object.pack by itself, once (the worker runs it; no provider yet, so it packs nothing) ──const packJobs = async (objectId) => (await queue.list('live', { type: 'object.pack', objectId })).jobs;nnew file mode 100644@@@ -0,0 +1,234 @@++'use strict';++// Shared segment locations and their reference counts (F3.4; docs/media-fabric.md §3/§4): any number of media_timeline++// rows — of any number of objects (a source and the clips over it) — may name one location, and a location's bytes are++// deleted only when no row names it. A clip object's rows name its source's locations; removing the clip leaves the++// source's files and durable keys, removing the source keeps the locations the clip still names, and in either order the++// bytes go when the last naming object does. object.pack re-keys every row that names a packed segment's old location++// (same sha256) to the chunk in one transaction, so a clip follows its source into the chunk and its playlist serves the++// chunk ranges; the old location is then freed. A row whose sha256 differs is left alone and keeps the location named.++// Migration 0003 adds the two location indexes and applies on a database that already has rows.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const crypto = require('crypto');++(async () => {++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-timeline-refs-'));++ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++ process.env.OBJECTS_PATH = dir('objects');++ process.env.VOD_PATH = dir('vods');++ process.env.CLIPS_PATH = dir('clips');++ process.env.MEDIA_HLS_ENABLED = '1';++ for (const p of ['B2', 'R2']) for (const k of ['ENDPOINT', 'BUCKET', 'KEY_ID', 'APP_KEY', 'ACCESS_KEY_ID', 'SECRET_ACCESS_KEY']) process.env[`MEDIA_${p}_${k}`] = '';++++ (async () => {++ const config = require('../server/config');++ const db = require('../server/db/database');++ const timeline = require('../server/objects/timeline');++ const vodStorage = require('../server/vod/vod-storage');++ const sha = (buf) => crypto.createHash('sha256').update(buf).digest('hex');++++ // The durable providers, stubbed: bytes in a Map, uploads kept, and deletes recorded only when a key actually++ // went (a delete of bytes already gone is a success, as S3's is — so a concurrent double delete records once).++ const blobs = new Map();++ const deletes = [];++ const failDeletes = new Set(); // keys whose durable delete fails once (returns false), for the retry case++ Object.assign(vodStorage, {++ providerConfigured: (p) => ['b2', 'r2'].includes(p),++ providerAvailable: (p) => ['b2', 'r2'].includes(p),++ uploadFile: async (p, key, file) => { blobs.set(`${p}:${key}`, fs.readFileSync(file)); },++ sha256Object: async (p, key) => (blobs.has(`${p}:${key}`) ? { sha256: sha(blobs.get(`${p}:${key}`)), size: blobs.get(`${p}:${key}`).length } : null),++ deleteObject: async (p, key) => {++ if (failDeletes.has(`${p}:${key}`)) return false;++ if (blobs.delete(`${p}:${key}`)) deletes.push(`${p}:${key}`);++ return true;++ },++ presignGet: async () => null,++ });++++ const insert = async (objectId, r) => db.run(++ `INSERT INTO media_timeline (object_id, rendition, seq, name, start_ms, end_ms, keyframe_ms, key, local_path,++ durable_provider, packed_object_id, byte_offset, byte_length, sha256, durability, job_id)++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)`,++ [objectId, 'source', r.seq, r.name, r.start_ms, r.end_ms, r.keyframe_ms ?? null, r.key, r.local_path,++ r.durable_provider, r.packed_object_id ?? null, r.byte_offset ?? null, r.byte_length, r.sha256, r.durability]);++++ /** One object's rows (init + three media segments), with this node's file and the durable copy really written. */++ const seedSource = async (id) => {++ const obj = { app_id: 'live', id };++ const payloads = [Buffer.from('init-bytes'), Buffer.from('segment-one-bytes'), Buffer.from('segment-two-bytes'), Buffer.from('segment-three-bytes')];++ const rows = [];++ for (let seq = 0; seq < payloads.length; seq++) {++ const bytes = payloads[seq];++ const s = sha(bytes);++ const name = timeline.segmentName(seq);++ const key = timeline.keyFor(id, 'source', name, s.slice(0, 12));++ const local = timeline.localPathFor(obj, 'source', name, s.slice(0, 12));++ fs.mkdirSync(path.dirname(local), { recursive: true });++ fs.writeFileSync(local, bytes);++ blobs.set(`b2:${key}`, bytes);++ const r = { seq, name, start_ms: seq ? (seq - 1) * 3000 : 0, end_ms: seq * 3000, keyframe_ms: seq ? (seq - 1) * 3000 : null,++ key, local_path: local, durable_provider: 'b2', byte_length: bytes.length, sha256: s, durability: 'durable' };++ await insert(id, r);++ rows.push(r);++ }++ return rows;++ };++++ /** A clip's row naming a source row's location: the location (key, local_path, durable provider, sha, byte++ * length) is the source's, the seq and times are the clip's, exactly as timeline.clipRows leaves them. */++ const clipRow = (src, seq, { sha256 = src.sha256 } = {}) => ({ seq, name: src.name, start_ms: src.start_ms, end_ms: src.end_ms,++ keyframe_ms: src.keyframe_ms, key: src.key, local_path: src.local_path, durable_provider: src.durable_provider,++ byte_length: src.byte_length, sha256, durability: src.durability });++++ // ── The migration: the two location indexes exist, and it applies again on a database that already has rows ──++ await seedSource('idx-A');++ const indexes = (await db.all(`SELECT indexname FROM pg_indexes WHERE schemaname = current_schema() AND tablename = 'media_timeline'`)).map((r) => r.indexname);++ assert.ok(indexes.includes('media_timeline_location_durable') && indexes.includes('media_timeline_location_local'), `migration 0003 applied: ${indexes.join(', ')}`);++ const migration = fs.readFileSync(path.join(__dirname, '..', 'migrations', '0003_media_timeline_locations.sql'), 'utf8');++ // CREATE INDEX IF NOT EXISTS: a no-op on the row-filled database, never an error. Only on PGlite: under test:pg the test++ // runs as the service role, and PostgreSQL checks table ownership before IF NOT EXISTS (the migration runner applies++ // migrations as the owner).++ if (process.env.MEDIA_TEST_STORE !== 'pg') await db.getDb().query(migration);++ await timeline.removeObject('idx-A');++ deletes.length = 0;++++ // ── namedElsewhere: another object's rows, and only another object's ──++ const A = await seedSource('ref-A');++ const B = A.slice(1).map((r, i) => clipRow(r, i + 1));++ for (const r of B) await insert('ref-B', r);++ const loc = { provider: 'b2', key: A[1].key };++ assert.strictEqual(await timeline.namedElsewhere(loc, 'ref-A'), true, 'the clip names the source location');++ assert.strictEqual(await timeline.namedElsewhere(loc, 'ref-B'), true, 'the source names it');++ assert.strictEqual(await timeline.namedElsewhere({ localPath: A[1].local_path }, 'ref-A'), true, 'the local location too');++ assert.strictEqual(await timeline.namedElsewhere({ provider: 'b2', key: 'nobody' }, 'ref-A'), false, 'an unnamed key is not');++ assert.strictEqual(await timeline.namedElsewhere(loc, null), true, 'isNamed counts every object');++++ // ── Case 1: removeObject(clip) keeps the source's bytes; removeObject(source) then deletes them ──++ assert.deepStrictEqual(await timeline.removeObject('ref-B'), { removed: B.length, pending: 0 });++ assert.strictEqual((await timeline.list('ref-B')).length, 0, 'the clip\'s rows are gone');++ assert.ok(A.every((r) => fs.existsSync(r.local_path)), 'the source\'s files stay');++ assert.ok(A.every((r) => blobs.has(`b2:${r.key}`)), 'the source\'s durable keys stay');++ assert.deepStrictEqual(deletes, [], 'nothing was deleted while the source still names them');++ assert.deepStrictEqual(await timeline.removeObject('ref-A'), { removed: A.length, pending: 0 });++ assert.ok(A.every((r) => !fs.existsSync(r.local_path)), 'the source\'s files go with the last naming object');++ assert.ok(A.every((r) => !blobs.has(`b2:${r.key}`)), 'and its durable keys');++++ // ── Case 2: removeObject(source) first keeps the locations the clip still names; removeObject(clip) deletes them ──++ const A2 = await seedSource('ref-A2');++ for (const [i, r] of A2.slice(1).entries()) await insert('ref-B2', clipRow(r, i + 1));++ assert.deepStrictEqual(await timeline.removeObject('ref-A2'), { removed: A2.length, pending: 0 });++ assert.ok(A2.slice(1).every((r) => fs.existsSync(r.local_path)) && A2.slice(1).every((r) => blobs.has(`b2:${r.key}`)), 'the source is gone, the segments the clip names are kept');++ assert.ok(!fs.existsSync(A2[0].local_path) && !blobs.has(`b2:${A2[0].key}`), 'the init segment, named by no clip row, went with the source');++ assert.ok(deletes.includes(`b2:${A2[0].key}`) && !deletes.includes(`b2:${A2[1].key}`), 'only the unshared location was deleted');++ assert.strictEqual((await timeline.list('ref-B2')).length, A2.length - 1, 'the clip keeps its rows');++ assert.deepStrictEqual(await timeline.removeObject('ref-B2'), { removed: A2.length - 1, pending: 0 });++ assert.ok(A2.slice(1).every((r) => !fs.existsSync(r.local_path)) && A2.slice(1).every((r) => !blobs.has(`b2:${r.key}`)), 'the last naming object deletes them');++++ // ── Pack: a clip follows its source into the chunk, in the same transaction; the old locations are freed ──++ const derive = require('../server/jobs/derive');++ const pack = require('../server/jobs/pack');++ derive.loadSource = async (job) => ({ id: job.object_id, app_id: 'live', kind: 'video', mime_type: 'video/mp4', lifecycle_status: 'ready' });++ timeline.durableProvider = async () => 'b2';++ const packA = await seedSource('pack-A');++ const packB = packA.slice(1).map((r, i) => clipRow(r, i + 1));++ for (const r of packB) await insert('pack-B', r);++ const job = { id: 'mjob_REFS', app_id: 'live', object_id: 'pack-A', params: { target_seconds: 60 } };++ const out = await pack.spec.run(job, { signal: new AbortController().signal });++ assert.deepStrictEqual([out.packs, out.segments, out.left_behind], [1, 3, 0], JSON.stringify(out));++ const packedA = await timeline.list('pack-A');++ const chunkId = packedA[1].packed_object_id;++ const chunkKey = packedA[1].key;++ const chunk = blobs.get(`b2:${chunkKey}`);++ assert.ok(chunk && sha(chunk) === chunkId, 'the chunk is one object of the three segments');++ const packedB = await timeline.list('pack-B');++ for (const [i, r] of packedA.slice(1).entries()) {++ const b = packedB[i];++ assert.deepStrictEqual([b.name, b.start_ms, b.end_ms, b.sha256, Number(b.byte_length)], [r.name, r.start_ms, r.end_ms, r.sha256, Number(r.byte_length)], 'the clip keeps its identity');++ assert.deepStrictEqual([b.key, b.durable_provider, b.packed_object_id, Number(b.byte_offset), b.local_path], [chunkKey, 'b2', chunkId, Number(r.byte_offset), r.local_path], 'the clip names the same chunk range');++ assert.strictEqual(sha(chunk.subarray(Number(b.byte_offset), Number(b.byte_offset) + Number(b.byte_length))), b.sha256, `${b.name} plays from the chunk range its own playlist names`);++ }++ assert.strictEqual(packB.length, 3, 'three clip rows');++ const bPlaylist = timeline.mediaPlaylist(packedB);++ assert.ok(bPlaylist.includes('000001.m4s') && bPlaylist.includes('000003.m4s'), 'the clip\'s playlist is written from its (re-keyed) rows');++ assert.ok(packA.slice(1).every((r) => !fs.existsSync(r.local_path)) && packA.slice(1).every((r) => !blobs.has(`b2:${r.key}`)), 'the per-segment locations are freed once every row names the chunk');++++ // ── A row with a mismatched sha256 is left alone and keeps the location named ──++ const misA = await seedSource('mis-A');++ await insert('mis-B', { ...clipRow(misA[1], 1), sha256: 'f'.repeat(64) });++ assert.deepStrictEqual(await timeline.removeObject('mis-A'), { removed: misA.length, pending: 0 });++ assert.ok(fs.existsSync(misA[1].local_path) && blobs.has(`b2:${misA[1].key}`), 'the location the mismatched row names is kept');++ assert.ok(!deletes.includes(`b2:${misA[1].key}`), 'it was not deleted while the mismatched row names it');++ assert.ok(deletes.includes(`b2:${misA[2].key}`), 'the locations no row names were deleted');++ assert.deepStrictEqual((await timeline.list('mis-B')).map((r) => [r.seq, r.key, r.sha256]),++ [[1, misA[1].key, 'f'.repeat(64)]], 'the mismatched row is untouched');++ assert.deepStrictEqual(await timeline.removeObject('mis-B'), { removed: 1, pending: 0 });++ assert.ok(!fs.existsSync(misA[1].local_path) && !blobs.has(`b2:${misA[1].key}`), 'removing the last naming object frees it');++++ // ── Pack leaves a mismatched row alone and keeps the old location ──++ const pmA = await seedSource('pmis-A');++ await insert('pmis-B', { ...clipRow(pmA[1], 1), sha256: 'e'.repeat(64) });++ const pout = await pack.spec.run({ id: 'mjob_REFS2', app_id: 'live', object_id: 'pmis-A', params: { target_seconds: 60 } }, { signal: new AbortController().signal });++ assert.strictEqual(pout.packs, 1, JSON.stringify(pout));++ const pmPacked = await timeline.list('pmis-A');++ const pmChunk = pmPacked[1];++ assert.deepStrictEqual((await timeline.list('pmis-B')).map((r) => [r.key, r.sha256, r.packed_object_id]), [[pmA[1].key, 'e'.repeat(64), null]], 'the mismatched row did not follow the pack');++ assert.ok(fs.existsSync(pmA[1].local_path) && blobs.has(`b2:${pmA[1].key}`), 'so the old location it names is kept');++ assert.ok(pmA.slice(2).every((r) => !blobs.has(`b2:${r.key}`)), 'the other segments (no mismatched row) were freed');++ assert.ok(pmChunk.packed_object_id);++++ // ── Two removals over one shared location, started together: whatever order the scheduler gives their commits++ // and checks (one PGlite connection serializes the queries), they converge: no row, no orphan, no lost bytes ──++ const ccA = await seedSource('cc-A');++ for (const [i, r] of ccA.slice(1).entries()) await insert('cc-B', clipRow(r, i + 1));++ deletes.length = 0;++ const [ccRa, ccRb] = await Promise.all([timeline.removeObject('cc-A'), timeline.removeObject('cc-B')]);++ assert.deepStrictEqual(ccRa, { removed: ccA.length, pending: 0 }, 'the source rows all went');++ assert.deepStrictEqual(ccRb, { removed: ccA.length - 1, pending: 0 }, 'the clip rows all went');++ assert.strictEqual((await timeline.list('cc-A')).length + (await timeline.list('cc-B')).length, 0, 'no row is left');++ assert.ok(!await timeline.isNamed({ provider: 'b2', key: ccA[1].key }), 'no row names the shared location afterwards');++ assert.ok(!blobs.has(`b2:${ccA[1].key}`), 'the shared bytes are gone, not orphaned');++ assert.ok(deletes.filter((d) => d === `b2:${ccA[1].key}`).length >= 1, 'and they were deleted (twice is fine: a delete is idempotent)');++ assert.ok(ccA.slice(1).every((r) => !fs.existsSync(r.local_path)), 'the shared local files are gone too');++++ // ── A pack whose source was re-cut must not commit on a clip's same-sha row alone ──++ const rcA = await seedSource('rcut-A');++ await insert('rcut-B', clipRow(rcA[1], 1));++ const rcStale = await timeline.list('rcut-A'); // the rows the pack reads before the re-cut++ // The re-cut: the source's own row now names different bytes (a new key and sha); the clip's row is unchanged.++ await db.run(`UPDATE media_timeline SET key = ?, sha256 = ?, local_path = ? WHERE object_id = ? AND rendition = ? AND seq = ?`,++ ['rcut-A/source/000001.m4s/v2', 'b'.repeat(64), timeline.localPathFor({ app_id: 'live', id: 'rcut-A' }, 'source', '000001.m4s', 'v2'), 'rcut-A', 'source', 1]);++ const realList = timeline.list;++ timeline.list = async (id, rendition) => (id === 'rcut-A' ? rcStale : realList.call(timeline, id, rendition));++ let conflict = null;++ try {++ await pack.spec.run({ id: 'mjob_REFS3', app_id: 'live', object_id: 'rcut-A', params: { target_seconds: 60 } }, { signal: new AbortController().signal });++ } catch (err) { conflict = err; } finally { timeline.list = realList; }++ assert.ok(conflict && conflict.code === 'media.timeline.changed', `pack refused a re-cut source: ${conflict && conflict.message}`);++ const rcNow = await timeline.list('rcut-A');++ assert.strictEqual(rcNow[1].sha256, 'b'.repeat(64), 'the source row is still the re-cut one');++ assert.strictEqual(rcNow[1].packed_object_id, null, 'and was not packed without the source');++ assert.deepStrictEqual((await timeline.list('rcut-B')).map((r) => [r.key, r.sha256, r.packed_object_id]),++ [[rcA[1].key, rcA[1].sha256, null]], 'the clip row is untouched');++ assert.ok(await timeline.isNamed({ provider: 'b2', key: rcA[1].key }) && blobs.has(`b2:${rcA[1].key}`), 'the old location the clip names survives');++++ // ── A shared location whose durable delete fails is retried and deleted once unnamed ──++ const rtA = await seedSource('rt-A');++ for (const [i, r] of rtA.slice(1).entries()) await insert('rt-B', clipRow(r, i + 1));++ assert.deepStrictEqual(await timeline.removeObject('rt-A'), { removed: rtA.length, pending: 0 }, 'the source goes; the bytes stay');++ assert.ok(rtA.slice(1).every((r) => blobs.has(`b2:${r.key}`)), 'the shared locations survive while the clip names them');++ failDeletes.add(`b2:${rtA[1].key}`);++ assert.deepStrictEqual(await timeline.removeObject('rt-B'), { removed: 3, pending: 1 }, 'the failed delete leaves its row behind');++ assert.deepStrictEqual((await timeline.list('rt-B')).map((r) => [Number(r.seq), r.key, r.durable_provider]),++ [[1, rtA[1].key, 'b2']], 'the row survives with its key');++ assert.ok(blobs.has(`b2:${rtA[1].key}`), 'the bytes are still there');++ assert.ok(!blobs.has(`b2:${rtA[2].key}`), 'the other shared locations were deleted');++ failDeletes.clear();++ assert.deepStrictEqual(await timeline.removeObject('rt-B'), { removed: 1, pending: 0 }, 'the retry deletes it once unnamed');++ assert.ok(!blobs.has(`b2:${rtA[1].key}`) && !fs.existsSync(rtA[1].local_path), 'the bytes are gone');++ assert.strictEqual((await timeline.list('rt-B')).length, 0, 'and its row is dropped');++++ console.log('timeline-refs: all checks passed');++ process.exit(0);++ })().catch((err) => { console.error(err); process.exit(1); });++})().catch((err) => { console.error(err); process.exit(1); });
@@ -136,6 +136,20 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
@@ -136,6 +136,20 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
chunk's delete fails). All of it behind `MEDIA_HLS_ENABLED`; no schema change (F3.1's columns carry it). A finished
chunk's delete fails). All of it behind `MEDIA_HLS_ENABLED`; no schema change (F3.1's columns carry it). A finished
`object.cmaf` queues it by itself (`objects/timeline-queue.js`; `dedupeActive`, so a rerun while a pack is active joins
`object.cmaf` queues it by itself (`objects/timeline-queue.js`; `dedupeActive`, so a rerun while a pack is active joins
it and a later cut queues a new one), so packing follows the cut without an operator.
it and a later cut queues a new one), so packing follows the cut without an operator.
+- **F3.4, shipped: shared locations and reference counts (the foundation of materialized clips, §4).** A location names
+ bytes, not an owner: any number of `media_timeline` rows — of any number of objects (a source and the clips over it) —
+ may name one location, because a location is content-addressed and names the same bytes. **A location's bytes are
+ deleted exactly when no row names it.** The reference count is a query over the rows (`timeline.namedElsewhere`: the
+ rows of another object naming a durable `(durable_provider, key)` or a `local_path`), never a counter table, so a count
+ cannot drift from the rows; migration `0003_media_timeline_locations.sql` indexes both so the query is an index
+ lookup. `timeline.removeObject(objectId)`/`deleteBytes` keep a location another object's rows still name: deleting a
+ clip never deletes its source's bytes, deleting a source a clip still names keeps them (its rows go; the bytes are
+ deleted when the last naming object is removed), and only a delete that fails keeps rows for a retry, as before.
+ `object.pack` re-keys **every** row that names a packed segment's old location with the same sha256 — not only the
+ source's — to the chunk in the same transaction, so a clip follows its source into the chunk and the per-segment
+ location is then freed only once nothing names it; a row naming the old location with a different sha256 (which should
+ not happen) is left alone and keeps the location named. `object.cmaf` deletes an older cut's locations under the same
+ rule. No schema besides the two indexes; additive, an older release never reads it.
- **F3.c, shipped: the timeline queued by itself, signed playlists, the MP4 fallback.** A recording's finalize queues
- **F3.c, shipped: the timeline queued by itself, signed playlists, the MP4 fallback.** A recording's finalize queues
`object.cmaf` for its object once it is ready (`server/objects/timeline-queue.js`; `dedupeActive` and the idempotency
`object.cmaf` for its object once it is ready (`server/objects/timeline-queue.js`; `dedupeActive` and the idempotency
key `object.cmaf:<id>`, so a second finalize joins the same job; a queue error is logged, never fails the finalize),
key `object.cmaf:<id>`, so a second finalize joins the same job; a queue error is logged, never fails the finalize),
@@ -157,9 +171,9 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
@@ -157,9 +171,9 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
and metadata are the seek path's, so the player contract is unchanged. A source without a timeline, or with
and metadata are the seek path's, so the player contract is unchanged. A source without a timeline, or with
`MEDIA_HLS_ENABLED` off, keeps the one-fast-seek-per-frame path; captions stay with AI's `media.analyze`.
`MEDIA_HLS_ENABLED` off, keeps the one-fast-seek-per-frame path; captions stay with AI's `media.analyze`.
**Still open in F3:** the growing live/DVR playlist written as OpenRe segments (F3.2), write-behind durability for
**Still open in F3:** the growing live/DVR playlist written as OpenRe segments (F3.2), write-behind durability for
- live segments and its upload-lag metric, a materialized clip that references the source's chunks (chunk reference
- counts, boundary-only re-encode), captions on the timeline, segment-bucket demand, and Live's player
- moving to HLS. Renditions are F4.
+ live segments and its upload-lag metric, a materialized clip that references the source's chunks (reference counts
+ shipped, F3.4; its own persisted rows and boundary-only re-encode are open), captions on the timeline,
+ segment-bucket demand, and Live's player moving to HLS. Renditions are F4.
## 4. Clips reuse the source
## 4. Clips reuse the source
@@ -190,8 +204,13 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
@@ -190,8 +204,13 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
through `clip_of`.
through `clip_of`.
- **Materialized clip** (shared externally, downloaded, edited, popular): middle GOPs are **referenced**, not copied
- **Materialized clip** (shared externally, downloaded, edited, popular): middle GOPs are **referenced**, not copied
(the new manifest points at the source's packed chunks); only the two boundary segments are re-encoded for
(the new manifest points at the source's packed chunks); only the two boundary segments are re-encoded for
- frame-accurate cuts. Segments and chunks are content-addressed (sha256) with reference counts, so deleting a VOD
- that a clip still references keeps the referenced chunks (and holds still freeze everything).
+ frame-accurate cuts. Segments and chunks are content-addressed (sha256), so the reference count of a location is just
+ the rows that name it (F3.4, shipped: `timeline.namedElsewhere`, no counter table; migration 0003 indexes it) and
+ deleting a VOD that a clip still references keeps the referenced chunks — they go when the last naming object is
+ removed, and holds still freeze everything. The clip's own persisted rows over the source's locations are the next
+ step (§3, F3.4 carries the deletion rule). Inserting a clip's rows over a source's locations must check, in the same
+ transaction, that a source row still names each location (and take a per-location lock with the delete check), or a
+ concurrent delete can drop bytes the new clip names.
## 5. Derivatives are cost-managed computed caches
## 5. Derivatives are cost-managed computed caches
new file mode 100644
new file mode 100644
@@ -0,0 +1,10 @@
@@ -0,0 +1,10 @@
+-- phase: expand
+-- Reference counts for shared segment locations (docs/media-fabric.md §3/§4; the foundation of materialized clips). A
+-- row names a location by the durable pair (durable_provider, key) and/or its local copy (local_path); a location's
+-- bytes are shared whenever several objects' rows name it (a source and the clips over it). "How many objects name this
+-- location?" (timeline.namedElsewhere) is answered by a query over media_timeline — there is no counter table, so a count
+-- can never drift from the rows — and these two indexes make those reference queries index lookups.
+-- Additive only: an older release never reads it (it just keeps deleting its own locations, as before). Never edited
+-- after it runs; a further change is a new migration.
+CREATE INDEX IF NOT EXISTS media_timeline_location_durable ON media_timeline (durable_provider, key);
+CREATE INDEX IF NOT EXISTS media_timeline_location_local ON media_timeline (local_path);
@@ -186,8 +186,12 @@ async function run(job, ctx) {
@@ -186,8 +186,12 @@ async function run(job, ctx) {
// A deletion that fails leaves an orphan the storage report names, never bytes a live row still points at.
// A deletion that fails leaves an orphan the storage report names, never bytes a live row still points at.
const named = new Set(rows.map((r) => `${r.key}\n${r.local_path || ''}`));
const named = new Set(rows.map((r) => `${r.key}\n${r.local_path || ''}`));
const stale = [...prev.values()].filter((old) => !named.has(`${old.key}\n${old.local_path || ''}`));
const stale = [...prev.values()].filter((old) => !named.has(`${old.key}\n${old.local_path || ''}`));
- const { failed: stuck } = await timeline.deleteBytes(stale);
+ // A location any row still names — another object's, or the source's own (another rendition/seq, or a
+ // different local_path with the same key) — is kept: the old cut's rows go, the shared bytes stay, and the
+ // last naming object's removal deletes them. `isNamed` counts every object, no exception.
+ const { failed: stuck, kept } = await timeline.deleteBytes(stale);
if (stuck.length) console.warn(`[Cmaf] ${job.id}: ${stuck.length} old segment(s) left behind (the storage orphan report names them), first ${stuck[0].provider}:${stuck[0].key}`);
if (stuck.length) console.warn(`[Cmaf] ${job.id}: ${stuck.length} old segment(s) left behind (the storage orphan report names them), first ${stuck[0].provider}:${stuck[0].key}`);
+ if (kept.length) console.warn(`[Cmaf] ${job.id}: ${kept.length} old segment location(s) kept — another object's rows name them`);
if (failed.length) {
if (failed.length) {
throw new JobError('upload_failed', `${failed.length} segment(s) not durable yet (${failed[0].slice(0, 200)}); a retry uploads only those`, { retryAfterS: 600 });
throw new JobError('upload_failed', `${failed.length} segment(s) not durable yet (${failed[0].slice(0, 200)}); a retry uploads only those`, { retryAfterS: 600 });
}
}
@@ -170,7 +170,7 @@ async function run(job, ctx) {
@@ -170,7 +170,7 @@ async function run(job, ctx) {
if (!back || back.sha256 !== sha) throw new JobError('pack_unverified', `${provider}:${key} does not read back as ${sha.slice(0, 12)}`, { retryAfterS: 600 });
if (!back || back.sha256 !== sha) throw new JobError('pack_unverified', `${provider}:${key} does not read back as ${sha.slice(0, 12)}`, { retryAfterS: 600 });
if (localPath) { fs.mkdirSync(path.dirname(localPath), { recursive: true }); fs.renameSync(tmp, localPath); }
if (localPath) { fs.mkdirSync(path.dirname(localPath), { recursive: true }); fs.renameSync(tmp, localPath); }
aborted(ctx);
aborted(ctx);
- committed = await timeline.markPacked(src.id, rendition, placed.map((p) => ({
+ committed = await timeline.markPacked(src.id, placed.map((p) => ({
row: p.row, key, local_path: localPath, durable_provider: provider, packed_object_id: sha, byte_offset: p.byte_offset,
row: p.row, key, local_path: localPath, durable_provider: provider, packed_object_id: sha, byte_offset: p.byte_offset,
})), { jobId: job.id });
})), { jobId: job.id });
if (!committed) throw new JobError('media.timeline.changed', 'The timeline changed while packing (a re-cut); a retry packs the new segments', { retryAfterS: 60 });
if (!committed) throw new JobError('media.timeline.changed', 'The timeline changed while packing (a re-cut); a retry packs the new segments', { retryAfterS: 60 });
@@ -182,9 +182,12 @@ async function run(job, ctx) {
@@ -182,9 +182,12 @@ async function run(job, ctx) {
try { fs.unlinkSync(tmp); } catch { /* moved or gone */ }
try { fs.unlinkSync(tmp); } catch { /* moved or gone */ }
}
}
}
}
- // Phase two, after the commit: the per-segment copies the chunk now replaces.
- const { failed } = await timeline.deleteBytes(chunk);
+ // Phase two, after the commit: the per-segment copies the chunk now replaces. A location any row still
+ // names — another object's, or the source's own if it somehow did not follow the sha — is kept, not
+ // deleted: `isNamed` counts every object, no exception.
+ const { failed, kept } = await timeline.deleteBytes(chunk);
if (failed.length) console.warn(`[Pack] ${job.id}: ${failed.length} packed segment key(s) left behind (the storage orphan report names them), first ${failed[0].provider}:${failed[0].key}`);
if (failed.length) console.warn(`[Pack] ${job.id}: ${failed.length} packed segment key(s) left behind (the storage orphan report names them), first ${failed[0].provider}:${failed[0].key}`);
+ if (kept.length) console.warn(`[Pack] ${job.id}: ${kept.length} packed segment location(s) kept — another object's rows name them`);
out.left_behind += failed.length;
out.left_behind += failed.length;
out.removed_segments += chunk.length - failed.length;
out.removed_segments += chunk.length - failed.length;
out.packs++;
out.packs++;
@@ -16,6 +16,15 @@
@@ -16,6 +16,15 @@
* row's key/local_path/durable_provider then name the chunk, packed_object_id its sha256 (content address) and
* row's key/local_path/durable_provider then name the chunk, packed_object_id its sha256 (content address) and
* byte_offset/byte_length the segment's bytes inside it; its name, times and sha256 stay the segment's. Several rows
* byte_offset/byte_length the segment's bytes inside it; its name, times and sha256 stay the segment's. Several rows
* share one chunk, so the bytes are deleted once per location and kept (for a retry) while any delete of them fails.
* share one chunk, so the bytes are deleted once per location and kept (for a retry) while any delete of them fails.
+ *
+ * Locations are shared (§3/§4): any number of rows — of any number of objects (a source and the clips over it, F3.5/F4) —
+ * may name one location, because a location is content-addressed and names bytes, not an owner. A location's bytes are
+ * deleted only when no row names it. "How many objects name this location?" is a query over these rows (namedElsewhere),
+ * never a counter table, so a count cannot drift from them; migration 0003 indexes (durable_provider, key) and
+ * (local_path) for it. removeObject and deleteBytes therefore keep a location another object's rows still name, and
+ * object.pack re-keys every row that names a packed segment's old location (same sha256) to the chunk in the same
+ * transaction, so a clip follows its source into the chunk and the per-segment location is freed only once nothing names
+ * it. A source's own purge deletes its rows; the shared bytes go when the last naming object does.
*/
*/
'use strict';
'use strict';
@@ -109,62 +118,98 @@ async function replace(objectId, rendition, rows, { jobId = null, expect = null
@@ -109,62 +118,98 @@ async function replace(objectId, rendition, rows, { jobId = null, expect = null
/**
/**
* Delete the bytes a set of rows names: this node's copy (only inside the timeline root) and the durable copy. A remote
* Delete the bytes a set of rows names: this node's copy (only inside the timeline root) and the durable copy. A remote
- * delete that fails is returned, never swallowed: the caller keeps those rows, and their keys, so a later pass retries.
- * → { failed: [{ row, provider, key, error }] }.
+ * delete that fails is returned, never swallowed: the caller keeps (or restores) those rows, and their keys, so a later
+ * pass retries. A location any row still names (`isNamed`, across EVERY object, the caller's own included) is kept,
+ * never deleted: it is shared (a source and the clips over it). Callers therefore delete the rows that named a location
+ * first, in one transaction, so the check sees the other objects' rows: concurrent removals of objects sharing a
+ * location converge on the last committer deleting it, and a double delete is idempotent. → { failed: [{ row, provider,
+ * key, error }], kept: [{ row, provider, key }] } (kept = intentionally not deleted; for a local location provider is 'local').
*/
*/
async function deleteBytes(rows) {
async function deleteBytes(rows) {
const root = localRoot() + path.sep;
const root = localRoot() + path.sep;
const vodStorage = require('../vod/vod-storage');
const vodStorage = require('../vod/vod-storage');
const dirs = new Set();
const dirs = new Set();
const failed = [];
const failed = [];
- const done = new Set(); // a packed chunk is named by every row it holds: each location is deleted once
+ const kept = [];
+ const done = new Set(); // a packed chunk is named by every row it holds: each location is handled once
for (const r of rows) {
for (const r of rows) {
if (r.local_path && path.resolve(r.local_path).startsWith(root) && !done.has(`local\n${r.local_path}`)) {
if (r.local_path && path.resolve(r.local_path).startsWith(root) && !done.has(`local\n${r.local_path}`)) {
done.add(`local\n${r.local_path}`);
done.add(`local\n${r.local_path}`);
- try { fs.unlinkSync(r.local_path); } catch { /* already gone */ }
- dirs.add(path.dirname(r.local_path));
+ if (await isNamed({ localPath: r.local_path })) kept.push({ row: r, provider: 'local', key: r.local_path });
+ else { try { fs.unlinkSync(r.local_path); } catch { /* already gone */ } dirs.add(path.dirname(r.local_path)); }
}
}
if (r.durable_provider && vodStorage.providerConfigured(r.durable_provider) && !done.has(`${r.durable_provider}\n${r.key}`)) {
if (r.durable_provider && vodStorage.providerConfigured(r.durable_provider) && !done.has(`${r.durable_provider}\n${r.key}`)) {
done.add(`${r.durable_provider}\n${r.key}`);
done.add(`${r.durable_provider}\n${r.key}`);
- const ok = await vodStorage.deleteObject(r.durable_provider, r.key);
- if (!ok) failed.push({ row: r, provider: r.durable_provider, key: r.key, error: 'delete failed' });
+ if (await isNamed({ provider: r.durable_provider, key: r.key })) kept.push({ row: r, provider: r.durable_provider, key: r.key });
+ else {
+ const ok = await vodStorage.deleteObject(r.durable_provider, r.key);
+ if (!ok) failed.push({ row: r, provider: r.durable_provider, key: r.key, error: 'delete failed' });
+ }
}
}
}
}
for (const d of dirs) {
for (const d of dirs) {
for (const p of [d, path.dirname(d)]) { try { fs.rmdirSync(p); } catch { /* not empty or gone */ } }
for (const p of [d, path.dirname(d)]) { try { fs.rmdirSync(p); } catch { /* not empty or gone */ } }
}
}
- return { failed };
+ return { failed, kept };
}
}
/**
/**
- * Whether any timeline row still names this location (a durable `provider` + `key`, or a local `localPath`). Keys and
- * paths are content-addressed, so a run that lost a race may have built exactly the bytes a winner committed: it deletes
- * what it staged only when no row names it.
+ * Whether any row of ANOTHER object names this location (a durable `provider` + `key`, or a local `localPath`), i.e.
+ * whether its bytes are shared with an object other than `exceptObjectId`. That is the reference count's only question:
+ * a location's bytes may be deleted exactly when this is false. Keys and paths are content-addressed, so two objects'
+ * rows can name one location (a source and the clips over it). `exceptObjectId`, when given, is an object whose own rows
+ * to ignore (a caller that has not deleted them yet); null counts every object (isNamed).
*/
*/
-async function isNamed({ provider = null, key = null, localPath = null }) {
- if (provider && key && await db.get('SELECT 1 AS x FROM media_timeline WHERE durable_provider = ? AND key = ? LIMIT 1', [provider, key])) return true;
- if (localPath && await db.get('SELECT 1 AS x FROM media_timeline WHERE local_path = ? LIMIT 1', [localPath])) return true;
+async function namedElsewhere({ provider = null, key = null, localPath = null }, exceptObjectId = null) {
+ const skip = exceptObjectId ? ' AND object_id <> ?' : '';
+ const rest = exceptObjectId ? [exceptObjectId] : [];
+ if (provider && key && await db.get(`SELECT 1 AS x FROM media_timeline WHERE durable_provider = ? AND key = ?${skip} LIMIT 1`, [provider, key, ...rest])) return true;
+ if (localPath && await db.get(`SELECT 1 AS x FROM media_timeline WHERE local_path = ?${skip} LIMIT 1`, [localPath, ...rest])) return true;
return false;
return false;
}
}
/**
/**
- * The object's whole timeline is gone with its bytes (a purge, or a vod/clip deleted for good): local files, then the
- * durable copies. A row whose durable delete failed is kept — with its key — so a later pass can retry; dropping it
- * would lose the only record of the bytes still in B2/R2. Callers have already refused a held object.
- * → { removed, pending } (pending = rows kept for a retry).
+ * Whether any timeline row still names this location. Keys and paths are content-addressed, so a run that lost a race
+ * may have built exactly the bytes a winner committed: it deletes what it staged only when no row names it.
+ */
+async function isNamed(loc) { return await namedElsewhere(loc, null); }
+
+/** Put a row back exactly as it was read, so a later pass retries a delete that failed (a packed chunk: every row). */
+async function reinsertRow(r) {
+ await db.run(`INSERT INTO media_timeline (object_id, rendition, seq, ${FIELDS.join(', ')}, job_id)
+ VALUES (?, ?, ?, ${FIELDS.map(() => '?').join(', ')}, ?)
+ ON CONFLICT (object_id, rendition, seq) DO UPDATE SET ${FIELDS.map((f) => `${f} = excluded.${f}`).join(', ')},
+ job_id = excluded.job_id, updated_at = ov_now()`,
+ [r.object_id, r.rendition, r.seq, ...FIELDS.map((f) => r[f] ?? null), r.job_id ?? null]);
+}
+
+/**
+ * The object's whole timeline is gone with its bytes (a purge, or a vod/clip deleted for good): row, then local files,
+ * then the durable copies. The rows go first, in one transaction, and the bytes are deleted only once `isNamed` — a
+ * check across ALL objects — sees no row naming a location; a location another object's rows still name is kept (its
+ * rows are already gone, the shared bytes stay, and they are deleted when the last naming object is removed). That is
+ * what makes concurrent removals converge: each remover deletes its rows before checking, so the last one to commit
+ * sees no naming row and deletes, and the loser's double delete is idempotent. A durable delete that failed gets its
+ * rows back — with their keys — so a later pass retries; dropping them would lose the only record of the bytes still in
+ * B2/R2. Callers have already refused a held object. → { removed, pending } (pending = rows restored for a retry).
*/
*/
async function removeObject(objectId) {
async function removeObject(objectId) {
if (!objectId) return { removed: 0, pending: 0 };
if (!objectId) return { removed: 0, pending: 0 };
const rows = await db.all('SELECT * FROM media_timeline WHERE object_id = ?', [objectId]);
const rows = await db.all('SELECT * FROM media_timeline WHERE object_id = ?', [objectId]);
if (!rows.length) return { removed: 0, pending: 0 };
if (!rows.length) return { removed: 0, pending: 0 };
+ // (a) The rows go first, in one transaction, before their bytes; the delete is not conditioned on what another
+ // object still names, so two concurrent removals of a shared location cannot each keep it and both drop their rows.
+ let removed = 0;
+ await db.getDb().tx(async () => {
+ removed = (await db.run('DELETE FROM media_timeline WHERE object_id = ?', [objectId])).changes || 0;
+ });
+ // (b) After the commit, each location the rows named, deleted only when no row of ANY object names it.
const { failed } = await deleteBytes(rows);
const { failed } = await deleteBytes(rows);
- // Every row naming a location whose delete failed is kept: for a packed chunk that is all the rows it holds.
+ // (c) Every row naming a location whose delete failed is restored: for a packed chunk that is all the rows it holds.
const stuck = new Set(failed.map((f) => `${f.provider}\n${f.key}`));
const stuck = new Set(failed.map((f) => `${f.provider}\n${f.key}`));
- let removed = 0;
let pending = 0;
let pending = 0;
for (const r of rows) {
for (const r of rows) {
- if (r.durable_provider && stuck.has(`${r.durable_provider}\n${r.key}`)) { pending++; continue; }
- removed += (await db.run('DELETE FROM media_timeline WHERE object_id = ? AND rendition = ? AND seq = ?', [objectId, r.rendition, r.seq])).changes || 0;
+ if (r.durable_provider && stuck.has(`${r.durable_provider}\n${r.key}`)) { await reinsertRow(r); pending++; }
}
}
if (pending) console.warn(`[Timeline] ${objectId}: ${pending} segment row(s) kept — their durable copy could not be deleted`);
if (pending) console.warn(`[Timeline] ${objectId}: ${pending} segment row(s) kept — their durable copy could not be deleted`);
return { removed, pending };
return { removed, pending };
@@ -172,21 +217,38 @@ async function removeObject(objectId) {
@@ -172,21 +217,38 @@ async function removeObject(objectId) {
/**
/**
* Point rows at the chunk they were packed into, in one transaction. `updates` = [{ row, key, local_path,
* Point rows at the chunk they were packed into, in one transaction. `updates` = [{ row, key, local_path,
- * durable_provider, packed_object_id, byte_offset }]; each row changes only if it is still exactly the unpacked segment
- * the packer read (same key and sha256, not packed): a re-cut in between makes the whole commit roll back and false is
- * returned, so the caller drops its chunk. Never deletes a row.
+ * durable_provider, packed_object_id, byte_offset }]; the location is the segment's old key (content-addressed, versioned
+ * by its sha), and EVERY row of any object that names it with the same sha256 — the source's, and a clip's over it —
+ * moves to the chunk in this same transaction, so a clip follows its source into the chunk. Each group changes only if
+ * it still names the unpacked segment the packer read (same key and sha256, not packed) AND at least one changed row is
+ * `objectId`'s own: a re-cut in between makes the whole commit roll back and false is returned, so the caller drops its
+ * chunk. Without that second condition a re-cut source row (new key/sha) would leave only a clip's same-sha row matching
+ * and the pack would commit while the source is not in the chunk. A row naming the old location with a different sha256
+ * (which should not happen) does not match, is left alone, and so keeps the old location named (its bytes are not
+ * deleted). Never deletes a row.
*/
*/
-async function markPacked(objectId, rendition, updates, { jobId = null } = {}) {
+async function markPacked(objectId, updates, { jobId = null } = {}) {
const CONFLICT = new Error('timeline changed');
const CONFLICT = new Error('timeline changed');
try {
try {
await db.getDb().tx(async () => {
await db.getDb().tx(async () => {
await lockObject(objectId);
await lockObject(objectId);
for (const u of updates) {
for (const u of updates) {
+ // The old location: the durable pair when the row was durable (packed rows always are), the local copy
+ // too. The (durable_provider, key) index serves the first, (local_path) the second.
+ const loc = u.row.durable_provider ? '(durable_provider = ? AND key = ?)' : 'key = ?';
+ const at = u.row.durable_provider ? [u.row.durable_provider, u.row.key] : [u.row.key];
+ const names = u.row.local_path ? `(${loc} OR local_path = ?)` : loc;
const r = await db.run(`UPDATE media_timeline SET key = ?, local_path = ?, durable_provider = ?, durability = 'durable',
const r = await db.run(`UPDATE media_timeline SET key = ?, local_path = ?, durable_provider = ?, durability = 'durable',
- packed_object_id = ?, byte_offset = ?, job_id = ?, updated_at = ov_now()
- WHERE object_id = ? AND rendition = ? AND seq = ? AND key = ? AND sha256 = ? AND packed_object_id IS NULL`,
- [u.key, u.local_path, u.durable_provider, u.packed_object_id, u.byte_offset, jobId, objectId, rendition, u.row.seq, u.row.key, u.row.sha256]);
- if (!r.changes) throw CONFLICT;
+ packed_object_id = ?, byte_offset = ?, byte_length = ?, job_id = ?, updated_at = ov_now()
+ WHERE sha256 = ? AND packed_object_id IS NULL AND ${names}
+ RETURNING object_id`,
+ [u.key, u.local_path, u.durable_provider, u.packed_object_id, u.byte_offset, u.row.byte_length, jobId,
+ u.row.sha256, ...at, ...(u.row.local_path ? [u.row.local_path] : [])]);
+ // The pack changes `objectId`'s OWN timeline: at least one re-keyed row must be its. A re-cut between
+ // the read and this commit leaves the source's row with a new key/sha; the same-sha match is then only
+ // some other object's row (a clip's), and committing would leave the source out of the chunk it names.
+ // Treat that as a conflict — nothing is written and the caller drops its chunk and retries.
+ if (!r.rows.some((row) => String(row.object_id) === String(objectId))) throw CONFLICT;
}
}
});
});
return true;
return true;
@@ -275,6 +337,6 @@ function clipRows(rows, startMs, endMs) {
@@ -275,6 +337,6 @@ function clipRows(rows, startMs, endMs) {
module.exports = {
module.exports = {
SOURCE, INIT_NAME, FIELDS, localRoot, localPathFor, keyFor, segmentName,
SOURCE, INIT_NAME, FIELDS, localRoot, localPathFor, keyFor, segmentName,
- list, segments, byName, has, segmentAt, replace, TimelineChanged, isNamed, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,
+ list, segments, byName, has, segmentAt, replace, TimelineChanged, isNamed, namedElsewhere, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,
mediaPlaylist, masterPlaylist, clipRows,
mediaPlaylist, masterPlaylist, clipRows,
};
};
@@ -165,6 +165,10 @@ const { spawn, spawnSync } = require('child_process');
@@ -165,6 +165,10 @@ const { spawn, spawnSync } = require('child_process');
const sto = await model.getObject(st.result.object_id);
const sto = await model.getObject(st.result.object_id);
assert.strictEqual(model.parseJson(sto.metadata, {}).source_duration_seconds, 7, 'the seek metadata carries the real duration');
assert.strictEqual(model.parseJson(sto.metadata, {}).source_duration_seconds, 7, 'the seek metadata carries the real duration');
assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');
assert.strictEqual(fs.readdirSync(path.join(process.env.OBJECTS_PATH, '.jobs')).length, 0, 'the work directory is removed');
+ // shorty's rows name pub's segment files (two objects, one location): dropping them keeps the bytes, which pub still names,
+ // and lets the purge below delete them, since after that no row of any object names them
+ await timeline.removeObject(shorty);
+ assert.ok(rows.every((r) => fs.existsSync(r.local_path)), 'pub still names its files after a sharing object is removed');
// ── The finished cut queues object.pack by itself, once (the worker runs it; no provider yet, so it packs nothing) ──
// ── The finished cut queues object.pack by itself, once (the worker runs it; no provider yet, so it packs nothing) ──
const packJobs = async (objectId) => (await queue.list('live', { type: 'object.pack', objectId })).jobs;
const packJobs = async (objectId) => (await queue.list('live', { type: 'object.pack', objectId })).jobs;
new file mode 100644
new file mode 100644
@@ -0,0 +1,234 @@
@@ -0,0 +1,234 @@
+'use strict';
+// Shared segment locations and their reference counts (F3.4; docs/media-fabric.md §3/§4): any number of media_timeline
+// rows — of any number of objects (a source and the clips over it) — may name one location, and a location's bytes are
+// deleted only when no row names it. A clip object's rows name its source's locations; removing the clip leaves the
+// source's files and durable keys, removing the source keeps the locations the clip still names, and in either order the
+// bytes go when the last naming object does. object.pack re-keys every row that names a packed segment's old location
+// (same sha256) to the chunk in one transaction, so a clip follows its source into the chunk and its playlist serves the
+// chunk ranges; the old location is then freed. A row whose sha256 differs is left alone and keeps the location named.
+// Migration 0003 adds the two location indexes and applies on a database that already has rows.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const crypto = require('crypto');
+(async () => {
+ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-timeline-refs-'));
+ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+ process.env.OBJECTS_PATH = dir('objects');
+ process.env.VOD_PATH = dir('vods');
+ process.env.CLIPS_PATH = dir('clips');
+ process.env.MEDIA_HLS_ENABLED = '1';
+ for (const p of ['B2', 'R2']) for (const k of ['ENDPOINT', 'BUCKET', 'KEY_ID', 'APP_KEY', 'ACCESS_KEY_ID', 'SECRET_ACCESS_KEY']) process.env[`MEDIA_${p}_${k}`] = '';
+
+ (async () => {
+ const config = require('../server/config');
+ const db = require('../server/db/database');
+ const timeline = require('../server/objects/timeline');
+ const vodStorage = require('../server/vod/vod-storage');
+ const sha = (buf) => crypto.createHash('sha256').update(buf).digest('hex');
+
+ // The durable providers, stubbed: bytes in a Map, uploads kept, and deletes recorded only when a key actually
+ // went (a delete of bytes already gone is a success, as S3's is — so a concurrent double delete records once).
+ const blobs = new Map();
+ const deletes = [];
+ const failDeletes = new Set(); // keys whose durable delete fails once (returns false), for the retry case
+ Object.assign(vodStorage, {
+ providerConfigured: (p) => ['b2', 'r2'].includes(p),
+ providerAvailable: (p) => ['b2', 'r2'].includes(p),
+ uploadFile: async (p, key, file) => { blobs.set(`${p}:${key}`, fs.readFileSync(file)); },
+ sha256Object: async (p, key) => (blobs.has(`${p}:${key}`) ? { sha256: sha(blobs.get(`${p}:${key}`)), size: blobs.get(`${p}:${key}`).length } : null),
+ deleteObject: async (p, key) => {
+ if (failDeletes.has(`${p}:${key}`)) return false;
+ if (blobs.delete(`${p}:${key}`)) deletes.push(`${p}:${key}`);
+ return true;
+ },
+ presignGet: async () => null,
+ });
+
+ const insert = async (objectId, r) => db.run(
+ `INSERT INTO media_timeline (object_id, rendition, seq, name, start_ms, end_ms, keyframe_ms, key, local_path,
+ durable_provider, packed_object_id, byte_offset, byte_length, sha256, durability, job_id)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)`,
+ [objectId, 'source', r.seq, r.name, r.start_ms, r.end_ms, r.keyframe_ms ?? null, r.key, r.local_path,
+ r.durable_provider, r.packed_object_id ?? null, r.byte_offset ?? null, r.byte_length, r.sha256, r.durability]);
+
+ /** One object's rows (init + three media segments), with this node's file and the durable copy really written. */
+ const seedSource = async (id) => {
+ const obj = { app_id: 'live', id };
+ const payloads = [Buffer.from('init-bytes'), Buffer.from('segment-one-bytes'), Buffer.from('segment-two-bytes'), Buffer.from('segment-three-bytes')];
+ const rows = [];
+ for (let seq = 0; seq < payloads.length; seq++) {
+ const bytes = payloads[seq];
+ const s = sha(bytes);
+ const name = timeline.segmentName(seq);
+ const key = timeline.keyFor(id, 'source', name, s.slice(0, 12));
+ const local = timeline.localPathFor(obj, 'source', name, s.slice(0, 12));
+ fs.mkdirSync(path.dirname(local), { recursive: true });
+ fs.writeFileSync(local, bytes);
+ blobs.set(`b2:${key}`, bytes);
+ const r = { seq, name, start_ms: seq ? (seq - 1) * 3000 : 0, end_ms: seq * 3000, keyframe_ms: seq ? (seq - 1) * 3000 : null,
+ key, local_path: local, durable_provider: 'b2', byte_length: bytes.length, sha256: s, durability: 'durable' };
+ await insert(id, r);
+ rows.push(r);
+ }
+ return rows;
+ };
+
+ /** A clip's row naming a source row's location: the location (key, local_path, durable provider, sha, byte
+ * length) is the source's, the seq and times are the clip's, exactly as timeline.clipRows leaves them. */
+ const clipRow = (src, seq, { sha256 = src.sha256 } = {}) => ({ seq, name: src.name, start_ms: src.start_ms, end_ms: src.end_ms,
+ keyframe_ms: src.keyframe_ms, key: src.key, local_path: src.local_path, durable_provider: src.durable_provider,
+ byte_length: src.byte_length, sha256, durability: src.durability });
+
+ // ── The migration: the two location indexes exist, and it applies again on a database that already has rows ──
+ await seedSource('idx-A');
+ const indexes = (await db.all(`SELECT indexname FROM pg_indexes WHERE schemaname = current_schema() AND tablename = 'media_timeline'`)).map((r) => r.indexname);
+ assert.ok(indexes.includes('media_timeline_location_durable') && indexes.includes('media_timeline_location_local'), `migration 0003 applied: ${indexes.join(', ')}`);
+ const migration = fs.readFileSync(path.join(__dirname, '..', 'migrations', '0003_media_timeline_locations.sql'), 'utf8');
+ // CREATE INDEX IF NOT EXISTS: a no-op on the row-filled database, never an error. Only on PGlite: under test:pg the test
+ // runs as the service role, and PostgreSQL checks table ownership before IF NOT EXISTS (the migration runner applies
+ // migrations as the owner).
+ if (process.env.MEDIA_TEST_STORE !== 'pg') await db.getDb().query(migration);
+ await timeline.removeObject('idx-A');
+ deletes.length = 0;
+
+ // ── namedElsewhere: another object's rows, and only another object's ──
+ const A = await seedSource('ref-A');
+ const B = A.slice(1).map((r, i) => clipRow(r, i + 1));
+ for (const r of B) await insert('ref-B', r);
+ const loc = { provider: 'b2', key: A[1].key };
+ assert.strictEqual(await timeline.namedElsewhere(loc, 'ref-A'), true, 'the clip names the source location');
+ assert.strictEqual(await timeline.namedElsewhere(loc, 'ref-B'), true, 'the source names it');
+ assert.strictEqual(await timeline.namedElsewhere({ localPath: A[1].local_path }, 'ref-A'), true, 'the local location too');
+ assert.strictEqual(await timeline.namedElsewhere({ provider: 'b2', key: 'nobody' }, 'ref-A'), false, 'an unnamed key is not');
+ assert.strictEqual(await timeline.namedElsewhere(loc, null), true, 'isNamed counts every object');
+
+ // ── Case 1: removeObject(clip) keeps the source's bytes; removeObject(source) then deletes them ──
+ assert.deepStrictEqual(await timeline.removeObject('ref-B'), { removed: B.length, pending: 0 });
+ assert.strictEqual((await timeline.list('ref-B')).length, 0, 'the clip\'s rows are gone');
+ assert.ok(A.every((r) => fs.existsSync(r.local_path)), 'the source\'s files stay');
+ assert.ok(A.every((r) => blobs.has(`b2:${r.key}`)), 'the source\'s durable keys stay');
+ assert.deepStrictEqual(deletes, [], 'nothing was deleted while the source still names them');
+ assert.deepStrictEqual(await timeline.removeObject('ref-A'), { removed: A.length, pending: 0 });
+ assert.ok(A.every((r) => !fs.existsSync(r.local_path)), 'the source\'s files go with the last naming object');
+ assert.ok(A.every((r) => !blobs.has(`b2:${r.key}`)), 'and its durable keys');
+
+ // ── Case 2: removeObject(source) first keeps the locations the clip still names; removeObject(clip) deletes them ──
+ const A2 = await seedSource('ref-A2');
+ for (const [i, r] of A2.slice(1).entries()) await insert('ref-B2', clipRow(r, i + 1));
+ assert.deepStrictEqual(await timeline.removeObject('ref-A2'), { removed: A2.length, pending: 0 });
+ assert.ok(A2.slice(1).every((r) => fs.existsSync(r.local_path)) && A2.slice(1).every((r) => blobs.has(`b2:${r.key}`)), 'the source is gone, the segments the clip names are kept');
+ assert.ok(!fs.existsSync(A2[0].local_path) && !blobs.has(`b2:${A2[0].key}`), 'the init segment, named by no clip row, went with the source');
+ assert.ok(deletes.includes(`b2:${A2[0].key}`) && !deletes.includes(`b2:${A2[1].key}`), 'only the unshared location was deleted');
+ assert.strictEqual((await timeline.list('ref-B2')).length, A2.length - 1, 'the clip keeps its rows');
+ assert.deepStrictEqual(await timeline.removeObject('ref-B2'), { removed: A2.length - 1, pending: 0 });
+ assert.ok(A2.slice(1).every((r) => !fs.existsSync(r.local_path)) && A2.slice(1).every((r) => !blobs.has(`b2:${r.key}`)), 'the last naming object deletes them');
+
+ // ── Pack: a clip follows its source into the chunk, in the same transaction; the old locations are freed ──
+ const derive = require('../server/jobs/derive');
+ const pack = require('../server/jobs/pack');
+ derive.loadSource = async (job) => ({ id: job.object_id, app_id: 'live', kind: 'video', mime_type: 'video/mp4', lifecycle_status: 'ready' });
+ timeline.durableProvider = async () => 'b2';
+ const packA = await seedSource('pack-A');
+ const packB = packA.slice(1).map((r, i) => clipRow(r, i + 1));
+ for (const r of packB) await insert('pack-B', r);
+ const job = { id: 'mjob_REFS', app_id: 'live', object_id: 'pack-A', params: { target_seconds: 60 } };
+ const out = await pack.spec.run(job, { signal: new AbortController().signal });
+ assert.deepStrictEqual([out.packs, out.segments, out.left_behind], [1, 3, 0], JSON.stringify(out));
+ const packedA = await timeline.list('pack-A');
+ const chunkId = packedA[1].packed_object_id;
+ const chunkKey = packedA[1].key;
+ const chunk = blobs.get(`b2:${chunkKey}`);
+ assert.ok(chunk && sha(chunk) === chunkId, 'the chunk is one object of the three segments');
+ const packedB = await timeline.list('pack-B');
+ for (const [i, r] of packedA.slice(1).entries()) {
+ const b = packedB[i];
+ assert.deepStrictEqual([b.name, b.start_ms, b.end_ms, b.sha256, Number(b.byte_length)], [r.name, r.start_ms, r.end_ms, r.sha256, Number(r.byte_length)], 'the clip keeps its identity');
+ assert.deepStrictEqual([b.key, b.durable_provider, b.packed_object_id, Number(b.byte_offset), b.local_path], [chunkKey, 'b2', chunkId, Number(r.byte_offset), r.local_path], 'the clip names the same chunk range');
+ assert.strictEqual(sha(chunk.subarray(Number(b.byte_offset), Number(b.byte_offset) + Number(b.byte_length))), b.sha256, `${b.name} plays from the chunk range its own playlist names`);
+ }
+ assert.strictEqual(packB.length, 3, 'three clip rows');
+ const bPlaylist = timeline.mediaPlaylist(packedB);
+ assert.ok(bPlaylist.includes('000001.m4s') && bPlaylist.includes('000003.m4s'), 'the clip\'s playlist is written from its (re-keyed) rows');
+ assert.ok(packA.slice(1).every((r) => !fs.existsSync(r.local_path)) && packA.slice(1).every((r) => !blobs.has(`b2:${r.key}`)), 'the per-segment locations are freed once every row names the chunk');
+
+ // ── A row with a mismatched sha256 is left alone and keeps the location named ──
+ const misA = await seedSource('mis-A');
+ await insert('mis-B', { ...clipRow(misA[1], 1), sha256: 'f'.repeat(64) });
+ assert.deepStrictEqual(await timeline.removeObject('mis-A'), { removed: misA.length, pending: 0 });
+ assert.ok(fs.existsSync(misA[1].local_path) && blobs.has(`b2:${misA[1].key}`), 'the location the mismatched row names is kept');
+ assert.ok(!deletes.includes(`b2:${misA[1].key}`), 'it was not deleted while the mismatched row names it');
+ assert.ok(deletes.includes(`b2:${misA[2].key}`), 'the locations no row names were deleted');
+ assert.deepStrictEqual((await timeline.list('mis-B')).map((r) => [r.seq, r.key, r.sha256]),
+ [[1, misA[1].key, 'f'.repeat(64)]], 'the mismatched row is untouched');
+ assert.deepStrictEqual(await timeline.removeObject('mis-B'), { removed: 1, pending: 0 });
+ assert.ok(!fs.existsSync(misA[1].local_path) && !blobs.has(`b2:${misA[1].key}`), 'removing the last naming object frees it');
+
+ // ── Pack leaves a mismatched row alone and keeps the old location ──
+ const pmA = await seedSource('pmis-A');
+ await insert('pmis-B', { ...clipRow(pmA[1], 1), sha256: 'e'.repeat(64) });
+ const pout = await pack.spec.run({ id: 'mjob_REFS2', app_id: 'live', object_id: 'pmis-A', params: { target_seconds: 60 } }, { signal: new AbortController().signal });
+ assert.strictEqual(pout.packs, 1, JSON.stringify(pout));
+ const pmPacked = await timeline.list('pmis-A');
+ const pmChunk = pmPacked[1];
+ assert.deepStrictEqual((await timeline.list('pmis-B')).map((r) => [r.key, r.sha256, r.packed_object_id]), [[pmA[1].key, 'e'.repeat(64), null]], 'the mismatched row did not follow the pack');
+ assert.ok(fs.existsSync(pmA[1].local_path) && blobs.has(`b2:${pmA[1].key}`), 'so the old location it names is kept');
+ assert.ok(pmA.slice(2).every((r) => !blobs.has(`b2:${r.key}`)), 'the other segments (no mismatched row) were freed');
+ assert.ok(pmChunk.packed_object_id);
+
+ // ── Two removals over one shared location, started together: whatever order the scheduler gives their commits
+ // and checks (one PGlite connection serializes the queries), they converge: no row, no orphan, no lost bytes ──
+ const ccA = await seedSource('cc-A');
+ for (const [i, r] of ccA.slice(1).entries()) await insert('cc-B', clipRow(r, i + 1));
+ deletes.length = 0;
+ const [ccRa, ccRb] = await Promise.all([timeline.removeObject('cc-A'), timeline.removeObject('cc-B')]);
+ assert.deepStrictEqual(ccRa, { removed: ccA.length, pending: 0 }, 'the source rows all went');
+ assert.deepStrictEqual(ccRb, { removed: ccA.length - 1, pending: 0 }, 'the clip rows all went');
+ assert.strictEqual((await timeline.list('cc-A')).length + (await timeline.list('cc-B')).length, 0, 'no row is left');
+ assert.ok(!await timeline.isNamed({ provider: 'b2', key: ccA[1].key }), 'no row names the shared location afterwards');
+ assert.ok(!blobs.has(`b2:${ccA[1].key}`), 'the shared bytes are gone, not orphaned');
+ assert.ok(deletes.filter((d) => d === `b2:${ccA[1].key}`).length >= 1, 'and they were deleted (twice is fine: a delete is idempotent)');
+ assert.ok(ccA.slice(1).every((r) => !fs.existsSync(r.local_path)), 'the shared local files are gone too');
+
+ // ── A pack whose source was re-cut must not commit on a clip's same-sha row alone ──
+ const rcA = await seedSource('rcut-A');
+ await insert('rcut-B', clipRow(rcA[1], 1));
+ const rcStale = await timeline.list('rcut-A'); // the rows the pack reads before the re-cut
+ // The re-cut: the source's own row now names different bytes (a new key and sha); the clip's row is unchanged.
+ await db.run(`UPDATE media_timeline SET key = ?, sha256 = ?, local_path = ? WHERE object_id = ? AND rendition = ? AND seq = ?`,
+ ['rcut-A/source/000001.m4s/v2', 'b'.repeat(64), timeline.localPathFor({ app_id: 'live', id: 'rcut-A' }, 'source', '000001.m4s', 'v2'), 'rcut-A', 'source', 1]);
+ const realList = timeline.list;
+ timeline.list = async (id, rendition) => (id === 'rcut-A' ? rcStale : realList.call(timeline, id, rendition));
+ let conflict = null;
+ try {
+ await pack.spec.run({ id: 'mjob_REFS3', app_id: 'live', object_id: 'rcut-A', params: { target_seconds: 60 } }, { signal: new AbortController().signal });
+ } catch (err) { conflict = err; } finally { timeline.list = realList; }
+ assert.ok(conflict && conflict.code === 'media.timeline.changed', `pack refused a re-cut source: ${conflict && conflict.message}`);
+ const rcNow = await timeline.list('rcut-A');
+ assert.strictEqual(rcNow[1].sha256, 'b'.repeat(64), 'the source row is still the re-cut one');
+ assert.strictEqual(rcNow[1].packed_object_id, null, 'and was not packed without the source');
+ assert.deepStrictEqual((await timeline.list('rcut-B')).map((r) => [r.key, r.sha256, r.packed_object_id]),
+ [[rcA[1].key, rcA[1].sha256, null]], 'the clip row is untouched');
+ assert.ok(await timeline.isNamed({ provider: 'b2', key: rcA[1].key }) && blobs.has(`b2:${rcA[1].key}`), 'the old location the clip names survives');
+
+ // ── A shared location whose durable delete fails is retried and deleted once unnamed ──
+ const rtA = await seedSource('rt-A');
+ for (const [i, r] of rtA.slice(1).entries()) await insert('rt-B', clipRow(r, i + 1));
+ assert.deepStrictEqual(await timeline.removeObject('rt-A'), { removed: rtA.length, pending: 0 }, 'the source goes; the bytes stay');
+ assert.ok(rtA.slice(1).every((r) => blobs.has(`b2:${r.key}`)), 'the shared locations survive while the clip names them');
+ failDeletes.add(`b2:${rtA[1].key}`);
+ assert.deepStrictEqual(await timeline.removeObject('rt-B'), { removed: 3, pending: 1 }, 'the failed delete leaves its row behind');
+ assert.deepStrictEqual((await timeline.list('rt-B')).map((r) => [Number(r.seq), r.key, r.durable_provider]),
+ [[1, rtA[1].key, 'b2']], 'the row survives with its key');
+ assert.ok(blobs.has(`b2:${rtA[1].key}`), 'the bytes are still there');
+ assert.ok(!blobs.has(`b2:${rtA[2].key}`), 'the other shared locations were deleted');
+ failDeletes.clear();
+ assert.deepStrictEqual(await timeline.removeObject('rt-B'), { removed: 1, pending: 0 }, 'the retry deletes it once unnamed');
+ assert.ok(!blobs.has(`b2:${rtA[1].key}`) && !fs.existsSync(rtA[1].local_path), 'the bytes are gone');
+ assert.strictEqual((await timeline.list('rt-B')).length, 0, 'and its row is dropped');
+
+ console.log('timeline-refs: all checks passed');
+ process.exit(0);
+ })().catch((err) => { console.error(err); process.exit(1); });
+})().catch((err) => { console.error(err); process.exit(1); });