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" } });