0.1.12: retire a consumer the mesh stops asking for, after five stable passes, and wait for a person over the bound

The hourly release let a mistaken bulk unassignment drain one consumer an hour
with nobody deciding. Per novox/hq ADR 0230 a consumer is retired (disabled,
marked, data kept) only after the same unasked set in five consecutive passes;
more than 3, or more than half of those held, waits for approve or reject.
Deletion is a separate act of a person, of a retired consumer only.
This commit is contained in:
jochen
2026-10-06 13:47:51 +02:00
parent a6b7b566b0
commit e0453316b8
6 changed files with 675 additions and 135 deletions
+13
View File
@@ -28,6 +28,19 @@ a database", "create a umami site" — which is the volatile, per-service half a
module. This is the ADR 0039 line drawn through provisioning: stable loop here, per-service create
in the module.
**A consumer the mesh stops asking for is retired, not withdrawn whole** (novox/hq ADR 0230). ACTIVE
becomes RETIRED — access disabled through the adapter's `retire` (or `remove`, for an adapter that
has none), marked with when and why, nothing deleted — only once the same set of consumers has gone
unasked in five consecutive passes that read the file (20 s at the default 5 s interval; a pass that
cannot read the file resets the count). A stable set of more than 3, or more than half of those held
where more than one is held, is not retired without a person: the provider waits for
`provisioner_retire_approve` or `provisioner_retire_reject`, and says `provisioner.retirement`
(`waiting`) through the `announce` option. Asked for again, a retired consumer is created — re-enabled
— at once. DELETED only through `provisioner_delete`, of a retired consumer. Given `module`, the four
tools (`provisioner_retirement`, `provisioner_retire_approve`, `provisioner_retire_reject`,
`provisioner_delete`) are registered for it; `provisionerTools(handle)` builds them otherwise.
Additions and changes are never delayed.
### `tools` — the tool-serving harness
`registerModuleTools`, the `ToolDefinition` type, and the registry/worker that loads a module's
tools and exposes them through the mesh's command surface. *How* a tool is declared and served is
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "@novox/mesh-sdk",
"version": "0.1.11",
"version": "0.1.12",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "@novox/mesh-sdk",
"version": "0.1.11",
"version": "0.1.12",
"devDependencies": {
"@types/node": "^22.0.0",
"typescript": "^5.6.0"
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@novox/mesh-sdk",
"version": "0.1.11",
"version": "0.1.12",
"description": "The stable spine a Novox Mesh module's own code builds against.",
"type": "module",
"exports": {
+369 -77
View File
@@ -1,6 +1,6 @@
// The reconcile harness every provider shares. The loop below is identical for postgres, redis,
// minio and umami: read the contributions the mesh delivered, bring each consumer's resource into
// being through the provider's adapter under the login and password the mesh minted, and withdraw
// being through the provider's adapter under the login and password the mesh minted, and retire
// what the mesh no longer asks for. A module writes ONLY the adapter — the per-service half — which
// is why this lives in the sdk and the adapter lives in the module (novox/hq ADR 0039).
//
@@ -13,6 +13,9 @@
import { readFile } from "node:fs/promises";
import type { ToolDefinition } from "../contracts/index.js";
import { registerModuleTools } from "../tools/index.js";
/** One consumer's resource to bring into being — everything the mesh derived and delivered. */
export interface Provision {
/** The login the mesh derived and gave the consumer to present. The provider creates exactly
@@ -44,7 +47,7 @@ export interface Provision {
* the credential is the mesh's, already delivered to the consumer. */
export interface Adapter {
create(p: Provision): Promise<void>;
/** Withdraw what `create` made for this consumer. `derived` is what the mesh last derived for
/** Retire what `create` made for this consumer when the adapter has no `retire`: disable, never destroy (issue 241). `derived` is what the mesh last derived for
* it, remembered by the harness: the mesh has already dropped the consumer from the
* contributions file by the time this is called, so it cannot be read again, and a provider
* that recomputed it would be keeping the second copy of a rule this removed
@@ -57,8 +60,72 @@ export interface Adapter {
* of being trusted from memory (novox/hq issue 120). An adapter without it is trusted from memory,
* as before. It must only read: it is asked often, and must never change the backend. */
holds?(p: Provision): Promise<boolean>;
/** Optional: RETIRE a consumer the mesh stopped asking for (novox/hq ADR 0230) — disable its
* access, reversibly, and mark it in the backend "to delete" with when and why; delete nothing.
* `create` must re-enable what this disabled and clear the mark, so a consumer asked for again is
* back as it was. An adapter without it is retired through `remove`, which since issue 241 already
* disables rather than destroys. */
retire?(p: {
readonly as: string;
readonly derived: Readonly<Record<string, unknown>>;
readonly why: string;
readonly at: Date;
}): Promise<void>;
/** Optional: every consumer the backend holds retired — what `cleanup list` shows. */
retired?(): Promise<RetiredConsumer[]>;
/** Optional: DELETE one retired consumer and its data. Only ever called for a consumer this
* harness or `retired()` knows retired, and only through `provisioner_delete`, a person's act. */
delete?(p: { readonly as: string; readonly why: string }): Promise<{ freedBytes?: number } | void>;
/** Optional: every consumer the backend holds ACTIVE that the mesh made, read at start, so one the
* mesh stopped asking for while this process was down is still retired (by the same rules). */
held?(): Promise<string[]>;
}
/** One retired consumer, as `provisioner_retirement` answers it. */
export interface RetiredConsumer {
readonly consumer: string;
readonly node?: string;
/** When it was retired, RFC 3339, or "" when the backend does not know. */
readonly retired_at: string;
readonly why: string;
/** Its size in bytes, or -1 when not knowable. */
readonly size_bytes: number;
/** "consumer", or a backend's own word for something else it keeps retired. */
readonly kind: string;
}
/** What `provisioner_retirement` answers: the provider's retirement state (novox/hq ADR 0230). */
export interface Retirement {
readonly resource: string;
readonly node: string;
readonly held: string[];
readonly stable_passes: number;
readonly bound: string;
readonly waiting: { consumers: { consumer: string; node?: string }[]; since: string; held: number } | null;
readonly rejected: { consumers: { consumer: string; node?: string }[]; by: string; why: string; at: string } | null;
readonly retired: RetiredConsumer[];
}
/** Who acts on a waiting retirement or a deletion, and why. `via` is "mesh-controller" when asked
* through the controller's verbs. */
export interface Act {
readonly why: string;
readonly by?: string;
readonly via?: string;
}
/** The running provisioner: a stop function, as before, that also answers and takes the person's
* acts on retirement (novox/hq ADR 0230). */
export type Provisioner = (() => void) & {
retirement(): Promise<Retirement>;
approve(consumers: readonly string[], act: Act): Promise<{ retired: string[] }>;
reject(consumers: readonly string[], act: Act): Promise<{ kept: string[] }>;
deleteRetired(consumer: string, act: Act & { readonly confirm?: string }): Promise<{ deleted: string; freed_bytes: number }>;
};
/** The one event a provider says about retirement (novox/hq ADR 0230). */
export const RETIREMENT_EVENT = "provisioner.retirement";
export interface ProvisionerOptions {
/** The contributions file the mesh writes for this resource (the module's `receives` path).
* Defaults to $MESH_RECEIVES. */
@@ -72,17 +139,25 @@ export interface ProvisionerOptions {
/** How long one `holds` may take before it counts as "could not ask". Defaults to 30000: the
* loop is sequential, so a check that hangs would otherwise stop every consumer's provisioning. */
holdsTimeoutMs?: number;
/** The most consumers one pass may withdraw (novox/hq to-be 45 Phase 2, ADR 0227 rule 4). Defaults
* to 1: a pass that would withdraw more withdraws nothing — the shape of issue 241, where one
* misread file withdrew seven at once — and says so. */
withdrawAtOnce?: number;
/** The largest share of the consumers held one pass may withdraw, where it holds more than one.
* Defaults to 0.5. */
withdrawFraction?: number;
/** While a braked withdrawal goes on being asked for, one consumer is let go this often, in ms.
* Defaults to an hour: an intended unassignment of many completes without a hand, and a mistaken
* one costs at most one consumer an hour while it is said. */
releaseEveryMs?: number;
/** How many consecutive passes that read the file must name the same set of consumers no longer
* asked for before they are retired (novox/hq ADR 0230). Defaults to 5: at the default interval a
* retirement comes 20 s after the first pass that saw it. Additions and changes are never delayed. */
stablePasses?: number;
/** A stable set larger than this is not retired without a person. Defaults to 3. */
retireAtOnce?: number;
/** Nor a set larger than this share of the consumers held, where more than one is held. Defaults
* to 0.5. */
retireFraction?: number;
/** How often a waiting retirement is said again, in ms. Defaults to 15 minutes. */
sayAgainEveryMs?: number;
/** Publishes one of the provider's events (`provisioner.retirement`); usually the SDK's `emit`.
* Without it the provider's retirement is said in its journal only. */
announce?: (event: string, body: Record<string, unknown>) => Promise<void> | void;
/** The machine this provider runs on, said in each event. Defaults to $MESH_NODE. */
node?: string;
/** The module's name: when given, the four retirement tools are registered for it through
* registerModuleTools, so the controller's `retire` and `cleanup` verbs can reach them. */
module?: string;
}
/** The longest a consumer whose `create` keeps failing to satisfy `holds` waits between checks. */
@@ -108,23 +183,37 @@ interface Contribution {
}
/**
* Run the reconcile loop for one provided resource. Returns a stop function. Never throws for a
* single bad contribution — it logs and keeps converging, because one consumer's failure must not
* stop the others' provisioning.
* Run the reconcile loop for one provided resource. Returns a stop function, which also carries the
* provider's retirement state and a person's acts on it. Never throws for a single bad contribution —
* it logs and keeps converging, because one consumer's failure must not stop the others'
* provisioning.
*
* **A consumer the mesh stops asking for is retired, never withdrawn whole (novox/hq ADR 0230).**
* ACTIVE → RETIRED once the same set of consumers has gone unasked in `stablePasses` consecutive
* passes that read the file; RETIRED is access disabled and marked, nothing deleted, and asked for
* again it is re-enabled at once. A stable set of more than `retireAtOnce`, or more than
* `retireFraction` of those held where more than one is held, is not retired without a person: the
* provider waits for approve or reject. RETIRED → DELETED only through `provisioner_delete`.
*/
export function runProvisioner(resource: string, adapter: Adapter, opts: ProvisionerOptions = {}): () => void {
export function runProvisioner(resource: string, adapter: Adapter, opts: ProvisionerOptions = {}): Provisioner {
const receives = opts.receives ?? envOrThrow("MESH_RECEIVES");
const everyMs = opts.everyMs ?? 5000;
const verifyEveryMs = opts.verifyEveryMs ?? 60_000;
const holdsTimeoutMs = opts.holdsTimeoutMs ?? 30_000;
const withdrawAtOnce = opts.withdrawAtOnce ?? 1;
const withdrawFraction = opts.withdrawFraction ?? 0.5;
const releaseEveryMs = opts.releaseEveryMs ?? 60 * 60_000;
// Consumers a braked pass kept, by when first kept; when the brake last let one go; and the set it
// last said, so a pass repeats nothing.
const braked = new Map<string, number>();
let releasedAt = 0;
let brakeSaid = "";
const stablePasses = opts.stablePasses ?? 5;
const retireAtOnce = opts.retireAtOnce ?? 3;
const retireFraction = opts.retireFraction ?? 0.5;
const sayAgainEveryMs = opts.sayAgainEveryMs ?? 15 * 60_000;
const node = opts.node ?? process.env.MESH_NODE ?? "";
const say = (line: string): void => console.error(`[provisioner:${resource}] ${line}`);
// Retirement (novox/hq ADR 0230). The unasked set last seen and in how many consecutive readable
// passes; a set too large, waiting for a person; a set a person rejected; and every consumer known
// retired, by name.
let stable = { set: "", count: 0 };
let pending: { consumers: string[]; since: number; held: number; saidAt: number } | null = null;
let rejected: { consumers: string[]; by: string; why: string; at: number } | null = null;
const retired = new Map<string, { node?: string; at: number; why: string }>();
let seeded = false;
let verifiedAt = 0;
// Consumers the backend reported lost, and how many times in a row. A `create` that succeeded and
// is still not held on the next check will never be: something in the adapter disagrees with
@@ -134,7 +223,7 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
// login (`as`) -> what was last applied: the hash that says whether it changed, and the
// derivation to hand back on withdrawal, which by then is no longer in the file.
const applied = new Map<string, { hash: string; derived: Readonly<Record<string, unknown>> }>();
const applied = new Map<string, { hash: string; derived: Readonly<Record<string, unknown>>; node?: string }>();
// How many passes each consumer's secret has been unreadable for, so the wait can stop sounding
// like a wait.
const waiting = new Map<string, number>();
@@ -155,7 +244,12 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
// "none" withdrew every consumer at once: on 2026-10-04 one failed read made the postgres provider
// drop every database it held. Only a file that was read can withdraw anything, so the whole pass
// waits for the next one.
if (given === null) return;
if (given === null) {
// Not a result: the passes that make a retirement stable must each have read the file.
stable = { set: "", count: 0 };
return;
}
if (!seeded) await seed();
const wantByAs = new Map(given.map((g) => [g.as, g]));
// On this pass, ask the backend rather than memory whether each applied consumer is still there.
let verifying = adapter.holds !== undefined && Date.now() - verifiedAt >= verifyEveryMs;
@@ -233,7 +327,13 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
}
try {
await adapter.create(p);
applied.set(g.as, { hash: h, derived: p.derived });
applied.set(g.as, { hash: h, derived: p.derived, node: g.node });
if (retired.has(g.as)) {
// Asked for again: the ordinary create re-enabled it and cleared its mark (ADR 0230).
retired.delete(g.as);
say(`${g.as}: asked for again; re-enabled as it was`);
await tell("reenabled", { consumers: [{ consumer: g.as, node: g.node }] });
}
if (reapplying === undefined) {
// Applied for a new login, password or values: whatever was counted before does not carry over.
lost.delete(g.as);
@@ -253,67 +353,216 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
}
}
// Withdraw every login the provider made that the mesh no longer asks for. The mesh drops a
// consumer from the file when it goes away; that is how a login is withdrawn rather than kept
// working for ever. Only logins this harness created are touched.
//
// **Within a bound** (novox/hq to-be 45 Phase 2, ADR 0227 rule 4): a pass that would withdraw more
// than `withdrawAtOnce` consumers, or more than `withdrawFraction` of those held, withdraws
// nothing and says so; one is let go every `releaseEveryMs` while the mesh goes on not asking.
const withdrawing = [...applied.keys()].filter((as) => !wantByAs.has(as)).sort();
for (const as of [...braked.keys()]) if (wantByAs.has(as)) braked.delete(as);
let releasing = withdrawing;
const over =
withdrawing.length > 0 &&
(withdrawing.length > withdrawAtOnce ||
(applied.size > 1 && withdrawing.length > withdrawFraction * applied.size));
if (over) {
const set = withdrawing.join(", ");
if (set !== brakeSaid) {
console.error(
`[provisioner:${resource}] WITHDRAWAL BRAKED: this pass would withdraw ${withdrawing.length} of the ` +
`${applied.size} consumer(s) this provider holds (${set}), more than ${withdrawAtOnce} at once or ` +
`${Math.round(withdrawFraction * 100)}% of them. Nothing is withdrawn; one is let go every ` +
`${Math.round(releaseEveryMs / 1000)}s while the mesh goes on not asking for them (novox/hq ADR 0227 rule 4)`,
);
brakeSaid = set;
}
if (braked.size === 0) releasedAt = Date.now();
for (const as of withdrawing) if (!braked.has(as)) braked.set(as, Date.now());
releasing = [];
if (Date.now() - releasedAt >= releaseEveryMs) {
releasedAt = Date.now();
releasing = [withdrawing[0]];
console.error(
`[provisioner:${resource}] ${withdrawing[0]}: released by the withdrawal brake; ` +
`${withdrawing.length - 1} more kept`,
);
}
} else if (braked.size > 0) {
console.error(`[provisioner:${resource}] the withdrawal is within its bound again`);
braked.clear();
brakeSaid = "";
// Retire what the mesh no longer asks for — once the same set has gone unasked for
// `stablePasses` passes, and within the bound, or with a person's word (novox/hq ADR 0230).
// Only consumers this harness applied, or the backend says the mesh made, are touched.
const unasked = [...applied.keys()].filter((as) => !wantByAs.has(as)).sort();
const key = unasked.join(",");
if (pending && pending.consumers.join(",") !== key) {
say(`the retirement waiting for a person (${pending.consumers.join(", ")}) is no longer what the mesh says; settled, nothing retired`);
await tell("settled", { consumers: named(pending.consumers) });
pending = null;
}
for (const as of releasing) {
if (rejected && rejected.consumers.join(",") !== key) {
say(`the rejected retirement (${rejected.consumers.join(", ")}) is no longer what the mesh says`);
await tell("settled", { consumers: named(rejected.consumers) });
rejected = null;
}
if (unasked.length === 0) {
stable = { set: "", count: 0 };
return;
}
stable = stable.set === key ? { set: key, count: stable.count + 1 } : { set: key, count: 1 };
if (stable.count < stablePasses) return;
if (rejected) return; // a person said keep them, and the mesh still says the same
const held = applied.size;
if (overTheBound(unasked.length, held)) {
const now = Date.now();
if (pending === null) {
pending = { consumers: unasked, since: now, held, saidAt: now };
say(
`RETIREMENT WAITS FOR A PERSON: the mesh no longer asks for ${unasked.length} of the ${held} consumer(s) ` +
`this provider holds (${unasked.join(", ")}), ${bound(held)}. Nothing is retired until ` +
`\`retire approve\` or \`retire reject\` (novox/hq ADR 0230)`,
);
await tell("waiting", { consumers: named(unasked), held, since: iso(now) });
} else if (now - pending.saidAt >= sayAgainEveryMs) {
pending.saidAt = now;
await tell("waiting", { consumers: named(unasked), held: pending.held, since: iso(pending.since) });
}
return;
}
const why = `the mesh stopped asking for it: ${stablePasses} passes from ${iso(Date.now() - (stablePasses - 1) * everyMs)}`;
const done = await retireAll(unasked, why);
if (done.length > 0) await tell("retired", { consumers: done.map((as) => retiredBody(as)), held, why });
}
function overTheBound(n: number, held: number): boolean {
return n > retireAtOnce || (held > 1 && n > retireFraction * held);
}
function bound(held: number): string {
return `more than ${retireAtOnce}, or more than ${Math.round(retireFraction * 100)}% of the ${held} held`;
}
// Retire each, as it is said: disabled and marked by the adapter, nothing deleted. Answers those
// that went; one that failed stays active and is asked again by the next stable pass.
async function retireAll(names: readonly string[], why: string): Promise<string[]> {
const done: string[] = [];
for (const as of names) {
const was = applied.get(as);
if (was === undefined) continue;
// Said before it is done: a removal is the one act here that destroys what a consumer kept.
console.error(`[provisioner:${resource}] ${as}: no longer in ${receives}; removing it from the backend`);
say(`${as}: no longer in ${receives}; retiring it (access disabled, nothing deleted): ${why}`);
try {
await adapter.remove({ as, derived: was.derived });
const at = new Date();
if (adapter.retire) await adapter.retire({ as, derived: was.derived, why, at });
else await adapter.remove({ as, derived: was.derived });
applied.delete(as);
lost.delete(as);
braked.delete(as);
retired.set(as, { node: was.node, at: at.getTime(), why });
done.push(as);
} catch (err) {
console.error(`[provisioner:${resource}] ${as}: remove failed, will retry: ${err}`);
say(`${as}: retiring failed, will retry: ${err}`);
}
}
stable = { set: "", count: 0 };
return done;
}
function named(names: readonly string[]): { consumer: string; node?: string }[] {
return names.map((as) => ({ consumer: as, node: applied.get(as)?.node ?? retired.get(as)?.node }));
}
function retiredBody(as: string): Record<string, unknown> {
const r = retired.get(as);
return { consumer: as, node: r?.node, retired_at: r ? iso(r.at) : "", why: r?.why ?? "" };
}
// Said on the bus, never failing the act it reports (the journal still has it).
async function tell(change: string, body: Record<string, unknown>): Promise<void> {
if (!opts.announce) return;
try {
await opts.announce(RETIREMENT_EVENT, {
provider: resource, "provider-node": node, change, bound: bound(applied.size), at: iso(Date.now()), ...body,
});
} catch (err) {
say(`could not announce ${RETIREMENT_EVENT} ${change}: ${err}`);
}
}
// What the backend holds that the mesh made: active ones become held, so one the mesh stopped asking
// for while this process was down is still retired, by the same rules; retired ones are known.
async function seed(): Promise<void> {
try {
for (const as of (await adapter.held?.()) ?? []) {
if (!applied.has(as)) applied.set(as, { hash: "", derived: {} });
}
for (const r of (await adapter.retired?.()) ?? []) {
if (r.kind === "consumer" && !retired.has(r.consumer)) {
retired.set(r.consumer, { node: r.node, at: r.retired_at ? Date.parse(r.retired_at) : 0, why: r.why });
}
}
seeded = true;
} catch (err) {
say(`cannot read what the backend holds yet, will ask again: ${err}`);
}
}
// One act at a time: a person's approval and a pass never interleave.
let queue: Promise<unknown> = Promise.resolve();
function serially<T>(f: () => Promise<T>): Promise<T> {
const run = queue.then(f, f);
queue = run.catch(() => undefined);
return run;
}
const sameSet = (a: readonly string[], b: readonly string[]): boolean =>
[...a].sort().join(",") === [...b].sort().join(",");
async function retirement(): Promise<Retirement> {
const fromBackend = (await adapter.retired?.()) ?? [];
const list: RetiredConsumer[] = [...fromBackend];
for (const [as, r] of retired) {
if (!list.some((x) => x.consumer === as && x.kind === "consumer")) {
list.push({ consumer: as, node: r.node, retired_at: r.at ? iso(r.at) : "", why: r.why, size_bytes: -1, kind: "consumer" });
}
}
return {
resource, node, held: [...applied.keys()].sort(), stable_passes: stablePasses, bound: bound(applied.size),
waiting: pending && { consumers: named(pending.consumers), since: iso(pending.since), held: pending.held },
rejected: rejected && { consumers: named(rejected.consumers), by: rejected.by, why: rejected.why, at: iso(rejected.at) },
retired: list,
};
}
function needWhy(act: Act, verb: string): void {
if (!act.why || !act.why.trim()) throw new Error(`${verb}: why is required`);
}
async function approve(consumers: readonly string[], act: Act): Promise<{ retired: string[] }> {
needWhy(act, "approve");
return serially(async () => {
const set = pending?.consumers ?? rejected?.consumers;
if (!set || !sameSet(set, consumers)) {
throw new Error(
`approve: nothing waits with exactly that set (${[...consumers].sort().join(", ")}); ` +
`waiting: ${pending?.consumers.join(", ") || "none"}; rejected: ${rejected?.consumers.join(", ") || "none"}. Look again`,
);
}
const held = applied.size;
say(`retirement of ${set.join(", ")} approved by ${act.by || "someone"}: ${act.why}`);
await tell("approved", { consumers: named(set), held, why: act.why, by: act.by ?? "", via: act.via ?? "" });
pending = null;
rejected = null;
const why = `approved by ${act.by || "someone"}: ${act.why}`;
const done = await retireAll(set, why);
if (done.length > 0) await tell("retired", { consumers: done.map((as) => retiredBody(as)), held, why });
return { retired: done };
});
}
async function reject(consumers: readonly string[], act: Act): Promise<{ kept: string[] }> {
needWhy(act, "reject");
return serially(async () => {
if (!pending || !sameSet(pending.consumers, consumers)) {
throw new Error(
`reject: nothing waits with exactly that set (${[...consumers].sort().join(", ")}); ` +
`waiting: ${pending?.consumers.join(", ") || "none"}. Look again`,
);
}
const kept = pending.consumers;
rejected = { consumers: kept, by: act.by ?? "", why: act.why, at: Date.now() };
pending = null;
say(`retirement of ${kept.join(", ")} rejected by ${act.by || "someone"}: ${act.why}; kept active while the mesh says the same`);
await tell("rejected", { consumers: named(kept), why: act.why, by: act.by ?? "", via: act.via ?? "" });
return { kept };
});
}
async function deleteRetired(consumer: string, act: Act & { readonly confirm?: string }): Promise<{ deleted: string; freed_bytes: number }> {
needWhy(act, "delete");
if (act.confirm !== consumer) throw new Error("delete: confirm must repeat the consumer's name");
return serially(async () => {
if (applied.has(consumer)) throw new Error(`delete: ${consumer} is active, not retired — only a retired consumer is deleted`);
const known = retired.has(consumer) || ((await adapter.retired?.()) ?? []).some((r) => r.consumer === consumer);
if (!known) throw new Error(`delete: ${consumer} is not retired here`);
if (!adapter.delete) throw new Error(`delete: this provider (${resource}) cannot delete; its adapter has no delete`);
say(`${consumer}: DELETING, asked by ${act.by || "someone"}: ${act.why}`);
const r = await adapter.delete({ as: consumer, why: act.why });
const freed = (r && typeof r.freedBytes === "number") ? r.freedBytes : -1;
const was = retired.get(consumer);
retired.delete(consumer);
await tell("deleted", {
consumers: [{ consumer, node: was?.node, size_bytes: freed }], why: act.why, by: act.by ?? "", via: act.via ?? "",
});
return { deleted: consumer, freed_bytes: freed };
});
}
const tick = async (): Promise<void> => {
if (stopped) return;
try {
await reconcile();
await serially(reconcile);
} catch (err) {
console.error(`[provisioner:${resource}] reconcile error: ${err}`);
}
@@ -321,16 +570,59 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
};
void tick();
return () => {
const stop = () => {
stopped = true;
};
const handle: Provisioner = Object.assign(stop, { retirement, approve, reject, deleteRetired });
if (opts.module) registerModuleTools(opts.module, () => provisionerTools(handle));
return handle;
}
const strings = (v: unknown): string[] => (Array.isArray(v) ? v.filter((x): x is string => typeof x === "string") : []);
const text = (v: unknown): string => (typeof v === "string" ? v : "");
/** The four tools every provider serves for retirement (novox/hq ADR 0230), the controller's `retire`
* and `cleanup` verbs asking them per machine. */
export function provisionerTools(p: Provisioner): ToolDefinition[] {
const act = { type: "string" };
return [
{
name: "provisioner_retirement",
description: "This provider's retirement: the consumers it holds, a retirement waiting for a person, one a person rejected, and every consumer retired (access disabled, data kept) with when, why and size.",
input: {},
run: async () => p.retirement(),
},
{
name: "provisioner_retire_approve",
description: "Retire exactly the set waiting for a person (or rejected earlier). Repeat the set in consumers; why is required.",
input: { consumers: { type: "array", items: { type: "string" } }, why: act, by: act, via: act },
run: async (a) => p.approve(strings(a.consumers), { why: text(a.why), by: text(a.by), via: text(a.via) }),
},
{
name: "provisioner_retire_reject",
description: "Keep the set waiting for a person active, though the mesh no longer asks for it. Repeat the set in consumers; why is required.",
input: { consumers: { type: "array", items: { type: "string" } }, why: act, by: act, via: act },
run: async (a) => p.reject(strings(a.consumers), { why: text(a.why), by: text(a.by), via: text(a.via) }),
},
{
name: "provisioner_delete",
description: "Delete one RETIRED consumer and its data from this provider's backend — never an active one. Repeat its name in confirm; why is required.",
input: { consumer: act, confirm: act, why: act, by: act, via: act },
run: async (a) =>
p.deleteRetired(text(a.consumer), { why: text(a.why), by: text(a.by), via: text(a.via), confirm: text(a.confirm) }),
},
];
}
function iso(ms: number): string {
return new Date(ms).toISOString();
}
/**
* Read the mesh's contributions file for one resource: the consumers it asks for, or `null` when the
* file says nothing usable — absent, unreadable, not JSON, for another requirement, or without a
* `given` list. `null` is not "no consumer": the mesh always writes the file, and writes `given: []`
* when nobody asks, so only that withdraws every consumer.
* when nobody asks, so only that makes every consumer unasked — retired by the rules of ADR 0230, never at once without a person.
*/
async function readContributions(
path: string,
+283
View File
@@ -0,0 +1,283 @@
// Retirement (novox/hq ADR 0230): a consumer the mesh stops asking for is retired — access disabled,
// data kept — only after the same result in five consecutive passes, and a set too large waits for a
// person. Deletion is a person's act, of a retired consumer only.
import { test } from "node:test";
import assert from "node:assert/strict";
import { mkdtemp, writeFile, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { runProvisioner, provisionerTools, RETIREMENT_EVENT, type Provisioner } from "../dist/provisioner/index.js";
import { resetTools, collectTools } from "../dist/tools/index.js";
const EVERY = 10;
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
async function until(cond: () => boolean | Promise<boolean>, ms = 3000): Promise<void> {
const start = Date.now();
while (!(await cond())) {
if (Date.now() - start > ms) throw new Error("timed out waiting");
await sleep(5);
}
}
interface World {
receives: string;
ask(names: string[]): Promise<void>;
created: string[];
retired: { as: string; why: string }[];
removed: string[];
deleted: string[];
events: Record<string, unknown>[];
said: string[];
p: Provisioner;
done(): void;
}
async function world(names: string[], opts: { retire?: boolean; held?: string[] } = {}): Promise<World> {
const dir = await mkdtemp(join(tmpdir(), "retire-"));
const receives = join(dir, "postgres-database.json");
const ask = async (list: string[]) => {
const given = [];
for (const n of list) {
await writeFile(join(dir, `${n}.secret`), `pw-${n}\n`);
given.push({ from: n, node: "anchor", as: n, secret: join(dir, `${n}.secret`) });
}
await writeFile(receives, JSON.stringify({ contributions: 1, requirement: "postgres-database", given }));
};
await ask(names);
const w = { receives, ask, created: [], retired: [], removed: [], deleted: [], events: [], said: [] } as unknown as World;
const was = console.error;
console.error = (...a: unknown[]) => w.said.push(a.join(" "));
const adapter = {
async create(p: { as: string }) { w.created.push(p.as); },
async remove(p: { as: string }) { w.removed.push(p.as); },
async delete(p: { as: string }) { w.deleted.push(p.as); return { freedBytes: 42 }; },
...(opts.retire === false ? {} : { async retire(p: { as: string; why: string }) { w.retired.push({ as: p.as, why: p.why }); } }),
...(opts.held ? { async held() { return opts.held!; } } : {}),
};
w.p = runProvisioner("postgres-database", adapter, {
receives, everyMs: EVERY, node: "anchor",
announce: (event, body) => { assert.equal(event, RETIREMENT_EVENT); w.events.push(body); },
});
// A pass is counted by the file being read: wait wall time for n passes and some slack.
w.done = () => { w.p(); console.error = was; };
await until(() => w.created.length >= names.length);
return w;
}
const changes = (w: World) => w.events.map((e) => e.change);
test("a transient empty list for four passes retires nothing; five stable passes retire, data kept", async () => {
const w = await world(["a", "b", "c"], {});
try {
// The unasked set is {a} — one of three, within the bound — seen for fewer than five passes.
const p = w.p;
// Run the loop by hand: stop the timer-driven one and drive passes directly is not exposed, so
// the transient is made short in wall time instead: ask without a, then with a, quickly.
await w.ask(["b", "c"]);
await sleep(EVERY * 2);
await w.ask(["a", "b", "c"]);
await sleep(EVERY * 8);
assert.deepEqual(w.retired, [], "retired on a transient result");
assert.deepEqual(changes(w), []);
// Stable: retired, through retire, and not removed.
await w.ask(["b", "c"]);
await until(() => w.retired.length === 1);
assert.equal(w.retired[0].as, "a");
assert.match(w.retired[0].why, /5 passes/);
assert.deepEqual(w.removed, []);
assert.deepEqual(w.deleted, []);
assert.deepEqual(changes(w), ["retired"]);
const r = await p.retirement();
assert.deepEqual(r.retired.map((x) => x.consumer), ["a"]);
assert.deepEqual(r.held, ["b", "c"]);
assert.equal(r.stable_passes, 5);
} finally {
w.done();
}
});
test("the fifth pass and not the fourth retires", async () => {
// Driven by the clock: five passes 10 ms apart take at least 40 ms after the first sees the set.
const w = await world(["a", "b", "c"]);
try {
await w.ask(["b", "c"]);
const start = Date.now();
await until(() => w.retired.length === 1);
assert.ok(Date.now() - start >= 4 * EVERY - 2, `retired after ${Date.now() - start} ms, before five passes`);
} finally {
w.done();
}
});
test("an unreadable file resets the count", async () => {
const w = await world(["a", "b", "c"]);
try {
await w.ask(["b", "c"]);
// Interleave unreadable passes so five readable ones never run in a row.
for (let i = 0; i < 6; i++) {
await sleep(EVERY * 2);
await rm(w.receives);
await sleep(EVERY * 2);
await w.ask(["b", "c"]);
}
assert.deepEqual(w.retired, [], "unreadable passes counted toward a stable result");
} finally {
w.done();
}
});
test("asked for again, a retired consumer is created again and announced re-enabled", async () => {
const w = await world(["a", "b", "c"]);
try {
await w.ask(["b", "c"]);
await until(() => w.retired.length === 1);
const before = w.created.length;
await w.ask(["a", "b", "c"]);
await until(() => w.created.length > before);
assert.equal(w.created.at(-1), "a");
await until(() => changes(w).includes("reenabled"));
assert.deepEqual((await w.p.retirement()).retired, []);
} finally {
w.done();
}
});
test("an adapter without retire is retired through remove, as before", async () => {
const w = await world(["a", "b", "c"], { retire: false });
try {
await w.ask(["b", "c"]);
await until(() => w.removed.length === 1);
assert.deepEqual(w.removed, ["a"]);
} finally {
w.done();
}
});
test("four of four waits for a person: approve the exact set retires, another set is refused", async () => {
const w = await world(["a", "b", "c", "d"]);
try {
await w.ask([]);
await until(async () => (await w.p.retirement()).waiting !== null);
await sleep(EVERY * 10);
assert.deepEqual(w.retired, [], "retired over the bound without a person");
assert.deepEqual(changes(w), ["waiting"], "waiting said more than once, or not");
const waiting = w.events[0];
assert.equal(waiting.held, 4);
assert.match(String(waiting.bound), /more than 3, or more than 50% of the 4 held/);
assert.deepEqual((waiting.consumers as { consumer: string }[]).map((c) => c.consumer), ["a", "b", "c", "d"]);
await assert.rejects(w.p.approve(["a", "b", "c"], { why: "x" }), /exactly that set/);
await assert.rejects(w.p.approve(["a", "b", "c", "d"], { why: "" }), /why is required/);
const done = await w.p.approve(["d", "c", "b", "a"], { why: "all four unassigned", by: "operator", via: "mesh-controller" });
assert.deepEqual(done.retired, ["a", "b", "c", "d"]);
assert.deepEqual(w.retired.map((r) => r.as), ["a", "b", "c", "d"]);
assert.deepEqual(changes(w), ["waiting", "approved", "retired"]);
assert.equal(w.events[1].via, "mesh-controller");
assert.equal(w.events[1].by, "operator");
} finally {
w.done();
}
});
test("rejected, the set is kept active and not asked again while the mesh says the same", async () => {
const w = await world(["a", "b", "c", "d"]);
try {
await w.ask([]);
await until(async () => (await w.p.retirement()).waiting !== null);
await assert.rejects(w.p.reject(["a"], { why: "no" }), /exactly that set/);
const kept = await w.p.reject(["a", "b", "c", "d"], { why: "the mesh is wrong", by: "operator" });
assert.deepEqual(kept.kept, ["a", "b", "c", "d"]);
await sleep(EVERY * 12);
assert.deepEqual(w.retired, []);
assert.deepEqual(changes(w), ["waiting", "rejected"]);
const r = await w.p.retirement();
assert.equal(r.waiting, null);
assert.deepEqual(r.rejected?.consumers.map((c) => c.consumer), ["a", "b", "c", "d"]);
// Approving a rejected set later still works.
await w.p.approve(["a", "b", "c", "d"], { why: "on second thought" });
assert.equal(w.retired.length, 4);
} finally {
w.done();
}
});
test("a waiting set the mesh asks for again settles, and nothing is retired", async () => {
const w = await world(["a", "b", "c", "d"]);
try {
await w.ask([]);
await until(async () => (await w.p.retirement()).waiting !== null);
await w.ask(["a", "b", "c", "d"]);
await until(() => changes(w).includes("settled"));
assert.equal((await w.p.retirement()).waiting, null);
assert.deepEqual(w.retired, []);
} finally {
w.done();
}
});
test("delete refuses an active consumer and deletes only the named retired one", async () => {
const w = await world(["a", "b", "c"]);
try {
await w.ask(["c"]);
// Two of three is more than half: approve, then delete one.
await until(async () => (await w.p.retirement()).waiting !== null);
await w.p.approve(["a", "b"], { why: "gone" });
await assert.rejects(w.p.deleteRetired("c", { why: "x", confirm: "c" }), /active, not retired/);
await assert.rejects(w.p.deleteRetired("zz", { why: "x", confirm: "zz" }), /not retired here/);
await assert.rejects(w.p.deleteRetired("a", { why: "x", confirm: "b" }), /confirm must repeat/);
await assert.rejects(w.p.deleteRetired("a", { why: "", confirm: "a" }), /why is required/);
const r = await w.p.deleteRetired("a", { why: "older than a month", by: "operator", confirm: "a" });
assert.deepEqual(r, { deleted: "a", freed_bytes: 42 });
assert.deepEqual(w.deleted, ["a"]);
assert.deepEqual((await w.p.retirement()).retired.map((x) => x.consumer), ["b"]);
assert.equal(w.events.at(-1)?.change, "deleted");
} finally {
w.done();
}
});
test("an addition acts on the first pass", async () => {
const w = await world(["a"]);
try {
const before = w.created.length;
await w.ask(["a", "b"]);
await until(() => w.created.length > before, EVERY * 3);
assert.equal(w.created.at(-1), "b");
} finally {
w.done();
}
});
test("what the backend holds at start is retired by the same rules", async () => {
// "x" was made by the mesh and the mesh stopped asking for it while the provider was down.
const w = await world(["a", "b", "c"], { held: ["a", "b", "c", "x"] });
try {
await until(() => w.retired.length === 1);
assert.equal(w.retired[0].as, "x");
} finally {
w.done();
}
});
test("the four tools are registered for the module and answer", async () => {
resetTools();
const dir = await mkdtemp(join(tmpdir(), "retire-tools-"));
const receives = join(dir, "r.json");
await writeFile(receives, JSON.stringify({ requirement: "r", given: [] }));
const p = runProvisioner("r", { async create() {}, async remove() {} }, { receives, everyMs: EVERY, module: "postgres" });
try {
const mine = collectTools({}).find((c) => c.module === "postgres");
assert.deepEqual(mine?.tools.map((t) => t.name), [
"provisioner_retirement", "provisioner_retire_approve", "provisioner_retire_reject", "provisioner_delete",
]);
const answer = (await mine!.tools[0].run({})) as { resource: string; waiting: unknown };
assert.equal(answer.resource, "r");
assert.equal(answer.waiting, null);
assert.equal(provisionerTools(p).length, 4);
} finally {
p();
resetTools();
}
});
+7 -55
View File
@@ -185,7 +185,7 @@ test("provisioner removes nothing while its contributions file cannot be used (n
removed.push(p.as);
},
},
{ receives, everyMs: 20, releaseEveryMs: 300 },
{ receives, everyMs: 20 },
);
await waitFor(() => created.length === 2, 2000);
@@ -203,65 +203,17 @@ test("provisioner removes nothing while its contributions file cannot be used (n
await new Promise((r) => setTimeout(r, 100));
}
// The file read and saying so still withdraws — within the brake (to-be 45 Phase 2): two at once is
// over its bound, so nothing goes at first, then one at a time as the brake lets them go.
// The file read and saying so is a result — but two of two held is over the bound (ADR 0230):
// nothing is retired without a person, and approving that exact set retires both.
await writeFile(receives, doc([]));
await new Promise((r) => setTimeout(r, 150));
assert.deepEqual(removed, [], "two consumers were withdrawn at once");
await waitFor(() => removed.length === 2, 3000);
for (let i = 0; i < 200 && (await stop.retirement()).waiting === null; i++) await new Promise((r) => setTimeout(r, 10));
assert.notEqual((await stop.retirement()).waiting, null, "the retirement never waited for a person");
assert.deepEqual(removed, [], "two of two consumers were retired without a person");
await stop.approve(["mesh_anchor_mail", "mesh_anchor_forge"], { why: "both unassigned on purpose", by: "test" });
assert.deepEqual([...removed].sort(), ["mesh_anchor_forge", "mesh_anchor_mail"]);
stop();
});
test("provisioner brakes a withdrawal of more than one consumer, and keeps one asked for again (to-be 45 Phase 2)", async () => {
const dir = await mkdtemp(join(tmpdir(), "prov-"));
const created: string[] = [];
const removed: string[] = [];
const receives = join(dir, "postgres-database.json");
const doc = (given: unknown[]): string =>
JSON.stringify({ contributions: 1, requirement: "postgres-database", given });
const seven: unknown[] = [];
for (let i = 1; i <= 7; i++) {
await writeFile(join(dir, `c${i}.secret`), `pw-${i}\n`);
seven.push({ from: `c${i}`, node: "anchor", as: `consumer${i}`, secret: join(dir, `c${i}.secret`) });
}
await writeFile(receives, doc(seven));
const said: string[] = [];
const was = console.error;
console.error = (...args: unknown[]) => said.push(args.join(" "));
const stop = runProvisioner(
"postgres-database",
{
async create(p) {
created.push(p.as);
},
async remove(p) {
removed.push(p.as);
},
},
{ receives, everyMs: 10, releaseEveryMs: 60_000 },
);
try {
await waitFor(() => created.length === 7, 2000);
// A file read whole that names nobody — the shape of 241 its first fix does not catch.
await writeFile(receives, doc([]));
await new Promise((r) => setTimeout(r, 200));
assert.deepEqual(removed, [], "a withdrawal of seven at once was not braked");
assert.ok(said.some((l) => l.includes("WITHDRAWAL BRAKED")), "the brake was not said");
// The file is right again: nothing was withdrawn, and nothing will be.
await writeFile(receives, doc(seven));
await new Promise((r) => setTimeout(r, 100));
assert.deepEqual(removed, []);
// One of seven goes as asked: within the bound.
await writeFile(receives, doc(seven.slice(1)));
await waitFor(() => removed.length === 1, 2000);
assert.deepEqual(removed, ["consumer1"]);
} finally {
stop();
console.error = was;
}
});
test("provisioner applies again what the backend no longer holds, and trusts memory without holds", async () => {
const dir = await mkdtemp(join(tmpdir(), "prov-holds-"));
await writeFile(join(dir, "webapp.secret"), "minted-pw\n");