Skip to content

Health samples: send them apart, at most one a window #162

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fxck merged 5 commits into from
Oct 9, 2026
Merged

Health samples: send them apart, at most one a window #162

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ef78387
fix(server): publish a strained Mate's sample only when its warning o…
fxck Oct 9, 2026
51cdcd4
feat(hq): send a Mate's health apart from its record to a reader that…
fxck Oct 9, 2026
a3e6ac9
feat(client-runtime): take a Mate's health apart from its record on H…
fxck Oct 9, 2026
c4587b5
fix(server): publish at most one health sample a CPU window, the latest
fxck Oct 9, 2026
2770ab9
fix(hq): relay at most one of a Mate's health samples a window, the l…
fxck Oct 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 25 additions & 11 deletions apps/hq/src/hqScopes.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import {
HqNavigationProject,
HqAttentionValue,
hqScopeKey,
HQ_ATTENTION_HEALTH_KEY,
type HqScope,
type HqScopeFailure,
type HqStreamMessage,
Expand Down Expand Up @@ -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) }),
);
Expand Down Expand Up @@ -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);
}
Expand All @@ -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);
}
Expand Down Expand Up @@ -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") {
Expand All @@ -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) };
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -1383,22 +1397,22 @@ 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;
entry.generation += 1;
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);
}),
);
Expand All @@ -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)
Expand Down Expand Up @@ -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]));
Expand Down
115 changes: 114 additions & 1 deletion apps/hq/src/link.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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);
Expand Down Expand Up @@ -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<HqAttentionScopeValue>(
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<HqAttentionScopeValue>(
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;
Expand Down
25 changes: 21 additions & 4 deletions apps/hq/src/link.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down Expand Up @@ -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. */
Expand Down Expand Up @@ -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<MateHealth>(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(
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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;
}),
);
6 changes: 5 additions & 1 deletion apps/server/src/zerops/ZeropsMateAttention.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. */
Expand Down Expand Up @@ -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)),
Expand Down
Loading
Loading