novox/mesh-sdk
9
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

T-3: the finding writer in src/views (TypeScript twin), transport handed in

Browse Source 
Builds: design 47 G §383, §384, §390, §439, §461, §485, §503

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 13:08:17 +01:00
parent ff7f38c757
commit 3cd4fea600
3 changed files with 439 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/views/index.ts
+1
View File
@@ -563,3 +563,4 @@ export function statusTools(
},
];
}
export * from "./writer.js";
src/views/writer.ts
+345
View File
@@ -0,0 +1,345 @@
// The finding writer, TypeScript side (design 47 G §461, §485, §503): the same acts as the Go `views` writer, with the
// same bodies (§439) and rules, for the observers written in TypeScript. It keeps an observer's findings in its own state
// `findings` and says them as events (§383, §384). It names no seat and no module: the observer hands in the subjects of
// `finding-written`, `finding-cleared`, `findings-stated` and `findings-asked`, which the seat's generated contract makes.
// Its acts, `write`, `clear` and a statement, run one at a time in the order they are asked, on one chain of promises.
import type { State } from "../state/index.js";
/** How long an event the bus did not acknowledge is sent again before it is logged as an error (§383). */
export const PUBLISH_RETRY_MS = 60 * 1000;
/** How often a failing `observe` is called again until a set is said (§485). */
export const OBSERVE_RETRY_MS = 60 * 1000;
/** What a finding is about. */
export interface FindingSubject {
scope: string;
id: string;
machine?: string;
also?: string[];
}
/** One observation, as it was said. */
export interface FindingEvidence {
at: string;
said: string;
}
/** One answer the operator may give from a notification. */
export interface FindingAction {
label: string;
verb: string;
machine?: string;
arguments?: Record<string, string>;
level?: "acknowledge" | "approve";
}
/**
* One finding as an observer says it: the layout whole of §439's list. It is the body of `finding-written`, of each entry
* of `findings-stated`, and the value of its key `<scope>.<id>.<kind>` in the state `findings`.
*/
export interface FindingBody {
kind: string;
scope: string;
id: string;
at?: string;
since?: string;
severity: "urgent" | "warning";
summary: string;
headline?: string;
explanation?: string;
subject: FindingSubject;
evidence?: FindingEvidence[];
source: string;
observer?: string;
owner?: string;
"waiting-on"?: string;
actions?: FindingAction[];
needs?: string;
resolver?: string;
ref?: string;
run?: string;
}
/** The body of `finding-cleared`. */
export interface FindingCleared {
kind: string;
scope: string;
id: string;
at: string;
why: string;
}
/** A key the observer cannot judge yet since its start: its `since` and its bound (a duration such as `5m`). */
export interface PendingKey {
key: string;
since: string;
bound: string;
}
/** The body of `findings-stated`. */
export interface FindingsStated {
at: string;
asked?: string;
findings: FindingBody[];
pending: PendingKey[];
}
/** The body of `findings-asked`. */
export interface FindingsAsked {
ask: string;
at: string;
by: string;
module?: string;
}
/** What the observer's own truth says now. */
export interface Observed {
findings: FindingBody[];
pending: PendingKey[];
}
/** The key of a finding in the state `findings`: `<scope>.<id>.<kind>`. */
export function findingKey(scope: string, id: string, kind: string): string {
return `${scope}.${id}.${kind}`;
}
export interface WriterOptions {
/** The observer's module name: it fills the `observer` of a finding and tells an ask for another module. */
module: string;
/** The observer's own state `findings`. */
state: State<FindingBody>;
/** The subjects of the four events. */
subjects: { written: string; cleared: string; stated: string; asked: string };
/** The findings the observer observes now and its pending keys. It is called inside the writer's one chain: it must not call `write` or `clear`. */
observe(): Promise<Observed>;
/** Say one message and resolve when the bus acknowledged it; `id` is the bus's de-duplication id. */
publish(subject: string, body: unknown, id: string): Promise<void>;
/** Make a consumer of `subject` that delivers only the messages published after it is made; resolves to its stop. */
listen(subject: string, handle: (body: unknown) => void): Promise<() => void>;
/** Put a set larger than one bus message in the observer's sent store; resolves to what to publish instead. */
send?(label: string, body: unknown): Promise<unknown>;
log?(line: string): void;
now?(): Date;
/** Pause for `ms`; resolve false when the writer closed. */
wait?(ms: number): Promise<boolean>;
}
/** The finding writer of an observer. */
export class Writer {
private readonly o: WriterOptions;
private chain: Promise<unknown> = Promise.resolve();
private seq = 0;
private stop?: () => void;
private asks: string[] = [];
private busy = false;
private closed = false;
private readonly closing: Promise<void>;
private close_!: () => void;
constructor(o: WriterOptions) {
this.o = o;
this.closing = new Promise((r) => (this.close_ = r));
}
private log(line: string): void {
this.o.log?.(line);
}
private now(): Date {
return this.o.now ? this.o.now() : new Date();
}
private stamp(d: Date): string {
return d.toISOString();
}
private async pause(ms: number): Promise<boolean> {
if (this.o.wait) return this.o.wait(ms);
return new Promise((resolve) => {
const t = setTimeout(() => resolve(!this.closed), ms);
void this.closing.then(() => {
clearTimeout(t);
resolve(false);
});
});
}
/** Run `act` after every act asked before it has ended. */
private act<T>(act: () => Promise<T>): Promise<T> {
const run = this.chain.then(act, act);
this.chain = run.catch(() => undefined);
return run;
}
/**
* Make the consumer of `findings-asked`, then state the findings now; later each ask for this module (or for none) is
* answered by a statement. Resolves once the consumer exists; the start statement, retried every minute until `observe`
* answers, runs on.
*/
async start(): Promise<void> {
await this.listen();
this.request("");
}
/** The bus connection was made again: make the consumer again, from the newest ask, and state the findings again (§485). */
async reconnected(): Promise<void> {
if (this.closed) return;
try {
await this.listen();
} catch (e) {
this.log(`error: the finding writer could not make its consumer of ${this.o.subjects.asked} again: ${e}`);
}
this.request("");
}
/** End the consumer and the statements. */
close(): void {
this.closed = true;
this.stop?.();
this.stop = undefined;
this.close_();
}
private async listen(): Promise<void> {
const stop = await this.o.listen(this.o.subjects.asked, (body) => this.heard(body));
this.stop?.();
this.stop = stop;
}
private heard(body: unknown): void {
const a = body as FindingsAsked | null;
if (!a || typeof a.ask !== "string") {
this.log("an ask for findings could not be read");
return;
}
if (a.module && a.module !== this.o.module) return;
this.request(a.ask);
}
private request(ask: string): void {
this.asks.push(ask);
if (this.busy) return;
this.busy = true;
void this.drain();
}
private async drain(): Promise<void> {
while (this.asks.length > 0 && !this.closed) {
const asked = this.asks[this.asks.length - 1];
this.asks = [];
for (;;) {
try {
await this.state(asked);
break;
} catch (e) {
this.log(`the findings could not be stated, no set said, trying again in ${OBSERVE_RETRY_MS / 1000}s: ${e}`);
if (!(await this.pause(OBSERVE_RETRY_MS))) {
this.busy = false;
return;
}
}
}
}
this.busy = false;
}
/** Keep the finding in its key and say `finding-written` only when what it says changed; the key's `since` stays. */
write(f: FindingBody): Promise<void> {
return this.act(async () => {
const [full, old] = await this.complete(f);
await this.o.state.put(key(full), full);
if (old && unsaid(old) === unsaid(full)) return;
await this.say(this.o.subjects.written, "finding-written", key(full), full);
});
}
/** Delete the key and say `finding-cleared`, also of a key that does not exist, so a keeper holding it clears it. */
clear(scope: string, id: string, kind: string, why: string): Promise<void> {
return this.act(async () => {
const k = findingKey(scope, id, kind);
await this.o.state.delete(k);
const body: FindingCleared = { kind, scope, id, at: this.stamp(this.now()), why };
await this.say(this.o.subjects.cleared, "finding-cleared", k, body);
});
}
/** Judge every finding anew, write each key, delete each key neither stated nor pending, and say `findings-stated`. */
state(asked = ""): Promise<void> {
return this.act(async () => {
const obs = await this.o.observe();
const keep = new Set<string>();
const body: FindingsStated = { at: this.stamp(this.now()), ...(asked ? { asked } : {}), findings: [], pending: [] };
for (const f of obs.findings) {
const [full] = await this.complete(f);
await this.o.state.put(key(full), full);
keep.add(key(full));
body.findings.push(full);
}
for (const p of obs.pending) {
keep.add(p.key);
body.pending.push(p);
}
for (const k of await this.o.state.keys()) {
if (!keep.has(k)) await this.o.state.delete(k);
}
let out: unknown = body;
if (this.o.send) {
try {
out = await this.o.send(`findings-stated.${this.o.module}`, body);
} catch (e) {
this.log(`error: the findings-stated of ${this.o.module} could not be sent, the next statement covers it: ${e}`);
return;
}
}
await this.say(this.o.subjects.stated, "findings-stated", this.o.module, out);
});
}
private async complete(f: FindingBody): Promise<[FindingBody, FindingBody | null]> {
if (!f.kind || !f.scope || !f.id) throw new Error(`a finding needs its kind, scope and id: ${f.scope}.${f.id}.${f.kind}`);
const e = await this.o.state.get(key(f));
const old = e ? e.value : null;
const at = f.at ?? this.stamp(this.now());
const since = old?.since ?? f.since ?? at;
return [{ ...f, observer: f.observer || this.o.module, at, since }, old];
}
/** Publish awaiting the acknowledgement, again for a minute when unacknowledged, then log the error naming what it said. */
private async say(subject: string, event: string, what: string, body: unknown): Promise<void> {
const id = `${this.o.module}-${event}-${this.now().getTime()}-${++this.seq}`;
const start = this.now().getTime();
for (let delay = 1000; ; delay = Math.min(delay * 2, 10000)) {
try {
await this.o.publish(subject, body, id);
return;
} catch (e) {
if (this.now().getTime() - start >= PUBLISH_RETRY_MS || this.closed || !(await this.pause(delay))) {
this.log(`error: ${event} for ${what} was not acknowledged by the bus, the next statement covers it: ${e}`);
return;
}
}
}
}
}
function key(f: FindingBody): string {
return findingKey(f.scope, f.id, f.kind);
}
/** A finding as compared to tell whether what it says changed: not the time it was last seen, nor what saw it. */
function unsaid(f: FindingBody): string {
const { at: _at, run: _run, ...rest } = f;
return canonical(rest);
}
/** JSON with every object's keys in order, so two equal findings compare equal whatever order they were built in. */
function canonical(v: unknown): string {
if (Array.isArray(v)) return `[${v.map(canonical).join(",")}]`;
if (v && typeof v === "object") {
const o = v as Record<string, unknown>;
return `{${Object.keys(o)
.filter((k) => o[k] !== undefined)
.sort()
.map((k) => `${JSON.stringify(k)}:${canonical(o[k])}`)
.join(",")}}`;
}
return JSON.stringify(v);
}
test/sdk.test.ts
+93
View File
@@ -833,3 +833,96 @@ test("a first view follows its owner's events, asks again on a gap and by hand,
assert.equal(wRaised.length, 1, "a view held before its bound raises nothing");
assert.equal(wCleared.length, 1, "nothing raised, nothing cleared");
});
test("the finding writer writes, clears and states findings as the Go writer does", async () => {
const { Writer, OBSERVE_RETRY_MS } = await import("../dist/views/index.js");
const store = new Map<string, any>();
const st: any = {
get: async (k: string) => (store.has(k) ? { key: k, value: store.get(k), revision: 1 } : null),
put: async (k: string, v: any) => (store.set(k, structuredClone(v)), 1),
delete: async (k: string) => void store.delete(k),
keys: async () => [...store.keys()].sort(),
};
let clock = Date.parse("2026-11-11T12:00:00Z");
const sent: { subject: string; body: any }[] = [];
let refuse = 0;
const waits: number[] = [];
let ask: ((b: unknown) => void) | undefined;
let consumers = 0;
let observes = 0;
const logs: string[] = [];
const w = new Writer({
module: "m",
state: st,
subjects: { written: "w", cleared: "c", stated: "s", asked: "a" },
observe: async () => {
if (++observes <= 2) throw new Error("not yet");
return { findings: [{ kind: "k", scope: "module", id: "a", severity: "warning", summary: "now", source: "p", subject: { scope: "module", id: "a" } }], pending: [{ key: "module.wait.k", since: "2026-11-11T11:00:00Z", bound: "5m" }] };
},
publish: async (subject, body) => {
if (refuse > 0) { refuse--; throw new Error("no ack"); }
sent.push({ subject, body });
},
listen: async (_s, h) => { consumers++; ask = h; return () => {}; },
log: (l) => logs.push(l),
now: () => new Date(clock),
wait: async (ms) => { waits.push(ms); clock += ms; return true; },
});
const f = { kind: "k", scope: "module", id: "a", severity: "warning" as const, summary: "one", source: "p", subject: { scope: "module", id: "a" } };
const first = new Date(clock).toISOString();
await w.write(f);
clock += 3600_000;
await w.write({ ...f, run: "other" }); // nothing it says changed
await w.write({ ...f, summary: "two" });
assert.deepEqual(sent.map((s) => s.subject), ["w", "w"]);
assert.equal(sent[1].body.since, first);
assert.equal(sent[1].body.observer, "m");
await w.clear("module", "never", "k", "held by the keeper");
assert.equal(sent[2].subject, "c");
assert.equal(sent[2].body.why, "held by the keeper");
// an unacknowledged event is sent again for a minute, then logged naming the finding
refuse = 1000;
const t0 = clock;
await w.write({ ...f, id: "b" });
assert.ok(clock - t0 >= 60_000 && clock - t0 < 70_000);
assert.ok(logs.some((l) => l.includes("module.b.k") && l.startsWith("error")));
refuse = 0;
// a start states what is observed now, retrying observe every minute, and deletes the keys not stated or pending
store.set("module.old.k", f);
waits.length = 0;
await w.start();
for (let i = 0; i < 50 && sent.length < 4; i++) await new Promise((r) => setImmediate(r));
const stated = sent[3];
assert.equal(stated.subject, "s");
assert.deepEqual(waits.slice(0, 2), [OBSERVE_RETRY_MS, OBSERVE_RETRY_MS]);
assert.equal(stated.body.findings.length, 1);
assert.equal(stated.body.findings[0].since, first);
assert.equal(stated.body.pending[0].key, "module.wait.k");
assert.deepEqual([...store.keys()].sort(), ["module.a.k"]);
// an ask for another module is not answered, one for this module is, with its id; a reconnect makes the consumer again
ask!({ ask: "a1", at: "", by: "keeper", module: "other" });
ask!({ ask: "a2", at: "", by: "keeper", module: "m" });
for (let i = 0; i < 50 && sent.length < 5; i++) await new Promise((r) => setImmediate(r));
assert.equal(sent.length, 5);
assert.equal(sent[4].body.asked, "a2");
await w.reconnected();
for (let i = 0; i < 50 && sent.length < 6; i++) await new Promise((r) => setImmediate(r));
assert.equal(consumers, 2);
assert.equal(sent.length, 6);
w.close();
// one act at a time: a write asked during a statement waits, so the statement never deletes the key it made
let release!: () => void;
const gate = new Promise<void>((r) => (release = r));
const w2 = new Writer({
module: "m", state: st, subjects: { written: "w", cleared: "c", stated: "s", asked: "a" },
observe: async () => { await gate; return { findings: [], pending: [] }; },
publish: async () => {}, listen: async () => () => {},
});
const stating = w2.state();
const writing = w2.write({ ...f, id: "during" });
for (let i = 0; i < 20; i++) await new Promise((r) => setImmediate(r));
assert.equal(store.has("module.during.k"), false);
release();
await Promise.all([stating, writing]);
assert.equal(store.has("module.during.k"), true);
});