123 lines
4.8 KiB
Go
123 lines
4.8 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// CallTimeout bounds a call to a tool or a seat verb (mesh-tools node-tools/internal/bus/bus.go:35,
|
|
// RequestTimeout): what a module already expects.
|
|
var CallTimeout = 30 * time.Second
|
|
|
|
// Call invokes a tool or a seat verb as this process's own account (novox/hq ADR 0436 clause 4), on the
|
|
// caller-named subject, falling back to the subject that names no caller when nothing answers it —
|
|
// exactly the provisioner client's own pattern (mesh-catalog modules/mesh-provisioner-client/cmd/
|
|
// mesh-provisioner-client/secrets.go:59-61, 93-97, 119, 125-133). key is "seat:<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.
|
|
func (c *Conn) Call(key string, body any) (json.RawMessage, error) {
|
|
called, fallback, err := callSubjects(key, c.callerToken())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return c.callOn(called, fallback, body)
|
|
}
|
|
|
|
// CallOn is Call to the tool or seat verb of one named machine (novox/hq ADR 0469 §3; issue 1470): the tool
|
|
// runner of that node alone answers, on mesh.mod.<module>.call.<tool>.<node>.<token> (a seat verb:
|
|
// mesh.seat.<seat>.call.<verb>.<node>.<token>), falling back to the subject that names no caller
|
|
// (mesh-tools node-tools/internal/bus/caller.go:15-19). node is one subject token: an empty one, or one
|
|
// holding '.', '*', '>' or white space, is refused before anything is published.
|
|
func (c *Conn) CallOn(node, key string, body any) (json.RawMessage, error) {
|
|
if err := checkNode(node); err != nil {
|
|
return nil, err
|
|
}
|
|
called, fallback, err := callSubjectsOn(key, node, c.callerToken())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return c.callOn(called, fallback, body)
|
|
}
|
|
|
|
func (c *Conn) callOn(called, fallback string, body any) (json.RawMessage, error) {
|
|
data, err := json.Marshal(body)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result, err := c.request(called, data)
|
|
if errors.Is(err, nats.ErrNoResponders) {
|
|
result, err = c.request(fallback, data)
|
|
}
|
|
return result, err
|
|
}
|
|
|
|
// callerToken is this account's user name as the last token of a call it publishes, each dot written ~
|
|
// (mesh-tools node-tools/internal/bus/caller.go:43, CallerToken).
|
|
func (c *Conn) callerToken() string { return strings.ReplaceAll(c.account.User, ".", "~") }
|
|
|
|
// callSubjects is the caller-named subject key is called on, and its plain fallback (mesh-tools
|
|
// node-tools/internal/bus/caller.go CalledSubject, toolShape, calledPattern).
|
|
func callSubjects(key, token string) (called, fallback string, err error) {
|
|
return callSubjectsOn(key, "", token)
|
|
}
|
|
|
|
// checkNode refuses a node name that is not one subject token.
|
|
func checkNode(node string) error {
|
|
if node == "" || strings.ContainsAny(node, ".*> \t\r\n") {
|
|
return fmt.Errorf("%q is not a node name (one subject token: no '.', '*', '>' or space)", node)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// callSubjectsOn is callSubjects for one named node ("" names none): the node is a token between the tool and
|
|
// the caller on the called subject, and the last one of the fallback.
|
|
func callSubjectsOn(key, node, token string) (called, fallback string, err error) {
|
|
at := ""
|
|
if node != "" {
|
|
at = "." + node
|
|
}
|
|
if rest, ok := strings.CutPrefix(key, "seat:"); ok {
|
|
seat, verb, ok := strings.Cut(rest, ".")
|
|
if !ok || seat == "" || verb == "" {
|
|
return "", "", fmt.Errorf("%q names no seat and verb (seat:<seat>.<verb>)", key)
|
|
}
|
|
return "mesh.seat." + seat + ".call." + verb + at + "." + token, "mesh.seat." + seat + ".tool." + verb + at, nil
|
|
}
|
|
module, tool, ok := strings.Cut(key, ".")
|
|
if !ok || module == "" || tool == "" {
|
|
return "", "", fmt.Errorf("%q is not <module>.<tool>, and does not start seat:<seat>.<verb>", key)
|
|
}
|
|
return "mesh.mod." + module + ".call." + tool + at + "." + token, "mesh.mod." + module + ".tool." + tool + at, nil
|
|
}
|
|
|
|
// request asks one subject once and reads its reply ({result, error, node}); a reply naming an error is
|
|
// the call's failure, not a transport one.
|
|
func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), CallTimeout)
|
|
defer cancel()
|
|
msg, err := c.nc.RequestWithContext(ctx, subject, data)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var r struct {
|
|
Result json.RawMessage `json:"result"`
|
|
Error string `json:"error"`
|
|
Node string `json:"node"`
|
|
}
|
|
if err := json.Unmarshal(msg.Data, &r); err != nil {
|
|
return nil, fmt.Errorf("%s answered something that is not a seat's answer: %w", subject, err)
|
|
}
|
|
if r.Error != "" {
|
|
return nil, errors.New(r.Error)
|
|
}
|
|
return r.Result, nil
|
|
}
|