openvibers/openvibe.media Commits

Job fencing (WS-P lifecycle follow-up): succeed() and fail() matched status = 'running' only, with no owner token, so a worker that had lost its job (lease run out, recovery requeued it, another claim took it) could still overwrite it; safe only while one process runs the queue. Now a claim writes a fresh random lease_token on the row and hands it to the worker; renew, saveCheckpoint (with its alsoInTx writes, rolled back), succeed, fail and markCancelled match the id AND that token. A stale holder is refused, not applied: logged ([Jobs] refused a stale <action> of <id> ...), counted (queue.staleStats(), worker status stale_refused, media_job_stale_completions_total{action} on /metrics) and emitted on the queue bus; its heartbeat aborts the handler (media.job.lease_lost). Recovery takes a job over only with the token it read and, past boot, only while the lease is still expired; rows claimed before the column (NULL token) are still recovered at boot. Migration is additive only: media_jobs.lease_token TEXT in schema.sql and migrateColumns (ADD COLUMN); an older release ignores it. test/jobs-fencing.test.js: the column added to an older database; two claimants, the stale one's heartbeat, checkpoint, succeed, fail and cancel refused and counted while the current holder's result stands; a declined takeover; the worker aborting a job whose lease it lost

OP OpenVibers committed 26 Sept 2026 · a348ea716e · 8 files, +312 −49


Changes

ddiff --git a/README.md b/README.md
iindex e9b1e06..b535707 100644
---- a/README.md
++++ b/README.md
@@@ -346,7 +346,7 @@ never the headers (`test/trust-proxy.test.js`).
|---|---|
| `GET /healthz` | liveness: the process answers (unchanged; it checks nothing else) |
| `GET /api/ready` | readiness from real checks, each with `status`, `required`, `latency_ms`, `checked_at`. **Required** (503 when one fails): `db` (a query against the SQLite database), `storage_vods`, `storage_clips`, `storage_pastes`, `storage_thumbnails`, `storage_files`, `storage_objects` (a probe file is written and removed). **Optional** (listed in `degraded`, still 200): `network_jwks` (user JWTs and service tokens; app keys work without it), `remote_b2` / `remote_r2` (HeadBucket, at most once a minute; present only when the provider is configured), `events_outbox` (backlog over 1000 events). Also `recordings_in_progress`. |
--| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs). |
++| `GET /metrics` | Prometheus text (openvibe-shared/metrics) for **direct loopback callers only**; 404 through nginx. HTTP golden signals by route template, process metrics, `release_info`, `release_client_updates_total{outcome,reason}`, plus `media_recordings_in_progress`, `media_object_locations{provider,state}`, `media_objects{lifecycle_status}`, `media_events_outbox{status}` (only while the outbox runs), `media_job_stale_completions_total{action}` (job writes refused by the lease-token fence). |
| `GET /release.json` | release manifest (`registry.release-manifest@1` 1.1.0, openvibe-shared/release, ADR-016): deployed commit, package versions, components, contract ranges, `metrics_url`. The shared navbar's release-watch polls it. No components are declared, so every release still prompts open tabs to reload. |
| `POST /release-metrics` | open tabs' update reports (release-watch beacons, at most 4 KB; 30 a minute per connecting address, which behind nginx is the proxy, so one budget for all tabs; Sec-GPC/DNT dropped) into `release_client_updates_total` on `/metrics`. 403 in a restore drill, like every write. |
ddiff --git a/docs/object-model.md b/docs/object-model.md
iindex b8ec6cb..a6c2cfa 100644
---- a/docs/object-model.md
++++ b/docs/object-model.md
@@@ -20,7 +20,7 @@ Code: `server/objects/` (`model.js`, `routes.js`, `backfill.js`, `reconcile.js`,
| `media_locations` | one row per provider copy (`UNIQUE(object_id, provider)`): `provider` (`local` `b2` `r2`), `bucket`, `key` (absolute path for local, object key for B2/R2), `storage_class` (`hot` local, `cold` B2, `cache` R2), `state`, `checksum`, `size_bytes`, `verified_at` |
| `media_relationships` | `from_object_id`, `relation`, `to_object_id`, `metadata`. In use: `clip_of` (clip to vod, with start/end), `thumbnail_of` (thumbnail to vod/clip). Reserved: `derived_from`, `screenshot_of` |
| `media_variants` | `object_id`, `variant_name`, `derived_object_id`, `recipe`. In use: `thumbnail` |
--| `media_jobs` | derivative/maintenance jobs: `id` (mjob_…), `app_id`, `object_id`, `job_type`, `status`, `idempotency_key`, `params`, `result`, `attempts`/`max_attempts`, `run_after`, `lease_until`, `checkpoint`, `error`/`error_code`, `cancel_requested`, `created_by`, `decided_by`. See [Jobs](#jobs) |
++| `media_jobs` | derivative/maintenance jobs: `id` (mjob_…), `app_id`, `object_id`, `job_type`, `status`, `idempotency_key`, `params`, `result`, `attempts`/`max_attempts`, `run_after`, `lease_until`, `lease_token`, `checkpoint`, `error`/`error_code`, `cancel_requested`, `created_by`, `decided_by`. See [Jobs](#jobs) |
| `media_uploads`, `media_upload_parts` | multipart upload sessions and the parts received (size, sha256). See [Multipart uploads](#multipart-uploads) |
| `media_holds` | retention holds (with a staff `note`), see [Holds](#retention-holds) |
| `media_invariant_violations` | public playback objects over the size policy, see [Invariant](#public-object-size-invariant) |
@@@ -590,6 +590,13 @@ 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
requeued (or failed when out of attempts). Handlers checkpoint progress and resume from it.
++**Fencing:** a claim writes a fresh random `lease_token` on the row and hands it to its holder; renewing
++the lease, saving a checkpoint and every way out of `running` (succeed, fail, cancel) match the job id
++**and** that token. A holder that lost the job (its lease ran out, recovery requeued it, maybe another
++claim took it) is refused instead of overwriting: its heartbeat aborts the handler, and its checkpoint
++(with whatever it wrote in the same transaction) and its completion are refused, logged as
++`[Jobs] refused a stale …` and counted in `media_job_stale_completions_total{action}`. Recovery takes a
++job over only with the token it read and, past boot, only while the lease is still expired.
`MEDIA_JOBS_ENABLED=0` stops the worker. Finished thumbnail and `object.hash` jobs are pruned after
`MEDIA_JOBS_RETENTION_DAYS`; other jobs are kept.
ddiff --git a/server/db/database.js b/server/db/database.js
iindex 0ebe415..f2c9370 100644
---- a/server/db/database.js
++++ b/server/db/database.js
@@@ -66,6 +66,8 @@ function migrateColumns() {
// Developer-project tenants (ADR-014): project_id prj_<ULID>, env sandbox|production. NULL on first-party tenants.
apps: [['project_id', 'TEXT'], ['env', 'TEXT']],
media_holds: [['note', 'TEXT']],
++ // Job fencing (server/jobs/queue.js): a claim's token; NULL on rows claimed before it existed.
++ media_jobs: [['lease_token', 'TEXT']],
};
for (const [table, cols] of Object.entries(wanted)) {
const existing = database.prepare(`PRAGMA table_info(${table})`).all().map(c => c.name);
ddiff --git a/server/db/schema.sql b/server/db/schema.sql
iindex 540d46a..8e2ebbc 100644
---- a/server/db/schema.sql
++++ b/server/db/schema.sql
@@@ -293,6 +293,7 @@ CREATE TABLE IF NOT EXISTS media_jobs (
max_attempts INTEGER NOT NULL DEFAULT 3,
run_after DATETIME, -- queued: not before (retry backoff)
lease_until DATETIME, -- running: renewed while the worker holds the job
++ lease_token TEXT, -- running: the claim's random fencing token; renew/checkpoint/succeed/fail must match it
checkpoint TEXT, -- handler progress (JSON); a retry resumes from it
error TEXT,
error_code TEXT, -- stable code of the last failure (media_unavailable, quota_exceeded, …)
ddiff --git a/server/jobs/queue.js b/server/jobs/queue.js
iindex e889f48..29ac6b8 100644
---- a/server/jobs/queue.js
++++ b/server/jobs/queue.js
@@@ -5,13 +5,21 @@
* proposed waiting for its owner to decide (approve -> queued, cancel -> cancelled). The worker never
* picks it up: the size-invariant validator proposes split/remux jobs this way.
* queued due when run_after has passed
-- * running held by the worker (lease_until renewed while it runs)
++ * running held by the worker (lease_until renewed while it runs; lease_token names the claim)
* succeeded | failed | cancelled finished
*
* Every state change and its media.job.<transition> event commit in ONE SQLite transaction (the same
* rule as webhooks.announce(), see events.js): an event exists if and only if its change committed.
* The relay is woken after the commit. Job events go to OpenVibe.Events only (no app webhook).
*
++ * Fencing: a claim writes a fresh random lease_token on the row and hands it to its holder. Renewing
++ * the lease, saving a checkpoint and every way out of 'running' (succeed, fail, markCancelled) match
++ * the job id AND that token, so a holder that lost the job (its lease ran out and recovery requeued
++ * it, and maybe another claim took it) can no longer write it: its completion is refused, logged and
++ * counted (staleStats(), the bus event 'stale', media_job_stale_completions_total) instead of
++ * overwriting what the current holder does. Rows claimed before the column existed have a NULL token,
++ * which only a NULL token matches (boot recovery reads the row's token and passes it back).
++ *
* Idempotency: a key is unique per tenant. A repeat with the same key and the same request (type,
* object, params) answers with the job it made; a different request under a used key is a conflict.
* `dedupeActive` instead joins an identical job that is still queued or running (thumbnail requests).
@@@ -182,10 +190,42 @@ function commit(id, transition, change) {
return row;
}
--function update(id, fromStatuses, sets, params = []) {
++/**
++ * `fence`: { token } adds AND lease_token IS <token> (a NULL token matches only an unleased row);
++ * { expired: true } also requires the lease to have run out (recovery taking a job over).
++ */
++function update(id, fromStatuses, sets, params = [], fence = null) {
const from = Array.isArray(fromStatuses) ? fromStatuses : [fromStatuses];
-- return db.run(`UPDATE media_jobs SET ${sets}, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status IN (${from.map(() => '?').join(', ')})`,
-- [...params, id, ...from]).changes > 0;
++ let where = `id = ? AND status IN (${from.map(() => '?').join(', ')})`;
++ const args = [...params, id, ...from];
++ if (fence) {
++ where += ' AND lease_token IS ?';
++ args.push(fence.token == null ? null : String(fence.token));
++ if (fence.expired) where += " AND (lease_until IS NULL OR lease_until < datetime('now'))";
++ }
++ return db.run(`UPDATE media_jobs SET ${sets}, updated_at = CURRENT_TIMESTAMP WHERE ${where}`, args).changes > 0;
++}
++
++// ── Fencing ──────────────────────────────────────────────────
++
++const stale = { succeed: 0, fail: 0, cancel: 0, renew: 0, checkpoint: 0 };
++function newLeaseToken() { return crypto.randomBytes(16).toString('hex'); }
++
++/** A holder's write was refused because its claim no longer holds the job: log, count, tell listeners. */
++function refuseStale(action, id) {
++ stale[action] = (stale[action] || 0) + 1;
++ const now = get(id);
++ const where = !now ? 'it no longer exists' : now.status === 'running' ? 'another claim holds it' : `it is ${now.status} now`;
++ console.warn(`[Jobs] refused a stale ${action} of ${id} from a holder that lost it (${where})`);
++ bus.emit('stale', { action, id, status: now ? now.status : null });
++}
++/** Stale writes refused since boot, by action. */
++function staleStats() { return { ...stale }; }
++
++/** The fence a worker-side call passes: its claim's token (required: undefined is refused as stale). */
++function holder(opts, action, id) {
++ if (!opts || !Object.prototype.hasOwnProperty.call(opts, 'token')) throw new Error(`queue.${action}(${id}) needs the claim's { token }`);
++ return { token: opts.token, expired: !!opts.expired, takeover: !!opts.takeover };
}
// ── Enqueue ──────────────────────────────────────────────────
@@@ -272,8 +312,8 @@ function cancel(id, { by = null, reason = null } = {}) {
// ── Worker side ──────────────────────────────────────────────
/**
-- * Take the oldest due queued job of these types (or exactly `id`): running, attempts + 1, leased.
-- * Returns the row or null.
++ * Take the oldest due queued job of these types (or exactly `id`): running, attempts + 1, leased, with
++ * a fresh lease_token. Returns the row (its lease_token is the holder's fence) or null.
*/
function claim(types, { leaseS = 120, id = null } = {}) {
if (!id && !types.length) return null;
@@@ -285,7 +325,7 @@ function claim(types, { leaseS = 120, id = null } = {}) {
AND (run_after IS NULL OR run_after <= datetime('now')) ORDER BY id LIMIT 1`, types);
if (!next) return;
if (!update(next.id, 'queued', `status = 'running', attempts = attempts + 1, started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
-- lease_until = datetime('now', '+${Math.round(leaseS)} seconds'), run_after = NULL`)) return;
++ lease_until = datetime('now', '+${Math.round(leaseS)} seconds'), run_after = NULL, lease_token = ?`, [newLeaseToken()])) return;
row = get(next.id);
events.recordJob('started', jobEvent(row));
})();
@@@ -293,45 +333,83 @@ function claim(types, { leaseS = 120, id = null } = {}) {
return row;
}
--/** Extend a running job's lease. Returns { cancelRequested } (false when it is no longer running). */
--function renew(id, { leaseS = 120 } = {}) {
-- db.run(`UPDATE media_jobs SET lease_until = datetime('now', '+${Math.round(leaseS)} seconds') WHERE id = ? AND status = 'running'`, [id]);
++/**
++ * Extend a running job's lease, for the holder of `token` only. Returns { running, held, cancelRequested }:
++ * held false = this claim lost the job (refused as stale; the worker aborts its handler).
++ */
++function renew(id, opts = {}) {
++ const fence = holder(opts, 'renew', id);
++ const leaseS = opts.leaseS == null ? 120 : opts.leaseS;
++ const held = db.run(`UPDATE media_jobs SET lease_until = datetime('now', '+${Math.round(leaseS)} seconds')
++ WHERE id = ? AND status = 'running' AND lease_token IS ?`, [id, fence.token == null ? null : String(fence.token)]).changes > 0;
++ if (!held) refuseStale('renew', id);
const row = get(id);
-- return { running: !!row && row.status === 'running', cancelRequested: !!(row && row.cancel_requested) };
++ return { running: !!row && row.status === 'running', held, cancelRequested: !!(row && row.cancel_requested) };
}
--/** Save handler progress; `alsoInTx()` (optional) runs in the same transaction (e.g. the object a part became). */
--function saveCheckpoint(id, checkpoint, alsoInTx = null) {
-- db.getDb().transaction(() => {
-- if (alsoInTx) alsoInTx();
-- db.run('UPDATE media_jobs SET checkpoint = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', [JSON.stringify(checkpoint), id]);
-- })();
++/**
++ * Save handler progress; `alsoInTx()` (optional) runs in the same transaction (e.g. the object a part
++ * became). With `{ token }` (the worker) only the job's holder may: a stale holder's checkpoint and
++ * everything alsoInTx did roll back, and it throws media.job.lease_lost.
++ */
++function saveCheckpoint(id, checkpoint, alsoInTx = null, opts = null) {
++ const fenced = !!(opts && Object.prototype.hasOwnProperty.call(opts, 'token'));
++ let lost = false;
++ try {
++ db.getDb().transaction(() => {
++ if (alsoInTx) alsoInTx();
++ const changes = fenced
++ ? db.run("UPDATE media_jobs SET checkpoint = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND status = 'running' AND lease_token IS ?",
++ [JSON.stringify(checkpoint), id, opts.token == null ? null : String(opts.token)]).changes
++ : db.run('UPDATE media_jobs SET checkpoint = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?', [JSON.stringify(checkpoint), id]).changes;
++ if (fenced && !changes) { lost = true; throw new JobError('media.job.lease_lost', 'This worker no longer holds the job', { permanent: true, status: 409 }); }
++ })();
++ } catch (err) {
++ if (lost) refuseStale('checkpoint', id);
++ throw err;
++ }
}
--function succeed(id, result) {
-- return commit(id, 'succeeded', () => update(id, 'running',
-- "status = 'succeeded', result = ?, error = NULL, error_code = NULL, lease_until = NULL, finished_at = CURRENT_TIMESTAMP",
-- [JSON.stringify(result ?? null)]));
++/** The holder's result: running → succeeded. { token } is the claim's; a stale holder is refused (null). */
++function succeed(id, result, opts) {
++ const fence = holder(opts, 'succeed', id);
++ const row = commit(id, 'succeeded', () => update(id, 'running',
++ "status = 'succeeded', result = ?, error = NULL, error_code = NULL, lease_until = NULL, lease_token = NULL, finished_at = CURRENT_TIMESTAMP",
++ [JSON.stringify(result ?? null)], fence));
++ if (!row && !fence.takeover) refuseStale('succeed', id);
++ return row;
}
--/** A failed attempt: queued again after `retryInS` (media.job.retrying), or failed for good. */
--function fail(id, { message, code = null, retryInS = null, result = null }) {
++/**
++ * A failed attempt: queued again after `retryInS` (media.job.retrying), or failed for good. { token } is
++ * the claim's; a stale holder is refused (null). Recovery passes the token it read with { takeover: true }
++ * (and { expired: true } past boot: only while the lease is still run out).
++ */
++function fail(id, { message, code = null, retryInS = null, result = null, ...opts }) {
++ const fence = holder(opts, 'fail', id);
const msg = String(message || 'failed').slice(0, 1000);
++ let row;
if (retryInS != null) {
-- return commit(id, 'retrying', () => update(id, 'running',
-- `status = 'queued', error = ?, error_code = ?, lease_until = NULL, run_after = datetime('now', '+${Math.max(0, Math.round(retryInS))} seconds')`,
-- [msg, code]));
++ row = commit(id, 'retrying', () => update(id, 'running',
++ `status = 'queued', error = ?, error_code = ?, lease_until = NULL, lease_token = NULL, run_after = datetime('now', '+${Math.max(0, Math.round(retryInS))} seconds')`,
++ [msg, code], fence));
++ } else {
++ row = commit(id, 'failed', () => update(id, 'running',
++ "status = 'failed', error = ?, error_code = ?, result = COALESCE(?, result), lease_until = NULL, lease_token = NULL, finished_at = CURRENT_TIMESTAMP",
++ [msg, code, result == null ? null : JSON.stringify(result)], fence));
}
-- return commit(id, 'failed', () => update(id, 'running',
-- "status = 'failed', error = ?, error_code = ?, result = COALESCE(?, result), lease_until = NULL, finished_at = CURRENT_TIMESTAMP",
-- [msg, code, result == null ? null : JSON.stringify(result)]));
++ if (!row && !fence.takeover) refuseStale('fail', id);
++ return row;
}
--/** A running job that stopped because its owner cancelled it. */
--function markCancelled(id, { result = null } = {}) {
-- return commit(id, 'cancelled', () => update(id, 'running',
-- "status = 'cancelled', error = 'cancelled by its owner', error_code = 'cancelled', result = COALESCE(?, result), lease_until = NULL, finished_at = CURRENT_TIMESTAMP",
-- [result == null ? null : JSON.stringify(result)]));
++/** A running job that stopped because its owner cancelled it. { token } as for fail(). */
++function markCancelled(id, { result = null, ...opts } = {}) {
++ const fence = holder(opts, 'cancel', id);
++ const row = commit(id, 'cancelled', () => update(id, 'running',
++ "status = 'cancelled', error = 'cancelled by its owner', error_code = 'cancelled', result = COALESCE(?, result), lease_until = NULL, lease_token = NULL, finished_at = CURRENT_TIMESTAMP",
++ [result == null ? null : JSON.stringify(result)], fence));
++ if (!row && !fence.takeover) refuseStale('cancel', id);
++ return row;
}
/**
@@@ -344,10 +422,14 @@ function recoverInterrupted({ all = false, except = new Set(), retryInS = 30 } =
for (const row of rows) {
if (except.has(row.id)) continue;
const why = all ? 'interrupted by a restart' : 'the worker stopped renewing its lease';
-- if (row.cancel_requested) markCancelled(row.id);
-- else if (row.attempts < row.max_attempts) fail(row.id, { message: why, code: 'interrupted', retryInS });
-- else fail(row.id, { message: why, code: 'interrupted' });
-- n++;
++ // A takeover, fenced on the token it read (and, past boot, on the lease still being expired):
++ // a holder that renewed or finished meanwhile keeps its job.
++ const fence = { token: row.lease_token, takeover: true, expired: !all };
++ let done;
++ if (row.cancel_requested) done = markCancelled(row.id, fence);
++ else if (row.attempts < row.max_attempts) done = fail(row.id, { message: why, code: 'interrupted', retryInS, ...fence });
++ else done = fail(row.id, { message: why, code: 'interrupted', ...fence });
++ if (done) n++;
}
return n;
}
@@@ -401,5 +483,5 @@ module.exports = {
register, typeSpec, typeNames, requestHash, parseJson,
get, getForApp, jobPublic, jobEvent, list, counts,
enqueue, approve, cancel,
-- claim, renew, saveCheckpoint, succeed, fail, markCancelled, recoverInterrupted, waitFor, prune,
++ claim, renew, saveCheckpoint, succeed, fail, markCancelled, recoverInterrupted, waitFor, prune, staleStats,
};
ddiff --git a/server/jobs/worker.js b/server/jobs/worker.js
iindex a56c973..1d7dd5e 100644
---- a/server/jobs/worker.js
++++ b/server/jobs/worker.js
@@@ -11,7 +11,9 @@
* clip live streams and wait for the result)
*
* 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. Every write about the job carries the claim's lease_token (queue.js,
++ * Fencing): a heartbeat that finds the job taken from it aborts the handler, and whatever it reports
++ * then is refused as stale. At start every job left `running` by the previous process is retried (or
* failed when out of attempts). A failed attempt is retried with backoff unless the handler says the
* failure is permanent. Handlers get { signal, checkpoint, saveCheckpoint(cp, alsoInTx) } and resume
* from the checkpoint on a retry.
@@@ -68,10 +70,12 @@ async function execute(row, laneName) {
const ac = new AbortController();
running.set(row.id, { ac, lane: laneName });
const leaseS = cfg().leaseS;
++ const token = row.lease_token;
let cancelRequested = !!row.cancel_requested;
const beat = setInterval(() => {
try {
-- const r = queue.renew(row.id, { leaseS });
++ const r = queue.renew(row.id, { leaseS, token });
++ if (!r.held) { clearInterval(beat); ac.abort(new queue.JobError('media.job.lease_lost', 'this worker no longer holds the job', { permanent: true })); return; }
if (r.cancelRequested && !cancelRequested) { cancelRequested = true; ac.abort(new Error('cancelled by its owner')); }
} catch { /* next beat */ }
}, Math.max(1000, Math.floor(leaseS * 1000 / 3)));
@@@ -85,24 +89,24 @@ async function execute(row, laneName) {
signal: ac.signal,
checkpoint: queue.parseJson(row.checkpoint, null),
attempt: row.attempts,
-- saveCheckpoint: (cp, alsoInTx) => { ctx.checkpoint = cp; queue.saveCheckpoint(row.id, cp, alsoInTx); },
++ saveCheckpoint: (cp, alsoInTx) => { queue.saveCheckpoint(row.id, cp, alsoInTx, { token }); ctx.checkpoint = cp; },
cancelled: () => cancelRequested,
};
try {
if (!spec) throw new queue.JobError('media.job.unknown_type', `No handler for ${row.job_type}`, { permanent: true });
const result = await spec.run(queue.jobPublic(row), ctx);
-- queue.succeed(row.id, result);
++ queue.succeed(row.id, result, { token });
} catch (err) {
const fresh = queue.get(row.id);
if (cancelRequested || (fresh && fresh.cancel_requested)) {
-- queue.markCancelled(row.id, { result: ctx.checkpoint ? { checkpoint: ctx.checkpoint } : null });
++ queue.markCancelled(row.id, { result: ctx.checkpoint ? { checkpoint: ctx.checkpoint } : null, token });
} else {
const permanent = !!(err && err.permanent);
const attempts = fresh ? fresh.attempts : row.attempts;
const more = !permanent && attempts < (fresh ? fresh.max_attempts : row.max_attempts);
const retryIn = err && err.retryAfterS != null ? err.retryAfterS : backoffS(spec || {}, attempts);
-- queue.fail(row.id, { message: (err && err.message) || String(err), code: (err && err.code) || null, retryInS: more ? retryIn : null });
-- if (!more) console.warn(`[Jobs] ${row.job_type} ${row.id} failed: ${(err && err.message) || err}`);
++ const failed = queue.fail(row.id, { message: (err && err.message) || String(err), code: (err && err.code) || null, retryInS: more ? retryIn : null, token });
++ if (failed && !more) console.warn(`[Jobs] ${row.job_type} ${row.id} failed: ${(err && err.message) || err}`);
}
} finally {
clearInterval(beat);
@@@ -211,7 +215,7 @@ function stop() {
}
function status() {
-- return { enabled: cfg().enabled, started, running: [...running.entries()].map(([id, r]) => ({ id, lane: r.lane })), counts: queue.counts() };
++ return { enabled: cfg().enabled, started, running: [...running.entries()].map(([id, r]) => ({ id, lane: r.lane })), counts: queue.counts(), stale_refused: queue.staleStats() };
}
module.exports = { start, stop, kick, tick, abort, runNow, execute, status, scheduleInvariantScans, isStarted: () => started, _running: running };
ddiff --git a/server/observability.js b/server/observability.js
iindex e35b385..fe3822d 100644
---- a/server/observability.js
++++ b/server/observability.js
@@@ -85,6 +85,11 @@ function domainMetrics(registry, { db, recorder, events }) {
name: 'media_jobs', help: 'Media jobs by type and status (proposed = waiting for the owner; see scripts/media-jobs.js)', labelNames: ['type', 'status'],
collect: () => db.all('SELECT job_type, status, COUNT(*) AS n FROM media_jobs GROUP BY job_type, status').map((r) => ({ labels: { type: r.job_type, status: r.status }, value: r.n })),
});
++ // Job writes the lease-token fence refused (server/jobs/queue.js): a holder that lost its job tried to
++ // renew, checkpoint or finish it. Anything above zero means two claimants overlapped.
++ const stale = registry.counter({ name: 'media_job_stale_completions_total', help: 'Media job writes refused because the writer no longer held the job (lease-token fencing), by action', labelNames: ['action'] });
++ for (const action of ['succeed', 'fail', 'cancel', 'renew', 'checkpoint']) stale.inc({ action }, 0);
++ require('./jobs/queue').bus.on('stale', (e) => stale.inc({ action: e.action }));
registry.gauge({
name: 'media_uploads_open', help: 'Multipart upload sessions still open (active or completing)',
collect: () => db.get("SELECT COUNT(*) AS n FROM media_uploads WHERE status IN ('active', 'completing')").n,
ddiff --git a/test/jobs-fencing.test.js b/test/jobs-fencing.test.js
nnew file mode 100644
iindex 0000000..e68e3f2
---- /dev/null
++++ b/test/jobs-fencing.test.js
@@@ -0,0 +1,162 @@
++'use strict';
++// Job fencing (server/jobs/queue.js, WS-P lifecycle follow-up): a claim writes a random lease_token on the
++// job row; renew, checkpoint, succeed, fail and markCancelled match the id AND that token. Two claimants:
++// A's lease runs out, recovery requeues the job and B claims it; A's late heartbeat, checkpoint and
++// completion are refused (logged, counted, metric), B's win. The column is added to an older database
++// (ADD COLUMN only), and a job a pre-fencing process left running (NULL token) is still recovered at boot.
++// The worker aborts a handler whose heartbeat finds the job taken, and its report is refused.
++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-fencing-'));
++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.FILES_PATH = dir('files');
++process.env.THUMBNAILS_PATH = dir('thumbnails');
++process.env.PASTES_PATH = dir('pastes');
++process.env.OBJECTS_PATH = dir('objects');
++process.env.MEDIA_JOBS_POLL_MS = '50';
++process.env.MEDIA_INVARIANT_SCAN_HOURS = '0';
++
++const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
++const waitFor = async (fn, ms = 8000) => {
++ const end = Date.now() + ms;
++ while (Date.now() < end) { if (await fn()) return true; await sleep(20); }
++ return false;
++};
++
++(async () => {
++ const db = require('../server/db/database');
++ const raw = () => db.getDb();
++
++ // ── 1. Additive migration: an older database gains lease_token; its NULL-token running job is recovered ──
++ raw();
++ db.close();
++ {
++ const Database = require('better-sqlite3');
++ const old = new Database(process.env.DB_PATH);
++ old.exec('ALTER TABLE media_jobs DROP COLUMN lease_token');
++ old.exec(`INSERT INTO media_jobs (id, app_id, job_type, status, attempts, max_attempts, lease_until)
++ VALUES ('mjob_prefence', 'live', 'test.fence', 'running', 1, 3, datetime('now', '+1 hour'))`);
++ old.close();
++ }
++ const cols = raw().prepare('PRAGMA table_info(media_jobs)').all().map((c) => c.name);
++ assert.ok(cols.includes('lease_token'), 'lease_token is added to an older media_jobs');
++ assert.strictEqual(db.get("SELECT lease_token FROM media_jobs WHERE id = 'mjob_prefence'").lease_token, null);
++
++ const queue = require('../server/jobs/queue');
++ queue.register('test.fence', { lane: 'light', maxAttempts: 3, run: async () => ({ ok: true }) });
++ assert.strictEqual(queue.recoverInterrupted({ all: true }), 1, 'a job a pre-fencing process left running (NULL token) is recovered at boot');
++ assert.deepStrictEqual([queue.get('mjob_prefence').status, queue.get('mjob_prefence').error_code], ['queued', 'interrupted']);
++ queue.cancel('mjob_prefence', { by: 'test' });
++ console.log('✅ lease_token is ADD COLUMN on an older database; a NULL-token running job is still recovered at boot');
++
++ // ── 2. Two claimants ──
++ const refused = [];
++ queue.bus.on('stale', (e) => refused.push(e));
++ const metrics = require('openvibe-shared/metrics').createRegistry();
++ require('../server/observability').domainMetrics(metrics, { db, recorder: { activeCount: () => 0 }, events: require('../server/events') });
++ const metric = (action) => metrics.getMetric('media_job_stale_completions_total').get({ action });
++ const warnings = [];
++ const warn = console.warn;
++ console.warn = (...a) => { warnings.push(a.join(' ')); };
++
++ const changes = [];
++ queue.bus.on('change', (row) => changes.push({ id: row.id, status: row.status }));
++ const { job } = queue.enqueue({ appId: 'live', type: 'test.fence', createdBy: 'test' });
++ const a = queue.claim(['test.fence'], { leaseS: 30 });
++ assert.strictEqual(a.id, job.id);
++ assert.match(a.lease_token, /^[0-9a-f]{32}$/, 'a claim writes a random token');
++ assert.ok(!('lease_token' in queue.jobPublic(a)), 'the token never leaves the service');
++ assert.strictEqual(queue.renew(job.id, { leaseS: 30, token: a.lease_token }).held, true, 'the holder renews');
++
++ // A stalls: its lease runs out, recovery takes the job over, B claims it.
++ raw().prepare("UPDATE media_jobs SET lease_until = datetime('now', '-1 minute') WHERE id = ?").run(job.id);
++ assert.strictEqual(queue.recoverInterrupted({ except: new Set() }), 1);
++ assert.deepStrictEqual([queue.get(job.id).status, queue.get(job.id).lease_token], ['queued', null]);
++ raw().prepare('UPDATE media_jobs SET run_after = NULL WHERE id = ?').run(job.id);
++ const b = queue.claim(['test.fence'], { leaseS: 30 });
++ assert.strictEqual(b.id, job.id);
++ assert.notStrictEqual(b.lease_token, a.lease_token, 'the second claim has its own token');
++ assert.strictEqual(b.attempts, 2);
++
++ // A comes back: everything it writes is refused.
++ assert.strictEqual(queue.renew(job.id, { leaseS: 30, token: a.lease_token }).held, false, "A's heartbeat is refused");
++ raw().exec('CREATE TABLE IF NOT EXISTS fence_probe (x TEXT)');
++ assert.throws(() => queue.saveCheckpoint(job.id, { by: 'A' }, () => raw().prepare("INSERT INTO fence_probe VALUES ('A')").run(), { token: a.lease_token }),
++ (err) => err.code === 'media.job.lease_lost');
++ assert.strictEqual(raw().prepare('SELECT COUNT(*) AS n FROM fence_probe').get().n, 0, "what A's checkpoint wrote alongside rolled back");
++ assert.strictEqual(queue.succeed(job.id, { by: 'A' }, { token: a.lease_token }), null, "A's completion is refused");
++ assert.strictEqual(queue.fail(job.id, { message: 'A gave up', token: a.lease_token }), null, "so is A's failure");
++ assert.strictEqual(queue.markCancelled(job.id, { token: a.lease_token }), null, 'and its cancel');
++ let now = queue.get(job.id);
++ assert.deepStrictEqual([now.status, now.result, now.checkpoint, now.lease_token], ['running', null, null, b.lease_token], "B's job is untouched");
++ assert.throws(() => queue.succeed(job.id, { by: '?' }), /needs the claim's \{ token \}/, 'a completion without a token is a programming error');
++
++ // B finishes; A's late completion is refused again and B's result stands.
++ assert.strictEqual(queue.renew(job.id, { leaseS: 30, token: b.lease_token }).held, true);
++ queue.saveCheckpoint(job.id, { by: 'B' }, null, { token: b.lease_token });
++ const done = queue.succeed(job.id, { by: 'B' }, { token: b.lease_token });
++ assert.deepStrictEqual([done.status, JSON.parse(done.result), done.lease_token], ['succeeded', { by: 'B' }, null]);
++ assert.strictEqual(queue.succeed(job.id, { by: 'A' }, { token: a.lease_token }), null);
++ now = queue.get(job.id);
++ assert.deepStrictEqual([now.status, JSON.parse(now.result)], ['succeeded', { by: 'B' }], "B's result stands");
++ assert.deepStrictEqual(changes.filter((r) => r.id === job.id).map((r) => r.status), ['queued', 'running', 'queued', 'running', 'succeeded'],
++ 'claimed by A, taken back by recovery, claimed by B, succeeded once (B); nothing from A');
++
++ assert.deepStrictEqual(queue.staleStats(), { succeed: 2, fail: 1, cancel: 1, renew: 1, checkpoint: 1 }, 'every refusal is counted');
++ assert.deepStrictEqual(refused.map((e) => e.action), ['renew', 'checkpoint', 'succeed', 'fail', 'cancel', 'succeed']);
++ assert.deepStrictEqual(['succeed', 'fail', 'cancel', 'renew', 'checkpoint'].map(metric), [2, 1, 1, 1, 1], 'media_job_stale_completions_total');
++ assert.ok(warnings.some((w) => w.includes(`refused a stale succeed of ${job.id}`) && w.includes('another claim holds it')), warnings.join('\n'));
++ assert.ok(warnings.some((w) => w.includes(`refused a stale succeed of ${job.id}`) && w.includes('it is succeeded now')));
++ console.warn = warn;
++ console.log('✅ two claimants: the stale holder\'s heartbeat, checkpoint and completion are refused (logged, counted); the current holder\'s stand');
++
++ // ── 3. Recovery does not take a job whose holder renewed meanwhile ──
++ const second = queue.enqueue({ appId: 'live', type: 'test.fence', createdBy: 'test' }).job;
++ const c = queue.claim(['test.fence'], { leaseS: 30 });
++ assert.strictEqual(c.id, second.id);
++ const seen = raw().prepare('SELECT * FROM media_jobs WHERE id = ?').get(second.id);
++ queue.renew(second.id, { leaseS: 30, token: c.lease_token }); // the lease is live again
++ assert.strictEqual(queue.fail(second.id, { message: 'x', code: 'interrupted', retryInS: 0, token: seen.lease_token, takeover: true, expired: true }), null,
++ 'a takeover needs the lease to have run out');
++ assert.strictEqual(queue.get(second.id).status, 'running');
++ assert.deepStrictEqual(queue.staleStats(), { succeed: 2, fail: 1, cancel: 1, renew: 1, checkpoint: 1 }, 'a declined takeover is not a stale completion');
++ assert.ok(queue.succeed(second.id, { ok: 1 }, { token: c.lease_token }));
++ console.log('✅ a takeover is fenced on the token it read and on the lease still being expired');
++
++ // ── 4. The worker: a heartbeat that finds its job taken aborts the handler; its report is refused ──
++ const worker = require('../server/jobs/worker');
++ let started = null, sawAbort = null;
++ queue.register('test.slowfence', {
++ lane: 'light', maxAttempts: 3,
++ run: (j, ctx) => new Promise((resolve, reject) => {
++ started = j.id;
++ ctx.signal.addEventListener('abort', () => { sawAbort = ctx.signal.reason; reject(ctx.signal.reason); });
++ }),
++ });
++ require('../server/config').jobs.leaseS = 3; // a heartbeat every second (the floor from the environment is 30 s)
++ const slow = queue.enqueue({ appId: 'live', type: 'test.slowfence', createdBy: 'test' }).job;
++ worker.start();
++ assert.ok(await waitFor(() => started === slow.id && queue.get(slow.id).status === 'running'), 'the worker runs it');
++ // Another claimant takes it over (as another process would after the lease ran out).
++ const theirs = 'f'.repeat(32);
++ raw().prepare('UPDATE media_jobs SET lease_token = ? WHERE id = ?').run(theirs, slow.id);
++ assert.ok(await waitFor(() => sawAbort !== null, 5000), 'the heartbeat notices and aborts the handler');
++ assert.strictEqual(sawAbort.code, 'media.job.lease_lost');
++ await waitFor(() => queue.staleStats().fail === 2);
++ const after = queue.get(slow.id);
++ assert.deepStrictEqual([after.status, after.lease_token, after.error_code], ['running', theirs, null], "the handler's failure did not overwrite the other claim");
++ assert.strictEqual(queue.staleStats().renew, 2);
++ worker.stop();
++ console.log('✅ the worker aborts a job whose lease it lost, and its failure is refused');
++
++ db.close();
++ fs.rmSync(tmp, { recursive: true, force: true });
++ console.log('jobs fencing: all checks passed');
++ process.exit(0);
++})().catch((err) => { console.error(err); process.exit(1); });