internal/link/builds_nats.go

@@ -362,6 +362,19 @@ 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
+		}
+		// **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.
@@ -371,11 +384,95 @@ 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 what is read from a pull without asking the server for more: the messages it already holds.
+type heldSource interface {
+	NextMsg(timeout time.Duration) (*nats.Msg, error)
+}
+
+// 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, 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 after its fetch gave up 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 {
+			msg, err := s.NextMsg(0)
+			if err != nil || msg == nil {
+				break
+			}
+			hand(msg)
+			n++
+		}
+	}
+	return n
+}
+
+// handBackWhileBuilding hands back what a pull is delivered in the first seconds of a build, the time a late delivery can
+// still arrive (a pull request lives a few milliseconds past its fetch). The stop 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)
+		last := time.Duration(0)
+		for _, at := range []time.Duration{100 * time.Millisecond, 500 * time.Millisecond, 2 * time.Second} {
+			select {
+			case <-done:
+				return
+			case <-time.After(at - last):
+			}
+			last = at
+			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 {
+	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)
+}
+
+// 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 {
+	if found && running.Build == r.ID && running.Builder != me && running.Same(r) {
+		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_integration_test.go

@@ -510,3 +510,70 @@ 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 real pull answers NextMsg from its buffer without asking the server, and a pull holding nothing hands back nothing.
+func TestARealPullHoldingNothingHandsBackNothing(t *testing.T) {
+	js := aBusWithTheBuildRole(t)
+	w := broker.CheckWorkers()[0]
+	sub, err := js.Context().PullSubscribe(w.Filters[0], w.Name, nats.Bind(w.Stream, w.Name), nats.ManualAck())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n := handBackHeld([]heldSource{sub}, func(*nats.Msg) { t.Fatal("held nothing") }); n != 0 {
+		t.Fatalf("handed back %d", n)
+	}
+}

internal/link/held_asks_test.go

@@ -0,0 +1,61 @@
+package link
+
+import (
+	"errors"
+	"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) NextMsg(time.Duration) (*nats.Msg, error) {
+	if len(f.msgs) == 0 {
+		return nil, nats.ErrTimeout
+	}
+	m := f.msgs[0]
+	f.msgs = f.msgs[1:]
+	return m, nil
+}
+
+type brokenHeld struct{}
+
+func (brokenHeld) NextMsg(time.Duration) (*nats.Msg, error) { return nil, errors.New("closed") }
+
+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, brokenHeld{}, 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)
+	}
+}
+
+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()}
+	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},
+		"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)
+		}
+	}
+}