Merge pull request 'The controller takes what the watchman says: key-ttl buckets, the checks running readable, stuck keys raised, silence from presence (design 47 R2b)' (#388) from feat/delivery-watch into main

cmd/mesh-controller/apply_failing.go

@@ -33,7 +33,7 @@ const packageManagerKey = "the package manager ("
 func watchApplyFailing(f *signalFacts) []conditions.Observation {
 	heard := map[string]bool{}
 	for _, m := range f.machines {
-		if !m.asleep() && !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) <= heartbeatBound(m.every) {
+		if !m.asleep() && !m.lastHeard.IsZero() && !m.silent(f.now) {
 			heard[m.name] = true
 		}
 	}

cmd/mesh-controller/doctor.go

@@ -120,6 +120,11 @@ var probeRegistry = []probe{
 	{ID: probeDeliveriesID, Asserts: "no delivery is held past its state's bound unsaid: mesh-delivery's " +
 		"`stalled`, each with the transition its table lets healer H2 take", From: "ADR 0239",
 		Kind: kindDeliveryStalled, Phase: 3, run: probeDeliveries},
+	// What the watchman says stopped moving (ADR 0433 §2, issue 1156): each key of its stuck bucket not moved, but a
+	// node's presence, which the silent check takes.
+	{ID: probeStuckID, Asserts: "no key the watchman says stopped moving is left unsaid: each key of its stuck bucket " +
+		"not moved is a condition of the delivery scope until it says moved", From: "ADR 0433 §2, issue 1156",
+		Kind: kindStuck, Phase: 3, run: probeStuck},
 	// A first merge into a repository the mesh builds nothing from, kept waiting for mesh-delivery's word on the
 	// operator's Approve (novox/hq ADR 0432, issue 1146): said past a bound, never only in a log line (ADR 0426 §3, §8).
 	{ID: probeFirstBuildsID, Asserts: "no first merge into a repository the mesh builds nothing from waits more than ten " +
@@ -183,6 +188,10 @@ var probeRegistry = []probe{
 	{ID: backupHeldProbe, Asserts: "no delivery plan is held for a backup before its send that failed or did not " +
 		"answer in ten minutes", From: "ADR 0404 decision 4, issue 1066", Kind: kindBackupHeld, Phase: 1,
 		run: probeBackupHeld},
+	// A plan's send key the bus refused (ADR 0433 §4, issue 1157): in place of the timer that stood in for it.
+	{ID: sendKeyRefusedID, Asserts: "no delivery plan's key that watches a machine for its report was refused by the " +
+		"bus at the newest look", From: "ADR 0433 §4, issue 1157", Kind: kindSendKeyRefused, Phase: 1,
+		run: probeSendKeyRefused},
 }
 
 // probeVerdict is one probe's outcome in a run.

cmd/mesh-controller/plain_words.go

@@ -610,6 +610,19 @@ 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. " +
+				"The delivery goes on, and the note is written again at its next look.",
+			Resolved: "Resolved: the delivery watches the machine for its report again"}
+	}),
 	kindConsumerBehind: worded(func(o conditions.Observation) words {
 		return words{Headline: "Messages pile up for a listener",
 			Explanation: "One of the mesh's listeners on the bus is far behind, so what it handles happens late. A " +

cmd/mesh-controller/plan_looks.go

@@ -400,34 +400,39 @@ func syncSilentKeys(ctx context.Context, open *stores) error {
 }
 
 // keepSilentKeys puts the keys on the bus as the waits are. Where the bus will not take a key — no bucket, the keys
-// cannot be read, one cannot be put — the wait is looked at by a timer of its own, once, and the error is given back so
+// cannot be read, one cannot be put — the refusal is raised as a condition naming the plan and the machine, never
-// the look is retried.
+// stood in for by a timer, and the error is given back so the look is retried and writes the key again (novox/hq ADR
+// 0433 §4). The set of refusals is what this look found: a key written at the next look is no longer in it.
 func keepSilentKeys(ctx context.Context, c silentClock, want map[silentWait]time.Time, now time.Time) error {
-	byTimer := func() {
+	refused := func(err error) map[silentWait]error {
-		for w, sent := range want {
+		out := map[silentWait]error{}
-			silentByTimer(w, sent, silentTTL(sent, now))
+		for w := range want {
+			out[w] = err
 		}
+		return out
 	}
 	if c == nil {
-		byTimer()
+		noteRefusedSendKeys(refused(errors.New("the bus's clock for the waits is not there")))
 		return nil
 	}
 	c = boundedClock{c}
 	held, err := c.Held(ctx)
 	if err != nil {
-		byTimer()
+		noteRefusedSendKeys(refused(err))
 		return err
 	}
 	arm, drop := silentDiff(want, held)
 	var first error
+	notTaken := map[silentWait]error{}
 	for _, w := range arm {
 		if err := c.Arm(ctx, w, silentTTL(want[w], now), want[w]); err != nil {
-			silentByTimer(w, want[w], silentTTL(want[w], now))
+			notTaken[w] = err
 			if first == nil {
 				first = err
 			}
 		}
 	}
+	noteRefusedSendKeys(notTaken)
 	for _, w := range drop {
 		if err := c.Reported(ctx, w); err != nil && first == nil {
 			first = err
@@ -481,31 +486,6 @@ func (b boundedClock) Reported(ctx context.Context, w silentWait) error {
 	return err
 }
 
-var silentTimers = struct {
-	mu     sync.Mutex
-	timers map[silentWait]time.Time
-}{timers: map[silentWait]time.Time{}}
-
-// silentByTimer looks at the plans when a wait's time is up, for a process whose bus will not take the key
-// (DegradableGrants says why that can be): never silently never. Once for each send.
-func silentByTimer(w silentWait, sent time.Time, ttl time.Duration) {
-	silentTimers.mu.Lock()
-	defer silentTimers.mu.Unlock()
-	if at, ok := silentTimers.timers[w]; ok && at.Equal(sent) {
-		return
-	}
-	silentTimers.timers[w] = sent
-	// retries: a rollout's wait for a machine, where the bus would not take its key. One per send, once.
-	time.AfterFunc(ttl, func() {
-		silentTimers.mu.Lock()
-		if silentTimers.timers[w].Equal(sent) {
-			delete(silentTimers.timers, w)
-		}
-		silentTimers.mu.Unlock()
-		askToLook()
-	})
-}
-
 // startRolloutSilent puts the wait for the machines' reports on the bus for as long as the context lives: the
 // controller is the lease holder here, the only one that writes it. The watcher asks for a look when a key expires.
 func startRolloutSilent(ctx context.Context, js *broker.JetStream) error {

cmd/mesh-controller/plan_looks_test.go

@@ -3,10 +3,12 @@ package main
 import (
 	"context"
 	"errors"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/novox/mesh-controller/internal/conditions"
 	"github.com/novox/mesh-controller/internal/inventory"
 )
 
@@ -159,22 +161,76 @@ func (brokenClock) Held(context.Context) (map[silentWait]time.Time, error) {
 	return nil, errors.New("the bus is down")
 }
 
-// A bus that will not say which keys it holds does not leave the waits unwatched: each is looked at by a timer of its own
+// 1157/AC1: a bus that will not say which keys it holds, or will not take one, does not leave the waits unwatched by
-// and the error is given back, so the look is marked failed and retried (issue 974, review 1).
+// nothing: the refusal raises a condition naming the plan and the machine, no timer stands in for the key, and the error
-func TestKeysThatCannotBeReadAreWatchedByTimersAndTheLookFails(t *testing.T) {
+// is given back so the look is marked failed and retried (issue 974, review 1; ADR 0433 §4).
+func TestARefusedSendKeyRaisesAConditionNamingThePlanAndTheMachine(t *testing.T) { // 1157/AC1
 	w := silentWait{Plan: "plan-held", Tier: 0, Node: "ace"}
 	sent := time.Now().UTC()
-	t.Cleanup(func() { silentTimers.mu.Lock(); delete(silentTimers.timers, w); silentTimers.mu.Unlock() })
+	for name, clock := range map[string]silentClock{"keys cannot be read": brokenClock{}, "a key cannot be put": armRefusingClock{},
-	err := keepSilentKeys(t.Context(), brokenClock{}, map[silentWait]time.Time{w: sent}, sent)
+		"no clock": nil} {
-	if err == nil {
+		t.Run(name, func(t *testing.T) {
-		t.Fatal("no error given back")
+			noteRefusedSendKeys(nil)
+			t.Cleanup(func() { noteRefusedSendKeys(nil) })
+			err := keepSilentKeys(t.Context(), clock, map[silentWait]time.Time{w: sent}, sent)
+			if clock != nil && err == nil {
+				t.Fatal("no error given back")
+			}
+			got := observeRefusedSendKeys()
+			if len(got) != 1 {
+				t.Fatalf("observations %+v", got)
+			}
+			o := got[0]
+			if o.Scope != conditions.ScopePlan || o.ID != "plan-held" || o.Machine != "ace" || o.Kind != kindSendKeyRefused ||
+				!strings.Contains(o.Summary, "plan-held") || !strings.Contains(o.Summary, "ace") {
+				t.Fatalf("the condition does not name the plan and the machine: %+v", o)
+			}
+		})
 	}
-	silentTimers.mu.Lock()
+}
-	armed := silentTimers.timers[w].Equal(sent)
+
-	silentTimers.mu.Unlock()
+// 1157/AC2: at the plan's next look the send key is written again, and the condition clears once it is.
-	if !armed {
+func TestTheNextLookWritesTheRefusedSendKeyAndTheConditionClears(t *testing.T) { // 1157/AC2
-		t.Error("no timer of its own watches the wait")
+	w := silentWait{Plan: "plan-held", Tier: 0, Node: "ace"}
+	sent := time.Now().UTC()
+	noteRefusedSendKeys(nil)
+	t.Cleanup(func() { noteRefusedSendKeys(nil) })
+	want := map[silentWait]time.Time{w: sent}
+	if err := keepSilentKeys(t.Context(), armRefusingClock{}, want, sent); err == nil || len(observeRefusedSendKeys()) != 1 {
+		t.Fatalf("the refusal was not raised: %v %+v", err, observeRefusedSendKeys())
 	}
+	clock := &keptClock{}
+	if err := keepSilentKeys(t.Context(), clock, want, sent); err != nil {
+		t.Fatal(err)
+	}
+	if !clock.armed[w].Equal(sent) {
+		t.Fatalf("the next look did not write the key again: %v", clock.armed)
+	}
+	if got := observeRefusedSendKeys(); len(got) != 0 {
+		t.Fatalf("the condition stays after the key was written: %+v", got)
+	}
+}
+
+type armRefusingClock struct{ silentClock }
+
+func (armRefusingClock) Held(context.Context) (map[silentWait]time.Time, error) { return nil, nil }
+func (armRefusingClock) Arm(context.Context, silentWait, time.Duration, time.Time) error {
+	return errors.New("the bus refuses the key")
+}
+
+// keptClock is a clock that takes every key.
+type keptClock struct {
+	silentClock
+	armed map[silentWait]time.Time
+}
+
+func (c *keptClock) Held(context.Context) (map[silentWait]time.Time, error) { return nil, nil }
+func (c *keptClock) Arm(_ context.Context, w silentWait, _ time.Duration, sent time.Time) error {
+	if c.armed == nil {
+		c.armed = map[silentWait]time.Time{}
+	}
+	c.armed[w] = sent
+	return nil
 }
 
 // A plan moved by a process the serving controller did not spawn — a verb run at the control node's terminal — is
@@ -189,14 +245,13 @@ func (hungClock) Held(context.Context) (map[silentWait]time.Time, error) {
 }
 
 // issue 1121: the look holds the plans while it keeps the silent keys, so a key listing that never ends held the plans
-// and every node for good. Every call to the clock has a bound, and the wait is looked at by a timer of its own.
+// and every node for good. Every call to the clock has a bound, and the refusal is raised as a condition.
 func TestAClockThatNeverAnswersDoesNotHoldTheLookForever(t *testing.T) {
 	was := silentCallBound
 	silentCallBound = 50 * time.Millisecond
 	t.Cleanup(func() { silentCallBound = was })
 	w := silentWait{Plan: "plan-hung", Tier: 0, Node: "ace"}
 	sent := time.Now().UTC()
-	t.Cleanup(func() { silentTimers.mu.Lock(); delete(silentTimers.timers, w); silentTimers.mu.Unlock() })
 	done := make(chan error, 1)
 	go func() { done <- keepSilentKeys(t.Context(), hungClock{}, map[silentWait]time.Time{w: sent}, sent) }()
 	select {

cmd/mesh-controller/probes.go

@@ -629,7 +629,7 @@ func heardMachines(d *doctor) map[string]bool {
 		return out
 	}
 	for _, m := range f.machines {
-		if !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) <= heartbeatBound(m.every) && !m.asleep() {
+		if !m.lastHeard.IsZero() && !m.silent(f.now) && !m.asleep() {
 			out[m.name] = true
 		}
 	}

cmd/mesh-controller/rollout_silent_bus_integration_test.go

@@ -42,7 +42,7 @@ func silentOnABus(t *testing.T) (*busSilent, <-chan silentWait) {
 	return s, heard
 }
 
-func TestAKeyLeftAloneExpiresAndTheWatcherNamesTheMachineSilent(t *testing.T) {
+func TestAKeyLeftAloneExpiresAndTheWatcherNamesTheMachineSilent(t *testing.T) { // 1157/AC3: the send keys stay until release 6 (ADR 0433 §5)
 	s, heard := silentOnABus(t)
 	want := silentWait{Plan: "plan.7", Tier: 2, Node: "ace"}
 	if err := s.Arm(t.Context(), want, time.Second, time.Now()); err != nil {

cmd/mesh-controller/send_key_refused.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"sync"
+
+	"github.com/novox/mesh-controller/internal/conditions"
+)
+
+// A plan's send key the bus refuses (novox/hq ADR 0433 §4, issue 1157): the look that could not write it says which, and
+// a probe of the controller's self-check raises each as a condition naming the plan and the machine. The next look writes
+// the key again; the look that writes it leaves it out of the set, and the condition clears.
+const (
+	kindSendKeyRefused = "send-key-refused"
+	sendKeyRefusedID   = "DL"
+)
+
+var refusedSendKeys = struct {
+	mu sync.Mutex
+	by map[silentWait]error
+}{}
+
+// noteRefusedSendKeys sets the keys the newest look could not write, replacing what the look before found.
+func noteRefusedSendKeys(refused map[silentWait]error) {
+	refusedSendKeys.mu.Lock()
+	defer refusedSendKeys.mu.Unlock()
+	refusedSendKeys.by = refused
+}
+
+// observeRefusedSendKeys is one condition per plan and machine whose send key was refused.
+func observeRefusedSendKeys() []conditions.Observation {
+	refusedSendKeys.mu.Lock()
+	waits := make([]silentWait, 0, len(refusedSendKeys.by))
+	why := map[silentWait]error{}
+	for w, err := range refusedSendKeys.by {
+		waits = append(waits, w)
+		why[w] = err
+	}
+	refusedSendKeys.mu.Unlock()
+	sort.Slice(waits, func(i, j int) bool { return silentKey(waits[i]) < silentKey(waits[j]) })
+	seen := map[[2]string]bool{}
+	var out []conditions.Observation
+	for _, w := range waits {
+		if seen[[2]string{w.Plan, w.Node}] {
+			continue // a plan sends a machine one tier at a time; one condition says it
+		}
+		seen[[2]string{w.Plan, w.Node}] = true
+		out = append(out, conditions.Observation{Scope: conditions.ScopePlan, ID: w.Plan, Token: kindSendKeyRefused + "-" + w.Node,
+			Kind: kindSendKeyRefused, Machine: w.Node, Severity: conditions.Warning,
+			Summary: fmt.Sprintf("the bus refused the key that watches the delivery %s for a report from %s, so nothing "+
+				"names the machine silent if it never reports; the plan writes the key again at its next look", w.Plan, w.Node),
+			Said: fmt.Sprintf("%s: %v", silentKey(w), why[w])})
+	}
+	return out
+}
+
+// probeSendKeyRefused reads what the newest look of the plans could not write.
+func probeSendKeyRefused(context.Context, *doctor) ([]conditions.Observation, error) {
+	return observeRefusedSendKeys(), nil
+}

cmd/mesh-controller/signals.go

@@ -335,7 +335,7 @@ func watchHeartbeats(f *signalFacts) []conditions.Observation {
 		}
 		bound := heartbeatBound(m.every)
 		silent := f.now.Sub(m.lastHeard)
-		if silent <= bound {
+		if !m.silent(f.now) {
 			continue
 		}
 		severity := conditions.Warning
@@ -357,7 +357,7 @@ func watchReports(f *signalFacts) []conditions.Observation {
 		if m.sentAt.IsZero() || m.reportedCurrent || m.asleep() {
 			continue
 		}
-		if !m.lastHeard.IsZero() && f.now.Sub(m.lastHeard) > heartbeatBound(m.every) {
+		if m.silent(f.now) {
 			continue // silent: S1 says it, and a silent machine reports nothing
 		}
 		bound := max(reportAtLeast, 3*m.lastApply)

cmd/mesh-controller/stuck_probe.go

@@ -0,0 +1,228 @@
+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
+)
+
+// presenceEntry is presence.<node> as the watchman wrote it: whether the machine is online and when the entry was made.
+// The watchman writes it only when it changes, so for an online machine the time is when it connected.
+type presenceEntry struct {
+	Online bool
+	At     time.Time
+}
+
+// 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).
+func applyPresence(m *machineFacts, stuck *sdkdelivery.Stuck, entry *presenceEntry, now time.Time) {
+	switch {
+	case stuck == nil:
+		m.presence = presenceUnsaid
+	case stuck.Moved:
+		m.presence = presenceHeard
+	default:
+		m.presence = presenceSilent
+	}
+	switch {
+	case entry == nil:
+	case entry.Online:
+		m.lastHeard = now
+	case entry.At.After(m.lastHeard):
+		m.lastHeard = entry.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
+}
+
+// readPresence is, by node, presence.<node>'s entry; a node without an entry is left out.
+func readPresence(ctx context.Context, conn *nats.Conn, nodes []string) (map[string]presenceEntry, 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]presenceEntry{}
+	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
+		}
+		p, err := sdkdelivery.DecodePresence(e.Value())
+		if err != nil {
+			fmt.Printf("the presence key %s is not read: %v\n", key, err)
+			continue
+		}
+		out[node] = presenceEntry{Online: p.Online, At: 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, now time.Time) 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
+	}
+	entries, err := readPresence(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
+			}
+		}
+		var entry *presenceEntry
+		if e, ok := entries[machines[i].name]; ok {
+			entry = &e
+		}
+		applyPresence(&machines[i], said, entry, now)
+	}
+	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
+}

cmd/mesh-controller/stuck_probe_integration_test.go

@@ -0,0 +1,82 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/testbus"
+)
+
+// 1156/AC4: on the embedded bus with the watchman's buckets, a stuck key written raises a condition and written as moved
+// clears it; a node's presence key not moved says the machine silent through the facts, and moved says it heard.
+func TestAStuckKeyWrittenAndThenMovedRaisesAndClearsACondition(t *testing.T) { // 1156/AC4
+	js, err := broker.Dial(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(js.Close)
+	if _, err := broker.RaiseBuckets(js, []broker.Bucket{
+		{Module: sdkdelivery.WatchmanModule, Name: "stuck"},
+		{Module: sdkdelivery.WatchmanModule, Name: "presence", History: sdkdelivery.PresenceHistory},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	api, err := jetstream.New(js.Conn())
+	if err != nil {
+		t.Fatal(err)
+	}
+	kv, err := api.KeyValue(ctx, sdkdelivery.StuckBucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	write := func(bucket, key string, moved bool) string {
+		t.Helper()
+		stuckKey, err := sdkdelivery.StuckKey(bucket, key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		body, err := sdkdelivery.EncodeStuck(sdkdelivery.Stuck{Bucket: bucket, Key: key, Waited: "an answer", Since: time.Now().UTC(), Moved: moved})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, err := kv.Put(ctx, stuckKey, body); err != nil {
+			t.Fatal(err)
+		}
+		return stuckKey
+	}
+	d := &doctor{js: js}
+	stuckKey := write(sdkdelivery.PulseBucket, "builds.x", false)
+	write(sdkdelivery.PresenceBucket, "ace", false)
+	got, err := probeStuck(ctx, d)
+	if err != nil || len(got) != 1 || got[0].ID != stuckKey || got[0].Kind != kindStuck {
+		t.Fatalf("a stuck key written: %+v %v", got, err)
+	}
+	machines := []machineFacts{{name: "ace", lastHeard: time.Now()}}
+	if err := withPresence(ctx, js.Conn(), machines, time.Now()); 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, time.Now()); err != nil {
+		t.Fatal(err)
+	}
+	if machines[0].silent(time.Now()) {
+		t.Fatal("a presence key that moved still says the machine silent")
+	}
+}

cmd/mesh-controller/stuck_probe_test.go

@@ -0,0 +1,122 @@
+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, nil, now)
+	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, nil, now)
+	if got := watchHeartbeats(f); len(got) != 0 {
+		t.Fatalf("a presence key that moved raised %+v", got)
+	}
+}
+
+// 1156/AC3: with a presence entry that says offline, last heard is the later of last_seen and the entry's time; 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
+	now := time.Now()
+	seen := now.Add(-time.Hour)
+	m := machineFacts{name: "laptop", lastHeard: seen, every: time.Minute}
+	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)
+	}
+	offlineAt := now.Add(-5 * time.Minute)
+	applyPresence(&m, nil, &presenceEntry{Online: false, At: offlineAt}, now)
+	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}
+	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)
+	}
+	for _, lastSeen := range []time.Duration{10 * time.Second, 5 * time.Hour} {
+		m := machineFacts{name: "laptop", lastHeard: now.Add(-lastSeen), every: time.Minute}
+		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)
+		}
+	}
+}

cmd/mesh-controller/watchdogs.go

@@ -155,6 +155,8 @@ type machineFacts struct {
 	tools      bool
 	toolsHeard time.Time
 	toolsEvery time.Duration
+	// presence is the watchman's word on whether it is online (novox/hq ADR 0433 §3); unsaid, the heartbeat rule judges.
+	presence presenceVerdict
 }
 
 // asleep says the machine said it would be away and has not said it is back (ADR 0211).
@@ -470,7 +472,7 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory
 			}
 			m.reportedCurrent = r.Current
 		}
-		if !m.lastHeard.IsZero() && now.Sub(m.lastHeard) > heartbeatBound(m.every) {
+		if m.silent(now) {
 			anyPast = true
 		}
 		if m.tools && now.Sub(later(m.toolsHeard, link.ToolsBeats.Started())) > heartbeatBound(m.toolsEvery) {
@@ -478,6 +480,19 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory
 		}
 		out = append(out, m)
 	}
+	// What the watchman says of each machine: its stuck presence key is the silent verdict, and presence.<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, now); 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.

go.mod

@@ -3,7 +3,7 @@ module github.com/novox/mesh-controller
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/nats-io/nats-server/v2 v2.11.17
 	github.com/nats-io/nats.go v1.54.0

go.sum

@@ -1,7 +1,7 @@
 git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4=
 git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83 h1:BrHyE3ifFMUQLIza6I/iNlEgdcysB9/Y+1FO6Q9xws8=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4 h1:yYDxFRJoH59kiCoEulD+FBwctyf9CrtKIxlEAsVL39Q=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

internal/broker/checks_running_mirror_test.go

@@ -0,0 +1,60 @@
+package broker
+
+import (
+	"slices"
+	"testing"
+)
+
+// 1155/AC1: a module whose manifest reads mesh-controller.checks-running may watch and get every key of the bucket and
+// write none.
+func TestAModuleReadingChecksRunningReadsAndWritesNone(t *testing.T) { // 1155/AC1
+	if !slices.Contains(ControllerMirrors(), "checks-running") || BucketName(ControllerSeat, "checks-running") != CheckRunningBucket {
+		t.Fatalf("a module cannot name %s in `reads`: %v", CheckRunningBucket, ControllerMirrors())
+	}
+	perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-watchman", PasswordHash: "x",
+		Reads: []string{"mesh-controller.checks-running"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	stream := "KV_" + CheckRunningBucket
+	has(t, perms.Publish, "$JS.API.STREAM.INFO."+stream)
+	has(t, perms.Publish, "$JS.API.DIRECT.GET."+stream+".>")
+	has(t, perms.Publish, "$JS.API.CONSUMER.CREATE."+stream+".>")
+	for _, p := range perms.Publish {
+		if SubjectsOverlap(p, "$KV."+CheckRunningBucket+".any.key") {
+			t.Errorf("the reading module may write %s through %q", CheckRunningBucket, p)
+		}
+	}
+}
+
+// 1155/AC3: the writers table names checks-running's writer as the build seat's holders, and the rule that a mirror is
+// written by the controller alone names it as the exception, with its reason.
+func TestChecksRunningIsAMirrorWrittenByTheBuildSeatsHolders(t *testing.T) { // 1155/AC3
+	why := MirrorsWrittenByAnother["checks-running"]
+	if why == "" {
+		t.Fatal("the mirrors written by another do not name checks-running with a reason")
+	}
+	subject := "$KV." + CheckRunningBucket + ".novox.a.7"
+	var rows int
+	for _, row := range WritersTable {
+		for _, s := range row.Subjects {
+			if s != "$KV."+CheckRunningBucket+".>" {
+				continue
+			}
+			rows++
+			holder := Principal{Kind: KindModule, Module: "builder", Node: "one", Holds: []Seat{{Name: TheBuildSeat}}}
+			other := Principal{Kind: KindModule, Module: "mesh-delivery-watchman", Node: "one"}
+			if row.Writes == nil || !row.Writes(holder, subject) || row.Writes(other, subject) {
+				t.Errorf("%q is not written by the build seat's holders alone", row.State)
+			}
+		}
+	}
+	if rows != 1 {
+		t.Fatalf("%d rows of the writers table write %s, want one", rows, CheckRunningBucket)
+	}
+	for m := range MirrorsWrittenByAnother {
+		if m != "checks-running" {
+			t.Errorf("the mirror %q is written by another and is not checks-running", m)
+		}
+	}
+}

internal/broker/controller_buckets.go

@@ -56,7 +56,7 @@ var (
 	// check ends; a key not renewed within the bucket's age is a builder that died, and is gone. A builder that takes a
 	// waiting request for a head and trunk commit a key names drops it. **Written by the build seat's holders, several by
 	// design** — each the key of the check it runs.
-	CheckRunningBucket = BucketName(ControllerSeat, "checks-running")
+	CheckRunningBucket = BucketName(ControllerSeat, ChecksRunning)
 	// DeliveryMergesBucket holds the merges mesh-delivery announced it made, each with the ask whose Approve it merged
 	// on (novox/hq ADR 0420 §7, issue 1025): per merge a key `<owner>/<repo>/<number>/<head sha>` holding the
 	// announcement, kept DeliveryMergesKeptFor. What lets the build of a moved module register it from its new
@@ -133,9 +133,22 @@ func ChainBuckets() []string { return []string{BuildsBucket, RolloutsBucket, Ass
 // ControllerMirrors are the local names of the controller's buckets a module may read: the record mirrors and the
 // chain's buckets.
 func ControllerMirrors() []string {
-	return []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts}
+	return []string{MirrorAssignments, MirrorDeclaredConfig, MirrorPredefined, ChainBuilds, ChainRollouts, ChecksRunning}
 }
 
+// MirrorsWrittenByAnother are the controller's mirrors whose writer is not the controller, each with its reason: the rule
+// that a mirror is written by the controller alone (TestTheMirrorsListAndTheWritersTableAgree) has these exceptions and no
+// other.
+var MirrorsWrittenByAnother = map[string]string{
+	ChecksRunning: "the checks running are said by the builders themselves: every machine holding the build seat puts the " +
+		"key of the check it runs, at its start and at each heartbeat, and deletes it when the check ends, so the " +
+		"controller asserts the bucket and writes none of its keys (ADR 0406 §3); it is readable through `reads` so " +
+		"the watchman sees a check that stopped (ADR 0433 §6)",
+}
+
+// ChecksRunning is the local name of the bucket of the checks running, as a reader names it in `reads`.
+const ChecksRunning = "checks-running"
+
 // The mirrors' caps (design 54 §5): one value per key, no age, 64 KiB a value, 64 MiB a bucket.
 const (
 	MirrorValueBytes = 64 << 10

internal/broker/controller_buckets_test.go

@@ -135,6 +135,12 @@ func TestTheMirrorsListAndTheWritersTableAgree(t *testing.T) {
 	for _, m := range ControllerMirrors() {
 		bucket := BucketName(ControllerSeat, m)
 		listed[bucket] = true
+		if why, other := MirrorsWrittenByAnother[m]; other { // 1155/AC3: the exception, with its reason
+			if why == "" {
+				t.Errorf("the mirror %s is written by another and says no reason", bucket)
+			}
+			continue
+		}
 		if !rowed[bucket] {
 			t.Errorf("the mirror %s is listed and no row of the writers table writes it", bucket)
 		}

internal/broker/jetstream.go

@@ -436,30 +436,50 @@ func retentionOf(r Retention) nats.RetentionPolicy {
 // discards what the bucket holds, and what a module's state holds is data. The mesh's caps are
 // asserted with the owner's options, so a bucket made by hand converges to them.
 func (j *JetStream) EnsureBucket(b Bucket) error {
-	history := b.History
-	if history == 0 {
-		history = 1
-	}
 	js, err := jetstream.New(j.conn)
 	if err != nil {
 		return err
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
+	// A stream that allows a time to live per message cannot be made to disallow it (the server refuses the update),
-		Bucket:       b.Bucket(),
+	// so a bucket that has it keeps it whatever the owner now declares (novox/hq ADR 0433 §1).
-		Description:  b.Why(),
+	var held time.Duration
-		History:      uint8(history),
+	if kv, err := js.KeyValue(ctx, b.Bucket()); err == nil {
-		TTL:          time.Duration(b.TTLSeconds) * time.Second,
+		if st, err := kv.Status(ctx); err == nil {
-		MaxValueSize: StateMaxValueBytes,
+			held = st.LimitMarkerTTL()
-		MaxBytes:     StateMaxBytes,
+		}
-		Storage:      jetstream.FileStorage,
+	}
-	}); err != nil {
+	if _, err := js.CreateOrUpdateKeyValue(ctx, BucketConfig(b, held)); err != nil {
 		return fmt.Errorf("asserting bucket %s: %w", b.Bucket(), err)
 	}
 	return nil
 }
 
+// BucketConfig is the options a module's bucket is asserted with. held is the marker age the bucket already has, zero
+// when it is absent or has none: **a bucket that allows a time to live per message keeps it** (the server refuses to
+// turn it off), so without key-ttl it is asserted with the age it holds, and with key-ttl with KeyTTLMarkerTTL.
+func BucketConfig(b Bucket, held time.Duration) jetstream.KeyValueConfig {
+	history := b.History
+	if history == 0 {
+		history = 1
+	}
+	markers := held
+	if b.KeyTTL {
+		markers = KeyTTLMarkerTTL
+	}
+	return jetstream.KeyValueConfig{
+		Bucket:         b.Bucket(),
+		Description:    b.Why(),
+		History:        uint8(history),
+		TTL:            time.Duration(b.TTLSeconds) * time.Second,
+		MaxValueSize:   StateMaxValueBytes,
+		MaxBytes:       StateMaxBytes,
+		Storage:        jetstream.FileStorage,
+		LimitMarkerTTL: markers,
+	}
+}
+
 // BucketNames is every key-value bucket on the server, the mesh's and anybody else's.
 func (j *JetStream) BucketNames() ([]string, error) {
 	js, err := jetstream.New(j.conn)

internal/broker/state.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 )
 
 // A module's state on the bus (novox/hq ADR 0201, design 32 §4, design 25 §3).
@@ -35,8 +36,14 @@ type Bucket struct {
 	// PerMachine says each machine's instance writes only the key named for its machine (novox/hq ADR
 	// 0260), and is granted that key and no other.
 	PerMachine bool
+	// KeyTTL says the bucket allows a time to live per key (novox/hq ADR 0433 §1): its owner writes each key with its
+	// own, and the server writes a delete marker, kept KeyTTLMarkerTTL, when it ends.
+	KeyTTL bool
 }
 
+// KeyTTLMarkerTTL is how long the marker an expired key leaves is kept: long enough for a watcher to hear it.
+const KeyTTLMarkerTTL = time.Minute
+
 // BucketName is the bucket a module's state lives in: the module and the local name joined by an
 // underscore, which neither may contain, so two modules can never derive one bucket.
 func BucketName(module, name string) string { return module + "_" + name }

internal/broker/state_integration_test.go

@@ -3,9 +3,14 @@
 package broker
 
 import (
+	"context"
+	"errors"
+	"time"
+
 	"testing"
 
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
 )
 
 func TestABucketIsAssertedInPlace(t *testing.T) {
@@ -53,3 +58,48 @@ func TestABucketIsAssertedOverAHeldConnection(t *testing.T) {
 		t.Fatalf("the bucket is not there: %v", err)
 	}
 }
+
+// 1154/AC4: on the pinned server a bucket declared with key-ttl takes a put with a time to live of a second, the key is
+// gone after it and the server has written a delete marker; and the bucket asserted again without key-ttl is not refused.
+func TestAKeyTTLBucketExpiresAKeyAndIsAssertedAgain(t *testing.T) { // 1154/AC4
+	js := aLiveBus(t)
+	b := Bucket{Module: "statetest", Name: "pulse", KeyTTL: true}
+	if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
+		t.Fatalf("a real server refused a key-ttl bucket: %v", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	kv, err := jetstream.New(js.Conn())
+	if err != nil {
+		t.Fatal(err)
+	}
+	bucket, err := kv.KeyValue(ctx, b.Bucket())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := bucket.Create(ctx, "presence.alpha", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil {
+		t.Fatalf("the bucket refused a key with a time to live: %v", err)
+	}
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		if _, err := bucket.Get(ctx, "presence.alpha"); errors.Is(err, jetstream.ErrKeyNotFound) {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("the key is still there after its time to live")
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	hist, err := bucket.History(ctx, "presence.alpha")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if last := hist[len(hist)-1]; last.Operation() == jetstream.KeyValuePut {
+		t.Fatalf("no delete marker was left: last entry is %v", last.Operation())
+	}
+	// The owner no longer says key-ttl: the server cannot turn the age off, so the bucket keeps it and is not refused.
+	b.KeyTTL = false
+	if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
+		t.Fatalf("asserting the bucket without key-ttl was refused, so a restart would fail: %v", err)
+	}
+}

internal/broker/state_keyttl_test.go

@@ -0,0 +1,31 @@
+package broker
+
+import (
+	"testing"
+	"time"
+)
+
+// 1154/AC2: a bucket declared with key-ttl is asserted with the marker age that allows a time to live per message;
+// one without is asserted with none.
+func TestAKeyTTLBucketIsAssertedWithTheMarkerAge(t *testing.T) { // 1154/AC2
+	with := BucketConfig(Bucket{Module: "mesh-delivery-watchman", Name: "pulse", KeyTTL: true}, 0)
+	if with.LimitMarkerTTL != KeyTTLMarkerTTL || KeyTTLMarkerTTL <= 0 {
+		t.Fatalf("a key-ttl bucket is asserted with marker age %v, want %v", with.LimitMarkerTTL, KeyTTLMarkerTTL)
+	}
+	without := BucketConfig(Bucket{Module: "mesh-delivery-watchman", Name: "stuck"}, 0)
+	if without.LimitMarkerTTL != 0 {
+		t.Fatalf("a bucket without key-ttl is asserted with marker age %v, want none", without.LimitMarkerTTL)
+	}
+	if with.Bucket != "mesh-delivery-watchman_pulse" || with.History != 1 || with.MaxBytes != StateMaxBytes {
+		t.Fatalf("the rest of the options changed: %+v", with)
+	}
+}
+
+// 1154/AC3: a bucket that already allows a time to live per message keeps it when its owner no longer says key-ttl,
+// because the server refuses to turn it off; it is not asserted without it.
+func TestABucketThatAllowsAKeyTTLKeepsIt(t *testing.T) { // 1154/AC3
+	got := BucketConfig(Bucket{Module: "m", Name: "s"}, 5*time.Minute)
+	if got.LimitMarkerTTL != 5*time.Minute {
+		t.Fatalf("a bucket that held marker age 5m is asserted with %v, which the server refuses", got.LimitMarkerTTL)
+	}
+}

internal/catalogue/state.go

@@ -36,6 +36,9 @@ type StateDeclaration struct {
 	// PerMachine says the state is keyed by machine: each machine's instance writes the key named for
 	// its machine and no other, and the bus grants it that key alone (novox/hq ADR 0260).
 	PerMachine bool `json:"per-machine,omitempty"`
+	// KeyTTL says each key of the bucket lives the time its writer gives it, and the server writes a delete marker
+	// when it ends (novox/hq ADR 0433 §1): the bucket is made with the server's time to live per message.
+	KeyTTL bool `json:"key-ttl,omitempty"`
 }
 
 // UnmarshalJSON reads a bucket as its bare name, or as {name, history, ttl-seconds}.
@@ -49,7 +52,7 @@ func (s *StateDeclaration) UnmarshalJSON(raw []byte) error {
 	dec := json.NewDecoder(bytes.NewReader(trimmed))
 	dec.DisallowUnknownFields()
 	if err := dec.Decode(&full); err != nil {
-		return fmt.Errorf("a state is either a name or {name, history, ttl-seconds, per-machine}: %w", typedUnknown(err))
+		return fmt.Errorf("a state is either a name or {name, history, ttl-seconds, per-machine, key-ttl}: %w", typedUnknown(err))
 	}
 	*s = StateDeclaration(full)
 	return nil
@@ -57,7 +60,7 @@ func (s *StateDeclaration) UnmarshalJSON(raw []byte) error {
 
 // MarshalJSON writes back the short form when there is nothing else to say.
 func (s StateDeclaration) MarshalJSON() ([]byte, error) {
-	if s.History == 0 && s.TTLSeconds == 0 && !s.PerMachine {
+	if s.History == 0 && s.TTLSeconds == 0 && !s.PerMachine && !s.KeyTTL {
 		return json.Marshal(s.Name)
 	}
 	type plain StateDeclaration
@@ -130,7 +133,7 @@ func StateProblems(m Manifest) []string {
 // controller's account alone (the writers table); declared as a module's state, they would be asserted a
 // second way and granted to the module's own principal as well. Named here so a read of one is a read of
 // kept state. The bus's list is broker.ControllerMirrors; a test keeps the two agreeing.
-var ControllerMirrors = []string{"assignments", "declared-config", "predefined", "builds", "rollouts"}
+var ControllerMirrors = []string{"assignments", "declared-config", "predefined", "builds", "rollouts", "checks-running"}
 
 // StateReadsNothingDeclares is every read across a catalogue whose owner is present and declares no
 // such state. An absent owner says nothing — a module may be installed long before the one whose

internal/catalogue/state_test.go

@@ -39,7 +39,7 @@ func TestAStateNameIsLocalAndOneToken(t *testing.T) {
 		{`{"module":"a.b","version":"1","state":["s"]}`, `no dot`},
 		{`{"module":"a","version":"1","reads":["bindings"]}`, `a read is <module>.<name>`},
 		{`{"module":"a","version":"1","reads":["a.s"]}`, `which is its own state`},
-		{`{"module":"a","version":"1","state":[{"name":"s","shared":true}]}`, `{name, history, ttl-seconds, per-machine}`},
+		{`{"module":"a","version":"1","state":[{"name":"s","shared":true}]}`, `{name, history, ttl-seconds, per-machine, key-ttl}`},
 	} {
 		_, err := ParseManifest([]byte(c.manifest))
 		if err == nil {
@@ -107,3 +107,19 @@ func TestTheCataloguePassNamesAReadItsOwnerDoesNotKeep(t *testing.T) {
 		t.Fatalf("the catalogue pass said %v", problems)
 	}
 }
+
+// 1154/AC1: a state entry may say key-ttl; it is read, written back, and a bare name or an entry without it reads as before.
+func TestAStateMayHaveKeysThatExpireOneByOne(t *testing.T) {
+	m, err := ParseManifest([]byte(`{"module":"mesh-delivery-watchman","version":"1",` +
+		`"state":[{"name":"pulse","key-ttl":true},"stuck",{"name":"seen","history":5}]}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !m.State[0].KeyTTL || m.State[1].KeyTTL || m.State[2].KeyTTL {
+		t.Fatalf("key-ttl not read as said: %+v", m.State)
+	}
+	out, _ := json.Marshal(m.State)
+	if string(out) != `[{"name":"pulse","key-ttl":true},"stuck",{"name":"seen","history":5}]` {
+		t.Fatalf("written back as %s", out)
+	}
+}

internal/inventory/busrecords.go

@@ -269,7 +269,7 @@ func bucketsOf(m catalogue.Manifest) []broker.Bucket {
 	var out []broker.Bucket
 	for _, s := range m.State {
 		out = append(out, broker.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds,
-			PerMachine: s.PerMachine})
+			PerMachine: s.PerMachine, KeyTTL: s.KeyTTL})
 	}
 	return out
 }

internal/inventory/state_keyttl_test.go

@@ -0,0 +1,16 @@
+package inventory
+
+import (
+	"testing"
+
+	"github.com/novox/mesh-controller/internal/catalogue"
+)
+
+// 1154/AC2: a module's state with key-ttl is composed into a bucket that carries KeyTTL; without it, none.
+func TestAKeyTTLStateIsComposedIntoABucketThatSaysSo(t *testing.T) { // 1154/AC2
+	got := bucketsOf(catalogue.Manifest{Module: "mesh-delivery-watchman",
+		State: []catalogue.StateDeclaration{{Name: "pulse", KeyTTL: true}, {Name: "stuck"}}})
+	if len(got) != 2 || !got[0].KeyTTL || got[1].KeyTTL {
+		t.Fatalf("buckets composed as %+v", got)
+	}
+}

internal/link/checks_nats.go

@@ -180,6 +180,9 @@ type CheckRunning struct {
 	Trunk   string `json:"trunk"`
 	Builder string `json:"builder"`
 	Build   string `json:"build"`
+	// Started is when the check started, kept on every heartbeat: the entry's own time is the last heartbeat's, and a
+	// reader that asks how long a check has run reads this (novox/hq ADR 0433 §6).
+	Started time.Time `json:"started"`
 }
 
 // RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.<name>` for a group's
@@ -253,11 +256,23 @@ func keepRunning(ctx context.Context, conn *nats.Conn, r BuildRequest, builder s
 		fmt.Fprintf(os.Stderr, "%s runs without saying so: %v\n", r.ID, err)
 		return func() {}
 	}
+	return keepRunningIn(ctx, kv, key, r, builder, every, time.Now())
+}
+
+// runningStore is what keepRunning needs of the bucket: a put and a delete.
+type runningStore interface {
+	Put(ctx context.Context, key string, value []byte) (uint64, error)
+	Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error
+}
+
+// keepRunningIn is keepRunning on a bucket in hand, the check having started at started: the value is made once, so every
+// heartbeat puts the same started.
+func keepRunningIn(ctx context.Context, kv runningStore, key string, r BuildRequest, builder string, every time.Duration, started time.Time) (stop func()) {
 	trunk := ""
 	if r.Check != nil {
 		trunk = r.Check.Trunk
 	}
-	body, _ := json.Marshal(CheckRunning{Head: r.Ref, Trunk: trunk, Builder: builder, Build: r.ID})
+	body, _ := json.Marshal(CheckRunning{Head: r.Ref, Trunk: trunk, Builder: builder, Build: r.ID, Started: started.UTC()})
 	put := func() {
 		putting, end := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
 		defer end()

internal/link/checks_running_started_test.go

@@ -0,0 +1,61 @@
+package link
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+type recordedRunning struct {
+	mu     sync.Mutex
+	bodies [][]byte
+}
+
+func (r *recordedRunning) Put(_ context.Context, _ string, value []byte) (uint64, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.bodies = append(r.bodies, value)
+	return uint64(len(r.bodies)), nil
+}
+
+func (r *recordedRunning) Delete(context.Context, string, ...jetstream.KVDeleteOpt) error { return nil }
+
+func (r *recordedRunning) puts() [][]byte {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([][]byte(nil), r.bodies...)
+}
+
+// 1155/AC2: a check starting puts its key, and puts it again at each heartbeat, and every value carries the same Started,
+// the time the check started.
+func TestACheckRunningKeepsItsStartAcrossHeartbeats(t *testing.T) { // 1155/AC2
+	store := &recordedRunning{}
+	started := time.Date(2026, 10, 18, 19, 0, 0, 0, time.UTC)
+	stop := keepRunningIn(context.Background(), store, "novox.a.7", aPullCheck("b1", 7, "head-7-000000000", "trunk-1-000000", false),
+		"builder-1", 5*time.Millisecond, started)
+	deadline := time.Now().Add(5 * time.Second)
+	for len(store.puts()) < 3 && time.Now().Before(deadline) {
+		time.Sleep(5 * time.Millisecond)
+	}
+	stop()
+	puts := store.puts()
+	if len(puts) < 3 {
+		t.Fatalf("the key was put %d times, want the first and two heartbeats", len(puts))
+	}
+	for i, body := range puts {
+		var got CheckRunning
+		if err := json.Unmarshal(body, &got); err != nil {
+			t.Fatal(err)
+		}
+		if !got.Started.Equal(started) {
+			t.Fatalf("put %d carries Started %v, want %v", i, got.Started, started)
+		}
+		if got.Builder != "builder-1" || got.Build != "b1" {
+			t.Fatalf("put %d lost the rest of the value: %+v", i, got)
+		}
+	}
+}

testdata/beside/CAPTURED

@@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself.
 
 mesh-catalog  e0bc919d4945bb5f13f971f74643a0b714cf45da  modules/*/module.json, modules/nats/Dockerfile
 mesh-host     9d9e3a46c4f5943cfffb413d962ce48b75e2fd39  examples/foundation-first-node.lock
-mesh-sdk      2b420603fe835d6b80e2c98080ab3094c6896c50  conformance/events/module-event.json
+mesh-sdk      238440d6f8c46a8dd1d829f4313d7b035a6ca847  conformance/events/module-event.json
 
 To move them, from this repository's root, with the two repositories checked out beside it:
 

vendor/git.novox.be/novox/mesh-sdk/go/delivery/delivery.go

@@ -35,8 +35,8 @@ import (
 // its own node only, and the watchman writes presence and pulse.
 const (
 	HaveBucket     = "node-engine_have"
-	PresenceBucket = "mesh-delivery-watchman_presence"
+	PresenceBucket = WatchmanModule + "_presence"
-	PulseBucket    = "mesh-delivery-watchman_pulse"
+	PulseBucket    = WatchmanModule + "_pulse"
 )
 
 // HaveHistory and PresenceHistory are the revisions the have and presence buckets keep per key (research 063

vendor/git.novox.be/novox/mesh-sdk/go/delivery/stuck.go

@@ -0,0 +1,65 @@
+package delivery
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+// WatchmanModule is the watchman's module: its buckets are its own state, `<module>_<name>`, and the controller's
+// broker grants it alone the bus's connect and disconnect events (novox/hq ADR 0431).
+const WatchmanModule = "mesh-delivery-watchman"
+
+// StuckBucket holds what stopped moving (novox/hq ADR 0433 §2): one key per key whose pulse expired, written by the
+// watchman, read by the controller's self-check, which raises each as a condition until it says moved. No key of it
+// is removed: a key that moves again is written as moved.
+const StuckBucket = WatchmanModule + "_stuck"
+
+// StuckKey is the key, in StuckBucket, of the key that stopped moving: `<bucket>.<key>`, formed as PulseKey forms a
+// pulse's, so a node or part gone offline is `mesh-delivery-watchman_presence.<presence key>`.
+func StuckKey(bucket, key string) (string, error) {
+	stuck, err := PulseKey(bucket, key)
+	if err != nil {
+		return "", fmt.Errorf("delivery: no stuck key: %w", err)
+	}
+	return stuck, nil
+}
+
+// ParseStuckKey is the inverse of StuckKey, splitting at the first dot.
+func ParseStuckKey(stuckKey string) (bucket, key string, err error) {
+	bucket, key, err = ParsePulseKey(stuckKey)
+	if err != nil {
+		return "", "", fmt.Errorf("delivery: %q is not a stuck key: %w", stuckKey, err)
+	}
+	return bucket, key, nil
+}
+
+// Stuck is the value of a stuck key: the key that stopped moving, what it waited for in words, since when, and
+// whether it has moved again since. The watchman writes it; the controller reads it.
+type Stuck struct {
+	V      int       `json:"v"`
+	Bucket string    `json:"bucket"`
+	Key    string    `json:"key"`
+	Waited string    `json:"waited,omitempty"`
+	Since  time.Time `json:"since"`
+	Moved  bool      `json:"moved,omitempty"`
+}
+
+// EncodeStuck is s as JSON at Version.
+func EncodeStuck(s Stuck) ([]byte, error) {
+	s.V = Version
+	return json.Marshal(s)
+}
+
+// DecodeStuck reads a stuck value; it refuses an unknown version (ErrUnknownVersion).
+func DecodeStuck(data []byte) (Stuck, error) {
+	var s Stuck
+	err := decode(data, &s)
+	return s, err
+}
+
+// SameStuck says whether two stuck values are the same, so a writer writes only when they differ.
+func SameStuck(a, b Stuck) bool {
+	return a.V == b.V && a.Bucket == b.Bucket && a.Key == b.Key && a.Waited == b.Waited && a.Since.Equal(b.Since) &&
+		a.Moved == b.Moved
+}

vendor/modules.txt

@@ -1,4 +1,4 @@
-# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018164924-2b420603fe83
+# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4
 ## explicit; go 1.22
 git.novox.be/novox/mesh-sdk/go/asks
 git.novox.be/novox/mesh-sdk/go/delivery