diff --git a/cmd/mesh-controller/build.go b/cmd/mesh-controller/build.go index 5a6802e1..b89e30b2 100644 --- a/cmd/mesh-controller/build.go +++ b/cmd/mesh-controller/build.go @@ -152,6 +152,8 @@ func buildFrom(result link.BuildResult) inventory.Build { Path: result.Path, // What it was made from (novox/hq issue 280): two builds with one are one build. SourceFingerprint: result.SourceFingerprint, + // And its reuse key (novox/hq ADR 0350 §1), which the chain names (ADR 0408 §1). + ReuseKey: result.ReuseKey, } // When it was asked, which is what orders it against another build of the same module // (novox/hq 04-ISSUES/219) — not when it was heard. @@ -254,6 +256,18 @@ func buildsCommand(ctx context.Context, args []string) error { for _, made := range b.Made { fmt.Printf(" %-10s %s\n", made.Kind, made.Reference) } + // What it published and under what reuse key (novox/hq ADR 0408 §1). + if p := b.Published; p != nil { + if p.Artifact != "" { + fmt.Printf(" published %s\n", p.Artifact) + } + if p.Package != nil { + fmt.Printf(" published %s@%s\n", p.Package.Name, p.Package.Version) + } + } + if b.ReuseKey != "" { + fmt.Printf(" reuse key %s\n", b.ReuseKey) + } if !b.Worked() { // The builder's own first line. The whole failure is often a build log, and printing // it here would bury every other row. diff --git a/cmd/mesh-controller/chain_integration_test.go b/cmd/mesh-controller/chain_integration_test.go new file mode 100644 index 00000000..1f5e957d --- /dev/null +++ b/cmd/mesh-controller/chain_integration_test.go @@ -0,0 +1,189 @@ +//go:build integration + +package main + +import ( + "context" + "fmt" + "slices" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/chain" + "github.com/novox/mesh-controller/internal/inventory" + "github.com/novox/mesh-controller/internal/lease" + "github.com/novox/mesh-controller/internal/link" +) + +// **Acceptance 10 of design 47: one read of an assignment names the commit, the build and the rollout that put it there, +// and one read of the rollout lists the assignments it changed** (novox/hq ADR 0408, issue 1095). It enters by the live +// trigger — a merge heard, its batch cut into a plan when the window closed, its build answered through the build seat's +// path, the plan advanced — and writes no key and no row by hand. Only the machine's end of the send is the test's: the send +// is recorded as the controller records one it made, and the gate's facts are the store's. +func TestOneReadOfAnAssignmentNamesItsCommitBuildAndRolloutAndOneReadOfTheRolloutListsItsAssignments(t *testing.T) { + open := windowed(t) + everyHeadWaits(t) + asked := asksRecorded(t) + ctx := t.Context() + inv := open.inventory + if _, err := assign(ctx, open, "anchor", "notes"); err != nil { + t.Fatal(err) + } + + var sent [][]string + wasSend := sendRollout + sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) { + sent = append(sent, append([]string(nil), names...)) + current, err := open.inventory.CurrentBuilds(ctx) + if err != nil { + return nil, err + } + for _, node := range names { + digest := fmt.Sprintf("d-%s-%d", node, len(sent)) + if err := open.inventory.RecordSent(ctx, nodeID(t, open, node), digest, + map[string]string{"notes": current["notes"].Commit}); err != nil { + return nil, err + } + } + return names, nil + } + t.Cleanup(func() { sendRollout = wasSend }) + wasGather := gatherGateFacts + gatherGateFacts = func(ctx context.Context, open *stores, component string) (gateFacts, error) { + f := gateFacts{now: time.Now(), reports: map[string]inventory.Reported{}, engines: map[string]string{}, + rolledBack: map[string][]lease.Rollback{}, served: map[string]served{}} + reports, err := open.inventory.LastReports(ctx) + if err != nil { + return f, err + } + for _, r := range reports { + f.reports[r.Node] = r + } + return f, nil + } + t.Cleanup(func() { gatherGateFacts = wasGather }) + + // A merge of notes is heard, and its window closes: the batch is cut into the walk. + hear(t, open, catalogueMerge("c-notes", "notes", t0), t0.Add(time.Second)) + cutAt(t, open, t0.Add(time.Second+90*time.Second)) + ws, _ := walks(t, open) + if len(ws) != 1 { + t.Fatalf("%d walk(s) after the cut", len(ws)) + } + rollout := ws[0].ID + + // The walk asks its first tier (at the cut, or at its next step); the build seat answers, and the controller takes the + // outcome in as it does. + asking, err := inv.PlanByID(ctx, rollout) + if err != nil { + t.Fatal(err) + } + for try := 0; try < 3 && (asking.Modules["notes"] == nil || asking.Modules["notes"].Build == ""); try++ { + advancePlans(ctx, open) + if asking, err = inv.PlanByID(ctx, rollout); err != nil { + t.Fatal(err) + } + } + buildID := asking.Modules["notes"].Build + if buildID == "" || len(*asked) != 1 || !strings.HasSuffix((*asked)[0], "#"+buildID) { + t.Fatalf("the walk asked %v, and notes' build is %q", *asked, buildID) + } + digest := "sha256:" + strings.Repeat("a", 64) + result := outcome(buildID, "modules/notes", "c-notes", "notes", "") + result.ReuseKey = "rk-notes" + result.Made = []link.MadeArtifact{{Name: "notes", Kind: "image", Reference: "artifact-store://notes/notes@" + digest}} + if err := (builds{inv: inv, open: open}).Built(ctx, result); err != nil { + t.Fatal(err) + } + + // The walk sends the tier to its first machine. + for try := 0; try < 3 && len(sent) == 0; try++ { + advancePlans(ctx, open) + } + if !slices.Equal(slices.Concat(sent...), []string{"anchor"}) { + t.Fatalf("sent %v", sent) + } + + // One read of the assignment. + on, err := inv.AssignmentsOn(ctx, "anchor") + if err != nil || len(on) != 1 { + t.Fatalf("the assignments of anchor: %+v %v", on, err) + } + a := on[0] + if a.Module != "notes" || a.Commit != "c-notes" || a.Build != buildID || a.Rollout != rollout || a.Artifact != digest || + a.Since.IsZero() || a.Assigned.IsZero() || !a.Stands() { + t.Fatalf("the assignment names %+v, not commit c-notes, build %s, artifact %s, rollout %s", a, buildID, digest, rollout) + } + // One read of the rollout. + p, err := inv.PlanByID(ctx, rollout) + if err != nil { + t.Fatal(err) + } + wantChange := chain.Change{Node: "anchor", Module: "notes", After: chain.Ref{Build: buildID, Commit: "c-notes"}} + if !slices.Contains(p.Changed, wantChange) || !slices.Contains(p.Builds, buildID) || len(p.Steps) != 1 || + p.Steps[0].Node != "anchor" || p.Steps[0].Sent == nil { + t.Fatalf("the rollout lists changed %+v, builds %v, steps %+v; want %+v", p.Changed, p.Builds, p.Steps, wantChange) + } + // And one read of the build names the same commit, with what it published and its reuse key. + built, found, err := inv.BuildByID(ctx, buildID) + if err != nil || !found || built.Commit != "c-notes" || built.Module != "notes" { + t.Fatalf("the build: %+v %v %v", built, found, err) + } + lately, err := inv.Builds(ctx, "notes", 1) + if err != nil || len(lately) != 1 || lately[0].ReuseKey != "rk-notes" || lately[0].Published == nil || + lately[0].Published.Artifact != digest { + t.Fatalf("what the build published: %+v %v", lately, err) + } + // The verbs say the links. + if lines := changedLines(p); len(lines) != 2 || !strings.Contains(lines[1], "anchor/notes") || + !strings.Contains(lines[1], buildID) { + t.Fatalf("`plans` says %v", lines) + } + if lines := runLines(on); len(lines) != 2 || !strings.Contains(lines[1], buildID) || !strings.Contains(lines[1], rollout) { + t.Fatalf("`node show` says %v", lines) + } +} + +// A machine sent again what it runs says nothing new, and a push by hand with no rollout fills what is missing and blanks +// nothing (issue 1095). +func TestASendOfWhatAMachineRunsAlreadyChangesNoLinkOfItsAssignment(t *testing.T) { + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + register(t, open, catalogue.Manifest{Module: "notes", Version: "1"}) + if _, err := assign(ctx, open, "anchor", "notes"); err != nil { + t.Fatal(err) + } + if err := inv.RecordBuild(ctx, inventory.Build{ID: "build-1", Module: "notes", Commit: "c1", Repository: "novox/notes"}); err != nil { + t.Fatal(err) + } + node := nodeID(t, open, "anchor") + inRollout := inventory.WithRollout(ctx, "plan-1", map[string]inventory.RolloutModule{"notes": {Commit: "c1", Build: "build-1"}}) + if err := inv.RecordSent(inRollout, node, "d1", map[string]string{"notes": "c1"}); err != nil { + t.Fatal(err) + } + // Sent again by hand, with no rollout: the assignment still names the rollout that put the build there. + if err := inv.RecordSent(ctx, node, "d2", map[string]string{"notes": "c1"}); err != nil { + t.Fatal(err) + } + on, _ := inv.AssignmentsOn(ctx, "anchor") + if len(on) != 1 || on[0].Rollout != "plan-1" || on[0].Build != "build-1" || on[0].Commit != "c1" { + t.Fatalf("the assignment names %+v", on) + } + // A declaration sent by hand with no builds known says nothing of them. + if err := inv.RecordSent(ctx, node, "d3", nil); err != nil { + t.Fatal(err) + } + if on, _ = inv.AssignmentsOn(ctx, "anchor"); len(on) != 1 || on[0].Commit != "c1" { + t.Fatalf("a send of unknown builds changed %+v", on) + } + // A module sent a commit no build is kept of is named by its commit alone. + if err := inv.RecordSent(ctx, node, "d4", map[string]string{"notes": "c2"}); err != nil { + t.Fatal(err) + } + if on, _ = inv.AssignmentsOn(ctx, "anchor"); len(on) != 1 || on[0].Commit != "c2" || on[0].Build != "" || on[0].Rollout != "" { + t.Fatalf("the assignment after a push of c2 by hand names %+v", on) + } +} diff --git a/cmd/mesh-controller/chain_links.go b/cmd/mesh-controller/chain_links.go new file mode 100644 index 00000000..b6d01adb --- /dev/null +++ b/cmd/mesh-controller/chain_links.go @@ -0,0 +1,121 @@ +package main + +import ( + "context" + "fmt" + "slices" + "time" + + "github.com/novox/mesh-controller/internal/chain" + "github.com/novox/mesh-controller/internal/inventory" +) + +// The links a rollout makes (novox/hq ADR 0408 §1, design 47 acceptance 10, issue 1095): the send of a plan's tier puts a +// build on a machine; the assignment of the module there names the build, its commit and the rollout (written by the record of +// the send, inventory.RecordSent, from the context a send is made in), and the rollout lists the assignments it changed with the +// build before and after. + +// rolloutModules is what a plan sends of each module: the commit and the build of it. +func rolloutModules(p *inventory.Plan) map[string]inventory.RolloutModule { + out := make(map[string]inventory.RolloutModule, len(p.Modules)) + for m, s := range p.Modules { + if s != nil && s.Commit != "" { + out[m] = inventory.RolloutModule{Commit: s.Commit, Build: s.Build} + } + } + return out +} + +// sendingAs is the context a plan's send is made in: what is recorded of the send names this rollout. +func sendingAs(ctx context.Context, p *inventory.Plan) context.Context { + return inventory.WithRollout(ctx, p.ID, rolloutModules(p)) +} + +// ranBefore is what each module on a machine ran before a send, as the chain names it: its assignment's build and commit, +// else the commit the machine was last sent for it. Read before the send, which writes the new ones. +func ranBefore(ctx context.Context, inv *inventory.Inventory, node string) map[string]chain.Ref { + was := map[string]chain.Ref{} + if sent, known, err := inv.SentBuilds(ctx, node); err == nil && known { + for m, commit := range sent { + was[m] = chain.Ref{Commit: commit} + } + } + if ran, err := inv.AssignmentRuns(ctx, node); err == nil { + for m, r := range ran { + was[m] = r + } + } + return was +} + +// noteChanges writes into a plan that a send to a machine changed the assignments of the modules it carried: the build and +// commit before and after, the builds the rollout carried, and the step taken. A change noted already for the machine and the +// module is replaced, so a send made again says it once. +func noteChanges(p *inventory.Plan, node string, modules []string, was map[string]chain.Ref, at time.Time) { + for _, m := range modules { + s := p.Modules[m] + if s == nil { + continue + } + change := chain.Change{Node: node, Module: m, Before: was[m], After: chain.Ref{Build: s.Build, Commit: s.Commit}} + if at := slices.IndexFunc(p.Changed, func(c chain.Change) bool { return c.Node == node && c.Module == m }); at >= 0 { + change.Before = p.Changed[at].Before + p.Changed[at] = change + } else { + p.Changed = append(p.Changed, change) + } + if s.Build != "" && !slices.Contains(p.Builds, s.Build) { + p.Builds = append(p.Builds, s.Build) + } + } + if !slices.ContainsFunc(p.Steps, func(s chain.Step) bool { return s.Tier == p.Tier && s.Node == node }) { + sent := at.UTC() + p.Steps = append(p.Steps, chain.Step{Tier: p.Tier, Node: node, Sent: &sent}) + } +} + +// refSaid is a build and the commit it was made from as a person reads them; "nothing" for a module that ran none. +func refSaid(r chain.Ref) string { + switch { + case r.Build == "" && r.Commit == "": + return "nothing" + case r.Build == "": + return "commit " + short(r.Commit) + } + return r.Build + " (" + short(r.Commit) + ")" +} + +// changedLines say the assignments a rollout changed: the machine, the module, and the build before and after. +func changedLines(p inventory.Plan) []string { + if len(p.Changed) == 0 { + return nil + } + out := []string{" changed:"} + for _, c := range p.Changed { + out = append(out, fmt.Sprintf(" %s/%s %s → %s", c.Node, c.Module, refSaid(c.Before), refSaid(c.After))) + } + return out +} + +// runLines say what each module assigned to a machine runs there, as its assignment names it: the build, the commit, the +// rollout that put it there and since when. +func runLines(assigned []chain.Assignment) []string { + var out []string + for _, a := range assigned { + if a.Build == "" && a.Commit == "" { + continue + } + line := fmt.Sprintf(" %-22s %s", a.Module, refSaid(chain.Ref{Build: a.Build, Commit: a.Commit})) + if a.Rollout != "" { + line += ", sent by " + a.Rollout + } + if !a.Since.IsZero() { + line += " at " + a.Since.Local().Format("2006-01-02 15:04") + } + out = append(out, line) + } + if len(out) == 0 { + return nil + } + return append([]string{" what they run, as the chain names it:"}, out...) +} diff --git a/cmd/mesh-controller/chain_links_test.go b/cmd/mesh-controller/chain_links_test.go new file mode 100644 index 00000000..8652bd60 --- /dev/null +++ b/cmd/mesh-controller/chain_links_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/chain" + "github.com/novox/mesh-controller/internal/inventory" +) + +// **A rollout lists the assignments it changed, each with the build before and after** (ADR 0408 §1, issue 1095): one +// change for a machine and a module however often the send is noted, the build before kept from the first, the builds the +// rollout carried once each, and one step for a machine and a tier. +func TestARolloutListsTheAssignmentsItChangedWithTheBuildBeforeAndAfter(t *testing.T) { + at := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + p := inventory.Plan{ID: "plan-1", Tier: 1, Modules: map[string]*inventory.PlanModule{ + "notes": {Commit: "c2", Build: "b2"}, "wiki": {Commit: "c2w", Build: "b2w"}, "nothing": nil}} + was := map[string]chain.Ref{"notes": {Build: "b1", Commit: "c1"}, "wiki": {Commit: "c1w"}} + + noteChanges(&p, "anchor", []string{"notes", "wiki", "nothing"}, was, at) + noteChanges(&p, "laptop", []string{"notes"}, nil, at) + // The send noted again, after what was sent had become what the machine runs: the first before stands. + noteChanges(&p, "anchor", []string{"notes"}, map[string]chain.Ref{"notes": {Build: "b2", Commit: "c2"}}, at.Add(time.Minute)) + + want := []chain.Change{ + {Node: "anchor", Module: "notes", Before: chain.Ref{Build: "b1", Commit: "c1"}, After: chain.Ref{Build: "b2", Commit: "c2"}}, + {Node: "anchor", Module: "wiki", Before: chain.Ref{Commit: "c1w"}, After: chain.Ref{Build: "b2w", Commit: "c2w"}}, + {Node: "laptop", Module: "notes", After: chain.Ref{Build: "b2", Commit: "c2"}}, + } + if len(p.Changed) != len(want) { + t.Fatalf("changed %+v", p.Changed) + } + for k := range want { + if p.Changed[k] != want[k] { + t.Errorf("change %d is %+v, want %+v", k, p.Changed[k], want[k]) + } + } + if len(p.Builds) != 2 || p.Builds[0] != "b2" || p.Builds[1] != "b2w" { + t.Errorf("builds %v", p.Builds) + } + if len(p.Steps) != 2 || p.Steps[0].Node != "anchor" || p.Steps[0].Tier != 1 || p.Steps[0].Sent == nil || p.Steps[1].Node != "laptop" { + t.Errorf("steps %+v", p.Steps) + } + lines := changedLines(p) + if len(lines) != 4 || !strings.Contains(lines[1], "anchor/notes b1 (c1) → b2 (c2)") || + !strings.Contains(lines[3], "laptop/notes nothing → b2 (c2)") { + t.Errorf("changed lines %q", lines) + } + if got := changedLines(inventory.Plan{}); got != nil { + t.Errorf("a plan that changed nothing says %q", got) + } +} + +func TestWhatAMachineRunsIsSaidAsItsAssignmentNamesIt(t *testing.T) { + at := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + lines := runLines([]chain.Assignment{ + {Node: "anchor", Module: "notes", Build: "b2", Commit: "c2c2c2c2c2c2", Rollout: "plan-1", Since: at}, + {Node: "anchor", Module: "web", Commit: "c3"}, + {Node: "anchor", Module: "never-sent"}, + }) + if len(lines) != 3 || !strings.Contains(lines[1], "notes") || !strings.Contains(lines[1], "b2 (c2c2c2c2)") || + !strings.Contains(lines[1], "sent by plan-1") || !strings.Contains(lines[2], "web") || !strings.Contains(lines[2], "commit c3") { + t.Fatalf("lines %q", lines) + } + if runLines([]chain.Assignment{{Node: "a", Module: "m"}}) != nil { + t.Fatal("a module never sent is said to run something") + } +} diff --git a/cmd/mesh-controller/nodes.go b/cmd/mesh-controller/nodes.go index c446da3d..a297e7b3 100644 --- a/cmd/mesh-controller/nodes.go +++ b/cmd/mesh-controller/nodes.go @@ -929,6 +929,12 @@ func showNode(ctx context.Context, stored *stores, name string) error { } if len(assigned) > 0 { fmt.Printf("\n assigned: %s\n", strings.Join(assigned, ", ")) + // And what each runs, as its assignment names it (novox/hq ADR 0408, issue 1095). + if on, err := inv.AssignmentsOn(ctx, name); err == nil { + for _, line := range runLines(on) { + fmt.Println(line) + } + } } // And which of them a push leaves out, and why (novox/hq issue 380): judged as the push judges it, the same // reason its condition says. Not computable is said, never read as "all applied". diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index 5b5d1590..a2a28ce1 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -16,6 +16,7 @@ import ( "github.com/novox/mesh-controller/internal/builder" "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/chain" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) @@ -1054,7 +1055,11 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, } } sort.Strings(machines) - sent, err := sendRollout(withScope(ctx, scope), open, machines) + was := map[string]map[string]chain.Ref{} + for _, n := range machines { + was[n] = ranBefore(ctx, inv, n) + } + sent, err := sendRollout(withScope(sendingAs(ctx, p), scope), open, machines) if err != nil { return false, fmt.Errorf("sending %s to %s after tier %d: %w", strings.Join(rest, ", "), strings.Join(machines, ", "), p.Tier, err) @@ -1076,6 +1081,7 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, } } noteSends(ctx, p.Tier, []string{n}, carried, now) + noteChanges(p, n, carried, was[n], now) } fmt.Printf("%s: tier %d built; sent %s to %s, one send each\n", p.ID, p.Tier, strings.Join(rest, ", "), strings.Join(sent, ", ")) @@ -1324,6 +1330,10 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string s := p.Modules[m] owns = append(owns, inventory.CarriedMove{Module: m, Node: node, From: s.Previous, To: s.Commit, Build: s.Build}) } + // What the assignments name before the send, for the rollout to say what it changed; and the send is made as this + // rollout, so what is recorded of it names it (novox/hq ADR 0408, issue 1095). + was := ranBefore(ctx, inv, node) + ctx = sendingAs(ctx, p) // A gated send (ADR 0236): everything waiting on the machine goes with the tier, and the gate judges // all of it there. carried, sent, grants, err := gatedSend(ctx, open, node, owns) @@ -1350,6 +1360,9 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string } } noteSends(ctx, p.Tier, sent, modules, now) + for _, n := range sent { + noteChanges(p, n, modules, was, now) + } p.State = inventory.PlanRolling p.Note = fmt.Sprintf("tier %d built; sent %s to %s first, in one send", p.Tier, strings.Join(modules, ", "), strings.Join(sent, ", ")) @@ -2016,6 +2029,10 @@ func plansCommand(ctx context.Context, args []string) error { } } } + // The assignments it changed, each with the build and commit before and after (novox/hq ADR 0408, issue 1095). + for _, line := range changedLines(p) { + fmt.Println(line) + } return nil } if *whatIf != "" { diff --git a/internal/inventory/chain_builds.go b/internal/inventory/chain_builds.go index 6a965d44..ececd272 100644 --- a/internal/inventory/chain_builds.go +++ b/internal/inventory/chain_builds.go @@ -185,7 +185,7 @@ func (i *Inventory) chainBuildList(ctx context.Context, c *chain.Chain, module s for _, b := range all { if module == "" || b.Module == module { out = append(out, Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, - On: b.On, Failed: b.Failed, Made: b.Made, Asked: b.Asked, At: b.At}) + On: b.On, Failed: b.Failed, Made: b.Made, Asked: b.Asked, At: b.At, ReuseKey: b.ReuseKey, Published: b.Published}) } } sort.SliceStable(out, func(a, b int) bool { diff --git a/internal/inventory/chain_links.go b/internal/inventory/chain_links.go new file mode 100644 index 00000000..c671c93f --- /dev/null +++ b/internal/inventory/chain_links.go @@ -0,0 +1,190 @@ +package inventory + +import ( + "context" + "errors" + "fmt" + "os" + "sort" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/novox/mesh-controller/internal/chain" +) + +// The links of the chain (novox/hq ADR 0408 §1, design 47 acceptance 10, issue 1095): a send that puts a build on a machine +// writes, in the assignment of the module there, the build, what it published, its commit and the rollout that sent it, and +// since when; the rollout lists the assignments it changed with the build before and after (the controller's send, in +// release_plan.go). One read of an assignment then names the commit, the build and the rollout that put it there, and one read +// of the rollout lists its assignments. + +type rolloutKey struct{} + +// RolloutModule is what a rollout sends of one module: the commit and the build of it. +type RolloutModule struct{ Commit, Build string } + +type rolloutOf struct { + id string + modules map[string]RolloutModule +} + +// WithRollout is a context for a send a rollout makes: what RecordSent writes in the assignments of the modules it carries +// names this rollout and the build it sent. A module the send carries beside these, that another walk is moving, is named +// by the build the mesh holds of its commit and by no rollout. +func WithRollout(ctx context.Context, id string, modules map[string]RolloutModule) context.Context { + return context.WithValue(ctx, rolloutKey{}, rolloutOf{id: id, modules: modules}) +} + +// AssignmentRuns is what each module assigned to a machine runs there, as its assignment names it: the build and the +// commit. Empty when the assignments are not on the bus, and for a module not sent yet. +func (i *Inventory) AssignmentRuns(ctx context.Context, node string) (map[string]chain.Ref, error) { + c, on, err := i.ChainReads(ctx, KindAssignments) + if err != nil || !on { + return nil, err + } + all, err := c.Assignments.All(ctx) + if err != nil { + return nil, err + } + out := map[string]chain.Ref{} + for _, k := range all { + a := k.Value + if a.Node == node && !a.Assigned.IsZero() && a.Stands() && (a.Build != "" || a.Commit != "") { + out[a.Module] = chain.Ref{Build: a.Build, Commit: a.Commit} + } + } + return out, nil +} + +// AssignmentsOn are the assignments of a machine as the chain names them, by module: the standing ones, with their links. +func (i *Inventory) AssignmentsOn(ctx context.Context, node string) ([]chain.Assignment, error) { + c, on, err := i.ChainReads(ctx, KindAssignments) + if err != nil || !on { + return nil, err + } + all, err := c.Assignments.All(ctx) + if err != nil { + return nil, err + } + var out []chain.Assignment + for _, k := range all { + if a := k.Value; a.Node == node && !a.Assigned.IsZero() && a.Stands() { + out = append(out, a) + } + } + sort.Slice(out, func(a, b int) bool { return out[a].Module < out[b].Module }) + return out, nil +} + +// recordRuns writes what a send put on a machine into the assignments, once the send is recorded: for each module it carried +// a build of — by the commit — the build, what it published, the commit, the rollout and since when. **A module that runs the +// commit already keeps what it names**, filled where it names nothing and never blanked, so a send that carries it again, or a +// push by hand, changes nothing it knows. A module the machine is not assigned is no assignment to write. +func (i *Inventory) recordRuns(ctx context.Context, nodeID string, builds map[string]string, at time.Time) error { + c, on, err := i.ChainReads(ctx, KindAssignments) + if err != nil || !on || len(builds) == 0 { + return err + } + var node string + if err := i.store.Pool().QueryRow(ctx, `select name from node where id = $1`, nodeID).Scan(&node); errors.Is(err, pgx.ErrNoRows) { + return nil + } else if err != nil { + return err + } + rollout, _ := ctx.Value(rolloutKey{}).(rolloutOf) + modules := make([]string, 0, len(builds)) + for m := range builds { + modules = append(modules, m) + } + sort.Strings(modules) + var failed []error + for _, m := range modules { + commit := builds[m] + if commit == "" { + continue + } + if err := i.recordRun(ctx, c, node, m, commit, rollout, at); err != nil { + failed = append(failed, fmt.Errorf("%s on %s: %w", m, node, err)) + } + } + return errors.Join(failed...) +} + +func (i *Inventory) recordRun(ctx context.Context, c *chain.Chain, node, module, commit string, rollout rolloutOf, at time.Time) error { + key := assignmentKey(node, module) + var build, rolloutID string + if sent, ours := rollout.modules[module]; ours && sent.Commit == commit { + build, rolloutID = sent.Build, rollout.id + } + var published *chain.Published + resolved := false + for try := 0; try < casTries; try++ { + cur, found, err := c.Assignments.Get(ctx, key) + if err != nil { + return err + } + if !found || cur.Value.Assigned.IsZero() || !cur.Value.Stands() { + return nil // nothing assigned there to say what it runs + } + same := cur.Value.Commit == commit + if same && cur.Value.Build != "" && (rolloutID == "" || cur.Value.Rollout != "") { + return nil // a send that carries what the machine runs already says nothing new + } + if !resolved { + // The build of the commit is looked up once, and only for a module that moved or names no build. + if build == "" { + if build, err = i.BuildOf(ctx, module, commit); err != nil { + return err + } + } + if build != "" { + if b, found, err := i.BuildByID(ctx, build); err != nil { + return err + } else if found { + published = publishedOf(b.Made) + } + } + resolved = true + } + next := cur.Value + if same { + // Already running it: fill what is missing, blank nothing. + if next.Build == "" { + next.Build = build + } + if next.Rollout == "" { + next.Rollout = rolloutID + } + if next.Since.IsZero() { + next.Since = at.UTC() + } + } else { + next.Commit, next.Build, next.Rollout, next.Since = commit, build, rolloutID, at.UTC() + next.Artifact, next.Package = "", nil + } + if published != nil && next.Artifact == "" && next.Package == nil && next.Build == build { + next.Artifact, next.Package = published.Artifact, published.Package + } + if next.Build == cur.Value.Build && next.Commit == cur.Value.Commit && next.Rollout == cur.Value.Rollout && + next.Artifact == cur.Value.Artifact && (next.Package == nil) == (cur.Value.Package == nil) && + next.Since.Equal(cur.Value.Since) { + return nil + } + if _, err := c.Assignments.Update(ctx, key, next, cur.Revision); errors.Is(err, chain.ErrMoved) { + continue + } else { + return err + } + } + return fmt.Errorf("the assignment kept moving under the write") +} + +// saidRunsNot says a send's links could not be written, to the log of whoever sent: the send is away and is recorded, and +// a record of what it put where that could not be written is said, never a reason to call the send failed. +func saidRunsNot(node string, err error) { + if err != nil { + fmt.Fprintf(os.Stderr, "mesh-controller: %s was sent its declaration, and the build, commit and rollout of each module "+ + "could not be written in its assignments: %v\n", node, err) + } +} diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index be27fdda..3f6c6dd2 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -1276,6 +1276,9 @@ func (i *Inventory) RecordSentUnder(ctx context.Context, node, digest string, bu _, err := i.store.Pool().Exec(ctx, `update node set sent = $2, sent_at = now(), sent_builds = $3::jsonb, sent_epoch = $4 where id = $1`, node, digest, carried, sentEpoch) + if err == nil { + saidRunsNot(node, i.recordRuns(ctx, node, builds, time.Now())) + } return err } diff --git a/internal/inventory/sends.go b/internal/inventory/sends.go index 7342c3d1..f9fdcdf4 100644 --- a/internal/inventory/sends.go +++ b/internal/inventory/sends.go @@ -174,6 +174,8 @@ func (i *Inventory) RecordSentWith(ctx context.Context, node, digest string, bui if err := tx.Commit(ctx); err != nil { return err } + // And what it put on the machine, in the assignments of the modules it carried a build of (issue 1095). + saidRunsNot(node, i.recordRuns(ctx, node, builds, time.Now())) if recordErr != nil { return &SendNotRecordedError{Err: recordErr} } diff --git a/testdata/store-tests b/testdata/store-tests index 42ed2763..0717a993 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -152,3 +152,5 @@ ./internal/inventory TestAnAssignmentIsAKeyOfTheBucketAndAnUnassignmentEndsIt ./internal/inventory TestThePlansAreKeptInTheAssignmentsValuesAndNothingElseOfThemMoves ./internal/inventory TestTheMoveCarriesTheAssignmentsAndWhatEachMachineWasSent +./cmd/mesh-controller TestOneReadOfAnAssignmentNamesItsCommitBuildAndRolloutAndOneReadOfTheRolloutListsItsAssignments +./cmd/mesh-controller TestASendOfWhatAMachineRunsAlreadyChangesNoLinkOfItsAssignment