openvibers/openvibe.media Commits
vod.duration.reconcile: stored VOD durations are compared with a measurement of the real file in bounded batches, offloaded VODs included, and clearly wrong values are repaired (the health scanner only repaired durations <= 0 and skipped the 674 of 906 VODs on B2/R2). A local file is probed where it is; a B2/R2 copy through a presigned GET URL, which ffprobe range-reads (nothing is downloaded). Verdicts: ok (within 2 s), mismatch (reported only), wrong (more than 60 s and 2% off), missing (nothing stored, the file measures), unmeasurable, skipped. A wrong or missing value is written only when a second measurement confirms it: a stream-copy pass over a local file's packets (the packets win when the header disagrees, duration_source remux), or for a remote copy the container agreeing with its own streams (confirm_remote reads the whole copy instead). A repair writes duration_seconds, probe_duration_seconds and duration_source only while the row still holds the value read, and re-projects the object. Every run writes a JSON report under <data>/reports. The job (heavy lane; params limit 1-200, after_id, apply, confirm_remote) walks each tenant's library from a kept cursor; it is not scheduled unless MEDIA_DURATION_RECONCILE_HOURS > 0 and scheduled runs only report unless MEDIA_DURATION_RECONCILE_APPLY=1. scripts/vod-duration-reconcile.js is the operator's: a dry run by default (--all walks every batch), --apply --backup <file.json> takes a checked online database copy first and keeps the report as the rollback file, --rollback <file> [--apply] restores rows that still hold what the repair wrote. scripts/vods-orphans-report.js reports on the B2 vods-orphans/ prefix (hazard H3): each object matched to a VOD or clip by file name, storage key or the recorder's vod-<app>-<id>-<ms> name, its canonical copy HEADed and its local file checked, with a recommendation (keep_only_copy, keep_until_offloaded, delete_duplicate, review_size_differs, delete_row_gone, review_row_mismatch, review_unknown) and totals in objects and bytes. It only lists and HEADs; nothing is deleted, moved or written. vod-storage gains a read-only listObjects. test/duration-reconcile.test.js runs both against a fake S3 through the real SDK with real ffmpeg files.
Changes
@@@ -458,6 +458,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) || `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part || `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |++| `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) || `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and too@@@ -493,6 +494,45 @@ It changes the database directly (the service's worker picks approved jobs up onwhen the service's outbox is on, queues each change's event in the same transaction. It never runs ajob. `media_jobs{type,status}` on `/metrics` counts jobs.++## Duration reconciliation++++Code: `server/vod/duration-reconcile.js`, the job `vod.duration.reconcile`, `scripts/vod-duration-reconcile.js`.++Test: `test/duration-reconcile.test.js` (a fake S3 through the real SDK, real ffmpeg files).++++Stored VOD durations are compared with a measurement of the real file, in bounded batches walked by++id. A local file is probed where it is; an offloaded one is probed in B2/R2 through a presigned GET++URL (ffprobe reads the index with ranged requests; nothing is downloaded). Verdicts: `ok` (within++2 s), `mismatch` (reported only), `wrong` (more than 60 s **and** 2% off), `missing` (nothing++stored, the file measures), `unmeasurable`, `skipped` (`zero_byte`/`missing_file`). A `wrong` or++`missing` value is repaired only when a second measurement confirms it: for a local file a++stream-copy pass over its packets (when the container disagrees with the packets, the packets are++used, `duration_source = remux`); for a remote copy the container must agree with its own streams'++durations (`--confirm-remote` / `confirm_remote` reads the whole copy instead, which is egress). A++repair writes `duration_seconds`, `probe_duration_seconds` and `duration_source` only while the row++still holds the value that was read, and re-projects the object. Every run writes a JSON report++(`<data>/reports/duration-reconcile-*.json`) listing each VOD with stored and measured values.++++```++node scripts/vod-duration-reconcile.js [--app live] [--batch 50] [--after <id>] [--all] [--ids 1,2] [--confirm-remote] # dry run++node scripts/vod-duration-reconcile.js --apply --backup <file.json> [same selection] # checked DB backup first; file = rollback++node scripts/vod-duration-reconcile.js --rollback <file.json> [--apply]++```++++The job is not scheduled unless `MEDIA_DURATION_RECONCILE_HOURS` > 0, and scheduled runs repair only++with `MEDIA_DURATION_RECONCILE_APPLY=1` (batch `MEDIA_DURATION_RECONCILE_BATCH`, 25).++++### `vods-orphans/` report++++`scripts/vods-orphans-report.js [--prefix vods-orphans/] [--provider b2] [--out file] [--json]`++(`server/vod/orphans-report.js`) lists the B2 prefix where orphaned recordings were parked (hazard++H3), matches each object to a VOD or clip (file name, storage key, or the recorder's++`vod-<app>-<id>-<ms>` name), HEADs that row's canonical copy and checks its local file, and++recommends: `keep_only_copy` (the row exists and this is its only copy: restore it, never delete),++`keep_until_offloaded`, `delete_duplicate` (the canonical copy has the same size),++`review_size_differs`, `delete_row_gone` (the row was deleted; its object says when),++`review_row_mismatch`, `review_unknown`. Totals per recommendation in objects and bytes. It only++lists and HEADs: nothing is deleted, moved or written. `delete_*` rows are for the owner to approve.++## R2 eviction drillThe exit criterion "the canonical B2 object survives R2 eviction", as a repeatable drill with a kept@@@ -542,6 +582,8 @@ Not run against production yet: it needs the owner's go-ahead.| `MEDIA_JOBS_POLL_MS` | 5000 | worker poll interval || `MEDIA_JOBS_LIGHT_CONCURRENCY` / `_HEAVY_CONCURRENCY` / `_FINALIZE_CONCURRENCY` | 2 / 1 / 1 | jobs per lane || `MEDIA_FINALIZE_ORPHAN_GRACE_S` / `MEDIA_FINALIZE_SWEEP_S` | 180 / 120 | an orphaned recording's file must be idle this long; the orphan sweep runs this often |++| `MEDIA_DURATION_RECONCILE_HOURS` | 0 | schedule `vod.duration.reconcile` per tenant this often (`0` = only on demand) |++| `MEDIA_DURATION_RECONCILE_APPLY` / `_BATCH` | off / 25 | scheduled runs repair (`1`) or only report; VODs per scheduled run || `MEDIA_JOBS_HEAVY_WHILE_RECORDING` | off | `1` lets split/remux run while a recording is being written || `MEDIA_JOBS_LEASE_S` | 120 | running-job lease (renewed by the heartbeat) || `MEDIA_INVARIANT_SCAN_HOURS` | 24 | the size-invariant validator's schedule per tenant (`0` = on demand only) |nnew file mode 100755@@@ -0,0 +1,130 @@++#!/usr/bin/env node++'use strict';++/**++ * VOD duration reconciliation, for the operator (server/vod/duration-reconcile.js has the rules; the++ * service runs the same thing as the job vod.duration.reconcile).++ *++ * node scripts/vod-duration-reconcile.js [--app live] [--batch 50] [--after <vod id>] [--all] [--ids 1,2,3]++ * [--confirm-remote] [--out <report.json>] [--json]++ * Dry run (the default): measures each finished VOD (local file, or the B2/R2 copy with ffprobe++ * over a presigned URL, ranged reads), classifies it (ok, mismatch, wrong, missing, unmeasurable,++ * skipped) and writes the report (default <data>/reports/duration-reconcile-…json). Changes++ * nothing. One batch from --after (default 0); --all walks every batch to the end.++ * Values that would be repaired are confirmed first (local: a stream-copy pass over the++ * packets; remote: the container must agree with its streams, or --confirm-remote reads the++ * whole copy, which is B2/R2 egress).++ * node scripts/vod-duration-reconcile.js --apply --backup <file.json> [same selection]++ * First an online backup of the database to <file>.media.db (checked with integrity_check),++ * then repairs the confirmed wrong/missing values (only rows still holding the value read);++ * <file.json> is the report, which lists every repair with its old values (the rollback file).++ * Refuses to overwrite either file.++ * node scripts/vod-duration-reconcile.js --rollback <file.json> [--apply]++ * Restores the old values of the repairs the file lists, where the row still holds what the++ * repair wrote. Without --apply it only reports what it would restore.++ * (every command takes --db <media.db>; the database is DB_PATH otherwise)++ *++ * B2/R2 credentials come from the environment (the service's /etc/openvibe/media.env names).++ * Exit codes: 0 done, 1 error, 2 bad usage.++ */++const fs = require('fs');++const path = require('path');++++const args = process.argv.slice(2);++const has = (name) => args.includes(`--${name}`);++const arg = (name) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : null; };++const usage = (msg) => { console.error(`vod-duration-reconcile: ${msg}`); process.exit(2); };++for (const flag of ['db', 'backup', 'rollback', 'batch', 'after', 'ids', 'app', 'out']) if (has(flag) && !arg(flag)) usage(`--${flag} needs a value`);++if (arg('db')) process.env.DB_PATH = path.resolve(arg('db'));++++const APPLY = has('apply');++const JSON_OUT = has('json');++const BACKUP = arg('backup') ? path.resolve(arg('backup')) : null;++const ROLLBACK = arg('rollback') ? path.resolve(arg('rollback')) : null;++const BATCH = arg('batch') ? Number(arg('batch')) : 50;++const AFTER = arg('after') ? Number(arg('after')) : 0;++const IDS = arg('ids') ? arg('ids').split(',').map(Number) : null;++if (!Number.isInteger(BATCH) || BATCH < 1 || BATCH > 200) usage('--batch must be an integer 1-200');++if (!Number.isInteger(AFTER) || AFTER < 0) usage('--after must be a vod id');++if (IDS && IDS.some(n => !Number.isInteger(n) || n < 1)) usage('--ids must be a comma list of vod ids');++if (ROLLBACK && BACKUP) usage('--rollback and --backup do not go together');++if (APPLY && !ROLLBACK && !BACKUP) usage('--apply needs --backup <file.json> (the rollback file; the database copy goes next to it)');++++const config = require('../server/config');++if (!fs.existsSync(config.db.path)) usage(`no database at ${config.db.path} (set DB_PATH or pass --db)`);++const db = require('../server/db/database');++const reconcile = require('../server/vod/duration-reconcile');++const log = (...a) => { if (!JSON_OUT) console.log(...a); };++++function writePrivate(file, text) {++ fs.mkdirSync(path.dirname(file), { recursive: true, mode: 0o700 });++ const tmp = `${file}.tmp-${process.pid}`;++ const fd = fs.openSync(tmp, 'wx', 0o600);++ try { fs.writeSync(fd, text); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }++ fs.renameSync(tmp, file);++}++++async function backupDatabase(dest) {++ fs.mkdirSync(path.dirname(dest), { recursive: true, mode: 0o700 });++ await db.getDb().backup(dest);++ fs.chmodSync(dest, 0o600);++ const Database = require('better-sqlite3');++ const copy = new Database(dest, { readonly: true, fileMustExist: true });++ try { return { path: dest, bytes: fs.statSync(dest).size, integrity_check: copy.pragma('integrity_check', { simple: true }), vods: copy.prepare('SELECT count(*) AS n FROM vods').get().n }; } finally { copy.close(); }++}++++const fmt = (s) => (s == null ? '-' : `${Math.round(s)}s`);++function printRow(r) {++ if (r.verdict === 'ok') return;++ log(` vod ${String(r.vod_id).padStart(6)} ${String(r.where || '-').padEnd(5)} stored ${fmt(r.stored).padStart(8)} (${r.stored_source || '?'}) measured ${fmt(r.measured).padStart(8)} (${r.measured_source || '-'}) ${r.verdict.padEnd(12)} ${r.action}${r.note ? ` [${r.note}]` : ''}`);++}++++async function main() {++ if (ROLLBACK) {++ let rep;++ try { rep = JSON.parse(fs.readFileSync(ROLLBACK, 'utf8')); } catch (err) { usage(`cannot read ${ROLLBACK}: ${err.message}`); }++ if (!rep || rep.kind !== 'media.vod.duration_reconcile' || !Array.isArray(rep.rows)) usage(`${ROLLBACK} is not a duration reconcile report`);++ const out = reconcile.rollback(rep.rows, { apply: APPLY });++ log(APPLY ? `Restored ${out.restored} VOD duration(s); ${out.changed_since} changed since (left alone).`++ : `Would restore ${out.would_restore} VOD duration(s); ${out.changed_since} changed since. Re-run with --apply to restore.`);++ if (JSON_OUT) console.log(JSON.stringify(out, null, 2));++ return 0;++ }++ let dbCopy = null;++ if (APPLY) {++ const copyPath = BACKUP.replace(/\.json$/i, '') + '.media.db';++ for (const f of [BACKUP, copyPath]) if (fs.existsSync(f)) usage(`${f} already exists; pick a new --backup name`);++ log(`Backing up ${config.db.path} -> ${copyPath} ...`);++ dbCopy = await backupDatabase(copyPath);++ if (dbCopy.integrity_check !== 'ok') { console.error(`backup failed integrity_check: ${dbCopy.integrity_check}; nothing changed`); return 1; }++ log(` ${dbCopy.bytes} bytes, integrity_check ok, ${dbCopy.vods} vods`);++ }++ const all = { kind: 'media.vod.duration_reconcile', version: 1, run_at: new Date().toISOString(), mode: APPLY ? 'apply' : 'dry-run', app_id: arg('app') || null,++ thresholds: reconcile.THRESHOLDS, confirm_remote: has('confirm-remote'), batches: 0, range: { after_id: IDS ? null : AFTER, last_id: null, done: false }, counts: null, rows: [], db_backup: dbCopy };++ let after = AFTER;++ log(`${APPLY ? 'Reconciling' : 'Dry run over'} VOD durations in ${config.db.path}${all.app_id ? ` (app ${all.app_id})` : ''}${IDS ? `, vods ${IDS.join(',')}` : `, from vod id > ${AFTER}, batches of ${BATCH}`}${has('all') ? ', every batch' : ''}`);++ for (;;) {++ const rep = await reconcile.reconcileBatch({ appId: all.app_id, afterId: after, limit: IDS ? Math.min(200, IDS.length) : BATCH, ids: IDS, apply: APPLY, confirmRemote: has('confirm-remote'), onRow: printRow });++ all.batches++;++ all.rows.push(...rep.rows);++ if (!all.counts) all.counts = { ...rep.counts };++ else for (const k of Object.keys(rep.counts)) all.counts[k] += rep.counts[k];++ all.range.last_id = rep.range.last_id ?? all.range.last_id;++ all.range.done = rep.range.done;++ if (APPLY) writePrivate(BACKUP, JSON.stringify(all, null, 1)); // the rollback file is current after every batch++ if (IDS || rep.range.done || !has('all')) break;++ after = rep.range.last_id;++ }++ const file = APPLY ? BACKUP : (arg('out') ? path.resolve(arg('out')) : null);++ const written = APPLY ? file : reconcile.writeReport(all, file);++ const c = all.counts;++ log('');++ log(`Checked ${c.checked} (${c.remote} remote): ok ${c.ok}, mismatch ${c.mismatch}, wrong ${c.wrong}, missing ${c.missing}, unmeasurable ${c.unmeasurable}, skipped ${c.skipped}.`);++ log(APPLY ? `Repaired ${c.repaired}; ${c.unconfirmed} unconfirmed (not written); ${c.changed_since} changed while running.`++ : `Would repair ${c.would_repair}; ${c.unconfirmed} unconfirmed (would not be written). Nothing changed.`);++ if (!all.range.done && !IDS) log(`More to check: next run --after ${all.range.last_id} (or --all).`);++ log(`Report: ${written}`);++ if (JSON_OUT) console.log(JSON.stringify({ ...all, rows: undefined, report: written }, null, 2));++ return 0;++}++++main().then((code) => { db.close(); process.exit(code); }).catch((err) => { console.error(err); process.exit(1); });nnew file mode 100755@@@ -0,0 +1,55 @@++#!/usr/bin/env node++'use strict';++/**++ * Report on the B2 `vods-orphans/` prefix (hazard H3). Read-only: it lists the prefix (ListObjectsV2),++ * HEADs each match's canonical copy, reads the database, and writes a report. It never deletes, moves,++ * copies or writes an object or a row (server/vod/orphans-report.js has the rules).++ *++ * node scripts/vods-orphans-report.js [--prefix vods-orphans/] [--provider b2] [--out <report.json>] [--json] [--db <media.db>]++ *++ * Prints one line per object (key, size, matched VOD/clip, recommendation) and the totals per++ * recommendation, and writes the full report (default <data>/reports/vods-orphans-<time>.json).++ * B2 credentials come from the environment (/etc/openvibe/media.env names). Exit 0 done, 1 error, 2 usage.++ */++const fs = require('fs');++const path = require('path');++++const args = process.argv.slice(2);++const has = (name) => args.includes(`--${name}`);++const arg = (name) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : null; };++const usage = (msg) => { console.error(`vods-orphans-report: ${msg}`); process.exit(2); };++for (const flag of ['prefix', 'provider', 'out', 'db']) if (has(flag) && !arg(flag)) usage(`--${flag} needs a value`);++if (arg('db')) process.env.DB_PATH = path.resolve(arg('db'));++const provider = arg('provider') || 'b2';++if (!['b2', 'r2'].includes(provider)) usage('--provider must be b2 or r2');++++const config = require('../server/config');++if (!fs.existsSync(config.db.path)) usage(`no database at ${config.db.path} (set DB_PATH or pass --db)`);++const db = require('../server/db/database');++const vodStorage = require('../server/vod/vod-storage');++const orphans = require('../server/vod/orphans-report');++const JSON_OUT = has('json');++const log = (...a) => { if (!JSON_OUT) console.log(...a); };++const mb = (b) => `${(b / 1048576).toFixed(1)} MB`;++++(async () => {++ if (!vodStorage.providerConfigured(provider)) usage(`${provider} is not configured (MEDIA_${provider.toUpperCase()}_* not in the environment)`);++ const report = await orphans.buildReport({ provider, prefix: arg('prefix') || orphans.PREFIX });++ const file = arg('out') ? path.resolve(arg('out')) : path.join(path.dirname(config.db.path), 'reports', `vods-orphans-${report.generated_at.replace(/[:.]/g, '-')}.json`);++ fs.mkdirSync(path.dirname(file), { recursive: true });++ fs.writeFileSync(file, JSON.stringify(report, null, 1));++ log(`${report.provider}://${report.bucket}/${report.prefix}: ${report.totals.objects} object(s), ${report.totals.gb} GB (read-only report)`);++ log('');++ for (const o of report.objects) {++ const m = o.match ? `${o.match.kind} ${o.match.id}${o.match.row === 'gone' ? ' (row gone)' : ''}` : '-';++ log(` ${o.key.padEnd(60)} ${mb(o.size).padStart(11)} ${m.padEnd(22)} ${o.recommendation.padEnd(20)} ${o.reason}`);++ }++ log('');++ for (const [rec, t] of Object.entries(report.by_recommendation)) if (t.objects) log(` ${rec.padEnd(22)} ${String(t.objects).padStart(4)} object(s) ${mb(t.bytes).padStart(12)}`);++ log('');++ log('Nothing was deleted or moved. delete_* rows are candidates for the owner; keep_only_copy rows are recordings whose only copy is here.');++ log(`Report: ${file}`);++ if (JSON_OUT) console.log(JSON.stringify({ ...report, report: file }, null, 2));++ db.close();++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });nnew file mode 100644@@@ -0,0 +1,80 @@++/**++ * OpenVibe.Media — job type vod.duration.reconcile (heavy lane): one bounded batch of the tenant's++ * finished VODs, stored duration vs a measurement of the real file (local, or the B2/R2 copy through++ * a presigned ranged read). server/vod/duration-reconcile.js has the rules.++ *++ * params { limit (1-200, default 25), after_id?, apply (default false), confirm_remote (default false) }++ * result { mode, counts, range, next_after_id, report } report = the file name under <data>/reports/++ *++ * Without after_id the job walks the tenant's library: it starts where the previous run stopped++ * (media_settings duration_reconcile.cursor.<app>) and wraps to the start when it reaches the end.++ * apply=false only reports. Scheduling is off unless MEDIA_DURATION_RECONCILE_HOURS > 0; scheduled++ * runs repair only with MEDIA_DURATION_RECONCILE_APPLY=1 (batch MEDIA_DURATION_RECONCILE_BATCH, 25).++ */++'use strict';++++const path = require('path');++const db = require('../db/database');++const queue = require('./queue');++const reconcile = require('../vod/duration-reconcile');++++const { JobError } = queue;++const TYPE = 'vod.duration.reconcile';++const cursorKey = (appId) => `duration_reconcile.cursor.${appId}`;++const bool = (v) => v === true || v === 1 || ['1', 'true', 'on', 'yes'].includes(String(v).toLowerCase());++++function envInt(name, fallback) {++ const v = parseInt(process.env[name] || '', 10);++ return Number.isFinite(v) ? v : fallback;++}++++function validate({ obj, params }) {++ if (obj) throw new JobError('media.job.invalid', `${TYPE} covers the tenant's VODs: leave object_id out`, { permanent: true });++ const p = params || {};++ const limit = p.limit == null ? 25 : Number(p.limit);++ if (!Number.isInteger(limit) || limit < 1 || limit > reconcile.MAX_BATCH) throw new JobError('media.job.invalid', `params.limit must be 1-${reconcile.MAX_BATCH}`, { permanent: true });++ const out = { limit, apply: bool(p.apply), confirm_remote: bool(p.confirm_remote) };++ if (p.after_id != null) {++ const a = Number(p.after_id);++ if (!Number.isInteger(a) || a < 0) throw new JobError('media.job.invalid', 'params.after_id must be a non-negative integer', { permanent: true });++ out.after_id = a;++ }++ return out;++}++++async function run(job, ctx = {}) {++ const p = validate({ obj: null, params: job.params });++ const walking = p.after_id == null;++ const afterId = walking ? (Number(db.getSetting(cursorKey(job.app_id))) || 0) : p.after_id;++ const report = await reconcile.reconcileBatch({++ appId: job.app_id, afterId, limit: p.limit, apply: p.apply, confirmRemote: p.confirm_remote, signal: ctx.signal,++ });++ report.job_id = job.id;++ const file = reconcile.writeReport(report);++ const next = report.range.done ? 0 : report.range.last_id;++ if (walking) db.setSetting(cursorKey(job.app_id), next, 'number');++ return { mode: report.mode, counts: report.counts, range: report.range, next_after_id: next, report: path.basename(file) };++}++++/** Queue one run per tenant with VODs every MEDIA_DURATION_RECONCILE_HOURS (0 = never). Returns how many. */++function schedule(nowMs = Date.now()) {++ const hours = envInt('MEDIA_DURATION_RECONCILE_HOURS', 0);++ if (!(hours > 0)) return 0;++ const period = Math.floor(nowMs / (hours * 3600 * 1000));++ const apply = bool(process.env.MEDIA_DURATION_RECONCILE_APPLY || '');++ const limit = Math.min(reconcile.MAX_BATCH, Math.max(1, envInt('MEDIA_DURATION_RECONCILE_BATCH', 25)));++ let n = 0;++ for (const { app_id: appId } of db.all('SELECT DISTINCT app_id FROM vods WHERE COALESCE(is_recording, 0) = 0 AND file_path IS NOT NULL')) {++ try {++ const r = queue.enqueue({ appId, type: TYPE, params: { limit, apply, confirm_remote: false }, idempotencyKey: `${TYPE}:${hours}h:${period}`, createdBy: 'system:schedule' });++ if (r.created) n++;++ } catch (err) { console.warn(`[Jobs] could not schedule ${TYPE} for ${appId}: ${err.message}`); }++ }++ return n;++}++++module.exports = {++ TYPE,++ spec: { lane: 'heavy', maxAttempts: 2, timeoutMs: 60 * 60 * 1000, needsObject: false, validate, run },++ schedule, cursorKey,++};@@@ -6,6 +6,7 @@* object.split heavy stream-copy parts of a large vod/clip as new private objects* object.remux heavy a stream-copy remux of a vod/clip as a new private object* vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff++ * vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch*/'use strict';@@@ -16,5 +17,6 @@ queue.register('invariant.scan', require('./invariant-scan').spec);queue.register('object.split', require('./derive').split);queue.register('object.remux', require('./derive').remux);queue.register('vod.finalize', require('./vod-finalize').spec);++queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);module.exports = { names: () => queue.typeNames() };@@@ -144,7 +144,11 @@ async function tick() {kicked = false;queue.recoverInterrupted({ except: new Set(running.keys()) });const now = Date.now();-- if (now - lastScheduleAt > 10 * 60 * 1000) { lastScheduleAt = now; scheduleInvariantScans(now); }++ if (now - lastScheduleAt > 10 * 60 * 1000) {++ lastScheduleAt = now;++ scheduleInvariantScans(now);++ try { require('./duration-reconcile').schedule(now); } catch (err) { console.warn('[Jobs] duration reconcile schedule:', err.message); }++ }try { require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }if (now - lastPruneAt > 6 * 3600 * 1000) {lastPruneAt = now;nnew file mode 100644@@@ -0,0 +1,201 @@++/**++ * OpenVibe.Media — VOD duration reconciliation (job vod.duration.reconcile; scripts/vod-duration-reconcile.js)++ *++ * Compares each finished VOD's stored duration with a measurement of its real file, in bounded++ * batches walked by id, and repairs the values that are clearly wrong. Offloaded VODs are measured++ * where they are: ffprobe reads the B2/R2 copy through a presigned HTTPS URL with ranged requests++ * (the file is not downloaded). Every run writes a report.++ *++ * Verdicts (stored vs measured, seconds):++ * ok |diff| <= 2++ * mismatch a small difference (reported, left alone)++ * wrong |diff| > 60 AND > 2% of the measured length: repairable++ * missing nothing stored (<= 0) but the file measures: repairable++ * unmeasurable no measurement (the file is unreadable, or neither copy can be reached)++ * skipped nothing to measure (zero_byte / missing_file health, no file)++ *++ * A repair needs a confirmed measurement: a local file's probe must agree with a stream-copy pass++ * over its packets (or the packets are used, source remux, when the header lies); a remote copy's++ * container duration must agree with its own streams' durations (--confirm-remote reads the whole++ * remote file instead). An unconfirmed value is reported, never written. A repair writes++ * duration_seconds, probe_duration_seconds and duration_source only if the row still holds the value++ * that was read, and re-projects the VOD's object.++ */++'use strict';++++const fs = require('fs');++const path = require('path');++const config = require('../config');++const db = require('../db/database');++const tools = require('./media-tools');++++const THRESHOLDS = Object.freeze({ mismatchS: 2, wrongAbsS: 60, wrongRel: 0.02, agreeAbsS: 5, agreeRel: 0.01, streamsAgreeRel: 0.05 });++const MAX_BATCH = 200;++++const vodStorage = () => require('./vod-storage');++++function classify(stored, measured, t = THRESHOLDS) {++ const s = Number(stored) || 0;++ if (!(measured > 0)) return 'unmeasurable';++ if (s <= 0) return 'missing';++ const diff = Math.abs(s - measured);++ if (diff <= t.mismatchS) return 'ok';++ if (diff > t.wrongAbsS && diff > t.wrongRel * measured) return 'wrong';++ return 'mismatch';++}++++const agree = (a, b, t = THRESHOLDS) => a > 0 && b > 0 && Math.abs(a - b) <= Math.max(t.agreeAbsS, t.agreeRel * Math.max(a, b));++++/** The longest stream duration ffprobe reported (0 when none). */++function streamsDuration(streams) {++ let max = 0;++ for (const st of streams || []) { const d = parseFloat(st && st.duration); if (Number.isFinite(d) && d > max) max = d; }++ return max;++}++++/**++ * Where to read a VOD from: { kind: 'file', value, where: 'local' } or { kind: 'url', value,++ * where: 'b2' | 'r2' } (a presigned GET: ffprobe range-reads it), or null.++ */++async function sourceFor(vod) {++ const vs = vodStorage();++ const local = vs.localPathForVod(vod);++ if (fs.existsSync(local)) return { kind: 'file', value: local, where: 'local' };++ if (vod.file_path && fs.existsSync(vod.file_path)) return { kind: 'file', value: vod.file_path, where: 'local' };++ if (vs.isRemote(vod)) {++ const plan = await vs.resolvePlayback(vod);++ if (plan && plan.kind === 'redirect') return { kind: 'url', value: plan.url, where: plan.provider };++ if (plan && plan.kind === 'file') return { kind: 'file', value: plan.path, where: 'local' };++ }++ return null;++}++++/**++ * Measure one VOD. Returns { where, seconds, source: probe|remux|null, confirmed, note }.++ * `confirm` runs the second, independent measurement (only needed when a repair is on the table).++ */++async function measure(vod, { confirm = false, confirmRemote = false, probeTimeoutMs } = {}) {++ const src = await sourceFor(vod);++ if (!src) return { where: null, seconds: 0, source: null, confirmed: false, note: 'no local file and no reachable remote copy' };++ const remote = src.kind === 'url';++ const probe = await tools.probeDuration(src.value, { timeoutMs: probeTimeoutMs || (remote ? 60000 : 20000) });++ const out = { where: src.where, seconds: probe.seconds || 0, source: probe.seconds > 0 ? 'probe' : null, confirmed: false, note: null };++ if (!probe.ok) { out.note = `ffprobe: ${probe.error}`; }++ if (!confirm) return out;++ if (!remote || confirmRemote) {++ // Read every packet: the media's real length, independent of the container header.++ const packets = await tools.streamCopyDuration(src.value);++ if (!packets.ok) {++ out.note = `${out.note ? `${out.note}; ` : ''}stream-copy pass: ${packets.error}`;++ return out;++ }++ if (out.seconds > 0 && agree(out.seconds, packets.seconds)) { out.confirmed = true; return out; }++ // The header disagrees with (or lacks) the packet timeline: the packets are the truth.++ out.note = out.seconds > 0 ? `container says ${out.seconds.toFixed(1)} s, packets ${packets.seconds.toFixed(1)} s: packets used` : 'no container duration: packets used';++ out.seconds = packets.seconds;++ out.source = 'remux';++ out.confirmed = true;++ return out;++ }++ // Remote, no full read: the container's duration must agree with its own streams'.++ const sd = streamsDuration(probe.streams);++ if (out.seconds > 0 && sd > 0 && Math.abs(out.seconds - sd) <= Math.max(THRESHOLDS.agreeAbsS, THRESHOLDS.streamsAgreeRel * out.seconds)) {++ out.confirmed = true;++ } else {++ out.note = `${out.note ? `${out.note}; ` : ''}remote container ${out.seconds.toFixed(1)} s vs streams ${sd.toFixed(1)} s: not confirmed (--confirm-remote reads the whole copy)`;++ }++ return out;++}++++function reportsDir() {++ return path.join(path.dirname(config.db.path), 'reports');++}++++function writeReport(report, file = null) {++ const out = file || path.join(reportsDir(), `duration-reconcile-${report.app_id || 'all'}-${report.run_at.replace(/[:.]/g, '-')}.json`);++ fs.mkdirSync(path.dirname(out), { recursive: true });++ fs.writeFileSync(out, JSON.stringify(report, null, 1));++ return out;++}++++/** Write one repair (only if the row still has the value we read). Returns true when it changed. */++function applyRepair(row) {++ const changed = db.run(`UPDATE vods SET duration_seconds = ?, probe_duration_seconds = ?, duration_source = ?++ WHERE id = ? AND COALESCE(duration_seconds, 0) = ? AND COALESCE(duration_source, '') = ? AND COALESCE(is_recording, 0) = 0`,++ [Math.round(row.measured), row.measured, row.measured_source, row.vod_id, row.stored, row.stored_source || '']).changes > 0;++ if (changed) require('../objects/model').safeSync('vod', row.vod_id);++ return changed;++}++++/** Undo repairs listed in a report (only rows that still hold what the repair wrote). */++function rollback(rows, { apply = false } = {}) {++ const out = { restored: 0, would_restore: 0, changed_since: 0 };++ for (const r of rows || []) {++ if (r.action !== 'repaired') continue;++ const cur = db.get('SELECT duration_seconds, duration_source FROM vods WHERE id = ?', [r.vod_id]);++ if (!cur || Number(cur.duration_seconds) !== Math.round(r.measured) || (cur.duration_source || '') !== (r.measured_source || '')) { out.changed_since++; continue; }++ if (!apply) { out.would_restore++; continue; }++ db.run('UPDATE vods SET duration_seconds = ?, probe_duration_seconds = ?, duration_source = ? WHERE id = ?',++ [r.stored, r.stored_probe ?? r.stored, r.stored_source || null, r.vod_id]);++ require('../objects/model').safeSync('vod', r.vod_id);++ out.restored++;++ }++ return out;++}++++/**++ * Reconcile one bounded batch. opts: appId (null = every tenant), afterId (id cursor), limit++ * (1-200), ids (exactly these VODs), apply, confirmRemote, onRow(row).++ * Returns the report: { run_at, mode, app_id, thresholds, range, counts, rows }.++ */++async function reconcileBatch({ appId = null, afterId = 0, limit = 50, ids = null, apply = false, confirmRemote = false, onRow = null, signal = null } = {}) {++ const n = Math.min(MAX_BATCH, Math.max(1, Number(limit) || 50));++ const conds = ['COALESCE(is_recording, 0) = 0', 'COALESCE(clips_only, 0) = 0', 'file_path IS NOT NULL'];++ const params = [];++ if (appId) { conds.push('app_id = ?'); params.push(appId); }++ if (ids && ids.length) { conds.push(`id IN (${ids.map(() => '?').join(', ')})`); params.push(...ids.map(Number)); } else { conds.push('id > ?'); params.push(Number(afterId) || 0); }++ const vods = db.all(`SELECT * FROM vods WHERE ${conds.join(' AND ')} ORDER BY id LIMIT ?`, [...params, n]);++ const report = {++ kind: 'media.vod.duration_reconcile', version: 1, run_at: new Date().toISOString(), mode: apply ? 'apply' : 'dry-run',++ app_id: appId, thresholds: THRESHOLDS, confirm_remote: !!confirmRemote,++ range: { after_id: ids ? null : Number(afterId) || 0, last_id: vods.length ? vods[vods.length - 1].id : null, limit: n, done: vods.length < n },++ counts: { checked: 0, ok: 0, mismatch: 0, wrong: 0, missing: 0, unmeasurable: 0, skipped: 0, repaired: 0, would_repair: 0, unconfirmed: 0, changed_since: 0, remote: 0 },++ rows: [],++ };++ for (const vod of vods) {++ if (signal && signal.aborted) break;++ report.counts.checked++;++ const stored = Number(vod.duration_seconds) || 0;++ const row = {++ vod_id: vod.id, app_id: vod.app_id, storage_provider: vod.storage_provider || 'local', health_status: vod.health_status || null,++ stored, stored_probe: Number(vod.probe_duration_seconds) || 0, stored_source: vod.duration_source || null,++ where: null, measured: 0, measured_source: null, diff: null, verdict: null, confirmed: false, action: 'none', note: null,++ };++ if (['zero_byte', 'missing_file'].includes(vod.health_status)) {++ row.verdict = 'skipped'; row.note = `health ${vod.health_status}`;++ } else {++ let m = await measure(vod);++ row.where = m.where;++ if (m.where && m.where !== 'local') report.counts.remote++;++ let verdict = classify(stored, m.seconds);++ if (verdict === 'wrong' || verdict === 'missing') {++ m = await measure(vod, { confirm: true, confirmRemote });++ verdict = classify(stored, m.seconds);++ }++ Object.assign(row, { measured: m.seconds ? Number(m.seconds.toFixed(3)) : 0, measured_source: m.source, confirmed: m.confirmed, note: m.note, verdict });++ row.diff = m.seconds > 0 ? Number((m.seconds - stored).toFixed(3)) : null;++ if (verdict === 'wrong' || verdict === 'missing') {++ if (!m.confirmed) { row.action = 'unconfirmed'; report.counts.unconfirmed++; }++ else if (!apply) { row.action = 'would_repair'; report.counts.would_repair++; }++ else if (applyRepair(row)) { row.action = 'repaired'; report.counts.repaired++; }++ else { row.action = 'changed_since'; report.counts.changed_since++; }++ }++ }++ report.counts[row.verdict]++;++ report.rows.push(row);++ if (onRow) onRow(row);++ }++ return report;++}++++module.exports = { THRESHOLDS, MAX_BATCH, classify, agree, streamsDuration, sourceFor, measure, reconcileBatch, applyRepair, rollback, writeReport, reportsDir };nnew file mode 100644@@@ -0,0 +1,139 @@++/**++ * OpenVibe.Media — report on the B2 `vods-orphans/` prefix (hazard H3: 55 objects, 14.5 GB on++ * 2026-09-21). Report only: it lists, matches and recommends; it never deletes, moves or writes.++ *++ * Each object under the prefix is matched to a VOD or clip by its file name (the basename of++ * vods.file_path / storage_key, clips.file_path; else the vod-<app>-<id>-<ms>.<ext> name the++ * recorder gives recordings), and checked against that row's canonical copy (HEAD vods/<basename>++ * on the same provider), its R2 copy and its local file. Recommendations:++ *++ * keep_only_copy the row exists and no canonical, R2 or local copy does: this IS the recording++ * (restore it to its canonical key; never delete)++ * keep_until_offloaded no canonical copy yet, but the local file has the same size++ * delete_duplicate the canonical copy exists with the same size++ * review_size_differs a copy exists but with another size (keep until someone compares them)++ * delete_row_gone the name points at a VOD/clip id whose row no longer exists (deleted); its++ * object says when, if the model saw it++ * review_row_mismatch the id exists but its row points at another file++ * review_unknown the name matches nothing++ * `delete_*` are candidates for the owner to approve, never acted on here.++ */++'use strict';++++const fs = require('fs');++const path = require('path');++const db = require('../db/database');++++const PREFIX = 'vods-orphans/';++const RECOMMENDATIONS = ['keep_only_copy', 'keep_until_offloaded', 'delete_duplicate', 'review_size_differs', 'delete_row_gone', 'review_row_mismatch', 'review_unknown'];++++const vodStorage = () => require('./vod-storage');++++/** vod-<app>-<id>-<epoch ms>.<ext> (+ .master.mkv) → { app, id, recorded_at } or null. */++function parseRecordingName(base) {++ const m = /^vod-(.+)-(\d+)-(\d{12,14})(?:\.master)?\.[a-z0-9]+$/i.exec(String(base || ''));++ if (!m) return null;++ return { app: m[1], id: Number(m[2]), recorded_at: new Date(Number(m[3])).toISOString() };++}++++function rowByBasename(base) {++ const vod = db.getVodByFileBasename(base);++ if (vod) return { kind: 'vod', row: vod };++ const byKey = db.all('SELECT * FROM vods WHERE storage_key LIKE ?', [`%${base}`]).find(r => path.basename(r.storage_key || '') === base);++ if (byKey) return { kind: 'vod', row: byKey };++ const clip = db.getClipByFileBasename(base);++ if (clip) return { kind: 'clip', row: clip };++ return null;++}++++function objectFor(kind, app, id) {++ try { return db.get('SELECT id, lifecycle_status, deleted_at FROM media_objects WHERE legacy_ref = ?', [`legacy:${app}:${kind}:${id}`]); } catch { return null; }++}++++async function head(provider, key) {++ const vs = vodStorage();++ if (!vs.providerConfigured(provider)) return { configured: false };++ try { const h = await vs.headObject(provider, key); return h ? { configured: true, size: Number(h.size) } : { configured: true, size: null }; } catch (err) { return { configured: true, error: err.message }; }++}++++function localSize(row) {++ const vs = vodStorage();++ for (const p of [vs.localPathForVod(row), row.file_path].filter(Boolean)) {++ try { const st = fs.statSync(p); if (st.isFile()) return st.size; } catch { /* next */ }++ }++ return null;++}++++/** One orphan object → its row, the copies that exist, and a recommendation. */++async function assess(obj, { provider = 'b2' } = {}) {++ const vs = vodStorage();++ const base = path.basename(obj.key);++ const parsed = parseRecordingName(base);++ const out = { key: obj.key, size: obj.size, last_modified: obj.last_modified, name: parsed, match: null, copies: {}, recommendation: null, reason: null };++ const found = rowByBasename(base);++ if (!found) {++ if (parsed) {++ const byId = db.get('SELECT * FROM vods WHERE id = ? AND app_id = ?', [parsed.id, parsed.app]);++ if (byId) {++ out.match = { kind: 'vod', id: byId.id, app_id: byId.app_id, file: path.basename(byId.file_path || ''), storage_provider: byId.storage_provider || 'local' };++ out.recommendation = 'review_row_mismatch';++ out.reason = `vod ${byId.id} exists but points at ${out.match.file || 'no file'}`;++ return out;++ }++ const o = objectFor('vod', parsed.app, parsed.id);++ out.match = { kind: 'vod', id: parsed.id, app_id: parsed.app, row: 'gone', object: o ? { id: o.id, lifecycle_status: o.lifecycle_status, deleted_at: o.deleted_at } : null };++ out.recommendation = 'delete_row_gone';++ out.reason = `vod ${parsed.id} (${parsed.app}) no longer exists${o && o.deleted_at ? ` (deleted ${o.deleted_at})` : ''}; recorded ${parsed.recorded_at}`;++ return out;++ }++ out.recommendation = 'review_unknown';++ out.reason = 'the name matches no VOD or clip';++ return out;++ }++ const { kind, row } = found;++ const canonicalKey = vs.keyForVod(row);++ out.match = { kind, id: row.id, app_id: row.app_id, storage_provider: row.storage_provider || 'local', canonical_key: canonicalKey, health_status: row.health_status || null,++ is_public: row.is_public, visibility: row.visibility || null, duration_seconds: row.duration_seconds || 0 };++ if (canonicalKey === obj.key) {++ // The row's storage_key points into the orphan prefix: this object is its canonical copy.++ out.recommendation = 'keep_only_copy';++ out.reason = `${kind} ${row.id}'s storage_key is this object: it is the canonical copy (never delete)`;++ return out;++ }++ out.copies.canonical = await head(provider, canonicalKey);++ if ((row.storage_provider || 'local') === 'r2') out.copies.r2 = await head('r2', canonicalKey);++ out.copies.local = localSize(row);++ const sizes = [out.copies.canonical && out.copies.canonical.size, out.copies.r2 && out.copies.r2.size].filter(s => s != null);++ if (sizes.includes(obj.size)) {++ out.recommendation = 'delete_duplicate';++ out.reason = `${kind} ${row.id}'s canonical copy ${canonicalKey} exists with the same size`;++ } else if (sizes.length || (out.copies.local != null && out.copies.local !== obj.size)) {++ out.recommendation = 'review_size_differs';++ out.reason = `${kind} ${row.id} has a copy of another size (${[...sizes, out.copies.local].filter(s => s != null).join(', ')} vs ${obj.size} bytes)`;++ } else if (out.copies.local === obj.size) {++ out.recommendation = 'keep_until_offloaded';++ out.reason = `${kind} ${row.id} is still local (same size); the orphan is redundant once it is offloaded to ${canonicalKey}`;++ } else {++ out.recommendation = 'keep_only_copy';++ out.reason = `the only copy of ${kind} ${row.id}: restore it to ${canonicalKey} (never delete)`;++ }++ return out;++}++++/** The whole report: { prefix, provider, bucket, generated_at, totals, by_recommendation, objects }. */++async function buildReport({ provider = 'b2', prefix = PREFIX, list = null } = {}) {++ const vs = vodStorage();++ const objects = list || await vs.listObjects(provider, prefix);++ const rows = [];++ for (const o of objects) rows.push(await assess(o, { provider }));++ const by = Object.fromEntries(RECOMMENDATIONS.map(r => [r, { objects: 0, bytes: 0 }]));++ let bytes = 0;++ for (const r of rows) { by[r.recommendation].objects++; by[r.recommendation].bytes += r.size; bytes += r.size; }++ return {++ kind: 'media.vods_orphans.report', version: 1, generated_at: new Date().toISOString(), provider, bucket: vs.bucketFor(provider), prefix,++ read_only: true, totals: { objects: rows.length, bytes, gb: Number((bytes / 1e9).toFixed(2)) }, by_recommendation: by, objects: rows,++ note: 'Report only: nothing was deleted, moved or written. delete_* rows are candidates for the owner to approve.',++ };++}++++module.exports = { PREFIX, RECOMMENDATIONS, parseRecordingName, assess, buildReport };@@@ -316,6 +316,21 @@ async function deleteLegacyPasteScreenshot(pasteId) {}}++/** Every object under `prefix` in one provider's bucket: [{ key, size, last_modified }] (read-only). */++async function listObjects(provider, prefix) {++ const client = clientFor(provider);++ if (!client) throw new Error(`${provider} is not configured`);++ loadSdk();++ const out = [];++ let token;++ do {++ const r = await client.send(new S3.ListObjectsV2Command({ Bucket: PROVIDER_ENV[provider].bucket, Prefix: prefix, ContinuationToken: token, MaxKeys: 1000 }));++ for (const o of (r.Contents || [])) out.push({ key: o.Key, size: Number(o.Size || 0), last_modified: o.LastModified ? new Date(o.LastModified).toISOString() : null });++ token = r.IsTruncated ? r.NextContinuationToken : null;++ } while (token);++ return out;++}++async function presignGet(provider, key, expiresInSeconds = 900) {const client = clientFor(provider);if (!client) return null;@@@ -1271,6 +1286,7 @@ module.exports = {bucketFor,endpointFor,headObject,++ listObjects,keyForVod,localPathForVod,resolvePlayback,nnew file mode 100644@@@ -0,0 +1,253 @@++'use strict';++// vod.duration.reconcile (server/vod/duration-reconcile.js, the job, scripts/vod-duration-reconcile.js)++// and the vods-orphans/ report (server/vod/orphans-report.js, scripts/vods-orphans-report.js), against++// a fake S3 spoken to by the real AWS SDK: stored durations are compared with a measurement of the++// real file, local or offloaded (ffprobe over a presigned URL, ranged reads); only confirmed, clearly++// wrong values are repaired; the dry run changes nothing; --apply needs a backup and the report++// rolls back; the job walks the library in bounded batches. The orphan report only reads.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const { spawnSync, execFile } = require('child_process');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-reconcile-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++const runCli = (script, args) => new Promise((resolve) => {++ execFile(process.execPath, [path.join(__dirname, '../scripts', script), ...args], { env: process.env, encoding: 'utf8' },++ (err, stdout, stderr) => resolve({ status: err ? err.code : 0, stdout, stderr }));++});++++// ── Fake S3: path-style /<bucket>/<key>; HEAD, GET (Range), ListObjectsV2; PUT/DELETE recorded ──++const buckets = { 'b2-bucket': new Map() };++const s3log = [];++const xml = (s) => String(s).replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>');++const s3 = http.createServer((req, res) => {++ req.resume();++ req.on('end', () => {++ const u = new URL(req.url, 'http://x');++ const [, bucket, ...rest] = decodeURIComponent(u.pathname).split('/');++ const key = rest.join('/');++ const store = buckets[bucket];++ s3log.push([req.method, bucket, key]);++ if (!store) { res.writeHead(404); return res.end(); }++ if (req.method === 'HEAD' && !key) { res.writeHead(200); return res.end(); } // HeadBucket++ if (req.method === 'GET' && !key && u.searchParams.get('list-type') === '2') {++ const prefix = u.searchParams.get('prefix') || '';++ const keys = [...store.keys()].filter(k => k.startsWith(prefix)).sort();++ res.writeHead(200, { 'Content-Type': 'application/xml' });++ return res.end(`<?xml version="1.0" encoding="UTF-8"?><ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>${bucket}</Name><Prefix>${xml(prefix)}</Prefix><KeyCount>${keys.length}</KeyCount><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated>${++ keys.map(k => `<Contents><Key>${xml(k)}</Key><LastModified>2026-09-21T00:00:00.000Z</LastModified><ETag>"e"</ETag><Size>${store.get(k).length}</Size><StorageClass>STANDARD</StorageClass></Contents>`).join('')}</ListBucketResult>`);++ }++ const obj = store.get(key);++ if (req.method === 'HEAD' || req.method === 'GET') {++ if (!obj) { res.writeHead(404, { 'Content-Type': 'application/xml' }); return res.end(req.method === 'GET' ? '<Error><Code>NoSuchKey</Code></Error>' : undefined); }++ const m = /bytes=(\d+)-(\d+)?/.exec(req.headers.range || '');++ if (m && req.method === 'GET') {++ const start = Number(m[1]), end = Math.min(obj.length - 1, m[2] != null ? Number(m[2]) : obj.length - 1);++ if (start >= obj.length) { res.writeHead(416, { 'Content-Range': `bytes */${obj.length}` }); return res.end(); }++ res.writeHead(206, { 'Content-Length': end - start + 1, 'Content-Range': `bytes ${start}-${end}/${obj.length}`, 'Accept-Ranges': 'bytes', ETag: '"e"', 'Content-Type': 'video/mp4' });++ return res.end(obj.subarray(start, end + 1));++ }++ res.writeHead(200, { 'Content-Length': obj.length, 'Accept-Ranges': 'bytes', ETag: '"e"', 'Content-Type': 'video/mp4' });++ return res.end(req.method === 'GET' ? obj : undefined);++ }++ res.writeHead(405); res.end();++ });++});++++(async () => {++ await new Promise((r) => s3.listen(0, '127.0.0.1', r));++ const endpoint = `http://127.0.0.1:${s3.address().port}`;++ Object.assign(process.env, {++ DB_PATH: path.join(tmp, 'data', 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),++ MEDIA_B2_ENDPOINT: endpoint, MEDIA_B2_BUCKET: 'b2-bucket', MEDIA_B2_KEY_ID: 'k', MEDIA_B2_APP_KEY: 's', MEDIA_B2_REGION: 'us-west-004',++ MEDIA_INVARIANT_SCAN_HOURS: '0', MEDIA_JOBS_ENABLED: '0',++ });++ const db = require('../server/db/database');++ const reconcile = require('../server/vod/duration-reconcile');++ const orphans = require('../server/vod/orphans-report');++ db.upsertApp({ app_id: 'live', api_key: 'live-key' });++++ // ── 1. classify ──++ assert.strictEqual(reconcile.classify(600, 601.5), 'ok');++ assert.strictEqual(reconcile.classify(600, 640), 'mismatch', '40 s off is reported, not repaired');++ assert.strictEqual(reconcile.classify(3600, 600), 'wrong');++ assert.strictEqual(reconcile.classify(20000, 19800), 'mismatch', '200 s off a 5.5 h VOD (1%) is a mismatch, not wrong');++ assert.strictEqual(reconcile.classify(0, 12), 'missing');++ assert.strictEqual(reconcile.classify(100, 0), 'unmeasurable');++ console.log('✅ classify: ok within 2 s; wrong only past 60 s and 2%; missing when nothing is stored');++++ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;++ const mk = (file, secs) => {++ const r = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', `testsrc=duration=${secs}:size=160x120:rate=10`, '-f', 'lavfi', '-i', `sine=frequency=440:duration=${secs}`,++ '-c:v', 'libx264', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest', '-movflags', '+faststart', file]);++ assert.strictEqual(r.status, 0, String(r.stderr));++ return fs.readFileSync(file);++ };++ const insVod = (o) => Number(db.run(`INSERT INTO vods (app_id, title, file_path, file_size, duration_seconds, duration_source, storage_provider, storage_key, health_status, is_recording, is_public, visibility)++ VALUES ('live', ?, ?, ?, ?, ?, ?, ?, ?, 0, 1, 'public')`, [o.title || 't', o.file, o.size || 1000, o.stored, o.source ?? null, o.provider || 'local', o.key || null, o.health || 'ok']).lastInsertRowid);++ const vod = (id) => db.get('SELECT * FROM vods WHERE id = ?', [id]);++++ let ids = {};++ if (!hasFfmpeg) {++ console.log('⚠️ ffmpeg not found: the measurement checks are skipped (orphan report still runs)');++ } else {++ const V = process.env.VOD_PATH;++ mk(path.join(V, 'a.mp4'), 6);++ mk(path.join(V, 'b.mp4'), 6);++ const cBytes = mk(path.join(tmp, 'c.mp4'), 8);++ buckets['b2-bucket'].set('vods/c.mp4', cBytes); // offloaded: no local file++ mk(path.join(V, 'd.mp4'), 6);++ mk(path.join(V, 'e.mp4'), 6);++ fs.writeFileSync(path.join(V, 'g.mp4'), Buffer.alloc(30000, 3)); // unreadable++ ids = {++ ok: insVod({ file: path.join(V, 'a.mp4'), stored: 6, source: 'probe' }),++ wallClock: insVod({ file: path.join(V, 'b.mp4'), stored: 3600 }),++ remote: insVod({ file: '/old/host/path/c.mp4', stored: 99999, provider: 'b2', key: 'vods/c.mp4' }),++ missing: insVod({ file: path.join(V, 'd.mp4'), stored: 0 }),++ small: insVod({ file: path.join(V, 'e.mp4'), stored: 9, source: 'probe' }),++ skipped: insVod({ file: path.join(V, 'f.mp4'), stored: 50, health: 'missing_file' }),++ unreadable: insVod({ file: path.join(V, 'g.mp4'), stored: 120 }),++ };++++ // ── 2. Dry run: measures local and remote, changes nothing ──++ const before = db.all('SELECT id, duration_seconds, duration_source FROM vods ORDER BY id');++ const dry = await reconcile.reconcileBatch({ appId: 'live', afterId: 0, limit: 50 });++ assert.deepStrictEqual(db.all('SELECT id, duration_seconds, duration_source FROM vods ORDER BY id'), before, 'a dry run writes nothing');++ const by = Object.fromEntries(dry.rows.map(r => [r.vod_id, r]));++ assert.strictEqual(by[ids.ok].verdict, 'ok');++ assert.deepStrictEqual([by[ids.wallClock].verdict, by[ids.wallClock].action, by[ids.wallClock].confirmed], ['wrong', 'would_repair', true]);++ assert.strictEqual(by[ids.remote].where, 'b2', 'the offloaded copy is measured in B2');++ assert.deepStrictEqual([by[ids.remote].verdict, by[ids.remote].action], ['wrong', 'would_repair']);++ assert.ok(Math.abs(by[ids.remote].measured - 8) < 0.6, `remote measured ${by[ids.remote].measured}`);++ assert.ok(s3log.some(([m, , k]) => m === 'GET' && k === 'vods/c.mp4'), 'read through a presigned GET');++ assert.ok(!s3log.some(([m]) => m === 'PUT' || m === 'DELETE'));++ assert.deepStrictEqual([by[ids.missing].verdict, by[ids.missing].action], ['missing', 'would_repair']);++ assert.deepStrictEqual([by[ids.small].verdict, by[ids.small].action], ['mismatch', 'none'], 'a 3 s difference is reported only');++ assert.strictEqual(by[ids.skipped].verdict, 'skipped');++ assert.strictEqual(by[ids.unreadable].verdict, 'unmeasurable');++ assert.strictEqual(dry.counts.would_repair, 3);++ assert.strictEqual(dry.counts.remote, 1);++ console.log('✅ dry run: local files and the B2 copy (presigned ranged read) measured, 3 repairs proposed, nothing written');++++ // ── 3. Bounded batches and --apply through the job, then rollback ──++ const small = await reconcile.reconcileBatch({ appId: 'live', afterId: 0, limit: 2 });++ assert.deepStrictEqual([small.rows.length, small.range.last_id, small.range.done], [2, ids.wallClock, false]);++ const queue = require('../server/jobs/queue');++ const worker = require('../server/jobs/worker');++ const spec = queue.typeSpec('vod.duration.reconcile');++ assert.throws(() => spec.validate({ obj: null, params: { limit: 500 } }), /limit/);++ const params = spec.validate({ obj: null, params: { limit: 4, apply: true } });++ let j = queue.enqueue({ appId: 'live', type: 'vod.duration.reconcile', params, createdBy: 'test' }).job;++ j = await worker.runNow(j.id);++ assert.strictEqual(j.status, 'succeeded', j.error);++ let res = JSON.parse(j.result);++ assert.deepStrictEqual([res.mode, res.counts.checked, res.counts.repaired, res.next_after_id], ['apply', 4, 3, ids.missing]);++ assert.ok(fs.existsSync(path.join(path.dirname(process.env.DB_PATH), 'reports', res.report)), 'the report is written');++ assert.deepStrictEqual([vod(ids.wallClock).duration_seconds, vod(ids.wallClock).duration_source], [6, 'probe']);++ assert.deepStrictEqual([vod(ids.remote).duration_seconds, vod(ids.remote).duration_source], [8, 'probe']);++ assert.deepStrictEqual([vod(ids.missing).duration_seconds, vod(ids.missing).duration_source], [6, 'probe']);++ assert.strictEqual(db.getSetting('duration_reconcile.cursor.live'), ids.missing, 'the cursor is kept for the next run');++ const obj = db.get('SELECT metadata FROM media_objects WHERE id = ?', [vod(ids.wallClock).object_id]);++ assert.strictEqual(JSON.parse(obj.metadata).duration_seconds, 6, 'the object is re-projected');++ j = queue.enqueue({ appId: 'live', type: 'vod.duration.reconcile', params: spec.validate({ obj: null, params: { limit: 4, apply: true } }), createdBy: 'test2' }).job;++ res = JSON.parse((await worker.runNow(j.id)).result);++ assert.deepStrictEqual([res.range.after_id, res.counts.checked, res.range.done, res.next_after_id], [ids.missing, 3, true, 0], 'continues, reaches the end, wraps');++ const report = JSON.parse(fs.readFileSync(path.join(path.dirname(process.env.DB_PATH), 'reports', JSON.parse(queue.list('live', { type: 'vod.duration.reconcile' }).jobs.slice(-1)[0].result).report), 'utf8'));++ const rb0 = reconcile.rollback(report.rows);++ assert.deepStrictEqual(rb0, { restored: 0, would_restore: 3, changed_since: 0 });++ const rb = reconcile.rollback(report.rows, { apply: true });++ assert.strictEqual(rb.restored, 3);++ assert.deepStrictEqual([vod(ids.wallClock).duration_seconds, vod(ids.wallClock).duration_source], [3600, null], 'rolled back');++ console.log('✅ the job repairs confirmed values in bounded batches, keeps its cursor, writes a report; the report rolls back');++++ // ── 4. The script: dry run by default, --apply needs --backup, --rollback ──++ let r = await runCli('vod-duration-reconcile.js', ['--app', 'live', '--all', '--batch', '3']);++ assert.strictEqual(r.status, 0, r.stderr);++ assert.ok(/Would repair 3/.test(r.stdout), r.stdout);++ assert.strictEqual(vod(ids.wallClock).duration_seconds, 3600, 'dry run changed nothing');++ r = await runCli('vod-duration-reconcile.js', ['--apply']);++ assert.strictEqual(r.status, 2, 'apply without a backup is refused');++ const backup = path.join(tmp, 'bk', 'reconcile.json');++ r = await runCli('vod-duration-reconcile.js', ['--app', 'live', '--all', '--apply', '--backup', backup]);++ assert.strictEqual(r.status, 0, r.stderr);++ assert.ok(fs.existsSync(backup) && fs.existsSync(backup.replace(/\.json$/, '.media.db')), 'rollback file and database copy');++ assert.strictEqual(vod(ids.wallClock).duration_seconds, 6);++ r = await runCli('vod-duration-reconcile.js', ['--app', 'live', '--apply', '--backup', backup]);++ assert.strictEqual(r.status, 2, 'never overwrites a backup');++ r = await runCli('vod-duration-reconcile.js', ['--rollback', backup]);++ assert.ok(/Would restore 3/.test(r.stdout), r.stdout);++ r = await runCli('vod-duration-reconcile.js', ['--rollback', backup, '--apply']);++ assert.ok(/Restored 3/.test(r.stdout), r.stdout);++ assert.strictEqual(vod(ids.wallClock).duration_seconds, 3600);++ console.log('✅ scripts/vod-duration-reconcile.js: dry run by default, --apply with a checked backup, --rollback');++ }++++ // ── 5. vods-orphans/ report: matches and recommendations, read-only ──++ {++ const B = buckets['b2-bucket'];++ const V = process.env.VOD_PATH;++ const bytes = (n, fill) => Buffer.alloc(n, fill);++ // canonical present, same size → duplicate++ const dupId = insVod({ file: '/x/vod-live-501-1726900000001.mp4', stored: 10, provider: 'b2' });++ B.set('vods/vod-live-501-1726900000001.mp4', bytes(5000, 1));++ B.set('vods-orphans/vod-live-501-1726900000001.mp4', bytes(5000, 1));++ // canonical present, other size → review++ insVod({ file: '/x/vod-live-502-1726900000002.mp4', stored: 10, provider: 'b2' });++ B.set('vods/vod-live-502-1726900000002.mp4', bytes(4000, 2));++ B.set('vods-orphans/vod-live-502-1726900000002.mp4', bytes(4500, 2));++ // no canonical, no local → the only copy++ const onlyId = insVod({ file: '/x/vod-live-503-1726900000003.mp4', stored: 10, provider: 'b2' });++ B.set('vods-orphans/vod-live-503-1726900000003.mp4', bytes(3000, 3));++ // still local with the same size → keep until offloaded++ fs.writeFileSync(path.join(V, 'vod-live-504-1726900000004.mp4'), bytes(2000, 4));++ insVod({ file: path.join(V, 'vod-live-504-1726900000004.mp4'), stored: 10 });++ B.set('vods-orphans/vod-live-504-1726900000004.mp4', bytes(2000, 4));++ // the row is gone (its object says when)++ B.set('vods-orphans/vod-live-9999-1726900000005.mp4', bytes(1000, 5));++ db.run(`INSERT INTO media_objects (id, app_id, namespace, kind, lifecycle_status, legacy_ref, deleted_at) VALUES ('med_01TESTGONE000000000000000A', 'live', 'live', 'vod', 'deleted', 'legacy:live:vod:9999', '2026-09-20 10:00:00')`);++ // a name that matches nothing++ B.set('vods-orphans/notes.txt', bytes(10, 6));++ // a row whose storage_key points INTO the prefix: that object is the canonical copy++ const keyedId = insVod({ file: '/x/vod-live-506-1726900000006.mp4', stored: 10, provider: 'b2', key: 'vods-orphans/vod-live-506-1726900000006.mp4' });++ B.set('vods-orphans/vod-live-506-1726900000006.mp4', bytes(700, 7));++++ const writesBefore = s3log.filter(([m]) => m !== 'GET' && m !== 'HEAD').length;++ const rep = await orphans.buildReport({ provider: 'b2' });++ const rec = Object.fromEntries(rep.objects.map(o => [path.basename(o.key), o.recommendation]));++ assert.deepStrictEqual(rec, {++ 'notes.txt': 'review_unknown',++ 'vod-live-501-1726900000001.mp4': 'delete_duplicate',++ 'vod-live-502-1726900000002.mp4': 'review_size_differs',++ 'vod-live-503-1726900000003.mp4': 'keep_only_copy',++ 'vod-live-504-1726900000004.mp4': 'keep_until_offloaded',++ 'vod-live-506-1726900000006.mp4': 'keep_only_copy',++ 'vod-live-9999-1726900000005.mp4': 'delete_row_gone',++ });++ const gone = rep.objects.find(o => o.key.includes('9999'));++ assert.ok(/deleted 2026-09-20/.test(gone.reason), gone.reason);++ assert.strictEqual(rep.objects.find(o => o.key.includes('-501-')).match.id, dupId);++ assert.strictEqual(rep.objects.find(o => o.key.includes('-503-')).match.id, onlyId);++ assert.strictEqual(rep.objects.find(o => o.key.includes('-506-')).match.id, keyedId);++ assert.deepStrictEqual([rep.totals.objects, rep.by_recommendation.keep_only_copy.objects, rep.by_recommendation.keep_only_copy.bytes], [7, 2, 3700]);++ assert.strictEqual(rep.read_only, true);++ assert.strictEqual(s3log.filter(([m]) => m !== 'GET' && m !== 'HEAD').length, writesBefore, 'the report never writes to the bucket');++ assert.strictEqual(B.size, [...B.keys()].length);++ const out = path.join(tmp, 'orphans.json');++ const r = await runCli('vods-orphans-report.js', ['--out', out]);++ assert.strictEqual(r.status, 0, r.stderr);++ assert.ok(/7 object\(s\)/.test(r.stdout) && /Nothing was deleted/.test(r.stdout), r.stdout);++ assert.strictEqual(JSON.parse(fs.readFileSync(out, 'utf8')).totals.objects, 7);++ assert.ok(B.has('vods-orphans/vod-live-501-1726900000001.mp4'), 'every orphan is still there');++ console.log('✅ vods-orphans report: each object matched and recommended (only copy, duplicate, row gone, …), nothing deleted');++ }++++ db.close();++ s3.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('\n✅ All duration reconcile and orphan report tests passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -458,6 +458,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu
@@ -458,6 +458,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu
| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) |
| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) |
| `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part |
| `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part |
| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |
| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |
+| `vod.duration.reconcile` | heavy | One bounded batch of the tenant's finished VODs: stored duration vs a measurement of the real file, local or the B2/R2 copy (ffprobe over a presigned URL, ranged reads). `params { limit (1-200, 25), after_id?, apply (false), confirm_remote (false) }`; without `after_id` it walks the library from where the last run stopped. Repairs only confirmed, clearly wrong values when `apply`. Result `{ counts, range, next_after_id, report }`. See [Duration reconciliation](#duration-reconciliation) |
| `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |
| `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |
Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and too
Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and too
@@ -493,6 +494,45 @@ It changes the database directly (the service's worker picks approved jobs up on
@@ -493,6 +494,45 @@ It changes the database directly (the service's worker picks approved jobs up on
when the service's outbox is on, queues each change's event in the same transaction. It never runs a
when the service's outbox is on, queues each change's event in the same transaction. It never runs a
job. `media_jobs{type,status}` on `/metrics` counts jobs.
job. `media_jobs{type,status}` on `/metrics` counts jobs.
+## Duration reconciliation
+
+Code: `server/vod/duration-reconcile.js`, the job `vod.duration.reconcile`, `scripts/vod-duration-reconcile.js`.
+Test: `test/duration-reconcile.test.js` (a fake S3 through the real SDK, real ffmpeg files).
+
+Stored VOD durations are compared with a measurement of the real file, in bounded batches walked by
+id. A local file is probed where it is; an offloaded one is probed in B2/R2 through a presigned GET
+URL (ffprobe reads the index with ranged requests; nothing is downloaded). Verdicts: `ok` (within
+2 s), `mismatch` (reported only), `wrong` (more than 60 s **and** 2% off), `missing` (nothing
+stored, the file measures), `unmeasurable`, `skipped` (`zero_byte`/`missing_file`). A `wrong` or
+`missing` value is repaired only when a second measurement confirms it: for a local file a
+stream-copy pass over its packets (when the container disagrees with the packets, the packets are
+used, `duration_source = remux`); for a remote copy the container must agree with its own streams'
+durations (`--confirm-remote` / `confirm_remote` reads the whole copy instead, which is egress). A
+repair writes `duration_seconds`, `probe_duration_seconds` and `duration_source` only while the row
+still holds the value that was read, and re-projects the object. Every run writes a JSON report
+(`<data>/reports/duration-reconcile-*.json`) listing each VOD with stored and measured values.
+
+```
+node scripts/vod-duration-reconcile.js [--app live] [--batch 50] [--after <id>] [--all] [--ids 1,2] [--confirm-remote] # dry run
+node scripts/vod-duration-reconcile.js --apply --backup <file.json> [same selection] # checked DB backup first; file = rollback
+node scripts/vod-duration-reconcile.js --rollback <file.json> [--apply]
+```
+
+The job is not scheduled unless `MEDIA_DURATION_RECONCILE_HOURS` > 0, and scheduled runs repair only
+with `MEDIA_DURATION_RECONCILE_APPLY=1` (batch `MEDIA_DURATION_RECONCILE_BATCH`, 25).
+
+### `vods-orphans/` report
+
+`scripts/vods-orphans-report.js [--prefix vods-orphans/] [--provider b2] [--out file] [--json]`
+(`server/vod/orphans-report.js`) lists the B2 prefix where orphaned recordings were parked (hazard
+H3), matches each object to a VOD or clip (file name, storage key, or the recorder's
+`vod-<app>-<id>-<ms>` name), HEADs that row's canonical copy and checks its local file, and
+recommends: `keep_only_copy` (the row exists and this is its only copy: restore it, never delete),
+`keep_until_offloaded`, `delete_duplicate` (the canonical copy has the same size),
+`review_size_differs`, `delete_row_gone` (the row was deleted; its object says when),
+`review_row_mismatch`, `review_unknown`. Totals per recommendation in objects and bytes. It only
+lists and HEADs: nothing is deleted, moved or written. `delete_*` rows are for the owner to approve.
+
## R2 eviction drill
## R2 eviction drill
The exit criterion "the canonical B2 object survives R2 eviction", as a repeatable drill with a kept
The exit criterion "the canonical B2 object survives R2 eviction", as a repeatable drill with a kept
@@ -542,6 +582,8 @@ Not run against production yet: it needs the owner's go-ahead.
@@ -542,6 +582,8 @@ Not run against production yet: it needs the owner's go-ahead.
| `MEDIA_JOBS_POLL_MS` | 5000 | worker poll interval |
| `MEDIA_JOBS_POLL_MS` | 5000 | worker poll interval |
| `MEDIA_JOBS_LIGHT_CONCURRENCY` / `_HEAVY_CONCURRENCY` / `_FINALIZE_CONCURRENCY` | 2 / 1 / 1 | jobs per lane |
| `MEDIA_JOBS_LIGHT_CONCURRENCY` / `_HEAVY_CONCURRENCY` / `_FINALIZE_CONCURRENCY` | 2 / 1 / 1 | jobs per lane |
| `MEDIA_FINALIZE_ORPHAN_GRACE_S` / `MEDIA_FINALIZE_SWEEP_S` | 180 / 120 | an orphaned recording's file must be idle this long; the orphan sweep runs this often |
| `MEDIA_FINALIZE_ORPHAN_GRACE_S` / `MEDIA_FINALIZE_SWEEP_S` | 180 / 120 | an orphaned recording's file must be idle this long; the orphan sweep runs this often |
+| `MEDIA_DURATION_RECONCILE_HOURS` | 0 | schedule `vod.duration.reconcile` per tenant this often (`0` = only on demand) |
+| `MEDIA_DURATION_RECONCILE_APPLY` / `_BATCH` | off / 25 | scheduled runs repair (`1`) or only report; VODs per scheduled run |
| `MEDIA_JOBS_HEAVY_WHILE_RECORDING` | off | `1` lets split/remux run while a recording is being written |
| `MEDIA_JOBS_HEAVY_WHILE_RECORDING` | off | `1` lets split/remux run while a recording is being written |
| `MEDIA_JOBS_LEASE_S` | 120 | running-job lease (renewed by the heartbeat) |
| `MEDIA_JOBS_LEASE_S` | 120 | running-job lease (renewed by the heartbeat) |
| `MEDIA_INVARIANT_SCAN_HOURS` | 24 | the size-invariant validator's schedule per tenant (`0` = on demand only) |
| `MEDIA_INVARIANT_SCAN_HOURS` | 24 | the size-invariant validator's schedule per tenant (`0` = on demand only) |
new file mode 100755
new file mode 100755
@@ -0,0 +1,130 @@
@@ -0,0 +1,130 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * VOD duration reconciliation, for the operator (server/vod/duration-reconcile.js has the rules; the
+ * service runs the same thing as the job vod.duration.reconcile).
+ *
+ * node scripts/vod-duration-reconcile.js [--app live] [--batch 50] [--after <vod id>] [--all] [--ids 1,2,3]
+ * [--confirm-remote] [--out <report.json>] [--json]
+ * Dry run (the default): measures each finished VOD (local file, or the B2/R2 copy with ffprobe
+ * over a presigned URL, ranged reads), classifies it (ok, mismatch, wrong, missing, unmeasurable,
+ * skipped) and writes the report (default <data>/reports/duration-reconcile-…json). Changes
+ * nothing. One batch from --after (default 0); --all walks every batch to the end.
+ * Values that would be repaired are confirmed first (local: a stream-copy pass over the
+ * packets; remote: the container must agree with its streams, or --confirm-remote reads the
+ * whole copy, which is B2/R2 egress).
+ * node scripts/vod-duration-reconcile.js --apply --backup <file.json> [same selection]
+ * First an online backup of the database to <file>.media.db (checked with integrity_check),
+ * then repairs the confirmed wrong/missing values (only rows still holding the value read);
+ * <file.json> is the report, which lists every repair with its old values (the rollback file).
+ * Refuses to overwrite either file.
+ * node scripts/vod-duration-reconcile.js --rollback <file.json> [--apply]
+ * Restores the old values of the repairs the file lists, where the row still holds what the
+ * repair wrote. Without --apply it only reports what it would restore.
+ * (every command takes --db <media.db>; the database is DB_PATH otherwise)
+ *
+ * B2/R2 credentials come from the environment (the service's /etc/openvibe/media.env names).
+ * Exit codes: 0 done, 1 error, 2 bad usage.
+ */
+const fs = require('fs');
+const path = require('path');
+
+const args = process.argv.slice(2);
+const has = (name) => args.includes(`--${name}`);
+const arg = (name) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : null; };
+const usage = (msg) => { console.error(`vod-duration-reconcile: ${msg}`); process.exit(2); };
+for (const flag of ['db', 'backup', 'rollback', 'batch', 'after', 'ids', 'app', 'out']) if (has(flag) && !arg(flag)) usage(`--${flag} needs a value`);
+if (arg('db')) process.env.DB_PATH = path.resolve(arg('db'));
+
+const APPLY = has('apply');
+const JSON_OUT = has('json');
+const BACKUP = arg('backup') ? path.resolve(arg('backup')) : null;
+const ROLLBACK = arg('rollback') ? path.resolve(arg('rollback')) : null;
+const BATCH = arg('batch') ? Number(arg('batch')) : 50;
+const AFTER = arg('after') ? Number(arg('after')) : 0;
+const IDS = arg('ids') ? arg('ids').split(',').map(Number) : null;
+if (!Number.isInteger(BATCH) || BATCH < 1 || BATCH > 200) usage('--batch must be an integer 1-200');
+if (!Number.isInteger(AFTER) || AFTER < 0) usage('--after must be a vod id');
+if (IDS && IDS.some(n => !Number.isInteger(n) || n < 1)) usage('--ids must be a comma list of vod ids');
+if (ROLLBACK && BACKUP) usage('--rollback and --backup do not go together');
+if (APPLY && !ROLLBACK && !BACKUP) usage('--apply needs --backup <file.json> (the rollback file; the database copy goes next to it)');
+
+const config = require('../server/config');
+if (!fs.existsSync(config.db.path)) usage(`no database at ${config.db.path} (set DB_PATH or pass --db)`);
+const db = require('../server/db/database');
+const reconcile = require('../server/vod/duration-reconcile');
+const log = (...a) => { if (!JSON_OUT) console.log(...a); };
+
+function writePrivate(file, text) {
+ fs.mkdirSync(path.dirname(file), { recursive: true, mode: 0o700 });
+ const tmp = `${file}.tmp-${process.pid}`;
+ const fd = fs.openSync(tmp, 'wx', 0o600);
+ try { fs.writeSync(fd, text); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }
+ fs.renameSync(tmp, file);
+}
+
+async function backupDatabase(dest) {
+ fs.mkdirSync(path.dirname(dest), { recursive: true, mode: 0o700 });
+ await db.getDb().backup(dest);
+ fs.chmodSync(dest, 0o600);
+ const Database = require('better-sqlite3');
+ const copy = new Database(dest, { readonly: true, fileMustExist: true });
+ try { return { path: dest, bytes: fs.statSync(dest).size, integrity_check: copy.pragma('integrity_check', { simple: true }), vods: copy.prepare('SELECT count(*) AS n FROM vods').get().n }; } finally { copy.close(); }
+}
+
+const fmt = (s) => (s == null ? '-' : `${Math.round(s)}s`);
+function printRow(r) {
+ if (r.verdict === 'ok') return;
+ log(` vod ${String(r.vod_id).padStart(6)} ${String(r.where || '-').padEnd(5)} stored ${fmt(r.stored).padStart(8)} (${r.stored_source || '?'}) measured ${fmt(r.measured).padStart(8)} (${r.measured_source || '-'}) ${r.verdict.padEnd(12)} ${r.action}${r.note ? ` [${r.note}]` : ''}`);
+}
+
+async function main() {
+ if (ROLLBACK) {
+ let rep;
+ try { rep = JSON.parse(fs.readFileSync(ROLLBACK, 'utf8')); } catch (err) { usage(`cannot read ${ROLLBACK}: ${err.message}`); }
+ if (!rep || rep.kind !== 'media.vod.duration_reconcile' || !Array.isArray(rep.rows)) usage(`${ROLLBACK} is not a duration reconcile report`);
+ const out = reconcile.rollback(rep.rows, { apply: APPLY });
+ log(APPLY ? `Restored ${out.restored} VOD duration(s); ${out.changed_since} changed since (left alone).`
+ : `Would restore ${out.would_restore} VOD duration(s); ${out.changed_since} changed since. Re-run with --apply to restore.`);
+ if (JSON_OUT) console.log(JSON.stringify(out, null, 2));
+ return 0;
+ }
+ let dbCopy = null;
+ if (APPLY) {
+ const copyPath = BACKUP.replace(/\.json$/i, '') + '.media.db';
+ for (const f of [BACKUP, copyPath]) if (fs.existsSync(f)) usage(`${f} already exists; pick a new --backup name`);
+ log(`Backing up ${config.db.path} -> ${copyPath} ...`);
+ dbCopy = await backupDatabase(copyPath);
+ if (dbCopy.integrity_check !== 'ok') { console.error(`backup failed integrity_check: ${dbCopy.integrity_check}; nothing changed`); return 1; }
+ log(` ${dbCopy.bytes} bytes, integrity_check ok, ${dbCopy.vods} vods`);
+ }
+ const all = { kind: 'media.vod.duration_reconcile', version: 1, run_at: new Date().toISOString(), mode: APPLY ? 'apply' : 'dry-run', app_id: arg('app') || null,
+ thresholds: reconcile.THRESHOLDS, confirm_remote: has('confirm-remote'), batches: 0, range: { after_id: IDS ? null : AFTER, last_id: null, done: false }, counts: null, rows: [], db_backup: dbCopy };
+ let after = AFTER;
+ log(`${APPLY ? 'Reconciling' : 'Dry run over'} VOD durations in ${config.db.path}${all.app_id ? ` (app ${all.app_id})` : ''}${IDS ? `, vods ${IDS.join(',')}` : `, from vod id > ${AFTER}, batches of ${BATCH}`}${has('all') ? ', every batch' : ''}`);
+ for (;;) {
+ const rep = await reconcile.reconcileBatch({ appId: all.app_id, afterId: after, limit: IDS ? Math.min(200, IDS.length) : BATCH, ids: IDS, apply: APPLY, confirmRemote: has('confirm-remote'), onRow: printRow });
+ all.batches++;
+ all.rows.push(...rep.rows);
+ if (!all.counts) all.counts = { ...rep.counts };
+ else for (const k of Object.keys(rep.counts)) all.counts[k] += rep.counts[k];
+ all.range.last_id = rep.range.last_id ?? all.range.last_id;
+ all.range.done = rep.range.done;
+ if (APPLY) writePrivate(BACKUP, JSON.stringify(all, null, 1)); // the rollback file is current after every batch
+ if (IDS || rep.range.done || !has('all')) break;
+ after = rep.range.last_id;
+ }
+ const file = APPLY ? BACKUP : (arg('out') ? path.resolve(arg('out')) : null);
+ const written = APPLY ? file : reconcile.writeReport(all, file);
+ const c = all.counts;
+ log('');
+ log(`Checked ${c.checked} (${c.remote} remote): ok ${c.ok}, mismatch ${c.mismatch}, wrong ${c.wrong}, missing ${c.missing}, unmeasurable ${c.unmeasurable}, skipped ${c.skipped}.`);
+ log(APPLY ? `Repaired ${c.repaired}; ${c.unconfirmed} unconfirmed (not written); ${c.changed_since} changed while running.`
+ : `Would repair ${c.would_repair}; ${c.unconfirmed} unconfirmed (would not be written). Nothing changed.`);
+ if (!all.range.done && !IDS) log(`More to check: next run --after ${all.range.last_id} (or --all).`);
+ log(`Report: ${written}`);
+ if (JSON_OUT) console.log(JSON.stringify({ ...all, rows: undefined, report: written }, null, 2));
+ return 0;
+}
+
+main().then((code) => { db.close(); process.exit(code); }).catch((err) => { console.error(err); process.exit(1); });
new file mode 100755
new file mode 100755
@@ -0,0 +1,55 @@
@@ -0,0 +1,55 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * Report on the B2 `vods-orphans/` prefix (hazard H3). Read-only: it lists the prefix (ListObjectsV2),
+ * HEADs each match's canonical copy, reads the database, and writes a report. It never deletes, moves,
+ * copies or writes an object or a row (server/vod/orphans-report.js has the rules).
+ *
+ * node scripts/vods-orphans-report.js [--prefix vods-orphans/] [--provider b2] [--out <report.json>] [--json] [--db <media.db>]
+ *
+ * Prints one line per object (key, size, matched VOD/clip, recommendation) and the totals per
+ * recommendation, and writes the full report (default <data>/reports/vods-orphans-<time>.json).
+ * B2 credentials come from the environment (/etc/openvibe/media.env names). Exit 0 done, 1 error, 2 usage.
+ */
+const fs = require('fs');
+const path = require('path');
+
+const args = process.argv.slice(2);
+const has = (name) => args.includes(`--${name}`);
+const arg = (name) => { const i = args.indexOf(`--${name}`); return i >= 0 && args[i + 1] && !args[i + 1].startsWith('--') ? args[i + 1] : null; };
+const usage = (msg) => { console.error(`vods-orphans-report: ${msg}`); process.exit(2); };
+for (const flag of ['prefix', 'provider', 'out', 'db']) if (has(flag) && !arg(flag)) usage(`--${flag} needs a value`);
+if (arg('db')) process.env.DB_PATH = path.resolve(arg('db'));
+const provider = arg('provider') || 'b2';
+if (!['b2', 'r2'].includes(provider)) usage('--provider must be b2 or r2');
+
+const config = require('../server/config');
+if (!fs.existsSync(config.db.path)) usage(`no database at ${config.db.path} (set DB_PATH or pass --db)`);
+const db = require('../server/db/database');
+const vodStorage = require('../server/vod/vod-storage');
+const orphans = require('../server/vod/orphans-report');
+const JSON_OUT = has('json');
+const log = (...a) => { if (!JSON_OUT) console.log(...a); };
+const mb = (b) => `${(b / 1048576).toFixed(1)} MB`;
+
+(async () => {
+ if (!vodStorage.providerConfigured(provider)) usage(`${provider} is not configured (MEDIA_${provider.toUpperCase()}_* not in the environment)`);
+ const report = await orphans.buildReport({ provider, prefix: arg('prefix') || orphans.PREFIX });
+ const file = arg('out') ? path.resolve(arg('out')) : path.join(path.dirname(config.db.path), 'reports', `vods-orphans-${report.generated_at.replace(/[:.]/g, '-')}.json`);
+ fs.mkdirSync(path.dirname(file), { recursive: true });
+ fs.writeFileSync(file, JSON.stringify(report, null, 1));
+ log(`${report.provider}://${report.bucket}/${report.prefix}: ${report.totals.objects} object(s), ${report.totals.gb} GB (read-only report)`);
+ log('');
+ for (const o of report.objects) {
+ const m = o.match ? `${o.match.kind} ${o.match.id}${o.match.row === 'gone' ? ' (row gone)' : ''}` : '-';
+ log(` ${o.key.padEnd(60)} ${mb(o.size).padStart(11)} ${m.padEnd(22)} ${o.recommendation.padEnd(20)} ${o.reason}`);
+ }
+ log('');
+ for (const [rec, t] of Object.entries(report.by_recommendation)) if (t.objects) log(` ${rec.padEnd(22)} ${String(t.objects).padStart(4)} object(s) ${mb(t.bytes).padStart(12)}`);
+ log('');
+ log('Nothing was deleted or moved. delete_* rows are candidates for the owner; keep_only_copy rows are recordings whose only copy is here.');
+ log(`Report: ${file}`);
+ if (JSON_OUT) console.log(JSON.stringify({ ...report, report: file }, null, 2));
+ db.close();
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
new file mode 100644
new file mode 100644
@@ -0,0 +1,80 @@
@@ -0,0 +1,80 @@
+/**
+ * OpenVibe.Media — job type vod.duration.reconcile (heavy lane): one bounded batch of the tenant's
+ * finished VODs, stored duration vs a measurement of the real file (local, or the B2/R2 copy through
+ * a presigned ranged read). server/vod/duration-reconcile.js has the rules.
+ *
+ * params { limit (1-200, default 25), after_id?, apply (default false), confirm_remote (default false) }
+ * result { mode, counts, range, next_after_id, report } report = the file name under <data>/reports/
+ *
+ * Without after_id the job walks the tenant's library: it starts where the previous run stopped
+ * (media_settings duration_reconcile.cursor.<app>) and wraps to the start when it reaches the end.
+ * apply=false only reports. Scheduling is off unless MEDIA_DURATION_RECONCILE_HOURS > 0; scheduled
+ * runs repair only with MEDIA_DURATION_RECONCILE_APPLY=1 (batch MEDIA_DURATION_RECONCILE_BATCH, 25).
+ */
+'use strict';
+
+const path = require('path');
+const db = require('../db/database');
+const queue = require('./queue');
+const reconcile = require('../vod/duration-reconcile');
+
+const { JobError } = queue;
+const TYPE = 'vod.duration.reconcile';
+const cursorKey = (appId) => `duration_reconcile.cursor.${appId}`;
+const bool = (v) => v === true || v === 1 || ['1', 'true', 'on', 'yes'].includes(String(v).toLowerCase());
+
+function envInt(name, fallback) {
+ const v = parseInt(process.env[name] || '', 10);
+ return Number.isFinite(v) ? v : fallback;
+}
+
+function validate({ obj, params }) {
+ if (obj) throw new JobError('media.job.invalid', `${TYPE} covers the tenant's VODs: leave object_id out`, { permanent: true });
+ const p = params || {};
+ const limit = p.limit == null ? 25 : Number(p.limit);
+ if (!Number.isInteger(limit) || limit < 1 || limit > reconcile.MAX_BATCH) throw new JobError('media.job.invalid', `params.limit must be 1-${reconcile.MAX_BATCH}`, { permanent: true });
+ const out = { limit, apply: bool(p.apply), confirm_remote: bool(p.confirm_remote) };
+ if (p.after_id != null) {
+ const a = Number(p.after_id);
+ if (!Number.isInteger(a) || a < 0) throw new JobError('media.job.invalid', 'params.after_id must be a non-negative integer', { permanent: true });
+ out.after_id = a;
+ }
+ return out;
+}
+
+async function run(job, ctx = {}) {
+ const p = validate({ obj: null, params: job.params });
+ const walking = p.after_id == null;
+ const afterId = walking ? (Number(db.getSetting(cursorKey(job.app_id))) || 0) : p.after_id;
+ const report = await reconcile.reconcileBatch({
+ appId: job.app_id, afterId, limit: p.limit, apply: p.apply, confirmRemote: p.confirm_remote, signal: ctx.signal,
+ });
+ report.job_id = job.id;
+ const file = reconcile.writeReport(report);
+ const next = report.range.done ? 0 : report.range.last_id;
+ if (walking) db.setSetting(cursorKey(job.app_id), next, 'number');
+ return { mode: report.mode, counts: report.counts, range: report.range, next_after_id: next, report: path.basename(file) };
+}
+
+/** Queue one run per tenant with VODs every MEDIA_DURATION_RECONCILE_HOURS (0 = never). Returns how many. */
+function schedule(nowMs = Date.now()) {
+ const hours = envInt('MEDIA_DURATION_RECONCILE_HOURS', 0);
+ if (!(hours > 0)) return 0;
+ const period = Math.floor(nowMs / (hours * 3600 * 1000));
+ const apply = bool(process.env.MEDIA_DURATION_RECONCILE_APPLY || '');
+ const limit = Math.min(reconcile.MAX_BATCH, Math.max(1, envInt('MEDIA_DURATION_RECONCILE_BATCH', 25)));
+ let n = 0;
+ for (const { app_id: appId } of db.all('SELECT DISTINCT app_id FROM vods WHERE COALESCE(is_recording, 0) = 0 AND file_path IS NOT NULL')) {
+ try {
+ const r = queue.enqueue({ appId, type: TYPE, params: { limit, apply, confirm_remote: false }, idempotencyKey: `${TYPE}:${hours}h:${period}`, createdBy: 'system:schedule' });
+ if (r.created) n++;
+ } catch (err) { console.warn(`[Jobs] could not schedule ${TYPE} for ${appId}: ${err.message}`); }
+ }
+ return n;
+}
+
+module.exports = {
+ TYPE,
+ spec: { lane: 'heavy', maxAttempts: 2, timeoutMs: 60 * 60 * 1000, needsObject: false, validate, run },
+ schedule, cursorKey,
+};
@@ -6,6 +6,7 @@
@@ -6,6 +6,7 @@
* object.split heavy stream-copy parts of a large vod/clip as new private objects
* object.split heavy stream-copy parts of a large vod/clip as new private objects
* object.remux heavy a stream-copy remux of a vod/clip as a new private object
* object.remux heavy a stream-copy remux of a vod/clip as a new private object
* vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff
* vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff
+ * vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch
*/
*/
'use strict';
'use strict';
@@ -16,5 +17,6 @@ queue.register('invariant.scan', require('./invariant-scan').spec);
@@ -16,5 +17,6 @@ queue.register('invariant.scan', require('./invariant-scan').spec);
queue.register('object.split', require('./derive').split);
queue.register('object.split', require('./derive').split);
queue.register('object.remux', require('./derive').remux);
queue.register('object.remux', require('./derive').remux);
queue.register('vod.finalize', require('./vod-finalize').spec);
queue.register('vod.finalize', require('./vod-finalize').spec);
+queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);
module.exports = { names: () => queue.typeNames() };
module.exports = { names: () => queue.typeNames() };
@@ -144,7 +144,11 @@ async function tick() {
@@ -144,7 +144,11 @@ async function tick() {
kicked = false;
kicked = false;
queue.recoverInterrupted({ except: new Set(running.keys()) });
queue.recoverInterrupted({ except: new Set(running.keys()) });
const now = Date.now();
const now = Date.now();
- if (now - lastScheduleAt > 10 * 60 * 1000) { lastScheduleAt = now; scheduleInvariantScans(now); }
+ if (now - lastScheduleAt > 10 * 60 * 1000) {
+ lastScheduleAt = now;
+ scheduleInvariantScans(now);
+ try { require('./duration-reconcile').schedule(now); } catch (err) { console.warn('[Jobs] duration reconcile schedule:', err.message); }
+ }
try { require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }
try { require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }
if (now - lastPruneAt > 6 * 3600 * 1000) {
if (now - lastPruneAt > 6 * 3600 * 1000) {
lastPruneAt = now;
lastPruneAt = now;
new file mode 100644
new file mode 100644
@@ -0,0 +1,201 @@
@@ -0,0 +1,201 @@
+/**
+ * OpenVibe.Media — VOD duration reconciliation (job vod.duration.reconcile; scripts/vod-duration-reconcile.js)
+ *
+ * Compares each finished VOD's stored duration with a measurement of its real file, in bounded
+ * batches walked by id, and repairs the values that are clearly wrong. Offloaded VODs are measured
+ * where they are: ffprobe reads the B2/R2 copy through a presigned HTTPS URL with ranged requests
+ * (the file is not downloaded). Every run writes a report.
+ *
+ * Verdicts (stored vs measured, seconds):
+ * ok |diff| <= 2
+ * mismatch a small difference (reported, left alone)
+ * wrong |diff| > 60 AND > 2% of the measured length: repairable
+ * missing nothing stored (<= 0) but the file measures: repairable
+ * unmeasurable no measurement (the file is unreadable, or neither copy can be reached)
+ * skipped nothing to measure (zero_byte / missing_file health, no file)
+ *
+ * A repair needs a confirmed measurement: a local file's probe must agree with a stream-copy pass
+ * over its packets (or the packets are used, source remux, when the header lies); a remote copy's
+ * container duration must agree with its own streams' durations (--confirm-remote reads the whole
+ * remote file instead). An unconfirmed value is reported, never written. A repair writes
+ * duration_seconds, probe_duration_seconds and duration_source only if the row still holds the value
+ * that was read, and re-projects the VOD's object.
+ */
+'use strict';
+
+const fs = require('fs');
+const path = require('path');
+const config = require('../config');
+const db = require('../db/database');
+const tools = require('./media-tools');
+
+const THRESHOLDS = Object.freeze({ mismatchS: 2, wrongAbsS: 60, wrongRel: 0.02, agreeAbsS: 5, agreeRel: 0.01, streamsAgreeRel: 0.05 });
+const MAX_BATCH = 200;
+
+const vodStorage = () => require('./vod-storage');
+
+function classify(stored, measured, t = THRESHOLDS) {
+ const s = Number(stored) || 0;
+ if (!(measured > 0)) return 'unmeasurable';
+ if (s <= 0) return 'missing';
+ const diff = Math.abs(s - measured);
+ if (diff <= t.mismatchS) return 'ok';
+ if (diff > t.wrongAbsS && diff > t.wrongRel * measured) return 'wrong';
+ return 'mismatch';
+}
+
+const agree = (a, b, t = THRESHOLDS) => a > 0 && b > 0 && Math.abs(a - b) <= Math.max(t.agreeAbsS, t.agreeRel * Math.max(a, b));
+
+/** The longest stream duration ffprobe reported (0 when none). */
+function streamsDuration(streams) {
+ let max = 0;
+ for (const st of streams || []) { const d = parseFloat(st && st.duration); if (Number.isFinite(d) && d > max) max = d; }
+ return max;
+}
+
+/**
+ * Where to read a VOD from: { kind: 'file', value, where: 'local' } or { kind: 'url', value,
+ * where: 'b2' | 'r2' } (a presigned GET: ffprobe range-reads it), or null.
+ */
+async function sourceFor(vod) {
+ const vs = vodStorage();
+ const local = vs.localPathForVod(vod);
+ if (fs.existsSync(local)) return { kind: 'file', value: local, where: 'local' };
+ if (vod.file_path && fs.existsSync(vod.file_path)) return { kind: 'file', value: vod.file_path, where: 'local' };
+ if (vs.isRemote(vod)) {
+ const plan = await vs.resolvePlayback(vod);
+ if (plan && plan.kind === 'redirect') return { kind: 'url', value: plan.url, where: plan.provider };
+ if (plan && plan.kind === 'file') return { kind: 'file', value: plan.path, where: 'local' };
+ }
+ return null;
+}
+
+/**
+ * Measure one VOD. Returns { where, seconds, source: probe|remux|null, confirmed, note }.
+ * `confirm` runs the second, independent measurement (only needed when a repair is on the table).
+ */
+async function measure(vod, { confirm = false, confirmRemote = false, probeTimeoutMs } = {}) {
+ const src = await sourceFor(vod);
+ if (!src) return { where: null, seconds: 0, source: null, confirmed: false, note: 'no local file and no reachable remote copy' };
+ const remote = src.kind === 'url';
+ const probe = await tools.probeDuration(src.value, { timeoutMs: probeTimeoutMs || (remote ? 60000 : 20000) });
+ const out = { where: src.where, seconds: probe.seconds || 0, source: probe.seconds > 0 ? 'probe' : null, confirmed: false, note: null };
+ if (!probe.ok) { out.note = `ffprobe: ${probe.error}`; }
+ if (!confirm) return out;
+ if (!remote || confirmRemote) {
+ // Read every packet: the media's real length, independent of the container header.
+ const packets = await tools.streamCopyDuration(src.value);
+ if (!packets.ok) {
+ out.note = `${out.note ? `${out.note}; ` : ''}stream-copy pass: ${packets.error}`;
+ return out;
+ }
+ if (out.seconds > 0 && agree(out.seconds, packets.seconds)) { out.confirmed = true; return out; }
+ // The header disagrees with (or lacks) the packet timeline: the packets are the truth.
+ out.note = out.seconds > 0 ? `container says ${out.seconds.toFixed(1)} s, packets ${packets.seconds.toFixed(1)} s: packets used` : 'no container duration: packets used';
+ out.seconds = packets.seconds;
+ out.source = 'remux';
+ out.confirmed = true;
+ return out;
+ }
+ // Remote, no full read: the container's duration must agree with its own streams'.
+ const sd = streamsDuration(probe.streams);
+ if (out.seconds > 0 && sd > 0 && Math.abs(out.seconds - sd) <= Math.max(THRESHOLDS.agreeAbsS, THRESHOLDS.streamsAgreeRel * out.seconds)) {
+ out.confirmed = true;
+ } else {
+ out.note = `${out.note ? `${out.note}; ` : ''}remote container ${out.seconds.toFixed(1)} s vs streams ${sd.toFixed(1)} s: not confirmed (--confirm-remote reads the whole copy)`;
+ }
+ return out;
+}
+
+function reportsDir() {
+ return path.join(path.dirname(config.db.path), 'reports');
+}
+
+function writeReport(report, file = null) {
+ const out = file || path.join(reportsDir(), `duration-reconcile-${report.app_id || 'all'}-${report.run_at.replace(/[:.]/g, '-')}.json`);
+ fs.mkdirSync(path.dirname(out), { recursive: true });
+ fs.writeFileSync(out, JSON.stringify(report, null, 1));
+ return out;
+}
+
+/** Write one repair (only if the row still has the value we read). Returns true when it changed. */
+function applyRepair(row) {
+ const changed = db.run(`UPDATE vods SET duration_seconds = ?, probe_duration_seconds = ?, duration_source = ?
+ WHERE id = ? AND COALESCE(duration_seconds, 0) = ? AND COALESCE(duration_source, '') = ? AND COALESCE(is_recording, 0) = 0`,
+ [Math.round(row.measured), row.measured, row.measured_source, row.vod_id, row.stored, row.stored_source || '']).changes > 0;
+ if (changed) require('../objects/model').safeSync('vod', row.vod_id);
+ return changed;
+}
+
+/** Undo repairs listed in a report (only rows that still hold what the repair wrote). */
+function rollback(rows, { apply = false } = {}) {
+ const out = { restored: 0, would_restore: 0, changed_since: 0 };
+ for (const r of rows || []) {
+ if (r.action !== 'repaired') continue;
+ const cur = db.get('SELECT duration_seconds, duration_source FROM vods WHERE id = ?', [r.vod_id]);
+ if (!cur || Number(cur.duration_seconds) !== Math.round(r.measured) || (cur.duration_source || '') !== (r.measured_source || '')) { out.changed_since++; continue; }
+ if (!apply) { out.would_restore++; continue; }
+ db.run('UPDATE vods SET duration_seconds = ?, probe_duration_seconds = ?, duration_source = ? WHERE id = ?',
+ [r.stored, r.stored_probe ?? r.stored, r.stored_source || null, r.vod_id]);
+ require('../objects/model').safeSync('vod', r.vod_id);
+ out.restored++;
+ }
+ return out;
+}
+
+/**
+ * Reconcile one bounded batch. opts: appId (null = every tenant), afterId (id cursor), limit
+ * (1-200), ids (exactly these VODs), apply, confirmRemote, onRow(row).
+ * Returns the report: { run_at, mode, app_id, thresholds, range, counts, rows }.
+ */
+async function reconcileBatch({ appId = null, afterId = 0, limit = 50, ids = null, apply = false, confirmRemote = false, onRow = null, signal = null } = {}) {
+ const n = Math.min(MAX_BATCH, Math.max(1, Number(limit) || 50));
+ const conds = ['COALESCE(is_recording, 0) = 0', 'COALESCE(clips_only, 0) = 0', 'file_path IS NOT NULL'];
+ const params = [];
+ if (appId) { conds.push('app_id = ?'); params.push(appId); }
+ if (ids && ids.length) { conds.push(`id IN (${ids.map(() => '?').join(', ')})`); params.push(...ids.map(Number)); } else { conds.push('id > ?'); params.push(Number(afterId) || 0); }
+ const vods = db.all(`SELECT * FROM vods WHERE ${conds.join(' AND ')} ORDER BY id LIMIT ?`, [...params, n]);
+ const report = {
+ kind: 'media.vod.duration_reconcile', version: 1, run_at: new Date().toISOString(), mode: apply ? 'apply' : 'dry-run',
+ app_id: appId, thresholds: THRESHOLDS, confirm_remote: !!confirmRemote,
+ range: { after_id: ids ? null : Number(afterId) || 0, last_id: vods.length ? vods[vods.length - 1].id : null, limit: n, done: vods.length < n },
+ counts: { checked: 0, ok: 0, mismatch: 0, wrong: 0, missing: 0, unmeasurable: 0, skipped: 0, repaired: 0, would_repair: 0, unconfirmed: 0, changed_since: 0, remote: 0 },
+ rows: [],
+ };
+ for (const vod of vods) {
+ if (signal && signal.aborted) break;
+ report.counts.checked++;
+ const stored = Number(vod.duration_seconds) || 0;
+ const row = {
+ vod_id: vod.id, app_id: vod.app_id, storage_provider: vod.storage_provider || 'local', health_status: vod.health_status || null,
+ stored, stored_probe: Number(vod.probe_duration_seconds) || 0, stored_source: vod.duration_source || null,
+ where: null, measured: 0, measured_source: null, diff: null, verdict: null, confirmed: false, action: 'none', note: null,
+ };
+ if (['zero_byte', 'missing_file'].includes(vod.health_status)) {
+ row.verdict = 'skipped'; row.note = `health ${vod.health_status}`;
+ } else {
+ let m = await measure(vod);
+ row.where = m.where;
+ if (m.where && m.where !== 'local') report.counts.remote++;
+ let verdict = classify(stored, m.seconds);
+ if (verdict === 'wrong' || verdict === 'missing') {
+ m = await measure(vod, { confirm: true, confirmRemote });
+ verdict = classify(stored, m.seconds);
+ }
+ Object.assign(row, { measured: m.seconds ? Number(m.seconds.toFixed(3)) : 0, measured_source: m.source, confirmed: m.confirmed, note: m.note, verdict });
+ row.diff = m.seconds > 0 ? Number((m.seconds - stored).toFixed(3)) : null;
+ if (verdict === 'wrong' || verdict === 'missing') {
+ if (!m.confirmed) { row.action = 'unconfirmed'; report.counts.unconfirmed++; }
+ else if (!apply) { row.action = 'would_repair'; report.counts.would_repair++; }
+ else if (applyRepair(row)) { row.action = 'repaired'; report.counts.repaired++; }
+ else { row.action = 'changed_since'; report.counts.changed_since++; }
+ }
+ }
+ report.counts[row.verdict]++;
+ report.rows.push(row);
+ if (onRow) onRow(row);
+ }
+ return report;
+}
+
+module.exports = { THRESHOLDS, MAX_BATCH, classify, agree, streamsDuration, sourceFor, measure, reconcileBatch, applyRepair, rollback, writeReport, reportsDir };
new file mode 100644
new file mode 100644
@@ -0,0 +1,139 @@
@@ -0,0 +1,139 @@
+/**
+ * OpenVibe.Media — report on the B2 `vods-orphans/` prefix (hazard H3: 55 objects, 14.5 GB on
+ * 2026-09-21). Report only: it lists, matches and recommends; it never deletes, moves or writes.
+ *
+ * Each object under the prefix is matched to a VOD or clip by its file name (the basename of
+ * vods.file_path / storage_key, clips.file_path; else the vod-<app>-<id>-<ms>.<ext> name the
+ * recorder gives recordings), and checked against that row's canonical copy (HEAD vods/<basename>
+ * on the same provider), its R2 copy and its local file. Recommendations:
+ *
+ * keep_only_copy the row exists and no canonical, R2 or local copy does: this IS the recording
+ * (restore it to its canonical key; never delete)
+ * keep_until_offloaded no canonical copy yet, but the local file has the same size
+ * delete_duplicate the canonical copy exists with the same size
+ * review_size_differs a copy exists but with another size (keep until someone compares them)
+ * delete_row_gone the name points at a VOD/clip id whose row no longer exists (deleted); its
+ * object says when, if the model saw it
+ * review_row_mismatch the id exists but its row points at another file
+ * review_unknown the name matches nothing
+ * `delete_*` are candidates for the owner to approve, never acted on here.
+ */
+'use strict';
+
+const fs = require('fs');
+const path = require('path');
+const db = require('../db/database');
+
+const PREFIX = 'vods-orphans/';
+const RECOMMENDATIONS = ['keep_only_copy', 'keep_until_offloaded', 'delete_duplicate', 'review_size_differs', 'delete_row_gone', 'review_row_mismatch', 'review_unknown'];
+
+const vodStorage = () => require('./vod-storage');
+
+/** vod-<app>-<id>-<epoch ms>.<ext> (+ .master.mkv) → { app, id, recorded_at } or null. */
+function parseRecordingName(base) {
+ const m = /^vod-(.+)-(\d+)-(\d{12,14})(?:\.master)?\.[a-z0-9]+$/i.exec(String(base || ''));
+ if (!m) return null;
+ return { app: m[1], id: Number(m[2]), recorded_at: new Date(Number(m[3])).toISOString() };
+}
+
+function rowByBasename(base) {
+ const vod = db.getVodByFileBasename(base);
+ if (vod) return { kind: 'vod', row: vod };
+ const byKey = db.all('SELECT * FROM vods WHERE storage_key LIKE ?', [`%${base}`]).find(r => path.basename(r.storage_key || '') === base);
+ if (byKey) return { kind: 'vod', row: byKey };
+ const clip = db.getClipByFileBasename(base);
+ if (clip) return { kind: 'clip', row: clip };
+ return null;
+}
+
+function objectFor(kind, app, id) {
+ try { return db.get('SELECT id, lifecycle_status, deleted_at FROM media_objects WHERE legacy_ref = ?', [`legacy:${app}:${kind}:${id}`]); } catch { return null; }
+}
+
+async function head(provider, key) {
+ const vs = vodStorage();
+ if (!vs.providerConfigured(provider)) return { configured: false };
+ try { const h = await vs.headObject(provider, key); return h ? { configured: true, size: Number(h.size) } : { configured: true, size: null }; } catch (err) { return { configured: true, error: err.message }; }
+}
+
+function localSize(row) {
+ const vs = vodStorage();
+ for (const p of [vs.localPathForVod(row), row.file_path].filter(Boolean)) {
+ try { const st = fs.statSync(p); if (st.isFile()) return st.size; } catch { /* next */ }
+ }
+ return null;
+}
+
+/** One orphan object → its row, the copies that exist, and a recommendation. */
+async function assess(obj, { provider = 'b2' } = {}) {
+ const vs = vodStorage();
+ const base = path.basename(obj.key);
+ const parsed = parseRecordingName(base);
+ const out = { key: obj.key, size: obj.size, last_modified: obj.last_modified, name: parsed, match: null, copies: {}, recommendation: null, reason: null };
+ const found = rowByBasename(base);
+ if (!found) {
+ if (parsed) {
+ const byId = db.get('SELECT * FROM vods WHERE id = ? AND app_id = ?', [parsed.id, parsed.app]);
+ if (byId) {
+ out.match = { kind: 'vod', id: byId.id, app_id: byId.app_id, file: path.basename(byId.file_path || ''), storage_provider: byId.storage_provider || 'local' };
+ out.recommendation = 'review_row_mismatch';
+ out.reason = `vod ${byId.id} exists but points at ${out.match.file || 'no file'}`;
+ return out;
+ }
+ const o = objectFor('vod', parsed.app, parsed.id);
+ out.match = { kind: 'vod', id: parsed.id, app_id: parsed.app, row: 'gone', object: o ? { id: o.id, lifecycle_status: o.lifecycle_status, deleted_at: o.deleted_at } : null };
+ out.recommendation = 'delete_row_gone';
+ out.reason = `vod ${parsed.id} (${parsed.app}) no longer exists${o && o.deleted_at ? ` (deleted ${o.deleted_at})` : ''}; recorded ${parsed.recorded_at}`;
+ return out;
+ }
+ out.recommendation = 'review_unknown';
+ out.reason = 'the name matches no VOD or clip';
+ return out;
+ }
+ const { kind, row } = found;
+ const canonicalKey = vs.keyForVod(row);
+ out.match = { kind, id: row.id, app_id: row.app_id, storage_provider: row.storage_provider || 'local', canonical_key: canonicalKey, health_status: row.health_status || null,
+ is_public: row.is_public, visibility: row.visibility || null, duration_seconds: row.duration_seconds || 0 };
+ if (canonicalKey === obj.key) {
+ // The row's storage_key points into the orphan prefix: this object is its canonical copy.
+ out.recommendation = 'keep_only_copy';
+ out.reason = `${kind} ${row.id}'s storage_key is this object: it is the canonical copy (never delete)`;
+ return out;
+ }
+ out.copies.canonical = await head(provider, canonicalKey);
+ if ((row.storage_provider || 'local') === 'r2') out.copies.r2 = await head('r2', canonicalKey);
+ out.copies.local = localSize(row);
+ const sizes = [out.copies.canonical && out.copies.canonical.size, out.copies.r2 && out.copies.r2.size].filter(s => s != null);
+ if (sizes.includes(obj.size)) {
+ out.recommendation = 'delete_duplicate';
+ out.reason = `${kind} ${row.id}'s canonical copy ${canonicalKey} exists with the same size`;
+ } else if (sizes.length || (out.copies.local != null && out.copies.local !== obj.size)) {
+ out.recommendation = 'review_size_differs';
+ out.reason = `${kind} ${row.id} has a copy of another size (${[...sizes, out.copies.local].filter(s => s != null).join(', ')} vs ${obj.size} bytes)`;
+ } else if (out.copies.local === obj.size) {
+ out.recommendation = 'keep_until_offloaded';
+ out.reason = `${kind} ${row.id} is still local (same size); the orphan is redundant once it is offloaded to ${canonicalKey}`;
+ } else {
+ out.recommendation = 'keep_only_copy';
+ out.reason = `the only copy of ${kind} ${row.id}: restore it to ${canonicalKey} (never delete)`;
+ }
+ return out;
+}
+
+/** The whole report: { prefix, provider, bucket, generated_at, totals, by_recommendation, objects }. */
+async function buildReport({ provider = 'b2', prefix = PREFIX, list = null } = {}) {
+ const vs = vodStorage();
+ const objects = list || await vs.listObjects(provider, prefix);
+ const rows = [];
+ for (const o of objects) rows.push(await assess(o, { provider }));
+ const by = Object.fromEntries(RECOMMENDATIONS.map(r => [r, { objects: 0, bytes: 0 }]));
+ let bytes = 0;
+ for (const r of rows) { by[r.recommendation].objects++; by[r.recommendation].bytes += r.size; bytes += r.size; }
+ return {
+ kind: 'media.vods_orphans.report', version: 1, generated_at: new Date().toISOString(), provider, bucket: vs.bucketFor(provider), prefix,
+ read_only: true, totals: { objects: rows.length, bytes, gb: Number((bytes / 1e9).toFixed(2)) }, by_recommendation: by, objects: rows,
+ note: 'Report only: nothing was deleted, moved or written. delete_* rows are candidates for the owner to approve.',
+ };
+}
+
+module.exports = { PREFIX, RECOMMENDATIONS, parseRecordingName, assess, buildReport };
@@ -316,6 +316,21 @@ async function deleteLegacyPasteScreenshot(pasteId) {
@@ -316,6 +316,21 @@ async function deleteLegacyPasteScreenshot(pasteId) {
}
}
}
}
+/** Every object under `prefix` in one provider's bucket: [{ key, size, last_modified }] (read-only). */
+async function listObjects(provider, prefix) {
+ const client = clientFor(provider);
+ if (!client) throw new Error(`${provider} is not configured`);
+ loadSdk();
+ const out = [];
+ let token;
+ do {
+ const r = await client.send(new S3.ListObjectsV2Command({ Bucket: PROVIDER_ENV[provider].bucket, Prefix: prefix, ContinuationToken: token, MaxKeys: 1000 }));
+ for (const o of (r.Contents || [])) out.push({ key: o.Key, size: Number(o.Size || 0), last_modified: o.LastModified ? new Date(o.LastModified).toISOString() : null });
+ token = r.IsTruncated ? r.NextContinuationToken : null;
+ } while (token);
+ return out;
+}
+
async function presignGet(provider, key, expiresInSeconds = 900) {
async function presignGet(provider, key, expiresInSeconds = 900) {
const client = clientFor(provider);
const client = clientFor(provider);
if (!client) return null;
if (!client) return null;
@@ -1271,6 +1286,7 @@ module.exports = {
@@ -1271,6 +1286,7 @@ module.exports = {
bucketFor,
bucketFor,
endpointFor,
endpointFor,
headObject,
headObject,
+ listObjects,
keyForVod,
keyForVod,
localPathForVod,
localPathForVod,
resolvePlayback,
resolvePlayback,
new file mode 100644
new file mode 100644
@@ -0,0 +1,253 @@
@@ -0,0 +1,253 @@
+'use strict';
+// vod.duration.reconcile (server/vod/duration-reconcile.js, the job, scripts/vod-duration-reconcile.js)
+// and the vods-orphans/ report (server/vod/orphans-report.js, scripts/vods-orphans-report.js), against
+// a fake S3 spoken to by the real AWS SDK: stored durations are compared with a measurement of the
+// real file, local or offloaded (ffprobe over a presigned URL, ranged reads); only confirmed, clearly
+// wrong values are repaired; the dry run changes nothing; --apply needs a backup and the report
+// rolls back; the job walks the library in bounded batches. The orphan report only reads.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const { spawnSync, execFile } = require('child_process');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-reconcile-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+const runCli = (script, args) => new Promise((resolve) => {
+ execFile(process.execPath, [path.join(__dirname, '../scripts', script), ...args], { env: process.env, encoding: 'utf8' },
+ (err, stdout, stderr) => resolve({ status: err ? err.code : 0, stdout, stderr }));
+});
+
+// ── Fake S3: path-style /<bucket>/<key>; HEAD, GET (Range), ListObjectsV2; PUT/DELETE recorded ──
+const buckets = { 'b2-bucket': new Map() };
+const s3log = [];
+const xml = (s) => String(s).replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>');
+const s3 = http.createServer((req, res) => {
+ req.resume();
+ req.on('end', () => {
+ const u = new URL(req.url, 'http://x');
+ const [, bucket, ...rest] = decodeURIComponent(u.pathname).split('/');
+ const key = rest.join('/');
+ const store = buckets[bucket];
+ s3log.push([req.method, bucket, key]);
+ if (!store) { res.writeHead(404); return res.end(); }
+ if (req.method === 'HEAD' && !key) { res.writeHead(200); return res.end(); } // HeadBucket
+ if (req.method === 'GET' && !key && u.searchParams.get('list-type') === '2') {
+ const prefix = u.searchParams.get('prefix') || '';
+ const keys = [...store.keys()].filter(k => k.startsWith(prefix)).sort();
+ res.writeHead(200, { 'Content-Type': 'application/xml' });
+ return res.end(`<?xml version="1.0" encoding="UTF-8"?><ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>${bucket}</Name><Prefix>${xml(prefix)}</Prefix><KeyCount>${keys.length}</KeyCount><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated>${
+ keys.map(k => `<Contents><Key>${xml(k)}</Key><LastModified>2026-09-21T00:00:00.000Z</LastModified><ETag>"e"</ETag><Size>${store.get(k).length}</Size><StorageClass>STANDARD</StorageClass></Contents>`).join('')}</ListBucketResult>`);
+ }
+ const obj = store.get(key);
+ if (req.method === 'HEAD' || req.method === 'GET') {
+ if (!obj) { res.writeHead(404, { 'Content-Type': 'application/xml' }); return res.end(req.method === 'GET' ? '<Error><Code>NoSuchKey</Code></Error>' : undefined); }
+ const m = /bytes=(\d+)-(\d+)?/.exec(req.headers.range || '');
+ if (m && req.method === 'GET') {
+ const start = Number(m[1]), end = Math.min(obj.length - 1, m[2] != null ? Number(m[2]) : obj.length - 1);
+ if (start >= obj.length) { res.writeHead(416, { 'Content-Range': `bytes */${obj.length}` }); return res.end(); }
+ res.writeHead(206, { 'Content-Length': end - start + 1, 'Content-Range': `bytes ${start}-${end}/${obj.length}`, 'Accept-Ranges': 'bytes', ETag: '"e"', 'Content-Type': 'video/mp4' });
+ return res.end(obj.subarray(start, end + 1));
+ }
+ res.writeHead(200, { 'Content-Length': obj.length, 'Accept-Ranges': 'bytes', ETag: '"e"', 'Content-Type': 'video/mp4' });
+ return res.end(req.method === 'GET' ? obj : undefined);
+ }
+ res.writeHead(405); res.end();
+ });
+});
+
+(async () => {
+ await new Promise((r) => s3.listen(0, '127.0.0.1', r));
+ const endpoint = `http://127.0.0.1:${s3.address().port}`;
+ Object.assign(process.env, {
+ DB_PATH: path.join(tmp, 'data', 'media.db'), VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
+ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),
+ MEDIA_B2_ENDPOINT: endpoint, MEDIA_B2_BUCKET: 'b2-bucket', MEDIA_B2_KEY_ID: 'k', MEDIA_B2_APP_KEY: 's', MEDIA_B2_REGION: 'us-west-004',
+ MEDIA_INVARIANT_SCAN_HOURS: '0', MEDIA_JOBS_ENABLED: '0',
+ });
+ const db = require('../server/db/database');
+ const reconcile = require('../server/vod/duration-reconcile');
+ const orphans = require('../server/vod/orphans-report');
+ db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+
+ // ── 1. classify ──
+ assert.strictEqual(reconcile.classify(600, 601.5), 'ok');
+ assert.strictEqual(reconcile.classify(600, 640), 'mismatch', '40 s off is reported, not repaired');
+ assert.strictEqual(reconcile.classify(3600, 600), 'wrong');
+ assert.strictEqual(reconcile.classify(20000, 19800), 'mismatch', '200 s off a 5.5 h VOD (1%) is a mismatch, not wrong');
+ assert.strictEqual(reconcile.classify(0, 12), 'missing');
+ assert.strictEqual(reconcile.classify(100, 0), 'unmeasurable');
+ console.log('✅ classify: ok within 2 s; wrong only past 60 s and 2%; missing when nothing is stored');
+
+ const hasFfmpeg = spawnSync('ffmpeg', ['-version']).status === 0;
+ const mk = (file, secs) => {
+ const r = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', `testsrc=duration=${secs}:size=160x120:rate=10`, '-f', 'lavfi', '-i', `sine=frequency=440:duration=${secs}`,
+ '-c:v', 'libx264', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-shortest', '-movflags', '+faststart', file]);
+ assert.strictEqual(r.status, 0, String(r.stderr));
+ return fs.readFileSync(file);
+ };
+ const insVod = (o) => Number(db.run(`INSERT INTO vods (app_id, title, file_path, file_size, duration_seconds, duration_source, storage_provider, storage_key, health_status, is_recording, is_public, visibility)
+ VALUES ('live', ?, ?, ?, ?, ?, ?, ?, ?, 0, 1, 'public')`, [o.title || 't', o.file, o.size || 1000, o.stored, o.source ?? null, o.provider || 'local', o.key || null, o.health || 'ok']).lastInsertRowid);
+ const vod = (id) => db.get('SELECT * FROM vods WHERE id = ?', [id]);
+
+ let ids = {};
+ if (!hasFfmpeg) {
+ console.log('⚠️ ffmpeg not found: the measurement checks are skipped (orphan report still runs)');
+ } else {
+ const V = process.env.VOD_PATH;
+ mk(path.join(V, 'a.mp4'), 6);
+ mk(path.join(V, 'b.mp4'), 6);
+ const cBytes = mk(path.join(tmp, 'c.mp4'), 8);
+ buckets['b2-bucket'].set('vods/c.mp4', cBytes); // offloaded: no local file
+ mk(path.join(V, 'd.mp4'), 6);
+ mk(path.join(V, 'e.mp4'), 6);
+ fs.writeFileSync(path.join(V, 'g.mp4'), Buffer.alloc(30000, 3)); // unreadable
+ ids = {
+ ok: insVod({ file: path.join(V, 'a.mp4'), stored: 6, source: 'probe' }),
+ wallClock: insVod({ file: path.join(V, 'b.mp4'), stored: 3600 }),
+ remote: insVod({ file: '/old/host/path/c.mp4', stored: 99999, provider: 'b2', key: 'vods/c.mp4' }),
+ missing: insVod({ file: path.join(V, 'd.mp4'), stored: 0 }),
+ small: insVod({ file: path.join(V, 'e.mp4'), stored: 9, source: 'probe' }),
+ skipped: insVod({ file: path.join(V, 'f.mp4'), stored: 50, health: 'missing_file' }),
+ unreadable: insVod({ file: path.join(V, 'g.mp4'), stored: 120 }),
+ };
+
+ // ── 2. Dry run: measures local and remote, changes nothing ──
+ const before = db.all('SELECT id, duration_seconds, duration_source FROM vods ORDER BY id');
+ const dry = await reconcile.reconcileBatch({ appId: 'live', afterId: 0, limit: 50 });
+ assert.deepStrictEqual(db.all('SELECT id, duration_seconds, duration_source FROM vods ORDER BY id'), before, 'a dry run writes nothing');
+ const by = Object.fromEntries(dry.rows.map(r => [r.vod_id, r]));
+ assert.strictEqual(by[ids.ok].verdict, 'ok');
+ assert.deepStrictEqual([by[ids.wallClock].verdict, by[ids.wallClock].action, by[ids.wallClock].confirmed], ['wrong', 'would_repair', true]);
+ assert.strictEqual(by[ids.remote].where, 'b2', 'the offloaded copy is measured in B2');
+ assert.deepStrictEqual([by[ids.remote].verdict, by[ids.remote].action], ['wrong', 'would_repair']);
+ assert.ok(Math.abs(by[ids.remote].measured - 8) < 0.6, `remote measured ${by[ids.remote].measured}`);
+ assert.ok(s3log.some(([m, , k]) => m === 'GET' && k === 'vods/c.mp4'), 'read through a presigned GET');
+ assert.ok(!s3log.some(([m]) => m === 'PUT' || m === 'DELETE'));
+ assert.deepStrictEqual([by[ids.missing].verdict, by[ids.missing].action], ['missing', 'would_repair']);
+ assert.deepStrictEqual([by[ids.small].verdict, by[ids.small].action], ['mismatch', 'none'], 'a 3 s difference is reported only');
+ assert.strictEqual(by[ids.skipped].verdict, 'skipped');
+ assert.strictEqual(by[ids.unreadable].verdict, 'unmeasurable');
+ assert.strictEqual(dry.counts.would_repair, 3);
+ assert.strictEqual(dry.counts.remote, 1);
+ console.log('✅ dry run: local files and the B2 copy (presigned ranged read) measured, 3 repairs proposed, nothing written');
+
+ // ── 3. Bounded batches and --apply through the job, then rollback ──
+ const small = await reconcile.reconcileBatch({ appId: 'live', afterId: 0, limit: 2 });
+ assert.deepStrictEqual([small.rows.length, small.range.last_id, small.range.done], [2, ids.wallClock, false]);
+ const queue = require('../server/jobs/queue');
+ const worker = require('../server/jobs/worker');
+ const spec = queue.typeSpec('vod.duration.reconcile');
+ assert.throws(() => spec.validate({ obj: null, params: { limit: 500 } }), /limit/);
+ const params = spec.validate({ obj: null, params: { limit: 4, apply: true } });
+ let j = queue.enqueue({ appId: 'live', type: 'vod.duration.reconcile', params, createdBy: 'test' }).job;
+ j = await worker.runNow(j.id);
+ assert.strictEqual(j.status, 'succeeded', j.error);
+ let res = JSON.parse(j.result);
+ assert.deepStrictEqual([res.mode, res.counts.checked, res.counts.repaired, res.next_after_id], ['apply', 4, 3, ids.missing]);
+ assert.ok(fs.existsSync(path.join(path.dirname(process.env.DB_PATH), 'reports', res.report)), 'the report is written');
+ assert.deepStrictEqual([vod(ids.wallClock).duration_seconds, vod(ids.wallClock).duration_source], [6, 'probe']);
+ assert.deepStrictEqual([vod(ids.remote).duration_seconds, vod(ids.remote).duration_source], [8, 'probe']);
+ assert.deepStrictEqual([vod(ids.missing).duration_seconds, vod(ids.missing).duration_source], [6, 'probe']);
+ assert.strictEqual(db.getSetting('duration_reconcile.cursor.live'), ids.missing, 'the cursor is kept for the next run');
+ const obj = db.get('SELECT metadata FROM media_objects WHERE id = ?', [vod(ids.wallClock).object_id]);
+ assert.strictEqual(JSON.parse(obj.metadata).duration_seconds, 6, 'the object is re-projected');
+ j = queue.enqueue({ appId: 'live', type: 'vod.duration.reconcile', params: spec.validate({ obj: null, params: { limit: 4, apply: true } }), createdBy: 'test2' }).job;
+ res = JSON.parse((await worker.runNow(j.id)).result);
+ assert.deepStrictEqual([res.range.after_id, res.counts.checked, res.range.done, res.next_after_id], [ids.missing, 3, true, 0], 'continues, reaches the end, wraps');
+ const report = JSON.parse(fs.readFileSync(path.join(path.dirname(process.env.DB_PATH), 'reports', JSON.parse(queue.list('live', { type: 'vod.duration.reconcile' }).jobs.slice(-1)[0].result).report), 'utf8'));
+ const rb0 = reconcile.rollback(report.rows);
+ assert.deepStrictEqual(rb0, { restored: 0, would_restore: 3, changed_since: 0 });
+ const rb = reconcile.rollback(report.rows, { apply: true });
+ assert.strictEqual(rb.restored, 3);
+ assert.deepStrictEqual([vod(ids.wallClock).duration_seconds, vod(ids.wallClock).duration_source], [3600, null], 'rolled back');
+ console.log('✅ the job repairs confirmed values in bounded batches, keeps its cursor, writes a report; the report rolls back');
+
+ // ── 4. The script: dry run by default, --apply needs --backup, --rollback ──
+ let r = await runCli('vod-duration-reconcile.js', ['--app', 'live', '--all', '--batch', '3']);
+ assert.strictEqual(r.status, 0, r.stderr);
+ assert.ok(/Would repair 3/.test(r.stdout), r.stdout);
+ assert.strictEqual(vod(ids.wallClock).duration_seconds, 3600, 'dry run changed nothing');
+ r = await runCli('vod-duration-reconcile.js', ['--apply']);
+ assert.strictEqual(r.status, 2, 'apply without a backup is refused');
+ const backup = path.join(tmp, 'bk', 'reconcile.json');
+ r = await runCli('vod-duration-reconcile.js', ['--app', 'live', '--all', '--apply', '--backup', backup]);
+ assert.strictEqual(r.status, 0, r.stderr);
+ assert.ok(fs.existsSync(backup) && fs.existsSync(backup.replace(/\.json$/, '.media.db')), 'rollback file and database copy');
+ assert.strictEqual(vod(ids.wallClock).duration_seconds, 6);
+ r = await runCli('vod-duration-reconcile.js', ['--app', 'live', '--apply', '--backup', backup]);
+ assert.strictEqual(r.status, 2, 'never overwrites a backup');
+ r = await runCli('vod-duration-reconcile.js', ['--rollback', backup]);
+ assert.ok(/Would restore 3/.test(r.stdout), r.stdout);
+ r = await runCli('vod-duration-reconcile.js', ['--rollback', backup, '--apply']);
+ assert.ok(/Restored 3/.test(r.stdout), r.stdout);
+ assert.strictEqual(vod(ids.wallClock).duration_seconds, 3600);
+ console.log('✅ scripts/vod-duration-reconcile.js: dry run by default, --apply with a checked backup, --rollback');
+ }
+
+ // ── 5. vods-orphans/ report: matches and recommendations, read-only ──
+ {
+ const B = buckets['b2-bucket'];
+ const V = process.env.VOD_PATH;
+ const bytes = (n, fill) => Buffer.alloc(n, fill);
+ // canonical present, same size → duplicate
+ const dupId = insVod({ file: '/x/vod-live-501-1726900000001.mp4', stored: 10, provider: 'b2' });
+ B.set('vods/vod-live-501-1726900000001.mp4', bytes(5000, 1));
+ B.set('vods-orphans/vod-live-501-1726900000001.mp4', bytes(5000, 1));
+ // canonical present, other size → review
+ insVod({ file: '/x/vod-live-502-1726900000002.mp4', stored: 10, provider: 'b2' });
+ B.set('vods/vod-live-502-1726900000002.mp4', bytes(4000, 2));
+ B.set('vods-orphans/vod-live-502-1726900000002.mp4', bytes(4500, 2));
+ // no canonical, no local → the only copy
+ const onlyId = insVod({ file: '/x/vod-live-503-1726900000003.mp4', stored: 10, provider: 'b2' });
+ B.set('vods-orphans/vod-live-503-1726900000003.mp4', bytes(3000, 3));
+ // still local with the same size → keep until offloaded
+ fs.writeFileSync(path.join(V, 'vod-live-504-1726900000004.mp4'), bytes(2000, 4));
+ insVod({ file: path.join(V, 'vod-live-504-1726900000004.mp4'), stored: 10 });
+ B.set('vods-orphans/vod-live-504-1726900000004.mp4', bytes(2000, 4));
+ // the row is gone (its object says when)
+ B.set('vods-orphans/vod-live-9999-1726900000005.mp4', bytes(1000, 5));
+ db.run(`INSERT INTO media_objects (id, app_id, namespace, kind, lifecycle_status, legacy_ref, deleted_at) VALUES ('med_01TESTGONE000000000000000A', 'live', 'live', 'vod', 'deleted', 'legacy:live:vod:9999', '2026-09-20 10:00:00')`);
+ // a name that matches nothing
+ B.set('vods-orphans/notes.txt', bytes(10, 6));
+ // a row whose storage_key points INTO the prefix: that object is the canonical copy
+ const keyedId = insVod({ file: '/x/vod-live-506-1726900000006.mp4', stored: 10, provider: 'b2', key: 'vods-orphans/vod-live-506-1726900000006.mp4' });
+ B.set('vods-orphans/vod-live-506-1726900000006.mp4', bytes(700, 7));
+
+ const writesBefore = s3log.filter(([m]) => m !== 'GET' && m !== 'HEAD').length;
+ const rep = await orphans.buildReport({ provider: 'b2' });
+ const rec = Object.fromEntries(rep.objects.map(o => [path.basename(o.key), o.recommendation]));
+ assert.deepStrictEqual(rec, {
+ 'notes.txt': 'review_unknown',
+ 'vod-live-501-1726900000001.mp4': 'delete_duplicate',
+ 'vod-live-502-1726900000002.mp4': 'review_size_differs',
+ 'vod-live-503-1726900000003.mp4': 'keep_only_copy',
+ 'vod-live-504-1726900000004.mp4': 'keep_until_offloaded',
+ 'vod-live-506-1726900000006.mp4': 'keep_only_copy',
+ 'vod-live-9999-1726900000005.mp4': 'delete_row_gone',
+ });
+ const gone = rep.objects.find(o => o.key.includes('9999'));
+ assert.ok(/deleted 2026-09-20/.test(gone.reason), gone.reason);
+ assert.strictEqual(rep.objects.find(o => o.key.includes('-501-')).match.id, dupId);
+ assert.strictEqual(rep.objects.find(o => o.key.includes('-503-')).match.id, onlyId);
+ assert.strictEqual(rep.objects.find(o => o.key.includes('-506-')).match.id, keyedId);
+ assert.deepStrictEqual([rep.totals.objects, rep.by_recommendation.keep_only_copy.objects, rep.by_recommendation.keep_only_copy.bytes], [7, 2, 3700]);
+ assert.strictEqual(rep.read_only, true);
+ assert.strictEqual(s3log.filter(([m]) => m !== 'GET' && m !== 'HEAD').length, writesBefore, 'the report never writes to the bucket');
+ assert.strictEqual(B.size, [...B.keys()].length);
+ const out = path.join(tmp, 'orphans.json');
+ const r = await runCli('vods-orphans-report.js', ['--out', out]);
+ assert.strictEqual(r.status, 0, r.stderr);
+ assert.ok(/7 object\(s\)/.test(r.stdout) && /Nothing was deleted/.test(r.stdout), r.stdout);
+ assert.strictEqual(JSON.parse(fs.readFileSync(out, 'utf8')).totals.objects, 7);
+ assert.ok(B.has('vods-orphans/vod-live-501-1726900000001.mp4'), 'every orphan is still there');
+ console.log('✅ vods-orphans report: each object matched and recommended (only copy, duplicate, row gone, …), nothing deleted');
+ }
+
+ db.close();
+ s3.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('\n✅ All duration reconcile and orphan report tests passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });