perf: index receipt recovery outside company accounting locks

Reuse a batch index of immutable receipt identities so settlements read only matching or newly published files. Verify unrelated receipts are read once, company locks remain available during indexing, and late publications still settle correctly.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/COST-ACCOUNTING.md

@@ -75,6 +75,8 @@ Before replacing a stopped run's recorder, recovery saves every pending spool re
 
 Settlement also drains every pending spool receipt for the run under the company accounting lock before checking completeness or acknowledging usage. A newer partial receipt prevents settlement of an older complete snapshot. Recovered journal writes and ledger settlement share a transaction; spool files are removed only after its commit, so a failed charge or projection write leaves the original evidence available for retry. A concurrent replay rename causes a rescan; repeated file movement defers settlement.
 
+Recovery indexes immutable spool-file identities once per batch, outside company locks. Each settlement skips indexed files belonging to other runs, revalidates its matching receipts, and checks newly published or renamed files. Standalone settlement builds the same index before taking its lock. This avoids repeatedly reading unrelated receipt contents while admissions and ledger writes wait.
+
 The spool survives process death on persistent local storage, including the tested boundary before the first database write. It does not survive destruction of that storage, and cannot recover provider activity that was never emitted as usage. Back up persistent instance storage together with PostgreSQL. No arbitrary provider-log scraping or automatic provider re-execution occurs during accounting recovery.
 
 Board-only routes under `/api/companies/:companyId/accounting` provide:

server/src/__tests__/cost-accounting-operations-edges.test.ts

@@ -303,6 +303,55 @@ const support = await getEmbeddedPostgresTestSupport();
     expect(await fs.readdir(spool)).toHaveLength(unrelated.length + 1);
   });
 
+  it("indexes a recovery backlog once outside company locks and still sees newly published receipts", async () => {
+    const f = await fixture(), other = await fixture(), spool = usageReceiptSpoolPath();
+    const runs = [f.run];
+    for (let i = 0; i < 7; i++) {
+      const [run] = await db.insert(heartbeatRuns).values({ companyId: f.company.id, agentId: f.agent.id, status: "running" }).returning();
+      runs.push(run);
+    }
+    for (const run of runs) {
+      const recorder = await createRunUsageRecorder(db, { companyId: f.company.id, runId: run.id, adapterType: "process" });
+      await recorder.capture({ complete: true, usage: { inputTokens: 50, outputTokens: 5 }, costUsd: 0.05 });
+      await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.id));
+    }
+    const [first] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id));
+    const envelope: UsageReceiptEnvelope = { schema: "paperclip/accounting-receipt/v1", id: randomUUID(), companyId: other.company.id,
+      runId: other.run.id, sourceId: randomUUID(), sequence: 1, receivedAt: new Date().toISOString(), adapterType: "process", receipt: { complete: false } };
+    const foreignFiles = new Set<string>();
+    for (let i = 0; i < 256; i++) foreignFiles.add(await spoolUsageReceipt({ ...envelope, id: randomUUID(), sequence: i + 1 }));
+    const reads = new Map<string, number>(), read = fs.readFile.bind(fs);
+    let checkedLock = false, lockWasAvailable = false;
+    const spy = vi.spyOn(fs, "readFile").mockImplementation(async (file, options) => {
+      if (typeof file === "string" && foreignFiles.has(file)) {
+        reads.set(file, (reads.get(file) ?? 0) + 1);
+        if (!checkedLock) {
+          checkedLock = true;
+          await db.transaction(async tx => {
+            await tx.execute(sql`set local lock_timeout = '100ms'`);
+            await tx.select().from(companies).where(eq(companies.id, f.company.id)).for("no key update");
+          });
+          lockWasAvailable = true;
+          // Published after the index's directory snapshot: the locked drain
+          // must discover it instead of trusting a stale batch inventory.
+          await spoolUsageReceipt({ ...envelope, id: randomUUID(), companyId: f.company.id, runId: f.run.id,
+            sourceId: String(first.usageJson?.accountingReceiptSourceId), sequence: 2,
+            receipt: { complete: true, usage: { inputTokens: 200, outputTokens: 20 }, costUsd: 0.2 } });
+        }
+      }
+      return read(file, options);
+    });
+    try { await reconcileRunCosts(db); } finally { spy.mockRestore(); }
+    expect(lockWasAvailable).toBe(true);
+    expect(reads.size).toBe(256);
+    expect([...reads.values()]).toEqual(Array(256).fill(1));
+    const charges = await db.select().from(costEvents).where(eq(costEvents.companyId, f.company.id));
+    expect(charges).toHaveLength(8);
+    expect(charges.find(charge => charge.heartbeatRunId === f.run.id)).toMatchObject({ inputTokens: 200, costCents: 20 });
+    expect((await costService(db).summary(f.company.id)).spendCents).toBe(55);
+    await Promise.all([...foreignFiles].map(file => fs.rm(file)));
+  });
+
   it("reports over-limit advisory budgets without blocking, and tolerates legacy active zero limits", async () => {
     const f = await fixture(), budgets = budgetService(db);
     const policy = await budgets.upsertPolicy(f.company.id, { scopeType: "company", scopeId: f.company.id, amount: 1, hardStopEnabled: false }, "board");

server/src/services/run-cost-accounting.ts

@@ -6,7 +6,7 @@ import { createCostEventInTransaction } from "./costs.js";
 import { budgetService, deliverBudgetEnforcement, type BudgetServiceHooks } from "./budgets.js";
 import { logger } from "../middleware/logger.js";
 import { promises as fs } from "node:fs";
-import { recoverPendingRunUsageReceipts } from "./usage-receipts.js";
+import { indexPendingUsageReceipts, recoverPendingRunUsageReceipts, type UsageReceiptIndex } from "./usage-receipts.js";
 
 const terminalStatuses = ["succeeded", "failed", "timed_out", "cancelled", "interrupted"];
 const object = (value: unknown): Record<string, unknown> => value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {};
@@ -15,9 +15,10 @@ const amount = (value: unknown) => typeof value === "number" && Number.isFinite(
 
 /** The run's persisted usage is a durable receipt. Ledger, runtime totals, and
  * acknowledgement commit together, including after a restart or failed run. */
-export async function accountRunCost(db: Db, runId: string, hooks: BudgetServiceHooks = {}) {
+export async function accountRunCost(db: Db, runId: string, hooks: BudgetServiceHooks = {}, receiptIndex?: UsageReceiptIndex) {
   const [identity] = await db.select({ companyId: heartbeatRuns.companyId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
   if (!identity) return false;
+  const index = receiptIndex ?? await indexPendingUsageReceipts();
   let recoveredFiles: string[] = [];
   const accounted = await withAccountingTransaction(db, identity.companyId, async (tx, publications) => {
     let [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update");
@@ -29,7 +30,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
     // The bounded global replay can leave a newer receipt behind an older
     // complete snapshot. Drain this run under the accounting lock before
     // deciding completeness, and retain disk evidence until the outer commit.
-    recoveredFiles = await recoverPendingRunUsageReceipts(tx, { ...identity, runId }, undefined, true);
+    recoveredFiles = await recoverPendingRunUsageReceipts(tx, { ...identity, runId }, undefined, { retainFiles: true, index });
     [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
     const preProviderFailure = object(object(run.resultJson).executionRecovery).providerWorkStarted === false
       || object(run.usageJson).accountingProviderWorkStarted === false;
@@ -145,8 +146,9 @@ export async function reconcileRunCosts(db: Db, hooks: BudgetServiceHooks = {})
     sql`(${heartbeatRuns.usageJson}->>'accountingReceiptReady' is distinct from 'false' or ${heartbeatRuns.resultJson}->'executionRecovery'->>'providerWorkStarted' = 'false' or ${heartbeatRuns.usageJson}->>'accountingProviderWorkStarted' = 'false')`,
   )).orderBy(asc(heartbeatRuns.updatedAt), asc(heartbeatRuns.id)).limit(100);
   let accounted = 0;
+  const receiptIndex = pending.length > 0 ? await indexPendingUsageReceipts() : undefined;
   for (const run of pending) {
-    try { if (await accountRunCost(db, run.id, hooks)) accounted++; }
+    try { if (await accountRunCost(db, run.id, hooks, receiptIndex)) accounted++; }
     catch (error) {
       // Move failed attempts behind other pending work so a poisoned receipt
       // cannot monopolize the bounded batch on every recovery tick.

server/src/services/usage-receipts.ts

@@ -113,7 +113,39 @@ export async function replayUsageReceipts(db: Db, directory = usageReceiptSpoolP
   return { replayed, failed };
 }
 
-export async function recoverPendingRunUsageReceipts(db: Db, input: { companyId: string; runId: string }, directory = usageReceiptSpoolPath(), retainFiles = false) {
+type ReceiptIdentity = { companyId: unknown; runId: unknown } | null;
+export type UsageReceiptIndex = Map<string, ReceiptIdentity>;
+function receiptIdentity(raw: unknown): ReceiptIdentity {
+  return raw && typeof raw === "object" && "companyId" in raw && "runId" in raw
+    ? { companyId: raw.companyId, runId: raw.runId } : null;
+}
+
+/** Build once per recovery batch, before acquiring any accounting lock.
+ * Spool publications are immutable, uniquely named files. Cache identities,
+ * never receipt contents: matching evidence is revalidated during settlement. */
+export async function indexPendingUsageReceipts(directory = usageReceiptSpoolPath()): Promise<UsageReceiptIndex> {
+  const index: UsageReceiptIndex = new Map();
+  let names: string[];
+  try { names = await fs.readdir(directory); }
+  // Indexing is only an optimization. The locked drain reports missing or
+  // unreadable storage through the existing per-run recovery error path.
+  catch { return index; }
+  for (const name of names.filter(name => name.endsWith(".json"))) {
+    const file = path.join(directory, name);
+    try {
+      const info = await fs.lstat(file);
+      if (!info.isFile() || info.size > 1024 * 1024) { index.set(file, null); continue; }
+      index.set(file, receiptIdentity(JSON.parse(await fs.readFile(file, "utf8"))));
+    } catch (error) {
+      if (error instanceof SyntaxError) index.set(file, null);
+      // Unreadable or concurrently moved entries remain unknown and are
+      // checked again under the lock, preserving per-run failure reporting.
+    }
+  }
+  return index;
+}
+
+export async function recoverPendingRunUsageReceipts(db: Db, input: { companyId: string; runId: string }, directory = usageReceiptSpoolPath(), options: { retainFiles?: boolean; index?: UsageReceiptIndex } = {}) {
   // The bounded sweep may not reach this run. Its durable evidence must
   // reach the journal before source replacement or ledger acknowledgement.
   // A failed save aborts the operation so recovery can retry.
@@ -128,6 +160,8 @@ export async function recoverPendingRunUsageReceipts(db: Db, input: { companyId:
     for (const name of names.filter(name => name.endsWith(".json"))) {
       const file = path.join(directory, name);
       if (recovered.has(file)) continue;
+      const indexed = options.index?.get(file);
+      if (indexed !== undefined && (indexed?.companyId !== input.companyId || indexed?.runId !== input.runId)) continue;
       let raw: unknown;
       try {
         const info = await fs.lstat(file);
@@ -140,13 +174,14 @@ export async function recoverPendingRunUsageReceipts(db: Db, input: { companyId:
         if (error instanceof SyntaxError) continue;
         throw error;
       }
-      if (!raw || typeof raw !== "object" || !("companyId" in raw) || !("runId" in raw)
-        || raw.companyId !== input.companyId || raw.runId !== input.runId) continue;
+      const identity = receiptIdentity(raw);
+      options.index?.set(file, identity);
+      if (identity?.companyId !== input.companyId || identity?.runId !== input.runId) continue;
       await persistUsageReceipt(db, envelopeSchema.parse(raw));
       // Settlement uses the caller's transaction: deleting before its commit
       // would lose the only durable evidence if the later ledger write fails.
       recovered.add(file);
-      if (!retainFiles) await fs.rm(file, { force: true });
+      if (!options.retainFiles) await fs.rm(file, { force: true });
     }
     if (!moved) return [...recovered];
   }