Merge pull request '0.1.14: the provider loop says a consumer it keeps failing on the bus, and its recovery (hq ADR 0224, issue 272)' (#24) from fix/272-the-provider-loop-says-its-standing into main
This commit was merged in pull request #24.
This commit is contained in:
@@ -39,11 +39,21 @@ never called to retire**; it is the delete of an adapter without `delete`, reach
|
||||
person's `provisioner_delete`. 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
|
||||
(`waiting`). Asked for again, a retired consumer is created — re-enabled
|
||||
— at once. 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.
|
||||
|
||||
**A provider that keeps failing a consumer says so on the bus** (novox/hq ADR 0224). A consumer whose
|
||||
create, check (`holds`) or secret has failed for five minutes without one success between
|
||||
(`failingAfterMs`) is announced `provisioner.failing` — the consumer, its machine, the class of error
|
||||
(`credentials-rejected`, `unreachable`, `secret-unreadable`, `refused`; an adapter's `classify` is
|
||||
believed over the words), the error with the password removed, since when and how many attempts — and
|
||||
again every fifteen minutes while it lasts. The first success after it is `provisioner.recovered`, as is
|
||||
a consumer no longer asked for or retired, and the first success for each consumer since the process
|
||||
started. Every event goes through the `announce` option, which defaults to the SDK's `emit`: a provider
|
||||
says them without asking, and a failed publish is said in its journal, never failing the act.
|
||||
|
||||
### `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
|
||||
|
||||
Generated
+2
-2
@@ -1,12 +1,12 @@
|
||||
{
|
||||
"name": "@novox/mesh-sdk",
|
||||
"version": "0.1.13",
|
||||
"version": "0.1.14",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "@novox/mesh-sdk",
|
||||
"version": "0.1.13",
|
||||
"version": "0.1.14",
|
||||
"devDependencies": {
|
||||
"@types/node": "^22.0.0",
|
||||
"typescript": "^5.6.0"
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@novox/mesh-sdk",
|
||||
"version": "0.1.13",
|
||||
"version": "0.1.14",
|
||||
"description": "The stable spine a Novox Mesh module's own code builds against.",
|
||||
"type": "module",
|
||||
"exports": {
|
||||
|
||||
+160
-11
@@ -10,10 +10,20 @@
|
||||
// does not hand one back — the consumer already receives its copy through the mesh's own channel.
|
||||
// This is why the old $MESH_SEAL_KEY, the symmetric seal, and the *.grant.json / *.credential files
|
||||
// are gone: they described a second, contradictory credential path the mesh does not have.
|
||||
//
|
||||
// **A provider that keeps failing a consumer says so on the bus (novox/hq ADR 0224).** A consumer
|
||||
// whose create, check or secret has failed without one success in between for `failingAfterMs` is
|
||||
// announced as `provisioner.failing` — naming the consumer, its machine and the class of error — and
|
||||
// again every `sayAgainEveryMs` while it lasts; the first success after that is
|
||||
// `provisioner.recovered`, as is a consumer the mesh stopped asking for, and the first success for
|
||||
// each consumer since this process started. The controller keeps the newest per provider and consumer
|
||||
// and `status` names it. Until novox/hq issue 272 only the Go copy of this loop (postgres, keycloak)
|
||||
// said it; every provider on this one failed its consumers in its journal alone.
|
||||
|
||||
import { readFile } from "node:fs/promises";
|
||||
|
||||
import type { ToolDefinition } from "../contracts/index.js";
|
||||
import { emit } from "../events/index.js";
|
||||
import { registerModuleTools } from "../tools/index.js";
|
||||
|
||||
/** One consumer's resource to bring into being — everything the mesh derived and delivered. */
|
||||
@@ -86,6 +96,9 @@ export interface Adapter {
|
||||
/** 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[]>;
|
||||
/** Optional: the class of one of the adapter's own errors (novox/hq ADR 0224), when its code knows
|
||||
* better than the error's words; "" or undefined falls back to `classOf` of the text. */
|
||||
classify?(err: unknown): string | undefined;
|
||||
}
|
||||
|
||||
/** One retired consumer, as `provisioner_retirement` answers it. */
|
||||
@@ -139,6 +152,21 @@ export type Provisioner = (() => void) & {
|
||||
/** The one event a provider says about retirement (novox/hq ADR 0230). */
|
||||
export const RETIREMENT_EVENT = "provisioner.retirement";
|
||||
|
||||
/** The events a provider's standing toward one consumer is announced as (novox/hq ADR 0224). The
|
||||
* controller derives the permission to emit them for every module that receives contributions. */
|
||||
export const FAILING_EVENT = "provisioner.failing";
|
||||
export const RECOVERED_EVENT = "provisioner.recovered";
|
||||
|
||||
/** The classes a failing consumer is announced with: what a person reading `status` needs to know
|
||||
* before reading the journal (novox/hq ADR 0224). An adapter may say better (`classify`). */
|
||||
export const CLASS_CREDENTIALS = "credentials-rejected";
|
||||
export const CLASS_UNREACHABLE = "unreachable";
|
||||
export const CLASS_SECRET = "secret-unreadable";
|
||||
export const CLASS_REFUSED = "refused";
|
||||
|
||||
/** How long a consumer fails without one success before it is announced failing (ADR 0224). */
|
||||
export const FAILING_AFTER_MS = 5 * 60_000;
|
||||
|
||||
export interface ProvisionerOptions {
|
||||
/** The contributions file the mesh writes for this resource (the module's `receives` path).
|
||||
* Defaults to $MESH_RECEIVES. */
|
||||
@@ -165,10 +193,16 @@ export interface ProvisionerOptions {
|
||||
/** 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. */
|
||||
/** How often a waiting retirement, and a consumer still failing, 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. */
|
||||
/** How long, in ms, a consumer fails without one success before it is announced
|
||||
* `provisioner.failing` (novox/hq ADR 0224). Defaults to FAILING_AFTER_MS, five minutes. */
|
||||
failingAfterMs?: number;
|
||||
/** Publishes one of the provider's events (`provisioner.failing`, `provisioner.recovered`,
|
||||
* `provisioner.retirement`). Defaults to the SDK's `emit`, so every provider says them on the bus
|
||||
* without asking (novox/hq issue 272); a failed publish is said in the journal and never fails
|
||||
* the act it reports. */
|
||||
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;
|
||||
@@ -227,6 +261,8 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
const retireAtOnce = opts.retireAtOnce ?? 3;
|
||||
const retireFraction = opts.retireFraction ?? 0.5;
|
||||
const sayAgainEveryMs = opts.sayAgainEveryMs ?? 15 * 60_000;
|
||||
const failingAfterMs = opts.failingAfterMs ?? FAILING_AFTER_MS;
|
||||
const announce = opts.announce ?? ((event: string, body: Record<string, unknown>) => emit(event, body));
|
||||
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
|
||||
@@ -245,6 +281,13 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
// itself. Each repeat doubles the wait before asking again, so such a bug costs one re-apply an
|
||||
// hour rather than one a minute, and it is said loudly rather than quietly repeated.
|
||||
const lost = new Map<string, { times: number; nextAt: number }>();
|
||||
// Each consumer's unbroken run of failures — since when, how often, the last error, and when it was
|
||||
// last announced (0: not yet) — and every consumer that has succeeded since this process started
|
||||
// (novox/hq ADR 0224).
|
||||
const trouble = new Map<string, {
|
||||
node?: string; since: number; attempts: number; cls: string; text: string; saidAt: number;
|
||||
}>();
|
||||
const cleared = new Set<string>();
|
||||
|
||||
// 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.
|
||||
@@ -311,6 +354,7 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
`file and who this process runs as (novox/hq issue 225)`,
|
||||
);
|
||||
}
|
||||
await failed(g.as, g.node, CLASS_SECRET, `secret not readable (${g.secret}): ${err}`);
|
||||
continue;
|
||||
}
|
||||
waiting.delete(g.as);
|
||||
@@ -328,11 +372,15 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
try {
|
||||
if (await withTimeout(adapter.holds!(p), holdsTimeoutMs)) {
|
||||
lost.delete(g.as);
|
||||
await succeeded(g.as);
|
||||
continue;
|
||||
}
|
||||
} catch (err) {
|
||||
// Unable to ask is not evidence of loss. Kept as applied, asked again next time.
|
||||
console.error(`[provisioner:${resource}] ${g.as}: could not check the backend, will ask again: ${scrub(err, password)}`);
|
||||
// Unable to ask is not evidence of loss. Kept as applied, asked again next time — but a
|
||||
// check that keeps failing is a consumer failed all the same (ADR 0224, issue 179).
|
||||
const text = scrub(err, password);
|
||||
console.error(`[provisioner:${resource}] ${g.as}: could not check the backend, will ask again: ${text}`);
|
||||
await failed(g.as, g.node, classed(err, text), text);
|
||||
// A backend that did not answer in time will not answer for the next consumer either, and
|
||||
// each would cost a timeout, so the rest of this pass is not asked. Any other failure may
|
||||
// be this consumer's alone, and the others are still asked.
|
||||
@@ -356,6 +404,7 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
await adapter.reenable({ as: g.as, derived: was.derived ?? p.derived });
|
||||
}
|
||||
await adapter.create(p);
|
||||
await succeeded(g.as);
|
||||
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).
|
||||
@@ -376,7 +425,9 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`[provisioner:${resource}] ${g.as}: create failed, will retry: ${scrub(err, password)}`);
|
||||
const text = scrub(err, password);
|
||||
console.error(`[provisioner:${resource}] ${g.as}: create failed, will retry: ${text}`);
|
||||
await failed(g.as, g.node, classed(err, text), text);
|
||||
// A failed create is not a disagreement: asked again at the next check, the count unchanged.
|
||||
if (reapplying !== undefined) lost.set(g.as, { times: reapplying - 1, nextAt: 0 });
|
||||
}
|
||||
@@ -386,6 +437,12 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
// `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();
|
||||
// A consumer the mesh stopped asking for that this provider holds nothing for is no longer failed
|
||||
// by anyone: said, so the standing the controller keeps for it is cleared rather than left naming a
|
||||
// consumer that is gone. One still held is said recovered when it is retired (ADR 0224, 0230).
|
||||
for (const as of [...trouble.keys()]) {
|
||||
if (!wantByAs.has(as) && !applied.has(as)) await recovered(as, "no longer asked for");
|
||||
}
|
||||
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`);
|
||||
@@ -464,6 +521,7 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
}
|
||||
applied.delete(as);
|
||||
lost.delete(as);
|
||||
await recovered(as, "retired");
|
||||
retired.set(as, { node: was.node, at: at.getTime(), why, derived: was.derived, access: adapter.retire ? "disabled" : "kept" });
|
||||
done.push(as);
|
||||
} catch (err) {
|
||||
@@ -485,16 +543,86 @@ export function runProvisioner(resource: string, adapter: Adapter, opts: Provisi
|
||||
|
||||
// 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;
|
||||
await said(RETIREMENT_EVENT, {
|
||||
provider: resource, "provider-node": node, change, bound: bound(applied.size), at: iso(Date.now()), ...body,
|
||||
}, change);
|
||||
}
|
||||
|
||||
async function said(event: string, body: Record<string, unknown>, what = ""): Promise<void> {
|
||||
try {
|
||||
await opts.announce(RETIREMENT_EVENT, {
|
||||
provider: resource, "provider-node": node, change, bound: bound(applied.size), at: iso(Date.now()), ...body,
|
||||
});
|
||||
await announce(event, body);
|
||||
} catch (err) {
|
||||
say(`could not announce ${RETIREMENT_EVENT} ${change}: ${err}`);
|
||||
say(`could not announce ${event}${what ? ` ${what}` : ""}: ${err}`);
|
||||
}
|
||||
}
|
||||
|
||||
// An error's class: the adapter's word when it has one, else read from the text (ADR 0224).
|
||||
function classed(err: unknown, text: string): string {
|
||||
try {
|
||||
const own = adapter.classify?.(err);
|
||||
if (own) return own;
|
||||
} catch {
|
||||
// An adapter that cannot class its own error is read by the words.
|
||||
}
|
||||
// A check that did not answer in time is a backend out of reach, whatever its words.
|
||||
if (err instanceof HoldsTimeout) return CLASS_UNREACHABLE;
|
||||
return classOf(text);
|
||||
}
|
||||
|
||||
// One more failure in a consumer's unbroken run, announced once the run has lasted failingAfterMs,
|
||||
// then again every sayAgainEveryMs while it lasts (novox/hq ADR 0224).
|
||||
async function failed(as: string, consumerNode: string | undefined, cls: string, text: string): Promise<void> {
|
||||
const now = Date.now();
|
||||
let s = trouble.get(as);
|
||||
if (!s) {
|
||||
s = { since: now, attempts: 0, cls, text, saidAt: 0 };
|
||||
trouble.set(as, s);
|
||||
}
|
||||
s.node = consumerNode;
|
||||
s.cls = cls;
|
||||
s.text = text;
|
||||
s.attempts++;
|
||||
if (now - s.since < failingAfterMs) return;
|
||||
if (s.saidAt !== 0 && now - s.saidAt < sayAgainEveryMs) return;
|
||||
if (s.saidAt === 0) {
|
||||
say(
|
||||
`${as}: FAILING for ${Math.round((now - s.since) / 1000)}s (${s.attempts} attempts, ${cls}): ${text}. ` +
|
||||
`Announced as ${FAILING_EVENT}; \`status\` names it until it succeeds (novox/hq ADR 0224)`,
|
||||
);
|
||||
}
|
||||
s.saidAt = now;
|
||||
await said(FAILING_EVENT, {
|
||||
provider: resource, "provider-node": node, consumer: as, node: consumerNode ?? "",
|
||||
class: cls, error: clip(text), since: iso(s.since), attempts: s.attempts,
|
||||
});
|
||||
}
|
||||
|
||||
// A success ends a consumer's run of failures; one that was announced is announced recovered. And
|
||||
// the first success for a consumer since this process started is announced too, failing or not: a
|
||||
// provider that announced a failure and was restarted has forgotten it, and without this the
|
||||
// controller would name the consumer failing for ever after it recovered unheard.
|
||||
async function succeeded(as: string): Promise<void> {
|
||||
if (!trouble.has(as) && !cleared.has(as)) {
|
||||
cleared.add(as);
|
||||
await said(RECOVERED_EVENT, { provider: resource, "provider-node": node, consumer: as, why: "first-success" });
|
||||
return;
|
||||
}
|
||||
cleared.add(as);
|
||||
await recovered(as, "");
|
||||
}
|
||||
|
||||
async function recovered(as: string, why: string): Promise<void> {
|
||||
const s = trouble.get(as);
|
||||
if (!s) return;
|
||||
trouble.delete(as);
|
||||
if (s.saidAt === 0) return; // never announced, so there is nothing to take back
|
||||
if (!why) say(`${as}: recovered after ${Math.round((Date.now() - s.since) / 1000)}s and ${s.attempts} failed attempt(s)`);
|
||||
await said(RECOVERED_EVENT, {
|
||||
provider: resource, "provider-node": node, consumer: as, node: s.node ?? "",
|
||||
since: iso(s.since), attempts: s.attempts, ...(why ? { why } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
// 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> {
|
||||
@@ -713,6 +841,27 @@ async function readContributions(
|
||||
return (doc.given as Contribution[]).filter((g) => g.as && g.secret);
|
||||
}
|
||||
|
||||
/** An error's class read from its text — the words the backends the mesh runs use (novox/hq ADR
|
||||
* 0224): the same words, in the same order, as the Go copy of this loop reads. */
|
||||
export function classOf(text: string): string {
|
||||
const t = text.toLowerCase();
|
||||
const credentials = ["invalid_grant", "invalid user credentials", "password authentication failed",
|
||||
"authentication failed", "unauthorized", " 401"];
|
||||
if (credentials.some((w) => t.includes(w))) return CLASS_CREDENTIALS;
|
||||
const unreachable = ["connection refused", "no such host", "i/o timeout", "deadline exceeded",
|
||||
"connection reset", "network is unreachable", "no route to host", "eof",
|
||||
// Node's own words for the same faults.
|
||||
"econnrefused", "enotfound", "etimedout", "econnreset", "ehostunreach", "enetunreach", "eai_again"];
|
||||
if (unreachable.some((w) => t.includes(w))) return CLASS_UNREACHABLE;
|
||||
return CLASS_REFUSED;
|
||||
}
|
||||
|
||||
/** An announced error kept to what belongs in a status line. */
|
||||
function clip(text: string): string {
|
||||
const most = 300;
|
||||
return text.length <= most ? text : `${text.slice(0, most)}…`;
|
||||
}
|
||||
|
||||
/** A check that did not answer in time: the backend, not the consumer, is the likely cause. */
|
||||
class HoldsTimeout extends Error {}
|
||||
|
||||
|
||||
+2
-1
@@ -64,7 +64,8 @@ async function world(
|
||||
};
|
||||
w.p = runProvisioner("postgres-database", adapter, {
|
||||
receives, everyMs: EVERY, node: "anchor", stableForMs: opts.stableForMs ?? 0,
|
||||
announce: (event, body) => { assert.equal(event, RETIREMENT_EVENT); w.events.push(body); },
|
||||
// Only retirement is these tests'; the standing (provisioner.failing, .recovered) is standing.test.ts's.
|
||||
announce: (event, body) => { if (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; };
|
||||
|
||||
@@ -0,0 +1,272 @@
|
||||
// A provider that keeps failing a consumer says so on the bus (novox/hq ADR 0224, issue 272): not on
|
||||
// the first failure, which may be a restart; after failingAfterMs of failures with no success between;
|
||||
// again every sayAgainEveryMs while it lasts; and recovered on the first success, or when the consumer
|
||||
// goes. The same behaviour as the Go copy of the loop's standing_test.go, with the clock shortened.
|
||||
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, classOf, FAILING_EVENT, RECOVERED_EVENT, RETIREMENT_EVENT,
|
||||
CLASS_CREDENTIALS, CLASS_UNREACHABLE, CLASS_SECRET, CLASS_REFUSED, type Adapter,
|
||||
} from "../dist/provisioner/index.js";
|
||||
import { useBroker } from "../dist/messaging/index.js";
|
||||
|
||||
const EVERY = 10;
|
||||
const FAILING_AFTER = 150;
|
||||
const SAY_AGAIN = 400;
|
||||
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
|
||||
|
||||
async function until(cond: () => boolean, ms = 4000): Promise<void> {
|
||||
const start = Date.now();
|
||||
while (!cond()) {
|
||||
if (Date.now() - start > ms) throw new Error("timed out waiting");
|
||||
await sleep(5);
|
||||
}
|
||||
}
|
||||
|
||||
interface Said { event: string; body: Record<string, unknown> }
|
||||
|
||||
interface World {
|
||||
dir: string;
|
||||
ask(list: { as: string; node?: string; values?: Record<string, unknown> }[]): Promise<void>;
|
||||
failing: Error | null;
|
||||
holdsErr: Error | null;
|
||||
created: string[];
|
||||
/** Every event but the first success since start, which has its own test. */
|
||||
said: Said[];
|
||||
all: Said[];
|
||||
stop(): void;
|
||||
}
|
||||
|
||||
async function world(opts: { classify?: boolean; announce?: false; stableForMs?: number } = {}): Promise<World> {
|
||||
const dir = await mkdtemp(join(tmpdir(), "standing-"));
|
||||
const receives = join(dir, "postgres-database.json");
|
||||
const w = { dir, failing: null, holdsErr: null, created: [], said: [], all: [] } as unknown as World;
|
||||
w.ask = async (list) => {
|
||||
const given = [];
|
||||
for (const g of list) {
|
||||
await writeFile(join(dir, `${g.as}.secret`), `pw-${g.as}\n`);
|
||||
given.push({ from: g.as, node: g.node ?? "anchor", as: g.as, secret: join(dir, `${g.as}.secret`), values: g.values });
|
||||
}
|
||||
await writeFile(receives, JSON.stringify({ contributions: 1, requirement: "postgres-database", given }));
|
||||
};
|
||||
const adapter: Adapter = {
|
||||
async create(p) {
|
||||
if (w.failing) throw w.failing;
|
||||
w.created.push(p.as);
|
||||
},
|
||||
async remove() {},
|
||||
async retire() {},
|
||||
async holds() {
|
||||
if (w.holdsErr) throw w.holdsErr;
|
||||
return true;
|
||||
},
|
||||
...(opts.classify ? { classify: () => "its-own" } : {}),
|
||||
};
|
||||
const was = console.error;
|
||||
console.error = () => {};
|
||||
const stop = runProvisioner("postgres-database", adapter, {
|
||||
receives, everyMs: EVERY, verifyEveryMs: 0, node: "home-server",
|
||||
failingAfterMs: FAILING_AFTER, sayAgainEveryMs: SAY_AGAIN, stableForMs: opts.stableForMs ?? 0,
|
||||
...(opts.announce === false ? {} : {
|
||||
announce: (event: string, body: Record<string, unknown>) => {
|
||||
w.all.push({ event, body });
|
||||
if (body.why !== "first-success") w.said.push({ event, body });
|
||||
},
|
||||
}),
|
||||
});
|
||||
w.stop = () => {
|
||||
stop();
|
||||
console.error = was;
|
||||
void rm(dir, { recursive: true, force: true });
|
||||
};
|
||||
return w;
|
||||
}
|
||||
|
||||
const events = (said: Said[]) => said.filter((s) => s.event !== RETIREMENT_EVENT).map((s) => s.event).join(",");
|
||||
|
||||
test("a consumer failed for minutes is announced, naming it, its machine and the class, again while it lasts, and recovered", async () => {
|
||||
const w = await world();
|
||||
try {
|
||||
w.failing = new Error(`token request failed: 401 {"error":"invalid_grant"} for pw-a`);
|
||||
await w.ask([{ as: "a", node: "anchor" }]);
|
||||
await sleep(FAILING_AFTER / 2);
|
||||
assert.equal(events(w.said), "", "announced before failingAfterMs");
|
||||
await until(() => w.said.length === 1);
|
||||
const b = w.said[0].body;
|
||||
assert.equal(w.said[0].event, FAILING_EVENT);
|
||||
assert.equal(b.consumer, "a");
|
||||
assert.equal(b.node, "anchor");
|
||||
assert.equal(b.class, CLASS_CREDENTIALS);
|
||||
assert.equal(b.provider, "postgres-database");
|
||||
assert.equal(b["provider-node"], "home-server");
|
||||
assert.ok((b.attempts as number) > 3, `attempts ${b.attempts}`);
|
||||
assert.match(String(b.since), /^\d{4}-\d{2}-\d{2}T/);
|
||||
assert.doesNotMatch(String(b.error), /pw-a/, "the consumer's password travelled");
|
||||
|
||||
// Said again while it lasts, not every pass.
|
||||
await sleep(SAY_AGAIN / 2);
|
||||
assert.equal(events(w.said), FAILING_EVENT, "repeated too soon");
|
||||
await until(() => w.said.length === 2);
|
||||
assert.equal(events(w.said), `${FAILING_EVENT},${FAILING_EVENT}`);
|
||||
|
||||
// The first success takes it back.
|
||||
w.failing = null;
|
||||
await until(() => w.said.length === 3);
|
||||
assert.equal(w.said[2].event, RECOVERED_EVENT);
|
||||
assert.equal(w.said[2].body.consumer, "a");
|
||||
assert.equal(w.said[2].body.node, "anchor");
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("one success between failures starts the run again", async () => {
|
||||
const w = await world();
|
||||
try {
|
||||
w.failing = new Error("connection refused");
|
||||
await w.ask([{ as: "a" }]);
|
||||
await sleep(FAILING_AFTER * 0.6);
|
||||
w.failing = null;
|
||||
await until(() => w.created.length === 1);
|
||||
w.failing = new Error("connection refused");
|
||||
await w.ask([{ as: "a", values: { name: "changed" } }]);
|
||||
await sleep(FAILING_AFTER * 0.6);
|
||||
assert.equal(events(w.said), "", "two short runs are not one long one");
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
|
||||
// The check that failed for a day on 2026-10-05 (issue 179): consumers already made, every check
|
||||
// refused. A check that cannot be asked is a failure too.
|
||||
test("a check that keeps failing is a failure too", async () => {
|
||||
const w = await world();
|
||||
try {
|
||||
await w.ask([{ as: "a" }]);
|
||||
await until(() => w.created.length === 1);
|
||||
w.holdsErr = new Error("401 invalid_grant");
|
||||
await until(() => w.said.length === 1);
|
||||
assert.equal(w.said[0].event, FAILING_EVENT);
|
||||
assert.equal(w.said[0].body.class, CLASS_CREDENTIALS);
|
||||
w.holdsErr = null;
|
||||
await until(() => w.said.length === 2);
|
||||
assert.equal(events(w.said), `${FAILING_EVENT},${RECOVERED_EVENT}`);
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("an unreadable secret is announced as such", async () => {
|
||||
const w = await world();
|
||||
try {
|
||||
await w.ask([{ as: "a" }]);
|
||||
await rm(join(w.dir, "a.secret"));
|
||||
await until(() => w.said.length === 1);
|
||||
assert.equal(w.said[0].event, FAILING_EVENT);
|
||||
assert.equal(w.said[0].body.class, CLASS_SECRET);
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("a consumer no longer asked for is no longer failing", async () => {
|
||||
const w = await world();
|
||||
try {
|
||||
w.failing = new Error("boom");
|
||||
await w.ask([{ as: "a" }, { as: "b" }]);
|
||||
await until(() => w.said.length === 2);
|
||||
assert.equal(events(w.said), `${FAILING_EVENT},${FAILING_EVENT}`);
|
||||
await w.ask([{ as: "a" }]);
|
||||
await until(() => w.said.some((s) => s.event === RECOVERED_EVENT));
|
||||
const last = w.said.find((s) => s.event === RECOVERED_EVENT)!;
|
||||
assert.equal(last.body.consumer, "b");
|
||||
assert.equal(last.body.why, "no longer asked for");
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("a failing consumer the mesh stopped asking for is said recovered when it is retired", async () => {
|
||||
const w = await world();
|
||||
try {
|
||||
await w.ask([{ as: "a" }, { as: "b" }, { as: "c" }]);
|
||||
await until(() => w.created.length === 3);
|
||||
w.holdsErr = new Error("boom");
|
||||
await until(() => w.said.filter((s) => s.event === FAILING_EVENT).length >= 3);
|
||||
// c is no longer asked for while it is still failing; the others then recover.
|
||||
await w.ask([{ as: "a" }, { as: "b" }]);
|
||||
w.holdsErr = null;
|
||||
await until(() => w.said.some((s) => s.event === RECOVERED_EVENT && s.body.consumer === "c"));
|
||||
const c = w.said.find((s) => s.event === RECOVERED_EVENT && s.body.consumer === "c")!;
|
||||
assert.equal(c.body.why, "retired");
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
|
||||
test("an error is classed by its words", () => {
|
||||
const table: Record<string, string> = {
|
||||
'Keycloak token request failed: 401 {"error":"invalid_grant"}': CLASS_CREDENTIALS,
|
||||
'FATAL: password authentication failed for user "postgres"': CLASS_CREDENTIALS,
|
||||
"dial tcp 127.0.0.1:5432: connect: connection refused": CLASS_UNREACHABLE,
|
||||
"context deadline exceeded": CLASS_UNREACHABLE,
|
||||
"Error: connect ECONNREFUSED 127.0.0.1:9000": CLASS_UNREACHABLE,
|
||||
"getaddrinfo ENOTFOUND minio": CLASS_UNREACHABLE,
|
||||
'extension "nope" is not available': CLASS_REFUSED,
|
||||
};
|
||||
for (const [text, want] of Object.entries(table)) assert.equal(classOf(text), want, text);
|
||||
});
|
||||
|
||||
test("an adapter that classes its own errors is believed", async () => {
|
||||
const w = await world({ classify: true });
|
||||
try {
|
||||
w.failing = new Error("anything");
|
||||
await w.ask([{ as: "a" }]);
|
||||
await until(() => w.said.length === 1);
|
||||
assert.equal(w.said[0].body.class, "its-own");
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
|
||||
// A provider restarted after announcing a failure has forgotten it; its first success for each
|
||||
// consumer is announced, so the controller clears what it kept rather than naming it for ever.
|
||||
test("the first success since start is announced once", async () => {
|
||||
const w = await world();
|
||||
try {
|
||||
await w.ask([{ as: "a" }]);
|
||||
await until(() => w.all.length >= 1);
|
||||
await sleep(EVERY * 10);
|
||||
assert.equal(w.all.length, 1, JSON.stringify(w.all));
|
||||
assert.equal(w.all[0].event, RECOVERED_EVENT);
|
||||
assert.equal(w.all[0].body.why, "first-success");
|
||||
assert.equal(w.all[0].body.consumer, "a");
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
|
||||
// No provider on main passed `announce`, so none said anything (issue 272): the loop says it through
|
||||
// the SDK's own emit unless it is given another way.
|
||||
test("without announce the standing is said through the SDK's emit", async () => {
|
||||
const published: { key: string; body: Record<string, unknown> }[] = [];
|
||||
useBroker(() => ({
|
||||
async publish(env: { key: string; body: Record<string, unknown> }) { published.push({ key: env.key, body: env.body }); },
|
||||
async subscribe() { return () => {}; },
|
||||
}) as never);
|
||||
const w = await world({ announce: false });
|
||||
try {
|
||||
w.failing = new Error("connection refused");
|
||||
await w.ask([{ as: "a" }]);
|
||||
await until(() => published.some((p) => p.key === FAILING_EVENT));
|
||||
const p = published.find((x) => x.key === FAILING_EVENT)!;
|
||||
assert.equal(p.body.consumer, "a");
|
||||
assert.equal(p.body.class, CLASS_UNREACHABLE);
|
||||
} finally {
|
||||
w.stop();
|
||||
}
|
||||
});
|
||||
Reference in New Issue
Block a user