openvibers/openvibe.media Commits
The CMAF cache gives way to the disk's drain: one drain for the whole disk (#238)
Changes
@@@ -338,7 +338,7 @@ const ready = (async () => {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)`);++ if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)${r.draining ? ' for the disk drain (durable copies serve them)' : ''}`);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).@@@ -107,6 +107,22 @@ async function sweepOrphans(entries, now) {return keep;}++// One drain for the whole disk (2026-10-31). Under disk pressure the VOD sweep drains to its targets (≤60% and ≥40 GB++// free), but this cache kept only its own budget (MEDIA_TIMELINE_LOCAL_MAX_BYTES / _MIN_FREE_BYTES). On 2026-10-31 the++// disk sat at 80% with 10 GB of segments here, all with verified durable copies, while the VOD sweep found nothing it++// may move (recordings stay local for a day). Now, while the VOD storage says the disk needs a drain, the cache gives++// way too: durable segments go, least recently served first, until the drain is satisfied. A segment served in the++// last hour stays (someone is watching it); a segment with no verified durable copy never goes.++const HOT_MS = 3600 * 1000;++function diskPressure() {++ try {++ const vs = require('../vod/vod-storage');++ const settings = vs.getSettings();++ const disk = () => vs.diskUsage(config.objects.path);++ return { draining: vs.needsDrain(disk(), settings), satisfied: () => vs.drainSatisfied(disk(), settings) };++ } catch { return { draining: false, satisfied: () => true }; } // no pressure reading: the budget alone decides++}++async function sweep({ now = Date.now() } = {}) {if (sweeping) return { skipped: true };sweeping = true;@@@ -117,8 +133,13 @@ async function sweep({ now = Date.now() } = {}) {let bytes = entries.reduce((n, f) => n + f.size, 0);let free = freeBytes();let removed = 0;++ const pressure = diskPressure();for (const file of entries) {-- if (bytes <= config.hls.localMaxBytes && free >= config.hls.localMinFreeBytes) break;++ if (bytes <= config.hls.localMaxBytes && free >= config.hls.localMinFreeBytes) {++ // Within the cache's own budget: only the disk's drain can still ask for room, and only from files++ // nobody has watched in the last hour (the list is oldest first, so the rest are newer still).++ if (!pressure.draining || pressure.satisfied() || now - file.servedAt < HOT_MS) break;++ }// A path can be shared by source, clip and multiple packed rows. All must name a verified durable// provider before the path can go. The row's durable state is set only after size+sha readback.const rows = await db.all('SELECT durability, durable_provider, key, sha256, packed_object_id, byte_offset, byte_length FROM media_timeline WHERE local_path = ?', [file.path]);@@@ -148,7 +169,7 @@ async function sweep({ now = Date.now() } = {}) {free = freeBytes();}memo = { at: Date.now(), bytes }; // what is left, counted by this walk-- return { removed, orphans, ...status() };++ return { removed, orphans, draining: pressure.draining, ...status() };} finally { sweeping = false; }}nnew file mode 100644@@@ -0,0 +1,85 @@++'use strict';++// The CMAF cache gives way to the disk's drain (2026-10-31): within its own budget it still evicts durable segments,++// least recently served first, while the VOD storage says the disk needs a drain, and stops once the drain is++// satisfied. A segment served in the last hour stays; one with no verified durable copy never goes. Without pressure++// nothing changes. No listening socket: real migrated PGlite rows, local files and a fake durable readback.++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-pressure-'));++ process.env.OBJECTS_PATH = root;++ process.env.MEDIA_HLS_ENABLED = '1';++ process.env.MEDIA_TIMELINE_LOCAL_MAX_BYTES = String(1024 ** 4); // far above what the test writes: the budget never asks++ process.env.MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES = '0';++ const timeline = require('../server/objects/timeline');++ const cache = require('../server/objects/timeline-cache');++ const storage = require('../server/vod/vod-storage');++ const blobs = new Map();++ storage.sha256Object = async (_provider, key) => {++ const bytes = blobs.get(key);++ return bytes ? { size: bytes.length, sha256: crypto.createHash('sha256').update(bytes).digest('hex') } : null;++ };++ // The disk's drain, faked: it needs a drain, and is satisfied once `room` evictions have happened.++ let draining = false;++ let room = 0;++ storage.getSettings = () => ({});++ storage.diskUsage = () => ({ total: 100, used: 80, available: 20, usePct: 80 });++ storage.needsDrain = () => draining;++ storage.drainSatisfied = () => cache.evictions() >= room;++ try {++ const obj = { app_id: 'live', id: 'pressure-test' };++ const make = async (seq, data, durable, at) => {++ const bytes = Buffer.from(data);++ const name = timeline.segmentName(seq);++ const sha = crypto.createHash('sha256').update(bytes).digest('hex');++ const key = `pressure/${name}`;++ const file = timeline.localPathFor(obj, 'source', name, sha.slice(0, 12));++ fs.mkdirSync(path.dirname(file), { recursive: true });++ fs.writeFileSync(file, bytes);++ cache.served(file, at);++ if (durable) blobs.set(key, bytes);++ await timeline.replace(`pressure-${seq}`, 'source', [{ seq, name, start_ms: 0, end_ms: 1000, key, local_path: file,++ durable_provider: durable ? 'b2' : null, packed_object_id: null, byte_offset: null,++ byte_length: bytes.length, sha256: sha, durability: durable ? 'durable' : 'local' }]);++ return file;++ };++ const now = Date.now();++ const hourAgo = 3600 * 1000;++ const oldest = await make(1, 'aaaaaa', true, now - 30 * hourAgo);++ const older = await make(2, 'bbbbbb', true, now - 20 * hourAgo);++ const onlyCopy = await make(3, 'cccccc', false, now - 40 * hourAgo);++ const old = await make(4, 'dddddd', true, now - 10 * hourAgo);++ const watched = await make(5, 'eeeeee', true, now - 60 * 1000);++++ // No pressure: within its budget the cache keeps everything.++ let r = await cache.sweep({ now });++ assert.strictEqual(r.removed, 0, 'no drain, within budget: nothing goes');++ assert.strictEqual(r.draining, false);++++ // Pressure until two files are gone: the two least recently served durable files go, oldest first.++ draining = true; room = 2;++ r = await cache.sweep({ now });++ assert.strictEqual(r.draining, true);++ assert.strictEqual(r.removed, 2, 'it stops once the drain is satisfied');++ assert.strictEqual(fs.existsSync(oldest), false);++ assert.strictEqual(fs.existsSync(older), false);++ assert.strictEqual(fs.existsSync(old), true, 'the drain was satisfied before this one');++ assert.strictEqual(fs.existsSync(onlyCopy), true, 'a file with no verified durable copy never goes');++ assert.strictEqual((await timeline.byName('pressure-1', 'source', timeline.segmentName(1))).local_path, null, 'the row now points at its durable copy only');++++ // Pressure that is never satisfied: everything durable goes except what was served in the last hour.++ room = Infinity;++ r = await cache.sweep({ now });++ assert.strictEqual(fs.existsSync(old), false);++ assert.strictEqual(fs.existsSync(watched), true, 'a segment someone watched in the last hour stays');++ assert.strictEqual(fs.existsSync(onlyCopy), true);++ console.log('timeline cache under disk pressure: gives way oldest first, stops when satisfied, keeps the watched and the only copies ok');++ } finally {++ fs.rmSync(root, { recursive: true, force: true });++ await require('../server/db/database').close?.();++ }++})().catch((e) => { console.error(e); process.exit(1); });
@@ -338,7 +338,7 @@ const ready = (async () => {
@@ -338,7 +338,7 @@ const ready = (async () => {
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 {
try {
const r = await require('./objects/timeline-cache').sweep();
const r = await require('./objects/timeline-cache').sweep();
- if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)`);
+ if (r.removed) console.log(`[Objects] Timeline cache: evicted ${r.removed} local file(s)${r.draining ? ' for the disk drain (durable copies serve them)' : ''}`);
if (r.orphans) console.log(`[Objects] Timeline cache: removed ${r.orphans} file(s) no row names (a cut that died before its commit)`);
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); }
} 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).
@@ -107,6 +107,22 @@ async function sweepOrphans(entries, now) {
@@ -107,6 +107,22 @@ async function sweepOrphans(entries, now) {
return keep;
return keep;
}
}
+// One drain for the whole disk (2026-10-31). Under disk pressure the VOD sweep drains to its targets (≤60% and ≥40 GB
+// free), but this cache kept only its own budget (MEDIA_TIMELINE_LOCAL_MAX_BYTES / _MIN_FREE_BYTES). On 2026-10-31 the
+// disk sat at 80% with 10 GB of segments here, all with verified durable copies, while the VOD sweep found nothing it
+// may move (recordings stay local for a day). Now, while the VOD storage says the disk needs a drain, the cache gives
+// way too: durable segments go, least recently served first, until the drain is satisfied. A segment served in the
+// last hour stays (someone is watching it); a segment with no verified durable copy never goes.
+const HOT_MS = 3600 * 1000;
+function diskPressure() {
+ try {
+ const vs = require('../vod/vod-storage');
+ const settings = vs.getSettings();
+ const disk = () => vs.diskUsage(config.objects.path);
+ return { draining: vs.needsDrain(disk(), settings), satisfied: () => vs.drainSatisfied(disk(), settings) };
+ } catch { return { draining: false, satisfied: () => true }; } // no pressure reading: the budget alone decides
+}
+
async function sweep({ now = Date.now() } = {}) {
async function sweep({ now = Date.now() } = {}) {
if (sweeping) return { skipped: true };
if (sweeping) return { skipped: true };
sweeping = true;
sweeping = true;
@@ -117,8 +133,13 @@ async function sweep({ now = Date.now() } = {}) {
@@ -117,8 +133,13 @@ async function sweep({ now = Date.now() } = {}) {
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;
+ const pressure = diskPressure();
for (const file of entries) {
for (const file of entries) {
- if (bytes <= config.hls.localMaxBytes && free >= config.hls.localMinFreeBytes) break;
+ if (bytes <= config.hls.localMaxBytes && free >= config.hls.localMinFreeBytes) {
+ // Within the cache's own budget: only the disk's drain can still ask for room, and only from files
+ // nobody has watched in the last hour (the list is oldest first, so the rest are newer still).
+ if (!pressure.draining || pressure.satisfied() || now - file.servedAt < HOT_MS) break;
+ }
// A path can be shared by source, clip and multiple packed rows. All must name a verified durable
// A path can be shared by source, clip and multiple packed rows. All must name a verified durable
// provider before the path can go. The row's durable state is set only after size+sha readback.
// provider before the path can go. The row's durable state is set only after size+sha readback.
const rows = await db.all('SELECT durability, durable_provider, key, sha256, packed_object_id, byte_offset, byte_length FROM media_timeline WHERE local_path = ?', [file.path]);
const rows = await db.all('SELECT durability, durable_provider, key, sha256, packed_object_id, byte_offset, byte_length FROM media_timeline WHERE local_path = ?', [file.path]);
@@ -148,7 +169,7 @@ async function sweep({ now = Date.now() } = {}) {
@@ -148,7 +169,7 @@ async function sweep({ now = Date.now() } = {}) {
free = freeBytes();
free = freeBytes();
}
}
memo = { at: Date.now(), bytes }; // what is left, counted by this walk
memo = { at: Date.now(), bytes }; // what is left, counted by this walk
- return { removed, orphans, ...status() };
+ return { removed, orphans, draining: pressure.draining, ...status() };
} finally { sweeping = false; }
} finally { sweeping = false; }
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,85 @@
@@ -0,0 +1,85 @@
+'use strict';
+// The CMAF cache gives way to the disk's drain (2026-10-31): within its own budget it still evicts durable segments,
+// least recently served first, while the VOD storage says the disk needs a drain, and stops once the drain is
+// satisfied. A segment served in the last hour stays; one with no verified durable copy never goes. Without pressure
+// nothing changes. No listening socket: real migrated PGlite rows, local files and a fake durable readback.
+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-pressure-'));
+ process.env.OBJECTS_PATH = root;
+ process.env.MEDIA_HLS_ENABLED = '1';
+ process.env.MEDIA_TIMELINE_LOCAL_MAX_BYTES = String(1024 ** 4); // far above what the test writes: the budget never asks
+ process.env.MEDIA_TIMELINE_LOCAL_MIN_FREE_BYTES = '0';
+ const timeline = require('../server/objects/timeline');
+ const cache = require('../server/objects/timeline-cache');
+ const storage = require('../server/vod/vod-storage');
+ const blobs = new Map();
+ storage.sha256Object = async (_provider, key) => {
+ const bytes = blobs.get(key);
+ return bytes ? { size: bytes.length, sha256: crypto.createHash('sha256').update(bytes).digest('hex') } : null;
+ };
+ // The disk's drain, faked: it needs a drain, and is satisfied once `room` evictions have happened.
+ let draining = false;
+ let room = 0;
+ storage.getSettings = () => ({});
+ storage.diskUsage = () => ({ total: 100, used: 80, available: 20, usePct: 80 });
+ storage.needsDrain = () => draining;
+ storage.drainSatisfied = () => cache.evictions() >= room;
+ try {
+ const obj = { app_id: 'live', id: 'pressure-test' };
+ const make = async (seq, data, durable, at) => {
+ const bytes = Buffer.from(data);
+ const name = timeline.segmentName(seq);
+ const sha = crypto.createHash('sha256').update(bytes).digest('hex');
+ const key = `pressure/${name}`;
+ const file = timeline.localPathFor(obj, 'source', name, sha.slice(0, 12));
+ fs.mkdirSync(path.dirname(file), { recursive: true });
+ fs.writeFileSync(file, bytes);
+ cache.served(file, at);
+ if (durable) blobs.set(key, bytes);
+ await timeline.replace(`pressure-${seq}`, 'source', [{ seq, name, start_ms: 0, end_ms: 1000, key, local_path: file,
+ durable_provider: durable ? 'b2' : null, packed_object_id: null, byte_offset: null,
+ byte_length: bytes.length, sha256: sha, durability: durable ? 'durable' : 'local' }]);
+ return file;
+ };
+ const now = Date.now();
+ const hourAgo = 3600 * 1000;
+ const oldest = await make(1, 'aaaaaa', true, now - 30 * hourAgo);
+ const older = await make(2, 'bbbbbb', true, now - 20 * hourAgo);
+ const onlyCopy = await make(3, 'cccccc', false, now - 40 * hourAgo);
+ const old = await make(4, 'dddddd', true, now - 10 * hourAgo);
+ const watched = await make(5, 'eeeeee', true, now - 60 * 1000);
+
+ // No pressure: within its budget the cache keeps everything.
+ let r = await cache.sweep({ now });
+ assert.strictEqual(r.removed, 0, 'no drain, within budget: nothing goes');
+ assert.strictEqual(r.draining, false);
+
+ // Pressure until two files are gone: the two least recently served durable files go, oldest first.
+ draining = true; room = 2;
+ r = await cache.sweep({ now });
+ assert.strictEqual(r.draining, true);
+ assert.strictEqual(r.removed, 2, 'it stops once the drain is satisfied');
+ assert.strictEqual(fs.existsSync(oldest), false);
+ assert.strictEqual(fs.existsSync(older), false);
+ assert.strictEqual(fs.existsSync(old), true, 'the drain was satisfied before this one');
+ assert.strictEqual(fs.existsSync(onlyCopy), true, 'a file with no verified durable copy never goes');
+ assert.strictEqual((await timeline.byName('pressure-1', 'source', timeline.segmentName(1))).local_path, null, 'the row now points at its durable copy only');
+
+ // Pressure that is never satisfied: everything durable goes except what was served in the last hour.
+ room = Infinity;
+ r = await cache.sweep({ now });
+ assert.strictEqual(fs.existsSync(old), false);
+ assert.strictEqual(fs.existsSync(watched), true, 'a segment someone watched in the last hour stays');
+ assert.strictEqual(fs.existsSync(onlyCopy), true);
+ console.log('timeline cache under disk pressure: gives way oldest first, stops when satisfied, keeps the watched and the only copies ok');
+ } finally {
+ fs.rmSync(root, { recursive: true, force: true });
+ await require('../server/db/database').close?.();
+ }
+})().catch((e) => { console.error(e); process.exit(1); });