From 79182a0b05f6ac53d482839782c6d5f3dc55662a Mon Sep 17 00:00:00 2001 From: Dan Anderson Date: Wed, 28 Oct 2026 21:59:16 -0400 Subject: [PATCH] Prioritize historical recovery reclassification --- apps/api/src/recovery/reclassify.ts | 87 +++++++++++++++---- .../api/src/scheduler/recovery-maintenance.ts | 6 +- apps/api/tests/recovery-reclassify.test.ts | 37 ++++++-- 3 files changed, 103 insertions(+), 27 deletions(-) diff --git a/apps/api/src/recovery/reclassify.ts b/apps/api/src/recovery/reclassify.ts index 7313ffb..83caa43 100644 --- a/apps/api/src/recovery/reclassify.ts +++ b/apps/api/src/recovery/reclassify.ts @@ -1,4 +1,5 @@ import type { Env } from "#api/env"; +import { mapWithLimit } from "#api/lib/net"; import { RECOVERY_CLASSIFIER_VERSION } from "#api/recovery/classifier"; import { importRecoveryTransactions, type RecoveryImportOutput } from "#api/recovery/import"; import { recoveryCandidates } from "#api/recovery/scanner"; @@ -21,6 +22,30 @@ export const RECOVERY_RECLASSIFY_QUEUE_SQL = ` SELECT txid,MIN(block_height) block_height,MIN(block_time) block_time FROM recovery_outputs WHERE classifier_version(); if (!pending) return { transactions: 0, outputs: 0, missing: 0 }; - const outdated = await env.RECOVERY_DB.prepare(RECOVERY_RECLASSIFY_QUEUE_SQL) - .bind(RECOVERY_CLASSIFIER_VERSION, limit) - .all(); + + const priorityDone = await env.RECOVERY_DB.prepare(`SELECT 1 done FROM recovery_state WHERE key=?`) + .bind(priorityStateKey) + .first<{ done: number }>(); + let outdated = priorityDone + ? { results: [] as OutdatedTransactionRow[] } + : await env.RECOVERY_DB.prepare(RECOVERY_RECLASSIFY_PRIORITY_QUEUE_SQL) + .bind(RECOVERY_CLASSIFIER_VERSION, limit) + .all(); + if (!priorityDone && outdated.results.length === 0) { + await env.RECOVERY_DB.prepare( + `INSERT INTO recovery_state(key,value,updated_at) VALUES (?,'1',unixepoch()) ON CONFLICT(key) DO NOTHING`, + ) + .bind(priorityStateKey) + .run(); + } + if (outdated.results.length === 0) { + outdated = await env.RECOVERY_DB.prepare(RECOVERY_RECLASSIFY_QUEUE_SQL) + .bind(RECOVERY_CLASSIFIER_VERSION, limit) + .all(); + } if (outdated.results.length === 0) return { transactions: 0, outputs: 0, missing: 0 }; let outputs = 0; let missing = 0; - for (const row of outdated.results) { - const object = await env.RECOVERY_TRANSACTIONS.get(`transactions/${row.txid}.hex`); - if (!object) { - // The R2 audit owns body coverage; skipping here keeps one absent body from stalling the sweep. - missing++; - continue; - } - const [rawHex, existingSpends] = await Promise.all([ - object.text(), + const prepared = await mapWithLimit(outdated.results, async (row) => { + const [object, existingSpends] = await Promise.all([ + env.RECOVERY_TRANSACTIONS.get(`transactions/${row.txid}.hex`), env.RECOVERY_DB.prepare( `SELECT vout,spent_by_txid,spent_height FROM recovery_outputs WHERE txid=? AND spent_by_txid IS NOT NULL`, @@ -79,13 +116,27 @@ export async function reclassifyRecoveryOutputs( .bind(row.txid) .all(), ]); - const candidates = retainRecoverySpendEvidence( - recoveryCandidates(rawHex, row.block_height ?? 0, row.block_time), - existingSpends.results, - ); - if (candidates.length === 0) continue; + if (!object) return null; + const rawHex = await object.text(); + return { + row, + rawHex, + candidates: retainRecoverySpendEvidence( + recoveryCandidates(rawHex, row.block_height ?? 0, row.block_time), + existingSpends.results, + ), + }; + }); + + for (const item of prepared) { + if (!item) { + // The R2 audit owns body coverage; skipping here keeps one absent body from stalling the sweep. + missing++; + continue; + } + if (item.candidates.length === 0) continue; outputs += await importRecoveryTransactions(env, [ - { txid: row.txid, raw_transaction_hex: rawHex, outputs: candidates }, + { txid: item.row.txid, raw_transaction_hex: item.rawHex, outputs: item.candidates }, ]); } return { transactions: outdated.results.length - missing, outputs, missing }; diff --git a/apps/api/src/scheduler/recovery-maintenance.ts b/apps/api/src/scheduler/recovery-maintenance.ts index 5dea3a2..ef766f2 100644 --- a/apps/api/src/scheduler/recovery-maintenance.ts +++ b/apps/api/src/scheduler/recovery-maintenance.ts @@ -11,6 +11,9 @@ import { runScheduledJob } from "#api/scheduler/job"; /** Independent recovery-database/provider lane; never delays canonical ingestion or projections. */ export async function runRecoveryMaintenance(env: Env): Promise { + // Classifier upgrades are rare and R2-backed. Give their bounded queue first use of the recovery + // invocation so a slow, unrelated CORE_DB projection cannot delay newly proven Bitcoin outputs. + await runScheduledJob("reclassifyRecoveryOutputs", () => reclassifyRecoveryOutputs(env, 50)); // Address projection gets a fresh CORE_DB invocation. Running it after the large asset projection can exhaust // D1's internal compound-statement budget even though both jobs are independently valid. await runScheduledJob("runCoreAddressSignalsStep", () => runCoreAddressSignalsStep(env.CORE_DB)); @@ -19,9 +22,6 @@ export async function runRecoveryMaintenance(env: Env): Promise { // durable cursor only after the complete page succeeds. Use a larger page while catching up so a restart // does not leave the recovery index days behind the canonical transaction mirror. await runScheduledJob("scanRecoveryTransactions", () => scanRecoveryTransactions(env, 100)); - // Re-deciding an already-indexed transaction reads only R2, so it can run alongside the chain-bound - // jobs without competing for the Electrs budget that caps everything else on this lane. - await runScheduledJob("reclassifyRecoveryOutputs", () => reclassifyRecoveryOutputs(env, 25)); await runScheduledJob("verifyRecoveryTransactions", () => verifyRecoveryTransactions(env, 10)); await runScheduledJob("reconcileRecoveryAttempts", () => reconcileRecoveryAttempts(env, 25)); await runScheduledJob("refreshRecoveryStats", () => refreshRecoveryStats(env)); diff --git a/apps/api/tests/recovery-reclassify.test.ts b/apps/api/tests/recovery-reclassify.test.ts index 1a1b8bd..fb1eafa 100644 --- a/apps/api/tests/recovery-reclassify.test.ts +++ b/apps/api/tests/recovery-reclassify.test.ts @@ -2,13 +2,18 @@ import assert from "node:assert/strict"; import { test } from "node:test"; import { DatabaseSync } from "node:sqlite"; import { RECOVERY_CLASSIFIER_VERSION } from "#api/recovery/classifier"; -import { RECOVERY_RECLASSIFY_QUEUE_SQL, retainRecoverySpendEvidence } from "#api/recovery/reclassify"; +import { + RECOVERY_RECLASSIFY_PRIORITY_QUEUE_SQL, + RECOVERY_RECLASSIFY_QUEUE_SQL, + retainRecoverySpendEvidence, +} from "#api/recovery/reclassify"; function database(): DatabaseSync { const db = new DatabaseSync(":memory:"); db.exec(` CREATE TABLE recovery_outputs(txid TEXT NOT NULL,vout INTEGER NOT NULL,block_height INTEGER,block_time INTEGER, - classifier_version INTEGER NOT NULL,PRIMARY KEY(txid,vout)) WITHOUT ROWID; + classifier_version INTEGER NOT NULL,layout TEXT,classification TEXT,PRIMARY KEY(txid,vout)) WITHOUT ROWID; + CREATE INDEX recovery_outputs_classifier ON recovery_outputs(classifier_version,txid); `); return db; } @@ -25,7 +30,7 @@ function queued(db: DatabaseSync, limit = 10) { test("only transactions decided by an older classifier are queued", () => { const db = database(); - const add = db.prepare(`INSERT INTO recovery_outputs VALUES (?,?,800000,1700000000,?)`); + const add = db.prepare(`INSERT INTO recovery_outputs VALUES (?,?,800000,1700000000,?,NULL,NULL)`); add.run("aa", 0, RECOVERY_CLASSIFIER_VERSION - 1); add.run("bb", 0, RECOVERY_CLASSIFIER_VERSION); @@ -34,7 +39,7 @@ test("only transactions decided by an older classifier are queued", () => { test("a transaction is queued once and carries the chain metadata its outputs recorded", () => { const db = database(); - const add = db.prepare(`INSERT INTO recovery_outputs VALUES (?,?,?,?,?)`); + const add = db.prepare(`INSERT INTO recovery_outputs VALUES (?,?,?,?,?,NULL,NULL)`); add.run("aa", 0, 800_000, 1_700_000_000, RECOVERY_CLASSIFIER_VERSION - 1); add.run("aa", 1, 800_000, 1_700_000_000, RECOVERY_CLASSIFIER_VERSION - 1); @@ -43,7 +48,7 @@ test("a transaction is queued once and carries the chain metadata its outputs re test("the sweep walks the backlog deterministically so it can resume", () => { const db = database(); - const add = db.prepare(`INSERT INTO recovery_outputs VALUES (?,0,1,1,?)`); + const add = db.prepare(`INSERT INTO recovery_outputs VALUES (?,0,1,1,?,NULL,NULL)`); for (const txid of ["cc", "aa", "bb"]) add.run(txid, RECOVERY_CLASSIFIER_VERSION - 1); assert.deepEqual( @@ -54,11 +59,31 @@ test("the sweep walks the backlog deterministically so it can resume", () => { test("an index already on the current classifier has nothing to do", () => { const db = database(); - db.prepare(`INSERT INTO recovery_outputs VALUES ('aa',0,1,1,?)`).run(RECOVERY_CLASSIFIER_VERSION); + db.prepare(`INSERT INTO recovery_outputs VALUES ('aa',0,1,1,?,NULL,NULL)`).run(RECOVERY_CLASSIFIER_VERSION); assert.deepEqual(queued(db), []); }); +test("the priority queue contains only unverified historical continuations after a proven anchor", () => { + const db = database(); + const add = db.prepare(`INSERT INTO recovery_outputs VALUES (?,?,?,?,?,?,?)`); + const old = RECOVERY_CLASSIFIER_VERSION - 1; + add.run("eligible", 0, 300_000, 1_400_000_000, old, "historical-1-of-2", "recoverable"); + add.run("eligible", 1, 300_000, 1_400_000_000, old, "historical-1-of-2", "unverified"); + add.run("gap", 0, 300_001, 1_400_000_001, old, "historical-1-of-2", "recoverable"); + add.run("gap", 2, 300_001, 1_400_000_001, old, "historical-1-of-2", "unverified"); + add.run("current", 0, 800_000, 1_700_000_000, old, "current-1-of-3", "recoverable"); + add.run("current", 1, 800_000, 1_700_000_000, old, "current-1-of-3", "unverified"); + + assert.deepEqual( + db + .prepare(RECOVERY_RECLASSIFY_PRIORITY_QUEUE_SQL) + .all(RECOVERY_CLASSIFIER_VERSION, 10) + .map((row) => ({ ...row })), + [{ txid: "eligible", block_height: 300_000, block_time: 1_400_000_000 }], + ); +}); + test("reclassification carries settled spend evidence into the new verdict", () => { assert.deepEqual( retainRecoverySpendEvidence(