diff --git a/cmd/mesh-controller/away.go b/cmd/mesh-controller/away.go new file mode 100644 index 00000000..85470ca8 --- /dev/null +++ b/cmd/mesh-controller/away.go @@ -0,0 +1,99 @@ +package main + +import ( + "context" + "fmt" + "os" + "slices" + "sort" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/inventory" +) + +// A machine out of touch when a tier is sent is not sent to and not waited for (novox/hq ADR 0435 §1, issue 1161): the +// delivery goes on with the machines heard, says the others were away, and the away ones get the builds by the send +// they are given when heard again (ADR 0353, ADR 0360 §5). Away is the silent check's one verdict, machineFacts.silent +// over the watchman's presence (ADR 0433 §3), never a second rule; a machine that said it is asleep counts as away. + +// awayOf is the machines out of touch at a moment: those the silent check calls silent, and those asleep. A machine +// never heard from has not joined and is not away. +func awayOf(machines []machineFacts, now time.Time) map[string]bool { + out := map[string]bool{} + for _, m := range machines { + if m.silent(now) || m.asleep() { + out[m.name] = true + } + } + return out +} + +// judgeAway is the walk's reading of which machines are away as a tier is sent: every node with its last-heard time and +// the watchman's verdict on its presence key. A bus that cannot be asked leaves none away, which is what a walk did +// before: a machine is not passed on a reading that was not made. +var judgeAway = func(ctx context.Context, open *stores, now time.Time) (map[string]bool, error) { + nodes, err := open.inventory.Nodes(ctx) + if err != nil { + return nil, err + } + address, err := broker.BusAddress() + if err != nil { + return nil, err + } + js, err := broker.Dial(address) + if err != nil { + return nil, err + } + defer js.Close() + machines := make([]machineFacts, len(nodes)) + for i, n := range nodes { + machines[i] = machineFacts{name: n.Name, lastHeard: n.LastSeen} + } + if err := withPresence(ctx, js.Conn(), machines, now); err != nil { + return nil, err + } + return awayOf(machines, now), nil +} + +// rememberAway keeps on the plan the machines of running that are away: sorted, each once. +func rememberAway(p *inventory.Plan, running []string, away map[string]bool) { + for _, n := range running { + if away[n] && !slices.Contains(p.Away, n) { + p.Away = append(p.Away, n) + } + } + sort.Strings(p.Away) +} + +// runningHeard is the machines of running that the walk waits on for a module: those a send of the walk reached, and +// those it did not pass as away. +func runningHeard(p inventory.Plan, running []string, s *inventory.PlanModule) []string { + return slices.DeleteFunc(slices.Clone(running), func(n string) bool { + if !slices.Contains(p.Away, n) { + return false + } + _, sent := s.Rest[n] + return !sent + }) +} + +// awaySaying is what a plan says of the machines it passed as away, empty when it passed none. +func awaySaying(p inventory.Plan) string { + var said []string + for _, n := range p.Away { + said = append(said, n+" was away: sent when it is heard again") + } + return strings.Join(said, "; ") +} + +// judgedAway reads which machines are away for a tier about to be sent, and says when it cannot. +func judgedAway(ctx context.Context, open *stores, now time.Time) map[string]bool { + away, err := judgeAway(ctx, open, now) + if err != nil { + fmt.Fprintf(os.Stderr, "which machines are away cannot be read, so none is passed over: %v\n", err) + return nil // empty-on-error: none is passed over on a reading that was not made; the error is said just above + } + return away +} diff --git a/cmd/mesh-controller/away_integration_test.go b/cmd/mesh-controller/away_integration_test.go new file mode 100644 index 00000000..3cbd81ba --- /dev/null +++ b/cmd/mesh-controller/away_integration_test.go @@ -0,0 +1,226 @@ +//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" +) + +// 1161/AC5, with the store and a real bus and no ticker: a plan of two tiers whose first tier runs on a machine away when it +// is sent finishes that tier without waiting for it and without its bound, sends the second tier to the machine heard +// alone, and says the machine was away (issue 1161). The machine away keeps its want: it was sent nothing of the plan's +// builds, so the send it gets when it is heard again carries them (AC3). +func TestAPlanGoesOnWithoutAMachineAwayWhenItsTierIsSent(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + + // laptop is away as every tier is sent; anchor is heard. + wasAway := judgeAway + judgeAway = func(context.Context, *stores, time.Time) (map[string]bool, error) { + return map[string]bool{"laptop": true}, nil + } + t.Cleanup(func() { judgeAway = wasAway }) + + 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 anchor alone", func() bool { + mu.Lock() + defer mu.Unlock() + return len(sends) == 1 && strings.Join(sends[0], ",") == "anchor" + }) + reports("anchor") + // The note of the move to tier 1 is read in the step that sees it: later steps of the tier say their own. + var moved string + eventually(t, "tier 1 reached on anchor's report alone", func() bool { + if q := plan(); q.Tier == 1 { + moved = q.Note + return true + } + return false + }) + if !strings.Contains(moved, "laptop was away: sent when it is heard again") || strings.Contains(moved, "anchor") { + t.Errorf("1161/AC2: the plan says %q; want laptop named away and not anchor", moved) + } + // The ask for tier 1's build is the next step, and records its build id on the module. + eventually(t, "tier 1 asked", func() bool { m := plan().Modules["b"]; return m != nil && m.Build != "" }) + 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" + }) + // 1161/AC3: laptop was sent nothing of the plan's builds, so it still wants them for the send it gets when heard again. + was, known, err := inv.SentBuilds(ctx, "laptop") + if err != nil || !known { + t.Fatalf("1161/AC3: laptop's last send: %v, known %v", err, known) + } + current, err := inv.CurrentBuilds(ctx) + if err != nil { + t.Fatal(err) + } + if was["a"] == current["a"].Commit || was["a"] != "c1" { + t.Errorf("1161/AC3: laptop was last sent a at %q and a is now %q; want it still at c1, owed c2", was["a"], current["a"].Commit) + } + if got := plan().Away; strings.Join(got, ",") != "laptop" { + t.Errorf("1161/AC2: the plan keeps %v away; want laptop", got) + } +} diff --git a/cmd/mesh-controller/away_test.go b/cmd/mesh-controller/away_test.go new file mode 100644 index 00000000..8ca3a84a --- /dev/null +++ b/cmd/mesh-controller/away_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "slices" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/link" +) + +// A machine out of touch when a tier is sent is not sent to and not waited for (novox/hq ADR 0435 §1, issue 1161): the +// walk judges it with the silent check's one verdict, machineFacts.silent, and a machine asleep counts as away. + +// 1161/AC1: away is the silent check's verdict, and a machine asleep; AC4: a machine heard is not away. +func TestAMachineIsAwayAsTheSilentCheckJudgesIt(t *testing.T) { + now := time.Now().UTC() + machines := []machineFacts{ + {name: "heard", lastHeard: now.Add(-time.Minute)}, + {name: "quiet", lastHeard: now.Add(-silentBound - time.Minute)}, + {name: "stuck", lastHeard: now, presence: presenceSilent}, + {name: "back", lastHeard: now.Add(-time.Hour), presence: presenceHeard}, + {name: "asleep", lastHeard: now.Add(-time.Minute), power: link.PowerState{State: "sleeping"}}, + {name: "new"}, + } + got := awayOf(machines, now) + for node, want := range map[string]bool{"heard": false, "quiet": true, "stuck": true, "back": false, "asleep": true, + "new": false} { + if got[node] != want { + t.Errorf("1161/AC1: %s away is %v; want %v (away: %v)", node, got[node], want, got) + } + } +} + +// 1161/AC1, AC2: the plan keeps the machines of the tier it passed as away, sorted and once. +func TestAPlanKeepsTheMachinesItPassedAsAway(t *testing.T) { + p := &inventory.Plan{} + away := map[string]bool{"g14": true, "ace": true} + rememberAway(p, []string{"anchor", "g14"}, away) + rememberAway(p, []string{"g14", "ace"}, away) + if !slices.Equal(p.Away, []string{"ace", "g14"}) { + t.Fatalf("1161/AC2: the plan keeps %v away; want ace, g14", p.Away) + } +} + +// 1161/AC2: the plan says it, where its other passed machines are said. +func TestAPlanSaysTheMachinesItPassedAsAway(t *testing.T) { + say := awaySaying(inventory.Plan{Away: []string{"g14"}}) + if !strings.Contains(say, "g14 was away: sent when it is heard again") { + t.Fatalf("1161/AC2: the plan says %q", say) + } + if got := awaySaying(inventory.Plan{}); got != "" { + t.Fatalf("1161/AC2: a plan that passed no machine says %q", got) + } +} + +// 1161/AC1: the gate does not wait for a machine passed as away, unless a send reached it; AC4: a machine heard is waited for. +func TestAGateDoesNotWaitForAMachinePassedAsAway(t *testing.T) { + p := inventory.Plan{Away: []string{"g14"}} + none := &inventory.PlanModule{} + if got := runningHeard(p, []string{"anchor", "g14"}, none); !slices.Equal(got, []string{"anchor"}) { + t.Fatalf("1161/AC1: the gate waits on %v; want anchor alone", got) + } + sent := &inventory.PlanModule{Rest: map[string]inventory.SentDeclaration{"g14": {Digest: "d"}}} + if got := runningHeard(p, []string{"anchor", "g14"}, sent); !slices.Equal(got, []string{"anchor", "g14"}) { + t.Fatalf("1161/AC4: a machine a send reached is waited for: %v", got) + } +} diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index c61bf214..3c65d501 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -938,6 +938,15 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, return false, err } silent := silentNodes(*p, allReports, time.Now().UTC(), provLookup(ctx)) + // **A machine away when a tier is sent is not sent to and not waited for** (ADR 0435 §1, issue 1161): read once, + // for a tier that has something left to send. + var away map[string]bool + for _, m := range tier { + if s := p.Modules[m]; s != nil && s.SentAt == nil && s.FirstAt == nil && s.State != planDeleted { + away = judgedAway(ctx, open, time.Now().UTC()) + break + } + } var pending []string firstOn := map[string][]string{} runningOf := map[string][]string{} @@ -974,6 +983,16 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, } if state.FirstAt == nil { running = withoutSilent(running, silent) + passed := withoutSilent(running, away) + if len(passed) < len(running) { + for _, n := range running { + if away[n] && !slices.Contains(p.Away, n) { + fmt.Printf("%s: %s is away; %s is not sent to it and not waited for\n", p.ID, n, m) + } + } + rememberAway(p, running, away) + running = passed + } } runningOf[m] = running // **A build whose source is unchanged is never a move** (novox/hq issue 280): made from the source @@ -1189,6 +1208,8 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, if state.State == planDeleted { continue } + // A machine passed as away was sent nothing to wait on (ADR 0435 §1, issue 1161). + running = runningHeard(*p, running, state) // The plan sends what it waits for. A rebuild from the same source commit is not a // move the catalogue announces — the build machine rebuilt for a controller change // is one — so the roll-out that opens this gate is the plan's to make, once, and @@ -1292,6 +1313,10 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, if end == "" { end = leftAsWas } + // A machine passed as away is said where the others passed are said (ADR 0435 §1, issue 1161). + if said := awaySaying(*p); said != "" { + end = strings.TrimPrefix(end+"; "+said, "; ") + } p.Tier++ p.State = inventory.PlanBuilding p.Note = end diff --git a/internal/inventory/plans.go b/internal/inventory/plans.go index cecc0d70..5af4ee7e 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -33,6 +33,10 @@ type Plan struct { Tiers [][]string `json:"tiers"` Modules map[string]*PlanModule `json:"modules"` Note string `json:"note,omitempty"` + // Away is the machines the walk passed because they were out of touch as a tier was sent (novox/hq ADR 0435 §1, + // issue 1161): not sent to and not waited for. Their want of the walk's builds stays for the send they get when + // they are heard again (ADR 0360 §5). Sorted. + Away []string `json:"away,omitempty"` // TierEntered is when the plan entered the tier it is at (novox/hq to-be 45 Phase 0), read and // never written from here: a save measures the tier it leaves and stamps the next. What the // watchdog of a plan's progress (S3) reads. diff --git a/testdata/store-tests b/testdata/store-tests index edf4aeff..a16b7f30 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -166,3 +166,4 @@ ./cmd/mesh-controller TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers ./cmd/mesh-controller TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule ./cmd/mesh-controller TestTheApprovedFirstBuildPassesTheGuardOfAVerbAndIsQueued +./cmd/mesh-controller TestAPlanGoesOnWithoutAMachineAwayWhenItsTierIsSent