From 706bbd0ff31dd69e709610b0f50e4462c5b5c5c5 Mon Sep 17 00:00:00 2001 From: Dan Anderson Date: Wed, 30 Sep 2026 13:01:24 -0400 Subject: [PATCH] Roll back burns and buyers with the index, and pay rewards only on settled mints A Core re-parse or reorg now also takes back the chain-derived rows the first rollback missed: token burns after the fork (with the launches' burned supply and the burn count restated), buyers whose every buy was after it, and the community rollup, which is marked stale so the next tick rebuilds it instead of waiting hours. 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 rollback leaves a marker instead; the next scan re-reads by block from there, resets both cursors to what the node now has, and clears the marker with them. Announcement keys keep it from reposting. Telegram announcements and reward records stay, since they record things that happened off the chain. reward-batch.mjs now counts only mints six confirmations deep (tip - 5 or older), so a reorg cannot change who a frozen batch paid. --- apps/api/src/indexer/ledger.ts | 70 +++++++++++-- apps/api/src/queries/communities.ts | 16 ++- apps/api/src/telegram/burns.ts | 143 +++++++++++++++++--------- apps/api/src/telegram/live.ts | 4 +- docs/reward-distributions.md | 4 +- scripts/lib/reward-eligibility.mjs | 46 +++++++++ scripts/reward-batch.mjs | 39 +++---- tests/api-indexer-safety.test.ts | 154 +++++++++++++++++++++++++++- 8 files changed, 392 insertions(+), 84 deletions(-) create mode 100644 scripts/lib/reward-eligibility.mjs 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(); + }); });