Resume the Bitcoin fee walk from a persisted cursor

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<BitcoinFeeCandidate[]> {
+  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<BitcoinFeeCandidate>();
+  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<number | null> {
+  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<void> {
+  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<number> {
+  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<number | null>;
+
+async function fetchFees(
+  env: Pick<Env, "ELECTRS_API_BASE">,
+  rows: BitcoinFeeCandidate[],
+  fetchFee: FetchTransactionFee,
+): Promise<BitcoinFeeRow[]> {
+  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<Env, "CORE_DB" | "ELECTRS_API_BASE">,
   limit = FEES_PER_RUN,
-): Promise<{ requested: number; updated: number }> {
+  fetchFee: FetchTransactionFee = fetchTransactionFee,
+): Promise<BitcoinFeeReconciliation> {
   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 };
 }

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<T>() {
+    return (this.db.prepare(this.sql).get(...(this.values as never[])) as T | undefined) ?? null;
+  }
+  async all<T>() {
+    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");
 });