The assigner's verbs and the checker's tools ask the indexer's tools instead of reading its state (issue 1966, recovery) #176
@@ -293,6 +293,9 @@ func (s memStore) put(state, key string, value []byte) error {
|
||||
return nil
|
||||
}
|
||||
func (s memStore) keys(state string) ([]string, error) {
|
||||
if s[state] == nil { // as the runtime refuses a state the module neither keeps nor reads
|
||||
return nil, fmt.Errorf("mesh-delivery-assigner keeps and reads no state called %q", state)
|
||||
}
|
||||
var out []string
|
||||
for k := range s[state].values {
|
||||
out = append(out, k)
|
||||
@@ -362,9 +365,14 @@ func putOfTheBus(t *testing.T) {
|
||||
|
||||
buildKV := mem()
|
||||
buildKV.values[bkey] = built
|
||||
modules := mem()
|
||||
modules.values["nats"] = mod
|
||||
s := newPutTools(memStore{putsState: putsKV, builds.BuilderModule + ".builds": buildKV, buckets.IndexerModule + ".modules": modules})
|
||||
// Issue 1966: newest is asked of the indexer's first view of modules, never read from its bus state.
|
||||
indexer := &fakeIndexer{views: map[string]map[string][]byte{"mesh-delivery-indexer.modules": {"mesh-catalog": mod, "nats": mod}}}
|
||||
s := newPutTools(memStore{putsState: putsKV, builds.BuilderModule + ".builds": buildKV}, func(key string, body any) (json.RawMessage, error) {
|
||||
if raw, ok, err := indexer.answer(key, body); ok {
|
||||
return raw, err
|
||||
}
|
||||
return nil, fmt.Errorf("%s is not served here", key)
|
||||
})
|
||||
s.now = func() time.Time { return now }
|
||||
was := answerWithin
|
||||
answerWithin = 0
|
||||
@@ -375,8 +383,11 @@ func putOfTheBus(t *testing.T) {
|
||||
if got, err := s.version("nats", "fd497ab80000"); err != nil || got != fp {
|
||||
t.Errorf("the prefix of the fingerprint built reads %q, %v", got, err)
|
||||
}
|
||||
if got, err := s.version("nats", "newest"); err != nil || got != fp {
|
||||
t.Errorf("newest reads %q, %v", got, err)
|
||||
if got, err := s.version("nats", "newest"); err != nil || got != fp || indexer.asks["mesh-delivery-indexer.modules"] != 2 {
|
||||
t.Errorf("newest reads %q, %v, asking %d pages of the indexer's first view (want its 2)", got, err, indexer.asks["mesh-delivery-indexer.modules"])
|
||||
}
|
||||
if _, err := s.version("ghost", "newest"); err == nil || !strings.Contains(err.Error(), "holds no module ghost") {
|
||||
t.Errorf("newest of a module the indexer does not hold: %v", err)
|
||||
}
|
||||
ans, err := s.doPut(map[string]any{"node": "novox", "module": "nats", "version": fp, "hold": "true", "why": "the bus's user list"}, "shanks.node-tools")
|
||||
if err != nil {
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"maps"
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"slices"
|
||||
"sort"
|
||||
@@ -17,6 +19,7 @@ import (
|
||||
mmac "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
|
||||
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
plc "mesh-delivery-assigner/internal/assignment"
|
||||
@@ -42,12 +45,40 @@ func (runtimeStore) history(state, key string) ([]stdio.StateRevision, error) {
|
||||
}
|
||||
|
||||
// The states the verbs read, as the runtime names them: the seat's own by their local names, the others'
|
||||
// `<module>.<state>`.
|
||||
// `<module>.<state>` where the manifest declares them under `reads`.
|
||||
const haveState = "node-engine.have"
|
||||
|
||||
// The indexer's states the verbs read, each asked of the indexer's first-view tool of that name, never read from its bus
|
||||
// state (novox/hq ADR 0596; ADR 0615 §1, ADR 0621 §1, §2: the manifest invokes them, and the indexer answers them to the
|
||||
// assigner alone of the delivery's parts).
|
||||
const (
|
||||
modulesState = buckets.IndexerModule + ".modules"
|
||||
haveState = "node-engine.have"
|
||||
indexerModules = "modules"
|
||||
indexerManifests = "manifests"
|
||||
)
|
||||
|
||||
// indexerView is one of the indexer's states whole, asked of its first-view tool through ask (stdio.Ask, or a test's) page
|
||||
// after page, each page continued after the one before's next. It is asked again at every call: the indexer's state is
|
||||
// the truth, and the verbs keep no copy of it.
|
||||
func indexerView(ask func(string, any) (json.RawMessage, error), state string) (map[string]json.RawMessage, error) {
|
||||
tool := buckets.IndexerModule + "." + state
|
||||
if ask == nil {
|
||||
return nil, fmt.Errorf("%s cannot be asked here", tool)
|
||||
}
|
||||
page := views.AskVia(ask, tool)
|
||||
out := map[string]json.RawMessage{}
|
||||
for after := ""; ; {
|
||||
a, err := page(context.Background(), after)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s does not answer: %w", tool, err)
|
||||
}
|
||||
maps.Copy(out, a.Values)
|
||||
if a.Next == "" || a.Next == after {
|
||||
return out, nil
|
||||
}
|
||||
after = a.Next
|
||||
}
|
||||
}
|
||||
|
||||
// seatTools serves the seat's verbs over a store, asking the controller through ask (stdio.Ask, or a test's).
|
||||
type seatTools struct {
|
||||
st seatStore
|
||||
@@ -97,14 +128,15 @@ func (s *seatTools) assignments() (map[string]mma.Assignments, map[string]uint64
|
||||
return out, revs, nil
|
||||
}
|
||||
|
||||
// lays is what a proposal is checked against: the modules the indexer holds and the nodes with a record.
|
||||
// lays is what a proposal is checked against: the modules the indexer holds, as its first view of modules answers them,
|
||||
// and the nodes with a record.
|
||||
func (s *seatTools) lays() (func(string) bool, func(string) bool, error) {
|
||||
mods, err := s.st.keys(modulesState)
|
||||
mods, err := indexerView(s.ask, indexerModules)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("the indexer's modules cannot be listed: %w", err)
|
||||
}
|
||||
held := map[string]bool{}
|
||||
for _, k := range mods {
|
||||
for k := range mods {
|
||||
if name, err := meshcatalogclient.ParseModulesKey(k); err == nil {
|
||||
held[name] = true
|
||||
}
|
||||
|
||||
@@ -26,6 +26,7 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
@@ -52,11 +53,13 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
anchor := mma.Assignments{V: keys.Version, Node: "anchor", Modules: []mma.PlacedModule{{Module: "web"}, {Module: "mesh-enrolment"}},
|
||||
Holds: []mma.Holding{{Claim: "mesh-node-enroller", Scope: "mesh", Module: "mesh-enrolment"}}}
|
||||
st := memStore{}
|
||||
for _, state := range []string{mma.StateAssignments, mma.StateProposals, mma.StateCandidates, modulesState, haveState} {
|
||||
for _, state := range []string{mma.StateAssignments, mma.StateProposals, mma.StateCandidates, haveState} {
|
||||
st[state] = &memKV{values: map[string][]byte{}}
|
||||
}
|
||||
// Issue 1966: the indexer's modules are asked of its first view, never read from its bus state (ADR 0596, ADR 0621).
|
||||
indexer := &fakeIndexer{views: map[string]map[string][]byte{"mesh-delivery-indexer.modules": {}}}
|
||||
for _, m := range []string{"web", "shell", "mesh-enrolment"} {
|
||||
st[modulesState].values[m] = []byte("{}")
|
||||
indexer.views["mesh-delivery-indexer.modules"][m] = []byte("{}")
|
||||
}
|
||||
for _, n := range []string{"anchor", "laptop"} {
|
||||
raw, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: n}})
|
||||
@@ -65,6 +68,9 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
var asked []string
|
||||
s := newSeatTools(st, func(key string, body any) (json.RawMessage, error) {
|
||||
asked = append(asked, key)
|
||||
if raw, ok, err := indexer.answer(key, body); ok {
|
||||
return raw, err
|
||||
}
|
||||
return json.RawMessage(`{"ok":true}`), nil
|
||||
})
|
||||
s.now = func() time.Time { return time.Date(2026, 10, 28, 20, 0, 0, 0, time.UTC) }
|
||||
@@ -193,6 +199,32 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
if tools := s.tools(); len(tools) != 5 || tools[2].Name != "mesh-module-assigner.propose" || tools[2].Run == nil {
|
||||
t.Errorf("the seat's verbs as served: %+v", tools)
|
||||
}
|
||||
// Issue 1966: propose checks its modules against the indexer's first view of modules, every page of it, asked again at
|
||||
// each call; a first view the indexer refuses refuses the proposal, naming the tool, and writes nothing.
|
||||
for _, c := range []struct {
|
||||
name, module, refused, want string
|
||||
}{
|
||||
{"a module on the last page", "web", "", ""},
|
||||
{"a module the indexer does not hold", "ghost", "", "ghost"},
|
||||
{"the first view refused", "web", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.modules does not answer"},
|
||||
} {
|
||||
t.Run("1966: "+c.name, func(t *testing.T) {
|
||||
indexer.refused, indexer.asks = c.refused, nil
|
||||
maps.DeleteFunc(st[mma.StateAssignments].values, func(k string, _ []byte) bool { return strings.HasPrefix(k, "joiner.") }) // the remove-node's joiner
|
||||
st[mma.StateProposals].values, st[mma.StateCandidates].values = map[string][]byte{}, map[string][]byte{}
|
||||
_, err := s.doPropose(map[string]any{"why": "x", "changes": []any{map[string]any{"op": "place", "node": "laptop", "module": c.module}}}, "op")
|
||||
switch {
|
||||
case c.want == "" && err != nil:
|
||||
t.Errorf("propose: %v", err)
|
||||
case c.want != "" && (err == nil || !strings.Contains(err.Error(), c.want) || len(st[mma.StateProposals].values) != 0):
|
||||
t.Errorf("propose: %v, want a refusal saying %q and nothing written", err, c.want)
|
||||
}
|
||||
if pages := indexer.asks["mesh-delivery-indexer.modules"]; c.refused == "" && pages != 3 {
|
||||
t.Errorf("the first view of modules was asked %d pages, want its 3", pages)
|
||||
}
|
||||
})
|
||||
}
|
||||
indexer.refused = ""
|
||||
t.Run("1953/D4-D6: the seat mesh-wants' upgrade and bus", wantsVerbs)
|
||||
t.Run("1955/L7 (ADR 0571 §6): rule 7 reads the node's own composition; an unplaced node is said", func(t *testing.T) {
|
||||
w := newWorld()
|
||||
@@ -439,9 +471,11 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
func wantsVerbs(t *testing.T) {
|
||||
fpOld, fpNew := meshcatalogclient.SourcePrefix2+strings.Repeat("1", 64), meshcatalogclient.SourcePrefix2+strings.Repeat("2", 64)
|
||||
st := memStore{}
|
||||
for _, state := range []string{modulesState, manifestsState, policiesState, haveState, putsState, "findings", buckets.ComposedRead} {
|
||||
for _, state := range []string{policiesState, haveState, putsState, "findings", buckets.ComposedRead} {
|
||||
st[state] = &memKV{values: map[string][]byte{}}
|
||||
}
|
||||
// Issue 1966: the indexer's modules and manifests are asked of its first views, never read from its bus state.
|
||||
indexer := &fakeIndexer{views: map[string]map[string][]byte{"mesh-delivery-indexer.modules": {}, "mesh-delivery-indexer.manifests": {}}}
|
||||
// ADR 0593 §6, §8: what each node published, every node's bucket as one state: anchor's own composition and its web at
|
||||
// a fingerprint, and laptop's web refused there.
|
||||
partNew, _ := meshcatalogclient.KeyPart(fpNew)
|
||||
@@ -453,8 +487,8 @@ func wantsVerbs(t *testing.T) {
|
||||
manifest := func(name, extra string) {
|
||||
raw := `{"module":"` + name + `","version":"1"` + extra + `}`
|
||||
digest := meshcatalogclient.ManifestDigest([]byte(raw))
|
||||
st[manifestsState].values[digest], _ = meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{V: keys.Version, Module: name, Manifest: raw})
|
||||
st[modulesState].values[name], _ = meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{V: keys.Version, Fp2: fpNew, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
indexer.views["mesh-delivery-indexer.manifests"][digest], _ = meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{V: keys.Version, Module: name, Manifest: raw})
|
||||
indexer.views["mesh-delivery-indexer.modules"][name], _ = meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{V: keys.Version, Fp2: fpNew, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
}
|
||||
manifest("web", "")
|
||||
manifest("photos", `,"upgrade":{"policy":"record","why":"a person checks the albums"}`)
|
||||
@@ -465,10 +499,14 @@ func wantsVerbs(t *testing.T) {
|
||||
st[haveState].values["anchor.node"], _ = nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Composed: "sha256:" + strings.Repeat("c", 64)})
|
||||
var asked []string
|
||||
var said []string
|
||||
w := newWantsTools(st, func(key string, body any) (json.RawMessage, error) {
|
||||
ask := func(key string, body any) (json.RawMessage, error) {
|
||||
asked = append(asked, key)
|
||||
if raw, ok, err := indexer.answer(key, body); ok {
|
||||
return raw, err
|
||||
}
|
||||
return json.RawMessage(`{"ok":true}`), nil
|
||||
}, func(event, id string, body any) error { said = append(said, event+" "+id); return nil })
|
||||
}
|
||||
w := newWantsTools(st, ask, func(event, id string, body any) error { said = append(said, event+" "+id); return nil })
|
||||
policyOf := func(module string) policyRow {
|
||||
out, err := w.doUpgrade(map[string]any{"module": module}, "op")
|
||||
if err != nil {
|
||||
@@ -497,7 +535,7 @@ func wantsVerbs(t *testing.T) {
|
||||
// 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 := newWantsTools(st, ask, nil)
|
||||
byWatch.watched = stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error {
|
||||
watches[state]++
|
||||
for k, v := range st[state].values {
|
||||
@@ -518,10 +556,13 @@ func wantsVerbs(t *testing.T) {
|
||||
t.Errorf("compare by watch = %+v", got)
|
||||
}
|
||||
}
|
||||
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{"mesh-delivery-indexer.modules", "mesh-delivery-indexer.manifests"} {
|
||||
if watches[state] != 0 || watches[state+" by key"] != 0 || indexer.asks[state] == 0 {
|
||||
t.Errorf("the indexer's %s was watched %d times and read by key %d, asked of its first view %d times: want asked alone",
|
||||
state, watches[state], watches[state+" by key"], indexer.asks[state])
|
||||
}
|
||||
}
|
||||
for _, state := range []string{modulesState, haveState, mcc.FindingsState} {
|
||||
for _, state := range []string{haveState, mcc.FindingsState} {
|
||||
if watches[state] != 1 {
|
||||
t.Errorf("%s was watched %d times, want once", state, watches[state])
|
||||
}
|
||||
@@ -742,3 +783,42 @@ func (k *copiesKV) Put(_ context.Context, key string, value []byte) (uint64, err
|
||||
k.values[key] = value
|
||||
return uint64(len(k.values)), nil
|
||||
}
|
||||
|
||||
// fakeIndexer answers the indexer's first-view tools as its runtime does (views.FirstViewTool): each tool its state's values
|
||||
// in key order, one key a page, each page continued after the page before's next; the tool named refused answers its
|
||||
// refusal. asks counts the pages asked by tool.
|
||||
type fakeIndexer struct {
|
||||
views map[string]map[string][]byte
|
||||
refused string
|
||||
asks map[string]int
|
||||
}
|
||||
|
||||
// answer is the answer to a call of key, and whether key is one of the indexer's first views at all.
|
||||
func (f *fakeIndexer) answer(key string, body any) (json.RawMessage, bool, error) {
|
||||
values, ok := f.views[key]
|
||||
if !ok {
|
||||
return nil, false, nil
|
||||
}
|
||||
if f.asks == nil {
|
||||
f.asks = map[string]int{}
|
||||
}
|
||||
f.asks[key]++
|
||||
if key == f.refused {
|
||||
return nil, true, fmt.Errorf("%s is refused to \"novox.mesh-delivery-assigner\"", key)
|
||||
}
|
||||
after, _ := body.(map[string]any)["after"].(string)
|
||||
a := views.Answer{V: views.Version, Owner: buckets.IndexerModule, State: strings.TrimPrefix(key, buckets.IndexerModule+"."), Revision: 7,
|
||||
Values: map[string]json.RawMessage{}}
|
||||
for _, k := range slices.Sorted(maps.Keys(values)) {
|
||||
if k <= after {
|
||||
continue
|
||||
}
|
||||
if len(a.Values) == 1 {
|
||||
a.Next = slices.Collect(maps.Keys(a.Values))[0]
|
||||
break
|
||||
}
|
||||
a.Values[k] = values[k]
|
||||
}
|
||||
raw, err := json.Marshal(a)
|
||||
return raw, true, err
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"mesh-delivery-assigner/internal/puts"
|
||||
"os"
|
||||
"regexp"
|
||||
@@ -45,7 +44,9 @@ type store interface {
|
||||
|
||||
// putTools serves the tools over a store, and keeps the newest value of every put it heard, to wait on.
|
||||
type putTools struct {
|
||||
st store
|
||||
st store
|
||||
// ask asks a module's tool (stdio.Ask, or a test's): the indexer's first view of modules for newest.
|
||||
ask func(key string, body any) (json.RawMessage, error)
|
||||
now func() time.Time
|
||||
mu sync.Mutex
|
||||
// heard is every put the watch handed, by key; changed is closed and replaced at each.
|
||||
@@ -53,8 +54,8 @@ type putTools struct {
|
||||
changed chan struct{}
|
||||
}
|
||||
|
||||
func newPutTools(st store) *putTools {
|
||||
return &putTools{st: st, now: time.Now, heard: map[string][]byte{}, changed: make(chan struct{})}
|
||||
func newPutTools(st store, ask func(string, any) (json.RawMessage, error)) *putTools {
|
||||
return &putTools{st: st, ask: ask, now: time.Now, heard: map[string][]byte{}, changed: make(chan struct{})}
|
||||
}
|
||||
|
||||
// hear takes a put's change in from the watch.
|
||||
@@ -78,10 +79,12 @@ func (s *putTools) version(module, version string) (string, error) {
|
||||
v := strings.ToLower(strings.TrimSpace(version))
|
||||
switch {
|
||||
case v == "newest":
|
||||
raw, _, found, err := s.st.get(buckets.ModulesRead, module)
|
||||
// Asked of the indexer's first view of modules, never read from its bus state (ADR 0596, ADR 0621; issue 1966).
|
||||
mods, err := indexerView(s.ask, indexerModules)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the indexer's modules cannot be read: %w", err)
|
||||
}
|
||||
raw, found := mods[module]
|
||||
if !found {
|
||||
return "", fmt.Errorf("the indexer holds no module %s at the trunk", module)
|
||||
}
|
||||
@@ -345,7 +348,7 @@ func (runtimeStore) keys(state string) ([]string, error) { return stdio.State(st
|
||||
|
||||
// serveTools serves the tools to the runtime until it stops, hearing every put to answer assigner_put with its first steps.
|
||||
func serveTools() error {
|
||||
s := newPutTools(runtimeStore{})
|
||||
s := newPutTools(runtimeStore{}, stdio.Ask)
|
||||
go func() {
|
||||
<-stdio.Served()
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
|
||||
@@ -32,8 +32,7 @@ import (
|
||||
|
||||
// The states the verbs read, as the runtime names them.
|
||||
const (
|
||||
policiesState = buckets.PoliciesState
|
||||
manifestsState = buckets.IndexerModule + ".manifests"
|
||||
policiesState = buckets.PoliciesState
|
||||
// PolicyChangedEvent is the seat's event of a person's policy set or taken back.
|
||||
PolicyChangedEvent = "policy-changed"
|
||||
// busStepCause is the cause the bus's step is recorded under in the hand-act log.
|
||||
@@ -88,19 +87,14 @@ func (s *wantsTools) direct(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.
|
||||
// manifests is every module the indexer holds, with its trunk manifest and fingerprint: both asked of the indexer's first
|
||||
// views of modules and manifests (indexerView), never read from its bus state.
|
||||
func (s *wantsTools) manifests() (map[string]catalogue.Manifest, map[string]meshcatalog.ModulesByModuleValue, error) {
|
||||
mods, err := s.values(modulesState)
|
||||
mods, err := indexerView(s.ask, indexerModules)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
kept, err := indexerView(s.ask, indexerManifests)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
@@ -114,16 +108,16 @@ func (s *wantsTools) manifests() (map[string]catalogue.Manifest, map[string]mesh
|
||||
if m.Manifest == "" {
|
||||
continue
|
||||
}
|
||||
mraw, found, err := s.manifest(m.Manifest)
|
||||
if err != nil || !found {
|
||||
mraw, found := kept[m.Manifest]
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
kept, err := meshcatalogclient.DecodeManifest(mraw)
|
||||
stored, err := meshcatalogclient.DecodeManifest(mraw)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
// Decoded, never judged (ADR 0578 §3); one that does not decode is left out of the answer, as before.
|
||||
if man, err := catalogue.Decode([]byte(kept.Manifest)); err == nil {
|
||||
if man, err := catalogue.Decode([]byte(stored.Manifest)); err == nil {
|
||||
out[name] = man
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
// Command mesh-delivery-checker-tools serves the checker's read tools (novox/hq ADR 0484 clause 13, issue 1686): the runtime
|
||||
// launches it for a call, it reads the checker's verdicts and kept states and the indexer's candidates through the checker's
|
||||
// state and reads. It writes only the tries (the try's tools) and, asking a check again by hand, one verdict. The checker is the other program: it serves no tool (ADR 0437 §4).
|
||||
// launches it for a call, it reads the checker's verdicts and kept states through the checker's state and reads, and the
|
||||
// indexer's candidates and modules through the indexer's read tools. It writes only the tries (the try's tools) and, asking a check again by hand, one verdict. The checker is the other program: it serves no tool (ADR 0437 §4).
|
||||
package main
|
||||
|
||||
import (
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
@@ -58,6 +59,52 @@ func read(s stdio.KeptState) check.StateRead {
|
||||
}}
|
||||
}
|
||||
|
||||
// ownerRead is a state of another module read through that module's ordinary read tool (views.ReadTool), asked through ask
|
||||
// (stdio.Ask, or a test's): its keys with no argument, one key's value with key. Another module's data is asked through its
|
||||
// tools, never read from its bus state (novox/hq ADR 0596); the manifest invokes the tool, and the owner answers every reader
|
||||
// that invokes it. Each read is asked when it is read: the owner's state is the truth, and the tools keep no copy.
|
||||
func ownerRead(ask func(string, any) (json.RawMessage, error), tool string) check.StateRead {
|
||||
call := func(args map[string]any, into any) error {
|
||||
reads.Add(1)
|
||||
raw, err := ask(tool, args)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s does not answer: %w", tool, err)
|
||||
}
|
||||
answer, said, ok := buscalls.Unwrap(raw)
|
||||
if !ok {
|
||||
return fmt.Errorf("%s answered: %s", tool, said)
|
||||
}
|
||||
if err := json.Unmarshal(answer, into); err != nil {
|
||||
return fmt.Errorf("%s answered what does not decode: %w", tool, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return check.StateRead{
|
||||
Keys: func() ([]string, error) {
|
||||
var a views.ReadAnswer
|
||||
if err := call(map[string]any{}, &a); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.Keys, nil
|
||||
},
|
||||
Get: func(key string) (json.RawMessage, error) {
|
||||
var a views.KeyAnswer
|
||||
if err := call(map[string]any{"key": key}, &a); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !a.Found {
|
||||
return nil, nil
|
||||
}
|
||||
return a.Value, nil
|
||||
}}
|
||||
}
|
||||
|
||||
// The indexer's read tools the tools ask for its candidates and modules (the manifest's invokes).
|
||||
const (
|
||||
indexerCandidatesTool = check.IndexerModule + ".indexer_read_candidates"
|
||||
indexerModulesTool = check.IndexerModule + ".indexer_read_modules"
|
||||
)
|
||||
|
||||
// runsOf says which commit the mesh runs of a module from the indexer's modules state, as the checker's own record of it does
|
||||
// (the commit the module's record was computed at); "" when the record is not readable.
|
||||
type runsOf struct{ modules check.StateRead }
|
||||
@@ -371,13 +418,13 @@ func stepBound() time.Duration {
|
||||
}
|
||||
|
||||
func main() {
|
||||
r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: read(stdio.State("mesh-delivery-indexer.candidates")),
|
||||
r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: ownerRead(stdio.Ask, indexerCandidatesTool),
|
||||
PreviewState: read(stdio.State("previews")), InputState: read(stdio.State(check.InputsStateName)),
|
||||
StepStates: map[string]check.StateRead{"mesh-build-go": read(stdio.State("mesh-build-go.steps")), "mesh-build-node": read(stdio.State("mesh-build-node.steps"))},
|
||||
HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound(),
|
||||
TurnState: read(stdio.State("mesh-delivery-gate.turns")), HearingState: read(stdio.State(buildrun.HearingState)),
|
||||
AskState: read(stdio.State(check.AsksState)), BuildState: read(stdio.State("mesh-delivery-builder.builds"))}
|
||||
r.Runs = runsOf{read(stdio.State("mesh-delivery-indexer.modules"))} // the node-engine at the commit the mesh runs, as the checker reads it
|
||||
r.Runs = runsOf{ownerRead(stdio.Ask, indexerModulesTool)} // the node-engine at the commit the mesh runs, as the checker reads it
|
||||
tries := stdio.State(check.TriesState)
|
||||
source := stdio.State("mesh-delivery-gate.source")
|
||||
trunks := func() (map[string]string, error) {
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"os"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -67,6 +68,73 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
|
||||
if _, err := RequestTools(check.StateRead{})[0].Run(map[string]any{}); err == nil {
|
||||
t.Errorf("answered with no state to read")
|
||||
}
|
||||
// Issue 1966: another module's state is asked of its read tool, never read from its bus state (ADR 0596). Every
|
||||
// stdio.State this program names of another module is one its manifest declares under reads, which the runtime grants;
|
||||
// the indexer's candidates and modules are asked of indexer_read_candidates and indexer_read_modules, which it invokes.
|
||||
t.Run("1966: another module's state is read only where the manifest declares it", func(t *testing.T) {
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var man struct {
|
||||
Reads []string `json:"reads"`
|
||||
Invokes []string `json:"invokes"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &man); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
src, err := os.ReadFile("main.go")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, m := range regexp.MustCompile(`stdio\.State\("([^"]+\.[^"]+)"\)`).FindAllStringSubmatch(string(src), -1) {
|
||||
if !slices.Contains(man.Reads, m[1]) {
|
||||
t.Errorf("main.go reads %s, which the manifest does not declare under reads: the runtime refuses it", m[1])
|
||||
}
|
||||
}
|
||||
for _, tool := range []string{indexerCandidatesTool, indexerModulesTool} {
|
||||
if !slices.Contains(man.Invokes, tool) {
|
||||
t.Errorf("the manifest does not invoke %s", tool)
|
||||
}
|
||||
}
|
||||
})
|
||||
for _, c := range []struct {
|
||||
name, key, answer, want string
|
||||
failed bool
|
||||
}{
|
||||
{name: "the keys", answer: `{"owner":"mesh-delivery-indexer","state":"candidates","keys":["a","b"]}`, want: "[a b]"},
|
||||
{name: "one key, wrapped as the tool runner hands it", key: "a",
|
||||
answer: `{"content":[{"type":"text","text":"{\"key\":\"a\",\"found\":true,\"value\":{\"commit\":\"c\"}}"}]}`, want: `{"commit":"c"}`},
|
||||
{name: "a key the indexer does not hold", key: "gone", answer: `{"key":"gone","found":false}`, want: ""},
|
||||
{name: "the tool refused", key: "a", answer: `{"content":[{"type":"text","text":"refused"}],"isError":true}`, failed: true},
|
||||
} {
|
||||
t.Run("1966: the indexer's read tool, "+c.name, func(t *testing.T) {
|
||||
var asked []string
|
||||
r := ownerRead(func(key string, body any) (json.RawMessage, error) {
|
||||
b, _ := json.Marshal(body)
|
||||
asked = append(asked, key+" "+string(b))
|
||||
return json.RawMessage(c.answer), nil
|
||||
}, indexerCandidatesTool)
|
||||
var got string
|
||||
var err error
|
||||
if c.key == "" {
|
||||
var keys []string
|
||||
keys, err = r.Keys()
|
||||
got = strings.Trim(strings.Join([]string{"[", strings.Join(keys, " "), "]"}, ""), " ")
|
||||
} else {
|
||||
var raw json.RawMessage
|
||||
raw, err = r.Get(c.key)
|
||||
got = string(raw)
|
||||
}
|
||||
want := []string{indexerCandidatesTool + " {}"}
|
||||
if c.key != "" {
|
||||
want = []string{indexerCandidatesTool + ` {"key":"` + c.key + `"}`}
|
||||
}
|
||||
if (err != nil) != c.failed || (!c.failed && got != c.want) || !slices.Equal(asked, want) {
|
||||
t.Errorf("read %q, %v, asking %v; want %q, failed %v, asking %v", got, err, asked, c.want, c.failed, want)
|
||||
}
|
||||
})
|
||||
}
|
||||
// With nothing to read from a tool says so and changes nothing.
|
||||
for _, tl := range Tools(check.Reader{}) {
|
||||
if tl.Name == check.StatusReadTool {
|
||||
|
||||
Reference in New Issue
Block a user