fix(bluebubbles): always run catchup on startup; advance cursor to page boundary on truncation

Lobster · claude · Lobster · commit 5ac997f733d8 · 2026-04-14T15:48:38.000-07:00
Addresses two more Codex P1 findings on PR #66760: - Run catchup even when cursor age is under MIN_INTERVAL Catchup is the *only* mechanism that recovers messages dropped during the gateway-down window, and it runs once per gateway startup. The 30s MIN_INTERVAL_MS gate was protecting against ~nothing (a few extra BB queries on rolling restarts) at the cost of permanent message loss when restarts happened within that window. A real failure: t0 startup, cursor=t0; t0+10s gateway down, webhook ECONNREFUSED at t0+15s; t0+20s restart skips catchup entirely. Gate removed; bounded by perRunLimit/maxAge + dedupe-protected so the cost of always running is capped. - Keep cursor behind unfetched pages when limit is hit Previously a long outage with >perRunLimit messages would fetch the oldest perRunLimit (sort:ASC), process them, and then advance the cursor to nowMs — permanently skipping the unfetched newer tail. Now we track the latest fetched timestamp regardless of fate, and on truncation (fetchedCount === perRunLimit) the cursor advances only to that page boundary so the next gateway startup picks up the rest. Updated the existing perRunLimit warn to reflect the new recoverable-on-next-startup semantics. Tests: replaced obsolete "skips rapid restart" with "runs catchup even on rapid restarts"; added "advances cursor only to last fetched ts when result is truncated". BB suite 410/410. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
diff --git a/extensions/bluebubbles/src/catchup.test.ts b/extensions/bluebubbles/src/catchup.test.ts
@@ -254,15 +254,66 @@ describe("runBlueBubblesCatchup", () => {
     expect(called.proc).toBe(0);
   });
 
-  it("skips a rapid second run within MIN_INTERVAL_MS", async () => {
+  it("runs catchup even on rapid restarts (no min-interval gate)", async () => {
+    // Catchup runs once per gateway startup, so a quick restart MUST run
+    // it again — otherwise messages dropped between the two startups
+    // (gateway down → BB ECONNREFUSED → gateway up <30s later) are lost
+    // permanently. Bounded by perRunLimit/maxAge + dedupe-protected.
     const now = 10_000;
-    await saveBlueBubblesCatchupCursor("test-account", now - 5_000); // 5s ago
+    await saveBlueBubblesCatchupCursor("test-account", now - 5_000);
+    let fetched = false;
     const summary = await runBlueBubblesCatchup(makeTarget(), {
       now: () => now,
-      fetchMessages: async () => ({ resolved: true, messages: [] }),
+      fetchMessages: async () => {
+        fetched = true;
+        return { resolved: true, messages: [] };
+      },
       processMessageFn: async () => {},
     });
-    expect(summary).toBeNull();
+    expect(fetched).toBe(true);
+    expect(summary).not.toBeNull();
+  });
+
+  it("advances cursor only to last fetched ts when result is truncated (perRunLimit hit)", async () => {
+    // Long-outage scenario: 4 messages arrived during downtime but
+    // perRunLimit=2. Sort:ASC means we get the 2 oldest. Cursor must
+    // advance to the 2nd's timestamp (not nowMs) so the next startup
+    // picks up the remaining 2.
+    const now = 100 * 60 * 1000;
+    await saveBlueBubblesCatchupCursor("test-account", 50 * 60 * 1000);
+    const summary = await runBlueBubblesCatchup(
+      makeTarget({
+        account: {
+          accountId: "test-account",
+          enabled: true,
+          configured: true,
+          baseUrl: "http://127.0.0.1:1234",
+          config: {
+            serverUrl: "http://127.0.0.1:1234",
+            password: "x",
+            network: { dangerouslyAllowPrivateNetwork: true },
+            catchup: { perRunLimit: 2 },
+          } as unknown as WebhookTarget["account"]["config"],
+        },
+      }),
+      {
+        now: () => now,
+        fetchMessages: async () => ({
+          resolved: true,
+          // Only the 2 the cap allows BB to return (oldest first via ASC).
+          messages: [
+            makeBbMessage({ guid: "p1", dateCreated: 60 * 60 * 1000 }),
+            makeBbMessage({ guid: "p2", dateCreated: 70 * 60 * 1000 }),
+          ],
+        }),
+        processMessageFn: async () => {},
+      },
+    );
+    expect(summary?.replayed).toBe(2);
+    expect(summary?.fetchedCount).toBe(2);
+    expect(summary?.cursorAfter).toBe(70 * 60 * 1000); // page boundary, not nowMs
+    const cursor = await loadBlueBubblesCatchupCursor("test-account");
+    expect(cursor?.lastSeenMs).toBe(70 * 60 * 1000);
   });
 
   it("filters isFromMe before dispatch and still advances cursor", async () => {
diff --git a/extensions/bluebubbles/src/catchup.ts b/extensions/bluebubbles/src/catchup.ts
@@ -21,9 +21,6 @@ const MAX_MAX_AGE_MINUTES = 12 * 60;
 const DEFAULT_PER_RUN_LIMIT = 50;
 const MAX_PER_RUN_LIMIT = 500;
 const DEFAULT_FIRST_RUN_LOOKBACK_MINUTES = 30;
-// Skip catchup on restarts <30s apart to avoid churn on healthy rolling
-// restarts (e.g. automated repair loops, deploy scripts).
-const MIN_INTERVAL_MS = 30_000;
 const FETCH_TIMEOUT_MS = 15_000;
 
 export type BlueBubblesCatchupConfig = {
@@ -225,14 +222,15 @@ export async function runBlueBubblesCatchup(
   const existing = await loadBlueBubblesCatchupCursor(accountId).catch(() => null);
   const cursorBefore = existing?.lastSeenMs ?? null;
 
-  if (existing && nowMs >= existing.lastSeenMs && nowMs - existing.lastSeenMs < MIN_INTERVAL_MS) {
-    // A recent run just committed; skip to avoid churn on rolling restarts.
-    // The `nowMs >= existing.lastSeenMs` guard avoids a wall-clock-skew
-    // hazard: if the host clock jumps backwards (NTP correction, manual
-    // adjust), a future-dated cursor would otherwise satisfy this gate
-    // forever and silently disable catchup until wall time caught up.
-    return null;
-  }
+  // Catchup runs once per gateway startup (called from monitor.ts after
+  // webhook target registration). We deliberately do NOT short-circuit on
+  // a "ran recently" gate, because catchup is the only mechanism that
+  // recovers messages dropped during the gateway-down window. A short
+  // gap (e.g. <30s) between two startups can still have lost messages in
+  // the middle, and skipping the second startup's catchup would lose
+  // them permanently. The bounded query (perRunLimit, maxAge) and the
+  // inbound-dedupe cache from #66230 cap the cost of running the query
+  // every startup.
 
   const earliestAllowed = nowMs - maxAgeMs;
   // A future-dated cursor (clock rollback via NTP correction or manual
@@ -299,12 +297,20 @@ export async function runBlueBubblesCatchup(
   // unlikely to ever normalize on retry, and blocking on them would wedge
   // catchup forever.
   let earliestProcessFailureTs: number | null = null;
+  // Track the latest fetched message timestamp regardless of fate, so a
+  // truncated query (fetchedCount === perRunLimit) can advance the cursor
+  // exactly to the page boundary. Without this, the unfetched tail past
+  // the cap is permanently unreachable.
+  let latestFetchedTs = windowStartMs;
 
   for (const rec of messages) {
     // Defense in depth: the server-side `after:` filter should already
     // exclude pre-cursor messages, but guard here against BB API variants
     // that return inclusive-of-boundary data.
     const ts = typeof rec.dateCreated === "number" ? rec.dateCreated : 0;
+    if (ts > 0 && ts > latestFetchedTs) {
+      latestFetchedTs = ts;
+    }
     if (ts > 0 && ts <= windowStartMs) {
       summary.skippedPreCursor++;
       continue;
@@ -339,17 +345,29 @@ export async function runBlueBubblesCatchup(
     }
   }
 
-  // Compute the new cursor. Default is `nowMs` so subsequent runs start
-  // from the moment this sweep finished (avoiding stuck rescans of a
-  // message with `dateCreated > nowMs` from minor clock skew between the
-  // BB Server host and the gateway host). If any `processMessage` call
-  // threw, hold the cursor just before the earliest failure so the next
-  // run retries from there. The inbound-dedupe cache from #66230 keeps
-  // already-replayed messages from being re-processed on that retry.
+  // Compute the new cursor.
+  //
+  // - Default: advance to `nowMs` so subsequent runs start from the moment
+  //   this sweep finished (avoiding stuck rescans of a message with
+  //   `dateCreated > nowMs` from minor clock skew between BB host and
+  //   gateway host).
+  // - On retryable failure (any `processMessage` throw): hold the cursor
+  //   just before the earliest failed timestamp so the next run retries
+  //   from there. The inbound-dedupe cache from #66230 keeps successfully
+  //   replayed messages from being re-processed.
+  // - On truncation (fetched === perRunLimit): advance only to the latest
+  //   fetched timestamp so the next run picks up from the page boundary.
+  //   Otherwise the unfetched tail past the cap (which can be substantial
+  //   during long outages) would be permanently unreachable.
+  const isTruncated = summary.fetchedCount >= perRunLimit;
   let nextCursorMs = nowMs;
   if (earliestProcessFailureTs !== null) {
     const heldCursor = Math.max(earliestProcessFailureTs - 1, cursorBefore ?? windowStartMs);
     nextCursorMs = Math.min(heldCursor, nowMs);
+  } else if (isTruncated) {
+    // Use latestFetchedTs (clamped to >= prior cursor and <= nowMs) so the
+    // next run starts where this page ended.
+    nextCursorMs = Math.min(Math.max(latestFetchedTs, cursorBefore ?? windowStartMs), nowMs);
   }
   summary.cursorAfter = nextCursorMs;
   await saveBlueBubblesCatchupCursor(accountId, nextCursorMs).catch((err) => {
@@ -363,17 +381,18 @@ export async function runBlueBubblesCatchup(
       `window_ms=${nowMs - windowStartMs}`,
   );
 
-  // Emit a distinct warning when the BB result hits perRunLimit. The cursor
-  // has already advanced to nowMs, so any older messages BB would have
-  // returned past the cap are unreachable on the next sweep. Loud signal
-  // for operators to raise perRunLimit before a long outage silently
-  // truncates inbound history.
-  if (summary.fetchedCount >= perRunLimit) {
+  // Distinct WARNING when the BB result hits perRunLimit so operators
+  // know a single startup didn't drain the full backlog. The cursor was
+  // advanced only to the page boundary above, so the unfetched tail will
+  // be picked up on the next gateway startup — but if startups are
+  // infrequent, raising perRunLimit drains larger backlogs in one pass.
+  if (isTruncated) {
     error?.(
       `[${accountId}] BlueBubbles catchup: WARNING fetched=${summary.fetchedCount} ` +
-        `hit perRunLimit=${perRunLimit}; older messages in the window may have ` +
-        `been truncated. Raise channels.bluebubbles...catchup.perRunLimit if ` +
-        `outages can exceed this many inbound messages.`,
+        `hit perRunLimit=${perRunLimit}; cursor advanced only to page boundary, ` +
+        `remaining messages will be picked up on next startup. Raise ` +
+        `channels.bluebubbles...catchup.perRunLimit to drain larger backlogs ` +
+        `in a single pass.`,
     );
   }
 
