From 8dc610c628e51cb3103cc41e3aba0df1d2d39b67 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 18:23:27 +0100 Subject: [PATCH] =?UTF-8?q?process:=20read=20a=20state=20its=20owner=20kee?= =?UTF-8?q?ps=20one=20bucket=20per=20node=20as=20one=20state,=20so=20the?= =?UTF-8?q?=20provisioner,=20the=20assigner=20and=20the=20view=20read=20ev?= =?UTF-8?q?ery=20node's=20compositions=20(ADR=200593=20=C2=A76)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- go/process/pernode.go | 190 +++++++++++++++++++++++++ go/process/pernode_integration_test.go | 88 ++++++++++++ go/process/state.go | 25 +++- 3 files changed, 296 insertions(+), 7 deletions(-) create mode 100644 go/process/pernode.go create mode 100644 go/process/pernode_integration_test.go diff --git a/go/process/pernode.go b/go/process/pernode.go new file mode 100644 index 0000000..9494dd6 --- /dev/null +++ b/go/process/pernode.go @@ -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-`, and the provisioner, the assigner and the view read them all). A reader names it as +// its manifest's `reads` does (`.`, such as `mesh-host.composed`) with the prefix of its buckets +// (`composed-`): each bucket `` 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, .", 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 ``, 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 .", 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 +} diff --git a/go/process/pernode_integration_test.go b/go/process/pernode_integration_test.go new file mode 100644 index 0000000..6eb6e8c --- /dev/null +++ b/go/process/pernode_integration_test.go @@ -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-) 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) + } +} diff --git a/go/process/state.go b/go/process/state.go index 48dbcff..4d44d44 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -28,6 +28,8 @@ const stateTimeout = 30 * time.Second type State struct { Module string Name string + // bucketName is the bucket when it is not _: 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_, 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 `.` 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, .", 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 _ 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 }