From 863db82737abd82a3cb754d9db797acf7e3cb78e Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 02:13:56 +0200 Subject: [PATCH 1/2] The controller's last-heard word for a machine comes from the watchman's presence in nodes, node, status and the watchdogs (issue 1183, AC1, AC2) --- cmd/mesh-controller/last_heard.go | 120 ++++++++++++++++++++++++ cmd/mesh-controller/last_heard_test.go | 97 +++++++++++++++++++ cmd/mesh-controller/nodes.go | 22 +---- cmd/mesh-controller/status.go | 11 +-- cmd/mesh-controller/stuck_probe.go | 11 +-- cmd/mesh-controller/stuck_probe_test.go | 6 +- 6 files changed, 228 insertions(+), 39 deletions(-) create mode 100644 cmd/mesh-controller/last_heard.go create mode 100644 cmd/mesh-controller/last_heard_test.go diff --git a/cmd/mesh-controller/last_heard.go b/cmd/mesh-controller/last_heard.go new file mode 100644 index 00000000..c5cb57d4 --- /dev/null +++ b/cmd/mesh-controller/last_heard.go @@ -0,0 +1,120 @@ +package main + +import ( + "context" + "fmt" + "os" + "time" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" +) + +// Last heard (novox/hq ADR 0433 §3, issue 1183): one answer to when the controller last heard a machine, read from the +// watchman's presence. The alive message that used to keep last_seen fresh is gone; a report, which is rare, is the +// answer only where presence has no key for the machine. + +// presenceReader reads presence.'s entry for each node; a node without an entry is left out. +type presenceReader func(ctx context.Context, nodes []string) (map[string]presenceEntry, error) + +// lastHeard is when a machine was last heard: now when presence says it is online, the entry's time when it says offline +// (the time it went), and the machine's last report only when presence has no entry for it. Zero is never heard. +func lastHeard(report time.Time, entry *presenceEntry, now time.Time) time.Time { + switch { + case entry == nil: + return report + case entry.Online: + return now + default: + return entry.At + } +} + +// nodesHeard returns the nodes with their last-heard time (LastSeen) as lastHeard says it. Unreadable presence is +// returned as an error with the nodes unchanged, so a caller says it and shows the reports. +func nodesHeard(ctx context.Context, nodes []inventory.Node, read presenceReader, now time.Time) ([]inventory.Node, error) { + names := make([]string, len(nodes)) + for i, n := range nodes { + names[i] = n.Name + } + entries, err := read(ctx, names) + if err != nil { + return nodes, fmt.Errorf("what the watchman says of the machines' presence cannot be read: %w", err) + } + out := make([]inventory.Node, len(nodes)) + for i, n := range nodes { + var entry *presenceEntry + if e, ok := entries[n.Name]; ok { + entry = &e + } + n.LastSeen = lastHeard(n.LastSeen, entry, now) + out[i] = n + } + return out, nil +} + +// heardNodes is every node, read with its last-heard time from the watchman's presence over a connection of its own. A bus +// that cannot be reached is said on standard error and the reports stand, which reads a machine as quiet rather than well. +func (h *stores) heardNodes(ctx context.Context) ([]inventory.Node, error) { + nodes, err := h.inventory.Nodes(ctx) + if err != nil { + return nil, err + } + return h.withPresence(ctx, nodes), nil +} + +// heardNode is one node, read the same way. +func (h *stores) heardNode(ctx context.Context, name string) (inventory.Node, error) { + n, err := h.inventory.NodeByName(ctx, name) + if err != nil { + return n, err + } + return h.withPresence(ctx, []inventory.Node{n})[0], nil +} + +func (h *stores) withPresence(ctx context.Context, nodes []inventory.Node) []inventory.Node { + address, err := broker.BusAddress() + var js *broker.JetStream + if err == nil { + js, err = broker.Dial(address) + } + if err != nil { + fmt.Fprintf(os.Stderr, "what the watchman says of the machines' presence cannot be read: %v\n", err) + return nodes + } + defer js.Close() + read := func(ctx context.Context, names []string) (map[string]presenceEntry, error) { + return readPresence(ctx, js.Conn(), names) + } + heard, err := nodesHeard(ctx, nodes, read, time.Now()) + if err != nil { + fmt.Fprintln(os.Stderr, err) + } + return heard +} + +// heardFromAt says when a node was last heard from, in a form somebody can act on. +// +// "never" and "an hour ago" are different answers and are kept different. A node that has never spoken did not finish +// joining; a node last heard from an hour ago is running an hour-old picture of the mesh. +func heardFromAt(n inventory.Node, now time.Time) string { + if n.LastSeen.IsZero() { + return "never spoken" + } + if silent := now.Sub(n.LastSeen); silent > SilentFor { + return "out of touch " + roughly(silent) + } + return "here" +} + +// notHeardLately is the nodes never heard from or not within the hour: different from failing, for a machine that says +// nothing may be new, switched off or unreachable, and none of those is a machine that tried and could not. +func notHeardLately(nodes []inventory.Node, now time.Time) []inventory.Node { + var out []inventory.Node + for _, n := range nodes { + if n.LastSeen.IsZero() || now.Sub(n.LastSeen) > time.Hour { + out = append(out, n) + } + } + return out +} diff --git a/cmd/mesh-controller/last_heard_test.go b/cmd/mesh-controller/last_heard_test.go new file mode 100644 index 00000000..187a2b9f --- /dev/null +++ b/cmd/mesh-controller/last_heard_test.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// aPresenceFake is the watchman's presence bucket as a reader sees it. +func aPresenceFake(entries map[string]presenceEntry, err error) presenceReader { + return func(_ context.Context, _ []string) (map[string]presenceEntry, error) { return entries, err } +} + +// 1183/AC1: one function says when a machine was last heard: online is now, offline is the entry's time, and only where +// presence has no key, the last report. +func TestLastHeardIsPresenceThenTheReport(t *testing.T) { // 1183/AC1 + now := time.Now() + report := now.Add(-time.Hour) + if got := lastHeard(report, nil, now); !got.Equal(report) { + t.Fatalf("no presence key: last heard is %v, want the report %v", got, report) + } + if got := lastHeard(report, &presenceEntry{Online: true, At: now.Add(-9 * time.Hour)}, now); !got.Equal(now) { + t.Fatalf("online: last heard is %v, want now", got) + } + offlineAt := now.Add(-4 * time.Minute) + if got := lastHeard(now.Add(-time.Minute), &presenceEntry{At: offlineAt}, now); !got.Equal(offlineAt) { + t.Fatalf("offline: last heard is %v, want the entry's time %v, not a report's", got, offlineAt) + } + if got := lastHeard(time.Time{}, nil, now); !got.IsZero() { + t.Fatalf("never heard and no key: %v", got) + } +} + +// The machines facts, the nodes list, node and status all take that one function. +func TestEveryReaderOfLastHeardTakesPresence(t *testing.T) { // 1183/AC1 + now := time.Now() + m := machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute} + applyPresence(&m, nil, &presenceEntry{At: now.Add(-4 * time.Minute)}, now) + if !m.lastHeard.Equal(now.Add(-4 * time.Minute)) { + t.Fatalf("machine facts take a report over the offline entry: %v", m.lastHeard) + } +} + +func twoNodes(now time.Time) []inventory.Node { + return []inventory.Node{ + {Name: "anchor", LastSeen: now.Add(-time.Hour)}, + {Name: "laptop", LastSeen: now.Add(-30 * time.Second)}, + {Name: "fresh"}, + } +} + +// 1183/AC2: a machine online by presence whose last report is an hour old reads here, is not quiet and is not silent; one +// offline for 4 minutes reads out of touch. +func TestOnlineByPresenceReadsHereWhateverTheReport(t *testing.T) { // 1183/AC2 + now := time.Now() + read := aPresenceFake(map[string]presenceEntry{ + "anchor": {Online: true, At: now.Add(-6 * time.Hour)}, + "laptop": {Online: false, At: now.Add(-4 * time.Minute)}, + }, nil) + nodes, err := nodesHeard(context.Background(), twoNodes(now), read, now) + if err != nil { + t.Fatal(err) + } + if got := heardFromAt(nodes[0], now); got != "here" { + t.Errorf("anchor online with an hour-old report reads %q", got) + } + if got := heardFromAt(nodes[1], now); got != "out of touch 4m" { + t.Errorf("laptop offline 4 minutes reads %q", got) + } + if got := heardFromAt(nodes[2], now); got != "never spoken" { + t.Errorf("a machine with no key and no report reads %q", got) + } + quiet := notHeardLately(nodes, now) + if len(quiet) != 1 || quiet[0].Name != "fresh" { + t.Errorf("quiet is %+v: an online machine is not quiet, and a machine offline 4 minutes is not quiet yet", quiet) + } + f := &machineFacts{name: "anchor", lastHeard: nodes[0].LastSeen, every: time.Minute} + if f.silent(now) { + t.Errorf("a machine online by presence is silent") + } +} + +// Unreadable presence is said, and the report stands as it was. +func TestUnreadablePresenceLeavesTheReport(t *testing.T) { // 1183/AC1 + now := time.Now() + nodes, err := nodesHeard(context.Background(), twoNodes(now), aPresenceFake(nil, errors.New("no bus")), now) + if err == nil || !strings.Contains(err.Error(), "no bus") { + t.Fatalf("an unreadable presence is not said: %v", err) + } + if !nodes[0].LastSeen.Equal(now.Add(-time.Hour)) { + t.Fatalf("the report was replaced: %v", nodes[0].LastSeen) + } +} diff --git a/cmd/mesh-controller/nodes.go b/cmd/mesh-controller/nodes.go index a297e7b3..8d8e19ec 100644 --- a/cmd/mesh-controller/nodes.go +++ b/cmd/mesh-controller/nodes.go @@ -50,7 +50,7 @@ func nodeCommand(ctx context.Context, args []string) error { return addNode(ctx, inv, args[1:]) case "list": - nodes, err := inv.Nodes(ctx) + nodes, err := open.heardNodes(ctx) if err != nil { return err } @@ -755,22 +755,8 @@ func consumerReset(ctx context.Context, args []string) error { return nil } -// heardFrom says when a node was last heard from, in a form somebody can act on. -// -// "never" and "an hour ago" are different answers and are kept different. A node that has never -// spoken did not finish joining; a node last heard from an hour ago is running an hour-old -// picture of the mesh. -func heardFrom(n inventory.Node) string { - silent, ever := n.Silent() - switch { - case !ever: - return "never spoken" - case silent > SilentFor: - return "out of touch " + roughly(silent) - default: - return "here" - } -} +// heardFrom is heardFromAt for now. +func heardFrom(n inventory.Node) string { return heardFromAt(n, time.Now()) } // roughly is a duration a person reads rather than parses. func roughly(d time.Duration) string { @@ -795,7 +781,7 @@ func roughly(d time.Duration) string { // assignment carries `card1-DP-1`, and a person composing settings for that machine needs it. func showNode(ctx context.Context, stored *stores, name string) error { inv := stored.inventory - node, err := inv.NodeByName(ctx, name) + node, err := stored.heardNode(ctx, name) if err != nil { return err } diff --git a/cmd/mesh-controller/status.go b/cmd/mesh-controller/status.go index 01cf7f98..2a103cdc 100644 --- a/cmd/mesh-controller/status.go +++ b/cmd/mesh-controller/status.go @@ -414,18 +414,11 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) { if err != nil { return answers{}, err } - out.nodes, err = inv.Nodes(ctx) + out.nodes, err = open.heardNodes(ctx) if err != nil { return answers{}, err } - for _, n := range out.nodes { - // Never heard from, or not lately. Different from failing: a machine that says nothing - // may be new, switched off, or unreachable, and none of those is a machine that tried - // and could not. - if n.LastSeen.IsZero() || time.Since(n.LastSeen) > time.Hour { - out.quiet = append(out.quiet, n) - } - } + out.quiet = notHeardLately(out.nodes, time.Now()) out.behind, err = inv.Behind(ctx) if err != nil { return answers{}, err diff --git a/cmd/mesh-controller/stuck_probe.go b/cmd/mesh-controller/stuck_probe.go index e97a02df..bdab8e5b 100644 --- a/cmd/mesh-controller/stuck_probe.go +++ b/cmd/mesh-controller/stuck_probe.go @@ -43,8 +43,7 @@ type presenceEntry struct { } // applyPresence sets what the watchman says of a machine: the verdict of its stuck presence key (nil when there is none). -// A presence entry that says online counts the machine as heard now; otherwise last heard is the later of the store's -// last_seen and the entry's time (the time it went offline). +// Last heard is lastHeard's: now when online, the entry's time when offline, the store's last report only without an entry. func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEntry, now time.Time) { switch { case stuck == nil: @@ -54,13 +53,7 @@ func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEnt default: m.presence = presenceSilent } - switch { - case entry == nil: - case entry.Online: - m.lastHeard = now - case entry.At.After(m.lastHeard): - m.lastHeard = entry.At - } + m.lastHeard = lastHeard(m.lastHeard, entry, now) } // silent says the machine is silent: the watchman's verdict where it gave one, else the heartbeat rule. A machine never diff --git a/cmd/mesh-controller/stuck_probe_test.go b/cmd/mesh-controller/stuck_probe_test.go index aab335ea..a48305ff 100644 --- a/cmd/mesh-controller/stuck_probe_test.go +++ b/cmd/mesh-controller/stuck_probe_test.go @@ -91,7 +91,7 @@ func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2 } } -// 1156/AC3: with a presence entry that says offline, last heard is the later of last_seen and the entry's time; with none, +// 1156/AC3: with a presence entry that says offline, last heard is the entry's time (1183/AC1); with none, // last_seen. An entry that says online counts as heard now: the watchman writes it only when it changes, so its time is // when the machine connected, and a machine online for hours with no stuck key is not silent, whatever last_seen says. func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3 @@ -109,8 +109,8 @@ func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3 } m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute} applyPresence(&m, nil, &presenceEntry{Online: false, At: now.Add(-time.Hour)}, now) - if !m.lastHeard.Equal(now.Add(-time.Minute)) { - t.Fatalf("an older entry replaced a newer last_seen: %v", m.lastHeard) + if !m.lastHeard.Equal(now.Add(-time.Hour)) { + t.Fatalf("an offline entry's time is last heard, not a report: %v", m.lastHeard) } for _, lastSeen := range []time.Duration{10 * time.Second, 5 * time.Hour} { m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen), every: time.Minute} -- 2.54.0 From 19e3eac74658e40c6db4e6311a9bc786c05d8d77 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 02:16:21 +0200 Subject: [PATCH 2/2] The controller no longer subscribes to the alive message: HostBeats, its handler, the heartbeat-gap duration and a fixed silent bound (issue 1183, AC3, AC4, AC5) --- cmd/mesh-controller/durations.go | 4 +- cmd/mesh-controller/handacts_test.go | 6 +- .../last_heard_integration_test.go | 82 +++++++++++++++++++ cmd/mesh-controller/last_heard_test.go | 19 ++++- cmd/mesh-controller/signals.go | 6 +- cmd/mesh-controller/signals_test.go | 2 +- cmd/mesh-controller/stuck_probe.go | 2 +- .../stuck_probe_integration_test.go | 2 +- cmd/mesh-controller/stuck_probe_test.go | 8 +- cmd/mesh-controller/watchdogs.go | 7 +- internal/catalogue/verbs.go | 4 +- internal/inventory/durations.go | 20 +---- internal/inventory/durations_test.go | 19 ----- internal/inventory/heartbeat_gap_gone_test.go | 12 +++ internal/link/bus.go | 10 +-- internal/link/contracts.go | 2 - internal/link/contracts_test.go | 16 +++- internal/link/enrolment.go | 9 +- internal/link/protocol.go | 14 ---- internal/link/receive.go | 1 - internal/link/receive_nats.go | 10 +-- .../link/receive_nats_integration_test.go | 25 ++++-- internal/link/receive_nats_test.go | 3 - internal/link/serve.go | 28 +------ internal/link/stale_report_test.go | 12 --- internal/link/watched.go | 11 +-- 26 files changed, 174 insertions(+), 160 deletions(-) create mode 100644 cmd/mesh-controller/last_heard_integration_test.go create mode 100644 internal/inventory/heartbeat_gap_gone_test.go diff --git a/cmd/mesh-controller/durations.go b/cmd/mesh-controller/durations.go index bb7e5454..36afadc7 100644 --- a/cmd/mesh-controller/durations.go +++ b/cmd/mesh-controller/durations.go @@ -72,8 +72,6 @@ func summarise(ds []inventory.Duration) []durationSummary { switch k.kind { case inventory.DurationApply: s.Suggests = "S2 bound max(2m, 3×last apply) ≈ " + round(max(2*time.Minute, 3*at(0.9))) + " at the p90" - case inventory.DurationHeartbeatGap: - s.Suggests = "S1 bound 3×interval ≈ " + round(3*at(0.5)) } out = append(out, s) } @@ -138,7 +136,7 @@ func durationsCommand(ctx context.Context, args []string) error { return nil } if len(summary) == 0 { - fmt.Printf("nothing measured in the last %d day(s): the serving controller records apply, heartbeat-gap, "+ + fmt.Printf("nothing measured in the last %d day(s): the serving controller records apply, "+ "plan-tier and build durations as it hears them\n", *days) return nil } diff --git a/cmd/mesh-controller/handacts_test.go b/cmd/mesh-controller/handacts_test.go index e51398eb..19d5a4f5 100644 --- a/cmd/mesh-controller/handacts_test.go +++ b/cmd/mesh-controller/handacts_test.go @@ -77,15 +77,11 @@ func TestDurationsAreSummarisedPerSubject(t *testing.T) { ds = append(ds, inventory.Duration{Kind: inventory.DurationApply, Subject: "anchor", Took: time.Duration(i) * time.Second}) } - ds = append(ds, inventory.Duration{Kind: inventory.DurationHeartbeatGap, Subject: "anchor", Took: time.Minute}) got := summarise(ds) - if len(got) != 2 || got[0].Kind != inventory.DurationApply || got[0].Count != 10 || + if len(got) != 1 || got[0].Kind != inventory.DurationApply || got[0].Count != 10 || got[0].Max != "10s" || got[0].Median != "5s" || got[0].P90 != "9s" { t.Fatalf("%+v", got) } - if !strings.Contains(got[1].Suggests, "3m0s") { - t.Fatalf("a minute between words suggests %q", got[1].Suggests) - } } // **A drill has its own verb** — `hand-act drill`, the seat's `drill` — and `hand-act record` refuses diff --git a/cmd/mesh-controller/last_heard_integration_test.go b/cmd/mesh-controller/last_heard_integration_test.go new file mode 100644 index 00000000..a1207e80 --- /dev/null +++ b/cmd/mesh-controller/last_heard_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/inventory" + "github.com/novox/mesh-controller/internal/testbus" +) + +// 1183/AC5: with the watchman's presence bucket holding a node online and no alive message ever sent, the node reads here +// in `nodes` four minutes later by the controller's clock, however old its last report; and an alive message still sent +// by a node-engine is published to the same bus without error and changes nothing. +func TestAnOnlineNodeReadsHereWithNoAliveMessage(t *testing.T) { // 1183/AC5 + 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: "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.PresenceBucket) + if err != nil { + t.Fatal(err) + } + put := func(node string, online bool) { + t.Helper() + key, err := sdkdelivery.PresenceNodeKey(node) + if err != nil { + t.Fatal(err) + } + body, err := sdkdelivery.EncodePresence(sdkdelivery.Presence{Online: online}) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, body); err != nil { + t.Fatal(err) + } + } + put("ace", true) + put("laptop", false) + // A node-engine still sending its alive message: harmless. + if err := js.Conn().Publish("mesh.control.ace.alive", []byte(`{"node":"ace"}`)); err != nil { + t.Fatalf("an alive message could not be published: %v", err) + } + if err := js.Conn().Flush(); err != nil { + t.Fatal(err) + } + + later := time.Now().Add(4 * time.Minute) + nodes := []inventory.Node{{Name: "ace", LastSeen: time.Now().Add(-time.Hour)}, + {Name: "laptop", LastSeen: time.Now().Add(-time.Hour)}} + read := func(ctx context.Context, names []string) (map[string]presenceEntry, error) { + return readPresence(ctx, js.Conn(), names) + } + heard, err := nodesHeard(ctx, nodes, read, later) + if err != nil { + t.Fatal(err) + } + if got := heardFromAt(heard[0], later); got != "here" { + t.Fatalf("a node online by presence reads %q four minutes later", got) + } + if got := heardFromAt(heard[1], later); got == "here" { + t.Fatalf("a node offline by presence reads %q", got) + } +} diff --git a/cmd/mesh-controller/last_heard_test.go b/cmd/mesh-controller/last_heard_test.go index 187a2b9f..3b926010 100644 --- a/cmd/mesh-controller/last_heard_test.go +++ b/cmd/mesh-controller/last_heard_test.go @@ -38,7 +38,7 @@ func TestLastHeardIsPresenceThenTheReport(t *testing.T) { // 1183/AC1 // The machines facts, the nodes list, node and status all take that one function. func TestEveryReaderOfLastHeardTakesPresence(t *testing.T) { // 1183/AC1 now := time.Now() - m := machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute} + m := machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute)} applyPresence(&m, nil, &presenceEntry{At: now.Add(-4 * time.Minute)}, now) if !m.lastHeard.Equal(now.Add(-4 * time.Minute)) { t.Fatalf("machine facts take a report over the offline entry: %v", m.lastHeard) @@ -78,7 +78,7 @@ func TestOnlineByPresenceReadsHereWhateverTheReport(t *testing.T) { // 1183/AC2 if len(quiet) != 1 || quiet[0].Name != "fresh" { t.Errorf("quiet is %+v: an online machine is not quiet, and a machine offline 4 minutes is not quiet yet", quiet) } - f := &machineFacts{name: "anchor", lastHeard: nodes[0].LastSeen, every: time.Minute} + f := &machineFacts{name: "anchor", lastHeard: nodes[0].LastSeen} if f.silent(now) { t.Errorf("a machine online by presence is silent") } @@ -95,3 +95,18 @@ func TestUnreadablePresenceLeavesTheReport(t *testing.T) { // 1183/AC1 t.Fatalf("the report was replaced: %v", nodes[0].LastSeen) } } + +// 1183/AC4: where the watchman gives no verdict a machine is silent after a fixed three minutes, whatever interval a +// node-engine once said. +func TestTheSilentBoundIsFixed(t *testing.T) { // 1183/AC4 + now := time.Now() + for _, c := range []struct { + ago time.Duration + want bool + }{{179 * time.Second, false}, {181 * time.Second, true}} { + m := machineFacts{name: "laptop", lastHeard: now.Add(-c.ago)} + if got := m.silent(now); got != c.want { + t.Errorf("heard %v ago: silent %v, want %v", c.ago, got, c.want) + } + } +} diff --git a/cmd/mesh-controller/signals.go b/cmd/mesh-controller/signals.go index 43ac6d41..c311efb4 100644 --- a/cmd/mesh-controller/signals.go +++ b/cmd/mesh-controller/signals.go @@ -318,6 +318,10 @@ func ago(d time.Duration) string { return d.Round(time.Minute).String() } +// silentBound is how long a machine may go unheard, where the watchman gives no verdict: three of the intervals nodes have +// always used, fixed now that no machine says its own (issue 1183). +const silentBound = heartbeatsMissed * heartbeatEvery + // heartbeatBound is a machine's S1 or S11 bound from the interval it says. func heartbeatBound(every time.Duration) time.Duration { if every <= 0 { @@ -333,7 +337,7 @@ func watchHeartbeats(f *signalFacts) []conditions.Observation { // Never heard is a machine that has not joined, which status says; asleep is not lost. continue } - bound := heartbeatBound(m.every) + bound := silentBound silent := f.now.Sub(m.lastHeard) if !m.silent(f.now) { continue diff --git a/cmd/mesh-controller/signals_test.go b/cmd/mesh-controller/signals_test.go index efa2ae7e..3fc0561c 100644 --- a/cmd/mesh-controller/signals_test.go +++ b/cmd/mesh-controller/signals_test.go @@ -28,7 +28,7 @@ import ( func calm(now time.Time) *signalFacts { return &signalFacts{now: now, started: now.Add(-time.Hour), host: "anchor", toolsHeardFrom: now.Add(-time.Hour), machines: []machineFacts{{name: "anchor", control: true, lastHeard: now.Add(-10 * time.Second), - every: time.Minute, sentAt: now.Add(-time.Hour), reportedCurrent: true, reportedAt: now.Add(-59 * time.Minute), + sentAt: now.Add(-time.Hour), reportedCurrent: true, reportedAt: now.Add(-59 * time.Minute), lastApply: 20 * time.Second, tools: true, toolsHeard: now.Add(-10 * time.Second), toolsEvery: time.Minute}}, plans: []planFacts{{id: "plan-1", repository: "novox/app", commit: "c0ffee00", tier: 0, tiers: 2, entered: now.Add(-time.Minute), bound: 30 * time.Minute, waiting: "building"}}, diff --git a/cmd/mesh-controller/stuck_probe.go b/cmd/mesh-controller/stuck_probe.go index bdab8e5b..db373b05 100644 --- a/cmd/mesh-controller/stuck_probe.go +++ b/cmd/mesh-controller/stuck_probe.go @@ -65,7 +65,7 @@ func (m machineFacts) silent(now time.Time) bool { case presenceHeard: return false } - return !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > heartbeatBound(m.every) + return !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > silentBound } // stuckObservations are the conditions of what stopped moving: one for each key not moved, in the delivery scope, kind diff --git a/cmd/mesh-controller/stuck_probe_integration_test.go b/cmd/mesh-controller/stuck_probe_integration_test.go index acf070c8..d29b9add 100644 --- a/cmd/mesh-controller/stuck_probe_integration_test.go +++ b/cmd/mesh-controller/stuck_probe_integration_test.go @@ -72,7 +72,7 @@ func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { / 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}} + machines = []machineFacts{{name: "ace", lastHeard: time.Now().Add(-time.Hour)}} if err := withPresence(ctx, js.Conn(), machines, time.Now()); err != nil { t.Fatal(err) } diff --git a/cmd/mesh-controller/stuck_probe_test.go b/cmd/mesh-controller/stuck_probe_test.go index a48305ff..4b61a713 100644 --- a/cmd/mesh-controller/stuck_probe_test.go +++ b/cmd/mesh-controller/stuck_probe_test.go @@ -60,7 +60,7 @@ func TestEachStuckKeyThatHasNotMovedIsAConditionOfTheDeliveryScope(t *testing.T) } func aMachineHeardAt(now time.Time, ago time.Duration) machineFacts { - return machineFacts{name: "laptop", lastHeard: now.Add(-ago), every: time.Minute} + return machineFacts{name: "laptop", lastHeard: now.Add(-ago)} } // 1156/AC2: a node's stuck presence key not moved makes the machine silent, unless it sleeps; moved, it is heard; no @@ -97,7 +97,7 @@ func TestAStuckPresenceKeyIsTheSilentVerdict(t *testing.T) { // 1156/AC2 func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3 now := time.Now() seen := now.Add(-time.Hour) - m := machineFacts{name: "laptop", lastHeard: seen, every: time.Minute} + m := machineFacts{name: "laptop", lastHeard: seen} applyPresence(&m, nil, nil, now) if !m.lastHeard.Equal(seen) { t.Fatalf("with no presence entry last heard is %v, want last_seen %v", m.lastHeard, seen) @@ -107,13 +107,13 @@ func TestLastHeardFollowsThePresenceEntry(t *testing.T) { // 1156/AC3 if !m.lastHeard.Equal(offlineAt) { t.Fatalf("offline since %v: last heard is %v", offlineAt, m.lastHeard) } - m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute), every: time.Minute} + m = machineFacts{name: "laptop", lastHeard: now.Add(-time.Minute)} applyPresence(&m, nil, &presenceEntry{Online: false, At: now.Add(-time.Hour)}, now) if !m.lastHeard.Equal(now.Add(-time.Hour)) { t.Fatalf("an offline entry's time is last heard, not a report: %v", m.lastHeard) } for _, lastSeen := range []time.Duration{10 * time.Second, 5 * time.Hour} { - m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen), every: time.Minute} + m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen)} applyPresence(&m, nil, &presenceEntry{Online: true, At: now.Add(-6 * time.Hour)}, now) if m.silent(now) { t.Fatalf("a machine online since hours ago (last_seen %v ago) with no stuck key is silent", lastSeen) diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index ce29a2f1..3ec0ff1c 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -142,9 +142,7 @@ type machineFacts struct { control bool // lastHeard is the store's last word from it, any word; zero when never. lastHeard time.Time - // every is the heartbeat interval it said; zero when it said none. - every time.Duration - power link.PowerState + power link.PowerState // sentAt is when it was last sent a declaration; reportedCurrent whether it has reported that one. sentAt time.Time reportedCurrent bool @@ -457,9 +455,6 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory for _, n := range nodes { m := machineFacts{name: n.Name, control: slices.Contains(control, n.Name), lastHeard: n.LastSeen, lastApply: lastApply[n.Name], tools: slices.Contains(tooled, n.Name)} - if beat, ok := link.HostBeats.Of(n.Name); ok { - m.every = beat.Every - } if beat, ok := link.ToolsBeats.Of(n.Name); ok { m.toolsHeard, m.toolsEvery = beat.At, beat.Every } diff --git a/internal/catalogue/verbs.go b/internal/catalogue/verbs.go index cd4c2d9c..aa470807 100644 --- a/internal/catalogue/verbs.go +++ b/internal/catalogue/verbs.go @@ -450,11 +450,11 @@ var ControllerVerbs = []Verb{ "recorded — who, why and the cause of each, and which causes repeat: each repeat is a healer the mesh lacks.", Input: schema(map[string]string{"days": "how many days back (default 14)"}, nil)}, {Name: "durations", Description: "How long things take, as the controller measured them: a send to its machine's " + - "report (apply), a machine's silence between words (heartbeat-gap), a plan's tier, a build, and each phase of an " + + "report (apply), a plan's tier, a build, and each phase of an " + "ended walk per class (walk-phase, novox/hq ADR 0282) — per machine, repository, module or class/phase, with " + "median, p90 and max. What the core's bounds are set from (novox/hq to-be 45 Phase 0).", Input: schema(map[string]string{ - "kind": "one kind: apply, heartbeat-gap, plan-tier, build or walk-phase; every kind when absent", + "kind": "one kind: apply, plan-tier, build or walk-phase; every kind when absent", "days": "how many days back (default 14)", }, nil)}, // What is wrong, and the self-check (novox/hq to-be 45 §2, §4). diff --git a/internal/inventory/durations.go b/internal/inventory/durations.go index 5be4b96f..5611cef8 100644 --- a/internal/inventory/durations.go +++ b/internal/inventory/durations.go @@ -9,10 +9,9 @@ import ( // The kinds of duration recorded. const ( - DurationApply = "apply" - DurationHeartbeatGap = "heartbeat-gap" - DurationPlanTier = "plan-tier" - DurationBuild = "build" + DurationApply = "apply" + DurationPlanTier = "plan-tier" + DurationBuild = "build" // DurationWalkPhase is one phase of an ended walk, per class (novox/hq ADR 0282 decision 6): subject // "/", and "/total" for its merge to every machine running it. DurationWalkPhase = "walk-phase" @@ -27,7 +26,7 @@ const ( const ApplyBusy = "busy" // DurationKinds are every kind, in the order `durations` shows them. -var DurationKinds = []string{DurationApply, DurationHeartbeatGap, DurationPlanTier, DurationBuild, DurationWalkPhase, +var DurationKinds = []string{DurationApply, DurationPlanTier, DurationBuild, DurationWalkPhase, DurationBusy} // DurationsKeptFor is how long a duration is kept: long enough to set a bound from, and to correct it @@ -94,17 +93,6 @@ func (i *Inventory) RecordBusyDuration(ctx context.Context, node, declared strin return err } -// RecordHeartbeatGap measures the silence before a machine's word: from the last word heard before -// it, which the caller read before recording this one. -func (i *Inventory) RecordHeartbeatGap(ctx context.Context, node string, before time.Time) error { - if before.IsZero() { - return nil - } - now := time.Now() - return i.RecordDuration(ctx, Duration{Kind: DurationHeartbeatGap, Subject: node, Node: node, - Ref: before.UTC().Format(time.RFC3339Nano), Started: before, Took: now.Sub(before)}) -} - // Durations is every measurement of a kind since a moment, oldest first; every kind when kind is empty. func (i *Inventory) Durations(ctx context.Context, kind string, since time.Time) ([]Duration, error) { rows, err := i.store.Pool().Query(ctx, diff --git a/internal/inventory/durations_test.go b/internal/inventory/durations_test.go index 356618f5..639866dc 100644 --- a/internal/inventory/durations_test.go +++ b/internal/inventory/durations_test.go @@ -44,25 +44,6 @@ func TestAnApplyIsMeasuredOncePerSend(t *testing.T) { } } -// A machine's silence is the time since its last word, once per word. -func TestASilenceIsMeasuredFromTheLastWord(t *testing.T) { - inv := ForTest(t) - ctx := t.Context() - before := time.Now().Add(-90 * time.Second) - for range 2 { - if err := inv.RecordHeartbeatGap(ctx, "anchor", before); err != nil { - t.Fatal(err) - } - } - if err := inv.RecordHeartbeatGap(ctx, "anchor", time.Time{}); err != nil { - t.Fatal(err) - } - ds, err := inv.Durations(ctx, DurationHeartbeatGap, time.Now().Add(-time.Hour)) - if err != nil || len(ds) != 1 || ds[0].Took < 90*time.Second || ds[0].Took > 2*time.Minute { - t.Fatalf("%v %+v", err, ds) - } -} - // A plan's tier is measured when the plan leaves it — moving on, or ending — and only then. func TestAPlansTierIsMeasuredWhenItIsLeft(t *testing.T) { inv := ForTest(t) diff --git a/internal/inventory/heartbeat_gap_gone_test.go b/internal/inventory/heartbeat_gap_gone_test.go new file mode 100644 index 00000000..1fcce32c --- /dev/null +++ b/internal/inventory/heartbeat_gap_gone_test.go @@ -0,0 +1,12 @@ +package inventory + +import "testing" + +// 1183/AC3: the heartbeat-gap duration is no kind the mesh measures. +func TestNoHeartbeatGapIsMeasured(t *testing.T) { // 1183/AC3 + for _, k := range DurationKinds { + if k == "heartbeat-gap" { + t.Fatalf("%s is still a kind of duration: %v", k, DurationKinds) + } + } +} diff --git a/internal/link/bus.go b/internal/link/bus.go index fb32ab03..fdb38c23 100644 --- a/internal/link/bus.go +++ b/internal/link/bus.go @@ -67,17 +67,12 @@ const ( // BuiltSubject is where a build's outcome lands, for results nobody was waiting for. BuiltSubject = "mesh.control.built" - // AliveSubjects is every node's heartbeat. Core NATS, never a stream: a lost heartbeat is the - // next heartbeat, and a stream of them is the mesh's least valuable message competing for - // retention with its most valuable (design 25 §3). - AliveSubjects = "mesh.control.*.alive" - // ToolsAliveSubjects is every machine's node tools saying they are there (novox/hq to-be 45 §3, // S11): core NATS like the host's, for the same reason. ToolsAliveSubjects = "mesh.control.*.tools-alive" // HealthSubjects is every machine's health statement between its reports (novox/hq ADR 0240): core - // NATS like the heartbeat, because a statement lost is said again within a minute while anything is + // NATS like the node tools' heartbeat, because a statement lost is said again within a minute while anything is // not healthy, and the next report carries it whatever happens. HealthSubjects = "mesh.control.*.health" ) @@ -89,9 +84,6 @@ func HealthSubject(node string) string { return "mesh.control." + node + ".healt // message the store-window guarantee is about (ADR 0083). func ReportSubject(node string) string { return "mesh.control." + node + ".report" } -// AliveSubject is one node's heartbeat. -func AliveSubject(node string) string { return "mesh.control." + node + ".alive" } - // ToolsAliveSubject is one machine's node tools' heartbeat. func ToolsAliveSubject(node string) string { return "mesh.control." + node + ".tools-alive" } diff --git a/internal/link/contracts.go b/internal/link/contracts.go index d5b7aff8..3e3f4752 100644 --- a/internal/link/contracts.go +++ b/internal/link/contracts.go @@ -32,8 +32,6 @@ var Contracts = map[string]Contract{ KindSourceMoved: {Ordered: "by the merge's commit against what was built from it: a merge already acted on " + "or older than the last look is history, not a second plan (issues 250, 266)", Tests: []string{"TestAMergeOlderThanTheLastLookIsHistory", "TestAMergeMatchesTheSourcesBuiltFromIt"}}, - KindHeartbeat: {Unordered: "a word that the machine is there: the newest heard is the newest said, and one " + - "lost is the next one"}, KindToolsHeartbeat: {Unordered: "a word that the node tools are there, as a machine's heartbeat"}, KindHealth: {Ordered: "by the time the node-engine looked (`at`, on the machine's clock): a statement older " + "than the one kept for that machine is refused, so an event arriving after a newer report does not undo it " + diff --git a/internal/link/contracts_test.go b/internal/link/contracts_test.go index 4fde2f1c..b05e2c0c 100644 --- a/internal/link/contracts_test.go +++ b/internal/link/contracts_test.go @@ -15,7 +15,7 @@ import ( // exists (novox/hq to-be 45 Phase 2, ADR 0227 rule 2). func TestEveryConsumedKindHasAContract(t *testing.T) { // What the controller can be handed, from the subjects it is granted and the ones it derives. - subjects := []string{EnrolSubject, BuiltSubject, ReportSubject("anchor"), AliveSubject("anchor"), + subjects := []string{EnrolSubject, BuiltSubject, ReportSubject("anchor"), ToolsAliveSubject("anchor"), HealthSubject("anchor"), "mesh.mod.postgres.event.provisioner.failing"} subjects = append(subjects, broker.ControllerFollows...) subjects = append(subjects, broker.ControllerHears...) @@ -83,3 +83,17 @@ func TestEveryConsumedKindHasAContract(t *testing.T) { } } } + +// 1183/AC3: the controller no longer takes the node-engine's alive message: its subject is no kind, so nothing is +// subscribed to it and nothing hears it. The tool runner's tools-alive stays. +func TestTheAliveMessageIsNoKindTheControllerHears(t *testing.T) { // 1183/AC3 + if kind, known := kindOfSubject("mesh.control.anchor.alive"); known { + t.Fatalf("the controller still takes the alive message as %s", kind) + } + if kind, known := kindOfSubject(ToolsAliveSubject("anchor")); !known || kind != KindToolsHeartbeat { + t.Fatalf("the tool runner's tools-alive is %q, %v", kind, known) + } + if _, ok := Contracts["heartbeat"]; ok { + t.Fatal("a contract still names the alive message") + } +} diff --git a/internal/link/enrolment.go b/internal/link/enrolment.go index f9b6ee84..845d9c01 100644 --- a/internal/link/enrolment.go +++ b/internal/link/enrolment.go @@ -384,9 +384,8 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err } // A bare word that a node is there is not an account of what the machine did or holds: it - // moves last_seen and touches nothing else. This arrives every minute (link.AliveEvery), - // while a real report is rare, so recording it as one would overwrite the node's last real - // apply with an empty one — wiping the declaration digest that decides whether the node is + // moves last_seen and touches nothing else. Recording one as a report would overwrite the + // node's last real apply with an empty one — wiping the declaration digest that decides whether the node is // current, the carried ports a push assigns around, and the clean-or-failed outcome — and a // node that had just caught up would read as behind within the minute. The alive path calls // this with only a node name; a real report always carries an account (something applied, or @@ -394,10 +393,6 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err if report.Applied == nil && report.Refused == "" && len(report.Failed) == 0 { if report.Superseded != "" { log.Printf("%s set aside declaration %s for the newer %s", report.Node, report.Declared, report.Superseded) - } else if err := e.Inventory.RecordHeartbeatGap(ctx, report.Node, node.LastSeen); err != nil { - // The silence before this word, which the machine-silent bound will be set from (novox/hq - // to-be 45 Phase 0). A measurement lost is said and costs the report nothing. - log.Printf("the silence before %s's word could not be recorded: %v", report.Node, err) } return false, e.Inventory.Seen(ctx, node.ID) } diff --git a/internal/link/protocol.go b/internal/link/protocol.go index 0687ee60..92177745 100644 --- a/internal/link/protocol.go +++ b/internal/link/protocol.go @@ -27,7 +27,6 @@ const ControlQueue = "control" const ( KeyEnrol = "enrol" KeyReport = "report" - KeyAlive = "alive" ) // QueueFor is the queue a node consumes from — the only one it may read. @@ -134,19 +133,6 @@ type Signed struct { Signature []byte `json:"signature"` } -// Alive is a node saying nothing except that it is there. -// -// How long a node has been out of touch is a fact only the mesh can hold — nobody else is -// watching — and without it a node running last month's assignments looks exactly like one that -// is current. -type Alive struct { - Node string `json:"node"` - // IntervalSeconds is how often the node says it is there (novox/hq to-be 45 §3, S1): the - // watchdog's bound is three of them. Zero from a host older than that, which is read as the - // interval hosts have always used. - IntervalSeconds int `json:"interval_seconds,omitempty"` -} - // ToolsAlive is a machine's node tools saying they are there (novox/hq to-be 45 §3, S11): the // runtime every module's tools and every held seat's verbs are served by. Its own word, apart from the // host's, because a host heard and a runtime gone is a machine nobody can ask anything. diff --git a/internal/link/receive.go b/internal/link/receive.go index 1585dc20..30e053cf 100644 --- a/internal/link/receive.go +++ b/internal/link/receive.go @@ -27,7 +27,6 @@ import ( const ( KindEnrolment = "enrolment" KindReport = "report" - KindHeartbeat = "heartbeat" // KindToolsHeartbeat is a machine's node tools saying they are there (novox/hq to-be 45 S11). KindToolsHeartbeat = "tools-heartbeat" // KindHealth is a machine's statement of its long-running resources' health (novox/hq ADR 0240). diff --git a/internal/link/receive_nats.go b/internal/link/receive_nats.go index 26c32238..31d6cc00 100644 --- a/internal/link/receive_nats.go +++ b/internal/link/receive_nats.go @@ -114,12 +114,8 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con // Heartbeats, on core NATS and off any stream (design 25 §3). Their own subscription because // they are their own guarantee: a lost one is the next one. beats := make(chan *nats.Msg, Prefetch) - alive, err := conn.ChanSubscribe(AliveSubjects, beats) - if err != nil { - return fmt.Errorf("subscribing to heartbeats: %w", err) - } - defer func() { _ = alive.Unsubscribe() }() - // And each machine's node tools, on the same channel: as cheap, and lost the same way. + // Each machine's node tools, as cheap and lost the same way. The node-engine's alive message is not subscribed to: a + // machine's last word is the watchman's presence (novox/hq ADR 0433 §3, issue 1183). toolsAlive, err := conn.ChanSubscribe(ToolsAliveSubjects, beats) if err != nil { return fmt.Errorf("subscribing to the node tools' heartbeats: %w", err) @@ -283,8 +279,6 @@ func kindOfSubject(subject string) (string, bool) { switch rest { case "report": return KindReport, true - case "alive": - return KindHeartbeat, true case "tools-alive": return KindToolsHeartbeat, true case "health": diff --git a/internal/link/receive_nats_integration_test.go b/internal/link/receive_nats_integration_test.go index 7806ef70..59c89e1d 100644 --- a/internal/link/receive_nats_integration_test.go +++ b/internal/link/receive_nats_integration_test.go @@ -157,24 +157,31 @@ func TestNatsAnEnrolmentIsAnsweredOnTheAddressInItsPayload(t *testing.T) { } } -func TestNatsAHeartbeatIsHeardAndNothingIsKept(t *testing.T) { +// 1183/AC3, AC5: a node-engine still sending its alive message is harmless: it errors nothing, nothing is recorded and +// nothing is persisted; the tool runner's tools-alive is still heard. +func TestNatsAnAliveMessageIsHarmlessAndToolsAliveIsStillHeard(t *testing.T) { // 1183/AC3 js := aBus(t) store := &counted{} _, stop := servingOn(t, js, store) defer stop() - // Given time to subscribe: a core subscription that is not yet up misses what is published, - // which is the guarantee a heartbeat has and not a fault. - eventually(t, "the heartbeat subscription coming up", func() bool { - body, _ := json.Marshal(Alive{Node: "anchor"}) - _ = js.Conn().Publish(AliveSubject("anchor"), body) + // Given time to subscribe: a core subscription that is not yet up misses what is published. + eventually(t, "the tools-alive subscription coming up", func() bool { + if err := js.Conn().Publish("mesh.control.anchor.alive", []byte(`{"node":"anchor"}`)); err != nil { + t.Fatalf("an alive message could not be published: %v", err) + } + body, _ := json.Marshal(ToolsAlive{Node: "anchor"}) + _ = js.Conn().Publish(ToolsAliveSubject("anchor"), body) _ = js.Conn().Flush() - return store.count() >= 1 + _, heard := ToolsBeats.Of("anchor") + return heard }) + if n := store.count(); n != 0 { + t.Fatalf("an alive message was recorded %d time(s)", n) + } info, err := js.Context().StreamInfo("CONTROL") if err != nil || info.State.Msgs != 0 { - t.Fatalf("a heartbeat was persisted, and the mesh's least valuable message now competes "+ - "for retention with its most valuable: %+v, %v", info, err) + t.Fatalf("an alive message was persisted: %+v, %v", info, err) } } diff --git a/internal/link/receive_nats_test.go b/internal/link/receive_nats_test.go index 382bceac..0f7d9ce3 100644 --- a/internal/link/receive_nats_test.go +++ b/internal/link/receive_nats_test.go @@ -125,9 +125,6 @@ type enrolsAs struct{ reply EnrolReply } func (e enrolsAs) Enrol(context.Context, EnrolRequest) (EnrolReply, error) { return e.reply, nil } -// A heartbeat is core NATS: it reaches the controller and nothing is persisted, so the stream the -// reports live in stays empty. - // The two events the controller follows arrive over one durable consumer with two filters, and it // can acknowledge them. // diff --git a/internal/link/serve.go b/internal/link/serve.go index 819b5648..3eaabcfe 100644 --- a/internal/link/serve.go +++ b/internal/link/serve.go @@ -205,8 +205,8 @@ func (s *Server) Close() { // Serve acts on what arrives until the context ends. func (s *Server) Serve(ctx context.Context) error { - s.log.Printf("consuming what nodes say: %s, %s, %s, %s", - KindEnrolment, KindReport, KindHeartbeat, KindBuilt) + s.log.Printf("consuming what nodes say: %s, %s, %s", + KindEnrolment, KindReport, KindBuilt) if s.upgrader != nil { s.log.Printf("following %s", KindModuleMoved) } @@ -226,8 +226,6 @@ func (s *Server) act(ctx context.Context, m Control) { s.enrolling(ctx, m) case KindReport: s.reported(ctx, m) - case KindHeartbeat: - s.heartbeat(m) case KindToolsHeartbeat: s.toolsHeartbeat(m) case KindHealth: @@ -324,28 +322,6 @@ func (s *Server) outstanding(ctx context.Context, node string) string { return digest } -// heartbeat records that a node was heard from, and nothing else. -// -// Deliberately silent: a node saying it is there every minute would fill the log with the ordinary -// case, and a log where the ordinary case is loud is a log nobody reads. Not held for the store -// either — the next heartbeat is a minute away, and a heartbeat kept for two minutes to be written -// late says nothing the one after it will not say better. -func (s *Server) heartbeat(m Control) { - var alive Alive - if err := json.Unmarshal(m.Body(), &alive); err != nil || alive.Node == "" { - _ = m.Drop() - return - } - // The interval it says, for the watchdog's bound (novox/hq to-be 45 S1). - HostBeats.Heard(alive.Node, time.Now(), time.Duration(alive.IntervalSeconds)*time.Second, "") - if s.listener != nil { - if _, err := s.listener.Heard(context.Background(), Report{Node: alive.Node}); err != nil { - s.log.Printf("could not record that %s is here: %v", alive.Node, err) - } - } - _ = m.Took() -} - // toolsHeartbeat records that a machine's node tools were heard from (novox/hq to-be 45 S11), and // nothing else: in this process's memory, for the watchdog — the next one is a minute away. func (s *Server) toolsHeartbeat(m Control) { diff --git a/internal/link/stale_report_test.go b/internal/link/stale_report_test.go index dcf69fb3..cf9f303b 100644 --- a/internal/link/stale_report_test.go +++ b/internal/link/stale_report_test.go @@ -121,15 +121,3 @@ func TestAReportCarryingWhatOnlyTheNodeKnowsIsActedOnHoweverOldItIs(t *testing.T } } } - -// A heartbeat is not held for the store: the next one is a minute away, and one kept for two -// minutes to be written late says nothing the one after it will not say better. -func TestAHeartbeatIsNotHeldForTheStore(t *testing.T) { - s, in := serving() - s.listener = heardWith{err: errors.Join(ErrTryAgain, errors.New("starting up"))} - to := &settled{} - s.act(context.Background(), in.sends(t, to, KindHeartbeat, Alive{Node: "anchor"})) - if !to.acked || len(in.held) != 0 { - t.Fatalf("a heartbeat was held for the store: %+v, %d held", *to, len(in.held)) - } -} diff --git a/internal/link/watched.go b/internal/link/watched.go index f21206bb..38bb8e5c 100644 --- a/internal/link/watched.go +++ b/internal/link/watched.go @@ -21,8 +21,7 @@ import ( // // **In memory, on purpose.** A heartbeat is the least valuable message the mesh sends — the next one // is a minute away — and what a watchdog needs of it is the newest moment and the interval it said. -// A machine's last word is kept in the store as well (`last_seen`), which is what S1 reads; these are -// what the store does not keep: the interval, the node tools' word, and when the event loop last took +// What the store does not keep: the node tools' word and its interval, and when the event loop last took // a message. // Beat is one emitter's newest heartbeat. @@ -43,11 +42,9 @@ type Beats struct { // NewBeats is an empty record, started now. func NewBeats() *Beats { return &Beats{started: time.Now(), heard: map[string]Beat{}} } -// HostBeats are the node-engines' heartbeats (S1); ToolsBeats the node tools' (S11). -var ( - HostBeats = NewBeats() - ToolsBeats = NewBeats() -) +// ToolsBeats are the node tools' heartbeats (S11). The node-engine's alive message is not heard: a machine's last word is +// the watchman's presence (novox/hq ADR 0433 §3). +var ToolsBeats = NewBeats() // Heard records one heartbeat. func (b *Beats) Heard(node string, at time.Time, every time.Duration, version string) { -- 2.54.0