fix(indexer): monotonic sync cursor, single-flight polls, and fair trader healing

app/db/concurrent-indexes.sql

@@ -1,3 +1,3 @@
 -- Indexes on the big, busy tables, built by scripts/migrate.mjs AFTER the schema transaction commits, one at a time
 -- with CREATE INDEX CONCURRENTLY: reads and writes on the table never wait for the build. (Inside the schema
 -- transaction a build would hold the exclusive lock an ALTER TABLE on that table already took, and every query on
@@ -24,3 +24,6 @@
 -- 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;

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<string, number>();
+  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}`));
+});

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<K, V>(
+  map: Map<K, V>,
+  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);
+}

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<ChainKey, Promise<void>>\(\);/);
+  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<LaunchSyncResult> \{\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;/,
+  );
+});

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<ChainKey, number>();
 const healRestUntil = new Map<string, number>(); // `${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<ChainKey, TraderHealCursor>();
+
+/**
+ * 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<nu
       // EntryPoint call learned of here is marked so.
       if (who.via === RECEIPT_PENDING) await db`UPDATE bb_launch_swaps SET trader_via = ${RECEIPT_PENDING}, tx_from = COALESCE(tx_from, ${who.tx_from}) WHERE chain_id = ${cid} AND tx_hash = ${r.tx_hash} AND log_index = ${r.log_index} AND trader_via IS NULL`;
       const n = (prev?.n ?? 0) + 1;
-      if (attributionTries.size > 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<LaunchSyncResult> {
   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<ChainKey, Promise<void>>();
+
+/**
+ * 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<Record<string, LaunchSyncResult>> {
   const out: Record<string, LaunchSyncResult> = {};
   await Promise.all(
@@ -714,17 +748,26 @@ export async function pollAll(): Promise<Record<string, LaunchSyncResult>> {
         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<void> {
+  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<number> {
 /**
  * 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<number> {
   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<HealSwapRow[]> => db<HealSwapRow[]>`
+    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<string, { tx_hash: string; log_index: number; token: string }[]>();
-  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<number> {
     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<LaunchSyncResult> {
       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<LaunchSyncResult> {
   } 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;
   }
 }