mesh-wants upgrade: read the large states by one watch each, started once, not a get per key, so compare answers in time (issue 1953 L5f)

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go

@@ -15,6 +15,7 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/health"
 	"git.novox.be/novox/mesh-sdk/go/provisions"
 	"github.com/nats-io/nats.go/jetstream"
 
@@ -400,6 +401,34 @@ func wantsVerbs(t *testing.T) {
 	if splits := out.(map[string]any)["splits"].([]assign.Split); len(splits) != 1 || splits[0].Module != "web" || out.(map[string]any)["findings"] == nil {
 		t.Errorf("splits = %+v", splits)
 	}
+	// Read by watch (issue 1953 L5f): the same answer, each watched state started once, and no get per key of the
+	// large states (the modules, their manifests, the composed record).
+	watches := map[string]int{}
+	byWatch := newWantsTools(st, nil, nil)
+	byWatch.follow = func(state string, handle func(string, []byte, bool)) error {
+		watches[state]++
+		for k, v := range st[state].values {
+			handle(k, v, false)
+		}
+		return nil
+	}
+	for range 2 {
+		wout, werr := byWatch.doUpgrade(map[string]any{"compare": "true"}, "op")
+		if werr != nil {
+			t.Fatal(werr)
+		}
+		if ws := wout.(map[string]any)["splits"].([]assign.Split); len(ws) != 1 || ws[0].Module != "web" {
+			t.Errorf("splits by watch = %+v", ws)
+		}
+	}
+	for _, state := range []string{modulesState, manifestsState, composedState, health.FindingsState} {
+		if watches[state] != 1 {
+			t.Errorf("%s was watched %d times, want once", state, watches[state])
+		}
+	}
+	if watches[policiesState] != 0 {
+		t.Errorf("the seat's own small policies state was watched: %v", watches)
+	}
 	// The process says the split as its finding, and deletes it once they agree.
 	findings := map[string][]byte{}
 	p := &passer{w: newWorld(), logf: func(string, ...any) {}, findings: assign.Findings{

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go

@@ -352,10 +352,17 @@ func serveTools() error {
 		}
 	}()
 	tools := append(s.tools(), newSeatTools(runtimeStore{}, stdio.Ask).tools()...)
-	tools = append(tools, newWantsTools(runtimeStore{}, stdio.Ask, func(e, id string, b any) error {
+	wants := newWantsTools(runtimeStore{}, stdio.Ask, func(e, id string, b any) error {
 		// The seat's event, said on the seat: a claimed seat's event is never the module's own.
 		return stdio.SeatPublish(wantsEventSubject(e), b, id)
-	}).tools()...)
+	})
+	wants.follow = func(state string, handle func(key string, value []byte, deleted bool)) error {
+		return stdio.State(state).Watch("", func(c stdio.StateChange) error {
+			handle(c.Key, c.Value, c.Op != stdio.OpPut)
+			return nil
+		})
+	}
+	tools = append(tools, wants.tools()...)
 	return stdio.Serve("", tools)
 }
 

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"sort"
 	"strconv"
+	"sync"
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
@@ -41,6 +42,17 @@ type wantsTools struct {
 	ask  func(key string, body any) (json.RawMessage, error)
 	emit func(event, id string, body any) error
 	now  func() time.Time
+	// follow starts a watch of a state: handle gets every current value, then every change (deleted for a key gone),
+	// and follow returns once the current values are all handled (stdio's KeptState.Watch). Nil reads key by key.
+	follow func(state string, handle func(key string, value []byte, deleted bool)) error
+	mu     sync.Mutex
+	mirror map[string]*watchedState
+}
+
+// watchedState is one state as a watch holds it: every key's value now, kept by the changes it hears.
+type watchedState struct {
+	mu     sync.Mutex
+	values map[string][]byte
 }
 
 // wantsEventSubject is the subject of one of the seat's events.
@@ -49,7 +61,7 @@ func wantsEventSubject(event string) string {
 }
 
 func newWantsTools(st seatStore, ask func(string, any) (json.RawMessage, error), emit func(event, id string, body any) error) *wantsTools {
-	return &wantsTools{st: st, ask: ask, emit: emit, now: time.Now}
+	return &wantsTools{st: st, ask: ask, emit: emit, now: time.Now, mirror: map[string]*watchedState{}}
 }
 
 // leased says the controller's lease is handed away.
@@ -59,8 +71,52 @@ func (s *wantsTools) leased() bool {
 	return away
 }
 
-// values reads every key of a state.
+// watched is a state as a watch holds it, started on the first read and kept by its changes after (NATS-native: a read of
+// a whole state is one watch's current values, never a get per key, which timed out on a large state, issue 1953 L5f).
+// The watch lasts as long as the process, so a state is watched once.
+func (s *wantsTools) watched(state string) (*watchedState, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if w, ok := s.mirror[state]; ok {
+		return w, nil
+	}
+	w := &watchedState{values: map[string][]byte{}}
+	err := s.follow(state, func(key string, value []byte, deleted bool) {
+		w.mu.Lock()
+		defer w.mu.Unlock()
+		if deleted {
+			delete(w.values, key)
+			return
+		}
+		w.values[key] = append([]byte(nil), value...)
+	})
+	if err != nil {
+		return nil, fmt.Errorf("%s cannot be watched: %w", state, err)
+	}
+	s.mirror[state] = w
+	return w, nil
+}
+
+// values reads every key of a state: by the watch when there is one, else key by key.
 func (s *wantsTools) values(state string) (map[string][]byte, error) {
+	if s.follow != nil {
+		w, err := s.watched(state)
+		if err != nil {
+			return nil, err
+		}
+		w.mu.Lock()
+		defer w.mu.Unlock()
+		out := make(map[string][]byte, len(w.values))
+		for k, v := range w.values {
+			out[k] = v
+		}
+		return out, nil
+	}
+	return s.direct(state)
+}
+
+// direct reads a state key by key, for a small state of the seat's own.
+func (s *wantsTools) direct(state string) (map[string][]byte, error) {
 	keys, err := s.st.keys(state)
 	if err != nil {
 		return nil, fmt.Errorf("%s cannot be listed: %w", state, err)
@@ -78,6 +134,22 @@ func (s *wantsTools) values(state string) (map[string][]byte, error) {
 	return out, nil
 }
 
+// one reads a key of a state: from the watch when there is one.
+func (s *wantsTools) one(state, key string) ([]byte, bool, error) {
+	if s.follow == nil {
+		raw, _, found, err := s.st.get(state, key)
+		return raw, found, err
+	}
+	w, err := s.watched(state)
+	if err != nil {
+		return nil, false, err
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	raw, found := w.values[key]
+	return raw, found, nil
+}
+
 // manifests is every module the indexer holds, with its trunk manifest and fingerprint.
 func (s *wantsTools) manifests() (map[string]catalogue.Manifest, map[string]delivery.Module, error) {
 	mods, err := s.values(modulesState)
@@ -94,7 +166,7 @@ func (s *wantsTools) manifests() (map[string]catalogue.Manifest, map[string]deli
 		if m.Manifest == "" {
 			continue
 		}
-		mraw, _, found, err := s.st.get(manifestsState, m.Manifest)
+		mraw, found, err := s.one(manifestsState, m.Manifest)
 		if err != nil || !found {
 			continue
 		}
@@ -126,7 +198,7 @@ func (s *wantsTools) doUpgrade(args map[string]any, by string) (any, error) {
 	if err != nil {
 		return nil, err
 	}
-	raw, err := s.values(policiesState)
+	raw, err := s.direct(policiesState) // the seat's own small state, read whole after its own writes
 	if err != nil {
 		return nil, err
 	}