From e5d13a1d129fc98384518c786efd77a06afb5c11 Mon Sep 17 00:00:00 2001 From: Dan Anderson Date: Sat, 22 Aug 2026 17:27:45 -0400 Subject: [PATCH] Resume the Bitcoin fee walk from a persisted cursor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fee runner restarted at the top of the missing-fee index every tick. The top ~930 candidates are 2019–2021 P2SH-segwit rows whose Counterparty tx_hash is a witness hash, so Electrs, mempool.space and blockstream all return 404 for them forever. Each run spent its whole 1,000-request budget on those same rows (logged as requested: 1000, updated: 0 after 128 s) and never reached the 1.08M pre-segwit rows below that Electrs can serve. Counterparty's own fee field was checked as an alternative source and rejected: against 5,000 stored Bitcoin-authoritative fees it disagreed on 7–44% of rows (mostly 0, some null, some different values), which is why the replayer stopped ingesting it on 2026-07-16. Now each run fetches transactions newer than a tip watermark first, then resumes the historical walk from core_state.bitcoin_fees_walk_cursor, so an unresolvable row costs one request per full cycle. Requests are smoothed to the provider's documented budget (3 per second) instead of six concurrent bursts that drew 429s; 300 candidates per two-minute tick clears the backlog in about five days. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016bNJBT1Ugmg2VG25BrdsJz --- apps/api/src/indexer/bitcoin-fees.ts | 158 +++++++++++++++++++++------ apps/api/tests/bitcoin-fees.test.ts | 133 +++++++++++++++++++++- 2 files changed, 255 insertions(+), 36 deletions(-) diff --git a/apps/api/src/indexer/bitcoin-fees.ts b/apps/api/src/indexer/bitcoin-fees.ts index 9831b0d..699bfd0 100644 --- a/apps/api/src/indexer/bitcoin-fees.ts +++ b/apps/api/src/indexer/bitcoin-fees.ts @@ -1,6 +1,10 @@ import type { D1Database } from "@cloudflare/workers-types"; import type { Env } from "#api/env"; -import { fetchTransactionFee } from "#api/integrations/electrs"; +import { + ELECTRS_REQUEST_BATCH_INTERVAL_MS, + ELECTRS_REQUEST_BATCH_SIZE, + fetchTransactionFee, +} from "#api/integrations/electrs"; import { hashToBytes } from "#api/indexer/identities"; export interface BitcoinFeeCandidate { @@ -15,12 +19,23 @@ export interface BitcoinFeeRow { const TX_HASH = /^[0-9a-f]{64}$/; const FEE_PAGE_SIZE = 100; -// Cloudflare permits six simultaneous outgoing connections per invocation. Matching that limit explicitly avoids -// relying on the runtime's connection queue. A 2026-07-16 production-provider sample completed 60/60 requests at -// ~69 requests/second with this concurrency, so 1,000 candidates is a conservative fraction of the paid Worker's -// 10,000-subrequest and 15-minute scheduled-invocation limits while materially shortening the finite backfill. -export const FEE_FETCH_CONCURRENCY = 6; -export const FEES_PER_RUN = 1_000; +/** + * Electrs replenishes roughly four requests a second and answers bursts with 429s, so fees are fetched + * at the provider's smoothed budget (see integrations/electrs.ts). 300 candidates is ~100 seconds of + * wall clock per two-minute tick; the 2026-07-16 figure of 1,000 predates the provider's tighter limit. + */ +export const FEE_FETCH_CONCURRENCY = ELECTRS_REQUEST_BATCH_SIZE; +export const FEES_PER_RUN = 300; +/** New transactions since the last run are fetched first so the tip never waits behind the historical walk. */ +export const FEE_TIP_PAGE_SIZE = FEE_PAGE_SIZE; +/** + * Some 2019–2021 P2SH-segwit transactions carry a witness hash as their Counterparty tx_hash, so no + * Bitcoin index can ever serve them by hash. Without a persisted cursor the walk restarted at the top + * every tick, spent its whole budget on those same rows, and never reached the million older rows + * Electrs can serve. The cursor makes each unresolvable row cost one request per full cycle instead. + */ +export const FEE_WALK_CURSOR_KEY = "bitcoin_fees_walk_cursor"; +export const FEE_TIP_WATERMARK_KEY = "bitcoin_fees_tip_watermark"; export async function listMissingBitcoinFees( db: D1Database, @@ -38,6 +53,21 @@ export async function listMissingBitcoinFees( return result.results; } +async function listMissingBitcoinFeesAbove( + db: D1Database, + above: number, + limit: number, +): Promise { + const result = await db + .prepare( + `SELECT tx_index,LOWER(HEX(tx_hash)) tx_hash FROM transactions + WHERE fee IS NULL AND tx_index>? ORDER BY tx_index DESC LIMIT ?`, + ) + .bind(above, limit) + .all(); + return result.results; +} + export function validBitcoinFeeRows(value: unknown): BitcoinFeeRow[] | null { if (!Array.isArray(value) || value.length === 0 || value.length > 100) return null; const rows: BitcoinFeeRow[] = []; @@ -63,40 +93,104 @@ export async function storeBitcoinFees(db: D1Database, rows: BitcoinFeeRow[]): P return results.reduce((sum, result) => sum + (result.results?.length ?? 0), 0); } -/** Keep the staging frontier current while the one-time historical exporter walks backward. */ +async function readCursor(db: D1Database, key: string): Promise { + const row = await db.prepare(`SELECT value FROM core_state WHERE key=?`).bind(key).first<{ value: string }>(); + const value = row === null ? Number.NaN : Number(row.value); + return Number.isSafeInteger(value) ? value : null; +} + +async function writeCursor(db: D1Database, key: string, value: number | null): Promise { + if (value === null) { + await db.prepare(`DELETE FROM core_state WHERE key=?`).bind(key).run(); + return; + } + await db + .prepare(`INSERT INTO core_state(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`) + .bind(key, String(value)) + .run(); +} + +async function readMaxTransactionIndex(db: D1Database): Promise { + const row = await db.prepare(`SELECT MAX(tx_index) tip FROM transactions`).first<{ tip: number | null }>(); + return row?.tip ?? 0; +} + +export type FetchTransactionFee = (baseUrl: string, txid: string) => Promise; + +async function fetchFees( + env: Pick, + rows: BitcoinFeeCandidate[], + fetchFee: FetchTransactionFee, +): Promise { + const fees: BitcoinFeeRow[] = []; + for (let offset = 0; offset < rows.length; offset += FEE_FETCH_CONCURRENCY) { + if (offset > 0) await new Promise((resolve) => setTimeout(resolve, ELECTRS_REQUEST_BATCH_INTERVAL_MS)); + const settled = await Promise.allSettled( + rows.slice(offset, offset + FEE_FETCH_CONCURRENCY).map(async (row) => ({ + tx_hash: row.tx_hash, + fee: await fetchFee(env.ELECTRS_API_BASE, row.tx_hash), + })), + ); + fees.push( + ...settled.flatMap((result) => + result.status === "fulfilled" && result.value.fee !== null + ? [{ tx_hash: result.value.tx_hash, fee: result.value.fee }] + : [], + ), + ); + } + return fees; +} + +export interface BitcoinFeeReconciliation { + requested: number; + updated: number; + /** Where the historical walk resumes next run; null once a full cycle completes and restarts at the top. */ + cursor: number | null; +} + +/** + * Fill Bitcoin-authoritative fees in two passes per run: transactions newer than the last tip watermark, + * then a bounded slice of the historical walk resumed from its persisted cursor. + */ export async function reconcileStagedBitcoinFees( env: Pick, limit = FEES_PER_RUN, -): Promise<{ requested: number; updated: number }> { + fetchFee: FetchTransactionFee = fetchTransactionFee, +): Promise { const boundedLimit = Math.min(FEES_PER_RUN, Math.max(1, Math.trunc(limit))); + const db = env.CORE_DB; let requested = 0; let updated = 0; - let after: number | null = null; + + const watermark = await readCursor(db, FEE_TIP_WATERMARK_KEY); + const maxTransactionIndex = await readMaxTransactionIndex(db); + if (watermark !== null && maxTransactionIndex > watermark) { + const tip = await listMissingBitcoinFeesAbove(db, watermark, Math.min(FEE_TIP_PAGE_SIZE, boundedLimit)); + requested += tip.length; + const fees = await fetchFees(env, tip, fetchFee); + if (fees.length > 0) updated += await storeBitcoinFees(db, fees); + } + // Anything the tip pass could not resolve falls through to the historical walk on its next cycle. + if (watermark !== maxTransactionIndex) await writeCursor(db, FEE_TIP_WATERMARK_KEY, maxTransactionIndex); + + let cursor = await readCursor(db, FEE_WALK_CURSOR_KEY); while (requested < boundedLimit) { const pageSize = Math.min(FEE_PAGE_SIZE, boundedLimit - requested); - const rows = await listMissingBitcoinFees(env.CORE_DB, after, pageSize); - if (rows.length === 0) break; + const rows = await listMissingBitcoinFees(db, cursor, pageSize); + if (rows.length === 0) { + cursor = null; + break; + } requested += rows.length; - after = rows.at(-1)!.tx_index; - - const fees: BitcoinFeeRow[] = []; - for (let offset = 0; offset < rows.length; offset += FEE_FETCH_CONCURRENCY) { - const settled = await Promise.allSettled( - rows.slice(offset, offset + FEE_FETCH_CONCURRENCY).map(async (row) => ({ - tx_hash: row.tx_hash, - fee: await fetchTransactionFee(env.ELECTRS_API_BASE, row.tx_hash), - })), - ); - fees.push( - ...settled.flatMap((result) => - result.status === "fulfilled" && result.value.fee !== null - ? [{ tx_hash: result.value.tx_hash, fee: result.value.fee }] - : [], - ), - ); + cursor = rows.at(-1)!.tx_index; + const fees = await fetchFees(env, rows, fetchFee); + if (fees.length > 0) updated += await storeBitcoinFees(db, fees); + if (rows.length < pageSize) { + cursor = null; + break; } - if (fees.length > 0) updated += await storeBitcoinFees(env.CORE_DB, fees); - if (rows.length < pageSize) break; } - return { requested, updated }; + await writeCursor(db, FEE_WALK_CURSOR_KEY, cursor); + return { requested, updated, cursor }; } diff --git a/apps/api/tests/bitcoin-fees.test.ts b/apps/api/tests/bitcoin-fees.test.ts index 49833e8..be4ec55 100644 --- a/apps/api/tests/bitcoin-fees.test.ts +++ b/apps/api/tests/bitcoin-fees.test.ts @@ -1,6 +1,15 @@ import assert from "node:assert/strict"; +import { readFileSync, readdirSync } from "node:fs"; +import { DatabaseSync } from "node:sqlite"; import { test } from "node:test"; -import { FEE_FETCH_CONCURRENCY, FEES_PER_RUN, validBitcoinFeeRows } from "#api/indexer/bitcoin-fees"; +import { + FEE_FETCH_CONCURRENCY, + FEE_TIP_WATERMARK_KEY, + FEE_WALK_CURSOR_KEY, + FEES_PER_RUN, + reconcileStagedBitcoinFees, + validBitcoinFeeRows, +} from "#api/indexer/bitcoin-fees"; const txHash = "1db7a85e9bbbcd9f60a62411e94f1ae8d3851642d0e3ca73e095d522bf234293"; @@ -18,7 +27,123 @@ test("rejects malformed fee batches", () => { assert.equal(validBitcoinFeeRows([{ tx_hash: "not-a-hash", fee: 1 }]), null); }); -test("scheduled fee maintenance stays within its reviewed resource budget", () => { - assert.equal(FEE_FETCH_CONCURRENCY, 6); - assert.equal(FEES_PER_RUN, 1_000); +test("scheduled fee maintenance stays within the provider's smoothed budget", () => { + assert.equal(FEE_FETCH_CONCURRENCY, 3); + assert.equal(FEES_PER_RUN, 300); +}); + +const migrations = readdirSync("migrations-core") + .filter((name) => name.endsWith(".sql")) + .sort() + .map((name) => readFileSync(`migrations-core/${name}`, "utf8")); + +class Statement { + private values: unknown[] = []; + constructor( + private readonly db: DatabaseSync, + private readonly sql: string, + ) {} + bind(...values: unknown[]) { + this.values = values; + return this; + } + async run() { + this.db.prepare(this.sql).run(...(this.values as never[])); + return { success: true }; + } + async first() { + return (this.db.prepare(this.sql).get(...(this.values as never[])) as T | undefined) ?? null; + } + async all() { + return { results: this.db.prepare(this.sql).all(...(this.values as never[])) as T[] }; + } +} + +const d1 = (db: DatabaseSync) => + ({ + prepare: (sql: string) => new Statement(db, sql), + batch: (statements: Statement[]) => Promise.all(statements.map((statement) => statement.all())), + }) as unknown as D1Database; + +const hashOf = (index: number) => index.toString(16).padStart(64, "0"); + +function seed(count: number): DatabaseSync { + const db = new DatabaseSync(":memory:"); + for (const migration of migrations) db.exec(migration); + const insert = db.prepare( + `INSERT INTO transactions(tx_index,tx_hash,block_index,block_time,supported) VALUES(?,?,?,?,1)`, + ); + for (let index = 1; index <= count; index++) insert.run(index, Buffer.from(hashOf(index), "hex"), index, index * 600); + return db; +} + +const state = (db: DatabaseSync, key: string) => + (db.prepare(`SELECT value FROM core_state WHERE key=?`).get(key) as { value: string } | undefined)?.value ?? null; +const missing = (db: DatabaseSync) => + (db.prepare(`SELECT COUNT(*) n FROM transactions WHERE fee IS NULL`).get() as { n: number }).n; + +/** Electrs-like provider: even indexes resolve, odd indexes are witness-hash rows that 404 forever. */ +const provider = (calls: string[]) => async (_base: string, hash: string) => { + calls.push(hash); + const index = Number.parseInt(hash, 16); + return index % 2 === 0 ? index * 10 : null; +}; + +test("the historical walk resumes from its cursor instead of re-requesting unresolvable rows", async () => { + const db = seed(12); + const env = { CORE_DB: d1(db), ELECTRS_API_BASE: "https://electrs.test" }; + const calls: string[] = []; + + const first = await reconcileStagedBitcoinFees(env, 4, provider(calls)); + assert.deepEqual(first, { requested: 4, updated: 2, cursor: 9 }); + assert.equal(state(db, FEE_WALK_CURSOR_KEY), "9"); + assert.equal(state(db, FEE_TIP_WATERMARK_KEY), "12"); + + const second = await reconcileStagedBitcoinFees(env, 4, provider(calls)); + assert.deepEqual(second, { requested: 4, updated: 2, cursor: 5 }); + assert.deepEqual( + calls.map((hash) => Number.parseInt(hash, 16)), + [12, 11, 10, 9, 8, 7, 6, 5], + ); + + const third = await reconcileStagedBitcoinFees(env, 4, provider(calls)); + assert.deepEqual(third, { requested: 4, updated: 2, cursor: 1 }); + const fourth = await reconcileStagedBitcoinFees(env, 4, provider(calls)); + assert.deepEqual(fourth, { requested: 0, updated: 0, cursor: null }); + assert.equal(state(db, FEE_WALK_CURSOR_KEY), null); + assert.equal(missing(db), 6); + assert.equal(state(db, "bitcoin_fees_remaining"), "6"); + + // A completed cycle restarts at the top; the unresolvable rows cost one request per cycle. + calls.length = 0; + const fifth = await reconcileStagedBitcoinFees(env, 4, provider(calls)); + assert.deepEqual(fifth, { requested: 4, updated: 0, cursor: 5 }); + assert.deepEqual( + calls.map((hash) => Number.parseInt(hash, 16)), + [11, 9, 7, 5], + ); +}); + +test("new transactions are fetched ahead of the historical walk", async () => { + const db = seed(6); + const env = { CORE_DB: d1(db), ELECTRS_API_BASE: "https://electrs.test" }; + const calls: string[] = []; + await reconcileStagedBitcoinFees(env, 2, provider(calls)); + assert.equal(state(db, FEE_WALK_CURSOR_KEY), "5"); + + db.prepare(`INSERT INTO transactions(tx_index,tx_hash,block_index,block_time,supported) VALUES(?,?,?,?,1)`).run( + 8, + Buffer.from(hashOf(8), "hex"), + 8, + 4_800, + ); + calls.length = 0; + const result = await reconcileStagedBitcoinFees(env, 2, provider(calls)); + assert.deepEqual(result, { requested: 2, updated: 2, cursor: 4 }); + assert.deepEqual( + calls.map((hash) => Number.parseInt(hash, 16)), + [8, 4], + ); + assert.equal(state(db, FEE_TIP_WATERMARK_KEY), "8"); + assert.equal((db.prepare(`SELECT fee FROM transactions WHERE tx_index=8`).get() as { fee: string }).fee, "80"); });