Prioritize historical recovery reclassification

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<? GROUP BY txid ORDER BY txid LIMIT ?`;
 
+/**
+ * Classifier v2 changes only one old verdict: an unverified historical 1-of-2 output immediately
+ * after a verified historical anchor may inherit that anchor's provenance. Work those transactions
+ * before the generic version sweep so the user-visible recovery improvement does not sit behind
+ * hundreds of thousands of unchanged current-layout transactions.
+ *
+ * This is deliberately only a queue hint. The TypeScript classifier still re-parses the complete
+ * raw transaction and proves the exact key and uninterrupted vout run; SQL never grants recovery.
+ */
+export const RECOVERY_RECLASSIFY_PRIORITY_QUEUE_SQL = `
+  SELECT continuation.txid,MIN(continuation.block_height) block_height,
+         MIN(continuation.block_time) block_time
+    FROM recovery_outputs continuation INDEXED BY recovery_outputs_classifier
+    JOIN recovery_outputs anchor
+      ON anchor.txid=continuation.txid AND anchor.vout=continuation.vout-1
+   WHERE continuation.classifier_version<?
+     AND continuation.layout='historical-1-of-2'
+     AND continuation.classification='unverified'
+     AND anchor.layout='historical-1-of-2'
+     AND anchor.classification IN ('recoverable','spent')
+   GROUP BY continuation.txid ORDER BY continuation.txid LIMIT ?`;
+
+const priorityStateKey = `recovery_reclassify_priority_${RECOVERY_CLASSIFIER_VERSION}_done`;
+
 interface OutdatedTransactionRow {
   txid: string;
   block_height: number | null;
@@ -56,36 +81,62 @@ export async function reclassifyRecoveryOutputs(
     .bind(RECOVERY_CLASSIFIER_VERSION)
     .first<{ pending: number }>();
   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<OutdatedTransactionRow>();
+
+  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<OutdatedTransactionRow>();
+  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<OutdatedTransactionRow>();
+  }
   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`,
       )
         .bind(row.txid)
         .all<ExistingSpendRow>(),
     ]);
-    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 };

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<void> {
+  // 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<void> {
   // 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));

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(