Fix the review of the checks-wait milestone: redeliveries, deleted modules, the kept queue

A check pending redelivery is read before the let-through fetches (none is handed back);
a deleted module is done for the plan's key; the queue an ask came from is let go of when
it ends. Tries say they run under their own id.

cmd/mesh-controller/plans_building.go

@@ -42,7 +42,7 @@ func buildPhaseOf(p inventory.Plan) bool {
 			if st != nil && st.State != "" {
 				asked = true
 			}
-			if st == nil || st.State != "built" {
+			if st == nil || (st.State != "built" && st.State != planDeleted) {
 				unbuilt = true
 			}
 		}

cmd/mesh-controller/plans_building_test.go

@@ -81,6 +81,8 @@ func TestAPlanIsInItsBuildPhaseFromTheFirstAskToTheLastBuild(t *testing.T) {
 		{"between two tiers", aPlan("p", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built"}), true},
 		{"the last tier asked", aPlan("p", inventory.PlanBuilding, tiers, map[string]string{"a": "built", "b": "built", "c": "asked"}), true},
 		{"every build built", aPlan("p", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built", "c": "built"}), false},
+		// A module the plan deletes is done as far as its tiers go (release_plan.go), so it holds no key through the rollout.
+		{"a deleted module is done", aPlan("p", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built", "c": planDeleted}), false},
 		{"failed", aPlan("p", inventory.PlanFailed, tiers, map[string]string{"a": "failed", "b": "asked"}), false},
 		{"superseded", aPlan("p", inventory.PlanSuperseded, tiers, map[string]string{"a": "asked"}), false},
 		{"a batch whose window is open", aPlan("p", inventory.PlanAssembling, tiers, map[string]string{"a": "asked"}), false},

internal/link/builds_nats.go

@@ -427,6 +427,9 @@ func (m *natsMachine) handOver(ctx context.Context, msg *nats.Msg, q *takeQueue,
 	do(ctx, build)
 	stopRunning()
 	close(working)
+	// **The queue it came from too, once the ask is settled**: a fetch that returned a late message leaves its pull request
+	// open, and the server can deliver one more into the subscription (issue 1747). Take binds every queue again first.
+	m.quiesce(nil)
 }
 
 // quiesce lets go of every queue but the one the running ask came from, for as long as it runs. **A builder never holds a

internal/link/checks_nats.go

@@ -287,12 +287,16 @@ type CheckRunning struct {
 }
 
 // RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.<name>` for a group's
-// composed check; none for a try, which no other request replaces.
+// composed check; `try.<id>` for a try, which no other request replaces.
 func RunningKeyOf(r BuildRequest) (string, bool) {
 	c := r.Check
 	switch {
-	case c == nil, c.Try:
+	case c == nil:
 		return "", false
+	case c.Try:
+		// A try says it runs under its own id, which no other request shares: so a reader of the bucket can tell a try that runs
+		// from a check left waiting for a redelivery (ADR 0483 clause 3).
+		return "try." + tokenOf(r.ID), true
 	case c.Group != "":
 		return "group." + tokenOf(c.Group), true
 	case c.Number <= 0:

internal/link/checks_wait_integration_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/nats-io/nats.go"
 	"reflect"
 	"strings"
 	"sync"
@@ -415,3 +416,60 @@ func TestNatsAMessageDeliveredOutsideAFetchIsNotHeldWhileTheBuilderRunsAnotherAs
 		t.Fatalf("the stray ran on %v, want once", runs[stray.ID])
 	}
 }
+
+// Review of 1688: a young check pending redelivery (handed back by a Nak) stands ahead of a check older than 30 minutes in the
+// same queue while a plan builds. The let-through fetch would return the young one first, so nothing is fetched: the young
+// check's delivery count stays 1, it is not handed back every second until it is dead-lettered, and it is not run.
+func TestNatsAYoungRedeliveryAheadOfAnOldCheckIsNeverFetchedByTheLetThrough(t *testing.T) {
+	js, plans := aBusWithTheBuckets(t)
+	ctx, stop := context.WithCancel(t.Context())
+	defer stop()
+	asks := BuildsOn(js, TheBuildMachine)
+	young := aCheckAskedAgo(time.Minute, 1, false)
+	if err := asks.Ask(ctx, young); err != nil {
+		t.Fatal(err)
+	}
+	// Someone took it and handed it back: it is pending redelivery.
+	worker := broker.CheckWorkers()[1]
+	sub, err := js.Context().PullSubscribe(worker.Filters[0], worker.Name, nats.Bind(worker.Stream, worker.Name), nats.ManualAck())
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := sub.Fetch(1, nats.MaxWait(5*time.Second))
+	if err != nil || len(got) != 1 {
+		t.Fatalf("the young check was not fetched to be handed back: %v", err)
+	}
+	if err := got[0].Nak(); err != nil {
+		t.Fatal(err)
+	}
+	_ = sub.Unsubscribe()
+	old := aCheckAskedAgo(45*time.Minute, 2, false)
+	if err := asks.Ask(ctx, old); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil {
+		t.Fatal(err)
+	}
+	started := make(chan string, 4)
+	machine := MachineOverNATS(js, "ace")
+	defer machine.Close()
+	go func() {
+		_ = machine.Take(ctx, func(ctx context.Context, work Build) {
+			started <- work.Request().ID
+			_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "ace", Failed: "no"})
+			_ = work.Done()
+		})
+	}()
+	select {
+	case id := <-started:
+		t.Fatalf("%s was run while the young redelivery stood ahead of it", id)
+	case <-time.After(6 * time.Second):
+	}
+	info, err := js.Context().ConsumerInfo(worker.Stream, worker.Name)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.Delivered.Consumer != 1 || info.NumRedelivered != 0 {
+		t.Fatalf("the young check was handed out %d time(s), %d redelivered: the let-through fetched it", info.Delivered.Consumer, info.NumRedelivered)
+	}
+}

internal/link/holds.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -90,22 +91,74 @@ func BuildingPlans(ctx context.Context, conn *nats.Conn) ([]string, error) {
 	return keys, nil
 }
 
-// OldestWaiting reads the first ask after the last one the check queue's worker handed out: a direct get on the stream, which
-// takes nothing from it and counts against no delivery. The ask's own id says when it was asked.
+// OldestWaiting is when the check a fetch of this queue would return was asked, read without taking anything.
+//
+// **It sees redeliveries** (review of issue 1688). A work queue keeps only what is not acknowledged: what the worker handed out
+// (sequence up to its delivered mark) is running or waiting to be handed out again, and a fetch returns those before anything
+// new. An ask there that is not said running (a key of the checks running names its build) is such a redelivery. If one is
+// younger than the bound, the queue's next fetch would return it, and a young check is never let through: the queue is not a
+// candidate, and nothing is fetched, so nothing is handed back and no delivery is spent. Otherwise the candidate is the oldest
+// of the redeliveries and the first ask after the delivered mark.
 func (g busGate) OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error) {
 	info, err := jsOf(g.conn).ConsumerInfo(q.worker.Stream, q.worker.Name, nats.Context(ctx))
 	if err != nil {
 		return time.Time{}, false, fmt.Errorf("the %s worker: %w", q.kind, err)
 	}
-	body, _ := json.Marshal(map[string]any{"seq": info.Delivered.Stream + 1, "next_by_subj": q.worker.Filters[0]})
+	var oldest time.Time
+	note := func(at time.Time) {
+		if oldest.IsZero() || at.Before(oldest) {
+			oldest = at
+		}
+	}
+	for seq := uint64(1); seq <= info.Delivered.Stream; {
+		data, stored, got, err := g.directGet(ctx, q, seq)
+		if err != nil {
+			return time.Time{}, false, err
+		}
+		if got == 0 || got > info.Delivered.Stream {
+			break
+		}
+		seq = got + 1
+		var r BuildRequest
+		if json.Unmarshal(data, &r) != nil {
+			continue
+		}
+		running, ok, err := ReadRunning(ctx, g.conn, r)
+		if err != nil {
+			return time.Time{}, false, fmt.Errorf("the checks running: %w", err)
+		}
+		if ok && running.Build == r.ID {
+			continue // running now
+		}
+		at := askedAtOf(data, stored)
+		if time.Since(at) < LetThroughAfter {
+			return time.Time{}, false, nil // a young redelivery is next in line
+		}
+		note(at)
+	}
+	data, stored, got, err := g.directGet(ctx, q, info.Delivered.Stream+1)
+	if err != nil {
+		return time.Time{}, false, err
+	}
+	if got != 0 {
+		note(askedAtOf(data, stored))
+	}
+	return oldest, !oldest.IsZero(), nil
+}
+
+// directGet reads the first ask at or after a sequence of a check queue's stream, without taking it: its body, the time the
+// stream stored it, and its sequence (0 when there is none).
+func (g busGate) directGet(ctx context.Context, q *takeQueue, seq uint64) ([]byte, string, uint64, error) {
+	body, _ := json.Marshal(map[string]any{"seq": seq, "next_by_subj": q.worker.Filters[0]})
 	reply, err := g.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+q.worker.Stream, body)
 	if err != nil {
-		return time.Time{}, false, fmt.Errorf("the first waiting ask of the %s queue: %w", q.kind, err)
+		return nil, "", 0, fmt.Errorf("the asks of the %s queue: %w", q.kind, err)
 	}
-	if reply.Header.Get("Status") == "404" || reply.Header.Get("Status") == "408" {
-		return time.Time{}, false, nil
+	if st := reply.Header.Get("Status"); st == "404" || st == "408" {
+		return nil, "", 0, nil
 	}
-	return askedAtOf(reply.Data, reply.Header.Get("Nats-Time-Stamp")), true, nil
+	got, _ := strconv.ParseUint(reply.Header.Get("Nats-Sequence"), 10, 64)
+	return reply.Data, reply.Header.Get("Nats-Time-Stamp"), got, nil
 }
 
 // askedAtOf is when a request was asked: its id says (NewBuildID), the stream's own time otherwise, and a request that says
@@ -275,7 +328,7 @@ func (m *natsMachine) letOneThrough(ctx context.Context) (*nats.Msg, *takeQueue)
 	// Another holder took the one that was oldest in the moment between the read and the fetch: what came is a younger check,
 	// handed back and the key released, never run during a delivery.
 	if askedAtOf(fetched[0].Data, "").Add(m.letThroughAfter()).After(time.Now()) {
-		_ = fetched[0].Nak()
+		_ = fetched[0].NakWithDelay(time.Minute) // a delay keeps it off the queue of redeliveries a fetch reads first
 		m.dropLetThrough(ctx)
 		return nil, nil
 	}

internal/link/stray_test.go

@@ -2,6 +2,7 @@ package link
 
 import (
 	"testing"
+	"time"
 
 	"github.com/nats-io/nats.go"
 )
@@ -34,3 +35,20 @@ func TestARunningAskLetsGoOfTheQueuesItDidNotComeFrom(t *testing.T) {
 		t.Errorf("fetched from a queue that was let go of: builds %d, other %d", builds.fetches(), other.fetches())
 	}
 }
+
+// Once the ask is settled the queue it came from is let go of too: a second message the server delivered into its subscription
+// on a pull request still open is not run after the ask ends (review of 1747).
+func TestTheQueueTheAskCameFromIsLetGoOfWhenTheAskEnds(t *testing.T) {
+	m, _, ready, _, _ := aMachine()
+	ready.waiting = []*nats.Msg{askedAgo(time.Minute, 2, true)}
+	m.quiesce(nil)
+	for _, q := range m.subs {
+		if q.bound || q.pull != nil {
+			t.Errorf("the %s queue is still held after the ask ended", q.kind)
+		}
+	}
+	m.subs[0].bound, m.subs[0].pull = true, &fakeQueue{}
+	if msg, _, _ := m.next(t.Context()); msg != nil || ready.fetches() != 0 {
+		t.Fatal("a message held in the kept queue's subscription was taken after the ask ended")
+	}
+}