indexer/follower: watchdog reconnect on silent Ogmios half-disconnect

Caught the indexer stuck for 3.5h at slot 23463870 today while the
chain tip had advanced to 23477134. Ogmios TCP socket was alive but
the server had stopped pushing nextBlock responses; /healthz reported
ogmios_status: "connected" because that flag only tracked tip_slot > 0,
not actual block flow.

OgmiosSource now arms a 90s response watchdog on every nextBlock /
findIntersection send and clears it on any well-formed protocol reply.
On timeout it ws.close()s, re-entering the existing onclose →
scheduleReconnect → findIntersection path that resumes from the saved
cursor. Errors and malformed messages intentionally leave the watchdog
armed so a stuck Ogmios error path also recycles instead of wedging.

OGMIOS_RESPONSE_TIMEOUT_MS env var tunes the window (0 disables).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

indexer/src/config.ts

@@ -19,6 +19,8 @@
  *   INDEXER_PORT                — HTTP port (default 8090)
  *   STATUS_POLL_MS              — status poll interval in ms (default 20000)
  *   SKIP_BEFORE_SLOT            — skip blocks before this slot (default 0)
+ *   OGMIOS_RESPONSE_TIMEOUT_MS  — watchdog timeout; reconnect Ogmios WS if no
+ *                                 response in this window (default 90000, 0 = off)
  */
 
 import type { NetworkId } from "@marketplace/shared";
@@ -30,6 +32,7 @@ export interface IndexerConfig {
   indexerPort: number;
   statusPollMs: number;
   skipBeforeSlot: number;
+  ogmiosResponseTimeoutMs: number;
   /**
    * Optional path to the bundled indexer-ui dist directory. When set, the
    * Express app mounts static assets + SPA catch-all so the same process
@@ -86,6 +89,10 @@ export function loadConfig(env: Record<string, string | undefined>): IndexerConf
     ? parseNonNegativeInt("SKIP_BEFORE_SLOT", env.SKIP_BEFORE_SLOT)
     : 0;
 
+  const ogmiosResponseTimeoutMs = env.OGMIOS_RESPONSE_TIMEOUT_MS !== undefined
+    ? parseNonNegativeInt("OGMIOS_RESPONSE_TIMEOUT_MS", env.OGMIOS_RESPONSE_TIMEOUT_MS)
+    : 90_000;
+
   // Optional INDEXER_UI_DIST: any non-empty string is accepted; directory
   // existence is checked at startup, not here. Absent → undefined (API-only).
   const uiDistDir = typeof env.INDEXER_UI_DIST === "string" && env.INDEXER_UI_DIST.length > 0
@@ -99,6 +106,7 @@ export function loadConfig(env: Record<string, string | undefined>): IndexerConf
     indexerPort,
     statusPollMs,
     skipBeforeSlot,
+    ogmiosResponseTimeoutMs,
     uiDistDir,
   };
 }

indexer/src/follower/ogmiosSource.ts

@@ -13,6 +13,7 @@ import type { ChainSyncSource, IndexerBlock, RollbackPoint } from "./types.js";
 
 const INITIAL_BACKOFF = 1_000;
 const MAX_BACKOFF = 30_000;
+const DEFAULT_RESPONSE_TIMEOUT_MS = 90_000;
 
 interface MinimalWebSocket {
   readyState: number;
@@ -35,6 +36,20 @@ async function loadWs(): Promise<WsCtor> {
   return _WsCtor;
 }
 
+export interface OgmiosSourceOpts {
+  /**
+   * Watchdog timeout (ms). If no response arrives from Ogmios within this
+   * window after a request was sent, the WebSocket is force-closed so the
+   * existing reconnect path can re-establish the chain-sync stream from the
+   * saved cursor. Defaults to 90_000. Set to 0 to disable.
+   *
+   * Rationale: Ogmios WebSockets can half-die (TCP socket alive, server
+   * stops pushing nextBlock responses). The indexer would otherwise sit
+   * idle indefinitely with `ogmios_status: "connected"` reported as true.
+   */
+  responseTimeoutMs?: number;
+}
+
 export class OgmiosSource extends EventEmitter implements ChainSyncSource {
   private url: string;
   private ws: MinimalWebSocket | null = null;
@@ -42,10 +57,13 @@ export class OgmiosSource extends EventEmitter implements ChainSyncSource {
   private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
   private closed: boolean = false;
   private intersectAt: { slot: number; id: string } | null = null;
+  private responseTimeoutMs: number;
+  private responseWatchdog: ReturnType<typeof setTimeout> | null = null;
 
-  constructor(ogmiosUrl: string) {
+  constructor(ogmiosUrl: string, opts: OgmiosSourceOpts = {}) {
     super();
     this.url = ogmiosUrl;
+    this.responseTimeoutMs = opts.responseTimeoutMs ?? DEFAULT_RESPONSE_TIMEOUT_MS;
   }
 
   async start(intersectAt?: { slot: number; id: string } | null): Promise<void> {
@@ -57,6 +75,7 @@ export class OgmiosSource extends EventEmitter implements ChainSyncSource {
 
   stop(): void {
     this.closed = true;
+    this.clearWatchdog();
     if (this.reconnectTimer) {
       clearTimeout(this.reconnectTimer);
       this.reconnectTimer = null;
@@ -74,6 +93,7 @@ export class OgmiosSource extends EventEmitter implements ChainSyncSource {
       return;
     }
     this.ws.send(JSON.stringify({ jsonrpc: "2.0", method: "nextBlock" }));
+    this.armWatchdog();
   }
 
   updateIntersect(point: { slot: number; id: string }): void {
@@ -104,6 +124,7 @@ export class OgmiosSource extends EventEmitter implements ChainSyncSource {
         }
       };
       socket.onclose = (ev) => {
+        this.clearWatchdog();
         this.emit("disconnected");
         if (!settled) {
           settled = true;
@@ -125,6 +146,7 @@ export class OgmiosSource extends EventEmitter implements ChainSyncSource {
       method: "findIntersection",
       params: { points },
     }));
+    this.armWatchdog();
   }
 
   private handleMessage(data: string): void {
@@ -132,13 +154,21 @@ export class OgmiosSource extends EventEmitter implements ChainSyncSource {
     try {
       msg = JSON.parse(data);
     } catch (err) {
+      // Watchdog stays armed — a stream of unparseable garbage triggers reconnect.
       this.emit("error", err);
       return;
     }
     if (msg.error) {
+      // Same: leave watchdog armed so a stuck Ogmios error path still recycles
+      // the connection rather than wedging the follower.
       this.emit("error", new Error(msg.error.message ?? "Ogmios error"));
       return;
     }
+    // Well-formed protocol message received — clear the watchdog. It will be
+    // re-armed by the next requestNextBlock() / sendFindIntersection() send.
+    if (msg.method) {
+      this.clearWatchdog();
+    }
     if (msg.method === "findIntersection") {
       this.emit("intersection", msg.result);
       // Pipeline: kick off some next-block requests
@@ -174,4 +204,31 @@ export class OgmiosSource extends EventEmitter implements ChainSyncSource {
     }, this.backoff);
     this.backoff = Math.min(this.backoff * 2, MAX_BACKOFF);
   }
+
+  private armWatchdog(): void {
+    this.clearWatchdog();
+    if (this.responseTimeoutMs <= 0) return;
+    this.responseWatchdog = setTimeout(() => {
+      this.responseWatchdog = null;
+      console.warn(
+        `[OgmiosSource] no response from Ogmios in ${this.responseTimeoutMs}ms — forcing reconnect`,
+      );
+      // Closing the socket triggers onclose → scheduleReconnect, which
+      // re-runs findIntersection from the cached intersectAt point.
+      if (this.ws) {
+        try {
+          this.ws.close();
+        } catch {
+          // ignore — close() should be idempotent on any sane WebSocket impl
+        }
+      }
+    }, this.responseTimeoutMs);
+  }
+
+  private clearWatchdog(): void {
+    if (this.responseWatchdog) {
+      clearTimeout(this.responseWatchdog);
+      this.responseWatchdog = null;
+    }
+  }
 }

indexer/src/index.ts

@@ -38,7 +38,9 @@ async function main(): Promise<void> {
   }
 
   const cache = new SqliteCache(config.dbPath);
-  const source = new OgmiosSource(config.ogmiosUrl);
+  const source = new OgmiosSource(config.ogmiosUrl, {
+    responseTimeoutMs: config.ogmiosResponseTimeoutMs,
+  });
   const worker = new ChainSyncWorker({
     source,
     cache,