Reduce Cloudflare query and storage operations

apps/api/src/indexer/core-address-signals.ts

@@ -19,13 +19,19 @@ const ACTIVITY_QUERIES = [
     WHERE tx0_address_id=?1 OR tx1_address_id=?1`,
   `SELECT min(block_index) first_block,max(block_index) last_block FROM dispensers
     WHERE source_id=?1 OR origin_id=?1`,
-  // A burn is the first thing most 2014 addresses did, and the burn address itself is every burn's
-  // destination — Counterparty only records the source, so without this the burn address "first
-  // appeared" when someone sent it an asset in 2018, four years late.
-  `SELECT min(block_index) first_block,max(block_index) last_block FROM burns
-    WHERE source_id=?1 OR ?1=(SELECT address_id FROM address_dictionary WHERE address='${BURN_ADDRESS}')`,
 ] as const;
 
+/**
+ * A burn is the first thing most 2014 addresses did, and the burn address itself is every burn's
+ * implicit destination. Keep that special case in TypeScript: putting it in a SQL OR makes SQLite
+ * scan the complete burns index for every ordinary address instead of seeking idx_burns_source.
+ * Both variants accept one binding so the surrounding D1 batch stays uniform.
+ */
+export const burnActivityQuery = (address: string) =>
+  address === BURN_ADDRESS
+    ? `SELECT min(block_index) first_block,max(block_index) last_block FROM burns WHERE ?1 IS NOT NULL`
+    : `SELECT min(block_index) first_block,max(block_index) last_block FROM burns WHERE source_id=?1`;
+
 // The DO UPDATE carries a WHERE guard so the repair lane skips rows whose signals are unchanged —
 // the weekly full-population cycle recycles ~500k addresses and nearly all are historically inert.
 // Every column here is absolute (block numbers, counts, flags), so unchanged history compares equal.
@@ -233,7 +239,7 @@ async function rebuildCoreAddressSignalsWithChanges(
     // subrequests per address, which capped a drain step near 110 addresses before hitting the
     // 1,000-subrequest invocation limit (and killed the whole maintenance lane when exceeded).
     const bounds = await db.batch<{ first_block: number | null; last_block: number | null }>(
-      ACTIVITY_QUERIES.map((query) => db.prepare(query).bind(identity.address_id)),
+      [...ACTIVITY_QUERIES, burnActivityQuery(address)].map((query) => db.prepare(query).bind(identity.address_id)),
     );
     const activity = bounds.flatMap((result) => result.results ?? []);
     const firstBlocks = activity.flatMap((row) => (row.first_block === null ? [] : [row.first_block]));

apps/api/src/recovery/import.ts

@@ -18,6 +18,11 @@ export interface RecoveryImportTransaction {
   outputs: RecoveryImportOutput[];
 }
 
+export interface RecoveryImportOptions {
+  /** The reclassifier reads the canonical body from this same bucket, so rewriting it is redundant. */
+  persistRawTransactions?: boolean;
+}
+
 const hashPattern = /^[0-9a-f]{64}$/i;
 
 export async function batchRecoveryStatements(
@@ -49,7 +54,11 @@ export const RECOVERY_OUTPUT_UPSERT_SQL = `
     ${keepSpend("spent_height")},verified_at=excluded.verified_at,classifier_version=excluded.classifier_version,
     ${keepSpend("chain_checked_at")}`;
 
-export async function importRecoveryTransactions(env: Env, rows: RecoveryImportTransaction[]): Promise<number> {
+export async function importRecoveryTransactions(
+  env: Env,
+  rows: RecoveryImportTransaction[],
+  options: RecoveryImportOptions = {},
+): Promise<number> {
   if (rows.length === 0 || rows.length > 100) throw new Error("expected 1 to 100 transactions");
   const now = Math.floor(Date.now() / 1000);
   const statements: D1PreparedStatement[] = [];
@@ -63,12 +72,14 @@ export async function importRecoveryTransactions(env: Env, rows: RecoveryImportT
     if (transaction.txid !== expectedTxid) throw new Error(`raw transaction hash mismatch: ${expectedTxid}`);
     if (!Array.isArray(row.outputs) || row.outputs.length === 0)
       throw new Error(`transaction has no outputs: ${expectedTxid}`);
-    blobs.push(
-      env.RECOVERY_TRANSACTIONS.put(`transactions/${expectedTxid}.hex`, row.raw_transaction_hex.toLowerCase(), {
-        httpMetadata: { contentType: "text/plain" },
-        customMetadata: { txid: expectedTxid },
-      }),
-    );
+    if (options.persistRawTransactions !== false) {
+      blobs.push(
+        env.RECOVERY_TRANSACTIONS.put(`transactions/${expectedTxid}.hex`, row.raw_transaction_hex.toLowerCase(), {
+          httpMetadata: { contentType: "text/plain" },
+          customMetadata: { txid: expectedTxid },
+        }),
+      );
+    }
 
     const validated = row.outputs.map((candidate) => {
       if (!Number.isSafeInteger(candidate.vout) || candidate.vout < 0) throw new Error("invalid output index");

apps/api/src/recovery/reclassify.ts

@@ -135,9 +135,11 @@ export async function reclassifyRecoveryOutputs(
       continue;
     }
     if (item.candidates.length === 0) continue;
-    outputs += await importRecoveryTransactions(env, [
-      { txid: item.row.txid, raw_transaction_hex: item.rawHex, outputs: item.candidates },
-    ]);
+    outputs += await importRecoveryTransactions(
+      env,
+      [{ txid: item.row.txid, raw_transaction_hex: item.rawHex, outputs: item.candidates }],
+      { persistRawTransactions: false },
+    );
   }
   return { transactions: outdated.results.length - missing, outputs, missing };
 }

apps/api/tests/core-address-signals.test.ts

@@ -4,6 +4,7 @@ import { readFileSync, readdirSync } from "node:fs";
 import { DatabaseSync } from "node:sqlite";
 import { test } from "node:test";
 import {
+  burnActivityQuery,
   enqueueCoreAddressSignals,
   rebuildCoreAddressSignals,
   runCoreAddressSignalsStep,
@@ -46,6 +47,19 @@ const d1 = (db: DatabaseSync) =>
     },
   }) as unknown as D1Database;
 
+test("ordinary address burn activity uses the source index instead of scanning every burn", () => {
+  const db = new DatabaseSync(":memory:");
+  for (const migration of migrations) db.exec(migration);
+  const plan = db
+    .prepare(`EXPLAIN QUERY PLAN ${burnActivityQuery("1ordinary")}`)
+    .all(1)
+    .map((row) => String(row.detail))
+    .join("\n");
+
+  assert.match(plan, /SEARCH burns USING COVERING INDEX idx_burns_source \(source_id=\?\)/);
+  assert.equal(plan.includes("SCAN burns"), false);
+});
+
 test("compact address signals recompute touched identities and converge to zero", async () => {
   const db = new DatabaseSync(":memory:");
   for (const migration of migrations) db.exec(migration);

apps/api/tests/recovery-import.test.ts

@@ -2,7 +2,11 @@ import { test } from "node:test";
 import assert from "node:assert/strict";
 import { DatabaseSync } from "node:sqlite";
 import type { D1PreparedStatement } from "@cloudflare/workers-types";
-import { batchRecoveryStatements, RECOVERY_OUTPUT_UPSERT_SQL } from "#api/recovery/import";
+import { secp256k1 } from "@noble/curves/secp256k1.js";
+import type { Env } from "#api/env";
+import { batchRecoveryStatements, importRecoveryTransactions, RECOVERY_OUTPUT_UPSERT_SQL } from "#api/recovery/import";
+import { parseRecoveryTransaction } from "#api/recovery/raw-transaction";
+import { recoveryCandidates } from "#api/recovery/scanner";
 
 test("recovery imports chunk large Counterparty transactions at the D1 batch ceiling", async () => {
   const batches: D1PreparedStatement[][] = [];
@@ -24,6 +28,66 @@ test("recovery imports chunk large Counterparty transactions at the D1 batch cei
   assert.equal((batches[1]![0] as unknown as { index: number }).index, 100);
 });
 
+const hex = (bytes: Uint8Array) => Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join("");
+const recoveryKey = hex(secp256k1.getPublicKey(new Uint8Array(32).fill(7), true));
+const counterpartyPrefix = hex(new TextEncoder().encode("CNTRPRTY"));
+const dataKey = `08${counterpartyPrefix}${"00".repeat(24)}`;
+const multisig = `51${(recoveryKey.length / 2).toString(16)}${recoveryKey}21${dataKey}52ae`;
+const input = `${"12".repeat(32)}0000000000ffffffff`;
+const raw = `0100000001${input}01e803000000000000${(multisig.length / 2).toString(16)}${multisig}00000000`;
+
+function importEnvironment(puts: string[], batches: number[]): Env {
+  const statement = {
+    bind() {
+      return this;
+    },
+  } as unknown as D1PreparedStatement;
+  return {
+    RECOVERY_DB: {
+      prepare: () => statement,
+      batch: async (items: D1PreparedStatement[]) => {
+        batches.push(items.length);
+        return [];
+      },
+    } as unknown as D1Database,
+    RECOVERY_TRANSACTIONS: {
+      put: async (key: string) => {
+        puts.push(key);
+        return null;
+      },
+    } as unknown as R2Bucket,
+  } as Env;
+}
+
+const importRow = () => ({
+  txid: parseRecoveryTransaction(raw).txid,
+  raw_transaction_hex: raw,
+  outputs: recoveryCandidates(raw, 900_000, 1_700_000_000),
+});
+
+test("normal recovery imports persist their canonical raw transaction", async () => {
+  const puts: string[] = [];
+  const batches: number[] = [];
+
+  assert.equal(await importRecoveryTransactions(importEnvironment(puts, batches), [importRow()]), 1);
+  assert.equal(puts.length, 1);
+  assert.deepEqual(batches, [1]);
+});
+
+test("reclassification can reuse its R2 body without rewriting the same object", async () => {
+  const puts: string[] = [];
+  const batches: number[] = [];
+
+  assert.equal(
+    await importRecoveryTransactions(importEnvironment(puts, batches), [importRow()], {
+      persistRawTransactions: false,
+    }),
+    1,
+  );
+  assert.equal(puts.length, 0);
+  assert.deepEqual(batches, [1]);
+});
+
 type OutputRow = { classification: string; spent_by_txid: string | null; chain_checked_at: number | null };
 
 /** Re-import the same output the way the scanner does: creation facts only, no spend evidence. */