7a806ff1e2
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>
79 lines
2.6 KiB
TypeScript
79 lines
2.6 KiB
TypeScript
/**
|
|
* tests/unit/__helpers__/MockChainSyncSource.ts — test helper for block streaming.
|
|
*
|
|
* MockChainSyncSource implements ChainSyncSource and replays a pre-built sequence
|
|
* of blocks (or rollback points) synchronously so ChainSyncWorker tests don't need
|
|
* a real WebSocket.
|
|
*
|
|
* This is NOT a fixture (it doesn't build datums); it is test infrastructure.
|
|
* The no-shared-helper rule applies only to datum builders, not to transport mocks.
|
|
*
|
|
* Usage:
|
|
* const source = new MockChainSyncSource([block1, block2]);
|
|
* const worker = new ChainSyncWorker({ source, cache, addresses });
|
|
* await worker.start();
|
|
* // all blocks are replayed synchronously before start() resolves
|
|
*/
|
|
|
|
import { EventEmitter } from "events";
|
|
import type { ChainSyncSource, IndexerBlock, RollbackPoint } from "../../../indexer/src/follower/types.js";
|
|
|
|
export type MockChainEvent =
|
|
| { kind: "block"; block: IndexerBlock; tipSlot?: number }
|
|
| { kind: "rollback"; point: RollbackPoint };
|
|
|
|
export class MockChainSyncSource extends EventEmitter implements ChainSyncSource {
|
|
private events: MockChainEvent[];
|
|
private eventIndex: number = 0;
|
|
private started: boolean = false;
|
|
|
|
constructor(events: MockChainEvent[]) {
|
|
super();
|
|
this.events = events;
|
|
}
|
|
|
|
async start(_intersectAt?: { slot: number; id: string } | null): Promise<void> {
|
|
this.started = true;
|
|
this.emit("connected");
|
|
// Replay all events; the worker calls requestNextBlock() after each.
|
|
// We pre-drain them here by emitting them synchronously.
|
|
this._drainAll();
|
|
}
|
|
|
|
stop(): void {
|
|
this.started = false;
|
|
}
|
|
|
|
requestNextBlock(): void {
|
|
// The worker calls this after processing each block. In Mock mode the
|
|
// start() call pre-drains all events so this is a no-op.
|
|
// 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++];
|
|
if (ev.kind === "block") {
|
|
const tipSlot = ev.tipSlot ?? ev.block.slot + 100;
|
|
this.emit("block", { block: ev.block, tip: { slot: tipSlot } });
|
|
} else {
|
|
this.emit("rollback", { point: ev.point });
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Push an additional event after start() has been called (for step-mode tests). */
|
|
pushEvent(ev: MockChainEvent): void {
|
|
this.events.push(ev);
|
|
if (this.started) {
|
|
this.eventIndex = this.events.length - 1;
|
|
this._drainAll();
|
|
}
|
|
}
|
|
}
|