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
This commit is contained in:
@@ -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
|
||||
|
||||
+1
-2
@@ -24,8 +24,7 @@ var CallTimeout = 30 * time.Second
|
||||
// mesh-provisioner-client/secrets.go:59-61, 93-97, 119, 125-133). key is "seat:<seat>.<verb>" for a seat's
|
||||
// verb, or "<module>.<tool>" 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())
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+2
-2
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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.<seat>.call.<verb>.*`, handing the
|
||||
// handler the caller its last token names, and on `mesh.seat.<seat>.tool.<verb>`, 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)
|
||||
}
|
||||
Reference in New Issue
Block a user