Cap the forge wait as a whole, and wait out only a forge not served (novox/hq issue 486)
The registration holds the controller's one receive loop. Checking the bound only between asks let an ask in flight hold it near two minutes, the deaf-controller limit. The whole wait now runs under a 60 s deadline, still says it waited when that deadline ends it, and a tool that did not answer in time is not asked again: it has already held the loop its whole wait.
This commit is contained in:
@@ -172,10 +172,10 @@ type waitOut struct {
|
|||||||
// forgeWait is the registration's wait for the forge (novox/hq issue 486). On 2026-10-11 a build's registration
|
// 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
|
// 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
|
// "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
|
// about 10 s, so 60 s waits one out with room to spare. It is a cap on the whole wait, an ask in flight included:
|
||||||
// controller's receive loop while it waits. The pauses are those the build machine waits the registry with
|
// the registration holds the controller's one receive loop while it waits, and a loop held two minutes reads as
|
||||||
// (novox/hq issue 457).
|
// deaf. 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,
|
var forgeWait = waitOut{bound: 60 * time.Second, first: time.Second, most: 10 * time.Second,
|
||||||
now: time.Now, sleep: sleepOrDone}
|
now: time.Now, sleep: sleepOrDone}
|
||||||
|
|
||||||
// sleepOrDone pauses for d, or less when ctx ends first.
|
// sleepOrDone pauses for d, or less when ctx ends first.
|
||||||
@@ -190,37 +190,53 @@ func sleepOrDone(ctx context.Context, d time.Duration) error {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// forgeNotNow is an answer that says the forge's tool is not there now, not that the forge refused: nothing
|
// forgeNotNow is an answer that says the forge's tool is not served now — the tool runner re-serving — not that
|
||||||
// serves it (the tool runner re-serving), or it did not answer within the wait. Anything else — a refusal, an
|
// the forge refused. Only that is asked again: a refusal, an unreadable answer or a repository with no id is the
|
||||||
// unreadable answer, a repository with no id — is the forge's word, and is never asked again.
|
// forge's word, and a tool that did not answer in time has already held the receive loop its whole wait.
|
||||||
func forgeNotNow(err error) bool {
|
func forgeNotNow(err error) bool { return errors.Is(err, link.ErrNothingServes) }
|
||||||
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"
|
// forgeWaitedOut asks the forge about repository, asking again while it answers "not served now", all of it
|
||||||
// (novox/hq issue 486). Past the bound it fails, saying it waited, for how long, and what the forge last said.
|
// within w's bound, an ask in flight included (novox/hq issue 486). Past the bound it fails, saying it waited,
|
||||||
|
// for how long, and what the forge last said; when the caller's own context ends, it says that instead.
|
||||||
func forgeWaitedOut(ctx context.Context, w waitOut, repository string,
|
func forgeWaitedOut(ctx context.Context, w waitOut, repository string,
|
||||||
ask func(context.Context) (forgeFacts, error)) (forgeFacts, error) {
|
ask func(context.Context) (forgeFacts, error)) (forgeFacts, error) {
|
||||||
|
capped, cancel := context.WithTimeout(ctx, w.bound)
|
||||||
|
defer cancel()
|
||||||
began := w.now()
|
began := w.now()
|
||||||
pause := w.first
|
pause := w.first
|
||||||
for asked := 1; ; asked++ {
|
for asked := 1; ; asked++ {
|
||||||
facts, err := ask(ctx)
|
facts, err := ask(capped)
|
||||||
if err == nil || !forgeNotNow(err) || ctx.Err() != nil {
|
if err == nil {
|
||||||
if err == nil && asked > 1 {
|
if asked > 1 {
|
||||||
fmt.Printf("the forge answered about %s again, after %s (novox/hq issue 486)\n", repository,
|
fmt.Printf("the forge answered about %s again, after %s (novox/hq issue 486)\n", repository,
|
||||||
w.now().Sub(began).Round(time.Second))
|
w.now().Sub(began).Round(time.Second))
|
||||||
}
|
}
|
||||||
return facts, err
|
return facts, nil
|
||||||
}
|
}
|
||||||
waited := w.now().Sub(began)
|
if ctx.Err() != nil {
|
||||||
if waited+pause > w.bound {
|
return forgeFacts{}, err // the caller's own context ended: no wait of ours to speak of
|
||||||
|
}
|
||||||
|
gaveUp := func() (forgeFacts, error) {
|
||||||
return forgeFacts{}, fmt.Errorf("waited %s for the forge to answer about %s, asking %d times, and it "+
|
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)
|
"did not (novox/hq issue 486): %w", w.now().Sub(began).Round(time.Second), repository, asked, err)
|
||||||
|
}
|
||||||
|
if capped.Err() != nil {
|
||||||
|
return gaveUp() // the bound ended an ask in flight
|
||||||
|
}
|
||||||
|
if !forgeNotNow(err) {
|
||||||
|
return forgeFacts{}, err
|
||||||
|
}
|
||||||
|
if w.now().Sub(began)+pause > w.bound {
|
||||||
|
return gaveUp()
|
||||||
}
|
}
|
||||||
fmt.Printf("the forge cannot be asked about %s now (%v); asking again in %s, for at most %s in all "+
|
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)
|
"(novox/hq issue 486)\n", repository, err, pause, w.bound)
|
||||||
if err := w.sleep(ctx, pause); err != nil {
|
if serr := w.sleep(capped, pause); serr != nil {
|
||||||
return forgeFacts{}, fmt.Errorf("stopped waiting for the forge to answer about %s: %w", repository, err)
|
if ctx.Err() != nil {
|
||||||
|
return forgeFacts{}, fmt.Errorf("stopped waiting for the forge to answer about %s: %w", repository,
|
||||||
|
ctx.Err())
|
||||||
|
}
|
||||||
|
return gaveUp()
|
||||||
}
|
}
|
||||||
pause = min(2*pause, w.most)
|
pause = min(2*pause, w.most)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -32,7 +32,7 @@ func notServed() error {
|
|||||||
// A registration's forge question that lands while the tool runner re-serves is asked again, and the forge's
|
// 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).
|
// answer, seconds later, is the registration's (novox/hq issue 486).
|
||||||
func TestTheForgeNotServedForAMomentIsWaitedOut(t *testing.T) {
|
func TestTheForgeNotServedForAMomentIsWaitedOut(t *testing.T) {
|
||||||
w, paused := aFakeWait(90 * time.Second)
|
w, paused := aFakeWait(60 * time.Second)
|
||||||
asked := 0
|
asked := 0
|
||||||
facts, err := forgeWaitedOut(t.Context(), w, "novox/mesh-catalog", func(context.Context) (forgeFacts, error) {
|
facts, err := forgeWaitedOut(t.Context(), w, "novox/mesh-catalog", func(context.Context) (forgeFacts, error) {
|
||||||
asked++
|
asked++
|
||||||
@@ -52,25 +52,50 @@ func TestTheForgeNotServedForAMomentIsWaitedOut(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// A tool that did not answer within its wait is waited out the same way: nothing refused.
|
// A tool that did not answer within its wait has held the receive loop that long already: not asked again.
|
||||||
func TestTheForgeNotAnsweringInTimeIsAskedAgain(t *testing.T) {
|
func TestTheForgeNotAnsweringInTimeIsNotAskedAgain(t *testing.T) {
|
||||||
w, _ := aFakeWait(90 * time.Second)
|
w, paused := aFakeWait(60 * time.Second)
|
||||||
asked := 0
|
asked := 0
|
||||||
_, err := forgeWaitedOut(t.Context(), w, "novox/x", func(context.Context) (forgeFacts, error) {
|
_, err := forgeWaitedOut(t.Context(), w, "novox/x", func(context.Context) (forgeFacts, error) {
|
||||||
asked++
|
asked++
|
||||||
if asked == 1 {
|
return forgeFacts{}, fmt.Errorf("gitea did not answer: %w", link.ErrNoAnswer)
|
||||||
return forgeFacts{}, fmt.Errorf("gitea did not answer: %w", link.ErrNoAnswer)
|
|
||||||
}
|
|
||||||
return forgeFacts{ID: 1, Guarded: true}, nil
|
|
||||||
})
|
})
|
||||||
if err != nil || asked != 2 {
|
if !errors.Is(err, link.ErrNoAnswer) || asked != 1 || len(*paused) != 0 {
|
||||||
t.Fatalf("asked %d times: %v", asked, err)
|
t.Fatalf("a forge that did not answer in time was asked %d times, paused %v: %v", asked, *paused, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// An ask that hangs past the bound is cut off at the bound, not at its own wait, and the failure still says it
|
||||||
|
// waited: the receive loop is held no longer than the bound (novox/hq issue 486).
|
||||||
|
func TestAnAskHangingPastTheBoundIsCutOffAtIt(t *testing.T) {
|
||||||
|
w := waitOut{bound: 200 * time.Millisecond, first: 10 * time.Millisecond, most: 50 * time.Millisecond,
|
||||||
|
now: time.Now, sleep: sleepOrDone}
|
||||||
|
asked := 0
|
||||||
|
began := time.Now()
|
||||||
|
_, err := forgeWaitedOut(t.Context(), w, "novox/mesh-catalog", func(ctx context.Context) (forgeFacts, error) {
|
||||||
|
asked++
|
||||||
|
if asked == 1 {
|
||||||
|
return forgeFacts{}, notServed()
|
||||||
|
}
|
||||||
|
select { // link.Ask's own wait is 30 s; the bound must end it first
|
||||||
|
case <-ctx.Done():
|
||||||
|
return forgeFacts{}, fmt.Errorf("gitea did not answer: %w", link.ErrNoAnswer)
|
||||||
|
case <-time.After(30 * time.Second):
|
||||||
|
return forgeFacts{ID: 1, Guarded: true}, nil
|
||||||
|
}
|
||||||
|
})
|
||||||
|
if took := time.Since(began); took > 2*time.Second {
|
||||||
|
t.Fatalf("the wait took %s, past its bound of %s", took, w.bound)
|
||||||
|
}
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "waited") || !strings.Contains(err.Error(), "asking 2 times") ||
|
||||||
|
!strings.Contains(err.Error(), "issue 486") {
|
||||||
|
t.Fatalf("the bound's end does not say it waited: %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Past the bound the registration fails, saying it waited, for how long, and what the forge last said.
|
// Past the bound the registration fails, saying it waited, for how long, and what the forge last said.
|
||||||
func TestTheForgeNotServedPastTheBoundFailsSayingItWaited(t *testing.T) {
|
func TestTheForgeNotServedPastTheBoundFailsSayingItWaited(t *testing.T) {
|
||||||
w, paused := aFakeWait(90 * time.Second)
|
w, paused := aFakeWait(60 * time.Second)
|
||||||
_, err := forgeWaitedOut(t.Context(), w, "novox/mesh-catalog", func(context.Context) (forgeFacts, error) {
|
_, err := forgeWaitedOut(t.Context(), w, "novox/mesh-catalog", func(context.Context) (forgeFacts, error) {
|
||||||
return forgeFacts{}, notServed()
|
return forgeFacts{}, notServed()
|
||||||
})
|
})
|
||||||
@@ -81,8 +106,8 @@ func TestTheForgeNotServedPastTheBoundFailsSayingItWaited(t *testing.T) {
|
|||||||
for _, p := range *paused {
|
for _, p := range *paused {
|
||||||
waited += p
|
waited += p
|
||||||
}
|
}
|
||||||
if waited > 90*time.Second || waited < 60*time.Second {
|
if waited > 60*time.Second || waited < 40*time.Second {
|
||||||
t.Fatalf("waited %s in all, want close to and within the bound of 90s", waited)
|
t.Fatalf("waited %s in all, want close to and within the bound of 60s", waited)
|
||||||
}
|
}
|
||||||
for _, want := range []string{"waited " + waited.String(), "novox/mesh-catalog", "nothing serves", "issue 486"} {
|
for _, want := range []string{"waited " + waited.String(), "novox/mesh-catalog", "nothing serves", "issue 486"} {
|
||||||
if !strings.Contains(err.Error(), want) {
|
if !strings.Contains(err.Error(), want) {
|
||||||
@@ -96,7 +121,7 @@ func TestTheForgeNotServedPastTheBoundFailsSayingItWaited(t *testing.T) {
|
|||||||
|
|
||||||
// A refusal is the forge's word: never asked again.
|
// A refusal is the forge's word: never asked again.
|
||||||
func TestARefusalFromTheForgeIsNeverAskedAgain(t *testing.T) {
|
func TestARefusalFromTheForgeIsNeverAskedAgain(t *testing.T) {
|
||||||
w, paused := aFakeWait(90 * time.Second)
|
w, paused := aFakeWait(60 * time.Second)
|
||||||
asked := 0
|
asked := 0
|
||||||
_, err := forgeWaitedOut(t.Context(), w, "novox/x", func(context.Context) (forgeFacts, error) {
|
_, err := forgeWaitedOut(t.Context(), w, "novox/x", func(context.Context) (forgeFacts, error) {
|
||||||
asked++
|
asked++
|
||||||
@@ -109,7 +134,7 @@ func TestARefusalFromTheForgeIsNeverAskedAgain(t *testing.T) {
|
|||||||
|
|
||||||
// A registration whose own context ends stops waiting at once.
|
// A registration whose own context ends stops waiting at once.
|
||||||
func TestAWaitForTheForgeEndsWithItsContext(t *testing.T) {
|
func TestAWaitForTheForgeEndsWithItsContext(t *testing.T) {
|
||||||
w, _ := aFakeWait(90 * time.Second)
|
w, _ := aFakeWait(60 * time.Second)
|
||||||
ctx, cancel := context.WithCancel(t.Context())
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
w.sleep = func(ctx context.Context, d time.Duration) error { cancel(); return sleepOrDone(ctx, d) }
|
w.sleep = func(ctx context.Context, d time.Duration) error { cancel(); return sleepOrDone(ctx, d) }
|
||||||
asked := 0
|
asked := 0
|
||||||
|
|||||||
Reference in New Issue
Block a user