Merge main (upgrade safety's declared packages, issue 1840) and pin the R4e mesh-sdk member's final head 875c680d, re-vendored here and in modules/route-proxy; the controller announces declared-packages readable (issue 1790, D1; E)

2026-10-22 20:44:59 +02:00
parent bc80683b9e 96441ea28a
commit a0b2d98002
34 changed files with 1133 additions and 28 deletions
@@ -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.<module>.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
+}
@@ -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 (`<module>.<name>`) 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()
+}
@@ -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 (`<name>_integration_test.go` beside `<name>_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=$?
@@ -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
@@ -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
+}
@@ -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
+}
@@ -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
@@ -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() {
@@ -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