diff --git a/go/process/call.go b/go/process/call.go new file mode 100644 index 0000000..8b79160 --- /dev/null +++ b/go/process/call.go @@ -0,0 +1,84 @@ +package process + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" +) + +// CallTimeout bounds a call to a tool or a seat verb (mesh-tools node-tools/internal/bus/bus.go:35, +// RequestTimeout): what a module already expects. +var CallTimeout = 30 * time.Second + +// Call invokes a tool or a seat verb as this process's own account (novox/hq ADR 0436 clause 4), on the +// caller-named subject, falling back to the subject that names no caller when nothing answers it — +// exactly the provisioner client's own pattern (mesh-catalog modules/mesh-provisioner-client/cmd/ +// mesh-provisioner-client/secrets.go:59-61, 93-97, 119, 125-133). key is "seat:." for a seat's +// verb, or "." for a module's own tool, as a process's bus.invokes names it. The reply is +// {result, error, node} (mesh-tools node-tools/internal/bus/bus.go answerOf); a non-empty error is the +// call's failure. It offers no way to answer a call — a process reaches the bus only as itself, and +// never serves anything. +func (c *Conn) Call(key string, body any) (json.RawMessage, error) { + called, fallback, err := callSubjects(key, c.callerToken()) + if err != nil { + return nil, err + } + data, err := json.Marshal(body) + if err != nil { + return nil, err + } + result, err := c.request(called, data) + if errors.Is(err, nats.ErrNoResponders) { + result, err = c.request(fallback, data) + } + return result, err +} + +// callerToken is this account's user name as the last token of a call it publishes, each dot written ~ +// (mesh-tools node-tools/internal/bus/caller.go:43, CallerToken). +func (c *Conn) callerToken() string { return strings.ReplaceAll(c.account.User, ".", "~") } + +// callSubjects is the caller-named subject key is called on, and its plain fallback (mesh-tools +// node-tools/internal/bus/caller.go CalledSubject, toolShape, calledPattern). +func callSubjects(key, token string) (called, fallback string, err error) { + if rest, ok := strings.CutPrefix(key, "seat:"); ok { + seat, verb, ok := strings.Cut(rest, ".") + if !ok || seat == "" || verb == "" { + return "", "", fmt.Errorf("%q names no seat and verb (seat:.)", key) + } + return "mesh.seat." + seat + ".call." + verb + "." + token, "mesh.seat." + seat + ".tool." + verb, nil + } + module, tool, ok := strings.Cut(key, ".") + if !ok || module == "" || tool == "" { + return "", "", fmt.Errorf("%q is not ., and does not start seat:.", key) + } + return "mesh.mod." + module + ".call." + tool + "." + token, "mesh.mod." + module + ".tool." + tool, nil +} + +// request asks one subject once and reads its reply ({result, error, node}); a reply naming an error is +// the call's failure, not a transport one. +func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) { + ctx, cancel := context.WithTimeout(context.Background(), CallTimeout) + defer cancel() + msg, err := c.nc.RequestWithContext(ctx, subject, data) + if err != nil { + return nil, err + } + var r struct { + Result json.RawMessage `json:"result"` + Error string `json:"error"` + Node string `json:"node"` + } + if err := json.Unmarshal(msg.Data, &r); err != nil { + return nil, fmt.Errorf("%s answered something that is not a seat's answer: %w", subject, err) + } + if r.Error != "" { + return nil, errors.New(r.Error) + } + return r.Result, nil +} diff --git a/go/process/call_integration_test.go b/go/process/call_integration_test.go new file mode 100644 index 0000000..20f8292 --- /dev/null +++ b/go/process/call_integration_test.go @@ -0,0 +1,237 @@ +//go:build integration + +package process + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// 1191/AC1: given a stand-in holder answering seat operator-channel verb tell on a test bus, when the +// process connection of novox.mesh-mail.send calls seat:operator-channel.tell with {headline, lines, +// subject}, then the request goes to mesh.seat.operator-channel.call.tell.novox~mesh-mail~send with the +// arguments as the body, and the holder's {result} comes back; a reply with error set comes back as an +// error with those words. +func TestCallReachesTheCallerNamedSubjectAndReadsTheReply(t *testing.T) { + url, account := grantedSendProcess(t) + adminNC, _ := adminJS(t, url) + + heard := make(chan *nats.Msg, 1) + sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) { + heard <- m + _ = m.Respond([]byte(`{"result":{"sent":true}}`)) + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = sub.Unsubscribe() }() + if err := adminNC.Flush(); err != nil { + t.Fatal(err) + } + + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + result, err := conn.Call("seat:operator-channel.tell", + map[string]any{"headline": "h", "lines": []string{"l1"}, "subject": "s"}) + if err != nil { + t.Fatalf("Call: %v", err) + } + if string(result) != `{"sent":true}` { + t.Fatalf("result: %s", result) + } + + select { + case msg := <-heard: + var body map[string]any + if err := json.Unmarshal(msg.Data, &body); err != nil { + t.Fatal(err) + } + if body["headline"] != "h" || body["subject"] != "s" { + t.Errorf("the arguments were not the call's body: %v", body) + } + case <-time.After(3 * time.Second): + t.Fatal("the holder never heard the call") + } +} + +// 1191/AC1: a reply with error set comes back as an error with those words. +func TestCallReadsAnErrorReplyAsAnError(t *testing.T) { + url, account := grantedSendProcess(t) + adminNC, _ := adminJS(t, url) + sub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) { + _ = m.Respond([]byte(`{"error":"refused: no such warrant"}`)) + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = sub.Unsubscribe() }() + if err := adminNC.Flush(); err != nil { + t.Fatal(err) + } + + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + _, err = conn.Call("seat:operator-channel.tell", map[string]any{}) + if err == nil || err.Error() != "refused: no such warrant" { + t.Fatalf("Call with an error reply: %v, want the words of the error", err) + } +} + +// 1191/AC2: given no responder on the caller-named subject, when the call is made, then it is asked +// again on mesh.seat.operator-channel.tool.tell (the tool runner's fallback). +func TestCallFallsBackToThePlainSubjectWhenNothingAnswersTheCallerNamedOne(t *testing.T) { + url, account := grantedSendProcess(t) + adminNC, _ := adminJS(t, url) + sub, err := adminNC.Subscribe("mesh.seat.operator-channel.tool.tell", func(m *nats.Msg) { + _ = m.Respond([]byte(`{"result":{"via":"fallback"}}`)) + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = sub.Unsubscribe() }() + if err := adminNC.Flush(); err != nil { + t.Fatal(err) + } + + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + result, err := conn.Call("seat:operator-channel.tell", map[string]any{}) + if err != nil { + t.Fatalf("Call: %v", err) + } + if string(result) != `{"via":"fallback"}` { + t.Fatalf("result: %s, want the fallback holder's answer", result) + } +} + +// 1191/AC4, release 2's acceptance test: given the test bus with accounts composed as the controller +// composes a process account (the subjects of mesh-controller internal/broker/process_grants.go), when +// one process connection emits, updates a state, reads an own ask's record, consumes and acknowledges its +// queue and calls tell, then each succeeds; and its attempt to subscribe a tool-call subject is refused +// by the bus. +func TestReleaseAcceptanceAProcessDoesExactlyItsBusDeclaration(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil { + t.Fatalf("the admin could not make the state bucket: %v", err) + } + asksKV, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "messenger_asks"}) + if err != nil { + t.Fatalf("the admin could not make the records bucket: %v", err) + } + if _, err := asksKV.Put(ctxFor(t), "mesh-mail.ask-1", []byte(`{"state":"decided"}`)); err != nil { + t.Fatal(err) + } + if _, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{ + Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute, + }); err != nil { + t.Fatalf("the admin could not make the events stream: %v", err) + } + makeQueue(t, url, account) + publishApproved(t, url, `{"send":"s1"}`) + + adminNC, _ := adminJS(t, url) + holderSub, err := adminNC.Subscribe("mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", func(m *nats.Msg) { + _ = m.Respond([]byte(`{"result":{}}`)) + }) + if err != nil { + t.Fatal(err) + } + defer func() { _ = holderSub.Unsubscribe() }() + if err := adminNC.Flush(); err != nil { + t.Fatal(err) + } + + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + // 1. Emits. + if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "ev-1"}); err != nil { + t.Fatalf("Emit: %v", err) + } + + // 2. Updates a state. + s := conn.State("sends") + rev, err := s.Create("k", map[string]any{"v": 1}) + if err != nil { + t.Fatalf("State.Create: %v", err) + } + if _, err := s.Update("k", map[string]any{"v": 2}, rev); err != nil { + t.Fatalf("State.Update: %v", err) + } + + // 3. Reads an own ask's record. + if e, err := conn.Ask("messenger_asks", "ask-1"); err != nil || e == nil { + t.Fatalf("Ask: %+v, %v", e, err) + } + + // 4. Consumes and acknowledges its queue. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + consumed := make(chan error, 1) + go func() { + consumed <- conn.Consume(ctx, "MAIL_SENDS", func(m *Message) { + if err := m.Ack(); err != nil { + t.Errorf("Ack: %v", err) + } + cancel() + }) + }() + select { + case <-consumed: + case <-time.After(8 * time.Second): + t.Fatal("the process did not consume its own queue") + } + + // 5. Calls tell. + if _, err := conn.Call("seat:operator-channel.tell", map[string]any{}); err != nil { + t.Fatalf("Call: %v", err) + } + + // 6. Refused: subscribing a tool-call subject — a fresh connection, so the error handler catches + // exactly this one refusal and nothing earlier. + refusals := make(chan error, 1) + watched, err := Connect(account, nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err })) + if err != nil { + t.Fatal(err) + } + defer watched.Close() + if _, err := watched.nc.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil { + t.Fatal(err) + } + if err := watched.nc.Flush(); err != nil { + t.Fatal(err) + } + select { + case err := <-refusals: + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("subscribing a tool-call subject was not refused as a permissions violation: %v", err) + } + case <-time.After(3 * time.Second): + t.Error("subscribing a tool-call subject the process does not serve was not refused") + } +} diff --git a/go/process/call_test.go b/go/process/call_test.go new file mode 100644 index 0000000..eb50aa9 --- /dev/null +++ b/go/process/call_test.go @@ -0,0 +1,41 @@ +package process + +import "testing" + +// 1191/AC3: given a module tool key ., when called, then the subject is +// mesh.mod..call.. with the same fallback (mesh.mod..tool.). +func TestCallSubjectsForAModuleTool(t *testing.T) { + called, fallback, err := callSubjects("mesh-mail.send", "novox~mesh-mail~send") + if err != nil { + t.Fatal(err) + } + if called != "mesh.mod.mesh-mail.call.send.novox~mesh-mail~send" { + t.Errorf("called: %q", called) + } + if fallback != "mesh.mod.mesh-mail.tool.send" { + t.Errorf("fallback: %q", fallback) + } +} + +// 1191/AC1's subject (the seat form): seat:. calls mesh.seat..call.., +// falling back to mesh.seat..tool.. +func TestCallSubjectsForASeatVerb(t *testing.T) { + called, fallback, err := callSubjects("seat:operator-channel.tell", "novox~mesh-mail~send") + if err != nil { + t.Fatal(err) + } + if called != "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send" { + t.Errorf("called: %q", called) + } + if fallback != "mesh.seat.operator-channel.tool.tell" { + t.Errorf("fallback: %q", fallback) + } +} + +func TestCallSubjectsRefusesAKeyNamingNeither(t *testing.T) { + for _, key := range []string{"", "nodot", "seat:", "seat:noverb", "."} { + if _, _, err := callSubjects(key, "token"); err == nil { + t.Errorf("callSubjects(%q) was accepted", key) + } + } +}