diff --git a/cmd/mesh-controller/apply_failing.go b/cmd/mesh-controller/apply_failing.go index 986207ce..5f76ef80 100644 --- a/cmd/mesh-controller/apply_failing.go +++ b/cmd/mesh-controller/apply_failing.go @@ -33,7 +33,7 @@ const packageManagerKey = "the package manager (" func watchApplyFailing(f *signalFacts) []conditions.Observation { heard := map[string]bool{} for _, m := range f.machines { - if !m.asleep() && !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) <= heartbeatBound(m.every) { + if !m.asleep() && !m.lastHeard.IsZero() && !m.silent(f.now) { heard[m.name] = true } } diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index 639d6f81..eb50a56e 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -120,6 +120,11 @@ var probeRegistry = []probe{ {ID: probeDeliveriesID, Asserts: "no delivery is held past its state's bound unsaid: mesh-delivery's " + "`stalled`, each with the transition its table lets healer H2 take", From: "ADR 0239", Kind: kindDeliveryStalled, Phase: 3, run: probeDeliveries}, + // What the watchman says stopped moving (ADR 0433 §2, issue 1156): each key of its stuck bucket not moved, but a + // node's presence, which the silent check takes. + {ID: probeStuckID, Asserts: "no key the watchman says stopped moving is left unsaid: each key of its stuck bucket " + + "not moved is a condition of the delivery scope until it says moved", From: "ADR 0433 §2, issue 1156", + Kind: kindStuck, Phase: 3, run: probeStuck}, // The delivery budgets (novox/hq ADR 0282 decision 7): the newest delivery of each class within its budget, // read from mesh-delivery's `times`; a measurement said, never a delivery held. {ID: probeBudgetsID, Asserts: "the newest delivery of a leaf module ran on every machine within three minutes of " + diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index dc308be9..f07fa940 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -610,6 +610,13 @@ var plainWordings = map[string]func(conditions.Observation) words{ "did not answer in ten minutes, so nothing of the delivery is built or sent.", Resolved: "Resolved: the backup is done and the delivery goes on"} }), + kindStuck: worded(func(o conditions.Observation) words { + return words{Headline: "Something in a delivery has stopped moving", + Explanation: "The watchman noticed a part of the delivery that did not move when it should have: a build " + + "asked and not answered, a check that stopped, or a part that went offline. Nothing is lost, and it is " + + "said again until it moves.", + Resolved: "Resolved: it moves again"} + }), kindSendKeyRefused: worded(func(o conditions.Observation) words { return words{Headline: "A delivery cannot watch a machine for its report", Explanation: "The bus refused the note that makes the mesh notice a machine that never reports on a send. " + diff --git a/cmd/mesh-controller/probes.go b/cmd/mesh-controller/probes.go index deafc4e9..3c22c7ef 100644 --- a/cmd/mesh-controller/probes.go +++ b/cmd/mesh-controller/probes.go @@ -629,7 +629,7 @@ func heardMachines(d *doctor) map[string]bool { return out } for _, m := range f.machines { - if !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) <= heartbeatBound(m.every) && !m.asleep() { + if !m.lastHeard.IsZero() && !m.silent(f.now) && !m.asleep() { out[m.name] = true } } diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index e086e5d9..43ac6d41 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -335,7 +335,7 @@ func watchHeartbeats(f *signalFacts) []conditions.Observation { } bound := heartbeatBound(m.every) silent := f.now.Sub(m.lastHeard) - if silent <= bound { + if !m.silent(f.now) { continue } severity := conditions.Warning @@ -357,7 +357,7 @@ func watchReports(f *signalFacts) []conditions.Observation { if m.sentAt.IsZero() || m.reportedCurrent || m.asleep() { continue } - if !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) > heartbeatBound(m.every) { + if m.silent(f.now) { continue // silent: S1 says it, and a silent machine reports nothing } bound := max(reportAtLeast, 3*m.lastApply) diff --git a/cmd/mesh-controller/stuck_probe.go b/cmd/mesh-controller/stuck_probe.go new file mode 100644 index 00000000..ec8eeb29 --- /dev/null +++ b/cmd/mesh-controller/stuck_probe.go @@ -0,0 +1,208 @@ +package main + +import ( + "context" + "errors" + "fmt" + "sort" + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// What stopped moving (novox/hq ADR 0433 §2, §3, issue 1156). The watchman writes `stuck..` in its own +// bucket for each key whose pulse expired, and again as moved when the key moves; it never deletes one. The controller +// reads that bucket in two places and holds no timer for either: a probe of the self-check raises each key not moved as +// a condition of the delivery scope, and the check that a machine is silent takes a node's presence key as its verdict. +const ( + kindStuck = "stuck" + probeStuckID = "DM" +) + +// presenceVerdict is the watchman's word on whether a machine is online, from its stuck presence key. +type presenceVerdict int + +const ( + // presenceUnsaid: the watchman has written nothing for the machine; the heartbeat rule judges it, as before. + presenceUnsaid presenceVerdict = iota + // presenceHeard: the key says moved, the machine is back online. + presenceHeard + // presenceSilent: the key says not moved, the machine was offline past its bound. + presenceSilent +) + +// applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none) +// and, when presence. has an entry, its time as when the machine was last heard from. Without a presence entry, +// last heard stays the store's last_seen. +func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, at time.Time) { + switch { + case stuck == nil: + m.presence = presenceUnsaid + case stuck.Moved: + m.presence = presenceHeard + default: + m.presence = presenceSilent + } + if !at.IsZero() { + m.lastHeard = at + } +} + +// silent says the machine is silent: the watchman's verdict where it gave one, else the heartbeat rule. A machine never +// heard from is not silent, it has not joined. +func (m machineFacts) silent(now time.Time) bool { + switch m.presence { + case presenceSilent: + return true + case presenceHeard: + return false + } + return !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > heartbeatBound(m.every) +} + +// stuckObservations are the conditions of what stopped moving: one for each key not moved, in the delivery scope, kind +// stuck, naming the key and what it waited for. A node's presence key is left to the silent check; a moved key raises +// nothing, so its condition clears. +func stuckObservations(entries map[string]sdkdelivery.Stuck) []conditions.Observation { + keys := make([]string, 0, len(entries)) + for k := range entries { + keys = append(keys, k) + } + sort.Strings(keys) + var out []conditions.Observation + for _, k := range keys { + s := entries[k] + if s.Moved { + continue + } + if s.Bucket == sdkdelivery.PresenceBucket { + if node, _, _, err := sdkdelivery.ParsePresenceKey(s.Key); err == nil && node != "" { + continue + } + } + waited := s.Waited + if waited == "" { + waited = "a move" + } + out = append(out, conditions.Observation{Scope: conditions.ScopeDelivery, ID: k, Kind: kindStuck, + Severity: conditions.Warning, + Summary: fmt.Sprintf("%s has not moved since %s: it waited for %s", k, s.Since.UTC().Format("2006-01-02 15:04 MST"), + waited), + Said: fmt.Sprintf("%s: waiting for %s since %s", k, waited, s.Since.UTC().Format(time.RFC3339))}) + } + return out +} + +// readStuck is every key of the stuck bucket and its value; none when the bucket is not there yet (the watchman not +// delivered). A value of another version is left out, said. +func readStuck(ctx context.Context, conn *nats.Conn) (map[string]sdkdelivery.Stuck, error) { + api, err := jetstream.New(conn) + if err != nil { + return nil, err + } + kv, err := api.KeyValue(ctx, sdkdelivery.StuckBucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("the watchman's stuck bucket: %w", err) + } + keys, err := boundedCall(ctx, "listing the stuck keys", func(ctx context.Context) ([]string, error) { return kv.Keys(ctx) }) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + if err != nil { + return nil, err + } + out := map[string]sdkdelivery.Stuck{} + for _, key := range keys { + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + if err != nil { + return nil, err + } + s, err := sdkdelivery.DecodeStuck(e.Value()) + if err != nil { + fmt.Printf("the stuck key %s is not read: %v\n", key, err) + continue + } + out[key] = s + } + return out, nil +} + +// readPresenceTimes is, by node, the time of presence.'s entry; a node without an entry is left out. +func readPresenceTimes(ctx context.Context, conn *nats.Conn, nodes []string) (map[string]time.Time, error) { + api, err := jetstream.New(conn) + if err != nil { + return nil, err + } + kv, err := api.KeyValue(ctx, sdkdelivery.PresenceBucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("the watchman's presence bucket: %w", err) + } + out := map[string]time.Time{} + for _, node := range nodes { + key, err := sdkdelivery.PresenceNodeKey(node) + if err != nil { + continue + } + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + if err != nil { + return nil, err + } + out[node] = e.Created() + } + return out, nil +} + +// withPresence says of each machine what the watchman says: its stuck presence key and its presence entry's time. +func withPresence(ctx context.Context, conn *nats.Conn, machines []machineFacts) error { + stuck, err := readStuck(ctx, conn) + if err != nil { + return err + } + names := make([]string, len(machines)) + for i, m := range machines { + names[i] = m.name + } + times, err := readPresenceTimes(ctx, conn, names) + if err != nil { + return err + } + for i := range machines { + var said *sdkdelivery.Stuck + if key, err := sdkdelivery.StuckKey(sdkdelivery.PresenceBucket, machines[i].name); err == nil { + if s, ok := stuck[key]; ok { + said = &s + } + } + applyPresence(&machines[i], said, times[machines[i].name]) + } + return nil +} + +// probeStuck is DM: every key the watchman says stopped moving is raised, but a node's presence, which the silent check +// takes. +func probeStuck(ctx context.Context, d *doctor) ([]conditions.Observation, error) { + if d.js == nil { + return nil, nil + } + entries, err := readStuck(ctx, d.js.Conn()) + if err != nil { + return nil, err + } + return stuckObservations(entries), nil +} diff --git a/cmd/mesh-controller/stuck_probe_integration_test.go b/cmd/mesh-controller/stuck_probe_integration_test.go new file mode 100644 index 00000000..a3e9bcac --- /dev/null +++ b/cmd/mesh-controller/stuck_probe_integration_test.go @@ -0,0 +1,82 @@ +//go:build integration + +package main + +import ( + "context" + "testing" + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/testbus" +) + +// 1156/AC4: on the embedded bus with the watchman's buckets, a stuck key written raises a condition and written as moved +// clears it; a node's presence key not moved says the machine silent through the facts, and moved says it heard. +func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { // 1156/AC4 + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if _, err := broker.RaiseBuckets(js, []broker.Bucket{ + {Module: sdkdelivery.WatchmanModule, Name: "stuck"}, + {Module: sdkdelivery.WatchmanModule, Name: "presence", History: sdkdelivery.PresenceHistory}, + }); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + kv, err := api.KeyValue(ctx, sdkdelivery.StuckBucket) + if err != nil { + t.Fatal(err) + } + write := func(bucket, key string, moved bool) string { + t.Helper() + stuckKey, err := sdkdelivery.StuckKey(bucket, key) + if err != nil { + t.Fatal(err) + } + body, err := sdkdelivery.EncodeStuck(sdkdelivery.Stuck{Bucket: bucket, Key: key, Waited: "an answer", Since: time.Now().UTC(), Moved: moved}) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, stuckKey, body); err != nil { + t.Fatal(err) + } + return stuckKey + } + d := &doctor{js: js} + stuckKey := write(sdkdelivery.PulseBucket, "builds.x", false) + write(sdkdelivery.PresenceBucket, "ace", false) + got, err := probeStuck(ctx, d) + if err != nil || len(got) != 1 || got[0].ID != stuckKey || got[0].Kind != kindStuck { + t.Fatalf("a stuck key written: %+v %v", got, err) + } + machines := []machineFacts{{name: "ace", lastHeard: time.Now()}} + if err := withPresence(ctx, js.Conn(), machines); err != nil { + t.Fatal(err) + } + if !machines[0].silent(time.Now()) { + t.Fatal("a presence key not moved did not say the machine silent") + } + write(sdkdelivery.PulseBucket, "builds.x", true) + write(sdkdelivery.PresenceBucket, "ace", true) + if got, err := probeStuck(ctx, d); err != nil || len(got) != 0 { + t.Fatalf("a stuck key written as moved: %+v %v", got, err) + } + machines = []machineFacts{{name: "ace", lastHeard: time.Now().Add(-time.Hour), every: time.Minute}} + if err := withPresence(ctx, js.Conn(), machines); err != nil { + t.Fatal(err) + } + if machines[0].silent(time.Now()) { + t.Fatal("a presence key that moved still says the machine silent") + } +} diff --git a/cmd/mesh-controller/stuck_probe_test.go b/cmd/mesh-controller/stuck_probe_test.go new file mode 100644 index 00000000..d3b258c2 --- /dev/null +++ b/cmd/mesh-controller/stuck_probe_test.go @@ -0,0 +1,108 @@ +package main + +import ( + "strings" + "testing" + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +func aStuckKey(t *testing.T, bucket, key string) string { + t.Helper() + k, err := sdkdelivery.StuckKey(bucket, key) + if err != nil { + t.Fatal(err) + } + return k +} + +// 1156/AC1: each not-moved stuck key that is not a node's presence is an observation of the delivery scope, kind stuck, +// naming the key and what it waited for; a moved key raises nothing, so its condition clears. +func TestEachStuckKeyThatHasNotMovedIsAConditionOfTheDeliveryScope(t *testing.T) { // 1156/AC1 + since := time.Date(2026, 10, 18, 19, 0, 0, 0, time.UTC) + pulse := aStuckKey(t, sdkdelivery.PulseBucket, "x") + moved := aStuckKey(t, sdkdelivery.HaveBucket, "ace.zsh") + node := aStuckKey(t, sdkdelivery.PresenceBucket, "ace") + part := aStuckKey(t, sdkdelivery.PresenceBucket, "part.mesh-controller.one") + entries := map[string]sdkdelivery.Stuck{ + pulse: {Bucket: sdkdelivery.PulseBucket, Key: "x", Waited: "the build of zsh to be answered", Since: since}, + moved: {Bucket: sdkdelivery.HaveBucket, Key: "ace.zsh", Waited: "a want", Since: since, Moved: true}, + node: {Bucket: sdkdelivery.PresenceBucket, Key: "ace", Waited: "its return", Since: since}, + part: {Bucket: sdkdelivery.PresenceBucket, Key: "part.mesh-controller.one", Waited: "its return", Since: since}, + } + got := stuckObservations(entries) + if len(got) != 2 { + t.Fatalf("observations %+v, want the pulse key and the part's presence", got) + } + byID := map[string]conditions.Observation{} + for _, o := range got { + if o.Scope != conditions.ScopeDelivery || o.Kind != kindStuck { + t.Errorf("%+v is not a stuck condition of the delivery scope", o) + } + byID[o.ID] = o + } + o, ok := byID[pulse] + if !ok || !strings.Contains(o.Summary, pulse) || !strings.Contains(o.Summary, "the build of zsh to be answered") { + t.Fatalf("the condition does not name the key and what it waited for: %+v", got) + } + if _, ok := byID[part]; !ok { + t.Errorf("a part's presence is left out: %v", byID) + } + // Moved again: the same key, written as moved, raises nothing. + entries[pulse] = sdkdelivery.Stuck{Bucket: sdkdelivery.PulseBucket, Key: "x", Since: since, Moved: true} + if got := stuckObservations(entries); len(got) != 1 { + t.Fatalf("a moved key still raises: %+v", got) + } +} + +func aMachineHeardAt(now time.Time, ago time.Duration) machineFacts { + return machineFacts{name: "laptop", lastHeard: now.Add(-ago), every: time.Minute} +} + +// 1156/AC2: a node's stuck presence key not moved makes the machine silent, unless it sleeps; moved, it is heard; no +// timer is armed (the verdict is the key's, whatever the time of the last word). +func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2 + now := time.Now() + f := calm(now) + f.machines = append(f.machines, aMachineHeardAt(now, 10*time.Second)) + notMoved := sdkdelivery.Stuck{Bucket: sdkdelivery.PresenceBucket, Key: "laptop", Waited: "its return", Since: now.Add(-3 * time.Minute)} + applyPresence(&f.machines[1], ¬Moved, now.Add(-3*time.Minute)) + got := watchHeartbeats(f) + if len(got) != 1 || got[0].Machine != "laptop" || got[0].Kind != "silent" { + t.Fatalf("a presence key not moved: %+v", got) + } + // Asleep: not silent. + f.machines[1].power = link.PowerState{State: "sleeping", At: now.Add(-time.Hour)} + if got := watchHeartbeats(f); len(got) != 0 { + t.Fatalf("a sleeping machine raised %+v", got) + } + // Moved: heard, whatever the store's last word says. + f.machines[1].power = link.PowerState{} + f.machines[1].lastHeard = now.Add(-2 * time.Hour) + moved := notMoved + moved.Moved = true + applyPresence(&f.machines[1], &moved, now.Add(-time.Minute)) + if got := watchHeartbeats(f); len(got) != 0 { + t.Fatalf("a presence key that moved raised %+v", got) + } +} + +// 1156/AC3: a machine's last heard is the presence key's entry time when there is one, and last_seen when there is none. +func TestLastHeardIsThePresenceKeysTimeWhenThereIsOne(t *testing.T) { // 1156/AC3 + now := time.Now() + seen := now.Add(-time.Hour) + m := machineFacts{name: "laptop", lastHeard: seen} + applyPresence(&m, nil, time.Time{}) + if !m.lastHeard.Equal(seen) { + t.Fatalf("with no presence key last heard is %v, want last_seen %v", m.lastHeard, seen) + } + at := now.Add(-5 * time.Minute) + applyPresence(&m, nil, at) + if !m.lastHeard.Equal(at) { + t.Fatalf("with a presence key last heard is %v, want %v", m.lastHeard, at) + } +} diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index 1c708fb5..2e5c2e4e 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -155,6 +155,8 @@ type machineFacts struct { tools bool toolsHeard time.Time toolsEvery time.Duration + // presence is the watchman's word on whether it is online (novox/hq ADR 0433 §3); unsaid, the heartbeat rule judges. + presence presenceVerdict } // asleep says the machine said it would be away and has not said it is back (ADR 0211). @@ -470,7 +472,7 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory } m.reportedCurrent = r.Current } - if !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > heartbeatBound(m.every) { + if m.silent(now) { anyPast = true } if m.tools && now.Sub(later(m.toolsHeard, link.ToolsBeats.Started())) > heartbeatBound(m.toolsEvery) { @@ -478,6 +480,19 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory } out = append(out, m) } + // What the watchman says of each machine: its stuck presence key is the silent verdict, and presence.'s time when + // it was last heard (ADR 0433 §3). Unreadable is said, and the heartbeat rule judges as before. + if w.js != nil { + if err := withPresence(ctx, w.js.Conn(), out); err != nil { + fmt.Printf("what the watchman says of the machines' presence cannot be read: %v\n", err) + } else { + for _, m := range out { + if m.silent(now) { + anyPast = true + } + } + } + } // What a machine past its bound last said of its power, read only then: a machine that said it // is asleep is not lost (ADR 0211). Unreadable is said: a sleeping laptop is then called silent, // which is the louder mistake and the right one to make.