From fdab58ae4f1e10f166adfce16bf0a9d44b1b0b67 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 05:22:28 +0200 Subject: [PATCH] Read what a node was sent from its assignment keys; a pulled-in module keeps a standing key; a hand send is a time (issue 1109) SentBuilds answers from the keys of the node's assignments that name a commit, known when the node was sent anything and no key says its commit unknown. A send that names its builds writes the commit, build and rollout on each carried module's key (a module nothing assigns gets a standing key marked pulled-in, never a state) and clears what every other key of the node named; a send by hand writes sent-unknown, a time, on every key and clears their commits. A failed write of these fails the record of the send. node.sent_builds is no longer written. The plan pass ends a pulled-in key only when no send carries it. --- cmd/mesh-controller/chain_integration_test.go | 6 +- internal/chain/pending_test.go | 24 ++++ internal/chain/records.go | 8 ++ internal/inventory/chain_assignments.go | 10 +- internal/inventory/chain_links.go | 126 +++++++++++++----- internal/inventory/chain_sent_test.go | 120 +++++++++++++++++ internal/inventory/nodes.go | 52 +++++--- internal/inventory/sends.go | 23 ++-- internal/inventory/sent_builds_test.go | 4 +- testdata/store-tests | 3 + 10 files changed, 300 insertions(+), 76 deletions(-) create mode 100644 internal/inventory/chain_sent_test.go diff --git a/cmd/mesh-controller/chain_integration_test.go b/cmd/mesh-controller/chain_integration_test.go index 565ca164..e5c3610a 100644 --- a/cmd/mesh-controller/chain_integration_test.go +++ b/cmd/mesh-controller/chain_integration_test.go @@ -176,12 +176,12 @@ func TestASendOfWhatAMachineRunsAlreadyChangesNoLinkOfItsAssignment(t *testing.T if a, _ := assignmentOf(on, "notes"); a.Rollout != "plan-1" || a.Build != "build-1" || a.Commit != "c1" { t.Fatalf("the assignment names %+v", on) } - // A declaration sent by hand with no builds known says nothing of them. + // A declaration sent by hand with no builds known makes what the machine runs unknown: a time on the key, no commit. if err := inv.RecordSent(ctx, node, "d3", nil); err != nil { t.Fatal(err) } - if on, _ = inv.AssignmentsOn(ctx, "anchor"); runOf(on, "notes").Commit != "c1" { - t.Fatalf("a send of unknown builds changed %+v", on) + if on, _ = inv.AssignmentsOn(ctx, "anchor"); runOf(on, "notes").Commit != "" || runOf(on, "notes").SentUnknown.IsZero() { + t.Fatalf("a send of unknown builds left %+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 { diff --git a/internal/chain/pending_test.go b/internal/chain/pending_test.go index c0e73f9d..655aae40 100644 --- a/internal/chain/pending_test.go +++ b/internal/chain/pending_test.go @@ -61,3 +61,27 @@ func TestTheTransitionsOfAPendingAssignment(t *testing.T) { t.Error("a claim moves a wait to applying and back") } } + +// **A module only pulled in is a standing value marked pulled-in, never a state, and a hand send makes a key say its commit +// unknown with a time** (issue 1109): every reader treats a non-empty state as no assignment, so a pulled-in key must have +// none; and unknown is never written in `commit`. +func TestAPulledInValueStandsAndAHandSendIsATimeNotACommit(t *testing.T) { + at := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + pulled := Assignment{Node: "anchor", Module: "mesh-wireguard", PulledIn: true, Commit: "c1", Build: "b1", Since: at} + if !pulled.Stands() || pulled.State != "" { + t.Fatalf("a pulled-in value does not stand: %+v", pulled) + } + raw, _ := json.Marshal(pulled) + if !strings.Contains(string(raw), `"pulled-in":true`) || strings.Contains(string(raw), `"state"`) || strings.Contains(string(raw), `"assigned"`) { + t.Fatalf("the pulled-in value is %s", raw) + } + unknown := Assignment{Node: "anchor", Module: "notes", Assigned: at, SentUnknown: at} + raw, _ = json.Marshal(unknown) + if !strings.Contains(string(raw), `"sent-unknown":"2026-10-18T03:00:00Z"`) || strings.Contains(string(raw), `"commit"`) || !unknown.Stands() { + t.Fatalf("the hand-sent value is %s", raw) + } + plain, _ := json.Marshal(Assignment{Node: "anchor", Module: "notes", Assigned: at}) + if strings.Contains(string(plain), "pulled-in") || strings.Contains(string(plain), "sent-unknown") { + t.Fatalf("a value that is neither says so: %s", plain) + } +} diff --git a/internal/chain/records.go b/internal/chain/records.go index ceb23533..337f4c3b 100644 --- a/internal/chain/records.go +++ b/internal/chain/records.go @@ -196,6 +196,14 @@ type Assignment struct { // Pending is the assignment's wait for its build (novox/hq ADR 0261), kept in the value of the key it waits for and, when // it is made, carried into the value that stands: the record of what became of the wait. Pending *Pending `json:"pending,omitempty"` + + // PulledIn marks a standing value of a module that is on the machine only because another requires it, or because a + // send carried it: no time assigned, but a key all the same, so what the machine was sent of it is kept and read + // (issue 1109). Never a state: every reader treats a non-empty state as no assignment, and would take it off the machine. + PulledIn bool `json:"pulled-in,omitempty"` + // SentUnknown is when a send by hand made what the machine runs unknown (a declaration sent with no builds known): the + // commit is cleared with it. Absent normally; a send that names its builds clears it. + SentUnknown time.Time `json:"sent-unknown,omitzero"` } // StatePending is the State of a value that holds no assignment but a pending one. diff --git a/internal/inventory/chain_assignments.go b/internal/inventory/chain_assignments.go index 3a4d058f..767580ab 100644 --- a/internal/inventory/chain_assignments.go +++ b/internal/inventory/chain_assignments.go @@ -172,7 +172,7 @@ func (i *Inventory) chainAssign(ctx context.Context, c *chain.Chain, node Node, switch { case found && cur.Value.Stands(): // A module only pulled in by another becomes an assignment, keeping what it holds. - cur.Value.Assigned = now + cur.Value.Assigned, cur.Value.PulledIn = now, false _, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision) case found: // One that ended is assigned again, and one that was pending is made: a fresh value over it, the earlier one @@ -291,11 +291,11 @@ func (i *Inventory) KeepAssignmentPlans(ctx context.Context) (written, removed i case keepSkip: continue case keepCreate: - next.Assigned = time.Time{} + next.Assigned, next.PulledIn = time.Time{}, true _, err = c.Assignments.Create(ctx, key, next) case keepFresh: // A value that ended, pulled in again: a fresh one over it, the ended one staying in history. - next.Assigned = time.Time{} + next.Assigned, next.PulledIn = time.Time{}, true _, err = c.Assignments.Update(ctx, key, next, cur.Revision) default: cur.Value.RequiredBy, cur.Value.Provisioning = next.RequiredBy, next.Provisioning @@ -310,8 +310,8 @@ func (i *Inventory) KeepAssignmentPlans(ctx context.Context) (written, removed i written++ } for key, cur := range have { - if _, wanted := want[key]; wanted || !cur.Value.Assigned.IsZero() || !cur.Value.Stands() { - continue + if _, wanted := want[key]; wanted || !cur.Value.Assigned.IsZero() || !cur.Value.Stands() || cur.Value.Commit != "" { + continue // one a send still carries is on the machine, and ends when a send does not carry it any more } // A value of a module no longer pulled in, which no assignment names: it ends, and stays. cur.Value.Ended = time.Now().UTC() diff --git a/internal/inventory/chain_links.go b/internal/inventory/chain_links.go index c671c93f..cb989951 100644 --- a/internal/inventory/chain_links.go +++ b/internal/inventory/chain_links.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "os" "sort" "time" @@ -50,7 +49,7 @@ func (i *Inventory) AssignmentRuns(ctx context.Context, node string) (map[string 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 != "") { + if a.Node == node && a.Stands() && (a.Build != "" || a.Commit != "") { out[a.Module] = chain.Ref{Build: a.Build, Commit: a.Commit} } } @@ -77,14 +76,18 @@ func (i *Inventory) AssignmentsOn(ctx context.Context, node string) ([]chain.Ass 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. +// recordRuns writes what a send put on a machine into the assignments, once the send is recorded, and it is how the mesh +// knows what a machine was sent (SentBuilds). **A send that names its builds** writes, for each module it carried a build of +// — by the commit — the build, what it published, the commit, the rollout and since when, on the module's key: filled where +// the key names nothing, replaced when the commit moved, never blanked by a push by hand; a module nothing assigns, carried +// because another requires it, gets a standing key marked pulled-in; and every other key of the machine loses what it named, +// for the machine is not sent it any more. **A send that names none** (a declaration by hand) makes what the machine runs +// unknown: every key of it says so (sent-unknown) and names no commit. **An error here fails the record of the send**, for +// what a machine was sent is read from these keys and from nowhere else. 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 + c, err := i.theChain(ctx) + if err != nil { + return fmt.Errorf("what the machine runs cannot be written: %w", 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) { @@ -93,24 +96,81 @@ func (i *Inventory) recordRuns(ctx context.Context, nodeID string, builds map[st return err } rollout, _ := ctx.Value(rolloutKey{}).(rolloutOf) + all, err := c.Assignments.All(ctx) + if err != nil { + return fmt.Errorf("the assignments of %s cannot be read: %w", node, err) + } + var failed []error + for _, k := range sortedValues(all) { + a := k.Value + if a.Node != node || !a.Stands() { + continue + } + if builds[a.Module] != "" { + continue // carried, and its commit is written below + } + if err := i.forgetRun(ctx, c, k, builds == nil, at); err != nil { + failed = append(failed, fmt.Errorf("%s on %s: %w", a.Module, node, err)) + } + } 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 == "" { + if builds[m] == "" { continue } - if err := i.recordRun(ctx, c, node, m, commit, rollout, at); err != nil { + if err := i.recordRun(ctx, c, node, m, builds[m], rollout, at); err != nil { failed = append(failed, fmt.Errorf("%s on %s: %w", m, node, err)) } } return errors.Join(failed...) } +func sortedValues(all map[string]chain.Kept[chain.Assignment]) []chain.Kept[chain.Assignment] { + out := make([]chain.Kept[chain.Assignment], 0, len(all)) + for key, k := range all { + k.Key = key + out = append(out, k) + } + sort.Slice(out, func(a, b int) bool { return out[a].Key < out[b].Key }) + return out +} + +// forgetRun writes that a machine is no longer sent what a key named: its links are cleared, and for a send by hand the key +// says what the machine runs is unknown. +func (i *Inventory) forgetRun(ctx context.Context, c *chain.Chain, k chain.Kept[chain.Assignment], hand bool, at time.Time) error { + for try := 0; try < casTries; try++ { + next := k.Value + next.Build, next.Artifact, next.Package, next.Commit, next.Rollout, next.Since = "", "", nil, "", "", time.Time{} + next.SentUnknown = time.Time{} + if hand { + next.SentUnknown = at.UTC() + } + if next.Build == k.Value.Build && next.Commit == k.Value.Commit && next.Rollout == k.Value.Rollout && + next.Artifact == k.Value.Artifact && next.Package == k.Value.Package && next.Since.Equal(k.Value.Since) && + next.SentUnknown.Equal(k.Value.SentUnknown) { + return nil + } + if _, err := c.Assignments.Update(ctx, k.Key, next, k.Revision); errors.Is(err, chain.ErrMoved) { + cur, found, gerr := c.Assignments.Get(ctx, k.Key) + if gerr != nil { + return gerr + } + if !found || !cur.Value.Stands() { + return nil + } + k = chain.Kept[chain.Assignment]{Key: k.Key, Value: cur.Value, Revision: cur.Revision} + continue + } else { + return err + } + } + return fmt.Errorf("the assignment kept moving under the write") +} + 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 @@ -124,11 +184,12 @@ func (i *Inventory) recordRun(ctx context.Context, c *chain.Chain, node, module, if err != nil { return err } - if !found || cur.Value.Assigned.IsZero() || !cur.Value.Stands() { - return nil // nothing assigned there to say what it runs + if found && cur.Value.State != "" { + return nil // a wait for a build: the module is not on the machine by it } - same := cur.Value.Commit == commit - if same && cur.Value.Build != "" && (rolloutID == "" || cur.Value.Rollout != "") { + standing := found && cur.Value.Stands() + same := standing && cur.Value.Commit == commit + if same && cur.Value.Build != "" && cur.Value.SentUnknown.IsZero() && (rolloutID == "" || cur.Value.Rollout != "") { return nil // a send that carries what the machine runs already says nothing new } if !resolved { @@ -148,7 +209,13 @@ func (i *Inventory) recordRun(ctx context.Context, c *chain.Chain, node, module, resolved = true } next := cur.Value - if same { + switch { + case !standing: + // Carried and assigned to nobody here: a standing key, no time assigned, marked pulled in. An ended one is + // written over, the ended one staying in history. + next = chain.Assignment{Node: node, Module: module, PulledIn: true, Commit: commit, Build: build, Rollout: rolloutID, + Since: at.UTC()} + case same: // Already running it: fill what is missing, blank nothing. if next.Build == "" { next.Build = build @@ -159,18 +226,26 @@ func (i *Inventory) recordRun(ctx context.Context, c *chain.Chain, node, module, if next.Since.IsZero() { next.Since = at.UTC() } - } else { + default: next.Commit, next.Build, next.Rollout, next.Since = commit, build, rolloutID, at.UTC() next.Artifact, next.Package = "", nil } + next.SentUnknown = time.Time{} 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 && + if standing && 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) { + next.Since.Equal(cur.Value.Since) && next.SentUnknown.Equal(cur.Value.SentUnknown) { return nil } + if !found { + if _, err := c.Assignments.Create(ctx, key, next); errors.Is(err, chain.ErrExists) { + continue + } else { + return err + } + } if _, err := c.Assignments.Update(ctx, key, next, cur.Revision); errors.Is(err, chain.ErrMoved) { continue } else { @@ -179,12 +254,3 @@ func (i *Inventory) recordRun(ctx context.Context, c *chain.Chain, node, module, } 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/chain_sent_test.go b/internal/inventory/chain_sent_test.go new file mode 100644 index 00000000..d65614f4 --- /dev/null +++ b/internal/inventory/chain_sent_test.go @@ -0,0 +1,120 @@ +//go:build integration + +package inventory + +import ( + "context" + "errors" + "reflect" + "testing" + + "github.com/novox/mesh-controller/internal/chain" +) + +// **What a machine was sent is read from the keys of its assignments that name a commit** (ADR 0408, issue 1109): an assigned +// module's, and a module only pulled in, whose key is a standing value marked pulled-in and never a state. +func TestSentBuildsAnswersFromTheAssignmentsKeysOfTheNode(t *testing.T) { + inv, node := aNodeWithModules(t, "web", "cache") + ctx := t.Context() + record, _ := inv.NodeByName(ctx, node) + if _, err := inv.Assign(ctx, node, "web"); err != nil { + t.Fatal(err) + } + if err := inv.RecordSent(ctx, record.ID, "d1", map[string]string{"web": "c1", "cache": "c2"}); err != nil { + t.Fatal(err) + } + if builds, known, err := inv.SentBuilds(ctx, node); err != nil || !known || !reflect.DeepEqual(builds, map[string]string{"web": "c1", "cache": "c2"}) { + t.Fatalf("sent %v %v %v", builds, known, err) + } + pulled, found, _ := inv.AssignmentValue(ctx, node, "cache") + if !found || !pulled.Stands() || pulled.State != "" || !pulled.PulledIn || !pulled.Assigned.IsZero() || pulled.Commit != "c2" { + t.Fatalf("the module only carried has the key %+v", pulled) + } + if on, _ := inv.Assigned(ctx, node); !reflect.DeepEqual(on, []string{"web"}) { + t.Fatalf("a module only pulled in is assigned: %v", on) + } + // Assigned later, it keeps what it was sent and is pulled in no longer. + if _, err := inv.Assign(ctx, node, "cache"); err != nil { + t.Fatal(err) + } + made, _, _ := inv.AssignmentValue(ctx, node, "cache") + if made.PulledIn || made.Assigned.IsZero() || made.Commit != "c2" { + t.Fatalf("assigned over a pulled-in key: %+v", made) + } +} + +// **A send replaces the whole map of the node, and a hand send makes it unknown until the next send that names its builds** +// (issue 1109): a module no longer carried loses its commit, the node reads unknown when any key says so. +func TestASendReplacesTheWholeMapOfTheNodeAndAHandSendMakesItUnknown(t *testing.T) { + inv, node := aNodeWithModules(t, "web", "cache") + ctx := t.Context() + record, _ := inv.NodeByName(ctx, node) + if _, err := inv.Assign(ctx, node, "web"); err != nil { + t.Fatal(err) + } + if err := inv.RecordSent(ctx, record.ID, "d1", map[string]string{"web": "c1", "cache": "c2"}); err != nil { + t.Fatal(err) + } + if err := inv.RecordSent(ctx, record.ID, "d2", map[string]string{"web": "c3"}); err != nil { + t.Fatal(err) + } + if builds, known, _ := inv.SentBuilds(ctx, node); !known || !reflect.DeepEqual(builds, map[string]string{"web": "c3"}) { + t.Fatalf("the second send did not replace the map: %v %v", builds, known) + } + if old, _, _ := inv.AssignmentValue(ctx, node, "cache"); old.Commit != "" || old.Build != "" { + t.Fatalf("a module no longer carried keeps %+v", old) + } + // By hand: unknown, with a time and no commit anywhere. + if err := inv.RecordSent(ctx, record.ID, "d3", nil); err != nil { + t.Fatal(err) + } + if builds, known, _ := inv.SentBuilds(ctx, node); known || builds != nil { + t.Fatalf("a hand send reads as %v %v", builds, known) + } + web, _, _ := inv.AssignmentValue(ctx, node, "web") + if web.SentUnknown.IsZero() || web.Commit != "" || !web.Stands() { + t.Fatalf("web after a hand send: %+v", web) + } + if sent, _ := inv.Outstanding(ctx, node); sent != "d3" { + t.Fatalf("the digest of the hand send: %q", sent) + } + // A send that names its builds makes it known again. + if err := inv.RecordSent(ctx, record.ID, "d4", map[string]string{"web": "c4"}); err != nil { + t.Fatal(err) + } + if builds, known, _ := inv.SentBuilds(ctx, node); !known || !reflect.DeepEqual(builds, map[string]string{"web": "c4"}) { + t.Fatalf("after a send that names its builds: %v %v", builds, known) + } + if again, _, _ := inv.AssignmentValue(ctx, node, "web"); !again.SentUnknown.IsZero() { + t.Fatalf("web still says unknown: %+v", again) + } + // The column is not written any more. + if rows := inv.QueryForTest(t, `select sent_builds is null from node where name = $1`, node); rows[0][0] != "true" { + t.Fatalf("node.sent_builds was written: %v", rows) + } +} + +// failingToWrite is an assignments bucket that fails every write. +type failingToWrite struct{ chain.Backend } + +func (f failingToWrite) Create(context.Context, string, []byte) (uint64, error) { + return 0, errors.New("the bus went away") +} + +func (f failingToWrite) Update(context.Context, string, []byte, uint64) (uint64, error) { + return 0, errors.New("the bus went away") +} + +// **A links write that fails fails the record of the send, loudly** (issue 1109): what a machine was sent is read from the +// keys, so a send recorded without them would read as one that never happened. +func TestAFailedWriteOfTheLinksFailsTheRecordOfTheSend(t *testing.T) { + inv, node := aNodeWithModules(t, "web") + ctx := t.Context() + record, _ := inv.NodeByName(ctx, node) + inv.UseChain(chain.New(chain.NewMemory(), chain.NewMemory(), failingToWrite{chain.NewMemory()}, + func(context.Context) error { return nil })) + err := inv.RecordSent(ctx, record.ID, "d1", map[string]string{"web": "c1"}) + if err == nil { + t.Fatal("a send whose links could not be written was recorded") + } +} diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index 3f6c6dd2..597fe925 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -1264,22 +1264,12 @@ func (i *Inventory) RecordSentUnder(ctx context.Context, node, digest string, bu e := int64(epoch) sentEpoch = &e } - var carried *string - if builds != nil { - raw, err := json.Marshal(builds) - if err != nil { - return err - } - text := string(raw) - carried = &text + if _, err := i.store.Pool().Exec(ctx, + `update node set sent = $2, sent_at = now(), sent_epoch = $3 where id = $1`, node, digest, sentEpoch); err != nil { + return err } - _, 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 + // What it was sent of each module is the assignments' (SentBuilds); a write of it that fails fails this record. + return i.recordRuns(ctx, node, builds, time.Now()) } // AwaitingSince is since when a machine has had something waiting for its next send (novox/hq issue @@ -1307,21 +1297,39 @@ func (i *Inventory) AwaitingSince(ctx context.Context, name string) (time.Time, } // SentBuilds is the build of each module a machine was last sent, by its name: module to the commit -// its build was made from (novox/hq issue 259). Known is false when that was not kept — a machine -// last sent before it was, sent a declaration by hand, or one the mesh does not know. +// its build was made from (novox/hq issue 259), read from the keys of the machine's assignments that name a commit — an +// assigned module and one only pulled in (ADR 0408, issue 1109). Known is false when that was not kept: a machine never sent +// anything, or last sent a declaration by hand, which makes every key of it say its commit unknown. func (i *Inventory) SentBuilds(ctx context.Context, name string) (builds map[string]string, known bool, err error) { - var raw []byte - err = i.store.Pool().QueryRow(ctx, `select sent_builds from node where name = $1`, name).Scan(&raw) + var sentAt *time.Time + err = i.store.Pool().QueryRow(ctx, `select sent_at from node where name = $1`, name).Scan(&sentAt) if errors.Is(err, pgx.ErrNoRows) { return nil, false, nil } - if err != nil || raw == nil { + if err != nil || sentAt == nil { return nil, false, err } - builds = map[string]string{} - if err := json.Unmarshal(raw, &builds); err != nil { + c, err := i.theChain(ctx) + if err != nil { return nil, false, err } + all, err := c.Assignments.All(ctx) + if err != nil { + return nil, false, fmt.Errorf("what %s was sent cannot be read from the bus: %w", name, err) + } + builds = map[string]string{} + for _, k := range all { + a := k.Value + if a.Node != name || !a.Stands() { + continue + } + if !a.SentUnknown.IsZero() { + return nil, false, nil + } + if a.Commit != "" { + builds[a.Module] = a.Commit + } + } return builds, true, nil } diff --git a/internal/inventory/sends.go b/internal/inventory/sends.go index f9fdcdf4..4b28e33e 100644 --- a/internal/inventory/sends.go +++ b/internal/inventory/sends.go @@ -2,7 +2,6 @@ package inventory import ( "context" - "encoding/json" "fmt" "time" @@ -150,32 +149,26 @@ func (i *Inventory) RecordSentWith(ctx context.Context, node, digest string, bui e := int64(epoch) sentEpoch = &e } - var carried *string - if builds != nil { - raw, err := json.Marshal(builds) - if err != nil { - return err - } - text := string(raw) - carried = &text - } tx, err := i.store.Pool().Begin(ctx) if err != nil { return err } defer func() { _ = tx.Rollback(ctx) }() if _, err := tx.Exec(ctx, - `update node set sent = $2, sent_at = now(), sent_builds = $3::jsonb, sent_epoch = $4, sent_generation = $5, - sent_provisioning = $6::jsonb - where id = $1`, node, digest, carried, sentEpoch, nullIfZero(generation), nullIfEmpty(s.Provisioning)); err != nil { + `update node set sent = $2, sent_at = now(), sent_epoch = $3, sent_generation = $4, + sent_provisioning = $5::jsonb + where id = $1`, node, digest, sentEpoch, nullIfZero(generation), nullIfEmpty(s.Provisioning)); err != nil { return err } recordErr := recordSend(ctx, tx, node, s) 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())) + // And what it put on the machine, in the assignments (issues 1095, 1109): a write of it that fails fails this record, + // for what a machine was sent is read from there. + if err := i.recordRuns(ctx, node, builds, time.Now()); err != nil { + return fmt.Errorf("%s was sent its declaration, and what each module of it runs there could not be written: %w", node, err) + } if recordErr != nil { return &SendNotRecordedError{Err: recordErr} } diff --git a/internal/inventory/sent_builds_test.go b/internal/inventory/sent_builds_test.go index f28ab693..cde4b6ff 100644 --- a/internal/inventory/sent_builds_test.go +++ b/internal/inventory/sent_builds_test.go @@ -27,8 +27,10 @@ func TestASendKeepsTheBuildsItCarried(t *testing.T) { if err := inv.RecordSent(ctx, node.ID, "d1", carried); err != nil { t.Fatal(err) } + // What it was sent is read from the keys of the machine's assignments that name a commit (issue 1109): a module carried + // with no commit known names none. builds, known, err := inv.SentBuilds(ctx, "anchor") - if err != nil || !known || !reflect.DeepEqual(builds, carried) { + if err != nil || !known || !reflect.DeepEqual(builds, map[string]string{"resolver": "abc123"}) { t.Fatalf("the builds sent were not kept: %v %v %v", builds, known, err) } diff --git a/testdata/store-tests b/testdata/store-tests index 1c1c3e58..3c07590e 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -162,3 +162,6 @@ ./internal/inventory TestAPendingIsMarkedOnceAndGoesWithItsMachine ./internal/inventory TestAReplacedWaitNeverLeavesTheOperatorWithoutItsCondition ./cmd/mesh-controller TestAnAssignmentWaitingForItsBuildIsOneKeyOfTheAssignmentsBucket +./internal/inventory TestSentBuildsAnswersFromTheAssignmentsKeysOfTheNode +./internal/inventory TestASendReplacesTheWholeMapOfTheNodeAndAHandSendMakesItUnknown +./internal/inventory TestAFailedWriteOfTheLinksFailsTheRecordOfTheSend