diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index 35515a4e..f822be2c 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -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 +} diff --git a/internal/link/checks_nats_integration_test.go b/internal/link/checks_nats_integration_test.go index 5e535632..c6aee833 100644 --- a/internal/link/checks_nats_integration_test.go +++ b/internal/link/checks_nats_integration_test.go @@ -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 { diff --git a/internal/link/checks_wait_integration_test.go b/internal/link/checks_wait_integration_test.go index cbdf0806..77d7c2ff 100644 --- a/internal/link/checks_wait_integration_test.go +++ b/internal/link/checks_wait_integration_test.go @@ -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)