Public: anyone can read this; the crew can change it.
/system updates over one server-sent-events stream and never reloads or polls #2723
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/2720-system-updates-over-one-server-sent-even → main 5 commits fork 5 threads · 5 unresolved
9 files
changed +300
−50
README.md +1 −1
| @@ -974,7 +974,7 @@ Where the whole forge waits is a forge admin's: `/flow` shows each stage from re | ||
| 974 | 974 | |
| 975 | 975 | The terminal also has an Access command (`:` → Access), a site admin's: it reads the same last-day non-browser requests as Settings → Access and the `access` MCP tool, newest first, and filters them with `s` (since), `u` (user agent) and `p` (path), `c` to clear. |
| 976 | 976 | |
| 977 | `/system` is the machine, for the owner (#1789): per-core CPU, hwmon temperatures, load, memory, disk per mount, network, and every process grouped by what the forge launched it for (`check:<run>`, `app:<slug>/<env>`, the forge, the host) with CPU, memory and age. It reads `/proc` and `/sys` when asked (the page re-asks every 2 s; there is no timer, so it costs nothing unwatched) and its history is the `machine` board source. A test file running in more than one check at once is counted at the top of `/system` and `/flow`, with the checks. Command lines are never kept — only `*.test.ts` arguments. | |
| 977 | `/system` is the machine, for the owner (#1789): per-core CPU, hwmon temperatures, load, memory, disk per mount, network, and every process grouped by what the forge launched it for (`check:<run>`, `app:<slug>/<env>`, the forge, the host) with CPU, memory and age. It reads `/proc` and `/sys` and updates in place over one event stream, `/system/stream` (`?pause` holds it still; no reload, no polling); the 2 s push timer runs only while someone is subscribed, so it costs nothing unwatched. Its history is the `machine` board source. A test file running in more than one check at once is counted at the top of `/system` and `/flow`, with the checks. Command lines are never kept — only `*.test.ts` arguments. | |
| 978 | 978 | In the terminal, `:` → System shows the same view (the owner's only), and `:` → Telemetry lists the agents and flips `telemetry:read` for the one you press Enter on. |
| 979 | 979 | An agent the owner grants `telemetry:read` (Settings → Telemetry, one switch per agent, revoked the same way) reads the same through `GET /api/v1/telemetry/{system,flow,sources,history}` and the MCP `flow` and `system` tools; private repositories' checks and apps are renamed `check:private-N` and lose their processes and test files, and `flow` (like `system`, an admin agent account needs the grant too: only a person who is an admin reads without it) names only public repositories' conflicted pulls, every read is listed on that page, and an agent without the grant is refused. `/contribute`, `GET /api/v1/contribute` and the MCP `contribute` tool are open to anyone: what the forge is, the bottleneck now, open issues labelled “help wanted” in public repositories, and how to propose (fork, pull, the same gate). The grant is a table, not a token scope, so no token has to be re-issued. |
| 980 | 980 |
docs/design/audit.md +1 −1
| @@ -85,7 +85,7 @@ Machine routes are not pages and have no entry: `/audit.csv`, | ||
| 85 | 85 | |
| 86 | 86 | ### Machine and operation (admin) |
| 87 | 87 | |
| 88 | - `/system` — `pages/system.ts` · the machine as it is now: cores, load, memory, processes · `/flow`, `/` · none seen; its own refresh script. | |
| 88 | - `/system` — `pages/system.ts` · the machine as it is now: cores, load, memory, processes · `/flow`, `/` · none seen; updates in place over `/system/stream`, no reload script. | |
| 89 | 89 | - `/resilience` — `pages/resilience.ts` · each protection in `docs/THREATS.md` against the last gate's tests · `/settings/resilience` (similar name, different content) · none seen. |
| 90 | 90 | - `/admin/limits` — `pages/admin-limits.ts` · every limit with value, default and reason · `/settings/capacity` (also a numbers-you-can-change page) · none seen. |
| 91 | 91 | - `/settings/capacity` — `pages/capacity.ts` · how many runs and checks go at once · `/admin/limits`, `/settings/usage` · none seen. |
src/http/server.ts +2 −1
| @@ -15,7 +15,7 @@ import { parentCallback, parentStart } from "./rp.ts"; | ||
| 15 | 15 | import { accessPage } from "../web/pages/access.ts"; |
| 16 | 16 | import { capacityPage, capacitySet } from "../web/pages/capacity.ts"; |
| 17 | 17 | import { flowPage } from "../web/pages/flow.ts"; |
| 18 | import { systemPage } from "../web/pages/system.ts"; | |
| 18 | import { systemPage, systemStream } from "../web/pages/system.ts"; | |
| 19 | 19 | import { telemetryPage, telemetrySet } from "../web/pages/telemetry.ts"; |
| 20 | 20 | import { contributePage } from "../web/pages/contribute.ts"; |
| 21 | 21 | import { flagsPage, flagsSet } from "../web/pages/flags.ts"; |
| @@ -730,6 +730,7 @@ export function buildRouter(): Router<Handler> { | ||
| 730 | 730 | // The pipeline's stages and its bottleneck, for a forge admin (#1554); literal, above /:owner. |
| 731 | 731 | router.add("GET", "/flow", flowPage); |
| 732 | 732 | router.add("GET", "/system", systemPage); |
| 733 | router.add("GET", "/system/stream", systemStream); | |
| 733 | 734 | router.add("GET", "/contribute", contributePage); |
| 734 | 735 | router.add("GET", "/settings/telemetry", telemetryPage); |
| 735 | 736 | router.add("POST", "/settings/telemetry", telemetrySet); |
src/http/sse.ts +6 −2
| @@ -82,7 +82,7 @@ export type Emit = (event: string, data: unknown, id?: string) => void; | ||
| 82 | 82 | */ |
| 83 | 83 | export function eventStream( |
| 84 | 84 | request: Request, |
| 85 | register: (emit: Emit, close: () => void) => () => void, | |
| 85 | register: (emit: Emit, close: () => void, ready: () => boolean) => () => void, | |
| 86 | 86 | ): Response { |
| 87 | 87 | const encoder = new TextEncoder(); |
| 88 | 88 | let cleanup = () => {}; |
| @@ -116,6 +116,10 @@ export function eventStream( | ||
| 116 | 116 | ); |
| 117 | 117 | }; |
| 118 | 118 | |
| 119 | // A source of samples (the latest wins, older ones are worthless) | |
| 120 | // skips a frame while the reader is behind instead of queueing it. | |
| 121 | const ready = () => open && (controller.desiredSize ?? 0) > MAX_BUFFERED_BYTES / 2; | |
| 122 | ||
| 119 | 123 | const heartbeat = setInterval(() => write(`: ping\n\n`), HEARTBEAT_MS); |
| 120 | 124 | // Assigned before `register` runs: a source that finishes immediately |
| 121 | 125 | // (a job that is already done) must find a cleanup that works. |
| @@ -148,7 +152,7 @@ export function eventStream( | ||
| 148 | 152 | request.signal.addEventListener("abort", cleanup); |
| 149 | 153 | release = holdForCaller(request, () => cleanup()); |
| 150 | 154 | write(`: connected\n\n`); |
| 151 | unsubscribe = register(emit, () => cleanup()); | |
| 155 | unsubscribe = register(emit, () => cleanup(), ready); | |
| 152 | 156 | // A registration that closed the stream synchronously leaves an |
| 153 | 157 | // unsubscribe nobody has run yet. |
| 154 | 158 | if (!open) unsubscribe(); |
src/system/live.ts +38 −2
| @@ -12,18 +12,53 @@ import { buildSystem, forAgent, type Roots, type SystemView } from "./view.ts"; | ||
| 12 | 12 | |
| 13 | 13 | const FRESH_MS = 1_000; |
| 14 | 14 | |
| 15 | /** How often a watched reading is pushed: only while someone is subscribed. */ | |
| 16 | export const PUSH_MS = 2_000; | |
| 17 | ||
| 15 | 18 | export type Live = { |
| 16 | 19 | read(now?: number): SystemView; |
| 20 | /** Hear every sample pushed; the first subscriber starts the clock, the last stops it. */ | |
| 21 | subscribe(listener: (view: SystemView) => void): () => void; | |
| 22 | /** Whether the push clock is running: only while someone is subscribed. */ | |
| 23 | running(): boolean; | |
| 24 | /** Hand a sample to every subscriber (the clock does this; a test may too). */ | |
| 25 | publish(view: SystemView): void; | |
| 17 | 26 | }; |
| 18 | 27 | |
| 19 | export function createLive(options: { paths?: Paths; roots?: () => Roots; self?: number } = {}): Live { | |
| 28 | export function createLive(options: { paths?: Paths; roots?: () => Roots; self?: number; pushMs?: number } = {}): Live { | |
| 20 | 29 | const paths = options.paths ?? HOST; |
| 21 | 30 | const roots = options.roots ?? cellRoots; |
| 22 | 31 | const self = options.self ?? process.pid; |
| 23 | 32 | let prev: Raw | undefined; |
| 24 | 33 | let last: Raw | undefined; |
| 25 | 34 | let view: SystemView | undefined; |
| 26 | return { | |
| 35 | const listeners = new Set<(view: SystemView) => void>(); | |
| 36 | let timer: ReturnType<typeof setInterval> | undefined; | |
| 37 | const self_: Live = { | |
| 38 | subscribe(listener) { | |
| 39 | listeners.add(listener); | |
| 40 | if (timer === undefined) { | |
| 41 | timer = setInterval(() => self_.publish(self_.read()), options.pushMs ?? PUSH_MS); | |
| 42 | timer.unref?.(); | |
| 43 | } | |
| 44 | return () => { | |
| 45 | listeners.delete(listener); | |
| 46 | if (listeners.size === 0 && timer !== undefined) { | |
| 47 | clearInterval(timer); | |
| 48 | timer = undefined; | |
| 49 | } | |
| 50 | }; | |
| 51 | }, | |
| 52 | running: () => timer !== undefined, | |
| 53 | publish(sample) { | |
| 54 | for (const listener of [...listeners]) { | |
| 55 | try { | |
| 56 | listener(sample); | |
| 57 | } catch { | |
| 58 | // one broken viewer never starves the others | |
| 59 | } | |
| 60 | } | |
| 61 | }, | |
| 27 | 62 | read(now = Date.now()) { |
| 28 | 63 | if (view !== undefined && last !== undefined && now - last.at < FRESH_MS) return view; |
| 29 | 64 | const raw = readRaw(paths, now); |
| @@ -34,6 +69,7 @@ export function createLive(options: { paths?: Paths; roots?: () => Roots; self?: | ||
| 34 | 69 | return view; |
| 35 | 70 | }, |
| 36 | 71 | }; |
| 72 | return self_; | |
| 37 | 73 | } |
| 38 | 74 | |
| 39 | 75 | let shared = createLive(); |
src/web/pages/system.ts +77 −42
| @@ -2,9 +2,10 @@ | ||
| 2 | 2 | * /system, for the owner: the machine as it is now — cores, temperatures, |
| 3 | 3 | * load, memory, disks, network — and every process grouped by what the |
| 4 | 4 | * forge launched it for, each with CPU, memory and age. The same test file |
| 5 | * in more than one check at once is named at the top. It refreshes itself | |
| 6 | * every 2 s (a few lines of static script; `?pause` turns that off), so a | |
| 7 | * viewer *is* the sampler (`system/live.ts`). Numbers are `system/view.ts`'s. | |
| 5 | * in more than one check at once is named at the top. It updates in place | |
| 6 | * over one event stream (`/system/stream`, `system-live.ts` patches it; `?pause` | |
| 7 | * leaves it still) and never reloads or polls; while someone watches, the | |
| 8 | * sampler (`system/live.ts`) pushes a reading every 2 s. Numbers are `system/view.ts`'s. | |
| 8 | 9 | * Below it, for a forge admin who is a person, the forge's resident memory |
| 9 | 10 | * (RSS) beside its JS heap, hourly for 24 h (#2151): RSS climbing while the |
| 10 | 11 | * heap stays flat is native memory growth. Drawn on the server as SVG. |
| @@ -15,10 +16,13 @@ import { telemetryReader } from "../../actions/telemetry.ts"; | ||
| 15 | 16 | import type { Ctx } from "../../http/context.ts"; |
| 16 | 17 | import { memoryCapHolder } from "../../apps/contain.ts"; |
| 17 | 18 | import { bytes, span } from "../../system/format.ts"; |
| 19 | import { eventStream } from "../../http/sse.ts"; | |
| 20 | import type { SystemView } from "../../system/view.ts"; | |
| 18 | 21 | import { live } from "../../system/live.ts"; |
| 19 | 22 | import { notFoundPage, signInFirst } from "../chrome.ts"; |
| 20 | 23 | import { html, raw, type Html } from "../html.ts"; |
| 21 | 24 | import { page } from "../layout.ts"; |
| 25 | import { systemLiveJs } from "../system-live.ts"; | |
| 22 | 26 | import { card, empty, listCard, pill, row } from "../ui.ts"; |
| 23 | 27 | |
| 24 | 28 | /** A group's memory, and a cell's beside its cap and who holds the cap. */ |
| @@ -51,14 +55,12 @@ function memorySection(ctx: Ctx): Html { | ||
| 51 | 55 | </table>`; |
| 52 | 56 | } |
| 53 | 57 | |
| 54 | const RELOAD = `if (!/[?&]pause\\b/.test(location.search)) setTimeout(function () { location.reload(); }, 2000);`; | |
| 55 | ||
| 56 | // GET /system | |
| 57 | export function systemPage(ctx: Ctx): Response { | |
| 58 | if (!ctx.auth.user) return signInFirst(ctx); | |
| 59 | const reader = telemetryReader(ctx.app, ctx.auth.user); | |
| 60 | if (reader?.kind !== "owner") return notFoundPage(ctx); | |
| 61 | const v = live().read(); | |
| 58 | /** | |
| 59 | * The parts of the page a sample moves, each the inner markup of the element | |
| 60 | * of that id. The first paint and every pushed sample are drawn by this one | |
| 61 | * function, so the stream can only ever say what the page would have. | |
| 62 | */ | |
| 63 | export function regions(v: SystemView): Record<string, string> { | |
| 62 | 64 | const dup: Html = |
| 63 | 65 | v.duplicates.length === 0 |
| 64 | 66 | ? html`<p id="duplicates" data-count="0" class="muted">No test file is running in more than one check.</p>` |
| @@ -69,39 +71,20 @@ export function systemPage(ctx: Ctx): Response { | ||
| 69 | 71 | <ul class="list small mono"> |
| 70 | 72 | ${v.duplicates.map((d) => html`<li>${d.file} · ${String(d.checks.length)} checks: ${d.checks.join(", ")}</li>`)} |
| 71 | 73 | </ul>`; |
| 72 | const holder = memoryCapHolder(); | |
| 73 | const caps: Html = | |
| 74 | holder.by === "kernel" | |
| 75 | ? html`<p class="small" id="cell-caps" data-by="kernel">Each cell's memory is capped by the kernel (its own cgroup's <code>memory.max</code>): a cell over it is killed alone.</p>` | |
| 76 | : html`<p class="small" id="cell-caps" data-by="watch">Each cell's memory is watched once a second and the cell stopped over its cap; the kernel does not cap it (${holder.reason}).</p>`; | |
| 77 | 74 | const cores = v.cpu.cores.map( |
| 78 | 75 | (c, i) => html`<span class="mono small" title="cpu${String(i)}">${c === null ? "—" : String(Math.round(c))}</span> `, |
| 79 | 76 | ); |
| 80 | const body = html`<div class="page-head"> | |
| 81 | <h1>System</h1> | |
| 82 | <p class="muted"> | |
| 83 | The machine now, sampled each time this page refreshes (every 2 s); | |
| 84 | <a href="/system?pause">pause</a>. Its history is on | |
| 85 | <a href="/boards">Boards</a> (the machine source); the pipeline is on | |
| 86 | <a href="/flow">Flow</a>. | |
| 87 | </p> | |
| 88 | </div> | |
| 89 | ${dup} | |
| 90 | ${card( | |
| 91 | html`<div id="summary" class="row-meta"> | |
| 92 | ${pill(`CPU ${pct(v.cpu.busyPct)}`, v.cpu.busyPct !== null && v.cpu.busyPct >= 90 ? "danger" : undefined)} | |
| 77 | const out: Record<string, Html> = { | |
| 78 | "sys-dup": dup, | |
| 79 | summary: html`${pill(`CPU ${pct(v.cpu.busyPct)}`, v.cpu.busyPct !== null && v.cpu.busyPct >= 90 ? "danger" : undefined)} | |
| 93 | 80 | ${pill(`${String(v.cpu.cores.length)} cores`)} |
| 94 | 81 | ${pill(`load ${v.load ? v.load.map((l) => l.toFixed(2)).join(" ") : "—"}`)} |
| 95 | 82 | ${pill(`hottest ${v.hottestC === null ? "—" : `${v.hottestC.toFixed(0)}°C`}`)} |
| 96 | 83 | ${pill(`memory ${v.memory ? `${bytes(v.memory.usedBytes)} of ${bytes(v.memory.totalBytes)}` : "—"}`)} |
| 97 | ${pill(`${String(v.processes)} processes`)} | |
| 98 | </div> | |
| 99 | <p class="small" id="cores" aria-label="Per-core CPU, percent">${cores}</p> | |
| 100 | ${caps} | |
| 101 | <p class="small mono" id="temps">${v.temps.map((t) => `${t.chip} ${t.label} ${t.celsius.toFixed(0)}°C`).join(" · ")}</p>`, | |
| 102 | { head: html`<h2>Machine</h2>` }, | |
| 103 | )} | |
| 104 | ${listCard( | |
| 84 | ${pill(`${String(v.processes)} processes`)}`, | |
| 85 | cores: html`${cores}`, | |
| 86 | temps: html`${v.temps.map((t) => `${t.chip} ${t.label} ${t.celsius.toFixed(0)}°C`).join(" · ")}`, | |
| 87 | "sys-disks": listCard( | |
| 105 | 88 | [ |
| 106 | 89 | ...v.disks.map((d) => |
| 107 | 90 | row({ |
| @@ -118,8 +101,8 @@ export function systemPage(ctx: Ctx): Response { | ||
| 118 | 101 | ), |
| 119 | 102 | ], |
| 120 | 103 | { head: html`<h2>Disks and network</h2>` }, |
| 121 | )} | |
| 122 | ${listCard( | |
| 104 | ), | |
| 105 | "sys-procs": listCard( | |
| 123 | 106 | v.groups.map( |
| 124 | 107 | (g) => html`<li data-kind="${g.kind}"><div class="row"> |
| 125 | 108 | <span class="row-main"><span class="row-title mono">${g.label}</span> |
| @@ -128,7 +111,59 @@ export function systemPage(ctx: Ctx): Response { | ||
| 128 | 111 | </div></li>`, |
| 129 | 112 | ), |
| 130 | 113 | { head: html`<h2>Processes</h2>` }, |
| 131 | )} | |
| 132 | ${memorySection(ctx)}`; | |
| 133 | return page({ ctx, title: "System", body, wide: true, scriptJs: RELOAD }); | |
| 114 | ), | |
| 115 | }; | |
| 116 | return Object.fromEntries(Object.entries(out).map(([k, h]) => [k, h.text])); | |
| 117 | } | |
| 118 | ||
| 119 | /** The sample as the stream sends it. */ | |
| 120 | export const sampleFrame = (v: SystemView): { regions: Record<string, string> } => ({ regions: regions(v) }); | |
| 121 | ||
| 122 | // GET /system/stream: the owner's live samples, one frame each. | |
| 123 | export function systemStream(ctx: Ctx): Response { | |
| 124 | if (!ctx.auth.user) return signInFirst(ctx); | |
| 125 | if (telemetryReader(ctx.app, ctx.auth.user)?.kind !== "owner") return notFoundPage(ctx); | |
| 126 | const sampler = live(); | |
| 127 | return eventStream(ctx.request, (emit, _close, ready) => { | |
| 128 | emit("sample", sampleFrame(sampler.read())); | |
| 129 | return sampler.subscribe((view) => { | |
| 130 | if (ready()) emit("sample", sampleFrame(view)); | |
Minor: |
||
| 131 | }); | |
| 132 | }); | |
| 133 | } | |
| 134 | ||
| 135 | // GET /system | |
| 136 | export function systemPage(ctx: Ctx): Response { | |
| 137 | if (!ctx.auth.user) return signInFirst(ctx); | |
| 138 | const reader = telemetryReader(ctx.app, ctx.auth.user); | |
| 139 | if (reader?.kind !== "owner") return notFoundPage(ctx); | |
| 140 | const v = live().read(); | |
| 141 | const holder = memoryCapHolder(); | |
| 142 | const caps: Html = | |
| 143 | holder.by === "kernel" | |
| 144 | ? html`<p class="small" id="cell-caps" data-by="kernel">Each cell's memory is capped by the kernel (its own cgroup's <code>memory.max</code>): a cell over it is killed alone.</p>` | |
| 145 | : html`<p class="small" id="cell-caps" data-by="watch">Each cell's memory is watched once a second and the cell stopped over its cap; the kernel does not cap it (${holder.reason}).</p>`; | |
| 146 | const r = regions(v); | |
| 147 | const at = (id: string): Html => raw(r[id]!); | |
| 148 | const body = html`<div class="page-head"> | |
| 149 | <h1>System</h1> | |
| 150 | <p class="muted"> | |
| 151 | The machine now, updated in place about every 2 s over one event | |
| 152 | stream; <a href="/system?pause">pause</a>. Its history is on | |
| 153 | <a href="/boards">Boards</a> (the machine source); the pipeline is on | |
| 154 | <a href="/flow">Flow</a>. | |
| 155 | </p> | |
| 156 | </div> | |
| 157 | <div id="sys-dup">${at("sys-dup")}</div> | |
| 158 | ${card( | |
| 159 | html`<div id="summary" class="row-meta">${at("summary")}</div> | |
| 160 | <p class="small" id="cores" aria-label="Per-core CPU, percent">${at("cores")}</p> | |
| 161 | ${caps} | |
| 162 | <p class="small mono" id="temps">${at("temps")}</p>`, | |
| 163 | { head: html`<h2>Machine</h2>` }, | |
| 164 | )} | |
| 165 | <div id="sys-disks">${at("sys-disks")}</div> | |
| 166 | <div id="sys-procs">${at("sys-procs")}</div> | |
| 167 | ${memorySection(ctx)}`; | |
| 168 | return page({ ctx, title: "System", body, wide: true, scriptJs: systemLiveJs }); | |
| 134 | 169 | } |
src/web/system-live.ts added +23
| @@ -0,0 +1,23 @@ | ||
| 1 | /** | |
| 2 | * /system's client: one EventSource, and each pushed sample patched into the | |
| 3 | * element of the id it names. No reload, no timer, no polling; EventSource | |
| 4 | * itself reconnects after a roll (the stream's `retry:` hint) and the first | |
| 5 | * frame of the new stream is the latest sample. `?pause` opens nothing. | |
| 6 | * The markup comes from our own server, escaped there, so it is set as HTML. | |
| 7 | */ | |
| 8 | export const systemLiveJs = /* js */ ` | |
| 9 | (function () { | |
| 10 | if (typeof EventSource === 'undefined' || /[?&]pause\\b/.test(location.search)) return; | |
| 11 | var es = new EventSource('/system/stream'); | |
| 12 | es.addEventListener('sample', function (e) { | |
| 13 | var data; | |
| 14 | try { data = JSON.parse(e.data); } catch (_) { return; } | |
| 15 | var regions = data && data.regions; | |
| 16 | if (!regions) return; | |
| 17 | for (var id in regions) { | |
| 18 | var el = document.getElementById(id); | |
| 19 | if (el && el.innerHTML !== regions[id]) el.innerHTML = regions[id]; | |
| 20 | } | |
| 21 | }); | |
| 22 | })(); | |
| 23 | `; |
test/design-audit.test.ts +1 −1
| @@ -11,7 +11,7 @@ const doc = readFileSync("docs/design/audit.md", "utf8"); | ||
| 11 | 11 | |
| 12 | 12 | /** Routes that answer a file, a fragment, a redirect hop or a callback. */ |
| 13 | 13 | const NOT_PAGES = |
| 14 | /^\/(assets\/|\.well-known\/|icons\/|og\/|live\/|__forge\/|go\/|events$|sw\.js$|robots\.txt$|sitemap\.xml$|manifest\.webmanifest$|llms\.txt$|releases\.json$|audit\.csv$|fleet\/graph\/live$|fleet\/graph\.csv$|settings\/origins\/callback$|home\/archived\/)/; | |
| 14 | /^\/(assets\/|\.well-known\/|icons\/|og\/|live\/|__forge\/|go\/|events$|system\/stream$|sw\.js$|robots\.txt$|sitemap\.xml$|manifest\.webmanifest$|llms\.txt$|releases\.json$|audit\.csv$|fleet\/graph\/live$|fleet\/graph\.csv$|settings\/origins\/callback$|home\/archived\/)/; | |
| 15 | 15 | |
| 16 | 16 | const getRoutes = [ |
| 17 | 17 | ...server.matchAll(/router\.add\(\s*"GET",\s*"([^"]+)"/g), |
test/system-stream.test.ts added +151
| @@ -0,0 +1,151 @@ | ||
| 1 | /** | |
| 2 | * /system is live over one event stream (#2720): its HTML never reloads or | |
| 3 | * polls, and a sample handed to the sampler arrives on the stream. | |
| 4 | */ | |
| 5 | import { afterAll, beforeAll, describe, expect, test } from "bun:test"; | |
| 6 | import { mkdtemp, rm } from "node:fs/promises"; | |
| 7 | import { tmpdir } from "node:os"; | |
| 8 | import { join } from "node:path"; | |
| 9 | import { createApp, type App } from "../src/app.ts"; | |
| 10 | import { eventStream, sseDroppedCount } from "../src/http/sse.ts"; | |
| 11 | import { createUser } from "../src/core/users.ts"; | |
| 12 | import { serve, type Servers } from "../src/http/server.ts"; | |
| 13 | import { createLive, useLive } from "../src/system/live.ts"; | |
| 14 | ||
| 15 | describe("/system over server-sent events", () => { | |
| 16 | let dataDir: string; | |
| 17 | let app: App; | |
| 18 | let servers: Servers; | |
| 19 | let cookie: string; | |
| 20 | const sampler = createLive({ pushMs: 20 }); | |
| 21 | ||
| 22 | beforeAll(async () => { | |
| 23 | dataDir = await mkdtemp(join(tmpdir(), "forge-sysstream-")); | |
| 24 | app = createApp({ dataDir, port: 0 }); | |
| 25 | servers = await serve(app); | |
| 26 | const u = await createUser(app.db, { | |
| 27 | origin: "seed", trusted: true, username: "alice", email: "[email protected]", | |
| 28 | password: "a-good-passphrase", adminIfFirst: true, | |
| 29 | }); | |
| 30 | if (!u.ok) throw new Error(u.error); | |
| 31 | await createUser(app.db, { | |
outdated
|
||
| 32 | origin: "seed", trusted: true, username: "bob", email: "[email protected]", password: "a-good-passphrase", | |
| 33 | }); | |
| 34 | useLive(sampler); | |
| 35 | const res = await fetch(`${servers.url}/signin`, { | |
| 36 | method: "POST", | |
| 37 | redirect: "manual", | |
| 38 | headers: { "content-type": "application/x-www-form-urlencoded", origin: servers.url }, | |
| 39 | body: new URLSearchParams({ username: "alice", password: "a-good-passphrase" }), | |
| 40 | }); | |
| 41 | cookie = (res.headers.get("set-cookie") ?? "").split(";")[0] ?? ""; | |
| 42 | }); | |
| 43 | ||
| 44 | afterAll(async () => { | |
| 45 | useLive(createLive()); | |
| 46 | servers.stop(); | |
| 47 | app.close(); | |
| 48 | await rm(dataDir, { recursive: true, force: true }); | |
| 49 | }); | |
| 50 | ||
| 51 | test("the page has no refresh and no interval reload, and opens the stream", async () => { | |
| 52 | const res = await fetch(`${servers.url}/system`, { headers: { cookie } }); | |
| 53 | expect(res.status).toBe(200); | |
| 54 | const body = await res.text(); | |
| 55 | expect(body).not.toMatch(/http-equiv=["']?refresh/i); | |
| 56 | expect(body).not.toContain("location.reload"); | |
| 57 | expect(body).not.toContain("setTimeout"); | |
| 58 | expect(body).not.toContain("setInterval"); | |
| 59 | expect(body).toContain("new EventSource('/system/stream')"); | |
| 60 | // First paint is server-rendered with the latest sample. | |
| 61 | expect(body).toContain('id="summary"'); | |
| 62 | expect(body).toMatch(/id="summary"[^>]*>\s*<[^>]*>CPU /); | |
| 63 | }); | |
| 64 | ||
| 65 | test("a sample written to the sampler arrives on the stream; a stranger is refused", async () => { | |
| 66 | const anon = await fetch(`${servers.url}/system/stream`, { redirect: "manual" }); | |
| 67 | expect(anon.headers.get("content-type") ?? "").not.toContain("text/event-stream"); | |
| 68 | await anon.body?.cancel(); | |
| 69 | const res = await fetch(`${servers.url}/system/stream`, { headers: { cookie } }); | |
| 70 | expect(res.headers.get("content-type")).toContain("text/event-stream"); | |
| 71 | const reader = res.body!.getReader(); | |
| 72 | const decoder = new TextDecoder(); | |
| 73 | const base = sampler.read(); | |
| 74 | sampler.publish({ ...base, processes: 98765 }); | |
| 75 | let text = ""; | |
| 76 | const deadline = Date.now() + 3000; | |
| 77 | while (!text.includes("98765") && Date.now() < deadline) { | |
| 78 | const { value, done } = await reader.read(); | |
| 79 | if (done) break; | |
| 80 | text += decoder.decode(value); | |
| 81 | } | |
| 82 | await reader.cancel(); | |
| 83 | expect(text).toContain("event: sample"); | |
| 84 | expect(text).toContain("98765 processes"); | |
| 85 | }); | |
| 86 | ||
outdated
This shows that an unsubscribed listener is not called. It does not show that the interval stops: the test would pass with the |
||
| 87 | test("a signed-in non-owner is refused the stream as the page is", async () => { | |
| 88 | const res = await fetch(`${servers.url}/signin`, { | |
| 89 | method: "POST", | |
| 90 | redirect: "manual", | |
| 91 | headers: { "content-type": "application/x-www-form-urlencoded", origin: servers.url }, | |
| 92 | body: new URLSearchParams({ username: "bob", password: "a-good-passphrase" }), | |
| 93 | }); | |
| 94 | const bob = (res.headers.get("set-cookie") ?? "").split(";")[0] ?? ""; | |
| 95 | expect(bob).not.toBe(""); | |
| 96 | const stream = await fetch(`${servers.url}/system/stream`, { headers: { cookie: bob } }); | |
| 97 | expect(stream.status).toBe(404); | |
| 98 | expect(stream.headers.get("content-type") ?? "").not.toContain("text/event-stream"); | |
| 99 | await stream.body?.cancel(); | |
| 100 | }); | |
| 101 | ||
| 102 | test("the sampler's clock runs only while someone watches, and ticks while they do", async () => { | |
| 103 | const own = createLive({ pushMs: 10 }); | |
| 104 | expect(own.running()).toBe(false); | |
| 105 | const first = own.subscribe(() => {}); | |
| 106 | expect(own.running()).toBe(true); | |
| 107 | first(); | |
| 108 | expect(own.running()).toBe(false); | |
| 109 | let ticks = 0; | |
| 110 | const second = own.subscribe(() => ticks++); | |
| 111 | expect(own.running()).toBe(true); | |
| 112 | await Bun.sleep(100); | |
| 113 | second(); | |
| 114 | expect(ticks).toBeGreaterThan(2); | |
| 115 | const stopped = ticks; | |
| 116 | await Bun.sleep(60); | |
| 117 | expect(ticks).toBe(stopped); | |
| 118 | expect(own.running()).toBe(false); | |
| 119 | }); | |
| 120 | ||
| 121 | test("a reader that never reads stops being queued to once half full, and is not dropped", async () => { | |
| 122 | let emitter: ((e: string, d: unknown) => void) | undefined; | |
| 123 | let ready: (() => boolean) | undefined; | |
| 124 | let closed = false; | |
| 125 | const res = eventStream(new Request("http://x/s"), (emit, _close, r) => { | |
| 126 | emitter = emit; | |
| 127 | ready = r; | |
| 128 | return () => { | |
| 129 | closed = true; | |
| 130 | }; | |
| 131 | }); | |
| 132 | const dropped = sseDroppedCount(); | |
| 133 | const frame = "x".repeat(8 * 1024); | |
| 134 | let sent = 0; | |
| 135 | while (ready!() && sent < 1000) { | |
| 136 | emitter!("sample", frame); | |
| 137 | sent++; | |
| 138 | } | |
| 139 | // Gated: it stopped well short of the 256 KB budget (about 16 frames), without a drop. | |
| 140 | expect(sent).toBeGreaterThan(5); | |
| 141 | expect(sent).toBeLessThan(25); | |
| 142 | expect(ready!()).toBe(false); | |
| 143 | expect(closed).toBe(false); | |
| 144 | expect(sseDroppedCount()).toBe(dropped); | |
| 145 | // Reading frees room again. | |
| 146 | const reader = res.body!.getReader(); | |
| 147 | for (let i = 0; i < sent && !ready!(); i++) await reader.read(); | |
| 148 | expect(ready!()).toBe(true); | |
| 149 | await reader.cancel(); | |
| 150 | }); | |
| 151 | }); |
new above
offline
This is new shared behaviour with no test. Add one where the reader never reads: frames gated on
ready()stop being queued past half the budget, and the stream is not dropped for it.