Compare commits

..
Author SHA1 Message Date
mesh-tools fd20b40dfe Bring the look back for a tier the bus refused whole, clear the wait when it ends, and say the wait in plain words (issue 1334, review round 1) 2026-10-20 12:04:00 +00:00
mesh-tools 3d49a971a1 Ask a module again at the next look when the bus did not answer, and fail the plan only after ten minutes (issue 1334) 2026-10-20 11:56:28 +00:00
mesh-tools 6c2a8d6210 Merge pull request 'Carry a module on the tool runner only when the runner serves something of it, so a part on its own credential has one writer (issues 1329, 1338)' (#404) from fix/1329-runner-carries-only-what-it-serves into main 2026-10-19 23:39:24 +00:00
mesh-tools c07f604047 Make the rollout tests wait for the plan look to return before their cleanups restore what it reads, which raced (issue 1338)
mesh/merge-gate head c07f604047be merged into trunk 70cb217601f9 — pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus…
mesh/repo-check head c07f604047be merged into trunk 70cb217601f9 — pass: its merge-check.sh passed
mesh/delivery delivered
2026-10-19 23:23:23 +00:00
mesh-tools b893cc829c Keep carrying a module whose bundle the tool runtime loads, as route-proxy's helper writes its state on the runner's account (issue 1329) 2026-10-19 23:14:51 +00:00
mesh-tools 70cb217601 Merge pull request 'Answer a single pull request's check with its request id on clean standard output (issue 1311)' (#403) from feat/no-terminal into main 2026-10-19 23:11:38 +00:00
mesh-tools 1d0df24a0d Carry a module on the tool runner only when the runner has something of it to serve, so a part on its own credential has one writer (issue 1329) 2026-10-19 23:10:48 +00:00
mesh-tools 202b453de1 Point standard output at standard error while a one-head delivery check runs, so its answer is the only JSON (issue 1311)
mesh/merge-gate out of date: judged against trunk 40350d82fa20, the trunk is now 70cb217601f9; judged again on a recheck or when it is merged
mesh/repo-check out of date: judged against trunk 40350d82fa20, the trunk is now 70cb217601f9; judged again on a recheck or when it is merged
mesh/delivery delivered
2026-10-19 22:52:00 +00:00
mesh-tools 7a0b275020 Keep delivery check standard output the answer alone and answer the request id asked (issue 1311) 2026-10-19 22:46:30 +00:00
mesh-tools 40350d82fa Merge pull request 'Ask a failed delivery's Retry on the phone, tell a refused answer, and pass a seat to its only claimant (ADR 0449, ADR 0451)' (#402) from feat/no-terminal into main 2026-10-19 22:44:07 +00:00
mesh-tools 7f23c537cc Let a retry see every plan made after the failed one, and pin the 7-day end of plan-failed (issue 1328, ADR 0451)
mesh/merge-gate head 7f23c537cced merged into trunk 1f708e2d8d72 — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 7f23c537cced merged into trunk 1f708e2d8d72 — pass: its merge-check.sh passed
mesh/delivery delivered
2026-10-19 21:48:41 +00:00
mesh-tools f1775281e2 Refuse a retry of a failed plan that a newer done plan of its branch has passed (issue 1328, ADR 0451) 2026-10-19 21:45:48 +00:00
mesh-tools 7b83aaa949 Tell a refusal as one and an unanswered call as not known; ask a plan that fails again within the hour (issues 1316, 1314, review round 1) 2026-10-19 21:40:27 +00:00
mesh-tools 3448f98000 Make a seat pass no repair, read every machine before passing, keep the gate's pass in memory (issue 1317, review round 1) 2026-10-19 21:38:41 +00:00
mesh-tools 12db1d3748 Tell the operator when the answer to a plan's ask is refused (issue 1316, ADR 0449) 2026-10-19 21:30:33 +00:00
mesh-tools 7770d193da Word and perform a failed plan's Retry in the asker (issue 1315, ADR 0449) 2026-10-19 21:28:57 +00:00
mesh-tools 248fe521a3 Raise plan-failed for a failed delivery plan, with Retry when a retry would run (issue 1314, ADR 0449) 2026-10-19 21:26:43 +00:00
mesh-tools 666dcff79b Pass a seat held once to its only claimant when its recorded holder is assigned nowhere (issue 1317, ADR 0449) 2026-10-19 21:21:32 +00:00
39 changed files with 2531 additions and 66 deletions
@@ -0,0 +1,134 @@
//go:build integration
package main
import (
"context"
"fmt"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// Issue 1334, the look: when the bus did not answer an ask of any module of a tier, no build is in flight and no event follows,
// so the look itself comes back (it fails, and the looker retries it), asks the modules again, and past the bound fails the plan.
func TestALookComesBackForATierTheBusRefusedWhole(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
inv := open.inventory
for _, name := range []string{"a", "b"} {
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: name, Version: "1"},
inventory.Source{Repository: "novox/" + name, Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
}
p := inventory.Plan{ID: "plan-bus", Repository: "novox/a", Branch: "main", Commit: "c1", Created: time.Now().UTC(),
State: inventory.PlanBuilding, Tiers: [][]string{{"a", "b"}}, Modules: map[string]*inventory.PlanModule{"a": {}, "b": {}}}
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
bus := false
was := askABuild
askABuild = func(_ context.Context, source buildSource, _, _ string) (string, error) {
if !bus {
return "", fmt.Errorf("%w: nats: no responders available for request", inventory.ErrNoChain)
}
return "build-" + source.Repository, nil
}
t.Cleanup(func() { askABuild = was })
look := func() bool {
t.Helper()
release, err := inv.HoldPlans(ctx, true)
if err != nil {
t.Fatal(err)
}
defer release()
return advanceHeld(ctx, open)
}
read := func() inventory.Plan {
t.Helper()
got, err := inv.PlanByID(ctx, "plan-bus")
if err != nil {
t.Fatal(err)
}
return got
}
// Every module refused: the look says it failed, which has the looker come back; the plan is building and says why.
if !look() {
t.Fatal("a look that left the whole tier unasked for the bus did not fail, so nothing would come back to it")
}
if got := read(); got.State != inventory.PlanBuilding || !strings.Contains(got.Note, "bus did not answer") ||
got.Modules["a"].State != "" || got.Modules["b"].State != "" {
t.Fatalf("the plan after a refused tier: %s %q %+v %+v", got.State, got.Note, got.Modules["a"], got.Modules["b"])
}
// Again, still refused: the same fact is not saved again, and the look still comes back.
before := read().Revision
if !look() || read().Revision != before {
t.Fatalf("a second refused look failed to come back or saved the plan again (revision %d, was %d)", read().Revision, before)
}
// The bus answers at the next look: both are asked, and nothing waits any more.
bus = true
if look() {
t.Fatal("a look that asked the whole tier failed")
}
if got := read(); got.Modules["a"].State != "asked" || got.Modules["b"].State != "asked" || got.Modules["a"].BusWaitFrom != nil {
t.Fatalf("the tier was not asked once the bus answered: %+v %+v", got.Modules["a"], got.Modules["b"])
}
}
func TestALookFailsTheTierTheBusRefusedPastTheBound(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
inv := open.inventory
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "a", Version: "1"},
inventory.Source{Repository: "novox/a", Seat: "git", Ref: "main", BuiltFrom: "c0", Head: "c0"}); err != nil {
t.Fatal(err)
}
long := time.Now().UTC().Add(-busWaitBound - time.Minute)
p := inventory.Plan{ID: "plan-bus", Repository: "novox/a", Branch: "main", Commit: "c1", Created: time.Now().UTC(),
State: inventory.PlanBuilding, Tiers: [][]string{{"a"}}, Modules: map[string]*inventory.PlanModule{"a": {BusWaitFrom: &long}}}
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
was := askABuild
askABuild = func(context.Context, buildSource, string, string) (string, error) {
return "", fmt.Errorf("%w: nats: no responders available for request", inventory.ErrNoChain)
}
t.Cleanup(func() { askABuild = was })
release, err := inv.HoldPlans(ctx, true)
if err != nil {
t.Fatal(err)
}
advanceHeld(ctx, open)
release()
got, err := inv.PlanByID(ctx, "plan-bus")
if err != nil {
t.Fatal(err)
}
if got.State != inventory.PlanFailed || !strings.Contains(got.Note, "the bus did not answer an ask for 10m0s (since ") {
t.Fatalf("a bus silent past its bound did not fail the plan in words: %s %q", got.State, got.Note)
}
// Rebuilt by hand while the bus still reloads: said as waiting, not as a failure and not with an empty reason.
joined, said, err := joinAPlan(ctx, open, "a")
if err != nil || !joined || !strings.Contains(said, "not asked yet: the bus did not answer") {
t.Fatalf("a rebuild left for the bus is said as %v %q %v", joined, said, err)
}
// And retried by hand: the module is named as waiting, not as asked again.
got, _ = inv.PlanByID(ctx, "plan-bus")
if got.State == inventory.PlanFailed {
got.Modules["a"].State, got.Modules["a"].Why = "failed", "x"
if err := inv.SavePlan(ctx, &got); err != nil {
t.Fatal(err)
}
said, err := retryPlan(ctx, open, "plan-bus")
if err != nil || !strings.Contains(said, "not asked yet: the bus did not answer") {
t.Fatalf("a retry left for the bus is said as %q %v", said, err)
}
}
}
+139
View File
@@ -0,0 +1,139 @@
package main
import (
"context"
"errors"
"fmt"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/inventory"
)
// Issue 1334: a bus that does not answer for a moment (it reloads its user list, and JetStream's API has no responder for a few
// seconds) is not a failed build. The module is left unasked and the plan building, and the next look asks it again; only a
// module's own fault, or a bus silent for ten minutes, fails the plan.
func aTierOfThree() (*inventory.Plan, map[string]inventory.Entry) {
p := &inventory.Plan{ID: "plan-1", State: inventory.PlanBuilding, Tiers: [][]string{{"a", "b", "c"}},
Modules: map[string]*inventory.PlanModule{}}
byName := map[string]inventory.Entry{}
for _, m := range []string{"a", "b", "c"} {
byName[m] = inventory.Entry{Source: inventory.Source{Repository: "novox/" + m, Seat: "git", Ref: "main"}}
}
return p, byName
}
func TestABusThatDoesNotAnswerForAMomentDoesNotFailTheTier(t *testing.T) {
p, byName := aTierOfThree()
bus := false
was := askABuild
askABuild = func(_ context.Context, source buildSource, _, _ string) (string, error) {
if source.Repository == "novox/b" && !bus {
return "", fmt.Errorf("%w: the bucket mesh-controller_builds is not on the bus: nats: no responders available for request",
inventory.ErrNoChain)
}
return "build-" + source.Repository, nil
}
t.Cleanup(func() { askABuild = was })
for _, m := range []string{"a", "b", "c"} {
askModule(t.Context(), p, m, byName)
}
if p.State != inventory.PlanBuilding {
t.Fatalf("a bus that did not answer for a moment failed the plan: %s, %q", p.State, p.Note)
}
if p.Modules["a"].State != "asked" || p.Modules["c"].State != "asked" {
t.Fatalf("the modules the bus did answer for were not asked: %+v %+v", p.Modules["a"], p.Modules["c"])
}
if s := p.Modules["b"]; s.State != "" {
t.Fatalf("the module the bus did not answer for is %q, not left unasked", s.State)
}
if got := unaskedOf(*p); len(got) != 1 || got[0] != "b" {
t.Fatalf("the next look would ask %v, not b", got)
}
if !strings.Contains(p.Note, "bus did not answer") {
t.Fatalf("the plan says nothing of the wait: %q", p.Note)
}
// The next look, the bus answering: b is asked and the plan goes on.
bus = true
for _, m := range unaskedOf(*p) {
askModule(t.Context(), p, m, byName)
}
if p.State != inventory.PlanBuilding || p.Modules["b"].State != "asked" || p.Modules["b"].Build == "" {
t.Fatalf("b was not asked once the bus answered: %s %+v", p.State, p.Modules["b"])
}
}
func TestABusSilentForTenMinutesFailsTheTierAsBefore(t *testing.T) {
p, byName := aTierOfThree()
was := askABuild
askABuild = func(context.Context, buildSource, string, string) (string, error) {
return "", fmt.Errorf("%w: nats: no responders available for request", inventory.ErrNoChain)
}
t.Cleanup(func() { askABuild = was })
askModule(t.Context(), p, "a", byName)
if p.State != inventory.PlanBuilding {
t.Fatalf("the first refusal failed the plan: %q", p.Note)
}
long := time.Now().Add(-busWaitBound - time.Minute)
p.Modules["a"].BusWaitFrom = &long
askModule(t.Context(), p, "a", byName)
if p.State != inventory.PlanFailed || p.Modules["a"].State != "failed" {
t.Fatalf("a bus silent past its bound did not fail the plan: %s %+v", p.State, p.Modules["a"])
}
// Said in plain words: the bus, the wait, and since when.
want := "a could not be asked for: the bus did not answer an ask for 10m0s (since " + long.UTC().Format(time.RFC3339) +
"): this kind of record lives on the bus"
if !strings.HasPrefix(p.Note, want) || p.Modules["a"].Why != strings.TrimPrefix(want, "a could not be asked for: ")+p.Note[len(want):] {
t.Fatalf("the plan says %q, not words starting %q", p.Note, want)
}
// The wait has ended: a Retry made while the bus still reloads waits again, and does not fail at its first refusal.
if p.Modules["a"].BusWaitFrom != nil {
t.Fatalf("a failed module still holds the start of its wait: %v", p.Modules["a"].BusWaitFrom)
}
p.State = inventory.PlanBuilding
askModule(t.Context(), p, "a", byName)
if p.State != inventory.PlanBuilding || p.Modules["a"].State != "" || p.Modules["a"].BusWaitFrom == nil {
t.Fatalf("a retry made while the bus is silent did not wait afresh: %s %+v", p.State, p.Modules["a"])
}
}
func TestAPlanWithAModuleWaitingForTheBusIsOneTheLookComesBackTo(t *testing.T) {
p, _ := aTierOfThree()
if busWaiting(*p) {
t.Fatal("a plan whose modules are not asked yet waits for the bus")
}
now := time.Now()
p.Modules["b"] = &inventory.PlanModule{BusWaitFrom: &now}
p.Modules["a"] = &inventory.PlanModule{State: "asked"}
if got := busWaitingIn(*p); len(got) != 1 || got[0] != "b" {
t.Fatalf("the modules waiting for the bus are %v, not b", got)
}
p.State = inventory.PlanFailed
if busWaiting(*p) {
t.Fatal("a plan that is not open is waited for")
}
}
func TestAModulesOwnFaultStillFailsThePlan(t *testing.T) {
p, byName := aTierOfThree()
delete(byName, "a")
askModule(t.Context(), p, "a", byName)
if p.State != inventory.PlanFailed || !strings.Contains(p.Note, "no longer in the catalogue") {
t.Fatalf("a module gone from the catalogue did not fail the plan: %s %q", p.State, p.Note)
}
p, byName = aTierOfThree()
was := askABuild
askABuild = func(context.Context, buildSource, string, string) (string, error) {
return "", errors.New("git.example/novox/a is no repository the catalogue builds a module from")
}
t.Cleanup(func() { askABuild = was })
askModule(t.Context(), p, "a", byName)
if p.State != inventory.PlanFailed || p.Modules["a"].State != "failed" {
t.Fatalf("a module's own refusal did not fail the plan: %s %+v", p.State, p.Modules["a"])
}
}
+87
View File
@@ -0,0 +1,87 @@
package main
import (
"errors"
"fmt"
"slices"
"sort"
"strings"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/inventory"
)
// busWaitBound is how long a module whose ask the bus did not answer is asked again at every look before the plan fails
// (issue 1334).
const busWaitBound = 10 * time.Minute
// busDidNotAnswer says an ask failed because the bus's JetStream had no responder at that moment, as it has for a few seconds
// when the bus reloads its user list (novox/hq issue 353), and not because of the module: the same ask is made again.
func busDidNotAnswer(err error) bool {
if err == nil {
return false
}
if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrJetStreamNotEnabled) ||
errors.Is(err, jetstream.ErrJetStreamNotEnabled) || errors.Is(err, inventory.ErrNoChain) {
return true
}
// An error that crossed a boundary as text only.
text := err.Error()
return strings.Contains(text, "no responders available") || strings.Contains(text, "jetstream not enabled")
}
// waitForTheBus leaves a module whose ask the bus did not answer unasked, the plan building, and says so in the plan's note: the
// next look asks the tier's unasked modules again. It says false, so the caller fails the plan as for any refusal, once the bus
// has been silent for busWaitBound.
func waitForTheBus(p *inventory.Plan, state *inventory.PlanModule, name string, err error, now time.Time) bool {
if state.BusWaitFrom == nil {
state.BusWaitFrom = &now
}
if now.Sub(*state.BusWaitFrom) > busWaitBound {
return false
}
state.State, state.Why = "", ""
p.Note = fmt.Sprintf("tier %d: %s not asked: the bus did not answer (%v); asked again at the next look, failing after %s",
p.Tier, name, firstLine(err.Error()), busWaitBound)
return true
}
// busWaiting says an open plan has a module of its current tier that the bus did not answer an ask of: the look must come back
// for it, as nothing else will when no build of the tier is in flight (issue 1334).
func busWaiting(p inventory.Plan) bool {
return len(busWaitingIn(p)) > 0
}
// busWaitingIn are the modules of the plan's current tier left unasked because the bus did not answer, sorted.
func busWaitingIn(p inventory.Plan) []string {
if !p.Open() || p.Tier >= len(p.Tiers) {
return nil
}
var out []string
for _, m := range p.Tiers[p.Tier] {
if s := p.Modules[m]; s != nil && s.State == "" && s.BusWaitFrom != nil {
out = append(out, m)
}
}
sort.Strings(out)
return out
}
// busWaitWords is how a verb says modules are waiting for the bus: not asked yet, and asked again by the plan at its next look.
func busWaitWords(p inventory.Plan, modules []string) string {
return fmt.Sprintf("%s not asked yet: the bus did not answer; %s asks again at its next look", strings.Join(modules, ", "), p.ID)
}
// slicesWithout is names without those in drop, in order.
func slicesWithout(names, drop []string) []string {
var out []string
for _, n := range names {
if !slices.Contains(drop, n) {
out = append(out, n)
}
}
return out
}
+63 -4
View File
@@ -208,8 +208,11 @@ type asker struct {
// raise keeps the asker's own condition (sourceAsker): which conditions needing the operator could not be
// asked, and why. Nil raises nothing (a test that does not look).
raise func(ctx context.Context, obs []conditions.Observation) error
now func() time.Time
logf func(string, ...any)
// tell tells the operator one plain notice that needs no answer (operator-channel.tell, novox/hq ADR 0380):
// the refusal of an answer to a plan's ask (ADR 0449 §3). Nil tells nothing (a test that does not look).
tell func(ctx context.Context, headline string, lines []string) error
now func() time.Time
logf func(string, ...any)
saidNoRouter bool
saidNoGrant bool
@@ -406,7 +409,9 @@ func (a *asker) reconcile(ctx context.Context) error {
for _, r := range all {
k := partKey(r.Condition, r.Part)
if r.State == string(asks.OutcomeChosen) && now.Sub(r.Ended) < askAgainAfterAnswer {
answered[k] = r
if prior, has := answered[k]; !has || r.Ended.After(prior.Ended) {
answered[k] = r
}
}
if r.State == string(asks.OutcomeRefused) {
if prior, has := refused[k]; !has || r.Opened.After(prior.Opened) {
@@ -497,7 +502,9 @@ func (a *asker) reconcile(ctx context.Context) error {
continue // refused lately, and nothing it was refused for has changed: asked again after the wait
}
}
if r, done := answered[key]; done && sameAsked(r.Actions, p.actions) {
// **A condition raised after the answered ask ended is a new one** (novox/hq ADR 0449 §2): a retried
// plan that fails again is asked again, not held back for the hour its earlier answer holds.
if r, done := answered[key]; done && sameAsked(r.Actions, p.actions) && !c.Raised.After(r.Ended) {
if _, held := byCondition[key]; !held {
continue
}
@@ -636,6 +643,8 @@ func doesWords(act conditions.Action) string {
return "the delivery starts"
case act.Verb == broker.ControllerSeat+".plans" && act.Arguments["stop"] != "":
return "the delivery is stopped"
case act.Verb == broker.ControllerSeat+".plans" && act.Arguments["retry"] != "":
return "the failed builds are asked again"
case strings.HasSuffix(act.Verb, ".restart"):
return "its service is restarted on " + act.Machine
}
@@ -884,6 +893,7 @@ func (a *asker) Decided(ctx context.Context, body []byte) error {
ended := a.now()
if acted != nil {
outcome = "failed: " + acted.Error()
a.tellRefused(ctx, *r, act, acted)
}
r.Ended, r.Acted = ended, outcome
if _, err := a.store.Change(ctx, r.ID, func(x *asked) bool {
@@ -918,6 +928,55 @@ func (a *asker) Decided(ctx context.Context, body []byte) error {
return nil
}
// tellRefused tells the operator what became of the answer they gave to a plan's ask, when it was not done (novox/hq
// ADR 0449 §3): the controller's `plans` verb refuses a Retry, Start or Stop when the plan moved on between the ask and
// the answer, and the refusal was only recorded, so the operator who pressed the answer heard nothing. **A refusal is
// told as one, and nothing else is**: a verb that said no changed nothing, but a call that got no answer (a timeout, a
// refused grant) may have run, and is told as not known. One plain notice with the ask's headline, beside the hand act
// the caller still records. An answer to any other ask is not told here. A notice that cannot be told is said in the
// log and changes nothing.
func (a *asker) tellRefused(ctx context.Context, r asked, act conditions.Action, failure error) {
if a.tell == nil || act.Verb != broker.ControllerSeat+".plans" {
return
}
var lines []string
var refused verbRefused
if errors.As(failure, &refused) {
lines = []string{
fmt.Sprintf("Your answer, %s, could not be done: %s.", act.Label, strings.TrimRight(refusalWords(refused.words, act), ".")),
"Nothing was changed.",
}
} else {
lines = []string{
fmt.Sprintf("Your answer, %s, was given, and it is not known whether it was done.", act.Label),
"Look at the delivery before you answer again.",
}
}
if err := a.tell(ctx, r.Ask.Headline, lines); err != nil {
a.logf("the operator could NOT be told that %s was refused (ask %s): %v", act.Label, r.ID, err)
}
}
// refusalWords is a verb's refusal as the operator reads it: the plans named as "the plan" (plainRefusal), and, where
// the words still carry an identifier the operator's channel withholds, one sentence of the answer's own.
func refusalWords(words string, act conditions.Action) string {
return plainRefusalOr(words, refusalFallback(act))
}
// refusalFallback is what an answer's refusal says when its own words cannot be said: one sentence for each answer
// to a plan's ask, never another answer's.
func refusalFallback(act conditions.Action) string {
switch {
case act.Arguments["retry"] != "":
return "a retry would not run now; the details say why"
case act.Arguments["go"] != "":
return "the delivery could not be started now; the details say why"
case act.Arguments["stop"] != "":
return "the delivery could not be stopped now; the details say why"
}
return "the controller refused it; the details say why"
}
// handActWarrant is the verb an act the operator chose on a warrant is recorded under: a person's decision,
// never a repair (handActVerbs).
const handActWarrant = "warrant"
@@ -0,0 +1,141 @@
//go:build integration
package main
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/asks"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
"github.com/novox/mesh-controller/internal/testbus"
)
// 1315/AC3: the operator's Retry on a failed plan, acted on through the real callAction and the controller's own
// `plans` verb, sets the plan building again at its tier and ends its plan-failed condition (novox/hq ADR 0449 §1).
func TestAWarrantForRetryRetriesAFailedPlanInTheStoreAndEndsItsCondition(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
asked := asksRecorded(t)
planMovedHeard(t) // the bus the verb says its moves on, and the hand-act log
twoTiers(t, open)
before := time.Now().UTC().Add(-time.Hour)
ended := before.Add(time.Minute)
failed := inventory.Plan{ID: "plan-1315-retry", Repository: "novox/a", Branch: "main", Commit: "c0ffee",
Created: before, State: inventory.PlanFailed, Tier: 0, Tiers: [][]string{{"a"}, {"b"}},
Note: "a failed to build in tier 0",
Delivery: &inventory.PlanDelivery{},
Modules: map[string]*inventory.PlanModule{"a": {State: "failed", AskedAt: &before, Build: "build-1",
Why: link.KilledByHand}},
Times: &inventory.PlanTimes{Cut: &before, Started: &before, Ended: &ended, Class: inventory.ClassLeaf}}
if err := open.inventory.SavePlan(ctx, &failed); err != nil {
t.Fatal(err)
}
conn, err := nats.Connect(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
defer conn.Close()
// The controller's seat answers `plans` as it does: by running the verb.
sub, err := conn.Subscribe(link.SeatToolSubject("mesh-controller", "plans"), func(m *nats.Msg) {
var args map[string]string
_ = json.Unmarshal(m.Data, &args)
answer := link.Answer{Result: json.RawMessage(`"ok"`)}
if len(args) != 1 || args["retry"] == "" {
answer = link.Answer{Error: "the retry was asked with more than the plan"}
} else if err := plansCommand(context.Background(), []string{"retry", args["retry"]}); err != nil {
answer = link.Answer{Error: err.Error()}
}
body, _ := json.Marshal(answer)
_ = m.Respond(body)
})
if err != nil {
t.Fatal(err)
}
defer func() { _ = sub.Unsubscribe() }()
// What S23 raises for it, read from the store as the watchdog reads it.
raised := func(context.Context) ([]conditions.Condition, error) {
facts, err := gatherFailedPlans(ctx, open.inventory, time.Now())
if err != nil {
return nil, err
}
var out []conditions.Condition
for _, o := range watchPlansFailed(&signalFacts{now: time.Now(), failed: facts}) {
out = append(out, conditions.Condition{Key: o.Key(), Kind: o.Kind, Severity: o.Severity, Headline: o.Headline,
Explanation: conditions.Verdict(o.Needs, o.Explanation), Needs: o.Needs, Actions: o.Actions})
}
return out, nil
}
now := time.Now()
store := memAskedStore{}
var sent []published
var acts []link.HandAct
a := &asker{open: raised, store: store,
silence: func(context.Context, string, time.Duration, string, string) error { return nil },
publish: func(_ context.Context, subject string, body []byte, id string) error {
sent = append(sent, published{subject, id, body})
return nil
},
call: callAction(conn),
record: func(_ context.Context, act link.HandAct) error { acts = append(acts, act); return nil },
now: func() time.Time { return now },
logf: t.Logf}
cs, _ := raised(ctx)
if len(cs) != 1 || len(cs[0].Actions) != 1 {
t.Fatalf("the failed plan raised %+v", cs)
}
if err := a.reconcile(ctx); err != nil {
t.Fatal(err)
}
var w asks.Warrant
for _, r := range store {
for _, o := range r.Ask.Options {
if o.Label == "Retry" {
w = asks.Warrant{Ask: r.ID, Asker: "mesh-controller", About: r.Condition, Outcome: asks.OutcomeChosen,
Option: o.ID, Label: o.Label, Level: o.Level, Channel: "telegram", Proofs: []string{"P1"}, At: now,
AskDigest: r.Ask.Digest(),
By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}}
}
}
}
if w.Ask == "" {
t.Fatalf("no ask offers Retry: %+v", store)
}
body, _ := json.Marshal(w)
if err := a.Decided(ctx, body); err != nil {
t.Fatal(err)
}
if err := a.Decided(ctx, body); err != nil { // heard again: nothing more
t.Fatal(err)
}
p, err := open.inventory.PlanByID(ctx, failed.ID)
if err != nil {
t.Fatal(err)
}
m := p.Modules["a"]
if p.State != inventory.PlanBuilding || p.Tier != 0 || m.State != "asked" || m.Build == "" || m.Build == "build-1" {
t.Fatalf("after the warrant the plan is %s at tier %d and a is %+v", p.State, p.Tier, m)
}
if len(*asked) != 1 {
t.Fatalf("the build was asked %d times: %v", len(*asked), *asked)
}
if got := store[w.Ask]; got.Acted != "done" {
t.Fatalf("the ask ended %q", got.Acted)
}
if len(acts) != 1 || acts[0].Verb != handActWarrant || !strings.Contains(strings.Join(acts[0].Args, " "), "retry="+failed.ID) {
t.Fatalf("the warrant was recorded as %+v", acts)
}
if cs, _ = raised(ctx); len(cs) != 0 {
t.Fatalf("the plan-failed condition did not end: %+v", cs)
}
}
+225
View File
@@ -77,6 +77,15 @@ type askerRig struct {
silenced []string
acts []link.HandAct
now time.Time
// refuses, when set, is what every verb the asker calls answers; told is every notice the operator was told.
refuses error
told []told
}
// told is one plain notice the asker told the operator.
type told struct {
headline string
lines []string
}
func newAskerRig(t *testing.T) *askerRig {
@@ -102,6 +111,10 @@ func newAskerRig(t *testing.T) *askerRig {
call: func(_ context.Context, a conditions.Action, args map[string]string) error {
raw, _ := json.Marshal(args)
r.called = append(r.called, a.Verb+"@"+a.Machine+" "+string(raw))
return r.refuses
},
tell: func(_ context.Context, headline string, lines []string) error {
r.told = append(r.told, told{headline, lines})
return nil
},
record: func(_ context.Context, act link.HandAct) error { r.acts = append(r.acts, act); return nil },
@@ -1013,3 +1026,215 @@ func TestTheServingAskerJudgesTheGrantForReal(t *testing.T) {
t.Errorf("held %v (%q, %v) with nothing to read", held, why, err)
}
}
// failedPlanCondition is the condition S23 raises for a failed delivery plan, as the keeper holds it.
func failedPlanCondition(plan string) conditions.Condition {
p := aFailedPlan(plan, nil)
f := calm(time.Date(2026, 10, 19, 18, 30, 0, 0, time.UTC))
f.failed = []failedPlanFacts{failedFactsFor(p)}
o := watchPlansFailed(f)[0]
return conditions.Condition{Key: o.Key(), Kind: o.Kind, Severity: o.Severity, Headline: o.Headline,
Explanation: conditions.Verdict(o.Needs, o.Explanation), Needs: o.Needs, Actions: o.Actions}
}
// 1315/AC1: a failed plan's Retry is asked at the approve level, in the words "the failed builds are asked again".
func TestAFailedPlansRetryIsAskedAtApproveInItsOwnWords(t *testing.T) {
r := newAskerRig(t)
c := failedPlanCondition("plan-1792433168676870614")
r.open = []conditions.Condition{c}
if err := r.a.reconcile(context.Background()); err != nil {
t.Fatal(err)
}
sent := r.asksSent(t)
if len(sent) != 1 || len(sent[0].Options) != 1 {
t.Fatalf("asked %+v", sent)
}
o := sent[0].Options[0]
if o.Label != "Retry" || o.Level != asks.Approve || o.Does != "the failed builds are asked again" {
t.Errorf("the option is %+v", o)
}
if err := sent[0].Check(r.now); err != nil {
t.Errorf("the ask is refused as made: %v", err)
}
if strings.Contains(sent[0].Explanation, "mesh MCP server") {
t.Errorf("the ask still says where else to answer: %q", sent[0].Explanation)
}
}
// 1315/AC2: the warrant calls plans with retry=<plan> and nothing else, once, and is recorded as the hand act;
// the same warrant heard again does nothing.
func TestAWarrantForRetryCallsPlansRetryOnceAndIsRecorded(t *testing.T) {
plan := "plan-1792433168676870614"
r := newAskerRig(t)
c := failedPlanCondition(plan)
r.open = []conditions.Condition{c}
_ = r.a.reconcile(context.Background())
w := r.warrantFor(t, c.Key, "Retry")
answerWith(t, r, w)
answerWith(t, r, w)
if want := `mesh-controller.plans@ {"retry":"` + plan + `"}`; len(r.called) != 1 || r.called[0] != want {
t.Fatalf("called %v, want exactly %s", r.called, want)
}
if len(r.acts) != 1 {
t.Fatalf("hand-acts %+v", r.acts)
}
act := r.acts[0]
if act.Verb != handActWarrant || strings.Join(act.Args, " ") != "mesh-controller.plans retry="+plan ||
act.Ask != w.Ask || act.Condition != c.Key || act.Outcome != "done" || act.Cause != conditions.CauseOperatorAnswer {
t.Errorf("the hand-act is %+v", act)
}
if got := r.store[w.Ask]; got.Acted != "done" {
t.Errorf("kept %+v", got)
}
}
// 1316/AC1 and AC2: a Retry or a Stop on a plan that moved on is refused by `plans`; the operator who pressed it is
// told once, with the ask's headline and the refusal's words, and the hand act records the failure as before.
func TestARefusedAnswerToAPlansAskIsToldOnceAndStillRecorded(t *testing.T) {
plan := "plan-1792433168676870614"
waiting := conditions.Condition{Key: "plan." + plan + ".waiting", Severity: conditions.Urgent,
Headline: "openrazer delivery waiting to start", Needs: "start it, or stop it.",
Explanation: "Needs you: start it, or stop it.", Actions: waitingActions(plan, conditions.Urgent)}
for _, tc := range []struct {
name string
c conditions.Condition
label string
}{
{"retry", failedPlanCondition(plan), "Retry"},
{"stop", waiting, "Stop"},
} {
t.Run(tc.name, func(t *testing.T) {
r := newAskerRig(t)
r.refuses = verbRefused{verb: "mesh-controller.plans", words: plan + " is done; there is nothing to retry"}
r.open = []conditions.Condition{tc.c}
_ = r.a.reconcile(context.Background())
w := r.warrantFor(t, tc.c.Key, tc.label)
answerWith(t, r, w)
answerWith(t, r, w) // heard again: told once
if len(r.told) != 1 {
t.Fatalf("told %+v", r.told)
}
got := r.told[0]
if got.headline != tc.c.Headline {
t.Errorf("the headline is %q, not the ask's %q", got.headline, tc.c.Headline)
}
said := strings.Join(got.lines, " ")
if !strings.Contains(said, "the plan is done; there is nothing to retry") || !strings.Contains(said, tc.label) {
t.Errorf("the lines do not carry the answer and the refusal's words: %q", said)
}
if why, ok := conditions.Plain(said); !ok {
t.Errorf("the notice is not plain, so the channel would withhold it: %s: %q", why, said)
}
if len(r.acts) != 1 || !strings.HasPrefix(r.acts[0].Outcome, "failed: ") {
t.Errorf("the failure is not recorded as before: %+v", r.acts)
}
if kept := r.store[w.Ask]; !strings.HasPrefix(kept.Acted, "failed: ") {
t.Errorf("kept %+v", kept)
}
})
}
}
// 1316/AC3: an answer that is done tells nothing; and a refusal of an answer that is no plan's is not told here.
func TestADoneAnswerTellsNothingAndOnlyAPlansRefusalIsTold(t *testing.T) {
r := newAskerRig(t)
c := failedPlanCondition("plan-1792433168676870614")
r.open = []conditions.Condition{c}
_ = r.a.reconcile(context.Background())
answerWith(t, r, r.warrantFor(t, c.Key, "Retry"))
if len(r.told) != 0 {
t.Fatalf("a retry that was done told %+v", r.told)
}
r = newAskerRig(t)
r.refuses = verbRefused{verb: "node-service-manager.restart", words: "unit not found"}
module := conditions.Condition{Key: "module.openrazer.g14.unhealthy", Severity: conditions.Warning,
Headline: "openrazer not working on g14", Needs: "restart its service openrazer-daemon on g14.",
Explanation: "Needs you: restart it.", Actions: []conditions.Action{{Label: "Restart",
Verb: "node-service-manager.restart", Machine: "g14", Level: conditions.LevelApprove,
Arguments: map[string]string{"unit": "openrazer-daemon.service", "scope": "user"}}}}
r.open = []conditions.Condition{module}
_ = r.a.reconcile(context.Background())
answerWith(t, r, r.warrantFor(t, module.Key, "Restart"))
if len(r.told) != 0 || len(r.acts) != 1 {
t.Fatalf("a refused restart told %+v, acts %+v", r.told, r.acts)
}
}
// 1316, review round 1: a call that got no answer may have run, so it is told as not known, never as "nothing was
// changed"; and the sentence a refusal falls back to fits the answer given.
func TestAnAnswerThatGotNoAnswerIsToldAsNotKnownNotAsRefused(t *testing.T) {
plan := "plan-1792433168676870614"
r := newAskerRig(t)
r.refuses = errors.New("mesh-controller did not answer plans within 30s")
c := failedPlanCondition(plan)
r.open = []conditions.Condition{c}
_ = r.a.reconcile(context.Background())
answerWith(t, r, r.warrantFor(t, c.Key, "Retry"))
if len(r.told) != 1 {
t.Fatalf("told %+v", r.told)
}
said := strings.Join(r.told[0].lines, " ")
if !strings.Contains(said, "not known whether") || strings.Contains(said, "Nothing was changed") {
t.Errorf("a call that may have run is told as %q", said)
}
if why, ok := conditions.Plain(said); !ok {
t.Errorf("not plain: %s: %q", why, said)
}
}
func TestEachAnswersRefusalFallsBackToItsOwnSentence(t *testing.T) {
plan := "plan-1792433168676870614"
for _, tc := range []struct {
act conditions.Action
want string
}{
{retryAction(plan), "a retry would not run now"},
{conditions.Action{Label: "Start", Verb: "mesh-controller.plans", Arguments: map[string]string{"go": plan}}, "could not be started now"},
{conditions.Action{Label: "Stop", Verb: "mesh-controller.plans", Arguments: map[string]string{"stop": plan}}, "could not be stopped now"},
} {
// Words the operator's channel withholds: a hash and a dotted name.
got := refusalWords("build 48bda475dunst at mesh.seat.x refused", tc.act)
if !strings.Contains(got, tc.want) {
t.Errorf("%s: %q, want %q", tc.act.Label, got, tc.want)
}
if _, ok := conditions.Plain(got); !ok {
t.Errorf("%s: the fallback is not plain: %q", tc.act.Label, got)
}
}
}
// 1314/AC6, ADR 0449 §2: a retried plan that fails again within the hour of the answer is asked again, because the
// condition that comes back is a new one, raised after the answered ask ended; the same condition still open is not.
func TestARetriedPlanThatFailsAgainWithinTheHourIsAskedAgain(t *testing.T) {
r := newAskerRig(t)
c := failedPlanCondition("plan-1792433168676870614")
c.Raised = r.now.Add(-time.Minute)
r.open = []conditions.Condition{c}
_ = r.a.reconcile(context.Background())
answerWith(t, r, r.warrantFor(t, c.Key, "Retry"))
// Still open right after the answer: not asked again at once.
r.now = r.now.Add(time.Minute)
_ = r.a.reconcile(context.Background())
if n := len(r.asksSent(t)); n != 1 {
t.Fatalf("asked again while the same condition stayed open: %d", n)
}
// The retry ran, the plan left failed and the condition ended.
r.now = r.now.Add(time.Minute)
r.open = nil
_ = r.a.reconcile(context.Background())
// It fails again, ten minutes after the answer: a condition raised after the answered ask ended.
r.now = r.now.Add(8 * time.Minute)
again := failedPlanCondition("plan-1792433168676870614")
again.Raised = r.now
r.open = []conditions.Condition{again}
if err := r.a.reconcile(context.Background()); err != nil {
t.Fatal(err)
}
if n := len(r.asksSent(t)); n != 2 {
t.Fatalf("the second failure was asked %d times in all, want 2", n)
}
if len(r.called) != 1 {
t.Fatalf("something retried without an answer: %v", r.called)
}
}
+24 -1
View File
@@ -135,6 +135,12 @@ func (b busAsked) All(ctx context.Context) ([]asked, error) {
return out, nil
}
// verbRefused is a verb's own refusal of an act: the seat answered, and said no. Any other error of callAction (a bus
// that did not answer, a grant that is refused) is not one: the verb may have run.
type verbRefused struct{ verb, words string }
func (e verbRefused) Error() string { return e.verb + " refused: " + e.words }
// callAction performs an action's verb as the controller, through the grant that names it.
func callAction(conn *nats.Conn) func(ctx context.Context, a conditions.Action, args map[string]string) error {
return func(ctx context.Context, a conditions.Action, args map[string]string) error {
@@ -168,7 +174,23 @@ func callAction(conn *nats.Conn) func(ctx context.Context, a conditions.Action,
return err
}
if answer.Error != "" {
return fmt.Errorf("%s refused: %s", a.Verb, answer.Error)
return verbRefused{verb: a.Verb, words: answer.Error}
}
return nil
}
}
// tellThrough tells the operator one plain notice through the operator channel's `tell`, over the controller's own
// connection and grant (novox/hq ADR 0380); a holder's refusal is an error naming it.
func tellThrough(conn *nats.Conn) func(ctx context.Context, headline string, lines []string) error {
return func(ctx context.Context, headline string, lines []string) error {
answer, err := link.AskMeshSeatTool(ctx, conn, "operator-channel", "tell",
map[string]any{"headline": headline, "lines": lines}, tellWithin)
if err != nil {
return err
}
if answer.Error != "" {
return fmt.Errorf("operator-channel.tell refused: %s", answer.Error)
}
return nil
}
@@ -576,6 +598,7 @@ func servingAsker(open *stores, conn *nats.Conn, js jetstream.JetStream, keeper
_, err := link.RecordHandAct(ctx, conn, act)
return err
},
tell: tellThrough(conn),
routerRecord: routerRecordOf(conn, open.inventory),
routerHere: routerHereIn(open.inventory),
grantHeld: grantHeldIn(open, conn),
+3 -1
View File
@@ -174,7 +174,9 @@ func TestAPlanGoesOnWithoutAMachineAwayWhenItsTierIsSent(t *testing.T) {
}}
first, stopFirst := context.WithCancel(ctx)
defer stopFirst()
startRolloutEvents(first, open, js)
waitFirst := startRolloutEvents(first, open, js)
// Ends with the test before the cleanups restore what the look reads, which run after it (issue 1338).
t.Cleanup(func() { stopFirst(); waitFirst() })
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
+33 -23
View File
@@ -4,6 +4,8 @@ import (
"context"
"encoding/json"
"fmt"
"io"
"os"
"path"
"regexp"
"runtime/debug"
@@ -141,18 +143,26 @@ func sourceOwner(repository string) string {
// PullUpdated decides a pull request's merge check, and asks for it when there is something to run.
func (f following) PullUpdated(ctx context.Context, p link.PullUpdated) error {
_, err := f.pullUpdated(ctx, p, os.Stdout)
return err
}
// pullUpdated is PullUpdated saying what it did to w. It returns the id of the check it asked of the build seat, or
// "" when it said a verdict itself and asked nothing. A command that answers JSON on standard output (`delivery
// check`) gives it standard error, so its output stays the document alone (novox/hq issue 1311).
func (f following) pullUpdated(ctx context.Context, p link.PullUpdated, w io.Writer) (string, error) {
inv := f.open.inventory
entries, err := inv.Catalogued(ctx)
if err != nil {
return err
return "", err
}
read, err := readForPlanning(ctx, inv)
if err != nil {
return err
return "", err
}
edges, err := inv.Dependencies(ctx)
if err != nil {
return err
return "", err
}
scope := pullScope(p, entries, read, edges)
// **A cycle of edges fails the check, naming every module and edge** (novox/hq ADR 0419 clause 6), a lone pull
@@ -161,12 +171,12 @@ func (f following) PullUpdated(ctx context.Context, p link.PullUpdated) error {
Head: p.Commit, Number: p.Number, Paths: p.Paths, PathsTruncated: p.PathsTruncated, Removed: p.Removed, BaseSHA: p.BaseSHA}
if failing, cyclic, err := cycleVerdict(ctx, "", []orderMember{alone},
map[string]orderReach{alone.ID: {Moved: scope.Modules}}, edges, entries); err != nil {
return err
return "", err
} else if cyclic {
failing.Commit, failing.Trunk = p.Commit, twelve(p.BaseSHA)
fmt.Printf("%s/%s#%d (%.8s): %s\n", p.Owner, p.Repo, p.Number, p.Commit, failing.Summary)
sayChecked(ctx, failing)
return nil
fmt.Fprintf(w, "%s/%s#%d (%.8s): %s\n", p.Owner, p.Repo, p.Number, p.Commit, failing.Summary)
sayChecked(ctx, w, failing)
return "", nil
}
// **A member of a group whose gate is the composed check's runs its repository's check alone** (novox/hq ADR 0403
// clause 1): mesh-delivery is asked once, and says so on the group's line. Told nothing, the pull request is
@@ -179,7 +189,7 @@ func (f following) PullUpdated(ctx context.Context, p link.PullUpdated) error {
u, err := inv.UpgradeOf(ctx, module)
return u, err == nil
})
fmt.Print(planText(plan))
fmt.Fprint(w, planText(plan))
direct := directVerdict(p, plan)
if composed {
scope.Modules, scope.Dependents, scope.New, scope.Manifests = nil, nil, nil, nil
@@ -187,19 +197,19 @@ func (f following) PullUpdated(ctx context.Context, p link.PullUpdated) error {
}
switch {
case !scope.gated() && !scope.Mesh:
fmt.Printf("%s/%s#%d (%.8s): %s, and the repository is not the mesh's: said, nothing run\n",
fmt.Fprintf(w, "%s/%s#%d (%.8s): %s, and the repository is not the mesh's: said, nothing run\n",
p.Owner, p.Repo, p.Number, p.Commit, noModuleTouched)
sayChecked(ctx, direct)
return nil
sayChecked(ctx, w, direct)
return "", nil
case !scope.gated() && p.MergeCheckSaid && !p.MergeCheck:
direct.RepoCheck = &link.CheckLayer{Verdict: "warning", Summary: noMergeCheck}
fmt.Printf("%s/%s#%d (%.8s): %s; %s\n", p.Owner, p.Repo, p.Number, p.Commit, noModuleTouched, noMergeCheck)
sayChecked(ctx, direct)
return nil
fmt.Fprintf(w, "%s/%s#%d (%.8s): %s; %s\n", p.Owner, p.Repo, p.Number, p.Commit, noModuleTouched, noMergeCheck)
sayChecked(ctx, w, direct)
return "", nil
}
request, err := checkRequestFor(ctx, f.open, p, scope, entries)
if err != nil {
return err
return "", err
}
request.Check.Plan = &plan
request.Check.Composed = composed
@@ -208,11 +218,11 @@ func (f following) PullUpdated(ctx context.Context, p link.PullUpdated) error {
seat := buildSeatHeld(ctx)
ask, err := askOverOn(seat)
if err != nil {
return err
return "", err
}
defer ask.Close()
if err := ask.Ask(ctx, request); err != nil {
return err
return "", err
}
// **A newer request replaces the one still waiting, and a newer key stops the one running** (novox/hq ADR 0406 §3, §4,
// issue 974): the check queue keeps one request a pull request, and a running check watches its key. Nothing is asked
@@ -226,9 +236,9 @@ func (f following) PullUpdated(ctx context.Context, p link.PullUpdated) error {
strings.Join(append(append([]string{}, scope.Modules...), prefixedAll("new:", scope.New)...), ", "),
scope.Width, scope.Tiers)
}
fmt.Printf("%s/%s#%d (%.8s): asked %s to check it before it merges — %s — as %s\n", p.Owner, p.Repo, p.Number,
fmt.Fprintf(w, "%s/%s#%d (%.8s): asked %s to check it before it merges — %s — as %s\n", p.Owner, p.Repo, p.Number,
p.Commit, seat, what, request.ID)
return nil
return request.ID, nil
}
// composedWords is what a group member's own verdict says of its gate layer (novox/hq ADR 0403 clause 1).
@@ -528,7 +538,7 @@ func checked(ctx context.Context, result link.BuildResult) {
fmt.Printf("%s: stopped, its pull request's key moved on; the newer check will judge\n", result.ID)
return
}
sayChecked(ctx, checkedOf(result))
sayChecked(ctx, os.Stdout, checkedOf(result))
}
// checkedOf is what a check's outcome says: each layer, and an error — never a pass — for a check that
@@ -644,7 +654,7 @@ func prefixedAll(prefix string, items []string) []string {
}
// sayChecked says a merge check's verdict on the bus, where the forge's holder hears it.
func sayChecked(ctx context.Context, c link.Checked) {
func sayChecked(ctx context.Context, w io.Writer, c link.Checked) {
repo := "none"
if c.RepoCheck != nil {
repo = strings.ToUpper(c.RepoCheck.Verdict) + " — " + c.RepoCheck.Summary
@@ -653,7 +663,7 @@ func sayChecked(ctx context.Context, c link.Checked) {
if c.Trunk != "" {
judged = " merged into trunk " + c.Trunk
}
fmt.Printf("%s: %s/%s#%d at %.8s%s checked on %s: gate %s — %s; repository %s\n", c.ID, c.Owner, c.Repo, c.Number,
fmt.Fprintf(w, "%s: %s/%s#%d at %.8s%s checked on %s: gate %s — %s; repository %s\n", c.ID, c.Owner, c.Repo, c.Number,
c.Commit, judged, orSomewhere(c.On), strings.ToUpper(c.Verdict), c.Summary, repo)
if checkEvents == nil {
return
@@ -665,7 +675,7 @@ func sayChecked(ctx context.Context, c link.Checked) {
stating, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer stop()
if err := checkEvents.PublishSeatEvent(stating, link.MeshControllerSeat, link.KeyChecked, body); err != nil {
fmt.Printf("%s: the verdict could not be said, so the pull request is not told it: %v\n", c.ID, err)
fmt.Fprintf(w, "%s: the verdict could not be said, so the pull request is not told it: %v\n", c.ID, err)
}
}
+31 -5
View File
@@ -902,10 +902,14 @@ func deliveryCommand(ctx context.Context, args []string) error {
if *group == "" && len(members) == 1 {
// One head, checked again as the forge's announcement would have it (a recheck): the controller's
// own path for a pull request, run because the delivery's owner asked.
if err := sayingOnTheBus(ctx, inv, func() error { return (following{open: open, oneShot: true}).PullUpdated(ctx, members[0].pull()) }); err != nil {
return err
}
return answer(map[string]any{"rechecked": members[0].ID})
return recheckOneHead(members[0].ID, answer, func() (string, error) {
var asked string
err := sayingOnTheBus(ctx, inv, func() (err error) {
asked, err = (following{open: open, oneShot: true}).pullUpdated(ctx, members[0].pull(), os.Stderr)
return err
})
return asked, err
})
}
if *group == "" || len(members) < 2 {
return errors.New("delivery check --group <id> --members <json> (two heads or more), or --members <one head>")
@@ -1156,7 +1160,7 @@ func askGroupCheck(ctx context.Context, open *stores, group string, members []or
if failing, cyclic, err := cycleVerdict(ctx, group, members, reach, edges, entries); err != nil {
return "", err
} else if cyclic {
sayChecked(ctx, failing)
sayChecked(ctx, os.Stderr, failing)
return failing.ID, nil
}
primary := groupPrimary(members, reach)
@@ -1376,3 +1380,25 @@ func overNATS(conn *nats.Conn) (link.Bus, error) {
}
return link.OverNATS{Conn: conn, JS: js}, nil
}
// recheckAnswer is what `delivery check` answers for one head: the id of the check asked of the build seat (empty when
// the controller said the verdict itself and asked nothing), which mesh-delivery reads, and the head rechecked.
func recheckAnswer(asked, member string) map[string]any {
return map[string]any{"asked": asked, "rechecked": member}
}
// recheckOneHead runs a one-head `delivery check` and answers it as data. Standard output is the answer's JSON alone,
// or the seat verb reads no answer from it (novox/hq issue 1311): while the check runs, whatever it prints to standard
// output (the controller's own path prints its prose there, as do the stores it reads) goes to standard error.
func recheckOneHead(member string, answer func(any) error, ask func() (string, error)) error {
real := os.Stdout
os.Stdout = os.Stderr
asked, err := func() (string, error) {
defer func() { os.Stdout = real }()
return ask()
}()
if err != nil {
return err
}
return answer(recheckAnswer(asked, member))
}
@@ -0,0 +1,85 @@
package main
import (
"bytes"
"context"
"encoding/json"
"fmt"
"os"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/link"
)
// `delivery check` with one head answers the id of the check it asked, as data (novox/hq issue 1311): the prose
// the check says goes to standard error, so standard output is the document alone and answerOf reads it.
func TestDeliveryCheckOfOneHeadAnswersTheRequestID(t *testing.T) {
argv := []string{"delivery", "check", "--members", "[]"}
doc, err := json.Marshal(recheckAnswer("check-42", "novox/x@abc"))
if err != nil {
t.Fatal(err)
}
// Seen failing first: the prose the command used to print on standard output before the document.
polluted := append([]byte("novox/x#7 (abc): asked the build seat as check-42\n"), doc...)
if got := answerOf(argv, polluted, "", true); got.Answer != nil {
t.Fatalf("prose before the document should leave no answer; got %v", got.Answer)
}
got := answerOf(argv, doc, "novox/x#7 (abc): asked the build seat as check-42\n", true)
m, ok := got.Answer.(map[string]any)
if !ok || m["asked"] != "check-42" || m["rechecked"] != "novox/x@abc" {
t.Fatalf("the answer is %v; want asked check-42 and rechecked novox/x@abc", got.Answer)
}
if !strings.Contains(got.Output, "asked the build seat") {
t.Fatalf("the prose is still said in the text: %q", got.Output)
}
}
// The text of the other callers does not change: sayChecked writes the same line to the writer it is given, and
// the controller's own path still gives PullUpdated standard output.
func TestSayCheckedWritesTheSameLineToItsWriter(t *testing.T) {
var w bytes.Buffer
sayChecked(context.Background(), &w, link.Checked{ID: "c1", Owner: "novox", Repo: "x", Number: 7, Commit: "0123456789abcdef",
Verdict: "pass", Summary: "fine"})
want := "c1: novox/x#7 at 01234567 checked on a machine that did not say which: gate PASS — fine; repository none\n"
if w.String() != want {
t.Fatalf("sayChecked said %q; want %q", w.String(), want)
}
}
// Whatever the check prints to standard output while it runs (checks.go namedForTheCheck, the inventory's pointer
// repair) must not reach the answer's standard output: it is exactly one JSON document (novox/hq issue 1311).
func TestOneHeadRecheckLeavesStandardOutputTheAnswerAlone(t *testing.T) {
r, w, err := os.Pipe()
if err != nil {
t.Fatal(err)
}
e, ew, err := os.Pipe()
if err != nil {
t.Fatal(err)
}
realOut, realErr := os.Stdout, os.Stderr
os.Stdout, os.Stderr = w, ew
runErr := recheckOneHead("novox/x@abc", func(v any) error { return json.NewEncoder(os.Stdout).Encode(v) },
func() (string, error) {
fmt.Println(" the earlier checks' builds could not be read, so the check builds")
fmt.Fprintln(os.Stdout, "repaired a pointer")
return "check-42", nil
})
os.Stdout, os.Stderr = realOut, realErr
w.Close()
ew.Close()
if runErr != nil {
t.Fatal(runErr)
}
var out, errOut bytes.Buffer
_, _ = out.ReadFrom(r)
_, _ = errOut.ReadFrom(e)
var doc map[string]any
if err := json.Unmarshal(out.Bytes(), &doc); err != nil || doc["asked"] != "check-42" || doc["rechecked"] != "novox/x@abc" {
t.Fatalf("standard output %q is not the answer alone (%v)", out.String(), err)
}
if !strings.Contains(errOut.String(), "repaired a pointer") {
t.Fatalf("the prose should be on standard error: %q", errOut.String())
}
}
+4
View File
@@ -72,6 +72,10 @@ var handActVerbs = []handActVerb{
// An act the operator chose on a warrant (novox/hq ADR 0259): asked by the controller, answered on a
// channel that proved who answered, performed by the controller as itself.
{Verb: handActWarrant, Decision: "the operator chose it, answering what the controller asked (ADR 0259)"},
// A seat passed by the controller to its only claimant when its holder on record left (novox/hq ADR 0449
// §4): the mesh working as decided. A person's own `seat --to` is not recorded under this cause.
{Verb: "seat", Decision: "the controller passes a seat held once to its only claimant when its recorded " +
"holder is assigned nowhere (ADR 0449 §4)", DecidedFor: []string{"seat-passed"}},
{Verb: "broker consumer-reset"},
// Silencing the same condition twice says the condition, or what it watches, wants mending — unless
// it is the operator's answer on a notification: a decision to live with it (novox/hq ADR 0258).
+5
View File
@@ -429,6 +429,11 @@ var plainWordings = map[string]func(conditions.Observation) words{
"out) has not let it start yet.",
Resolved: "Resolved: the delivery is no longer waiting"}
}),
kindPlanFailed: worded(func(o conditions.Observation) words {
return words{Headline: "A delivery failed",
Explanation: "A delivery did not finish. Nothing is tried again until you say so.",
Resolved: "Resolved: the delivery is no longer failed"}
}),
"ask-lost": worded(func(o conditions.Observation) words {
return words{Headline: "A build has no result",
Explanation: "A build was asked for and never answered. It is asked again with the next delivery.",
+18 -2
View File
@@ -285,12 +285,28 @@ func theRestOfTheMesh(ctx context.Context, inv *inventory.Inventory,
assigned []string
}
var others []candidate
// **Every machine's assignments, before the holdings are read by any resolution** (novox/hq ADR 0449 §4): a
// seat held once whose holder on record is assigned on no machine needs the whole mesh to be known, and it
// passes to its only claimant here, in the composition that finds it.
everyAssigned := map[string][]string{}
var assignedErr error
for _, n := range nodes {
theirs, err := inv.Assigned(ctx, n.Name)
if err != nil {
assignedErr = errors.Join(assignedErr, fmt.Errorf("%s: %w", n.Name, err))
continue
}
if len(theirs) > 0 {
everyAssigned[n.Name] = theirs
}
}
holdings = theMeshPassesSeatsOn(ctx, inv, shelf, everyAssigned, assignedErr, holdings)
for _, n := range nodes {
if n.Name == exclude {
continue
}
theirs, err := inv.Assigned(ctx, n.Name)
if err != nil || len(theirs) == 0 {
theirs := everyAssigned[n.Name]
if len(theirs) == 0 {
continue
}
caps, _ := inv.ProfileOf(ctx, n.Name)
+227
View File
@@ -0,0 +1,227 @@
package main
import (
"context"
"errors"
"fmt"
"regexp"
"sort"
"strings"
"time"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
)
// failedModule is one module of a failed plan that failed to build, where it runs and what it said.
type failedModule struct {
module string
machines []string
why string
}
// failedPlanFacts is one failed delivery plan asked of the operator (S23, novox/hq ADR 0449 §1).
type failedPlanFacts struct {
id, repository, commit string
modules []string
failed []failedModule
// note is where the plan stopped when no module failed to build (a rollout stopped at its first machine).
note string
// since is the plan's last change: when it failed.
since time.Time
// refusal is why `plans retry` would not run it, empty when it would.
refusal string
// passed is set when a newer plan of its repository and branch is done and the retry is refused for that
// (novox/hq ADR 0451): the condition asks nothing.
passed bool
}
// machines are the machines the failed modules run on, sorted, each once.
func (p failedPlanFacts) machines() []string {
seen := map[string]bool{}
var out []string
for _, m := range p.failed {
for _, n := range m.machines {
if !seen[n] {
seen[n] = true
out = append(out, n)
}
}
}
sort.Strings(out)
return out
}
// failureSaid is what failed, in the mesh's own words, for the condition's evidence.
func (p failedPlanFacts) failureSaid() string {
var parts []string
for _, m := range p.failed {
line := m.module
if len(m.machines) > 0 {
line += " on " + strings.Join(m.machines, ", ")
}
if m.why != "" {
line += ": " + oneLine(m.why)
}
parts = append(parts, line)
}
if len(parts) == 0 {
return "stopped: " + oneLine(p.note)
}
return strings.Join(parts, "; ")
}
// refusalSaid is why a retry would not run, for the evidence.
func (p failedPlanFacts) refusalSaid() string {
if p.refusal == "" {
return ""
}
return "; a retry would not run: " + oneLine(p.refusal)
}
// retryAction is the one answer a failed plan offers: the controller's `plans` verb with `retry` naming the plan,
// at the level approve. Only that argument: `plans` refuses a retry that carries others (seatverbs).
func retryAction(plan string) conditions.Action {
return conditions.Action{Label: "Retry", Verb: "mesh-controller.plans", Level: conditions.LevelApprove,
Arguments: map[string]string{"retry": plan}}
}
var (
planIDWord = regexp.MustCompile(`\bplan-[0-9]+\b`)
walkWd = regexp.MustCompile(`\bwalks\b`)
walkOne = regexp.MustCompile(`\bwalk\b`)
)
// plainRefusal is a retry's refusal in words the operator reads: the plans named as "the plan", a walk said as a
// delivery. Whatever still breaks the plain rule (a hash, a dotted name) is left to the condition's evidence, and the
// words say only that a retry would not run.
func plainRefusal(refusal string) string {
return plainRefusalOr(refusal, "a retry would not run now; the details say why")
}
// plainRefusalOr is plainRefusal with the sentence to say when the refusal's own words cannot be said plainly.
func plainRefusalOr(refusal, fallback string) string {
r := planIDWord.ReplaceAllString(refusal, "the plan")
r = walkWd.ReplaceAllString(r, "deliveries")
r = walkOne.ReplaceAllString(r, "delivery")
r = strings.TrimSpace(r)
if _, ok := conditions.Plain(r); !ok {
return fallback
}
return r
}
// planFailedWords are the plain words of a failed plan: which modules failed and where, what they said when that
// is plain, and Retry when a retry would run, else the refusal's words.
func planFailedWords(p failedPlanFacts) (headline, explanation, needs string, actions []conditions.Action, resolved string) {
name := deliveryName(p.modules, p.repository)
headline = conditions.Capital(name) + " failed"
what := deliveryWhat(p.modules, p.repository)
var parts []string
for i, m := range p.failed {
if i == 2 {
parts = append(parts, fmt.Sprintf("%d more", len(p.failed)-2))
break
}
line := m.module
if len(m.machines) > 0 {
line += " for " + readableList(m.machines)
}
if w := strings.TrimRight(cutWords(oneLine(m.why), 110), ".;:, "); w != "" {
if _, ok := conditions.Plain(w); ok {
line += " (" + w + ")"
}
}
parts = append(parts, line)
}
if len(parts) > 0 {
explanation = fmt.Sprintf("The delivery of %s failed: %s did not build.", what, readableList(parts))
} else {
explanation = fmt.Sprintf("The delivery of %s stopped before it finished.", what)
}
switch {
case p.passed:
explanation += " A newer delivery has passed it, so a retry would send older code. Nothing is needed from you."
case p.refusal == "":
explanation += " Nothing is tried again until you say so."
needs = "try the failed builds again " + FromMeshMCPServer
actions = []conditions.Action{retryAction(p.id)}
default:
explanation += " " + conditions.Capital(plainRefusal(p.refusal)) + "."
needs = "look at why it failed; a retry would not run now."
}
resolved = conditions.Capital(name) + " is no longer failed"
return
}
// cutWords is s cut to at most n bytes at a word.
func cutWords(s string, n int) string {
if len(s) <= n {
return s
}
t := s[:n]
if i := strings.LastIndexByte(t, ' '); i > n/2 {
t = t[:i]
}
return strings.TrimRight(t, " ,;:") + "…"
}
// gatherFailedPlans is every delivery plan that failed within planFailedWithin and is asked of the operator
// (S23, novox/hq ADR 0449 §1): a failed plan that carries a delivery, is no rollout of held updates and was not
// stopped by a person. Each carries why `plans retry` would not run it, when it would not.
func gatherFailedPlans(ctx context.Context, inv *inventory.Inventory, now time.Time) ([]failedPlanFacts, error) {
plans, err := inv.PlansSince(ctx, now.Add(-planFailedWithin))
if err != nil {
return nil, fmt.Errorf("the plans cannot be read: %w", err)
}
var failed []inventory.Plan
for _, p := range plans {
if failedAndAsked(p) {
failed = append(failed, p)
}
}
if len(failed) == 0 {
return nil, nil
}
entries, err := inv.Catalogued(ctx)
if err != nil {
return nil, fmt.Errorf("the catalogue cannot be read: %w", err)
}
onPath := ownPathOf(entries)
runsOn := map[string][]string{}
for _, e := range entries {
runsOn[e.Manifest.Module] = e.On
}
var out []failedPlanFacts
for _, p := range failed {
out = append(out, failedFactsOf(p, plans, onPath, runsOn))
}
return out, nil
}
// failedAndAsked says a plan is one S23 asks about: failed, a delivery's, not a rollout of held updates, not
// stopped by a person.
func failedAndAsked(p inventory.Plan) bool {
return p.State == inventory.PlanFailed && p.Delivery != nil && p.Release == nil && p.Delivery.Stopped == ""
}
// failedFactsOf is what S23 says of one failed plan, given every plan, the controller's own path and where each
// module runs.
func failedFactsOf(p inventory.Plan, plans []inventory.Plan, onPath map[string]string,
runsOn map[string][]string) failedPlanFacts {
f := failedPlanFacts{id: p.ID, repository: p.Repository, commit: p.Commit, modules: planModules(p),
note: p.Note, since: p.Updated}
for _, m := range f.modules {
if s := p.Modules[m]; s != nil && s.State == "failed" {
machines := append([]string(nil), runsOn[m]...)
sort.Strings(machines)
f.failed = append(f.failed, failedModule{module: m, machines: machines, why: s.Why})
}
}
if err := retryRefusal(p, plans, onPath); err != nil {
f.refusal = err.Error()
var passed passedBy
f.passed = errors.As(err, &passed)
}
return f
}
+338
View File
@@ -0,0 +1,338 @@
package main
import (
"context"
"fmt"
"sort"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/conditions"
"github.com/novox/mesh-controller/internal/inventory"
)
// **A failed delivery plan is asked of the operator** (novox/hq ADR 0449 §1 and §2, issue 1314).
func aFailedPlan(id string, mutate func(p *inventory.Plan)) inventory.Plan {
p := inventory.Plan{ID: id, Repository: "novox/mesh-catalog", Commit: "48bda475dunst0000", State: inventory.PlanFailed,
Tiers: [][]string{{"openrazer"}}, Updated: time.Date(2026, 10, 19, 18, 26, 0, 0, time.UTC),
Modules: map[string]*inventory.PlanModule{"openrazer": {State: "failed", Why: "the build could not fetch its source"}},
Delivery: &inventory.PlanDelivery{Awaits: "mesh-delivery"}}
if mutate != nil {
mutate(&p)
}
return p
}
func failedFactsFor(p inventory.Plan, others ...inventory.Plan) failedPlanFacts {
return failedFactsOf(p, append([]inventory.Plan{p}, others...), map[string]string{},
map[string][]string{"openrazer": {"g14", "shanks"}})
}
// 1314/AC1
func TestAFailedDeliveryPlanIsRaisedWithRetryAsItsOnlyAction(t *testing.T) {
now := time.Date(2026, 10, 19, 18, 30, 0, 0, time.UTC)
p := aFailedPlan("plan-1792433168676870614", nil)
f := calm(now)
f.failed = []failedPlanFacts{failedFactsFor(p)}
got := watchPlansFailed(f)
if len(got) != 1 {
t.Fatalf("%+v", got)
}
o := got[0]
if o.Kind != "plan-failed" || o.Severity != conditions.Warning || o.Resolver != conditions.ResolverOperator ||
o.Scope != conditions.ScopePlan || o.ID != p.ID {
t.Fatalf("kind, severity, resolver or subject: %+v", o)
}
if len(o.Actions) != 1 {
t.Fatalf("actions: %+v", o.Actions)
}
a := o.Actions[0]
if a.Label != "Retry" || a.Verb != "mesh-controller.plans" || a.Level != conditions.LevelApprove ||
len(a.Arguments) != 1 || a.Arguments["retry"] != p.ID {
t.Fatalf("the action is not Retry with only the plan: %+v", a)
}
if o.Needs == "" {
t.Fatal("no Needs: the asker wants an action with a need")
}
for _, want := range []string{"openrazer", "g14", "shanks", "could not fetch its source"} {
if !strings.Contains(o.Headline+o.Explanation+o.Said, want) {
t.Errorf("the words do not name %q:\nheadline %q\nexplanation %q\nsaid %q", want, o.Headline, o.Explanation, o.Said)
}
}
if !strings.Contains(o.Said, "could not fetch its source") || !strings.Contains(o.Explanation, "could not fetch its source") {
t.Errorf("the failure's own words are not in both the evidence and the explanation: %q / %q", o.Said, o.Explanation)
}
words := conditions.Words{Headline: o.Headline, Explanation: o.Explanation, Resolved: o.Resolved, Needs: o.Needs, Actions: o.Actions}
if why, ok := conditions.PlainWords(words, "g14", "shanks"); !ok {
t.Errorf("the words are not plain: %s\n%+v", why, words)
}
}
// 1314/AC2
func TestAFailedRolloutOfHeldUpdatesOrAPersonStoppedPlanIsNotRaised(t *testing.T) {
for name, p := range map[string]inventory.Plan{
"held updates": aFailedPlan("plan-1", func(p *inventory.Plan) { p.Release = &inventory.PlanRelease{} }),
"stopped": aFailedPlan("plan-2", func(p *inventory.Plan) { p.Delivery.Stopped = "jochen" }),
"no delivery": aFailedPlan("plan-3", func(p *inventory.Plan) { p.Delivery = nil }),
"building": aFailedPlan("plan-4", func(p *inventory.Plan) { p.State = inventory.PlanBuilding }),
} {
if failedAndAsked(p) {
t.Errorf("%s: a plan not asked of the operator is asked", name)
}
}
if !failedAndAsked(aFailedPlan("plan-5", nil)) {
t.Error("a failed delivery plan is not asked")
}
}
// 1314/AC3: every refusal of `plans retry` gives a condition with no action that says it.
func TestAFailedPlanThatCannotBeRetriedHasNoActionAndSaysWhy(t *testing.T) {
now := time.Date(2026, 10, 19, 18, 30, 0, 0, time.UTC)
older := aFailedPlan("plan-1", func(p *inventory.Plan) { p.Created = now.Add(-time.Hour) })
newer := aFailedPlan("plan-2", func(p *inventory.Plan) {
p.State, p.Created = inventory.PlanBuilding, now.Add(-time.Minute)
})
superseded := aFailedPlan("plan-3", func(p *inventory.Plan) { p.State, p.Note = inventory.PlanSuperseded, "superseded by plan-9" })
done := aFailedPlan("plan-4", func(p *inventory.Plan) { p.State = inventory.PlanDone })
first := time.Date(2026, 10, 19, 18, 20, 0, 0, time.UTC)
gate := aFailedPlan("plan-5", func(p *inventory.Plan) {
p.Note = "openrazer stopped at its first machine"
p.Modules = map[string]*inventory.PlanModule{"openrazer": {State: "built", FirstAt: &first, First: []string{"g14"},
Why: "the build did not become healthy on g14", Gate: &inventory.PlanGate{Machines: []string{"g14"},
Verdict: inventory.GateFailed, Why: "not healthy", From: "aaaaaaa", To: "bbbbbbb"}}}
})
for name, c := range map[string]struct {
p inventory.Plan
others []inventory.Plan
says string
}{
"a newer plan": {older, []inventory.Plan{newer}, "supersedes it"},
"superseded": {superseded, nil, "superseded"},
"done": {done, nil, "nothing to retry"},
"gate failed": {gate, nil, "failed its gate"},
} {
f := calm(now)
f.failed = []failedPlanFacts{failedFactsFor(c.p, c.others...)}
got := watchPlansFailed(f)
if len(got) != 1 {
t.Fatalf("%s: %+v", name, got)
}
o := got[0]
if len(o.Actions) != 0 {
t.Errorf("%s: a refused retry carries an action: %+v", name, o.Actions)
}
if !strings.Contains(o.Said, c.says) {
t.Errorf("%s: the evidence does not carry the refusal's words (%q): %q", name, c.says, o.Said)
}
if !strings.Contains(o.Explanation, "retry would not run") && !strings.Contains(o.Explanation, c.says) {
t.Errorf("%s: the explanation does not say the refusal: %q", name, o.Explanation)
}
words := conditions.Words{Headline: o.Headline, Explanation: o.Explanation, Resolved: o.Resolved, Needs: o.Needs}
if why, ok := conditions.PlainWords(words, "g14", "shanks"); !ok {
t.Errorf("%s: the words are not plain: %s\n%+v", name, why, words)
}
}
}
// 1314/AC4: the condition ends when the plan is no longer failed, kept by the keeper reconciling S23.
func TestAFailedPlanConditionEndsWhenThePlanLeavesFailed(t *testing.T) {
now := time.Date(2026, 10, 19, 18, 30, 0, 0, time.UTC)
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
Teller: &conditions.Told{}, Now: func() time.Time { return now }})
defer k.Close(t.Context())
p := aFailedPlan("plan-1792433168676870614", nil)
f := calm(now)
f.failed = []failedPlanFacts{failedFactsFor(p)}
if err := k.Reconcile(t.Context(), "S23", watchPlansFailed(f)); err != nil {
t.Fatal(err)
}
open, _ := k.Open(t.Context())
if len(open) != 1 || open[0].Kind != "plan-failed" {
t.Fatalf("not raised: %+v", open)
}
// Retried: the plan is building, so it is no longer among the failed.
f = calm(now.Add(time.Minute))
if err := k.Reconcile(t.Context(), "S23", watchPlansFailed(f)); err != nil {
t.Fatal(err)
}
if open, _ = k.Open(t.Context()); len(open) != 0 {
t.Fatalf("the condition outlived the failure: %+v", open)
}
// Failed again: asked again, with Retry, and nothing retried it.
f = calm(now.Add(2 * time.Minute))
again := aFailedPlan(p.ID, func(q *inventory.Plan) { q.Updated = now.Add(2 * time.Minute) })
f.failed = []failedPlanFacts{failedFactsFor(again)}
if err := k.Reconcile(t.Context(), "S23", watchPlansFailed(f)); err != nil {
t.Fatal(err)
}
open, _ = k.Open(t.Context())
if len(open) != 1 || len(open[0].Actions) != 1 || open[0].Actions[0].Label != "Retry" {
t.Fatalf("a second failure was not asked again with Retry: %+v", open)
}
}
// 1314/AC7 and ADR 0449 §2: no look at a failed plan retries it, however much time passes; the only thing
// S23 produces is an observation that offers an answer.
func TestNoClockRetriesAFailedPlan(t *testing.T) {
now := time.Date(2026, 10, 19, 18, 30, 0, 0, time.UTC)
f := calm(now)
f.failed = []failedPlanFacts{failedFactsFor(aFailedPlan("plan-1", nil))}
for _, later := range []time.Duration{time.Hour, 24 * time.Hour, 6 * 24 * time.Hour} {
f.now = now.Add(later)
got := watchPlansFailed(f)
if len(got) != 1 || len(got[0].Actions) != 1 || got[0].Actions[0].Arguments["retry"] != "plan-1" {
t.Fatalf("after %s: %+v", later, got)
}
}
// A plan failed longer ago than the week it is kept is left to `plans`.
f.now = now.Add(planFailedWithin + time.Second)
if got := watchPlansFailed(f); len(got) != 0 {
t.Fatalf("a plan failed beyond the window is still raised: %+v", got)
}
}
// A later plan of the same repository and branch, in a state, made after the failed one.
func laterPlan(id, branch, state string, after inventory.Plan) inventory.Plan {
q := aFailedPlan(id, func(p *inventory.Plan) {
p.State, p.Branch, p.Created = state, branch, after.Created.Add(time.Hour)
})
return q
}
// 1328/AC1: a later done plan of the same repository and branch makes `plans retry` refuse, naming the later plan.
func TestARetryIsRefusedWhenALaterPlanOfTheBranchIsDone(t *testing.T) {
failed := aFailedPlan("plan-1", func(p *inventory.Plan) { p.Branch, p.Created = "main", time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC) })
later := laterPlan("plan-2", "main", inventory.PlanDone, failed)
err := retryRefusal(failed, []inventory.Plan{failed, later}, map[string]string{})
if err == nil || !strings.Contains(err.Error(), "plan-2") || !strings.Contains(err.Error(), "passed") {
t.Fatalf("the retry was not refused, naming the later plan: %v", err)
}
}
// 1328/AC2: with the later plan still open the refusal is today's; a later done plan of another branch refuses nothing.
func TestARetryIsRefusedAsBeforeWhileTheLaterPlanIsOpenAndNotForAnotherBranch(t *testing.T) {
failed := aFailedPlan("plan-1", func(p *inventory.Plan) { p.Branch, p.Created = "main", time.Date(2026, 10, 19, 12, 0, 0, 0, time.UTC) })
open := laterPlan("plan-2", "main", inventory.PlanBuilding, failed)
err := retryRefusal(failed, []inventory.Plan{failed, open}, map[string]string{})
if err == nil || !strings.Contains(err.Error(), "supersedes it") {
t.Fatalf("with the later plan open the refusal is not today's: %v", err)
}
other := laterPlan("plan-3", "feature/x", inventory.PlanDone, failed)
if err := retryRefusal(failed, []inventory.Plan{failed, other}, map[string]string{}); err != nil {
t.Fatalf("a later done plan of another branch refused the retry: %v", err)
}
earlier := laterPlan("plan-4", "main", inventory.PlanDone, failed)
earlier.Created = failed.Created.Add(-time.Hour)
if err := retryRefusal(failed, []inventory.Plan{failed, earlier}, map[string]string{}); err != nil {
t.Fatalf("an earlier done plan refused the retry: %v", err)
}
}
// 1328/AC3: such a plan's plan-failed condition has no action and says it was passed by a newer delivery.
func TestAPassedFailedPlanHasNoActionAndSaysItWasPassed(t *testing.T) {
now := time.Date(2026, 10, 19, 18, 30, 0, 0, time.UTC)
failed := aFailedPlan("plan-1", func(p *inventory.Plan) { p.Branch, p.Created = "main", now.Add(-5*time.Hour) })
later := laterPlan("plan-2", "main", inventory.PlanDone, failed)
f := calm(now)
f.failed = []failedPlanFacts{failedFactsFor(failed, later)}
got := watchPlansFailed(f)
if len(got) != 1 || len(got[0].Actions) != 0 {
t.Fatalf("%+v", got)
}
o := got[0]
if !strings.Contains(o.Explanation, "newer delivery") || !strings.Contains(o.Explanation, "passed") {
t.Errorf("the words do not say it was passed by a newer delivery: %q", o.Explanation)
}
if strings.Contains(o.Explanation, "details say why") {
t.Errorf("the generic fallback was said: %q", o.Explanation)
}
words := conditions.Words{Headline: o.Headline, Explanation: o.Explanation, Resolved: o.Resolved, Needs: o.Needs}
if why, ok := conditions.PlainWords(words, "g14", "shanks"); !ok {
t.Errorf("not plain: %s\n%+v", why, words)
}
}
// plansBehind serves plans as the inventory does: the open ones, the most recent n, and every one made since.
type plansBehind []inventory.Plan
func (s plansBehind) OpenPlans(context.Context) ([]inventory.Plan, error) {
var out []inventory.Plan
for _, p := range s {
if p.Open() {
out = append(out, p)
}
}
return out, nil
}
func (s plansBehind) RecentPlans(_ context.Context, limit int) ([]inventory.Plan, error) {
all := append([]inventory.Plan(nil), s...)
sort.SliceStable(all, func(i, j int) bool { return all[i].Created.After(all[j].Created) })
if len(all) > limit {
all = all[:limit]
}
return all, nil
}
func (s plansBehind) PlansSince(_ context.Context, since time.Time) ([]inventory.Plan, error) {
var out []inventory.Plan
for _, p := range s {
if p.Created.After(since) || p.Open() {
out = append(out, p)
}
}
return out, nil
}
// 1328, review round 3: the plan that passed a failed one is seen by `plans retry` however many plans were made after it.
func TestARetrySeesThePassingPlanBeyondTheFiftyMostRecent(t *testing.T) {
failed := aFailedPlan("plan-1", func(p *inventory.Plan) { p.Branch, p.Created = "main", time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) })
all := plansBehind{failed, laterPlan("plan-done", "main", inventory.PlanDone, failed)}
for i := 0; i < 60; i++ { // 60 newer plans of other repositories push the passing one out of the newest 50
other := laterPlan(fmt.Sprintf("plan-x%d", i), "main", inventory.PlanFailed, failed)
other.Repository = "novox/other"
other.Created = failed.Created.Add(time.Duration(48+i) * time.Hour)
all = append(all, other)
}
plans, err := plansForRetry(t.Context(), all, failed)
if err != nil {
t.Fatal(err)
}
if err := retryRefusal(failed, plans, map[string]string{}); err == nil || !strings.Contains(err.Error(), "plan-done") {
t.Fatalf("the retry was not refused naming the passing plan beyond the 50 most recent: %v", err)
}
}
// ADR 0451 §3: a failed plan that last changed 8 days ago raises nothing, and a raised condition ends once its
// plan's last change passes 7 days.
func TestAFailedPlanLastChangedEightDaysAgoRaisesNothingAndARaisedOneEnds(t *testing.T) {
now := time.Date(2026, 10, 19, 18, 30, 0, 0, time.UTC)
old := aFailedPlan("plan-1", func(p *inventory.Plan) { p.Updated = now.Add(-8 * 24 * time.Hour) })
f := calm(now)
f.failed = []failedPlanFacts{failedFactsFor(old)}
if got := watchPlansFailed(f); len(got) != 0 {
t.Fatalf("a plan last changed 8 days ago raised %+v", got)
}
store := conditions.NewInMemory()
k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
Teller: &conditions.Told{}, Now: func() time.Time { return now }})
defer k.Close(t.Context())
fresh := aFailedPlan("plan-1", func(p *inventory.Plan) { p.Updated = now.Add(-6 * 24 * time.Hour) })
f.failed = []failedPlanFacts{failedFactsFor(fresh)}
if err := k.Reconcile(t.Context(), "S23", watchPlansFailed(f)); err != nil {
t.Fatal(err)
}
if open, _ := k.Open(t.Context()); len(open) != 1 {
t.Fatalf("a plan last changed 6 days ago is not raised: %+v", open)
}
f.now = now.Add(2 * 24 * time.Hour) // its last change is now 8 days old
if err := k.Reconcile(t.Context(), "S23", watchPlansFailed(f)); err != nil {
t.Fatal(err)
}
if open, _ := k.Open(t.Context()); len(open) != 0 {
t.Fatalf("the condition outlived the 7 days: %+v", open)
}
}
+15 -5
View File
@@ -138,7 +138,9 @@ func askToLook() {
// startPlanLooks starts the look at the open plans, and asks for the first: the controller took the lease, and a
// plan left by a replaced controller resumes from there (ADR 0406 §6). It replaces the rollout's 30 second ticker.
func startPlanLooks(ctx context.Context, open *stores) {
// It answers a channel that is closed once the look has returned after the context ended, for a test that restores what
// the look reads (issue 1338).
func startPlanLooks(ctx context.Context, open *stores) <-chan struct{} {
l := newLooker(func(ctx context.Context) bool { return lookAtPlans(ctx, open) })
l.first = func(ctx context.Context) { rearmWalkEnds(ctx, open) }
lookerMu.Lock()
@@ -152,8 +154,13 @@ func startPlanLooks(ctx context.Context, open *stores) {
}
lookerMu.Unlock()
}()
go l.run(ctx)
done := make(chan struct{})
go func() {
defer close(done)
l.run(ctx)
}()
l.ask()
return done
}
// applySilentAfter is how long a machine may say nothing on a send before the plan names it silent: the mesh's
@@ -163,8 +170,10 @@ var applySilentAfter = inventory.ApplySilentAfter
// startRolloutEvents starts what a rollout advances on, for as long as the context lives (novox/hq ADR 0406 §5, issue 974):
// the wait for each machine's report as a key on the bus that expires (where the bus will not take it a timer of its own
// for that wait looks instead), a provisioner client's words, and the look at the plans, asked for first so a plan left
// by a replaced controller resumes. The serving controller's start and the tests of a rollout call this one.
func startRolloutEvents(ctx context.Context, open *stores, js *broker.JetStream) {
// by a replaced controller resumes. The serving controller's start and the tests of a rollout call this one. It answers
// a function that waits until the look at the plans has returned, which only a test that ends the context needs (issue
// 1338): the serving controller ignores it.
func startRolloutEvents(ctx context.Context, open *stores, js *broker.JetStream) (wait func()) {
if err := startRolloutSilent(ctx, js); err != nil {
fmt.Printf("plans: the wait for the machines' reports could not be put on the bus (%v); each is looked at by a timer of its own\n", err)
}
@@ -174,7 +183,8 @@ func startRolloutEvents(ctx context.Context, open *stores, js *broker.JetStream)
if err := startPlanMovedLooks(ctx, js); err != nil {
fmt.Printf("plans: the plans moved by other processes could not be followed (%v)\n", err)
}
startPlanLooks(ctx, open)
done := startPlanLooks(ctx, open)
return func() { <-done }
}
// servingPlanSaved is what the serving controller does for each plan it keeps: said as plan-moved, looked at (a build, a
+71 -9
View File
@@ -154,6 +154,31 @@ func newerOpenPlan(p inventory.Plan, plans []inventory.Plan) (inventory.Plan, bo
return inventory.Plan{}, false
}
// newerDonePlan is a done plan of the same repository and branch made after this one: the newer commit it delivered
// runs now (novox/hq ADR 0451).
func newerDonePlan(p inventory.Plan, plans []inventory.Plan) (inventory.Plan, bool) {
var found inventory.Plan
var ok bool
for _, q := range plans {
if q.ID == p.ID || q.State != inventory.PlanDone || !strings.EqualFold(q.Repository, p.Repository) ||
(p.Branch != "" && q.Branch != "" && p.Branch != q.Branch) || !q.Created.After(p.Created) {
continue
}
if !ok || q.Created.After(found.Created) {
found, ok = q, true
}
}
return found, ok
}
// passedBy is the refusal of a retry of a plan that a newer done plan of its repository and branch has passed.
type passedBy struct{ plan, by, repository, commit string }
func (e passedBy) Error() string {
return fmt.Sprintf("%s was passed by %s, a newer plan of %s (at %s) that is done: retrying it would build and "+
"send the older commit over the newer one that runs", e.plan, e.by, e.repository, short(e.commit))
}
// retryRefusal is why a plan cannot be retried, or nothing.
func retryRefusal(p inventory.Plan, plans []inventory.Plan, onPath map[string]string) error {
switch {
@@ -165,6 +190,11 @@ func retryRefusal(p inventory.Plan, plans []inventory.Plan, onPath map[string]st
return fmt.Errorf("%s is still %s; nothing in it failed to retry — `rebuild <module>` asks one module again", p.ID, p.State)
}
// **A newer delivery of the branch has passed it** (novox/hq ADR 0451): retrying would build and send the older
// commit over the newer one that runs.
if q, found := newerDonePlan(p, plans); found {
return passedBy{plan: p.ID, by: q.ID, repository: q.Repository, commit: q.Commit}
}
if len(failedIn(p)) > 0 {
if q, found := newerOpenPlan(p, plans); found {
return fmt.Errorf("%s supersedes it: a newer merge of %s (%s at %s) is open, and retrying %s would build "+
@@ -275,6 +305,32 @@ func goesOnAgain(p *inventory.Plan) {
}
}
// planReader is what a retry reads the mesh's plans through.
type planReader interface {
OpenPlans(ctx context.Context) ([]inventory.Plan, error)
RecentPlans(ctx context.Context, limit int) ([]inventory.Plan, error)
PlansSince(ctx context.Context, since time.Time) ([]inventory.Plan, error)
}
// plansForRetry are the plans a retry of p judges itself against.
func plansForRetry(ctx context.Context, inv planReader, p inventory.Plan) ([]inventory.Plan, error) {
plans, err := inv.OpenPlans(ctx)
if err != nil {
return nil, err
}
recent, err := inv.RecentPlans(ctx, 50)
if err != nil {
return nil, err
}
// **Every plan made after this one** (novox/hq ADR 0451 §1): a plan that passed it is seen however many plans
// were made since, not only when it is among the newest.
later, err := inv.PlansSince(ctx, p.Created)
if err != nil {
return nil, err
}
return append(append(plans, recent...), later...), nil
}
// retryPlan asks a failed plan's failed modules again, under new ids, and sets it building again at
// that tier: what follows is the plan going on as if they had built the first time.
func retryPlan(ctx context.Context, open *stores, id string) (string, error) {
@@ -288,15 +344,10 @@ func retryPlan(ctx context.Context, open *stores, id string) (string, error) {
if err != nil {
return "", err
}
plans, err := inv.OpenPlans(ctx)
plans, err := plansForRetry(ctx, inv, p)
if err != nil {
return "", err
}
recent, err := inv.RecentPlans(ctx, 50)
if err != nil {
return "", err
}
plans = append(plans, recent...)
// **A failed walk whose earlier merges are walked alone is not retried** (novox/hq ADR 0276): the search for
// the merge that brought the failure answers them now, and a retried walk would name them twice.
// **A walk stopped without some merges is not retried** (novox/hq ADR 0299): the merges it left out were stopped,
@@ -355,15 +406,23 @@ func retryPlan(ctx context.Context, open *stores, id string) (string, error) {
asked = append(asked, m+" as "+s.Build)
}
}
resumed(&p, fmt.Sprintf("tier %d retried by hand: %s asked again", p.Tier, strings.Join(failed, ", ")))
waiting := busWaitingIn(p)
resumed(&p, fmt.Sprintf("tier %d retried by hand: %s asked again", p.Tier, strings.Join(slicesWithout(failed, waiting), ", ")))
if len(waiting) > 0 {
p.Note += "; " + busWaitWords(p, waiting)
}
if err := inv.SavePlan(ctx, &p); err != nil {
return "", err
}
if p.State != inventory.PlanBuilding {
return "", fmt.Errorf("%s could not be resumed: %s", p.ID, p.Note)
}
return fmt.Sprintf("%s retried at tier %d of %d: asked %s; the plan goes on from there as any plan does",
p.ID, p.Tier, len(p.Tiers), strings.Join(asked, ", ")), nil
said := fmt.Sprintf("%s retried at tier %d of %d: asked %s; the plan goes on from there as any plan does",
p.ID, p.Tier, len(p.Tiers), strings.Join(asked, ", "))
if len(waiting) > 0 {
said += "; " + busWaitWords(p, waiting)
}
return said, nil
}
// joinAPlan asks a module again for the plan that holds it unbuilt or failed, if one does: open plans
@@ -409,6 +468,9 @@ func joinAPlan(ctx context.Context, open *stores, module string) (bool, string,
if err := inv.SavePlan(ctx, &p); err != nil {
return false, "", err
}
if s.State == "" && s.BusWaitFrom != nil {
return true, busWaitWords(p, []string{module}), nil
}
if s.State != "asked" {
return true, "", fmt.Errorf("%s could not be asked for %s: %s", module, p.ID, s.Why)
}
+6 -3
View File
@@ -79,11 +79,14 @@ func TestRetryRefusesWhatItCannotResume(t *testing.T) {
if err := retryRefusal(stopped, nil, ownPathOf(nil)); err == nil || !strings.Contains(err.Error(), "nothing in tier 0") {
t.Errorf("a plan with nothing failed to build was retried: %v", err)
}
// A failed or done plan does not supersede it.
if err := retryRefusal(failed, []inventory.Plan{plan("plan-4", inventory.PlanFailed, at.Add(time.Hour)),
plan("plan-5", inventory.PlanDone, at.Add(time.Hour))}, ownPathOf(nil)); err != nil {
// A later failed plan does not supersede it; a later done one passes it (novox/hq ADR 0451).
if err := retryRefusal(failed, []inventory.Plan{plan("plan-4", inventory.PlanFailed, at.Add(time.Hour))}, ownPathOf(nil)); err != nil {
t.Errorf("refused for a plan that does not supersede it: %v", err)
}
if err := retryRefusal(failed, []inventory.Plan{plan("plan-5", inventory.PlanDone, at.Add(time.Hour))}, ownPathOf(nil)); err == nil ||
!strings.Contains(err.Error(), "plan-5") {
t.Errorf("a plan a later done plan passed was retried: %v", err)
}
// Another branch's open walk, or an older one, does not supersede it, and is one walk open: one at a time
// (novox/hq ADR 0276).
other := plan("plan-3", inventory.PlanBuilding, at.Add(time.Hour))
+23 -3
View File
@@ -580,12 +580,25 @@ func askModule(ctx context.Context, p *inventory.Plan, name string, byName map[s
id, err := askABuild(withReuse(ctx, reuseWithVerdicts(ctx, p.ID, reuseAskOf(p), mergedHeadsOf(p))), source,
e.Source.Path, ref)
if err != nil {
// **A bus that does not answer for a moment is not a failed build** (issue 1334): the module stays unasked and the
// tier's next look asks it again. A fault of the module's own fails the plan at once, as does a bus silent too long.
reason := err.Error()
if busDidNotAnswer(err) {
if waitForTheBus(p, state, name, err, now) {
return
}
reason = fmt.Sprintf("the bus did not answer an ask for %s (since %s): %s", busWaitBound,
state.BusWaitFrom.UTC().Format(time.RFC3339), firstLine(err.Error()))
}
// The wait has ended: a Retry or a rebuild waits for the bus afresh.
state.BusWaitFrom = nil
state.State = "failed"
state.Why = err.Error()
state.Why = reason
p.State = inventory.PlanFailed
p.Note = fmt.Sprintf("%s could not be asked for: %v", name, err)
p.Note = fmt.Sprintf("%s could not be asked for: %s", name, reason)
return
}
state.BusWaitFrom = nil
// The commit of the module's own repository the walk carries (novox/hq ADR 0276): a batch's walk carries
// one per repository.
commit := carried
@@ -810,6 +823,11 @@ func advanceHeld(ctx context.Context, open *stores) (failed bool) {
if p.State == inventory.PlanFailed {
sayUnsent(p, rollsOut)
}
// A module the bus did not answer an ask of is asked again by a look, and no event may bring one (a tier all
// refused has no build in flight): the look fails, so the looker retries it from five seconds (issue 1334).
if busWaiting(*p) {
failed = true
}
if planSnapshot(*p) != before {
if err := inv.SavePlan(ctx, p); err != nil {
fmt.Printf("%s: cannot keep the plan: %v\n", p.ID, err)
@@ -887,10 +905,12 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
if len(unasked) < len(tier) {
fmt.Printf("%s: tier %d was asked in part; asking the rest: %s\n", p.ID, p.Tier, strings.Join(unasked, ", "))
}
note := p.Note
if err := askTier(ctx, inv, p, unasked); err != nil {
return false, err
}
return true, nil
// Nothing asked and nothing new to say (the bus still silent, issue 1334): not saved again at every look.
return len(unaskedOf(*p)) < len(unasked) || p.Note != note || p.State != inventory.PlanBuilding, nil
}
// **Asked: settle from the build records first** (novox/hq 04-ISSUES/214). An outcome is taken
// in by whichever controller hears it, and a merge to the controller's own repository replaces
+9 -3
View File
@@ -179,7 +179,7 @@ func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) {
// The first controller. Its plan has tier 0 built and tier 1 not asked: nothing but the save asks anything.
first, stopFirst := context.WithCancel(ctx)
defer stopFirst()
startRolloutEvents(first, open, js)
waitFirst := startRolloutEvents(first, open, js)
built := time.Now().UTC()
p := inventory.Plan{ID: "plan-974", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", Created: built,
State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
@@ -199,12 +199,15 @@ func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) {
// The controller is replaced: its looks and its watcher end. anchor reports meanwhile, to nobody.
stopFirst()
waitFirst()
time.Sleep(300 * time.Millisecond)
reports("anchor")
second, stopSecond := context.WithCancel(ctx)
defer stopSecond()
startRolloutEvents(second, open, js)
waitSecond := startRolloutEvents(second, open, js)
// Ends with the test before the cleanups restore what the look reads, which run after it (issue 1338).
t.Cleanup(func() { stopSecond(); waitSecond() })
// Its first look finds anchor reported, so its key goes; laptop's stays, and is the new controller's to hear expire.
eventually(t, "the new controller's first look deletes the key of the machine that reported", func() bool {
h := heldKeys()
@@ -321,7 +324,7 @@ func TestAWalkLetGoByTheDeliverVerbIsAskedWithNoOtherEvent(t *testing.T) {
t.Cleanup(func() { inventory.PlanSaved = wasSaved })
started, stop := context.WithCancel(ctx)
defer stop()
startRolloutEvents(started, open, js)
wait := startRolloutEvents(started, open, js)
time.Sleep(500 * time.Millisecond) // the first look: the walk waits, and is asked nothing
// The verb as its child process runs it: the delivery command, whose saves are not the serving controller's.
@@ -333,6 +336,9 @@ func TestAWalkLetGoByTheDeliverVerbIsAskedWithNoOtherEvent(t *testing.T) {
return verbAnswer{}, deliveryCommand(ctx, []string{"go", waiting.ID, "--why", "its turn"})
}
t.Cleanup(func() { runTheVerb = wasRun })
// Registered after every variable cleanup above, so it runs before them (last in, first out): the look at the plans
// has returned when checkEvents, inventory.PlanSaved and runTheVerb are restored (issue 1338).
t.Cleanup(func() { stop(); wait() })
handlers, _, err := seatToolHandlers()
if err != nil {
t.Fatal(err)
+129
View File
@@ -0,0 +1,129 @@
package main
import (
"context"
"fmt"
"io"
"os"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// seatPassing is what passing seats on needs of the outside: the record's compare-and-set, and the hand act.
type seatPassing struct {
pass func(ctx context.Context, p catalogue.SeatPass) (bool, error)
act func(ctx context.Context, p catalogue.SeatPass) error
say io.Writer
}
// passSeatsOn records, in the composition that finds it, the new holder of each seat held once whose holder on
// record is assigned nowhere and which has exactly one claimant able to hold it, and writes the controller's own
// hand act for it (novox/hq ADR 0449 §4, issue 1317). It answers the holdings as they are after it, so the
// resolution that follows reads the record it just moved. A failed write is said and leaves the record as it was:
// a composition is never refused for it.
func passSeatsOn(ctx context.Context, sp seatPassing, holdings []catalogue.Held,
assignments []catalogue.Assignment) []catalogue.Held {
passes, said := catalogue.SeatsToPassOn(holdings, assignments)
for _, line := range said {
fmt.Fprintln(sp.say, line)
}
for _, p := range passes {
moved, err := sp.pass(ctx, p)
if err != nil {
fmt.Fprintf(sp.say, "%s could not be passed from %s on %s to %s on %s: %v\n", p.Seat.Name,
p.From.Module, p.From.Node, p.To.Module, p.To.Node, err)
continue
}
if !moved {
continue // another composition passed it first, and wrote the act
}
if err := sp.act(ctx, p); err != nil {
fmt.Fprintf(sp.say, "%s passed from %s on %s to %s on %s, and the hand act could NOT be recorded: %v\n",
p.Seat.Name, p.From.Module, p.From.Node, p.To.Module, p.To.Node, err)
}
var next []catalogue.Held
for _, h := range holdings {
if hs, ok := catalogue.SeatNamed(h.Claim); ok && hs.Name == p.Seat.Name {
continue
}
next = append(next, h)
}
p.To.Site = p.From.Site
holdings = append(next, p.To)
}
return holdings
}
// seatPassHandAct is the controller's own entry for a pass: it is the act that puts the holder on record,
// where a person's `seat --to` would have, so the record of who held what stays complete (ADR 0449 §4).
func seatPassHandAct(p catalogue.SeatPass) link.HandAct {
return link.HandAct{
By: "the controller",
Verb: "seat",
Args: []string{p.Seat.Name, "--to", p.To.Node + "/" + p.To.Module, "--was", p.From.Node + "/" + p.From.Module},
Why: fmt.Sprintf("%s on %s, the holder on record of %s, is assigned on no machine, and %s on %s is the only "+
"assignment that claims it and can hold it (ADR 0449 §4)", p.From.Module, p.From.Node, p.Seat.Name,
p.To.Module, p.To.Node),
Cause: "seat-passed",
Outcome: "done",
}
}
// assignmentsOf are the assignments of every machine, as the passing of a seat reads them. **A module assigned and
// missing from the shelf is still assigned**: it is an assignment with no claims, so its holder is not taken for gone.
func assignmentsOf(shelf map[string]catalogue.Manifest, assigned map[string][]string) []catalogue.Assignment {
var out []catalogue.Assignment
for node, modules := range assigned {
for _, module := range modules {
m, ok := shelf[module]
if !ok {
m = catalogue.Manifest{Module: module}
}
out = append(out, catalogue.Assignment{Node: node, Manifest: m})
}
}
return out
}
// seatPassingOn is how a composition on this inventory passes a seat: against the store's record with the hand act
// on the bus, or, on a throwaway mesh (the merge gate's), in memory only, writing nothing anywhere.
func seatPassingOn(inv *inventory.Inventory, w io.Writer) seatPassing {
if inv.Throwaway() {
return seatPassing{say: w,
pass: func(context.Context, catalogue.SeatPass) (bool, error) { return true, nil },
act: func(context.Context, catalogue.SeatPass) error { return nil }}
}
return seatPassing{
say: w,
pass: func(ctx context.Context, p catalogue.SeatPass) (bool, error) {
return inv.PassSeat(ctx, p.Seat.Name, p.Seat.Scope, p.From, p.To)
},
act: func(ctx context.Context, p catalogue.SeatPass) error {
return onTheBus(func(conn *nats.Conn) error {
_, err := link.RecordHandAct(ctx, conn, seatPassHandAct(p))
return err
})
},
}
}
// theMeshPassesSeatsOn is passSeatsOn as the composition runs it. **Only when every machine's assignments were
// read** (assignedErr nil): a machine whose read failed would look as if its modules were assigned nowhere, and
// its holder would lose the seat. Said to where the composition says its other lines (sayLeftOut).
func theMeshPassesSeatsOn(ctx context.Context, inv *inventory.Inventory, shelf map[string]catalogue.Manifest,
assigned map[string][]string, assignedErr error, holdings []catalogue.Held) []catalogue.Held {
var w io.Writer = os.Stderr
if lw, ok := ctx.Value(leftOutKey{}).(io.Writer); ok {
w = lw
}
if assignedErr != nil {
fmt.Fprintf(w, "no seat is passed on in this composition: a machine's assignments could not be read, so a "+
"holder on record might look assigned nowhere: %v\n", assignedErr)
return holdings
}
return passSeatsOn(ctx, seatPassingOn(inv, w), holdings, assignmentsOf(shelf, assigned))
}
+193
View File
@@ -0,0 +1,193 @@
package main
import (
"bytes"
"context"
"errors"
"strings"
"testing"
"time"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
func aBrokerClaiming(module string) catalogue.Manifest {
return catalogue.Manifest{Module: module,
Provides: []catalogue.Offer{{Name: "mesh-bus", Scope: catalogue.ScopeMesh}},
Claims: []catalogue.Claim{{Name: "mesh-broker", Scope: catalogue.ScopeMesh}}}
}
func withBusSeat(t *testing.T) {
t.Helper()
was := catalogue.Seats()
t.Cleanup(func() { catalogue.UseSeats(was) })
catalogue.UseSeats([]catalogue.Seat{{Name: "mesh-broker", Scope: catalogue.ScopeMesh, Delivers: "mesh-bus", Decision: "test"}})
}
// 1317/AC1: the composition records the claimant as the holder and writes one hand act by the controller naming
// the seat, the former holder and the new one; the holdings it answers name the new holder.
func TestACompositionPassesTheSeatAndWritesOneHandActByTheController(t *testing.T) {
withBusSeat(t)
var passed []catalogue.SeatPass
var acts []link.HandAct
var said bytes.Buffer
sp := seatPassing{say: &said,
pass: func(_ context.Context, p catalogue.SeatPass) (bool, error) {
passed = append(passed, p)
return true, nil
},
act: func(_ context.Context, p catalogue.SeatPass) error {
acts = append(acts, seatPassHandAct(p))
return nil
},
}
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
got := passSeatsOn(t.Context(), sp, holdings, []catalogue.Assignment{{Node: "novox", Manifest: aBrokerClaiming("gate")}})
if len(got) != 1 || got[0].Module != "gate" || got[0].Node != "novox" {
t.Fatalf("the holdings after the composition do not name the gate: %+v", got)
}
if len(passed) != 1 || len(acts) != 1 {
t.Fatalf("passes=%d acts=%d, want one each", len(passed), len(acts))
}
a := acts[0]
text := strings.Join(append([]string{a.By, a.Verb, a.Why}, a.Args...), " ")
for _, want := range []string{"the controller", "mesh-broker", "novox/old-broker", "novox/gate"} {
if !strings.Contains(text, want) {
t.Fatalf("the hand act does not name %q: %+v", want, a)
}
}
}
// 1317/AC2: two claimants, nothing recorded, the refusal said.
func TestACompositionWithTwoClaimantsRecordsNoneAndSaysSo(t *testing.T) {
withBusSeat(t)
sp := seatPassing{say: &bytes.Buffer{},
pass: func(context.Context, catalogue.SeatPass) (bool, error) {
t.Fatal("a seat with two claimants was passed")
return false, nil
},
act: func(context.Context, catalogue.SeatPass) error { t.Fatal("an act was written"); return nil },
}
var said bytes.Buffer
sp.say = &said
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
got := passSeatsOn(t.Context(), sp, holdings, []catalogue.Assignment{
{Node: "novox", Manifest: aBrokerClaiming("gate")}, {Node: "ace", Manifest: aBrokerClaiming("other")}})
if len(got) != 1 || got[0].Module != "old-broker" {
t.Fatalf("the record changed: %+v", got)
}
if !strings.Contains(said.String(), "gate") || !strings.Contains(said.String(), "other") ||
!strings.Contains(said.String(), "seat mesh-broker --to") {
t.Fatalf("the refusal is not said with both and the remedy: %q", said.String())
}
}
// 1317/AC3 and AC4: nothing changes while the holder is assigned, nor with no claimant.
func TestACompositionChangesNothingWhileTheHolderIsAssignedOrNobodyClaims(t *testing.T) {
withBusSeat(t)
sp := seatPassing{say: &bytes.Buffer{},
pass: func(context.Context, catalogue.SeatPass) (bool, error) { t.Fatal("passed"); return false, nil },
act: func(context.Context, catalogue.SeatPass) error { t.Fatal("act"); return nil },
}
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
got := passSeatsOn(t.Context(), sp, holdings, []catalogue.Assignment{
{Node: "novox", Manifest: aBrokerClaiming("old-broker")}, {Node: "novox", Manifest: aBrokerClaiming("gate")}})
if len(got) != 1 || got[0].Module != "old-broker" {
t.Fatalf("holder assigned: record changed: %+v", got)
}
got = passSeatsOn(t.Context(), sp, holdings, nil)
if len(got) != 1 || got[0].Module != "old-broker" {
t.Fatalf("no claimant: record changed: %+v", got)
}
}
// A composition that lost the compare-and-set writes no act, and a failed write leaves the record.
func TestALostPassWritesNoActAndAFailedOneLeavesTheRecord(t *testing.T) {
withBusSeat(t)
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
as := []catalogue.Assignment{{Node: "novox", Manifest: aBrokerClaiming("gate")}}
lost := seatPassing{say: &bytes.Buffer{},
pass: func(context.Context, catalogue.SeatPass) (bool, error) { return false, nil },
act: func(context.Context, catalogue.SeatPass) error { t.Fatal("act without a pass"); return nil },
}
passSeatsOn(t.Context(), lost, holdings, as)
var said bytes.Buffer
sp := seatPassing{say: &said,
pass: func(context.Context, catalogue.SeatPass) (bool, error) { return false, context.DeadlineExceeded },
act: func(context.Context, catalogue.SeatPass) error { t.Fatal("act"); return nil },
}
got := passSeatsOn(t.Context(), sp, holdings, as)
if got[0].Module != "old-broker" || !strings.Contains(said.String(), "could not be passed") {
t.Fatalf("got %+v said %q", got, said.String())
}
}
// A pass is the mesh working as ADR 0449 §4 decided, never a repair: two of them raise no healer-wanted and
// count as no repair.
func TestTwoSeatPassesAreNoRepairAndRaiseNoHealerWanted(t *testing.T) {
withBusSeat(t)
now := time.Date(2026, 10, 19, 20, 0, 0, 0, time.UTC)
mk := func(at time.Time) link.HandAct {
a := seatPassHandAct(catalogue.SeatPass{Seat: catalogue.Seat{Name: "mesh-broker"},
From: catalogue.Held{Node: "novox", Module: "old"}, To: catalogue.Held{Node: "novox", Module: "gate"}})
a.At = at
return a
}
acts := []link.HandAct{mk(now.Add(-26 * time.Hour)), mk(now.Add(-time.Hour))}
if r := repairs(acts); len(r) != 0 {
t.Fatalf("a seat pass counts as a repair: %+v", r)
}
f := calm(now)
f.handActs = acts
if got := watchHandActs(f); len(got) != 0 {
t.Fatalf("two passes raised %+v", got)
}
}
// A failed read of one machine's assignments passes no seat: its holder would look assigned nowhere.
func TestNoSeatIsPassedWhenAMachinesAssignmentsCouldNotBeRead(t *testing.T) {
withBusSeat(t)
var said bytes.Buffer
ctx := leftOutSaidBy(t.Context(), &said)
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "ace", Module: "old-broker"}}
shelf := map[string]catalogue.Manifest{"gate": aBrokerClaiming("gate")}
got := theMeshPassesSeatsOn(ctx, nil, shelf, map[string][]string{"novox": {"gate"}},
errors.New("ace: the bus did not answer"), holdings)
if len(got) != 1 || got[0].Module != "old-broker" {
t.Fatalf("the record moved: %+v", got)
}
if !strings.Contains(said.String(), "no seat is passed") || !strings.Contains(said.String(), "ace") {
t.Errorf("not said: %q", said.String())
}
}
// A holder that is assigned but missing from the shelf is still assigned.
func TestAHolderAssignedButNotOnTheShelfIsNotTakenForGone(t *testing.T) {
withBusSeat(t)
as := assignmentsOf(map[string]catalogue.Manifest{"gate": aBrokerClaiming("gate")},
map[string][]string{"novox": {"gate", "old-broker"}})
passes, _ := catalogue.SeatsToPassOn(
[]catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}, as)
if len(passes) != 0 {
t.Fatalf("a holder missing from the shelf lost its seat: %+v", passes)
}
}
// The merge gate's throwaway mesh passes the seat in memory and writes nothing: no record, no hand act.
func TestAThrowawayMeshPassesTheSeatInMemoryAndWritesNothing(t *testing.T) {
withBusSeat(t)
inv := &inventory.Inventory{}
inv.MarkThrowaway()
var said bytes.Buffer
holdings := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
got := passSeatsOn(t.Context(), seatPassingOn(inv, &said), holdings,
[]catalogue.Assignment{{Node: "novox", Manifest: aBrokerClaiming("gate")}})
if len(got) != 1 || got[0].Module != "gate" {
t.Fatalf("the gate's view does not hold the pass: %+v", got)
}
if said.Len() != 0 {
t.Errorf("the throwaway composition said %q (a write was tried)", said.String())
}
}
+38
View File
@@ -234,6 +234,16 @@ var signalsTable = []signalRow{
newest: func(f *signalFacts) time.Time {
return newestOf(f.batches, func(b batchFacts) time.Time { return b.created })
}},
{Row: "S23", Signal: "a delivery plan that failed is asked of the operator", Emitter: "controller's plan",
Trigger: "each delivery plan in state failed, not a rollout of held updates and not one a person stopped (novox/hq ADR 0449)",
Bound: "none: raised when the plan fails, naming the modules that failed, their machines and the failure's own " +
"words, with Retry when `plans retry` would run and the refusal's words when it would not; kept a week " +
"from the plan's last change, and ended when the plan is no longer failed",
Kind: kindPlanFailed, Severity: conditions.Warning, Phase: 3,
needs: func(f *signalFacts) error { return f.failedErr }, watch: watchPlansFailed,
newest: func(f *signalFacts) time.Time {
return newestOf(f.failed, func(p failedPlanFacts) time.Time { return p.since })
}},
{Row: "S20", Signal: "a send refused for the assignment generation it was composed from",
Emitter: "node-engine", Trigger: "each refusal (novox/hq issue 234)",
Bound: "none: raised at the first refusal, naming its sender, the generation it came from and the one the " +
@@ -404,6 +414,34 @@ func watchPlans(f *signalFacts) []conditions.Observation {
return out
}
// kindPlanFailed is S23's kind: plan.<id>.plan-failed.
const kindPlanFailed = "plan-failed"
// planFailedWithin is how long a failed plan is kept as a condition from its last change (ADR 0449 §1 names no
// bound; a failed plan stays failed for ever, and a condition for every one the mesh has ever had is no question to
// the operator). A plan failed longer ago than this is left to `plans`.
const planFailedWithin = 7 * 24 * time.Hour
// watchPlansFailed is S23: a delivery plan that failed, said to the operator at once with a Retry answer when
// `plans retry` would run it (novox/hq ADR 0449 §1). Never a retry of its own: the answer is the operator's (§2).
// The condition ends when the plan is no longer failed, because it is no longer among the facts.
func watchPlansFailed(f *signalFacts) []conditions.Observation {
var out []conditions.Observation
for _, p := range f.failed {
if f.now.Sub(p.since) > planFailedWithin {
continue
}
headline, explanation, needs, actions, resolved := planFailedWords(p)
out = append(out, conditions.Observation{Scope: conditions.ScopePlan, ID: p.id, Kind: kindPlanFailed,
Also: p.machines(), Severity: conditions.Warning, Resolver: conditions.ResolverOperator,
Summary: fmt.Sprintf("the delivery plan %s of %s %s failed: %s", p.id, p.repository, short(p.commit),
p.failureSaid()),
Said: p.failureSaid() + p.refusalSaid(),
Headline: headline, Explanation: explanation, Needs: needs, Actions: actions, Resolved: resolved})
}
return out
}
// kindWalkWaiting is S16's kind: plan.<id>.waiting.
const kindWalkWaiting = "waiting"
+11
View File
@@ -160,6 +160,17 @@ var suppressions = map[string]suppression{
created: f.now.Add(-11 * time.Minute), atMostWindow: 10 * time.Minute}}
},
},
// A failed delivery plan is said when it fails and kept a week from its last change (novox/hq ADR 0449).
"S23": {
inside: func(f *signalFacts) {
f.failed = []failedPlanFacts{{id: "plan-1", repository: "novox/app", commit: "c0ffee11", modules: []string{"app"},
since: f.now.Add(-planFailedWithin - time.Second)}}
},
past: func(f *signalFacts) {
f.failed = []failedPlanFacts{{id: "plan-1", repository: "novox/app", commit: "c0ffee11", modules: []string{"app"},
since: f.now.Add(-time.Second)}}
},
},
// A batch queued behind an open walk past that walk's bound, named.
"S19": {
inside: func(f *signalFacts) {
@@ -171,7 +171,9 @@ func TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToT
}}
first, stopFirst := context.WithCancel(ctx)
defer stopFirst()
startRolloutEvents(first, open, js)
waitFirst := startRolloutEvents(first, open, js)
// Ends with the test before the cleanups restore what the look reads, which run after it (issue 1338).
t.Cleanup(func() { stopFirst(); waitFirst() })
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
+4
View File
@@ -55,6 +55,9 @@ type signalFacts struct {
waits []waitFacts
// batches are the batches not yet cut (S18, S19, novox/hq ADR 0276).
batches []batchFacts
// failed are the delivery plans that failed and are asked of the operator (S23, novox/hq ADR 0449).
failed []failedPlanFacts
failedErr error
// refusedSends are the sends a machine refused within the hour for the generation they were composed
// from, each naming its sender (S20, novox/hq issue 234).
refusedSends []inventory.Send
@@ -353,6 +356,7 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
if f.plansErr == nil {
f.batches, f.plansErr = gatherBatches(ctx, inv, now)
}
f.failed, f.failedErr = gatherFailedPlans(ctx, inv, now)
f.loop, f.loopErr = w.gatherLoop()
f.asks, f.asksErr = w.gatherAsks(ctx, inv)
if open, err := w.keeper.Open(ctx); err != nil {
+2 -2
View File
@@ -67,8 +67,8 @@ func TestTheGatesBucketsAreWritableByItsMachinesToolRunner(t *testing.T) {
if !canRead(runner, read, "k") {
t.Errorf("novox's tool runner may not read %s, which the gate declares reading", read)
}
// (The runner may write the teller's answers: the teller declares no runs-as, so the runner carries it too. See the report of issue 1326.)
if canWrite(runner, read, "k") && read != "mesh-delivery-teller_answers" {
// 1329/AC3: the runner no longer carries the teller, so it may not write the teller's answers either.
if canWrite(runner, read, "k") {
t.Errorf("novox's tool runner may write %s, which the gate only reads", read)
}
}
+205 -2
View File
@@ -13,8 +13,9 @@ import (
// and streams, `reads` watching another module's bucket and never writing it, `consumes` hearing the named events,
// `invokes` calling the named tools, `uses` the seat operator-channel's by-caller subjects under the module's own name.
//
// The three manifests are copied from mesh-delivery's main at 6a23627426c6 (modules/<module>/module.json, read
// 2026-10-19), whole, so a change to one of them there that changes its grant is a change to read here.
// The four manifests (the watchman's is the fourth, issue 1329) are copied from mesh-delivery's main at 9c15112d935a
// (modules/<module>/module.json, read 2026-10-20; the first three are unchanged since 6a23627426c6), whole, so a change
// to one of them there that changes its grant is a change to read here.
const bridgeManifest = `{
"module": "mesh-delivery-forge-bridge",
"version": "1",
@@ -160,6 +161,69 @@ const indexerManifest = `{
}
}`
const watchmanManifest = `{
"module": "mesh-delivery-watchman",
"version": "1",
"slug": "watchman",
"reads": [
"mesh-controller.checks-running"
],
"state": [
{ "name": "presence", "history": 20 },
{ "name": "pulse", "key-ttl": true },
{ "name": "stuck", "history": 20 }
],
"own-secrets": {
"broker": "${dir:state}/broker"
},
"secrets-owner": "mesh-delivery-watchman",
"resources": [
{
"id": "account",
"type": "user",
"name": "mesh-delivery-watchman",
"shell": "/usr/bin/nologin",
"home": "/var/lib/mesh-delivery-watchman"
},
{
"id": "state",
"type": "directory",
"mode": "0700",
"owner": "mesh-delivery-watchman",
"place": ".",
"unknown": "program"
},
{
"id": "watchman",
"type": "process",
"name": "mesh-delivery-watchman",
"artifact": "program",
"run": [
"./mesh-delivery-watchman"
],
"env": {
"MESH_BROKER_FILE": "${dir:state}/broker"
},
"user": "mesh-delivery-watchman",
"health": {
"kind": "unit"
}
}
],
"build": {
"artifacts": [
{
"name": "program",
"kind": "bundle",
"language": "go",
"system": "arch",
"from": "cmd/mesh-delivery-watchman",
"binary": "mesh-delivery-watchman"
}
]
}
}`
const tellerManifest = `{
"module": "mesh-delivery-teller",
"version": "1",
@@ -488,3 +552,142 @@ func TestNoPartsCredentialReachesBeyondItsDeclaration(t *testing.T) {
}
}
}
// toolServer is a module that serves a tool on its own broker credential: the runner carries it (ADR 0175).
func toolServer() Declared {
return Declared{Module: "some-tools", Serves: []string{"some_tool"},
State: []Bucket{{Module: "some-tools", Name: "kept", History: 1}}}
}
// 1329/AC1: composed from the manifests of mesh-delivery's main, the machine's tool runner may write no bucket of the
// forge bridge, the indexer, the teller or the watchman, publish none of their stream events nor the teller's
// operator-channel accept subjects, and pull none of their consumers; each part's own credential still may.
func TestTheToolRunnerCarriesNoneOfTheDeliverysParts(t *testing.T) {
streams := map[string]ModuleStream{"mesh-delivery-forge-bridge": {Module: "mesh-delivery-forge-bridge", Name: "FORGE"}}
all := composed(t, Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {
{Module: RuntimeModule}, declaredOf(t, bridgeManifest, streams), declaredOf(t, indexerManifest, streams),
declaredOf(t, tellerManifest, streams), declaredOf(t, watchmanManifest, streams),
}}})
runner := all["runner@novox"]
buckets := map[string][]string{
"mesh-delivery-forge-bridge": {"heads"}, "mesh-delivery-indexer": {"candidates", "modules"},
"mesh-delivery-teller": {"answers"}, "mesh-delivery-watchman": {"presence", "pulse", "stuck"},
}
for part, names := range buckets {
own, there := all[part+"@novox"]
if !there {
t.Errorf("%s has no credential of its own", part)
continue
}
for _, name := range names {
if canWrite(runner, part+"_"+name, "some.key") {
t.Errorf("novox's tool runner may write %s_%s", part, name)
}
if !canWrite(own, part+"_"+name, "some.key") {
t.Errorf("%s may not write its own bucket %s", part, name)
}
}
}
for _, subject := range []string{
"mesh.stream.mesh-delivery-forge-bridge.event.forge.merged",
"mesh.seat.operator-channel.accept.ask.mesh-delivery-teller",
"mesh.seat.operator-channel.accept.cancel.mesh-delivery-teller",
"mesh.seat.operator-channel.accept.condition.mesh-delivery-teller",
"$JS.API.CONSUMER.MSG.NEXT.FORGE.novox_mesh-delivery-indexer",
} {
if allowed(runner.Publish, subject) {
t.Errorf("novox's tool runner may publish %s", subject)
}
}
if !allowed(all["mesh-delivery-forge-bridge@novox"].Publish, "mesh.stream.mesh-delivery-forge-bridge.event.forge.merged") ||
!allowed(all["mesh-delivery-teller@novox"].Publish, "mesh.seat.operator-channel.accept.ask.mesh-delivery-teller") ||
!allowed(all["mesh-delivery-indexer@novox"].Publish, "$JS.API.CONSUMER.MSG.NEXT.FORGE.novox_mesh-delivery-indexer") {
t.Error("a part's own credential lost what its manifest grants")
}
}
// 1329/AC2: a module with tools and a broker secret of its own stays carried by the tool runner, beside its own
// credential (ADR 0175); a module with neither tools, a seat, waits nor a check is not.
func TestTheToolRunnerStillCarriesAModuleWithToolsAndItsOwnCredential(t *testing.T) {
process := Declared{Module: "only-process", State: []Bucket{{Module: "only-process", Name: "mine", History: 1}}}
all := composed(t, Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {
{Module: RuntimeModule}, toolServer(), process,
}}})
runner := all["runner@novox"]
if !canWrite(runner, "some-tools_kept", "k") {
t.Error("novox's tool runner does not carry a module with tools")
}
if !canWrite(all["some-tools@novox"], "some-tools_kept", "k") {
t.Error("the module with tools lost its own credential")
}
if canWrite(runner, "only-process_mine", "k") {
t.Error("novox's tool runner carries a module that has nothing to serve")
}
if !canWrite(all["only-process@novox"], "only-process_mine", "k") {
t.Error("the process module lost its own credential")
}
for name, d := range map[string]Declared{
"a held seat": {Module: "m", Holds: []Seat{{Name: "some-seat"}}},
"a wait": {Module: "m", Waits: []Wait{{}}},
"a check": {Module: "m", Checks: []string{"m.check"}},
} {
if !runnerServes(d) {
t.Errorf("the tool runner does not serve a module with %s", name)
}
}
}
// routeProxyExcerpt is the part of mesh-control's modules/route-proxy/module.json at 70cb217601f9 (read 2026-10-20)
// that decides its grants: its state, what it reads, its own broker secret, and the bundle `provider` that the machine's
// runtime loads (`loads`) though the module declares no tools.
const routeProxyExcerpt = `{
"module": "route-proxy",
"state": ["bound"],
"reads": ["mesh-controller.assignments"],
"own-secrets": { "broker": "${dir:mesh-state}/broker" },
"build": { "artifacts": [
{ "name": "provider", "kind": "bundle", "language": "go", "from": "cmd/route-provider", "binary": "route-provider", "loads": ["route-provider"] }
] }
}`
// 1329 (review): a module with no tools whose bundle the runtime loads stays carried, because its helper writes its state
// and reads on the runner's account; the delivery's five parts are still dropped.
func TestTheToolRunnerStillCarriesAModuleWhoseBundleItLoads(t *testing.T) {
var m struct {
Module string `json:"module"`
State []string `json:"state"`
Reads []string `json:"reads"`
Build struct {
Artifacts []struct {
Loads []string `json:"loads"`
} `json:"artifacts"`
} `json:"build"`
}
if err := json.Unmarshal([]byte(routeProxyExcerpt), &m); err != nil {
t.Fatal(err)
}
proxy := Declared{Module: m.Module, Reads: m.Reads}
for _, s := range m.State {
proxy.State = append(proxy.State, Bucket{Module: m.Module, Name: s, History: 1})
}
for _, a := range m.Build.Artifacts {
proxy.Loads = proxy.Loads || len(a.Loads) > 0
}
streams := map[string]ModuleStream{"mesh-delivery-forge-bridge": {Module: "mesh-delivery-forge-bridge", Name: "FORGE"}}
all := composed(t, Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {
{Module: RuntimeModule}, proxy, declaredOf(t, bridgeManifest, streams), declaredOf(t, indexerManifest, streams),
declaredOf(t, tellerManifest, streams), declaredOf(t, watchmanManifest, streams),
}}})
runner := all["runner@novox"]
if !canWrite(runner, "route-proxy_bound", "k") {
t.Error("novox's tool runner does not carry route-proxy, whose bundle it loads")
}
if !canWrite(all["route-proxy@novox"], "route-proxy_bound", "k") {
t.Error("route-proxy lost its own credential")
}
for _, b := range []string{"mesh-delivery-teller_answers", "mesh-delivery-forge-bridge_heads", "mesh-delivery-indexer_modules", "mesh-delivery-watchman_stuck"} {
if canWrite(runner, b, "k") {
t.Errorf("novox's tool runner may write %s", b)
}
}
}
+1 -1
View File
@@ -154,7 +154,7 @@ func TestTheToolRunnerNeverHearsARestrictedStream(t *testing.T) {
records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{
"novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}, {Module: "news", Streams: []ModuleStream{newsStream}}},
"ace": {{Module: RuntimeModule},
{Module: "carried", Hears: []Heard{mailHeard, newsHeard}, HearsAStream: true},
{Module: "carried", NoAccount: true, Hears: []Heard{mailHeard, newsHeard}, HearsAStream: true},
{Module: "mailer", RunsAs: "mailer", Hears: []Heard{mailHeard}, HearsAStream: true}},
}}
users, err := Users(records)
+16 -1
View File
@@ -69,6 +69,10 @@ type Declared struct {
// Checks are the module's own tools its health asks, each `<module>.<tool>` (novox/hq ADR 0240, to-be
// 48 §3): the machine's node-engine asks them of its own node tools, and is granted that and no more.
Checks []string
// Loads says the module has a compiled bundle whose entrypoints the machine's tool runtime loads and runs
// (novox/hq ADR 0175; issue 1329): a helper that reaches the bus on the runner's account, so the runner
// must carry the module although it declares no tools.
Loads bool
// RunsAs is the account the module runs as in a runtime of its own (novox/hq ADR 0259 §8): it is never
// carried by the machine's runtime, and reaches the bus on its own account.
RunsAs string
@@ -206,7 +210,7 @@ func Users(r Records) ([]Principal, error) {
if runtimeHere {
var carried []Declared
for _, d := range r.Assigned[node] {
if d.RunsAs == "" {
if d.RunsAs == "" && runnerServes(d) {
// Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1).
d.Hears = ComposedHears(d)
carried = append(carried, d)
@@ -391,3 +395,14 @@ func boundBucketsOf(r Records) []string {
sort.Strings(out)
return out
}
// runnerServes says the machine's tool runner has something of this module to serve (novox/hq ADR 0436 §1,
// ADR 0437 §4, issue 1329): tools or loads it answers, a seat it holds, event families a session waits on,
// a bundle the runtime loads (Loads), or a check its health asks. A module with none of these is a process of its own on its own credential, and
// carrying it would hand every session on the machine the runner's account with that module's state, seat
// traffic and streams. A module with no account of its own (NoAccount) is spoken for by the runner, so it
// stays carried, and so does the runtime module itself. Whether the module has its own broker secret is not
// the test: a module with tools and its own credential is still carried (ADR 0175).
func runnerServes(d Declared) bool {
return d.Module == RuntimeModule || d.NoAccount || d.Loads || len(d.Serves) > 0 || len(d.Holds) > 0 || len(d.Waits) > 0 || len(d.Checks) > 0
}
+85
View File
@@ -0,0 +1,85 @@
package catalogue
import (
"fmt"
"sort"
)
// An assignment as the passing of a seat reads it: a module on a machine, with its manifest.
type Assignment struct {
Node string
Manifest Manifest
}
// SeatPass is one seat whose holder on record is assigned nowhere, and the only assignment that can take it.
type SeatPass struct {
Seat Seat
From Held
To Held
}
// SeatsToPassOn reads the mesh for seats held once whose holder on record has left (novox/hq ADR 0449 §4).
//
// **A holder on record that is assigned on no machine holds nothing, and a seat that answers nothing is
// the day a delivery removed its module.** Of such a seat, exactly one assignment in the mesh able to hold
// it (it claims the seat at its scope and CanHold accepts it) takes it over: that is a pass, and the caller
// records it and writes the hand act. With no such assignment the record stays; with two or more nothing is
// recorded and the answer says so, naming both and the `seat --to` that settles it. A holder still assigned
// anywhere changes nothing: the other claimants stand beside it, eligible and silent (ADR 0131). Only mesh
// seats held once are read: a replicated seat's holders are added by an act (ADR 0223), a kinded bench is held
// by its claims, and a seat with several holders on record is a store that disagrees with the mesh (refused
// by the resolver).
func SeatsToPassOn(holdings []Held, assignments []Assignment) ([]SeatPass, []string) {
assigned := map[string]bool{}
for _, a := range assignments {
assigned[a.Manifest.Module] = true
}
bySeat := map[string][]Held{}
var order []string
for _, h := range holdings {
s, ok := SeatNamed(h.Claim)
if !ok || s.Replicated || s.Scope != ScopeMesh || h.Scope != ScopeMesh || KindedBenches[s.Name] {
continue
}
if _, seen := bySeat[s.Name]; !seen {
order = append(order, s.Name)
}
bySeat[s.Name] = append(bySeat[s.Name], h)
}
sort.Strings(order)
var passes []SeatPass
var said []string
for _, name := range order {
recs := bySeat[name]
if len(recs) != 1 || assigned[recs[0].Module] {
continue
}
seat, _ := SeatNamed(name)
var claimants []Held
for _, a := range assignments {
if CanHold(a.Manifest, seat) != nil {
continue
}
claimants = append(claimants, Held{Claim: seat.Name, Scope: seat.Scope, Node: a.Node, Module: a.Manifest.Module})
}
sort.Slice(claimants, func(i, j int) bool {
if claimants[i].Node != claimants[j].Node {
return claimants[i].Node < claimants[j].Node
}
return claimants[i].Module < claimants[j].Module
})
switch len(claimants) {
case 0:
case 1:
passes = append(passes, SeatPass{Seat: seat, From: recs[0], To: claimants[0]})
default:
said = append(said, fmt.Sprintf(
"%s on %s is on record as the holder of %q and is assigned nowhere, and %s on %s and %s on %s both claim it "+
"— nothing is recorded; `seat %s --to <node>/<module>` records the one that holds it",
recs[0].Module, recs[0].Node, seat.Name, claimants[0].Module, claimants[0].Node,
claimants[1].Module, claimants[1].Node, seat.Name))
}
}
return passes, said
}
+81
View File
@@ -0,0 +1,81 @@
package catalogue
import (
"strings"
"testing"
)
// **A seat held once whose holder on record is assigned nowhere passes to its only claimant** (novox/hq ADR
// 0449 §4, issue 1317).
func onRecord(node, module string) []Held {
return []Held{{Claim: "mesh-broker", Scope: ScopeMesh, Node: node, Module: module}}
}
// 1317/AC1
func TestASeatWhoseHolderIsAssignedNowhereIsPassedToItsOnlyClaimant(t *testing.T) {
busSeatDelivering(t, "mesh-bus")
passes, said := SeatsToPassOn(onRecord("novox", "old-broker"),
[]Assignment{{Node: "novox", Manifest: newBroker()}})
if len(said) != 0 || len(passes) != 1 {
t.Fatalf("the only claimant was not given the seat: passes=%v said=%v", passes, said)
}
p := passes[0]
if p.Seat.Name != "mesh-broker" || p.From.Module != "old-broker" || p.From.Node != "novox" ||
p.To.Module != "new-broker" || p.To.Node != "novox" {
t.Fatalf("the pass does not name the seat, the former holder and the new one: %+v", p)
}
}
// 1317/AC2
func TestWithTwoClaimantsAndTheHolderGoneNothingIsPassedAndTheRefusalNamesBoth(t *testing.T) {
busSeatDelivering(t, "mesh-bus")
passes, said := SeatsToPassOn(onRecord("novox", "gone"), []Assignment{
{Node: "novox", Manifest: newBroker()}, {Node: "ace", Manifest: oldBroker()}})
if len(passes) != 0 || len(said) != 1 {
t.Fatalf("two claimants: passes=%v said=%v", passes, said)
}
for _, want := range []string{"old-broker", "new-broker", "seat mesh-broker --to"} {
if !strings.Contains(said[0], want) {
t.Fatalf("the refusal does not say %q: %s", want, said[0])
}
}
}
// 1317/AC3
func TestAHolderStillAssignedSomewhereChangesNothing(t *testing.T) {
busSeatDelivering(t, "mesh-bus")
// Assigned on another machine than the record names: still assigned somewhere.
passes, said := SeatsToPassOn(onRecord("novox", "old-broker"), []Assignment{
{Node: "ace", Manifest: oldBroker()}, {Node: "novox", Manifest: newBroker()}})
if len(passes) != 0 || len(said) != 0 {
t.Fatalf("a holder still assigned lost its seat: passes=%v said=%v", passes, said)
}
}
// 1317/AC4
func TestWithNoClaimantTheRecordStays(t *testing.T) {
busSeatDelivering(t, "mesh-bus")
passes, said := SeatsToPassOn(onRecord("novox", "old-broker"), []Assignment{
{Node: "novox", Manifest: Manifest{Module: "other"}}})
if len(passes) != 0 || len(said) != 0 {
t.Fatalf("with no claimant: passes=%v said=%v", passes, said)
}
// A claimant that cannot hold the seat (it does not provide what the seat delivers) is no claimant.
bad := Manifest{Module: "no-bus", Claims: []Claim{{Name: "mesh-broker", Scope: ScopeMesh}}}
passes, said = SeatsToPassOn(onRecord("novox", "old-broker"), []Assignment{{Node: "novox", Manifest: bad}})
if len(passes) != 0 || len(said) != 0 {
t.Fatalf("a module that cannot hold the seat was passed it: passes=%v said=%v", passes, said)
}
}
// A replicated seat is never passed: its holders are added by an act (ADR 0223).
func TestAReplicatedSeatIsNeverPassed(t *testing.T) {
record := []Held{{Claim: "mesh-dns-resolver", Scope: ScopeMesh, Node: "novox", Module: "gone"}}
resolver := Manifest{Module: "resolver", Provides: []Offer{{Name: "wildcard-resolution", Scope: ScopeMesh}},
Claims: []Claim{{Name: "mesh-dns-resolver", Scope: ScopeMesh}}}
passes, _ := SeatsToPassOn(record, []Assignment{{Node: "novox", Manifest: resolver}})
if len(passes) != 0 {
t.Fatalf("a replicated seat was passed: %v", passes)
}
}
+6
View File
@@ -201,6 +201,12 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
// name (novox/hq ADR 0305).
Issued: issuedTools(m),
}
// Whether the node's runtime loads a bundle of it: a helper that runs on the runner's account (issue 1329).
for _, b := range m.Bundles {
if len(b.Loads) > 0 {
d.Loads = true
}
}
// Whether it can be given an account at all: delivered as its own secret named broker, so one
// that declares none has nowhere to read it (novox/hq issue 195).
if _, reads := m.OwnSecrets["broker"]; !reads {
+7
View File
@@ -85,6 +85,7 @@ func (i *Inventory) UseChain(c *chain.Chain) {
// beside the real one, as the merge gate raises the facts snapshot's mesh in a throwaway store (issue 1121). Nothing of it
// reaches the bus.
func (i *Inventory) UseThrowawayChain() {
i.MarkThrowaway()
i.UseChain(chain.InMemory(nil))
i.markLaidOut()
}
@@ -129,3 +130,9 @@ func (i *Inventory) theChain(ctx context.Context) (*chain.Chain, error) {
}
return c, nil
}
// MarkThrowaway says this inventory is a mesh raised for a moment: a composition on it writes nothing outside it.
func (i *Inventory) MarkThrowaway() { i.throwaway = true }
// Throwaway says whether this inventory is a mesh raised for a moment (UseThrowawayChain).
func (i *Inventory) Throwaway() bool { return i != nil && i.throwaway }
+21
View File
@@ -153,3 +153,24 @@ func TestAReplicatedSeatHasSeveralHoldersOnRecord(t *testing.T) {
t.Fatalf("a handover left other holders on record: %+v", held)
}
}
// A pass is a compare-and-set: it moves the record only while the holder it was read with is still on it.
func TestPassingASeatMovesTheRecordOnceAndOnlyFromTheHolderRead(t *testing.T) {
inv, ctx := twoBrokersOnTwoNodes(t)
if err := inv.HoldSeat(ctx, "mesh-broker", catalogue.ScopeMesh, "anchor", "old-broker"); err != nil {
t.Fatal(err)
}
from := catalogue.Held{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "anchor", Module: "old-broker"}
to := catalogue.Held{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "laptop", Module: "new-broker"}
passed, err := inv.PassSeat(ctx, "mesh-broker", catalogue.ScopeMesh, from, to)
if err != nil || !passed {
t.Fatalf("the seat was not passed: %v %v", passed, err)
}
held, _ := inv.Holdings(ctx)
if len(held) != 1 || held[0].Node != "laptop" || held[0].Module != "new-broker" {
t.Fatalf("the record does not name the new holder alone: %+v", held)
}
if passed, err = inv.PassSeat(ctx, "mesh-broker", catalogue.ScopeMesh, from, to); err != nil || passed {
t.Fatalf("a second pass from the same read moved the record again: %v %v", passed, err)
}
}
+3
View File
@@ -39,6 +39,9 @@ type Inventory struct {
state *chainState
// recorded, when set, takes the durations a save of the chain measures in place of the store: a test with no store.
recorded func(context.Context, Duration) error
// throwaway is set on an inventory raised for a moment beside the real mesh (the merge gate's): nothing it
// composes is to be written to the store or the bus.
throwaway bool
}
// Open connects to the inventory store.
+3
View File
@@ -340,6 +340,9 @@ type PlanModule struct {
State string `json:"state,omitempty"`
AskedAt *time.Time `json:"asked_at,omitempty"`
BuiltAt *time.Time `json:"built_at,omitempty"`
// BusWaitFrom is when the bus first did not answer an ask of this module (issue 1334): the module stays unasked and is
// asked again at the next look, until the bus has been silent for the bound, when the plan fails as for any refusal.
BusWaitFrom *time.Time `json:"bus_wait_from,omitempty"`
// SentAt is when the plan sent the machines running this module its new build, because a
// later tier is built by it (ADR 0163's gate): the reports that open the gate are the ones
// after this.
+42
View File
@@ -284,6 +284,48 @@ func (i *Inventory) HoldSeat(ctx context.Context, seat, scope, nodeName, module
return tx.Commit(ctx)
}
// PassSeat records `to` as the holder of a seat held once in place of `from`, only while `from` is still the
// one on record: a compare-and-set in one transaction, so two compositions reading the same record cannot both
// hand the seat over (novox/hq ADR 0449 §4). It answers false, with no error, when the record no longer names
// `from` (another composition passed it first). The new holder must be an assignment.
func (i *Inventory) PassSeat(ctx context.Context, seat, scope string, from, to catalogue.Held) (bool, error) {
fromNode, err := i.NodeByName(ctx, from.Node)
if err != nil {
return false, err
}
toNode, err := i.NodeByName(ctx, to.Node)
if err != nil {
return false, err
}
if err := i.mustBeAssigned(ctx, seat, to.Node, to.Module); err != nil {
return false, err
}
tx, err := i.store.Pool().Begin(ctx)
if err != nil {
return false, err
}
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
gone, err := tx.Exec(ctx,
`delete from seat_holding where seat = $1 and node = $2 and module = $3`,
seat, fromNode.ID, from.Module)
if err != nil {
return false, fmt.Errorf("passing %s from %s on %s: %w", seat, from.Module, from.Node, err)
}
if gone.RowsAffected() == 0 {
return false, nil
}
if _, err := tx.Exec(ctx,
`delete from seat_holding where seat = $1`, seat); err != nil {
return false, fmt.Errorf("passing %s: %w", seat, err)
}
if _, err := tx.Exec(ctx,
`insert into seat_holding (seat, scope, node, module) values ($1, $2, $3, $4)`,
seat, scope, toNode.ID, to.Module); err != nil {
return false, fmt.Errorf("recording %s on %s as the holder of %s: %w", to.Module, to.Node, seat, err)
}
return true, tx.Commit(ctx)
}
// AddSeatHolder records one more assignment holding a replicated seat, beside those already on
// record (novox/hq ADR 0223). Whether the seat may have several holders is the caller's to judge —
// the seat's definition is compiled, and the store holds no column for it. Recording a holder