openvibers/openvibe.media Commits
vod.finalize job: a recording whose finalize failed or never ran is retried on media_jobs with backoff, instead of a one-shot boot sweep whose failure was swallowed and a stat failure that only cleared is_recording. The orphan sweep (in the worker: at boot and every MEDIA_FINALIZE_SWEEP_S, 120 s) queues one job per vods row still is_recording = 1 that no recorder, chunk upload or finalize holds and whose file has been idle for MEDIA_FINALIZE_ORPHAN_GRACE_S (180 s; a browser chunk upload resuming after a restart keeps writing its file, so it is no longer finalized under the uploader). Finalize queues one itself (first run in 5 min) when it cannot settle a recording: nothing measurable, the file could not be stat'ed, or it threw. The job re-runs finalize (fromJob, so it queues nothing more), succeeds with the outcome (ready with duration and source, needs_review for short, corrupt, deleted for an empty file, skipped when a recording or chunk upload still holds it or it is already final) and otherwise fails with a retryable code (unmeasurable, stat_failed, finalize_threw): 5 min, 15 min, 45 min, 2 h 15, 6 h, 6 attempts, after which the VOD stays needs_review and hidden. It never stops a live recording. New finalize lane (MEDIA_JOBS_FINALIZE_CONCURRENCY, 1) that runs while recording, like the recorder's own finalize. Duplicate requests join the queued or running job. Without the worker (MEDIA_JOBS_ENABLED=0) the boot sweep finalizes directly, as before. The v2 jobs API accepts vod.finalize for the tenant's own VODs. test/vod-finalize-job.test.js.
Changes
@@@ -400,7 +400,7 @@ W7) is what will make every new recording fit.## Jobs--Code: `server/jobs/`. Tests: `test/jobs.test.js`, `test/jobs-invariant.test.js`.++Code: `server/jobs/`. Tests: `test/jobs.test.js`, `test/jobs-invariant.test.js`, `test/vod-finalize-job.test.js`.**States.** `proposed` (waiting for the owner; the worker never takes it) → `queued` → `running` →`succeeded` | `failed` | `cancelled`. A failed attempt goes back to `queued` with `run_after` (backoff@@@ -443,9 +443,10 @@ jobs answer 409. The owner is the tenant (app key, service token for the namespaapp token); a caller acting for one of the app's users (`X-OV-User-Id`) sees and decides only the jobsit created and the jobs on objects it owns.--**Worker.** In-process, polling every `MEDIA_JOBS_POLL_MS`, in two lanes: `light`--(`MEDIA_JOBS_LIGHT_CONCURRENCY`, 2) and `heavy` (`MEDIA_JOBS_HEAVY_CONCURRENCY`, 1; waits while a--recording runs unless `MEDIA_JOBS_HEAVY_WHILE_RECORDING=1`). A running job holds a lease++**Worker.** In-process, polling every `MEDIA_JOBS_POLL_MS`, in three lanes: `light`++(`MEDIA_JOBS_LIGHT_CONCURRENCY`, 2), `heavy` (`MEDIA_JOBS_HEAVY_CONCURRENCY`, 1; waits while a++recording runs unless `MEDIA_JOBS_HEAVY_WHILE_RECORDING=1`) and `finalize`++(`MEDIA_JOBS_FINALIZE_CONCURRENCY`, 1; runs while recording, as the recorder's own finalize does). A running job holds a lease(`MEDIA_JOBS_LEASE_S`) renewed by a heartbeat; at start, jobs the previous process left `running` arerequeued (or failed when out of attempts). Handlers checkpoint progress and resume from it.`MEDIA_JOBS_ENABLED=0` stops the worker. Finished thumbnail jobs are pruned after@@@ -457,6 +458,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) || `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part || `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |++| `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and toolittle free disk (the source size plus `MEDIA_UPLOAD_MIN_FREE_MB`; retried after 30 min).@@@ -538,7 +540,8 @@ Not run against production yet: it needs the owner's go-ahead.| `MEDIA_UPLOAD_MIN_FREE_MB` | 10240 | free disk kept in reserve by multipart uploads and split/remux jobs || `MEDIA_JOBS_ENABLED` | on | `0` stops the job worker || `MEDIA_JOBS_POLL_MS` | 5000 | worker poll interval |--| `MEDIA_JOBS_LIGHT_CONCURRENCY` / `_HEAVY_CONCURRENCY` | 2 / 1 | jobs per lane |++| `MEDIA_JOBS_LIGHT_CONCURRENCY` / `_HEAVY_CONCURRENCY` / `_FINALIZE_CONCURRENCY` | 2 / 1 / 1 | jobs per lane |++| `MEDIA_FINALIZE_ORPHAN_GRACE_S` / `MEDIA_FINALIZE_SWEEP_S` | 180 / 120 | an orphaned recording's file must be idle this long; the orphan sweep runs this often || `MEDIA_JOBS_HEAVY_WHILE_RECORDING` | off | `1` lets split/remux run while a recording is being written || `MEDIA_JOBS_LEASE_S` | 120 | running-job lease (renewed by the heartbeat) || `MEDIA_INVARIANT_SCAN_HOURS` | 24 | the size-invariant validator's schedule per tenant (`0` = on demand only) |@@@ -70,6 +70,7 @@ const config = {pollMs: intEnv('MEDIA_JOBS_POLL_MS', 5000),lightConcurrency: Math.max(1, intEnv('MEDIA_JOBS_LIGHT_CONCURRENCY', 2)), // thumbnails, scansheavyConcurrency: Math.max(1, intEnv('MEDIA_JOBS_HEAVY_CONCURRENCY', 1)), // split, remux++ finalizeConcurrency: Math.max(1, intEnv('MEDIA_JOBS_FINALIZE_CONCURRENCY', 1)), // vod.finalize retries// Heavy jobs wait while a recording is running (as the health job does), unless this is on.heavyWhileRecording: ['1', 'true', 'on'].includes(String(process.env.MEDIA_JOBS_HEAVY_WHILE_RECORDING || '').toLowerCase()),leaseS: Math.max(30, intEnv('MEDIA_JOBS_LEASE_S', 120)),@@@ -234,9 +234,10 @@ if (!drill.enabled) {} catch (err) { console.warn('[Objects] Backfill failed:', err.message); }}, 15 * 1000).unref?.();-- // Recover from an unclean shutdown: rows stuck in is_recording with no live-- // ffmpeg are finalized from whatever hit the disk.-- setTimeout(() => {++ // Recover from an unclean shutdown: rows stuck in is_recording with no live ffmpeg are finalized++ // from whatever hit the disk. With the job worker on, its orphan sweep queues a vod.finalize job++ // for each (retried with backoff; server/jobs/vod-finalize.js); without it, once, here.++ if (!config.jobs.enabled) setTimeout(() => {try {const stuck = db.all('SELECT id FROM vods WHERE is_recording = 1');for (const row of stuck) {@@@ -5,6 +5,7 @@* invariant.scan light the size-invariant validator: proposes object.split / object.remux* object.split heavy stream-copy parts of a large vod/clip as new private objects* object.remux heavy a stream-copy remux of a vod/clip as a new private object++ * vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff*/'use strict';@@@ -14,5 +15,6 @@ queue.register('thumbnail.regenerate', require('./thumbnail').spec);queue.register('invariant.scan', require('./invariant-scan').spec);queue.register('object.split', require('./derive').split);queue.register('object.remux', require('./derive').remux);++queue.register('vod.finalize', require('./vod-finalize').spec);module.exports = { names: () => queue.typeNames() };nnew file mode 100644@@@ -0,0 +1,170 @@++/**++ * OpenVibe.Media — job type vod.finalize (finalize lane): finalize a recording whose finalize failed++ * or never ran, with backoff (docs/object-model.md#jobs).++ *++ * Queued by++ * - the orphan sweep: a vods row still is_recording = 1 that no recorder, chunk upload or finalize++ * holds and whose file has not grown for MEDIA_FINALIZE_ORPHAN_GRACE_S (default 180 s). At boot++ * (every recording of the previous process) and every MEDIA_FINALIZE_SWEEP_S (120 s) after that.++ * - finalize itself, when it could not settle a recording: nothing measurable (duration stored 0,++ * needs_review), the file could not be stat'ed, or it threw. First retry after 5 minutes.++ * - the v2 jobs API (POST /api/v2/:app/jobs { type: 'vod.finalize', params: { vod_id } }).++ *++ * params { vod_id } (derived from the object's legacy ref when omitted)++ * result { outcome: ready | needs_review | corrupt | deleted | skipped, vod_id, duration_seconds?,++ * duration_source?, health_status?, reason? }++ * A run that still cannot measure the recording fails with a retryable code (unmeasurable,++ * stat_failed, finalize_threw); the worker retries it (5 min, 15 min, 45 min, 2 h 15, then 6 h;++ * 6 attempts). When the attempts run out the VOD stays needs_review, hidden, for a person.++ * The job never stops a running recording and never runs while a chunk upload is still adding to it.++ */++'use strict';++++const fs = require('fs');++const db = require('../db/database');++const queue = require('./queue');++++const { JobError } = queue;++const TYPE = 'vod.finalize';++// Outcomes of a finalize that another attempt may fix (see server/vod/finalize.js FINALIZE_ISSUES).++const RETRYABLE = { probe_failed: 'unmeasurable', inflated_duration: 'unmeasurable', stat_failed: 'stat_failed', finalize_failed: 'finalize_threw' };++++function envInt(name, fallback) {++ const v = parseInt(process.env[name] || '', 10);++ return Number.isFinite(v) && v >= 0 ? v : fallback;++}++const graceMs = () => envInt('MEDIA_FINALIZE_ORPHAN_GRACE_S', 180) * 1000;++const sweepEveryMs = () => envInt('MEDIA_FINALIZE_SWEEP_S', 120) * 1000;++++function issuesOf(vod) {++ try { const v = JSON.parse(vod.health_issues_json || '[]'); return Array.isArray(v) ? v : []; } catch { return []; }++}++++function chunkUploadActive(vodId) {++ try { return require('../vod/routes').activeChunkUploads.has(Number(vodId)); } catch { return false; }++}++++/** Something in this process is still writing or finalizing the recording. */++function busy(vodId) {++ try { if (require('../vod/recorder').isRecording(vodId)) return 'recording'; } catch { /* */ }++ try { if (require('../vod/finalize').isFinalizing(vodId)) return 'finalizing'; } catch { /* */ }++ if (chunkUploadActive(vodId)) return 'chunk upload in progress';++ return null;++}++++function vodIdOf(obj, params = {}) {++ if (params.vod_id != null) return Number(params.vod_id);++ if (obj) {++ const ref = require('../objects/model').parseLegacyRef(obj.legacy_ref);++ if (ref && ref.kind === 'vod') return Number(ref.id);++ }++ return null;++}++++/** API validation: the VOD must be this tenant's (and the named object's). */++function validate({ appId, obj, params }) {++ const id = vodIdOf(obj, params || {});++ if (!Number.isInteger(id) || id < 1) throw new JobError('media.job.invalid', 'vod.finalize needs params.vod_id (or a vod object)', { permanent: true });++ const vod = db.getVodById(id, appId);++ if (!vod) throw new JobError('media.job.not_found', `vod ${id} not found`, { status: 404, permanent: true });++ if (obj && vod.object_id && vod.object_id !== obj.id) throw new JobError('media.job.invalid', `the object is not vod ${id}`, { permanent: true });++ return { vod_id: id };++}++++/** Does this VOD still need a finalize? Returns the reason, or null. */++function needsFinalize(vod) {++ if (!vod) return null;++ if (vod.is_recording) return 'orphan';++ const retry = issuesOf(vod).find(i => RETRYABLE[i]);++ return retry && vod.health_status === 'needs_review' ? retry : null;++}++++async function run(job) {++ const vodId = Number(job.params && job.params.vod_id);++ const vod = db.getVodById(vodId, job.app_id);++ if (!vod) return { outcome: 'skipped', vod_id: vodId, reason: 'the vod no longer exists' };++ const holder = busy(vodId);++ if (holder === 'finalizing') throw new JobError('busy', 'a finalize of this vod is already running', { retryAfterS: 60 });++ if (holder) return { outcome: 'skipped', vod_id: vodId, reason: `${holder}: that path finalizes it` };++ const why = needsFinalize(vod);++ if (!why) return { outcome: 'skipped', vod_id: vodId, reason: 'already finalized', health_status: vod.health_status };++++ try {++ await require('../vod/finalize').finalizeVod(vodId, { fromJob: true });++ } catch (err) {++ throw new JobError('finalize_threw', `finalize failed: ${err.message}`);++ }++ const after = db.getVodById(vodId, job.app_id);++ if (!after) return { outcome: 'deleted', vod_id: vodId, reason: 'no media was ever written (empty or missing file)' };++ if (after.is_recording) throw new JobError('busy', 'the vod is still marked recording after finalize', { retryAfterS: 60 });++ const still = issuesOf(after).find(i => RETRYABLE[i]);++ if (still && after.health_status === 'needs_review') {++ throw new JobError(RETRYABLE[still], `vod ${vodId} still cannot be settled (${issuesOf(after).join(', ')})`);++ }++ return {++ outcome: after.health_status === 'ok' ? 'ready' : after.health_status,++ vod_id: vodId, duration_seconds: after.duration_seconds, duration_source: after.duration_source || null,++ health_status: after.health_status,++ };++}++++/**++ * Queue a finalize for this VOD (joins one that is already queued or running). Never throws;++ * returns { job, created } or null. `reason` goes into created_by (system:vod.finalize:<reason>).++ */++function queueFinalize(vodOrId, reason, { runAfterS = 0 } = {}) {++ try {++ const vod = typeof vodOrId === 'object' ? vodOrId : db.getVodById(Number(vodOrId));++ if (!vod) return null;++ const r = queue.enqueue({++ appId: vod.app_id, type: TYPE, objectId: vod.object_id || null, params: { vod_id: vod.id },++ dedupeActive: true, createdBy: `system:${TYPE}:${reason}`, runAfterS,++ });++ if (r.created) {++ console.log(`[Jobs] vod.finalize queued for vod ${vod.id} (${reason}${runAfterS ? `, in ${runAfterS} s` : ''})`);++ try { require('./worker').kick(); } catch { /* next poll */ }++ }++ return { job: r.job, created: r.created };++ } catch (err) {++ console.warn(`[Jobs] could not queue vod.finalize for vod ${typeof vodOrId === 'object' ? vodOrId.id : vodOrId}: ${err.message}`);++ return null;++ }++}++++/**++ * Recordings left is_recording = 1 with nothing holding them. `graceMs`: the file must not have++ * changed for this long (a chunk upload resuming after a restart keeps writing it). Returns ids.++ */++function orphans({ grace = graceMs(), now = Date.now() } = {}) {++ const out = [];++ // clips_only recordings too: their finalize discards them (file and row).++ for (const row of db.all('SELECT id, file_path FROM vods WHERE is_recording = 1 ORDER BY id')) {++ if (busy(row.id)) continue;++ let mtime = 0;++ try { mtime = row.file_path ? fs.statSync(row.file_path).mtimeMs : 0; } catch { mtime = 0; }++ if (mtime && now - mtime < grace) continue;++ out.push(row.id);++ }++ return out;++}++++let lastSweepAt = 0;++++/** Queue vod.finalize for every orphan (due every MEDIA_FINALIZE_SWEEP_S; `force` = now). */++function sweepOrphans({ force = false, now = Date.now(), grace } = {}) {++ if (!force && now - lastSweepAt < sweepEveryMs()) return 0;++ lastSweepAt = now;++ let n = 0;++ for (const id of orphans({ grace, now })) { const r = queueFinalize(id, 'orphan'); if (r && r.created) n++; }++ return n;++}++++// 5 min, 15 min, 45 min, 2 h 15, then 6 h.++const backoffS = (attempt) => Math.min(6 * 3600, 300 * 3 ** Math.max(0, attempt - 1));++++module.exports = {++ TYPE,++ spec: { lane: 'finalize', maxAttempts: 6, timeoutMs: 60 * 60 * 1000, needsObject: false, backoffS, validate, run },++ queueFinalize, sweepOrphans, orphans, needsFinalize, backoffS, RETRYABLE,++ _resetSweep: () => { lastSweepAt = 0; },++};@@@ -5,6 +5,8 @@* light thumbnail.regenerate, invariant.scan MEDIA_JOBS_LIGHT_CONCURRENCY (2)* heavy object.split, object.remux MEDIA_JOBS_HEAVY_CONCURRENCY (1); waits while a* recording runs unless MEDIA_JOBS_HEAVY_WHILE_RECORDING=1++ * finalize vod.finalize MEDIA_JOBS_FINALIZE_CONCURRENCY (1); runs while recording++ * (it is the work the recorder does when a stream ends)** A running job holds a lease (MEDIA_JOBS_LEASE_S) that a heartbeat renews; the heartbeat also notices* an owner's cancel request. At start every job left `running` by the previous process is retried (or@@@ -13,7 +15,8 @@* from the checkpoint on a retry.** Scheduling: the size-invariant validator (invariant.scan) is enqueued per tenant every-- * MEDIA_INVARIANT_SCAN_HOURS (it proposes split/remux jobs; it never runs them).++ * MEDIA_INVARIANT_SCAN_HOURS (it proposes split/remux jobs; it never runs them). Orphaned recordings++ * get a vod.finalize every MEDIA_FINALIZE_SWEEP_S (server/jobs/vod-finalize.js).*/'use strict';@@@ -36,6 +39,7 @@ function lanes() {return [{ name: 'light', types: by('light'), max: cfg().lightConcurrency },{ name: 'heavy', types: by('heavy'), max: cfg().heavyConcurrency },++ { name: 'finalize', types: by('finalize'), max: cfg().finalizeConcurrency },];}@@@ -141,6 +145,7 @@ async function tick() {queue.recoverInterrupted({ except: new Set(running.keys()) });const now = Date.now();if (now - lastScheduleAt > 10 * 60 * 1000) { lastScheduleAt = now; scheduleInvariantScans(now); }++ try { require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }if (now - lastPruneAt > 6 * 3600 * 1000) {lastPruneAt = now;try { const n = queue.prune({ days: cfg().retentionDays }); if (n) console.log(`[Jobs] pruned ${n} finished thumbnail job(s)`); } catch { /* next time */ }@@@ -126,6 +126,7 @@ function _commitVodOutcome(vod, event, change, row = () => db.getVodById(vod.id)* @param {number} [opts.startTimeMs] wall-clock recording start (for duration sanity)* @param {boolean} [opts.ffmpegCorrupted] recorder flagged heavy corruption* @param {string} [opts.segmentPath] trailing chunk segment not yet merged++ * @param {boolean} [opts.fromJob] run by the vod.finalize job (which retries itself; queue no retry)*/async function finalizeVod(vodId, opts = {}) {vodId = Number(vodId);@@@ -145,6 +146,10 @@ async function finalizeVod(vodId, opts = {}) {try {return await _doFinalize(vodId, opts);++ } catch (err) {++ // A finalize that throws leaves the recording unsettled: the vod.finalize job tries again.++ _queueRetry(vodId, 'finalize_failed', opts);++ throw err;} finally {_finalizing.delete(vodId);// Every outcome (ready, quarantined, failed) re-projects the object; a deleted row was already marked by trigger.@@@ -152,6 +157,15 @@ async function finalizeVod(vodId, opts = {}) {}}++/**++ * Queue a vod.finalize retry (server/jobs/vod-finalize.js; first attempt in 5 minutes), unless this++ * finalize IS that job (it retries itself with backoff).++ */++function _queueRetry(vodId, reason, opts = {}) {++ if (opts.fromJob) return;++ try { require('../jobs/vod-finalize').queueFinalize(vodId, reason, { runAfterS: 300 }); } catch { /* the orphan sweep and the reconcile job remain */ }++}++// Health issues finalize itself records when it cannot settle a recording. A later successful// finalize (the vod.finalize retry job) lifts the quarantine they caused.const FINALIZE_ISSUES = ['probe_failed', 'inflated_duration', 'stat_failed', 'finalize_failed'];@@@ -283,6 +297,7 @@ async function _doFinalize(vodId, opts) {console.warn(`[VOD] vod ${vodId}: no measurable duration (${issues.join(', ')}) — stored 0, needs_review`);_commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review', health_issues_json = ?, probe_duration_seconds = 0, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,[JSON.stringify(issues), probeFormatJson, vodId]));++ _queueRetry(vodId, issues.includes('probe_failed') ? 'probe_failed' : 'inflated_duration', opts);return db.getVodById(vodId);}@@@ -304,6 +319,7 @@ async function _doFinalize(vodId, opts) {console.error(`[VOD] Failed to stat finalized VOD ${vodId}:`, err.message);db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',health_issues_json = ?, last_health_scan_at = datetime('now') WHERE id = ?`, [JSON.stringify(['stat_failed', ...issues]), vodId]);++ _queueRetry(vodId, 'stat_failed', opts);return null;}// The lossless .master.mkv archive is only a fallback. Delete it ONLY oncennew file mode 100644@@@ -0,0 +1,156 @@++'use strict';++// vod.finalize (server/jobs/vod-finalize.js): orphaned recordings (is_recording = 1, nothing holding++// them, file idle past the grace) are queued once each; a finalize that cannot settle a recording++// (nothing measurable, stat failure, a throw) queues a retry instead of just clearing is_recording;++// the job retries with backoff, never stops a live recording or a chunk upload, and lifts the++// quarantine once a later attempt measures the file.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++++const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-finalize-job-'));++const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };++process.env.DB_PATH = path.join(tmp, 'media.db');++process.env.VOD_PATH = dir('vods');++process.env.CLIPS_PATH = dir('clips');++process.env.THUMBNAILS_PATH = dir('thumbnails');++process.env.FILES_PATH = dir('files');++process.env.PASTES_PATH = dir('pastes');++process.env.OBJECTS_PATH = dir('objects');++process.env.MEDIA_INVARIANT_SCAN_HOURS = '0';++++const db = require('../server/db/database');++const queue = require('../server/jobs/queue');++const worker = require('../server/jobs/worker');++const job = require('../server/jobs/vod-finalize');++const tools = require('../server/vod/media-tools');++const recorder = require('../server/vod/recorder');++const vodRoutes = require('../server/vod/routes');++++db.upsertApp({ app_id: 'live', api_key: 'live-key' });++db.upsertApp({ app_id: 'games', api_key: 'games-key' });++++const realTools = { ...tools };++const restore = () => Object.assign(tools, realTools);++const measurable = (s = 1800) => {++ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: s, error: null });++ tools.probeDuration = async () => ({ ok: true, seconds: s, format: {}, streams: [], error: null });++};++const unmeasurable = () => {++ tools.remuxForSeekingDetailed = async () => ({ ok: false, seconds: 0, error: 'ffmpeg remux exited 1' });++ tools.probeDuration = async () => ({ ok: false, seconds: 0, format: null, streams: [], error: 'unreadable' });++ tools.streamCopyDuration = async () => ({ ok: false, seconds: 0, error: 'no packets' });++};++++function makeVod(name, { idleS = 3600, app = 'live' } = {}) {++ const file = path.join(process.env.VOD_PATH, name);++ fs.writeFileSync(file, Buffer.alloc(30000, 9));++ const t = (Date.now() - idleS * 1000) / 1000;++ fs.utimesSync(file, t, t);++ const r = db.run(`INSERT INTO vods (app_id, user_id, title, file_path, is_recording, duration_seconds, is_public, visibility, created_at)++ VALUES (?, 1, 'rec', ?, 1, 77777, 1, 'public', datetime('now', '-2 days'))`, [app, file]);++ return Number(r.lastInsertRowid);++}++const vod = (id) => db.get('SELECT * FROM vods WHERE id = ?', [id]);++const jobsFor = (id) => db.all("SELECT * FROM media_jobs WHERE job_type = 'vod.finalize' AND json_extract(params, '$.vod_id') = ? ORDER BY id", [id]);++++(async () => {++ assert.ok(queue.typeNames().includes('vod.finalize'));++ assert.strictEqual(queue.typeSpec('vod.finalize').lane, 'finalize');++++ // ── 1. The orphan sweep: idle orphans only; live recordings, chunk uploads and fresh files are left alone ──++ const idle = makeVod('idle.mp4');++ const fresh = makeVod('fresh.mp4', { idleS: 10 });++ const live = makeVod('live.mp4');++ const chunked = makeVod('chunked.webm');++ recorder.activeRecordings.set(live, { vodId: live });++ vodRoutes.activeChunkUploads.set(chunked, { filePath: 'x' });++ assert.deepStrictEqual(job.orphans(), [idle], 'only the idle orphan');++ assert.strictEqual(job.sweepOrphans({ force: true }), 1);++ assert.strictEqual(job.sweepOrphans({ force: true }), 0, 'a queued finalize is joined, not repeated');++ const [j1] = jobsFor(idle);++ assert.strictEqual(j1.created_by, 'system:vod.finalize:orphan');++ assert.strictEqual(j1.app_id, 'live');++ assert.ok(job.orphans({ grace: 1000 }).includes(fresh), 'a file idle past the grace is an orphan');++ console.log('✅ orphan sweep: idle recordings nothing holds are queued once; live, chunk-upload and fresh ones are not');++++ // ── 2. Running it finalizes from the measured file ──++ measurable(1800.2);++ const done = await worker.runNow(j1.id);++ assert.strictEqual(done.status, 'succeeded');++ const res = JSON.parse(done.result);++ assert.deepStrictEqual([res.outcome, res.duration_seconds, res.duration_source], ['ready', 1800, 'probe']);++ assert.deepStrictEqual([vod(idle).is_recording, vod(idle).duration_seconds, vod(idle).health_status], [0, 1800, 'ok']);++ console.log('✅ vod.finalize settles an orphan to its measured duration');++++ // ── 3. A finalize that cannot measure queues a retry (5 min out); the job retries with backoff ──++ unmeasurable();++ const bad = makeVod('bad.mp4');++ await require('../server/vod/finalize').finalizeVod(bad);++ assert.strictEqual(vod(bad).health_status, 'needs_review');++ let js = jobsFor(bad);++ assert.strictEqual(js.length, 1, 'one retry queued');++ assert.strictEqual(js[0].created_by, 'system:vod.finalize:probe_failed');++ const due = db.get("SELECT (julianday(run_after) - julianday('now')) * 86400 AS s FROM media_jobs WHERE id = ?", [js[0].id]).s;++ assert.ok(due > 250 && due <= 301, `first retry in ~5 minutes (${due})`);++ let r = await worker.runNow(js[0].id);++ assert.strictEqual(r.status, 'queued', 'still unmeasurable: back in the queue');++ assert.strictEqual(r.error_code, 'unmeasurable');++ assert.strictEqual(r.attempts, 1);++ assert.strictEqual(jobsFor(bad).length, 1, 'the job does not queue more of itself');++ const wait = db.get("SELECT (julianday(run_after) - julianday('now')) * 86400 AS s FROM media_jobs WHERE id = ?", [r.id]).s;++ assert.ok(wait > 250 && wait <= 301, `backoff after attempt 1 is 5 minutes (${wait})`);++ assert.deepStrictEqual([job.backoffS(1), job.backoffS(2), job.backoffS(3), job.backoffS(6)], [300, 900, 2700, 21600]);++ measurable(610);++ r = await worker.runNow(r.id);++ assert.strictEqual(r.status, 'succeeded');++ const v = vod(bad);++ assert.deepStrictEqual([v.duration_seconds, v.health_status, v.is_public, v.quarantined_at], [610, 'ok', 1, null], 'measured, un-quarantined');++ console.log('✅ unmeasurable finalize → retry job (5 min), retried with backoff, settles once the file measures');++++ // ── 4. Stat failure and a thrown finalize queue a retry too (not just is_recording = 0) ──++ const gone = makeVod('gone.mp4');++ tools.probeDuration = async (p) => { if (p.endsWith('gone.mp4')) { try { fs.unlinkSync(p); } catch { /* */ } } return { ok: true, seconds: 300, format: {}, streams: [], error: null }; };++ await require('../server/vod/finalize').finalizeVod(gone);++ assert.deepStrictEqual(jobsFor(gone).map(j => j.created_by), ['system:vod.finalize:stat_failed']);++ const thrower = makeVod('thrower.mp4');++ tools.remuxForSeekingDetailed = async () => { throw new Error('boom'); };++ await assert.rejects(require('../server/vod/finalize').finalizeVod(thrower), /boom/);++ assert.deepStrictEqual(jobsFor(thrower).map(j => j.created_by), ['system:vod.finalize:finalize_failed']);++ assert.strictEqual(vod(thrower).is_recording, 1, 'still unsettled, and the job knows it');++ measurable(95);++ const tr = await worker.runNow(jobsFor(thrower)[0].id);++ assert.strictEqual(tr.status, 'succeeded');++ assert.strictEqual(vod(thrower).duration_seconds, 95);++ console.log('✅ a stat failure or a thrown finalize queues vod.finalize, which settles it later');++++ // ── 5. Never touches a live recording; out of attempts the VOD waits for a person ──++ const jl = queue.enqueue({ appId: 'live', type: 'vod.finalize', params: { vod_id: live }, createdBy: 'test' }).job;++ const lr = await worker.runNow(jl.id);++ assert.strictEqual(JSON.parse(lr.result).outcome, 'skipped');++ assert.strictEqual(vod(live).is_recording, 1, 'the recording keeps going');++ recorder.activeRecordings.delete(live);++ vodRoutes.activeChunkUploads.delete(chunked);++++ unmeasurable();++ const last = makeVod('last.mp4');++ const jlast = queue.enqueue({ appId: 'live', type: 'vod.finalize', params: { vod_id: last }, createdBy: 'test', maxAttempts: 1 }).job;++ const out = await worker.runNow(jlast.id);++ assert.deepStrictEqual([out.status, out.error_code], ['failed', 'unmeasurable']);++ assert.deepStrictEqual([vod(last).is_recording, vod(last).duration_seconds, vod(last).health_status, vod(last).is_public], [0, 0, 'needs_review', 0]);++ restore();++ console.log('✅ a live recording is never finalized by the job; out of attempts the VOD stays needs_review, hidden');++++ // ── 6. API validation: the VOD must be the tenant's ──++ const gv = makeVod('games.mp4', { app: 'games' });++ assert.throws(() => job.spec.validate({ appId: 'live', obj: null, params: { vod_id: gv } }), (e) => e.code === 'media.job.not_found');++ assert.deepStrictEqual(job.spec.validate({ appId: 'games', obj: null, params: { vod_id: gv } }), { vod_id: gv });++ assert.throws(() => job.spec.validate({ appId: 'games', obj: null, params: {} }), (e) => e.code === 'media.job.invalid');++ console.log('✅ vod.finalize through the jobs API is scoped to the tenant\'s own VODs');++++ db.close();++ fs.rmSync(tmp, { recursive: true, force: true });++ console.log('\n✅ All vod.finalize job tests passed');++ process.exit(0);++})().catch((err) => { console.error(err); process.exit(1); });
@@ -400,7 +400,7 @@ W7) is what will make every new recording fit.
@@ -400,7 +400,7 @@ W7) is what will make every new recording fit.
## Jobs
## Jobs
-Code: `server/jobs/`. Tests: `test/jobs.test.js`, `test/jobs-invariant.test.js`.
+Code: `server/jobs/`. Tests: `test/jobs.test.js`, `test/jobs-invariant.test.js`, `test/vod-finalize-job.test.js`.
**States.** `proposed` (waiting for the owner; the worker never takes it) → `queued` → `running` →
**States.** `proposed` (waiting for the owner; the worker never takes it) → `queued` → `running` →
`succeeded` | `failed` | `cancelled`. A failed attempt goes back to `queued` with `run_after` (backoff
`succeeded` | `failed` | `cancelled`. A failed attempt goes back to `queued` with `run_after` (backoff
@@ -443,9 +443,10 @@ jobs answer 409. The owner is the tenant (app key, service token for the namespa
@@ -443,9 +443,10 @@ jobs answer 409. The owner is the tenant (app key, service token for the namespa
app token); a caller acting for one of the app's users (`X-OV-User-Id`) sees and decides only the jobs
app token); a caller acting for one of the app's users (`X-OV-User-Id`) sees and decides only the jobs
it created and the jobs on objects it owns.
it created and the jobs on objects it owns.
-**Worker.** In-process, polling every `MEDIA_JOBS_POLL_MS`, in two lanes: `light`
-(`MEDIA_JOBS_LIGHT_CONCURRENCY`, 2) and `heavy` (`MEDIA_JOBS_HEAVY_CONCURRENCY`, 1; waits while a
-recording runs unless `MEDIA_JOBS_HEAVY_WHILE_RECORDING=1`). A running job holds a lease
+**Worker.** In-process, polling every `MEDIA_JOBS_POLL_MS`, in three lanes: `light`
+(`MEDIA_JOBS_LIGHT_CONCURRENCY`, 2), `heavy` (`MEDIA_JOBS_HEAVY_CONCURRENCY`, 1; waits while a
+recording runs unless `MEDIA_JOBS_HEAVY_WHILE_RECORDING=1`) and `finalize`
+(`MEDIA_JOBS_FINALIZE_CONCURRENCY`, 1; runs while recording, as the recorder's own finalize does). A running job holds a lease
(`MEDIA_JOBS_LEASE_S`) renewed by a heartbeat; at start, jobs the previous process left `running` are
(`MEDIA_JOBS_LEASE_S`) renewed by a heartbeat; at start, jobs the previous process left `running` are
requeued (or failed when out of attempts). Handlers checkpoint progress and resume from it.
requeued (or failed when out of attempts). Handlers checkpoint progress and resume from it.
`MEDIA_JOBS_ENABLED=0` stops the worker. Finished thumbnail jobs are pruned after
`MEDIA_JOBS_ENABLED=0` stops the worker. Finished thumbnail jobs are pruned after
@@ -457,6 +458,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu
@@ -457,6 +458,7 @@ requeued (or failed when out of attempts). Handlers checkpoint progress and resu
| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) |
| `invariant.scan` | light | The [size-invariant validator](#public-object-size-invariant): records violations, proposes `object.split` / `object.remux`, withdraws moot proposals. Tenant-wide (no object) |
| `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part |
| `object.split` | heavy | Stream-copies a vod/clip (or a video/audio object) into parts (`params.parts` 2-1000 or `segment_seconds`), each a new **private** object with a `derived_from` relationship (`job_id`, `part`, `start_seconds`, `duration_seconds`). Cuts land on keyframes, so neighbouring parts can overlap slightly. The source is never changed. Checkpointed per part |
| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |
| `object.remux` | heavy | Stream-copy remux of the whole source (seek index, duration, MP4 faststart) into one new private object; also the source's `remux` variant |
+| `vod.finalize` | finalize | Finalizes a recording whose finalize failed or never ran. `params { vod_id }`. Queued by the orphan sweep (a row still `is_recording = 1` that no recorder, chunk upload or finalize holds and whose file has been idle for `MEDIA_FINALIZE_ORPHAN_GRACE_S`, 180 s; at boot and every `MEDIA_FINALIZE_SWEEP_S`, 120 s) and by finalize itself when it could not settle a recording (nothing measurable, a stat failure, a throw; first run after 5 min). Still unmeasurable: retried (5 min, 15 min, 45 min, 2 h 15, 6 h; 6 attempts), then the VOD stays `needs_review`, hidden. Never stops a live recording. Result `{ outcome: ready \| needs_review \| corrupt \| deleted \| skipped, duration_seconds, duration_source }` |
Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and too
Split and remux refuse a source that is not ready, a tenant quota the output would exceed, and too
little free disk (the source size plus `MEDIA_UPLOAD_MIN_FREE_MB`; retried after 30 min).
little free disk (the source size plus `MEDIA_UPLOAD_MIN_FREE_MB`; retried after 30 min).
@@ -538,7 +540,8 @@ Not run against production yet: it needs the owner's go-ahead.
@@ -538,7 +540,8 @@ Not run against production yet: it needs the owner's go-ahead.
| `MEDIA_UPLOAD_MIN_FREE_MB` | 10240 | free disk kept in reserve by multipart uploads and split/remux jobs |
| `MEDIA_UPLOAD_MIN_FREE_MB` | 10240 | free disk kept in reserve by multipart uploads and split/remux jobs |
| `MEDIA_JOBS_ENABLED` | on | `0` stops the job worker |
| `MEDIA_JOBS_ENABLED` | on | `0` stops the job worker |
| `MEDIA_JOBS_POLL_MS` | 5000 | worker poll interval |
| `MEDIA_JOBS_POLL_MS` | 5000 | worker poll interval |
-| `MEDIA_JOBS_LIGHT_CONCURRENCY` / `_HEAVY_CONCURRENCY` | 2 / 1 | jobs per lane |
+| `MEDIA_JOBS_LIGHT_CONCURRENCY` / `_HEAVY_CONCURRENCY` / `_FINALIZE_CONCURRENCY` | 2 / 1 / 1 | jobs per lane |
+| `MEDIA_FINALIZE_ORPHAN_GRACE_S` / `MEDIA_FINALIZE_SWEEP_S` | 180 / 120 | an orphaned recording's file must be idle this long; the orphan sweep runs this often |
| `MEDIA_JOBS_HEAVY_WHILE_RECORDING` | off | `1` lets split/remux run while a recording is being written |
| `MEDIA_JOBS_HEAVY_WHILE_RECORDING` | off | `1` lets split/remux run while a recording is being written |
| `MEDIA_JOBS_LEASE_S` | 120 | running-job lease (renewed by the heartbeat) |
| `MEDIA_JOBS_LEASE_S` | 120 | running-job lease (renewed by the heartbeat) |
| `MEDIA_INVARIANT_SCAN_HOURS` | 24 | the size-invariant validator's schedule per tenant (`0` = on demand only) |
| `MEDIA_INVARIANT_SCAN_HOURS` | 24 | the size-invariant validator's schedule per tenant (`0` = on demand only) |
@@ -70,6 +70,7 @@ const config = {
@@ -70,6 +70,7 @@ const config = {
pollMs: intEnv('MEDIA_JOBS_POLL_MS', 5000),
pollMs: intEnv('MEDIA_JOBS_POLL_MS', 5000),
lightConcurrency: Math.max(1, intEnv('MEDIA_JOBS_LIGHT_CONCURRENCY', 2)), // thumbnails, scans
lightConcurrency: Math.max(1, intEnv('MEDIA_JOBS_LIGHT_CONCURRENCY', 2)), // thumbnails, scans
heavyConcurrency: Math.max(1, intEnv('MEDIA_JOBS_HEAVY_CONCURRENCY', 1)), // split, remux
heavyConcurrency: Math.max(1, intEnv('MEDIA_JOBS_HEAVY_CONCURRENCY', 1)), // split, remux
+ finalizeConcurrency: Math.max(1, intEnv('MEDIA_JOBS_FINALIZE_CONCURRENCY', 1)), // vod.finalize retries
// Heavy jobs wait while a recording is running (as the health job does), unless this is on.
// Heavy jobs wait while a recording is running (as the health job does), unless this is on.
heavyWhileRecording: ['1', 'true', 'on'].includes(String(process.env.MEDIA_JOBS_HEAVY_WHILE_RECORDING || '').toLowerCase()),
heavyWhileRecording: ['1', 'true', 'on'].includes(String(process.env.MEDIA_JOBS_HEAVY_WHILE_RECORDING || '').toLowerCase()),
leaseS: Math.max(30, intEnv('MEDIA_JOBS_LEASE_S', 120)),
leaseS: Math.max(30, intEnv('MEDIA_JOBS_LEASE_S', 120)),
@@ -234,9 +234,10 @@ if (!drill.enabled) {
@@ -234,9 +234,10 @@ if (!drill.enabled) {
} catch (err) { console.warn('[Objects] Backfill failed:', err.message); }
} catch (err) { console.warn('[Objects] Backfill failed:', err.message); }
}, 15 * 1000).unref?.();
}, 15 * 1000).unref?.();
- // Recover from an unclean shutdown: rows stuck in is_recording with no live
- // ffmpeg are finalized from whatever hit the disk.
- setTimeout(() => {
+ // Recover from an unclean shutdown: rows stuck in is_recording with no live ffmpeg are finalized
+ // from whatever hit the disk. With the job worker on, its orphan sweep queues a vod.finalize job
+ // for each (retried with backoff; server/jobs/vod-finalize.js); without it, once, here.
+ if (!config.jobs.enabled) setTimeout(() => {
try {
try {
const stuck = db.all('SELECT id FROM vods WHERE is_recording = 1');
const stuck = db.all('SELECT id FROM vods WHERE is_recording = 1');
for (const row of stuck) {
for (const row of stuck) {
@@ -5,6 +5,7 @@
@@ -5,6 +5,7 @@
* invariant.scan light the size-invariant validator: proposes object.split / object.remux
* invariant.scan light the size-invariant validator: proposes object.split / object.remux
* object.split heavy stream-copy parts of a large vod/clip as new private objects
* object.split heavy stream-copy parts of a large vod/clip as new private objects
* object.remux heavy a stream-copy remux of a vod/clip as a new private object
* object.remux heavy a stream-copy remux of a vod/clip as a new private object
+ * vod.finalize finalize finalize a recording whose finalize failed or never ran (orphans), with backoff
*/
*/
'use strict';
'use strict';
@@ -14,5 +15,6 @@ queue.register('thumbnail.regenerate', require('./thumbnail').spec);
@@ -14,5 +15,6 @@ queue.register('thumbnail.regenerate', require('./thumbnail').spec);
queue.register('invariant.scan', require('./invariant-scan').spec);
queue.register('invariant.scan', require('./invariant-scan').spec);
queue.register('object.split', require('./derive').split);
queue.register('object.split', require('./derive').split);
queue.register('object.remux', require('./derive').remux);
queue.register('object.remux', require('./derive').remux);
+queue.register('vod.finalize', require('./vod-finalize').spec);
module.exports = { names: () => queue.typeNames() };
module.exports = { names: () => queue.typeNames() };
new file mode 100644
new file mode 100644
@@ -0,0 +1,170 @@
@@ -0,0 +1,170 @@
+/**
+ * OpenVibe.Media — job type vod.finalize (finalize lane): finalize a recording whose finalize failed
+ * or never ran, with backoff (docs/object-model.md#jobs).
+ *
+ * Queued by
+ * - the orphan sweep: a vods row still is_recording = 1 that no recorder, chunk upload or finalize
+ * holds and whose file has not grown for MEDIA_FINALIZE_ORPHAN_GRACE_S (default 180 s). At boot
+ * (every recording of the previous process) and every MEDIA_FINALIZE_SWEEP_S (120 s) after that.
+ * - finalize itself, when it could not settle a recording: nothing measurable (duration stored 0,
+ * needs_review), the file could not be stat'ed, or it threw. First retry after 5 minutes.
+ * - the v2 jobs API (POST /api/v2/:app/jobs { type: 'vod.finalize', params: { vod_id } }).
+ *
+ * params { vod_id } (derived from the object's legacy ref when omitted)
+ * result { outcome: ready | needs_review | corrupt | deleted | skipped, vod_id, duration_seconds?,
+ * duration_source?, health_status?, reason? }
+ * A run that still cannot measure the recording fails with a retryable code (unmeasurable,
+ * stat_failed, finalize_threw); the worker retries it (5 min, 15 min, 45 min, 2 h 15, then 6 h;
+ * 6 attempts). When the attempts run out the VOD stays needs_review, hidden, for a person.
+ * The job never stops a running recording and never runs while a chunk upload is still adding to it.
+ */
+'use strict';
+
+const fs = require('fs');
+const db = require('../db/database');
+const queue = require('./queue');
+
+const { JobError } = queue;
+const TYPE = 'vod.finalize';
+// Outcomes of a finalize that another attempt may fix (see server/vod/finalize.js FINALIZE_ISSUES).
+const RETRYABLE = { probe_failed: 'unmeasurable', inflated_duration: 'unmeasurable', stat_failed: 'stat_failed', finalize_failed: 'finalize_threw' };
+
+function envInt(name, fallback) {
+ const v = parseInt(process.env[name] || '', 10);
+ return Number.isFinite(v) && v >= 0 ? v : fallback;
+}
+const graceMs = () => envInt('MEDIA_FINALIZE_ORPHAN_GRACE_S', 180) * 1000;
+const sweepEveryMs = () => envInt('MEDIA_FINALIZE_SWEEP_S', 120) * 1000;
+
+function issuesOf(vod) {
+ try { const v = JSON.parse(vod.health_issues_json || '[]'); return Array.isArray(v) ? v : []; } catch { return []; }
+}
+
+function chunkUploadActive(vodId) {
+ try { return require('../vod/routes').activeChunkUploads.has(Number(vodId)); } catch { return false; }
+}
+
+/** Something in this process is still writing or finalizing the recording. */
+function busy(vodId) {
+ try { if (require('../vod/recorder').isRecording(vodId)) return 'recording'; } catch { /* */ }
+ try { if (require('../vod/finalize').isFinalizing(vodId)) return 'finalizing'; } catch { /* */ }
+ if (chunkUploadActive(vodId)) return 'chunk upload in progress';
+ return null;
+}
+
+function vodIdOf(obj, params = {}) {
+ if (params.vod_id != null) return Number(params.vod_id);
+ if (obj) {
+ const ref = require('../objects/model').parseLegacyRef(obj.legacy_ref);
+ if (ref && ref.kind === 'vod') return Number(ref.id);
+ }
+ return null;
+}
+
+/** API validation: the VOD must be this tenant's (and the named object's). */
+function validate({ appId, obj, params }) {
+ const id = vodIdOf(obj, params || {});
+ if (!Number.isInteger(id) || id < 1) throw new JobError('media.job.invalid', 'vod.finalize needs params.vod_id (or a vod object)', { permanent: true });
+ const vod = db.getVodById(id, appId);
+ if (!vod) throw new JobError('media.job.not_found', `vod ${id} not found`, { status: 404, permanent: true });
+ if (obj && vod.object_id && vod.object_id !== obj.id) throw new JobError('media.job.invalid', `the object is not vod ${id}`, { permanent: true });
+ return { vod_id: id };
+}
+
+/** Does this VOD still need a finalize? Returns the reason, or null. */
+function needsFinalize(vod) {
+ if (!vod) return null;
+ if (vod.is_recording) return 'orphan';
+ const retry = issuesOf(vod).find(i => RETRYABLE[i]);
+ return retry && vod.health_status === 'needs_review' ? retry : null;
+}
+
+async function run(job) {
+ const vodId = Number(job.params && job.params.vod_id);
+ const vod = db.getVodById(vodId, job.app_id);
+ if (!vod) return { outcome: 'skipped', vod_id: vodId, reason: 'the vod no longer exists' };
+ const holder = busy(vodId);
+ if (holder === 'finalizing') throw new JobError('busy', 'a finalize of this vod is already running', { retryAfterS: 60 });
+ if (holder) return { outcome: 'skipped', vod_id: vodId, reason: `${holder}: that path finalizes it` };
+ const why = needsFinalize(vod);
+ if (!why) return { outcome: 'skipped', vod_id: vodId, reason: 'already finalized', health_status: vod.health_status };
+
+ try {
+ await require('../vod/finalize').finalizeVod(vodId, { fromJob: true });
+ } catch (err) {
+ throw new JobError('finalize_threw', `finalize failed: ${err.message}`);
+ }
+ const after = db.getVodById(vodId, job.app_id);
+ if (!after) return { outcome: 'deleted', vod_id: vodId, reason: 'no media was ever written (empty or missing file)' };
+ if (after.is_recording) throw new JobError('busy', 'the vod is still marked recording after finalize', { retryAfterS: 60 });
+ const still = issuesOf(after).find(i => RETRYABLE[i]);
+ if (still && after.health_status === 'needs_review') {
+ throw new JobError(RETRYABLE[still], `vod ${vodId} still cannot be settled (${issuesOf(after).join(', ')})`);
+ }
+ return {
+ outcome: after.health_status === 'ok' ? 'ready' : after.health_status,
+ vod_id: vodId, duration_seconds: after.duration_seconds, duration_source: after.duration_source || null,
+ health_status: after.health_status,
+ };
+}
+
+/**
+ * Queue a finalize for this VOD (joins one that is already queued or running). Never throws;
+ * returns { job, created } or null. `reason` goes into created_by (system:vod.finalize:<reason>).
+ */
+function queueFinalize(vodOrId, reason, { runAfterS = 0 } = {}) {
+ try {
+ const vod = typeof vodOrId === 'object' ? vodOrId : db.getVodById(Number(vodOrId));
+ if (!vod) return null;
+ const r = queue.enqueue({
+ appId: vod.app_id, type: TYPE, objectId: vod.object_id || null, params: { vod_id: vod.id },
+ dedupeActive: true, createdBy: `system:${TYPE}:${reason}`, runAfterS,
+ });
+ if (r.created) {
+ console.log(`[Jobs] vod.finalize queued for vod ${vod.id} (${reason}${runAfterS ? `, in ${runAfterS} s` : ''})`);
+ try { require('./worker').kick(); } catch { /* next poll */ }
+ }
+ return { job: r.job, created: r.created };
+ } catch (err) {
+ console.warn(`[Jobs] could not queue vod.finalize for vod ${typeof vodOrId === 'object' ? vodOrId.id : vodOrId}: ${err.message}`);
+ return null;
+ }
+}
+
+/**
+ * Recordings left is_recording = 1 with nothing holding them. `graceMs`: the file must not have
+ * changed for this long (a chunk upload resuming after a restart keeps writing it). Returns ids.
+ */
+function orphans({ grace = graceMs(), now = Date.now() } = {}) {
+ const out = [];
+ // clips_only recordings too: their finalize discards them (file and row).
+ for (const row of db.all('SELECT id, file_path FROM vods WHERE is_recording = 1 ORDER BY id')) {
+ if (busy(row.id)) continue;
+ let mtime = 0;
+ try { mtime = row.file_path ? fs.statSync(row.file_path).mtimeMs : 0; } catch { mtime = 0; }
+ if (mtime && now - mtime < grace) continue;
+ out.push(row.id);
+ }
+ return out;
+}
+
+let lastSweepAt = 0;
+
+/** Queue vod.finalize for every orphan (due every MEDIA_FINALIZE_SWEEP_S; `force` = now). */
+function sweepOrphans({ force = false, now = Date.now(), grace } = {}) {
+ if (!force && now - lastSweepAt < sweepEveryMs()) return 0;
+ lastSweepAt = now;
+ let n = 0;
+ for (const id of orphans({ grace, now })) { const r = queueFinalize(id, 'orphan'); if (r && r.created) n++; }
+ return n;
+}
+
+// 5 min, 15 min, 45 min, 2 h 15, then 6 h.
+const backoffS = (attempt) => Math.min(6 * 3600, 300 * 3 ** Math.max(0, attempt - 1));
+
+module.exports = {
+ TYPE,
+ spec: { lane: 'finalize', maxAttempts: 6, timeoutMs: 60 * 60 * 1000, needsObject: false, backoffS, validate, run },
+ queueFinalize, sweepOrphans, orphans, needsFinalize, backoffS, RETRYABLE,
+ _resetSweep: () => { lastSweepAt = 0; },
+};
@@ -5,6 +5,8 @@
@@ -5,6 +5,8 @@
* light thumbnail.regenerate, invariant.scan MEDIA_JOBS_LIGHT_CONCURRENCY (2)
* light thumbnail.regenerate, invariant.scan MEDIA_JOBS_LIGHT_CONCURRENCY (2)
* heavy object.split, object.remux MEDIA_JOBS_HEAVY_CONCURRENCY (1); waits while a
* heavy object.split, object.remux MEDIA_JOBS_HEAVY_CONCURRENCY (1); waits while a
* recording runs unless MEDIA_JOBS_HEAVY_WHILE_RECORDING=1
* recording runs unless MEDIA_JOBS_HEAVY_WHILE_RECORDING=1
+ * finalize vod.finalize MEDIA_JOBS_FINALIZE_CONCURRENCY (1); runs while recording
+ * (it is the work the recorder does when a stream ends)
*
*
* A running job holds a lease (MEDIA_JOBS_LEASE_S) that a heartbeat renews; the heartbeat also notices
* A running job holds a lease (MEDIA_JOBS_LEASE_S) that a heartbeat renews; the heartbeat also notices
* an owner's cancel request. At start every job left `running` by the previous process is retried (or
* an owner's cancel request. At start every job left `running` by the previous process is retried (or
@@ -13,7 +15,8 @@
@@ -13,7 +15,8 @@
* from the checkpoint on a retry.
* from the checkpoint on a retry.
*
*
* Scheduling: the size-invariant validator (invariant.scan) is enqueued per tenant every
* Scheduling: the size-invariant validator (invariant.scan) is enqueued per tenant every
- * MEDIA_INVARIANT_SCAN_HOURS (it proposes split/remux jobs; it never runs them).
+ * MEDIA_INVARIANT_SCAN_HOURS (it proposes split/remux jobs; it never runs them). Orphaned recordings
+ * get a vod.finalize every MEDIA_FINALIZE_SWEEP_S (server/jobs/vod-finalize.js).
*/
*/
'use strict';
'use strict';
@@ -36,6 +39,7 @@ function lanes() {
@@ -36,6 +39,7 @@ function lanes() {
return [
return [
{ name: 'light', types: by('light'), max: cfg().lightConcurrency },
{ name: 'light', types: by('light'), max: cfg().lightConcurrency },
{ name: 'heavy', types: by('heavy'), max: cfg().heavyConcurrency },
{ name: 'heavy', types: by('heavy'), max: cfg().heavyConcurrency },
+ { name: 'finalize', types: by('finalize'), max: cfg().finalizeConcurrency },
];
];
}
}
@@ -141,6 +145,7 @@ async function tick() {
@@ -141,6 +145,7 @@ async function tick() {
queue.recoverInterrupted({ except: new Set(running.keys()) });
queue.recoverInterrupted({ except: new Set(running.keys()) });
const now = Date.now();
const now = Date.now();
if (now - lastScheduleAt > 10 * 60 * 1000) { lastScheduleAt = now; scheduleInvariantScans(now); }
if (now - lastScheduleAt > 10 * 60 * 1000) { lastScheduleAt = now; scheduleInvariantScans(now); }
+ try { require('./vod-finalize').sweepOrphans({ now }); } catch (err) { console.warn('[Jobs] orphan sweep:', err.message); }
if (now - lastPruneAt > 6 * 3600 * 1000) {
if (now - lastPruneAt > 6 * 3600 * 1000) {
lastPruneAt = now;
lastPruneAt = now;
try { const n = queue.prune({ days: cfg().retentionDays }); if (n) console.log(`[Jobs] pruned ${n} finished thumbnail job(s)`); } catch { /* next time */ }
try { const n = queue.prune({ days: cfg().retentionDays }); if (n) console.log(`[Jobs] pruned ${n} finished thumbnail job(s)`); } catch { /* next time */ }
@@ -126,6 +126,7 @@ function _commitVodOutcome(vod, event, change, row = () => db.getVodById(vod.id)
@@ -126,6 +126,7 @@ function _commitVodOutcome(vod, event, change, row = () => db.getVodById(vod.id)
* @param {number} [opts.startTimeMs] wall-clock recording start (for duration sanity)
* @param {number} [opts.startTimeMs] wall-clock recording start (for duration sanity)
* @param {boolean} [opts.ffmpegCorrupted] recorder flagged heavy corruption
* @param {boolean} [opts.ffmpegCorrupted] recorder flagged heavy corruption
* @param {string} [opts.segmentPath] trailing chunk segment not yet merged
* @param {string} [opts.segmentPath] trailing chunk segment not yet merged
+ * @param {boolean} [opts.fromJob] run by the vod.finalize job (which retries itself; queue no retry)
*/
*/
async function finalizeVod(vodId, opts = {}) {
async function finalizeVod(vodId, opts = {}) {
vodId = Number(vodId);
vodId = Number(vodId);
@@ -145,6 +146,10 @@ async function finalizeVod(vodId, opts = {}) {
@@ -145,6 +146,10 @@ async function finalizeVod(vodId, opts = {}) {
try {
try {
return await _doFinalize(vodId, opts);
return await _doFinalize(vodId, opts);
+ } catch (err) {
+ // A finalize that throws leaves the recording unsettled: the vod.finalize job tries again.
+ _queueRetry(vodId, 'finalize_failed', opts);
+ throw err;
} finally {
} finally {
_finalizing.delete(vodId);
_finalizing.delete(vodId);
// Every outcome (ready, quarantined, failed) re-projects the object; a deleted row was already marked by trigger.
// Every outcome (ready, quarantined, failed) re-projects the object; a deleted row was already marked by trigger.
@@ -152,6 +157,15 @@ async function finalizeVod(vodId, opts = {}) {
@@ -152,6 +157,15 @@ async function finalizeVod(vodId, opts = {}) {
}
}
}
}
+/**
+ * Queue a vod.finalize retry (server/jobs/vod-finalize.js; first attempt in 5 minutes), unless this
+ * finalize IS that job (it retries itself with backoff).
+ */
+function _queueRetry(vodId, reason, opts = {}) {
+ if (opts.fromJob) return;
+ try { require('../jobs/vod-finalize').queueFinalize(vodId, reason, { runAfterS: 300 }); } catch { /* the orphan sweep and the reconcile job remain */ }
+}
+
// Health issues finalize itself records when it cannot settle a recording. A later successful
// Health issues finalize itself records when it cannot settle a recording. A later successful
// finalize (the vod.finalize retry job) lifts the quarantine they caused.
// finalize (the vod.finalize retry job) lifts the quarantine they caused.
const FINALIZE_ISSUES = ['probe_failed', 'inflated_duration', 'stat_failed', 'finalize_failed'];
const FINALIZE_ISSUES = ['probe_failed', 'inflated_duration', 'stat_failed', 'finalize_failed'];
@@ -283,6 +297,7 @@ async function _doFinalize(vodId, opts) {
@@ -283,6 +297,7 @@ async function _doFinalize(vodId, opts) {
console.warn(`[VOD] vod ${vodId}: no measurable duration (${issues.join(', ')}) — stored 0, needs_review`);
console.warn(`[VOD] vod ${vodId}: no measurable duration (${issues.join(', ')}) — stored 0, needs_review`);
_commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review', health_issues_json = ?, probe_duration_seconds = 0, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
_commitVodOutcome(vod, 'vod.failed', () => db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review', health_issues_json = ?, probe_duration_seconds = 0, probe_format_json = ?, last_health_scan_at = datetime('now'), quarantined_at = datetime('now'), is_public = 0 WHERE id = ?`,
[JSON.stringify(issues), probeFormatJson, vodId]));
[JSON.stringify(issues), probeFormatJson, vodId]));
+ _queueRetry(vodId, issues.includes('probe_failed') ? 'probe_failed' : 'inflated_duration', opts);
return db.getVodById(vodId);
return db.getVodById(vodId);
}
}
@@ -304,6 +319,7 @@ async function _doFinalize(vodId, opts) {
@@ -304,6 +319,7 @@ async function _doFinalize(vodId, opts) {
console.error(`[VOD] Failed to stat finalized VOD ${vodId}:`, err.message);
console.error(`[VOD] Failed to stat finalized VOD ${vodId}:`, err.message);
db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',
db.run(`UPDATE vods SET is_recording = 0, duration_seconds = 0, duration_source = 'unknown', health_status = 'needs_review',
health_issues_json = ?, last_health_scan_at = datetime('now') WHERE id = ?`, [JSON.stringify(['stat_failed', ...issues]), vodId]);
health_issues_json = ?, last_health_scan_at = datetime('now') WHERE id = ?`, [JSON.stringify(['stat_failed', ...issues]), vodId]);
+ _queueRetry(vodId, 'stat_failed', opts);
return null;
return null;
}
}
// The lossless .master.mkv archive is only a fallback. Delete it ONLY once
// The lossless .master.mkv archive is only a fallback. Delete it ONLY once
new file mode 100644
new file mode 100644
@@ -0,0 +1,156 @@
@@ -0,0 +1,156 @@
+'use strict';
+// vod.finalize (server/jobs/vod-finalize.js): orphaned recordings (is_recording = 1, nothing holding
+// them, file idle past the grace) are queued once each; a finalize that cannot settle a recording
+// (nothing measurable, stat failure, a throw) queues a retry instead of just clearing is_recording;
+// the job retries with backoff, never stops a live recording or a chunk upload, and lifts the
+// quarantine once a later attempt measures the file.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+const tmp = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-media-finalize-job-'));
+const dir = (n) => { const d = path.join(tmp, n); fs.mkdirSync(d, { recursive: true }); return d; };
+process.env.DB_PATH = path.join(tmp, 'media.db');
+process.env.VOD_PATH = dir('vods');
+process.env.CLIPS_PATH = dir('clips');
+process.env.THUMBNAILS_PATH = dir('thumbnails');
+process.env.FILES_PATH = dir('files');
+process.env.PASTES_PATH = dir('pastes');
+process.env.OBJECTS_PATH = dir('objects');
+process.env.MEDIA_INVARIANT_SCAN_HOURS = '0';
+
+const db = require('../server/db/database');
+const queue = require('../server/jobs/queue');
+const worker = require('../server/jobs/worker');
+const job = require('../server/jobs/vod-finalize');
+const tools = require('../server/vod/media-tools');
+const recorder = require('../server/vod/recorder');
+const vodRoutes = require('../server/vod/routes');
+
+db.upsertApp({ app_id: 'live', api_key: 'live-key' });
+db.upsertApp({ app_id: 'games', api_key: 'games-key' });
+
+const realTools = { ...tools };
+const restore = () => Object.assign(tools, realTools);
+const measurable = (s = 1800) => {
+ tools.remuxForSeekingDetailed = async () => ({ ok: true, seconds: s, error: null });
+ tools.probeDuration = async () => ({ ok: true, seconds: s, format: {}, streams: [], error: null });
+};
+const unmeasurable = () => {
+ tools.remuxForSeekingDetailed = async () => ({ ok: false, seconds: 0, error: 'ffmpeg remux exited 1' });
+ tools.probeDuration = async () => ({ ok: false, seconds: 0, format: null, streams: [], error: 'unreadable' });
+ tools.streamCopyDuration = async () => ({ ok: false, seconds: 0, error: 'no packets' });
+};
+
+function makeVod(name, { idleS = 3600, app = 'live' } = {}) {
+ const file = path.join(process.env.VOD_PATH, name);
+ fs.writeFileSync(file, Buffer.alloc(30000, 9));
+ const t = (Date.now() - idleS * 1000) / 1000;
+ fs.utimesSync(file, t, t);
+ const r = db.run(`INSERT INTO vods (app_id, user_id, title, file_path, is_recording, duration_seconds, is_public, visibility, created_at)
+ VALUES (?, 1, 'rec', ?, 1, 77777, 1, 'public', datetime('now', '-2 days'))`, [app, file]);
+ return Number(r.lastInsertRowid);
+}
+const vod = (id) => db.get('SELECT * FROM vods WHERE id = ?', [id]);
+const jobsFor = (id) => db.all("SELECT * FROM media_jobs WHERE job_type = 'vod.finalize' AND json_extract(params, '$.vod_id') = ? ORDER BY id", [id]);
+
+(async () => {
+ assert.ok(queue.typeNames().includes('vod.finalize'));
+ assert.strictEqual(queue.typeSpec('vod.finalize').lane, 'finalize');
+
+ // ── 1. The orphan sweep: idle orphans only; live recordings, chunk uploads and fresh files are left alone ──
+ const idle = makeVod('idle.mp4');
+ const fresh = makeVod('fresh.mp4', { idleS: 10 });
+ const live = makeVod('live.mp4');
+ const chunked = makeVod('chunked.webm');
+ recorder.activeRecordings.set(live, { vodId: live });
+ vodRoutes.activeChunkUploads.set(chunked, { filePath: 'x' });
+ assert.deepStrictEqual(job.orphans(), [idle], 'only the idle orphan');
+ assert.strictEqual(job.sweepOrphans({ force: true }), 1);
+ assert.strictEqual(job.sweepOrphans({ force: true }), 0, 'a queued finalize is joined, not repeated');
+ const [j1] = jobsFor(idle);
+ assert.strictEqual(j1.created_by, 'system:vod.finalize:orphan');
+ assert.strictEqual(j1.app_id, 'live');
+ assert.ok(job.orphans({ grace: 1000 }).includes(fresh), 'a file idle past the grace is an orphan');
+ console.log('✅ orphan sweep: idle recordings nothing holds are queued once; live, chunk-upload and fresh ones are not');
+
+ // ── 2. Running it finalizes from the measured file ──
+ measurable(1800.2);
+ const done = await worker.runNow(j1.id);
+ assert.strictEqual(done.status, 'succeeded');
+ const res = JSON.parse(done.result);
+ assert.deepStrictEqual([res.outcome, res.duration_seconds, res.duration_source], ['ready', 1800, 'probe']);
+ assert.deepStrictEqual([vod(idle).is_recording, vod(idle).duration_seconds, vod(idle).health_status], [0, 1800, 'ok']);
+ console.log('✅ vod.finalize settles an orphan to its measured duration');
+
+ // ── 3. A finalize that cannot measure queues a retry (5 min out); the job retries with backoff ──
+ unmeasurable();
+ const bad = makeVod('bad.mp4');
+ await require('../server/vod/finalize').finalizeVod(bad);
+ assert.strictEqual(vod(bad).health_status, 'needs_review');
+ let js = jobsFor(bad);
+ assert.strictEqual(js.length, 1, 'one retry queued');
+ assert.strictEqual(js[0].created_by, 'system:vod.finalize:probe_failed');
+ const due = db.get("SELECT (julianday(run_after) - julianday('now')) * 86400 AS s FROM media_jobs WHERE id = ?", [js[0].id]).s;
+ assert.ok(due > 250 && due <= 301, `first retry in ~5 minutes (${due})`);
+ let r = await worker.runNow(js[0].id);
+ assert.strictEqual(r.status, 'queued', 'still unmeasurable: back in the queue');
+ assert.strictEqual(r.error_code, 'unmeasurable');
+ assert.strictEqual(r.attempts, 1);
+ assert.strictEqual(jobsFor(bad).length, 1, 'the job does not queue more of itself');
+ const wait = db.get("SELECT (julianday(run_after) - julianday('now')) * 86400 AS s FROM media_jobs WHERE id = ?", [r.id]).s;
+ assert.ok(wait > 250 && wait <= 301, `backoff after attempt 1 is 5 minutes (${wait})`);
+ assert.deepStrictEqual([job.backoffS(1), job.backoffS(2), job.backoffS(3), job.backoffS(6)], [300, 900, 2700, 21600]);
+ measurable(610);
+ r = await worker.runNow(r.id);
+ assert.strictEqual(r.status, 'succeeded');
+ const v = vod(bad);
+ assert.deepStrictEqual([v.duration_seconds, v.health_status, v.is_public, v.quarantined_at], [610, 'ok', 1, null], 'measured, un-quarantined');
+ console.log('✅ unmeasurable finalize → retry job (5 min), retried with backoff, settles once the file measures');
+
+ // ── 4. Stat failure and a thrown finalize queue a retry too (not just is_recording = 0) ──
+ const gone = makeVod('gone.mp4');
+ tools.probeDuration = async (p) => { if (p.endsWith('gone.mp4')) { try { fs.unlinkSync(p); } catch { /* */ } } return { ok: true, seconds: 300, format: {}, streams: [], error: null }; };
+ await require('../server/vod/finalize').finalizeVod(gone);
+ assert.deepStrictEqual(jobsFor(gone).map(j => j.created_by), ['system:vod.finalize:stat_failed']);
+ const thrower = makeVod('thrower.mp4');
+ tools.remuxForSeekingDetailed = async () => { throw new Error('boom'); };
+ await assert.rejects(require('../server/vod/finalize').finalizeVod(thrower), /boom/);
+ assert.deepStrictEqual(jobsFor(thrower).map(j => j.created_by), ['system:vod.finalize:finalize_failed']);
+ assert.strictEqual(vod(thrower).is_recording, 1, 'still unsettled, and the job knows it');
+ measurable(95);
+ const tr = await worker.runNow(jobsFor(thrower)[0].id);
+ assert.strictEqual(tr.status, 'succeeded');
+ assert.strictEqual(vod(thrower).duration_seconds, 95);
+ console.log('✅ a stat failure or a thrown finalize queues vod.finalize, which settles it later');
+
+ // ── 5. Never touches a live recording; out of attempts the VOD waits for a person ──
+ const jl = queue.enqueue({ appId: 'live', type: 'vod.finalize', params: { vod_id: live }, createdBy: 'test' }).job;
+ const lr = await worker.runNow(jl.id);
+ assert.strictEqual(JSON.parse(lr.result).outcome, 'skipped');
+ assert.strictEqual(vod(live).is_recording, 1, 'the recording keeps going');
+ recorder.activeRecordings.delete(live);
+ vodRoutes.activeChunkUploads.delete(chunked);
+
+ unmeasurable();
+ const last = makeVod('last.mp4');
+ const jlast = queue.enqueue({ appId: 'live', type: 'vod.finalize', params: { vod_id: last }, createdBy: 'test', maxAttempts: 1 }).job;
+ const out = await worker.runNow(jlast.id);
+ assert.deepStrictEqual([out.status, out.error_code], ['failed', 'unmeasurable']);
+ assert.deepStrictEqual([vod(last).is_recording, vod(last).duration_seconds, vod(last).health_status, vod(last).is_public], [0, 0, 'needs_review', 0]);
+ restore();
+ console.log('✅ a live recording is never finalized by the job; out of attempts the VOD stays needs_review, hidden');
+
+ // ── 6. API validation: the VOD must be the tenant's ──
+ const gv = makeVod('games.mp4', { app: 'games' });
+ assert.throws(() => job.spec.validate({ appId: 'live', obj: null, params: { vod_id: gv } }), (e) => e.code === 'media.job.not_found');
+ assert.deepStrictEqual(job.spec.validate({ appId: 'games', obj: null, params: { vod_id: gv } }), { vod_id: gv });
+ assert.throws(() => job.spec.validate({ appId: 'games', obj: null, params: {} }), (e) => e.code === 'media.job.invalid');
+ console.log('✅ vod.finalize through the jobs API is scoped to the tenant\'s own VODs');
+
+ db.close();
+ fs.rmSync(tmp, { recursive: true, force: true });
+ console.log('\n✅ All vod.finalize job tests passed');
+ process.exit(0);
+})().catch((err) => { console.error(err); process.exit(1); });