From bb4d2e1da48ecfb1d42d7f93f63e18eaec8fc1d1 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 19:56:23 +0200 Subject: [PATCH] Let go of the other queues while an ask runs, and read the let-through race right A message that reaches a builder's subscription outside a fetch was held unheartbeated for the whole run of another ask, behind which a delivery waited (issue 1747). The queues the running ask did not come from are let go of and bound again after it. --- internal/link/builds_nats.go | 44 +++++- internal/link/checks_wait_integration_test.go | 131 ++++++++++++++++-- internal/link/stray_test.go | 36 +++++ 3 files changed, 201 insertions(+), 10 deletions(-) create mode 100644 internal/link/stray_test.go diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index e7b0908f..8883c0be 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -307,6 +307,19 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) } continue } + // The seat's own queue is bound again when a running ask made the machine let go of it (quiesce): a machine that cannot + // is not taking, which is said and tried again. + if !m.subs[0].bound { + if err := m.bind(m.subs[0]); err != nil { + fmt.Fprintf(os.Stderr, "this machine cannot bind %s again after a build, and tries again: %v\n", m.seat, err) + select { + case <-ctx.Done(): + return nil + case <-time.After(pausedPoll): + } + continue + } + } if time.Now().After(bindAgain) { for _, q := range m.subs[1:] { if !q.bound { @@ -331,7 +344,7 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) // say why (2026-10-03, the first build agents). Asked again. continue default: - if m.subs[0].sub.IsValid() { + if m.subs[0].sub != nil && m.subs[0].sub.IsValid() { // A transient fault in asking — a reconnect, a slow server — is asked past rather // than ending the machine; one that outlasts the ack wait redelivers nothing lost. time.Sleep(time.Second) @@ -340,6 +353,8 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) return fmt.Errorf("the bus stopped delivering build work: %w", err) } m.handOver(ctx, msg, q, do) + // And the queues it let go of are bound again at once, not after the half minute. + bindAgain = time.Time{} } } @@ -407,11 +422,34 @@ func (m *natsMachine) handOver(ctx context.Context, msg *nats.Msg, q *takeQueue, // 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) + // **And holds nothing else while it runs** (issue 1747): see quiesce. + m.quiesce(q) do(ctx, build) stopRunning() close(working) } +// quiesce lets go of every queue but the one the running ask came from, for as long as it runs. **A builder never holds a +// message it does not start** (issue 1747). A message the server sends after a fetch gave up, within the few milliseconds its +// pull request still lived, lands in that queue's subscription and stays there until the next fetch of it: held, never +// acknowledged as in progress, for as long as the builder runs its current ask, five or ten minutes for a check, while the +// holders that could build it are all busy and a delivery waits behind it. The legacy pull subscription cannot be read +// without sending a new pull request, which could take one more, so it cannot be searched for such a message. It is dropped +// instead: the subscription is closed, a message in it or on its way is lost to this builder and the server hands it out again +// when its worker's wait for an acknowledgement ends (a minute), to any holder that is free, and the queue is bound again +// before the next fetch (Take). The queue the running ask came from keeps its subscription, which its acknowledgements use. +func (m *natsMachine) quiesce(keep *takeQueue) { + for _, q := range m.subs { + if q == keep || !q.bound { + continue + } + if q.sub != nil { + _ = q.sub.Unsubscribe() + } + q.sub, q.pull, q.bound = nil, nil, false + } +} + // runningElsewhere says why an ask need not run here, or nothing: the bucket of the checks running says another builder // runs this very ask (its id, its head and its trunk commit). A key that cannot be read runs the check. func (m *natsMachine) runningElsewhere(ctx context.Context, r BuildRequest) string { @@ -499,6 +537,10 @@ func (m *natsMachine) next(ctx context.Context) (*nats.Msg, *takeQueue, error) { // nextBuild waits for a second at most for a delivery build, the seat's own queue. func (m *natsMachine) nextBuild(ctx context.Context) (*nats.Msg, *takeQueue, error) { + if !m.subs[0].bound { + // Let go of while an ask ran and not bound again yet (Take binds it before it asks): nothing to take from. + return nil, nil, nil + } waiting, end := context.WithTimeout(ctx, takeIdleWait) defer end() fetched, err := m.subs[0].pull.Fetch(1, nats.Context(waiting)) diff --git a/internal/link/checks_wait_integration_test.go b/internal/link/checks_wait_integration_test.go index 8839fc54..1006a438 100644 --- a/internal/link/checks_wait_integration_test.go +++ b/internal/link/checks_wait_integration_test.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "reflect" + "strings" "sync" "testing" "time" @@ -171,16 +172,19 @@ func TestNatsTwoHoldersRacingForOldChecksLetExactlyOneThrough(t *testing.T) { t.Fatalf("the let-through key is not there while the check runs: %v", err) } release <- struct{}{} - // And it is deleted when the check ends. - deadline := time.Now().Add(10 * time.Second) - for { - if _, err := kv.Get(ctx, LetThroughKey); err != nil { - break + // It is deleted when the check ends: the next check let through creates the key again at once, and a create fails + // while the key is there, so its start (the next round) is the proof; after the last one the key stays gone. + if len(order) == 3 { + deadline := time.Now().Add(10 * time.Second) + for { + if _, err := kv.Get(ctx, LetThroughKey); err != nil { + break + } + if time.Now().After(deadline) { + t.Fatal("the let-through key was not deleted when the last check ended") + } + time.Sleep(50 * time.Millisecond) } - if time.Now().After(deadline) { - t.Fatal("the let-through key was not deleted when the check ended") - } - time.Sleep(50 * time.Millisecond) } } if want := []string{"a1", "a2", "a3"}; !reflect.DeepEqual(order, want) { @@ -302,3 +306,112 @@ func TestNatsTheOldestWaitingAskIsReadWithoutTakingIt(t *testing.T) { t.Fatalf("an empty queue read as %v (%v)", found, err) } } + +// Issue 1747: a message that reaches a builder's subscription outside a fetch is not held while the builder runs another ask. A +// stray check lands in the other queue's subscription of builder A, which then runs a long check; the server hands the stray +// out again when its wait for an acknowledgement ends (two seconds here), builder B runs it, and A, finished, does not run it a +// second time from the subscription it held. +func TestNatsAMessageDeliveredOutsideAFetchIsNotHeldWhileTheBuilderRunsAnotherAsk(t *testing.T) { // 1747 + js := aBusWithTheBuildRole(t) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + for _, w := range broker.CheckWorkers() { + w.AckWaitSeconds = 2 + if err := js.EnsureConsumer(w); err != nil { + t.Fatal(err) + } + } + asks := BuildsOn(js, TheBuildMachine) + long := aPullCheck("build-"+fmt.Sprint(time.Now().UnixNano()), 1, "head-1-0000000", "trunk-0-000000", true) + stray := aPullCheck("build-"+fmt.Sprint(time.Now().UnixNano()+1), 2, "head-2-0000000", "trunk-0-000000", false) + for _, r := range []BuildRequest{long, stray} { + if err := asks.Ask(ctx, r); err != nil { + t.Fatal(err) + } + } + + var mu sync.Mutex + runs := map[string][]string{} // ask id -> the machines that ran it + release := make(chan struct{}) + started := make(chan string, 8) + do := func(on string) func(context.Context, Build) { + return func(ctx context.Context, work Build) { + id := work.Request().ID + mu.Lock() + runs[id] = append(runs[id], on) + mu.Unlock() + started <- on + ":" + id + if id == long.ID { + <-release + } + _ = work.Announce(ctx, BuildResult{ID: id, On: on, Failed: "no"}) + _ = work.Done() + } + } + + var a *natsMachine + strayLanded := make(chan error, 1) + a = MachineOverNATSWith(js, "ace", TheBuildMachine, MachineOptions{Claim: func() bool { + // The moment the long check is claimed: the server hands the other queue's waiting check to A's subscription there, as + // one that arrives after a fetch gave up does. Done once. + select { + case <-strayLanded: + return true + default: + } + other := a.subs[2] + inbox := strings.TrimSuffix(other.sub.Subject, "*") + "stray" + body, _ := json.Marshal(map[string]any{"batch": 1, "expires": int64(30 * time.Second)}) + if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, inbox, body); err != nil { + strayLanded <- err + return true + } + _ = js.Conn().Flush() + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { + if n, _, _ := other.sub.Pending(); n > 0 { + break + } + } + strayLanded <- nil + return true + }}).(*natsMachine) + defer a.Close() + go func() { _ = a.Take(ctx, do("ace")) }() + + select { + case got := <-started: + if got != "ace:"+long.ID { + t.Fatalf("started %s, want the long check on ace first", got) + } + case <-time.After(15 * time.Second): + t.Fatal("the long check did not start") + } + if err := <-strayLanded; err != nil { + t.Fatal(err) + } + + // B, free, runs the stray once the server hands it out again. + b := MachineOverNATS(js, "shanks") + defer b.Close() + go func() { _ = b.Take(ctx, do("shanks")) }() + select { + case got := <-started: + if got != "shanks:"+stray.ID { + t.Fatalf("started %s, want the stray on shanks", got) + } + case <-time.After(20 * time.Second): + t.Fatal("the stray was held by the builder that ran the long check and nobody ran it") + } + // A finishes, and does not run the stray from the subscription it held. + release <- struct{}{} + select { + case got := <-started: + t.Fatalf("%s started after the long check: the stray ran twice", got) + case <-time.After(6 * time.Second): + } + mu.Lock() + defer mu.Unlock() + if len(runs[stray.ID]) != 1 { + t.Fatalf("the stray ran on %v, want once", runs[stray.ID]) + } +} diff --git a/internal/link/stray_test.go b/internal/link/stray_test.go new file mode 100644 index 00000000..31d2a495 --- /dev/null +++ b/internal/link/stray_test.go @@ -0,0 +1,36 @@ +package link + +import ( + "testing" + + "github.com/nats-io/nats.go" +) + +// Issue 1747: a builder never holds a message it does not start. A message the server sends after a fetch gave up lands in that +// queue's subscription; while an ask runs, the queues it does not come from are let go of, so such a message is not held +// unheartbeated for the whole run, and the queues are bound again before the next fetch. + +func TestARunningAskLetsGoOfTheQueuesItDidNotComeFrom(t *testing.T) { + m, builds, _, other, _ := aMachine() + // A delivery build reached the builds queue's subscription after its fetch gave up: it sits in the buffer. + builds.waiting = []*nats.Msg{buildAsk()} + took := m.subs[1] // the running ask came from the ready queue + + m.quiesce(took) + + if !took.bound || took.pull == nil { + t.Error("the queue the running ask came from was let go of; its acknowledgements need it") + } + for _, q := range []*takeQueue{m.subs[0], m.subs[2]} { + if q.bound || q.pull != nil || q.sub != nil { + t.Errorf("the %s queue is still held while an ask runs", q.kind) + } + } + // Nothing is fetched from a queue that was let go of: a message in it cannot be taken unheartbeated. + if msg, _, _ := m.next(t.Context()); msg != nil { + t.Fatalf("a message was taken from a queue that was let go of: %s", msg.Data) + } + if builds.fetches() != 0 || other.fetches() != 0 { + t.Errorf("fetched from a queue that was let go of: builds %d, other %d", builds.fetches(), other.fetches()) + } +}