diff --git a/modules/route-proxy/go.mod b/modules/route-proxy/go.mod index 178b9e9..a1f29bc 100644 --- a/modules/route-proxy/go.mod +++ b/modules/route-proxy/go.mod @@ -3,7 +3,7 @@ module route-proxy go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022075101-c61212eba77c + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 ) diff --git a/modules/route-proxy/go.sum b/modules/route-proxy/go.sum index 1db4a20..b72ac07 100644 --- a/modules/route-proxy/go.sum +++ b/modules/route-proxy/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022075101-c61212eba77c h1:YdSbkI42rlVmClybH8BS717ut0vouKIMavO691fpdps= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022075101-c61212eba77c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 h1:dtiIJzrX7ey1wmsdzapiFgCKDllCph7/vxfUp1ib/t8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/announce/announce.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/announce/announce.go new file mode 100644 index 0000000..37dfd21 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/announce/announce.go @@ -0,0 +1,126 @@ +// Package announce is how a module says on the bus which states it keeps and which of them others may read (novox/hq +// ADR 0495 clauses 2 and 3; issue 1790): the subject it says it on, the value it says, and where the controller keeps +// the last one said by each module. What a module announced is the truth of what it keeps; a reader is granted a read +// from it, never from a list or from the owner's manifest. +// +// A module announces from its code (process.Conn.Keeps, stdio.Keeps, the TypeScript SDK's keeps), never from its +// manifest, which only the core reads (clause 1). A module that keeps no state on the bus announces nothing. +package announce + +import ( + "encoding/json" + "fmt" + "regexp" + "sort" +) + +// Version is the version of the value this package writes. +const Version = 1 + +// Event is the event type of an announcement, under the module's own namespace. +const Event = "states" + +// Subject is where module announces what it keeps: its own event subject, mesh.mod..event.states, which its +// account may already publish to and the events stream keeps. +func Subject(module string) string { return "mesh.mod." + module + ".event." + Event } + +// SubjectAll is every module's announcement, what the controller consumes. +const SubjectAll = "mesh.mod.*.event." + Event + +// KeptBucket is the controller's bucket of the last announcement of each module, by Key. +const KeptBucket = "mesh-controller_states-announced" + +var name = regexp.MustCompile(`^[a-z0-9][a-z0-9-]*$`) + +// Key is module's key in KeptBucket: the module itself, which must be one plain name, since a module's name is part of +// its buckets' names (catalogue.StateProblems). +func Key(module string) (string, error) { + if !name.MatchString(module) { + return "", fmt.Errorf("announce: %q is not a module that keeps state: lower-case letters, digits and hyphens", module) + } + return module, nil +} + +// State is one state a module keeps, by its local name, and whether others may read it. Others never write it: a +// state another module writes is the owner's own code's business. +type State struct { + Name string `json:"name"` + Readable bool `json:"readable,omitempty"` +} + +// States is an announcement: the module, the machine whose instance said it, and every state the module keeps, sorted +// by name. An empty list says the module keeps nothing. +type States struct { + V int `json:"v"` + Module string `json:"module"` + Node string `json:"node,omitempty"` + States []State `json:"states"` +} + +// Keeps says whether the module announced keeping the state. +func (s States) Keeps(state string) bool { + for _, k := range s.States { + if k.Name == state { + return true + } + } + return false +} + +// Readable says whether the module announced the state readable by others. +func (s States) Readable(state string) bool { + for _, k := range s.States { + if k.Name == state { + return k.Readable + } + } + return false +} + +func (s *States) check() error { + if s.Module == "" { + return fmt.Errorf("announce: an announcement names no module") + } + if _, err := Key(s.Module); err != nil { + return err + } + seen := map[string]bool{} + for _, k := range s.States { + if !name.MatchString(k.Name) { + return fmt.Errorf("announce: %s announces the state %q, and a state is named locally: lower-case letters, digits and hyphens", s.Module, k.Name) + } + if seen[k.Name] { + return fmt.Errorf("announce: %s announces the state %q twice", s.Module, k.Name) + } + seen[k.Name] = true + } + return nil +} + +// Encode is s as JSON at Version, its states sorted; it refuses a module or a state no bucket could be named by and a +// state announced twice, naming it. +func Encode(s States) ([]byte, error) { + if err := s.check(); err != nil { + return nil, err + } + s.V = Version + s.States = append([]State{}, s.States...) + sort.Slice(s.States, func(i, j int) bool { return s.States[i].Name < s.States[j].Name }) + return json.Marshal(s) +} + +// Decode reads an announcement; it refuses another version and what Encode refuses. +func Decode(data []byte) (States, error) { + var s States + if err := json.Unmarshal(data, &s); err != nil { + return States{}, fmt.Errorf("announce: the announcement is not JSON: %w", err) + } + if s.V != Version { + return States{}, fmt.Errorf("announce: the announcement is version %d, and this reads version %d", s.V, Version) + } + if err := s.check(); err != nil { + return States{}, err + } + sort.Slice(s.States, func(i, j int) bool { return s.States[i].Name < s.States[j].Name }) + return s, nil +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/keeps.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/keeps.go new file mode 100644 index 0000000..cffdf9c --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/keeps.go @@ -0,0 +1,80 @@ +package stdio + +import ( + "encoding/json" + "fmt" + "os" + "strings" + "sync" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// What this bundle said it keeps (novox/hq ADR 0495 clause 2; issue 1790). +var ( + keepsMu sync.Mutex + kept map[string]bool // nil until Keeps: no state is refused then + announced []byte // the announcement, published once the bundle serves +) + +// Keeps says on the bus, as this bundle's module, every state it keeps and which of them others may read (novox/hq +// ADR 0495 clause 2): what a reader is granted is what the owner announced (clause 3). Call it before Serve; the +// announcement is published through the runtime once the bundle serves, or at once when it already does. From then +// on a state of the module's own that it did not announce is refused at the call, naming it, so code and announcement +// cannot drift (clause 5). It refuses a state no bucket could be named by and a state given twice. +func Keeps(states ...announce.State) error { + module := os.Getenv("MESH_MODULE") + if module == "" { + module = os.Getenv(ServedAs) + } + data, err := announce.Encode(announce.States{Module: module, Node: os.Getenv("MESH_NODE"), States: states}) + if err != nil { + return err + } + names := map[string]bool{} + for _, s := range states { + names[s.Name] = true + } + keepsMu.Lock() + kept, announced = names, data + keepsMu.Unlock() + currentMu.Lock() + serving := current != nil + currentMu.Unlock() + if serving { + return announce1() + } + return nil +} + +// announce1 publishes the announcement Keeps made, if any. +func announce1() error { + keepsMu.Lock() + data := announced + keepsMu.Unlock() + if data == nil { + return nil + } + return Emit(announce.Event, json.RawMessage(data)) +} + +// refusedUnannounced is the error of a call on a state of the module's own that Keeps did not name, or nil. A read of +// another module's state (`.`) is the grant's business. +func refusedUnannounced(name string) error { + if strings.Contains(name, ".") { + return nil + } + keepsMu.Lock() + defer keepsMu.Unlock() + if kept != nil && !kept[name] { + return fmt.Errorf("the state %q is not among the states this module announced it keeps (stdio.Keeps)", name) + } + return nil +} + +// forgetKeeps is for a test: nothing announced. +func forgetKeeps() { + keepsMu.Lock() + kept, announced = nil, nil + keepsMu.Unlock() +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh index 92e9d9c..0058732 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/merge-check.sh @@ -10,13 +10,17 @@ # 1. the unit level: `go test ./...` untagged, no bus or other process, timed and held to a bound # (unit_bound below); over it the check fails and names the slowest tests (issue 895); # 2. the integration level: the whole suite with `-tags integration`, each test against a bus of its -# own (internal/testbus, which issue 1171 brings with the first test that needs a bus) so nothing is +# own (testbus, which issue 1171 brings with the first test that needs a bus) so nothing is # serialised. With no tagged test yet, the level runs the unit tests again and says their count. # # **A test that starts a bus or another process carries `//go:build integration`** as the first line of # its file; a file that held both kinds is two files (`_integration_test.go` beside `_test.go`). # Each level says its test count and its time. # +# **A test that needs bubblewrap carries `//go:build integration && isolation`** (issue 1309's proof that a run step +# is shut off): this script vets it but does not run it, since the toolchain it runs in has no bubblewrap; the checker +# runs the `isolation` tag on the build nodes, which have it (R4e). +# # No dependency is committed (no vendor directory: .gitignore; the operator, 2026-10-19; issue 1304). Every dependency # is resolved by the version and checksum go.sum names (novox/hq ADR 0405 §5), from its public source: this binding # requires nothing of the mesh's own forge. `go mod verify` fails by module name when a downloaded dependency does not @@ -98,6 +102,7 @@ fi go vet ./... # The files tagged `integration` are vetted too, or nothing would read them. go vet -tags integration ./... +go vet -tags integration,isolation ./... # NOT RACE-CHECKED here: the SDK's suite is small and carries no store or other shared state between # packages, so the race detector is not required of this check (novox/hq issue 1175); a module that needs @@ -117,7 +122,7 @@ if [ "$status" -ne 0 ]; then fi judge_unit_run "$work/unit.out" "$unit_bound" -# The integration level: the whole suite with the tag, each test against its own bus (internal/testbus). +# The integration level: the whole suite with the tag, each test against its own bus (testbus). integration_start=$(date +%s) status=0 go test -tags integration -count=1 -v ./... >"$work/integration.out" 2>&1 || status=$? diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/conn.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/conn.go index 58efb40..ca46922 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/conn.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/conn.go @@ -24,6 +24,12 @@ type Conn struct { mu sync.Mutex next uint64 pending map[string]map[uint64]context.CancelCauseFunc + + // What Keeps announced (keeps.go): the states' names (nil before Keeps), the announcement, and whether a + // reconnect announces it again. + kept map[string]bool + announced []byte + reannounces bool } // errPermissionRefused marks a context cancelled because the bus refused this account's publish of the diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/keeps.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/keeps.go new file mode 100644 index 0000000..22a1df7 --- /dev/null +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/keeps.go @@ -0,0 +1,71 @@ +package process + +import ( + "encoding/json" + "fmt" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/announce" +) + +// Keeps says on the bus, as this process's module, every state it keeps and which of them others may read (novox/hq +// ADR 0495 clause 2; issue 1790), on announce.Subject; what a reader is granted is what the owner announced (clause 3). +// It publishes at once and again after every reconnect, so the controller hears it whenever it missed it. From then on +// a state of the module that it did not announce is refused at the call, naming it, so code and announcement cannot +// drift (clause 5). It refuses a state no bucket could be named by and a state given twice. +func (c *Conn) Keeps(states ...announce.State) error { + if err := c.remember(states); err != nil { + return err + } + c.mu.Lock() + first := !c.reannounces + c.reannounces = true + c.mu.Unlock() + if first { + before := c.nc.Opts.ReconnectedCB + c.nc.SetReconnectHandler(func(nc *nats.Conn) { + if before != nil { + before(nc) + } + _ = c.publishKeeps() + }) + } + return c.publishKeeps() +} + +// remember makes the announcement of states and binds only them from now on. +func (c *Conn) remember(states []announce.State) error { + data, err := announce.Encode(announce.States{Module: c.account.Module, Node: c.account.Node, States: states}) + if err != nil { + return err + } + names := map[string]bool{} + for _, s := range states { + names[s.Name] = true + } + c.mu.Lock() + c.kept, c.announced = names, data + c.mu.Unlock() + return nil +} + +func (c *Conn) publishKeeps() error { + c.mu.Lock() + data := c.announced + c.mu.Unlock() + if err := c.Emit(announce.Event, json.RawMessage(data)); err != nil { + return fmt.Errorf("announcing what %s keeps: %w", c.account.Module, err) + } + return nil +} + +// refusedUnannounced is the error of a bind of a state Keeps did not name, or nil (also before Keeps). +func (c *Conn) refusedUnannounced(name string) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.kept != nil && !c.kept[name] { + return fmt.Errorf("the state %q is not among the states %s announced it keeps (Conn.Keeps)", name, c.account.Module) + } + return nil +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go index 43221fa..777a9fd 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/process/state.go @@ -5,6 +5,8 @@ import ( "encoding/json" "errors" "fmt" + "sort" + "strings" "time" "github.com/nats-io/nats.go" @@ -27,6 +29,8 @@ type State struct { // 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) + // closed says the connection is gone, so a watch ends; nil in a unit test. + closed func() bool } // bucket is this state's name on the bus. @@ -38,7 +42,10 @@ func (s State) bucket() string { return s.Module + "_" + s.Name } // 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) { + return State{Module: c.account.Module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) { + if err := c.refusedUnannounced(name); err != nil { + return nil, err + } js, err := jetstream.New(c.nc) if err != nil { return nil, err @@ -128,3 +135,148 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { } return rev, nil } + +// Put writes a key whole, whatever it held, and returns the revision written. +func (s State) Put(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.Put(ctx, key, data) + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} + +// Keys is every key with a current value, sorted; none is an empty list. +func (s State) Keys() ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + if err != nil { + return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err) + } + sort.Strings(keys) + return keys, nil +} + +// watchPattern is a key narrowing as the bus names it: `*` stays one name, `**` is the rest, empty is every key. +func watchPattern(key string) string { + if key == "" { + return jetstream.AllKeys + } + tokens := strings.Split(key, ".") + for i, t := range tokens { + if t == "**" { + tokens[i] = ">" + } + } + return strings.Join(tokens, ".") +} + +// Watch hands handle every current value — none that is deleted, each with Current set — and then every +// change in order, and returns once the current values have all been handled. key narrows it: a key, with +// `*` for one name and `**` for the rest; empty is every key. The changes go on for the life of the +// connection; an error from handle, or the connection closing, ends them. Before the current values are +// handled, either is returned as the error. +func (s State) Watch(key string, handle func(stdio.StateChange) error) error { + // the watch outlives this call, so its context is not the call's timeout: the connection ends it. + ctx, cancel := context.WithCancel(context.Background()) + bind, stopBind := context.WithTimeout(ctx, stateTimeout) + kv, err := s.store(bind) + stopBind() + if err != nil { + cancel() + return err + } + w, err := kv.Watch(ctx, watchPattern(key)) + if err != nil { + cancel() + return fmt.Errorf("%s's state %s, watching %q: %w", s.Module, s.Name, key, err) + } + end := func() { w.Stop(); cancel() } + next := func() (jetstream.KeyValueEntry, bool, error) { + tick := time.NewTicker(250 * time.Millisecond) + defer tick.Stop() + for { + select { + case e, open := <-w.Updates(): + if !open { + return nil, false, fmt.Errorf("%s's state %s: the watch ended", s.Module, s.Name) + } + return e, true, nil + case <-tick.C: + if s.closed != nil && s.closed() { + return nil, false, fmt.Errorf("%s's state %s: the connection closed", s.Module, s.Name) + } + } + } + } + deliver := func(e jetstream.KeyValueEntry, current bool) error { + c := stdio.StateChange{State: s.Name, Key: e.Key(), Op: "put", Revision: e.Revision(), Current: current} + if e.Operation() == jetstream.KeyValuePut { + c.Value = json.RawMessage(e.Value()) + } else { + c.Op = "delete" + } + return handle(c) + } + for { + e, _, err := next() + if err != nil { + end() + return err + } + if e == nil { + break + } + if e.Operation() != jetstream.KeyValuePut { + continue + } + if err := deliver(e, true); err != nil { + end() + return err + } + } + go func() { + defer end() + for { + e, _, err := next() + if err != nil || e == nil { + return + } + if deliver(e, false) != nil { + return + } + } + }() + return nil +} + +// Delete removes a key. A key that was not there is not an error. +func (s State) Delete(key string) error { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return err + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return nil +} diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go index 70e1b35..56490b6 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/state.go @@ -50,6 +50,9 @@ func State(name string) KeptState { return KeptState{name: name} } const stateTimeout = 30 * time.Second func (s KeptState) ask(verb string, params map[string]any) (json.RawMessage, error) { + if err := refusedUnannounced(s.name); err != nil { + return nil, err + } c, err := live() if err != nil { return nil, err diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go index a61f4de..a072f40 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/stdio.go @@ -180,6 +180,9 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { currentMu.Lock() current = c currentMu.Unlock() + // What the bundle said it keeps is announced once it serves: the runtime answers the publish on this channel, so + // it is asked from beside the loop that reads the answer. + go func() { _ = announce1() }() scanner := bufio.NewScanner(in) scanner.Buffer(make([]byte, 1<<20), 1<<24) for scanner.Scan() { diff --git a/modules/route-proxy/vendor/modules.txt b/modules/route-proxy/vendor/modules.txt index 909cdc3..c65612e 100644 --- a/modules/route-proxy/vendor/modules.txt +++ b/modules/route-proxy/vendor/modules.txt @@ -1,6 +1,7 @@ -# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022075101-c61212eba77c +# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022183749-875c680d9876 ## explicit; go 1.26.0 git.novox.be/novox/mesh-sdk/go +git.novox.be/novox/mesh-sdk/go/announce git.novox.be/novox/mesh-sdk/go/membership git.novox.be/novox/mesh-sdk/go/process git.novox.be/novox/mesh-sdk/go/provider