openvibers/openvibe.media Commits
Retry cmaf with no healthy provider and guard races with pack
Changes
@@@ -114,6 +114,8 @@ async function run(job, ctx) {const prev = new Map((await timeline.list(src.id, rendition)).map((x) => [Number(x.seq), x]));const vodStorage = require('../vod/vod-storage');const target = await timeline.durableProvider(src);++ // B2/R2 configured but none healthy (breakers open) is "not durable yet", retried later, never a quiet success.++ const remoteConfigured = vodStorage.REMOTE_PROVIDERS.some((p) => vodStorage.providerConfigured(p));const rows = [];let uploaded = 0;const failed = [];@@@ -150,6 +152,7 @@ async function run(job, ctx) {fs.renameSync(tmp, dest);stagedLocal.push(dest);}++ if (!durable && !target && remoteConfigured) failed.push(`${f.name}: no healthy durable provider`);if (!durable && target) {try {if (!vodStorage.providerAvailable(target)) throw new Error(`${target} is unavailable`);@@@ -169,7 +172,14 @@ async function run(job, ctx) {});}aborted(ctx);-- const written = await timeline.replace(src.id, rendition, rows, { jobId: job.id });++ let written;++ try {++ written = await timeline.replace(src.id, rendition, rows, { jobId: job.id, expect: prev });++ } catch (err) {++ // An object.pack committed after `prev` was read: these rows would name the segment keys it deleted.++ if (err instanceof timeline.TimelineChanged) throw new JobError('media.timeline.changed', 'The timeline changed while cutting (a pack); a retry cuts against the new rows', { retryAfterS: 60 });++ throw err;++ }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.@@@ -189,8 +199,14 @@ async function run(job, ctx) {};} 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 */ } }++ // Content-addressed: another run may have committed these very keys and files, so only unnamed ones go.++ for (const k of stagedRemote) {++ if (!await timeline.isNamed({ provider: k.provider, key: k.key }).catch(() => true)) await vodStorage.deleteObject(k.provider, k.key).catch(() => { /* best effort */ });++ }++ for (const p of stagedLocal) {++ if (await timeline.isNamed({ localPath: p }).catch(() => true)) continue;++ try { fs.unlinkSync(p); } catch { /* already gone */ }++ }}throw err;}@@@ -176,8 +176,9 @@ async function run(job, ctx) {if (!committed) throw new JobError('media.timeline.changed', 'The timeline changed while packing (a re-cut); a retry packs the new segments', { retryAfterS: 60 });} finally {if (!committed) {-- if (uploaded) await vodStorage.deleteObject(provider, key).catch(() => { /* best effort: the orphan report names it */ });-- if (localPath) { try { fs.unlinkSync(localPath); } catch { /* not moved */ } }++ // Content-addressed: a concurrent pack of the same run may have committed this very chunk; never delete it.++ if (uploaded && !await timeline.isNamed({ provider, key }).catch(() => true)) await vodStorage.deleteObject(provider, key).catch(() => { /* best effort: the orphan report names it */ });++ if (localPath && !await timeline.isNamed({ localPath }).catch(() => true)) { try { fs.unlinkSync(localPath); } catch { /* not moved */ } }try { fs.unlinkSync(tmp); } catch { /* moved or gone */ }}}@@@ -741,6 +741,7 @@ const publicRouter = express.Router();// A segment's copy is the placement router's choice; a packed one (F3.3) is a ranged read of its ~60 s chunk.const HLS_TYPE = 'application/vnd.apple.mpegurl';const SEGMENT_NAME = /^(init\.mp4|\d{6,}\.m4s)$/;++const SLICE_FETCH_MS = 15000; // a ranged read of one packed segment from B2/R2/** The object of an HLS request, or null when answered (or passed on to the 404 while the flag is off). */async function hlsObject(req, res, next) {@@@ -810,8 +811,20 @@ async function sendSlice(req, res, { file, url, offset, length, headers }) {stream.on('error', () => res.destroy());return stream.pipe(res);}-- const up = await fetch(url, { headers: { range: `bytes=${offset + start}-${offset + end}` } }).catch(() => null);-- const body = up && up.status === 206 ? Buffer.from(await up.arrayBuffer()) : null;++ // Bounded in time and dropped when the viewer goes away; any failure, mid-body too, moves on to the next copy.++ const ac = new AbortController();++ const gone = () => ac.abort();++ req.on('close', gone);++ let body = null;++ try {++ const up = await fetch(url, { headers: { range: `bytes=${offset + start}-${offset + end}` }, signal: AbortSignal.any([ac.signal, AbortSignal.timeout(SLICE_FETCH_MS)]) });++ body = up.status === 206 ? Buffer.from(await up.arrayBuffer()) : null;++ } catch {++ body = null;++ } finally {++ req.off('close', gone);++ }++ if (res.destroyed) return undefined; // the viewer left: nothing more to try for themif (!body || body.length !== end - start + 1) return false;res.writeHead(r ? 206 : 200, head);return res.end(body);@@@ -67,14 +67,30 @@ function same(a, b) {return FIELDS.every((f) => (a[f] == null ? null : String(a[f])) === (b[f] == null ? null : String(b[f])));}++/** Inside a transaction: replace() and markPacked() of one object take turns, so each sees the other's commit. */++async function lockObject(objectId) {++ await db.get('SELECT pg_advisory_xact_lock(hashtext(?)) AS locked', [`media.timeline:${objectId}`]);++}++++/** Thrown by replace() when the rows changed since the caller read them (`expect`). */++class TimelineChanged extends Error {}++/*** Make the rows of one rendition exactly `rows` (each { seq, name, start_ms, … }), in one transaction: a row that already* says the same is not written, a changed one is updated, rows past the new end are dropped. → { inserted, updated, unchanged, removed }.++ * `expect` (the rows the caller read, by seq) makes it compare-and-set: if any row's location or packing changed meanwhile++ * (an object.pack committed), nothing is written and TimelineChanged is thrown.*/--async function replace(objectId, rendition, rows, { jobId = null } = {}) {++async function replace(objectId, rendition, rows, { jobId = null, expect = null } = {}) {const out = { inserted: 0, updated: 0, unchanged: 0, removed: 0 };await db.getDb().tx(async () => {++ await lockObject(objectId);const before = new Map((await list(objectId, rendition)).map((r) => [Number(r.seq), r]));++ if (expect) {++ const at = (r) => (r ? ['key', 'local_path', 'durable_provider', 'packed_object_id', 'sha256'].map((f) => (r[f] == null ? '' : String(r[f]))).join('\n') : null);++ const seqs = new Set([...before.keys(), ...expect.keys()]);++ for (const q of seqs) if (at(before.get(q)) !== at(expect.get(q))) throw new TimelineChanged(`timeline of ${objectId}/${rendition} changed (seq ${q})`);++ }for (const r of rows) {const prev = before.get(r.seq);if (prev && same(prev, r)) { out.unchanged++; continue; }@@@ -120,6 +136,17 @@ async function deleteBytes(rows) {return { failed };}++/**++ * 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.++ */++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;++ 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@@@ -153,6 +180,7 @@ async function markPacked(objectId, rendition, updates, { jobId = null } = {}) {const CONFLICT = new Error('timeline changed');try {await db.getDb().tx(async () => {++ await lockObject(objectId);for (const u of updates) {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()@@@ -229,6 +257,6 @@ function masterPlaylist(renditions, { query = '' } = {}) {module.exports = {SOURCE, INIT_NAME, FIELDS, localRoot, localPathFor, keyFor, segmentName,-- list, segments, byName, has, segmentAt, replace, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,++ list, segments, byName, has, segmentAt, replace, TimelineChanged, isNamed, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,mediaPlaylist, masterPlaylist,};
@@ -114,6 +114,8 @@ async function run(job, ctx) {
@@ -114,6 +114,8 @@ async function run(job, ctx) {
const prev = new Map((await timeline.list(src.id, rendition)).map((x) => [Number(x.seq), x]));
const prev = new Map((await timeline.list(src.id, rendition)).map((x) => [Number(x.seq), x]));
const vodStorage = require('../vod/vod-storage');
const vodStorage = require('../vod/vod-storage');
const target = await timeline.durableProvider(src);
const target = await timeline.durableProvider(src);
+ // B2/R2 configured but none healthy (breakers open) is "not durable yet", retried later, never a quiet success.
+ const remoteConfigured = vodStorage.REMOTE_PROVIDERS.some((p) => vodStorage.providerConfigured(p));
const rows = [];
const rows = [];
let uploaded = 0;
let uploaded = 0;
const failed = [];
const failed = [];
@@ -150,6 +152,7 @@ async function run(job, ctx) {
@@ -150,6 +152,7 @@ async function run(job, ctx) {
fs.renameSync(tmp, dest);
fs.renameSync(tmp, dest);
stagedLocal.push(dest);
stagedLocal.push(dest);
}
}
+ if (!durable && !target && remoteConfigured) failed.push(`${f.name}: no healthy durable provider`);
if (!durable && target) {
if (!durable && target) {
try {
try {
if (!vodStorage.providerAvailable(target)) throw new Error(`${target} is unavailable`);
if (!vodStorage.providerAvailable(target)) throw new Error(`${target} is unavailable`);
@@ -169,7 +172,14 @@ async function run(job, ctx) {
@@ -169,7 +172,14 @@ async function run(job, ctx) {
});
});
}
}
aborted(ctx);
aborted(ctx);
- const written = await timeline.replace(src.id, rendition, rows, { jobId: job.id });
+ let written;
+ try {
+ written = await timeline.replace(src.id, rendition, rows, { jobId: job.id, expect: prev });
+ } catch (err) {
+ // An object.pack committed after `prev` was read: these rows would name the segment keys it deleted.
+ if (err instanceof timeline.TimelineChanged) throw new JobError('media.timeline.changed', 'The timeline changed while cutting (a pack); a retry cuts against the new rows', { retryAfterS: 60 });
+ throw err;
+ }
committed = true;
committed = true;
// Phase two, only once the new rows are published: the bytes a previous cut named that this one does not.
// 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.
// A deletion that fails leaves an orphan the storage report names, never bytes a live row still points at.
@@ -189,8 +199,14 @@ async function run(job, ctx) {
@@ -189,8 +199,14 @@ async function run(job, ctx) {
};
};
} catch (err) {
} catch (err) {
if (!committed) {
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 */ } }
+ // Content-addressed: another run may have committed these very keys and files, so only unnamed ones go.
+ for (const k of stagedRemote) {
+ if (!await timeline.isNamed({ provider: k.provider, key: k.key }).catch(() => true)) await vodStorage.deleteObject(k.provider, k.key).catch(() => { /* best effort */ });
+ }
+ for (const p of stagedLocal) {
+ if (await timeline.isNamed({ localPath: p }).catch(() => true)) continue;
+ try { fs.unlinkSync(p); } catch { /* already gone */ }
+ }
}
}
throw err;
throw err;
}
}
@@ -176,8 +176,9 @@ async function run(job, ctx) {
@@ -176,8 +176,9 @@ async function run(job, ctx) {
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 });
} finally {
} finally {
if (!committed) {
if (!committed) {
- if (uploaded) await vodStorage.deleteObject(provider, key).catch(() => { /* best effort: the orphan report names it */ });
- if (localPath) { try { fs.unlinkSync(localPath); } catch { /* not moved */ } }
+ // Content-addressed: a concurrent pack of the same run may have committed this very chunk; never delete it.
+ if (uploaded && !await timeline.isNamed({ provider, key }).catch(() => true)) await vodStorage.deleteObject(provider, key).catch(() => { /* best effort: the orphan report names it */ });
+ if (localPath && !await timeline.isNamed({ localPath }).catch(() => true)) { try { fs.unlinkSync(localPath); } catch { /* not moved */ } }
try { fs.unlinkSync(tmp); } catch { /* moved or gone */ }
try { fs.unlinkSync(tmp); } catch { /* moved or gone */ }
}
}
}
}
@@ -741,6 +741,7 @@ const publicRouter = express.Router();
@@ -741,6 +741,7 @@ const publicRouter = express.Router();
// A segment's copy is the placement router's choice; a packed one (F3.3) is a ranged read of its ~60 s chunk.
// A segment's copy is the placement router's choice; a packed one (F3.3) is a ranged read of its ~60 s chunk.
const HLS_TYPE = 'application/vnd.apple.mpegurl';
const HLS_TYPE = 'application/vnd.apple.mpegurl';
const SEGMENT_NAME = /^(init\.mp4|\d{6,}\.m4s)$/;
const SEGMENT_NAME = /^(init\.mp4|\d{6,}\.m4s)$/;
+const SLICE_FETCH_MS = 15000; // a ranged read of one packed segment from B2/R2
/** The object of an HLS request, or null when answered (or passed on to the 404 while the flag is off). */
/** The object of an HLS request, or null when answered (or passed on to the 404 while the flag is off). */
async function hlsObject(req, res, next) {
async function hlsObject(req, res, next) {
@@ -810,8 +811,20 @@ async function sendSlice(req, res, { file, url, offset, length, headers }) {
@@ -810,8 +811,20 @@ async function sendSlice(req, res, { file, url, offset, length, headers }) {
stream.on('error', () => res.destroy());
stream.on('error', () => res.destroy());
return stream.pipe(res);
return stream.pipe(res);
}
}
- const up = await fetch(url, { headers: { range: `bytes=${offset + start}-${offset + end}` } }).catch(() => null);
- const body = up && up.status === 206 ? Buffer.from(await up.arrayBuffer()) : null;
+ // Bounded in time and dropped when the viewer goes away; any failure, mid-body too, moves on to the next copy.
+ const ac = new AbortController();
+ const gone = () => ac.abort();
+ req.on('close', gone);
+ let body = null;
+ try {
+ const up = await fetch(url, { headers: { range: `bytes=${offset + start}-${offset + end}` }, signal: AbortSignal.any([ac.signal, AbortSignal.timeout(SLICE_FETCH_MS)]) });
+ body = up.status === 206 ? Buffer.from(await up.arrayBuffer()) : null;
+ } catch {
+ body = null;
+ } finally {
+ req.off('close', gone);
+ }
+ if (res.destroyed) return undefined; // the viewer left: nothing more to try for them
if (!body || body.length !== end - start + 1) return false;
if (!body || body.length !== end - start + 1) return false;
res.writeHead(r ? 206 : 200, head);
res.writeHead(r ? 206 : 200, head);
return res.end(body);
return res.end(body);
@@ -67,14 +67,30 @@ function same(a, b) {
@@ -67,14 +67,30 @@ function same(a, b) {
return FIELDS.every((f) => (a[f] == null ? null : String(a[f])) === (b[f] == null ? null : String(b[f])));
return FIELDS.every((f) => (a[f] == null ? null : String(a[f])) === (b[f] == null ? null : String(b[f])));
}
}
+/** Inside a transaction: replace() and markPacked() of one object take turns, so each sees the other's commit. */
+async function lockObject(objectId) {
+ await db.get('SELECT pg_advisory_xact_lock(hashtext(?)) AS locked', [`media.timeline:${objectId}`]);
+}
+
+/** Thrown by replace() when the rows changed since the caller read them (`expect`). */
+class TimelineChanged extends Error {}
+
/**
/**
* Make the rows of one rendition exactly `rows` (each { seq, name, start_ms, … }), in one transaction: a row that already
* Make the rows of one rendition exactly `rows` (each { seq, name, start_ms, … }), in one transaction: a row that already
* says the same is not written, a changed one is updated, rows past the new end are dropped. → { inserted, updated, unchanged, removed }.
* says the same is not written, a changed one is updated, rows past the new end are dropped. → { inserted, updated, unchanged, removed }.
+ * `expect` (the rows the caller read, by seq) makes it compare-and-set: if any row's location or packing changed meanwhile
+ * (an object.pack committed), nothing is written and TimelineChanged is thrown.
*/
*/
-async function replace(objectId, rendition, rows, { jobId = null } = {}) {
+async function replace(objectId, rendition, rows, { jobId = null, expect = null } = {}) {
const out = { inserted: 0, updated: 0, unchanged: 0, removed: 0 };
const out = { inserted: 0, updated: 0, unchanged: 0, removed: 0 };
await db.getDb().tx(async () => {
await db.getDb().tx(async () => {
+ await lockObject(objectId);
const before = new Map((await list(objectId, rendition)).map((r) => [Number(r.seq), r]));
const before = new Map((await list(objectId, rendition)).map((r) => [Number(r.seq), r]));
+ if (expect) {
+ const at = (r) => (r ? ['key', 'local_path', 'durable_provider', 'packed_object_id', 'sha256'].map((f) => (r[f] == null ? '' : String(r[f]))).join('\n') : null);
+ const seqs = new Set([...before.keys(), ...expect.keys()]);
+ for (const q of seqs) if (at(before.get(q)) !== at(expect.get(q))) throw new TimelineChanged(`timeline of ${objectId}/${rendition} changed (seq ${q})`);
+ }
for (const r of rows) {
for (const r of rows) {
const prev = before.get(r.seq);
const prev = before.get(r.seq);
if (prev && same(prev, r)) { out.unchanged++; continue; }
if (prev && same(prev, r)) { out.unchanged++; continue; }
@@ -120,6 +136,17 @@ async function deleteBytes(rows) {
@@ -120,6 +136,17 @@ async function deleteBytes(rows) {
return { failed };
return { failed };
}
}
+/**
+ * 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.
+ */
+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;
+ 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
* 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
* durable copies. A row whose durable delete failed is kept — with its key — so a later pass can retry; dropping it
@@ -153,6 +180,7 @@ async function markPacked(objectId, rendition, updates, { jobId = null } = {}) {
@@ -153,6 +180,7 @@ async function markPacked(objectId, rendition, 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);
for (const u of updates) {
for (const u of updates) {
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()
packed_object_id = ?, byte_offset = ?, job_id = ?, updated_at = ov_now()
@@ -229,6 +257,6 @@ function masterPlaylist(renditions, { query = '' } = {}) {
@@ -229,6 +257,6 @@ function masterPlaylist(renditions, { query = '' } = {}) {
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, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,
+ list, segments, byName, has, segmentAt, replace, TimelineChanged, isNamed, deleteBytes, removeObject, markPacked, durableProvider, locationsOf,
mediaPlaylist, masterPlaylist,
mediaPlaylist, masterPlaylist,
};
};