novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

mesh-wants upgrade reads through mesh-sdk's watched reader, keeping no mirror of its own

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-01 03:34:21 +01:00
parent f95d5b4d23
commit b8d4a77e75
5 changed files with 40 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go
+18 -4
View File
@@ -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)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go
+1 -6
View File
@@ -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)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go
+16 -66
View File
@@ -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
}
modules/mesh-delivery-assigner/go.mod
+1 -1
View File
@@ -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
)
modules/mesh-delivery-assigner/go.sum
+4
View File
@@ -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=