// 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 { runProviderLoop, classOf, FAILING_EVENT, RECOVERED_EVENT, RETIREMENT_EVENT, CLASS_CREDENTIALS, CLASS_UNREACHABLE, CLASS_SECRET, CLASS_REFUSED, type Adapter, } from "../dist/provider-loop/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 { 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 } interface World { dir: string; ask(list: { as: string; node?: string; values?: Record }[]): Promise; 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 { 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 = runProviderLoop("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) => { 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 = { '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 }[] = []; useBroker(() => ({ async publish(env: { key: string; body: Record }) { 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(); } });