From ec80dcfc22aa6194ed45cf5ba8b267abea35a785 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 6 Oct 2026 18:14:58 +0200 Subject: [PATCH] 0.1.13: one rule for taking an event, and no handler left by a refused subscribe (hq issue 276) A handler takes an event by returning and asks for it again by throwing; any mesh/event answer but an error is taken. Written down in the README where event handling is, with what follows for a handler: throw only when the work was not done, and be safe to run twice. Both SDKs added the handler before asking the runtime and kept it when the runtime refused, so a module that asks again until it is taken (every Go consumer does) ran its handler once per attempt for every event. A refused subscribe now removes its handler; one the Go runtime did not answer in time is kept, since the runtime may have bound it. --- README.md | 30 ++++++++++ go/README.md | 2 +- go/stdio.go | 43 +++++++++++++-- go/stdio_test.go | 85 +++++++++++++++++++++++++++++ package-lock.json | 4 +- package.json | 2 +- src/events/index.ts | 4 ++ src/stdio/index.ts | 12 +++- test/fixtures/stdio-resubscribe.mjs | 18 ++++++ test/stdio.test.ts | 34 ++++++++++++ 10 files changed, 223 insertions(+), 11 deletions(-) create mode 100644 test/fixtures/stdio-resubscribe.mjs diff --git a/README.md b/README.md index 13420bd..01d1d3e 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,36 @@ they call, live in the module. The broker client (connect, request/reply, publish/subscribe), the event-consumer a module uses to react to mesh events, and the envelope/routing-key conventions. Transport primitives. +#### Taking an event — the one rule + +A module's handler (`on(pattern, handler)` here, `stdio.Subscribe` in Go, the same in every language) +**takes an event by returning, and asks for it again by throwing** (Go: returning an error). Nothing +else counts: what a handler returns is not read. + +On the wire, the runtime hands the bundle `mesh/event` with the envelope and acknowledges the event on +the bus when the answer is **anything but a JSON-RPC error** — `{}` (what the SDKs send), `null`, or no +`result` at all. An `error` answer, no answer within the runtime's event timeout (two minutes), or the +bundle exiting leaves it unacknowledged: it is offered again after five seconds, up to the module +consumer's maximum deliveries, after which the bus gives it up and the controller raises a condition +(`max-deliveries`). The runtime's line for an event not taken quotes a handler's error as the +handler's own words, so it is never mistaken for the runtime's. + +So, for a handler: + +- **Throw only when the work was not done** — the service refused, was unreachable, the store failed. + Never for something that went right but answered oddly: an API that answers an empty body is a + success, and reading it as JSON is a throw that has every event offered five times, its work done + five times, and a false alarm raised (novox/hq issue 276). +- **Delivery is at-least-once**, so the work must be safe to do twice, or the handler skips an event + whose `x-event-id` (`event.id`) it already did — remembering it only once the work succeeded, so one + that failed is done when it is offered again. +- An event no handler matches is answered at once and taken. +- A subscribe the runtime refused leaves no handler behind, so asking again until it is taken hands + each event to the handler once. + +Checked by the SDKs' stdio tests (TypeScript `test/stdio.test.ts`, Go `go/stdio_test.go`) and the +runtime's (`node-tools/internal/launch`, mesh-tools). + ### `state` — a module's current state on the bus `state(name)` reaches a state the module declared it keeps (`state` in its manifest) or reads (`reads`, as `.`): `get`, `put`, `delete`, `keys`, and `watch`, which hands the current diff --git a/go/README.md b/go/README.md index 1d1ae78..4c50d16 100644 --- a/go/README.md +++ b/go/README.md @@ -3,7 +3,7 @@ `stdio.Serve("", tools)` speaks MCP over stdio to the node's tool runtime, as the module the runtime names (novox/hq ADR 0188, ADR 0193). A `Tool` is a name, a description, an argument schema and a function; calls run concurrently. A tool named `.` is the seat's implementation. -`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (acknowledged only after the handler returns nil) and `stdio.Ask(key, body)` calls a tool, and `stdio.State(name)` gets, puts, deletes, lists and watches the module's declared state — all through the runtime (novox/hq ADR 0193, 0198, 0201). Standard library only. +`stdio.Emit(type, body)` publishes an event as the module, `stdio.Subscribe(pattern, handler)` receives the module's events (taken when the handler returns nil, offered again when it returns an error — the one rule is in the repository README, "Taking an event") and `stdio.Ask(key, body)` calls a tool, and `stdio.State(name)` gets, puts, deletes, lists and watches the module's declared state — all through the runtime (novox/hq ADR 0193, 0198, 0201). Standard library only. ## Installing diff --git a/go/stdio.go b/go/stdio.go index 136309e..310d76a 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -60,11 +60,14 @@ type channel struct { next int waiting map[string]chan message listeners []listener + // subscribed numbers the listeners, so one whose subscribe failed is the one removed. + subscribed int // watches are this bundle's state watches by the id each was asked with (ADR 0201). watches map[string]func(StateChange) error } type listener struct { + n int pattern string handle func(Envelope) error } @@ -108,7 +111,7 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j select { case m := <-answer: if m.Error != nil { - return nil, errors.New(m.Error.Message) + return nil, &refused{m.Error.Message} } return m.Result, nil case <-time.After(timeout): @@ -119,6 +122,11 @@ func (c *channel) askAs(id, method string, params any, timeout time.Duration) (j } } +// refused is the runtime's error answer, as opposed to no answer: what it refused, it did not do. +type refused struct{ message string } + +func (r *refused) Error() string { return r.message } + var ( current *channel currentMu sync.Mutex @@ -287,18 +295,43 @@ func live() (*channel, error) { } // Subscribe asks the runtime for this module's events whose type matches the pattern (`*` one -// segment, `#` any), and hands each to handle (ADR 0198). The runtime acknowledges an event on the bus -// only after every matching handler returned nil; an error has it delivered again. Call it from a -// goroutine started before Serve, or from a tool, once Serve runs. +// segment, `#` any), and hands each to handle (ADR 0198). +// +// **Taking an event** (see the repository's README): handle returning nil takes the event — the +// runtime acknowledges it once every matching handler did; an error is the one way to have it offered +// again, after a few seconds and a bounded number of times, after which the bus gives up and the +// controller raises a condition. So return an error only when the work was not done, never for an +// answer that went right but read oddly; and since delivery is at-least-once, make the work safe to +// do twice, or skip an event whose x-event-id was already done. +// +// A subscribe the runtime refused leaves no handler behind, so asking again — as a module does until +// the runtime takes it — does not hand one event to the same handler twice. (One the runtime did not +// answer in time is kept, as the runtime may have bound it.) Call it from a goroutine +// started before Serve, or from a tool, once Serve runs. func Subscribe(pattern string, handle func(Envelope) error) error { c, err := live() if err != nil { return err } c.mu.Lock() - c.listeners = append(c.listeners, listener{pattern: pattern, handle: handle}) + c.subscribed++ + n := c.subscribed + c.listeners = append(c.listeners, listener{n: n, pattern: pattern, handle: handle}) c.mu.Unlock() _, err = c.ask("mesh/subscribe", map[string]any{"pattern": pattern}, 30*time.Second) + // Removed only when the runtime answered that it refused: one that did not answer in time may + // have bound the consumer, and a handler removed then would have its events taken unhandled. + var no *refused + if errors.As(err, &no) { + c.mu.Lock() + for i, l := range c.listeners { + if l.n == n { + c.listeners = append(c.listeners[:i:i], c.listeners[i+1:]...) + break + } + } + c.mu.Unlock() + } return err } diff --git a/go/stdio_test.go b/go/stdio_test.go index e05a16a..f07b6e3 100644 --- a/go/stdio_test.go +++ b/go/stdio_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "io" + "sync" "testing" "time" ) @@ -139,3 +140,87 @@ func TestABundleSubscribesAndAsksThroughTheRuntime(t *testing.T) { t.Error("topic matching differs from the TypeScript SDK's") } } + +// A subscribe the runtime refused leaves no handler behind: asked again until the runtime takes it, a +// module's handler is handed each event once, not once per attempt (novox/hq issue 276). +func TestARefusedSubscribeLeavesNoHandlerBehind(t *testing.T) { + t.Setenv(ServedAs, "plex") + inR, inW := io.Pipe() + outR, outW := io.Pipe() + var mu sync.Mutex + handled := 0 + subscribed := make(chan error, 2) + go func() { + _ = ServeOn("", []Tool{{Name: "start", Run: func(map[string]any) (any, error) { + go func() { + for i := 0; i < 2; i++ { + err := Subscribe("*.download.completed", func(Envelope) error { + mu.Lock() + handled++ + mu.Unlock() + return nil + }) + subscribed <- err + if err == nil { + return + } + } + }() + return "started", nil + }}}, inR, outW) + }() + lines := bufio.NewScanner(outR) + next := func() map[string]any { + done := make(chan map[string]any, 1) + go func() { + if lines.Scan() { + var m map[string]any + _ = json.Unmarshal(lines.Bytes(), &m) + done <- m + } + }() + select { + case m := <-done: + return m + case <-time.After(3 * time.Second): + t.Fatal("the bundle said nothing") + return nil + } + } + send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) } + send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "start"}}) + answered := 0 + for answered < 2 { + m := next() + switch m["method"] { + case "mesh/subscribe": + if answered == 0 { + send(map[string]any{"jsonrpc": "2.0", "id": m["id"], "error": map[string]any{"code": -32000, "message": "not yet"}}) + } else { + send(map[string]any{"jsonrpc": "2.0", "id": m["id"], "result": map[string]any{}}) + } + answered++ + } + } + if err := <-subscribed; err == nil || err.Error() != "not yet" { + t.Fatalf("the refused subscribe said %v", err) + } + if err := <-subscribed; err != nil { + t.Fatalf("the second subscribe said %v", err) + } + send(map[string]any{"jsonrpc": "2.0", "id": 9, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{"key": "radarr.download.completed", "node": "anchor", "body": map[string]any{}}}}) + for { + if r := next(); r["id"] == float64(9) { + if r["error"] != nil { + t.Fatalf("the event was answered %v", r) + } + break + } + } + _ = inW.Close() + mu.Lock() + defer mu.Unlock() + if handled != 1 { + t.Fatalf("one event was handled %d times", handled) + } +} diff --git a/package-lock.json b/package-lock.json index 1037bb4..a109d0a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@novox/mesh-sdk", - "version": "0.1.12", + "version": "0.1.13", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@novox/mesh-sdk", - "version": "0.1.12", + "version": "0.1.13", "devDependencies": { "@types/node": "^22.0.0", "typescript": "^5.6.0" diff --git a/package.json b/package.json index a3ec789..43be6bf 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@novox/mesh-sdk", - "version": "0.1.12", + "version": "0.1.13", "description": "The stable spine a Novox Mesh module's own code builds against.", "type": "module", "exports": { diff --git a/src/events/index.ts b/src/events/index.ts index 42e902a..7360688 100644 --- a/src/events/index.ts +++ b/src/events/index.ts @@ -65,6 +65,10 @@ export async function emit(type: string, body: T, opts: EmitOptions = {}): Pr * React to events whose type matches a topic pattern (`*` one segment, `#` any). The audit logger * is just `on("#", …)`. The handler receives the reconstructed event — its metadata read back from * the headers, its body the domain payload. + * + * **The handler takes the event by returning and asks for it again by throwing** — nothing else + * counts (README, "Taking an event"). Throw only when the work was not done; delivery is + * at-least-once, so make the work safe to repeat or skip an `id` already done. */ export async function on(pattern: string, handler: (event: Event) => Promise): Promise<() => void> { return broker().subscribe(pattern, async (envelope) => { diff --git a/src/stdio/index.ts b/src/stdio/index.ts index e99ed2f..0e5c600 100644 --- a/src/stdio/index.ts +++ b/src/stdio/index.ts @@ -106,11 +106,19 @@ function startSession(): Session { subscribe: (async (pattern: string, handler: (env: Envelope) => Promise) => { const listener = { pattern, handler: handler as (env: Envelope) => Promise }; listeners.push(listener); - await ask("mesh/subscribe", { pattern }); - return () => { + const remove = (): void => { const at = listeners.indexOf(listener); if (at >= 0) listeners.splice(at, 1); }; + try { + await ask("mesh/subscribe", { pattern }); + } catch (err) { + // Refused: nothing is bound, and a handler left behind would be handed each event twice once + // the module asks again (novox/hq issue 276). + remove(); + throw err; + } + return remove; }) as Broker["subscribe"], close: async () => {}, }; diff --git a/test/fixtures/stdio-resubscribe.mjs b/test/fixtures/stdio-resubscribe.mjs new file mode 100644 index 0000000..f642bc5 --- /dev/null +++ b/test/fixtures/stdio-resubscribe.mjs @@ -0,0 +1,18 @@ +// A launched bundle that asks for its events until the runtime takes the subscription — as a module +// whose first ask may come too early does — and counts how often its one handler is handed an event. +import { on } from "../../dist/events/index.js"; +import { serveRegisteredOverStdio } from "../../dist/stdio/index.js"; + +let handled = 0; +for (;;) { + try { + await on("radarr.download.completed", async () => { + handled++; + process.stderr.write(`handled ${handled}\n`); + }); + break; + } catch (err) { + process.stderr.write(`refused: ${err.message}\n`); + } +} +await serveRegisteredOverStdio(); diff --git a/test/stdio.test.ts b/test/stdio.test.ts index 885bd26..df6f0e4 100644 --- a/test/stdio.test.ts +++ b/test/stdio.test.ts @@ -160,6 +160,40 @@ test("a launched bundle subscribes as it is imported, handles events the runtime } }); +test("a subscribe the runtime refused leaves no handler behind, so one event is handled once (novox/hq issue 276)", async () => { + const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-resubscribe.mjs", import.meta.url))], + { stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, MESH_SERVED_MODULE: "plex" } }); + const seen: any[] = []; + let buffered = ""; + let said = ""; + child.stderr.on("data", (d) => (said += d.toString())); + child.stdout.on("data", (d) => { + buffered += d.toString(); + let at: number; + while ((at = buffered.indexOf("\n")) >= 0) { + seen.push(JSON.parse(buffered.slice(0, at))); + buffered = buffered.slice(at + 1); + } + }); + const until = async (match: (m: any) => boolean) => { + for (let i = 0; i < 150 && !seen.some(match); i++) await new Promise((r) => setTimeout(r, 20)); + return seen.find(match); + }; + const send = (m: unknown) => child.stdin.write(JSON.stringify(m) + "\n"); + try { + const first = await until((m) => m.method === "mesh/subscribe"); + send({ jsonrpc: "2.0", id: first.id, error: { code: -32000, message: "not yet" } }); + const second = await until((m) => m.method === "mesh/subscribe" && m.id !== first.id); + send({ jsonrpc: "2.0", id: second.id, result: {} }); + send({ jsonrpc: "2.0", id: 20, method: "mesh/event", params: { envelope: { key: "radarr.download.completed", node: "anchor", body: {}, headers: { "x-event-id": "e20" } } } }); + assert.deepEqual((await until((m) => m.id === 20)).result, {}); + assert.match(said, /refused: not yet/); + assert.equal(said.match(/handled/g)?.length, 1, `one event was handled more than once:\n${said}`); + } finally { + child.kill(); + } +}); + test("a launched bundle watches its state as it is imported, has the current values before it goes on, and keeps state through the runtime (ADR 0201)", async () => { const child = spawn(process.execPath, [fileURLToPath(new URL("./fixtures/stdio-state.mjs", import.meta.url))], { stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, MESH_SERVED_MODULE: "keeper" } }); -- 2.54.0