From 027360909060e3a290a8aeef07bfbc2e89b78cfe Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 20:55:18 +0200 Subject: [PATCH] The checks running are readable through reads and carry when each started (issue 1155) --- internal/broker/checks_running_mirror_test.go | 60 ++++++++++++++++++ internal/broker/controller_buckets.go | 17 +++++- internal/broker/controller_buckets_test.go | 6 ++ internal/catalogue/state.go | 2 +- internal/link/checks_nats.go | 17 +++++- internal/link/checks_running_started_test.go | 61 +++++++++++++++++++ 6 files changed, 159 insertions(+), 4 deletions(-) create mode 100644 internal/broker/checks_running_mirror_test.go create mode 100644 internal/link/checks_running_started_test.go diff --git a/internal/broker/checks_running_mirror_test.go b/internal/broker/checks_running_mirror_test.go new file mode 100644 index 00000000..a19f1dcf --- /dev/null +++ b/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) + } + } +} diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index 29ce9439..ef129ba7 100644 --- a/internal/broker/controller_buckets.go +++ b/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 `///` 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 diff --git a/internal/broker/controller_buckets_test.go b/internal/broker/controller_buckets_test.go index 1db7465c..228f05c0 100644 --- a/internal/broker/controller_buckets_test.go +++ b/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) } diff --git a/internal/catalogue/state.go b/internal/catalogue/state.go index 4c3dcb8d..14e3432d 100644 --- a/internal/catalogue/state.go +++ b/internal/catalogue/state.go @@ -133,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 diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index fe900eb2..924af561 100644 --- a/internal/link/checks_nats.go +++ b/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.` 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() diff --git a/internal/link/checks_running_started_test.go b/internal/link/checks_running_started_test.go new file mode 100644 index 00000000..16dc3e0a --- /dev/null +++ b/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) + } + } +}