From 7a806ff1e2f82500d8f8045dbcf51358e846b026 Mon Sep 17 00:00:00 2001
From: david <david.cejtlin@hal8.io>
Date: Tue, 19 May 2026 14:59:25 +0200
Subject: [PATCH] indexer/follower: refresh OgmiosSource intersect on each
 saved cursor
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The mainnet marketplace-indexer was stuck in an event-replay loop since
2026-05-14. Diagnosis: OgmiosSource.intersectAt is captured ONCE in
start() from the cursor at process startup. On every WebSocket reconnect
(Cloudflare-fronted WS times out at ~100s idle, plus periodic Ogmios
server restarts), scheduleReconnect calls sendFindIntersection with the
stale start-time intersect, never the latest persisted cursor. Ogmios
seeks to that old point, sends nextBlock forward from there, the worker
re-processes the same blocks and writes saveCursor over the advanced
cursor with the old slot — DB cursor effectively never moves past the
startup point, and events get duplicated each iteration.

Fix:
- Add ChainSyncSource.updateIntersect({ slot, id }) to the interface.
- OgmiosSource.updateIntersect mutates this.intersectAt, so the next
  reconnect's findIntersection picks up from the latest saved cursor.
- worker.handleBlock calls source.updateIntersect after every saveCursor.
- worker.handleRollback also calls it so a WS drop right after a rollback
  doesn't resume past the rollback point.
- MockChainSyncSource implements as no-op (no reconnect path in tests).

Verified: same shape as the testnet projection-gap bug (memory 95fe6946),
manifesting differently. After this fix the cursor will only ever advance,
and reconnects resume from the freshest persisted point.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 indexer/src/follower/ogmiosSource.ts          | 4 ++++
 indexer/src/follower/types.ts                 | 3 +++
 indexer/src/follower/worker.ts                | 9 +++++++++
 tests/unit/__helpers__/MockChainSyncSource.ts | 5 +++++
 4 files changed, 21 insertions(+)

diff --git a/indexer/src/follower/ogmiosSource.ts b/indexer/src/follower/ogmiosSource.ts
index 572f35b..a7b873b 100644
--- a/indexer/src/follower/ogmiosSource.ts
+++ b/indexer/src/follower/ogmiosSource.ts
@@ -76,6 +76,10 @@ export class OgmiosSource extends EventEmitter implements ChainSyncSource {
     this.ws.send(JSON.stringify({ jsonrpc: "2.0", method: "nextBlock" }));
   }
 
+  updateIntersect(point: { slot: number; id: string }): void {
+    this.intersectAt = point;
+  }
+
   // ─── Internal ───────────────────────────────────────────────────────────
 
   private async connect(): Promise<void> {
diff --git a/indexer/src/follower/types.ts b/indexer/src/follower/types.ts
index 9d39674..83555e3 100644
--- a/indexer/src/follower/types.ts
+++ b/indexer/src/follower/types.ts
@@ -68,6 +68,9 @@ export interface ChainSyncSource extends EventEmitter {
   start(intersectAt?: { slot: number; id: string } | null): Promise<void>;
   stop(): void;
   requestNextBlock(): void;
+  /** Update the cached intersect point used on reconnect. Without this,
+   * Ogmios WebSocket disconnects rewind the cursor to the startup point. */
+  updateIntersect(point: { slot: number; id: string }): void;
 }
 
 /**
diff --git a/indexer/src/follower/worker.ts b/indexer/src/follower/worker.ts
index 4310d23..a524e0e 100644
--- a/indexer/src/follower/worker.ts
+++ b/indexer/src/follower/worker.ts
@@ -149,6 +149,10 @@ export class ChainSyncWorker extends EventEmitter {
     }
 
     this.cache.saveCursor(slot, block.id);
+    // Mirror the saved cursor into the source's reconnect-intersect so that
+    // a WebSocket drop+reconnect resumes from here rather than rewinding to
+    // the startup intersect point.
+    this.source.updateIntersect({ slot, id: block.id });
     this.currentSlot = slot;
     this.emit("sync-progress", { currentSlot: slot, tipSlot: this.tipSlot });
 
@@ -239,6 +243,11 @@ export class ChainSyncWorker extends EventEmitter {
     const slot = data.point?.slot ?? 0;
     this.cache.rollbackToSlot(slot);
     this.currentSlot = slot;
+    // Also slide the source's reconnect-intersect back so a subsequent WS
+    // drop doesn't resume past the rollback point.
+    if (data.point && typeof data.point.id === "string") {
+      this.source.updateIntersect({ slot, id: data.point.id });
+    }
     this.emit("sync-progress", { currentSlot: slot, tipSlot: this.tipSlot });
     this.source.requestNextBlock();
   }
diff --git a/tests/unit/__helpers__/MockChainSyncSource.ts b/tests/unit/__helpers__/MockChainSyncSource.ts
index 7d75e28..841e80d 100644
--- a/tests/unit/__helpers__/MockChainSyncSource.ts
+++ b/tests/unit/__helpers__/MockChainSyncSource.ts
@@ -50,6 +50,11 @@ export class MockChainSyncSource extends EventEmitter implements ChainSyncSource
     // Subclasses that need step-by-step control can override.
   }
 
+  updateIntersect(_point: { slot: number; id: string }): void {
+    // No-op in mock mode; the source has no reconnect path. Kept to satisfy
+    // the ChainSyncSource interface contract.
+  }
+
   private _drainAll(): void {
     while (this.eventIndex < this.events.length) {
       const ev = this.events[this.eventIndex++];