Compare commits

...
Author SHA1 Message Date
jschoubben e7cc315fd0 Guard the taking loop's subscriptions, and update the two tests that held a busy machine's pulls
The loop now lets go of and binds queues around every ask, so Close takes the lock the loop does;
the busy machine's test says it holds no pull, and the stray test keeps its long check alive
within its two-second ack wait.
2026-10-21 20:53:58 +02:00
jschoubben 60253c82b7 Fix the review of the checks-wait milestone: redeliveries, deleted modules, the kept queue
A check pending redelivery is read before the let-through fetches (none is handed back);
a deleted module is done for the plan's key; the queue an ask came from is let go of when
it ends. Tries say they run under their own id.
2026-10-21 20:17:37 +02:00
jschoubben bb8f076cc0 Merge remote-tracking branch 'origin/main' into fix/1652-checks-wait 2026-10-21 20:14:12 +02:00
jschoubben bb4d2e1da4 Let go of the other queues while an ask runs, and read the let-through race right
A message that reaches a builder's subscription outside a fetch was held unheartbeated for
the whole run of another ask, behind which a delivery waited (issue 1747). The queues the
running ask did not come from are let go of and bound again after it.
2026-10-21 19:56:23 +02:00
jschoubben a14dc04b7b Hold the build seat's checks while a delivery builds, and let one check that waited 30 minutes through
ADR 0483: the controller keeps a key per plan in its build phase; holders read it and
take no new check while one is there, but for the oldest check past 30 minutes, one
at a time by a let-through key. The queue says why checks wait.
2026-10-21 19:32:47 +02:00
23 changed files with 2216 additions and 88 deletions
+1
View File
@@ -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)
}
+238
View File
@@ -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" && st.State != planDeleted) {
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:
}
}
}
+175
View File
@@ -0,0 +1,175 @@
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},
// A module the plan deletes is done as far as its tiers go (release_plan.go), so it holds no key through the rollout.
{"a deleted module is done", aPlan("p", inventory.PlanRolling, tiers, map[string]string{"a": "built", "b": "built", "c": planDeleted}), false},
{"failed", aPlan("p", inventory.PlanFailed, tiers, map[string]string{"a": "failed", "b": "asked"}), false},
{"superseded", aPlan("p", inventory.PlanSuperseded, tiers, map[string]string{"a": "asked"}), false},
{"a batch whose window is open", aPlan("p", inventory.PlanAssembling, tiers, map[string]string{"a": "asked"}), false},
}
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")
}
}
+2
View File
@@ -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) })
+40 -6
View File
@@ -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 {
+77
View File
@@ -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")
}
}
+3
View File
@@ -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 {
+9
View File
@@ -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
}
+34
View File
@@ -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)
}
}
}
}
+38 -2
View File
@@ -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 id>` (`plan-<digits>`), 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): `<plan id>` 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,
+5
View File
@@ -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 " +
+134
View File
@@ -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"))
}
}
+1 -1
View File
@@ -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" }
} }
+4
View File
@@ -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.
+2
View File
@@ -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",
+174 -62
View File
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"os"
"sync"
"time"
"github.com/nats-io/nats.go"
@@ -162,11 +163,21 @@ 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
// mu guards the subscriptions of subs, bound and let go of by the taking loop and closed by Close.
mu sync.Mutex
}
// 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 +185,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 +211,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.
@@ -222,6 +241,8 @@ const (
)
func (m *natsMachine) Close() {
m.mu.Lock()
defer m.mu.Unlock()
for _, q := range m.subs {
if q.sub != nil {
_ = q.sub.Unsubscribe()
@@ -244,6 +265,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"}})
@@ -288,6 +312,19 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
}
continue
}
// The seat's own queue is bound again when a running ask made the machine let go of it (quiesce): a machine that cannot
// is not taking, which is said and tried again.
if !m.subs[0].bound {
if err := m.bind(m.subs[0]); err != nil {
fmt.Fprintf(os.Stderr, "this machine cannot bind %s again after a build, and tries again: %v\n", m.seat, err)
select {
case <-ctx.Done():
return nil
case <-time.After(pausedPoll):
}
continue
}
}
if time.Now().After(bindAgain) {
for _, q := range m.subs[1:] {
if !q.bound {
@@ -312,7 +349,7 @@ func (m *natsMachine) Take(ctx context.Context, do func(context.Context, Build))
// say why (2026-10-03, the first build agents). Asked again.
continue
default:
if m.subs[0].sub.IsValid() {
if m.subs[0].sub != nil && m.subs[0].sub.IsValid() {
// A transient fault in asking — a reconnect, a slow server — is asked past rather
// than ending the machine; one that outlasts the ack wait redelivers nothing lost.
time.Sleep(time.Second)
@@ -320,67 +357,106 @@ 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)
m.handOver(ctx, msg, q, do)
// And the queues it let go of are bound again at once, not after the half minute.
bindAgain = time.Time{}
}
}
// 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)
// **And holds nothing else while it runs** (issue 1747): see quiesce.
m.quiesce(q)
do(ctx, build)
stopRunning()
close(working)
// **The queue it came from too, once the ask is settled**: a fetch that returned a late message leaves its pull request
// open, and the server can deliver one more into the subscription (issue 1747). Take binds every queue again first.
m.quiesce(nil)
}
// quiesce lets go of every queue but the one the running ask came from, for as long as it runs. **A builder never holds a
// message it does not start** (issue 1747). A message the server sends after a fetch gave up, within the few milliseconds its
// pull request still lived, lands in that queue's subscription and stays there until the next fetch of it: held, never
// acknowledged as in progress, for as long as the builder runs its current ask, five or ten minutes for a check, while the
// holders that could build it are all busy and a delivery waits behind it. The legacy pull subscription cannot be read
// without sending a new pull request, which could take one more, so it cannot be searched for such a message. It is dropped
// instead: the subscription is closed, a message in it or on its way is lost to this builder and the server hands it out again
// when its worker's wait for an acknowledgement ends (a minute), to any holder that is free, and the queue is bound again
// before the next fetch (Take). The queue the running ask came from keeps its subscription, which its acknowledgements use.
func (m *natsMachine) quiesce(keep *takeQueue) {
m.mu.Lock()
defer m.mu.Unlock()
for _, q := range m.subs {
if q == keep || !q.bound {
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)
if q.sub != nil {
_ = q.sub.Unsubscribe()
}
q.sub, q.pull, q.bound = nil, nil, false
}
}
@@ -420,25 +496,39 @@ func duplicateOf(running CheckRunning, found bool, r BuildRequest, me string) st
// subscribing `…accept.build` against a consumer filtered on `…accept.>` is rejected even though it is narrower. Learned
// twice now, on two different consumers, which is why it is written down here.
func (m *natsMachine) bind(q *takeQueue) error {
m.mu.Lock()
defer m.mu.Unlock()
sub, err := m.js.Context().PullSubscribe(q.worker.Filters[0], q.worker.Name,
nats.Bind(q.worker.Stream, q.worker.Name), nats.ManualAck())
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 +544,18 @@ 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) {
if !m.subs[0].bound {
// Let go of while an ask ran and not bound again yet (Take binds it before it asks): nothing to take from.
return nil, nil, nil
}
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
}
@@ -619,3 +718,16 @@ func endCancelled(ctx context.Context, b *natsBuild) {
}
_ = b.msg.Term()
}
// boundKinds are the kinds of queue this machine holds a subscription to now.
func (m *natsMachine) boundKinds() []string {
m.mu.Lock()
defer m.mu.Unlock()
var out []string
for _, q := range m.subs {
if q.bound {
out = append(out, q.kind)
}
}
return out
}
+6 -2
View File
@@ -287,12 +287,16 @@ type CheckRunning struct {
}
// RunningKeyOf is the key a check is said running under: its pull request's tokens, or `group.<name>` for a group's
// composed check; none for a try, which no other request replaces.
// composed check; `try.<id>` for a try, which no other request replaces.
func RunningKeyOf(r BuildRequest) (string, bool) {
c := r.Check
switch {
case c == nil, c.Try:
case c == nil:
return "", false
case c.Try:
// A try says it runs under its own id, which no other request shares: so a reader of the bucket can tell a try that runs
// from a check left waiting for a redelivery (ADR 0483 clause 3).
return "try." + tokenOf(r.ID), true
case c.Group != "":
return "group." + tokenOf(c.Group), true
case c.Number <= 0:
+5 -15
View File
@@ -6,7 +6,6 @@ import (
"context"
"encoding/json"
"reflect"
"strings"
"testing"
"time"
@@ -595,21 +594,12 @@ func TestABusyMachineTakesNoWaitingAskWhateverItsPullsHold(t *testing.T) {
case <-time.After(10 * time.Second):
t.Fatal("builder a took nothing")
}
var other *takeQueue
for _, q := range a.subs {
if q.kind == queueOther && q.bound {
other = q
}
}
if other == nil {
t.Fatal("builder a has no pull on the other checks")
}
// The queue is empty, so this request is answered at its expiry with a status message that lands in the pull.
reply := strings.TrimSuffix(other.sub.Subject, "*") + "late"
if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, reply,
[]byte(`{"batch":1,"expires":1000000}`)); err != nil {
t.Fatal(err)
// **A busy machine holds no pull at all** (issues 1649 and 1747): the queues the running ask did not come from are let go of,
// so no message, a status message or a late delivery, can sit in one while it runs.
if kinds := a.boundKinds(); len(kinds) != 1 || kinds[0] != queueBuilds {
t.Fatalf("builder a holds %v while it builds; want only the queue its build came from", kinds)
}
other := &takeQueue{kind: queueOther, worker: broker.CheckWorkers()[1]}
time.Sleep(500 * time.Millisecond)
for i, id := range []string{"wait-1", "wait-2"} {
if err := asks.Ask(ctx, aPullCheck(id, 20+i, "head-x-000000000", "trunk-1-000000", false)); err != nil {
@@ -0,0 +1,479 @@
//go:build integration
package link
import (
"context"
"encoding/json"
"fmt"
"github.com/nats-io/nats.go"
"reflect"
"strings"
"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{}{}
// It is deleted when the check ends: the next check let through creates the key again at once, and a create fails
// while the key is there, so its start (the next round) is the proof; after the last one the key stays gone.
if len(order) == 3 {
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 last 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)
}
}
// Issue 1747: a message that reaches a builder's subscription outside a fetch is not held while the builder runs another ask. A
// stray check lands in the other queue's subscription of builder A, which then runs a long check; the server hands the stray
// out again when its wait for an acknowledgement ends (two seconds here), builder B runs it, and A, finished, does not run it a
// second time from the subscription it held.
func TestNatsAMessageDeliveredOutsideAFetchIsNotHeldWhileTheBuilderRunsAnotherAsk(t *testing.T) { // 1747
js := aBusWithTheBuildRole(t)
ctx, stop := context.WithCancel(t.Context())
defer stop()
for _, w := range broker.CheckWorkers() {
w.AckWaitSeconds = 2
if err := js.EnsureConsumer(w); err != nil {
t.Fatal(err)
}
}
// The long check says it is alive well within the ack wait of two seconds.
was := heartbeatWhileWorking
heartbeatWhileWorking = 400 * time.Millisecond
t.Cleanup(func() { heartbeatWhileWorking = was })
asks := BuildsOn(js, TheBuildMachine)
long := aPullCheck("build-"+fmt.Sprint(time.Now().UnixNano()), 1, "head-1-0000000", "trunk-0-000000", true)
stray := aPullCheck("build-"+fmt.Sprint(time.Now().UnixNano()+1), 2, "head-2-0000000", "trunk-0-000000", false)
for _, r := range []BuildRequest{long, stray} {
if err := asks.Ask(ctx, r); err != nil {
t.Fatal(err)
}
}
var mu sync.Mutex
runs := map[string][]string{} // ask id -> the machines that ran it
release := make(chan struct{})
started := make(chan string, 8)
do := func(on string) func(context.Context, Build) {
return func(ctx context.Context, work Build) {
id := work.Request().ID
mu.Lock()
runs[id] = append(runs[id], on)
mu.Unlock()
started <- on + ":" + id
if id == long.ID {
<-release
}
_ = work.Announce(ctx, BuildResult{ID: id, On: on, Failed: "no"})
_ = work.Done()
}
}
var a *natsMachine
strayLanded := make(chan error, 1)
a = MachineOverNATSWith(js, "ace", TheBuildMachine, MachineOptions{Claim: func() bool {
// The moment the long check is claimed: the server hands the other queue's waiting check to A's subscription there, as
// one that arrives after a fetch gave up does. Done once.
select {
case <-strayLanded:
return true
default:
}
other := a.subs[2]
inbox := strings.TrimSuffix(other.sub.Subject, "*") + "stray"
body, _ := json.Marshal(map[string]any{"batch": 1, "expires": int64(30 * time.Second)})
if err := js.Conn().PublishRequest("$JS.API.CONSUMER.MSG.NEXT."+other.worker.Stream+"."+other.worker.Name, inbox, body); err != nil {
strayLanded <- err
return true
}
_ = js.Conn().Flush()
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) {
if n, _, _ := other.sub.Pending(); n > 0 {
break
}
}
strayLanded <- nil
return true
}}).(*natsMachine)
defer a.Close()
go func() { _ = a.Take(ctx, do("ace")) }()
select {
case got := <-started:
if got != "ace:"+long.ID {
t.Fatalf("started %s, want the long check on ace first", got)
}
case <-time.After(15 * time.Second):
t.Fatal("the long check did not start")
}
if err := <-strayLanded; err != nil {
t.Fatal(err)
}
// B, free, runs the stray once the server hands it out again.
b := MachineOverNATS(js, "shanks")
defer b.Close()
go func() { _ = b.Take(ctx, do("shanks")) }()
select {
case got := <-started:
if got != "shanks:"+stray.ID {
t.Fatalf("started %s, want the stray on shanks", got)
}
case <-time.After(20 * time.Second):
t.Fatal("the stray was held by the builder that ran the long check and nobody ran it")
}
// A finishes, and does not run the stray from the subscription it held.
release <- struct{}{}
select {
case got := <-started:
t.Fatalf("%s started after the long check: the stray ran twice", got)
case <-time.After(6 * time.Second):
}
mu.Lock()
defer mu.Unlock()
if len(runs[stray.ID]) != 1 {
t.Fatalf("the stray ran on %v, want once", runs[stray.ID])
}
}
// Review of 1688: a young check pending redelivery (handed back by a Nak) stands ahead of a check older than 30 minutes in the
// same queue while a plan builds. The let-through fetch would return the young one first, so nothing is fetched: the young
// check's delivery count stays 1, it is not handed back every second until it is dead-lettered, and it is not run.
func TestNatsAYoungRedeliveryAheadOfAnOldCheckIsNeverFetchedByTheLetThrough(t *testing.T) {
js, plans := aBusWithTheBuckets(t)
ctx, stop := context.WithCancel(t.Context())
defer stop()
asks := BuildsOn(js, TheBuildMachine)
young := aCheckAskedAgo(time.Minute, 1, false)
if err := asks.Ask(ctx, young); err != nil {
t.Fatal(err)
}
// Someone took it and handed it back: it is pending redelivery.
worker := broker.CheckWorkers()[1]
sub, err := js.Context().PullSubscribe(worker.Filters[0], worker.Name, nats.Bind(worker.Stream, worker.Name), nats.ManualAck())
if err != nil {
t.Fatal(err)
}
got, err := sub.Fetch(1, nats.MaxWait(5*time.Second))
if err != nil || len(got) != 1 {
t.Fatalf("the young check was not fetched to be handed back: %v", err)
}
if err := got[0].Nak(); err != nil {
t.Fatal(err)
}
_ = sub.Unsubscribe()
old := aCheckAskedAgo(45*time.Minute, 2, false)
if err := asks.Ask(ctx, old); err != nil {
t.Fatal(err)
}
if _, err := plans.Put(ctx, "plan-1", []byte(`{"plan":"plan-1"}`)); err != nil {
t.Fatal(err)
}
started := make(chan string, 4)
machine := MachineOverNATS(js, "ace")
defer machine.Close()
go func() {
_ = machine.Take(ctx, func(ctx context.Context, work Build) {
started <- work.Request().ID
_ = work.Announce(ctx, BuildResult{ID: work.Request().ID, On: "ace", Failed: "no"})
_ = work.Done()
})
}()
select {
case id := <-started:
t.Fatalf("%s was run while the young redelivery stood ahead of it", id)
case <-time.After(6 * time.Second):
}
info, err := js.Context().ConsumerInfo(worker.Stream, worker.Name)
if err != nil {
t.Fatal(err)
}
if info.Delivered.Consumer != 1 || info.NumRedelivered != 0 {
t.Fatalf("the young check was handed out %d time(s), %d redelivered: the let-through fetched it", info.Delivered.Consumer, info.NumRedelivered)
}
}
+384
View File
@@ -0,0 +1,384 @@
package link
import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"strconv"
"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-<digits>`), 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 (`<owner>.<repo>.<number>`) and no group's (`group.<name>`).
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 is when the check a fetch of this queue would return was asked, read without taking anything.
//
// **It sees redeliveries** (review of issue 1688). A work queue keeps only what is not acknowledged: what the worker handed out
// (sequence up to its delivered mark) is running or waiting to be handed out again, and a fetch returns those before anything
// new. An ask there that is not said running (a key of the checks running names its build) is such a redelivery. If one is
// younger than the bound, the queue's next fetch would return it, and a young check is never let through: the queue is not a
// candidate, and nothing is fetched, so nothing is handed back and no delivery is spent. Otherwise the candidate is the oldest
// of the redeliveries and the first ask after the delivered mark.
func (g busGate) OldestWaiting(ctx context.Context, q *takeQueue) (time.Time, bool, error) {
info, err := jsOf(g.conn).ConsumerInfo(q.worker.Stream, q.worker.Name, nats.Context(ctx))
if err != nil {
return time.Time{}, false, fmt.Errorf("the %s worker: %w", q.kind, err)
}
var oldest time.Time
note := func(at time.Time) {
if oldest.IsZero() || at.Before(oldest) {
oldest = at
}
}
for seq := uint64(1); seq <= info.Delivered.Stream; {
data, stored, got, err := g.directGet(ctx, q, seq)
if err != nil {
return time.Time{}, false, err
}
if got == 0 || got > info.Delivered.Stream {
break
}
seq = got + 1
var r BuildRequest
if json.Unmarshal(data, &r) != nil {
continue
}
running, ok, err := ReadRunning(ctx, g.conn, r)
if err != nil {
return time.Time{}, false, fmt.Errorf("the checks running: %w", err)
}
if ok && running.Build == r.ID {
continue // running now
}
at := askedAtOf(data, stored)
if time.Since(at) < LetThroughAfter {
return time.Time{}, false, nil // a young redelivery is next in line
}
note(at)
}
data, stored, got, err := g.directGet(ctx, q, info.Delivered.Stream+1)
if err != nil {
return time.Time{}, false, err
}
if got != 0 {
note(askedAtOf(data, stored))
}
return oldest, !oldest.IsZero(), nil
}
// directGet reads the first ask at or after a sequence of a check queue's stream, without taking it: its body, the time the
// stream stored it, and its sequence (0 when there is none).
func (g busGate) directGet(ctx context.Context, q *takeQueue, seq uint64) ([]byte, string, uint64, error) {
body, _ := json.Marshal(map[string]any{"seq": seq, "next_by_subj": q.worker.Filters[0]})
reply, err := g.conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET."+q.worker.Stream, body)
if err != nil {
return nil, "", 0, fmt.Errorf("the asks of the %s queue: %w", q.kind, err)
}
if st := reply.Header.Get("Status"); st == "404" || st == "408" {
return nil, "", 0, nil
}
got, _ := strconv.ParseUint(reply.Header.Get("Nats-Sequence"), 10, 64)
return reply.Data, reply.Header.Get("Nats-Time-Stamp"), got, nil
}
// askedAtOf is when a request was asked: its id says (NewBuildID), the stream's own time otherwise, and a request that says
// 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].NakWithDelay(time.Minute) // a delay keeps it off the queue of redeliveries a fetch reads first
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, ", ") }
+307
View File
@@ -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")
}
}
+44
View File
@@ -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
}
+54
View File
@@ -0,0 +1,54 @@
package link
import (
"testing"
"time"
"github.com/nats-io/nats.go"
)
// Issue 1747: a builder never holds a message it does not start. A message the server sends after a fetch gave up lands in that
// queue's subscription; while an ask runs, the queues it does not come from are let go of, so such a message is not held
// unheartbeated for the whole run, and the queues are bound again before the next fetch.
func TestARunningAskLetsGoOfTheQueuesItDidNotComeFrom(t *testing.T) {
m, builds, _, other, _ := aMachine()
// A delivery build reached the builds queue's subscription after its fetch gave up: it sits in the buffer.
builds.waiting = []*nats.Msg{buildAsk()}
took := m.subs[1] // the running ask came from the ready queue
m.quiesce(took)
if !took.bound || took.pull == nil {
t.Error("the queue the running ask came from was let go of; its acknowledgements need it")
}
for _, q := range []*takeQueue{m.subs[0], m.subs[2]} {
if q.bound || q.pull != nil || q.sub != nil {
t.Errorf("the %s queue is still held while an ask runs", q.kind)
}
}
// Nothing is fetched from a queue that was let go of: a message in it cannot be taken unheartbeated.
if msg, _, _ := m.next(t.Context()); msg != nil {
t.Fatalf("a message was taken from a queue that was let go of: %s", msg.Data)
}
if builds.fetches() != 0 || other.fetches() != 0 {
t.Errorf("fetched from a queue that was let go of: builds %d, other %d", builds.fetches(), other.fetches())
}
}
// Once the ask is settled the queue it came from is let go of too: a second message the server delivered into its subscription
// on a pull request still open is not run after the ask ends (review of 1747).
func TestTheQueueTheAskCameFromIsLetGoOfWhenTheAskEnds(t *testing.T) {
m, _, ready, _, _ := aMachine()
ready.waiting = []*nats.Msg{askedAgo(time.Minute, 2, true)}
m.quiesce(nil)
for _, q := range m.subs {
if q.bound || q.pull != nil {
t.Errorf("the %s queue is still held after the ask ended", q.kind)
}
}
m.subs[0].bound, m.subs[0].pull = true, &fakeQueue{}
if msg, _, _ := m.next(t.Context()); msg != nil || ready.fetches() != 0 {
t.Fatal("a message held in the kept queue's subscription was taken after the ask ended")
}
}