The controller takes what the watchman says: key-ttl buckets, the checks running readable, stuck keys raised, silence from presence (design 47 R2b) #388

Merged
mesh-tools merged 7 commits from feat/delivery-watch into main 2026-10-18 23:34:08 +00:00
9 changed files with 430 additions and 5 deletions
Showing only changes of commit 7be2df627f - Show all commits
+1 -1
View File
@@ -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
}
}
+5
View File
@@ -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 " +
+7
View File
@@ -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. " +
+1 -1
View File
@@ -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
}
}
+2 -2
View File
@@ -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)
+208
View File
@@ -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.<bucket>.<key>` 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.<node> 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.<node>'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
}
@@ -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")
}
}
+108
View File
@@ -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], &notMoved, 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)
}
}
+16 -1
View File
@@ -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.<node>'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.