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

@@ -362,6 +362,14 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
 			_ = msg.Nak()
 			continue
 		}
+		// **A duplicate of a check another builder is running is dropped** (issue 1649): an ask redelivered to this machine
+		// because its first holder did not say, in time, that it had it. The running key names that builder and this same
+		// ask; running it a second time would spend a second place on one check.
+		if why := m.runningElsewhere(ctx, request); why != "" {
+			fmt.Fprintf(os.Stderr, "%s is handed out again while another builder runs it and is dropped: %s\n", request.ID, why)
+			_ = msg.Ack()
+			continue
+		}
 		// 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.
@@ -376,6 +384,35 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
 	}
 }
 
+// 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 {
+	if r.Check == nil {
+		return ""
+	}
+	running, ok, err := ReadRunning(ctx, m.js.Conn(), r)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "could not read whether %s is running elsewhere, so it runs: %v\n", r.ID, err)
+		return ""
+	}
+	return duplicateOf(running, ok, r, m.on)
+}
+
+// runningFreshFor is how recent a running key's last heartbeat must be for its builder to be taken as alive: two missed
+// heartbeats of ten seconds are not yet a dead builder, and a redelivery after a crash comes a minute after the last one.
+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 {
+		return fmt.Sprintf("%s is running it as %s since %s", running.Builder, running.Build, running.Started.Local().Format("15:04:05"))
+	}
+	return ""
+}
+
 // bind binds a queue's worker, which exists and is not created here.
 //
 // **The consumer's own filter, not the one subject this machine cares about.** The client checks what is asked for

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 {
@@ -268,6 +282,8 @@ type CheckRunning struct {
 	// Started is when the check started, kept on every heartbeat: the entry's own time is the last heartbeat's, and a
 	// reader that asks how long a check has run reads this (novox/hq ADR 0433 §6).
 	Started time.Time `json:"started"`
+	// Seen is when the key was last put, which is the last heartbeat: read from the entry, never written (issue 1649).
+	Seen time.Time `json:"-"`
 }
 
 // RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.<name>` for a group's
@@ -319,6 +335,7 @@ func ReadRunning(ctx context.Context, conn *nats.Conn, r BuildRequest) (CheckRun
 	if err := json.Unmarshal(e.Value(), &running); err != nil {
 		return CheckRunning{}, false, nil // a key of a builder from before this means no match
 	}
+	running.Seen = e.Created()
 	return running, true, nil
 }
 

internal/link/checks_nats_integration_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"encoding/json"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
@@ -510,3 +511,122 @@ func TestAnUnacknowledgedCheckGoesToASecondBuilder(t *testing.T) {
 		stop()
 	}
 }
+
+// An ask handed out a second time to another builder while the first still runs it is acknowledged and not run: the
+// running key names the other builder and this same ask (issue 1649).
+func TestAnAskHandedOutAgainWhileAnotherBuilderRunsItIsNotRunTwice(t *testing.T) {
+	js := aBusWithTheBuildRole(t)
+	aKeyBucket(t, js)
+	aRunningBucket(t, js, 2*time.Minute)
+	ctx, stop := context.WithCancel(t.Context())
+	defer stop()
+	asks := BuildsOn(js, TheBuildMachine)
+	running := make(chan string, 4)
+	release := make(chan struct{})
+	a := MachineOverNATS(js, "builder-a")
+	defer a.Close()
+	go func() {
+		_ = a.Take(ctx, func(ctx context.Context, work Build) {
+			running <- work.Request().ID
+			<-release
+			_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"})
+			_ = work.Done()
+		})
+	}()
+	check := aPullCheck("build-twice", 7, "head-7-000000000", "trunk-1-000000", false)
+	if err := asks.Ask(ctx, check); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-running:
+	case <-time.After(10 * time.Second):
+		t.Fatal("builder a took nothing")
+	}
+	// The bus's redelivery, as it reaches the other machine: the same ask, the same id.
+	if err := asks.Ask(ctx, check); 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.Done()
+		})
+	}()
+	select {
+	case id := <-took:
+		t.Fatalf("builder b ran %s, which builder a is running", id)
+	case <-time.After(3 * time.Second):
+	}
+	if n := len(queueOf(t, js).Of(AskWaiting)); n != 0 {
+		t.Errorf("%d request(s) still wait; builder b was to acknowledge the duplicate", n)
+	}
+	close(release)
+}
+
+// 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)
+	ctx, stop := context.WithCancel(t.Context())
+	defer stop()
+	asks := BuildsOn(js, TheBuildMachine)
+	busy := make(chan string, 4)
+	release := make(chan struct{})
+	a := MachineOverNATS(js, "builder-a").(*natsMachine)
+	defer a.Close()
+	go func() {
+		_ = a.Take(ctx, func(ctx context.Context, work Build) {
+			busy <- work.Request().ID
+			<-release
+			_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "builder-a", Failed: "no"})
+			_ = work.Done()
+		})
+	}()
+	if err := asks.Ask(ctx, BuildRequest{ID: "busy", Repository: "/d"}); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-busy:
+	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)
+	}
+	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)
+		}
+	}
+	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

@@ -0,0 +1,43 @@
+package link
+
+import (
+	"testing"
+	"time"
+)
+
+// 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).
+
+// 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)
+		}
+	}
+}
+
+func TestAnAskAnotherBuilderIsRunningIsADuplicateAndYoursIsNot(t *testing.T) {
+	r := aPullCheck("build-1", 423, "head-4771ee53aa", "trunk-0123456789", false)
+	run := CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1", Started: time.Now().Add(-time.Minute), Seen: time.Now().Add(-4 * time.Second)}
+	dead := run
+	dead.Seen = time.Now().Add(-60 * time.Second)
+	for name, c := range map[string]struct {
+		running CheckRunning
+		found   bool
+		me      string
+		dup     bool
+	}{
+		"another builder runs this very ask":                {run, true, "ace", true},
+		"its builder stopped renewing the key a minute ago": {dead, true, "ace", false},
+		"this builder runs it itself":                       {run, true, "shanks", false},
+		"nobody runs it":                                    {CheckRunning{}, false, "ace", false},
+		"another ask of the same head":                      {CheckRunning{Head: r.Ref, Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-0"}, true, "ace", false},
+		"the same ask at another head":                      {CheckRunning{Head: "head-other-0000", Trunk: "trunk-0123456789", Builder: "shanks", Build: "build-1"}, true, "ace", false},
+	} {
+		if got := duplicateOf(c.running, c.found, r, c.me) != ""; got != c.dup {
+			t.Errorf("%s: duplicate = %v; want %v", name, got, c.dup)
+		}
+	}
+}