fix(usage): yield between costly ledger rows

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

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<void>(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<void>(resolve => setTimeout(resolve, 0));
+        lastYieldAt = performance.now();
         bytesSinceYield = 0;
         linesSinceYield = 0;
       }

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)

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),