Merge branch 'feat/47-G-fix11-b6b' into feat/47-G-join11

This commit is contained in:
2026-11-12 10:28:34 +01:00
27 changed files with 699 additions and 1230 deletions
@@ -18,8 +18,10 @@ import (
// waitBody is the body of wait-held and wait-ended on the builder's stream waits (design 47 G §672): the fields of §629's
// wait with `step` building, `on` toolchain:<language>@<version>, `module` and `fingerprint`, and, on wait-ended, `how`.
type waitBody struct {
Step string `json:"step"`
On string `json:"on"`
Step string `json:"step"`
On string `json:"on"`
// Part is the package a key queued behind it waits for (design 47 G A-192); none for a toolchain wait.
Part string `json:"part,omitempty"`
Module string `json:"module"`
Fingerprint string `json:"fingerprint"`
Since time.Time `json:"since"`
@@ -46,7 +48,7 @@ func newWaitsKept(kv *process.AwaitedBucket, say func(context.Context, string, s
}
func (w *waitsKept) body(t build.ToolchainWait, how string) waitBody {
return waitBody{Step: waitStepBuilding, On: t.On(), Module: t.Module, Fingerprint: t.Fingerprint, Since: t.Since, Bound: w.bound.String(), How: how}
return waitBody{Step: waitStepBuilding, On: t.On(), Part: t.Package, Module: t.Module, Fingerprint: t.Fingerprint, Since: t.Since, Bound: w.bound.String(), How: how}
}
// start reads the state whole and says wait-held again for each key.
@@ -110,7 +112,13 @@ func (w *waitsKept) End(ctx context.Context, module, fp, how string) bool {
}
w.mu.Lock()
t, had := w.held[key]
delete(w.held, key)
// A key queued behind a package ends only when it is published, and a toolchain wait never by that (design 47 G A-192).
if had && (t.Package != "") != (how == build.WaitPublished) {
had = false
}
if had {
delete(w.held, key)
}
w.mu.Unlock()
if !had {
return false
@@ -0,0 +1,21 @@
package main
import (
"testing"
"time"
"mesh-delivery-builder/internal/build"
)
// Design 47 G A-192: a key queued behind a package says its wait in the one wait form, the package as its `part` and `on`; a toolchain
// wait names none.
func TestAWaitBodyNamesItsPart(t *testing.T) {
w := &waitsKept{bound: time.Hour, held: map[string]build.ToolchainWait{}}
pkg := w.body(build.ToolchainWait{Module: "m", Fingerprint: "src2:f", Package: "@novox/sdk@0.1.15"}, "")
if pkg.Part != "@novox/sdk@0.1.15" || pkg.On != "package:@novox/sdk@0.1.15" || pkg.Step != waitStepBuilding {
t.Errorf("a package wait says %+v", pkg)
}
if tc := w.body(build.ToolchainWait{Module: "m", Fingerprint: "src2:f", Language: "go", Version: "go1.26.0"}, ""); tc.Part != "" || tc.On != "toolchain:go@go1.26.0" {
t.Errorf("a toolchain wait says %+v", tc)
}
}
@@ -173,6 +173,8 @@ func sdkModule(t *testing.T, r *rig, ref string) []byte {
func TestAModuleWhoseLockNamesAnUnpublishedPackageWaitsQueuedAndIsBuiltOncePublished(t *testing.T) { // B7, ADR 0500 clause 6
const ref = "@novox/mesh-sdk@0.1.15"
r := newRig(t, tsBundle)
w := &memWaits{}
r.b.Waits = w
for _, n := range []string{"ace", "novox"} {
r.bm.fails[n+"/build_node_fetch"] = waitsFor + ref + ": the lock names a version the registry does not hold"
}
@@ -180,6 +182,15 @@ func TestAModuleWhoseLockNamesAnUnpublishedPackageWaitsQueuedAndIsBuiltOncePubli
if v.State != builds.BuildQueued || v.Waits != ref || v.Error != "" {
t.Fatalf("build = %+v, want queued, waiting for %s, not failed", v, ref)
}
// Design 47 G A-192: the queued key is a wait said in the one wait form, with its package as the part, and it stands while the package is unpublished.
if len(w.held) != 1 {
t.Fatalf("waits held %v, want the one of the package", w.held)
}
for _, h := range w.held {
if h.Package != ref || h.On() != "package:"+ref {
t.Errorf("the wait is %+v on %s, want the package %s", h, h.On(), ref)
}
}
delete(r.bm.fails, "ace/build_node_fetch")
delete(r.bm.fails, "novox/build_node_fetch")
writes := r.st.writes
@@ -188,11 +199,17 @@ func TestAModuleWhoseLockNamesAnUnpublishedPackageWaitsQueuedAndIsBuiltOncePubli
if r.st.writes != writes {
t.Fatalf("a key waiting for an unpublished package was claimed (%d writes, was %d)", r.st.writes, writes)
}
if len(w.held) != 1 {
t.Fatalf("the wait ended while the package is unpublished: %v", w.held)
}
r.b.Module(context.Background(), "sdk", sdkModule(t, r, ref)) // the package's trunk key is built and published
r.b.Wait()
if v = valueOf(t, r.st, r.key); v.State != builds.BuildDone || v.Waits != "" {
t.Errorf("build = %+v, want built once the package it waits for is published", v)
}
if len(w.held) != 0 || len(w.ended) != 1 || w.ended[0] != WaitPublished {
t.Errorf("held %v, ended %v: want the wait ended published", w.held, w.ended)
}
}
func TestAKeyWaitingForAPackageAlreadyPublishedIsBuiltWhenTheBuilderStarts(t *testing.T) { // B7: the scan at start
@@ -617,6 +617,9 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
return
}
b.handClaimed(key, cl.Rev)
if b.Waits != nil {
b.Waits.End(ctx, module, m.Fp2, WaitPublished) // a key queued behind a package is claimed once it is published
}
b.logBuild(LogStarted(module, m, "", b.Node, b.now()))
if cl.Unasked != nil {
b.logf("%v", cl.Unasked)
@@ -717,6 +720,10 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
}
if result.State == builds.BuildQueued {
b.logf("%s: waits for the package %s to be published", cl.Key, result.Waits)
if b.Waits != nil {
// Said as a wait in the one wait form, with its part, for the watchman to time (design 47 G A-192; ADR 0603 §8).
b.Waits.Hold(ctx, ToolchainWait{Module: module, Fingerprint: m.Fp2, Package: result.Waits, Since: b.now()})
}
}
if b.notePublished(result) {
// A key may wait for the package this build published (ADR 0500 clause 6).
@@ -26,8 +26,10 @@ const (
// ModeRelease is the Mode of a reasks value of builder_release.
ModeRelease = "release"
// WaitRan ends a wait when a toolchain-agreed names the version; WaitReleased when builder_release ended it.
WaitRan = "ran"
WaitReleased = "released"
WaitRan = "ran"
// WaitPublished ends the wait of a key queued behind a package, when the key is claimed (the package is published).
WaitPublished = "published"
WaitReleased = "released"
)
// ToolchainWait is one wait: the entry's module and fingerprint, and the toolchain it waits for.
@@ -37,10 +39,18 @@ type ToolchainWait struct {
Language string `json:"language"`
Version string `json:"version"`
Since time.Time `json:"since"`
// Package is the package version a key queued behind it waits for, `<name>@<version>` (ADR 0603 §8; design 47 G A-192): the wait is the
// package's, not a toolchain's, and Language and Version are empty.
Package string `json:"package,omitempty"`
}
// On is what the wait waits for, as the wait's `on` says it.
func (w ToolchainWait) On() string { return "toolchain:" + w.Language + "@" + w.Version }
func (w ToolchainWait) On() string {
if w.Package != "" {
return "package:" + w.Package
}
return "toolchain:" + w.Language + "@" + w.Version
}
// Waits keeps the builder's waits: Hold writes one in the state waits and says wait-held, once per wait; End deletes it and
// says wait-ended with how, and answers whether one was held.
@@ -28,7 +28,8 @@ func (w *memWaits) End(_ context.Context, module, fp, how string) bool {
w.mu.Lock()
defer w.mu.Unlock()
k := module + "." + fp
if _, ok := w.held[k]; !ok {
t, ok := w.held[k]
if !ok || (t.Package != "") != (how == WaitPublished) { // as the builder's waitsKept ends them: a package wait by its publishing alone
return false
}
delete(w.held, k)
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
@@ -45,7 +46,16 @@ func consumed(w *watch.Watchman, subject string, body []byte, logf func(string,
return
}
w.Accepted(a)
case subject == meshcatalog.SubjectManifestRefused:
var r meshcatalog.ManifestRefusedEvent
if err := json.Unmarshal(body, &r); err != nil || r.Module == "" {
logf("a manifest-refused that does not read is left: %v", err)
return
}
w.ManifestRefused(r)
case pipelineSubject(w, subject, body):
case engineWaitSubject(w, subject, body):
case moduleWaitSubject(w, subject, body):
case recordSubject(w, subject, body):
case buildSeatSubject(w, subject, body):
case brokerSubject(subject) && strings.HasSuffix(subject, ".event."+eventCreated):
@@ -190,6 +200,45 @@ func recordSubject(w *watch.Watchman, subject string, body []byte) bool {
return true
}
// 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, ".")
if len(parts) != 7 || parts[0] != "mesh" || parts[1] != "seat" || parts[2] != nodeengine.Seat || parts[3] != "event" {
return false
}
switch parts[4] {
case nodeengine.EventWaitHeld:
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
}
return true
}
// moduleWaitSubject hands a module's `wait-held` and `wait-ended` on its own stream, on
// mesh.stream.<module>.event.<event>.<module>.<fingerprint>, to the watchman: the form the builder says a key queued behind a package
// in (design 47 G A-192). It says whether the subject was one.
func moduleWaitSubject(w *watch.Watchman, subject string, body []byte) bool {
parts := strings.Split(subject, ".")
if len(parts) != 7 || parts[0] != "mesh" || parts[1] != "stream" || parts[3] != "event" {
return false
}
switch parts[4] {
case nodeengine.EventWaitHeld:
w.ModuleWaitHeld(parts[2], body)
case nodeengine.EventWaitEnded:
w.ModuleWaitEnded(parts[2], body)
default:
return false
}
return true
}
// buildSeatSubject hands the build seats' `ask-taken`, `step-progress` and `step-ended`, on
// mesh.seat.<seat>.event.<event>.<node>, to the watchman (design 47 G §483, §496, §602, A-136): it times every ask of a
// build seat from them. It says whether the subject was one of them.
@@ -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"
@@ -63,12 +62,12 @@ accounts {
}
`
// serveOwners serves the assigner's wants and the indexer's modules as their owners do, by their read tools and views
// serveOwners serves the assigner's wants as its owner does, by their read tools and views
// families (DLV-M5, ADR 0615): the watchman (n1) follows both, never their buckets.
func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) {
t.Helper()
viewstest.Serve(t, ctx, js, nc, watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants", "n1_"+watch.WatchmanModule)
viewstest.Serve(t, ctx, js, nc, watch.IndexerModule, modulesState, watch.IndexerModule+".indexer_read_modules", "n1_"+watch.WatchmanModule)
}
// makeEvents makes EVENTS and the watchman's durable consumer of the build seats' events, as the controller and the grants part
@@ -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)
}
}
@@ -194,8 +194,7 @@ func createdOr(e jetstream.KeyValueEntry, at time.Time) time.Time {
var firstViewsBucket = watch.WatchmanModule + "_" + views.StatusState
const (
wantsState = "wants"
modulesState = "modules"
wantsState = "wants"
// The owners' views families, as their manifests' `streams` name them.
)
@@ -222,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
@@ -231,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
@@ -261,16 +259,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err != nil {
return err
}
modules, err := reader.Follow(watch.IndexerModule, modulesState, watch.IndexerModule+".indexer_read_modules")
if err != nil {
return err
}
w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: held, js: js, cred: cred}, time.Now)
w.Logf(logger.Printf)
// What the wants and the modules hold is ended by their own late read (watch.ReadLate), never by the start's end.
w.Awaits(watch.AssignerWantsBucket)
w.Awaits(watch.ModulesBucket)
// A test shortens how long a move may take; the mesh never sets it.
if v := strings.TrimSpace(getenv("MESH_DELIVERY_PULSE_BOUND")); v != "" {
d, err := time.ParseDuration(v)
@@ -314,26 +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 builder's builds (ADR 0603 §8): a key queued behind a package is a wait, mirrored from the time it was written.
var buildSeen jetstream.KeyWatcher
if kv := process.OptionalBucket(ctx, js, watch.BuilderBuildsBucket, "a build waiting for a package is not watched until the watchman starts again", logger.Printf); kv != nil {
buildEach := func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), opOf(e), e.Created()) }
if buildSeen, err = settle(ctx, kv, buildEach); err != nil {
process.NotWatched(logger.Printf, watch.BuilderBuildsBucket, err, "a build waiting for a package is not watched until the watchman starts again")
buildSeen = nil
} else {
w.ReadWhole(watch.BuilderBuildsBucket)
}
}
// 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) {
@@ -394,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()
@@ -427,6 +381,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
logger.Printf("take-in: no connection to ask %s (%v), so a wait the indexer held before this start is said only once a take-in event of it is heard", pendingTool, err)
} else {
seedTakeIn(w, calls.Call, logger.Printf)
seedVerdicts(w, calls.Call, logger.Printf)
calls.Close()
}
// The take-in events are the indexer's ordinary events (its emits), heard on the shared events stream through this
@@ -455,27 +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()
// The indexer's modules (ADR 0578 §1): a module whose value carries a refusal of its manifest is a finding.
verdictEach := func(e jetstream.KeyValueEntry) { w.ModuleVerdict(e.Key(), e.Value(), opOf(e)) }
go func() { errs <- readLate(ctx, w, modules, watch.ModulesBucket, verdictEach) }()
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)) })
@@ -491,11 +435,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go func() {
errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) })
}()
if buildSeen != nil {
go func() {
errs <- follow(ctx, buildSeen, func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), opOf(e), e.Created()) })
}()
}
go func() {
for {
select {
@@ -538,6 +477,20 @@ func seedTakeIn(w *watch.Watchman, call func(key string, body any) (json.RawMess
logf("take-in: %s cannot be asked (%v), so a wait the indexer held before this start is said only once a take-in event of it is heard", pendingTool, err)
}
// seedVerdicts asks the indexer's verdicts and raises the findings they hold (design 47 G §500): a verdict said while the watchman was
// down is then not lost. An indexer that does not answer is said: the findings are then raised from the verdicts heard after.
func seedVerdicts(w *watch.Watchman, call func(key string, body any) (json.RawMessage, error), logf func(string, ...any)) {
raw, err := call(watch.VerdictsTool, map[string]any{"all": true})
if err == nil {
var rows []watch.VerdictRow
if rows, err = watch.DecodeVerdictRows(raw); err == nil {
w.SeedVerdicts(rows)
return
}
}
logf("verdicts: %s cannot be asked (%v), so a manifest refused before this start is said only once a verdict of it is heard", watch.VerdictsTool, err)
}
// takeInHeard hands one take-in event of the indexer to the watchman, by its name relative to the indexer, and acknowledges it;
// one that is not a take-in event is said and acknowledged, since hearing it again would not read it.
func takeInHeard(w *watch.Watchman, m jetstream.Msg, logf func(string, ...any)) {
@@ -75,6 +75,41 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
t.Fatalf("created left %v", out.findings)
}
})
t.Run("design 47 G §500: the indexer's manifest-refused reaches the watchman by its subject", func(t *testing.T) {
out := &findingsOut{findings: map[string]bool{}}
w := watch.New(out, time.Now)
consumed(w, "mesh.seat.mesh-catalog.event.manifest-refused", []byte(`{"module":"m","digest":"d","problems":["x"]}`), func(string, ...any) {})
if !out.findings[watch.FindingKey(watch.ScopeModule, "m", watch.KindManifestRefused)] {
t.Fatalf("a refused manifest reached %v", out.findings)
}
})
t.Run("design 47 G §499, §541: a node-engine's wait-held and wait-ended reach the watchman by their subjects", func(t *testing.T) {
out := &findingsOut{findings: map[string]bool{}}
w := watch.New(out, time.Now)
say := func(string, ...any) {}
body := fmt.Sprintf(`{"step":"compose","reason":"entry-missing","since":%q}`, time.Now().Add(-2*time.Hour).UTC().Format(time.RFC3339))
consumed(w, "mesh.seat.node-engine.event.wait-held.ace.web", []byte(body), say)
if !out.findings[watch.FindingKey(watch.ScopeDelivery, "ace.web", watch.KindDeliveryWaits)] {
t.Fatalf("a hold two hours old reached %v", out.findings)
}
consumed(w, "mesh.seat.node-engine.event.wait-ended.ace.web", []byte(body), say)
if len(out.findings) != 0 {
t.Fatalf("wait-ended left %v", out.findings)
}
})
t.Run("design 47 G A-192: a module's wait-held on its stream, naming a part, reaches the watchman", func(t *testing.T) {
out := &findingsOut{findings: map[string]bool{}}
w := watch.New(out, time.Now)
body := fmt.Sprintf(`{"step":"building","part":"mesh-sdk@v0.1.14","module":"web","fingerprint":"src2:f","bound":"1h0m0s","since":%q}`, time.Now().Add(-2*time.Hour).UTC().Format(time.RFC3339))
consumed(w, "mesh.stream.mesh-delivery-builder.event.wait-held.web.src2:f", []byte(body), func(string, ...any) {})
if len(out.findings) != 1 {
t.Fatalf("a package wait two hours old reached %v", out.findings)
}
consumed(w, "mesh.stream.mesh-delivery-builder.event.wait-ended.web.src2:f", []byte(body), func(string, ...any) {})
if len(out.findings) != 0 {
t.Fatalf("wait-ended left %v", out.findings)
}
})
err := run(context.Background(), func(string) string { return "" }, io.Discard)
if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err)
@@ -64,6 +64,7 @@ func refusedKey(module, local string) string {
func (w *Watchman) Accepted(a meshcatalog.ManifestAcceptedEvent) {
w.mu.Lock()
defer w.mu.Unlock()
w.manifestAccepted(a.Module, a.Problems) // §500: the problems a stricter validator finds are the module's warning
asked := map[string]bool{}
for _, e := range a.BusState {
if e.Name == "" {
@@ -3,16 +3,12 @@ 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-builder/builds"
"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"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
@@ -31,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)
@@ -188,43 +175,32 @@ func TestTheWatchmansFindings(t *testing.T) {
next := "m.00000000000000000043"
w.TakeIn(takeInEvent(t, takein.TakeInPending, next), takeInBody(t, takein.TakeInPending, next, t0))
}, map[string]string{FindingKey(ScopeDelivery, "m.00000000000000000043", KindTakeInWaits): Warning}},
{"ADR 0578 §1: a module whose value carries a refusal of its manifest is a finding until it is accepted", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
value := func(verdict string, problems ...string) []byte {
b, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: "src2:" + strings.Repeat("a", 64), Manifest: strings.Repeat("b", 64),
Repository: "novox/mesh-catalog", Commit: "c", Verdict: verdict, Problems: problems})
if err != nil {
t.Fatal(err)
}
return b
}
w.ModuleVerdict("nats", value(meshcatalogclient.VerdictRefused, "removes a stream"), Put)
w.ModuleVerdict("power", value(meshcatalogclient.VerdictRefused, "shows nothing"), Put)
{"design 47 G §500: manifest-refused raises the refusal, manifest-accepted clears it and carries an outgrown manifest's problems", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.ManifestRefused(meshcatalog.ManifestRefusedEvent{Module: "nats", Problems: []string{"removes a stream"}})
w.ManifestRefused(meshcatalog.ManifestRefusedEvent{Module: "power", Problems: []string{"shows nothing"}})
k := FindingKey(ScopeModule, "nats", KindManifestRefused)
if f := o.findings[k]; !strings.Contains(f.Summary, "refused the manifest of nats: removes a stream") {
t.Fatalf("findings = %v", o.findings)
}
w.ModuleVerdict("nats", value(""), Put)
w.ModuleVerdict("power", nil, Delete)
w.Accepted(meshcatalog.ManifestAcceptedEvent{Module: "nats", Digest: strings.Repeat("d", 64)})
w.Accepted(meshcatalog.ManifestAcceptedEvent{Module: "power", Digest: strings.Repeat("d", 64)})
// ADR 0579 §1: an accepted manifest a stricter judge outgrew is a warning until mended.
outgrown, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: "src2:" + strings.Repeat("a", 64), Manifest: strings.Repeat("b", 64),
Repository: "novox/mesh-tools", Commit: "c", Warnings: []string{"says no credential form"}})
w.ModuleVerdict("node-tools", outgrown, Put)
w.Accepted(meshcatalog.ManifestAcceptedEvent{Module: "node-tools", Digest: strings.Repeat("d", 64), Problems: []string{"says no credential form"}})
if f := o.findings[FindingKey(ScopeModule, "node-tools", KindManifestOutgrown)]; f.Severity != Warning || !strings.Contains(f.Summary, "says no credential form") {
t.Fatalf("findings = %v", o.findings)
}
w.ModuleVerdict("node-tools", value(""), Put)
// Clause 3 of the record on a release's Approve: a manifest that waits for a judge that knows its fields is its own
// warning (a refusal in the value, naming the fields), not manifest-refused, and goes when it is accepted.
waiting, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: "src2:" + strings.Repeat("a", 64), Manifest: strings.Repeat("b", 64),
Repository: "novox/mesh-catalog", Commit: "c", Verdict: meshcatalogclient.VerdictRefused, Problems: []string{"waits"}, Unknown: []string{"core-credentials"}})
w.ModuleVerdict("nats", waiting, Put)
if f := o.findings[FindingKey(ScopeModule, "nats", KindManifestWaits)]; f.Severity != Warning || !strings.Contains(f.Summary, "core-credentials") {
w.Accepted(meshcatalog.ManifestAcceptedEvent{Module: "node-tools", Digest: strings.Repeat("e", 64), Problems: []string{}})
// The start seeds from the indexer's verdicts: the newest of each module counts.
w.SeedVerdicts([]VerdictRow{
{Module: "nats", State: meshcatalogclient.VerdictRefused, Problems: []string{"old"}, At: t0},
{Module: "nats", State: meshcatalogclient.VerdictAccepted, At: t0.Add(time.Hour)},
{Module: "power", State: meshcatalogclient.VerdictRefused, Problems: []string{"shows nothing"}, At: t0},
{Module: "node-tools", State: meshcatalogclient.VerdictAccepted, Warnings: []string{"w"}, At: t0}})
if len(o.findings) != 2 || o.findings[FindingKey(ScopeModule, "power", KindManifestRefused)].Kind == "" || o.findings[FindingKey(ScopeModule, "node-tools", KindManifestOutgrown)].Kind == "" {
t.Fatalf("findings = %v", o.findings)
}
if _, refused := o.findings[FindingKey(ScopeModule, "nats", KindManifestRefused)]; refused {
t.Fatalf("a waiting manifest was found refused: %v", o.findings)
}
w.ModuleVerdict("nats", value(""), Put)
w.Accepted(meshcatalog.ManifestAcceptedEvent{Module: "power"})
w.Accepted(meshcatalog.ManifestAcceptedEvent{Module: "node-tools"})
}, map[string]string{}},
{"DLV-W1, 1966/AC37-AC38 (ADR 0595 §4, design 25 § 12.2): an accepted entry the bus module says nothing of in five minutes is overdue until created; a refused one stands until created; a later verdict no longer asking ends both", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
accepted := func(locals ...string) meshcatalog.ManifestAcceptedEvent {
@@ -335,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)
@@ -459,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)})
@@ -480,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"},
@@ -540,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)
@@ -709,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"},
@@ -737,7 +493,7 @@ func TestTheWatchmansFindings(t *testing.T) {
w.DeadLettersHeld(map[string]int{"SEAT_X.y_worker": 1}, who)
w.NoticesWaiting(true)
}, map[string]string{"bus.SEAT_X.y_worker.max-deliveries": Warning, "bus.DEAD_LETTER_NOTICES.controller.notices-waiting": Warning}},
{"ADR 0603 §8: a want held by a rule, a provision asked and a build queued behind a package each raise a finding naming the wait past their bound, and clear when it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
{"ADR 0603 §8: a want held by a rule and a provision asked each raise a finding naming the wait past their bound, and clear when it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.SetStepBound(30 * time.Minute)
held, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fpOld, Self: true, Newest: fpNew, Rule: meshwantsclient.RuleCanary, Words: "the canary ace does not run it yet"})
released, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fpNew, Self: true})
@@ -745,15 +501,11 @@ 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})
bkey, _ := builds.BuilderBuildKey("web", fpNew)
queued, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildQueued, Waits: "mesh-sdk@v0.1.14"})
building, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildBuilding, By: "n1"})
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)
w.Build(bkey, queued, Put, t0)
waits := map[string]string{AssignerWantsBucket: "want.ace.web", ProvisionsBucket: "ace.photos.oidc", BuilderBuildsBucket: bkey}
if len(o.pulses) != 3 || len(o.findings) != 0 {
waits := map[string]string{AssignerWantsBucket: "want.ace.web", ProvisionsBucket: "ace.photos.oidc"}
if len(o.pulses) != 2 || len(o.findings) != 0 {
t.Fatalf("pulses %v findings %v", o.pulses, o.findings)
}
// The keys are written again as their parts move on, with new words and a new time: the wait still began at t0.
@@ -763,7 +515,6 @@ func TestTheWatchmansFindings(t *testing.T) {
Provider: meshprovisionerclient.End{Module: "keycloak", Node: "novox"}, State: meshprovisionerclient.Asked, Why: "the provider is applying", At: c.t})
w.Want("want.ace.web", held2, Put, c.t)
w.Provision("ace.photos.oidc", asked2, Put)
w.Build(bkey, queued, Put, c.t)
for bucket, key := range waits {
expire(w, c, pulseKeyOf(t, bucket, key), 31*time.Minute)
}
@@ -775,7 +526,6 @@ func TestTheWatchmansFindings(t *testing.T) {
}
w.Want("want.ace.web", released, Put, c.t)
w.Provision("ace.photos.oidc", provided, Put)
w.Build(bkey, building, Put, c.t)
}, map[string]string{}},
}
for _, tc := range cases {
@@ -801,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"
@@ -845,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
@@ -8,7 +8,6 @@ import (
"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-builder/builds"
"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"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo"
@@ -165,84 +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)
}
// A build key queued behind a package is a wait of the builder's (ADR 0603 §8, ADR 0500 clause 6): its Waits names the package
// version the registry does not yet hold, and the key is not claimed until it is published. Its pulse is the pulse of the
// builder's builds bucket under the key, from the time the key was written, bounded as a step is (StepBound); the key
// claimed, done, failed or deleted ends it.
// Build is a change of a key of the builder's builds bucket, read from the watch (at the start the keys that stand, then
// each change), at the time the entry was written.
func (w *Watchman) Build(key string, value []byte, op Op, at time.Time) {
module, fp, err := builds.ParseBuilderBuildKey(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 op == Expired {
return
}
if op == Delete {
w.moved(BuilderBuildsBucket, key)
return
}
w.stands(BuilderBuildsBucket, key)
b, err := builds.DecodeBuilderBuild(value)
if err != nil {
w.say("the build %s cannot be read, and is left: %v", key, err)
return
}
if b.State != builds.BuildQueued || b.Waits == "" {
w.moved(BuilderBuildsBucket, key)
return
}
if at.IsZero() {
at = w.now()
}
w.mirrorKept(BuilderBuildsBucket, key,
describe("the package %s to be published before the build of %s at %s, bound %s", b.Waits, module, shortFP(fp), w.stepBound), b.Waits, at)
}
// 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)
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)
}
}
@@ -14,27 +14,18 @@ const (
ProvisionsBucket = "mesh-provisioner_provisions"
// IndexerModule is the holder of the seat mesh-catalog, named from this module's manifest (consumes and invokes of mesh-delivery-indexer).
IndexerModule = "mesh-delivery-indexer"
builderModule = "mesh-delivery-builder"
// The watchman's own states of the seat mesh-watchdogs.
PresenceBucket = WatchmanModule + "_" + meshwatchdogs.StatePresence
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"
// BuilderBuildsBucket is the builder's state builds (reads: mesh-delivery-builder.builds).
BuilderBuildsBucket = builderModule + "_builds"
// AssignerWantsBucket is the assigner's state wants, followed by its views family (consumes: mesh-delivery-assigner.views.wants).
AssignerWantsBucket = AssignerModule + "_wants"
// ModulesBucket is the indexer's state modules, followed by its views family (consumes: mesh-delivery-indexer.views.modules).
ModulesBucket = IndexerModule + "_modules"
)
// The findings a value of another part is raised as, named by their words.
const (
FindingManifestOutgrown = "manifest-outgrown"
FindingManifestWaits = "manifest-waits"
FindingNodeAccountFallback = "node-account-fallback"
)
@@ -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)
}
}
}
@@ -24,7 +24,10 @@ const HeadBound = 5 * time.Minute
const headRow = "head-pipeline"
// head is a watched branch's head still waiting for its pipeline.
type head struct{ revision int64 }
type head struct {
revision int64
at time.Time
}
// SourceMoved hears the seat mesh-delivery's `source-moved`: the branch's head is a new source revision, and its pipeline
// is due within HeadBound. A move made by a gathered group is a member governed by the take-in's own bound (ADR 0633 §5),
@@ -48,15 +51,26 @@ func (w *Watchman) SourceMoved(body []byte) {
if m.At != nil && !m.At.IsZero() {
at = *m.At
}
w.heads[m.Branch] = head{revision: m.Revision}
f := Finding{Scope: ScopeDelivery, ID: id, Kind: KindPipelineStands, Severity: Warning, Source: headRow, Since: at,
Summary: fmt.Sprintf("the head of %s (source revision %d) has started no pipeline since %s", m.Branch, m.Revision, stamp(at)),
w.heads[m.Branch] = head{revision: m.Revision, at: at}
w.timeHead(m.Branch)
}
// timeHead times the head of a branch that waits for its pipeline, w.mu held: five minutes from its own time, unless the
// branch has a pending revision, whose take-in wait governs it (design 47 G §497) and which times it again once taken in.
func (w *Watchman) timeHead(branch string) {
h, ok := w.heads[branch]
if !ok || w.branchPending(branch) {
return
}
id := headID(branch)
f := Finding{Scope: ScopeDelivery, ID: id, Kind: KindPipelineStands, Severity: Warning, Source: headRow, Since: h.at,
Summary: fmt.Sprintf("the head of %s (source revision %d) has started no pipeline since %s", branch, h.revision, stamp(h.at)),
Said: fmt.Sprintf("no pipeline in %s", HeadBound),
Headline: "A change has started no pipeline",
Explanation: "The head of the watched branch " + m.Branch + " (source revision " + fmt.Sprint(m.Revision) + ") moved " +
Explanation: "The head of the watched branch " + branch + " (source revision " + fmt.Sprint(h.revision) + ") moved " +
HeadBound.String() + " ago or more and the pipeline runner has said no pipeline started for it.",
Needs: "the runner to start the pipeline (its pipeline_stands and list tools say where it is)"}
left := HeadBound - w.now().Sub(at)
left := HeadBound - w.now().Sub(h.at)
if left > 0 {
w.signal(headRow, id, left, signalDue{Do: dueFind, Finding: f})
return
@@ -64,6 +78,16 @@ func (w *Watchman) SourceMoved(body []byte) {
w.find(f)
}
// branchPending is whether an open take-in wait names the branch (design 47 G §497): its head is behind a pending revision.
func (w *Watchman) branchPending(branch string) bool {
for _, b := range w.pendingBranch {
if b == branch {
return true
}
}
return false
}
// PipelineStarted hears the runner's `pipeline-started`: it ends the wait of the head it is for (the branch's head at that
// source revision or a later one). A run on request is no head's pipeline.
func (w *Watchman) PipelineStarted(body []byte) {
@@ -7,6 +7,8 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
)
func moved(branch string, rev int64, group string, at time.Time) []byte {
@@ -14,6 +16,21 @@ func moved(branch string, rev int64, group string, at time.Time) []byte {
return b
}
// takeInBodyOn is a take-in event's body that names its watched branch (design 47 G §493).
func takeInBodyOn(t *testing.T, kind, key, branch string, since time.Time) []byte {
t.Helper()
var m map[string]any
if err := json.Unmarshal(takeInBody(t, kind, key, since), &m); err != nil {
t.Fatal(err)
}
m["branch"] = branch
b, err := json.Marshal(m)
if err != nil {
t.Fatal(err)
}
return b
}
func started(branch string, rev int64, run string) []byte {
b, _ := json.Marshal(meshpipelinerunner.PipelineStartedEvent{Branch: branch, Source: rev, Run: run, Module: "m", Pipeline: "p"})
return b
@@ -58,6 +75,30 @@ func TestAHeadWithNoPipelineStands(t *testing.T) {
t.Fatalf("a pulse for a member: %v", o.pulses)
}
}, false},
{"a head behind a pending revision of its branch is not timed", func(w *Watchman, o *fakeOut, c *clock) {
w.TakeIn(takeInEvent(t, takein.TakeInPending, "m.1"), takeInBodyOn(t, takein.TakeInPending, "m.1", "main", c.t))
w.SourceMoved(moved("main", 7, "", c.t))
if _, ok := o.pulses[signalKey(headRow, headID("main"))]; ok {
t.Fatalf("a pulse for a head behind a pending revision: %v", o.pulses)
}
}, false},
{"a pending revision of another branch does not hold it", func(w *Watchman, o *fakeOut, c *clock) {
w.TakeIn(takeInEvent(t, takein.TakeInPending, "m.1"), takeInBodyOn(t, takein.TakeInPending, "m.1", "other", c.t))
w.SourceMoved(moved("main", 7, "", c.t))
expire(w, c, signalKey(headRow, headID("main")), HeadBound)
}, true},
{"a head whose pending revision is taken in is timed again", func(w *Watchman, o *fakeOut, c *clock) {
w.SourceMoved(moved("main", 7, "", c.t))
w.TakeIn(takeInEvent(t, takein.TakeInPending, "m.1"), takeInBodyOn(t, takein.TakeInPending, "m.1", "main", c.t))
if _, ok := o.pulses[signalKey(headRow, headID("main"))]; ok {
t.Fatalf("the head is timed behind a pending revision: %v", o.pulses)
}
c.t = c.t.Add(HeadBound)
w.TakeIn(takeInEvent(t, takein.TakeInTaken, "m.1"), takeInBodyOn(t, takein.TakeInTaken, "m.1", "main", c.t))
if _, ok := o.findings[key]; !ok {
t.Fatalf("the head, past its five minutes when taken in, stands: %v", o.findings)
}
}, true},
{"a run on request is no head's pipeline", func(w *Watchman, o *fakeOut, c *clock) {
w.SourceMoved(moved("main", 7, "", c.t))
w.PipelineStarted(started("main", 7, "run-1"))
@@ -207,3 +248,38 @@ func TestARecordOpenADayIsAFinding(t *testing.T) {
t.Fatal("the pulse of an ended record stands")
}
}
// design 47 G §502, A-200 (WM16): a record is tied to the entries of the take-in.taken of its milestone `<branch>.<source revision>`; it
// is covered only while one of those entries is held by an `acting-part-late` wait, so another record's late entry covers nothing.
func TestARecordIsCoveredOnlyByTheWaitsOfItsOwnEntries(t *testing.T) {
w, o, c := setup()
taken := func(key string, source int64, module string) {
var m map[string]any
if err := json.Unmarshal(takeInBody(t, takein.TakeInTaken, key, c.t), &m); err != nil {
t.Fatal(err)
}
m["branch"], m["source"] = "main", source
m["entries"] = []map[string]any{{"commit": "c", "fingerprint": "f", "module": module}}
b, _ := json.Marshal(m)
w.TakeIn(takeInEvent(t, takein.TakeInTaken, key), b)
}
taken("m.1", 7, "web")
taken("m.2", 8, "db")
rec := func(id, milestone string) []byte {
b, _ := json.Marshal(meshdelivery.RecordOpenedEvent{Record: id, Milestone: milestone, At: &c.t})
return b
}
w.RecordOpen(rec("rWeb", "main.7"))
w.RecordOpen(rec("rDb", "main.8"))
late, _ := json.Marshal(meshwants.Wait{Step: "place", Part: "mesh-delivery-indexer", Node: "ace", Module: "web", Since: c.t, Bound: "30m"})
w.WaitHeld("mesh-wants", late)
expire(w, c, signalKey(actingRow, "mesh-wants-place-mesh-delivery-indexer-ace-web"), 30*time.Minute)
expire(w, c, signalKey(recordRow, "rWeb"), RecordOpenBound)
expire(w, c, signalKey(recordRow, "rDb"), 0)
if _, ok := o.findings[FindingKey(ScopeDelivery, "rWeb", KindRecordOpen)]; ok {
t.Fatalf("a record whose entry is held by acting-part-late is covered: %v", o.findings)
}
if _, ok := o.findings[FindingKey(ScopeDelivery, "rDb", KindRecordOpen)]; !ok {
t.Fatalf("a record of other entries is open at 24 hours whatever else is late: %v", o.findings)
}
}
@@ -3,6 +3,7 @@ package watch
import (
"encoding/json"
"fmt"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
@@ -93,6 +94,7 @@ func (w *Watchman) RecordOpen(body []byte) {
if r.At != nil && !r.At.IsZero() {
since = *r.At
}
w.recordMilestone[r.Record] = r.Milestone
f := Finding{Scope: ScopeDelivery, ID: r.Record, Kind: KindRecordOpen, Severity: Warning, Source: recordRow, Since: since,
Summary: fmt.Sprintf("the delivery record of %s has stood open since %s", r.Milestone, stamp(since)),
Said: "open for " + RecordOpenBound.String(), Headline: "A delivery is not on every machine yet",
@@ -120,17 +122,31 @@ func (w *Watchman) RecordEnded(body []byte) {
defer w.mu.Unlock()
w.unsignal(recordRow, r.Record)
delete(w.opened, r.Record)
delete(w.recordMilestone, r.Record)
w.unfind(FindingKey(ScopeDelivery, r.Record, KindRecordOpen))
}
// recordCovered says another condition tells the operator why a record is open: a wait past its bound or a full store.
func (w *Watchman) recordCovered() bool {
// recordCovered says another condition tells the operator why a record is open (design 47 G §502, A-200): the record is tied to the
// entries of its source revision's `take-in.taken` by its milestone, `<branch>.<source revision>`, and is covered while one of those
// entries is held by a wait that raised `acting-part-late`, or a store says `full`; a record whose revision is not known to be taken in
// is covered by a standing `take-in-waits`.
func (w *Watchman) recordCovered(record string) bool {
entries, taken := w.takenEntries[w.recordMilestone[record]]
for _, v := range w.findings {
f, err := DecodeFinding(v)
if err != nil {
continue
}
if f.Kind == KindActingPartLate || (f.Kind == KindStoreFilling && f.Severity == Urgent) {
switch {
case f.Kind == KindStoreFilling && f.Severity == Urgent:
return true
case f.Kind == KindActingPartLate && taken:
for _, module := range entries {
if strings.HasSuffix(f.ID, "-"+module) {
return true
}
}
case f.Kind == KindTakeInWaits && !taken:
return true
}
}
@@ -140,11 +156,10 @@ func (w *Watchman) recordCovered() bool {
// rejudgeRecords writes the finding of each record past its bound that nothing covers, and ends it when something does.
func (w *Watchman) rejudgeRecords() {
for id, f := range w.opened {
if w.recordCovered() {
if w.recordCovered(id) {
w.unfind(f.Key())
continue
}
w.find(f)
_ = id
}
}
@@ -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,30 +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 haveKey == uncovered {
return !w.recordCovered()
}
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
}
@@ -215,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") }
@@ -267,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)
@@ -277,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
}
@@ -481,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
@@ -716,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) {
@@ -748,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 := ""
@@ -764,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)
}
@@ -872,9 +712,6 @@ func BusUserWords(user string) (string, string) {
return "the node-engine on " + m, m
}
if m, module, ok := strings.Cut(user, "."); ok {
if module == "node-tools" {
return "the tool runner on " + m, m
}
return module + " on " + m, m
}
return "a client of the bus", ""
@@ -934,13 +771,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},
@@ -1,12 +1,14 @@
package watch
import (
"encoding/json"
"fmt"
"maps"
"slices"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
)
@@ -42,10 +44,24 @@ func (w *Watchman) TakeIn(event string, body []byte) {
w.say("the body of %s says the kind %q, left", event, t.Kind)
return
}
// The wait names the watched branch it is of (design 47 G §493), by the contract's body.
var named meshcatalog.TakeInPending
_ = json.Unmarshal(body, &named)
var taken meshcatalog.TakeInTaken
_ = json.Unmarshal(body, &taken)
key := t.Key
w.mu.Lock()
defer w.mu.Unlock()
w.takeIn(kind, key, t)
w.pendingOf(kind, key, named.Branch)
if kind == takein.TakeInTaken && taken.Branch != "" && taken.Source != 0 && len(taken.Entries) > 0 {
var modules []string
for _, e := range taken.Entries {
modules = append(modules, e.Module)
}
w.takenEntries[fmt.Sprintf("%s.%d", taken.Branch, taken.Source)] = modules // a record's milestone (A-200)
w.rejudgeRecords()
}
}
// SeedTakeIn starts, at the watchman's start and before its consumer on the take-in stream, the wait of every revision
@@ -88,6 +104,28 @@ func (w *Watchman) SeedTakeIn(a takein.PendingAnswer) {
}
}
// pendingOf records or ends the branch an open take-in wait names (design 47 G §497), w.mu held: the head of a branch behind a
// pending revision raises no `pipeline-stands`, and is timed again, from the move of the head, once the branch has none.
func (w *Watchman) pendingOf(kind, key, branch string) {
if kind == takein.TakeInTaken {
branch = w.pendingBranch[key]
delete(w.pendingBranch, key)
if branch != "" && !w.branchPending(branch) {
w.timeHead(branch)
}
return
}
if branch == "" {
return
}
w.pendingBranch[key] = branch
if _, ok := w.heads[branch]; ok {
id := headID(branch)
w.unsignal(headRow, id)
w.unfind(FindingKey(ScopeDelivery, id, KindPipelineStands))
}
}
// takeIn is one take-in event of key, w.mu held.
func (w *Watchman) takeIn(kind, key string, t takein.TakeIn) {
finding := FindingKey(ScopeDelivery, key, KindTakeInWaits)
@@ -1,9 +1,13 @@
package watch
import (
"encoding/json"
"fmt"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
)
// KindManifestRefused is a module whose manifest the indexer's judge refused (novox/hq ADR 0578 §1): every reader holds
@@ -14,66 +18,96 @@ const KindManifestRefused = "manifest-refused"
// (novox/hq ADR 0579 §1): it stays accepted and runs; the warning stands until the manifest is mended.
const KindManifestOutgrown = FindingManifestOutgrown
// KindManifestWaits is a module whose manifest carries a field the indexer's judge does not know yet (novox/hq ADR 0578
// §1, amended): every reader holds it as a refused one and keeps what runs, and the warning stands until a judge knows the
// field or the manifest is mended.
const KindManifestWaits = FindingManifestWaits
// The manifest findings are the indexer's verdicts heard as events (design 47 G §500): `manifest-refused` raises the refusal, and
// `manifest-accepted`, which carries `problems` (what a stricter validator finds in a manifest it still accepts), raises
// `manifest-outgrown` or, with none, clears it and any refusal. The watchman seeds both from the indexer's verdicts at its start
// (VerdictsTool). `manifest-waits` is no longer written: every revision is judged by a validator built from its own heads (§16).
// ModuleVerdict is a change of the indexer's value of a module (ModulesBucket): a value carrying a refusal is a
// finding, one that does not (accepted, or not judged yet) or a deleted key ends it.
func (w *Watchman) ModuleVerdict(module string, value []byte, op Op) {
// VerdictsTool is the indexer's tool that lists its verdicts, which seeds the findings at the start.
const VerdictsTool = IndexerModule + ".indexer_verdicts"
// VerdictRow is one verdict as the tool lists it (the fields the watchman reads).
type VerdictRow struct {
Module string `json:"module"`
State string `json:"state"`
Problems []string `json:"problems"`
Warnings []string `json:"warnings"`
At time.Time `json:"at"`
}
// ManifestRefused hears the seat mesh-catalog's manifest-refused: the module's manifest is refused until a mended one is accepted.
func (w *Watchman) ManifestRefused(e meshcatalog.ManifestRefusedEvent) {
if e.Module == "" {
w.say("a manifest-refused names no module, left")
return
}
w.mu.Lock()
defer w.mu.Unlock()
key := FindingKey(ScopeModule, module, KindManifestRefused)
outgrown := FindingKey(ScopeModule, module, KindManifestOutgrown)
waits := FindingKey(ScopeModule, module, KindManifestWaits)
if op != Put {
w.unfind(key)
w.unfind(outgrown)
w.unfind(waits)
return
}
m, err := meshcatalogclient.DecodeModule(value)
if err != nil {
w.say("the indexer's value of %s cannot be read, so whether its manifest is refused is left as it was: %v", module, err)
return
}
v := meshcatalogclient.ModuleVerdict(module, m)
if v != nil && meshcatalogclient.VerdictWaits(*v) {
w.unfind(key)
w.unfind(outgrown)
w.find(Finding{Scope: ScopeModule, ID: module, Kind: KindManifestWaits, Severity: Warning, Source: ModulesBucket,
Summary: meshcatalogclient.VerdictWhy(*v),
Said: strings.Join(v.Unknown, "; "),
Headline: "The manifest of " + module + " waits for a judge that knows its fields",
Explanation: "The manifest carries " + strings.Join(v.Unknown, ", ") + ", which the indexer's judge does not know yet, so every " +
"reader holds " + module + " as a refused one and keeps what runs; an indexer that knows the fields judges it again " +
"at its start, and the newest build of the indexer refuses a field it does not know.",
Needs: "nothing while a newer indexer is on its way; else a pull request that mends the manifest of " + module})
return
}
w.unfind(waits)
if refused, _ := meshcatalogclient.TakeVerdict(v); !refused {
w.unfind(key)
if v == nil || len(v.Warnings) == 0 {
w.unfind(outgrown)
return
w.verdict(e.Module, e.Problems, nil, true)
}
// manifestAccepted hears the problems a manifest-accepted carries (Accepted calls it, w.mu held): none clears the module's findings, some are its warning.
func (w *Watchman) manifestAccepted(module string, problems []string) {
w.verdict(module, nil, problems, false)
}
// SeedVerdicts raises, at the start, the findings of the newest verdict of each module the indexer's tool listed: a refused one, or an
// accepted one with warnings. A module whose newest verdict is accepted clears both.
func (w *Watchman) SeedVerdicts(rows []VerdictRow) {
newest := map[string]VerdictRow{}
for _, r := range rows {
if old, ok := newest[r.Module]; !ok || r.At.After(old.At) {
newest[r.Module] = r
}
w.find(Finding{Scope: ScopeModule, ID: module, Kind: KindManifestOutgrown, Severity: Warning, Source: ModulesBucket,
Summary: "the manifest of " + module + " breaks rules the indexer's judge holds it to now: " + strings.Join(v.Warnings, "; "),
Said: strings.Join(v.Warnings, "; "),
Headline: "The manifest of " + module + " needs mending",
Explanation: "A stricter judge found problems in a manifest it accepted before; " + module + " stays accepted and " +
"runs, and its next change is refused until they are mended (novox/hq ADR 0579).",
}
w.mu.Lock()
defer w.mu.Unlock()
for module, r := range newest {
switch r.State {
case meshcatalogclient.VerdictRefused:
w.verdict(module, r.Problems, nil, true)
case meshcatalogclient.VerdictAccepted:
w.verdict(module, nil, r.Warnings, false)
}
}
}
// DecodeVerdictRows reads the answer of the indexer's verdicts tool.
func DecodeVerdictRows(raw []byte) ([]VerdictRow, error) {
var a struct {
Verdicts []VerdictRow `json:"verdicts"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return nil, fmt.Errorf("not an answer of the indexer's verdicts: %w", err)
}
return a.Verdicts, nil
}
// verdict writes and clears a module's manifest findings, w.mu held: a refusal (problems, refused) or an outgrown manifest (warnings).
func (w *Watchman) verdict(module string, problems, warnings []string, refused bool) {
refusedKey := FindingKey(ScopeModule, module, KindManifestRefused)
outgrownKey := FindingKey(ScopeModule, module, KindManifestOutgrown)
if refused {
w.unfind(outgrownKey)
w.find(Finding{Scope: ScopeModule, ID: module, Kind: KindManifestRefused, Severity: Warning, Source: "indexer.verdicts",
Summary: "the indexer's judge refused the manifest of " + module + ": " + strings.Join(problems, "; "),
Said: strings.Join(problems, "; "),
Headline: "The manifest of " + module + " is refused",
Explanation: "The indexer's judge refused it, so every node, the grants part and the assigner hold " + module + ": " +
"nothing of it changes until a mended manifest is merged, and every other module goes on.",
Needs: "a pull request that mends the manifest of " + module})
return
}
w.unfind(outgrown)
w.find(Finding{Scope: ScopeModule, ID: module, Kind: KindManifestRefused, Severity: Warning, Source: ModulesBucket,
Summary: meshcatalogclient.VerdictWhy(*v),
Said: strings.Join(v.Problems, "; "),
Headline: "The manifest of " + module + " is refused",
Explanation: "The indexer's judge refused it, so every node, the grants part and the assigner hold " + module + ": " +
"nothing of it changes until a mended manifest is merged, and every other module goes on.",
w.unfind(refusedKey)
if len(warnings) == 0 {
w.unfind(outgrownKey)
return
}
w.find(Finding{Scope: ScopeModule, ID: module, Kind: KindManifestOutgrown, Severity: Warning, Source: "indexer.verdicts",
Summary: "the manifest of " + module + " breaks rules the indexer's judge holds it to now: " + strings.Join(warnings, "; "),
Said: strings.Join(warnings, "; "),
Headline: "The manifest of " + module + " needs mending",
Explanation: "A stricter judge found problems in a manifest it accepted before; " + module + " stays accepted and " +
"runs, and its next change is refused until they are mended (novox/hq ADR 0579).",
Needs: "a pull request that mends the manifest of " + module})
}
@@ -6,6 +6,7 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
)
@@ -45,6 +46,7 @@ func (w *Watchman) WaitEnded(seat string, body []byte) {
w.unsignal(actingRow, id)
w.unfind(FindingKey(ScopeDelivery, id, KindActingPartLate))
w.unfind(FindingKey(ScopeDelivery, id, KindPutStands))
w.rejudgeRecords()
}
func decodeWait(w *Watchman, seat, event string, body []byte) (meshwants.Wait, bool) {
@@ -88,6 +90,7 @@ func (w *Watchman) waitHeld(seat string, a meshwants.Wait) {
return
}
w.find(f)
w.rejudgeRecords() // a record whose entry this wait holds is covered now (§502)
}
// putStands is a put's wait held (`step: put`, design 47 G §498): a put without hold that stands past its bound is said as the
@@ -110,3 +113,99 @@ func (w *Watchman) putStands(id string, a meshwants.Wait, bound time.Duration, f
}
w.find(f)
}
// What a node-engine says of a hold it keeps (design 47 G §499, §541), heard on the seat node-engine's `wait-held` and `wait-ended`
// (`mesh.seat.node-engine.event.<event>.<node>.<module>`, the node the one that says it): a hold of the composition (`step:
// compose`, with its reason) standing past composing's 30 minutes (§266) is the finding `delivery-waits`, naming the node, the
// module and the reason, and a wait for a kept build (`step: kept-build`, naming the build node) standing past building's default
// bound of 60 minutes is `kept-build-waits`. Each clears on `wait-ended`.
// The findings of a node-engine's waits, and the bounds that stand when its event names none.
const (
KindDeliveryWaits = "delivery-waits"
KindKeptBuildWaits = "kept-build-waits"
ComposeBound = 30 * time.Minute
KeptBuildBound = 60 * time.Minute
)
// engineRow is the row of the signals table of a node-engine's wait.
const engineRow = "engine-wait"
func engineWaitID(node, module string) string { return node + "." + module }
// EngineWaitHeld hears the node-engine of node saying `wait-held` for module.
func (w *Watchman) EngineWaitHeld(node, module string, body []byte) {
var a nodeengine.Wait
if err := json.Unmarshal(body, &a); err != nil || a.Since.IsZero() {
w.say("a wait-held of the node-engine of %s for %s does not read, left: %v", node, module, err)
return
}
kind, bound := KindDeliveryWaits, ComposeBound
if a.Step == nodeengine.WaitStepKeptBuild {
kind, bound = KindKeptBuildWaits, KeptBuildBound
}
if d, err := time.ParseDuration(a.Bound); err == nil && d > 0 {
bound = d
}
from := a.Since
if a.ClockFrom != nil {
from = *a.ClockFrom
}
id := engineWaitID(node, module)
what := "holds the composition of " + module
if a.Step == nodeengine.WaitStepKeptBuild {
what = "waits for the kept build of " + module + " on " + orWords(a.Node, "no node")
}
f := Finding{Scope: ScopeDelivery, ID: id, Kind: kind, Severity: Warning, Machine: node, Source: engineRow, Since: a.Since,
Summary: fmt.Sprintf("the node-engine of %s %s: %s, since %s (bound %s)", node, what, orWords(a.Reason, "no reason said"), stamp(a.Since), bound),
Said: "held for " + bound.String(),
Headline: "A module's delivery is held on a machine",
Explanation: "The node-engine of " + node + " " + what + " (" + orWords(a.Reason, "no reason said") + ") and its bound of " +
bound.String() + " has run, so the module is not at the build wanted there.",
Needs: "what it waits on to come: " + orWords(a.Words, orWords(a.Reason, "the node-engine's wait to end"))}
w.mu.Lock()
defer w.mu.Unlock()
w.unsignal(engineRow, id)
w.unfind(f.Key())
if left := bound - w.now().Sub(from); left > 0 {
w.signal(engineRow, id, left, signalDue{Do: dueFind, Finding: f})
return
}
w.find(f)
}
// EngineWaitEnded hears the node-engine of node saying `wait-ended` for module: the hold is over, whichever kind it was.
func (w *Watchman) EngineWaitEnded(node, module string) {
id := engineWaitID(node, module)
w.mu.Lock()
defer w.mu.Unlock()
w.unsignal(engineRow, id)
w.unfind(FindingKey(ScopeDelivery, id, KindDeliveryWaits))
w.unfind(FindingKey(ScopeDelivery, id, KindKeptBuildWaits))
}
// ModuleWaitHeld hears a module's `wait-held` on its own stream (`mesh.stream.<module>.event.wait-held.<module>.<fingerprint>`), the form
// the builder says a key queued behind a package in, with the package as its `part` (design 47 G A-192, ADR 0603 §8): it is timed as any
// acting part's wait. A wait naming no part is a toolchain wait, which the builder times itself (§672), and is left.
func (w *Watchman) ModuleWaitHeld(holder string, body []byte) {
var a meshwants.Wait
if json.Unmarshal(body, &a) == nil && a.Part == "" {
return
}
a2, ok := decodeWait(w, holder, "wait-held", body)
if !ok {
return
}
w.mu.Lock()
defer w.mu.Unlock()
w.waitHeld(holder, a2)
}
// ModuleWaitEnded hears the same module's `wait-ended`.
func (w *Watchman) ModuleWaitEnded(holder string, body []byte) {
var a meshwants.Wait
if json.Unmarshal(body, &a) == nil && a.Part == "" {
return
}
w.WaitEnded(holder, body)
}
@@ -2,9 +2,11 @@ package watch
import (
"encoding/json"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
)
@@ -93,3 +95,66 @@ func TestAPutWithoutHoldThatStandsIsPutStands(t *testing.T) {
t.Fatalf("wait-ended left %v", o.findings)
}
}
// design 47 G §499, §541 (WM4, WM5): a node-engine's hold of the composition standing 30 minutes is `delivery-waits` naming the node,
// the module and the reason, a wait for a kept build standing 60 minutes is `kept-build-waits`, and `wait-ended` clears either.
func TestANodeEnginesWaitIsAFindingPastItsBound(t *testing.T) {
held := func(step, reason string, since time.Time) []byte {
b, _ := json.Marshal(nodeengine.Wait{Step: step, Reason: reason, Node: "build1", Since: since, Owner: "node-engine"})
return b
}
for _, tc := range []struct {
name, step, reason, kind string
bound time.Duration
}{
{"a hold of the composition", nodeengine.WaitStepCompose, "restore point missing", KindDeliveryWaits, 30 * time.Minute},
{"a wait for a kept build", nodeengine.WaitStepKeptBuild, "node away", KindKeptBuildWaits, 60 * time.Minute},
} {
t.Run(tc.name, func(t *testing.T) {
w, o, c := setup()
id := "ace.web"
w.EngineWaitHeld("ace", "web", held(tc.step, tc.reason, c.t))
if o.ttls[signalKey(engineRow, id)] != tc.bound || len(o.findings) != 0 {
t.Fatalf("within the bound: ttl %v, findings %v", o.ttls[signalKey(engineRow, id)], o.findings)
}
expire(w, c, signalKey(engineRow, id), tc.bound)
f, ok := o.findings[FindingKey(ScopeDelivery, id, tc.kind)]
if !ok || len(o.findings) != 1 || !strings.Contains(f.Summary, tc.reason) {
t.Fatalf("past its bound: %v", o.findings)
}
w.EngineWaitEnded("ace", "web")
if len(o.findings) != 0 {
t.Fatalf("wait-ended left %v", o.findings)
}
})
}
}
// design 47 G A-192 (WM13): a key queued behind a package is a wait the builder says on its own stream with the package as its part;
// past its bound it is `acting-part-late`, `wait-ended` clears it, and a toolchain wait (no part) is left to the builder.
func TestAKeyQueuedBehindAPackageIsAnActingPartsWait(t *testing.T) {
body := func(part string, since time.Time) []byte {
b, _ := json.Marshal(meshwants.Wait{Step: "building", Part: part, On: "package:" + part, Module: "web", Fingerprint: "src2:f", Since: since, Bound: "1h0m0s"})
return b
}
const holder = "builder"
w, o, c := setup()
id := waitID(holder, meshwants.Wait{Step: "building", Part: "mesh-sdk@v0.1.14", Module: "web"})
w.ModuleWaitHeld(holder, body("mesh-sdk@v0.1.14", c.t))
if o.ttls[signalKey(actingRow, id)] != time.Hour || len(o.findings) != 0 {
t.Fatalf("within the bound: ttl %v, findings %v", o.ttls[signalKey(actingRow, id)], o.findings)
}
expire(w, c, signalKey(actingRow, id), time.Hour)
if _, ok := o.findings[FindingKey(ScopeDelivery, id, KindActingPartLate)]; !ok || len(o.findings) != 1 {
t.Fatalf("past its bound: %v", o.findings)
}
w.ModuleWaitEnded(holder, body("mesh-sdk@v0.1.14", c.t))
if len(o.findings) != 0 {
t.Fatalf("wait-ended left %v", o.findings)
}
w, o, c = setup()
w.ModuleWaitHeld(holder, body("", c.t.Add(-2*time.Hour)))
if len(o.findings) != 0 || len(o.ttls) != 0 {
t.Fatalf("a toolchain wait was timed by the watchman: %v %v", o.ttls, o.findings)
}
}
@@ -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.
@@ -118,6 +106,12 @@ type Watchman struct {
awaited map[string]bool
// heads is, by watched branch, the head still waiting for its pipeline (pipelines.go).
heads map[string]head
// pendingBranch is, by key, the watched branch of each open take-in wait that names one (takein.go, design 47 G §497).
pendingBranch map[string]string
// takenEntries are the modules of each taken-in revision, by `<branch>.<source revision>`, and recordMilestone each open record's
// milestone, which joins a record to them (records.go, design 47 G §502).
takenEntries map[string][]string
recordMilestone map[string]string
// queued is, by build seat, each step queued on it for a build node and since when (pipelines.go).
queued map[string]map[string]time.Time
// awaySaid is each machine whose machine-away was said and not yet its machine-back, with when it was last heard (presence.go).
@@ -130,10 +124,10 @@ type Watchman struct {
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{}, 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{},
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{},
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{}}
}
@@ -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 {
@@ -753,3 +745,13 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
})
}
}
// design 47 G §7, §428: no part's code names another module: a bus user is said by its own module's name on its machine, the tool
// runner's no differently from any module's.
func TestABusUserIsSaidByItsModuleWithoutAModulesNameInTheCode(t *testing.T) {
for user, want := range map[string]string{"ace.node-tools": "node-tools on ace", "ace.web": "web on ace", "node.ace": "the node-engine on ace"} {
if got, _ := BusUserWords(user); got != want {
t.Errorf("BusUserWords(%q) = %q, want %q", user, got, want)
}
}
}
@@ -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-delivery-builder.builds", "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.
@@ -126,11 +126,11 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
}
// The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). The waits the
// indexer holds at the watchman's start are asked of its tool (ADR 0633 §5), a verb edge as the view's.
if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_read_modules", "seat:mesh-hand-acts.hand-act", "seat:mesh-conditions.conditions"}; !reflect.DeepEqual(m.Invokes, want) {
if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_pending", "mesh-delivery-indexer.indexer_verdicts", "seat:mesh-hand-acts.hand-act", "seat:mesh-conditions.conditions"}; !reflect.DeepEqual(m.Invokes, want) {
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-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", "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-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,12 +172,8 @@ 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")},
"builds": {watch.BuilderBuildsBucket, bucket(m.Reads, "mesh-delivery-builder.builds")},
"wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")},
"provisions": {watch.ProvisionsBucket, bucket(m.Reads, "mesh-provisioner.provisions")},
"modules": {watch.ModulesBucket, bucket(m.Consumes, "mesh-delivery-indexer.views.modules")},
"assigner": {watch.AssignerModule + ".assigner_read_wants", entry(m.Invokes, "mesh-delivery-assigner.assigner_read_wants")},
"indexer": {watch.IndexerModule + ".indexer_pending", entry(m.Invokes, "mesh-delivery-indexer.indexer_pending")},
} {
+8 -6
View File
@@ -16,15 +16,12 @@
}
],
"reads": [
"mesh-delivery-builder.builds",
"mesh-host.health",
"mesh-provisioner.provisions",
"node-engine.have"
"mesh-provisioner.provisions"
],
"invokes": [
"mesh-delivery-assigner.assigner_read_wants",
"mesh-delivery-indexer.indexer_pending",
"mesh-delivery-indexer.indexer_read_modules",
"mesh-delivery-indexer.indexer_verdicts",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-conditions.conditions"
],
@@ -32,8 +29,8 @@
"mesh-delivery-assigner.views.wants",
"mesh-delivery-indexer.take-in.pending",
"mesh-delivery-indexer.take-in.taken",
"mesh-delivery-indexer.views.modules",
"mesh-catalog.manifest-accepted",
"mesh-catalog.manifest-refused",
"mesh-delivery.source-moved",
"mesh-delivery.filling",
"mesh-delivery.full",
@@ -67,6 +64,11 @@
"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"