Builds, rollouts and assignments live on the bus, each assignment naming its commit, build and rollout (design 47 milestone 5, issues 1089-1095) #379

Merged
mesh-tools merged 12 commits from feat/chain-on-the-bus-controller into main 2026-10-18 10:40:33 +00:00
11 changed files with 615 additions and 2 deletions
Showing only changes of commit 703badcf5a - Show all commits
+14
View File
@@ -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.
@@ -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)
}
}
+121
View File
@@ -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...)
}
+69
View File
@@ -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")
}
}
+6
View File
@@ -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".
+18 -1
View File
@@ -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 != "" {
+1 -1
View File
@@ -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 {
+190
View File
@@ -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)
}
}
+3
View File
@@ -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
}
+2
View File
@@ -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}
}
+2
View File
@@ -152,3 +152,5 @@
./internal/inventory TestAnAssignmentIsAKeyOfTheBucketAndAnUnassignmentEndsIt
./internal/inventory TestThePlansAreKeptInTheAssignmentsValuesAndNothingElseOfThemMoves
./internal/inventory TestTheMoveCarriesTheAssignmentsAndWhatEachMachineWasSent
./cmd/mesh-controller TestOneReadOfAnAssignmentNamesItsCommitBuildAndRolloutAndOneReadOfTheRolloutListsItsAssignments
./cmd/mesh-controller TestASendOfWhatAMachineRunsAlreadyChangesNoLinkOfItsAssignment