From 2b1eeb7cd53edc106854c74a0ef87a3578196b38 Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 22:03:55 +0000 Subject: [PATCH] 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 | 73 ++++++++++++++++------- cmd/mesh-controller/forge_address_test.go | 55 +++++++++++++++-- 2 files changed, 101 insertions(+), 27 deletions(-) diff --git a/cmd/mesh-controller/build_source.go b/cmd/mesh-controller/build_source.go index e3fff9f2..a58292cd 100644 --- a/cmd/mesh-controller/build_source.go +++ b/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) diff --git a/cmd/mesh-controller/forge_address_test.go b/cmd/mesh-controller/forge_address_test.go index c901ea35..ea50d84f 100644 --- a/cmd/mesh-controller/forge_address_test.go +++ b/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) + } +}