Finish reorg recovery without per-launch pool request bursts

apps/api/src/indexer/sync.ts

@@ -9,6 +9,7 @@ import {
   fetchFairmints,
   fetchNodeStatus,
   fetchPool,
+  fetchRecoveryPools,
 } from "#api/integrations/counterparty";
 import { coreVersionGate } from "#api/indexer/core-version";
 import { checkLedger, rollbackIndexTo, verifyLedgerTip, type LedgerCheck } from "#api/indexer/ledger";
@@ -166,6 +167,7 @@ export async function syncLaunches(
   await db.prepare("INSERT INTO chain_state(key,value) VALUES('pending_index_height',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value").bind(String(height)).run();
   const all = await fetchAllFairminters();
   const reindexed = ledger.rolled_back_to !== null;
+  const recoveryPools = reindexed ? await fetchRecoveryPools() : null;
   // Remembered so what runs after this pass can judge "now" without a node
   // read of its own. The feed announces straight after this, which is exactly
   // when the node is likeliest to be refusing us; see currentHeight.
@@ -239,7 +241,11 @@ export async function syncLaunches(
       truthy(fm.pool_quantity) &&
       (reindexed || priorLaunch?.phase !== "refunded")
     ) {
-      const lookup = await fetchPool(fm.asset);
+      const lookup = recoveryPools
+        ? recoveryPools.known
+          ? { known: true as const, pool: recoveryPools.pools.get(fm.asset) ?? null }
+          : recoveryPools
+        : await fetchPool(fm.asset);
       poolUnknown = !lookup.known;
       if (!lookup.known) {
         if (lookup.deferred) poolLookupsDeferred++;

apps/api/src/integrations/counterparty.ts

@@ -690,6 +690,38 @@ export interface CpPool {
  */
 export type PoolLookup = { known: true; pool: CpPool | null } | { known: false; deferred?: boolean };
 
+/** Recovery must reconsider absent pools too. Read the complete set once,
+ * rather than issuing one request for every historical refunded launch.
+ * A failed/incomplete listing cannot establish absence for any asset. */
+export async function fetchRecoveryPools(): Promise<
+  { known: true; pools: Map<string, CpPool> } | { known: false; deferred?: boolean }
+> {
+  try {
+    const pools = new Map<string, CpPool>();
+    const seen = new Set<string>();
+    let cursor: string | number | null = null;
+    for (;;) {
+      const suffix = cursor === null ? "" : `&cursor=${encodeURIComponent(cursor)}`;
+      const data: { result: CpPool[]; next_cursor: string | number | null } =
+        await get(`/pools?limit=1000&verbose=true${suffix}`);
+      if (!Array.isArray(data.result) || !("next_cursor" in data)) throw new Error("Incomplete pool listing");
+      for (const pool of data.result) {
+        if (typeof pool.asset_a !== "string" || typeof pool.asset_b !== "string"
+          || pool.reserve_a == null || pool.reserve_b == null) throw new Error("Invalid pool listing");
+        if (pool.asset_a === "XCP") pools.set(pool.asset_b, pool);
+        else if (pool.asset_b === "XCP") pools.set(pool.asset_a, pool);
+      }
+      if (data.next_cursor === null) return { known: true, pools };
+      if (!["string", "number"].includes(typeof data.next_cursor)
+        || seen.has(String(data.next_cursor))) throw new Error("Invalid pool cursor");
+      cursor = data.next_cursor;
+      seen.add(String(cursor));
+    }
+  } catch (error) {
+    return { known: false, deferred: error instanceof CounterpartyReadDeferred };
+  }
+}
+
 export async function fetchPool(asset: string): Promise<PoolLookup> {
   const path = `/pools/${encodeURIComponent(asset)}/XCP?verbose=true`;
   try {

tests/api-announce-without-node.test.ts

@@ -98,11 +98,15 @@ describe("the index pass stops re-asking about pools it has already found missin
       listing({ tx_hash: refundedHash, tx_index: fixture.launch.tx_index + 1, asset: "NOPOOL", lp_asset: null,
         earned_quantity: "100000000000000", paid_quantity: "1000000" }),
     ];
+    let refuseSnapshot = false;
     const fetch = vi.fn(async (input: string) => {
       const url = new URL(input);
       if (url.pathname === "/v2/blocks") return Response.json({result:[{block_index:HEIGHT,block_hash:"aa",ledger_hash:"bb",messages_hash:"cc"}]});
       if (url.pathname === "/v2/") return Response.json({ result: { counterparty_height: HEIGHT } });
       if (url.pathname === "/v2/fairminters") return Response.json({ result: fairminters, next_cursor: null });
+      if (url.pathname === "/v2/pools" && refuseSnapshot) return new Response(null, { status: 503 });
+      if (url.pathname === "/v2/pools") return Response.json({ result: [{ asset_a: "XCP", asset_b: "FEWGOODMAN",
+        reserve_a: fixture.launch.pool_xcp_reserve, reserve_b: fixture.launch.pool_token_reserve }], next_cursor: null });
       if (url.pathname === "/v2/pools/FEWGOODMAN/XCP") {
         return Response.json({ result: { asset_a: "XCP", asset_b: "FEWGOODMAN",
           reserve_a: fixture.launch.pool_xcp_reserve, reserve_b: fixture.launch.pool_token_reserve } });
@@ -127,9 +131,25 @@ describe("the index pass stops re-asking about pools it has already found missin
     // Recovery must revisit a formerly absent pool without first replacing
     // either public phase with pending or invalidating unaffected mint totals.
     database.raw.prepare("INSERT INTO chain_state(key,value) VALUES('ledger_recovery',?)").run(String(HEIGHT-1));
+    for (let i = 1; i < 208; i++) {
+      const over = { tx_hash: i.toString(16).padStart(64, "0"), tx_index: fixture.launch.tx_index + i + 1,
+        asset: `NOPOOL${i}`, phase: "refunded", earned_quantity: "100000000000000", paid_quantity: "1000000",
+        mints: 1, minters: 1, pool_xcp_reserve: null, pool_token_reserve: null, pool_xcp_sats: 0, lp_asset: null };
+      insertLaunch(over);
+      fairminters.push(listing(over));
+    }
+    refuseSnapshot = true;
+    expect(await syncLaunches(database.db, metadata)).toMatchObject({ partial: true, pool_lookups_failed: 209 });
+    expect(database.raw.prepare("SELECT COUNT(*) AS n FROM launches WHERE phase='refunded'").get()).toEqual({n:208});
+    expect(database.raw.prepare("SELECT phase,pool_xcp_reserve FROM launches WHERE tx_hash=?").get(fixture.launch.tx_hash))
+      .toEqual({phase:'graduated',pool_xcp_reserve:fixture.launch.pool_xcp_reserve});
+    expect(database.raw.prepare("SELECT value FROM chain_state WHERE key='ledger_recovery'").get()).toBeDefined();
+    refuseSnapshot = false;
     fetch.mockClear();
-    await syncLaunches(database.db, metadata);
-    expect(fetch.mock.calls.map(([input])=>new URL(input).pathname)).toContain('/v2/pools/NOPOOL/XCP');
+    expect(await syncLaunches(database.db, metadata)).toMatchObject({ partial: false, pool_lookups_failed: 0 });
+    expect(fetch.mock.calls.map(([input])=>new URL(input).pathname).filter(path=>path.startsWith('/v2/pools'))).toEqual(['/v2/pools']);
+    expect(database.raw.prepare("SELECT COUNT(*) AS n FROM launches WHERE phase='refunded'").get()).toEqual({n:208});
+    expect(database.raw.prepare("SELECT key FROM chain_state WHERE key IN ('ledger_recovery','pending_index_height')").all()).toEqual([]);
     expect(fetch.mock.calls.map(([input])=>new URL(input).pathname).filter(path=>path.endsWith('/fairmints'))).toEqual([]);
     expect(database.raw.prepare("SELECT phase FROM launches WHERE tx_hash=?").get(refundedHash)).toEqual({phase:'refunded'});
     expect(database.raw.prepare("SELECT phase FROM launches WHERE tx_hash=?").get(fixture.launch.tx_hash)).toEqual({phase:'graduated'});

tests/api-recovery-pools.test.ts

@@ -0,0 +1,32 @@
+import { afterEach, beforeEach, expect, it, vi } from "vitest";
+
+beforeEach(() => vi.resetModules());
+afterEach(() => vi.unstubAllGlobals());
+
+const pool = { asset_a: "TOKEN", asset_b: "XCP", reserve_a: "99999999999999999", reserve_b: 123 };
+
+it("exhausts pool pages before establishing absence and preserves raw reserves", async () => {
+  const fetch = vi.fn()
+    .mockResolvedValueOnce(Response.json({ result: [{ ...pool, asset_a: "OTHER", asset_b: "BTC" }], next_cursor: 0 }))
+    .mockResolvedValueOnce(Response.json({ result: [pool], next_cursor: null }));
+  vi.stubGlobal("fetch", fetch);
+  const { fetchRecoveryPools } = await import("#api/integrations/counterparty");
+  const result = await fetchRecoveryPools();
+  expect(result).toEqual({ known: true, pools: new Map([["TOKEN", pool]]) });
+  expect(new URL(fetch.mock.calls[1]![0]).searchParams.get("cursor")).toBe("0");
+});
+
+it.each(["failed-page", "repeated-cursor", "missing-cursor", "malformed-page"])(
+  "never treats an incomplete snapshot (%s) as pool absence", async failure => {
+    const second = failure === "failed-page" ? new Response(null, { status: 503 })
+      : failure === "repeated-cursor" ? Response.json({ result: [], next_cursor: 4 })
+        : failure === "missing-cursor" ? Response.json({ result: [] })
+          : Response.json({ result: [{}], next_cursor: null });
+    const fetch = vi.fn().mockResolvedValueOnce(Response.json({ result: [pool], next_cursor: 4 }))
+      .mockResolvedValueOnce(second);
+    vi.stubGlobal("fetch", fetch);
+    const { fetchRecoveryPools } = await import("#api/integrations/counterparty");
+    expect(await fetchRecoveryPools()).toEqual({ known: false, deferred: false });
+    expect(fetch).toHaveBeenCalledTimes(2);
+  },
+);