diff --git a/app/db/concurrent-indexes.sql b/app/db/concurrent-indexes.sql index fe9157db..f3899ab8 100644 --- a/app/db/concurrent-indexes.sql +++ b/app/db/concurrent-indexes.sql @@ -24,3 +24,6 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS bb_launch_swaps_usd_missing_idx ON bb_la -- the site totals' read of quote-side fees with no USD value (getLaunchTotals): neither scans its whole table on a page view CREATE INDEX CONCURRENTLY IF NOT EXISTS bb_launch_swaps_usd_missing_token_idx ON bb_launch_swaps (chain_id, token) WHERE usd IS NULL; CREATE INDEX CONCURRENTLY IF NOT EXISTS bb_launch_fee_events_usd_missing_idx ON bb_launch_fee_events (chain_id, token) WHERE usd IS NULL AND kind IN ('collected', 'burned'); +-- trader healing (2026-10-20): the rows whose transaction could not be read at index time; healSwapTraders pages this +-- set oldest-first by (block, log), and the partial index stays empty once the backlog is drained +CREATE INDEX CONCURRENTLY IF NOT EXISTS bb_launch_swaps_trader_null_idx ON bb_launch_swaps (chain_id, block_number, log_index) WHERE trader IS NULL; diff --git a/app/src/lib/launchpad/boundedMap.test.ts b/app/src/lib/launchpad/boundedMap.test.ts new file mode 100644 index 00000000..343413ad --- /dev/null +++ b/app/src/lib/launchpad/boundedMap.test.ts @@ -0,0 +1,39 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { setBounded } from "./boundedMap.ts"; + +test("expired entries are dropped before an active one is evicted", () => { + const map = new Map([ + ["expired", 1], + ["live", 100], + ]); + setBounded(map, "new", 100, 2, (value, now) => value <= now, 50); + assert.deepEqual([...map.keys()], ["live", "new"]); +}); + +test("a full map with nothing expired evicts the oldest insertion", () => { + const map = new Map([ + ["a", 1], + ["b", 2], + ]); + setBounded(map, "c", 3, 2, () => false, 0); + assert.deepEqual([...map.keys()], ["b", "c"]); +}); + +test("refreshing an existing key moves it out of the eviction lane", () => { + const map = new Map([ + ["a", 1], + ["b", 2], + ]); + setBounded(map, "a", 3, 2, () => false, 0); + assert.deepEqual([...map.keys()], ["b", "a"]); + setBounded(map, "c", 4, 2, () => false, 0); + assert.deepEqual([...map.keys()], ["a", "c"], "b was the oldest, not the refreshed a"); +}); + +test("a burst of new keys stays at the bound", () => { + const map = new Map(); + for (let i = 0; i < 100; i++) setBounded(map, `k${i}`, i, 10, () => false, 0); + assert.equal(map.size, 10); + assert.deepEqual([...map.keys()], Array.from({ length: 10 }, (_, i) => `k${90 + i}`)); +}); diff --git a/app/src/lib/launchpad/boundedMap.ts b/app/src/lib/launchpad/boundedMap.ts new file mode 100644 index 00000000..e4150174 --- /dev/null +++ b/app/src/lib/launchpad/boundedMap.ts @@ -0,0 +1,25 @@ +/** + * Put a key into a bounded retry map without throwing the map away. On overflow, entries the caller marks expired are + * dropped first; if the map is still full, the oldest insertions go. Refreshing an existing key also refreshes its + * insertion order, so a row that is actively being retried is not the next one evicted. + */ +export function setBounded( + map: Map, + key: K, + value: V, + max: number, + expired: (value: V, now: number) => boolean, + now: number, +): void { + if (max < 1) return; + if (!map.has(key) && map.size >= max) { + for (const [k, v] of map) if (expired(v, now)) map.delete(k); + while (map.size >= max) { + const oldest = map.keys().next().value; + if (oldest === undefined) break; + map.delete(oldest); + } + } + map.delete(key); + map.set(key, value); +} diff --git a/app/src/lib/launchpad/indexer-durability.test.ts b/app/src/lib/launchpad/indexer-durability.test.ts new file mode 100644 index 00000000..c93f0a15 --- /dev/null +++ b/app/src/lib/launchpad/indexer-durability.test.ts @@ -0,0 +1,55 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import test from "node:test"; + +/** + * Source contracts for the indexer durability changes. The paging planner itself is unit-tested in + * traderHeal.test.ts; these assertions pin the SQL and the once-at-a-time heal wiring that node --test cannot import + * (indexer.ts is server-only). + */ +const read = (file: string) => readFileSync(new URL(file, import.meta.url), "utf8").replaceAll("\r\n", "\n"); +const indexer = read("./indexer.ts"); +const indexes = read("../../../db/concurrent-indexes.sql"); + +test("cursor and head updates are monotonic across overlapping sync machines", () => { + assert.match(indexer, /cursor_block = GREATEST\(cursor_block, \$\{to\}\)/); + assert.match(indexer, /head_block = GREATEST\(COALESCE\(head_block, 0\), \$\{head\}\)/); + assert.match(indexer, /last_error = CASE WHEN cursor_block <= \$\{to\} THEN NULL ELSE last_error END/); + assert.match(indexer, /head_block = CASE WHEN \$\{knownHead\}::bigint IS NULL THEN head_block ELSE GREATEST\(COALESCE\(head_block, 0\), \$\{knownHead\}::bigint\) END/); +}); + +test("each chain's heal pass runs once at a time, and a running one is skipped, never awaited", () => { + // runOnce.test.ts covers the helper's behaviour; this pins that pollAll uses it for the heals and nothing else + assert.match(indexer, /const healsRunning = new Map>\(\);/); + assert.match(indexer, /out\[c\] = await pollLaunches\(c\)/); + assert.match(indexer, /await startUnlessRunning\(healsRunning, c, \(\) => healChain\(c\)\);/); + assert.ok(indexer.indexOf("out[c] = await pollLaunches(c)") < indexer.indexOf("startUnlessRunning(healsRunning"), "the chain is polled before its heals"); + // the poll itself is shared per chain: a caller joins a run() still in flight instead of starting a second one + assert.match(indexer, /export async function pollLaunches\(chain: ChainKey\): Promise \{\n const cur = inFlight\.get\(chain\);\n if \(cur\) return cur;/); + // the whole pass is not single-flight: one slow chain or a stuck heal must not hold the other chains' next poll + assert.doesNotMatch(indexer, /allInFlight/); +}); + +test("trader healing pages oldest-first with a keyset, not the newest 100 rows", () => { + assert.match(indexer, /\(block_number, log_index\) > \(\$\{after\.block\}, \$\{after\.log\}\)/); + assert.match(indexer, /ORDER BY block_number ASC, log_index ASC LIMIT \$\{HEAL_TRADER_PAGE\}/); + assert.doesNotMatch(indexer, /ORDER BY block_number DESC LIMIT 100/); + assert.match(indexer, /if \(plan\.cursor\) healTraderCursor\.set\(chain, plan\.cursor\)/); + assert.match(indexer, /healTraderCursor\.delete\(chain\)/); +}); + +test("retry state is pruned, never cleared wholesale", () => { + assert.match(indexer, /function restHeal\(key: string\): void/); + assert.match(indexer, /function noteAttributionTry\(key: string, n: number\): void/); + assert.match(indexer, /setBounded\(healRestUntil,/); + assert.match(indexer, /setBounded\(attributionTries,/); + assert.doesNotMatch(indexer, /healRestUntil\.clear\(\)/); + assert.doesNotMatch(indexer, /attributionTries\.clear\(\)/); +}); + +test("the partial index serves the missing-trader page and order", () => { + assert.match( + indexes, + /CREATE INDEX CONCURRENTLY IF NOT EXISTS bb_launch_swaps_trader_null_idx ON bb_launch_swaps \(chain_id, block_number, log_index\) WHERE trader IS NULL;/, + ); +}); diff --git a/app/src/lib/launchpad/indexer.ts b/app/src/lib/launchpad/indexer.ts index 1bd38a97..4516d306 100644 --- a/app/src/lib/launchpad/indexer.ts +++ b/app/src/lib/launchpad/indexer.ts @@ -10,6 +10,9 @@ import { DEAD_ADDR, ZERO_ADDR } from "./holders"; import { SYNC_CHUNK_BLOCKS, SYNC_MAX_CHUNKS_PER_CALL, syncOverlapBlocks } from "@/lib/config"; import { byChainOrder, fetchBySelectors, fetchLogsSplit, isRangeTooLarge } from "./log-range"; import { publicError, redactUrls } from "./redact"; +import { setBounded } from "./boundedMap"; +import { planTraderHeal, type HealSwapRow, type TraderHealCursor } from "./traderHeal"; +import { startUnlessRunning } from "./runOnce"; import { ENTRY_POINTS, attributeSwap, type ReceiptLog } from "./attribution"; import { appTagOf, swapInApp, type AppTag } from "./appTag"; import { BUILDER_DATA_SUFFIX } from "@/lib/chainPublic"; @@ -412,8 +415,30 @@ const ATTRIBUTION_MAX_TRIES = 3; const ATTRIBUTION_TRY_GAP_MS = 10 * 60_000; // tries this far apart: a short RPC outage never freezes a row const ATTRIBUTION_PASS_BUDGET_MS = 5_000; // chain reads per pass stop here, so a long backlog never holds up indexing const ATTRIBUTION_IDLE_MS = 5 * 60_000; // a pass that found nothing to check rests this long (per chain, per machine) +const ATTRIBUTION_TRIES_MAX = 20_000; const attributionIdleUntil = new Map(); const healRestUntil = new Map(); // `${chain}:${tx}` whose transaction could not be read → next try +const HEAL_REST_MAX = 5_000; +const HEAL_TRADER_PAGE = 500; // rows read per tick; only 20 transactions are actually read +const HEAL_TRADER_READS = 20; +// The next trader-heal page starts after this (block, log) pair. A page that is all parked rows still advances, so +// the rows behind it are reached on the next tick instead of the same newest 100 rows being selected forever. +const healTraderCursor = new Map(); + +/** + * Rest a heal key (a transaction the trader or app-tag heal could not read) for the retry gap. The map is bounded + * without clearing it: expired entries go first, then the oldest insertions. + */ +function restHeal(key: string): void { + const now = Date.now(); + setBounded(healRestUntil, key, now + ATTRIBUTION_TRY_GAP_MS, HEAL_REST_MAX, (until) => until <= now, now); +} + +/** Same bound for the attribution retry counters: expired backoff state is dropped, never the whole map. */ +function noteAttributionTry(key: string, n: number): void { + const now = Date.now(); + setBounded(attributionTries, key, { n, at: now }, ATTRIBUTION_TRIES_MAX, (v) => now - v.at >= ATTRIBUTION_TRY_GAP_MS, now); +} /** * Swaps not checked yet (trader_via NULL, or RECEIPT_PENDING) get checked a batch per poll: always for the last day @@ -479,8 +504,7 @@ export async function attributeBacklog(chain: ChainKey, batch = 100): Promise 20_000) attributionTries.clear(); - attributionTries.set(key, { n, at: Date.now() }); + noteAttributionTry(key, n); if (n >= ATTRIBUTION_MAX_TRIES) { await db`UPDATE bb_launch_swaps SET trader_via = 'unread', tx_from = COALESCE(tx_from, trader) WHERE chain_id = ${cid} AND tx_hash = ${r.tx_hash} AND log_index = ${r.log_index} AND (trader_via IS NULL OR trader_via = ${RECEIPT_PENDING})`; attributionTries.delete(key); @@ -703,7 +727,17 @@ export async function pollLaunches(chain: ChainKey): Promise { return p; } -/** Poll every configured chain (used by the loop and POST /api/launch/sync without a tx). */ +/** Chains whose heal and backfill pass is running in this process. */ +const healsRunning = new Map>(); + +/** + * Poll every configured chain (used by the loop and POST /api/launch/sync without a tx). + * + * Overlapping callers share work per chain, not the pass as a whole: a chain whose run() is in flight is joined + * (pollLaunches), and a chain whose heal pass is running is skipped (healsRunning). A chain whose run has finished is + * polled again by the next caller, which reads only the blocks since, as every tick does. Sharing the whole pass + * instead would make each chain wait for the slowest one, and a stuck query on one chain would stop indexing on all. + */ export async function pollAll(): Promise> { const out: Record = {}; await Promise.all( @@ -714,17 +748,26 @@ export async function pollAll(): Promise> { console.warn(`[launch-sync] poll ${c}:`, redactUrls(errMessage(e))); return { status: "skipped", reason: publicError(e, "sync failed") }; }); - await healLaunchReads(c).catch((e) => console.warn(`[launch-sync] heal ${c}:`, redactUrls(errMessage(e)))); - await healQuoteTokens(c).catch((e) => console.warn(`[launch-sync] quote tokens ${c}:`, redactUrls(errMessage(e)))); - await healSwapTraders(c).catch((e) => console.warn(`[launch-sync] heal traders ${c}:`, redactUrls(errMessage(e)))); - await attributeBacklog(c).catch((e) => console.warn(`[launch-sync] attribute swaps ${c}:`, redactUrls(errMessage(e)))); - await healAppTags(c).catch((e) => console.warn(`[launch-sync] app tags ${c}:`, redactUrls(errMessage(e)))); - await backfillHolders(c).catch((e) => console.warn(`[launch-sync] holders backfill ${c}:`, redactUrls(errMessage(e)))); + // pollLaunches already shares one run() per chain. The heals and backfills are as expensive, so they run one pass + // per chain at a time too: the 15s loop's ticks overlap when a pass is slow, and a web sync can land beside one. + // A caller that finds a pass still running skips it instead of waiting, so a slow or stuck heal never holds up + // indexing on the next tick. + await startUnlessRunning(healsRunning, c, () => healChain(c)); }), ); return out; } +/** One chain's heals and backfills, in order; each logs its own failure and the next still runs. */ +async function healChain(c: ChainKey): Promise { + await healLaunchReads(c).catch((e) => console.warn(`[launch-sync] heal ${c}:`, redactUrls(errMessage(e)))); + await healQuoteTokens(c).catch((e) => console.warn(`[launch-sync] quote tokens ${c}:`, redactUrls(errMessage(e)))); + await healSwapTraders(c).catch((e) => console.warn(`[launch-sync] heal traders ${c}:`, redactUrls(errMessage(e)))); + await attributeBacklog(c).catch((e) => console.warn(`[launch-sync] attribute swaps ${c}:`, redactUrls(errMessage(e)))); + await healAppTags(c).catch((e) => console.warn(`[launch-sync] app tags ${c}:`, redactUrls(errMessage(e)))); + await backfillHolders(c).catch((e) => console.warn(`[launch-sync] holders backfill ${c}:`, redactUrls(errMessage(e)))); +} + /** * Self-heal rows whose eth_calls failed at index time (a launch is indexed in the same block it is mined; * a lagging RPC node then reverts `name()`/`symbol()`/`recipientsOf()` and the fallback "?" / [] was stored). @@ -812,40 +855,47 @@ export async function healQuoteTokens(chain: ChainKey): Promise { /** * Self-heal swaps stored without a trader (the sender lookup failed at index time, typically a rate-limited * node during a busy launch block). Those rows carry no wallet facts: the holder panel skips them and the - * ranking cannot tell whether they were outside trades. Re-reads a bounded batch of the newest ones per poll. + * ranking cannot tell whether they were outside trades. Reads a bounded page oldest-first per poll. + * + * A plain newest-first LIMIT 100 could not get past a head of unreadable transactions: they were parked in + * healRestUntil, but the same 100 rows were selected again every poll, so older rows were never reached. The + * keyset cursor advances past every visited transaction group (parked or not), wraps at the end, and reaches + * the whole set. The partial index in db/concurrent-indexes.sql serves the filter and the order. */ export async function healSwapTraders(chain: ChainKey): Promise { const db = maybeDb(); if (!db || skipReason(chain)) return 0; const cid = chainIdOf(chain); - const rows = await db<{ tx_hash: string; log_index: number; token: string }[]>` - SELECT tx_hash, log_index, token FROM bb_launch_swaps - WHERE chain_id = ${cid} AND trader IS NULL ORDER BY block_number DESC LIMIT 100`; + const cursor = healTraderCursor.get(chain) ?? { block: 0n, log: -1 }; + const page = async (after: { block: bigint; log: number }): Promise => db` + SELECT tx_hash, log_index, token, block_number FROM bb_launch_swaps + WHERE chain_id = ${cid} AND trader IS NULL + AND (block_number, log_index) > (${after.block}, ${after.log}) + ORDER BY block_number ASC, log_index ASC LIMIT ${HEAL_TRADER_PAGE}`; + let rows = await page(cursor); + if (rows.length === 0 && (cursor.block > 0n || cursor.log >= 0)) { + // End of the set: wrap, so parked rows and rows added behind the cursor get another look. + healTraderCursor.delete(chain); + rows = await page({ block: 0n, log: -1 }); + } if (rows.length === 0) return 0; - // one read per transaction (a transaction with many swaps no longer takes every slot and every call), up to 20 - // transactions a poll, and one that could not be read rests ten minutes before it is read again - const byTx = new Map(); - for (const r of rows) byTx.set(r.tx_hash, [...(byTx.get(r.tx_hash) ?? []), r]); + const plan = planTraderHeal(rows, healRestUntil, chain, Date.now(), HEAL_TRADER_READS); let healed = 0; - let read = 0; - for (const [tx, swaps] of byTx) { - if (read >= 20) break; - const key = `${chain}:${tx}`; - if (Date.now() < (healRestUntil.get(key) ?? 0)) continue; - read++; - if (!(await txOf(chain, tx as Hex))) { - if (healRestUntil.size > 5_000) healRestUntil.clear(); - healRestUntil.set(key, Date.now() + ATTRIBUTION_TRY_GAP_MS); + for (const group of plan.groups) { + if (group.resting) continue; + if (!(await txOf(chain, group.tx as Hex))) { + restHeal(group.key); continue; } - healRestUntil.delete(key); - for (const r of swaps) { + healRestUntil.delete(group.key); + for (const r of group.swaps) { const who = await traderOf(chain, r.tx_hash as Hex, Number(r.log_index), r.token); // the transaction is cached now if (!who.trader) continue; await db`UPDATE bb_launch_swaps SET trader = ${who.trader}, tx_from = ${who.tx_from}, trader_via = ${who.via}, via_app = ${who.app} WHERE chain_id = ${cid} AND tx_hash = ${r.tx_hash} AND log_index = ${r.log_index} AND trader IS NULL`; healed++; } } + if (plan.cursor) healTraderCursor.set(chain, plan.cursor); if (healed) console.log(`[launch-sync] healed traders on ${healed} swap(s) on ${chain}`); return healed; } @@ -892,8 +942,7 @@ export async function healAppTags(chain: ChainKey): Promise { read++; const tx = await txOf(chain, hash as Hex); if (!tx || tx.tag === null) { - if (healRestUntil.size > 5_000) healRestUntil.clear(); - healRestUntil.set(key, Date.now() + ATTRIBUTION_TRY_GAP_MS); + restHeal(key); continue; } healRestUntil.delete(key); @@ -936,7 +985,7 @@ async function run(chain: ChainKey): Promise { totals.swaps += r.swaps; totals.fees += r.fees; totals.boosts += r.boosts; - await db`UPDATE bb_launch_sync_cursor SET cursor_block = ${to}, head_block = ${head}, last_run_at = now(), last_error = NULL WHERE chain_id = ${cid}`; + await db`UPDATE bb_launch_sync_cursor SET cursor_block = GREATEST(cursor_block, ${to}), head_block = GREATEST(COALESCE(head_block, 0), ${head}), last_run_at = now(), last_error = CASE WHEN cursor_block <= ${to} THEN NULL ELSE last_error END WHERE chain_id = ${cid}`; from = to + 1n; chunks++; } @@ -946,7 +995,7 @@ async function run(chain: ChainKey): Promise { } catch (err) { const msg = errMessage(err); // the stored message reaches /api/health unauthenticated: never with the upstream URL (a keyed provider URL carries its API key) - await db`UPDATE bb_launch_sync_cursor SET last_run_at = now(), head_block = COALESCE(${knownHead}::bigint, head_block), last_error = ${redactUrls(msg).slice(0, 500)} WHERE chain_id = ${cid}`.catch(() => {}); + await db`UPDATE bb_launch_sync_cursor SET last_run_at = now(), head_block = CASE WHEN ${knownHead}::bigint IS NULL THEN head_block ELSE GREATEST(COALESCE(head_block, 0), ${knownHead}::bigint) END, last_error = ${redactUrls(msg).slice(0, 500)} WHERE chain_id = ${cid}`.catch(() => {}); throw err; } } diff --git a/app/src/lib/launchpad/runOnce.test.ts b/app/src/lib/launchpad/runOnce.test.ts new file mode 100644 index 00000000..a9425f28 --- /dev/null +++ b/app/src/lib/launchpad/runOnce.test.ts @@ -0,0 +1,36 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { startUnlessRunning } from "./runOnce.ts"; + +test("a key that is still running is skipped, not run twice and not waited on", async () => { + const running = new Map>(); + let calls = 0; + let finish!: () => void; + const first = startUnlessRunning(running, "base", () => { + calls++; + return new Promise((resolve) => (finish = resolve)); + }); + assert.ok(first); + assert.equal(startUnlessRunning(running, "base", async () => { calls++; }), null, "the second caller gets null at once"); + assert.equal(calls, 1); + finish(); + await first; + assert.equal(running.size, 0, "the key is free once its run settles"); + await startUnlessRunning(running, "base", async () => { calls++; }); + assert.equal(calls, 2); +}); + +test("a run that never settles holds up neither another key nor a later caller of its own", async () => { + const running = new Map>(); + startUnlessRunning(running, "base", () => new Promise(() => {})); + let arc = 0; + await startUnlessRunning(running, "arc", async () => { arc++; }); + assert.equal(arc, 1, "another key runs as usual"); + assert.equal(startUnlessRunning(running, "base", async () => {}), null, "the stuck key is skipped, and nothing awaits it"); +}); + +test("a run that rejects frees its key and passes the rejection to its caller", async () => { + const running = new Map>(); + await assert.rejects(startUnlessRunning(running, "base", async () => { throw new Error("boom"); })!, /boom/); + assert.equal(running.size, 0); +}); diff --git a/app/src/lib/launchpad/runOnce.ts b/app/src/lib/launchpad/runOnce.ts new file mode 100644 index 00000000..49a0716a --- /dev/null +++ b/app/src/lib/launchpad/runOnce.ts @@ -0,0 +1,15 @@ +/** Pure (node --test loads it directly): one run per key at a time, never queued behind a run that may be stuck. */ + +/** + * Start `fn` for `key` unless a run for that key is still going. A key already running is skipped, not waited on, and + * the caller gets null: two callers never do the same work side by side, and a slow or stuck run never holds up the + * callers after it. The key is free again once its run settles, whether it resolved or rejected. + */ +export function startUnlessRunning(running: Map>, key: K, fn: () => Promise): Promise | null { + if (running.has(key)) return null; + const p: Promise = fn().finally(() => { + if (running.get(key) === p) running.delete(key); + }); + running.set(key, p); + return p; +} diff --git a/app/src/lib/launchpad/traderHeal.test.ts b/app/src/lib/launchpad/traderHeal.test.ts new file mode 100644 index 00000000..746f6599 --- /dev/null +++ b/app/src/lib/launchpad/traderHeal.test.ts @@ -0,0 +1,39 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { planTraderHeal, type HealSwapRow } from "./traderHeal.ts"; + +const row = (tx: string, block: number, log: number): HealSwapRow => ({ tx_hash: tx, log_index: log, token: "0x" + "11".repeat(20), block_number: BigInt(block) }); + +test("groups keep row order and only live groups spend a read", () => { + const rows = [row("a", 10, 0), row("a", 10, 1), row("b", 11, 0), row("c", 12, 0)]; + const plan = planTraderHeal(rows, new Map([["base:b", 2_000]]), "base", 1_000, 20); + assert.deepEqual(plan.groups.map((g) => g.tx), ["a", "b", "c"]); + assert.equal(plan.groups[0]!.resting, false); + assert.equal(plan.groups[1]!.resting, true); + assert.equal(plan.reads, 2, "the parked group does not spend a chain read"); + assert.deepEqual(plan.cursor, { block: 12n, log: 0 }); +}); + +test("a page made entirely of parked groups still advances the cursor", () => { + const rows = [row("a", 10, 0), row("b", 11, 0), row("c", 12, 0)]; + const rest = new Map([["base:a", 2_000], ["base:b", 2_000], ["base:c", 2_000]]); + const plan = planTraderHeal(rows, rest, "base", 1_000, 20); + assert.equal(plan.reads, 0); + assert.deepEqual(plan.groups.map((g) => g.tx), ["a", "b", "c"]); + assert.deepEqual(plan.cursor, { block: 12n, log: 0 }, "the next tick starts after these rows, not on them again"); +}); + +test("maxReads stops before the next live group, so the rest waits for the next tick", () => { + const rows = [row("a", 10, 0), row("b", 11, 0), row("c", 12, 0)]; + const plan = planTraderHeal(rows, new Map(), "base", 1_000, 2); + assert.deepEqual(plan.groups.map((g) => g.tx), ["a", "b"]); + assert.equal(plan.reads, 2); + assert.deepEqual(plan.cursor, { block: 11n, log: 0 }); +}); + +test("an empty page has no cursor to store", () => { + const plan = planTraderHeal([], new Map(), "base", 1_000, 20); + assert.deepEqual(plan.groups, []); + assert.equal(plan.reads, 0); + assert.equal(plan.cursor, null); +}); diff --git a/app/src/lib/launchpad/traderHeal.ts b/app/src/lib/launchpad/traderHeal.ts new file mode 100644 index 00000000..85ca75d7 --- /dev/null +++ b/app/src/lib/launchpad/traderHeal.ts @@ -0,0 +1,49 @@ +/** + * Pure planning for the missing-trader heal pass. No `server-only` import: node --test loads it directly. + * + * The old query took the newest 100 rows every time. Rows whose transaction could not be read were parked in a + * retry map and skipped, but they still occupied those 100 slots, so rows behind them were never reached. This + * planner always advances past every visited transaction group (parked or not); the caller stores the returned + * cursor and asks for the page after it next tick. The page therefore walks the whole set instead of staring at + * its head. + */ +export type HealSwapRow = { tx_hash: string; log_index: number; token: string; block_number: bigint }; +export type TraderHealCursor = { block: bigint; log: number }; + +export type TraderHealGroup = { + tx: string; + /** `${chain}:${tx}`: the key used by the rest map. */ + key: string; + swaps: HealSwapRow[]; + resting: boolean; + /** The last row of the group, which the next page starts after. */ + last: TraderHealCursor; +}; + +export function planTraderHeal( + rows: readonly HealSwapRow[], + restUntil: ReadonlyMap, + chain: string, + now: number, + maxReads: number, +): { groups: TraderHealGroup[]; reads: number; cursor: TraderHealCursor | null } { + const byTx = new Map(); + for (const r of rows) { + const group = byTx.get(r.tx_hash); + if (group) group.push(r); + else byTx.set(r.tx_hash, [r]); + } + const groups: TraderHealGroup[] = []; + let reads = 0; + let cursor: TraderHealCursor | null = null; + for (const [tx, swaps] of byTx) { + if (reads >= maxReads) break; + const last = swaps[swaps.length - 1]!; + cursor = { block: last.block_number, log: Number(last.log_index) }; + const key = `${chain}:${tx}`; + const resting = now < (restUntil.get(key) ?? 0); + groups.push({ tx, key, swaps, resting, last: cursor }); + if (!resting) reads++; + } + return { groups, reads, cursor }; +}