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 <noreply@anthropic.com>
feat/client-detail-redesign
Thomas Joise 4 days ago
parent a7b3ade22e
commit aaad1e7a3a

@ -550,7 +550,12 @@ export function apiRouter(
} }
// Warn, never truncate: every unreached client is named in the response. // Warn, never truncate: every unreached client is named in the response.
res.json({ ok: true, sent, failed, total: targets.length }) 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) { } catch (err) {
res.status(400).json({ ok: false, error: err instanceof Error ? err.message : String(err) }) res.status(400).json({ ok: false, error: err instanceof Error ? err.message : String(err) })
} }

@ -1,3 +1,4 @@
import { AsyncLocalStorage } from 'node:async_hooks'
import Database from 'better-sqlite3' import Database from 'better-sqlite3'
import fs from 'node:fs' import fs from 'node:fs'
import path from 'node:path' import path from 'node:path'
@ -30,7 +31,13 @@ export type SqliteRaw = Database.Database
export class SqliteDb implements DB { export class SqliteDb implements DB {
readonly engine = 'sqlite' as const readonly engine = 'sqlite' as const
private stmts = new Map<string, Database.Statement>() private stmts = new Map<string, Database.Statement>()
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<void> = Promise.resolve()
private als = new AsyncLocalStorage<{ depth: number }>()
constructor(readonly raw: SqliteRaw) {} constructor(readonly raw: SqliteRaw) {}
private stmt(sql: string): Database.Statement { private stmt(sql: string): Database.Statement {
let s = this.stmts.get(sql) let s = this.stmts.get(sql)
@ -47,22 +54,45 @@ export class SqliteDb implements DB {
return { changes: this.stmt(sql).run(...args).changes } return { changes: this.stmt(sql).run(...args).changes }
} }
async exec(sql: string): Promise<void> { this.raw.exec(sql) } async exec(sql: string): Promise<void> { this.raw.exec(sql) }
async transaction<T>(fn: () => Promise<T> | T): Promise<T> { async transaction<T>(fn: () => Promise<T> | T): Promise<T> {
// better-sqlite3's own transaction() rejects async fns, so BEGIN/SAVEPOINT by const store = this.als.getStore()
// hand with a depth counter. Single connection: interleaved writers would join if (store !== undefined) {
// the open txn — fine for the dev/test engine (requests are awaited serially). // Nested on the same call chain → savepoint (the outer BEGIN already holds
const depth = this.txnDepth++ // the mutex, so we must not queue on it).
const sp = `sp_d${depth}` const sp = `sp_d${++store.depth}`
this.raw.exec(depth === 0 ? 'BEGIN' : `SAVEPOINT ${sp}`) this.raw.exec(`SAVEPOINT ${sp}`)
try { try {
const out = await fn() const out = await fn()
this.raw.exec(depth === 0 ? 'COMMIT' : `RELEASE ${sp}`) this.raw.exec(`RELEASE ${sp}`)
return out return out
} catch (err) { } catch (err) {
this.raw.exec(depth === 0 ? 'ROLLBACK' : `ROLLBACK TO ${sp}; RELEASE ${sp}`) this.raw.exec(`ROLLBACK TO ${sp}; RELEASE ${sp}`)
throw err throw err
} finally { } finally {
this.txnDepth-- 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<void>((r) => { release = r })
await prior.catch(() => undefined)
try {
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 {
release()
} }
} }
async close(): Promise<void> { this.raw.close() } async close(): Promise<void> { this.raw.close() }

@ -94,6 +94,10 @@ async function generateRecurring(db: DB, today: string, now: string, created: Re
) )
const autoQueue: string[] = [] const autoQueue: string[] = []
for (const { id } of plans) { for (const { id } of plans) {
// 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 let guard = 0
for (;;) { for (;;) {
if (guard++ > 240) break // safety cap: ~20 years of monthly against a corrupt next_run if (guard++ > 240) break // safety cap: ~20 years of monthly against a corrupt next_run
@ -104,6 +108,13 @@ async function generateRecurring(db: DB, today: string, now: string, created: Re
if (claim.auto) autoQueue.push(claim.reminderId) 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 return autoQueue
} }

@ -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()
})
})
Loading…
Cancel
Save