Skip to content

Roll back burns and buyers with the index, and pay rewards only on settled mints #51

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
droplister merged 1 commit into from
Sep 30, 2026
Merged

Roll back burns and buyers with the index, and pay rewards only on settled mints #51

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 64 additions & 6 deletions apps/api/src/indexer/ledger.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -126,6 +128,8 @@ export async function checkLedger(db: D1Database): Promise<LedgerCheck> {
export interface RollbackResult {
mints_removed: number;
launches_removed: number;
burns_removed: number;
buyers_removed: number;
event_assets_reset: number;
}

Expand All @@ -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
Expand All @@ -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<RollbackResult> {
const cursors = await q<{ key: string }>(
Expand All @@ -160,6 +174,11 @@ export async function rollbackIndexTo(db: D1Database, block: number): Promise<Ro
);
const assets = cursors.map((row) => 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
Expand All @@ -180,9 +199,46 @@ export async function rollbackIndexTo(db: D1Database, block: number): Promise<Ro
WHERE last_mint_block > ?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(",");
Expand All @@ -196,8 +252,10 @@ export async function rollbackIndexTo(db: D1Database, block: number): Promise<Ro

const results = await db.batch(statements);
return {
mints_removed: results[1]!.meta.rows_written ?? 0,
launches_removed: results[3]!.meta.rows_written ?? 0,
mints_removed: results[MINTS]!.meta.rows_written ?? 0,
launches_removed: results[LAUNCHES]!.meta.rows_written ?? 0,
burns_removed: results[BURNS]!.meta.rows_written ?? 0,
buyers_removed: results[BUYERS]!.meta.rows_written ?? 0,
event_assets_reset: assets.length,
};
}
16 changes: 13 additions & 3 deletions apps/api/src/queries/communities.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -287,6 +287,7 @@ export async function writeCommunityRollup(db: D1Database, rollup: CommunityRoll
OR community_totals.paid_xcp IS NOT excluded.paid_xcp`,
)
.bind(rollup.minters, rollup.represented, rollup.creators, rollup.collectors, rollup.paid_xcp),
db.prepare(`DELETE FROM chain_state WHERE key = ?1`).bind(COMMUNITY_ROLLUP_STALE_KEY),
]);
return results.reduce((sum, r) => sum + (r.meta.rows_written ?? 0), 0);
}
Expand Down Expand Up @@ -331,8 +332,17 @@ export async function readCommunityRollup(db: D1Database): Promise<CommunityRoll
};
}

/** False until a refresh has written every column the current shape carries. */
/** Set by a ledger rollback: the stored rollup summarises mints that are gone. */
export const COMMUNITY_ROLLUP_STALE_KEY = "community_rollup_stale";

/** False until a refresh has written every column the current shape carries,
* and again after a rollback until the next one has. */
export async function hasCommunityRollup(db: D1Database): Promise<boolean> {
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;
}
143 changes: 95 additions & 48 deletions apps/api/src/telegram/burns.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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<T>(
fetchPage: (cursor: number | undefined) => Promise<{ result: T[]; next_cursor: number | null }>,
reached: (row: T) => boolean,
take: (row: T) => void,
what: string,
): Promise<void> {
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);
Expand Down Expand Up @@ -242,10 +277,11 @@ async function seedBurnEvents(db: D1Database): Promise<void> {
export async function scanBurnReceives(db: D1Database): Promise<BurnScan> {
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);
Expand All @@ -254,6 +290,8 @@ export async function scanBurnReceives(db: D1Database): Promise<BurnScan> {
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);
Expand All @@ -265,30 +303,22 @@ export async function scanBurnReceives(db: D1Database): Promise<BurnScan> {
const latestReceive = latestReceives.result[0];
let nextCursor = receiveStored;
const freshSends = new Map<string, CpAddressReceive>();
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");
}
}
}
Expand All @@ -297,28 +327,31 @@ export async function scanBurnReceives(db: D1Database): Promise<BurnScan> {
const latestDestruction = latestDestructions.result[0];
let nextDestructionCursor = destructionStored;
const freshDestructions = new Map<number, CpAssetDestructionEvent>();
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",
);
}
}
}
Expand Down Expand Up @@ -390,29 +423,43 @@ export async function scanBurnReceives(db: D1Database): Promise<BurnScan> {
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<void> {
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);
}
4 changes: 3 additions & 1 deletion apps/api/src/telegram/live.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,7 @@ export async function announceLive(env: Env, height: number): Promise<LiveResult
event: "burn_scan_skipped",
error: error instanceof Error ? error.message : String(error),
});
return { announcements: [], nextCursor: null, nextDestructionCursor: null, seeded: false };
return { announcements: [], nextCursor: null, nextDestructionCursor: null, seeded: false, rescanFrom: null };
});
for (const burn of burnScan.announcements) {
items.push({ key: burn.key, a: burn.a, mintOf: null, earned: "0", paid: "0" });
Expand Down Expand Up @@ -267,6 +267,7 @@ export async function announceLive(env: Env, height: number): Promise<LiveResult
env.DB,
burnScan.nextCursor,
burnScan.nextDestructionCursor,
burnScan.rescanFrom,
);
return { announced: 0, queued: 0 };
}
Expand All @@ -279,6 +280,7 @@ export async function announceLive(env: Env, height: number): Promise<LiveResult
env.DB,
burnScan.nextCursor,
burnScan.nextDestructionCursor,
burnScan.rescanFrom,
);
return { announced: queued.accepted, queued: queued.depth };
}
Loading
Loading