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/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])
+	}
+}