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

@@ -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")
+	}
+}

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 {