src/ops-snapshot.ts

Functions

Imports

It imports src/env.ts, src/health-format.ts, src/health.ts, src/log.ts, src/ops-sites.ts, src/ops-types/sites.ts, src/read-capped.ts, src/time.ts. Imported by src/inbox.ts, src/maintenance-deploys.ts, src/ops-feed.ts, src/public-stats.ts, src/site-check.ts, src/tools/ops.ts, src/watcher-detect.ts, src/watcher-gather.ts, src/watcher.ts, src/whats-next.ts.

Source

1import type { Env } from "./env";
2import type { HealthReport } from "./health";
3import { parseHealth } from "./health-format";
4import { readCapped } from "./read-capped";
5import type { OpsSite } from "./ops-sites";
6import type { CiObservation, MirrorObservation, OpsSnapshot, PackageSnapshot, SiteCloudflare, SiteMapDrift, SiteProbe, SiteSnapshot } from "./ops-types/sites";
7import { logEvent } from "./log";
8import { MINUTE_MS } from "./time";
9
10// The watcher's pass, kept (capsid/research/design-ops-console.md, PR 1 of the Watch
11// Floor build). Until this, a pass kept only its timestamp: every check's outcome, the
12// mirror's newest dump and last run, and each repo's latest CI run were computed and
13// thrown away. The dashboard reads this one KV value instead of assembling the same
14// facts from about a hundred reads per load.
15
16export const OPS_SNAPSHOT_KEY = "ops:snapshot";
17
18// Uptime is a ring of half-hour slots aligned to the clock, 7 days long. A slot holds
19// '1' (the site answered), '0' (it did not) or '-' (no pass ran in that slot), so a
20// watcher that stopped shows as missing data rather than as uptime.
21const RING_SLOT_MINUTES = 30;
22export const RING_SLOTS = (7 * 24 * 60) / RING_SLOT_MINUTES;
23
24const PROBE_TIMEOUT_MS = 5000;
25// A health answer is a small JSON object. The read stops here, so a site that streams a
26// huge body cannot hold the watcher pass's memory.
27export const MAX_HEALTH_BYTES = 65_536;
28
29// The shapes live in src/ops-types/, the contract the dashboard app reads.
30// ok: the health route answered 2xx. degraded: it did not, but the root did.
31// liveness: the site has no health route and its root answered 2xx, which proves it is
32// up and nothing more. down: nothing answered 2xx.
33export type { CiObservation, MirrorObservation, OpsSnapshot, SiteProbe, SiteSnapshot } from "./ops-types/sites";
34
35export const ringSlot = (at: Date): number => Math.floor(at.getTime() / (RING_SLOT_MINUTES * MINUTE_MS));
36
37/** Adds this pass's result to a site's ring. Slots no pass reached become '-'. Two
38 *  passes in one slot (a Refresh run on demand) keep the worse result, so a failure is
39 *  never overwritten by a later success in the same half hour. */
40export function advanceRing(
41  prev: { ring: string; ring_slot: number } | undefined,
42  slot: number,
43  up: boolean
44): { ring: string; ring_slot: number } {
45  const mark = up ? "1" : "0";
46  if (!prev || prev.ring.length === 0) return { ring: mark, ring_slot: slot };
47  if (slot <= prev.ring_slot) {
48    // The same slot, or a clock that stepped back: fold into the last slot.
49    const last = prev.ring.slice(-1);
50    const worse = last === "0" || !up ? "0" : "1";
51    return { ring: prev.ring.slice(0, -1) + worse, ring_slot: prev.ring_slot };
52  }
53  const gap = Math.min(slot - prev.ring_slot - 1, RING_SLOTS);
54  const ring = (prev.ring + "-".repeat(gap) + mark).slice(-RING_SLOTS);
55  return { ring, ring_slot: slot };
56}
57
58const RING_SLOT_MS = RING_SLOT_MINUTES * MINUTE_MS;
59
60export interface RingReading {
61  namespace: string;
62  // 0 is the ring's newest slot, the one the last pass that reached this site wrote.
63  slot: number;
64  // The half hour the slot covers, [from, to).
65  from: string;
66  to: string;
67  // up: the site answered; down: it did not; no-pass: no pass reached it in that half
68  // hour; outside-ring: older than the history this ring holds yet.
69  value: "up" | "down" | "no-pass" | "outside-ring";
70  mark: string | null;
71}
72
73/** One site's ring value, by slot counted back from the newest (0) or by the time the
74 *  slot covers. A refusal is returned as a string, never thrown, for the tool to relay. */
75export function ringReading(
76  site: Pick<SiteSnapshot, "namespace" | "ring" | "ring_slot">,
77  pick: { slot: number } | { at: Date }
78): RingReading | string {
79  let back: number;
80  if ("slot" in pick) {
81    back = pick.slot;
82  } else {
83    if (!Number.isFinite(pick.at.getTime())) return "at is not a time; pass an ISO time such as 2026-09-28T12:10:00Z.";
84    back = site.ring_slot - ringSlot(pick.at);
85    if (back < 0) {
86      return `at ${pick.at.toISOString()} is after the newest slot of ${site.namespace}'s ring, which starts ${new Date(site.ring_slot * RING_SLOT_MS).toISOString()}.`;
87    }
88  }
89  if (!Number.isInteger(back) || back < 0 || back >= RING_SLOTS) {
90    return `the ring holds ${RING_SLOTS} half-hour slots (0 to ${RING_SLOTS - 1}, 0 the newest), so ${"slot" in pick ? `slot ${pick.slot}` : `at ${pick.at.toISOString()}`} is outside it.`;
91  }
92  const absolute = site.ring_slot - back;
93  const index = site.ring.length - 1 - back;
94  const mark = index >= 0 ? site.ring[index] : null;
95  const value = mark === "1" ? "up" : mark === "0" ? "down" : mark === "-" ? "no-pass" : "outside-ring";
96  return {
97    namespace: site.namespace,
98    slot: back,
99    from: new Date(absolute * RING_SLOT_MS).toISOString(),
100    to: new Date((absolute + 1) * RING_SLOT_MS).toISOString(),
101    value,
102    mark,
103  };
104}
105
106type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
107
108interface Reached {
109  status: number | null;
110  latency_ms: number | null;
111  body: string | null;
112  error: string | null;
113}
114
115async function reach(fetchImpl: FetchLike, url: string, withBody: boolean): Promise<Reached> {
116  const started = Date.now();
117  try {
118    const res = await fetchImpl(url, {
119      method: "GET",
120      redirect: "follow",
121      headers: { "User-Agent": "capsid-watcher (ops probe)" },
122      signal: AbortSignal.timeout(PROBE_TIMEOUT_MS),
123    });
124    const latency_ms = Date.now() - started;
125    if (!withBody) {
126      await res.body?.cancel();
127      return { status: res.status, latency_ms, body: null, error: null };
128    }
129    const { text, truncated } = await readCapped(res, MAX_HEALTH_BYTES);
130    // Past the cap the body is not read as health, so no sha is taken from part of it.
131    return { status: res.status, latency_ms, body: truncated ? null : text, error: null };
132  } catch (err) {
133    // A probe that got no answer is a result (the site is down from here), recorded
134    // with its reason, not an error in the watcher.
135    return { status: null, latency_ms: null, body: null, error: (err instanceof Error ? err.message : String(err)).slice(0, 160) };
136  }
137}
138
139const is2xx = (status: number | null): boolean => status !== null && status >= 200 && status < 300;
140
141/** One site. The health route when it has one, and the root as well when that route
142 *  fails, so a broken health route on a site that is up reads as degraded, not down. */
143export async function probeSite(site: OpsSite, fetchImpl: FetchLike, now: Date, selfHealth: HealthReport | null): Promise<SiteProbe> {
144  const base = {
145    namespace: site.namespace,
146    name: site.name,
147    origin: site.origin,
148    health_path: site.healthPath,
149    platform: site.platform,
150    checked_at: now.toISOString(),
151  };
152  if (site.self) {
153    if (!selfHealth) return { ...base, state: "down", http_status: null, latency_ms: null, sha: null, error: "healthReport could not be read this pass" };
154    const ok = selfHealth.status === "ok";
155    return { ...base, state: ok ? "ok" : "degraded", http_status: ok ? 200 : 503, latency_ms: null, sha: selfHealth.sha === "unknown" ? null : selfHealth.sha, error: null };
156  }
157  if (site.healthPath === null) {
158    const root = await reach(fetchImpl, site.origin + "/", false);
159    return { ...base, state: is2xx(root.status) ? "liveness" : "down", http_status: root.status, latency_ms: root.latency_ms, sha: null, error: root.error };
160  }
161  const health = await reach(fetchImpl, site.origin + site.healthPath, true);
162  if (is2xx(health.status)) {
163    return { ...base, state: "ok", http_status: health.status, latency_ms: health.latency_ms, sha: parseHealth(health.body).sha, error: null };
164  }
165  const root = await reach(fetchImpl, site.origin + "/", false);
166  const why = health.error ?? `health route answered ${health.status}`;
167  return is2xx(root.status)
168    ? { ...base, state: "degraded", http_status: health.status, latency_ms: root.latency_ms, sha: null, error: why }
169    : { ...base, state: "down", http_status: root.status ?? health.status, latency_ms: null, sha: null, error: `${why}; root ${root.error ?? `answered ${root.status}`}` };
170}
171
172const probeIsUp = (p: SiteProbe): boolean => p.state !== "down";
173
174export interface SnapshotInput {
175  now: Date;
176  pass_ms: number;
177  cadence_min: number;
178  checks: OpsSnapshot["checks"];
179  health: HealthReport | null;
180  mirror: MirrorObservation | null;
181  ci: CiObservation[];
182  site_map: SiteMapDrift | null;
183  // Null when the probes could not run at all this pass.
184  probes: SiteProbe[] | null;
185  // This pass's Cloudflare state per site, keyed by namespace (src/ops-cloudflare.ts).
186  // A probed site with no entry is written without the field, never with last pass's.
187  cloudflare?: Record<string, SiteCloudflare>;
188  // This pass's packages; undefined when they could not be read, and then the last
189  // pass's are kept rather than shown as gone.
190  packages?: PackageSnapshot[];
191}
192
193/** The snapshot for this pass, carrying each site's ring forward from `prev`. A site
194 *  not probed this pass keeps its last probe and gets a '-' slot, so the ring says a
195 *  pass happened without an answer for it. */
196export function buildSnapshot(prev: OpsSnapshot | null, input: SnapshotInput): OpsSnapshot {
197  const slot = ringSlot(input.now);
198  const prevSites = new Map((prev?.sites ?? []).map((s) => [s.namespace, s]));
199  const sites: SiteSnapshot[] = [];
200  for (const probe of input.probes ?? []) {
201    const before = prevSites.get(probe.namespace);
202    const cloudflare = input.cloudflare?.[probe.namespace];
203    sites.push({ ...probe, ...advanceRing(before, slot, probeIsUp(probe)), ...(cloudflare ? { cloudflare } : {}) });
204    prevSites.delete(probe.namespace);
205  }
206  if (input.probes === null) {
207    for (const before of prevSites.values()) {
208      const gap = Math.min(Math.max(slot - before.ring_slot, 0), RING_SLOTS);
209      const cloudflare = input.cloudflare?.[before.namespace];
210      sites.push({
211        ...before,
212        ring: (before.ring + "-".repeat(gap)).slice(-RING_SLOTS),
213        ring_slot: Math.max(slot, before.ring_slot),
214        ...(cloudflare ? { cloudflare } : {}),
215      });
216    }
217  }
218  return {
219    version: 1,
220    pass_at: input.now.toISOString(),
221    pass_ms: input.pass_ms,
222    cadence_min: input.cadence_min,
223    checks: input.checks,
224    health: input.health,
225    mirror: input.mirror,
226    ci: input.ci,
227    site_map: input.site_map,
228    sites,
229    ...(input.packages ? { packages: input.packages } : prev?.packages ? { packages: prev.packages } : {}),
230  };
231}
232
233/** The last snapshot, for its rings. A value that does not parse is logged and treated
234 *  as absent: the next write replaces it, and the rings restart, which the dashboard
235 *  shows as missing history rather than as uptime. */
236export async function readSnapshot(env: Pick<Env, "APP_KV">): Promise<OpsSnapshot | null> {
237  const raw = await env.APP_KV.get(OPS_SNAPSHOT_KEY);
238  if (raw === null) return null;
239  try {
240    const parsed = JSON.parse(raw) as OpsSnapshot;
241    if (parsed?.version !== 1 || !Array.isArray(parsed.sites)) throw new Error(`unexpected shape (version ${String(parsed?.version)})`);
242    return parsed;
243  } catch (err) {
244    logEvent("error", "OPS_SNAPSHOT_UNREADABLE", { message: `OPS_SNAPSHOT_UNREADABLE ${err instanceof Error ? err.message : String(err)}; starting the rings again` });
245    return null;
246  }
247}
248
249export async function writeSnapshot(env: Pick<Env, "APP_KV">, snapshot: OpsSnapshot): Promise<void> {
250  await env.APP_KV.put(OPS_SNAPSHOT_KEY, JSON.stringify(snapshot));
251}