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

A-197: the watchman reads no node-engine state: have and health go, running ends a want's pulse

Browse Source 
Removed with the reads of node-engine.have and mesh-host.health: the findings module-unhealthy (module <module>.<node>, token
unhealthy, urgent after four hours), kept-as-it-runs (module <module>.<node>, token kept), apply-failing from a have that failed
(machine <node>; the provisioner clients' half stays), machine-network, network-rewritten, network-unreachable, machine-units,
relogin-needed, directory-used-as-found, node-account-fallback, and the urgency of a silent control node. A want's pulse now ends on
the node-engine's running event (consumes node-engine.running).

Builds: design 47 G A-197, A-192, §541

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 05:58:55 +01:00
parent 161d6e6451
commit 3eca094c5c
12 changed files with 68 additions and 1035 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go
+3 -1
View File
@@ -200,7 +200,7 @@ func recordSubject(w *watch.Watchman, subject string, body []byte) bool {
return true
}
// engineWaitSubject hands the node-engine seat's `wait-held` and `wait-ended`, on
// engineWaitSubject hands the node-engine seat's `wait-held`, `wait-ended` and `running`, on
// mesh.seat.node-engine.event.<event>.<node>.<module>, to the watchman (design 47 G §499, §541). It says whether the subject was one.
func engineWaitSubject(w *watch.Watchman, subject string, body []byte) bool {
parts := strings.Split(subject, ".")
@@ -212,6 +212,8 @@ func engineWaitSubject(w *watch.Watchman, subject string, body []byte) bool {
w.EngineWaitHeld(parts[5], parts[6], body)
case nodeengine.EventWaitEnded:
w.EngineWaitEnded(parts[5], parts[6])
case nodeengine.EventRunning:
w.EngineRunning(parts[5], parts[6], body)
default:
return false
}
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go
+3 -10
View File
@@ -17,7 +17,6 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
@@ -150,7 +149,6 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
{Bucket: watch.StuckBucket, History: 20},
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: watch.HaveBucket, History: 20},
{Bucket: watch.FindingsBucket, History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -390,7 +388,6 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
{Bucket: watch.StuckBucket, History: 20},
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: watch.HaveBucket, History: 20},
{Bucket: watch.FindingsBucket, History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -472,7 +469,6 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
{Bucket: watch.StuckBucket, History: 20},
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: watch.HaveBucket, History: 20},
{Bucket: watch.FindingsBucket, History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -498,14 +494,11 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
t.Fatal(err)
}
}
// The node-engine says `running` on its seat (design 47 G §541), which the watchman hears on its consumer of EVENTS.
putHave := func(node, module, fp string) {
t.Helper()
body, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
if err != nil {
t.Fatal(err)
}
key, _ := nodeengineclient.HaveKey(node, module)
if _, err := kvs[watch.HaveBucket].Put(ctx, key, body); err != nil {
body, _ := json.Marshal(nodeengine.Running{Fingerprint: fp, Module: module, Node: node, At: time.Now().UTC(), V: 1})
if _, err := js.Publish(ctx, "mesh.seat.node-engine.event.running."+node+"."+module, body); err != nil {
t.Fatal(err)
}
}
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go
+6 -43
View File
@@ -221,7 +221,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// Its own presence, pulses and stuck rows are what it keeps: each stops the start, naming its grant. The assigner's wants
// and the node-engine's haves only make the pulses of a move: without the haves the watchman watches presence and the
// build seats' steps, and says that moves are not watched (issue 1945).
var kvs struct{ presence, pulse, stuck, haves jetstream.KeyValue }
var kvs struct{ presence, pulse, stuck jetstream.KeyValue }
for _, b := range []struct {
name string
into *jetstream.KeyValue
@@ -230,7 +230,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
}
kvs.haves = process.OptionalBucket(ctx, js, watch.HaveBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf)
// The assigner's wants and the indexer's modules are their owners' states, read by the first view of their read tools and
// then their views families, never their buckets (ADR 0615 §1, DLV-M5). An owner that is down never holds the start: each
// is read on a goroutine once its first view is answered, and until then the unanswered first view is the watchman's
@@ -308,15 +307,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err != nil {
return err
}
// What the node agents run (issue 1906, ADR 0539 §19): a want is a pulse owed by its node's agent until the have names its
// fingerprint, so the haves are read at the start and the wants after them, once their first view is answered.
var haveSeen jetstream.KeyWatcher
if kvs.haves != nil {
if haveSeen, err = settle(ctx, kvs.haves, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) }); err != nil {
process.NotWatched(logger.Printf, watch.HaveBucket, err, "moves are not watched until the watchman starts again")
haveSeen = nil
}
}
// The events first, then the one request of the open connections, so no connection falls between them.
onEvent := func(open bool) nats.MsgHandler {
return func(m *nats.Msg) {
@@ -377,25 +367,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}()
})
}
// Each node's own health report (the node-engine's state health): read before the findings are reconciled when the
// bucket is there, bound once it appears when it is not; each change is heard on the watch, never polled.
healthEach := func(e jetstream.KeyValueEntry) { w.NodeHealth(e.Key(), e.Value(), opOf(e)) }
var healthWatch jetstream.KeyWatcher
if kv, err := js.KeyValue(ctx, watch.NodeHealthBucket); err == nil {
if healthWatch, err = settle(ctx, kv, healthEach); err != nil {
process.NotWatched(logger.Printf, watch.NodeHealthBucket, err, "no node's own health is said until the watchman starts again")
}
} else {
process.AwaitBucket(ctx, js, conn, watch.NodeHealthBucket, "no node's own health is said",
logger.Printf).Ready(func(kv jetstream.KeyValue) {
hw, err := settle(ctx, kv, healthEach)
if err != nil {
process.NotWatched(logger.Printf, watch.NodeHealthBucket, err, "no node's own health is said until the watchman starts again")
return
}
go func() { _ = follow(ctx, hw, healthEach) }()
})
}
dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf}
w.SetRecount(dead.count)
dead.count()
@@ -439,24 +410,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
errs := make(chan error, 16)
if healthWatch != nil {
go func() { errs <- follow(ctx, healthWatch, healthEach) }()
}
// The first views of the wants and the modules, asked now and read each on its own goroutine once answered (ADR 0615).
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "watchman_first_views reads nothing and watchman_first_view_again is not heard", logger.Printf))
// Every part has registered: the hub pulls the module's durable, and the watchman's other consumers open.
hearConsumed(ctx, js, cred, hub, w, logger.Printf)
defer hub.Stop()
if haveSeen != nil {
go func() {
errs <- follow(ctx, haveSeen, func(e jetstream.KeyValueEntry) { w.Have(e.Key(), e.Value(), opOf(e)) })
}()
go func() {
errs <- readLate(ctx, w, wants, watch.AssignerWantsBucket, func(e jetstream.KeyValueEntry) {
w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now()))
})
}()
}
go func() {
errs <- readLate(ctx, w, wants, watch.AssignerWantsBucket, func(e jetstream.KeyValueEntry) {
w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now()))
})
}()
if provWatch != nil {
go func() {
errs <- follow(ctx, provWatch, func(e jetstream.KeyValueEntry) { w.Provision(e.Key(), e.Value(), opOf(e)) })
modules/mesh-delivery-watchman/internal/watch/findings_test.go
+3 -243
View File
@@ -3,14 +3,11 @@ package watch
import (
"encoding/json"
"errors"
"slices"
"strings"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
@@ -30,15 +27,6 @@ func mustEncode(t *testing.T, f Finding) []byte {
return v
}
func haveOf(t *testing.T, phase string, health string, detail string) []byte {
t.Helper()
b, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpNew, Phase: phase, Health: health, Detail: detail})
if err != nil {
t.Fatal(err)
}
return b
}
// expire hears the expiry of the pulse at pk.
func expire(w *Watchman, c *clock, pk string, after time.Duration) {
c.t = c.t.Add(after)
@@ -323,29 +311,6 @@ func TestTheWatchmansFindings(t *testing.T) {
}
}
}, map[string]string{}},
{"ADR 0601 §4: what a node keeps of a module as it runs, for a manifest that does not say it yet, is a warning until the have says none", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
have := func(kept ...string) []byte {
b, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: "src2:" + strings.Repeat("a", 64), Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK, Kept: kept})
if err != nil {
t.Fatal(err)
}
return b
}
k := FindingKey(ScopeModule, "runner-x.ace", "kept")
w.Have("ace.runner-x", have("runner-x runs the tool runner's process here"), Put)
f := o.findings[k]
if f.Kind != KindKeptAsRuns || f.Severity != Warning || !strings.Contains(f.Summary, "the tool runner's process") {
t.Fatalf("findings = %v", o.findings)
}
c.t = c.t.Add(time.Minute)
w.Have("ace.runner-x", have("runner-x runs the tool runner's process here"), Put)
if !o.findings[k].Since.Equal(f.Since) {
t.Errorf("a have written again moved the finding's since: %v, was %v", o.findings[k].Since, f.Since)
}
w.Have("ace.prov-y", have("prov-y carries its own secrets sealed"), Put)
w.Have("ace.runner-x", have(), Put) // its manifest says tool-runner now
w.Have("ace.prov-y", nil, Delete)
}, map[string]string{}},
{"WM4: a stuck key not moved is its finding, and its move deletes it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
askProgress(w, "ask1", t0)
expire(w, c, pulseKeyOf(t, checksKey, "ask1"), 120*time.Second)
@@ -447,18 +412,6 @@ func TestTheWatchmansFindings(t *testing.T) {
}
expire(w, c, signalKey("S9", k), AdvisoryQuiet)
}, map[string]string{}},
{"WM3 S21: a heard machine's have failed for one apply interval is apply-failing, naming the package manager; applied again it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.Connect("node.ace")
w.Have("ace.web", haveOf(t, nodeengineclient.PhaseFailed, nodeengineclient.HealthUnknown, "refused: the package manager (pacman) does not answer"), Put)
if len(o.findings) != 0 {
t.Fatalf("raised before its bound: %v", o.findings)
}
expire(w, c, signalKey("S21", "ace"), ApplyEvery+time.Second)
if f := o.findings["machine.ace.apply-failing"]; !strings.Contains(f.Summary, "pacman does not answer") || f.Headline != "ace cannot install packages" {
t.Fatalf("findings = %v", o.findings)
}
w.Have("ace.web", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put)
}, map[string]string{}},
{"WM3 S21: a provisioner client's failed files are apply-failing past the bound; a silent machine's are not", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.Connect("node.ace")
w.Provisioned(ProvReport{Node: "ace", Module: "web", Status: ProvFailed, Reason: "disk full", At: t0.Add(-10 * time.Minute)})
@@ -468,48 +421,6 @@ func TestTheWatchmansFindings(t *testing.T) {
w.Provisioned(ProvReport{Node: "g14", Module: "web", Status: ProvFailed, Reason: "x", At: t0.Add(-10 * time.Minute)})
w.Provisioned(ProvReport{Node: "ace", Module: "web", Status: "applied", At: t0})
}, map[string]string{}},
{"WM3: unhealthy once a failed have is confirmed after a minute (written once and steady); recovered within it none; urgent after four hours", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.Have("ace.web", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "the unit failed"), Put)
if len(o.findings) != 0 {
t.Fatalf("raised before its bound: %v", o.findings)
}
w.Have("ace.quick", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "blip"), Put)
w.Have("ace.quick", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put) // recovered within the bound
expire(w, c, signalKey("unhealthy-confirm", "quick.ace"), UnhealthyConfirmAfter)
expire(w, c, signalKey("unhealthy-confirm", "web.ace"), UnhealthyConfirmAfter) // the one write, never repeated
if o.findings["module.web.ace.unhealthy"].Severity != Warning || len(o.findings) != 1 {
t.Fatalf("findings = %v", o.findings)
}
expire(w, c, signalKey("unhealthy", "web.ace"), UnhealthyUrgentAfter)
w.Have("ace.mail", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthWaits, "a part waits for the secret token"), Put) // needs-operator is not observed here
}, map[string]string{"module.web.ace.unhealthy": Urgent}},
{"WM3: a warning unhealthy finding read back at the start is kept as it was, urgent when its four hours from Since are out", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
seedUnhealthy(t, w, o, Warning, t0.Add(-time.Hour))
w.Have("ace.web", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "x"), Put)
w.Reconcile()
if o.findings["module.web.ace.unhealthy"].Severity != Warning {
t.Fatalf("cleared: findings %v", o.findings)
}
if ttl := o.ttls[signalKey("unhealthy", "web.ace")]; ttl != UnhealthyUrgentAfter-time.Hour {
t.Fatalf("urgent signal restarted from zero: %v", ttl)
}
if _, ok := w.signals[signalKey("unhealthy-confirm", "web.ace")]; ok {
t.Fatal("a standing finding is confirmed again")
}
expire(w, c, signalKey("unhealthy", "web.ace"), UnhealthyUrgentAfter-time.Hour)
}, map[string]string{"module.web.ace.unhealthy": Urgent}},
{"WM3: an urgent unhealthy finding read back at the start stays urgent, with no signal restarted", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
seedUnhealthy(t, w, o, Urgent, t0.Add(-5*time.Hour))
w.Have("ace.web", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "x"), Put)
w.Reconcile()
if o.findings["module.web.ace.unhealthy"].Severity != Urgent || len(w.signals) != 0 {
t.Fatalf("findings %v signals %v", o.findings, w.signals)
}
}, map[string]string{"module.web.ace.unhealthy": Urgent}},
{"WM3: a restart replays a have already failed: it is confirmed after a minute like a new one", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.Have("novox.fail2ban", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "down since 19:40"), Put)
expire(w, c, signalKey("unhealthy-confirm", "fail2ban.novox"), UnhealthyConfirmAfter)
}, map[string]string{"module.fail2ban.novox.unhealthy": Warning}},
{"1954/D2: a provision in waits is its consumer's needs-operator; provided, it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
enc := func(state, why string) []byte {
b, err := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"},
@@ -528,141 +439,6 @@ func TestTheWatchmansFindings(t *testing.T) {
w.Provision("ace.web.oidc", enc(meshprovisionerclient.Waits, "x"), Put)
w.Provision("ace.web.oidc", nil, Delete)
}, map[string]string{}},
{"774: a node's own report: its network, the resolver file rewritten, failed units, a relogin, a directory used as found, a login fallen back", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
report := func(h nodeengine.HealthByNodeValue) []byte {
b, err := nodeengineclient.EncodeHealth(h)
if err != nil {
t.Fatal(err)
}
return b
}
w.Have("novox.nats", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put) // the control node: the bus node
w.NodeHealth("novox", report(nodeengine.HealthByNodeValue{Node: "novox", Network: &nodeengine.NetworkHealth{State: "unhealthy", Parts: []nodeengine.NetworkPart{
{Part: "resolv-conf", State: "unhealthy", Writer: "NetworkManager", Owner: "systemd-resolved", Said: "rewritten"},
{Part: "names", State: "unhealthy", Reason: "names do not resolve.", Said: "x"},
{Part: "route", State: "unhealthy", Reason: "no default route", Said: "y"}}},
Units: &nodeengine.UnitsHealth{State: "degraded", Failed: []nodeengine.FailedUnit{{Unit: "cups.service", Scope: "system", Result: "exit-code"}}},
Relogin: []nodeengine.NodeAccount{{Account: "jochen", Module: "docker", Reason: "a group"}},
Found: []nodeengine.NodeDirectory{{Module: "photos", Resource: "data", Path: "/srv/photos", Reason: "owner differs"}},
Fallback: &nodeengine.LoginFallback{Node: "novox", From: "node.novox--b", Why: "refused twice", At: time.Date(2026, 11, 2, 12, 0, 0, 0, time.UTC)}}), Put)
for key, sev := range map[string]string{"machine.novox.network": Urgent, "machine.novox.systemd-resolved.rewritten": Urgent,
"machine.novox.units": Warning, "module.docker.novox.relogin-needed": Warning, "module.photos.novox.directory-used-as-found": Warning,
"machine.novox.node-account-fallback": Warning} {
if o.findings[key].Severity != sev {
t.Errorf("%s = %+v", key, o.findings[key])
}
}
if f := o.findings["machine.novox.network"]; strings.Contains(f.Summary, "names") {
t.Errorf("the names failing through the rewritten file are said twice: %s", f.Summary)
}
w.NodeHealth("novox", report(nodeengine.HealthByNodeValue{Node: "novox"}), Put) // all well again
}, map[string]string{}},
{"774: a failing part is held at the machines it points at, as the controller holds it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
toward := func(node, part string, targets ...string) []byte {
b, _ := nodeengineclient.EncodeHealth(nodeengine.HealthByNodeValue{Node: node, Network: &nodeengine.NetworkHealth{State: "unhealthy",
Parts: []nodeengine.NetworkPart{{Part: part, State: "unhealthy", Reason: part + " fails", Said: "x", Targets: targets}}}})
return b
}
// own part: one machine alone failing toward a healthy one is its own.
w.NodeHealth("ace", toward("ace", "names", "dns1"), Put)
if _, ok := o.findings["machine.ace.network"]; !ok || len(o.findings) != 1 {
t.Fatalf("own part: findings %v", o.findings)
}
// two resolvers' machines: a second machine failing toward dns1 holds both there, said once as unreachable.
w.NodeHealth("g14", toward("g14", "names", "dns1"), Put)
if f, ok := o.findings["machine.dns1.unreachable"]; !ok || len(f.Also) != 2 || len(o.findings) != 1 {
t.Fatalf("two toward one resolver: findings %v", o.findings)
}
w.NodeHealth("ace", toward("ace", "names", "dns1", "dns2"), Put) // toward two resolvers, one not down: its own
if _, ok := o.findings["machine.ace.network"]; !ok {
t.Fatalf("two resolvers, one up: findings %v", o.findings)
}
w.NodeHealth("ace", nil, Delete)
w.NodeHealth("g14", nil, Delete)
// the hub silent on the record: a tunnel toward it is held there, and its silence says it.
w.Connect("node.hub")
w.Disconnect("node.hub")
expire(w, c, pulseKeyOf(t, presenceNS, "hub"), 120*time.Second)
w.NodeHealth("lap", toward("lap", "tunnel", "hub"), Put)
if _, own := o.findings["machine.lap.network"]; own {
t.Errorf("a tunnel toward the silent hub is the reporter's own: %v", o.findings)
}
if _, ok := o.findings["machine.hub.unreachable"]; ok {
t.Error("the silent hub is said unreachable as well")
}
// design 47 G §706: the silence lists the machines failing toward it in `also`, as `unreachable` does.
if f := o.findings["machine.hub.silent"]; !slices.Equal(f.Also, []string{"lap"}) {
t.Errorf("the hub's silence lists %v in also", f.Also)
}
w.NodeHealth("lap", nil, Delete)
if f := o.findings["machine.hub.silent"]; len(f.Also) != 0 {
t.Errorf("the hub's silence still lists %v once nothing fails toward it", f.Also)
}
w.NodeHealth("lap", toward("lap", "tunnel", "hub"), Put)
// the hub's own network open, two machines failing toward it: its `network` finding lists both.
w.NodeHealth("hub", toward("hub", "bus"), Put)
w.NodeHealth("ace", toward("ace", "tunnel", "hub"), Put)
if f := o.findings["machine.hub.network"]; !slices.Equal(f.Also, []string{"ace", "lap"}) {
t.Errorf("the hub's network lists %v in also: %v", f.Also, o.findings)
}
w.NodeHealth("hub", nil, Delete)
w.NodeHealth("ace", nil, Delete)
w.Connect("node.hub") // back: lap's tunnel failing toward a hub up is its own again
if _, own := o.findings["machine.lap.network"]; !own {
t.Errorf("hub back: findings %v", o.findings)
}
w.NodeHealth("lap", nil, Delete)
}, map[string]string{}},
{"774: a node two nodes' tunnels no longer see is unreachable, said once there; not while it is silent", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
for node, key := range map[string]string{"ace": "kA", "g14": "kG", "lap": "kL"} {
b, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: node, TunnelKey: key}})
w.Have(node+".node", b, Put)
}
peers := func(from string, seen bool) []byte {
b, _ := nodeengineclient.EncodeHealth(nodeengine.HealthByNodeValue{Node: from, Peers: []nodeengine.PeerSeen{{Key: "kL", Seen: seen}}})
return b
}
w.NodeHealth("ace", peers("ace", false), Put)
if len(o.findings) != 0 {
t.Fatalf("one node alone said it: %v", o.findings)
}
w.NodeHealth("g14", peers("g14", false), Put)
if f := o.findings["machine.lap.unreachable"]; len(f.Also) != 2 {
t.Fatalf("findings = %v", o.findings)
}
w.Connect("node.lap")
w.Disconnect("node.lap")
expire(w, c, pulseKeyOf(t, presenceNS, "lap"), 120*time.Second)
if _, ok := o.findings["machine.lap.unreachable"]; ok {
t.Error("a silent node is also said unreachable")
}
w.Connect("node.lap")
w.NodeHealth("g14", peers("g14", true), Put)
}, map[string]string{}},
{"WM3: a have removed ends its findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.Have("ace.mail", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "x"), Put)
expire(w, c, signalKey("unhealthy-confirm", "mail.ace"), UnhealthyConfirmAfter)
w.Have("ace.mail", haveOf(t, nodeengineclient.PhaseRemoved, nodeengineclient.HealthUnknown, ""), Put)
}, map[string]string{}},
{"WM2 S1: a control node silent is urgent after half an hour", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
// 1126 (ADR 0590): the control node is the bus node, the one whose have holds the bus module; a have of the
// ended controller names none.
w.Have("g14.mesh-controller", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put)
w.Connect("node.g14")
w.Disconnect("node.g14")
expire(w, c, pulseKeyOf(t, presenceNS, "g14"), 120*time.Second)
if _, ok := w.signals[signalKey("S1", "g14")]; ok {
t.Fatalf("a node running the controller's have is held a control node")
}
w.Have("novox.nats", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthOK, ""), Put)
w.Connect("node.novox")
w.Disconnect("node.novox")
expire(w, c, pulseKeyOf(t, presenceNS, "novox"), 120*time.Second)
if o.findings["machine.novox.silent"].Severity != Warning {
t.Fatalf("findings = %v", o.findings)
}
expire(w, c, signalKey("S1", "novox"), ControlNodeUrgentAfter)
}, map[string]string{"machine.novox.silent": Urgent, "machine.g14.silent": Warning}},
{"WM3 S9: the bus module's closures, each counted once, ended an hour after the newest", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
cl := BusClosure{User: "ace.web", CID: 7, Reason: "Slow Consumer (Write Deadline)", Opened: t0.Add(-time.Hour), Closed: t0}
w.Closed(KindSlowConsumer, cl)
@@ -697,14 +473,6 @@ func TestTheWatchmansFindings(t *testing.T) {
t.Fatal("still counted again with nothing held")
}
}, map[string]string{}},
{"WM4: resync after a cleared bucket writes the same findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.Have("ace.mail", haveOf(t, nodeengineclient.PhaseRunning, nodeengineclient.HealthFailed, "x"), Put)
expire(w, c, signalKey("unhealthy-confirm", "mail.ace"), UnhealthyConfirmAfter)
o.findings = map[string]Finding{}
if n := w.Rewrite(); n != 1 {
t.Fatalf("rewrote %d", n)
}
}, map[string]string{"module.mail.ace.unhealthy": Warning}},
{"WM4: a finding read at the start and not observed again is deleted; a signal's stays", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
for _, f := range []Finding{
{Scope: ScopeDelivery, ID: "x.y", Kind: KindStuck, Severity: Warning, Summary: "old", Source: "stuck"},
@@ -733,7 +501,7 @@ func TestTheWatchmansFindings(t *testing.T) {
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Asked, At: t0})
provided, _ := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "oidc", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"},
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Provided, At: t0})
w.Have("ace.web", haveValue(t, fpOld), Put) // the node runs what the held want names: the hold is still the wait
w.EngineRunning("ace", "web", []byte(`{"fingerprint":"`+fpOld+`"}`)) // the node runs what the held want names: the hold is still the wait
w.Want("want.ace.web", held, Put, t0)
w.Provision("ace.photos.oidc", asked, Put)
waits := map[string]string{AssignerWantsBucket: "want.ace.web", ProvisionsBucket: "ace.photos.oidc"}
@@ -783,14 +551,6 @@ func TestTheWatchmansFindings(t *testing.T) {
}
}
func seedUnhealthy(t *testing.T, w *Watchman, o *fakeOut, sev string, since time.Time) {
t.Helper()
f := Finding{Scope: ScopeModule, ID: "web.ace", Token: "unhealthy", Kind: KindUnhealthy, Machine: "ace", Severity: sev, Source: "health", Since: since, Summary: "web on ace is not healthy: x"}
v, _ := EncodeFinding(f)
o.findings[f.Key()], _ = DecodeFinding(v)
w.SeedFinding(f.Key(), v)
}
const (
revKey = "m.00000000000000000042"
groupKey = "g.feat/take-in"
@@ -827,9 +587,9 @@ func takeInBody(t *testing.T, kind, key string, since time.Time) []byte {
func TestAFindingIsTheContractsWithItsRefInItsID(t *testing.T) {
since := time.Date(2026, 11, 11, 20, 0, 0, 0, time.UTC)
for _, f := range []Finding{
{Scope: ScopeModule, ID: "web.ace", Token: "unhealthy", Kind: KindUnhealthy, Machine: "ace", Severity: Warning, Source: "health", Since: since, Summary: "s", Said: "said"},
{Scope: ScopeModule, ID: "web.ace", Token: "unhealthy", Kind: KindStuck, Machine: "ace", Severity: Warning, Source: "stuck", Since: since, Summary: "s", Said: "said"},
{Scope: ScopeModule, ID: "web", Token: KindBusStateOverdue + "-state", Kind: KindBusStateOverdue, Severity: Warning, Source: "bus", Since: since, Summary: "s"},
{Scope: ScopeMachine, ID: "ace", Kind: KindMachineUnits, Severity: Warning, Source: "units", Since: since, Summary: "s", Needs: "mend"},
{Scope: ScopeMachine, ID: "ace", Kind: KindSilent, Severity: Warning, Source: "S1", Since: since, Summary: "s", Needs: "mend"},
} {
c := f.Contract()
wantID := f.ID
modules/mesh-delivery-watchman/internal/watch/mirror.go
+12 -28
View File
@@ -164,46 +164,30 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
describe("the node agent of %s to run %s at %s", node, module, shortFP(want.Fp)), want.Fp, at, 0)
}
// Have is a change of a key of the node-engine's have bucket: the fingerprint a node runs of a module. One that names the
// fingerprint of the node's want for the module ends the want's pulse. A node's own key and a key deleted are left.
func (w *Watchman) Have(key string, value []byte, op Op) {
node, module, isNode, err := nodeengineclient.ParseHaveKey(key)
// EngineRunning hears the node-engine of node saying `running` for module (design 47 G §541, A-192, A-197): the fingerprint it runs, or
// that it runs none (stopped). One that names the fingerprint of the node's want for the module ends the want's pulse.
func (w *Watchman) EngineRunning(node, module string, body []byte) {
var r nodeengine.Running
if err := json.Unmarshal(body, &r); err != nil {
w.say("a running of the node-engine of %s for %s does not read, left: %v", node, module, err)
return
}
key, err := nodeengineclient.HaveKey(node, module)
if err != nil {
return
}
w.mu.Lock()
defer w.mu.Unlock()
if isNode {
// The node's own record: its tunnel key, whose peers' sight of it is network-unreachable.
var record *nodeengine.NodeRecord
if op == Put {
if hn, err := nodeengineclient.DecodeHaveNode(value); err == nil {
record = hn.Record
}
}
w.tunnelKey(node, record)
return
}
switch op {
case Expired:
return
case Delete:
if r.Stopped || r.Fingerprint == "" {
delete(w.haves, key)
w.haveFindings(key, node, module, nil)
return
}
have, err := nodeengineclient.DecodeHave(value)
if err != nil {
w.say("the have %s cannot be read, and is left: %v", key, err)
return
}
w.haves[key] = have.Fp
w.haveFindings(key, node, module, &have)
w.haves[key] = r.Fingerprint
wk, err := meshwantsclient.WantKey(node, module)
if err != nil {
return
}
if fp, ok := w.wants[wk]; ok && fp == have.Fp {
if fp, ok := w.wants[wk]; ok && fp == r.Fingerprint {
w.moved(AssignerWantsBucket, wk)
}
}
modules/mesh-delivery-watchman/internal/watch/names.go
-3
View File
@@ -20,9 +20,6 @@ const (
PulseBucket = WatchmanModule + "_" + meshwatchdogs.StatePulse
StuckBucket = WatchmanModule + "_" + meshwatchdogs.StateStuck
// HaveBucket and NodeHealthBucket are the node-engine's states have and health (reads: node-engine.have, mesh-host.health).
HaveBucket = "node-engine_have"
NodeHealthBucket = "mesh-host_health"
// AssignerWantsBucket is the assigner's state wants, followed by its views family (consumes: mesh-delivery-assigner.views.wants).
AssignerWantsBucket = AssignerModule + "_wants"
)
modules/mesh-delivery-watchman/internal/watch/nodehealth.go
-485
View File
@@ -1,485 +0,0 @@
package watch
import (
"fmt"
"maps"
"slices"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
)
// A node's own health report (novox/hq ADR 0241, issue 315, the operator's alert table of 2026-10-29): each node's engine
// writes `mesh-host_health.<node>` when what it says changes, and the watchman raises from it what the controller raised
// from the passing health message, under the same keys and words (mesh-control cmd/mesh-controller machine_network.go,
// machine_units.go, module_health.go at e04846e9):
// - machine-network `machine.<node>.network`: the machine's own unhealthy parts; urgent on the control node or for the bus;
// - network-rewritten `machine.<node>[.<owner>].rewritten`: the resolver file rewritten by another program;
// - network-unreachable `machine.<node>.unreachable`: two or more nodes' tunnels no longer see it (their peers' handshake),
// said once there, never while its own network or its silence says it;
// - machine-units `machine.<node>.units`: failed units no module places;
// - relogin-needed `module.<module>.<node>.relogin-needed` and directory-used-as-found
// `module.<module>.<node>.directory-used-as-found`;
// - node-account-fallback `machine.<node>.node-account-fallback`: the engine fell back to its node's old login during a
// rotation (novox/hq ADR 0583 §6), raised while its report carries the fall back and cleared once it no longer does.
//
// A failing part that points at other mesh machines (NetworkPart.Targets: the hub's node, each resolver's node) is held
// there, as the controller holds it: when every machine it points at is down on the record (silent, its own network
// unhealthy, or two machines failing toward it), it is that machine's `unreachable`, said once there, and no finding of
// the reporter's. A tunnel peer two or more nodes no longer see points at its node the same way.
// The kinds a node's health report raises.
const (
KindMachineNetwork = "machine-network"
KindNetworkRewritten = "network-rewritten"
KindNetworkUnreachable = "network-unreachable"
KindMachineUnits = "machine-units"
KindReloginNeeded = "relogin-needed"
KindUsedAsFound = "directory-used-as-found"
KindAccountFallback = FindingNodeAccountFallback
)
// The parts and states a report names (mesh-control internal/link).
const (
partResolvConf = "resolv-conf"
partNames = "names"
partTunnel = "tunnel"
partBus = "bus"
stateUnhealthy = "unhealthy"
)
func stampSec(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05 MST") }
// NodeHealth is a change of a node's report: written, or deleted (value nil).
func (w *Watchman) NodeHealth(node string, value []byte, op Op) {
w.mu.Lock()
defer w.mu.Unlock()
if op == Put {
h, err := nodeengineclient.DecodeHealth(value)
if err != nil {
w.say("the health report of %s cannot be read, and is left: %v", node, err)
return
}
w.health[node] = h
} else {
delete(w.health, node)
}
w.nodeFindings(node)
w.networks()
}
// TunnelKey is a node's own tunnel key, from its record in its have: whose peers see it is read by it.
func (w *Watchman) tunnelKey(node string, record *nodeengine.NodeRecord) {
key := ""
if record != nil {
key = record.TunnelKey
}
if w.tunnels[node] == key {
return
}
if key == "" {
delete(w.tunnels, node)
} else {
w.tunnels[node] = key
}
w.networks()
}
// nodeFindings writes and deletes one node's own findings from its report.
func (w *Watchman) nodeFindings(node string) {
h, ok := w.health[node]
want := map[string]bool{}
if ok {
if f, raise := unitsOf(node, h.Units); raise {
want[f.Key()] = true
w.find(f)
}
for _, f := range reloginOf(node, h.Relogin) {
want[f.Key()] = true
w.find(f)
}
for _, f := range foundOf(node, h.Found) {
want[f.Key()] = true
w.find(f)
}
if f, raise := fallbackOf(node, h.Fallback); raise {
want[f.Key()] = true
w.find(f)
}
}
for key := range w.findings {
if !want[key] && w.ownsHealthKey(key, node) {
w.unfind(key)
}
}
}
// ownsHealthKey says a finding key is one of node's own report findings.
func (w *Watchman) ownsHealthKey(key, node string) bool {
f, err := DecodeFinding(w.findings[key])
if err != nil || f.Machine != node || f.Source != "health-report" {
return false
}
return f.Kind != KindNetworkUnreachable && f.Kind != KindMachineNetwork && f.Kind != KindNetworkRewritten
}
func (w *Watchman) rewrittenOf(m string, p nodeengine.NetworkPart, all []nodeengine.NetworkPart) Finding {
writer := ""
if p.Writer != "" {
writer = " (" + p.Writer + ")"
}
cost := "the names still resolve through what it wrote"
said := []string{p.Part + ": " + p.Said}
for _, q := range all {
if q.Part == partNames {
cost = strings.TrimSuffix(q.Reason, ".")
said = append(said, q.Part+": "+q.Said)
}
}
id := m
summary := fmt.Sprintf("the resolver file on %s was rewritten by another program%s — %s until the node-engine writes it "+
"back at its next reconcile, or that program gives it back", m, writer, cost)
if p.Owner != "" {
id = m + "." + p.Owner
summary = fmt.Sprintf("the resolver file %s writes on %s was rewritten by another program%s — %s until the node-engine "+
"writes it back at its next reconcile, or that program gives it back", p.Owner, m, writer, cost)
}
severity := Warning
if w.control[m] {
severity = Urgent
}
return Finding{Scope: ScopeMachine, ID: id, Token: "rewritten", Kind: KindNetworkRewritten, Machine: m, Severity: severity,
Source: "health-report", Since: p.Since, Summary: summary,
Said: fmt.Sprintf("since %s: %s", stampSec(p.Since), strings.Join(said, " | "))}
}
// unitsOf is a machine's failed units no module places, one finding.
func unitsOf(node string, u *nodeengine.UnitsHealth) (Finding, bool) {
if u == nil || len(u.Failed) == 0 {
return Finding{}, false
}
var names, said []string
since := u.Failed[0].Since
for _, f := range u.Failed {
where := ""
if f.Scope == "user" {
where = " (the account's own manager)"
}
names = append(names, f.Unit+where)
scope, result := f.Scope, f.Result
if scope == "" {
scope = "system"
}
if result == "" {
result = "no reason said"
}
line := fmt.Sprintf("%s in the %s manager: %s", f.Unit, scope, result)
if f.Load != "" && f.Load != "loaded" {
line += ", its unit " + f.Load
}
if f.Resource != "" {
line += ", named by the mesh's own " + f.Resource
}
said = append(said, line+", since "+stampSec(f.Since))
if f.Since.Before(since) {
since = f.Since
}
}
failed := "1 failed service"
if len(u.Failed) > 1 {
failed = fmt.Sprintf("%d failed services", len(u.Failed))
}
return Finding{Scope: ScopeMachine, ID: node, Token: "units", Kind: KindMachineUnits, Machine: node, Severity: Warning,
Source: "health-report", Since: since,
Summary: fmt.Sprintf("%s's service manager is degraded: %d failed unit(s) no module places — %s. Each is the machine's own: "+
"mend or remove it there, or have a module place it", node, len(u.Failed), strings.Join(names, ", ")),
Said: strings.Join(said, "; "),
Headline: strings.ToUpper(failed[:1]) + failed[1:] + " on " + node,
Needs: fmt.Sprintf("mend or remove them on %s, or silence this if they do not matter.", node)}, true
}
// reloginOf is one finding per module whose accounts wait for a new login.
func reloginOf(node string, accounts []nodeengine.NodeAccount) []Finding {
by := map[string][]nodeengine.NodeAccount{}
for _, a := range accounts {
by[a.Module] = append(by[a.Module], a)
}
var out []Finding
for _, module := range slices.Sorted(maps.Keys(by)) {
var names []string
since := by[module][0].Since
for _, a := range by[module] {
names = append(names, a.Account)
if a.Since.Before(since) {
since = a.Since
}
}
sort.Strings(names)
said := fmt.Sprintf("relogin needed on %s: %s waits for a new login of %s, which is in its group and whose login began "+
"before it was; log out of %[1]s completely and log in again, or restart it", node, orModule(module), strings.Join(names, ", "))
out = append(out, Finding{Scope: ScopeModule, ID: orModule(module) + "." + node, Kind: KindReloginNeeded, Machine: node,
Severity: Warning, Source: "health-report", Since: since, Summary: said, Said: said})
}
return out
}
// foundOf is one finding per module with a directory used as found.
func foundOf(node string, dirs []nodeengine.NodeDirectory) []Finding {
by := map[string][]nodeengine.NodeDirectory{}
for _, d := range dirs {
by[d.Module] = append(by[d.Module], d)
}
var out []Finding
for _, module := range slices.Sorted(maps.Keys(by)) {
var ids, why []string
since := by[module][0].Since
for _, d := range by[module] {
ids = append(ids, d.Resource)
why = append(why, d.Reason)
if d.Since.Before(since) {
since = d.Since
}
}
m := orModule(module)
said := fmt.Sprintf("%s on %s uses %s as found: %s", m, node, strings.Join(ids, ", "), strings.Join(why, "; "))
out = append(out, Finding{Scope: ScopeModule, ID: m + "." + node, Kind: KindUsedAsFound, Machine: node, Severity: Warning,
Source: "health-report", Since: since, Summary: said, Said: said,
Headline: fmt.Sprintf("%s waits for a directory on %s", m, node),
Explanation: fmt.Sprintf("Needs you: hand the directory over from the control-node, as the operator; the details name it "+
"and the line to type. A directory of %s was already on %s, with another owner or mode than %s declares.", m, node, m),
Needs: "hand the directory over from the control-node, as the operator; the details name it and the line to type."})
}
return out
}
// fallbackOf is a node's fall back to its old login during a rotation of its account (ADR 0583 §6), one finding.
func fallbackOf(node string, f *nodeengine.LoginFallback) (Finding, bool) {
if f == nil || f.Node != node {
return Finding{}, false
}
said := fmt.Sprintf("%s fell back from its new login %s to its old one during a rotation of its bus account: %s", node, f.From, f.Why)
return Finding{Scope: ScopeMachine, ID: node, Token: KindAccountFallback, Kind: KindAccountFallback, Machine: node, Severity: Warning,
Source: "health-report", Since: f.At, Summary: said, Said: said,
Headline: fmt.Sprintf("%s fell back to its old bus login", node),
Needs: fmt.Sprintf("abandon the rotation of %s's account, or rotate it again once its new login authenticates.", node)}, true
}
func orModule(m string) string {
if m == "" {
return "_"
}
return m
}
// silentAlso sets the machines `also` lists on the open `silent` finding of x, when there is one.
func (w *Watchman) silentAlso(x string, also []string) {
value, ok := w.findings[FindingKey(ScopeMachine, x, KindSilent)]
if !ok {
return
}
f, err := DecodeFinding(value)
if err != nil || slices.Equal(f.Also, also) {
return
}
f.Also = also
w.find(f)
}
// keepSilentAlso gives a `silent` finding about to be written the `also` the open one holds, so a write of its other
// words (its urgency after half an hour) never drops the machines listed there.
func (w *Watchman) keepSilentAlso(f Finding) Finding {
if f.Kind != KindSilent {
return f
}
if value, ok := w.findings[f.Key()]; ok {
if cur, err := DecodeFinding(value); err == nil {
f.Also = cur.Also
}
}
return f
}
// pointed is one machine's failing part that points at another machine.
type pointed struct {
from string
part nodeengine.NetworkPart
}
// networks is every machine's network findings from every report together (the controller's networkObservations): the
// resolver file rewritten, each machine's own unhealthy parts, and the machine others point at, held there.
func (w *Watchman) networks() {
machines := slices.Sorted(maps.Keys(w.health))
unhealthy := func(m string) []nodeengine.NetworkPart {
h := w.health[m]
if h.Network == nil {
return nil
}
var out []nodeengine.NetworkPart
for _, p := range h.Network.Parts {
if p.State == stateUnhealthy {
out = append(out, p)
}
}
return out
}
targets := func(m string, p nodeengine.NetworkPart) ([]string, bool) {
var out []string
for _, x := range p.Targets {
if x == "" || x == m {
return nil, false
}
if !slices.Contains(out, x) {
out = append(out, x)
}
}
return out, len(out) > 0
}
pointing := map[string][]pointed{}
for _, m := range machines {
for _, p := range unhealthy(m) {
if xs, ok := targets(m, p); ok {
for _, x := range xs {
pointing[x] = append(pointing[x], pointed{from: m, part: p})
}
}
}
}
// A tunnel peer a node no longer sees points at that peer's node, as a tunnel part toward it would.
byKey := map[string]string{}
for node, key := range w.tunnels {
byKey[key] = node
}
for _, m := range machines {
for _, p := range w.health[m].Peers {
if x, ok := byKey[p.Key]; ok && x != m && !p.Seen {
pointing[x] = append(pointing[x], pointed{from: m, part: nodeengine.NetworkPart{Part: partTunnel, State: stateUnhealthy,
Said: "no handshake with " + x + " within five minutes", Targets: []string{x}}})
}
}
}
from := func(x string) []string {
var out []string
for _, pt := range pointing[x] {
if !slices.Contains(out, pt.from) {
out = append(out, pt.from)
}
}
sort.Strings(out)
return out
}
silent := func(x string) bool {
_, ok := w.findings[FindingKey(ScopeMachine, x, KindSilent)]
return ok
}
down := func(x string) bool { return silent(x) || len(unhealthy(x)) > 0 || len(from(x)) >= 2 }
allDown := func(xs []string) bool {
for _, x := range xs {
if !down(x) {
return false
}
}
return len(xs) > 0
}
// A machine silent on the record holds the failures toward it under its silence, so its finding lists in `also` the
// machines whose failing parts point at it, as `unreachable` does (design 47 G §706); a change of `also` rewrites it.
for key, value := range w.findings {
if f, err := DecodeFinding(value); err == nil && f.Kind == KindSilent && key == f.Key() {
w.silentAlso(f.ID, from(f.ID))
}
}
want := map[string]bool{}
saysOwn := map[string]bool{}
for _, m := range machines {
parts := unhealthy(m)
if len(parts) == 0 {
continue
}
var own []nodeengine.NetworkPart
var rewritten *nodeengine.NetworkPart
for i, p := range parts {
if p.Part == partResolvConf {
rewritten = &parts[i]
continue
}
if xs, ok := targets(m, p); ok && allDown(xs) {
continue // held at the machines it points at
}
own = append(own, p)
}
if rewritten != nil {
f := w.rewrittenOf(m, *rewritten, parts)
want[f.Key()] = true
w.find(f)
kept := own[:0]
for _, p := range own {
if p.Part != partNames {
kept = append(kept, p)
}
}
own = kept
}
if len(own) == 0 {
continue
}
var words, said []string
severity := Warning
for _, p := range own {
words = append(words, p.Reason)
said = append(said, fmt.Sprintf("%s since %s: %s", p.Part, stampSec(p.Since), p.Said))
if p.Part == partBus {
severity = Urgent
}
}
if w.control[m] {
severity = Urgent
}
summary := fmt.Sprintf("%s's network is not healthy: %s", m, strings.Join(words, "; "))
if waiting := from(m); len(waiting) > 0 {
severity = Urgent
summary += fmt.Sprintf("; %s cannot reach it", strings.Join(waiting, ", "))
}
f := Finding{Scope: ScopeMachine, ID: m, Token: "network", Kind: KindMachineNetwork, Machine: m, Also: from(m), Severity: severity,
Source: "health-report", Since: w.health[m].Network.Since, Summary: summary, Said: strings.Join(said, " | ")}
want[f.Key()] = true
saysOwn[m] = true
w.find(f)
}
for _, x := range slices.Sorted(maps.Keys(pointing)) {
if !down(x) || saysOwn[x] || silent(x) {
continue
}
var what, said []string
for _, pt := range pointing[x] {
wd := map[string]string{partTunnel: "the tunnel to it", partBus: "the bus on it", partNames: "its resolver"}[pt.part.Part]
if wd == "" {
wd = pt.part.Part
}
if !slices.Contains(what, wd) {
what = append(what, wd)
}
said = append(said, fmt.Sprintf("%s: %s: %s", pt.from, pt.part.Part, pt.part.Said))
}
severity := Warning
if w.control[x] || slices.Contains(what, "the bus on it") {
severity = Urgent
}
f := Finding{Scope: ScopeMachine, ID: x, Token: "unreachable", Kind: KindNetworkUnreachable, Machine: x, Also: from(x),
Severity: severity, Source: "health-report",
Summary: fmt.Sprintf("%s cannot be reached from %s: %s", x, strings.Join(from(x), ", "), strings.Join(what, ", ")),
Said: strings.Join(said, " | ")}
want[f.Key()] = true
w.find(f)
}
for key, value := range w.findings {
f, err := DecodeFinding(value)
if err != nil || f.Source != "health-report" {
continue
}
if (f.Kind == KindMachineNetwork || f.Kind == KindNetworkRewritten || f.Kind == KindNetworkUnreachable) && !want[key] {
w.unfind(key)
}
}
}
modules/mesh-delivery-watchman/internal/watch/signals.go
+22 -180
View File
@@ -8,8 +8,6 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient"
)
@@ -45,14 +43,8 @@ const (
HeartbeatEvery = 60 * time.Second
// ApplyEvery is one apply interval (S21): a failure still there after it is not one the next apply clears.
ApplyEvery = 5 * time.Minute
// UnhealthyConfirmAfter is how long a have must keep saying failed before the module is said unhealthy.
UnhealthyConfirmAfter = time.Minute
// UnhealthyUrgentAfter is how long a module stays unhealthy before it is urgent.
UnhealthyUrgentAfter = 4 * time.Hour
// SilentBound is the bound S1 says, three of the intervals a machine has always used, and ControlNodeUrgentAfter
// how long a control node may be silent before it is urgent (the controller's silentBound, controlNodeUrgentAfter).
SilentBound = 3 * HeartbeatEvery
ControlNodeUrgentAfter = 30 * time.Minute
// SilentBound is the bound S1 says, three of the intervals a machine has always used (the controller's silentBound).
SilentBound = 3 * HeartbeatEvery
)
// signalBucket is the pseudo-bucket of a signal's pulse key.
@@ -179,27 +171,16 @@ func (w *Watchman) signalExpired(pk string) {
if due.While != "" && !w.stillWrong(due.While, due.Finding.Kind) {
return
}
w.find(w.keepSilentAlso(due.Finding))
w.find(due.Finding)
}
}
// stillWrong says a have's key still says what a row's finding was waited on: failed for apply-failing, unhealthy for
// unhealthy.
func (w *Watchman) stillWrong(haveKey, kind string) bool {
if node, ok := strings.CutPrefix(haveKey, "silent:"); ok {
// stillWrong says a silent machine's finding still stands, which a row's finding waited on.
func (w *Watchman) stillWrong(key, _ string) bool {
if node, ok := strings.CutPrefix(key, "silent:"); ok {
_, open := w.findings[FindingKey(ScopeMachine, node, KindSilent)]
return open
}
h, ok := w.haveState[haveKey]
if !ok {
return false
}
switch kind {
case KindApplyFailing:
return h.Phase == nodeengineclient.PhaseFailed
case KindUnhealthy:
return h.Health == nodeengineclient.HealthFailed
}
return false
}
@@ -212,10 +193,8 @@ const (
KindSelfCheckSilent = "self-check-silent"
KindToolsSilent = "tools-silent"
KindApplyFailing = "apply-failing"
KindUnhealthy = "module-unhealthy"
KindNeedsOperator = "needs-operator"
KindDeadLetters = "max-deliveries"
KindKeptAsRuns = "kept-as-it-runs"
)
func stamp(t time.Time) string { return t.UTC().Format("2006-01-02 15:04 MST") }
@@ -264,7 +243,6 @@ func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) {
} else {
w.machineAway(node, s.Since)
}
defer w.networks() // a silent machine is down on the record: what points at it is held there, said by its silence
if s.Moved || w.away[node] {
w.unfind(key)
w.unsignal("S1", node)
@@ -274,18 +252,7 @@ func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) {
f := Finding{Scope: ScopeMachine, ID: node, Kind: KindSilent, Machine: node, Severity: Warning, Source: "S1",
Summary: fmt.Sprintf("%s has not been heard from since %s (bound %s)", node, stamp(s.Since), SilentBound),
Said: fmt.Sprintf("offline since %s", s.Since.UTC().Format(time.RFC3339)), Since: s.Since}
// A control node, the bus node (ControlNodeModule), is urgent after half an hour silent, as the controller
// says it (its watchHeartbeats): the expiry of a pulse from its silence's start, never a loop.
if w.control[node] {
if age := w.now().Sub(s.Since); age > ControlNodeUrgentAfter {
f.Severity = Urgent
} else {
u := f
u.Severity = Urgent
w.signal("S1", node, ControlNodeUrgentAfter-age+time.Second, signalDue{Do: dueFind, While: "silent:" + node, Finding: u})
}
}
w.find(w.keepSilentAlso(f))
w.find(f)
w.applyCheck(node)
return
}
@@ -478,114 +445,7 @@ func ToolsNode(subject string) string {
return node
}
// ---- S21 and unhealthy, from the haves and the provisioner clients' reports ----
// haveFindings reads a have's change for the rows the haves feed: S21 (the machine's applies failing, applyCheck),
// unhealthy (two writes in a row, urgent after four hours), and where the bus module runs (S1's control node). A have
// removed or deleted ends its part of them.
//
// needs-operator is not observed from the haves: a module's part that waits for the operator is a provision in the state
// `waits` of the seat mesh-provisioner (Provision, ADR 0570 §7).
func (w *Watchman) haveFindings(key string, node, module string, h *nodeengine.HaveByNodeModuleValue) {
unhealthyKey := FindingKey(ScopeModule, module+"."+node, "unhealthy")
if module == ControlNodeModule {
w.control[node] = h != nil && h.Phase != nodeengineclient.PhaseRemoved
}
if h == nil || h.Phase == nodeengineclient.PhaseRemoved {
delete(w.haveState, key)
delete(w.unhealthy, key)
delete(w.failedSince, key)
w.unsignal("unhealthy-confirm", module+"."+node)
w.unsignal("unhealthy", module+"."+node)
w.unfind(unhealthyKey)
w.keptFinding(key, node, module, nil)
w.applyCheck(node)
return
}
w.haveState[key] = *h
now := w.now()
w.keptFinding(key, node, module, h.Kept)
// S21: a have refused or applied in part says phase failed, with the node's words in its detail.
if h.Phase == nodeengineclient.PhaseFailed {
if _, ok := w.failedSince[key]; !ok {
w.failedSince[key] = now
}
} else {
delete(w.failedSince, key)
}
w.applyCheck(node)
// unhealthy: a have that says failed and still says it after UnhealthyConfirmAfter (ADR 0564 §4's confirmation, by time:
// a have written once and then unchanged never gets a second write). The confirm signal's expiry writes the warning; a
// signal already standing (kept across a restart) is not renewed, so a steady have does not push its own expiry out.
if h.Health == nodeengineclient.HealthFailed {
if w.unhealthy[key] == 0 {
w.unhealthy[key] = 1
f := Finding{Scope: ScopeModule, ID: module + "." + node, Token: "unhealthy", Kind: KindUnhealthy, Machine: node,
Severity: Warning, Source: "health", Since: now,
Summary: fmt.Sprintf("%s on %s is not healthy: %s", module, node, orWords(h.Detail, "its check failed")),
Said: firstLine(orWords(h.Detail, "failed")),
Headline: fmt.Sprintf("%s not working on %s", module, node)}
id := module + "." + node
urgentIn := UnhealthyUrgentAfter
if w.seededFindings[unhealthyKey] {
// Read back at the start: it stands, so it is observed again, kept as it was (not cleared and raised again), and
// its four hours run on from its own Since, not from zero; one that already went urgent needs no signal.
delete(w.seededFindings, unhealthyKey)
old, err := DecodeFinding(w.findings[unhealthyKey])
if err == nil && old.Severity == Urgent {
urgentIn = 0
} else if err == nil && !old.Since.IsZero() {
urgentIn = max(time.Second, UnhealthyUrgentAfter-now.Sub(old.Since))
}
} else if _, ok := w.signals[signalKey("unhealthy-confirm", id)]; !ok {
w.signal("unhealthy-confirm", id, UnhealthyConfirmAfter, signalDue{Do: dueFind, While: key, Finding: f})
}
if _, ok := w.signals[signalKey("unhealthy", id)]; !ok && urgentIn > 0 {
f.Severity = Urgent
w.signal("unhealthy", id, urgentIn, signalDue{Do: dueFind, While: key, Finding: f})
}
}
} else {
delete(w.unhealthy, key)
w.unsignal("unhealthy-confirm", module+"."+node)
w.unsignal("unhealthy", module+"."+node)
w.unfind(unhealthyKey)
}
}
// keptFinding raises, as a warning, what a node keeps of a module as it runs because its manifest does not yet say what shared
// code no longer infers (its have's Kept, novox/hq ADR 0601 §4: the tool runner, the core's credentials), naming each; and
// clears it once the have says none, or is gone. Its since is the first have that said it, so a have written again moves
// nothing.
func (w *Watchman) keptFinding(key, node, module string, kept []string) {
id := module + "." + node
fkey := FindingKey(ScopeModule, id, "kept")
if len(kept) == 0 {
delete(w.keptSince, key)
w.unfind(fkey)
return
}
if w.keptSince == nil {
w.keptSince = map[string]time.Time{}
}
since, ok := w.keptSince[key]
if !ok {
since = w.now()
w.keptSince[key] = since
}
words := strings.Join(kept, "; ")
w.find(Finding{Scope: ScopeModule, ID: id, Token: "kept", Kind: KindKeptAsRuns, Machine: node, Severity: Warning,
Source: "node-engine", Since: since, Summary: fmt.Sprintf("%s on %s is kept as it runs: %s", module, node, words),
Said: firstLine(words), Headline: fmt.Sprintf("%s kept as it runs on %s", module, node),
Needs: "a build of " + module + " whose manifest says it (tool-runner, core-credentials)"})
}
// ControlNodeModule is the module whose have names a control node: the bus module, so the control node is the bus node,
// the one stably reachable node every part depends on (novox/hq ADR 0590, issue 1126: the controller that named it ended).
// The same name the bus module's events are heard under (SlowConsumerSubject).
const ControlNodeModule = "nats"
// ---- S21, from the provisioner clients' reports ----
// Provision is a change of a key of mesh-provisioner's `provisions` (ADR 0570 §7): a provision in `waits` is a part of its
// consumer waiting for the operator, said as `module.<consumer>.<node>.needs-operator` naming each waiting provision and
@@ -713,24 +573,12 @@ func (w *Watchman) heard(node string) bool {
func (w *Watchman) applyCheck(node string) {
key := FindingKey(ScopeMachine, node, KindApplyFailing)
type failure struct{ module, words string }
var applied, placed []failure
var placed []failure
var since time.Time
for hk, at := range w.failedSince {
n, module, _, err := nodeengineclient.ParseHaveKey(hk)
if err != nil || n != node {
continue
}
applied = append(applied, failure{module, firstLine(orWords(w.haveState[hk].Detail, "no reason given"))})
if since.IsZero() || at.Before(since) {
since = at
}
}
if len(applied) == 0 {
for module, r := range w.provFailed[node] {
placed = append(placed, failure{module, firstLine(orWords(r.Reason, "no reason given"))})
if since.IsZero() || r.At.Before(since) {
since = r.At
}
for module, r := range w.provFailed[node] {
placed = append(placed, failure{module, firstLine(orWords(r.Reason, "no reason given"))})
if since.IsZero() || r.At.Before(since) {
since = r.At
}
}
if since.IsZero() || !w.heard(node) {
@@ -745,7 +593,7 @@ func (w *Watchman) applyCheck(node string) {
}
return
}
failed := append(applied, placed...)
failed := placed
sort.Slice(failed, func(i, j int) bool { return failed[i].module < failed[j].module })
var names, words []string
tool := ""
@@ -761,19 +609,14 @@ func (w *Watchman) applyCheck(node string) {
}
f := Finding{Scope: ScopeMachine, ID: node, Kind: KindApplyFailing, Machine: node, Severity: Warning, Source: "S21", Since: since,
Said: strings.Join(words, "; "), Headline: node + " is not applying its update"}
switch {
case len(applied) == 0:
f.Summary = fmt.Sprintf("%s's provisioner has failed to place the files of %s since %s (bound %s)", node,
strings.Join(names, ", "), stamp(since), ApplyEvery)
f.Explanation = fmt.Sprintf("The provisioner on %s has not been able to write the files of %s. The details say why.",
node, strings.Join(names, ", "))
case tool != "":
f.Summary = fmt.Sprintf("%s: %s does not answer, so no package reaches it; everything else it was sent is applied, "+
f.Summary = fmt.Sprintf("%s's provisioner has failed to place the files of %s since %s (bound %s)", node,
strings.Join(names, ", "), stamp(since), ApplyEvery)
f.Explanation = fmt.Sprintf("The provisioner on %s has not been able to write the files of %s. The details say why.",
node, strings.Join(names, ", "))
if tool != "" {
f.Summary = fmt.Sprintf("%s: %s does not answer, so no package reaches it; everything else it was sent is placed, "+
"since %s (bound %s)", node, tool, stamp(since), ApplyEvery)
f.Headline = node + " cannot install packages"
default:
f.Summary = fmt.Sprintf("%s has failed to apply %s since %s (bound %s)", node, strings.Join(names, ", "), stamp(since), ApplyEvery)
f.Explanation = fmt.Sprintf("%s has failed to apply its updates, and trying again has not helped. The details say what fails.", node)
}
w.find(f)
}
@@ -931,13 +774,12 @@ func orWords(s, otherwise string) string {
// SignalRows are the rows the watchman holds, in the table's order, for `signals`.
var SignalRows = []struct{ Row, Signal, Bound, Kind string }{
{"S1", "a machine's presence (not while it said it sleeps or shuts down); a control node urgent after " + ControlNodeUrgentAfter.String(), SilentBound.String(), KindSilent},
{"S1", "a machine's presence (not while it said it sleeps or shuts down)", SilentBound.String(), KindSilent},
{"S8", "a provider's failing word repeated", ProviderSaysAgainWithin.String(), KindProviderSilent},
{"S9", "the bus's advisories", "any occurrence; clears after " + AdvisoryQuiet.String() + " without another", "slow-consumer, connection-closed, max-deliveries, refused, consumer-lost"},
{"S10", "the self-check's checked", (3 * SelfCheckEvery).String(), KindSelfCheckSilent},
{"S11", "a machine's tool runner heartbeat", "3 × its interval (" + (3 * HeartbeatEvery).String() + " when it says none)", KindToolsSilent},
{"S21", "a machine's have failed or its provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing},
{"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy},
{"S21", "a machine's provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing},
{"take-in", "a source revision or a delivery group the indexer holds back (ADR 0633 §5)", TakeInRevisionBound.String() + " a revision, " + TakeInGroupBound.String() + " a group", KindTakeInWaits},
{actingRow, "an acting part's wait held, or a put's, a put without hold that stands (the seats' wait-held, ended by wait-ended)", "the wait's own bound", KindActingPartLate + ", " + KindPutStands},
{storeRow, "a store of the delivery's parts filling or full (the seats' filling, full and room)", "filling from " + fmt.Sprintf("%.0f", FillingFrom*100) + " %, full urgent", KindStoreFilling},
modules/mesh-delivery-watchman/internal/watch/watch.go
+10 -22
View File
@@ -8,7 +8,6 @@ import (
"sync"
"time"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-sdk/go/keys"
)
@@ -64,10 +63,8 @@ type Watchman struct {
presence map[string]presence
pulses map[string]*pulse // live pulses, by pulse key
stuck map[string]meshwatchdogsclient.Stuck
// keptSince is, by have key, when its have first said what the node keeps of the module as it runs (keptFinding).
keptSince map[string]time.Time
wants map[string]string // the fingerprint of each want that stands, by want key
haves map[string]string // the fingerprint each node runs of a module, by have key
wants map[string]string // the fingerprint of each want that stands, by want key
haves map[string]string // the fingerprint each node runs of a module, by have key, from the node-engine's running
// starts is, by pulse key, the first start of a wait whose mark still stands (mirrorKept).
starts map[string]startMark
held map[string]bool // each want key held by a rule (ADR 0603 §8), mirrored as the hold's wait
@@ -85,24 +82,15 @@ type Watchman struct {
takeInCaught bool
// away is each machine that said it sleeps or shuts down, until it says it is back.
away map[string]bool
// haveState is each module's have as last read, unhealthy its writes in a row that said it failed, advised each
// advisory's count while its finding stands and advisedSince since when.
haveState map[string]nodeengine.HaveByNodeModuleValue
unhealthy map[string]int
// advised is each advisory's count while its finding stands and advisedSince since when.
advised map[string]int
advisedSince map[string]time.Time
// control is each node running the controller; failedSince each have's first failure still standing; provFailed each
// machine's provisioner client's failed reports; closures each bus closure counted, by finding key and identity.
control map[string]bool
failedSince map[string]time.Time
provFailed map[string]map[string]ProvReport
closures map[string]map[string]time.Time
recount func()
// provFailed is each machine's provisioner client's failed reports; closures each bus closure counted, by finding key and identity.
provFailed map[string]map[string]ProvReport
closures map[string]map[string]time.Time
recount func()
// waiting is, by `<consumer>.<node>`, each of its provisions in waits, with its words.
waiting map[string]map[string]string
// health is each node's own health report, and tunnels each node's tunnel key from its record.
health map[string]nodeengine.HealthByNodeValue
tunnels map[string]string
// stepBound is how long a build step may run once a node took the ask; a test shortens it.
stepBound time.Duration
// progressBound is how long a running ask may say nothing of its progress; a test shortens it.
@@ -137,9 +125,9 @@ func New(out Out, now func() time.Time) *Watchman {
return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{},
pulses: map[string]*pulse{}, stuck: map[string]meshwatchdogsclient.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: meshwatchdogsclient.PulseMoveBound, stepBound: StepBound, progressBound: ProgressBound,
findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, busCreated: map[string]bool{}, takeInHeld: map[string]Finding{}, pendingBranch: map[string]string{}, takenEntries: map[string][]string{}, recordMilestone: map[string]string{}, away: map[string]bool{},
haveState: map[string]nodeengine.HaveByNodeModuleValue{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{},
control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{},
health: map[string]nodeengine.HealthByNodeValue{}, tunnels: map[string]string{},
advised: map[string]int{}, advisedSince: map[string]time.Time{},
provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{},
standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}, heads: map[string]head{}, queued: map[string]map[string]time.Time{}, awaySaid: map[string]time.Time{}, opened: map[string]Finding{}}
}
modules/mesh-delivery-watchman/internal/watch/watch_test.go
+5 -13
View File
@@ -8,8 +8,6 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
@@ -614,14 +612,8 @@ func wantValue(t *testing.T, fp string, removed bool) []byte {
return b
}
func haveValue(t *testing.T, fp string) []byte {
t.Helper()
b, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
if err != nil {
t.Fatal(err)
}
return b
}
// runningBody is the node-engine's `running` for a build of fingerprint fp (design 47 G §541).
func runningBody(fp string) []byte { return []byte(`{"fingerprint":"` + fp + `"}`) }
var (
fpOld = meshcatalogclient.SourcePrefix2 + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
@@ -637,11 +629,11 @@ func TestAWantIsAPulseOwedByTheNodeAndTheHaveNamingItEndsIt(t *testing.T) {
if !ok || p.Key != "want.ace.web" || !strings.Contains(p.Owes, "ace") || !strings.Contains(p.Owes, "web") {
t.Fatalf("pulses = %v", o.pulses)
}
w.Have("ace.web", haveValue(t, fpOld), Put) // the node still runs the old one
w.EngineRunning("ace", "web", runningBody(fpOld)) // the node still runs the old one
if _, ok := o.pulses[pk]; !ok {
t.Fatalf("a have of another fingerprint ended the pulse")
}
w.Have("ace.web", haveValue(t, fpNew), Put)
w.EngineRunning("ace", "web", runningBody(fpNew))
if _, ok := o.pulses[pk]; ok {
t.Errorf("the have naming the want's fingerprint left the pulse standing: %v", o.pulses)
}
@@ -653,7 +645,7 @@ func TestAWantIsAPulseOwedByTheNodeAndTheHaveNamingItEndsIt(t *testing.T) {
func TestAHaveThatCameBeforeTheWantEndsItAtOnceAndARemovedWantHasNoPulse(t *testing.T) {
w, o, c := setup()
c.t = t0.Add(time.Second)
w.Have("ace.web", haveValue(t, fpNew), Put)
w.EngineRunning("ace", "web", runningBody(fpNew))
w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
w.Want("want.ace.mail", wantValue(t, "", true), Put, t0)
if len(o.pulses) != 0 {
modules/mesh-delivery-watchman/manifest_test.go
+2 -4
View File
@@ -103,7 +103,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
reads := slices.Clone(m.Reads)
slices.Sort(reads)
// The build seats' steps are no read (design 47 G §483, A-136): the watchman times an ask from the seats' step-progress.
if !reflect.DeepEqual(reads, []string{"mesh-host.health", "mesh-provisioner.provisions", "node-engine.have"}) {
if !reflect.DeepEqual(reads, []string{"mesh-provisioner.provisions"}) {
t.Errorf("reads = %v", reads)
}
// Design 47 G §368: the version is MAJOR.MINOR and nothing else.
@@ -130,7 +130,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
// DLV-M5 its first views, 1585/T20 the indexer's take-in events, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably.
if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) {
if want := []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.take-in.pending", "mesh-delivery-indexer.take-in.taken", "mesh-catalog.manifest-accepted", "mesh-catalog.manifest-refused", "mesh-delivery.source-moved", "mesh-delivery.filling", "mesh-delivery.full", "mesh-delivery.room", "mesh-delivery.record-opened", "mesh-delivery.record-ended", "mesh-catalog.filling", "mesh-catalog.full", "mesh-catalog.room", "mesh-wants.filling", "mesh-wants.full", "mesh-wants.room", "mesh-grants.filling", "mesh-grants.full", "mesh-grants.room", "node-build-go.ask-taken", "node-build-go.step-progress", "node-build-go.step-ended", "node-build-image.ask-taken", "node-build-image.step-progress", "node-build-image.step-ended", "node-build-node.ask-taken", "node-build-node.step-progress", "node-build-node.step-ended", "mesh-pipeline-runner.pipeline-started", "mesh-pipeline-runner.step-started", "mesh-pipeline-runner.step-ended", "mesh-pipeline-runner.wait-held", "mesh-pipeline-runner.wait-ended", "mesh-wants.wait-held", "mesh-wants.wait-ended", "mesh-grants.wait-held", "mesh-grants.wait-ended", "mesh-delivery-builder.wait-held", "mesh-delivery-builder.wait-ended", "node-engine.wait-held", "node-engine.wait-ended", "node-engine.running", "mesh-broker.created", "mesh-broker.refused", "mesh-conditions.findings-asked"}; !reflect.DeepEqual(m.Consumes, want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
}
// 1952/WM1: the watchman holds the seat mesh-watchdogs at mesh scope, its first seat (novox/hq ADR 0564 §1, §4).
@@ -172,8 +172,6 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
"presence": {watch.PresenceBucket, m.Module + "_" + m.State[0].Name},
"pulse": {watch.PulseBucket, m.Module + "_" + m.State[1].Name},
"stuck": {watch.StuckBucket, m.Module + "_" + m.State[2].Name},
"have": {watch.HaveBucket, bucket(m.Reads, "node-engine.have")},
"health": {watch.NodeHealthBucket, bucket(m.Reads, "mesh-host.health")},
"wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")},
"provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")},
"assigner": {watch.AssignerModule + ".assigner_read_wants", entry(m.Invokes, "mesh-delivery-assigner.assigner_read_wants")},
modules/mesh-delivery-watchman/module.json
+2 -3
View File
@@ -16,9 +16,7 @@
}
],
"reads": [
"mesh-host.health",
"mesh-provisioner.provisions",
"node-engine.have"
"mesh-provisioner.provisions"
],
"invokes": [
"mesh-delivery-assigner.assigner_read_wants",
@@ -70,6 +68,7 @@
"mesh-delivery-builder.wait-ended",
"node-engine.wait-held",
"node-engine.wait-ended",
"node-engine.running",
"mesh-broker.created",
"mesh-broker.refused",
"mesh-conditions.findings-asked"