novox/mesh-controller
17
0
Fork 0
Code Issues Pull Requests 3 Releases Wiki Activity

Merge pull request 'Read the forge's address again before calling a source not on the forge, within one bound (issue 1277)' (#411) from fix/1277-forge-re-serve into main

Browse Source
This commit was merged in pull request #411.
This commit is contained in:
mesh-tools
2026-10-20 22:34:18 +00:00
parent f29d193ddb 2b1eeb7cd5
commit f4584e6fa2
2 changed files with 275 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/build_source.go
+115 -20
View File
@@ -75,35 +75,94 @@ 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 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)
}
// newSourceForms reads the seats' bases from the mesh once, when first needed.
// 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 {
f := &sourceForms{bases: map[string]string{}}
var world *catalogue.World
f.base = func(seat string) string {
if b, known := f.bases[seat]; known {
return b
}
if world == nil {
w := catalogue.World{}
if shelf, err := inv.Catalogue(ctx); err == nil {
if read, err := restOfTheMeshOf(ctx, inv, shelf, ""); err == nil {
w = read
}
}
world = &w
}
b, err := seatBase(*world, seat)
return sourceFormsRead(ctx, time.Now, func(ctx context.Context) (catalogue.World, error) {
shelf, err := inv.Catalogue(ctx)
if err != nil {
b = ""
return catalogue.World{}, fmt.Errorf("the catalogue cannot be read: %w", err)
}
f.bases[seat] = b
read, err := restOfTheMeshOf(ctx, inv, shelf, "")
if err != nil {
return catalogue.World{}, fmt.Errorf("the mesh cannot be read: %w", err)
}
return read, nil
})
}
// sourceFormsRead is newSourceForms on how the mesh is read, handed in.
func sourceFormsRead(ctx context.Context, now func() time.Time,
readWorld func(context.Context) (catalogue.World, error)) *sourceForms {
f := &sourceForms{bases: map[string]string{}}
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 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
}
}
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(ctx, seat, false)
return b
}
return f
}
// forgeAddressWait is the registration's wait for the forge's address (novox/hq issue 1277): a fresh read every
// 5 s for at most 60 s, as the wait for the forge's answer (issue 486). A variable so a test waits in no real time.
var forgeAddressWait = waitOut{bound: 60 * time.Second, first: 5 * time.Second, most: 5 * time.Second,
now: time.Now, sleep: sleepOrDone}
// forgeAddress is the git seat's base, read again within w's bound while it cannot be read. Where the forms have
// no reader of their own (a test's), base says it.
func (f *sourceForms) forgeAddress(ctx context.Context, w waitOut) (string, error) {
if f.read == nil {
return f.base(gitSeat), nil
}
began := w.now()
for {
b, err := f.read(ctx, gitSeat, true)
if err == nil {
return b, nil
}
if ctx.Err() != nil {
return "", err
}
if w.now().Sub(began)+w.first > w.bound {
return "", fmt.Errorf("after %s: %w", w.now().Sub(began).Round(time.Second), err)
}
fmt.Printf("the forge's address cannot be read now (%v); reading it again in %s, for at most %s in all "+
"(novox/hq issue 1277)\n", err, w.first, w.bound)
if serr := w.sleep(ctx, w.first); serr != nil {
return "", err
}
}
}
// canonical is one spelling of a repository: lower case, no `.git`, no trailing slash, and a seat's path as
// the URL its holder serves it at where that is known.
func (f *sourceForms) canonical(repository, seat string) string {
@@ -156,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 {
@@ -478,6 +555,24 @@ func mayRegisterIn(ctx context.Context, inv registrations, forms *sourceForms, m
return refuse(why)
}
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).
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)
return 0, nil
}
return refuse(fmt.Sprintf("the forge's address could not be read (%v), so whether %s's trunk is "+
"protected cannot be read", err, sourceWords(built.Repository, built.Seat)))
}
}
owner, name, onForge := forms.onTheForge(built.Repository, built.Seat)
if !onForge {
if terminal {
cmd/mesh-controller/forge_address_test.go
+160
View File
@@ -0,0 +1,160 @@
package main
import (
"context"
"errors"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// 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(context.Context) (catalogue.World, error), *int) {
reads := 0
return func(context.Context) (catalogue.World, error) {
reads++
if reads <= failing {
return catalogue.World{}, errors.New("the mesh cannot be read: nothing serves")
}
return forgeHolding(float64(20000)), nil
}, &reads
}
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)
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)
}
}
type aMeshRegisteredFromTheURL struct {
anEmptyMesh
from string
}
func (m *aMeshRegisteredFromTheURL) SourceOf(context.Context, string) (inventory.Source, error) {
return inventory.Source{Repository: m.from}, nil
}
// The wait on a clock of its own, in place of the registration's.
func withFakeAddressWait(t *testing.T) *[]time.Duration {
w, paused := aFakeWait(60 * time.Second)
w.first, w.most = 5*time.Second, 5*time.Second
was := forgeAddressWait
forgeAddressWait = w
t.Cleanup(func() { forgeAddressWait = was })
return paused
}
func registerFrom(t *testing.T, forms *sourceForms) (int64, error) {
mesh := &aMeshRegisteredFromTheURL{anEmptyMesh{asked: true,
request: inventory.BuildRequest{ID: "build-1", Repository: aRepoOnTheForge, For: "build"}}, aRepoOnTheForge}
return mayRegisterIn(t.Context(), mesh, forms, "mesh-delivery-view", inventory.Source{Repository: aRepoOnTheForge},
"build-1", "", "main")
}
// 1277/AC2: an address unread twice, then read, is waited out and the forge is asked.
func TestAForgeAddressUnreadForAMomentIsWaitedOut(t *testing.T) {
paused := withFakeAddressWait(t)
read, reads := aWorldFailing(2)
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)
}
if *reads != 3 || len(*paused) != 2 {
t.Fatalf("1277/AC2: %d reads and %d pauses, want 3 and 2", *reads, len(*paused))
}
}
// 1277/AC3: an address unread for the whole bound is said so, with the read's error.
func TestAForgeAddressUnreadForTheWholeBoundIsSaidSo(t *testing.T) {
paused := withFakeAddressWait(t)
read, _ := aWorldFailing(1 << 30)
_, err := registerFrom(t, sourceFormsRead(context.Background(), time.Now, read))
if !errors.Is(err, errNotItsSource) {
t.Fatalf("1277/AC3: not refused: %v", err)
}
for _, want := range []string{"the forge's address could not be read", "nothing serves"} {
if !strings.Contains(err.Error(), want) {
t.Errorf("1277/AC3: the refusal lacks %q: %v", want, err)
}
}
if strings.Contains(err.Error(), "not on the mesh's forge") {
t.Errorf("1277/AC3: the refusal says the source is not on the forge: %v", err)
}
var total time.Duration
for _, d := range *paused {
total += d
}
if total > 60*time.Second || total < 50*time.Second {
t.Errorf("1277/AC3: waited %s, want about the 60 s bound", total)
}
}
// 1277/AC4: a known address and a URL on another host is still not on the forge.
func TestAKnownAddressAndAnotherHostIsStillNotOnTheForge(t *testing.T) {
withFakeAddressWait(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(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)
}
}