Move the route proxy into its module and run it from the trust image, on mesh-sdk's public membership (issues 1633, 1634, 1635)
mesh/delivery-group group feat/r4d rejected: a member's own check failed
mesh/repo-check out of date: judged beside mesh-catalog#434 at c948f3a61c74, it is now at 7451e7edf4e5; judged again on a recheck or when it is merged
mesh/merge-gate group feat/r4d: composed again for the group's new heads
mesh/delivery superseded: a newer head of the same pull request
mesh/check failed: unit: i3/cmd/i3-tools.TestTheMainFileAndEveryModulesContributionLoadTogether: manifest_test.go:247: expected the launcher, clipboar…

A bundle is built only from its module's directory (ADR 0480 clause 4), so the proxy leaves
examples/ and the controller's internal/broker; the broker keeps no second copy of what the
proxy shares with mesh-sdk. The image and its Dockerfile go, and a test holds that only
build-agent declares one (clause 11).
This commit is contained in:
2026-10-22 09:53:56 +02:00
parent 3d4ad6fefb
commit 499ff5c218
313 changed files with 89380 additions and 419 deletions
@@ -0,0 +1,165 @@
// Package membership is what the mesh issues an assignment to serve and to reach (novox/hq ADR 0160), as a
// process reads it from the bus: the Membership shape, the one subject it is published on, the stream that
// holds it, and the bus address helpers a reader needs. The controller composes it and the route proxy reads
// it; both use this one definition, so there is no second copy to drift (novox/hq issue 1633).
package membership
import (
"encoding/json"
"strings"
)
// AssignmentsStream holds every assignment's membership, the newest per subject.
const AssignmentsStream = "ASSIGNMENTS"
// Subject is the one address a runtime derives for itself: where its own membership is published, from
// the two names its credential carries. Everything else is in the membership.
func Subject(node, module string) string {
return "mesh.assignment." + node + "." + module
}
// DirectGetSubject is the subject-addressed direct get of a membership from its stream: the one request an
// account may make of the stream.
func DirectGetSubject(node, module string) string {
return "$JS.API.DIRECT.GET." + AssignmentsStream + "." + Subject(node, module)
}
// Membership is one assignment's subjects: what this instance of a module on this machine serves,
// and what it may reach.
type Membership struct {
Node string `json:"node"`
Module string `json:"module"`
// Serves is every address a tool of this instance answers on. `{tool}` stands for the tool's
// own name. An address with a queue is shared with the module's other instances.
Serves []Served `json:"serves"`
// Seats is every verb of a seat this instance holds, at the subject the seat's callers use.
Seats []SeatServed `json:"seats,omitempty"`
// Emits is where an event of this module lands; `{event}` stands for the event's name.
Emits string `json:"emits"`
// Reaches is each tool this module may call, `<module>.<tool>`, to the subjects that reach it.
Reaches map[string][]string `json:"reaches,omitempty"`
// Tools is where this instance answers what it serves.
Tools string `json:"tools"`
// Receives is what this assignment is given for each requirement it receives, by requirement (novox/hq
// ADR 0167): kept as JSON because the catalogue owns the shape of a contribution.
Receives map[string]json.RawMessage `json:"receives,omitempty"`
// Mesh is every machine's address on the private network (novox/hq ADR 0167).
Mesh []string `json:"mesh,omitempty"`
// State is every bucket this module's code may reach, and whether it may write it (novox/hq ADR 0201).
State []StateIssued `json:"state,omitempty"`
// Logs is every log this module's code may reach (novox/hq ADR 0297).
Logs []LogIssued `json:"logs,omitempty"`
// SeatTraffic is what this module's code may submit, say, hear, take, ask, answer and read on seats
// that name their caller or their kind (novox/hq ADR 0259 §3).
SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"`
// Streams are this module's own streams, Hears the module streams it hears (novox/hq ADR 0363).
Streams []StreamIssued `json:"streams,omitempty"`
Hears []HeardIssued `json:"hears,omitempty"`
// Waits are the event families a session may wait on through the mesh MCP server (novox/hq ADR 0367 §2).
Waits []WaitIssued `json:"waits,omitempty"`
}
// Served is one address a tool is answered on.
type Served struct {
Subject string `json:"subject"`
Queue string `json:"queue,omitempty"`
}
// SeatServed is one verb of a held seat, where its callers ask.
type SeatServed struct {
Seat string `json:"seat"`
Verb string `json:"verb"`
Subject string `json:"subject"`
}
// StateIssued is one bucket an assignment may reach, by the name its module uses for it (novox/hq ADR 0201).
// Key, when set, is the one key it may reach (novox/hq ADR 0260): this machine's.
type StateIssued struct {
Name string `json:"name"`
Bucket string `json:"bucket"`
Writes bool `json:"writes,omitempty"`
Key string `json:"key,omitempty"`
}
// LogIssued is one log an assignment may reach, by the name its module uses for it (novox/hq ADR 0297).
type LogIssued struct {
Name string `json:"name"`
Stream string `json:"stream"`
Subject string `json:"subject"`
Writes bool `json:"writes"`
}
// Worker is one durable consumer a holder pulls a seat's work from.
type Worker struct {
Stream string
Consumer string
Filter string
}
// SeatTraffic is one module's seat traffic beyond tools. Publish and Subscribe are subject patterns, in the
// server's wildcards.
type SeatTraffic struct {
Publish []string `json:"publish,omitempty"`
Subscribe []string `json:"subscribe,omitempty"`
Answers []string `json:"answers,omitempty"`
Workers []Worker `json:"workers,omitempty"`
Records []string `json:"records,omitempty"`
Kinds []KindHeld `json:"kinds,omitempty"`
}
// KindHeld is one kind of a kinded bench and who holds it.
type KindHeld struct {
Seat string `json:"seat"`
Kind string `json:"kind"`
Module string `json:"module"`
Node string `json:"node"`
Capabilities []string `json:"capabilities,omitempty"`
}
// StreamIssued is one of a module's own streams as its membership names it (ADR 0363 §2).
type StreamIssued struct {
Stream string `json:"stream"`
Events []string `json:"events"`
Subject string `json:"subject"`
}
// HeardIssued is one module stream an assignment hears (ADR 0363 §8).
type HeardIssued struct {
Stream string `json:"stream"`
Consumer string `json:"consumer"`
}
// WaitIssued is one family as a membership issues it: the name a session asks for, and its subject.
type WaitIssued struct {
Event string `json:"event"`
Subject string `json:"subject"`
}
// CredentialIn reads the credential out of a bus address, `[scheme://]user:password@host:port`, splitting on
// the last at-sign so a password containing one stays whole, and returns the address without it.
func CredentialIn(address string) (user, password, bare string) {
at := strings.LastIndex(address, "@")
if at < 0 {
return "", "", address
}
scheme := ""
rest := address[:at]
if i := strings.Index(rest, "://"); i >= 0 {
scheme, rest = rest[:i+3], rest[i+3:]
}
user, password, _ = strings.Cut(rest, ":")
return user, password, scheme + address[at+1:]
}
// BareAddress is a bus address with any credential stripped, for something that only needs to know
// whether a server is answering there.
func BareAddress(address string) string {
_, _, bare := CredentialIn(address)
if bare == "" {
return address
}
if strings.Contains(bare, "://") {
return bare
}
return "nats://" + bare
}
@@ -0,0 +1,54 @@
// Package process is a module's own process's connection to the bus (novox/hq ADR 0436 clause 4): it reads
// 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,
// 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
import (
"encoding/json"
"fmt"
"os"
)
// Account is a process's own bus account, exactly as the controller delivers it as the module's own secret
// broker-<process id> (mesh-controller cmd/mesh-controller/modules.go issueProcess, heldProcessAccount, and
// issueWith for a module's own broker account, which also carries claims). A process account carries no
// claims at all; this type reads either file without error.
type Account struct {
URL string `json:"url"`
Fingerprint string `json:"fingerprint,omitempty"`
Node string `json:"node"`
Module string `json:"module"`
User string `json:"user"`
Password string `json:"password"`
// Claims rides on a module's own broker account only; a process holds no seat, so its own account
// carries none (novox/hq ADR 0436 §2, §3). Kept raw: this package never reads it.
Claims json.RawMessage `json:"claims,omitempty"`
}
// InboxPrefix is this account's own inbox, and nothing wider: `_INBOX.<user>`.
func (a Account) InboxPrefix() string { return "_INBOX." + a.User }
// ReadAccount reads and validates the account file the controller delivers: it refuses a file missing url,
// user or password, naming the field — never the password, which does not appear in an error.
func ReadAccount(path string) (Account, error) {
data, err := os.ReadFile(path)
if err != nil {
return Account{}, fmt.Errorf("reading the process account at %s: %w", path, err)
}
var a Account
if err := json.Unmarshal(data, &a); err != nil {
return Account{}, fmt.Errorf("the process account at %s is not JSON: %w", path, err)
}
for _, f := range []struct{ name, value string }{
{"url", a.URL}, {"user", a.User}, {"password", a.Password},
} {
if f.value == "" {
return Account{}, fmt.Errorf("the process account at %s has no %s", path, f.name)
}
}
return a, nil
}
@@ -0,0 +1,68 @@
package process
import (
"context"
"encoding/json"
"fmt"
"strings"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// Ask reads the router's record of one of this process's module's own asks, under its own module and no
// other's (novox/hq ADR 0436 clause 4, ADR 0259 §3): by direct get on bucket, the same subject the tool
// runner reads (mesh-tools node-tools/internal/bus/seat.go ~300-345):
//
// $JS.API.DIRECT.GET.KV_<bucket>.$KV.<bucket>.<module>.<id>
//
// bucket is the router's records bucket as the caller's own manifest names it (operator-channel's is
// messenger_asks in the controller's own test fixtures) — never hard-coded to one bucket here, since a
// process may read records of more than one seat's router. A missing, deleted or purged key is nil, not
// an error.
func (c *Conn) Ask(bucket, id string) (*stdio.StateEntry, error) {
if err := checkAskID(id); err != nil {
return nil, err
}
subject := "$JS.API.DIRECT.GET.KV_" + bucket + ".$KV." + bucket + "." + c.account.Module + "." + id
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
defer cancel()
reply, err := c.nc.RequestWithContext(ctx, subject, nil)
if err != nil {
return nil, fmt.Errorf("reading %s's ask %s from %s: %w", c.account.Module, id, bucket, err)
}
if status := reply.Header.Get("Status"); status == "404" {
return nil, nil
} else if status != "" {
return nil, fmt.Errorf("the record %s.%s could not be read: %s %s", c.account.Module, id, status,
reply.Header.Get("Description"))
}
if op := reply.Header.Get("KV-Operation"); op == "DEL" || op == "PURGE" {
return nil, nil
}
var revision uint64
if _, err := fmt.Sscan(reply.Header.Get("Nats-Sequence"), &revision); err != nil {
return nil, fmt.Errorf("the record %s.%s carries no readable revision: %w", c.account.Module, id, err)
}
return &stdio.StateEntry{Key: c.account.Module + "." + id, Value: askValueOf(reply.Data), Revision: revision}, nil
}
// checkAskID refuses an id holding a dot or a wildcard before anything is sent: either would name a
// different key under the direct-get subject's own dot-joined path, never this one (novox/hq issue 1189
// AC2).
func checkAskID(id string) error {
if id == "" || strings.ContainsAny(id, ".*>") {
return fmt.Errorf("%q is not an ask id: an ask id holds no dot and no wildcard, naming one key and "+
"no other", id)
}
return nil
}
// askValueOf is a record's raw bytes as JSON, quoting them when they are not already valid JSON
// (mesh-tools node-tools/internal/bus/state.go:351-357, valueOf).
func askValueOf(raw []byte) json.RawMessage {
if len(raw) == 0 || !json.Valid(raw) {
b, _ := json.Marshal(string(raw))
return b
}
return json.RawMessage(append([]byte(nil), raw...))
}
@@ -0,0 +1,122 @@
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
}
@@ -0,0 +1,164 @@
package process
import (
"context"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"encoding/hex"
"errors"
"fmt"
"regexp"
"sync"
"github.com/nats-io/nats.go"
)
// 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.
type Conn struct {
nc *nats.Conn
account Account
mu sync.Mutex
next uint64
pending map[string]map[uint64]context.CancelCauseFunc
}
// errPermissionRefused marks a context cancelled because the bus refused this account's publish of the
// subject a pending call was waiting an answer on: a denied publish is refused only as an async protocol
// error with no reply (mesh-tools node-tools/internal/bus/caller.go publishRefused), so without this a
// call would wait out its whole timeout instead of failing at once.
var errPermissionRefused = errors.New("the bus refused this account's publish")
// refusedPublish reads the subject out of the bus's "-ERR 'Permissions Violation for Publish to
// "<subject>"'" (mesh-tools node-tools/internal/bus/caller.go's own regexp).
var refusedPublish = regexp.MustCompile(`(?i)Publish to "(\S+)"`)
// Connect dials the bus with account: pinned to its fingerprint when it carries one (refused with any
// other certificate), authenticated as its user and password, on the inbox prefix _INBOX.<user> and no
// other, reconnecting without limit as the tool runner does (novox/hq issue 1192: a process outlives a
// bus restart, never giving up the way the client library's own default of 60 attempts would). Ported
// from mesh-controller internal/broker/jetstream.go (DialPinned, PinnedToFingerprint), which now uses this one since this
// package may not import the controller.
//
// An ErrorHandler among opts is chained behind this package's own, which watches every async error for a
// permission violation refusing a publish this connection is waiting an answer to, and cancels that wait
// at once.
func Connect(account Account, opts ...nats.Option) (*Conn, error) {
c := &Conn{account: account, pending: map[string]map[uint64]context.CancelCauseFunc{}}
nc, err := nats.Connect(account.URL, c.options(opts)...)
if err != nil {
return nil, fmt.Errorf("connecting the process account %s: %w", account.User, err)
}
c.nc = nc
return c, nil
}
// options is the full option list Connect dials with, apart from Connect so a unit test can check what
// it carries without a live bus: this package's own (the inbox prefix, the credentials, the pin,
// reconnecting without limit, and an ErrorHandler chaining any the caller passed), then opts, then this
// package's own ErrorHandler again last, so it is the one AsyncErrorCB nats.Connect keeps however opts
// ordered theirs.
func (c *Conn) options(opts []nats.Option) []nats.Option {
// opts are functions over a *nats.Options; applied to a scratch value, this reads off a caller's own
// ErrorHandler (if any) to chain, without needing nats.Connect's own URL/Servers parsing here.
var scratch nats.Options
for _, opt := range opts {
if opt != nil {
_ = opt(&scratch)
}
}
chained := scratch.AsyncErrorCB
all := []nats.Option{
nats.CustomInboxPrefix(c.account.InboxPrefix()),
nats.UserInfo(c.account.User, c.account.Password),
nats.MaxReconnects(-1),
}
if c.account.Fingerprint != "" {
all = append(all, nats.Secure(PinnedToFingerprint(c.account.Fingerprint)))
}
all = append(all, opts...)
return append(all, nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) {
c.publishRefused(err)
if chained != nil {
chained(nc, sub, err)
}
}))
}
// Close ends the connection.
func (c *Conn) Close() { c.nc.Close() }
// withRefusal derives a context from ctx that is also cancelled, with errPermissionRefused as its cause,
// the moment the bus refuses this account's publish of subject — before ctx's own deadline, which a
// denied publish would otherwise run out, refused only as an async error with no reply. Every call that
// requests on the bus registers the subject it is about to publish to here first.
func (c *Conn) withRefusal(ctx context.Context, subject string) (context.Context, func()) {
derived, cancel := context.WithCancelCause(ctx)
c.mu.Lock()
c.next++
id := c.next
if c.pending[subject] == nil {
c.pending[subject] = map[uint64]context.CancelCauseFunc{}
}
c.pending[subject][id] = cancel
c.mu.Unlock()
return derived, func() {
c.mu.Lock()
if list := c.pending[subject]; list != nil {
delete(list, id)
if len(list) == 0 {
delete(c.pending, subject)
}
}
c.mu.Unlock()
cancel(nil)
}
}
// publishRefused ends every call waiting on the subject the bus's error names, with errPermissionRefused
// as the cause (mesh-tools node-tools/internal/bus/caller.go publishRefused): true when there was one
// waiting.
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.mu.Lock()
cancels := c.pending[m[1]]
delete(c.pending, m[1])
c.mu.Unlock()
for _, cancel := range cancels {
cancel(errPermissionRefused)
}
return len(cancels) > 0
}
// PinnedToFingerprint accepts exactly the certificate with this SHA-256 and no other — the leaf alone, its
// name and chain deliberately not consulted, as mesh-controller's own pin does (jetstream.go:116-132).
// PinnedToFingerprint is public so that a process dialling the bus on its own terms, and the controller,
// share this one pin (novox/hq issue 1633).
func PinnedToFingerprint(want string) *tls.Config {
return &tls.Config{
InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
MinVersion: tls.VersionTLS12,
VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
if len(rawCerts) == 0 {
return errors.New("the bus presented no certificate")
}
sum := sha256.Sum256(rawCerts[0])
got := "sha256:" + hex.EncodeToString(sum[:])
if got != want {
return fmt.Errorf("the bus presented a certificate this mesh does not know (%s…), expected %s…",
got[:23], want[:23])
}
return nil
},
}
}
@@ -0,0 +1,69 @@
package process
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// Emit publishes an event as this process's module (novox/hq ADR 0436 clause 4), under its module's own
// namespace (mesh.mod.<module>.event.<type>, the same subject the module's own principal would publish it
// on), with the SDK's own envelope and headers (stdio.go:498-509). A chosen message id
// (EmitOptions.MessageID) is sent as x-event-id, and as the bus's own dedup id (Nats-Msg-Id), so a second
// publish inside the stream's duplicate window is dropped; empty is a random one, as stdio.Emit on main.
func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error {
data, err := json.Marshal(body)
if err != nil {
return err
}
id := ""
for _, o := range opts {
if o.MessageID != "" {
id = o.MessageID
}
}
if id == "" {
id = eventID()
}
msg := nats.NewMsg("mesh.mod." + c.account.Module + ".event." + eventType)
msg.Header.Set("x-event-id", id)
msg.Header.Set("x-source", c.account.Module)
msg.Header.Set("x-node", c.account.Node)
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
msg.Header.Set("content-type", "application/json")
for _, o := range opts {
if o.CausationID != "" {
msg.Header.Set("x-causation-id", o.CausationID)
}
if o.Schema != "" {
msg.Header.Set("x-schema", o.Schema)
}
}
msg.Data = data
js, err := jetstream.New(c.nc)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
defer cancel()
_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
return err
}
// eventID is a random UUID, version 4, as stdio's own eventID (go/stdio.go:515-522) — duplicated rather
// than imported, since stdio's is unexported and this package touches none of stdio's bundle machinery.
func eventID() string {
var b [16]byte
_, _ = rand.Read(b[:])
b[6] = (b[6] & 0x0f) | 0x40
b[8] = (b[8] & 0x3f) | 0x80
h := hex.EncodeToString(b[:])
return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32]
}
@@ -0,0 +1,53 @@
package process
import (
"encoding/json"
"fmt"
"time"
"github.com/nats-io/nats.go"
"git.novox.be/novox/mesh-sdk/go/membership"
)
// FollowMembership reads this account's own membership and follows it (novox/hq ADR 0160): apply is called
// with every membership the mesh issues this account's node and module, the one already issued first and
// each later one as it is published, so a change reaches a running process without a restart. A body that is
// not a membership is handed to onBad (nil to ignore it) and skipped.
//
// The subject is followed before it is read, so an issue landing between the two is applied, not missed. The
// read is the subject-addressed direct get, the one request this account may make of the stream; no
// membership issued yet, or the read failing, is not an error: the process follows the subject and is told
// when one arrives. The returned error is a refusal to follow at all.
func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(error)) error {
if c.account.Node == "" || c.account.Module == "" {
return fmt.Errorf("the account names no node or module, so it has no membership")
}
bad := func(err error) {
if onBad != nil {
onBad(err)
}
}
take := func(body []byte) {
var issued membership.Membership
if err := json.Unmarshal(body, &issued); err != nil {
bad(fmt.Errorf("a membership arrived that is not one: %w", err))
return
}
apply(issued)
}
subject := membership.Subject(c.account.Node, c.account.Module)
if _, err := c.nc.Subscribe(subject, func(m *nats.Msg) { take(m.Data) }); err != nil {
return fmt.Errorf("cannot follow %s: %w", subject, err)
}
got, err := c.nc.Request(membership.DirectGetSubject(c.account.Node, c.account.Module), nil, 5*time.Second)
switch {
case err != nil:
bad(fmt.Errorf("cannot read the membership issued on %s yet (%v); following it", subject, err))
case got.Header.Get("Status") != "" || len(got.Data) == 0:
// None issued yet: the follow above delivers it.
default:
take(got.Data)
}
return nil
}
@@ -0,0 +1,127 @@
package process
import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/nats-io/nats.go/jetstream"
)
// Message is one piece of work this process's queue consumer delivered, with how many times the bus has
// delivered it. Nothing acts on its behalf: the handler decides its fate exactly once, by calling Ack,
// NakWithDelay or Term (mesh-tools node-tools/internal/bus/seat.go's seat work Consume, with the same
// three outcomes, offered here explicitly instead of by a handler's return value, since the caller
// chooses its own delay before a message comes back).
type Message struct {
Subject string
Body []byte
Headers map[string]string
// Delivered is how many times the bus has handed this message over, this time included.
Delivered uint64
ack func() error
nak func(time.Duration) error
term func() error
}
// Ack acknowledges the message: the bus never delivers it again.
func (m *Message) Ack() error { return m.ack() }
// NakWithDelay asks for the message to be delivered again no sooner than delay, its delivery count
// raised by one.
func (m *Message) NakWithDelay(delay time.Duration) error { return m.nak(delay) }
// Term terminates the message: the bus never delivers it again, and does not count it a dead letter the
// way running out of deliveries does.
func (m *Message) Term() error { return m.term() }
// ProcessID is this account's own process id, the last segment of its user name <node>.<module>.<id>
// (mesh-controller internal/broker, Principal.Username for KindProcess).
func (a Account) ProcessID() string {
prefix := a.Node + "." + a.Module + "."
if !strings.HasPrefix(a.User, prefix) {
return ""
}
return strings.TrimPrefix(a.User, prefix)
}
// QueueConsumerName is the one durable consumer a process's bus.queues names it on its work-queue stream
// (novox/hq ADR 0436 clauses 1, 2, 6; mesh-controller internal/broker/modstreams.go QueueConsumer):
// <node>_<module>_<process id>.
func (a Account) QueueConsumerName() string { return a.Node + "_" + a.Module + "_" + a.ProcessID() }
// errNoMessage is one pull that fetched nothing within its wait: not an error to the caller, just a
// reason to pull again.
var errNoMessage = errors.New("no message this pull")
// Consume binds this process's one durable consumer on stream (QueueConsumerName) and pulls from it, one
// message at a time, handing each to handle until ctx ends: it then stops pulling and returns ctx.Err(),
// never acknowledging a message still in handle's hands — only handle decides a message's fate, by
// calling Ack, NakWithDelay or Term on it.
func (c *Conn) Consume(ctx context.Context, stream string, handle func(*Message)) error {
js, err := jetstream.New(c.nc)
if err != nil {
return err
}
name := c.account.QueueConsumerName()
cons, err := js.Consumer(ctx, stream, name)
if err != nil {
return fmt.Errorf("binding %s's queue consumer %s on %s: %w", c.account.Module, name, stream, err)
}
return runQueue(ctx, fetchOne(cons), handle)
}
// fetchOne is one pull of at most one message, non-blocking past ctx: Fetch itself does not take a
// context, so this selects on its message channel and ctx together, and returns promptly once ctx ends
// even while a fetch is still outstanding.
func fetchOne(cons jetstream.Consumer) func(ctx context.Context) (*Message, error) {
return func(ctx context.Context) (*Message, error) {
batch, err := cons.Fetch(1, jetstream.FetchMaxWait(5*time.Second))
if err != nil {
return nil, err
}
select {
case msg, ok := <-batch.Messages():
if !ok {
return nil, errNoMessage
}
headers := map[string]string{}
for k := range msg.Headers() {
headers[k] = msg.Headers().Get(k)
}
var delivered uint64 = 1
if meta, err := msg.Metadata(); err == nil && meta != nil {
delivered = meta.NumDelivered
}
return &Message{Subject: msg.Subject(), Body: append([]byte(nil), msg.Data()...), Headers: headers,
Delivered: delivered, ack: msg.Ack, nak: msg.NakWithDelay, term: msg.Term}, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}
}
// runQueue is Consume's loop, apart from the bus so a unit test can drive it with a fake next
// (novox/hq issue 1190 AC4): it checks ctx before every pull, and stops as soon as next or the check
// says ctx has ended, without touching the message past handing it to handle.
func runQueue(ctx context.Context, next func(context.Context) (*Message, error), handle func(*Message)) error {
for {
if err := ctx.Err(); err != nil {
return err
}
msg, err := next(ctx)
if err != nil {
if ctx.Err() != nil {
return ctx.Err()
}
if errors.Is(err, errNoMessage) {
continue
}
return err
}
handle(msg)
}
}
@@ -0,0 +1,130 @@
package process
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
stdio "git.novox.be/novox/mesh-sdk/go"
)
// stateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a denied publish is
// caught at once by Conn's own ErrorHandler (conn.go withRefusal/publishRefused), never waited out, so
// this stays the ordinary request bound and is not exposed for a test to shrink.
const stateTimeout = 30 * time.Second
// State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436
// clause 4), in the bucket <module>_<name> (mesh-controller internal/broker/state.go:49) — only as the
// account's grants allow. It satisfies go/asker's Store.
type State struct {
Module string
Name string
// kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a
// unit test, a fake that never reaches a bus.
kv func(ctx context.Context) (jetstream.KeyValue, error)
}
// bucket is this state's name on the bus.
func (s State) bucket() string { return s.Module + "_" + s.Name }
// State is the named state this connection's module keeps, bound lazily to its bucket on each call so a
// bucket the account is not granted fails at the call that needed it, naming the bucket. The bind itself
// (js.KeyValue) is what asks $JS.API.STREAM.INFO.KV_<bucket>, so it is what withRefusal watches: a bucket
// not granted is refused there, promptly, as nats.ErrPermissionViolation.
func (c *Conn) State(name string) State {
bucket := c.account.Module + "_" + name
return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
js, err := jetstream.New(c.nc)
if err != nil {
return nil, err
}
subject := "$JS.API.STREAM.INFO.KV_" + bucket
ctx, stop := c.withRefusal(ctx, subject)
defer stop()
kv, err := js.KeyValue(ctx, bucket)
if err != nil {
if errors.Is(context.Cause(ctx), errPermissionRefused) {
return nil, fmt.Errorf("%w: publishing %s", nats.ErrPermissionViolation, subject)
}
return nil, err
}
return kv, nil
}}
}
func (s State) store(ctx context.Context) (jetstream.KeyValue, error) {
kv, err := s.kv(ctx)
if err != nil {
return nil, fmt.Errorf("%s's state %s (bucket %s): %w", s.Module, s.Name, s.bucket(), err)
}
return kv, nil
}
// Get is a key's current value, or nil when it has none.
func (s State) Get(key string) (*stdio.StateEntry, error) {
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
defer cancel()
kv, err := s.store(ctx)
if err != nil {
return nil, err
}
entry, err := kv.Get(ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
}
return &stdio.StateEntry{Key: key, Value: json.RawMessage(entry.Value()), Revision: entry.Revision()}, nil
}
// Create writes a key only when it has none, and returns the revision; a key that has one is refused as
// stdio.ErrChanged.
func (s State) Create(key string, value any) (uint64, error) {
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
defer cancel()
kv, err := s.store(ctx)
if err != nil {
return 0, err
}
data, err := json.Marshal(value)
if err != nil {
return 0, err
}
rev, err := kv.Create(ctx, key, data)
if errors.Is(err, jetstream.ErrKeyExists) {
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
}
if err != nil {
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
}
return rev, nil
}
// Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile
// is refused as stdio.ErrChanged.
func (s State) Update(key string, value any, revision uint64) (uint64, error) {
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
defer cancel()
kv, err := s.store(ctx)
if err != nil {
return 0, err
}
data, err := json.Marshal(value)
if err != nil {
return 0, err
}
rev, err := kv.Update(ctx, key, data, revision)
if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
}
if err != nil {
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
}
return rev, nil
}