diff --git a/docs-site/src/content/docs/reference/management-api.md b/docs-site/src/content/docs/reference/management-api.md index 35101ba3176..9c199eb0c09 100644 --- a/docs-site/src/content/docs/reference/management-api.md +++ b/docs-site/src/content/docs/reference/management-api.md @@ -480,6 +480,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..2d469de105d 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,24 @@ export async function scanUsageLedgerCooperatively( } lineLength = 0; cursor = newline + 1; + // Bytes do not bound normalization/aggregation time on a busy host. Let + // 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(); + 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 d231a30be99..8db118c606e 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 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. > 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..f81a85d9d8b 100644 --- a/tests/usage/usage-ledger-scanner.test.ts +++ b/tests/usage/usage-ledger-scanner.test.ts @@ -165,6 +165,52 @@ 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(""))); + // 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(); + } + }); + + 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),