openvibers/openvibe.media Commits

Backfill missing VOD thumbnails in due-based batches (#394)

OP OpenVibers committed 3 days ago · 32eabf3e5b · 6 files, +244 −11

* Backfill missing VOD thumbnails in due-based batches

* Backfill log only when it queues a picture

* Backfill walks the whole VOD library a window at a time

* Advance the backfill cursor through the whole library

Changes

ddiff --git a/server/config.js b/server/config.js
iindex 98b5615..9b6f4d4 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -134,6 +134,11 @@ const config = {
leaseS: Math.max(30, intEnv('MEDIA_JOBS_LEASE_S', 120)),
// The size-invariant validator runs this often (hours; 0 = only on demand). It proposes jobs, never runs them.
invariantScanHours: intEnv('MEDIA_INVARIANT_SCAN_HOURS', 24),
++ // The missing-picture backfill (server/jobs/thumbnail.js sweepMissing): a small batch this often
++ // (minutes; 0 = never). It queues the existing thumbnail.regenerate jobs for stored VODs whose
++ // picture is missing or gone, so the home's replay rows stop showing a grey 1×1.
++ thumbnailBackfillMin: intEnv('MEDIA_THUMBNAIL_BACKFILL_MIN', 2),
++ thumbnailBackfillBatch: Math.max(1, Math.min(20, intEnv('MEDIA_THUMBNAIL_BACKFILL_BATCH', 10))),
retentionDays: intEnv('MEDIA_JOBS_RETENTION_DAYS', 30), // finished thumbnail jobs are pruned after this
},
ddiff --git a/server/db/database.js b/server/db/database.js
iindex c945395..0f9d3d2 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -303,23 +303,27 @@ async function listVods(appId, filters = {}) {
return await all(`SELECT * FROM vods WHERE ${conds.join(' AND ')} ORDER BY ${_listOrder(order)} LIMIT ? OFFSET ?`, params);
}
--/** Latest finished public VOD (id + thumbnail) per managed stream — one batch query. */
++/**
++ * Latest finished public VOD (id + thumbnail + whether it is audio-only) per managed stream — one batch
++ * query. The pick is deterministic (newest created_at, then newest id); two rows of one stream can share
++ * a created_at, and the old join on created_at alone let either answer. `audio_only` lets the card say
++ * "no picture to show" instead of naming a 1×1 fallback for a broadcast with no camera.
++ */
async function latestVodThumbsByManagedStreams(appId, managedStreamIds) {
const ids = (managedStreamIds || []).map(n => parseInt(n, 10)).filter(Number.isFinite);
if (!ids.length) return {};
const ph = ids.map(() => '?').join(',');
const rows = await all(`
-- SELECT v.managed_stream_id, v.id, v.thumbnail_url
++ SELECT DISTINCT ON (v.managed_stream_id) v.managed_stream_id, v.id, v.thumbnail_url, v.audio_only
FROM vods v
-- JOIN (SELECT managed_stream_id ms, MAX(created_at) mc FROM vods
-- WHERE app_id = ? AND managed_stream_id IN (${ph})
-- AND is_public = 1 AND COALESCE(is_recording, 0) = 0 AND COALESCE(clips_only, 0) = 0
-- GROUP BY managed_stream_id) latest
-- ON v.managed_stream_id = latest.ms AND v.created_at = latest.mc
-- WHERE v.app_id = ? AND v.is_public = 1 AND COALESCE(v.is_recording, 0) = 0`,
-- [appId, ...ids, appId]);
++ WHERE v.app_id = ? AND v.managed_stream_id IN (${ph})
++ AND v.is_public = 1 AND COALESCE(v.is_recording, 0) = 0 AND COALESCE(v.clips_only, 0) = 0
++ ORDER BY v.managed_stream_id, v.created_at DESC NULLS LAST, v.id DESC`,
++ [appId, ...ids]);
const out = {};
-- for (const r of rows) { if (out[r.managed_stream_id] == null) out[r.managed_stream_id] = { vod_id: r.id, thumbnail_url: r.thumbnail_url }; }
++ for (const r of rows) {
++ out[r.managed_stream_id] = { vod_id: r.id, thumbnail_url: r.thumbnail_url, audio_only: !!r.audio_only };
++ }
return out;
}
ddiff --git a/server/jobs/thumbnail.js b/server/jobs/thumbnail.js
iindex 8430185..65e555d 100644
---- a/server/jobs/thumbnail.js
++++ b/server/jobs/thumbnail.js
@@@ -16,7 +16,10 @@
*/
'use strict';
++const fs = require('fs');
++const path = require('path');
const db = require('../db/database');
++const config = require('../config');
const { JobError } = require('./queue');
function targetOf(obj, params = {}) {
@@@ -125,9 +128,112 @@ async function failedBefore(appId, kind, id) {
return Number((r && r.n) || 0);
}
++// ── Missing-picture backfill (due based, small batches) ──────
++// Most pictures are made at finalize (server/vod/finalize.js). One whose generation failed there, whose
++// files were lost, or whose row was made before thumbnails existed had no picture forever: the page then
++// showed Media's 1×1 fallback (the owner's empty grey boxes, 2026-11-10). This sweep asks the existing
++// thumbnail.regenerate job for a few of them at a time, never retrying one too soon, and logs its count.
++const TYPE = 'thumbnail.regenerate';
++const MIN_AGE_MS = 10 * 60 * 1000; // a just-finalized recording may still be working on its picture
++const RETRY_FAILED_MS = 6 * 3600 * 1000; // ffmpeg produced no frame; try again in six hours
++const RETRY_UNAVAILABLE_MS = 24 * 3600 * 1000; // the file is gone/empty/too small; once a day is enough
++const RETRY_SUCCEEDED_MS = 60 * 60 * 1000; // the row had a picture but its file is missing again
++const CANDIDATE_WINDOW = 400; // rows looked at per sweep per tenant
++const cursorKey = (appId) => `thumbnail_backfill.cursor.${appId}`;
++
++function dbTimeMs(value) {
++ if (!value) return NaN;
++ const t = String(value);
++ const ms = Date.parse(/Z$|[+-]\d\d:?\d\d$/.test(t) ? t : t.replace(' ', 'T') + 'Z');
++ return Number.isFinite(ms) ? ms : NaN;
++}
++
++/** The stored thumbnail's file is on disk: a URL whose file is gone is not a picture (it answers 1×1). */
++function hasPicture(row) {
++ const url = String((row && row.thumbnail_url) || '');
++ if (!url) return false;
++ let name;
++ try { name = path.basename(new URL(url, 'http://media.invalid').pathname); } catch { name = path.basename(url); }
++ if (!name) return false;
++ try { return fs.existsSync(path.join(require('../thumbnails/thumbnail-service').THUMB_DIR, name)); } catch { return false; }
++}
++
++/** This VOD's latest thumbnail job (any status), for the due time. */
++async function lastAttempt(appId, id) {
++ return await db.get(`SELECT status, error_code, created_at FROM media_jobs
++ WHERE app_id = ? AND job_type = ? AND params::jsonb ->> 'kind' = 'vod' AND params::jsonb ->> 'id' = ?
++ ORDER BY created_at DESC, id DESC LIMIT 1`, [appId, TYPE, String(id)]);
++}
++
++/** The earliest moment the next attempt may run; Infinity while one is queued or running. */
++function dueAt(last) {
++ if (!last) return 0;
++ if (last.status === 'queued' || last.status === 'running') return Infinity;
++ const at = dbTimeMs(last.created_at);
++ if (!Number.isFinite(at)) return 0;
++ if (last.status === 'failed') {
++ return at + (last.error_code === 'media_unavailable' || last.error_code === 'no_video' ? RETRY_UNAVAILABLE_MS : RETRY_FAILED_MS);
++ }
++ return at + RETRY_SUCCEEDED_MS;
++}
++
++/**
++ * Queue one small batch of thumbnail.regenerate jobs for public, finished VODs whose stored picture is
++ * missing or gone. A VOD is due when no job is active and enough time has passed since its last attempt.
++ * → { missing, queued }
++ */
++async function sweepMissing({ now = Date.now(), limit = null, appId = null, minAgeMs = MIN_AGE_MS } = {}) {
++ const batch = Math.max(1, Math.min(20, Number(limit) || config.jobs.thumbnailBackfillBatch || 3));
++ const apps = appId ? [{ app_id: appId }]
++ : await db.all(`SELECT DISTINCT app_id FROM vods WHERE is_public = 1 AND COALESCE(is_recording, 0) = 0 AND COALESCE(clips_only, 0) = 0`);
++ let missing = 0, queued = 0;
++ for (const { app_id: app } of apps) {
++ // Walk the whole library a window at a time (newest first, then older), not just the newest rows:
++ // an old recording with a lost picture is a stored VOD too. The cursor wraps when the walk ends.
++ const cursor = Number(await db.getSetting(cursorKey(app))) || 0;
++ const rows = await db.all(`SELECT id, created_at, thumbnail_url FROM vods
++ WHERE app_id = ? AND is_public = 1
++ AND COALESCE(is_recording, 0) = 0 AND COALESCE(clips_only, 0) = 0
++ AND (file_path IS NOT NULL OR object_id IS NOT NULL)
++ AND COALESCE(health_status, '') NOT IN ('missing_file', 'zero_byte')
++ AND id < ?
++ ORDER BY id DESC LIMIT ?`, [app, cursor || '9223372036854775807', CANDIDATE_WINDOW]);
++ let appMissing = 0, appQueued = 0, examined = 0, lastId = 0;
++ for (const row of rows) {
++ examined++;
++ lastId = Number(row.id);
++ if (hasPicture(row)) continue;
++ const created = dbTimeMs(row.created_at);
++ if (Number.isFinite(created) && now - created < minAgeMs) continue; // finalize may still make it
++ appMissing++;
++ if (now < dueAt(await lastAttempt(app, row.id))) continue;
++ try {
++ const out = await require('./queue').enqueue({
++ appId: app, type: TYPE, params: { kind: 'vod', id: row.id },
++ dedupeActive: true, createdBy: 'system:thumbnail-backfill',
++ });
++ if (out && out.created && appQueued < batch) appQueued++;
++ if (appQueued >= batch) break;
++ } catch (err) {
++ console.warn(`[Thumbnails] backfill could not queue vod ${row.id}: ${err.message}`);
++ }
++ }
++ // A stopped batch resumes at the row after the one it stopped on; a finished walk wraps to the top.
++ const done = examined === rows.length && rows.length < CANDIDATE_WINDOW;
++ await db.setSetting(cursorKey(app), done ? 0 : lastId, 'number');
++ if (appQueued) console.log(`[Thumbnails] backfill: queued ${appQueued} thumbnail job(s); ${appMissing} recording(s) on ${app} have no usable picture`);
++ missing += appMissing;
++ queued += appQueued;
++ }
++ return { missing, queued };
++}
++
module.exports = {
spec: { lane: 'light', maxAttempts: 2, timeoutMs: 2 * 60 * 1000, needsObject: false, backoffS: () => 20, validate, run },
targetOf,
failedBefore,
GIVE_UP_AFTER,
++ sweepMissing,
++ hasPicture,
++ dueAt,
};
ddiff --git a/server/jobs/worker.js b/server/jobs/worker.js
iindex edecc20..b43feef 100644
---- a/server/jobs/worker.js
++++ b/server/jobs/worker.js
@@@ -23,6 +23,8 @@
* get a vod.finalize every MEDIA_FINALIZE_SWEEP_S (server/jobs/vod-finalize.js). The storage orphan
* report (storage.orphans.scan, report only) runs every MEDIA_ORPHAN_SCAN_DAYS under queue.SYSTEM_APP;
* storage.move.cleanup (alerts on placement moves that failed three times) every MEDIA_MOVE_CLEANUP_MINUTES.
++ * Stored VODs with no usable picture get a thumbnail.regenerate a few at a time every
++ * MEDIA_THUMBNAIL_BACKFILL_MIN (server/jobs/thumbnail.js sweepMissing; 0 = never).
*/
'use strict';
@@@ -46,6 +48,7 @@ async function sweepWorkDirs() {
} catch (err) { console.warn('[Jobs] work directory sweep:', err.message); }
}
let lastPruneAt = 0;
++let lastThumbBackfillAt = 0;
let kicked = false;
function lanes() {
@@@ -172,6 +175,14 @@ async function tick() {
try { await require('./recording-cleanup').schedule(now); } catch (err) { console.warn('[Jobs] recording clean-up schedule:', err.message); }
}
try { await require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }
++ // Stored VODs with no usable picture: a few thumbnail.regenerate jobs at a time (due based, so a
++ // broken file is not retried every sweep). Runs from the first tick after boot.
++ const backfillMs = cfg().thumbnailBackfillMin * 60 * 1000;
++ if (backfillMs > 0 && now - lastThumbBackfillAt > backfillMs) {
++ lastThumbBackfillAt = now;
++ try { await require('./thumbnail').sweepMissing({ now, limit: cfg().thumbnailBackfillBatch }); }
++ catch (err) { console.warn('[Jobs] thumbnail backfill:', err.message); }
++ }
if (now - lastSweepAt > 3600 * 1000) {
lastSweepAt = now;
await sweepWorkDirs();
ddiff --git a/server/vod/routes.js b/server/vod/routes.js
iindex 5e732d6..40e0b3a 100644
---- a/server/vod/routes.js
++++ b/server/vod/routes.js
@@@ -324,7 +324,8 @@ router.get('/', tenantAuth({ verb: 'list', allowUser: true }), async (req, res)
});
// ── Latest VOD thumbnail per managed stream (batch) ──────────
--// GET /latest-thumbs?managed_stream_ids=1,2,3 → { "<msid>": { vod_id, thumbnail_url } }
++// GET /latest-thumbs?managed_stream_ids=1,2,3 → { "<msid>": { vod_id, thumbnail_url, audio_only } }
++// audio_only: the recording has no frame to show (a camera-less broadcast); its picture is the avatar card.
router.get('/latest-thumbs', tenantAuth({ verb: 'list', allowUser: true }), async (req, res) => {
try {
const ids = String(req.query.managed_stream_ids || '').split(',').filter(Boolean);
ddiff --git a/test/thumbnail-backfill.test.js b/test/thumbnail-backfill.test.js
nnew file mode 100644
iindex 0000000..b3752fc
---- /dev/null
++++ b/test/thumbnail-backfill.test.js
@@@ -0,0 +1,106 @@
++'use strict';
++// The missing-picture backfill (server/jobs/thumbnail.js sweepMissing): a scheduled, due-based sweep of
++// stored public VODs whose thumbnail is absent or whose file is gone. It queues the existing
++// thumbnail.regenerate job for a small batch, does not retry one that is active or recently failed, and
++// logs its count. The job it queues makes a real picture (checked with a 2-second test video), and a row
++// that already has a thumbnail file is left alone.
++// node --import ./test/helpers/pg-preload.mjs test/thumbnail-backfill.test.js
++const assert = require('assert');
++const fs = require('fs');
++const os = require('os');
++const path = require('path');
++const { spawnSync } = require('child_process');
++
++(async () => {
++ const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-thumb-backfill-'));
++ const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
++ Object.assign(process.env, {
++ VOD_PATH: dir('vods'), CLIPS_PATH: dir('clips'), FILES_PATH: dir('files'),
++ THUMBNAILS_PATH: dir('thumbnails'), PASTES_PATH: dir('pastes'), OBJECTS_PATH: dir('objects'),
++ MEDIA_INVARIANT_SCAN_HOURS: '0', MEDIA_JOBS_ENABLED: '0',
++ });
++
++ const db = require('../server/db/database');
++ const model = require('../server/objects/model');
++ const jobs = require('../server/jobs/thumbnail');
++ const THUMB_DIR = require('../server/thumbnails/thumbnail-service').THUMB_DIR;
++ const AGE = Date.now() - 3600 * 1000;
++ const oldIso = new Date(AGE).toISOString();
++
++ await db.upsertApp({ app_id: 'live', api_key: 'live-key' });
++
++ const video = path.join(process.env.VOD_PATH, 'clip.mp4');
++ const hasFfmpeg = spawnSync('ffmpeg', ['-version'], { stdio: 'ignore' }).status === 0;
++ if (hasFfmpeg) {
++ const r = spawnSync('ffmpeg', ['-v', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=duration=2:size=160x120:rate=10',
++ '-c:v', 'libx264', '-pix_fmt', 'yuv420p', video], { stdio: 'ignore' });
++ assert.strictEqual(r.status, 0, 'the test video was made');
++ }
++
++ async function insert({ title, thumb = null, file = video, audio = false, recording = false, isPublic = true, created_at = oldIso, health = 'ok', managed_stream_id = null }) {
++ const id = Number((await db.run(`INSERT INTO vods (app_id, title, file_path, file_size, duration_seconds, thumbnail_url, health_status,
++ is_recording, is_public, visibility, audio_only, created_at, managed_stream_id)
++ VALUES ('live', ?, ?, ?, 60, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id`,
++ [title, file, file && fs.existsSync(file) ? fs.statSync(file).size : 0, thumb, health,
++ recording ? 1 : 0, isPublic ? 1 : 0, isPublic ? 'public' : 'private', audio ? 1 : 0, created_at, managed_stream_id])).lastInsertRowid);
++ await model.sync('vod', id);
++ return id;
++ }
++
++ // A picture file that is really there.
++ fs.writeFileSync(path.join(THUMB_DIR, 'vod-present-1.jpg'), Buffer.from('not-a-real-jpeg-but-a-file'));
++
++ const noPicture = await insert({ title: 'no picture yet' });
++ const lostFile = await insert({ title: 'picture file gone', thumb: '/t/vod-lost-file-1.jpg' });
++ const present = await insert({ title: 'picture present', thumb: '/t/vod-present-1.jpg' });
++ const audioOnly = await insert({ title: 'audio only', audio: true });
++ const recording = await insert({ title: 'still recording', recording: true });
++ const tooNew = await insert({ title: 'just finalized', created_at: new Date().toISOString() });
++ const privateVod = await insert({ title: 'private', isPublic: false });
++ const broken = await insert({ title: 'missing file', health: 'missing_file' });
++
++ // The newest recording per slot wins deterministically (same created_at: the newer id), and it says
++ // when it has no frame to show, so Live's card can carry the flag instead of a 1×1 fallback.
++ const olderSlotVod = await insert({ title: 'slot older', thumb: '/t/vod-present-1.jpg', managed_stream_id: 77 });
++ const newestSlotVod = await insert({ title: 'slot newest audio', thumb: '/t/vod-present-1.jpg', audio: true, managed_stream_id: 77 });
++ assert.ok(olderSlotVod < newestSlotVod);
++ const thumb = (await db.latestVodThumbsByManagedStreams('live', [77]))[77];
++ assert.deepStrictEqual(thumb, { vod_id: newestSlotVod, thumbnail_url: '/t/vod-present-1.jpg', audio_only: true },
++ 'latest-thumbs picks the newest row and carries audio_only for the card');
++
++ assert.strictEqual(jobs.hasPicture({ thumbnail_url: '/t/vod-present-1.jpg' }), true, 'present file is a picture');
++ assert.strictEqual(jobs.hasPicture({ thumbnail_url: '/t/vod-lost-file-1.jpg' }), false, 'a URL whose file is gone is not');
++
++ // ── 1. A small, due batch: missing pictures only, not recordings/new/private/broken ones ──
++ const now = Date.now();
++ const first = await jobs.sweepMissing({ appId: 'live', now, limit: 10 });
++ assert.deepStrictEqual(first, { missing: 3, queued: 3 }, 'no picture, lost file, audio-only');
++ const queued = (await db.all("SELECT params FROM media_jobs WHERE app_id = 'live' AND job_type = 'thumbnail.regenerate' ORDER BY id"))
++ .map((r) => JSON.parse(r.params).id).sort((a, b) => a - b);
++ assert.deepStrictEqual(queued, [noPicture, lostFile, audioOnly].sort((a, b) => a - b),
++ `only the missing ones were queued (saw ${JSON.stringify(queued)}; ${present}/${recording}/${tooNew}/${privateVod}/${broken} were skipped)`);
++
++ // ── 2. Due based: an active job is not queued twice, and a failed one waits ──
++ const again = await jobs.sweepMissing({ appId: 'live', now: now + 60 * 1000, limit: 10 });
++ assert.deepStrictEqual(again, { missing: 3, queued: 0 }, 'active jobs are not queued again');
++ await db.run(`UPDATE media_jobs SET status = 'failed', error_code = 'generate_failed', created_at = ? WHERE job_type = 'thumbnail.regenerate' AND params = ?`,
++ [new Date(now - 60 * 60 * 1000).toISOString(), JSON.stringify({ kind: 'vod', id: noPicture })]);
++ const third = await jobs.sweepMissing({ appId: 'live', now: now + 2 * 60 * 1000, limit: 10 });
++ assert.strictEqual(third.queued, 0, 'a generate_failed attempt waits six hours');
++
++ // ── 3. The queued job makes the picture: a real frame from the stored video ──
++ if (hasFfmpeg) {
++ const before = await db.get('SELECT thumbnail_url FROM vods WHERE id = ?', [lostFile]);
++ assert.strictEqual(before.thumbnail_url, '/t/vod-lost-file-1.jpg');
++ const out = await jobs.spec.run({ app_id: 'live', params: { kind: 'vod', id: lostFile } }, {});
++ assert.match(out.url, /^\/t\/vod-\d+-\d+-[0-9a-f]{6}\.jpg$/);
++ const after = await db.get('SELECT thumbnail_url FROM vods WHERE id = ?', [lostFile]);
++ assert.strictEqual(after.thumbnail_url, out.url, 'the row names the new picture');
++ assert.ok(fs.existsSync(path.join(THUMB_DIR, path.basename(out.url))), 'the file is there');
++ assert.strictEqual(jobs.hasPicture(after), true, 'and the sweep no longer sees it as missing');
++ } else {
++ console.log('thumbnail-backfill: generation check skipped (ffmpeg not found)');
++ }
++
++ console.log('thumbnail-backfill: all checks passed');
++})().catch((e) => { console.error(e); process.exit(1); });