diff --git a/merge-check.sh b/merge-check.sh index 1bd567e..8b3100c 100644 --- a/merge-check.sh +++ b/merge-check.sh @@ -36,7 +36,7 @@ if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; t # And what of the runtime launches no bundle: how a bundle's seat traffic reaches the bus, what its # environment holds, what a runtime on a module's own account serves (novox/hq ADR 0259), and that a write # by compare-and-set loses no update under contention (novox/hq research 042). - runtimeless='^(TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing)$' + runtimeless='^(TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing|TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller|TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone)$' fi if command -v gcc >/dev/null 2>&1; then CGO_ENABLED=1 go test -race -count=1 $packages diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index ddf22ce..896b37e 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -5,6 +5,8 @@ // mesh.mod..event. an event a module emits // mesh.mod..tool. a tool a module serves // mesh.seat..tool. a role's verb, answered by whoever holds the seat +// mesh.mod..call.., mesh.seat..call.. +// the same, naming the bus user that called (caller.go, hq issue 365) package bus import ( @@ -170,6 +172,8 @@ type Conn struct { // RetryAfter is how long after a refusal a subscription is asked for again, by attempt; past the // last, it is given up and said. A variable so a test need not wait minutes. RetryAfter []time.Duration + // calls are the caller-named calls in flight (caller.go, novox/hq issue 365). + calls callers } // answered is one subject the runtime answers on: what to subscribe again with, and how often it was @@ -204,6 +208,9 @@ func (c *Conn) refused(err error) { if !errors.Is(err, nats.ErrPermissionViolation) { return } + if c.publishRefused(err) { + return + } m := refusedSubject.FindStringSubmatch(err.Error()) if len(m) < 2 { return @@ -441,10 +448,17 @@ type reply struct { // answerOn answers one subject with one handler, and says which machine answered (ADR 0159). func (c *Conn) answerOn(subject, queue string, h Handler) (func(), error) { + return c.answerCalledOn(subject, queue, func(_ string, body json.RawMessage) (any, error) { return h(body) }) +} + +// answerCalledOn answers one subject, handing the handler the caller the subject a call arrived on names: +// under a pattern calledPattern made, its last token; on any other subject, "" — none was proven (novox/hq +// issue 365). Never from the body: a body is the caller's to write. +func (c *Conn) answerCalledOn(subject, queue string, h CalledHandler) (func(), error) { cb := func(msg *nats.Msg) { go func() { var r reply - result, err := h(json.RawMessage(msg.Data)) + result, err := h(callerOn(subject, msg.Subject), json.RawMessage(msg.Data)) if err != nil { r.Error = err.Error() } else { @@ -497,18 +511,50 @@ func nullable(v any) any { // HandleSubject answers a subject outright: a seat's verb where the mesh issued it. func (c *Conn) HandleSubject(subject string, h Handler) (func(), error) { - return c.answerOn(subject, "", h) + return c.HandleSubjectCalled(subject, func(_ string, body json.RawMessage) (any, error) { return h(body) }) +} + +// HandleSubjectCalled answers a tool subject and its caller-named counterpart (novox/hq issue 365): a call on +// the counterpart is handed the caller its subject names, one on the subject itself "" — none proven. The +// subject that names no caller is served beside the other for one release, so callers move without a gap +// (its retirement: hq issue 464). +func (c *Conn) HandleSubjectCalled(subject string, h CalledHandler) (func(), error) { + return c.answerBoth(subject, "", h) +} + +// answerBoth answers a subject and, when it is a tool's, the caller-named pattern beside it, in one queue. +func (c *Conn) answerBoth(subject, queue string, h CalledHandler) (func(), error) { + stop, err := c.answerCalledOn(subject, queue, h) + if err != nil { + return func() {}, err + } + pattern := calledPattern(subject) + if pattern == "" { + return stop, nil + } + stopCalled, err := c.answerCalledOn(pattern, queue, h) + if err != nil { + c.Logf("[mesh-tools] cannot serve the caller-named %s: %v; its callers are asked on %s alone", pattern, err, subject) + return stop, nil + } + return func() { stop(); stopCalled() }, nil } // Handle serves `.` where the mesh issued that module, and follows its membership: // when a new one arrives, it serves where it now says and stops where it no longer does. func (c *Conn) Handle(key string, h Handler) (func(), error) { + return c.HandleCalled(key, func(_ string, body json.RawMessage) (any, error) { return h(body) }) +} + +// HandleCalled is Handle telling the handler who called: the bus user a caller-named subject names, or "" for +// a call on the subject that names none (novox/hq issue 365). +func (c *Conn) HandleCalled(key string, h CalledHandler) (func(), error) { if strings.HasPrefix(key, "seat:") { subject, err := ToolSubject(key, c.self) if err != nil { return func() {}, err } - return c.answerOn(subject, "", h) + return c.answerBoth(subject, "", h) } module, tool := c.self, key if dot := strings.Index(key, "."); dot >= 0 { @@ -528,7 +574,7 @@ func (c *Conn) Handle(key string, h Handler) (func(), error) { } stops = nil for _, s := range c.servedOn(module, tool) { - if stop, err := c.answerOn(s.Subject, s.Queue, h); err == nil { + if stop, err := c.answerBoth(s.Subject, s.Queue, h); err == nil { stops = append(stops, stop) } else { c.Logf("[mesh-tools] cannot serve %s on %s: %v", key, s.Subject, err) @@ -585,6 +631,23 @@ func (c *Conn) Ask(key string, body any, on string) (Answered, error) { if err != nil { return Answered{}, err } + // **Named by its caller first** (novox/hq issue 365): on the caller-named subject, which the bus lets this + // account publish with its own name alone; asked again on the subject that names no caller only when the + // first reached nobody — nothing serves it yet, or the bus does not grant it yet — so no call is done twice. + if called := CalledSubject(subject, c.cred.User); called != "" { + answered, handingOver, fallBack, err := c.askCalled(called, data) + if !fallBack { + if !handingOver { + return answered, err + } + time.Sleep(HandoverPause) + answered, again, _, err := c.askCalled(called, data) + if err != nil && again { + return Answered{}, fmt.Errorf("asked twice, %s apart, and both times refused as a handover: %w", HandoverPause, err) + } + return answered, err + } + } answered, handingOver, err := c.askOnce(subject, data) if !handingOver { return answered, err @@ -621,8 +684,14 @@ func (c *Conn) askOnce(subject string, data []byte) (Answered, bool, error) { } return Answered{}, false, err } + return c.answerOf(subject, msg.Data) +} + +// answerOf reads one answer, asking the rest of a paged one on the subject it was asked on, and says whether it +// was a refusal marked to be asked again. +func (c *Conn) answerOf(subject string, data []byte) (Answered, bool, error) { // An answer larger than one message comes in parts (pages.go, novox/hq issue 314). - body, err := c.whole(subject, msg.Data) + body, err := c.whole(subject, data) if err != nil { return Answered{}, false, err } diff --git a/node-tools/internal/bus/caller.go b/node-tools/internal/bus/caller.go new file mode 100644 index 0000000..5a817bb --- /dev/null +++ b/node-tools/internal/bus/caller.go @@ -0,0 +1,212 @@ +package bus + +import ( + "context" + "encoding/json" + "errors" + "regexp" + "strings" + "sync" + + "github.com/nats-io/nats.go" +) + +// A tool call names its caller (novox/hq issue 365, by ADR 0259 §3's precedent for asks). +// +// mesh.mod..call.. to whichever instance answers +// mesh.mod..call... to the instance on one machine +// mesh.seat..call.. a mesh seat's verb +// mesh.seat..call... a node seat's verb on one machine +// +// The last token is the bus user that published the call, and the controller grants each user the publish of +// these subjects with its own name there and no other — so the caller is a fact the bus server enforces, and +// the runtime hands it to the module from the subject a call arrived on, never from the payload. +// +// **A kind of its own, `call`, not the `tool` subject with a token appended.** Every grant to call a tool is a +// wildcard over the `tool` kind — `mesh.mod..tool..*` for the instance on any machine, `mesh.mod.*.tool.>` +// for every tool — and either matches a `tool` subject with any caller appended, so a caller could name +// another. Under `call` no grant but the caller-named ones reaches the caller token. +// +// The `tool` subjects are served beside these for one release, and say Unproven, so callers move without a +// gap; their retirement is hq issue 464. +const CallKind = "call" + +// A bus user's name carries dots (`.`, `person.`); a subject token cannot. In the token +// each dot is written `~`, which no part of a bus user's name may hold (the controller refuses any part that +// is not letters, digits, `_` or `-`), so the token is the user's name and nothing else. +const userDot = "~" + +var userName = regexp.MustCompile(`^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)*$`) + +// CallerToken is a bus user's name as the last token of a call it publishes, or "" for a name that is not a +// bus user's. +func CallerToken(user string) string { + if !userName.MatchString(user) { + return "" + } + return strings.ReplaceAll(user, ".", userDot) +} + +// UserOfToken is the bus user a caller token names, or "" when it names none. +func UserOfToken(token string) string { + user := strings.ReplaceAll(token, userDot, ".") + if !userName.MatchString(user) { + return "" + } + return user +} + +// toolShape splits a tool subject — `mesh.mod..tool.` or `mesh.seat..tool.` — into what +// comes before the kind and what after; ok is false for any other subject. +func toolShape(subject string) (head, rest string, ok bool) { + parts := strings.SplitN(subject, ".", 5) + if len(parts) != 5 || parts[0] != "mesh" || (parts[1] != "mod" && parts[1] != "seat") || parts[3] != "tool" || + parts[2] == "" || parts[4] == "" { + return "", "", false + } + return parts[0] + "." + parts[1] + "." + parts[2], parts[4], true +} + +// CalledSubject is where a call to a tool subject goes naming its caller: the same address under the `call` +// kind, the caller's token last. "" for a subject that is not a tool's, or a caller that is not a bus user. +func CalledSubject(subject, user string) string { + head, rest, ok := toolShape(subject) + token := CallerToken(user) + if !ok || token == "" || strings.ContainsAny(rest, "*>") { + return "" + } + return head + "." + CallKind + "." + rest + "." + token +} + +// calledPattern is what a runtime answering a tool subject also subscribes, to hear the calls that name their +// caller: the same address under the `call` kind, with any caller. "" for a subject that is not a tool's. +func calledPattern(subject string) string { + head, rest, ok := toolShape(subject) + if !ok || strings.ContainsAny(rest, "*>") { + return "" + } + return head + "." + CallKind + "." + rest + ".*" +} + +// callerOn is the bus user a call names, from the subject it arrived on: its last token, under a pattern +// calledPattern made. "" — said by the module as Unproven — for a call on the subject that names no caller. +func callerOn(pattern, subject string) string { + // Only the `call` kind names a caller: a `tool` pattern ending in a wildcard (a node seat's verb on any + // machine) is never read as one. + parts := strings.SplitN(pattern, ".", 5) + if len(parts) != 5 || parts[0] != "mesh" || (parts[1] != "mod" && parts[1] != "seat") || parts[3] != CallKind || + !strings.HasSuffix(pattern, ".*") { + return "" + } + dot := strings.LastIndex(subject, ".") + if dot < 0 || subject[:dot] != strings.TrimSuffix(pattern, ".*") { + return "" + } + return UserOfToken(subject[dot+1:]) +} + +// CalledHandler answers one call, told who called: the bus user the subject named, or "" when it named none. +type CalledHandler func(caller string, body json.RawMessage) (any, error) + +// callers is what a connection keeps of its caller-named calls: those in flight, by subject, so a refused +// publish ends its call at once instead of after RequestTimeout. +type callers struct { + mu sync.Mutex + // fellBack are the caller-named subjects already said to have reached nobody, so each is said once. + fellBack map[string]bool + next uint64 + pending map[string]map[uint64]context.CancelCauseFunc +} + +// errCallRefused is the bus refusing a caller-named publish: this user is not granted it yet (the controller's +// grant not delivered to the bus), so the call is asked again on the subject that names no caller. +var errCallRefused = errors.New("the bus refused the caller-named subject") + +var refusedPublish = regexp.MustCompile(`(?i)Publish to "(\S+)"`) + +// publishRefused ends every call in flight on a subject the bus refused this connection the publish of. +func (c *Conn) publishRefused(err error) bool { + if !errors.Is(err, nats.ErrPermissionViolation) { + return false + } + m := refusedPublish.FindStringSubmatch(err.Error()) + if len(m) < 2 { + return false + } + c.calls.mu.Lock() + cancels := c.calls.pending[m[1]] + delete(c.calls.pending, m[1]) + c.calls.mu.Unlock() + if len(cancels) > 0 { + c.Logf("[mesh-tools] the bus refused this account the caller-named %s; asked again on the subject that "+ + "names no caller, until the mesh grants it (novox/hq issue 365)", m[1]) + } + for _, cancel := range cancels { + cancel(errCallRefused) + } + return len(cancels) > 0 +} + +// askCalled asks once on a caller-named subject. fallBack says the call reached nobody — nothing serves that +// subject yet, or the bus refused this user its publish — so it may be asked on the subject that names no +// caller without being done twice. +func (c *Conn) askCalled(subject string, data []byte) (answered Answered, handingOver, fallBack bool, err error) { + ctx, cancel := context.WithCancelCause(context.Background()) + ctx, stop := context.WithTimeout(ctx, RequestTimeout) + defer stop() + c.calls.mu.Lock() + if c.calls.pending == nil { + c.calls.pending = map[string]map[uint64]context.CancelCauseFunc{} + } + if c.calls.pending[subject] == nil { + c.calls.pending[subject] = map[uint64]context.CancelCauseFunc{} + } + c.calls.next++ + id := c.calls.next + c.calls.pending[subject][id] = cancel + c.calls.mu.Unlock() + defer func() { + c.calls.mu.Lock() + if list := c.calls.pending[subject]; list != nil { + delete(list, id) + if len(list) == 0 { + delete(c.calls.pending, subject) + } + } + c.calls.mu.Unlock() + cancel(nil) + }() + msg, err := c.nc.RequestWithContext(ctx, subject, data) + if err != nil { + switch { + case errors.Is(context.Cause(ctx), errCallRefused): + return Answered{}, false, true, nil + case errors.Is(err, nats.ErrNoResponders): + c.saidFallBack(subject) + return Answered{}, false, true, nil + case errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout): + return Answered{}, false, false, errors.New("timeout") + } + return Answered{}, false, false, err + } + answered, handingOver, err = c.answerOf(subject, msg.Data) + return answered, handingOver, false, err +} + +// saidFallBack says, once per subject, that a caller-named call reached nobody and is asked on the subject that +// names no caller: a runtime or holder not yet serving it, or a grant that drifted, is seen in the log rather +// than read later as a quiet "unproven" (novox/hq issue 365; expected for the controller's seat until hq issue +// 464). +func (c *Conn) saidFallBack(subject string) { + c.calls.mu.Lock() + if c.calls.fellBack == nil { + c.calls.fellBack = map[string]bool{} + } + first := !c.calls.fellBack[subject] + c.calls.fellBack[subject] = true + c.calls.mu.Unlock() + if first { + c.Logf("[mesh-tools] nothing answers the caller-named %s; asked on the subject that names no caller, where "+ + "the module is told no caller (said once per subject; novox/hq issue 365)", subject) + } +} diff --git a/node-tools/internal/bus/caller_bus_test.go b/node-tools/internal/bus/caller_bus_test.go new file mode 100644 index 0000000..625c7fd --- /dev/null +++ b/node-tools/internal/bus/caller_bus_test.go @@ -0,0 +1,218 @@ +package bus_test + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// A bus whose users are granted as the controller grants them (mesh-controller internal/broker): each named +// user publishes the subjects listed for it. +func grantedBus(t *testing.T, publish map[string][]string) string { + t.Helper() + var users []*server.User + for user, allow := range publish { + // And the membership's direct get every connection asks once, answered "no responders" here. + allow = append(allow, "$JS.API.DIRECT.GET.>") + users = append(users, &server.User{Username: user, Password: "pw", Permissions: &server.Permissions{ + Publish: &server.SubjectPermission{Allow: allow}, + Subscribe: &server.SubjectPermission{Allow: []string{">"}}, + }}) + } + s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, NoLog: true, NoSigs: true, + Users: users}) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(s.Shutdown) + return s.ClientURL() +} + +func dial(t *testing.T, url, module, node, user string) *bus.Conn { + t.Helper() + c, err := bus.Connect(bus.Credential{URL: url, Module: module, Node: node, User: user, Password: "pw"}) + if err != nil { + t.Fatal(err) + } + c.Logf = func(string, ...any) {} + t.Cleanup(c.Close) + return c +} + +type heard struct { + mu sync.Mutex + callers []string +} + +func (h *heard) handler(caller string, _ json.RawMessage) (any, error) { + h.mu.Lock() + defer h.mu.Unlock() + h.callers = append(h.callers, caller) + return map[string]any{"caller": caller}, nil +} + +func (h *heard) all() []string { + h.mu.Lock() + defer h.mu.Unlock() + return append([]string(nil), h.callers...) +} + +// A caller asks on the caller-named subject first; where nothing serves it yet — a runtime from before issue +// 365 — the same call is asked once on the subject that names no caller, at once, and done once. +func TestACallerFallsBackAtOnceWhereNothingServesTheCallerNamedSubject(t *testing.T) { + url := mt.URL(t) + server := dial(t, url, "ledger", "anchor", "") + h := &heard{} + // An older runtime: the subject that names no caller, alone. + stop, err := server.HandleSubject("mesh.mod.ledger.tool.who", func(body json.RawMessage) (any, error) { + return h.handler("", body) + }) + if err != nil { + t.Fatal(err) + } + defer stop() + server.Flush() + asker := dial(t, url, "node-tools", "shanks", "shanks.node-tools") + began := time.Now() + answered, err := asker.Ask("ledger.who", map[string]any{}, "") + if err != nil { + t.Fatal(err) + } + if took := time.Since(began); took > 2*time.Second { + t.Errorf("the fall-back took %s", took) + } + if got := h.all(); len(got) != 1 || got[0] != "" || !strings.Contains(string(answered.Result), `"caller":""`) { + t.Errorf("the call was answered %v, heard %v; want once, on the old subject", string(answered.Result), got) + } +} + +// Where the bus does not grant the caller-named subject yet — the controller's grants not yet delivered — the +// refusal ends the call at once and it is asked on the subject that names no caller: no caller waits out +// RequestTimeout for a grant that is on its way. +func TestACallerTheBusRefusesTheCallerNamedSubjectFallsBackAtOnce(t *testing.T) { + url := grantedBus(t, map[string][]string{ + "anchor.node-tools": {"_INBOX.>"}, + "shanks.node-tools": {"mesh.mod.*.tool.>"}, + }) + server := dial(t, url, "node-tools", "anchor", "anchor.node-tools") + h := &heard{} + stop, err := server.HandleSubjectCalled("mesh.mod.ledger.tool.who", h.handler) + if err != nil { + t.Fatal(err) + } + defer stop() + server.Flush() + asker := dial(t, url, "node-tools", "shanks", "shanks.node-tools") + var said []string + var mu sync.Mutex + asker.Logf = func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() } + began := time.Now() + if _, err := asker.Ask("ledger.who", map[string]any{}, ""); err != nil { + t.Fatal(err) + } + mu.Lock() + if !strings.Contains(strings.Join(said, "\n"), "the bus refused this account the caller-named mesh.mod.ledger.call.who.shanks~node-tools") { + t.Errorf("the refusal was not said: %v", said) + } + mu.Unlock() + if took := time.Since(began); took > 3*time.Second { + t.Errorf("the refused caller-named call held the caller %s", took) + } + if got := h.all(); len(got) != 1 || got[0] != "" { + t.Errorf("heard %v; want one call, on the subject that names no caller", got) + } +} + +// With the controller's grant — the caller-named subjects with its own name last — a caller is heard as itself; +// and a caller cannot publish on a subject naming another, however it asks. +func TestAGrantedCallerIsHeardAsItselfAndCannotNameAnother(t *testing.T) { + url := grantedBus(t, map[string][]string{ + "anchor.node-tools": {"_INBOX.>"}, + "shanks.node-tools": {"mesh.mod.*.call.*.shanks~node-tools", "mesh.mod.*.call.*.*.shanks~node-tools"}, + }) + server := dial(t, url, "node-tools", "anchor", "anchor.node-tools") + h := &heard{} + stop, err := server.HandleSubjectCalled("mesh.mod.ledger.tool.who", h.handler) + if err != nil { + t.Fatal(err) + } + defer stop() + server.Flush() + asker := dial(t, url, "node-tools", "shanks", "shanks.node-tools") + if _, err := asker.Ask("ledger.who", map[string]any{}, ""); err != nil { + t.Fatal(err) + } + // Naming another on the subject itself: refused by the bus, so nothing is heard. + refusals := make(chan error, 4) + forger, err := nats.Connect(url, nats.UserInfo("shanks.node-tools", "pw"), + nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err })) + if err != nil { + t.Fatal(err) + } + defer forger.Close() + if err := forger.Publish("mesh.mod.ledger.call.who.novox~node-tools", []byte(`{}`)); err != nil { + t.Fatal(err) + } + _ = forger.Flush() + select { + case err := <-refusals: + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("the bus said %v, want a permissions violation", err) + } + case <-time.After(3 * time.Second): + t.Error("the bus did not refuse a call naming another caller") + } + if got := h.all(); len(got) != 1 || got[0] != "shanks.node-tools" { + t.Errorf("heard %v; want the asker alone, as itself", got) + } +} + +// A fall-back for want of a responder on the caller-named subject is said, once per subject, so a runtime or +// grant that drifted behind is seen rather than read as a quiet "unproven". +func TestAFallBackForWantOfAResponderIsSaidOncePerSubject(t *testing.T) { + url := mt.URL(t) + // A runtime from before issue 365: the subject that names no caller, alone. + older, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer older.Close() + if _, err := older.Subscribe("mesh.mod.ledger.tool.who", func(m *nats.Msg) { _ = m.Respond([]byte(`{"result":{}}`)) }); err != nil { + t.Fatal(err) + } + _ = older.Flush() + asker := dial(t, url, "node-tools", "shanks", "shanks.node-tools") + var mu sync.Mutex + var said []string + asker.Logf = func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() } + for i := 0; i < 3; i++ { + if _, err := asker.Ask("ledger.who", map[string]any{}, ""); err != nil { + t.Fatal(err) + } + } + mu.Lock() + defer mu.Unlock() + n := 0 + for _, s := range said { + if strings.Contains(s, "nothing answers the caller-named mesh.mod.ledger.call.who.shanks~node-tools") { + n++ + } + } + if n != 1 { + t.Errorf("the fall-back was said %d times, want once: %v", n, said) + } +} diff --git a/node-tools/internal/bus/caller_test.go b/node-tools/internal/bus/caller_test.go new file mode 100644 index 0000000..5c4da68 --- /dev/null +++ b/node-tools/internal/bus/caller_test.go @@ -0,0 +1,72 @@ +package bus + +import "testing" + +// A call names its caller under the `call` kind, the caller's bus user its last token (novox/hq issue 365). +func TestACallerNamedSubjectIsTheToolsAddressUnderCallWithTheCallerLast(t *testing.T) { + for subject, want := range map[string]string{ + "mesh.mod.ledger.tool.who": "mesh.mod.ledger.call.who.shanks~node-tools", + "mesh.mod.ledger.tool.who.anchor": "mesh.mod.ledger.call.who.anchor.shanks~node-tools", + "mesh.seat.issue-tracker.tool.open": "mesh.seat.issue-tracker.call.open.shanks~node-tools", + "mesh.seat.node-desk.tool.who.anchor": "mesh.seat.node-desk.call.who.anchor.shanks~node-tools", + "mesh.mod.ledger.event.opened": "", + "mesh.seat.intake.accept.ask.x": "", + "mesh.mod.ledger.tool.>": "", + "mesh.mod.*.tool.who": "mesh.mod.*.call.who.shanks~node-tools", + "mesh.node.anchor.ask.report": "", + "mesh.seat.operator-channel.proof.x.y": "", + } { + if subject == "mesh.mod.*.tool.who" { + continue // a wildcard is never asked on + } + if got := CalledSubject(subject, "shanks.node-tools"); got != want { + t.Errorf("%s → %q, want %q", subject, got, want) + } + } + for _, user := range []string{"", "a b", "x.*", "x.>", "a~b", "a..b"} { + if got := CalledSubject("mesh.mod.ledger.tool.who", user); got != "" { + t.Errorf("a caller %q that is not a bus user was named: %s", user, got) + } + } +} + +// The caller is read back from the subject a call arrived on, and only under the pattern that names one. +func TestTheCallerIsTheLastTokenOfTheSubjectTheCallArrivedOn(t *testing.T) { + pattern := calledPattern("mesh.mod.ledger.tool.who.anchor") + if pattern != "mesh.mod.ledger.call.who.anchor.*" { + t.Fatalf("pattern %q", pattern) + } + for subject, want := range map[string]string{ + "mesh.mod.ledger.call.who.anchor.shanks~node-tools": "shanks.node-tools", + "mesh.mod.ledger.call.who.anchor.controller": "controller", + "mesh.mod.ledger.call.who.anchor.person~jochen": "person.jochen", + "mesh.mod.ledger.call.who.anchor.a~~b": "", + "mesh.mod.ledger.call.what.anchor.controller": "", + } { + if got := callerOn(pattern, subject); got != want { + t.Errorf("%s → %q, want %q", subject, got, want) + } + } + if got := callerOn("mesh.mod.ledger.tool.who.anchor", "mesh.mod.ledger.tool.who.anchor"); got != "" { + t.Errorf("a call on the subject that names no caller was handed %q", got) + } + for _, user := range []string{"controller", "shanks.node-tools", "person.jochen", "node.ace"} { + if back := UserOfToken(CallerToken(user)); back != user { + t.Errorf("%s → %s → %s", user, CallerToken(user), back) + } + } +} + +// Only a pattern of the `call` kind names a caller: a `tool` subscription with a wildcard last — a node seat's +// verb on any machine, say — never yields one, whatever token a call arrived with. +func TestAToolPatternNeverYieldsACaller(t *testing.T) { + for pattern, subject := range map[string]string{ + "mesh.mod.ledger.tool.who.*": "mesh.mod.ledger.tool.who.controller", + "mesh.seat.node-desk.tool.who.*": "mesh.seat.node-desk.tool.who.shanks~node-tools", + "mesh.mod.ledger.event.*": "mesh.mod.ledger.event.controller", + } { + if got := callerOn(pattern, subject); got != "" { + t.Errorf("%s on %s yielded caller %q", pattern, subject, got) + } + } +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index dc68ae5..b07cc3c 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -37,7 +37,24 @@ type Tool struct { Name string Description string Input json.RawMessage - Run func(args json.RawMessage) (json.RawMessage, error) + // Run calls the tool for caller: the bus user the call's subject named, or "" when it named none. A + // caller is handed to the bundle in the call's `_meta` (MetaCaller), never among its arguments. + Run func(caller string, args json.RawMessage) (json.RawMessage, error) +} + +// MetaCaller is where in a `tools/call`'s `_meta` the runtime names the caller it took from the subject the +// call arrived on (novox/hq issue 365): mesh-sdk's stdio.MetaCaller, which hands it to the tool as +// stdio.CallerOf, and says Unproven when it is absent. +const MetaCaller = "mesh/caller" + +// callParams is a `tools/call`'s params: the tool, its arguments as the caller sent them, and — only when the +// subject named one — the caller, in `_meta`, which the runtime writes and no caller can. +func callParams(name, caller string, args json.RawMessage) map[string]any { + params := map[string]any{"name": name, "arguments": args} + if caller != "" { + params["_meta"] = map[string]any{MetaCaller: caller} + } + return params } // Registration is the tools a bundle listed under one name: its module's, or a seat's. @@ -484,11 +501,11 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an input = json.RawMessage("{}") } tool := Tool{Name: name, Description: t.Description, Input: input, - Run: func(args json.RawMessage) (json.RawMessage, error) { + Run: func(caller string, args json.RawMessage) (json.RawMessage, error) { if len(args) == 0 || string(args) == "null" { args = json.RawMessage("{}") } - res, err := asking("tools/call", map[string]any{"name": full, "arguments": args}, CallTimeout) + res, err := asking("tools/call", callParams(full, caller, args), CallTimeout) if err != nil { return nil, err } diff --git a/node-tools/internal/runtime/caller_test.go b/node-tools/internal/runtime/caller_test.go new file mode 100644 index 0000000..54ca2fe --- /dev/null +++ b/node-tools/internal/runtime/caller_test.go @@ -0,0 +1,99 @@ +package runtime + +import ( + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +type saidCaller struct { + Caller string `json:"caller"` + Arguments map[string]any `json:"arguments"` +} + +// **A module is told who called it, from the subject the call arrived on** (novox/hq issue 365): a call on the +// caller-named subject reaches the bundle with that caller in the call's `_meta`, for a module's tool, on its +// machine's address, and for a seat's verb; a caller named in the arguments is only an argument. +func TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller(t *testing.T) { + mesh := mt.New(t) + mesh.Issue(t, mt.MembershipOf("ledger", "anchor", true, map[string][]string{"node-desk": {"who"}})) + nodeTools := connect(t, "node-tools", "anchor") + stop, err := Run(nodeTools, []Served{{"ledger", []string{mt.Fixture("caller-echo.py")}}}, nil, func(string, ...any) {}) + if err != nil { + t.Fatal(err) + } + defer settled(stop) + asker, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "shanks", + User: "shanks.node-tools", Password: "unused"}) + if err != nil { + t.Fatal(err) + } + asker.Logf = func(string, ...any) {} + defer asker.Close() + + for _, key := range []string{"ledger.who", "ledger.who@anchor", "seat:node-desk.who@anchor"} { + var got saidCaller + mt.Until(t, func() error { + answered, err := asker.Ask(key, map[string]any{"caller": "novox.node-tools", + "_meta": map[string]any{"mesh/caller": "novox.controller"}}, "") + if err == nil { + err = json.Unmarshal(answered.Result, &got) + } + return err + }) + if got.Caller != "shanks.node-tools" { + t.Errorf("%s reached the module with caller %q, want the asker's bus user", key, got.Caller) + } + if got.Arguments["caller"] != "novox.node-tools" { + t.Errorf("%s: the arguments did not reach the module as sent: %v", key, got.Arguments) + } + } +} + +// The subject that names no caller is still served, for one release (retired by hq issue 464), and says so: +// the module is named no caller, whatever the body claims. +func TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone(t *testing.T) { + mesh := mt.New(t) + mesh.Issue(t, mt.MembershipOf("ledger", "anchor", true, map[string][]string{"node-desk": {"who"}})) + nodeTools := connect(t, "node-tools", "anchor") + stop, err := Run(nodeTools, []Served{{"ledger", []string{mt.Fixture("caller-echo.py")}}}, nil, func(string, ...any) {}) + if err != nil { + t.Fatal(err) + } + defer settled(stop) + old, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + defer old.Close() + body, _ := json.Marshal(map[string]any{"_meta": map[string]any{"mesh/caller": "novox.controller"}}) + for _, subject := range []string{"mesh.mod.ledger.tool.who", "mesh.mod.ledger.tool.who.anchor", "mesh.seat.node-desk.tool.who.anchor"} { + var reply struct { + Result saidCaller `json:"result"` + Error string `json:"error"` + } + mt.Until(t, func() error { + msg, err := old.Request(subject, body, 5*time.Second) + if err == nil { + err = json.Unmarshal(msg.Data, &reply) + } + return err + }) + if reply.Error != "" || reply.Result.Caller != "" { + t.Errorf("%s: the module was named caller %q (error %q), want none", subject, reply.Result.Caller, reply.Error) + } + } +} + +// settled stops a runtime and waits for its bundle's exit to be handled: the launcher reads its restart pauses +// (launch.RestartFirst) as a bundle exits, and a later test in this package sets them, so a test that leaves its +// bundle exiting behind it races the next one. +func settled(stop func()) { + stop() + time.Sleep(500 * time.Millisecond) +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index edc1498..2c106fa 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -230,8 +230,9 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log for _, r := range own { for _, t := range r.tools { t := t - stop, err := conn.Handle(r.module+"."+t.Name, func(body json.RawMessage) (any, error) { - return t.Run(argsOf(body)) + // Told who called, from the subject the call arrived on, never from its body (novox/hq issue 365). + stop, err := conn.HandleCalled(r.module+"."+t.Name, func(caller string, body json.RawMessage) (any, error) { + return t.Run(caller, argsOf(body)) }) if err != nil { logf("[mesh-tools] cannot serve %s.%s: %v", r.module, t.Name, err) @@ -424,8 +425,8 @@ func serveSeats(conn *bus.Conn, modules []string, registrations []registration, logf("[mesh-tools] %s claims %s and implements no %s, which that seat promises; not served", module, v.Seat, v.Verb) continue } - stop, err := conn.HandleSubject(v.Subject, func(body json.RawMessage) (any, error) { - return t.Run(argsOf(body)) + stop, err := conn.HandleSubjectCalled(v.Subject, func(caller string, body json.RawMessage) (any, error) { + return t.Run(caller, argsOf(body)) }) if err != nil { logf("[mesh-tools] cannot serve %s's %s on %s: %v", v.Seat, v.Verb, v.Subject, err) diff --git a/node-tools/test/fixtures/caller-echo.py b/node-tools/test/fixtures/caller-echo.py new file mode 100755 index 0000000..16706cd --- /dev/null +++ b/node-tools/test/fixtures/caller-echo.py @@ -0,0 +1,24 @@ +#!/usr/bin/env python3 +# A tools bundle that says who the runtime said called it (novox/hq issue 365): MCP over stdio, no SDK. Its tool +# and its seat verb answer the caller the call's `_meta` named — "" when it named none — and the arguments as given. +import json, sys + +def say(m): + m["jsonrpc"] = "2.0"; sys.stdout.write(json.dumps(m) + "\n"); sys.stdout.flush() + +TOOLS = [ + {"name": "who", "description": "who called", "inputSchema": {"type": "object", "properties": {}}}, + {"name": "node-desk.who", "description": "who called the seat's verb", "inputSchema": {"type": "object", "properties": {}}}, +] +for line in sys.stdin: + req = json.loads(line); rid = req.get("id"); m = req.get("method"); p = req.get("params") or {} + if m == "initialize": + say({"id": rid, "result": {"protocolVersion": "2025-03-26", "capabilities": {"tools": {}}, "serverInfo": {"name": "ledger", "version": "1"}}}) + elif m == "tools/list": + say({"id": rid, "result": {"tools": TOOLS}}) + elif m == "tools/call": + meta = p.get("_meta") or {} + said = {"caller": meta.get("mesh/caller", ""), "arguments": p.get("arguments") or {}} + say({"id": rid, "result": {"content": [{"type": "text", "text": json.dumps(said)}]}}) + elif rid is not None: + say({"id": rid, "result": {}})