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

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{
-		Bucket:       b.Bucket(),
-		Description:  b.Why(),
-		History:      uint8(history),
-		TTL:          time.Duration(b.TTLSeconds) * time.Second,
-		MaxValueSize: StateMaxValueBytes,
-		MaxBytes:     StateMaxBytes,
-		Storage:      jetstream.FileStorage,
-	}); err != nil {
+	// A stream that allows a time to live per message cannot be made to disallow it (the server refuses the update),
+	// so a bucket that has it keeps it whatever the owner now declares (novox/hq ADR 0433 §1).
+	var held time.Duration
+	if kv, err := js.KeyValue(ctx, b.Bucket()); err == nil {
+		if st, err := kv.Status(ctx); err == nil {
+			held = st.LimitMarkerTTL()
+		}
+	}
+	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)
+		}
+	}
+}