openvibers/openvibe.tools Commits
Jobs retention: referenced results are never pruned (PUT/DELETE /api/v1/jobs/:id/references/:ref, e.g. community:paste:p_123, owner-scoped, idempotent; a referenced job shows expires_at null), and the pruner no longer drops a job whose Media object Media keeps (retention hold 409 media.object.held, Media unreachable or not configured): the job is kept and looked at again 24 h later
Changes
@@@ -75,6 +75,8 @@ routes answer on every host of the img, audio and docs satellites:| `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. |++| `PUT /api/v1/jobs/:id/references/:ref` | Keep a **succeeded** job's result while `<ref>` (`<service>:<kind>:<id>`, e.g. `community:paste:p_123`) points at it → `201` (`200` if already there). The job then shows `references` and `expires_at: null`. At most 50 per job; not for sandbox jobs. |++| `DELETE /api/v1/jobs/:id/references/:ref` | Drop it → `200`; after the last one the job expires one ttl later at the earliest. |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),@@@ -94,7 +96,10 @@ reload reattaches.overrides), `TOOLS_JOBS_MAX_ACTIVE` unfinished jobs per owner (default 10, then `429`), plus the satellites'existing burst and processing rate limits on submit.- **Retention.** Finished jobs expire after 1 hour (browser sessions) or 24 hours (signed-in people, principals);-- the pruner deletes the row, its events, its files and its Media objects.++ the pruner deletes the row, its events, its files and its Media objects. It never touches a job that has a++ reference (above). If Media will not delete a result object (a retention hold, `409 media.object.held`), cannot++ be reached, or is no longer configured here, the job is kept, record and all, and looked at again 24 hours++ later, so no object is left without the record that would delete it.- **Results in OpenVibe.Media** when `TOOLS_JOB_RESULTS=media`: each result file becomes a private Media object(v2 object API, namespace `tools`, owner `X-OV-Subject` for signed-in people) and the job's result carries its`media.media_id`; previews redirect to a short-lived signed Media URL, downloads stream through the satellite.@@@ -114,7 +114,7 @@if (!f) return null;return f.url + (opts && opts.inline ? '?inline=1' : '');}-- /** Milliseconds until a finished job's files expire (null while it is still running). */++ /** Milliseconds until a finished job's files expire (null while it is still running, or while something references it). */function expiresIn(job) {return job && job.expires_at ? Math.max(0, Date.parse(job.expires_at) - Date.now()) : null;}@@@ -10,6 +10,9 @@// 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++// PUT /api/v1/jobs/:id/references/:ref keep a succeeded job's result while <ref> points at it++// (e.g. community:paste:p_123) → 201 new | 200 already there++// DELETE /api/v1/jobs/:id/references/:ref drop it → 200 (the expiry clock restarts after the last one)// 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@@@ -197,7 +200,7 @@ 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.++ // Retry and references are writes: 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');@@@ -214,6 +217,24 @@ function mountJobRoutes(app, o) {}});++ function referenceRoute(fn) {++ return (req, res) => {++ noStore(res);++ const row = load(req, res, 'create');++ if (!row) return;++ try {++ const r = fn(row.id, String(req.params.ref || ''));++ return res.status(r.created ? 201 : 200).json(system.view(r.job));++ } catch (err) {++ if (err instanceof system.JobError) return send(res, err.status, err.code, err.detail, err.extra);++ console.error('[Jobs] reference failed:', err.message);++ return send(res, 500, 'tools.job.reference_failed', 'The reference could not be changed');++ }++ };++ }++ app.put('/api/v1/jobs/:id/references/:ref', referenceRoute((id, ref) => system.reference(id, ref)));++ app.delete('/api/v1/jobs/:id/references/:ref', referenceRoute((id, ref) => system.unreference(id, ref)));++app.get('/api/v1/jobs/:id/events', (req, res) => {const row = load(req, res, 'read');if (!row) return;@@@ -2,10 +2,12 @@// ═══════════════════════════════════════════════════════════════// Job store — the satellite's own SQLite (better-sqlite3 handle passed in).//--// Two tables, created on first use (idempotent):--// tool_jobs one row per job; the durable truth a restart recovers from--// tool_job_events the ordered event log per job; its AUTOINCREMENT seq is the SSE event id,--// so a reconnect with Last-Event-ID replays exactly what was missed++// Three tables, created on first use (idempotent):++// tool_jobs one row per job; the durable truth a restart recovers from++// tool_job_events the ordered event log per job; its AUTOINCREMENT seq is the SSE event id,++// so a reconnect with Last-Event-ID replays exactly what was missed++// tool_job_references who still points at a job's result (a paste, a project, …); a referenced++// job never expires, so its files and Media objects are not pruned// ═══════════════════════════════════════════════════════════════const TERMINAL = ['succeeded', 'failed', 'cancelled'];@@@ -50,6 +52,12 @@ CREATE TABLE IF NOT EXISTS tool_job_events (at INTEGER NOT NULL);CREATE INDEX IF NOT EXISTS tool_job_events_job ON tool_job_events(job_id, seq);++CREATE TABLE IF NOT EXISTS tool_job_references (++ job_id TEXT NOT NULL,++ ref TEXT NOT NULL,++ created_at INTEGER NOT NULL,++ PRIMARY KEY (job_id, ref)++);`;// Columns added after the first release, each added when missing (idempotent).@@@ -88,9 +96,19 @@ function createStore(db) {event: db.prepare('INSERT INTO tool_job_events (job_id, event, data, at) VALUES (?, ?, ?, ?)'),eventsAfter: db.prepare('SELECT seq, event, data, at FROM tool_job_events WHERE job_id = ? AND seq > ? ORDER BY seq'),lastSeq: db.prepare('SELECT MAX(seq) AS seq FROM tool_job_events WHERE job_id = ?'),-- expired: db.prepare('SELECT * FROM tool_jobs WHERE expires_at IS NOT NULL AND expires_at <= ? LIMIT 500'),++ // A job something still references is never expired (see tool_job_references).++ expired: db.prepare(`SELECT * FROM tool_jobs j WHERE expires_at IS NOT NULL AND expires_at <= ?++ AND NOT EXISTS (SELECT 1 FROM tool_job_references r WHERE r.job_id = j.id) LIMIT 500`),++ deferExpiry: db.prepare('UPDATE tool_jobs SET expires_at = @until, updated_at = @now WHERE id = @id'),del: db.prepare('DELETE FROM tool_jobs WHERE id = ?'),delEvents: db.prepare('DELETE FROM tool_job_events WHERE job_id = ?'),++ delRefs: db.prepare('DELETE FROM tool_job_references WHERE job_id = ?'),++ addRef: db.prepare('INSERT OR IGNORE INTO tool_job_references (job_id, ref, created_at) VALUES (?, ?, ?)'),++ dropRef: db.prepare('DELETE FROM tool_job_references WHERE job_id = ? AND ref = ?'),++ refs: db.prepare('SELECT ref, created_at FROM tool_job_references WHERE job_id = ? ORDER BY created_at, ref'),++ refCount: db.prepare('SELECT COUNT(*) AS n FROM tool_job_references WHERE job_id = ?'),++ // After the last reference goes, the result is kept at least one more ttl.++ releaseExpiry: db.prepare('UPDATE tool_jobs SET expires_at = MAX(COALESCE(expires_at, 0), @now + ttl_ms), updated_at = @now WHERE id = @id AND finished_at IS NOT NULL'),counts: db.prepare('SELECT state, COUNT(*) AS n FROM tool_jobs GROUP BY state'),};@@@ -123,7 +141,18 @@ function createStore(db) {eventsAfter: (jobId, seq) => q.eventsAfter.all(jobId, seq).map(e => ({ seq: e.seq, event: e.event, data: parse(e.data, {}), at: e.at })),lastSeq: (jobId) => q.lastSeq.get(jobId).seq || 0,expired: (now = Date.now()) => q.expired.all(now),-- remove: db.transaction((id) => { q.delEvents.run(id); q.del.run(id); }),++ deferExpiry: (id, until, now = Date.now()) => q.deferExpiry.run({ id, until, now }),++ remove: db.transaction((id) => { q.delEvents.run(id); q.delRefs.run(id); q.del.run(id); }),++ /** → true when the reference is new. */++ addReference: (id, ref, now = Date.now()) => q.addRef.run(id, ref, now).changes === 1,++ /** → true when it existed; dropping the last one restarts the job's expiry clock. */++ dropReference: db.transaction((id, ref, now = Date.now()) => {++ const gone = q.dropRef.run(id, ref).changes === 1;++ if (gone && q.refCount.get(id).n === 0) q.releaseExpiry.run({ id, now });++ return gone;++ }),++ references: (id) => q.refs.all(id),++ referenceCount: (id) => q.refCount.get(id).n,counts() { const out = Object.fromEntries(STATES.map(s => [s, 0])); for (const r of q.counts.all()) out[r.state] = r.n; return out; },transaction: (fn) => db.transaction(fn),parse,@@@ -11,7 +11,10 @@//// Bounded: at most `concurrency` jobs run at once in this process, and each owner may have at// 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.++// and the pruner deletes the row, its events, its files and any Media objects it made — except++// while something references the result (reference(), e.g. a paste or a project that points at the++// Media object), and never while Media keeps an object (a retention hold, or Media unreachable):++// then the job is kept and looked at again later, so no referenced result loses its record.//// 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@@@ -27,6 +30,9 @@ const crypto = require('crypto');const { createStore, TERMINAL } = require('./store');const HOUR = 60 * 60 * 1000;++const PRUNE_RECHECK_MS = 24 * HOUR; // an expired job the pruner could not finish is looked at again after this++const REF_RE = /^[a-z][a-z0-9_-]*(?::[A-Za-z0-9_.-]+){1,4}$/; // <service>:<kind>:<id>, e.g. community:paste:p_123++const MAX_REFS = 50;const EVENT_FOR = { queued: 'job.queued', running: 'job.running', succeeded: 'job.succeeded', failed: 'job.failed', cancelled: 'job.cancelled' };const SAFE_NAME = (s) => String(s || 'file').replace(/[/\\\0]/g, '_').replace(/[^\w.\- ]/g, '_').slice(-120) || 'file';@@@ -122,6 +128,7 @@ function createJobSystem(o) {function view(row) {if (!row) return null;const live = row.state === 'queued' || row.state === 'running';++ const references = store.references(row.id).map(r => ({ ref: r.ref, created_at: iso(r.created_at) }));return {id: row.id,object: 'tools.job',@@@ -136,12 +143,14 @@ function createJobSystem(o) {created_at: iso(row.created_at),started_at: iso(row.started_at),finished_at: iso(row.finished_at),-- expires_at: iso(row.expires_at),++ // null while something references the result: it is kept until the last reference goes.++ expires_at: references.length ? null : iso(row.expires_at),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,++ references,links: {self: `/api/v1/jobs/${row.id}`,events: `/api/v1/jobs/${row.id}/events`,@@@ -439,6 +448,35 @@ function createJobSystem(o) {return { job: store.get(next), replayed: false };}++ // ── References (results something still points at) ─────────++ function checkRef(ref) {++ if (!REF_RE.test(String(ref || '')) || String(ref).length > 200) {++ throw new JobError(400, 'tools.job.invalid', 'A reference is <service>:<kind>:<id>, e.g. community:paste:p_123 (at most 200 characters)');++ }++ return String(ref);++ }++ /**++ * Keep a succeeded job's result while `ref` points at it. → { job, created }. Idempotent.++ */++ function reference(id, ref) {++ ref = checkRef(ref);++ const row = store.get(id);++ if (!row) return null;++ if (row.state !== 'succeeded') throw new JobError(409, 'tools.job.not_succeeded', `Only a succeeded job's result can be referenced; this one is ${row.state}`, { state: row.state });++ if (row.env === 'sandbox') throw new JobError(409, 'tools.job.sandbox', 'Sandbox results are kept briefly and cannot be referenced');++ const has = store.references(id).some(r => r.ref === ref);++ if (!has && store.referenceCount(id) >= MAX_REFS) throw new JobError(409, 'tools.job.too_many_references', `At most ${MAX_REFS} references per job`);++ const created = store.addReference(id, ref);++ return { job: store.get(id), created };++ }++ /** Stop keeping the result for `ref`. → { job, removed }. Idempotent. */++ function unreference(id, ref) {++ ref = checkRef(ref);++ if (!store.get(id)) return null;++ const removed = store.dropReference(id, ref);++ return { job: store.get(id), removed };++ }++// ── Boot recovery, pruning ─────────────────────────────────function recover() {let requeued = 0, failed = 0;@@@ -471,11 +509,30 @@ function createJobSystem(o) {return { requeued, failed };}++ /**++ * Delete expired jobs. Referenced jobs are never expired (store.expired skips them). A job whose++ * Media objects cannot all be deleted — Media keeps one under a retention hold (409++ * media.object.held), Media is unreachable, or this process no longer has Media configured —++ * is kept, record and all, and looked at again after PRUNE_RECHECK_MS: dropping the record++ * would leave an object nobody deletes, and a held object is one somebody still needs.++ */async function prune(now = Date.now()) {let n = 0;for (const row of store.expired(now)) {const result = store.parse(row.result_json, null);-- if (media && result) for (const f of result.files || []) if (f.media && f.media.media_id) await media.remove(f.media.media_id).catch(err => log.warn(`[Jobs] could not delete ${f.media.media_id}:`, err.message));++ let kept = null;++ for (const f of (result && result.files) || []) {++ if (!f.media || !f.media.media_id) continue;++ if (!media) { kept = 'results in OpenVibe.Media, but Media is not configured here'; continue; }++ try { await media.remove(f.media.media_id); } catch (err) {++ kept = err.code === 'media.object.held' ? `${f.media.media_id} is under a retention hold` : `could not delete ${f.media.media_id}: ${err.message}`;++ }++ }++ if (kept) {++ log.warn(`[Jobs] kept expired job ${row.id} (${kept}); checking again later`);++ store.deferExpiry(row.id, now + PRUNE_RECHECK_MS);++ continue;++ }await fsp.rm(jobDir(row.id), { recursive: true, force: true }).catch(() => {});store.remove(row.id);listeners.delete(row.id);@@@ -520,7 +577,7 @@ function createJobSystem(o) {}const api = {-- define, submit, cancel, retry, subscribe, start, stop, onStop, prune, recover, view, resultFile,++ define, submit, cancel, retry, reference, unreference, 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),@@@ -3,7 +3,8 @@// 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, results stored--// as Media objects through Media's v2 object API (with the local fallback), and retrying a failed job.++// as Media objects through Media's v2 object API (with the local fallback), retention of results++// that are referenced or held in Media, and retrying a failed job.const assert = require('assert');const fs = require('fs');const os = require('os');@@@ -80,7 +81,7 @@ const flaky = new Set();function fakeMedia() {const objects = new Map();const app = express();-- const state = { failInit: false, deleted: [] };++ const state = { failInit: false, deleted: [], held: new Set(), deleteDown: false };app.use(express.json());const authed = (req, res, next) => (req.headers.authorization === 'Bearer media-token' ? next() : res.status(401).json({ error: 'no' }));app.post('/api/v2/tools/objects', authed, (req, res) => {@@@ -103,7 +104,11 @@ function fakeMedia() {res.json({ id: o.id, lifecycle_status: 'ready' });});app.get('/api/v2/tools/objects/:id/download', authed, (req, res) => res.json({ url: `https://openvibe.media/o/${req.params.id}?exp=1&sig=s`, expires_at: 'x', public: false }));-- app.delete('/api/v2/tools/objects/:id', authed, (req, res) => { state.deleted.push(req.params.id); objects.delete(req.params.id); res.json({ ok: true }); });++ app.delete('/api/v2/tools/objects/:id', authed, (req, res) => {++ if (state.deleteDown) return res.status(503).json({ code: 'media.unavailable' });++ if (state.held.has(req.params.id)) return res.status(409).type('application/problem+json').send(JSON.stringify({ status: 409, code: 'media.object.held', detail: 'Object is under a retention hold' }));++ state.deleted.push(req.params.id); objects.delete(req.params.id); res.json({ ok: true });++ });app.get('/o/:id', (req, res) => { const o = objects.get(req.params.id); if (!o || req.query.sig !== 's') return res.status(404).end(); res.type('text/plain').send(o.bytes); });return { app, objects, state };}@@@ -369,6 +374,60 @@ async function readSse(url, { headers = {}, stop = () => false, ms = 5000 } = {}await s.system.prune(Date.now() + 25 * 60 * 60 * 1000);assert.ok(fm.state.deleted.includes(f0.media.media_id), 'pruning deletes the Media object');await s.close();++++ // ── Retention: referenced or held results are not pruned ──++ fm.state.failInit = false;++ const DAY = 24 * 60 * 60 * 1000;++ s = await satellite(path.join(root, 'g'), { media });++ const gina = client(s.base, { user: USER_A });++ const finished = async (text) => {++ const id = (await gina('/api/v1/jobs', json({ type: 'test.upper', input: { text } }))).body.id;++ return until(async () => { const g = await gina(`/api/v1/jobs/${id}`); return g.body.state === 'succeeded' && g.body; }, `job ${text}`);++ };++ const kept = await finished('kept by a paste');++ const keptMedia = kept.result.files[0].media.media_id;++ r = await gina(`/api/v1/jobs/${kept.id}/references/community:paste:p_123`, { method: 'PUT' });++ assert.strictEqual(r.status, 201, 'a reference is recorded');++ assert.deepStrictEqual(r.body.references.map(x => x.ref), ['community:paste:p_123']);++ assert.strictEqual(r.body.expires_at, null, 'a referenced result does not expire');++ assert.strictEqual((await gina(`/api/v1/jobs/${kept.id}/references/community:paste:p_123`, { method: 'PUT' })).status, 200, 'referencing twice is fine');++ assert.strictEqual((await client(s.base, { user: USER_B })(`/api/v1/jobs/${kept.id}/references/community:paste:p_9`, { method: 'PUT' })).status, 404, 'only the owner can reference it');++ r = await gina(`/api/v1/jobs/${kept.id}/references/${encodeURIComponent('no spaces allowed')}`, { method: 'PUT' });++ assert.strictEqual(r.status, 400); assert.strictEqual(r.body.code, 'tools.job.invalid');++ const held = await finished('held by Media');++ const heldMedia = held.result.files[0].media.media_id;++ fm.state.held.add(heldMedia);++ const plain = await finished('nobody needs this');++ fm.state.deleted.length = 0;++ await s.system.prune(Date.now() + 2 * DAY);++ assert.strictEqual((await gina(`/api/v1/jobs/${kept.id}`)).status, 200, 'a referenced job survives pruning');++ assert.ok(fm.objects.has(keptMedia) && !fm.state.deleted.includes(keptMedia), 'and so does its Media object');++ assert.strictEqual((await gina(`/api/v1/jobs/${kept.id}/files/0`)).body, 'KEPT BY A PASTE', 'still downloadable');++ const stillHeld = (await gina(`/api/v1/jobs/${held.id}`)).body;++ assert.ok(stillHeld.id && Date.parse(stillHeld.expires_at) > Date.now() + 2 * DAY, 'a job whose Media object is held keeps its record and is checked again later');++ assert.ok(fm.objects.has(heldMedia));++ assert.strictEqual((await gina(`/api/v1/jobs/${plain.id}`)).status, 404, 'an unreferenced job is pruned as before');++ assert.deepStrictEqual(fm.state.deleted, [plain.result.files[0].media.media_id]);++ // Media unreachable: nothing is dropped.++ const outage = await finished('media is down');++ fm.state.deleteDown = true;++ await s.system.prune(Date.now() + 2 * DAY);++ assert.strictEqual((await gina(`/api/v1/jobs/${outage.id}`)).status, 200, 'a failed Media delete keeps the job');++ fm.state.deleteDown = false;++ // Dropping the last reference restarts the clock; after that the job expires normally.++ r = await gina(`/api/v1/jobs/${kept.id}/references/community:paste:p_123`, { method: 'DELETE' });++ assert.strictEqual(r.status, 200); assert.deepStrictEqual(r.body.references, []);++ assert.ok(Date.parse(r.body.expires_at) >= Date.now() + DAY - 60000, 'kept one more ttl after the last reference goes');++ assert.strictEqual((await gina(`/api/v1/jobs/${kept.id}/references/community:paste:p_123`, { method: 'DELETE' })).status, 200, 'dropping twice is fine');++ fm.state.held.delete(heldMedia);++ await s.system.prune(Date.now() + 5 * DAY);++ for (const j of [kept, held, outage]) assert.strictEqual((await gina(`/api/v1/jobs/${j.id}`)).status, 404, `${j.id} pruned once nothing keeps it`);++ assert.ok([keptMedia, heldMedia].every(id => fm.state.deleted.includes(id)));++ r = await gina('/api/v1/jobs', json({ type: 'test.boom' }));++ await until(async () => (await gina(`/api/v1/jobs/${r.body.id}`)).body.state === 'failed', 'failed job');++ r = await gina(`/api/v1/jobs/${r.body.id}/references/community:paste:p_1`, { method: 'PUT' });++ assert.strictEqual(r.status, 409); assert.strictEqual(r.body.code, 'tools.job.not_succeeded');++ await s.close();mediaServer.close();// ── Retry a failed job ──────────────────────────────────
@@ -75,6 +75,8 @@ routes answer on every host of the img, audio and docs satellites:
@@ -75,6 +75,8 @@ routes answer on every host of the img, audio and docs satellites:
| `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/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. |
| `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. |
| `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. |
+| `PUT /api/v1/jobs/:id/references/:ref` | Keep a **succeeded** job's result while `<ref>` (`<service>:<kind>:<id>`, e.g. `community:paste:p_123`) points at it → `201` (`200` if already there). The job then shows `references` and `expires_at: null`. At most 50 per job; not for sandbox jobs. |
+| `DELETE /api/v1/jobs/:id/references/:ref` | Drop it → `200`; after the last one the job expires one ttl later at the earliest. |
Job types: `img.process` (input `{ tool: convert|compress|resize|crop, format, quality, width, … }`, one image),
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),
`audio.process` (`{ tool, …options }` as `/api/process` takes them, one audio/video file; progress from ffmpeg),
@@ -94,7 +96,10 @@ reload reattaches.
@@ -94,7 +96,10 @@ reload reattaches.
overrides), `TOOLS_JOBS_MAX_ACTIVE` unfinished jobs per owner (default 10, then `429`), plus the satellites'
overrides), `TOOLS_JOBS_MAX_ACTIVE` unfinished jobs per owner (default 10, then `429`), plus the satellites'
existing burst and processing rate limits on submit.
existing burst and processing rate limits on submit.
- **Retention.** Finished jobs expire after 1 hour (browser sessions) or 24 hours (signed-in people, principals);
- **Retention.** Finished jobs expire after 1 hour (browser sessions) or 24 hours (signed-in people, principals);
- the pruner deletes the row, its events, its files and its Media objects.
+ the pruner deletes the row, its events, its files and its Media objects. It never touches a job that has a
+ reference (above). If Media will not delete a result object (a retention hold, `409 media.object.held`), cannot
+ be reached, or is no longer configured here, the job is kept, record and all, and looked at again 24 hours
+ later, so no object is left without the record that would delete it.
- **Results in OpenVibe.Media** when `TOOLS_JOB_RESULTS=media`: each result file becomes a private Media object
- **Results in OpenVibe.Media** when `TOOLS_JOB_RESULTS=media`: each result file becomes a private Media object
(v2 object API, namespace `tools`, owner `X-OV-Subject` for signed-in people) and the job's result carries its
(v2 object API, namespace `tools`, owner `X-OV-Subject` for signed-in people) and the job's result carries its
`media.media_id`; previews redirect to a short-lived signed Media URL, downloads stream through the satellite.
`media.media_id`; previews redirect to a short-lived signed Media URL, downloads stream through the satellite.
@@ -114,7 +114,7 @@
@@ -114,7 +114,7 @@
if (!f) return null;
if (!f) return null;
return f.url + (opts && opts.inline ? '?inline=1' : '');
return f.url + (opts && opts.inline ? '?inline=1' : '');
}
}
- /** Milliseconds until a finished job's files expire (null while it is still running). */
+ /** Milliseconds until a finished job's files expire (null while it is still running, or while something references it). */
function expiresIn(job) {
function expiresIn(job) {
return job && job.expires_at ? Math.max(0, Date.parse(job.expires_at) - Date.now()) : null;
return job && job.expires_at ? Math.max(0, Date.parse(job.expires_at) - Date.now()) : null;
}
}
@@ -10,6 +10,9 @@
@@ -10,6 +10,9 @@
// POST /api/v1/jobs/:id/retry retry a failed job as a new job → 202 + Location (new) |
// 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)
// 200 + Idempotent-Replayed: true (it was already retried: that job)
// | 409 tools.job.not_failed | 410 tools.job.inputs_gone
// | 409 tools.job.not_failed | 410 tools.job.inputs_gone
+// PUT /api/v1/jobs/:id/references/:ref keep a succeeded job's result while <ref> points at it
+// (e.g. community:paste:p_123) → 201 new | 200 already there
+// DELETE /api/v1/jobs/:id/references/:ref drop it → 200 (the expiry clock restarts after the last one)
// GET /api/v1/jobs/:id/events SSE: job.queued|running|progress|cancel_requested|succeeded|failed|cancelled;
// 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
// every event has an id; reconnecting with Last-Event-ID replays
// only what came after it; 204 once there is nothing left to say
// only what came after it; 204 once there is nothing left to say
@@ -197,7 +200,7 @@ function mountJobRoutes(app, o) {
@@ -197,7 +200,7 @@ function mountJobRoutes(app, o) {
return res.status(job.state === 'cancelled' ? 200 : 202).json(system.view(job));
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.
+ // Retry and references are writes: a principal needs tools.job.create, like a submit.
app.post('/api/v1/jobs/:id/retry', ...(o.limiters || []), (req, res) => {
app.post('/api/v1/jobs/:id/retry', ...(o.limiters || []), (req, res) => {
noStore(res);
noStore(res);
const row = load(req, res, 'create');
const row = load(req, res, 'create');
@@ -214,6 +217,24 @@ function mountJobRoutes(app, o) {
@@ -214,6 +217,24 @@ function mountJobRoutes(app, o) {
}
}
});
});
+ function referenceRoute(fn) {
+ return (req, res) => {
+ noStore(res);
+ const row = load(req, res, 'create');
+ if (!row) return;
+ try {
+ const r = fn(row.id, String(req.params.ref || ''));
+ return res.status(r.created ? 201 : 200).json(system.view(r.job));
+ } catch (err) {
+ if (err instanceof system.JobError) return send(res, err.status, err.code, err.detail, err.extra);
+ console.error('[Jobs] reference failed:', err.message);
+ return send(res, 500, 'tools.job.reference_failed', 'The reference could not be changed');
+ }
+ };
+ }
+ app.put('/api/v1/jobs/:id/references/:ref', referenceRoute((id, ref) => system.reference(id, ref)));
+ app.delete('/api/v1/jobs/:id/references/:ref', referenceRoute((id, ref) => system.unreference(id, ref)));
+
app.get('/api/v1/jobs/:id/events', (req, res) => {
app.get('/api/v1/jobs/:id/events', (req, res) => {
const row = load(req, res, 'read');
const row = load(req, res, 'read');
if (!row) return;
if (!row) return;
@@ -2,10 +2,12 @@
@@ -2,10 +2,12 @@
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
// Job store — the satellite's own SQLite (better-sqlite3 handle passed in).
// Job store — the satellite's own SQLite (better-sqlite3 handle passed in).
//
//
-// Two tables, created on first use (idempotent):
-// tool_jobs one row per job; the durable truth a restart recovers from
-// tool_job_events the ordered event log per job; its AUTOINCREMENT seq is the SSE event id,
-// so a reconnect with Last-Event-ID replays exactly what was missed
+// Three tables, created on first use (idempotent):
+// tool_jobs one row per job; the durable truth a restart recovers from
+// tool_job_events the ordered event log per job; its AUTOINCREMENT seq is the SSE event id,
+// so a reconnect with Last-Event-ID replays exactly what was missed
+// tool_job_references who still points at a job's result (a paste, a project, …); a referenced
+// job never expires, so its files and Media objects are not pruned
// ═══════════════════════════════════════════════════════════════
// ═══════════════════════════════════════════════════════════════
const TERMINAL = ['succeeded', 'failed', 'cancelled'];
const TERMINAL = ['succeeded', 'failed', 'cancelled'];
@@ -50,6 +52,12 @@ CREATE TABLE IF NOT EXISTS tool_job_events (
@@ -50,6 +52,12 @@ CREATE TABLE IF NOT EXISTS tool_job_events (
at INTEGER NOT NULL
at INTEGER NOT NULL
);
);
CREATE INDEX IF NOT EXISTS tool_job_events_job ON tool_job_events(job_id, seq);
CREATE INDEX IF NOT EXISTS tool_job_events_job ON tool_job_events(job_id, seq);
+CREATE TABLE IF NOT EXISTS tool_job_references (
+ job_id TEXT NOT NULL,
+ ref TEXT NOT NULL,
+ created_at INTEGER NOT NULL,
+ PRIMARY KEY (job_id, ref)
+);
`;
`;
// Columns added after the first release, each added when missing (idempotent).
// Columns added after the first release, each added when missing (idempotent).
@@ -88,9 +96,19 @@ function createStore(db) {
@@ -88,9 +96,19 @@ function createStore(db) {
event: db.prepare('INSERT INTO tool_job_events (job_id, event, data, at) VALUES (?, ?, ?, ?)'),
event: db.prepare('INSERT INTO tool_job_events (job_id, event, data, at) VALUES (?, ?, ?, ?)'),
eventsAfter: db.prepare('SELECT seq, event, data, at FROM tool_job_events WHERE job_id = ? AND seq > ? ORDER BY seq'),
eventsAfter: db.prepare('SELECT seq, event, data, at FROM tool_job_events WHERE job_id = ? AND seq > ? ORDER BY seq'),
lastSeq: db.prepare('SELECT MAX(seq) AS seq FROM tool_job_events WHERE job_id = ?'),
lastSeq: db.prepare('SELECT MAX(seq) AS seq FROM tool_job_events WHERE job_id = ?'),
- expired: db.prepare('SELECT * FROM tool_jobs WHERE expires_at IS NOT NULL AND expires_at <= ? LIMIT 500'),
+ // A job something still references is never expired (see tool_job_references).
+ expired: db.prepare(`SELECT * FROM tool_jobs j WHERE expires_at IS NOT NULL AND expires_at <= ?
+ AND NOT EXISTS (SELECT 1 FROM tool_job_references r WHERE r.job_id = j.id) LIMIT 500`),
+ deferExpiry: db.prepare('UPDATE tool_jobs SET expires_at = @until, updated_at = @now WHERE id = @id'),
del: db.prepare('DELETE FROM tool_jobs WHERE id = ?'),
del: db.prepare('DELETE FROM tool_jobs WHERE id = ?'),
delEvents: db.prepare('DELETE FROM tool_job_events WHERE job_id = ?'),
delEvents: db.prepare('DELETE FROM tool_job_events WHERE job_id = ?'),
+ delRefs: db.prepare('DELETE FROM tool_job_references WHERE job_id = ?'),
+ addRef: db.prepare('INSERT OR IGNORE INTO tool_job_references (job_id, ref, created_at) VALUES (?, ?, ?)'),
+ dropRef: db.prepare('DELETE FROM tool_job_references WHERE job_id = ? AND ref = ?'),
+ refs: db.prepare('SELECT ref, created_at FROM tool_job_references WHERE job_id = ? ORDER BY created_at, ref'),
+ refCount: db.prepare('SELECT COUNT(*) AS n FROM tool_job_references WHERE job_id = ?'),
+ // After the last reference goes, the result is kept at least one more ttl.
+ releaseExpiry: db.prepare('UPDATE tool_jobs SET expires_at = MAX(COALESCE(expires_at, 0), @now + ttl_ms), updated_at = @now WHERE id = @id AND finished_at IS NOT NULL'),
counts: db.prepare('SELECT state, COUNT(*) AS n FROM tool_jobs GROUP BY state'),
counts: db.prepare('SELECT state, COUNT(*) AS n FROM tool_jobs GROUP BY state'),
};
};
@@ -123,7 +141,18 @@ function createStore(db) {
@@ -123,7 +141,18 @@ function createStore(db) {
eventsAfter: (jobId, seq) => q.eventsAfter.all(jobId, seq).map(e => ({ seq: e.seq, event: e.event, data: parse(e.data, {}), at: e.at })),
eventsAfter: (jobId, seq) => q.eventsAfter.all(jobId, seq).map(e => ({ seq: e.seq, event: e.event, data: parse(e.data, {}), at: e.at })),
lastSeq: (jobId) => q.lastSeq.get(jobId).seq || 0,
lastSeq: (jobId) => q.lastSeq.get(jobId).seq || 0,
expired: (now = Date.now()) => q.expired.all(now),
expired: (now = Date.now()) => q.expired.all(now),
- remove: db.transaction((id) => { q.delEvents.run(id); q.del.run(id); }),
+ deferExpiry: (id, until, now = Date.now()) => q.deferExpiry.run({ id, until, now }),
+ remove: db.transaction((id) => { q.delEvents.run(id); q.delRefs.run(id); q.del.run(id); }),
+ /** → true when the reference is new. */
+ addReference: (id, ref, now = Date.now()) => q.addRef.run(id, ref, now).changes === 1,
+ /** → true when it existed; dropping the last one restarts the job's expiry clock. */
+ dropReference: db.transaction((id, ref, now = Date.now()) => {
+ const gone = q.dropRef.run(id, ref).changes === 1;
+ if (gone && q.refCount.get(id).n === 0) q.releaseExpiry.run({ id, now });
+ return gone;
+ }),
+ references: (id) => q.refs.all(id),
+ referenceCount: (id) => q.refCount.get(id).n,
counts() { const out = Object.fromEntries(STATES.map(s => [s, 0])); for (const r of q.counts.all()) out[r.state] = r.n; return out; },
counts() { const out = Object.fromEntries(STATES.map(s => [s, 0])); for (const r of q.counts.all()) out[r.state] = r.n; return out; },
transaction: (fn) => db.transaction(fn),
transaction: (fn) => db.transaction(fn),
parse,
parse,
@@ -11,7 +11,10 @@
@@ -11,7 +11,10 @@
//
//
// Bounded: at most `concurrency` jobs run at once in this process, and each owner may have at
// Bounded: at most `concurrency` jobs run at once in this process, and each owner may have at
// most `maxActivePerOwner` queued + running jobs. Finished jobs expire (ttl chosen at submit),
// 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.
+// and the pruner deletes the row, its events, its files and any Media objects it made — except
+// while something references the result (reference(), e.g. a paste or a project that points at the
+// Media object), and never while Media keeps an object (a retention hold, or Media unreachable):
+// then the job is kept and looked at again later, so no referenced result loses its record.
//
//
// Retry: a failed job keeps its input files until it expires; retry(id) moves them to a new job
// 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
// (retry_of → the failed one, which records retried_by). Retrying the same failed job again
@@ -27,6 +30,9 @@ const crypto = require('crypto');
@@ -27,6 +30,9 @@ const crypto = require('crypto');
const { createStore, TERMINAL } = require('./store');
const { createStore, TERMINAL } = require('./store');
const HOUR = 60 * 60 * 1000;
const HOUR = 60 * 60 * 1000;
+const PRUNE_RECHECK_MS = 24 * HOUR; // an expired job the pruner could not finish is looked at again after this
+const REF_RE = /^[a-z][a-z0-9_-]*(?::[A-Za-z0-9_.-]+){1,4}$/; // <service>:<kind>:<id>, e.g. community:paste:p_123
+const MAX_REFS = 50;
const EVENT_FOR = { queued: 'job.queued', running: 'job.running', succeeded: 'job.succeeded', failed: 'job.failed', cancelled: 'job.cancelled' };
const EVENT_FOR = { queued: 'job.queued', running: 'job.running', succeeded: 'job.succeeded', failed: 'job.failed', cancelled: 'job.cancelled' };
const SAFE_NAME = (s) => String(s || 'file').replace(/[/\\\0]/g, '_').replace(/[^\w.\- ]/g, '_').slice(-120) || 'file';
const SAFE_NAME = (s) => String(s || 'file').replace(/[/\\\0]/g, '_').replace(/[^\w.\- ]/g, '_').slice(-120) || 'file';
@@ -122,6 +128,7 @@ function createJobSystem(o) {
@@ -122,6 +128,7 @@ function createJobSystem(o) {
function view(row) {
function view(row) {
if (!row) return null;
if (!row) return null;
const live = row.state === 'queued' || row.state === 'running';
const live = row.state === 'queued' || row.state === 'running';
+ const references = store.references(row.id).map(r => ({ ref: r.ref, created_at: iso(r.created_at) }));
return {
return {
id: row.id,
id: row.id,
object: 'tools.job',
object: 'tools.job',
@@ -136,12 +143,14 @@ function createJobSystem(o) {
@@ -136,12 +143,14 @@ function createJobSystem(o) {
created_at: iso(row.created_at),
created_at: iso(row.created_at),
started_at: iso(row.started_at),
started_at: iso(row.started_at),
finished_at: iso(row.finished_at),
finished_at: iso(row.finished_at),
- expires_at: iso(row.expires_at),
+ // null while something references the result: it is kept until the last reference goes.
+ expires_at: references.length ? null : iso(row.expires_at),
result: publicResult(store.parse(row.result_json, null), row.id),
result: publicResult(store.parse(row.result_json, null), row.id),
error: store.parse(row.error_json, null),
error: store.parse(row.error_json, null),
retryable: !!row.retryable,
retryable: !!row.retryable,
retry_of: row.retry_of || null,
retry_of: row.retry_of || null,
retried_by: row.retried_by || null,
retried_by: row.retried_by || null,
+ references,
links: {
links: {
self: `/api/v1/jobs/${row.id}`,
self: `/api/v1/jobs/${row.id}`,
events: `/api/v1/jobs/${row.id}/events`,
events: `/api/v1/jobs/${row.id}/events`,
@@ -439,6 +448,35 @@ function createJobSystem(o) {
@@ -439,6 +448,35 @@ function createJobSystem(o) {
return { job: store.get(next), replayed: false };
return { job: store.get(next), replayed: false };
}
}
+ // ── References (results something still points at) ─────────
+ function checkRef(ref) {
+ if (!REF_RE.test(String(ref || '')) || String(ref).length > 200) {
+ throw new JobError(400, 'tools.job.invalid', 'A reference is <service>:<kind>:<id>, e.g. community:paste:p_123 (at most 200 characters)');
+ }
+ return String(ref);
+ }
+ /**
+ * Keep a succeeded job's result while `ref` points at it. → { job, created }. Idempotent.
+ */
+ function reference(id, ref) {
+ ref = checkRef(ref);
+ const row = store.get(id);
+ if (!row) return null;
+ if (row.state !== 'succeeded') throw new JobError(409, 'tools.job.not_succeeded', `Only a succeeded job's result can be referenced; this one is ${row.state}`, { state: row.state });
+ if (row.env === 'sandbox') throw new JobError(409, 'tools.job.sandbox', 'Sandbox results are kept briefly and cannot be referenced');
+ const has = store.references(id).some(r => r.ref === ref);
+ if (!has && store.referenceCount(id) >= MAX_REFS) throw new JobError(409, 'tools.job.too_many_references', `At most ${MAX_REFS} references per job`);
+ const created = store.addReference(id, ref);
+ return { job: store.get(id), created };
+ }
+ /** Stop keeping the result for `ref`. → { job, removed }. Idempotent. */
+ function unreference(id, ref) {
+ ref = checkRef(ref);
+ if (!store.get(id)) return null;
+ const removed = store.dropReference(id, ref);
+ return { job: store.get(id), removed };
+ }
+
// ── Boot recovery, pruning ─────────────────────────────────
// ── Boot recovery, pruning ─────────────────────────────────
function recover() {
function recover() {
let requeued = 0, failed = 0;
let requeued = 0, failed = 0;
@@ -471,11 +509,30 @@ function createJobSystem(o) {
@@ -471,11 +509,30 @@ function createJobSystem(o) {
return { requeued, failed };
return { requeued, failed };
}
}
+ /**
+ * Delete expired jobs. Referenced jobs are never expired (store.expired skips them). A job whose
+ * Media objects cannot all be deleted — Media keeps one under a retention hold (409
+ * media.object.held), Media is unreachable, or this process no longer has Media configured —
+ * is kept, record and all, and looked at again after PRUNE_RECHECK_MS: dropping the record
+ * would leave an object nobody deletes, and a held object is one somebody still needs.
+ */
async function prune(now = Date.now()) {
async function prune(now = Date.now()) {
let n = 0;
let n = 0;
for (const row of store.expired(now)) {
for (const row of store.expired(now)) {
const result = store.parse(row.result_json, null);
const result = store.parse(row.result_json, null);
- if (media && result) for (const f of result.files || []) if (f.media && f.media.media_id) await media.remove(f.media.media_id).catch(err => log.warn(`[Jobs] could not delete ${f.media.media_id}:`, err.message));
+ let kept = null;
+ for (const f of (result && result.files) || []) {
+ if (!f.media || !f.media.media_id) continue;
+ if (!media) { kept = 'results in OpenVibe.Media, but Media is not configured here'; continue; }
+ try { await media.remove(f.media.media_id); } catch (err) {
+ kept = err.code === 'media.object.held' ? `${f.media.media_id} is under a retention hold` : `could not delete ${f.media.media_id}: ${err.message}`;
+ }
+ }
+ if (kept) {
+ log.warn(`[Jobs] kept expired job ${row.id} (${kept}); checking again later`);
+ store.deferExpiry(row.id, now + PRUNE_RECHECK_MS);
+ continue;
+ }
await fsp.rm(jobDir(row.id), { recursive: true, force: true }).catch(() => {});
await fsp.rm(jobDir(row.id), { recursive: true, force: true }).catch(() => {});
store.remove(row.id);
store.remove(row.id);
listeners.delete(row.id);
listeners.delete(row.id);
@@ -520,7 +577,7 @@ function createJobSystem(o) {
@@ -520,7 +577,7 @@ function createJobSystem(o) {
}
}
const api = {
const api = {
- define, submit, cancel, retry, subscribe, start, stop, onStop, prune, recover, view, resultFile,
+ define, submit, cancel, retry, reference, unreference, subscribe, start, stop, onStop, prune, recover, view, resultFile,
get: (id) => store.get(id),
get: (id) => store.get(id),
eventsAfter: (id, seq) => store.eventsAfter(id, seq),
eventsAfter: (id, seq) => store.eventsAfter(id, seq),
lastSeq: (id) => store.lastSeq(id),
lastSeq: (id) => store.lastSeq(id),
@@ -3,7 +3,8 @@
@@ -3,7 +3,8 @@
// an accepted job survives a restart (DB and app closed and reopened), running jobs are re-queued
// 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,
// 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, results stored
// 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.
+// as Media objects through Media's v2 object API (with the local fallback), retention of results
+// that are referenced or held in Media, and retrying a failed job.
const assert = require('assert');
const assert = require('assert');
const fs = require('fs');
const fs = require('fs');
const os = require('os');
const os = require('os');
@@ -80,7 +81,7 @@ const flaky = new Set();
@@ -80,7 +81,7 @@ const flaky = new Set();
function fakeMedia() {
function fakeMedia() {
const objects = new Map();
const objects = new Map();
const app = express();
const app = express();
- const state = { failInit: false, deleted: [] };
+ const state = { failInit: false, deleted: [], held: new Set(), deleteDown: false };
app.use(express.json());
app.use(express.json());
const authed = (req, res, next) => (req.headers.authorization === 'Bearer media-token' ? next() : res.status(401).json({ error: 'no' }));
const authed = (req, res, next) => (req.headers.authorization === 'Bearer media-token' ? next() : res.status(401).json({ error: 'no' }));
app.post('/api/v2/tools/objects', authed, (req, res) => {
app.post('/api/v2/tools/objects', authed, (req, res) => {
@@ -103,7 +104,11 @@ function fakeMedia() {
@@ -103,7 +104,11 @@ function fakeMedia() {
res.json({ id: o.id, lifecycle_status: 'ready' });
res.json({ id: o.id, lifecycle_status: 'ready' });
});
});
app.get('/api/v2/tools/objects/:id/download', authed, (req, res) => res.json({ url: `https://openvibe.media/o/${req.params.id}?exp=1&sig=s`, expires_at: 'x', public: false }));
app.get('/api/v2/tools/objects/:id/download', authed, (req, res) => res.json({ url: `https://openvibe.media/o/${req.params.id}?exp=1&sig=s`, expires_at: 'x', public: false }));
- app.delete('/api/v2/tools/objects/:id', authed, (req, res) => { state.deleted.push(req.params.id); objects.delete(req.params.id); res.json({ ok: true }); });
+ app.delete('/api/v2/tools/objects/:id', authed, (req, res) => {
+ if (state.deleteDown) return res.status(503).json({ code: 'media.unavailable' });
+ if (state.held.has(req.params.id)) return res.status(409).type('application/problem+json').send(JSON.stringify({ status: 409, code: 'media.object.held', detail: 'Object is under a retention hold' }));
+ state.deleted.push(req.params.id); objects.delete(req.params.id); res.json({ ok: true });
+ });
app.get('/o/:id', (req, res) => { const o = objects.get(req.params.id); if (!o || req.query.sig !== 's') return res.status(404).end(); res.type('text/plain').send(o.bytes); });
app.get('/o/:id', (req, res) => { const o = objects.get(req.params.id); if (!o || req.query.sig !== 's') return res.status(404).end(); res.type('text/plain').send(o.bytes); });
return { app, objects, state };
return { app, objects, state };
}
}
@@ -369,6 +374,60 @@ async function readSse(url, { headers = {}, stop = () => false, ms = 5000 } = {}
@@ -369,6 +374,60 @@ async function readSse(url, { headers = {}, stop = () => false, ms = 5000 } = {}
await s.system.prune(Date.now() + 25 * 60 * 60 * 1000);
await s.system.prune(Date.now() + 25 * 60 * 60 * 1000);
assert.ok(fm.state.deleted.includes(f0.media.media_id), 'pruning deletes the Media object');
assert.ok(fm.state.deleted.includes(f0.media.media_id), 'pruning deletes the Media object');
await s.close();
await s.close();
+
+ // ── Retention: referenced or held results are not pruned ──
+ fm.state.failInit = false;
+ const DAY = 24 * 60 * 60 * 1000;
+ s = await satellite(path.join(root, 'g'), { media });
+ const gina = client(s.base, { user: USER_A });
+ const finished = async (text) => {
+ const id = (await gina('/api/v1/jobs', json({ type: 'test.upper', input: { text } }))).body.id;
+ return until(async () => { const g = await gina(`/api/v1/jobs/${id}`); return g.body.state === 'succeeded' && g.body; }, `job ${text}`);
+ };
+ const kept = await finished('kept by a paste');
+ const keptMedia = kept.result.files[0].media.media_id;
+ r = await gina(`/api/v1/jobs/${kept.id}/references/community:paste:p_123`, { method: 'PUT' });
+ assert.strictEqual(r.status, 201, 'a reference is recorded');
+ assert.deepStrictEqual(r.body.references.map(x => x.ref), ['community:paste:p_123']);
+ assert.strictEqual(r.body.expires_at, null, 'a referenced result does not expire');
+ assert.strictEqual((await gina(`/api/v1/jobs/${kept.id}/references/community:paste:p_123`, { method: 'PUT' })).status, 200, 'referencing twice is fine');
+ assert.strictEqual((await client(s.base, { user: USER_B })(`/api/v1/jobs/${kept.id}/references/community:paste:p_9`, { method: 'PUT' })).status, 404, 'only the owner can reference it');
+ r = await gina(`/api/v1/jobs/${kept.id}/references/${encodeURIComponent('no spaces allowed')}`, { method: 'PUT' });
+ assert.strictEqual(r.status, 400); assert.strictEqual(r.body.code, 'tools.job.invalid');
+ const held = await finished('held by Media');
+ const heldMedia = held.result.files[0].media.media_id;
+ fm.state.held.add(heldMedia);
+ const plain = await finished('nobody needs this');
+ fm.state.deleted.length = 0;
+ await s.system.prune(Date.now() + 2 * DAY);
+ assert.strictEqual((await gina(`/api/v1/jobs/${kept.id}`)).status, 200, 'a referenced job survives pruning');
+ assert.ok(fm.objects.has(keptMedia) && !fm.state.deleted.includes(keptMedia), 'and so does its Media object');
+ assert.strictEqual((await gina(`/api/v1/jobs/${kept.id}/files/0`)).body, 'KEPT BY A PASTE', 'still downloadable');
+ const stillHeld = (await gina(`/api/v1/jobs/${held.id}`)).body;
+ assert.ok(stillHeld.id && Date.parse(stillHeld.expires_at) > Date.now() + 2 * DAY, 'a job whose Media object is held keeps its record and is checked again later');
+ assert.ok(fm.objects.has(heldMedia));
+ assert.strictEqual((await gina(`/api/v1/jobs/${plain.id}`)).status, 404, 'an unreferenced job is pruned as before');
+ assert.deepStrictEqual(fm.state.deleted, [plain.result.files[0].media.media_id]);
+ // Media unreachable: nothing is dropped.
+ const outage = await finished('media is down');
+ fm.state.deleteDown = true;
+ await s.system.prune(Date.now() + 2 * DAY);
+ assert.strictEqual((await gina(`/api/v1/jobs/${outage.id}`)).status, 200, 'a failed Media delete keeps the job');
+ fm.state.deleteDown = false;
+ // Dropping the last reference restarts the clock; after that the job expires normally.
+ r = await gina(`/api/v1/jobs/${kept.id}/references/community:paste:p_123`, { method: 'DELETE' });
+ assert.strictEqual(r.status, 200); assert.deepStrictEqual(r.body.references, []);
+ assert.ok(Date.parse(r.body.expires_at) >= Date.now() + DAY - 60000, 'kept one more ttl after the last reference goes');
+ assert.strictEqual((await gina(`/api/v1/jobs/${kept.id}/references/community:paste:p_123`, { method: 'DELETE' })).status, 200, 'dropping twice is fine');
+ fm.state.held.delete(heldMedia);
+ await s.system.prune(Date.now() + 5 * DAY);
+ for (const j of [kept, held, outage]) assert.strictEqual((await gina(`/api/v1/jobs/${j.id}`)).status, 404, `${j.id} pruned once nothing keeps it`);
+ assert.ok([keptMedia, heldMedia].every(id => fm.state.deleted.includes(id)));
+ r = await gina('/api/v1/jobs', json({ type: 'test.boom' }));
+ await until(async () => (await gina(`/api/v1/jobs/${r.body.id}`)).body.state === 'failed', 'failed job');
+ r = await gina(`/api/v1/jobs/${r.body.id}/references/community:paste:p_1`, { method: 'PUT' });
+ assert.strictEqual(r.status, 409); assert.strictEqual(r.body.code, 'tools.job.not_succeeded');
+ await s.close();
mediaServer.close();
mediaServer.close();
// ── Retry a failed job ──────────────────────────────────
// ── Retry a failed job ──────────────────────────────────