From b786f385abbe1e6d6b42b892ee107d523fb3e199 Mon Sep 17 00:00:00 2001 From: seedds Date: Mon, 9 Nov 2026 17:20:56 +0100 Subject: [PATCH 1/2] fix(usage): yield between costly ledger rows --- .../content/docs/reference/management-api.md | 3 ++ src/usage/ledger-scanner.ts | 12 ++++++ structure/dashboard-and-usage.md | 6 +-- tests/usage/usage-ledger-scanner.test.ts | 42 +++++++++++++++++++ 4 files changed, 60 insertions(+), 3 deletions(-) diff --git a/docs-site/src/content/docs/reference/management-api.md b/docs-site/src/content/docs/reference/management-api.md index 1ecced18e34..7ae6e40a0a7 100644 --- a/docs-site/src/content/docs/reference/management-api.md +++ b/docs-site/src/content/docs/reference/management-api.md @@ -473,6 +473,9 @@ an earlier file prefix from 7-day, 30-day, or all-history totals. `managementUsa accepted for compatibility with bounded legacy readers, but changing it no longer expands or reduces the history summarized by this endpoint. +Cold ledger scans yield between records so health checks and other local requests can continue +while a large usage report is calculated. A single record is processed atomically. + Pass both `since` and `until` to select an inclusive custom interval. Each accepts integer Unix epoch **milliseconds**, or a full ISO datetime with an explicit timezone. Invalid dates, negative or out-of-range values, reversed bounds, and a single bound are rejected. Custom bounds override diff --git a/src/usage/ledger-scanner.ts b/src/usage/ledger-scanner.ts index c2c143139a6..eb5808e1cba 100644 --- a/src/usage/ledger-scanner.ts +++ b/src/usage/ledger-scanner.ts @@ -20,6 +20,7 @@ export const USAGE_LEDGER_ROW_READ_WINDOW_BYTES = 64 * 1024; // one unterminated line an unbounded allocation. export const USAGE_LEDGER_MAX_LINE_BYTES = 1024 * 1024; export const USAGE_LEDGER_BOUNDARY_DIGEST_BYTES = 64 * 1024; +const MAX_SCAN_SLICE_MS = 8; export interface ScanUsageLedgerOptions { signal?: AbortSignal; @@ -289,6 +290,7 @@ export async function scanUsageLedgerCooperatively( let bytesRead = 0; let bytesSinceYield = 0; let linesSinceYield = 0; + let lastYieldAt = performance.now(); let processedThroughBytes = startAtBytes; const capturedHash = createHash("sha256"); const committedTail = new RollingByteWindow(USAGE_LEDGER_BOUNDARY_DIGEST_BYTES); @@ -395,11 +397,21 @@ export async function scanUsageLedgerCooperatively( } lineLength = 0; cursor = newline + 1; + // Bytes do not bound normalization/aggregation time on a busy host. Let + // network callbacks and cancellation run even within the final chunk. + if (linesSinceYield >= 1_000 || performance.now() - lastYieldAt >= MAX_SCAN_SLICE_MS) { + await new Promise(resolve => setTimeout(resolve, 0)); + throwIfAborted(options.signal); + lastYieldAt = performance.now(); + bytesSinceYield = 0; + linesSinceYield = 0; + } } if (position < scanEnd && (bytesSinceYield >= USAGE_LEDGER_READ_CHUNK_BYTES || linesSinceYield >= 1_000)) { await new Promise(resolve => setTimeout(resolve, 0)); + lastYieldAt = performance.now(); bytesSinceYield = 0; linesSinceYield = 0; } diff --git a/structure/dashboard-and-usage.md b/structure/dashboard-and-usage.md index bb0df0429f2..12c2c187c21 100644 --- a/structure/dashboard-and-usage.md +++ b/structure/dashboard-and-usage.md @@ -340,9 +340,9 @@ normalized per-request rows after a response. File identity changes, shrinkage, changes, pricing-overlay changes, and local-time-zone changes force a cold rebuild. Ordinary growth is treated as an append: the scanner verifies the previous LF and its trailing 64 KiB digest, then folds only the suffix into a cloned accumulator and publishes it after validation. Concurrent callers -share that work. Cold rebuilds scan the whole ledger in fixed-size chunks and yield between bounded -batches, so memory stays bounded and unrelated management requests remain serviceable even for a -large existing log. The first read is proportional to ledger size; later refreshes hash the bounded retained window once and parse only the appended suffix, rather than rescanning the whole ledger. +share that work. Cold rebuilds scan fixed-size chunks and yield between rows after 8 ms or 1,000 rows, +including within the final chunk, then recheck cancellation. This keeps costly normalization and +aggregation cooperative; one row remains atomic. The first read is proportional to ledger size; later refreshes hash the bounded retained window once and parse only the appended suffix, rather than rescanning the whole ledger. `tests/usage/usage-ledger-scanner.test.ts` covers slow-row scheduling and cancellation. The Dashboard polls its 30-day usage independently once per minute, separate from five-second state polls. An unchanged retained snapshot reuses its verified region digest only for identical bounds; appends or trimming hash the returned region, preserving same-inode rewrite detection. > Decision record: [ADR-0102](decisions/ADR-0102-incremental-stream-accounting.md) diff --git a/tests/usage/usage-ledger-scanner.test.ts b/tests/usage/usage-ledger-scanner.test.ts index 0883ce9f3df..46bf54aabc3 100644 --- a/tests/usage/usage-ledger-scanner.test.ts +++ b/tests/usage/usage-ledger-scanner.test.ts @@ -165,6 +165,48 @@ describe("usage ledger cooperative scanner", () => { expect(totalTokens).toBe(4_200); }); + test("yields between expensive rows even when the ledger fits in one read", async () => { + const rows = Array.from({ length: 40 }, (_, index) => line(`slow-${index}`)); + writeFileSync(usageLogPath(), rows.join("")); + let elapsed = 0; + let visited = 0; + let visitedAtTimer = -1; + const clock = spyOn(performance, "now").mockImplementation(() => elapsed); + const timer = setTimeout(() => { visitedAtTimer = visited; }, 0); + try { + const result = await scanUsageLedgerCooperatively({ + onEntry() { visited += 1; elapsed += 10; }, + }); + expect(visitedAtTimer).toBeGreaterThan(0); + expect(visitedAtTimer).toBeLessThan(rows.length); + expect(result.parsedRows).toBe(rows.length); + expect(result.processedThroughBytes).toBe(Buffer.byteLength(rows.join(""))); + } finally { + clearTimeout(timer); + clock.mockRestore(); + } + }); + + test("honors cancellation delivered between expensive rows in one read", async () => { + writeFileSync(usageLogPath(), Array.from({ length: 40 }, (_, index) => line(`abort-${index}`)).join("")); + let elapsed = 0; + let visited = 0; + const clock = spyOn(performance, "now").mockImplementation(() => elapsed); + const abort = new AbortController(); + const reason = new Error("cancel slow scan"); + const timer = setTimeout(() => abort.abort(reason), 0); + try { + await expect(scanUsageLedgerCooperatively({ + signal: abort.signal, + onEntry() { visited += 1; elapsed += 10; }, + })).rejects.toBe(reason); + expect(visited).toBeLessThan(40); + } finally { + clearTimeout(timer); + clock.mockRestore(); + } + }); + test("skips malformed, invalid UTF-8, oversized, and torn final rows with bounded recovery", async () => { const exactlyAtLimit = Buffer.concat([ Buffer.alloc(USAGE_LEDGER_MAX_LINE_BYTES, 0x20), From 2349db49c9e0ad0132d31a55b57789ee6d1c1785 Mon Sep 17 00:00:00 2001 From: lidge-jun <243035832+lidge-jun@users.noreply.github.com> Date: Thu, 12 Nov 2026 00:58:27 +0900 Subject: [PATCH 2/2] fix(usage): yield inside a chunk on elapsed time only A per-row 1,000-row check yielded several times per chunk of cheap rows and made cold scans of large ledgers slower than dev. The row limit stays at the chunk boundary; inside a chunk only the 8 ms budget yields. Assert that a paused scan publishes the same checkpoint digest as an unpaused one. Co-authored-by: seedds --- src/usage/ledger-scanner.ts | 7 +++++-- structure/dashboard-and-usage.md | 4 ++-- tests/usage/usage-ledger-scanner.test.ts | 4 ++++ 3 files changed, 11 insertions(+), 4 deletions(-) diff --git a/src/usage/ledger-scanner.ts b/src/usage/ledger-scanner.ts index eb5808e1cba..2d469de105d 100644 --- a/src/usage/ledger-scanner.ts +++ b/src/usage/ledger-scanner.ts @@ -398,8 +398,11 @@ export async function scanUsageLedgerCooperatively( lineLength = 0; cursor = newline + 1; // Bytes do not bound normalization/aggregation time on a busy host. Let - // network callbacks and cancellation run even within the final chunk. - if (linesSinceYield >= 1_000 || performance.now() - lastYieldAt >= MAX_SCAN_SLICE_MS) { + // network callbacks, timers and cancellation run even within the final chunk. + // Only elapsed time triggers this yield: a row count here would yield several + // times per chunk of cheap rows and slow a cold scan (the chunk boundary below + // keeps the row limit). setTimeout, unlike setImmediate, lets due timers run. + if (performance.now() - lastYieldAt >= MAX_SCAN_SLICE_MS) { await new Promise(resolve => setTimeout(resolve, 0)); throwIfAborted(options.signal); lastYieldAt = performance.now(); diff --git a/structure/dashboard-and-usage.md b/structure/dashboard-and-usage.md index 97785aa37b5..8db118c606e 100644 --- a/structure/dashboard-and-usage.md +++ b/structure/dashboard-and-usage.md @@ -340,8 +340,8 @@ normalized per-request rows after a response. File identity changes, shrinkage, changes, pricing-overlay changes, and local-time-zone changes force a cold rebuild. Ordinary growth is treated as an append: the scanner verifies the previous LF and its trailing 64 KiB digest, then folds only the suffix into a cloned accumulator and publishes it after validation. Concurrent callers -share that work. Cold rebuilds scan fixed-size chunks and yield between rows after 8 ms or 1,000 rows, -including within the final chunk, then recheck cancellation. This keeps costly normalization and +share that work. Cold rebuilds and append scans read fixed-size chunks, yield between chunks after a full +chunk or 1,000 rows, and yield between rows once 8 ms have passed, including within the final chunk, then recheck cancellation. This keeps costly normalization and aggregation cooperative; one row remains atomic. The first read is proportional to ledger size; later refreshes hash the bounded retained window once and parse only the appended suffix, rather than rescanning the whole ledger. `tests/usage/usage-ledger-scanner.test.ts` covers slow-row scheduling and cancellation. The Dashboard polls its 30-day usage independently once per minute, separate from five-second state polls. An unchanged retained snapshot reuses its verified region digest only for identical bounds; appends or trimming hash the returned region, preserving same-inode rewrite detection. diff --git a/tests/usage/usage-ledger-scanner.test.ts b/tests/usage/usage-ledger-scanner.test.ts index 46bf54aabc3..f81a85d9d8b 100644 --- a/tests/usage/usage-ledger-scanner.test.ts +++ b/tests/usage/usage-ledger-scanner.test.ts @@ -181,6 +181,10 @@ describe("usage ledger cooperative scanner", () => { expect(visitedAtTimer).toBeLessThan(rows.length); expect(result.parsedRows).toBe(rows.length); expect(result.processedThroughBytes).toBe(Buffer.byteLength(rows.join(""))); + // Pausing between rows must not change the checkpoint a scan publishes. + clock.mockRestore(); + const unpaused = await scanUsageLedgerCooperatively({ onEntry() {} }); + expect(result.processedThroughDigest).toBe(unpaused.processedThroughDigest); } finally { clearTimeout(timer); clock.mockRestore();