From 3d49a971a1fe0ab00de883510385042a1990c78c Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 11:56:28 +0000 Subject: [PATCH] Ask a module again at the next look when the bus did not answer, and fail the plan only after ten minutes (issue 1334) --- cmd/mesh-controller/ask_bus_moment_test.go | 107 +++++++++++++++++++++ cmd/mesh-controller/ask_bus_wait.go | 48 +++++++++ cmd/mesh-controller/release_plan.go | 10 +- internal/inventory/plans.go | 3 + 4 files changed, 167 insertions(+), 1 deletion(-) create mode 100644 cmd/mesh-controller/ask_bus_moment_test.go create mode 100644 cmd/mesh-controller/ask_bus_wait.go diff --git a/cmd/mesh-controller/ask_bus_moment_test.go b/cmd/mesh-controller/ask_bus_moment_test.go new file mode 100644 index 00000000..fb50dfd6 --- /dev/null +++ b/cmd/mesh-controller/ask_bus_moment_test.go @@ -0,0 +1,107 @@ +package main + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// Issue 1334: a bus that does not answer for a moment (it reloads its user list, and JetStream's API has no responder for a few +// seconds) is not a failed build. The module is left unasked and the plan building, and the next look asks it again; only a +// module's own fault, or a bus silent for ten minutes, fails the plan. + +func aTierOfThree() (*inventory.Plan, map[string]inventory.Entry) { + p := &inventory.Plan{ID: "plan-1", State: inventory.PlanBuilding, Tiers: [][]string{{"a", "b", "c"}}, + Modules: map[string]*inventory.PlanModule{}} + byName := map[string]inventory.Entry{} + for _, m := range []string{"a", "b", "c"} { + byName[m] = inventory.Entry{Source: inventory.Source{Repository: "novox/" + m, Seat: "git", Ref: "main"}} + } + return p, byName +} + +func TestABusThatDoesNotAnswerForAMomentDoesNotFailTheTier(t *testing.T) { + p, byName := aTierOfThree() + bus := false + was := askABuild + askABuild = func(_ context.Context, source buildSource, _, _ string) (string, error) { + if source.Repository == "novox/b" && !bus { + return "", fmt.Errorf("%w: the bucket mesh-controller_builds is not on the bus: nats: no responders available for request", + inventory.ErrNoChain) + } + return "build-" + source.Repository, nil + } + t.Cleanup(func() { askABuild = was }) + + for _, m := range []string{"a", "b", "c"} { + askModule(t.Context(), p, m, byName) + } + if p.State != inventory.PlanBuilding { + t.Fatalf("a bus that did not answer for a moment failed the plan: %s, %q", p.State, p.Note) + } + if p.Modules["a"].State != "asked" || p.Modules["c"].State != "asked" { + t.Fatalf("the modules the bus did answer for were not asked: %+v %+v", p.Modules["a"], p.Modules["c"]) + } + if s := p.Modules["b"]; s.State != "" { + t.Fatalf("the module the bus did not answer for is %q, not left unasked", s.State) + } + if got := unaskedOf(*p); len(got) != 1 || got[0] != "b" { + t.Fatalf("the next look would ask %v, not b", got) + } + if !strings.Contains(p.Note, "bus did not answer") { + t.Fatalf("the plan says nothing of the wait: %q", p.Note) + } + + // The next look, the bus answering: b is asked and the plan goes on. + bus = true + for _, m := range unaskedOf(*p) { + askModule(t.Context(), p, m, byName) + } + if p.State != inventory.PlanBuilding || p.Modules["b"].State != "asked" || p.Modules["b"].Build == "" { + t.Fatalf("b was not asked once the bus answered: %s %+v", p.State, p.Modules["b"]) + } +} + +func TestABusSilentForTenMinutesFailsTheTierAsBefore(t *testing.T) { + p, byName := aTierOfThree() + was := askABuild + askABuild = func(context.Context, buildSource, string, string) (string, error) { + return "", fmt.Errorf("%w: nats: no responders available for request", inventory.ErrNoChain) + } + t.Cleanup(func() { askABuild = was }) + askModule(t.Context(), p, "a", byName) + if p.State != inventory.PlanBuilding { + t.Fatalf("the first refusal failed the plan: %q", p.Note) + } + long := time.Now().Add(-busWaitBound - time.Minute) + p.Modules["a"].BusWaitFrom = &long + askModule(t.Context(), p, "a", byName) + if p.State != inventory.PlanFailed || p.Modules["a"].State != "failed" { + t.Fatalf("a bus silent past its bound did not fail the plan: %s %+v", p.State, p.Modules["a"]) + } +} + +func TestAModulesOwnFaultStillFailsThePlan(t *testing.T) { + p, byName := aTierOfThree() + delete(byName, "a") + askModule(t.Context(), p, "a", byName) + if p.State != inventory.PlanFailed || !strings.Contains(p.Note, "no longer in the catalogue") { + t.Fatalf("a module gone from the catalogue did not fail the plan: %s %q", p.State, p.Note) + } + + p, byName = aTierOfThree() + was := askABuild + askABuild = func(context.Context, buildSource, string, string) (string, error) { + return "", errors.New("git.example/novox/a is no repository the catalogue builds a module from") + } + t.Cleanup(func() { askABuild = was }) + askModule(t.Context(), p, "a", byName) + if p.State != inventory.PlanFailed || p.Modules["a"].State != "failed" { + t.Fatalf("a module's own refusal did not fail the plan: %s %+v", p.State, p.Modules["a"]) + } +} diff --git a/cmd/mesh-controller/ask_bus_wait.go b/cmd/mesh-controller/ask_bus_wait.go new file mode 100644 index 00000000..03df932f --- /dev/null +++ b/cmd/mesh-controller/ask_bus_wait.go @@ -0,0 +1,48 @@ +package main + +import ( + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// busWaitBound is how long a module whose ask the bus did not answer is asked again at every look before the plan fails +// (issue 1334). +const busWaitBound = 10 * time.Minute + +// busDidNotAnswer says an ask failed because the bus's JetStream had no responder at that moment, as it has for a few seconds +// when the bus reloads its user list (novox/hq issue 353), and not because of the module: the same ask is made again. +func busDidNotAnswer(err error) bool { + if err == nil { + return false + } + if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrJetStreamNotEnabled) || + errors.Is(err, jetstream.ErrJetStreamNotEnabled) || errors.Is(err, inventory.ErrNoChain) { + return true + } + // An error that crossed a boundary as text only. + text := err.Error() + return strings.Contains(text, "no responders available") || strings.Contains(text, "jetstream not enabled") +} + +// waitForTheBus leaves a module whose ask the bus did not answer unasked, the plan building, and says so in the plan's note: the +// next look asks the tier's unasked modules again. It says false, so the caller fails the plan as for any refusal, once the bus +// has been silent for busWaitBound. +func waitForTheBus(p *inventory.Plan, state *inventory.PlanModule, name string, err error, now time.Time) bool { + if state.BusWaitFrom == nil { + state.BusWaitFrom = &now + } + if now.Sub(*state.BusWaitFrom) > busWaitBound { + return false + } + state.State, state.Why = "", "" + p.Note = fmt.Sprintf("tier %d: %s not asked: the bus did not answer (%v); asked again at the next look, failing after %s", + p.Tier, name, firstLine(err.Error()), busWaitBound) + return true +} diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index e6364de5..2290f025 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -580,12 +580,18 @@ func askModule(ctx context.Context, p *inventory.Plan, name string, byName map[s id, err := askABuild(withReuse(ctx, reuseWithVerdicts(ctx, p.ID, reuseAskOf(p), mergedHeadsOf(p))), source, e.Source.Path, ref) if err != nil { + // **A bus that does not answer for a moment is not a failed build** (issue 1334): the module stays unasked and the + // tier's next look asks it again. A fault of the module's own fails the plan at once, as does a bus silent too long. + if busDidNotAnswer(err) && waitForTheBus(p, state, name, err, now) { + return + } state.State = "failed" state.Why = err.Error() p.State = inventory.PlanFailed p.Note = fmt.Sprintf("%s could not be asked for: %v", name, err) return } + state.BusWaitFrom = nil // The commit of the module's own repository the walk carries (novox/hq ADR 0276): a batch's walk carries // one per repository. commit := carried @@ -887,10 +893,12 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, if len(unasked) < len(tier) { fmt.Printf("%s: tier %d was asked in part; asking the rest: %s\n", p.ID, p.Tier, strings.Join(unasked, ", ")) } + note := p.Note if err := askTier(ctx, inv, p, unasked); err != nil { return false, err } - return true, nil + // Nothing asked and nothing new to say (the bus still silent, issue 1334): not saved again at every look. + return len(unaskedOf(*p)) < len(unasked) || p.Note != note || p.State != inventory.PlanBuilding, nil } // **Asked: settle from the build records first** (novox/hq 04-ISSUES/214). An outcome is taken // in by whichever controller hears it, and a merge to the controller's own repository replaces diff --git a/internal/inventory/plans.go b/internal/inventory/plans.go index 5af4ee7e..ac646f7e 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -340,6 +340,9 @@ type PlanModule struct { State string `json:"state,omitempty"` AskedAt *time.Time `json:"asked_at,omitempty"` BuiltAt *time.Time `json:"built_at,omitempty"` + // BusWaitFrom is when the bus first did not answer an ask of this module (issue 1334): the module stays unasked and is + // asked again at the next look, until the bus has been silent for the bound, when the plan fails as for any refusal. + BusWaitFrom *time.Time `json:"bus_wait_from,omitempty"` // SentAt is when the plan sent the machines running this module its new build, because a // later tier is built by it (ADR 0163's gate): the reports that open the gate are the ones // after this.