feat(kalshi): opt-in parallel backfill passes

.env.example

@@ -45,6 +45,15 @@ AUTO_RESTART_DELAY=10
 # Number of days to look back for forked blocks (default: 30)
 FORKED_BLOCKS_DAYS_BACK=30
 
+# Kalshi Backfill Configuration
+# Run the trades / markets / events backfill passes concurrently within one
+# cycle instead of sequentially. Concurrent mode is faster when upstream RTT
+# dominates but couples per-pass failure detection through the shared batch
+# queue's `lastFlushError` flag — one pass's flush failure forces the others'
+# next-page check to abort too. Off by default; opt in once parallel cycles
+# have been observed safe.
+# KALSHI_BACKFILL_PARALLEL=false
+
 # Token Metadata Overrides
 # URL of a tokens.json file (e.g. GitHub raw) used to override on-chain name/symbol
 # with curated values for known tokens by updating matching metadata rows at startup.

services/kalshi/backfill.test.ts

@@ -4,8 +4,10 @@ import {
     POISONED_SENTINEL,
     runEventsBackfill,
     runMarketsBackfill,
+    runPasses,
     runTradesBackfill,
 } from './backfill';
+import type { CursorCheckpoint } from './cursor';
 import type { EventEntity, Market, Trade } from './types';
 
 function trade(overrides: Partial<Trade>): Trade {
@@ -651,3 +653,271 @@ describe('runEventsBackfill', () => {
         expect(lastCall?.[1]).toMatch(/^2\d{3}-/);
     });
 });
+
+// ---------------------------------------------------------------------------
+// runPasses — sequential vs parallel orchestration. The shared per-pass
+// machinery is already covered above; here we only assert the cross-pass
+// scheduling + error-handling semantics that differ between modes.
+// ---------------------------------------------------------------------------
+
+interface GateClient {
+    getTradesHistorical: () => Promise<{ trades: Trade[]; cursor: string }>;
+    getMarketsHistorical: () => Promise<{ markets: Market[]; cursor: string }>;
+    getEvents: (params: {
+        min_updated_ts?: number;
+        cursor?: string;
+    }) => Promise<{ events: EventEntity[]; cursor: string }>;
+}
+
+/** Promise that resolves only when `release()` is called. Tracks observers
+ * for assertions like "did this fetch even start?". */
+function gate<T>(value: T) {
+    let resolveFn!: () => void;
+    let started = false;
+    const promise = new Promise<T>((res) => {
+        resolveFn = () => res(value);
+    });
+    return {
+        wait: () => {
+            started = true;
+            return promise;
+        },
+        release: () => resolveFn(),
+        get started() {
+            return started;
+        },
+    };
+}
+
+async function flushMicrotasks(n = 8): Promise<void> {
+    for (let i = 0; i < n; i++) {
+        await Promise.resolve();
+    }
+}
+
+const EMPTY_CURSORS = new Map<string, CursorCheckpoint>();
+
+describe('runPasses — scheduling', () => {
+    test('parallel mode kicks off all three passes before any completes', async () => {
+        const tradesGate = gate({ trades: [], cursor: '' });
+        const marketsGate = gate({ markets: [], cursor: '' });
+        const eventsGate = gate({ events: [], cursor: '' });
+        const client: GateClient = {
+            getTradesHistorical: () => tradesGate.wait(),
+            getMarketsHistorical: () => marketsGate.wait(),
+            getEvents: () => eventsGate.wait(),
+        };
+
+        const promise = runPasses(
+            client as unknown as Parameters<typeof runPasses>[0],
+            fakeQueue(),
+            EMPTY_CURSORS,
+            true,
+        );
+        await flushMicrotasks();
+
+        // All three fetchPage calls have hit the gate before any resolved.
+        expect(tradesGate.started).toBe(true);
+        expect(marketsGate.started).toBe(true);
+        expect(eventsGate.started).toBe(true);
+
+        tradesGate.release();
+        marketsGate.release();
+        eventsGate.release();
+        await promise;
+    });
+
+    test('sequential mode runs passes one at a time', async () => {
+        const tradesGate = gate({ trades: [], cursor: '' });
+        const marketsGate = gate({ markets: [], cursor: '' });
+        const eventsGate = gate({ events: [], cursor: '' });
+        const client: GateClient = {
+            getTradesHistorical: () => tradesGate.wait(),
+            getMarketsHistorical: () => marketsGate.wait(),
+            getEvents: () => eventsGate.wait(),
+        };
+
+        const promise = runPasses(
+            client as unknown as Parameters<typeof runPasses>[0],
+            fakeQueue(),
+            EMPTY_CURSORS,
+            false,
+        );
+        await flushMicrotasks();
+
+        // Only trades has started — markets + events must wait.
+        expect(tradesGate.started).toBe(true);
+        expect(marketsGate.started).toBe(false);
+        expect(eventsGate.started).toBe(false);
+
+        tradesGate.release();
+        await flushMicrotasks();
+        expect(marketsGate.started).toBe(true);
+        expect(eventsGate.started).toBe(false);
+
+        marketsGate.release();
+        await flushMicrotasks();
+        expect(eventsGate.started).toBe(true);
+
+        eventsGate.release();
+        await promise;
+    });
+});
+
+describe('runPasses — error propagation', () => {
+    test('parallel mode: failure in one pass does NOT cancel the others', async () => {
+        let marketsStarted = false;
+        let eventsStarted = false;
+        const client = {
+            getTradesHistorical: () => Promise.reject(new Error('trades boom')),
+            getMarketsHistorical: () => {
+                marketsStarted = true;
+                return Promise.resolve({ markets: [], cursor: '' });
+            },
+            getEvents: (_p: { min_updated_ts?: number; cursor?: string }) => {
+                eventsStarted = true;
+                return Promise.resolve({ events: [], cursor: '' });
+            },
+        };
+
+        await expect(
+            runPasses(
+                client as unknown as Parameters<typeof runPasses>[0],
+                fakeQueue(),
+                EMPTY_CURSORS,
+                true,
+            ),
+        ).rejects.toThrow(/trades boom/);
+
+        // markets + events still ran to completion (their fetches happened)
+        // even though trades rejected first — `allSettled` semantic.
+        expect(marketsStarted).toBe(true);
+        expect(eventsStarted).toBe(true);
+    });
+
+    test('sequential mode: failure in one pass halts subsequent passes', async () => {
+        let marketsStarted = false;
+        let eventsStarted = false;
+        const client = {
+            getTradesHistorical: () => Promise.reject(new Error('trades boom')),
+            getMarketsHistorical: () => {
+                marketsStarted = true;
+                return Promise.resolve({ markets: [], cursor: '' });
+            },
+            getEvents: (_p: { min_updated_ts?: number; cursor?: string }) => {
+                eventsStarted = true;
+                return Promise.resolve({ events: [], cursor: '' });
+            },
+        };
+
+        await expect(
+            runPasses(
+                client as unknown as Parameters<typeof runPasses>[0],
+                fakeQueue(),
+                EMPTY_CURSORS,
+                false,
+            ),
+        ).rejects.toThrow(/trades boom/);
+
+        // Sequential mode short-circuits on first failure.
+        expect(marketsStarted).toBe(false);
+        expect(eventsStarted).toBe(false);
+    });
+
+    test('error carries the failing pass label (parallel)', async () => {
+        const client = {
+            getTradesHistorical: () =>
+                Promise.resolve({ trades: [], cursor: '' }),
+            getMarketsHistorical: () =>
+                Promise.reject(new Error('markets boom')),
+            getEvents: () => Promise.resolve({ events: [], cursor: '' }),
+        };
+
+        try {
+            await runPasses(
+                client as unknown as Parameters<typeof runPasses>[0],
+                fakeQueue(),
+                EMPTY_CURSORS,
+                true,
+            );
+            throw new Error('expected runPasses to throw');
+        } catch (e) {
+            expect((e as Error & { pass?: string }).pass).toBe('markets');
+        }
+    });
+});
+
+describe('runPasses — sentinel short-circuits apply in both modes', () => {
+    function buildClient() {
+        const tradesCalled = { v: false };
+        const marketsCalled = { v: false };
+        const eventsCalled = { v: false };
+        const client = {
+            getTradesHistorical: () => {
+                tradesCalled.v = true;
+                return Promise.resolve({ trades: [], cursor: '' });
+            },
+            getMarketsHistorical: () => {
+                marketsCalled.v = true;
+                return Promise.resolve({ markets: [], cursor: '' });
+            },
+            getEvents: (_p: { min_updated_ts?: number; cursor?: string }) => {
+                eventsCalled.v = true;
+                return Promise.resolve({ events: [], cursor: '' });
+            },
+        };
+        return { client, tradesCalled, marketsCalled, eventsCalled };
+    }
+
+    function buildCursors(
+        overrides: Record<string, Partial<CursorCheckpoint>> = {},
+    ): Map<string, CursorCheckpoint> {
+        const m = new Map<string, CursorCheckpoint>();
+        for (const [scope, partial] of Object.entries(overrides)) {
+            m.set(scope, {
+                scope,
+                last_cursor: '',
+                last_processed_ts_ms: 0,
+                last_processed_ts_iso: '2026-01-01T00:00:00.000000Z',
+                ...partial,
+            });
+        }
+        return m;
+    }
+
+    for (const parallel of [true, false]) {
+        test(`drained scope is skipped (parallel=${parallel})`, async () => {
+            const { client, tradesCalled, marketsCalled, eventsCalled } =
+                buildClient();
+            const cursors = buildCursors({
+                trades_backfill: { last_cursor: DRAINED_SENTINEL },
+            });
+            await runPasses(
+                client as unknown as Parameters<typeof runPasses>[0],
+                fakeQueue(),
+                cursors,
+                parallel,
+            );
+            expect(tradesCalled.v).toBe(false);
+            expect(marketsCalled.v).toBe(true);
+            expect(eventsCalled.v).toBe(true);
+        });
+
+        test(`poisoned scope is skipped (parallel=${parallel})`, async () => {
+            const { client, tradesCalled, marketsCalled, eventsCalled } =
+                buildClient();
+            const cursors = buildCursors({
+                markets_backfill: { last_cursor: POISONED_SENTINEL },
+            });
+            await runPasses(
+                client as unknown as Parameters<typeof runPasses>[0],
+                fakeQueue(),
+                cursors,
+                parallel,
+            );
+            expect(tradesCalled.v).toBe(true);
+            expect(marketsCalled.v).toBe(false);
+            expect(eventsCalled.v).toBe(true);
+        });
+    }
+});