perf(backfill): drain owner history concurrently, not one owner at a time

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<Hex, string[]>
+): 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),

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<T, R>(
+  items: readonly T[],
+  limit: number,
+  worker: (item: T, index: number) => Promise<R>,
+): Promise<R[]> {
+  const results = new Array<R>(items.length);
+  let next = 0;
+
+  const runner = async (): Promise<void> => {
+    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;
+}

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_<chainId>.
  */
-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_<chainId>.
+ */
+export const DEFAULT_OWNER_BACKFILL_CONCURRENCY = 10;
 
 /**
  * Maximum number of TWAP parts that precomputeOrderUids will attempt to enumerate.

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);
+  });
+});