fix(aggregator): refresh stale quota on demand (#1171)

pm-stack/aggregator/src/quota-refresh.ts

@@ -0,0 +1,151 @@
+// Lazy quota refresh (#1171): a page/API load whose newest measurement is
+// older than five minutes asks the existing gatherer to run again, without
+// blocking the response on provider calls. Concurrent loads share one request
+// and keep a refreshing status until the data advances or the unit timeout.
+//
+// The trigger reuses the deployed gatherer unit (quota-gatherer.service)
+// rather than fetching providers here — provider-fetch logic lives only in
+// scripts/gather-quota.ts.
+
+import type { QuotaView } from "./model.ts";
+
+/** The issue's freshness bound: five minutes. */
+export const STALE_AFTER_MS = 5 * 60_000;
+/** Gatherer systemd TimeoutStartSec is 300s; add one minute of grace. */
+export const MAX_REFRESH_MS = 6 * 60_000;
+/** Avoid rapid retries after a trigger command fails to start. */
+export const TRIGGER_COOLDOWN_MS = 60_000;
+
+export interface RefreshStatus {
+  /** Newest measurement was older than maxAgeMs (or absent). */
+  stale: boolean;
+  /** A gatherer run is expected to be in flight. */
+  refreshing: boolean;
+  /** This call started the gatherer. */
+  triggered: boolean;
+  /** ISO time of the most recent trigger, or null. */
+  lastTriggerAt: string | null;
+}
+
+export type RefreshTrigger = (reason: "stale" | "manual") => void | Promise<void>;
+
+/** Newest non-error provider/pool measurement in the view, in epoch ms, or null. */
+export function newestMeasurement(view: QuotaView, nowMs = Date.now()): number | null {
+  const times: number[] = [];
+  const add = (raw: string | null | undefined, state?: "ok" | "stale" | "error") => {
+    if (!raw || state === "error") return;
+    const t = Date.parse(raw);
+    // Future timestamps (clock skew) count as fresh, not as very stale.
+    if (!Number.isNaN(t)) times.push(Math.min(t, nowMs));
+  };
+  for (const pool of view.pools ?? []) {
+    add(pool.freshness?.measuredAt, pool.freshness?.state);
+    for (const credential of pool.credentials ?? []) {
+      add(credential.freshness?.measuredAt, credential.freshness?.state);
+    }
+  }
+  if (view.gatherer) {
+    for (const provider of view.gatherer.providers ?? []) add(provider.measuredAt, provider.state);
+  }
+  return times.length ? Math.max(...times) : null;
+}
+
+/** True when the data is missing or older than maxAgeMs. */
+export function isQuotaStale(view: QuotaView, nowMs = Date.now(), maxAgeMs = STALE_AFTER_MS): boolean {
+  const newest = newestMeasurement(view, nowMs);
+  return newest === null || nowMs - newest > maxAgeMs;
+}
+
+export interface QuotaRefresher {
+  /** Trigger when stale; safe to call on every load. */
+  request(view: QuotaView, nowMs?: number): RefreshStatus;
+  /** Trigger regardless of freshness (manual refresh-now control). */
+  manual(view: QuotaView, nowMs?: number): RefreshStatus;
+  /** Test/introspection hook: is a run expected to be in flight? */
+  refreshing(): boolean;
+}
+
+export function createQuotaRefresher(opts: {
+  trigger: RefreshTrigger;
+  now?: () => number;
+  maxAgeMs?: number;
+  maxRefreshMs?: number;
+  cooldownMs?: number;
+}): QuotaRefresher {
+  const now = opts.now ?? (() => Date.now());
+  const maxAgeMs = opts.maxAgeMs ?? STALE_AFTER_MS;
+  const maxRefreshMs = opts.maxRefreshMs ?? MAX_REFRESH_MS;
+  const cooldownMs = opts.cooldownMs ?? TRIGGER_COOLDOWN_MS;
+  // Track the gatherer result, not the short-lived systemctl --no-block child.
+  let active: { startedAt: number; baseline: number | null } | null = null;
+  let lastTriggerAt: number | null = null;
+
+  const iso = () => (lastTriggerAt === null ? null : new Date(lastTriggerAt).toISOString());
+  const status = (stale: boolean, refreshing: boolean, triggered = false): RefreshStatus => ({
+    stale,
+    refreshing,
+    triggered,
+    lastTriggerAt: iso(),
+  });
+
+  function observe(view: QuotaView, nowMs: number): boolean {
+    if (!active) return false;
+    const measurement = newestMeasurement(view, nowMs);
+    const dataAdvanced = measurement !== null && (active.baseline === null || measurement > active.baseline);
+    if (dataAdvanced || nowMs - active.startedAt >= maxRefreshMs) {
+      active = null;
+      return false;
+    }
+    return true;
+  }
+
+  function kick(reason: "stale" | "manual", view: QuotaView, nowMs: number): void {
+    const run = { startedAt: nowMs, baseline: newestMeasurement(view, nowMs) };
+    // Set synchronously so concurrent request handlers share the trigger.
+    active = run;
+    lastTriggerAt = nowMs;
+    const fail = (err: unknown) => {
+      console.error("aggregator: quota refresh trigger failed (" + reason + "):", err);
+      if (active === run) active = null;
+    };
+    try {
+      // The trigger only asks systemd to queue the existing gatherer; it does
+      // not wait for provider calls or the gatherer's five-minute run.
+      void Promise.resolve(opts.trigger(reason)).catch(fail);
+    } catch (err) {
+      fail(err);
+    }
+  }
+
+  return {
+    request(view, nowMs = now()) {
+      const stale = isQuotaStale(view, nowMs, maxAgeMs);
+      const refreshing = observe(view, nowMs);
+      if (!stale) return status(false, refreshing);
+      if (refreshing) return status(true, true);
+      if (lastTriggerAt !== null && nowMs - lastTriggerAt < cooldownMs) return status(true, false);
+      kick("stale", view, nowMs);
+      return status(true, true, true);
+    },
+    manual(view, nowMs = now()) {
+      const stale = isQuotaStale(view, nowMs, maxAgeMs);
+      if (observe(view, nowMs)) return status(stale, true);
+      kick("manual", view, nowMs);
+      return status(stale, true, true);
+    },
+    refreshing: () => active !== null,
+  };
+}
+
+/** Queue the deployed gatherer; provider fetching remains in that service. */
+export function systemdTrigger(
+  command = process.env.QUOTA_REFRESH_CMD ?? "systemctl --user start --no-block quota-gatherer.service",
+): RefreshTrigger {
+  return async () => {
+    const argv = command.trim().split(/\s+/);
+    if (argv.length === 0 || !argv[0]) throw new Error("empty refresh command");
+    const proc = Bun.spawn(argv, { stdin: "ignore", stdout: "ignore", stderr: "ignore" });
+    const exitCode = await proc.exited;
+    if (exitCode !== 0) throw new Error("systemd gatherer start exited with code " + exitCode);
+  };
+}

pm-stack/aggregator/src/render.ts

@@ -2342,11 +2342,27 @@ function chargesStripHtml(usage: UsageView): string {
   return `<h2>Charges this period</h2><p class="muted">Evidence, not accounting — charges observed on cards/receipts in the local records (full balances stay in the finance tooling). Total observed: $${total.toFixed(2)}.</p><div class="tblwrap"><table><caption>Charges this period</caption><tr><th scope="col">Date</th><th scope="col">Charge</th><th scope="col">Amount</th><th scope="col">Method</th><th scope="col">Evidence</th></tr>${rows}</table></div>`;
 }
 
+/** Lazy-refresh strip (#1171): a visible "refreshing" marker while a
+ *  gatherer run is in flight, plus the manual refresh-now control. Both
+ *  use the same trigger path; neither blocks the page on provider calls. */
+function quotaRefreshHtml(refresh?: import("./quota-refresh.ts").RefreshStatus): string {
+  if (!refresh) return "";
+  const marker = refresh.refreshing
+    ? `<span class="badge refreshing" data-quota-refreshing="true">refreshing — the quota gatherer is running; reload for updated data</span>`
+    : refresh.stale
+      ? `<span class="muted">quota data is older than five minutes</span>`
+      : "";
+  return `<div class="card quota-refresh"><div>${marker || "<span class=\"muted\">quota data is fresh</span>"}</div><form method="post" action="/quota/refresh"><button type="submit">Refresh now</button></form></div>`;
+}
+
 export function renderQuota(
   view: QuotaView,
   provenance: Provenance,
   usage?: UsageView,
   detail?: ProvenanceDetail,
+  /** Lazy-refresh state (#1171): renders the "refreshing" marker and the
+   *  manual refresh-now control. Absent means no request was made. */
+  refresh?: import("./quota-refresh.ts").RefreshStatus,
 ): string {
   const credentialLifecycle = view.credentialLifecycle?.length
     ? `<h2>Credential lifecycle</h2><div class="card">${view.credentialLifecycle
@@ -2451,6 +2467,7 @@ ${burnAndChoices}`,
     `<h1>Usage &amp; billing</h1><p class="muted">Token quotas, PAYG dollars, and SaaS usage vs limits — from dated local records; every card carries its measured-at stamp, and stale cards dim. Machine-first: <a href="/api/quota.json">/api/quota.json</a>.</p>
 ${attentionStripHtml(usage)}
 ${credentialLifecycle}
+${quotaRefreshHtml(refresh)}
 <h2>Token quotas</h2>${pools}
 ${paygCardHtml(usage)}
 <h2>SaaS &amp; services</h2>${groups || '<p class="muted">No services extracted yet.</p>'}

pm-stack/aggregator/src/server.ts

@@ -7,6 +7,7 @@
 // (GitHub, Kaneo, Grist) stays read-only from here — the fleet jobs
 // apply actions; the dashboard only records them.
 import { stat } from "node:fs/promises";
+import { createQuotaRefresher, systemdTrigger } from "./quota-refresh.ts";
 import { decodePathSegment } from "./path-segment.ts";
 import { errorBoundaryResponse, notFoundResponse } from "./error-boundary.ts";
 import { ActionsStore, validateNewAction, type DashAction } from "./actions.ts";
@@ -241,6 +242,10 @@ function pickSource(): Source {
 
 const source = pickSource();
 
+/** Lazy quota refresh (#1171): stale page/API loads ask the deployed
+ *  gatherer unit to run again; single-flight + cooldown live in the module. */
+const quotaRefresher = createQuotaRefresher({ trigger: systemdTrigger() });
+
 // #1279: prefer the latest-comment-per-ref read so a busy batch cannot
 // truncate an item's most recent activity out of its brief.
 const briefCommentsFor = async (refs: RefKey[]): Promise<Comment[]> =>
@@ -1344,7 +1349,16 @@ async function handler(req: Request): Promise<Response> {
       source.quota().catch(() => emptyQuotaView()),
       source.usage().catch(() => emptyUsageView()),
     ]);
-    return html(renderQuota(quota, prov(), usage, provDetail()));
+    // Never block on providers (#1171): decide from the stored measurement,
+    // ask the gatherer to run in the background, render what we have.
+    const refresh = quotaRefresher.request(quota);
+    return html(renderQuota(quota, prov(), usage, provDetail(), refresh));
+  }
+
+  if (req.method === "POST" && path === "/quota/refresh") {
+    const quota = await source.quota().catch(() => emptyQuotaView());
+    quotaRefresher.manual(quota);
+    return new Response(null, { status: 303, headers: { Location: "/quota" } });
   }
 
   const serviceMatch = /^\/quota\/service\/([\w-]+)$/.exec(path);
@@ -1678,8 +1692,12 @@ async function handler(req: Request): Promise<Response> {
     // hub sections ride as additional top-level keys. Snapshot history
     // stays out of this twin — it is per-service, on the service twin.
     const [quota, usage] = await Promise.all([source.quota(), source.usage()]);
+    // #1171: a stale read returns immediately with refresh state instead of
+    // waiting on provider calls; the gatherer runs in the background.
+    const refresh = quotaRefresher.request(quota);
     return json({
       ...quota,
+      refresh,
       saas: usage.services,
       payg: usage.payg,
       charges: usage.charges,
@@ -1688,6 +1706,11 @@ async function handler(req: Request): Promise<Response> {
       subscriptions: usage.subscriptions,
     });
   }
+  if (req.method === "POST" && path === "/api/quota/refresh") {
+    const quota = await source.quota().catch(() => emptyQuotaView());
+    const refresh = quotaRefresher.manual(quota);
+    return json({ requested: refresh.triggered, refresh });
+  }
   if (path === "/api/quota/service.json") {
     const id = url.searchParams.get("id") ?? "";
     const usage = await source.usage();

pm-stack/aggregator/test/nav.test.ts

@@ -37,7 +37,7 @@ function serverPaths(): { exact: string[]; prefixes: string[] } {
 
 /** Routes that are not pages a person navigates to: JSON twins, POST
  * endpoints, datastar fragments, static assets, and the /board alias. */
-const NON_PAGE = /^\/api\/|^\/frag\/|^\/assets\/|^\/sw\.js$|^\/board$|^\/manifest\.webmanifest$/;
+const NON_PAGE = /^\/api\/|^\/frag\/|^\/assets\/|^\/sw\.js$|^\/board$|^\/manifest\.webmanifest$|^\/quota\/refresh$/;
 
 describe("the manifest groups every page (#1135)", () => {
   test("every page names one of PAGE_GROUPS, and groups have no page-less id", () => {