openvibers/openvibe.media Commits
Timeline cache: files no row names (a cut that died before its commit) are removed once older than a day and their object is idle; the sweep skipped them for good (7 GB from VOD 5726's failed cuts) (#134)
Changes
@@@ -313,7 +313,11 @@ const ready = (async () => {try { const n = await require('./objects/model').purgeExpired(); if (n) console.log(`[Objects] Purged ${n} expired deleted object(s)`); } catch (err) { console.warn('[Objects] purge:', err.message); }// A projected object (a deleted clip's/VOD's) whose durable timeline delete failed keeps its rows; retry them.try { const r = await require('./objects/timeline').sweepPendingRemovals(); if (r.objects) console.log(`[Objects] Timeline retry: ${r.removed} row(s) of ${r.objects} gone object(s) removed, ${r.pending} still pending`); } catch (err) { console.warn('[Objects] timeline retry:', err.message); }-- try { const r = await require('./objects/timeline-cache').sweep(); if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`); } catch (err) { console.warn('[Objects] timeline cache:', err.message); }++ try {++ const r = await require('./objects/timeline-cache').sweep();++ if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`);++ if (r.orphans) console.log(`[Objects] Timeline cache: removed ${r.orphans} file(s) no row names (a cut that died before its commit)`);++ } catch (err) { console.warn('[Objects] timeline cache:', err.message); }// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }// S3 multipart uploads past their expiry (MEDIA_MULTIPART_TTL_HOURS): their parts are removed (server/s3/multipart.js).@@@ -8,7 +8,13 @@ const db = require('../db/database');const timeline = require('./timeline');let evictions = 0;++let orphansRemoved = 0;let sweeping = false;++// A cut stages its segments here and names them in one commit at the end (jobs/segments.js publishRendition); a failure++// before that commit removes them again, but a process that dies there (a restart, a killed worker) leaves files no row++// names. They are removed once older than any cut can run (a heavy job's limit is 13 h) and their object has no job++// queued or running. Before this, the sweep skipped them for good: VOD 5726's three failed cuts on 2026-10-15 left 7 GB.++const ORPHAN_AGE_MS = 24 * 3600 * 1000;function freeBytes() {try {@@@ -58,11 +64,34 @@ function served(file, now = Date.now()) {try { fs.utimesSync(file, now / 1000, now / 1000); } catch { /* an eviction may have just unlinked it */ }}--async function sweep() {++/** The object a timeline file belongs to: <root>/<app>/<object>/<rendition>/[<version>/]<name>, or null. */++function objectOf(file) {++ const parts = path.relative(timeline.localRoot(), file).split(path.sep);++ return parts.length >= 4 && !parts.includes('..') ? parts[1] : null;++}++++/** Remove the unnamed files that are old enough and whose object is idle. Returns the files that remain. */++async function sweepOrphans(entries, now) {++ const keep = [];++ const idle = new Map();++ for (const file of entries) {++ const object = objectOf(file.path);++ if (!object || now - file.servedAt < ORPHAN_AGE_MS) { keep.push(file); continue; }++ if (await db.get('SELECT 1 AS named FROM media_timeline WHERE local_path = ? LIMIT 1', [file.path])) { keep.push(file); continue; }++ if (!idle.has(object)) idle.set(object, !(await db.get("SELECT 1 AS busy FROM media_jobs WHERE object_id = ? AND status IN ('queued', 'running') LIMIT 1", [object])));++ if (!idle.get(object)) { keep.push(file); continue; }++ try { fs.unlinkSync(file.path); orphansRemoved++; } catch { keep.push(file); }++ }++ return keep;++}++++async function sweep({ now = Date.now() } = {}) {if (sweeping) return { skipped: true };sweeping = true;try {-- const entries = files().sort((a, b) => a.servedAt - b.servedAt);++ const before = orphansRemoved;++ const entries = (await sweepOrphans(files(), now)).sort((a, b) => a.servedAt - b.servedAt);++ const orphans = orphansRemoved - before;let bytes = entries.reduce((n, f) => n + f.size, 0);let free = freeBytes();let removed = 0;@@@ -96,9 +125,9 @@ async function sweep() {});free = freeBytes();}-- if (removed) memo = null;-- return { removed, ...status({ fresh: true }) };++ if (removed || orphans) memo = null;++ return { removed, orphans, ...status({ fresh: true }) };} finally { sweeping = false; }}--module.exports = { status, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions };++module.exports = { status, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions, orphansRemoved: () => orphansRemoved, ORPHAN_AGE_MS };nnew file mode 100644@@@ -0,0 +1,70 @@++'use strict';++// Timeline files no row names: a cut stages its segments and names them in one commit; a process that dies before that++// commit leaves them (VOD 5726's three failed cuts on 2026-10-15 left 7 GB, and the cache sweep skipped them for good).++// The sweep now removes them once they are older than any cut can run and their object has no job queued or running.++// Named files, recent files and the files of a busy object stay.++const assert = require('assert');++const crypto = require('crypto');++const fs = require('fs');++const os = require('os');++const path = require('path');++++(async () => {++ const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-timeline-orphans-'));++ process.env.OBJECTS_PATH = root;++ process.env.MEDIA_HLS_ENABLED = '1';++ process.env.MEDIA_TIMELINE_LOCAL_MAX_BYTES = String(1024 ** 3);++ process.env.MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES = '0';++ const db = require('../server/db/database');++ const timeline = require('../server/objects/timeline');++ const cache = require('../server/objects/timeline-cache');++ const now = Date.parse('2026-10-17T03:00:00Z');++ const DAY = 24 * 3600 * 1000;++++ const write = (obj, name, data, atMs) => {++ const bytes = Buffer.from(data);++ const sha = crypto.createHash('sha256').update(bytes).digest('hex');++ const file = timeline.localPathFor(obj, 'source', name, sha.slice(0, 12));++ fs.mkdirSync(path.dirname(file), { recursive: true });++ fs.writeFileSync(file, bytes);++ fs.utimesSync(file, atMs / 1000, atMs / 1000);++ return { file, bytes, sha };++ };++ try {++ const dead = { app_id: 'live', id: 'med_dead' };++ const named = { app_id: 'live', id: 'med_named' };++ const busy = { app_id: 'live', id: 'med_busy' };++++ // A cut that died before its commit, two days ago: nothing names these.++ const lost = [write(dead, '000001.m4s', 'lost-1', now - 2 * DAY), write(dead, '000002.m4s', 'lost-2', now - 2 * DAY)];++ // The same object's newer staging (a cut running right now could be writing these): too recent to judge.++ const fresh = write(dead, '000003.m4s', 'fresh', now - 3600 * 1000);++ // An old file a row still names: the cache's own business, never an orphan.++ const kept = write(named, '000001.m4s', 'kept', now - 3 * DAY);++ await timeline.replace(named.id, 'source', [{ seq: 1, name: '000001.m4s', start_ms: 0, end_ms: 1000, key: `${named.id}/source/${kept.sha.slice(0, 12)}/000001.m4s`,++ local_path: kept.file, durable_provider: null, packed_object_id: null, byte_offset: null, byte_length: kept.bytes.length, sha256: kept.sha, durability: 'local' }]);++ // An old unnamed file whose object has a cut queued: left for that cut.++ const waiting = write(busy, '000001.m4s', 'waiting', now - 2 * DAY);++ await db.run("INSERT INTO media_jobs (id, app_id, object_id, job_type, status) VALUES ('job_busy', 'live', ?, 'object.cmaf', 'queued')", [busy.id]);++ // A finished job does not hold anything back.++ await db.run("INSERT INTO media_jobs (id, app_id, object_id, job_type, status) VALUES ('job_dead', 'live', ?, 'object.cmaf', 'failed')", [dead.id]);++++ const out = await cache.sweep({ now });++ assert.strictEqual(out.orphans, 2, JSON.stringify(out));++ assert.ok(lost.every((f) => !fs.existsSync(f.file)), 'the dead cut\'s files are gone');++ assert.ok(fs.existsSync(fresh.file), 'a recent file stays (a cut may be staging it)');++ assert.ok(fs.existsSync(kept.file), 'a named file stays');++ assert.ok(fs.existsSync(waiting.file), 'a busy object\'s file stays');++ assert.strictEqual(cache.orphansRemoved(), 2);++ assert.strictEqual(cache.status({ fresh: true }).bytes, fresh.bytes.length + kept.bytes.length + waiting.bytes.length, 'the cache counts what is left');++++ // The queued cut finishes (and fails): the next sweep takes its object's leftover too.++ await db.run("UPDATE media_jobs SET status = 'failed' WHERE id = 'job_busy'");++ assert.strictEqual((await cache.sweep({ now })).orphans, 1);++ assert.ok(!fs.existsSync(waiting.file));++ console.log('timeline orphans: all checks passed');++ } finally {++ fs.rmSync(root, { recursive: true, force: true });++ }++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -313,7 +313,11 @@ const ready = (async () => {
@@ -313,7 +313,11 @@ const ready = (async () => {
try { const n = await require('./objects/model').purgeExpired(); if (n) console.log(`[Objects] Purged ${n} expired deleted object(s)`); } catch (err) { console.warn('[Objects] purge:', err.message); }
try { const n = await require('./objects/model').purgeExpired(); if (n) console.log(`[Objects] Purged ${n} expired deleted object(s)`); } catch (err) { console.warn('[Objects] purge:', err.message); }
// A projected object (a deleted clip's/VOD's) whose durable timeline delete failed keeps its rows; retry them.
// A projected object (a deleted clip's/VOD's) whose durable timeline delete failed keeps its rows; retry them.
try { const r = await require('./objects/timeline').sweepPendingRemovals(); if (r.objects) console.log(`[Objects] Timeline retry: ${r.removed} row(s) of ${r.objects} gone object(s) removed, ${r.pending} still pending`); } catch (err) { console.warn('[Objects] timeline retry:', err.message); }
try { const r = await require('./objects/timeline').sweepPendingRemovals(); if (r.objects) console.log(`[Objects] Timeline retry: ${r.removed} row(s) of ${r.objects} gone object(s) removed, ${r.pending} still pending`); } catch (err) { console.warn('[Objects] timeline retry:', err.message); }
- try { const r = await require('./objects/timeline-cache').sweep(); if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`); } catch (err) { console.warn('[Objects] timeline cache:', err.message); }
+ try {
+ const r = await require('./objects/timeline-cache').sweep();
+ if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`);
+ if (r.orphans) console.log(`[Objects] Timeline cache: removed ${r.orphans} file(s) no row names (a cut that died before its commit)`);
+ } catch (err) { console.warn('[Objects] timeline cache:', err.message); }
// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).
// Incomplete multipart uploads past MEDIA_MULTIPART_TTL_HOURS: their parts are deleted (the objects stay uploading).
try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }
try { const r = await require('./objects/multipart').purgeExpired(); if (r.expired || r.orphan_dirs) console.log(`[Objects] Multipart: ${r.expired} expired session(s), ${r.orphan_dirs} orphan part dir(s) removed`); } catch (err) { console.warn('[Objects] multipart purge:', err.message); }
// S3 multipart uploads past their expiry (MEDIA_MULTIPART_TTL_HOURS): their parts are removed (server/s3/multipart.js).
// S3 multipart uploads past their expiry (MEDIA_MULTIPART_TTL_HOURS): their parts are removed (server/s3/multipart.js).
@@ -8,7 +8,13 @@ const db = require('../db/database');
@@ -8,7 +8,13 @@ const db = require('../db/database');
const timeline = require('./timeline');
const timeline = require('./timeline');
let evictions = 0;
let evictions = 0;
+let orphansRemoved = 0;
let sweeping = false;
let sweeping = false;
+// A cut stages its segments here and names them in one commit at the end (jobs/segments.js publishRendition); a failure
+// before that commit removes them again, but a process that dies there (a restart, a killed worker) leaves files no row
+// names. They are removed once older than any cut can run (a heavy job's limit is 13 h) and their object has no job
+// queued or running. Before this, the sweep skipped them for good: VOD 5726's three failed cuts on 2026-10-15 left 7 GB.
+const ORPHAN_AGE_MS = 24 * 3600 * 1000;
function freeBytes() {
function freeBytes() {
try {
try {
@@ -58,11 +64,34 @@ function served(file, now = Date.now()) {
@@ -58,11 +64,34 @@ function served(file, now = Date.now()) {
try { fs.utimesSync(file, now / 1000, now / 1000); } catch { /* an eviction may have just unlinked it */ }
try { fs.utimesSync(file, now / 1000, now / 1000); } catch { /* an eviction may have just unlinked it */ }
}
}
-async function sweep() {
+/** The object a timeline file belongs to: <root>/<app>/<object>/<rendition>/[<version>/]<name>, or null. */
+function objectOf(file) {
+ const parts = path.relative(timeline.localRoot(), file).split(path.sep);
+ return parts.length >= 4 && !parts.includes('..') ? parts[1] : null;
+}
+
+/** Remove the unnamed files that are old enough and whose object is idle. Returns the files that remain. */
+async function sweepOrphans(entries, now) {
+ const keep = [];
+ const idle = new Map();
+ for (const file of entries) {
+ const object = objectOf(file.path);
+ if (!object || now - file.servedAt < ORPHAN_AGE_MS) { keep.push(file); continue; }
+ if (await db.get('SELECT 1 AS named FROM media_timeline WHERE local_path = ? LIMIT 1', [file.path])) { keep.push(file); continue; }
+ if (!idle.has(object)) idle.set(object, !(await db.get("SELECT 1 AS busy FROM media_jobs WHERE object_id = ? AND status IN ('queued', 'running') LIMIT 1", [object])));
+ if (!idle.get(object)) { keep.push(file); continue; }
+ try { fs.unlinkSync(file.path); orphansRemoved++; } catch { keep.push(file); }
+ }
+ return keep;
+}
+
+async function sweep({ now = Date.now() } = {}) {
if (sweeping) return { skipped: true };
if (sweeping) return { skipped: true };
sweeping = true;
sweeping = true;
try {
try {
- const entries = files().sort((a, b) => a.servedAt - b.servedAt);
+ const before = orphansRemoved;
+ const entries = (await sweepOrphans(files(), now)).sort((a, b) => a.servedAt - b.servedAt);
+ const orphans = orphansRemoved - before;
let bytes = entries.reduce((n, f) => n + f.size, 0);
let bytes = entries.reduce((n, f) => n + f.size, 0);
let free = freeBytes();
let free = freeBytes();
let removed = 0;
let removed = 0;
@@ -96,9 +125,9 @@ async function sweep() {
@@ -96,9 +125,9 @@ async function sweep() {
});
});
free = freeBytes();
free = freeBytes();
}
}
- if (removed) memo = null;
- return { removed, ...status({ fresh: true }) };
+ if (removed || orphans) memo = null;
+ return { removed, orphans, ...status({ fresh: true }) };
} finally { sweeping = false; }
} finally { sweeping = false; }
}
}
-module.exports = { status, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions };
+module.exports = { status, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions, orphansRemoved: () => orphansRemoved, ORPHAN_AGE_MS };
new file mode 100644
new file mode 100644
@@ -0,0 +1,70 @@
@@ -0,0 +1,70 @@
+'use strict';
+// Timeline files no row names: a cut stages its segments and names them in one commit; a process that dies before that
+// commit leaves them (VOD 5726's three failed cuts on 2026-10-15 left 7 GB, and the cache sweep skipped them for good).
+// The sweep now removes them once they are older than any cut can run and their object has no job queued or running.
+// Named files, recent files and the files of a busy object stay.
+const assert = require('assert');
+const crypto = require('crypto');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+(async () => {
+ const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-timeline-orphans-'));
+ process.env.OBJECTS_PATH = root;
+ process.env.MEDIA_HLS_ENABLED = '1';
+ process.env.MEDIA_TIMELINE_LOCAL_MAX_BYTES = String(1024 ** 3);
+ process.env.MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES = '0';
+ const db = require('../server/db/database');
+ const timeline = require('../server/objects/timeline');
+ const cache = require('../server/objects/timeline-cache');
+ const now = Date.parse('2026-10-17T03:00:00Z');
+ const DAY = 24 * 3600 * 1000;
+
+ const write = (obj, name, data, atMs) => {
+ const bytes = Buffer.from(data);
+ const sha = crypto.createHash('sha256').update(bytes).digest('hex');
+ const file = timeline.localPathFor(obj, 'source', name, sha.slice(0, 12));
+ fs.mkdirSync(path.dirname(file), { recursive: true });
+ fs.writeFileSync(file, bytes);
+ fs.utimesSync(file, atMs / 1000, atMs / 1000);
+ return { file, bytes, sha };
+ };
+ try {
+ const dead = { app_id: 'live', id: 'med_dead' };
+ const named = { app_id: 'live', id: 'med_named' };
+ const busy = { app_id: 'live', id: 'med_busy' };
+
+ // A cut that died before its commit, two days ago: nothing names these.
+ const lost = [write(dead, '000001.m4s', 'lost-1', now - 2 * DAY), write(dead, '000002.m4s', 'lost-2', now - 2 * DAY)];
+ // The same object's newer staging (a cut running right now could be writing these): too recent to judge.
+ const fresh = write(dead, '000003.m4s', 'fresh', now - 3600 * 1000);
+ // An old file a row still names: the cache's own business, never an orphan.
+ const kept = write(named, '000001.m4s', 'kept', now - 3 * DAY);
+ await timeline.replace(named.id, 'source', [{ seq: 1, name: '000001.m4s', start_ms: 0, end_ms: 1000, key: `${named.id}/source/${kept.sha.slice(0, 12)}/000001.m4s`,
+ local_path: kept.file, durable_provider: null, packed_object_id: null, byte_offset: null, byte_length: kept.bytes.length, sha256: kept.sha, durability: 'local' }]);
+ // An old unnamed file whose object has a cut queued: left for that cut.
+ const waiting = write(busy, '000001.m4s', 'waiting', now - 2 * DAY);
+ await db.run("INSERT INTO media_jobs (id, app_id, object_id, job_type, status) VALUES ('job_busy', 'live', ?, 'object.cmaf', 'queued')", [busy.id]);
+ // A finished job does not hold anything back.
+ await db.run("INSERT INTO media_jobs (id, app_id, object_id, job_type, status) VALUES ('job_dead', 'live', ?, 'object.cmaf', 'failed')", [dead.id]);
+
+ const out = await cache.sweep({ now });
+ assert.strictEqual(out.orphans, 2, JSON.stringify(out));
+ assert.ok(lost.every((f) => !fs.existsSync(f.file)), 'the dead cut\'s files are gone');
+ assert.ok(fs.existsSync(fresh.file), 'a recent file stays (a cut may be staging it)');
+ assert.ok(fs.existsSync(kept.file), 'a named file stays');
+ assert.ok(fs.existsSync(waiting.file), 'a busy object\'s file stays');
+ assert.strictEqual(cache.orphansRemoved(), 2);
+ assert.strictEqual(cache.status({ fresh: true }).bytes, fresh.bytes.length + kept.bytes.length + waiting.bytes.length, 'the cache counts what is left');
+
+ // The queued cut finishes (and fails): the next sweep takes its object's leftover too.
+ await db.run("UPDATE media_jobs SET status = 'failed' WHERE id = 'job_busy'");
+ assert.strictEqual((await cache.sweep({ now })).orphans, 1);
+ assert.ok(!fs.existsSync(waiting.file));
+ console.log('timeline orphans: all checks passed');
+ } finally {
+ fs.rmSync(root, { recursive: true, force: true });
+ }
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });