From 1d01a36696cbcd82d2a043ed326d7bdffbe449eb Mon Sep 17 00:00:00 2001 From: Epinephrine Date: Mon, 9 Nov 2026 13:51:10 +0900 Subject: [PATCH 01/13] fix(usage): page filtered index candidates --- .../management/usage-aggregate-cache.ts | 53 +++++----- src/usage/filter-index.ts | 97 ++++++++++++++----- structure/usage-client-attribution.md | 5 +- tests/usage/usage-filter-index.test.ts | 27 ++++++ 4 files changed, 134 insertions(+), 48 deletions(-) diff --git a/src/server/management/usage-aggregate-cache.ts b/src/server/management/usage-aggregate-cache.ts index b54ed7d9685..66b2f845a2c 100644 --- a/src/server/management/usage-aggregate-cache.ts +++ b/src/server/management/usage-aggregate-cache.ts @@ -110,7 +110,6 @@ const MAX_REBUILD_ATTEMPTS = 2; const MAX_FILTER_INDEX_ATTEMPTS = 3; // Selected rows are folded in bounded batches, so a filter that matches most of a large ledger // costs the same peak memory as the streaming scan it replaces. -const FILTER_ROW_READ_BATCH = 2_000; const MAX_RETAINED_FILTERED_AGGREGATES = 4; let retainedAggregate: RetainedUsageAggregate | null = null; @@ -396,7 +395,7 @@ function filterStats( ): UsageFilterQueryStats { return { totalRows: read.snapshot.indexedRows, - candidateRows: read.offsets.length, + candidateRows: read.candidateRows, scope: update, ingestedRows: read.snapshot.ingestedRows, ingestPasses: read.snapshot.ingestPasses, @@ -470,28 +469,36 @@ async function accumulateCandidates( read: UsageFilterCandidateRead, filter: NormalizedUsageFilter, ): Promise { - // Always read at least once, even with no candidates: the reader validates the snapshot - // against the ledger, and an empty selection must not skip that check. - const batchCount = Math.max(1, Math.ceil(read.offsets.length / FILTER_ROW_READ_BATCH)); - for (let batch = 0; batch < batchCount; batch += 1) { - const start = batch * FILTER_ROW_READ_BATCH; - const rows = await readUsageLedgerRowsAtOffsets({ - offsets: read.offsets.slice(start, start + FILTER_ROW_READ_BATCH), - expectedIdentityKey: read.snapshot.identityKey, - // The offsets belong to this index snapshot. If the ledger moved since the snapshot, - // the reader re-validates the boundary before trusting any of them, and a rewrite of - // the indexed tail becomes a rebuild instead of a silently different row set. - expectedRevisionKey: usageLogRevisionKey(read.snapshot.revision), - expectedProcessedThroughBytes: read.snapshot.indexedOffset, - expectedProcessedThroughDigest: read.snapshot.processedThroughDigest, - // The physical size at publish time; a torn suffix makes it larger than the boundary. - expectedFileSize: read.snapshot.revision?.size, - }); - if (rows.unreadableOffsets.length > 0) { - // Serving only the rows that happened to parse would silently under-report the filter. - throw new UsageFilterIndexCorruptError(); + try { + let first = true; + for (;;) { + // Candidate selection and row decoding are synchronous bounded chunks. A timer yield + // between them prevents a broad authenticated query from monopolizing the event loop. + if (!first) await new Promise(resolve => setTimeout(resolve, 0)); + const offsets = read.nextPage(); + if (!first && offsets.length === 0) break; + first = false; + const rows = await readUsageLedgerRowsAtOffsets({ + offsets, + expectedIdentityKey: read.snapshot.identityKey, + // The offsets belong to this index snapshot. If the ledger moved since the snapshot, + // the reader re-validates the boundary before trusting any of them, and a rewrite of + // the indexed tail becomes a rebuild instead of a silently different row set. + expectedRevisionKey: usageLogRevisionKey(read.snapshot.revision), + expectedProcessedThroughBytes: read.snapshot.indexedOffset, + expectedProcessedThroughDigest: read.snapshot.processedThroughDigest, + // The physical size at publish time; a torn suffix makes it larger than the boundary. + expectedFileSize: read.snapshot.revision?.size, + }); + if (rows.unreadableOffsets.length > 0) { + // Serving only the rows that happened to parse would silently under-report the filter. + throw new UsageFilterIndexCorruptError(); + } + for (const entry of rows.entries) accumulator.add(entry); + if (offsets.length === 0) break; } - for (const entry of rows.entries) accumulator.add(entry); + } finally { + read.close(); } // Without a source filter the reported span covers the whole ledger, including rows this // filter never selected. A source filter derives its span from the rows that match it, so diff --git a/src/usage/filter-index.ts b/src/usage/filter-index.ts index 30db3c6e78b..b780c2798e4 100644 --- a/src/usage/filter-index.ts +++ b/src/usage/filter-index.ts @@ -89,7 +89,7 @@ const MAX_REFRESH_ATTEMPTS = 3; const SQLITE_BUSY_TIMEOUT_MS = 250; const BATCH_BUSY_RETRIES = 3; const BATCH_BUSY_RETRY_MS = 5; -const BUSY_RETRY_DELAYS_MS = [10, 25, 50] as const; +const BUSY_RETRY_DELAYS_MS = [10, 25, 50, 100, 200, 400, 800, 1_600, 3_200] as const; const SCHEMA_DDL = ` CREATE TABLE IF NOT EXISTS ${META_TABLE} ( @@ -218,8 +218,12 @@ export interface UsageFilterIndexQuery { export interface UsageFilterCandidateRead { snapshot: UsageFilterIndexSnapshot; - /** Ascending row-start offsets selected by the index. */ - offsets: number[]; + /** Number of candidates returned by pages consumed so far. */ + candidateRows: number; + /** Returns the next bounded page of ascending row-start offsets. */ + nextPage(): number[]; + /** Releases the read snapshot held while pages are consumed. */ + close(): void; } /** @@ -1171,14 +1175,41 @@ function rowAccessIndexWithoutClient(query: UsageFilterIndexQuery): string | nul return null; } -function selectCandidates( +const CANDIDATE_PAGE_SIZE = 2_000; + +function candidatePager( db: Database, snapshot: UsageFilterIndexSnapshot, query: UsageFilterIndexQuery, -): number[] { +): Pick { const { sql, values } = buildCandidateSelect(query, snapshot); - const rows = db.query(sql).all(...values) as Array<{ byteOffset: number }>; - return rows.map(row => row.byteOffset); + const statement = db.query( + `SELECT byteOffset FROM (${sql}) WHERE byteOffset > ? ORDER BY byteOffset LIMIT ?`, + ); + let cursor = -1; + const pager = { + candidateRows: 0, + nextPage(): number[] { + db.exec("BEGIN"); + try { + const meta = readMeta(db); + if (meta.schemaVersion !== USAGE_FILTER_INDEX_SCHEMA_VERSION + || meta.generation !== snapshot.generation) { + throw new UsageLedgerRebuildRequiredError("content_changed"); + } + const rows = statement.all(...values, cursor, CANDIDATE_PAGE_SIZE) as Array<{ byteOffset: number }>; + const offsets = rows.map(row => row.byteOffset); + db.exec("COMMIT"); + if (offsets.length > 0) cursor = offsets[offsets.length - 1]!; + pager.candidateRows += offsets.length; + return offsets; + } catch (error) { + try { db.exec("ROLLBACK"); } catch { /* nothing to roll back */ } + throw error; + } + }, + }; + return pager; } /** A published generation with a nonzero row count must actually hold rows. */ @@ -1208,22 +1239,28 @@ async function readActive( ingestedRows: refreshed.ingestedRows, ingestPasses: refreshed.ingestPasses, }; - const offsets = selectCandidates(db, snapshot, query); - if (offsets.length === 0 && meta.indexedRows > 0 && !hasPublishedRows(db, meta.generation)) { + if (meta.indexedRows > 0 && !hasPublishedRows(db, meta.generation)) { // The published generation claims rows it no longer has. Never serve that as a // complete zero; drop the projection and rebuild. The refresh alone cannot see this // (the cursor and identity still match), so the rebuild must be requested explicitly. requestUsageFilterIndexRebuild(); throw new UsageLedgerRebuildRequiredError("content_changed"); } + const pager = candidatePager(db, snapshot, query); db.exec("COMMIT"); - return { snapshot, offsets }; + return { + snapshot, + get candidateRows() { return pager.candidateRows; }, + nextPage: () => pager.nextPage(), + close: () => db.close(), + }; } catch (error) { try { db.exec("ROLLBACK"); } catch { /* nothing to roll back */ } throw error; } - } finally { + } catch (error) { db.close(); + throw error; } } @@ -1295,16 +1332,21 @@ export async function readUsageFilterCandidatesPassive( || !tables.has("usage_rows") || !tables.has("usage_attributions") || !tables.has("usage_filter_index_writers")) { - return null; + throw new UsageFilterIndexForeignError(); } db.exec("BEGIN"); try { const meta = readMeta(db); const revision = currentUsageLogRevision(); - if (meta.schemaVersion !== USAGE_FILTER_INDEX_SCHEMA_VERSION || !meta.generation) return null; - if (usageLogIdentityKey(meta.revision) !== usageLogIdentityKey(revision)) return null; - if (!revision || meta.indexedOffset !== revision.size) return null; - if (usageLogRevisionKey(meta.revision) !== usageLogRevisionKey(revision)) return null; + if (meta.schemaVersion !== USAGE_FILTER_INDEX_SCHEMA_VERSION || !meta.generation) { + throw new UsageLedgerRebuildRequiredError("content_changed"); + } + if (usageLogIdentityKey(meta.revision) !== usageLogIdentityKey(revision) + || !revision + || meta.indexedOffset !== revision.size + || usageLogRevisionKey(meta.revision) !== usageLogRevisionKey(revision)) { + throw new UsageLedgerRebuildRequiredError("content_changed"); + } const snapshot = snapshotFrom(meta, 0, 0); // The exact revision match above already proves the ledger is the one indexed; the // empty read keeps the passive path on the same validation as the active one. @@ -1316,20 +1358,27 @@ export async function readUsageFilterCandidatesPassive( expectedProcessedThroughDigest: snapshot.processedThroughDigest, expectedFileSize: snapshot.revision?.size, }); - const offsets = selectCandidates(db, snapshot, query); - if (offsets.length === 0 && meta.indexedRows > 0 && !hasPublishedRows(db, meta.generation)) { + if (meta.indexedRows > 0 && !hasPublishedRows(db, meta.generation)) { // Same guard as the active read: a published generation that lost its rows must not // be reported as a complete zero. - return null; + throw new UsageLedgerRebuildRequiredError("content_changed"); } - return { snapshot, offsets }; - } finally { - try { db.exec("COMMIT"); } catch { /* read-only transaction */ } + const pager = candidatePager(db, snapshot, query); + db.exec("COMMIT"); + return { + snapshot, + get candidateRows() { return pager.candidateRows; }, + nextPage: () => pager.nextPage(), + close: () => db.close(), + }; + } catch (error) { + try { db.exec("ROLLBACK"); } catch { /* read-only transaction */ } + throw error; } } catch { - return null; - } finally { + try { db.exec("ROLLBACK"); } catch { /* transaction may not have started */ } db.close(); + return null; } } diff --git a/structure/usage-client-attribution.md b/structure/usage-client-attribution.md index 9a7ca1a5b8f..5b70fab4bb1 100644 --- a/structure/usage-client-attribution.md +++ b/structure/usage-client-attribution.md @@ -39,7 +39,10 @@ match. `src/usage/ledger-scanner.ts` supplies both halves: `scanUsageLedgerCoope row offset during ingest, and `readUsageLedgerRowsAtOffsets` re-reads only the selected rows under the same identity, row-ceiling and torn-suffix rules. The first filtered query after the index is missing or invalid pays one cooperative full ingest; appends ingest only new bytes, and every later filter -reuses the projection. A replaced, truncated or same-size rewritten ledger rebuilds it, a write +reuses the projection. Candidate offsets are selected and consumed in bounded keyset pages, with a +timer yield between pages; each page rechecks the published generation before it is read, so broad +filters neither retain the complete offset set nor monopolize the event loop. A replaced, truncated +or same-size rewritten ledger rebuilds it, a write observed during a row read re-verifies the selected bytes before publishing them, and a recorded offset that no longer frames a complete row drops the projection instead of serving a partial answer. `usage.jsonl` stays authoritative and is never truncated or rewritten. diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 3272cbd3d51..577f7127625 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -210,6 +210,33 @@ describe("filtered usage candidate index", () => { expect(summary(result).summary.requests).toBe(inside); }); + test("a broad filter consumes bounded candidate pages with timer yields", async () => { + writeRows(Array.from({ length: SYNTHETIC_ROWS }, (_, index) => syntheticRow(index))); + await getFilteredUsageAggregate({ provider: "kimi" }); + resetUsageAggregateCacheForTests(); + + const originalRead = ledgerScannerModule.readUsageLedgerRowsAtOffsets; + const pageSizes: number[] = []; + let yieldedBetweenPages = false; + let timerFired = false; + const readSpy = spyOn(ledgerScannerModule, "readUsageLedgerRowsAtOffsets") + .mockImplementation(async options => { + if (pageSizes.length > 0) yieldedBetweenPages ||= timerFired; + pageSizes.push(options.offsets.length); + const result = await originalRead(options); + if (pageSizes.length === 1) setTimeout(() => { timerFired = true; }, 0); + return result; + }); + try { + const result = await getFilteredUsageAggregate({}); + expect(result.filterIndex?.candidateRows).toBe(SYNTHETIC_ROWS); + expect(pageSizes).toEqual([2_000, 2_000]); + expect(yieldedBetweenPages).toBe(true); + } finally { + readSpy.mockRestore(); + } + }); + test("an append ingests only new bytes and a retained filter folds only new matches", async () => { const rows = Array.from({ length: SYNTHETIC_ROWS }, (_, index) => syntheticRow(index)); writeRows(rows); From 996ebf9638e332d600cefe945960f6f93ae879ca Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 05:29:40 +0000 Subject: [PATCH 02/13] fix(usage): release filtered readers and fall back on page errors - Close the candidate reader on the unchanged and generation-mismatch branches of refreshFilteredAggregate; only the paging paths closed it, so repeated unchanged reads leaked one SQLite connection each. - Wrap index-side candidate-page failures in UsageFilterIndexPageError and answer from the streaming ledger scan instead of rejecting the request; generation changes keep the rebuild retry. - Give the write-contention degradation test a 20s timeout: the extended busy-retry window takes several seconds by design. Co-Authored-By: Epinephrine --- .../management/usage-aggregate-cache.ts | 104 ++++++++++-------- src/usage/filter-index.ts | 18 ++- tests/usage/usage-filter-index.test.ts | 70 ++++++++++++ 3 files changed, 147 insertions(+), 45 deletions(-) diff --git a/src/server/management/usage-aggregate-cache.ts b/src/server/management/usage-aggregate-cache.ts index 66b2f845a2c..4109ddc4907 100644 --- a/src/server/management/usage-aggregate-cache.ts +++ b/src/server/management/usage-aggregate-cache.ts @@ -15,6 +15,7 @@ import { readUsageFilterCandidatesPassive, requestUsageFilterIndexRebuild, resetUsageFilterIndexForTests, + UsageFilterIndexPageError, type UsageFilterCandidateRead, type UsageFilterIndexQuery, type UsageFilterIndexSnapshot, @@ -638,52 +639,69 @@ async function refreshFilteredAggregate( filter: NormalizedUsageFilter, window?: UsageTimeWindow, ): Promise { - return withFilterIndexRetry(async () => { - const overlayVersion = userCostOverlayVersion(); - const timeZone = currentTimeZone(); - const state = retainedFilteredAggregates.get(key); - if (!state - || state.overlayVersion !== overlayVersion - || state.timeZone !== timeZone - || state.indexGeneration === null) { - return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone); - } - const read = await readCandidates({ - ...filterQuery(filter, window), - afterOffset: state.processedThroughBytes, + try { + return await withFilterIndexRetry(async () => { + const overlayVersion = userCostOverlayVersion(); + const timeZone = currentTimeZone(); + const state = retainedFilteredAggregates.get(key); + if (!state + || state.overlayVersion !== overlayVersion + || state.timeZone !== timeZone + || state.indexGeneration === null) { + return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone); + } + const read = await readCandidates({ + ...filterQuery(filter, window), + afterOffset: state.processedThroughBytes, + }); + if (!read) { + retainedFilteredAggregates.delete(key); + return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone); + } + if (state.indexGeneration !== read.snapshot.generation + || state.identityKey !== read.snapshot.identityKey + || state.processedThroughBytes > read.snapshot.indexedOffset) { + // This branch never pages, so it owns closing the reader before the rebuild discards it. + read.close(); + retainedFilteredAggregates.delete(key); + return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone); + } + if (state.processedThroughBytes === read.snapshot.indexedOffset) { + // Even an unchanged retained projection validates the ledger against the snapshot, so a + // rewrite in the window between the index refresh and this return is not served stale. + try { + await readUsageLedgerRowsAtOffsets({ + offsets: [], + expectedIdentityKey: read.snapshot.identityKey, + expectedRevisionKey: usageLogRevisionKey(read.snapshot.revision), + expectedProcessedThroughBytes: read.snapshot.indexedOffset, + expectedProcessedThroughDigest: read.snapshot.processedThroughDigest, + expectedFileSize: read.snapshot.revision?.size, + }); + } finally { + // This branch never pages, so it owns closing the reader before publishing its stats. + read.close(); + } + state.retainedAt = Date.now(); + // Whole-ledger diagnostics can move without the boundary moving: a torn oversized or + // malformed suffix is observed by every append scan and never advances the offset. + state.usageIncomplete = read.snapshot.oversizedRows > 0; + state.usageInvalidRows = read.snapshot.invalidRows > 0; + state.revision = read.snapshot.revision; + state.revisionKey = usageLogRevisionKey(read.snapshot.revision); + return publishFilteredAggregate(key, state, "unchanged", read); + } + return appendFilteredAggregate(key, state, filter, read, overlayVersion, timeZone); }); - if (!read) { - retainedFilteredAggregates.delete(key); - return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone); - } - if (state.indexGeneration !== read.snapshot.generation - || state.identityKey !== read.snapshot.identityKey - || state.processedThroughBytes > read.snapshot.indexedOffset) { + } catch (error) { + if (error instanceof UsageFilterIndexPageError) { + // A page failed after its reader was handed out: the index cannot serve this query, + // so answer from the authoritative ledger scan instead of failing the request. retainedFilteredAggregates.delete(key); - return rebuildFilteredAggregate(key, filter, window, overlayVersion, timeZone); + return rebuildFilteredAggregateStreaming(key, filter, window, userCostOverlayVersion(), currentTimeZone()); } - if (state.processedThroughBytes === read.snapshot.indexedOffset) { - // Even an unchanged retained projection validates the ledger against the snapshot, so a - // rewrite in the window between the index refresh and this return is not served stale. - await readUsageLedgerRowsAtOffsets({ - offsets: [], - expectedIdentityKey: read.snapshot.identityKey, - expectedRevisionKey: usageLogRevisionKey(read.snapshot.revision), - expectedProcessedThroughBytes: read.snapshot.indexedOffset, - expectedProcessedThroughDigest: read.snapshot.processedThroughDigest, - expectedFileSize: read.snapshot.revision?.size, - }); - state.retainedAt = Date.now(); - // Whole-ledger diagnostics can move without the boundary moving: a torn oversized or - // malformed suffix is observed by every append scan and never advances the offset. - state.usageIncomplete = read.snapshot.oversizedRows > 0; - state.usageInvalidRows = read.snapshot.invalidRows > 0; - state.revision = read.snapshot.revision; - state.revisionKey = usageLogRevisionKey(read.snapshot.revision); - return publishFilteredAggregate(key, state, "unchanged", read); - } - return appendFilteredAggregate(key, state, filter, read, overlayVersion, timeZone); - }); + throw error; + } } function jevStatsResultFrom( diff --git a/src/usage/filter-index.ts b/src/usage/filter-index.ts index b780c2798e4..1b9da27580d 100644 --- a/src/usage/filter-index.ts +++ b/src/usage/filter-index.ts @@ -182,6 +182,17 @@ export class UsageFilterIndexHomeChangedError extends Error { } } +/** + * A candidate page could not be read from the index after the reader was handed out. The + * projection cannot serve this query, so the caller streams the ledger instead of failing. + */ +export class UsageFilterIndexPageError extends Error { + constructor(readonly pageError: unknown) { + super("usage filter index candidate page could not be read"); + this.name = "UsageFilterIndexPageError"; + } +} + /** * A retained filtered projection read this generation. A full rebuild replaces every offset, * so a projection from an older generation must be discarded even when the file identity and @@ -1190,8 +1201,8 @@ function candidatePager( const pager = { candidateRows: 0, nextPage(): number[] { - db.exec("BEGIN"); try { + db.exec("BEGIN"); const meta = readMeta(db); if (meta.schemaVersion !== USAGE_FILTER_INDEX_SCHEMA_VERSION || meta.generation !== snapshot.generation) { @@ -1205,7 +1216,10 @@ function candidatePager( return offsets; } catch (error) { try { db.exec("ROLLBACK"); } catch { /* nothing to roll back */ } - throw error; + // A generation change keeps the rebuild retry; every other index-side failure means + // this projection cannot serve the page, so the caller streams the ledger instead. + if (error instanceof UsageLedgerRebuildRequiredError) throw error; + throw new UsageFilterIndexPageError(error); } }, }; diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 577f7127625..1ea94e6cab4 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -37,7 +37,9 @@ import { setUsageFilterIndexTestHooksForTests, usageFilterCandidatePlan, usageFilterIndexPath, + UsageFilterIndexPageError, } from "../../src/usage/filter-index"; +import * as filterIndexModule from "../../src/usage/filter-index"; import { currentUsageLogRevision, resetUsageReadCacheForTests, @@ -1155,6 +1157,74 @@ describe("filtered usage index availability", () => { lock.exec("ROLLBACK"); lock.close(); } + // Sustained contention exhausts the extended busy-retry window before degrading to the + // streaming read, which takes several seconds by design; the default 5s timeout is too tight. + }, 20_000); + + test("an unchanged filtered read releases its index reader", async () => { + const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); + writeRows(rows); + await getFilteredUsageAggregate({ provider: "openai" }); + + const realRead = filterIndexModule.readUsageFilterCandidates; + let closeCalls = 0; + const spy = spyOn(filterIndexModule, "readUsageFilterCandidates").mockImplementation(async (query) => { + const read = await realRead(query); + if (read) { + const originalClose = read.close; + read.close = () => { + closeCalls += 1; + originalClose(); + }; + } + return read; + }); + try { + // The retained projection is unchanged, so this exercises the branch that validates the + // ledger without paging and must still release the reader it opened. + const result = await getFilteredUsageAggregate({ provider: "openai" }); + expect(result.update).toBe("unchanged"); + expect(closeCalls).toBe(1); + } finally { + spy.mockRestore(); + } + }); + + test("a candidate page failure falls back to the streaming read", async () => { + const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); + writeRows(rows); + await getFilteredUsageAggregate({ provider: "openai" }); + resetUsageAggregateCacheForTests(); + + const expected = rows.filter(row => row.provider === "openai").length; + const spy = spyOn(filterIndexModule, "readUsageFilterCandidates").mockImplementation(async () => ({ + snapshot: { + revision: null, + identityKey: "", + generation: "", + indexedOffset: 0, + processedThroughDigest: "", + indexedRows: 0, + invalidRows: 0, + oversizedRows: 0, + minTimestamp: null, + maxTimestamp: null, + ingestPasses: 0, + ingestedRows: 0, + }, + candidateRows: 0, + nextPage() { + throw new UsageFilterIndexPageError(new Error("SQLITE_CORRUPT")); + }, + close() {}, + })); + try { + const result = await getFilteredUsageAggregate({ provider: "openai" }); + expect(summary(result).summary.requests).toBe(expected); + expect(result.filterIndex).toBeUndefined(); + } finally { + spy.mockRestore(); + } }); test("a projection that vanishes before the read is not recreated", async () => { From 018e0c43c77956bd5ab94f60dfda5b6639c84090 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 06:12:39 +0000 Subject: [PATCH 03/13] test(usage): give the two-process rebuild contention test room The extended busy-retry backoff plus a 30k-row double scan pushes this test past the default 5s timeout on CI (observed 5002ms). Co-Authored-By: Epinephrine --- tests/usage/usage-filter-index.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 1ea94e6cab4..96cac4e8ac1 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -648,7 +648,9 @@ console.log(JSON.stringify({ requests: summary.summary.requests, candidates: res } expect(indexGenerations()).toHaveLength(1); expect(indexRowCount(publishedGeneration())).toBe(rows.length); - }); + // Two processes contend on one 30k-row rebuild; the loser's busy-retry backoff plus the + // double scan legitimately exceeds the default 5s timeout. + }, 30_000); test("crashed and abandoned staging never accumulates", async () => { const rows = Array.from({ length: 3000 }, (_, index) => syntheticRow(index)); From 63d2a21f1c2791a0db180a9c869f84793f7ca531 Mon Sep 17 00:00:00 2001 From: Epinephrine Date: Mon, 9 Nov 2026 15:31:21 +0900 Subject: [PATCH 04/13] fix(usage): retry streaming fallback after index page failures Keep bounded retry handling for ledger and aggregation-input changes after the index pager becomes unavailable. Cover ledger replacement, overlay changes, and real SQLite pager rollback/error wrapping in active and passive reads. --- .../management/usage-aggregate-cache.ts | 4 +- structure/usage-client-attribution.md | 4 + tests/usage/usage-filter-index.test.ts | 122 ++++++++++++++++++ 3 files changed, 129 insertions(+), 1 deletion(-) diff --git a/src/server/management/usage-aggregate-cache.ts b/src/server/management/usage-aggregate-cache.ts index 4109ddc4907..2ecec670630 100644 --- a/src/server/management/usage-aggregate-cache.ts +++ b/src/server/management/usage-aggregate-cache.ts @@ -698,7 +698,9 @@ async function refreshFilteredAggregate( // A page failed after its reader was handed out: the index cannot serve this query, // so answer from the authoritative ledger scan instead of failing the request. retainedFilteredAggregates.delete(key); - return rebuildFilteredAggregateStreaming(key, filter, window, userCostOverlayVersion(), currentTimeZone()); + return withFilterIndexRetry(() => + rebuildFilteredAggregateStreaming(key, filter, window, userCostOverlayVersion(), currentTimeZone()), + ); } throw error; } diff --git a/structure/usage-client-attribution.md b/structure/usage-client-attribution.md index 5b70fab4bb1..fb83ecd2231 100644 --- a/structure/usage-client-attribution.md +++ b/structure/usage-client-attribution.md @@ -138,3 +138,7 @@ filtered HEAD reads, whose empty body cannot confirm the scope. Requests without newly scoped by that rule. `src/server/hub-usage.ts` binds a client-filtered data-plane read to its authenticated key and emits a safe scope echo; `src/remote/hub-usage.ts` projects it without key IDs, and `src/client/hub-client.ts` rejects a missing or mismatched echo. + +Page-time index failure fallback retries the authoritative streaming scan within the same bounded +retry policy for concurrent ledger replacement and aggregation-input changes. It does not retry the +unusable index, and unrelated ledger I/O failures still propagate. diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 96cac4e8ac1..80127c666b7 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -1229,6 +1229,128 @@ describe("filtered usage index availability", () => { } }); + test("page fallback retries a concurrent ledger replacement", async () => { + const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); + writeRows(rows); + await getFilteredUsageAggregate({ provider: "openai" }); + resetUsageAggregateCacheForTests(); + + const expected = rows.filter(row => row.provider === "openai").length; + const spy = spyOn(filterIndexModule, "readUsageFilterCandidates").mockImplementation(async () => ({ + snapshot: { + revision: null, + identityKey: "", + generation: "", + indexedOffset: 0, + processedThroughDigest: "", + indexedRows: 0, + invalidRows: 0, + oversizedRows: 0, + minTimestamp: null, + maxTimestamp: null, + ingestPasses: 0, + ingestedRows: 0, + }, + candidateRows: 0, + nextPage() { + throw new UsageFilterIndexPageError(new Error("SQLITE_CORRUPT")); + }, + close() {}, + })); + const realScan = ledgerScannerModule.scanUsageLedgerCooperatively; + let scans = 0; + const scanSpy = spyOn(ledgerScannerModule, "scanUsageLedgerCooperatively").mockImplementation(async options => { + if (++scans === 1) throw new ledgerScannerModule.UsageLedgerRebuildRequiredError("content_changed"); + return realScan(options); + }); + try { + const result = await getFilteredUsageAggregate({ provider: "openai" }); + expect(summary(result).summary.requests).toBe(expected); + expect(result.filterIndex).toBeUndefined(); + expect(scans).toBe(2); + } finally { + spy.mockRestore(); + scanSpy.mockRestore(); + } + }); + + test("page fallback retries an overlay change", async () => { + const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); + writeRows(rows); + await getFilteredUsageAggregate({ provider: "openai" }); + resetUsageAggregateCacheForTests(); + + const expected = rows.filter(row => row.provider === "openai").length; + const spy = spyOn(filterIndexModule, "readUsageFilterCandidates").mockImplementation(async () => ({ + snapshot: { + revision: null, + identityKey: "", + generation: "", + indexedOffset: 0, + processedThroughDigest: "", + indexedRows: 0, + invalidRows: 0, + oversizedRows: 0, + minTimestamp: null, + maxTimestamp: null, + ingestPasses: 0, + ingestedRows: 0, + }, + candidateRows: 0, + nextPage() { + throw new UsageFilterIndexPageError(new Error("SQLITE_CORRUPT")); + }, + close() {}, + })); + const realScan = ledgerScannerModule.scanUsageLedgerCooperatively; + let scans = 0; + const scanSpy = spyOn(ledgerScannerModule, "scanUsageLedgerCooperatively").mockImplementation(async options => { + const result = await realScan(options); + if (++scans === 1) refreshUserCostOverlays({ providers: { openai: {} } } as unknown as OcxConfig); + return result; + }); + try { + const result = await getFilteredUsageAggregate({ provider: "openai" }); + expect(summary(result).summary.requests).toBe(expected); + expect(result.filterIndex).toBeUndefined(); + expect(scans).toBe(2); + } finally { + spy.mockRestore(); + scanSpy.mockRestore(); + } + }); + + for (const passive of [false, true]) { + test(`real SQLite page failure rolls back and wraps the error (passive=${passive})`, async () => { + writeRows(Array.from({ length: 5 }, (_, i) => syntheticRow(i))); + await getFilteredUsageAggregate({ provider: "openai" }); + const read = await (passive + ? filterIndexModule.readUsageFilterCandidatesPassive({}) + : filterIndexModule.readUsageFilterCandidates({})); + expect(read).not.toBeNull(); + const writer = new Database(usageFilterIndexPath(testDir)); + writer.exec("DROP TABLE usage_rows"); + writer.close(); + const realExec = Database.prototype.exec; + let rolledBack = false; + const execSpy = spyOn(Database.prototype, "exec").mockImplementation(function (this: Database, sql: string) { + const result = realExec.call(this, sql); + if (sql === "ROLLBACK") rolledBack = !this.inTransaction; + return result; + }); + try { + let caught: unknown; + try { read!.nextPage(); } catch (error) { caught = error; } + expect(caught).toBeInstanceOf(UsageFilterIndexPageError); + expect(String((caught as UsageFilterIndexPageError).pageError)).toContain("no such table"); + expect(rolledBack).toBe(true); + } finally { + execSpy.mockRestore(); + read!.close(); + } + }); + } + test("a projection that vanishes before the read is not recreated", async () => { const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); writeRows(rows); From 5c4c2deb7939feb4d48fc179aae461b4f717a634 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 06:33:08 +0000 Subject: [PATCH 05/13] test(usage): drop stray patch markers from filter-index tests Leftover '+' diff-application artifacts between test blocks; they parse as unary-plus expressions but carry no meaning. Co-Authored-By: Epinephrine --- tests/usage/usage-filter-index.test.ts | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 80127c666b7..5538b60661d 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -764,7 +764,7 @@ console.log(JSON.stringify({ requests: summary.summary.requests, candidates: res expect(read.entries).toHaveLength(1); expect(read.unreadableOffsets).toHaveLength(49); }); -+ + test("a rewritten indexed row makes the next append rebuild instead of streaming forever", async () => { const rows = Array.from({ length: 200 }, (_, index) => syntheticRow(index)); writeRows(rows); @@ -811,7 +811,7 @@ console.log(JSON.stringify({ requests: summary.summary.requests, candidates: res setUsageFilterIndexTestHooksForTests(null); } }); -+ + test("a same-size rewrite before the trailing boundary forces a rebuild", async () => { // Large enough that row zero sits far outside the trailing 64 KiB digest window. const rows = Array.from({ length: 4000 }, (_, index) => syntheticRow(index)); @@ -897,7 +897,7 @@ console.log(JSON.stringify({ requests: summary.summary.requests, candidates: res for (const seen of observed) expect(seen).toBe("none/none"); expect(totalIndexRows()).toBe(ledgerRowCount()); }); -+ + test("a replaced ledger between the index refresh and the read is caught", async () => { const rows = Array.from({ length: 20 }, (_, index) => syntheticRow(index)); writeRows(rows); @@ -984,7 +984,7 @@ console.log(JSON.stringify({ requests: summary.summary.requests, candidates: res probe.close(); } }); -+ + test("an unchanged retained projection still validates the ledger", async () => { const rows = Array.from({ length: 40 }, (_, index) => syntheticRow(index)); writeRows(rows); @@ -1008,7 +1008,7 @@ console.log(JSON.stringify({ requests: summary.summary.requests, candidates: res setUsageFilterIndexTestHooksForTests(null); } }); -+ + test("a torn suffix does not let a same-size rewrite masquerade as an append", async () => { const rows = Array.from({ length: 1000 }, (_, index) => ( index === 0 ? { ...syntheticRow(index), provider: "kimi" } : syntheticRow(index) @@ -1398,7 +1398,7 @@ describe("filtered usage index availability", () => { // The original home's projection was not advanced with another home's ledger. expect(publishedOffset()).toBe(offsetBefore); }); -+ + test("a passive read does not report a published generation that lost its rows", async () => { const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); writeRows(rows); @@ -1415,7 +1415,7 @@ describe("filtered usage index availability", () => { expect(summary(passive).summary.requests) .toBe(rows.filter(row => row.provider === "openai").length); }); -+ + test("a foreign SQLite file with a coincidental meta table is not adopted", async () => { const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); writeRows(rows); From 69869dc676bdd754925348a25b896fe22e590686 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 08:49:27 +0000 Subject: [PATCH 06/13] fix(usage): heal corrupt index pages and split the rebuild retry schedule - nextPage() requests an index rebuild when a page fails with SQLITE_CORRUPT/NOTADB, so a damaged projection heals instead of degrading every query to streaming\n- UsageLedgerRebuildRequiredError retries on the original short schedule; the 9-delay backoff stays for SQLITE_BUSY contention\n- document per-page transactions on close(), name the fallback retry contract in the structure doc, and fix the stale one-transaction claim\n- cover the per-page generation recheck against real SQLite with a rebuild between pages Co-Authored-By: Epinephrine --- src/usage/filter-index.ts | 13 +++++++++-- structure/usage-client-attribution.md | 20 ++++++++++------- tests/usage/usage-filter-index.test.ts | 30 ++++++++++++++++++++++++++ 3 files changed, 53 insertions(+), 10 deletions(-) diff --git a/src/usage/filter-index.ts b/src/usage/filter-index.ts index 1b9da27580d..85cd42c3586 100644 --- a/src/usage/filter-index.ts +++ b/src/usage/filter-index.ts @@ -90,6 +90,9 @@ const SQLITE_BUSY_TIMEOUT_MS = 250; const BATCH_BUSY_RETRIES = 3; const BATCH_BUSY_RETRY_MS = 5; const BUSY_RETRY_DELAYS_MS = [10, 25, 50, 100, 200, 400, 800, 1_600, 3_200] as const; +// A rebuild-required error resolves as soon as the competing publish settles, so it retries on +// the original short schedule instead of parking behind the long busy backoff. +const REBUILD_RETRY_DELAYS_MS = [10, 25, 50] as const; const SCHEMA_DDL = ` CREATE TABLE IF NOT EXISTS ${META_TABLE} ( @@ -233,7 +236,10 @@ export interface UsageFilterCandidateRead { candidateRows: number; /** Returns the next bounded page of ascending row-start offsets. */ nextPage(): number[]; - /** Releases the read snapshot held while pages are consumed. */ + /** + * Closes the database handle. Pages are not read from one long snapshot: each `nextPage()` + * re-checks schema and generation inside its own short transaction. + */ close(): void; } @@ -1188,6 +1194,7 @@ function rowAccessIndexWithoutClient(query: UsageFilterIndexQuery): string | nul const CANDIDATE_PAGE_SIZE = 2_000; +/** Keyset pager over the candidate select: bounded pages, each validated in its own transaction. */ function candidatePager( db: Database, snapshot: UsageFilterIndexSnapshot, @@ -1219,6 +1226,8 @@ function candidatePager( // A generation change keeps the rebuild retry; every other index-side failure means // this projection cannot serve the page, so the caller streams the ledger instead. if (error instanceof UsageLedgerRebuildRequiredError) throw error; + // Paging cannot repair a damaged projection; ask the next active refresh to rebuild it. + if (isCorruptError(error)) requestUsageFilterIndexRebuild(); throw new UsageFilterIndexPageError(error); } }, @@ -1304,7 +1313,7 @@ export async function readUsageFilterCandidates( return null; } if (error instanceof UsageLedgerRebuildRequiredError) { - const delay = BUSY_RETRY_DELAYS_MS[attempt]; + const delay = REBUILD_RETRY_DELAYS_MS[attempt]; if (delay === undefined) return null; await new Promise(resolve => setTimeout(resolve, delay)); continue; diff --git a/structure/usage-client-attribution.md b/structure/usage-client-attribution.md index fb83ecd2231..5bb5e8bdeae 100644 --- a/structure/usage-client-attribution.md +++ b/structure/usage-client-attribution.md @@ -71,8 +71,9 @@ writer's own publish fails the fence and retries, and a reader only ever sees on generation. A crash before publish leaves the previously published generation complete. An append writes into the published generation and publishes its cursor, digest, counts and span in the same transaction, -fenced on the cursor so a stale append cannot regress it. Readers take meta and the candidate -select in one read transaction and reject a generation that changed under them. A published +fenced on the cursor so a stale append cannot regress it. Each candidate page re-reads meta and +selects inside its own short read transaction, and a generation or schema change aborts the read -- +the per-page recheck described with the paging contract above. A published generation that claims rows it no longer holds is rebuilt rather than served as a complete zero, in both the active and the passive read. @@ -93,10 +94,17 @@ sidecars are registered for uninstall only when this module creates them, and a that is not this projection is never modified. A passive management read opens an existing projection read-only, requires this projection's full table set rather than a coincidental meta table, and only uses it when it is current for the exact ledger revision; otherwise it streams, -and it never creates, writes or registers anything. Any other index-side failure -- unwritable home, +and it never creates, writes or registers anything. Index reads retry contention on nine delays +totaling about 6.4 s and a rebuild-required read on three short delays before degrading. Any other +index-side failure -- unwritable home, full disk, contention, a corrupt or foreign file -- falls back to the authoritative streaming scan instead of failing a request the scan can answer; a failure to read `usage.jsonl` itself is -propagated. A nonselective filter (`client=all`, `device=all`, no provider/model/key) still reads +propagated. A `UsageFilterIndexPageError` from a `src/usage/filter-index.ts` page read clears the +retained filtered aggregate, then retries the authoritative streaming scan under +`withFilterIndexRetry` (`MAX_FILTER_INDEX_ATTEMPTS` attempts) in +`src/server/management/usage-aggregate-cache.ts`: only a concurrent ledger replacement or an +aggregation-input change is retried, the unusable index is never re-entered, and unrelated ledger +I/O failures still propagate. A nonselective filter (`client=all`, `device=all`, no provider/model/key) still reads every row in its window; `candidateRows` counts the rows selected and normalized for that call and `scope` says whether that is the whole candidate set or only the appended suffix, not how many SQL index entries were examined. `tests/usage/usage-filter-index.test.ts` asserts candidate counts against the whole @@ -138,7 +146,3 @@ filtered HEAD reads, whose empty body cannot confirm the scope. Requests without newly scoped by that rule. `src/server/hub-usage.ts` binds a client-filtered data-plane read to its authenticated key and emits a safe scope echo; `src/remote/hub-usage.ts` projects it without key IDs, and `src/client/hub-client.ts` rejects a missing or mismatched echo. - -Page-time index failure fallback retries the authoritative streaming scan within the same bounded -retry policy for concurrent ledger replacement and aggregation-input changes. It does not retry the -unusable index, and unrelated ledger I/O failures still propagate. diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 5538b60661d..79e9ad28af9 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -1351,6 +1351,36 @@ describe("filtered usage index availability", () => { }); } + test("a generation change between pages keeps the rebuild retry and correct totals", async () => { + const rows = Array.from({ length: 2_500 }, (_, index) => syntheticRow(index)); + writeRows(rows); + await getFilteredUsageAggregate({}); + + // Explicit nulls match the normalized query shape; a bare `{}` is not a valid caller query. + const read = await filterIndexModule.readUsageFilterCandidates({ + provider: null, model: null, apiKeyId: null, + }); + expect(read).not.toBeNull(); + // A full first page proves this reader has another page left to validate. + expect(read!.nextPage()).toHaveLength(2_000); + + try { + // Publish a new generation under the open reader through the normal rebuild flow. + filterIndexModule.requestUsageFilterIndexRebuild(); + await getFilteredUsageAggregate({}); + let caught: unknown; + try { read!.nextPage(); } catch (error) { caught = error; } + expect(caught).toBeInstanceOf(ledgerScannerModule.UsageLedgerRebuildRequiredError); + expect(caught).not.toBeInstanceOf(UsageFilterIndexPageError); + } finally { + read!.close(); + } + + // The retried aggregate still answers from the fresh generation, matching the oracle. + resetUsageAggregateCacheForTests(); + await expectMatchesFullScan({}); + }); + test("a projection that vanishes before the read is not recreated", async () => { const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); writeRows(rows); From a1674457626bdd2cee949bcc7c4f85411ced7397 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 08:51:06 +0000 Subject: [PATCH 07/13] test(usage): cover the mid-aggregate rebuild retry Publish a new generation between the first and second candidate pages of a live aggregate, asserting the restarted paging matches the full-scan oracle with fresh totals. Co-Authored-By: Epinephrine --- tests/usage/usage-filter-index.test.ts | 31 ++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 79e9ad28af9..66d87a19067 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -1381,6 +1381,37 @@ describe("filtered usage index availability", () => { await expectMatchesFullScan({}); }); + test("a rebuild mid-aggregate restarts paging with fresh totals", async () => { + const rows = Array.from({ length: 2_500 }, (_, index) => syntheticRow(index)); + writeRows(rows); + await getFilteredUsageAggregate({}); + resetUsageAggregateCacheForTests(); + + const realReadRows = ledgerScannerModule.readUsageLedgerRowsAtOffsets; + let rebuilt = false; + const rowsSpy = spyOn(ledgerScannerModule, "readUsageLedgerRowsAtOffsets").mockImplementation(async options => { + if (!rebuilt) { + rebuilt = true; + // Publish a new generation between the first and second candidate pages. + filterIndexModule.requestUsageFilterIndexRebuild(); + const fresh = await filterIndexModule.readUsageFilterCandidates({ + provider: null, model: null, apiKeyId: null, + }); + fresh?.close(); + } + return realReadRows(options); + }); + try { + const result = await getFilteredUsageAggregate({}); + const oracle = await fullScan({}); + expect(summary(result)).toEqual(oracle.summarize("all", NOW)); + // The retry consumed the rebuilt generation end to end, not partial old-generation rows. + expect(result.filterIndex?.candidateRows).toBe(rows.length); + } finally { + rowsSpy.mockRestore(); + } + }); + test("a projection that vanishes before the read is not recreated", async () => { const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); writeRows(rows); From 8e90aefb65924368961fb8a5f363d61303334725 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:59:52 +0000 Subject: [PATCH 08/13] docs(usage): docstring the paged read path Document candidatePager.nextPage, readActive, accumulateCandidates and refreshFilteredAggregate. Co-Authored-By: Epinephrine --- src/server/management/usage-aggregate-cache.ts | 8 ++++++++ src/usage/filter-index.ts | 8 ++++++++ 2 files changed, 16 insertions(+) diff --git a/src/server/management/usage-aggregate-cache.ts b/src/server/management/usage-aggregate-cache.ts index 2ecec670630..17e530fa866 100644 --- a/src/server/management/usage-aggregate-cache.ts +++ b/src/server/management/usage-aggregate-cache.ts @@ -465,6 +465,10 @@ async function readCandidates( : readUsageFilterCandidates(query); } +/** + * Folds every candidate page into the accumulator, yielding between pages and re-reading rows + * against the snapshot identity; closes the reader when done. + */ async function accumulateCandidates( accumulator: UsageSummaryAccumulator, read: UsageFilterCandidateRead, @@ -634,6 +638,10 @@ async function appendFilteredAggregate( } } +/** + * Serves the filtered report from the retained aggregate, appending new candidates or rebuilding + * as needed; a candidate-page failure clears the retained state and retries the streaming scan. + */ async function refreshFilteredAggregate( key: string, filter: NormalizedUsageFilter, diff --git a/src/usage/filter-index.ts b/src/usage/filter-index.ts index 85cd42c3586..6d34da8133c 100644 --- a/src/usage/filter-index.ts +++ b/src/usage/filter-index.ts @@ -1207,6 +1207,10 @@ function candidatePager( let cursor = -1; const pager = { candidateRows: 0, + /** + * Returns the next bounded page of ascending offsets, re-validating schema and generation + * inside its own transaction; a changed generation throws `UsageLedgerRebuildRequiredError`. + */ nextPage(): number[] { try { db.exec("BEGIN"); @@ -1241,6 +1245,10 @@ function hasPublishedRows(db: Database, generation: string): boolean { .get(generation) !== null; } +/** + * Refreshes the projection, then returns a paged reader for the published generation. Throws + * `UsageLedgerRebuildRequiredError` when the published meta moved under the open. + */ async function readActive( dir: string, path: string, From e8c4ce939c2c8fda7b4eef7c47a1d1526d7e5df8 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:10:41 +0000 Subject: [PATCH 09/13] fix(usage): re-check transient index-open classifications before degrading Two racing writers can observe the projection mid-creation: classifyDatabase sees the schema's statements commit one by one, and a journal sidecar can vanish mid-stat in preparePrivateSqliteDatabase. Both produced an instant foreign/unsafe degrade to streaming even though the index was healthy a moment later; the two-process rebuild test observed exactly that as candidates:null. Re-check both on the short rebuild schedule; only a home change still degrades immediately. Co-Authored-By: Epinephrine --- src/usage/filter-index.ts | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/usage/filter-index.ts b/src/usage/filter-index.ts index 6d34da8133c..fb0913b446b 100644 --- a/src/usage/filter-index.ts +++ b/src/usage/filter-index.ts @@ -1315,12 +1315,17 @@ export async function readUsageFilterCandidates( return await readActive(dir, path, query); } catch (error) { if (error instanceof UsageFilterIndexSourceError) throw error; - if (error instanceof UsageFilterIndexForeignError - || error instanceof UnsafeSqlitePathError - || error instanceof UsageFilterIndexHomeChangedError) { + // A changed home never heals on a re-check; every later open targets the new directory. + if (error instanceof UsageFilterIndexHomeChangedError) { return null; } - if (error instanceof UsageLedgerRebuildRequiredError) { + // A foreign classification or unsafe-path flag can be a transient view of a projection + // another process is still creating (the schema's statements commit one by one, and a + // journal sidecar can vanish mid-stat). Give them the short re-check schedule before + // degrading, exactly like a moved publish. + if (error instanceof UsageLedgerRebuildRequiredError + || error instanceof UsageFilterIndexForeignError + || error instanceof UnsafeSqlitePathError) { const delay = REBUILD_RETRY_DELAYS_MS[attempt]; if (delay === undefined) return null; await new Promise(resolve => setTimeout(resolve, delay)); From f0b623b453089439fb2787781cff48281d255f85 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:36:56 +0000 Subject: [PATCH 10/13] test(usage): cover transient index-open classification recovery - transient foreign file clears mid-retry and a later attempt builds the index - transient unsafe sidecar clears mid-retry and recovers the same way - permanent unsafe sidecar still falls back to streaming, unmodified Co-Authored-By: Epinephrine --- tests/usage/usage-filter-index.test.ts | 59 ++++++++++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 66d87a19067..70c0f2134ca 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -4,6 +4,7 @@ import { appendFileSync, chmodSync, existsSync, + mkdirSync, mkdtempSync, readFileSync, renameSync, @@ -1106,6 +1107,64 @@ describe("filtered usage index availability", () => { expect(existsSync(`${indexPath}-wal`)).toBe(false); }); + test("a transient foreign classification recovers on the open retry", async () => { + const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); + writeRows(rows); + const expected = rows.filter(row => row.provider === "openai").length; + const indexPath = usageFilterIndexPath(testDir); + const foreign = "this is not a sqlite database\n"; + writeFileSync(indexPath, foreign); + + const request = getFilteredUsageAggregate({ provider: "openai" }); + // Early attempts still see the foreign file; clear it inside the short re-check window so + // a later attempt builds the projection instead of degrading to the streaming read. + const clear = setTimeout(() => rmSync(indexPath, { force: true }), 60); + try { + const result = await request; + expect(summary(result).summary.requests).toBe(expected); + expect(result.filterIndex?.totalRows).toBe(rows.length); + expect(result.filterIndex?.candidateRows).toBe(expected); + } finally { + clearTimeout(clear); + } + }); + + test("a transient unsafe sidecar recovers on the open retry", async () => { + const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); + writeRows(rows); + const expected = rows.filter(row => row.provider === "openai").length; + const indexPath = usageFilterIndexPath(testDir); + const sidecar = `${indexPath}-wal`; + // A directory at a sidecar path is not a private regular file, so the open refuses it. + mkdirSync(sidecar); + + const request = getFilteredUsageAggregate({ provider: "openai" }); + const clear = setTimeout(() => rmSync(sidecar, { recursive: true, force: true }), 60); + try { + const result = await request; + expect(summary(result).summary.requests).toBe(expected); + expect(result.filterIndex?.totalRows).toBe(rows.length); + expect(result.filterIndex?.candidateRows).toBe(expected); + } finally { + clearTimeout(clear); + } + }); + + test("a permanently unsafe sidecar still falls back without being modified", async () => { + const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); + writeRows(rows); + const expected = rows.filter(row => row.provider === "openai").length; + const indexPath = usageFilterIndexPath(testDir); + const sidecar = `${indexPath}-wal`; + mkdirSync(sidecar); + + const result = await getFilteredUsageAggregate({ provider: "openai" }); + expect(summary(result).summary.requests).toBe(expected); + expect(result.filterIndex).toBeUndefined(); + expect(statSync(sidecar).isDirectory()).toBe(true); + expect(existsSync(indexPath)).toBe(false); + }); + test("an unwritable home degrades to the streaming read", async () => { const rows = Array.from({ length: 5 }, (_, index) => syntheticRow(index)); writeRows(rows); From 9463710196caee157d7fcdc154274c6139600bf1 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 10 Nov 2026 01:44:24 +0000 Subject: [PATCH 11/13] docs(usage): align retry wording with the atomic schema contract The schema now commits in one BEGIN IMMEDIATE transaction, so the stale partial-schema justification for the foreign/unsafe re-check is replaced; the structure note names the classifications that share the short schedule. Co-Authored-By: Epinephrine --- src/usage/filter-index.ts | 8 ++++---- structure/usage-client-attribution.md | 5 +++-- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/src/usage/filter-index.ts b/src/usage/filter-index.ts index 4f7e7b1d79c..82a6836fe8c 100644 --- a/src/usage/filter-index.ts +++ b/src/usage/filter-index.ts @@ -1415,10 +1415,10 @@ export async function readUsageFilterCandidates( if (error instanceof UsageFilterIndexHomeChangedError) { return null; } - // A foreign classification or unsafe-path flag can be a transient view of a projection - // another process is still creating (the schema's statements commit one by one, and a - // journal sidecar can vanish mid-stat). Give them the short re-check schedule before - // degrading, exactly like a moved publish. + // A foreign classification or unsafe-path flag can still be a transient view: the file at + // this path can be replaced mid-flight, or a sidecar appears or vanishes mid-stat while + // another writer transacts. Give them the short re-check schedule before degrading, exactly + // like a moved publish. if (error instanceof UsageLedgerRebuildRequiredError || error instanceof UsageFilterIndexForeignError || error instanceof UnsafeSqlitePathError) { diff --git a/structure/usage-client-attribution.md b/structure/usage-client-attribution.md index a8bdc8d9ea6..4d5ad7a9247 100644 --- a/structure/usage-client-attribution.md +++ b/structure/usage-client-attribution.md @@ -118,9 +118,10 @@ not this projection is never registered or modified. A passive management read o projection read-only, requires this projection's full table set rather than a coincidental meta table, and only uses it when it is current for the exact ledger revision; otherwise it streams, and it never creates, writes or registers anything. Index reads retry contention on nine delays -totaling about 6.4 s and a rebuild-required read on three short delays before degrading. Any other +totaling about 6.4 s and run a foreign or unsafe open and a rebuild-required read through three +short delays before degrading. Any other index-side failure -- unwritable home, -full disk, contention, a corrupt or foreign file -- falls back to the authoritative streaming scan +full disk, a corrupt file -- falls back to the authoritative streaming scan instead of failing a request the scan can answer; a failure to read `usage.jsonl` itself is propagated. A `UsageFilterIndexPageError` from a `src/usage/filter-index.ts` page read clears the retained filtered aggregate, then retries the authoritative streaming scan under From 9994ae98e92c405fafdb9447bf55877d21c1e069 Mon Sep 17 00:00:00 2001 From: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 10 Nov 2026 13:16:35 +0900 Subject: [PATCH 12/13] docs(usage): align busy-retry bounds and test query shape with current contract --- src/usage/filter-index.ts | 5 ++--- structure/usage-client-attribution.md | 8 +++++--- tests/usage/usage-filter-index.test.ts | 6 ++++-- 3 files changed, 11 insertions(+), 8 deletions(-) diff --git a/src/usage/filter-index.ts b/src/usage/filter-index.ts index 82a6836fe8c..c23218b7efa 100644 --- a/src/usage/filter-index.ts +++ b/src/usage/filter-index.ts @@ -1429,9 +1429,8 @@ export async function readUsageFilterCandidates( } if (isBusyError(error)) { // Contention, including a peer still creating the projection, is waited out rather than - // read as a foreign file. The bound is four busy timeouts (the fourth attempt has no - // back-off) plus these delays, about 1.1 s, before the request falls back to the - // streaming read. + // read as a foreign file. The bound is the BUSY_RETRY_DELAYS_MS sum (about 6.4 s) plus up + // to ten busy timeouts before the request falls back to the streaming read. const delay = BUSY_RETRY_DELAYS_MS[attempt]; if (delay === undefined) return null; testHooks?.beforeBusyRetry?.({ dbPath: path, attempt }); diff --git a/structure/usage-client-attribution.md b/structure/usage-client-attribution.md index 4d5ad7a9247..2be55ce16a3 100644 --- a/structure/usage-client-attribution.md +++ b/structure/usage-client-attribution.md @@ -117,9 +117,11 @@ existing file that classified as this projection, once they appear; a file at th not this projection is never registered or modified. A passive management read opens an existing projection read-only, requires this projection's full table set rather than a coincidental meta table, and only uses it when it is current for the exact ledger revision; otherwise it streams, -and it never creates, writes or registers anything. Index reads retry contention on nine delays -totaling about 6.4 s and run a foreign or unsafe open and a rebuild-required read through three -short delays before degrading. Any other +and it never creates, writes or registers anything. The passive read makes a single attempt and +degrades to the streaming scan on any error; it does not retry. Only the active index read +(`readUsageFilterCandidates`) retries contention on nine delays totaling about 6.4 s and runs a +foreign or unsafe open and a rebuild-required read through three short delays before degrading. +Any other index-side failure -- unwritable home, full disk, a corrupt file -- falls back to the authoritative streaming scan instead of failing a request the scan can answer; a failure to read `usage.jsonl` itself is diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 35ff433e16b..289b2972594 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -1488,9 +1488,11 @@ describe("filtered usage index availability", () => { test(`real SQLite page failure rolls back and wraps the error (passive=${passive})`, async () => { writeRows(Array.from({ length: 5 }, (_, i) => syntheticRow(i))); await getFilteredUsageAggregate({ provider: "openai" }); + // Explicit nulls match the normalized query shape; a bare `{}` is not a valid caller query. + const query = { provider: null, model: null, apiKeyId: null }; const read = await (passive - ? filterIndexModule.readUsageFilterCandidatesPassive({}) - : filterIndexModule.readUsageFilterCandidates({})); + ? filterIndexModule.readUsageFilterCandidatesPassive(query) + : filterIndexModule.readUsageFilterCandidates(query)); expect(read).not.toBeNull(); const writer = new Database(usageFilterIndexPath(testDir)); writer.exec("DROP TABLE usage_rows"); From 9c6254831e780da2fd28feff1ec82777721c0b12 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 10 Nov 2026 04:25:58 +0000 Subject: [PATCH 13/13] perf(usage): push the page cursor into each candidate branch The outer WHERE byteOffset > ? on the compound select could not reach the UNION branches, so every page materialized the whole candidate set before dropping the rows below the cursor. Each branch now carries the cursor bound and its own LIMIT, and the wrapped UNION still merges and deduplicates the bounded branch pages. Adds a >2,000-candidate provider-only test and a paged plan assertion on the branch seeks. Co-Authored-By: Epinephrine --- src/usage/filter-index.ts | 112 +++++++++++++++++++------ tests/usage/usage-filter-index.test.ts | 30 +++++++ 2 files changed, 116 insertions(+), 26 deletions(-) diff --git a/src/usage/filter-index.ts b/src/usage/filter-index.ts index c23218b7efa..52e162f1e09 100644 --- a/src/usage/filter-index.ts +++ b/src/usage/filter-index.ts @@ -1166,10 +1166,21 @@ async function refreshUsageFilterIndex( } } -function buildCandidateSelect( +interface CandidateBranch { + sql: string; + values: Array; +} + +/** + * One UNION branch of the candidate select. When `paged` is set each branch ends with + * `byte_offset > ?` so the pager can seek from its cursor instead of re-reading the candidates + * before it; the caller appends the cursor value at the end of each branch's values. + */ +function candidateBranches( query: UsageFilterIndexQuery, snapshot: UsageFilterIndexSnapshot, -): { sql: string; values: Array } { + paged: boolean, +): CandidateBranch[] { const attrClauses: string[] = []; const attrValues: Array = []; const rowClauses: string[] = []; @@ -1244,18 +1255,15 @@ function buildCandidateSelect( // estimate prefers the row primary key, which examines every row in the ledger even when // one API key or client matches a handful. const baseIndex = rowAccessIndexWithoutClient(query); - const branches = clientAlternatives.map(alternative => { + const cursor = paged ? " AND r.byte_offset > ?" : ""; + return clientAlternatives.map(alternative => { const index = alternative.clauses.length > 0 ? "idx_usage_rows_client" : baseIndex; const clauses = [...rowClauses, ...alternative.clauses]; return { - sql: `SELECT r.byte_offset AS byteOffset FROM usage_rows r${index ? ` INDEXED BY ${index}` : ""} WHERE r.generation = ? AND ${clauses.join(" AND ")}`, + sql: `SELECT r.byte_offset AS byteOffset FROM usage_rows r${index ? ` INDEXED BY ${index}` : ""} WHERE r.generation = ? AND ${clauses.join(" AND ")}${cursor}`, values: [snapshot.generation, ...rowValues, ...alternative.values], }; }); - return { - sql: `${branches.map(branch => branch.sql).join(" UNION ")} ORDER BY byteOffset`, - values: branches.flatMap(branch => branch.values), - }; } // Drive from the attribution index rather than scanning the row range: SQLite's default // estimate otherwise prefers the row primary key, which examines every row in the ledger. @@ -1263,18 +1271,36 @@ function buildCandidateSelect( // Rows with no usable attribution are selected too: the accumulator is what decides, and a // hand-edited row with a missing model must fail a provider/model filter exactly as the // streaming read does rather than disappear from the result. + return [ + { + sql: `SELECT DISTINCT a.byte_offset AS byteOffset FROM usage_attributions a INDEXED BY ${attrIndex} WHERE a.generation = ? AND ${attrClauses.join(" AND ")} AND a.byte_offset <= ? AND EXISTS (SELECT 1 FROM usage_rows r WHERE r.generation = a.generation AND r.byte_offset = a.byte_offset AND ${rowSql})${paged ? " AND a.byte_offset > ?" : ""}`, + values: [ + snapshot.generation, + ...attrValues, + snapshot.indexedOffset, + ...rowValues, + ...clientAlternatives.flatMap(alternative => alternative.values), + ], + }, + { + sql: `SELECT r.byte_offset AS byteOffset FROM usage_rows r INDEXED BY idx_usage_rows_unattributable WHERE r.generation = ? AND r.unattributable = 1 AND ${rowSql}${paged ? " AND r.byte_offset > ?" : ""}`, + values: [ + snapshot.generation, + ...rowValues, + ...clientAlternatives.flatMap(alternative => alternative.values), + ], + }, + ]; +} + +function buildCandidateSelect( + query: UsageFilterIndexQuery, + snapshot: UsageFilterIndexSnapshot, +): { sql: string; values: Array } { + const branches = candidateBranches(query, snapshot, false); return { - sql: `SELECT DISTINCT a.byte_offset AS byteOffset FROM usage_attributions a INDEXED BY ${attrIndex} WHERE a.generation = ? AND ${attrClauses.join(" AND ")} AND a.byte_offset <= ? AND EXISTS (SELECT 1 FROM usage_rows r WHERE r.generation = a.generation AND r.byte_offset = a.byte_offset AND ${rowSql}) UNION SELECT r.byte_offset AS byteOffset FROM usage_rows r INDEXED BY idx_usage_rows_unattributable WHERE r.generation = ? AND r.unattributable = 1 AND ${rowSql} ORDER BY byteOffset`, - values: [ - snapshot.generation, - ...attrValues, - snapshot.indexedOffset, - ...rowValues, - ...clientAlternatives.flatMap(alternative => alternative.values), - snapshot.generation, - ...rowValues, - ...clientAlternatives.flatMap(alternative => alternative.values), - ], + sql: `${branches.map(branch => branch.sql).join(" UNION ")} ORDER BY byteOffset`, + values: branches.flatMap(branch => branch.values), }; } @@ -1290,16 +1316,36 @@ function rowAccessIndexWithoutClient(query: UsageFilterIndexQuery): string | nul const CANDIDATE_PAGE_SIZE = 2_000; +/** + * The page statement behind the candidate pager: each UNION branch seeks from the page cursor + * and stops at the page size, so a late page reads at most `CANDIDATE_PAGE_SIZE` rows per branch + * rather than re-materializing every candidate before the cursor. The cursor value is appended + * at the end of each branch's values when the statement runs. + */ +function candidatePageSql( + query: UsageFilterIndexQuery, + snapshot: UsageFilterIndexSnapshot, +): { sql: string; branchValues: Array> } { + // A compound operand cannot carry its own ORDER BY/LIMIT, so each branch is wrapped before + // the UNION deduplicates and the outer ORDER BY/LIMIT merges the bounded branch pages. + const branches = candidateBranches(query, snapshot, true); + const paged = branches.map( + branch => `SELECT * FROM (${branch.sql} ORDER BY byteOffset LIMIT ${CANDIDATE_PAGE_SIZE})`, + ); + return { + sql: `SELECT byteOffset FROM (${paged.join(" UNION ")}) ORDER BY byteOffset LIMIT ?`, + branchValues: branches.map(branch => branch.values), + }; +} + /** Keyset pager over the candidate select: bounded pages, each validated in its own transaction. */ function candidatePager( db: Database, snapshot: UsageFilterIndexSnapshot, query: UsageFilterIndexQuery, ): Pick { - const { sql, values } = buildCandidateSelect(query, snapshot); - const statement = db.query( - `SELECT byteOffset FROM (${sql}) WHERE byteOffset > ? ORDER BY byteOffset LIMIT ?`, - ); + const { sql, branchValues } = candidatePageSql(query, snapshot); + const statement = db.query(sql); let cursor = -1; const pager = { candidateRows: 0, @@ -1315,7 +1361,10 @@ function candidatePager( || meta.generation !== snapshot.generation) { throw new UsageLedgerRebuildRequiredError("content_changed"); } - const rows = statement.all(...values, cursor, CANDIDATE_PAGE_SIZE) as Array<{ byteOffset: number }>; + const rows = statement.all( + ...branchValues.flatMap(values => [...values, cursor]), + CANDIDATE_PAGE_SIZE, + ) as Array<{ byteOffset: number }>; const offsets = rows.map(row => row.byteOffset); db.exec("COMMIT"); if (offsets.length > 0) cursor = offsets[offsets.length - 1]!; @@ -1518,13 +1567,24 @@ export async function readUsageFilterCandidatesPassive( } } -/** Test seam: the SQLite plan for one filter's candidate select. */ -export function usageFilterCandidatePlan(query: UsageFilterIndexQuery): string[] { +/** Test seam: the SQLite plan for one filter's candidate select; `paged` explains the pager statement. */ +export function usageFilterCandidatePlan( + query: UsageFilterIndexQuery, + options: { paged?: boolean } = {}, +): string[] { const path = indexPathForCurrentHome(); const db = new Database(path, { readonly: true }); try { const meta = readMeta(db); const snapshot = snapshotFrom(meta, 0, 0); + if (options.paged) { + const { sql, branchValues } = candidatePageSql(query, snapshot); + const rows = db.query(`EXPLAIN QUERY PLAN ${sql}`).all( + ...branchValues.flatMap(values => [...values, 0]), + CANDIDATE_PAGE_SIZE, + ) as Array<{ detail: string }>; + return rows.map(row => row.detail); + } const { sql, values } = buildCandidateSelect(query, snapshot); const rows = db.query(`EXPLAIN QUERY PLAN ${sql}`).all(...values) as Array<{ detail: string }>; return rows.map(row => row.detail); diff --git a/tests/usage/usage-filter-index.test.ts b/tests/usage/usage-filter-index.test.ts index 289b2972594..900d313f94c 100644 --- a/tests/usage/usage-filter-index.test.ts +++ b/tests/usage/usage-filter-index.test.ts @@ -243,6 +243,20 @@ describe("filtered usage candidate index", () => { } }); + test("a broad provider filter answers complete totals across bounded pages", async () => { + // More matching candidates than one page (2,000) forces late pages; each must still seek + // from the page cursor rather than re-reading the candidates before it. + const rows = Array.from({ length: 4_100 }, (_, index) => ({ + ...syntheticRow(index), + provider: "openai", + })); + writeRows(rows); + const result = await getFilteredUsageAggregate({ provider: "openai" }); + expect(result.filterIndex?.candidateRows).toBe(rows.length); + expect(summary(result)).toEqual((await fullScan({ provider: "openai" })).summarize("all", NOW)); + expect(summary(result).summary.requests).toBe(rows.length); + }); + test("an append ingests only new bytes and a retained filter folds only new matches", async () => { const rows = Array.from({ length: SYNTHETIC_ROWS }, (_, index) => syntheticRow(index)); writeRows(rows); @@ -1758,6 +1772,22 @@ describe("filtered usage index selection evidence", () => { .toContain("idx_usage_rows_ts"); }); + test("the paged candidate select seeks each branch from the page cursor", async () => { + writeRows(Array.from({ length: 50 }, (_, index) => syntheticRow(index))); + await getFilteredUsageAggregate({ provider: "openai" }); + + const paged = (query: Parameters[0]) => + usageFilterCandidatePlan(query, { paged: true }).join("\n"); + // Every UNION branch carries the `byte_offset > cursor` bound and its own LIMIT, so a late + // page seeks from the cursor instead of re-materializing the candidates before it. + const providerOnly = paged({ provider: "openai", model: null, apiKeyId: null }); + expect(providerOnly).toContain("idx_usage_attr_provider"); + expect(providerOnly).toContain("idx_usage_rows_unattributable"); + expect(providerOnly).toContain("byte_offset>"); + expect(paged({ provider: null, model: null, apiKeyId: null })) + .toContain("PRIMARY KEY (generation=? AND byte_offset>"); + }); + test("every source filter selects exactly the rows matchesUsageSource keeps", async () => { const rows: PersistedUsageEntry[] = []; let ordinal = 0;