Guard the taking loop's subscriptions, and update the two tests that held a busy machine's pulls

The loop now lets go of and binds queues around every ask, so Close takes the lock the loop does;
the busy machine's test says it holds no pull, and the stray test keeps its long check alive
within its two-second ack wait.
This commit is contained in:
2026-10-21 20:53:58 +02:00
parent 60253c82b7
commit e7cc315fd0
3 changed files with 31 additions and 15 deletions
+22
View File
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"os"
"sync"
"time"
"github.com/nats-io/nats.go"
@@ -168,6 +169,8 @@ type natsMachine struct {
// took last (ADR 0483 clause 3).
said map[string]bool
letKey bool
// mu guards the subscriptions of subs, bound and let go of by the taking loop and closed by Close.
mu sync.Mutex
}
// takeQueue is one queue a machine pulls from.
@@ -238,6 +241,8 @@ const (
)
func (m *natsMachine) Close() {
m.mu.Lock()
defer m.mu.Unlock()
for _, q := range m.subs {
if q.sub != nil {
_ = q.sub.Unsubscribe()
@@ -442,6 +447,8 @@ func (m *natsMachine) handOver(ctx context.Context, msg *nats.Msg, q *takeQueue,
// 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) {
m.mu.Lock()
defer m.mu.Unlock()
for _, q := range m.subs {
if q == keep || !q.bound {
continue
@@ -489,6 +496,8 @@ func duplicateOf(running CheckRunning, found bool, r BuildRequest, me string) st
// subscribing `…accept.build` against a consumer filtered on `…accept.>` is rejected even though it is narrower. Learned
// twice now, on two different consumers, which is why it is written down here.
func (m *natsMachine) bind(q *takeQueue) error {
m.mu.Lock()
defer m.mu.Unlock()
sub, err := m.js.Context().PullSubscribe(q.worker.Filters[0], q.worker.Name,
nats.Bind(q.worker.Stream, q.worker.Name), nats.ManualAck())
if err != nil {
@@ -709,3 +718,16 @@ func endCancelled(ctx context.Context, b *natsBuild) {
}
_ = b.msg.Term()
}
// boundKinds are the kinds of queue this machine holds a subscription to now.
func (m *natsMachine) boundKinds() []string {
m.mu.Lock()
defer m.mu.Unlock()
var out []string
for _, q := range m.subs {
if q.bound {
out = append(out, q.kind)
}
}
return out
}
+5 -15
View File
@@ -6,7 +6,6 @@ import (
"context"
"encoding/json"
"reflect"
"strings"
"testing"
"time"
@@ -595,21 +594,12 @@ func TestABusyMachineTakesNoWaitingAskWhateverItsPullsHold(t *testing.T) {
case <-time.After(10 * time.Second):
t.Fatal("builder a took nothing")
}
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")
}
// The queue is empty, so this request is answered at its expiry with a status message that lands in the pull.
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":1000000}`)); err != nil {
t.Fatal(err)
// **A busy machine holds no pull at all** (issues 1649 and 1747): the queues the running ask did not come from are let go of,
// so no message, a status message or a late delivery, can sit in one while it runs.
if kinds := a.boundKinds(); len(kinds) != 1 || kinds[0] != queueBuilds {
t.Fatalf("builder a holds %v while it builds; want only the queue its build came from", kinds)
}
other := &takeQueue{kind: queueOther, worker: broker.CheckWorkers()[1]}
time.Sleep(500 * time.Millisecond)
for i, id := range []string{"wait-1", "wait-2"} {
if err := asks.Ask(ctx, aPullCheck(id, 20+i, "head-x-000000000", "trunk-1-000000", false)); err != nil {
@@ -322,6 +322,10 @@ func TestNatsAMessageDeliveredOutsideAFetchIsNotHeldWhileTheBuilderRunsAnotherAs
t.Fatal(err)
}
}
// The long check says it is alive well within the ack wait of two seconds.
was := heartbeatWhileWorking
heartbeatWhileWorking = 400 * time.Millisecond
t.Cleanup(func() { heartbeatWhileWorking = was })
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)