alezmad/claudemesh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(broker): stop evicting live reattached leases; close evicted sockets; presence indexes + retention

Browse Source 
Root causes of the 2026-09-08 mesh blackout on the broker side (spec
.artifacts/specs/2026-09-08-daemon-restart-mesh-blackout.md, Addenda 5-6):

- ws.on("close") checked conn.ws !== ws BEFORE an await and mutated the
  lease AFTER it; and under Bun (prod runtime) the ws shim emits the replaced
  socket's close event synchronously inside close(). Either way the close of
  the DEAD daemon's socket landed on the NEW daemon's reattached lease: grace
  started, 90 s eviction timer armed, healthy session evicted while its
  socket stayed open — a ghost that peer list and sends could not see.
  Fix: decision taken synchronously (src/lease.ts decideCloseAction), lease
  swapped onto the new socket BEFORE the old one is closed, dedup drops the
  entry before closing, grace timer re-checks leaseState before evicting.
- evictPresenceFully never closed a still-open socket → zombie. Now closes
  it (4004 presence_evicted) so the client re-hellos; a 30 s zombie sweeper
  closes any OPEN socket with no presence (4005 no_presence).
- Session-id dedup scoped by role: the launch wrapper's control-plane hello
  and the daemon's session_hello share the Claude session UUID and used to
  kick each other on every relaunch.
- mesh.presence had 2,491,950 rows and no index beyond the PK; every hot
  query full-scanned it, pushing hello latency to ~5 s under a reconnect
  storm (which tripped the daemon's 5 s ack timeout). Migration 0030 adds
  partial indexes on the online rows; hourly retention sweep deletes rows
  disconnected > 30 d (batched 20k). Hello handlers log latency_ms.
- sweepOrphanMessages rewritten with the query builder (the raw Date param
  threw on every hourly run).

Tests: tests/lease.test.ts (pure decision helpers incl. the incident
sequence), tests/presence-retention.test.ts (retention + orphan sweep
against a real DB). Verified locally: migrations 0000-0030 apply through
the runtime runner on a fresh pgvector DB; SIGKILL/respawn drills against
this broker show reattach with 0 evictions.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Alejandro Gutiérrez
2026-09-08 22:22:57 +01:00
parent 1e1fbacddb
commit 0bf7ee9c84
7 changed files with 627 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
apps/broker/src/broker.ts
View File

@@ -311,15 +311,62 @@ export async function sweepStalePresences(): Promise<void> {
*/ */
export async function sweepOrphanMessages(): Promise<number> { export async function sweepOrphanMessages(): Promise<number> {
const cutoff = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000); const cutoff = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000);
const result = await db.execute(sql` // 2026-09-08: rewritten with the query builder. The previous raw
DELETE FROM mesh.message_queue // `sql\`… created_at < ${cutoff}\`` threw on every hourly run
WHERE delivered_at IS NULL // (`DrizzleQueryError … The "string" argument must be of type string …
AND created_at < ${cutoff} // Received an instance of Date`) — the Date param was never serialized
RETURNING id // for the raw path, so orphan rows were never swept. The builder path
`); // (same shape as sweepPendingStatuses) serializes timestamps correctly.
const rows = (result as unknown as { rows?: unknown[]; length?: number }).rows ?? result; const deleted = await db
const count = Array.isArray(rows) ? rows.length : 0; .delete(messageQueue)
return count; .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<number> {
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. */ /** 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}`); }) .then((n) => { if (n > 0) console.log(`[broker] orphan msgs swept: ${n}`); })
.catch((e) => console.error("[broker] orphan msg sweep:", e)); .catch((e) => console.error("[broker] orphan msg sweep:", e));
}, 60 * 60_000).unref(); }, 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. */ /** Stop background sweepers and mark all active presences disconnected. */

217
apps/broker/src/index.ts
View File

@@ -123,6 +123,25 @@ import { canonicalInvite, canonicalInviteV2, claimInviteV2Core as _claimInviteV2
const claimInviteV2Core = _claimInviteV2Core; const claimInviteV2Core = _claimInviteV2Core;
import { handleWebhook } from "./webhooks"; import { handleWebhook } from "./webhooks";
import { audit, loadLastHashes, ensureAuditLogTable, verifyChain, queryAuditLog } from "./audit"; 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<WebSocket, number>();
/** 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 PORT = env.BROKER_PORT;
const WS_PATH = "/ws"; 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 * Idempotent: re-entering when the connections entry is already gone
* is a no-op. * is a no-op.
*/ */
async function evictPresenceFully(presenceId: string): Promise<void> { async function evictPresenceFully(presenceId: string, reason = "unspecified"): Promise<void> {
const conn = connections.get(presenceId); const conn = connections.get(presenceId);
if (!conn) return; // already evicted if (!conn) return; // already evicted
if (conn.evictionTimer) { if (conn.evictionTimer) {
@@ -593,6 +612,17 @@ async function evictPresenceFully(presenceId: string): Promise<void> {
connections.delete(presenceId); connections.delete(presenceId);
decMeshCount(conn.meshId); 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 = { const leaveMsg: WSPushMessage = {
type: "push", type: "push",
subtype: "system", subtype: "system",
@@ -656,7 +686,7 @@ async function evictPresenceFully(presenceId: string): Promise<void> {
log.info("clock auto-paused (mesh empty)", { mesh_id: conn.meshId }); 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( async function maybePushQueuedMessages(
@@ -1824,6 +1854,7 @@ async function handleHello(
* augmentation. The session was never visibly absent from peers. */ * augmentation. The session was never visibly absent from peers. */
silent?: boolean; silent?: boolean;
} | null> { } | null> {
const helloStartedAt = Date.now();
// Validate sessionPubkey shape — it becomes a routable identity in // Validate sessionPubkey shape — it becomes a routable identity in
// listPeers/drainForMember, so arbitrary strings let a client claim // listPeers/drainForMember, so arbitrary strings let a client claim
// nonsense pubkeys. Required-if-present: empty is allowed (falls back // 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, // Session-id dedup: if this session_id already has an active presence
// disconnect the ghost. Happens when a client reconnects after a // OF THE SAME ROLE, disconnect the ghost. Happens when a client
// network blip or broker restart before the 90s stale sweeper runs. // reconnects after a network blip or broker restart before the 90s
// One Claude Code instance = one session_id = one presence, always. // stale sweeper runs. One control-plane connection per session_id.
for (const [oldPid, oldConn] of connections) { //
if (oldConn.meshId === hello.meshId && oldConn.sessionId === hello.sessionId) { // 2026-09-08 (Addendum 2/5, item 4): scoped by role. The launch
log.info("hello dedup", { old_presence: oldPid, session_id: hello.sessionId }); // wrapper's member-level `hello` carries the Claude session UUID as
try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } // session_id — the same value the daemon's `session_hello` uses — so an
dropConnection(oldPid); // unscoped dedup kicked the live session presence, whose re-hello then
void disconnectPresence(oldPid); // 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({ const presenceId = await connectPresence({
@@ -2033,6 +2074,10 @@ async function handleHello(
presence_id: presenceId, presence_id: presenceId,
session_id: hello.sessionId, session_id: hello.sessionId,
restored: !!saved, 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 // Drain any queued messages in the background. The hello_ack is
// sent by the CALLER after it assigns presenceId — sending it here // sent by the CALLER after it assigns presenceId — sending it here
@@ -2080,6 +2125,7 @@ async function handleSessionHello(
* absent from peers. */ * absent from peers. */
silent?: boolean; silent?: boolean;
} | null> { } | null> {
const helloStartedAt = Date.now();
// Shape checks. The crypto helpers also enforce these but bailing // Shape checks. The crypto helpers also enforce these but bailing
// early gives a clearer error code on the wire. // early gives a clearer error code on the wire.
if (!/^[0-9a-f]{64}$/.test(hello.sessionPubkey ?? "")) { 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 // presence per session pubkey" invariant — it kills the same-name
// ghost that used to win queued-DM claim races. // ghost that used to win queued-DM claim races.
const wasState = oldConn.leaseState; const wasState = oldConn.leaseState;
if (oldConn.evictionTimer) { const replacedWs = oldConn.ws !== ws ? oldConn.ws : null;
clearTimeout(oldConn.evictionTimer); // ORDER MATTERS (2026-09-08, RC-C): swap the lease onto the new socket
oldConn.evictionTimer = null; // and mark it online BEFORE closing the old socket. Under Bun (the
} // production runtime) the `ws` shim emits the old socket's `close`
if (oldConn.ws !== ws) { // event SYNCHRONOUSLY inside `close()` — verified 2026-09-08 with a
try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } // 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.ws = ws;
oldConn.leaseState = "online"; oldConn.leaseState = "online";
oldConn.leaseUntil = 0; oldConn.leaseUntil = 0;
oldConn.lastPongAt = Date.now(); 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. // Refresh mutable fields from the new hello.
oldConn.sessionId = hello.sessionId; oldConn.sessionId = hello.sessionId;
oldConn.cwd = hello.cwd; oldConn.cwd = hello.cwd;
@@ -2259,15 +2321,19 @@ async function handleSessionHello(
}; };
} }
// Session-id dedup: if the same session_id is already connected, kick // Session-id dedup: if the same session_id already has a SESSION-role
// the ghost. Reconnect after a network blip lands here cleanly. // presence (different pubkey — e.g. keypair rotated), kick the ghost.
for (const [oldPid, oldConn] of connections) { // Reconnect after a network blip lands here cleanly. Scoped by role
if (oldConn.meshId === hello.meshId && oldConn.sessionId === hello.sessionId) { // (see the matching comment in handleHello): the wrapper's
log.info("session_hello dedup", { old_presence: oldPid, session_id: hello.sessionId }); // control-plane presence for this session_id is left alone.
try { oldConn.ws.close(1000, "session_replaced"); } catch { /* already dead */ } for (const oldPid of dedupTargets(connections, hello, "session")) {
dropConnection(oldPid); const oldConn = connections.get(oldPid);
void disconnectPresence(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({ const presenceId = await connectPresence({
@@ -2319,6 +2385,7 @@ async function handleSessionHello(
presence_id: presenceId, presence_id: presenceId,
session_id: hello.sessionId, session_id: hello.sessionId,
session_pubkey: hello.sessionPubkey.slice(0, 12), session_pubkey: hello.sessionPubkey.slice(0, 12),
latency_ms: Date.now() - helloStartedAt,
}); });
// Drain any DMs queued for this session pubkey (or the parent member). // Drain any DMs queued for this session pubkey (or the parent member).
void maybePushQueuedMessages(presenceId); void maybePushQueuedMessages(presenceId);
@@ -2679,6 +2746,7 @@ async function handleSend(
function handleConnection(ws: WebSocket): void { function handleConnection(ws: WebSocket): void {
metrics.connectionsTotal.inc(); metrics.connectionsTotal.inc();
socketOpenedAt.set(ws, Date.now());
let presenceId: string | null = null; let presenceId: string | null = null;
ws.on("message", async (raw) => { ws.on("message", async (raw) => {
try { try {
@@ -5432,32 +5500,46 @@ function handleConnection(ws: WebSocket): void {
}); });
} }
}); });
ws.on("close", async () => { ws.on("close", (code, reasonBuf) => {
if (!presenceId) return; 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; const pid = presenceId;
if (conn.leaseState === "online") { const conn = connections.get(pid);
conn.leaseState = "offline"; const reason = reasonBuf.toString("utf8");
conn.leaseUntil = Date.now() + GRACE_MS;
conn.evictionTimer = setTimeout(() => { // 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 }); log.info("lease grace expired — evicting", { presence_id: pid });
void evictPresenceFully(pid); void evictPresenceFully(pid, "grace_expired");
}, GRACE_MS); }, GRACE_MS);
log.info("ws close — lease grace started", { log.info("ws close — lease grace started", {
presence_id: pid, presence_id: pid,
@@ -5466,9 +5548,9 @@ function handleConnection(ws: WebSocket): void {
return; return;
} }
// Not online (already in grace from an earlier close, or odd state). // "evict_now": already offline (second close while in grace, or odd
// Run full eviction immediately. // state). Run full eviction immediately.
await evictPresenceFully(pid); void evictPresenceFully(pid, "close_while_offline");
}); });
ws.on("error", (err) => { ws.on("error", (err) => {
log.warn("ws error", { error: err.message }); log.warn("ws error", { error: err.message });
@@ -5678,6 +5760,31 @@ async function main(): Promise<void> {
const STALE_PONG_THRESHOLD_MS = 75_000; const STALE_PONG_THRESHOLD_MS = 75_000;
const pingInterval = setInterval(() => { const pingInterval = setInterval(() => {
const now = Date.now(); 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<WebSocket>();
for (const [, conn] of connections) liveSockets.add(conn.ws);
const zombies = findZombieSockets(
wss.clients as Set<WebSocket>,
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) { for (const [pid, conn] of connections) {
// Skip offline-leased entries: their WS is intentionally dead // Skip offline-leased entries: their WS is intentionally dead
// during grace; the eviction timer handles their lifecycle. // during grace; the eviction timer handles their lifecycle.

135
apps/broker/src/lease.ts Normal file
View File

@@ -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: 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<WS>(
conn: LeaseConnView<WS> | 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<WS>(conn: LeaseConnView<WS> | 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<K>(
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<WS extends ZombieSocketView>(
clients: Iterable<WS>,
liveSockets: Set<WS>,
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;

8
apps/broker/src/metrics.ts
View File

@@ -106,6 +106,14 @@ export const metrics = {
"broker_ttl_sweeps_total", "broker_ttl_sweeps_total",
"TTL sweeper runs completed", "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( hookRequestsTotal: new Counter(
"broker_hook_requests_total", "broker_hook_requests_total",
"POST /hook/set-status requests received", "POST /hook/set-status requests received",

128
apps/broker/tests/lease.test.ts Normal file
View File

@@ -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<object, number>([[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<object, number>([[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<object, number>([[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<object, number>([[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]);
});
});

91
apps/broker/tests/presence-retention.test.ts Normal file
View File

@@ -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]);
});
});

38
packages/db/migrations/0030_presence_indexes.sql Normal file
View File

@@ -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;