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

Merge branch 'feat/47-G-conditions' into feat/47-G-dev2-join

Browse Source
This commit is contained in:
jschoubben
2026-11-11 21:06:13 +01:00
parent bea1c446ea 1af839fb72
commit 65570636b2
21 changed files with 1102 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go
+7 -2
View File
@@ -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")
modules/artifact-store-tools/module.json
+22 -12
View File
@@ -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
}
}
]
}
modules/mesh-conditions/cmd/mesh-conditions/asks.go
+13 -4
View File
@@ -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()
@@ -79,8 +87,9 @@ 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) {
"until": now.Add(k.askedWithin())}, k.askedWithin()); err != nil && !errors.Is(err, stdio.ErrChanged) {
problems = append(problems, err.Error())
}
}
@@ -137,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"}}}
modules/mesh-conditions/cmd/mesh-conditions/bound.go
+73
View File
@@ -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)
}
}
modules/mesh-conditions/cmd/mesh-conditions/branches.go
+128
View File
@@ -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,
}
}
modules/mesh-conditions/cmd/mesh-conditions/branches_test.go
+43
View File
@@ -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)
}
}
}
modules/mesh-conditions/cmd/mesh-conditions/catalog.go
+408
View File
@@ -0,0 +1,408 @@
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"
"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"
)
// 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
}
// 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. 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
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
// 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) {
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
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()
// 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).
// 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 && (!listed || k.declares(observer)) {
return true
}
}
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
}
// 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) {
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
var dropped [][2]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)
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()
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()
}
}
// ViewsState is the keeper's list of sources (design 47 G §388), in the form of its `conditions`: one key per observer and
// node, `<observer>.<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()
}
}
modules/mesh-conditions/cmd/mesh-conditions/events.go
+14 -4
View File
@@ -14,6 +14,8 @@ 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"
watchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
@@ -47,15 +49,18 @@ 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 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)
}
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 {
@@ -106,6 +111,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)
}
@@ -199,7 +205,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, broker.Seat + "." + broker.EventCreated}
for _, pattern := range patterns {
go func() {
<-stdio.Served()
@@ -234,6 +241,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.
@@ -339,6 +348,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 {
modules/mesh-conditions/cmd/mesh-conditions/keeper.go
+15 -19
View File
@@ -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 `<module>.<name>`.
const (
AttemptsRead = healersModule + "." + meshhealers.StateAttempts
@@ -125,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
@@ -158,6 +142,17 @@ 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
// `<node>.<module>` the newest assignment graph revision places, placedHeld whether one is held; stopped are the
// `<node>.<module>` the node-engines say stopped.
halves map[string]map[string]catseat.CatalogEntry
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
// followed are the finding streams the catalog half declared for each module, by module (catalog.go).
followed map[string]string
}
func (k *Keeper) init() {
@@ -168,6 +163,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 {
modules/mesh-conditions/cmd/mesh-conditions/keeper_test.go
+161 -54
View File
@@ -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"
@@ -41,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++
@@ -112,9 +117,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 +369,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 +389,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 +400,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 +542,140 @@ 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},
// 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},
// 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},
// 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"}, ""},
// 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 {
modules/mesh-conditions/cmd/mesh-conditions/main.go
+24 -1
View File
@@ -18,6 +18,10 @@ 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"
)
@@ -59,7 +63,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
@@ -105,6 +111,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)
@@ -122,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))
@@ -146,6 +155,18 @@ 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)
}
// 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
}
@@ -172,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)
}()
modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go
+4 -3
View File
@@ -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"}) ||
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-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"}) {
"*.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 {
@@ -63,7 +64,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
modules/mesh-conditions/cmd/mesh-conditions/settings.go
+36
View File
@@ -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)
}
modules/mesh-conditions/cmd/mesh-conditions/settings_test.go
+32
View File
@@ -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)
}
}
}
modules/mesh-conditions/cmd/mesh-conditions/whole_integration_test.go
+7 -1
View File
@@ -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)
modules/mesh-conditions/go.mod
+3
View File
@@ -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
@@ -12,6 +13,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
modules/mesh-conditions/module.json
+26 -2
View File
@@ -20,7 +20,11 @@
"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",
"seat:mesh-catalog.registrations",
"seat:mesh-catalog.revision",
"seat:mesh-broker.list"
],
"consumes": [
"mesh-delivery-assigner.views.wants",
@@ -29,7 +33,12 @@
"*.finding-cleared",
"*.findings-stated",
"mesh-watchdogs.machine-away",
"mesh-watchdogs.machine-back"
"mesh-watchdogs.machine-back",
"mesh-wants.assignment-revision",
"mesh-catalog.registration-changed",
"mesh-catalog.catalog-revision",
"mesh-broker.created",
"node-engine.running"
],
"state": [
"conditions",
@@ -38,11 +47,26 @@
"ttl-seconds": 7776000
},
"held",
"findings-views",
{
"name": "bounds",
"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"
},
modules/mesh-healers/module.json
+22 -10
View File
@@ -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
}
}
]
}
modules/mesh-watcher/module.json
+22 -12
View File
@@ -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
}
}
]
}
modules/minio/module.json
+20 -12
View File
@@ -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": [
{
modules/nfs-server/module.json
+22 -10
View File
@@ -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
}
}
]
}