A runtime serves what the mesh issued it, and a seat's verbs are implemented under the seat's name (hq ADR 0160)
The one address a runtime derives for itself is mesh.assignment.<node>.<module>. It reads the
membership there with a direct get on the ASSIGNMENTS stream, serves each tool exactly where the
membership says — the plain subject in the module's queue when the mesh issued one, this machine's
beside it — and follows the subject, re-serving when a new membership arrives. A mesh that has issued
nothing yet gets the shape it always derived, and the log says so.
A seat's verbs are the role's, not the software's (ADR 0159): a module implements them with
registerModuleTools("<seat>", …), the runtime serves that on the seat's subjects when the credential
claims the seat, and never lists it among the module's own tools. A module named like its seat
registers once and is both.
The tools answer carries each tool's subjects, and the console and CLI call the subject the listing
gave them instead of composing one.
This commit is contained in:
+130
-5
@@ -46,6 +46,24 @@ export interface Credential {
|
||||
claims?: { seat: string; scope?: string; serves?: string[] }[];
|
||||
}
|
||||
|
||||
/**
|
||||
* What the mesh issued this assignment (novox/hq ADR 0160): where its tools are served, in which
|
||||
* queue, the verbs of the seats it holds, where its events land, what it may reach. Read from the
|
||||
* ASSIGNMENTS stream at `mesh.assignment.<node>.<module>` — the one subject a runtime derives for
|
||||
* itself — and followed live. Absent for a mesh older than the membership, and then the runtime
|
||||
* serves the shape it always derived, and says so.
|
||||
*/
|
||||
export interface Membership {
|
||||
node: string;
|
||||
module: string;
|
||||
/** Addresses a tool is answered on; `{tool}` stands for the tool's name. */
|
||||
serves: { subject: string; queue?: string }[];
|
||||
seats?: { seat: string; verb: string; subject: string }[];
|
||||
emits: string;
|
||||
reaches?: Record<string, string[]>;
|
||||
tools: string;
|
||||
}
|
||||
|
||||
/** What a tool call answers: the module's own result, and which machine answered it
|
||||
* (novox/hq ADR 0159) — a module on several machines is otherwise an answer from nowhere. */
|
||||
export interface Answered<Res> {
|
||||
@@ -57,8 +75,20 @@ export interface Answered<Res> {
|
||||
* an answer that says which machine gave it, and serving a subject that is not a module's own tool
|
||||
* (a seat's verb). */
|
||||
export interface RuntimeBroker extends Broker {
|
||||
ask<Req, Res>(key: string, body: Req): Promise<Answered<Res>>;
|
||||
/** Call a tool by key, or — when `on` names a subject the mesh listed for it (ADR 0160) — there. */
|
||||
ask<Req, Res>(key: string, body: Req, on?: string): Promise<Answered<Res>>;
|
||||
handleSubject<Req, Res>(subject: string, handler: (body: Req) => Promise<Res>): Promise<() => void>;
|
||||
/** What the mesh issued this assignment, or undefined when nothing has been issued yet. */
|
||||
membership(): Membership | undefined;
|
||||
/** Called when the mesh issues a new membership; the runtime re-serves on it. */
|
||||
onMembership(handler: (m: Membership) => void): void;
|
||||
}
|
||||
|
||||
const ASSIGNMENTS_STREAM = "ASSIGNMENTS";
|
||||
|
||||
/** The one address a runtime derives for itself (ADR 0160). */
|
||||
export function membershipSubject(node: string, module: string): string {
|
||||
return `mesh.assignment.${node}.${module}`;
|
||||
}
|
||||
|
||||
/** Whether a connection failure is worth retrying, or is a fact about this configuration that
|
||||
@@ -123,6 +153,80 @@ export async function connectNats(
|
||||
|
||||
const node = cred.node;
|
||||
|
||||
// The membership, read once at connect and followed. A direct get is one request on the
|
||||
// stream's API, which is the whole of what this account may ask JetStream for its own subject;
|
||||
// a 404 is a mesh that has not issued one, which is a fact to say and not an error to retry.
|
||||
let issued: Membership | undefined;
|
||||
const issuedHandlers: ((m: Membership) => void)[] = [];
|
||||
const subjectOfMine = node ? membershipSubject(node, self) : "";
|
||||
if (subjectOfMine) {
|
||||
try {
|
||||
const got = await conn.request(`$JS.API.DIRECT.GET.${ASSIGNMENTS_STREAM}`,
|
||||
sc.encode(JSON.stringify({ last_by_subj: subjectOfMine })), { timeout: 5_000 });
|
||||
const status = got.headers?.code ?? 0;
|
||||
if (status === 0 && got.data.length > 0) {
|
||||
issued = JSON.parse(sc.decode(got.data)) as Membership;
|
||||
}
|
||||
} catch {
|
||||
// Not readable here: an older mesh, a stream not yet asserted, or no grant. Said below.
|
||||
}
|
||||
if (!issued) {
|
||||
console.log(`[mesh-tools] no membership issued for ${self} on ${node} yet; serving the derived shape until one arrives`);
|
||||
}
|
||||
try {
|
||||
const live = conn.subscribe(subjectOfMine);
|
||||
subs.push(live);
|
||||
void (async () => {
|
||||
for await (const msg of live) {
|
||||
try {
|
||||
issued = JSON.parse(sc.decode(msg.data)) as Membership;
|
||||
console.log(`[mesh-tools] ${self} on ${node} was issued a new membership; re-serving on it`);
|
||||
for (const h of issuedHandlers) h(issued);
|
||||
} catch (err) {
|
||||
console.log(`[mesh-tools] a membership arrived that is not one: ${err}`);
|
||||
}
|
||||
}
|
||||
})();
|
||||
} catch {
|
||||
// A subscription this account may not make is a mesh older than the membership.
|
||||
}
|
||||
}
|
||||
|
||||
/** The subjects a tool of this module is served on: from the membership when issued, derived
|
||||
* otherwise (the shape the mesh issues on day one, so the two agree). */
|
||||
const servedOn = (tool: string): { subject: string; queue?: string }[] => {
|
||||
if (issued) {
|
||||
const m = issued;
|
||||
const out = m.serves.map((s) => ({ subject: s.subject.replace("{tool}", tool), queue: s.queue }));
|
||||
// The verb that lists what this module serves is answered on the mesh's plain address for it
|
||||
// whatever the placement — one answer suffices, so a queue — and on this machine's beside it.
|
||||
if (tool === "tools" && m.tools && !out.some((s) => s.subject === m.tools)) {
|
||||
out.unshift({ subject: m.tools, queue: `serve.${self}` });
|
||||
}
|
||||
return out;
|
||||
}
|
||||
const base = `mesh.mod.${self}.tool.${tool}`;
|
||||
const out: { subject: string; queue?: string }[] = [{ subject: base, queue: `serve.${self}` }];
|
||||
if (node) out.push({ subject: `${base}.${node}` });
|
||||
return out;
|
||||
};
|
||||
|
||||
/** Where a call by key goes: a subject the membership says this module reaches, when it says
|
||||
* one — the machine's when named — else the derived shape. */
|
||||
const reachedAt = (key: string): string => {
|
||||
const [name, wanted] = key.split("@", 2);
|
||||
const reach = issued?.reaches?.[name];
|
||||
if (reach && reach.length > 0) {
|
||||
if (wanted) {
|
||||
const at = reach.find((s) => s.endsWith(`.${wanted}`));
|
||||
if (at) return at;
|
||||
} else {
|
||||
return reach[0];
|
||||
}
|
||||
}
|
||||
return toolSubject(key, self);
|
||||
};
|
||||
|
||||
/** Answer one subject with one handler, and say which machine answered (novox/hq ADR 0159). */
|
||||
const answerOn = <Req, Res>(
|
||||
subject: string,
|
||||
@@ -156,8 +260,8 @@ export async function connectNats(
|
||||
* request carries, which the responder may answer because its account has `allow_responses`
|
||||
* — one reply to a message it actually received, and nothing wider.
|
||||
*/
|
||||
const ask = async <Req, Res>(key: string, body: Req): Promise<Answered<Res>> => {
|
||||
const msg = await conn.request(toolSubject(key, self), sc.encode(JSON.stringify(body)), {
|
||||
const ask = async <Req, Res>(key: string, body: Req, on?: string): Promise<Answered<Res>> => {
|
||||
const msg = await conn.request(on ?? reachedAt(key), sc.encode(JSON.stringify(body)), {
|
||||
timeout: REQUEST_TIMEOUT_MS,
|
||||
});
|
||||
const reply = JSON.parse(sc.decode(msg.data)) as { result?: Res; error?: string; node?: string };
|
||||
@@ -180,11 +284,32 @@ export async function connectNats(
|
||||
* which is how it always behaved.
|
||||
*/
|
||||
async handle<Req, Res>(key: string, handler: (body: Req) => Promise<Res>): Promise<() => void> {
|
||||
const stops = [answerOn(toolSubject(key, self), `serve.${self}`, handler)];
|
||||
if (node) stops.push(answerOn(`${toolSubject(key, self)}.${node}`, undefined, handler));
|
||||
// A seat's verb named outright is served on the seat's subject as given, for a holder that
|
||||
// knows its role without a membership; everything else is this module's own tool, served
|
||||
// where the mesh issued it (ADR 0160). A key naming another module is not served here at all.
|
||||
if (key.startsWith("seat:")) {
|
||||
const stop = answerOn(toolSubject(key, self), undefined, handler);
|
||||
return () => stop();
|
||||
}
|
||||
const dot = key.indexOf(".");
|
||||
if (dot >= 0 && key.slice(0, dot) !== self) {
|
||||
throw new Error(`${self} cannot serve ${key}: a module serves its own tools`);
|
||||
}
|
||||
const tool = dot < 0 ? key : key.slice(dot + 1);
|
||||
let stops = servedOn(tool).map((s) => answerOn(s.subject, s.queue, handler));
|
||||
// When a new membership arrives, serve where it now says and stop serving where it no longer does.
|
||||
issuedHandlers.push(() => {
|
||||
stops.forEach((stop) => stop());
|
||||
stops = servedOn(tool).map((s) => answerOn(s.subject, s.queue, handler));
|
||||
});
|
||||
return () => stops.forEach((stop) => stop());
|
||||
},
|
||||
|
||||
membership: () => issued,
|
||||
onMembership: (handler: (m: Membership) => void) => {
|
||||
issuedHandlers.push(handler);
|
||||
},
|
||||
|
||||
async handleSubject<Req, Res>(subject: string, handler: (body: Req) => Promise<Res>): Promise<() => void> {
|
||||
return answerOn(subject, undefined, handler);
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user