Read a pull's held asks with Fetch, hand them back all through a build, and drop a duplicate only while its builder's key is fresh (issue 1649)

internal/link/builds_nats.go

@@ -390,9 +390,28 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
 	}
 }
 
-// heldSource is what is read from a pull without asking the server for more: the messages it already holds.
+// heldSource is a pull that may hold a message the server already delivered to it.
 type heldSource interface {
-	NextMsg(timeout time.Duration) (*nats.Msg, error)
+	// 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.
@@ -400,7 +419,7 @@ func (m *natsMachine) heldSources() []heldSource {
 	var out []heldSource
 	for _, q := range m.subs {
 		if q.bound && q.sub != nil {
-			out = append(out, q.sub)
+			out = append(out, pullHeld{q.sub})
 		}
 	}
 	return out
@@ -412,16 +431,16 @@ func handBack(msg *nats.Msg) {
 		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")
+	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 {
-			msg, err := s.NextMsg(0)
-			if err != nil || msg == nil {
+		for i := 0; i < 32; i++ {
+			msg := s.Held()
+			if msg == nil {
 				break
 			}
 			hand(msg)
@@ -431,21 +450,24 @@ func handBackHeld(sources []heldSource, hand func(*nats.Msg)) int {
 	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.
+// 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)
-		last := time.Duration(0)
-		for _, at := range []time.Duration{100 * time.Millisecond, 500 * time.Millisecond, 2 * time.Second} {
+		tick := time.NewTicker(heldLookEvery)
+		defer tick.Stop()
+		for {
 			select {
 			case <-done:
 				return
-			case <-time.After(at - last):
+			case <-tick.C:
+				handBackHeld(m.heldSources(), hand)
 			}
-			last = at
-			handBackHeld(m.heldSources(), hand)
 		}
 	}()
 	return func() { close(done); <-finished }
@@ -465,9 +487,15 @@ func (m *natsMachine) runningElsewhere(ctx context.Context, r BuildRequest) stri
 	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 {
-	if found && running.Build == r.ID && running.Builder != me && running.Same(r) {
+	// 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 ""

internal/link/checks_nats.go

@@ -321,6 +321,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

@@ -3,10 +3,10 @@
 package link
 
 import (
-	"strings"
 	"context"
 	"encoding/json"
 	"reflect"
+	"strings"
 	"testing"
 	"time"
 
@@ -566,19 +566,6 @@ func TestAnAskHandedOutAgainWhileAnotherBuilderRunsItIsNotRunTwice(t *testing.T)
 	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)
-	}
-}
-
 // 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) {

internal/link/held_asks_test.go

@@ -1,7 +1,6 @@
 package link
 
 import (
-	"errors"
 	"testing"
 	"time"
 
@@ -13,24 +12,24 @@ import (
 
 type fakeHeld struct{ msgs []*nats.Msg }
 
-func (f *fakeHeld) NextMsg(time.Duration) (*nats.Msg, error) {
+func (f *fakeHeld) Held() *nats.Msg {
 	if len(f.msgs) == 0 {
-		return nil, nats.ErrTimeout
+		return nil
 	}
 	m := f.msgs[0]
 	f.msgs = f.msgs[1:]
-	return m, nil
+	return m
 }
 
-type brokenHeld struct{}
+type emptyHeld struct{}
 
-func (brokenHeld) NextMsg(time.Duration) (*nats.Msg, error) { return nil, errors.New("closed") }
+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, brokenHeld{}, b}, func(m *nats.Msg) { back = append(back, m.Subject) })
+	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)
 	}