openvibers/openvibe.tools Commits

Jobs: POST /api/v1/jobs/:id/retry retries a failed job as a new job with the same type, input and files (retry_of / retried_by link them); owner-scoped (tools.job.create for principals), idempotent (a second call answers 200 Idempotent-Replayed with the same retry), 409 tools.job.not_failed for anything but a failed job, 410 tools.job.inputs_gone

OP OpenVibers committed 23 Sept 2026 · 2aff45dc69 · 6 files, +169 −15

A failed job now keeps its input files until it expires; the retry moves them to the new job
with one rename, and everything up to the insert is synchronous, so concurrent retries yield one
job. The per-owner active cap and the sandbox cap apply. tool_jobs gains retry_of and retried_by
(added when missing); job views carry them and links.retry. OVJobs.retry(id) in the browser
helper. Tests: failure then retry on the original input, two concurrent retries, repeat after
success, not-failed and running jobs refused, another owner 404, inputs gone 410.

Changes

ddiff --git a/README.md b/README.md
iindex b2f5120..7cc51f8 100644
---- a/README.md
++++ b/README.md
@@@ -74,6 +74,7 @@ routes answer on every host of the img, audio and docs satellites:
| `DELETE /api/v1/jobs/:id` | Cancel: queued → `200` cancelled at once; running → `202` (signalled; ffmpeg is killed, sharp/pdf-lib work finishes and is discarded) and ends `cancelled`; finished → `409 tools.job.already_finished`. |
| `GET /api/v1/jobs/:id/events` | SSE: `job.queued`, `job.running`, `job.progress`, `job.cancel_requested`, `job.succeeded`, `job.failed`, `job.cancelled`, each with an `id`. Reconnecting with `Last-Event-ID` (or `?last_event_id=`) replays only later events; a finished job with nothing newer answers `204`. |
| `GET /api/v1/jobs/:id/files/:n` | A result file (attachment); `?inline=1` for previews. |
++| `POST /api/v1/jobs/:id/retry` | Retry a **failed** job: a new job with the same type, input and files (`retry_of` → the failed one, which gets `retried_by`) → `202` + `Location`. Idempotent: asking again returns that same retry with `200` + `Idempotent-Replayed: true`. Not failed → `409 tools.job.not_failed`; inputs gone → `410 tools.job.inputs_gone`. A failed job keeps its input files until it expires. |
Job types: `img.process` (input `{ tool: convert|compress|resize|crop, format, quality, width, … }`, one image),
`audio.process` (`{ tool, …options }` as `/api/process` takes them, one audio/video file; progress from ffmpeg),
@@@ -106,9 +107,9 @@ Environment (all in `/etc/openvibe/tools.env`): `TOOLS_JOBS_CONCURRENCY`, `TOOLS
`TOOLS_JOBS_MAX_ACTIVE`, `TOOLS_JOB_RESULTS`, `TOOLS_MEDIA_NAMESPACE`, `OV_MEDIA_INTERNAL_URL`, `OV_MEDIA_URL`,
`OV_NETWORK_INTERNAL_URL`, `OV_OAUTH_CLIENT_ID`, `OV_OAUTH_CLIENT_SECRET`.
--Not done yet: job lifecycle events are not published to OpenVibe.Events; there is no retry endpoint (a failed
--retryable job is submitted again); quotas for external developer apps are the per-owner limits above, not a
--Codes-issued quota.
++Not done yet: job lifecycle events are not published to OpenVibe.Events (openvibe-contracts registers no
++`tools.job.*` event types, and Tools does not depend on openvibe-sdk yet); quotas for external developer apps are
++the per-owner limits above, not a Codes-issued quota.
## Canonical hosts and the service registry
ddiff --git a/apps/_shared/jobs/client.js b/apps/_shared/jobs/client.js
iindex 843e39f..439023e 100644
---- a/apps/_shared/jobs/client.js
++++ b/apps/_shared/jobs/client.js
@@@ -4,6 +4,7 @@
* OVJobs.remember(job.id); // ?job=<id> in the address + sessionStorage: a reload reattaches
* OVJobs.watch(job.id, { onUpdate(job) {…}, onDone(job) {…} });
* const pending = OVJobs.recall(); // on load: the job to reattach to, if any
++ * const next = await OVJobs.retry(failed.id); // a failed job, again, as a new job (idempotent)
*
* Progress arrives over SSE (EventSource resumes with Last-Event-ID by itself); if the stream cannot be
* kept open the helper polls GET /api/v1/jobs/:id instead. Errors are problem+json; submit() throws an
@@@ -51,6 +52,12 @@
.then(function (res) { return readJson(res).then(function (body) { if (!res.ok) throw problemError(res, body); return body; }); });
}
++ /** Retry a failed job. → the new job (or, if it was already retried, that retry). */
++ function retry(id) {
++ return fetch('/api/v1/jobs/' + encodeURIComponent(id) + '/retry', { method: 'POST', credentials: 'same-origin', headers: { Accept: 'application/json' } })
++ .then(function (res) { return readJson(res).then(function (body) { if (!res.ok) throw problemError(res, body); return body; }); });
++ }
++
/** Follow a job until it ends. → { close() } */
function watch(id, h) {
h = h || {};
@@@ -112,5 +119,5 @@
return job && job.expires_at ? Math.max(0, Date.parse(job.expires_at) - Date.now()) : null;
}
-- window.OVJobs = { submit: submit, get: get, cancel: cancel, watch: watch, remember: remember, recall: recall, forget: forget, fileUrl: fileUrl, expiresIn: expiresIn, TERMINAL: TERMINAL };
++ window.OVJobs = { submit: submit, get: get, cancel: cancel, retry: retry, watch: watch, remember: remember, recall: recall, forget: forget, fileUrl: fileUrl, expiresIn: expiresIn, TERMINAL: TERMINAL };
})();
ddiff --git a/apps/_shared/jobs/http.js b/apps/_shared/jobs/http.js
iindex c40320d..68258c6 100644
---- a/apps/_shared/jobs/http.js
++++ b/apps/_shared/jobs/http.js
@@@ -7,6 +7,9 @@
// → 202 + Location (new) | 200 + Idempotent-Replayed: true
// GET /api/v1/jobs/:id the job (reattach by id)
// DELETE /api/v1/jobs/:id cancel → 200 cancelled | 202 cancel requested (running)
++// POST /api/v1/jobs/:id/retry retry a failed job as a new job → 202 + Location (new) |
++// 200 + Idempotent-Replayed: true (it was already retried: that job)
++// | 409 tools.job.not_failed | 410 tools.job.inputs_gone
// GET /api/v1/jobs/:id/events SSE: job.queued|running|progress|cancel_requested|succeeded|failed|cancelled;
// every event has an id; reconnecting with Last-Event-ID replays
// only what came after it; 204 once there is nothing left to say
@@@ -194,6 +197,23 @@ function mountJobRoutes(app, o) {
return res.status(job.state === 'cancelled' ? 200 : 202).json(system.view(job));
});
++ // Retry is a write: a principal needs tools.job.create, like a submit.
++ app.post('/api/v1/jobs/:id/retry', ...(o.limiters || []), (req, res) => {
++ noStore(res);
++ const row = load(req, res, 'create');
++ if (!row) return;
++ try {
++ const { job, replayed } = system.retry(row.id);
++ res.set('Location', `/api/v1/jobs/${job.id}`);
++ if (replayed) res.set('Idempotent-Replayed', 'true');
++ return res.status(replayed ? 200 : 202).json(system.view(job));
++ } catch (err) {
++ if (err instanceof system.JobError) return send(res, err.status, err.code, err.detail, err.extra);
++ console.error('[Jobs] retry failed:', err.message);
++ return send(res, 500, 'tools.job.retry_failed', 'The job could not be retried');
++ }
++ });
++
app.get('/api/v1/jobs/:id/events', (req, res) => {
const row = load(req, res, 'read');
if (!row) return;
ddiff --git a/apps/_shared/jobs/store.js b/apps/_shared/jobs/store.js
iindex b326868..dfa9a7e 100644
---- a/apps/_shared/jobs/store.js
++++ b/apps/_shared/jobs/store.js
@@@ -52,20 +52,26 @@ CREATE TABLE IF NOT EXISTS tool_job_events (
CREATE INDEX IF NOT EXISTS tool_job_events_job ON tool_job_events(job_id, seq);
`;
++// Columns added after the first release, each added when missing (idempotent).
++const ADDED_COLUMNS = [
++ ['env', "TEXT NOT NULL DEFAULT 'production' CHECK (env IN ('production','sandbox'))"], // developer-app sandboxes
++ ['retry_of', 'TEXT'], // the failed job this one retries
++ ['retried_by', 'TEXT'], // set on a failed job once it has been retried: the retry's id
++];
++
const parse = (s, fallback) => { if (s == null) return fallback; try { return JSON.parse(s); } catch { return fallback; } };
function createStore(db) {
db.pragma('journal_mode = WAL');
db.pragma('busy_timeout = 5000');
db.exec(SCHEMA);
-- // Databases created before developer-app sandboxes: add the env column.
-- if (!db.prepare('PRAGMA table_info(tool_jobs)').all().some((c) => c.name === 'env')) {
-- db.exec("ALTER TABLE tool_jobs ADD COLUMN env TEXT NOT NULL DEFAULT 'production' CHECK (env IN ('production','sandbox'))");
-- }
++ const have = new Set(db.prepare('PRAGMA table_info(tool_jobs)').all().map((c) => c.name));
++ for (const [name, decl] of ADDED_COLUMNS) if (!have.has(name)) db.exec(`ALTER TABLE tool_jobs ADD COLUMN ${name} ${decl}`);
const q = {
-- insert: db.prepare(`INSERT INTO tool_jobs (id, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env)
-- VALUES (@id, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env)`),
++ insert: db.prepare(`INSERT INTO tool_jobs (id, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of)
++ VALUES (@id, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of)`),
++ markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND state = 'failed' AND retried_by IS NULL"),
get: db.prepare('SELECT * FROM tool_jobs WHERE id = ?'),
byIdem: db.prepare('SELECT * FROM tool_jobs WHERE owner = ? AND idempotency_key = ?'),
nextQueued: db.prepare("SELECT * FROM tool_jobs WHERE state = 'queued' ORDER BY id LIMIT ?"),
@@@ -96,7 +102,8 @@ function createStore(db) {
return {
db,
TERMINAL, STATES,
-- insert(row) { q.insert.run(row); },
++ insert(row) { q.insert.run({ retry_of: null, ...row }); },
++ markRetried: (id, next, now = Date.now()) => q.markRetried.run({ id, next, now }).changes === 1,
get: (id) => q.get.get(id) || null,
byIdempotencyKey: (owner, key) => q.byIdem.get(owner, key) || null,
nextQueued: (limit) => q.nextQueued.all(limit),
ddiff --git a/apps/_shared/jobs/system.js b/apps/_shared/jobs/system.js
iindex 9b5d8d7..d1c0161 100644
---- a/apps/_shared/jobs/system.js
++++ b/apps/_shared/jobs/system.js
@@@ -13,6 +13,10 @@
// most `maxActivePerOwner` queued + running jobs. Finished jobs expire (ttl chosen at submit),
// and the pruner deletes the row, its events, its files and any Media objects it made.
//
++// Retry: a failed job keeps its input files until it expires; retry(id) moves them to a new job
++// (retry_of → the failed one, which records retried_by). Retrying the same failed job again
++// returns that same new job, so the call is idempotent.
++//
// No dependencies of its own: the app passes its better-sqlite3 handle and openvibe-contracts.
// ═══════════════════════════════════════════════════════════════
@@@ -136,10 +140,14 @@ function createJobSystem(o) {
result: publicResult(store.parse(row.result_json, null), row.id),
error: store.parse(row.error_json, null),
retryable: !!row.retryable,
++ retry_of: row.retry_of || null,
++ retried_by: row.retried_by || null,
links: {
self: `/api/v1/jobs/${row.id}`,
events: `/api/v1/jobs/${row.id}/events`,
cancel: live ? `/api/v1/jobs/${row.id}` : null,
++ retry: row.state === 'failed' ? `/api/v1/jobs/${row.id}/retry` : null,
++ retried_by: row.retried_by ? `/api/v1/jobs/${row.retried_by}` : null,
},
};
}
@@@ -308,7 +316,8 @@ function createJobSystem(o) {
emit(row.id, EVENT_FOR[outcome.state]);
} finally {
active.delete(row.id);
-- fsp.rm(path.join(jobDir(row.id), 'in'), { recursive: true, force: true }).catch(() => {});
++ // A failed job keeps its inputs (until it expires) so it can be retried.
++ if (outcome.state !== 'failed') fsp.rm(path.join(jobDir(row.id), 'in'), { recursive: true, force: true }).catch(() => {});
if (outcome.state !== 'succeeded') fsp.rm(outDir, { recursive: true, force: true }).catch(() => {});
kick();
}
@@@ -377,6 +386,59 @@ function createJobSystem(o) {
return { job: store.get(id), changed: true };
}
++ // ── Retry ──────────────────────────────────────────────────
++ /**
++ * Retry a failed job as a new job with the same type, input, files and lifetime.
++ * → { job, replayed } | null (no such job). Idempotent: a failed job is retried once, and asking
++ * again returns that retry, whatever state it is in now. Only failed jobs can be retried.
++ * Everything up to the insert is synchronous, so two concurrent calls cannot both retry it.
++ */
++ function retry(id) {
++ const row = store.get(id);
++ if (!row) return null;
++ if (row.retried_by) {
++ const next = store.get(row.retried_by);
++ if (next) return { job: next, replayed: true };
++ throw new JobError(410, 'tools.job.retry_gone', 'This job was retried and the retry has since expired', { retried_by: row.retried_by });
++ }
++ if (row.state !== 'failed') throw new JobError(409, 'tools.job.not_failed', `Only failed jobs can be retried; this one is ${row.state}`, { state: row.state });
++ if (stopped) throw new JobError(503, 'tools.job.unavailable', 'The job system is shutting down');
++ const def = types.get(row.type);
++ if (!def) throw new JobError(409, 'tools.job.unknown_type', `This service no longer runs "${row.type}" jobs`);
++ const limit = row.env === 'sandbox' ? Math.min(SANDBOX_MAX_ACTIVE, maxActivePerOwner) : maxActivePerOwner;
++ if (store.activeForOwner(row.owner) >= limit) throw new JobError(429, 'tools.job.too_many_active', `At most ${limit} unfinished jobs at a time; wait for one to finish or cancel one`);
++ const inputs = store.parse(row.files_json, []);
++ if (inputs.some(f => !f._path || !fs.existsSync(f._path))) throw new JobError(410, 'tools.job.inputs_gone', 'The input files of this job are no longer kept; submit it again');
++
++ const next = `job_${contracts.ids.ulid()}`;
++ const oldIn = path.join(jobDir(row.id), 'in');
++ const newIn = path.join(jobDir(next), 'in');
++ let moved = false;
++ if (inputs.length) {
++ fs.mkdirSync(jobDir(next), { recursive: true });
++ fs.renameSync(oldIn, newIn); // same directory tree, so one atomic rename
++ moved = true;
++ }
++ const stored = inputs.map(f => ({ ...f, _path: path.join(newIn, path.basename(f._path)) }));
++ try {
++ store.transaction(() => {
++ if (!store.markRetried(row.id, next)) throw new JobError(409, 'tools.job.not_failed', 'This job cannot be retried any more');
++ store.insert({
++ id: next, type: row.type, type_version: def.version, owner: row.owner, input_json: row.input_json, files_json: JSON.stringify(stored),
++ idempotency_key: null, request_hash: row.request_hash, max_attempts: def.maxAttempts, ttl_ms: row.ttl_ms, now: Date.now(),
++ env: row.env, retry_of: row.id,
++ });
++ })();
++ } catch (err) {
++ if (moved) { try { fs.renameSync(newIn, oldIn); } catch { /* best effort */ } }
++ fs.rmSync(jobDir(next), { recursive: true, force: true });
++ throw err;
++ }
++ emit(next, 'job.queued');
++ kick();
++ return { job: store.get(next), replayed: false };
++ }
++
// ── Boot recovery, pruning ─────────────────────────────────
function recover() {
let requeued = 0, failed = 0;
@@@ -458,7 +520,7 @@ function createJobSystem(o) {
}
const api = {
-- define, submit, cancel, subscribe, start, stop, onStop, prune, recover, view, resultFile,
++ define, submit, cancel, retry, subscribe, start, stop, onStop, prune, recover, view, resultFile,
get: (id) => store.get(id),
eventsAfter: (id, seq) => store.eventsAfter(id, seq),
lastSeq: (id) => store.lastSeq(id),
ddiff --git a/apps/_shared/test/jobs.test.js b/apps/_shared/test/jobs.test.js
iindex 1ddf559..cb5a57f 100644
---- a/apps/_shared/test/jobs.test.js
++++ b/apps/_shared/test/jobs.test.js
@@@ -2,8 +2,8 @@
// Wave 11 exit criteria for the shared job runtime, end to end over HTTP:
// an accepted job survives a restart (DB and app closed and reopened), running jobs are re-queued
// or failed-retryable per type, reattach by id, cancel (queued and running), idempotency keys,
--// owner scoping, SSE resume with Last-Event-ID, bounded concurrency, pruning, and results stored
--// as Media objects through Media's v2 object API (with the local fallback).
++// owner scoping, SSE resume with Last-Event-ID, bounded concurrency, pruning, results stored
++// as Media objects through Media's v2 object API (with the local fallback), and retrying a failed job.
const assert = require('assert');
const fs = require('fs');
const os = require('os');
@@@ -63,7 +63,18 @@ function define(system) {
},
});
system.define({ type: 'test.boom', async run() { throw new Error('This input at /srv/secret/path/x.png is not an image'); } });
++ // Upper-cases its file; fails while its input.tag is in `flaky` (a transient failure, then fixed).
++ system.define({
++ type: 'test.flaky', maxFiles: 1,
++ async run({ input, files, outDir }) {
++ if (flaky.has(input.tag)) throw Object.assign(new Error('upstream hiccup'), { retryable: true });
++ const out = path.join(outDir, 'flaky.txt');
++ fs.writeFileSync(out, fs.readFileSync(files[0].path, 'utf8').toUpperCase());
++ return { files: [{ path: out, name: 'flaky.txt', mime: 'text/plain' }] };
++ },
++ });
}
++const flaky = new Set();
// ── A fake OpenVibe.Media (object API v2) ────────────────────
function fakeMedia() {
@@@ -360,6 +371,52 @@ async function readSse(url, { headers = {}, stop = () => false, ms = 5000 } = {}
await s.close();
mediaServer.close();
++ // ── Retry a failed job ──────────────────────────────────
++ s = await satellite(path.join(root, 'h'));
++ const hal = client(s.base, { user: USER_B });
++ flaky.add('t1');
++ const fd3 = new FormData();
++ fd3.append('type', 'test.flaky'); fd3.append('input', JSON.stringify({ tag: 't1' })); fd3.append('file', new Blob(['retry me']), 'in.txt');
++ const failedId = (await hal('/api/v1/jobs', { method: 'POST', body: fd3 })).body.id;
++ const failedJob = await until(async () => { const g = await hal(`/api/v1/jobs/${failedId}`); return g.body.state === 'failed' && g.body; }, 'flaky job fails');
++ assert.strictEqual(failedJob.links.retry, `/api/v1/jobs/${failedId}/retry`);
++ assert.ok(fs.readdirSync(path.join(root, 'h', 'jobs', failedId, 'in')).length === 1, 'a failed job keeps its input for a retry');
++ assert.strictEqual((await client(s.base, { user: USER_A })(`/api/v1/jobs/${failedId}/retry`, { method: 'POST' })).status, 404, 'only the owner can retry');
++ flaky.delete('t1');
++ // Two retries at once: one new job, and the other call gets the same one.
++ const [ra, rb] = await Promise.all([hal(`/api/v1/jobs/${failedId}/retry`, { method: 'POST' }), hal(`/api/v1/jobs/${failedId}/retry`, { method: 'POST' })]);
++ assert.deepStrictEqual([ra.status, rb.status].sort(), [200, 202]);
++ assert.strictEqual(ra.body.id, rb.body.id, 'retrying is idempotent');
++ const retried = ra.body.id;
++ assert.notStrictEqual(retried, failedId);
++ assert.strictEqual((ra.status === 202 ? ra : rb).headers.get('location'), `/api/v1/jobs/${retried}`);
++ assert.strictEqual((ra.status === 200 ? ra : rb).headers.get('idempotent-replayed'), 'true');
++ assert.strictEqual(ra.body.retry_of, failedId);
++ const ok = await until(async () => { const g = await hal(`/api/v1/jobs/${retried}`); return g.body.state === 'succeeded' && g.body; }, 'retry succeeds');
++ assert.strictEqual((await hal(ok.result.files[0].url)).body, 'RETRY ME', 'the retry ran on the original input');
++ const old = (await hal(`/api/v1/jobs/${failedId}`)).body;
++ assert.strictEqual(old.state, 'failed', 'the failed job stays failed');
++ assert.strictEqual(old.retried_by, retried);
++ r = await hal(`/api/v1/jobs/${failedId}/retry`, { method: 'POST' });
++ assert.strictEqual(r.status, 200); assert.strictEqual(r.body.id, retried, 'asking again later still answers with the same retry');
++ r = await hal(`/api/v1/jobs/${retried}/retry`, { method: 'POST' });
++ assert.strictEqual(r.status, 409); assert.strictEqual(r.body.code, 'tools.job.not_failed', 'only failed jobs can be retried');
++ const running = (await hal('/api/v1/jobs', json({ type: 'test.upper', input: { gate: 'retry-running' } }))).body.id;
++ await until(async () => (await hal(`/api/v1/jobs/${running}`)).body.state === 'running', 'running job');
++ assert.strictEqual((await hal(`/api/v1/jobs/${running}/retry`, { method: 'POST' })).body.code, 'tools.job.not_failed');
++ gate('retry-running').open();
++ // A failed job whose inputs are gone cannot be retried.
++ flaky.add('t2');
++ const fd4 = new FormData();
++ fd4.append('type', 'test.flaky'); fd4.append('input', JSON.stringify({ tag: 't2' })); fd4.append('file', new Blob(['x']), 'x.txt');
++ const lost = (await hal('/api/v1/jobs', { method: 'POST', body: fd4 })).body.id;
++ await until(async () => (await hal(`/api/v1/jobs/${lost}`)).body.state === 'failed', 'second flaky job fails');
++ fs.rmSync(path.join(root, 'h', 'jobs', lost, 'in'), { recursive: true, force: true });
++ r = await hal(`/api/v1/jobs/${lost}/retry`, { method: 'POST' });
++ assert.strictEqual(r.status, 410); assert.strictEqual(r.body.code, 'tools.job.inputs_gone');
++ assert.strictEqual((await hal(`/api/v1/jobs/${lost}`)).body.retried_by, null, 'a refused retry changes nothing');
++ await s.close();
++
// ── Service/app principals (tools.job.* capabilities) ───
s = await satellite(path.join(root, 'f'));
const token = (claims) => contracts.serviceAuth.signServiceToken({