openvibers/openvibe.media Commits
Timeline cache: the local walk no longer blocks the event loop (1.9 s every 30 s on each /metrics scrape, from 193,900 empty directories left by evictions); the hourly sweep removes emptied directories (#143)
Changes
@@@ -23,38 +23,60 @@ function freeBytes() {} catch { return 0; } // Unknown free space must not authorize another local copy.}--function files() {++// An emptied directory (its files evicted or swept) older than this is removed by the sweep. Evictions used to leave++// every <app>/<object>/<rendition>/<version> directory behind: 193,900 directories for 1,080 files on 2026-10-17, and++// walking them took 1.9 s of the event loop on every /metrics scrape (every 30 s).++const EMPTY_DIR_AGE_MS = 3600 * 1000;++++/** Every local timeline file, read without blocking the event loop. `prune`: remove emptied directories on the way. */++async function files({ prune = false, now = Date.now() } = {}) {const root = timeline.localRoot();const out = [];-- function visit(dir) {++ async function visit(dir, depth) {let entries;-- try { entries = fs.readdirSync(dir, { withFileTypes: true }); } catch { return; }++ try { entries = await fs.promises.readdir(dir, { withFileTypes: true }); } catch { return 1; }++ // Its age from before this walk removed anything inside it (a removal touches the parent's mtime).++ let touched = null;++ if (prune && depth >= 2) { try { touched = (await fs.promises.stat(dir)).mtimeMs; } catch { /* gone */ } }++ let left = 0;for (const entry of entries) {const file = path.join(dir, entry.name);-- if (entry.isDirectory()) visit(file);-- else if (entry.isFile()) {-- try { const st = fs.statSync(file); out.push({ path: file, size: st.size, servedAt: st.mtimeMs }); }-- catch { /* disappeared while scanning */ }-- }++ if (entry.isDirectory()) { left += await visit(file, depth + 1); continue; }++ left++;++ if (!entry.isFile()) continue;++ try { const st = await fs.promises.stat(file); out.push({ path: file, size: st.size, servedAt: st.mtimeMs }); }++ catch { /* disappeared while scanning */ }++ }++ // <root> and <root>/<app> stay; below them an empty directory nothing has touched for an hour goes (a cut++ // writing into a new one is minutes old).++ if (!left && touched !== null && now - touched >= EMPTY_DIR_AGE_MS) {++ try { await fs.promises.rmdir(dir); return 0; } catch { /* written to meanwhile */ }}++ return 1; // still there: it counts in its parent}-- visit(root);++ await visit(root, 0);return out;}--// status() walks every local timeline file, so readiness and /metrics read a copy at most 30 s old; a write decision--// (shouldKeepLocal) and the sweep read it fresh, and a sweep that removed files drops the copy.++// status() is synchronous (readiness, /metrics) and never walks: it answers with the last measurement and starts a new++// one in the background once that is 30 s old. refresh() measures now; the sweep leaves its own count behind.const STATUS_TTL_MS = 30_000;let memo = null;--function status({ fresh = false } = {}) {-- const now = Date.now();-- // Only the measured numbers are kept; the budget and the floor are read from config on every call.-- if (fresh || !memo || now - memo.at >= STATUS_TTL_MS) memo = { at: now, bytes: files().reduce((n, f) => n + f.size, 0), free: freeBytes() };-- return { bytes: memo.bytes, max_bytes: config.hls.localMaxBytes, free_bytes: memo.free, min_free_bytes: config.hls.localMinFreeBytes };++let measuring = null;++function refresh() {++ if (!measuring) {++ measuring = files().then((list) => { memo = { at: Date.now(), bytes: list.reduce((n, f) => n + f.size, 0) }; return status(); })++ .finally(() => { measuring = null; });++ }++ return measuring;++}++function status() {++ if (!memo || Date.now() - memo.at >= STATUS_TTL_MS) refresh().catch(() => {});++ return { bytes: memo ? memo.bytes : 0, max_bytes: config.hls.localMaxBytes, free_bytes: freeBytes(), min_free_bytes: config.hls.localMinFreeBytes, ...(memo ? {} : { measuring: true }) };}--function shouldKeepLocal() {-- const s = status({ fresh: true });++async function shouldKeepLocal() {++ const s = await refresh();return s.max_bytes > 0 && s.bytes < s.max_bytes && s.free_bytes >= s.min_free_bytes;}@@@ -90,7 +112,7 @@ async function sweep({ now = Date.now() } = {}) {sweeping = true;try {const before = orphansRemoved;-- const entries = (await sweepOrphans(files(), now)).sort((a, b) => a.servedAt - b.servedAt);++ const entries = (await sweepOrphans(await files({ prune: true, now }), 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();@@@ -125,9 +147,9 @@ async function sweep({ now = Date.now() } = {}) {});free = freeBytes();}-- if (removed || orphans) memo = null;-- return { removed, orphans, ...status({ fresh: true }) };++ memo = { at: Date.now(), bytes }; // what is left, counted by this walk++ return { removed, orphans, ...status() };} finally { sweeping = false; }}--module.exports = { status, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions, orphansRemoved: () => orphansRemoved, ORPHAN_AGE_MS };++module.exports = { status, refresh, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions, orphansRemoved: () => orphansRemoved, ORPHAN_AGE_MS };@@@ -45,8 +45,8 @@ const { EventEmitter } = require('events');const old = await make(1, 'aaaaaa', true, 1000);const recent = await make(2, 'bbbbbb', true, 3000);const only = await make(3, 'cccccc', false, 500);-- assert.strictEqual(cache.status().bytes, 18);-- assert.strictEqual(cache.shouldKeepLocal(), false);++ assert.strictEqual((await cache.refresh()).bytes, 18);++ assert.strictEqual(await cache.shouldKeepLocal(), false);const first = await cache.sweep();assert.strictEqual(first.removed, 2);assert.strictEqual(fs.existsSync(old), false);@@@ -126,6 +126,24 @@ const { EventEmitter } = require('events');assert.strictEqual(fs.existsSync(only), true);console.log('timeline cache: oldest verified copies and packed range evicted; sole copy retained');++ // Emptied directories: the sweep removes the ones nothing touched for an hour; a new one (a cut may be writing++ // into it) and the <app> level stay. The walk itself never blocks (no synchronous readdir/stat in the module).++ const tlRoot = timeline.localRoot();++ const stale = path.join(tlRoot, 'live', 'gone-object', 'source', 'v1');++ const young = path.join(tlRoot, 'live', 'new-object', 'source');++ const appDir = path.join(tlRoot, 'empty-app');++ for (const d of [stale, young, appDir]) fs.mkdirSync(d, { recursive: true });++ const hourAgo = (Date.now() - 2 * 3600 * 1000) / 1000;++ for (const d of [stale, path.dirname(stale), path.dirname(path.dirname(stale)), appDir]) fs.utimesSync(d, hourAgo, hourAgo);++ await cache.sweep();++ assert.strictEqual(fs.existsSync(path.join(tlRoot, 'live', 'gone-object')), false, 'an old emptied object directory goes, down to its leaves');++ assert.strictEqual(fs.existsSync(young), true, 'a new empty directory stays');++ assert.strictEqual(fs.existsSync(appDir), true, 'the app level stays');++ assert.strictEqual(fs.existsSync(only), true, 'files are never touched by the pruning');++ const src = fs.readFileSync(path.join(__dirname, '../server/objects/timeline-cache.js'), 'utf8');++ assert.ok(!/readdirSync|statSync\(/.test(src), 'the walk is asynchronous');++ console.log('timeline cache: emptied directories pruned, the walk does not block');++// The CMAF publisher stages normally, verifies the upload, then publishes remote-only rows below the floor.const publishDir = path.join(root, 'publish');fs.mkdirSync(publishDir);@@@ -56,7 +56,7 @@ const path = require('path');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');++ assert.strictEqual((await cache.refresh()).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'");
@@ -23,38 +23,60 @@ function freeBytes() {
@@ -23,38 +23,60 @@ function freeBytes() {
} catch { return 0; } // Unknown free space must not authorize another local copy.
} catch { return 0; } // Unknown free space must not authorize another local copy.
}
}
-function files() {
+// An emptied directory (its files evicted or swept) older than this is removed by the sweep. Evictions used to leave
+// every <app>/<object>/<rendition>/<version> directory behind: 193,900 directories for 1,080 files on 2026-10-17, and
+// walking them took 1.9 s of the event loop on every /metrics scrape (every 30 s).
+const EMPTY_DIR_AGE_MS = 3600 * 1000;
+
+/** Every local timeline file, read without blocking the event loop. `prune`: remove emptied directories on the way. */
+async function files({ prune = false, now = Date.now() } = {}) {
const root = timeline.localRoot();
const root = timeline.localRoot();
const out = [];
const out = [];
- function visit(dir) {
+ async function visit(dir, depth) {
let entries;
let entries;
- try { entries = fs.readdirSync(dir, { withFileTypes: true }); } catch { return; }
+ try { entries = await fs.promises.readdir(dir, { withFileTypes: true }); } catch { return 1; }
+ // Its age from before this walk removed anything inside it (a removal touches the parent's mtime).
+ let touched = null;
+ if (prune && depth >= 2) { try { touched = (await fs.promises.stat(dir)).mtimeMs; } catch { /* gone */ } }
+ let left = 0;
for (const entry of entries) {
for (const entry of entries) {
const file = path.join(dir, entry.name);
const file = path.join(dir, entry.name);
- if (entry.isDirectory()) visit(file);
- else if (entry.isFile()) {
- try { const st = fs.statSync(file); out.push({ path: file, size: st.size, servedAt: st.mtimeMs }); }
- catch { /* disappeared while scanning */ }
- }
+ if (entry.isDirectory()) { left += await visit(file, depth + 1); continue; }
+ left++;
+ if (!entry.isFile()) continue;
+ try { const st = await fs.promises.stat(file); out.push({ path: file, size: st.size, servedAt: st.mtimeMs }); }
+ catch { /* disappeared while scanning */ }
+ }
+ // <root> and <root>/<app> stay; below them an empty directory nothing has touched for an hour goes (a cut
+ // writing into a new one is minutes old).
+ if (!left && touched !== null && now - touched >= EMPTY_DIR_AGE_MS) {
+ try { await fs.promises.rmdir(dir); return 0; } catch { /* written to meanwhile */ }
}
}
+ return 1; // still there: it counts in its parent
}
}
- visit(root);
+ await visit(root, 0);
return out;
return out;
}
}
-// status() walks every local timeline file, so readiness and /metrics read a copy at most 30 s old; a write decision
-// (shouldKeepLocal) and the sweep read it fresh, and a sweep that removed files drops the copy.
+// status() is synchronous (readiness, /metrics) and never walks: it answers with the last measurement and starts a new
+// one in the background once that is 30 s old. refresh() measures now; the sweep leaves its own count behind.
const STATUS_TTL_MS = 30_000;
const STATUS_TTL_MS = 30_000;
let memo = null;
let memo = null;
-function status({ fresh = false } = {}) {
- const now = Date.now();
- // Only the measured numbers are kept; the budget and the floor are read from config on every call.
- if (fresh || !memo || now - memo.at >= STATUS_TTL_MS) memo = { at: now, bytes: files().reduce((n, f) => n + f.size, 0), free: freeBytes() };
- return { bytes: memo.bytes, max_bytes: config.hls.localMaxBytes, free_bytes: memo.free, min_free_bytes: config.hls.localMinFreeBytes };
+let measuring = null;
+function refresh() {
+ if (!measuring) {
+ measuring = files().then((list) => { memo = { at: Date.now(), bytes: list.reduce((n, f) => n + f.size, 0) }; return status(); })
+ .finally(() => { measuring = null; });
+ }
+ return measuring;
+}
+function status() {
+ if (!memo || Date.now() - memo.at >= STATUS_TTL_MS) refresh().catch(() => {});
+ return { bytes: memo ? memo.bytes : 0, max_bytes: config.hls.localMaxBytes, free_bytes: freeBytes(), min_free_bytes: config.hls.localMinFreeBytes, ...(memo ? {} : { measuring: true }) };
}
}
-function shouldKeepLocal() {
- const s = status({ fresh: true });
+async function shouldKeepLocal() {
+ const s = await refresh();
return s.max_bytes > 0 && s.bytes < s.max_bytes && s.free_bytes >= s.min_free_bytes;
return s.max_bytes > 0 && s.bytes < s.max_bytes && s.free_bytes >= s.min_free_bytes;
}
}
@@ -90,7 +112,7 @@ async function sweep({ now = Date.now() } = {}) {
@@ -90,7 +112,7 @@ async function sweep({ now = Date.now() } = {}) {
sweeping = true;
sweeping = true;
try {
try {
const before = orphansRemoved;
const before = orphansRemoved;
- const entries = (await sweepOrphans(files(), now)).sort((a, b) => a.servedAt - b.servedAt);
+ const entries = (await sweepOrphans(await files({ prune: true, now }), now)).sort((a, b) => a.servedAt - b.servedAt);
const orphans = orphansRemoved - before;
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();
@@ -125,9 +147,9 @@ async function sweep({ now = Date.now() } = {}) {
@@ -125,9 +147,9 @@ async function sweep({ now = Date.now() } = {}) {
});
});
free = freeBytes();
free = freeBytes();
}
}
- if (removed || orphans) memo = null;
- return { removed, orphans, ...status({ fresh: true }) };
+ memo = { at: Date.now(), bytes }; // what is left, counted by this walk
+ return { removed, orphans, ...status() };
} finally { sweeping = false; }
} finally { sweeping = false; }
}
}
-module.exports = { status, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions, orphansRemoved: () => orphansRemoved, ORPHAN_AGE_MS };
+module.exports = { status, refresh, freeBytes, shouldKeepLocal, served, sweep, evictions: () => evictions, orphansRemoved: () => orphansRemoved, ORPHAN_AGE_MS };
@@ -45,8 +45,8 @@ const { EventEmitter } = require('events');
@@ -45,8 +45,8 @@ const { EventEmitter } = require('events');
const old = await make(1, 'aaaaaa', true, 1000);
const old = await make(1, 'aaaaaa', true, 1000);
const recent = await make(2, 'bbbbbb', true, 3000);
const recent = await make(2, 'bbbbbb', true, 3000);
const only = await make(3, 'cccccc', false, 500);
const only = await make(3, 'cccccc', false, 500);
- assert.strictEqual(cache.status().bytes, 18);
- assert.strictEqual(cache.shouldKeepLocal(), false);
+ assert.strictEqual((await cache.refresh()).bytes, 18);
+ assert.strictEqual(await cache.shouldKeepLocal(), false);
const first = await cache.sweep();
const first = await cache.sweep();
assert.strictEqual(first.removed, 2);
assert.strictEqual(first.removed, 2);
assert.strictEqual(fs.existsSync(old), false);
assert.strictEqual(fs.existsSync(old), false);
@@ -126,6 +126,24 @@ const { EventEmitter } = require('events');
@@ -126,6 +126,24 @@ const { EventEmitter } = require('events');
assert.strictEqual(fs.existsSync(only), true);
assert.strictEqual(fs.existsSync(only), true);
console.log('timeline cache: oldest verified copies and packed range evicted; sole copy retained');
console.log('timeline cache: oldest verified copies and packed range evicted; sole copy retained');
+ // Emptied directories: the sweep removes the ones nothing touched for an hour; a new one (a cut may be writing
+ // into it) and the <app> level stay. The walk itself never blocks (no synchronous readdir/stat in the module).
+ const tlRoot = timeline.localRoot();
+ const stale = path.join(tlRoot, 'live', 'gone-object', 'source', 'v1');
+ const young = path.join(tlRoot, 'live', 'new-object', 'source');
+ const appDir = path.join(tlRoot, 'empty-app');
+ for (const d of [stale, young, appDir]) fs.mkdirSync(d, { recursive: true });
+ const hourAgo = (Date.now() - 2 * 3600 * 1000) / 1000;
+ for (const d of [stale, path.dirname(stale), path.dirname(path.dirname(stale)), appDir]) fs.utimesSync(d, hourAgo, hourAgo);
+ await cache.sweep();
+ assert.strictEqual(fs.existsSync(path.join(tlRoot, 'live', 'gone-object')), false, 'an old emptied object directory goes, down to its leaves');
+ assert.strictEqual(fs.existsSync(young), true, 'a new empty directory stays');
+ assert.strictEqual(fs.existsSync(appDir), true, 'the app level stays');
+ assert.strictEqual(fs.existsSync(only), true, 'files are never touched by the pruning');
+ const src = fs.readFileSync(path.join(__dirname, '../server/objects/timeline-cache.js'), 'utf8');
+ assert.ok(!/readdirSync|statSync\(/.test(src), 'the walk is asynchronous');
+ console.log('timeline cache: emptied directories pruned, the walk does not block');
+
// The CMAF publisher stages normally, verifies the upload, then publishes remote-only rows below the floor.
// The CMAF publisher stages normally, verifies the upload, then publishes remote-only rows below the floor.
const publishDir = path.join(root, 'publish');
const publishDir = path.join(root, 'publish');
fs.mkdirSync(publishDir);
fs.mkdirSync(publishDir);
@@ -56,7 +56,7 @@ const path = require('path');
@@ -56,7 +56,7 @@ const path = require('path');
assert.ok(fs.existsSync(kept.file), 'a named file stays');
assert.ok(fs.existsSync(kept.file), 'a named file stays');
assert.ok(fs.existsSync(waiting.file), 'a busy object\'s file stays');
assert.ok(fs.existsSync(waiting.file), 'a busy object\'s file stays');
assert.strictEqual(cache.orphansRemoved(), 2);
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');
+ assert.strictEqual((await cache.refresh()).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.
// 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'");
await db.run("UPDATE media_jobs SET status = 'failed' WHERE id = 'job_busy'");