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

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

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

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

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.