From 1b9ce22e4325f40d94b15b95ec08933d652cfd06 Mon Sep 17 00:00:00 2001 From: Ayush7614 Date: Tue, 20 Oct 2026 16:15:01 +0530 Subject: [PATCH 1/3] fix(indexer): keep sync cursors monotonic and page trader healing fairly Make cursor_block and head_block GREATEST updates so overlapping sync machines cannot move them backwards. Share one full poll pass across concurrent callers. Replace the newest-100 trader heal with a keyset cursor that advances past every visited tx group, add the matching partial index, and prune retry maps instead of clearing them. --- app/db/concurrent-indexes.sql | 3 + app/src/lib/launchpad/boundedMap.test.ts | 39 +++++++++ app/src/lib/launchpad/boundedMap.ts | 25 ++++++ .../lib/launchpad/indexer-durability.test.ts | 50 +++++++++++ app/src/lib/launchpad/indexer.ts | 87 ++++++++++++++----- app/src/lib/launchpad/traderHeal.test.ts | 39 +++++++++ app/src/lib/launchpad/traderHeal.ts | 49 +++++++++++ 7 files changed, 268 insertions(+), 24 deletions(-) create mode 100644 app/src/lib/launchpad/boundedMap.test.ts create mode 100644 app/src/lib/launchpad/boundedMap.ts create mode 100644 app/src/lib/launchpad/indexer-durability.test.ts create mode 100644 app/src/lib/launchpad/traderHeal.test.ts create mode 100644 app/src/lib/launchpad/traderHeal.ts 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..0eda32e6 --- /dev/null +++ b/app/src/lib/launchpad/indexer-durability.test.ts @@ -0,0 +1,50 @@ +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 single-flight 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("the whole poll pass is single-flight, not just each chain's run()", () => { + assert.match(indexer, /let allInFlight: Promise> \| null = null;/); + assert.match(indexer, /if \(allInFlight\) return allInFlight;/); + assert.match(indexer, /async function pollAllOnce\(\): Promise>/); + assert.match(indexer, /const p = pollAllOnce\(\)\.finally\(\(\) => \{/); +}); + +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 restTraderHeal\(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..aabea4ee 100644 --- a/app/src/lib/launchpad/indexer.ts +++ b/app/src/lib/launchpad/indexer.ts @@ -10,6 +10,8 @@ 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 { ENTRY_POINTS, attributeSwap, type ReceiptLog } from "./attribution"; import { appTagOf, swapInApp, type AppTag } from "./appTag"; import { BUILDER_DATA_SUFFIX } from "@/lib/chainPublic"; @@ -412,8 +414,27 @@ 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(); + +/** Bound a retry map without clearing it: expired entries go first, then the oldest insertions. */ +function restTraderHeal(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 +500,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); @@ -704,7 +724,19 @@ export async function pollLaunches(chain: ChainKey): Promise { } /** Poll every configured chain (used by the loop and POST /api/launch/sync without a tx). */ -export async function pollAll(): Promise> { +let allInFlight: Promise> | null = null; +export function pollAll(): Promise> { + // The per-chain guard only protects run(); the heals/backfills below are just as expensive. A web sync and the + // 15s loop overlapping must share one full pass, not run two. + if (allInFlight) return allInFlight; + const p = pollAllOnce().finally(() => { + if (allInFlight === p) allInFlight = null; + }); + allInFlight = p; + return p; +} + +async function pollAllOnce(): Promise> { const out: Record = {}; await Promise.all( CONFIGURED_CHAINS.map(async (c) => { @@ -812,40 +844,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))) { + restTraderHeal(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; } @@ -936,7 +975,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 +985,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/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 }; +} From 58f5c3c9de6bcd2f1c205270731974286a00e80d Mon Sep 17 00:00:00 2001 From: Kevin Codex Date: Fri, 13 Nov 2026 10:14:46 +0800 Subject: [PATCH 2/3] fix(indexer): heal passes run one at a time per chain, never blocking indexing On top of the rebase onto main: - pollAll is no longer single-flight as a whole. Sharing one full pass meant a heal that hung (there is no statement timeout) held every later tick's run() behind it, so a stuck heal stopped indexing on all chains. run() keeps its per-chain guard; each chain's heals and backfills now run one pass at a time (startUnlessRunning), and a caller that finds a pass still running skips it instead of waiting. - healAppTags, added on main after this branch, still cleared the shared rest map at 5,000 keys. It now uses the same bounded rest (restHeal, renamed from restTraderHeal since both heals use it). Checked against a throwaway Postgres 16 with the real indexer: a slower machine no longer pulls the cursor or head back or hides a newer error, a failed run keeps the highest head, rows behind 120 unreadable transactions heal on the first tick, a stuck heal no longer stops the next pass, and the paging query uses bb_launch_swaps_trader_null_idx. --- .../lib/launchpad/indexer-durability.test.ts | 17 ++++--- app/src/lib/launchpad/indexer.ts | 51 ++++++++++--------- app/src/lib/launchpad/runOnce.test.ts | 36 +++++++++++++ app/src/lib/launchpad/runOnce.ts | 15 ++++++ 4 files changed, 88 insertions(+), 31 deletions(-) create mode 100644 app/src/lib/launchpad/runOnce.test.ts create mode 100644 app/src/lib/launchpad/runOnce.ts diff --git a/app/src/lib/launchpad/indexer-durability.test.ts b/app/src/lib/launchpad/indexer-durability.test.ts index 0eda32e6..4a6f4476 100644 --- a/app/src/lib/launchpad/indexer-durability.test.ts +++ b/app/src/lib/launchpad/indexer-durability.test.ts @@ -4,7 +4,7 @@ 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 single-flight wiring that node --test cannot import + * 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"); @@ -18,11 +18,14 @@ test("cursor and head updates are monotonic across overlapping sync machines", ( assert.match(indexer, /head_block = CASE WHEN \$\{knownHead\}::bigint IS NULL THEN head_block ELSE GREATEST\(COALESCE\(head_block, 0\), \$\{knownHead\}::bigint\) END/); }); -test("the whole poll pass is single-flight, not just each chain's run()", () => { - assert.match(indexer, /let allInFlight: Promise> \| null = null;/); - assert.match(indexer, /if \(allInFlight\) return allInFlight;/); - assert.match(indexer, /async function pollAllOnce\(\): Promise>/); - assert.match(indexer, /const p = pollAllOnce\(\)\.finally\(\(\) => \{/); +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 whole pass is not single-flight: a stuck heal must not hold the next tick's run() behind it + assert.doesNotMatch(indexer, /allInFlight/); }); test("trader healing pages oldest-first with a keyset, not the newest 100 rows", () => { @@ -34,7 +37,7 @@ test("trader healing pages oldest-first with a keyset, not the newest 100 rows", }); test("retry state is pruned, never cleared wholesale", () => { - assert.match(indexer, /function restTraderHeal\(key: string\): void/); + 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,/); diff --git a/app/src/lib/launchpad/indexer.ts b/app/src/lib/launchpad/indexer.ts index aabea4ee..0d48547c 100644 --- a/app/src/lib/launchpad/indexer.ts +++ b/app/src/lib/launchpad/indexer.ts @@ -12,6 +12,7 @@ import { byChainOrder, fetchBySelectors, fetchLogsSplit, isRangeTooLarge } from 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"; @@ -424,8 +425,11 @@ const HEAL_TRADER_READS = 20; // the rows behind it are reached on the next tick instead of the same newest 100 rows being selected forever. const healTraderCursor = new Map(); -/** Bound a retry map without clearing it: expired entries go first, then the oldest insertions. */ -function restTraderHeal(key: string): void { +/** + * 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); } @@ -723,20 +727,11 @@ 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). */ -let allInFlight: Promise> | null = null; -export function pollAll(): Promise> { - // The per-chain guard only protects run(); the heals/backfills below are just as expensive. A web sync and the - // 15s loop overlapping must share one full pass, not run two. - if (allInFlight) return allInFlight; - const p = pollAllOnce().finally(() => { - if (allInFlight === p) allInFlight = null; - }); - allInFlight = p; - return p; -} +/** Chains whose heal and backfill pass is running in this process. */ +const healsRunning = new Map>(); -async function pollAllOnce(): Promise> { +/** Poll every configured chain (used by the loop and POST /api/launch/sync without a tx). */ +export async function pollAll(): Promise> { const out: Record = {}; await Promise.all( CONFIGURED_CHAINS.map(async (c) => { @@ -746,17 +741,26 @@ async function pollAllOnce(): 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). @@ -873,7 +877,7 @@ export async function healSwapTraders(chain: ChainKey): Promise { for (const group of plan.groups) { if (group.resting) continue; if (!(await txOf(chain, group.tx as Hex))) { - restTraderHeal(group.key); + restHeal(group.key); continue; } healRestUntil.delete(group.key); @@ -931,8 +935,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); 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; +} From b3e792d96d6e84ca94ffad44e29af700c8a98c60 Mon Sep 17 00:00:00 2001 From: Kevin Codex Date: Fri, 13 Nov 2026 10:25:33 +0800 Subject: [PATCH 3/3] docs(indexer): state why pollAll shares work per chain, not the whole pass CodeRabbit asked for the whole pollAll pass to be shared between overlapping callers. It is shared per chain on purpose: a run() in flight is joined (pollLaunches) and a running heal pass 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 would make each chain wait for the slowest one, and a stuck query on one chain would stop indexing on all. The comment on pollAll says so, and the durability test now also pins that a caller joins a run() still in flight. --- app/src/lib/launchpad/indexer-durability.test.ts | 4 +++- app/src/lib/launchpad/indexer.ts | 9 ++++++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/app/src/lib/launchpad/indexer-durability.test.ts b/app/src/lib/launchpad/indexer-durability.test.ts index 4a6f4476..c93f0a15 100644 --- a/app/src/lib/launchpad/indexer-durability.test.ts +++ b/app/src/lib/launchpad/indexer-durability.test.ts @@ -24,7 +24,9 @@ test("each chain's heal pass runs once at a time, and a running one is skipped, 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 whole pass is not single-flight: a stuck heal must not hold the next tick's run() behind it + // 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/); }); diff --git a/app/src/lib/launchpad/indexer.ts b/app/src/lib/launchpad/indexer.ts index 0d48547c..4516d306 100644 --- a/app/src/lib/launchpad/indexer.ts +++ b/app/src/lib/launchpad/indexer.ts @@ -730,7 +730,14 @@ export async function pollLaunches(chain: ChainKey): Promise { /** 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). */ +/** + * 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(