From dd6d82a094b05f95ac0efd751c185d1a1c0f4cb2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alejandro=20Guti=C3=A9rrez?= Date: Tue, 8 Sep 2026 23:17:37 +0100 Subject: [PATCH] fix(broker): DB stale sweeper reconciled with in-memory leases; flipped live presences get their socket closed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addendum 5 follow-up. sweepStalePresences() flipped presence.disconnected_at purely on last_ping_at age, with no knowledge of the in-memory lease, and a flip never closed anything — so a session could be hidden from list_peers (DB-backed) while pushes still reached it (memory-backed) and its daemon was never told to reconnect: the half-visible ghost seen in prod at 2026-09-08 22:05Z. - lease.ts: liveLeasePresenceIds() (online + OPEN + pong within 75 s) and reconcileFlippedPresences() (flipped ids that still hold an ONLINE lease); STALE_PONG_THRESHOLD_MS becomes the single shared constant. - broker.ts: sweepStalePresences(keep) skips the live set and returns the ids it flipped; setStaleSweepReconciler() lets index.ts (owner of `connections`) feed the keep-set and react to flips without a circular import. - index.ts: registers the reconciler before startSweepers(); every flipped presence that still has an online lease is evicted via evictPresenceFully(), which closes the socket (4004) so the daemon reconnects and re-hellos. Offline (in-grace) leases are left to their grace timer on purpose — the 90 s cutoff can land inside the window. Tests: 4 pure cases in tests/lease.test.ts (keep-set selection incl. the 75 s edge, the ghost case, in-grace exemption, empty). Broker tsc clean; suite 79 passed, 6 files fail only for missing DATABASE_URL / bun broker (pre-existing, environment). Cambio: barrido DB de presencias reconciliado con las leases en memoria HARD RULES touched: ninguna (broker only, no schema) Blast radius: broker prod (all meshes) · revert: trivial (1 commit) Smoke: 🟡 tests puros + tsc; verificación en vivo post-deploy: last_ping_at avanza ≥10 min tras un reattach y ningún 'stale sweep flipped' en logs Veredicto: safe — deploy junto al resto de fix/daemon-restart-mesh-blackout-1.37.1 Co-Authored-By: Claude Fable 5.1 --- apps/broker/src/broker.ts | 49 ++++++++++++++++++------ apps/broker/src/index.ts | 26 ++++++++++++- apps/broker/src/lease.ts | 66 +++++++++++++++++++++++++++++++++ apps/broker/tests/lease.test.ts | 42 +++++++++++++++++++++ 4 files changed, 171 insertions(+), 12 deletions(-) diff --git a/apps/broker/src/broker.ts b/apps/broker/src/broker.ts index a7e75e6..eaf130a 100644 --- a/apps/broker/src/broker.ts +++ b/apps/broker/src/broker.ts @@ -25,6 +25,7 @@ import { isNotNull, isNull, lt, + notInArray, or, sql, } from "drizzle-orm"; @@ -285,18 +286,43 @@ export async function refreshQueueDepth(): Promise { /** * Sweep stale presences: mark as disconnected if last_ping_at is older * than 90s (3 missed pings at the 30s interval = dead session). + * + * 2026-09-08 (Addendum 5 follow-up): the DB column is only a projection of + * the in-memory lease. `keep` = presence ids whose lease is online with an + * OPEN, recently-ponged socket — those are never flipped, whatever the + * column says (a lagging or failed heartbeat write must not hide a live + * session from `list_peers`). Returns the ids that WERE flipped so the + * caller can reconcile them against memory and close any socket that is + * still open (otherwise the client is never told and lives on as a ghost). */ -export async function sweepStalePresences(): Promise { +export async function sweepStalePresences(keep: ReadonlySet = new Set()): Promise { const cutoff = new Date(Date.now() - 90_000); // 3 missed pings - await db + const conds = [ + isNull(presence.disconnectedAt), + lt(presence.lastPingAt, cutoff), + ]; + if (keep.size > 0) conds.push(notInArray(presence.id, [...keep])); + const flipped = await db .update(presence) .set({ disconnectedAt: new Date() }) - .where( - and( - isNull(presence.disconnectedAt), - lt(presence.lastPingAt, cutoff), - ), - ); + .where(and(...conds)) + .returning({ id: presence.id }); + return flipped.map((r) => r.id); +} + +/** + * Hook the stale sweeper into the in-memory lease map (lives in index.ts, + * which owns `connections`; broker.ts must not import it). `keep()` is + * evaluated right before each sweep; `onFlipped` receives the ids the + * sweep disconnected, if any. + */ +export interface StaleSweepReconciler { + keep: () => ReadonlySet; + onFlipped: (ids: string[]) => void; +} +let staleSweepReconciler: StaleSweepReconciler | null = null; +export function setStaleSweepReconciler(r: StaleSweepReconciler | null): void { + staleSweepReconciler = r; } /** @@ -2651,9 +2677,10 @@ export function startSweepers(): void { ); }, PENDING_SWEEP_INTERVAL_MS); staleTimer = setInterval(() => { - sweepStalePresences().catch((e) => - console.error("[broker] stale presence sweep:", e), - ); + const r = staleSweepReconciler; + sweepStalePresences(r ? r.keep() : undefined) + .then((flipped) => { if (r && flipped.length > 0) r.onFlipped(flipped); }) + .catch((e) => console.error("[broker] stale presence sweep:", e)); }, 30_000); claimSweepTimer = setInterval(() => { sweepExpiredClaims() diff --git a/apps/broker/src/index.ts b/apps/broker/src/index.ts index 99d9c43..3588a53 100644 --- a/apps/broker/src/index.ts +++ b/apps/broker/src/index.ts @@ -42,6 +42,7 @@ import { handleHookSetStatus, heartbeat, restorePresence, + setStaleSweepReconciler, insertFileKeys, joinGroup, joinMesh, @@ -129,6 +130,9 @@ import { dedupTargets, findZombieSockets, reattachLease, + liveLeasePresenceIds, + reconcileFlippedPresences, + STALE_PONG_THRESHOLD_MS, CLOSE_CODE_PRESENCE_EVICTED, CLOSE_CODE_NO_PRESENCE, } from "./lease"; @@ -5742,7 +5746,8 @@ async function main(): Promise { // ping interval) are considered half-dead — kernel hasn't yet RST'd // the socket but no application traffic is flowing. Force-terminate // them to fire the close handler and free the connection slot. - const STALE_PONG_THRESHOLD_MS = 75_000; + // STALE_PONG_THRESHOLD_MS (75 s) is shared with the DB stale-sweep + // reconciler and lives in lease.ts. const pingInterval = setInterval(() => { const now = Date.now(); @@ -5806,6 +5811,25 @@ async function main(): Promise { }, 30_000); queueDepthTimer.unref(); + // 2026-09-08 (Addendum 5 follow-up): the DB stale sweeper must not flip + // a presence whose lease is alive in memory, and whatever it does flip + // that still holds an online lease must be evicted (socket closed) so + // the owning daemon reconnects instead of becoming a half-visible ghost. + setStaleSweepReconciler({ + keep: () => liveLeasePresenceIds(connections, Date.now(), STALE_PONG_THRESHOLD_MS), + onFlipped: (ids) => { + const ghosts = reconcileFlippedPresences(ids, (pid) => connections.get(pid)); + for (const pid of ghosts) { + const conn = connections.get(pid); + log.warn("stale sweep flipped a presence with a live lease — evicting so the client reconnects", { + presence_id: pid, + session_pubkey: conn?.sessionPubkey?.slice(0, 12) ?? "(member-WS)", + last_pong_ago_ms: conn ? Date.now() - conn.lastPongAt : -1, + }); + void evictPresenceFully(pid); + } + }, + }); startSweepers(); startDbHealth(); serviceManager.startHealthChecks(); diff --git a/apps/broker/src/lease.ts b/apps/broker/src/lease.ts index 1cd2c34..b1bb5e7 100644 --- a/apps/broker/src/lease.ts +++ b/apps/broker/src/lease.ts @@ -177,3 +177,69 @@ export function reattachLease( conn.leaseUntil = 0; return wasState; } + +// ── DB stale-sweep reconciliation (2026-09-08, Addendum 5 follow-up) ────── +// +// `sweepStalePresences` (broker.ts) flips `presence.disconnected_at` purely +// on `last_ping_at` age. That column is bumped by `heartbeat()` on every +// pong, so a live lease normally never goes stale — but the DB write can +// lag or fail while the in-memory lease is perfectly healthy, and the +// reverse can happen too: the row is flipped while a socket is still OPEN +// and its owner (the daemon) is never told. Both halves are closed here: +// 1. `liveLeasePresenceIds` = the set the sweeper must NOT flip +// (online lease, OPEN socket, pong within the stale-pong window). +// 2. `reconcileFlippedPresences` = of the ids the sweeper DID flip, the +// ones that still hold an online in-memory lease. The caller evicts +// them (which closes the socket) so the client reconnects and +// re-hellos instead of living on as a half-visible ghost. +// Offline-leased entries are left to their grace timer on purpose: the +// sweep cutoff (90 s since last ping) can land inside the grace window, +// and evicting there would shorten the grace the lease model promises. + +/** Broker-side stale-pong window: 2.5x the 30 s ping cadence. */ +export const STALE_PONG_THRESHOLD_MS = 75_000; + +export interface LivenessConnView + extends LeaseConnView { + lastPongAt: number; +} + +/** + * Presence ids whose lease is online, whose socket is OPEN and whose last + * pong is within `stalePongMs`. The DB stale sweeper must skip these — the + * in-memory lease is the source of truth for liveness, the DB column is a + * projection of it. + */ +export function liveLeasePresenceIds( + conns: Iterable<[string, LivenessConnView]>, + now: number, + stalePongMs: number = STALE_PONG_THRESHOLD_MS, +): Set { + const keep = new Set(); + for (const [pid, conn] of conns) { + if (conn.leaseState !== "online") continue; + if (conn.ws.readyState !== conn.ws.OPEN) continue; + if (now - conn.lastPongAt > stalePongMs) continue; + keep.add(pid); + } + return keep; +} + +/** + * Of the presence ids the DB sweeper flipped to disconnected, return those + * that still have an ONLINE in-memory lease. Each one is a DB/memory + * split-brain: discovery (`list_peers`, DB-backed) already hides it while + * pushes (memory-backed) still reach it, and nothing tells the client. + * The caller must evict them (closing the socket) so the client reconnects. + */ +export function reconcileFlippedPresences( + flipped: Iterable, + lookup: (presenceId: string) => LeaseConnView | undefined, +): string[] { + const out: string[] = []; + for (const pid of flipped) { + const conn = lookup(pid); + if (conn?.leaseState === "online") out.push(pid); + } + return out; +} diff --git a/apps/broker/tests/lease.test.ts b/apps/broker/tests/lease.test.ts index dc62e77..19da72a 100644 --- a/apps/broker/tests/lease.test.ts +++ b/apps/broker/tests/lease.test.ts @@ -14,6 +14,9 @@ import { dedupTargets, findZombieSockets, reattachLease, + liveLeasePresenceIds, + reconcileFlippedPresences, + STALE_PONG_THRESHOLD_MS, type DedupConnView, } from "../src/lease"; @@ -183,3 +186,42 @@ describe("reattachLease (RC-C ordering: swap before close)", () => { expect(conn.leaseState).toBe("online"); }); }); + +describe("DB stale-sweep reconciliation (Addendum 5 follow-up: sweeper vs in-memory lease)", () => { + const OPEN = 1, CLOSED = 3; + const sock = (readyState: number) => ({ readyState, OPEN }); + const now = 1_000_000; + + test("liveLeasePresenceIds keeps only online + OPEN + recently-ponged leases", () => { + const conns: Array<[string, { ws: { readyState: number; OPEN: number }; leaseState: "online" | "offline"; lastPongAt: number }]> = [ + ["alive", { ws: sock(OPEN), leaseState: "online", lastPongAt: now - 10_000 }], + ["in-grace", { ws: sock(CLOSED), leaseState: "offline", lastPongAt: now - 10_000 }], + ["half-dead", { ws: sock(OPEN), leaseState: "online", lastPongAt: now - STALE_PONG_THRESHOLD_MS - 1 }], + ["closed-sock",{ ws: sock(CLOSED), leaseState: "online", lastPongAt: now - 1_000 }], + ["edge", { ws: sock(OPEN), leaseState: "online", lastPongAt: now - STALE_PONG_THRESHOLD_MS }], + ]; + const keep = liveLeasePresenceIds(conns, now); + expect([...keep].sort()).toEqual(["alive", "edge"]); + }); + + test("THE GHOST: sweeper flips a row whose lease is online → reconcile returns it so the caller evicts (closes the socket)", () => { + // Prod 2026-09-08 22:05Z: presence rows flipped to disconnected while + // the daemon's socket stayed OPEN and nobody told it. list_peers hid + // the session; pushes still reached it; it never reconnected. + const conns = new Map([ + ["ghost", { ws: {}, leaseState: "online" }], + ["graced", { ws: {}, leaseState: "offline" }], + ]); + const out = reconcileFlippedPresences(["ghost", "graced", "already-gone"], (id) => conns.get(id)); + expect(out).toEqual(["ghost"]); + }); + + test("offline (in-grace) leases are left to their grace timer, never evicted by the sweeper", () => { + const conns = new Map([["g", { ws: {}, leaseState: "offline" as const }]]); + expect(reconcileFlippedPresences(["g"], (id) => conns.get(id))).toEqual([]); + }); + + test("nothing flipped → nothing to reconcile", () => { + expect(reconcileFlippedPresences([], () => undefined)).toEqual([]); + }); +});