Merge pull request 'node-tools: a tool call names its caller, and the module is told it from the subject (hq issue 365)' (#27) from fix/365-a-tool-call-names-its-caller into main
This commit was merged in pull request #27.
This commit is contained in:
+1
-1
@@ -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
|
||||
|
||||
@@ -5,6 +5,8 @@
|
||||
// mesh.mod.<module>.event.<type> an event a module emits
|
||||
// mesh.mod.<module>.tool.<tool> a tool a module serves
|
||||
// mesh.seat.<seat>.tool.<verb> a role's verb, answered by whoever holds the seat
|
||||
// mesh.mod.<module>.call.<tool>.<caller>, mesh.seat.<seat>.call.<verb>.<caller>
|
||||
// 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 `<module>.<tool>` 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
|
||||
}
|
||||
|
||||
@@ -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.<module>.call.<tool>.<caller> to whichever instance answers
|
||||
// mesh.mod.<module>.call.<tool>.<node>.<caller> to the instance on one machine
|
||||
// mesh.seat.<seat>.call.<verb>.<caller> a mesh seat's verb
|
||||
// mesh.seat.<seat>.call.<verb>.<node>.<caller> 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.<m>.tool.<t>.*` 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 (`<node>.<module>`, `person.<name>`); 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.<m>.tool.<rest>` or `mesh.seat.<s>.tool.<rest>` — 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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
+24
@@ -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": {}})
|
||||
Reference in New Issue
Block a user