openvibers/openvibe.media Commits
VOD storage: make the B2 drain self-monitoring — free-space budget, fast re-runs, bounded uploads, stall alerts
Changes
@@@ -258,3 +258,16 @@ idempotent (`INSERT OR IGNORE`). After the copy, runstream registry; apps re-trigger ingest on their side.- Comments on VODs/clips, notifications, permissions/moderation ranks: ownedby the apps.++++## VOD storage tiering & disk safety++++`server/vod/vod-storage.js` keeps the local VOD volume healthy without anyone watching it:++++- **Drain policy** — a drain starts when the disk is above `hotDiskPressurePct` (70%) **or** has less than `minFreeGb` (25 GB) free, and stops only when it is under `localLowWaterPct` (60%) **and** has `targetFreeGb` (40 GB) free. Above `criticalDiskPct` (90%) recordings finished more than `criticalMinAgeHours` (2 h) ago are eligible; otherwise 1 day. Least-recently-watched VODs go first. All values live in `media_settings` as `storage_tier.*` (`PUT /api/v1/:app/admin/storage/tiers/settings`).++- **Only real files are candidates.** Rows without a file (a recording that was refused or never produced output, legacy paths) are quarantined as `health_status='missing_file'` and excluded — they used to fill the candidate list and starve the drain.++- **Cadence** — while a drain is still needed the sweep re-runs every `pressureRetryMs` (2 min), otherwise every `sweepIntervalMs` (15 min). A VOD whose upload failed is skipped for 30 min so one bad file cannot pin the drain.++- **Bounded uploads** — each upload is aborted past `uploadTimeoutFloorMs + size / uploadMinThroughputMBps`; a watchdog logs a sweep that has run for hours.++- **Stall alerts** — after `alertAfterStalledPasses` (2) consecutive pressure passes that freed nothing, and whenever the disk is critical, Media sends `storage.alert` to every app webhook (Live logs it and forwards to `OPS_ALERT_WEBHOOK_URL` / the `ops_alert_webhook_url` setting); `storage.recovered` follows once the drain works again. Each pass under pressure logs candidates / uploaded / failed / backed-off / quarantined.++- **Status** — `GET /api/v1/:app/admin/storage` includes `sweep` (last result, next run, stalled flag, needsDrain/critical).++++Run `npm test` for the policy regression tests.@@@ -6,7 +6,8 @@"scripts": {"start": "node server/index.js","init-db": "node server/db/init.js",-- "smoke": "bash scripts/smoke-test.sh"++ "smoke": "bash scripts/smoke-test.sh",++ "test": "for f in test/*.test.js; do node \"$f\" || exit 1; done"},"license": "MIT","dependencies": {@@@ -85,6 +85,23 @@ const DEFAULTS = {// beats keeping yesterday's VOD hot.criticalDiskPct: 90,criticalMinAgeHours: 2,++ // Free-space budget, in GB, evaluated alongside the percentages. Percentages++ // alone scale badly: 30% of a 96 GB disk is one long stream, and the recorder's++ // own guard rails are in GB (VOD_DISK_WARN_GB / VOD_DISK_CRIT_GB). Drain whenever++ // free space is below minFreeGb and keep going until targetFreeGb is free again.++ minFreeGb: 25,++ targetFreeGb: 40,++ // While a drain is still needed after a pass, re-run this soon instead of waiting++ // for the regular interval — recovery from a full disk should take minutes.++ pressureRetryMs: 2 * 60 * 1000,++ // Upload deadline = floor + size / minimum acceptable throughput; a stalled++ // multipart upload is aborted instead of holding the sweep lock forever.++ uploadTimeoutFloorMs: 20 * 60 * 1000,++ uploadMinThroughputMBps: 2,++ // Raise storage.alert after this many consecutive pressure passes that freed++ // nothing (per-kind cooldown so a stuck night doesn't page every 2 minutes).++ alertAfterStalledPasses: 2,++ alertCooldownMs: 6 * 60 * 60 * 1000,maxPerSweep: 40,// R2 popularity tierr2Enabled: true,@@@ -175,10 +192,23 @@ async function headObject(provider, key) {}}++/** Deadline for one upload: a floor plus the time the file takes at the minimum acceptable throughput. */++function uploadTimeoutMs(bytes, settings = getSettings()) {++ const floor = Number(settings.uploadTimeoutFloorMs) || DEFAULTS.uploadTimeoutFloorMs;++ const mbps = Number(settings.uploadMinThroughputMBps) || DEFAULTS.uploadMinThroughputMBps;++ return floor + Math.ceil((Number(bytes) || 0) / (mbps * 1024 * 1024)) * 1000;++}++async function uploadFile(provider, key, filePath, contentType = 'video/webm') {const client = clientFor(provider);if (!client) throw new Error(`Provider ${provider} not configured`);loadSdk();++ const size = fs.statSync(filePath).size;++ // A hung multipart upload used to hold the sweep lock indefinitely (nothing else++ // 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();++ 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,params: {@@@ -190,8 +220,16 @@ async function uploadFile(provider, key, filePath, contentType = 'video/webm') {partSize: 64 * 1024 * 1024,queueSize: 3,leavePartsOnError: false,++ abortController: abort,});-- await upload.done();++ try {++ await upload.done();++ } catch (err) {++ if (abort.signal.aborted) throw (abort.signal.reason instanceof Error ? abort.signal.reason : new Error('upload aborted'));++ throw err;++ } finally {++ clearTimeout(deadline);++ }// Verify size before anything destructive happensconst localSize = fs.statSync(filePath).size;const head = await headObject(provider, key);@@@ -609,6 +647,71 @@ function reconcileGhosts() {const OFFLOAD_RETRY_BACKOFF_MS = 30 * 60 * 1000;const offloadFailedAt = new Map(); // vodId → epoch ms of the last failed upload++// ── Drain policy (pure — unit-tested) ────────────────────────++++const GB = 1024 * 1024 * 1024;++++/** Drain when either the percentage ceiling or the free-space floor is crossed. */++function needsDrain(disk, settings) {++ if (!disk || !disk.total) return false;++ const freeGb = disk.available / GB;++ return disk.usePct >= settings.hotDiskPressurePct || freeGb < settings.minFreeGb;++}++++/** A drain may stop only when BOTH the percentage and the free-space target are met. */++function drainSatisfied(disk, settings) {++ if (!disk || !disk.total) return true;++ const freeGb = disk.available / GB;++ return disk.usePct <= settings.localLowWaterPct && freeGb >= settings.targetFreeGb;++}++++function isCritical(disk, settings) {++ return !!disk && !!disk.total && disk.usePct >= settings.criticalDiskPct;++}++++/**++ * When to run the next pass: soon while a drain is still needed (progress was made,++ * or candidates are merely in retry back-off), the regular interval otherwise.++ */++function planNextDelayMs(result, settings) {++ if (result && result.stillNeedsDrain && (result.migrated || result.skippedBackoff || result.errors)) {++ return settings.pressureRetryMs;++ }++ return settings.sweepIntervalMs;++}++++// ── Sweep state + alerts ─────────────────────────────────────++++const sweepState = {++ lastRunAt: null,++ lastResult: null,++ nextRunAt: null,++ startedAt: null,++ stalledPasses: 0, // consecutive pressure passes that freed nothing++ stalled: false,++};++const alertLastSentAt = new Map(); // kind → epoch ms++++/**++ * Tell every app with a webhook that storage needs a human (or that it recovered).++ * Delivered as the regular Media→app webhook (`storage.alert` / `storage.recovered`),++ * so Live can log it loudly and page whatever ops channel it has configured.++ */++async function emitStorageEvent(event, kind, data, settings = getSettings()) {++ const cooldownKey = `${event}:${kind}`;++ const last = alertLastSentAt.get(cooldownKey) || 0;++ if (event === 'storage.alert' && Date.now() - last < settings.alertCooldownMs) return false;++ alertLastSentAt.set(cooldownKey, Date.now());++ const payload = { kind, ...data, at: new Date().toISOString() };++ (event === 'storage.alert' ? console.error : console.warn)(`[VodStorage] ${event} (${kind}): ${JSON.stringify(data)}`);++ let apps = [];++ try { apps = db.listApps().filter(a => a.webhook_url); } catch { apps = []; }++ let webhooks;++ try { webhooks = require('../webhooks'); } catch { return false; }++ await Promise.all(apps.map(app => webhooks.sendWebhook(app.app_id, event, payload).catch(() => false)));++ return true;++}++async function runSweep() {if (sweepRunning) return { skipped: true, reason: 'already running' };sweepRunning = true;@@@ -622,16 +725,17 @@ async function runSweep() {let migrated = 0, bytesFreed = 0, promoted = 0, demoted = 0;const errors = [];++ sweepState.startedAt = Date.now();const disk = diskUsage(config.vod.path);-- const underPressure = disk.usePct >= settings.hotDiskPressurePct;-- const critical = disk.usePct >= settings.criticalDiskPct;++ const underPressure = needsDrain(disk, settings);++ const critical = isCritical(disk, settings);const ghosts = reconcileGhosts();// 1) Cold offload to B2let candidates;if (underPressure) {const minAge = critical ? `-${settings.criticalMinAgeHours} hours` : '-1 day';-- console.log(`[VodStorage] Disk at ${disk.usePct}% (${critical ? 'CRITICAL ≥ ' + settings.criticalDiskPct : 'pressure ≥ ' + settings.hotDiskPressurePct}%) — draining to ${settings.localLowWaterPct}%`++ console.log(`[VodStorage] Disk at ${disk.usePct}%, ${(disk.available / GB).toFixed(1)} GB free (${critical ? 'CRITICAL' : 'pressure'}) — draining to ≤${settings.localLowWaterPct}% / ≥${settings.targetFreeGb} GB free`+ (ghosts ? `; quarantined ${ghosts} file-less VOD row(s)` : ''));candidates = db.all(`SELECT id, file_path, file_size FROM vods@@@ -655,7 +759,7 @@ async function runSweep() {let skippedBackoff = 0, quarantined = ghosts;for (const vod of candidates) {-- if (underPressure && diskUsage(config.vod.path).usePct <= settings.localLowWaterPct) break;++ if (underPressure && drainSatisfied(diskUsage(config.vod.path), settings)) break;const failedAt = offloadFailedAt.get(vod.id);if (failedAt && Date.now() - failedAt < OFFLOAD_RETRY_BACKOFF_MS) { skippedBackoff++; continue; }// Verify the file before spending an upload slot on it; a row whose file is@@@ -693,6 +797,46 @@ async function runSweep() {}}++ // Stall detection: a pressure pass that freed nothing is the failure mode that++ // went unnoticed for weeks. Count them, and after a couple in a row raise an++ // alert through the app webhooks; announce recovery once the drain works again.++ const diskAfter = diskUsage(config.vod.path);++ const stillNeedsDrain = underPressure && !drainSatisfied(diskAfter, settings);++ if (underPressure && !migrated) {++ sweepState.stalledPasses++;++ if (sweepState.stalledPasses >= settings.alertAfterStalledPasses) {++ sweepState.stalled = true;++ await emitStorageEvent('storage.alert', 'drain_stalled', {++ disk_pct: diskAfter.usePct,++ free_gb: Number((diskAfter.available / GB).toFixed(1)),++ stalled_passes: sweepState.stalledPasses,++ candidates: candidates.length,++ errors: errors.slice(0, 3),++ hint: !candidates.length++ ? 'nothing eligible to offload (stuck recordings? non-VOD files in the VOD directory?)'++ : errors.length ? 'uploads failing — check B2 credentials/bucket and network' : 'all candidates in retry back-off',++ }, settings);++ }++ } else {++ if (sweepState.stalled) {++ await emitStorageEvent('storage.recovered', 'drain_recovered', {++ disk_pct: diskAfter.usePct,++ free_gb: Number((diskAfter.available / GB).toFixed(1)),++ uploaded: migrated,++ }, settings);++ }++ sweepState.stalledPasses = 0;++ sweepState.stalled = false;++ }++ if (critical && stillNeedsDrain) {++ await emitStorageEvent('storage.alert', 'disk_critical', {++ disk_pct: diskAfter.usePct,++ free_gb: Number((diskAfter.available / GB).toFixed(1)),++ uploaded_this_pass: migrated,++ hint: 'recorder refuses new recordings below VOD_DISK_CRIT_GB; drain in progress',++ }, settings);++ }++// 2) R2 promotion for popular VODsif (settings.r2Enabled && providerConfigured('r2') && providerHealthy.r2 !== false) {const popular = db.all(`@@@ -730,9 +874,17 @@ async function runSweep() {demoted,bytesFreed,underPressure,++ critical,++ stillNeedsDrain,++ skippedBackoff,++ quarantined,++ diskPct: diskAfter.usePct,++ freeGb: Number((diskAfter.available / GB).toFixed(1)),errors: errors.length ? errors : undefined,timestamp: new Date().toISOString(),};++ sweepState.lastRunAt = Date.now();++ sweepState.lastResult = summary;if (migrated || promoted || demoted) {console.log(`[VodStorage] Sweep: ${migrated} → B2, ${promoted} → R2, ${demoted} R2→B2, ${(bytesFreed / 1048576).toFixed(1)} MB freed locally`);}@@@ -742,6 +894,7 @@ async function runSweep() {return { error: err.message };} finally {sweepRunning = false;++ sweepState.startedAt = null;}}@@@ -812,6 +965,32 @@ async function migrateLegacy() {console.log(`[VodStorage] Legacy migration: ${flipped} cold VOD(s) mapped to B2${restoredLocal ? `, ${restoredLocal} still local` : ''}`);}++// The sweep is a self-rescheduling chain, not a fixed interval: after a pass that++// left the disk still needing a drain it comes back in pressureRetryMs, otherwise in++// sweepIntervalMs. A sweep that overran its deadline is reported by the watchdog++// (the upload abort in uploadFile is what actually frees it).++const SWEEP_WATCHDOG_MS = 3 * 60 * 60 * 1000;++++function scheduleNext(delayMs) {++ if (sweepTimer) { clearTimeout(sweepTimer); sweepTimer = null; }++ sweepState.nextRunAt = Date.now() + delayMs;++ sweepTimer = setTimeout(async () => {++ sweepTimer = null;++ let result = null;++ try {++ result = await runSweep();++ } catch (err) {++ console.error('[VodStorage] Sweep failed:', err.message);++ }++ if (result && result.skipped && result.reason === 'already running'++ && sweepState.startedAt && Date.now() - sweepState.startedAt > SWEEP_WATCHDOG_MS) {++ console.error(`[VodStorage] Watchdog: a sweep has been running for ${Math.round((Date.now() - sweepState.startedAt) / 60000)} min — an upload is probably hung`);++ }++ scheduleNext(planNextDelayMs(result, getSettings()));++ }, delayMs);++ if (sweepTimer.unref) sweepTimer.unref();++}++function start() {stop();const settings = getSettings();@@@ -819,18 +998,13 @@ function start() {console.log('[VodStorage] Disabled — not starting sweep timer');return;}-- console.log(`[VodStorage] Starting sweep timer (every ${(settings.sweepIntervalMs / 60000).toFixed(0)} min)`);-- setTimeout(() => {-- runSweep().catch(err => console.error('[VodStorage] Sweep failed:', err.message));-- sweepTimer = setInterval(() => {-- runSweep().catch(err => console.error('[VodStorage] Sweep failed:', err.message));-- }, settings.sweepIntervalMs);-- if (sweepTimer.unref) sweepTimer.unref();-- }, 30_000).unref?.();++ console.log(`[VodStorage] Starting sweep (every ${(settings.sweepIntervalMs / 60000).toFixed(0)} min, every ${(settings.pressureRetryMs / 60000).toFixed(0)} min while draining)`);++ scheduleNext(30_000);}function stop() {-- if (sweepTimer) { clearInterval(sweepTimer); sweepTimer = null; }++ if (sweepTimer) { clearTimeout(sweepTimer); sweepTimer = null; }++ sweepState.nextRunAt = null;}// ── Status ───────────────────────────────────────────────────@@@ -987,11 +1161,28 @@ function getStatus() {r2: { count: clipCounts.r2Count || 0 },},sweepRunning,++ sweep: {++ running: sweepRunning,++ startedAt: sweepState.startedAt ? new Date(sweepState.startedAt).toISOString() : null,++ lastRunAt: sweepState.lastRunAt ? new Date(sweepState.lastRunAt).toISOString() : null,++ nextRunAt: sweepState.nextRunAt ? new Date(sweepState.nextRunAt).toISOString() : null,++ lastResult: sweepState.lastResult,++ stalled: sweepState.stalled,++ stalledPasses: sweepState.stalledPasses,++ needsDrain: needsDrain(localDisk, settings),++ critical: isCritical(localDisk, settings),++ },};}module.exports = {DEFAULTS,++ OFFLOADABLE_WHERE,++ needsDrain,++ drainSatisfied,++ isCritical,++ planNextDelayMs,++ uploadTimeoutMs,providerOf,isRemote,keyForVod,nnew file mode 100644@@@ -0,0 +1,77 @@++'use strict';++++// Regression tests for the VOD offload policy in server/vod/vod-storage.js.++//++// Background (2026-08-29): the disk sat at 96% for weeks while the sweep logged++// "draining" every 15 minutes and never uploaded a byte, because ~350 VOD rows with++// no file matched the candidate query and filled its LIMIT ahead of the real++// recordings. These tests pin the pieces that prevent a repeat: candidate++// selection, the free-space budget, the re-run cadence, and the upload deadline.++++const assert = require('assert');++const Database = require('better-sqlite3');++++process.env.MEDIA_B2_ENDPOINT = process.env.MEDIA_B2_ENDPOINT || '';++const storage = require('../server/vod/vod-storage');++const { DEFAULTS, OFFLOADABLE_WHERE, needsDrain, drainSatisfied, isCritical, planNextDelayMs, uploadTimeoutMs } = storage;++++const GB = 1024 * 1024 * 1024;++const disk = (usePct, freeGb, totalGb = 96) => ({ total: totalGb * GB, used: (totalGb - freeGb) * GB, available: freeGb * GB, usePct, mount: '/' });++++// ── Candidate selection: only rows with a real, non-quarantined local file ──++const db = new Database(':memory:');++db.exec(`CREATE TABLE vods (++ id INTEGER PRIMARY KEY, file_path TEXT, file_size INTEGER, storage_provider TEXT,++ is_recording INTEGER, health_status TEXT, view_count INTEGER, last_accessed_at TEXT,++ created_at TEXT DEFAULT (datetime('now', '-2 days'))++)`);++const ins = db.prepare('INSERT INTO vods (id, file_path, file_size, storage_provider, is_recording, health_status) VALUES (?, ?, ?, ?, ?, ?)');++ins.run(1, '/data/vods/real-a.mp4', 15 * GB, 'local', 0, 'ok'); // real recording++ins.run(2, '/data/vods/real-b.webm', 2 * GB, null, 0, null); // real, provider/health NULL (legacy rows)++ins.run(3, null, 0, 'local', 0, null); // ghost: never got a file++ins.run(4, '/opt/hobostreamer/data/vods/old.webm', 0, 'local', 0, 'missing_file'); // quarantined legacy path++ins.run(5, '/data/vods/zero.webm', 0, 'local', 0, 'zero_byte'); // quarantined zero-byte++ins.run(6, '/data/vods/live.mp4', 1 * GB, 'local', 1, 'ok'); // still recording++ins.run(7, '/data/vods/cold.mp4', 8 * GB, 'b2', 0, 'ok'); // already offloaded++++const picked = db.prepare(`SELECT id FROM vods WHERE ${OFFLOADABLE_WHERE} ORDER BY id`).all().map(r => r.id);++assert.deepStrictEqual(picked, [1, 2], `only real local finished files are offload candidates, got ${picked}`);++console.log('✅ ghost / quarantined / recording / offloaded rows are never offload candidates');++++// ── Free-space budget: percent OR GB floor triggers, both must be satisfied to stop ──++assert.strictEqual(needsDrain(disk(96, 4), DEFAULTS), true, '96% full must drain');++assert.strictEqual(needsDrain(disk(50, 48), DEFAULTS), false, 'half-empty disk is fine');++assert.strictEqual(needsDrain(disk(65, 20), DEFAULTS), true, 'below the percent ceiling but under the GB floor still drains');++assert.strictEqual(needsDrain(disk(75, 400, 1600), DEFAULTS), true, 'percent ceiling applies even with lots of absolute space');++assert.strictEqual(needsDrain({ total: 0, available: 0, usePct: 0 }, DEFAULTS), false, 'unknown disk never drains');++++assert.strictEqual(drainSatisfied(disk(60, 38), DEFAULTS), false, 'at the percent low-water but under the GB target keeps draining');++assert.strictEqual(drainSatisfied(disk(55, 43), DEFAULTS), true, 'both targets met stops the drain');++assert.strictEqual(drainSatisfied(disk(70, 500, 1600), DEFAULTS), false, 'above the percent low-water keeps draining on big disks');++++assert.strictEqual(isCritical(disk(90, 9), DEFAULTS), true);++assert.strictEqual(isCritical(disk(89, 10), DEFAULTS), false);++console.log('✅ free-space budget: percent ceiling OR GB floor start a drain; both targets end it');++++// ── Cadence: come back quickly while a drain is making (or could make) progress ──++assert.strictEqual(planNextDelayMs({ stillNeedsDrain: true, migrated: 3 }, DEFAULTS), DEFAULTS.pressureRetryMs, 'progress under pressure → soon');++assert.strictEqual(planNextDelayMs({ stillNeedsDrain: true, migrated: 0, skippedBackoff: 2 }, DEFAULTS), DEFAULTS.pressureRetryMs, 'candidates in back-off → soon');++assert.strictEqual(planNextDelayMs({ stillNeedsDrain: true, migrated: 0, errors: [{ id: 1 }] }, DEFAULTS), DEFAULTS.pressureRetryMs, 'failed uploads → retry soon');++assert.strictEqual(planNextDelayMs({ stillNeedsDrain: true, migrated: 0 }, DEFAULTS), DEFAULTS.sweepIntervalMs, 'nothing eligible → no point hammering');++assert.strictEqual(planNextDelayMs({ stillNeedsDrain: false, migrated: 5 }, DEFAULTS), DEFAULTS.sweepIntervalMs, 'drain satisfied → regular interval');++assert.strictEqual(planNextDelayMs(null, DEFAULTS), DEFAULTS.sweepIntervalMs, 'skipped/failed sweep → regular interval');++assert.ok(DEFAULTS.pressureRetryMs < DEFAULTS.sweepIntervalMs);++console.log('✅ sweep re-runs within minutes while a drain is still needed');++++// ── Upload deadline scales with size so a hung 15 GB multipart cannot hold the sweep forever ──++const t100mb = uploadTimeoutMs(100 * 1024 * 1024, DEFAULTS);++const t15gb = uploadTimeoutMs(15 * GB, DEFAULTS);++assert.ok(t100mb >= DEFAULTS.uploadTimeoutFloorMs && t100mb < DEFAULTS.uploadTimeoutFloorMs + 2 * 60 * 1000, 'small file ≈ floor');++assert.ok(t15gb > 2 * 60 * 60 * 1000 && t15gb < 4 * 60 * 60 * 1000, `15 GB at 2 MB/s ≈ 2–3 h, got ${Math.round(t15gb / 60000)} min`);++console.log('✅ upload deadline = floor + size / minimum throughput');++++// ── Status surface exists for dashboards / health checks ──++assert.ok(typeof storage.getStatus === 'function');++console.log('✅ policy helpers exported for status reporting');++++console.log('\n✅ All VOD storage policy tests passed');
@@ -258,3 +258,16 @@ idempotent (`INSERT OR IGNORE`). After the copy, run
@@ -258,3 +258,16 @@ idempotent (`INSERT OR IGNORE`). After the copy, run
stream registry; apps re-trigger ingest on their side.
stream registry; apps re-trigger ingest on their side.
- Comments on VODs/clips, notifications, permissions/moderation ranks: owned
- Comments on VODs/clips, notifications, permissions/moderation ranks: owned
by the apps.
by the apps.
+
+## VOD storage tiering & disk safety
+
+`server/vod/vod-storage.js` keeps the local VOD volume healthy without anyone watching it:
+
+- **Drain policy** — a drain starts when the disk is above `hotDiskPressurePct` (70%) **or** has less than `minFreeGb` (25 GB) free, and stops only when it is under `localLowWaterPct` (60%) **and** has `targetFreeGb` (40 GB) free. Above `criticalDiskPct` (90%) recordings finished more than `criticalMinAgeHours` (2 h) ago are eligible; otherwise 1 day. Least-recently-watched VODs go first. All values live in `media_settings` as `storage_tier.*` (`PUT /api/v1/:app/admin/storage/tiers/settings`).
+- **Only real files are candidates.** Rows without a file (a recording that was refused or never produced output, legacy paths) are quarantined as `health_status='missing_file'` and excluded — they used to fill the candidate list and starve the drain.
+- **Cadence** — while a drain is still needed the sweep re-runs every `pressureRetryMs` (2 min), otherwise every `sweepIntervalMs` (15 min). A VOD whose upload failed is skipped for 30 min so one bad file cannot pin the drain.
+- **Bounded uploads** — each upload is aborted past `uploadTimeoutFloorMs + size / uploadMinThroughputMBps`; a watchdog logs a sweep that has run for hours.
+- **Stall alerts** — after `alertAfterStalledPasses` (2) consecutive pressure passes that freed nothing, and whenever the disk is critical, Media sends `storage.alert` to every app webhook (Live logs it and forwards to `OPS_ALERT_WEBHOOK_URL` / the `ops_alert_webhook_url` setting); `storage.recovered` follows once the drain works again. Each pass under pressure logs candidates / uploaded / failed / backed-off / quarantined.
+- **Status** — `GET /api/v1/:app/admin/storage` includes `sweep` (last result, next run, stalled flag, needsDrain/critical).
+
+Run `npm test` for the policy regression tests.
@@ -6,7 +6,8 @@
@@ -6,7 +6,8 @@
"scripts": {
"scripts": {
"start": "node server/index.js",
"start": "node server/index.js",
"init-db": "node server/db/init.js",
"init-db": "node server/db/init.js",
- "smoke": "bash scripts/smoke-test.sh"
+ "smoke": "bash scripts/smoke-test.sh",
+ "test": "for f in test/*.test.js; do node \"$f\" || exit 1; done"
},
},
"license": "MIT",
"license": "MIT",
"dependencies": {
"dependencies": {
@@ -85,6 +85,23 @@ const DEFAULTS = {
@@ -85,6 +85,23 @@ const DEFAULTS = {
// beats keeping yesterday's VOD hot.
// beats keeping yesterday's VOD hot.
criticalDiskPct: 90,
criticalDiskPct: 90,
criticalMinAgeHours: 2,
criticalMinAgeHours: 2,
+ // Free-space budget, in GB, evaluated alongside the percentages. Percentages
+ // alone scale badly: 30% of a 96 GB disk is one long stream, and the recorder's
+ // own guard rails are in GB (VOD_DISK_WARN_GB / VOD_DISK_CRIT_GB). Drain whenever
+ // free space is below minFreeGb and keep going until targetFreeGb is free again.
+ minFreeGb: 25,
+ targetFreeGb: 40,
+ // While a drain is still needed after a pass, re-run this soon instead of waiting
+ // for the regular interval — recovery from a full disk should take minutes.
+ pressureRetryMs: 2 * 60 * 1000,
+ // Upload deadline = floor + size / minimum acceptable throughput; a stalled
+ // multipart upload is aborted instead of holding the sweep lock forever.
+ uploadTimeoutFloorMs: 20 * 60 * 1000,
+ uploadMinThroughputMBps: 2,
+ // Raise storage.alert after this many consecutive pressure passes that freed
+ // nothing (per-kind cooldown so a stuck night doesn't page every 2 minutes).
+ alertAfterStalledPasses: 2,
+ alertCooldownMs: 6 * 60 * 60 * 1000,
maxPerSweep: 40,
maxPerSweep: 40,
// R2 popularity tier
// R2 popularity tier
r2Enabled: true,
r2Enabled: true,
@@ -175,10 +192,23 @@ async function headObject(provider, key) {
@@ -175,10 +192,23 @@ async function headObject(provider, key) {
}
}
}
}
+/** Deadline for one upload: a floor plus the time the file takes at the minimum acceptable throughput. */
+function uploadTimeoutMs(bytes, settings = getSettings()) {
+ const floor = Number(settings.uploadTimeoutFloorMs) || DEFAULTS.uploadTimeoutFloorMs;
+ const mbps = Number(settings.uploadMinThroughputMBps) || DEFAULTS.uploadMinThroughputMBps;
+ 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') {
const client = clientFor(provider);
const client = clientFor(provider);
if (!client) throw new Error(`Provider ${provider} not configured`);
if (!client) throw new Error(`Provider ${provider} not configured`);
loadSdk();
loadSdk();
+ const size = fs.statSync(filePath).size;
+ // A hung multipart upload used to hold the sweep lock indefinitely (nothing else
+ // 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();
+ 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({
const upload = new LibStorage.Upload({
client,
client,
params: {
params: {
@@ -190,8 +220,16 @@ async function uploadFile(provider, key, filePath, contentType = 'video/webm') {
@@ -190,8 +220,16 @@ async function uploadFile(provider, key, filePath, contentType = 'video/webm') {
partSize: 64 * 1024 * 1024,
partSize: 64 * 1024 * 1024,
queueSize: 3,
queueSize: 3,
leavePartsOnError: false,
leavePartsOnError: false,
+ abortController: abort,
});
});
- await upload.done();
+ try {
+ await upload.done();
+ } catch (err) {
+ if (abort.signal.aborted) throw (abort.signal.reason instanceof Error ? abort.signal.reason : new Error('upload aborted'));
+ throw err;
+ } finally {
+ clearTimeout(deadline);
+ }
// Verify size before anything destructive happens
// Verify size before anything destructive happens
const localSize = fs.statSync(filePath).size;
const localSize = fs.statSync(filePath).size;
const head = await headObject(provider, key);
const head = await headObject(provider, key);
@@ -609,6 +647,71 @@ function reconcileGhosts() {
@@ -609,6 +647,71 @@ function reconcileGhosts() {
const OFFLOAD_RETRY_BACKOFF_MS = 30 * 60 * 1000;
const OFFLOAD_RETRY_BACKOFF_MS = 30 * 60 * 1000;
const offloadFailedAt = new Map(); // vodId → epoch ms of the last failed upload
const offloadFailedAt = new Map(); // vodId → epoch ms of the last failed upload
+// ── Drain policy (pure — unit-tested) ────────────────────────
+
+const GB = 1024 * 1024 * 1024;
+
+/** Drain when either the percentage ceiling or the free-space floor is crossed. */
+function needsDrain(disk, settings) {
+ if (!disk || !disk.total) return false;
+ const freeGb = disk.available / GB;
+ return disk.usePct >= settings.hotDiskPressurePct || freeGb < settings.minFreeGb;
+}
+
+/** A drain may stop only when BOTH the percentage and the free-space target are met. */
+function drainSatisfied(disk, settings) {
+ if (!disk || !disk.total) return true;
+ const freeGb = disk.available / GB;
+ return disk.usePct <= settings.localLowWaterPct && freeGb >= settings.targetFreeGb;
+}
+
+function isCritical(disk, settings) {
+ return !!disk && !!disk.total && disk.usePct >= settings.criticalDiskPct;
+}
+
+/**
+ * When to run the next pass: soon while a drain is still needed (progress was made,
+ * or candidates are merely in retry back-off), the regular interval otherwise.
+ */
+function planNextDelayMs(result, settings) {
+ if (result && result.stillNeedsDrain && (result.migrated || result.skippedBackoff || result.errors)) {
+ return settings.pressureRetryMs;
+ }
+ return settings.sweepIntervalMs;
+}
+
+// ── Sweep state + alerts ─────────────────────────────────────
+
+const sweepState = {
+ lastRunAt: null,
+ lastResult: null,
+ nextRunAt: null,
+ startedAt: null,
+ stalledPasses: 0, // consecutive pressure passes that freed nothing
+ stalled: false,
+};
+const alertLastSentAt = new Map(); // kind → epoch ms
+
+/**
+ * Tell every app with a webhook that storage needs a human (or that it recovered).
+ * Delivered as the regular Media→app webhook (`storage.alert` / `storage.recovered`),
+ * so Live can log it loudly and page whatever ops channel it has configured.
+ */
+async function emitStorageEvent(event, kind, data, settings = getSettings()) {
+ const cooldownKey = `${event}:${kind}`;
+ const last = alertLastSentAt.get(cooldownKey) || 0;
+ if (event === 'storage.alert' && Date.now() - last < settings.alertCooldownMs) return false;
+ alertLastSentAt.set(cooldownKey, Date.now());
+ const payload = { kind, ...data, at: new Date().toISOString() };
+ (event === 'storage.alert' ? console.error : console.warn)(`[VodStorage] ${event} (${kind}): ${JSON.stringify(data)}`);
+ let apps = [];
+ try { apps = db.listApps().filter(a => a.webhook_url); } catch { apps = []; }
+ let webhooks;
+ try { webhooks = require('../webhooks'); } catch { return false; }
+ await Promise.all(apps.map(app => webhooks.sendWebhook(app.app_id, event, payload).catch(() => false)));
+ return true;
+}
+
async function runSweep() {
async function runSweep() {
if (sweepRunning) return { skipped: true, reason: 'already running' };
if (sweepRunning) return { skipped: true, reason: 'already running' };
sweepRunning = true;
sweepRunning = true;
@@ -622,16 +725,17 @@ async function runSweep() {
@@ -622,16 +725,17 @@ async function runSweep() {
let migrated = 0, bytesFreed = 0, promoted = 0, demoted = 0;
let migrated = 0, bytesFreed = 0, promoted = 0, demoted = 0;
const errors = [];
const errors = [];
+ sweepState.startedAt = Date.now();
const disk = diskUsage(config.vod.path);
const disk = diskUsage(config.vod.path);
- const underPressure = disk.usePct >= settings.hotDiskPressurePct;
- const critical = disk.usePct >= settings.criticalDiskPct;
+ const underPressure = needsDrain(disk, settings);
+ const critical = isCritical(disk, settings);
const ghosts = reconcileGhosts();
const ghosts = reconcileGhosts();
// 1) Cold offload to B2
// 1) Cold offload to B2
let candidates;
let candidates;
if (underPressure) {
if (underPressure) {
const minAge = critical ? `-${settings.criticalMinAgeHours} hours` : '-1 day';
const minAge = critical ? `-${settings.criticalMinAgeHours} hours` : '-1 day';
- console.log(`[VodStorage] Disk at ${disk.usePct}% (${critical ? 'CRITICAL ≥ ' + settings.criticalDiskPct : 'pressure ≥ ' + settings.hotDiskPressurePct}%) — draining to ${settings.localLowWaterPct}%`
+ console.log(`[VodStorage] Disk at ${disk.usePct}%, ${(disk.available / GB).toFixed(1)} GB free (${critical ? 'CRITICAL' : 'pressure'}) — draining to ≤${settings.localLowWaterPct}% / ≥${settings.targetFreeGb} GB free`
+ (ghosts ? `; quarantined ${ghosts} file-less VOD row(s)` : ''));
+ (ghosts ? `; quarantined ${ghosts} file-less VOD row(s)` : ''));
candidates = db.all(`
candidates = db.all(`
SELECT id, file_path, file_size FROM vods
SELECT id, file_path, file_size FROM vods
@@ -655,7 +759,7 @@ async function runSweep() {
@@ -655,7 +759,7 @@ async function runSweep() {
let skippedBackoff = 0, quarantined = ghosts;
let skippedBackoff = 0, quarantined = ghosts;
for (const vod of candidates) {
for (const vod of candidates) {
- if (underPressure && diskUsage(config.vod.path).usePct <= settings.localLowWaterPct) break;
+ if (underPressure && drainSatisfied(diskUsage(config.vod.path), settings)) break;
const failedAt = offloadFailedAt.get(vod.id);
const failedAt = offloadFailedAt.get(vod.id);
if (failedAt && Date.now() - failedAt < OFFLOAD_RETRY_BACKOFF_MS) { skippedBackoff++; continue; }
if (failedAt && Date.now() - failedAt < OFFLOAD_RETRY_BACKOFF_MS) { skippedBackoff++; continue; }
// Verify the file before spending an upload slot on it; a row whose file is
// Verify the file before spending an upload slot on it; a row whose file is
@@ -693,6 +797,46 @@ async function runSweep() {
@@ -693,6 +797,46 @@ async function runSweep() {
}
}
}
}
+ // Stall detection: a pressure pass that freed nothing is the failure mode that
+ // went unnoticed for weeks. Count them, and after a couple in a row raise an
+ // alert through the app webhooks; announce recovery once the drain works again.
+ const diskAfter = diskUsage(config.vod.path);
+ const stillNeedsDrain = underPressure && !drainSatisfied(diskAfter, settings);
+ if (underPressure && !migrated) {
+ sweepState.stalledPasses++;
+ if (sweepState.stalledPasses >= settings.alertAfterStalledPasses) {
+ sweepState.stalled = true;
+ await emitStorageEvent('storage.alert', 'drain_stalled', {
+ disk_pct: diskAfter.usePct,
+ free_gb: Number((diskAfter.available / GB).toFixed(1)),
+ stalled_passes: sweepState.stalledPasses,
+ candidates: candidates.length,
+ errors: errors.slice(0, 3),
+ hint: !candidates.length
+ ? 'nothing eligible to offload (stuck recordings? non-VOD files in the VOD directory?)'
+ : errors.length ? 'uploads failing — check B2 credentials/bucket and network' : 'all candidates in retry back-off',
+ }, settings);
+ }
+ } else {
+ if (sweepState.stalled) {
+ await emitStorageEvent('storage.recovered', 'drain_recovered', {
+ disk_pct: diskAfter.usePct,
+ free_gb: Number((diskAfter.available / GB).toFixed(1)),
+ uploaded: migrated,
+ }, settings);
+ }
+ sweepState.stalledPasses = 0;
+ sweepState.stalled = false;
+ }
+ if (critical && stillNeedsDrain) {
+ await emitStorageEvent('storage.alert', 'disk_critical', {
+ disk_pct: diskAfter.usePct,
+ free_gb: Number((diskAfter.available / GB).toFixed(1)),
+ uploaded_this_pass: migrated,
+ hint: 'recorder refuses new recordings below VOD_DISK_CRIT_GB; drain in progress',
+ }, settings);
+ }
+
// 2) R2 promotion for popular VODs
// 2) R2 promotion for popular VODs
if (settings.r2Enabled && providerConfigured('r2') && providerHealthy.r2 !== false) {
if (settings.r2Enabled && providerConfigured('r2') && providerHealthy.r2 !== false) {
const popular = db.all(`
const popular = db.all(`
@@ -730,9 +874,17 @@ async function runSweep() {
@@ -730,9 +874,17 @@ async function runSweep() {
demoted,
demoted,
bytesFreed,
bytesFreed,
underPressure,
underPressure,
+ critical,
+ stillNeedsDrain,
+ skippedBackoff,
+ quarantined,
+ diskPct: diskAfter.usePct,
+ freeGb: Number((diskAfter.available / GB).toFixed(1)),
errors: errors.length ? errors : undefined,
errors: errors.length ? errors : undefined,
timestamp: new Date().toISOString(),
timestamp: new Date().toISOString(),
};
};
+ sweepState.lastRunAt = Date.now();
+ sweepState.lastResult = summary;
if (migrated || promoted || demoted) {
if (migrated || promoted || demoted) {
console.log(`[VodStorage] Sweep: ${migrated} → B2, ${promoted} → R2, ${demoted} R2→B2, ${(bytesFreed / 1048576).toFixed(1)} MB freed locally`);
console.log(`[VodStorage] Sweep: ${migrated} → B2, ${promoted} → R2, ${demoted} R2→B2, ${(bytesFreed / 1048576).toFixed(1)} MB freed locally`);
}
}
@@ -742,6 +894,7 @@ async function runSweep() {
@@ -742,6 +894,7 @@ async function runSweep() {
return { error: err.message };
return { error: err.message };
} finally {
} finally {
sweepRunning = false;
sweepRunning = false;
+ sweepState.startedAt = null;
}
}
}
}
@@ -812,6 +965,32 @@ async function migrateLegacy() {
@@ -812,6 +965,32 @@ async function migrateLegacy() {
console.log(`[VodStorage] Legacy migration: ${flipped} cold VOD(s) mapped to B2${restoredLocal ? `, ${restoredLocal} still local` : ''}`);
console.log(`[VodStorage] Legacy migration: ${flipped} cold VOD(s) mapped to B2${restoredLocal ? `, ${restoredLocal} still local` : ''}`);
}
}
+// The sweep is a self-rescheduling chain, not a fixed interval: after a pass that
+// left the disk still needing a drain it comes back in pressureRetryMs, otherwise in
+// sweepIntervalMs. A sweep that overran its deadline is reported by the watchdog
+// (the upload abort in uploadFile is what actually frees it).
+const SWEEP_WATCHDOG_MS = 3 * 60 * 60 * 1000;
+
+function scheduleNext(delayMs) {
+ if (sweepTimer) { clearTimeout(sweepTimer); sweepTimer = null; }
+ sweepState.nextRunAt = Date.now() + delayMs;
+ sweepTimer = setTimeout(async () => {
+ sweepTimer = null;
+ let result = null;
+ try {
+ result = await runSweep();
+ } catch (err) {
+ console.error('[VodStorage] Sweep failed:', err.message);
+ }
+ if (result && result.skipped && result.reason === 'already running'
+ && sweepState.startedAt && Date.now() - sweepState.startedAt > SWEEP_WATCHDOG_MS) {
+ console.error(`[VodStorage] Watchdog: a sweep has been running for ${Math.round((Date.now() - sweepState.startedAt) / 60000)} min — an upload is probably hung`);
+ }
+ scheduleNext(planNextDelayMs(result, getSettings()));
+ }, delayMs);
+ if (sweepTimer.unref) sweepTimer.unref();
+}
+
function start() {
function start() {
stop();
stop();
const settings = getSettings();
const settings = getSettings();
@@ -819,18 +998,13 @@ function start() {
@@ -819,18 +998,13 @@ function start() {
console.log('[VodStorage] Disabled — not starting sweep timer');
console.log('[VodStorage] Disabled — not starting sweep timer');
return;
return;
}
}
- console.log(`[VodStorage] Starting sweep timer (every ${(settings.sweepIntervalMs / 60000).toFixed(0)} min)`);
- setTimeout(() => {
- runSweep().catch(err => console.error('[VodStorage] Sweep failed:', err.message));
- sweepTimer = setInterval(() => {
- runSweep().catch(err => console.error('[VodStorage] Sweep failed:', err.message));
- }, settings.sweepIntervalMs);
- if (sweepTimer.unref) sweepTimer.unref();
- }, 30_000).unref?.();
+ console.log(`[VodStorage] Starting sweep (every ${(settings.sweepIntervalMs / 60000).toFixed(0)} min, every ${(settings.pressureRetryMs / 60000).toFixed(0)} min while draining)`);
+ scheduleNext(30_000);
}
}
function stop() {
function stop() {
- if (sweepTimer) { clearInterval(sweepTimer); sweepTimer = null; }
+ if (sweepTimer) { clearTimeout(sweepTimer); sweepTimer = null; }
+ sweepState.nextRunAt = null;
}
}
// ── Status ───────────────────────────────────────────────────
// ── Status ───────────────────────────────────────────────────
@@ -987,11 +1161,28 @@ function getStatus() {
@@ -987,11 +1161,28 @@ function getStatus() {
r2: { count: clipCounts.r2Count || 0 },
r2: { count: clipCounts.r2Count || 0 },
},
},
sweepRunning,
sweepRunning,
+ sweep: {
+ running: sweepRunning,
+ startedAt: sweepState.startedAt ? new Date(sweepState.startedAt).toISOString() : null,
+ lastRunAt: sweepState.lastRunAt ? new Date(sweepState.lastRunAt).toISOString() : null,
+ nextRunAt: sweepState.nextRunAt ? new Date(sweepState.nextRunAt).toISOString() : null,
+ lastResult: sweepState.lastResult,
+ stalled: sweepState.stalled,
+ stalledPasses: sweepState.stalledPasses,
+ needsDrain: needsDrain(localDisk, settings),
+ critical: isCritical(localDisk, settings),
+ },
};
};
}
}
module.exports = {
module.exports = {
DEFAULTS,
DEFAULTS,
+ OFFLOADABLE_WHERE,
+ needsDrain,
+ drainSatisfied,
+ isCritical,
+ planNextDelayMs,
+ uploadTimeoutMs,
providerOf,
providerOf,
isRemote,
isRemote,
keyForVod,
keyForVod,
new file mode 100644
new file mode 100644
@@ -0,0 +1,77 @@
@@ -0,0 +1,77 @@
+'use strict';
+
+// Regression tests for the VOD offload policy in server/vod/vod-storage.js.
+//
+// Background (2026-08-29): the disk sat at 96% for weeks while the sweep logged
+// "draining" every 15 minutes and never uploaded a byte, because ~350 VOD rows with
+// no file matched the candidate query and filled its LIMIT ahead of the real
+// recordings. These tests pin the pieces that prevent a repeat: candidate
+// selection, the free-space budget, the re-run cadence, and the upload deadline.
+
+const assert = require('assert');
+const Database = require('better-sqlite3');
+
+process.env.MEDIA_B2_ENDPOINT = process.env.MEDIA_B2_ENDPOINT || '';
+const storage = require('../server/vod/vod-storage');
+const { DEFAULTS, OFFLOADABLE_WHERE, needsDrain, drainSatisfied, isCritical, planNextDelayMs, uploadTimeoutMs } = storage;
+
+const GB = 1024 * 1024 * 1024;
+const disk = (usePct, freeGb, totalGb = 96) => ({ total: totalGb * GB, used: (totalGb - freeGb) * GB, available: freeGb * GB, usePct, mount: '/' });
+
+// ── Candidate selection: only rows with a real, non-quarantined local file ──
+const db = new Database(':memory:');
+db.exec(`CREATE TABLE vods (
+ id INTEGER PRIMARY KEY, file_path TEXT, file_size INTEGER, storage_provider TEXT,
+ is_recording INTEGER, health_status TEXT, view_count INTEGER, last_accessed_at TEXT,
+ created_at TEXT DEFAULT (datetime('now', '-2 days'))
+)`);
+const ins = db.prepare('INSERT INTO vods (id, file_path, file_size, storage_provider, is_recording, health_status) VALUES (?, ?, ?, ?, ?, ?)');
+ins.run(1, '/data/vods/real-a.mp4', 15 * GB, 'local', 0, 'ok'); // real recording
+ins.run(2, '/data/vods/real-b.webm', 2 * GB, null, 0, null); // real, provider/health NULL (legacy rows)
+ins.run(3, null, 0, 'local', 0, null); // ghost: never got a file
+ins.run(4, '/opt/hobostreamer/data/vods/old.webm', 0, 'local', 0, 'missing_file'); // quarantined legacy path
+ins.run(5, '/data/vods/zero.webm', 0, 'local', 0, 'zero_byte'); // quarantined zero-byte
+ins.run(6, '/data/vods/live.mp4', 1 * GB, 'local', 1, 'ok'); // still recording
+ins.run(7, '/data/vods/cold.mp4', 8 * GB, 'b2', 0, 'ok'); // already offloaded
+
+const picked = db.prepare(`SELECT id FROM vods WHERE ${OFFLOADABLE_WHERE} ORDER BY id`).all().map(r => r.id);
+assert.deepStrictEqual(picked, [1, 2], `only real local finished files are offload candidates, got ${picked}`);
+console.log('✅ ghost / quarantined / recording / offloaded rows are never offload candidates');
+
+// ── Free-space budget: percent OR GB floor triggers, both must be satisfied to stop ──
+assert.strictEqual(needsDrain(disk(96, 4), DEFAULTS), true, '96% full must drain');
+assert.strictEqual(needsDrain(disk(50, 48), DEFAULTS), false, 'half-empty disk is fine');
+assert.strictEqual(needsDrain(disk(65, 20), DEFAULTS), true, 'below the percent ceiling but under the GB floor still drains');
+assert.strictEqual(needsDrain(disk(75, 400, 1600), DEFAULTS), true, 'percent ceiling applies even with lots of absolute space');
+assert.strictEqual(needsDrain({ total: 0, available: 0, usePct: 0 }, DEFAULTS), false, 'unknown disk never drains');
+
+assert.strictEqual(drainSatisfied(disk(60, 38), DEFAULTS), false, 'at the percent low-water but under the GB target keeps draining');
+assert.strictEqual(drainSatisfied(disk(55, 43), DEFAULTS), true, 'both targets met stops the drain');
+assert.strictEqual(drainSatisfied(disk(70, 500, 1600), DEFAULTS), false, 'above the percent low-water keeps draining on big disks');
+
+assert.strictEqual(isCritical(disk(90, 9), DEFAULTS), true);
+assert.strictEqual(isCritical(disk(89, 10), DEFAULTS), false);
+console.log('✅ free-space budget: percent ceiling OR GB floor start a drain; both targets end it');
+
+// ── Cadence: come back quickly while a drain is making (or could make) progress ──
+assert.strictEqual(planNextDelayMs({ stillNeedsDrain: true, migrated: 3 }, DEFAULTS), DEFAULTS.pressureRetryMs, 'progress under pressure → soon');
+assert.strictEqual(planNextDelayMs({ stillNeedsDrain: true, migrated: 0, skippedBackoff: 2 }, DEFAULTS), DEFAULTS.pressureRetryMs, 'candidates in back-off → soon');
+assert.strictEqual(planNextDelayMs({ stillNeedsDrain: true, migrated: 0, errors: [{ id: 1 }] }, DEFAULTS), DEFAULTS.pressureRetryMs, 'failed uploads → retry soon');
+assert.strictEqual(planNextDelayMs({ stillNeedsDrain: true, migrated: 0 }, DEFAULTS), DEFAULTS.sweepIntervalMs, 'nothing eligible → no point hammering');
+assert.strictEqual(planNextDelayMs({ stillNeedsDrain: false, migrated: 5 }, DEFAULTS), DEFAULTS.sweepIntervalMs, 'drain satisfied → regular interval');
+assert.strictEqual(planNextDelayMs(null, DEFAULTS), DEFAULTS.sweepIntervalMs, 'skipped/failed sweep → regular interval');
+assert.ok(DEFAULTS.pressureRetryMs < DEFAULTS.sweepIntervalMs);
+console.log('✅ sweep re-runs within minutes while a drain is still needed');
+
+// ── Upload deadline scales with size so a hung 15 GB multipart cannot hold the sweep forever ──
+const t100mb = uploadTimeoutMs(100 * 1024 * 1024, DEFAULTS);
+const t15gb = uploadTimeoutMs(15 * GB, DEFAULTS);
+assert.ok(t100mb >= DEFAULTS.uploadTimeoutFloorMs && t100mb < DEFAULTS.uploadTimeoutFloorMs + 2 * 60 * 1000, 'small file ≈ floor');
+assert.ok(t15gb > 2 * 60 * 60 * 1000 && t15gb < 4 * 60 * 60 * 1000, `15 GB at 2 MB/s ≈ 2–3 h, got ${Math.round(t15gb / 60000)} min`);
+console.log('✅ upload deadline = floor + size / minimum throughput');
+
+// ── Status surface exists for dashboards / health checks ──
+assert.ok(typeof storage.getStatus === 'function');
+console.log('✅ policy helpers exported for status reporting');
+
+console.log('\n✅ All VOD storage policy tests passed');