diff --git a/cmd/mesh-controller/batches_test.go b/cmd/mesh-controller/batches_test.go index 846804ca..23557cfb 100644 --- a/cmd/mesh-controller/batches_test.go +++ b/cmd/mesh-controller/batches_test.go @@ -226,7 +226,7 @@ func TestTheAssemblingBatchIsShown(t *testing.T) { asksWithPaths(t) var said []inventory.Plan was := inventory.PlanSaved - inventory.PlanSaved = func(p inventory.Plan) { said = append(said, p) } + inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { said = append(said, p) } t.Cleanup(func() { inventory.PlanSaved = was }) now := time.Now().UTC() claude := catalogueMerge("553b7191claude", "app", now.Add(-20*time.Second)) diff --git a/cmd/mesh-controller/delivery.go b/cmd/mesh-controller/delivery.go index 23a0a629..c5f4765f 100644 --- a/cmd/mesh-controller/delivery.go +++ b/cmd/mesh-controller/delivery.go @@ -10,6 +10,7 @@ import ( "slices" "sort" "strings" + "sync" "time" "github.com/nats-io/nats.go" @@ -1253,9 +1254,10 @@ func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error return err } bus := link.OverNATS{Conn: conn, JS: js} + was := inventory.PlanSaved checkEvents = bus - inventory.PlanSaved = func(p inventory.Plan) { publishPlanMoved(ctx, bus, facts, p) } - defer func() { checkEvents, inventory.PlanSaved = nil, nil }() + inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { publishPlanMoved(ctx, bus, facts, p) } + defer func() { checkEvents, inventory.PlanSaved = nil, was }() ran = true return f() }) @@ -1265,3 +1267,65 @@ func sayingOnTheBus(ctx context.Context, facts walkFacts, f func() error) error } return err } + +// sayingWhatAVerbKept is PlanSaved for a process that is not the serving controller: each walk the verb keeps is +// said as `plan-moved` before the save returns, with the facts of the inventory that kept it, on the bus the verb +// reaches. It reaches the bus on the first save and never for a verb that keeps no walk. When the bus cannot be +// reached, a save that started or ended a walk says so on the verb's error output: its mark is kept and no later +// save says it, so the router holds what the walk touched until its bound, or not at all for a start. done lets +// the bus go and takes the hook back. +func sayingWhatAVerbKept(ctx context.Context) (said func(*inventory.Inventory, inventory.Plan), done func()) { + var ( + mu sync.Mutex + reached bool + bus link.Bus + closing func() + failed error + ) + said = func(inv *inventory.Inventory, p inventory.Plan) { + mu.Lock() + defer mu.Unlock() + if !reached { + reached = true + if handActConn != nil { + bus, failed = overNATS(handActConn) + } else if js, err := aBus(); err != nil { + failed = err + } else if bus, failed = overNATS(js.Conn()); failed == nil { + closing = js.Close + } else { + js.Close() + } + } + if bus == nil { + switch p.WalkSaid { + case inventory.WalkStarted: + fmt.Fprintf(os.Stderr, "%s: its start is kept and could NOT be said as plan-moved (%v): the router "+ + "does not hold what it touches\n", p.ID, failed) + case inventory.WalkEnded: + fmt.Fprintf(os.Stderr, "%s: its end is kept and could NOT be said as plan-moved (%v): the router holds "+ + "what it touched until its bound\n", p.ID, failed) + } + return + } + publishPlanMoved(ctx, bus, inv, p) + } + done = func() { + mu.Lock() + defer mu.Unlock() + inventory.PlanSaved = nil + if closing != nil { + closing() + } + } + return said, done +} + +// overNATS is the bus a connection carries, with its JetStream. +func overNATS(conn *nats.Conn) (link.Bus, error) { + js, err := conn.JetStream() + if err != nil { + return nil, err + } + return link.OverNATS{Conn: conn, JS: js}, nil +} diff --git a/cmd/mesh-controller/main.go b/cmd/mesh-controller/main.go index 3e4cebec..7269d761 100644 --- a/cmd/mesh-controller/main.go +++ b/cmd/mesh-controller/main.go @@ -63,6 +63,15 @@ func run() error { // 45 §6), so the next controller takes it at once rather than after its age. defer theLease.release() + // Every walk a verb keeps is said on the bus, whichever verb kept it (novox/hq ADR 0308 decision 1): the save + // that starts or ends a walk keeps its mark, and no later save says it again. The serving controller says its + // own saves. + if args[0] != "serve" { + said, done := sayingWhatAVerbKept(ctx) + inventory.PlanSaved = said + defer done() + } + switch args[0] { case "build": return buildCommand(ctx, args[1:]) diff --git a/cmd/mesh-controller/plan_retry.go b/cmd/mesh-controller/plan_retry.go index a41acb69..93a3bcdc 100644 --- a/cmd/mesh-controller/plan_retry.go +++ b/cmd/mesh-controller/plan_retry.go @@ -251,6 +251,16 @@ func resumed(p *inventory.Plan, why string) { if p.State == inventory.PlanFailed && len(failedIn(*p)) == 0 { p.State = inventory.PlanBuilding p.Note = why + goesOnAgain(p) + } +} + +// goesOnAgain clears a stop from a walk that walks again: a stop said why the walk ended, and a walk resumed has +// not ended. Kept, it would end a retried walk that fails again as stopped, not failed (novox/hq ADR 0308 +// decision 1), and keep the search for the merge that brought the failure from running (searchAfterFailures). +func goesOnAgain(p *inventory.Plan) { + if p.Delivery != nil { + p.Delivery.Stopped, p.Delivery.StoppedWhy = "", "" } } @@ -460,6 +470,7 @@ func retryRollouts(ctx context.Context, open *stores, p *inventory.Plan) (string said = append(said, m+" to "+strings.Join(again, ", ")) } p.State = inventory.PlanRolling + goesOnAgain(p) p.Note = fmt.Sprintf("tier %d retried by hand; sent %s first again", p.Tier, strings.Join(said, "; ")) if err := open.inventory.SavePlan(ctx, p); err != nil { return "", err @@ -525,6 +536,7 @@ func retryTierWhole(ctx context.Context, open *stores, p *inventory.Plan, again } if p.State == inventory.PlanFailed && len(failedIn(*p)) == 0 { p.State = inventory.PlanBuilding + goesOnAgain(p) p.Note = fmt.Sprintf("tier %d retried by hand: %s asked again; %s sent again with the tier", p.Tier, strings.Join(again, ", "), orNone(strings.Join(resent, ", "))) } diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index f550fe03..723eb7bc 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -249,7 +249,7 @@ func serve(ctx context.Context) (err error) { // (novox/hq to-be 45 §9). checkEvents = bus // And every walk this controller keeps in a new state, for the delivery it walks (novox/hq ADR 0239). - inventory.PlanSaved = func(p inventory.Plan) { sayPlanMoved(ctx, bus, open.inventory, p) } + inventory.PlanSaved = func(_ *inventory.Inventory, p inventory.Plan) { sayPlanMoved(ctx, bus, open.inventory, p) } if err := server.Checks(following{open}); err != nil { return err } diff --git a/cmd/mesh-controller/walk_said_test.go b/cmd/mesh-controller/walk_said_test.go index a5075508..22ef9a15 100644 --- a/cmd/mesh-controller/walk_said_test.go +++ b/cmd/mesh-controller/walk_said_test.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "os" "reflect" "testing" "time" @@ -389,3 +390,125 @@ func TestABacklogReleasedByAPersonIsSaidStarted(t *testing.T) { t.Fatalf("the release's start: %v", started) } } + +// asTyped runs a verb as the controller's command line runs it, through run, as a person's terminal would. +func asTyped(t *testing.T, args ...string) error { + t.Helper() + was := os.Args + os.Args = append([]string{"mesh-controller"}, args...) + t.Cleanup(func() { os.Args = was }) + err := run() + // run gives back the lease as the process would end; the test's process goes on, and takes it again. + theLease.mu.Lock() + theLease.held, theLease.kv, theLease.noBus = nil, nil, false + theLease.mu.Unlock() + return err +} + +// aStartedWalkAsking keeps a live walk, said started, whose first tier asked a as build. +func aStartedWalkAsking(t *testing.T, open *stores, id, build string, asked time.Time) inventory.Plan { + t.Helper() + walk := inventory.Plan{ID: id, Repository: "novox/a", Branch: "main", Commit: "c0ffee", Created: asked, + State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}}, + Modules: map[string]*inventory.PlanModule{"a": {State: "asked", AskedAt: &asked, Build: build}}, + Times: &inventory.PlanTimes{Cut: &asked, Started: &asked, Class: inventory.ClassLeaf}} + if err := open.inventory.SavePlan(t.Context(), &walk); err != nil { + t.Fatal(err) + } + return walk +} + +// Live, with the store and a bus: `cancel ` and `clear` fail the walk that asked the build, and the verb +// says its end, failed (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247). +func TestAWalkEndedByCancelOrClearIsSaidEnded(t *testing.T) { + _, ask := aBuildQueue(t) + open := aMesh(t) + next := planMovedHeard(t) + twoTiers(t, open) + for _, verb := range [][]string{{"cancel"}, {"clear"}} { + id := link.NewBuildID(time.Now()) + ask(id, "https://forge.example/novox/a.git") + asked, _ := link.BuildAskedAt(id) + walk := aStartedWalkAsking(t, open, "plan-308-"+verb[0], id, asked) + args := verb + if verb[0] == "cancel" { + args = append(args, id) + } + if err := asTyped(t, args...); err != nil { + t.Fatal(err) + } + ended := next(inventory.WalkEnded) + if ended["walk"] != walk.ID || ended["outcome"] != inventory.WalkFailed { + t.Fatalf("%s: the end is %v", verb[0], ended) + } + // Ended, and so the next walk may be cut on the same repository. + p, _ := open.inventory.PlanByID(t.Context(), walk.ID) + if p.State != inventory.PlanFailed { + t.Fatalf("%s left the walk %s", verb[0], p.State) + } + } +} + +// Live, with the store and a bus: `rebuild ` joining a failed walk sets it building again, and the verb +// says its start (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247). +func TestAWalkRestartedByARebuildIsSaidStarted(t *testing.T) { + open := aMesh(t) + asksRecorded(t) + next := planMovedHeard(t) + twoTiers(t, open) + before := time.Now().UTC().Add(-time.Hour) + ended := before.Add(time.Minute) + failed := inventory.Plan{ID: "plan-308-rebuild", Repository: "novox/a", Branch: "main", Commit: "c0ffee", + Created: before, State: inventory.PlanFailed, Tiers: [][]string{{"a"}, {"b"}}, + Modules: map[string]*inventory.PlanModule{"a": {State: "failed", AskedAt: &before, Build: "build-1", + Why: link.CancelledByHand}}, + Times: &inventory.PlanTimes{Cut: &before, Started: &before, Ended: &ended, Class: inventory.ClassLeaf}} + if err := open.inventory.SavePlan(t.Context(), &failed); err != nil { + t.Fatal(err) + } + if err := asTyped(t, "rebuild", "a"); err != nil { + t.Fatal(err) + } + if started := next(inventory.WalkStarted); started["walk"] != failed.ID { + t.Fatalf("the rebuilt walk's start: %v", started) + } +} + +// Live, with the store: a walk a person stopped and then retried, by `plans retry` or by `rebuild`, walks again +// without the stop, so when it fails again it ends failed, not stopped, and the search for the merge that +// brought the failure is not kept from it (novox/hq ADR 0308 decision 1, review 21 on mesh-controller#247). +func TestAWalkStoppedByHandAndRetriedEndsFailedWhenItFailsAgain(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + asksRecorded(t) + twoTiers(t, open) + for _, again := range [][]string{{"plans", "retry"}, {"rebuild", "a"}} { + before := time.Now().UTC().Add(-time.Hour) + walk := aStartedWalkAsking(t, open, "plan-308-again-"+again[0], "build-1", before) + if err := asTyped(t, "plans", "stop", walk.ID, "--why", "a test"); err != nil { + t.Fatal(err) + } + // The build asked before the stop fails after it. + p, _ := open.inventory.PlanByID(ctx, walk.ID) + p.Modules["a"].State, p.Modules["a"].Why = "failed", "it broke" + if err := open.inventory.SavePlan(ctx, &p); err != nil { + t.Fatal(err) + } + args := again + if again[0] == "plans" { + args = append(append([]string(nil), again...), walk.ID) + } + if err := asTyped(t, args...); err != nil { + t.Fatal(err) + } + p, _ = open.inventory.PlanByID(ctx, walk.ID) + if p.State != inventory.PlanBuilding || (p.Delivery != nil && p.Delivery.Stopped != "") { + t.Fatalf("%s: the walk is %s and keeps its stop: %+v", again[0], p.State, p.Delivery) + } + asking, _ := link.BuildAskedAt(p.Modules["a"].Build) + planBuilt(ctx, open, "a", "c0ffee", "it broke again", asking, p.Modules["a"].Build) + if p, _ = open.inventory.PlanByID(ctx, walk.ID); p.State != inventory.PlanFailed || p.Outcome() != inventory.WalkFailed { + t.Fatalf("%s: failing again, the walk is %s and ends %s", again[0], p.State, p.Outcome()) + } + } +} diff --git a/internal/inventory/plans.go b/internal/inventory/plans.go index a0ec1bf8..fa7f3231 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -244,10 +244,12 @@ func (p Plan) Waiting() bool { return p.Delivery != nil && p.Delivery.Awaits != "" && p.Delivery.Go == nil } -// PlanSaved is told every plan this process kept, after it is kept (novox/hq ADR 0239): the serving -// controller says it on the bus as `plan-moved`. Nil in a command, which says nothing; whoever follows a -// walk also asks for it, so a save made by a command is found by comparison. -var PlanSaved func(Plan) +// PlanSaved is told every plan this process kept, after it is kept, with the inventory that kept it (novox/hq +// ADR 0239): the serving controller says it on the bus as `plan-moved`, and so does a command (the controller's +// run installs it for every verb, novox/hq ADR 0308 decision 1), so the save that starts or ends a walk is said +// whichever process made it. Nil says nothing; whoever follows a walk also asks for it, and finds an unsaid save +// by comparison. +var PlanSaved func(*Inventory, Plan) // ErrPlanMoved is a save against a plan written by somebody else since it was read. var ErrPlanMoved = errors.New("the plan was written by somebody else since it was read") @@ -519,7 +521,7 @@ func (i *Inventory) SavePlansMoving(ctx context.Context, plans []*Plan, moves [] p.Epoch = 0 } if PlanSaved != nil { - PlanSaved(*p) + PlanSaved(i, *p) } } return nil diff --git a/internal/inventory/walksaid_test.go b/internal/inventory/walksaid_test.go index 98fa04b6..69cf2e5e 100644 --- a/internal/inventory/walksaid_test.go +++ b/internal/inventory/walksaid_test.go @@ -96,7 +96,7 @@ func TestAWalksStartIsKeptAndARefusedSaveLeavesItUnsaid(t *testing.T) { Times: &PlanTimes{Cut: &cut, Class: ClassLeaf}} var said []string was := PlanSaved - PlanSaved = func(p Plan) { said = append(said, p.WalkSaid) } + PlanSaved = func(_ *Inventory, p Plan) { said = append(said, p.WalkSaid) } t.Cleanup(func() { PlanSaved = was }) if err := inv.SavePlan(ctx, &p); err != nil { t.Fatal(err) diff --git a/testdata/store-tests b/testdata/store-tests index 8b270fb1..d60bca8d 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -19,3 +19,6 @@ ./cmd/mesh-controller TestAWalkStoppedOrClosedByAPersonIsSaidEndedStopped ./cmd/mesh-controller TestAWalkRetriedByAPersonIsSaidStartedAgain ./cmd/mesh-controller TestABacklogReleasedByAPersonIsSaidStarted +./cmd/mesh-controller TestAWalkEndedByCancelOrClearIsSaidEnded +./cmd/mesh-controller TestAWalkRestartedByARebuildIsSaidStarted +./cmd/mesh-controller TestAWalkStoppedByHandAndRetriedEndsFailedWhenItFailsAgain