Watchman: observe S1, S9 and S21 as the controller does and stop the needs-operator row, so the keeper's compare can empty with no condition lost (issue 1952 WM2, WM3, WM5)

S1 is urgent after half an hour for a control node (the node running mesh-controller). S9 also hears the bus module's
slow consumers and closed connections, each closure counted once. S21 is one finding per heard machine from its failed
haves and its provisioner clients' reports (PROV_APPLIED). DEAD_LETTERS is counted again at a letter's age and every
five minutes while letters are held, on the watchman's own pulse. needs-operator waits for L6: its inputs are not on the
bus. Pins mesh-sdk b6ef01d.

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go

@@ -298,8 +298,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	dead := &deadKeeper{ctx: ctx, js: js, w: w, logf: logger.Printf}
+	w.SetRecount(dead.count)
 	dead.count()
+	dead.ages()
 	dead.check()
+	provisioned(ctx, js, w, logger.Printf)
 	w.Reconcile()
 	if err := hear(conn, w, dead, logger.Printf); err != nil {
 		return err

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/signals.go

@@ -36,11 +36,23 @@ const (
 	routeChecked  = "checked"
 	routeAdvisory = "advisory"
 	routeDead     = "dead-letter"
+	routeBusFault = "bus-fault"
+	routeProv     = "provisioned"
+)
+
+// The bus module's events of the connections the bus closed (S9).
+const (
+	SlowConsumerSubject     = "mesh.mod.nats.event.bus.slow-consumer"
+	ConnectionClosedSubject = "mesh.mod.nats.event.bus.connection-closed"
 )
 
 // routeOf is the row a heard subject goes to; empty for one the watchman does not read.
 func routeOf(subject string) string {
 	switch {
+	case subject == SlowConsumerSubject || subject == ConnectionClosedSubject:
+		return routeBusFault
+	case strings.HasPrefix(subject, "prov.applied."):
+		return routeProv
 	case strings.HasPrefix(subject, "mesh.mod.power.event."):
 		return routePower
 	case strings.HasSuffix(subject, ".event.provisioner.failing") || strings.HasSuffix(subject, ".event.provisioner.recovered"):
@@ -87,7 +99,30 @@ func heard(w *watch.Watchman, dead *deadKeeper, m *nats.Msg, logf func(string, .
 			return
 		}
 		w.Advised(a)
+	case routeBusFault:
+		var c watch.BusClosure
+		if err := json.Unmarshal(m.Data, &c); err != nil {
+			logf("the bus module's word on %s cannot be read, left: %v", m.Subject, err)
+			return
+		}
+		kind := watch.KindSlowConsumer
+		if m.Subject == ConnectionClosedSubject {
+			kind = watch.KindConnectionClosed
+		}
+		w.Closed(kind, c)
+	case routeProv:
+		var r watch.ProvReport
+		if err := json.Unmarshal(m.Data, &r); err != nil {
+			logf("the provisioner client's report on %s cannot be read, left: %v", m.Subject, err)
+			return
+		}
+		w.Provisioned(r)
 	case routeDead:
+		if stream, consumer, ok := buscompose.DeadLetterOf(m.Subject); ok {
+			if at, err := time.Parse(time.RFC3339Nano, m.Header.Get(deadletters.HeaderAgesOut)); err == nil {
+				w.DeadAging(stream+"."+consumer, at)
+			}
+		}
 		dead.count()
 	}
 }
@@ -115,6 +150,44 @@ type deadKeeper struct {
 	taking     jetstream.ConsumeContext
 }
 
+// ages pulses each consumer's restricted letters by when they age out, read from DEAD_LETTERS' headers (at the start and
+// at resync; each letter kept later is pulsed as it is heard).
+func (d *deadKeeper) ages() {
+	letters, _, err := deadletters.List(d.ctx, d.js, "", 0)
+	if err != nil {
+		d.logf("the dead letters' ages cannot be read, so one aging out ends its finding only at the next count: %v", err)
+		return
+	}
+	for _, l := range letters {
+		if l.Restricted && !l.AgesOut.IsZero() {
+			d.w.DeadAging(l.Stream+"."+l.Consumer, l.AgesOut)
+		}
+	}
+}
+
+// provisioned reads each provisioner client's newest report kept by PROV_APPLIED (as the controller's
+// ProvAppliedReports): S21 judges what stood at the start as what is said later.
+func provisioned(ctx context.Context, js jetstream.JetStream, w *watch.Watchman, logf func(string, ...any)) {
+	s, err := js.Stream(ctx, buscompose.ProvAppliedStream)
+	if err == nil {
+		var info *jetstream.StreamInfo
+		if info, err = s.Info(ctx, jetstream.WithSubjectFilter(buscompose.ProvAppliedSubjects)); err == nil {
+			for subject := range info.State.Subjects {
+				msg, gerr := s.GetLastMsgForSubject(ctx, subject)
+				if gerr != nil {
+					continue // aged out between the two reads
+				}
+				var r watch.ProvReport
+				if json.Unmarshal(msg.Data, &r) == nil {
+					w.Provisioned(r)
+				}
+			}
+			return
+		}
+	}
+	logf("the provisioner clients' reports cannot be read (%v): S21 hears only those said from now", err)
+}
+
 // count reads DEAD_LETTERS' count per consumer into the findings.
 func (d *deadKeeper) count() {
 	held, err := deadletters.Held(d.ctx, d.js)
@@ -180,6 +253,7 @@ func answer(a socket.Ask, w *watch.Watchman, dead *deadKeeper, ask func([]byte)
 			return nil, err
 		}
 		dead.count()
+		dead.ages()
 		dead.check()
 		return map[string]any{"findings-written": w.Rewrite(), "presence": "rebuilt from the bus's connections"}, nil
 	case socket.AskDeadLetters:

modules/mesh-delivery-watchman/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-watchman
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0
 )

modules/mesh-delivery-watchman/go.sum

@@ -1,5 +1,7 @@
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o=
 git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7 h1:/XEAhD7/8QoMjqI7XiX9HM4HkpKubWheQosqgfe2WrU=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029205750-b6ef01d6c6a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
 github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
 github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=

modules/mesh-delivery-watchman/internal/watch/dead.go

@@ -4,12 +4,18 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 )
 
 // The dead letters' findings (novox/hq ADR 0564 §5, issue 330): a consumer DEAD_LETTERS holds a message for is said for as
 // long as it holds any, `bus.<stream>.<consumer>.max-deliveries`, and the notices waiting behind the controller's consumer
 // after the lease write are said as `bus.DEAD_LETTER_NOTICES.controller.notices-waiting`.
 
+// DeadRecountEvery is how often DEAD_LETTERS is counted again while a consumer's finding stands: a letter delivered again
+// or dropped by the controller's verb, which the watchman does not hear, ends its finding within it. The expiry of the
+// watchman's own pulse, never a loop.
+const DeadRecountEvery = 5 * time.Minute
+
 // KindNoticesWaiting is the finding of notices waiting while the controller's consumer stands after the lease write.
 const KindNoticesWaiting = "notices-waiting"
 
@@ -38,7 +44,8 @@ func (w *Watchman) DeadLettersHeld(held map[string]int, who func(stream, consume
 			Machine: consumerMachine(stream, consumer),
 			Summary: fmt.Sprintf("%s gave up on %s; %s kept in DEAD_LETTERS until delivered again or dropped, with why",
 				who(stream, consumer), messages, them),
-			Said:  fmt.Sprintf("%d held for %s", n, key),
+			Said: fmt.Sprintf("%d held for %s (counted again every %s while any is held, so one delivered again or dropped "+
+				"ends this within it)", n, key, DeadRecountEvery),
 			Needs: "deliver them again or drop them, from the mesh MCP server."}
 		want[f.Key()] = true
 		w.find(f)
@@ -48,6 +55,12 @@ func (w *Watchman) DeadLettersHeld(held map[string]int, who func(stream, consume
 			w.unfind(key)
 		}
 	}
+	if len(want) == 0 {
+		w.unsignal("dead", "recount")
+		return
+	}
+	w.signal("dead", "recount", DeadRecountEvery, signalDue{Do: dueRecount,
+		Finding: Finding{Scope: ScopeBus, ID: "DEAD_LETTERS", Kind: KindDeadLetters, Severity: Warning, Summary: "counted again", Source: "S9"}})
 }
 
 // consumerMachine is the machine a module's consumer on EVENTS is on; empty for the others.

modules/mesh-delivery-watchman/internal/watch/findings_test.go

@@ -121,16 +121,26 @@ func TestTheWatchmansFindings(t *testing.T) {
 			}
 			expire(w, c, signalKey("S9", k), AdvisoryQuiet)
 		}, map[string]string{}},
-		{"WM3 S21: a have failed for one apply interval is apply-failing; applied again it ends", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+		{"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)
-			expire(w, c, signalKey("S21", "ace.web"), ApplyEvery)
-			if f := o.findings["machine.ace.web.apply-failing"]; !strings.Contains(f.Said, "pacman") {
+			if len(o.findings) != 0 {
+				t.Fatalf("raised before its bound: %v", o.findings)
+			}
+			expire(w, c, signalKey("S21", "ace"), ApplyEvery+time.Second)
+			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.mail", haveOf(t, delivery.PhaseFailed, delivery.HealthUnknown, "refused"), Put)
-			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthOK, ""), Put)
-			expire(w, c, signalKey("S21", "ace.mail"), ApplyEvery) // applied before its bound: nothing
+		}, 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")
+			w.Provisioned(ProvReport{Node: "ace", Module: "web", Status: ProvFailed, Reason: "disk full", At: t0.Add(-10 * time.Minute)})
+			if f := o.findings["machine.ace.apply-failing"]; !strings.Contains(f.Summary, "provisioner has failed to place the files of web") {
+				t.Fatalf("findings = %v", o.findings)
+			}
+			w.Provisioned(ProvReport{Node: "g14", Module: "web", Status: ProvFailed, Reason: "x", At: t0.Add(-10 * time.Minute)})
+			w.Provisioned(ProvReport{Node: "ace", Module: "web", Status: "applied", At: t0})
 		}, map[string]string{}},
 		{"WM3: unhealthy only at the second write in a row, 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)
@@ -142,19 +152,63 @@ func TestTheWatchmansFindings(t *testing.T) {
 				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)
-		}, map[string]string{"module.web.ace.unhealthy": Urgent, "module.mail.ace.needs-operator": Warning}},
+			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.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 have removed ends its findings", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
-			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthWaits, "waits"), Put)
+			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
+			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
 			w.Have("ace.mail", haveOf(t, delivery.PhaseRemoved, delivery.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) {
+			w.Have("novox.mesh-controller", haveOf(t, delivery.PhaseRunning, delivery.HealthOK, ""), Put)
+			w.Connect("node.novox")
+			w.Disconnect("node.novox")
+			expire(w, c, pulseKeyOf(t, presenceNS, "novox"), 120*time.Second)
+			if o.findings["machine.novox.silent"].Severity != Warning {
+				t.Fatalf("findings = %v", o.findings)
+			}
+			expire(w, c, signalKey("S1", "novox"), ControlNodeUrgentAfter)
+		}, map[string]string{"machine.novox.silent": Urgent}},
+		{"WM3 S9: the bus module's closures, each counted once, ended an hour after the newest", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			cl := BusClosure{User: "ace.web", CID: 7, Reason: "Slow Consumer (Write Deadline)", Opened: t0.Add(-time.Hour), Closed: t0}
+			w.Closed(KindSlowConsumer, cl)
+			w.Closed(KindSlowConsumer, cl) // said again when the bus module starts
+			cl.CID = 8
+			w.Closed(KindSlowConsumer, cl)
+			f := o.findings["bus.ace.web.slow-consumer"]
+			if f.Machine != "ace" || !strings.Contains(f.Summary, "web on ace") || !strings.Contains(f.Summary, "2 times") {
+				t.Fatalf("findings = %v", o.findings)
+			}
+			expire(w, c, signalKey("S9", "bus.ace.web.slow-consumer"), AdvisoryQuiet)
+		}, map[string]string{}},
+		{"WM5: while letters are held DEAD_LETTERS is counted again at its own pulse's expiry, and at a letter's age", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			n := 0
+			w.SetRecount(func() { n++ })
+			w.DeadLettersHeld(map[string]int{"EVENTS.ace_web": 1}, func(s, c string) string { return c })
+			if o.ttls[signalKey("dead", "recount")] != DeadRecountEvery || !strings.Contains(o.findings["bus.EVENTS.ace_web.max-deliveries"].Said, "every 5m0s") {
+				t.Fatalf("pulses %v findings %v", o.ttls, o.findings)
+			}
+			w.DeadAging("EVENTS.ace_web", c.t.Add(time.Minute))
+			expire(w, c, signalKey("dead", "EVENTS.ace_web"), time.Minute)
+			expire(w, c, signalKey("dead", "recount"), DeadRecountEvery)
+			time.Sleep(10 * time.Millisecond)
+			if n != 2 {
+				t.Fatalf("counted again %d times", n)
+			}
+			w.DeadLettersHeld(map[string]int{"EVENTS.ace_web": 1}, func(s, c string) string { return c }) // the count renews its pulse
+			w.DeadLettersHeld(map[string]int{}, nil)
+			if _, ok := o.pulses[signalKey("dead", "recount")]; ok {
+				t.Fatal("still counted again with nothing held")
+			}
+		}, 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.HealthWaits, "waits"), Put)
+			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
+			w.Have("ace.mail", haveOf(t, delivery.PhaseRunning, delivery.HealthFailed, "x"), Put)
 			o.findings = map[string]Finding{}
 			if n := w.Rewrite(); n != 1 {
 				t.Fatalf("rewrote %d", n)
 			}
-		}, map[string]string{"module.mail.ace.needs-operator": Warning}},
+		}, map[string]string{"module.mail.ace.unhealthy": Warning}},
 		{"WM4: a finding read at the start and not observed again is deleted; a signal's stays", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
 			for _, f := range []Finding{
 				{Scope: ScopeDelivery, ID: "x.y", Kind: KindStuck, Severity: Warning, Summary: "old", Source: "stuck"},

modules/mesh-delivery-watchman/internal/watch/presence.go

@@ -87,6 +87,9 @@ 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 != "" {
+		defer w.applyCheck(node) // S21 is judged only while the machine is heard
+	}
 	if online {
 		w.moved(delivery.PresenceBucket, key)
 		return

modules/mesh-delivery-watchman/internal/watch/signals.go

@@ -44,6 +44,10 @@ const (
 	ApplyEvery = 5 * time.Minute
 	// UnhealthyUrgentAfter is how long a module stays unhealthy before it is urgent.
 	UnhealthyUrgentAfter = 4 * time.Hour
+	// SilentBound is the bound S1 says, three of the intervals a machine has always used, and ControlNodeUrgentAfter
+	// how long a control node may be silent before it is urgent (the controller's silentBound, controlNodeUrgentAfter).
+	SilentBound            = 3 * HeartbeatEvery
+	ControlNodeUrgentAfter = 30 * time.Minute
 )
 
 // signalBucket is the pseudo-bucket of a signal's pulse key.
@@ -62,6 +66,9 @@ type signalDue struct {
 const (
 	dueFind = "find"
 	dueEnd  = "end"
+	// dueApply judges a machine's S21 again (applyCheck); dueRecount counts DEAD_LETTERS again (a restricted letter aged out).
+	dueApply   = "apply"
+	dueRecount = "recount"
 )
 
 // signalKey is the pulse key of a row's signal about id.
@@ -127,7 +134,7 @@ func dueOf(value []byte) (signalDue, bool) {
 		return signalDue{}, false
 	}
 	var due signalDue
-	if json.Unmarshal([]byte(p.Owes), &due) != nil || (due.Do != dueFind && due.Do != dueEnd) {
+	if json.Unmarshal([]byte(p.Owes), &due) != nil || (due.Do != dueFind && due.Do != dueEnd && due.Do != dueApply && due.Do != dueRecount) {
 		return signalDue{}, false
 	}
 	return due, true
@@ -141,6 +148,12 @@ func (w *Watchman) signalExpired(pk string) {
 	}
 	delete(w.signals, pk)
 	switch due.Do {
+	case dueApply:
+		w.applyCheck(due.While)
+	case dueRecount:
+		if w.recount != nil {
+			go w.recount()
+		}
 	case dueEnd:
 		w.unfind(due.Finding.Key())
 	case dueFind:
@@ -154,6 +167,10 @@ func (w *Watchman) signalExpired(pk string) {
 // stillWrong says a have's key still says what a row's finding was waited on: failed for apply-failing, unhealthy for
 // unhealthy.
 func (w *Watchman) stillWrong(haveKey, kind string) bool {
+	if node, ok := strings.CutPrefix(haveKey, "silent:"); ok {
+		_, open := w.findings[FindingKey(ScopeMachine, node, KindSilent)]
+		return open
+	}
 	h, ok := w.haveState[haveKey]
 	if !ok {
 		return false
@@ -202,8 +219,10 @@ func (w *Watchman) Power(node, state string) {
 	if err != nil {
 		return
 	}
+	defer w.applyCheck(node)
 	if away {
 		w.unfind(FindingKey(ScopeMachine, node, KindSilent))
+		w.unsignal("S1", node)
 		return
 	}
 	if p, ok := w.presence[key]; ok && !p.online {
@@ -222,11 +241,26 @@ func (w *Watchman) stuckFinding(sk string, s delivery.Stuck) {
 			key := FindingKey(ScopeMachine, node, KindSilent)
 			if s.Moved || w.away[node] {
 				w.unfind(key)
+				w.unsignal("S1", node)
+				w.applyCheck(node)
 				return
 			}
-			w.find(Finding{Scope: ScopeMachine, ID: node, Kind: KindSilent, Machine: node, Severity: Warning, Source: "S1",
-				Summary: fmt.Sprintf("%s has not been heard from since %s (bound %s)", node, stamp(s.Since), delivery.PulseMoveBound),
-				Said:    fmt.Sprintf("offline since %s", s.Since.UTC().Format(time.RFC3339)), Since: s.Since})
+			f := Finding{Scope: ScopeMachine, ID: node, Kind: KindSilent, Machine: node, Severity: Warning, Source: "S1",
+				Summary: fmt.Sprintf("%s has not been heard from since %s (bound %s)", node, stamp(s.Since), SilentBound),
+				Said:    fmt.Sprintf("offline since %s", s.Since.UTC().Format(time.RFC3339)), Since: s.Since}
+			// A control node, the machine running the controller, is urgent after half an hour silent, as the controller
+			// says it (its watchHeartbeats): the expiry of a pulse from its silence's start, never a loop.
+			if w.control[node] {
+				if age := w.now().Sub(s.Since); age > ControlNodeUrgentAfter {
+					f.Severity = Urgent
+				} else {
+					u := f
+					u.Severity = Urgent
+					w.signal("S1", node, ControlNodeUrgentAfter-age+time.Second, signalDue{Do: dueFind, While: "silent:" + node, Finding: u})
+				}
+			}
+			w.find(f)
+			w.applyCheck(node)
 			return
 		}
 	}
@@ -418,42 +452,41 @@ func ToolsNode(subject string) string {
 	return node
 }
 
-// ---- S21, unhealthy and needs-operator, from the haves ----
+// ---- S21 and unhealthy, from the haves and the provisioner clients' reports ----
 
-// haveFindings reads a have's change for the rows the haves feed: S21 (failed for one apply interval), unhealthy (two
-// writes in a row, urgent after four hours) and needs-operator (it waits for the operator). A have removed or deleted ends
-// them all.
+// haveFindings reads a have's change for the rows the haves feed: S21 (the machine's applies failing, applyCheck),
+// unhealthy (two writes in a row, urgent after four hours), and where the controller runs (S1's control node). A have
+// removed or deleted ends its part of them.
+//
+// needs-operator is not observed here: the controller excuses a wait only after checking it against the module's manifest
+// and against whether a person gave the secret on that machine (its operator_wait.go), and neither each resource's wait nor
+// what a person gave is on the bus (the have carries the node's words only). The row waits for those facts' home.
 func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Have) {
-	applyKey := FindingKey(ScopeMachine, node+"."+module, KindApplyFailing)
 	unhealthyKey := FindingKey(ScopeModule, module+"."+node, "unhealthy")
-	waitsKey := FindingKey(ScopeModule, module+"."+node, KindNeedsOperator)
+	if module == ControllerModule {
+		w.control[node] = h != nil && h.Phase != delivery.PhaseRemoved
+	}
 	if h == nil || h.Phase == delivery.PhaseRemoved {
 		delete(w.haveState, key)
 		delete(w.unhealthy, key)
-		w.unsignal("S21", node+"."+module)
+		delete(w.failedSince, key)
 		w.unsignal("unhealthy", module+"."+node)
-		w.unfind(applyKey)
 		w.unfind(unhealthyKey)
-		w.unfind(waitsKey)
+		w.applyCheck(node)
 		return
 	}
-	was, had := w.haveState[key]
 	w.haveState[key] = *h
 	now := w.now()
 
 	// 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 !had || was.Phase != delivery.PhaseFailed {
-			w.signal("S21", node+"."+module, ApplyEvery, signalDue{Do: dueFind, While: key, Finding: Finding{Scope: ScopeMachine,
-				ID: node + "." + module, Kind: KindApplyFailing, Machine: node, Severity: Warning, Source: "S21", Since: now,
-				Summary:  fmt.Sprintf("%s has failed to apply %s since %s (bound %s)", node, module, stamp(now), ApplyEvery),
-				Said:     firstLine(orWords(h.Detail, "no reason given")),
-				Headline: node + " is not applying its update", Resolved: node + " applies what it is sent again"}})
+		if _, ok := w.failedSince[key]; !ok {
+			w.failedSince[key] = now
 		}
 	} else {
-		w.unsignal("S21", node+"."+module)
-		w.unfind(applyKey)
+		delete(w.failedSince, key)
 	}
+	w.applyCheck(node)
 
 	// unhealthy: two writes in a row that say it failed.
 	if h.Health == delivery.HealthFailed {
@@ -473,16 +506,268 @@ func (w *Watchman) haveFindings(key string, node, module string, h *delivery.Hav
 		w.unsignal("unhealthy", module+"."+node)
 		w.unfind(unhealthyKey)
 	}
+}
 
-	// needs-operator: a part waits for the operator's secret or setting.
-	if h.Health == delivery.HealthWaits {
-		w.find(Finding{Scope: ScopeModule, ID: module + "." + node, Kind: KindNeedsOperator, Machine: node, Severity: Warning,
-			Source: "health", Since: now,
-			Summary: fmt.Sprintf("%s on %s waits for the operator: %s", module, node, orWords(h.Detail, "a part waits for a secret or a setting")),
-			Said:    firstLine(orWords(h.Detail, "waits"))})
-	} else {
-		w.unfind(waitsKey)
+// ControllerModule is the module whose have names a control node.
+const ControllerModule = "mesh-controller"
+
+// ProvReport is a machine's provisioner client's word on one module's files (`prov.applied.<node>.<module>`, ADR 0362 §3).
+type ProvReport struct {
+	Node   string    `json:"node"`
+	Module string    `json:"module"`
+	Status string    `json:"status"`
+	Reason string    `json:"reason,omitempty"`
+	At     time.Time `json:"at"`
+}
+
+// ProvFailed is the status of a report whose files were not placed.
+const ProvFailed = "failed"
+
+// Provisioned is a provisioner client's newest report: a failure is kept for S21, any other word ends it.
+func (w *Watchman) Provisioned(r ProvReport) {
+	if r.Node == "" || r.Module == "" {
+		return
 	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.provFailed[r.Node] == nil {
+		w.provFailed[r.Node] = map[string]ProvReport{}
+	}
+	if r.Status == ProvFailed {
+		w.provFailed[r.Node][r.Module] = r
+	} else {
+		delete(w.provFailed[r.Node], r.Module)
+	}
+	w.applyCheck(r.Node)
+}
+
+// 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)
+	if err != nil || w.away[node] {
+		return false
+	}
+	p, ok := w.presence[key]
+	return ok && p.online
+}
+
+// applyCheck is S21 for one machine, `machine.<node>.apply-failing` (the controller's watchApplyFailing): raised once the
+// first failure it still has is older than one apply interval while the machine is heard, naming the modules its node
+// failed to apply (their haves failed), or else those whose files its provisioner client failed to place; ended by the
+// clean apply. Before the bound a pulse judges it again at the bound.
+func (w *Watchman) applyCheck(node string) {
+	key := FindingKey(ScopeMachine, node, KindApplyFailing)
+	type failure struct{ module, words string }
+	var applied, placed []failure
+	var since time.Time
+	for hk, at := range w.failedSince {
+		n, module, _, err := delivery.ParseHaveKey(hk)
+		if err != nil || n != node {
+			continue
+		}
+		applied = append(applied, failure{module, firstLine(orWords(w.haveState[hk].Detail, "no reason given"))})
+		if since.IsZero() || at.Before(since) {
+			since = at
+		}
+	}
+	if len(applied) == 0 {
+		for module, r := range w.provFailed[node] {
+			placed = append(placed, failure{module, firstLine(orWords(r.Reason, "no reason given"))})
+			if since.IsZero() || r.At.Before(since) {
+				since = r.At
+			}
+		}
+	}
+	if since.IsZero() || !w.heard(node) {
+		w.unsignal("S21", node)
+		w.unfind(key)
+		return
+	}
+	if age := w.now().Sub(since); age <= ApplyEvery {
+		if _, pending := w.signals[signalKey("S21", node)]; !pending {
+			w.signal("S21", node, ApplyEvery-age+time.Second, signalDue{Do: dueApply, While: node,
+				Finding: Finding{Scope: ScopeMachine, ID: node, Kind: KindApplyFailing, Severity: Warning, Summary: "judged again at its bound", Source: "S21"}})
+		}
+		return
+	}
+	failed := append(applied, placed...)
+	sort.Slice(failed, func(i, j int) bool { return failed[i].module < failed[j].module })
+	var names, words []string
+	tool := ""
+	for _, f := range failed {
+		names = append(names, f.module)
+		words = append(words, f.module+": "+f.words)
+		if rest, ok := strings.CutPrefix(f.words, "refused: "); ok {
+			f.words = rest
+		}
+		if i := strings.Index(f.words, PackageManagerKey); i >= 0 && tool == "" {
+			tool = strings.TrimSuffix(strings.SplitN(f.words[i+len(PackageManagerKey):], ")", 2)[0], ")")
+		}
+	}
+	f := Finding{Scope: ScopeMachine, ID: node, Kind: KindApplyFailing, Machine: node, Severity: Warning, Source: "S21", Since: since,
+		Said: strings.Join(words, "; "), Resolved: node + " applies what it is sent again", Headline: node + " is not applying its update"}
+	switch {
+	case len(applied) == 0:
+		f.Summary = fmt.Sprintf("%s's provisioner has failed to place the files of %s since %s (bound %s)", node,
+			strings.Join(names, ", "), stamp(since), ApplyEvery)
+		f.Explanation = fmt.Sprintf("The provisioner on %s has not been able to write the files of %s. The details say why.",
+			node, strings.Join(names, ", "))
+	case tool != "":
+		f.Summary = fmt.Sprintf("%s: %s does not answer, so no package reaches it; everything else it was sent is applied, "+
+			"since %s (bound %s)", node, tool, stamp(since), ApplyEvery)
+		f.Headline = node + " cannot install packages"
+	default:
+		f.Summary = fmt.Sprintf("%s has failed to apply %s since %s (bound %s)", node, strings.Join(names, ", "), stamp(since), ApplyEvery)
+		f.Explanation = fmt.Sprintf("%s has failed to apply its updates, and trying again has not helped. The details say what fails.", node)
+	}
+	w.find(f)
+}
+
+// PackageManagerKey is how a node-engine names a package manager that does not answer among what failed.
+const PackageManagerKey = "the package manager ("
+
+// ---- S9: the bus module's closures ----
+
+// BusClosure is one connection the bus closed, as the bus module says it (`bus.slow-consumer`, `bus.connection-closed`).
+type BusClosure struct {
+	User    string    `json:"user"`
+	Name    string    `json:"name,omitempty"`
+	CID     uint64    `json:"cid"`
+	Reason  string    `json:"reason"`
+	Opened  time.Time `json:"opened"`
+	Closed  time.Time `json:"closed"`
+	Pending int64     `json:"pending_bytes,omitempty"`
+}
+
+// The kinds of the bus module's closures, as conditions name them.
+const (
+	KindSlowConsumer     = "slow-consumer"
+	KindConnectionClosed = "connection-closed"
+)
+
+// Closed is one closure the bus module said (S9, the controller's BusClosures and busUserClosed): one finding per user and
+// kind, `bus.<user>.<kind>`, each closure counted once however often it is said, ended an hour after the newest.
+func (w *Watchman) Closed(kind string, c BusClosure) {
+	if c.User == "" || c.Closed.IsZero() || (kind != KindSlowConsumer && kind != KindConnectionClosed) {
+		return
+	}
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	f := Finding{Scope: ScopeBus, ID: c.User, Kind: kind, Severity: Warning, Source: "S9"}
+	key := f.Key()
+	if _, open := w.findings[key]; !open {
+		w.closures[key] = map[string]time.Time{}
+	}
+	identity := fmt.Sprintf("%s/%d/%s", c.User, c.CID, c.Opened.UTC().Format(time.RFC3339Nano))
+	if _, seen := w.closures[key][identity]; seen {
+		return
+	}
+	w.closures[key][identity] = c.Closed
+	var first, last time.Time
+	for _, at := range w.closures[key] {
+		if first.IsZero() || at.Before(first) {
+			first = at
+		}
+		if at.After(last) {
+			last = at
+		}
+	}
+	n := len(w.closures[key])
+	who, machine := BusUserWords(c.User)
+	f.Machine, f.Since = machine, first
+	times, count := "once", "once"
+	if n > 1 {
+		times = fmt.Sprintf("%d times since %s", n, first.UTC().Format("15:04 MST"))
+		count = fmt.Sprintf("%d times", n)
+	}
+	if kind == KindSlowConsumer {
+		f.Said = fmt.Sprintf("the bus closed %s's connection %d as a slow consumer: %s", c.User, c.CID, c.Reason)
+		f.Summary = fmt.Sprintf("the bus closed the connection of %s as a slow consumer %s in the last hour, newest at %s: it "+
+			"could not take what it was sent fast enough, and what was on its way to it was lost", who, times, last.UTC().Format("15:04 MST"))
+		f.Headline = clip(capital(who)+" was too slow for the bus", 60)
+		f.Explanation = capital(fmt.Sprintf("%s could not keep up with what the bus sent it, so the bus closed its connection (%s "+
+			"in the last hour) and what was on its way was lost. It connects again by itself; this clears an hour after the last time.", who, count))
+		f.Resolved = "Resolved: " + who + " kept up for an hour"
+	} else {
+		reason := strings.ToLower(c.Reason)
+		f.Said = fmt.Sprintf("the bus closed %s's connection %d: %s", c.User, c.CID, c.Reason)
+		f.Summary = fmt.Sprintf("the bus closed the connection of %s for breaking its rules %s in the last hour, newest at %s: %s",
+			who, times, last.UTC().Format("15:04 MST"), reason)
+		f.Headline = clip(capital(who)+" was cut off by the bus", 60)
+		f.Explanation = capital(fmt.Sprintf("The bus closed the connection of %s because it broke the bus's rules (%s, %s in the "+
+			"last hour), so what it was sending was lost. This clears an hour after the last time.", who, reason, count))
+		f.Resolved = "Resolved: " + who + " kept to the bus's rules for an hour"
+	}
+	w.find(f)
+	if quiet := AdvisoryQuiet - w.now().Sub(last); quiet > 0 {
+		w.signal("S9", key, quiet, signalDue{Do: dueEnd, Finding: f})
+	} else {
+		w.unfind(key)
+	}
+}
+
+// BusUserWords is a bus user as the operator says it, and its machine (the controller's busUserWords).
+func BusUserWords(user string) (string, string) {
+	switch {
+	case user == "controller":
+		return "the controller", ""
+	case strings.HasPrefix(user, "node."):
+		m := strings.TrimPrefix(user, "node.")
+		return "the node-engine on " + m, m
+	}
+	if m, module, ok := strings.Cut(user, "."); ok {
+		if module == "node-tools" {
+			return "the tool runner on " + m, m
+		}
+		return module + " on " + m, m
+	}
+	return "a client of the bus", ""
+}
+
+func capital(s string) string {
+	if s == "" {
+		return s
+	}
+	return strings.ToUpper(s[:1]) + s[1:]
+}
+
+func clip(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	cut := s[:n]
+	if i := strings.LastIndex(cut, " "); i > 0 {
+		cut = cut[:i]
+	}
+	return cut
+}
+
+// ---- dead letters aging out ----
+
+// SetRecount sets how DEAD_LETTERS is counted again when a restricted letter ages out.
+func (w *Watchman) SetRecount(f func()) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.recount = f
+}
+
+// DeadAging is a restricted letter of a consumer (`<stream>.<consumer>`) that ages out at at: a pulse expiring a second
+// after it counts DEAD_LETTERS again, so the consumer's finding ends when its letters are gone (ADR 0376 §6), with no
+// timer. The earliest of a consumer's letters is the one pulsed.
+func (w *Watchman) DeadAging(key string, at time.Time) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	left := at.Sub(w.now())
+	if left <= 0 {
+		left = 0
+	}
+	pk := signalKey("dead", key)
+	if cur, ok := w.signals[pk]; ok && !cur.Finding.Since.IsZero() && cur.Finding.Since.Before(at) && cur.Finding.Since.After(w.now()) {
+		return
+	}
+	w.signal("dead", key, left+time.Second, signalDue{Do: dueRecount,
+		Finding: Finding{Scope: ScopeBus, ID: key, Kind: KindDeadLetters, Severity: Warning, Summary: "counted again when it ages out", Source: "S9", Since: at}})
 }
 
 func orWords(s, otherwise string) string {
@@ -494,14 +779,13 @@ func orWords(s, otherwise string) string {
 
 // SignalRows are the rows the watchman holds, in the table's order, for `signals`.
 var SignalRows = []struct{ Row, Signal, Bound, Kind string }{
-	{"S1", "a machine's presence (not while it said it sleeps or shuts down)", delivery.PulseMoveBound.String(), KindSilent},
+	{"S1", "a machine's presence (not while it said it sleeps or shuts down); a control node urgent after " + ControlNodeUrgentAfter.String(), SilentBound.String(), KindSilent},
 	{"S8", "a provider's failing word repeated", ProviderSaysAgainWithin.String(), KindProviderSilent},
 	{"S9", "the bus's advisories", "any occurrence; clears after " + AdvisoryQuiet.String() + " without another", "slow-consumer, connection-closed, max-deliveries, refused, consumer-lost"},
 	{"S10", "the self-check's checked", (3 * SelfCheckEvery).String(), KindSelfCheckSilent},
 	{"S11", "a machine's tool runner heartbeat", "3 × its interval (" + (3 * HeartbeatEvery).String() + " when it says none)", KindToolsSilent},
-	{"S21", "a have refused or applied in part", ApplyEvery.String(), KindApplyFailing},
+	{"S21", "a machine's have failed or its provisioner client's files not placed", ApplyEvery.String(), KindApplyFailing},
 	{"health", "a have unhealthy in two writes in a row; urgent after " + UnhealthyUrgentAfter.String(), "two writes", KindUnhealthy},
-	{"health", "a have that waits for the operator", "at once", KindNeedsOperator},
 }
 
 // Signal is one signal's pulse as `signals` shows it.

modules/mesh-delivery-watchman/internal/watch/watch.go

@@ -77,6 +77,13 @@ type Watchman struct {
 	unhealthy    map[string]int
 	advised      map[string]int
 	advisedSince map[string]time.Time
+	// control is each node running the controller; failedSince each have's first failure still standing; provFailed each
+	// machine's provisioner client's failed reports; closures each bus closure counted, by finding key and identity.
+	control     map[string]bool
+	failedSince map[string]time.Time
+	provFailed  map[string]map[string]ProvReport
+	closures    map[string]map[string]time.Time
+	recount     func()
 	// stepBound is how long a build step may run once a node took the ask; a test shortens it.
 	stepBound time.Duration
 }
@@ -86,7 +93,8 @@ 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{}, bound: delivery.PulseMoveBound, stepBound: StepBound,
 		findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, 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]delivery.Have{}, 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{}}
 }
 
 // Logf sets where a write that failed is said; it is silent until set.