perf: skip receipt indexing for acknowledged run cleanup

Use the initial run state to skip unnecessary disk reads, while retaining the locked recheck, stale pending-marker cleanup, and budget-stop delivery. Extend the backlog regression to verify duplicate finalization performs no spool listing.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip committed 2026-10-04 02:27:34 -07:00
1 parent 3415e577b9
commit 66930ab7f5
2 files changed
+18 -3

No files matched your search

@@ -349,6 +349,16 @@ const support = await getEmbeddedPostgresTestSupport();
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);
const listing = vi.spyOn(fs, "readdir");
try {
for (const run of runs) expect(await accountRunCost(db, run.id)).toBe(false);
// An acknowledged run with a stale pending flag still needs its locked
// cleanup, but neither form of duplicate finalization needs spool I/O.
await db.update(heartbeatRuns).set({ costAccountingPending: true }).where(eq(heartbeatRuns.id, f.run.id));
expect(await accountRunCost(db, f.run.id)).toBe(false);
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].costAccountingPending).toBe(false);
expect(listing).not.toHaveBeenCalled();
} finally { listing.mockRestore(); }
await Promise.all([...foreignFiles].map(file => fs.rm(file)));
});
+8 -3
View File
@@ -16,9 +16,14 @@ 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 = {}, receiptIndex?: UsageReceiptIndex) {
const [identity] = await db.select({ companyId: heartbeatRuns.companyId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
const [identity] = await db.select({ companyId: heartbeatRuns.companyId, status: heartbeatRuns.status,
pending: heartbeatRuns.costAccountingPending, accountedAt: heartbeatRuns.costAccountedAt,
}).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
if (!identity) return false;
const index = receiptIndex ?? await indexPendingUsageReceipts();
// Finalization can call twice. Avoid disk work for settled or active runs,
// but retain the locked recheck, stale-marker cleanup, and stop delivery.
const index = identity.pending && !identity.accountedAt && terminalStatuses.includes(identity.status)
? receiptIndex ?? await indexPendingUsageReceipts() : undefined;
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");
@@ -30,7 +35,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, { retainFiles: true, index });
recoveredFiles = await recoverPendingRunUsageReceipts(tx, { companyId: identity.companyId, 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;