diff --git a/apps/api/src/indexer/ledger.ts b/apps/api/src/indexer/ledger.ts index 2419e1b3..261a011e 100644 --- a/apps/api/src/indexer/ledger.ts +++ b/apps/api/src/indexer/ledger.ts @@ -17,6 +17,8 @@ */ import { q } from "#api/db"; import { fetchBlockHashes, type CpBlockHashes } from "#api/integrations/counterparty"; +import { COMMUNITY_ROLLUP_STALE_KEY } from "#api/queries/communities"; +import { BURN_RESCAN_KEY } from "#api/telegram/burns"; /** Blocks compared each tick. Five-minute ticks see one or two new blocks. */ const WINDOW = 12; @@ -126,6 +128,8 @@ export async function checkLedger(db: D1Database): Promise { export interface RollbackResult { mints_removed: number; launches_removed: number; + burns_removed: number; + buyers_removed: number; event_assets_reset: number; } @@ -137,6 +141,9 @@ const POOL_PREFIX = "events_pool:"; * re-reads it from the node's current ledger. One batch, so a failure leaves * the index exactly as it was. * + * What comes from the chain is rolled back; what records something that + * happened off it is not. + * * - Mints after the block are deleted, and their launches' `earned_quantity` * cleared: that value is what tells the next pass a launch's mint history * needs reading, and it must not match by coincidence. The crown is @@ -145,11 +152,18 @@ const POOL_PREFIX = "events_pool:"; * back the ones the node still has, and any it no longer has stay gone. * - A traded asset whose event cursor passed the block loses its events after * it, its candles, and both cursors, so its next pass is a first run: the - * authoritative refold from block 0. + * authoritative refold from block 0. A buyer whose every buy was after the + * block leaves `behavior_buyers`; the refold adds back the ones still true. + * - Burns after the block are deleted and their launches' burned supply + * summed again from the burns that remain. The burn scan's cursors are + * Core's tx_index and event_index, which a re-parse hands out again from + * the fork, so they cannot be rewound to a block: the scan is told to + * re-read by block from the one after this, and it resets both cursors to + * what it finds. * - * Rollups are not touched here; the pass that follows rebuilds them. - * Telegram announcements already sent and recorded reward batches are history - * and stay. + * The rollups summarising mints and trades are not touched here; the pass + * that follows rebuilds them from what remains. The community rollup, which + * otherwise waits hours, is marked stale so the next tick rebuilds it. */ export async function rollbackIndexTo(db: D1Database, block: number): Promise { const cursors = await q<{ key: string }>( @@ -160,6 +174,11 @@ export async function rollbackIndexTo(db: D1Database, block: number): Promise row.key.slice(CURSOR_PREFIX.length)); + // Deliberately absent: `announced`, `announcement_work`, and the reward + // tables. A Telegram post that went out and a reward batch that was frozen + // or sent are records of things that happened off the chain; a re-parse + // cannot unsay the post or unsend the payment, so their records stay. Reward + // batches only ever take mints six blocks deep (scripts/reward-batch.mjs). const statements: D1PreparedStatement[] = [ db.prepare( `UPDATE launches SET earned_quantity = NULL @@ -180,9 +199,46 @@ export async function rollbackIndexTo(db: D1Database, block: number): Promise ?1`, ).bind(block), db.prepare(`DELETE FROM launches WHERE announce_block > ?1`).bind(block), + // Before the events go: which buyers to reconsider is read from them. + db.prepare( + `DELETE FROM behavior_buyers + WHERE address IN (SELECT address FROM asset_events WHERE block_index > ?1 AND kind = 'buy') + AND NOT EXISTS (SELECT 1 FROM asset_events e + WHERE e.address = behavior_buyers.address + AND e.kind = 'buy' AND e.block_index <= ?1)`, + ).bind(block), db.prepare(`DELETE FROM asset_events WHERE block_index > ?1`).bind(block), + // Same sum as reconcileBurnedSupply, without the burns about to go, and + // only for the launches they touched. + db.prepare( + `UPDATE launches + SET burned_quantity = CAST(COALESCE(( + SELECT SUM(CAST(b.quantity AS INTEGER)) + FROM token_burns b + WHERE b.asset = launches.asset + AND b.tx_index > launches.tx_index + AND b.block_index <= ?1 + ), 0) AS TEXT) + WHERE asset IN (SELECT asset FROM token_burns WHERE block_index > ?1)`, + ).bind(block), + db.prepare(`DELETE FROM token_burns WHERE block_index > ?1`).bind(block), + // The count is kept by an insert trigger, so a delete has to restate it. + db.prepare( + `UPDATE burn_totals SET burns = (SELECT COUNT(*) FROM token_burns) + WHERE id = 1 AND burns IS NOT (SELECT COUNT(*) FROM token_burns)`, + ), + db.prepare( + `INSERT INTO chain_state (key, value) VALUES (?1, ?2) + ON CONFLICT(key) DO UPDATE SET value = excluded.value + WHERE CAST(excluded.value AS INTEGER) < CAST(chain_state.value AS INTEGER)`, + ).bind(BURN_RESCAN_KEY, String(block + 1)), + db.prepare(`INSERT OR IGNORE INTO chain_state (key, value) VALUES (?1, '1')`).bind(COMMUNITY_ROLLUP_STALE_KEY), db.prepare(`DELETE FROM indexed_blocks WHERE block_index > ?1`).bind(block), ]; + const MINTS = 1; + const LAUNCHES = 3; + const BUYERS = 4; + const BURNS = 7; for (let i = 0; i < assets.length; i += SQL_VAR_LIMIT) { const chunk = assets.slice(i, i + SQL_VAR_LIMIT); const places = chunk.map((_, idx) => `?${idx + 1}`).join(","); @@ -196,8 +252,10 @@ export async function rollbackIndexTo(db: D1Database, block: number): Promise sum + (r.meta.rows_written ?? 0), 0); } @@ -331,8 +332,17 @@ export async function readCommunityRollup(db: D1Database): Promise { - const row = await one<{ paid_xcp: string }>(db, `SELECT paid_xcp FROM community_totals WHERE id = 1`); - return row !== null && row.paid_xcp !== "0"; + const row = await one<{ paid_xcp: string | null; stale: string | null }>( + db, + `SELECT (SELECT paid_xcp FROM community_totals WHERE id = 1) AS paid_xcp, + (SELECT value FROM chain_state WHERE key = ?1) AS stale`, + COMMUNITY_ROLLUP_STALE_KEY, + ); + return row !== null && row.paid_xcp !== null && row.paid_xcp !== "0" && row.stale === null; } diff --git a/apps/api/src/telegram/burns.ts b/apps/api/src/telegram/burns.ts index d8f2ce76..d94748c7 100644 --- a/apps/api/src/telegram/burns.ts +++ b/apps/api/src/telegram/burns.ts @@ -18,6 +18,11 @@ const DESTRUCTION_CURSOR_KEY = "telegram_asset_destruction_event_index"; // Versioned because v1 incorrectly seeded from this address's live balance, // while v2 included SENDs but missed explicit ASSET_DESTRUCTION events. const SUPPLY_SEED_KEY = "burned_supply_from_chain_events_seeded"; +/** Set by a ledger rollback (indexer/ledger.ts) to the first block it removed. + * Both cursors are Core indexes that a re-parse hands out again from the + * fork, so the scan cannot trust them past it: it re-reads by block instead, + * and then resets both cursors to what it found. */ +export const BURN_RESCAN_KEY = "telegram_burn_rescan_from_block"; const PAGE_SIZE = 100; const MAX_PAGES = 20; @@ -32,6 +37,36 @@ export interface BurnScan { nextCursor: number | null; nextDestructionCursor: number | null; seeded: boolean; + /** The block a rollback asked this scan to re-read from, or null on a + * normal scan. Passed back to advanceBurnCursor, which clears it. */ + rescanFrom: number | null; +} + +/** + * Newest-first pages until `reached` says a row is already covered. Rows past + * it are handed to `take`; the page bound keeps a runaway feed from spinning. + */ +async function readBack( + fetchPage: (cursor: number | undefined) => Promise<{ result: T[]; next_cursor: number | null }>, + reached: (row: T) => boolean, + take: (row: T) => void, + what: string, +): Promise { + let cursor: number | undefined; + for (let pageNumber = 0; pageNumber < MAX_PAGES; pageNumber++) { + const page = await fetchPage(cursor); + let done = false; + for (const row of page.result) { + if (reached(row)) { + done = true; + continue; + } + take(row); + } + if (done || page.next_cursor === null) return; + cursor = page.next_cursor; + } + throw new Error(`${what} pagination exceeded safety limit; cursor not advanced`); } const txIndex = (row: CpAddressReceive) => Number(row.tx_index); @@ -242,10 +277,11 @@ async function seedBurnEvents(db: D1Database): Promise { export async function scanBurnReceives(db: D1Database): Promise { const state = await q<{ key: string; value: string }>( db, - `SELECT key, value FROM chain_state WHERE key IN (?1, ?2, ?3)`, + `SELECT key, value FROM chain_state WHERE key IN (?1, ?2, ?3, ?4)`, CURSOR_KEY, DESTRUCTION_CURSOR_KEY, SUPPLY_SEED_KEY, + BURN_RESCAN_KEY, ); const states = new Map(state.map((row) => [row.key, row.value])); const receiveValue = states.get(CURSOR_KEY); @@ -254,6 +290,8 @@ export async function scanBurnReceives(db: D1Database): Promise { const destructionStored = destructionValue === undefined ? null : Number(destructionValue); + const rescanValue = states.get(BURN_RESCAN_KEY); + const rescanFrom = rescanValue === undefined ? null : Number(rescanValue); let reconciled = false; if (states.get(SUPPLY_SEED_KEY) !== "1") { await seedBurnEvents(db); @@ -265,30 +303,22 @@ export async function scanBurnReceives(db: D1Database): Promise { const latestReceive = latestReceives.result[0]; let nextCursor = receiveStored; const freshSends = new Map(); + const receivePage = (cursor: number | undefined) => + fetchAddressReceives(COUNTERPARTY_BURN_ADDRESS, PAGE_SIZE, cursor); + const takeSend = (row: CpAddressReceive) => { + freshSends.set(`${row.tx_hash}:${row.msg_index}:${row.asset}`, row); + }; if (latestReceive) { const newest = txIndex(latestReceive); - nextCursor = newest > (receiveStored ?? -1) ? newest : receiveStored; - if (receiveStored !== null && newest > receiveStored) { - let cursor: number | undefined; - let reachedStored = false; - for (let pageNumber = 0; pageNumber < MAX_PAGES; pageNumber++) { - const page = await fetchAddressReceives( - COUNTERPARTY_BURN_ADDRESS, - PAGE_SIZE, - cursor, - ); - for (const row of page.result) { - if (txIndex(row) <= receiveStored) { - reachedStored = true; - continue; - } - freshSends.set(`${row.tx_hash}:${row.msg_index}:${row.asset}`, row); - } - if (reachedStored || page.next_cursor === null) break; - cursor = page.next_cursor; - if (pageNumber === MAX_PAGES - 1) { - throw new Error("Burn receive pagination exceeded safety limit; cursor not advanced"); - } + if (rescanFrom !== null) { + // After a rollback the node's newest row is the cursor, even below the + // old one: the old one may number a transaction that no longer exists. + nextCursor = newest; + await readBack(receivePage, (row) => row.block_index < rescanFrom, takeSend, "Burn receive"); + } else { + nextCursor = newest > (receiveStored ?? -1) ? newest : receiveStored; + if (receiveStored !== null && newest > receiveStored) { + await readBack(receivePage, (row) => txIndex(row) <= receiveStored, takeSend, "Burn receive"); } } } @@ -297,28 +327,31 @@ export async function scanBurnReceives(db: D1Database): Promise { const latestDestruction = latestDestructions.result[0]; let nextDestructionCursor = destructionStored; const freshDestructions = new Map(); + const destructionPage = (cursor: number | undefined) => fetchAssetDestructions(PAGE_SIZE, cursor); + const takeDestruction = (row: CpAssetDestructionEvent) => { + freshDestructions.set(row.event_index, row); + }; if (latestDestruction) { const newest = latestDestruction.event_index; - nextDestructionCursor = newest > (destructionStored ?? -1) - ? newest - : destructionStored; - if (destructionStored !== null && newest > destructionStored) { - let cursor: number | undefined; - let reachedStored = false; - for (let pageNumber = 0; pageNumber < MAX_PAGES; pageNumber++) { - const page = await fetchAssetDestructions(PAGE_SIZE, cursor); - for (const row of page.result) { - if (row.event_index <= destructionStored) { - reachedStored = true; - continue; - } - freshDestructions.set(row.event_index, row); - } - if (reachedStored || page.next_cursor === null) break; - cursor = page.next_cursor; - if (pageNumber === MAX_PAGES - 1) { - throw new Error("Asset destruction pagination exceeded safety limit; cursor not advanced"); - } + if (rescanFrom !== null) { + nextDestructionCursor = newest; + await readBack( + destructionPage, + (row) => row.block_index < rescanFrom, + takeDestruction, + "Asset destruction", + ); + } else { + nextDestructionCursor = newest > (destructionStored ?? -1) + ? newest + : destructionStored; + if (destructionStored !== null && newest > destructionStored) { + await readBack( + destructionPage, + (row) => row.event_index <= destructionStored, + takeDestruction, + "Asset destruction", + ); } } } @@ -390,29 +423,43 @@ export async function scanBurnReceives(db: D1Database): Promise { nextCursor, nextDestructionCursor, seeded: receiveStored === null && destructionStored === null, + rescanFrom, }; } -/** Monotonic even if a retry races another caller after durable acceptance. */ +/** Monotonic even if a retry races another caller after durable acceptance, + * except after a rollback's rescan (`rescanFrom`): its cursors are the node's + * newest and may sit below the old ones. Those writes and the marker's + * removal are one batch, and the marker goes only if it still names the block + * this scan re-read from, so a second rollback in between is not forgotten. */ export async function advanceBurnCursor( db: D1Database, nextCursor: number | null, nextDestructionCursor?: number | null, + rescanFrom: number | null = null, ): Promise { + const guard = rescanFrom === null + ? ` + WHERE CAST(excluded.value AS INTEGER) > CAST(chain_state.value AS INTEGER)` + : ""; const statements: D1PreparedStatement[] = []; if (nextCursor !== null) { statements.push(db.prepare( `INSERT INTO chain_state (key, value) VALUES (?1, ?2) - ON CONFLICT(key) DO UPDATE SET value = excluded.value - WHERE CAST(excluded.value AS INTEGER) > CAST(chain_state.value AS INTEGER)`, + ON CONFLICT(key) DO UPDATE SET value = excluded.value${guard}`, ).bind(CURSOR_KEY, String(nextCursor))); } if (nextDestructionCursor !== undefined && nextDestructionCursor !== null) { statements.push(db.prepare( `INSERT INTO chain_state (key, value) VALUES (?1, ?2) - ON CONFLICT(key) DO UPDATE SET value = excluded.value - WHERE CAST(excluded.value AS INTEGER) > CAST(chain_state.value AS INTEGER)`, + ON CONFLICT(key) DO UPDATE SET value = excluded.value${guard}`, ).bind(DESTRUCTION_CURSOR_KEY, String(nextDestructionCursor))); } + if (rescanFrom !== null) { + statements.push( + db.prepare(`DELETE FROM chain_state WHERE key = ?1 AND value = ?2`) + .bind(BURN_RESCAN_KEY, String(rescanFrom)), + ); + } if (statements.length > 0) await db.batch(statements); } diff --git a/apps/api/src/telegram/live.ts b/apps/api/src/telegram/live.ts index 287f6ab7..bf81ca9d 100644 --- a/apps/api/src/telegram/live.ts +++ b/apps/api/src/telegram/live.ts @@ -174,7 +174,7 @@ export async function announceLive(env: Env, height: number): Promise (Number(BigInt(raw) / 100_000_000n)).toLocaleString("en-US"); console.log(`reward batch through mint #${cutoff}`); console.log(` cutoff tx ${last.tx_hash} (block ${last.block_index})`); -console.log(` chain tip ${tip}`); +console.log(` chain tip ${tip} (mints through block ${settled})`); console.log(` recipients ${payouts.length}`); console.log(` total ${whole(total)} ${ASSET}`); console.log(` mpma.csv ${mpma.length} addresses, ${whole( diff --git a/tests/api-indexer-safety.test.ts b/tests/api-indexer-safety.test.ts index 6fa862c7..b1a67a26 100644 --- a/tests/api-indexer-safety.test.ts +++ b/tests/api-indexer-safety.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import fixture from "./fixtures/fewgoodman-index-recovery.json"; import { apiReplayDb } from "./helpers/api-replay-db"; import { CORE_ACTIVATIONS, coreVersionGate, parseCoreVersion } from "#api/indexer/core-version"; +import { eligibleMintsSql, settledThrough } from "../scripts/lib/reward-eligibility.mjs"; vi.mock("#api/integrations/mempool", () => ({ fetchTxFee: async () => ({ feeSats: 100, weightWu: 400 }) })); vi.mock("#api/integrations/price", () => ({ fetchXcpUsd: async () => null, fetchXcpUsdHistory: async () => [], historicalXcpUsdAt: () => null })); @@ -11,6 +12,14 @@ let database: ReturnType; const metadata = { get: async () => null } as unknown as R2Bucket; const count = (sql: string, ...binds: Array) => (database.raw.prepare(sql).get(...binds) as { n: number }).n; +const stateValue = (key: string) => + (database.raw.prepare("SELECT value FROM chain_state WHERE key = ?").get(key) as { value: string } | undefined)?.value ?? null; +function insertFixtureLaunch() { + const writable = new Set(database.raw.prepare("PRAGMA table_xinfo(launches)").all().filter(c => c.hidden === 0).map(c => c.name)); + const entries = Object.entries(fixture.launch).filter(([key]) => writable.has(key)); + database.raw.prepare(`INSERT INTO launches (${entries.map(([key]) => key).join(",")}) VALUES (${entries.map(() => "?").join(",")})`) + .run(...entries.map(([, value]) => (typeof value === "boolean" ? Number(value) : value) as string | number | null)); +} beforeEach(() => { vi.resetModules(); @@ -236,7 +245,150 @@ describe("rollbackIndexTo", () => { expect(await rollbackIndexTo(database.db, 965_950)).toMatchObject({ event_assets_reset: 1 }); expect(raw.prepare("SELECT id FROM asset_events ORDER BY id").all().map(r => r.id)).toEqual(["old", "quiet"]); expect(raw.prepare("SELECT asset FROM price_candles").all().map(r => r.asset)).toEqual(["QUIETONE"]); - expect(raw.prepare("SELECT key FROM chain_state ORDER BY key").all().map(r => r.key)) + expect(raw.prepare("SELECT key FROM chain_state WHERE key LIKE 'events_%' ORDER BY key").all().map(r => r.key)) .toEqual(["events_hw:QUIETONE", "events_pool:QUIETONE"]); }); + + it("drops buyers whose every buy was after the block, and keeps the rest", async () => { + const raw = database.raw; + const event = raw.prepare(`INSERT INTO asset_events (id, event, address, asset, block_index, token_delta, xcp_delta, kind) + VALUES (?, 'x', ?, 'FEWGOODMAN', ?, '1', '-1', ?)`); + event.run("early", "EARLY", 965_900, "buy"); + event.run("early-again", "EARLY", 965_990, "buy"); + event.run("late", "LATE", 965_990, "buy"); + event.run("seller", "SELLER", 965_990, "sell"); + raw.exec("INSERT INTO behavior_buyers (address) VALUES ('EARLY'), ('LATE'), ('UNTOUCHED')"); + + const { rollbackIndexTo } = await import("#api/indexer/ledger"); + expect(await rollbackIndexTo(database.db, 965_950)).toMatchObject({ buyers_removed: 1 }); + expect(raw.prepare("SELECT address FROM behavior_buyers ORDER BY address").all().map(r => r.address)) + .toEqual(["EARLY", "UNTOUCHED"]); + }); + + it("removes burns after the block, restates the burned supply and count, and asks the burn scan to re-read", async () => { + const raw = database.raw; + insertFixtureLaunch(); + const launchTx = fixture.launch.tx_index; + const burn = raw.prepare(`INSERT INTO token_burns (key, tx_hash, tx_index, msg_index, block_index, source, destination, asset, quantity) + VALUES (?, ?, ?, 0, ?, 'holder', 'burn', 'FEWGOODMAN', ?)`); + burn.run("burn:kept", "a".repeat(64), launchTx + 10, 965_940, "100"); + burn.run("burn:gone", "b".repeat(64), launchTx + 20, 965_960, "50"); + raw.prepare("UPDATE launches SET burned_quantity = '150' WHERE asset = 'FEWGOODMAN'").run(); + expect(count("SELECT burns AS n FROM burn_totals WHERE id = 1")).toBe(2); + // History of what happened off the chain. + raw.exec("INSERT INTO announced (key, at) VALUES ('burn:gone', 1)"); + raw.exec("INSERT INTO community_totals (id, minters, represented, creators, collectors, paid_xcp) VALUES (1, 1, 1, 1, 1, '5')"); + + const { rollbackIndexTo } = await import("#api/indexer/ledger"); + const { hasCommunityRollup } = await import("#api/queries/communities"); + expect(await hasCommunityRollup(database.db)).toBe(true); + expect(await rollbackIndexTo(database.db, 965_950)).toMatchObject({ burns_removed: 1 }); + expect(raw.prepare("SELECT key FROM token_burns").all().map(r => r.key)).toEqual(["burn:kept"]); + expect(raw.prepare("SELECT burned_quantity FROM launches WHERE asset = 'FEWGOODMAN'").get()) + .toEqual({ burned_quantity: "100" }); + expect(count("SELECT burns AS n FROM burn_totals WHERE id = 1")).toBe(1); + expect(stateValue("telegram_burn_rescan_from_block")).toBe("965951"); + expect(count("SELECT COUNT(*) AS n FROM announced")).toBe(1); + expect(await hasCommunityRollup(database.db)).toBe(false); + + // A deeper rollback before the scan ran widens the re-read; a shallower one does not narrow it. + await rollbackIndexTo(database.db, 965_900); + await rollbackIndexTo(database.db, 965_945); + expect(stateValue("telegram_burn_rescan_from_block")).toBe("965901"); + }); +}); + +describe("burn scan after a rollback", () => { + const BURN = "1CounterpartyXXXXXXXXXXXXXXXUWLpVr"; + const launchTx = fixture.launch.tx_index; + const receive = (txIndex: number, block: number, quantity: string) => ({ + tx_index: txIndex, tx_hash: String(txIndex).padStart(64, "0"), block_index: block, source: "holder", + destination: BURN, asset: "FEWGOODMAN", quantity, status: "valid", msg_index: 0, send_type: "send", + }); + const destruction = (eventIndex: number, block: number, quantity: string) => ({ + event_index: eventIndex, event: "ASSET_DESTRUCTION", tx_hash: String(eventIndex).padStart(64, "e"), block_index: block, + params: { tx_hash: String(eventIndex).padStart(64, "e"), tx_index: launchTx + 50, block_index: block, source: "holder", + asset: "FEWGOODMAN", quantity, status: "valid", tag: "" }, + }); + + it("re-reads by block from the rollback, resets both cursors below the old ones, and then clears the marker", async () => { + const raw = database.raw; + insertFixtureLaunch(); + raw.prepare(`INSERT INTO token_burns (key, tx_hash, tx_index, msg_index, block_index, source, destination, asset, quantity) + VALUES ('burn:kept', ?, ?, 0, 965940, 'holder', ?, 'FEWGOODMAN', '100')`).run("a".repeat(64), launchTx + 10, BURN); + const state = raw.prepare("INSERT INTO chain_state (key, value) VALUES (?, ?)"); + state.run("burned_supply_from_chain_events_seeded", "1"); + // Cursors from the chain before the re-parse: higher than anything the node now has. + state.run("telegram_burn_receive_tx_index", String(launchTx + 900)); + state.run("telegram_asset_destruction_event_index", "90000"); + state.run("telegram_burn_rescan_from_block", "965951"); + + const receives = [receive(launchTx + 40, 965_960, "70"), receive(launchTx + 10, 965_940, "100")]; + const destructions = [destruction(80_000, 965_955, "5"), destruction(70_000, 965_900, "9")]; + const fetch = vi.fn(async (input: string) => { + const url = new URL(input); + const limit = Number(url.searchParams.get("limit")); + if (url.pathname === `/v2/addresses/${BURN}/receives`) return Response.json({ result: receives.slice(0, limit), next_cursor: null }); + if (url.pathname === "/v2/events/ASSET_DESTRUCTION") return Response.json({ result: destructions.slice(0, limit), next_cursor: null }); + throw new Error(`unexpected read ${url}`); + }); + vi.stubGlobal("fetch", fetch); + + const { advanceBurnCursor, scanBurnReceives } = await import("#api/telegram/burns"); + const scan = await scanBurnReceives(database.db); + expect(scan).toMatchObject({ nextCursor: launchTx + 40, nextDestructionCursor: 80_000, rescanFrom: 965_951 }); + expect(scan.announcements.map(a => a.key)).toEqual([ + `burn:${receive(launchTx + 40, 0, "").tx_hash}:0:FEWGOODMAN`, + "destroy:80000:FEWGOODMAN", + ]); + expect(raw.prepare("SELECT key FROM token_burns ORDER BY block_index").all().map(r => r.key)).toEqual([ + "burn:kept", "destroy:80000:FEWGOODMAN", `burn:${receive(launchTx + 40, 0, "").tx_hash}:0:FEWGOODMAN`, + ]); + expect(raw.prepare("SELECT burned_quantity FROM launches WHERE asset = 'FEWGOODMAN'").get()) + .toEqual({ burned_quantity: "175" }); + + await advanceBurnCursor(database.db, scan.nextCursor, scan.nextDestructionCursor, scan.rescanFrom); + expect(stateValue("telegram_burn_receive_tx_index")).toBe(String(launchTx + 40)); + expect(stateValue("telegram_asset_destruction_event_index")).toBe("80000"); + expect(stateValue("telegram_burn_rescan_from_block")).toBeNull(); + + // Back to normal: monotonic, and nothing re-read below the cursor. + await advanceBurnCursor(database.db, launchTx + 1, 1); + expect(stateValue("telegram_burn_receive_tx_index")).toBe(String(launchTx + 40)); + fetch.mockClear(); + const quiet = await scanBurnReceives(database.db); + expect(quiet).toMatchObject({ announcements: [], rescanFrom: null }); + expect(fetch).toHaveBeenCalledTimes(2); + }); + + it("keeps a marker a second rollback moved while the scan ran", async () => { + database.raw.prepare("INSERT INTO chain_state (key, value) VALUES ('telegram_burn_rescan_from_block', '965901')").run(); + const { advanceBurnCursor } = await import("#api/telegram/burns"); + await advanceBurnCursor(database.db, 5, 6, 965_951); + expect(stateValue("telegram_burn_rescan_from_block")).toBe("965901"); + }); +}); + +describe("reward eligibility", () => { + it("counts only mints six confirmations deep, in the programme's order", () => { + const raw = database.raw; + insertFixtureLaunch(); + const TIP = 970_000; + expect(settledThrough(TIP)).toBe(969_995); + const mint = raw.prepare(`INSERT INTO launch_mints (tx_hash, launch_tx, block_index, source, earn_quantity, paid_quantity, tx_index) + VALUES (?, ?, ?, ?, '1', '1', ?)`); + mint.run("c".repeat(64), fixture.launch.tx_hash, 969_990, "third", 3); + mint.run("a".repeat(64), fixture.launch.tx_hash, 969_980, "first", 1); + mint.run("b".repeat(64), fixture.launch.tx_hash, 969_995, "six-deep", 2); + mint.run("d".repeat(64), fixture.launch.tx_hash, 969_996, "five-deep", 4); + mint.run("e".repeat(64), fixture.launch.tx_hash, TIP, "tip", 5); + const sources = (cutoff: number, tip: number) => + raw.prepare(eligibleMintsSql(cutoff, tip)).all().map(r => r.source); + expect(sources(10, TIP)).toEqual(["first", "third", "six-deep"]); + // Fewer than the cutoff: the script refuses rather than reach for shallow mints. + expect(sources(4, TIP)).toHaveLength(3); + expect(sources(4, TIP + 1)).toEqual(["first", "third", "six-deep", "five-deep"]); + expect(() => eligibleMintsSql(10, Number.NaN)).toThrow(); + expect(() => eligibleMintsSql(0, TIP)).toThrow(); + }); });