fix(history): bound the integrity check wait (#7984)

src/routing/history/integrity.ts

@@ -11,6 +11,7 @@ import { RebuildWorkerJob } from "./rebuild-job";
 
 export const REQUEST_HISTORY_INTEGRITY_INTERVAL_MS = 5 * 60_000;
 export const REQUEST_HISTORY_INTEGRITY_FAILURE_COOLDOWN_MS = 30_000;
+export const REQUEST_HISTORY_INTEGRITY_CHECK_TIMEOUT_MS = 30_000;
 
 type IntegrityJob = Pick<RebuildWorkerJob, "check" | "dispose" | "afterExit">;
 
@@ -19,14 +20,15 @@ export interface RequestHistoryIntegrityOptions {
   now?: () => number;
   intervalMs?: number;
   failureCooldownMs?: number;
+  checkTimeoutMs?: number;
+  scheduleTimeout?: (callback: () => void, ms: number) => () => void;
 }
 
 export type RequestHistoryIntegrityState =
   | { state: "ready" }
   | { state: "corrupt" }
   | { state: "unavailable"; error: unknown }
-  // Never rejects, including on cancellation or check failure. Completion
-  // proves the scanner exited; inspect again to obtain its verdict or error.
+  // Never rejects. Inspect again for the verdict or operational failure.
   | { state: "pending"; promise: Promise<void> };
 
 interface Generation {
@@ -43,26 +45,36 @@ interface ActiveCheck {
   cancelled: boolean;
   cancel: () => void;
   promise: Promise<void>;
+  barrier: Promise<void>;
 }
 
 type CheckOutcome =
   | { kind: "checked"; healthy: boolean }
   | { kind: "failed"; error: unknown }
+  | { kind: "timeout"; error: Error }
   | { kind: "cancelled" };
 
 export class RequestHistoryIntegrityMonitor {
   private readonly createJob: () => IntegrityJob;
   private readonly now: () => number;
   private readonly intervalMs: number;
   private readonly failureCooldownMs: number;
+  private readonly checkTimeoutMs: number;
+  private readonly scheduleTimeout: (callback: () => void, ms: number) => () => void;
   private current: Generation | null = null;
   private active: ActiveCheck | null = null;
+  private readonly draining = new Set<ActiveCheck>();
 
   constructor(options: RequestHistoryIntegrityOptions = {}) {
     this.createJob = options.createJob ?? (() => new RebuildWorkerJob());
     this.now = options.now ?? Date.now;
     this.intervalMs = options.intervalMs ?? REQUEST_HISTORY_INTEGRITY_INTERVAL_MS;
     this.failureCooldownMs = options.failureCooldownMs ?? REQUEST_HISTORY_INTEGRITY_FAILURE_COOLDOWN_MS;
+    this.checkTimeoutMs = options.checkTimeoutMs ?? REQUEST_HISTORY_INTEGRITY_CHECK_TIMEOUT_MS;
+    this.scheduleTimeout = options.scheduleTimeout ?? ((callback, ms) => {
+      const timer = setTimeout(callback, ms);
+      return () => clearTimeout(timer);
+    });
   }
 
   inspect(handle: object, path: string): RequestHistoryIntegrityState {
@@ -72,6 +84,8 @@ export class RequestHistoryIntegrityMonitor {
     }
     const generation = this.current;
     if (generation.corrupt) return { state: "corrupt" };
+    const old = [...this.draining].filter(check => check.generation !== generation);
+    if (old.length) return { state: "pending", promise: Promise.all(old.map(check => check.barrier)).then(() => {}) };
     if (this.active) {
       if (this.active.generation === generation && !this.active.cancelled && generation.verified) {
         return { state: "ready" };
@@ -97,7 +111,7 @@ export class RequestHistoryIntegrityMonitor {
    * use inspect instead. The caller drains scans before rebuilding/swapping.
    */
   acceptRebuilt(handle: object, path: string): void {
-    if (this.active) throw new Error("history integrity worker must exit before accepting a rebuilt index");
+    if (this.draining.size) throw new Error("history integrity worker must exit before accepting a rebuilt index");
     this.current = {
       handle, path, verified: true, corrupt: false,
       nextCheckAt: this.now() + this.intervalMs, failure: null,
@@ -113,7 +127,7 @@ export class RequestHistoryIntegrityMonitor {
   quiesce(): Promise<void> | null {
     this.current = null;
     this.active?.cancel();
-    return this.active?.promise ?? null;
+    return this.draining.size ? Promise.all([...this.draining].map(check => check.barrier)).then(() => {}) : null;
   }
 
   reset(): void {
@@ -130,14 +144,17 @@ export class RequestHistoryIntegrityMonitor {
     }
     let cancel!: () => void;
     const cancelled = new Promise<CheckOutcome>(resolve => { cancel = () => resolve({ kind: "cancelled" }); });
+    let release!: () => void;
     const check: ActiveCheck = {
       generation,
       cancelled: false,
       cancel: () => { check.cancelled = true; cancel(); },
-      promise: Promise.resolve(),
+      promise: new Promise(resolve => { release = resolve; }),
+      barrier: Promise.resolve(),
     };
     this.active = check;
-    check.promise = this.run(check, job, cancelled);
+    this.draining.add(check);
+    check.barrier = this.run(check, job, cancelled, release);
     return check;
   }
 
@@ -146,7 +163,10 @@ export class RequestHistoryIntegrityMonitor {
     generation.nextCheckAt = this.now() + this.failureCooldownMs;
   }
 
-  private async run(check: ActiveCheck, job: IntegrityJob, cancelled: Promise<CheckOutcome>): Promise<void> {
+  private async run(check: ActiveCheck, job: IntegrityJob, cancelled: Promise<CheckOutcome>, release: () => void): Promise<void> {
+    let expire!: (outcome: CheckOutcome) => void;
+    const timeout = new Promise<CheckOutcome>(resolve => { expire = resolve; });
+    const clear = this.scheduleTimeout(() => expire({ kind: "timeout", error: new Error("request-history integrity check timed out") }), this.checkTimeoutMs);
     const observed = Promise.resolve().then(async (): Promise<CheckOutcome> => {
       if (check.cancelled) return { kind: "cancelled" };
       try {
@@ -155,7 +175,16 @@ export class RequestHistoryIntegrityMonitor {
         return { kind: "failed", error };
       }
     });
-    const outcome = await Promise.race([observed, cancelled]);
+    const outcome = await Promise.race([observed, cancelled, timeout]);
+    clear();
+    if (outcome.kind === "timeout") {
+      if (this.current === check.generation && !check.cancelled && !check.generation.corrupt) {
+        this.recordFailure(check.generation, outcome.error);
+      }
+      if (this.active === check) this.active = null;
+      void job.dispose().catch(() => undefined);
+      release();
+    }
     await new Promise<void>(resolve => {
       // afterExit may return at its disposal timeout, before native I/O lets
       // the worker exit. Only its callback releases the file replacement gate.
@@ -173,5 +202,7 @@ export class RequestHistoryIntegrityMonitor {
       }
     }
     if (this.active === check) this.active = null;
+    this.draining.delete(check);
+    release();
   }
 }

structure/dashboard-and-usage.md

@@ -279,7 +279,7 @@ inheriting its bounds, so deleting a ledger row removes it from the grouping.
 `usageLedgerMaxBytes` is unset by default; when set, an append that crosses it publishes the newest whole rows byte for byte through the shared atomic writer, refuses the rename unless the source is the exact revision that was copied, and then discards the Logs ring, the retained aggregates and the request-history index so no surface serves rows the ledger no longer has.
 That index (`src/routing/history/`) rebuilds a ledger of at least `REQUEST_HISTORY_WORKER_REBUILD_MIN_BYTES` (1 MiB) on a worker thread (#6990): `rebuild-worker.ts`, driven by `rebuild-job.ts`, ingests the ledger snapshot the refresh recorded into a private `<index>.rebuild-<pid>-<n>` file through the same `ingest.ts` the inline path uses, switches it to WAL and closes it; the serving thread then closes the old index and renames the build over it. Async readers wait for the build; the routing-time sync entry never does and throws `RequestHistoryIndexBusyError`, which health scoring reads as unknown evidence. Health scoring (`src/routing/health.ts`) counts physical sends only: a request answered locally or refused before any send is no evidence for any target, and a row's terminal status and close reason score its last physical send, so a 200 stream cut mid-flight after a failover counts against the target that dropped it (`tests/server/health-scoring.test.ts`). A close discards the build, rejects waiting readers with `RequestHistoryIndexClosedError`, and removes the build files only after the worker thread exits. For 30 s after a worker build fails (other than on a replaced ledger), async readers and manual rebuilds rebuild inline instead, while routing-time reads keep reporting busy.
 
-`src/routing/history/integrity.ts` validates an existing persisted index on first use with a read-only worker `PRAGMA quick_check` (#7966); serving refreshes retain schema/source/offset checks without a synchronous integrity scan. Async readers await initial validation; synchronous routing reports busy/unknown meanwhile. A successful fresh inline reconstruction is immediately usable, and a private worker build is integrity-checked before publication. After validation or reconstruction, the next refresh at least five minutes later starts one background check; verified reads remain available during it, and idle indexes start no timer. Operational check failures leave the index intact and delay retry for 30 s; an unverified index remains unavailable. A corrupt verdict or `SQLITE_CORRUPT`/`SQLITE_NOTADB` invalidates the live handle and enters the existing rebuild path; async history and analytics reads retry once after recovery. Close/rebuild discards stale verdicts and drains scanners through actual thread exit before index replacement. Neither validation nor recovery rewrites canonical `usage.jsonl`. `tests/usage/request-history-integrity.test.ts` covers scan scheduling, corruption recovery and lifecycle isolation.
+`src/routing/history/integrity.ts` validates an existing persisted index on first use with a read-only worker `PRAGMA quick_check` (#7966); serving refreshes retain schema/source/offset checks without a synchronous integrity scan. Async readers await initial validation; synchronous routing reports busy/unknown meanwhile. A successful fresh inline reconstruction is immediately usable, and a private worker build is integrity-checked before publication. After validation or reconstruction, the next refresh at least five minutes later starts one background check; verified reads remain available during it, and idle indexes start no timer. A check that has not answered in 30 s is stopped and reported as an operational failure; waiting readers are released without waiting for the worker's actual exit. Operational check failures leave the index intact and delay retry for 30 s; an unverified index remains unavailable. A corrupt verdict or `SQLITE_CORRUPT`/`SQLITE_NOTADB` invalidates the live handle and enters the existing rebuild path; async history and analytics reads retry once after recovery. Close/rebuild discards stale verdicts and drains all scanners through actual thread exit before index replacement, including timed-out scanners. Neither validation nor recovery rewrites canonical `usage.jsonl`. `tests/usage/request-history-integrity.test.ts` covers scan scheduling, corruption recovery and lifecycle isolation.
 
 Ledger revisions in `src/usage/log.ts` capture device/inode identifiers from bigint descriptor stats as exact decimal strings; history metadata, ingestion, cooperative scans and retention compare them without narrowing to Numbers, preserving distinct NTFS identifiers above 2^53. An older index containing a rounded identifier fails the identity check and rebuilds once. `src/usage/summary.ts` turns that file into the `/api/usage` shape — totals, daily zero-filled grid, model and provider breakdowns, and `measured / reported / unreported / unsupported / estimated` counts. Each scope also aggregates end-to-end output throughput (#6309): measured output tokens and wall-clock `durationMs` are summed over attempts reporting both, exposing `throughputTokensPerSec` — a token-sum-over-duration-sum, never a mean of per-request rates. `throughputSamples` counts qualifying attempts (legacy rows without attempts contribute one); nonpositive or nonfinite tokens/timing are excluded. The Usage summary and Models/Providers tables display the rate and sample basis, or unavailable when none qualify.
 All three throughput displays use the same finite-number guard as their tooltip; malformed or nonfinite rates are unavailable rather than formatted.

tests/usage/request-history-integrity.test.ts

@@ -138,6 +138,123 @@ function healthQueryIndex(handle: Database): string {
 }
 
 describe("request-history integrity lifecycle (#7966)", () => {
+  test("a silent scanner times out for readers but retains the replacement barrier", async () => {
+    let clock = 0;
+    let fire: (() => void) | undefined;
+    const exits: Array<() => void> = [];
+    let jobs = 0;
+    let disposals = 0;
+    const monitor = new RequestHistoryIntegrityMonitor({
+      now: () => clock,
+      failureCooldownMs: 100,
+      checkTimeoutMs: 50,
+      scheduleTimeout: (callback: () => void, ms: number) => {
+        expect(ms).toBe(50);
+        fire = callback;
+        return () => { fire = undefined; };
+      },
+      createJob: () => {
+        jobs++;
+        return {
+          check: () => new Promise<boolean>(() => {}),
+          dispose: async () => { disposals++; },
+          afterExit: async (cleanup: () => void) => { exits.push(cleanup); },
+        };
+      },
+    });
+    const handle = {};
+    const first = monitor.inspect(handle, "index.sqlite");
+    if (first.state !== "pending") throw new Error("initial check must be pending");
+    expect(fire).toBeDefined();
+    clock = 50;
+    fire!();
+    await first.promise;
+    expect(disposals).toBe(1);
+    expect(monitor.inspect(handle, "index.sqlite").state).toBe("unavailable");
+    expect(jobs).toBe(1);
+    clock = 150;
+    expect(monitor.inspect(handle, "index.sqlite").state).toBe("pending");
+    expect(jobs).toBe(2);
+    const barrier = monitor.quiesce();
+    expect(barrier).not.toBeNull();
+    let released = false;
+    void barrier!.then(() => { released = true; });
+    await Promise.resolve();
+    expect(released).toBe(false);
+    exits[0]!();
+    await Promise.resolve();
+    expect(released).toBe(false);
+    // Let cancellation schedule the second worker's exit callback.
+    await Promise.resolve();
+    exits[1]!();
+    await barrier;
+    expect(released).toBe(true);
+  }, 1000);
+
+  test("a verdict arriving after the timeout never changes the generation's state", async () => {
+    for (const lateVerdict of [true, false]) {
+      let clock = 0;
+      const fires: Array<() => void> = [];
+      const answers: Array<(healthy: boolean) => void> = [];
+      const exits: Array<() => void> = [];
+      const monitor = new RequestHistoryIntegrityMonitor({
+        now: () => clock,
+        failureCooldownMs: 100,
+        checkTimeoutMs: 50,
+        scheduleTimeout: callback => { fires.push(callback); return () => {}; },
+        createJob: () => ({
+          check: () => new Promise<boolean>(resolve => { answers.push(resolve); }),
+          dispose: async () => {},
+          afterExit: async (cleanup: () => void) => { exits.push(cleanup); },
+        }),
+      });
+      const handle = {};
+      const first = monitor.inspect(handle, "index.sqlite");
+      if (first.state !== "pending") throw new Error("initial check must be pending");
+      await Promise.resolve();
+      fires[0]!();
+      await first.promise;
+      expect(monitor.inspect(handle, "index.sqlite").state).toBe("unavailable");
+      // The timed-out worker answers late, then exits.
+      answers[0]!(lateVerdict);
+      exits[0]!();
+      for (let turn = 0; turn < 5; turn++) await Promise.resolve();
+      expect(monitor.inspect(handle, "index.sqlite").state).toBe("unavailable");
+      // After the cooldown a fresh check decides, and only its verdict counts.
+      clock = 150;
+      const fresh = monitor.inspect(handle, "index.sqlite");
+      if (fresh.state !== "pending") throw new Error("fresh check must be pending");
+      await Promise.resolve();
+      answers[1]!(true);
+      for (let turn = 0; turn < 5 && !exits[1]; turn++) await Promise.resolve();
+      exits[1]!();
+      await fresh.promise;
+      expect(monitor.inspect(handle, "index.sqlite").state).toBe("ready");
+    }
+  }, 1000);
+
+  test("shared async readers leave a silent initial integrity check with a bounded error", async () => {
+    await existingIndex();
+    let fire!: () => void;
+    let started!: () => void;
+    let exit!: () => void;
+    const scheduled = new Promise<void>(resolve => { started = resolve; });
+    setRequestHistoryIntegrityOptionsForTests({
+      scheduleTimeout: callback => { fire = callback; started(); return () => {}; },
+      createJob: () => ({
+        check: () => new Promise<boolean>(() => {}),
+        dispose: async () => {},
+        afterExit: async cleanup => { exit = cleanup; },
+      }),
+    });
+    const first = openRequestHistoryIndex();
+    const second = openRequestHistoryIndex();
+    const errors = Promise.all([first.catch(error => error), second.catch(error => error)]);
+    await scheduled;
+    fire();
+    expect((await errors).every(error => error instanceof RequestHistoryIndexBusyError)).toBe(true);
+    exit();
+  }, 5000);
   test("real worker checks leave repeated page, row, status and routing reads free of full serving-thread scans", async () => {
     await existingIndex();
     const query = spyOn(Database.prototype, "query");