//go:build integration 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 kept, said and looked at by the serving controller's own hook. saved := &movesSaid{revs: map[string][]int64{}} wasSaved := inventory.PlanSaved serving := servingPlanSaved(ctx, checkEvents, open) inventory.PlanSaved = func(i *inventory.Inventory, p inventory.Plan) { saved.hear(i, p); serving(i, 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() 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"}}, 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() startRolloutEvents(second, open, js) // 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) } // Acceptance of the verb path (issue 974, review 1), live, with the store and a bus: a walk the delivery's owner lets // go by the verb `deliver` after the operator's approval is asked its first tier with no other event. The verb runs as a // child process, whose saves the serving controller does not hear; the handler asks the look as the verb ends. func TestAWalkLetGoByTheDeliverVerbIsAskedWithNoOtherEvent(t *testing.T) { open := aMesh(t) ctx := t.Context() 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 }) asked := asksRecorded(t) register(t, open, catalogue.Manifest{Module: "app", Version: "1"}) if _, err := assign(ctx, open, "anchor", "app"); err != nil { t.Fatal(err) } cut := time.Now().UTC() waiting := inventory.Plan{ID: "plan-974-deliver", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c8c8c8c8", Created: cut, State: inventory.PlanBuilding, Tiers: [][]string{{"app"}}, Modules: map[string]*inventory.PlanModule{"app": {}}, Delivery: &inventory.PlanDelivery{Awaits: catalogue.DeliverySeat}, Times: &inventory.PlanTimes{Cut: &cut, Class: inventory.ClassLeaf}} if err := open.inventory.SavePlan(ctx, &waiting); err != nil { t.Fatal(err) } // The serving controller, as serve starts it. wasSaved := inventory.PlanSaved inventory.PlanSaved = servingPlanSaved(ctx, checkEvents, open) t.Cleanup(func() { inventory.PlanSaved = wasSaved }) started, stop := context.WithCancel(ctx) defer stop() 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. wasRun := runTheVerb runTheVerb = func(ctx context.Context, argv []string) (verbAnswer, error) { was := inventory.PlanSaved inventory.PlanSaved = func(*inventory.Inventory, inventory.Plan) {} defer func() { inventory.PlanSaved = was }() return verbAnswer{}, deliveryCommand(ctx, []string{"go", waiting.ID, "--why", "its turn"}) } t.Cleanup(func() { runTheVerb = wasRun }) handlers, _, err := seatToolHandlers() if err != nil { t.Fatal(err) } if _, err := handlers["deliver"](ctx, []byte(`{"plan":"`+waiting.ID+`","why":"its turn"}`)); err != nil { t.Fatal(err) } eventually(t, "the walk's first tier asked, with no event but the verb's end", func() bool { p, err := open.inventory.PlanByID(ctx, waiting.ID) return err == nil && p.Modules["app"] != nil && p.Modules["app"].Build != "" && len(*asked) == 1 }) }