Merge pull request 'Wait out a forge not served for a moment when registering a build (issue 486)' (#244) from fix/486-the-forge-is-waited-out into main

2026-10-12 09:43:50 +00:00
parent da1473e9db 4aacd98f8d
commit d854598d8c
5 changed files with 268 additions and 7 deletions
@@ -153,8 +153,97 @@ 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
+// 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:
+// the registration holds the controller's one receive loop while it waits, and a loop held two minutes reads as
+// deaf. The pauses are those the build machine waits the registry with (novox/hq issue 457).
+var forgeWait = waitOut{bound: 60 * 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 served now — the tool runner re-serving — not that
+// the forge refused. Only that is asked again: a refusal, an unreadable answer or a repository with no id is the
+// 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 { return errors.Is(err, link.ErrNothingServes) }
+
+// forgeWaitedOut asks the forge about repository, asking again while it answers "not served now", all of it
+// 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,
+	ask func(context.Context) (forgeFacts, error)) (forgeFacts, error) {
+	capped, cancel := context.WithTimeout(ctx, w.bound)
+	defer cancel()
+	began := w.now()
+	pause := w.first
+	for asked := 1; ; asked++ {
+		facts, err := ask(capped)
+		if err == nil {
+			if 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, nil
+		}
+		if ctx.Err() != nil {
+			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 "+
+				"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 "+
+			"(novox/hq issue 486)\n", repository, err, pause, w.bound)
+		if serr := w.sleep(capped, pause); serr != nil {
+			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)
+	}
+}
+
+// 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
@@ -0,0 +1,148 @@
+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(60 * 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 has held the receive loop that long already: not asked again.
+func TestTheForgeNotAnsweringInTimeIsNotAskedAgain(t *testing.T) {
+	w, paused := aFakeWait(60 * time.Second)
+	asked := 0
+	_, err := forgeWaitedOut(t.Context(), w, "novox/x", func(context.Context) (forgeFacts, error) {
+		asked++
+		return forgeFacts{}, fmt.Errorf("gitea did not answer: %w", link.ErrNoAnswer)
+	})
+	if !errors.Is(err, link.ErrNoAnswer) || asked != 1 || len(*paused) != 0 {
+		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.
+func TestTheForgeNotServedPastTheBoundFailsSayingItWaited(t *testing.T) {
+	w, paused := aFakeWait(60 * 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 > 60*time.Second || waited < 40*time.Second {
+		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"} {
+		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(60 * 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(60 * 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)
+	}
+}
@@ -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)
 	}
@@ -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)
 	}
@@ -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.