diff --git a/cmd/mesh-controller/build_source.go b/cmd/mesh-controller/build_source.go index 6cc7685f..dc940c98 100644 --- a/cmd/mesh-controller/build_source.go +++ b/cmd/mesh-controller/build_source.go @@ -153,8 +153,81 @@ type forgeFacts struct { } // askTheForge asks the forge, through its module's tools on the bus, for a repository's id and its branch's -// protection. A variable so a test needs no forge. +// 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 askTheForgeOnce(ctx, owner, repo, branch) + }) +} + +// 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 { + bound, first, most time.Duration + now func() time.Time + sleep func(context.Context, time.Duration) error +} + +// forgeWait is the registration's wait for the forge (novox/hq issue 486). On 2026-10-11 a build's registration +// asked the forge while node-tools on the control-node re-served after a rollout of the tool runner, heard +// "nothing serves gitea.gitea_api", and failed its plan; the forge answered again seconds later. A re-serve takes +// seconds, so 90 s waits one out with room to spare, and stays short because the registration holds the +// controller's receive loop while it waits. The pauses are those the build machine waits the registry with +// (novox/hq issue 457). +var forgeWait = waitOut{bound: 90 * time.Second, first: time.Second, most: 10 * time.Second, + now: time.Now, sleep: sleepOrDone} + +// sleepOrDone pauses for d, or less when ctx ends first. +func sleepOrDone(ctx context.Context, d time.Duration) error { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-t.C: + return nil + } +} + +// forgeNotNow is an answer that says the forge's tool is not there now, not that the forge refused: nothing +// serves it (the tool runner re-serving), or it did not answer within the wait. Anything else — a refusal, an +// unreadable answer, a repository with no id — is the forge's word, and is never asked again. +func forgeNotNow(err error) bool { + return errors.Is(err, link.ErrNothingServes) || errors.Is(err, link.ErrNoAnswer) +} + +// forgeWaitedOut asks the forge about repository, asking again within w's bound while it answers "not now" +// (novox/hq issue 486). Past the bound it fails, saying it waited, for how long, and what the forge last said. +func forgeWaitedOut(ctx context.Context, w waitOut, repository string, + ask func(context.Context) (forgeFacts, error)) (forgeFacts, error) { + began := w.now() + pause := w.first + for asked := 1; ; asked++ { + facts, err := ask(ctx) + if err == nil || !forgeNotNow(err) || ctx.Err() != nil { + if err == nil && asked > 1 { + fmt.Printf("the forge answered about %s again, after %s (novox/hq issue 486)\n", repository, + w.now().Sub(began).Round(time.Second)) + } + return facts, err + } + waited := w.now().Sub(began) + if waited+pause > w.bound { + return forgeFacts{}, fmt.Errorf("waited %s for the forge to answer about %s, asking %d times, and it "+ + "did not (novox/hq issue 486): %w", waited.Round(time.Second), repository, asked, err) + } + fmt.Printf("the forge cannot be asked about %s now (%v); asking again in %s, for at most %s in all "+ + "(novox/hq issue 486)\n", repository, err, pause, w.bound) + if err := w.sleep(ctx, pause); err != nil { + return forgeFacts{}, fmt.Errorf("stopped waiting for the forge to answer about %s: %w", repository, err) + } + pause = min(2*pause, w.most) + } +} + +// askTheForgeOnce asks the forge once, through its module's tools on the bus. +func askTheForgeOnce(ctx context.Context, owner, repo, branch string) (forgeFacts, error) { js, err := aBus() if err != nil { return forgeFacts{}, err diff --git a/cmd/mesh-controller/forge_wait_test.go b/cmd/mesh-controller/forge_wait_test.go new file mode 100644 index 00000000..3b977fed --- /dev/null +++ b/cmd/mesh-controller/forge_wait_test.go @@ -0,0 +1,123 @@ +package main + +import ( + "context" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/link" +) + +// A wait on a clock of its own: each pause moves it on, and nothing waits in real time. +func aFakeWait(bound time.Duration) (waitOut, *[]time.Duration) { + clock := time.Date(2026, 10, 11, 11, 52, 2, 0, time.UTC) + var paused []time.Duration + return waitOut{bound: bound, first: time.Second, most: 10 * time.Second, + now: func() time.Time { return clock }, + sleep: func(_ context.Context, d time.Duration) error { + paused = append(paused, d) + clock = clock.Add(d) + return nil + }}, &paused +} + +// notServed is what the bus says while node-tools re-serves, in the words seen on 2026-10-11. +func notServed() error { + return fmt.Errorf("cannot ask gitea.gitea_api: %w", link.ErrNothingServes) +} + +// A registration's forge question that lands while the tool runner re-serves is asked again, and the forge's +// answer, seconds later, is the registration's (novox/hq issue 486). +func TestTheForgeNotServedForAMomentIsWaitedOut(t *testing.T) { + w, paused := aFakeWait(90 * time.Second) + asked := 0 + facts, err := forgeWaitedOut(t.Context(), w, "novox/mesh-catalog", func(context.Context) (forgeFacts, error) { + asked++ + if asked <= 2 { + return forgeFacts{}, notServed() + } + return forgeFacts{ID: 42, Guarded: true}, nil + }) + if err != nil { + t.Fatalf("a forge not served for two asks failed the registration: %v", err) + } + if facts.ID != 42 || !facts.Guarded || asked != 3 { + t.Fatalf("the forge's answer was %+v after %d asks, want its answer on the third", facts, asked) + } + if want := []time.Duration{time.Second, 2 * time.Second}; fmt.Sprint(*paused) != fmt.Sprint(want) { + t.Fatalf("paused %v, want %v", *paused, want) + } +} + +// A tool that did not answer within its wait is waited out the same way: nothing refused. +func TestTheForgeNotAnsweringInTimeIsAskedAgain(t *testing.T) { + w, _ := aFakeWait(90 * time.Second) + asked := 0 + _, err := forgeWaitedOut(t.Context(), w, "novox/x", func(context.Context) (forgeFacts, error) { + asked++ + if asked == 1 { + return forgeFacts{}, fmt.Errorf("gitea did not answer: %w", link.ErrNoAnswer) + } + return forgeFacts{ID: 1, Guarded: true}, nil + }) + if err != nil || asked != 2 { + t.Fatalf("asked %d times: %v", asked, err) + } +} + +// Past the bound the registration fails, saying it waited, for how long, and what the forge last said. +func TestTheForgeNotServedPastTheBoundFailsSayingItWaited(t *testing.T) { + w, paused := aFakeWait(90 * time.Second) + _, err := forgeWaitedOut(t.Context(), w, "novox/mesh-catalog", func(context.Context) (forgeFacts, error) { + return forgeFacts{}, notServed() + }) + if err == nil { + t.Fatal("a forge never served was read as an answer") + } + var waited time.Duration + for _, p := range *paused { + waited += p + } + if waited > 90*time.Second || waited < 60*time.Second { + t.Fatalf("waited %s in all, want close to and within the bound of 90s", waited) + } + for _, want := range []string{"waited " + waited.String(), "novox/mesh-catalog", "nothing serves", "issue 486"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("the failure does not say %q: %v", want, err) + } + } + if !errors.Is(err, link.ErrNothingServes) { + t.Errorf("the failure lost what the forge last said: %v", err) + } +} + +// A refusal is the forge's word: never asked again. +func TestARefusalFromTheForgeIsNeverAskedAgain(t *testing.T) { + w, paused := aFakeWait(90 * time.Second) + asked := 0 + _, err := forgeWaitedOut(t.Context(), w, "novox/x", func(context.Context) (forgeFacts, error) { + asked++ + return forgeFacts{}, errors.New("gitea.gitea_api: 404 Not Found") + }) + if err == nil || asked != 1 || len(*paused) != 0 { + t.Fatalf("a refusal was asked %d times, paused %v: %v", asked, *paused, err) + } +} + +// A registration whose own context ends stops waiting at once. +func TestAWaitForTheForgeEndsWithItsContext(t *testing.T) { + w, _ := aFakeWait(90 * time.Second) + ctx, cancel := context.WithCancel(t.Context()) + w.sleep = func(ctx context.Context, d time.Duration) error { cancel(); return sleepOrDone(ctx, d) } + asked := 0 + _, err := forgeWaitedOut(ctx, w, "novox/x", func(context.Context) (forgeFacts, error) { + asked++ + return forgeFacts{}, notServed() + }) + if !errors.Is(err, context.Canceled) || asked != 1 { + t.Fatalf("asked %d times after the context ended: %v", asked, err) + } +} diff --git a/internal/link/ask.go b/internal/link/ask.go index 41adda6e..b3c561bf 100644 --- a/internal/link/ask.go +++ b/internal/link/ask.go @@ -16,6 +16,19 @@ type Answer struct { Error string `json:"error,omitempty"` } +// ErrNoAnswer is a tool asked that did not answer within the asker's wait: nothing refused, and asking again +// may be answered. With ErrNothingServes, what an asker may wait out (novox/hq issue 486). +var ErrNoAnswer = errors.New("no answer within the wait") + +// saidAs is an error said in its own words that is still the error it is a case of, for errors.Is. +type saidAs struct { + text string + is error +} + +func (e saidAs) Error() string { return e.text } +func (e saidAs) Unwrap() error { return e.is } + // Ask calls one of a module's tools over the broker and waits for its answer. // // **The control plane is the way in** (novox/hq 04-ISSUES/049, ADR 0095). A module's broker @@ -37,16 +50,16 @@ func Ask(ctx context.Context, bus Bus, module, tool string, args json.RawMessage reply, err := bus.AskTool(ctx, module, tool, args, timeout) if err != nil { if errors.Is(err, nats.ErrNoResponders) { - return Answer{}, fmt.Errorf( + return Answer{}, saidAs{fmt.Sprintf( "nothing serves %s: no runtime has bound %q on the bus. The module is not "+ "assigned, its runtime is not up, or it serves no such tool — `status` says "+ - "whether the machine carrying it has applied", module, key) + "whether the machine carrying it has applied", module, key), ErrNothingServes} } if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) { - return Answer{}, fmt.Errorf( + return Answer{}, saidAs{fmt.Sprintf( "%s did not answer within %s. Its runtime serves %q when it is up and has bound "+ "the bus — `status` says whether the machine carrying it has applied", - module, timeout, key) + module, timeout, key), ErrNoAnswer} } return Answer{}, fmt.Errorf("cannot ask %s: %w", key, err) } diff --git a/internal/link/bus.go b/internal/link/bus.go index e2e227d9..6d94cada 100644 --- a/internal/link/bus.go +++ b/internal/link/bus.go @@ -206,8 +206,9 @@ func (b OverNATS) AskTool(ctx context.Context, module, tool string, args []byte, if err != nil { if errors.Is(err, nats.ErrNoResponders) { // Said at once rather than after the whole wait: nothing is subscribed to that - // subject, which is a different fact from a tool being slow. - return nil, fmt.Errorf("nothing serves %s.%s", module, tool) + // subject, which is a different fact from a tool being slow. Said as ErrNothingServes, so an + // asker can tell a tool not served now from one that refused (novox/hq issue 486). + return nil, saidAs{fmt.Sprintf("nothing serves %s.%s", module, tool), ErrNothingServes} } return nil, fmt.Errorf("asking %s.%s: %w", module, tool, err) } diff --git a/internal/link/bus_test.go b/internal/link/bus_test.go index 15a3bc7f..a8fa0745 100644 --- a/internal/link/bus_test.go +++ b/internal/link/bus_test.go @@ -3,6 +3,7 @@ package link import ( "context" "encoding/json" + "errors" "strings" "testing" "time" @@ -115,6 +116,15 @@ func TestAskingAToolNobodyServesFailsAtOnce(t *testing.T) { if !strings.Contains(err.Error(), "nothing serves") { t.Errorf("the refusal does not say nobody is there: %v", err) } + // And it is ErrNothingServes, through Ask too, so an asker can wait out a tool runner re-serving and + // never a refusal (novox/hq issue 486). + if !errors.Is(err, ErrNothingServes) { + t.Errorf("nothing serving the tool is not ErrNothingServes: %v", err) + } + if _, err := Ask(context.Background(), bus, "nobody", "home", nil, 30*time.Second); !errors.Is(err, ErrNothingServes) || + !strings.Contains(err.Error(), "nothing serves nobody.home") { + t.Errorf("Ask does not say ErrNothingServes in its words: %v", err) + } } // And a served tool answers, with no reply queue to declare and no correlation to check.