Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b8d4a77e75 | ||
|
|
f95d5b4d23 |
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -15,6 +16,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 +402,47 @@ 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.watched = stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error {
|
||||
watches[state]++
|
||||
for k, v := range st[state].values {
|
||||
_ = 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 {
|
||||
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)
|
||||
}
|
||||
}
|
||||
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)
|
||||
}
|
||||
// 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{
|
||||
|
||||
@@ -352,10 +352,12 @@ 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.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)
|
||||
}
|
||||
|
||||
|
||||
@@ -41,6 +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
|
||||
// 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.
|
||||
@@ -59,8 +63,16 @@ func (s *wantsTools) leased() bool {
|
||||
return away
|
||||
}
|
||||
|
||||
// values reads every key of a state.
|
||||
// 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.watched != nil {
|
||||
return s.watched.Values(state)
|
||||
}
|
||||
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 +90,16 @@ func (s *wantsTools) values(state string) (map[string][]byte, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
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.
|
||||
func (s *wantsTools) manifests() (map[string]catalogue.Manifest, map[string]delivery.Module, error) {
|
||||
mods, err := s.values(modulesState)
|
||||
@@ -94,7 +116,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.manifest(m.Manifest)
|
||||
if err != nil || !found {
|
||||
continue
|
||||
}
|
||||
@@ -126,7 +148,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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
Reference in New Issue
Block a user