From 8114b45884d543772f69daf5203a732af2d15202 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 18:27:34 +0100 Subject: [PATCH 1/4] Serve a per-node state to a bundle as one state: get by the key's node, keys and watch over every node's bucket, a later bucket too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The provisioner, the assigner and the view read every node's composition (ADR 0593 §6); the runtime followed mesh-sdk's membership.StateIssued PerNode and process.NodeFamily on its own connection. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- node-tools/internal/bus/per_node.go | 185 ++++++++++++++++++ node-tools/internal/bus/state.go | 39 +++- node-tools/internal/bus/state_pernode_test.go | 111 +++++++++++ 3 files changed, 334 insertions(+), 1 deletion(-) create mode 100644 node-tools/internal/bus/per_node.go create mode 100644 node-tools/internal/bus/state_pernode_test.go diff --git a/node-tools/internal/bus/per_node.go b/node-tools/internal/bus/per_node.go new file mode 100644 index 0000000..01a65b2 --- /dev/null +++ b/node-tools/internal/bus/per_node.go @@ -0,0 +1,185 @@ +package bus + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + "sync" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// A state its owner keeps one bucket per node (novox/hq ADR 0593 §6: each node's engine writes its own `composed-`, +// and the provisioner, the assigner and the view read them all) is served to a bundle as one state, as mesh-sdk's +// process.NodeFamily serves it to a process: a key is read from the bucket its first part names, the keys and a watch +// go over every bucket of the prefix, and a watch takes in a bucket created later, heard by the bus's creation +// advisory, so a node added later is heard with no timer. This runtime holds its own connection, not mesh-sdk's, so +// the same rules are written here over it; the two subjects below are the ones the grants part grants a reader. +const ( + familyListSubject = "$JS.API.STREAM.NAMES" + familyCreatedAdvisory = "$JS.EVENT.ADVISORY.STREAM.CREATED." +) + +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) +} + +// nodeBucketOf is the state of the one node a key names: its first part, which must be a node's name. +func nodeBucketOf(s StateIssued, key string) (StateIssued, error) { + node, _, ok := strings.Cut(key, ".") + if !ok || node == "" { + return s, fmt.Errorf("%q names no node: a key of the per-node state %s starts .", key, s.Name) + } + s.Bucket += node + return s, nil +} + +// perNodeNodes are the nodes whose bucket of the family the bus holds now, sorted. +func (c *Conn) perNodeNodes(s StateIssued) ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + defer cancel() + js, err := jetstream.New(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, s.Bucket); 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", s.Name, s.Bucket, err) + } + sort.Strings(nodes) + return nodes, nil +} + +func (c *Conn) perNodeKeys(s StateIssued) ([]string, error) { + nodes, err := c.perNodeNodes(s) + if err != nil { + return nil, err + } + keys := []string{} + for _, node := range nodes { + one := s + one.Bucket += node + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + kv, err := c.bucket(ctx, one) + if err == nil { + var k []string + k, err = keysOf(ctx, kv) + keys = append(keys, k...) + } + cancel() + if err != nil { + return nil, err + } + } + sort.Strings(keys) + return keys, nil +} + +// perNodeWatch is StateWatch over every bucket of the family, and over a bucket created later. A pattern whose first +// part names a node watches that node's bucket only. +func (c *Conn) perNodeWatch(module string, s StateIssued, filter string, deliver func(StateChange) error) (func(), error) { + var mu sync.Mutex + one := func(ch StateChange) error { + mu.Lock() + defer mu.Unlock() + return deliver(ch) + } + var ( + wmu sync.Mutex + watched = map[string]bool{} + stops []func() + done bool + sub *nats.Subscription + ) + add := func(node string) error { + wmu.Lock() + if watched[node] || done { + wmu.Unlock() + return nil + } + watched[node] = true + wmu.Unlock() + bs := s + bs.Bucket += node + ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) + kv, err := c.bucket(ctx, bs) + cancel() + if err != nil { + return err + } + stop, err := c.watchBucket(module, s.Name, kv, filter, one) + if err != nil { + return err + } + wmu.Lock() + if done { + wmu.Unlock() + stop() + return nil + } + stops = append(stops, stop) + wmu.Unlock() + return nil + } + stopAll := func() { + wmu.Lock() + done = true + all := stops + stops = nil + s := sub + wmu.Unlock() + if s != nil { + _ = s.Unsubscribe() + } + for _, stop := range all { + stop() + } + } + if first, _, _ := strings.Cut(filter, "."); filter != ">" && first != "*" && first != ">" { + if err := add(first); err != nil { + stopAll() + return nil, err + } + return stopAll, nil + } + // Heard before the buckets are listed, so a bucket created between the two is not missed. + subject := familyCreatedAdvisory + "*" + sb, err := c.nc.Subscribe(subject, func(m *nats.Msg) { + bucket, isKV := strings.CutPrefix(m.Subject, familyCreatedAdvisory+"KV_") + if node, ok := strings.CutPrefix(bucket, s.Bucket); isKV && ok && node != "" { + go func() { + if err := add(node); err != nil { + c.Logf("[mesh-tools] %s did not take in the new bucket %s%s of %s: %v", module, s.Bucket, node, s.Name, err) + } + }() + } + }) + if err != nil { + return nil, fmt.Errorf("the state %s: hearing new buckets %s*: %w", s.Name, s.Bucket, err) + } + wmu.Lock() + sub = sb + wmu.Unlock() + nodes, err := c.perNodeNodes(s) + if err != nil { + stopAll() + return nil, err + } + for _, node := range nodes { + if err := add(node); err != nil { + stopAll() + return nil, err + } + } + return stopAll, nil +} diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index f396cc6..82970d8 100644 --- a/node-tools/internal/bus/state.go +++ b/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. @@ -81,6 +85,9 @@ func (c *Conn) issuedState(module, name string) (StateIssued, error) { module, name, issued) } +// errNoBucket marks a bucket the bus does not hold yet. +var errNoBucket = errors.New("no such bucket") + func (c *Conn) bucket(ctx context.Context, s StateIssued) (jetstream.KeyValue, error) { js, err := jetstream.New(c.nc) if err != nil { @@ -90,7 +97,7 @@ func (c *Conn) bucket(ctx context.Context, s StateIssued) (jetstream.KeyValue, e if err != nil { if errors.Is(err, jetstream.ErrBucketNotFound) { return nil, fmt.Errorf("the state %q is issued and its bucket is not on the bus yet: the mesh "+ - "creates it from the catalogue on its next raise", s.Name) + "creates it from the catalogue on its next raise: %w", s.Name, errNoBucket) } return nil, err } @@ -114,10 +121,18 @@ func (c *Conn) StateGet(module, name, key string) (*StateEntry, error) { if err := checkKey(key); err != nil { return nil, err } + if s.PerNode { + if s, err = nodeBucketOf(s, key); err != nil { + return nil, err + } + } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() kv, err := c.bucket(ctx, s) if err != nil { + if s.PerNode && errors.Is(err, errNoBucket) { + return nil, nil + } return nil, err } e, err := kv.Get(ctx, key) @@ -149,6 +164,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 +217,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 +346,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 +421,20 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange } filter = strings.Join(parts, ".") } + if s.PerNode { + return c.perNodeWatch(module, s, filter, 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 { diff --git a/node-tools/internal/bus/state_pernode_test.go b/node-tools/internal/bus/state_pernode_test.go new file mode 100644 index 0000000..de04809 --- /dev/null +++ b/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-`, 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 + }) +} From 3f538027e9e4fef5c5a827dead67d334a5ca9fb5 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 18:31:18 +0100 Subject: [PATCH 2/4] Read a per-node state through mesh-sdk's pernode package, so the runtime and a process share one copy of its rules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the runtime's own bucket listing, advisory hearing and per-bucket bookkeeping; the bound on progress and the runtime's shapes stay here. ADR 0593 §6. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- node-tools/internal/bus/per_node.go | 218 +++++++++------------------- node-tools/internal/bus/state.go | 14 +- 2 files changed, 72 insertions(+), 160 deletions(-) diff --git a/node-tools/internal/bus/per_node.go b/node-tools/internal/bus/per_node.go index 01a65b2..ba64d35 100644 --- a/node-tools/internal/bus/per_node.go +++ b/node-tools/internal/bus/per_node.go @@ -2,184 +2,104 @@ package bus import ( "context" - "errors" "fmt" - "sort" - "strings" - "sync" + "time" - "github.com/nats-io/nats.go" "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-`, -// and the provisioner, the assigner and the view read them all) is served to a bundle as one state, as mesh-sdk's -// process.NodeFamily serves it to a process: a key is read from the bucket its first part names, the keys and a watch -// go over every bucket of the prefix, and a watch takes in a bucket created later, heard by the bus's creation -// advisory, so a node added later is heard with no timer. This runtime holds its own connection, not mesh-sdk's, so -// the same rules are written here over it; the two subjects below are the ones the grants part grants a reader. -const ( - familyListSubject = "$JS.API.STREAM.NAMES" - familyCreatedAdvisory = "$JS.EVENT.ADVISORY.STREAM.CREATED." -) +// 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) } -// nodeBucketOf is the state of the one node a key names: its first part, which must be a node's name. -func nodeBucketOf(s StateIssued, key string) (StateIssued, error) { - node, _, ok := strings.Cut(key, ".") - if !ok || node == "" { - return s, fmt.Errorf("%q names no node: a key of the per-node state %s starts .", key, s.Name) - } - s.Bucket += node - return s, nil +func (c *Conn) family(s StateIssued) pernode.Family { + return pernode.Family{Read: s.Name, Prefix: s.Bucket, Bus: pernode.OverNATS(c.nc)} } -// perNodeNodes are the nodes whose bucket of the family the bus holds now, sorted. -func (c *Conn) perNodeNodes(s StateIssued) ([]string, error) { +func (c *Conn) perNodeGet(s StateIssued, key string) (*StateEntry, error) { ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() - js, err := jetstream.New(c.nc) - if err != nil { + e, err := c.family(s).Get(ctx, key) + if err != nil || e == nil { return nil, err } - names := js.KeyValueStoreNames(ctx) - var nodes []string - for name := range names.Name() { - if node, ok := strings.CutPrefix(name, s.Bucket); 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", s.Name, s.Bucket, err) - } - sort.Strings(nodes) - return nodes, nil + return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision()}, nil } func (c *Conn) perNodeKeys(s StateIssued) ([]string, error) { - nodes, err := c.perNodeNodes(s) - if err != nil { - return nil, err - } - keys := []string{} - for _, node := range nodes { - one := s - one.Bucket += node - ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) - kv, err := c.bucket(ctx, one) - if err == nil { - var k []string - k, err = keysOf(ctx, kv) - keys = append(keys, k...) - } - cancel() - if err != nil { - return nil, err - } - } - sort.Strings(keys) - return keys, nil + 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. A pattern whose first -// part names a node watches that node's bucket only. -func (c *Conn) perNodeWatch(module string, s StateIssued, filter string, deliver func(StateChange) error) (func(), error) { - var mu sync.Mutex - one := func(ch StateChange) error { - mu.Lock() - defer mu.Unlock() - return deliver(ch) - } - var ( - wmu sync.Mutex - watched = map[string]bool{} - stops []func() - done bool - sub *nats.Subscription - ) - add := func(node string) error { - wmu.Lock() - if watched[node] || done { - wmu.Unlock() +// 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 } - watched[node] = true - wmu.Unlock() - bs := s - bs.Bucket += node - ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) - kv, err := c.bucket(ctx, bs) - cancel() - if err != nil { - return err + 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) } - stop, err := c.watchBucket(module, s.Name, kv, filter, one) - if err != nil { - return err - } - wmu.Lock() - if done { - wmu.Unlock() - stop() - return nil - } - stops = append(stops, stop) - wmu.Unlock() return nil } - stopAll := func() { - wmu.Lock() - done = true - all := stops - stops = nil - s := sub - wmu.Unlock() - if s != nil { - _ = s.Unsubscribe() - } - for _, stop := range all { - stop() - } - } - if first, _, _ := strings.Cut(filter, "."); filter != ">" && first != "*" && first != ">" { - if err := add(first); err != nil { - stopAll() - return nil, err - } - return stopAll, nil - } - // Heard before the buckets are listed, so a bucket created between the two is not missed. - subject := familyCreatedAdvisory + "*" - sb, err := c.nc.Subscribe(subject, func(m *nats.Msg) { - bucket, isKV := strings.CutPrefix(m.Subject, familyCreatedAdvisory+"KV_") - if node, ok := strings.CutPrefix(bucket, s.Bucket); isKV && ok && node != "" { - go func() { - if err := add(node); err != nil { - c.Logf("[mesh-tools] %s did not take in the new bucket %s%s of %s: %v", module, s.Bucket, node, s.Name, err) + 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: } - }() - } - }) - if err != nil { - return nil, fmt.Errorf("the state %s: hearing new buckets %s*: %w", s.Name, s.Bucket, err) - } - wmu.Lock() - sub = sb - wmu.Unlock() - nodes, err := c.perNodeNodes(s) - if err != nil { - stopAll() - return nil, err - } - for _, node := range nodes { - if err := add(node); err != nil { - stopAll() - return nil, err + } + 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) } } - return stopAll, nil } diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index 82970d8..935ddff 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -85,9 +85,6 @@ func (c *Conn) issuedState(module, name string) (StateIssued, error) { module, name, issued) } -// errNoBucket marks a bucket the bus does not hold yet. -var errNoBucket = errors.New("no such bucket") - func (c *Conn) bucket(ctx context.Context, s StateIssued) (jetstream.KeyValue, error) { js, err := jetstream.New(c.nc) if err != nil { @@ -97,7 +94,7 @@ func (c *Conn) bucket(ctx context.Context, s StateIssued) (jetstream.KeyValue, e if err != nil { if errors.Is(err, jetstream.ErrBucketNotFound) { return nil, fmt.Errorf("the state %q is issued and its bucket is not on the bus yet: the mesh "+ - "creates it from the catalogue on its next raise: %w", s.Name, errNoBucket) + "creates it from the catalogue on its next raise", s.Name) } return nil, err } @@ -122,17 +119,12 @@ func (c *Conn) StateGet(module, name, key string) (*StateEntry, error) { return nil, err } if s.PerNode { - if s, err = nodeBucketOf(s, key); err != nil { - return nil, err - } + return c.perNodeGet(s, key) } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() kv, err := c.bucket(ctx, s) if err != nil { - if s.PerNode && errors.Is(err, errNoBucket) { - return nil, nil - } return nil, err } e, err := kv.Get(ctx, key) @@ -422,7 +414,7 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange filter = strings.Join(parts, ".") } if s.PerNode { - return c.perNodeWatch(module, s, filter, deliver) + return c.perNodeWatch(module, s, pattern, deliver) } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) kv, err := c.bucket(ctx, s) From bd24ceafa9017fcdf038658065fa3bfeb0949723 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 19:29:53 +0100 Subject: [PATCH 3/4] Pin mesh-sdk at the sdk1 group's head 1c7356c, so the group's check builds every module against the narrowed SDK it was written for Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- node-tools/go.mod | 2 +- node-tools/go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/node-tools/go.mod b/node-tools/go.mod index 717ea86..ed52861 100644 --- a/node-tools/go.mod +++ b/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.20261102181925-1c7356c24722 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 diff --git a/node-tools/go.sum b/node-tools/go.sum index ebd4e0f..9125505 100644 --- a/node-tools/go.sum +++ b/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.20261102181925-1c7356c24722 h1:ptP67HDdrlX3w421JtVaolCeHxahndFYf/uYPS7U2sc= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102181925-1c7356c24722/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= From 24c0ec0d324a7621918b0a4219aef9f269e8fe3f Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 20:14:16 +0100 Subject: [PATCH 4/4] Pin mesh-sdk at f02f142, the sdk1 head whose tests no longer leak package globals, so the group's check builds against it Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- node-tools/go.mod | 2 +- node-tools/go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/node-tools/go.mod b/node-tools/go.mod index ed52861..4b2d0b9 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -8,7 +8,7 @@ require ( ) require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102181925-1c7356c24722 + 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 diff --git a/node-tools/go.sum b/node-tools/go.sum index 9125505..43cc4ec 100644 --- a/node-tools/go.sum +++ b/node-tools/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102181925-1c7356c24722 h1:ptP67HDdrlX3w421JtVaolCeHxahndFYf/uYPS7U2sc= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261102181925-1c7356c24722/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=