From 678a51784313519c6dc2e292a1053a99335adac1 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 16:57:40 +0200 Subject: [PATCH] Tests first for issue 1649: a held ask really handed back, a stale running key no duplicate --- internal/link/checks_nats.go | 2 + internal/link/checks_nats_integration_test.go | 68 +++++++++++++++++++ internal/link/held_asks_test.go | 15 ++-- 3 files changed, 79 insertions(+), 6 deletions(-) diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index 176c70ec..3d472515 100644 --- a/internal/link/checks_nats.go +++ b/internal/link/checks_nats.go @@ -268,6 +268,8 @@ type CheckRunning struct { // 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"` + // Seen is when the key was last put, which is the last heartbeat: read from the entry, never written (issue 1649). + Seen time.Time `json:"-"` } // RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.` for a group's diff --git a/internal/link/checks_nats_integration_test.go b/internal/link/checks_nats_integration_test.go index 7afb3d07..6ba9b83c 100644 --- a/internal/link/checks_nats_integration_test.go +++ b/internal/link/checks_nats_integration_test.go @@ -3,6 +3,7 @@ package link import ( + "strings" "context" "encoding/json" "reflect" @@ -577,3 +578,70 @@ func TestARealPullHoldingNothingHandsBackNothing(t *testing.T) { t.Fatalf("handed back %d", n) } } + +// An ask the server delivers to one of a busy machine's pulls outside a fetch (here by a pull request of our own, as one +// that outlived its fetch) is handed back at once for another machine, not held until the ack wait (issue 1649). +func TestAnAskDeliveredToABusyMachinesIdlePullIsHandedBackAtOnce(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) + busy := make(chan string, 4) + release := make(chan struct{}) + a := MachineOverNATS(js, "builder-a").(*natsMachine) + defer a.Close() + go func() { + _ = a.Take(ctx, func(ctx context.Context, work Build) { + busy <- work.Request().ID + <-release + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"}) + _ = work.Done() + }) + }() + if err := asks.Ask(ctx, BuildRequest{ID: "busy", Repository: "/d"}); err != nil { + t.Fatal(err) + } + select { + case <-busy: + case <-time.After(10 * time.Second): + t.Fatal("builder a took nothing") + } + if err := asks.Ask(ctx, aPullCheck("held-check", 9, "head-9-000000000", "trunk-1-000000", false)); err != nil { + t.Fatal(err) + } + var other *takeQueue + for _, q := range a.subs { + if q.kind == queueOther && q.bound { + other = q + } + } + if other == nil { + t.Fatal("builder a has no pull on the other checks") + } + reply := strings.TrimSuffix(other.sub.Subject, "*") + "late" + if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, reply, + []byte(`{"batch":1,"expires":5000000000}`)); 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: + if id != "held-check" { + t.Fatalf("builder b took %s", id) + } + case <-time.After(15 * time.Second): + t.Fatal("the ask held by busy builder a was not handed back; it would wait out the 60 s ack wait") + } + close(release) +} diff --git a/internal/link/held_asks_test.go b/internal/link/held_asks_test.go index 6f84a620..47947a21 100644 --- a/internal/link/held_asks_test.go +++ b/internal/link/held_asks_test.go @@ -41,18 +41,21 @@ func TestEveryMessageAPullHoldsIsHandedBackBeforeABuildStarts(t *testing.T) { func TestAnAskAnotherBuilderIsRunningIsADuplicateAndYoursIsNot(t *testing.T) { r := aPullCheck("build-1", 423, "head-4771ee53aa", "trunk-0123456789", false) - run := CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1", Started: time.Now()} + run := CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1", Started: time.Now().Add(-time.Minute), Seen: time.Now().Add(-4 * time.Second)} + dead := run + dead.Seen = time.Now().Add(-60 * time.Second) for name, c := range map[string]struct { running CheckRunning found bool me string dup bool }{ - "another builder runs this very ask": {run, true, "ace", true}, - "this builder runs it itself": {run, true, "shanks", false}, - "nobody runs it": {CheckRunning{}, false, "ace", false}, - "another ask of the same head": {CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-0"}, true, "ace", false}, - "the same ask at another head": {CheckRunning{Head: "head-other-0000", Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1"}, true, "ace", false}, + "another builder runs this very ask": {run, true, "ace", true}, + "its builder stopped renewing the key a minute ago": {dead, true, "ace", false}, + "this builder runs it itself": {run, true, "shanks", false}, + "nobody runs it": {CheckRunning{}, false, "ace", false}, + "another ask of the same head": {CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-0"}, true, "ace", false}, + "the same ask at another head": {CheckRunning{Head: "head-other-0000", Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1"}, true, "ace", false}, } { if got := duplicateOf(c.running, c.found, r, c.me) != ""; got != c.dup { t.Errorf("%s: duplicate = %v; want %v", name, got, c.dup)