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_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)
+		}
+	}
+}

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"])
+	}
+}

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
+}

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)
 	}

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

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.