indexer/follower: refresh OgmiosSource intersect on each saved cursor
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>
This commit is contained in:
@@ -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> {
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user