diff --git a/src/application/handlers/block/ownerBackfill.ts b/src/application/handlers/block/ownerBackfill.ts index 856ab42..6822f95 100644 --- a/src/application/handlers/block/ownerBackfill.ts +++ b/src/application/handlers/block/ownerBackfill.ts @@ -6,12 +6,14 @@ import { type SupportedChainId } from "../../../data"; import { BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS, DEFAULT_MAX_OWNERS_BACKFILL_PER_BLOCK, + DEFAULT_OWNER_BACKFILL_CONCURRENCY, } from "../../../constants"; import { fetchComposableOrders, upsertDiscreteOrders, } from "../../helpers/orderbookClient"; import { TimeoutError, withTimeout } from "../../helpers/withTimeout"; +import { mapWithConcurrency } from "../../helpers/concurrency"; import { log } from "../../helpers/logger"; import { NON_DETERMINISTIC_TYPES } from "../../../utils/order-types"; @@ -41,6 +43,60 @@ function resolveOwnerCap(chainId: number): number { : DEFAULT_MAX_OWNERS_BACKFILL_PER_BLOCK; } +function resolveOwnerConcurrency(chainId: number): number { + const raw = Number(process.env[`MAX_OWNERS_BACKFILL_CONCURRENCY_${chainId}`]); + return Number.isFinite(raw) && raw > 0 + ? raw + : DEFAULT_OWNER_BACKFILL_CONCURRENCY; +} + +// Drain one owner's full history: fetch (timeout-bounded), upsert discovered orders, +// and — only on a full drain — flip historyBackfilled. Returns the per-owner tallies +// so the concurrent caller can aggregate. A per-owner timeout is swallowed (the owner +// stays eligible and is retried next firing); any other error propagates to abort the +// batch (the block handler is idempotent, so the block simply retries). +async function drainOwner( + context: Context, + chainId: SupportedChainId, + currentBlock: bigint, + owner: Hex, + ownerGeneratorIds: Map +): Promise<{ discovered: number; drained: number }> { + try { + const { orders, complete } = await withTimeout( + fetchComposableOrders(context, chainId, owner), + BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS, + `OwnerBackfill:owner:${owner}` + ); + const discovered = await upsertDiscreteOrders(context, chainId, orders); + + // Only flip the flag when the owner's history was drained in full. A partial + // drain (rate limit / timeout) leaves the owner eligible → retried next block. + if (complete) { + await markOwnerHistoryBackfilled(context, chainId, owner, ownerGeneratorIds); + return { discovered, drained: 1 }; + } + + log("warn", "OwnerBackfill:owner_incomplete", { + block: String(currentBlock), + chainId, + owner, + }); + return { discovered, drained: 0 }; + } catch (err) { + if (err instanceof TimeoutError) { + log("warn", "OwnerBackfill:owner_timeout", { + block: String(currentBlock), + chainId, + owner, + timeoutMs: BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS, + }); + return { discovered: 0, drained: 0 }; // leave eligible — retried next block + } + throw err; + } +} + async function drainOwnerBatch( event: Event<"OwnerBackfillLive:block">, context: Context @@ -90,55 +146,25 @@ async function drainOwnerBatch( ownerGeneratorIds.set(row.owner, existing); } + const concurrency = resolveOwnerConcurrency(chainId); + log("info", "OwnerBackfill:START", { block: String(currentBlock), chainId, owners: owners.length, cap, + concurrency, }); - let discovered = 0; - let drained = 0; - - for (const owner of owners) { - try { - const { orders, complete } = await withTimeout( - fetchComposableOrders(context, chainId, owner), - BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS, - `OwnerBackfill:owner:${owner}` - ); - discovered += await upsertDiscreteOrders(context, chainId, orders); - - // Only flip the flag when the owner's history was drained in full. A partial - // drain (rate limit / timeout) leaves the owner eligible → retried next block. - if (complete) { - await markOwnerHistoryBackfilled( - context, - chainId, - owner, - ownerGeneratorIds - ); - drained++; - } else { - log("warn", "OwnerBackfill:owner_incomplete", { - block: String(currentBlock), - chainId, - owner, - }); - } - } catch (err) { - if (err instanceof TimeoutError) { - log("warn", "OwnerBackfill:owner_timeout", { - block: String(currentBlock), - chainId, - owner, - timeoutMs: BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS, - }); - continue; // leave eligible — retried next block - } - throw err; - } - } + // Owner fetches are independent HTTP round-trips, so run them concurrently + // (bounded) rather than one-at-a-time. Wall-clock per firing drops from + // cap × timeout to ~ceil(cap / concurrency) × timeout. + const tallies = await mapWithConcurrency(owners, concurrency, (owner) => + drainOwner(context, chainId, currentBlock, owner, ownerGeneratorIds) + ); + + const discovered = tallies.reduce((sum, t) => sum + t.discovered, 0); + const drained = tallies.reduce((sum, t) => sum + t.drained, 0); log("info", "OwnerBackfill:DONE", { block: String(currentBlock), diff --git a/src/application/helpers/concurrency.ts b/src/application/helpers/concurrency.ts new file mode 100644 index 0000000..f8ce6af --- /dev/null +++ b/src/application/helpers/concurrency.ts @@ -0,0 +1,23 @@ +/** + * Map `worker` over `items` with at most `limit` invocations in flight at once. + * Results are returned in input order regardless of completion order. + */ +export async function mapWithConcurrency( + items: readonly T[], + limit: number, + worker: (item: T, index: number) => Promise, +): Promise { + const results = new Array(items.length); + let next = 0; + + const runner = async (): Promise => { + while (next < items.length) { + const index = next++; + results[index] = await worker(items[index]!, index); + } + }; + + const poolSize = Math.min(limit, items.length); + await Promise.all(Array.from({ length: poolSize }, runner)); + return results; +} diff --git a/src/constants.ts b/src/constants.ts index 0e019da..7db1684 100644 --- a/src/constants.ts +++ b/src/constants.ts @@ -105,9 +105,25 @@ export const BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS = 30_000; * firing. Bounds the transaction size and the per-block orderbook request rate so * the historical drain spreads across live-sync blocks instead of one burst. * + * The drain runs inline on the realtime indexing path. Owner fetches now run + * concurrently (see DEFAULT_OWNER_BACKFILL_CONCURRENCY), so the per-firing wall-clock + * is roughly ceil(cap / concurrency) × BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS rather than + * cap × timeout — but keep the cap modest so a single firing can't stall event + * indexing when the orderbook API is slow. + * * Override per chain with env var MAX_OWNERS_BACKFILL_PER_BLOCK_. */ -export const DEFAULT_MAX_OWNERS_BACKFILL_PER_BLOCK = 100; +export const DEFAULT_MAX_OWNERS_BACKFILL_PER_BLOCK = 10; + +/** + * How many owner history fetches OwnerBackfill runs concurrently within a single + * firing. Bounds in-flight orderbook API load while collapsing the per-firing + * wall-clock: at concurrency >= cap, a firing takes ~one BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS + * worst case instead of cap × that. + * + * Override per chain with env var MAX_OWNERS_BACKFILL_CONCURRENCY_. + */ +export const DEFAULT_OWNER_BACKFILL_CONCURRENCY = 10; /** * Maximum number of TWAP parts that precomputeOrderUids will attempt to enumerate. diff --git a/tests/helpers/concurrency.test.ts b/tests/helpers/concurrency.test.ts new file mode 100644 index 0000000..0506c4b --- /dev/null +++ b/tests/helpers/concurrency.test.ts @@ -0,0 +1,60 @@ +import { describe, it, expect } from "vitest"; +import { mapWithConcurrency } from "../../src/application/helpers/concurrency"; + +describe("mapWithConcurrency", () => { + it("runs every item and returns results in input order", async () => { + const results = await mapWithConcurrency([1, 2, 3], 2, async (n) => n * 10); + expect(results).toEqual([10, 20, 30]); + }); + + it("never runs more than `limit` workers at once", async () => { + let active = 0; + let maxActive = 0; + const items = [1, 2, 3, 4, 5, 6]; + + const results = await mapWithConcurrency(items, 2, async (n) => { + active++; + maxActive = Math.max(maxActive, active); + await new Promise((r) => setTimeout(r, 5)); + active--; + return n; + }); + + expect(maxActive).toBeLessThanOrEqual(2); + expect(results).toEqual(items); + }); + + it("propagates a worker rejection", async () => { + const boom = new Error("boom"); + await expect( + mapWithConcurrency([1, 2, 3], 2, async (n) => { + if (n === 2) throw boom; + return n; + }), + ).rejects.toBe(boom); + }); + + it("with concurrency 1, a rejection halts the remaining items", async () => { + const started: number[] = []; + await expect( + mapWithConcurrency([1, 2, 3, 4, 5, 6], 1, async (n) => { + started.push(n); + if (n === 2) throw new Error("boom"); + return n; + }), + ).rejects.toThrow("boom"); + + // Single runner: the throw breaks its loop, so items 3..6 never start. + expect(started).toEqual([1, 2]); + }); + + it("returns an empty array without calling the worker for empty input", async () => { + let calls = 0; + const results = await mapWithConcurrency([], 3, async (n) => { + calls++; + return n; + }); + expect(results).toEqual([]); + expect(calls).toBe(0); + }); +});