openvibers/openvibe.media Commits

Version cmaf segment keys and delete old bytes only after commit

OP OpenVibers committed 4 Oct 2026 · d62f988acf · 7 files, +145 −78

Requested by finish-f1. Handed to the local harness by the cloud job pr-OpenVibe.Media-25; done by DeepSeek harness.

Changes

ddiff --git a/docs/media-fabric.md b/docs/media-fabric.md
iindex d6a8a17..6cc506c 100644
---- a/docs/media-fabric.md
++++ b/docs/media-fabric.md
@@@ -89,24 +89,28 @@ med_xyz/{captions/*.vtt, storyboard.webp+.vtt, waveform, metadata}
HTTP ranges.
- **F3.1, shipped: the timeline index and the source representation of finished video.** Migration
`0002_media_timeline.sql` adds `media_timeline` (object, rendition, segment number `seq` — 0 is the init segment —,
-- `start_ms`/`end_ms` (contiguous, half-open), `keyframe_ms`, the segment's `key` `<object>/<rendition>/<name>`,
-- `local_path`, `durable_provider`, `packed_object_id` + `byte_offset`/`byte_length` (packing is F3.3; NULL until then),
++ `start_ms`/`end_ms` (contiguous, half-open), `keyframe_ms`, the segment's `key` `<object>/<rendition>/<version>/<name>`
++ (`version` = the sha prefix of the bytes, so a re-cut writes beside the old segment, never over it), `local_path`,
++ `durable_provider`, `packed_object_id` + `byte_offset`/`byte_length` (packing is F3.3; NULL until then),
`sha256`, `durability` `local|durable`), keyed by (object, rendition, seq) for "segments in order" with an index on
(object, rendition, start_ms) for "the segment at t". `server/objects/timeline.js` is the store (`segmentAt(objectId,
rendition, tMs)`, `segments`, `replace`, `removeObject`, the playlist writers). The heavy-lane job **`object.cmaf`**
(`server/jobs/cmaf.js`) stream-copies a ready vod/clip/video object (`ffmpeg -c copy -f hls -hls_segment_type fmp4`,
`params.segment_seconds` 1–10, 4 by default, cut at the first source keyframe after it) into `init.mp4` +
-- `000001.m4s`…, stores each under `OBJECTS_PATH/.timeline/<app>/<object>/source/` and, with B2 configured, under the
-- key `<object>/source/<name>` on B2 (`durable` once B2 confirms the size), then writes all rows of the rendition in one
-- transaction. It never touches the source object; a rerun leaves rows whose sha256 matches as they are and uploads
-- only what is not durable yet; it honours the abort signal and the stream-copy budget and removes its work directory.
++ `000001.m4s`…, stores each under `OBJECTS_PATH/.timeline/<app>/<object>/source/<version>/` and, with B2 configured,
++ under the key `<object>/source/<version>/<name>` on B2 (`durable` once B2 confirms the size), then writes all rows of
++ the rendition in one transaction and only then deletes the previous cut's bytes. It never touches the source object; a
++ rerun reuses rows whose sha256 matches and their location as they are and uploads only what is not durable yet; it
++ honours the abort signal (which stops an upload in flight) and, before the rows commit, removes the files and keys it
++ staged; the stream-copy budget is honoured and its work directory removed.
Behind **`MEDIA_HLS_ENABLED`** (off by default: the job is refused with `409 media.hls.disabled` and the routes 404),
`GET /o/:id/master.m3u8`, `/o/:id/source/index.m3u8` and `/o/:id/source/{init.mp4,NNNNNN.m4s}` serve the timeline
under the same check as `GET /o/:id` (private and sandbox objects only with the object's `?exp&sig`, carried onto
every URI of a signed playlist; deleted 410). The playlists are written from the rows (`#EXT-X-PLAYLIST-TYPE:VOD`,
`#EXT-X-ENDLIST`, EXTINF = the row's duration, BANDWIDTH = peak segment bitrate); `GET …/download?format=json` adds
-- `hls_url` when a timeline exists. A purge, or a vod/clip deleted for good, removes the segments and rows; a held
-- object keeps them. Storage orphan reports count timeline files and keys as wanted.
++ `hls_url` when a timeline exists. A purge, or a vod/clip deleted for good, removes the segments and rows; a durable
++ copy that will not delete keeps its row so the next pass retries the key; a held object keeps them all. Storage orphan
++ reports count timeline files and keys as wanted.
**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, packing into ~60 s chunks (F3.3), virtual and materialized clips over the
timeline, sprites and captions on it, the faststart MP4 fallback as a derivative, segment-bucket demand, a finalize
ddiff --git a/docs/object-model.md b/docs/object-model.md
iindex 48fcd54..4a2dbab 100644
---- a/docs/object-model.md
++++ b/docs/object-model.md
@@@ -27,7 +27,7 @@ Code: `server/objects/` (`model.js`, `routes.js`, `backfill.js`, `reconcile.js`,
| `media_holds` | retention holds (with a staff `note`), see [Holds](#retention-holds) |
| `media_namespaces` | every namespace: `namespace`, `app_id` (tenant), `parent`, `owner` (`service:<app>` or `project:<prj_…>`), `policy` (JSON), `quota_bytes`, `quota_objects`, and the usage snapshot `used_bytes`, `used_objects`, `reserved_bytes`, `reserved_objects`, `reconciled_at`. See [Namespaces](#namespaces-grants-and-quotas) |
| `media_quota_reservations` | quota held by an upload in progress: `object_id`, `app_id`, `namespace`, `bytes`, `expires_at` |
--| `media_timeline` | the timeline index of segment-native video (F3.1, [media-fabric §3](media-fabric.md#3-video-is-segment-native-cmaf-under-hls)): one row per CMAF segment of one rendition (`object_id`, `rendition`, `seq`: 0 is `init.mp4`), `start_ms`/`end_ms`/`keyframe_ms`, `key` (`<object>/<rendition>/<name>`), `local_path`, `durable_provider`, `packed_object_id` + `byte_offset`/`byte_length`, `sha256`, `durability` (`local` `durable`). Written by `object.cmaf`; read by the HLS routes and `timeline.segmentAt` |
++| `media_timeline` | the timeline index of segment-native video (F3.1, [media-fabric §3](media-fabric.md#3-video-is-segment-native-cmaf-under-hls)): one row per CMAF segment of one rendition (`object_id`, `rendition`, `seq`: 0 is `init.mp4`), `start_ms`/`end_ms`/`keyframe_ms`, `key` (`<object>/<rendition>/<version>/<name>`, `version` = the sha prefix so a re-cut writes beside the old segment), `local_path`, `durable_provider`, `packed_object_id` + `byte_offset`/`byte_length`, `sha256`, `durability` (`local` `durable`). Written by `object.cmaf`; read by the HLS routes and `timeline.segmentAt` |
| `media_invariant_violations` | public playback objects over the size policy, see [Invariant](#public-object-size-invariant) |
Location `state`:
@@@ -759,7 +759,7 @@ job over only with the token it read and, past boot, only while the lease is sti
| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) |
| `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) |
| `storage.orphans.scan` | light | The [storage orphan report](#storage-orphan-report), service-wide, under the system app `_media`; report only. `params { limit (1-100000, 2000) }`. Result `{ totals, providers, summary, report }` (the report file under `<data>/reports/`). Scheduled every `MEDIA_ORPHAN_SCAN_DAYS`; tenants cannot enqueue it (403) |
--| `object.cmaf` | heavy | Segment-native video (F3.1; only with `MEDIA_HLS_ENABLED`, else `409 media.hls.disabled`): a stream copy of a ready vod/clip/video object into CMAF segments on its own keyframes (`init.mp4` + `NNNNNN.m4s`; `segment_seconds` 1–10, default 4: a segment ends at the first keyframe after it), stored under `OBJECTS_PATH/.timeline/<app>/<object>/source/` and, with B2 configured, as `<object>/source/<name>` on B2; one `media_timeline` row per segment (times, size, sha256, `durability`), all written in one transaction. The source is untouched. A rerun keeps rows whose sha256 matches and uploads only what is not durable; a failed upload leaves its row `local` and retries the job. Result `{ source_id, rendition, segments, duration_ms, bytes, durable, local_only, rows: { inserted, updated, unchanged, removed }, uploaded, master_path }`. Served at `/o/:id/master.m3u8` (see [media-fabric §3](media-fabric.md#3-video-is-segment-native-cmaf-under-hls)) |
++| `object.cmaf` | heavy | Segment-native video (F3.1; only with `MEDIA_HLS_ENABLED`, else `409 media.hls.disabled`): a stream copy of a ready vod/clip/video object into CMAF segments on its own keyframes (`init.mp4` + `NNNNNN.m4s`; `segment_seconds` 1–10, default 4: a segment ends at the first keyframe after it), stored under `OBJECTS_PATH/.timeline/<app>/<object>/source/<version>/` and, with B2 configured, as `<object>/source/<version>/<name>` on B2 (`version` = the sha prefix: a re-cut writes beside the old bytes); one `media_timeline` row per segment (times, size, sha256, `durability`), all written in one transaction before the old bytes are deleted. The source is untouched. A rerun reuses rows whose sha256 matches and uploads only what is not durable; a failed upload leaves its row `local` and retries the job; an abort removes staged files and keys. Result `{ source_id, rendition, segments, duration_ms, bytes, durable, local_only, rows: { inserted, updated, unchanged, removed }, uploaded, master_path }`. Served at `/o/:id/master.m3u8` (see [media-fabric §3](media-fabric.md#3-video-is-segment-native-cmaf-under-hls)) |
| `storage.move.cleanup` | light | Placement moves whose delete keeps failing: an object with three `failed` rows in `media_object_tier_decisions` in the last 7 days (a later `done` resets them) raises one `storage.alert` of kind `move_cleanup_failed`. Service-wide under the system app `_media`; report only. `params {}`. Result `{ scanned, alerted, objects }`. Scheduled every `MEDIA_MOVE_CLEANUP_MINUTES` (60; `0` = never); tenants cannot enqueue it (403) |
| `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |
ddiff --git a/server/jobs/cmaf.js b/server/jobs/cmaf.js
iindex 4325d25..63f02af 100644
---- a/server/jobs/cmaf.js
++++ b/server/jobs/cmaf.js
@@@ -12,10 +12,12 @@
* rows: { inserted, updated, unchanged, removed }, uploaded, master_path }
*
* Bytes: each segment lands on this node's disk (OBJECTS_PATH/.timeline/…); with B2 configured it is uploaded under
-- * <object>/source/<name> and its row says `durable` once B2 confirms the size. Idempotent: a rerun makes the same
-- * bytes (bitexact), leaves a row whose sha256 matches as it is, and uploads only what is not durable yet. The rows of
-- * a rendition are written in one transaction after every segment is stored, so a half-finished run leaves no
-- * timeline. An upload that fails leaves its row `local` and fails the attempt, so the retry finishes it.
++ * <object>/source/<version>/<name> (version = the sha's prefix) and its row says `durable` once B2 confirms the size.
++ * Idempotent: a rerun makes the same bytes (bitexact), reuses a row whose sha256 matches and its location as it is, and
++ * uploads only what is not durable yet. Two-phase: a re-cut writes beside the published segments and the rows commit in
++ * one transaction before the old bytes go, so playback never reads bytes that disagree with the playlist; a run that
++ * aborts or fails before the commit removes its staged files and keys. An upload that fails leaves its row `local` and
++ * fails the attempt, so the retry finishes it.
* Refused: MEDIA_HLS_ENABLED off, a source that is not ready or not media (permanent), less free disk than the source
* size plus MEDIA_UPLOAD_MIN_FREE_MB (retried later), a codec fMP4 cannot carry (permanent).
*/
@@@ -113,54 +115,78 @@ async function run(job, ctx) {
const rows = [];
let uploaded = 0;
const failed = [];
-- for (const f of files) {
++ // Bytes written before the new rows commit are staged: an abort or a failure before the commit removes them
++ // again, so a cancelled run leaves neither untracked keys nor files.
++ const stagedRemote = [];
++ const stagedLocal = [];
++ let committed = false;
++ try {
++ for (const f of files) {
++ aborted(ctx);
++ const tmp = path.join(out, f.file || f.name);
++ if (!derive.existing(tmp)) throw new JobError('ffmpeg_failed', `ffmpeg listed ${f.file || f.name} but did not write it`);
++ const size = fs.statSync(tmp).size;
++ const sha = await derive.sha256File(tmp);
++ const old = prev.get(f.seq);
++ const match = !!old && old.sha256 === sha;
++ // Content-addressed: the same bytes always land at the same versioned location, so a rerun rewrites
++ // nothing and a re-cut writes beside the old segment until the new rows are published.
++ const version = sha.slice(0, 12);
++ const dest = timeline.localPathFor(src, rendition, f.name, version);
++ const key = timeline.keyFor(src.id, rendition, f.name, version);
++ let localPath = dest;
++ let durable = match && old.durability === 'durable' ? old.durable_provider : null;
++ if (match && old.local_path && derive.existing(old.local_path)) {
++ localPath = old.local_path; // the previous copy is exactly these bytes: keep it and its location
++ } else {
++ fs.mkdirSync(path.dirname(dest), { recursive: true });
++ fs.renameSync(tmp, dest);
++ stagedLocal.push(dest);
++ }
++ if (!durable && durableOn) {
++ try {
++ if (!vodStorage.providerAvailable(DURABLE)) throw new Error(`${DURABLE} is unavailable`);
++ await vodStorage.uploadFile(DURABLE, key, localPath, f.seq === 0 ? 'video/mp4' : 'video/iso.segment', { signal: ctx.signal });
++ stagedRemote.push({ provider: DURABLE, key });
++ durable = DURABLE;
++ uploaded++;
++ } catch (err) {
++ if (ctx.signal.aborted) throw ctx.signal.reason || err; // a cancelled upload is not "not durable yet"
++ failed.push(`${f.name}: ${err.message}`);
++ }
++ }
++ rows.push({
++ seq: f.seq, name: f.name, start_ms: f.start_ms, end_ms: f.end_ms, keyframe_ms: f.keyframe_ms, key, local_path: localPath,
++ durable_provider: durable, packed_object_id: null, byte_offset: null, byte_length: size, sha256: sha,
++ durability: durable ? 'durable' : 'local',
++ });
++ }
aborted(ctx);
-- const tmp = path.join(out, f.file || f.name);
-- if (!derive.existing(tmp)) throw new JobError('ffmpeg_failed', `ffmpeg listed ${f.file || f.name} but did not write it`);
-- const size = fs.statSync(tmp).size;
-- const sha = await derive.sha256File(tmp);
-- const old = prev.get(f.seq);
-- const match = !!old && old.sha256 === sha;
-- const dest = timeline.localPathFor(src, rendition, f.name);
-- if (!(match && derive.existing(dest))) {
-- fs.mkdirSync(path.dirname(dest), { recursive: true });
-- fs.renameSync(tmp, dest);
++ const written = await timeline.replace(src.id, rendition, rows, { jobId: job.id });
++ committed = true;
++ // Phase two, only once the new rows are published: the bytes a previous cut named that this one does not.
++ // 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);
++ 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 (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 });
}
-- const key = timeline.keyFor(src.id, rendition, f.name);
-- let durable = match && old.durability === 'durable' ? old.durable_provider : null;
-- if (!durable && durableOn) {
-- try {
-- if (!vodStorage.providerAvailable(DURABLE)) throw new Error(`${DURABLE} is unavailable`);
-- await vodStorage.uploadFile(DURABLE, key, dest, f.seq === 0 ? 'video/mp4' : 'video/iso.segment');
-- durable = DURABLE;
-- uploaded++;
-- } catch (err) { failed.push(`${f.name}: ${err.message}`); }
++ const segs = rows.filter((x) => x.seq > 0);
++ return {
++ source_id: src.id, rendition, segments: segs.length, duration_ms: segs[segs.length - 1].end_ms,
++ bytes: rows.reduce((a, x) => a + x.byte_length, 0),
++ durable: rows.filter((x) => x.durability === 'durable').length, local_only: rows.filter((x) => x.durability === 'local').length,
++ rows: written, uploaded, master_path: `/o/${src.id}/master.m3u8`, source_unchanged: true,
++ };
++ } catch (err) {
++ if (!committed) {
++ for (const k of stagedRemote) await vodStorage.deleteObject(k.provider, k.key).catch(() => { /* best effort */ });
++ for (const p of stagedLocal) { try { fs.unlinkSync(p); } catch { /* already gone */ } }
}
-- rows.push({
-- seq: f.seq, name: f.name, start_ms: f.start_ms, end_ms: f.end_ms, keyframe_ms: f.keyframe_ms, key, local_path: dest,
-- durable_provider: durable, packed_object_id: null, byte_offset: null, byte_length: size, sha256: sha,
-- durability: durable ? 'durable' : 'local',
-- });
-- }
-- aborted(ctx);
-- // Segments past the new end (an older run cut the source differently) go with their rows.
-- const last = files[files.length - 1].seq;
-- for (const [seq, old] of prev) {
-- if (seq <= last) continue;
-- if (old.local_path && path.resolve(old.local_path).startsWith(timeline.localRoot() + path.sep)) { try { fs.unlinkSync(old.local_path); } catch { /* gone */ } }
-- if (old.durable_provider && vodStorage.providerConfigured(old.durable_provider)) await vodStorage.deleteObject(old.durable_provider, old.key).catch(() => {});
-- }
-- const written = await timeline.replace(src.id, rendition, rows, { jobId: job.id });
-- 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 err;
}
-- const segs = rows.filter((x) => x.seq > 0);
-- return {
-- source_id: src.id, rendition, segments: segs.length, duration_ms: segs[segs.length - 1].end_ms,
-- bytes: rows.reduce((a, x) => a + x.byte_length, 0),
-- durable: rows.filter((x) => x.durability === 'durable').length, local_only: rows.filter((x) => x.durability === 'local').length,
-- rows: written, uploaded, master_path: `/o/${src.id}/master.m3u8`, source_unchanged: true,
-- };
} finally {
derive.cleanupWork(job.id);
}
ddiff --git a/server/objects/model.js b/server/objects/model.js
iindex b2a2959..45bbfbe 100644
---- a/server/objects/model.js
++++ b/server/objects/model.js
@@@ -675,7 +675,10 @@ async function purgeExpired({ retentionDays = config.objects.retentionDays } = {
if (l.provider === 'local' && path.resolve(l.key).startsWith(root)) { try { fs.unlinkSync(l.key); } catch { /* already gone */ } }
if (l.provider === 'local') await db.run('DELETE FROM media_locations WHERE id = ?', [l.id]);
}
-- await require('./timeline').removeObject(obj.id); // its CMAF segments (F3.1) go with the bytes
++ // Its CMAF segments (F3.1) go with the bytes. A durable copy whose delete failed keeps its row, so this purge
++ // is not marked done and the next pass retries it rather than losing the key.
++ const tl = await require('./timeline').removeObject(obj.id);
++ if (tl.pending) continue;
md.purged_at = new Date().toISOString();
await db.run('UPDATE media_objects SET metadata = ?, updated_at = ov_now() WHERE id = ?', [JSON.stringify(md), obj.id]);
purged++;
ddiff --git a/server/objects/timeline.js b/server/objects/timeline.js
iindex c91f957..c077765 100644
---- a/server/objects/timeline.js
++++ b/server/objects/timeline.js
@@@ -5,9 +5,11 @@
* "the bytes for 01:47:12.350" (segmentAt) and "the segments of a rendition in order" (segments) without
* opening a file, and writes the playlists from the rows. Segment 0 of a rendition is its init segment.
*
-- * Bytes: <object>/<rendition>/<name> is the segment's key; a copy on this node's disk sits at
-- * OBJECTS_PATH/.timeline/<app>/<object>/<rendition>/<name> (local_path), and a durable one under the same key on
-- * durable_provider. The object.cmaf job (server/jobs/cmaf.js) writes the source rendition; the public routes
++ * Bytes: <object>/<rendition>/<version>/<name> is the segment's key, `version` the sha prefix of the bytes, so a re-cut
++ * writes beside the old one instead of over it; a copy on this node's disk sits at
++ * OBJECTS_PATH/.timeline/<app>/<object>/<rendition>/<version>/<name> (local_path), and a durable one under the same key
++ * on durable_provider. The name alone is the playlist URI, so a row carries the versioned location it serves from. The
++ * object.cmaf job (server/jobs/cmaf.js) writes the source rendition; the public routes
* /o/:id/master.m3u8 and /o/:id/source/… (objects/routes.js) serve it behind MEDIA_HLS_ENABLED.
*/
'use strict';
@@@ -22,8 +24,11 @@ const INIT_NAME = 'init.mp4';
const FIELDS = ['name', 'start_ms', 'end_ms', 'keyframe_ms', 'key', 'local_path', 'durable_provider', 'packed_object_id', 'byte_offset', 'byte_length', 'sha256', 'durability'];
function localRoot() { return path.join(path.resolve(config.objects.path), '.timeline'); }
--function localPathFor(obj, rendition, name) { return path.join(localRoot(), obj.app_id, obj.id, rendition, name); }
--function keyFor(objectId, rendition, name) { return `${objectId}/${rendition}/${name}`; }
++/** `version` names the immutable set of bytes (the sha's prefix): a re-cut lands beside the old one instead of on it. */
++function localPathFor(obj, rendition, name, version = null) {
++ return path.join(localRoot(), obj.app_id, obj.id, rendition, ...(version ? [version] : []), name);
++}
++function keyFor(objectId, rendition, name, version = null) { return `${objectId}/${rendition}/${version ? `${version}/` : ''}${name}`; }
function segmentName(seq) { return seq === 0 ? INIT_NAME : `${String(seq).padStart(6, '0')}.m4s`; }
/** Every row of a rendition in order, the init segment (seq 0) first. */
@@@ -82,29 +87,51 @@ async function replace(objectId, rendition, rows, { jobId = null } = {}) {
}
/**
-- * The object's whole timeline is gone with its bytes (a purge, or a vod/clip deleted for good): local files, durable
-- * copies (best effort), then the rows. Callers have already refused a held object. → rows removed.
++ * 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 }] }.
*/
--async function removeObject(objectId) {
-- if (!objectId) return 0;
-- const rows = await db.all('SELECT * FROM media_timeline WHERE object_id = ?', [objectId]);
-- if (!rows.length) return 0;
++async function deleteBytes(rows) {
const root = localRoot() + path.sep;
const vodStorage = require('../vod/vod-storage');
const dirs = new Set();
++ const failed = [];
for (const r of rows) {
if (r.local_path && path.resolve(r.local_path).startsWith(root)) {
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)) {
-- await vodStorage.deleteObject(r.durable_provider, r.key).catch((err) => console.warn(`[Timeline] ${r.durable_provider}:${r.key} not deleted: ${err.message}`));
++ 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 (await db.run('DELETE FROM media_timeline WHERE object_id = ?', [objectId])).changes || 0;
++ return { failed };
++}
++
++/**
++ * 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).
++ */
++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 };
++ const { failed } = await deleteBytes(rows);
++ const stuck = new Set(failed.map((f) => `${f.row.rendition}\n${f.row.seq}`));
++ let removed = 0;
++ let pending = 0;
++ for (const r of rows) {
++ if (stuck.has(`${r.rendition}\n${r.seq}`)) { 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 (pending) console.warn(`[Timeline] ${objectId}: ${pending} segment row(s) kept — their durable copy could not be deleted`);
++ return { removed, pending };
}
// ── Playlists (from the rows, never from files) ──────────────
@@@ -144,5 +171,5 @@ function masterPlaylist(renditions, { query = '' } = {}) {
module.exports = {
SOURCE, INIT_NAME, localRoot, localPathFor, keyFor, segmentName,
-- list, segments, byName, has, segmentAt, replace, removeObject, mediaPlaylist, masterPlaylist,
++ list, segments, byName, has, segmentAt, replace, deleteBytes, removeObject, mediaPlaylist, masterPlaylist,
};
ddiff --git a/server/vod/vod-storage.js b/server/vod/vod-storage.js
iindex f39d6db..5f8171c 100644
---- a/server/vod/vod-storage.js
++++ b/server/vod/vod-storage.js
@@@ -305,7 +305,8 @@ function uploadTimeoutMs(bytes, settings = getSettings()) {
return floor + Math.ceil((Number(bytes) || 0) / (mbps * 1024 * 1024)) * 1000;
}
--async function uploadFile(provider, key, filePath, contentType = 'video/webm') {
++async function uploadFile(provider, key, filePath, contentType = 'video/webm', { signal = null } = {}) {
++ if (signal && signal.aborted) throw (signal.reason instanceof Error ? signal.reason : new Error('upload aborted'));
const client = clientFor(provider);
if (!client) throw new Error(`Provider ${provider} not configured`);
loadSdk();
@@@ -318,6 +319,9 @@ async function uploadFile(provider, key, filePath, contentType = 'video/webm') {
// could drain). Abort past the deadline so the sweep moves on to the next VOD;
// leavePartsOnError cleans the partial multipart up on the bucket.
const abort = new AbortController();
++ // A cancelled job (the caller's signal) must stop an upload already in flight, not only the next one.
++ const onAbort = () => abort.abort(signal.reason instanceof Error ? signal.reason : new Error('upload aborted'));
++ if (signal) signal.addEventListener('abort', onAbort, { once: true });
const deadline = setTimeout(() => abort.abort(new Error(`upload deadline exceeded (${Math.round(uploadTimeoutMs(size) / 60000)} min for ${(size / 1048576).toFixed(0)} MB)`)), uploadTimeoutMs(size));
const upload = new LibStorage.Upload({
client,
@@@ -339,6 +343,7 @@ async function uploadFile(provider, key, filePath, contentType = 'video/webm') {
throw err;
} finally {
clearTimeout(deadline);
++ if (signal) signal.removeEventListener('abort', onAbort);
}
const deletedSize = recentlyDeleted.get(`${provider}:${key}`);
if (before || deletedSize != null) {
@@@ -411,9 +416,10 @@ function rememberDeleted(id, size) {
setTimeout(() => { if (recentlyDeleted.get(id) === size) recentlyDeleted.delete(id); }, 10 * 60e3).unref?.();
}
++/** → true once the key is gone, false when there is no client or the delete failed (callers keep the key to retry). */
async function deleteObject(provider, key) {
const client = clientFor(provider);
-- if (!client) return;
++ if (!client) return false;
// A shielded provider's cached slices leave the shield with the object (placement/shield.js): the size decides
// how many slices to refresh, so it is read before the delete.
const shield = require('../placement/shield');
@@@ -422,12 +428,13 @@ async function deleteObject(provider, key) {
await client.send(new S3.DeleteObjectCommand({ Bucket: PROVIDER_ENV[provider].bucket, Key: key }));
} catch (err) {
console.warn(`[VodStorage] Delete ${provider}:${key} failed:`, err.message);
-- return;
++ return false;
}
if (head) rememberDeleted(`${provider}:${key}`, head.size);
if (head) shield.purge({ provider, key, size: head.size, presign: presignGet })
.then((r) => { if (r.failed) console.warn(`[VodStorage] shield purge ${provider}:${key}: ${r.failed} of ${r.slices} slices not refreshed`); })
.catch(() => { /* purge never throws */ });
++ return true;
}
/**
ddiff --git a/test/media-timeline.test.js b/test/media-timeline.test.js
iindex a137a69..9b7d06a 100644
---- a/test/media-timeline.test.js
++++ b/test/media-timeline.test.js
@@@ -60,7 +60,7 @@ const { spawn, spawnSync } = require('child_process');
assert.ok(mp.includes('#EXT-X-TARGETDURATION:3\n') && mp.includes('#EXT-X-MAP:URI="init.mp4"') && mp.includes('#EXTINF:1.932,\n000003.m4s') && mp.trim().endsWith('#EXT-X-ENDLIST'), mp);
assert.ok(timeline.masterPlaylist([{ name: 'source', rows: hand }]).includes('#EXT-X-STREAM-INF:BANDWIDTH=80000,AVERAGE-BANDWIDTH='), 'peak = 30000 B over 3 s');
assert.deepStrictEqual(await timeline.replace('med_HAND', 'source', hand.slice(0, 2)), { inserted: 0, updated: 0, unchanged: 2, removed: 2 }, 'rows past a shorter end are dropped');
-- assert.strictEqual(await timeline.removeObject('med_HAND'), 2);
++ assert.deepStrictEqual(await timeline.removeObject('med_HAND'), { removed: 2, pending: 0 });
if (!hasFfmpeg) { console.log('media timeline: skipped (ffmpeg not found)'); process.exit(0); }
await new Promise((r) => stub.listen(0, '127.0.0.1', r));
@@@ -131,7 +131,7 @@ const { spawn, spawnSync } = require('child_process');
const bytes = fs.readFileSync(r.local_path);
assert.strictEqual(sha(bytes), r.sha256, `${r.name} sha256`);
assert.strictEqual(bytes.length, Number(r.byte_length));
-- assert.strictEqual(r.key, `${pub}/source/${r.name}`);
++ assert.strictEqual(r.key, `${pub}/source/${r.sha256.slice(0, 12)}/${r.name}`, 'the key is versioned by the sha');
assert.strictEqual(r.durability, 'local');
assert.ok(r.local_path.startsWith(path.join(path.resolve(process.env.OBJECTS_PATH), '.timeline', 'live', pub, 'source')));
}