Keep a registration's two forge waits within one bound, and read an unreadable address once per five seconds (issue 1277)

cmd/mesh-controller/build_source.go

@@ -75,13 +75,19 @@ func markServed() {
 type sourceForms struct {
 	bases map[string]string // the seat's clone base, `scheme://host:port`, by seat; "" where it is not known
 	base  func(seat string) string
-	read  func(seat string) (string, error) // base with the reason it is unknown; nil where only base is given
+	// read is base with the reason it is unknown; nil where only base is given. again forces a fresh read past a
+	// failure kept for failedKept.
+	read func(ctx context.Context, seat string, again bool) (string, error)
 }
 
+// failedKept is how long an unreadable base is not read again for the callers that loop over every catalogue
+// entry: one composition of the mesh serves them all (novox/hq issue 1277).
+const failedKept = 5 * time.Second
+
 // newSourceForms reads the seats' bases from the mesh when first needed. A base that could not be read is not
 // kept as if it were known: it is read again at the next ask (novox/hq issue 1277).
 func newSourceForms(ctx context.Context, inv *inventory.Inventory) *sourceForms {
-	return sourceFormsRead(func() (catalogue.World, error) {
+	return sourceFormsRead(ctx, time.Now, func(ctx context.Context) (catalogue.World, error) {
 		shelf, err := inv.Catalogue(ctx)
 		if err != nil {
 			return catalogue.World{}, fmt.Errorf("the catalogue cannot be read: %w", err)
@@ -95,30 +101,32 @@ func newSourceForms(ctx context.Context, inv *inventory.Inventory) *sourceForms
 }
 
 // sourceFormsRead is newSourceForms on how the mesh is read, handed in.
-func sourceFormsRead(readWorld func() (catalogue.World, error)) *sourceForms {
+func sourceFormsRead(ctx context.Context, now func() time.Time,
+	readWorld func(context.Context) (catalogue.World, error)) *sourceForms {
 	f := &sourceForms{bases: map[string]string{}}
-	var world *catalogue.World
-	f.read = func(seat string) (string, error) {
+	var failedAt time.Time // when the last read failed; zero when it did not
+	var failure error
+	f.read = func(ctx context.Context, seat string, again bool) (string, error) {
 		if b, known := f.bases[seat]; known {
 			return b, nil
 		}
-		if world == nil {
-			read, err := readWorld()
-			if err != nil {
-				return "", err
+		if failure != nil && !again && now().Sub(failedAt) < failedKept {
+			return "", failure
+		}
+		world, err := readWorld(ctx)
+		if err == nil {
+			var b string
+			if b, err = seatBase(world, seat); err == nil {
+				failure = nil
+				f.bases[seat] = b
+				return b, nil
 			}
-			world = &read
 		}
-		b, err := seatBase(*world, seat)
-		if err != nil {
-			world = nil // the holder may be changing now: read the mesh again at the next ask
-			return "", err
-		}
-		f.bases[seat] = b
-		return b, nil
+		failedAt, failure = now(), err // the holder may be changing now: only an unread base is kept, for a moment
+		return "", err
 	}
 	f.base = func(seat string) string {
-		b, _ := f.read(seat)
+		b, _ := f.read(ctx, seat, false)
 		return b
 	}
 	return f
@@ -137,7 +145,7 @@ func (f *sourceForms) forgeAddress(ctx context.Context, w waitOut) (string, erro
 	}
 	began := w.now()
 	for {
-		b, err := f.read(gitSeat)
+		b, err := f.read(ctx, gitSeat, true)
 		if err == nil {
 			return b, nil
 		}
@@ -207,11 +215,29 @@ type forgeFacts struct {
 // protection, waiting out a forge not served for a moment (novox/hq issue 486). A variable so a test needs no
 // forge.
 var askTheForge = func(ctx context.Context, owner, repo, branch string) (forgeFacts, error) {
-	return forgeWaitedOut(ctx, forgeWait, owner+"/"+repo, func(ctx context.Context) (forgeFacts, error) {
+	return askedWithinBudget(ctx, owner+"/"+repo, func(ctx context.Context) (forgeFacts, error) {
 		return askTheForgeOnce(ctx, owner, repo, branch)
 	})
 }
 
+type forgeBudgetKey struct{}
+
+// withForgeBudget says how much of the registration's one wait is left for the forge's answer: the wait for its
+// address has used the rest (novox/hq issue 1277). The registration holds the controller's one receive loop, so the
+// two waits together keep to one bound.
+func withForgeBudget(ctx context.Context, left time.Duration) context.Context {
+	return context.WithValue(ctx, forgeBudgetKey{}, max(left, 0))
+}
+
+// askedWithinBudget is forgeWaitedOut with forgeWait's bound, or what is left of it when the address was waited for.
+func askedWithinBudget(ctx context.Context, repository string, ask func(context.Context) (forgeFacts, error)) (forgeFacts, error) {
+	w := forgeWait
+	if left, ok := ctx.Value(forgeBudgetKey{}).(time.Duration); ok {
+		w.bound = min(w.bound, left)
+	}
+	return forgeWaitedOut(ctx, w, repository, ask)
+}
+
 // waitOut bounds how long a question is asked again while its answer is "not now": a pause of first, doubling
 // to most, for at most bound in all. now and sleep are the clock, so a test waits in no real time.
 type waitOut struct {
@@ -532,7 +558,12 @@ func mayRegisterIn(ctx context.Context, inv registrations, forms *sourceForms, m
 	if built.Seat == "" {
 		// A URL is judged against the forge's address: an address that cannot be read is no proof the source is
 		// elsewhere (novox/hq issue 1277).
-		if _, err := forms.forgeAddress(ctx, forgeAddressWait); err != nil {
+		began := forgeAddressWait.now()
+		addressCtx, cancel := context.WithTimeout(ctx, forgeAddressWait.bound)
+		_, err := forms.forgeAddress(addressCtx, forgeAddressWait)
+		cancel()
+		ctx = withForgeBudget(ctx, forgeAddressWait.bound-forgeAddressWait.now().Sub(began))
+		if err != nil {
 			if terminal {
 				fmt.Printf("%s: the forge's address could not be read (%v) — registered, as asked at the controller's "+
 					"terminal\n", buildID, err)

cmd/mesh-controller/forge_address_test.go

@@ -13,9 +13,9 @@ import (
 
 // A world reader that fails the first `failing` reads, as the mesh does while the forge's seat holder changes in
 // the plan's tier (novox/hq issue 1277), then answers with the forge held.
-func aWorldFailing(failing int) (func() (catalogue.World, error), *int) {
+func aWorldFailing(failing int) (func(context.Context) (catalogue.World, error), *int) {
 	reads := 0
-	return func() (catalogue.World, error) {
+	return func(context.Context) (catalogue.World, error) {
 		reads++
 		if reads <= failing {
 			return catalogue.World{}, errors.New("the mesh cannot be read: nothing serves")
@@ -29,10 +29,12 @@ const aRepoOnTheForge = "http://anchor.internal:20000/novox/mesh-delivery.git"
 // 1277/AC1: a base that could not be read is not remembered as known.
 func TestAnUnreadForgeAddressIsNotKeptAsKnown(t *testing.T) {
 	read, _ := aWorldFailing(1)
-	forms := sourceFormsRead(read)
+	clock := time.Now()
+	forms := sourceFormsRead(context.Background(), func() time.Time { return clock }, read)
 	if _, _, ok := forms.onTheForge(aRepoOnTheForge, ""); ok {
 		t.Fatal("1277/AC1: the source was on the forge with the address unread")
 	}
+	clock = clock.Add(failedKept) // the failure is kept only for a moment, not as a known empty base
 	owner, name, ok := forms.onTheForge(aRepoOnTheForge, "")
 	if !ok || owner != "novox" || name != "mesh-delivery" {
 		t.Fatalf("1277/AC1: the second ask answered %q %q %v, not the owner and name: the empty base was kept", owner, name, ok)
@@ -69,7 +71,7 @@ func registerFrom(t *testing.T, forms *sourceForms) (int64, error) {
 func TestAForgeAddressUnreadForAMomentIsWaitedOut(t *testing.T) {
 	paused := withFakeAddressWait(t)
 	read, reads := aWorldFailing(2)
-	id, err := registerFrom(t, sourceFormsRead(read))
+	id, err := registerFrom(t, sourceFormsRead(context.Background(), time.Now, read))
 	if err != nil || id == 0 {
 		t.Fatalf("1277/AC2: the registration was refused (id %d): %v", id, err)
 	}
@@ -82,7 +84,7 @@ func TestAForgeAddressUnreadForAMomentIsWaitedOut(t *testing.T) {
 func TestAForgeAddressUnreadForTheWholeBoundIsSaidSo(t *testing.T) {
 	paused := withFakeAddressWait(t)
 	read, _ := aWorldFailing(1 << 30)
-	_, err := registerFrom(t, sourceFormsRead(read))
+	_, err := registerFrom(t, sourceFormsRead(context.Background(), time.Now, read))
 	if !errors.Is(err, errNotItsSource) {
 		t.Fatalf("1277/AC3: not refused: %v", err)
 	}
@@ -109,9 +111,50 @@ func TestAKnownAddressAndAnotherHostIsStillNotOnTheForge(t *testing.T) {
 	read, _ := aWorldFailing(0)
 	const elsewhere = "http://elsewhere.example/x/y.git"
 	mesh := &aMeshRegisteredFromTheURL{anEmptyMesh{asked: true, request: inventory.BuildRequest{ID: "build-1", For: "build"}}, elsewhere}
-	_, err := mayRegisterIn(t.Context(), mesh, sourceFormsRead(read), "mesh-delivery-view",
+	_, err := mayRegisterIn(t.Context(), mesh, sourceFormsRead(context.Background(), time.Now, read), "mesh-delivery-view",
 		inventory.Source{Repository: elsewhere}, "build-1", "", "main")
 	if !errors.Is(err, errNotItsSource) || !strings.Contains(err.Error(), "is not on the mesh's forge") {
 		t.Fatalf("1277/AC4: the refusal changed: %v", err)
 	}
 }
+
+// The registration holds the controller's one receive loop: the wait for the forge's address and the wait for its
+// answer together keep to one bound, not one each (novox/hq issue 1277, review F1).
+func TestTheTwoWaitsOfARegistrationKeepToOneBound(t *testing.T) {
+	w, paused := aFakeWait(60 * time.Second)
+	address, answer := w, w
+	address.first, address.most = 5*time.Second, 5*time.Second
+	wasAddress, wasWait, wasAsk := forgeAddressWait, forgeWait, askTheForge
+	forgeAddressWait, forgeWait = address, answer
+	askTheForge = func(ctx context.Context, owner, repo, _ string) (forgeFacts, error) {
+		return askedWithinBudget(ctx, owner+"/"+repo, func(context.Context) (forgeFacts, error) {
+			return forgeFacts{}, notServed()
+		})
+	}
+	t.Cleanup(func() { forgeAddressWait, forgeWait, askTheForge = wasAddress, wasWait, wasAsk })
+	read, _ := aWorldFailing(6) // the address comes back after 30 s of waiting; the forge then never answers
+	_, err := registerFrom(t, sourceFormsRead(context.Background(), time.Now, read))
+	if !errors.Is(err, errNotItsSource) {
+		t.Fatalf("not refused: %v", err)
+	}
+	var total time.Duration
+	for _, d := range *paused {
+		total += d
+	}
+	if total > 60*time.Second {
+		t.Fatalf("1277 review F1: the registration waited %s in all, past the one 60 s bound", total)
+	}
+}
+
+// An unreadable address is kept for a moment, so the callers that loop over every catalogue entry compose the mesh
+// once, not once each (novox/hq issue 1277, review F2).
+func TestAnUnreadableAddressIsComposedOnceForACatalogueLoop(t *testing.T) {
+	read, reads := aWorldFailing(1 << 30)
+	forms := sourceFormsRead(context.Background(), time.Now, read)
+	for range 180 {
+		forms.base(gitSeat)
+	}
+	if *reads != 1 {
+		t.Fatalf("1277 review F2: the mesh was composed %d times for 180 asks of the address, want once", *reads)
+	}
+}