node-tools/go.mod

@@ -8,7 +8,7 @@ require (
 )
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102112556-abf5eace7791
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128
 	github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect
 	github.com/google/go-tpm v0.9.8 // indirect
 	github.com/klauspost/compress v1.20.0 // indirect

node-tools/go.sum

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102112556-abf5eace7791 h1:7SYIDPMxbPBr8dMB6k9fYsgVt4ZyWVL41q4LMFsfAvs=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102112556-abf5eace7791/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128 h1:aI1DhZAuqVBXbybSqhLNOGX5rUOrn9T0cHNCDUQF7IQ=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102191058-f02f142af128/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=

node-tools/internal/bus/per_node.go

@@ -0,0 +1,105 @@
+package bus
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/pernode"
+)
+
+// A state its owner keeps one bucket per node (novox/hq ADR 0593 §6: each node's engine writes its own `composed-<node>`,
+// and the provisioner, the assigner and the view read them all) is served to a bundle as one state. The rules (a key's
+// first part names its node's bucket, keys and a watch over every bucket of the prefix, a bucket created later heard by
+// the bus's creation advisory) are mesh-sdk's pernode package, written once for a process and for this runtime; this
+// file adapts its entries to the runtime's shapes. A reader is granted the listing of stream names
+// (pernode.ListSubject) and the creation advisory (pernode.CreatedAdvisory) beside each bucket's read.
+
+func perNodeRefusal(module, name, what string) error {
+	return fmt.Errorf("%s's state %s is kept one bucket per node and read only: no %s of it as one state (novox/hq ADR 0593 §6)",
+		module, name, what)
+}
+
+func (c *Conn) family(s StateIssued) pernode.Family {
+	return pernode.Family{Read: s.Name, Prefix: s.Bucket, Bus: pernode.OverNATS(c.nc)}
+}
+
+func (c *Conn) perNodeGet(s StateIssued, key string) (*StateEntry, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	e, err := c.family(s).Get(ctx, key)
+	if err != nil || e == nil {
+		return nil, err
+	}
+	return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision()}, nil
+}
+
+func (c *Conn) perNodeKeys(s StateIssued) ([]string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	return c.family(s).Keys(ctx)
+}
+
+// perNodeWatch is StateWatch over every bucket of the family, and over a bucket created later. As in a single bucket's
+// watch, StateTimeout bounds the time without progress (issue 1726), and a value the bundle does not take is said and
+// the watch goes on.
+func (c *Conn) perNodeWatch(module string, s StateIssued, pattern string, deliver func(StateChange) error) (func(), error) {
+	ctx, cancel := context.WithCancel(context.Background())
+	progress := make(chan struct{}, 1)
+	handle := func(ch pernode.Change) error {
+		select {
+		case progress <- struct{}{}:
+		default:
+		}
+		e := ch.Entry
+		change := StateChange{State: s.Name, Key: e.Key(), Revision: e.Revision(), Current: ch.Current}
+		switch e.Operation() {
+		case jetstream.KeyValuePut:
+			change.Op = "put"
+			change.Value = valueOf(e.Value())
+		default:
+			if ch.Current {
+				return nil
+			}
+			change.Op = "delete"
+			if e.Operation() == jetstream.KeyValuePurge {
+				change.Op = "expire"
+			}
+		}
+		if ctx.Err() != nil {
+			return nil
+		}
+		if err := deliver(change); err != nil {
+			c.Logf("[mesh-tools] %s did not take %s %s of %s: %v; its next change, or its next start, reads it again",
+				module, change.Op, change.Key, s.Name, err)
+		}
+		return nil
+	}
+	done := make(chan error, 1)
+	go func() { done <- c.family(s).Watch(ctx, pattern, handle) }()
+	bound := time.NewTimer(StateTimeout)
+	defer bound.Stop()
+	for {
+		select {
+		case err := <-done:
+			if err != nil {
+				cancel()
+				return nil, err
+			}
+			return cancel, nil
+		case <-progress:
+			if !bound.Stop() {
+				select {
+				case <-bound.C:
+				default:
+				}
+			}
+			bound.Reset(StateTimeout)
+		case <-bound.C:
+			cancel()
+			return nil, fmt.Errorf("the current values of %s did not arrive in %s: none was handed over in that time", s.Name, StateTimeout)
+		}
+	}
+}

node-tools/internal/bus/state.go

@@ -28,6 +28,10 @@ type StateIssued struct {
 	Name   string `json:"name"`
 	Bucket string `json:"bucket"`
 	Writes bool   `json:"writes,omitempty"`
+	// PerNode says the state is kept one bucket per node (novox/hq ADR 0593 §6, mesh-sdk membership.StateIssued): Bucket is
+	// then the buckets' prefix, each bucket the prefix and a node's name, and a key's first part names its node. It is read
+	// only, served as one state (per_node.go).
+	PerNode bool `json:"per-node,omitempty"`
 }
 
 // StateEntry is one key's current value.
@@ -114,6 +118,9 @@ func (c *Conn) StateGet(module, name, key string) (*StateEntry, error) {
 	if err := checkKey(key); err != nil {
 		return nil, err
 	}
+	if s.PerNode {
+		return c.perNodeGet(s, key)
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
 	defer cancel()
 	kv, err := c.bucket(ctx, s)
@@ -149,6 +156,9 @@ func (c *Conn) StateHistory(module, name, key string) ([]StateKept, error) {
 	if err := checkKey(key); err != nil {
 		return nil, err
 	}
+	if s.PerNode {
+		return nil, perNodeRefusal(module, name, "history")
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
 	defer cancel()
 	kv, err := c.bucket(ctx, s)
@@ -199,6 +209,9 @@ func (c *Conn) StateInfo(module, name string) (*StateInfo, error) {
 	if err != nil {
 		return nil, err
 	}
+	if s.PerNode {
+		return nil, perNodeRefusal(module, name, "fill")
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
 	defer cancel()
 	kv, err := c.bucket(ctx, s)
@@ -325,12 +338,20 @@ func (c *Conn) StateKeys(module, name string) ([]string, error) {
 	if err != nil {
 		return nil, err
 	}
+	if s.PerNode {
+		return c.perNodeKeys(s)
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
 	defer cancel()
 	kv, err := c.bucket(ctx, s)
 	if err != nil {
 		return nil, err
 	}
+	return keysOf(ctx, kv)
+}
+
+// keysOf is every key with a current value in one bucket, sorted.
+func keysOf(ctx context.Context, kv jetstream.KeyValue) ([]string, error) {
 	lister, err := kv.ListKeys(ctx)
 	if err != nil {
 		return nil, err
@@ -392,12 +413,20 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange
 		}
 		filter = strings.Join(parts, ".")
 	}
+	if s.PerNode {
+		return c.perNodeWatch(module, s, pattern, deliver)
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
 	kv, err := c.bucket(ctx, s)
 	cancel()
 	if err != nil {
 		return nil, err
 	}
+	return c.watchBucket(module, name, kv, filter, deliver)
+}
+
+// watchBucket is StateWatch over one bucket: the current values, then every change.
+func (c *Conn) watchBucket(module, name string, kv jetstream.KeyValue, filter string, deliver func(StateChange) error) (stop func(), err error) {
 	watching, stopWatching := context.WithCancel(context.Background())
 	w, err := kv.Watch(watching, filter)
 	if err != nil {

node-tools/internal/bus/state_pernode_test.go

@@ -0,0 +1,111 @@
+//go:build integration
+
+package bus_test
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/novox/mesh-tools/node-tools/internal/bus"
+	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
+)
+
+// **A state its owner keeps one bucket per node is served to a bundle as one state** (novox/hq ADR 0593 §6: each node's
+// engine writes its own `composed-<node>`, and the provisioner reads them all). Get goes to the bucket the key's first
+// part names, Keys and Watch over every bucket of the prefix, a bucket created during the watch included, and a bucket
+// that only starts like the prefix is never read.
+func TestAPerNodeStateIsServedAcrossEveryNodesBucket(t *testing.T) {
+	mesh := mt.New(t)
+	a, b := mesh.Bucket(t, "composed-a"), mesh.Bucket(t, "composed-b")
+	other := mesh.Bucket(t, "composedx-z")
+	for kv, key := range map[interface {
+		Put(string, []byte) (uint64, error)
+	}]string{a: "a.m1.f1", b: "b.m2.f2", other: "z.m3.f3"} {
+		if _, err := kv.Put(key, []byte(`{"v":1}`)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	prov := mt.MembershipOf("mesh-provisioner", "anchor", false, nil)
+	prov.State = []bus.StateIssued{{Name: "mesh-host.composed", Bucket: "composed-", PerNode: true}}
+	mesh.Issue(t, prov)
+	conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn.Logf = func(string, ...any) {}
+	t.Cleanup(conn.Close)
+	conn.Follow("mesh-provisioner")
+	mt.Until(t, func() error {
+		if conn.Membership("mesh-provisioner") == nil {
+			return errors.New("not issued yet")
+		}
+		return nil
+	})
+
+	if e, err := conn.StateGet("mesh-provisioner", "mesh-host.composed", "b.m2.f2"); err != nil || e == nil || e.Key != "b.m2.f2" {
+		t.Fatalf("get from the second node's bucket: %+v, %v", e, err)
+	}
+	if e, err := conn.StateGet("mesh-provisioner", "mesh-host.composed", "q.m.f"); err != nil || e != nil {
+		t.Fatalf("a key of a node with no bucket should read as none: %+v, %v", e, err)
+	}
+	if e, err := conn.StateGet("mesh-provisioner", "mesh-host.composed", "z.m3.f3"); err != nil || e != nil {
+		t.Fatalf("the look-alike bucket was read: %+v, %v", e, err)
+	}
+	if _, err := conn.StateGet("mesh-provisioner", "mesh-host.composed", "nonode"); err == nil {
+		t.Fatal("a key naming no node should be refused")
+	}
+	if keys, err := conn.StateKeys("mesh-provisioner", "mesh-host.composed"); err != nil || !reflect.DeepEqual(keys, []string{"a.m1.f1", "b.m2.f2"}) {
+		t.Fatalf("keys across the buckets: %v, %v", keys, err)
+	}
+	if _, err := conn.StatePut("mesh-provisioner", "mesh-host.composed", "a.m1.f9", []byte(`{}`)); err == nil {
+		t.Fatal("a per-node state is read only")
+	}
+
+	var mu sync.Mutex
+	var current, live []string
+	stop, err := conn.StateWatch("mesh-provisioner", "mesh-host.composed", "", func(ch bus.StateChange) error {
+		mu.Lock()
+		defer mu.Unlock()
+		if ch.Current {
+			current = append(current, ch.Key)
+		} else {
+			live = append(live, ch.Key)
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stop)
+	mu.Lock()
+	sort.Strings(current)
+	if !reflect.DeepEqual(current, []string{"a.m1.f1", "b.m2.f2"}) {
+		t.Fatalf("current values: %v", current)
+	}
+	mu.Unlock()
+
+	// A third node's bucket, created during the watch, and a change in a first one.
+	c := mesh.Bucket(t, "composed-c")
+	if _, err := c.Put("c.m4.f4", []byte(`{"v":1}`)); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := a.Put("a.m1.f5", []byte(`{"v":1}`)); err != nil {
+		t.Fatal(err)
+	}
+	mt.Until(t, func() error {
+		mu.Lock()
+		defer mu.Unlock()
+		got := strings.Join(append(append([]string(nil), current...), live...), ",")
+		for _, want := range []string{"a.m1.f5", "c.m4.f4"} {
+			if !strings.Contains(got, want) {
+				return fmt.Errorf("%s not heard yet: current %v, changes %v", want, current, live)
+			}
+		}
+		return nil
+	})
+}