feat(filler): opt-in partition-parallel drain for large sales-filter backlogs#71
feat(filler): opt-in partition-parallel drain for large sales-filter backlogs#71igorls wants to merge 1 commit into
Conversation
There was a problem hiding this comment.
Pull request overview
Note
Copilot was unable to run its full agentic suite in this review.
Adds an opt-in, partition-parallel draining path to speed up recovery from large atomicmarket_sales_filters_updates backlogs, including SQL functions, an operator CLI, and integration coverage for lock/disjointness/parity guarantees.
Changes:
- Introduces
update_atomicmarket_sales_filters_partitionandnormalize_atomicmarket_sales_filters_offers, plus an index to speed offer→sale resolution. - Adds a
drain-sales-filtersCLI to run multi-worker partition drains (and optional offer normalization). - Adds integration tests validating parity with the stock drain, partition scoping, and advisory-lock behavior.
Reviewed changes
Copilot reviewed 3 out of 4 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| src/filler/handlers/atomicmarket/sales-filters-parallel-drain.integration.test.ts | New integration tests for partition drain semantics, parity, and advisory-lock protocol. |
| src/bin/drain-sales-filters.ts | New operator CLI to normalize offer rows and drain queued sale rows in parallel by hash partition. |
| definitions/migrations/1.7.13/database.sql | Bumps DB version to 1.7.13 with brief migration notes. |
| definitions/migrations/1.7.13/atomicmarket.sql | Adds index + new PL/pgSQL functions implementing normalization and partition-parallel draining. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
…tionale Review follow-ups on atomicassets#71 (comment-only, no behavior change): - Replace the loose 'pre-build under the same names' note with the actual ON ONLY + CREATE INDEX CONCURRENTLY + ATTACH PARTITION recipe (attachment matches on definition equivalence, not names) as used for the production validation run. - State the plain-temp-table rationale (1.7.11 precedent: reused pool connections, no DROP-IF-EXISTS notice spam; the claimed (key,seq) set must survive across statements so a CTE chain cannot replace it; per-call DDL is negligible at drain cadence).
|
v2 port opened as a separate PR per the rc2 cherry-pick workflow: the same three commits apply cleanly onto |
robrigo
left a comment
robrigo
left a comment
There was a problem hiding this comment.
Review (FACINGS) — LGTM with one merge-sequencing flag.
Verified mechanically, not just by reading:
- The byte-for-byte recompute claim checks out. Extracted the recompute block from
1.7.11/atomicmarket.sqland fromupdate_atomicmarket_sales_filters_partition, diffed them: the only deltas are the RELEASE section (workers delete only their claimed_part_sales; stock deletes its three claim tables) and comments. The filter math,create_atomicmarket_sales_filtercall, and nx/nb flags are identical. - Lock semantics verified against the 1.7.11 source: stock takes
pg_try_advisory_xact_lock(hashtext('update_atomicmarket_sales_filters'))(exclusive); workers/normalizer take the same key_sharedplus their own exclusive sub-keys. Shared-shared coexists, shared-exclusive conflicts — matches the documented matrix, and the integration tests exercise it on real concurrent backends. - Ran the suite on PostgreSQL 18 (our production CNPG major): 9/9 new parallel-drain tests pass (incl. the PARITY byte-identical check), full integration 332 passing, typecheck clean.
- The hardcoded
/home/node/app/config/...path matches the existingfiller.ts/server.tsconvention; the index pre-build recipe mirrors the 1.7.12 pattern. Worker disjointness reasoning (sale-keyed writes, asset rows left to the stock drain) is sound.
One sequencing hazard before merge: main has no 1.7.12 yet — that's #70, still open. If this merges first and any DB upgrades to 1.7.13, the runner's strictly-greater-than filter means #70's 1.7.12 would be skipped forever on that DB when it lands later. Merge #70 before #71, or renumber this one above whatever #70 ships as.
The production numbers (24.9M backlog, weeks → 4 minutes) speak for themselves. Nice work.
…tionale Review follow-ups on #71 (comment-only, no behavior change): - Replace the loose 'pre-build under the same names' note with the actual ON ONLY + CREATE INDEX CONCURRENTLY + ATTACH PARTITION recipe (attachment matches on definition equivalence, not names) as used for the production validation run. - State the plain-temp-table rationale (1.7.11 precedent: reused pool connections, no DROP-IF-EXISTS notice spam; the claimed (key,seq) set must survive across statements so a CTE chain cannot replace it; per-call DDL is negligible at drain cadence).
….0.1 The v2 line's migration chain is at 2.0.0 (rc tags already initialize databases at that version), so the main-line 1.7.13 directory would sort before 2.0.0 and never apply to them. Renumbered to 2.0.1; SQL content unchanged from PR #71.
port(v2): partition-parallel drain for large sales-filter backlogs (#71 on the v2 line)
…backlogs A deep catchup or bulk load leaves atomicmarket_sales_filters_updates with a multi-million-row backlog that the (correctly) single-flight drain clears at random-read I/O speed - days to weeks of stale /v2/sales filters. The recompute parallelizes perfectly across disjoint sales: * update_atomicmarket_sales_filters_partition(part_count, part_index, batch): the stock drain restricted to sale rows with sale_id % part_count = part_index. Recompute CTE block is byte-for-byte the stock 1.7.11 recompute. Workers take the stock drain's advisory key SHARED plus an exclusive per-partition sub-lock: N workers coexist, never with the stock drain, duplicates no-op. Nothing runs unless an operator launches it. * normalize_atomicmarket_sales_filters_offers(batch): converts queued offer rows into equivalent queued sale rows so workers can own them; seq-guard protocol. * index atomicmarket_sales_filters (assets_contract, offer_id): turns the drain's offer->sales resolution from a multi-GB hash join into index probes. Also speeds the stock drain. * bin/drain-sales-filters: operator runner - normalizes offers, runs N partition workers to empty, ANALYZEs after. Additive only (new functions + index + CLI); does not redefine the stock drain, so it composes with the 1.7.13 two-lane prio queue (atomicassets#73). Every queue interaction uses explicit column lists and key/seq deletes, so the prio column added by 1.7.13 is transparent here. The partition CLI claims by hash partition + seq (prio-agnostic) by design: it is an operator bulk-recovery tool where draining the whole backlog matters, not steady-state freshness ordering. Rebased onto main and renumbered 1.7.13 -> 1.7.18: the original 1.7.13 slot was taken upstream by atomicassets#73, and atomicassets#70's transfers-PK converge already landed as 1.7.12 - which resolves the merge-sequencing hazard flagged in review. 1.7.18 sorts after the current head migration (1.7.17). Validated against a production WAX dataset (multi-million-row backlog).
f1b89fe to
63fbf1b
Compare
|
Rebased onto current
Compatibility with #73's prio lane (which merged after your review): this PR is additive only — it adds The two prior commits are collapsed into one clean commit on top of |
Problem
A deep catchup, bulk load, or extended drain outage leaves
atomicmarket_sales_filters_updateswith a multi-million-row backlog. The drain is (correctly) single-flight — the 1.7.11 advisory lock exists because concurrent unpartitioned drains could recompute the same sale from two backends and deadlock onatomicmarket_sales_filters— but that leaves no recovery path that uses the hardware: the backlog clears at single-connection random-read speed while/v2/salesfilters go stale. On our production WAX deployment (460M assets), a 24.9M-row post-catchup backlog drained at a net ~17 rows/s ≈ weeks.The recompute parallelizes perfectly across disjoint sales, which hash-partitioning the queue provides.
Changes (no steady-state behavior change; nothing runs unless an operator launches it)
update_atomicmarket_sales_filters_partition(part_count, part_index, batch_size)— the 1.7.11 drain restricted to sale rows withsale_id % part_count = part_index. The recompute CTE block is byte-for-byte 1.7.11 (preserving the 1.3.3-identical output guarantee); only the claim predicate and lock protocol differ.normalize_atomicmarket_sales_filters_offers(batch_size)— converts queued offer rows into the equivalent queued sale rows (the same offer→sales resolution the stock drain performs per batch), so partition workers can own them. 1.7.11 seq-guard claim/release protocol.atomicmarket_sales_filters (assets_contract, offer_id)— the drain's offer→sales resolution was a hash join scanning the multi-GB filter partitions on every batch containing offer rows; now it is index probes. This also speeds the stock drain. (Live-DB recipe: pre-build per-partitionCONCURRENTLYunder the same names; the migration'sCREATE INDEX IF NOT EXISTSthen attaches without rebuilding — documented in the migration header.)bin/drain-sales-filters.js [--partitions N] [--batch B] [--skip-offers]— operator runner: normalizes offers, runs N partition workers until empty, ANALYZEs the churned tables after.Lock protocol
Workers coexist with each other (shared-shared); the stock drain no-ops while any worker batch is in flight and vice versa (its normal contended path — the filler retries next cycle); duplicate workers on a partition no-op. All xact-scoped try-locks: no waiting, nothing orphaned on crash, every claim/recompute/release is one crash-safe transaction exactly like 1.7.11. Workers never claim asset rows (one asset fans out to arbitrary sales — not partitionable), so the stock drain keeps owning that path.
Validation
Integration tests (
sales-filters-parallel-drain.integration.test.ts, full suite 332 passing):to_jsonbequality after delete + re-enqueue + partition drain).Production (WAX mainnet, 460M assets, live filler at head throughout):
Notes for review
1.7.13to avoid colliding with fix(migration): converge legacy atomicassets_transfers PK to (contract, transfer_id) #70's1.7.12; renumber freely.reconcilefunction in a follow-up if there's interest.