diff --git a/cmd/mesh-controller/plan_looks.go b/cmd/mesh-controller/plan_looks.go index f0098525..29297b24 100644 --- a/cmd/mesh-controller/plan_looks.go +++ b/cmd/mesh-controller/plan_looks.go @@ -206,6 +206,7 @@ func servingPlanSaved(ctx context.Context, bus link.Bus, open *stores) func(*inv return func(_ *inventory.Inventory, p inventory.Plan) { sayPlanMoved(ctx, bus, open.inventory, p) planKept(p) + keepBuildingKey(ctx, p) if p.WalkSaid == inventory.WalkEnded { go forgetChecksOfWalk(context.WithoutCancel(ctx), open.inventory, p) } diff --git a/cmd/mesh-controller/plans_building.go b/cmd/mesh-controller/plans_building.go new file mode 100644 index 00000000..f2c9d58e --- /dev/null +++ b/cmd/mesh-controller/plans_building.go @@ -0,0 +1,238 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "sync/atomic" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" +) + +// The plans building (novox/hq ADR 0483 clause 2, issue 1687). +// +// **A key per delivery plan in its build phase, so that the build seat takes no new check while a delivery builds.** The phase +// runs from the ask of the plan's first build until every build of its last tier is built, or the plan fails, stops or waits +// on the operator. It is derived from the plan alone (buildPhaseOf), so a controller that restarts finds the same answer. The +// key is put when the first build is asked (sayBuilding), kept current at every save of the plan (keepBuildingKey) and by a +// look every link.PlansBuildingRefresh while this controller holds the lease (keepingPlansBuilding), and deleted at the phase's +// end; the bucket's age removes the key of a controller that stopped, so checks are held at most that long. +// +// The layout is documented in internal/link/holds.go. + +// buildPhaseOf says whether a plan is in its build phase: open, not waiting for the operator's word, with a build asked and a +// build of its tiers still to be built. A tier built and sent leaves the plan rolling while later tiers are still to be asked; +// it is in the phase still, which is what keeps a check from taking a place between two tiers. +func buildPhaseOf(p inventory.Plan) bool { + if !p.Open() || p.Waiting() || p.Release != nil { + return false + } + asked, unbuilt := false, false + for _, tier := range p.Tiers { + for _, m := range tier { + st := p.Modules[m] + if st != nil && st.State != "" { + asked = true + } + if st == nil || st.State != "built" { + unbuilt = true + } + } + } + return asked && unbuilt +} + +// buildingSince is when the plan's first build was asked, or fallback when no module says. +func buildingSince(p inventory.Plan, fallback time.Time) time.Time { + since := time.Time{} + for _, st := range p.Modules { + if st != nil && st.AskedAt != nil && (since.IsZero() || st.AskedAt.Before(since)) { + since = *st.AskedAt + } + } + if since.IsZero() { + return fallback + } + return since +} + +// buildingStore is the bucket of the plans building. +type buildingStore interface { + Put(ctx context.Context, plan string, value []byte) error + Delete(ctx context.Context, plan string) error + Keys(ctx context.Context) ([]string, error) +} + +// syncBuilding makes the bucket say what the plans say: a key for each plan in its build phase, put again, and none for any +// other plan, deleted. It returns the faults, each once. +func syncBuilding(ctx context.Context, store buildingStore, plans []inventory.Plan, now time.Time) []error { + var faults []error + live := map[string]bool{} + for _, p := range plans { + if !buildPhaseOf(p) { + continue + } + live[p.ID] = true + if err := putBuilding(ctx, store, p.ID, buildingSince(p, now), now); err != nil { + faults = append(faults, err) + } + } + keys, err := store.Keys(ctx) + if err != nil { + return append(faults, err) + } + for _, key := range keys { + if live[key] { + continue + } + if err := store.Delete(ctx, key); err != nil { + faults = append(faults, fmt.Errorf("the key of %s could not be taken away: %w", key, err)) + } + } + return faults +} + +func putBuilding(ctx context.Context, store buildingStore, plan string, since, now time.Time) error { + body, _ := json.Marshal(link.PlanBuilding{Plan: plan, Since: since.UTC(), Seen: now.UTC()}) + if err := store.Put(ctx, plan, body); err != nil { + return fmt.Errorf("the key of %s could not be put: %w", plan, err) + } + return nil +} + +// busBuilding is the store on the controller's bucket. +type busBuilding struct { + conn *nats.Conn + mu sync.Mutex + kv jetstream.KeyValue +} + +func (b *busBuilding) bucket(ctx context.Context) (jetstream.KeyValue, error) { + b.mu.Lock() + defer b.mu.Unlock() + if b.kv != nil { + return b.kv, nil + } + api, err := jetstream.New(b.conn) + if err != nil { + return nil, err + } + kv, err := api.KeyValue(ctx, broker.PlansBuildingBucket) + if err != nil { + return nil, fmt.Errorf("the plans building: %w", err) + } + b.kv = kv + return kv, nil +} + +func (b *busBuilding) Put(ctx context.Context, plan string, value []byte) error { + kv, err := b.bucket(ctx) + if err != nil { + return err + } + _, err = kv.Put(ctx, plan, value) + return err +} + +func (b *busBuilding) Delete(ctx context.Context, plan string) error { + kv, err := b.bucket(ctx) + if err != nil { + return err + } + if err := kv.Delete(ctx, plan); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + return nil +} + +func (b *busBuilding) Keys(ctx context.Context) ([]string, error) { + kv, err := b.bucket(ctx) + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return keys, err +} + +// buildingKeys is the store of the serving controller; nil in a process that is not it, which says nothing. +var buildingKeys atomic.Pointer[buildingStore] + +func setBuildingStore(s buildingStore) { + if s == nil { + buildingKeys.Store(nil) + return + } + buildingKeys.Store(&s) +} + +// sayBuilding puts a plan's key: its first build is being asked (or another of its builds). A key that could not be put is +// said; the look that follows puts it again. +func sayBuilding(ctx context.Context, plan string, since time.Time) { + s := buildingKeys.Load() + if s == nil { + return + } + if err := putBuilding(ctx, *s, plan, since, time.Now()); err != nil { + fmt.Printf("plans: %v; checks may take a place of %s until the next look\n", err, plan) + } +} + +// keepBuildingKey is told each plan this controller keeps: its key put again while it builds and taken away once it does not. +func keepBuildingKey(ctx context.Context, p inventory.Plan) { + s := buildingKeys.Load() + if s == nil { + return + } + if buildPhaseOf(p) { + sayBuilding(ctx, p.ID, buildingSince(p, time.Now())) + return + } + if err := (*s).Delete(ctx, p.ID); err != nil { + fmt.Printf("plans: the key of %s could not be taken away; it ages out: %v\n", p.ID, err) + } +} + +// keepingPlansBuilding looks at the plans every link.PlansBuildingRefresh for as long as this controller serves and holds the +// lease, so a key is put again before its age and the key of a plan that left the phase unseen is taken away. +func keepingPlansBuilding(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn) { + store := &busBuilding{conn: conn} + setBuildingStore(store) + defer setBuildingStore(nil) + tick := time.NewTicker(broker.PlansBuildingRefresh) + defer tick.Stop() + failing := "" + for { + if epoch, err := theLease.epoch(ctx); err == nil && epoch != 0 { + why := "" + if plans, err := inv.OpenPlans(ctx); err != nil { + why = fmt.Sprintf("the plans cannot be read: %v", err) + } else if faults := syncBuilding(ctx, store, plans, time.Now()); len(faults) > 0 { + why = fmt.Sprint(faults) + } + if why != failing { + if why != "" { + fmt.Printf("plans: the keys of the plans building cannot be kept: %s\n", why) + } else { + fmt.Printf("plans: the keys of the plans building are kept again\n") + } + failing = why + } + } + select { + case <-ctx.Done(): + return + case <-tick.C: + } + } +} diff --git a/cmd/mesh-controller/plans_building_test.go b/cmd/mesh-controller/plans_building_test.go new file mode 100644 index 00000000..c65c553a --- /dev/null +++ b/cmd/mesh-controller/plans_building_test.go @@ -0,0 +1,173 @@ +package main + +import ( + "context" + "encoding/json" + "sort" + "sync" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" +) + +// Issue 1687 (ADR 0483 clause 2): a delivery plan's key lives from the ask of its first build until every build of its last +// tier is built, or the plan fails, stops or waits on the operator. + +type fakeBuildingStore struct { + mu sync.Mutex + keys map[string]link.PlanBuilding + puts map[string]int +} + +func newFakeBuildingStore() *fakeBuildingStore { + return &fakeBuildingStore{keys: map[string]link.PlanBuilding{}, puts: map[string]int{}} +} + +func (f *fakeBuildingStore) Put(_ context.Context, plan string, value []byte) error { + f.mu.Lock() + defer f.mu.Unlock() + var v link.PlanBuilding + if err := json.Unmarshal(value, &v); err != nil { + return err + } + f.keys[plan] = v + f.puts[plan]++ + return nil +} + +func (f *fakeBuildingStore) Delete(_ context.Context, plan string) error { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.keys, plan) + return nil +} + +func (f *fakeBuildingStore) Keys(context.Context) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + var out []string + for k := range f.keys { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +func aPlan(id, state string, tiers [][]string, states map[string]string) inventory.Plan { + p := inventory.Plan{ID: id, State: state, Tiers: tiers, Modules: map[string]*inventory.PlanModule{}} + now := time.Now().Add(-time.Minute) + for m, st := range states { + pm := &inventory.PlanModule{State: st} + if st != "" { + pm.AskedAt = &now + } + p.Modules[m] = pm + } + return p +} + +func TestAPlanIsInItsBuildPhaseFromTheFirstAskToTheLastBuild(t *testing.T) { + tiers := [][]string{{"a", "b"}, {"c"}} + cases := []struct { + name string + plan inventory.Plan + want bool + }{ + {"no build asked yet", aPlan("p", inventory.PlanBuilding, tiers, nil), false}, + {"the first tier asked", aPlan("p", inventory.PlanBuilding, tiers, map[string]string{"a": "asked", "b": "asked"}), true}, + // Between two tiers: the first built and sent (rolling), the second not asked yet. + {"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}, + {"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}, + } + for _, c := range cases { + if got := buildPhaseOf(c.plan); got != c.want { + t.Errorf("%s: in the build phase = %v, want %v", c.name, got, c.want) + } + } + waiting := aPlan("p", inventory.PlanBuilding, tiers, map[string]string{"a": "asked"}) + waiting.Delivery = &inventory.PlanDelivery{Awaits: "the operator's word"} + if buildPhaseOf(waiting) { + t.Error("a plan that waits on the operator is in its build phase") + } +} + +// The key's life: put, refreshed, deleted at the phase's end; a plan that left the phase unseen loses its key at the next look. +func TestAPlanKeyIsPutRefreshedAndDeletedAtThePhasesEnd(t *testing.T) { + ctx := context.Background() + store := newFakeBuildingStore() + tiers := [][]string{{"a"}, {"b"}} + building := aPlan("plan-1", inventory.PlanBuilding, tiers, map[string]string{"a": "asked"}) + + if faults := syncBuilding(ctx, store, []inventory.Plan{building}, time.Now()); len(faults) > 0 { + t.Fatal(faults) + } + if _, ok := store.keys["plan-1"]; !ok { + t.Fatal("no key while the first build is asked") + } + first := store.keys["plan-1"] + // Refreshed: put again, the phase's start kept. + time.Sleep(2 * time.Millisecond) + syncBuilding(ctx, store, []inventory.Plan{building}, time.Now()) + if store.puts["plan-1"] != 2 || !store.keys["plan-1"].Seen.After(first.Seen) || !store.keys["plan-1"].Since.Equal(first.Since) { + t.Errorf("the key was not refreshed with its start kept: %d puts, %+v then %+v", store.puts["plan-1"], first, store.keys["plan-1"]) + } + // Between the tiers: still there. + between := aPlan("plan-1", inventory.PlanRolling, tiers, map[string]string{"a": "built"}) + syncBuilding(ctx, store, []inventory.Plan{between}, time.Now()) + if _, ok := store.keys["plan-1"]; !ok { + t.Error("the key was gone between two tiers") + } + // A key of another plan that is no longer in the phase. + store.keys["plan-0"] = link.PlanBuilding{Plan: "plan-0"} + done := aPlan("plan-1", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built"}) + syncBuilding(ctx, store, []inventory.Plan{done}, time.Now()) + if len(store.keys) != 0 { + t.Errorf("keys remain after the last build was built: %v", store.keys) + } +} + +// keepBuildingKey, told of each save, puts the key while the plan builds and deletes it when the plan fails. +func TestASavedPlanPutsOrDeletesItsKey(t *testing.T) { + ctx := context.Background() + store := newFakeBuildingStore() + var s buildingStore = store + buildingKeys.Store(&s) + defer buildingKeys.Store(nil) + + p := aPlan("plan-7", inventory.PlanBuilding, [][]string{{"a"}}, map[string]string{"a": "asked"}) + keepBuildingKey(ctx, p) + if _, ok := store.keys["plan-7"]; !ok { + t.Fatal("the key was not put when the plan was saved building") + } + p.State = inventory.PlanFailed + keepBuildingKey(ctx, p) + if _, ok := store.keys["plan-7"]; ok { + t.Fatal("the key stayed after the plan failed") + } +} + +// The plan's key is there before its first build is asked, so a check cannot take a place between the ask and the plan's save. +func TestThePlansKeyIsPutBeforeItsFirstBuildIsAsked(t *testing.T) { + store := newFakeBuildingStore() + var s buildingStore = store + buildingKeys.Store(&s) + defer buildingKeys.Store(nil) + p, byName := aTierOfThree() + keyThere := false + was := askABuild + askABuild = func(context.Context, buildSource, string, string) (string, error) { + _, keyThere = store.keys["plan-1"] + return "build-x", nil + } + t.Cleanup(func() { askABuild = was }) + askModule(t.Context(), p, "a", byName) + if !keyThere { + t.Fatal("the plan's key was not there when its first build was asked") + } +} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index e3c12494..9f38cd91 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -284,6 +284,8 @@ func serve(ctx context.Context) (err error) { go mirroringRecords(ctx, inv, bus.Conn, bus) // And the bucket of the files that travel by digest, judged against what each machine's latest declaration names (ADR 0486 §3). go sweepingDeclarationFiles(ctx, inv, bus.Conn) + // And the key of each plan in its build phase, which holds the build seat's checks while a delivery builds (ADR 0483 clause 2). + go keepingPlansBuilding(ctx, inv, bus.Conn) // Every call carries the lease's epoch, and its record is written only under the lease (novox/hq // to-be 45 §6). link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) }) diff --git a/cmd/mesh-controller/queue.go b/cmd/mesh-controller/queue.go index ea2f8fed..f82e2919 100644 --- a/cmd/mesh-controller/queue.go +++ b/cmd/mesh-controller/queue.go @@ -91,7 +91,9 @@ type queueAnswer struct { Dead int `json:"dead"` MaxDeliver int `json:"max-deliver"` Asks []queuedAnswer `json:"asks"` - Text string `json:"text"` + // HeldFor are the delivery plans building, while which the waiting checks are held (ADR 0483 clause 5); empty when none. + HeldFor []string `json:"held-for,omitempty"` + Text string `json:"text"` } // queuedAnswer is one ask as data. Never what the request carries beyond its source — not `held`, every @@ -120,6 +122,9 @@ type queuedAnswer struct { // Delivery is the how-manyth time an ask in flight was handed out; more than 1 is a restart // (novox/hq issue 594). Delivery int `json:"delivery,omitempty"` + // LetThrough is a check that waited past 30 minutes and is let through while a delivery builds: the next one, or the one + // that runs (ADR 0483 clause 5). + LetThrough bool `json:"let-through,omitempty"` } // aCommitRef is a ref that names a commit by its full hash (SHA-1 or SHA-256). @@ -130,7 +135,8 @@ var aCommitRef = regexp.MustCompile(`^(?:[0-9a-f]{40}|[0-9a-f]{64})$`) func queueAnswerOf(q link.Queue, now time.Time) queueAnswer { waiting, running, dead := q.Of(link.AskWaiting), q.Of(link.AskInFlight), q.Of(link.AskDead) out := queueAnswer{Seat: q.Seat, Waiting: len(waiting), InFlight: len(running), Dead: len(dead), - MaxDeliver: q.MaxDeliver, Asks: []queuedAnswer{}, Text: queueText(q, now)} + MaxDeliver: q.MaxDeliver, Asks: []queuedAnswer{}, HeldFor: q.Hold.Plans, Text: queueText(q, now)} + letNext, letRunning := q.LetThroughOf(now) one := func(a link.QueuedAsk, position int) queuedAnswer { kind := link.KindOf(a.Request) since := a.AskedAt @@ -143,7 +149,7 @@ func queueAnswerOf(q link.Queue, now time.Time) queueAnswer { } return queuedAnswer{ID: a.ID, State: a.State, Kind: kind, Repository: link.WithoutPassword(a.Repository), Path: a.Path, Ref: a.Ref, Commit: commit, Machine: a.On, WasOn: a.Was, AskedAt: a.AskedAt, Since: since, - Position: position, Seq: a.Seq, Delivery: a.Delivery} + Position: position, Seq: a.Seq, Delivery: a.Delivery, LetThrough: a.ID != "" && (a.ID == letNext || a.ID == letRunning)} } for _, a := range running { out.Asks = append(out.Asks, one(a, 0)) @@ -179,6 +185,11 @@ func queueText(q link.Queue, now time.Time) string { } return s } + letNext, letRunning := q.LetThroughOf(now) + const passed = "waited past 30 minutes, let through" + if q.HoldFault != "" { + fmt.Fprintf(&b, "why checks wait cannot be read, so none is said held: %s\n", q.HoldFault) + } if len(running) > 0 { fmt.Fprintln(&b, "\nin flight:") for _, a := range running { @@ -193,14 +204,27 @@ func queueText(q link.Queue, now time.Time) string { case a.Was != "": where = fmt.Sprintf("handed back by %s, to be handed out again", a.Was) } - fmt.Fprintf(&b, " %-26s %-8s %s — %s, %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), where, - ago(a.AskedAt), a.Seq) + note := "" + if a.ID != "" && a.ID == letRunning { + note = " — " + passed + } + fmt.Fprintf(&b, " %-26s %-8s %s — %s, %s (seq %d)%s\n", a.ID, link.KindOf(a.Request), what(a), where, + ago(a.AskedAt), a.Seq, note) } } if len(waiting) > 0 { fmt.Fprintln(&b, "\nwaiting, in the order they are handed out:") + // **Why the checks wait** (ADR 0483 clause 5): above them, the plans they are held for. + if len(q.Hold.Plans) > 0 && hasCheck(waiting) { + fmt.Fprintf(&b, " checks are held while delivery %s builds; the oldest check that waited 30 minutes goes first, one at a time\n", + strings.Join(q.Hold.Plans, ", ")) + } for _, a := range waiting { - fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq) + note := "" + if a.ID != "" && a.ID == letNext { + note = " — " + passed + " next" + } + fmt.Fprintf(&b, " %-26s %-8s %s — %s (seq %d)%s\n", a.ID, link.KindOf(a.Request), what(a), ago(a.AskedAt), a.Seq, note) } } if len(dead) > 0 { @@ -218,6 +242,16 @@ func queueText(q link.Queue, now time.Time) string { return b.String() } +// hasCheck says a check is among the asks. +func hasCheck(asks []link.QueuedAsk) bool { + for _, a := range asks { + if a.Request.Check != nil { + return true + } + } + return false +} + // cancelCommand drops one waiting or dead ask. func cancelCommand(ctx context.Context, args []string) error { if len(args) != 1 { diff --git a/cmd/mesh-controller/queue_held_test.go b/cmd/mesh-controller/queue_held_test.go new file mode 100644 index 00000000..2561492e --- /dev/null +++ b/cmd/mesh-controller/queue_held_test.go @@ -0,0 +1,77 @@ +package main + +import ( + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/link" +) + +// Issue 1688 / 1652 AC5 (ADR 0483 clause 5): the queue names, above the waiting checks, the plan they are held for, and says of a +// check let through that it waited past 30 minutes. + +func aHeldQueue(now time.Time) link.Queue { + check := func(id string, ago time.Duration, n int) link.QueuedAsk { + return link.QueuedAsk{Seq: uint64(n), ID: id, Repository: "ssh://forge/novox/r", AskedAt: now.Add(-ago), State: link.AskWaiting, + Request: link.BuildRequest{ID: id, Check: &link.CheckRequest{Owner: "novox", Repo: "r", Number: n}}} + } + return link.Queue{Seat: link.TheBuildMachine, Asks: []link.QueuedAsk{ + check("build-young", 5*time.Minute, 1), + check("build-old", 45*time.Minute, 2), + check("build-older", 50*time.Minute, 3), + }, Hold: link.CheckHold{Plans: []string{"plan-42"}}} +} + +func TestTheQueueNamesThePlanTheChecksAreHeldFor(t *testing.T) { + now := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC) + text := queueText(aHeldQueue(now), now) + heldAt := strings.Index(text, "held while delivery plan-42 builds") + firstAt := strings.Index(text, "build-young") + if heldAt < 0 || firstAt < 0 || heldAt > firstAt { + t.Fatalf("the plan the checks are held for is not named above the waiting checks:\n%s", text) + } + // Nothing building: nothing held. + q := aHeldQueue(now) + q.Hold = link.CheckHold{} + if strings.Contains(queueText(q, now), "held while") { + t.Fatal("checks are said held with no delivery building") + } +} + +func TestACheckLetThroughIsSaidToHaveWaitedPastThirtyMinutes(t *testing.T) { + now := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC) + q := aHeldQueue(now) + text := queueText(q, now) + var saidOf []string + for _, l := range strings.Split(text, "\n") { + if strings.Contains(l, "waited past 30 minutes, let through") { + saidOf = append(saidOf, strings.Fields(l)[0]) + } + } + if len(saidOf) != 1 || saidOf[0] != "build-older" { + t.Fatalf("the check let through next (the oldest) is not the one said: %v\n%s", saidOf, text) + } + answer := queueAnswerOf(q, now) + if len(answer.HeldFor) != 1 || answer.HeldFor[0] != "plan-42" { + t.Errorf("held-for = %v", answer.HeldFor) + } + for _, a := range answer.Asks { + if a.LetThrough != (a.ID == "build-older") { + t.Errorf("%s let-through = %v", a.ID, a.LetThrough) + } + } + // A check in flight under the let-through key is said too. + q.Asks = append(q.Asks, link.QueuedAsk{Seq: 9, ID: "build-running", State: link.AskInFlight, On: "ace", AskedAt: now.Add(-time.Hour), + Started: now.Add(-time.Minute), Request: link.BuildRequest{ID: "build-running", Check: &link.CheckRequest{Owner: "novox", Repo: "r", Number: 9}}}) + q.Hold.LetThrough = &link.LetThrough{Builder: "ace"} + if !strings.Contains(queueText(q, now), "build-running") || !strings.Contains(queueText(q, now), "waited past 30 minutes, let through\n") { + t.Errorf("the check in flight let through is not said:\n%s", queueText(q, now)) + } + // Younger than the bound: none is said. + young := aHeldQueue(now) + young.Asks = young.Asks[:1] + if strings.Contains(queueText(young, now), "waited past") { + t.Error("a check of 5 minutes is said to have waited past 30") + } +} diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index 7331626c..78ec934d 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -577,6 +577,9 @@ func askModule(ctx context.Context, p *inventory.Plan, name string, byName map[s } // **The walk takes the check's build of the same key** (novox/hq ADR 0350 §2): its ask says so, with the pull // requests it answers, whose checks name the part of a differing key. + // **The plan's key is there before its first build is** (ADR 0483 clause 2): a check taken between the ask and the plan's save + // would hold a place for its whole run. + sayBuilding(ctx, p.ID, now) id, err := askABuild(withReuse(ctx, reuseWithVerdicts(ctx, p.ID, reuseAskOf(p), mergedHeadsOf(p))), source, e.Source.Path, ref) if err != nil { diff --git a/internal/broker/checks.go b/internal/broker/checks.go index f9ab8da9..cb3b16ea 100644 --- a/internal/broker/checks.go +++ b/internal/broker/checks.go @@ -46,6 +46,7 @@ func CheckStreams() []Stream { Retention: RetentionWorkQueue, MaxAge: 7 * 24 * 60 * 60, MaxMsgsPerSubject: 1, + Direct: true, Why: "the checks of pull requests marked ready, taken after a delivery build and before any other check " + "(novox/hq ADR 0406 §3); one subject a pull request, the newest request kept, so a newer one replaces a " + "request still waiting", @@ -56,6 +57,7 @@ func CheckStreams() []Stream { Retention: RetentionWorkQueue, MaxAge: 7 * 24 * 60 * 60, MaxMsgsPerSubject: 1, + Direct: true, Why: "every other check, taken last and in the order asked (novox/hq ADR 0406 §3); one subject a pull " + "request, the newest request kept; one left behind for a head already judged is dropped when taken", }, @@ -98,6 +100,13 @@ func holderCheckGrants() []string { "$KV."+CheckRunningBucket+".>", "$JS.API.STREAM.INFO.KV_"+CheckRunningBucket, "$JS.API.DIRECT.GET.KV_"+CheckRunningBucket+".>") + // And, to hold checks while a delivery builds (ADR 0483, issues 1687 and 1688): the bucket of the plans building, **read and + // never written** (the controller's alone, so no `$KV.` subject of it is granted), and the oldest waiting ask of each check + // queue, read by a direct get that takes nothing from the queue and so counts against no delivery of the ask. + pub = append(pub, readGrants(PlansBuildingBucket, "")...) + for _, w := range CheckWorkers() { + pub = append(pub, "$JS.API.DIRECT.GET."+w.Stream, "$JS.API.DIRECT.GET."+w.Stream+".>") + } return pub } diff --git a/internal/broker/checks_test.go b/internal/broker/checks_test.go index 23bb9e84..9fd3e095 100644 --- a/internal/broker/checks_test.go +++ b/internal/broker/checks_test.go @@ -40,3 +40,37 @@ func TestTheBuildAgentReadsAndWatchesThePullRequestsKeys(t *testing.T) { } } } + +// Issue 1687 (ADR 0483 clause 2): a holder of the build seat reads the bucket of the plans building and writes none of it, and +// reads the oldest waiting ask of each check queue by a direct get, which takes nothing from the queue. +func TestTheBuildSeatsHoldersReadThePlansBuildingAndWriteNone(t *testing.T) { + seat := Seat{Name: TheBuildSeat, Scope: "node", Accepts: []string{"build"}, Emits: []string{"built", "started"}} + p := Principal{Kind: KindModule, Node: "one", Module: "build-agent", Holds: []Seat{seat}, PasswordHash: "x"} + perms, err := PermissionsFor(p) + if err != nil { + t.Fatal(err) + } + stream := "KV_" + PlansBuildingBucket + for _, s := range []string{ + "$JS.API.STREAM.INFO." + stream, + "$JS.API.DIRECT.GET." + stream + ".>", + "$JS.API.CONSUMER.CREATE." + stream + ".>", + "$JS.API.DIRECT.GET.CHECKS_READY", + "$JS.API.DIRECT.GET.CHECKS_OTHER", + } { + has(t, perms.Publish, s) + } + for _, s := range perms.Publish { + if SubjectsOverlap(s, "$KV."+PlansBuildingBucket+".plan-1") { + t.Errorf("a holder of the build seat may write %s through %q", PlansBuildingBucket, s) + } + } + // And the controller alone writes it, as the writers table says. + for _, row := range WritersTable { + for _, s := range row.Subjects { + if s == "$KV."+PlansBuildingBucket+".>" && row.Writes(p, "$KV."+PlansBuildingBucket+".plan-1") { + t.Errorf("%q lets the build seat's holder write the plans building", row.State) + } + } + } +} diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index 9dd511de..3cb81c58 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -63,6 +63,26 @@ var ( // repository: the controller reads the ask's warrant from the router and accepts only the operator's Approve of // those heads. Written by the controller alone, from the announcement, when it holds the lease. DeliveryMergesBucket = BucketName(ControllerSeat, "delivery-merges") + // PlansBuildingBucket holds one key per delivery plan in its build phase (novox/hq ADR 0483 clause 2, issue 1687): the key + // is `` (`plan-`), its value the plan, the time its build phase began and the time of the last refresh. + // The controller (lease holder) puts it when the plan's first build is asked, puts it again about every thirty seconds + // while the plan builds, and deletes it when the phase ends (every build of the last tier built, or the plan fails, stops + // or waits on the operator); the bucket's age, PlansBuildingKeptFor, is how long a key outlives a controller that + // stopped. **Read by the build seat's holders and the checker, written by the controller alone**: a holder takes no new + // check while any key is there, but for one that waited past ADR 0483's bound. A reader lists the keys and reads none of + // their values to decide that (`link.BuildingPlans`). + PlansBuildingBucket = BucketName(ControllerSeat, PlansBuilding) +) + +// PlansBuilding is the local name of the bucket of the plans building, as a reader names it in `reads` +// (`mesh-controller.plans-building`). +const PlansBuilding = "plans-building" + +// PlansBuildingKeptFor is the age of a plan-building key: the controller puts it again at PlansBuildingRefresh, so a +// controller that stopped leaves checks held at most this long (ADR 0483 clause 2). +const ( + PlansBuildingKeptFor = 2 * time.Minute + PlansBuildingRefresh = 30 * time.Second ) // DeliveryMergesKeptFor is how long a merge mesh-delivery announced is kept: a week, as the events are. @@ -204,14 +224,14 @@ func IsControllerBucket(bucket string) bool { bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket || bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket || bucket == DeliveryMergesBucket || bucket == BuildsBucket || bucket == RolloutsBucket || - bucket == delivery.HaveBucket + bucket == PlansBuildingBucket || bucket == delivery.HaveBucket } // ControllerBuckets are the controller's own buckets, in the order they are asserted. func ControllerBuckets() []string { return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket, AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket, - DeliveryMergesBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket} + DeliveryMergesBucket, PlansBuildingBucket, BuildsBucket, RolloutsBucket, delivery.HaveBucket} } // ControllerBucketsAsserter is what raising the controller's buckets needs of a connection. @@ -354,6 +374,22 @@ func (j *JetStream) EnsureControllerBuckets() error { }); err != nil { return fmt.Errorf("asserting bucket %s: %w", CheckRunningBucket, err) } + // **The plans building: keys that age** (ADR 0483 clause 2, issue 1687). One message a key, each refresh a new one, so the + // age of the bucket is how long a key outlives a controller that stopped refreshing it. + if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: PlansBuildingBucket, + Description: "the delivery plans in their build phase (novox/hq ADR 0483): `` holds the plan and when its " + + "build phase began; put when the plan's first build is asked and about every thirty seconds, deleted when the " + + "phase ends, gone within two minutes of a controller that stopped; read by the build seat's holders, which " + + "take no new check while a key is there; written by the controller alone", + History: 1, + TTL: PlansBuildingKeptFor, + MaxValueSize: 4 << 10, + MaxBytes: 8 << 20, + Storage: jetstream.FileStorage, + }); err != nil { + return fmt.Errorf("asserting bucket %s: %w", PlansBuildingBucket, err) + } // **The merges mesh-delivery announced: keys that age** (ADR 0420 §7, issue 1025). One message a key; a week. if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ Bucket: DeliveryMergesBucket, diff --git a/internal/broker/nats.go b/internal/broker/nats.go index c9007a14..644f0dd4 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -412,6 +412,11 @@ var DegradableGrants = []DegradableGrant{ Reason: "refused: the controller asserts this bucket through the API and writes none of its keys — the build seat's " + "holders do (ADR 0406 §3, issue 974)", }, + { + Subject: "$KV." + PlansBuildingBucket + ".>", + Reason: "refused: the controller then says no plan builds, and no check is held while a delivery builds — a check run " + + "wrongly costs a delivery a place, a check held wrongly costs a pull request its turn (ADR 0483 clause 2, issue 1687)", + }, { Subject: "mesh.seat." + ControllerSeat + ".event.sent", Reason: "refused: the controller says a tier sent to a machine only for outsiders, and a send that could not " + diff --git a/internal/broker/plans_building_live_test.go b/internal/broker/plans_building_live_test.go new file mode 100644 index 00000000..adaad82d --- /dev/null +++ b/internal/broker/plans_building_live_test.go @@ -0,0 +1,134 @@ +//go:build integration + +package broker + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "golang.org/x/crypto/bcrypt" +) + +// **A holder of the build seat reads the plans building and the server refuses its write** (novox/hq ADR 0483 clause 2, issue +// 1687): on a real server whose users are composed as the controller composes them, the controller makes the bucket and puts a +// plan's key; the build agent's credential lists the keys and gets one, and its put of a key is refused with a permissions +// violation. It also reads the first waiting ask of a check queue by a direct get, which takes nothing from the queue. +func TestNatsTheBuildSeatsHoldersReadThePlansBuildingAndTheServerRefusesTheirWrite(t *testing.T) { // 1687 + hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost) + if err != nil { + t.Fatal(err) + } + seat := Seat{Name: TheBuildSeat, Scope: "node", Accepts: []string{"build"}, Emits: []string{"built", "started"}} + users, err := Users(Records{Nodes: []string{"alpha"}, Assigned: map[string][]Declared{ + "alpha": {{Module: RuntimeModule}, {Module: "build-agent", Holds: []Seat{seat}}}, + }}) + if err != nil { + t.Fatal(err) + } + for i := range users { + users[i].PasswordHash = string(hash) + } + accounts, err := ComposeAccounts(users) + if err != nil { + t.Fatal(err) + } + conf := filepath.Join(t.TempDir(), "bus.conf") + if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+ + `"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil { + t.Fatal(err) + } + opts, err := server.ProcessConfigFile(conf) + if err != nil { + t.Fatalf("the composed accounts do not parse: %v", err) + } + opts.NoLog, opts.NoSigs = true, true + s, err := server.NewServer(opts) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(s.Shutdown) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + connect := func(user string) (*nats.Conn, chan error) { + t.Helper() + refused := make(chan error, 8) + nc, err := nats.Connect(s.ClientURL(), nats.UserInfo(user, "pw"), nats.CustomInboxPrefix("_INBOX."+user), + nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused <- err })) + if err != nil { + t.Fatalf("%s cannot connect: %v", user, err) + } + t.Cleanup(nc.Close) + return nc, refused + } + controller, _ := connect(ControllerName) + cj := OnConn(controller) + if err := AssertMeshStreams(cj); err != nil { + t.Fatalf("the controller cannot make the streams: %v", err) + } + if err := cj.EnsureControllerBuckets(); err != nil { + t.Fatalf("the controller cannot make its buckets: %v", err) + } + cjs, _ := jetstream.New(controller) + ckv, err := cjs.KeyValue(ctx, PlansBuildingBucket) + if err != nil { + t.Fatal(err) + } + if _, err := ckv.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil { + t.Fatalf("the controller cannot put a plan's key: %v", err) + } + + holder, refused := connect("alpha.build-agent") + hjs, err := jetstream.New(holder) + if err != nil { + t.Fatal(err) + } + kv, err := hjs.KeyValue(ctx, PlansBuildingBucket) + if err != nil { + t.Fatalf("a holder of the build seat cannot bind %s: %v", PlansBuildingBucket, err) + } + keys, err := kv.Keys(ctx) + if err != nil || len(keys) != 1 || keys[0] != "plan-1" { + t.Fatalf("the holder lists %v (%v), want plan-1", keys, err) + } + if e, err := kv.Get(ctx, "plan-1"); err != nil || e == nil { + t.Fatalf("the holder cannot get the key: %v", err) + } + // Its write is refused by the server. + if err := holder.Publish("$KV."+PlansBuildingBucket+".plan-2", []byte(`{}`)); err != nil { + t.Fatal(err) + } + _ = holder.Flush() + select { + case err := <-refused: + if !errors.Is(err, nats.ErrPermissionViolation) { + t.Errorf("the server said %v, want a permissions violation", err) + } + case <-time.After(3 * time.Second): + t.Error("the server did not refuse the holder's put of a plan's key") + } + if _, err := ckv.Get(ctx, "plan-2"); !errors.Is(err, jetstream.ErrKeyNotFound) { + t.Errorf("the refused put left a key: %v", err) + } + + // And the direct get of a check queue's first waiting ask, which is granted. + reply, err := holder.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+ChecksOtherStream, []byte(`{"seq":1,"next_by_subj":"`+ + CheckSubjectPrefix(TheBuildSeat, false)+`>"}`)) + if err != nil { + t.Fatalf("the holder's direct get of %s was not answered: %v", ChecksOtherStream, err) + } + if reply.Header.Get("Status") != "404" { + t.Errorf("an empty queue answered %q, want 404", reply.Header.Get("Status")) + } +} diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index 1e2b427d..6547383a 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -29,7 +29,7 @@ accounts { ] users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_plans-building.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$KV.mesh-controller_rollouts.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } } diff --git a/internal/broker/writers.go b/internal/broker/writers.go index 06a44fde..af35bf19 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -177,6 +177,10 @@ var WritersTable = []WriterRow{ KeptIn: "key-value " + CheckRunningBucket + ", aged out when not renewed", Others: "the build seat's holders read it to drop a duplicate request", Subjects: kvOf(CheckRunningBucket), Writes: holdsTheBuildSeat, Shared: "every machine holding the build seat puts the key of the check it runs and deletes it when the check ends"}, + {State: "a delivery plan in its build phase", Writer: "controller (lease holder), at the first build's ask and at each refresh", + KeptIn: "key-value " + PlansBuildingBucket + ", aged out when not refreshed", + Others: "the build seat's holders read it and take no new check while a key is there (ADR 0483)", + Subjects: kvOf(PlansBuildingBucket), Writes: isController}, {State: "a merge announced", Writer: "one announcer per forge (the hook, or the poll when the hook is absent — never both)", KeptIn: "the bus", Others: "—", Subjects: []string{"mesh.mod.*.event.pull.merged"}, Writes: ownModule}, // A pull request's head, before it merges (novox/hq to-be 45 §9): the same one announcer. diff --git a/internal/broker/writers_test.go b/internal/broker/writers_test.go index fd56ff3f..78d3de1c 100644 --- a/internal/broker/writers_test.go +++ b/internal/broker/writers_test.go @@ -29,6 +29,8 @@ var designRows = []string{ "builds and their outcomes", "a pull request's check asked", "a pull request's check running", + // The plans in their build phase, a bucket the build seat's holders read (ADR 0483 clause 2, issue 1687). + "a delivery plan in its build phase", "a merge announced", "a pull request's head announced", "a pull request's merge check", diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index b901e9fe..e7b0908f 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -162,11 +162,19 @@ type natsMachine struct { // build waits, then the checks of pull requests marked ready, then every other check (novox/hq ADR 0406 §3). subs []*takeQueue opts MachineOptions + // gate is what the loop asks of the bus to hold checks while a delivery builds; nil is the bus's (a test sets a fake). + gate checkGate + // said is the faults of the hold already said, letKey whether this holder created the let-through key for the check it + // took last (ADR 0483 clause 3). + said map[string]bool + letKey bool } // takeQueue is one queue a machine pulls from. type takeQueue struct { sub *nats.Subscription + // pull is what the queue is fetched through: the subscription once bound (a test sets a fake). + pull puller // kind is what its asks are: a build, a ready check, or another check. kind string // worker is the consumer bound; bound is false until a bind succeeded, for a queue the mesh may not have made yet. @@ -174,6 +182,11 @@ type takeQueue struct { bound bool } +// puller is the fetch of a queue. +type puller interface { + Fetch(batch int, opts ...nats.PullOpt) ([]*nats.Msg, error) +} + // The kinds of queue a machine takes from, in the order it takes from them. const ( queueBuilds = "builds" @@ -195,6 +208,9 @@ type MachineOptions struct { // the request in hand first and reads its drain after, so an ask of `busy` racing the fetch sees one or // the other, never neither. Nil claims every one. Claim func() bool + // LetThroughAfter is how long a check waits while a delivery builds before it is let through; zero is ADR 0483's thirty + // minutes. Set by a test. + LetThroughAfter time.Duration } // MachineOverNATS takes build work from the current build role. @@ -244,6 +260,9 @@ func (m *natsMachine) Close() { // **In this order** (novox/hq ADR 0406 §3): a delivery build, then a check of a pull request marked ready, then any // other check, each queue tried without waiting; when all three are empty the first is waited on for a second at most // and the three are tried again, so a delivery build is never much more than a second behind and nothing spins. +// +// **Unless a delivery builds** (novox/hq ADR 0483): while the controller's bucket of the plans building holds a key, no check queue +// is fetched from, but for the one check that waited LetThroughAfter, which goes first (see next). func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) error { worker, found := broker.HolderConsumerFor(m.on, "build-agent", broker.DeclaredSeat{Name: m.seat, Accepts: []string{"build"}}) @@ -320,70 +339,79 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build)) } return fmt.Errorf("the bus stopped delivering build work: %w", err) } - // **Paused while the pull was answered** (ADR 0219): "takes no new build" holds even for - // the one that arrived in that moment. Handed back at once for another holder — counted as - // a delivery, which a pause landing exactly then costs and nothing else does. - if m.opts.Paused != nil && m.opts.Paused() { - _ = msg.Nak() - continue - } - var request BuildRequest - if err := json.Unmarshal(msg.Data, &request); err != nil { - // Unreadable: terminated rather than retried, because the next attempt reads the same - // bytes. Nobody waiting hears an answer, which is right — there was no request. - _ = msg.Term() - continue - } - // **Cancelled in the moment it was fetched** (novox/hq ADR 0219): the controller wrote the - // id into the seat's cancelled set before deleting the ask, so one taken in between is - // ended here, terminated rather than redelivered, and its outcome said as failed — never - // built. A set that cannot be read is said and the ask built: a cancel is a person's - // exception, and a holder that refused every build while its grant was missing would - // stop the mesh building for it. - build := &natsBuild{request: request, msg: msg, on: m.on, js: m.js, seat: m.seat, delivery: deliveryOf(msg)} - if cancelled, err := IsCancelled(m.js.Conn(), m.seat, request.ID); err != nil { - fmt.Fprintf(os.Stderr, "could not read whether %s was cancelled, so it is built: %v\n", request.ID, err) - } else if cancelled { - endCancelled(ctx, build) - continue - } - // **A check left behind is dropped when it is taken** (novox/hq ADR 0406 §3): a request on the third queue for a - // head and a trunk commit its pull request's key already says were judged is acknowledged and not run. - if q.kind == queueOther { - if why := m.judgedAlready(ctx, request); why != "" { - fmt.Fprintf(os.Stderr, "%s was left behind in the check queue and is dropped: %s\n", request.ID, why) - _ = msg.Ack() - continue - } - } - // **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause - // landing then is. - if m.opts.Claim != nil && !m.opts.Claim() { - _ = 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. - working := make(chan struct{}) - go stillWorking(msg, working, heartbeatWhileWorking) - // And a check says it is running, to the other builders (ADR 0406 §3): its key put now and at each heartbeat, - // deleted when it ends, whatever the outcome. - stopRunning := keepRunning(ctx, m.js.Conn(), request, m.on, heartbeatWhileWorking) - do(ctx, build) - stopRunning() - close(working) + m.handOver(ctx, msg, q, do) } } +// handOver settles one fetched ask: dropped, handed back or built. **A check let through keeps its key for as long as it +// stays here** (ADR 0483 clause 3): refreshed while it runs and deleted when it ends, however it ends. +func (m *natsMachine) handOver(ctx context.Context, msg *nats.Msg, q *takeQueue, do func(context.Context, Build)) { + if m.letKey { + defer m.keepLetThrough(ctx, heartbeatWhileWorking)() + } + // **Paused while the pull was answered** (ADR 0219): "takes no new build" holds even for + // the one that arrived in that moment. Handed back at once for another holder — counted as + // a delivery, which a pause landing exactly then costs and nothing else does. + if m.opts.Paused != nil && m.opts.Paused() { + _ = msg.Nak() + return + } + var request BuildRequest + if err := json.Unmarshal(msg.Data, &request); err != nil { + // Unreadable: terminated rather than retried, because the next attempt reads the same + // bytes. Nobody waiting hears an answer, which is right — there was no request. + _ = msg.Term() + return + } + // **Cancelled in the moment it was fetched** (novox/hq ADR 0219): the controller wrote the + // id into the seat's cancelled set before deleting the ask, so one taken in between is + // ended here, terminated rather than redelivered, and its outcome said as failed — never + // built. A set that cannot be read is said and the ask built: a cancel is a person's + // exception, and a holder that refused every build while its grant was missing would + // stop the mesh building for it. + build := &natsBuild{request: request, msg: msg, on: m.on, js: m.js, seat: m.seat, delivery: deliveryOf(msg)} + if cancelled, err := IsCancelled(m.js.Conn(), m.seat, request.ID); err != nil { + fmt.Fprintf(os.Stderr, "could not read whether %s was cancelled, so it is built: %v\n", request.ID, err) + } else if cancelled { + endCancelled(ctx, build) + return + } + // **A check left behind is dropped when it is taken** (novox/hq ADR 0406 §3): a request on the third queue for a + // head and a trunk commit its pull request's key already says were judged is acknowledged and not run. + if q.kind == queueOther { + if why := m.judgedAlready(ctx, request); why != "" { + fmt.Fprintf(os.Stderr, "%s was left behind in the check queue and is dropped: %s\n", request.ID, why) + _ = msg.Ack() + return + } + } + // **Drained in the moment it was fetched** (novox/hq ADR 0335): handed back at once, as a pause + // landing then is. + if m.opts.Claim != nil && !m.opts.Claim() { + _ = msg.Nak() + return + } + // **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() + return + } + // 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. + working := make(chan struct{}) + go stillWorking(msg, working, heartbeatWhileWorking) + // And a check says it is running, to the other builders (ADR 0406 §3): its key put now and at each heartbeat, + // deleted when it ends, whatever the outcome. + stopRunning := keepRunning(ctx, m.js.Conn(), request, m.on, heartbeatWhileWorking) + do(ctx, build) + stopRunning() + close(working) +} + // 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 { @@ -425,20 +453,32 @@ func (m *natsMachine) bind(q *takeQueue) error { if err != nil { return err } - q.sub, q.bound = sub, true + q.sub, q.pull, q.bound = sub, sub, true return nil } // next takes the next request in the order of the queues: each tried for a moment, and when all are empty the first waited // on for a second. Nothing, with no error, when there was nothing. func (m *natsMachine) next(ctx context.Context) (*nats.Msg, *takeQueue, error) { + // **While a delivery builds, only delivery builds are taken, but for one check that waited** (novox/hq ADR 0483): the + // bucket of the plans building is read before either check queue is fetched from, and a key there holds both. The check + // that waited LetThroughAfter goes first, ahead of the builds that wait; the rest of the holders take builds only. + if m.seat == broker.TheBuildSeat && len(m.subs) > 1 { + if plans := m.building(ctx); len(plans) > 0 { + if msg, q := m.letOneThrough(ctx); msg != nil { + fmt.Fprintf(os.Stderr, "a check that waited %s is let through while %s builds\n", m.letThroughAfter(), whyHeld(plans)) + return msg, q, nil + } + return m.nextBuild(ctx) + } + } var last error for _, q := range m.subs { if !q.bound { continue } trying, end := context.WithTimeout(ctx, checkTakeWait) - fetched, err := q.sub.Fetch(1, nats.Context(trying)) + fetched, err := q.pull.Fetch(1, nats.Context(trying)) end() if len(fetched) > 0 { return fetched[0], q, nil @@ -454,9 +494,14 @@ func (m *natsMachine) next(ctx context.Context) (*nats.Msg, *takeQueue, error) { if last != nil { return nil, nil, last } + return m.nextBuild(ctx) +} + +// nextBuild waits for a second at most for a delivery build, the seat's own queue. +func (m *natsMachine) nextBuild(ctx context.Context) (*nats.Msg, *takeQueue, error) { waiting, end := context.WithTimeout(ctx, takeIdleWait) defer end() - fetched, err := m.subs[0].sub.Fetch(1, nats.Context(waiting)) + fetched, err := m.subs[0].pull.Fetch(1, nats.Context(waiting)) if len(fetched) > 0 { return fetched[0], m.subs[0], nil } diff --git a/internal/link/checks_wait_integration_test.go b/internal/link/checks_wait_integration_test.go new file mode 100644 index 00000000..8839fc54 --- /dev/null +++ b/internal/link/checks_wait_integration_test.go @@ -0,0 +1,304 @@ +//go:build integration + +package link + +import ( + "context" + "encoding/json" + "fmt" + "reflect" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1652, 1687, 1688), against a real bus. + +// aBusWithTheBuckets is the bus with the build role and the controller's buckets, the bucket of the checks running with an age +// of two seconds so a key a holder stopped refreshing is seen to expire. +func aBusWithTheBuckets(t *testing.T) (*broker.JetStream, jetstream.KeyValue) { + t.Helper() + js := aBusWithTheBuildRole(t) + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + ctx := t.Context() + if _, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.CheckRunningBucket, History: 1, TTL: 2 * time.Second}); err != nil { + t.Fatal(err) + } + plans, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.PlansBuildingBucket, History: 1, TTL: broker.PlansBuildingKeptFor}) + if err != nil { + t.Fatal(err) + } + return js, plans +} + +func aCheckAskedAgo(ago time.Duration, number int, ready bool) BuildRequest { + r := aPullCheck(fmt.Sprintf("build-%d", time.Now().Add(-ago).UnixNano()), number, fmt.Sprintf("head-%d-0000000", number), "trunk-0-000000", ready) + return r +} + +// 1652/AC6: a plan's builds asked in two tiers with a check waiting, the check is handed out only after the last tier's builds, +// and not in the gap between the tiers. +func TestNatsACheckIsHandedOutOnlyAfterTheLastTierOfAPlan(t *testing.T) { + js, plans := aBusWithTheBuckets(t) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + asks := BuildsOn(js, TheBuildMachine) + + if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil { + t.Fatal(err) + } + if err := asks.Ask(ctx, aCheckAskedAgo(time.Minute, 7, true)); err != nil { + t.Fatal(err) + } + if err := asks.Ask(ctx, BuildRequest{ID: "tier-1-a", Repository: "/a"}); err != nil { + t.Fatal(err) + } + took := make(chan string, 16) + machine := MachineOverNATS(js, "anchor") + defer machine.Close() + go func() { + _ = machine.Take(ctx, func(ctx context.Context, work Build) { + took <- work.Request().ID + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "anchor", Failed: "no"}) + _ = work.Done() + }) + }() + expect := func(want string) { + t.Helper() + select { + case got := <-took: + if got != want { + t.Fatalf("taken %s, want %s", got, want) + } + case <-time.After(15 * time.Second): + t.Fatalf("%s was not taken", want) + } + } + expect("tier-1-a") + // The gap between two tiers: the delivery's queue is empty, the plan's key is there, and the check is not taken. + select { + case got := <-took: + t.Fatalf("%s was taken in the gap between two tiers", got) + case <-time.After(3 * time.Second): + } + if err := asks.Ask(ctx, BuildRequest{ID: "tier-2-a", Repository: "/b"}); err != nil { + t.Fatal(err) + } + expect("tier-2-a") + select { + case got := <-took: + t.Fatalf("%s was taken before the plan's last build was built and its key deleted", got) + case <-time.After(2 * time.Second): + } + // The phase ends: the key is deleted, and the check is handed out. + if err := plans.Delete(ctx, "plan-1"); err != nil { + t.Fatal(err) + } + select { + case got := <-took: + if got == "tier-1-a" || got == "tier-2-a" { + t.Fatalf("a build was taken twice: %s", got) + } + case <-time.After(15 * time.Second): + t.Fatal("the check was not handed out once the plan's key was deleted") + } +} + +// 1688: two holders racing for checks that waited past the bound: exactly one is let through at a time, by a create that fails +// while the key is there, and the key is deleted when its check ends, after which the next oldest goes. +func TestNatsTwoHoldersRacingForOldChecksLetExactlyOneThrough(t *testing.T) { + js, plans := aBusWithTheBuckets(t) + ctx, stop := context.WithCancel(t.Context()) + defer stop() + // The buckets' age is two seconds here, so the keys are refreshed well within it. + was := heartbeatWhileWorking + heartbeatWhileWorking = 400 * time.Millisecond + t.Cleanup(func() { heartbeatWhileWorking = was }) + asks := BuildsOn(js, TheBuildMachine) + if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil { + t.Fatal(err) + } + for i, ago := range []time.Duration{50 * time.Minute, 40 * time.Minute, 35 * time.Minute} { + if err := asks.Ask(ctx, aCheckAskedAgo(ago, i+1, false)); err != nil { + t.Fatal(err) + } + } + var mu sync.Mutex + running, most := 0, 0 + started := make(chan string, 8) + release := make(chan struct{}) + do := func(ctx context.Context, work Build) { + mu.Lock() + running++ + most = max(most, running) + mu.Unlock() + started <- work.Request().Check.Repo + fmt.Sprint(work.Request().Check.Number) + <-release + mu.Lock() + running-- + mu.Unlock() + _ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "x", Failed: "no"}) + _ = work.Done() + } + for _, on := range []string{"ace", "shanks"} { + machine := MachineOverNATS(js, on) + defer machine.Close() + go func() { _ = machine.Take(ctx, do) }() + } + var order []string + for range 3 { + select { + case id := <-started: + order = append(order, id) + case <-time.After(20 * time.Second): + t.Fatalf("started so far: %v", order) + } + // While it runs, no second check starts and the key is there. + select { + case id := <-started: + t.Fatalf("%s started beside a check let through", id) + case <-time.After(2500 * time.Millisecond): + } + kv := runningKV(t, js) + if _, err := kv.Get(ctx, LetThroughKey); err != nil { + t.Fatalf("the let-through key is not there while the check runs: %v", err) + } + release <- struct{}{} + // And it is deleted when the check ends. + deadline := time.Now().Add(10 * time.Second) + for { + if _, err := kv.Get(ctx, LetThroughKey); err != nil { + break + } + if time.Now().After(deadline) { + t.Fatal("the let-through key was not deleted when the check ended") + } + time.Sleep(50 * time.Millisecond) + } + } + if want := []string{"a1", "a2", "a3"}; !reflect.DeepEqual(order, want) { + t.Fatalf("checks went %v, want the oldest first %v", order, want) + } + mu.Lock() + defer mu.Unlock() + if most != 1 { + t.Fatalf("%d checks ran at once while a delivery built", most) + } +} + +func runningKV(t *testing.T, js *broker.JetStream) jetstream.KeyValue { + t.Helper() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + kv, err := api.KeyValue(t.Context(), broker.CheckRunningBucket) + if err != nil { + t.Fatal(err) + } + return kv +} + +// 1688: the create of the let-through key succeeds for exactly one of holders asking at once, and a key its holder stopped +// refreshing expires, after which the next create succeeds. +func TestNatsOneCreateOfTheLetThroughKeySucceedsAndAStoppedHoldersKeyExpires(t *testing.T) { + js, _ := aBusWithTheBuckets(t) + gate := busGate{conn: js.Conn()} + ctx := t.Context() + var wg sync.WaitGroup + var mu sync.Mutex + won := 0 + for i := range 8 { + wg.Add(1) + go func() { + defer wg.Done() + ok, err := gate.TakeLetThrough(ctx, fmt.Sprint("holder-", i)) + if err != nil { + t.Error(err) + return + } + if ok { + mu.Lock() + won++ + mu.Unlock() + } + }() + } + wg.Wait() + if won != 1 { + t.Fatalf("%d of 8 holders created the let-through key, want exactly one", won) + } + // The holder stopped: nothing refreshes the key, and the bucket's age (two seconds here) removes it. + deadline := time.Now().Add(15 * time.Second) + for { + ok, err := gate.TakeLetThrough(ctx, "next") + if err != nil { + t.Fatal(err) + } + if ok { + break + } + if time.Now().After(deadline) { + t.Fatal("the key of a stopped holder did not expire") + } + time.Sleep(250 * time.Millisecond) + } + // And a refreshed key does not expire. + for range 6 { + if err := gate.KeepLetThrough(ctx, "next"); err != nil { + t.Fatal(err) + } + time.Sleep(700 * time.Millisecond) + } + if ok, _ := gate.TakeLetThrough(ctx, "other"); ok { + t.Fatal("a refreshed key expired") + } + var v LetThrough + e, err := runningKV(t, js).Get(ctx, LetThroughKey) + if err != nil || json.Unmarshal(e.Value(), &v) != nil || v.Builder != "next" { + t.Fatalf("the key says %+v (%v)", v, err) + } +} + +// 1687: the oldest waiting ask is read without being taken: no delivery is counted against it. +func TestNatsTheOldestWaitingAskIsReadWithoutTakingIt(t *testing.T) { + js, _ := aBusWithTheBuckets(t) + ctx := t.Context() + asks := BuildsOn(js, TheBuildMachine) + old := aCheckAskedAgo(45*time.Minute, 1, false) + if err := asks.Ask(ctx, old); err != nil { + t.Fatal(err) + } + gate := busGate{conn: js.Conn()} + q := &takeQueue{kind: queueOther, worker: broker.CheckWorkers()[1]} + at, found, err := gate.OldestWaiting(ctx, q) + if err != nil || !found { + t.Fatalf("the first waiting ask was not read: %v %v", found, err) + } + if want, _ := BuildAskedAt(old.ID); !at.Equal(want) { + t.Fatalf("read %v, asked %v", at, want) + } + for range 3 { + if _, _, err := gate.OldestWaiting(ctx, q); err != nil { + t.Fatal(err) + } + } + info, err := js.Context().ConsumerInfo(q.worker.Stream, q.worker.Name) + if err != nil { + t.Fatal(err) + } + if info.Delivered.Consumer != 0 || info.NumAckPending != 0 || info.NumPending != 1 { + t.Fatalf("reading the age took the ask: delivered %d, pending acks %d, waiting %d", info.Delivered.Consumer, info.NumAckPending, info.NumPending) + } + // And an empty queue reads as none. + if _, found, err := gate.OldestWaiting(ctx, &takeQueue{kind: queueReady, worker: broker.CheckWorkers()[0]}); err != nil || found { + t.Fatalf("an empty queue read as %v (%v)", found, err) + } +} diff --git a/internal/link/holds.go b/internal/link/holds.go new file mode 100644 index 00000000..dabf05d3 --- /dev/null +++ b/internal/link/holds.go @@ -0,0 +1,331 @@ +package link + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1652, 1687 and 1688). +// +// **The layout other parts read** (ADR 0484 clause 12 has the checker read the same key; a layout shared between modules +// belongs in mesh-sdk, and this one is not there yet): +// +// - bucket `mesh-controller_plans-building` (broker.PlansBuildingBucket): one key per delivery plan in its build phase, named +// by the plan's id (`plan-`), value a PlanBuilding as JSON. Written by the controller (lease holder) alone: put when +// the plan's first build is asked, put again about every thirty seconds, deleted when the phase ends; the bucket's age +// (two minutes) removes the key of a controller that stopped. A reader lists the keys and needs no value to decide. +// - key `let-through` in bucket `mesh-controller_checks-running` (broker.CheckRunningBucket), value a LetThrough as JSON: +// created by the holder that takes a check which waited LetThroughAfter, refreshed at each heartbeat of that check, deleted +// when it ends. At most one such check runs at a time on the build seat, because the create fails while the key is there. + +// LetThroughAfter is how long a check waits, while a delivery builds, before it is let through (ADR 0483 clause 3). +const LetThroughAfter = 30 * time.Minute + +// LetThroughKey is the key of the one check let through, in the bucket of the checks running. It has no dot, so it is no +// pull request's key (`..`) and no group's (`group.`). +const LetThroughKey = "let-through" + +// PlanBuilding is the value of a plan's key in the bucket of the plans building. +type PlanBuilding struct { + Plan string `json:"plan"` + Since time.Time `json:"since"` + Seen time.Time `json:"seen"` +} + +// LetThrough is the value of the let-through key: the builder that runs the check and when it took it. +type LetThrough struct { + Builder string `json:"builder"` + Since time.Time `json:"since"` +} + +// checkGate is what a taking loop asks of the bus to hold checks while a delivery builds; a variable of the machine so a test +// hands out against fakes. +type checkGate interface { + // Building is the ids of the plans in their build phase: the keys of the bucket. An error is a bucket that cannot be read. + Building(ctx context.Context) ([]string, error) + // OldestWaiting is when the first waiting ask of a check queue was asked, read without taking it. + OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error) + // TakeLetThrough creates the let-through key for a builder: false when it is there (another builder runs a check let through). + TakeLetThrough(ctx context.Context, builder string) (bool, error) + // KeepLetThrough puts the key again, and DropLetThrough deletes it. + KeepLetThrough(ctx context.Context, builder string) error + DropLetThrough(ctx context.Context) error +} + +// busGate is the gate over a connection. +type busGate struct{ conn *nats.Conn } + +func (g busGate) Building(ctx context.Context) ([]string, error) { + return BuildingPlans(ctx, g.conn) +} + +// BuildingPlans is the ids of the delivery plans in their build phase, from the controller's bucket. A reader that is not the +// build seat's (the checker, ADR 0484 clause 12) calls this. A bucket that cannot be read is an error; none is none building. +func BuildingPlans(ctx context.Context, conn *nats.Conn) ([]string, error) { + api, err := jetstream.New(conn) + if err != nil { + return nil, err + } + kv, err := api.KeyValue(ctx, broker.PlansBuildingBucket) + if err != nil { + return nil, fmt.Errorf("the plans building: %w", err) + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("the plans building: %w", err) + } + 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. +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]}) + 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) + } + if reply.Header.Get("Status") == "404" || reply.Header.Get("Status") == "408" { + return time.Time{}, false, nil + } + return askedAtOf(reply.Data, reply.Header.Get("Nats-Time-Stamp")), true, nil +} + +// askedAtOf is when a request was asked: its id says (NewBuildID), the stream's own time otherwise, and a request that says +// neither is taken as asked now, so it is never let through on a guess. +func askedAtOf(data []byte, stored string) time.Time { + var r BuildRequest + if json.Unmarshal(data, &r) == nil { + if at, ok := BuildAskedAt(r.ID); ok { + return at + } + } + if at, err := time.Parse(time.RFC3339Nano, stored); err == nil { + return at + } + return time.Now() +} + +func jsOf(conn *nats.Conn) nats.JetStreamContext { + js, _ := conn.JetStream() + return js +} + +func (g busGate) TakeLetThrough(ctx context.Context, builder string) (bool, error) { + kv, err := runningBucket(ctx, g.conn) + if err != nil { + return false, err + } + body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()}) + if _, err := kv.Create(ctx, LetThroughKey, body); err != nil { + if errors.Is(err, jetstream.ErrKeyExists) { + return false, nil + } + return false, err + } + return true, nil +} + +func (g busGate) KeepLetThrough(ctx context.Context, builder string) error { + kv, err := runningBucket(ctx, g.conn) + if err != nil { + return err + } + body, _ := json.Marshal(LetThrough{Builder: builder, Since: time.Now().UTC()}) + _, err = kv.Put(ctx, LetThroughKey, body) + return err +} + +func (g busGate) DropLetThrough(ctx context.Context) error { + kv, err := runningBucket(ctx, g.conn) + if err != nil { + return err + } + if err := kv.Delete(ctx, LetThroughKey); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + return nil +} + +// CheckHold is what a reader of the queue (the controller's own verb) says of why checks wait: the plans building and the check +// let through now, if one. +type CheckHold struct { + Plans []string + LetThrough *LetThrough +} + +// ReadCheckHold reads the plans building and the let-through key. A bucket that cannot be read is an error the caller says. +func ReadCheckHold(ctx context.Context, conn *nats.Conn) (CheckHold, error) { + var h CheckHold + plans, err := BuildingPlans(ctx, conn) + if err != nil { + return h, err + } + h.Plans = plans + kv, err := runningBucket(ctx, conn) + if err != nil { + return h, err + } + e, err := kv.Get(ctx, LetThroughKey) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return h, nil + } + if err != nil { + return h, err + } + var l LetThrough + if json.Unmarshal(e.Value(), &l) == nil { + h.LetThrough = &l + } + return h, nil +} + +// gateOf is the machine's gate: the one a test set, or the bus's. +func (m *natsMachine) gateOf() checkGate { + if m.gate != nil { + return m.gate + } + return busGate{conn: m.js.Conn()} +} + +// sayOnce says a fault of the hold the first time it happens and again once it has recovered and happened anew, so a bucket +// that cannot be read is in the log once and not every second. +func (m *natsMachine) sayOnce(kind string, err error) { + if m.said == nil { + m.said = map[string]bool{} + } + if err == nil { + delete(m.said, kind) + return + } + if !m.said[kind] { + m.said[kind] = true + fmt.Fprintf(os.Stderr, "%s; read as empty, so no check is held for it: %v\n", kind, err) + } +} + +// building says whether a delivery builds: any key in the bucket of the plans building. **A bucket that cannot be read is said +// and read as empty** (ADR 0483 clause 2): a check wrongly held costs a pull request its turn, a fault that held every check +// would stop the mesh checking. +func (m *natsMachine) building(ctx context.Context) []string { + plans, err := m.gateOf().Building(ctx) + m.sayOnce("the bucket of the plans building cannot be read", err) + if err != nil { + // empty-on-error: ADR 0483 clause 2 — a bucket that cannot be read is said (sayOnce) and read as empty, because a check wrongly held costs a pull request its turn and a fault that held every check would stop the mesh checking. + return nil + } + return plans +} + +// letOneThrough takes the oldest waiting check of the two check queues once it waited LetThroughAfter, by the one let-through +// key: nil when none waited that long, when another holder runs the check let through, or when the check was taken by +// another holder in the moment. The caller owns the key while it returns a message and releases it with dropLetThrough. +func (m *natsMachine) letOneThrough(ctx context.Context) (*nats.Msg, *takeQueue) { + gate := m.gateOf() + var oldest *takeQueue + var oldestAt time.Time + for _, q := range m.subs[1:] { + if !q.bound { + continue + } + at, found, err := gate.OldestWaiting(ctx, q) + m.sayOnce("the first waiting check of the "+q.kind+" queue cannot be read", err) + if err != nil || !found { + continue + } + if oldest == nil || at.Before(oldestAt) { + oldest, oldestAt = q, at + } + } + if oldest == nil || time.Since(oldestAt) < m.letThroughAfter() { + return nil, nil + } + created, err := gate.TakeLetThrough(ctx, m.on) + m.sayOnce("the let-through key cannot be taken", err) + if err != nil || !created { + return nil, nil + } + trying, end := context.WithTimeout(ctx, checkTakeWait) + defer end() + fetched, err := oldest.pull.Fetch(1, nats.Context(trying)) + if len(fetched) == 0 { + if err != nil && !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.Canceled) { + m.sayOnce("the "+oldest.kind+" check queue cannot be taken from", err) + } + m.dropLetThrough(ctx) + return nil, nil + } + // 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() + m.dropLetThrough(ctx) + return nil, nil + } + m.letKey = true + return fetched[0], oldest +} + +func (m *natsMachine) letThroughAfter() time.Duration { + if m.opts.LetThroughAfter > 0 { + return m.opts.LetThroughAfter + } + return LetThroughAfter +} + +// dropLetThrough deletes the let-through key this holder created. +func (m *natsMachine) dropLetThrough(ctx context.Context) { + m.letKey = false + if err := m.gateOf().DropLetThrough(context.WithoutCancel(ctx)); err != nil { + fmt.Fprintf(os.Stderr, "could not take the let-through key away; it ages out: %v\n", err) + } +} + +// keepLetThrough refreshes the key at every heartbeat while the check it was created for runs, and returns the end of that: it +// stops the refreshing and deletes the key. +func (m *natsMachine) keepLetThrough(ctx context.Context, every time.Duration) (end func()) { + done := make(chan struct{}) + finished := make(chan struct{}) + go func() { + defer close(finished) + tick := time.NewTicker(every) + defer tick.Stop() + for { + select { + case <-done: + return + case <-tick.C: + putting, stop := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + if err := m.gateOf().KeepLetThrough(putting, m.on); err != nil { + fmt.Fprintf(os.Stderr, "could not keep the let-through key: %v\n", err) + } + stop() + } + } + }() + return func() { + close(done) + <-finished + m.dropLetThrough(ctx) + } +} + +// whyHeld words the plans a check is held for, for the log. +func whyHeld(plans []string) string { return strings.Join(plans, ", ") } diff --git a/internal/link/holds_test.go b/internal/link/holds_test.go new file mode 100644 index 00000000..a7fc624a --- /dev/null +++ b/internal/link/holds_test.go @@ -0,0 +1,307 @@ +package link + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/broker" +) + +// Checks wait while a delivery builds (novox/hq ADR 0483, issues 1687 and 1688): the order the taking loop hands out in, against +// fake queues and a fake bus. + +// fakeQueue is a queue the loop fetches from: the asks waiting, handed out in order, and how often it was fetched from. +type fakeQueue struct { + mu sync.Mutex + waiting []*nats.Msg + fetched int +} + +func (f *fakeQueue) Fetch(int, ...nats.PullOpt) ([]*nats.Msg, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.fetched++ + if len(f.waiting) == 0 { + return nil, nats.ErrTimeout + } + m := f.waiting[0] + f.waiting = f.waiting[1:] + return []*nats.Msg{m}, nil +} + +func (f *fakeQueue) fetches() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.fetched +} + +// askedAgo is a request asked this long ago, as the id says, for a check of this pull request. +func askedAgo(ago time.Duration, number int, ready bool) *nats.Msg { + id := fmt.Sprintf("build-%d", time.Now().Add(-ago).UnixNano()) + body, _ := json.Marshal(BuildRequest{ID: id, Repository: "r", Check: &CheckRequest{Owner: "o", Repo: "r", Number: number, Ready: ready}}) + return &nats.Msg{Data: body} +} + +func buildAsk() *nats.Msg { + body, _ := json.Marshal(BuildRequest{ID: fmt.Sprintf("build-%d", time.Now().UnixNano()), Repository: "r"}) + return &nats.Msg{Data: body} +} + +// fakeGate is the bus as the loop sees it for the hold. +type fakeGate struct { + mu sync.Mutex + plans []string + plansErr error + letKey bool + taken int + kept int + dropped int + oldestFrom map[*takeQueue]*fakeQueue +} + +func (g *fakeGate) Building(context.Context) ([]string, error) { + g.mu.Lock() + defer g.mu.Unlock() + return g.plans, g.plansErr +} + +// OldestWaiting reads the first waiting ask of the fake queue without taking it. +func (g *fakeGate) OldestWaiting(_ context.Context, q *takeQueue) (time.Time, bool, error) { + f := g.oldestFrom[q] + f.mu.Lock() + defer f.mu.Unlock() + if len(f.waiting) == 0 { + return time.Time{}, false, nil + } + return askedAtOf(f.waiting[0].Data, ""), true, nil +} + +func (g *fakeGate) TakeLetThrough(context.Context, string) (bool, error) { + g.mu.Lock() + defer g.mu.Unlock() + if g.letKey { + return false, nil + } + g.letKey = true + g.taken++ + return true, nil +} + +func (g *fakeGate) KeepLetThrough(context.Context, string) error { + g.mu.Lock() + defer g.mu.Unlock() + g.kept++ + return nil +} + +func (g *fakeGate) DropLetThrough(context.Context) error { + g.mu.Lock() + defer g.mu.Unlock() + g.letKey = false + g.dropped++ + return nil +} + +func (g *fakeGate) setPlans(plans ...string) { + g.mu.Lock() + defer g.mu.Unlock() + g.plans = plans +} + +// aMachine is a loop on the build seat with a delivery queue and the two check queues, all fake. +func aMachine() (m *natsMachine, builds, ready, other *fakeQueue, gate *fakeGate) { + builds, ready, other = &fakeQueue{}, &fakeQueue{}, &fakeQueue{} + subs := []*takeQueue{ + {kind: queueBuilds, pull: builds, bound: true}, + {kind: queueReady, pull: ready, bound: true}, + {kind: queueOther, pull: other, bound: true}, + } + gate = &fakeGate{oldestFrom: map[*takeQueue]*fakeQueue{subs[1]: ready, subs[2]: other}} + m = &natsMachine{on: "ace", seat: broker.TheBuildSeat, subs: subs, gate: gate} + return m, builds, ready, other, gate +} + +func nextAsk(t *testing.T, m *natsMachine) (*nats.Msg, *takeQueue) { + t.Helper() + msg, q, err := m.next(context.Background()) + if err != nil { + t.Fatal(err) + } + return msg, q +} + +// 1652/AC6, 1687: a waiting check is not taken while a plan key is there, also when the delivery's queue is empty for the moment +// (between two tiers). +func TestNoCheckIsTakenWhileAPlanBuilds(t *testing.T) { + m, builds, ready, other, gate := aMachine() + gate.setPlans("plan-1") + ready.waiting = []*nats.Msg{askedAgo(2*time.Minute, 1, true)} + other.waiting = []*nats.Msg{askedAgo(5*time.Minute, 2, false)} + // The delivery's queue is empty: between two tiers. + if msg, _ := nextAsk(t, m); msg != nil { + t.Fatalf("a check was taken while plan-1 builds: %s", msg.Data) + } + if ready.fetches() != 0 || other.fetches() != 0 { + t.Fatalf("the check queues were fetched from while plan-1 builds: ready %d, other %d", ready.fetches(), other.fetches()) + } + if builds.fetches() == 0 { + t.Fatal("the delivery's queue was not looked at") + } +} + +// A delivery build is still taken first, and the check waits behind it. +func TestADeliveryBuildIsTakenWhileAPlanBuilds(t *testing.T) { + m, builds, ready, _, gate := aMachine() + gate.setPlans("plan-1") + builds.waiting = []*nats.Msg{buildAsk()} + ready.waiting = []*nats.Msg{askedAgo(2*time.Minute, 1, true)} + msg, q := nextAsk(t, m) + if msg == nil || q.kind != queueBuilds { + t.Fatalf("the delivery build was not taken first: %v %v", msg, q) + } + if ready.fetches() != 0 { + t.Fatal("a check queue was fetched from") + } +} + +// It is taken once the key is gone (deleted at the phase's end, or expired). +func TestACheckIsTakenOnceTheKeyIsGone(t *testing.T) { + m, _, ready, _, gate := aMachine() + gate.setPlans("plan-1") + ready.waiting = []*nats.Msg{askedAgo(2*time.Minute, 1, true)} + if msg, _ := nextAsk(t, m); msg != nil { + t.Fatal("taken while the key is there") + } + gate.setPlans() + msg, q := nextAsk(t, m) + if msg == nil || q.kind != queueReady { + t.Fatalf("the check was not taken once the key was gone: %v", q) + } + if m.letKey { + t.Fatal("a check taken without a delivery building took the let-through key") + } +} + +// 1687: a bucket that cannot be read is said and read as empty, so the checks are taken. +func TestAnUnreadableBucketHoldsNoCheck(t *testing.T) { + m, _, ready, _, gate := aMachine() + gate.plansErr = errors.New("permissions violation") + gate.plans = []string{"plan-1"} + ready.waiting = []*nats.Msg{askedAgo(2*time.Minute, 1, true)} + msg, q := nextAsk(t, m) + if msg == nil || q.kind != queueReady { + t.Fatal("a check was held because the bucket could not be read") + } + if !m.said["the bucket of the plans building cannot be read"] { + t.Fatalf("the fault was not said: %v", m.said) + } +} + +// 1688: a check asked 30 minutes ago is let through while a key is there; the oldest across both queues by ask time, ahead of a +// waiting delivery build; a younger one is never taken before it. +func TestTheOldestCheckPastTheBoundGoesFirstAcrossBothQueues(t *testing.T) { + m, builds, ready, other, gate := aMachine() + gate.setPlans("plan-1") + builds.waiting = []*nats.Msg{buildAsk()} + ready.waiting = []*nats.Msg{askedAgo(31*time.Minute, 1, true)} + other.waiting = []*nats.Msg{askedAgo(50*time.Minute, 2, false)} + msg, q := nextAsk(t, m) + if msg == nil || q.kind != queueOther || !strings.Contains(string(msg.Data), `"number":2`) { + t.Fatalf("the oldest check (50 minutes, other queue) was not let through first: %v", q) + } + if !m.letKey || gate.taken != 1 { + t.Fatalf("the let-through key was not created for it: held %v, created %d", m.letKey, gate.taken) + } +} + +// 1688: several checks past the bound go one at a time, oldest first: while the key is there every other holder takes builds. +func TestChecksPastTheBoundGoOneAtATimeOldestFirst(t *testing.T) { + m, builds, ready, other, gate := aMachine() + gate.setPlans("plan-1") + ready.waiting = []*nats.Msg{askedAgo(33*time.Minute, 1, true), askedAgo(31*time.Minute, 3, true)} + other.waiting = []*nats.Msg{askedAgo(40*time.Minute, 2, false)} + var order []string + for range 3 { + msg, q := nextAsk(t, m) + if msg == nil { + t.Fatalf("no check let through after %v", order) + } + var r BuildRequest + _ = json.Unmarshal(msg.Data, &r) + order = append(order, fmt.Sprint(r.Check.Number, q.kind)) + // While it runs another holder (the same fake gate, key held) takes builds only. + builds.waiting = []*nats.Msg{buildAsk()} + other2 := &natsMachine{on: "shanks", seat: broker.TheBuildSeat, subs: m.subs, gate: gate} + if next, nq := nextAsk(t, other2); next == nil || nq.kind != queueBuilds { + t.Fatalf("a second holder did not take a delivery build while the let-through key is held: %v", nq) + } + m.dropLetThrough(context.Background()) + } + if want := []string{"2other", "1ready", "3ready"}; strings.Join(order, " ") != strings.Join(want, " ") { + t.Fatalf("checks went %v, want %v (oldest first, one at a time)", order, want) + } +} + +// 1688: a check younger than the bound is never let through, and no key is created for it. +func TestACheckYoungerThanTheBoundIsHeld(t *testing.T) { + m, _, ready, _, gate := aMachine() + gate.setPlans("plan-1") + ready.waiting = []*nats.Msg{askedAgo(29*time.Minute, 1, true)} + if msg, _ := nextAsk(t, m); msg != nil { + t.Fatal("a check of 29 minutes was let through") + } + if gate.taken != 0 { + t.Fatal("a let-through key was created for a young check") + } + if ready.fetches() != 0 { + t.Fatal("the age was read by a fetch; it must be read without taking") + } +} + +// 1688: the holder that lost the race for the key, or the check, takes builds only; and a younger check that arrives in the +// moment between the read and the fetch is handed back with the key released. +func TestAYoungerCheckFetchedInTheRaceIsHandedBackAndTheKeyReleased(t *testing.T) { + m, _, ready, _, gate := aMachine() + gate.setPlans("plan-1") + // The read sees the old check; before the fetch another holder takes it and a young one is first. + gate.oldestFrom[m.subs[1]] = &fakeQueue{waiting: []*nats.Msg{askedAgo(40*time.Minute, 1, true)}} + ready.waiting = []*nats.Msg{askedAgo(time.Minute, 2, true)} + if msg, _ := nextAsk(t, m); msg != nil { + t.Fatalf("a young check was run during a delivery: %s", msg.Data) + } + if gate.letKey || m.letKey { + t.Fatal("the let-through key was left held") + } +} + +// The key of a check let through is refreshed while it runs and deleted when it ends. +func TestTheLetThroughKeyLivesAsLongAsTheCheckRuns(t *testing.T) { + m, _, ready, _, gate := aMachine() + gate.setPlans("plan-1") + ready.waiting = []*nats.Msg{askedAgo(40*time.Minute, 1, true)} + if msg, _ := nextAsk(t, m); msg == nil { + t.Fatal("not let through") + } + end := m.keepLetThrough(context.Background(), 20*time.Millisecond) + time.Sleep(120 * time.Millisecond) + gate.mu.Lock() + held, kept := gate.letKey, gate.kept + gate.mu.Unlock() + if !held || kept == 0 { + t.Errorf("while the check ran the key was held %v and refreshed %d times", held, kept) + } + end() + gate.mu.Lock() + defer gate.mu.Unlock() + if gate.letKey || m.letKey { + t.Error("the key was not deleted when the check ended") + } +} diff --git a/internal/link/queue.go b/internal/link/queue.go index c9c3b0fe..c9656e28 100644 --- a/internal/link/queue.go +++ b/internal/link/queue.go @@ -88,6 +88,40 @@ type Queue struct { AckPending int `json:"ack-pending"` MaxDeliver int `json:"max-deliver"` Asks []QueuedAsk `json:"asks"` + // Hold is why the checks wait: the delivery plans building and the check let through now (ADR 0483, issues 1687 and 1688); + // HoldFault says the bucket could not be read, in which case Hold is empty. + Hold CheckHold `json:"hold"` + HoldFault string `json:"hold-fault,omitempty"` +} + +// LetThroughOf says which ask is let through under ADR 0483 clause 3: the check that waited LetThroughAfter or longer and is next +// (the oldest waiting check across the two check queues, while a delivery builds), and the check in flight that was let through +// (the let-through key names the machine that runs it). Empty when none. +func (q Queue) LetThroughOf(now time.Time) (next, running string) { + if len(q.Hold.Plans) == 0 && q.Hold.LetThrough == nil { + return "", "" + } + var oldest *QueuedAsk + for i := range q.Asks { + a := &q.Asks[i] + if a.State != AskWaiting || a.Request.Check == nil || a.AskedAt.IsZero() { + continue + } + if oldest == nil || a.AskedAt.Before(oldest.AskedAt) { + oldest = a + } + } + if oldest != nil && len(q.Hold.Plans) > 0 && now.Sub(oldest.AskedAt) >= LetThroughAfter { + next = oldest.ID + } + if q.Hold.LetThrough != nil { + for _, a := range q.Asks { + if a.State == AskInFlight && a.Request.Check != nil && a.On == q.Hold.LetThrough.Builder { + running = a.ID + } + } + } + return next, running } // Of is the asks in one state, in queue order. @@ -319,6 +353,16 @@ func ReadQueue(ctx context.Context, js *broker.JetStream, seat string) (Queue, e for _, r := range reads { q.Asks = append(q.Asks, ClassifyQueue(r.asks, r.delivered, r.ackPending, heard)...) } + // And why the checks wait, for the build seat that has them (ADR 0483 clause 5). A bucket that cannot be read is said by the + // queue, which then names no plan. + if seat == broker.TheBuildSeat { + hold, err := ReadCheckHold(ctx, js.Conn()) + if err != nil { + q.HoldFault = err.Error() + } else { + q.Hold = hold + } + } return q, nil }