The controller takes what the watchman says: key-ttl buckets, the checks running readable, stuck keys raised, silence from presence (design 47 R2b) #388
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user