From bd257afbd52d2e8fa94ffb3db5362da4190cd1fe Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 17 Oct 2026 14:10:26 +0200 Subject: [PATCH] issue 974: the builders say the checks they run in a bucket that ages, and a builder drops a waiting request for a head and trunk commit another is running --- internal/broker/checks.go | 6 + internal/broker/checks_test.go | 3 + internal/broker/controller_buckets.go | 30 ++++- internal/broker/nats.go | 5 + internal/broker/testdata/composed.conf | 2 +- internal/broker/writers.go | 12 ++ internal/broker/writers_test.go | 1 + internal/link/builds_nats.go | 33 ++++- internal/link/checks_nats.go | 120 +++++++++++++++++ internal/link/checks_nats_test.go | 177 +++++++++++++++++++++++++ 10 files changed, 380 insertions(+), 9 deletions(-) diff --git a/internal/broker/checks.go b/internal/broker/checks.go index e9efd275..f9ab8da9 100644 --- a/internal/broker/checks.go +++ b/internal/broker/checks.go @@ -92,6 +92,12 @@ func holderCheckGrants() []string { "$JS.API.CONSUMER.MSG.NEXT."+w.Stream+"."+w.Name, "$JS.ACK."+w.Stream+"."+w.Name+".>") } + // And the bucket of the checks they are running: each writes the key of its own check, reads another's, and the bucket + // is a controller's own bucket the build seat's holders write by design (ADR 0406 §3). + pub = append(pub, + "$KV."+CheckRunningBucket+".>", + "$JS.API.STREAM.INFO.KV_"+CheckRunningBucket, + "$JS.API.DIRECT.GET.KV_"+CheckRunningBucket+".>") return pub } diff --git a/internal/broker/checks_test.go b/internal/broker/checks_test.go index e778abeb..23bb9e84 100644 --- a/internal/broker/checks_test.go +++ b/internal/broker/checks_test.go @@ -26,6 +26,9 @@ func TestTheBuildAgentReadsAndWatchesThePullRequestsKeys(t *testing.T) { "$JS.API.CONSUMER.MSG.NEXT.CHECKS_READY.CHECKS_READY_worker", "$JS.API.CONSUMER.MSG.NEXT.CHECKS_OTHER.CHECKS_OTHER_worker", "$JS.ACK.CHECKS_OTHER.CHECKS_OTHER_worker.>", + // And the bucket of the checks running, which it writes and reads (ADR 0406 §3). + "$KV." + CheckRunningBucket + ".>", + "$JS.API.DIRECT.GET.KV_" + CheckRunningBucket + ".>", } { has(t, perms.Publish, s) } diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index 7d8ba4e4..8aa61828 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -49,6 +49,13 @@ var ( // controller's watcher hearing the expiry ends the wait and names the machine silent. Written by the lease // holder alone. RolloutSilentBucket = BucketName(ControllerSeat, "rollout-silent") + // CheckRunningBucket holds the pull request checks the builders are running now (novox/hq ADR 0406 §3, issue 974): per + // pull request a key `..` (or `group.`) holding the head and trunk commit being checked, the + // builder and its build. A builder puts it when it starts a check and again on each heartbeat, and deletes it when the + // 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") ) // The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of @@ -102,6 +109,10 @@ const RolloutSilentMarkerTTL = time.Minute // takes nothing live. const RolloutSilentKeptFor = time.Hour +// CheckRunningKeptFor is the age of a running check's key: a few heartbeats (ten seconds each) of a builder, so a builder that +// died is forgotten within two minutes and the check it ran is taken again then. +const CheckRunningKeptFor = 2 * time.Minute + // AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own. const AskedKeptFor = 30 * 24 * time.Hour @@ -130,13 +141,13 @@ func IsControllerBucket(bucket string) bool { return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket || bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket || bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket || - bucket == MergeWindowBucket || bucket == RolloutSilentBucket + bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket } // ControllerBuckets are the controller's own buckets, in the order they are asserted. func ControllerBuckets() []string { return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket, - AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket} + AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket} } // ControllerBucketsAsserter is what raising the controller's buckets needs of a connection. @@ -264,6 +275,21 @@ func (j *JetStream) EnsureControllerBuckets() error { }); err != nil { return fmt.Errorf("asserting bucket %s: %w", RolloutSilentBucket, err) } + // **The checks running: a key that ages** (ADR 0406 §3, issue 974). One message a key, each renewal a new one, so the + // age of the bucket is how long a key outlives the builder that stopped renewing it. + if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: CheckRunningBucket, + Description: "the pull request checks the builders are running now (novox/hq ADR 0406): `` holds the " + + "head and trunk commit, the builder and its build; put by the builder when it starts and at each heartbeat, " + + "deleted when the check ends, gone within two minutes of a builder that died; written by the build seat's holders", + History: 1, + TTL: CheckRunningKeptFor, + MaxValueSize: 4 << 10, + MaxBytes: 8 << 20, + Storage: jetstream.FileStorage, + }); err != nil { + return fmt.Errorf("asserting bucket %s: %w", CheckRunningBucket, err) + } // **The record mirrors: no age** (design 54 §5). A key is removed when its record is, by a purge whose // marker lives a day; a key that expired would be a record the bus forgot while the store still holds it. for _, m := range []struct{ bucket, why string }{ diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 7507be33..b73d975e 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -387,6 +387,11 @@ var DegradableGrants = []DegradableGrant{ "first push, and a check asked before that is asked again by the next announcement of its head (ADR 0406 §3, " + "issue 974)", }, + { + Subject: "$KV." + CheckRunningBucket + ".>", + Reason: "refused: the controller asserts this bucket through the API and writes none of its keys — the build seat's " + + "holders do (ADR 0406 §3, issue 974)", + }, { Subject: "mesh.seat." + ControllerSeat + ".event.sent", Reason: "refused: the controller says a tier sent to a machine only for outsiders, and a send that could not " + diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index dadf2781..32fd9212 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -24,7 +24,7 @@ accounts { jetstream: enabled users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } } diff --git a/internal/broker/writers.go b/internal/broker/writers.go index 437612e1..8dc3c4b0 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -41,6 +41,14 @@ func ownMachine(p Principal, pattern string) bool { return p.Kind == KindNode && len(tokens) > 2 && tokens[2] == p.Node } +// holdsTheBuildSeat is a principal holding the build seat, or the controller, which asserts the bucket the holders write. +func holdsTheBuildSeat(p Principal, _ string) bool { + if p.Kind == KindController { + return true + } + return holdsSeatOf(p, "mesh.seat."+TheBuildSeat+".x") +} + // holdsSeatOf is a principal holding the seat a `mesh.seat..…` pattern names: its module's own // principal, or the node tools carrying a module that holds it (ADR 0175, the runtime is its modules). func holdsSeatOf(p Principal, pattern string) bool { @@ -153,6 +161,10 @@ var WritersTable = []WriterRow{ Others: "the build seat's holders take it", Subjects: []string{CheckSubjectPrefix(TheBuildSeat, true) + ">", CheckSubjectPrefix(TheBuildSeat, false) + ">"}, Writes: isController}, + {State: "a pull request's check running", Writer: "the build seat's holders — several by design, each the key of the check it runs", + KeptIn: "key-value " + CheckRunningBucket + ", aged out when not renewed", Others: "the build seat's holders read it to drop a duplicate request", + Subjects: kvOf(CheckRunningBucket), Writes: holdsTheBuildSeat, + Shared: "every machine holding the build seat puts the key of the check it runs and deletes it when the check ends"}, {State: "a merge announced", Writer: "one announcer per forge (the hook, or the poll when the hook is absent — never both)", KeptIn: "the bus", Others: "—", Subjects: []string{"mesh.mod.*.event.pull.merged"}, Writes: ownModule}, // A pull request's head, before it merges (novox/hq to-be 45 §9): the same one announcer. diff --git a/internal/broker/writers_test.go b/internal/broker/writers_test.go index 6f4c585f..6f929662 100644 --- a/internal/broker/writers_test.go +++ b/internal/broker/writers_test.go @@ -28,6 +28,7 @@ var designRows = []string{ "stream definitions and bus permissions", "builds and their outcomes", "a pull request's check asked", + "a pull request's check running", "a merge announced", "a pull request's head announced", "a pull request's merge check", diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index ce33dbec..b1985f0d 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -367,7 +367,11 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) // the ask to a second machine nor counts the wait against its deliveries. working := make(chan struct{}) go stillWorking(msg, working, heartbeatWhileWorking) + // And a check says it is running, to the other builders (ADR 0406 §3): its key put now and at each heartbeat, + // deleted when it ends, whatever the outcome. + stopRunning := keepRunning(ctx, m.js.Conn(), request, m.on, heartbeatWhileWorking) do(ctx, build) + stopRunning() close(working) } } @@ -431,17 +435,34 @@ func (m *natsMachine) next(ctx context.Context) (*nats.Msg, *takeQueue, error) { // mesh-delivery's, and a check wrongly dropped is worse than one run twice). func (m *natsMachine) judgedAlready(ctx context.Context, r BuildRequest) string { c := r.Check - if c == nil || c.Group != "" || c.Try || c.Number <= 0 { + if c == nil || c.Try || (c.Group == "" && c.Number <= 0) { return "" } - key, found, err := ReadPullKey(ctx, m.js.Conn(), c.Owner, c.Repo, c.Number) + if c.Group == "" { + key, found, err := ReadPullKey(ctx, m.js.Conn(), c.Owner, c.Repo, c.Number) + if err != nil { + fmt.Fprintf(os.Stderr, "could not read the key of %s/%s#%d, so its check runs: %v\n", c.Owner, c.Repo, c.Number, err) + } else if found && key.Judged(r.Ref, c.Trunk) { + return fmt.Sprintf("%s/%s#%d is judged already at %.12s on %.12s (%s)", c.Owner, c.Repo, c.Number, r.Ref, c.Trunk, + key.Verdict) + } + } + return m.beingChecked(ctx, r) +} + +// beingChecked says why a request need not run, or nothing: another builder's check of the same head and trunk commit is +// running, and says so under its pull request's key in the bucket of the checks running (ADR 0406 §3). A key a builder +// stopped renewing is gone within the bucket's age, and the request is taken then. Never the ask's own id: a redelivered ask +// is its own, and a key an older builder wrote that names no head or trunk is no match. +func (m *natsMachine) beingChecked(ctx context.Context, r BuildRequest) string { + running, ok, err := ReadRunning(ctx, m.js.Conn(), r) if err != nil { - fmt.Fprintf(os.Stderr, "could not read the key of %s/%s#%d, so its check runs: %v\n", c.Owner, c.Repo, c.Number, err) + fmt.Fprintf(os.Stderr, "could not read whether %s is being checked, so its check runs: %v\n", r.ID, err) return "" } - if found && key.Judged(r.Ref, c.Trunk) { - return fmt.Sprintf("%s/%s#%d is judged already at %.12s on %.12s (%s)", c.Owner, c.Repo, c.Number, r.Ref, c.Trunk, - key.Verdict) + if ok && running.Build != r.ID && running.Same(r) { + return fmt.Sprintf("%s is being checked at %.12s on %.12s by %s as %s", r.ID, r.Ref, r.Check.Trunk, running.Builder, + running.Build) } return "" } diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index f0d90cf2..cf09cad0 100644 --- a/internal/link/checks_nats.go +++ b/internal/link/checks_nats.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "os" "strconv" "strings" "time" @@ -168,3 +169,122 @@ const ( checkTakeWait = 30 * time.Millisecond takeIdleWait = time.Second ) + +// CheckRunning is what a builder says of the check it is running, in the bucket of the checks running (novox/hq ADR 0406 +// §3): the head and trunk commit it checks, the builder and its build. The key is the pull request's (RunningKeyOf). +type CheckRunning struct { + Head string `json:"head"` + Trunk string `json:"trunk"` + Builder string `json:"builder"` + Build string `json:"build"` +} + +// RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.` for a group's +// composed check; none for a try, which no other request replaces. +func RunningKeyOf(r BuildRequest) (string, bool) { + c := r.Check + switch { + case c == nil, c.Try: + return "", false + case c.Group != "": + return "group." + tokenOf(c.Group), true + case c.Number <= 0: + return "", false + } + return PullKeyOf(c.Owner, c.Repo, c.Number), true +} + +// runningBucket opens the bucket of the checks running. +func runningBucket(ctx context.Context, conn *nats.Conn) (jetstream.KeyValue, error) { + api, err := jetstream.New(conn) + if err != nil { + return nil, err + } + kv, err := api.KeyValue(ctx, broker.CheckRunningBucket) + if err != nil { + return nil, fmt.Errorf("the checks running: %w", err) + } + return kv, nil +} + +// ReadRunning reads the key of the check running for a request's pull request: false when there is none. +func ReadRunning(ctx context.Context, conn *nats.Conn, r BuildRequest) (CheckRunning, bool, error) { + key, ok := RunningKeyOf(r) + if !ok { + return CheckRunning{}, false, nil + } + kv, err := runningBucket(ctx, conn) + if err != nil { + return CheckRunning{}, false, err + } + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return CheckRunning{}, false, nil + } + if err != nil { + return CheckRunning{}, false, err + } + var running CheckRunning + if err := json.Unmarshal(e.Value(), &running); err != nil { + return CheckRunning{}, false, nil // a key of a builder from before this means no match + } + return running, true, nil +} + +// Same says a running check is for this request's head and trunk commit: both said, and the same. A key that names neither +// (an older builder's) is no match. +func (c CheckRunning) Same(r BuildRequest) bool { + return r.Check != nil && c.Head != "" && c.Trunk != "" && r.Check.Trunk != "" && sameCommit(c.Head, r.Ref) && + sameCommit(c.Trunk, r.Check.Trunk) +} + +// keepRunning says a check is running: its key put now, put again at each heartbeat, and deleted by the stop it returns, +// whatever the outcome. Said to nobody that cannot be told: a bucket that is not there is no reason not to run the check. +func keepRunning(ctx context.Context, conn *nats.Conn, r BuildRequest, builder string, every time.Duration) (stop func()) { + key, ok := RunningKeyOf(r) + if !ok { + return func() {} + } + kv, err := runningBucket(ctx, conn) + if err != nil { + fmt.Fprintf(os.Stderr, "%s runs without saying so: %v\n", r.ID, err) + return func() {} + } + trunk := "" + if r.Check != nil { + trunk = r.Check.Trunk + } + body, _ := json.Marshal(CheckRunning{Head: r.Ref, Trunk: trunk, Builder: builder, Build: r.ID}) + put := func() { + putting, end := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer end() + if _, err := kv.Put(putting, key, body); err != nil { + fmt.Fprintf(os.Stderr, "%s could not say it is running: %v\n", r.ID, err) + } + } + put() + done := make(chan struct{}) + finished := make(chan struct{}) + go func() { + defer close(finished) + tick := time.NewTicker(every) + defer tick.Stop() + for { + select { + case <-done: + return + case <-tick.C: + put() + } + } + }() + return func() { + close(done) + <-finished + deleting, end := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer end() + if err := kv.Delete(deleting, key); err != nil { + fmt.Fprintf(os.Stderr, "%s could not take its running key away; it ages out: %v\n", r.ID, err) + } + } +} diff --git a/internal/link/checks_nats_test.go b/internal/link/checks_nats_test.go index 9326ade6..34c36447 100644 --- a/internal/link/checks_nats_test.go +++ b/internal/link/checks_nats_test.go @@ -307,3 +307,180 @@ func TestTheQueueIsReadAcrossTheThreeStreamsAndAnAskIsDeletedFromItsOwn(t *testi t.Errorf("%d asks left; want 2", n) } } + +// aRunningBucket makes the bucket the builders say the checks they run in, with an age of its own short enough for a test +// (the mesh's is two minutes): a key not renewed within it is gone. +func aRunningBucket(t *testing.T, js *broker.JetStream, age time.Duration) jetstream.KeyValue { + t.Helper() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + kv, err := api.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: broker.CheckRunningBucket, History: 1, TTL: age}) + if err != nil { + t.Fatal(err) + } + return kv +} + +// A builder taking a request for a pull request's head and trunk commit that another builder is running a check of drops it: +// the bucket of running checks names the head and the trunk, the builder and the build (issue 974). +func TestARunningCheckOnOneBuilderMakesAnotherDropTheSameRequestWaiting(t *testing.T) { + js := aBusWithTheBuildRole(t) + aKeyBucket(t, js) + aRunningBucket(t, js, 2*time.Minute) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + asks := BuildsOn(js, TheBuildMachine) + running := make(chan string, 4) + release := make(chan struct{}) + a := MachineOverNATS(js, "builder-a") + defer a.Close() + go func() { + _ = a.Take(ctx, func(ctx context.Context, work Build) { + running <- work.Request().ID + <-release + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"}) + _ = work.Done() + }) + }() + if err := asks.Ask(ctx, aPullCheck("first", 7, "head-7-000000000", "trunk-1-000000", false)); err != nil { + t.Fatal(err) + } + select { + case <-running: + case <-time.After(10 * time.Second): + t.Fatal("builder a took nothing") + } + // The same head and trunk commit asked again while it runs, and a pull request's other head asked beside it. + if err := asks.Ask(ctx, aPullCheck("again", 7, "head-7-000000000", "trunk-1-000000", false)); err != nil { + t.Fatal(err) + } + b := MachineOverNATS(js, "builder-b") + defer b.Close() + took := make(chan string, 4) + go func() { + _ = b.Take(ctx, func(ctx context.Context, work Build) { + took <- work.Request().ID + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-b", Failed: "no"}) + _ = work.Done() + }) + }() + select { + case id := <-took: + t.Fatalf("builder b took %s, which builder a is running a check of", id) + case <-time.After(3 * time.Second): + } + if n := len(queueOf(t, js).Of(AskWaiting)); n != 0 { + t.Errorf("%d request(s) still wait; builder b was to drop the one for a head and trunk commit being checked", n) + } + // Another trunk commit is another check. + if err := asks.Ask(ctx, aPullCheck("moved", 7, "head-7-000000000", "trunk-2-000000", false)); err != nil { + t.Fatal(err) + } + select { + case id := <-took: + if id != "moved" { + t.Fatalf("builder b took %s", id) + } + case <-time.After(10 * time.Second): + t.Fatal("a check against another trunk commit was dropped too") + } + close(release) +} + +// A key that is not renewed ages out, and the request for its head and trunk commit is then taken: a builder that died +// running a check leaves nothing that holds the next one back for longer than the bucket's age. +func TestAKeyNotRenewedAgesOutAndTheRequestIsThenTaken(t *testing.T) { + js := aBusWithTheBuildRole(t) + aKeyBucket(t, js) + kv := aRunningBucket(t, js, 2*time.Second) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + body, _ := json.Marshal(CheckRunning{Head: "head-7-000000000", Trunk: "trunk-1-000000", Builder: "dead", Build: "gone"}) + if _, err := kv.Put(ctx, PullKeyOf("novox", "a", 7), body); err != nil { + t.Fatal(err) + } + asks := BuildsOn(js, TheBuildMachine) + took := make(chan string, 4) + b := MachineOverNATS(js, "builder-b") + defer b.Close() + go func() { + _ = b.Take(ctx, func(ctx context.Context, work Build) { + took <- work.Request().ID + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-b", Failed: "no"}) + _ = work.Done() + }) + }() + if err := asks.Ask(ctx, aPullCheck("while-it-lives", 7, "head-7-000000000", "trunk-1-000000", false)); err != nil { + t.Fatal(err) + } + select { + case id := <-took: + t.Fatalf("%s was taken while the key was fresh", id) + case <-time.After(1200 * time.Millisecond): + } + time.Sleep(2500 * time.Millisecond) // the key was not renewed + if err := asks.Ask(ctx, aPullCheck("after-it-aged", 7, "head-7-000000000", "trunk-1-000000", false)); err != nil { + t.Fatal(err) + } + select { + case id := <-took: + if id != "after-it-aged" { + t.Fatalf("took %s", id) + } + case <-time.After(10 * time.Second): + t.Fatal("the request was not taken after the key aged out") + } +} + +// A builder running a check renews its key on each heartbeat, so it outlives the bucket's age, and deletes it when the check +// ends, whatever the outcome. +func TestARunningCheckRenewsItsKeyAndDeletesItWhenItEnds(t *testing.T) { + js := aBusWithTheBuildRole(t) + kv := aRunningBucket(t, js, 2*time.Second) + was := heartbeatWhileWorking + heartbeatWhileWorking = 400 * time.Millisecond + t.Cleanup(func() { heartbeatWhileWorking = was }) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + running := make(chan struct{}, 1) + release := make(chan struct{}) + m := MachineOverNATS(js, "builder-a") + defer m.Close() + go func() { + _ = m.Take(ctx, func(ctx context.Context, work Build) { + running <- struct{}{} + <-release + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"}) + _ = work.Done() + }) + }() + if err := BuildsOn(js, TheBuildMachine).Ask(ctx, aPullCheck("long", 7, "head-7-000000000", "trunk-1-000000", false)); err != nil { + t.Fatal(err) + } + <-running + key := PullKeyOf("novox", "a", 7) + e, err := kv.Get(ctx, key) + if err != nil { + t.Fatalf("no key while the check runs: %v", err) + } + var got CheckRunning + _ = json.Unmarshal(e.Value(), &got) + if got.Head != "head-7-000000000" || got.Trunk != "trunk-1-000000" || got.Builder != "builder-a" || got.Build != "long" { + t.Errorf("the key says %+v", got) + } + time.Sleep(3 * time.Second) // longer than the bucket's age + if _, err := kv.Get(ctx, key); err != nil { + t.Fatalf("the key aged out under a check still running: %v", err) + } + close(release) + for end := time.Now().Add(5 * time.Second); ; time.Sleep(50 * time.Millisecond) { + if _, err := kv.Get(ctx, key); err != nil { + break + } + if time.Now().After(end) { + t.Fatal("the key stays after the check ended") + } + } +}