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

The assigner's seat verbs ask the indexer's first views instead of reading its state (issue 1966, recovery)

Browse Source 
mesh-module-assigner.propose and mesh-wants' upgrade and bus read the indexer's
modules and manifests straight from its bus state, which the runtime refuses
since those reads moved to the indexer's first-view tools (ADR 0615 §1, ADR 0621):
"mesh-delivery-assigner keeps and reads no state called mesh-delivery-indexer.modules".
Another module's data is asked through its tools, never read from its state
(ADR 0596). The verbs now ask mesh-delivery-indexer.modules and .manifests through
the tool runner, every page, at each call; the manifest already invokes both and
the indexer answers them to the assigner.

Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5
This commit is contained in:
jschoubben
2026-11-08 11:54:37 +01:00
parent 7f41218552
commit 0c4660eea8
4 changed files with 144 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+3
View File
@@ -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)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go
+38 -6
View File
@@ -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
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go
+91 -11
View File
@@ -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
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go
+12 -18
View File
@@ -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
}
}