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

Watchman: the node-engine's, wants', build steps', builder's and catalog's names come from their clients and seats, the watchdogs' codec from meshwatchdogsclient; the holders and buckets it names are its own constants checked against its manifest (D-11; ADR 0634 §1, §2, §4; issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 16:28:51 +01:00
parent 4382ac1fcb
commit 88d100b9d5
16 changed files with 405 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go
+1 -2
View File
@@ -10,7 +10,6 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/membership"
"git.novox.be/novox/mesh-sdk/go/process"
@@ -36,7 +35,7 @@ const (
func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, ...any)) {
switch {
case subject == meshcatalog.SubjectManifestAccepted:
var a delivery.ManifestAccepted
var a meshcatalog.ManifestAcceptedEvent
if err := json.Unmarshal(body, &a); err != nil || a.Module == "" {
logf("a manifest-accepted that does not read is left: %v", err)
return
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go
+68 -63
View File
@@ -17,8 +17,13 @@ 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-contracts/placement/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/views/viewstest"
@@ -62,8 +67,8 @@ accounts {
// 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, delivery.AssignerModule, wantsState, delivery.AssignerModule+".assigner_read_wants", "n1_"+delivery.WatchmanModule)
viewstest.Serve(t, ctx, js, nc, delivery.IndexerModule, modulesState, delivery.IndexerModule+".indexer_read_modules", "n1_"+delivery.WatchmanModule)
viewstest.Serve(t, ctx, js, nc, watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants", "n1_"+watch.WatchmanModule)
viewstest.Serve(t, ctx, js, nc, meshcatalogclient.Module, modulesState, meshcatalogclient.Module+".indexer_read_modules", "n1_"+watch.WatchmanModule)
}
// TestWatchmanWhole starts the program the way the node does (its entry, run, with its environment and a broker file)
@@ -111,12 +116,12 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
defer cancel()
kvs := map[string]jetstream.KeyValue{}
for _, c := range []jetstream.KeyValueConfig{
{Bucket: delivery.PresenceBucket, History: 20},
{Bucket: delivery.StuckBucket, History: 20},
{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
{Bucket: delivery.AssignerWantsBucket, History: 20},
{Bucket: delivery.HaveBucket, History: 20},
{Bucket: watch.PresenceBucket, History: 20},
{Bucket: watch.StuckBucket, History: 20},
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: watch.StepsBucket("mesh-build-go"), History: 5},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: watch.HaveBucket, History: 20},
{Bucket: watch.FindingsBucket, History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -174,11 +179,11 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
return e, err == nil
}
online := func(key string) (bool, bool) {
e, ok := has(delivery.PresenceBucket, key)
e, ok := has(watch.PresenceBucket, key)
if !ok {
return false, false
}
p, err := delivery.DecodePresence(e.Value())
p, err := meshwatchdogsclient.DecodeNodePresence(e.Value())
return p.Online, err == nil
}
until := func(what string, ok func() bool) {
@@ -191,14 +196,14 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
t.Fatalf("never saw: %s", what)
}
pulseOf := func(bucket, key string) string {
k, err := delivery.PulseKey(bucket, key)
k, err := meshwatchdogsclient.PulseKey(bucket, key)
if err != nil {
t.Fatal(err)
}
return k
}
stuckOf := func(bucket, key string) string {
k, err := delivery.StuckKey(bucket, key)
k, err := meshwatchdogsclient.StuckKey(bucket, key)
if err != nil {
t.Fatal(err)
}
@@ -214,12 +219,12 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
until("node n1 offline", func() bool { on, ok := online("n1"); return ok && !on })
// An ask put on a build seat gets a pulse; one no node takes lets the pulse expire, and that writes a stuck key.
askStream, _ := delivery.AskStream(delivery.AskGo)
steps := kvs[delivery.StepsBucket("mesh-build-go")]
askStream, _ := buildstep.AskStream(buildstep.AskGo)
steps := kvs[watch.StepsBucket("mesh-build-go")]
// A node takes an ask: it writes the step key running (and passed when the ask ends).
putStep := func(id, state string) {
t.Helper()
body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: time.Now().UTC()})
body, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: time.Now().UTC()})
if err != nil {
t.Fatal(err)
}
@@ -228,16 +233,16 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
}
}
const ask1, ask2 = "m0000001aaaaaa", "m0000002bbbbbb"
putStep(ask1, delivery.StepRunning)
until("a pulse for the ask a node took", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask1)); return ok })
var stuck delivery.Stuck
putStep(ask1, buildstep.StepRunning)
until("a pulse for the ask a node took", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask1)); return ok })
var stuck meshwatchdogsclient.Stuck
until("a stuck key for the ask no node ended", func() bool {
e, ok := has(delivery.StuckBucket, stuckOf(askStream, ask1))
e, ok := has(watch.StuckBucket, stuckOf(askStream, ask1))
if !ok {
return false
}
var err error
stuck, err = delivery.DecodeStuck(e.Value())
stuck, err = meshwatchdogsclient.DecodeStuck(e.Value())
return err == nil
})
if stuck.Moved || stuck.Bucket != askStream || stuck.Key != ask1 {
@@ -245,7 +250,7 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
}
// The node n1, gone offline and never back, is stuck as well.
until("a stuck key for node n1 that did not come back", func() bool {
_, ok := has(delivery.StuckBucket, stuckOf(delivery.PresenceBucket, "n1"))
_, ok := has(watch.StuckBucket, stuckOf(watch.PresenceBucket, "n1"))
return ok
})
@@ -253,18 +258,18 @@ func TestWatchmanWhole(t *testing.T) { // 1135/AC8
n2 := connect("node.n2")
until("node n2 online", func() bool { on, ok := online("n2"); return ok && on })
n2.Close()
until("a pulse for node n2 gone offline", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return ok })
until("a pulse for node n2 gone offline", func() bool { _, ok := has(watch.PulseBucket, pulseOf(watch.PresenceBucket, "n2")); return ok })
connect("node.n2")
until("node n2 online again", func() bool { on, ok := online("n2"); return ok && on })
until("the pulse of node n2 deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return !ok })
until("the pulse of node n2 deleted", func() bool { _, ok := has(watch.PulseBucket, pulseOf(watch.PresenceBucket, "n2")); return !ok })
// An ask a node takes at once is the same: no stuck key, however long the bound passes.
putStep(ask2, delivery.StepRunning)
until("a pulse for the second ask", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return ok })
putStep(ask2, delivery.StepPassed)
until("the pulse of the second ask deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(askStream, ask2)); return !ok })
putStep(ask2, buildstep.StepRunning)
until("a pulse for the second ask", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask2)); return ok })
putStep(ask2, buildstep.StepPassed)
until("the pulse of the second ask deleted", func() bool { _, ok := has(watch.PulseBucket, pulseOf(askStream, ask2)); return !ok })
time.Sleep(6 * time.Second) // twice the bound and the marker time
for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(askStream, ask2)} {
if _, ok := has(delivery.StuckBucket, k); ok {
for _, k := range []string{stuckOf(watch.PresenceBucket, "n2"), stuckOf(askStream, ask2)} {
if _, ok := has(watch.StuckBucket, k); ok {
t.Errorf("a stuck key %s was written for a move that happened before the bound", k)
}
}
@@ -362,12 +367,12 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
defer cancel()
kvs := map[string]jetstream.KeyValue{}
for _, c := range []jetstream.KeyValueConfig{
{Bucket: delivery.PresenceBucket, History: 20},
{Bucket: delivery.StuckBucket, History: 20},
{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
{Bucket: delivery.AssignerWantsBucket, History: 20},
{Bucket: delivery.HaveBucket, History: 20},
{Bucket: watch.PresenceBucket, History: 20},
{Bucket: watch.StuckBucket, History: 20},
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: watch.StepsBucket("mesh-build-go"), History: 5},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: watch.HaveBucket, History: 20},
{Bucket: watch.FindingsBucket, History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -379,11 +384,11 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
}
serveOwners(t, ctx, js, ctl)
const id = "m0000009ffffff"
body, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: time.Now().UTC()})
body, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: buildstep.StepRunning, Started: time.Now().UTC()})
if err != nil {
t.Fatal(err)
}
if _, err := kvs[delivery.StepsBucket("mesh-build-go")].Put(ctx, "n1."+id, body); err != nil {
if _, err := kvs[watch.StepsBucket("mesh-build-go")].Put(ctx, "n1."+id, body); err != nil {
t.Fatal(err)
}
broker := filepath.Join(dir, "broker")
@@ -396,13 +401,13 @@ func anAskRunningBeforeTheStartGetsItsPulse(t *testing.T) { // 1909
done := make(chan error, 1)
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
t.Cleanup(func() { stop(); <-done })
stream, _ := delivery.AskStream(delivery.AskGo)
pk, err := delivery.PulseKey(stream, id)
stream, _ := buildstep.AskStream(buildstep.AskGo)
pk, err := meshwatchdogsclient.PulseKey(stream, id)
if err != nil {
t.Fatal(err)
}
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if _, err := kvs[delivery.PulseBucket].Get(ctx, pk); err == nil {
if _, err := kvs[watch.PulseBucket].Get(ctx, pk); err == nil {
return
}
}
@@ -449,12 +454,12 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
defer cancel()
kvs := map[string]jetstream.KeyValue{}
for _, c := range []jetstream.KeyValueConfig{
{Bucket: delivery.PresenceBucket, History: 20},
{Bucket: delivery.StuckBucket, History: 20},
{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: delivery.StepsBucket("mesh-build-go"), History: 5},
{Bucket: delivery.AssignerWantsBucket, History: 20},
{Bucket: delivery.HaveBucket, History: 20},
{Bucket: watch.PresenceBucket, History: 20},
{Bucket: watch.StuckBucket, History: 20},
{Bucket: watch.PulseBucket, LimitMarkerTTL: time.Second},
{Bucket: watch.StepsBucket("mesh-build-go"), History: 5},
{Bucket: watch.AssignerWantsBucket, History: 20},
{Bucket: watch.HaveBucket, History: 20},
{Bucket: watch.FindingsBucket, History: 5},
{Bucket: firstViewsBucket, History: 1},
} {
@@ -465,27 +470,27 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
kvs[c.Bucket] = kv
}
serveOwners(t, ctx, js, ctl)
fpNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64)
fpNew := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64)
fpOld := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
putWant := func(node, module, fp string) {
t.Helper()
body, err := delivery.EncodeWant(delivery.Want{FP: fp, Composed: "sha256:" + strings.Repeat("c", 64)})
body, err := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fp, Composed: "sha256:" + strings.Repeat("c", 64)})
if err != nil {
t.Fatal(err)
}
key, _ := delivery.WantKey(node, module)
if _, err := kvs[delivery.AssignerWantsBucket].Put(ctx, key, body); err != nil {
key, _ := meshwantsclient.WantKey(node, module)
if _, err := kvs[watch.AssignerWantsBucket].Put(ctx, key, body); err != nil {
t.Fatal(err)
}
}
putHave := func(node, module, fp string) {
t.Helper()
body, err := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
body, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
if err != nil {
t.Fatal(err)
}
key, _ := delivery.HaveKey(node, module)
if _, err := kvs[delivery.HaveBucket].Put(ctx, key, body); err != nil {
key, _ := nodeengineclient.HaveKey(node, module)
if _, err := kvs[watch.HaveBucket].Put(ctx, key, body); err != nil {
t.Fatal(err)
}
}
@@ -502,8 +507,8 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
t.Cleanup(func() { stop(); <-done })
pulse := func(node, module string) string {
wk, _ := delivery.WantKey(node, module)
pk, err := delivery.PulseKey(delivery.AssignerWantsBucket, wk)
wk, _ := meshwantsclient.WantKey(node, module)
pk, err := meshwatchdogsclient.PulseKey(watch.AssignerWantsBucket, wk)
if err != nil {
t.Fatal(err)
}
@@ -518,15 +523,15 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
}
t.Fatalf("never saw: %s", what)
}
has := func(key string) bool { _, err := kvs[delivery.PulseBucket].Get(ctx, key); return err == nil }
has := func(key string) bool { _, err := kvs[watch.PulseBucket].Get(ctx, key); return err == nil }
until("a pulse for the want that stood before the start", func() bool { return has(pulse("n2", "web")) })
putWant("n2", "mail", fpNew)
until("a pulse for a want written after the start", func() bool { return has(pulse("n2", "mail")) })
e, err := kvs[delivery.PulseBucket].Get(ctx, pulse("n2", "web"))
e, err := kvs[watch.PulseBucket].Get(ctx, pulse("n2", "web"))
if err != nil {
t.Fatal(err)
}
if p, err := delivery.DecodePulse(e.Value()); err != nil || !strings.Contains(p.Owes, "n2") {
if p, err := meshwatchdogsclient.DecodePulse(e.Value()); err != nil || !strings.Contains(p.Owes, "n2") {
t.Errorf("the pulse = %+v, %v; want it owed by the agent of n2", p, err)
}
putHave("n2", "web", fpNew)
@@ -535,9 +540,9 @@ func aWantIsMirroredAsAPulseAndTheHaveNamingItEndsIt(t *testing.T) {
t.Errorf("the pulse of another want ended with web's have")
}
for _, m := range []string{"web"} {
wk, _ := delivery.WantKey("n2", m)
sk, _ := delivery.StuckKey(delivery.AssignerWantsBucket, wk)
if _, err := kvs[delivery.StuckBucket].Get(ctx, sk); err == nil {
wk, _ := meshwantsclient.WantKey("n2", m)
sk, _ := meshwatchdogsclient.StuckKey(watch.AssignerWantsBucket, wk)
if _, err := kvs[watch.StuckBucket].Get(ctx, sk); err == nil {
t.Errorf("a stuck key was written for the want of %s that was applied", m)
}
}
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go
+29 -27
View File
@@ -25,8 +25,10 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/provisions"
"mesh-delivery-watchman/internal/socket"
@@ -64,7 +66,7 @@ func (o *kvOut) PutFinding(key string, value []byte) error {
func (o *kvOut) DeleteFinding(key string) error { return o.findings.Delete(o.ctx, key) }
func (o *kvOut) PutPresence(key string, online bool) error {
value, err := delivery.EncodePresence(delivery.Presence{Online: online})
value, err := meshwatchdogsclient.EncodeNodePresence(meshwatchdogsclient.NodePresence{Online: online})
if err != nil {
return err
}
@@ -184,7 +186,7 @@ func createdOr(e jetstream.KeyValueEntry, at time.Time) time.Time {
// The watchman's own states of its first views (views.StatusState); its findings of a first view unanswered past its
// bound go to its findings bucket with the others (ADR 0615 §6). The owners' states it follows are the assigner's wants
// and the indexer's modules (DLV-M5).
var firstViewsBucket = delivery.WatchmanModule + "_" + views.StatusState
var firstViewsBucket = watch.WatchmanModule + "_" + views.StatusState
const (
wantsState = "wants"
@@ -219,12 +221,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
for _, b := range []struct {
name string
into *jetstream.KeyValue
}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck}, {watch.FindingsBucket, &kvs.findings}} {
}{{watch.PresenceBucket, &kvs.presence}, {watch.PulseBucket, &kvs.pulse}, {watch.StuckBucket, &kvs.stuck}, {watch.FindingsBucket, &kvs.findings}} {
if *b.into, err = process.EssentialBucket(ctx, js, "watchman", b.name, "say who is online or what stopped moving"); err != nil {
return err
}
}
kvs.haves = process.OptionalBucket(ctx, js, delivery.HaveBucket, "a want makes no pulse, so a move that stops is not seen, until the watchman starts again", logger.Printf)
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
@@ -237,15 +239,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// The module has one durable consumer of EVENTS: one hub pulls it for the views reader, the indexer's take-in events
// and the seats' events, started once every part is registered.
hub := process.NewEventHub(logger.Printf)
reader := &views.ProcessReader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf, Hub: hub,
reader := &views.ProcessReader{Module: watch.WatchmanModule, Call: calls.Call, Logf: logger.Printf, Hub: hub,
Findings: func() views.FindingWriter {
return sdkfindings.Writer(ctx, func() jetstream.KeyValue { return kvs.findings })
}}
wants, err := reader.Follow(delivery.AssignerModule, wantsState, delivery.AssignerModule+".assigner_read_wants")
wants, err := reader.Follow(watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants")
if err != nil {
return err
}
modules, err := reader.Follow(delivery.IndexerModule, modulesState, delivery.IndexerModule+".indexer_read_modules")
modules, err := reader.Follow(meshcatalogclient.Module, modulesState, meshcatalogclient.Module+".indexer_read_modules")
if err != nil {
return err
}
@@ -253,8 +255,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: kvs.findings}, 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(delivery.AssignerWantsBucket)
w.Awaits(delivery.ModulesBucket)
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)
@@ -270,7 +272,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
each func(jetstream.KeyValueEntry)
}{
{kvs.presence, func(e jetstream.KeyValueEntry) {
if p, err := delivery.DecodePresence(e.Value()); err == nil && opOf(e) == watch.Put {
if p, err := meshwatchdogsclient.DecodeNodePresence(e.Value()); err == nil && opOf(e) == watch.Put {
w.SeedPresence(e.Key(), p.Online, e.Created())
}
}},
@@ -305,27 +307,27 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
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, delivery.HaveBucket, err, "moves are not watched until the watchman starts again")
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, delivery.BuilderBuildsBucket, "a build waiting for a package is not watched until the watchman starts again", logger.Printf); kv != nil {
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, delivery.BuilderBuildsBucket, err, "a build waiting for a package is not watched until the watchman starts again")
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(delivery.BuilderBuildsBucket)
w.ReadWhole(watch.BuilderBuildsBucket)
}
}
// The build seats' steps: a step key `<node>.<ask id>` written is the ask taken (issue 1887 A8b). A seat whose build module
// keeps no steps bucket (yet) is said and left; the asks of its stream are mirrored all the same.
taken := map[string]jetstream.KeyWatcher{}
for _, lang := range delivery.AskLanguages {
module, _ := delivery.BuildModuleOf(lang)
kv, err := js.KeyValue(ctx, delivery.StepsBucket(module))
for _, lang := range buildstep.AskLanguages {
module, _ := buildstep.BuildModuleOf(lang)
kv, err := js.KeyValue(ctx, watch.StepsBucket(module))
if err != nil {
logger.Printf("the steps of %s cannot be read (%v): its asks are mirrored and never seen taken", module, err)
continue
@@ -337,7 +339,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
continue
}
taken[lang] = sw
if bucket, err := delivery.AskStream(lang); err == nil {
if bucket, err := buildstep.AskStream(lang); err == nil {
w.ReadWhole(bucket)
}
}
@@ -406,16 +408,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// 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, delivery.NodeHealthBucket); err == nil {
if kv, err := js.KeyValue(ctx, watch.NodeHealthBucket); err == nil {
if healthWatch, err = settle(ctx, kv, healthEach); err != nil {
process.NotWatched(logger.Printf, delivery.NodeHealthBucket, err, "no node's own health is said until the watchman starts again")
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, delivery.NodeHealthBucket, "no node's own health is said",
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, delivery.NodeHealthBucket, err, "no node's own health is said until the watchman starts again")
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) }()
@@ -453,7 +455,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
w.TakeInCaughtUp()
}
}
if err := hub.Handle("mesh.mod."+delivery.IndexerModule+".event.take-in.*", func(m jetstream.Msg) { takeInHeard(w, m, logger.Printf) }); err != nil {
if err := hub.Handle("mesh.mod."+meshcatalogclient.Module+".event.take-in.*", func(m jetstream.Msg) { takeInHeard(w, m, logger.Printf) }); err != nil {
return err
}
}
@@ -473,13 +475,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
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, delivery.ModulesBucket, verdictEach) }()
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, delivery.AssignerWantsBucket, func(e jetstream.KeyValueEntry) {
errs <- readLate(ctx, w, wants, watch.AssignerWantsBucket, func(e jetstream.KeyValueEntry) {
w.Want(e.Key(), e.Value(), opOf(e), createdOr(e, time.Now()))
})
}()
@@ -533,7 +535,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
const eventsStream = "EVENTS"
// pendingTool is the indexer's tool that answers the waits it holds (ADR 0633 §5), as the view asks it.
const pendingTool = delivery.IndexerModule + ".indexer_pending"
const pendingTool = meshcatalogclient.Module + ".indexer_pending"
// seedTakeIn asks the indexer the waits it holds now and starts each (Watchman.SeedTakeIn). An indexer that does not
// answer is said: the waits are then known only from the events heard after.
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go
+3 -4
View File
@@ -10,7 +10,6 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/views"
@@ -82,11 +81,11 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
// The watchman's want wait starts at the entry's time (createdOr): through mesh-sdk's views.ProcessReader it is
// the time the owner's bus wrote the key, never the watchman's start.
written := time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC)
r := &views.ProcessReader{Module: delivery.WatchmanModule, Call: func(string, any) (json.RawMessage, error) {
return json.Marshal(views.Answer{V: views.Version, Owner: delivery.AssignerModule, State: wantsState, Revision: 3, Created: written,
r := &views.ProcessReader{Module: watch.WatchmanModule, Call: func(string, any) (json.RawMessage, error) {
return json.Marshal(views.Answer{V: views.Version, Owner: watch.AssignerModule, State: wantsState, Revision: 3, Created: written,
At: written.Add(time.Hour), Values: map[string]json.RawMessage{"ace.mail": json.RawMessage(`{}`)}, Written: map[string]time.Time{"ace.mail": written}})
}}
p, err := r.Follow(delivery.AssignerModule, wantsState, delivery.AssignerModule+".assigner_read_wants")
p, err := r.Follow(watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants")
if err != nil {
t.Fatal(err)
}
modules/mesh-delivery-watchman/go.mod
+7
View File
@@ -3,6 +3,13 @@ module mesh-delivery-watchman
go 1.26.0
require (
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/placement/go/clients/meshwantsclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/go/seats/meshbroker v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
modules/mesh-delivery-watchman/internal/watch/busstate.go
+4 -3
View File
@@ -7,7 +7,8 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// **A refused or overdue creation of bus state is loud** (novox/hq ADR 0595 §4; design 25 § 12.2; DLV-W1). The watchman
@@ -60,7 +61,7 @@ func refusedKey(module, local string) string {
// Accepted is the indexer's manifest-accepted heard: each entry it asks for waits OverdueBound for the bus module's word,
// and every entry of the module this watchman waits on or holds a finding of that it no longer asks for is ended.
func (w *Watchman) Accepted(a delivery.ManifestAccepted) {
func (w *Watchman) Accepted(a meshcatalog.ManifestAcceptedEvent) {
w.mu.Lock()
defer w.mu.Unlock()
asked := map[string]bool{}
@@ -100,7 +101,7 @@ func (w *Watchman) Accepted(a delivery.ManifestAccepted) {
func (w *Watchman) markCreated(module, local string) {
pk := signalKey(rowBusCreated, busStateID(module, local))
owes, _ := json.Marshal(signalDue{Do: dueCreated})
value, err := delivery.EncodePulse(delivery.Pulse{Bucket: signalBucket, Key: strings.TrimPrefix(pk, signalBucket+"."), Owes: string(owes)})
value, err := meshwatchdogsclient.EncodePulse(meshwatchdogsclient.Pulse{Bucket: signalBucket, Key: strings.TrimPrefix(pk, signalBucket+"."), Owes: string(owes)})
if err == nil {
err = w.out.PutPulse(pk, value, 0)
}
modules/mesh-delivery-watchman/internal/watch/findings_test.go
+54 -49
View File
@@ -7,9 +7,14 @@ 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/placement/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-delivery/sdk/heard"
@@ -25,9 +30,9 @@ func mustEncode(t *testing.T, f Finding) []byte {
return v
}
func haveOf(t *testing.T, phase delivery.Phase, health delivery.Health, detail string) []byte {
func haveOf(t *testing.T, phase string, health string, detail string) []byte {
t.Helper()
b, err := delivery.EncodeHave(delivery.Have{FP: fpNew, Phase: phase, Health: health, Detail: detail})
b, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpNew, Phase: phase, Health: health, Detail: detail})
if err != nil {
t.Fatal(err)
}
@@ -136,15 +141,15 @@ func TestTheWatchmansFindings(t *testing.T) {
}, 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 := delivery.EncodeModule(delivery.Module{FP2: "src2:" + strings.Repeat("a", 64), Manifest: strings.Repeat("b", 64),
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(delivery.VerdictRefused, "removes a stream"), Put)
w.ModuleVerdict("power", value(delivery.VerdictRefused, "shows nothing"), Put)
w.ModuleVerdict("nats", value(meshcatalogclient.VerdictRefused, "removes a stream"), Put)
w.ModuleVerdict("power", value(meshcatalogclient.VerdictRefused, "shows nothing"), Put)
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)
@@ -152,7 +157,7 @@ func TestTheWatchmansFindings(t *testing.T) {
w.ModuleVerdict("nats", value(""), Put)
w.ModuleVerdict("power", nil, Delete)
// ADR 0579 §1: an accepted manifest a stricter judge outgrew is a warning until mended.
outgrown, _ := delivery.EncodeModule(delivery.Module{FP2: "src2:" + strings.Repeat("a", 64), Manifest: strings.Repeat("b", 64),
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)
if f := o.findings[FindingKey(ScopeModule, "node-tools", KindManifestOutgrown)]; f.Severity != Warning || !strings.Contains(f.Summary, "says no credential form") {
@@ -161,8 +166,8 @@ func TestTheWatchmansFindings(t *testing.T) {
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, _ := delivery.EncodeModule(delivery.Module{FP2: "src2:" + strings.Repeat("a", 64), Manifest: strings.Repeat("b", 64),
Repository: "novox/mesh-catalog", Commit: "c", Verdict: delivery.VerdictRefused, Problems: []string{"waits"}, Unknown: []string{"core-credentials"}})
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") {
t.Fatalf("findings = %v", o.findings)
@@ -173,10 +178,10 @@ func TestTheWatchmansFindings(t *testing.T) {
w.ModuleVerdict("nats", value(""), Put)
}, 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) delivery.ManifestAccepted {
a := delivery.ManifestAccepted{Module: "mesh-knowledge", Digest: strings.Repeat("d", 64)}
accepted := func(locals ...string) meshcatalog.ManifestAcceptedEvent {
a := meshcatalog.ManifestAcceptedEvent{Module: "mesh-knowledge", Digest: strings.Repeat("d", 64)}
for _, l := range locals {
a.BusState = append(a.BusState, catalogue.BusStateEntry{Name: l, Storage: catalogue.StorageLedger})
a.BusState = append(a.BusState, meshcatalog.ManifestAcceptedEventBusStateItem{Name: l, Storage: "ledger"})
}
return a
}
@@ -282,7 +287,7 @@ 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 := delivery.EncodeHave(delivery.Have{FP: "src2:" + strings.Repeat("a", 64), Phase: delivery.PhaseRunning, Health: delivery.HealthOK, Kept: kept})
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)
}
@@ -310,7 +315,7 @@ func TestTheWatchmansFindings(t *testing.T) {
if f := o.findings[k]; f.Kind != KindStuck || !strings.Contains(f.Summary, "the ask ask1 to end on n1") {
t.Fatalf("findings = %v", o.findings)
}
w.Step("go", "n1.ask1", endedStep("ask1", delivery.StepPassed, t0), Put)
w.Step("go", "n1.ask1", endedStep("ask1", buildstep.StepPassed, t0), Put)
}, map[string]string{}},
{"WM2 S1: a node offline past its bound is silent, and back online it is not", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
w.Connect("node.ace")
@@ -406,7 +411,7 @@ func TestTheWatchmansFindings(t *testing.T) {
}, 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, delivery.PhaseFailed, delivery.HealthUnknown, "refused: the package manager (pacman) does not answer"), Put)
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)
}
@@ -414,7 +419,7 @@ func TestTheWatchmansFindings(t *testing.T) {
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, delivery.PhaseRunning, delivery.HealthOK, ""), Put)
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")
@@ -426,23 +431,23 @@ func TestTheWatchmansFindings(t *testing.T) {
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, delivery.PhaseRunning, delivery.HealthFailed, "the unit failed"), Put)
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, delivery.PhaseRunning, delivery.HealthFailed, "blip"), Put)
w.Have("ace.quick", haveOf(t, delivery.PhaseRunning, delivery.HealthOK, ""), Put) // recovered within the bound
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, delivery.PhaseRunning, delivery.HealthWaits, "a part waits for the secret token"), Put) // needs-operator is not observed here
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, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
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)
@@ -457,14 +462,14 @@ func TestTheWatchmansFindings(t *testing.T) {
}, 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, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
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, delivery.PhaseRunning, delivery.HealthFailed, "down since 19:40"), Put)
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) {
@@ -486,22 +491,22 @@ func TestTheWatchmansFindings(t *testing.T) {
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 delivery.NodeHealth) []byte {
b, err := delivery.EncodeNodeHealth(h)
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, delivery.PhaseRunning, delivery.HealthOK, ""), Put) // the control node: the bus node
w.NodeHealth("novox", report(delivery.NodeHealth{Node: "novox", Network: &delivery.NetworkHealth{State: "unhealthy", Parts: []delivery.NetworkPart{
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: &delivery.UnitsHealth{State: "degraded", Failed: []delivery.FailedUnit{{Unit: "cups.service", Scope: "system", Result: "exit-code"}}},
Relogin: []delivery.NodeAccount{{Account: "jochen", Module: "docker", Reason: "a group"}},
Found: []delivery.NodeDirectory{{Module: "photos", Resource: "data", Path: "/srv/photos", Reason: "owner differs"}},
Fallback: &delivery.LoginFallback{Node: "novox", From: "node.novox--b", Why: "refused twice", At: time.Date(2026, 11, 2, 12, 0, 0, 0, time.UTC)}}), Put)
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} {
@@ -512,12 +517,12 @@ func TestTheWatchmansFindings(t *testing.T) {
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(delivery.NodeHealth{Node: "novox"}), Put) // all well again
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, _ := delivery.EncodeNodeHealth(delivery.NodeHealth{Node: node, Network: &delivery.NetworkHealth{State: "unhealthy",
Parts: []delivery.NetworkPart{{Part: part, State: "unhealthy", Reason: part + " fails", Said: "x", Targets: targets}}}})
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.
@@ -555,11 +560,11 @@ func TestTheWatchmansFindings(t *testing.T) {
}, 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, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: node, TunnelKey: key}})
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, _ := delivery.EncodeNodeHealth(delivery.NodeHealth{Node: from, Peers: []delivery.PeerSeen{{Key: "kL", Seen: seen}}})
b, _ := nodeengineclient.EncodeHealth(nodeengine.HealthByNodeValue{Node: from, Peers: []nodeengine.PeerSeen{{Key: "kL", Seen: seen}}})
return b
}
w.NodeHealth("ace", peers("ace", false), Put)
@@ -580,21 +585,21 @@ func TestTheWatchmansFindings(t *testing.T) {
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, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
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, delivery.PhaseRemoved, delivery.HealthUnknown, ""), Put)
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, delivery.PhaseRunning, delivery.HealthOK, ""), Put)
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, delivery.PhaseRunning, delivery.HealthOK, ""), Put)
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)
@@ -638,7 +643,7 @@ func TestTheWatchmansFindings(t *testing.T) {
}
}, 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, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
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 {
@@ -667,26 +672,26 @@ func TestTheWatchmansFindings(t *testing.T) {
}, 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) {
w.SetStepBound(30 * time.Minute)
held, _ := delivery.EncodeWant(delivery.Want{FP: fpOld, Self: true, Newest: fpNew, Rule: delivery.RuleCanary, Words: "the canary ace does not run it yet"})
released, _ := delivery.EncodeWant(delivery.Want{FP: fpNew, Self: true})
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})
asked, _ := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"},
Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: provisions.Asked, At: t0})
provided, _ := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"},
Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: provisions.Provided, At: t0})
bkey, _ := delivery.BuilderBuildKey("web", fpNew)
queued, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildQueued, Waits: "mesh-sdk@v0.1.14"})
building, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "n1"})
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.Want("want.ace.web", held, Put, t0)
w.Provision("ace.photos.oidc", asked, Put)
w.Build(bkey, queued, Put, t0)
waits := map[string]string{delivery.AssignerWantsBucket: "want.ace.web", provisions.ProvisionsBucket: "ace.photos.oidc", delivery.BuilderBuildsBucket: bkey}
waits := map[string]string{AssignerWantsBucket: "want.ace.web", provisions.ProvisionsBucket: "ace.photos.oidc", BuilderBuildsBucket: bkey}
if len(o.pulses) != 3 || 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.
c.t = c.t.Add(20 * time.Minute)
held2, _ := delivery.EncodeWant(delivery.Want{FP: fpOld, Self: true, Newest: fpNew, Rule: delivery.RuleCanary, Words: "the canary ace runs it, not healthy yet"})
held2, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fpOld, Self: true, Newest: fpNew, Rule: meshwantsclient.RuleCanary, Words: "the canary ace runs it, not healthy yet"})
asked2, _ := provisions.Encode(provisions.Provision{Provision: "oidc", Consumer: provisions.End{Module: "photos", Node: "ace"},
Provider: provisions.End{Module: "keycloak", Node: "novox"}, State: provisions.Asked, Why: "the provider is applying", At: c.t})
w.Want("want.ace.web", held2, Put, c.t)
modules/mesh-delivery-watchman/internal/watch/mirror.go
+45 -39
View File
@@ -4,13 +4,19 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"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/placement/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// A build ask taken by a node is a move (novox/hq ADR 0525 §4, issues 1887 A8b, 1908, 1909): the node writes the ask's step
// key `<node>.<ask id>` running in the build module's steps (delivery.StepsBucket) when it takes the ask, and writes it passed,
// key `<node>.<ask id>` running in the build module's steps (StepsBucket) when it takes the ask, and writes it passed,
// failed or refused when the ask ends. The pulse of the move is the pulse of the bucket named as the seat's stream
// (delivery.AskStream), under the ask's id, and it begins when the node started the step.
// (buildstep.AskStream), under the ask's id, and it begins when the node started the step.
//
// **An ask still waiting in the stream is a backlog, not a stall** (issue 1908): the rebuild after the switch puts about 170
// asks on a few nodes' slots, and most wait for hours with every worker healthy. No pulse is made for a waiting ask, so none
@@ -22,12 +28,12 @@ import (
// stand, then each change). A running step makes the pulse of its ask, from the step's own start; a step that ended, or a
// key deleted, ends it. The node's `paused.<node>` key, a key that is no step key and a value that cannot be read are left.
func (w *Watchman) Step(language, key string, value []byte, op Op) {
bucket, err := delivery.AskStream(language)
bucket, err := buildstep.AskStream(language)
if err != nil {
w.say("no pulse for the step %s: %v", key, err)
return
}
node, id, err := delivery.ParseStepKey(key)
node, id, err := buildstep.ParseStepKey(key)
if err != nil {
return
}
@@ -41,12 +47,12 @@ func (w *Watchman) Step(language, key string, value []byte, op Op) {
return
}
w.stands(bucket, id)
step, err := delivery.DecodeStep(value)
step, err := buildstep.DecodeStep(value)
if err != nil {
w.say("the step %s cannot be read, and is left: %v", key, err)
return
}
if step.State != delivery.StepRunning {
if step.State != buildstep.StepRunning {
w.moved(bucket, id)
return
}
@@ -67,7 +73,7 @@ func (w *Watchman) Step(language, key string, value []byte, op Op) {
// Want is a change of a key of the assigner's wants 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) Want(key string, value []byte, op Op, at time.Time) {
node, module, err := delivery.ParseWantKey(key)
node, module, err := meshwantsclient.ParseWantKey(key)
if err != nil {
return
}
@@ -79,11 +85,11 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
case Delete:
delete(w.wants, key)
delete(w.held, key)
w.moved(delivery.AssignerWantsBucket, key)
w.moved(AssignerWantsBucket, key)
return
}
w.stands(delivery.AssignerWantsBucket, key)
want, err := delivery.DecodeWant(value)
w.stands(AssignerWantsBucket, key)
want, err := meshwantsclient.DecodeWant(value)
if err != nil {
w.say("the want %s cannot be read, and is left: %v", key, err)
return
@@ -91,10 +97,10 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
if want.Removed {
delete(w.wants, key)
delete(w.held, key)
w.moved(delivery.AssignerWantsBucket, key)
w.moved(AssignerWantsBucket, key)
return
}
w.wants[key] = want.FP
w.wants[key] = want.Fp
if want.Newest != "" || want.Rule != 0 {
// A want held by a rule (ADR 0603 §8) waits on the rule, not on the node: its wait is the hold's, from the time the
// assigner wrote it, and the node running what the want names does not end it.
@@ -102,27 +108,27 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
at = w.now()
}
w.held[key] = true
w.mirrorKept(delivery.AssignerWantsBucket, key,
describe("the hold on %s on %s at %s (rule %d) to end, bound %s: %s", module, node, shortFP(want.FP), want.Rule, w.stepBound, want.Words),
"held|"+want.FP+"|"+want.Newest, at)
w.mirrorKept(AssignerWantsBucket, key,
describe("the hold on %s on %s at %s (rule %d) to end, bound %s: %s", module, node, shortFP(want.Fp), want.Rule, w.stepBound, want.Words),
"held|"+want.Fp+"|"+want.Newest, at)
return
}
if w.held[key] {
delete(w.held, key)
if pk, err := delivery.PulseKey(delivery.AssignerWantsBucket, key); err == nil {
if pk, err := meshwatchdogsclient.PulseKey(AssignerWantsBucket, key); err == nil {
delete(w.starts, pk)
}
w.stuckMoved(delivery.AssignerWantsBucket, key) // the hold ended: its finding is cleared
w.stuckMoved(AssignerWantsBucket, key) // the hold ended: its finding is cleared
}
if w.have(node, module) == want.FP {
w.moved(delivery.AssignerWantsBucket, key)
if w.have(node, module) == want.Fp {
w.moved(AssignerWantsBucket, key)
return
}
if at.IsZero() {
at = w.now()
}
w.mirrorWithin(w.stepBound, delivery.AssignerWantsBucket, key,
describe("the node agent of %s to run %s at %s", node, module, shortFP(want.FP)), want.FP, at, 0)
w.mirrorWithin(w.stepBound, AssignerWantsBucket, key,
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
@@ -133,7 +139,7 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) {
// 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 := delivery.ParseBuilderBuildKey(key)
module, fp, err := builds.ParseBuilderBuildKey(key)
if err != nil {
return
}
@@ -143,30 +149,30 @@ func (w *Watchman) Build(key string, value []byte, op Op, at time.Time) {
return
}
if op == Delete {
w.moved(delivery.BuilderBuildsBucket, key)
w.moved(BuilderBuildsBucket, key)
return
}
w.stands(delivery.BuilderBuildsBucket, key)
b, err := delivery.DecodeBuilderBuild(value)
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 != delivery.BuildQueued || b.Waits == "" {
w.moved(delivery.BuilderBuildsBucket, key)
if b.State != builds.BuildQueued || b.Waits == "" {
w.moved(BuilderBuildsBucket, key)
return
}
if at.IsZero() {
at = w.now()
}
w.mirrorKept(delivery.BuilderBuildsBucket, key,
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 := delivery.ParseHaveKey(key)
node, module, isNode, err := nodeengineclient.ParseHaveKey(key)
if err != nil {
return
}
@@ -174,9 +180,9 @@ func (w *Watchman) Have(key string, value []byte, op Op) {
defer w.mu.Unlock()
if isNode {
// The node's own record: its tunnel key, whose peers' sight of it is network-unreachable.
var record *delivery.NodeRecord
var record *nodeengine.NodeRecord
if op == Put {
if hn, err := delivery.DecodeHaveNode(value); err == nil {
if hn, err := nodeengineclient.DecodeHaveNode(value); err == nil {
record = hn.Record
}
}
@@ -191,24 +197,24 @@ func (w *Watchman) Have(key string, value []byte, op Op) {
w.haveFindings(key, node, module, nil)
return
}
have, err := delivery.DecodeHave(value)
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.haves[key] = have.Fp
w.haveFindings(key, node, module, &have)
wk, err := delivery.WantKey(node, module)
wk, err := meshwantsclient.WantKey(node, module)
if err != nil {
return
}
if fp, ok := w.wants[wk]; ok && fp == have.FP {
w.moved(delivery.AssignerWantsBucket, wk)
if fp, ok := w.wants[wk]; ok && fp == have.Fp {
w.moved(AssignerWantsBucket, wk)
}
}
func (w *Watchman) have(node, module string) string {
key, err := delivery.HaveKey(node, module)
key, err := nodeengineclient.HaveKey(node, module)
if err != nil {
return ""
}
@@ -216,7 +222,7 @@ func (w *Watchman) have(node, module string) string {
}
func shortFP(fp string) string {
hex := strings.TrimPrefix(fp, delivery.SourcePrefix2)
hex := strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2)
if len(hex) > 12 {
hex = hex[:12]
}
modules/mesh-delivery-watchman/internal/watch/names.go
+40
View File
@@ -0,0 +1,40 @@
package watch
import (
meshwatchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
)
// The holders and buckets the watchman names, taken from its own manifest (module.json: its name, its `reads`, its
// states and the tools it invokes) and the seats' state names; a unit row checks them against it (TestManifestStateAndReads).
// No contract names a holder (novox/hq ADR 0625 §4, ADR 0634 §1).
const (
WatchmanModule = "mesh-delivery-watchman"
AssignerModule = "mesh-delivery-assigner"
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 = meshcatalogclient.Module + "_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"
)
// StepsBucket is a build module's state steps (reads: mesh-build-go.steps, mesh-build-image.steps, mesh-build-node.steps).
func StepsBucket(module string) string { return module + "_steps" }
modules/mesh-delivery-watchman/internal/watch/nodehealth.go
+19 -18
View File
@@ -8,7 +8,8 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"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
@@ -38,7 +39,7 @@ const (
KindMachineUnits = "machine-units"
KindReloginNeeded = "relogin-needed"
KindUsedAsFound = "directory-used-as-found"
KindAccountFallback = delivery.NodeAccountFallbackFinding
KindAccountFallback = FindingNodeAccountFallback
)
// The parts and states a report names (mesh-control internal/link).
@@ -57,7 +58,7 @@ func (w *Watchman) NodeHealth(node string, value []byte, op Op) {
w.mu.Lock()
defer w.mu.Unlock()
if op == Put {
h, err := delivery.DecodeNodeHealth(value)
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
@@ -71,7 +72,7 @@ func (w *Watchman) NodeHealth(node string, value []byte, op Op) {
}
// 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 *delivery.NodeRecord) {
func (w *Watchman) tunnelKey(node string, record *nodeengine.NodeRecord) {
key := ""
if record != nil {
key = record.TunnelKey
@@ -125,7 +126,7 @@ func (w *Watchman) ownsHealthKey(key, node string) bool {
return f.Kind != KindNetworkUnreachable && f.Kind != KindMachineNetwork && f.Kind != KindNetworkRewritten
}
func (w *Watchman) rewrittenOf(m string, p delivery.NetworkPart, all []delivery.NetworkPart) Finding {
func (w *Watchman) rewrittenOf(m string, p nodeengine.NetworkPart, all []nodeengine.NetworkPart) Finding {
writer := ""
if p.Writer != "" {
writer = " (" + p.Writer + ")"
@@ -156,7 +157,7 @@ func (w *Watchman) rewrittenOf(m string, p delivery.NetworkPart, all []delivery.
}
// unitsOf is a machine's failed units no module places, one finding.
func unitsOf(node string, u *delivery.UnitsHealth) (Finding, bool) {
func unitsOf(node string, u *nodeengine.UnitsHealth) (Finding, bool) {
if u == nil || len(u.Failed) == 0 {
return Finding{}, false
}
@@ -202,8 +203,8 @@ func unitsOf(node string, u *delivery.UnitsHealth) (Finding, bool) {
}
// reloginOf is one finding per module whose accounts wait for a new login.
func reloginOf(node string, accounts []delivery.NodeAccount) []Finding {
by := map[string][]delivery.NodeAccount{}
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)
}
@@ -227,8 +228,8 @@ func reloginOf(node string, accounts []delivery.NodeAccount) []Finding {
}
// foundOf is one finding per module with a directory used as found.
func foundOf(node string, dirs []delivery.NodeDirectory) []Finding {
by := map[string][]delivery.NodeDirectory{}
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)
}
@@ -257,7 +258,7 @@ func foundOf(node string, dirs []delivery.NodeDirectory) []Finding {
}
// 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 *delivery.LoginFallback) (Finding, bool) {
func fallbackOf(node string, f *nodeengine.LoginFallback) (Finding, bool) {
if f == nil || f.Node != node {
return Finding{}, false
}
@@ -279,19 +280,19 @@ func orModule(m string) string {
// pointed is one machine's failing part that points at another machine.
type pointed struct {
from string
part delivery.NetworkPart
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) []delivery.NetworkPart {
unhealthy := func(m string) []nodeengine.NetworkPart {
h := w.health[m]
if h.Network == nil {
return nil
}
var out []delivery.NetworkPart
var out []nodeengine.NetworkPart
for _, p := range h.Network.Parts {
if p.State == stateUnhealthy {
out = append(out, p)
@@ -299,7 +300,7 @@ func (w *Watchman) networks() {
}
return out
}
targets := func(m string, p delivery.NetworkPart) ([]string, bool) {
targets := func(m string, p nodeengine.NetworkPart) ([]string, bool) {
var out []string
for _, x := range p.Targets {
if x == "" || x == m {
@@ -329,7 +330,7 @@ func (w *Watchman) networks() {
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: delivery.NetworkPart{Part: partTunnel, State: stateUnhealthy,
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}}})
}
}
@@ -364,8 +365,8 @@ func (w *Watchman) networks() {
if len(parts) == 0 {
continue
}
var own []delivery.NetworkPart
var rewritten *delivery.NetworkPart
var own []nodeengine.NetworkPart
var rewritten *nodeengine.NetworkPart
for i, p := range parts {
if p.Part == partResolvConf {
rewritten = &parts[i]
modules/mesh-delivery-watchman/internal/watch/presence.go
+7 -7
View File
@@ -7,7 +7,7 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
)
// partPrefix is what the watchman keeps presence of among the module users: the parts of delivery.
@@ -21,7 +21,7 @@ func presenceOf(user string) (key, waited string, ok bool) {
return "", "", false
}
if first == "node" {
k, err := delivery.PresenceNodeKey(rest)
k, err := meshwatchdogsclient.PresenceNodeKey(rest)
if err != nil {
return "", "", false
}
@@ -30,7 +30,7 @@ func presenceOf(user string) (key, waited string, ok bool) {
if !strings.HasPrefix(rest, partPrefix) {
return "", "", false
}
k, err := delivery.PresencePartKey(rest, first)
k, err := meshwatchdogsclient.PresencePartKey(rest, first)
if err != nil {
return "", "", false
}
@@ -39,7 +39,7 @@ func presenceOf(user string) (key, waited string, ok bool) {
// waitedOf is the words a presence key's return is waited in.
func waitedOf(key string) string {
node, part, instance, err := delivery.ParsePresenceKey(key)
node, part, instance, err := meshwatchdogsclient.ParsePresenceKey(key)
switch {
case err != nil:
return key + " to come back online"
@@ -87,11 +87,11 @@ func (w *Watchman) settle(key string) {
return
}
w.presence[key] = presence{online: online, since: w.now()}
if node, part, _, err := delivery.ParsePresenceKey(key); err == nil && part == "" && node != "" {
if node, part, _, err := meshwatchdogsclient.ParsePresenceKey(key); err == nil && part == "" && node != "" {
defer w.applyCheck(node) // S21 is judged only while the machine is heard
}
if online {
w.moved(delivery.PresenceBucket, key)
w.moved(PresenceBucket, key)
return
}
w.ensureOffline(key)
@@ -100,7 +100,7 @@ func (w *Watchman) settle(key string) {
// ensureOffline mirrors the return of a node or part that is offline.
func (w *Watchman) ensureOffline(key string) {
p := w.presence[key]
w.mirror(delivery.PresenceBucket, key, waitedOf(key), "", p.since, 0)
w.mirror(PresenceBucket, key, waitedOf(key), "", p.since, 0)
}
// SeedPresence tells the watchman a presence key as it stands in its bucket at start, with the time it was written.
modules/mesh-delivery-watchman/internal/watch/signals.go
+19 -17
View File
@@ -8,7 +8,9 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"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-sdk/go/provisions"
)
@@ -99,7 +101,7 @@ func (w *Watchman) signal(row, id string, ttl time.Duration, due signalDue) {
w.say("no signal %s: %v", pk, err)
return
}
value, err := delivery.EncodePulse(delivery.Pulse{Bucket: signalBucket, Key: strings.TrimPrefix(pk, signalBucket+"."), Owes: string(owes)})
value, err := meshwatchdogsclient.EncodePulse(meshwatchdogsclient.Pulse{Bucket: signalBucket, Key: strings.TrimPrefix(pk, signalBucket+"."), Owes: string(owes)})
if err != nil {
w.say("no signal %s: %v", pk, err)
return
@@ -128,7 +130,7 @@ func (w *Watchman) unsignal(row, id string) {
func (w *Watchman) SeedSignal(key string, value []byte) {
w.mu.Lock()
defer w.mu.Unlock()
if p, err := delivery.DecodePulse(value); err == nil && p.Bucket == signalBucket {
if p, err := meshwatchdogsclient.DecodePulse(value); err == nil && p.Bucket == signalBucket {
var due signalDue
if json.Unmarshal([]byte(p.Owes), &due) == nil && due.Do == dueCreated {
w.busCreated[key] = true
@@ -141,7 +143,7 @@ func (w *Watchman) SeedSignal(key string, value []byte) {
}
func dueOf(value []byte) (signalDue, bool) {
p, err := delivery.DecodePulse(value)
p, err := meshwatchdogsclient.DecodePulse(value)
if err != nil || p.Bucket != signalBucket {
return signalDue{}, false
}
@@ -189,9 +191,9 @@ func (w *Watchman) stillWrong(haveKey, kind string) bool {
}
switch kind {
case KindApplyFailing:
return h.Phase == delivery.PhaseFailed
return h.Phase == nodeengineclient.PhaseFailed
case KindUnhealthy:
return h.Health == delivery.HealthFailed
return h.Health == nodeengineclient.HealthFailed
}
return false
}
@@ -228,7 +230,7 @@ func (w *Watchman) Power(node, state string) {
defer w.mu.Unlock()
away := state == "sleeping" || state == "shutting-down"
w.away[node] = away
key, err := delivery.PresenceNodeKey(node)
key, err := meshwatchdogsclient.PresenceNodeKey(node)
if err != nil {
return
}
@@ -248,9 +250,9 @@ func (w *Watchman) Power(node, state string) {
// stuckFinding is the finding of a stuck key as it stands (ADR 0433 §2, §3, now the watchman's own): a node's presence not
// moved is S1, the machine silent, unless it said it is away; any other key not moved is `delivery.<key>.stuck`; a key
// moved ends its finding.
func (w *Watchman) stuckFinding(sk string, s delivery.Stuck) {
if s.Bucket == delivery.PresenceBucket {
if node, part, _, err := delivery.ParsePresenceKey(s.Key); err == nil && part == "" && node != "" {
func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) {
if s.Bucket == PresenceBucket {
if node, part, _, err := meshwatchdogsclient.ParsePresenceKey(s.Key); err == nil && part == "" && node != "" {
key := FindingKey(ScopeMachine, node, KindSilent)
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] {
@@ -474,12 +476,12 @@ func ToolsNode(subject string) string {
//
// 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 *delivery.Have) {
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 != delivery.PhaseRemoved
w.control[node] = h != nil && h.Phase != nodeengineclient.PhaseRemoved
}
if h == nil || h.Phase == delivery.PhaseRemoved {
if h == nil || h.Phase == nodeengineclient.PhaseRemoved {
delete(w.haveState, key)
delete(w.unhealthy, key)
delete(w.failedSince, key)
@@ -495,7 +497,7 @@ func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Hav
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 == delivery.PhaseFailed {
if h.Phase == nodeengineclient.PhaseFailed {
if _, ok := w.failedSince[key]; !ok {
w.failedSince[key] = now
}
@@ -507,7 +509,7 @@ func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Hav
// 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 == delivery.HealthFailed {
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,
@@ -686,7 +688,7 @@ func (w *Watchman) Provisioned(r ProvReport) {
// heard says the watchman hears the machine now: online, and not said away. S21 is judged only while it is (the
// controller's watchApplyFailing): S1 says silent, and asleep is not lost.
func (w *Watchman) heard(node string) bool {
key, err := delivery.PresenceNodeKey(node)
key, err := meshwatchdogsclient.PresenceNodeKey(node)
if err != nil || w.away[node] {
return false
}
@@ -704,7 +706,7 @@ func (w *Watchman) applyCheck(node string) {
var applied, placed []failure
var since time.Time
for hk, at := range w.failedSince {
n, module, _, err := delivery.ParseHaveKey(hk)
n, module, _, err := nodeengineclient.ParseHaveKey(hk)
if err != nil || n != node {
continue
}
modules/mesh-delivery-watchman/internal/watch/verdicts.go
+13 -13
View File
@@ -3,7 +3,7 @@ package watch
import (
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
)
// KindManifestRefused is a module whose manifest the indexer's judge refused (novox/hq ADR 0578 §1): every reader holds
@@ -12,14 +12,14 @@ const KindManifestRefused = "manifest-refused"
// KindManifestOutgrown is a module whose manifest the indexer accepted before and a stricter judge finds problems in
// (novox/hq ADR 0579 §1): it stays accepted and runs; the warning stands until the manifest is mended.
const KindManifestOutgrown = delivery.ManifestOutgrown
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 = delivery.ManifestWaits
const KindManifestWaits = FindingManifestWaits
// ModuleVerdict is a change of the indexer's value of a module (delivery.ModulesBucket): a value carrying a refusal is a
// 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) {
w.mu.Lock()
@@ -33,17 +33,17 @@ func (w *Watchman) ModuleVerdict(module string, value []byte, op Op) {
w.unfind(waits)
return
}
m, err := delivery.DecodeModule(value)
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 := m.VerdictOf(module)
if v != nil && v.Waits() {
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: delivery.ModulesBucket,
Summary: v.Why(),
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 " +
@@ -53,13 +53,13 @@ func (w *Watchman) ModuleVerdict(module string, value []byte, op Op) {
return
}
w.unfind(waits)
if refused, _ := delivery.TakeManifestVerdict(v); !refused {
if refused, _ := meshcatalogclient.TakeVerdict(v); !refused {
w.unfind(key)
if v == nil || len(v.Warnings) == 0 {
w.unfind(outgrown)
return
}
w.find(Finding{Scope: ScopeModule, ID: module, Kind: KindManifestOutgrown, Severity: Warning, Source: delivery.ModulesBucket,
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",
@@ -69,8 +69,8 @@ func (w *Watchman) ModuleVerdict(module string, value []byte, op Op) {
return
}
w.unfind(outgrown)
w.find(Finding{Scope: ScopeModule, ID: module, Kind: KindManifestRefused, Severity: Warning, Source: delivery.ModulesBucket,
Summary: v.Why(),
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 + ": " +
modules/mesh-delivery-watchman/internal/watch/watch.go
+27 -25
View File
@@ -8,7 +8,9 @@ import (
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"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"
)
// Op is what happened to a watched key.
@@ -59,7 +61,7 @@ type Watchman struct {
counts map[string]int // open connections per presence key
presence map[string]presence
pulses map[string]*pulse // live pulses, by pulse key
stuck map[string]delivery.Stuck
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
@@ -83,7 +85,7 @@ type Watchman struct {
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]delivery.Have
haveState map[string]nodeengine.HaveByNodeModuleValue
unhealthy map[string]int
advised map[string]int
advisedSince map[string]time.Time
@@ -97,7 +99,7 @@ type Watchman struct {
// 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]delivery.NodeHealth
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
@@ -115,11 +117,11 @@ type Watchman struct {
// New is a watchman writing to out, reading the time from now.
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]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: delivery.PulseMoveBound, stepBound: StepBound,
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,
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]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{},
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]delivery.NodeHealth{}, tunnels: map[string]string{},
health: map[string]nodeengine.HealthByNodeValue{}, tunnels: map[string]string{},
standing: map[string]map[string]bool{}, readWhole: map[string]bool{}, awaited: map[string]bool{}}
}
@@ -138,9 +140,9 @@ func (w *Watchman) mirror(bucket, key, waited, head string, since time.Time, rev
w.mirrorWithin(0, bucket, key, waited, head, since, revision)
}
// mirrorWithin is mirror for a move that may take bound (zero: the watchman's own bound, delivery.PulseMoveBound).
// mirrorWithin is mirror for a move that may take bound (zero: the watchman's own bound, meshwatchdogsclient.PulseMoveBound).
func (w *Watchman) mirrorWithin(bound time.Duration, bucket, key, waited, head string, since time.Time, revision uint64) {
pk, err := delivery.PulseKey(bucket, key)
pk, err := meshwatchdogsclient.PulseKey(bucket, key)
if err != nil {
w.say("no pulse for %s %s: %v", bucket, key, err)
return
@@ -161,7 +163,7 @@ func (w *Watchman) mirrorWithin(bound time.Duration, bucket, key, waited, head s
w.writeStuck(bucket, key, waited, since)
return
}
value, err := delivery.EncodePulse(delivery.Pulse{Bucket: bucket, Key: key, Revision: revision, Owes: waited})
value, err := meshwatchdogsclient.EncodePulse(meshwatchdogsclient.Pulse{Bucket: bucket, Key: key, Revision: int64(revision), Owes: waited})
if err != nil {
w.say("no pulse for %s %s: %v", bucket, key, err)
return
@@ -183,7 +185,7 @@ type startMark struct {
// (a provision asked, a want held at the same fingerprints, a build waiting for the same package), a rewrite of the key (new
// words, a new time) does not start the bound again; the earliest start is kept. A different mark is a new wait.
func (w *Watchman) mirrorKept(bucket, key, waited, mark string, since time.Time) {
if pk, err := delivery.PulseKey(bucket, key); err == nil {
if pk, err := meshwatchdogsclient.PulseKey(bucket, key); err == nil {
if cur, ok := w.starts[pk]; ok && cur.mark == mark && cur.since.Before(since) {
since = cur.since
}
@@ -202,7 +204,7 @@ func (w *Watchman) deletePulse(pk string) {
// moved is the mirrored key moving: its pulse is deleted, so it never expires, and a stuck key it left is written as
// moved (never deleted).
func (w *Watchman) moved(bucket, key string) {
pk, err := delivery.PulseKey(bucket, key)
pk, err := meshwatchdogsclient.PulseKey(bucket, key)
if err != nil {
return
}
@@ -215,7 +217,7 @@ func (w *Watchman) moved(bucket, key string) {
// stuckMoved writes a stuck key of the key as moved, if it stands and is not yet.
func (w *Watchman) stuckMoved(bucket, key string) {
sk, err := delivery.StuckKey(bucket, key)
sk, err := meshwatchdogsclient.StuckKey(bucket, key)
if err != nil {
return
}
@@ -228,7 +230,7 @@ func (w *Watchman) stuckMoved(bucket, key string) {
}
func (w *Watchman) writeStuck(bucket, key, waited string, since time.Time) {
sk, err := delivery.StuckKey(bucket, key)
sk, err := meshwatchdogsclient.StuckKey(bucket, key)
if err != nil {
w.say("no stuck key for %s %s: %v", bucket, key, err)
return
@@ -236,11 +238,11 @@ func (w *Watchman) writeStuck(bucket, key, waited string, since time.Time) {
if cur, ok := w.stuck[sk]; ok && cur.Since.Equal(since) {
return
}
w.putStuck(sk, delivery.Stuck{Bucket: bucket, Key: key, Waited: waited, Since: since})
w.putStuck(sk, meshwatchdogsclient.Stuck{Bucket: bucket, Key: key, Waited: waited, Since: since})
}
func (w *Watchman) putStuck(sk string, s delivery.Stuck) {
value, err := delivery.EncodeStuck(s)
func (w *Watchman) putStuck(sk string, s meshwatchdogsclient.Stuck) {
value, err := meshwatchdogsclient.EncodeStuck(s)
if err != nil {
w.say("cannot encode the stuck key %s: %v", sk, err)
return
@@ -249,7 +251,7 @@ func (w *Watchman) putStuck(sk string, s delivery.Stuck) {
w.say("cannot write the stuck key %s: %v", sk, err)
return
}
s.V = delivery.Version
s.V = keys.Version
w.stuck[sk] = s
w.stuckFinding(sk, s)
}
@@ -278,7 +280,7 @@ func (w *Watchman) Pulse(key string, op Op, at time.Time) {
func (w *Watchman) SeedStuck(key string, value []byte) {
w.mu.Lock()
defer w.mu.Unlock()
if s, err := delivery.DecodeStuck(value); err == nil {
if s, err := meshwatchdogsclient.DecodeStuck(value); err == nil {
w.stuck[key] = s
w.stuckFinding(key, s)
}
@@ -289,7 +291,7 @@ func (w *Watchman) SeedStuck(key string, value []byte) {
func (w *Watchman) SeedPulse(key string) {
w.mu.Lock()
defer w.mu.Unlock()
bucket, k, err := delivery.ParsePulseKey(key)
bucket, k, err := meshwatchdogsclient.ParsePulseKey(key)
if err != nil || bucket == signalBucket {
return
}
@@ -351,7 +353,7 @@ func (w *Watchman) closeGone() {
w.standing, w.readWhole, w.reconciled = standing, map[string]bool{}, true
}
func (w *Watchman) closeStuck(sk string, s delivery.Stuck) {
func (w *Watchman) closeStuck(sk string, s meshwatchdogsclient.Stuck) {
s.Moved = true
w.say("the stuck key %s is closed: its key %s is no longer in %s (it aged out of the bucket or was deleted while nobody heard), so nothing waits for it since %s",
sk, s.Key, s.Bucket, s.Since.UTC().Format(time.RFC3339))
@@ -444,7 +446,7 @@ func (w *Watchman) SetStepBound(d time.Duration) {
w.stepBound = d
}
// SetBound sets how long a move, and a running step, may take; it is delivery.PulseMoveBound unless a test shortens it.
// SetBound sets how long a move, and a running step, may take; it is meshwatchdogsclient.PulseMoveBound unless a test shortens it.
func (w *Watchman) SetBound(d time.Duration) {
w.mu.Lock()
defer w.mu.Unlock()
@@ -452,10 +454,10 @@ func (w *Watchman) SetBound(d time.Duration) {
}
// ttl is the time to live of a pulse for a move of the given age, and whether the move is stalled already
// (delivery.PulseTTL, with the bound this watchman keeps).
// (meshwatchdogsclient.PulseTTL, with the bound this watchman keeps).
func (w *Watchman) ttl(age time.Duration) (time.Duration, bool) {
if w.bound == delivery.PulseMoveBound {
return delivery.PulseTTL(age)
if w.bound == meshwatchdogsclient.PulseMoveBound {
return meshwatchdogsclient.PulseTTL(age)
}
if age >= w.bound {
return 0, true
modules/mesh-delivery-watchman/internal/watch/watch_test.go
+31 -26
View File
@@ -8,23 +8,28 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"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/placement/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
type fakeOut struct {
presence map[string]bool
presenceWrite int
pulses map[string]delivery.Pulse
pulses map[string]meshwatchdogsclient.Pulse
ttls map[string]time.Duration
pulseDeletes []string
stuck map[string]delivery.Stuck
stuck map[string]meshwatchdogsclient.Stuck
stuckWrites int
findings map[string]Finding
findingWrites int
}
func newOut() *fakeOut {
return &fakeOut{presence: map[string]bool{}, pulses: map[string]delivery.Pulse{}, ttls: map[string]time.Duration{}, stuck: map[string]delivery.Stuck{},
return &fakeOut{presence: map[string]bool{}, pulses: map[string]meshwatchdogsclient.Pulse{}, ttls: map[string]time.Duration{}, stuck: map[string]meshwatchdogsclient.Stuck{},
findings: map[string]Finding{}}
}
@@ -53,7 +58,7 @@ func (o *fakeOut) PutPresence(key string, online bool) error {
}
func (o *fakeOut) PutPulse(key string, value []byte, ttl time.Duration) error {
p, err := delivery.DecodePulse(value)
p, err := meshwatchdogsclient.DecodePulse(value)
if err != nil {
return err
}
@@ -68,7 +73,7 @@ func (o *fakeOut) DeletePulse(key string) error {
}
func (o *fakeOut) PutStuck(key string, value []byte) error {
s, err := delivery.DecodeStuck(value)
s, err := meshwatchdogsclient.DecodeStuck(value)
if err != nil {
return err
}
@@ -86,7 +91,7 @@ func (c *clock) now() time.Time { return c.t }
func setup() (*Watchman, *fakeOut, *clock) {
o, c := newOut(), &clock{t: t0}
w := New(o, c.now)
w.SetStepBound(delivery.PulseMoveBound) // the older tests move a step as they move any key: stuck at the pulse's bound
w.SetStepBound(meshwatchdogsclient.PulseMoveBound) // the older tests move a step as they move any key: stuck at the pulse's bound
return w, o, c
}
@@ -99,12 +104,12 @@ func checkValue(started time.Time, head string) []byte {
const (
checksKey = "SEAT_NODE_BUILD_GO"
presenceNS = delivery.PresenceBucket
presenceNS = PresenceBucket
)
// runningStep is a build module's step key as a node writes it when it takes an ask: the ask's id, running, since started.
func runningStep(id string, started time.Time) []byte {
b, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: delivery.StepRunning, Started: started})
b, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: buildstep.StepRunning, Started: started})
if err != nil {
panic(err)
}
@@ -112,7 +117,7 @@ func runningStep(id string, started time.Time) []byte {
}
func endedStep(id, state string, started time.Time) []byte {
b, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: started, Ended: started.Add(time.Second)})
b, err := buildstep.EncodeStep(buildstep.Step{ID: id, Tool: "build_go_test", Node: "n1", State: state, Started: started, Ended: started.Add(time.Second)})
if err != nil {
panic(err)
}
@@ -121,7 +126,7 @@ func endedStep(id, state string, started time.Time) []byte {
func pulseKeyOf(t *testing.T, bucket, key string) string {
t.Helper()
k, err := delivery.PulseKey(bucket, key)
k, err := meshwatchdogsclient.PulseKey(bucket, key)
if err != nil {
t.Fatal(err)
}
@@ -130,7 +135,7 @@ func pulseKeyOf(t *testing.T, bucket, key string) string {
func stuckKeyOf(t *testing.T, bucket, key string) string {
t.Helper()
k, err := delivery.StuckKey(bucket, key)
k, err := meshwatchdogsclient.StuckKey(bucket, key)
if err != nil {
t.Fatal(err)
}
@@ -501,7 +506,7 @@ func TestAnAskTakenByANodeHasAPulseFromWhenItStartedAndEndsWithTheStep(t *testin
if len(o.pulses) != 1 {
t.Fatalf("pulses = %v, want the taken ask's", o.pulses)
}
for _, state := range []string{delivery.StepPassed, delivery.StepFailed, delivery.StepRefused} {
for _, state := range []string{buildstep.StepPassed, buildstep.StepFailed, buildstep.StepRefused} {
w2, o2, c2 := setup()
c2.t = t0.Add(10 * time.Second)
w2.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
@@ -573,11 +578,11 @@ func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908
// its fingerprint.
func wantValue(t *testing.T, fp string, removed bool) []byte {
t.Helper()
w := delivery.Want{FP: fp, Composed: "sha256:" + strings.Repeat("c", 64), Removed: removed}
w := meshwantsclient.Want{Fp: fp, Composed: "sha256:" + strings.Repeat("c", 64), Removed: removed}
if removed {
w.Composed = ""
}
b, err := delivery.EncodeWant(w)
b, err := meshwantsclient.EncodeWant(w)
if err != nil {
t.Fatal(err)
}
@@ -586,7 +591,7 @@ func wantValue(t *testing.T, fp string, removed bool) []byte {
func haveValue(t *testing.T, fp string) []byte {
t.Helper()
b, err := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
b, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
if err != nil {
t.Fatal(err)
}
@@ -594,15 +599,15 @@ func haveValue(t *testing.T, fp string) []byte {
}
var (
fpOld = delivery.SourcePrefix2 + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
fpNew = delivery.SourcePrefix2 + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
fpOld = meshcatalogclient.SourcePrefix2 + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
fpNew = meshcatalogclient.SourcePrefix2 + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
)
func TestAWantIsAPulseOwedByTheNodeAndTheHaveNamingItEndsIt(t *testing.T) {
w, o, c := setup()
c.t = t0.Add(10 * time.Second)
w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
pk := pulseKeyOf(t, delivery.AssignerWantsBucket, "want.ace.web")
pk := pulseKeyOf(t, AssignerWantsBucket, "want.ace.web")
p, ok := o.pulses[pk]
if !ok || p.Key != "want.ace.web" || !strings.Contains(p.Owes, "ace") || !strings.Contains(p.Owes, "web") {
t.Fatalf("pulses = %v", o.pulses)
@@ -642,10 +647,10 @@ func TestAWantNotAppliedByItsBoundIsStuckAndAKeyDeletedEndsIt(t *testing.T) {
o, c := newOut(), &clock{t: t0}
w := New(o, c.now)
w.Want("want.ace.web", wantValue(t, fpNew, false), Put, t0)
pk := pulseKeyOf(t, delivery.AssignerWantsBucket, "want.ace.web")
pk := pulseKeyOf(t, AssignerWantsBucket, "want.ace.web")
c.t = t0.Add(31 * time.Minute)
w.Pulse(pk, Expired, c.t)
if s, ok := o.stuck[stuckKeyOf(t, delivery.AssignerWantsBucket, "want.ace.web")]; !ok || s.Moved || !s.Since.Equal(t0) {
if s, ok := o.stuck[stuckKeyOf(t, AssignerWantsBucket, "want.ace.web")]; !ok || s.Moved || !s.Since.Equal(t0) {
t.Fatalf("stuck = %v", o.stuck)
}
w2, o2, c2 := setup()
@@ -673,18 +678,18 @@ func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) {
{"an ask's step aged out of the steps bucket", checksKey, "askold1", func(*Watchman) {}, true, true, false},
{"an ask's step still running", checksKey, "askold1", func(w *Watchman) { w.Step("go", "n1.askold1", runningStep("askold1", t0), Put) }, true, false, false},
{"the steps bucket could not be read", checksKey, "askold1", func(*Watchman) {}, false, false, false},
{"a want deleted while nobody heard", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, false},
{"a want still standing", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, false},
{"a want deleted while nobody heard", AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, false},
{"a want still standing", AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, false},
// DLV-M5 (ADR 0615): the wants are read from the assigner's first view, which may come after the start's end; what
// they hold waits for it, and is then reconciled as at the start.
{"a want deleted while nobody heard, read from a late first view", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, true},
{"a want still standing, read from a late first view", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, true},
{"a want deleted while nobody heard, read from a late first view", AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true, true},
{"a want still standing, read from a late first view", AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false, true},
} {
t.Run(row.name, func(t *testing.T) {
w, o, c := setup()
c.t = t0.Add(9 * 24 * time.Hour)
sk := stuckKeyOf(t, row.bucket, row.key)
v, _ := delivery.EncodeStuck(delivery.Stuck{Bucket: row.bucket, Key: row.key, Waited: "a move", Since: t0})
v, _ := meshwatchdogsclient.EncodeStuck(meshwatchdogsclient.Stuck{Bucket: row.bucket, Key: row.key, Waited: "a move", Since: t0})
w.SeedStuck(sk, v)
if _, ok := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)]; !ok {
t.Fatal("the seeded stuck key raised no finding")
modules/mesh-delivery-watchman/manifest_test.go
+38
View File
@@ -10,6 +10,10 @@ import (
"slices"
"strings"
"testing"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"mesh-delivery-watchman/internal/watch"
)
// The controller's manifest reader is mesh-controller internal/catalogue (ParseManifest), an internal package of another
@@ -130,6 +134,40 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6
if m.SecretsOwner != "mesh-delivery-watchman" {
t.Errorf("secrets-owner = %q", m.SecretsOwner)
}
// D-11 (ADR 0634 §1): the holders and buckets the watchman names are its manifest's: a state of the seat it holds, a
// `reads` entry, a `consumes` views family or an `invokes` tool, each as "<module>_<state>".
// bucket is "<module>_<state>" of a manifest entry "<module>.<state>" or "<module>.views.<state>".
bucket := func(list []string, entry string) string {
if !slices.Contains(list, entry) {
return "(not in the manifest: " + entry + ")"
}
parts := strings.Split(entry, ".")
return parts[0] + "_" + parts[len(parts)-1]
}
entry := func(list []string, e string) string {
if !slices.Contains(list, e) {
return "(not in the manifest: " + e + ")"
}
return e
}
for name, row := range map[string]struct{ got, want string }{
"module": {watch.WatchmanModule, m.Module},
"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")},
"steps": {watch.StepsBucket("mesh-build-go"), bucket(m.Reads, "mesh-build-go.steps")},
"wants": {watch.AssignerWantsBucket, bucket(m.Consumes, "mesh-delivery-assigner.views.wants")},
"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": {meshcatalogclient.Module + ".indexer_pending", entry(m.Invokes, "mesh-delivery-indexer.indexer_pending")},
} {
if row.got != row.want {
t.Errorf("the name for %s is %q, its manifest's is %q", name, row.got, row.want)
}
}
}
// The manifest is whole in its repository: the program, and the tools bundle the runtime loads (1952/WM1).