openvibers/openvibe.media Commits
Scheduled copy verification: every 10 min verify the 50 least recently verified ready objects (local stat + sha256 up to 64 MB, B2/R2 HEAD + size), record verdicts on media_locations, media_verifications and media_verify_runs, restore a missing remote copy from a verified-good local copy (2 per run; corrupt remote only with MEDIA_VERIFY_REPAIR_CORRUPT=1), never delete anything; media_objects_no_good_copy on /metrics and an optional object_copies check in /api/ready (degraded, never not-ready); scripts/no-good-copy-report.js lists those objects with owners, references and locations, read-only
Changes
@@@ -41,10 +41,11 @@ server/thumbnails/ thumbnail service + /api/v1/:app/thumbnailspublic/routes.js public /v /c /p /t /fobjects/ canonical object model: model (projections, holds), routes (/api/v2 + /o),-- backfill, reconcile, invariant, signing — see docs/object-model.md++ backfill, reconcile, invariant, signing, verify-job (scheduled copy++ verification), copy-report — see docs/object-model.md(openvibe-shared: pinned OpenVibe.Shared v1.0.0 release, installed by npm)scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)--scripts/backfill-objects.js / reconcile-objects.js / object-invariant.js object-model operator tools++scripts/backfill-objects.js / reconcile-objects.js / object-invariant.js / no-good-copy-report.js object-model operator tools```## Visitor sign-in@@@ -7,7 +7,7 @@ projection over an object through its `object_id` column. Every existing URL andunchanged. New code talks to objects through the [v2 API](#object-api-v2).Code: `server/objects/` (`model.js`, `routes.js`, `backfill.js`, `reconcile.js`, `invariant.js`,--`signing.js`). Schema: the bottom of `server/db/schema.sql`. Tests: `test/objects-*.test.js`.++`signing.js`, `verify-job.js`, `copy-report.js`). Schema: the bottom of `server/db/schema.sql`. Tests: `test/objects-*.test.js`.## Tables@@@ -25,7 +25,7 @@ Location `state`:- `present`: verified. The local file exists, or a HEAD answered with the right size.- `missing`: verified absent.--- `pending`: believed there but not verified yet. Every remote copy starts here until reconciliation runs with `--verify`.++- `pending`: believed there but not verified yet. Every remote copy starts here until reconciliation runs with `--verify` or the [scheduled verification](#scheduled-verification) reaches it.- `corrupt`: the size or hash does not match.A copy that reconciliation has verified keeps that state when the row is projected again. The@@@ -209,6 +209,34 @@ The exit code is 1 when issues were found.listing) and incomplete multipart uploads. `server/objects/reconcile.js` takes a `head` function,so tests run it against a fake provider.++## Scheduled verification++++`server/objects/verify-job.js` runs inside the service (started from `server/index.js`, two minutes++after boot, then every `MEDIA_VERIFY_INTERVAL_MIN`). Each run takes the `MEDIA_VERIFY_BATCH` ready++objects verified least recently (never-verified first), so every object is covered over time: with++the defaults, 7,200 objects a day.++++For each object it:++++1. **Checks every copy**, with the same rules as `reconcile-objects.js --verify`. A local copy is `present` when the file exists, and `corrupt` when the object carries a `content_hash` and the file (≤ `MEDIA_VERIFY_HASH_MAX_MB`) hashes differently. A B2/R2 copy is `missing` on a 404 and `corrupt` when its size differs. An unconfigured or unreachable provider is *unverifiable* and keeps its state. The verdict goes to `media_locations` (`state`, `size_bytes`, `verified_at`), but only if the location still has the key that was checked.++2. **Restores a missing B2/R2 copy** by uploading a verified-good local copy (right size, hash not contradicted) to the same key through the storage engine's `uploadFile`. That call HEADs the upload afterwards and fails on a size mismatch. At most `MEDIA_VERIFY_MAX_REUPLOADS` uploads run per run, and the rest wait for the next one. A `corrupt` remote copy is overwritten only when `MEDIA_VERIFY_REPAIR_CORRUPT=1`.++3. **Records the outcome.** `media_verifications` gets one row per object (`verified_at`, `status` `good` / `no_good_copy` / `unverifiable`, `good_providers`, per-location `detail`). `media_verify_runs` gets one summary row per run.++++**It never deletes anything.** No file, remote object or row is removed, and no lifecycle is changed.++++**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:++++- `media_objects_no_good_copy` on `/metrics`, next to `media_verify_last_run_timestamp_seconds`.++- The optional `object_copies` check in `/api/ready`. It is degraded while the count is above zero and never makes Media not-ready.++- The report, which lists each one for an operator to decide on:++++ ```++ node scripts/no-good-copy-report.js [--app live] [--json] [--out report.json] [--db ./data/media.db]++ ```++++ For each object it prints the owner (app, `owner_user_id`, `owner_subject`), `legacy_ref`, the `vods`/`clips`/`files`/`pastes` rows that point at it, its relationships, variants and holds, every location with its state and `verified_at`, and the last verification. The report opens the database read-only. It exits 1 when there are any.++## Public object-size invariantThis is roadmap W4 deliverable 5, stated as **policy** rather than a copied constant:@@@ -247,6 +275,12 @@ which splits playback into objects that fit.| `MEDIA_SIGNED_URL_TTL_S` | 300 | default signed download lifetime || `MEDIA_UPLOAD_TOKEN_TTL_S` | 3600 | upload token lifetime || `MEDIA_PUBLIC_OBJECT_{MAX,TARGET,WARN}_MB` | 500 / 256 / 384 | invariant thresholds |++| `MEDIA_VERIFY_ENABLED` | on | `0` turns the scheduled verification off |++| `MEDIA_VERIFY_INTERVAL_MIN` | 10 | minutes between verification runs |++| `MEDIA_VERIFY_BATCH` | 50 | objects per run |++| `MEDIA_VERIFY_HASH_MAX_MB` | 64 | sha256 local copies up to this size (objects with a `content_hash`) |++| `MEDIA_VERIFY_MAX_REUPLOADS` | 2 | missing remote copies restored per run |++| `MEDIA_VERIFY_REPAIR_CORRUPT` | off | `1` also overwrites a corrupt remote copy from a good local copy |## Not in this passnnew file mode 100644@@@ -0,0 +1,37 @@++#!/usr/bin/env node++'use strict';++/**++ * List every ready object that has no good copy anywhere (docs/object-model.md#scheduled-verification),++ * with its owner, the rows that reference it, its locations and their verified states, so an operator++ * can decide what to do with each one. READ-ONLY: the database is opened read-only and nothing is++ * deleted, moved or re-projected.++ *++ * node scripts/no-good-copy-report.js [--app live] [--json] [--out report.json] [--db ./data/media.db]++ *++ * Exit code 0 when there are none, 1 when some were found, 2 on error.++ */++const fs = require('fs');++const path = require('path');++++const args = process.argv.slice(2);++const has = (name) => args.includes(`--${name}`);++const arg = (name, dflt) => { const i = args.indexOf(`--${name}`); return i >= 0 ? args[i + 1] : dflt; };++++try {++ const dbPath = arg('db') ? path.resolve(arg('db')) : require('../server/config').db.path;++ const Database = require('better-sqlite3');++ const h = new Database(dbPath, { readonly: true, fileMustExist: true });++ const q = {++ all: (sql, params = []) => h.prepare(sql).all(...params),++ get: (sql, params = []) => h.prepare(sql).get(...params),++ };++ const { buildReport, formatReport } = require('../server/objects/copy-report');++ const report = buildReport(q, { appId: arg('app', null) });++ h.close();++ if (arg('out')) fs.writeFileSync(arg('out'), JSON.stringify(report, null, 2));++ console.log(has('json') ? JSON.stringify(report, null, 2) : formatReport(report));++ process.exit(report.count ? 1 : 0);++} catch (err) {++ console.error(err.message);++ process.exit(2);++}@@@ -57,6 +57,16 @@ const config = {publicWarnMb: intEnv('MEDIA_PUBLIC_OBJECT_WARN_MB', 384),},++ // Scheduled copy verification (server/objects/verify-job.js; docs/object-model.md#scheduled-verification).++ verify: {++ enabled: !['0', 'false', 'off'].includes(String(process.env.MEDIA_VERIFY_ENABLED || '').toLowerCase()),++ intervalMin: intEnv('MEDIA_VERIFY_INTERVAL_MIN', 10), // one batch this often++ batch: intEnv('MEDIA_VERIFY_BATCH', 50), // objects per run (least recently verified first)++ hashMaxMb: intEnv('MEDIA_VERIFY_HASH_MAX_MB', 64), // sha256 local copies up to this size (objects with a content_hash)++ maxReuploads: intEnv('MEDIA_VERIFY_MAX_REUPLOADS', 2), // missing remote copies restored from a good local copy, per run++ repairCorrupt: ['1', 'true', 'on'].includes(String(process.env.MEDIA_VERIFY_REPAIR_CORRUPT || '').toLowerCase()),++ },++rtp: {portMin: intEnv('RTP_PORT_MIN', 12000),portMax: intEnv('RTP_PORT_MAX', 12199),@@@ -313,3 +313,33 @@ CREATE TABLE IF NOT EXISTS media_invariant_violations (last_seen_at DATETIME DEFAULT CURRENT_TIMESTAMP,resolved_at DATETIME);++++-- Scheduled copy verification (server/objects/verify-job.js). One row per ready object: when it was++-- last checked and what that check found; the least recently verified objects go first in each run.++-- Verification only records and restores copies — it never deletes bytes or rows.++CREATE TABLE IF NOT EXISTS media_verifications (++ object_id TEXT PRIMARY KEY,++ verified_at TEXT NOT NULL, -- ISO-8601 with ms (orders the rotation)++ status TEXT NOT NULL CHECK(status IN ('good', 'no_good_copy', 'unverifiable')),++ good_providers TEXT, -- comma list of providers holding a good copy++ detail TEXT NOT NULL DEFAULT '{}', -- per-location verdicts and any re-upload++ run_id INTEGER++);++CREATE INDEX IF NOT EXISTS idx_media_verifications_at ON media_verifications(verified_at);++CREATE INDEX IF NOT EXISTS idx_media_verifications_status ON media_verifications(status);++++-- One row per verification run (summary).++CREATE TABLE IF NOT EXISTS media_verify_runs (++ id INTEGER PRIMARY KEY AUTOINCREMENT,++ started_at TEXT NOT NULL,++ finished_at TEXT,++ objects_checked INTEGER NOT NULL DEFAULT 0,++ locations_checked INTEGER NOT NULL DEFAULT 0,++ good INTEGER NOT NULL DEFAULT 0,++ no_good_copy INTEGER NOT NULL DEFAULT 0,++ unverifiable INTEGER NOT NULL DEFAULT 0,++ reuploaded INTEGER NOT NULL DEFAULT 0,++ reupload_failed INTEGER NOT NULL DEFAULT 0,++ no_good_copy_total INTEGER, -- ready objects with no good copy anywhere, after this run++ error TEXT++);@@@ -172,6 +172,7 @@ vodStorage.checkProviders()vodStorage.start(); // tiering sweephealthJob.start(); // health scan + quarantine cleanuprequire('./vod/clip-jobs').start(); // failed clip re-cuts (auto-retry, hot-fetch)++require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.every(5 * 60 * 1000, () => {try {@@@ -227,6 +228,7 @@ function shutdown(signal) {try { recorder.stopAll(); } catch { /* */ }try { vodStorage.stop(); } catch { /* */ }try { healthJob.stop(); } catch { /* */ }++ try { require('./objects/verify-job').stop(); } catch { /* */ }try { auth.stopJwksRefresh(); } catch { /* */ }try { require('./events')._reset(); } catch { /* */ }for (const t of timers) clearInterval(t);nnew file mode 100644@@@ -0,0 +1,104 @@++/**++ * OpenVibe.Media — "no good copy" query and report (docs/object-model.md#scheduled-verification).++ *++ * A ready object has NO GOOD COPY when none of its locations is `present` (verified there) or++ * `pending` (not verified yet, so possibly fine). That covers objects whose every copy was found++ * missing/corrupt, and ready objects with no location row at all.++ *++ * Pure SQL over a small `q` interface — { all(sql, params), get(sql, params) } — so the service++ * (server/db/database.js), the metrics/readiness checks and the read-only report script++ * (scripts/no-good-copy-report.js, which opens the database file read-only) share one definition.++ * Nothing here writes.++ */++'use strict';++++const NO_GOOD_COPY_WHERE = `o.lifecycle_status = 'ready'++ AND NOT EXISTS (SELECT 1 FROM media_locations l WHERE l.object_id = o.id AND l.state IN ('present', 'pending'))`;++++function countNoGoodCopy(q, { appId = null } = {}) {++ const r = q.get(`SELECT COUNT(*) AS n FROM media_objects o WHERE ${NO_GOOD_COPY_WHERE}${appId ? ' AND o.app_id = ?' : ''}`, appId ? [appId] : []);++ return Number(r && r.n) || 0;++}++++function listNoGoodCopy(q, { appId = null } = {}) {++ return q.all(`SELECT o.* FROM media_objects o WHERE ${NO_GOOD_COPY_WHERE}${appId ? ' AND o.app_id = ?' : ''} ORDER BY o.app_id, o.id`, appId ? [appId] : []);++}++++function tableExists(q, name) {++ return !!q.get("SELECT 1 AS x FROM sqlite_master WHERE type = 'table' AND name = ?", [name]);++}++++function parseJson(s, dflt) { try { return s ? JSON.parse(s) : dflt; } catch { return dflt; } }++++/** Rows elsewhere that point at the object: the inherited projections, relationships, variants, holds. */++function references(q, obj) {++ const refs = {++ vods: q.all(`SELECT id, app_id, user_id, stream_id, managed_stream_id, title, visibility, is_public, file_path, file_size,++ storage_provider, storage_key, created_at FROM vods WHERE object_id = ?`, [obj.id]),++ clips: q.all(`SELECT id, app_id, vod_id, stream_id, user_id, channel_user_id, title, visibility, is_public, file_path, status, created_at++ FROM clips WHERE object_id = ?`, [obj.id]),++ files: q.all('SELECT key, app_id, user_id, original_name, size, mime, created_at FROM files WHERE object_id = ?', [obj.id]),++ pastes: q.all('SELECT id, slug, app_id, user_id, title, type, visibility, screenshot_path, created_at FROM pastes WHERE object_id = ?', [obj.id]),++ relationships: q.all(`SELECT from_object_id, relation, to_object_id FROM media_relationships++ WHERE from_object_id = ? OR to_object_id = ? ORDER BY id`, [obj.id, obj.id]),++ variants: q.all(`SELECT object_id, variant_name, derived_object_id FROM media_variants++ WHERE object_id = ? OR derived_object_id = ? ORDER BY id`, [obj.id, obj.id]),++ holds: q.all('SELECT id, kind, reason, created_by, created_at FROM media_holds WHERE object_id = ? AND released_at IS NULL ORDER BY id', [obj.id]),++ };++ return refs;++}++++/** Everything an operator needs to decide what to do with each object that has no good copy. */++function buildReport(q, { appId = null } = {}) {++ const hasVerifications = tableExists(q, 'media_verifications');++ const items = listNoGoodCopy(q, { appId }).map((o) => {++ const last = hasVerifications ? q.get('SELECT verified_at, status, detail FROM media_verifications WHERE object_id = ?', [o.id]) : null;++ const md = parseJson(o.metadata, {});++ return {++ object_id: o.id,++ app_id: o.app_id,++ kind: o.kind,++ visibility: o.visibility,++ title: md.title || md.filename || null,++ size_bytes: Number(o.size_bytes) || 0,++ content_hash: o.content_hash || null,++ created_at: o.created_at,++ owner: { app: o.owner_app || o.app_id, user_id: o.owner_user_id ?? null, subject: o.owner_subject || null },++ legacy_ref: o.legacy_ref || null,++ canonical: { provider: o.canonical_provider || null, key: o.canonical_key || null },++ locations: q.all('SELECT provider, bucket, key, state, size_bytes, verified_at FROM media_locations WHERE object_id = ? ORDER BY id', [o.id]),++ references: references(q, o),++ last_verification: last ? { verified_at: last.verified_at, status: last.status, detail: parseJson(last.detail, {}) } : null,++ };++ });++ return { generated_at: new Date().toISOString(), app_id: appId, count: items.length, objects: items };++}++++function mb(n) { return `${((Number(n) || 0) / 1048576).toFixed(1)} MB`; }++++function formatReport(report) {++ const out = [`${report.count} ready object(s) with no good copy${report.app_id ? ` in ${report.app_id}` : ''} (generated ${report.generated_at})`];++ if (!report.count) return out.join('\n');++ out.push('Nothing was changed. Decide per object: restore the bytes from elsewhere, or retire it through its owning app.');++ for (const it of report.objects) {++ out.push('');++ out.push(`${it.object_id} ${it.kind} ${it.app_id} ${mb(it.size_bytes)} created ${it.created_at} ${it.visibility}`);++ if (it.title) out.push(` title: ${it.title}`);++ out.push(` owner: app=${it.owner.app} user_id=${it.owner.user_id ?? '-'} subject=${it.owner.subject || '-'}`);++ out.push(` legacy_ref: ${it.legacy_ref || '(native object)'}`);++ if (!it.locations.length) out.push(' locations: (none recorded)');++ for (const l of it.locations) out.push(` location: ${l.provider}${l.bucket ? `/${l.bucket}` : ''} ${l.key} ${l.state}${l.size_bytes != null ? ` ${l.size_bytes} bytes` : ''}${l.verified_at ? ` verified ${l.verified_at}` : ''}`);++ const r = it.references;++ for (const v of r.vods) out.push(` vods#${v.id}: user ${v.user_id ?? '-'} "${v.title || ''}" ${v.visibility || ''} provider=${v.storage_provider || 'local'} file=${v.file_path || '-'}`);++ for (const c of r.clips) out.push(` clips#${c.id}: vod ${c.vod_id ?? '-'} user ${c.user_id ?? '-'} "${c.title || ''}" ${c.visibility || ''} file=${c.file_path || '-'}`);++ for (const f of r.files) out.push(` files/${f.key}: user ${f.user_id ?? '-'} ${f.original_name || ''}`);++ for (const p of r.pastes) out.push(` pastes/${p.slug}: user ${p.user_id ?? '-'} ${p.type || ''} ${p.visibility || ''}`);++ for (const x of r.relationships) out.push(` relationship: ${x.from_object_id} ${x.relation} ${x.to_object_id}`);++ for (const x of r.variants) out.push(` variant: ${x.object_id} ${x.variant_name} → ${x.derived_object_id}`);++ for (const h of r.holds) out.push(` HOLD ${h.kind}: ${h.reason || ''} (by ${h.created_by || '-'})`);++ if (it.last_verification) out.push(` last verified ${it.last_verification.verified_at}: ${it.last_verification.status}`);++ }++ return out.join('\n');++}++++module.exports = { NO_GOOD_COPY_WHERE, countNoGoodCopy, listNoGoodCopy, buildReport, formatReport };nnew file mode 100644@@@ -0,0 +1,238 @@++/**++ * OpenVibe.Media — scheduled copy verification (docs/object-model.md#scheduled-verification).++ *++ * Every MEDIA_VERIFY_INTERVAL_MIN minutes one run takes the MEDIA_VERIFY_BATCH ready objects that++ * were verified least recently (never-verified first), so every object is covered over time, and++ * for each one:++ * - checks every copy: local = the file exists (size, and sha256 when the object carries a hash++ * and the file is ≤ MEDIA_VERIFY_HASH_MAX_MB); B2/R2 = a HEAD with the right size. The verdict++ * (present / missing / corrupt) is written to media_locations, exactly as++ * `reconcile-objects.js --verify` would. An unconfigured provider is unverifiable, never missing;++ * - restores a MISSING B2/R2 copy by uploading a verified-good local copy to the same key++ * (at most MEDIA_VERIFY_MAX_REUPLOADS per run). A corrupt remote copy is only overwritten when++ * MEDIA_VERIFY_REPAIR_CORRUPT=1;++ * - records the outcome in media_verifications (one row per object) and the run in media_verify_runs.++ *++ * It NEVER deletes anything: no file unlink, no remote delete, no row delete. Objects with no good++ * copy anywhere are only counted (media_objects_no_good_copy on /metrics, the optional++ * `object_copies` readiness check) and listed by scripts/no-good-copy-report.js for an operator.++ */++'use strict';++++const fs = require('fs');++const crypto = require('crypto');++const db = require('../db/database');++const config = require('../config');++const copyReport = require('./copy-report');++++const MB = 1024 * 1024;++const REMOTE = ['b2', 'r2'];++++let _timer = null, _startTimer = null, _busy = false;++++const nowIso = () => new Date().toISOString();++++function sha256File(p) {++ return new Promise((resolve, reject) => {++ const h = crypto.createHash('sha256');++ const s = fs.createReadStream(p);++ s.on('data', d => h.update(d));++ s.on('end', () => resolve(h.digest('hex')));++ s.on('error', reject);++ });++}++++/** Real provider HEAD: undefined when the provider is not configured here, null on 404. */++async function defaultHead(provider, key) {++ return require('./reconcile').defaultHead(provider, key);++}++++/** Real re-upload through the storage engine (it HEADs afterwards and throws on a size mismatch). */++async function defaultUpload(provider, key, filePath, contentType, loc) {++ const vs = require('../vod/vod-storage');++ if (!vs.providerConfigured(provider)) return undefined;++ if (loc && loc.bucket && vs.bucketFor(provider) && loc.bucket !== vs.bucketFor(provider)) {++ throw new Error(`location is in bucket ${loc.bucket}, the configured ${provider} bucket is ${vs.bucketFor(provider)}`);++ }++ return vs.uploadFile(provider, key, filePath, contentType || 'application/octet-stream');++}++++/** Write a verdict only if the location still points where it did when we looked (a tier move may have re-projected it). */++function writeState(loc, state, size) {++ db.run(`UPDATE media_locations SET state = ?, size_bytes = COALESCE(?, size_bytes), verified_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP++ WHERE id = ? AND key = ?`, [state, size ?? null, loc.id, loc.key]);++}++++/**++ * Verify one object's copies and restore missing remote ones from a good local copy.++ * Returns { status, good_providers, locations: [...], reuploads: [...] }.++ */++async function verifyObject(obj, opts) {++ const { head, upload, hashMaxBytes, repairCorrupt, budget } = opts;++ const expected = Number(obj.size_bytes) || 0;++ const locs = db.all('SELECT * FROM media_locations WHERE object_id = ? ORDER BY id', [obj.id]);++ const results = [];++ let goodLocal = null;++++ for (const loc of locs) {++ const r = { provider: loc.provider, key: loc.key, before: loc.state, state: loc.state, size_bytes: null, note: null };++ results.push(r);++ if (loc.provider === 'local') {++ let st = null;++ try { const s = fs.statSync(loc.key); if (s.isFile()) st = s; } catch { /* missing */ }++ if (!st) { r.state = 'missing'; writeState(loc, 'missing', null); continue; }++ r.size_bytes = st.size;++ // Same rule as reconcile: a local size that differs from the recorded size is reported, not++ // condemned (recordings and remuxes change it) — but such a file is never used as a re-upload source.++ let state = 'present';++ let sizeOk = !expected || st.size === expected;++ if (!sizeOk) r.note = `size ${st.size} ≠ expected ${expected}`;++ let hashOk = null;++ if (obj.content_hash && st.size <= hashMaxBytes) {++ const actual = await sha256File(loc.key).catch(() => null);++ if (actual) {++ hashOk = actual === obj.content_hash;++ if (!hashOk) { state = 'corrupt'; r.note = 'sha256 mismatch'; }++ }++ }++ r.state = state;++ writeState(loc, state, st.size);++ if (state === 'present' && sizeOk && hashOk !== false) goodLocal = loc;++ continue;++ }++ // b2 / r2++ let h;++ try { h = await head(loc.provider, loc.key, loc.bucket); }++ catch (err) { r.state = loc.state; r.note = `unverifiable: ${err.message}`; r.unverifiable = true; continue; }++ if (h === undefined) { r.state = loc.state; r.note = 'unverifiable: provider not configured'; r.unverifiable = true; continue; }++ if (!h) { r.state = 'missing'; writeState(loc, 'missing', null); continue; }++ r.size_bytes = Number(h.size);++ const state = expected && Number(h.size) !== expected ? 'corrupt' : 'present';++ if (state === 'corrupt') r.note = `size ${h.size} ≠ expected ${expected}`;++ r.state = state;++ writeState(loc, state, Number(h.size));++ }++++ // Restore missing (and, when allowed, corrupt) remote copies from a verified-good local copy.++ const reuploads = [];++ if (goodLocal) {++ for (const loc of locs) {++ if (!REMOTE.includes(loc.provider)) continue;++ const r = results.find(x => x.provider === loc.provider);++ const want = r.state === 'missing' || (repairCorrupt && r.state === 'corrupt');++ if (!want) continue;++ if (budget.left <= 0) { reuploads.push({ provider: loc.provider, key: loc.key, ok: false, error: 'deferred: re-upload budget for this run used up' }); continue; }++ budget.left--;++ try {++ const out = await upload(loc.provider, loc.key, goodLocal.key, obj.mime_type, loc);++ if (out === undefined) { reuploads.push({ provider: loc.provider, key: loc.key, ok: false, error: 'provider not configured' }); continue; }++ const size = Number(out && out.size) || fs.statSync(goodLocal.key).size;++ writeState(loc, 'present', size);++ r.state = 'present'; r.size_bytes = size; r.note = `re-uploaded from ${goodLocal.key}`;++ reuploads.push({ provider: loc.provider, key: loc.key, ok: true, size_bytes: size, from: goodLocal.key });++ } catch (err) {++ reuploads.push({ provider: loc.provider, key: loc.key, ok: false, error: err.message });++ }++ }++ }++++ const good = results.filter(x => x.state === 'present').map(x => x.provider);++ const unknown = results.some(x => x.unverifiable || x.state === 'pending');++ const status = good.length ? 'good' : unknown ? 'unverifiable' : 'no_good_copy';++ return { status, good_providers: good, locations: results, reuploads };++}++++/** One bounded run. Deps are injectable for tests: head(provider, key), upload(provider, key, file, mime, loc). */++async function runOnce({ batch = config.verify.batch, head = defaultHead, upload = defaultUpload,++ hashMaxBytes = config.verify.hashMaxMb * MB, maxReuploads = config.verify.maxReuploads, repairCorrupt = config.verify.repairCorrupt } = {}) {++ if (_busy) return { skipped: 'busy' };++ _busy = true;++ const summary = { started_at: nowIso(), finished_at: null, objects_checked: 0, locations_checked: 0, good: 0, no_good_copy: 0, unverifiable: 0,++ reuploaded: 0, reupload_failed: 0, no_good_copy_total: null, error: null, objects: [] };++ let runId = null;++ try {++ runId = db.run('INSERT INTO media_verify_runs (started_at) VALUES (?)', [summary.started_at]).lastInsertRowid;++ const n = Math.max(1, Number(batch) || 1);++ const objects = db.all(`SELECT o.* FROM media_objects o LEFT JOIN media_verifications v ON v.object_id = o.id++ WHERE o.lifecycle_status = 'ready'++ ORDER BY (v.verified_at IS NOT NULL), v.verified_at, o.id LIMIT ?`, [n]);++ const budget = { left: Math.max(0, Number(maxReuploads) || 0) };++ for (const obj of objects) {++ let res;++ try { res = await verifyObject(obj, { head, upload, hashMaxBytes, repairCorrupt, budget }); }++ catch (err) { res = { status: 'unverifiable', good_providers: [], locations: [], reuploads: [], error: err.message }; }++ summary.objects_checked++;++ summary.locations_checked += res.locations.length;++ summary[res.status]++;++ for (const u of res.reuploads) { if (u.ok) summary.reuploaded++; else if (!/^deferred/.test(u.error || '')) summary.reupload_failed++; }++ db.run(`INSERT INTO media_verifications (object_id, verified_at, status, good_providers, detail, run_id) VALUES (?, ?, ?, ?, ?, ?)++ ON CONFLICT(object_id) DO UPDATE SET verified_at = excluded.verified_at, status = excluded.status,++ good_providers = excluded.good_providers, detail = excluded.detail, run_id = excluded.run_id`,++ [obj.id, nowIso(), res.status, res.good_providers.join(',') || null,++ JSON.stringify({ locations: res.locations, reuploads: res.reuploads, error: res.error || undefined }), runId]);++ summary.objects.push({ object_id: obj.id, status: res.status, reuploads: res.reuploads.length });++ if (res.status === 'no_good_copy') console.warn(`[Verify] ${obj.id} (${obj.legacy_ref || obj.kind}) has no good copy — see scripts/no-good-copy-report.js`);++ for (const u of res.reuploads) {++ if (u.ok) console.log(`[Verify] ${obj.id}: restored ${u.provider}:${u.key} from the local copy`);++ else if (!/^deferred/.test(u.error || '')) console.warn(`[Verify] ${obj.id}: re-upload to ${u.provider} failed: ${u.error}`);++ }++ }++ } catch (err) {++ summary.error = err.message;++ console.warn('[Verify] run failed:', err.message);++ } finally {++ summary.finished_at = nowIso();++ try { summary.no_good_copy_total = copyReport.countNoGoodCopy(db); } catch { /* */ }++ if (runId != null) {++ try {++ db.run(`UPDATE media_verify_runs SET finished_at = ?, objects_checked = ?, locations_checked = ?, good = ?, no_good_copy = ?, unverifiable = ?,++ reuploaded = ?, reupload_failed = ?, no_good_copy_total = ?, error = ? WHERE id = ?`,++ [summary.finished_at, summary.objects_checked, summary.locations_checked, summary.good, summary.no_good_copy, summary.unverifiable,++ summary.reuploaded, summary.reupload_failed, summary.no_good_copy_total, summary.error, runId]);++ } catch { /* */ }++ }++ summary.run_id = runId;++ _busy = false;++ }++ if (summary.no_good_copy || summary.reuploaded || summary.reupload_failed) {++ console.log(`[Verify] run ${runId}: ${summary.objects_checked} object(s), ${summary.good} good, ${summary.no_good_copy} no good copy, `++ + `${summary.unverifiable} unverifiable, ${summary.reuploaded} restored, ${summary.reupload_failed} restore failed; `++ + `${summary.no_good_copy_total} ready object(s) with no good copy in total`);++ }++ return summary;++}++++/** Last finished run and the current no-good-copy count (readiness detail, operators). */++function status() {++ const last = db.get('SELECT * FROM media_verify_runs WHERE finished_at IS NOT NULL ORDER BY id DESC LIMIT 1');++ return {++ enabled: !!config.verify.enabled,++ running: !!_timer,++ no_good_copy: copyReport.countNoGoodCopy(db),++ verified_objects: db.get('SELECT COUNT(*) AS n FROM media_verifications').n,++ last_run: last ? { id: last.id, finished_at: last.finished_at, objects_checked: last.objects_checked, no_good_copy: last.no_good_copy,++ reuploaded: last.reuploaded, reupload_failed: last.reupload_failed, error: last.error || null } : null,++ };++}++++function start({ initialDelayMs = 2 * 60 * 1000 } = {}) {++ if (!config.verify.enabled) { console.log('[Verify] scheduled copy verification disabled (MEDIA_VERIFY_ENABLED=0)'); return false; }++ if (_timer || _startTimer) return true;++ const tick = () => { runOnce().catch((err) => console.warn('[Verify] run error:', err.message)); };++ const intervalMs = Math.max(1, config.verify.intervalMin) * 60 * 1000;++ _startTimer = setTimeout(() => {++ _startTimer = null;++ tick();++ _timer = setInterval(tick, intervalMs);++ if (_timer.unref) _timer.unref();++ }, initialDelayMs);++ if (_startTimer.unref) _startTimer.unref();++ return true;++}++++function stop() {++ if (_startTimer) { clearTimeout(_startTimer); _startTimer = null; }++ if (_timer) { clearInterval(_timer); _timer = null; }++}++++module.exports = { runOnce, verifyObject, status, start, stop };@@@ -9,7 +9,10 @@* local storage directory accepts a write. Optional (one capability degrades, the rest keeps* working): the Network public key (user-JWT and service-token routes; app-key tenants are* unaffected), each CONFIGURED remote tier (B2 cold storage, R2 cache; a HeadBucket at most once a-- * minute), and the Events outbox (webhooks still deliver when it backs up).++ * minute), the Events outbox (webhooks still deliver when it backs up), and object_copies: every++ * ready object has a good copy somewhere (server/objects/verify-job.js records it; objects with no++ * good copy are listed by scripts/no-good-copy-report.js). object_copies never fails readiness —++ * Media keeps serving everything else.** `instrument()` must run before any route is mounted; `mountReady()` any time after.*/@@@ -19,6 +22,8 @@ const crypto = require('crypto');const metrics = require('openvibe-shared/metrics');const { createReadiness } = require('openvibe-shared/ready');++const copyReport = require('./objects/copy-report');++const OUTBOX_BACKLOG_LIMIT = 1000;function instrument(app, { release }) {@@@ -59,6 +64,18 @@ function domainMetrics(registry, { db, recorder, events }) {name: 'media_objects', help: 'Media objects by lifecycle status', labelNames: ['lifecycle_status'],collect: () => db.all('SELECT lifecycle_status, COUNT(*) AS n FROM media_objects GROUP BY lifecycle_status').map((r) => ({ labels: { lifecycle_status: r.lifecycle_status }, value: r.n })),});++ registry.gauge({++ name: 'media_objects_no_good_copy', help: 'Ready media objects with no present or unverified copy anywhere (see scripts/no-good-copy-report.js)',++ collect: () => copyReport.countNoGoodCopy(db),++ });++ registry.gauge({++ name: 'media_verify_last_run_timestamp_seconds', help: 'When the scheduled copy verification last finished a run',++ collect: () => {++ const r = db.get('SELECT finished_at FROM media_verify_runs WHERE finished_at IS NOT NULL ORDER BY id DESC LIMIT 1');++ const t = r ? Date.parse(r.finished_at) : NaN;++ return Number.isFinite(t) ? Math.floor(t / 1000) : null; // no run yet: no series++ },++ });registry.gauge({name: 'media_events_outbox', help: 'Events outbox rows waiting to reach OpenVibe.Events, and rows Events rejected', labelNames: ['status'],collect: () => {@@@ -90,6 +107,16 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, remreturn { ok: true, detail: { pending: s.pending, rejected: s.rejected } };},});++ checks.push({++ name: 'object_copies', required: false, cacheMs: 60 * 1000, description: 'every ready object has a good copy (scheduled verification)',++ check: () => {++ const n = copyReport.countNoGoodCopy(db);++ const last = db.get('SELECT id, finished_at, objects_checked, reuploaded FROM media_verify_runs WHERE finished_at IS NOT NULL ORDER BY id DESC LIMIT 1');++ const detail = { no_good_copy: n, last_verify_run: last ? last.finished_at : null };++ if (n) return { ok: false, error: `${n} ready object(s) with no good copy (scripts/no-good-copy-report.js)`, detail };++ return { ok: true, detail };++ },++ });return createReadiness({service: 'media',release,nnew file mode 100644@@@ -0,0 +1,223 @@++'use strict';++// Scheduled copy verification (server/objects/verify-job.js) against a fake B2/R2 provider:++// bounded batches that rotate through every ready object, verdicts recorded on media_locations,++// media_verifications and media_verify_runs, a missing remote copy restored from a good local copy,++// nothing ever deleted, the no-good-copy count on /metrics and as a degraded (still ready) check in++// /api/ready, and scripts/no-good-copy-report.js listing those objects read-only.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const http = require('http');++const crypto = require('crypto');++const { execFileSync } = require('child_process');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-verify-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++const 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'), ASSETS_PATH: dir('assets'),++ MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9',++};++for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL', 'MEDIA_VERIFY_ENABLED', 'MEDIA_VERIFY_REPAIR_CORRUPT']) process.env[k] = '';++Object.assign(process.env, env);++++const db = require('../server/db/database');++const config = require('../server/config');++const model = require('../server/objects/model');++const job = require('../server/objects/verify-job');++const copyReport = require('../server/objects/copy-report');++++const d = db.getDb();++const sha = (b) => crypto.createHash('sha256').update(b).digest('hex');++const put = (name, content) => { const p = path.join(env.OBJECTS_PATH, 'live', name); fs.mkdirSync(path.dirname(p), { recursive: true }); fs.writeFileSync(p, content); return p; };++function obj(name, { size, hash = null, status = 'ready', locs = [], owner = 7 }) {++ const id = model.createObject({ app_id: 'live', kind: 'file', lifecycle_status: status, size_bytes: size, content_hash: hash,++ owner_user_id: owner, owner_subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', mime_type: 'application/octet-stream', metadata: { filename: name } });++ for (const l of locs) model.upsertLocation(id, l);++ return id;++}++++// Fake provider: bucket[provider][key] = size (absent → 404). r2 is "not configured" (HEAD undefined).++const bucket = { b2: {} };++const heads = [], uploads = [];++const head = async (provider, key) => { heads.push(`${provider}:${key}`); if (!bucket[provider]) return undefined; const s = bucket[provider][key]; return s === undefined ? null : { size: s }; };++const upload = async (provider, key, file) => { uploads.push(`${provider}:${key}`); if (!bucket[provider]) return undefined; const size = fs.statSync(file).size; bucket[provider][key] = size; return { size }; };++++const A = Buffer.from('aaaaaaaaaa'), B = Buffer.from('bbbbbbbbbbbb'), E = Buffer.from('eeee'), G = Buffer.from('gggggg');++const ids = {++ good: obj('a', { size: A.length, hash: sha(A), locs: [{ provider: 'local', key: put('a', A), state: 'present' }] }),++ restore: obj('b', { size: B.length, locs: [{ provider: 'local', key: put('b', B), state: 'present' }, { provider: 'b2', key: 'objs/b', state: 'pending' }] }),++ remoteCorrupt: obj('c', { size: 300, locs: [{ provider: 'b2', key: 'objs/c', state: 'pending' }] }),++ localGone: obj('d', { size: 40, locs: [{ provider: 'local', key: path.join(env.OBJECTS_PATH, 'live', 'd-gone'), state: 'present' }] }),++ localBadHash: obj('e', { size: E.length, hash: '0'.repeat(64), locs: [{ provider: 'local', key: put('e', E), state: 'present' }, { provider: 'b2', key: 'objs/e', state: 'pending' }] }),++ unverifiable: obj('f', { size: 50, locs: [{ provider: 'r2', key: 'objs/f', state: 'pending' }] }),++ corruptNoRepair: obj('g', { size: G.length, locs: [{ provider: 'local', key: put('g', G), state: 'present' }, { provider: 'b2', key: 'objs/g', state: 'pending' }] }),++ noLocations: obj('h', { size: 10 }),++};++bucket.b2['objs/c'] = 12345; // wrong size → corrupt++bucket.b2['objs/e'] = E.length; // the remote copy of e is fine++bucket.b2['objs/g'] = 999; // corrupt, and MEDIA_VERIFY_REPAIR_CORRUPT is off++obj('deleted', { size: 1, status: 'deleted', locs: [{ provider: 'local', key: '/nowhere', state: 'missing' }] });++obj('uploading', { size: 1, status: 'uploading' });++// Legacy row that points at the object whose local file is gone (the report must show it).++d.prepare("INSERT INTO vods (id, app_id, user_id, title, file_path, file_size, visibility, is_public, storage_provider, object_id) VALUES (41, 'live', 7, 'Lost stream', ?, 40, 'public', 1, 'local', ?)")++ .run(path.join(env.OBJECTS_PATH, 'live', 'd-gone'), ids.localGone);++const READY = Object.keys(ids).length;++++const loc = (id, p) => d.prepare('SELECT * FROM media_locations WHERE object_id = ? AND provider = ?').get(id, p);++const ver = (id) => d.prepare('SELECT * FROM media_verifications WHERE object_id = ?').get(id);++const snapshot = () => ({++ objects: d.prepare('SELECT COUNT(*) c FROM media_objects').get().c,++ locations: d.prepare('SELECT COUNT(*) c FROM media_locations').get().c,++ vods: d.prepare('SELECT COUNT(*) c FROM vods').get().c,++ files: fs.readdirSync(path.join(env.OBJECTS_PATH, 'live')).sort().join(','),++});++++function request(base, p) {++ return new Promise((resolve, reject) => {++ http.get(base + p, (res) => { let b = ''; res.on('data', (c) => { b += c; }); res.on('end', () => resolve({ status: res.statusCode, body: b })); }).on('error', reject);++ });++}++++(async () => {++ const before = snapshot();++ // Nothing may unlink or remove anything while the job runs.++ const deletes = [];++ const orig = { unlinkSync: fs.unlinkSync, unlink: fs.unlink, rmSync: fs.rmSync, rm: fs.rm };++ fs.unlinkSync = (...a) => { deletes.push(a[0]); return orig.unlinkSync(...a); };++ fs.unlink = (...a) => { deletes.push(a[0]); return orig.unlink(...a); };++ fs.rmSync = (...a) => { deletes.push(a[0]); return orig.rmSync(...a); };++ fs.rm = (...a) => { deletes.push(a[0]); return orig.rm(...a); };++++ // ── Bounded batches, rotating: never-verified first, then least recently verified ──++ const r1 = await job.runOnce({ batch: 3, head, upload });++ assert.strictEqual(r1.objects_checked, 3, 'a run never checks more than the batch');++ assert.strictEqual(d.prepare('SELECT COUNT(*) c FROM media_verifications').get().c, 3);++ const firstBatch = r1.objects.map(o => o.object_id);++ const r2 = await job.runOnce({ batch: 3, head, upload });++ assert.ok(r2.objects.every(o => !firstBatch.includes(o.object_id)), 'the next run takes objects not verified yet');++ const r3 = await job.runOnce({ batch: 3, head, upload });++ assert.strictEqual(d.prepare('SELECT COUNT(*) c FROM media_verifications').get().c, READY, `${READY} ready objects covered in ceil(${READY}/3) runs`);++ assert.ok(r3.objects.slice(READY - 6).every(o => firstBatch.includes(o.object_id)), 'once all are verified, the oldest verified come round again');++ assert.ok(!d.prepare("SELECT 1 FROM media_verifications v JOIN media_objects o ON o.id = v.object_id WHERE o.lifecycle_status != 'ready'").get(), 'only ready objects are verified');++ console.log('✅ bounded batches rotate through every ready object (never-verified first, then oldest)');++++ // ── Verdicts recorded ──++ assert.strictEqual(ver(ids.good).status, 'good');++ assert.strictEqual(loc(ids.good, 'local').state, 'present');++ assert.ok(loc(ids.good, 'local').verified_at);++ assert.strictEqual(loc(ids.remoteCorrupt, 'b2').state, 'corrupt');++ assert.strictEqual(loc(ids.remoteCorrupt, 'b2').size_bytes, 12345);++ assert.strictEqual(ver(ids.remoteCorrupt).status, 'no_good_copy');++ assert.strictEqual(loc(ids.localGone, 'local').state, 'missing');++ assert.strictEqual(ver(ids.localGone).status, 'no_good_copy');++ assert.strictEqual(loc(ids.localBadHash, 'local').state, 'corrupt', 'a local copy whose sha256 differs is corrupt');++ assert.strictEqual(loc(ids.localBadHash, 'b2').state, 'present');++ assert.strictEqual(ver(ids.localBadHash).good_providers, 'b2');++ assert.strictEqual(ver(ids.unverifiable).status, 'unverifiable', 'an unconfigured provider is unverifiable, never missing');++ assert.strictEqual(loc(ids.unverifiable, 'r2').state, 'pending');++ assert.strictEqual(ver(ids.noLocations).status, 'no_good_copy');++ const runs = d.prepare('SELECT * FROM media_verify_runs ORDER BY id').all();++ assert.strictEqual(runs.length, 3);++ assert.ok(runs.every(r => r.finished_at && r.objects_checked >= 1 && r.no_good_copy_total != null));++ assert.strictEqual(runs.reduce((n, r) => n + r.objects_checked, 0), 9);++ console.log('✅ verdicts recorded on media_locations, media_verifications and media_verify_runs');++++ // ── Re-upload of a missing remote copy from the good local copy ──++ assert.ok(uploads.includes('b2:objs/b'), 'missing b2 copy re-uploaded');++ assert.strictEqual(bucket.b2['objs/b'], B.length);++ assert.strictEqual(loc(ids.restore, 'b2').state, 'present');++ assert.strictEqual(JSON.parse(ver(ids.restore).detail).reuploads[0].ok, true);++ assert.ok(!uploads.includes('b2:objs/g'), 'a corrupt remote copy is not overwritten unless MEDIA_VERIFY_REPAIR_CORRUPT=1');++ assert.strictEqual(loc(ids.corruptNoRepair, 'b2').state, 'corrupt');++ assert.ok(!uploads.includes('b2:objs/e'), 'never re-uploads from a local copy that failed its hash');++ assert.strictEqual(runs.reduce((n, r) => n + r.reuploaded, 0), 1);++ // The per-run budget: with none left the restore is deferred, not attempted.++ delete bucket.b2['objs/b'];++ const deferred = await job.verifyObject(model.getObject(ids.restore), { head, upload, hashMaxBytes: 1e9, repairCorrupt: false, budget: { left: 0 } });++ assert.match(deferred.reuploads[0].error, /^deferred/);++ assert.strictEqual(loc(ids.restore, 'b2').state, 'missing');++ // Opt-in corrupt repair.++ const repaired = await job.verifyObject(model.getObject(ids.corruptNoRepair), { head, upload, hashMaxBytes: 1e9, repairCorrupt: true, budget: { left: 1 } });++ assert.strictEqual(repaired.reuploads[0].ok, true);++ assert.strictEqual(bucket.b2['objs/g'], G.length);++ bucket.b2['objs/b'] = B.length;++ console.log('✅ missing remote copy restored from a verified-good local copy (budgeted; corrupt only when opted in)');++++ // ── Never deletes ──++ fs.unlinkSync = orig.unlinkSync; fs.unlink = orig.unlink; fs.rmSync = orig.rmSync; fs.rm = orig.rm;++ assert.deepStrictEqual(deletes, [], 'no unlink/rm during verification');++ assert.deepStrictEqual(snapshot(), before, 'every object, location, legacy row and file is still there');++ assert.ok(model.getObject(ids.remoteCorrupt) && model.getObject(ids.localGone), 'objects with no good copy are kept');++ assert.strictEqual(model.getObject(ids.localGone).lifecycle_status, 'ready', 'and their lifecycle is untouched');++ console.log('✅ never deletes: files, objects, locations and legacy rows all remain');++++ // ── Count, /metrics and /api/ready ──++ assert.strictEqual(copyReport.countNoGoodCopy(db), 3, 'c (remote corrupt), d (local gone), h (no locations)');++ const st = job.status();++ assert.strictEqual(st.no_good_copy, 3);++ assert.strictEqual(st.verified_objects, READY);++ assert.ok(st.last_run.finished_at);++++ const express = require('express');++ const observability = require('../server/observability');++ const app = express();++ const inst = observability.instrument(app, { release: 'test' });++ observability.mountReady(app, inst.registry, {++ release: 'test', db, config, auth: { jwksLoaded: () => true }, recorder: { activeCount: () => 0 },++ events: { status: () => ({ enabled: false }) }, remote: { configured: () => false, probe: async () => true },++ });++ const server = await new Promise((resolve) => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); });++ const base = `http://127.0.0.1:${server.address().port}`;++ const ready = await request(base, '/api/ready');++ const body = JSON.parse(ready.body);++ assert.strictEqual(ready.status, 200, 'no good copy never fails readiness');++ assert.strictEqual(body.ready, true);++ assert.strictEqual(body.status, 'degraded');++ assert.deepStrictEqual(body.degraded, ['object_copies']);++ assert.strictEqual(body.checks.object_copies.required, false);++ assert.strictEqual(body.checks.object_copies.detail.no_good_copy, 3);++ assert.match(body.checks.object_copies.error, /^3 ready object\(s\) with no good copy/);++ const m = await request(base, '/metrics');++ assert.ok(m.body.includes('media_objects_no_good_copy 3\n'), m.body);++ assert.match(m.body, /media_verify_last_run_timestamp_seconds \d{10}\n/);++ server.close(); inst.stop();++ console.log('✅ no-good-copy count on /metrics and as a degraded, never-failing /api/ready check');++++ // ── Report script (read-only) ──++ const script = path.join(__dirname, '..', 'scripts', 'no-good-copy-report.js');++ const run = (a) => { try { return { code: 0, out: execFileSync(process.execPath, [script, ...a], { env: { ...process.env, ...env }, encoding: 'utf8' }) }; } catch (e) { return { code: e.status, out: String(e.stdout), err: String(e.stderr) }; } };++ const snapBefore = JSON.stringify(d.prepare('SELECT * FROM media_locations ORDER BY id').all()) + JSON.stringify(d.prepare('SELECT * FROM media_verifications ORDER BY object_id').all());++ const js = run(['--json', '--db', env.DB_PATH]);++ assert.strictEqual(js.code, 1, `objects found → exit 1 ${js.err || ''}`);++ const rep = JSON.parse(js.out);++ assert.strictEqual(rep.count, 3);++ assert.deepStrictEqual(rep.objects.map(o => o.object_id).sort(), [ids.remoteCorrupt, ids.localGone, ids.noLocations].sort());++ const gone = rep.objects.find(o => o.object_id === ids.localGone);++ assert.deepStrictEqual(gone.owner, { app: 'live', user_id: 7, subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ' });++ assert.strictEqual(gone.references.vods[0].id, 41);++ assert.strictEqual(gone.references.vods[0].title, 'Lost stream');++ assert.strictEqual(gone.locations[0].state, 'missing');++ assert.strictEqual(gone.last_verification.status, 'no_good_copy');++ assert.strictEqual(rep.objects.find(o => o.object_id === ids.noLocations).locations.length, 0);++ const txt = run(['--db', env.DB_PATH]);++ assert.strictEqual(txt.code, 1);++ assert.ok(txt.out.includes('3 ready object(s) with no good copy'));++ assert.ok(txt.out.includes(ids.localGone) && txt.out.includes('vods#41') && txt.out.includes('locations: (none recorded)'));++ assert.ok(run(['--db', env.DB_PATH, '--app', 'games']).out.startsWith('0 ready object(s)'), '--app filters');++ assert.strictEqual(JSON.stringify(d.prepare('SELECT * FROM media_locations ORDER BY id').all()) + JSON.stringify(d.prepare('SELECT * FROM media_verifications ORDER BY object_id').all()), snapBefore, 'the report writes nothing');++ console.log('✅ no-good-copy report: owners, legacy references, locations; text + --json; read-only');++++ // ── Scheduling: disabled by env; start/stop are idempotent ──++ config.verify.enabled = false;++ assert.strictEqual(job.start(), false);++ config.verify.enabled = true;++ assert.strictEqual(job.start({ initialDelayMs: 60 * 60 * 1000 }), true);++ job.stop();++++ db.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('objects-verify: all checks passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });@@@ -62,7 +62,7 @@ const listen = (app) => new Promise((resolve) => { const s = app.listen(0, '127.assert.strictEqual(body.service, 'media');assert.strictEqual(body.recordings_in_progress, 1);const names = Object.keys(body.checks).sort();-- assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'remote_b2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods'], 'r2 is not configured, so it has no check');++ assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods'], 'r2 is not configured, so it has no check');for (const [n, c] of Object.entries(body.checks)) {assert.strictEqual(c.status, 'ok', n);assert.strictEqual(typeof c.latency_ms, 'number');
@@ -41,10 +41,11 @@ server/
@@ -41,10 +41,11 @@ server/
thumbnails/ thumbnail service + /api/v1/:app/thumbnails
thumbnails/ thumbnail service + /api/v1/:app/thumbnails
public/routes.js public /v /c /p /t /f
public/routes.js public /v /c /p /t /f
objects/ canonical object model: model (projections, holds), routes (/api/v2 + /o),
objects/ canonical object model: model (projections, holds), routes (/api/v2 + /o),
- backfill, reconcile, invariant, signing — see docs/object-model.md
+ backfill, reconcile, invariant, signing, verify-job (scheduled copy
+ verification), copy-report — see docs/object-model.md
(openvibe-shared: pinned OpenVibe.Shared v1.0.0 release, installed by npm)
(openvibe-shared: pinned OpenVibe.Shared v1.0.0 release, installed by npm)
scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)
scripts/smoke-test.sh end-to-end smoke test (boots a temp instance)
-scripts/backfill-objects.js / reconcile-objects.js / object-invariant.js object-model operator tools
+scripts/backfill-objects.js / reconcile-objects.js / object-invariant.js / no-good-copy-report.js object-model operator tools
```
```
## Visitor sign-in
## Visitor sign-in
@@ -7,7 +7,7 @@ projection over an object through its `object_id` column. Every existing URL and
@@ -7,7 +7,7 @@ projection over an object through its `object_id` column. Every existing URL and
unchanged. New code talks to objects through the [v2 API](#object-api-v2).
unchanged. New code talks to objects through the [v2 API](#object-api-v2).
Code: `server/objects/` (`model.js`, `routes.js`, `backfill.js`, `reconcile.js`, `invariant.js`,
Code: `server/objects/` (`model.js`, `routes.js`, `backfill.js`, `reconcile.js`, `invariant.js`,
-`signing.js`). Schema: the bottom of `server/db/schema.sql`. Tests: `test/objects-*.test.js`.
+`signing.js`, `verify-job.js`, `copy-report.js`). Schema: the bottom of `server/db/schema.sql`. Tests: `test/objects-*.test.js`.
## Tables
## Tables
@@ -25,7 +25,7 @@ Location `state`:
@@ -25,7 +25,7 @@ Location `state`:
- `present`: verified. The local file exists, or a HEAD answered with the right size.
- `present`: verified. The local file exists, or a HEAD answered with the right size.
- `missing`: verified absent.
- `missing`: verified absent.
-- `pending`: believed there but not verified yet. Every remote copy starts here until reconciliation runs with `--verify`.
+- `pending`: believed there but not verified yet. Every remote copy starts here until reconciliation runs with `--verify` or the [scheduled verification](#scheduled-verification) reaches it.
- `corrupt`: the size or hash does not match.
- `corrupt`: the size or hash does not match.
A copy that reconciliation has verified keeps that state when the row is projected again. The
A copy that reconciliation has verified keeps that state when the row is projected again. The
@@ -209,6 +209,34 @@ The exit code is 1 when issues were found.
@@ -209,6 +209,34 @@ The exit code is 1 when issues were found.
listing) and incomplete multipart uploads. `server/objects/reconcile.js` takes a `head` function,
listing) and incomplete multipart uploads. `server/objects/reconcile.js` takes a `head` function,
so tests run it against a fake provider.
so tests run it against a fake provider.
+## Scheduled verification
+
+`server/objects/verify-job.js` runs inside the service (started from `server/index.js`, two minutes
+after boot, then every `MEDIA_VERIFY_INTERVAL_MIN`). Each run takes the `MEDIA_VERIFY_BATCH` ready
+objects verified least recently (never-verified first), so every object is covered over time: with
+the defaults, 7,200 objects a day.
+
+For each object it:
+
+1. **Checks every copy**, with the same rules as `reconcile-objects.js --verify`. A local copy is `present` when the file exists, and `corrupt` when the object carries a `content_hash` and the file (≤ `MEDIA_VERIFY_HASH_MAX_MB`) hashes differently. A B2/R2 copy is `missing` on a 404 and `corrupt` when its size differs. An unconfigured or unreachable provider is *unverifiable* and keeps its state. The verdict goes to `media_locations` (`state`, `size_bytes`, `verified_at`), but only if the location still has the key that was checked.
+2. **Restores a missing B2/R2 copy** by uploading a verified-good local copy (right size, hash not contradicted) to the same key through the storage engine's `uploadFile`. That call HEADs the upload afterwards and fails on a size mismatch. At most `MEDIA_VERIFY_MAX_REUPLOADS` uploads run per run, and the rest wait for the next one. A `corrupt` remote copy is overwritten only when `MEDIA_VERIFY_REPAIR_CORRUPT=1`.
+3. **Records the outcome.** `media_verifications` gets one row per object (`verified_at`, `status` `good` / `no_good_copy` / `unverifiable`, `good_providers`, per-location `detail`). `media_verify_runs` gets one summary row per run.
+
+**It never deletes anything.** No file, remote object or row is removed, and no lifecycle is changed.
+
+**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:
+
+- `media_objects_no_good_copy` on `/metrics`, next to `media_verify_last_run_timestamp_seconds`.
+- The optional `object_copies` check in `/api/ready`. It is degraded while the count is above zero and never makes Media not-ready.
+- The report, which lists each one for an operator to decide on:
+
+ ```
+ node scripts/no-good-copy-report.js [--app live] [--json] [--out report.json] [--db ./data/media.db]
+ ```
+
+ For each object it prints the owner (app, `owner_user_id`, `owner_subject`), `legacy_ref`, the `vods`/`clips`/`files`/`pastes` rows that point at it, its relationships, variants and holds, every location with its state and `verified_at`, and the last verification. The report opens the database read-only. It exits 1 when there are any.
+
## Public object-size invariant
## Public object-size invariant
This is roadmap W4 deliverable 5, stated as **policy** rather than a copied constant:
This is roadmap W4 deliverable 5, stated as **policy** rather than a copied constant:
@@ -247,6 +275,12 @@ which splits playback into objects that fit.
@@ -247,6 +275,12 @@ which splits playback into objects that fit.
| `MEDIA_SIGNED_URL_TTL_S` | 300 | default signed download lifetime |
| `MEDIA_SIGNED_URL_TTL_S` | 300 | default signed download lifetime |
| `MEDIA_UPLOAD_TOKEN_TTL_S` | 3600 | upload token lifetime |
| `MEDIA_UPLOAD_TOKEN_TTL_S` | 3600 | upload token lifetime |
| `MEDIA_PUBLIC_OBJECT_{MAX,TARGET,WARN}_MB` | 500 / 256 / 384 | invariant thresholds |
| `MEDIA_PUBLIC_OBJECT_{MAX,TARGET,WARN}_MB` | 500 / 256 / 384 | invariant thresholds |
+| `MEDIA_VERIFY_ENABLED` | on | `0` turns the scheduled verification off |
+| `MEDIA_VERIFY_INTERVAL_MIN` | 10 | minutes between verification runs |
+| `MEDIA_VERIFY_BATCH` | 50 | objects per run |
+| `MEDIA_VERIFY_HASH_MAX_MB` | 64 | sha256 local copies up to this size (objects with a `content_hash`) |
+| `MEDIA_VERIFY_MAX_REUPLOADS` | 2 | missing remote copies restored per run |
+| `MEDIA_VERIFY_REPAIR_CORRUPT` | off | `1` also overwrites a corrupt remote copy from a good local copy |
## Not in this pass
## Not in this pass
new file mode 100644
new file mode 100644
@@ -0,0 +1,37 @@
@@ -0,0 +1,37 @@
+#!/usr/bin/env node
+'use strict';
+/**
+ * List every ready object that has no good copy anywhere (docs/object-model.md#scheduled-verification),
+ * with its owner, the rows that reference it, its locations and their verified states, so an operator
+ * can decide what to do with each one. READ-ONLY: the database is opened read-only and nothing is
+ * deleted, moved or re-projected.
+ *
+ * node scripts/no-good-copy-report.js [--app live] [--json] [--out report.json] [--db ./data/media.db]
+ *
+ * Exit code 0 when there are none, 1 when some were found, 2 on error.
+ */
+const fs = require('fs');
+const path = require('path');
+
+const args = process.argv.slice(2);
+const has = (name) => args.includes(`--${name}`);
+const arg = (name, dflt) => { const i = args.indexOf(`--${name}`); return i >= 0 ? args[i + 1] : dflt; };
+
+try {
+ const dbPath = arg('db') ? path.resolve(arg('db')) : require('../server/config').db.path;
+ const Database = require('better-sqlite3');
+ const h = new Database(dbPath, { readonly: true, fileMustExist: true });
+ const q = {
+ all: (sql, params = []) => h.prepare(sql).all(...params),
+ get: (sql, params = []) => h.prepare(sql).get(...params),
+ };
+ const { buildReport, formatReport } = require('../server/objects/copy-report');
+ const report = buildReport(q, { appId: arg('app', null) });
+ h.close();
+ if (arg('out')) fs.writeFileSync(arg('out'), JSON.stringify(report, null, 2));
+ console.log(has('json') ? JSON.stringify(report, null, 2) : formatReport(report));
+ process.exit(report.count ? 1 : 0);
+} catch (err) {
+ console.error(err.message);
+ process.exit(2);
+}
@@ -57,6 +57,16 @@ const config = {
@@ -57,6 +57,16 @@ const config = {
publicWarnMb: intEnv('MEDIA_PUBLIC_OBJECT_WARN_MB', 384),
publicWarnMb: intEnv('MEDIA_PUBLIC_OBJECT_WARN_MB', 384),
},
},
+ // Scheduled copy verification (server/objects/verify-job.js; docs/object-model.md#scheduled-verification).
+ verify: {
+ enabled: !['0', 'false', 'off'].includes(String(process.env.MEDIA_VERIFY_ENABLED || '').toLowerCase()),
+ intervalMin: intEnv('MEDIA_VERIFY_INTERVAL_MIN', 10), // one batch this often
+ batch: intEnv('MEDIA_VERIFY_BATCH', 50), // objects per run (least recently verified first)
+ hashMaxMb: intEnv('MEDIA_VERIFY_HASH_MAX_MB', 64), // sha256 local copies up to this size (objects with a content_hash)
+ maxReuploads: intEnv('MEDIA_VERIFY_MAX_REUPLOADS', 2), // missing remote copies restored from a good local copy, per run
+ repairCorrupt: ['1', 'true', 'on'].includes(String(process.env.MEDIA_VERIFY_REPAIR_CORRUPT || '').toLowerCase()),
+ },
+
rtp: {
rtp: {
portMin: intEnv('RTP_PORT_MIN', 12000),
portMin: intEnv('RTP_PORT_MIN', 12000),
portMax: intEnv('RTP_PORT_MAX', 12199),
portMax: intEnv('RTP_PORT_MAX', 12199),
@@ -313,3 +313,33 @@ CREATE TABLE IF NOT EXISTS media_invariant_violations (
@@ -313,3 +313,33 @@ CREATE TABLE IF NOT EXISTS media_invariant_violations (
last_seen_at DATETIME DEFAULT CURRENT_TIMESTAMP,
last_seen_at DATETIME DEFAULT CURRENT_TIMESTAMP,
resolved_at DATETIME
resolved_at DATETIME
);
);
+
+-- Scheduled copy verification (server/objects/verify-job.js). One row per ready object: when it was
+-- last checked and what that check found; the least recently verified objects go first in each run.
+-- Verification only records and restores copies — it never deletes bytes or rows.
+CREATE TABLE IF NOT EXISTS media_verifications (
+ object_id TEXT PRIMARY KEY,
+ verified_at TEXT NOT NULL, -- ISO-8601 with ms (orders the rotation)
+ status TEXT NOT NULL CHECK(status IN ('good', 'no_good_copy', 'unverifiable')),
+ good_providers TEXT, -- comma list of providers holding a good copy
+ detail TEXT NOT NULL DEFAULT '{}', -- per-location verdicts and any re-upload
+ run_id INTEGER
+);
+CREATE INDEX IF NOT EXISTS idx_media_verifications_at ON media_verifications(verified_at);
+CREATE INDEX IF NOT EXISTS idx_media_verifications_status ON media_verifications(status);
+
+-- One row per verification run (summary).
+CREATE TABLE IF NOT EXISTS media_verify_runs (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ started_at TEXT NOT NULL,
+ finished_at TEXT,
+ objects_checked INTEGER NOT NULL DEFAULT 0,
+ locations_checked INTEGER NOT NULL DEFAULT 0,
+ good INTEGER NOT NULL DEFAULT 0,
+ no_good_copy INTEGER NOT NULL DEFAULT 0,
+ unverifiable INTEGER NOT NULL DEFAULT 0,
+ reuploaded INTEGER NOT NULL DEFAULT 0,
+ reupload_failed INTEGER NOT NULL DEFAULT 0,
+ no_good_copy_total INTEGER, -- ready objects with no good copy anywhere, after this run
+ error TEXT
+);
@@ -172,6 +172,7 @@ vodStorage.checkProviders()
@@ -172,6 +172,7 @@ vodStorage.checkProviders()
vodStorage.start(); // tiering sweep
vodStorage.start(); // tiering sweep
healthJob.start(); // health scan + quarantine cleanup
healthJob.start(); // health scan + quarantine cleanup
require('./vod/clip-jobs').start(); // failed clip re-cuts (auto-retry, hot-fetch)
require('./vod/clip-jobs').start(); // failed clip re-cuts (auto-retry, hot-fetch)
+require('./objects/verify-job').start(); // scheduled copy verification (bounded batches; never deletes)
// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.
// Descriptor watchdog: a leak here once pinned 80 GB of deleted recordings to the disk.
every(5 * 60 * 1000, () => {
every(5 * 60 * 1000, () => {
try {
try {
@@ -227,6 +228,7 @@ function shutdown(signal) {
@@ -227,6 +228,7 @@ function shutdown(signal) {
try { recorder.stopAll(); } catch { /* */ }
try { recorder.stopAll(); } catch { /* */ }
try { vodStorage.stop(); } catch { /* */ }
try { vodStorage.stop(); } catch { /* */ }
try { healthJob.stop(); } catch { /* */ }
try { healthJob.stop(); } catch { /* */ }
+ try { require('./objects/verify-job').stop(); } catch { /* */ }
try { auth.stopJwksRefresh(); } catch { /* */ }
try { auth.stopJwksRefresh(); } catch { /* */ }
try { require('./events')._reset(); } catch { /* */ }
try { require('./events')._reset(); } catch { /* */ }
for (const t of timers) clearInterval(t);
for (const t of timers) clearInterval(t);
new file mode 100644
new file mode 100644
@@ -0,0 +1,104 @@
@@ -0,0 +1,104 @@
+/**
+ * OpenVibe.Media — "no good copy" query and report (docs/object-model.md#scheduled-verification).
+ *
+ * A ready object has NO GOOD COPY when none of its locations is `present` (verified there) or
+ * `pending` (not verified yet, so possibly fine). That covers objects whose every copy was found
+ * missing/corrupt, and ready objects with no location row at all.
+ *
+ * Pure SQL over a small `q` interface — { all(sql, params), get(sql, params) } — so the service
+ * (server/db/database.js), the metrics/readiness checks and the read-only report script
+ * (scripts/no-good-copy-report.js, which opens the database file read-only) share one definition.
+ * Nothing here writes.
+ */
+'use strict';
+
+const NO_GOOD_COPY_WHERE = `o.lifecycle_status = 'ready'
+ AND NOT EXISTS (SELECT 1 FROM media_locations l WHERE l.object_id = o.id AND l.state IN ('present', 'pending'))`;
+
+function countNoGoodCopy(q, { appId = null } = {}) {
+ const r = q.get(`SELECT COUNT(*) AS n FROM media_objects o WHERE ${NO_GOOD_COPY_WHERE}${appId ? ' AND o.app_id = ?' : ''}`, appId ? [appId] : []);
+ return Number(r && r.n) || 0;
+}
+
+function listNoGoodCopy(q, { appId = null } = {}) {
+ return q.all(`SELECT o.* FROM media_objects o WHERE ${NO_GOOD_COPY_WHERE}${appId ? ' AND o.app_id = ?' : ''} ORDER BY o.app_id, o.id`, appId ? [appId] : []);
+}
+
+function tableExists(q, name) {
+ return !!q.get("SELECT 1 AS x FROM sqlite_master WHERE type = 'table' AND name = ?", [name]);
+}
+
+function parseJson(s, dflt) { try { return s ? JSON.parse(s) : dflt; } catch { return dflt; } }
+
+/** Rows elsewhere that point at the object: the inherited projections, relationships, variants, holds. */
+function references(q, obj) {
+ const refs = {
+ vods: q.all(`SELECT id, app_id, user_id, stream_id, managed_stream_id, title, visibility, is_public, file_path, file_size,
+ storage_provider, storage_key, created_at FROM vods WHERE object_id = ?`, [obj.id]),
+ clips: q.all(`SELECT id, app_id, vod_id, stream_id, user_id, channel_user_id, title, visibility, is_public, file_path, status, created_at
+ FROM clips WHERE object_id = ?`, [obj.id]),
+ files: q.all('SELECT key, app_id, user_id, original_name, size, mime, created_at FROM files WHERE object_id = ?', [obj.id]),
+ pastes: q.all('SELECT id, slug, app_id, user_id, title, type, visibility, screenshot_path, created_at FROM pastes WHERE object_id = ?', [obj.id]),
+ relationships: q.all(`SELECT from_object_id, relation, to_object_id FROM media_relationships
+ WHERE from_object_id = ? OR to_object_id = ? ORDER BY id`, [obj.id, obj.id]),
+ variants: q.all(`SELECT object_id, variant_name, derived_object_id FROM media_variants
+ WHERE object_id = ? OR derived_object_id = ? ORDER BY id`, [obj.id, obj.id]),
+ holds: q.all('SELECT id, kind, reason, created_by, created_at FROM media_holds WHERE object_id = ? AND released_at IS NULL ORDER BY id', [obj.id]),
+ };
+ return refs;
+}
+
+/** Everything an operator needs to decide what to do with each object that has no good copy. */
+function buildReport(q, { appId = null } = {}) {
+ const hasVerifications = tableExists(q, 'media_verifications');
+ const items = listNoGoodCopy(q, { appId }).map((o) => {
+ const last = hasVerifications ? q.get('SELECT verified_at, status, detail FROM media_verifications WHERE object_id = ?', [o.id]) : null;
+ const md = parseJson(o.metadata, {});
+ return {
+ object_id: o.id,
+ app_id: o.app_id,
+ kind: o.kind,
+ visibility: o.visibility,
+ title: md.title || md.filename || null,
+ size_bytes: Number(o.size_bytes) || 0,
+ content_hash: o.content_hash || null,
+ created_at: o.created_at,
+ owner: { app: o.owner_app || o.app_id, user_id: o.owner_user_id ?? null, subject: o.owner_subject || null },
+ legacy_ref: o.legacy_ref || null,
+ canonical: { provider: o.canonical_provider || null, key: o.canonical_key || null },
+ locations: q.all('SELECT provider, bucket, key, state, size_bytes, verified_at FROM media_locations WHERE object_id = ? ORDER BY id', [o.id]),
+ references: references(q, o),
+ last_verification: last ? { verified_at: last.verified_at, status: last.status, detail: parseJson(last.detail, {}) } : null,
+ };
+ });
+ return { generated_at: new Date().toISOString(), app_id: appId, count: items.length, objects: items };
+}
+
+function mb(n) { return `${((Number(n) || 0) / 1048576).toFixed(1)} MB`; }
+
+function formatReport(report) {
+ const out = [`${report.count} ready object(s) with no good copy${report.app_id ? ` in ${report.app_id}` : ''} (generated ${report.generated_at})`];
+ if (!report.count) return out.join('\n');
+ out.push('Nothing was changed. Decide per object: restore the bytes from elsewhere, or retire it through its owning app.');
+ for (const it of report.objects) {
+ out.push('');
+ out.push(`${it.object_id} ${it.kind} ${it.app_id} ${mb(it.size_bytes)} created ${it.created_at} ${it.visibility}`);
+ if (it.title) out.push(` title: ${it.title}`);
+ out.push(` owner: app=${it.owner.app} user_id=${it.owner.user_id ?? '-'} subject=${it.owner.subject || '-'}`);
+ out.push(` legacy_ref: ${it.legacy_ref || '(native object)'}`);
+ if (!it.locations.length) out.push(' locations: (none recorded)');
+ for (const l of it.locations) out.push(` location: ${l.provider}${l.bucket ? `/${l.bucket}` : ''} ${l.key} ${l.state}${l.size_bytes != null ? ` ${l.size_bytes} bytes` : ''}${l.verified_at ? ` verified ${l.verified_at}` : ''}`);
+ const r = it.references;
+ for (const v of r.vods) out.push(` vods#${v.id}: user ${v.user_id ?? '-'} "${v.title || ''}" ${v.visibility || ''} provider=${v.storage_provider || 'local'} file=${v.file_path || '-'}`);
+ for (const c of r.clips) out.push(` clips#${c.id}: vod ${c.vod_id ?? '-'} user ${c.user_id ?? '-'} "${c.title || ''}" ${c.visibility || ''} file=${c.file_path || '-'}`);
+ for (const f of r.files) out.push(` files/${f.key}: user ${f.user_id ?? '-'} ${f.original_name || ''}`);
+ for (const p of r.pastes) out.push(` pastes/${p.slug}: user ${p.user_id ?? '-'} ${p.type || ''} ${p.visibility || ''}`);
+ for (const x of r.relationships) out.push(` relationship: ${x.from_object_id} ${x.relation} ${x.to_object_id}`);
+ for (const x of r.variants) out.push(` variant: ${x.object_id} ${x.variant_name} → ${x.derived_object_id}`);
+ for (const h of r.holds) out.push(` HOLD ${h.kind}: ${h.reason || ''} (by ${h.created_by || '-'})`);
+ if (it.last_verification) out.push(` last verified ${it.last_verification.verified_at}: ${it.last_verification.status}`);
+ }
+ return out.join('\n');
+}
+
+module.exports = { NO_GOOD_COPY_WHERE, countNoGoodCopy, listNoGoodCopy, buildReport, formatReport };
new file mode 100644
new file mode 100644
@@ -0,0 +1,238 @@
@@ -0,0 +1,238 @@
+/**
+ * OpenVibe.Media — scheduled copy verification (docs/object-model.md#scheduled-verification).
+ *
+ * Every MEDIA_VERIFY_INTERVAL_MIN minutes one run takes the MEDIA_VERIFY_BATCH ready objects that
+ * were verified least recently (never-verified first), so every object is covered over time, and
+ * for each one:
+ * - checks every copy: local = the file exists (size, and sha256 when the object carries a hash
+ * and the file is ≤ MEDIA_VERIFY_HASH_MAX_MB); B2/R2 = a HEAD with the right size. The verdict
+ * (present / missing / corrupt) is written to media_locations, exactly as
+ * `reconcile-objects.js --verify` would. An unconfigured provider is unverifiable, never missing;
+ * - restores a MISSING B2/R2 copy by uploading a verified-good local copy to the same key
+ * (at most MEDIA_VERIFY_MAX_REUPLOADS per run). A corrupt remote copy is only overwritten when
+ * MEDIA_VERIFY_REPAIR_CORRUPT=1;
+ * - records the outcome in media_verifications (one row per object) and the run in media_verify_runs.
+ *
+ * It NEVER deletes anything: no file unlink, no remote delete, no row delete. Objects with no good
+ * copy anywhere are only counted (media_objects_no_good_copy on /metrics, the optional
+ * `object_copies` readiness check) and listed by scripts/no-good-copy-report.js for an operator.
+ */
+'use strict';
+
+const fs = require('fs');
+const crypto = require('crypto');
+const db = require('../db/database');
+const config = require('../config');
+const copyReport = require('./copy-report');
+
+const MB = 1024 * 1024;
+const REMOTE = ['b2', 'r2'];
+
+let _timer = null, _startTimer = null, _busy = false;
+
+const nowIso = () => new Date().toISOString();
+
+function sha256File(p) {
+ return new Promise((resolve, reject) => {
+ const h = crypto.createHash('sha256');
+ const s = fs.createReadStream(p);
+ s.on('data', d => h.update(d));
+ s.on('end', () => resolve(h.digest('hex')));
+ s.on('error', reject);
+ });
+}
+
+/** Real provider HEAD: undefined when the provider is not configured here, null on 404. */
+async function defaultHead(provider, key) {
+ return require('./reconcile').defaultHead(provider, key);
+}
+
+/** Real re-upload through the storage engine (it HEADs afterwards and throws on a size mismatch). */
+async function defaultUpload(provider, key, filePath, contentType, loc) {
+ const vs = require('../vod/vod-storage');
+ if (!vs.providerConfigured(provider)) return undefined;
+ if (loc && loc.bucket && vs.bucketFor(provider) && loc.bucket !== vs.bucketFor(provider)) {
+ throw new Error(`location is in bucket ${loc.bucket}, the configured ${provider} bucket is ${vs.bucketFor(provider)}`);
+ }
+ return vs.uploadFile(provider, key, filePath, contentType || 'application/octet-stream');
+}
+
+/** Write a verdict only if the location still points where it did when we looked (a tier move may have re-projected it). */
+function writeState(loc, state, size) {
+ db.run(`UPDATE media_locations SET state = ?, size_bytes = COALESCE(?, size_bytes), verified_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
+ WHERE id = ? AND key = ?`, [state, size ?? null, loc.id, loc.key]);
+}
+
+/**
+ * Verify one object's copies and restore missing remote ones from a good local copy.
+ * Returns { status, good_providers, locations: [...], reuploads: [...] }.
+ */
+async function verifyObject(obj, opts) {
+ const { head, upload, hashMaxBytes, repairCorrupt, budget } = opts;
+ const expected = Number(obj.size_bytes) || 0;
+ const locs = db.all('SELECT * FROM media_locations WHERE object_id = ? ORDER BY id', [obj.id]);
+ const results = [];
+ let goodLocal = null;
+
+ for (const loc of locs) {
+ const r = { provider: loc.provider, key: loc.key, before: loc.state, state: loc.state, size_bytes: null, note: null };
+ results.push(r);
+ if (loc.provider === 'local') {
+ let st = null;
+ try { const s = fs.statSync(loc.key); if (s.isFile()) st = s; } catch { /* missing */ }
+ if (!st) { r.state = 'missing'; writeState(loc, 'missing', null); continue; }
+ r.size_bytes = st.size;
+ // Same rule as reconcile: a local size that differs from the recorded size is reported, not
+ // condemned (recordings and remuxes change it) — but such a file is never used as a re-upload source.
+ let state = 'present';
+ let sizeOk = !expected || st.size === expected;
+ if (!sizeOk) r.note = `size ${st.size} ≠ expected ${expected}`;
+ let hashOk = null;
+ if (obj.content_hash && st.size <= hashMaxBytes) {
+ const actual = await sha256File(loc.key).catch(() => null);
+ if (actual) {
+ hashOk = actual === obj.content_hash;
+ if (!hashOk) { state = 'corrupt'; r.note = 'sha256 mismatch'; }
+ }
+ }
+ r.state = state;
+ writeState(loc, state, st.size);
+ if (state === 'present' && sizeOk && hashOk !== false) goodLocal = loc;
+ continue;
+ }
+ // b2 / r2
+ let h;
+ try { h = await head(loc.provider, loc.key, loc.bucket); }
+ catch (err) { r.state = loc.state; r.note = `unverifiable: ${err.message}`; r.unverifiable = true; continue; }
+ if (h === undefined) { r.state = loc.state; r.note = 'unverifiable: provider not configured'; r.unverifiable = true; continue; }
+ if (!h) { r.state = 'missing'; writeState(loc, 'missing', null); continue; }
+ r.size_bytes = Number(h.size);
+ const state = expected && Number(h.size) !== expected ? 'corrupt' : 'present';
+ if (state === 'corrupt') r.note = `size ${h.size} ≠ expected ${expected}`;
+ r.state = state;
+ writeState(loc, state, Number(h.size));
+ }
+
+ // Restore missing (and, when allowed, corrupt) remote copies from a verified-good local copy.
+ const reuploads = [];
+ if (goodLocal) {
+ for (const loc of locs) {
+ if (!REMOTE.includes(loc.provider)) continue;
+ const r = results.find(x => x.provider === loc.provider);
+ const want = r.state === 'missing' || (repairCorrupt && r.state === 'corrupt');
+ if (!want) continue;
+ if (budget.left <= 0) { reuploads.push({ provider: loc.provider, key: loc.key, ok: false, error: 'deferred: re-upload budget for this run used up' }); continue; }
+ budget.left--;
+ try {
+ const out = await upload(loc.provider, loc.key, goodLocal.key, obj.mime_type, loc);
+ if (out === undefined) { reuploads.push({ provider: loc.provider, key: loc.key, ok: false, error: 'provider not configured' }); continue; }
+ const size = Number(out && out.size) || fs.statSync(goodLocal.key).size;
+ writeState(loc, 'present', size);
+ r.state = 'present'; r.size_bytes = size; r.note = `re-uploaded from ${goodLocal.key}`;
+ reuploads.push({ provider: loc.provider, key: loc.key, ok: true, size_bytes: size, from: goodLocal.key });
+ } catch (err) {
+ reuploads.push({ provider: loc.provider, key: loc.key, ok: false, error: err.message });
+ }
+ }
+ }
+
+ const good = results.filter(x => x.state === 'present').map(x => x.provider);
+ const unknown = results.some(x => x.unverifiable || x.state === 'pending');
+ const status = good.length ? 'good' : unknown ? 'unverifiable' : 'no_good_copy';
+ return { status, good_providers: good, locations: results, reuploads };
+}
+
+/** One bounded run. Deps are injectable for tests: head(provider, key), upload(provider, key, file, mime, loc). */
+async function runOnce({ batch = config.verify.batch, head = defaultHead, upload = defaultUpload,
+ hashMaxBytes = config.verify.hashMaxMb * MB, maxReuploads = config.verify.maxReuploads, repairCorrupt = config.verify.repairCorrupt } = {}) {
+ if (_busy) return { skipped: 'busy' };
+ _busy = true;
+ const summary = { started_at: nowIso(), finished_at: null, objects_checked: 0, locations_checked: 0, good: 0, no_good_copy: 0, unverifiable: 0,
+ reuploaded: 0, reupload_failed: 0, no_good_copy_total: null, error: null, objects: [] };
+ let runId = null;
+ try {
+ runId = db.run('INSERT INTO media_verify_runs (started_at) VALUES (?)', [summary.started_at]).lastInsertRowid;
+ const n = Math.max(1, Number(batch) || 1);
+ const objects = db.all(`SELECT o.* FROM media_objects o LEFT JOIN media_verifications v ON v.object_id = o.id
+ WHERE o.lifecycle_status = 'ready'
+ ORDER BY (v.verified_at IS NOT NULL), v.verified_at, o.id LIMIT ?`, [n]);
+ const budget = { left: Math.max(0, Number(maxReuploads) || 0) };
+ for (const obj of objects) {
+ let res;
+ try { res = await verifyObject(obj, { head, upload, hashMaxBytes, repairCorrupt, budget }); }
+ catch (err) { res = { status: 'unverifiable', good_providers: [], locations: [], reuploads: [], error: err.message }; }
+ summary.objects_checked++;
+ summary.locations_checked += res.locations.length;
+ summary[res.status]++;
+ for (const u of res.reuploads) { if (u.ok) summary.reuploaded++; else if (!/^deferred/.test(u.error || '')) summary.reupload_failed++; }
+ db.run(`INSERT INTO media_verifications (object_id, verified_at, status, good_providers, detail, run_id) VALUES (?, ?, ?, ?, ?, ?)
+ ON CONFLICT(object_id) DO UPDATE SET verified_at = excluded.verified_at, status = excluded.status,
+ good_providers = excluded.good_providers, detail = excluded.detail, run_id = excluded.run_id`,
+ [obj.id, nowIso(), res.status, res.good_providers.join(',') || null,
+ JSON.stringify({ locations: res.locations, reuploads: res.reuploads, error: res.error || undefined }), runId]);
+ summary.objects.push({ object_id: obj.id, status: res.status, reuploads: res.reuploads.length });
+ if (res.status === 'no_good_copy') console.warn(`[Verify] ${obj.id} (${obj.legacy_ref || obj.kind}) has no good copy — see scripts/no-good-copy-report.js`);
+ for (const u of res.reuploads) {
+ if (u.ok) console.log(`[Verify] ${obj.id}: restored ${u.provider}:${u.key} from the local copy`);
+ else if (!/^deferred/.test(u.error || '')) console.warn(`[Verify] ${obj.id}: re-upload to ${u.provider} failed: ${u.error}`);
+ }
+ }
+ } catch (err) {
+ summary.error = err.message;
+ console.warn('[Verify] run failed:', err.message);
+ } finally {
+ summary.finished_at = nowIso();
+ try { summary.no_good_copy_total = copyReport.countNoGoodCopy(db); } catch { /* */ }
+ if (runId != null) {
+ try {
+ db.run(`UPDATE media_verify_runs SET finished_at = ?, objects_checked = ?, locations_checked = ?, good = ?, no_good_copy = ?, unverifiable = ?,
+ reuploaded = ?, reupload_failed = ?, no_good_copy_total = ?, error = ? WHERE id = ?`,
+ [summary.finished_at, summary.objects_checked, summary.locations_checked, summary.good, summary.no_good_copy, summary.unverifiable,
+ summary.reuploaded, summary.reupload_failed, summary.no_good_copy_total, summary.error, runId]);
+ } catch { /* */ }
+ }
+ summary.run_id = runId;
+ _busy = false;
+ }
+ if (summary.no_good_copy || summary.reuploaded || summary.reupload_failed) {
+ console.log(`[Verify] run ${runId}: ${summary.objects_checked} object(s), ${summary.good} good, ${summary.no_good_copy} no good copy, `
+ + `${summary.unverifiable} unverifiable, ${summary.reuploaded} restored, ${summary.reupload_failed} restore failed; `
+ + `${summary.no_good_copy_total} ready object(s) with no good copy in total`);
+ }
+ return summary;
+}
+
+/** Last finished run and the current no-good-copy count (readiness detail, operators). */
+function status() {
+ const last = db.get('SELECT * FROM media_verify_runs WHERE finished_at IS NOT NULL ORDER BY id DESC LIMIT 1');
+ return {
+ enabled: !!config.verify.enabled,
+ running: !!_timer,
+ no_good_copy: copyReport.countNoGoodCopy(db),
+ verified_objects: db.get('SELECT COUNT(*) AS n FROM media_verifications').n,
+ last_run: last ? { id: last.id, finished_at: last.finished_at, objects_checked: last.objects_checked, no_good_copy: last.no_good_copy,
+ reuploaded: last.reuploaded, reupload_failed: last.reupload_failed, error: last.error || null } : null,
+ };
+}
+
+function start({ initialDelayMs = 2 * 60 * 1000 } = {}) {
+ if (!config.verify.enabled) { console.log('[Verify] scheduled copy verification disabled (MEDIA_VERIFY_ENABLED=0)'); return false; }
+ if (_timer || _startTimer) return true;
+ const tick = () => { runOnce().catch((err) => console.warn('[Verify] run error:', err.message)); };
+ const intervalMs = Math.max(1, config.verify.intervalMin) * 60 * 1000;
+ _startTimer = setTimeout(() => {
+ _startTimer = null;
+ tick();
+ _timer = setInterval(tick, intervalMs);
+ if (_timer.unref) _timer.unref();
+ }, initialDelayMs);
+ if (_startTimer.unref) _startTimer.unref();
+ return true;
+}
+
+function stop() {
+ if (_startTimer) { clearTimeout(_startTimer); _startTimer = null; }
+ if (_timer) { clearInterval(_timer); _timer = null; }
+}
+
+module.exports = { runOnce, verifyObject, status, start, stop };
@@ -9,7 +9,10 @@
@@ -9,7 +9,10 @@
* local storage directory accepts a write. Optional (one capability degrades, the rest keeps
* local storage directory accepts a write. Optional (one capability degrades, the rest keeps
* working): the Network public key (user-JWT and service-token routes; app-key tenants are
* working): the Network public key (user-JWT and service-token routes; app-key tenants are
* unaffected), each CONFIGURED remote tier (B2 cold storage, R2 cache; a HeadBucket at most once a
* unaffected), each CONFIGURED remote tier (B2 cold storage, R2 cache; a HeadBucket at most once a
- * minute), and the Events outbox (webhooks still deliver when it backs up).
+ * minute), the Events outbox (webhooks still deliver when it backs up), and object_copies: every
+ * ready object has a good copy somewhere (server/objects/verify-job.js records it; objects with no
+ * good copy are listed by scripts/no-good-copy-report.js). object_copies never fails readiness —
+ * Media keeps serving everything else.
*
*
* `instrument()` must run before any route is mounted; `mountReady()` any time after.
* `instrument()` must run before any route is mounted; `mountReady()` any time after.
*/
*/
@@ -19,6 +22,8 @@ const crypto = require('crypto');
@@ -19,6 +22,8 @@ const crypto = require('crypto');
const metrics = require('openvibe-shared/metrics');
const metrics = require('openvibe-shared/metrics');
const { createReadiness } = require('openvibe-shared/ready');
const { createReadiness } = require('openvibe-shared/ready');
+const copyReport = require('./objects/copy-report');
+
const OUTBOX_BACKLOG_LIMIT = 1000;
const OUTBOX_BACKLOG_LIMIT = 1000;
function instrument(app, { release }) {
function instrument(app, { release }) {
@@ -59,6 +64,18 @@ function domainMetrics(registry, { db, recorder, events }) {
@@ -59,6 +64,18 @@ function domainMetrics(registry, { db, recorder, events }) {
name: 'media_objects', help: 'Media objects by lifecycle status', labelNames: ['lifecycle_status'],
name: 'media_objects', help: 'Media objects by lifecycle status', labelNames: ['lifecycle_status'],
collect: () => db.all('SELECT lifecycle_status, COUNT(*) AS n FROM media_objects GROUP BY lifecycle_status').map((r) => ({ labels: { lifecycle_status: r.lifecycle_status }, value: r.n })),
collect: () => db.all('SELECT lifecycle_status, COUNT(*) AS n FROM media_objects GROUP BY lifecycle_status').map((r) => ({ labels: { lifecycle_status: r.lifecycle_status }, value: r.n })),
});
});
+ registry.gauge({
+ name: 'media_objects_no_good_copy', help: 'Ready media objects with no present or unverified copy anywhere (see scripts/no-good-copy-report.js)',
+ collect: () => copyReport.countNoGoodCopy(db),
+ });
+ registry.gauge({
+ name: 'media_verify_last_run_timestamp_seconds', help: 'When the scheduled copy verification last finished a run',
+ collect: () => {
+ const r = db.get('SELECT finished_at FROM media_verify_runs WHERE finished_at IS NOT NULL ORDER BY id DESC LIMIT 1');
+ const t = r ? Date.parse(r.finished_at) : NaN;
+ return Number.isFinite(t) ? Math.floor(t / 1000) : null; // no run yet: no series
+ },
+ });
registry.gauge({
registry.gauge({
name: 'media_events_outbox', help: 'Events outbox rows waiting to reach OpenVibe.Events, and rows Events rejected', labelNames: ['status'],
name: 'media_events_outbox', help: 'Events outbox rows waiting to reach OpenVibe.Events, and rows Events rejected', labelNames: ['status'],
collect: () => {
collect: () => {
@@ -90,6 +107,16 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, rem
@@ -90,6 +107,16 @@ function createMediaReadiness({ release, db, config, auth, recorder, events, rem
return { ok: true, detail: { pending: s.pending, rejected: s.rejected } };
return { ok: true, detail: { pending: s.pending, rejected: s.rejected } };
},
},
});
});
+ checks.push({
+ name: 'object_copies', required: false, cacheMs: 60 * 1000, description: 'every ready object has a good copy (scheduled verification)',
+ check: () => {
+ const n = copyReport.countNoGoodCopy(db);
+ const last = db.get('SELECT id, finished_at, objects_checked, reuploaded FROM media_verify_runs WHERE finished_at IS NOT NULL ORDER BY id DESC LIMIT 1');
+ const detail = { no_good_copy: n, last_verify_run: last ? last.finished_at : null };
+ if (n) return { ok: false, error: `${n} ready object(s) with no good copy (scripts/no-good-copy-report.js)`, detail };
+ return { ok: true, detail };
+ },
+ });
return createReadiness({
return createReadiness({
service: 'media',
service: 'media',
release,
release,
new file mode 100644
new file mode 100644
@@ -0,0 +1,223 @@
@@ -0,0 +1,223 @@
+'use strict';
+// Scheduled copy verification (server/objects/verify-job.js) against a fake B2/R2 provider:
+// bounded batches that rotate through every ready object, verdicts recorded on media_locations,
+// media_verifications and media_verify_runs, a missing remote copy restored from a good local copy,
+// nothing ever deleted, the no-good-copy count on /metrics and as a degraded (still ready) check in
+// /api/ready, and scripts/no-good-copy-report.js listing those objects read-only.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const http = require('http');
+const crypto = require('crypto');
+const { execFileSync } = require('child_process');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-verify-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+const 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'), ASSETS_PATH: dir('assets'),
+ MEDIA_PUBLIC_URL: 'https://media.test', OV_NETWORK_URL: 'http://127.0.0.1:9',
+};
+for (const k of ['MEDIA_B2_ENDPOINT', 'MEDIA_B2_BUCKET', 'MEDIA_R2_ENDPOINT', 'MEDIA_R2_BUCKET', 'EVENTS_URL', 'MEDIA_VERIFY_ENABLED', 'MEDIA_VERIFY_REPAIR_CORRUPT']) process.env[k] = '';
+Object.assign(process.env, env);
+
+const db = require('../server/db/database');
+const config = require('../server/config');
+const model = require('../server/objects/model');
+const job = require('../server/objects/verify-job');
+const copyReport = require('../server/objects/copy-report');
+
+const d = db.getDb();
+const sha = (b) => crypto.createHash('sha256').update(b).digest('hex');
+const put = (name, content) => { const p = path.join(env.OBJECTS_PATH, 'live', name); fs.mkdirSync(path.dirname(p), { recursive: true }); fs.writeFileSync(p, content); return p; };
+function obj(name, { size, hash = null, status = 'ready', locs = [], owner = 7 }) {
+ const id = model.createObject({ app_id: 'live', kind: 'file', lifecycle_status: status, size_bytes: size, content_hash: hash,
+ owner_user_id: owner, owner_subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ', mime_type: 'application/octet-stream', metadata: { filename: name } });
+ for (const l of locs) model.upsertLocation(id, l);
+ return id;
+}
+
+// Fake provider: bucket[provider][key] = size (absent → 404). r2 is "not configured" (HEAD undefined).
+const bucket = { b2: {} };
+const heads = [], uploads = [];
+const head = async (provider, key) => { heads.push(`${provider}:${key}`); if (!bucket[provider]) return undefined; const s = bucket[provider][key]; return s === undefined ? null : { size: s }; };
+const upload = async (provider, key, file) => { uploads.push(`${provider}:${key}`); if (!bucket[provider]) return undefined; const size = fs.statSync(file).size; bucket[provider][key] = size; return { size }; };
+
+const A = Buffer.from('aaaaaaaaaa'), B = Buffer.from('bbbbbbbbbbbb'), E = Buffer.from('eeee'), G = Buffer.from('gggggg');
+const ids = {
+ good: obj('a', { size: A.length, hash: sha(A), locs: [{ provider: 'local', key: put('a', A), state: 'present' }] }),
+ restore: obj('b', { size: B.length, locs: [{ provider: 'local', key: put('b', B), state: 'present' }, { provider: 'b2', key: 'objs/b', state: 'pending' }] }),
+ remoteCorrupt: obj('c', { size: 300, locs: [{ provider: 'b2', key: 'objs/c', state: 'pending' }] }),
+ localGone: obj('d', { size: 40, locs: [{ provider: 'local', key: path.join(env.OBJECTS_PATH, 'live', 'd-gone'), state: 'present' }] }),
+ localBadHash: obj('e', { size: E.length, hash: '0'.repeat(64), locs: [{ provider: 'local', key: put('e', E), state: 'present' }, { provider: 'b2', key: 'objs/e', state: 'pending' }] }),
+ unverifiable: obj('f', { size: 50, locs: [{ provider: 'r2', key: 'objs/f', state: 'pending' }] }),
+ corruptNoRepair: obj('g', { size: G.length, locs: [{ provider: 'local', key: put('g', G), state: 'present' }, { provider: 'b2', key: 'objs/g', state: 'pending' }] }),
+ noLocations: obj('h', { size: 10 }),
+};
+bucket.b2['objs/c'] = 12345; // wrong size → corrupt
+bucket.b2['objs/e'] = E.length; // the remote copy of e is fine
+bucket.b2['objs/g'] = 999; // corrupt, and MEDIA_VERIFY_REPAIR_CORRUPT is off
+obj('deleted', { size: 1, status: 'deleted', locs: [{ provider: 'local', key: '/nowhere', state: 'missing' }] });
+obj('uploading', { size: 1, status: 'uploading' });
+// Legacy row that points at the object whose local file is gone (the report must show it).
+d.prepare("INSERT INTO vods (id, app_id, user_id, title, file_path, file_size, visibility, is_public, storage_provider, object_id) VALUES (41, 'live', 7, 'Lost stream', ?, 40, 'public', 1, 'local', ?)")
+ .run(path.join(env.OBJECTS_PATH, 'live', 'd-gone'), ids.localGone);
+const READY = Object.keys(ids).length;
+
+const loc = (id, p) => d.prepare('SELECT * FROM media_locations WHERE object_id = ? AND provider = ?').get(id, p);
+const ver = (id) => d.prepare('SELECT * FROM media_verifications WHERE object_id = ?').get(id);
+const snapshot = () => ({
+ objects: d.prepare('SELECT COUNT(*) c FROM media_objects').get().c,
+ locations: d.prepare('SELECT COUNT(*) c FROM media_locations').get().c,
+ vods: d.prepare('SELECT COUNT(*) c FROM vods').get().c,
+ files: fs.readdirSync(path.join(env.OBJECTS_PATH, 'live')).sort().join(','),
+});
+
+function request(base, p) {
+ return new Promise((resolve, reject) => {
+ http.get(base + p, (res) => { let b = ''; res.on('data', (c) => { b += c; }); res.on('end', () => resolve({ status: res.statusCode, body: b })); }).on('error', reject);
+ });
+}
+
+(async () => {
+ const before = snapshot();
+ // Nothing may unlink or remove anything while the job runs.
+ const deletes = [];
+ const orig = { unlinkSync: fs.unlinkSync, unlink: fs.unlink, rmSync: fs.rmSync, rm: fs.rm };
+ fs.unlinkSync = (...a) => { deletes.push(a[0]); return orig.unlinkSync(...a); };
+ fs.unlink = (...a) => { deletes.push(a[0]); return orig.unlink(...a); };
+ fs.rmSync = (...a) => { deletes.push(a[0]); return orig.rmSync(...a); };
+ fs.rm = (...a) => { deletes.push(a[0]); return orig.rm(...a); };
+
+ // ── Bounded batches, rotating: never-verified first, then least recently verified ──
+ const r1 = await job.runOnce({ batch: 3, head, upload });
+ assert.strictEqual(r1.objects_checked, 3, 'a run never checks more than the batch');
+ assert.strictEqual(d.prepare('SELECT COUNT(*) c FROM media_verifications').get().c, 3);
+ const firstBatch = r1.objects.map(o => o.object_id);
+ const r2 = await job.runOnce({ batch: 3, head, upload });
+ assert.ok(r2.objects.every(o => !firstBatch.includes(o.object_id)), 'the next run takes objects not verified yet');
+ const r3 = await job.runOnce({ batch: 3, head, upload });
+ assert.strictEqual(d.prepare('SELECT COUNT(*) c FROM media_verifications').get().c, READY, `${READY} ready objects covered in ceil(${READY}/3) runs`);
+ assert.ok(r3.objects.slice(READY - 6).every(o => firstBatch.includes(o.object_id)), 'once all are verified, the oldest verified come round again');
+ assert.ok(!d.prepare("SELECT 1 FROM media_verifications v JOIN media_objects o ON o.id = v.object_id WHERE o.lifecycle_status != 'ready'").get(), 'only ready objects are verified');
+ console.log('✅ bounded batches rotate through every ready object (never-verified first, then oldest)');
+
+ // ── Verdicts recorded ──
+ assert.strictEqual(ver(ids.good).status, 'good');
+ assert.strictEqual(loc(ids.good, 'local').state, 'present');
+ assert.ok(loc(ids.good, 'local').verified_at);
+ assert.strictEqual(loc(ids.remoteCorrupt, 'b2').state, 'corrupt');
+ assert.strictEqual(loc(ids.remoteCorrupt, 'b2').size_bytes, 12345);
+ assert.strictEqual(ver(ids.remoteCorrupt).status, 'no_good_copy');
+ assert.strictEqual(loc(ids.localGone, 'local').state, 'missing');
+ assert.strictEqual(ver(ids.localGone).status, 'no_good_copy');
+ assert.strictEqual(loc(ids.localBadHash, 'local').state, 'corrupt', 'a local copy whose sha256 differs is corrupt');
+ assert.strictEqual(loc(ids.localBadHash, 'b2').state, 'present');
+ assert.strictEqual(ver(ids.localBadHash).good_providers, 'b2');
+ assert.strictEqual(ver(ids.unverifiable).status, 'unverifiable', 'an unconfigured provider is unverifiable, never missing');
+ assert.strictEqual(loc(ids.unverifiable, 'r2').state, 'pending');
+ assert.strictEqual(ver(ids.noLocations).status, 'no_good_copy');
+ const runs = d.prepare('SELECT * FROM media_verify_runs ORDER BY id').all();
+ assert.strictEqual(runs.length, 3);
+ assert.ok(runs.every(r => r.finished_at && r.objects_checked >= 1 && r.no_good_copy_total != null));
+ assert.strictEqual(runs.reduce((n, r) => n + r.objects_checked, 0), 9);
+ console.log('✅ verdicts recorded on media_locations, media_verifications and media_verify_runs');
+
+ // ── Re-upload of a missing remote copy from the good local copy ──
+ assert.ok(uploads.includes('b2:objs/b'), 'missing b2 copy re-uploaded');
+ assert.strictEqual(bucket.b2['objs/b'], B.length);
+ assert.strictEqual(loc(ids.restore, 'b2').state, 'present');
+ assert.strictEqual(JSON.parse(ver(ids.restore).detail).reuploads[0].ok, true);
+ assert.ok(!uploads.includes('b2:objs/g'), 'a corrupt remote copy is not overwritten unless MEDIA_VERIFY_REPAIR_CORRUPT=1');
+ assert.strictEqual(loc(ids.corruptNoRepair, 'b2').state, 'corrupt');
+ assert.ok(!uploads.includes('b2:objs/e'), 'never re-uploads from a local copy that failed its hash');
+ assert.strictEqual(runs.reduce((n, r) => n + r.reuploaded, 0), 1);
+ // The per-run budget: with none left the restore is deferred, not attempted.
+ delete bucket.b2['objs/b'];
+ const deferred = await job.verifyObject(model.getObject(ids.restore), { head, upload, hashMaxBytes: 1e9, repairCorrupt: false, budget: { left: 0 } });
+ assert.match(deferred.reuploads[0].error, /^deferred/);
+ assert.strictEqual(loc(ids.restore, 'b2').state, 'missing');
+ // Opt-in corrupt repair.
+ const repaired = await job.verifyObject(model.getObject(ids.corruptNoRepair), { head, upload, hashMaxBytes: 1e9, repairCorrupt: true, budget: { left: 1 } });
+ assert.strictEqual(repaired.reuploads[0].ok, true);
+ assert.strictEqual(bucket.b2['objs/g'], G.length);
+ bucket.b2['objs/b'] = B.length;
+ console.log('✅ missing remote copy restored from a verified-good local copy (budgeted; corrupt only when opted in)');
+
+ // ── Never deletes ──
+ fs.unlinkSync = orig.unlinkSync; fs.unlink = orig.unlink; fs.rmSync = orig.rmSync; fs.rm = orig.rm;
+ assert.deepStrictEqual(deletes, [], 'no unlink/rm during verification');
+ assert.deepStrictEqual(snapshot(), before, 'every object, location, legacy row and file is still there');
+ assert.ok(model.getObject(ids.remoteCorrupt) && model.getObject(ids.localGone), 'objects with no good copy are kept');
+ assert.strictEqual(model.getObject(ids.localGone).lifecycle_status, 'ready', 'and their lifecycle is untouched');
+ console.log('✅ never deletes: files, objects, locations and legacy rows all remain');
+
+ // ── Count, /metrics and /api/ready ──
+ assert.strictEqual(copyReport.countNoGoodCopy(db), 3, 'c (remote corrupt), d (local gone), h (no locations)');
+ const st = job.status();
+ assert.strictEqual(st.no_good_copy, 3);
+ assert.strictEqual(st.verified_objects, READY);
+ assert.ok(st.last_run.finished_at);
+
+ const express = require('express');
+ const observability = require('../server/observability');
+ const app = express();
+ const inst = observability.instrument(app, { release: 'test' });
+ observability.mountReady(app, inst.registry, {
+ release: 'test', db, config, auth: { jwksLoaded: () => true }, recorder: { activeCount: () => 0 },
+ events: { status: () => ({ enabled: false }) }, remote: { configured: () => false, probe: async () => true },
+ });
+ const server = await new Promise((resolve) => { const s = app.listen(0, '127.0.0.1', () => resolve(s)); });
+ const base = `http://127.0.0.1:${server.address().port}`;
+ const ready = await request(base, '/api/ready');
+ const body = JSON.parse(ready.body);
+ assert.strictEqual(ready.status, 200, 'no good copy never fails readiness');
+ assert.strictEqual(body.ready, true);
+ assert.strictEqual(body.status, 'degraded');
+ assert.deepStrictEqual(body.degraded, ['object_copies']);
+ assert.strictEqual(body.checks.object_copies.required, false);
+ assert.strictEqual(body.checks.object_copies.detail.no_good_copy, 3);
+ assert.match(body.checks.object_copies.error, /^3 ready object\(s\) with no good copy/);
+ const m = await request(base, '/metrics');
+ assert.ok(m.body.includes('media_objects_no_good_copy 3\n'), m.body);
+ assert.match(m.body, /media_verify_last_run_timestamp_seconds \d{10}\n/);
+ server.close(); inst.stop();
+ console.log('✅ no-good-copy count on /metrics and as a degraded, never-failing /api/ready check');
+
+ // ── Report script (read-only) ──
+ const script = path.join(__dirname, '..', 'scripts', 'no-good-copy-report.js');
+ const run = (a) => { try { return { code: 0, out: execFileSync(process.execPath, [script, ...a], { env: { ...process.env, ...env }, encoding: 'utf8' }) }; } catch (e) { return { code: e.status, out: String(e.stdout), err: String(e.stderr) }; } };
+ const snapBefore = JSON.stringify(d.prepare('SELECT * FROM media_locations ORDER BY id').all()) + JSON.stringify(d.prepare('SELECT * FROM media_verifications ORDER BY object_id').all());
+ const js = run(['--json', '--db', env.DB_PATH]);
+ assert.strictEqual(js.code, 1, `objects found → exit 1 ${js.err || ''}`);
+ const rep = JSON.parse(js.out);
+ assert.strictEqual(rep.count, 3);
+ assert.deepStrictEqual(rep.objects.map(o => o.object_id).sort(), [ids.remoteCorrupt, ids.localGone, ids.noLocations].sort());
+ const gone = rep.objects.find(o => o.object_id === ids.localGone);
+ assert.deepStrictEqual(gone.owner, { app: 'live', user_id: 7, subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ' });
+ assert.strictEqual(gone.references.vods[0].id, 41);
+ assert.strictEqual(gone.references.vods[0].title, 'Lost stream');
+ assert.strictEqual(gone.locations[0].state, 'missing');
+ assert.strictEqual(gone.last_verification.status, 'no_good_copy');
+ assert.strictEqual(rep.objects.find(o => o.object_id === ids.noLocations).locations.length, 0);
+ const txt = run(['--db', env.DB_PATH]);
+ assert.strictEqual(txt.code, 1);
+ assert.ok(txt.out.includes('3 ready object(s) with no good copy'));
+ assert.ok(txt.out.includes(ids.localGone) && txt.out.includes('vods#41') && txt.out.includes('locations: (none recorded)'));
+ assert.ok(run(['--db', env.DB_PATH, '--app', 'games']).out.startsWith('0 ready object(s)'), '--app filters');
+ assert.strictEqual(JSON.stringify(d.prepare('SELECT * FROM media_locations ORDER BY id').all()) + JSON.stringify(d.prepare('SELECT * FROM media_verifications ORDER BY object_id').all()), snapBefore, 'the report writes nothing');
+ console.log('✅ no-good-copy report: owners, legacy references, locations; text + --json; read-only');
+
+ // ── Scheduling: disabled by env; start/stop are idempotent ──
+ config.verify.enabled = false;
+ assert.strictEqual(job.start(), false);
+ config.verify.enabled = true;
+ assert.strictEqual(job.start({ initialDelayMs: 60 * 60 * 1000 }), true);
+ job.stop();
+
+ db.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('objects-verify: all checks passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });
@@ -62,7 +62,7 @@ const listen = (app) => new Promise((resolve) => { const s = app.listen(0, '127.
@@ -62,7 +62,7 @@ const listen = (app) => new Promise((resolve) => { const s = app.listen(0, '127.
assert.strictEqual(body.service, 'media');
assert.strictEqual(body.service, 'media');
assert.strictEqual(body.recordings_in_progress, 1);
assert.strictEqual(body.recordings_in_progress, 1);
const names = Object.keys(body.checks).sort();
const names = Object.keys(body.checks).sort();
- assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'remote_b2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods'], 'r2 is not configured, so it has no check');
+ assert.deepStrictEqual(names, ['db', 'events_outbox', 'network_jwks', 'object_copies', 'remote_b2', 'storage_clips', 'storage_files', 'storage_objects', 'storage_pastes', 'storage_thumbnails', 'storage_vods'], 'r2 is not configured, so it has no check');
for (const [n, c] of Object.entries(body.checks)) {
for (const [n, c] of Object.entries(body.checks)) {
assert.strictEqual(c.status, 'ok', n);
assert.strictEqual(c.status, 'ok', n);
assert.strictEqual(typeof c.latency_ms, 'number');
assert.strictEqual(typeof c.latency_ms, 'number');