diff --git a/apps/broker/src/broker.ts b/apps/broker/src/broker.ts index ae2a3c5..a7e75e6 100644 --- a/apps/broker/src/broker.ts +++ b/apps/broker/src/broker.ts @@ -311,15 +311,62 @@ export async function sweepStalePresences(): Promise { */ export async function sweepOrphanMessages(): Promise { const cutoff = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000); - const result = await db.execute(sql` - DELETE FROM mesh.message_queue - WHERE delivered_at IS NULL - AND created_at < ${cutoff} - RETURNING id - `); - const rows = (result as unknown as { rows?: unknown[]; length?: number }).rows ?? result; - const count = Array.isArray(rows) ? rows.length : 0; - return count; + // 2026-09-08: rewritten with the query builder. The previous raw + // `sql\`… created_at < ${cutoff}\`` threw on every hourly run + // (`DrizzleQueryError … The "string" argument must be of type string … + // Received an instance of Date`) — the Date param was never serialized + // for the raw path, so orphan rows were never swept. The builder path + // (same shape as sweepPendingStatuses) serializes timestamps correctly. + const deleted = await db + .delete(messageQueue) + .where(and(isNull(messageQueue.deliveredAt), lt(messageQueue.createdAt, cutoff))) + .returning({ id: messageQueue.id }); + return deleted.length; +} + +/** Presence rows are retained this long after disconnect, then deleted. */ +export const PRESENCE_RETENTION_MS = 30 * 24 * 60 * 60 * 1000; +/** Rows deleted per statement; bounded so the delete never holds a long + * lock on the hot presence table. */ +export const PRESENCE_PRUNE_BATCH = 20_000; + +/** + * Retention sweep for `mesh.presence` (2026-09-08, RC-B). + * + * Every connect inserts a presence row; nothing ever deleted them. Prod + * had 2,491,950 rows (2,488,046 disconnected > 30 d ago) and no index + * beyond the PK, so `sweepStalePresences` (every 30 s), `listPeersInMesh` + * and the hook status lookup were full-scanning 2.5 M rows. Under a + * reconnect storm those scans pushed hello latency to ~5 s, which tripped + * client ack timeouts and widened the close-handler race window. Indexes + * (migration 0030) make the hot queries O(online rows); this sweep keeps + * the table from growing without bound so it stays that way. + * + * Batched (`PRESENCE_PRUNE_BATCH` per statement, up to `maxBatches` per + * call) so a first run against a huge backlog does not hold locks for + * minutes. Returns the number of rows deleted. + */ +export async function sweepOldPresenceRows(maxBatches = 25): Promise { + const cutoff = new Date(Date.now() - PRESENCE_RETENTION_MS); + let total = 0; + for (let i = 0; i < maxBatches; i++) { + const result = await db.execute(sql` + DELETE FROM mesh.presence + WHERE id IN ( + SELECT id FROM mesh.presence + WHERE disconnected_at IS NOT NULL + AND disconnected_at < ${cutoff.toISOString()}::timestamp + LIMIT ${PRESENCE_PRUNE_BATCH} + ) + RETURNING id + `); + const rows = ((result as unknown as { rows?: unknown[] }).rows ?? (result as unknown as unknown[])) as unknown[]; + const n = Array.isArray(rows) ? rows.length : 0; + total += n; + if (n < PRESENCE_PRUNE_BATCH) break; + } + if (total > 0) metrics.presenceRowsPruned.inc({}, total); + return total; } /** Sweep expired pending_status entries. */ @@ -2621,6 +2668,15 @@ export function startSweepers(): void { .then((n) => { if (n > 0) console.log(`[broker] orphan msgs swept: ${n}`); }) .catch((e) => console.error("[broker] orphan msg sweep:", e)); }, 60 * 60_000).unref(); + // Presence retention: first pass 60s after boot (let the reconnect + // storm that follows a deploy settle first), then hourly. + const runPresencePrune = () => { + sweepOldPresenceRows() + .then((n) => { if (n > 0) console.log(`[broker] presence rows pruned: ${n}`); }) + .catch((e) => console.error("[broker] presence prune:", e)); + }; + setTimeout(runPresencePrune, 60_000).unref(); + setInterval(runPresencePrune, 60 * 60_000).unref(); } /** Stop background sweepers and mark all active presences disconnected. */ diff --git a/apps/broker/src/index.ts b/apps/broker/src/index.ts index 67c91cb..62fc42f 100644 --- a/apps/broker/src/index.ts +++ b/apps/broker/src/index.ts @@ -123,6 +123,25 @@ import { canonicalInvite, canonicalInviteV2, claimInviteV2Core as _claimInviteV2 const claimInviteV2Core = _claimInviteV2Core; import { handleWebhook } from "./webhooks"; import { audit, loadLastHashes, ensureAuditLogTable, verifyChain, queryAuditLog } from "./audit"; +import { + decideCloseAction, + shouldEvictOnGraceExpiry, + dedupTargets, + findZombieSockets, + CLOSE_CODE_PRESENCE_EVICTED, + CLOSE_CODE_NO_PRESENCE, +} from "./lease"; + +/** + * When each raw WS socket was accepted. Used by the zombie sweeper to + * leave a grace period for sockets that are still mid-hello. WeakMap so + * closed sockets are GC'd without bookkeeping. + */ +const socketOpenedAt = new WeakMap(); +/** A socket must have been open this long with no presence before the + * zombie sweeper closes it. Hello + auth + DB work never takes this long + * even under the 2026-09-08 storm (worst observed: ~5s). */ +const ZOMBIE_MIN_AGE_MS = 60_000; const PORT = env.BROKER_PORT; const WS_PATH = "/ws"; @@ -583,7 +602,7 @@ function sendToPeer(presenceId: string, msg: WSServerMessage): void { * Idempotent: re-entering when the connections entry is already gone * is a no-op. */ -async function evictPresenceFully(presenceId: string): Promise { +async function evictPresenceFully(presenceId: string, reason = "unspecified"): Promise { const conn = connections.get(presenceId); if (!conn) return; // already evicted if (conn.evictionTimer) { @@ -593,6 +612,17 @@ async function evictPresenceFully(presenceId: string): Promise { connections.delete(presenceId); decMeshCount(conn.meshId); + // 2026-09-08 (Addendum 2/5): if the lease's socket is still OPEN, the + // client believes it is connected while the broker has just forgotten + // it — a zombie that can neither be discovered nor receive pushes, and + // whose pings the `ws` library keeps answering. Close it with a + // distinct code so the client's reconnect path re-hellos and recreates + // the presence. Never leave an evicted lease's socket open. + if (conn.ws.readyState === conn.ws.OPEN || conn.ws.readyState === conn.ws.CONNECTING) { + try { conn.ws.close(CLOSE_CODE_PRESENCE_EVICTED, "presence_evicted"); } catch { /* already dead */ } + log.warn("ws evict: closed still-open socket", { presence_id: presenceId, reason }); + } + const leaveMsg: WSPushMessage = { type: "push", subtype: "system", @@ -656,7 +686,7 @@ async function evictPresenceFully(presenceId: string): Promise { log.info("clock auto-paused (mesh empty)", { mesh_id: conn.meshId }); } } - log.info("ws evict full", { presence_id: presenceId }); + log.info("ws evict full", { presence_id: presenceId, reason }); } async function maybePushQueuedMessages( @@ -1824,6 +1854,7 @@ async function handleHello( * augmentation. The session was never visibly absent from peers. */ silent?: boolean; } | null> { + const helloStartedAt = Date.now(); // Validate sessionPubkey shape — it becomes a routable identity in // listPeers/drainForMember, so arbitrary strings let a client claim // nonsense pubkeys. Required-if-present: empty is allowed (falls back @@ -1970,17 +2001,27 @@ async function handleHello( }; } - // Session-id dedup: if this session_id already has an active presence, - // disconnect the ghost. Happens when a client reconnects after a - // network blip or broker restart before the 90s stale sweeper runs. - // One Claude Code instance = one session_id = one presence, always. - for (const [oldPid, oldConn] of connections) { - if (oldConn.meshId === hello.meshId && oldConn.sessionId === hello.sessionId) { - log.info("hello dedup", { old_presence: oldPid, session_id: hello.sessionId }); - try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } - dropConnection(oldPid); - void disconnectPresence(oldPid); - } + // Session-id dedup: if this session_id already has an active presence + // OF THE SAME ROLE, disconnect the ghost. Happens when a client + // reconnects after a network blip or broker restart before the 90s + // stale sweeper runs. One control-plane connection per session_id. + // + // 2026-09-08 (Addendum 2/5, item 4): scoped by role. The launch + // wrapper's member-level `hello` carries the Claude session UUID as + // session_id — the same value the daemon's `session_hello` uses — so an + // unscoped dedup kicked the live session presence, whose re-hello then + // kicked the wrapper's presence, and so on (2 evictions + 2 DB + // disconnects + a forced close per relaunch). A control-plane presence + // and a session presence for one session must coexist. + for (const oldPid of dedupTargets(connections, hello, "control-plane")) { + const oldConn = connections.get(oldPid); + if (!oldConn) continue; + log.info("hello dedup", { old_presence: oldPid, session_id: hello.sessionId, role: "control-plane" }); + // Drop the entry BEFORE closing: under Bun `close()` fires the close + // handler synchronously (RC-C), and with the entry gone it is a no-op. + dropConnection(oldPid); + try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } + void disconnectPresence(oldPid); } const presenceId = await connectPresence({ @@ -2033,6 +2074,10 @@ async function handleHello( presence_id: presenceId, session_id: hello.sessionId, restored: !!saved, + // Hello processing latency. The 2026-09-08 storm pushed this to ~5s + // (DB full scans on an unindexed 2.5M-row presence table), tripping + // the daemon's ack timeout. Watch this number. + latency_ms: Date.now() - helloStartedAt, }); // Drain any queued messages in the background. The hello_ack is // sent by the CALLER after it assigns presenceId — sending it here @@ -2080,6 +2125,7 @@ async function handleSessionHello( * absent from peers. */ silent?: boolean; } | null> { + const helloStartedAt = Date.now(); // Shape checks. The crypto helpers also enforce these but bailing // early gives a clearer error code on the wire. if (!/^[0-9a-f]{64}$/.test(hello.sessionPubkey ?? "")) { @@ -2228,17 +2274,33 @@ async function handleSessionHello( // presence per session pubkey" invariant — it kills the same-name // ghost that used to win queued-DM claim races. const wasState = oldConn.leaseState; - if (oldConn.evictionTimer) { - clearTimeout(oldConn.evictionTimer); - oldConn.evictionTimer = null; - } - if (oldConn.ws !== ws) { - try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } - } + const replacedWs = oldConn.ws !== ws ? oldConn.ws : null; + // ORDER MATTERS (2026-09-08, RC-C): swap the lease onto the new socket + // and mark it online BEFORE closing the old socket. Under Bun (the + // production runtime) the `ws` shim emits the old socket's `close` + // event SYNCHRONOUSLY inside `close()` — verified 2026-09-08 with a + // 20-line repro; Node emits it on a later tick. With the old order the + // close handler ran while `conn.ws` was still the old socket, saw an + // online lease, and put THIS lease into grace with a 90 s eviction + // timer that the swap below never cleared — the reattached session was + // evicted 90 s later while its new socket stayed open (ghost). Swapping + // first makes that handler see `conn.ws !== closingWs` → ignored. oldConn.ws = ws; oldConn.leaseState = "online"; oldConn.leaseUntil = 0; oldConn.lastPongAt = Date.now(); + if (replacedWs) { + try { replacedWs.close(1000, "session_replaced"); } catch { /* already dead */ } + } + // Clear any grace timer — including one a synchronous close event may + // have just armed — AFTER the close call, and restore online state in + // case that handler flipped it. + if (oldConn.evictionTimer) { + clearTimeout(oldConn.evictionTimer); + oldConn.evictionTimer = null; + } + oldConn.leaseState = "online"; + oldConn.leaseUntil = 0; // Refresh mutable fields from the new hello. oldConn.sessionId = hello.sessionId; oldConn.cwd = hello.cwd; @@ -2259,15 +2321,19 @@ async function handleSessionHello( }; } - // Session-id dedup: if the same session_id is already connected, kick - // the ghost. Reconnect after a network blip lands here cleanly. - for (const [oldPid, oldConn] of connections) { - if (oldConn.meshId === hello.meshId && oldConn.sessionId === hello.sessionId) { - log.info("session_hello dedup", { old_presence: oldPid, session_id: hello.sessionId }); - try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } - dropConnection(oldPid); - void disconnectPresence(oldPid); - } + // Session-id dedup: if the same session_id already has a SESSION-role + // presence (different pubkey — e.g. keypair rotated), kick the ghost. + // Reconnect after a network blip lands here cleanly. Scoped by role + // (see the matching comment in handleHello): the wrapper's + // control-plane presence for this session_id is left alone. + for (const oldPid of dedupTargets(connections, hello, "session")) { + const oldConn = connections.get(oldPid); + if (!oldConn) continue; + log.info("session_hello dedup", { old_presence: oldPid, session_id: hello.sessionId, role: "session" }); + // Drop the entry BEFORE closing (RC-C: Bun fires `close` synchronously). + dropConnection(oldPid); + try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } + void disconnectPresence(oldPid); } const presenceId = await connectPresence({ @@ -2319,6 +2385,7 @@ async function handleSessionHello( presence_id: presenceId, session_id: hello.sessionId, session_pubkey: hello.sessionPubkey.slice(0, 12), + latency_ms: Date.now() - helloStartedAt, }); // Drain any DMs queued for this session pubkey (or the parent member). void maybePushQueuedMessages(presenceId); @@ -2679,6 +2746,7 @@ async function handleSend( function handleConnection(ws: WebSocket): void { metrics.connectionsTotal.inc(); + socketOpenedAt.set(ws, Date.now()); let presenceId: string | null = null; ws.on("message", async (raw) => { try { @@ -5432,32 +5500,46 @@ function handleConnection(ws: WebSocket): void { }); } }); - ws.on("close", async () => { + ws.on("close", (code, reasonBuf) => { if (!presenceId) return; - const conn = connections.get(presenceId); - if (!conn) return; // already evicted - - // If the conn's `ws` is no longer THIS ws, the close belongs to an - // older socket that was already replaced by a reattach. Ignore — the - // lease is healthy with the new WS, no eviction needed. - if (conn.ws !== ws) { - log.debug("ws close on replaced socket — ignoring", { presence_id: presenceId }); - return; - } - - await savePeerState(conn, conn.memberId, conn.meshId); - - // If lease is currently online, enter grace. Other peers see the - // session as still online; DMs queue (sendToPeer no-ops on dead - // WS, drain on reattach). After GRACE_MS without a reattach, the - // timer fires evictPresenceFully and cleanup runs as before. const pid = presenceId; - if (conn.leaseState === "online") { - conn.leaseState = "offline"; - conn.leaseUntil = Date.now() + GRACE_MS; - conn.evictionTimer = setTimeout(() => { + const conn = connections.get(pid); + const reason = reasonBuf.toString("utf8"); + + // 2026-09-08 (RC-A): the decision MUST be taken synchronously, at the + // instant the close fires. The previous handler checked `conn.ws !== + // ws` here, then `await savePeerState(...)`, then mutated the lease. + // Under DB latency the close of the DEAD daemon's socket resumed after + // the NEW daemon had reattached (conn.ws swapped, lease online) and put + // the healthy lease into grace / evicted it — while the new socket + // stayed open as a zombie. No await may sit between the identity check + // and the state transition. + const action = decideCloseAction(conn, ws); + log.info("ws close", { presence_id: pid, code, reason, action }); + if (action === "no_conn") return; // already evicted + if (action === "ignore_replaced") return; // older socket of a reattached lease + + // Persist peer state best-effort, AFTER the decision, off the hot path. + void savePeerState(conn!, conn!.memberId, conn!.meshId).catch((e) => + log.warn("savePeerState on close failed", { presence_id: pid, error: e instanceof Error ? e.message : String(e) }), + ); + + if (action === "enter_grace") { + // Other peers see the session as still online; DMs queue (sendToPeer + // no-ops on dead WS, drain on reattach). After GRACE_MS without a + // reattach, the timer fires evictPresenceFully and cleanup runs. + conn!.leaseState = "offline"; + conn!.leaseUntil = Date.now() + GRACE_MS; + conn!.evictionTimer = setTimeout(() => { + const cur = connections.get(pid); + if (!shouldEvictOnGraceExpiry(cur)) { + // A reattach landed and restored the lease after this timer was + // armed (clearTimeout raced the firing). Never evict a live lease. + log.info("lease grace timer fired on online lease — skipped", { presence_id: pid }); + return; + } log.info("lease grace expired — evicting", { presence_id: pid }); - void evictPresenceFully(pid); + void evictPresenceFully(pid, "grace_expired"); }, GRACE_MS); log.info("ws close — lease grace started", { presence_id: pid, @@ -5466,9 +5548,9 @@ function handleConnection(ws: WebSocket): void { return; } - // Not online (already in grace from an earlier close, or odd state). - // Run full eviction immediately. - await evictPresenceFully(pid); + // "evict_now": already offline (second close while in grace, or odd + // state). Run full eviction immediately. + void evictPresenceFully(pid, "close_while_offline"); }); ws.on("error", (err) => { log.warn("ws error", { error: err.message }); @@ -5678,6 +5760,31 @@ async function main(): Promise { const STALE_PONG_THRESHOLD_MS = 75_000; const pingInterval = setInterval(() => { const now = Date.now(); + + // 2026-09-08 zombie sweeper (Addendum 5, item 3): a socket that is OPEN + // but referenced by no lease can neither be discovered nor pushed to, + // yet its owner thinks it is connected (the `ws` lib auto-pongs). Any + // eviction path that forgets to close the socket, or any future bug of + // the same family, is healed here: closing forces the client to + // reconnect and re-hello. Sockets younger than ZOMBIE_MIN_AGE_MS are + // skipped (they may be mid-hello). + const liveSockets = new Set(); + for (const [, conn] of connections) liveSockets.add(conn.ws); + const zombies = findZombieSockets( + wss.clients as Set, + liveSockets, + (s) => socketOpenedAt.get(s), + now, + ZOMBIE_MIN_AGE_MS, + ); + for (const z of zombies) { + log.warn("ws zombie socket closed (open, no presence)", { + age_ms: now - (socketOpenedAt.get(z) ?? now), + }); + try { z.close(CLOSE_CODE_NO_PRESENCE, "no_presence"); } catch { /* already gone */ } + } + for (let i = 0; i < zombies.length; i++) metrics.zombieSocketsClosed.inc(); + for (const [pid, conn] of connections) { // Skip offline-leased entries: their WS is intentionally dead // during grace; the eviction timer handles their lifecycle. diff --git a/apps/broker/src/lease.ts b/apps/broker/src/lease.ts new file mode 100644 index 0000000..5338bfa --- /dev/null +++ b/apps/broker/src/lease.ts @@ -0,0 +1,135 @@ +/** + * Pure lease-lifecycle helpers for the broker's WS presence handling. + * + * Extracted from `index.ts` after the 2026-09-08 daemon-restart blackout + * (spec: .artifacts/specs/2026-09-08-daemon-restart-mesh-blackout.md, + * Addendum 5) so the decision logic is unit-testable without booting the + * broker. Every function here is synchronous and side-effect free; the + * caller (index.ts) applies the returned decision. + * + * Root cause these guard against (RC-A): the `ws.on("close")` handler used + * to check `conn.ws !== ws` BEFORE an `await savePeerState()` and mutate + * the lease AFTER it. Under DB latency the close of the *dead* daemon's + * socket resumed after the *new* daemon had already reattached, and put a + * healthy lease into grace (or evicted it outright). The new socket then + * lived on as a zombie — OPEN on both ends, no presence. The decision must + * be taken synchronously against the current state, which is what + * `decideCloseAction` encodes. + */ + +export type LeaseState = "online" | "offline"; + +/** Minimal structural view of a PeerConn the helpers need. */ +export interface LeaseConnView { + ws: WS; + leaseState: LeaseState; +} + +export type CloseAction = + /** No connections entry for this presence — already evicted. */ + | "no_conn" + /** The closing socket is not the lease's current socket (a reattach + * already swapped in a fresh one). Nothing to do. */ + | "ignore_replaced" + /** Lease was online → enter the grace window. */ + | "enter_grace" + /** Lease was already offline (second close in grace, odd state) → evict now. */ + | "evict_now"; + +/** + * Decide what a socket-close event means for a lease. MUST be evaluated + * synchronously at the moment the close event fires — never after an + * await — so the `conn.ws === ws` identity check and the lease-state read + * see the same instant. + */ +export function decideCloseAction( + conn: LeaseConnView | undefined, + closingWs: WS, +): CloseAction { + if (!conn) return "no_conn"; + if (conn.ws !== closingWs) return "ignore_replaced"; + return conn.leaseState === "online" ? "enter_grace" : "evict_now"; +} + +/** + * Should the grace-expiry timer actually evict? A reattach that landed + * after the timer was scheduled but whose clearTimeout raced the firing + * leaves the lease `online` — evicting then would kill a live session. + */ +export function shouldEvictOnGraceExpiry(conn: LeaseConnView | undefined): boolean { + if (!conn) return false; + return conn.leaseState === "offline"; +} + +export type PresenceRole = "control-plane" | "session" | "service"; + +export interface DedupConnView { + meshId: string; + sessionId: string; + peerRole: PresenceRole; +} + +/** + * Session-id dedup, scoped by role. + * + * The launch wrapper's member-level `hello` and the daemon's `session_hello` + * for the same Claude Code session both carry the same `sessionId` (the + * Claude session UUID). Before this fix each path kicked the other's + * presence ("hello dedup" ↔ "session_hello dedup"), producing a chain of + * evictions + forced WS closes on every relaunch (Addendum 2). A + * control-plane presence and a session presence for one session must + * coexist; only presences of the SAME role dedup each other. + */ +export function dedupTargets( + connections: Iterable<[K, DedupConnView]>, + hello: { meshId: string; sessionId: string }, + role: PresenceRole, +): K[] { + const out: K[] = []; + for (const [pid, c] of connections) { + if (c.meshId !== hello.meshId) continue; + if (c.sessionId !== hello.sessionId) continue; + if (c.peerRole !== role) continue; + out.push(pid); + } + return out; +} + +export interface ZombieSocketView { + readyState: number; + OPEN: number; +} + +/** + * Zombie sweeper selection: sockets that are OPEN, have been connected for + * at least `minAgeMs`, and are not referenced by any live lease. Such a + * socket cannot receive pushes (no presence → sendToPeer never finds it) + * and cannot be discovered, yet the client thinks it is connected because + * the `ws` library answers its pings. Closing it forces the client to + * reconnect and re-hello, which recreates the presence (self-heal). + * + * `minAgeMs` leaves room for a socket that is mid-hello. + */ +export function findZombieSockets( + clients: Iterable, + liveSockets: Set, + openedAt: (ws: WS) => number | undefined, + now: number, + minAgeMs: number, +): WS[] { + const out: WS[] = []; + for (const ws of clients) { + if (ws.readyState !== ws.OPEN) continue; + if (liveSockets.has(ws)) continue; + const t = openedAt(ws); + // Unknown open time = accepted before this code path existed; treat + // as old enough (the sweeper runs every 30s, so it has had time). + if (t !== undefined && now - t < minAgeMs) continue; + out.push(ws); + } + return out; +} + +/** Close codes the broker uses when it is the one ending a socket. */ +export const CLOSE_CODE_PRESENCE_EVICTED = 4004; +export const CLOSE_CODE_NO_PRESENCE = 4005; diff --git a/apps/broker/src/metrics.ts b/apps/broker/src/metrics.ts index e8c71ab..35f2155 100644 --- a/apps/broker/src/metrics.ts +++ b/apps/broker/src/metrics.ts @@ -106,6 +106,14 @@ export const metrics = { "broker_ttl_sweeps_total", "TTL sweeper runs completed", ), + zombieSocketsClosed: new Counter( + "broker_zombie_sockets_closed_total", + "Open WS sockets with no presence closed by the zombie sweeper (2026-09-08 blackout fix)", + ), + presenceRowsPruned: new Counter( + "broker_presence_rows_pruned_total", + "Old disconnected presence rows deleted by the retention sweep", + ), hookRequestsTotal: new Counter( "broker_hook_requests_total", "POST /hook/set-status requests received", diff --git a/apps/broker/tests/lease.test.ts b/apps/broker/tests/lease.test.ts new file mode 100644 index 0000000..9a20fa1 --- /dev/null +++ b/apps/broker/tests/lease.test.ts @@ -0,0 +1,128 @@ +/** + * Lease lifecycle decisions — regression tests for the 2026-09-08 + * daemon-restart mesh blackout (spec Addendum 5, RC-A + items 2-4). + * + * Pure-logic tests over `src/lease.ts`, same pattern as + * kick-control-plane-skip.test.ts: no broker boot, no DB. + */ + +import { describe, expect, test } from "vitest"; + +import { + decideCloseAction, + shouldEvictOnGraceExpiry, + dedupTargets, + findZombieSockets, + type DedupConnView, +} from "../src/lease"; + +const sockA = { id: "A" }; +const sockB = { id: "B" }; + +describe("decideCloseAction (RC-A: stale close after reattach)", () => { + test("no connections entry → no_conn", () => { + expect(decideCloseAction(undefined, sockA)).toBe("no_conn"); + }); + + test("online lease, its own socket closes → enter_grace", () => { + expect(decideCloseAction({ ws: sockA, leaseState: "online" }, sockA)).toBe("enter_grace"); + }); + + test("offline lease (already in grace), its socket closes again → evict_now", () => { + expect(decideCloseAction({ ws: sockA, leaseState: "offline" }, sockA)).toBe("evict_now"); + }); + + test("THE INCIDENT: dead daemon's socket closes late, lease already reattached to a new socket → ignore_replaced", () => { + // 19:28:08 reattach swapped conn.ws = B; 19:28:17 the close event of + // the old socket A finally ran. The old handler evicted/graced the + // healthy lease; the decision must ignore it. + const conn = { ws: sockB, leaseState: "online" as const }; + expect(decideCloseAction(conn, sockA)).toBe("ignore_replaced"); + }); + + test("stale close on an offline lease whose socket differs → still ignore_replaced (never evict on a foreign socket)", () => { + const conn = { ws: sockB, leaseState: "offline" as const }; + expect(decideCloseAction(conn, sockA)).toBe("ignore_replaced"); + }); +}); + +describe("shouldEvictOnGraceExpiry", () => { + test("offline lease → evict", () => { + expect(shouldEvictOnGraceExpiry({ ws: sockA, leaseState: "offline" })).toBe(true); + }); + test("lease reattached (online) before the timer fired → do NOT evict", () => { + expect(shouldEvictOnGraceExpiry({ ws: sockB, leaseState: "online" })).toBe(false); + }); + test("entry gone → nothing to evict", () => { + expect(shouldEvictOnGraceExpiry(undefined)).toBe(false); + }); +}); + +describe("dedupTargets (role-scoped session-id dedup)", () => { + const conns: Array<[string, DedupConnView]> = [ + ["cp1", { meshId: "m1", sessionId: "uuid-1", peerRole: "control-plane" }], + ["s1", { meshId: "m1", sessionId: "uuid-1", peerRole: "session" }], + ["s2", { meshId: "m1", sessionId: "uuid-2", peerRole: "session" }], + ["s1-other-mesh", { meshId: "m2", sessionId: "uuid-1", peerRole: "session" }], + ["svc", { meshId: "m1", sessionId: "uuid-1", peerRole: "service" }], + ]; + const hello = { meshId: "m1", sessionId: "uuid-1" }; + + test("member hello dedups only the control-plane presence with the same session_id", () => { + expect(dedupTargets(conns, hello, "control-plane")).toEqual(["cp1"]); + }); + + test("session_hello dedups only the session presence with the same session_id", () => { + expect(dedupTargets(conns, hello, "session")).toEqual(["s1"]); + }); + + test("never crosses meshes or session ids", () => { + expect(dedupTargets(conns, { meshId: "m1", sessionId: "uuid-9" }, "session")).toEqual([]); + expect(dedupTargets(conns, { meshId: "m3", sessionId: "uuid-1" }, "session")).toEqual([]); + }); + + test("Addendum 2 chain: wrapper hello + daemon session_hello for one session coexist", () => { + // The wrapper's control-plane hello must not kick the session, and the + // session_hello must not kick the wrapper. + const kickedByWrapper = dedupTargets(conns, hello, "control-plane"); + const kickedBySession = dedupTargets(conns, hello, "session"); + expect(kickedByWrapper).not.toContain("s1"); + expect(kickedBySession).not.toContain("cp1"); + }); +}); + +describe("findZombieSockets", () => { + const OPEN = 1; + const CLOSED = 3; + const mk = (readyState: number) => ({ readyState, OPEN }); + const now = 1_000_000; + + test("open socket with no lease, older than minAge → zombie", () => { + const z = mk(OPEN); + const opened = new Map([[z, now - 120_000]]); + expect(findZombieSockets([z], new Set(), (s) => opened.get(s), now, 60_000)).toEqual([z]); + }); + + test("open socket referenced by a lease → not a zombie", () => { + const live = mk(OPEN); + const opened = new Map([[live, now - 120_000]]); + expect(findZombieSockets([live], new Set([live]), (s) => opened.get(s), now, 60_000)).toEqual([]); + }); + + test("young socket (mid-hello) → skipped", () => { + const young = mk(OPEN); + const opened = new Map([[young, now - 5_000]]); + expect(findZombieSockets([young], new Set(), (s) => opened.get(s), now, 60_000)).toEqual([]); + }); + + test("non-open socket → skipped", () => { + const closed = mk(CLOSED); + const opened = new Map([[closed, now - 120_000]]); + expect(findZombieSockets([closed], new Set(), (s) => opened.get(s), now, 60_000)).toEqual([]); + }); + + test("socket with unknown open time (pre-existing) → treated as old", () => { + const z = mk(OPEN); + expect(findZombieSockets([z], new Set(), () => undefined, now, 60_000)).toEqual([z]); + }); +}); diff --git a/apps/broker/tests/presence-retention.test.ts b/apps/broker/tests/presence-retention.test.ts new file mode 100644 index 0000000..956fb09 --- /dev/null +++ b/apps/broker/tests/presence-retention.test.ts @@ -0,0 +1,91 @@ +/** + * Presence retention + orphan-message sweep — 2026-09-08 blackout RC-B + * (spec Addendum 5, items 5-6). Integration test against the test DB + * (same setup as dup-delivery.test.ts). + */ + +import { afterAll, afterEach, describe, expect, test } from "vitest"; +import { eq, and, isNull } from "drizzle-orm"; + +import { db } from "../src/db"; +import { messageQueue, presence } from "@turbostarter/db/schema/mesh"; +import { + connectPresence, + disconnectPresence, + listPeersInMesh, + sweepOldPresenceRows, + sweepOrphanMessages, + PRESENCE_RETENTION_MS, +} from "../src/broker"; +import { cleanupAllTestMeshes, setupTestMesh, type TestMesh } from "./helpers"; + +afterAll(async () => { + await cleanupAllTestMeshes(); +}); + +describe("sweepOldPresenceRows", () => { + let m: TestMesh; + afterEach(async () => { await m?.cleanup(); }); + + test("deletes rows disconnected > 30d ago, keeps recent + online rows", async () => { + m = await setupTestMesh("presence-retention"); + const mk = () => connectPresence({ + memberId: m.peerA.memberId, sessionId: `s-${Math.random()}`, pid: 1, cwd: "/tmp", + }); + const online = await mk(); + const recentlyGone = await mk(); + const longGone1 = await mk(); + const longGone2 = await mk(); + await disconnectPresence(recentlyGone); + await disconnectPresence(longGone1); + await disconnectPresence(longGone2); + const old = new Date(Date.now() - PRESENCE_RETENTION_MS - 60_000); + await db.update(presence).set({ disconnectedAt: old }).where(eq(presence.id, longGone1)); + await db.update(presence).set({ disconnectedAt: old }).where(eq(presence.id, longGone2)); + + const deleted = await sweepOldPresenceRows(); + expect(deleted).toBeGreaterThanOrEqual(2); + + const remaining = await db.select({ id: presence.id }).from(presence) + .where(eq(presence.memberId, m.peerA.memberId)); + const ids = remaining.map((r) => r.id).sort(); + expect(ids).toEqual([online, recentlyGone].sort()); + + // Online row still discoverable (the partial indexes must not change results). + const peers = await listPeersInMesh(m.meshId); + expect(peers.map((p) => p.sessionId)).toHaveLength(1); + }); +}); + +describe("sweepOrphanMessages (Date param regression)", () => { + let m: TestMesh; + afterEach(async () => { await m?.cleanup(); }); + + test("deletes undelivered rows older than 7d and does not throw on the Date cutoff", async () => { + m = await setupTestMesh("orphan-sweep"); + const base = { + meshId: m.meshId, senderMemberId: m.peerA.memberId, targetSpec: m.peerB.pubkey, + nonce: "n", ciphertext: "c", + }; + const [fresh] = await db.insert(messageQueue).values({ ...base }).returning({ id: messageQueue.id }); + const [stale] = await db.insert(messageQueue).values({ + ...base, createdAt: new Date(Date.now() - 8 * 24 * 60 * 60 * 1000), + }).returning({ id: messageQueue.id }); + const [staleDelivered] = await db.insert(messageQueue).values({ + ...base, createdAt: new Date(Date.now() - 8 * 24 * 60 * 60 * 1000), deliveredAt: new Date(), + }).returning({ id: messageQueue.id }); + + const n = await sweepOrphanMessages(); + expect(n).toBeGreaterThanOrEqual(1); + + const left = await db.select({ id: messageQueue.id }).from(messageQueue) + .where(eq(messageQueue.meshId, m.meshId)); + const ids = left.map((r) => r.id).sort(); + expect(ids).toEqual([fresh!.id, staleDelivered!.id].sort()); + expect(ids).not.toContain(stale!.id); + // Sanity: undelivered fresh row still pending. + const pending = await db.select({ id: messageQueue.id }).from(messageQueue) + .where(and(eq(messageQueue.meshId, m.meshId), isNull(messageQueue.deliveredAt))); + expect(pending.map((r) => r.id)).toEqual([fresh!.id]); + }); +}); diff --git a/packages/db/migrations/0030_presence_indexes.sql b/packages/db/migrations/0030_presence_indexes.sql new file mode 100644 index 0000000..dc4e518 --- /dev/null +++ b/packages/db/migrations/0030_presence_indexes.sql @@ -0,0 +1,38 @@ +-- 2026-09-08 daemon-restart mesh blackout, RC-B (spec: +-- .artifacts/specs/2026-09-08-daemon-restart-mesh-blackout.md, Addendum 5). +-- +-- mesh.presence had 2.49 M rows and no index beyond the PK. Every hot +-- query filters on `disconnected_at IS NULL` (13 rows online) and was +-- full-scanning the table: +-- * sweepStalePresences — every 30 s: WHERE disconnected_at IS NULL AND last_ping_at < cutoff +-- * listPeersInMesh — JOIN member … WHERE disconnected_at IS NULL +-- * handleHookSetStatus — WHERE cwd = ? AND disconnected_at IS NULL +-- * heartbeat / restorePresence / disconnectPresence — by PK (fine) +-- Under a reconnect storm those scans pushed hello latency to ~5 s. +-- +-- Partial indexes keep only the online rows (tiny) so each of those +-- queries is an index scan regardless of table size. The plain index on +-- disconnected_at serves the new retention sweep (delete rows disconnected +-- > 30 d ago in batches). +-- +-- Plain CREATE INDEX (not CONCURRENTLY): the migration runner wraps each +-- file in a transaction, and CONCURRENTLY cannot run inside one. On the +-- 2.5 M-row prod table these three builds take a few seconds each and +-- block writes to presence for that window; heartbeats retry on the next +-- pong, so this is acceptable for a one-time build. + +CREATE INDEX IF NOT EXISTS presence_online_member_idx + ON mesh.presence (member_id) + WHERE disconnected_at IS NULL; +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS presence_online_last_ping_idx + ON mesh.presence (last_ping_at) + WHERE disconnected_at IS NULL; +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS presence_online_cwd_idx + ON mesh.presence (cwd) + WHERE disconnected_at IS NULL; +--> statement-breakpoint +CREATE INDEX IF NOT EXISTS presence_disconnected_at_idx + ON mesh.presence (disconnected_at) + WHERE disconnected_at IS NOT NULL;