From 2819915e8e1918b31418c2e33ef65c328232ecc2 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 12 Oct 2026 23:53:06 +0200 Subject: [PATCH] Stop the controller's tests waiting out fixed intervals Reading the build events waited two seconds on every queue read, the race detector's exit sleep cost a second per command line a test runs, and the lease and plan-wait tests waited out production intervals (novox/hq issue 591). --- cmd/mesh-controller/acting.go | 6 +++++- cmd/mesh-controller/lease_test.go | 14 ++++++++++---- cmd/mesh-controller/plans_follow.go | 5 +++-- cmd/mesh-controller/plans_follow_test.go | 8 +++++++- cmd/mesh-controller/sayable_test.go | 8 ++++++++ internal/link/queue.go | 19 ++++++++++++++----- 6 files changed, 47 insertions(+), 13 deletions(-) diff --git a/cmd/mesh-controller/acting.go b/cmd/mesh-controller/acting.go index bae46dea..46b2bb07 100644 --- a/cmd/mesh-controller/acting.go +++ b/cmd/mesh-controller/acting.go @@ -64,6 +64,10 @@ type actor struct { // the highest epoch issued, and what was said of it; zero when it was not (S12). reset time.Time resetSaid string + // renewEvery and poll, when set, are how often the serving controller renews the lease it holds and + // looks again at one another holds, in place of the lease's own: a test's, so two controllers hand the + // lease over in a second rather than in a renewal of five (novox/hq issue 591). + renewEvery, poll time.Duration } // theLease is this process's standing to the lease. @@ -223,7 +227,7 @@ func (a *actor) serveUnderTheLease(ctx context.Context, inv *inventory.Inventory } say := func(format string, args ...any) { fmt.Printf(format+"\n", args...) } l, err := lease.Open(ctx, api, broker.LeaseBucket, lease.Options{Holder: holderOf(instance), - Floor: inv.HighestEpoch, Say: say, Health: controllerHealth, Moved: func(was, floor uint64) { + RenewEvery: a.renewEvery, Poll: a.poll, Floor: inv.HighestEpoch, Say: say, Health: controllerHealth, Moved: func(was, floor uint64) { a.mu.Lock() defer a.mu.Unlock() a.reset = time.Now() diff --git a/cmd/mesh-controller/lease_test.go b/cmd/mesh-controller/lease_test.go index 29aeb2e9..3d715d54 100644 --- a/cmd/mesh-controller/lease_test.go +++ b/cmd/mesh-controller/lease_test.go @@ -55,8 +55,13 @@ func TestTwoControllersOneActs(t *testing.T) { inv := inventory.ForTest(t) ctx := t.Context() + // Each renews five times and looks again ten times as often as a serving controller does, so the + // handover and the loss below are seen in a second, not in a renewal of five (novox/hq issue 591): what + // is checked is the order of events, which does not change with the interval. + const renewEvery, poll = lease.RenewEvery / 5, lease.Poll / 10 + // Controller A takes the lease. - a := &actor{} + a := &actor{renewEvery: renewEvery, poll: poll} aCtx, stopA := context.WithCancel(ctx) defer stopA() lostA, err := a.serveUnderTheLease(aCtx, inv, url) @@ -69,7 +74,7 @@ func TestTwoControllersOneActs(t *testing.T) { } // Controller B starts while A holds it, and waits — acting on nothing meanwhile. - b := &actor{} + b := &actor{renewEvery: renewEvery, poll: poll} bCtx, stopB := context.WithCancel(ctx) defer stopB() tookB := make(chan (<-chan struct{}), 1) @@ -82,10 +87,11 @@ func TestTwoControllersOneActs(t *testing.T) { } tookB <- lost }() + // B looks fifteen times, across one of A's renewals at least. select { case <-tookB: t.Fatal("B took the lease while A held it") - case <-time.After(3 * time.Second): + case <-time.After(15 * poll): } if _, err := b.epoch(ctx); !errors.Is(err, lease.ErrNotHeld) { t.Fatalf("B, waiting, may act: %v", err) @@ -125,7 +131,7 @@ func TestTwoControllersOneActs(t *testing.T) { } select { case <-lostB: - case <-time.After(2 * lease.RenewEvery): + case <-time.After(2 * renewEvery): t.Fatal("B was not told it lost the lease") } if _, err := b.epoch(ctx); !errors.Is(err, lease.ErrNotHeld) { diff --git a/cmd/mesh-controller/plans_follow.go b/cmd/mesh-controller/plans_follow.go index daaaeed7..cf48e194 100644 --- a/cmd/mesh-controller/plans_follow.go +++ b/cmd/mesh-controller/plans_follow.go @@ -30,8 +30,9 @@ var maxPlanWait = link.LongestAnswer - planWaitStart // answered that the call is still running, and `calls` with its id says what came of it. const planWaitStart = 5 * time.Second -// planWaitEvery is how often a wait reads the plan again. -const planWaitEvery = time.Second +// planWaitEvery is how often a wait reads the plan again. A variable so a test reads it more often +// (novox/hq issue 591). +var planWaitEvery = time.Second // The things a wait waits until. var planWaitUntil = []string{"moved", "tier", "done"} diff --git a/cmd/mesh-controller/plans_follow_test.go b/cmd/mesh-controller/plans_follow_test.go index b3932d59..cc8e0db0 100644 --- a/cmd/mesh-controller/plans_follow_test.go +++ b/cmd/mesh-controller/plans_follow_test.go @@ -138,6 +138,12 @@ func TestThePlansVerbComposesAWait(t *testing.T) { func TestAPlanFollowedFromTheStoreAnswersWhenItMoves(t *testing.T) { open := aMesh(t) ctx := t.Context() + // Read every 50ms rather than every second, so the wait answers as soon as the plan moves and the bound + // below passes in a fraction of a second (novox/hq issue 591); the move still waits 1.5s, long enough + // for the wait to have read the plan before it. + was := planWaitEvery + planWaitEvery = 50 * time.Millisecond + t.Cleanup(func() { planWaitEvery = was }) now := time.Now().UTC() walk := inventory.Plan{ID: "plan-229", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c229c229", Created: now, State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}}, @@ -173,7 +179,7 @@ func TestAPlanFollowedFromTheStoreAnswersWhenItMoves(t *testing.T) { t.Errorf("it answered after %s, not when the plan moved", waited) } out = printed(t, func() error { - return plansCommand(ctx, []string{"wait", walk.ID, "--bound", "2"}) + return plansCommand(ctx, []string{"wait", walk.ID, "--bound", "300ms"}) }) if err := json.Unmarshal([]byte(out), &got); err != nil || got.Came != "bound" || got.Until != "moved" { t.Fatalf("a plan that did not move answered %+v (%v)", got, err) diff --git a/cmd/mesh-controller/sayable_test.go b/cmd/mesh-controller/sayable_test.go index d0919ec1..8f98b992 100644 --- a/cmd/mesh-controller/sayable_test.go +++ b/cmd/mesh-controller/sayable_test.go @@ -26,6 +26,14 @@ import ( // shape fails the suite, naming its source. The keeper would say such a summary in words at run time; // this is where the producer is made to say it rightly in the first place. func TestMain(m *testing.M) { + // **A command line this suite runs ends when it is done** (novox/hq issue 591). Many tests run this test + // binary as a verb's command line, and a binary built with the race detector sleeps a second before it + // exits (GORACE's atexit_sleep_ms), which was more than the rest of each such test. The children still + // run under the race detector and still fail on a race they find; they only stop waiting at their end. + // Set before any child starts, and only where nobody chose GORACE. + if _, chosen := os.LookupEnv("GORACE"); !chosen { + _ = os.Setenv("GORACE", "atexit_sleep_ms=0") + } // The process a mesh-cli test runs as a command line: it says the verb and the caller it was given, and // ends (meshcli_test.go). // The process a mesh-cli test runs as `secret accept`: it reads its value exactly as `secret accept` does diff --git a/internal/link/queue.go b/internal/link/queue.go index 41abbb9e..77c21eb3 100644 --- a/internal/link/queue.go +++ b/internal/link/queue.go @@ -305,11 +305,23 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin return heard, fmt.Errorf("cannot read %s from the bus: %w", subject, err) } defer func() { _ = sub.Unsubscribe() }() - for { - batch, err := sub.Fetch(500, nats.MaxWait(2*time.Second)) + // **Read what is there, then stop** (novox/hq issue 591): the consumer says how many events it holds + // from the moment asked, and the reading ends once they are read. A fetch for a full batch waits out + // its whole wait when fewer are there, so every reading of the queue took two seconds and more. + info, err := sub.ConsumerInfo() + if err != nil { + return heard, fmt.Errorf("cannot read %s from the bus: %w", subject, err) + } + for remaining := info.NumPending; remaining > 0; { + batch, err := sub.Fetch(int(min(remaining, 500)), nats.MaxWait(2*time.Second)) if err != nil && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.DeadlineExceeded) { return heard, fmt.Errorf("reading %s: %w", subject, err) } + // None in the whole wait: what was counted is gone from the stream since (its limits), not coming. + if len(batch) == 0 { + break + } + remaining -= uint64(len(batch)) for _, msg := range batch { switch msg.Subject { case BuildStartedOf(seat): @@ -326,9 +338,6 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin } } } - if len(batch) < 500 { - break - } if ctx.Err() != nil { return heard, ctx.Err() }