Describe the client on its own terms

Same cleanup: the comments explained each decision by contrast with what
came before instead of stating it. The certificate constraint stays — it is
a fact about the mesh's certificates, not a comparison.
This commit is contained in:
2026-09-26 23:51:00 +02:00
parent 19560ca6a7
commit 2197c36fef
+21 -25
View File
@@ -5,8 +5,8 @@
// That is why a module built before any of this runs on the new runtime without a rebuild, and // That is why a module built before any of this runs on the new runtime without a rebuild, and
// why the sdk's own diff for the whole bus change is three comments. // why the sdk's own diff for the whole bus change is three comments.
// //
// What changes is underneath: exchanges and per-tool queues become subjects, and durability // Underneath, everything is a subject and durability is JetStream (novox/hq design 25,
// becomes JetStream (novox/hq design 25, design 29). // design 29).
// //
// mesh.mod.<module>.event.<type> an event this module emits // mesh.mod.<module>.event.<type> an event this module emits
// mesh.mod.<module>.tool.<tool> a tool this module serves // mesh.mod.<module>.tool.<tool> a tool this module serves
@@ -23,8 +23,8 @@ import type { Broker, Envelope, EventHeaders } from "@novox/mesh-sdk/messaging";
const sc = StringCodec(); const sc = StringCodec();
/** Requests wait this long for an answer before failing, matching the AMQP client's behaviour so /** Requests wait this long for an answer before failing. Unchanged from what modules already
* a module's timeout handling does not change with the transport. */ * expect, so a module's timeout handling is not something the bus quietly redefines. */
const REQUEST_TIMEOUT_MS = 30_000; const REQUEST_TIMEOUT_MS = 30_000;
export class PinMismatchError extends Error {} export class PinMismatchError extends Error {}
@@ -42,8 +42,8 @@ export interface Credential {
} }
/** Whether a connection failure is worth retrying, or is a fact about this configuration that /** Whether a connection failure is worth retrying, or is a fact about this configuration that
* retrying cannot change. Mirrors the AMQP client's judgement so the runtime's supervisor does * retrying cannot change. The runtime's supervisor asks this and does not need to know what it
* not have to know which transport it is on. */ * is connected to. */
export function fatalBrokerReason(err: unknown): string | null { export function fatalBrokerReason(err: unknown): string | null {
if (err instanceof PinMismatchError) return "the bus's certificate does not match the pin"; if (err instanceof PinMismatchError) return "the bus's certificate does not match the pin";
const e = err as { code?: string; message?: string }; const e = err as { code?: string; message?: string };
@@ -85,8 +85,7 @@ export async function connectNats(
name: `${cred.node ?? "?"}.${self}`, name: `${cred.node ?? "?"}.${self}`,
tls: cred.fingerprint ? await pinnedTls(cred.url, cred.fingerprint) : undefined, tls: cred.fingerprint ? await pinnedTls(cred.url, cred.fingerprint) : undefined,
// Reconnect forever: the bus being restarted is an upgrade, not a reason for every module on // Reconnect forever: the bus being restarted is an upgrade, not a reason for every module on
// the mesh to exit. The AMQP client's supervisor did this a level up; here the library does // the mesh to exit. `close()` stays the only thing that ends the connection.
// it, and `close()` is still the only thing that ends the connection.
maxReconnectAttempts: -1, maxReconnectAttempts: -1,
}); });
const js = conn.jetstream(); const js = conn.jetstream();
@@ -116,7 +115,7 @@ export async function connectNats(
* Answer a question. * Answer a question.
* *
* A queue group, so several nodes may serve one tool and exactly one of them answers each * A queue group, so several nodes may serve one tool and exactly one of them answers each
* call — the same property the AMQP client got from a shared durable queue. * call.
*/ */
async handle<Req, Res>(key: string, handler: (body: Req) => Promise<Res>): Promise<() => void> { async handle<Req, Res>(key: string, handler: (body: Req) => Promise<Res>): Promise<() => void> {
const sub = conn.subscribe(toolSubject(key, self), { queue: `serve.${self}` }); const sub = conn.subscribe(toolSubject(key, self), { queue: `serve.${self}` });
@@ -144,17 +143,15 @@ export async function connectNats(
* *
* Published into JetStream and awaited, so a publish the bus never accepted fails the emit * Published into JetStream and awaited, so a publish the bus never accepted fails the emit
* rather than vanishing — at-least-once starts at the emitter, not only the consumer * rather than vanishing — at-least-once starts at the emitter, not only the consumer
* (ADR 0042), which is what the AMQP client's confirm channel was for. * (ADR 0042).
* *
* `msgID` is the event's own id, so a redelivery after a crash between publishing and * `msgID` is the event's own id, so a redelivery after a crash between publishing and
* acknowledging is de-duplicated by the server inside its window rather than seen twice. * acknowledging is de-duplicated by the server inside its window rather than seen twice.
*/ */
async publish<T>(env: Envelope<T>): Promise<void> { async publish<T>(env: Envelope<T>): Promise<void> {
// **The body is the payload and the metadata rides as headers**, exactly as on AMQP // **The body is the payload and the metadata rides as headers** (ADR 0042). That shape
// (ADR 0042). NATS has headers of its own, so the envelope's shape on the wire is // is what the conformance suite pins: an implementation that nested the whole envelope in
// preserved rather than re-encoded — which matters because that shape is what the // the body would pass every one of its own tests and agree with nobody.
// conformance suite pins, and an implementation that nested the whole envelope in the
// body would pass every one of its own tests and agree with nobody.
const meta = (env.headers ?? {}) as Record<string, string>; const meta = (env.headers ?? {}) as Record<string, string>;
const h = natsHeaders(); const h = natsHeaders();
for (const [k, v] of Object.entries(meta)) { for (const [k, v] of Object.entries(meta)) {
@@ -288,13 +285,12 @@ function normalizeFingerprint(fingerprint: string): string {
* A certificate authority is not consulted: the mesh issued this and knows its fingerprint, * A certificate authority is not consulted: the mesh issued this and knows its fingerprint,
* which is stronger than trusting whoever a machine's trust store happens to contain. * which is stronger than trusting whoever a machine's trust store happens to contain.
* *
* **One behaviour differs from the AMQP client, and it is a constraint on the mesh rather than a * **A constraint on the mesh, not a detail of this file.** Pinning the exact certificate makes
* detail of this file.** That client passed `checkServerIdentity: () => undefined`, because * hostname verification redundant in principle, but the NATS client exposes no hook to replace
* pinning the exact certificate makes hostname verification redundant. The NATS client exposes no * it — its TLS options are file paths and PEM strings, with no verify callback. So the
* such hook — its TLS options are file paths and PEM strings, with no verify callback — so the * certificate the mesh issues the bus **must carry a subject-alternative name matching the
* certificate the mesh issues the bus **must carry a subject-alternative name matching the address * address nodes dial it by**. The fingerprint check below still happens and is still the real
* nodes dial it by**. The fingerprint check below still happens and is still the real guarantee; * guarantee; what cannot be switched off is the check *beside* it.
* what cannot be switched off is the check *beside* it.
*/ */
async function pinnedTls(rawUrl: string, fingerprint: string): Promise<{ ca: string }> { async function pinnedTls(rawUrl: string, fingerprint: string): Promise<{ ca: string }> {
const url = new URL(rawUrl.includes("://") ? rawUrl : `nats://${rawUrl}`); const url = new URL(rawUrl.includes("://") ? rawUrl : `nats://${rawUrl}`);
@@ -320,9 +316,9 @@ async function pinnedTls(rawUrl: string, fingerprint: string): Promise<{ ca: str
return { ca: pem }; return { ca: pem };
} }
/** The mesh's topic matching, unchanged from AMQP: `*` is one token, `#` the rest. Kept because /** The mesh's topic matching: `*` is one token, `#` the rest. This is the module's vocabulary —
* it is the module's vocabulary — a module's `consumes` pattern reads the same as it always did, * a module's `consumes` pattern is matched here, and the subject it becomes is the mesh's
* and the subject it becomes is the mesh's business. */ * business, not the module's. */
export function topicMatches(pattern: string, key: string): boolean { export function topicMatches(pattern: string, key: string): boolean {
return matchFrom(pattern.split("."), 0, key.split("."), 0); return matchFrom(pattern.split("."), 0, key.split("."), 0);
} }