From a24f7809a6ddb6238e5049afc0b7fe49cad0951b Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 12:28:45 +0100 Subject: [PATCH 1/9] COND6 (partial, not landed): the keeper's sources, holder and clears from the catalog half and the assignment half MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fixed Observers list and the engineModule constant are removed; sources are the modules whose catalog entry declares the finding events, a module in no branch's revision is cleared (module left the catalog), a stopped module no longer placed is cleared (module left the node), Keeper.Holder is derived from the entry's claims. The per-branch catalog followers in main.go wait on a design question (how a part adopts one catalog view per registered branch); the whole integration test is red until then. Builds: design 47 G §365, §196, §200, §364, §384 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/catalog.go | 231 ++++++++++++++++++ .../cmd/mesh-conditions/events.go | 10 +- .../cmd/mesh-conditions/keeper.go | 28 +-- .../cmd/mesh-conditions/keeper_test.go | 119 +++++---- .../cmd/mesh-conditions/main.go | 10 +- .../cmd/mesh-conditions/manifest_test.go | 4 +- modules/mesh-conditions/go.mod | 2 + modules/mesh-conditions/module.json | 7 +- 8 files changed, 329 insertions(+), 82 deletions(-) create mode 100644 modules/mesh-conditions/cmd/mesh-conditions/catalog.go diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go new file mode 100644 index 000000000..2b176aa2d --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -0,0 +1,231 @@ +package main + +// The keeper finds its sources in the catalog (design 47 G §365): the modules whose catalog entry declares the finding +// events of §383, in any watched branch's newest revision. A module in no branch's revision has its conditions cleared with +// `module left the catalog`; a module's conditions on a node are cleared with `module left the node` once the node-engine +// says the module stopped there and the newest assignment graph revision no longer places it. No module is named here (§7). + +import ( + "encoding/json" + "slices" + "sort" + "strings" + + catseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + wants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + + engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" +) + +// The reasons a catalog or an assignment takes a module's conditions away (design 47 G §365). +const ( + whyLeftCatalog = "module left the catalog" + whyLeftNode = "module left the node" +) + +// declaresFindings says whether the entry declares the finding events of §383 in its streams. +func declaresFindings(e catseat.CatalogEntry) bool { + if slices.ContainsFunc(e.Streams, func(s string) bool { return strings.Contains(s, EventFindingWritten) }) { + return true + } + streams, _ := e.Manifest["streams"].([]any) + for _, s := range streams { + m, _ := s.(map[string]any) + events, _ := m["events"].([]any) + for _, ev := range events { + if name, _ := ev.(string); strings.HasPrefix(name, EventFindingWritten) { + return true + } + } + } + return false +} + +// CatalogHeld takes the newest catalog revision of one watched branch: the sources are the modules declaring the finding +// events in any branch, and the conditions of a module in no branch are cleared. +func (k *Keeper) CatalogHeld(branch string, entries map[string]catseat.CatalogEntry) error { + k.mu.Lock() + k.init() + k.halves[branch] = entries + var gone []string + for observer := range k.findings { + if observer == Module { + continue + } + if _, ok := k.entry(observer); !ok { + gone = append(gone, observer) + } + } + k.mu.Unlock() + sort.Strings(gone) + var first error + for _, observer := range gone { + k.clearObserver(observer, "", whyLeftCatalog, &first) + } + return first +} + +// entry is the observer's entry in the first branch (in name order) that holds one. Called with mu held. +func (k *Keeper) entry(observer string) (catseat.CatalogEntry, bool) { + branches := make([]string, 0, len(k.halves)) + for b := range k.halves { + branches = append(branches, b) + } + sort.Strings(branches) + for _, b := range branches { + if e, ok := k.halves[b][observer]; ok { + return e, true + } + } + return catseat.CatalogEntry{}, false +} + +// isSource says whether the keeper takes findings from the module: its catalog entry declares the finding events. +func (k *Keeper) isSource(observer string) bool { + k.mu.Lock() + defer k.mu.Unlock() + k.init() + branches := make([]string, 0, len(k.halves)) + for b := range k.halves { + branches = append(branches, b) + } + for _, b := range branches { + if e, ok := k.halves[b][observer]; ok && declaresFindings(e) { + return true + } + } + return false +} + +// catalogHolder is Keeper.Holder: an observer whose catalog entry claims only seats held once for the mesh is held on the +// node the assignments name as the holder of one of them (design 47 G §384); unknown, it is taken from every node. +func (k *Keeper) catalogHolder(observer string) (string, bool) { + k.mu.Lock() + defer k.mu.Unlock() + k.init() + e, ok := k.entry(observer) + if !ok { + return "", false + } + claims, _ := e.Manifest["claims"].([]any) + if len(claims) == 0 { + return "", false + } + for _, c := range claims { + m, _ := c.(map[string]any) + if scope, _ := m["scope"].(string); scope != "mesh" { + return "", false + } + } + nodes := make([]string, 0, 1) + for hk, module := range k.holders { + if module == observer { + node, _, _ := strings.Cut(hk, ".") + nodes = append(nodes, node) + } + } + if len(nodes) == 0 { + return "", false + } + sort.Strings(nodes) + return nodes[0], true +} + +// AssignmentHeld takes the newest assignment graph revision: the modules it places on each node. +func (k *Keeper) AssignmentHeld(rev wants.AssignmentRevision) error { + k.mu.Lock() + k.init() + k.placed = map[string]bool{} + for _, e := range rev.Entries { + k.placed[e.Node+"."+e.Module] = true + } + k.placedHeld = true + k.mu.Unlock() + return k.sweepStopped() +} + +// EngineRunning takes a node-engine's `running` for a module on a node: stopped, it is kept until the assignment graph +// places the module there no longer. +func (k *Keeper) EngineRunning(node string, body []byte) error { + var r engine.Running + if err := json.Unmarshal(body, &r); err != nil || r.Module == "" || r.Node == "" { + k.Logf("the node-engine's running on %s is not readable: %v", node, err) + return nil + } + if node != "" && r.Node != node { + k.Logf("the node-engine's running on %s names the node %s: refused", node, r.Node) + return nil + } + k.mu.Lock() + k.init() + if r.Stopped { + k.stopped[r.Node+"."+r.Module] = true + } else { + delete(k.stopped, r.Node+"."+r.Module) + } + k.mu.Unlock() + return k.sweepStopped() +} + +// sweepStopped clears the conditions of each module stopped on a node that the held assignment revision does not place +// there; before an assignment revision is held, nothing is cleared. +func (k *Keeper) sweepStopped() error { + k.mu.Lock() + k.init() + var left []string + if k.placedHeld { + for pair := range k.stopped { + if !k.placed[pair] { + left = append(left, pair) + } + } + } + for _, pair := range left { + delete(k.stopped, pair) + } + k.mu.Unlock() + sort.Strings(left) + var first error + for _, pair := range left { + node, module, _ := strings.Cut(pair, ".") + k.clearObserver(module, node, whyLeftNode, &first) + } + return first +} + +// clearObserver clears every condition the module has, or has from the node when node is not empty, with the reason, and +// forgets what it heard of the module there. +func (k *Keeper) clearObserver(observer, node, why string, first *error) { + k.mu.Lock() + var keys []string + for key := range k.findings[observer] { + if node == "" || k.keyNode[observer][key] == node { + keys = append(keys, key) + } + } + for name, v := range k.views { + if v.observer == observer && (node == "" || v.node == node) { + for key := range v.Pending { + _ = k.Store.Delete(BoundsState, pendingKey(observer, v.node, key)) + } + for ask := range v.Asks { + _ = k.Store.Delete(BoundsState, askedKey(observer, v.node, ask)) + } + delete(k.views, name) + } + } + k.mu.Unlock() + sort.Strings(keys) + for _, key := range keys { + k.mu.Lock() + k.whys[key] = why + k.mu.Unlock() + if err := k.Finding(observer, key, nil, true); err != nil && *first == nil { + *first = err + } + k.mu.Lock() + delete(k.whys, key) + delete(k.keyNode[observer], key) + k.mu.Unlock() + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/events.go b/modules/mesh-conditions/cmd/mesh-conditions/events.go index 4520787dc..ec7abe5d8 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/events.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/events.go @@ -14,6 +14,7 @@ import ( "strings" "time" + engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" watchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" @@ -47,15 +48,15 @@ func (k *Keeper) FindingEvent(e stdio.Envelope) error { if _, ok := emitterOf(e.Key, watchdogs.EventMachineBack); ok && strings.HasPrefix(e.Key, watchdogs.Seat+".") { return k.MachinePresence(e.Body, false) } + if e.Key == engine.Seat+"."+engine.EventRunning { + return k.EngineRunning(e.Node, e.Body) + } if observer, ok := emitterOf(e.Key, EventFindingsStated); ok { return k.FindingsStated(observer, e.Node, e.Body) } return nil } -// isSource says whether the keeper takes findings from the module. -func (k *Keeper) isSource(observer string) bool { return slices.Contains(Observers, observer) } - // FindingWritten raises or changes the condition of the finding an observer says on a node. A body that is not a finding // the keeper can raise, or names another observer, is said in the journal and changes nothing. func (k *Keeper) FindingWritten(observer, node string, body []byte) error { @@ -199,7 +200,8 @@ func (k *Keeper) MachinePresence(body []byte, away bool) error { // hear subscribes to the observers' events once the runtime serves, asked again until it takes each. func hear(k *Keeper) { patterns := []string{"*." + EventFindingWritten, "*." + EventFindingCleared, "*." + EventFindingsStated, - watchdogs.Seat + "." + watchdogs.EventMachineAway, watchdogs.Seat + "." + watchdogs.EventMachineBack} + watchdogs.Seat + "." + watchdogs.EventMachineAway, watchdogs.Seat + "." + watchdogs.EventMachineBack, + engine.Seat + "." + engine.EventRunning} for _, pattern := range patterns { go func() { <-stdio.Served() diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index ae523a835..6753dffc2 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -25,6 +25,7 @@ import ( cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" meshhealers "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers" + catseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" ) // The modules whose states the keeper reads by name: the holders of the health seats (their seats' names are the @@ -35,8 +36,6 @@ const ( healersModule = "mesh-healers" // The holders of the assignments and the catalog, whose views and findings the keeper follows (the manifest test holds // these equal to the manifest's `consumes` and `invokes`). - // engineModule is the node-engine, an observer of §387 (design 47 G §387 names mesh-host). - engineModule = "mesh-host" assignerModule = "mesh-delivery-assigner" indexerModule = "mesh-delivery-indexer" ) @@ -55,23 +54,6 @@ const ( // HistoryKeptFor is how long a transition is kept (clause 3). const HistoryKeptFor = 90 * 24 * time.Hour -// Observers are the modules whose `findings` the keeper reads (clause 2; ADR 0565 §8 for the assigner's; the lead's -// table of 2026-10-29 for the provisioner's and the router's; ADR 0571 clause 2 for the seat artifact-store's holder, -// which observes S25, image-missing). -var Observers = []string{watchmanModule, selfCheckModule, healersModule, assignerModule, - "mesh-provisioner", "messenger", artifactStore, configModule, vaultModule, - // The readers of the indexer's and the assigner's state say a first view unanswered past its bound in their own findings - // (ADR 0615 §6). - "mesh-watcher", "minio", "nfs-server", gitea, "mesh-build-go", - // The builder's, the indexer's (asked of its read tool and followed by its views events, ADR 0615 §1) and the other - // readers' of the indexer's and the assigner's state (the registry milestone's join). - builder, indexer, "mesh-delivery-checker", "mesh-delivery-teller", "mesh-delivery-view", enrolment, - // The tool runner's and claude-code's own first views (views.UnansweredFinding in their findings, ADR 0615 §6). - "node-tools", "claude-code", - // The node-engine's: a node that has read no seat list five minutes after its start, seat-list-unread (ADR 0636 §1; - // issue 1966: written on ace and raised by nobody). - engineModule} - // The other states it reads, as `.`. const ( AttemptsRead = healersModule + "." + meshhealers.StateAttempts @@ -158,6 +140,13 @@ type Keeper struct { Record func(what, why, verb string) error // away are the machines the watchman says are away (design 47 G §389). away map[string]bool + // halves are the newest catalog revisions of the watched branches, by branch and module (catalog.go); placed are the + // `.` the newest assignment graph revision places, placedHeld whether one is held; stopped are the + // `.` the node-engines say stopped. + halves map[string]map[string]catseat.CatalogEntry + placed map[string]bool + placedHeld bool + stopped map[string]bool } func (k *Keeper) init() { @@ -168,6 +157,7 @@ func (k *Keeper) init() { map[string]string{}, map[string]held{}, map[string]clearedMemo{} k.passed, k.outOfHold, k.holders = map[string]string{}, map[string]string{}, map[string]string{} k.assignments = assignerclient.NewKeys() + k.halves, k.placed, k.stopped = map[string]map[string]catseat.CatalogEntry{}, map[string]bool{}, map[string]bool{} k.views, k.keyNode, k.whys = map[string]*sourceView{}, map[string]map[string]string{}, map[string]string{} } if k.Now == nil { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index f8206d4d4..446b155fd 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -3,10 +3,9 @@ package main import ( "encoding/json" "fmt" - "io/fs" - "os" - "path/filepath" - "regexp" + nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + catseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "slices" "sort" "strings" @@ -112,9 +111,27 @@ func newRig() *rig { r.said = append(r.said, e) return nil }} + // The catalog half of the watched branch: the modules the tests hear findings from declare the finding events (§365); + // the self-check's holder claims a seat held once for the mesh. + r.half(watchmanModule, "node") + r.half(selfCheckModule, "mesh") return r } +// half puts a module in the watched branch's catalog half, declaring the finding events and claiming one seat of scope. +func (r *rig) half(module, scope string) { + r.k.mu.Lock() + r.k.init() + if r.k.halves["main"] == nil { + r.k.halves["main"] = map[string]catseat.CatalogEntry{} + } + r.k.halves["main"][module] = catseat.CatalogEntry{Module: module, Manifest: map[string]any{ + "claims": []any{map[string]any{"name": module, "scope": scope}}, + "streams": []any{map[string]any{"name": module + "-findings", "events": []any{"finding-written.*", "finding-cleared.*", "findings-stated.*"}}}, + }} + r.k.mu.Unlock() +} + func (r *rig) changes() string { var out []string for _, e := range r.said { @@ -346,10 +363,6 @@ func TestEveryControllerKindHasItsHome(t *testing.T) { if h, ok := Homes[kind]; !ok || h.Observer != observer || h.Carried { t.Errorf("%s's home is %+v", kind, h) } - // The keeper reads every observer's findings that a kind's home names (the registry milestone's join). - if !slices.Contains(Observers, observer) { - t.Errorf("%s's observer %s is not read by the keeper", kind, observer) - } } for _, kind := range []string{"resolver-wrong", "holder-silent", "archives-missing", "archives-unheld", "binding-kept", "binding-moved", "binding-moving", "grant-composes-nothing", "root-not-free", "cleanup-waiting", "data-held-twice"} { @@ -370,18 +383,6 @@ func TestEveryControllerKindHasItsHome(t *testing.T) { t.Errorf("%s's home is %+v", kind, h) } } - // Each module that writes a reader's unanswered first view (ADR 0615 §6), or that a delivery part's consumer stopped - // hearing, is read: the delivery parts and the tool runner by name (other repositories), and every catalogue module - // whose code writes a views finding, found in the catalogue's tree, so a new writer cannot go unobserved. - // And the node-engine, whose node writes seat-list-unread when it has read no seat list (ADR 0636 §1, issue 1966). - readers := []string{watchman, "mesh-delivery-teller", enrolment, builder, "mesh-delivery-checker", "mesh-delivery-view", - indexer, assignerModule, configModule, vaultModule, gitea, "node-tools", "mesh-host"} - readers = append(readers, viewFindingWriters(t, "../../..")...) - for _, reader := range readers { - if !slices.Contains(Observers, reader) { - t.Errorf("the reader %s's findings are not read by the keeper", reader) - } - } // 19 of ADR 0564, the 63 of the table, plan-failed and stalled, mesh-config's four (design 32), the readers' unanswered // first view, the owners' view too large, the builder's four, the indexer's four and the package seats' token refused. // And the builder's three publishing kinds and the delivery parts' family not heard: 103. @@ -393,40 +394,6 @@ func TestEveryControllerKindHasItsHome(t *testing.T) { } } -// viewFindingWriters are the catalogue modules (under modules) whose non-test Go code writes a views finding into its own -// findings: views.UnansweredFinding, views.TooLargeFinding, views.UnsaidFinding or views.OwnFindings. -func viewFindingWriters(t *testing.T, modules string) []string { - t.Helper() - writes := regexp.MustCompile(`(views\.(UnansweredFinding|TooLargeFinding|UnsaidFinding)|meshconditionsclient\.FindingWriter)\b`) - var out []string - entries, err := os.ReadDir(modules) - if err != nil { - t.Fatal(err) - } - for _, e := range entries { - if !e.IsDir() || e.Name() == Module { - continue - } - found := false - _ = filepath.WalkDir(filepath.Join(modules, e.Name()), func(path string, d fs.DirEntry, err error) error { - if err != nil || found || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { - return nil - } - if src, err := os.ReadFile(path); err == nil && writes.Match(src) { - found = true - } - return nil - }) - if found { - out = append(out, e.Name()) - } - } - if len(out) < 5 { - t.Fatalf("found %d catalogue modules writing views findings in %s: the tree was not read (%v)", len(out), modules, out) - } - return out -} - // ADR 0634 §1, §4 (issue 1966): the keeper reads a condition written before the seat named its layout (the controller's // body: `silenced` null, `actions` empty, the times with an offset) and writes it back in the seat's layout. func TestTheKeeperReadsTheOldLayoutAndWritesTheSeatsLayout(t *testing.T) { @@ -569,6 +536,50 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { } return nil }, []string{"machine.d.silent", "machine.g.silent"}, ""}, + // Design 47 G §365: the sources are the catalog's, a module that left the catalog or the node loses its conditions. + {"a module not declaring the finding events is no source, one the catalog half declares is heard with no code change", func() error { + if err := r.k.FindingWritten("fixture", "ace", mustJSON(t, f("fixture", "z"))); err != nil { + return err + } + if _, ok := r.open("machine.z.silent"); ok { + t.Errorf("a module in no catalog half raised a condition") + } + r.half("fixture", "node") + return r.k.FindingWritten("fixture", "ace", mustJSON(t, f("fixture", "z"))) + }, []string{"machine.d.silent", "machine.g.silent", "machine.z.silent"}, ""}, + {"an observer claiming only mesh seats is held where the assignments place it, with no fixed name", func() error { + r.k.holders["g14.mesh-self-check"] = selfCheck + if node, once := r.k.catalogHolder(selfCheck); node != "g14" || !once { + t.Errorf("the holder of %s is %q, %v", selfCheck, node, once) + } + if _, once := r.k.catalogHolder("fixture"); once { + t.Errorf("a node seat's module is held once") + } + return nil + }, []string{"machine.d.silent", "machine.g.silent", "machine.z.silent"}, ""}, + {"stopped but still placed, or placed by no revision held, clears nothing", func() error { + body := mustJSON(t, nodeengine.Running{Module: "fixture", Node: "ace", Stopped: true, At: r.now}) + if err := r.k.EngineRunning("ace", body); err != nil { + return err + } + return r.k.AssignmentHeld(meshwants.AssignmentRevision{Entries: []meshwants.NodeEntryRef{{Module: "fixture", Node: "ace"}}}) + }, []string{"machine.d.silent", "machine.g.silent", "machine.z.silent"}, ""}, + {"stopped and no longer placed is cleared as left the node", func() error { + return r.k.AssignmentHeld(meshwants.AssignmentRevision{Revision: 2}) + }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftNode}, + {"an assignment revision alone clears nothing, and a module that left the catalog is cleared", func() error { + if err := r.k.FindingWritten("fixture", "ace", mustJSON(t, f("fixture", "z"))); err != nil { + return err + } + r.said = nil + if err := r.k.AssignmentHeld(meshwants.AssignmentRevision{Revision: 3}); err != nil { + return err + } + if _, ok := r.open("machine.z.silent"); !ok { + t.Errorf("an assignment revision alone cleared a module's condition") + } + return r.k.CatalogHeld("main", map[string]catseat.CatalogEntry{watchmanModule: {}, selfCheckModule: {}}) + }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, } for _, s := range steps { if err := s.do(); err != nil { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index cbec7f8c2..9b5a81ddc 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -18,6 +18,8 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -59,7 +61,9 @@ func sayOnSeat(e Event) error { // NewKeeper is the keeper over the runtime. func NewKeeper() *Keeper { - return &Keeper{Store: runtimeStore{}, Say: sayOnSeat, Now: time.Now, Logf: logf} + k := &Keeper{Store: runtimeStore{}, Say: sayOnSeat, Now: time.Now, Logf: logf} + k.Holder = k.catalogHolder + return k } // watch keeps a watch of one state, said blind while it cannot be read: a state not on the bus yet (an observer not @@ -146,6 +150,10 @@ func NewViews(k *Keeper) *views.Reader { v := &views.Reader{Module: Module, Logf: logf, Findings: ownFindings{k}, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} v.Follow(assignerModule, "wants", WantsTool) v.Follow(assignerModule, "assignments", AssignmentsTool) + // The newest assignment graph revision, which says where a module that stopped is placed no more (design 47 G §365). + if _, err := meshwantsclient.WireAssignment(v, stdio.Ask, func(rev meshwants.AssignmentRevision) { _ = k.AssignmentHeld(rev) }); err != nil { + logf("the assignment half is not followed: %v", err) + } return v } diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 066f9b6b9..6dc3c9aa5 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -36,9 +36,9 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { // The assigner's wants and assignments are asked once of their read tools and followed by their events (ADR 0615 §1); // the observers' findings are heard as events and no `findings` is read (design 47 G §383). reads := []string{HavesRead, AttemptsRead} - if !slices.Equal(m.Invokes, []string{WantsTool, AssignmentsTool, "seat:mesh-hand-acts.hand-act"}) || + if !slices.Equal(m.Invokes, []string{WantsTool, AssignmentsTool, "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision"}) || !slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-assigner.views.assignments", - "*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back"}) { + "*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "node-engine.running"}) { t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes) } for _, r := range m.Reads { diff --git a/modules/mesh-conditions/go.mod b/modules/mesh-conditions/go.mod index 717626fd6..7ce3ace27 100644 --- a/modules/mesh-conditions/go.mod +++ b/modules/mesh-conditions/go.mod @@ -12,6 +12,8 @@ require ( git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhealers v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 github.com/nats-io/nats.go v1.54.0 diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index 3414ae96b..b0431616d 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -20,7 +20,8 @@ "invokes": [ "mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-assigner.assigner_read_assignments", - "seat:mesh-hand-acts.hand-act" + "seat:mesh-hand-acts.hand-act", + "seat:mesh-wants.revision" ], "consumes": [ "mesh-delivery-assigner.views.wants", @@ -29,7 +30,9 @@ "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", - "mesh-watchdogs.machine-back" + "mesh-watchdogs.machine-back", + "mesh-wants.assignment-revision", + "node-engine.running" ], "state": [ "conditions", From 9e2311029a493a8cfd8d79eff6ca9825c6afdefc Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 15:46:35 +0100 Subject: [PATCH 2/9] COND2: the keeper keeps its sources in the state findings-views and hears them before its first view of the catalog half MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §388, §365 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/asks.go | 1 + .../cmd/mesh-conditions/catalog.go | 105 +++++++++++++++++- .../cmd/mesh-conditions/events.go | 4 + .../cmd/mesh-conditions/keeper_test.go | 27 +++++ .../cmd/mesh-conditions/main.go | 2 + .../cmd/mesh-conditions/manifest_test.go | 2 +- modules/mesh-conditions/module.json | 1 + 7 files changed, 138 insertions(+), 4 deletions(-) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/asks.go b/modules/mesh-conditions/cmd/mesh-conditions/asks.go index a380abd87..f9868baa5 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/asks.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/asks.go @@ -79,6 +79,7 @@ func (k *Keeper) Ask(by, module string) (string, error) { } var problems []string for _, w := range wait { + k.saveView(w[0], w[1]) if err := k.Store.CreateTTL(BoundsState, askedKey(w[0], w[1], ask), map[string]any{"observer": w[0], "node": w[1], "ask": ask, "until": now.Add(AskedWithin)}, AskedWithin); err != nil && !errors.Is(err, stdio.ErrChanged) { problems = append(problems, err.Error()) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go index 2b176aa2d..b513677b5 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -7,12 +7,17 @@ package main import ( "encoding/json" + "errors" + "maps" "slices" "sort" "strings" + "time" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" catseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" wants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + stdio "git.novox.be/novox/mesh-sdk/go" engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" ) @@ -48,14 +53,23 @@ func (k *Keeper) CatalogHeld(branch string, entries map[string]catseat.CatalogEn k.init() k.halves[branch] = entries var gone []string - for observer := range k.findings { - if observer == Module { - continue + seen := map[string]bool{} + consider := func(observer string) { + if observer == Module || observer == "" || seen[observer] { + return } + seen[observer] = true if _, ok := k.entry(observer); !ok { gone = append(gone, observer) } } + for observer := range k.findings { + consider(observer) + } + // A kept source (design 47 G §388) no revision lists any more leaves the list with its conditions. + for _, v := range k.views { + consider(v.observer) + } k.mu.Unlock() sort.Strings(gone) var first error @@ -85,6 +99,12 @@ func (k *Keeper) isSource(observer string) bool { k.mu.Lock() defer k.mu.Unlock() k.init() + // A source the keeper kept in `findings-views` is heard before its first view of the catalog half (design 47 G §388). + for _, v := range k.views { + if v.observer == observer && v.kept { + return true + } + } branches := make([]string, 0, len(k.halves)) for b := range k.halves { branches = append(branches, b) @@ -198,6 +218,7 @@ func (k *Keeper) sweepStopped() error { func (k *Keeper) clearObserver(observer, node, why string, first *error) { k.mu.Lock() var keys []string + var dropped [][2]string for key := range k.findings[observer] { if node == "" || k.keyNode[observer][key] == node { keys = append(keys, key) @@ -212,9 +233,13 @@ func (k *Keeper) clearObserver(observer, node, why string, first *error) { _ = k.Store.Delete(BoundsState, askedKey(observer, v.node, ask)) } delete(k.views, name) + dropped = append(dropped, [2]string{v.observer, v.node}) } } k.mu.Unlock() + for _, d := range dropped { + _ = k.Store.Delete(ViewsState, viewKey(d[0], d[1])) + } sort.Strings(keys) for _, key := range keys { k.mu.Lock() @@ -229,3 +254,77 @@ func (k *Keeper) clearObserver(observer, node, why string, first *error) { k.mu.Unlock() } } + +// ViewsState is the keeper's list of sources (design 47 G §388), in the form of its `conditions`: one key per observer and +// node, `.`, its value what the keeper heard there: the last stated set and the open asks. +const ViewsState = "findings-views" + +func viewKey(observer, node string) string { return observer + "." + node } + +// keptView is the value of one key of `findings-views`. +type keptView struct { + Observer string `json:"observer"` + Node string `json:"node"` + At time.Time `json:"at"` + Pending map[string]time.Time `json:"pending,omitempty"` + Asks map[string]time.Time `json:"asks,omitempty"` + Findings []seat.Finding `json:"findings,omitempty"` +} + +// saveView keeps what the keeper heard of the observer on the node, by compare-and-set; a source is kept once it is heard. +func (k *Keeper) saveView(observer, node string) { + k.mu.Lock() + k.init() + v := k.views[observer+"."+node] + if v == nil { + k.mu.Unlock() + return + } + v.kept = true + value := keptView{Observer: observer, Node: node, At: v.At, Pending: maps.Clone(v.Pending), Asks: maps.Clone(v.Asks), + Findings: slices.Clone(v.Findings)} + k.mu.Unlock() + key := viewKey(observer, node) + for range ConditionWrites { + _, rev, found, err := k.Store.Get(ViewsState, key) + if err == nil && !found { + _, err = k.Store.Create(ViewsState, key, value) + } else if err == nil { + _, err = k.Store.Update(ViewsState, key, value, rev) + } + if err == nil || !errors.Is(err, stdio.ErrChanged) { + if err != nil { + k.Logf("the source %s on %s is not kept: %v", observer, node, err) + } + return + } + } +} + +// RestoreViews reads the kept sources at the start, before the first view of the catalog half: their events are applied +// at once from the durable consumers, and the catalog half only adds and removes sources (design 47 G §388). +func (k *Keeper) RestoreViews() { + keys, err := k.Store.Keys(ViewsState) + if err != nil { + k.Logf("the kept sources are not read: %v", err) + return + } + for _, key := range keys { + raw, _, found, err := k.Store.Get(ViewsState, key) + if err != nil || !found { + continue + } + var kv keptView + if json.Unmarshal(raw, &kv) != nil || kv.Observer == "" || kv.Node == "" { + k.Logf("the kept source %s is not readable: left out", key) + continue + } + k.mu.Lock() + k.init() + v := k.view(kv.Observer, kv.Node) + v.At, v.Findings, v.kept = kv.At, kv.Findings, true + maps.Copy(v.Pending, kv.Pending) + maps.Copy(v.Asks, kv.Asks) + k.mu.Unlock() + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/events.go b/modules/mesh-conditions/cmd/mesh-conditions/events.go index ec7abe5d8..e0eb2c7b1 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/events.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/events.go @@ -107,6 +107,7 @@ func (k *Keeper) take(observer, node string, f seat.Finding) error { if !named.IsZero() { _ = k.Store.Delete(BoundsState, pendingKey(observer, node, v.Key)) } + k.saveView(observer, node) return k.Finding(observer, v.Key, value, false) } @@ -236,6 +237,8 @@ type sourceView struct { Findings []seat.Finding // Asks are the asks it has not answered, with the end of their bound (asks.go). Asks map[string]time.Time + // kept says the source is in the state `findings-views` (catalog.go). + kept bool } // view is the observer's view on the node, made when there is none. Called with mu held. @@ -341,6 +344,7 @@ func (k *Keeper) FindingsStated(observer, node string, body []byte) error { _ = k.Store.Delete(BoundsState, askedKey(observer, node, ask)) } k.clearUnstated(observer, node) + k.saveView(observer, node) sort.Strings(drop) for key := range oldPending { if _, still := v.Pending[key]; !still { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index 446b155fd..1f1836648 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -40,6 +40,12 @@ func (f *fakeStore) Get(state, key string) ([]byte, uint64, bool, error) { return v, f.revs[state+"/"+key], ok, nil } +// Get2 says whether the key is held. +func (f *fakeStore) Get2(state, key string) ([]byte, bool) { + v, _, ok, _ := f.Get(state, key) + return v, ok +} + func (f *fakeStore) put(state, key string, value any) uint64 { raw, _ := json.Marshal(value) f.next++ @@ -580,6 +586,27 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { } return r.k.CatalogHeld("main", map[string]catseat.CatalogEntry{watchmanModule: {}, selfCheckModule: {}}) }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, + // Design 47 G §388: a restarted keeper hears its kept sources before its first view of the catalog half. + {"a restarted keeper with no view of the catalog half hears its kept sources, and the view takes one away", func() error { + if _, ok := r.st.Get2(ViewsState, viewKey(watchman, "ace")); !ok { + t.Errorf("the source %s on ace was not kept in %s", watchman, ViewsState) + } + k2 := &Keeper{Store: r.st, Now: func() time.Time { return r.now }, Say: func(e Event) error { r.said = append(r.said, e); return nil }} + k2.RestoreViews() + if err := k2.FindingWritten(watchman, "ace", mustJSON(t, f(watchman, "k"))); err != nil { + return err + } + if _, ok := r.open("machine.k.silent"); !ok { + t.Errorf("a kept source's finding was not raised with no view of the catalog half") + } + if err := k2.FindingWritten("fixture", "ace", mustJSON(t, f("fixture", "q"))); err != nil { + return err + } + if _, ok := r.open("machine.q.silent"); ok { + t.Errorf("a module that was never a source raised a condition") + } + return k2.CatalogHeld("main", map[string]catseat.CatalogEntry{selfCheckModule: {}}) + }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, } for _, s := range steps { if err := s.do(); err != nil { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index 9b5a81ddc..92c9ed5b3 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -109,6 +109,8 @@ func Keep(k *Keeper, v *views.Reader) { } } } + // The kept sources are heard at once, before the first view of the catalog half (design 47 G §388). + k.RestoreViews() said := func(err error) error { if err != nil { logf("%v", err) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 6dc3c9aa5..6b2d320ea 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -63,7 +63,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { } // Clause 12: each state with the tool that reads it, each act with the tool that starts it. readBy := map[string]string{ConditionsState: Seat + ".conditions", HistoryState: HistoryTool, - HeldState: HeldTool, BoundsState: HeldTool, "a resync": Seat + ".resync", "what it holds of observers": FindingsViewsTool, "what it reads": InputsTool} + HeldState: HeldTool, BoundsState: HeldTool, ViewsState: FindingsViewsTool, "a resync": Seat + ".resync", "what it holds of observers": FindingsViewsTool, "what it reads": InputsTool} tools := map[string]bool{} for _, tool := range Tools(&Keeper{}) { tools[tool.Name] = true diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index b0431616d..f9aed5620 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -41,6 +41,7 @@ "ttl-seconds": 7776000 }, "held", + "findings-views", { "name": "bounds", "key-ttl": true From 76be864bff12f1119d8d8184eb895089fe5e180f Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 15:48:35 +0100 Subject: [PATCH 3/9] HEAL2 (manifest only): mesh-healers asks findings and finding-events in requires; the writer's use waits on the contracts' finding subjects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §529, §384, §461 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/mesh-healers/module.json | 32 ++++++++++++++++++++++---------- 1 file changed, 22 insertions(+), 10 deletions(-) diff --git a/modules/mesh-healers/module.json b/modules/mesh-healers/module.json index 7fe853339..4bd6f46ea 100644 --- a/modules/mesh-healers/module.json +++ b/modules/mesh-healers/module.json @@ -35,7 +35,6 @@ "name": "would", "key-ttl": true }, - "findings", { "name": "wakes", "key-ttl": true @@ -62,14 +61,6 @@ "unknown": "program" } ], - "streams": [ - { - "name": "mesh-healers-findings", - "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], - "max-age-days": 7, - "max-mib": 16 - } - ], "build": { "artifacts": [ { @@ -84,5 +75,26 @@ ] } ] - } + }, + "requires": [ + { + "name": "findings", + "storage": "key-value", + "template": "state" + }, + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + } + ] } From 26c82a507a7719cc263e2decce8b8be80b7b0b39 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 15:48:35 +0100 Subject: [PATCH 4/9] OBS4: artifact-store-tools, mesh-watcher, minio and nfs-server ask findings and finding-events in requires MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §529, §387 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/artifact-store-tools/store_test.go | 9 +++-- modules/artifact-store-tools/module.json | 34 ++++++++++++------- modules/mesh-watcher/module.json | 34 ++++++++++++------- modules/minio/module.json | 32 ++++++++++------- modules/nfs-server/module.json | 32 +++++++++++------ 5 files changed, 93 insertions(+), 48 deletions(-) diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go b/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go index 8156df52c..e7e183262 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go @@ -480,8 +480,13 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) { } // The operator, 2026-11-04 (issue 1966): it reaches the store through the registry seats' verbs, and requires no // provision artifact-store. - if len(m.Requires) != 0 { - t.Errorf("requires %v, want nothing: the store is reached through the registry seats", m.Requires) + // What it requires is its finding state and stream alone, in their template form (design 47 G §529). + var required []string + for _, r := range m.Requires { + required = append(required, fmt.Sprint(r.(map[string]any)["name"])) + } + if !reflect.DeepEqual(required, []string{"findings", "finding-events"}) { + t.Errorf("requires %v, want its findings and finding-events alone: the store is reached through the registry seats", m.Requires) } // Its address file is filled from the image seat holder's listen (ADR 0632 §1), never by the retired provision. address, err := os.ReadFile("../../files/store-address") diff --git a/modules/artifact-store-tools/module.json b/modules/artifact-store-tools/module.json index 8ae3fe620..b0e2e25a8 100644 --- a/modules/artifact-store-tools/module.json +++ b/modules/artifact-store-tools/module.json @@ -42,9 +42,6 @@ "mesh-delivery-indexer.views.manifests", "mesh-conditions.findings-asked" ], - "state": [ - "findings" - ], "tools": [ "artifact_store_repositories", "artifact_store_usage", @@ -86,14 +83,6 @@ "template": "files/store-address" } ], - "streams": [ - { - "name": "artifact-store-tools-findings", - "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], - "max-age-days": 7, - "max-mib": 16 - } - ], "build": { "artifacts": [ { @@ -117,5 +106,26 @@ "from": "files" } ] - } + }, + "requires": [ + { + "name": "findings", + "storage": "key-value", + "template": "state" + }, + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + } + ] } diff --git a/modules/mesh-watcher/module.json b/modules/mesh-watcher/module.json index 5a855c5fb..c4d8d9f98 100644 --- a/modules/mesh-watcher/module.json +++ b/modules/mesh-watcher/module.json @@ -12,9 +12,6 @@ "mesh-watcher.watcher_ping", "seat:mesh-hand-acts.hand-act" ], - "state": [ - "findings" - ], "own-secrets": { "telegram-token": "${dir:state}/telegram-token" }, @@ -54,14 +51,6 @@ "template": "files/settings.json" } ], - "streams": [ - { - "name": "mesh-watcher-findings", - "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], - "max-age-days": 7, - "max-mib": 16 - } - ], "build": { "artifacts": [ { @@ -85,5 +74,26 @@ "from": "files" } ] - } + }, + "requires": [ + { + "name": "findings", + "storage": "key-value", + "template": "state" + }, + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + } + ] } diff --git a/modules/minio/module.json b/modules/minio/module.json index 47f19cdf6..60800260d 100644 --- a/modules/minio/module.json +++ b/modules/minio/module.json @@ -19,7 +19,26 @@ } ], "requires": [ - "route" + "route", + { + "name": "findings", + "storage": "key-value", + "template": "state" + }, + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + } ], "seats": [ { @@ -47,9 +66,6 @@ "mesh-delivery-assigner.views.seat-holders", "mesh-conditions.findings-asked" ], - "state": [ - "findings" - ], "contributes": { "route": { "api": { @@ -193,14 +209,6 @@ "package": "minio-client" } ], - "streams": [ - { - "name": "minio-findings", - "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], - "max-age-days": 7, - "max-mib": 16 - } - ], "build": { "artifacts": [ { diff --git a/modules/nfs-server/module.json b/modules/nfs-server/module.json index 6f77dba2c..9e7d61b69 100644 --- a/modules/nfs-server/module.json +++ b/modules/nfs-server/module.json @@ -48,8 +48,7 @@ "name": "exports", "ttl-seconds": 120, "per-machine": true - }, - "findings" + } ], "reads": [ "disk-load.readings", @@ -172,14 +171,6 @@ "template": "files/shares.conf" } ], - "streams": [ - { - "name": "nfs-server-findings", - "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], - "max-age-days": 7, - "max-mib": 16 - } - ], "build": { "artifacts": [ { @@ -207,5 +198,26 @@ "consumes": [ "mesh-delivery-assigner.views.seat-holders", "mesh-conditions.findings-asked" + ], + "requires": [ + { + "name": "findings", + "storage": "key-value", + "template": "state" + }, + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + } ] } From ef60e5d09eea3dd757d7e604d60a702d117c5263 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 15:49:13 +0100 Subject: [PATCH 5/9] COND2 (whole): the keeper's whole test starts with a kept source and no view of the catalog half MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §388 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/whole_integration_test.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/whole_integration_test.go b/modules/mesh-conditions/cmd/mesh-conditions/whole_integration_test.go index 36292d779..48af4aaf2 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/whole_integration_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/whole_integration_test.go @@ -294,7 +294,7 @@ func TestTheKeeperWholeRaisesAndClearsOnTheBus(t *testing.T) { } buckets := []string{"mesh-controller_conditions", "mesh-delivery-assigner_wants", "node-engine_have", "mesh-delivery-assigner_assignments", "mesh-healers_attempts"} - for _, s := range []string{ConditionsState, HistoryState, HeldState, BoundsState} { + for _, s := range []string{ConditionsState, HistoryState, HeldState, ViewsState, BoundsState} { buckets = append(buckets, Module+"_"+s) } for _, b := range buckets { @@ -302,6 +302,12 @@ func TestTheKeeperWholeRaisesAndClearsOnTheBus(t *testing.T) { t.Fatal(err) } } + // Design 47 G §388: the catalog's holder is stopped, and the keeper hears the source it kept at an earlier run. + views, _ := js.KeyValue(ctx, Module+"_"+ViewsState) + kept, _ := json.Marshal(keptView{Observer: watchmanModule, Node: "ace"}) + if _, err := views.Create(ctx, viewKey(watchmanModule, "ace"), kept); err != nil { + t.Fatal(err) + } events, err := nc.SubscribeSync("mesh.seat." + Seat + ".event.>") if err != nil { t.Fatal(err) From 8f4f70f08875b88b955183e3b7d8cd2105ec49c1 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 20:26:40 +0100 Subject: [PATCH 6/9] COND6: the keeper follows one catalog view per watched branch and adopts a branch registered while it runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §365, §479, §521, §532 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/branches.go | 128 ++++++++++++++++++ .../cmd/mesh-conditions/branches_test.go | 43 ++++++ .../cmd/mesh-conditions/catalog.go | 14 ++ .../cmd/mesh-conditions/keeper_test.go | 10 ++ .../cmd/mesh-conditions/main.go | 10 ++ .../cmd/mesh-conditions/manifest_test.go | 5 +- modules/mesh-conditions/module.json | 6 +- 7 files changed, 213 insertions(+), 3 deletions(-) create mode 100644 modules/mesh-conditions/cmd/mesh-conditions/branches.go create mode 100644 modules/mesh-conditions/cmd/mesh-conditions/branches_test.go diff --git a/modules/mesh-conditions/cmd/mesh-conditions/branches.go b/modules/mesh-conditions/cmd/mesh-conditions/branches.go new file mode 100644 index 000000000..f65d63ec8 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/branches.go @@ -0,0 +1,128 @@ +package main + +// The keeper follows one view of the catalog half per watched branch, the branches the registrations name, and adopts a +// branch registered while it runs with no restart (design 47 G §532, §365, §479). No branch is named here (§7). + +import ( + "encoding/json" + "slices" + "sort" + "sync" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + catseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// Branches keeps the keeper's views of the catalog half, one per watched branch. +type Branches struct { + // Open adopts the view of the branch's catalog half and answers its name in the reader; Drop ends it, and Dropped + // forgets what the keeper held of the branch. + Open func(branch string) (string, error) + Drop func(name string) error + Dropped func(branch string) error + Logf func(string, ...any) + + mu sync.Mutex + names map[string]string +} + +// Watched sets the watched branches to the ones the registrations name: a branch new to them is adopted, a branch no +// registration names any more is dropped. +func (b *Branches) Watched(regs []catseat.Registration) { + want := map[string]bool{} + for _, r := range regs { + for _, branch := range r.Branches { + want[keys.EscapeName(branch)] = true + } + } + b.mu.Lock() + if b.names == nil { + b.names = map[string]string{} + } + var adopt, drop []string + for branch := range want { + if _, ok := b.names[branch]; !ok { + adopt = append(adopt, branch) + } + } + for branch := range b.names { + if !want[branch] { + drop = append(drop, branch) + } + } + b.mu.Unlock() + sort.Strings(adopt) + sort.Strings(drop) + for _, branch := range adopt { + name, err := b.Open(branch) + if err != nil { + b.Logf("the catalog half of %s is not followed: %v", branch, err) + continue + } + b.mu.Lock() + b.names[branch] = name + b.mu.Unlock() + } + for _, branch := range drop { + b.mu.Lock() + name := b.names[branch] + delete(b.names, branch) + b.mu.Unlock() + if err := b.Drop(name); err != nil { + b.Logf("the view %s is not dropped: %v", name, err) + } + if err := b.Dropped(branch); err != nil { + b.Logf("the conditions of the dropped branch %s are not cleared: %v", branch, err) + } + } +} + +// Followed is the branches followed, in name order. +func (b *Branches) Followed() []string { + b.mu.Lock() + defer b.mu.Unlock() + out := make([]string, 0, len(b.names)) + for branch := range b.names { + out = append(out, branch) + } + slices.Sort(out) + return out +} + +// followCatalog is Branches over the keeper's reader: each branch's view is the catalog client's, whose changes hand the +// branch's whole half to the keeper. +func followCatalog(k *Keeper, v *views.Reader, call meshcatalogclient.Call) *Branches { + return &Branches{ + Logf: logf, + Open: func(branch string) (string, error) { + var mu sync.Mutex + var c *meshcatalogclient.Catalog + changed := func(string, json.RawMessage, bool, time.Time) { + mu.Lock() + view := c + mu.Unlock() + if view == nil { + return + } + if half, ok := view.View(); ok { + if err := k.CatalogHeld(branch, half.Entries); err != nil { + logf("%v", err) + } + } + } + mu.Lock() + defer mu.Unlock() + made, err := meshcatalogclient.WireCatalog(v, call, branch, changed) + if err != nil { + return "", err + } + c = made + return made.Follower.Owner + "." + made.Follower.State, nil + }, + Drop: v.Drop, + Dropped: k.CatalogDropped, + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/branches_test.go b/modules/mesh-conditions/cmd/mesh-conditions/branches_test.go new file mode 100644 index 000000000..f7bbd1d8f --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/branches_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "slices" + "testing" + + catseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" +) + +// Design 47 G §532: the keeper adopts a branch a registration adds while it runs, and drops one no registration names. +func TestTheKeeperFollowsTheBranchesTheRegistrationsName(t *testing.T) { + var opened, dropped, forgotten []string + b := &Branches{ + Logf: func(string, ...any) {}, + Open: func(branch string) (string, error) { + opened = append(opened, branch) + return "mesh-catalog." + branch, nil + }, + Drop: func(name string) error { dropped = append(dropped, name); return nil }, + Dropped: func(branch string) error { forgotten = append(forgotten, branch); return nil }, + } + reg := func(module string, branches ...string) catseat.Registration { + return catseat.Registration{Module: module, Branches: branches} + } + steps := []struct { + name string + regs []catseat.Registration + followed, opened, dropped, forgotten []string + }{ + {"the first registrations", []catseat.Registration{reg("a", "trunk")}, []string{"trunk"}, []string{"trunk"}, nil, nil}, + {"a branch registered while it runs", []catseat.Registration{reg("a", "trunk"), reg("b", "feature/x")}, []string{"feature/x", "trunk"}, []string{"trunk", "feature/x"}, nil, nil}, + {"a branch nobody names is dropped", []catseat.Registration{reg("a", "trunk")}, []string{"trunk"}, []string{"trunk", "feature/x"}, []string{"mesh-catalog.feature/x"}, []string{"feature/x"}}, + } + for _, s := range steps { + b.Watched(s.regs) + if got := b.Followed(); !slices.Equal(got, s.followed) { + t.Errorf("%s: followed %v, want %v", s.name, got, s.followed) + } + if !slices.Equal(opened, s.opened) || !slices.Equal(dropped, s.dropped) || !slices.Equal(forgotten, s.forgotten) { + t.Errorf("%s: opened %v dropped %v forgotten %v", s.name, opened, dropped, forgotten) + } + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go index b513677b5..f7f89381c 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -52,6 +52,20 @@ func (k *Keeper) CatalogHeld(branch string, entries map[string]catseat.CatalogEn k.mu.Lock() k.init() k.halves[branch] = entries + return k.sweepCatalog() +} + +// CatalogDropped forgets a branch no registration watches any more (design 47 G §532): the modules only it listed leave +// the sources with their conditions. +func (k *Keeper) CatalogDropped(branch string) error { + k.mu.Lock() + k.init() + delete(k.halves, branch) + return k.sweepCatalog() +} + +// sweepCatalog clears the conditions of every observer no held revision lists. Called with mu held; it releases it. +func (k *Keeper) sweepCatalog() error { var gone []string seen := map[string]bool{} consider := func(observer string) { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index 1f1836648..31e11071f 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -607,6 +607,16 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { } return k2.CatalogHeld("main", map[string]catseat.CatalogEntry{selfCheckModule: {}}) }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, + // Design 47 G §532: a module only a second watched branch lists is a source, and the branch dropped takes it away. + {"a module listed only by a second branch is a source", func() error { + if err := r.k.CatalogHeld("feature", map[string]catseat.CatalogEntry{"fixture": {Streams: []string{EventFindingWritten}}}); err != nil { + return err + } + return r.k.FindingWritten("fixture", "ace", mustJSON(t, f("fixture", "z"))) + }, []string{"machine.d.silent", "machine.g.silent", "machine.z.silent"}, ""}, + {"the second branch dropped takes its source away", func() error { + return r.k.CatalogDropped("feature") + }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, } for _, s := range steps { if err := s.do(); err != nil { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index 92c9ed5b3..55f780583 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -18,7 +18,9 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + catseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -156,6 +158,14 @@ func NewViews(k *Keeper) *views.Reader { if _, err := meshwantsclient.WireAssignment(v, stdio.Ask, func(rev meshwants.AssignmentRevision) { _ = k.AssignmentHeld(rev) }); err != nil { logf("the assignment half is not followed: %v", err) } + // One view of the catalog half per watched branch, the branches the registrations name, adopted and dropped while the + // keeper runs (design 47 G §365, §532). + branches := followCatalog(k, v, stdio.Ask) + var regs *meshcatalogclient.Registrations + regs, err := meshcatalogclient.WireRegistrations(v, stdio.Ask, func(catseat.Registration, bool) { branches.Watched(regs.All()) }) + if err != nil { + logf("the registrations are not followed: %v", err) + } return v } diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 6b2d320ea..273a924a0 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -36,9 +36,10 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { // The assigner's wants and assignments are asked once of their read tools and followed by their events (ADR 0615 §1); // the observers' findings are heard as events and no `findings` is read (design 47 G §383). reads := []string{HavesRead, AttemptsRead} - if !slices.Equal(m.Invokes, []string{WantsTool, AssignmentsTool, "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision"}) || + if !slices.Equal(m.Invokes, []string{WantsTool, AssignmentsTool, "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision", + "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision"}) || !slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-assigner.views.assignments", - "*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "node-engine.running"}) { + "*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "node-engine.running"}) { t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes) } for _, r := range m.Reads { diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index f9aed5620..66cced877 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -21,7 +21,9 @@ "mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-assigner.assigner_read_assignments", "seat:mesh-hand-acts.hand-act", - "seat:mesh-wants.revision" + "seat:mesh-wants.revision", + "seat:mesh-catalog.registrations", + "seat:mesh-catalog.revision" ], "consumes": [ "mesh-delivery-assigner.views.wants", @@ -32,6 +34,8 @@ "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", + "mesh-catalog.registration-changed", + "mesh-catalog.catalog-revision", "node-engine.running" ], "state": [ From 811b59b4540dfc7e13e5d9068b193f758f03af50 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 20:27:47 +0100 Subject: [PATCH 7/9] COND9: the keeper takes a part whose finding stream the bus module made as a source (the moved-stream half stops on a question) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §515, §663 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/bound.go | 73 +++++++++++++++++++ .../cmd/mesh-conditions/catalog.go | 4 + .../cmd/mesh-conditions/events.go | 6 +- .../cmd/mesh-conditions/keeper.go | 2 + .../cmd/mesh-conditions/keeper_test.go | 18 +++++ .../cmd/mesh-conditions/main.go | 1 + .../cmd/mesh-conditions/manifest_test.go | 4 +- modules/mesh-conditions/go.mod | 1 + modules/mesh-conditions/module.json | 4 +- 9 files changed, 109 insertions(+), 4 deletions(-) create mode 100644 modules/mesh-conditions/cmd/mesh-conditions/bound.go diff --git a/modules/mesh-conditions/cmd/mesh-conditions/bound.go b/modules/mesh-conditions/cmd/mesh-conditions/bound.go new file mode 100644 index 000000000..edff8dac4 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/bound.go @@ -0,0 +1,73 @@ +package main + +// The keeper's sources are the parts whose finding stream the bus module made (design 47 G §515, §663): it takes a source +// when it hears the seat mesh-broker's `created` of a part's `finding-events` entry, and at its start reads the entries +// the bus module holds by `list`. No part is named here (§7). + +import ( + "encoding/json" + "fmt" + "time" + + broker "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" +) + +// FindingEventsTemplate is the template of a part's finding stream in its `requires` (design 47 G §529). +const FindingEventsTemplate = "finding-events" + +// BrokerCreated takes one `created` of the bus module: a `finding-events` entry makes its module a source, asked once with +// `findings-asked` naming it. An entry of another template, or a module already a source, changes nothing. +func (k *Keeper) BrokerCreated(body []byte) error { + var said broker.Said + if err := json.Unmarshal(body, &said); err != nil || said.Module == "" { + k.Logf("a created of the bus module is not readable: %v", err) + return nil + } + if said.Template != FindingEventsTemplate { + return nil + } + return k.takeSource(said.Module) +} + +// takeSource makes the module a source the bus module made a stream for, and asks it once. +func (k *Keeper) takeSource(module string) error { + k.mu.Lock() + k.init() + if k.created == nil { + k.created = map[string]bool{} + } + if k.created[module] { + k.mu.Unlock() + return nil + } + k.created[module] = true + k.mu.Unlock() + if _, err := k.Ask(Module, module); err != nil { + return fmt.Errorf("asking %s for its findings: %w", module, err) + } + return nil +} + +// ListBound reads the entries the bus module holds at the keeper's start: every module with a `finding-events` entry is a +// source. A call that fails is asked again with a growing wait, the bus module being delivered after the keeper at times. +func ListBound(k *Keeper, call func(key string, body any) (json.RawMessage, error)) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + raw, err := call("seat:"+broker.Seat+"."+broker.VerbList, broker.ListInput{}) + var answer broker.ListAnswer + if err == nil { + err = json.Unmarshal(raw, &answer) + } + if err == nil { + for _, e := range answer.Entries { + if e.Template == FindingEventsTemplate { + if err := k.takeSource(e.Module); err != nil { + k.Logf("%v", err) + } + } + } + return + } + k.Logf("the entries of the bus module are not read yet (%v); asked again in %s", err, wait) + time.Sleep(wait) + } +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go index f7f89381c..19e16487c 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -113,6 +113,10 @@ func (k *Keeper) isSource(observer string) bool { k.mu.Lock() defer k.mu.Unlock() k.init() + // A part whose finding stream the bus module made is a source (design 47 G §663). + if k.created[observer] { + return true + } // A source the keeper kept in `findings-views` is heard before its first view of the catalog half (design 47 G §388). for _, v := range k.views { if v.observer == observer && v.kept { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/events.go b/modules/mesh-conditions/cmd/mesh-conditions/events.go index e0eb2c7b1..f2be01310 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/events.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/events.go @@ -14,6 +14,7 @@ import ( "strings" "time" + broker "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" engine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" cc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" @@ -51,6 +52,9 @@ func (k *Keeper) FindingEvent(e stdio.Envelope) error { if e.Key == engine.Seat+"."+engine.EventRunning { return k.EngineRunning(e.Node, e.Body) } + if e.Key == broker.Seat+"."+broker.EventCreated { + return k.BrokerCreated(e.Body) + } if observer, ok := emitterOf(e.Key, EventFindingsStated); ok { return k.FindingsStated(observer, e.Node, e.Body) } @@ -202,7 +206,7 @@ func (k *Keeper) MachinePresence(body []byte, away bool) error { func hear(k *Keeper) { patterns := []string{"*." + EventFindingWritten, "*." + EventFindingCleared, "*." + EventFindingsStated, watchdogs.Seat + "." + watchdogs.EventMachineAway, watchdogs.Seat + "." + watchdogs.EventMachineBack, - engine.Seat + "." + engine.EventRunning} + engine.Seat + "." + engine.EventRunning, broker.Seat + "." + broker.EventCreated} for _, pattern := range patterns { go func() { <-stdio.Served() diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index 6753dffc2..a8598441d 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -147,6 +147,8 @@ type Keeper struct { placed map[string]bool placedHeld bool stopped map[string]bool + // created are the modules whose `finding-events` entry the bus module made (bound.go). + created map[string]bool } func (k *Keeper) init() { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index 31e11071f..3d47e4f62 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -617,6 +617,24 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { {"the second branch dropped takes its source away", func() error { return r.k.CatalogDropped("feature") }, []string{"machine.d.silent", "machine.g.silent"}, whyLeftCatalog}, + // Design 47 G §663: a part whose finding-events entry the bus module made is a source, asked once. + {"a created finding-events entry makes its part a source, asked once", func() error { + before := len(asked) + created := mustJSON(t, map[string]any{"module": "newpart", "template": FindingEventsTemplate, "bus-name": "newpart_finding-events", "local-name": "finding-events"}) + other := mustJSON(t, map[string]any{"module": "otherpart", "template": "state", "bus-name": "otherpart_x", "local-name": "x"}) + for _, body := range [][]byte{created, other, created} { + if err := r.k.FindingEvent(stdio.Envelope{Key: "mesh-broker.created", Body: body}); err != nil { + return err + } + } + if len(asked) != before+1 || asked[before]["module"] != "newpart" { + t.Errorf("asked %v, want newpart once", asked[before:]) + } + if err := r.k.FindingWritten("otherpart", "ace", mustJSON(t, f("otherpart", "o"))); err != nil { + return err + } + return r.k.FindingWritten("newpart", "ace", mustJSON(t, f("newpart", "w"))) + }, []string{"machine.d.silent", "machine.g.silent", "machine.w.silent"}, ""}, } for _, s := range steps { if err := s.do(); err != nil { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index 55f780583..a185af451 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -130,6 +130,7 @@ func Keep(k *Keeper, v *views.Reader) { return said(k.Attempt(c.Key, c.Value, c.Op != "put")) }) hear(k) + go func() { <-stdio.Served(); ListBound(k, stdio.Ask) }() watch(k, v, BoundsState, "", func(c stdio.StateChange) error { if c.Op == stdio.OpExpire { return said(k.BoundPassed(c.Key)) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 273a924a0..73d3b0920 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -37,9 +37,9 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { // the observers' findings are heard as events and no `findings` is read (design 47 G §383). reads := []string{HavesRead, AttemptsRead} if !slices.Equal(m.Invokes, []string{WantsTool, AssignmentsTool, "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision", - "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision"}) || + "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-broker.list"}) || !slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-assigner.views.assignments", - "*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "node-engine.running"}) { + "*.finding-written", "*.finding-cleared", "*.findings-stated", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", "mesh-broker.created", "node-engine.running"}) { t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes) } for _, r := range m.Reads { diff --git a/modules/mesh-conditions/go.mod b/modules/mesh-conditions/go.mod index 7ce3ace27..13d6fdc7e 100644 --- a/modules/mesh-conditions/go.mod +++ b/modules/mesh-conditions/go.mod @@ -5,6 +5,7 @@ go 1.26.0 require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index 66cced877..569f6071a 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -23,7 +23,8 @@ "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.revision", "seat:mesh-catalog.registrations", - "seat:mesh-catalog.revision" + "seat:mesh-catalog.revision", + "seat:mesh-broker.list" ], "consumes": [ "mesh-delivery-assigner.views.wants", @@ -36,6 +37,7 @@ "mesh-wants.assignment-revision", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", + "mesh-broker.created", "node-engine.running" ], "state": [ From ecf597adce36b757d9e3656670d39525235145e6 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 20:29:24 +0100 Subject: [PATCH 8/9] COND11 (in part): the keeper's ask bound and first-view bound are settings read at its start; the two hours wait on COND7 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §557 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/asks.go | 16 ++++++--- .../cmd/mesh-conditions/keeper.go | 2 ++ .../cmd/mesh-conditions/main.go | 2 ++ .../cmd/mesh-conditions/settings.go | 36 +++++++++++++++++++ .../cmd/mesh-conditions/settings_test.go | 32 +++++++++++++++++ modules/mesh-conditions/module.json | 14 ++++++++ 6 files changed, 98 insertions(+), 4 deletions(-) create mode 100644 modules/mesh-conditions/cmd/mesh-conditions/settings.go create mode 100644 modules/mesh-conditions/cmd/mesh-conditions/settings_test.go diff --git a/modules/mesh-conditions/cmd/mesh-conditions/asks.go b/modules/mesh-conditions/cmd/mesh-conditions/asks.go index f9868baa5..b9bff6d23 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/asks.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/asks.go @@ -18,9 +18,17 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" ) -// AskedWithin is how long an observer has to state its findings after an ask. +// AskedWithin is how long an observer has to state its findings after an ask, when the setting says none (design 47 G §557). const AskedWithin = 5 * time.Minute +// askedWithin is the keeper's bound for an ask: its setting, AskedWithin when none. +func (k *Keeper) askedWithin() time.Duration { + if k.AskBound > 0 { + return k.AskBound + } + return AskedWithin +} + // The kind of the keeper's finding of an observer that stated nothing, and the prefix of its ask's key in `bounds`. const ( KindFindingsUnstated = "findings-unstated" @@ -62,7 +70,7 @@ func (k *Keeper) Ask(by, module string) (string, error) { if v.observer == "" || module != "" && v.observer != module || k.away[v.node] { continue } - v.Asks[ask] = now.Add(AskedWithin) + v.Asks[ask] = now.Add(k.askedWithin()) wait = append(wait, [2]string{v.observer, v.node}) } k.mu.Unlock() @@ -81,7 +89,7 @@ func (k *Keeper) Ask(by, module string) (string, error) { for _, w := range wait { k.saveView(w[0], w[1]) if err := k.Store.CreateTTL(BoundsState, askedKey(w[0], w[1], ask), map[string]any{"observer": w[0], "node": w[1], "ask": ask, - "until": now.Add(AskedWithin)}, AskedWithin); err != nil && !errors.Is(err, stdio.ErrChanged) { + "until": now.Add(k.askedWithin())}, k.askedWithin()); err != nil && !errors.Is(err, stdio.ErrChanged) { problems = append(problems, err.Error()) } } @@ -138,7 +146,7 @@ func (k *Keeper) AskPassed(bound string) error { now := k.Now().UTC() f := seat.FindingsByKeyValue{V: cc.FindingVersion, Key: unstatedKey(observer, node), Kind: KindFindingsUnstated, Subject: seat.Subject{Scope: cc.ScopeModule, ID: observer + "." + node, Machine: node}, - Severity: cc.Warning, Summary: fmt.Sprintf("%s on %s stated no findings within %s of the ask", observer, node, AskedWithin), + Severity: cc.Warning, Summary: fmt.Sprintf("%s on %s stated no findings within %s of the ask", observer, node, k.askedWithin()), Headline: observer + " on " + node + " did not state its findings", Explanation: "The keeper asked " + observer + " on " + node + " to state its findings and heard nothing; its conditions stand as they were.", Source: "findings-asked", Observer: Module, At: now, Evidence: []seat.Evidence{{At: now, Said: "ask " + ask + " unanswered"}}} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index a8598441d..154773e4d 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -107,6 +107,8 @@ type Keeper struct { Homes map[string]Home // Bound is a delivery's hold bound; motion.DefaultDeliveryBound when zero (hold.go). Bound time.Duration + // AskBound is how long an observer has to state its findings after an ask (the setting ask-bound-seconds, §557); AskedWithin when zero. + AskBound time.Duration mu sync.Mutex findings map[string]map[string]seat.FindingsByKeyValue diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index a185af451..90fd771f6 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -193,6 +193,8 @@ func main() { v := NewViews(k) go func() { <-stdio.Served() + // The bounds are settings, read at the start once the runtime serves (design 47 G §557). + v.Bound = ReadSettings(k, stdio.Config) v.Start(context.Background()) Keep(k, v) }() diff --git a/modules/mesh-conditions/cmd/mesh-conditions/settings.go b/modules/mesh-conditions/cmd/mesh-conditions/settings.go new file mode 100644 index 000000000..b29fe7596 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/settings.go @@ -0,0 +1,36 @@ +package main + +// The bounds the keeper times are settings read at its start, their defaults the design's values, so a whole test sets +// them in seconds and never waits on real time (design 47 G §557). + +import ( + "encoding/json" + "time" +) + +// The settings, in seconds. +const ( + SettingAskBound = "ask-bound-seconds" + SettingViewWaits = "view-waits-seconds" + defaultViewWaits = 5 * time.Minute +) + +// seconds reads one setting as a duration; def when the setting is unreadable, absent or not above zero. +func seconds(read func(key string) (json.RawMessage, error), key string, def time.Duration) time.Duration { + raw, err := read(key) + if err != nil { + logf("the setting %s cannot be read (%v); its default %s stands", key, err, def) + return def + } + var n float64 + if json.Unmarshal(raw, &n) != nil || n <= 0 { + return def + } + return time.Duration(n * float64(time.Second)) +} + +// ReadSettings sets the keeper's bounds and the first views' wait from the settings. +func ReadSettings(k *Keeper, read func(key string) (json.RawMessage, error)) (viewWaits time.Duration) { + k.AskBound = seconds(read, SettingAskBound, AskedWithin) + return seconds(read, SettingViewWaits, defaultViewWaits) +} diff --git a/modules/mesh-conditions/cmd/mesh-conditions/settings_test.go b/modules/mesh-conditions/cmd/mesh-conditions/settings_test.go new file mode 100644 index 000000000..9d9f48e82 --- /dev/null +++ b/modules/mesh-conditions/cmd/mesh-conditions/settings_test.go @@ -0,0 +1,32 @@ +package main + +import ( + "encoding/json" + "errors" + "testing" + "time" +) + +// Design 47 G §557: the bounds the keeper times are settings, their defaults the design's values. +func TestTheKeepersBoundsAreSettings(t *testing.T) { + for _, c := range []struct { + name string + read func(string) (json.RawMessage, error) + ask, view time.Duration + }{ + {"unset", func(string) (json.RawMessage, error) { return nil, errors.New("none") }, AskedWithin, defaultViewWaits}, + {"set in seconds", func(key string) (json.RawMessage, error) { + if key == SettingAskBound { + return json.RawMessage("2"), nil + } + return json.RawMessage("3"), nil + }, 2 * time.Second, 3 * time.Second}, + {"not above zero", func(string) (json.RawMessage, error) { return json.RawMessage("0"), nil }, AskedWithin, defaultViewWaits}, + } { + k := &Keeper{} + view := ReadSettings(k, c.read) + if k.askedWithin() != c.ask || view != c.view { + t.Errorf("%s: ask bound %s, view bound %s; want %s, %s", c.name, k.askedWithin(), view, c.ask, c.view) + } + } +} diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index 569f6071a..a59a68797 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -53,6 +53,20 @@ "key-ttl": true } ], + "config": { + "ask-bound-seconds": { + "type": "number", + "description": "How long an observer has to state its findings after an ask, in seconds, before the keeper raises findings-unstated for it: 300 is five minutes.", + "minimum": 1, + "default": 300 + }, + "view-waits-seconds": { + "type": "number", + "description": "How long a first view may stay unanswered, in seconds, before the keeper raises view-waits for it: 300 is five minutes.", + "minimum": 1, + "default": 300 + } + }, "own-secrets": { "broker": "${dir:state}/broker" }, From 1af839fb727a0a80e6deef0dc604cd49bf067c0d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 20:44:44 +0100 Subject: [PATCH 9/9] COND9 (moved-stream half): the keeper keeps a moved part's conditions, drops the old source and asks once on the new stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §663, §690 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-conditions/catalog.go | 66 ++++++++++++++++++- .../cmd/mesh-conditions/keeper.go | 2 + .../cmd/mesh-conditions/keeper_test.go | 41 ++++++++++++ 3 files changed, 106 insertions(+), 3 deletions(-) diff --git a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go index 19e16487c..4b9cb39dc 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/catalog.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/catalog.go @@ -46,13 +46,60 @@ func declaresFindings(e catseat.CatalogEntry) bool { return false } +// findingStreams names the streams the entry declares its finding events in, one string, in name order; none is "". +func findingStreams(e catseat.CatalogEntry) string { + var names []string + for _, s := range e.Streams { + if strings.Contains(s, EventFindingWritten) { + names = append(names, s) + } + } + streams, _ := e.Manifest["streams"].([]any) + for _, s := range streams { + m, _ := s.(map[string]any) + name, _ := m["name"].(string) + events, _ := m["events"].([]any) + for _, ev := range events { + if n, _ := ev.(string); strings.HasPrefix(n, EventFindingWritten) { + names = append(names, name) + break + } + } + } + sort.Strings(names) + return strings.Join(names, ",") +} + // CatalogHeld takes the newest catalog revision of one watched branch: the sources are the modules declaring the finding -// events in any branch, and the conditions of a module in no branch are cleared. +// events in any branch, and the conditions of a module in no branch are cleared. A module whose entry names another +// stream than the one the keeper followed for it has moved its findings: the keeper keeps its open conditions and asks +// once on the new stream (design 47 G §663, §690). func (k *Keeper) CatalogHeld(branch string, entries map[string]catseat.CatalogEntry) error { k.mu.Lock() k.init() k.halves[branch] = entries - return k.sweepCatalog() + if k.followed == nil { + k.followed = map[string]string{} + } + var moved []string + for module, e := range entries { + now := findingStreams(e) + was := k.followed[module] + if now != "" { + k.followed[module] = now + } + if was != "" && now != "" && now != was { + moved = append(moved, module) + } + } + sort.Strings(moved) + err := k.sweepCatalog() + for _, module := range moved { + if _, askErr := k.Ask(Module, module); askErr != nil && err == nil { + err = askErr + } + } + return err } // CatalogDropped forgets a branch no registration watches any more (design 47 G §532): the modules only it listed leave @@ -118,8 +165,11 @@ func (k *Keeper) isSource(observer string) bool { return true } // A source the keeper kept in `findings-views` is heard before its first view of the catalog half (design 47 G §388). + // Once a revision lists the part and no longer declares its finding stream, a kept source of the old stream is dropped, + // its open conditions kept (§690); a part no revision lists is kept until the first view takes it away. + _, listed := k.entry(observer) for _, v := range k.views { - if v.observer == observer && v.kept { + if v.observer == observer && v.kept && (!listed || k.declares(observer)) { return true } } @@ -135,6 +185,16 @@ func (k *Keeper) isSource(observer string) bool { return false } +// declares says whether a held revision lists the observer with its finding events. Called with mu held. +func (k *Keeper) declares(observer string) bool { + for _, half := range k.halves { + if e, ok := half[observer]; ok && declaresFindings(e) { + return true + } + } + return false +} + // catalogHolder is Keeper.Holder: an observer whose catalog entry claims only seats held once for the mesh is held on the // node the assignments name as the holder of one of them (design 47 G §384); unknown, it is taken from every node. func (k *Keeper) catalogHolder(observer string) (string, bool) { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index 154773e4d..4b600e93c 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -151,6 +151,8 @@ type Keeper struct { stopped map[string]bool // created are the modules whose `finding-events` entry the bus module made (bound.go). created map[string]bool + // followed are the finding streams the catalog half declared for each module, by module (catalog.go). + followed map[string]string } func (k *Keeper) init() { diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go index 3d47e4f62..0578a1dd9 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go @@ -635,6 +635,47 @@ func TestTheKeeperSetsItsViewFromWhatAnObserverStates(t *testing.T) { } return r.k.FindingWritten("newpart", "ace", mustJSON(t, f("newpart", "w"))) }, []string{"machine.d.silent", "machine.g.silent", "machine.w.silent"}, ""}, + // Design 47 G §663, §690: a part whose findings move keeps its condition, is asked once on the new stream, and its + // old source is dropped once its entry declares no finding stream. + {"a part that moves its finding stream is asked once and keeps its condition", func() error { + entry := func(stream string) map[string]catseat.CatalogEntry { + e := catseat.CatalogEntry{Module: "movedpart", Manifest: map[string]any{}} + if stream != "" { + e.Manifest["streams"] = []any{map[string]any{"name": stream, "events": []any{"finding-written.*"}}} + } + return map[string]catseat.CatalogEntry{"movedpart": e} + } + if err := r.k.CatalogHeld("moves", entry("old-findings")); err != nil { + return err + } + if err := r.k.FindingWritten("movedpart", "ace", mustJSON(t, f("movedpart", "m"))); err != nil { + return err + } + before := len(asked) + if err := r.k.CatalogHeld("moves", entry("old-findings")); err != nil { + return err + } + if len(asked) != before { + t.Errorf("the same entry asked %v", asked[before:]) + } + if err := r.k.CatalogHeld("moves", entry("new-findings")); err != nil { + return err + } + if len(asked) != before+1 || asked[before]["module"] != "movedpart" { + t.Errorf("asked %v, want movedpart once on the new stream", asked[before:]) + } + // Its entry declares no finding stream now (the findings moved to a template): the old source is dropped. + if err := r.k.CatalogHeld("moves", entry("")); err != nil { + return err + } + if err := r.k.FindingWritten("movedpart", "ace", mustJSON(t, f("movedpart", "late"))); err != nil { + return err + } + if _, ok := r.open("machine.late.silent"); ok { + t.Errorf("the dropped source raised a condition") + } + return nil + }, []string{"machine.d.silent", "machine.g.silent", "machine.m.silent"}, ""}, // newpart, listed by no revision, leaves with the first view } for _, s := range steps { if err := s.do(); err != nil {