From f95d5b4d2357cf5191451eb1fb2278271ee2b363 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 03:22:12 +0100 Subject: [PATCH 1/2] 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 --- .../mesh-delivery-assigner/seat_tools_test.go | 29 +++++++ .../cmd/mesh-delivery-assigner/tools.go | 11 ++- .../cmd/mesh-delivery-assigner/wants_tools.go | 80 ++++++++++++++++++- 3 files changed, 114 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index 1de1966..c21f035 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/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{ diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 74a5180..c55c45d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/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) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go index 1be3cb3..0f52ffc 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go +++ b/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 } From b8d4a77e7584554da3ea27c71c390acc65a206c9 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 03:34:21 +0100 Subject: [PATCH 2/2] mesh-wants upgrade reads through mesh-sdk's watched reader, keeping no mirror of its own One helper serves every part that reads whole states (ADR 0580 clause 3): the assigner's own watched mirror goes, its compare reads the large states from stdio.WatchedStates and each manifest once per digest. mesh-sdk pinned at 8311ecb. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- .../mesh-delivery-assigner/seat_tools_test.go | 22 ++++- .../cmd/mesh-delivery-assigner/tools.go | 7 +- .../cmd/mesh-delivery-assigner/wants_tools.go | 82 ++++--------------- modules/mesh-delivery-assigner/go.mod | 2 +- modules/mesh-delivery-assigner/go.sum | 4 + 5 files changed, 40 insertions(+), 77 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index c21f035..5379540 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "os" + "reflect" "slices" "strings" "testing" @@ -405,13 +406,17 @@ func wantsVerbs(t *testing.T) { // 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 { + byWatch.watched = stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error { watches[state]++ for k, v := range st[state].values { - handle(k, v, false) + _ = handle(stdio.StateChange{Key: k, Op: stdio.OpPut, Value: v, Current: true}) } return nil - } + }, func(state, key string) ([]byte, bool, error) { + watches[state+" by key"]++ + v, ok := st[state].values[key] + return v, ok, nil + }) for range 2 { wout, werr := byWatch.doUpgrade(map[string]any{"compare": "true"}, "op") if werr != nil { @@ -421,11 +426,20 @@ func wantsVerbs(t *testing.T) { t.Errorf("splits by watch = %+v", ws) } } - for _, state := range []string{modulesState, manifestsState, composedState, health.FindingsState} { + if n := watches[manifestsState+" by key"]; n == 0 || n > len(st[manifestsState].values) { + t.Errorf("the manifests were read by key %d times, want each once", n) + } + for _, state := range []string{modulesState, composedState, health.FindingsState} { if watches[state] != 1 { t.Errorf("%s was watched %d times, want once", state, watches[state]) } } + // ADR 0580 clause 3: the compare reads through mesh-sdk's watched reader and the seat keeps no mirror of its own. + for f := range reflect.TypeFor[wantsTools]().Fields() { + if f.Type.Kind() == reflect.Map || (f.Type.Kind() == reflect.Pointer && f.Type.Elem().Kind() == reflect.Map) { + t.Errorf("wantsTools keeps %s, a copy of its own beside the watched reader", f.Name) + } + } if watches[policiesState] != 0 { t.Errorf("the seat's own small policies state was watched: %v", watches) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index c55c45d..2ed4d10 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -356,12 +356,7 @@ func serveTools() 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) }) - 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 - }) - } + wants.watched = stdio.NewWatchedStates() // the large states read whole from one watched copy (ADR 0580 clause 3) tools = append(tools, wants.tools()...) return stdio.Serve("", tools) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go index 0f52ffc..30ad161 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go @@ -6,7 +6,6 @@ import ( "fmt" "sort" "strconv" - "sync" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -42,17 +41,10 @@ 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 + // watched is the copy of the large states it reads whole, each by one watch started on its first read and kept by + // its changes after (mesh-sdk's WatchedStates, novox/hq ADR 0580 clause 3), never a get per key, which timed out on + // a large state (issue 1953 L5f). Nil reads key by key. + watched *stdio.WatchedStates } // wantsEventSubject is the subject of one of the seat's events. @@ -61,7 +53,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, mirror: map[string]*watchedState{}} + return &wantsTools{st: st, ask: ask, emit: emit, now: time.Now} } // leased says the controller's lease is handed away. @@ -71,46 +63,10 @@ func (s *wantsTools) leased() bool { return away } -// 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. +// values reads every key of a state: from the watched copy 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 + if s.watched != nil { + return s.watched.Values(state) } return s.direct(state) } @@ -134,20 +90,14 @@ func (s *wantsTools) direct(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 +// manifest reads a manifest by its digest: kept once read when there is a watched copy, a manifest never changing +// under its digest. +func (s *wantsTools) manifest(digest string) ([]byte, bool, error) { + if s.watched != nil { + return s.watched.Once(manifestsState, digest) } - 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 + raw, _, found, err := s.st.get(manifestsState, digest) + return raw, found, err } // manifests is every module the indexer holds, with its trunk manifest and fingerprint. @@ -166,7 +116,7 @@ func (s *wantsTools) manifests() (map[string]catalogue.Manifest, map[string]deli if m.Manifest == "" { continue } - mraw, found, err := s.one(manifestsState, m.Manifest) + mraw, found, err := s.manifest(m.Manifest) if err != nil || !found { continue } diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index b3749c2..361c833 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-assigner go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101020750-62cddf4efc62 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum index eaa1d08..3e9c572 100644 --- a/modules/mesh-delivery-assigner/go.sum +++ b/modules/mesh-delivery-assigner/go.sum @@ -12,6 +12,10 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101014310-19c204273bdc h1:kVq7sS4I git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101014310-19c204273bdc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101020750-62cddf4efc62 h1:R5C8ICjiaDITHRFR/uIATwARPPXKxssUC70m0svstb0= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101020750-62cddf4efc62/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101022905-ba8b8f28a364 h1:UUDPWfUjyerptar+wPwI891jTUAtbbuXori0j3phTE4= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101022905-ba8b8f28a364/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911 h1:+82ygf2hiY5Mhwfmc1xfCe9e/KY/5ObBuE2LecM37qI= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101023317-8311ecb84911/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=