Wait out a forge not served for a moment when registering a build (novox/hq issue 486)
A registration asked the forge while node-tools re-served after a tool-runner rollout, heard "nothing serves gitea.gitea_api", and failed a good build's plan. The bus's not-served answer is now link.ErrNothingServes (and a timeout link.ErrNoAnswer), and the registration asks again for up to 90 s on those alone, never on a refusal; past the bound it fails saying how long it waited.
This commit is contained in:
@@ -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
|
// 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) {
|
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()
|
js, err := aBus()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return forgeFacts{}, err
|
return forgeFacts{}, err
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
+17
-4
@@ -16,6 +16,19 @@ type Answer struct {
|
|||||||
Error string `json:"error,omitempty"`
|
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.
|
// 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
|
// **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)
|
reply, err := bus.AskTool(ctx, module, tool, args, timeout)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, nats.ErrNoResponders) {
|
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 "+
|
"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 "+
|
"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) {
|
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 "+
|
"%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",
|
"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)
|
return Answer{}, fmt.Errorf("cannot ask %s: %w", key, err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -206,8 +206,9 @@ func (b OverNATS) AskTool(ctx context.Context, module, tool string, args []byte,
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, nats.ErrNoResponders) {
|
if errors.Is(err, nats.ErrNoResponders) {
|
||||||
// Said at once rather than after the whole wait: nothing is subscribed to that
|
// 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.
|
// subject, which is a different fact from a tool being slow. Said as ErrNothingServes, so an
|
||||||
return nil, fmt.Errorf("nothing serves %s.%s", module, tool)
|
// 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)
|
return nil, fmt.Errorf("asking %s.%s: %w", module, tool, err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package link
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -115,6 +116,15 @@ func TestAskingAToolNobodyServesFailsAtOnce(t *testing.T) {
|
|||||||
if !strings.Contains(err.Error(), "nothing serves") {
|
if !strings.Contains(err.Error(), "nothing serves") {
|
||||||
t.Errorf("the refusal does not say nobody is there: %v", err)
|
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.
|
// And a served tool answers, with no reply queue to declare and no correlation to check.
|
||||||
|
|||||||
Reference in New Issue
Block a user