From e9638e6696a80c70af34738b2e73cc3375eec803 Mon Sep 17 00:00:00 2001 From: Etienne Donneger Date: Wed, 3 Jun 2026 17:06:59 -0400 Subject: [PATCH] feat(kalshi): opt-in parallel backfill passes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The trades / markets / events backfill passes run sequentially by default, which stretches one full cycle to ~10-15min once all three are walking (vs ~3min for trades alone on the previous single-pass shape). The passes are HTTP-bound and write to independent CH tables + cursor scopes, so concurrency is safe. Adds `KALSHI_BACKFILL_PARALLEL=true` opt-in. When set, the three passes run via `Promise.allSettled` within one cycle. A failure in one pass no longer cancels the others — partial-cycle progress is durable per-pass — and the first failure propagates with its pass label intact for cycle-error attribution. Default stays sequential so behavior is unchanged for existing deployments until operators explicitly opt in. Trade-off in concurrent mode: the BatchInsertQueue's `lastFlushError` health is module-level, so one pass's flush failure forces the others' next-page health check to abort too. That's "fail safe" — overly eager rather than silently dropping rows — and worth the simplicity vs per-pass queue isolation. `runPasses()` is now exported so the parallel/sequential scheduling can be unit-tested directly. Tests assert: - Parallel kicks off all three passes before any completes. - Sequential runs strictly one-at-a-time. - Parallel + one pass failing: other passes still complete. - Sequential + one pass failing: subsequent passes halted. - Errors carry the failing pass label. - Drained / poisoned short-circuits apply in both modes. Co-Authored-By: Claude Opus 4.7 (1M context) --- .env.example | 9 ++ services/kalshi/backfill.test.ts | 270 +++++++++++++++++++++++++++++++ services/kalshi/backfill.ts | 123 ++++++++++---- 3 files changed, 372 insertions(+), 30 deletions(-) diff --git a/.env.example b/.env.example index b1a19b0..cb82e7a 100644 --- a/.env.example +++ b/.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. diff --git a/services/kalshi/backfill.test.ts b/services/kalshi/backfill.test.ts index faffa9b..318bd26 100644 --- a/services/kalshi/backfill.test.ts +++ b/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 { @@ -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(value: T) { + let resolveFn!: () => void; + let started = false; + const promise = new Promise((res) => { + resolveFn = () => res(value); + }); + return { + wait: () => { + started = true; + return promise; + }, + release: () => resolveFn(), + get started() { + return started; + }, + }; +} + +async function flushMicrotasks(n = 8): Promise { + for (let i = 0; i < n; i++) { + await Promise.resolve(); + } +} + +const EMPTY_CURSORS = new Map(); + +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[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[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[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[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[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> = {}, + ): Map { + const m = new Map(); + 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[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[0], + fakeQueue(), + cursors, + parallel, + ); + expect(tradesCalled.v).toBe(true); + expect(marketsCalled.v).toBe(false); + expect(eventsCalled.v).toBe(true); + }); + } +}); diff --git a/services/kalshi/backfill.ts b/services/kalshi/backfill.ts index 0c299d2..1b9b08c 100644 --- a/services/kalshi/backfill.ts +++ b/services/kalshi/backfill.ts @@ -63,6 +63,15 @@ const SENTINELS = new Set([DRAINED_SENTINEL, POISONED_SENTINEL]); * resumable from the last checkpoint. */ const MAX_PAGES_PER_CYCLE = 1000; +/** When true, the three backfill passes run concurrently within one cycle + * (via `Promise.allSettled`). Default sequential — concurrent mode is + * faster when upstream RTT dominates but shares the BatchInsertQueue's + * `lastFlushError` health across passes (one pass's flush failure forces + * the others' next-page check to abort too). Opt in via the env var. */ +function isParallelEnabled(): boolean { + return process.env.KALSHI_BACKFILL_PARALLEL === 'true'; +} + type Queue = ReturnType; /** Callback that durably persists the current pagination cursor + watermark @@ -78,6 +87,7 @@ export async function run(): Promise { const client = new KalshiClient(); const queue = getBatchInsertQueue(); + const parallel = isParallelEnabled(); let primaryError: unknown; @@ -91,36 +101,7 @@ export async function run(): Promise { setScopePoisoned(scope, c?.last_cursor === POISONED_SENTINEL); } - // Sequential — three short HTTP-bound walks share the same batch - // queue, so running them back-to-back keeps the queue's table-keyed - // buffers from interleaving. Each pass yields the queue clean. - await runPass(SCOPE_TRADES, 'trades', cursors, () => - runTradesBackfill( - client, - queue, - (c, t) => setCursor(SCOPE_TRADES, c, t), - cursors.get(SCOPE_TRADES)?.last_cursor, - cursors.get(SCOPE_TRADES)?.last_processed_ts_iso, - ), - ); - await runPass(SCOPE_MARKETS, 'markets', cursors, () => - runMarketsBackfill( - client, - queue, - (c, t) => setCursor(SCOPE_MARKETS, c, t), - cursors.get(SCOPE_MARKETS)?.last_cursor, - cursors.get(SCOPE_MARKETS)?.last_processed_ts_iso, - ), - ); - await runPass(SCOPE_EVENTS, 'events', cursors, () => - runEventsBackfill( - client, - queue, - (c, t) => setCursor(SCOPE_EVENTS, c, t), - cursors.get(SCOPE_EVENTS)?.last_cursor, - cursors.get(SCOPE_EVENTS)?.last_processed_ts_iso, - ), - ); + await runPasses(client, queue, cursors, parallel); // Heartbeat on every cycle — even when all three passes are drained // (no walker ran). Without the bump an idle-but-healthy service @@ -160,6 +141,88 @@ export async function run(): Promise { incrementSuccess(serviceName); } +/** Build the three pass invocations + run them either sequentially or + * concurrently based on `parallel`. Exported for unit testing. */ +export async function runPasses( + client: KalshiClient, + queue: Queue, + cursors: Map, + parallel: boolean, +): Promise { + const passes: Array<{ + scope: string; + label: string; + body: () => Promise; + }> = [ + { + scope: SCOPE_TRADES, + label: 'trades', + body: () => + runTradesBackfill( + client, + queue, + (c, t) => setCursor(SCOPE_TRADES, c, t), + cursors.get(SCOPE_TRADES)?.last_cursor, + cursors.get(SCOPE_TRADES)?.last_processed_ts_iso, + ), + }, + { + scope: SCOPE_MARKETS, + label: 'markets', + body: () => + runMarketsBackfill( + client, + queue, + (c, t) => setCursor(SCOPE_MARKETS, c, t), + cursors.get(SCOPE_MARKETS)?.last_cursor, + cursors.get(SCOPE_MARKETS)?.last_processed_ts_iso, + ), + }, + { + scope: SCOPE_EVENTS, + label: 'events', + body: () => + runEventsBackfill( + client, + queue, + (c, t) => setCursor(SCOPE_EVENTS, c, t), + cursors.get(SCOPE_EVENTS)?.last_cursor, + cursors.get(SCOPE_EVENTS)?.last_processed_ts_iso, + ), + }, + ]; + + if (parallel) { + // `allSettled` so a failure in one pass doesn't cancel the others + // — passes write to independent CH tables + cursor scopes, so + // partial-cycle progress is durable per-pass. Aggregate failures + // afterward; the first one carries through with its pass label. + const results = await Promise.allSettled( + passes.map((p) => runPass(p.scope, p.label, cursors, p.body)), + ); + const failures = results + .map((r, i) => ({ r, label: passes[i]?.label ?? '' })) + .filter(({ r }) => r.status === 'rejected'); + if (failures.length > 1) { + for (const f of failures) { + const reason = (f.r as PromiseRejectedResult).reason; + log.error('parallel backfill pass failed', { + pass: f.label, + message: (reason as Error)?.message ?? String(reason), + }); + } + } + if (failures.length > 0) { + throw (failures[0]!.r as PromiseRejectedResult).reason; + } + return; + } + + for (const p of passes) { + await runPass(p.scope, p.label, cursors, p.body); + } +} + /** Run one pass with quarantine + drained short-circuits + per-pass error * tagging. */ async function runPass(