openvibers/openvibe.tools Commits
Tools jobs: a failed boot recovery no longer strands queued jobs (plan T8 review): start() rejects and leaves the system unstarted (started is set only after recovery, so the worker never claims during it) and can be called again; setupJobs retries it with a backoff (1 s doubling to 60 s) until it runs or the app closes. jobs-start-retry.test.js.
Changes
@@@ -86,6 +86,7 @@ function mediaOrigin(env = process.env) {* @param {object} [o.guard] apps/_shared/guard: its caller resolver, admission (session rule,* sniffing, quota) and busy handling; without it the job routes* resolve owners themselves and a busy store is always refused++ * @param {number} [o.bootRetryMs] first retry delay after a failed boot recovery (default 1000, doubling to 60 s)* @param {(req, type, input) => string|null} [o.jobTool] the descriptor id a submit runs (the host's* tool, or the one its input.tool names)*/@@@ -122,7 +123,17 @@ function setupJobs(o) {onAddressFull: (req, res, full) => !g.refuse(req, res, { status: 429, code: 'tools.job.too_many_active', reason: 'jobs.address', tool: o.jobTool ? o.jobTool(req, null, {}) : null, retryAfter: 30, detail: `At most ${full.limit} unfinished jobs from one address at a time; wait for one to finish.`, extra: { scope: 'address' } }),}),});-- system.start().catch((err) => log.error(`[Jobs] ${o.service}: boot recovery failed:`, err.message));++ // A failed boot recovery (the database not reachable yet) leaves the system unstarted; try again with++ // a backoff (1 s doubling to 60 s) until it starts or the app closes, so queued jobs always resume.++ let bootTimer = null, closed = false;++ const boot = (attempt = 0) => system.start().catch((err) => {++ if (closed) return;++ const wait = Math.min(60_000, (o.bootRetryMs || 1000) * 2 ** attempt);++ log.error(`[Jobs] ${o.service}: boot recovery failed (retrying in ${Math.round(wait / 1000)} s):`, err.message);++ bootTimer = setTimeout(() => boot(attempt + 1), wait);++ if (bootTimer.unref) bootTimer.unref();++ });++ boot();let pruneOutbox = null;if (events.outbox) {events.outbox.start();@@@ -133,6 +144,8 @@ function setupJobs(o) {system.db = db;// The `tools` database is shared with the rest of the app and closed there; this only stops the worker.system.close = () => {++ closed = true;++ clearTimeout(bootTimer);system.stop();if (events.outbox) { events.outbox.stop().catch(() => {}); clearInterval(pruneOutbox); }};@@@ -679,11 +679,20 @@ function createJobSystem(o) {return active >= maxActivePerAddress ? { active, limit: maxActivePerAddress } : null;}-- async function start() {-- if (started) return api;-- started = true;++ /**++ * Recover, prune, start the timer and the worker. A failed recovery (the database not reachable yet)++ * rejects and leaves the system unstarted, so start() can simply be called again (index.js retries++ * it with a backoff): queued jobs are never stranded behind a `started` flag nothing will reset.++ */++ let starting = null;++ function start() {++ if (!starting) starting = boot().catch((err) => { starting = null; throw err; });++ return starting;++ }++ async function boot() {if (diskBudget) refreshDisk();const r = await recover();++ started = true; // only now: the worker never claims a job while recovery is still reading 'running' rowsif (r.requeued || r.failed) log.log(`[Jobs] ${o.service}: ${r.requeued} job(s) re-queued, ${r.failed} failed after restart`);await prune().catch(() => {});await flushUsage();nnew file mode 100644@@@ -0,0 +1,81 @@++'use strict';++// A boot recovery that fails (the database not reachable yet) must not strand queued jobs: start()++// rejects and leaves the system unstarted, a second start() recovers and runs them, and setupJobs++// (apps/_shared/jobs/index.js) retries on its own with a backoff until the system is running.++const assert = require('assert');++const fs = require('fs');++const os = require('os');++const path = require('path');++const { dep } = require('./deps');++const jobs = require('../jobs');++const { testDb, closeAllTestDbs } = require('./testdb');++++const express = dep('express');++const contracts = dep('openvibe-contracts');++const quiet = { log() {}, warn() {}, error() {} };++++/** The db handle with the recovery's "running jobs" query failing the first `n` times. */++function flaky(db, n) {++ const out = Object.create(db);++ out.fails = n;++ out.prepare = (sql) => {++ const s = db.prepare(sql);++ if (!/^SELECT \* FROM tool_jobs WHERE app = @app AND state = 'running'$/.test(sql.trim())) return s;++ return {++ get: (...a) => s.get(...a),++ run: (...a) => s.run(...a),++ all: async (...a) => { if (out.fails > 0) { out.fails--; throw new Error('database not reachable yet'); } return s.all(...a); },++ };++ };++ return out;++}++++async function until(fn, what, ms = 5000) {++ const end = Date.now() + ms;++ while (Date.now() < end) { if (await fn()) return; await new Promise((r) => setTimeout(r, 20)); }++ throw new Error(`timed out: ${what}`);++}++++(async () => {++ const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-jobs-retry-'));++ try {++ // 1. The job system: a failed start() leaves it unstarted; the next one recovers.++ {++ const dir = path.join(root, 'system');++ const db = flaky(await testDb(dir), 1);++ const system = jobs.createJobSystem({ db, contracts, service: 'retry-a', dataDir: dir, log: quiet });++ system.define({ type: 'test.noop', maxFiles: 1, async run() { return { files: [], data: { ok: true } }; } });++ const { job } = await system.submit({ owner: 'session:x', type: 'test.noop', input: {} });++ assert.strictEqual(job.state, 'queued');++ await assert.rejects(system.start(), /not reachable yet/);++ assert.strictEqual(system.isRunning(), false, 'a failed recovery leaves the system unstarted');++ assert.strictEqual((await system.get(job.id)).state, 'queued', 'the queued job waits');++ await system.start();++ assert.strictEqual(system.isRunning(), true);++ await until(async () => (await system.get(job.id)).state === 'succeeded', 'the queued job runs after the second start');++ system.stop();++ }++++ // 2. setupJobs retries by itself.++ {++ const dir = path.join(root, 'setup');++ const db = flaky(await testDb(dir), 2);++ const app = express();++ const system = jobs.setupJobs({++ app, service: 'retry-b', dataDir: dir, db, contracts, getPublicKey: () => null, issuer: 'https://openvibe.network',++ define: (s) => s.define({ type: 'test.noop', maxFiles: 1, async run() { return { files: [], data: {} }; } }),++ bootRetryMs: 20, log: quiet,++ });++ await until(() => system.isRunning(), 'setupJobs retried the boot until it started');++ assert.strictEqual(db.fails, 0, 'after both failures');++ const { job } = await system.submit({ owner: 'session:y', type: 'test.noop', input: {} });++ await until(async () => (await system.get(job.id)).state === 'succeeded', 'a job runs');++ system.close();++ }++++ console.log('jobs start retry: a failed boot recovery leaves the system unstarted, start() again resumes queued jobs, setupJobs retries with a backoff: all checks passed');++ } finally {++ await closeAllTestDbs();++ fs.rmSync(root, { recursive: true, force: true });++ }++})().catch((err) => { console.error(err); process.exit(1); });
@@ -86,6 +86,7 @@ function mediaOrigin(env = process.env) {
@@ -86,6 +86,7 @@ function mediaOrigin(env = process.env) {
* @param {object} [o.guard] apps/_shared/guard: its caller resolver, admission (session rule,
* @param {object} [o.guard] apps/_shared/guard: its caller resolver, admission (session rule,
* sniffing, quota) and busy handling; without it the job routes
* sniffing, quota) and busy handling; without it the job routes
* resolve owners themselves and a busy store is always refused
* resolve owners themselves and a busy store is always refused
+ * @param {number} [o.bootRetryMs] first retry delay after a failed boot recovery (default 1000, doubling to 60 s)
* @param {(req, type, input) => string|null} [o.jobTool] the descriptor id a submit runs (the host's
* @param {(req, type, input) => string|null} [o.jobTool] the descriptor id a submit runs (the host's
* tool, or the one its input.tool names)
* tool, or the one its input.tool names)
*/
*/
@@ -122,7 +123,17 @@ function setupJobs(o) {
@@ -122,7 +123,17 @@ function setupJobs(o) {
onAddressFull: (req, res, full) => !g.refuse(req, res, { status: 429, code: 'tools.job.too_many_active', reason: 'jobs.address', tool: o.jobTool ? o.jobTool(req, null, {}) : null, retryAfter: 30, detail: `At most ${full.limit} unfinished jobs from one address at a time; wait for one to finish.`, extra: { scope: 'address' } }),
onAddressFull: (req, res, full) => !g.refuse(req, res, { status: 429, code: 'tools.job.too_many_active', reason: 'jobs.address', tool: o.jobTool ? o.jobTool(req, null, {}) : null, retryAfter: 30, detail: `At most ${full.limit} unfinished jobs from one address at a time; wait for one to finish.`, extra: { scope: 'address' } }),
}),
}),
});
});
- system.start().catch((err) => log.error(`[Jobs] ${o.service}: boot recovery failed:`, err.message));
+ // A failed boot recovery (the database not reachable yet) leaves the system unstarted; try again with
+ // a backoff (1 s doubling to 60 s) until it starts or the app closes, so queued jobs always resume.
+ let bootTimer = null, closed = false;
+ const boot = (attempt = 0) => system.start().catch((err) => {
+ if (closed) return;
+ const wait = Math.min(60_000, (o.bootRetryMs || 1000) * 2 ** attempt);
+ log.error(`[Jobs] ${o.service}: boot recovery failed (retrying in ${Math.round(wait / 1000)} s):`, err.message);
+ bootTimer = setTimeout(() => boot(attempt + 1), wait);
+ if (bootTimer.unref) bootTimer.unref();
+ });
+ boot();
let pruneOutbox = null;
let pruneOutbox = null;
if (events.outbox) {
if (events.outbox) {
events.outbox.start();
events.outbox.start();
@@ -133,6 +144,8 @@ function setupJobs(o) {
@@ -133,6 +144,8 @@ function setupJobs(o) {
system.db = db;
system.db = db;
// The `tools` database is shared with the rest of the app and closed there; this only stops the worker.
// The `tools` database is shared with the rest of the app and closed there; this only stops the worker.
system.close = () => {
system.close = () => {
+ closed = true;
+ clearTimeout(bootTimer);
system.stop();
system.stop();
if (events.outbox) { events.outbox.stop().catch(() => {}); clearInterval(pruneOutbox); }
if (events.outbox) { events.outbox.stop().catch(() => {}); clearInterval(pruneOutbox); }
};
};
@@ -679,11 +679,20 @@ function createJobSystem(o) {
@@ -679,11 +679,20 @@ function createJobSystem(o) {
return active >= maxActivePerAddress ? { active, limit: maxActivePerAddress } : null;
return active >= maxActivePerAddress ? { active, limit: maxActivePerAddress } : null;
}
}
- async function start() {
- if (started) return api;
- started = true;
+ /**
+ * Recover, prune, start the timer and the worker. A failed recovery (the database not reachable yet)
+ * rejects and leaves the system unstarted, so start() can simply be called again (index.js retries
+ * it with a backoff): queued jobs are never stranded behind a `started` flag nothing will reset.
+ */
+ let starting = null;
+ function start() {
+ if (!starting) starting = boot().catch((err) => { starting = null; throw err; });
+ return starting;
+ }
+ async function boot() {
if (diskBudget) refreshDisk();
if (diskBudget) refreshDisk();
const r = await recover();
const r = await recover();
+ started = true; // only now: the worker never claims a job while recovery is still reading 'running' rows
if (r.requeued || r.failed) log.log(`[Jobs] ${o.service}: ${r.requeued} job(s) re-queued, ${r.failed} failed after restart`);
if (r.requeued || r.failed) log.log(`[Jobs] ${o.service}: ${r.requeued} job(s) re-queued, ${r.failed} failed after restart`);
await prune().catch(() => {});
await prune().catch(() => {});
await flushUsage();
await flushUsage();
new file mode 100644
new file mode 100644
@@ -0,0 +1,81 @@
@@ -0,0 +1,81 @@
+'use strict';
+// A boot recovery that fails (the database not reachable yet) must not strand queued jobs: start()
+// rejects and leaves the system unstarted, a second start() recovers and runs them, and setupJobs
+// (apps/_shared/jobs/index.js) retries on its own with a backoff until the system is running.
+const assert = require('assert');
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+const { dep } = require('./deps');
+const jobs = require('../jobs');
+const { testDb, closeAllTestDbs } = require('./testdb');
+
+const express = dep('express');
+const contracts = dep('openvibe-contracts');
+const quiet = { log() {}, warn() {}, error() {} };
+
+/** The db handle with the recovery's "running jobs" query failing the first `n` times. */
+function flaky(db, n) {
+ const out = Object.create(db);
+ out.fails = n;
+ out.prepare = (sql) => {
+ const s = db.prepare(sql);
+ if (!/^SELECT \* FROM tool_jobs WHERE app = @app AND state = 'running'$/.test(sql.trim())) return s;
+ return {
+ get: (...a) => s.get(...a),
+ run: (...a) => s.run(...a),
+ all: async (...a) => { if (out.fails > 0) { out.fails--; throw new Error('database not reachable yet'); } return s.all(...a); },
+ };
+ };
+ return out;
+}
+
+async function until(fn, what, ms = 5000) {
+ const end = Date.now() + ms;
+ while (Date.now() < end) { if (await fn()) return; await new Promise((r) => setTimeout(r, 20)); }
+ throw new Error(`timed out: ${what}`);
+}
+
+(async () => {
+ const root = fs.mkdtempSync(path.join(os.tmpdir(), 'ov-jobs-retry-'));
+ try {
+ // 1. The job system: a failed start() leaves it unstarted; the next one recovers.
+ {
+ const dir = path.join(root, 'system');
+ const db = flaky(await testDb(dir), 1);
+ const system = jobs.createJobSystem({ db, contracts, service: 'retry-a', dataDir: dir, log: quiet });
+ system.define({ type: 'test.noop', maxFiles: 1, async run() { return { files: [], data: { ok: true } }; } });
+ const { job } = await system.submit({ owner: 'session:x', type: 'test.noop', input: {} });
+ assert.strictEqual(job.state, 'queued');
+ await assert.rejects(system.start(), /not reachable yet/);
+ assert.strictEqual(system.isRunning(), false, 'a failed recovery leaves the system unstarted');
+ assert.strictEqual((await system.get(job.id)).state, 'queued', 'the queued job waits');
+ await system.start();
+ assert.strictEqual(system.isRunning(), true);
+ await until(async () => (await system.get(job.id)).state === 'succeeded', 'the queued job runs after the second start');
+ system.stop();
+ }
+
+ // 2. setupJobs retries by itself.
+ {
+ const dir = path.join(root, 'setup');
+ const db = flaky(await testDb(dir), 2);
+ const app = express();
+ const system = jobs.setupJobs({
+ app, service: 'retry-b', dataDir: dir, db, contracts, getPublicKey: () => null, issuer: 'https://openvibe.network',
+ define: (s) => s.define({ type: 'test.noop', maxFiles: 1, async run() { return { files: [], data: {} }; } }),
+ bootRetryMs: 20, log: quiet,
+ });
+ await until(() => system.isRunning(), 'setupJobs retried the boot until it started');
+ assert.strictEqual(db.fails, 0, 'after both failures');
+ const { job } = await system.submit({ owner: 'session:y', type: 'test.noop', input: {} });
+ await until(async () => (await system.get(job.id)).state === 'succeeded', 'a job runs');
+ system.close();
+ }
+
+ console.log('jobs start retry: a failed boot recovery leaves the system unstarted, start() again resumes queued jobs, setupJobs retries with a backoff: all checks passed');
+ } finally {
+ await closeAllTestDbs();
+ fs.rmSync(root, { recursive: true, force: true });
+ }
+})().catch((err) => { console.error(err); process.exit(1); });