novox/mesh-controller
18
0
Fork 0
Code Issues Pull Requests 5 Releases Wiki Activity

Drop what a send carried from the node row (migration 0109) and read an image list from the compositions (release 6, issue 1945 B8)

Browse Source
This commit is contained in:
jschoubben
2026-10-24 17:29:03 +02:00
parent b4a62e9e6e
commit 8f90d3ecd6
18 changed files with 148 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/adopting_test.go
+1 -1
View File
@@ -87,7 +87,7 @@ func reportsReaching(t *testing.T, open *stores, reachable []link.Reach, held ..
if err != nil {
t.Fatal(err)
}
if err := open.inventory.RecordSent(ctx, record.ID, digestOf(body), nil); err != nil {
if err := open.inventory.RecordSentForTest(ctx, record.ID, digestOf(body)); err != nil {
t.Fatal(err)
}
if _, err := (link.Enrolment{Inventory: open.inventory}).Heard(ctx, link.Report{
cmd/mesh-controller/awaiting_push_integration_test.go
+1 -1
View File
@@ -149,7 +149,7 @@ func TestALastSendIsReadFromTheSendAndTheReport(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if err := open.inventory.RecordSent(ctx, record.ID, "carried", map[string]string{overlay.Name: "c-overlay"}); err != nil {
if err := open.inventory.RecordSentForTest(ctx, record.ID, "carried"); err != nil {
t.Fatal(err)
}
if _, err := assign(ctx, open, "laptop", "keeper"); err != nil {
cmd/mesh-controller/awaiting_push_test.go
+1 -1
View File
@@ -59,7 +59,7 @@ func pushedBy(t *testing.T, open *stores, node string) string {
builds[m.Module] = current[m.Module].Commit
}
}
if err := open.inventory.RecordSent(ctx, record.ID, digest, builds); err != nil {
if err := open.inventory.RecordSentForTest(ctx, record.ID, digest); err != nil {
t.Fatal(err)
}
return digest
cmd/mesh-controller/compose_input.go
-3
View File
@@ -245,9 +245,6 @@ func readNode(ctx context.Context, open *stores, n inventory.Node, place invento
if out.Sequence, err = inv.Sequence(ctx, n.ID); err != nil {
return out, err
}
if out.Epoch, err = inv.SentEpoch(ctx, n.ID); err != nil {
return out, err
}
// Every node-engine of release 6 reads a module per resource and a build per module.
out.ReadsModules = true
stated, has, err := inv.HealthOf(ctx, n.Name)
cmd/mesh-controller/probes.go
+1 -3
View File
@@ -93,9 +93,7 @@ func probeDeclarations(ctx context.Context, d *doctor) ([]conditions.Observation
// With the order it was last sent, so the validator reads the envelope a machine is sent
// — its epoch included (novox/hq to-be 45 §6).
var serr error
if declared.Sequence, serr = open.inventory.Sequence(ctx, n.ID); serr == nil {
declared.Epoch, serr = open.inventory.SentEpoch(ctx, n.ID)
}
declared.Sequence, serr = open.inventory.Sequence(ctx, n.ID)
if serr != nil {
return nil, serr // the store, not the machine: the probe could not run
}
cmd/mesh-controller/registry_verbs.go
+51 -12
View File
@@ -6,6 +6,8 @@ import (
"errors"
"flag"
"fmt"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go/jetstream"
"os"
"slices"
"sort"
@@ -413,18 +415,12 @@ func imagesCommand(ctx context.Context, args []string) error {
return err
}
}
// And what it was last sent, as `plan --diff` compares against.
prev, err := open.inventory.SentSummary(ctx, node)
// And what its compositions name, which the node applies by its wants (ADR 0539 §2): read from its bucket on the bus.
sent, kept, err := keptImages(ctx, node)
if err != nil {
return err
}
var sent []sentResource
if len(prev) > 0 {
if err := json.Unmarshal(prev, &sent); err != nil {
return fmt.Errorf("%s: what it was last sent is kept in a form this version does not read: %w", node, err)
}
}
answer, err := imagesOf(node, body, sent, len(prev) > 0)
answer, err := imagesOf(node, body, sent, kept)
if err != nil {
return err
}
@@ -437,15 +433,58 @@ func imagesCommand(ctx context.Context, args []string) error {
fmt.Printf("%s (%s; %s)\n", d.Image, strings.Join(d.In, ", "), strings.Join(d.Resources, ", "))
}
if !answer.SentKnown {
fmt.Printf("%s: which images it was last sent is not kept — the next push keeps it\n", node)
fmt.Printf("%s: no composition of it is kept on the bus yet, so which images it applies is not known\n", node)
}
return nil
}
// sentResource is what the store kept of one resource a machine was last sent, as far as `images` reads it: the container's
// image. Nothing writes the summary any more (ADR 0539 §5); a machine's images are read from its compositions.
// sentResource is a container among the resources of a node's compositions, as `images` reads it: the image the node applies.
type sentResource struct {
ID string `json:"id"`
Type string `json:"type"`
Image string `json:"image,omitempty"`
}
// imagesOfCompositions are the containers of the compositions' resources, in the shape `images` compares with what would be
// declared; the bool says a composition was kept at all.
func imagesOfCompositions(values [][]byte) ([]sentResource, bool, error) {
var out []sentResource
for _, value := range values {
c, err := sdkdelivery.DecodeComposed(value)
if err != nil {
return nil, false, fmt.Errorf("a composition is not readable: %w", err)
}
for _, raw := range c.Resources {
var r struct {
ID string `json:"id"`
Type string `json:"type"`
Image string `json:"image"`
}
if err := json.Unmarshal(raw, &r); err != nil {
return nil, false, fmt.Errorf("a resource of a composition is not readable: %w", err)
}
if r.Type == "container" {
out = append(out, sentResource{ID: r.ID, Type: r.Type, Image: r.Image})
}
}
}
return out, len(values) > 0, nil
}
// keptImages reads a node's compositions from its bucket on the bus and answers their containers.
func keptImages(ctx context.Context, node string) ([]sentResource, bool, error) {
bus, err := aBus()
if err != nil {
return nil, false, fmt.Errorf("a node's compositions are kept on the bus, and this process has none: %w", err)
}
defer bus.Close()
js, err := jetstream.New(bus.Conn())
if err != nil {
return nil, false, err
}
values, err := keptCompositionsOf(ctx, js, node)
if err != nil {
return nil, false, err
}
return imagesOfCompositions(values)
}
cmd/mesh-controller/registry_verbs_test.go
+33
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"net/http"
"net/http/httptest"
"slices"
@@ -229,3 +230,35 @@ func TestImagesAreEveryContainerImageOfTheDeclarationAndOfWhatWasSent(t *testing
t.Errorf("with nothing sent yet: %+v %v", none, err)
}
}
// **`images` reads what a node applies from its compositions** (ADR 0539 §2, issue 1945 B8): the containers among the
// resources of the values kept in its bucket, whatever module and fingerprint; none kept is not known.
func TestImagesAreReadFromTheCompositionsAMachineKeeps(t *testing.T) {
file := json.RawMessage(`{"id":"web.config","type":"file","module":"web","path":"/etc/web","content":"x"}`)
box := json.RawMessage(`{"id":"web.server","type":"container","name":"web","module":"web","image":"registry/web@sha256:` + strings.Repeat("a", 64) + `"}`)
other := json.RawMessage(`{"id":"db.server","type":"container","name":"db","module":"db","image":"registry/db@sha256:` + strings.Repeat("b", 64) + `"}`)
encode := func(module string, fp string, resources ...json.RawMessage) []byte {
v, err := sdkdelivery.EncodeComposed(sdkdelivery.Composed{Node: "anchor", Module: module, FP: fp, Resources: resources})
if err != nil {
t.Fatal(err)
}
return v
}
web := encode("web", sdkdelivery.SourcePrefix2+strings.Repeat("1", 64), file, box)
db := encode("db", sdkdelivery.SourcePrefix2+strings.Repeat("2", 64), other)
got, kept, err := imagesOfCompositions([][]byte{web, db})
if err != nil || !kept || len(got) != 2 {
t.Fatalf("images %+v kept %v: %v", got, kept, err)
}
answer, err := imagesOf("anchor", nil, got, kept)
if err != nil || len(answer.Images) != 2 || !answer.SentKnown {
t.Fatalf("the answer %+v: %v", answer, err)
}
if none, kept, err := imagesOfCompositions(nil); err != nil || kept || len(none) != 0 {
t.Fatalf("no composition kept: %+v %v %v", none, kept, err)
}
if _, _, err := imagesOfCompositions([][]byte{[]byte("nope")}); err == nil {
t.Fatal("an unreadable composition was read as holding no image")
}
}
internal/inventory/doing_test.go
+2 -2
View File
@@ -191,7 +191,7 @@ func TestAMachineIsWaitingWhenWhatItWasSentIsNotWhatItShouldBe(t *testing.T) {
}
// Sent what it should be: not waiting.
if err := inv.RecordSent(ctx, anchor.ID, "aaa", nil); err != nil {
if err := inv.RecordSentForTest(ctx, anchor.ID, "aaa"); err != nil {
t.Fatal(err)
}
waiting, err = inv.Waiting(ctx, map[string]string{"anchor": "aaa", "laptop": "bbb"})
@@ -236,7 +236,7 @@ func TestAMachineWithNothingComputedForItIsNotWaiting(t *testing.T) {
// It has been sent something before, which is what makes this the case the guard is for: a
// machine with a digest and nothing computed for it would compare against the empty string
// and look out of date, when the truth is that nobody worked out what it should be.
if err := inv.RecordSent(ctx, node.ID, "what-it-got-last-time", nil); err != nil {
if err := inv.RecordSentForTest(ctx, node.ID, "what-it-got-last-time"); err != nil {
t.Fatal(err)
}
waiting, err := inv.Waiting(ctx, map[string]string{})
internal/inventory/fortest.go
+7
View File
@@ -208,3 +208,10 @@ func (i *Inventory) PlanValuesForTest(t *testing.T) map[string][]byte {
}
return out
}
// RecordSentForTest gives a machine the digest and time of a last send, as a test needs one to stand in for a send: nothing
// sends a declaration any more (ADR 0539 §5), so nothing in the controller writes them.
func (i *Inventory) RecordSentForTest(ctx context.Context, node, digest string) error {
_, err := i.store.Pool().Exec(ctx, `update node set sent = $2, sent_at = now() where id = $1`, node, digest)
return err
}
internal/inventory/given_test.go
+1 -1
View File
@@ -52,7 +52,7 @@ func sent(t *testing.T, inv *Inventory, ctx context.Context, node, digest string
if err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, n.ID, digest, nil); err != nil {
if err := inv.RecordSentForTest(ctx, n.ID, digest); err != nil {
t.Fatal(err)
}
return digest
internal/inventory/migrations/0109-the-node-keeps-no-record-of-a-send.sql
+12
View File
@@ -0,0 +1,12 @@
-- The node row keeps no record of a send's builds, epoch, summary or generation (novox/hq ADR 0539 §5, §17, issue 1945).
--
-- Nothing sends a declaration any more, so nothing writes what a send carried: the commit of each module (sent_builds, which
-- 0105 kept only as the source of the one-time copy onto the assignments, long done), the lease epoch (sent_epoch, 0068), the
-- summary `plan --diff` compared (sent_summary, 0078) and the assignment generation (sent_generation, 0094). **It is one-way**,
-- as 0108. What stays: sent and sent_at (the digest and time of the last send, which `status`' reported-and-current reads and the
-- given secrets' first good start is judged against), sent_bus_users, and sent_provisioning, which now holds the provisioner's
-- plans of each composition (inventory.RecordComposedPlans).
alter table node drop column if exists sent_builds;
alter table node drop column if exists sent_epoch;
alter table node drop column if exists sent_summary;
alter table node drop column if exists sent_generation;
internal/inventory/nodes.go
-44
View File
@@ -1242,41 +1242,6 @@ func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) {
return out, rows.Err()
}
// RecordSent keeps a digest of the declaration a machine was last sent, and the build of each module
// it carried.
//
// **A digest rather than the declaration.** The mesh can compute what a machine should be at any
// moment; keeping a copy would be a second account of it, able to disagree with the first. What
// cannot be recomputed is what was *actually sent*, and that is the whole difference between a
// machine that is out of date and one that has never been told.
//
// **And which build of each module** (novox/hq issue 259, ADR 0221): module name to the commit its
// build was made from. A digest cannot say whether a machine differs because a module moved to a build
// its upgrade policy holds back, or because of something a push made — a grant — and only the second
// is a push's to send to a machine it did not name. Nil records that it is not known, as for a
// declaration sent by hand.
func (i *Inventory) RecordSent(ctx context.Context, node, digest string, builds map[string]string) error {
return i.RecordSentUnder(ctx, node, digest, builds, 0)
}
// RecordSentUnder is RecordSent for a declaration that carried an epoch (novox/hq to-be 45 §6): kept
// beside the digest, so what the mesh would send is composed with it. Zero is one that carried none.
func (i *Inventory) RecordSentUnder(ctx context.Context, node, digest string, builds map[string]string,
epoch uint64) error {
var sentEpoch *int64
if epoch > 0 {
e := int64(epoch)
sentEpoch = &e
}
if _, err := i.store.Pool().Exec(ctx,
`update node set sent = $2, sent_at = now(), sent_builds = null, sent_epoch = $3 where id = $1`, node, digest, sentEpoch); err != nil {
return err
}
// What it was sent of each module is no longer kept in the assignments: they hold placement only (ADR 0539 §12).
_ = builds
return nil
}
// AwaitingSince is since when a machine has had something waiting for its next send (novox/hq issue
// 275): the oldest assignment on it made after it was last sent, or — when nothing was assigned since —
// when it was last sent, or when it joined if it never was. The moment a bound on "not pushed yet" is
@@ -1286,15 +1251,6 @@ func (i *Inventory) AwaitingSince(ctx context.Context, name string) (time.Time,
return i.chainAwaitingSince(ctx, name)
}
// 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), 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) {
// Never known: the assignments hold placement only (ADR 0539 §12), and what a machine runs is its have.
return nil, false, nil
}
// RecordSentBusUsers keeps a digest of the bus's user list a machine was just sent, by its name
// (novox/hq issue 249): whether the machine holding the bus must go first is whether this differs
// from the list composed now.
internal/inventory/order.go
-12
View File
@@ -142,18 +142,6 @@ func scanEpochs(rows pgx.Rows) ([]Epoch, error) {
return out, rows.Err()
}
// SentEpoch is the epoch a machine was last sent, by its id; zero for one sent without.
func (i *Inventory) SentEpoch(ctx context.Context, id string) (uint64, error) {
var epoch *int64
if err := i.store.Pool().QueryRow(ctx, `select sent_epoch from node where id = $1`, id).Scan(&epoch); err != nil {
return 0, fmt.Errorf("reading the epoch %s was last sent: %w", id, err)
}
if epoch == nil {
return 0, nil
}
return uint64(*epoch), nil
}
// ReadsEpoch says a machine's node-engine said it reads an epoch in a declaration, by its id.
func (i *Inventory) ReadsEpoch(ctx context.Context, id string) (bool, error) {
var reads bool
internal/inventory/plans_leave_the_store_test.go
+33
View File
@@ -45,3 +45,36 @@ func TestMigration0108DropsTheTablesOfThePlansMachineryAndKeepsWhatTheController
t.Errorf("the function raise_assignment_generation is still in the store")
}
}
// **Migration 0109 drops the columns of the node row that kept what a send carried** (ADR 0539 §17, issue 1945): the builds,
// the epoch, the summary and the generation. The digest and time of the last send stay, with the provisioner's plans.
func TestMigration0109DropsWhatASendCarriedFromTheNodeRow(t *testing.T) {
inv, migrate := ForTestBefore(t, 109)
columns := func() map[string]bool {
t.Helper()
out := map[string]bool{}
for _, row := range inv.QueryForTest(t, `select column_name::text from information_schema.columns where table_name = 'node'`) {
out[strings.Trim(row[0], `"`)] = true
}
return out
}
gone := []string{"sent_builds", "sent_epoch", "sent_summary", "sent_generation"}
before := columns()
for _, name := range gone {
if !before[name] {
t.Fatalf("node.%s is not in the store before the migration", name)
}
}
migrate()
after := columns()
for _, name := range gone {
if after[name] {
t.Errorf("node.%s is still in the store", name)
}
}
for _, stays := range []string{"sent", "sent_at", "sent_bus_users", "sent_provisioning"} {
if !after[stays] {
t.Errorf("node.%s is gone", stays)
}
}
}
internal/inventory/unseen.go
-19
View File
@@ -136,22 +136,3 @@ func (i *Inventory) layerNode(ctx context.Context, nodeName string) (*string, er
}
return &n.ID, nil
}
// RecordSentSummary keeps what a machine was last sent, summarised (migration 0078): read only to
// compare what would be sent with it, never as what the machine should be.
func (i *Inventory) RecordSentSummary(ctx context.Context, node string, summary []byte) error {
_, err := i.store.Pool().Exec(ctx, `update node set sent_summary = $2 where id = $1`, node, summary)
return err
}
// SentSummary is what a machine was last sent, summarised, by its name; empty for a machine sent
// nothing since the summary was first kept.
func (i *Inventory) SentSummary(ctx context.Context, name string) ([]byte, error) {
var raw []byte
err := i.store.Pool().QueryRow(ctx,
`select sent_summary from node where name = $1`, name).Scan(&raw)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
return raw, err
}
internal/inventory/unseen_test.go
-21
View File
@@ -83,24 +83,3 @@ func TestTheLayerASetReplacesIsKeptAndReadable(t *testing.T) {
t.Fatalf("the node's history changed with the mesh's: %v %v", past, err)
}
}
// What a machine was last sent is kept, summarised, and read back by its name; a machine sent
// nothing since has nothing to compare with.
func TestWhatAMachineWasSentIsKeptForComparison(t *testing.T) {
inv := fresh(t)
ctx := t.Context()
n, err := inv.AddNode(ctx, "anchor")
if err != nil {
t.Fatal(err)
}
if got, err := inv.SentSummary(ctx, "anchor"); err != nil || len(got) != 0 {
t.Fatalf("a machine never sent anything: %s %v", got, err)
}
if err := inv.RecordSentSummary(ctx, n.ID, []byte(`[{"id":"web.server","type":"container"}]`)); err != nil {
t.Fatal(err)
}
got, err := inv.SentSummary(ctx, "anchor")
if err != nil || len(got) == 0 {
t.Fatalf("read back: %s %v", got, err)
}
}
internal/link/heard_order_integration_test.go
+2 -2
View File
@@ -72,7 +72,7 @@ func TestAnOlderReportOfTheSameDeclarationIsRefused(t *testing.T) {
func TestAnOrderedAccountIsNotJudgedByTheDigestSent(t *testing.T) {
inv, heard, id := anOrderedMachine(t)
ctx := context.Background()
if err := inv.RecordSent(ctx, id, "d13", nil); err != nil {
if err := inv.RecordSentForTest(ctx, id, "d13"); err != nil {
t.Fatal(err)
}
if _, err := heard.Heard(ctx, ordered("d12", 57, 12, 41, "a")); err != nil {
@@ -93,7 +93,7 @@ func TestAnUnorderedAccountClearsTheOrderAndTheEpoch(t *testing.T) {
if _, err := heard.Heard(ctx, ordered("d12", 57, 12, 41, "a")); err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, id, "d13", nil); err != nil {
if err := inv.RecordSentForTest(ctx, id, "d13"); err != nil {
t.Fatal(err)
}
if _, err := heard.Heard(ctx, link.Report{Node: "home-server", Declared: "d13", Applied: []string{"a", "b"}}); err != nil {
internal/link/heard_test.go
+3 -3
View File
@@ -102,7 +102,7 @@ func TestABareAliveDoesNotWipeTheDeclarationThatSaysANodeIsCurrent(t *testing.T)
}
// The mesh sent this node a declaration, and the node applied it and named which by digest.
const digest = "d640d1b6a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f6071829304152"
if err := inv.RecordSent(ctx, node.ID, digest, nil); err != nil {
if err := inv.RecordSentForTest(ctx, node.ID, digest); err != nil {
t.Fatal(err)
}
if _, err := (link.Enrolment{Inventory: inv}).Heard(ctx, link.Report{
@@ -274,7 +274,7 @@ func TestAnAccountOfAnOlderDeclarationDoesNotReplaceTheNewer(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, node.ID, "d2", nil); err != nil {
if err := inv.RecordSentForTest(ctx, node.ID, "d2"); err != nil {
t.Fatal(err)
}
heard := link.Enrolment{Inventory: inv}
@@ -312,7 +312,7 @@ func TestAnAccountOfTheSentDeclarationReplacesTheKeptOne(t *testing.T) {
if _, err := heard.Heard(ctx, link.Report{Node: "home-server", Declared: "d1", Applied: []string{"a"}}); err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, node.ID, "d2", nil); err != nil {
if err := inv.RecordSentForTest(ctx, node.ID, "d2"); err != nil {
t.Fatal(err)
}
if _, err := heard.Heard(ctx, link.Report{Node: "home-server", Declared: "d2", Applied: []string{"a", "b"}}); err != nil {