From 60253c82b739c7e3f9f3eac7a4d18521c97c5d96 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 20:17:37 +0200 Subject: [PATCH] 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 | 2 +- cmd/mesh-controller/plans_building_test.go | 2 + internal/link/builds_nats.go | 3 + internal/link/checks_nats.go | 8 ++- internal/link/checks_wait_integration_test.go | 58 ++++++++++++++++ internal/link/holds.go | 69 ++++++++++++++++--- internal/link/stray_test.go | 18 +++++ 7 files changed, 149 insertions(+), 11 deletions(-) diff --git a/cmd/mesh-controller/plans_building.go b/cmd/mesh-controller/plans_building.go index f2c9d58e..46ba06c3 100644 --- a/cmd/mesh-controller/plans_building.go +++ b/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 } } diff --git a/cmd/mesh-controller/plans_building_test.go b/cmd/mesh-controller/plans_building_test.go index c65c553a..8aab90c0 100644 --- a/cmd/mesh-controller/plans_building_test.go +++ b/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}, diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index 8883c0be..35515a4e 100644 --- a/internal/link/builds_nats.go +++ b/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 diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index b720bbf4..f1271e9b 100644 --- a/internal/link/checks_nats.go +++ b/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.` for a group's -// composed check; none for a try, which no other request replaces. +// composed check; `try.` 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: diff --git a/internal/link/checks_wait_integration_test.go b/internal/link/checks_wait_integration_test.go index 1006a438..cbdf0806 100644 --- a/internal/link/checks_wait_integration_test.go +++ b/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) + } +} diff --git a/internal/link/holds.go b/internal/link/holds.go index dabf05d3..0f8296f2 100644 --- a/internal/link/holds.go +++ b/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 } diff --git a/internal/link/stray_test.go b/internal/link/stray_test.go index 31d2a495..83e11678 100644 --- a/internal/link/stray_test.go +++ b/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") + } +}