Files
mesh-sdk/src/views/index.ts
T
2026-11-11 13:55:44 +01:00

568 lines
24 KiB
TypeScript

// A reader's first view of an owner's state, and its read and ask-again tools (novox/hq ADR 0615, ADR 0621; design 25
// § 12.2; design 47 *The forge module's tools come from mesh-sdk's TypeScript side*): the TypeScript side of the Go
// `views` package's Follower and StatusTools, with the same rules, so a TypeScript part answers `<module>_first_views`
// and `<module>_first_view_again` as the Go parts do and keeps no follower of its own.
//
// A follower asks the owner's first view at its start (every page), then keeps its picture from the owner's events
// alone, asking once more only on a gap in revisions (said in its log), never on a timer. Started while the owner is
// down it keeps asking, waiting between asks, and past its bound raises its condition; the answer clears it. It names
// no seat and no owner: the part hands it how to ask (a seat's verb through the seat's TypeScript contract, or an
// owner's tool) and the events it hears, each in the form below.
import type { ToolDefinition } from "../contracts/index.js";
import type { State } from "../state/index.js";
import { callerOf } from "../callers/index.js";
/** The version of every value this writes and the only one it reads. */
export const VERSION = 1;
/** The state in which a reader keeps each first view's status, as the Go side's `views.StatusState`. */
export const STATUS_STATE = "first-views";
/** How long a reader asks before it raises its condition naming the owner and the tool (ADR 0615 §6), in ms. */
export const BOUND_MS = 2 * 60 * 1000;
/**
* How long a part waits for a view before it writes its finding view-waits (design 47 *A part with no view says it
* waits*), in ms: the Go side's `views.WaitsBound`, so an owner down for a restart raises nothing.
*/
export const WAITS_BOUND_MS = 5 * 60 * 1000;
/** The kind of a part's finding of a view it has waited for past WAITS_BOUND_MS: the Go side's `views.ViewWaitsKind`. */
export const VIEW_WAITS_KIND = "view-waits";
/** The key of a reader's status of its first view of `owner`'s `state`. */
export function statusKey(owner: string, state: string): string {
return `${owner}.${state}`;
}
/**
* One page of a first view: the state's values by key, from `after` on, at the revision of the state the owner said up
* to before it read them; `next` is the key to ask after for the rest, empty when the answer is whole. A reader keeps
* the first page's revision.
*/
export interface Answer {
readonly v?: number;
readonly owner?: string;
readonly state?: string;
readonly revision: number;
/** When the owner's state was made on the bus (RFC 3339): another one is a state made again. */
readonly created: string;
readonly values: Readonly<Record<string, unknown>>;
readonly next?: string;
readonly at?: string;
/** When the owner's bus wrote each key (RFC 3339); a key whose time the owner does not say is absent. */
readonly written?: Readonly<Record<string, string>>;
}
/** One change of one key of the owner's state, at the revision it was written at. */
export interface ViewEvent {
readonly v?: number;
readonly owner?: string;
readonly state?: string;
readonly key: string;
readonly revision: number;
readonly created: string;
readonly deleted?: boolean;
readonly value?: unknown;
readonly at?: string;
/** The value was too large to carry: the reader asks its first view once. */
readonly gap?: boolean;
}
/** What a reader's read tool says of one first view: the Go side's `views.Status`, field for field. */
export interface Status {
owner: string;
tool: string;
state: string;
revision: number;
created?: string;
keys: number;
asked?: string;
answered?: string;
asks: number;
"by-hand": number;
raised: boolean;
heard?: number;
"heard-created"?: string;
behind?: string;
why?: string;
error?: string;
/**
* The revision of the revision view the part holds from this first view, as its seat's client names it (design 47 *A
* part with no view says it waits*); `waits-for` the one it waits for and acts on nothing new until it holds, since
* `waiting-since`; `waits-raised` says its view-waits finding is written (past WAITS_BOUND_MS).
*/
view?: string;
"waits-for"?: string;
"waiting-since"?: string;
"waits-raised"?: boolean;
}
/** How a part makes a follower: the owner and the tool it asks (for the status and the log), and the hooks. */
export interface FollowerOptions {
readonly owner: string;
readonly tool: string;
readonly state: string;
/** Ask the owner for the page after `after` (empty: the first). */
readonly ask: (after: string) => Promise<Answer>;
/** The reader's condition of this first view: raised past the bound, cleared by the answer. */
readonly raise?: (s: Status) => void | Promise<void>;
readonly clear?: (s: Status) => void | Promise<void>;
/** The part's view-waits finding: written past WAITS_BOUND_MS of a waitFor, cleared when it holds the view. */
readonly raiseWaits?: (s: Status) => void | Promise<void>;
readonly clearWaits?: (s: Status) => void | Promise<void>;
/** Told of each key the picture changes, after the change is held. */
readonly changed?: (key: string, value: unknown, deleted: boolean, at: string) => void | Promise<void>;
/** Told of the status each time an ask starts, is answered or fails, and of each event taken: where a reader keeps it. */
readonly said?: (s: Status) => void | Promise<void>;
readonly log?: (line: string) => void;
readonly now?: () => Date;
/** How long it asks before it raises, in ms; BOUND_MS when absent. */
readonly bound?: number;
/** The first wait between two asks that failed, in ms, doubling to a minute; five seconds when absent. */
readonly wait?: number;
/** Waits ms between two asks; a test hands one that does not sleep. */
readonly pause?: (ms: number) => Promise<void>;
/** Waits out WAITS_BOUND_MS of a waitFor; a test hands one it ends itself. Absent: a timer that holds no process open. */
readonly waitsBound?: (ms: number) => Promise<void>;
}
const ms = (t: string | undefined): number => (t ? Date.parse(t) || 0 : 0);
/** A reader's picture of one state of an owner, kept from its first view and its events. */
export class Follower {
private readonly o: FollowerOptions;
private values = new Map<string, unknown>();
private st: Status;
private asking = false;
private catching = false;
private during: ViewEvent[] = [];
private heardRev = 0;
private heardCreated = "";
private answered = false;
private closed = false;
private markReady!: () => void;
/** Resolves once a first view was answered. */
readonly ready: Promise<void>;
constructor(o: FollowerOptions) {
this.o = o;
this.st = { owner: o.owner, tool: o.tool, state: o.state, revision: 0, keys: 0, asks: 0, "by-hand": 0, raised: false };
this.ready = new Promise((resolve) => (this.markReady = resolve));
}
get owner(): string {
return this.o.owner;
}
get state(): string {
return this.o.state;
}
get tool(): string {
return this.o.tool;
}
private now(): string {
return (this.o.now ? this.o.now() : new Date()).toISOString();
}
private log(line: string): void {
this.o.log?.(line);
}
private pause(wait: number): Promise<void> {
return this.o.pause ? this.o.pause(wait) : new Promise((resolve) => setTimeout(resolve, wait));
}
/** The status now, with the keys held, the highest revision heard and why the picture is behind it. */
status(): Status {
const s: Status = { ...this.st, keys: this.values.size };
if (this.heardRev) s.heard = this.heardRev;
if (this.heardCreated) s["heard-created"] = this.heardCreated;
const behind = this.behind();
if (behind) s.behind = behind;
else delete s.behind;
return s;
}
private async say(): Promise<void> {
if (this.o.said) {
try {
await this.o.said(this.status());
} catch (err) {
this.log(`views: the status of the first view ${this.o.tool} is not kept: ${err instanceof Error ? err.message : String(err)}`);
}
}
}
/** A copy of the picture. */
picture(): Map<string, unknown> {
return new Map(this.values);
}
/** One key of the picture. */
get(key: string): unknown {
return this.values.get(key);
}
/** Why the picture is behind an event heard, or "" when it holds every revision heard. */
behind(): string {
if (!this.heardCreated) return "";
if (ms(this.heardCreated) > ms(this.st.created)) {
return `the state was made again at ${this.heardCreated}, the picture holds the one made at ${this.st.created ?? "none"}`;
}
if (ms(this.heardCreated) === ms(this.st.created) && this.heardRev > this.st.revision) {
return `revision ${this.heardRev} heard, the picture holds ${this.st.revision}`;
}
return "";
}
/**
* Say the part waits for `view`, a revision of the revision view its seat's client names, and acts on nothing new
* until it holds it (holds): the status says what and since when, and if it still waits for it past WAITS_BOUND_MS,
* raiseWaits is told (the part writes view-waits, viewWaitsFinding). Waiting for the view it waits for already changes
* nothing, not even since when. Resolves once the status is said; the bound runs on by itself.
*/
async waitFor(view: string): Promise<void> {
if (!view || this.st["waits-for"] === view) return;
this.st["waits-for"] = view;
this.st["waiting-since"] = this.now();
this.st["waits-raised"] = false;
this.log(`views: ${this.o.tool} waits for ${this.o.owner}'s view ${view}, from ${this.st["waiting-since"]}`);
await this.say();
void this.boundWaits(view);
}
private async boundWaits(view: string): Promise<void> {
const bound = this.o.waitsBound ?? ((wait: number) => new Promise<void>((resolve) => setTimeout(resolve, wait).unref?.()));
await bound(WAITS_BOUND_MS);
if (this.closed || this.st["waits-for"] !== view || this.st["waits-raised"]) return;
this.st["waits-raised"] = true;
this.log(`views: ${this.o.tool} has waited for ${this.o.owner}'s view ${view} since ${this.st["waiting-since"]}`);
try {
await this.o.raiseWaits?.(this.status());
} catch (err) {
this.log(`views: the view-waits of ${this.o.tool} is not written: ${err instanceof Error ? err.message : String(err)}`);
}
await this.say();
}
/** Say the part holds `view` now: it waits no more, and a raised view-waits is cleared (clearWaits). */
async holds(view: string): Promise<void> {
const raised = !!this.st["waits-raised"];
this.st.view = view;
delete this.st["waits-for"];
delete this.st["waiting-since"];
delete this.st["waits-raised"];
if (raised) {
try {
await this.o.clearWaits?.(this.status());
} catch (err) {
this.log(`views: the view-waits of ${this.o.tool} is not cleared: ${err instanceof Error ? err.message : String(err)}`);
}
}
await this.say();
}
/** Ask the first view until it is answered (ADR 0615 §6); resolves once answered, or once the follower is closed. */
start(): Promise<void> {
return this.askUntil("start", false);
}
/** Ask the first view once more by hand (the re-read tool, recorded by its caller as a hand act). */
again(why: string): Promise<void> {
return this.askUntil(`by hand: ${why}`, true);
}
/** Ask the first view once more for a gap the reader saw: counted among its asks, never among those by hand. */
gap(why: string): Promise<void> {
return this.askUntil(`a gap: ${why}`, false);
}
/** Stop asking: a loop waiting between two asks ends at its next turn. */
close(): void {
this.closed = true;
}
private async askUntil(why: string, byHand: boolean): Promise<void> {
if (this.asking) return; // one ask at a time; the one running brings the picture to its newest
this.asking = true;
this.during = [];
this.st.why = why;
if (byHand) this.st["by-hand"]++;
const bound = this.o.bound ?? BOUND_MS;
let wait = this.o.wait ?? 5000;
const first = ms(this.now());
try {
while (!this.closed) {
let failed: unknown;
try {
await this.askOnce();
} catch (err) {
failed = err ?? new Error("the ask failed");
}
if (failed === undefined) {
// An event heard that the answer does not hold yet leaves the picture behind: asked once more, by itself.
this.asking = false;
const behind = this.behind();
if (behind && !this.catching) {
this.catching = true;
this.log(`views: ${this.o.owner}'s ${this.o.state} is behind an event heard (${behind}): its first view ${this.o.tool} is asked once more`);
void this.catchUp();
}
return;
}
this.st.error = failed instanceof Error ? failed.message : String(failed);
const raise = !this.st.raised && ms(this.now()) - first >= bound;
if (raise) this.st.raised = true;
this.log(`the first view ${this.o.tool} of ${this.o.owner}'s ${this.o.state} is not answered (${this.st.error}): asked again in ${wait} ms`);
if (raise) await this.o.raise?.(this.status());
await this.say();
await this.pause(wait);
wait = Math.min(wait * 2, 60_000);
}
} finally {
this.asking = false;
}
}
// catchUp asks once more, each time after a pause (doubling to a minute, so an owner that answers from before its
// last write is not asked in a hot loop), until the picture holds every revision heard. One at a time: while it runs,
// an answered ask starts no other.
private async catchUp(): Promise<void> {
let wait = this.o.wait ?? 5000;
try {
while (!this.closed) {
await this.pause(wait);
const behind = this.behind();
if (!behind || this.closed) return;
this.log(`views: ${this.o.owner}'s ${this.o.state} is still behind (${behind}): its first view ${this.o.tool} is asked once more`);
await this.askUntil(`behind: ${behind}`, false);
wait = Math.min(wait * 2, 60_000);
}
} finally {
this.catching = false;
}
}
// askOnce asks every page and, once whole, takes it as the picture.
private async askOnce(): Promise<void> {
this.st.asked = this.now();
this.st.asks++;
await this.say();
let firstPage: Answer | undefined;
const values = new Map<string, unknown>();
const written = new Map<string, string>();
let after = "";
for (;;) {
const a = await this.o.ask(after);
if (a.v !== undefined && a.v !== VERSION) throw new Error(`views: unknown value version ${a.v}: this reader reads ${VERSION}`);
firstPage ??= a;
for (const [k, v] of Object.entries(a.values ?? {})) values.set(k, v);
for (const [k, t] of Object.entries(a.written ?? {})) written.set(k, t);
if (!a.next || a.next === after) break;
after = a.next;
}
const old = this.values;
this.values = values;
const wasRaised = this.st.raised;
this.st.revision = firstPage!.revision;
this.st.created = firstPage!.created;
this.st.answered = this.now();
this.st.raised = false;
delete this.st.error;
// The events heard while it asked, after the answer's revision and in order, are taken again.
const during = [...this.during].sort((x, y) => x.revision - y.revision);
this.during = [];
for (const ev of during) {
if (ms(ev.created) === ms(this.st.created) && ev.revision === this.st.revision + 1 && !ev.gap) {
if (ev.deleted) this.values.delete(ev.key);
else this.values.set(ev.key, ev.value);
if (ev.at) written.set(ev.key, ev.at);
this.st.revision = ev.revision;
}
}
if (!this.answered) {
this.answered = true;
this.markReady();
}
if (wasRaised) await this.o.clear?.(this.status());
await this.say();
if (this.o.changed) {
for (const [k, v] of this.values) {
if (!old.has(k) || JSON.stringify(old.get(k)) !== JSON.stringify(v)) await this.o.changed(k, v, false, written.get(k) ?? "");
}
for (const k of old.keys()) if (!this.values.has(k)) await this.o.changed(k, undefined, true, "");
}
}
/**
* Take one event into the picture and say whether it was a gap: another `created` (the owner's state made again), a
* revision past the one after the picture's, or a value too large to carry. A revision at or below the picture's is
* in it already and is passed over; so is any event before the first view is answered (the answer holds it).
*/
async apply(ev: ViewEvent): Promise<{ gap: boolean; why: string }> {
if (ev.v !== undefined && ev.v !== VERSION) throw new Error(`views: unknown value version ${ev.v}: this reader reads ${VERSION}`);
if (!ev.key || !ev.revision) throw new Error("views: an event names no key or revision");
if (ms(ev.created) > ms(this.heardCreated)) {
this.heardCreated = ev.created;
this.heardRev = ev.revision;
} else if (ms(ev.created) === ms(this.heardCreated) && ev.revision > this.heardRev) {
this.heardRev = ev.revision;
}
if (this.asking) this.during.push(ev);
if (!this.answered) {
this.log(`views: ${this.o.owner}'s ${this.o.state} at revision ${ev.revision} was heard before its first view was answered: passed over, the answer holds it or is asked again`);
return { gap: false, why: "" };
}
let why = "";
if (ms(ev.created) !== ms(this.st.created)) {
why = `${this.o.owner}'s ${this.o.state} was made again (or not marked when the first view was answered): created ${ev.created}, held ${this.st.created ?? "none"}`;
} else if (ev.revision <= this.st.revision) {
this.log(`views: ${this.o.owner}'s ${this.o.state} at revision ${ev.revision} is in the picture already (it holds ${this.st.revision}): passed over`);
return { gap: false, why: "" };
} else if (ev.revision > this.st.revision + 1) {
why = `${this.o.owner}'s ${this.o.state} at revision ${ev.revision}, the picture holds ${this.st.revision}: ${ev.revision - this.st.revision - 1} changes missed`;
} else if (ev.gap) {
why = `${this.o.owner}'s ${this.o.state}.${ev.key} at revision ${ev.revision} was too large to carry`;
}
if (why) return { gap: true, why };
if (ev.deleted) this.values.delete(ev.key);
else this.values.set(ev.key, ev.value);
this.st.revision = ev.revision;
await this.say(); // the revision the picture holds follows the events, not only the asks
await this.o.changed?.(ev.key, ev.value, !!ev.deleted, ev.at ?? "");
return { gap: false, why: "" };
}
/**
* A reader's handler of one event: taken into the picture, and a gap asks the first view once more, said in the log
* and not waited for (the event is taken; the ask brings the picture to its newest).
*/
async hear(ev: ViewEvent): Promise<void> {
const { gap, why } = await this.apply(ev);
if (gap) {
this.log(`views: a gap in ${this.o.state} (${why}): its first view ${this.o.tool} is asked once more`);
void this.gap(why);
}
}
}
/**
* What a views part says is wrong with a state it follows, in plain fields, as the Go side's `views.Finding`: the writer
* the part gives encodes it as the condition keeper reads it, which is the seat mesh-conditions' contract, not this
* package's (novox/hq ADR 0634 §4).
*/
export interface Finding {
readonly observer: string;
readonly kind: string;
readonly id: string;
readonly summary: string;
readonly headline: string;
readonly explanation: string;
readonly at: string;
}
/**
* A part's finding of the view it has waited for since `s["waiting-since"]` (design 47), as the Go side's
* `views.ViewWaitsFinding`: it names the part, the owner and the revision, and clears when the part holds it.
*/
export function viewWaitsFinding(part: string, s: Status, at: Date): Finding {
const since = s["waiting-since"] ? new Date(s["waiting-since"]).toISOString().replace(/\.\d{3}Z$/, "Z") : "";
return {
observer: part,
kind: VIEW_WAITS_KIND,
id: `${part}.${s.owner}.${s.state}`,
summary: `${part} waits for ${s.owner}'s view ${s["waits-for"] ?? ""} since ${since} and acts on nothing new until it holds it; it holds ${s.view || "none"}`,
headline: `${part} waits for a view of ${s.owner}`,
explanation:
`${part} keeps running what it runs and takes up nothing new until ${s.owner}'s view ${s["waits-for"] ?? ""} reaches it. Look at ${s.owner}: is it running, and has it ` +
`said that revision? ${part}'s first views tool says what it holds and waits for.`,
at: at.toISOString(),
};
}
/** A Follower's `said` that keeps its status at statusKey in the reader's state STATUS_STATE. */
export function keepStatus(st: State<Status>): (s: Status) => Promise<void> {
return async (s) => {
await st.put(statusKey(s.owner, s.state), s);
};
}
/** A hand act a part records before it does it (ADR 0565 clause 11): what is done, why, and its cause and verb. */
export interface HandAct {
readonly what: string;
readonly why: string;
readonly cause: string;
readonly verb: string;
}
/**
* A reader's read tool of its first views (`read`) and its re-read tool (`again`), as the Go side's StatusTools: read
* lists what each follower last said, kept in `st` (STATUS_STATE); again records a hand act through `acts` (none
* given: refused, nothing asked), then asks that follower's first view again, naming the caller the runtime proved.
*/
export function statusTools(
reader: string,
read: string,
again: string,
followers: readonly Follower[],
st: State<Status>,
acts: ((act: HandAct) => Promise<void>) | undefined,
): ToolDefinition[] {
return [
{
name: read,
description:
`${reader}'s first views of the owners' states it follows (novox/hq ADR 0615): per owner and state, the tool it asks, the ` +
"revision its picture holds now (events taken in included), the highest revision it heard and why it is behind it, if it " +
"is, when it asked and was answered, how many asks (by hand among them) and whether its condition (unanswered past two " +
"minutes) is raised; and the revision of the revision view it holds, the one it waits for and acts on nothing new until it " +
"holds, since when, and whether its view-waits finding (past five minutes) is written.",
input: { type: "object", properties: {} },
run: async () => {
const out: Status[] = [];
for (const k of await st.keys()) {
const e = await st.get(k);
if (e?.value && typeof e.value === "object") out.push(e.value);
}
return { "first-views": out };
},
},
{
name: again,
description:
`Ask ${reader}'s first view of an owner's state again now, by hand (a hand act, its caller and why kept in the status ` +
`${read} reads): for a picture a person doubts. The part asks once and follows the owner's events after the answer.`,
input: {
type: "object",
required: ["owner", "state", "why"],
properties: {
owner: { type: "string", description: "the owner, as the read tool names it" },
state: { type: "string", description: "the owner's state, as the read tool names it" },
why: { type: "string", description: "why it is asked again, in plain words" },
},
},
run: async (args) => {
const owner = typeof args.owner === "string" ? args.owner : "";
const state = typeof args.state === "string" ? args.state : "";
const why = typeof args.why === "string" ? args.why.trim() : "";
if (!owner || !state || !why) throw new Error("again names the owner, the state and why");
const f = followers.find((x) => x.owner === owner && x.state === state);
if (!f) throw new Error(`${reader} holds no first view of ${owner}'s ${state}`);
const by = callerOf(args);
if (!acts) throw new Error("the hand act was not recorded, so nothing was asked again: no way to record a hand act is given");
try {
await acts({ what: `asked ${reader}'s first view of ${owner}'s ${state} again by hand`, why, cause: "first-view-again", verb: again });
} catch (err) {
throw new Error(`the hand act was not recorded, so nothing was asked again: ${err instanceof Error ? err.message : String(err)}`);
}
void f.again(`${why} (${by})`);
return { asked: { owner, state, why, by }, read };
},
},
];
}
export * from "./writer.js";
export * from "./bundle.js";