From e76b9be70f2def1aa6bdd941875ea8ea04744e14 Mon Sep 17 00:00:00 2001 From: Dan Anderson Date: Wed, 28 Oct 2026 22:33:24 -0400 Subject: [PATCH] Reduce Cloudflare query and storage operations --- apps/api/src/indexer/core-address-signals.ts | 18 ++++-- apps/api/src/recovery/import.ts | 25 +++++--- apps/api/src/recovery/reclassify.ts | 8 ++- apps/api/tests/core-address-signals.test.ts | 14 +++++ apps/api/tests/recovery-import.test.ts | 66 +++++++++++++++++++- 5 files changed, 114 insertions(+), 17 deletions(-) diff --git a/apps/api/src/indexer/core-address-signals.ts b/apps/api/src/indexer/core-address-signals.ts index a7ee41f..7245a93 100644 --- a/apps/api/src/indexer/core-address-signals.ts +++ b/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])); diff --git a/apps/api/src/recovery/import.ts b/apps/api/src/recovery/import.ts index bc39e76..36bb444 100644 --- a/apps/api/src/recovery/import.ts +++ b/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 { +export async function importRecoveryTransactions( + env: Env, + rows: RecoveryImportTransaction[], + options: RecoveryImportOptions = {}, +): Promise { 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"); diff --git a/apps/api/src/recovery/reclassify.ts b/apps/api/src/recovery/reclassify.ts index 83caa43..d46d50c 100644 --- a/apps/api/src/recovery/reclassify.ts +++ b/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 }; } diff --git a/apps/api/tests/core-address-signals.test.ts b/apps/api/tests/core-address-signals.test.ts index 7bd27c0..a8bba93 100644 --- a/apps/api/tests/core-address-signals.test.ts +++ b/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); diff --git a/apps/api/tests/recovery-import.test.ts b/apps/api/tests/recovery-import.test.ts index 359bde6..75229fc 100644 --- a/apps/api/tests/recovery-import.test.ts +++ b/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. */