diff --git a/apps/api/src/indexer/sync.ts b/apps/api/src/indexer/sync.ts index 1aa0512..83c743c 100644 --- a/apps/api/src/indexer/sync.ts +++ b/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++; diff --git a/apps/api/src/integrations/counterparty.ts b/apps/api/src/integrations/counterparty.ts index 62f11f4..316e176 100644 --- a/apps/api/src/integrations/counterparty.ts +++ b/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 } | { known: false; deferred?: boolean } +> { + try { + const pools = new Map(); + const seen = new Set(); + 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 { const path = `/pools/${encodeURIComponent(asset)}/XCP?verbose=true`; try { diff --git a/tests/api-announce-without-node.test.ts b/tests/api-announce-without-node.test.ts index c03b6b5..be62dc6 100644 --- a/tests/api-announce-without-node.test.ts +++ b/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'}); diff --git a/tests/api-recovery-pools.test.ts b/tests/api-recovery-pools.test.ts new file mode 100644 index 0000000..7880d67 --- /dev/null +++ b/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); + }, +);