From ef783872461b65ac8e140d44ce3a1348f2630dbd Mon Sep 17 00:00:00 2001 From: Ales Rechtorik Date: Fri, 9 Oct 2026 21:07:54 +0200 Subject: [PATCH 1/5] fix(server): publish a strained Mate's sample only when its warning or a quoted number changes A Mate under strain published every 2 s sample whole, so an I/O stall sent fresh CPU windows, memory and disk numbers nobody reads, and HQ relayed each. The warning quotes numbers only for CPU (the measured window and its top process, to two decimals), so the Mate now publishes again when the warning changes or one of those quoted numbers does. cpuWarningQuote is the one place the copy and the publish rule read the quote from. --- .../src/zerops/mateResourceHealth.test.ts | 96 ++++++++++++++++++- apps/server/src/zerops/mateResourceHealth.ts | 38 +++++--- .../src/data/projections/mateHealth.ts | 13 +-- packages/contracts/src/mateHealth.ts | 23 +++++ 4 files changed, 148 insertions(+), 22 deletions(-) diff --git a/apps/server/src/zerops/mateResourceHealth.test.ts b/apps/server/src/zerops/mateResourceHealth.test.ts index 1d35db8f2..26dc2ac65 100644 --- a/apps/server/src/zerops/mateResourceHealth.test.ts +++ b/apps/server/src/zerops/mateResourceHealth.test.ts @@ -9,6 +9,7 @@ import * as Queue from "effect/Queue"; import { vi } from "vite-plus/test"; import * as TestClock from "effect/testing/TestClock"; import { + publishesAgain, resourceHealthChanges, readResourceHealth as readKernelResourceHealth, } from "./mateResourceHealth.ts"; @@ -441,9 +442,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 +484,95 @@ 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); + }); +}); diff --git a/apps/server/src/zerops/mateResourceHealth.ts b/apps/server/src/zerops/mateResourceHealth.ts index 47caf74c1..d7457a5b2 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,31 @@ 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)); +} diff --git a/packages/client-runtime/src/data/projections/mateHealth.ts b/packages/client-runtime/src/data/projections/mateHealth.ts index e3364339d..d17daf0dc 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 ee38fb002..96721a547 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, From 51cdcd46a84e0381d7dc8f40c55d1f9463249170 Mon Sep 17 00:00:00 2001 From: Ales Rechtorik Date: Fri, 9 Oct 2026 21:18:25 +0200 Subject: [PATCH 2/5] feat(hq): send a Mate's health apart from its record to a reader that asks HQ relayed every health sample inside the Mate's whole attention record, so a strained Mate's 2 s samples resent its overview each time: 3.7 KB a sample on the structure socket in the 2026-10-09 recording. An attention scope may now ask for health apart; HQ then keeps the record and the health as two values of the scope, so a sample changes only the health value and the record goes out when it changes. The two shapes are separate journals of one scope key. A reader that does not ask, and an HQ from before (which passes the key by), keep the record whole, health inside. --- apps/hq/src/hqScopes.ts | 36 +++++++++++----- apps/hq/src/link.test.ts | 76 ++++++++++++++++++++++++++++++++- packages/shared/src/hqStream.ts | 20 ++++++++- 3 files changed, 119 insertions(+), 13 deletions(-) diff --git a/apps/hq/src/hqScopes.ts b/apps/hq/src/hqScopes.ts index dbc774952..0b8e761be 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 722351c7e..f359afcaf 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,59 @@ 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("closes on an overview that does not decode", () => Effect.gen(function* () { const { link } = yield* linked; diff --git a/packages/shared/src/hqStream.ts b/packages/shared/src/hqStream.ts index c5e2adffb..09ef5ff72 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; From a3e6ac968bf61a752c5b45e0741814ddd9959f7e Mon Sep 17 00:00:00 2001 From: Ales Rechtorik Date: Fri, 9 Oct 2026 21:18:26 +0200 Subject: [PATCH 3/5] feat(client-runtime): take a Mate's health apart from its record on HQ's attention scope The relayed Mate subscribes with health apart and reads health from its own value; from an HQ that does not take it apart it still reads health inside the record. The record's families pass the health value by. --- .../src/data/adapters/hq.test.ts | 106 +++++++++++++++++- .../src/data/families/hqMate.ts | 7 +- .../src/data/families/mateAttention.ts | 4 +- .../src/data/families/mateHealth.ts | 11 +- 4 files changed, 114 insertions(+), 14 deletions(-) diff --git a/packages/client-runtime/src/data/adapters/hq.test.ts b/packages/client-runtime/src/data/adapters/hq.test.ts index e81601d9e..2bf599526 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 51662b002..99b628ad5 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 4b4f39d78..507228217 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 bf637cc78..23fcc8b43 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 => From c4587b5556ee1984e783563980018ab474988dc1 Mon Sep 17 00:00:00 2001 From: Ales Rechtorik Date: Fri, 9 Oct 2026 21:31:42 +0200 Subject: [PATCH 4/5] fix(server): publish at most one health sample a CPU window, the latest In the 2026-10-09 run 3 Milo sat at its memory limit and the kernel woke the health reader on every memory event, so samples left at up to 33 a second. The Mate now publishes the first sample at once and then at most one per 2 s window, the latest, and only one that says something new. --- apps/server/src/zerops/ZeropsMateAttention.ts | 6 +++- .../src/zerops/mateResourceHealth.test.ts | 28 +++++++++++++++++++ apps/server/src/zerops/mateResourceHealth.ts | 15 ++++++++++ 3 files changed, 48 insertions(+), 1 deletion(-) diff --git a/apps/server/src/zerops/ZeropsMateAttention.ts b/apps/server/src/zerops/ZeropsMateAttention.ts index 1b563477b..6550cfc7c 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 26dc2ac65..4275d3797 100644 --- a/apps/server/src/zerops/mateResourceHealth.test.ts +++ b/apps/server/src/zerops/mateResourceHealth.test.ts @@ -9,6 +9,7 @@ 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, @@ -576,3 +577,30 @@ describe("a Mate publishes a sample again only when its warning, or a number the 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 d7457a5b2..bdb354b9c 100644 --- a/apps/server/src/zerops/mateResourceHealth.ts +++ b/apps/server/src/zerops/mateResourceHealth.ts @@ -379,3 +379,18 @@ export function publishesAgain(previous: MateResourceHealth, next: MateResourceH }; 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)))), + ), + ); From 2770ab9b5cae7497b56e05ef406856aea32ccc9a Mon Sep 17 00:00:00 2001 From: Ales Rechtorik Date: Fri, 9 Oct 2026 21:42:21 +0200 Subject: [PATCH 5/5] fix(hq): relay at most one of a Mate's health samples a window, the latest An older Mate at its memory limit sends a sample on every kernel wake (up to 33 a second in the 2026-10-09 run 3), and HQ relayed each to every reader. Each link now relays the first sample at once and then at most one per 2 s, the latest; its reports run one at a time, the relay beside the frames it reads. --- apps/hq/src/link.test.ts | 39 +++++++++++++++++++++++++++++++++++++++ apps/hq/src/link.ts | 25 +++++++++++++++++++++---- 2 files changed, 60 insertions(+), 4 deletions(-) diff --git a/apps/hq/src/link.test.ts b/apps/hq/src/link.test.ts index f359afcaf..e863ea437 100644 --- a/apps/hq/src/link.test.ts +++ b/apps/hq/src/link.test.ts @@ -264,6 +264,45 @@ describe("a Mate's link", () => { }), ); + 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 3743c7107..f17e2954c 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; }), );