From aa229386a9ef128b5f7f6cc15fc5102c1716dee0 Mon Sep 17 00:00:00 2001 From: Jake Gaylor Date: Sun, 6 Sep 2026 23:55:57 -0400 Subject: [PATCH] fix: bound and share live reads across requests and polling --- README.md | 25 +++++- src/chant-read-integration.test.ts | 120 +++++++++++++++++++++++++++++ src/chant.test.ts | 4 +- src/chant.ts | 94 +++++++++++++++++----- src/estate.ts | 13 +--- src/member-ir.ts | 55 ++----------- src/member-source.ts | 38 +++++++++ src/poll.test.ts | 15 ++++ src/poll.ts | 18 ++++- src/read-scheduler.test.ts | 100 ++++++++++++++++++++++++ src/read-scheduler.ts | 112 +++++++++++++++++++++++++++ src/server.ts | 20 ++++- web/app.js | 21 ++--- web/refresh-queue.js | 26 +++++++ web/refresh-queue.test.js | 54 +++++++++++++ 15 files changed, 617 insertions(+), 98 deletions(-) create mode 100644 src/chant-read-integration.test.ts create mode 100644 src/member-source.ts create mode 100644 src/read-scheduler.test.ts create mode 100644 src/read-scheduler.ts create mode 100644 web/refresh-queue.js create mode 100644 web/refresh-queue.test.js diff --git a/README.md b/README.md index 9505bca..4884fc8 100644 --- a/README.md +++ b/README.md @@ -264,9 +264,32 @@ graph updates, no reload. Add `--poll ` (with `--env`) to also re-query li drift on an interval and push updates when a node's status changes: ```sh -behold serve ./infra --env prod --poll 30 # watch source + poll drift every 30s +behold serve ./infra --env prod --poll 30 # wait 30s between completed drift sweeps ``` +**Read budget.** HTTP observations, background polling, and frame captures share +one Chant subprocess budget: two reads at once by default (one on a one-CPU +host). `BEHOLD_ESTATE_CONCURRENCY` overrides that process-wide limit. Up to 64 +distinct reads can wait; excess reads fail explicitly rather than growing an +unbounded queue. Identical in-flight reads share work only when the project, +resolved Chant, source stamp, argv, and effective environment match. Completed +live results are not cached. Source watcher invalidation also separates new +requests from work begun before an edit. + +A running read has a 180-second deadline, configurable with +`BEHOLD_READ_TIMEOUT_MS`. A disconnected GET releases its subscription; when no +callers remain, the read is canceled. On Unix, cancellation stops the whole +npm/tsx/Node process group, escalating from TERM to KILL after one second. On +Windows only the direct child is terminated. Delegated writes retain their +existing lifecycle and are never deduplicated as reads. + +The browser runs one refresh at a time and collapses intervening notifications +into one follow-up. It no longer starts additional reads at 3/8/15-second offsets. +Polling uses the same estate namespace bindings as the HTTP overlay and reuses +the primary member's observation for lanes capture. Slow sweeps extend the poll +period; they do not overlap the next sweep. These bounds control duplicate work; +they do not eliminate Chant's underlying live discovery cost. + behold shells the **project's own** chant (resolved from the project's `node_modules` first), so the project decides the chant version — pin it to `@intentius/chant ^0.18.1` or later for the live overlay (`graph --live` observed diff --git a/src/chant-read-integration.test.ts b/src/chant-read-integration.test.ts new file mode 100644 index 0000000..ce7ed8b --- /dev/null +++ b/src/chant-read-integration.test.ts @@ -0,0 +1,120 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdtempSync, mkdirSync, writeFileSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { runChantRaw, isScheduledRead, resolveChant } from "./chant.ts"; +import { createApp } from "./server.ts"; +import { invalidateReadGeneration, withReadSignal } from "./read-scheduler.ts"; + +// Real subprocesses, no cluster. The fixture writes its instrumentation outside +// the member so observing it cannot accidentally change the source stamp. +let scratch: string, project: string, log: string; +beforeEach(() => { + scratch = mkdtempSync(join(tmpdir(), "behold-read-test-")); + project = join(scratch, "project"); log = join(scratch, "events"); + const pkg = join(project, "node_modules/@intentius/chant"); + mkdirSync(pkg, { recursive: true }); + writeFileSync(join(project, "source.ts"), "// fixture source\n"); + writeFileSync(join(pkg, "package.json"), JSON.stringify({ name: "@intentius/chant", version: "0.54.0", main: "chant.cjs", bin: { chant: "chant.cjs" } })); + writeFileSync(join(pkg, "chant.cjs"), `#!/usr/bin/env node +const fs = require('node:fs'); +const log = ${JSON.stringify(log)}; +const event = (value) => fs.appendFileSync(log, JSON.stringify(value) + '\\n'); +event({ event: 'start', pid: process.pid, args: process.argv.slice(2) }); +if (process.argv.includes('--tree')) { + const child = require('node:child_process').spawn(process.execPath, ['-e', + "process.on('SIGTERM', () => {}); require('node:fs').appendFileSync(" + JSON.stringify(log) + ", JSON.stringify({event:'child',pid:process.pid})+'\\\\n'); setInterval(() => {}, 1000);" + ], { stdio: 'ignore' }); + process.on('SIGTERM', () => process.exit(0)); + setInterval(() => {}, 1000); +} else { + setTimeout(() => { + event({ event: 'finish', pid: process.pid }); + process.stdout.write(JSON.stringify({ pid: process.pid, variant: process.env.READ_TEST_VARIANT })); + }, process.argv.includes('--slow') ? 1000 : 150); +} +`, { mode: 0o755 }); + expect(resolveChant(project).source).toBe("project"); + vi.stubEnv("BEHOLD_ESTATE_CONCURRENCY", "2"); +}); +afterEach(() => { vi.unstubAllEnvs(); rmSync(scratch, { recursive: true, force: true }); }); +const events = (): { event: string; pid: number; args?: string[] }[] => { + try { return readFileSync(log, "utf8").trim().split("\n").map((s) => JSON.parse(s)); } catch { return []; } +}; + +describe("real Chant read scheduling", () => { + it("matching HTTP and background reads share a process, then the next read is fresh", async () => { + const args = ["graph", "source.ts", "--live", "--namespace", "app"]; + const first = withReadSignal(new AbortController().signal, () => runChantRaw(args, project)); + const second = runChantRaw(args, project); + const [a, b] = await Promise.all([first, second]); + expect(a.stdout).toBe(b.stdout); + expect(events().filter((e) => e.event === "start")).toHaveLength(1); + a.stdout = "caller mutation"; + expect(b.stdout).not.toBe(a.stdout); + const fresh = await runChantRaw(args, project); + expect(fresh.stdout).not.toBe(b.stdout); + }); + + it("the HTTP middleware isolates cancellation between matching requests", async () => { + const app = createApp({ projectDir: project, port: 0 }); + app.onError((error, c) => c.json({ error: error.message }, 500)); + const caller = new AbortController(); + const first = app.request(new Request("http://localhost/api/diff?env=home", { signal: caller.signal })); + const second = app.request("/api/diff?env=home"); + await vi.waitFor(() => expect(events().filter((e) => e.event === "start")).toHaveLength(1)); + caller.abort(new Error("client disconnected")); + expect((await first).status).toBe(500); + expect((await second).status).toBe(200); + expect(events().filter((e) => e.event === "start")).toHaveLength(1); + }); + + it("namespace, effective environment, source edits, and watcher invalidation isolate reads", async () => { + const args = ["graph", "source.ts", "--live"]; + const reads = [runChantRaw(args, project)]; + reads.push(runChantRaw([...args, "--namespace", "other"], project)); + reads.push(runChantRaw(args, project, { READ_TEST_VARIANT: "other" })); + writeFileSync(join(project, "source.ts"), "// changed while reading\n"); + reads.push(runChantRaw(args, project)); + invalidateReadGeneration(); + reads.push(runChantRaw(args, project)); + const results = await Promise.all(reads); + expect(new Set(results.map((r) => JSON.parse(r.stdout).pid)).size).toBe(5); + let active = 0, peak = 0; + for (const e of events()) { active += e.event === "start" ? 1 : -1; peak = Math.max(peak, active); } + expect(peak).toBe(2); + }); + + it("a timed-out process releases the shared budget", async () => { + vi.stubEnv("BEHOLD_READ_TIMEOUT_MS", "50"); + await expect(runChantRaw(["graph", "source.ts", "--slow"], project)).rejects.toThrow("exceeded 50ms"); + vi.stubEnv("BEHOLD_READ_TIMEOUT_MS", "3000"); + expect((await runChantRaw(["graph", "source.ts"], project)).code).toBe(0); + }); + + it.skipIf(process.platform === "win32")("disconnect kills the whole worker group, including a descendant ignoring TERM", async () => { + const caller = new AbortController(); + const read = withReadSignal(caller.signal, () => runChantRaw(["graph", "source.ts", "--tree"], project)); + // Handle rejection immediately; disconnect rejects the subscriber before + // the worker group has necessarily finished closing its pipes. + const rejected = read.catch((error) => error); + let child = 0; + try { + await vi.waitFor(() => { child = events().find((e) => e.event === "child")?.pid ?? 0; expect(child).toBeGreaterThan(0); }); + caller.abort(new Error("client left")); + expect((await rejected).message).toBe("client left"); + await vi.waitFor(() => expect(() => process.kill(child, 0)).toThrow(), { timeout: 3000 }); + } finally { + caller.abort(new Error("client left")); + for (const e of events()) { try { process.kill(e.pid, "SIGKILL"); } catch {} } + } + }); + + it("delegated mutations are never scheduled or deduplicated as reads", async () => { + for (const args of [["approve", "op", "gate"], ["run", "apply"], ["emulator", "up"], ["carve", "emit"], ["build"]]) { + expect(isScheduledRead(args)).toBe(false); + } + const results = await Promise.all([runChantRaw(["approve", "op", "gate"], project), runChantRaw(["approve", "op", "gate"], project)]); + expect(results[0].stdout).not.toBe(results[1].stdout); + }); +}); diff --git a/src/chant.test.ts b/src/chant.test.ts index a9a35e4..47541ca 100644 --- a/src/chant.test.ts +++ b/src/chant.test.ts @@ -562,11 +562,11 @@ describe("applyArgs", () => { describe("runChantRaw — env override reaches the spawn", () => { beforeEach(() => vi.mocked(spawnMock).mockReset()); - it("spawns with no explicit `env` option when no override is given — inherits process.env as before", async () => { + it("snapshots the inherited environment before a read can queue", async () => { vi.mocked(spawnMock).mockReturnValue(fakeProc(0, "{}")); await runChantRaw(["graph", "src", "--format", "ir"], "/proj"); const opts = vi.mocked(spawnMock).mock.calls[0]![2] as { env?: unknown } | undefined; - expect(opts?.env).toBeUndefined(); + expect(opts?.env).toEqual(process.env); }); it("merges the env override over process.env for exactly this spawn (M2 tier/target lenses)", async () => { diff --git a/src/chant.ts b/src/chant.ts index ca078c7..18b5a7b 100644 --- a/src/chant.ts +++ b/src/chant.ts @@ -26,6 +26,9 @@ import { detectProject } from "./project.ts"; // (package.json `exports["./yaml"]` has no compiled-JS condition), which a // plain `node dist/cli.js` cannot import unbundled (Node refuses to // type-strip files under node_modules). +import { createHash } from "node:crypto"; +import { memberSourceStamp } from "./member-source.ts"; +import { ReadScheduler, currentReadSignal, readGeneration } from "./read-scheduler.ts"; import { parseYAML } from "@intentius/chant/yaml"; import { carveStatusArgs, type CarveStatusJson } from "./carve-manifest.ts"; import { dropForeignDeclarations } from "./foreign.ts"; @@ -302,41 +305,94 @@ export interface ChantRun { } /** Run the chant bin, capturing stdout/stderr and the exit code. Never rejects on - * a non-zero exit (only on a spawn failure) — a failing exit is data. + * a non-zero exit — a failing exit is data. Scheduled reads can also reject + * on queue saturation, cancellation, or deadline expiry. * `envOverride` (M2, #54: the tier/target lenses' `envOverridesFor`) merges over * `process.env` for this one spawn only — never a global mutation, so a picked * lens on one request can't bleed into a concurrent request on another. */ +// Positive read allowlist: delegated mutations (including build/carve emit) must +// never be coalesced or canceled by the observation scheduler. +export function isScheduledRead(args: string[]): boolean { + return args[0] === "graph" || + (args[0] === "components" && args[1] === "status") || + (args[0] === "lifecycle" && ["diff", "plan"].includes(args[1])) || + (args[0] === "helm" && ["renders", "diff"].includes(args[1])) || + (args[0] === "run" && args[1] === "status") || + (args[0] === "operator" && ["status", "log"].includes(args[1])); +} +const reads = new ReadScheduler(); +let unstampableRead = 0; + export function runChantRaw( args: string[], projectDir?: string, envOverride?: Record, ): Promise { + const chant = resolveChant(projectDir); + if (!isScheduledRead(args)) return spawnChant(chant.bin, args, projectDir, envOverride); + const dir = resolve(projectDir ?? process.cwd()); + const stamp = memberSourceStamp(dir); + const effectiveEnv = { ...process.env, ...envOverride }; + const environment = Object.entries(effectiveEnv).sort(([a], [b]) => a.localeCompare(b)); + // Include the resolved compiler, exact argv (namespace/lens/env included), + // source identity and effective environment. Unreadable source cannot share. + const key = createHash("sha256").update(JSON.stringify([ + dir, chant.bin, chant.version, args, environment, readGeneration(), stamp ?? ++unstampableRead, + ])).digest("hex"); + return reads.read(key, (signal) => spawnChant(chant.bin, args, projectDir, effectiveEnv, signal), currentReadSignal()) + .then((result) => ({ ...result })); // callers own their result; JSON is parsed separately +} + +function spawnChant( + bin: string, + args: string[], + projectDir?: string, + envOverride?: NodeJS.ProcessEnv, + signal?: AbortSignal, +): Promise { + if (signal?.aborted) return Promise.reject(signal.reason); return new Promise((resolvePromise, reject) => { - // Run in the project dir: `chant graph --live` reads the current working - // directory (not the path arg), so the cwd must be the project for the live - // and overlay paths to observe the right environment. - const proc = spawn(chantBin(projectDir), args, { + // Read workers own a process group: killing only npx leaves tsx/Node alive. + // Writes retain their existing lifetime and process behavior. + const grouped = !!signal && process.platform !== "win32"; + const proc = spawn(bin, args, { ...(projectDir ? { cwd: projectDir } : {}), - ...(envOverride ? { env: { ...process.env, ...envOverride } } : {}), + ...(envOverride ? { env: signal ? envOverride : { ...process.env, ...envOverride } } : {}), + ...(grouped ? { detached: true } : {}), stdio: ["ignore", "pipe", "pipe"], }); - // Accumulate raw Buffer chunks and decode once at the end. Coercing each - // chunk to a string as it arrives (`s += d`) corrupts a multi-byte UTF-8 - // character that straddles a chunk boundary — which for loomster's ~200KB - // entity-graph IR reliably mangles the JSON near the 64KB highWaterMark and - // makes `JSON.parse` throw. Concatenating bytes first avoids the split. const outChunks: Buffer[] = []; const errChunks: Buffer[] = []; + let escalation: ReturnType | undefined; + const kill = (name: NodeJS.Signals): void => { + try { + if (grouped && proc.pid) process.kill(-proc.pid, name); + else proc.kill(name); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ESRCH") proc.kill(name); + } + }; + const abort = (): void => { + kill("SIGTERM"); + escalation = setTimeout(() => kill("SIGKILL"), 1000); + }; + signal?.addEventListener("abort", abort, { once: true }); + const cleanup = (): void => { + signal?.removeEventListener("abort", abort); + clearTimeout(escalation); + // A wrapper can exit before a descendant that ignores TERM. Finish the + // group before releasing its budget, even if its pipes already closed. + if (signal?.aborted && grouped) kill("SIGKILL"); + }; + // Decode once: a UTF-8 character can straddle stdout chunks. proc.stdout.on("data", (d: Buffer) => outChunks.push(d)); proc.stderr.on("data", (d: Buffer) => errChunks.push(d)); - proc.on("error", reject); - proc.on("close", (code) => - resolvePromise({ - code: code ?? 1, - stdout: Buffer.concat(outChunks).toString("utf8"), - stderr: Buffer.concat(errChunks).toString("utf8"), - }), - ); + proc.on("error", (error) => { cleanup(); reject(error); }); + proc.on("close", (code) => { + cleanup(); + if (signal?.aborted) { reject(signal.reason); return; } + resolvePromise({ code: code ?? 1, stdout: Buffer.concat(outChunks).toString("utf8"), stderr: Buffer.concat(errChunks).toString("utf8") }); + }); }); } diff --git a/src/estate.ts b/src/estate.ts index 136fa84..c6117a8 100644 --- a/src/estate.ts +++ b/src/estate.ts @@ -21,7 +21,7 @@ import { joinCarvedSources } from "./carve-manifest.ts"; import { carveStatesFor } from "./carve-discovery.ts"; import { statSync } from "node:fs"; -import { availableParallelism } from "node:os"; +import { readConcurrency } from "./read-scheduler.ts"; import { join, resolve, sep } from "node:path"; import { composeStacks, shortStackNames, type GraphIR } from "@intentius/pinhole"; import { graphIr, meetsFloor, resolveChant, type GraphOptions } from "./chant.ts"; @@ -41,15 +41,10 @@ import { CLUSTER_SCOPED } from "./zoom-notes.ts"; // host busy without drowning it. // --------------------------------------------------------------------------- -/** How many member chant processes one estate read runs at once: the host's - * parallelism capped at 4 (each spawn wants a core-plus for its TS eval), - * never more lanes than members. `BEHOLD_ESTATE_CONCURRENCY` overrides the - * cap for tuning a live estate; anything unparseable or < 1 is ignored - * rather than honoured into a stall. Exported for testing. */ +/** Per-composition pipelining; runChantRaw also enforces this budget across + * all simultaneous estate requests, background polls, and frame captures. */ export function estateReadPool(members: number, env: Record = process.env): number { - const override = Number.parseInt(env.BEHOLD_ESTATE_CONCURRENCY ?? "", 10); - const cap = Number.isInteger(override) && override >= 1 ? override : Math.min(4, availableParallelism()); - return Math.max(1, Math.min(cap, members)); + return Math.max(1, Math.min(readConcurrency(env), members)); } /** `Promise.all(items.map(fn))` with at most `width` calls in flight. diff --git a/src/member-ir.ts b/src/member-ir.ts index 80bf0e3..c353128 100644 --- a/src/member-ir.ts +++ b/src/member-ir.ts @@ -55,18 +55,13 @@ * with the process. * --------------------------------------------------------------------------- */ -import { createHash } from "node:crypto"; -import { readdirSync, statSync } from "node:fs"; -import { join, relative, resolve } from "node:path"; +import { resolve } from "node:path"; +import { memberSourceStamp } from "./member-source.ts"; +export { memberSourceStamp } from "./member-source.ts"; +import { invalidateReadGeneration } from "./read-scheduler.ts"; import type { GraphIR } from "@intentius/chant"; import { graphIr, resolveChant, type GraphOptions } from "./chant.ts"; -/** Directories a member's source stamp never walks: build output and installed - * packages are not the member's declared source, and a `node_modules` sweep is - * exactly the walk that would make stamping cost more than the spawn it saves - * (the same set `watchSource` ignores, src/events.ts). */ -const SKIP = new Set(["node_modules", "dist", ".git"]); - /** How many entries the cache holds before the least recently used are dropped. * A bound, not a tuning knob: #306 fixed an OOM and this must not reintroduce * one. An estate reads each member under a handful of option shapes, so a @@ -78,47 +73,6 @@ export function memberIrCacheSize(env: Record = proc return Number.isInteger(override) && override >= 0 ? override : 64; } -/** - * A fingerprint of everything a member declares on disk: each file's - * member-relative path, mtime and size, hashed. Undefined when the member - * cannot be walked at all — an unstampable member is never cached, which is the - * safe direction (a spawn, not a guess). - * - * The whole member root, not just the resolved graph source dir: `chant.config.ts` - * decides what the source dir even is, a multi-stack member graphs from several - * of them, and a member's `cluster/` build root is source too. Over-broad by - * design — a stray edit under the member costs one spawn, and the alternative - * (walking only what this read happens to graph) is a stamp that can miss the - * file that changed the answer. Exported for testing. - */ -export function memberSourceStamp(dir: string): string | undefined { - const root = resolve(dir); - const h = createHash("sha1"); - let any = false; - const walk = (at: string): void => { - // Sorted, so the same tree stamps the same however the filesystem enumerates it. - const entries = readdirSync(at, { withFileTypes: true }).sort((a, b) => (a.name < b.name ? -1 : 1)); - for (const e of entries) { - if (SKIP.has(e.name)) continue; - const path = join(at, e.name); - if (e.isDirectory()) { - walk(path); - continue; - } - if (!e.isFile()) continue; // sockets, fifos, dangling symlinks: nothing to stamp - const st = statSync(path); - h.update(`${relative(root, path)}\0${st.mtimeMs}\0${st.size}\n`); - any = true; - } - }; - try { - walk(root); - } catch { - return undefined; - } - return any ? h.digest("hex") : undefined; -} - /** The identity of one cached read: the member, the chant that would answer it, * and every graph option that reaches the invocation. `GraphOptions` keys are * sorted so two equal option sets built in different orders are one key. */ @@ -150,6 +104,7 @@ export function memberIrCacheStats(): { hits: number; misses: number; entries: n /** Drop `dir`'s entries (every option shape), or the whole cache when called * with nothing. Wired to the estate source watcher — see rule 5 above. */ export function invalidateMember(dir?: string): void { + invalidateReadGeneration(); if (dir === undefined) { cache.clear(); return; diff --git a/src/member-source.ts b/src/member-source.ts new file mode 100644 index 0000000..3725029 --- /dev/null +++ b/src/member-source.ts @@ -0,0 +1,38 @@ +// Shared source identity for completed source reads and in-flight live reads. +import { createHash } from "node:crypto"; +import { readdirSync, statSync } from "node:fs"; +import { join, relative, resolve } from "node:path"; +const SKIP = new Set(["node_modules", "dist", ".git"]); + +/** Hash the whole member root by relative path, mtime and size. Config can + * choose source outside src/, so narrowing the walk would miss declarations. + * Unreadable or empty roots return undefined and are never shared/cached. + * Installed dependencies and generated dist/ output are deliberately excluded; + * the caller separately keys on the resolved Chant compiler identity. */ +export function memberSourceStamp(dir: string): string | undefined { + const root = resolve(dir); + const h = createHash("sha1"); + let any = false; + const walk = (at: string): void => { + // Sorted, so the same tree stamps the same however the filesystem enumerates it. + const entries = readdirSync(at, { withFileTypes: true }).sort((a, b) => (a.name < b.name ? -1 : 1)); + for (const e of entries) { + if (SKIP.has(e.name)) continue; + const path = join(at, e.name); + if (e.isDirectory()) { + walk(path); + continue; + } + if (!e.isFile()) continue; // sockets, fifos, dangling symlinks: nothing to stamp + const st = statSync(path); + h.update(`${relative(root, path)}\0${st.mtimeMs}\0${st.size}\n`); + any = true; + } + }; + try { + walk(root); + } catch { + return undefined; + } + return any ? h.digest("hex") : undefined; +} diff --git a/src/poll.test.ts b/src/poll.test.ts index 888a300..0f74978 100644 --- a/src/poll.test.ts +++ b/src/poll.test.ts @@ -105,6 +105,21 @@ describe("startDriftPoll", () => { stop(); }); + it("prepares namespace scopes once per sweep and supplies the existing IR for capture", async () => { + const observed = ir([{ id: "svc", status: "good" }]); + let namespace = "old"; + const prepare = vi.fn(async () => { namespace = "app"; }); + const query = vi.fn(async () => { expect(namespace).toBe("app"); return observed; }); + const onRead = vi.fn(); + const stop = startDriftPoll({ intervalMs: 1000, beforeSweep: prepare, + members: [{ dir: "/app", query }], onRead, onChange: vi.fn() }); + await vi.advanceTimersByTimeAsync(2000); + expect(prepare).toHaveBeenCalledTimes(2); + expect(query).toHaveBeenCalledTimes(2); // capture costs no third/fourth query + expect(onRead).toHaveBeenCalledWith("/app", observed); + stop(); + }); + it("sweeps members sequentially — a slow read never stampedes the next member", async () => { let release!: (ir: GraphIR) => void; const slow = { dir: "/estate/a", query: vi.fn(() => new Promise((res) => (release = res))) }; diff --git a/src/poll.ts b/src/poll.ts index 32d9065..e51d83d 100644 --- a/src/poll.ts +++ b/src/poll.ts @@ -5,6 +5,7 @@ * `--live` is describe calls, not a cloud watch — the cadence is the drift * resolution. Off unless `--poll` is set. */ +import { withReadSignal } from "./read-scheduler.ts"; import type { GraphIR } from "@intentius/chant"; /** A stable fingerprint of the drift state: each node's id + its `_status` @@ -84,6 +85,10 @@ export interface DriftPollOptions { /** Fired per member whose drift moved, with that member's dir and the * substrates that moved (#117) — sorted, and never empty when called. */ onChange: (dir: string, movedLexicons: string[]) => void; + /** Refresh estate namespace bindings once before a sweep. */ + beforeSweep?: () => Promise; + /** Reuse the observation for frame capture instead of describing again. */ + onRead?: (dir: string, ir: GraphIR) => void; onError?: (dir: string, err: unknown) => void; } @@ -104,14 +109,24 @@ export interface DriftPollOptions { */ export function startDriftPoll(opts: DriftPollOptions): () => void { let stopped = false; + const controller = new AbortController(); const last = new Map>(); let timer: ReturnType; const tick = async (): Promise => { + try { await withReadSignal(controller.signal, () => opts.beforeSweep?.()); } + catch (err) { + opts.onError?.(opts.members[0]?.dir ?? "", err); + if (!stopped) timer = setTimeout(tick, opts.intervalMs); + return; + } for (const member of opts.members) { if (stopped) return; try { - const digests = driftDigestsByLexicon(await member.query()); + const ir = await withReadSignal(controller.signal, member.query); + if (stopped) return; + opts.onRead?.(member.dir, ir); + const digests = driftDigestsByLexicon(ir); const prev = last.get(member.dir); if (prev !== undefined) { const moved = changedLexicons(prev, digests); @@ -129,5 +144,6 @@ export function startDriftPoll(opts: DriftPollOptions): () => void { return () => { stopped = true; clearTimeout(timer); + controller.abort(new Error("Drift poll stopped")); }; } diff --git a/src/read-scheduler.test.ts b/src/read-scheduler.test.ts new file mode 100644 index 0000000..420471c --- /dev/null +++ b/src/read-scheduler.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, it } from "vitest"; +import { ReadScheduler, readConcurrency } from "./read-scheduler.ts"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + return { promise, resolve, reject }; +} + +describe("shared Chant read budget", () => { + it("caps distinct HTTP/poll/capture reads together and shares matching work only while pending", async () => { + const pool = new ReadScheduler(() => 2); + const gates = Array.from({ length: 5 }, () => deferred()); + const starts: number[] = []; + let active = 0, peak = 0; + const read = (i: number) => pool.read(String(i), async () => { + starts.push(i); peak = Math.max(peak, ++active); + try { return await gates[i].promise; } finally { active--; } + }); + const results = [read(0), read(1), read(2), read(3), read(4), read(0), read(2)]; + await Promise.resolve(); + expect(starts).toEqual([0, 1]); + gates.forEach((gate, i) => gate.resolve(i)); + expect(await Promise.all(results)).toEqual([0, 1, 2, 3, 4, 0, 2]); + expect(starts).toEqual([0, 1, 2, 3, 4]); + expect(peak).toBe(2); + expect(await pool.read("0", async () => 99)).toBe(99); + }); + + it("one disconnected caller does not cancel another subscriber", async () => { + const pool = new ReadScheduler(() => 1); + const gate = deferred(); + const caller = new AbortController(); + let worker!: AbortSignal; + const first = pool.read("same", async (signal) => { worker = signal; return gate.promise; }, caller.signal); + const second = pool.read("same", async () => { throw new Error("duplicate spawn"); }); + await Promise.resolve(); + caller.abort(new Error("gone")); + await expect(first).rejects.toThrow("gone"); + expect(worker.aborted).toBe(false); + gate.resolve(7); + expect(await second).toBe(7); + }); + + it("canceled queued work never starts; canceled running work holds its slot until cleanup finishes", async () => { + const pool = new ReadScheduler(() => 1); + const cleanup = deferred(); + const running = new AbortController(), queued = new AbortController(); + let worker!: AbortSignal, queuedStarted = false, replacementStarted = false; + const first = pool.read("first", async (signal) => { worker = signal; return cleanup.promise; }, running.signal); + const second = pool.read("queued", async () => { queuedStarted = true; return 2; }, queued.signal); + await Promise.resolve(); + queued.abort(new Error("queued gone")); running.abort(new Error("running gone")); + await expect(first).rejects.toThrow("running gone"); + await expect(second).rejects.toThrow("queued gone"); + expect(worker.aborted).toBe(true); + const replacement = pool.read("first", async () => { replacementStarted = true; return 3; }); + await Promise.resolve(); + expect(replacementStarted).toBe(false); + cleanup.resolve(1); + expect(await replacement).toBe(3); + expect(queuedStarted).toBe(false); + }); + + it("a failed read releases the slot and is not retained", async () => { + const pool = new ReadScheduler(() => 1); + await expect(pool.read("failed", () => { throw new Error("failure"); })).rejects.toThrow("failure"); + expect(await pool.read("failed", async () => 2)).toBe(2); + }); + + it("bounds queued distinct work but still accepts subscribers to existing work", async () => { + const pool = new ReadScheduler(() => 1, () => 1000, 1); + const gate = deferred(); + const first = pool.read("one", () => gate.promise); + const second = pool.read("two", async () => 2); + const shared = pool.read("two", async () => 999); + await expect(pool.read("three", async () => 3)).rejects.toThrow("queue is full"); + gate.resolve(1); + expect(await Promise.all([first, second, shared])).toEqual([1, 2, 2]); + }); + + it("a deadline cancels the worker and allows the next read after it stops", async () => { + const pool = new ReadScheduler(() => 1, () => 10); + const first = pool.read("slow", (signal) => new Promise((_, reject) => { + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + })); + const next = pool.read("next", async () => 2); + await expect(first).rejects.toThrow("exceeded 10ms"); + expect(await next).toBe(2); + }); + + it("uses a conservative default and rejects malformed concurrency settings", () => { + expect(readConcurrency({})).toBeLessThanOrEqual(2); + expect(readConcurrency({ BEHOLD_ESTATE_CONCURRENCY: "3" })).toBe(3); + for (const value of ["0", "-1", "NaN", "2garbage", "1.5"]) { + expect(readConcurrency({ BEHOLD_ESTATE_CONCURRENCY: value })).toBe(readConcurrency({})); + } + }); +}); diff --git a/src/read-scheduler.ts b/src/read-scheduler.ts new file mode 100644 index 0000000..a95678b --- /dev/null +++ b/src/read-scheduler.ts @@ -0,0 +1,112 @@ +import { AsyncLocalStorage } from "node:async_hooks"; +import { availableParallelism } from "node:os"; + +// Request-local cancellation never becomes part of a shared read's identity. +const signals = new AsyncLocalStorage(); +export const withReadSignal = (signal: AbortSignal, fn: () => T): T => signals.run(signal, fn); +export const currentReadSignal = (): AbortSignal | undefined => signals.getStore(); +let generation = 0; +export const invalidateReadGeneration = (): void => { generation++; }; +export const readGeneration = (): number => generation; + +/** The existing estate knob now caps ALL simultaneous Chant reads in a process. */ +export function readConcurrency(env: NodeJS.ProcessEnv = process.env): number { + const override = Number(env.BEHOLD_ESTATE_CONCURRENCY); + return Number.isInteger(override) && override > 0 ? override : Math.min(2, availableParallelism()); +} + +export function readTimeoutMs(env: NodeJS.ProcessEnv = process.env): number { + const override = Number(env.BEHOLD_READ_TIMEOUT_MS); + return Number.isInteger(override) && override > 0 ? override : 180_000; +} + +interface Job { + key: string; + run: (signal: AbortSignal) => Promise; + controller: AbortController; + promise: Promise; + resolve: (result: T) => void; + reject: (error: unknown) => void; + users: number; + started: boolean; +} + +/** FIFO process-wide budget with in-flight sharing, never a completed-result cache. + * Each subscriber can leave independently. A slot stays occupied until the task + * has actually stopped, even when all subscribers have gone or its deadline hit. */ +export class ReadScheduler { + private active = 0; + private pending = new Map>(); + private queue: Job[] = []; + + constructor( + private readonly width = readConcurrency, + private readonly timeout = readTimeoutMs, + private readonly queueLimit = 64, + ) {} + + read(key: string, run: (signal: AbortSignal) => Promise, signal?: AbortSignal): Promise { + if (signal?.aborted) return Promise.reject(signal.reason); + let job = this.pending.get(key); + if (job?.controller.signal.aborted) { this.pending.delete(key); job = undefined; } + if (!job) { + if (this.active >= this.width() && this.queue.length >= this.queueLimit) { + return Promise.reject(new Error("Chant read queue is full; retry after current reads finish")); + } + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + job = { key, run, promise, resolve, reject, controller: new AbortController(), users: 0, started: false }; + this.pending.set(key, job); + this.queue.push(job); + } + const shared = job; + shared.users++; + const result = new Promise((resolve, reject) => { + let done = false; + const finish = (error: unknown, value?: T, failed = false): void => { + if (done) return; + done = true; + signal?.removeEventListener("abort", abort); + shared.users--; + if (failed) reject(error); else resolve(value as T); + }; + const abort = (): void => { + finish(signal?.reason, undefined, true); + if (shared.users === 0) { + if (this.pending.get(key) === shared) this.pending.delete(key); + shared.controller.abort(signal?.reason); + if (!shared.started) { + this.queue = this.queue.filter((item) => item !== shared); + shared.reject(signal?.reason); + } + } + }; + signal?.addEventListener("abort", abort, { once: true }); + shared.promise.then((value) => finish(undefined, value), (error) => finish(error, undefined, true)); + }); + this.drain(); + return result; + } + + private drain(): void { + while (this.active < this.width() && this.queue.length) { + const job = this.queue.shift()!; + job.started = true; + this.active++; + const timeout = this.timeout(); + const timer = setTimeout(() => job.controller.abort(new Error(`Chant read exceeded ${timeout}ms`)), timeout); + const settle = (): void => { + clearTimeout(timer); + if (this.pending.get(job.key) === job) this.pending.delete(job.key); + this.active--; + this.drain(); + }; + // Defer invocation so even a synchronous throw releases the slot. + Promise.resolve().then(() => job.run(job.controller.signal)).then( + (value) => { settle(); job.resolve(value); }, + (error) => { settle(); job.reject(error); }, + ); + } + } +} diff --git a/src/server.ts b/src/server.ts index fd6e85b..6ac0d05 100644 --- a/src/server.ts +++ b/src/server.ts @@ -12,6 +12,7 @@ * delegated gated writes". */ import { Hono, type Context, type Next } from "hono"; +import { withReadSignal } from "./read-scheduler.ts"; import { resolveSubstrateTargets } from "./targets.ts"; import { loadKubeconfig, resolveK8sTarget, type K8sTarget } from "./k8s-target.ts"; import { streamSSE } from "hono/streaming"; @@ -128,7 +129,7 @@ import { OpRunner } from "./op-runner.ts"; import { detectSubstrates, projectLexicons } from "./substrates.ts"; import { pickAutoSyncOps, splitForgeRouted, suspendedByRollback, type AutoSyncMode } from "./autosync.ts"; import { sourceCommits, openRollbackBranches } from "./history.ts"; -import { composeEstate, composeEstateOverlay, estateMembers, withoutJoinedMembers } from "./estate.ts"; +import { composeEstate, composeEstateOverlay, estateNamespaceScopes, estateMembers, withoutJoinedMembers } from "./estate.ts"; import { addEstateMemberEdges } from "./estate-edges.ts"; import { invalidateMember, memberIr } from "./member-ir.ts"; import { carveStatesFor, carveStatesUnder } from "./carve-discovery.ts"; @@ -833,6 +834,10 @@ export function createApp( }), ): Hono { const app = new Hono(); + app.use("/api/*", async (c, next) => { + if (c.req.method === "GET") return withReadSignal(c.req.raw.signal, next); + await next(); + }); // Carve mode (#252) claims /api/graph, /api/project and friends before the // project-shaped handlers are registered — see carveRoutes. @@ -2946,7 +2951,7 @@ export async function startServer(cfg: ServerOptions): Promise { // whose overlay moved (#297) — attribution flows through to the now-line and // the rollback interlock. const onPollDrift = (dir: string, movedLexicons: string[]): void => { - onEstateChange(dir); + broadcaster.emit("changed", dir); if (autoSync === "off") return; void routeAutoSync(dir, movedLexicons); }; @@ -3014,16 +3019,25 @@ export async function startServer(cfg: ServerOptions): Promise { // graph`, which would fail once a second on a directory that isn't a project). const carve = !!cfg.carveReport; let stopWatch = carve ? () => {} : watchSources(cfg.projectDirs ?? [cfg.projectDir], onMemberSourceChange); + let pollScopes = new Map(); let stopPoll = !carve && cfg.env && cfg.pollSecs ? startDriftPoll({ intervalMs: cfg.pollSecs * 1000, + beforeSweep: async () => { + pollScopes = await estateNamespaceScopes(cfg.projectDirs ?? [cfg.projectDir], { env: cfg.env }); + }, + onRead: (dir, ir) => { + // Lanes currently records the primary only. Reuse its actual poll + // result, including the baseline, without an extra live invocation. + if (dir === cfg.projectDir && frames.capture(ir) !== null) broadcaster.emit("frames"); + }, // Per-member queries, swept sequentially inside the poll (#297): a // member read can take seconds (#295), so an estate-wide tick must // stretch, not stampede N describes at once. members: (cfg.projectDirs ?? [cfg.projectDir]).map((dir) => ({ dir, - query: () => graphIr(dir, { live: true, overlay: true, env: cfg.env }), + query: () => graphIr(dir, { live: true, overlay: true, env: cfg.env, ...(pollScopes.has(dir) ? { namespace: pollScopes.get(dir) } : {}) }), })), onChange: onPollDrift, onError: (dir, err) => diff --git a/web/app.js b/web/app.js index 483ad0f..1ebee1d 100644 --- a/web/app.js +++ b/web/app.js @@ -8,6 +8,7 @@ // floating control panel's chrome (panel.js — drag/snap/collapse/tabs, persisted // position), and the theme picker into the panel's View-tab slot (a stable element // renderPanelView never rewrites, so the select mounts once and survives re-renders). +import { createRefreshQueue } from "./refresh-queue.js"; import { initTheme, mountThemePicker, readableOn, colorForCategory, onThemeChange, getTokens } from "./theme.js"; import { addPanelTab, initPanel, setPanelTab, togglePanelCollapsed, isPanelCollapsed } from "./panel.js"; // #254: the carve walkthrough's stepper — everything it DECIDES is a pure @@ -3511,7 +3512,8 @@ function renderPreconditionError(body) { } // Fetch the current view (source graph, or the picked env's live overlay). -async function load(opts = {}) { +const load = createRefreshQueue(loadOnce); +async function loadOnce(opts = {}, isCurrent = () => true) { const meta = document.getElementById("meta"); // A background settle re-pull (post-apply) shouldn't flash the meta/overlay. if (!opts.quiet) { @@ -3583,6 +3585,7 @@ async function load(opts = {}) { } const res = await apiFetch(`${endpoint}?${q}`); const body = await res.json(); + if (!isCurrent()) return; // a newer view/event is pending; do not paint the old result if (!res.ok) { // #72: a classified precondition failure (lint gate, not installed, a // tier that needs credentials) gets the calmer entry/error card instead @@ -3603,13 +3606,13 @@ async function load(opts = {}) { autoZoomFallback = false; applyZoom("resources"); renderStatusbar(); - return load(opts); + return loadOnce(opts, isCurrent); } autoZoomFallback = false; render(body.ir, body.svg, body.meta); } catch (err) { // A background settle poll must not blow away a good graph on a transient error. - if (!opts.quiet) { + if (!opts.quiet && isCurrent()) { // Text, never innerHTML — `err.message` embeds chant's own stderr, which // is not ours to interpolate as markup (the sibling precondition card has // said so since #72; this branch had been left behind). @@ -3711,20 +3714,12 @@ initPickers(); // No backend in a static export → no live event stream; a no-op keeps the // `events.addEventListener(...)` wiring below harmless. const events = staticMode ? { addEventListener() {} } : new EventSource("/api/events"); -// Post-op settle re-pull: an apply's CLI can exit while the last stacks are still -// flipping to *_COMPLETE, so the immediate reload catches a few components mid- -// deploy ("all done, 3 still pending"). Quietly re-pull a couple more times so -// the graph lands on the final colours without a manual Re-check live. -let settleTimers = []; -function scheduleSettle() { - settleTimers.forEach(clearTimeout); - settleTimers = [3000, 8000, 15000].map((ms) => setTimeout(() => load({ quiet: true }), ms)); -} +// One notification requests one refresh. Further notifications during a slow +// read collapse to one follow-up; no wall-clock timers multiply estate scans. events.addEventListener("changed", () => { bulkDiffCache = null; // an op ran → per-node live state may have changed load(); loadSubstrates(); // a bring-up (or any op) finished → re-detect readiness - scheduleSettle(); }); // Substrate readiness (M5, #54): is each substrate the project needs actually diff --git a/web/refresh-queue.js b/web/refresh-queue.js new file mode 100644 index 0000000..ef65850 --- /dev/null +++ b/web/refresh-queue.js @@ -0,0 +1,26 @@ +// One active refresh and one pending refresh, regardless of how many events +// arrive. All callers wait for the latest requested view to finish loading. +export function createRefreshQueue(run) { + let active; + let pending; + let revision = 0; + return (options = {}) => { + revision++; + pending = { ...options, quiet: pending ? !!pending.quiet && !!options.quiet : !!options.quiet }; + if (!active) { + active = Promise.resolve().then(async () => { + try { + while (pending) { + const next = pending; + pending = undefined; + const current = revision; + await run(next, () => current === revision); + } + } finally { + active = undefined; + } + }); + } + return active; + }; +} diff --git a/web/refresh-queue.test.js b/web/refresh-queue.test.js new file mode 100644 index 0000000..aacff3d --- /dev/null +++ b/web/refresh-queue.test.js @@ -0,0 +1,54 @@ +import { describe, expect, it } from "vitest"; +import { createRefreshQueue } from "./refresh-queue.js"; + +function deferred() { + let resolve; + const promise = new Promise((done) => { resolve = done; }); + return { promise, resolve }; +} + +describe("browser refresh queue", () => { + it("a slow read plus an event burst makes one follow-up, never overlapping reads", async () => { + const first = deferred(), second = deferred(); + let active = 0, peak = 0; + const calls = [], current = []; + const load = createRefreshQueue(async (opts, isCurrent) => { + peak = Math.max(peak, ++active); + calls.push(opts); current.push(isCurrent); + await (calls.length === 1 ? first : second).promise; + active--; + }); + const initial = load(); + await Promise.resolve(); + const burst = Array.from({ length: 20 }, () => load({ quiet: true })); + load({ quiet: false }); load({ quiet: true }); + expect(calls).toHaveLength(1); + expect(current[0]()).toBe(false); // suppress the superseded response + first.resolve(); + await Promise.resolve(); await Promise.resolve(); + expect(calls).toHaveLength(2); + expect(calls[1].quiet).toBe(false); // foreground request wins + expect(current[1]()).toBe(true); + second.resolve(); + await Promise.all([initial, ...burst]); + expect(peak).toBe(1); + expect(calls).toHaveLength(2); + }); + + it("one change event makes exactly one read", async () => { + let calls = 0; + const load = createRefreshQueue(async () => { calls++; }); + await load(); + expect(calls).toBe(1); + await load(); // a later notification remains fresh + expect(calls).toBe(2); + }); + + it("a failed load does not wedge future refreshes", async () => { + let calls = 0; + const load = createRefreshQueue(async () => { if (++calls === 1) throw new Error("offline"); }); + await expect(load()).rejects.toThrow("offline"); + await load(); + expect(calls).toBe(2); + }); +});