From aaad1e7a3add05b86b8a380e8f040246fe962666 Mon Sep 17 00:00:00 2001 From: Thomas Joise Date: Sat, 18 Jul 2026 00:31:06 +0530 Subject: [PATCH] fix(sec/d24): concurrency + workflow isolation Red-team correctness findings: - SqliteDb.transaction now serializes top-level transactions with an async mutex and tracks re-entrancy via AsyncLocalStorage (nested calls -> savepoints, no deadlock). The old shared txnDepth counter let concurrent async callers interleave BEGIN/COMMIT on the one connection -> silent data loss; a 50-way concurrent read-modify-write test now lands exactly. - generateRecurring wraps each plan in try/catch: one misconfigured plan is parked (audited 'generate_failed') and the scan continues, instead of aborting all remaining generation and the entire auto-send drain, every scan. - POST /modules/:id/notify broadcast IIFE gets a trailing catch so a mid-loop throw answers the request instead of hanging it. Full suite 388 green. Co-Authored-By: Claude Fable 5 --- apps/hq/src/api.ts | 7 +++- apps/hq/src/db.ts | 58 ++++++++++++++++++++++++-------- apps/hq/src/scheduler.ts | 27 ++++++++++----- apps/hq/test/concurrency.test.ts | 46 +++++++++++++++++++++++++ 4 files changed, 115 insertions(+), 23 deletions(-) create mode 100644 apps/hq/test/concurrency.test.ts diff --git a/apps/hq/src/api.ts b/apps/hq/src/api.ts index fc96072..b8101a4 100644 --- a/apps/hq/src/api.ts +++ b/apps/hq/src/api.ts @@ -550,7 +550,12 @@ export function apiRouter( } // Warn, never truncate: every unreached client is named in the response. res.json({ ok: true, sent, failed, total: targets.length }) - })() + })().catch((err: unknown) => { + // D24: a throw inside the broadcast loop must answer the request, not hang it. + if (!res.headersSent) { + res.status(500).json({ ok: false, error: err instanceof Error ? err.message : String(err) }) + } + }) } catch (err) { res.status(400).json({ ok: false, error: err instanceof Error ? err.message : String(err) }) } diff --git a/apps/hq/src/db.ts b/apps/hq/src/db.ts index febd270..63fcf4d 100644 --- a/apps/hq/src/db.ts +++ b/apps/hq/src/db.ts @@ -1,3 +1,4 @@ +import { AsyncLocalStorage } from 'node:async_hooks' import Database from 'better-sqlite3' import fs from 'node:fs' import path from 'node:path' @@ -30,7 +31,13 @@ export type SqliteRaw = Database.Database export class SqliteDb implements DB { readonly engine = 'sqlite' as const private stmts = new Map() - private txnDepth = 0 + // D24 (red-team): a single SQLite connection cannot host two overlapping + // BEGIN..COMMIT blocks. `mutex` is a promise chain that serializes TOP-LEVEL + // transactions so only one is ever open; `als` tracks re-entrancy on a call + // chain so a genuinely nested transaction() becomes a SAVEPOINT (and does NOT + // wait on the mutex — that would deadlock, since it runs inside the held lock). + private mutex: Promise = Promise.resolve() + private als = new AsyncLocalStorage<{ depth: number }>() constructor(readonly raw: SqliteRaw) {} private stmt(sql: string): Database.Statement { let s = this.stmts.get(sql) @@ -47,22 +54,45 @@ export class SqliteDb implements DB { return { changes: this.stmt(sql).run(...args).changes } } async exec(sql: string): Promise { this.raw.exec(sql) } + async transaction(fn: () => Promise | T): Promise { - // better-sqlite3's own transaction() rejects async fns, so BEGIN/SAVEPOINT by - // hand with a depth counter. Single connection: interleaved writers would join - // the open txn — fine for the dev/test engine (requests are awaited serially). - const depth = this.txnDepth++ - const sp = `sp_d${depth}` - this.raw.exec(depth === 0 ? 'BEGIN' : `SAVEPOINT ${sp}`) + const store = this.als.getStore() + if (store !== undefined) { + // Nested on the same call chain → savepoint (the outer BEGIN already holds + // the mutex, so we must not queue on it). + const sp = `sp_d${++store.depth}` + this.raw.exec(`SAVEPOINT ${sp}`) + try { + const out = await fn() + this.raw.exec(`RELEASE ${sp}`) + return out + } catch (err) { + this.raw.exec(`ROLLBACK TO ${sp}; RELEASE ${sp}`) + throw err + } finally { + store.depth-- + } + } + // Top-level: take the mutex so no other BEGIN overlaps this one. Chain onto the + // previous transaction's tail (ignoring its rejection) and expose our own tail. + let release!: () => void + const prior = this.mutex + this.mutex = new Promise((r) => { release = r }) + await prior.catch(() => undefined) try { - const out = await fn() - this.raw.exec(depth === 0 ? 'COMMIT' : `RELEASE ${sp}`) - return out - } catch (err) { - this.raw.exec(depth === 0 ? 'ROLLBACK' : `ROLLBACK TO ${sp}; RELEASE ${sp}`) - throw err + return await this.als.run({ depth: 0 }, async () => { + this.raw.exec('BEGIN') + try { + const out = await fn() + this.raw.exec('COMMIT') + return out + } catch (err) { + this.raw.exec('ROLLBACK') + throw err + } + }) } finally { - this.txnDepth-- + release() } } async close(): Promise { this.raw.close() } diff --git a/apps/hq/src/scheduler.ts b/apps/hq/src/scheduler.ts index 47ef519..612ee79 100644 --- a/apps/hq/src/scheduler.ts +++ b/apps/hq/src/scheduler.ts @@ -94,15 +94,26 @@ async function generateRecurring(db: DB, today: string, now: string, created: Re ) const autoQueue: string[] = [] for (const { id } of plans) { - let guard = 0 - for (;;) { - if (guard++ > 240) break // safety cap: ~20 years of monthly against a corrupt next_run - const claim = await db.transaction(() => claimAndGenerate(db, id, today, now)) - if (claim === 'done') break - if (claim.created) { - bump(created, 'recurring_generated') - if (claim.auto) autoQueue.push(claim.reminderId) + // D24 (red-team): isolate each plan. One misconfigured plan throwing must not abort + // generation for every remaining plan (and, downstream, the whole auto-send drain) — + // park it (audited) and move on, exactly as the auto-send loop isolates per reminder. + try { + let guard = 0 + for (;;) { + if (guard++ > 240) break // safety cap: ~20 years of monthly against a corrupt next_run + const claim = await db.transaction(() => claimAndGenerate(db, id, today, now)) + if (claim === 'done') break + if (claim.created) { + bump(created, 'recurring_generated') + if (claim.auto) autoQueue.push(claim.reminderId) + } } + } catch (err) { + bump(created, 'recurring_failed') + // eslint-disable-next-line no-console + console.error(`[scheduler] recurring plan ${id} failed, skipped:`, err instanceof Error ? err.message : err) + await writeAudit(db, 'system', 'generate_failed', 'recurring_plan', id, undefined, + { error: err instanceof Error ? err.message : String(err) }).catch(() => undefined) } } return autoQueue diff --git a/apps/hq/test/concurrency.test.ts b/apps/hq/test/concurrency.test.ts new file mode 100644 index 0000000..37fbae7 --- /dev/null +++ b/apps/hq/test/concurrency.test.ts @@ -0,0 +1,46 @@ +// apps/hq/test/concurrency.test.ts — D24 (red-team): SQLite transaction serialization. +import { describe, it, expect } from 'vitest' +import { openDb } from '../src/db' + +describe('SqliteDb.transaction serialization (D24)', () => { + it('overlapping transactions do not interleave — a concurrent counter stays consistent', async () => { + const db = openDb(':memory:') + await db.exec(`CREATE TABLE ctr (id INTEGER PRIMARY KEY, n INTEGER NOT NULL)`) + await db.run(`INSERT INTO ctr (id, n) VALUES (1, 0)`) + // 50 read-modify-write transactions fired concurrently. Without serialization the + // read-then-write would race and lose increments; the mutex makes the final count exact. + const bump = () => db.transaction(async () => { + const row = (await db.get<{ n: number }>(`SELECT n FROM ctr WHERE id=1`))! + // Yield the event loop mid-transaction — the moment a naive impl would interleave. + await Promise.resolve() + await db.run(`UPDATE ctr SET n=? WHERE id=1`, row.n + 1) + }) + await Promise.all(Array.from({ length: 50 }, bump)) + expect((await db.get<{ n: number }>(`SELECT n FROM ctr WHERE id=1`))!.n).toBe(50) + await db.close() + }) + + it('a nested transaction still commits with its parent (savepoint), and rolls back alone on inner failure', async () => { + const db = openDb(':memory:') + await db.exec(`CREATE TABLE t (k TEXT PRIMARY KEY)`) + // Outer commits; a caught inner failure rolls back only the savepoint. + await db.transaction(async () => { + await db.run(`INSERT INTO t (k) VALUES ('outer')`) + await db.transaction(async () => { + await db.run(`INSERT INTO t (k) VALUES ('inner')`) + throw new Error('inner boom') + }).catch(() => undefined) + }) + expect(await db.get(`SELECT k FROM t WHERE k='outer'`)).toBeDefined() + expect(await db.get(`SELECT k FROM t WHERE k='inner'`)).toBeUndefined() + // An outer failure rolls back everything including the nested savepoint work. + await db.transaction(async () => { + await db.run(`INSERT INTO t (k) VALUES ('a')`) + await db.transaction(async () => { await db.run(`INSERT INTO t (k) VALUES ('b')`) }) + throw new Error('outer boom') + }).catch(() => undefined) + expect(await db.get(`SELECT k FROM t WHERE k='a'`)).toBeUndefined() + expect(await db.get(`SELECT k FROM t WHERE k='b'`)).toBeUndefined() + await db.close() + }) +})