diff --git a/cmd/mesh-controller/plan_looks.go b/cmd/mesh-controller/plan_looks.go index 10ad6832..6ae5db79 100644 --- a/cmd/mesh-controller/plan_looks.go +++ b/cmd/mesh-controller/plan_looks.go @@ -156,6 +156,10 @@ func startPlanLooks(ctx context.Context, open *stores) { l.ask() } +// applySilentAfter is how long a machine may say nothing on a send before the plan names it silent: the mesh's +// ApplySilentAfter, a variable so a test of the wait needs no quarter of an hour. +var applySilentAfter = inventory.ApplySilentAfter + // planLookPasses is the most times one look goes over the plans again because the first pass changed which are open. const planLookPasses = 6 @@ -316,7 +320,7 @@ func silentWaitsOf(p inventory.Plan, reports []inventory.Reported, now time.Time // silentTTL is how long a wait's key lives: what is left of ApplySilentAfter from the send, and a second over, so the // look it wakes finds the machine silent and not a fraction early. func silentTTL(sent, now time.Time) time.Duration { - left := sent.Add(inventory.ApplySilentAfter).Sub(now) + time.Second + left := sent.Add(applySilentAfter).Sub(now) + time.Second if left < time.Second { return time.Second } @@ -429,7 +433,11 @@ func startRolloutSilent(ctx context.Context, js *broker.JetStream) error { setSilentClock(clock) go func() { <-ctx.Done() - setSilentClock(nil) + silentMu.Lock() + if silentKeyed == silentClock(clock) { + silentKeyed = nil + } + silentMu.Unlock() }() return nil } diff --git a/cmd/mesh-controller/plan_looks_test.go b/cmd/mesh-controller/plan_looks_test.go index fb3dc659..e6612960 100644 --- a/cmd/mesh-controller/plan_looks_test.go +++ b/cmd/mesh-controller/plan_looks_test.go @@ -101,7 +101,7 @@ func lookerCounting(failures *atomic.Int32, looks *atomic.Int32, block chan stru func eventually(t *testing.T, what string, ok func() bool) { t.Helper() - for end := time.Now().Add(3 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) { + for end := time.Now().Add(15 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) { if ok() { return } diff --git a/cmd/mesh-controller/release.go b/cmd/mesh-controller/release.go index 5751b1cd..f7939794 100644 --- a/cmd/mesh-controller/release.go +++ b/cmd/mesh-controller/release.go @@ -1023,12 +1023,12 @@ func advanceRest(ctx context.Context, open *stores, p *inventory.Plan, now time. outcome, reported = provisionedOutcome(node, sent.Only(carried...), byNode[node], reported, provLookup(ctx)) } if !reported { - if now.Sub(*g.Since) > inventory.ApplySilentAfter { + if now.Sub(*g.Since) > applySilentAfter { g.Silent = true r.Silent = append(r.Silent, node) changed = true notes = append(notes, fmt.Sprintf("%s said nothing on its send within %s: left silent", node, - inventory.ApplySilentAfter)) + applySilentAfter)) continue } judging = append(judging, node+": not yet reported on its send") diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index 64e0fa8d..c0b90e63 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -1174,7 +1174,7 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, said = append(said, "failed or refused its send (apply-failing says it): "+strings.Join(failed, ", ")) } if len(silent) > 0 { - said = append(said, fmt.Sprintf("said nothing on its send within %s: %s", inventory.ApplySilentAfter, + said = append(said, fmt.Sprintf("said nothing on its send within %s: %s", applySilentAfter, strings.Join(silent, ", "))) } said = append(said, recordedSaid(*p)...) @@ -1239,7 +1239,7 @@ func walkReports(p inventory.Plan, reports []inventory.Reported, now time.Time, case strings.HasPrefix(outcome, inventory.OutcomeProvisioningFailed): failed = append(failed, node+" ("+outcome+")") } - case !at.IsZero() && now.Sub(at) > inventory.ApplySilentAfter: + case !at.IsZero() && now.Sub(at) > applySilentAfter: seen[node] = true silent = append(silent, node) default: diff --git a/cmd/mesh-controller/rollout_events_test.go b/cmd/mesh-controller/rollout_events_test.go new file mode 100644 index 00000000..7617a7dd --- /dev/null +++ b/cmd/mesh-controller/rollout_events_test.go @@ -0,0 +1,281 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" + "github.com/novox/mesh-controller/internal/testbus" +) + +// Acceptance of the step (novox/hq ADR 0406 §5, §6, acceptance 7; issue 974), live, with the store and a real bus and no +// ticker: a rollout of two tiers advances on the events alone — the plan kept, the machines' reports, a silent key's +// expiry — a machine that reports in time is never named silent and one that never reports is named silent when its +// key expires, and a controller replaced in the middle of it resumes with no tier asked twice and none skipped. +func TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + + // A quarter of an hour of silence is two seconds here. + wasSilent := applySilentAfter + applySilentAfter = 2 * time.Second + t.Cleanup(func() { applySilentAfter = wasSilent }) + + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + if err := broker.AssertMeshStreams(js); err != nil { + t.Fatal(err) + } + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatal(err) + } + legacy, err := js.Conn().JetStream() + if err != nil { + t.Fatal(err) + } + checkEvents = link.OverNATS{Conn: js.Conn(), JS: legacy} + t.Cleanup(func() { checkEvents = nil }) + sentHeard := make(chan *nats.Msg, 32) + sub, err := js.Conn().ChanSubscribe(link.SeatEventSubject(link.MeshControllerSeat, link.KeySent), sentHeard) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + // Every save of a plan is said and asks the look, as the serving controller's hook does. + saved := &movesSaid{revs: map[string][]int64{}} + wasSaved := inventory.PlanSaved + inventory.PlanSaved = func(i *inventory.Inventory, p inventory.Plan) { saved.hear(i, p); planKept(p) } + t.Cleanup(func() { inventory.PlanSaved = wasSaved }) + + // The build machine and the machines, as far as the plan meets them: an ask is counted, a send is recorded and kept. + var mu sync.Mutex + var asks []string + var sends [][]string + wasAsk := askABuild + askABuild = func(_ context.Context, source buildSource, _, _ string) (string, error) { + mu.Lock() + defer mu.Unlock() + id := link.NewBuildID(time.Now().Add(time.Duration(len(asks)) * time.Millisecond)) + asks = append(asks, source.Repository+"#"+id) + return id, nil + } + t.Cleanup(func() { askABuild = wasAsk }) + digests := 0 + wasSend := sendRollout + sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) { + mu.Lock() + defer mu.Unlock() + sends = append(sends, append([]string(nil), names...)) + current, err := open.inventory.CurrentBuilds(ctx) + if err != nil { + return nil, err + } + builds := map[string]string{} + for _, m := range []string{"a", "b"} { + builds[m] = current[m].Commit + } + for _, node := range names { + digests++ + if err := open.inventory.RecordSent(ctx, nodeID(t, open, node), fmt.Sprintf("d-%s-%d", node, digests), builds); err != nil { + return nil, err + } + } + return names, nil + } + t.Cleanup(func() { sendRollout = wasSend }) + + register := func(m, commit string, asked time.Time) { + t.Helper() + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: m, Version: commit}, + inventory.Source{Repository: "novox/" + m, Seat: "git", Path: "modules/" + m, BuiltFrom: commit, Head: commit, + Asked: asked}); err != nil { + t.Fatal(err) + } + } + record := func(id, m, commit string, at time.Time) { + t.Helper() + b := inventory.Build{ID: id, Module: m, Commit: commit, Repository: "novox/" + m, Path: "modules/" + m, Asked: at, At: at} + b.Manifest, _ = json.Marshal(catalogue.Manifest{Module: m, Version: commit}) + if err := inv.RecordBuild(ctx, b); err != nil { + t.Fatal(err) + } + } + old := time.Now().Add(-2 * time.Hour) + for _, m := range []string{"a", "b"} { + record("build-"+m+"-1", m, "c1", old) + register(m, "c1", old) + for _, n := range []string{"anchor", "laptop"} { + if _, err := inv.Assign(ctx, n, m); err != nil { + t.Fatal(err) + } + } + } + for _, n := range []string{"anchor", "laptop"} { + if err := inv.RecordSent(ctx, nodeID(t, open, n), "d-"+n+"-c1", map[string]string{"a": "c1", "b": "c1"}); err != nil { + t.Fatal(err) + } + } + record("build-a-2", "a", "c2", time.Now().Add(-time.Minute)) + register("a", "c2", time.Now().Add(-time.Minute)) + + // A machine's report on what it was sent, heard as the controller hears it. + listener := nudgingListener{Enrolment: link.Enrolment{Inventory: inv}} + seq := int64(0) + reports := func(node string) { + t.Helper() + last, found, err := inv.SentTo(ctx, node) + if err != nil || !found { + t.Fatalf("%s was sent nothing: %v", node, err) + } + seq++ + body, _ := json.Marshal(map[string]any{"node": node, "applied": []string{"x.server"}, "declared": last.Digest, + "report_sequence": seq}) + var report link.Report + if err := json.Unmarshal(body, &report); err != nil { + t.Fatal(err) + } + if _, err := listener.Heard(ctx, report); err != nil { + t.Fatal(err) + } + } + heldKeys := func() map[silentWait]time.Time { + c := silentClockOf() + if c == nil { + return nil + } + held, err := c.Held(ctx) + if err != nil { + t.Fatal(err) + } + return held + } + plan := func() inventory.Plan { + t.Helper() + p, err := inv.PlanByID(ctx, "plan-974") + if err != nil { + t.Fatal(err) + } + return p + } + + // 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() + if err := startRolloutSilent(first, js); err != nil { + t.Fatal(err) + } + startPlanLooks(first, open) + 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"}}, + Modules: map[string]*inventory.PlanModule{ + "a": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-a-2"}, + "b": {}, + }} + if err := inv.SavePlan(ctx, &p); err != nil { + t.Fatal(err) + } + eventually(t, "tier 0 sent to both machines and tier 1 asked, on the save alone", func() bool { + mu.Lock() + defer mu.Unlock() + return len(sends) == 1 && len(asks) == 1 && plan().Tier == 1 + }) + eventually(t, "a key on the bus for each machine tier 0 was sent to", func() bool { return len(heldKeys()) == 2 }) + + // The controller is replaced: its looks and its watcher end. anchor reports meanwhile, to nobody. + stopFirst() + time.Sleep(300 * time.Millisecond) + reports("anchor") + + second, stopSecond := context.WithCancel(ctx) + defer stopSecond() + if err := startRolloutSilent(second, js); err != nil { + t.Fatal(err) + } + startPlanLooks(second, open) + // 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() + _, anchor := h[silentWait{Plan: "plan-974", Tier: 0, Node: "anchor"}] + return !anchor + }) + + // tier 1's build comes in, as the build machine's outcome is taken in. + b := plan().Modules["b"] + if b == nil || b.Build == "" { + t.Fatalf("tier 1 was not asked: %+v", b) + } + askedAt, _ := link.BuildAskedAt(b.Build) + record(b.Build, "b", "c2", askedAt) + register("b", "c2", askedAt) + planBuilt(ctx, open, "b", "c2", "", askedAt, b.Build) + eventually(t, "tier 1 sent to both machines", func() bool { + mu.Lock() + defer mu.Unlock() + return len(sends) == 2 + }) + reports("anchor") + + // laptop never reports: its key expires and the new controller names it silent, which ends the rollout. + eventually(t, "the rollout ended, naming laptop silent and not anchor", func() bool { + p := plan() + return p.State == inventory.PlanDone + }) + p = plan() + if !strings.Contains(p.Note, "laptop") || strings.Contains(p.Note, "anchor") || !strings.Contains(p.Note, "said nothing") { + t.Errorf("the plan ended with the note %q; want laptop named silent and not anchor", p.Note) + } + eventually(t, "no key left on the bus", func() bool { return len(heldKeys()) == 0 }) + + // No tier asked twice and none skipped: one ask for tier 1 (tier 0 was built when the plan was kept), each tier sent + // once to both machines, and said as sent once for each. + mu.Lock() + if len(asks) != 1 || !strings.HasPrefix(asks[0], "novox/b#") { + t.Errorf("asks = %v; want tier 1's module asked once", asks) + } + if len(sends) != 2 || strings.Join(sends[0], ",") != "anchor,laptop" || strings.Join(sends[1], ",") != "anchor,laptop" { + t.Errorf("sends = %v; want both machines sent each tier once", sends) + } + mu.Unlock() + said := map[string]int{} + for deadline := time.After(5 * time.Second); len(said) < 4; { + select { + case m := <-sentHeard: + var body planSentBody + if err := json.Unmarshal(m.Data, &body); err != nil { + t.Fatal(err) + } + said[fmt.Sprintf("%d/%s", body.Tier, body.Node)]++ + case <-deadline: + t.Fatalf("sent was said for %v; want tiers 0 and 1 to both machines", said) + } + } + time.Sleep(300 * time.Millisecond) + for len(sentHeard) > 0 { + var body planSentBody + _ = json.Unmarshal((<-sentHeard).Data, &body) + said[fmt.Sprintf("%d/%s", body.Tier, body.Node)]++ + } + for k, n := range said { + if n != 1 { + t.Errorf("sent for %s was said %d times", k, n) + } + } + // And every move of the plan was said as plan-moved. + saved.noUnsaid(t, open) +} diff --git a/testdata/store-tests b/testdata/store-tests index 3d3dff3a..8a116cec 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -134,3 +134,4 @@ ./cmd/mesh-controller TestMergesHeardWhileAnotherHolderHasThePlansAreRecordedAndCarried ./cmd/mesh-controller TestABatchWindowClosesByTheExpiryOfItsKey ./cmd/mesh-controller TestEveryMoveAMergeAndTheCutMakeOfAPlanIsSaid +./cmd/mesh-controller TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt