process: read a state its owner keeps one bucket per node as one state, so the provisioner, the assigner and the view read every node's compositions (ADR 0593 §6)

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-02 18:23:27 +01:00
parent b56841596e
commit 8dc610c628
3 changed files with 296 additions and 7 deletions
@@ -0,0 +1,190 @@
+package process
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// **A state its owner keeps one bucket per node is read as one state** (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). A reader names it as
+// its manifest's `reads` does (`<module>.<state>`, such as `mesh-host.composed`) with the prefix of its buckets
+// (`composed-`): each bucket `<prefix><node>` holds keys whose first part is that node, so a key names its bucket.
+// Get reads one key from its node's bucket; Keys and Watch go over every bucket of the family the bus holds, and Watch
+// takes in a node's bucket created after it began, as the bus says it (its creation advisory), so a node added later is
+// heard with no timer. Reads only: the owner's nodes write, as their grants allow.
+
+// What a reader of a per-node state is granted beside each bucket's read (the grants part composes them): the listing
+// of bucket names (FamilyListSubject) and the creation of any stream (FamilyCreatedAdvisory), which says a name only.
+const (
+	FamilyListSubject     = "$JS.API.STREAM.NAMES"
+	FamilyCreatedAdvisory = "$JS.EVENT.ADVISORY.STREAM.CREATED.*"
+)
+
+// NodeFamily is a per-node state read across every node's bucket.
+type NodeFamily struct {
+	Read   string
+	Prefix string
+	c      *Conn
+}
+
+// ReadPerNode is another module's state kept one bucket per node, named read in the reader's manifest, its buckets
+// prefix followed by the node's name.
+func (c *Conn) ReadPerNode(read, prefix string) NodeFamily {
+	return NodeFamily{Read: read, Prefix: prefix, c: c}
+}
+
+// checked refuses a family with no read name or prefix.
+func (f NodeFamily) checked() error {
+	module, name, ok := strings.Cut(f.Read, ".")
+	if !ok || module == "" || name == "" || strings.Contains(name, ".") {
+		return fmt.Errorf("%q is not a state another module keeps, <module>.<state>", f.Read)
+	}
+	if f.Prefix == "" || strings.ContainsAny(f.Prefix, ".*> ") {
+		return fmt.Errorf("the state %s names no prefix of its buckets, or one a bucket name cannot start with: %q", f.Read, f.Prefix)
+	}
+	return nil
+}
+
+// Node is the state of one node of the family: its bucket `<prefix><node>`, bound as Read binds a bucket.
+func (f NodeFamily) Node(node string) State {
+	module, name, _ := strings.Cut(f.Read, ".")
+	return f.c.bound(module, name, f.Prefix+node, func() error {
+		if err := f.checked(); err != nil {
+			return err
+		}
+		if node == "" || strings.ContainsAny(node, ".*> ") {
+			return fmt.Errorf("the state %s: %q is not a node a bucket is named for", f.Read, node)
+		}
+		return nil
+	})
+}
+
+// nodeOf is the node a key of the family names: its first part.
+func nodeOf(key string) (string, error) {
+	node, _, ok := strings.Cut(key, ".")
+	if !ok || node == "" {
+		return "", fmt.Errorf("%q names no node: a key of a per-node state starts <node>.", key)
+	}
+	return node, nil
+}
+
+// Nodes are the nodes whose bucket of the family the bus holds now, sorted: the names of the key-value buckets that
+// start with the prefix, the prefix taken off.
+func (f NodeFamily) Nodes() ([]string, error) {
+	if err := f.checked(); err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
+	defer cancel()
+	js, err := jetstream.New(f.c.nc)
+	if err != nil {
+		return nil, err
+	}
+	names := js.KeyValueStoreNames(ctx)
+	var nodes []string
+	for name := range names.Name() {
+		if node, ok := strings.CutPrefix(name, f.Prefix); ok && node != "" {
+			nodes = append(nodes, node)
+		}
+	}
+	if err := names.Error(); err != nil && !errors.Is(err, jetstream.ErrEndOfData) {
+		return nil, fmt.Errorf("the state %s: listing the buckets %s*: %w", f.Read, f.Prefix, err)
+	}
+	sort.Strings(nodes)
+	return nodes, nil
+}
+
+// Get is a key's current value, read from the bucket of the node the key names, or nil when it has none or that
+// node has no bucket.
+func (f NodeFamily) Get(key string) (*stdio.StateEntry, error) {
+	node, err := nodeOf(key)
+	if err != nil {
+		return nil, err
+	}
+	e, err := f.Node(node).Get(key)
+	if errors.Is(err, jetstream.ErrBucketNotFound) {
+		return nil, nil
+	}
+	return e, err
+}
+
+// Keys is every key with a current value in every bucket of the family, sorted.
+func (f NodeFamily) Keys() ([]string, error) {
+	nodes, err := f.Nodes()
+	if err != nil {
+		return nil, err
+	}
+	keys := []string{}
+	for _, node := range nodes {
+		k, err := f.Node(node).Keys()
+		if err != nil {
+			return nil, err
+		}
+		keys = append(keys, k...)
+	}
+	sort.Strings(keys)
+	return keys, nil
+}
+
+// Watch is State.Watch over every bucket of the family: handle hears the current values of every bucket the bus holds,
+// then every change, and the current values and changes of a bucket the bus creates later. Calls to handle are one at
+// a time. key narrows it as State.Watch's does, within each bucket; a key whose first part names a node watches only
+// that node's bucket, but is still told of no bucket that appears later for another node.
+func (f NodeFamily) Watch(key string, handle func(stdio.StateChange) error) error {
+	if err := f.checked(); err != nil {
+		return err
+	}
+	var mu sync.Mutex
+	one := func(c stdio.StateChange) error {
+		mu.Lock()
+		defer mu.Unlock()
+		return handle(c)
+	}
+	watched := map[string]bool{}
+	var wmu sync.Mutex
+	watch := func(node string) error {
+		wmu.Lock()
+		if watched[node] {
+			wmu.Unlock()
+			return nil
+		}
+		watched[node] = true
+		wmu.Unlock()
+		return f.Node(node).Watch(key, one)
+	}
+	if first, _, _ := strings.Cut(key, "."); key != "" && first != "*" && first != "**" {
+		return watch(first)
+	}
+	// Heard before the buckets are listed, so a bucket created between the two is not missed. A subject's wildcard is a
+	// whole token, so every stream's creation is heard (FamilyCreatedAdvisory) and the family's picked by name.
+	sub, err := f.c.nc.Subscribe(FamilyCreatedAdvisory, func(m *nats.Msg) {
+		bucket, isKV := strings.CutPrefix(m.Subject, strings.TrimSuffix(FamilyCreatedAdvisory, "*")+"KV_")
+		if node, ok := strings.CutPrefix(bucket, f.Prefix); isKV && ok && node != "" {
+			go func() { _ = watch(node) }()
+		}
+	})
+	if err != nil {
+		return fmt.Errorf("the state %s: hearing new buckets %s*: %w", f.Read, f.Prefix, err)
+	}
+	nodes, err := f.Nodes()
+	if err != nil {
+		_ = sub.Unsubscribe()
+		return err
+	}
+	for _, node := range nodes {
+		if err := watch(node); err != nil {
+			_ = sub.Unsubscribe()
+			return err
+		}
+	}
+	return nil
+}
@@ -0,0 +1,88 @@
+//go:build integration
+
+package process
+
+import (
+	"slices"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+)
+
+// ADR 0593 §6: a state its owner keeps one bucket per node (composed-<node>) is read as one state: a key is read from
+// the bucket its node part names, the keys of every node's bucket are listed together, and a watch hears every bucket,
+// one created after it began included, while a bucket outside the family is never read.
+func TestAPerNodeStateIsReadAcrossEveryNodesBucket(t *testing.T) {
+	url, _ := grantedSendProcess(t)
+	_, admin := adminJS(t, url)
+	put := func(bucket, key, value string) {
+		t.Helper()
+		kv, err := admin.CreateOrUpdateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: bucket})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := kv.Put(ctxFor(t), key, []byte(value)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	put("composed-ace", "ace.netcheck.aa", `{"n":1}`)
+	put("composed-g14", "g14.netcheck.bb", `{"n":2}`)
+	put("composedx-ace", "ace.other.cc", `{"n":3}`)
+
+	conn, err := Connect(Account{URL: url, Node: "novox", Module: "mesh-provisioner", User: adminUser, Password: adminPass})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	family := conn.ReadPerNode("mesh-host.composed", "composed-")
+
+	if nodes, err := family.Nodes(); err != nil || !slices.Equal(nodes, []string{"ace", "g14"}) {
+		t.Fatalf("nodes = %v, %v", nodes, err)
+	}
+	if e, err := family.Get("g14.netcheck.bb"); err != nil || e == nil || string(e.Value) != `{"n":2}` {
+		t.Fatalf("get from g14's bucket: %+v, %v", e, err)
+	}
+	if e, err := family.Get("laptop.netcheck.bb"); err != nil || e != nil {
+		t.Fatalf("a node with no bucket: %+v, %v", e, err)
+	}
+	if keys, err := family.Keys(); err != nil || !slices.Equal(keys, []string{"ace.netcheck.aa", "g14.netcheck.bb"}) {
+		t.Fatalf("keys = %v, %v", keys, err)
+	}
+
+	var mu sync.Mutex
+	heard := map[string]bool{}
+	err = family.Watch("", func(c stdio.StateChange) error {
+		mu.Lock()
+		defer mu.Unlock()
+		heard[c.Key] = true
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	put("composed-shanks", "shanks.netcheck.dd", `{"n":4}`)
+	put("composed-ace", "ace.netcheck.ee", `{"n":5}`)
+	deadline := time.Now().Add(5 * time.Second)
+	want := []string{"ace.netcheck.aa", "ace.netcheck.ee", "g14.netcheck.bb", "shanks.netcheck.dd"}
+	for {
+		mu.Lock()
+		var got []string
+		for k := range heard {
+			got = append(got, k)
+		}
+		mu.Unlock()
+		slices.Sort(got)
+		if slices.Equal(got, want) {
+			break
+		}
+		if time.Now().After(deadline) || slices.ContainsFunc(got, func(k string) bool { return strings.HasPrefix(k, "ace.other") }) {
+			t.Fatalf("heard %v, want %v", got, want)
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+}
@@ -28,6 +28,8 @@ const stateTimeout = 30 * time.Second
 type State struct {
 	Module string
 	Name   string
+	// bucketName is the bucket when it is not <module>_<name>: one node's bucket of a per-node state (NodeFamily).
+	bucketName 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)
@@ -36,14 +38,19 @@ type State struct {
 }

 // bucket is this state's name on the bus.
-func (s State) bucket() string { return s.Module + "_" + s.Name }
+func (s State) bucket() string {
+	if s.bucketName != "" {
+		return s.bucketName
+	}
+	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 {
-	return c.bound(c.account.Module, name, func() error { return c.refusedUnannounced(name) })
+	return c.bound(c.account.Module, name, "", func() error { return c.refusedUnannounced(name) })
 }

 // Read is another module's state this connection's module reads, named as its manifest's `reads` names it
@@ -52,7 +59,7 @@ func (c *Conn) State(name string) State {
 // the seat mesh-provisioner's plans). A name that is not `<module>.<state>` fails at the call that needed it.
 func (c *Conn) Read(read string) State {
 	module, name, ok := strings.Cut(read, ".")
-	return c.bound(module, name, func() error {
+	return c.bound(module, name, "", func() error {
 		if !ok || module == "" || name == "" || strings.Contains(name, ".") {
 			return fmt.Errorf("%q is not a state another module keeps, <module>.<state>", read)
 		}
@@ -60,10 +67,14 @@ func (c *Conn) Read(read string) State {
 	})
 }

-// bound is module's state name on this connection, checked by refused before every bind.
-func (c *Conn) bound(module, name string, refused func() error) State {
-	bucket := module + "_" + name
-	return State{Module: module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
+// bound is module's state name on this connection, in its bucket <module>_<name> or the bucket given, checked by
+// refused before every bind.
+func (c *Conn) bound(module, name, bucket string, refused func() error) State {
+	given := bucket
+	if bucket == "" {
+		bucket = module + "_" + name
+	}
+	return State{Module: module, Name: name, bucketName: given, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
 		if err := refused(); err != nil {
 			return nil, err
 		}