Make a fetch's pull request end before the fetch does, and drop the hand-back loop that could itself take asks (issue 1649)

internal/link/builds_nats.go

@@ -370,11 +370,6 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
 			_ = msg.Ack()
 			continue
 		}
-		// **Nothing else is held while this builds** (issue 1649): an ask the server delivered to another queue's pull in
-		// the moment its fetch gave up sits in that pull's buffer, gets no heartbeat, and is handed to another machine after
-		// the ack wait while this one would still run it later. Handed back now and in the first seconds of the build.
-		handBackHeld(m.heldSources(), handBack)
-		stopHanding := m.handBackWhileBuilding(handBack)
 		// A build outlives the acknowledgement window many times over; said while it runs,
 		// as the controller says it for its own long handlers, so the server neither hands
 		// the ask to a second machine nor counts the wait against its deliveries.
@@ -384,95 +379,11 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
 		// deleted when it ends, whatever the outcome.
 		stopRunning := keepRunning(ctx, m.js.Conn(), request, m.on, heartbeatWhileWorking)
 		do(ctx, build)
-		stopHanding()
 		stopRunning()
 		close(working)
 	}
 }
 
-// heldSource is a pull that may hold a message the server already delivered to it.
-type heldSource interface {
-	// Held takes the message the pull holds, or nil when it holds none.
-	Held() *nats.Msg
-}
-
-// pullHeld is a pull subscription as a heldSource. NextMsg is refused on a pull subscription (ErrTypeSubscription), so the
-// buffer is read with Fetch, which returns what the subscription holds before it asks the server for anything; it is only
-// called when Pending says something is held.
-type pullHeld struct{ sub *nats.Subscription }
-
-func (p pullHeld) Held() *nats.Msg {
-	if pending, _, err := p.sub.Pending(); err != nil || pending == 0 {
-		return nil
-	}
-	ctx, end := context.WithTimeout(context.Background(), 20*time.Millisecond)
-	defer end()
-	got, err := p.sub.Fetch(1, nats.Context(ctx))
-	if err != nil || len(got) == 0 {
-		return nil
-	}
-	return got[0]
-}
-
-// heldSources are the machine's pulls, which may hold a message already delivered to them.
-func (m *natsMachine) heldSources() []heldSource {
-	var out []heldSource
-	for _, q := range m.subs {
-		if q.bound && q.sub != nil {
-			out = append(out, pullHeld{q.sub})
-		}
-	}
-	return out
-}
-
-// handBack gives a message nobody here will start back to the queue, for another holder.
-func handBack(msg *nats.Msg) {
-	if err := msg.Nak(); err != nil {
-		fmt.Fprintf(os.Stderr, "could not hand a held ask back to the queue: %v\n", err)
-		return
-	}
-	fmt.Fprintln(os.Stderr, "an ask delivered to this machine's pull while it was not fetching is handed back, not held")
-}
-
-// handBackHeld hands back every message the sources hold, none waited for, and says how many.
-func handBackHeld(sources []heldSource, hand func(*nats.Msg)) int {
-	n := 0
-	for _, s := range sources {
-		for i := 0; i < 32; i++ {
-			msg := s.Held()
-			if msg == nil {
-				break
-			}
-			hand(msg)
-			n++
-		}
-	}
-	return n
-}
-
-// heldLookEvery is how often a machine that is building looks for an ask delivered to one of its pulls.
-const heldLookEvery = time.Second
-
-// handBackWhileBuilding hands back what a pull is delivered during a build, looking every second until the stop, which
-// ends it and waits for it.
-func (m *natsMachine) handBackWhileBuilding(hand func(*nats.Msg)) (stop func()) {
-	done, finished := make(chan struct{}), make(chan struct{})
-	go func() {
-		defer close(finished)
-		tick := time.NewTicker(heldLookEvery)
-		defer tick.Stop()
-		for {
-			select {
-			case <-done:
-				return
-			case <-tick.C:
-				handBackHeld(m.heldSources(), hand)
-			}
-		}
-	}()
-	return func() { close(done); <-finished }
-}
-
 // 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 {
@@ -493,6 +404,7 @@ const runningFreshFor = 25 * time.Second
 
 // duplicateOf says why an ask is a second hand-out of one another builder is running, or nothing.
 func duplicateOf(running CheckRunning, found bool, r BuildRequest, me string) string {
+	// Seen is the server's time and runningFreshFor is judged against this builder's clock: a skew over about 35 s would matter.
 	// A key not renewed for runningFreshFor is a builder that has gone: its ask is redelivered after the ack wait, and
 	// that is the recovery, not a duplicate (the key itself lives two minutes).
 	if found && running.Build == r.ID && running.Builder != me && running.Same(r) && time.Since(running.Seen) < runningFreshFor {

internal/link/checks_nats.go

@@ -253,11 +253,25 @@ func watchPullIn(ctx context.Context, buckets []string, open func(ctx context.Co
 
 // checkTakeWait is how long a taking loop waits on one queue for an ask that is not there, and takeIdleWait how long it
 // waits on the first when all three are empty.
+//
+// **A fetch's pull request must end before the fetch does** (issue 1649). nats.go gives the server-side request an expiry a tenth
+// shorter than the fetch's wait (fetchExpiryOf), and the server then answers with a status message at the expiry; a wait so
+// short that the answer can land after the fetch gave up left a status message in the pull's buffer, and anything a request
+// could still deliver would be held by a machine that is busy. Each wait leaves at least minFetchMargin between the two.
 const (
-	checkTakeWait = 30 * time.Millisecond
+	checkTakeWait = 100 * time.Millisecond
 	takeIdleWait  = time.Second
 )
 
+// minFetchMargin is the least time between a fetch's server-side expiry and the end of its wait.
+const minFetchMargin = 10 * time.Millisecond
+
+// fetchExpiryOf is the expiry nats.go gives the pull request of a fetch that waits this long: the wait less a tenth of it, or
+// less five seconds.
+func fetchExpiryOf(wait time.Duration) time.Duration {
+	return wait - min(wait/10, 5*time.Second)
+}
+
 // CheckRunning is what a builder says of the check it is running, in the bucket of the checks running (novox/hq ADR 0406
 // §3): the head and trunk commit it checks, the builder and its build. The key is the pull request's (RunningKeyOf).
 type CheckRunning struct {

internal/link/checks_nats_integration_test.go

@@ -566,9 +566,9 @@ func TestAnAskHandedOutAgainWhileAnotherBuilderRunsItIsNotRunTwice(t *testing.T)
 	close(release)
 }
 
-// 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) {
+// A busy machine fetches nothing, so no ask waiting in the queue is taken by it, whatever status message sits in its pulls
+// (issue 1649). A pull request of our own that expires at once leaves such a message in the other checks' pull.
+func TestABusyMachineTakesNoWaitingAskWhateverItsPullsHold(t *testing.T) {
 	js := aBusWithTheBuildRole(t)
 	aKeyBucket(t, js)
 	aRunningBucket(t, js, 2*time.Minute)
@@ -595,9 +595,6 @@ func TestAnAskDeliveredToABusyMachinesIdlePullIsHandedBackAtOnce(t *testing.T) {
 	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 {
@@ -607,28 +604,29 @@ func TestAnAskDeliveredToABusyMachinesIdlePullIsHandedBackAtOnce(t *testing.T) {
 	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":5000000000}`)); err != nil {
+		[]byte(`{"batch":1,"expires":1000000}`)); 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)
+	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 {
+			t.Fatal(err)
 		}
-	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")
+	}
+	time.Sleep(5 * time.Second)
+	info, err := js.Context().ConsumerInfo(other.worker.Stream, other.worker.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.NumAckPending != 0 || info.NumRedelivered != 0 || info.Delivered.Consumer != 0 {
+		t.Fatalf("a busy machine took from the queue: %d delivered, %d pending, %d redelivered", info.Delivered.Consumer,
+			info.NumAckPending, info.NumRedelivered)
+	}
+	if n := len(queueOf(t, js).Of(AskWaiting)); n != 2 {
+		t.Fatalf("%d asks wait; want the 2 nobody took", n)
 	}
 	close(release)
 }

internal/link/held_asks_test.go

@@ -3,38 +3,18 @@ package link
 import (
 	"testing"
 	"time"
-
-	"github.com/nats-io/nats.go"
 )
 
 // An ask a machine holds but does not start gets no heartbeat, and the bus hands it to another machine after the ack
 // wait, so two machines run one check (issue 1649).
 
-type fakeHeld struct{ msgs []*nats.Msg }
-
-func (f *fakeHeld) Held() *nats.Msg {
-	if len(f.msgs) == 0 {
-		return nil
-	}
-	m := f.msgs[0]
-	f.msgs = f.msgs[1:]
-	return m
-}
-
-type emptyHeld struct{}
-
-func (emptyHeld) Held() *nats.Msg { return nil }
-
-func TestEveryMessageAPullHoldsIsHandedBackBeforeABuildStarts(t *testing.T) {
-	a := &fakeHeld{msgs: []*nats.Msg{{Subject: "a1"}, {Subject: "a2"}}}
-	b := &fakeHeld{msgs: []*nats.Msg{{Subject: "b1"}}}
-	var back []string
-	n := handBackHeld([]heldSource{a, emptyHeld{}, b}, func(m *nats.Msg) { back = append(back, m.Subject) })
-	if n != 3 || len(back) != 3 || back[0] != "a1" || back[2] != "b1" {
-		t.Fatalf("handed back %d %v; want a1 a2 b1", n, back)
-	}
-	if n := handBackHeld([]heldSource{a, b}, func(*nats.Msg) { t.Fatal("handed back what was not held") }); n != 0 {
-		t.Fatalf("a second pass handed back %d", n)
+// A fetch's server-side pull request ends clearly before the fetch does, so the server's answer and any message it could still
+// deliver come inside the fetch and never into a buffer nobody reads while the machine builds.
+func TestAFetchsPullRequestEndsClearlyBeforeTheFetchDoes(t *testing.T) {
+	for name, wait := range map[string]time.Duration{"a check queue": checkTakeWait, "the idle wait": takeIdleWait} {
+		if margin := wait - fetchExpiryOf(wait); margin < minFetchMargin {
+			t.Errorf("%s: the pull request ends %v before a fetch of %v does; want at least %v", name, margin, wait, minFetchMargin)
+		}
 	}
 }