openvibers/openvibe.media Commits
object.hash job: sha256 for local copies in bounded batches, recorded on the object (content_hash, with metadata.hash_basis = the file and size it read) and on its local location (checksum, verified_at). Only 14 of 2,839 ready objects had a hash, so scheduled verification's hash check and hash reconciliation were inert. Queued per tenant with work every MEDIA_HASH_INTERVAL_MIN (15; 0 = on demand), light lane, 50 objects and MEDIA_HASH_BUDGET_MB (4096) per run (the first file always fits, up to MEDIA_HASH_MAX_FILE_MB 20480), or one object through the jobs API. It skips a file changed in the last MEDIA_HASH_SETTLE_S (120 s) or while it was read, and a local copy whose size differs from the object's (another version), and writes only if the object still has no hash and the location still points at the file it read. A re-projection that finds the local copy at another path or size (a remux in place, a re-cut, a regenerated thumbnail) now drops a hash this job made and clears the location checksum, so a hash never goes stale and verification never condemns a legitimately rewritten file; a copy that only moved to B2/R2 keeps it. Finished object.hash jobs are pruned like thumbnail jobs. test/content-hash.test.js: recording, skips, count and byte bounds, stale-hash invalidation, the B2 move, verification catching same-size corruption, scheduling and pruning.
Changes
@@@ -348,6 +348,17 @@ For each object it:**It never deletes anything.** No file, remote object or row is removed, and no lifecycle is changed.++**Content hashes.** Most objects were projected without a hash, which left step 1's hash check++inert. The `object.hash` job (`server/jobs/content-hash.js`, light lane, queued per tenant every++`MEDIA_HASH_INTERVAL_MIN`, 15) reads ready objects' present local copies in bounded batches (50++objects, `MEDIA_HASH_BUDGET_MB` 4096 per run; one file up to `MEDIA_HASH_MAX_FILE_MB` 20480) and++records the sha256 on the object (`content_hash`, with `metadata.hash_basis` = the file and size it++read) and on the local location (`checksum`, `verified_at`). It skips a file changed in the last++`MEDIA_HASH_SETTLE_S` (120 s) or while it was read, and a local copy whose size differs from the++object's (another version). A re-projection that finds the local copy at another path or size (a++remux, a re-cut, a regenerated thumbnail) drops a hash this job made, so it never goes stale; a copy++that only moved to B2/R2 keeps it. Test: `test/content-hash.test.js`.++**No good copy.** A ready object has *no good copy* when none of its locations is `present` or`pending`. This includes ready objects with no location row at all. They show up in three places:@@@ -449,7 +460,7 @@ recording runs unless `MEDIA_JOBS_HEAVY_WHILE_RECORDING=1`) and `finalize`(`MEDIA_JOBS_FINALIZE_CONCURRENCY`, 1; runs while recording, as the recorder's own finalize does). A running job holds a lease(`MEDIA_JOBS_LEASE_S`) renewed by a heartbeat; at start, jobs the previous process left `running` arerequeued (or failed when out of attempts). Handlers checkpoint progress and resume from it.--`MEDIA_JOBS_ENABLED=0` stops the worker. Finished thumbnail jobs are pruned after++`MEDIA_JOBS_ENABLED=0` stops the worker. Finished thumbnail and `object.hash` jobs are pruned after`MEDIA_JOBS_RETENTION_DAYS`; other jobs are kept.| type | lane | what it does |@@@ -458,6 +469,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 |++| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) || `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 }` |@@@ -582,6 +594,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_HASH_INTERVAL_MIN` | 15 | queue `object.hash` per tenant this often (`0` = only on demand) |++| `MEDIA_HASH_BUDGET_MB` / `_MAX_FILE_MB` / `_SETTLE_S` | 4096 / 20480 / 120 | bytes read per run; largest file hashed; a file changed this recently waits || `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 |nnew file mode 100644@@@ -0,0 +1,148 @@++/**++ * OpenVibe.Media — job type object.hash (light lane): sha256 of local copies, in bounded batches.++ *++ * Ready objects that have no content_hash and a present local copy get one: the file is read once++ * (streaming), and the hash is recorded on the object (content_hash, plus metadata.hash_basis++ * { key, size, at }: the file it was computed from) and on its local location (checksum,++ * verified_at). Scheduled copy verification then compares local copies against it, and a++ * re-projection that finds the local file at another path or size drops a hash this job made++ * (server/objects/model.js project()), so a remux or re-cut never leaves a stale one.++ *++ * Skipped (and counted): a file changed in the last MEDIA_HASH_SETTLE_S (120 s) or while it was++ * read, a local size that differs from the object's recorded size (the copy is another version),++ * files over MEDIA_HASH_MAX_FILE_MB, and anything past the run's byte budget (next run).++ *++ * object_id set hash that object only++ * params { limit (1-500, default 50), budget_mb (default MEDIA_HASH_BUDGET_MB, 4096) }++ * result { hashed, bytes, skipped: { reason: n }, remaining }++ * Scheduled per tenant every MEDIA_HASH_INTERVAL_MIN (15; 0 = only on demand).++ */++'use strict';++++const fs = require('fs');++const crypto = require('crypto');++const db = require('../db/database');++const queue = require('./queue');++++const { JobError } = queue;++const TYPE = 'object.hash';++const MB = 1024 * 1024;++++function envInt(name, fallback) {++ const v = parseInt(process.env[name] || '', 10);++ return Number.isFinite(v) ? v : fallback;++}++const settings = () => ({++ budgetMb: Math.max(1, envInt('MEDIA_HASH_BUDGET_MB', 4096)),++ maxFileMb: Math.max(1, envInt('MEDIA_HASH_MAX_FILE_MB', 20480)),++ settleS: Math.max(0, envInt('MEDIA_HASH_SETTLE_S', 120)),++ intervalMin: envInt('MEDIA_HASH_INTERVAL_MIN', 15),++});++++function sha256File(p, signal) {++ return new Promise((resolve, reject) => {++ const h = crypto.createHash('sha256');++ const s = fs.createReadStream(p, { highWaterMark: MB });++ const onAbort = () => s.destroy(new Error('aborted'));++ if (signal) { if (signal.aborted) return reject(new Error('aborted')); signal.addEventListener('abort', onAbort, { once: true }); }++ s.on('data', d => h.update(d));++ s.on('end', () => { if (signal) signal.removeEventListener('abort', onAbort); resolve(h.digest('hex')); });++ s.on('error', (err) => { if (signal) signal.removeEventListener('abort', onAbort); reject(err); });++ });++}++++const CANDIDATES = `SELECT o.id, o.size_bytes, l.id AS loc_id, l.key AS loc_key++ FROM media_objects o JOIN media_locations l ON l.object_id = o.id AND l.provider = 'local' AND l.state = 'present'++ WHERE o.lifecycle_status = 'ready' AND o.content_hash IS NULL`;++++function remaining(appId) {++ return db.get(`SELECT COUNT(*) AS n FROM (${CANDIDATES} AND o.app_id = ?)`, [appId]).n;++}++++function validate({ obj, params }) {++ const p = params || {};++ const limit = p.limit == null ? 50 : Number(p.limit);++ if (!Number.isInteger(limit) || limit < 1 || limit > 500) throw new JobError('media.job.invalid', 'params.limit must be 1-500', { permanent: true });++ const out = { limit };++ if (p.budget_mb != null) {++ const b = Number(p.budget_mb);++ if (!Number.isInteger(b) || b < 1 || b > 1024 * 1024) throw new JobError('media.job.invalid', 'params.budget_mb must be a positive integer', { permanent: true });++ out.budget_mb = b;++ }++ if (obj && obj.content_hash) throw new JobError('media.job.invalid', 'the object already has a content hash', { permanent: true });++ return out;++}++++/** Record one hash, only if nothing changed since the file was read. Returns true when written. */++function record(c, hash, st) {++ let wrote = false;++ db.getDb().transaction(() => {++ const o = db.get('SELECT content_hash, metadata, lifecycle_status FROM media_objects WHERE id = ?', [c.id]);++ const l = db.get('SELECT key, state FROM media_locations WHERE id = ?', [c.loc_id]);++ if (!o || o.content_hash || o.lifecycle_status !== 'ready' || !l || l.key !== c.loc_key || l.state !== 'present') return;++ let md = {};++ try { md = JSON.parse(o.metadata || '{}') || {}; } catch { md = {}; }++ md.hash_basis = { key: c.loc_key, size: st.size, at: new Date().toISOString() };++ db.run('UPDATE media_objects SET content_hash = ?, metadata = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND content_hash IS NULL', [hash, JSON.stringify(md), c.id]);++ db.run(`UPDATE media_locations SET checksum = ?, size_bytes = ?, verified_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP++ WHERE id = ? AND key = ?`, [hash, st.size, c.loc_id, c.loc_key]);++ wrote = true;++ })();++ return wrote;++}++++/** One bounded batch for a tenant (or one object). Used by the job and by tests. */++async function hashBatch({ appId, objectId = null, limit = 50, budgetMb = settings().budgetMb, signal = null, now = Date.now() } = {}) {++ const s = settings();++ const budget = budgetMb * MB;++ const rows = objectId++ ? db.all(`${CANDIDATES} AND o.id = ? AND o.app_id = ?`, [objectId, appId])++ : db.all(`${CANDIDATES} AND o.app_id = ? ORDER BY o.id LIMIT ?`, [appId, limit]);++ const out = { hashed: 0, bytes: 0, skipped: {}, remaining: 0 };++ const skip = (why) => { out.skipped[why] = (out.skipped[why] || 0) + 1; };++ for (const c of rows) {++ if (signal && signal.aborted) break;++ let st;++ try { st = fs.statSync(c.loc_key); } catch { skip('file_missing'); continue; }++ if (!st.isFile()) { skip('not_a_file'); continue; }++ if (now - st.mtimeMs < s.settleS * 1000) { skip('recently_changed'); continue; }++ if (Number(c.size_bytes) > 0 && st.size !== Number(c.size_bytes)) { skip('size_differs_from_object'); continue; }++ if (st.size > s.maxFileMb * MB) { skip('over_max_file_size'); continue; }++ if (out.bytes > 0 && out.bytes + st.size > budget) { skip('budget'); continue; }++ let hash;++ try { hash = await sha256File(c.loc_key, signal); } catch (err) { if (signal && signal.aborted) break; skip('read_error'); continue; }++ let after;++ try { after = fs.statSync(c.loc_key); } catch { skip('file_missing'); continue; }++ if (after.size !== st.size || after.mtimeMs !== st.mtimeMs) { skip('changed_while_reading'); continue; }++ out.bytes += st.size;++ if (record(c, hash, st)) out.hashed++; else skip('changed_in_database');++ }++ out.remaining = remaining(appId);++ return out;++}++++async function run(job, ctx = {}) {++ const p = validate({ obj: null, params: job.params });++ return hashBatch({ appId: job.app_id, objectId: job.object_id || null, limit: p.limit, budgetMb: p.budget_mb || settings().budgetMb, signal: ctx.signal });++}++++/** Queue one batch per tenant that has unhashed local copies, every MEDIA_HASH_INTERVAL_MIN. */++function schedule(nowMs = Date.now()) {++ const minutes = settings().intervalMin;++ if (!(minutes > 0)) return 0;++ const period = Math.floor(nowMs / (minutes * 60 * 1000));++ let n = 0;++ for (const { app_id: appId } of db.all(`SELECT DISTINCT o.app_id FROM (${CANDIDATES}) c JOIN media_objects o ON o.id = c.id`)) {++ try {++ const r = queue.enqueue({ appId, type: TYPE, params: { limit: 50 }, idempotencyKey: `${TYPE}:${minutes}m:${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: 'light', maxAttempts: 2, timeoutMs: 30 * 60 * 1000, needsObject: false, validate, run },++ hashBatch, schedule, sha256File, remaining,++};@@@ -387,8 +387,8 @@ function counts() {return out;}--/** Delete finished jobs of high-volume types older than `days` (thumbnail requests). Others are kept. */--function prune({ days = 30, types = ['thumbnail.regenerate'] } = {}) {++/** Delete finished jobs of high-volume types older than `days` (thumbnail requests, scheduled hashing). Others are kept. */++function prune({ days = 30, types = ['thumbnail.regenerate', 'object.hash'] } = {}) {return db.run(`DELETE FROM media_jobs WHERE status IN ('succeeded', 'failed', 'cancelled') AND job_type IN (${types.map(() => '?').join(', ')})AND finished_at < datetime('now', ?)`, [...types, `-${Math.max(1, days)} days`]).changes;}@@@ -7,6 +7,7 @@* 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++ * object.hash light sha256 of local copies with no content hash yet, one bounded batch (or one object)*/'use strict';@@@ -18,5 +19,6 @@ 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);++queue.register('object.hash', require('./content-hash').spec);module.exports = { names: () => queue.typeNames() };@@@ -148,11 +148,12 @@ async function tick() {lastScheduleAt = now;scheduleInvariantScans(now);try { require('./duration-reconcile').schedule(now); } catch (err) { console.warn('[Jobs] duration reconcile schedule:', err.message); }++ try { require('./content-hash').schedule(now); } catch (err) { console.warn('[Jobs] content hash 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;-- try { const n = queue.prune({ days: cfg().retentionDays }); if (n) console.log(`[Jobs] pruned ${n} finished thumbnail job(s)`); } catch { /* next time */ }++ try { const n = queue.prune({ days: cfg().retentionDays }); if (n) console.log(`[Jobs] pruned ${n} finished thumbnail/hash job(s)`); } catch { /* next time */ }}for (const lane of lanes()) {if (lane.name === 'heavy' && !cfg().heavyWhileRecording && recordingActive()) continue;@@@ -158,7 +158,7 @@ function upsertLocation(objectId, loc) {updated_at = CURRENT_TIMESTAMP WHERE id = ?`,[loc.bucket ?? existing.bucket, loc.key, loc.storage_class || existing.storage_class,keep ? existing.state : (loc.state || 'pending'),-- loc.checksum ?? (existing.key === loc.key ? existing.checksum : null),++ loc.clearChecksum ? null : (loc.checksum ?? (existing.key === loc.key ? existing.checksum : null)),keep ? existing.size_bytes : (loc.size_bytes ?? null),keep ? existing.verified_at : (verifiedAt || (existing.key === loc.key && loc.state === existing.state ? existing.verified_at : null)),existing.id]);@@@ -205,11 +205,24 @@ function project(p) {legacy_ref: p.legacy_ref, metadata: p.metadata || {},};let id, created = false;++ let staleHash = false;if (found) {id = found.id;// Knowledge the row doesn't carry survives a re-projection.if (!fields.content_hash) fields.content_hash = found.content_hash;fields.metadata = { ...parseJson(found.metadata, {}), ...fields.metadata };++ // A hash the content-hash job computed (hash_basis: the local file it read) holds only while++ // that file is unchanged: a present local copy at another path or of another size (a remux,++ // a re-cut, a regenerated thumbnail) makes it stale. A copy that moved to B2/R2 keeps it.++ const basis = fields.metadata.hash_basis;++ if (basis && fields.content_hash === found.content_hash && !p.content_hash) {++ const local = p.locations.find(l => l.provider === 'local' && l.state === 'present');++ if (local && (local.key !== basis.key || Number(local.size_bytes) !== Number(basis.size))) {++ fields.content_hash = null;++ delete fields.metadata.hash_basis;++ staleHash = true;++ }++ }if (found.lifecycle_status === 'deleted') fields.deleted_at = null; // the row exists, so the object does// owner_subject (filled by the owner-subject job) is the subject of owner_user_id; a different owner// drops it until the job resolves the new one.@@@ -222,7 +235,7 @@ function project(p) {created = true;}const keep = new Set(p.locations.map(l => l.provider));-- for (const loc of p.locations) upsertLocation(id, loc);++ for (const loc of p.locations) upsertLocation(id, staleHash && loc.provider === 'local' ? { ...loc, clearChecksum: true } : loc);for (const l of listLocations(id)) if (!keep.has(l.provider)) db.run('DELETE FROM media_locations WHERE id = ?', [l.id]);return { id, created };}nnew file mode 100644@@@ -0,0 +1,127 @@++'use strict';++// object.hash (server/jobs/content-hash.js): ready objects with a present local copy and no hash get++// sha256 on the object (content_hash + metadata.hash_basis) and on the local location (checksum), in++// bounded batches (count and bytes); files still changing or of another size than the object are++// skipped; a re-projection that finds the local file changed drops the job's hash (a remux never++// leaves a stale one) while a move to B2 keeps it; scheduled copy verification then checks against it.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const crypto = require('crypto');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-hash-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++Object.assign(process.env, {++ DB_PATH: path.join(tmp, '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_INVARIANT_SCAN_HOURS: '0', MEDIA_HASH_SETTLE_S: '60', MEDIA_HASH_INTERVAL_MIN: '15',++});++++const db = require('../server/db/database');++const model = require('../server/objects/model');++const queue = require('../server/jobs/queue');++const worker = require('../server/jobs/worker');++const hashJob = require('../server/jobs/content-hash');++const verify = require('../server/objects/verify-job');++++db.upsertApp({ app_id: 'live', api_key: 'live-key' });++db.upsertApp({ app_id: 'games', api_key: 'games-key' });++++const sha = (buf) => crypto.createHash('sha256').update(buf).digest('hex');++const age = (file, s = 3600) => { const t = (Date.now() - s * 1000) / 1000; fs.utimesSync(file, t, t); };++function vodWith(name, bytes, { app = 'live', old = true } = {}) {++ const file = path.join(process.env.VOD_PATH, name);++ fs.writeFileSync(file, bytes);++ if (old) age(file);++ const id = Number(db.run(`INSERT INTO vods (app_id, title, file_path, file_size, duration_seconds, is_public, visibility, health_status) VALUES (?, 't', ?, ?, 60, 1, 'public', 'ok')`,++ [app, file, bytes.length]).lastInsertRowid);++ model.sync('vod', id);++ return { id, file, obj: () => model.getObject(db.get('SELECT object_id FROM vods WHERE id = ?', [id]).object_id) };++}++const localLoc = (objId) => db.get("SELECT * FROM media_locations WHERE object_id = ? AND provider = 'local'", [objId]);++++(async () => {++ const a = vodWith('a.mp4', Buffer.alloc(300000, 1));++ const b = vodWith('b.mp4', Buffer.alloc(200000, 2));++ const fresh = vodWith('fresh.mp4', Buffer.alloc(1000, 3), { old: false });++ const other = vodWith('other.mp4', Buffer.alloc(5000, 4));++ db.run('UPDATE media_objects SET size_bytes = 999 WHERE id = ?', [other.obj().id]); // the local copy is another version++ const games = vodWith('g.mp4', Buffer.alloc(4000, 5), { app: 'games' });++ assert.strictEqual(a.obj().content_hash, null);++ const before = hashJob.remaining('live');++ assert.strictEqual(before, 4);++++ // ── 1. A batch hashes local copies onto the object and the location, and skips what it must ──++ const r = await hashJob.hashBatch({ appId: 'live' });++ assert.deepStrictEqual([r.hashed, r.bytes], [2, 500000]);++ assert.deepStrictEqual(r.skipped, { recently_changed: 1, size_differs_from_object: 1 });++ assert.strictEqual(r.remaining, 2);++ const oa = a.obj();++ assert.strictEqual(oa.content_hash, sha(fs.readFileSync(a.file)));++ assert.strictEqual(localLoc(oa.id).checksum, oa.content_hash, 'the location carries it too');++ assert.ok(localLoc(oa.id).verified_at);++ assert.deepStrictEqual([JSON.parse(oa.metadata).hash_basis.key, JSON.parse(oa.metadata).hash_basis.size], [path.resolve(a.file), 300000]);++ assert.strictEqual(games.obj().content_hash, null, 'one tenant per run');++ console.log('✅ sha256 of local copies recorded on object and location; changing files and other versions skipped');++++ // ── 2. Bounded by bytes: a run stops at its budget (the first file always fits), the next continues ──++ const big1 = vodWith('big1.mp4', Buffer.alloc(3 * 1048576, 6));++ const big2 = vodWith('big2.mp4', Buffer.alloc(3 * 1048576, 7));++ const r2 = await hashJob.hashBatch({ appId: 'live', budgetMb: 4 });++ assert.deepStrictEqual([r2.hashed, r2.skipped.budget], [1, 1]);++ assert.strictEqual([big1, big2].filter(v => v.obj().content_hash).length, 1, 'one of the two 3 MB files fits a 4 MB budget');++ const r3 = await hashJob.hashBatch({ appId: 'live', budgetMb: 4 });++ assert.strictEqual(r3.hashed, 1);++ assert.ok(big1.obj().content_hash && big2.obj().content_hash, 'the next run takes the other');++ const r4 = await hashJob.hashBatch({ appId: 'live', limit: 1 });++ assert.strictEqual(r4.hashed + Object.values(r4.skipped).reduce((x, y) => x + y, 0), 1, 'limit bounds the count');++ console.log('✅ bounded batches: object count and byte budget');++++ // ── 3. A changed local file drops the job's hash on re-projection; a move to B2 keeps it ──++ fs.appendFileSync(b.file, Buffer.alloc(1000, 9)); // a remux in place: new size++ db.run('UPDATE vods SET file_size = ? WHERE id = ?', [201000, b.id]);++ model.sync('vod', b.id);++ const ob = b.obj();++ assert.strictEqual(ob.content_hash, null, 'stale hash dropped');++ assert.strictEqual(localLoc(ob.id).checksum, null);++ assert.strictEqual(JSON.parse(ob.metadata).hash_basis, undefined);++ age(b.file);++ await hashJob.hashBatch({ appId: 'live', objectId: ob.id });++ assert.strictEqual(b.obj().content_hash, sha(fs.readFileSync(b.file)), 'hashed again from the new bytes');++ const kept = oa.content_hash;++ fs.unlinkSync(a.file); // offloaded: local copy gone, B2 canonical++ db.run("UPDATE vods SET storage_provider = 'b2', storage_key = 'vods/a.mp4' WHERE id = ?", [a.id]);++ model.sync('vod', a.id);++ assert.strictEqual(a.obj().content_hash, kept, 'the bytes moved; the hash stays');++ console.log('✅ a remuxed local file loses the stale hash (and is re-hashed); a move to B2 keeps it');++++ // ── 4. Verification uses it: a corrupted local copy of the same size is caught ──++ const c = vodWith('c.mp4', Buffer.alloc(100000, 8));++ await hashJob.hashBatch({ appId: 'live', objectId: c.obj().id });++ const good = await verify.verifyObject(c.obj(), { head: async () => undefined, upload: async () => undefined, hashMaxBytes: 64 * 1048576, repairCorrupt: false, budget: { left: 0 } });++ assert.strictEqual(good.locations.find(l => l.provider === 'local').state, 'present');++ const fd = fs.openSync(c.file, 'r+'); fs.writeSync(fd, Buffer.from([0]), 0, 1, 500); fs.closeSync(fd);++ const bad = await verify.verifyObject(c.obj(), { head: async () => undefined, upload: async () => undefined, hashMaxBytes: 64 * 1048576, repairCorrupt: false, budget: { left: 0 } });++ assert.strictEqual(bad.locations.find(l => l.provider === 'local').state, 'corrupt', 'bit rot of the same size is now visible');++ console.log('✅ scheduled verification checks local copies against the recorded hash');++++ // ── 5. As a job: scheduled per tenant with work, run through the worker, pruned like thumbnails ──++ assert.throws(() => hashJob.spec.validate({ obj: null, params: { limit: 0 } }), /limit/);++ const n = hashJob.schedule(Date.now());++ assert.strictEqual(n, 2, 'one per tenant with unhashed local copies (live, games)');++ assert.strictEqual(hashJob.schedule(Date.now()), 0, 'once per period');++ const gj = queue.list('games', { type: 'object.hash' }).jobs[0];++ const done = await worker.runNow(gj.id);++ assert.strictEqual(done.status, 'succeeded');++ assert.strictEqual(JSON.parse(done.result).hashed, 1);++ assert.ok(games.obj().content_hash);++ db.run("UPDATE media_jobs SET finished_at = datetime('now', '-40 days') WHERE id = ?", [gj.id]);++ assert.strictEqual(queue.prune({ days: 30 }), 1, 'finished hash jobs are pruned');++ console.log('✅ object.hash runs as a scheduled job per tenant and is pruned when finished');++++ db.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('\n✅ All content-hash tests passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -348,6 +348,17 @@ For each object it:
@@ -348,6 +348,17 @@ For each object it:
**It never deletes anything.** No file, remote object or row is removed, and no lifecycle is changed.
**It never deletes anything.** No file, remote object or row is removed, and no lifecycle is changed.
+**Content hashes.** Most objects were projected without a hash, which left step 1's hash check
+inert. The `object.hash` job (`server/jobs/content-hash.js`, light lane, queued per tenant every
+`MEDIA_HASH_INTERVAL_MIN`, 15) reads ready objects' present local copies in bounded batches (50
+objects, `MEDIA_HASH_BUDGET_MB` 4096 per run; one file up to `MEDIA_HASH_MAX_FILE_MB` 20480) and
+records the sha256 on the object (`content_hash`, with `metadata.hash_basis` = the file and size it
+read) and on the local location (`checksum`, `verified_at`). It skips a file changed in the last
+`MEDIA_HASH_SETTLE_S` (120 s) or while it was read, and a local copy whose size differs from the
+object's (another version). A re-projection that finds the local copy at another path or size (a
+remux, a re-cut, a regenerated thumbnail) drops a hash this job made, so it never goes stale; a copy
+that only moved to B2/R2 keeps it. Test: `test/content-hash.test.js`.
+
**No good copy.** A ready object has *no good copy* when none of its locations is `present` or
**No good copy.** A ready object has *no good copy* when none of its locations is `present` or
`pending`. This includes ready objects with no location row at all. They show up in three places:
`pending`. This includes ready objects with no location row at all. They show up in three places:
@@ -449,7 +460,7 @@ recording runs unless `MEDIA_JOBS_HEAVY_WHILE_RECORDING=1`) and `finalize`
@@ -449,7 +460,7 @@ recording runs unless `MEDIA_JOBS_HEAVY_WHILE_RECORDING=1`) and `finalize`
(`MEDIA_JOBS_FINALIZE_CONCURRENCY`, 1; runs while recording, as the recorder's own finalize does). A running job holds a lease
(`MEDIA_JOBS_FINALIZE_CONCURRENCY`, 1; runs while recording, as the recorder's own finalize does). A running job holds a lease
(`MEDIA_JOBS_LEASE_S`) renewed by a heartbeat; at start, jobs the previous process left `running` are
(`MEDIA_JOBS_LEASE_S`) renewed by a heartbeat; at start, jobs the previous process left `running` are
requeued (or failed when out of attempts). Handlers checkpoint progress and resume from it.
requeued (or failed when out of attempts). Handlers checkpoint progress and resume from it.
-`MEDIA_JOBS_ENABLED=0` stops the worker. Finished thumbnail jobs are pruned after
+`MEDIA_JOBS_ENABLED=0` stops the worker. Finished thumbnail and `object.hash` jobs are pruned after
`MEDIA_JOBS_RETENTION_DAYS`; other jobs are kept.
`MEDIA_JOBS_RETENTION_DAYS`; other jobs are kept.
| type | lane | what it does |
| type | lane | what it does |
@@ -458,6 +469,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu
@@ -458,6 +469,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 |
+| `object.hash` | light | sha256 of ready objects' present local copies that have no hash yet, onto the object (`content_hash`, `metadata.hash_basis`) and the local location (`checksum`). Tenant-wide batch (`params { limit (1-500, 50), budget_mb }`) or one object. Scheduled per tenant every `MEDIA_HASH_INTERVAL_MIN`. Result `{ hashed, bytes, skipped, remaining }`. See [Scheduled verification](#scheduled-verification) |
| `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.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 }` |
@@ -582,6 +594,8 @@ Not run against production yet: it needs the owner's go-ahead.
@@ -582,6 +594,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_HASH_INTERVAL_MIN` | 15 | queue `object.hash` per tenant this often (`0` = only on demand) |
+| `MEDIA_HASH_BUDGET_MB` / `_MAX_FILE_MB` / `_SETTLE_S` | 4096 / 20480 / 120 | bytes read per run; largest file hashed; a file changed this recently waits |
| `MEDIA_DURATION_RECONCILE_HOURS` | 0 | schedule `vod.duration.reconcile` per tenant this often (`0` = only on demand) |
| `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_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 |
new file mode 100644
new file mode 100644
@@ -0,0 +1,148 @@
@@ -0,0 +1,148 @@
+/**
+ * OpenVibe.Media — job type object.hash (light lane): sha256 of local copies, in bounded batches.
+ *
+ * Ready objects that have no content_hash and a present local copy get one: the file is read once
+ * (streaming), and the hash is recorded on the object (content_hash, plus metadata.hash_basis
+ * { key, size, at }: the file it was computed from) and on its local location (checksum,
+ * verified_at). Scheduled copy verification then compares local copies against it, and a
+ * re-projection that finds the local file at another path or size drops a hash this job made
+ * (server/objects/model.js project()), so a remux or re-cut never leaves a stale one.
+ *
+ * Skipped (and counted): a file changed in the last MEDIA_HASH_SETTLE_S (120 s) or while it was
+ * read, a local size that differs from the object's recorded size (the copy is another version),
+ * files over MEDIA_HASH_MAX_FILE_MB, and anything past the run's byte budget (next run).
+ *
+ * object_id set hash that object only
+ * params { limit (1-500, default 50), budget_mb (default MEDIA_HASH_BUDGET_MB, 4096) }
+ * result { hashed, bytes, skipped: { reason: n }, remaining }
+ * Scheduled per tenant every MEDIA_HASH_INTERVAL_MIN (15; 0 = only on demand).
+ */
+'use strict';
+
+const fs = require('fs');
+const crypto = require('crypto');
+const db = require('../db/database');
+const queue = require('./queue');
+
+const { JobError } = queue;
+const TYPE = 'object.hash';
+const MB = 1024 * 1024;
+
+function envInt(name, fallback) {
+ const v = parseInt(process.env[name] || '', 10);
+ return Number.isFinite(v) ? v : fallback;
+}
+const settings = () => ({
+ budgetMb: Math.max(1, envInt('MEDIA_HASH_BUDGET_MB', 4096)),
+ maxFileMb: Math.max(1, envInt('MEDIA_HASH_MAX_FILE_MB', 20480)),
+ settleS: Math.max(0, envInt('MEDIA_HASH_SETTLE_S', 120)),
+ intervalMin: envInt('MEDIA_HASH_INTERVAL_MIN', 15),
+});
+
+function sha256File(p, signal) {
+ return new Promise((resolve, reject) => {
+ const h = crypto.createHash('sha256');
+ const s = fs.createReadStream(p, { highWaterMark: MB });
+ const onAbort = () => s.destroy(new Error('aborted'));
+ if (signal) { if (signal.aborted) return reject(new Error('aborted')); signal.addEventListener('abort', onAbort, { once: true }); }
+ s.on('data', d => h.update(d));
+ s.on('end', () => { if (signal) signal.removeEventListener('abort', onAbort); resolve(h.digest('hex')); });
+ s.on('error', (err) => { if (signal) signal.removeEventListener('abort', onAbort); reject(err); });
+ });
+}
+
+const CANDIDATES = `SELECT o.id, o.size_bytes, l.id AS loc_id, l.key AS loc_key
+ FROM media_objects o JOIN media_locations l ON l.object_id = o.id AND l.provider = 'local' AND l.state = 'present'
+ WHERE o.lifecycle_status = 'ready' AND o.content_hash IS NULL`;
+
+function remaining(appId) {
+ return db.get(`SELECT COUNT(*) AS n FROM (${CANDIDATES} AND o.app_id = ?)`, [appId]).n;
+}
+
+function validate({ obj, params }) {
+ const p = params || {};
+ const limit = p.limit == null ? 50 : Number(p.limit);
+ if (!Number.isInteger(limit) || limit < 1 || limit > 500) throw new JobError('media.job.invalid', 'params.limit must be 1-500', { permanent: true });
+ const out = { limit };
+ if (p.budget_mb != null) {
+ const b = Number(p.budget_mb);
+ if (!Number.isInteger(b) || b < 1 || b > 1024 * 1024) throw new JobError('media.job.invalid', 'params.budget_mb must be a positive integer', { permanent: true });
+ out.budget_mb = b;
+ }
+ if (obj && obj.content_hash) throw new JobError('media.job.invalid', 'the object already has a content hash', { permanent: true });
+ return out;
+}
+
+/** Record one hash, only if nothing changed since the file was read. Returns true when written. */
+function record(c, hash, st) {
+ let wrote = false;
+ db.getDb().transaction(() => {
+ const o = db.get('SELECT content_hash, metadata, lifecycle_status FROM media_objects WHERE id = ?', [c.id]);
+ const l = db.get('SELECT key, state FROM media_locations WHERE id = ?', [c.loc_id]);
+ if (!o || o.content_hash || o.lifecycle_status !== 'ready' || !l || l.key !== c.loc_key || l.state !== 'present') return;
+ let md = {};
+ try { md = JSON.parse(o.metadata || '{}') || {}; } catch { md = {}; }
+ md.hash_basis = { key: c.loc_key, size: st.size, at: new Date().toISOString() };
+ db.run('UPDATE media_objects SET content_hash = ?, metadata = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND content_hash IS NULL', [hash, JSON.stringify(md), c.id]);
+ db.run(`UPDATE media_locations SET checksum = ?, size_bytes = ?, verified_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
+ WHERE id = ? AND key = ?`, [hash, st.size, c.loc_id, c.loc_key]);
+ wrote = true;
+ })();
+ return wrote;
+}
+
+/** One bounded batch for a tenant (or one object). Used by the job and by tests. */
+async function hashBatch({ appId, objectId = null, limit = 50, budgetMb = settings().budgetMb, signal = null, now = Date.now() } = {}) {
+ const s = settings();
+ const budget = budgetMb * MB;
+ const rows = objectId
+ ? db.all(`${CANDIDATES} AND o.id = ? AND o.app_id = ?`, [objectId, appId])
+ : db.all(`${CANDIDATES} AND o.app_id = ? ORDER BY o.id LIMIT ?`, [appId, limit]);
+ const out = { hashed: 0, bytes: 0, skipped: {}, remaining: 0 };
+ const skip = (why) => { out.skipped[why] = (out.skipped[why] || 0) + 1; };
+ for (const c of rows) {
+ if (signal && signal.aborted) break;
+ let st;
+ try { st = fs.statSync(c.loc_key); } catch { skip('file_missing'); continue; }
+ if (!st.isFile()) { skip('not_a_file'); continue; }
+ if (now - st.mtimeMs < s.settleS * 1000) { skip('recently_changed'); continue; }
+ if (Number(c.size_bytes) > 0 && st.size !== Number(c.size_bytes)) { skip('size_differs_from_object'); continue; }
+ if (st.size > s.maxFileMb * MB) { skip('over_max_file_size'); continue; }
+ if (out.bytes > 0 && out.bytes + st.size > budget) { skip('budget'); continue; }
+ let hash;
+ try { hash = await sha256File(c.loc_key, signal); } catch (err) { if (signal && signal.aborted) break; skip('read_error'); continue; }
+ let after;
+ try { after = fs.statSync(c.loc_key); } catch { skip('file_missing'); continue; }
+ if (after.size !== st.size || after.mtimeMs !== st.mtimeMs) { skip('changed_while_reading'); continue; }
+ out.bytes += st.size;
+ if (record(c, hash, st)) out.hashed++; else skip('changed_in_database');
+ }
+ out.remaining = remaining(appId);
+ return out;
+}
+
+async function run(job, ctx = {}) {
+ const p = validate({ obj: null, params: job.params });
+ return hashBatch({ appId: job.app_id, objectId: job.object_id || null, limit: p.limit, budgetMb: p.budget_mb || settings().budgetMb, signal: ctx.signal });
+}
+
+/** Queue one batch per tenant that has unhashed local copies, every MEDIA_HASH_INTERVAL_MIN. */
+function schedule(nowMs = Date.now()) {
+ const minutes = settings().intervalMin;
+ if (!(minutes > 0)) return 0;
+ const period = Math.floor(nowMs / (minutes * 60 * 1000));
+ let n = 0;
+ for (const { app_id: appId } of db.all(`SELECT DISTINCT o.app_id FROM (${CANDIDATES}) c JOIN media_objects o ON o.id = c.id`)) {
+ try {
+ const r = queue.enqueue({ appId, type: TYPE, params: { limit: 50 }, idempotencyKey: `${TYPE}:${minutes}m:${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: 'light', maxAttempts: 2, timeoutMs: 30 * 60 * 1000, needsObject: false, validate, run },
+ hashBatch, schedule, sha256File, remaining,
+};
@@ -387,8 +387,8 @@ function counts() {
@@ -387,8 +387,8 @@ function counts() {
return out;
return out;
}
}
-/** Delete finished jobs of high-volume types older than `days` (thumbnail requests). Others are kept. */
-function prune({ days = 30, types = ['thumbnail.regenerate'] } = {}) {
+/** Delete finished jobs of high-volume types older than `days` (thumbnail requests, scheduled hashing). Others are kept. */
+function prune({ days = 30, types = ['thumbnail.regenerate', 'object.hash'] } = {}) {
return db.run(`DELETE FROM media_jobs WHERE status IN ('succeeded', 'failed', 'cancelled') AND job_type IN (${types.map(() => '?').join(', ')})
return db.run(`DELETE FROM media_jobs WHERE status IN ('succeeded', 'failed', 'cancelled') AND job_type IN (${types.map(() => '?').join(', ')})
AND finished_at < datetime('now', ?)`, [...types, `-${Math.max(1, days)} days`]).changes;
AND finished_at < datetime('now', ?)`, [...types, `-${Math.max(1, days)} days`]).changes;
}
}
@@ -7,6 +7,7 @@
@@ -7,6 +7,7 @@
* 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
* vod.duration.reconcile heavy stored VOD durations vs a measurement of the real file (local or B2/R2), one batch
+ * object.hash light sha256 of local copies with no content hash yet, one bounded batch (or one object)
*/
*/
'use strict';
'use strict';
@@ -18,5 +19,6 @@ queue.register('object.split', require('./derive').split);
@@ -18,5 +19,6 @@ 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);
queue.register('vod.duration.reconcile', require('./duration-reconcile').spec);
+queue.register('object.hash', require('./content-hash').spec);
module.exports = { names: () => queue.typeNames() };
module.exports = { names: () => queue.typeNames() };
@@ -148,11 +148,12 @@ async function tick() {
@@ -148,11 +148,12 @@ async function tick() {
lastScheduleAt = now;
lastScheduleAt = now;
scheduleInvariantScans(now);
scheduleInvariantScans(now);
try { require('./duration-reconcile').schedule(now); } catch (err) { console.warn('[Jobs] duration reconcile schedule:', err.message); }
try { require('./duration-reconcile').schedule(now); } catch (err) { console.warn('[Jobs] duration reconcile schedule:', err.message); }
+ try { require('./content-hash').schedule(now); } catch (err) { console.warn('[Jobs] content hash 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;
- try { const n = queue.prune({ days: cfg().retentionDays }); if (n) console.log(`[Jobs] pruned ${n} finished thumbnail job(s)`); } catch { /* next time */ }
+ try { const n = queue.prune({ days: cfg().retentionDays }); if (n) console.log(`[Jobs] pruned ${n} finished thumbnail/hash job(s)`); } catch { /* next time */ }
}
}
for (const lane of lanes()) {
for (const lane of lanes()) {
if (lane.name === 'heavy' && !cfg().heavyWhileRecording && recordingActive()) continue;
if (lane.name === 'heavy' && !cfg().heavyWhileRecording && recordingActive()) continue;
@@ -158,7 +158,7 @@ function upsertLocation(objectId, loc) {
@@ -158,7 +158,7 @@ function upsertLocation(objectId, loc) {
updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
[loc.bucket ?? existing.bucket, loc.key, loc.storage_class || existing.storage_class,
[loc.bucket ?? existing.bucket, loc.key, loc.storage_class || existing.storage_class,
keep ? existing.state : (loc.state || 'pending'),
keep ? existing.state : (loc.state || 'pending'),
- loc.checksum ?? (existing.key === loc.key ? existing.checksum : null),
+ loc.clearChecksum ? null : (loc.checksum ?? (existing.key === loc.key ? existing.checksum : null)),
keep ? existing.size_bytes : (loc.size_bytes ?? null),
keep ? existing.size_bytes : (loc.size_bytes ?? null),
keep ? existing.verified_at : (verifiedAt || (existing.key === loc.key && loc.state === existing.state ? existing.verified_at : null)),
keep ? existing.verified_at : (verifiedAt || (existing.key === loc.key && loc.state === existing.state ? existing.verified_at : null)),
existing.id]);
existing.id]);
@@ -205,11 +205,24 @@ function project(p) {
@@ -205,11 +205,24 @@ function project(p) {
legacy_ref: p.legacy_ref, metadata: p.metadata || {},
legacy_ref: p.legacy_ref, metadata: p.metadata || {},
};
};
let id, created = false;
let id, created = false;
+ let staleHash = false;
if (found) {
if (found) {
id = found.id;
id = found.id;
// Knowledge the row doesn't carry survives a re-projection.
// Knowledge the row doesn't carry survives a re-projection.
if (!fields.content_hash) fields.content_hash = found.content_hash;
if (!fields.content_hash) fields.content_hash = found.content_hash;
fields.metadata = { ...parseJson(found.metadata, {}), ...fields.metadata };
fields.metadata = { ...parseJson(found.metadata, {}), ...fields.metadata };
+ // A hash the content-hash job computed (hash_basis: the local file it read) holds only while
+ // that file is unchanged: a present local copy at another path or of another size (a remux,
+ // a re-cut, a regenerated thumbnail) makes it stale. A copy that moved to B2/R2 keeps it.
+ const basis = fields.metadata.hash_basis;
+ if (basis && fields.content_hash === found.content_hash && !p.content_hash) {
+ const local = p.locations.find(l => l.provider === 'local' && l.state === 'present');
+ if (local && (local.key !== basis.key || Number(local.size_bytes) !== Number(basis.size))) {
+ fields.content_hash = null;
+ delete fields.metadata.hash_basis;
+ staleHash = true;
+ }
+ }
if (found.lifecycle_status === 'deleted') fields.deleted_at = null; // the row exists, so the object does
if (found.lifecycle_status === 'deleted') fields.deleted_at = null; // the row exists, so the object does
// owner_subject (filled by the owner-subject job) is the subject of owner_user_id; a different owner
// owner_subject (filled by the owner-subject job) is the subject of owner_user_id; a different owner
// drops it until the job resolves the new one.
// drops it until the job resolves the new one.
@@ -222,7 +235,7 @@ function project(p) {
@@ -222,7 +235,7 @@ function project(p) {
created = true;
created = true;
}
}
const keep = new Set(p.locations.map(l => l.provider));
const keep = new Set(p.locations.map(l => l.provider));
- for (const loc of p.locations) upsertLocation(id, loc);
+ for (const loc of p.locations) upsertLocation(id, staleHash && loc.provider === 'local' ? { ...loc, clearChecksum: true } : loc);
for (const l of listLocations(id)) if (!keep.has(l.provider)) db.run('DELETE FROM media_locations WHERE id = ?', [l.id]);
for (const l of listLocations(id)) if (!keep.has(l.provider)) db.run('DELETE FROM media_locations WHERE id = ?', [l.id]);
return { id, created };
return { id, created };
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,127 @@
@@ -0,0 +1,127 @@
+'use strict';
+// object.hash (server/jobs/content-hash.js): ready objects with a present local copy and no hash get
+// sha256 on the object (content_hash + metadata.hash_basis) and on the local location (checksum), in
+// bounded batches (count and bytes); files still changing or of another size than the object are
+// skipped; a re-projection that finds the local file changed drops the job's hash (a remux never
+// leaves a stale one) while a move to B2 keeps it; scheduled copy verification then checks against it.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const crypto = require('crypto');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-hash-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+Object.assign(process.env, {
+ DB_PATH: path.join(tmp, '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_INVARIANT_SCAN_HOURS: '0', MEDIA_HASH_SETTLE_S: '60', MEDIA_HASH_INTERVAL_MIN: '15',
+});
+
+const db = require('../server/db/database');
+const model = require('../server/objects/model');
+const queue = require('../server/jobs/queue');
+const worker = require('../server/jobs/worker');
+const hashJob = require('../server/jobs/content-hash');
+const verify = require('../server/objects/verify-job');
+
+db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+db.upsertApp({ app_id: 'games', api_key: 'games-key' });
+
+const sha = (buf) => crypto.createHash('sha256').update(buf).digest('hex');
+const age = (file, s = 3600) => { const t = (Date.now() - s * 1000) / 1000; fs.utimesSync(file, t, t); };
+function vodWith(name, bytes, { app = 'live', old = true } = {}) {
+ const file = path.join(process.env.VOD_PATH, name);
+ fs.writeFileSync(file, bytes);
+ if (old) age(file);
+ const id = Number(db.run(`INSERT INTO vods (app_id, title, file_path, file_size, duration_seconds, is_public, visibility, health_status) VALUES (?, 't', ?, ?, 60, 1, 'public', 'ok')`,
+ [app, file, bytes.length]).lastInsertRowid);
+ model.sync('vod', id);
+ return { id, file, obj: () => model.getObject(db.get('SELECT object_id FROM vods WHERE id = ?', [id]).object_id) };
+}
+const localLoc = (objId) => db.get("SELECT * FROM media_locations WHERE object_id = ? AND provider = 'local'", [objId]);
+
+(async () => {
+ const a = vodWith('a.mp4', Buffer.alloc(300000, 1));
+ const b = vodWith('b.mp4', Buffer.alloc(200000, 2));
+ const fresh = vodWith('fresh.mp4', Buffer.alloc(1000, 3), { old: false });
+ const other = vodWith('other.mp4', Buffer.alloc(5000, 4));
+ db.run('UPDATE media_objects SET size_bytes = 999 WHERE id = ?', [other.obj().id]); // the local copy is another version
+ const games = vodWith('g.mp4', Buffer.alloc(4000, 5), { app: 'games' });
+ assert.strictEqual(a.obj().content_hash, null);
+ const before = hashJob.remaining('live');
+ assert.strictEqual(before, 4);
+
+ // ── 1. A batch hashes local copies onto the object and the location, and skips what it must ──
+ const r = await hashJob.hashBatch({ appId: 'live' });
+ assert.deepStrictEqual([r.hashed, r.bytes], [2, 500000]);
+ assert.deepStrictEqual(r.skipped, { recently_changed: 1, size_differs_from_object: 1 });
+ assert.strictEqual(r.remaining, 2);
+ const oa = a.obj();
+ assert.strictEqual(oa.content_hash, sha(fs.readFileSync(a.file)));
+ assert.strictEqual(localLoc(oa.id).checksum, oa.content_hash, 'the location carries it too');
+ assert.ok(localLoc(oa.id).verified_at);
+ assert.deepStrictEqual([JSON.parse(oa.metadata).hash_basis.key, JSON.parse(oa.metadata).hash_basis.size], [path.resolve(a.file), 300000]);
+ assert.strictEqual(games.obj().content_hash, null, 'one tenant per run');
+ console.log('✅ sha256 of local copies recorded on object and location; changing files and other versions skipped');
+
+ // ── 2. Bounded by bytes: a run stops at its budget (the first file always fits), the next continues ──
+ const big1 = vodWith('big1.mp4', Buffer.alloc(3 * 1048576, 6));
+ const big2 = vodWith('big2.mp4', Buffer.alloc(3 * 1048576, 7));
+ const r2 = await hashJob.hashBatch({ appId: 'live', budgetMb: 4 });
+ assert.deepStrictEqual([r2.hashed, r2.skipped.budget], [1, 1]);
+ assert.strictEqual([big1, big2].filter(v => v.obj().content_hash).length, 1, 'one of the two 3 MB files fits a 4 MB budget');
+ const r3 = await hashJob.hashBatch({ appId: 'live', budgetMb: 4 });
+ assert.strictEqual(r3.hashed, 1);
+ assert.ok(big1.obj().content_hash && big2.obj().content_hash, 'the next run takes the other');
+ const r4 = await hashJob.hashBatch({ appId: 'live', limit: 1 });
+ assert.strictEqual(r4.hashed + Object.values(r4.skipped).reduce((x, y) => x + y, 0), 1, 'limit bounds the count');
+ console.log('✅ bounded batches: object count and byte budget');
+
+ // ── 3. A changed local file drops the job's hash on re-projection; a move to B2 keeps it ──
+ fs.appendFileSync(b.file, Buffer.alloc(1000, 9)); // a remux in place: new size
+ db.run('UPDATE vods SET file_size = ? WHERE id = ?', [201000, b.id]);
+ model.sync('vod', b.id);
+ const ob = b.obj();
+ assert.strictEqual(ob.content_hash, null, 'stale hash dropped');
+ assert.strictEqual(localLoc(ob.id).checksum, null);
+ assert.strictEqual(JSON.parse(ob.metadata).hash_basis, undefined);
+ age(b.file);
+ await hashJob.hashBatch({ appId: 'live', objectId: ob.id });
+ assert.strictEqual(b.obj().content_hash, sha(fs.readFileSync(b.file)), 'hashed again from the new bytes');
+ const kept = oa.content_hash;
+ fs.unlinkSync(a.file); // offloaded: local copy gone, B2 canonical
+ db.run("UPDATE vods SET storage_provider = 'b2', storage_key = 'vods/a.mp4' WHERE id = ?", [a.id]);
+ model.sync('vod', a.id);
+ assert.strictEqual(a.obj().content_hash, kept, 'the bytes moved; the hash stays');
+ console.log('✅ a remuxed local file loses the stale hash (and is re-hashed); a move to B2 keeps it');
+
+ // ── 4. Verification uses it: a corrupted local copy of the same size is caught ──
+ const c = vodWith('c.mp4', Buffer.alloc(100000, 8));
+ await hashJob.hashBatch({ appId: 'live', objectId: c.obj().id });
+ const good = await verify.verifyObject(c.obj(), { head: async () => undefined, upload: async () => undefined, hashMaxBytes: 64 * 1048576, repairCorrupt: false, budget: { left: 0 } });
+ assert.strictEqual(good.locations.find(l => l.provider === 'local').state, 'present');
+ const fd = fs.openSync(c.file, 'r+'); fs.writeSync(fd, Buffer.from([0]), 0, 1, 500); fs.closeSync(fd);
+ const bad = await verify.verifyObject(c.obj(), { head: async () => undefined, upload: async () => undefined, hashMaxBytes: 64 * 1048576, repairCorrupt: false, budget: { left: 0 } });
+ assert.strictEqual(bad.locations.find(l => l.provider === 'local').state, 'corrupt', 'bit rot of the same size is now visible');
+ console.log('✅ scheduled verification checks local copies against the recorded hash');
+
+ // ── 5. As a job: scheduled per tenant with work, run through the worker, pruned like thumbnails ──
+ assert.throws(() => hashJob.spec.validate({ obj: null, params: { limit: 0 } }), /limit/);
+ const n = hashJob.schedule(Date.now());
+ assert.strictEqual(n, 2, 'one per tenant with unhashed local copies (live, games)');
+ assert.strictEqual(hashJob.schedule(Date.now()), 0, 'once per period');
+ const gj = queue.list('games', { type: 'object.hash' }).jobs[0];
+ const done = await worker.runNow(gj.id);
+ assert.strictEqual(done.status, 'succeeded');
+ assert.strictEqual(JSON.parse(done.result).hashed, 1);
+ assert.ok(games.obj().content_hash);
+ db.run("UPDATE media_jobs SET finished_at = datetime('now', '-40 days') WHERE id = ?", [gj.id]);
+ assert.strictEqual(queue.prune({ days: 30 }), 1, 'finished hash jobs are pruned');
+ console.log('✅ object.hash runs as a scheduled job per tenant and is pruned when finished');
+
+ db.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('\n✅ All content-hash tests passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });