diff --git a/apps/hq/src/hqScopes.ts b/apps/hq/src/hqScopes.ts index dbc774952d..0b8e761be0 100644 --- a/apps/hq/src/hqScopes.ts +++ b/apps/hq/src/hqScopes.ts @@ -6,6 +6,7 @@ import { HqNavigationProject, HqAttentionValue, hqScopeKey, + HQ_ATTENTION_HEALTH_KEY, type HqScope, type HqScopeFailure, type HqStreamMessage, @@ -139,6 +140,14 @@ const json = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); const readUsageAccess = Schema.decodeUnknownOption( Schema.Struct({ generation: Schema.Struct({ access: Schema.String }) }), ); +/** + * A journal's key: its scope's, and for a Mate's attention whether the reader takes health apart — + * the two are different journals of one scope, so neither reader sees the other's shape. + */ +const journalKey = (scope: HqScope): string => + scope.kind === "attention" && scope.health === "apart" + ? JSON.stringify([scope.kind, scope.projectId, HQ_ATTENTION_HEALTH_KEY]) + : hqScopeKey(scope); const readAttentionSource = Schema.decodeUnknownOption( Schema.Struct({ attention: Schema.NullOr(HqAttentionValue) }), ); @@ -326,7 +335,7 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => Queue.offerAllUnsafe(connection.queue, messages); for (const message of messages) { if (!("scope" in message)) continue; - const key = hqScopeKey(message.scope); + const key = journalKey(message.scope); if (message.type === "scope-error") connection.failures.set(key, message.code); else connection.failures.delete(key); } @@ -347,7 +356,7 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => for (const connection of connections) { if ( connection.userId === entry.userId && - connection.scopes.has(hqScopeKey(entry.scope)) + connection.scopes.has(journalKey(entry.scope)) ) offer(connection, messages); } @@ -644,7 +653,12 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => }); } else if (scope.kind === "attention") { const mate = all.get(scope.projectId); - if (mate !== undefined) values.push({ key: scope.projectId, value: mate }); + if (mate !== undefined && scope.health === "apart") { + // A health sample changes this value alone; the record goes out when it changes. + const { health, healthState, ...record } = mate; + values.push({ key: scope.projectId, value: record }); + values.push({ key: HQ_ATTENTION_HEALTH_KEY, value: { health, healthState } }); + } else if (mate !== undefined) values.push({ key: scope.projectId, value: mate }); // No report is not a deletion. return { values, removals: [] as HqRemoval[] }; } else if (scope.kind === "operation") { @@ -656,7 +670,7 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => values.push(...(yield* operations.read(entry.userId, scope.appId))); return { values, removals: [] as HqRemoval[] }; } else { - const key = hqScopeKey(scope); + const key = journalKey(scope); let slot = contents.get(key); if (slot === undefined) { slot = { one: Semaphore.makeUnsafe(1) }; @@ -845,7 +859,7 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => const demanded = (entry: Entry) => [...connections].some( (connection) => - connection.userId === entry.userId && connection.scopes.has(hqScopeKey(entry.scope)), + connection.userId === entry.userId && connection.scopes.has(journalKey(entry.scope)), ); const prune = () => { pruneIdleScopes( @@ -1241,7 +1255,7 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => let demandSequence = 0; const subscribe = (subscription: HqSubscription, demand: number) => Effect.gen(function* () { - const key = hqScopeKey(subscription.scope); + const key = journalKey(subscription.scope); if (wanted.get(key) !== demand) return; const id = json([userId, key]); let entry = journals.get(id); @@ -1383,14 +1397,14 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => const keys = request.scopes === undefined ? undefined - : new Set(request.scopes.map(hqScopeKey)); + : new Set(request.scopes.map(journalKey)); sourceFence.invalidate(); contents.clear(); overviewCache = undefined; for (const entry of journals.values()) { if ( entry.userId !== userId || - (keys !== undefined && !keys.has(hqScopeKey(entry.scope))) + (keys !== undefined && !keys.has(journalKey(entry.scope))) ) continue; entry.dirty = true; @@ -1398,7 +1412,7 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => yield* entry.one.withPermits(1)( Effect.gen(function* () { delete entry.failure; - connection.failures.delete(hqScopeKey(entry.scope)); + connection.failures.delete(journalKey(entry.scope)); if (demanded(entry)) yield* refreshUnlocked(entry); }), ); @@ -1408,7 +1422,7 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => case "subscribe": { const demanded: Array<{ subscription: HqSubscription; demand: number }> = []; for (const subscription of request.scopes) { - const key = hqScopeKey(subscription.scope); + const key = journalKey(subscription.scope); const usageScopes = new Set( [...connections] .filter((value) => value.userId === userId) @@ -1457,7 +1471,7 @@ export const hqScopesLayer = (build?: string, recheck = Duration.seconds(30)) => } case "unsubscribe": for (const scope of request.scopes) { - const key = hqScopeKey(scope); + const key = journalKey(scope); wanted.delete(key); connection.failures.delete(key); const held = journals.get(json([userId, key])); diff --git a/apps/hq/src/link.test.ts b/apps/hq/src/link.test.ts index 722351c7e9..e863ea4372 100644 --- a/apps/hq/src/link.test.ts +++ b/apps/hq/src/link.test.ts @@ -13,7 +13,12 @@ import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as Socket from "effect/socket/Socket"; -import type { HqAttentionScopeValue } from "@t3tools/shared/hqStream"; +import { + HQ_ATTENTION_HEALTH_KEY, + type HqAttentionHealthValue, + type HqAttentionScopeValue, + type HqScopeDelivery, +} from "@t3tools/shared/hqStream"; import { memoryStore, overviewOf, row } from "../test/harness/overviews.ts"; import { enrollMate, @@ -37,6 +42,22 @@ import { AutoUpdatePolicy } from "./autoUpdate.ts"; const toJson = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); +/** A strained Mate's sample at `revision` of its run, as its link sends it. */ +const healthAt = (revision: number) => ({ + source: { environmentId: "env-1", epoch: 1, incarnation: "run-1", revision }, + sampledAt: `2026-10-09T13:00:0${revision}.000Z`, + evidence: { + status: "strained", + severity: "warning", + resources: ["io"], + memory: null, + cpu: null, + io: { some: { avg10: 1.75, total: 2_792_695_561 + revision }, full: null }, + disk: null, + unavailable: [], + }, +}); + /** A running Core with one Mate enrolled, its link open and its first state read. */ const linked = Effect.gen(function* () { const core = yield* startCore(true); @@ -190,6 +211,98 @@ describe("a Mate's link", () => { }), ); + it.effect( + "sends each health sample to a reader that takes health apart as that sample alone, never the Mate's unchanged record again", + () => + Effect.gen(function* () { + const { call, owner, link, socket } = yield* linked; + const apart = { kind: "attention", projectId: "P_MATE", health: "apart" } as const; + const whole = { kind: "attention", projectId: "P_MATE" } as const; + const reader = yield* socket(`/api/structure/ws?ticket=${yield* ticketFor(call, owner)}`); + const older = yield* socket(`/api/structure/ws?ticket=${yield* ticketFor(call, owner)}`); + yield* scopeReset(reader, apart); + yield* scopeReset(older, whole); + const overview = overviewOf(); + yield* link.send({ type: "overview", full: true, overview }); + yield* nextScopeValue( + reader, + apart, + "P_MATE", + (value) => JSON.stringify(value.overview) === JSON.stringify(overview), + ); + for (const revision of [1, 2, 3]) { + const sample = healthAt(revision); + yield* link.send({ type: "health", health: sample }); + const frame = (yield* reader.takeWhere( + `health ${revision}`, + (message) => + message.type === "scope-values" && + ( + message.values as ReadonlyArray<{ key: string; value: HqAttentionHealthValue }> + ).some((value) => value.value.health?.sampledAt === sample.sampledAt), + )) as HqScopeDelivery; + assert.deepStrictEqual(frame.values, [ + { key: HQ_ATTENTION_HEALTH_KEY, value: { health: sample, healthState: "live" } }, + ]); + // A reader that does not ask takes the sample inside the whole record, as before. + const relayed = yield* nextScopeValue( + older, + whole, + "P_MATE", + (value) => value.health?.sampledAt === sample.sampledAt, + ); + assert.deepStrictEqual(relayed.overview, overview); + } + const resent = (yield* reader.collected).filter( + (message) => + message.type === "scope-values" && + (message.values as ReadonlyArray<{ key: string }>).some( + (value) => value.key === "P_MATE", + ), + ); + assert.deepStrictEqual(resent, []); + }), + ); + + it.effect( + "relays at most one health sample a window from a Mate that sends many, the latest", + () => + Effect.gen(function* () { + const { call, owner, link, socket } = yield* linked; + const apart = { kind: "attention", projectId: "P_MATE", health: "apart" } as const; + const reader = yield* socket(`/api/structure/ws?ticket=${yield* ticketFor(call, owner)}`); + yield* scopeReset(reader, apart); + yield* link.send({ type: "overview", full: true, overview: overviewOf() }); + const carries = (message: { readonly type: string }, revision?: number) => + message.type === "scope-values" && + ( + (message as HqScopeDelivery).values as ReadonlyArray<{ + key: string; + value: HqAttentionHealthValue; + }> + ).some( + (value) => + value.key === HQ_ATTENTION_HEALTH_KEY && + (revision === undefined || value.value.health?.source.revision === revision), + ); + const relayed = (revision: number) => + reader.takeWhere(`health ${revision}`, (message) => carries(message, revision)); + yield* link.send({ type: "health", health: healthAt(1) }); + yield* relayed(1); + // Thirty-three a second, as a Mate at its memory limit sent them. + for (const revision of [2, 3, 4, 5, 6, 7, 8, 9]) { + yield* link.send({ type: "health", health: healthAt(revision) }); + yield* Effect.sleep(Duration.millis(30)); + } + yield* relayed(9); + // Every sample between went by unrelayed: nothing else carried health before the latest. + assert.deepStrictEqual( + (yield* reader.collected).filter((message) => carries(message)), + [], + ); + }), + ); + it.effect("closes on an overview that does not decode", () => Effect.gen(function* () { const { link } = yield* linked; diff --git a/apps/hq/src/link.ts b/apps/hq/src/link.ts index 3743c71072..f17e2954c7 100644 --- a/apps/hq/src/link.ts +++ b/apps/hq/src/link.ts @@ -24,13 +24,16 @@ import { linkFrameBytes, readLinkUp, } from "@t3tools/shared/mateLink"; +import type { MateHealth } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as Socket from "effect/socket/Socket"; @@ -60,6 +63,8 @@ export interface LinkOptions { readonly pingEvery?: Duration.Duration; /** How often HQ checks that it still leads and the credential still holds; 30 s. */ readonly recheck?: Duration.Duration; + /** The window in which HQ relays at most one of a Mate's health samples, the latest; 2 s. */ + readonly healthEvery?: Duration.Duration; } /** Who the logins an overview names were last signed in by; a login nobody signed in names none. */ @@ -130,6 +135,18 @@ export const serveMateLink = ( if ((yield* Ref.getAndSet(sent, frame)) !== frame) yield* writer.write(frame); }); const heard = yield* Ref.make(yield* Clock.currentTimeMillis); + // One report at a time from this link: the relay below runs beside the frames it reads. + const reporting = yield* Semaphore.make(1); + const report = reporting.withPermits(1); + // A Mate's health reaches its readers at most once a window, the latest sample: an older + // Mate at its memory limit sends one on every kernel wake. + const latestHealth = yield* Queue.sliding(1); + const healthRelay = Effect.forever( + Queue.take(latestHealth).pipe( + Effect.flatMap((health) => report(overviews.reportHealth(projectId, link, health))), + Effect.andThen(Effect.sleep(options.healthEvery ?? Duration.seconds(2))), + ), + ); // Every change's tick, starting with the current one: its state at once, then as it moves. const states = Stream.runForEach( @@ -213,15 +230,15 @@ export const serveMateLink = ( continue; } if (read.kind === "message" && read.message.type === "health") { - yield* overviews.reportHealth(projectId, link, read.message.health); + yield* Queue.offer(latestHealth, read.message.health); continue; } if (read.kind === "message" && read.message.type === "attention") { - yield* overviews.reportAttention(projectId, link, read.message.attention); + yield* report(overviews.reportAttention(projectId, link, read.message.attention)); continue; } if (read.kind === "message" && read.message.type === "overview") { - yield* overviews.report(projectId, link, read.message); + yield* report(overviews.report(projectId, link, read.message)); // A write that fails is the next overview's to repeat; it never ends the link. yield* Effect.ignore( structure.recordSigners( @@ -278,7 +295,7 @@ export const serveMateLink = ( Effect.andThen(Effect.never), ); - yield* Effect.raceAll([listen, ping, states, recheck, accessFrames, capture]); + yield* Effect.raceAll([listen, ping, states, recheck, accessFrames, capture, healthRelay]); return yield* ending; }), ); diff --git a/apps/server/src/zerops/ZeropsMateAttention.ts b/apps/server/src/zerops/ZeropsMateAttention.ts index 1b563477b4..6550cfc7cb 100644 --- a/apps/server/src/zerops/ZeropsMateAttention.ts +++ b/apps/server/src/zerops/ZeropsMateAttention.ts @@ -53,7 +53,8 @@ import { OrchestrationEngineService } from "../orchestration/Services/Orchestrat import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts"; import { MateEngine } from "../engine/MateEngine.ts"; import { engineAttentionReads } from "./engineOverview.ts"; -import { resourceHealthChanges } from "./mateResourceHealth.ts"; +import { CPU_SAMPLE_WINDOW_USEC } from "./mateCpuHealth.ts"; +import { latestAtMostEvery, publishesAgain, resourceHealthChanges } from "./mateResourceHealth.ts"; import { mateAttentionOf } from "./zeropsAttentionValue.ts"; /** The most chats followed one by one between two reads; past it, all are read whole again. */ @@ -280,6 +281,9 @@ export const layer = Layer.effect( undefined, Stream.fromPubSub(healthDemand), ).pipe( + // One sample a CPU window at most, the latest, and only one that says something new. + latestAtMostEvery(CPU_SAMPLE_WINDOW_USEC / 1000), + Stream.changesWith((a, b) => !publishesAgain(a, b)), Stream.mapEffect((evidence) => Effect.map(Clock.currentTimeMillis, (ms) => ({ sampledAt: DateTime.formatIso(DateTime.makeUnsafe(ms)), diff --git a/apps/server/src/zerops/mateResourceHealth.test.ts b/apps/server/src/zerops/mateResourceHealth.test.ts index 1d35db8f25..4275d37975 100644 --- a/apps/server/src/zerops/mateResourceHealth.test.ts +++ b/apps/server/src/zerops/mateResourceHealth.test.ts @@ -9,6 +9,8 @@ import * as Queue from "effect/Queue"; import { vi } from "vite-plus/test"; import * as TestClock from "effect/testing/TestClock"; import { + latestAtMostEvery, + publishesAgain, resourceHealthChanges, readResourceHealth as readKernelResourceHealth, } from "./mateResourceHealth.ts"; @@ -441,9 +443,10 @@ it.effect("a tick during an event read advances CPU and memory together at the n const dir = yield* Effect.promise(() => fixture()); const samples = yield* Queue.unbounded(); let calls = 0; + // The tick's window is strained, so the Mate publishes it beside the memory strain it carries. const cpu = async () => ({ cpu: { some: { avg10: 0, total: ++calls }, full: null, window: null }, - strained: false, + strained: calls === 2, unavailable: [], }); yield* Effect.forkScoped( @@ -482,3 +485,122 @@ it.effect("a tick during an event read advances CPU and memory together at the n expect(recovered.resources).toEqual([]); }), ); + +describe("a Mate publishes a sample again only when its warning, or a number the warning quotes, changes", () => { + const quiet = { avg10: 0, total: 0 }; + const sample = (change: { + readonly status?: MateResourceHealth["status"]; + readonly severity?: MateResourceHealth["severity"]; + readonly resources?: MateResourceHealth["resources"]; + readonly memoryCurrent?: number; + readonly memoryHigh?: number; + readonly ioAvg10?: number; + readonly cpuTotal?: number; + readonly usageUsec?: number; + readonly elapsedUsec?: number; + readonly consumer?: { readonly pid: number; readonly name: string; readonly cpuCores: number }; + }): MateResourceHealth => ({ + status: change.status ?? "ok", + severity: change.severity ?? "warning", + resources: change.resources ?? [], + memory: { + current: change.memoryCurrent ?? 900, + high: change.memoryHigh ?? 1000, + max: 3000, + events: { high: 0, max: 0, oom: 0, oomKill: 0 }, + growth: { high: 0, max: 0, oom: 0, oomKill: 0 }, + pressure: { some: quiet, full: quiet }, + swapCurrent: 0, + swapMax: 0, + swapGrowth: 0, + }, + cpu: { + some: { avg10: 6.18, total: change.cpuTotal ?? 5_996_062_707 }, + full: quiet, + window: { + scope: "/sys/fs/cgroup", + elapsedUsec: change.elapsedUsec ?? 2_000_000, + usageUsec: change.usageUsec ?? 3_800_000, + someUsec: 155_374, + fullUsec: 3_894, + capacityCpus: 2, + throttledPeriods: 0, + saturated: true, + consumer: change.consumer ?? { pid: 42, name: "node", cpuCores: 1.5 }, + }, + }, + io: { some: { avg10: change.ioAvg10 ?? 1.75, total: 2_792_695_561 }, full: quiet }, + disk: { free: 263_085_883_392, total: 268_435_456_000 }, + unavailable: [], + }); + const io = { status: "strained", resources: ["io"] } as const; + const cpu = { status: "strained", resources: ["cpu"] } as const; + const again = "publishes again"; + const nothing = "publishes nothing"; + it.each([ + ["a healthy Mate whose usage moved", nothing, {}, { memoryCurrent: 950, cpuTotal: 1 }], + ["a healthy Mate whose RAM cap changed", again, {}, { memoryHigh: 2000 }], + [ + "an I/O-strained Mate whose numbers moved, none of them quoted,", + nothing, + io, + { ...io, ioAvg10: 9.5, memoryCurrent: 950, cpuTotal: 1, usageUsec: 1_000_000 }, + ], + [ + "an I/O-strained Mate now strained on CPU too", + again, + io, + { ...cpu, resources: ["io", "cpu"] }, + ], + ["an I/O-strained Mate that recovered", again, io, {}], + ["a strained Mate that turned critical", again, io, { ...io, severity: "critical" }], + ["a CPU-strained Mate whose measured use moved", again, cpu, { ...cpu, usageUsec: 3_000_000 }], + [ + "a CPU-strained Mate whose window took a different time", + again, + cpu, + { ...cpu, elapsedUsec: 2_100_000, usageUsec: 3_990_000 }, + ], + [ + "a CPU-strained Mate whose top process changed", + again, + cpu, + { ...cpu, consumer: { pid: 43, name: "vite", cpuCores: 1.5 } }, + ], + [ + "a CPU-strained Mate whose unquoted numbers moved, the quote the same to two decimals,", + nothing, + cpu, + { ...cpu, usageUsec: 3_800_400, ioAvg10: 9.5, memoryCurrent: 950, cpuTotal: 1 }, + ], + ] as const)("%s %s", (_case, verdict, previous, next) => { + expect(publishesAgain(sample(previous), sample(next))).toBe(verdict === again); + }); +}); + +it.effect( + "publishes at most one sample a window however often the kernel wakes it, and that the latest", + () => + Effect.gen(function* () { + const wakes = yield* Queue.unbounded(); + const published = yield* Queue.unbounded(); + yield* Effect.forkScoped( + Stream.fromQueue(wakes).pipe( + latestAtMostEvery(2000), + Stream.runForEach((value) => Queue.offer(published, value)), + ), + ); + yield* Queue.offerAll(wakes, [1]); + expect(yield* Queue.take(published)).toBe(1); + // Thirty-three wakes in the next second: one sample leaves, the last, when the window ends. + for (let value = 2; value <= 34; value += 1) { + yield* Queue.offer(wakes, value); + yield* TestClock.adjust(30); + } + expect(yield* Queue.size(published)).toBe(0); + yield* TestClock.adjust(2000); + expect(yield* Queue.take(published)).toBe(34); + yield* TestClock.adjust(4000); + expect(yield* Queue.size(published)).toBe(0); + }), +); diff --git a/apps/server/src/zerops/mateResourceHealth.ts b/apps/server/src/zerops/mateResourceHealth.ts index 47caf74c12..bdb354b9c8 100644 --- a/apps/server/src/zerops/mateResourceHealth.ts +++ b/apps/server/src/zerops/mateResourceHealth.ts @@ -5,6 +5,7 @@ import * as NodeFS from "node:fs"; import * as NodeChildProcess from "node:child_process"; import * as NodePath from "node:path"; import { + cpuWarningQuote, sustainedPressureLevel, type MateResourceHealth, type ResourcePressure, @@ -350,20 +351,46 @@ export function resourceHealthChanges( }; }), ), - Stream.changesWith((a, b) => { - const key = (value: MateResourceHealth) => - value.status === "strained" - ? value - : { - status: value.status, - unavailable: value.unavailable, - high: value.memory?.high, - max: value.memory?.max, - swapMax: value.memory?.swapMax, - }; - return JSON.stringify(key(a)) === JSON.stringify(key(b)); - }), + Stream.changesWith((a, b) => !publishesAgain(a, b)), ); }), ); } + +/** + * Whether a sample says something the last published one did not: the warning itself (its status, + * severity, strained resources, what could not be read, the RAM and swap caps) or a number it + * quotes. Only the CPU warning quotes numbers, so a Mate under CPU strain publishes each window + * whose quote moved and a Mate strained otherwise publishes only when its warning changes. + */ +export function publishesAgain(previous: MateResourceHealth, next: MateResourceHealth): boolean { + const said = (value: MateResourceHealth) => { + const window = value.cpu?.window; + return { + status: value.status, + severity: value.status === "strained" ? value.severity : null, + resources: value.resources, + unavailable: value.unavailable, + high: value.memory?.high, + max: value.memory?.max, + swapMax: value.memory?.swapMax, + quote: value.resources.includes("cpu") && window != null ? cpuWarningQuote(window) : null, + }; + }; + return JSON.stringify(said(previous)) !== JSON.stringify(said(next)); +} + +/** + * At most one value a window, the latest: the first leaves at once, and whatever arrives while a + * window runs is replaced by what follows it, the last leaving when the window ends. A Mate whose + * kernel wakes it many times a second (memory events at its limit) publishes once a window. + */ +export const latestAtMostEvery = + (millis: number) => + (self: Stream.Stream): Stream.Stream => + self.pipe( + Stream.buffer({ capacity: 1, strategy: "sliding" }), + Stream.flatMap((value) => + Stream.concat(Stream.make(value), Stream.drain(Stream.fromEffect(Effect.sleep(millis)))), + ), + ); diff --git a/packages/client-runtime/src/data/adapters/hq.test.ts b/packages/client-runtime/src/data/adapters/hq.test.ts index e81601d9ea..2bf599526d 100644 --- a/packages/client-runtime/src/data/adapters/hq.test.ts +++ b/packages/client-runtime/src/data/adapters/hq.test.ts @@ -4,7 +4,11 @@ import { hqMateHealthScope } from "../families/mateHealth.ts"; import { hqMateSetup } from "../projections/hqMateSetup.ts"; import { describe, expect, it } from "@effect/vitest"; import type { MateAttention } from "@t3tools/contracts"; -import type { HqScopeDelivery, HqStreamMessage } from "@t3tools/shared/hqStream"; +import { + HQ_ATTENTION_HEALTH_KEY, + type HqScopeDelivery, + type HqStreamMessage, +} from "@t3tools/shared/hqStream"; import * as Effect from "effect/Effect"; import * as Clock from "effect/Clock"; import * as Fiber from "effect/Fiber"; @@ -275,7 +279,7 @@ describe("hqNavigationLink", () => { knownKeys: ["app:shop", "project:ada"], }, // The Mate HQ places is observed with the navigation, across segments too. - { scope: { kind: "attention", projectId: "ada" }, knownKeys: [] }, + { scope: { kind: "attention", projectId: "ada", health: "apart" }, knownKeys: [] }, ], }, ]); @@ -412,7 +416,7 @@ describe("hqNavigationLink", () => { const store = makeAccountStore(AtomRegistry.make()); const fixture = hqFixtureWire(); const { fiber } = yield* live(store, fixture); - const attention = { kind: "attention", projectId: "ada" } as const; + const attention = { kind: "attention", projectId: "ada", health: "apart" } as const; expect(fixture.sent.at(-1)?.request).toEqual({ type: "subscribe", scopes: [{ scope: attention, knownKeys: [] }], @@ -563,7 +567,7 @@ describe("asking HQ on the open socket", () => { }); describe("a Mate's attention, relayed", () => { - const ATTENTION = { kind: "attention", projectId: "ada" } as const; + const ATTENTION = { kind: "attention", projectId: "ada", health: "apart" } as const; const presence = { online: true, since: "2026-10-06T00:00:00Z", overview: "live" } as const; const relayed = (value: unknown, state: "live" | "stored") => ({ presence, @@ -892,7 +896,7 @@ describe("an HQ outage", () => { cursor: { incarnation: "i1", revision: 3 }, knownKeys: ["app:shop", "project:ada"], }, - { scope: { kind: "attention", projectId: "ada" }, knownKeys: [] }, + { scope: { kind: "attention", projectId: "ada", health: "apart" }, knownKeys: [] }, ], }, ]); @@ -1047,6 +1051,98 @@ describe("HQ lifecycle receipt delivery", () => { ); }); +const decodeHealth = Schema.decodeUnknownSync(MateHealth); +it.effect( + "asks HQ for a Mate's health apart and reads it from that value, or from the whole record of an HQ that sends it inside", + () => + Effect.gen(function* () { + const sample = (revision: number) => + decodeHealth({ + source: { environmentId: "env", epoch: 1, incarnation: "run", revision }, + sampledAt: `2026-10-09T13:00:0${revision}Z`, + evidence: { + status: "strained", + severity: "warning", + resources: ["io"], + memory: null, + cpu: null, + io: { some: { avg10: 1.75, total: revision }, full: null }, + disk: null, + unavailable: [], + }, + }); + const record = { + presence: { online: true, since: "2026-10-09T13:00:00Z", overview: "none" }, + overview: null, + attention: null, + attentionState: "none", + } as const; + const store = makeAccountStore(AtomRegistry.make()); + const fixture = hqFixtureWire(); + const { fiber } = yield* live(store, fixture); + const asked = fixture.sent.flatMap(({ request }) => + request.type === "subscribe" + ? request.scopes.filter(({ scope }) => scope.kind === "attention") + : [], + ); + expect(asked.map(({ scope }) => scope)).toEqual([ + { kind: "attention", projectId: "ada", health: "apart" }, + ]); + const health = () => { + const read = publicRead(factOf(store.state(), "mateHealth", "ada")); + return read.kind === "known" ? (read.value as MateHealth).source.revision : read.kind; + }; + // An HQ from before the option answers on the scope it knows, health inside the record. + const older = { kind: "attention", projectId: "ada" } as const; + yield* fixture.send({ + type: "scope-reset", + scope: older, + incarnation: "hq1", + revision: 1, + values: [{ key: "ada", value: { ...record, health: sample(1), healthState: "live" } }], + removals: [], + }); + yield* fixture.send({ type: "scope-ready", scope: older, incarnation: "hq1", revision: 1 }); + yield* settle; + expect(health()).toBe(1); + // A newer HQ sends the record without it, and each sample as its own value. + const apart = { ...older, health: "apart" } as const; + yield* fixture.send({ + type: "scope-values", + scope: apart, + incarnation: "hq1", + revision: 2, + values: [ + { key: "ada", value: record }, + { key: HQ_ATTENTION_HEALTH_KEY, value: { health: sample(2), healthState: "live" } }, + ], + removals: [], + }); + yield* settle; + expect(health()).toBe(2); + yield* fixture.send({ + type: "scope-values", + scope: apart, + incarnation: "hq1", + revision: 3, + values: [ + { key: HQ_ATTENTION_HEALTH_KEY, value: { health: sample(3), healthState: "live" } }, + ], + removals: [], + }); + yield* settle; + expect(health()).toBe(3); + expect(publicRead(factOf(store.state(), "hqMate", "ada"))).toMatchObject({ + kind: "known", + value: record, + }); + expect(publicRead(factOf(store.state(), "hqMate", HQ_ATTENTION_HEALTH_KEY)).kind).not.toBe( + "known", + ); + yield* Fiber.interrupt(fiber); + }), +); + it.effect("relays health without an overview or attention and keeps it when HQ goes away", () => Effect.gen(function* () { const health = yield* Schema.decodeUnknownEffect(MateHealth)({ diff --git a/packages/client-runtime/src/data/families/hqMate.ts b/packages/client-runtime/src/data/families/hqMate.ts index 51662b002b..99b628ad54 100644 --- a/packages/client-runtime/src/data/families/hqMate.ts +++ b/packages/client-runtime/src/data/families/hqMate.ts @@ -6,7 +6,7 @@ * * @module data/families/hqMate */ -import { HqAttentionScopeValue } from "@t3tools/shared/hqStream"; +import { HQ_ATTENTION_HEALTH_KEY, HqAttentionScopeValue } from "@t3tools/shared/hqStream"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; @@ -29,10 +29,11 @@ export const hqMateFamily: FamilySpec<"hqMate"> = { scope: { source: "hq", suffix: "hq-mates", leaving: "removed", demand: "detail" }, hq: { scope: "attention", - idOf: (key) => key, + // The Mate's health, taken apart from its record, is `mateHealth`'s alone. + idOf: (key) => (key === HQ_ATTENTION_HEALTH_KEY ? null : key), keyOf: (id) => id, decode: (raw) => Option.getOrNull(decodeMate(raw)), - wireScope: (projectId) => ({ kind: "attention", projectId }), + wireScope: (projectId) => ({ kind: "attention", projectId, health: "apart" }), }, }; diff --git a/packages/client-runtime/src/data/families/mateAttention.ts b/packages/client-runtime/src/data/families/mateAttention.ts index 4b4f39d780..507228217a 100644 --- a/packages/client-runtime/src/data/families/mateAttention.ts +++ b/packages/client-runtime/src/data/families/mateAttention.ts @@ -9,7 +9,7 @@ * @module data/families/mateAttention */ import type { MateAttention } from "@t3tools/contracts"; -import { HqAttentionScopeValue } from "@t3tools/shared/hqStream"; +import { HQ_ATTENTION_HEALTH_KEY, HqAttentionScopeValue } from "@t3tools/shared/hqStream"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; @@ -31,7 +31,7 @@ export const mateAttentionFamily: FamilySpec<"mateAttention"> = { // HQ's relay rides the attention scope `hqMate` demands for each Mate it places. hq: { scope: "attention", - idOf: (key) => key, + idOf: (key) => (key === HQ_ATTENTION_HEALTH_KEY ? null : key), keyOf: (id) => id, decode: (raw) => Option.getOrNull(decodeRelayed(raw))?.attention ?? null, revisionOf: ({ source }, raw) => ({ diff --git a/packages/client-runtime/src/data/families/mateHealth.ts b/packages/client-runtime/src/data/families/mateHealth.ts index bf637cc78f..23fcc8b435 100644 --- a/packages/client-runtime/src/data/families/mateHealth.ts +++ b/packages/client-runtime/src/data/families/mateHealth.ts @@ -2,14 +2,15 @@ * A Mate's health (`MateHealth`): one value per Mate, by its project, * ordered by its own source revision. It arrives on two paths into this one family — straight from * a Mate the person has open (`adapters/mateHealth.ts`, `mate-direct`), and relayed by HQ for - * every Mate HQ places (its `attention` scope, `hq-stream`) — and the reducer keeps whichever is + * every Mate HQ places (its `attention` scope, `hq-stream`, as its own value or inside the Mate's + * record) — and the reducer keeps whichever is * newer by that revision: never by a clock, never by which path is live. A Mate from before the * health value relays none; its family holds nothing for it. * * @module data/families/mateHealth */ import type { MateHealth } from "@t3tools/contracts"; -import { HqAttentionScopeValue } from "@t3tools/shared/hqStream"; +import { HQ_ATTENTION_HEALTH_KEY, HqAttentionHealthValue } from "@t3tools/shared/hqStream"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; @@ -31,7 +32,8 @@ export const mateHealthFamily: FamilySpec<"mateHealth"> = { // HQ's relay rides the attention scope `hqMate` demands for each Mate it places. hq: { scope: "attention", - idOf: (key) => key, + // Its own value where HQ takes it apart, else inside the Mate's record (an HQ from before). + idOf: (key, { ownerId }) => (key === HQ_ATTENTION_HEALTH_KEY ? ownerId : key), keyOf: (id) => id, decode: (raw) => Option.getOrNull(decodeRelayed(raw))?.health ?? null, revisionOf: ({ source }, raw) => ({ @@ -46,7 +48,8 @@ export const mateHealthFamily: FamilySpec<"mateHealth"> = { }, }; -const decodeRelayed = Schema.decodeUnknownOption(HqAttentionScopeValue); +// Both shapes carry the two fields; the record's others are passed by. +const decodeRelayed = Schema.decodeUnknownOption(HqAttentionHealthValue); /** The health HQ relays of one Mate, under the organization's HQ link. */ export const hqMateHealthScope = (orgId: string, projectId: string): ScopeKey => diff --git a/packages/client-runtime/src/data/projections/mateHealth.ts b/packages/client-runtime/src/data/projections/mateHealth.ts index e3364339de..d17daf0dc3 100644 --- a/packages/client-runtime/src/data/projections/mateHealth.ts +++ b/packages/client-runtime/src/data/projections/mateHealth.ts @@ -1,4 +1,4 @@ -import type { MateHealth } from "@t3tools/contracts"; +import { cpuWarningQuote, type MateHealth } from "@t3tools/contracts"; import { hqMateScope } from "../families/hqMate.ts"; import { hqMateHealthScope, mateHealthScope } from "../families/mateHealth.ts"; import { linkKeys } from "../model.ts"; @@ -80,16 +80,13 @@ export function mateHealthCopy( : (() => { const window = evidence.cpu?.window; if (window == null) return []; - const cores = (value: number) => Number(value.toFixed(2)); - const usage = cores(window.usageUsec / window.elapsedUsec); - const capacity = cores(window.capacityCpus); - const measured = `Measured ${usage} CPU cores used out of a limit of ${capacity} over ${cores(window.elapsedUsec / 1_000_000)} seconds.`; - const consumer = window.consumer; + const quote = cpuWarningQuote(window); + const measured = `Measured ${quote.usedCores} CPU cores used out of a limit of ${quote.limitCores} over ${quote.seconds} seconds.`; return [ measured, - consumer === null + quote.consumer === null ? "The kernel reports runnable work waiting for CPU. No current process could be attributed; inspect container workloads in Zerops." - : `Top measured process: ${consumer.name} (PID ${consumer.pid}) used ${cores(consumer.cpuCores)} CPU cores. Check its workload before changing CPU in Zerops.`, + : `Top measured process: ${quote.consumer.name} (PID ${quote.consumer.pid}) used ${quote.consumer.cores} CPU cores. Check its workload before changing CPU in Zerops.`, ]; })(), ); diff --git a/packages/contracts/src/mateHealth.ts b/packages/contracts/src/mateHealth.ts index ee38fb002a..96721a547c 100644 --- a/packages/contracts/src/mateHealth.ts +++ b/packages/contracts/src/mateHealth.ts @@ -81,6 +81,29 @@ export const MateResourceHealth = Schema.Struct({ unavailable: Schema.Array(Schema.String), }); export type MateResourceHealth = typeof MateResourceHealth.Type; +type CpuWindow = NonNullable["window"]>; + +/** + * The numbers a CPU warning quotes from a measured window, to the two decimals it shows them in: + * a Mate under CPU strain publishes a new sample when one of these changes, and no other number. + */ +export function cpuWarningQuote(window: CpuWindow) { + const cores = (value: number) => Number(value.toFixed(2)); + return { + usedCores: cores(window.usageUsec / window.elapsedUsec), + limitCores: cores(window.capacityCpus), + seconds: cores(window.elapsedUsec / 1_000_000), + consumer: + window.consumer === null + ? null + : { + name: window.consumer.name, + pid: window.consumer.pid, + cores: cores(window.consumer.cpuCores), + }, + }; +} + export const MateHealth = Schema.Struct({ source: MateAttentionSource, sampledAt: IsoDateTime, diff --git a/packages/shared/src/hqStream.ts b/packages/shared/src/hqStream.ts index c5e2adffbd..09ef5ff724 100644 --- a/packages/shared/src/hqStream.ts +++ b/packages/shared/src/hqStream.ts @@ -69,7 +69,17 @@ export const HqScope = Schema.Union([ number: Schema.Int, }), Schema.Struct({ kind: Schema.Literal("operation"), appId: Schema.String }), - Schema.Struct({ kind: Schema.Literal("attention"), projectId: Schema.String }), + Schema.Struct({ + kind: Schema.Literal("attention"), + projectId: Schema.String, + /** + * `apart`: the reader takes the Mate's health as its own value ({@link HQ_ATTENTION_HEALTH_KEY}), + * so a health sample never resends the Mate's record. An HQ from before it passes the key by + * and sends health inside the record, as it does to a reader that does not ask; the scope's + * key ({@link hqScopeKey}) is the same either way, so a reader reads either HQ's answer. + */ + health: Schema.optionalKey(Schema.Literal("apart")), + }), ]); export type HqScope = typeof HqScope.Type; export const HqCursor = Schema.Struct({ @@ -429,3 +439,11 @@ export const HqAttentionScopeValue = Schema.Struct({ attentionState: Schema.Literals(["live", "stored", "none"]), }); export type HqAttentionScopeValue = typeof HqAttentionScopeValue.Type; +/** The key of a Mate's health in its attention scope, for a reader that takes health apart. */ +export const HQ_ATTENTION_HEALTH_KEY = "health"; +/** A Mate's health as its own value: the sample, and whether HQ hears it from the Mate now. */ +export const HqAttentionHealthValue = Schema.Struct({ + health: HqAttentionScopeValue.fields.health, + healthState: HqAttentionScopeValue.fields.healthState, +}); +export type HqAttentionHealthValue = typeof HqAttentionHealthValue.Type;