From 0796341c8db6792068705456ec471582aa3c4439 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 04:26:54 +0200 Subject: [PATCH] Let an SDK Emit choose its event's message id, so a repeated publish is kept to one (issue 1187) --- go/emit_message_id_test.go | 110 +++++++++++++++++++++++++++++++++++++ go/stdio.go | 18 +++++- 2 files changed, 126 insertions(+), 2 deletions(-) create mode 100644 go/emit_message_id_test.go diff --git a/go/emit_message_id_test.go b/go/emit_message_id_test.go new file mode 100644 index 0000000..45add73 --- /dev/null +++ b/go/emit_message_id_test.go @@ -0,0 +1,110 @@ +package stdio + +import ( + "bufio" + "encoding/json" + "io" + "testing" + "time" +) + +// emitRuntime drives a bundle over pipes as the runtime does, reading every line it says. +type emitRuntime struct { + t *testing.T + lines *bufio.Scanner + in io.Writer +} + +func (r *emitRuntime) next() map[string]any { + done := make(chan map[string]any, 1) + go func() { + if r.lines.Scan() { + var m map[string]any + _ = json.Unmarshal(r.lines.Bytes(), &m) + done <- m + } + }() + select { + case m := <-done: + return m + case <-time.After(3 * time.Second): + r.t.Fatal("the bundle said nothing") + return nil + } +} + +func serveEmit(t *testing.T, tools []Tool) *emitRuntime { + t.Helper() + t.Setenv(ServedAs, "chime") + inR, inW := io.Pipe() + outR, outW := io.Pipe() + go func() { _ = ServeOn("", tools, inR, outW) }() + t.Cleanup(func() { _ = inW.Close() }) + return &emitRuntime{t: t, lines: bufio.NewScanner(outR), in: inW} +} + +func (r *emitRuntime) call(id int, tool string) { + b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": id, "method": "tools/call", + "params": map[string]any{"name": tool, "arguments": map[string]any{}}}) + _, _ = r.in.Write(append(b, '\n')) +} + +func (r *emitRuntime) answer(id any) { + b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": id, "result": map[string]any{}}) + _, _ = r.in.Write(append(b, '\n')) +} + +// eventIDOf is a "mesh/publish" ask's x-event-id header. +func eventIDOf(asked map[string]any) string { + return asked["params"].(map[string]any)["headers"].(map[string]any)["x-event-id"].(string) +} + +// 1187/AC1: given EmitOptions with a message id, when a bundle emits, then the event's x-event-id header is +// that id; and a random one when none is given, as on main. +func TestEmitSendsTheChosenMessageIDAsXEventID(t *testing.T) { + rt := serveEmit(t, []Tool{ + {Name: "chosen", Run: func(map[string]any) (any, error) { + return nil, Emit("sent", map[string]any{}, EmitOptions{MessageID: "send-42"}) + }}, + {Name: "unchosen", Run: func(map[string]any) (any, error) { + return nil, Emit("sent", map[string]any{}) + }}, + }) + rt.call(1, "chosen") + asked := rt.next() + if got := eventIDOf(asked); got != "send-42" { + t.Fatalf("x-event-id: %q, want the chosen id %q", got, "send-42") + } + rt.answer(asked["id"]) + rt.next() // the tool's own answer + + rt.call(2, "unchosen") + asked = rt.next() + if got := eventIDOf(asked); len(got) != 36 { + t.Fatalf("x-event-id with no message id chosen: %q, want a random UUID as on main", got) + } + rt.answer(asked["id"]) + rt.next() +} + +// 1187/AC2: given two emits with the same chosen id, when sent through the runtime's protocol stand-in, +// then both carry the same x-event-id, which the tool runner gives the bus as Nats-Msg-Id (mesh-tools +// node-tools/internal/bus/bus.go ~748-773, :770): a repeated publish is kept to one by the bus's dedup. +func TestEmitRepeatedWithTheSameChosenIDCarriesTheSameXEventIDTwice(t *testing.T) { + rt := serveEmit(t, []Tool{ + {Name: "chosen", Run: func(map[string]any) (any, error) { + return nil, Emit("sent", map[string]any{}, EmitOptions{MessageID: "send-77"}) + }}, + }) + var ids []string + for i := 1; i <= 2; i++ { + rt.call(i, "chosen") + asked := rt.next() + ids = append(ids, eventIDOf(asked)) + rt.answer(asked["id"]) + rt.next() + } + if ids[0] != "send-77" || ids[1] != "send-77" { + t.Fatalf("the two emits' x-event-id: %v, want %q both times", ids, "send-77") + } +} diff --git a/go/stdio.go b/go/stdio.go index 03db9bd..3960fba 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -472,10 +472,15 @@ func matchFrom(p []string, pi int, k []string, ki int) bool { return matchFrom(p, pi+1, k, ki+1) } -// EmitOptions is the provenance an emitter may attach (ADR 0042). +// EmitOptions is the provenance an emitter may attach (ADR 0042), and the message id it may choose. type EmitOptions struct { CausationID string Schema string + // MessageID is the id this emit is sent under, as x-event-id (and, by the tool runner, as the bus's + // own Nats-Msg-Id): chosen by the emitter so a repeated publish inside the stream's duplicate window + // is kept to one, never generated once an id is given (novox/hq ADR 0436 clause 4). Empty is a random + // one, as on main. + MessageID string } // Emit asks the runtime to publish an event as this bundle's module, and returns once the bus has @@ -495,8 +500,17 @@ func Emit(eventType string, body any, opts ...EmitOptions) error { if node == "" { node = "unknown" } + id := "" + for _, o := range opts { + if o.MessageID != "" { + id = o.MessageID + } + } + if id == "" { + id = eventID() + } headers := map[string]string{ - "x-event-id": eventID(), "x-source": source, "x-node": node, + "x-event-id": id, "x-source": source, "x-node": node, "x-time": time.Now().UTC().Format(time.RFC3339Nano), "content-type": "application/json", } for _, o := range opts {