From b80c526a9186c234090e77f4ae1cbb472fca6d3c Mon Sep 17 00:00:00 2001 From: lodar Date: Sat, 29 Aug 2026 17:37:13 +0000 Subject: [PATCH] DIVE-3809: serialise the pending drain ACROSS processes, and stop calling collection delivery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `draining`/`rerun` pair is module state — it serialises the three callers inside one process and is blind to a second one. Two plugin processes for the same agent each fetch the SAME pending rows and push every message twice, because the ack lands only after the notifications are sent. DIVE-3806 REFUTED this as the cause of the loss it observed (lifecycle.log showed exactly one live process across that window). It is still a real race and it is scope 3 of DIVE-3809. - an O_EXCL lock file at $STATE_DIR/pending-drain.lock, TIME-BOUNDED at 2min so a drain killed mid-flight cannot wedge the collect path permanently — which would turn an intermittent loss into a total one. - a filesystem that cannot support the lock degrades to today's per-process guard rather than blocking the collect. - log wording: the ack attests COLLECTION, never display. The SDK's send() has no reject path and a client with nothing subscribed drops the notification silently, so 'healed N undelivered' was a claim this code cannot make. test/dive3809-drain-lock.test.ts runs TWO real server.ts processes against one stub control plane and one shared state dir; dashboard-collect-now.test.ts runs a single process and stays green with the lock deleted, so it could not grade this. Graded by mutation: with acquireDrainLock stubbed to `return true`, both arms go red. Co-Authored-By: Claude Opus 5 --- plugins/dashboard/server.ts | 88 ++++++++++++++-- test/dive3809-drain-lock.test.ts | 168 +++++++++++++++++++++++++++++++ 2 files changed, 249 insertions(+), 7 deletions(-) create mode 100644 test/dive3809-drain-lock.test.ts diff --git a/plugins/dashboard/server.ts b/plugins/dashboard/server.ts index 6f25c43..fa61cab 100644 --- a/plugins/dashboard/server.ts +++ b/plugins/dashboard/server.ts @@ -27,7 +27,7 @@ import { ListToolsRequestSchema, CallToolRequestSchema, } from '@modelcontextprotocol/sdk/types.js' -import { readFileSync, mkdirSync, readdirSync, unlinkSync, watch, chmodSync, copyFileSync } from 'fs' +import { readFileSync, mkdirSync, readdirSync, unlinkSync, watch, chmodSync, copyFileSync, openSync, closeSync, writeSync, statSync } from 'fs' import { homedir } from 'os' import { join } from 'path' import { installLifecycle } from './lifecycle.ts' @@ -47,6 +47,16 @@ const AGENT_INBOX_DIR = join(STATE_DIR, 'agent-inbox') // purpose: a nudge carries no payload and must never be mistaken for a message // drop-file, so the two ingesters can never read each other's files. const COLLECT_NOW_DIR = join(STATE_DIR, 'collect-now') +// DIVE-3809: the drain's `draining`/`rerun` guard below is PER-PROCESS. Two +// plugin processes for the same agent (an overlapping restart, a stray +// supervisor respawn) each fetch the SAME pending rows and push every message +// into the session twice, because the ack lands only after the notifications +// are sent. This file is the cross-process arm of that guard: O_EXCL create +// wins the drain, everyone else skips this pass and picks it up on the next +// sweep. Refuted as the CAUSE of the loss DIVE-3806 observed (lifecycle.log +// showed exactly one live process across that window) — it is still a real +// race, and it is scope 3 of this row. +const DRAIN_LOCK = join(STATE_DIR, 'pending-drain.lock') mkdirSync(AGENT_INBOX_DIR, { recursive: true, mode: 0o700 }) mkdirSync(COLLECT_NOW_DIR, { recursive: true, mode: 0o700 }) @@ -206,10 +216,17 @@ function startCollectNowWatch(): void { // DIVE-848 offline heal: a message sent while this box was unreachable never // produced a drop file — it sits in the control plane with delivered_at NULL. // Pull those on boot (and on a slow sweep), push them into the session, then -// ack so they stamp delivered. Ack only AFTER the notifications are sent; a +// ack so they stamp COLLECTED. Ack only AFTER the notifications are sent; a // crash in between redelivers rather than losing the message. A row whose -// drop landed but whose delivered-stamp write failed may arrive twice — rare +// drop landed but whose collected-stamp write failed may arrive twice — rare // and preferable to silence. +// DIVE-3809: the ack no longer stamps delivered_at. It could never attest +// delivery, and stamping the column `/pending` reads meant one wrong ack +// deleted the only copy. A collected row is now merely hidden for a TTL and +// comes back, bounded by an attempt count. Note the consequence for the +// empty-text branch below: it acks a row it never pushed, so such a row is +// re-offered until the attempt bound retires it — bounded and visible, where +// before it was silently destroyed. // DIVE-3574: drainPending is now reachable from three places (boot, the 5-min // timer, and a collect nudge that can fire several times a second while someone // types in the dashboard) where it used to be reachable from two that could @@ -220,11 +237,63 @@ function startCollectNowWatch(): void { // landed after the fetch from waiting out the full timer. let draining = false let rerun = false + +// A drain that dies mid-flight (SIGKILL, box reboot) leaves the lock file +// behind, and a stale lock that nothing clears would wedge the collect path +// permanently — the exact failure shape this row exists to remove. So the lock +// is TIME-BOUNDED: older than this and it is treated as abandoned and broken. +// One drain is a fetch + N notifications + an ack, all with short timeouts; +// two minutes is far past any healthy pass. +const DRAIN_LOCK_STALE_MS = 2 * 60_000 + +// Returns true if THIS process now holds the lock. Never throws: a filesystem +// that cannot support the lock must degrade to today's per-process-only +// behaviour, not stop the customer's message from being collected. +function acquireDrainLock(): boolean { + for (let attempt = 0; attempt < 2; attempt++) { + try { + const fd = openSync(DRAIN_LOCK, 'wx') + try { writeSync(fd, `${process.pid} ${new Date().toISOString()}\n`) } catch {} + closeSync(fd) + return true + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== 'EEXIST') { + process.stderr.write(`dashboard channel: drain lock unavailable (${err}); per-process guard only\n`) + return true + } + // Held. Break it only if it is provably stale, then retry the create + // once — if another process wins that race, we simply skip this pass. + try { + const age = Date.now() - statSync(DRAIN_LOCK).mtimeMs + if (age <= DRAIN_LOCK_STALE_MS) return false + process.stderr.write(`dashboard channel: breaking stale drain lock (age ${Math.round(age / 1000)}s)\n`) + unlinkSync(DRAIN_LOCK) + } catch { return false } + } + } + return false +} + +function releaseDrainLock(): void { + try { unlinkSync(DRAIN_LOCK) } catch {} +} + async function drainPending(): Promise { if (draining) { rerun = true; return } draining = true try { - await drainPendingOnce() + // Cross-process (DIVE-3809). Skipping is safe and NOT a lost message: the + // holder is draining the same rows right now, and anything it misses is + // re-offered by the control plane once the collect TTL expires. + if (!acquireDrainLock()) { + process.stderr.write('dashboard channel: another process holds the drain lock; skipping this pass\n') + return + } + try { + await drainPendingOnce() + } finally { + releaseDrainLock() + } } finally { draining = false } @@ -265,7 +334,7 @@ async function drainPendingOnce(): Promise { }) acked.push(m.id) } catch (err) { - process.stderr.write(`dashboard channel: pending deliver failed for ${m.id}: ${err}\n`) + process.stderr.write(`dashboard channel: pending push failed for ${m.id}: ${err}\n`) } } if (acked.length === 0) return @@ -275,9 +344,14 @@ async function drainPendingOnce(): Promise { headers: { 'content-type': 'application/json', authorization: `Bearer ${TOKEN}` }, body: JSON.stringify({ agent: AGENT, ids: acked }), }) - process.stderr.write(`dashboard channel: healed ${acked.length} undelivered message(s)\n`) + // DIVE-3809: "collected", not "delivered" or "healed". This ack attests + // that the notification's bytes entered the stdout pipe — the SDK's send() + // has no reject path, and a client with nothing subscribed drops the + // notification silently — so it can never say the session displayed it. + // The control plane now re-offers a collected row whose TTL expires. + process.stderr.write(`dashboard channel: collected ${acked.length} pending message(s) (collection is not display)\n`) } catch (err) { - process.stderr.write(`dashboard channel: pending ack failed (will redeliver next boot): ${err}\n`) + process.stderr.write(`dashboard channel: pending ack failed (row stays uncollected; re-offered next sweep): ${err}\n`) } } diff --git a/test/dive3809-drain-lock.test.ts b/test/dive3809-drain-lock.test.ts new file mode 100644 index 0000000..a53b07f --- /dev/null +++ b/test/dive3809-drain-lock.test.ts @@ -0,0 +1,168 @@ +// DIVE-3809 scope 3: the drain guard must serialise ACROSS PROCESSES. +// +// server.ts's `draining`/`rerun` pair is module state — it serialises the three +// callers inside ONE process and is blind to a second one. Two plugin processes +// for the same agent (an overlapping restart, a stray respawn) each fetch the +// SAME pending rows and push every message into the session TWICE, because the +// ack only lands after the notifications are sent. +// +// DIVE-3806 REFUTED this as the cause of the loss it observed — lifecycle.log +// showed exactly one live process across that window — so this is not a fix for +// that. It is a real race, it is scope 3, and dashboard-collect-now.test.ts +// cannot see it: that harness runs a single process, so it grades the +// per-process half and would stay green with the lock deleted. +// +// The arm therefore runs TWO real server.ts processes against ONE stub control +// plane and ONE shared state dir, and counts pushes across BOTH. +import { describe, test, expect } from 'bun:test' +import { mkdtempSync, rmSync, mkdirSync, writeFileSync, existsSync, utimesSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +const SERVER = join(import.meta.dir, '..', 'plugins', 'dashboard', 'server.ts') +const BOOT_MS = 5_000 + +function spawnPlugin(dir: string, port: number, delivered: string[]) { + const proc = Bun.spawn(['bun', SERVER], { + env: { + ...process.env, + DASHBOARD_STATE_DIR: dir, + DASHBOARD_API_BASE: `http://127.0.0.1:${port}`, + CONNECTORD_TOKEN: 'test-token-abcdefghijkl', + USER: 'agent-dev', + }, + stdin: 'pipe', + stdout: 'pipe', + stderr: 'pipe', + }) + proc.stdin.write( + JSON.stringify({ + jsonrpc: '2.0', id: 1, method: 'initialize', + params: { protocolVersion: '2024-11-05', capabilities: {}, clientInfo: { name: 'lock-test', version: '0' } }, + }) + '\n' + ) + proc.stdin.flush() + void (async () => { + const reader = proc.stdout.getReader() + const dec = new TextDecoder() + let buf = '' + for (;;) { + const { done, value } = await reader.read() + if (done) return + buf += dec.decode(value, { stream: true }) + const lines = buf.split('\n') + buf = lines.pop() ?? '' + for (const line of lines) { + if (!line.trim()) continue + try { + const msg = JSON.parse(line) + if (msg.method === 'notifications/claude/channel') delivered.push(String(msg.params?.content ?? '')) + if (msg.id === 1) { + proc.stdin.write(JSON.stringify({ jsonrpc: '2.0', method: 'notifications/initialized' }) + '\n') + proc.stdin.flush() + } + } catch {} + } + } + })() + return proc +} + +async function waitFor(pred: () => boolean, ms: number) { + const deadline = Date.now() + ms + while (Date.now() < deadline) { + if (pred()) return true + await Bun.sleep(25) + } + return pred() +} + +describe('cross-process drain lock (DIVE-3809)', () => { + test('two plugin processes on one state dir push each message EXACTLY once', async () => { + const dir = mkdtempSync(join(tmpdir(), 'drain-lock-')) + const delivered: string[] = [] + let queue = [ + { id: 1, text: 'first' }, + { id: 2, text: 'second' }, + ] + let pendingHits = 0 + // Snapshot at REQUEST time then delay, exactly as dashboard-collect-now + // does: a stub that re-reads after the delay hands the second drain the + // state AFTER the first one's ack and silently removes the race. + const api = Bun.serve({ + port: 0, + async fetch(req) { + const url = new URL(req.url) + if (url.pathname === '/server/messages/pending') { + pendingHits++ + const snapshot = queue + await Bun.sleep(900) + return Response.json({ pending: snapshot }) + } + if (url.pathname === '/server/messages/pending/ack') { + const body = (await req.json()) as { ids: number[] } + queue = queue.filter(m => !body.ids.includes(m.id)) + return Response.json({ ok: true }) + } + return new Response('not found', { status: 404 }) + }, + }) + const a = spawnPlugin(dir, api.port, delivered) + const b = spawnPlugin(dir, api.port, delivered) + try { + // Both boot drains fire ~5s after start, i.e. genuinely overlapping. + await waitFor(() => delivered.length >= 2, BOOT_MS + 10_000) + // Settle: give a second (unserialised) drain every chance to double-push. + await Bun.sleep(3_000) + const counts = new Map() + for (const d of delivered) counts.set(d, (counts.get(d) ?? 0) + 1) + expect(delivered.length).toBeGreaterThanOrEqual(2) + expect([...counts.entries()].filter(([, n]) => n > 1)).toEqual([]) + expect(pendingHits).toBeGreaterThan(0) + } finally { + a.kill(); b.kill(); api.stop(true) + rmSync(dir, { recursive: true, force: true }) + } + }, 40_000) + + test('a stale lock left by a killed drain is broken, not honoured forever', async () => { + // The failure this forecloses: a lock file is the classic way to turn an + // intermittent loss into a permanent one. A drain killed mid-flight leaves + // the file behind, and if nothing breaks it the collect path is wedged for + // the life of the box — strictly worse than the bug being fixed. + const dir = mkdtempSync(join(tmpdir(), 'stale-lock-')) + mkdirSync(dir, { recursive: true }) + const lock = join(dir, 'pending-drain.lock') + writeFileSync(lock, '999999 stale\n') + // Age it well past the 2-minute staleness bound. utimesSync, not `touch + // -d` — bun's built-in shell does not implement that flag. + const old = new Date(Date.now() - 10 * 60_000) + utimesSync(lock, old, old) + const delivered: string[] = [] + let queue = [{ id: 7, text: 'after a crash' }] + const api = Bun.serve({ + port: 0, + async fetch(req) { + const url = new URL(req.url) + if (url.pathname === '/server/messages/pending') return Response.json({ pending: queue }) + if (url.pathname === '/server/messages/pending/ack') { + const body = (await req.json()) as { ids: number[] } + queue = queue.filter(m => !body.ids.includes(m.id)) + return Response.json({ ok: true }) + } + return new Response('not found', { status: 404 }) + }, + }) + const p = spawnPlugin(dir, api.port, delivered) + try { + const got = await waitFor(() => delivered.length >= 1, BOOT_MS + 10_000) + expect(got).toBe(true) + expect(delivered[0]).toBe('after a crash') + // And the drain cleaned up after itself, so the next pass is not blocked. + expect(existsSync(lock)).toBe(false) + } finally { + p.kill(); api.stop(true) + rmSync(dir, { recursive: true, force: true }) + } + }, 40_000) +})