openvibers/openvibe.sdk Commits
No floating promises left in the SDK (plan T0, the shared checker reports 0): the Valkey queue's fire-and-forget runs log a rejection (a malformed stream entry used to be an unhandled rejection; test/valkey.test.js drives it), the mock jobs service fails a job whose background run throws, and the synchronous calls the checker cannot tell apart (better-sqlite3 exec, the SQLite markFailure, Web Streams enqueue) are marked '// floating-ok: <why>'. 45/45 test files (verified).
Changes
@@@ -28,6 +28,7 @@ function createRevocationStore(db = null, { table = 'ov_token_revocations', nowlet q = null;function ready() {if (!db || q) return q;++ // floating-ok: better-sqlite3 db.exec is synchronousdb.exec(`CREATE TABLE IF NOT EXISTS ${table} (subject_id TEXT PRIMARY KEY,valid_after_ms INTEGER NOT NULL,@@@ -63,6 +63,7 @@ function createOutbox(db, {let running = false;function ensureSchema() {++ // floating-ok: better-sqlite3 db.exec is synchronousdb.exec(`CREATE TABLE IF NOT EXISTS ${table} (id INTEGER PRIMARY KEY AUTOINCREMENT,event_id TEXT NOT NULL UNIQUE,@@@ -135,6 +136,7 @@ function createOutbox(db, {stats.sent += s.sent; stats.failed += s.failed; stats.rejected += s.rejected;}} else {++ // floating-ok: markFailure here is the better-sqlite3 one above (synchronous .run)for (const row of rows) markFailure(row, err);if (isPermanent(err)) stats.rejected += rows.length; else stats.failed += rows.length;}@@@ -189,6 +191,7 @@ function createInbox(db, { table = 'idempotency_receipts', now = () => Date.now(let claim = null;function ensureSchema() {++ // floating-ok: better-sqlite3 db.exec is synchronousdb.exec(`CREATE TABLE IF NOT EXISTS ${table} (consumer TEXT NOT NULL,event_id TEXT NOT NULL,@@@ -101,7 +101,9 @@ function valkeyQueue(valkey, name, group, maxLen, log) {continue;}if (!res) continue;-- for (const [, entries] of res) for (const [id, raw] of entries) run(id, raw);++ // Fire-and-forget on purpose (concurrency is bounded by `active`); a malformed entry++ // (raw null) rejects in fieldsOf(), so log it instead of leaving an unhandled rejection.++ for (const [, entries] of res) for (const [id, raw] of entries) run(id, raw).catch((err) => log.warn(`[queue ${name}] job ${id} failed outside the handler: ${err.message}`));}}@@@ -110,7 +112,7 @@ function valkeyQueue(valkey, name, group, maxLen, log) {if (stopped || active >= concurrency) return;try {const [, entries] = await c.xautoclaim(key, group, consumer, visibilityMs, '0-0', 'COUNT', Math.max(1, concurrency - active));-- for (const [id, raw] of entries || []) if (raw) run(id, raw);++ for (const [id, raw] of entries || []) if (raw) run(id, raw).catch((err) => log.warn(`[queue ${name}] reclaimed job ${id} failed outside the handler: ${err.message}`));} catch { /* the next tick tries again */ }}, Math.max(1000, Math.floor(visibilityMs / 2)));promote.unref(); reclaim.unref();@@@ -565,6 +565,7 @@ function createMockPlatform(opts = {}) {let conn;const body = new ReadableStream({start(controller) {++ // floating-ok: controller.enqueue is synchronousconst send = (s) => { try { controller.enqueue(enc.encode(s)); } catch { streams.delete(conn); } };conn = {lastSeq: last ?? lastSeq,@@@ -148,7 +148,12 @@ function createJobsService(ctx) {jobs.set(job.id, job);if (key) idem.set(`${origin}|${owner}|${key}`, { id: job.id, hash });emit(job, 'job.queued');-- runJob(job);++ // The run goes to the background on purpose (the POST answers 202 while it works); a throw that++ // escapes runJob must fail the job, never become an unhandled rejection in the host test.++ runJob(job).catch((err) => {++ try { finish(job, 'failed', { error: problemBody(500, 'tools.job.failed', String((err && err.message) || err)) }); }++ catch { Object.assign(job, { state: 'failed', finishedAt: Date.now(), error: problemBody(500, 'tools.job.failed', String((err && err.message) || err)) }); }++ });return { job, replayed: false };}@@@ -176,11 +181,13 @@ function createJobsService(ctx) {push(e) {if (e.seq <= conn.last) return;conn.last = e.seq;++ // floating-ok: controller.enqueue is synchronoustry { controller.enqueue(enc.encode(`id: ${e.seq}\nevent: ${e.event}\ndata: ${JSON.stringify(e.data)}\n\n`)); } catch { streams.delete(conn); return; }if (TERMINAL.has(e.data.state) && e.event !== 'job.progress') setImmediate(close);},close,};++ // floating-ok: controller.enqueue is synchronouscontroller.enqueue(enc.encode(`retry: ${stepMs}\n\n`));streams.add(conn);for (const e of job.events) conn.push(e);@@@ -53,6 +53,36 @@ const tests = [assert.equal((await q.dead())[0].error, 'always');await w.stop();}],++ ['valkey queue: a malformed stream entry is logged, never an unhandled rejection', async () => {++ const warns = []; const unhandled = [];++ const onUnhandled = (err) => unhandled.push(err);++ process.on('unhandledRejection', onUnhandled);++ let rejectRead = null;++ const multi = () => { const m = {}; for (const fn of ['xack', 'xdel', 'xadd', 'zadd']) m[fn] = () => m; m.exec = async () => []; return m; };++ let reads = 0;++ const client = {++ defineCommand() {}, ovQueuePromote: async () => 0, xgroup: async () => 'OK',++ xautoclaim: async () => { throw new Error('nothing pending'); },++ multi, set: async () => 'OK', zadd: async () => 1, xlen: async () => 0, zcard: async () => 0, xpending: async () => [0],++ };++ const reader = {++ async xreadgroup() {++ if (++reads === 1) return [['q:badentry', [['1-1', null]]]]; // ioredis gives null fields for an undecodable entry++ await new Promise((resolve, reject) => { rejectRead = reject; });++ },++ disconnect() { if (rejectRead) rejectRead(new Error('disconnect')); },++ };++ const v = { client, key: (...p) => p.join(':'), duplicate: () => reader };++ const q = createQueue({ valkey: v, name: 'badentry', log: { log() {}, warn: (m) => warns.push(m), error() {} } });++ try {++ const w = q.process(async () => { throw new Error('must not run'); }, { concurrency: 2, blockMs: 200 });++ assert.ok(await until(async () => warns.length > 0), 'the malformed entry was logged');++ assert.match(warns[0], /badentry.*1-1.*failed outside the handler/);++ await sleep(100);++ assert.deepEqual(unhandled, [], 'no unhandled rejection escaped the worker');++ await w.stop();++ } finally { process.off('unhandledRejection', onUnhandled); }++ }],['memory pubsub', async () => {const ps = createPubSub();const got = [];
@@ -28,6 +28,7 @@ function createRevocationStore(db = null, { table = 'ov_token_revocations', now
@@ -28,6 +28,7 @@ function createRevocationStore(db = null, { table = 'ov_token_revocations', now
let q = null;
let q = null;
function ready() {
function ready() {
if (!db || q) return q;
if (!db || q) return q;
+ // floating-ok: better-sqlite3 db.exec is synchronous
db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
subject_id TEXT PRIMARY KEY,
subject_id TEXT PRIMARY KEY,
valid_after_ms INTEGER NOT NULL,
valid_after_ms INTEGER NOT NULL,
@@ -63,6 +63,7 @@ function createOutbox(db, {
@@ -63,6 +63,7 @@ function createOutbox(db, {
let running = false;
let running = false;
function ensureSchema() {
function ensureSchema() {
+ // floating-ok: better-sqlite3 db.exec is synchronous
db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
id INTEGER PRIMARY KEY AUTOINCREMENT,
id INTEGER PRIMARY KEY AUTOINCREMENT,
event_id TEXT NOT NULL UNIQUE,
event_id TEXT NOT NULL UNIQUE,
@@ -135,6 +136,7 @@ function createOutbox(db, {
@@ -135,6 +136,7 @@ function createOutbox(db, {
stats.sent += s.sent; stats.failed += s.failed; stats.rejected += s.rejected;
stats.sent += s.sent; stats.failed += s.failed; stats.rejected += s.rejected;
}
}
} else {
} else {
+ // floating-ok: markFailure here is the better-sqlite3 one above (synchronous .run)
for (const row of rows) markFailure(row, err);
for (const row of rows) markFailure(row, err);
if (isPermanent(err)) stats.rejected += rows.length; else stats.failed += rows.length;
if (isPermanent(err)) stats.rejected += rows.length; else stats.failed += rows.length;
}
}
@@ -189,6 +191,7 @@ function createInbox(db, { table = 'idempotency_receipts', now = () => Date.now(
@@ -189,6 +191,7 @@ function createInbox(db, { table = 'idempotency_receipts', now = () => Date.now(
let claim = null;
let claim = null;
function ensureSchema() {
function ensureSchema() {
+ // floating-ok: better-sqlite3 db.exec is synchronous
db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
db.exec(`CREATE TABLE IF NOT EXISTS ${table} (
consumer TEXT NOT NULL,
consumer TEXT NOT NULL,
event_id TEXT NOT NULL,
event_id TEXT NOT NULL,
@@ -101,7 +101,9 @@ function valkeyQueue(valkey, name, group, maxLen, log) {
@@ -101,7 +101,9 @@ function valkeyQueue(valkey, name, group, maxLen, log) {
continue;
continue;
}
}
if (!res) continue;
if (!res) continue;
- for (const [, entries] of res) for (const [id, raw] of entries) run(id, raw);
+ // Fire-and-forget on purpose (concurrency is bounded by `active`); a malformed entry
+ // (raw null) rejects in fieldsOf(), so log it instead of leaving an unhandled rejection.
+ for (const [, entries] of res) for (const [id, raw] of entries) run(id, raw).catch((err) => log.warn(`[queue ${name}] job ${id} failed outside the handler: ${err.message}`));
}
}
}
}
@@ -110,7 +112,7 @@ function valkeyQueue(valkey, name, group, maxLen, log) {
@@ -110,7 +112,7 @@ function valkeyQueue(valkey, name, group, maxLen, log) {
if (stopped || active >= concurrency) return;
if (stopped || active >= concurrency) return;
try {
try {
const [, entries] = await c.xautoclaim(key, group, consumer, visibilityMs, '0-0', 'COUNT', Math.max(1, concurrency - active));
const [, entries] = await c.xautoclaim(key, group, consumer, visibilityMs, '0-0', 'COUNT', Math.max(1, concurrency - active));
- for (const [id, raw] of entries || []) if (raw) run(id, raw);
+ for (const [id, raw] of entries || []) if (raw) run(id, raw).catch((err) => log.warn(`[queue ${name}] reclaimed job ${id} failed outside the handler: ${err.message}`));
} catch { /* the next tick tries again */ }
} catch { /* the next tick tries again */ }
}, Math.max(1000, Math.floor(visibilityMs / 2)));
}, Math.max(1000, Math.floor(visibilityMs / 2)));
promote.unref(); reclaim.unref();
promote.unref(); reclaim.unref();
@@ -565,6 +565,7 @@ function createMockPlatform(opts = {}) {
@@ -565,6 +565,7 @@ function createMockPlatform(opts = {}) {
let conn;
let conn;
const body = new ReadableStream({
const body = new ReadableStream({
start(controller) {
start(controller) {
+ // floating-ok: controller.enqueue is synchronous
const send = (s) => { try { controller.enqueue(enc.encode(s)); } catch { streams.delete(conn); } };
const send = (s) => { try { controller.enqueue(enc.encode(s)); } catch { streams.delete(conn); } };
conn = {
conn = {
lastSeq: last ?? lastSeq,
lastSeq: last ?? lastSeq,
@@ -148,7 +148,12 @@ function createJobsService(ctx) {
@@ -148,7 +148,12 @@ function createJobsService(ctx) {
jobs.set(job.id, job);
jobs.set(job.id, job);
if (key) idem.set(`${origin}|${owner}|${key}`, { id: job.id, hash });
if (key) idem.set(`${origin}|${owner}|${key}`, { id: job.id, hash });
emit(job, 'job.queued');
emit(job, 'job.queued');
- runJob(job);
+ // The run goes to the background on purpose (the POST answers 202 while it works); a throw that
+ // escapes runJob must fail the job, never become an unhandled rejection in the host test.
+ runJob(job).catch((err) => {
+ try { finish(job, 'failed', { error: problemBody(500, 'tools.job.failed', String((err && err.message) || err)) }); }
+ catch { Object.assign(job, { state: 'failed', finishedAt: Date.now(), error: problemBody(500, 'tools.job.failed', String((err && err.message) || err)) }); }
+ });
return { job, replayed: false };
return { job, replayed: false };
}
}
@@ -176,11 +181,13 @@ function createJobsService(ctx) {
@@ -176,11 +181,13 @@ function createJobsService(ctx) {
push(e) {
push(e) {
if (e.seq <= conn.last) return;
if (e.seq <= conn.last) return;
conn.last = e.seq;
conn.last = e.seq;
+ // floating-ok: controller.enqueue is synchronous
try { controller.enqueue(enc.encode(`id: ${e.seq}\nevent: ${e.event}\ndata: ${JSON.stringify(e.data)}\n\n`)); } catch { streams.delete(conn); return; }
try { controller.enqueue(enc.encode(`id: ${e.seq}\nevent: ${e.event}\ndata: ${JSON.stringify(e.data)}\n\n`)); } catch { streams.delete(conn); return; }
if (TERMINAL.has(e.data.state) && e.event !== 'job.progress') setImmediate(close);
if (TERMINAL.has(e.data.state) && e.event !== 'job.progress') setImmediate(close);
},
},
close,
close,
};
};
+ // floating-ok: controller.enqueue is synchronous
controller.enqueue(enc.encode(`retry: ${stepMs}\n\n`));
controller.enqueue(enc.encode(`retry: ${stepMs}\n\n`));
streams.add(conn);
streams.add(conn);
for (const e of job.events) conn.push(e);
for (const e of job.events) conn.push(e);
@@ -53,6 +53,36 @@ const tests = [
@@ -53,6 +53,36 @@ const tests = [
assert.equal((await q.dead())[0].error, 'always');
assert.equal((await q.dead())[0].error, 'always');
await w.stop();
await w.stop();
}],
}],
+ ['valkey queue: a malformed stream entry is logged, never an unhandled rejection', async () => {
+ const warns = []; const unhandled = [];
+ const onUnhandled = (err) => unhandled.push(err);
+ process.on('unhandledRejection', onUnhandled);
+ let rejectRead = null;
+ const multi = () => { const m = {}; for (const fn of ['xack', 'xdel', 'xadd', 'zadd']) m[fn] = () => m; m.exec = async () => []; return m; };
+ let reads = 0;
+ const client = {
+ defineCommand() {}, ovQueuePromote: async () => 0, xgroup: async () => 'OK',
+ xautoclaim: async () => { throw new Error('nothing pending'); },
+ multi, set: async () => 'OK', zadd: async () => 1, xlen: async () => 0, zcard: async () => 0, xpending: async () => [0],
+ };
+ const reader = {
+ async xreadgroup() {
+ if (++reads === 1) return [['q:badentry', [['1-1', null]]]]; // ioredis gives null fields for an undecodable entry
+ await new Promise((resolve, reject) => { rejectRead = reject; });
+ },
+ disconnect() { if (rejectRead) rejectRead(new Error('disconnect')); },
+ };
+ const v = { client, key: (...p) => p.join(':'), duplicate: () => reader };
+ const q = createQueue({ valkey: v, name: 'badentry', log: { log() {}, warn: (m) => warns.push(m), error() {} } });
+ try {
+ const w = q.process(async () => { throw new Error('must not run'); }, { concurrency: 2, blockMs: 200 });
+ assert.ok(await until(async () => warns.length > 0), 'the malformed entry was logged');
+ assert.match(warns[0], /badentry.*1-1.*failed outside the handler/);
+ await sleep(100);
+ assert.deepEqual(unhandled, [], 'no unhandled rejection escaped the worker');
+ await w.stop();
+ } finally { process.off('unhandledRejection', onUnhandled); }
+ }],
['memory pubsub', async () => {
['memory pubsub', async () => {
const ps = createPubSub();
const ps = createPubSub();
const got = [];
const got = [];