From 2676120ed19f0954817f6c9fa9c0396f1f8f09a6 Mon Sep 17 00:00:00 2001 From: Drewry Pope Date: Sun, 1 Nov 2026 15:23:22 -0600 Subject: [PATCH 1/2] fix(quota-gatherer): bound provider concurrency + retain last-known snapshot on error (#1268) --- pm-stack/aggregator/scripts/gather-quota.ts | 51 +++++++++++++-- pm-stack/aggregator/src/quota-gather-map.ts | 43 +++++++++++++ .../aggregator/test/quota-gather-map.test.ts | 63 +++++++++++++++++++ 3 files changed, 153 insertions(+), 4 deletions(-) diff --git a/pm-stack/aggregator/scripts/gather-quota.ts b/pm-stack/aggregator/scripts/gather-quota.ts index 276eca2b6..8215be3a3 100755 --- a/pm-stack/aggregator/scripts/gather-quota.ts +++ b/pm-stack/aggregator/scripts/gather-quota.ts @@ -46,7 +46,11 @@ // store cannot be read, the tick writes a heartbeat carrying the // error and exits non-zero WITHOUT touching quota-current.json. A // single provider failing or exceeding its cap degrades only its own -// rows; the tick still succeeds. +// rows; the tick still succeeds. A provider whose run errors keeps +// its last-known ok entries from the previous tick's +// quota-current.json (#1268) — retained verbatim, so their original +// measuredAt makes the hub classify them stale instead of the fresh +// error rows replacing the last good data. // // Usage: bun run scripts/gather-quota.ts // Env: OPENCHAMBER_LIB_DIR (default: first of ~/git/devtoday- @@ -55,7 +59,8 @@ // QUOTA_CURRENT_PATH, QUOTA_HEARTBEAT_PATH, // QUOTA_GATHERER_STATE_PATH, QUOTA_SNAPSHOTS_DIR (./snapshots), // USAGE_SNAPSHOT_OUTBOX_PATH, META_QUOTA_EXPORT_PATH, -// QUOTA_GATHER_TIMEOUT_MS (15000), TURSO_API_TOKEN (else the +// QUOTA_GATHER_TIMEOUT_MS (15000), +// QUOTA_GATHER_CONCURRENCY (6), TURSO_API_TOKEN (else the // Bitwarden secret turso.platform-api-token-muse-cli, resolved // by name via ~/tools/bws-env conventions — BWS_BIN and // BWS_ACCESS_TOKEN_FILE override its plumbing), @@ -73,6 +78,8 @@ import { entryFromResult, entryFromFileExport, entryFromTurso, + lastKnownEntries, + mapWithConcurrency, recordsFromEntries, type FileExportDoc, type QuotaProviderResult, @@ -104,6 +111,7 @@ const statePath = process.env.QUOTA_GATHERER_STATE_PATH ?? join(PKG, "quota-gath const snapshotsDir = process.env.QUOTA_SNAPSHOTS_DIR ?? join(PKG, "snapshots"); const outboxPath = process.env.USAGE_SNAPSHOT_OUTBOX_PATH ?? null; const timeoutMs = Number(process.env.QUOTA_GATHER_TIMEOUT_MS ?? 15_000); +const gatherConcurrency = Number(process.env.QUOTA_GATHER_CONCURRENCY ?? 6); const OBSERVED_BY = "quota-gatherer"; /** File-export producers ingested each tick (the #1301 convention). */ @@ -305,6 +313,8 @@ interface ProviderRun { ms: number; entries: QuotaCurrentEntry[]; error?: string; + /** #1268: entries are last-known retentions, not fresh reads. */ + retained?: boolean; } async function runProvider(quota: QuotaModule, id: string, tickIso: string): Promise { @@ -363,6 +373,20 @@ async function runFileExport(fx: (typeof FILE_EXPORTS)[number]): Promise { + try { + const parsed: unknown = JSON.parse(await readFile(currentPath, "utf8")); + if (typeof parsed === "object" && parsed !== null && Array.isArray((parsed as QuotaCurrentFile).providers)) { + return (parsed as QuotaCurrentFile).providers; + } + } catch { + // no previous tick yet, or an unreadable file — nothing to retain + } + return []; +} + async function readState(): Promise> { try { const parsed: unknown = JSON.parse(await readFile(statePath, "utf8")); @@ -399,9 +423,28 @@ async function main(): Promise { ); } - const runs = await Promise.all(providerIds.map((id) => runProvider(quota, id, tickAt))); + // #1268: bounded provider concurrency (default 6, matching the + // OpenChamber CLI) instead of one Promise.all over the catalogue. + const runs = await mapWithConcurrency(providerIds, gatherConcurrency, (id) => runProvider(quota, id, tickAt)); for (const fx of FILE_EXPORTS) runs.push(await runFileExport(fx)); runs.push(await runTurso(tickAt)); + + // #1268: an errored run with no ok row of its own retains the + // provider's last-known ok entries from the previous tick instead + // of replacing them with fresh error rows (or with nothing, on a + // whole-fetch timeout). The heartbeat below still reports the run + // as an error, with its reason and the retained row count. + const prevEntries = await readPreviousEntries(); + if (prevEntries.length) { + for (const run of runs) { + if (run.state !== "error" || run.entries.some((e) => e.state === "ok")) continue; + const retained = lastKnownEntries(prevEntries, run.id); + if (retained.length) { + run.entries = retained; + run.retained = true; + } + } + } const entries = runs.flatMap((r) => r.entries); const perProvider: QuotaGatherHeartbeat["perProvider"] = {}; @@ -413,7 +456,7 @@ async function main(): Promise { ...(run.error ? { error: run.error } : {}), }; console.log( - `quota-gatherer: ${run.id} ${run.state} ${run.ms}ms rows=${run.entries.length}${run.error ? ` error=${run.error.slice(0, 140)}` : ""}`, + `quota-gatherer: ${run.id} ${run.state} ${run.ms}ms rows=${run.entries.length}${run.retained ? " retained-last-known" : ""}${run.error ? ` error=${run.error.slice(0, 140)}` : ""}`, ); } diff --git a/pm-stack/aggregator/src/quota-gather-map.ts b/pm-stack/aggregator/src/quota-gather-map.ts index bb6e43895..95dd5d00a 100644 --- a/pm-stack/aggregator/src/quota-gather-map.ts +++ b/pm-stack/aggregator/src/quota-gather-map.ts @@ -528,3 +528,46 @@ export function entryFromTurso( entry.usage = usage; return entry; } + +/** + * #1268: map over items with at most `limit` invocations in flight, + * preserving input order in the results. The gatherer fetches ~30 + * catalogue providers per tick; an unbounded Promise.all fires all + * of them at once. A non-positive or non-finite limit degrades to + * sequential (1) rather than to unbounded. Errors propagate like + * Promise.all: the first rejection rejects the whole map (callers + * that fail soft per item catch inside their worker, as runProvider + * does). + */ +export async function mapWithConcurrency( + items: readonly T[], + limit: number, + fn: (item: T, index: number) => Promise, +): Promise { + const cap = Number.isFinite(limit) && limit >= 1 ? Math.floor(limit) : 1; + const results = new Array(items.length); + let next = 0; + const workers = Array.from({ length: Math.min(cap, items.length) }, async () => { + while (next < items.length) { + const i = next++; + results[i] = await fn(items[i], i); + } + }); + await Promise.all(workers); + return results; +} + +/** + * #1268: the last-known entries for a provider — the previous tick's + * `ok` entries, verbatim. When this tick's run for a provider errors + * out (timeout, fetch failure), the gatherer retains these instead + * of writing fresh error rows over them: the entries keep their + * original measuredAt, so the hub's freshness classification + * (quota-current.ts) surfaces them as stale — last-known data with + * visible staleness, never fresh-looking zeros and never a vanished + * provider. Error/unconfigured previous entries are not data and are + * never retained. + */ +export function lastKnownEntries(prev: readonly QuotaCurrentEntry[], providerId: string): QuotaCurrentEntry[] { + return prev.filter((e) => e.providerId === providerId && e.state === "ok"); +} diff --git a/pm-stack/aggregator/test/quota-gather-map.test.ts b/pm-stack/aggregator/test/quota-gather-map.test.ts index eeb55bbac..93effc6f8 100644 --- a/pm-stack/aggregator/test/quota-gather-map.test.ts +++ b/pm-stack/aggregator/test/quota-gather-map.test.ts @@ -3,6 +3,8 @@ import { entryFromFileExport, entryFromTurso, entryFromResult, + lastKnownEntries, + mapWithConcurrency, normalizeResetAt, recordsFromEntries, serviceIdFor, @@ -105,6 +107,67 @@ describe("entryFromResult", () => { }); }); +describe("mapWithConcurrency (#1268)", () => { + test("preserves order and never exceeds the in-flight bound", async () => { + let inFlight = 0; + let maxInFlight = 0; + const out = await mapWithConcurrency([1, 2, 3, 4, 5, 6, 7], 3, async (n) => { + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + await Bun.sleep(n % 3); + inFlight--; + return n * 10; + }); + expect(out).toEqual([10, 20, 30, 40, 50, 60, 70]); + expect(maxInFlight).toBe(3); + }); + test("a limit above the item count still completes; empty input maps to empty", async () => { + expect(await mapWithConcurrency([1, 2], 6, async (n) => n + 1)).toEqual([2, 3]); + expect(await mapWithConcurrency([], 6, async (n: number) => n)).toEqual([]); + }); + test("a non-positive limit degrades to sequential, never unbounded", async () => { + let inFlight = 0; + let maxInFlight = 0; + const out = await mapWithConcurrency([1, 2, 3], 0, async (n) => { + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + await Bun.sleep(1); + inFlight--; + return n; + }); + expect(out).toEqual([1, 2, 3]); + expect(maxInFlight).toBe(1); + }); + test("a worker rejection rejects the map", async () => { + await expect( + mapWithConcurrency([1, 2], 2, async (n) => { + if (n === 2) throw new Error("boom"); + return n; + }), + ).rejects.toThrow("boom"); + }); +}); + +describe("lastKnownEntries (#1268)", () => { + const tick = "2026-10-27T05:00:00.000Z"; + const okEntry = entryFromResult(OK_RESULT, tick); + const errEntry = entryFromResult( + { providerId: "cline-pass", ok: false, configured: true, error: "timed out", usage: null }, + tick, + ); + const otherOk = entryFromResult({ ...OK_RESULT, providerId: "claude" }, tick); + test("returns the provider's ok entries verbatim, original measuredAt kept", () => { + const kept = lastKnownEntries([okEntry, errEntry, otherOk], "cline-pass"); + expect(kept).toEqual([okEntry]); + expect(kept[0].measuredAt).toBe(okEntry.measuredAt); + }); + test("error-only history retains nothing; unknown providers retain nothing", () => { + expect(lastKnownEntries([errEntry], "cline-pass")).toEqual([]); + expect(lastKnownEntries([okEntry], "nope")).toEqual([]); + expect(lastKnownEntries([], "cline-pass")).toEqual([]); + }); +}); + describe("recordsFromEntries", () => { test("a pct record per changed window; label-only windows yield none", () => { const entry = entryFromResult(OK_RESULT, "2026-10-27T05:00:00.000Z"); From 6190a8cc625e6bdba321b2d5f6276933590d42b6 Mon Sep 17 00:00:00 2001 From: Drewry Pope Date: Sun, 1 Nov 2026 15:29:38 -0600 Subject: [PATCH 2/2] fix(quota-gatherer): retained entries carry a failed-tick marker and classify stale immediately (#1268) --- pm-stack/aggregator/scripts/gather-quota.ts | 3 ++- pm-stack/aggregator/src/model.ts | 7 ++++++ pm-stack/aggregator/src/quota-current.ts | 4 ++++ pm-stack/aggregator/src/quota-gather-map.ts | 11 +++++++++ .../aggregator/test/quota-current.test.ts | 24 +++++++++++++++++++ .../aggregator/test/quota-gather-map.test.ts | 9 +++++++ 6 files changed, 57 insertions(+), 1 deletion(-) diff --git a/pm-stack/aggregator/scripts/gather-quota.ts b/pm-stack/aggregator/scripts/gather-quota.ts index 8215be3a3..f1763315b 100755 --- a/pm-stack/aggregator/scripts/gather-quota.ts +++ b/pm-stack/aggregator/scripts/gather-quota.ts @@ -80,6 +80,7 @@ import { entryFromTurso, lastKnownEntries, mapWithConcurrency, + markRetained, recordsFromEntries, type FileExportDoc, type QuotaProviderResult, @@ -440,7 +441,7 @@ async function main(): Promise { if (run.state !== "error" || run.entries.some((e) => e.state === "ok")) continue; const retained = lastKnownEntries(prevEntries, run.id); if (retained.length) { - run.entries = retained; + run.entries = markRetained(retained, tickAt); run.retained = true; } } diff --git a/pm-stack/aggregator/src/model.ts b/pm-stack/aggregator/src/model.ts index 57acd08e0..c7f2495d2 100644 --- a/pm-stack/aggregator/src/model.ts +++ b/pm-stack/aggregator/src/model.ts @@ -1304,6 +1304,13 @@ export interface QuotaCurrentEntry { planLabel?: string; state: "ok" | "error" | "unconfigured"; measuredAt: string; + /** #1268: ISO time of the tick that retained this entry from the + * previous tick after the provider's run errored. The data is + * last-known, never a fresh read: readers classify it stale + * regardless of measuredAt freshness (a retained entry is usually + * only one tick old, well inside the freshness bound). Absent on + * fresh reads. */ + retainedFromFailedTick?: string; source: string; error?: string; windows: QuotaCurrentWindow[]; diff --git a/pm-stack/aggregator/src/quota-current.ts b/pm-stack/aggregator/src/quota-current.ts index 5dbec60a0..3e930bf1d 100644 --- a/pm-stack/aggregator/src/quota-current.ts +++ b/pm-stack/aggregator/src/quota-current.ts @@ -68,6 +68,10 @@ type FreshState = PoolFreshness["state"]; function entryState(entry: QuotaCurrentEntry, nowMs: number): FreshState { if (entry.state === "ok") { + // #1268: retained last-known entries are stale by definition — + // the provider's latest run failed, however recent the retained + // measuredAt is (one tick is 15 min, inside the 20-min bound). + if (entry.retainedFromFailedTick) return "stale"; const t = Date.parse(entry.measuredAt); return Number.isFinite(t) && t - nowMs <= QUOTA_CURRENT_FUTURE_SKEW_MS && nowMs - t <= QUOTA_CURRENT_FRESH_MS ? "ok" diff --git a/pm-stack/aggregator/src/quota-gather-map.ts b/pm-stack/aggregator/src/quota-gather-map.ts index 95dd5d00a..5123b62f7 100644 --- a/pm-stack/aggregator/src/quota-gather-map.ts +++ b/pm-stack/aggregator/src/quota-gather-map.ts @@ -571,3 +571,14 @@ export async function mapWithConcurrency( export function lastKnownEntries(prev: readonly QuotaCurrentEntry[], providerId: string): QuotaCurrentEntry[] { return prev.filter((e) => e.providerId === providerId && e.state === "ok"); } + +/** + * #1268: stamp retained entries with the failed tick that kept them. + * Copies, never the previous tick's objects; measuredAt is preserved + * (it is the honest age of the data) — the stamp is the explicit + * signal readers use to classify the entry stale immediately, + * instead of it reading as fresh until the freshness bound lapses. + */ +export function markRetained(entries: readonly QuotaCurrentEntry[], failedTickIso: string): QuotaCurrentEntry[] { + return entries.map((e) => ({ ...e, retainedFromFailedTick: failedTickIso })); +} diff --git a/pm-stack/aggregator/test/quota-current.test.ts b/pm-stack/aggregator/test/quota-current.test.ts index 4a48ee697..4477040e9 100644 --- a/pm-stack/aggregator/test/quota-current.test.ts +++ b/pm-stack/aggregator/test/quota-current.test.ts @@ -171,6 +171,30 @@ describe("quota-current overlay", () => { expect(cline.windows[0].percent).toBe(69); // dated read, not the aged gatherer value }); + test("a retained entry is stale immediately, even inside the freshness bound (#1268)", () => { + // One tick is 15 min and the freshness bound is 20, so a retained + // entry's measuredAt alone would read fresh for ~5 min after the + // failed tick; the retention stamp is what surfaces the failure. + const at = iso(5 * 60_000); + const out = overlayQuotaCurrent( + baseView(), + file([ + entry({ + serviceId: "cline-pass:cline-pass", + measuredAt: at, + windows: [win("monthly", 71, at)], + retainedFromFailedTick: iso(1 * 60_000), + }), + ]), + NOW, + ); + const cline = out.pools.find((p) => p.id === "cline-pass:cline-pass")!; + expect(cline.state).toBe("stale"); + expect(cline.freshness).toEqual({ measuredAt: at, source: "openchamber-quota", state: "stale" }); + expect(cline.windows[0].percent).toBe(69); // dated read, not the retained gatherer value + expect(out.gatherer?.providers.find((p) => p.serviceId === "cline-pass:cline-pass")?.state).toBe("stale"); + }); + test("an ok entry dated past the future-skew tolerance is stale, windows untouched (#1541)", () => { for (const futureMs of [60 * 60_000, 30 * 24 * 60 * 60_000]) { const futureAt = iso(-futureMs); diff --git a/pm-stack/aggregator/test/quota-gather-map.test.ts b/pm-stack/aggregator/test/quota-gather-map.test.ts index 93effc6f8..62f8fb5ad 100644 --- a/pm-stack/aggregator/test/quota-gather-map.test.ts +++ b/pm-stack/aggregator/test/quota-gather-map.test.ts @@ -5,6 +5,7 @@ import { entryFromResult, lastKnownEntries, mapWithConcurrency, + markRetained, normalizeResetAt, recordsFromEntries, serviceIdFor, @@ -166,6 +167,14 @@ describe("lastKnownEntries (#1268)", () => { expect(lastKnownEntries([okEntry], "nope")).toEqual([]); expect(lastKnownEntries([], "cline-pass")).toEqual([]); }); + test("markRetained stamps copies with the failed tick; originals and measuredAt untouched", () => { + const stamped = markRetained([okEntry], "2026-10-27T05:15:00.000Z"); + expect(stamped).toHaveLength(1); + expect(stamped[0].retainedFromFailedTick).toBe("2026-10-27T05:15:00.000Z"); + expect(stamped[0].measuredAt).toBe(okEntry.measuredAt); + expect(stamped[0]).not.toBe(okEntry); + expect(okEntry.retainedFromFailedTick).toBeUndefined(); + }); }); describe("recordsFromEntries", () => {