Read a per-node state through mesh-sdk's pernode package, so the runtime and a process share one copy of its rules

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

@@ -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-<node>`,
-// 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 <node>.", 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
 }

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)