diff --git a/cmd/mesh-controller/ask_bus_look_integration_test.go b/cmd/mesh-controller/ask_bus_look_integration_test.go new file mode 100644 index 00000000..9f78de1e --- /dev/null +++ b/cmd/mesh-controller/ask_bus_look_integration_test.go @@ -0,0 +1,134 @@ +//go:build integration + +package main + +import ( + "context" + "fmt" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// Issue 1334, the look: when the bus did not answer an ask of any module of a tier, no build is in flight and no event follows, +// so the look itself comes back (it fails, and the looker retries it), asks the modules again, and past the bound fails the plan. +func TestALookComesBackForATierTheBusRefusedWhole(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + for _, name := range []string{"a", "b"} { + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: name, Version: "1"}, + inventory.Source{Repository: "novox/" + name, Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + } + p := inventory.Plan{ID: "plan-bus", Repository: "novox/a", Branch: "main", Commit: "c1", Created: time.Now().UTC(), + State: inventory.PlanBuilding, Tiers: [][]string{{"a", "b"}}, Modules: map[string]*inventory.PlanModule{"a": {}, "b": {}}} + if err := inv.SavePlan(ctx, &p); err != nil { + t.Fatal(err) + } + bus := false + was := askABuild + askABuild = func(_ context.Context, source buildSource, _, _ string) (string, error) { + if !bus { + return "", fmt.Errorf("%w: nats: no responders available for request", inventory.ErrNoChain) + } + return "build-" + source.Repository, nil + } + t.Cleanup(func() { askABuild = was }) + look := func() bool { + t.Helper() + release, err := inv.HoldPlans(ctx, true) + if err != nil { + t.Fatal(err) + } + defer release() + return advanceHeld(ctx, open) + } + read := func() inventory.Plan { + t.Helper() + got, err := inv.PlanByID(ctx, "plan-bus") + if err != nil { + t.Fatal(err) + } + return got + } + + // Every module refused: the look says it failed, which has the looker come back; the plan is building and says why. + if !look() { + t.Fatal("a look that left the whole tier unasked for the bus did not fail, so nothing would come back to it") + } + if got := read(); got.State != inventory.PlanBuilding || !strings.Contains(got.Note, "bus did not answer") || + got.Modules["a"].State != "" || got.Modules["b"].State != "" { + t.Fatalf("the plan after a refused tier: %s %q %+v %+v", got.State, got.Note, got.Modules["a"], got.Modules["b"]) + } + // Again, still refused: the same fact is not saved again, and the look still comes back. + before := read().Revision + if !look() || read().Revision != before { + t.Fatalf("a second refused look failed to come back or saved the plan again (revision %d, was %d)", read().Revision, before) + } + + // The bus answers at the next look: both are asked, and nothing waits any more. + bus = true + if look() { + t.Fatal("a look that asked the whole tier failed") + } + if got := read(); got.Modules["a"].State != "asked" || got.Modules["b"].State != "asked" || got.Modules["a"].BusWaitFrom != nil { + t.Fatalf("the tier was not asked once the bus answered: %+v %+v", got.Modules["a"], got.Modules["b"]) + } +} + +func TestALookFailsTheTierTheBusRefusedPastTheBound(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "a", Version: "1"}, + inventory.Source{Repository: "novox/a", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil { + t.Fatal(err) + } + long := time.Now().UTC().Add(-busWaitBound - time.Minute) + p := inventory.Plan{ID: "plan-bus", Repository: "novox/a", Branch: "main", Commit: "c1", Created: time.Now().UTC(), + State: inventory.PlanBuilding, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {BusWaitFrom: &long}}} + if err := inv.SavePlan(ctx, &p); err != nil { + t.Fatal(err) + } + 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 }) + release, err := inv.HoldPlans(ctx, true) + if err != nil { + t.Fatal(err) + } + advanceHeld(ctx, open) + release() + got, err := inv.PlanByID(ctx, "plan-bus") + if err != nil { + t.Fatal(err) + } + if got.State != inventory.PlanFailed || !strings.Contains(got.Note, "the bus did not answer an ask for 10m0s (since ") { + t.Fatalf("a bus silent past its bound did not fail the plan in words: %s %q", got.State, got.Note) + } + + // Rebuilt by hand while the bus still reloads: said as waiting, not as a failure and not with an empty reason. + joined, said, err := joinAPlan(ctx, open, "a") + if err != nil || !joined || !strings.Contains(said, "not asked yet: the bus did not answer") { + t.Fatalf("a rebuild left for the bus is said as %v %q %v", joined, said, err) + } + // And retried by hand: the module is named as waiting, not as asked again. + got, _ = inv.PlanByID(ctx, "plan-bus") + if got.State == inventory.PlanFailed { + got.Modules["a"].State, got.Modules["a"].Why = "failed", "x" + if err := inv.SavePlan(ctx, &got); err != nil { + t.Fatal(err) + } + said, err := retryPlan(ctx, open, "plan-bus") + if err != nil || !strings.Contains(said, "not asked yet: the bus did not answer") { + t.Fatalf("a retry left for the bus is said as %q %v", said, err) + } + } +} 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..42ced920 --- /dev/null +++ b/cmd/mesh-controller/ask_bus_moment_test.go @@ -0,0 +1,139 @@ +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"]) + } + // Said in plain words: the bus, the wait, and since when. + want := "a could not be asked for: the bus did not answer an ask for 10m0s (since " + long.UTC().Format(time.RFC3339) + + "): this kind of record lives on the bus" + if !strings.HasPrefix(p.Note, want) || p.Modules["a"].Why != strings.TrimPrefix(want, "a could not be asked for: ")+p.Note[len(want):] { + t.Fatalf("the plan says %q, not words starting %q", p.Note, want) + } + // The wait has ended: a Retry made while the bus still reloads waits again, and does not fail at its first refusal. + if p.Modules["a"].BusWaitFrom != nil { + t.Fatalf("a failed module still holds the start of its wait: %v", p.Modules["a"].BusWaitFrom) + } + p.State = inventory.PlanBuilding + askModule(t.Context(), p, "a", byName) + if p.State != inventory.PlanBuilding || p.Modules["a"].State != "" || p.Modules["a"].BusWaitFrom == nil { + t.Fatalf("a retry made while the bus is silent did not wait afresh: %s %+v", p.State, p.Modules["a"]) + } +} + +func TestAPlanWithAModuleWaitingForTheBusIsOneTheLookComesBackTo(t *testing.T) { + p, _ := aTierOfThree() + if busWaiting(*p) { + t.Fatal("a plan whose modules are not asked yet waits for the bus") + } + now := time.Now() + p.Modules["b"] = &inventory.PlanModule{BusWaitFrom: &now} + p.Modules["a"] = &inventory.PlanModule{State: "asked"} + if got := busWaitingIn(*p); len(got) != 1 || got[0] != "b" { + t.Fatalf("the modules waiting for the bus are %v, not b", got) + } + p.State = inventory.PlanFailed + if busWaiting(*p) { + t.Fatal("a plan that is not open is waited for") + } +} + +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..afdfe96e --- /dev/null +++ b/cmd/mesh-controller/ask_bus_wait.go @@ -0,0 +1,87 @@ +package main + +import ( + "errors" + "fmt" + "slices" + "sort" + "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 +} + +// busWaiting says an open plan has a module of its current tier that the bus did not answer an ask of: the look must come back +// for it, as nothing else will when no build of the tier is in flight (issue 1334). +func busWaiting(p inventory.Plan) bool { + return len(busWaitingIn(p)) > 0 +} + +// busWaitingIn are the modules of the plan's current tier left unasked because the bus did not answer, sorted. +func busWaitingIn(p inventory.Plan) []string { + if !p.Open() || p.Tier >= len(p.Tiers) { + return nil + } + var out []string + for _, m := range p.Tiers[p.Tier] { + if s := p.Modules[m]; s != nil && s.State == "" && s.BusWaitFrom != nil { + out = append(out, m) + } + } + sort.Strings(out) + return out +} + +// busWaitWords is how a verb says modules are waiting for the bus: not asked yet, and asked again by the plan at its next look. +func busWaitWords(p inventory.Plan, modules []string) string { + return fmt.Sprintf("%s not asked yet: the bus did not answer; %s asks again at its next look", strings.Join(modules, ", "), p.ID) +} + +// slicesWithout is names without those in drop, in order. +func slicesWithout(names, drop []string) []string { + var out []string + for _, n := range names { + if !slices.Contains(drop, n) { + out = append(out, n) + } + } + return out +} diff --git a/cmd/mesh-controller/plan_retry.go b/cmd/mesh-controller/plan_retry.go index 6e5e99ed..38171760 100644 --- a/cmd/mesh-controller/plan_retry.go +++ b/cmd/mesh-controller/plan_retry.go @@ -406,15 +406,23 @@ func retryPlan(ctx context.Context, open *stores, id string) (string, error) { asked = append(asked, m+" as "+s.Build) } } - resumed(&p, fmt.Sprintf("tier %d retried by hand: %s asked again", p.Tier, strings.Join(failed, ", "))) + waiting := busWaitingIn(p) + resumed(&p, fmt.Sprintf("tier %d retried by hand: %s asked again", p.Tier, strings.Join(slicesWithout(failed, waiting), ", "))) + if len(waiting) > 0 { + p.Note += "; " + busWaitWords(p, waiting) + } if err := inv.SavePlan(ctx, &p); err != nil { return "", err } if p.State != inventory.PlanBuilding { return "", fmt.Errorf("%s could not be resumed: %s", p.ID, p.Note) } - return fmt.Sprintf("%s retried at tier %d of %d: asked %s; the plan goes on from there as any plan does", - p.ID, p.Tier, len(p.Tiers), strings.Join(asked, ", ")), nil + said := fmt.Sprintf("%s retried at tier %d of %d: asked %s; the plan goes on from there as any plan does", + p.ID, p.Tier, len(p.Tiers), strings.Join(asked, ", ")) + if len(waiting) > 0 { + said += "; " + busWaitWords(p, waiting) + } + return said, nil } // joinAPlan asks a module again for the plan that holds it unbuilt or failed, if one does: open plans @@ -460,6 +468,9 @@ func joinAPlan(ctx context.Context, open *stores, module string) (bool, string, if err := inv.SavePlan(ctx, &p); err != nil { return false, "", err } + if s.State == "" && s.BusWaitFrom != nil { + return true, busWaitWords(p, []string{module}), nil + } if s.State != "asked" { return true, "", fmt.Errorf("%s could not be asked for %s: %s", module, p.ID, s.Why) } diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index a6440771..5dea2cea 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -580,12 +580,25 @@ 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. + reason := err.Error() + if busDidNotAnswer(err) { + if waitForTheBus(p, state, name, err, now) { + return + } + reason = fmt.Sprintf("the bus did not answer an ask for %s (since %s): %s", busWaitBound, + state.BusWaitFrom.UTC().Format(time.RFC3339), firstLine(err.Error())) + } + // The wait has ended: a Retry or a rebuild waits for the bus afresh. + state.BusWaitFrom = nil state.State = "failed" - state.Why = err.Error() + state.Why = reason p.State = inventory.PlanFailed - p.Note = fmt.Sprintf("%s could not be asked for: %v", name, err) + p.Note = fmt.Sprintf("%s could not be asked for: %s", name, reason) 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 @@ -810,6 +823,11 @@ func advanceHeld(ctx context.Context, open *stores) (failed bool) { if p.State == inventory.PlanFailed { sayUnsent(p, rollsOut) } + // A module the bus did not answer an ask of is asked again by a look, and no event may bring one (a tier all + // refused has no build in flight): the look fails, so the looker retries it from five seconds (issue 1334). + if busWaiting(*p) { + failed = true + } if planSnapshot(*p) != before { if err := inv.SavePlan(ctx, p); err != nil { fmt.Printf("%s: cannot keep the plan: %v\n", p.ID, err) @@ -887,10 +905,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.