Compare commits

..
11 Commits
Author SHA1 Message Date
jochen f9bd41d436 Say every walk a verb keeps, and clear a stop from a walk that walks again
mesh/merge-gate pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus step; every machine composes with the change as it…
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery superseded: a newer head of the same pull request
cancel, clear and rebuild kept a walk's end or restart without saying it, as plans stop, retry and
release-backlog did before them. The controller's run now says every save a verb makes on the bus, so the
next verb cannot lose a mark the same way. A stop is cleared when a walk is resumed, so one that fails again
ends failed and the search for its merge runs (review 21, ADR 0308, issue 532).
2026-10-12 12:29:18 +02:00
jochen 6a898e3d5b Say a walk's start or end when a person stops, closes, retries or releases it, and a hand stop as stopped
plans stop|close, plans retry and upgrade release-backlog kept a walk's start or end mark without saying it,
so the router would hold what the walk touched until its bound; each now runs on the bus as plans go does.
A hand stop keeps who stopped it, so its outcome is stopped, not failed (review 15, ADR 0308, issue 532).
2026-10-12 12:22:21 +02:00
jochen 015bfdbec7 Say what a walk touches, its bound and its outcome when it starts and ends (ADR 0308, issue 532)
The router holds what an open walk touches and tells its result once; it can do so
only if plan-moved names the walk's start and end, the modules and machines it moves,
what a moved tool runner or node-engine carries there, and how long to hold.
2026-10-12 12:22:21 +02:00
mesh-admin f549647d18 Merge pull request 'Give the unheld log test its own writer and memory, so it passes on a repeat (issue 486)' (#246) from fix/486-the-unheld-log-test-stands-alone into main 2026-10-12 10:10:39 +00:00
mesh-admin 22a2f73f7b Merge pull request 'plans names the hold a waiting walk waits on (issue 529)' (#245) from fix/529-a-waiting-walk-is-answered-or-named into main 2026-10-12 09:55:17 +00:00
mesh-admin d854598d8c 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
jschoubben ac5d34f613 Give the unheld log test its own writer and memory, so it passes on a repeat (novox/hq issue 486)
mesh/delivery delivered
mesh/merge-gate pass: builds mesh-controller → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed
logUnheld kept what it had logged in package state and the test swapped os.Stderr, so a second run in one process logged nothing and failed.
2026-10-12 11:41:41 +02:00
jschoubben b03ec3b64e Offer no hand start for a walk that is only waiting its turn (issue 529)
mesh/merge-gate pass: builds mesh-controller → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery delivered
mesh/delivery-group group fix/529-a-waiting-walk-is-answered-or-named merged: every member is merged
A hand start there would skip its group's order; point at it only for a
walk nothing holds, and say why nothing holds it.
2026-10-12 11:34:54 +02:00
jschoubben 4aacd98f8d Cap the forge wait as a whole, and wait out only a forge not served (novox/hq issue 486)
mesh/merge-gate pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without …
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery delivered
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.
2026-10-12 11:32:14 +02:00
jschoubben ed037784e2 Name the hold a waiting walk waits on in plans (issue 529)
mesh/merge-gate pass: builds mesh-controller → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery superseded: a newer head of the same pull request
mesh/delivery-group group fix/529-a-waiting-walk-is-answered-or-named checking: 0 of 2 member(s) ready
A walk whose delivery was held for a person said only that it waits for
mesh-delivery's word and that plans go starts it by hand, pointing the
reader at a hand start instead of the person's decision. Ask mesh-delivery
what holds each waiting walk and say the held delivery, its reason and
mesh-delivery.release, or the delivery not at its turn.
2026-10-12 11:27:05 +02:00
jschoubben 47205403d3 Wait out a forge not served for a moment when registering a build (novox/hq issue 486)
mesh/merge-gate pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without …
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery superseded: a newer head of the same pull request
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.
2026-10-12 11:23:03 +02:00
19 changed files with 731 additions and 41 deletions
+1 -1
View File
@@ -226,7 +226,7 @@ func TestTheAssemblingBatchIsShown(t *testing.T) {
asksWithPaths(t)
var said []inventory.Plan
was := inventory.PlanSaved
inventory.PlanSaved = func(p inventory.Plan) { said = append(said, p) }
inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { said = append(said, p) }
t.Cleanup(func() { inventory.PlanSaved = was })
now := time.Now().UTC()
claude := catalogueMerge("553b7191claude", "app", now.Add(-20*time.Second))
+90 -1
View File
@@ -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
+98 -2
View File
@@ -10,6 +10,7 @@ import (
"slices"
"sort"
"strings"
"sync"
"time"
"github.com/nats-io/nats.go"
@@ -82,6 +83,38 @@ func waitingNote(p inventory.Plan) string {
p.Delivery.Awaits, p.ID)
}
// waitingSaid is what a walk waiting for its word says when the delivery's owner said what holds it (novox/hq
// issue 529): a delivery held for a person, its reason and the verb that releases it — the person's decision, not
// a hand start — or the deliveries not at their turn. Told nothing, it is waitingNote.
func waitingSaid(p inventory.Plan, lines []stalledLine) string {
if len(lines) == 0 {
return waitingNote(p)
}
var held, unheld, other []string
for _, l := range lines {
switch l.State {
case "held":
held = append(held, l.Says)
case "unheld":
unheld = append(unheld, l.Says)
default:
other = append(other, l.Says)
}
}
switch {
case len(held) > 0:
return fmt.Sprintf("published; its walk waits for %s's word, and a delivery it carries is held for a person: %s",
p.Delivery.Awaits, strings.Join(held, "; "))
case len(other) > 0:
// Not at its turn: it goes when its turn comes. A hand start here would skip its group's order (review of
// novox/hq issue 529), so none is offered.
return fmt.Sprintf("published; its walk waits for %s's word: %s", p.Delivery.Awaits, strings.Join(other, "; "))
}
// Held by nothing: nothing will give it its word, so the hand start is the way on.
return fmt.Sprintf("published; its walk waits for %s's word: %s — `plans go %s --why …` starts it by hand",
p.Delivery.Awaits, strings.Join(unheld, "; "), p.ID)
}
// letGo gives a waiting walk its word: by the delivery's owner, or a person. The next advance asks its
// first tier. Refused for a walk that does not wait.
func letGo(ctx context.Context, inv *inventory.Inventory, id, by, why string) (inventory.Plan, error) {
@@ -1221,9 +1254,10 @@ func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error
return err
}
bus := link.OverNATS{Conn: conn, JS: js}
was := inventory.PlanSaved
checkEvents = bus
inventory.PlanSaved = func(p inventory.Plan) { publishPlanMoved(ctx, bus, facts, p) }
defer func() { checkEvents, inventory.PlanSaved = nil, nil }()
inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { publishPlanMoved(ctx, bus, facts, p) }
defer func() { checkEvents, inventory.PlanSaved = nil, was }()
ran = true
return f()
})
@@ -1233,3 +1267,65 @@ func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error
}
return err
}
// sayingWhatAVerbKept is PlanSaved for a process that is not the serving controller: each walk the verb keeps is
// said as `plan-moved` before the save returns, with the facts of the inventory that kept it, on the bus the verb
// reaches. It reaches the bus on the first save and never for a verb that keeps no walk. When the bus cannot be
// reached, a save that started or ended a walk says so on the verb's error output: its mark is kept and no later
// save says it, so the router holds what the walk touched until its bound, or not at all for a start. done lets
// the bus go and takes the hook back.
func sayingWhatAVerbKept(ctx context.Context) (said func(*inventory.Inventory, inventory.Plan), done func()) {
var (
mu sync.Mutex
reached bool
bus link.Bus
closing func()
failed error
)
said = func(inv *inventory.Inventory, p inventory.Plan) {
mu.Lock()
defer mu.Unlock()
if !reached {
reached = true
if handActConn != nil {
bus, failed = overNATS(handActConn)
} else if js, err := aBus(); err != nil {
failed = err
} else if bus, failed = overNATS(js.Conn()); failed == nil {
closing = js.Close
} else {
js.Close()
}
}
if bus == nil {
switch p.WalkSaid {
case inventory.WalkStarted:
fmt.Fprintf(os.Stderr, "%s: its start is kept and could NOT be said as plan-moved (%v): the router "+
"does not hold what it touches\n", p.ID, failed)
case inventory.WalkEnded:
fmt.Fprintf(os.Stderr, "%s: its end is kept and could NOT be said as plan-moved (%v): the router holds "+
"what it touched until its bound\n", p.ID, failed)
}
return
}
publishPlanMoved(ctx, bus, inv, p)
}
done = func() {
mu.Lock()
defer mu.Unlock()
inventory.PlanSaved = nil
if closing != nil {
closing()
}
}
return said, done
}
// overNATS is the bus a connection carries, with its JetStream.
func overNATS(conn *nats.Conn) (link.Bus, error) {
js, err := conn.JetStream()
if err != nil {
return nil, err
}
return link.OverNATS{Conn: conn, JS: js}, nil
}
@@ -37,6 +37,9 @@ type stalledLine struct {
// Number is the pull request's, for a line of a head the forge never announced (novox/hq issue 347).
Number int `json:"number,omitempty"`
State string `json:"state"`
// Walk is the walk a line is about: a waiting walk no live delivery holds, or, asked with `walks`, each walk
// waiting for the owner's word and what holds it (novox/hq issues 362, 529).
Walk string `json:"walk,omitempty"`
// Waiting are the merge checks a line of the state `unanswered` waits on, as mesh-delivery says each:
// "mesh/merge-gate pending since <UTC time>", "mesh/repo-check never set" (novox/hq issue 438).
Waiting []string `json:"waiting,omitempty"`
@@ -125,6 +128,34 @@ func deliveriesStalled(ctx context.Context, conn *nats.Conn, held bool) ([]stall
return lines, nil
}
// walkWaits are, by walk, the lines mesh-delivery says of a walk waiting for its word: what holds it.
type walkWaits map[string][]stalledLine
// walksWaiting asks the delivery's owner what holds each walk waiting for its word (novox/hq issue 529): its
// `stalled` asked with `walks`, so a line of `plans` names a held delivery, its reason and the verb that releases
// it, instead of pointing at a hand start. Nil when no holder is on record or none answers: the line then reads as
// before. An owner from before this answers its stalled lines, of which only those naming a walk are kept.
func walksWaiting(ctx context.Context, conn *nats.Conn, held bool) walkWaits {
if !held {
return nil
}
raw, err := askDeliveryOwner(ctx, conn, "stalled", map[string]any{"walks": true})
if err != nil {
return nil
}
var lines []stalledLine
if json.Unmarshal(raw, &lines) != nil {
return nil
}
out := walkWaits{}
for _, l := range lines {
if l.Walk != "" {
out[l.Walk] = append(out[l.Walk], l)
}
}
return out
}
// stalledObservations are the conditions of what is stalled.
func stalledObservations(lines []stalledLine) []conditions.Observation {
out := make([]conditions.Observation, 0, len(lines))
+148
View File
@@ -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)
}
}
+102
View File
@@ -0,0 +1,102 @@
package main
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/inventory"
)
// novox/hq issue 529: a walk waiting because its delivery is held for a person said only "its walk waits for
// mesh-delivery's word — `plans go … --why …` starts it by hand", pointing a reader at a hand start instead of at
// the person's decision. The line names what mesh-delivery says holds the walk: the held delivery, its reason and
// `mesh-delivery.release`; or the delivery not at its turn.
func waitingWalk(now time.Time) inventory.Plan {
return inventory.Plan{ID: "plan-1791793277221379309", Repository: "novox/mesh-sdk", Branch: "main",
Commit: "9747f3e1b1ac", State: inventory.PlanBuilding, Tiers: [][]string{{"mesh-sdk"}},
Created: now.Add(-9 * time.Minute), Modules: map[string]*inventory.PlanModule{},
Delivery: &inventory.PlanDelivery{Awaits: "mesh-delivery"}}
}
func TestAWalkWaitingOnAHeldDeliveryNamesTheHold(t *testing.T) {
now := time.Date(2026, 10, 12, 8, 35, 0, 0, time.UTC)
p := waitingWalk(now)
held := "novox/mesh-sdk@d41d255f28ad held since 2026-10-12T08:23:00Z: its group adr-0307-state-history's composed " +
"check did not pass for the heads that merged — a person decides: `mesh-delivery.release` (id " +
"novox/mesh-sdk@d41d255f28ad, why) lets it go, `mesh-delivery.stop` ends it"
waits := walkWaits{p.ID: {{ID: "novox/mesh-sdk@d41d255f28ad", State: "held", Walk: p.ID, Says: held}}}
line := planLineSaid(p, now, pauseView{}, tierAtLeast, nil, waits)
for _, want := range []string{"held for a person", "composed check did not pass", "mesh-delivery.release"} {
if !strings.Contains(line, want) {
t.Fatalf("the line does not say %q: %q", want, line)
}
}
if strings.Contains(line, "starts it by hand") {
t.Fatalf("a held walk's line points at a hand start: %q", line)
}
// Told nothing — an older mesh-delivery, or none answering — it reads as before.
if line := planLineSaid(p, now, pauseView{}, tierAtLeast, nil, nil); !strings.Contains(line, "starts it by hand") {
t.Fatalf("told nothing, the line reads %q", line)
}
}
func TestAWalkWaitingOnADeliveryNotAtItsTurnSaysSo(t *testing.T) {
now := time.Date(2026, 10, 12, 8, 57, 0, 0, time.UTC)
p := waitingWalk(now)
waits := walkWaits{p.ID: {{ID: "novox/mesh-sdk@9747f3e1b1ac", State: "published", Walk: p.ID,
Says: "not let go yet: novox/mesh-tools@3f0c1a6cec18 (published) carried by it not at its turn"}}}
line := planLineSaid(p, now, pauseView{}, tierAtLeast, nil, waits)
if !strings.Contains(line, "not at its turn") || !strings.Contains(line, "novox/mesh-tools@3f0c1a6cec18") {
t.Fatalf("the line reads %q", line)
}
// No hand start offered: it would skip the group's order (review of issue 529).
if strings.Contains(line, "plans go") {
t.Fatalf("a walk not at its turn points at a hand start: %q", line)
}
}
// A walk nothing holds — the nzbgeek case, its merges' deliveries ended as nothing-to-move — is said so, and the hand
// start is the way on.
func TestAWalkHeldByNothingSaysWhyAndTheHandStart(t *testing.T) {
now := time.Date(2026, 10, 11, 20, 30, 0, 0, time.UTC)
p := waitingWalk(now)
waits := walkWaits{p.ID: {{ID: "novox/mesh-media-catalog@cdcdcdcd0000", State: "unheld", Walk: p.ID,
Says: "held by nothing: its merges' deliveries ended as nothing-to-move (novox/mesh-media-catalog@abababab0000), " +
"yet the walk moves nzbgeek, nzbget — nothing gives it its word"}}}
line := planLineSaid(p, now, pauseView{}, tierAtLeast, nil, waits)
if !strings.Contains(line, "ended as nothing-to-move") || !strings.Contains(line, "plans go "+p.ID) {
t.Fatalf("the line reads %q", line)
}
}
// What holds each waiting walk is asked of mesh-delivery's `stalled` with `walks`, a verb the controller's grant
// already names; only lines naming a walk are kept, by walk.
func TestWhatHoldsTheWaitingWalksIsAskedOfTheDeliveryOwner(t *testing.T) {
was := askDeliveryOwner
defer func() { askDeliveryOwner = was }()
var asked map[string]any
askDeliveryOwner = func(_ context.Context, _ *nats.Conn, verb string, args map[string]any) (json.RawMessage, error) {
if verb != "stalled" {
t.Fatalf("asked %s", verb)
}
asked = args
return json.Marshal([]stalledLine{{ID: "novox/a@1", State: "held", Walk: "plan-1", Says: "held: why"},
{ID: "novox/b@2", State: "held", Says: "a delivery past its bound, no walk"}})
}
got := walksWaiting(t.Context(), nil, true)
if asked["walks"] != true {
t.Fatalf("stalled was asked with %v", asked)
}
if len(got) != 1 || len(got["plan-1"]) != 1 || got["plan-1"][0].Says != "held: why" {
t.Fatalf("the waiting walks are %+v", got)
}
if got := walksWaiting(t.Context(), nil, false); got != nil {
t.Fatalf("with no holder on record it asked: %+v", got)
}
}
+9
View File
@@ -63,6 +63,15 @@ func run() error {
// 45 §6), so the next controller takes it at once rather than after its age.
defer theLease.release()
// Every walk a verb keeps is said on the bus, whichever verb kept it (novox/hq ADR 0308 decision 1): the save
// that starts or ends a walk keeps its mark, and no later save says it again. The serving controller says its
// own saves.
if args[0] != "serve" {
said, done := sayingWhatAVerbKept(ctx)
inventory.PlanSaved = said
defer done()
}
switch args[0] {
case "build":
return buildCommand(ctx, args[1:])
+25 -10
View File
@@ -7,6 +7,7 @@ import (
"errors"
"flag"
"fmt"
"io"
"os"
"slices"
"sort"
@@ -1691,17 +1692,27 @@ func providerModuleOf(resolved catalogue.Resolution, open *stores, ctx context.C
return ""
}
// unheldLogged is what was last logged about each node's unmet seat dependencies, so the log says
// each change once (novox/hq ADR 0207), on stderr so `status --json` stays a document.
// unheldLogger remembers what was last logged about each node's unmet seat dependencies, so the
// log says each change once (novox/hq ADR 0207), on stderr so `status --json` stays a document.
// A test gives its own writer and memory, and no longer shares the process's (novox/hq issue 486).
//
// **The serving controller's log only.** planFor runs for every node on every push, assignment and
// status — `blockedElsewhere` alone resolves the whole mesh — and a one-shot command starts with an
// empty memory, so every node's report was "a change" and a push printed the whole mesh's list,
// burying the line about the node it acted on. A command says what concerns its own act instead
// (unheldChange, reportUnheldPushed); the full list is `status`'s.
type unheldLogger struct {
mu sync.Mutex
logged map[string]string
out io.Writer
}
func newUnheldLogger(out io.Writer) *unheldLogger {
return &unheldLogger{logged: map[string]string{}, out: out}
}
var (
unheldLogged = map[string]string{}
unheldLoggedMu sync.Mutex
unheldLog = newUnheldLogger(os.Stderr)
logUnheldChanges bool
)
@@ -1711,23 +1722,27 @@ func logUnheld(node string, unheld []catalogue.Unheld) {
if !logUnheldChanges {
return
}
unheldLog.log(node, unheld)
}
func (l *unheldLogger) log(node string, unheld []catalogue.Unheld) {
lines := make([]string, 0, len(unheld))
for _, u := range unheld {
lines = append(lines, u.String())
}
now := strings.Join(lines, "\n")
unheldLoggedMu.Lock()
before, seen := unheldLogged[node]
unheldLogged[node] = now
unheldLoggedMu.Unlock()
l.mu.Lock()
defer l.mu.Unlock()
before, seen := l.logged[node]
l.logged[node] = now
if (seen && before == now) || (!seen && now == "") {
return
}
if now == "" {
fmt.Fprintf(os.Stderr, "%s: every seat its modules depend on is held (novox/hq ADR 0207)\n", node)
fmt.Fprintf(l.out, "%s: every seat its modules depend on is held (novox/hq ADR 0207)\n", node)
return
}
fmt.Fprintf(os.Stderr, "%s: %d unmet seat dependenc(ies), reported and not refused (novox/hq ADR 0207):\n %s\n",
fmt.Fprintf(l.out, "%s: %d unmet seat dependenc(ies), reported and not refused (novox/hq ADR 0207):\n %s\n",
node, len(lines), strings.Join(lines, "\n "))
}
+12
View File
@@ -251,6 +251,16 @@ func resumed(p *inventory.Plan, why string) {
if p.State == inventory.PlanFailed && len(failedIn(*p)) == 0 {
p.State = inventory.PlanBuilding
p.Note = why
goesOnAgain(p)
}
}
// goesOnAgain clears a stop from a walk that walks again: a stop said why the walk ended, and a walk resumed has
// not ended. Kept, it would end a retried walk that fails again as stopped, not failed (novox/hq ADR 0308
// decision 1), and keep the search for the merge that brought the failure from running (searchAfterFailures).
func goesOnAgain(p *inventory.Plan) {
if p.Delivery != nil {
p.Delivery.Stopped, p.Delivery.StoppedWhy = "", ""
}
}
@@ -460,6 +470,7 @@ func retryRollouts(ctx context.Context, open *stores, p *inventory.Plan) (string
said = append(said, m+" to "+strings.Join(again, ", "))
}
p.State = inventory.PlanRolling
goesOnAgain(p)
p.Note = fmt.Sprintf("tier %d retried by hand; sent %s first again", p.Tier, strings.Join(said, "; "))
if err := open.inventory.SavePlan(ctx, p); err != nil {
return "", err
@@ -525,6 +536,7 @@ func retryTierWhole(ctx context.Context, open *stores, p *inventory.Plan, again
}
if p.State == inventory.PlanFailed && len(failedIn(*p)) == 0 {
p.State = inventory.PlanBuilding
goesOnAgain(p)
p.Note = fmt.Sprintf("tier %d retried by hand: %s asked again; %s sent again with the tier", p.Tier,
strings.Join(again, ", "), orNone(strings.Join(resent, ", ")))
}
+1 -1
View File
@@ -249,7 +249,7 @@ func serve(ctx context.Context) (err error) {
// (novox/hq to-be 45 §9).
checkEvents = bus
// And every walk this controller keeps in a new state, for the delivery it walks (novox/hq ADR 0239).
inventory.PlanSaved = func(p inventory.Plan) { sayPlanMoved(ctx, bus, open.inventory, p) }
inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { sayPlanMoved(ctx, bus, open.inventory, p) }
if err := server.Checks(following{open}); err != nil {
return err
}
+39 -4
View File
@@ -11,6 +11,8 @@ import (
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
@@ -1258,6 +1260,13 @@ func planLineWith(p inventory.Plan, now time.Time, pause pauseView, bound time.D
// planLineOn is planLineWith naming the plan by what it moves and where (planHeadline), given what runs each
// module; nil names no machine.
func planLineOn(p inventory.Plan, now time.Time, pause pauseView, bound time.Duration, running func(string) []string) string {
return planLineSaid(p, now, pause, bound, running, nil)
}
// planLineSaid is planLineOn saying, of a walk waiting for its delivery's word, what the delivery's owner says
// holds it (novox/hq issue 529); nil waits says it as before.
func planLineSaid(p inventory.Plan, now time.Time, pause pauseView, bound time.Duration, running func(string) []string,
waits walkWaits) string {
name := planHeadline(p, running)
where := fmt.Sprintf("tier %d of %d", min(p.Tier+1, len(p.Tiers)), len(p.Tiers))
switch p.State {
@@ -1274,7 +1283,7 @@ func planLineOn(p inventory.Plan, now time.Time, pause pauseView, bound time.Dur
since := now.Sub(inTierSince(p)).Round(time.Second)
if p.Waiting() {
// Waiting for its delivery's word is no lateness of the walk's (novox/hq ADR 0239).
return fmt.Sprintf("%s · %s, %s, for %s", name, where, waitingNote(p), since)
return fmt.Sprintf("%s · %s, %s, for %s", name, where, waitingSaid(p, waits[p.ID]), since)
}
if deferred(p) {
// Nor is waiting for the walk before it to end (one walk at a time, novox/hq ADR 0276).
@@ -1479,11 +1488,11 @@ func plansCommand(ctx context.Context, args []string) error {
return err
}
bounds := readTierBounds(ctx, inv, now)
fmt.Printf("%s — %s\n", p.ID, planLineOn(p, now, buildSeatPause(ctx, inv, []inventory.Plan{p}),
fmt.Printf("%s — %s\n", p.ID, planLineSaid(p, now, buildSeatPause(ctx, inv, []inventory.Plan{p}),
bounds.of(p.Repository), func(module string) []string {
on, _ := inv.Running(ctx, module)
return on
}))
}, waitsOf(ctx, inv, []inventory.Plan{p})))
if r := p.Release; r != nil {
// A release plan's walk (ADR 0236): machines done, the one judged, those to come.
fmt.Printf(" machines in order: %s; done: %s; skipped: %s\n", strings.Join(r.Order, ", "),
@@ -1638,15 +1647,41 @@ func plansCommand(ctx context.Context, args []string) error {
on, _ := inv.Running(ctx, module)
return on
}
waits := waitsOf(ctx, inv, plans)
for _, p := range plans {
if p.Batch() {
continue
}
fmt.Printf("%-28s %s\n", p.ID, planLineOn(p, now, pause, bounds.of(p.Repository), running))
fmt.Printf("%-28s %s\n", p.ID, planLineSaid(p, now, pause, bounds.of(p.Repository), running, waits))
}
return nil
}
// waitsOf asks the delivery's owner what holds the walks among these that wait for its word (novox/hq issue 529),
// once, and only when one waits; nil when none waits, no holder is on record, or the bus or the owner does not
// answer within a few seconds — the lines then read as before.
func waitsOf(ctx context.Context, inv *inventory.Inventory, plans []inventory.Plan) walkWaits {
waiting := false
for _, p := range plans {
waiting = waiting || p.Open() && p.Waiting()
}
if !waiting {
return nil
}
entries, err := inv.Catalogued(ctx)
if err != nil {
return nil
}
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
var waits walkWaits
_ = onTheBus(func(conn *nats.Conn) error {
waits = walksWaiting(ctx, conn, deliverySeatHeld(entries))
return nil
})
return waits
}
// planWhatIf is the plan a merge would produce, computed the way the merge handler computes one
// and saved nowhere: the modules the repository's changed files touch (or the modules named), what
// packages their source, everything reachable from them, in tiers. For reading before merging.
+11 -10
View File
@@ -2,7 +2,6 @@ package main
import (
"bytes"
"os"
"strings"
"testing"
@@ -81,28 +80,30 @@ func TestAPushSaysANamedNodesDependenciesAndCountsTheRest(t *testing.T) {
}
}
// The test gives logUnheld its own writer and memory, so a repeat in one process and the
// package's other tests see neither the last run's memory nor a swapped os.Stderr (novox/hq issue 486).
func TestOnlyTheServingControllerLogsEachChange(t *testing.T) {
var out bytes.Buffer
saved, savedOn := unheldLog, logUnheldChanges
unheldLog, logUnheldChanges = newUnheldLogger(&out), false
t.Cleanup(func() { unheldLog, logUnheldChanges = saved, savedOn })
read := func(f func()) string {
old := os.Stderr
r, w, _ := os.Pipe()
os.Stderr = w
out.Reset()
f()
_ = w.Close()
os.Stderr = old
var b bytes.Buffer
_, _ = b.ReadFrom(r)
return b.String()
return out.String()
}
u := []catalogue.Unheld{{Node: "n1", Module: "web", Seat: catalogue.ContainerRuntimeSeat}}
if got := read(func() { logUnheld("n1", u) }); got != "" {
t.Errorf("a command logged:\n%s", got)
}
logUnheldChanges = true
defer func() { logUnheldChanges = false }()
if got := read(func() { logUnheld("n1", u) }); !strings.Contains(got, "web on n1") {
t.Errorf("the serving controller did not log a change:\n%s", got)
}
if got := read(func() { logUnheld("n1", u) }); got != "" {
t.Errorf("an unchanged report was logged again:\n%s", got)
}
if got := read(func() { logUnheld("n1", nil) }); !strings.Contains(got, "every seat its modules depend on is held") {
t.Errorf("a report that became none was not logged:\n%s", got)
}
}
+123
View File
@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"os"
"reflect"
"testing"
"time"
@@ -389,3 +390,125 @@ func TestABacklogReleasedByAPersonIsSaidStarted(t *testing.T) {
t.Fatalf("the release's start: %v", started)
}
}
// asTyped runs a verb as the controller's command line runs it, through run, as a person's terminal would.
func asTyped(t *testing.T, args ...string) error {
t.Helper()
was := os.Args
os.Args = append([]string{"mesh-controller"}, args...)
t.Cleanup(func() { os.Args = was })
err := run()
// run gives back the lease as the process would end; the test's process goes on, and takes it again.
theLease.mu.Lock()
theLease.held, theLease.kv, theLease.noBus = nil, nil, false
theLease.mu.Unlock()
return err
}
// aStartedWalkAsking keeps a live walk, said started, whose first tier asked a as build.
func aStartedWalkAsking(t *testing.T, open *stores, id, build string, asked time.Time) inventory.Plan {
t.Helper()
walk := inventory.Plan{ID: id, Repository: "novox/a", Branch: "main", Commit: "c0ffee", Created: asked,
State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
Modules: map[string]*inventory.PlanModule{"a": {State: "asked", AskedAt: &asked, Build: build}},
Times: &inventory.PlanTimes{Cut: &asked, Started: &asked, Class: inventory.ClassLeaf}}
if err := open.inventory.SavePlan(t.Context(), &walk); err != nil {
t.Fatal(err)
}
return walk
}
// Live, with the store and a bus: `cancel <build>` and `clear` fail the walk that asked the build, and the verb
// says its end, failed (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247).
func TestAWalkEndedByCancelOrClearIsSaidEnded(t *testing.T) {
_, ask := aBuildQueue(t)
open := aMesh(t)
next := planMovedHeard(t)
twoTiers(t, open)
for _, verb := range [][]string{{"cancel"}, {"clear"}} {
id := link.NewBuildID(time.Now())
ask(id, "https://forge.example/novox/a.git")
asked, _ := link.BuildAskedAt(id)
walk := aStartedWalkAsking(t, open, "plan-308-"+verb[0], id, asked)
args := verb
if verb[0] == "cancel" {
args = append(args, id)
}
if err := asTyped(t, args...); err != nil {
t.Fatal(err)
}
ended := next(inventory.WalkEnded)
if ended["walk"] != walk.ID || ended["outcome"] != inventory.WalkFailed {
t.Fatalf("%s: the end is %v", verb[0], ended)
}
// Ended, and so the next walk may be cut on the same repository.
p, _ := open.inventory.PlanByID(t.Context(), walk.ID)
if p.State != inventory.PlanFailed {
t.Fatalf("%s left the walk %s", verb[0], p.State)
}
}
}
// Live, with the store and a bus: `rebuild <module>` joining a failed walk sets it building again, and the verb
// says its start (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247).
func TestAWalkRestartedByARebuildIsSaidStarted(t *testing.T) {
open := aMesh(t)
asksRecorded(t)
next := planMovedHeard(t)
twoTiers(t, open)
before := time.Now().UTC().Add(-time.Hour)
ended := before.Add(time.Minute)
failed := inventory.Plan{ID: "plan-308-rebuild", Repository: "novox/a", Branch: "main", Commit: "c0ffee",
Created: before, State: inventory.PlanFailed, Tiers: [][]string{{"a"}, {"b"}},
Modules: map[string]*inventory.PlanModule{"a": {State: "failed", AskedAt: &before, Build: "build-1",
Why: link.CancelledByHand}},
Times: &inventory.PlanTimes{Cut: &before, Started: &before, Ended: &ended, Class: inventory.ClassLeaf}}
if err := open.inventory.SavePlan(t.Context(), &failed); err != nil {
t.Fatal(err)
}
if err := asTyped(t, "rebuild", "a"); err != nil {
t.Fatal(err)
}
if started := next(inventory.WalkStarted); started["walk"] != failed.ID {
t.Fatalf("the rebuilt walk's start: %v", started)
}
}
// Live, with the store: a walk a person stopped and then retried, by `plans retry` or by `rebuild`, walks again
// without the stop, so when it fails again it ends failed, not stopped, and the search for the merge that
// brought the failure is not kept from it (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247).
func TestAWalkStoppedByHandAndRetriedEndsFailedWhenItFailsAgain(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
asksRecorded(t)
twoTiers(t, open)
for _, again := range [][]string{{"plans", "retry"}, {"rebuild", "a"}} {
before := time.Now().UTC().Add(-time.Hour)
walk := aStartedWalkAsking(t, open, "plan-308-again-"+again[0], "build-1", before)
if err := asTyped(t, "plans", "stop", walk.ID, "--why", "a test"); err != nil {
t.Fatal(err)
}
// The build asked before the stop fails after it.
p, _ := open.inventory.PlanByID(ctx, walk.ID)
p.Modules["a"].State, p.Modules["a"].Why = "failed", "it broke"
if err := open.inventory.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
args := again
if again[0] == "plans" {
args = append(append([]string(nil), again...), walk.ID)
}
if err := asTyped(t, args...); err != nil {
t.Fatal(err)
}
p, _ = open.inventory.PlanByID(ctx, walk.ID)
if p.State != inventory.PlanBuilding || (p.Delivery != nil && p.Delivery.Stopped != "") {
t.Fatalf("%s: the walk is %s and keeps its stop: %+v", again[0], p.State, p.Delivery)
}
asking, _ := link.BuildAskedAt(p.Modules["a"].Build)
planBuilt(ctx, open, "a", "c0ffee", "it broke again", asking, p.Modules["a"].Build)
if p, _ = open.inventory.PlanByID(ctx, walk.ID); p.State != inventory.PlanFailed || p.Outcome() != inventory.WalkFailed {
t.Fatalf("%s: failing again, the walk is %s and ends %s", again[0], p.State, p.Outcome())
}
}
}
+7 -5
View File
@@ -244,10 +244,12 @@ func (p Plan) Waiting() bool {
return p.Delivery != nil && p.Delivery.Awaits != "" && p.Delivery.Go == nil
}
// PlanSaved is told every plan this process kept, after it is kept (novox/hq ADR 0239): the serving
// controller says it on the bus as `plan-moved`. Nil in a command, which says nothing; whoever follows a
// walk also asks for it, so a save made by a command is found by comparison.
var PlanSaved func(Plan)
// PlanSaved is told every plan this process kept, after it is kept, with the inventory that kept it (novox/hq
// ADR 0239): the serving controller says it on the bus as `plan-moved`, and so does a command (the controller's
// run installs it for every verb, novox/hq ADR 0308 decision 1), so the save that starts or ends a walk is said
// whichever process made it. Nil says nothing; whoever follows a walk also asks for it, and finds an unsaid save
// by comparison.
var PlanSaved func(*Inventory, Plan)
// ErrPlanMoved is a save against a plan written by somebody else since it was read.
var ErrPlanMoved = errors.New("the plan was written by somebody else since it was read")
@@ -519,7 +521,7 @@ func (i *Inventory) SavePlansMoving(ctx context.Context, plans []*Plan, moves []
p.Epoch = 0
}
if PlanSaved != nil {
PlanSaved(*p)
PlanSaved(i, *p)
}
}
return nil
+1 -1
View File
@@ -96,7 +96,7 @@ func TestAWalksStartIsKeptAndARefusedSaveLeavesItUnsaid(t *testing.T) {
Times: &PlanTimes{Cut: &cut, Class: ClassLeaf}}
var said []string
was := PlanSaved
PlanSaved = func(p Plan) { said = append(said, p.WalkSaid) }
PlanSaved = func(_ *Inventory, p Plan) { said = append(said, p.WalkSaid) }
t.Cleanup(func() { PlanSaved = was })
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
+17 -4
View File
@@ -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)
}
+3 -2
View File
@@ -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)
}
+10
View File
@@ -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.
+3
View File
@@ -17,3 +17,6 @@
./cmd/mesh-controller TestAWalkStoppedOrClosedByAPersonIsSaidEndedStopped
./cmd/mesh-controller TestAWalkRetriedByAPersonIsSaidStartedAgain
./cmd/mesh-controller TestABacklogReleasedByAPersonIsSaidStarted
./cmd/mesh-controller TestAWalkEndedByCancelOrClearIsSaidEnded
./cmd/mesh-controller TestAWalkRestartedByARebuildIsSaidStarted
./cmd/mesh-controller TestAWalkStoppedByHandAndRetriedEndsFailedWhenItFailsAgain