openvibers/openvibe.tools Commits
Tools job store: events and references go through the owning app (plan T8 review): tool_job_events and tool_job_references carry no app, so their reads, inserts and deletes now join or check this app's tool_jobs row, and every write by job id filters on app too; another app that knows an id reads and changes nothing. jobs-store-scope.test.js.
Changes
@@@ -30,46 +30,52 @@ function createStore(db, { app } = {}) {const q = {insert: db.prepare(`INSERT INTO tool_jobs (id, app, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool, project_id, trace_id)VALUES (@id, @app, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool, @project_id, @trace_id)`),-- markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND state = 'failed' AND retried_by IS NULL"),++ markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND app = @app AND state = 'failed' AND retried_by IS NULL"),get: db.prepare('SELECT * FROM tool_jobs WHERE id = ? AND app = ?'),byIdem: db.prepare('SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND idempotency_key = @idempotency_key'),// A result file stored in Media, by its media_id (a run's { media_id } reference); the owner's only.byResultMedia: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND state = 'succeeded' AND result_json ILIKE @like ORDER BY id DESC LIMIT 5"),nextQueued: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND state = 'queued' ORDER BY id LIMIT @limit"),-- claim: db.prepare("UPDATE tool_jobs SET state = 'running', attempts = attempts + 1, started_at = @now, updated_at = @now, progress = NULL, progress_message = NULL WHERE id = @id AND state = 'queued'"),++ claim: db.prepare("UPDATE tool_jobs SET state = 'running', attempts = attempts + 1, started_at = @now, updated_at = @now, progress = NULL, progress_message = NULL WHERE id = @id AND app = @app AND state = 'queued'"),running: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND state = 'running'"),activeForOwner: db.prepare("SELECT COUNT(*) AS n FROM tool_jobs WHERE app = @app AND owner = @owner AND state IN ('queued','running')"),activeForIp: db.prepare("SELECT COUNT(*) AS n FROM tool_jobs WHERE app = @app AND ip_key = @ip_key AND state IN ('queued','running')"),-- progress: db.prepare('UPDATE tool_jobs SET progress = @progress, progress_message = @message, updated_at = @now WHERE id = @id AND state = \'running\''),-- requestCancel: db.prepare("UPDATE tool_jobs SET cancel_requested = 1, updated_at = @now WHERE id = @id AND state IN ('queued','running')"),-- requeue: db.prepare("UPDATE tool_jobs SET state = 'queued', started_at = NULL, progress = NULL, progress_message = NULL, updated_at = @now WHERE id = @id AND state = 'running'"),++ progress: db.prepare('UPDATE tool_jobs SET progress = @progress, progress_message = @message, updated_at = @now WHERE id = @id AND app = @app AND state = \'running\''),++ requestCancel: db.prepare("UPDATE tool_jobs SET cancel_requested = 1, updated_at = @now WHERE id = @id AND app = @app AND state IN ('queued','running')"),++ requeue: db.prepare("UPDATE tool_jobs SET state = 'queued', started_at = NULL, progress = NULL, progress_message = NULL, updated_at = @now WHERE id = @id AND app = @app AND state = 'running'"),finish: db.prepare(`UPDATE tool_jobs SET state = @state, result_json = @result_json, error_json = @error_json, retryable = @retryable,progress = CASE WHEN @state = 'succeeded' THEN 100 ELSE progress END, progress_message = NULL,-- finished_at = @now, updated_at = @now, expires_at = @now + ttl_ms::bigint WHERE id = @id AND state IN ('queued','running')`),-- setFiles: db.prepare('UPDATE tool_jobs SET files_json = @files_json, updated_at = @now WHERE id = @id'),-- event: db.prepare('INSERT INTO tool_job_events (job_id, event, data, at) VALUES (@job_id, @event, @data, @at) RETURNING 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 = ?'),++ finished_at = @now, updated_at = @now, expires_at = @now + ttl_ms::bigint WHERE id = @id AND app = @app AND state IN ('queued','running')`),++ setFiles: db.prepare('UPDATE tool_jobs SET files_json = @files_json, updated_at = @now WHERE id = @id AND app = @app'),++ // tool_job_events and tool_job_references carry no `app` of their own: every statement on them++ // goes through this app's tool_jobs row, so an id from another app reads and changes nothing.++ event: db.prepare(`INSERT INTO tool_job_events (job_id, event, data, at)++ SELECT CAST(@job_id AS text), CAST(@event AS text), CAST(@data AS text), CAST(@at AS bigint) WHERE EXISTS (SELECT 1 FROM tool_jobs WHERE id = @job_id AND app = @app) RETURNING seq`),++ eventsAfter: db.prepare(`SELECT e.seq, e.event, e.data, e.at FROM tool_job_events e JOIN tool_jobs j ON j.id = e.job_id++ WHERE e.job_id = @job_id AND j.app = @app AND e.seq > @seq ORDER BY e.seq`),++ lastSeq: db.prepare('SELECT MAX(e.seq) AS seq FROM tool_job_events e JOIN tool_jobs j ON j.id = e.job_id WHERE e.job_id = @job_id AND j.app = @app'),// A job something still references is never expired (see tool_job_references).expired: db.prepare(`SELECT * FROM tool_jobs j WHERE app = @app AND expires_at IS NOT NULL AND expires_at <= @nowAND 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 INTO tool_job_references (job_id, ref, created_at) VALUES (?, ?, ?) ON CONFLICT DO NOTHING'),-- 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 = ?'),++ deferExpiry: db.prepare('UPDATE tool_jobs SET expires_at = @until, updated_at = @now WHERE id = @id AND app = @app'),++ del: db.prepare('DELETE FROM tool_jobs WHERE id = @id AND app = @app'),++ delEvents: db.prepare('DELETE FROM tool_job_events WHERE job_id = @id AND EXISTS (SELECT 1 FROM tool_jobs WHERE id = @id AND app = @app)'),++ delRefs: db.prepare('DELETE FROM tool_job_references WHERE job_id = @id AND EXISTS (SELECT 1 FROM tool_jobs WHERE id = @id AND app = @app)'),++ addRef: db.prepare(`INSERT INTO tool_job_references (job_id, ref, created_at)++ SELECT CAST(@id AS text), CAST(@ref AS text), CAST(@now AS bigint) WHERE EXISTS (SELECT 1 FROM tool_jobs WHERE id = @id AND app = @app) ON CONFLICT DO NOTHING`),++ dropRef: db.prepare('DELETE FROM tool_job_references WHERE job_id = @id AND ref = @ref AND EXISTS (SELECT 1 FROM tool_jobs WHERE id = @id AND app = @app)'),++ refs: db.prepare(`SELECT r.ref, r.created_at FROM tool_job_references r JOIN tool_jobs j ON j.id = r.job_id++ WHERE r.job_id = @id AND j.app = @app ORDER BY r.created_at, r.ref`),++ refCount: db.prepare('SELECT COUNT(*) AS n FROM tool_job_references r JOIN tool_jobs j ON j.id = r.job_id WHERE r.job_id = @id AND j.app = @app'),// After the last reference goes, the result is kept at least one more ttl.-- releaseExpiry: db.prepare('UPDATE tool_jobs SET expires_at = GREATEST(COALESCE(expires_at, 0), @now + ttl_ms::bigint), updated_at = @now WHERE id = @id AND finished_at IS NOT NULL'),++ releaseExpiry: db.prepare('UPDATE tool_jobs SET expires_at = GREATEST(COALESCE(expires_at, 0), @now + ttl_ms::bigint), updated_at = @now WHERE id = @id AND app = @app AND finished_at IS NOT NULL'),counts: db.prepare('SELECT state, COUNT(*) AS n FROM tool_jobs WHERE app = @app GROUP BY state'),};/** Append one event; returns its seq (the SSE id). */async function appendEvent(jobId, event, data, now = Date.now()) {-- const r = await q.event.run({ job_id: jobId, event, data: JSON.stringify(data), at: now });-- return Number(r.lastInsertRowid);++ const r = await q.event.run({ app, job_id: jobId, event, data: JSON.stringify(data), at: now });++ return r.changes === 1 ? Number(r.lastInsertRowid) : 0; // 0: no such job in this app}return {@@@ -78,39 +84,39 @@ function createStore(db, { app } = {}) {TERMINAL, STATES,async insert(row) { await q.insert.run({ app, retry_of: null, ip_key: null, tool: null, project_id: null, trace_id: null, ...row }); },byResultMedia: async (owner, mediaId) => q.byResultMedia.all({ app, owner, like: `%"media_id":"${String(mediaId).replace(/[%"\\]/g, '')}"%` }),-- markRetried: async (id, next, now = Date.now()) => (await q.markRetried.run({ id, next, now })).changes === 1,++ markRetried: async (id, next, now = Date.now()) => (await q.markRetried.run({ app, id, next, now })).changes === 1,get: async (id) => (await q.get.get(id, app)) || null,byIdempotencyKey: async (owner, key) => (await q.byIdem.get({ app, owner, idempotency_key: key })) || null,nextQueued: (limit) => q.nextQueued.all({ app, limit }),-- claim: async (id, now = Date.now()) => (await q.claim.run({ id, now })).changes === 1,++ claim: async (id, now = Date.now()) => (await q.claim.run({ app, id, now })).changes === 1,running: () => q.running.all({ app }),activeForOwner: async (owner) => (await q.activeForOwner.get({ app, owner })).n,activeForIp: async (ipKey) => (ipKey ? (await q.activeForIp.get({ app, ip_key: ipKey })).n : 0),-- setProgress: async (id, progress, message, now = Date.now()) => (await q.progress.run({ id, progress, message, now })).changes === 1,-- requestCancel: async (id, now = Date.now()) => (await q.requestCancel.run({ id, now })).changes === 1,-- requeue: async (id, now = Date.now()) => (await q.requeue.run({ id, now })).changes === 1,++ setProgress: async (id, progress, message, now = Date.now()) => (await q.progress.run({ app, id, progress, message, now })).changes === 1,++ requestCancel: async (id, now = Date.now()) => (await q.requestCancel.run({ app, id, now })).changes === 1,++ requeue: async (id, now = Date.now()) => (await q.requeue.run({ app, id, now })).changes === 1,finish: async (id, { state, result = null, error = null, retryable = false }, now = Date.now()) => (await q.finish.run({-- id, state, now, retryable: retryable ? 1 : 0,++ app, id, state, now, retryable: retryable ? 1 : 0,result_json: result == null ? null : JSON.stringify(result),error_json: error == null ? null : JSON.stringify(error),})).changes === 1,-- setFiles: (id, files, now = Date.now()) => q.setFiles.run({ id, files_json: JSON.stringify(files), now }),++ setFiles: (id, files, now = Date.now()) => q.setFiles.run({ app, id, files_json: JSON.stringify(files), now }),appendEvent,-- eventsAfter: async (jobId, seq) => (await q.eventsAfter.all(jobId, seq)).map(e => ({ seq: Number(e.seq), event: e.event, data: parse(e.data, {}), at: Number(e.at) })),-- lastSeq: async (jobId) => Number((await q.lastSeq.get(jobId)).seq) || 0,++ eventsAfter: async (jobId, seq) => (await q.eventsAfter.all({ app, job_id: jobId, seq })).map(e => ({ seq: Number(e.seq), event: e.event, data: parse(e.data, {}), at: Number(e.at) })),++ lastSeq: async (jobId) => Number((await q.lastSeq.get({ app, job_id: jobId })).seq) || 0,expired: (now = Date.now()) => q.expired.all({ app, now }),-- deferExpiry: (id, until, now = Date.now()) => q.deferExpiry.run({ id, until, now }),-- remove: (id) => db.tx(async () => { await q.delEvents.run(id); await q.delRefs.run(id); await q.del.run(id); }),++ deferExpiry: (id, until, now = Date.now()) => q.deferExpiry.run({ app, id, until, now }),++ remove: (id) => db.tx(async () => { await q.delEvents.run({ app, id }); await q.delRefs.run({ app, id }); await q.del.run({ app, id }); }),/** → true when the reference is new. */-- addReference: async (id, ref, now = Date.now()) => (await q.addRef.run(id, ref, now)).changes === 1,++ addReference: async (id, ref, now = Date.now()) => (await q.addRef.run({ app, id, ref, now })).changes === 1,/** → true when it existed; dropping the last one restarts the job's expiry clock. */dropReference: (id, ref, now = Date.now()) => db.tx(async () => {-- const gone = (await q.dropRef.run(id, ref)).changes === 1;-- if (gone && (await q.refCount.get(id)).n === 0) await q.releaseExpiry.run({ id, now });++ const gone = (await q.dropRef.run({ app, id, ref })).changes === 1;++ if (gone && Number((await q.refCount.get({ app, id })).n) === 0) await q.releaseExpiry.run({ app, id, now });return gone;}),-- references: (id) => q.refs.all(id),-- referenceCount: async (id) => (await q.refCount.get(id)).n,++ references: (id) => q.refs.all({ app, id }),++ referenceCount: async (id) => Number((await q.refCount.get({ app, id })).n),counts: async () => { const out = Object.fromEntries(STATES.map(s => [s, 0])); for (const r of await q.counts.all({ app })) out[r.state] = Number(r.n); return out; },tx: (fn) => db.tx(fn),parse,nnew file mode 100644@@@ -0,0 +1,62 @@++'use strict';++// The job store in the one shared `tools` database (plan T8): tool_job_events and tool_job_references++// carry no `app`, so every statement on them goes through the owning tool_jobs row. Another app that++// knows a job id reads no events or references, cannot add or drop one, cannot claim, finish or remove++// the job, and its own writes on a foreign id change nothing.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { createStore } = require('../jobs/store');++const { testDb, closeAllTestDbs } = require('./testdb');++++(async () => {++ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-jobs-scope-'));++ try {++ const db = await testDb(dir);++ const img = createStore(db, { app: 'img' });++ const docs = createStore(db, { app: 'docs' });++ const now = Date.now();++ await img.insert({ id: 'job_scope_1', type: 't', type_version: 1, owner: 'session:o', input_json: '{}', files_json: '[]', idempotency_key: null, request_hash: null, max_attempts: 1, ttl_ms: 60_000, now, env: 'production' });++++ // Events.++ const seq = await img.appendEvent('job_scope_1', 'progress', { p: 1 });++ assert.ok(seq > 0);++ assert.strictEqual(await docs.appendEvent('job_scope_1', 'progress', { p: 2 }), 0, 'another app appends nothing to a foreign job');++ assert.strictEqual((await img.eventsAfter('job_scope_1', 0)).length, 1);++ assert.deepStrictEqual(await docs.eventsAfter('job_scope_1', 0), [], "another app reads none of its events");++ assert.strictEqual(await img.lastSeq('job_scope_1'), seq);++ assert.strictEqual(await docs.lastSeq('job_scope_1'), 0);++++ // References.++ assert.strictEqual(await img.addReference('job_scope_1', 'paste:1'), true);++ assert.strictEqual(await docs.addReference('job_scope_1', 'paste:2'), false, 'another app cannot reference it');++ assert.deepStrictEqual((await img.references('job_scope_1')).map((r) => r.ref), ['paste:1']);++ assert.deepStrictEqual(await docs.references('job_scope_1'), []);++ assert.strictEqual(await img.referenceCount('job_scope_1'), 1);++ assert.strictEqual(await docs.referenceCount('job_scope_1'), 0);++ assert.strictEqual(await docs.dropReference('job_scope_1', 'paste:1'), false, 'nor drop one');++ assert.strictEqual(await img.referenceCount('job_scope_1'), 1);++++ // The job row itself.++ assert.strictEqual(await docs.get('job_scope_1'), null);++ assert.strictEqual(await docs.claim('job_scope_1'), false);++ assert.strictEqual(await docs.requestCancel('job_scope_1'), false);++ assert.strictEqual(await img.claim('job_scope_1'), true);++ assert.strictEqual(await docs.setProgress('job_scope_1', 50, 'x'), false);++ assert.strictEqual(await docs.finish('job_scope_1', { state: 'failed' }), false);++ assert.strictEqual(await img.finish('job_scope_1', { state: 'succeeded', result: {} }), true);++ await docs.remove('job_scope_1');++ assert.ok(await img.get('job_scope_1'), 'another app cannot remove it');++ assert.strictEqual((await img.eventsAfter('job_scope_1', 0)).length, 1, 'nor its events');++ assert.strictEqual(await img.dropReference('job_scope_1', 'paste:1'), true);++ await img.remove('job_scope_1');++ assert.strictEqual(await img.get('job_scope_1'), null);++ assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_events WHERE job_id = ?').get('job_scope_1')).n), 0, 'its own remove takes the events');++++ console.log('jobs store scope: events, references and job writes go through the owning app: all checks passed');++ } finally {++ await closeAllTestDbs();++ fs.rmSync(dir, { recursive: true, force: true });++ }++})().catch((err) => { console.error(err); process.exit(1); });
@@ -30,46 +30,52 @@ function createStore(db, { app } = {}) {
@@ -30,46 +30,52 @@ function createStore(db, { app } = {}) {
const q = {
const q = {
insert: db.prepare(`INSERT INTO tool_jobs (id, app, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool, project_id, trace_id)
insert: db.prepare(`INSERT INTO tool_jobs (id, app, type, type_version, owner, state, input_json, files_json, idempotency_key, request_hash, max_attempts, ttl_ms, created_at, updated_at, env, retry_of, ip_key, tool, project_id, trace_id)
VALUES (@id, @app, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool, @project_id, @trace_id)`),
VALUES (@id, @app, @type, @type_version, @owner, 'queued', @input_json, @files_json, @idempotency_key, @request_hash, @max_attempts, @ttl_ms, @now, @now, @env, @retry_of, @ip_key, @tool, @project_id, @trace_id)`),
- markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND state = 'failed' AND retried_by IS NULL"),
+ markRetried: db.prepare("UPDATE tool_jobs SET retried_by = @next, updated_at = @now WHERE id = @id AND app = @app AND state = 'failed' AND retried_by IS NULL"),
get: db.prepare('SELECT * FROM tool_jobs WHERE id = ? AND app = ?'),
get: db.prepare('SELECT * FROM tool_jobs WHERE id = ? AND app = ?'),
byIdem: db.prepare('SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND idempotency_key = @idempotency_key'),
byIdem: db.prepare('SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND idempotency_key = @idempotency_key'),
// A result file stored in Media, by its media_id (a run's { media_id } reference); the owner's only.
// A result file stored in Media, by its media_id (a run's { media_id } reference); the owner's only.
byResultMedia: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND state = 'succeeded' AND result_json ILIKE @like ORDER BY id DESC LIMIT 5"),
byResultMedia: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND owner = @owner AND state = 'succeeded' AND result_json ILIKE @like ORDER BY id DESC LIMIT 5"),
nextQueued: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND state = 'queued' ORDER BY id LIMIT @limit"),
nextQueued: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND state = 'queued' ORDER BY id LIMIT @limit"),
- claim: db.prepare("UPDATE tool_jobs SET state = 'running', attempts = attempts + 1, started_at = @now, updated_at = @now, progress = NULL, progress_message = NULL WHERE id = @id AND state = 'queued'"),
+ claim: db.prepare("UPDATE tool_jobs SET state = 'running', attempts = attempts + 1, started_at = @now, updated_at = @now, progress = NULL, progress_message = NULL WHERE id = @id AND app = @app AND state = 'queued'"),
running: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND state = 'running'"),
running: db.prepare("SELECT * FROM tool_jobs WHERE app = @app AND state = 'running'"),
activeForOwner: db.prepare("SELECT COUNT(*) AS n FROM tool_jobs WHERE app = @app AND owner = @owner AND state IN ('queued','running')"),
activeForOwner: db.prepare("SELECT COUNT(*) AS n FROM tool_jobs WHERE app = @app AND owner = @owner AND state IN ('queued','running')"),
activeForIp: db.prepare("SELECT COUNT(*) AS n FROM tool_jobs WHERE app = @app AND ip_key = @ip_key AND state IN ('queued','running')"),
activeForIp: db.prepare("SELECT COUNT(*) AS n FROM tool_jobs WHERE app = @app AND ip_key = @ip_key AND state IN ('queued','running')"),
- progress: db.prepare('UPDATE tool_jobs SET progress = @progress, progress_message = @message, updated_at = @now WHERE id = @id AND state = \'running\''),
- requestCancel: db.prepare("UPDATE tool_jobs SET cancel_requested = 1, updated_at = @now WHERE id = @id AND state IN ('queued','running')"),
- requeue: db.prepare("UPDATE tool_jobs SET state = 'queued', started_at = NULL, progress = NULL, progress_message = NULL, updated_at = @now WHERE id = @id AND state = 'running'"),
+ progress: db.prepare('UPDATE tool_jobs SET progress = @progress, progress_message = @message, updated_at = @now WHERE id = @id AND app = @app AND state = \'running\''),
+ requestCancel: db.prepare("UPDATE tool_jobs SET cancel_requested = 1, updated_at = @now WHERE id = @id AND app = @app AND state IN ('queued','running')"),
+ requeue: db.prepare("UPDATE tool_jobs SET state = 'queued', started_at = NULL, progress = NULL, progress_message = NULL, updated_at = @now WHERE id = @id AND app = @app AND state = 'running'"),
finish: db.prepare(`UPDATE tool_jobs SET state = @state, result_json = @result_json, error_json = @error_json, retryable = @retryable,
finish: db.prepare(`UPDATE tool_jobs SET state = @state, result_json = @result_json, error_json = @error_json, retryable = @retryable,
progress = CASE WHEN @state = 'succeeded' THEN 100 ELSE progress END, progress_message = NULL,
progress = CASE WHEN @state = 'succeeded' THEN 100 ELSE progress END, progress_message = NULL,
- finished_at = @now, updated_at = @now, expires_at = @now + ttl_ms::bigint WHERE id = @id AND state IN ('queued','running')`),
- setFiles: db.prepare('UPDATE tool_jobs SET files_json = @files_json, updated_at = @now WHERE id = @id'),
- event: db.prepare('INSERT INTO tool_job_events (job_id, event, data, at) VALUES (@job_id, @event, @data, @at) RETURNING 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 = ?'),
+ finished_at = @now, updated_at = @now, expires_at = @now + ttl_ms::bigint WHERE id = @id AND app = @app AND state IN ('queued','running')`),
+ setFiles: db.prepare('UPDATE tool_jobs SET files_json = @files_json, updated_at = @now WHERE id = @id AND app = @app'),
+ // tool_job_events and tool_job_references carry no `app` of their own: every statement on them
+ // goes through this app's tool_jobs row, so an id from another app reads and changes nothing.
+ event: db.prepare(`INSERT INTO tool_job_events (job_id, event, data, at)
+ SELECT CAST(@job_id AS text), CAST(@event AS text), CAST(@data AS text), CAST(@at AS bigint) WHERE EXISTS (SELECT 1 FROM tool_jobs WHERE id = @job_id AND app = @app) RETURNING seq`),
+ eventsAfter: db.prepare(`SELECT e.seq, e.event, e.data, e.at FROM tool_job_events e JOIN tool_jobs j ON j.id = e.job_id
+ WHERE e.job_id = @job_id AND j.app = @app AND e.seq > @seq ORDER BY e.seq`),
+ lastSeq: db.prepare('SELECT MAX(e.seq) AS seq FROM tool_job_events e JOIN tool_jobs j ON j.id = e.job_id WHERE e.job_id = @job_id AND j.app = @app'),
// A job something still references is never expired (see tool_job_references).
// A job something still references is never expired (see tool_job_references).
expired: db.prepare(`SELECT * FROM tool_jobs j WHERE app = @app AND expires_at IS NOT NULL AND expires_at <= @now
expired: db.prepare(`SELECT * FROM tool_jobs j WHERE app = @app AND expires_at IS NOT NULL AND expires_at <= @now
AND NOT EXISTS (SELECT 1 FROM tool_job_references r WHERE r.job_id = j.id) LIMIT 500`),
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 INTO tool_job_references (job_id, ref, created_at) VALUES (?, ?, ?) ON CONFLICT DO NOTHING'),
- 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 = ?'),
+ deferExpiry: db.prepare('UPDATE tool_jobs SET expires_at = @until, updated_at = @now WHERE id = @id AND app = @app'),
+ del: db.prepare('DELETE FROM tool_jobs WHERE id = @id AND app = @app'),
+ delEvents: db.prepare('DELETE FROM tool_job_events WHERE job_id = @id AND EXISTS (SELECT 1 FROM tool_jobs WHERE id = @id AND app = @app)'),
+ delRefs: db.prepare('DELETE FROM tool_job_references WHERE job_id = @id AND EXISTS (SELECT 1 FROM tool_jobs WHERE id = @id AND app = @app)'),
+ addRef: db.prepare(`INSERT INTO tool_job_references (job_id, ref, created_at)
+ SELECT CAST(@id AS text), CAST(@ref AS text), CAST(@now AS bigint) WHERE EXISTS (SELECT 1 FROM tool_jobs WHERE id = @id AND app = @app) ON CONFLICT DO NOTHING`),
+ dropRef: db.prepare('DELETE FROM tool_job_references WHERE job_id = @id AND ref = @ref AND EXISTS (SELECT 1 FROM tool_jobs WHERE id = @id AND app = @app)'),
+ refs: db.prepare(`SELECT r.ref, r.created_at FROM tool_job_references r JOIN tool_jobs j ON j.id = r.job_id
+ WHERE r.job_id = @id AND j.app = @app ORDER BY r.created_at, r.ref`),
+ refCount: db.prepare('SELECT COUNT(*) AS n FROM tool_job_references r JOIN tool_jobs j ON j.id = r.job_id WHERE r.job_id = @id AND j.app = @app'),
// After the last reference goes, the result is kept at least one more ttl.
// After the last reference goes, the result is kept at least one more ttl.
- releaseExpiry: db.prepare('UPDATE tool_jobs SET expires_at = GREATEST(COALESCE(expires_at, 0), @now + ttl_ms::bigint), updated_at = @now WHERE id = @id AND finished_at IS NOT NULL'),
+ releaseExpiry: db.prepare('UPDATE tool_jobs SET expires_at = GREATEST(COALESCE(expires_at, 0), @now + ttl_ms::bigint), updated_at = @now WHERE id = @id AND app = @app AND finished_at IS NOT NULL'),
counts: db.prepare('SELECT state, COUNT(*) AS n FROM tool_jobs WHERE app = @app GROUP BY state'),
counts: db.prepare('SELECT state, COUNT(*) AS n FROM tool_jobs WHERE app = @app GROUP BY state'),
};
};
/** Append one event; returns its seq (the SSE id). */
/** Append one event; returns its seq (the SSE id). */
async function appendEvent(jobId, event, data, now = Date.now()) {
async function appendEvent(jobId, event, data, now = Date.now()) {
- const r = await q.event.run({ job_id: jobId, event, data: JSON.stringify(data), at: now });
- return Number(r.lastInsertRowid);
+ const r = await q.event.run({ app, job_id: jobId, event, data: JSON.stringify(data), at: now });
+ return r.changes === 1 ? Number(r.lastInsertRowid) : 0; // 0: no such job in this app
}
}
return {
return {
@@ -78,39 +84,39 @@ function createStore(db, { app } = {}) {
@@ -78,39 +84,39 @@ function createStore(db, { app } = {}) {
TERMINAL, STATES,
TERMINAL, STATES,
async insert(row) { await q.insert.run({ app, retry_of: null, ip_key: null, tool: null, project_id: null, trace_id: null, ...row }); },
async insert(row) { await q.insert.run({ app, retry_of: null, ip_key: null, tool: null, project_id: null, trace_id: null, ...row }); },
byResultMedia: async (owner, mediaId) => q.byResultMedia.all({ app, owner, like: `%"media_id":"${String(mediaId).replace(/[%"\\]/g, '')}"%` }),
byResultMedia: async (owner, mediaId) => q.byResultMedia.all({ app, owner, like: `%"media_id":"${String(mediaId).replace(/[%"\\]/g, '')}"%` }),
- markRetried: async (id, next, now = Date.now()) => (await q.markRetried.run({ id, next, now })).changes === 1,
+ markRetried: async (id, next, now = Date.now()) => (await q.markRetried.run({ app, id, next, now })).changes === 1,
get: async (id) => (await q.get.get(id, app)) || null,
get: async (id) => (await q.get.get(id, app)) || null,
byIdempotencyKey: async (owner, key) => (await q.byIdem.get({ app, owner, idempotency_key: key })) || null,
byIdempotencyKey: async (owner, key) => (await q.byIdem.get({ app, owner, idempotency_key: key })) || null,
nextQueued: (limit) => q.nextQueued.all({ app, limit }),
nextQueued: (limit) => q.nextQueued.all({ app, limit }),
- claim: async (id, now = Date.now()) => (await q.claim.run({ id, now })).changes === 1,
+ claim: async (id, now = Date.now()) => (await q.claim.run({ app, id, now })).changes === 1,
running: () => q.running.all({ app }),
running: () => q.running.all({ app }),
activeForOwner: async (owner) => (await q.activeForOwner.get({ app, owner })).n,
activeForOwner: async (owner) => (await q.activeForOwner.get({ app, owner })).n,
activeForIp: async (ipKey) => (ipKey ? (await q.activeForIp.get({ app, ip_key: ipKey })).n : 0),
activeForIp: async (ipKey) => (ipKey ? (await q.activeForIp.get({ app, ip_key: ipKey })).n : 0),
- setProgress: async (id, progress, message, now = Date.now()) => (await q.progress.run({ id, progress, message, now })).changes === 1,
- requestCancel: async (id, now = Date.now()) => (await q.requestCancel.run({ id, now })).changes === 1,
- requeue: async (id, now = Date.now()) => (await q.requeue.run({ id, now })).changes === 1,
+ setProgress: async (id, progress, message, now = Date.now()) => (await q.progress.run({ app, id, progress, message, now })).changes === 1,
+ requestCancel: async (id, now = Date.now()) => (await q.requestCancel.run({ app, id, now })).changes === 1,
+ requeue: async (id, now = Date.now()) => (await q.requeue.run({ app, id, now })).changes === 1,
finish: async (id, { state, result = null, error = null, retryable = false }, now = Date.now()) => (await q.finish.run({
finish: async (id, { state, result = null, error = null, retryable = false }, now = Date.now()) => (await q.finish.run({
- id, state, now, retryable: retryable ? 1 : 0,
+ app, id, state, now, retryable: retryable ? 1 : 0,
result_json: result == null ? null : JSON.stringify(result),
result_json: result == null ? null : JSON.stringify(result),
error_json: error == null ? null : JSON.stringify(error),
error_json: error == null ? null : JSON.stringify(error),
})).changes === 1,
})).changes === 1,
- setFiles: (id, files, now = Date.now()) => q.setFiles.run({ id, files_json: JSON.stringify(files), now }),
+ setFiles: (id, files, now = Date.now()) => q.setFiles.run({ app, id, files_json: JSON.stringify(files), now }),
appendEvent,
appendEvent,
- eventsAfter: async (jobId, seq) => (await q.eventsAfter.all(jobId, seq)).map(e => ({ seq: Number(e.seq), event: e.event, data: parse(e.data, {}), at: Number(e.at) })),
- lastSeq: async (jobId) => Number((await q.lastSeq.get(jobId)).seq) || 0,
+ eventsAfter: async (jobId, seq) => (await q.eventsAfter.all({ app, job_id: jobId, seq })).map(e => ({ seq: Number(e.seq), event: e.event, data: parse(e.data, {}), at: Number(e.at) })),
+ lastSeq: async (jobId) => Number((await q.lastSeq.get({ app, job_id: jobId })).seq) || 0,
expired: (now = Date.now()) => q.expired.all({ app, now }),
expired: (now = Date.now()) => q.expired.all({ app, now }),
- deferExpiry: (id, until, now = Date.now()) => q.deferExpiry.run({ id, until, now }),
- remove: (id) => db.tx(async () => { await q.delEvents.run(id); await q.delRefs.run(id); await q.del.run(id); }),
+ deferExpiry: (id, until, now = Date.now()) => q.deferExpiry.run({ app, id, until, now }),
+ remove: (id) => db.tx(async () => { await q.delEvents.run({ app, id }); await q.delRefs.run({ app, id }); await q.del.run({ app, id }); }),
/** → true when the reference is new. */
/** → true when the reference is new. */
- addReference: async (id, ref, now = Date.now()) => (await q.addRef.run(id, ref, now)).changes === 1,
+ addReference: async (id, ref, now = Date.now()) => (await q.addRef.run({ app, id, ref, now })).changes === 1,
/** → true when it existed; dropping the last one restarts the job's expiry clock. */
/** → true when it existed; dropping the last one restarts the job's expiry clock. */
dropReference: (id, ref, now = Date.now()) => db.tx(async () => {
dropReference: (id, ref, now = Date.now()) => db.tx(async () => {
- const gone = (await q.dropRef.run(id, ref)).changes === 1;
- if (gone && (await q.refCount.get(id)).n === 0) await q.releaseExpiry.run({ id, now });
+ const gone = (await q.dropRef.run({ app, id, ref })).changes === 1;
+ if (gone && Number((await q.refCount.get({ app, id })).n) === 0) await q.releaseExpiry.run({ app, id, now });
return gone;
return gone;
}),
}),
- references: (id) => q.refs.all(id),
- referenceCount: async (id) => (await q.refCount.get(id)).n,
+ references: (id) => q.refs.all({ app, id }),
+ referenceCount: async (id) => Number((await q.refCount.get({ app, id })).n),
counts: async () => { const out = Object.fromEntries(STATES.map(s => [s, 0])); for (const r of await q.counts.all({ app })) out[r.state] = Number(r.n); return out; },
counts: async () => { const out = Object.fromEntries(STATES.map(s => [s, 0])); for (const r of await q.counts.all({ app })) out[r.state] = Number(r.n); return out; },
tx: (fn) => db.tx(fn),
tx: (fn) => db.tx(fn),
parse,
parse,
new file mode 100644
new file mode 100644
@@ -0,0 +1,62 @@
@@ -0,0 +1,62 @@
+'use strict';
+// The job store in the one shared `tools` database (plan T8): tool_job_events and tool_job_references
+// carry no `app`, so every statement on them goes through the owning tool_jobs row. Another app that
+// knows a job id reads no events or references, cannot add or drop one, cannot claim, finish or remove
+// the job, and its own writes on a foreign id change nothing.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { createStore } = require('../jobs/store');
+const { testDb, closeAllTestDbs } = require('./testdb');
+
+(async () => {
+ const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-jobs-scope-'));
+ try {
+ const db = await testDb(dir);
+ const img = createStore(db, { app: 'img' });
+ const docs = createStore(db, { app: 'docs' });
+ const now = Date.now();
+ await img.insert({ id: 'job_scope_1', type: 't', type_version: 1, owner: 'session:o', input_json: '{}', files_json: '[]', idempotency_key: null, request_hash: null, max_attempts: 1, ttl_ms: 60_000, now, env: 'production' });
+
+ // Events.
+ const seq = await img.appendEvent('job_scope_1', 'progress', { p: 1 });
+ assert.ok(seq > 0);
+ assert.strictEqual(await docs.appendEvent('job_scope_1', 'progress', { p: 2 }), 0, 'another app appends nothing to a foreign job');
+ assert.strictEqual((await img.eventsAfter('job_scope_1', 0)).length, 1);
+ assert.deepStrictEqual(await docs.eventsAfter('job_scope_1', 0), [], "another app reads none of its events");
+ assert.strictEqual(await img.lastSeq('job_scope_1'), seq);
+ assert.strictEqual(await docs.lastSeq('job_scope_1'), 0);
+
+ // References.
+ assert.strictEqual(await img.addReference('job_scope_1', 'paste:1'), true);
+ assert.strictEqual(await docs.addReference('job_scope_1', 'paste:2'), false, 'another app cannot reference it');
+ assert.deepStrictEqual((await img.references('job_scope_1')).map((r) => r.ref), ['paste:1']);
+ assert.deepStrictEqual(await docs.references('job_scope_1'), []);
+ assert.strictEqual(await img.referenceCount('job_scope_1'), 1);
+ assert.strictEqual(await docs.referenceCount('job_scope_1'), 0);
+ assert.strictEqual(await docs.dropReference('job_scope_1', 'paste:1'), false, 'nor drop one');
+ assert.strictEqual(await img.referenceCount('job_scope_1'), 1);
+
+ // The job row itself.
+ assert.strictEqual(await docs.get('job_scope_1'), null);
+ assert.strictEqual(await docs.claim('job_scope_1'), false);
+ assert.strictEqual(await docs.requestCancel('job_scope_1'), false);
+ assert.strictEqual(await img.claim('job_scope_1'), true);
+ assert.strictEqual(await docs.setProgress('job_scope_1', 50, 'x'), false);
+ assert.strictEqual(await docs.finish('job_scope_1', { state: 'failed' }), false);
+ assert.strictEqual(await img.finish('job_scope_1', { state: 'succeeded', result: {} }), true);
+ await docs.remove('job_scope_1');
+ assert.ok(await img.get('job_scope_1'), 'another app cannot remove it');
+ assert.strictEqual((await img.eventsAfter('job_scope_1', 0)).length, 1, 'nor its events');
+ assert.strictEqual(await img.dropReference('job_scope_1', 'paste:1'), true);
+ await img.remove('job_scope_1');
+ assert.strictEqual(await img.get('job_scope_1'), null);
+ assert.strictEqual(Number((await db.prepare('SELECT COUNT(*) AS n FROM tool_job_events WHERE job_id = ?').get('job_scope_1')).n), 0, 'its own remove takes the events');
+
+ console.log('jobs store scope: events, references and job writes go through the owning app: all checks passed');
+ } finally {
+ await closeAllTestDbs();
+ fs.rmSync(dir, { recursive: true, force: true });
+ }
+})().catch((err) => { console.error(err); process.exit(1); });