diff --git a/.env.example b/.env.example
index f5a2895..914a226 100644
--- a/.env.example
+++ b/.env.example
@@ -45,6 +45,14 @@ DATABASE_SCHEMA=programmatic_orders
 # MAX_FLASH_LOAN_ORDERS_PER_BLOCK_1=200
 # MAX_FLASH_LOAN_ORDERS_PER_BLOCK_100=400
 
+# OwnerBackfill drain worker (src/worker/drain.ts — run with `pnpm drain`, or the `drain`
+# sidecar in docker-compose). Fetches non-deterministic owners' historical orders from the
+# orderbook into cow_cache; the indexer projects them into discreteOrder. Needs only
+# DATABASE_URL (same Postgres) — no DATABASE_SCHEMA, no RPC URLs. All tuning is optional:
+# DRAIN_OWNER_CONCURRENCY=10     # owners drained concurrently (rate limit is not binding)
+# DRAIN_IDLE_SLEEP_MS=5000       # backoff when the queue is empty
+# DRAIN_LEASE_TTL_MS=300000      # stale-lease reclaim window for crash recovery
+
 # eth_getLogs block range cap (optional; default: 1000)
 # Increase if your RPC provider supports a larger range to speed up backfill.
 # Override per chain with the numeric chain-id suffix:
diff --git a/agent_docs/orderbook-integration-flow.md b/agent_docs/orderbook-integration-flow.md
index 71af63d..014b2e3 100644
--- a/agent_docs/orderbook-integration-flow.md
+++ b/agent_docs/orderbook-integration-flow.md
@@ -72,23 +72,21 @@ The system has seven components. Each has a single responsibility.
 
 **Writes to**: `discreteOrder`, `cow_cache.order_uid_cache` (caches newly terminal).
 
-### Component OwnerBackfill (`block/ownerBackfill.ts`)
+### Component OwnerBackfill (`block/ownerBackfill.ts`) + drain worker (`worker/drain.ts`)
 
-**Responsibility**: Discovery of historical discrete orders for non-deterministic generators (the realtime poller only ever returns the *current* tradeable order, never past fulfilled/expired ones). Bounded batch per firing so the work spreads across blocks instead of one burst.
+**Responsibility**: Discovery of historical discrete orders for non-deterministic generators (the realtime poller only ever returns the *current* tradeable order, never past fulfilled/expired ones). Split into a **drain worker** (does the slow orderbook HTTP, out of band) and a **projection** Ponder handler (DB-only). This is the COW-1118 design: the drain used to be an inline block handler, which serialized 30 s `/account` calls into Ponder's single indexing slot and let a single un-drainable owner wedge `/readyz` forever.
 
-**Two registrations, one drain**:
-- **OwnerBackfill (historical)** — `startBlock` = ComposableCow start block, `endBlock: "latest"`, coarse interval. Runs *during* the event backfill, so the orderbook drain overlaps historical sync and is largely done by the time Ponder reaches the tip. Orders an owner creates after its drain are at blocks ahead of `"latest"` and get discovered by the realtime catch-up (poller → confirmer → status tracker) as Ponder processes those blocks — so draining early loses nothing.
-- **OwnerBackfillLive** — `startBlock: "latest"`, fine interval. Mops up owners created late in the backfill or not finished before the tip.
+**The `drain` worker** (`src/worker/drain.ts`, a standalone process — `pnpm drain` / the `drain` sidecar): loops claiming `pending` owners from `cow_cache.owner_drain_state` (`SELECT … FOR UPDATE SKIP LOCKED`, set `status='draining'`, `claimed_at=now`; a stale `draining` lease is reclaimed after `DRAIN_LEASE_TTL_MS`). It drains `DRAIN_OWNER_CONCURRENCY` owners at once. Per owner it **offset-walks** `/account/{owner}/orders` newest→oldest, committing each page to `cow_cache.composable_order` and bumping `next_offset`; a short page → `status='complete'`. Offset-based resume (not a `MAX(creation_date)` cursor) is required: with per-page commit, a max-cursor would declare the owner complete after re-reading only the newest page and drop the older pages. Re-fetches on restart are harmless (upsert by `order_uid`). It touches **only** `cow_cache` — never Ponder's versioned schema — so blue-green deploys and reindexes are invisible to it, and one worker serves every deployment. It computes everything from the orderbook response + static maps (`getOrderTypeFromHandler`, `COMPOSABLE_COW_HANDLER_ADDRESSES`, and the reconstructed `generator_hash`); it never looks up `eventId`.
 
-**How it works**: Each firing selects up to `MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>` (default 25) distinct owners with `status = 'Active'`, non-deterministic `orderType`, and `historyBackfilled = false`. For each, it calls `fetchComposableOrders(owner)` (which drains the owner's history incrementally — see below), upserts into `discreteOrder`, and sets `historyBackfilled = true` on that owner's generators **only if the drain completed in full**. A partial drain (rate limit / timeout) leaves the owner eligible → retried on a later block. No retry queue — the flag *is* the queue.
+**The projection** (`OwnerBackfill:block`, HTTP-free): one registration, `startBlock` = ComposableCow start block, **no `endBlock`** (so it fires through the backfill *and* into realtime — the old `OwnerBackfillLive` registration is gone, the worker handles late/live owners). Each firing selects up to `MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>` (default 25) distinct owners with `status='Active'`, non-deterministic `orderType`, `historyBackfilled=false`; keeps those that are `complete` in `owner_drain_state`; reads their cached rows, maps `generator_hash → the current eventId` via `remapToCurrentGenerators` (drops orphans), upserts into `discreteOrder`, and flips `historyBackfilled`. DB-only, cheap, and re-runs correctly after a reindex to repopulate a fresh `discreteOrder` from the warm cache.
 
-Eligibility is gated on the dedicated `conditionalOrderGenerator.historyBackfilled` flag, **not** on "has zero discrete orders". An active generator the realtime `OrderDiscoveryPoller` has already inserted a row for is still backfilled. The flag is set at generator creation (composableCow.ts) for the cases that never need a drain — deterministic types (precompute handles them) and generators created during live sync (owned by the poller from birth) — so the only `false` rows are non-deterministic historical generators.
+**Enqueue**: the `ConditionalOrderCreated` handler (composableCow.ts) inserts each non-deterministic owner into `owner_drain_state` (`'pending', 0`) `ON CONFLICT DO NOTHING` — so a reindex replaying history never resets a `complete` owner.
 
-**Why bounded + repeating**: at production scale there are thousands of non-deterministic generators; draining them all in one firing would exceed the orderbook rate limit and hold one giant transaction. Spreading a bounded batch per block is wall-clock-paced (rate-limit friendly) and keeps transactions small. Promotion readiness is gated on the drain completing (`/readyz`), so a blue-green deploy never promotes a pod with history still filling.
+Eligibility for the projection is the dedicated `conditionalOrderGenerator.historyBackfilled` flag, **not** "has zero discrete orders". The flag is set at generator creation for the cases that never need a drain — deterministic types (precompute handles them) and generators created during live sync (owned by the poller from birth) — so the only `false` rows are non-deterministic historical generators. `/readyz` gates promotion on the pending count reaching 0, satisfied only once the worker (drained) **and** the projection (flipped) both finish. Because the drain is a separate process, no owner can wedge the pipeline or readiness.
 
-**Cost across deploys**: Ponder rebuilds onchain tables from scratch on every schema-hash redeploy (`historyBackfilled` and `discreteOrder` included), so this handler re-runs each deploy. To avoid re-fetching an owner's entire history every time, the full composable-order rows are kept in the durable `cow_cache.composable_order` table (survives reindex) and only the delta newer than the cached high-water mark is fetched — see the Orderbook Client below.
+**`reconcile` dropped**: the drain no longer re-checks still-open cached rows via `/orders/by_uids` — `OrderStatusTracker:block` already polls every `discreteOrder` with `status='open'` regardless of source, so it was redundant.
 
-**Writes to**: `discreteOrder`, `conditionalOrderGenerator` (`historyBackfilled`), `cow_cache.composable_order`.
+**Writes to**: worker → `cow_cache.composable_order`, `cow_cache.owner_drain_state`. Projection → `discreteOrder`, `conditionalOrderGenerator` (`historyBackfilled`).
 
 ### Component FlashLoanOrderBackfiller + FlashLoanOrderEnricher (`block/flashLoanOrderBackfiller.ts`, `block/flashLoanOrderEnricher.ts`)
 
@@ -106,16 +104,12 @@ Eligibility is gated on the dedicated `conditionalOrderGenerator.historyBackfill
 
 **Responsibility**: The single interface to the CoW Protocol Orderbook API. All API calls go through this module. It handles fetching, filtering, EIP-1271 signature decoding, generator matching, and per-UID caching.
 
-**Public functions**:
-- `fetchComposableOrders(context, chainId, owner)` — incremental owner drain (see below), rebuilt from the durable cache
-- `fetchAccountOrders(apiBaseUrl, owner, maxPages?, signingScheme?, pageSize?, sinceCreationDate?)` — paginated `/account/{owner}/orders` fetch; with `sinceCreationDate` it stops once a (newest-first) page dips below the cursor
-- `fetchOrderStatusByUids(context, chainId, uids)` — batch UID status lookup with cache
-- `fetchFlashLoanEnrichmentByUids(context, chainId, uids)` — batch UID enrichment for flash-loan orders, cache-first
-- `upsertDiscreteOrders(context, chainId, orders)` — write to discreteOrder table
+The module is split so the drain worker (which runs outside the Ponder runtime, where the virtual `ponder:*` modules don't resolve) can share the fiddly decode/hash/cache logic without importing Ponder's schema:
+- **`orderbookHttp.ts`** (Ponder-free) — raw HTTP: `fetchOrderbook` (retry/backoff), `fetchAccountOrders`, `fetchAccountOrderPage` (single page at an offset, for the worker's offset-walk), `fetchOrdersByUids`.
+- **`composableCache.ts`** (Ponder-free) — the `cow_cache` tables (`composable_order`, `owner_drain_state`) + `filterAndProcessForCache` (decode EIP-1271 → reconstruct `generator_hash` → resolve type from the static handler map, **no DB**), `upsertComposableCache`, `readOwnerComposableCache`, the drain-state claim/complete/release helpers, and `createCowCacheTables` (the shared DDL, called by both `setup.ts` and the worker).
+- **`orderbookClient.ts`** (Ponder-side) — the part that touches Ponder's versioned schema: `fetchOrderStatusByUids`, `fetchFlashLoanEnrichmentByUids`, `upsertDiscreteOrders`, and `remapToCurrentGenerators` (the `generator_hash → current eventId` join the worker deliberately skips).
 
-**`fetchComposableOrders` — incremental drain**: rather than re-fetching the whole history each deploy, it reads `MAX(creation_date)` for the owner from `cow_cache.composable_order` as a cursor, fetches only orders newer than it (orders are returned newest-first, so pagination stops at the cursor), persists the delta, then rebuilds the **full** owner set from the durable cache. It re-checks any still-open cached rows via `/orders/by_uids`, and re-maps `generator_hash → the current generator eventId` (the eventId is per-deployment and changes each reindex; the hash is stable). First deploy full-drains; later deploys fetch only the delta.
-
-**Writes to**: `discreteOrder`, `cow_cache.composable_order`, `cow_cache.order_uid_cache`.
+**Writes to**: `discreteOrder`, `cow_cache.composable_order`, `cow_cache.owner_drain_state`, `cow_cache.order_uid_cache`.
 
 ### Component E: API Endpoints (`api/index.ts`)
 
@@ -181,18 +175,32 @@ Shared per-UID terminal-order cache. Survives Ponder resyncs (external `cow_cach
 
 ### `cow_cache.composable_order`
 
-Durable **full** composable-order rows for the OwnerBackfill incremental drain. Like `order_uid_cache` it lives in the external `cow_cache` schema and survives reindex — but it stores every field needed to rebuild a `discreteOrder` row (not just terminal status), so redeploys fetch only the delta newer than `MAX(creation_date)` per owner instead of the full history. It stores the stable `generator_hash` rather than the per-deployment `eventId`, which `fetchComposableOrders` re-maps to the current generator on read. Created in `setup.ts`.
+Durable **full** composable-order rows, filled by the drain worker and read by the OwnerBackfill projection. Like `order_uid_cache` it lives in the external `cow_cache` schema and survives reindex — but it stores every field needed to rebuild a `discreteOrder` row (not just terminal status), so a reindex repopulates `discreteOrder` from the warm cache with no orderbook calls. It stores the stable `generator_hash` rather than the per-deployment `eventId`, which the projection re-maps to the current generator on read. A deployment-independent **superset**: the worker never reads Ponder's schema, so it caches every composable order regardless of which deployment's generators exist. Created by `createCowCacheTables` (called from both `setup.ts` and the worker).
 
 | Column | Purpose |
 |--------|---------|
 | `chain_id`, `order_uid` | Primary key |
-| `owner` | Indexed (`chain_id`, `owner`) — cursor lookup + per-owner rebuild |
+| `owner` | Indexed (`chain_id`, `owner`) — per-owner read/upsert |
 | `generator_hash` | Stable `keccak256(handler, salt, staticInput)`; re-mapped to the current generator `eventId` on read |
-| `order_type` | Handler-derived order type |
+| `order_type` | Handler-derived order type (from the static map, no DB) |
 | `status`, `sell_amount`, `buy_amount`, `fee_amount`, `valid_to`, `executed_sell_amount`, `executed_buy_amount` | Full order fields to rebuild `discreteOrder` |
-| `creation_date` | Orderbook creation timestamp (seconds) — the incremental high-water cursor |
+| `creation_date` | Orderbook creation timestamp (seconds) |
 | `fetched_at` | When it was cached |
 
+### `cow_cache.owner_drain_state`
+
+Durable per-owner work queue coordinating the indexer and the drain worker (COW-1118). The `ConditionalOrderCreated` handler enqueues non-deterministic owners (`'pending'`); the worker claims and drains them (`'draining' → 'complete'`); the projection flips `historyBackfilled` once an owner is `'complete'`. Survives reindex, so a redeploy never re-drains a `'complete'` owner. Created by `createCowCacheTables`.
+
+| Column | Purpose |
+|--------|---------|
+| `chain_id`, `owner` | Primary key |
+| `status` | `'pending'` \| `'draining'` \| `'complete'` |
+| `next_offset` | Offset-walk resume point; committed per page so a restart resumes rather than restarts |
+| `claimed_at` | Lease timestamp; a stale `'draining'` lease is reclaimed after `DRAIN_LEASE_TTL_MS` |
+| `updated_at` | Last transition; claim ordering (oldest first) |
+
+Indexed on (`status`, `updated_at`) for the claim query.
+
 ---
 
 ## 3. Detailed Flows
@@ -283,30 +291,24 @@ Durable **full** composable-order rows for the OwnerBackfill incremental drain.
 
 4. **Expire by validTo.** Any `discreteOrder` where `status = 'open'` and `validTo <= currentTimestamp` → set to `expired`.
 
-### 3.6 OwnerBackfill — Historical Discovery (per-block, bounded)
+### 3.6 OwnerBackfill — Historical Discovery (worker drain + projection)
 
-**When**: During the event backfill (historical handler, `startBlock` = ComposableCow start block, coarse interval) and from the tip onward (live handler, `startBlock: "latest"`, fine interval). Both run the same drain.
+**Drain worker** (`src/worker/drain.ts`, out of band):
 
-1. **Select a bounded batch.** Up to `MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>` (default 25) distinct owners with `status = 'Active'` AND non-deterministic `orderType` AND `historyBackfilled = false`, ordered by owner. (Gated on the flag, **not** on "no discreteOrder rows" — so generators the realtime poller already touched are still backfilled.)
+1. **Claim.** `SELECT … FOR UPDATE SKIP LOCKED` up to `DRAIN_OWNER_CONCURRENCY` owners from `owner_drain_state` that are `pending` (or `draining` with a stale lease), set `status='draining'`, `claimed_at=now`.
+2. **Offset-walk.** For each claimed owner, `GET /account/{owner}/orders` newest→oldest, one page (1000) at a time from `next_offset`; decode + reconstruct `generator_hash` (`filterAndProcessForCache`, no DB), commit the page to `cow_cache.composable_order`, and bump `next_offset`. A short page → `status='complete'`. A transient error returns the owner to `pending` at the last committed `next_offset`.
+3. **Sleep** when no owner is claimable.
 
-2. **Fetch by owner.** For each, call `fetchComposableOrders(owner)`. This drains the owner's history incrementally (delta since the durable-cache cursor), rebuilds the full set, matches to generators, and upserts into `discreteOrder`. Set `historyBackfilled = true` on that owner's generators **only if the drain returned complete**; otherwise leave it eligible for a later block.
+**Projection** (`OwnerBackfill:block`, DB-only, one registration `startBlock` = ComposableCow start block, no `endBlock`):
 
-3. **Repeat.** The next firing takes the next batch (drained owners have dropped out). When no eligible owner remains, the query is a cheap no-op. Readiness (`/readyz`) turns green once the pending count hits 0.
+1. **Select a bounded batch.** Up to `MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>` (default 25) distinct owners with `status='Active'` AND non-deterministic `orderType` AND `historyBackfilled = false`, ordered by owner. (Gated on the flag, **not** on "no discreteOrder rows".)
+2. **Keep the drained ones.** Filter to owners `complete` in `owner_drain_state` — partial owners stay pending.
+3. **Project.** Read each ready owner's cached rows, map `generator_hash → the current eventId` (drops orphans), upsert into `discreteOrder`, flip `historyBackfilled` on that owner's generators.
+4. **Repeat.** When no eligible owner remains, the query is a cheap no-op. `/readyz` turns green once the pending count hits 0 — satisfied only once worker (drained) + projection (flipped) both finish.
 
 ### 3.7 Orderbook Client — Fetch & Cache Logic
 
-The Orderbook Client is used by all other components. Two main entry points:
-
-**`fetchComposableOrders(context, chainId, owner)`** — Incremental owner drain:
-
-1. Resolve API URL from `ORDERBOOK_API_URLS[chainId]`
-2. Read the cursor: `MAX(creation_date)` for this owner from `cow_cache.composable_order` (undefined ⇒ full drain)
-3. `GET /account/{owner}/orders` (paginated, 1000/page, newest-first) fetching only the delta — pagination **stops** once a page dips below the cursor
-4. Filter: keep `signingScheme = "eip1271"`, skip `presignaturePending`; decode EIP-1271 signatures → match to generators by hash
-5. Persist the delta into `cow_cache.composable_order`
-6. Rebuild the **full** owner set from `cow_cache.composable_order` (delta + all older rows)
-7. Re-check any still-open cached rows via `POST /orders/by_uids`; re-persist any that became terminal
-8. Re-map `generator_hash → current generator eventId` (drops rows whose generator no longer exists) and return `ComposableOrder[]`
+The Orderbook Client is used by all other components. Main entry points:
 
 **`fetchOrderStatusByUids(context, chainId, uids)`** — Batch UID lookup:
 
@@ -374,7 +376,7 @@ The Orderbook Client is used by all other components. Two main entry points:
 ### Scenario C: PerpetualSwap, created 3 months ago (non-deterministic)
 
 1. Backfill: Generator created, `status = 'Active'`. UID Pre-computation returns null.
-2. **OwnerBackfill at live sync start**: Finds this generator has no `discreteOrder` rows. Fetches by owner from API. Discovers 15 historical orders. Upserts all into `discreteOrder`.
+2. **Drain worker (in parallel with backfill)**: the owner was enqueued at generator creation; the worker fetches its history from the API (15 historical orders) into `cow_cache.composable_order` and marks it `complete`. The **OwnerBackfill projection** then upserts all 15 into `discreteOrder` and flips `historyBackfilled`.
 3. **OrderDiscoveryPoller at live sync**: Polls `getTradeableOrderWithSignature`. Gets `Success` during active window → candidate created.
 4. **CandidateConfirmer**: Checks API → confirms → `discreteOrder`.
 5. **OrderStatusTracker**: Tracks open orders → `fulfilled` when settled.
@@ -383,7 +385,7 @@ The Orderbook Client is used by all other components. Two main entry points:
 ### Scenario D: Unknown order type, created during backfill
 
 1. Backfill: Generator created with `orderType = 'Unknown'`. UID Pre-computation returns null.
-2. **OwnerBackfill**: Fetches by owner. If the API has composable orders for this owner, they're upserted into `discreteOrder`.
+2. **Drain worker + OwnerBackfill projection**: the worker fetches the owner's history into `cow_cache.composable_order`; the projection then upserts them into `discreteOrder`.
 3. **OrderDiscoveryPoller**: May poll if generator is still Active. Gets `Success` or error responses.
 4. Normal lifecycle from there.
 
@@ -405,11 +407,11 @@ A **boolean** is the correct type. It answers one question: "does OrderDiscovery
 
 ### OwnerBackfill — Completeness (Resolved)
 
-The bootstrap discovers orders via `fetchComposableOrders(owner)`, which relies on the Orderbook API having the orders. If an order never reached the API, the bootstrap won't discover it.
+The drain worker discovers orders from `/account/{owner}/orders`, which relies on the Orderbook API having the orders. If an order never reached the API, the worker won't discover it.
 
 **This is correct behavior, not a gap.** All CoW Protocol orders go through the Orderbook API. An order that never reached the API is the same as an order that never existed from the protocol's perspective — it was never submitted to solvers and never had a chance to be settled. The watch-tower is the standard submission path and is operated by the CoW Protocol team.
 
-**History depth:** backfill is gated on the `historyBackfilled` flag (independent of whether the generator already has discrete orders) and drains the owner's full `/account/{owner}/orders` history at 1000/page — so an active owner's entire history is discovered, not just the most recent page. Redeploys stay cheap via the incremental `cow_cache.composable_order` cursor (delta-only fetch).
+**History depth:** eligibility is gated on the `historyBackfilled` flag (independent of whether the generator already has discrete orders); the worker offset-walks the owner's full `/account/{owner}/orders` history at 1000/page — so an active owner's entire history is discovered, not just the most recent page. Redeploys stay cheap because the durable `cow_cache.composable_order` + `owner_drain_state` survive reindex, so a `complete` owner is never re-drained.
 
 ---
 
@@ -425,7 +427,7 @@ flowchart TB
         BH1["Block Handler 1<br/><i>OrderDiscoveryPoller — every block</i>"]
         BH2["Block Handler 2<br/><i>CandidateConfirmer — every block</i>"]
         BH3["Block Handler 3<br/><i>OrderStatusTracker — every block</i>"]
-        BH4["Block Handler 4<br/><i>OwnerBackfill — per block (bounded)</i>"]
+        BH4["Block Handler 4<br/><i>OwnerBackfill projection — per block (DB-only)</i>"]
         GPV["GPv2Settlement<br/><i>flash loans only</i>"]
         CSF["CoWShedFactory"]
     end
diff --git a/agent_docs/project-structure.md b/agent_docs/project-structure.md
index c54ff6b..a90597e 100644
--- a/agent_docs/project-structure.md
+++ b/agent_docs/project-structure.md
@@ -11,7 +11,11 @@
 | `schema/relations.ts` | Drizzle relations: `transaction → many(conditionalOrderGenerators)`, `conditionalOrderGenerator → one(transaction) + many(discreteOrders)`, `discreteOrder → one(conditionalOrderGenerator)` |
 | `src/data.ts` | Contract addresses + start blocks per chain; exports `ComposableCowContract` consumed by `ponder.config.ts` |
 | `src/api/index.ts` | Hono app exposing `/sql/*` (Ponder SQL client) and `/` + `/graphql` (GraphQL) |
-| `src/application/handlers/composableCow.ts` | `ConditionalOrderCreated` event handler — computes hash, upserts `transaction` row, inserts `conditionalOrderGenerator` row |
+| `src/application/handlers/composableCow.ts` | `ConditionalOrderCreated` event handler — computes hash, upserts `transaction` row, inserts `conditionalOrderGenerator` row, enqueues non-deterministic owners into `cow_cache.owner_drain_state` |
+| `src/application/handlers/block/ownerBackfill.ts` | OwnerBackfill projection (DB-only) — projects `complete` owners' cached rows into `discreteOrder` and flips `historyBackfilled` |
+| `src/application/helpers/orderbookHttp.ts` | Ponder-free orderbook HTTP client (retry/backoff, account pagination, by_uids) — shared by the indexer and the drain worker |
+| `src/application/helpers/composableCache.ts` | Ponder-free `cow_cache` tables (`composable_order`, `owner_drain_state`) + decode/hash + drain-state helpers + shared DDL |
+| `src/worker/drain.ts` | Standalone OwnerBackfill drain worker (`pnpm drain`) — fetches owner history into `cow_cache`, coordinating with the indexer only via `cow_cache` (COW-1118) |
 | `abis/ComposableCowAbi.ts` | Full ComposableCoW ABI (all events + functions) |
 | `vite.config.ts` | Vite config with `tsconfigPaths` plugin for Ponder bundler |
 | `tsconfig.json` | TypeScript config (`moduleResolution: bundler`, strict, ES2022) |
@@ -129,6 +133,7 @@ src/data.ts
 ```bash
 pnpm dev          # Start Ponder in development mode (hot reload)
 pnpm start        # Start Ponder in production mode
+pnpm drain        # Start the OwnerBackfill drain worker (needs DATABASE_URL; separate process)
 pnpm codegen      # Generate ponder-env.d.ts (run after config/schema changes)
 pnpm typecheck    # tsc --noEmit
 pnpm lint         # ESLint on all .ts files
diff --git a/docker-compose.yml b/docker-compose.yml
index 2070ee0..a07387a 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -75,6 +75,38 @@ services:
         max-file: "5"
     profiles: [deploy]
 
+  # OwnerBackfill drain worker — a long-lived sidecar that fetches non-deterministic
+  # owners' historical orders from the orderbook and fills cow_cache. Separate process so
+  # the slow HTTP drain never serializes into Ponder's single indexing slot (COW-1118).
+  # Touches ONLY the cow_cache schema (never Ponder's versioned schema), so it is
+  # deployment-independent: one worker serves every blue-green deploy and survives reindex.
+  # It needs no DATABASE_SCHEMA and no RPC URLs — only DATABASE_URL and outbound HTTPS to
+  # api.cow.fi. Reuses the same image; just a different command.
+  drain:
+    image: ${PROJECT_PREFIX:?error}-ponder:${APP_REVISION:?error}
+    restart: unless-stopped
+    build:
+      context: .
+      dockerfile: Dockerfile
+      args:
+        PIPELINE_BUILD_TAG: ${APP_REVISION}
+    command: ["pnpm", "drain"]
+    environment:
+      DATABASE_URL: postgresql://${POSTGRES_USER:-postgres}:${POSTGRES_PASSWORD:-postgres}@postgres:5432/${POSTGRES_DB:-programmatic-orders} # ggignore
+      # Optional tuning (defaults in src/constants.ts):
+      # DRAIN_OWNER_CONCURRENCY: "10"
+      # DRAIN_IDLE_SLEEP_MS: "5000"
+      # DRAIN_LEASE_TTL_MS: "300000"
+    depends_on:
+      postgres:
+        condition: service_healthy
+    logging:
+      driver: json-file
+      options:
+        max-size: "50m"
+        max-file: "5"
+    profiles: [deploy]
+
 volumes:
   postgres-data:
     name: ${PROJECT_PREFIX:-local}-postgres-data
diff --git a/docs/architecture.md b/docs/architecture.md
index b06b13b..6e70d29 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -58,7 +58,8 @@ composableCow.ts handler        cowshed.ts handler             settlement.ts han
                                            cascades parent Cancelled to orphan candidates
                  OrderStatusTracker      — polls API for status updates on open discrete orders;
                                            cascades parent Cancelled to orphan open rows
-                 OwnerBackfill           — per-block bounded backfill of non-deterministic historical orders
+                 OwnerBackfill           — DB-only projection of drained non-deterministic history
+                                           (the orderbook drain runs in a separate worker — src/worker/drain.ts)
                  CancellationWatcher     — periodic singleOrders() read for deterministic generators;
                                            flips to Cancelled when remove() has been called on-chain
                         |
@@ -182,7 +183,7 @@ Stats (total settlements, trade logs found, EOA skips, adapter mappings, avg FAC
 
 ### block/ -- live block handlers
 
-The block handlers live in `src/application/handlers/block/` (one file per handler); `blockHandler.ts` is a thin barrel that imports them so their `ponder.on(...)` registrations run. Most run only during live sync (`startBlock: "latest"`) to avoid hammering the orderbook API during historical backfill — the exception is **OwnerBackfill (historical)**, which deliberately runs *during* the event backfill so the owner-history drain overlaps sync and completes before the tip (its live counterpart, OwnerBackfillLive, continues from `"latest"`). `OrderDiscoveryPoller` and `CancellationWatcher` share a per-chain batch cap (`MAX_GENERATORS_PER_BLOCK_<chainId>`, default 200) and pull from a priority queue ordered by oldest `lastCheckBlock` first. Generators past the cap defer to the next block.
+The block handlers live in `src/application/handlers/block/` (one file per handler); `blockHandler.ts` is a thin barrel that imports them so their `ponder.on(...)` registrations run. Most run only during live sync (`startBlock: "latest"`) to avoid hammering the orderbook API during historical backfill — the exception is the **OwnerBackfill projection**, which runs from the ComposableCow start block through realtime (no `endBlock`) but is DB-only, so it never touches the orderbook (the drain itself is a separate worker process — see below). `OrderDiscoveryPoller` and `CancellationWatcher` share a per-chain batch cap (`MAX_GENERATORS_PER_BLOCK_<chainId>`, default 200) and pull from a priority queue ordered by oldest `lastCheckBlock` first. Generators past the cap defer to the next block.
 
 **OrderDiscoveryPoller** (every block, mainnet + gnosis): Multicalls `getTradeableOrderWithSignature` on ComposableCoW for each `Active` generator where `allCandidatesKnown=false`. A success result creates a `candidateDiscreteOrder` entry. A `SingleOrderNotAuthed` error marks the generator as `Cancelled` with `lastPollResult='cancelled:SingleOrderNotAuthed'`. Other errors (tryNextBlock, tryAtEpoch, etc.) advance the generator's `nextCheckBlock` accordingly. Single-shot non-deterministic types (GoodAfterTime, TradeAboveThreshold) set `allCandidatesKnown=true` after first success.
 
@@ -196,12 +197,12 @@ The block handlers live in `src/application/handlers/block/` (one file per handl
 
 **FlashLoanOrderEnricher** (every block, mainnet + gnosis): Steady-state enrichment for orders that settle *during* live sync, plus any stragglers the backfiller left (timeouts / not-yet-on-API). Selects pending rows (`enrichedAt IS NULL`, oldest `blockNumber` first) up to `MAX_FLASH_LOAN_ORDERS_PER_BLOCK_<chainId>` (default 200). Both handlers share one enrichment routine: batch-fetch via `/orders/by_uids` (cache-first against `cow_cache.order_uid_cache`, the shared per-UID cache, which survives reindex), upsert the orderbook fields + `enrichedAt` on hits, and bump `enrichmentAttempts` on misses until `MAX_FLASH_LOAN_ENRICHMENT_ATTEMPTS` (then left permanently un-enriched rather than polled forever).
 
-**OwnerBackfill** (two handlers, mainnet + gnosis): Fetches historical orders for non-deterministic generators (the realtime poller only ever returns the *current* tradeable order, never past ones). Each firing drains a bounded batch of owners (`MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>`, default 25), so the drain spreads across blocks (rate-limit friendly) rather than one burst, and no single transaction holds thousands of owners.
+**OwnerBackfill** (a projection handler + a standalone worker, mainnet + gnosis): Fetches historical orders for non-deterministic generators (the realtime poller only ever returns the *current* tradeable order, never past ones). Split into two so the slow orderbook HTTP never runs inside Ponder's single indexing slot (COW-1118):
 
-- **OwnerBackfill (historical)** runs *during* the event backfill (`startBlock` = the ComposableCow start block, coarse interval, `endBlock: "latest"`), so the orderbook drain overlaps historical sync and is largely done by the tip. Orders an owner creates after its drain are at blocks ahead of `"latest"` and are picked up by the realtime catch-up (`OrderDiscoveryPoller` → `CandidateConfirmer` → `OrderStatusTracker`) as Ponder processes those blocks — the same path as any live order — so draining early loses nothing.
-- **OwnerBackfillLive** runs from `"latest"` onward at a fine cadence, mopping up owners created late in the backfill or not finished before the tip.
+- **Drain worker** (`src/worker/drain.ts`, a separate process — the `drain` sidecar): claims owners from `cow_cache.owner_drain_state` (`FOR UPDATE SKIP LOCKED` + stale-lease reclaim), offset-walks each owner's full `/account/{owner}/orders` history at 1000/page committing every page to `cow_cache.composable_order`, and marks the owner `complete`. It touches **only** `cow_cache` — never Ponder's versioned schema — so one worker serves every blue-green deploy and a reindex is invisible to it. Runs in parallel with the event backfill, so time-to-ready ≈ `max(backfill, drain)`.
+- **OwnerBackfill projection** (`block/ownerBackfill.ts`, DB-only): one registration, `startBlock` = the ComposableCow start block, **no `endBlock`** (fires through backfill and into realtime). Each firing takes a bounded batch of pending owners (`MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>`, default 25), keeps those the worker has marked `complete`, projects their cached rows into `discreteOrder` (mapping the stable `generator_hash → the current eventId`, dropping orphans), and flips `historyBackfilled`.
 
-Eligibility is gated on the `conditionalOrderGenerator.historyBackfilled` flag — **not** on "has no discrete orders yet" — so a generator the realtime poller already inserted rows for is still backfilled. The flag is `true` at creation for generators that never need a drain (deterministic types, and generators created during live sync), and is flipped to `true` only after an owner's history is drained **in full** (a partial drain from rate-limit/timeout leaves the owner eligible → retried next block). Per owner it drains the full `/account/{owner}/orders` history at 1000/page. Because Ponder rebuilds onchain tables on every schema-hash redeploy, the full composable-order rows are kept in the durable `cow_cache.composable_order` table (survives reindex) and only the delta newer than `MAX(creation_date)` is fetched each deploy; the rows store the stable `generator_hash` and are re-mapped to the current generator `eventId` on read. Promotion readiness is gated on this drain completing — see `/readyz` in deployment.md.
+The `ConditionalOrderCreated` handler enqueues each non-deterministic owner into `owner_drain_state` (`ON CONFLICT DO NOTHING`, so a reindex never resets a `complete` owner). Eligibility for the projection is gated on the `conditionalOrderGenerator.historyBackfilled` flag — **not** on "has no discrete orders yet". The flag is `true` at creation for generators that never need a drain (deterministic types, and generators created during live sync). Because `cow_cache` survives reindex, a redeploy re-projects `discreteOrder` from the warm cache with no new HTTP and never re-drains a `complete` owner. Promotion readiness is gated on the pending count reaching 0 — satisfied only once worker (drained) and projection (flipped) both finish; no single owner can wedge it — see `/readyz` in deployment.md.
 
 **CancellationWatcher** (every block, mainnet + gnosis): Closes `OrderDiscoveryPoller`'s blind spot. `OrderDiscoveryPoller` skips `allCandidatesKnown=true` generators, so removals via `ComposableCoW.remove()` on deterministic types (TWAP, StopLoss, CirclesBackingOrder) would otherwise go undetected — `remove()` emits no event. `CancellationWatcher` multicalls `singleOrders(owner, hash)` on a per-generator cadence of `DETERMINISTIC_CANCEL_SWEEP_INTERVAL` blocks (default 100). A `false` return means the owner called `remove()` on-chain: the generator is flipped to `Cancelled` with `lastPollResult='cancelled:removeMapping'`, after which `CandidateConfirmer` and `OrderStatusTracker`'s parent-cancelled cascades reconcile the children on the next block. `true` reschedules the next check.
 
diff --git a/docs/deployment.md b/docs/deployment.md
index ae839aa..e186e71 100644
--- a/docs/deployment.md
+++ b/docs/deployment.md
@@ -34,7 +34,10 @@ Example: `DATABASE_URL=postgresql://cow_programmatic:secretpass@localhost:5433/c
 |----------|----------|-------------|
 | `MAX_GENERATORS_PER_BLOCK_<chainId>` | No | Per-block cap on how many generators `OrderDiscoveryPoller` and `CancellationWatcher` will touch on the given chain (e.g. `MAX_GENERATORS_PER_BLOCK_1=200`, `MAX_GENERATORS_PER_BLOCK_100=400`). Default is 200. Excess generators defer to the next block, prioritized by oldest `lastCheckBlock` first. |
 | `MAX_DISCRETE_ORDERS_PER_BLOCK_<chainId>` | No | Per-block cap on how many open discrete orders `OrderStatusTracker` will check on the given chain (e.g. `MAX_DISCRETE_ORDERS_PER_BLOCK_1=200`). Default is 200. Excess orders are deferred to the next block, prioritised by oldest `promotedAt` first. |
-| `MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>` | No | Per-block cap on how many distinct owners `OwnerBackfill` drains on the given chain (e.g. `MAX_OWNERS_BACKFILL_PER_BLOCK_1=25`). Default is 25. Bounds the per-block orderbook request rate and transaction size while the historical drain spreads across live-sync blocks. |
+| `MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>` | No | Per-firing cap on how many pending owners the `OwnerBackfill` **projection** processes on the given chain (e.g. `MAX_OWNERS_BACKFILL_PER_BLOCK_1=25`). Default is 25. The projection is DB-only (the orderbook drain runs in the separate `drain` worker), so this just bounds the transaction size. |
+| `DRAIN_OWNER_CONCURRENCY` | No | (drain worker) How many owners it drains concurrently. Default 10. The orderbook rate limit is not the binding constraint, so a modest fan-out is safe. |
+| `DRAIN_IDLE_SLEEP_MS` | No | (drain worker) Backoff when the queue has no claimable owners. Default 5000. |
+| `DRAIN_LEASE_TTL_MS` | No | (drain worker) Stale-lease reclaim window — a crashed worker's `draining` owner is re-claimed after this. Default 300000. |
 | `DISABLE_SETTLEMENT_FACTORY_CHECK` | No | Skips `getCode` + `FACTORY()` RPC calls in the GPv2Settlement handler. Useful for benchmarking base sync throughput. |
 | `PINO_LOG_LEVEL` | No | Log verbosity: `debug`, `info`, `warn`, `error`. Defaults to Ponder's built-in default. |
 
@@ -82,7 +85,7 @@ deployment/
   deploy-remotely.ts       # Rsync + SSH deploy to a remote host
 ```
 
-The deploy services (`postgres-deploy` and `ponder`) live in the root `docker-compose.yml` under the `deploy` profile. Start them with:
+The deploy services (`postgres`, `ponder`, and `drain`) live in the root `docker-compose.yml` under the `deploy` profile. Start them with:
 
 ```bash
 docker compose --profile deploy up -d
@@ -90,6 +93,8 @@ docker compose --profile deploy up -d
 
 The `Dockerfile` in the project root builds the Ponder image: two-stage Node 22 Alpine, installs dependencies with `--frozen-lockfile`, exposes port 3000, runs `pnpm start`. The health check hits `/ready` with a 24-hour start period (initial sync takes hours).
 
+The **`drain`** service reuses the same image with `command: ["pnpm", "drain"]` — the OwnerBackfill drain worker (`src/worker/drain.ts`). It needs only `DATABASE_URL` (the same Postgres) and outbound HTTPS to `api.cow.fi` — no `DATABASE_SCHEMA`, no RPC URLs, since it touches only the deployment-independent `cow_cache` schema. **One worker serves every blue-green deploy** — it does not need to be redeployed per Ponder revision, though redeploying it alongside is harmless (its work queue and cache are durable). See "Historical Discrete Order Gap" below for how it splits work with the OwnerBackfill projection.
+
 ### Kubernetes Probes
 
 The indexer exposes two health endpoints with distinct semantics:
@@ -181,17 +186,17 @@ A fresh deployment (no prior `ponder_sync` cache) reindexes from the configured
 
 | Phase | Typical duration | Notes |
 |-------|-----------------|-------|
-| Event backfill | 4–10 hours | Fetches `eth_getLogs` from start block to tip. Bottleneck is RPC throughput; a generous RPC endpoint shortens this. OwnerBackfill (historical) drains owner history *during* this phase, so it overlaps sync rather than running after it. |
-| Live-sync catch-up | 5–15 minutes | Most block handlers (OrderDiscoveryPoller, CandidateConfirmer, OrderStatusTracker, OwnerBackfillLive, CancellationWatcher) run at "latest". Stale TWAP candidates drain at 500/block; OwnerBackfillLive mops up any owners not drained during backfill. |
-| Full data completeness | Gated by `/readyz` | All generators have candidates or discrete orders; every non-deterministic owner's history is drained (`historyBackfilled` complete). `/readyz` turns 200 only here — use it as the promotion probe. |
+| Event backfill | 4–10 hours | Fetches `eth_getLogs` from start block to tip. Bottleneck is RPC throughput; a generous RPC endpoint shortens this. The `drain` worker fetches owner history *in parallel* with this phase (separate process), so time-to-ready ≈ `max(backfill, drain)` rather than `backfill + drain`. |
+| Live-sync catch-up | 5–15 minutes | Most block handlers (OrderDiscoveryPoller, CandidateConfirmer, OrderStatusTracker, OwnerBackfill projection, CancellationWatcher) run at "latest". Stale TWAP candidates drain at 500/block; the OwnerBackfill projection flips owners the worker has finished draining. |
+| Full data completeness | Gated by `/readyz` | All generators have candidates or discrete orders; every non-deterministic owner's history is drained (by the worker) and projected (`historyBackfilled` complete). `/readyz` turns 200 only here — use it as the promotion probe. |
 
 A reindex that reuses an existing `ponder_sync` cache (same chain, same start blocks) skips the event backfill and completes in minutes.
 
 ### `/ready` vs `/readyz` Semantics
 
-`GET /ready` (Ponder built-in) returns `200` when Ponder has processed all historical blocks up to the tip and the live indexer is running. It does **not** guarantee historical discrete-order data is complete — OwnerBackfill drains that across subsequent live blocks.
+`GET /ready` (Ponder built-in) returns `200` when Ponder has processed all historical blocks up to the tip and the live indexer is running. It does **not** guarantee historical discrete-order data is complete — the `drain` worker fills that and the OwnerBackfill projection flips it across subsequent live blocks.
 
-`GET /readyz` (app) returns `200` only when Ponder is synced **and** the owner backfill has finished (no non-deterministic historical generator with `historyBackfilled = false`). This is the promotion gate: it guarantees a newly-promoted pod has the full historical discrete-order set, so blue-green promotion never drops a complete pod for one that's still filling. It returns `503` (with the pending count) while the drain is in progress.
+`GET /readyz` (app) returns `200` only when Ponder is synced **and** the owner backfill has finished (no non-deterministic historical generator with `historyBackfilled = false`). The count is satisfied only once the worker has drained an owner *and* the projection has flipped it. This is the promotion gate: it guarantees a newly-promoted pod has the full historical discrete-order set, so blue-green promotion never drops a complete pod for one that's still filling. It returns `503` (with the pending count) while the drain is in progress. Because the drain runs in a separate process, no single un-drainable owner can wedge the indexing pipeline (the previous inline design could — COW-1118).
 
 During backfill both return `503`. GraphQL queries are still available but data is incomplete (generators and transactions accumulate; discrete orders fill in as live sync progresses).
 
@@ -205,8 +210,11 @@ Block handlers only run during live sync. TWAP parts computed during backfill la
 
 **Residual gap**: Orders that no longer appear in `/account/{owner}/orders` (beyond the CoW API's retention window) will be recorded as `expired` regardless of their actual fill status. This affects only very old orders for users with a large order history.
 
-Non-deterministic generators (PerpetualSwap, GoodAfterTime, TradeAboveThreshold, fee-burners, CoW AMM, Unknown) are handled by OwnerBackfill, which drains each owner's full `/account/{owner}/orders` history and upserts discovered orders directly into `discrete_order`. It runs as a repeating live-sync handler draining a bounded batch of owners per block (`MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>`, default 25), so a large owner population spreads across blocks instead of one burst; `/readyz` gates promotion until it's done.
+Non-deterministic generators (PerpetualSwap, GoodAfterTime, TradeAboveThreshold, fee-burners, CoW AMM, Unknown) are handled by the split of the **`drain` worker** and the **OwnerBackfill projection**:
+
+- **`drain` worker** (`src/worker/drain.ts`, the `drain` sidecar) — a long-running process, separate from the indexer, that claims pending owners from `cow_cache.owner_drain_state`, offset-walks each owner's full `/account/{owner}/orders` history, and commits every page to the durable `cow_cache.composable_order` table. It touches **only** `cow_cache` — never Ponder's versioned schema — so it is deployment-independent: one worker serves every blue-green deploy, and a reindex is invisible to it. It is safe to run N-wide (claims use `FOR UPDATE SKIP LOCKED` + stale-lease reclaim).
+- **OwnerBackfill projection** (`block/ownerBackfill.ts`) — a DB-only Ponder block handler that projects a fully-drained (`complete`) owner's cached rows into `discrete_order` (mapping the stable `generator_hash → the current eventId`) and flips `historyBackfilled`. It processes a bounded batch per firing (`MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>`, default 25).
 
-**Redeploy cost**: Ponder rebuilds onchain tables from scratch on every schema-hash deploy, so OwnerBackfill re-runs each time. To avoid re-fetching an owner's entire history per deploy, the full composable-order rows are kept in the durable `cow_cache.composable_order` table (external schema, survives reindex). On redeploy only the delta newer than the cached `MAX(creation_date)` is fetched; the rest is rebuilt from the cache. The first-ever deploy (empty cache) does the full drain, which is the dominant cost and the main thing `/readyz` waits on.
+**Redeploy cost**: Ponder rebuilds onchain tables from scratch on every schema-hash deploy, so the projection re-runs each time — but it is DB-only and **the worker never re-drains a `complete` owner** (its cache and drain state survive reindex). The first-ever deploy (empty cache) does the full drain in the worker, in parallel with the event backfill; `/readyz` waits on `max(backfill, drain)`.
 
 
diff --git a/package.json b/package.json
index e54fb1a..256815b 100644
--- a/package.json
+++ b/package.json
@@ -6,6 +6,7 @@
   "scripts": {
     "dev": "ponder dev",
     "start": "ponder start -p 3000 --schema ${DATABASE_SCHEMA:-public} --log-format json",
+    "drain": "tsx src/worker/drain.ts",
     "db": "ponder db",
     "codegen": "ponder codegen",
     "lint": "eslint . --ext .ts",
@@ -21,6 +22,7 @@
     "@hono/zod-openapi": "^0.19.10",
     "drizzle-orm": "0.41.0",
     "hono": "^4.5.0",
+    "pg": "^8.11.0",
     "ponder": "^0.16.6",
     "ponder-enrich-gql-docs-middleware": "^0.1.3",
     "viem": "^2.21.3",
@@ -28,6 +30,7 @@
   },
   "devDependencies": {
     "@types/node": "^20.9.0",
+    "@types/pg": "^8.11.0",
     "eslint": "^8.53.0",
     "eslint-config-ponder": "^0.16.2",
     "tsx": "^4.0.0",
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index 5bb5c42..03cb9dd 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -24,13 +24,16 @@ importers:
         version: 0.19.10(hono@4.12.3)(zod@3.25.76)
       drizzle-orm:
         specifier: 0.41.0
-        version: 0.41.0(@electric-sql/pglite@0.2.13)(@opentelemetry/api@1.9.0)(kysely@0.26.3)(pg@8.19.0)
+        version: 0.41.0(@electric-sql/pglite@0.2.13)(@opentelemetry/api@1.9.0)(@types/pg@8.20.0)(kysely@0.26.3)(pg@8.19.0)
       hono:
         specifier: ^4.5.0
         version: 4.12.3
+      pg:
+        specifier: ^8.11.0
+        version: 8.19.0
       ponder:
         specifier: ^0.16.6
-        version: 0.16.6(patch_hash=f0612e7102efb92911aeae79416c3c5c52fcaf41c4a22e7903ba90ef8aa952d5)(@opentelemetry/api@1.9.0)(@types/node@20.19.35)(hono@4.12.3)(typescript@5.9.3)(viem@2.46.3(typescript@5.9.3)(zod@3.25.76))(zod@3.25.76)
+        version: 0.16.6(patch_hash=f0612e7102efb92911aeae79416c3c5c52fcaf41c4a22e7903ba90ef8aa952d5)(@opentelemetry/api@1.9.0)(@types/node@20.19.35)(@types/pg@8.20.0)(hono@4.12.3)(typescript@5.9.3)(viem@2.46.3(typescript@5.9.3)(zod@3.25.76))(zod@3.25.76)
       ponder-enrich-gql-docs-middleware:
         specifier: ^0.1.3
         version: 0.1.3(graphql@16.8.2)(hono@4.12.3)
@@ -44,6 +47,9 @@ importers:
       '@types/node':
         specifier: ^20.9.0
         version: 20.19.35
+      '@types/pg':
+        specifier: ^8.11.0
+        version: 8.20.0
       eslint:
         specifier: ^8.53.0
         version: 8.57.1
@@ -1011,6 +1017,9 @@ packages:
   '@types/normalize-package-data@2.4.4':
     resolution: {integrity: sha512-37i+OaWTh9qeK4LSHPsyRC7NahnGotNuZvjLSgcPzblpHB3rrCJxAOgI5gCdKm7coonsaX1Of0ILiTcnZjbfxA==}
 
+  '@types/pg@8.20.0':
+    resolution: {integrity: sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==}
+
   '@types/semver@7.7.1':
     resolution: {integrity: sha512-FmgJfu+MOcQ370SD0ev7EI8TlCAfKYU+B4m5T3yXc1CiRN94g/SZPtsCkk506aUDtlMnFZvasDwHHUcZUEaYuA==}
 
@@ -3598,6 +3607,12 @@ snapshots:
 
   '@types/normalize-package-data@2.4.4': {}
 
+  '@types/pg@8.20.0':
+    dependencies:
+      '@types/node': 20.19.35
+      pg-protocol: 1.12.0
+      pg-types: 2.2.0
+
   '@types/semver@7.7.1': {}
 
   '@typescript-eslint/eslint-plugin@6.21.0(@typescript-eslint/parser@6.21.0(eslint@8.57.1)(typescript@5.9.3))(eslint@8.57.1)(typescript@5.9.3)':
@@ -3982,10 +3997,11 @@ snapshots:
 
   dotenv@16.6.1: {}
 
-  drizzle-orm@0.41.0(@electric-sql/pglite@0.2.13)(@opentelemetry/api@1.9.0)(kysely@0.26.3)(pg@8.19.0):
+  drizzle-orm@0.41.0(@electric-sql/pglite@0.2.13)(@opentelemetry/api@1.9.0)(@types/pg@8.20.0)(kysely@0.26.3)(pg@8.19.0):
     optionalDependencies:
       '@electric-sql/pglite': 0.2.13
       '@opentelemetry/api': 1.9.0
+      '@types/pg': 8.20.0
       kysely: 0.26.3
       pg: 8.19.0
 
@@ -4849,7 +4865,7 @@ snapshots:
       graphql: 16.8.2
       hono: 4.12.3
 
-  ponder@0.16.6(patch_hash=f0612e7102efb92911aeae79416c3c5c52fcaf41c4a22e7903ba90ef8aa952d5)(@opentelemetry/api@1.9.0)(@types/node@20.19.35)(hono@4.12.3)(typescript@5.9.3)(viem@2.46.3(typescript@5.9.3)(zod@3.25.76))(zod@3.25.76):
+  ponder@0.16.6(patch_hash=f0612e7102efb92911aeae79416c3c5c52fcaf41c4a22e7903ba90ef8aa952d5)(@opentelemetry/api@1.9.0)(@types/node@20.19.35)(@types/pg@8.20.0)(hono@4.12.3)(typescript@5.9.3)(viem@2.46.3(typescript@5.9.3)(zod@3.25.76))(zod@3.25.76):
     dependencies:
       '@babel/code-frame': 7.29.0
       '@commander-js/extra-typings': 12.1.0(commander@12.1.0)
@@ -4866,7 +4882,7 @@ snapshots:
       dataloader: 2.2.3
       detect-package-manager: 3.0.2
       dotenv: 16.6.1
-      drizzle-orm: 0.41.0(@electric-sql/pglite@0.2.13)(@opentelemetry/api@1.9.0)(kysely@0.26.3)(pg@8.19.0)
+      drizzle-orm: 0.41.0(@electric-sql/pglite@0.2.13)(@opentelemetry/api@1.9.0)(@types/pg@8.20.0)(kysely@0.26.3)(pg@8.19.0)
       glob: 10.5.0
       graphql: 16.8.2
       graphql-yoga: 5.17.1(graphql@16.8.2)
diff --git a/ponder.config.ts b/ponder.config.ts
index aeaa5dc..907875c 100644
--- a/ponder.config.ts
+++ b/ponder.config.ts
@@ -146,36 +146,29 @@ export default createConfig({
       ),
       interval: 1,
     },
-    // OwnerBackfill (historical) — drains non-deterministic generators' history during
-    // the event backfill (startBlock = composableCow start block), so the orderbook drain
-    // overlaps historical sync and is largely done by the tip. Coarse interval keeps the
-    // idle-block overhead low; each firing drains a bounded batch
-    // (MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>). Ends at "latest"; OwnerBackfillLive
-    // continues from there. Readiness is gated on completion (/readyz).
+    // OwnerBackfill — projection-only (HTTP-free). Projects fully-drained owners'
+    // cached composable orders (cow_cache.composable_order) into discreteOrder and flips
+    // historyBackfilled. The orderbook drain itself runs in the standalone worker
+    // (src/worker/drain.ts), not here (COW-1118).
+    //
+    // One registration, no endBlock: it fires through the historical backfill AND
+    // continues into realtime, so owners the worker completes after the tip still get
+    // projected (the old OwnerBackfillLive registration is gone). Coarse interval keeps
+    // the per-firing overhead low across the long backfill; because each firing is a
+    // cheap DB-only query, a completed owner is projected within one interval, bounding
+    // its contribution to time-to-ready — it can never wedge readiness.
     OwnerBackfill: {
       chain: Object.fromEntries(
         ACTIVE_CHAINS.map((c) => [
           c.name,
           {
             startBlock: c.composableCow.startBlock,
-            endBlock: "latest" as const,
             interval: 250,
           },
         ])
       ),
       interval: 1,
     },
-    // OwnerBackfillLive — same drain, from the tip onward at a fine cadence: mops up
-    // owners created late in the backfill and any not finished before the tip.
-    OwnerBackfillLive: {
-      chain: Object.fromEntries(
-        ACTIVE_CHAINS.map((c) => [
-          c.name,
-          { startBlock: "latest" as const, interval: c.blockTime < 8 ? 10 : 4 },
-        ])
-      ),
-      interval: 1,
-    },
     // CancellationWatcher — singleOrders() mapping read for deterministic generators
     // (allCandidatesKnown=true). Cadence per generator is DETERMINISTIC_CANCEL_SWEEP_INTERVAL
     // blocks; the handler itself is cheap when nothing is due.
diff --git a/src/application/handlers/block/ownerBackfill.ts b/src/application/handlers/block/ownerBackfill.ts
index 5a8ba93..7881b37 100644
--- a/src/application/handlers/block/ownerBackfill.ts
+++ b/src/application/handlers/block/ownerBackfill.ts
@@ -3,54 +3,46 @@ import { conditionalOrderGenerator } from "ponder:schema";
 import { and, eq, inArray } from "ponder";
 import type { Hex } from "viem";
 import { type SupportedChainId } from "../../../data";
-import {
-  BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS,
-  DEFAULT_MAX_OWNERS_BACKFILL_PER_BLOCK,
-} from "../../../constants";
-import { fetchComposableOrders, upsertDiscreteOrders } from "../../helpers/orderbookClient";
-import { TimeoutError, withTimeout } from "../../helpers/withTimeout";
+import { DEFAULT_MAX_OWNERS_BACKFILL_PER_BLOCK } from "../../../constants";
+import { remapToCurrentGenerators, upsertDiscreteOrders } from "../../helpers/orderbookClient";
+import { readOwnerComposableCache, selectCompleteDrainOwners } from "../../helpers/composableCache";
 import { log } from "../../helpers/logger";
 import { NON_DETERMINISTIC_TYPES } from "../../../utils/order-types";
 
-// ─── OwnerBackfill ───────────────────────────────────────────────────────────
-// Discovers historical discrete orders for non-deterministic generators (the realtime
-// poller only ever returns the *current* tradeable order, never past ones). Each firing
-// drains a bounded batch of not-yet-backfilled owners, so the work spreads across blocks
-// (rate-limit friendly) and no single transaction holds thousands of owners.
+// ─── OwnerBackfill (projection-only) ───────────────────────────────────────────
+// Projects historical discrete orders for non-deterministic generators from the durable
+// cow_cache.composable_order table into discreteOrder. HTTP-free: the standalone drain
+// worker (src/worker/drain.ts) does the orderbook fetching and fills the cache; this
+// handler only reads the cache and flips historyBackfilled once an owner is fully drained.
 //
-// Two registrations share this drain:
-//   - OwnerBackfill (historical): startBlock = composableCow start block, coarse interval.
-//     Runs *during* the event backfill so the drain overlaps historical sync and is
-//     largely done by the time Ponder reaches the tip — orders an owner creates after its
-//     drain are at blocks ahead of "latest" and get picked up by the realtime catch-up
-//     (OrderDiscoveryPoller etc.) exactly as any live order would.
-//   - OwnerBackfillLive: startBlock = "latest", fine interval. Mops up owners created late
-//     in the backfill and any that weren't finished before the tip.
+// Two consequences of moving the drain out of the pipeline (COW-1118):
+//   - No HTTP here means no owner can park the single indexing slot for 30 s, and no
+//     un-drainable owner can wedge readiness — the worker resumes a fat owner across
+//     restarts via next_offset, then this projection flips the flag.
+//   - It re-runs correctly after a reindex: the warm cache survives, so a fresh
+//     discreteOrder table is repopulated from cache with no new orderbook calls.
 //
-// Readiness is gated on the drain completing (see /readyz), so promotion never ships an
-// indexer with history still missing.
+// One registration (startBlock = composableCow start block, no endBlock) covers both the
+// historical backfill and realtime — the old OwnerBackfillLive registration is gone; the
+// worker, not a second block handler, handles late/live owners. The coarse interval keeps
+// per-firing overhead low; a completed owner is projected within one interval, so a
+// post-tip straggler adds at most one interval to time-to-ready — it can never wedge.
 //
-// Eligibility is the historyBackfilled flag, set at generator creation for the cases
-// that never need a drain (deterministic types, and generators created live) — see
-// composableCow.ts. So the only false rows are non-deterministic historical generators,
-// which this handler flips to true once their owner is fully drained.
+// Eligibility is the historyBackfilled flag, set at generator creation for the cases that
+// never need a drain (deterministic types, and generators created live) — see
+// composableCow.ts. The only false rows are non-deterministic historical generators; this
+// handler flips them to true once their owner is 'complete' in owner_drain_state.
 
 function resolveOwnerCap(chainId: number): number {
   const raw = Number(process.env[`MAX_OWNERS_BACKFILL_PER_BLOCK_${chainId}`]);
   return Number.isFinite(raw) && raw > 0 ? raw : DEFAULT_MAX_OWNERS_BACKFILL_PER_BLOCK;
 }
 
-// Shared drain — registered for both the historical and live block handlers below.
-async function drainOwnerBatch(
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  event: any,
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  context: any,
-  phase: "historical" | "live",
-): Promise<void> {
+ponder.on("OwnerBackfill:block", async ({ event, context }) => {
   const chainId = context.chain.id as SupportedChainId;
   const currentBlock = event.block.number;
   const cap = resolveOwnerCap(chainId);
+  const db = context.db.sql;
 
   const eligibleWhere = and(
     eq(conditionalOrderGenerator.chainId, chainId),
@@ -59,27 +51,32 @@ async function drainOwnerBatch(
     eq(conditionalOrderGenerator.historyBackfilled, false),
   );
 
-  // Take up to `cap` distinct owners this block; ordering by owner keeps progress
-  // deterministic and lets already-drained owners fall out of the set.
-  const ownerRows = (await context.db.sql
+  // Pending owners (the readiness set — shrinks to 0). Ordering by owner keeps progress
+  // deterministic and lets already-projected owners fall out of the set.
+  const ownerRows = (await db
     .selectDistinct({ owner: conditionalOrderGenerator.owner })
     .from(conditionalOrderGenerator)
     .where(eligibleWhere)
     .orderBy(conditionalOrderGenerator.owner)
     .limit(cap)) as { owner: Hex }[];
 
-  if (ownerRows.length === 0) return; // nothing pending — cheap no-op every block
+  if (ownerRows.length === 0) return; // nothing pending — cheap no-op every firing
 
   const owners = ownerRows.map((r) => r.owner);
 
-  // Generator ids for the selected owners, to flip historyBackfilled after a clean drain.
-  const genRows = (await context.db.sql
+  // Only project owners the worker has fully drained; partially-drained owners stay pending.
+  const complete = await selectCompleteDrainOwners(db, chainId, owners);
+  const readyOwners = owners.filter((o) => complete.has(o.toLowerCase()));
+  if (readyOwners.length === 0) return; // worker hasn't finished any of these yet
+
+  // Generator ids for the ready owners, to flip historyBackfilled after projection.
+  const genRows = (await db
     .select({
       generatorId: conditionalOrderGenerator.eventId,
       owner: conditionalOrderGenerator.owner,
     })
     .from(conditionalOrderGenerator)
-    .where(and(eligibleWhere, inArray(conditionalOrderGenerator.owner, owners)))) as {
+    .where(and(eligibleWhere, inArray(conditionalOrderGenerator.owner, readyOwners)))) as {
     generatorId: string;
     owner: Hex;
   }[];
@@ -91,48 +88,27 @@ async function drainOwnerBatch(
     ownerGeneratorIds.set(row.owner, existing);
   }
 
-  log("info", "OwnerBackfill:START", { block: String(currentBlock), chainId, phase, owners: owners.length, cap });
-
-  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;
-    }
+  let projected = 0;
+  for (const owner of readyOwners) {
+    // Read the owner's cached history and map the stable generator_hash → the current
+    // per-deployment eventId (drops orphans whose generator isn't in this deployment).
+    const cachedRows = await readOwnerComposableCache(db, chainId, owner);
+    const orders = await remapToCurrentGenerators(db, chainId, cachedRows);
+    projected += await upsertDiscreteOrders(context, chainId, orders);
+    await markOwnerHistoryBackfilled(context, chainId, owner, ownerGeneratorIds);
   }
 
-  log("info", "OwnerBackfill:DONE", { block: String(currentBlock), chainId, phase, owners: owners.length, drained, discovered });
-}
-
-// Historical: runs during the event backfill (startBlock = composableCow start block).
-ponder.on("OwnerBackfill:block", ({ event, context }) => drainOwnerBatch(event, context, "historical"));
-
-// Live: runs from "latest" onward, mopping up late/unfinished owners.
-ponder.on("OwnerBackfillLive:block", ({ event, context }) => drainOwnerBatch(event, context, "live"));
+  log("info", "OwnerBackfill:PROJECTED", {
+    block: String(currentBlock),
+    chainId,
+    pending: owners.length,
+    ready: readyOwners.length,
+    projected,
+  });
+});
 
 // Mark every eligible generator of this owner as history-backfilled so it drops out
-// of the eligibility set (and the readiness count). Set only after a full drain.
+// of the eligibility set (and the readiness count). Set only after projection.
 async function markOwnerHistoryBackfilled(
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
   context: any,
diff --git a/src/application/handlers/composableCow.ts b/src/application/handlers/composableCow.ts
index dbc1e9f..349abf2 100644
--- a/src/application/handlers/composableCow.ts
+++ b/src/application/handlers/composableCow.ts
@@ -45,6 +45,7 @@ import {
 import { encodeAbiParameters, keccak256, type Hex } from "viem";
 import { getOrderTypeFromHandler, isNonDeterministic, type OrderType } from "../../utils/order-types";
 import { decodeStaticInput } from "../../decoders/index";
+import { enqueueOwnerDrain } from "../helpers/composableCache";
 import { precomputeAndDiscover } from "../helpers/uidPrecompute";
 import { CirclesBackingOrderAbi } from "../../../abis/CirclesBackingOrderAbi";
 import { log } from "../helpers/logger";
@@ -233,6 +234,15 @@ async function insertGenerator(
     })
     .onConflictDoNothing();
 
+  // Enqueue the owner for the standalone drain worker. HTTP-free — just a durable INSERT.
+  // ON CONFLICT DO NOTHING never resets an already-'complete' owner when a reindex replays
+  // history. Both historical and live non-deterministic owners are enqueued: the worker is
+  // the single source of the durable composable cache, and the projection only flips the
+  // (historical) generators whose historyBackfilled is still false.
+  if (isNonDeterministic(orderType)) {
+    await enqueueOwnerDrain(context.db.sql, chainId, ownerAddress);
+  }
+
   return { ownerAddress, chainId, decodedParams, orderType };
 }
 
diff --git a/src/application/handlers/setup.ts b/src/application/handlers/setup.ts
index 2f66b4e..195434d 100644
--- a/src/application/handlers/setup.ts
+++ b/src/application/handlers/setup.ts
@@ -1,5 +1,6 @@
 import { ponder } from "ponder:registry";
 import { sql } from "ponder";
+import { createCowCacheTables } from "../helpers/composableCache";
 import { log } from "../helpers/logger";
 
 /**
@@ -16,8 +17,10 @@ import { log } from "../helpers/logger";
  *   - Open orders: not cached — always re-fetched
  */
 ponder.on("ComposableCow:setup", async ({ context }) => {
-  // Create a separate schema that Ponder's per-deployment schema management won't touch.
-  await context.db.sql.execute(sql`CREATE SCHEMA IF NOT EXISTS cow_cache`);
+  // Create the cow_cache schema + the tables the drain worker owns (composable_order,
+  // owner_drain_state). Shared with src/worker/drain.ts so table existence is not
+  // startup-ordering-dependent — either process can create them first.
+  await createCowCacheTables(context.db.sql);
 
   // Per-UID cache of terminal order data, keyed by (chain_id, order_uid). Used by
   // both the discrete-order path (status + executed amounts) and the flash-loan
@@ -51,40 +54,10 @@ ponder.on("ComposableCow:setup", async ({ context }) => {
   // dedicated table if a prior build created it.
   await context.db.sql.execute(sql`DROP TABLE IF EXISTS cow_cache.flash_loan_order_cache`);
 
-  // Durable per-owner composable-order rows, keyed by (chain_id, order_uid). Unlike
-  // order_uid_cache (per-UID terminal status only), this holds every field needed to
-  // rebuild a discreteOrder row without re-hitting the orderbook — so OwnerBackfill
-  // drains only the delta newer than MAX(creation_date) on each redeploy instead of
-  // the owner's full history. generator_hash (not the per-deployment eventId) is stored
-  // so rows survive reindex and re-map to the current generator by hash.
-  await context.db.sql.execute(sql`
-    CREATE TABLE IF NOT EXISTS cow_cache.composable_order (
-      chain_id              INTEGER NOT NULL,
-      order_uid             TEXT NOT NULL,
-      owner                 TEXT NOT NULL,
-      generator_hash        TEXT NOT NULL,
-      order_type            TEXT NOT NULL,
-      status                TEXT NOT NULL,
-      sell_amount           TEXT NOT NULL,
-      buy_amount            TEXT NOT NULL,
-      fee_amount            TEXT NOT NULL,
-      valid_to              INTEGER,
-      creation_date         BIGINT NOT NULL,
-      executed_sell_amount   TEXT,
-      executed_buy_amount    TEXT,
-      fetched_at            BIGINT NOT NULL,
-      PRIMARY KEY (chain_id, order_uid)
-    )
-  `);
-  await context.db.sql.execute(sql`
-    CREATE INDEX IF NOT EXISTS composable_order_owner_idx
-      ON cow_cache.composable_order (chain_id, owner)
-  `);
-
   // Log surviving cache entries — non-zero means cache persisted across restart/resync
   const result = await context.db.sql.execute(
     sql`SELECT COUNT(*)::int AS count FROM cow_cache.order_uid_cache`,
-  ) as { count: number }[];
+  ) as unknown as { count: number }[];
   const count = result[0]?.count ?? 0;
 
   log("info", "setup:cacheReady", { count, entries: `${count} entr${count === 1 ? "y" : "ies"} from previous run` });
diff --git a/src/application/helpers/composableCache.ts b/src/application/helpers/composableCache.ts
new file mode 100644
index 0000000..2d18495
--- /dev/null
+++ b/src/application/helpers/composableCache.ts
@@ -0,0 +1,409 @@
+/**
+ * Durable composable-order cache + drain-state coordination — Ponder-free.
+ *
+ * Lives in the `cow_cache` schema, which is NOT versioned by Ponder, so these rows
+ * survive `ponder start` redeploys and reindexes. Deliberately free of `ponder` /
+ * `ponder:schema` imports so both the Ponder handlers (via orderbookClient.ts) and
+ * the standalone drain worker (src/worker/drain.ts) can use it — the worker runs
+ * outside the Ponder runtime where the virtual `ponder:*` modules don't resolve.
+ *
+ * Two tables:
+ *   - composable_order: a deployment-independent superset of every composable order,
+ *     keyed by (chain_id, order_uid) and by the stable generator_hash (never the
+ *     per-deployment generator eventId). The drain worker fills it; the OwnerBackfill
+ *     projection reads it and maps hash → the current eventId.
+ *   - owner_drain_state: the durable work queue coordinating the indexer and the worker.
+ *     The ConditionalOrderCreated handler enqueues owners ('pending'); the worker claims
+ *     and drains them ('draining' → 'complete'); the projection flips historyBackfilled
+ *     once an owner is 'complete'.
+ */
+
+import { and, eq, inArray, sql } from "drizzle-orm";
+import { pgSchema, integer, text, bigint } from "drizzle-orm/pg-core";
+import { encodeAbiParameters, keccak256, type Hex } from "viem";
+import { getOrderTypeFromHandler, type OrderType } from "../../utils/order-types";
+import { COMPOSABLE_COW_HANDLER_ADDRESSES } from "../../data";
+import { SIGNING_SCHEME_EIP1271 } from "../../constants";
+import { decodeEip1271Signature } from "../decoders/erc1271Signature";
+import { log } from "./logger";
+import type { OrderbookOrder } from "./orderbookHttp";
+
+// A Drizzle handle — Ponder passes `context.db.sql`; the worker passes its own
+// node-postgres `drizzle(pool)`. Both expose the query-builder + `.execute`; we use
+// `any` (as the surrounding handler code does) to avoid coupling to either driver's
+// generic types.
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+export type DrizzleHandle = any;
+
+export const cowCacheSchema = pgSchema("cow_cache");
+
+/** Durable full composable-order rows (survives reindex). See createCowCacheTables for the DDL. */
+export const composableOrderCache = cowCacheSchema.table("composable_order", {
+  chainId: integer("chain_id").notNull(),
+  orderUid: text("order_uid").notNull(),
+  owner: text("owner").notNull(),
+  generatorHash: text("generator_hash").notNull(),
+  orderType: text("order_type").notNull(),
+  status: text("status").notNull(),
+  sellAmount: text("sell_amount").notNull(),
+  buyAmount: text("buy_amount").notNull(),
+  feeAmount: text("fee_amount").notNull(),
+  validTo: integer("valid_to"),
+  creationDate: bigint("creation_date", { mode: "bigint" }).notNull(),
+  executedSellAmount: text("executed_sell_amount"),
+  executedBuyAmount: text("executed_buy_amount"),
+  fetchedAt: bigint("fetched_at", { mode: "bigint" }).notNull(),
+});
+
+/** Durable per-owner drain work queue. See createCowCacheTables for the DDL. */
+export const ownerDrainState = cowCacheSchema.table("owner_drain_state", {
+  chainId: integer("chain_id").notNull(),
+  owner: text("owner").notNull(),
+  status: text("status").notNull(), // 'pending' | 'draining' | 'complete'
+  nextOffset: integer("next_offset").notNull(),
+  claimedAt: bigint("claimed_at", { mode: "bigint" }),
+  updatedAt: bigint("updated_at", { mode: "bigint" }).notNull(),
+});
+
+/** Durable-cache row shape for cow_cache.composable_order (owner passed separately). */
+export interface ComposableCacheRow {
+  orderUid: string;
+  generatorHash: string;
+  orderType: OrderType;
+  status: string;
+  sellAmount: string;
+  buyAmount: string;
+  feeAmount: string;
+  validTo: number | null;
+  creationDate: bigint;
+  executedSellAmount: string | null;
+  executedBuyAmount: string | null;
+}
+
+const TERMINAL_STATUSES = new Set(["fulfilled", "expired", "cancelled"]);
+
+/** `db.execute()` returns a plain array on Ponder's handle but a pg QueryResult
+ *  ({ rows }) on node-postgres. Normalize to the row array. */
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+function rowsOf(result: any): any[] {
+  if (Array.isArray(result)) return result;
+  return result?.rows ?? [];
+}
+
+// ─── Decode + hash (shared by the worker's drain) ─────────────────────────────
+
+/**
+ * Filter raw API orders to composable EIP-1271, decode signatures, reconstruct the
+ * generator hash, and resolve the order type from the static handler map — no DB.
+ *
+ * The cache is keyed by (stable) generator_hash, not the per-deployment eventId, so
+ * this needs no generator lookup: the OwnerBackfill projection does the hash → eventId
+ * join (and drops orphans) later. That keeps the worker off Ponder's versioned schema.
+ */
+export function filterAndProcessForCache(
+  chainId: number,
+  apiOrders: OrderbookOrder[],
+): ComposableCacheRow[] {
+  const results: ComposableCacheRow[] = [];
+
+  for (const order of apiOrders) {
+    if (order.signingScheme !== SIGNING_SCHEME_EIP1271) continue;
+    if (order.status === "presignaturePending") continue;
+
+    const decoded = decodeEip1271Signature(order.signature as Hex);
+    if (!decoded) continue;
+
+    if (!COMPOSABLE_COW_HANDLER_ADDRESSES.has(decoded.handler)) continue;
+
+    // Reproduce the same hash stored in conditionalOrderGenerator.hash.
+    const generatorHash = keccak256(
+      encodeAbiParameters(
+        [
+          {
+            type: "tuple",
+            components: [
+              { name: "handler", type: "address" },
+              { name: "salt", type: "bytes32" },
+              { name: "staticInput", type: "bytes" },
+            ],
+          },
+        ],
+        [{ handler: decoded.handler, salt: decoded.salt, staticInput: decoded.staticInput }],
+      ),
+    );
+
+    results.push({
+      orderUid: order.uid,
+      generatorHash,
+      orderType: getOrderTypeFromHandler(decoded.handler, chainId),
+      status: order.status,
+      sellAmount: order.sellAmount,
+      buyAmount: order.buyAmount,
+      feeAmount: order.feeAmount,
+      validTo: order.validTo,
+      creationDate: BigInt(Math.floor(new Date(order.creationDate).getTime() / 1000)),
+      executedSellAmount: order.executedSellAmount,
+      executedBuyAmount: order.executedBuyAmount,
+    });
+  }
+
+  return results;
+}
+
+// ─── composable_order cache ────────────────────────────────────────────────────
+
+/** Upsert durable composable rows; excluded status/validTo/executed overwrite on conflict. */
+export async function upsertComposableCache(
+  db: DrizzleHandle,
+  chainId: number,
+  owner: Hex,
+  rows: ComposableCacheRow[],
+): Promise<void> {
+  if (rows.length === 0) return;
+  const now = BigInt(Math.floor(Date.now() / 1000));
+  try {
+    await db
+      .insert(composableOrderCache)
+      .values(rows.map((r) => ({
+        chainId,
+        orderUid: r.orderUid,
+        owner: owner.toLowerCase(),
+        generatorHash: r.generatorHash,
+        orderType: r.orderType,
+        status: r.status,
+        sellAmount: r.sellAmount,
+        buyAmount: r.buyAmount,
+        feeAmount: r.feeAmount,
+        validTo: r.validTo,
+        creationDate: r.creationDate,
+        executedSellAmount: r.executedSellAmount,
+        executedBuyAmount: r.executedBuyAmount,
+        fetchedAt: now,
+      })))
+      .onConflictDoUpdate({
+        target: [composableOrderCache.chainId, composableOrderCache.orderUid],
+        set: {
+          status: sql`excluded.status`,
+          validTo: sql`excluded.valid_to`,
+          executedSellAmount: sql`excluded.executed_sell_amount`,
+          executedBuyAmount: sql`excluded.executed_buy_amount`,
+          fetchedAt: now,
+        },
+      });
+  } catch (err) {
+    log("warn", "ob:composableCacheWriteFailed", { chainId, rows: rows.length, err: String(err) });
+  }
+}
+
+/** All durably-cached composable rows for an owner. */
+export async function readOwnerComposableCache(
+  db: DrizzleHandle,
+  chainId: number,
+  owner: Hex,
+): Promise<ComposableCacheRow[]> {
+  try {
+    return (await db
+      .select({
+        orderUid: composableOrderCache.orderUid,
+        generatorHash: composableOrderCache.generatorHash,
+        orderType: composableOrderCache.orderType,
+        status: composableOrderCache.status,
+        sellAmount: composableOrderCache.sellAmount,
+        buyAmount: composableOrderCache.buyAmount,
+        feeAmount: composableOrderCache.feeAmount,
+        validTo: composableOrderCache.validTo,
+        creationDate: composableOrderCache.creationDate,
+        executedSellAmount: composableOrderCache.executedSellAmount,
+        executedBuyAmount: composableOrderCache.executedBuyAmount,
+      })
+      .from(composableOrderCache)
+      .where(
+        and(
+          eq(composableOrderCache.chainId, chainId),
+          eq(composableOrderCache.owner, owner.toLowerCase()),
+        ),
+      )) as ComposableCacheRow[];
+  } catch {
+    return [];
+  }
+}
+
+// ─── owner_drain_state work queue ────────────────────────────────────────────────
+
+/** A claimed drain job returned by claimDrainOwners. */
+export interface DrainClaim {
+  chainId: number;
+  owner: Hex;
+  nextOffset: number;
+}
+
+/**
+ * Enqueue an owner for a historical drain (idempotent). ON CONFLICT DO NOTHING so a
+ * reindex replaying ConditionalOrderCreated never resets an already-'complete' owner
+ * back to 'pending'. Called from the Ponder handler with `context.db.sql`.
+ */
+export async function enqueueOwnerDrain(
+  db: DrizzleHandle,
+  chainId: number,
+  owner: Hex,
+): Promise<void> {
+  const now = BigInt(Math.floor(Date.now() / 1000));
+  await db
+    .insert(ownerDrainState)
+    .values({ chainId, owner: owner.toLowerCase(), status: "pending", nextOffset: 0, claimedAt: null, updatedAt: now })
+    .onConflictDoNothing();
+}
+
+/** Owners that are fully drained ('complete'), among the given candidate owners. */
+export async function selectCompleteDrainOwners(
+  db: DrizzleHandle,
+  chainId: number,
+  owners: Hex[],
+): Promise<Set<string>> {
+  if (owners.length === 0) return new Set();
+  const rows = (await db
+    .select({ owner: ownerDrainState.owner })
+    .from(ownerDrainState)
+    .where(
+      and(
+        eq(ownerDrainState.chainId, chainId),
+        eq(ownerDrainState.status, "complete"),
+        inArray(ownerDrainState.owner, owners.map((o) => o.toLowerCase())),
+      ),
+    )) as { owner: string }[];
+  return new Set(rows.map((r) => r.owner));
+}
+
+/**
+ * Atomically claim up to `batch` drainable owners: those 'pending', plus 'draining'
+ * owners whose lease went stale (claimed before `staleBeforeSec` — crash recovery).
+ * FOR UPDATE SKIP LOCKED so multiple worker instances never claim the same owner,
+ * making the worker safe to scale horizontally later.
+ */
+export async function claimDrainOwners(
+  db: DrizzleHandle,
+  opts: { batch: number; nowSec: number; staleBeforeSec: number },
+): Promise<DrainClaim[]> {
+  const { batch, nowSec, staleBeforeSec } = opts;
+  const result = await db.execute(sql`
+    UPDATE cow_cache.owner_drain_state AS s
+    SET status = 'draining', claimed_at = ${nowSec}, updated_at = ${nowSec}
+    WHERE (s.chain_id, s.owner) IN (
+      SELECT chain_id, owner
+      FROM cow_cache.owner_drain_state
+      WHERE status = 'pending'
+         OR (status = 'draining' AND (claimed_at IS NULL OR claimed_at < ${staleBeforeSec}))
+      ORDER BY updated_at ASC
+      LIMIT ${batch}
+      FOR UPDATE SKIP LOCKED
+    )
+    RETURNING s.chain_id AS "chainId", s.owner AS owner, s.next_offset AS "nextOffset"
+  `);
+  return rowsOf(result).map((r) => ({
+    chainId: Number(r.chainId),
+    owner: r.owner as Hex,
+    nextOffset: Number(r.nextOffset),
+  }));
+}
+
+/** Persist drain progress mid-walk: advance next_offset and refresh the lease heartbeat. */
+export async function commitDrainProgress(
+  db: DrizzleHandle,
+  chainId: number,
+  owner: Hex,
+  nextOffset: number,
+  nowSec: number,
+): Promise<void> {
+  await db
+    .update(ownerDrainState)
+    .set({ nextOffset, claimedAt: BigInt(nowSec), updatedAt: BigInt(nowSec) })
+    .where(and(eq(ownerDrainState.chainId, chainId), eq(ownerDrainState.owner, owner.toLowerCase())));
+}
+
+/** Mark an owner fully drained. Terminal — never revisited (its cache survives reindex). */
+export async function completeDrainOwner(
+  db: DrizzleHandle,
+  chainId: number,
+  owner: Hex,
+  nowSec: number,
+): Promise<void> {
+  await db
+    .update(ownerDrainState)
+    .set({ status: "complete", claimedAt: null, updatedAt: BigInt(nowSec) })
+    .where(and(eq(ownerDrainState.chainId, chainId), eq(ownerDrainState.owner, owner.toLowerCase())));
+}
+
+/** Return a partially-drained owner to the queue (transient error). next_offset is kept,
+ *  so the next claim resumes where the walk stopped. */
+export async function releaseDrainOwner(
+  db: DrizzleHandle,
+  chainId: number,
+  owner: Hex,
+  nowSec: number,
+): Promise<void> {
+  await db
+    .update(ownerDrainState)
+    .set({ status: "pending", claimedAt: null, updatedAt: BigInt(nowSec) })
+    .where(and(eq(ownerDrainState.chainId, chainId), eq(ownerDrainState.owner, owner.toLowerCase())));
+}
+
+// ─── DDL ─────────────────────────────────────────────────────────────────────
+
+/**
+ * Create the cow_cache schema and the tables the drain owns, idempotently.
+ * Called by BOTH the Ponder setup handler (with `context.db.sql`) and the worker
+ * bootstrap (with its own pool) so table existence is not startup-ordering-dependent.
+ */
+export async function createCowCacheTables(db: DrizzleHandle): Promise<void> {
+  await db.execute(sql`CREATE SCHEMA IF NOT EXISTS cow_cache`);
+
+  // Durable per-owner composable-order rows, keyed by (chain_id, order_uid). Holds every
+  // field needed to rebuild a discreteOrder row without re-hitting the orderbook. Stored
+  // by the stable generator_hash (not the per-deployment eventId) so rows survive reindex
+  // and re-map to the current generator by hash.
+  await db.execute(sql`
+    CREATE TABLE IF NOT EXISTS cow_cache.composable_order (
+      chain_id              INTEGER NOT NULL,
+      order_uid             TEXT NOT NULL,
+      owner                 TEXT NOT NULL,
+      generator_hash        TEXT NOT NULL,
+      order_type            TEXT NOT NULL,
+      status                TEXT NOT NULL,
+      sell_amount           TEXT NOT NULL,
+      buy_amount            TEXT NOT NULL,
+      fee_amount            TEXT NOT NULL,
+      valid_to              INTEGER,
+      creation_date         BIGINT NOT NULL,
+      executed_sell_amount   TEXT,
+      executed_buy_amount    TEXT,
+      fetched_at            BIGINT NOT NULL,
+      PRIMARY KEY (chain_id, order_uid)
+    )
+  `);
+  await db.execute(sql`
+    CREATE INDEX IF NOT EXISTS composable_order_owner_idx
+      ON cow_cache.composable_order (chain_id, owner)
+  `);
+
+  // Durable per-owner drain work queue, keyed by (chain_id, owner). Coordinates the
+  // indexer (enqueue on discovery, project on complete) with the standalone worker
+  // (claim → drain → complete). Survives reindex, so a redeploy never re-drains a
+  // 'complete' owner — its rows are already in composable_order.
+  await db.execute(sql`
+    CREATE TABLE IF NOT EXISTS cow_cache.owner_drain_state (
+      chain_id     INTEGER NOT NULL,
+      owner        TEXT NOT NULL,
+      status       TEXT NOT NULL DEFAULT 'pending',
+      next_offset  INTEGER NOT NULL DEFAULT 0,
+      claimed_at   BIGINT,
+      updated_at   BIGINT NOT NULL,
+      PRIMARY KEY (chain_id, owner)
+    )
+  `);
+  await db.execute(sql`
+    CREATE INDEX IF NOT EXISTS owner_drain_state_status_idx
+      ON cow_cache.owner_drain_state (status, updated_at)
+  `);
+}
+
+export { TERMINAL_STATUSES };
diff --git a/src/application/helpers/orderbookClient.ts b/src/application/helpers/orderbookClient.ts
index 64c7fe0..4e221b9 100644
--- a/src/application/helpers/orderbookClient.ts
+++ b/src/application/helpers/orderbookClient.ts
@@ -1,11 +1,16 @@
 /**
- * Orderbook client — fetches and caches composable orders from the CoW Orderbook API.
+ * Orderbook client (Ponder-side) — status refresh, flash-loan enrichment, discrete-order
+ * upsert, and the generator re-mapping used by the OwnerBackfill projection.
+ *
+ * The raw HTTP layer lives in orderbookHttp.ts and the durable composable cache +
+ * drain-state queue in composableCache.ts — both Ponder-free so the standalone drain
+ * worker can share them. This module is the part that touches Ponder's versioned schema
+ * (conditionalOrderGenerator / discreteOrder), so it stays inside the Ponder runtime.
  *
  * Cache strategy (per-UID):
  * - Uses cow_cache.order_uid_cache to store per-UID terminal statuses
  * - Terminal orders (fulfilled/expired/cancelled) are cached and never re-fetched
  * - Open/non-cached orders are refreshed via POST /api/v1/orders/by_uids
- * - Cache is invalidated per-owner when ConditionalOrderCreated fires
  *
  * KNOWN LIMITATION — Off-chain cancellation gap:
  *   Orders cancelled via the CoW Orderbook API's DELETE endpoint (off-chain
@@ -19,42 +24,23 @@ import {
   discreteOrder,
 } from "ponder:schema";
 import { and, eq, inArray, sql } from "ponder";
-import { pgSchema, integer, text, bigint } from "drizzle-orm/pg-core";
-import { encodeAbiParameters, keccak256, type Hex } from "viem";
+import { pgSchema, integer, text } from "drizzle-orm/pg-core";
+import { type Hex } from "viem";
 import { type OrderType } from "../../utils/order-types";
-import { COMPOSABLE_COW_HANDLER_ADDRESSES, ORDERBOOK_API_URLS } from "../../data";
-import {
-  ORDERBOOK_HTTP_TIMEOUT_MS,
-  ORDERBOOK_MAX_RETRIES,
-  ORDERBOOK_RETRY_BASE_MS,
-  ORDERBOOK_RETRY_BUDGET_MS,
-  ORDERBOOK_RETRY_MAX_DELAY_MS,
-  SIGNING_SCHEME_EIP1271,
-} from "../../constants";
-import { decodeEip1271Signature } from "../decoders/erc1271Signature";
-import { fetchWithTimeout, TimeoutError, withTimeout } from "./withTimeout";
+import { ORDERBOOK_API_URLS } from "../../data";
+import { ORDERBOOK_HTTP_TIMEOUT_MS } from "../../constants";
+import { TimeoutError, withTimeout } from "./withTimeout";
 import { log } from "./logger";
+import { fetchAccountOrders, fetchOrdersByUids, type OrderbookOrder } from "./orderbookHttp";
+import { type ComposableCacheRow, type DrizzleHandle } from "./composableCache";
 
-// ─── Types ───────────────────────────────────────────────────────────────────
+// Re-export the account paginator so callers that only need the raw HTTP path can
+// import it from here (the historical import site) without reaching into orderbookHttp.
+export { fetchAccountOrders } from "./orderbookHttp";
 
-/** Raw API response shape (subset of fields we use). */
-interface OrderbookOrder {
-  uid: string;
-  status: "open" | "fulfilled" | "expired" | "cancelled" | "presignaturePending";
-  kind: "sell" | "buy";
-  receiver: string | null;
-  sellAmount: string;
-  buyAmount: string;
-  feeAmount: string;
-  validTo: number;
-  creationDate: string; // ISO 8601
-  signingScheme: string;
-  signature: string;
-  executedSellAmount: string;
-  executedBuyAmount: string;
-}
+// ─── Types ───────────────────────────────────────────────────────────────────
 
-/** Processed composable order stored in cache and returned to callers.
+/** Processed composable order returned to callers (discrete-order projection).
  *  Shares field types with the discreteOrder schema for the DB-mapped fields. */
 export type ComposableOrder = Pick<
   typeof discreteOrder.$inferInsert,
@@ -75,96 +61,9 @@ export interface OrderStatusInfo {
 }
 
 const TERMINAL_STATUSES = new Set(["fulfilled", "expired", "cancelled"]);
-const PAGE_LIMIT = 1000;
-const BATCH_SIZE = 50;
 
 // ─── Public API ──────────────────────────────────────────────────────────────
 
-/**
- * Fetch composable orders for an owner, using per-UID cache for terminal orders.
- * Incremental drain: Ponder rebuilds the onchain discreteOrder table
- * from scratch on every schema-hash redeploy, so a naive implementation re-fetches
- * an owner's entire history each deploy. Instead the full composable-order rows are
- * kept in the durable cow_cache.composable_order table (survives reindex), and only
- * the delta newer than MAX(creation_date) is fetched from the orderbook:
- *
- * 1. cursor = newest creation_date already cached for this owner (undefined = full drain)
- * 2. Fetch /account/{owner}/orders newest-first, stopping once older than the cursor
- * 3. Decode → filter to composable → match to generators, then persist the delta
- * 4. Rebuild the full owner set from the durable cache (delta + all older rows)
- * 5. Re-check any still-open cached rows via by_uids so statuses don't go stale
- * 6. Re-map generator_hash → the current generator eventId (changes each reindex)
- */
-export async function fetchComposableOrders(
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  context: any,
-  chainId: number,
-  owner: Hex,
-): Promise<{ orders: ComposableOrder[]; complete: boolean }> {
-  const apiBaseUrl = ORDERBOOK_API_URLS[chainId];
-  if (!apiBaseUrl) {
-    log("warn", "ob:noApiUrl", { chainId });
-    return { orders: [], complete: false };
-  }
-
-  // Only fetch orders newer than what we've already durably cached for this owner.
-  const cursor = await readOwnerBackfillCursor(context, chainId, owner);
-  log("info", "ob:fetch", { owner, chainId, since: cursor ?? null });
-
-  // complete=false (pagination cut short by rate limit / timeout) means the caller must
-  // NOT mark the owner backfilled — it stays eligible and is retried on a later block.
-  const { orders: deltaApiOrders, complete } = await fetchAccountOrders(apiBaseUrl, owner, 0, SIGNING_SCHEME_EIP1271, PAGE_LIMIT, cursor);
-  const delta = await filterAndProcess(context, chainId, deltaApiOrders);
-
-  // Persist the delta (account-endpoint status is the live status) into the durable cache.
-  await upsertComposableCache(context, chainId, owner, delta.map(toCacheRow));
-
-  // Rebuild the full owner set from the durable cache (delta + everything older).
-  const cachedRows = await readOwnerComposableCache(context, chainId, owner);
-
-  // Re-check any still-open cached rows — long-lived orders that terminated below the
-  // cursor since a prior drain would otherwise keep a stale "open" status forever.
-  const reconciled = await reconcileOpenCachedRows(context, chainId, owner, apiBaseUrl, cachedRows);
-
-  // The per-deployment generator eventId changes each reindex; re-map by the stable hash.
-  const results = await remapToCurrentGenerators(context, chainId, reconciled);
-
-  log("info", "ob:fetchResult", { owner, chainId, since: cursor ?? null, delta: delta.length, total: results.length, complete });
-  return { orders: results, complete };
-}
-
-/** Durable-cache row shape for cow_cache.composable_order (owner passed separately). */
-interface ComposableCacheRow {
-  orderUid: string;
-  generatorHash: string;
-  orderType: OrderType;
-  status: string;
-  sellAmount: string;
-  buyAmount: string;
-  feeAmount: string;
-  validTo: number | null;
-  creationDate: bigint;
-  executedSellAmount: string | null;
-  executedBuyAmount: string | null;
-}
-
-/** Project a freshly-decoded ComposableOrder into the durable-cache row shape. */
-function toCacheRow(o: ComposableOrder): ComposableCacheRow {
-  return {
-    orderUid: o.uid,
-    generatorHash: o.generatorHash,
-    orderType: o.orderType,
-    status: o.status,
-    sellAmount: o.sellAmount,
-    buyAmount: o.buyAmount,
-    feeAmount: o.feeAmount,
-    validTo: o.validTo ?? null,
-    creationDate: o.creationDate,
-    executedSellAmount: o.executedSellAmount ?? null,
-    executedBuyAmount: o.executedBuyAmount ?? null,
-  };
-}
-
 /**
  * Upsert composable orders into the discrete_order table.
  * Uses onConflictDoUpdate so the API's authoritative status overwrites
@@ -397,273 +296,55 @@ export async function fetchFlashLoanEnrichmentByUids(
   return result;
 }
 
-// ─── API calls ───────────────────────────────────────────────────────────────
-
-/**
- * The orderbook API refused to answer (HTTP 429 or 5xx) after bounded retries.
- * Distinct from "the API has no such order" (a UID simply absent from a 2xx
- * body) so callers / dashboards can alarm on an unavailable API rather than
- * silently treating it as "order not on API yet".
- */
-export class OrderbookUnavailableError extends Error {
-  constructor(
-    public readonly status: number,
-    public readonly endpoint: string,
-  ) {
-    super(`[COW:orderbook-unavailable] ${endpoint} responded ${status}`);
-    this.name = "OrderbookUnavailableError";
-  }
-}
-
-const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
-
-/** Parse an orderbook order's ISO creationDate into Unix seconds. */
-function orderCreationSeconds(order: OrderbookOrder): number {
-  return Math.floor(new Date(order.creationDate).getTime() / 1000);
-}
-
-/** Parse a `Retry-After` header (delta-seconds or HTTP-date) into milliseconds; null if absent/unparseable. */
-function parseRetryAfter(value: string | null): number | null {
-  if (!value) return null;
-  const seconds = Number(value);
-  if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000);
-  const date = Date.parse(value);
-  if (!Number.isNaN(date)) return Math.max(0, date - Date.now());
-  return null;
-}
-
-/**
- * `fetchWithTimeout` plus bounded retry/backoff for transient orderbook errors.
- *
- * Returns the Response on a 2xx. On 429 it honors `Retry-After` (capped at
- * ORDERBOOK_RETRY_MAX_DELAY_MS); on 5xx it uses exponential backoff. Retries
- * stop once ORDERBOOK_MAX_RETRIES is reached or the next sleep would push the
- * loop past ORDERBOOK_RETRY_BUDGET_MS — at which point it throws
- * OrderbookUnavailableError instead of holding the block transaction open.
- * A TimeoutError from the underlying fetch propagates unchanged.
- */
-async function fetchOrderbook(
-  url: string,
-  init: RequestInit | undefined,
-  endpoint: string,
-): Promise<Response> {
-  let spent = 0;
-  for (let attempt = 0; ; attempt++) {
-    const response = await fetchWithTimeout(url, init, ORDERBOOK_HTTP_TIMEOUT_MS, endpoint);
-    if (response.ok) return response;
-
-    const retryable = response.status === 429 || response.status >= 500;
-    if (!retryable || attempt >= ORDERBOOK_MAX_RETRIES) {
-      throw new OrderbookUnavailableError(response.status, endpoint);
-    }
-
-    const retryAfterMs =
-      response.status === 429 ? parseRetryAfter(response.headers.get("retry-after")) : null;
-    const backoffMs = ORDERBOOK_RETRY_BASE_MS * 2 ** attempt;
-    const delay = Math.min(retryAfterMs ?? backoffMs, ORDERBOOK_RETRY_MAX_DELAY_MS);
-
-    // Fail fast rather than hold the block transaction open past our budget.
-    if (spent + delay > ORDERBOOK_RETRY_BUDGET_MS) {
-      throw new OrderbookUnavailableError(response.status, endpoint);
-    }
-
-    log("warn", "ob:retry", { endpoint, status: response.status, attempt: attempt + 1, delayMs: delay, retryAfterMs });
-    await sleep(delay);
-    spent += delay;
-  }
-}
-
-/** Fetch orders for an owner with pagination. maxPages limits how many pages are fetched (0 = unlimited).
- *  signingScheme, if provided, is appended as a query param — the API filters server-side when supported,
- *  reducing payload for owners with many ECDSA orders mixed with composable ones.
- *  pageSize overrides the default PAGE_LIMIT per request.
- *
- *  sinceCreationDate (Unix seconds), if provided, enables an incremental drain: the
- *  API returns orders newest-first (creationDate DESC), so once a page contains an
- *  order strictly older than the cursor, everything beyond it is already known and
- *  pagination stops. Orders at or after the cursor are kept (the boundary is
- *  re-included so ties at exactly the cursor second are never dropped). */
-export async function fetchAccountOrders(
-  apiBaseUrl: string,
-  owner: Hex,
-  maxPages = 0,
-  signingScheme?: string,
-  pageSize = PAGE_LIMIT,
-  sinceCreationDate?: number,
-): Promise<{ orders: OrderbookOrder[]; complete: boolean }> {
-  const allOrders: OrderbookOrder[] = [];
-  let offset = 0;
-  let pagesFetched = 0;
-  // complete=false means pagination was cut short by an error (rate limit / timeout /
-  // network) — the caller must NOT treat the result as the owner's full history.
-  let complete = false;
-
-  // eslint-disable-next-line no-constant-condition
-  while (true) {
-    const params = new URLSearchParams({ limit: String(pageSize), offset: String(offset) });
-    if (signingScheme) params.set("signingScheme", signingScheme);
-    const url = `${apiBaseUrl}/api/v1/account/${owner}/orders?${params.toString()}`;
-    try {
-      const response = await fetchOrderbook(url, undefined, "ob:account");
-      const page = (await response.json()) as OrderbookOrder[];
-
-      if (sinceCreationDate !== undefined) {
-        // DESC order → orders at/after the cursor form a prefix of the page.
-        const fresh = page.filter((o) => orderCreationSeconds(o) >= sinceCreationDate);
-        allOrders.push(...fresh);
-        if (fresh.length < page.length) { complete = true; break; } // crossed the cursor — older orders already cached
-      } else {
-        allOrders.push(...page);
-      }
-
-      pagesFetched++;
-      if (page.length < pageSize) { complete = true; break; } // last page
-      if (maxPages > 0 && pagesFetched >= maxPages) { complete = true; break; } // page cap reached
-      offset += page.length;
-    } catch (err) {
-      if (err instanceof OrderbookUnavailableError) {
-        log("error", "ob:unavailable", { endpoint: "ob:account", status: err.status, owner });
-        break;
-      }
-      if (err instanceof TimeoutError) {
-        log("warn", "ob:accountTimeout", { owner, offset, after: ORDERBOOK_HTTP_TIMEOUT_MS });
-        break;
-      }
-      log("warn", "ob:accountFetchFailed", { owner, err: String(err) });
-      break;
-    }
-  }
-
-  return { orders: allOrders, complete };
-}
-
-/** Batch-fetch orders by UID to refresh status of open orders.
- *  Chunks into BATCH_SIZE to avoid HTTP 413, then fires all chunks in parallel
- *  so N chunks take the time of one instead of N × one. */
-async function fetchOrdersByUids(
-  apiBaseUrl: string,
-  uids: string[],
-): Promise<OrderbookOrder[]> {
-  if (uids.length === 0) return [];
+// ─── Projection: durable cache → current generators ──────────────────────────
 
-  const url = `${apiBaseUrl}/api/v1/orders/by_uids`;
-  const chunks: string[][] = [];
-  for (let i = 0; i < uids.length; i += BATCH_SIZE) {
-    chunks.push(uids.slice(i, i + BATCH_SIZE));
-  }
-
-  const chunkResults = await Promise.all(
-    chunks.map(async (chunk, idx) => {
-      try {
-        const response = await fetchOrderbook(
-          url,
-          {
-            method: "POST",
-            headers: { "Content-Type": "application/json" },
-            body: JSON.stringify(chunk),
-          },
-          "ob:byUids",
-        );
-        const raw = (await response.json()) as { order: OrderbookOrder }[];
-        return raw.flatMap((item) => (item?.order != null ? [item.order] : []));
-      } catch (err) {
-        if (err instanceof OrderbookUnavailableError) {
-          log("error", "ob:unavailable", { endpoint: "ob:byUids", status: err.status, uids: chunk.length, offset: idx * BATCH_SIZE });
-          return [] as OrderbookOrder[];
-        }
-        if (err instanceof TimeoutError) {
-          log("warn", "ob:batchFetchTimeout", { uids: chunk.length, offset: idx * BATCH_SIZE, after: ORDERBOOK_HTTP_TIMEOUT_MS });
-          return [] as OrderbookOrder[];
-        }
-        log("warn", "ob:batchFetchFailed", { err: String(err), offset: idx * BATCH_SIZE });
-        return [] as OrderbookOrder[];
-      }
-    }),
-  );
-
-  return chunkResults.flat();
-}
-
-// ─── Processing ──────────────────────────────────────────────────────────────
-
-/** Filter API orders to composable eip1271, decode signatures, match to generators. */
-async function filterAndProcess(
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  context: any,
+/** Map durable rows (keyed by the stable generator_hash) to ComposableOrder with the
+ *  current per-deployment generator eventId. Rows with no current generator are dropped.
+ *  This is the hash → eventId join the drain worker deliberately skips. */
+export async function remapToCurrentGenerators(
+  db: DrizzleHandle,
   chainId: number,
-  apiOrders: OrderbookOrder[],
+  rows: ComposableCacheRow[],
 ): Promise<ComposableOrder[]> {
-  const results: ComposableOrder[] = [];
-
-  for (const order of apiOrders) {
-    if (order.signingScheme !== SIGNING_SCHEME_EIP1271) continue;
-    if (order.status === "presignaturePending") continue;
-
-    const decoded = decodeEip1271Signature(order.signature as Hex);
-    if (!decoded) continue;
-
-    if (!COMPOSABLE_COW_HANDLER_ADDRESSES.has(decoded.handler)) continue;
-
-    // Reproduce the same hash stored in conditionalOrderGenerator.hash
-    const paramHash = keccak256(
-      encodeAbiParameters(
-        [
-          {
-            type: "tuple",
-            components: [
-              { name: "handler", type: "address" },
-              { name: "salt", type: "bytes32" },
-              { name: "staticInput", type: "bytes" },
-            ],
-          },
-        ],
-        [{ handler: decoded.handler, salt: decoded.salt, staticInput: decoded.staticInput }],
-      ),
-    );
-
-    // Find the generator — there should be exactly one per (chainId, hash).
-    // Uses context.db.sql (raw SQL) because Ponder ORM has no non-PK findMany.
-    // Wrapped in try-catch: in multichain realtime mode a shared-qb race can cause
-    // a SAVEPOINT error here; skipping the order is safe — it's retried next block.
-    let generators: { eventId: string; orderType: OrderType }[];
-    try {
-      generators = (await context.db.sql
-        .select({
-          eventId: conditionalOrderGenerator.eventId,
-          orderType: conditionalOrderGenerator.orderType,
-        })
-        .from(conditionalOrderGenerator)
-        .where(
-          and(
-            eq(conditionalOrderGenerator.chainId, chainId),
-            eq(conditionalOrderGenerator.hash, paramHash),
-          ),
-        )
-        .limit(1)) as { eventId: string; orderType: OrderType }[];
-    } catch {
-      continue;
-    }
+  if (rows.length === 0) return [];
+  const hashes = [...new Set(rows.map((r) => r.generatorHash))] as Hex[];
 
-    if (generators.length === 0) continue;
+  let generators: { eventId: string; hash: string }[];
+  try {
+    generators = (await db
+      .select({ eventId: conditionalOrderGenerator.eventId, hash: conditionalOrderGenerator.hash })
+      .from(conditionalOrderGenerator)
+      .where(
+        and(
+          eq(conditionalOrderGenerator.chainId, chainId),
+          inArray(conditionalOrderGenerator.hash, hashes),
+        ),
+      )) as { eventId: string; hash: string }[];
+  } catch {
+    return [];
+  }
 
-    const generator = generators[0]!;
+  const eventIdByHash = new Map(generators.map((g) => [g.hash, g.eventId]));
 
+  const results: ComposableOrder[] = [];
+  for (const row of rows) {
+    const generatorId = eventIdByHash.get(row.generatorHash);
+    if (!generatorId) continue;
     results.push({
-      uid: order.uid,
-      status: order.status as ComposableOrder["status"],
-      generatorId: generator.eventId,
-      generatorHash: paramHash,
-      orderType: generator.orderType,
-      sellAmount: order.sellAmount,
-      buyAmount: order.buyAmount,
-      feeAmount: order.feeAmount,
-      validTo: order.validTo,
-      creationDate: BigInt(Math.floor(new Date(order.creationDate).getTime() / 1000)),
-      executedSellAmount: order.executedSellAmount,
-      executedBuyAmount: order.executedBuyAmount,
+      uid: row.orderUid,
+      status: row.status as ComposableOrder["status"],
+      generatorId,
+      generatorHash: row.generatorHash,
+      orderType: row.orderType,
+      sellAmount: row.sellAmount,
+      buyAmount: row.buyAmount,
+      feeAmount: row.feeAmount,
+      validTo: row.validTo,
+      creationDate: row.creationDate,
+      executedSellAmount: row.executedSellAmount,
+      executedBuyAmount: row.executedBuyAmount,
     });
   }
-
   return results;
 }
 
@@ -674,24 +355,6 @@ async function filterAndProcess(
 // columns are nullable; the two UID populations are disjoint.
 const cowCacheSchema = pgSchema("cow_cache");
 
-// Durable full composable-order rows (survives reindex). See setup.ts for the DDL.
-const composableOrderCache = cowCacheSchema.table("composable_order", {
-  chainId: integer("chain_id").notNull(),
-  orderUid: text("order_uid").notNull(),
-  owner: text("owner").notNull(),
-  generatorHash: text("generator_hash").notNull(),
-  orderType: text("order_type").notNull(),
-  status: text("status").notNull(),
-  sellAmount: text("sell_amount").notNull(),
-  buyAmount: text("buy_amount").notNull(),
-  feeAmount: text("fee_amount").notNull(),
-  validTo: integer("valid_to"),
-  creationDate: bigint("creation_date", { mode: "bigint" }).notNull(),
-  executedSellAmount: text("executed_sell_amount"),
-  executedBuyAmount: text("executed_buy_amount"),
-  fetchedAt: bigint("fetched_at", { mode: "bigint" }).notNull(),
-});
-
 const orderUidCache = cowCacheSchema.table("order_uid_cache", {
   chainId: integer("chain_id").notNull(),
   orderUid: text("order_uid").notNull(),
@@ -878,195 +541,3 @@ async function cacheUidStatuses(
     // Best-effort cache write
   }
 }
-
-// ─── Durable composable-order cache helpers ───────────────────────────────────
-// cow_cache.composable_order (created in setup.ts) holds full composable-order rows
-// keyed by (chain_id, order_uid), so the backfill drains only the delta newer than
-// MAX(creation_date) per owner instead of the full history on each reindex.
-
-/** Newest creation_date already cached for this owner (Unix seconds), or undefined
- *  when nothing is cached — the signal to do a full-history drain. */
-async function readOwnerBackfillCursor(
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  context: any,
-  chainId: number,
-  owner: Hex,
-): Promise<number | undefined> {
-  try {
-    const rows = (await context.db.sql
-      .select({ cursor: sql<string | null>`max(${composableOrderCache.creationDate})` })
-      .from(composableOrderCache)
-      .where(
-        and(
-          eq(composableOrderCache.chainId, chainId),
-          eq(composableOrderCache.owner, owner.toLowerCase()),
-        ),
-      )) as { cursor: string | null }[];
-    const raw = rows[0]?.cursor;
-    return raw == null ? undefined : Number(raw);
-  } catch {
-    return undefined; // no cache table / error → full drain
-  }
-}
-
-/** All durably-cached composable rows for an owner. */
-async function readOwnerComposableCache(
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  context: any,
-  chainId: number,
-  owner: Hex,
-): Promise<ComposableCacheRow[]> {
-  try {
-    return (await context.db.sql
-      .select({
-        orderUid: composableOrderCache.orderUid,
-        generatorHash: composableOrderCache.generatorHash,
-        orderType: composableOrderCache.orderType,
-        status: composableOrderCache.status,
-        sellAmount: composableOrderCache.sellAmount,
-        buyAmount: composableOrderCache.buyAmount,
-        feeAmount: composableOrderCache.feeAmount,
-        validTo: composableOrderCache.validTo,
-        creationDate: composableOrderCache.creationDate,
-        executedSellAmount: composableOrderCache.executedSellAmount,
-        executedBuyAmount: composableOrderCache.executedBuyAmount,
-      })
-      .from(composableOrderCache)
-      .where(
-        and(
-          eq(composableOrderCache.chainId, chainId),
-          eq(composableOrderCache.owner, owner.toLowerCase()),
-        ),
-      )) as ComposableCacheRow[];
-  } catch {
-    return [];
-  }
-}
-
-/** Upsert durable composable rows; excluded status/validTo/executed overwrite on conflict. */
-async function upsertComposableCache(
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  context: any,
-  chainId: number,
-  owner: Hex,
-  rows: ComposableCacheRow[],
-): Promise<void> {
-  if (rows.length === 0) return;
-  const now = BigInt(Math.floor(Date.now() / 1000));
-  try {
-    await context.db.sql
-      .insert(composableOrderCache)
-      .values(rows.map((r) => ({
-        chainId,
-        orderUid: r.orderUid,
-        owner: owner.toLowerCase(),
-        generatorHash: r.generatorHash,
-        orderType: r.orderType,
-        status: r.status,
-        sellAmount: r.sellAmount,
-        buyAmount: r.buyAmount,
-        feeAmount: r.feeAmount,
-        validTo: r.validTo,
-        creationDate: r.creationDate,
-        executedSellAmount: r.executedSellAmount,
-        executedBuyAmount: r.executedBuyAmount,
-        fetchedAt: now,
-      })))
-      .onConflictDoUpdate({
-        target: [composableOrderCache.chainId, composableOrderCache.orderUid],
-        set: {
-          status: sql`excluded.status`,
-          validTo: sql`excluded.valid_to`,
-          executedSellAmount: sql`excluded.executed_sell_amount`,
-          executedBuyAmount: sql`excluded.executed_buy_amount`,
-          fetchedAt: now,
-        },
-      });
-  } catch (err) {
-    log("warn", "ob:composableCacheWriteFailed", { chainId, rows: rows.length, err: String(err) });
-  }
-}
-
-/** Re-check non-terminal cached rows via by_uids; update status/validTo/executed and
- *  re-persist any that became terminal. Mutates and returns `rows`. */
-async function reconcileOpenCachedRows(
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  context: any,
-  chainId: number,
-  owner: Hex,
-  apiBaseUrl: string,
-  rows: ComposableCacheRow[],
-): Promise<ComposableCacheRow[]> {
-  const openUids = rows.filter((r) => !TERMINAL_STATUSES.has(r.status)).map((r) => r.orderUid);
-  if (openUids.length === 0) return rows;
-
-  const refreshed = await fetchOrdersByUids(apiBaseUrl, openUids);
-  if (refreshed.length === 0) return rows;
-  const byUid = new Map(refreshed.map((o) => [o.uid, o]));
-
-  const newlyTerminal: ComposableCacheRow[] = [];
-  for (const row of rows) {
-    const fresh = byUid.get(row.orderUid);
-    if (!fresh) continue;
-    row.status = fresh.status;
-    row.validTo = fresh.validTo;
-    row.executedSellAmount = fresh.executedSellAmount;
-    row.executedBuyAmount = fresh.executedBuyAmount;
-    if (TERMINAL_STATUSES.has(fresh.status)) newlyTerminal.push(row);
-  }
-
-  if (newlyTerminal.length > 0) {
-    await upsertComposableCache(context, chainId, owner, newlyTerminal);
-  }
-  return rows;
-}
-
-/** Map durable rows (keyed by the stable generator_hash) to ComposableOrder with the
- *  current per-deployment generator eventId. Rows with no current generator are dropped. */
-async function remapToCurrentGenerators(
-  // eslint-disable-next-line @typescript-eslint/no-explicit-any
-  context: any,
-  chainId: number,
-  rows: ComposableCacheRow[],
-): Promise<ComposableOrder[]> {
-  if (rows.length === 0) return [];
-  const hashes = [...new Set(rows.map((r) => r.generatorHash))] as Hex[];
-
-  let generators: { eventId: string; hash: string }[];
-  try {
-    generators = (await context.db.sql
-      .select({ eventId: conditionalOrderGenerator.eventId, hash: conditionalOrderGenerator.hash })
-      .from(conditionalOrderGenerator)
-      .where(
-        and(
-          eq(conditionalOrderGenerator.chainId, chainId),
-          inArray(conditionalOrderGenerator.hash, hashes),
-        ),
-      )) as { eventId: string; hash: string }[];
-  } catch {
-    return [];
-  }
-
-  const eventIdByHash = new Map(generators.map((g) => [g.hash, g.eventId]));
-
-  const results: ComposableOrder[] = [];
-  for (const row of rows) {
-    const generatorId = eventIdByHash.get(row.generatorHash);
-    if (!generatorId) continue;
-    results.push({
-      uid: row.orderUid,
-      status: row.status as ComposableOrder["status"],
-      generatorId,
-      generatorHash: row.generatorHash,
-      orderType: row.orderType,
-      sellAmount: row.sellAmount,
-      buyAmount: row.buyAmount,
-      feeAmount: row.feeAmount,
-      validTo: row.validTo,
-      creationDate: row.creationDate,
-      executedSellAmount: row.executedSellAmount,
-      executedBuyAmount: row.executedBuyAmount,
-    });
-  }
-  return results;
-}
diff --git a/src/application/helpers/orderbookHttp.ts b/src/application/helpers/orderbookHttp.ts
new file mode 100644
index 0000000..d4a1aaa
--- /dev/null
+++ b/src/application/helpers/orderbookHttp.ts
@@ -0,0 +1,246 @@
+/**
+ * Orderbook HTTP layer — the CoW Orderbook API client, with no Ponder dependency.
+ *
+ * Deliberately free of `ponder` / `ponder:schema` imports so it can be imported by
+ * both the Ponder handlers (via orderbookClient.ts) AND the standalone drain worker
+ * (src/worker/drain.ts), which runs outside the Ponder runtime where the virtual
+ * `ponder:*` modules don't resolve.
+ *
+ * Only raw HTTP lives here: fetch + bounded retry/backoff, account pagination, and
+ * by_uids batching. Decoding, generator matching, and the durable cache live in
+ * composableCache.ts (also ponder-free) and orderbookClient.ts (Ponder-side).
+ */
+
+import type { Hex } from "viem";
+import {
+  ORDERBOOK_HTTP_TIMEOUT_MS,
+  ORDERBOOK_MAX_RETRIES,
+  ORDERBOOK_RETRY_BASE_MS,
+  ORDERBOOK_RETRY_BUDGET_MS,
+  ORDERBOOK_RETRY_MAX_DELAY_MS,
+} from "../../constants";
+import { fetchWithTimeout, TimeoutError } from "./withTimeout";
+import { log } from "./logger";
+
+/** Raw API response shape (subset of fields we use). */
+export interface OrderbookOrder {
+  uid: string;
+  status: "open" | "fulfilled" | "expired" | "cancelled" | "presignaturePending";
+  kind: "sell" | "buy";
+  receiver: string | null;
+  sellAmount: string;
+  buyAmount: string;
+  feeAmount: string;
+  validTo: number;
+  creationDate: string; // ISO 8601
+  signingScheme: string;
+  signature: string;
+  executedSellAmount: string;
+  executedBuyAmount: string;
+}
+
+export const PAGE_LIMIT = 1000;
+export const BATCH_SIZE = 50;
+
+/**
+ * The orderbook API refused to answer (HTTP 429 or 5xx) after bounded retries.
+ * Distinct from "the API has no such order" (a UID simply absent from a 2xx
+ * body) so callers / dashboards can alarm on an unavailable API rather than
+ * silently treating it as "order not on API yet".
+ */
+export class OrderbookUnavailableError extends Error {
+  constructor(
+    public readonly status: number,
+    public readonly endpoint: string,
+  ) {
+    super(`[COW:orderbook-unavailable] ${endpoint} responded ${status}`);
+    this.name = "OrderbookUnavailableError";
+  }
+}
+
+const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
+
+/** Parse an orderbook order's ISO creationDate into Unix seconds. */
+export function orderCreationSeconds(order: OrderbookOrder): number {
+  return Math.floor(new Date(order.creationDate).getTime() / 1000);
+}
+
+/** Parse a `Retry-After` header (delta-seconds or HTTP-date) into milliseconds; null if absent/unparseable. */
+function parseRetryAfter(value: string | null): number | null {
+  if (!value) return null;
+  const seconds = Number(value);
+  if (Number.isFinite(seconds)) return Math.max(0, seconds * 1000);
+  const date = Date.parse(value);
+  if (!Number.isNaN(date)) return Math.max(0, date - Date.now());
+  return null;
+}
+
+/**
+ * `fetchWithTimeout` plus bounded retry/backoff for transient orderbook errors.
+ *
+ * Returns the Response on a 2xx. On 429 it honors `Retry-After` (capped at
+ * ORDERBOOK_RETRY_MAX_DELAY_MS); on 5xx it uses exponential backoff. Retries
+ * stop once ORDERBOOK_MAX_RETRIES is reached or the next sleep would push the
+ * loop past ORDERBOOK_RETRY_BUDGET_MS — at which point it throws
+ * OrderbookUnavailableError instead of holding the caller open indefinitely.
+ * A TimeoutError from the underlying fetch propagates unchanged.
+ */
+export async function fetchOrderbook(
+  url: string,
+  init: RequestInit | undefined,
+  endpoint: string,
+): Promise<Response> {
+  let spent = 0;
+  for (let attempt = 0; ; attempt++) {
+    const response = await fetchWithTimeout(url, init, ORDERBOOK_HTTP_TIMEOUT_MS, endpoint);
+    if (response.ok) return response;
+
+    const retryable = response.status === 429 || response.status >= 500;
+    if (!retryable || attempt >= ORDERBOOK_MAX_RETRIES) {
+      throw new OrderbookUnavailableError(response.status, endpoint);
+    }
+
+    const retryAfterMs =
+      response.status === 429 ? parseRetryAfter(response.headers.get("retry-after")) : null;
+    const backoffMs = ORDERBOOK_RETRY_BASE_MS * 2 ** attempt;
+    const delay = Math.min(retryAfterMs ?? backoffMs, ORDERBOOK_RETRY_MAX_DELAY_MS);
+
+    // Fail fast rather than hold the caller open past our budget.
+    if (spent + delay > ORDERBOOK_RETRY_BUDGET_MS) {
+      throw new OrderbookUnavailableError(response.status, endpoint);
+    }
+
+    log("warn", "ob:retry", { endpoint, status: response.status, attempt: attempt + 1, delayMs: delay, retryAfterMs });
+    await sleep(delay);
+    spent += delay;
+  }
+}
+
+/**
+ * Fetch a single page of an owner's account orders at a given offset (newest-first).
+ *
+ * Used by the drain worker's offset-walk: it advances `offset` by the raw page length
+ * and stops on a short page, so each call must surface the untrimmed page. Throws
+ * OrderbookUnavailableError / TimeoutError so the worker can leave the owner for retry
+ * (its next_offset is already durably committed, so re-fetches are harmless).
+ */
+export async function fetchAccountOrderPage(
+  apiBaseUrl: string,
+  owner: Hex,
+  offset: number,
+  pageSize = PAGE_LIMIT,
+  signingScheme?: string,
+): Promise<OrderbookOrder[]> {
+  const params = new URLSearchParams({ limit: String(pageSize), offset: String(offset) });
+  if (signingScheme) params.set("signingScheme", signingScheme);
+  const url = `${apiBaseUrl}/api/v1/account/${owner}/orders?${params.toString()}`;
+  const response = await fetchOrderbook(url, undefined, "ob:account");
+  return (await response.json()) as OrderbookOrder[];
+}
+
+/** Fetch orders for an owner with pagination. maxPages limits how many pages are fetched (0 = unlimited).
+ *  signingScheme, if provided, is appended as a query param — the API filters server-side when supported,
+ *  reducing payload for owners with many ECDSA orders mixed with composable ones.
+ *  pageSize overrides the default PAGE_LIMIT per request.
+ *
+ *  sinceCreationDate (Unix seconds), if provided, enables an incremental drain: the
+ *  API returns orders newest-first (creationDate DESC), so once a page contains an
+ *  order strictly older than the cursor, everything beyond it is already known and
+ *  pagination stops. Orders at or after the cursor are kept (the boundary is
+ *  re-included so ties at exactly the cursor second are never dropped). */
+export async function fetchAccountOrders(
+  apiBaseUrl: string,
+  owner: Hex,
+  maxPages = 0,
+  signingScheme?: string,
+  pageSize = PAGE_LIMIT,
+  sinceCreationDate?: number,
+): Promise<{ orders: OrderbookOrder[]; complete: boolean }> {
+  const allOrders: OrderbookOrder[] = [];
+  let offset = 0;
+  let pagesFetched = 0;
+  // complete=false means pagination was cut short by an error (rate limit / timeout /
+  // network) — the caller must NOT treat the result as the owner's full history.
+  let complete = false;
+
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
+    try {
+      const page = await fetchAccountOrderPage(apiBaseUrl, owner, offset, pageSize, signingScheme);
+
+      if (sinceCreationDate !== undefined) {
+        // DESC order → orders at/after the cursor form a prefix of the page.
+        const fresh = page.filter((o) => orderCreationSeconds(o) >= sinceCreationDate);
+        allOrders.push(...fresh);
+        if (fresh.length < page.length) { complete = true; break; } // crossed the cursor — older orders already cached
+      } else {
+        allOrders.push(...page);
+      }
+
+      pagesFetched++;
+      if (page.length < pageSize) { complete = true; break; } // last page
+      if (maxPages > 0 && pagesFetched >= maxPages) { complete = true; break; } // page cap reached
+      offset += page.length;
+    } catch (err) {
+      if (err instanceof OrderbookUnavailableError) {
+        log("error", "ob:unavailable", { endpoint: "ob:account", status: err.status, owner });
+        break;
+      }
+      if (err instanceof TimeoutError) {
+        log("warn", "ob:accountTimeout", { owner, offset, after: ORDERBOOK_HTTP_TIMEOUT_MS });
+        break;
+      }
+      log("warn", "ob:accountFetchFailed", { owner, err: String(err) });
+      break;
+    }
+  }
+
+  return { orders: allOrders, complete };
+}
+
+/** Batch-fetch orders by UID to refresh status of open orders.
+ *  Chunks into BATCH_SIZE to avoid HTTP 413, then fires all chunks in parallel
+ *  so N chunks take the time of one instead of N × one. */
+export async function fetchOrdersByUids(
+  apiBaseUrl: string,
+  uids: string[],
+): Promise<OrderbookOrder[]> {
+  if (uids.length === 0) return [];
+
+  const url = `${apiBaseUrl}/api/v1/orders/by_uids`;
+  const chunks: string[][] = [];
+  for (let i = 0; i < uids.length; i += BATCH_SIZE) {
+    chunks.push(uids.slice(i, i + BATCH_SIZE));
+  }
+
+  const chunkResults = await Promise.all(
+    chunks.map(async (chunk, idx) => {
+      try {
+        const response = await fetchOrderbook(
+          url,
+          {
+            method: "POST",
+            headers: { "Content-Type": "application/json" },
+            body: JSON.stringify(chunk),
+          },
+          "ob:byUids",
+        );
+        const raw = (await response.json()) as { order: OrderbookOrder }[];
+        return raw.flatMap((item) => (item?.order != null ? [item.order] : []));
+      } catch (err) {
+        if (err instanceof OrderbookUnavailableError) {
+          log("error", "ob:unavailable", { endpoint: "ob:byUids", status: err.status, uids: chunk.length, offset: idx * BATCH_SIZE });
+          return [] as OrderbookOrder[];
+        }
+        if (err instanceof TimeoutError) {
+          log("warn", "ob:batchFetchTimeout", { uids: chunk.length, offset: idx * BATCH_SIZE, after: ORDERBOOK_HTTP_TIMEOUT_MS });
+          return [] as OrderbookOrder[];
+        }
+        log("warn", "ob:batchFetchFailed", { err: String(err), offset: idx * BATCH_SIZE });
+        return [] as OrderbookOrder[];
+      }
+    }),
+  );
+
+  return chunkResults.flat();
+}
diff --git a/src/constants.ts b/src/constants.ts
index e950780..b7297ce 100644
--- a/src/constants.ts
+++ b/src/constants.ts
@@ -100,14 +100,36 @@ export const SETTLEMENT_INNER_RPC_TIMEOUT_MS = 5_000;
 export const BOOTSTRAP_OWNER_FETCH_TIMEOUT_MS = 30_000;
 
 /**
- * Per-block ceiling on how many distinct owners OwnerBackfill drains in a single
- * 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.
+ * Per-firing ceiling on how many pending owners the OwnerBackfill projection
+ * processes. The projection is DB-only (read the durable cache → discreteOrder,
+ * flip historyBackfilled), so this just bounds the transaction size; owners past
+ * the cap are picked up on the next firing.
  *
  * Override per chain with env var MAX_OWNERS_BACKFILL_PER_BLOCK_<chainId>.
  */
 export const DEFAULT_MAX_OWNERS_BACKFILL_PER_BLOCK = 25;
 
+// ─── Drain worker (src/worker/drain.ts) ───────────────────────────────────────
+
+/**
+ * How many owners the standalone drain worker drains concurrently. The orderbook
+ * rate limit is not the binding constraint (no 429s observed even at 300 concurrent
+ * requests); per-page latency/payload is, so a modest fan-out keeps time-to-drain
+ * low without risk. Override with env var DRAIN_OWNER_CONCURRENCY.
+ */
+export const DRAIN_OWNER_CONCURRENCY = 10;
+
+/** Idle backoff when the drain worker finds no claimable owners. Override with DRAIN_IDLE_SLEEP_MS. */
+export const DRAIN_IDLE_SLEEP_MS = 5_000;
+
+/**
+ * Lease TTL for a claimed ('draining') owner. A worker crash leaves an owner
+ * 'draining' forever; after this TTL another claim reclaims it (its next_offset is
+ * durably committed per page, so the walk resumes rather than restarts). Long
+ * enough to cover the slowest single-owner walk. Override with DRAIN_LEASE_TTL_MS.
+ */
+export const DRAIN_LEASE_TTL_MS = 300_000;
+
 /**
  * Maximum number of TWAP parts that precomputeOrderUids will attempt to enumerate.
  * Pathological orders with n > this value skip precompute and fall back to the
diff --git a/src/worker/drain.ts b/src/worker/drain.ts
new file mode 100644
index 0000000..2e56ea5
--- /dev/null
+++ b/src/worker/drain.ts
@@ -0,0 +1,198 @@
+/**
+ * OwnerBackfill drain worker — a standalone long-running process, separate from the
+ * Ponder indexer.
+ *
+ * Why a separate process: draining an owner's full history from the orderbook is slow
+ * (a fat owner is several ~5 s pages). Run inside a Ponder block handler it serialized
+ * into the single indexing slot and a single un-drainable owner could wedge the whole
+ * pipeline — and, worse, wedge `/readyz` forever (COW-1118). Here it runs on its own
+ * clock and coordinates with the indexer only through the durable `cow_cache` schema:
+ *
+ *   1. The ConditionalOrderCreated handler enqueues non-deterministic owners into
+ *      cow_cache.owner_drain_state ('pending').
+ *   2. This worker claims owners (FOR UPDATE SKIP LOCKED), offset-walks their history
+ *      newest→oldest committing each page to cow_cache.composable_order and bumping
+ *      next_offset, and marks the owner 'complete' at the bottom.
+ *   3. The Ponder OwnerBackfill projection sees 'complete' owners, projects their cached
+ *      rows into discreteOrder (hash → current eventId), and flips historyBackfilled.
+ *
+ * The worker touches ONLY cow_cache — never Ponder's versioned schema — so blue-green
+ * deploys and reindexes are invisible to it. One long-lived worker serves every
+ * deployment. It is safe to run N-wide (SKIP LOCKED + stale-lease reclaim).
+ *
+ * Run with `pnpm drain` (dev) or as a sidecar container sharing DATABASE_URL (prod).
+ */
+
+import pg from "pg";
+import { drizzle } from "drizzle-orm/node-postgres";
+import type { Hex } from "viem";
+import { ORDERBOOK_API_URLS } from "../data";
+import {
+  DRAIN_IDLE_SLEEP_MS,
+  DRAIN_LEASE_TTL_MS,
+  DRAIN_OWNER_CONCURRENCY,
+  SIGNING_SCHEME_EIP1271,
+} from "../constants";
+import {
+  claimDrainOwners,
+  commitDrainProgress,
+  completeDrainOwner,
+  createCowCacheTables,
+  filterAndProcessForCache,
+  releaseDrainOwner,
+  upsertComposableCache,
+  type DrainClaim,
+  type DrizzleHandle,
+} from "../application/helpers/composableCache";
+import {
+  fetchAccountOrderPage,
+  OrderbookUnavailableError,
+  PAGE_LIMIT,
+} from "../application/helpers/orderbookHttp";
+import { TimeoutError } from "../application/helpers/withTimeout";
+import { log } from "../application/helpers/logger";
+
+const { Pool } = pg;
+
+function envInt(name: string, fallback: number): number {
+  const raw = Number(process.env[name]);
+  return Number.isFinite(raw) && raw > 0 ? raw : fallback;
+}
+
+const OWNER_CONCURRENCY = envInt("DRAIN_OWNER_CONCURRENCY", DRAIN_OWNER_CONCURRENCY);
+const IDLE_SLEEP_MS = envInt("DRAIN_IDLE_SLEEP_MS", DRAIN_IDLE_SLEEP_MS);
+const LEASE_TTL_MS = envInt("DRAIN_LEASE_TTL_MS", DRAIN_LEASE_TTL_MS);
+
+const nowSec = (): number => Math.floor(Date.now() / 1000);
+const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
+
+let shuttingDown = false;
+
+/**
+ * Drain one owner: offset-walk /account newest→oldest, committing each page to the
+ * durable cache and advancing next_offset, until a short page marks the bottom.
+ *
+ * Offset-based resume (not a MAX(creation_date) cursor) is required: with per-page
+ * commit, a max-cursor would declare the owner complete after re-reading only the
+ * newest page and silently drop the older pages. Re-fetches on restart are harmless —
+ * composable_order upserts by (chain_id, order_uid).
+ */
+async function drainOwner(db: DrizzleHandle, claim: DrainClaim): Promise<void> {
+  const { chainId, owner } = claim;
+  const apiBaseUrl = ORDERBOOK_API_URLS[chainId];
+  if (!apiBaseUrl) {
+    // No orderbook for this chain — nothing to drain. Mark complete so it stops being
+    // re-claimed; the projection will flip historyBackfilled against an empty cache.
+    log("warn", "drain:noApiUrl", { chainId, owner });
+    await completeDrainOwner(db, chainId, owner, nowSec());
+    return;
+  }
+
+  let offset = claim.nextOffset;
+  let pages = 0;
+  let cached = 0;
+
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
+    let page;
+    try {
+      page = await fetchAccountOrderPage(apiBaseUrl, owner, offset, PAGE_LIMIT, SIGNING_SCHEME_EIP1271);
+    } catch (err) {
+      if (err instanceof OrderbookUnavailableError || err instanceof TimeoutError) {
+        // Transient — return to the queue at the last committed offset. Re-claimed later.
+        log("warn", "drain:owner_interrupted", { chainId, owner, offset, err: err.name });
+        await releaseDrainOwner(db, chainId, owner, nowSec());
+        return;
+      }
+      throw err;
+    }
+
+    const rows = filterAndProcessForCache(chainId, page);
+    await upsertComposableCache(db, chainId, owner, rows);
+    cached += rows.length;
+
+    offset += page.length;
+    pages++;
+    await commitDrainProgress(db, chainId, owner, offset, nowSec());
+
+    if (page.length < PAGE_LIMIT) {
+      await completeDrainOwner(db, chainId, owner, nowSec());
+      log("info", "drain:owner_complete", { chainId, owner, pages, cached, offset });
+      return;
+    }
+  }
+}
+
+async function runLoop(db: DrizzleHandle): Promise<void> {
+  log("info", "drain:start", { concurrency: OWNER_CONCURRENCY, idleSleepMs: IDLE_SLEEP_MS, leaseTtlMs: LEASE_TTL_MS });
+
+  while (!shuttingDown) {
+    let claims: DrainClaim[];
+    try {
+      claims = await claimDrainOwners(db, {
+        batch: OWNER_CONCURRENCY,
+        nowSec: nowSec(),
+        staleBeforeSec: nowSec() - Math.floor(LEASE_TTL_MS / 1000),
+      });
+    } catch (err) {
+      log("error", "drain:claimFailed", { err: String(err) });
+      await sleep(IDLE_SLEEP_MS);
+      continue;
+    }
+
+    if (claims.length === 0) {
+      await sleep(IDLE_SLEEP_MS);
+      continue;
+    }
+
+    log("info", "drain:claimed", { owners: claims.length });
+
+    // Drain the batch concurrently; the batch size IS the concurrency bound. A single
+    // owner failing must not sink the batch, so isolate each with its own catch.
+    await Promise.all(
+      claims.map((claim) =>
+        drainOwner(db, claim).catch(async (err) => {
+          log("error", "drain:owner_failed", { chainId: claim.chainId, owner: claim.owner, err: String(err) });
+          // Unexpected error — return to the queue so the lease doesn't strand the owner
+          // until the TTL. next_offset is preserved, so the walk resumes.
+          await releaseDrainOwner(db, claim.chainId, claim.owner as Hex, nowSec()).catch(() => {});
+        }),
+      ),
+    );
+  }
+}
+
+async function main(): Promise<void> {
+  const connectionString = process.env.DATABASE_URL;
+  if (!connectionString) {
+    log("error", "drain:noDatabaseUrl", {});
+    process.exit(1);
+  }
+
+  const pool = new Pool({ connectionString });
+  const db = drizzle(pool);
+
+  // Idempotent — the tables also exist via the Ponder setup handler, but the worker may
+  // start first, so it must not depend on startup ordering.
+  await createCowCacheTables(db);
+
+  const stop = (signal: string) => {
+    if (shuttingDown) return;
+    log("info", "drain:shutdown", { signal });
+    shuttingDown = true;
+  };
+  process.on("SIGINT", () => stop("SIGINT"));
+  process.on("SIGTERM", () => stop("SIGTERM"));
+
+  try {
+    await runLoop(db);
+  } finally {
+    await pool.end().catch(() => {});
+    log("info", "drain:stopped", {});
+  }
+}
+
+main().catch((err) => {
+  log("error", "drain:fatal", { err: String(err) });
+  process.exit(1);
+});
diff --git a/tests/helpers/orderbookClient.test.ts b/tests/helpers/orderbookClient.test.ts
index 6f868e0..63166d6 100644
--- a/tests/helpers/orderbookClient.test.ts
+++ b/tests/helpers/orderbookClient.test.ts
@@ -19,7 +19,9 @@ vi.mock("ponder", () => ({
 
 import * as data from "../../src/data";
 import { ORDERBOOK_MAX_RETRIES } from "../../src/constants";
-import { fetchAccountOrders, fetchComposableOrders, fetchFlashLoanEnrichmentByUids, fetchOrderStatusByUids, fetchOwnerOrderStatuses } from "../../src/application/helpers/orderbookClient";
+import { fetchAccountOrders, fetchFlashLoanEnrichmentByUids, fetchOrderStatusByUids, fetchOwnerOrderStatuses } from "../../src/application/helpers/orderbookClient";
+import { fetchAccountOrderPage } from "../../src/application/helpers/orderbookHttp";
+import { filterAndProcessForCache } from "../../src/application/helpers/composableCache";
 
 // ─── Helpers ─────────────────────────────────────────────────────────────────
 
@@ -479,19 +481,17 @@ describe("fetchOwnerOrderStatuses", () => {
   });
 });
 
-// ─── fetchComposableOrders full-history drain tests ───────────────────────────
+// ─── drain worker: offset-walk page fetch ─────────────────────────────────────
 
-describe("fetchComposableOrders — full-history drain", () => {
+describe("fetchAccountOrderPage — single-page offset fetch", () => {
   const DRAIN_OWNER = "0x3333333333333333333333333333333333333333" as Hex;
 
-  it("paginates the full account history at limit=1000, past the old 100-order cap", async () => {
+  it("returns the raw page at the requested offset (untrimmed, for offset-walk termination)", async () => {
     const receivedOffsets: number[] = [];
     const receivedLimits = new Set<string>();
 
-    // 3 pages of the account endpoint: 1000 + 1000 + 500 = 2500 orders,
-    // mimicking a large-history owner. All are non-composable (signature "0x"
-    // decodes to null), so they filter out and no generator lookup is needed —
-    // but fetchAccountOrders still drains every page first.
+    // Mimic a large-history owner. The worker advances offset by the raw page length and
+    // stops on a short page, so each call must surface the untrimmed page.
     const { url, close } = await startServer((req, res) => {
       const parsed = new URL(req.url ?? "/", "http://127.0.0.1");
       const offset = parseInt(parsed.searchParams.get("offset") ?? "0", 10);
@@ -508,12 +508,13 @@ describe("fetchComposableOrders — full-history drain", () => {
 
     try {
       await withFakeApi(TEST_CHAIN_ID, url, async () => {
-        await fetchComposableOrders(makeContext(), TEST_CHAIN_ID, DRAIN_OWNER);
+        const apiBaseUrl = data.ORDERBOOK_API_URLS[TEST_CHAIN_ID]!;
+        const first = await fetchAccountOrderPage(apiBaseUrl, DRAIN_OWNER, 0, 1000);
+        const last = await fetchAccountOrderPage(apiBaseUrl, DRAIN_OWNER, 2000, 1000);
 
-        // Old behavior capped at 4 pages × 25 = offsets 0,25,50,75 — never past 100.
-        expect(receivedOffsets).toContain(0);
-        expect(receivedOffsets).toContain(1000);
-        expect(receivedOffsets).toContain(2000);
+        expect(first).toHaveLength(1000); // full page → walk continues
+        expect(last).toHaveLength(500); // short page → walk stops here
+        expect(receivedOffsets).toEqual([0, 2000]);
         expect(receivedLimits).toEqual(new Set(["1000"]));
       });
     } finally {
@@ -522,77 +523,20 @@ describe("fetchComposableOrders — full-history drain", () => {
   });
 });
 
-// ─── fetchComposableOrders durable-cache rebuild test ─────────────────────────
-
-describe("fetchComposableOrders — rebuild from durable cache", () => {
-  const OWNER = "0x2222222222222222222222222222222222222222" as Hex;
-  const GEN_HASH = `0x${"cc".repeat(32)}`;
-
-  // Simulates a post-reindex deploy: discreteOrder is empty, but cow_cache.composable_order
-  // still holds the owner's history and the orderbook has nothing new past the cursor.
-  function makeIncrementalContext(opts: {
-    cursor: string;
-    cacheRows: Record<string, unknown>[];
-    generators: { eventId: string; hash: string }[];
-  }) {
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    const select = (proj: any) => ({
-      from: () => ({
-        where: async () => {
-          if (proj.cursor !== undefined) return [{ cursor: opts.cursor }];
-          if (proj.eventId !== undefined) return opts.generators;
-          return opts.cacheRows;
-        },
-      }),
-    });
-    return {
-      db: { sql: {
-        select,
-        insert: () => ({ values: () => ({ onConflictDoUpdate: async () => {} }) }),
-      } },
-    };
-  }
-
-  it("returns the cached history re-mapped to the current generator eventId when no new orders exist", async () => {
-    // Orderbook returns nothing newer than the cursor.
-    const { url, close } = await startServer((_req, res) => {
-      res.writeHead(200, { "content-type": "application/json" });
-      res.end("[]");
-    });
-
-    const cacheRows = [
-      {
-        orderUid: "0xcached-order",
-        generatorHash: GEN_HASH,
-        orderType: "PerpetualSwap",
-        status: "fulfilled",
-        sellAmount: "1000",
-        buyAmount: "2000",
-        feeAmount: "0",
-        validTo: 9999999999,
-        creationDate: 1700000000n,
-        executedSellAmount: "1000",
-        executedBuyAmount: "2000",
-      },
-    ];
-    const ctx = makeIncrementalContext({
-      cursor: "1700000000",
-      cacheRows,
-      // eventId differs from any prior deployment — the row is keyed by the stable hash.
-      generators: [{ eventId: "gen-current", hash: GEN_HASH }],
-    });
-
-    try {
-      await withFakeApi(TEST_CHAIN_ID, url, async () => {
-        const { orders } = await fetchComposableOrders(ctx, TEST_CHAIN_ID, OWNER);
-        expect(orders).toHaveLength(1);
-        expect(orders[0]!.uid).toBe("0xcached-order");
-        expect(orders[0]!.generatorId).toBe("gen-current");
-        expect(orders[0]!.status).toBe("fulfilled");
-      });
-    } finally {
-      await close();
-    }
+// ─── filterAndProcessForCache: composable filtering (no DB) ───────────────────
+
+describe("filterAndProcessForCache", () => {
+  it("drops non-eip1271 and undecodable orders (signature '0x' → null)", () => {
+    const rows = filterAndProcessForCache(TEST_CHAIN_ID, [
+      // Non-eip1271 → dropped before decode.
+      { ...makeOrderStub({ uid: UID_A, status: "open" }), signingScheme: "eip712" },
+      // eip1271 but signature "0x" decodes to null → dropped.
+      makeOrderStub({ uid: UID_B, status: "open" }),
+      // presignaturePending is skipped explicitly.
+      makeOrderStub({ uid: `0x${"cc".repeat(56)}`, status: "presignaturePending" as never }),
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    ] as any);
+    expect(rows).toEqual([]);
   });
 });