fix(broker): DB stale sweeper reconciled with in-memory leases; flipped live presences get their socket closed

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 <noreply@anthropic.com>

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<void> {
 /**
  * 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<void> {
+export async function sweepStalePresences(keep: ReadonlySet<string> = new Set()): Promise<string[]> {
   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<string>;
+  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()

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<void> {
   // 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<void> {
   }, 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();

apps/broker/src/lease.ts

@@ -177,3 +177,69 @@ export function reattachLease<WS>(
   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<WS extends { readyState: number; OPEN: number }>
+  extends LeaseConnView<WS> {
+  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<WS extends { readyState: number; OPEN: number }>(
+  conns: Iterable<[string, LivenessConnView<WS>]>,
+  now: number,
+  stalePongMs: number = STALE_PONG_THRESHOLD_MS,
+): Set<string> {
+  const keep = new Set<string>();
+  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<WS>(
+  flipped: Iterable<string>,
+  lookup: (presenceId: string) => LeaseConnView<WS> | undefined,
+): string[] {
+  const out: string[] = [];
+  for (const pid of flipped) {
+    const conn = lookup(pid);
+    if (conn?.leaseState === "online") out.push(pid);
+  }
+  return out;
+}

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<string, { ws: object; leaseState: "online" | "offline" }>([
+      ["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([]);
+  });
+});