diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index a2a28ce1..c61bf214 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -250,6 +250,68 @@ func appliedEach(module string, since func(node string) time.Time, running []str return len(waiting) == 0, waiting } +// gateNeed is a module the next tier waits for: the machines running it, and from when each must have reported. +type gateNeed struct { + module string + running []string + since func(node string) time.Time +} + +// silentNodes is the machines a walk sent and heard nothing from past ApplySilentAfter, on whichever tier it sent +// them (novox/hq ADR 0360 §5, issue 1160): the one rule both the gates between tiers and the walk's end read, through +// walkReports. A machine that reports on its send is not silent from then on. +func silentNodes(p inventory.Plan, reports []inventory.Reported, now time.Time, prov inventory.ProvLookup) map[string]bool { + _, _, silent := walkReports(p, reports, now, prov) + out := map[string]bool{} + for _, n := range silent { + out[n] = true + } + return out +} + +// withoutSilent is the machines of running that are not silent: a later tier is not sent to a machine the walk heard +// nothing from, which keeps its want of the plan's builds for the send it gets when it is heard again (ADR 0360 §5). +func withoutSilent(running []string, silent map[string]bool) []string { + return slices.DeleteFunc(slices.Clone(running), func(n string) bool { return silent[n] }) +} + +// settleTierGate is the wait between tiers on the reports of the machines running what the next tier needs: what is +// still waited for ("module on a, b"), the silent machines left out of the wait, sorted, and whether every machine the +// gate waits on is silent, so that nothing it needs was delivered anywhere (issue 1160). +func settleTierGate(needs []gateNeed, reports []inventory.Reported, silent map[string]bool) (waiting, quiet []string, lost bool) { + all, silentAll := 0, 0 + seenQuiet := map[string]bool{} + seenNode := map[string]bool{} + for _, need := range needs { + for _, n := range need.running { + if !seenNode[n] { + seenNode[n] = true + all++ + if silent[n] { + silentAll++ + } + } + } + _, on := appliedEach(need.module, need.since, need.running, reports) + var still []string + for _, n := range on { + if silent[n] { + if !seenQuiet[n] { + seenQuiet[n] = true + quiet = append(quiet, n) + } + continue + } + still = append(still, n) + } + if len(still) > 0 { + waiting = append(waiting, fmt.Sprintf("%s on %s", need.module, strings.Join(still, ", "))) + } + } + sort.Strings(quiet) + return waiting, quiet, all > 0 && silentAll == all +} + func applied(module string, builtAt time.Time, running []string, reports []inventory.Reported) (bool, []string) { at := map[string]*time.Time{} for _, r := range reports { @@ -869,6 +931,13 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, // being behind. Now every module of the tier whose first machine is the same is sent there in one // send, judged there by one gate — kept on the first of them, the others gated by it — and the rest // of the tier's machines are sent once, together, when the gates have passed. + // **A machine silent past the bound is left as it was, on every tier** (novox/hq ADR 0360 §5, issue 1160): the + // walk's end names it, and the gates between tiers and the sends of the tiers after do not wait for it or go to it. + allReports, err := inv.LastReports(ctx) + if err != nil { + return false, err + } + silent := silentNodes(*p, allReports, time.Now().UTC(), provLookup(ctx)) var pending []string firstOn := map[string][]string{} runningOf := map[string][]string{} @@ -903,6 +972,9 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, if goes != nil { running = slices.DeleteFunc(running, func(n string) bool { return !goes[n] }) } + if state.FirstAt == nil { + running = withoutSilent(running, silent) + } runningOf[m] = running // **A build whose source is unchanged is never a move** (novox/hq issue 280): made from the source // of the build every machine running it was sent, it is registered with that build's artifacts — @@ -1096,13 +1168,14 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, return changed, nil } // And wait for what the next tier needs running. + leftAsWas := "" needed := gates(*p, edges, rollsOut) if len(needed) > 0 { reports, err := inv.LastReports(ctx) if err != nil { return false, err } - var waiting []string + var needs []gateNeed for _, m := range needed { running, err := inv.Running(ctx, m) if err != nil { @@ -1133,14 +1206,12 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, for _, r := range reports { byNode[r.Node] = r } - if ok, on := appliedEach(m, func(node string) time.Time { + needs = append(needs, gateNeed{m, running, func(node string) time.Time { if s := byNode[node].Sent; s != nil { return *s } return time.Time{} - }, running, reports); !ok { - waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", "))) - } + }}) continue } // **Each machine from its own send** (novox/hq issue 256). With one machine first (ADR 0218) @@ -1160,9 +1231,23 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, } return at } - if ok, on := appliedEach(m, sinceFor, running, reports); !ok { - waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", "))) - } + needs = append(needs, gateNeed{m, running, sinceFor}) + } + waiting, quiet, lost := settleTierGate(needs, reports, silent) + if lost { + // Every machine the next tier waits on is silent: nothing it needs was delivered anywhere, so the plan does + // not go on as if it had been (issue 1160). + p.State = inventory.PlanFailed + p.Note = fmt.Sprintf("tier %d built; every machine it waits on said nothing on its send within %s: %s", + p.Tier, applySilentAfter, strings.Join(quiet, ", ")) + fmt.Printf("%s: %s\n", p.ID, p.Note) + return true, nil + } + if len(quiet) > 0 { + fmt.Printf("%s: tier %d: %s said nothing on its send within %s and is left as it was; the plan goes on without it\n", + p.ID, p.Tier, strings.Join(quiet, ", "), applySilentAfter) + leftAsWas = fmt.Sprintf("%s said nothing on its send within %s and is left as it was", strings.Join(quiet, ", "), + applySilentAfter) } if len(waiting) > 0 { note := "tier " + fmt.Sprint(p.Tier) + " built; waiting for " + strings.Join(waiting, "; ") + " to be applied" @@ -1204,6 +1289,9 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, carryUserList(ctx, open, p) } } + if end == "" { + end = leftAsWas + } p.Tier++ p.State = inventory.PlanBuilding p.Note = end diff --git a/cmd/mesh-controller/silent_mid_tier_integration_test.go b/cmd/mesh-controller/silent_mid_tier_integration_test.go new file mode 100644 index 00000000..645c908b --- /dev/null +++ b/cmd/mesh-controller/silent_mid_tier_integration_test.go @@ -0,0 +1,202 @@ +//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" +) + +// 1160/AC5, live, with the store and a real bus and no ticker: a plan of two tiers whose first tier has a machine that +// never reports, and whose second tier waits on the first (the plan's own order), passes its first tier after the bound, +// the silent key's expiry the only trigger, and sends its second tier to the machine that reported alone (issue 1160). +func TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers(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() }) + + wasSaved := inventory.PlanSaved + inventory.PlanSaved = servingPlanSaved(ctx, checkEvents, open) + 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) + } + } + plan := func() inventory.Plan { + t.Helper() + p, err := inv.PlanByID(ctx, "plan-1160") + if err != nil { + t.Fatal(err) + } + return p + } + + built := time.Now().UTC() + p := inventory.Plan{ID: "plan-1160", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", Created: time.Now().UTC(), + State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}}, + Delivery: &inventory.PlanDelivery{Order: []inventory.Edge{{From: "b", To: "a", Kind: edgeChangeOrder}}}, + Modules: map[string]*inventory.PlanModule{ + "a": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-a-2"}, + "b": {}, + }} + first, stopFirst := context.WithCancel(ctx) + defer stopFirst() + startRolloutEvents(first, open, js) + if err := inv.SavePlan(ctx, &p); err != nil { + t.Fatal(err) + } + eventually(t, "tier 0 sent to both machines", func() bool { + mu.Lock() + defer mu.Unlock() + return len(sends) == 1 + }) + reports("anchor") + time.Sleep(time.Second) + if got := plan().Tier; got != 0 { + t.Fatalf("1160/AC5: the plan went on to tier %d before the bound, with laptop not heard from", got) + } + eventually(t, "tier 1 asked once laptop's key expired", func() bool { return plan().Tier == 1 }) + if note := plan().Note; !strings.Contains(note, "laptop") || strings.Contains(note, "anchor") { + t.Errorf("1160/AC5: the plan says %q; want laptop named silent and not anchor", note) + } + b := plan().Modules["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 anchor alone", func() bool { + mu.Lock() + defer mu.Unlock() + return len(sends) == 2 && strings.Join(sends[1], ",") == "anchor" + }) +} diff --git a/cmd/mesh-controller/silent_mid_tier_test.go b/cmd/mesh-controller/silent_mid_tier_test.go new file mode 100644 index 00000000..f0d64c99 --- /dev/null +++ b/cmd/mesh-controller/silent_mid_tier_test.go @@ -0,0 +1,85 @@ +package main + +import ( + "slices" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// A machine silent past the bound at a tier between the first and the last is named silent and left as it was, as the +// last tier's is (novox/hq ADR 0282 §1, ADR 0360 §4-5, issue 1160): one rule, walkReports, read by the walk's end and by +// the gates between tiers. + +func silentWalk(now time.Time) (inventory.Plan, []inventory.Reported) { + sent := now.Add(-time.Minute) + long := now.Add(-inventory.ApplySilentAfter - time.Minute) + at := now.Add(-10 * time.Second) + p := inventory.Plan{Tier: 0, Tiers: [][]string{{"engine"}, {"app"}}, Modules: map[string]*inventory.PlanModule{ + "engine": {Rest: map[string]inventory.SentDeclaration{ + "anchor": {Digest: "d-anchor", At: sent}, + "g14": {Digest: "d-g14", At: long}, + "fresh": {Digest: "d-fresh", At: sent}}}, + "app": {}}} + reports := []inventory.Reported{{Node: "anchor", Outcome: inventory.OutcomeApplied, At: &at, Declared: "d-anchor"}} + return p, reports +} + +func needsOf(running ...string) []gateNeed { + return []gateNeed{{module: "engine", running: running, since: func(string) time.Time { return time.Time{} }}} +} + +// 1160/AC1: at a middle tier the machine silent past the bound is named, not waited for, and the plan goes on with the +// machines that reported; one still inside the bound is waited for. +func TestAMiddleTierGateNamesTheSilentMachineAndGoesOnWithThoseThatReported(t *testing.T) { + now := time.Now().UTC() + p, reports := silentWalk(now) + silent := silentNodes(p, reports, now, nil) + waiting, quiet, lost := settleTierGate(needsOf("anchor", "g14"), reports, silent) + if len(waiting) != 0 || !slices.Equal(quiet, []string{"g14"}) || lost { + t.Fatalf("1160/AC1: waiting %v, silent %v, lost %v; want g14 named silent and nothing waited for", waiting, quiet, lost) + } + waiting, quiet, _ = settleTierGate(needsOf("anchor", "g14", "fresh"), reports, silent) + if len(waiting) != 1 || waiting[0] != "engine on fresh" || !slices.Equal(quiet, []string{"g14"}) { + t.Fatalf("1160/AC1: a machine inside the bound is still waited for: waiting %v, silent %v", waiting, quiet) + } +} + +// 1160/AC2: the walk's end and the gate read the same silent machines. +func TestTheGateAndTheWalksEndShareOneRuleOfSilence(t *testing.T) { + now := time.Now().UTC() + p, reports := silentWalk(now) + _, _, ended := walkReports(p, reports, now, nil) + got := silentNodes(p, reports, now, nil) + if len(got) != len(ended) || !got["g14"] || !slices.Equal(ended, []string{"g14"}) { + t.Fatalf("1160/AC2: the gate reads %v silent, the walk's end %v", got, ended) + } +} + +// 1160/AC3: a later tier is not sent to a machine silent. +func TestALaterTierIsNotSentToASilentMachine(t *testing.T) { + got := withoutSilent([]string{"anchor", "g14"}, map[string]bool{"g14": true}) + if !slices.Equal(got, []string{"anchor"}) { + t.Fatalf("1160/AC3: a later tier goes to %v; want anchor alone", got) + } + in := []string{"anchor", "g14"} + withoutSilent(in, map[string]bool{"g14": true}) + if !slices.Equal(in, []string{"anchor", "g14"}) { + t.Fatalf("1160/AC3: the machines running it were changed: %v", in) + } +} + +// 1160/AC4: every machine the gate waits on silent: the plan is lost, not gone on as if delivered. +func TestAGateWhoseEveryMachineIsSilentIsLost(t *testing.T) { + now := time.Now().UTC() + p, reports := silentWalk(now) + reports = nil + p.Modules["engine"].Rest = map[string]inventory.SentDeclaration{ + "g14": {Digest: "d", At: now.Add(-time.Hour)}, "laptop": {Digest: "d", At: now.Add(-time.Hour)}} + silent := silentNodes(p, reports, now, nil) + _, quiet, lost := settleTierGate(needsOf("g14", "laptop"), reports, silent) + if !lost || !slices.Equal(quiet, []string{"g14", "laptop"}) { + t.Fatalf("1160/AC4: lost %v, silent %v; want both named and the gate lost", lost, quiet) + } +} diff --git a/testdata/store-tests b/testdata/store-tests index 4bf968a0..970640e6 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -163,3 +163,4 @@ ./internal/inventory TestASendEmptiesTheColumnSoALaterCopyLeavesAClearedKeyBare ./internal/inventory TestWhetherAMachineReadsModulesIsItsLatestWord ./cmd/mesh-controller TestAComposedDeclarationCarriesModulesAndBuildsOnlyForANodeThatReadsThem +./cmd/mesh-controller TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers