[PROJ-1851] Preserve Jetstream progress under firehose load

docs/dev-journal.md

@@ -1618,3 +1618,44 @@ The v4 product path now starts directly with Community Governed Feed, exposes al
 ### Release boundary
 
 No commit, merge, or deployment has occurred. Production deployment and complete browser signoff remain blocked on PROJ-1753 until fresh-load and HTTP revalidation prove the CSP/304 blank-page defect closed. The live-feed panel is proof of the public feed; the guided comparison set is deliberately frozen and does not claim to rescan the firehose or reproduce all candidates that might enter a new production scoring run.
+
+## 2026-07-13 #06 - PROJ-1851 Jetstream ingestion backpressure
+
+**Branch:** `dev/PROJ-1851-jetstream-backpressure`
+**Commits:** `8bd9aa842db37eb053989da768bfad71e151c2d2`, `7cf05e021ed659cf268ecf9cfdff852880365a78`; final adversarial hardening is included here pending review closeout.
+
+### Incident evidence
+
+Production remained live and ready, but the Jetstream worker repeatedly filled its 10,000-event pending queue and reconnected from an old cursor. The service logged 1,828 queue-saturation events between 2026-07-12 07:57 EDT and incident capture. Twenty handlers were active, cursor lag was approximately 35,700 seconds (9.9 hours), and a 30-second sample advanced the cursor by only 27.4 seconds, so lag was still growing despite green liveness/readiness and zero systemd restarts.
+
+### What changed
+
+The WebSocket now pauses inbound delivery when the pending queue reaches a bounded high-water mark and resumes only after it drains to a lower threshold. The existing 10,000-event overflow/reconnect path remains as a last-resort circuit breaker. Pause activation occurs in the same synchronous queue mutation that enqueues work, avoiding an ordering dependency in the message callback.
+
+Hosted review identified two additional lifecycle races before release. Messages buffered on a replaced socket are now rejected before `processEvent`, and graceful shutdown detaches inbound delivery, drains active handlers, then saves the resulting safely completed cursor. The test reset path now closes sockets, cancels reconnect/metrics timers, and restores primary/fallback state. An explicit reserved-headroom calculation covers frames already buffered when `pause()` takes effect.
+
+Reconnects reset in-memory cursor bookkeeping to the durable resume cursor, so a failed pre-close save cannot cause replayed events to inherit a newer unsaved cursor or save interval. Runtime telemetry now reports active and pending work, pause/resume counts, overload reconnects, cumulative drops, safely completed cursor, and cursor lag through admin health, feed-health, and vitals surfaces.
+
+Jetstream health now requires both recent event handling and a cursor less than five minutes behind. A stale cursor degrades operator/public health while `/health/ready` remains dependent only on PostgreSQL and Redis, allowing a healthy process to catch up without a restart loop.
+
+Two independent adversarial reviews then found correctness hazards that the first pass did not cover. Queued cursors are now registered before they wait for a processing slot, so a later event cannot advance the safe cursor past same-timestamp queued work. Commit events for the same DID/collection/rkey are serialized in arrival order, and replayed duplicate replies no longer increment a root post's reply count when the post insert is a no-op.
+
+Reconnect cancellation is now distinct from a hard-limit drop, and pause/resume failures have their own reconnect counter. Delayed overload recovery is bound to the socket and connection generation that overflowed, so it cannot close a replacement socket. Graceful shutdown raises a typed error when the final cursor cannot be persisted while still clearing its stopped lifecycle state. Failed dead-letter persistence retains a bounded, fail-closed cursor floor; stale generations cannot overwrite it, operators can see it in health surfaces, and only an explicit operator clear or a new connection generation releases it.
+
+### Verification
+
+- Focused backpressure, lifecycle, message-processing, metrics, and health suite: 6 files / 63 tests passed.
+- Full backend suite: 153 files / 1,716 tests passed on the final exact worktree.
+- `npm run build`, `npm run verify`, `npm run docs:verify`, `web-next` lint, and `web-next` strict TypeScript passed. The canonical `web-next` package has no test script; its production build passed through `npm run verify`.
+- Socket-level coverage proves high-water pause, buffered-frame headroom, low-water resume, pause/resume exception recovery, sustained asynchronous drainage, zero overflow reconnects/drops on the bounded path, closed/no-socket handling, hard-limit drop telemetry, startup null-cursor serialization, five-minute health boundaries, durable-cursor reset after reconnect, safe in-memory cursor fallback when a reconnect-time database read fails, stale-socket close and message isolation after a replacement connection is active, final-cursor persistence after active shutdown work drains, complete test lifecycle reset, and cancellation of obsolete reconnect timers before an operator-triggered replacement.
+- The first local CodeRabbit review reported one major health-message gap and five maintainability/test gaps. A second current-diff review found one duplicated freshness threshold plus three test/documentation gaps. All ten findings were addressed before commit. A third local convergence request and a final exact-diff retry both stalled without results and were terminated after bounded waits; hosted exact-head review remains required before closeout.
+- Hosted review of commit `8bd9aa842db37eb053989da768bfad71e151c2d2` reported nine inline threads plus two outside-diff lifecycle findings. All valid findings were addressed in the follow-up diff. A local review of that follow-up remained in summarization for five minutes without emitting a finding and was terminated; a new hosted current-head review remains mandatory after push.
+- Post-adversarial focused verification passed: 7 files / 78 tests covering backpressure, lifecycle, cursor safety, dead-letter failure, same-record ordering, duplicate replies, health telemetry, and replacement-socket isolation. Root TypeScript build also passed.
+- A production-shaped local burst used the real `startJetstream()` WebSocket path, PostgreSQL 16, Redis 7, the full migration set, and 5,000 valid cursor-bearing events. It completed in 989.38 ms (5,053.69 events/second), exercised 26 high-water pauses and 26 low-water resumes, held 20 active handlers with a maximum pending depth of 275, and recorded zero hard-limit drops, overload reconnects, flow-control-failure reconnects, or fail-closed cursor floors. The expected, runtime, and persisted cursors all matched at `1800000000027714`; Redis returned `PONG`; and cleanup exited cleanly.
+- The earlier sequential replay receipt remains complementary rather than a substitute for the socket burst: 5,000 events at 802.44 events/second, p95 handler latency 3.08 ms, p99 4.34 ms, zero drops, zero handler/state/outcome mismatches, and 2,369 durable mutations.
+
+### Release boundary
+
+This entry records local implementation and deterministic verification only. Production success requires the exact merged SHA to show no new hard-limit reconnects or dropped events, increasing pause/resume counters, a declining cursor lag, unchanged public-feed availability, and healthy readiness while catch-up proceeds. No cursor jump or backlog drop is authorized.
+
+Cursor-staleness health is now observable, but no tracked alert consumer yet pages on stalled cursor progression. That operational alert is intentionally split from this ingestion-correctness PR so liveness/readiness restart semantics are not changed in the incident fix.

src/admin/routes/feed-health.ts

@@ -16,6 +16,7 @@ import { adminSecurity } from '../../lib/openapi.js';
 import {
   getJetstreamDisconnectedAt,
   getJetstreamEventsLast5Min,
+  getJetstreamRuntimeState,
   getLastEventReceivedAt,
   isJetstreamConnected,
   triggerJetstreamReconnect,
@@ -53,6 +54,7 @@ export function registerFeedHealthRoutes(app: FastifyInstance): void {
     const connected = isJetstreamConnected();
     const lastEventAt = getLastEventReceivedAt();
     const disconnectedAt = getJetstreamDisconnectedAt();
+    const jetstreamRuntime = getJetstreamRuntimeState();
     const disconnectedForSeconds =
       !connected && disconnectedAt
         ? Math.max(0, Math.floor((Date.now() - disconnectedAt.getTime()) / 1000))
@@ -112,6 +114,17 @@ export function registerFeedHealthRoutes(app: FastifyInstance): void {
         lastEvent: lastEventAt ? lastEventAt.toISOString() : null,
         eventsLast5min: getJetstreamEventsLast5Min(),
         disconnectedForSeconds,
+        cursorUs: jetstreamRuntime.cursorUs,
+        cursorLagMs: jetstreamRuntime.cursorLagMs,
+        activeEvents: jetstreamRuntime.activeEvents,
+        pendingEvents: jetstreamRuntime.pendingEvents,
+        inboundPaused: jetstreamRuntime.inboundPaused,
+        pauseCount: jetstreamRuntime.pauseCount,
+        resumeCount: jetstreamRuntime.resumeCount,
+        overloadReconnectCount: jetstreamRuntime.overloadReconnectCount,
+        flowControlFailureReconnectCount: jetstreamRuntime.flowControlFailureReconnectCount,
+        totalDroppedEvents: jetstreamRuntime.totalDroppedEvents,
+        failedCursorPersistenceFloorUs: jetstreamRuntime.failedCursorPersistenceFloorUs,
       },
       subscribers: {
         total: parseInt(subStats.rows[0].total, 10),

src/admin/routes/health.ts

@@ -39,6 +39,17 @@ export function registerAdminHealthRoutes(app: FastifyInstance): void {
                     status: { type: 'string', enum: ['healthy', 'unhealthy'] },
                     connected: { type: 'boolean', description: 'Whether the WebSocket is connected' },
                     last_event_age_ms: { type: 'number', description: 'Age of the last received event in ms' },
+                    cursor_us: { type: 'string', description: 'Latest safely completed Jetstream cursor in microseconds' },
+                    cursor_lag_ms: { type: 'number', description: 'Lag between the processed cursor and wall-clock time in ms' },
+                    active_events: { type: 'number', description: 'Jetstream events currently being processed' },
+                    pending_events: { type: 'number', description: 'Jetstream events waiting for a processing slot' },
+                    inbound_paused: { type: 'boolean', description: 'Whether socket reads are paused for backpressure' },
+                    pause_count: { type: 'number', description: 'Number of inbound backpressure pauses since process start' },
+                    resume_count: { type: 'number', description: 'Number of inbound backpressure resumes since process start' },
+                    overload_reconnect_count: { type: 'number', description: 'Reconnects triggered by the hard queue limit since process start' },
+                    flow_control_failure_reconnect_count: { type: 'number', description: 'Reconnects triggered by socket pause/resume failures since process start' },
+                    total_dropped_events: { type: 'number', description: 'Incoming events rejected at the hard queue limit since process start' },
+                    failed_cursor_persistence_floor_us: { type: 'string', description: 'Fail-closed cursor floor retained after a dead-letter persistence failure' },
                     error: { type: 'string' },
                   },
                   required: ['status', 'connected'],

src/admin/routes/vitals.ts

@@ -15,6 +15,7 @@ import { getHealthStatus } from '../../lib/health.js';
 import {
   getLastEventReceivedAt,
   getJetstreamEventsLast5Min,
+  getJetstreamRuntimeState,
   isJetstreamConnected,
   getJetstreamDisconnectedAt,
 } from '../../ingestion/jetstream.js';
@@ -123,6 +124,7 @@ export function registerVitalsRoutes(app: FastifyInstance): void {
     const connected = isJetstreamConnected();
     const lastEventAt = getLastEventReceivedAt();
     const disconnectedAt = getJetstreamDisconnectedAt();
+    const jetstreamRuntime = getJetstreamRuntimeState();
 
     // Feed freshness
     let feedAgeMs: number | undefined;
@@ -181,6 +183,17 @@ export function registerVitalsRoutes(app: FastifyInstance): void {
         last_event_at: lastEventAt?.toISOString() ?? null,
         last_event_age_ms: lastEventAt ? Date.now() - lastEventAt.getTime() : null,
         disconnected_since: !connected && disconnectedAt ? disconnectedAt.toISOString() : null,
+        cursor_us: jetstreamRuntime.cursorUs,
+        cursor_lag_ms: jetstreamRuntime.cursorLagMs,
+        active_events: jetstreamRuntime.activeEvents,
+        pending_events: jetstreamRuntime.pendingEvents,
+        inbound_paused: jetstreamRuntime.inboundPaused,
+        pause_count: jetstreamRuntime.pauseCount,
+        resume_count: jetstreamRuntime.resumeCount,
+        overload_reconnect_count: jetstreamRuntime.overloadReconnectCount,
+        flow_control_failure_reconnect_count: jetstreamRuntime.flowControlFailureReconnectCount,
+        total_dropped_events: jetstreamRuntime.totalDroppedEvents,
+        failed_cursor_persistence_floor_us: jetstreamRuntime.failedCursorPersistenceFloorUs,
       },
 
       scoring: scoringRun

src/index.ts

@@ -1,12 +1,25 @@
 import { config } from './config.js';
 import { logger } from './lib/logger.js';
 import { createServer } from './feed/server.js';
-import { startJetstream, stopJetstream, isJetstreamConnected, getLastEventReceivedAt } from './ingestion/jetstream.js';
+import {
+  getJetstreamRuntimeState,
+  getJetstreamStartedAt,
+  getLastEventReceivedAt,
+  isJetstreamConnected,
+  startJetstream,
+  stopJetstream,
+} from './ingestion/jetstream.js';
 import { startScoring, stopScoring, isScoringInProgress } from './scoring/scheduler.js';
 import { getLastScoringRunAt } from './scoring/pipeline.js';
 import { runStartupChecks } from './lib/startup-checks.js';
 import { registerShutdownHandlers } from './lib/shutdown.js';
-import { registerJetstreamHealth, registerScoringHealth, registerDiskHealth, JetstreamHealth, ScoringHealth } from './lib/health.js';
+import {
+  calculateJetstreamHealth,
+  registerDiskHealth,
+  registerJetstreamHealth,
+  registerScoringHealth,
+  type ScoringHealth,
+} from './lib/health.js';
 import { getDiskStatus } from './maintenance/disk-monitor.js';
 import { registerBotRoutes } from './bot/server.js';
 import { initializeBot } from './bot/agent.js';
@@ -99,21 +112,13 @@ async function main() {
   }
 
   // 4. Register Jetstream health check
-  registerJetstreamHealth((): JetstreamHealth => {
-    const connected = isJetstreamConnected();
-    const lastEventAt = getLastEventReceivedAt();
-    const lastEventAgeMs = lastEventAt ? Date.now() - lastEventAt.getTime() : undefined;
-
-    // Consider unhealthy if no events for more than 5 minutes
-    const isHealthy = connected && (lastEventAgeMs === undefined || lastEventAgeMs < 300_000);
-
-    return {
-      status: isHealthy ? 'healthy' : 'unhealthy',
-      connected,
-      last_event_age_ms: lastEventAgeMs,
-      error: !connected ? 'WebSocket not connected' : undefined,
-    };
-  });
+  registerJetstreamHealth(() => calculateJetstreamHealth(
+    isJetstreamConnected(),
+    getLastEventReceivedAt(),
+    getJetstreamRuntimeState(),
+    Date.now(),
+    getJetstreamStartedAt()
+  ));
 
   // 5. Register scoring health check (before starting scoring so it's available during initial run)
   registerScoringHealth((): ScoringHealth => {

src/ingestion/handlers/post-handler.ts

@@ -207,7 +207,7 @@ export async function handlePost(
     );
 
     // If this is a reply, increment reply count on the root post
-    if (replyRoot) {
+    if (replyRoot && postResult.rowCount) {
       await db.query(
         `UPDATE post_engagement SET reply_count = reply_count + 1, updated_at = NOW()
          WHERE post_uri = $1`,

src/ingestion/jetstream-health.ts

@@ -0,0 +1 @@
+export const JETSTREAM_FRESHNESS_LIMIT_MS = 5 * 60_000;