From 88d100b9d5c56cb7a273d006b0bb8da8a7b4ef8b Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 16:28:51 +0100 Subject: [PATCH] =?UTF-8?q?Watchman:=20the=20node-engine's,=20wants',=20bu?= =?UTF-8?q?ild=20steps',=20builder's=20and=20catalog's=20names=20come=20fr?= =?UTF-8?q?om=20their=20clients=20and=20seats,=20the=20watchdogs'=20codec?= =?UTF-8?q?=20from=20meshwatchdogsclient;=20the=20holders=20and=20buckets?= =?UTF-8?q?=20it=20names=20are=20its=20own=20constants=20checked=20against?= =?UTF-8?q?=20its=20manifest=20(D-11;=20ADR=200634=20=C2=A71,=20=C2=A72,?= =?UTF-8?q?=20=C2=A74;=20issue=201966)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-watchman/consumes.go | 3 +- .../integration_test.go | 131 +++++++++--------- .../cmd/mesh-delivery-watchman/main.go | 56 ++++---- .../cmd/mesh-delivery-watchman/main_test.go | 7 +- modules/mesh-delivery-watchman/go.mod | 7 + .../internal/watch/busstate.go | 7 +- .../internal/watch/findings_test.go | 103 +++++++------- .../internal/watch/mirror.go | 84 +++++------ .../internal/watch/names.go | 40 ++++++ .../internal/watch/nodehealth.go | 37 ++--- .../internal/watch/presence.go | 14 +- .../internal/watch/signals.go | 36 ++--- .../internal/watch/verdicts.go | 26 ++-- .../internal/watch/watch.go | 52 +++---- .../internal/watch/watch_test.go | 57 ++++---- .../mesh-delivery-watchman/manifest_test.go | 38 +++++ 16 files changed, 405 insertions(+), 293 deletions(-) create mode 100644 modules/mesh-delivery-watchman/internal/watch/names.go diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go index cbd06fde..a154a11e 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -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 diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index 86d96c09..ca458306 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -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) } } diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index f522d41e..b00534cb 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -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 `.` 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. diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index c96c4f0d..7e8ab4b3 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -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) } diff --git a/modules/mesh-delivery-watchman/go.mod b/modules/mesh-delivery-watchman/go.mod index defa3868..2899b58e 100644 --- a/modules/mesh-delivery-watchman/go.mod +++ b/modules/mesh-delivery-watchman/go.mod @@ -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 diff --git a/modules/mesh-delivery-watchman/internal/watch/busstate.go b/modules/mesh-delivery-watchman/internal/watch/busstate.go index 8828c3e5..8b1b66b4 100644 --- a/modules/mesh-delivery-watchman/internal/watch/busstate.go +++ b/modules/mesh-delivery-watchman/internal/watch/busstate.go @@ -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) } diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 5662391c..789ca57a 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -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) diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index 76fd2cf5..532583a5 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -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 `.` running in the build module's steps (delivery.StepsBucket) when it takes the ask, and writes it passed, +// key `.` 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.` 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] } diff --git a/modules/mesh-delivery-watchman/internal/watch/names.go b/modules/mesh-delivery-watchman/internal/watch/names.go new file mode 100644 index 00000000..8d547735 --- /dev/null +++ b/modules/mesh-delivery-watchman/internal/watch/names.go @@ -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" } diff --git a/modules/mesh-delivery-watchman/internal/watch/nodehealth.go b/modules/mesh-delivery-watchman/internal/watch/nodehealth.go index eda9df49..a53a3d8a 100644 --- a/modules/mesh-delivery-watchman/internal/watch/nodehealth.go +++ b/modules/mesh-delivery-watchman/internal/watch/nodehealth.go @@ -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] diff --git a/modules/mesh-delivery-watchman/internal/watch/presence.go b/modules/mesh-delivery-watchman/internal/watch/presence.go index 369e73c4..5fe16e07 100644 --- a/modules/mesh-delivery-watchman/internal/watch/presence.go +++ b/modules/mesh-delivery-watchman/internal/watch/presence.go @@ -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. diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index 3874b062..1faad419 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -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..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 } diff --git a/modules/mesh-delivery-watchman/internal/watch/verdicts.go b/modules/mesh-delivery-watchman/internal/watch/verdicts.go index b31b648e..905db480 100644 --- a/modules/mesh-delivery-watchman/internal/watch/verdicts.go +++ b/modules/mesh-delivery-watchman/internal/watch/verdicts.go @@ -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 + ": " + diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 1b43cbd2..e73bc987 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -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 `.`, 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 diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 29c2759a..1fd555ea 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -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") diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 0724860c..cff47de1 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -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 "_". + // bucket is "_" of a manifest entry "." or ".views.". + 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).