From ba6c2a5207586e0caf5009d938ce4191cf966fc4 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 23:50:15 +0100 Subject: [PATCH] process: AnswerSeat answers a seat's verbs over the bus with the proved caller A process answers its seat's verbs through one call: caller-named and unproven subjects, one queue group per seat, reply {result, error, node}. The rule that a process answers no call (issue 1171 AC3) is replaced by one that AnswerSeat is the only way to serve. A compare-and-set write on a process's own state already exists (State.Create, Update). Builds: design 47 G A-107 (G-20-lead-answers) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/process/account.go | 2 +- go/process/call.go | 3 +- go/process/call_integration_test.go | 85 +++++++++++++++++++++ go/process/call_test.go | 18 +++++ go/process/conn.go | 4 +- go/process/no_serve_test.go | 13 ++-- go/process/serve.go | 113 ++++++++++++++++++++++++++++ 7 files changed, 228 insertions(+), 10 deletions(-) create mode 100644 go/process/serve.go diff --git a/go/process/account.go b/go/process/account.go index eba3acb0..9bbbf72e 100644 --- a/go/process/account.go +++ b/go/process/account.go @@ -2,7 +2,7 @@ // the account the controller delivers, connects as it with its pinned fingerprint and its inbox prefix, and // offers emitting an event, the module's states with compare-and-set, consuming a work queue, reading the // router's records of the module's own asks, and calling a tool or verb — each only as the account's grants -// allow. It offers no Serve and no way to serve a tool: a process never answers a call. A sibling of stdio, +// allow. It answers its seat's verbs through AnswerSeat alone (design 47 G, lead answer A-107). A sibling of stdio, // which it imports only for stdio's pure types (ErrChanged, CodeChanged, StateEntry, EmitOptions); it never // touches stdio's current, channel or live(), which stay the bundle's own way to the bus. package process diff --git a/go/process/call.go b/go/process/call.go index 870598c7..791af6e1 100644 --- a/go/process/call.go +++ b/go/process/call.go @@ -24,8 +24,7 @@ var CallTimeout = 30 * time.Second // 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. A key naming its machine after '@' (`seat:node-service-manager.stop@novox`) is CallOn of that +// call's failure. A process answers calls through AnswerSeat (serve.go). A key naming its machine after '@' (`seat:node-service-manager.stop@novox`) is CallOn of that // machine; any other malformed key is refused before anything is published (calladdress; novox/hq issue 774). func (c *Conn) Call(key string, body any) (json.RawMessage, error) { called, fallback, err := callSubjectsOn(key, "", c.callerToken()) diff --git a/go/process/call_integration_test.go b/go/process/call_integration_test.go index 09345a47..5da784ec 100644 --- a/go/process/call_integration_test.go +++ b/go/process/call_integration_test.go @@ -6,6 +6,9 @@ import ( "context" "encoding/json" "errors" + "strings" + "sync" + "sync/atomic" "testing" "time" @@ -281,3 +284,85 @@ func TestCallOnReachesTheNamedNodeAlone(t *testing.T) { default: } } + +// A-107: a process answers its seat's verbs over the bus. Given two instances of one holder answering seat +// `broker-seat`'s verbs, a call on the caller-named subject reaches exactly one, which is handed the caller the +// subject names (dots written `~`) and replies {result, node}; a call on the subject naming none is handed ""; a +// handler's error is the reply's error, a panic too; and after stop nothing answers. +func TestAnswerSeatServesVerbsWithTheProvedCallerInOneQueueGroup(t *testing.T) { + url, account := grantedSendProcess(t) + adminNC, _ := adminJS(t, url) + account.User, account.Password, account.Node = adminUser, adminPass, "novox" + + var heard sync.Map + count := func(name string) SeatHandler { + return func(caller string, body json.RawMessage) (any, error) { + n, _ := heard.LoadOrStore(name, new(int32)) + atomic.AddInt32(n.(*int32), 1) + return map[string]string{"caller": caller, "body": string(body)}, nil + } + } + verbs := func(name string) map[string]SeatHandler { + return map[string]SeatHandler{ + "ping": count(name), + "fail": func(string, json.RawMessage) (any, error) { return nil, errors.New("refused: not yours") }, + "boom": func(string, json.RawMessage) (any, error) { panic("oops") }, + } + } + var stops []func() + for _, name := range []string{"a", "b"} { + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + stop, err := conn.AnswerSeat("broker-seat", verbs(name)) + if err != nil { + t.Fatalf("AnswerSeat: %v", err) + } + stops = append(stops, stop) + } + ask := func(subject string) (struct { + Result map[string]string `json:"result"` + Error string `json:"error"` + Node string `json:"node"` + }, error) { + var r struct { + Result map[string]string `json:"result"` + Error string `json:"error"` + Node string `json:"node"` + } + msg, err := adminNC.Request(subject, []byte(`{"x":1}`), 5*time.Second) + if err != nil { + return r, err + } + return r, json.Unmarshal(msg.Data, &r) + } + + r, err := ask("mesh.seat.broker-seat.call.ping.novox~mesh-delivery-builder") + if err != nil || r.Result["caller"] != "novox.mesh-delivery-builder" || r.Result["body"] != `{"x":1}` || r.Node != "novox" { + t.Fatalf("a caller-named call: %+v, %v", r, err) + } + r, err = ask("mesh.seat.broker-seat.tool.ping") + if err != nil || r.Result["caller"] != "" { + t.Fatalf("an unproven call: %+v, %v", r, err) + } + total := int32(0) + heard.Range(func(_, v any) bool { total += atomic.LoadInt32(v.(*int32)); return true }) + if total != 2 { + t.Errorf("two calls were answered %d times in all, want once each (one queue group)", total) + } + if r, err = ask("mesh.seat.broker-seat.call.fail.x"); err != nil || r.Error != "refused: not yours" { + t.Errorf("a handler's error: %+v, %v", r, err) + } + if r, err = ask("mesh.seat.broker-seat.call.boom.x"); err != nil || !strings.Contains(r.Error, "oops") { + t.Errorf("a handler's panic: %+v, %v", r, err) + } + for _, stop := range stops { + stop() + } + _ = adminNC.Flush() + if _, err := adminNC.Request("mesh.seat.broker-seat.call.ping.x", nil, time.Second); !errors.Is(err, nats.ErrNoResponders) { + t.Errorf("after stop a call reached someone: %v", err) + } +} diff --git a/go/process/call_test.go b/go/process/call_test.go index e4eefaf3..750e981f 100644 --- a/go/process/call_test.go +++ b/go/process/call_test.go @@ -136,3 +136,21 @@ func TestACallIsAskedAgainOnlyWhenItReachedNoOne(t *testing.T) { } } } + +// A-107: AnswerSeat reads the caller from a call's last token (dots written `~`), "" for anything that is not a +// bus user's name; and it refuses what it cannot serve before subscribing. +func TestUserOfTokenAndAnswerSeatRefusals(t *testing.T) { + for token, want := range map[string]string{"novox~mesh-delivery-builder": "novox.mesh-delivery-builder", "person~jo": "person.jo", + "plain": "plain", "": "", "a b": "", "a*": "", "~": ""} { + if got := userOfToken(token); got != want { + t.Errorf("userOfToken(%q) = %q, want %q", token, got, want) + } + } + c := &Conn{} + ok := func(string, json.RawMessage) (any, error) { return nil, nil } + for name, verbs := range map[string]map[string]SeatHandler{"none": nil, "nil handler": {"v": nil}, "bad verb": {"a b": ok}, "wildcard": {"*": ok}} { + if _, err := c.AnswerSeat("seat", verbs); err == nil { + t.Errorf("AnswerSeat accepted %s", name) + } + } +} diff --git a/go/process/conn.go b/go/process/conn.go index 3809c429..2b6ecaf5 100644 --- a/go/process/conn.go +++ b/go/process/conn.go @@ -15,8 +15,8 @@ import ( ) // Conn is one process's own connection to the bus (novox/hq ADR 0436 clause 4): dialled with the account -// the controller delivered, pinned to its fingerprint, on its own inbox prefix. It offers no Serve and no -// way to serve a tool — a process reaches the bus only as itself, and never answers a call. +// the controller delivered, pinned to its fingerprint, on its own inbox prefix. It answers its seat's verbs +// through AnswerSeat alone (design 47 G, lead answer A-107) and offers no other way to serve a tool. type Conn struct { nc *nats.Conn account Account diff --git a/go/process/no_serve_test.go b/go/process/no_serve_test.go index f7eb72fe..923ced68 100644 --- a/go/process/no_serve_test.go +++ b/go/process/no_serve_test.go @@ -7,14 +7,17 @@ import ( "github.com/nats-io/nats.go" ) -// 1171/AC3: it offers no Serve and no way to serve a tool — a process reaches the bus only as itself, and -// never answers a call. Checked by reflection over *Conn's whole method set and field list, so a later -// addition cannot bring one back without this test naming it. -func TestConnOffersNoServeAndExposesNoRawConnection(t *testing.T) { +// 1171/AC3 as lead answer A-107 changes it: a process answers its seat's verbs through AnswerSeat alone — no Serve, +// Handle, Subscribe or Respond of its own — and exposes no raw connection. Checked by reflection over *Conn's whole +// method set and field list, so a later addition cannot bring one back without this test naming it. +func TestConnOffersOnlyAnswerSeatToServeAndExposesNoRawConnection(t *testing.T) { ptrType := reflect.TypeOf(&Conn{}) + if _, ok := ptrType.MethodByName("AnswerSeat"); !ok { + t.Error("*Conn has no AnswerSeat: a process answers its seat's verbs through it (A-107)") + } for _, forbidden := range []string{"Serve", "ServeOn", "Handle", "Subscribe", "Respond", "Answer"} { if _, ok := ptrType.MethodByName(forbidden); ok { - t.Errorf("*Conn has a method %s: a process offers no way to serve a tool", forbidden) + t.Errorf("*Conn has a method %s: a process serves only through AnswerSeat", forbidden) } } diff --git a/go/process/serve.go b/go/process/serve.go new file mode 100644 index 00000000..03bbc79b --- /dev/null +++ b/go/process/serve.go @@ -0,0 +1,113 @@ +package process + +import ( + "encoding/json" + "fmt" + "regexp" + "sort" + "strings" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/calladdress" +) + +// SeatHandler answers one call of a seat's verb: caller is the bus user the subject the call arrived on names +// (the bus server enforces it, so it is a fact, never read from the body), or "" when the call came on the subject +// that names none (an unproven caller); body is the call's arguments. A returned error is the call's failure. +type SeatHandler func(caller string, body json.RawMessage) (any, error) + +// callerUser is the bus user in a caller token: each `~` is a dot (mesh-tools node-tools/internal/bus/caller.go +// UserOfToken), "" for a token that is not a bus user's name. +var callerUser = regexp.MustCompile(`^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)*$`) + +func userOfToken(token string) string { + user := strings.ReplaceAll(token, "~", ".") + if !callerUser.MatchString(user) { + return "" + } + return user +} + +// AnswerSeat answers the verbs of a mesh seat this process's module holds, over the bus, as the tool runner +// answers a bundle's (novox/hq design 47 G, lead answer A-107): on `mesh.seat..call..*`, handing the +// handler the caller its last token names, and on `mesh.seat..tool.`, which names none. Every instance +// of the seat's holder subscribes in one queue group per seat, so a call reaches one of them. The reply is +// {result, error, node}, as Call reads it. The returned stop ends every subscription. Nothing is subscribed when +// any verb is malformed or a subscription is refused. +func (c *Conn) AnswerSeat(seat string, verbs map[string]SeatHandler) (stop func(), err error) { + if len(verbs) == 0 { + return nil, fmt.Errorf("AnswerSeat(%q) answers no verb", seat) + } + names := make([]string, 0, len(verbs)) + for verb, h := range verbs { + if h == nil { + return nil, fmt.Errorf("AnswerSeat(%q): the verb %s has no handler", seat, verb) + } + if _, err := calladdress.Parse("seat:" + seat + "." + verb); err != nil { + return nil, err + } + names = append(names, verb) + } + sort.Strings(names) + queue := "serve.seat." + seat + var subs []*nats.Subscription + undo := func() { + for _, s := range subs { + _ = s.Unsubscribe() + } + } + for _, verb := range names { + h := verbs[verb] + base := "mesh.seat." + seat + for _, subject := range []string{base + ".call." + verb + ".*", base + ".tool." + verb} { + subject := subject + sub, err := c.nc.QueueSubscribe(subject, queue, func(msg *nats.Msg) { go c.answerCall(msg, subject, h) }) + if err != nil { + undo() + return nil, fmt.Errorf("answering %s: %w", subject, err) + } + subs = append(subs, sub) + } + } + if err := c.nc.Flush(); err != nil { + undo() + return nil, err + } + return undo, nil +} + +// answerCall runs h for one call and replies {result, error, node}; a panic in h is the call's error, never silence. +func (c *Conn) answerCall(msg *nats.Msg, pattern string, h SeatHandler) { + caller := "" + if strings.HasSuffix(pattern, ".*") { + caller = userOfToken(msg.Subject[strings.LastIndex(msg.Subject, ".")+1:]) + } + var reply struct { + Result any `json:"result,omitempty"` + Error string `json:"error,omitempty"` + Node string `json:"node,omitempty"` + } + func() { + defer func() { + if r := recover(); r != nil { + reply.Error = fmt.Sprintf("%s failed: %v", msg.Subject, r) + } + }() + result, err := h(caller, json.RawMessage(msg.Data)) + switch { + case err != nil: + reply.Error = err.Error() + case result == nil: + reply.Result = json.RawMessage("null") + default: + reply.Result = result + } + }() + reply.Node = c.account.Node + body, err := json.Marshal(reply) + if err != nil { + body, _ = json.Marshal(map[string]string{"error": "the answer is not JSON: " + err.Error(), "node": c.account.Node}) + } + _ = msg.Respond(body) +}