From fe3faaef83d5b9f76bbce9feeca4f0de7b1fa930 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 24 Oct 2026 17:51:40 +0200 Subject: [PATCH] Read what a machine has from its haves: reported-and-current, the bus list, a given value's first good start; drop the digest of a send (migration 0110) (release 6, issue 1945 B8) --- cmd/mesh-controller/adopting_test.go | 7 - cmd/mesh-controller/adoption.go | 2 +- .../awaiting_push_integration_test.go | 186 ------------------ cmd/mesh-controller/awaiting_push_test.go | 128 ------------ cmd/mesh-controller/bus_step.go | 2 +- cmd/mesh-controller/doctor.go | 2 +- cmd/mesh-controller/given.go | 91 +++++++-- cmd/mesh-controller/given_test.go | 33 ++-- cmd/mesh-controller/haves.go | 117 +++++++++++ cmd/mesh-controller/haves_test.go | 102 ++++++++++ cmd/mesh-controller/helpers_kept_test.go | 34 ++++ cmd/mesh-controller/plain_words.go | 7 - cmd/mesh-controller/probes.go | 33 +--- cmd/mesh-controller/push.go | 62 ++++-- cmd/mesh-controller/rollout.go | 26 ++- cmd/mesh-controller/status.go | 2 +- cmd/mesh-controller/status_summary.go | 5 - cmd/mesh-controller/watchdogs.go | 2 +- internal/inventory/chain_assignments.go | 32 --- internal/inventory/doing_test.go | 90 --------- internal/inventory/fortest.go | 7 - internal/inventory/given.go | 26 +-- internal/inventory/given_test.go | 37 ++-- ...109-the-node-keeps-no-record-of-a-send.sql | 5 +- ...110-the-node-keeps-no-digest-of-a-send.sql | 11 ++ internal/inventory/nodes.go | 171 +--------------- .../inventory/plans_leave_the_store_test.go | 35 +++- internal/link/heard_order_integration_test.go | 6 - internal/link/heard_test.go | 13 +- 29 files changed, 498 insertions(+), 776 deletions(-) delete mode 100644 cmd/mesh-controller/awaiting_push_integration_test.go delete mode 100644 cmd/mesh-controller/awaiting_push_test.go create mode 100644 cmd/mesh-controller/haves.go create mode 100644 cmd/mesh-controller/haves_test.go create mode 100644 internal/inventory/migrations/0110-the-node-keeps-no-digest-of-a-send.sql diff --git a/cmd/mesh-controller/adopting_test.go b/cmd/mesh-controller/adopting_test.go index 6ae7e7fc..67a5b104 100644 --- a/cmd/mesh-controller/adopting_test.go +++ b/cmd/mesh-controller/adopting_test.go @@ -83,13 +83,6 @@ func reportsReaching(t *testing.T, open *stores, reachable []link.Reach, held .. if err != nil { t.Fatal(err) } - record, err := open.inventory.NodeByName(ctx, "anchor") - if err != nil { - t.Fatal(err) - } - 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{ Node: "anchor", Applied: []string{"hello-web.x"}, Declared: digestOf(body), Firewall: "ufw", Held: held, Reachable: reachable, diff --git a/cmd/mesh-controller/adoption.go b/cmd/mesh-controller/adoption.go index ad679ea5..0dd4f50f 100644 --- a/cmd/mesh-controller/adoption.go +++ b/cmd/mesh-controller/adoption.go @@ -615,7 +615,7 @@ func converge(ctx context.Context, open *stores, node string, yes bool, digest s if !record.Adopted { return "", fmt.Errorf("%s is converged already; there is nothing to flip", node) } - reports, err := inv.LastReports(ctx) + reports, err := lastReports(ctx, inv) if err != nil { return "", err } diff --git a/cmd/mesh-controller/awaiting_push_integration_test.go b/cmd/mesh-controller/awaiting_push_integration_test.go deleted file mode 100644 index bae548ac..00000000 --- a/cmd/mesh-controller/awaiting_push_integration_test.go +++ /dev/null @@ -1,186 +0,0 @@ -//go:build integration - -package main - -import ( - "errors" - "strings" - "testing" - "time" - - "git.novox.be/novox/mesh-sdk/go/catalogue" - "git.novox.be/novox/mesh-sdk/go/compose" - "git.novox.be/novox/mesh-sdk/go/overlay" - "github.com/novox/mesh-controller/internal/conditions" - "github.com/novox/mesh-controller/internal/inventory" -) - -func TestAModuleAssignedAndNotPushedIsAwaitingAPushNotUncomposable(t *testing.T) { - open := aMesh(t) - ctx := t.Context() - register(t, open, aKeeper()) - pushedBy(t, open, "laptop") // the machine was pushed before; then the module is assigned - if _, err := assign(ctx, open, "laptop", "keeper"); err != nil { - t.Fatal(err) - } - - // The read the rest of the mesh asks still refuses it, with the composer's typed error: nothing - // can be compared about a secret that does not exist (status), and nothing here string-matches. - _, _, err := composeQuestion(ctx, open, "laptop", compose.Reading, false) - var notMade *catalogue.NotMadeError - if !errors.As(err, ¬Made) || notMade.Module != "keeper" || notMade.Name != "repository" { - t.Fatalf("a read composition did not say which secret is not made, typed: %v", err) - } - // Foreseen, it composes, names what the push makes, and made nothing. - foreseen, _, err := composeQuestion(ctx, open, "laptop", compose.Foreseeing, false) - if err != nil || len(foreseen.foreseen) != 1 || foreseen.foreseen[0] != "keeper/repository" { - t.Fatalf("foreseen: %v %v", foreseen.foreseen, err) - } - if _, held, err := open.inventory.ModuleSecretIfIssued(ctx, "laptop", "keeper", "repository"); err != nil || held { - t.Fatalf("asking ahead of the push made the secret (held %v, %v)", held, err) - } - - // Within the bound: nothing at all — no urgent, no warning, nobody notified. - if got := d1(t, open); len(got) != 0 { - t.Fatalf("a machine waiting for a push raised %+v", got) - } - - // Past the bound: a warning, not urgent, saying what the push will make. - before := awaitingPushBound - awaitingPushBound = -time.Minute - t.Cleanup(func() { awaitingPushBound = before }) - got := d1(t, open) - if len(got) != 1 || got[0].Key() != "machine.laptop.awaiting-push" || got[0].Severity != conditions.Warning || - !strings.Contains(got[0].Summary, "keeper/repository") || !strings.Contains(got[0].Summary, "push laptop") { - t.Fatalf("a machine left un-pushed past the bound: %+v", got) - } - - // Pushed: the secret is made and D1 says nothing. - pushedBy(t, open, "laptop") - if got := d1(t, open); len(got) != 0 { - t.Fatalf("a pushed machine still raised %+v", got) - } -} - -func TestASecretThePushCannotMakeIsStillUncomposable(t *testing.T) { - open := aMesh(t) - ctx := t.Context() - register(t, open, aTalker()) - if _, err := assign(ctx, open, "laptop", "talker"); err != nil { - t.Fatal(err) - } - got := d1(t, open) - if len(got) != 1 || got[0].Key() != "machine.laptop.uncomposable" || got[0].Severity != conditions.Urgent || - !strings.Contains(got[0].Summary, "module issue talker --node laptop") { - t.Fatalf("a push that will be refused was not said urgently: %+v", got) - } - - // And a machine whose composition fails for anything else, with a secret waiting beside it, is - // uncomposable for that — the stand-in hides nothing. - register(t, open, aKeeper()) - if _, err := assign(ctx, open, "anchor", "keeper"); err != nil { - t.Fatal(err) - } - one, two := rivals() - register(t, open, one) - register(t, open, two) - _, _ = assign(ctx, open, "anchor", "rival-one") - _, _ = assign(ctx, open, "anchor", "rival-two") - keys := map[string]conditions.Severity{} - for _, o := range d1(t, open) { - keys[o.Key()] = o.Severity - } - if keys["machine.anchor.uncomposable"] != conditions.Urgent { - t.Fatalf("a real failure beside a waiting secret: %v", keys) - } -} - -func TestAGivenSecretUnderAnOldKeyIsNotForeseen(t *testing.T) { - open := aMesh(t) - ctx := t.Context() - register(t, open, aKeeper()) - if _, err := assign(ctx, open, "laptop", "keeper"); err != nil { - t.Fatal(err) - } - if err := open.inventory.AcceptSecretForModule(ctx, "laptop", "keeper", "repository", "given"); err != nil { - t.Fatal(err) - } - record, err := open.inventory.NodeByName(ctx, "laptop") - if err != nil { - t.Fatal(err) - } - if err := open.inventory.RecordSealingKey(ctx, record.ID, aPublicKey(t)); err != nil { - t.Fatal(err) - } - got := d1(t, open) - if len(got) != 1 || got[0].Key() != "machine.laptop.uncomposable" || !strings.Contains(got[0].Summary, "issue it again") { - t.Fatalf("a given secret under an old key: %+v", got) - } -} - -func TestAMachineAwaitsAPushSinceItsOldestAssignmentSinceTheLastSend(t *testing.T) { - open := aMesh(t) - ctx := t.Context() - register(t, open, aKeeper()) - pushedBy(t, open, "laptop") - sent := time.Now() - since, err := open.inventory.AwaitingSince(ctx, "laptop") - if err != nil || since.After(sent) { - t.Fatalf("nothing assigned since the send: waiting since %v (%v), the send was before %v", since, err, sent) - } - if _, err := assign(ctx, open, "laptop", "keeper"); err != nil { - t.Fatal(err) - } - since, err = open.inventory.AwaitingSince(ctx, "laptop") - if err != nil || since.Before(sent.Add(-time.Second)) { - t.Fatalf("assigned after the send: waiting since %v (%v), not from the assignment", since, err) - } -} - -func TestALastSendIsReadFromTheSendAndTheReport(t *testing.T) { - open := aMesh(t) - ctx := t.Context() - // What a machine was sent is read from the commits its assignments name (issue 1109): a module is carried by a send only - // when it was built from a commit, and a hand send on a machine with nothing assigned leaves no key to mark unknown. - if err := open.inventory.RegisterModule(ctx, aKeeper(), inventory.Source{BuiltFrom: "c-keeper"}); err != nil { - t.Fatal(err) - } - record, err := open.inventory.NodeByName(ctx, "laptop") - if err != nil { - t.Fatal(err) - } - if err := open.inventory.RecordSentForTest(ctx, record.ID, "carried"); err != nil { - t.Fatal(err) - } - if _, err := assign(ctx, open, "laptop", "keeper"); err != nil { - t.Fatal(err) - } - sends, err := readDeliveries(ctx, open.inventory) - if err != nil { - t.Fatal(err) - } - if sends["laptop"].settled("keeper", time.Now()) { - t.Fatal("a holder assigned and not pushed is expected to answer") - } - if !sends["laptop"].settled(overlay.Name, time.Now().Add(time.Hour)) { - t.Fatal("a module the machine was sent long ago is not expected to answer") - } - pushedBy(t, open, "laptop") - sends, err = readDeliveries(ctx, open.inventory) - if err != nil { - t.Fatal(err) - } - if sends["laptop"].settled("keeper", time.Now()) { - t.Fatal("pushed a moment ago and not reported, a holder is already expected") - } - if !sends["laptop"].settled("keeper", time.Now().Add(reportGrace+time.Minute)) { - t.Fatal("pushed past the grace, a holder is not expected") - } - if _, ok := sends["laptop"].carried["keeper"]; !ok { - t.Fatalf("the send's builds do not carry keeper: %v", sends["laptop"].carried) - } - pushedAndApplied(t, open, "laptop") - if sends, err = readDeliveries(ctx, open.inventory); err != nil || !sends["laptop"].settled("keeper", time.Now()) { - t.Fatalf("pushed and reported applied, a holder is not expected to answer (%v)", err) - } -} diff --git a/cmd/mesh-controller/awaiting_push_test.go b/cmd/mesh-controller/awaiting_push_test.go deleted file mode 100644 index 37007d65..00000000 --- a/cmd/mesh-controller/awaiting_push_test.go +++ /dev/null @@ -1,128 +0,0 @@ -package main - -import ( - "testing" - "time" - - "git.novox.be/novox/mesh-sdk/go/catalogue" - "github.com/novox/mesh-controller/internal/conditions" - "github.com/novox/mesh-controller/internal/inventory" -) - -// Between `assign` and `push` a module's own secrets are not made yet: the push makes them. D1 composed -// the machine's declaration without making anything, failed on the missing secret, and raised an urgent -// "nothing can be sent to " — seen live on 2026-10-06, a desktop notification for a machine -// the next push sent to without a word (novox/hq issue 275). D1 now composes as the push would, with a -// stand-in for what the push makes: pending, not broken, and said only past a bound, as a warning. - -// aKeeper is a module with an own secret the mesh makes — a backup repository's password. -func aKeeper() catalogue.Manifest { - return catalogue.Manifest{Module: "keeper", Version: "1", - OwnSecrets: catalogue.OwnSecrets{"repository": {Path: "/var/lib/mesh/keeper/repository"}}, - Resources: []map[string]any{ - {"id": "state", "type": "directory", "path": "/var/lib/mesh/keeper", "mode": "0700"}, - }} -} - -// pushedBy records a send to a machine as a push does — composed on the send path, so its own secrets -// are made — without a bus to carry it. -func pushedBy(t *testing.T, open *stores, node string) string { - t.Helper() - ctx := t.Context() - plan, settings, err := resolveNode(ctx, open, node) - if err != nil { - t.Fatal(err) - } - declared, err := declarationFor(ctx, open, node, plan, settings) - if err != nil { - t.Fatal(err) - } - body, err := declared.Body() - if err != nil { - t.Fatal(err) - } - record, err := open.inventory.NodeByName(ctx, node) - if err != nil { - t.Fatal(err) - } - digest := digestOf(body) - // A push records the build of each module it carries (plan.go, Allocating); a module is carried by a send only when it was - // built from a commit, for what a machine was sent is read from the commits its assignments name (issue 1109). - builds := declared.Builds - if builds == nil { - current, err := open.inventory.CurrentBuilds(ctx) - if err != nil { - t.Fatal(err) - } - builds = map[string]string{} - for _, m := range plan.Modules { - builds[m.Module] = current[m.Module].Commit - } - } - if err := open.inventory.RecordSentForTest(ctx, record.ID, digest); err != nil { - t.Fatal(err) - } - return digest -} - -// pushedAndApplied is pushedBy, and the machine reporting it applied that declaration. -func pushedAndApplied(t *testing.T, open *stores, node string) { - t.Helper() - digest := pushedBy(t, open, node) - record, err := open.inventory.NodeByName(t.Context(), node) - if err != nil { - t.Fatal(err) - } - if _, err := open.inventory.RecordDoing(t.Context(), record.ID, inventory.Doing{ - Outcome: inventory.OutcomeApplied, Declared: digest}); err != nil { - t.Fatal(err) - } -} - -// d1 runs D1 once. -func d1(t *testing.T, open *stores) []conditions.Observation { - t.Helper() - got, err := probeDeclarations(t.Context(), &doctor{open: open}) - if err != nil { - t.Fatal(err) - } - return got -} - -// **THE WINDOW, REPRODUCED**: assigned and not pushed, a module whose own secret the push makes is -// waiting, not uncomposable; past the bound it is a warning naming what the push makes; pushed, nothing. - -// **A REAL FAILURE IS STILL URGENT**: a secret the push would be refused on is not one it will make. -// A bus credential nobody issued (issue 203) fails the push, so D1 says it — urgent, in the push's words. - -// A given secret sealed to a key the machine no longer has is not the mesh's to make again: the push is -// refused on it, so D1 is too. - -// The bound is read from when the machine began waiting: the oldest assignment since its last send. - -// **D3 AND D13 WAIT FOR THE SEND**: a holder is expected to answer on a machine once the machine was sent -// it and had time to report — never between assign and push. -func TestAHolderIsExpectedOnlyOnceSentAndReported(t *testing.T) { - now := time.Now() - sent := now.Add(-time.Minute) - long := now.Add(-time.Hour) - cases := []struct { - name string - send lastSend - want bool - }{ - {"never sent", lastSend{}, false}, - {"sent without it", lastSend{sent: &long, current: true, carried: map[string]string{"other": "c"}}, false}, - {"sent with it, not reported yet", lastSend{sent: &sent, carried: map[string]string{"keeper": "c"}}, false}, - {"sent with it and reported", lastSend{sent: &sent, current: true, carried: map[string]string{"keeper": "c"}}, true}, - {"sent with it long ago, never reported", lastSend{sent: &long, carried: map[string]string{"keeper": "c"}}, true}, - {"sent before builds were kept", lastSend{sent: &long, current: true}, true}, - } - for _, c := range cases { - if got := c.send.settled("keeper", now); got != c.want { - t.Errorf("%s: settled %v, want %v", c.name, got, c.want) - } - } -} - -// And through the stores: assigned, not in the last send; pushed and reported, carried and settled. diff --git a/cmd/mesh-controller/bus_step.go b/cmd/mesh-controller/bus_step.go index fa594e6c..0d1e3202 100644 --- a/cmd/mesh-controller/bus_step.go +++ b/cmd/mesh-controller/bus_step.go @@ -391,7 +391,7 @@ func probeBusStep(ctx context.Context, d *doctor) ([]conditions.Observation, err if err != nil { return nil, err } - reports, err := inv.LastReports(ctx) + reports, err := lastReports(ctx, inv) if err != nil { return nil, err } diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index 45659067..76a147e1 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -79,7 +79,7 @@ var probeRegistry = []probe{ {ID: "D1", Asserts: "every machine's declaration composes, and passes the node-engine's validation; no module " + "assigned to a machine is left out of its declaration unsaid", From: "issues 236, 263, 275, 380", Kind: "declaration-refused", - Raises: []string{kindAwaitingPush, kindLeftOut, kindNeedsOperator}, Phase: 1, + Raises: []string{kindLeftOut, kindNeedsOperator}, Phase: 1, run: probeDeclarations}, {ID: "D2", Asserts: "every holder of the mesh's resolver answers a machine name for IPv4, and NODATA for IPv6", From: "issue 262", Kind: "resolver-wrong", Phase: 1, run: probeResolvers}, diff --git a/cmd/mesh-controller/given.go b/cmd/mesh-controller/given.go index 30ed5cf6..1c89e174 100644 --- a/cmd/mesh-controller/given.go +++ b/cmd/mesh-controller/given.go @@ -3,6 +3,8 @@ package main import ( "context" "encoding/json" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" "log" "strings" "sync" @@ -13,9 +15,8 @@ import ( // A value given by hand lives only until the module's first good start (novox/hq ADR 0228). // -// The serving controller hears every report; a clean one about the declaration a machine was last -// sent is the signal the store asks about (inventory.ReplaceGivenAfterStart). What it replaced is -// sent at once, said in the log and stated on the bus as the controller seat's `secret-replaced`, +// The serving controller watches every have; a module's have that is running and healthy is the signal the store asks about +// (inventory.ReplaceGivenAfterStart). What it replaced is composed at once, said in the log and stated on the bus as the controller seat's `secret-replaced`, // so a replacement is never silent. **Not in the hand-act log**: nobody acted by hand, and that log // is read as the count of repairs a healer is wanted for. @@ -41,29 +42,32 @@ var givenEvents link.Bus // same rows, and the store's claim makes one of them the replacer, but the sends need not race. var replacing sync.Map -// startedWell says a report is a machine's clean account of a declaration: everything applied, -// nothing failed or refused. Whether it is the declaration last sent is the store's to answer. -func startedWell(report link.Report) bool { - return report.Declared != "" && report.Refused == "" && len(report.Failed) == 0 && report.Applied != nil +// startedOn is the signal of a module's first good start: its have is running and healthy, as of the time the node-engine +// wrote it. The zero time when it is not. +func startedOn(h sdkdelivery.Have, written time.Time) time.Time { + if h.Phase == sdkdelivery.PhaseRunning && h.Health == sdkdelivery.HealthOK { + return written + } + return time.Time{} } const givenWhy = "a value given by hand lives only until its module's first good start under the mesh (novox/hq ADR 0228)" -// replaceGiven replaces what the report makes due, sends the machines, and says so. -func replaceGiven(ctx context.Context, open *stores, report link.Report) { - if _, busy := replacing.LoadOrStore(report.Node, true); busy { +// replaceGiven replaces what a module's good start makes due, composes the machines, and says so. +func replaceGiven(ctx context.Context, open *stores, node string, started map[string]time.Time) { + if _, busy := replacing.LoadOrStore(node, true); busy { return } - defer replacing.Delete(report.Node) - replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, report.Node, report.Declared) + defer replacing.Delete(node) + replaced, err := open.inventory.ReplaceGivenAfterStart(ctx, node, started) if err != nil { - log.Printf("a value given by hand on %s could not be replaced after its module started: %v", report.Node, err) + log.Printf("a value given by hand on %s could not be replaced after its module started: %v", node, err) } for _, r := range replaced { - said := SecretReplaced{Node: report.Node, Module: r.Module, Name: r.Name, Given: r.Given.UTC(), + said := SecretReplaced{Node: node, Module: r.Module, Name: r.Name, Given: r.Given.UTC(), Sent: r.Machines, Why: givenWhy} log.Printf("replaced %q of %s on %s, given %s, with a value the mesh made: %s; composing %s", - r.Name, r.Module, report.Node, r.Given.UTC().Format(time.RFC3339), givenWhy, strings.Join(r.Machines, ", ")) + r.Name, r.Module, node, r.Given.UTC().Format(time.RFC3339), givenWhy, strings.Join(r.Machines, ", ")) // Composed (novox/hq ADR 0539 clause 5): the compositions that use it get a new revision, which each node's // want then names. if _, err := composeModules(ctx, open, r.Machines); err != nil { @@ -90,3 +94,60 @@ func stateReplaced(ctx context.Context, said SecretReplaced) { log.Printf("could not say that %s's %q was replaced: %v", said.Module, said.Name, err) } } + +// watchHavesForGiven replaces the given values of a module on a node each time its have is running and healthy, until ctx +// ends: never on the watch's own goroutine. +func watchHavesForGiven(ctx context.Context, open *stores) { + bus, err := aBus() + if err != nil { + log.Printf("given values: the haves are not watched: %v", err) + return + } + js, err := jetstream.New(bus.Conn()) + if err != nil { + bus.Close() + return + } + kv, err := js.KeyValue(ctx, sdkdelivery.HaveBucket) + if err != nil { + bus.Close() + log.Printf("given values: the haves are not watched: %v", err) + return + } + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + bus.Close() + log.Printf("given values: the haves are not watched: %v", err) + return + } + go func() { + defer bus.Close() + defer w.Stop() + for { + select { + case <-ctx.Done(): + return + case e, ok := <-w.Updates(): + if !ok { + return + } + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + node, module, isNode, err := sdkdelivery.ParseHaveKey(e.Key()) + if err != nil || isNode { + continue + } + have, err := sdkdelivery.DecodeHave(e.Value()) + if err != nil { + continue + } + at := startedOn(have, e.Created()) + if at.IsZero() { + continue + } + inBackground(func() { replaceGiven(ctx, open, node, map[string]time.Time{module: at}) }) + } + } + }() +} diff --git a/cmd/mesh-controller/given_test.go b/cmd/mesh-controller/given_test.go index 8a45a465..b1e664a8 100644 --- a/cmd/mesh-controller/given_test.go +++ b/cmd/mesh-controller/given_test.go @@ -1,10 +1,11 @@ package main import ( + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "strings" "testing" - - "github.com/novox/mesh-controller/internal/link" ) // A rotation asked through the seat carries why to the command, which records it in the hand-act @@ -27,22 +28,24 @@ func TestARotationThroughTheSeatCarriesWhy(t *testing.T) { } } -// Only a clean account of a declaration is a good start; a refusal, a failure or a bare word that -// the machine is there is not (novox/hq ADR 0228). -func TestAGoodStartIsACleanAccountOfADeclaration(t *testing.T) { +// Only a have that is running and healthy is a good start, at the time it was written; any other is none (novox/hq ADR 0228, +// ADR 0539 §9). +func TestAGoodStartIsAHaveRunningAndHealthy(t *testing.T) { + at := time.Date(2026, 10, 24, 17, 0, 0, 0, time.UTC) for _, c := range []struct { - report link.Report - good bool + have sdkdelivery.Have + good bool }{ - {link.Report{Node: "anchor", Declared: "d", Applied: []string{"container:letta"}}, true}, - {link.Report{Node: "anchor", Declared: "d", Applied: []string{}}, true}, - {link.Report{Node: "anchor"}, false}, - {link.Report{Node: "anchor", Applied: []string{"x"}}, false}, - {link.Report{Node: "anchor", Declared: "d", Applied: []string{"x"}, Failed: map[string]string{"y": "no"}}, false}, - {link.Report{Node: "anchor", Declared: "d", Refused: "older"}, false}, + {sdkdelivery.Have{Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK}, true}, + {sdkdelivery.Have{Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthFailed}, false}, + {sdkdelivery.Have{Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthWaits}, false}, + {sdkdelivery.Have{Phase: sdkdelivery.PhaseApplying, Health: sdkdelivery.HealthOK}, false}, + {sdkdelivery.Have{Phase: sdkdelivery.PhaseFailed, Health: sdkdelivery.HealthFailed}, false}, + {sdkdelivery.Have{Phase: sdkdelivery.PhaseHeld, Health: sdkdelivery.HealthOK}, false}, } { - if got := startedWell(c.report); got != c.good { - t.Errorf("%+v: a good start = %v, want %v", c.report, got, c.good) + got := startedOn(c.have, at) + if good := !got.IsZero(); good != c.good || good && !got.Equal(at) { + t.Errorf("%+v: started at %v, want a good start = %v", c.have, got, c.good) } } } diff --git a/cmd/mesh-controller/haves.go b/cmd/mesh-controller/haves.go new file mode 100644 index 00000000..5d1f74d0 --- /dev/null +++ b/cmd/mesh-controller/haves.go @@ -0,0 +1,117 @@ +package main + +import ( + "context" + "errors" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// What a machine has, against what it is wanted to have (novox/hq ADR 0539 §1, §9). Nothing is sent, so nothing records that a +// machine was sent anything: whether a machine has "acted on the current words" is read from its haves, the node-engine's +// own bucket, beside the assigner's wants. A node is current when every module wanted on it is had at the wanted fingerprint +// and composition, applied (not failed, not held, not still applying), and every module wanted removed is gone. A node that +// nothing is wanted on has nothing to apply and is current. + +// currentNodes is, for each node a want names, whether it has everything wanted of it. Pure. +func currentNodes(wants map[string]sdkdelivery.Want, haves map[string]sdkdelivery.Have) map[string]bool { + out := map[string]bool{} + for key, w := range wants { + node, _, err := sdkdelivery.ParseWantKey(key) + if err != nil { + continue // the node's own want: its composition is the node-engine's own + } + if _, seen := out[node]; !seen { + out[node] = true + } + h, had := haves[key] + switch { + case w.Removed: + if had && h.Phase != sdkdelivery.PhaseRemoved { + out[node] = false + } + case !had, h.FP != w.FP, h.Composed != w.Composed, h.Phase != sdkdelivery.PhaseRunning: + out[node] = false + } + } + return out +} + +// readWantsAndHaves reads the assigner's wants and the node-engines' haves from the bus; a bucket not there yet is empty. +func readWantsAndHaves(ctx context.Context, js jetstream.JetStream) (map[string]sdkdelivery.Want, map[string]sdkdelivery.Have, error) { + wants := map[string]sdkdelivery.Want{} + haves := map[string]sdkdelivery.Have{} + each := func(bucket string, take func(key string, value []byte)) error { + kv, err := js.KeyValue(ctx, bucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + return nil + } + if err != nil { + return err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil + } + if err != nil { + return err + } + for _, key := range keys { + e, err := kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + if err != nil { + return err + } + take(key, e.Value()) + } + return nil + } + if err := each(sdkdelivery.AssignerWantsBucket, func(key string, value []byte) { + if w, err := sdkdelivery.DecodeWant(value); err == nil { + wants[key] = w + } + }); err != nil { + return nil, nil, err + } + if err := each(sdkdelivery.HaveBucket, func(key string, value []byte) { + if h, err := sdkdelivery.DecodeHave(value); err == nil { + haves[key] = h + } + }); err != nil { + return nil, nil, err + } + return wants, haves, nil +} + +// lastReports is inventory.LastReports with Current read from the haves: every machine's last report, and whether the machine +// has everything the assigner wants of it. A process with no bus cannot tell, and leaves Current false. +func lastReports(ctx context.Context, inv *inventory.Inventory) ([]inventory.Reported, error) { + reports, err := inv.LastReports(ctx) + if err != nil { + return nil, err + } + bus, err := aBus() + if err != nil { + return reports, nil // empty-on-error: no bus to read the haves from; Current stays false, which is "not known to be" + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return nil, err + } + wants, haves, err := readWantsAndHaves(ctx, js) + if err != nil { + return nil, err + } + current := currentNodes(wants, haves) + for i := range reports { + c, wanted := current[reports[i].Node] + reports[i].Current = c || !wanted && reports[i].At != nil + } + return reports, nil +} diff --git a/cmd/mesh-controller/haves_test.go b/cmd/mesh-controller/haves_test.go new file mode 100644 index 00000000..9d9e8ac8 --- /dev/null +++ b/cmd/mesh-controller/haves_test.go @@ -0,0 +1,102 @@ +package main + +import ( + "errors" + "strings" + "testing" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **A machine is current when it has what is wanted of it** (novox/hq ADR 0539 §1, §9, issue 1945): every module wanted on it is +// had at the wanted fingerprint and composition, running, and every module wanted removed is gone. Read from the wants and the +// haves; there is no send to compare a report with. +func TestAMachineIsCurrentWhenItHasWhatIsWantedOfIt(t *testing.T) { + fp := func(c string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(c, 64) } + want := func(f, composed string) sdkdelivery.Want { return sdkdelivery.Want{V: 1, FP: f, Composed: composed} } + have := func(f, composed string, phase sdkdelivery.Phase) sdkdelivery.Have { + return sdkdelivery.Have{V: 1, FP: f, Composed: composed, Phase: phase, Health: sdkdelivery.HealthOK} + } + key := func(node, module string) string { + k, err := sdkdelivery.WantKey(node, module) + if err != nil { + t.Fatal(err) + } + return k + } + for _, c := range []struct { + what string + wants map[string]sdkdelivery.Want + haves map[string]sdkdelivery.Have + current map[string]bool + }{ + {"everything had as wanted", + map[string]sdkdelivery.Want{key("anchor", "web"): want(fp("a"), "d1")}, + map[string]sdkdelivery.Have{key("anchor", "web"): have(fp("a"), "d1", sdkdelivery.PhaseRunning)}, + map[string]bool{"anchor": true}}, + {"a module not had yet", + map[string]sdkdelivery.Want{key("anchor", "web"): want(fp("a"), "d1")}, + nil, map[string]bool{"anchor": false}}, + {"an older fingerprint", + map[string]sdkdelivery.Want{key("anchor", "web"): want(fp("b"), "d1")}, + map[string]sdkdelivery.Have{key("anchor", "web"): have(fp("a"), "d1", sdkdelivery.PhaseRunning)}, + map[string]bool{"anchor": false}}, + {"the same fingerprint, an older composition", + map[string]sdkdelivery.Want{key("anchor", "web"): want(fp("a"), "d2")}, + map[string]sdkdelivery.Have{key("anchor", "web"): have(fp("a"), "d1", sdkdelivery.PhaseRunning)}, + map[string]bool{"anchor": false}}, + {"still applying", + map[string]sdkdelivery.Want{key("anchor", "web"): want(fp("a"), "d1")}, + map[string]sdkdelivery.Have{key("anchor", "web"): have(fp("a"), "d1", sdkdelivery.PhaseApplying)}, + map[string]bool{"anchor": false}}, + {"held for a backup that failed", + map[string]sdkdelivery.Want{key("anchor", "web"): want(fp("a"), "d1")}, + map[string]sdkdelivery.Have{key("anchor", "web"): have(fp("a"), "d1", sdkdelivery.PhaseHeld)}, + map[string]bool{"anchor": false}}, + {"a module wanted removed that is gone", + map[string]sdkdelivery.Want{key("anchor", "web"): {V: 1, Removed: true}}, + map[string]sdkdelivery.Have{key("anchor", "web"): have("", "", sdkdelivery.PhaseRemoved)}, + map[string]bool{"anchor": true}}, + {"a module wanted removed that still runs", + map[string]sdkdelivery.Want{key("anchor", "web"): {V: 1, Removed: true}}, + map[string]sdkdelivery.Have{key("anchor", "web"): have(fp("a"), "d1", sdkdelivery.PhaseRunning)}, + map[string]bool{"anchor": false}}, + {"one machine behind does not hold another back", + map[string]sdkdelivery.Want{key("anchor", "web"): want(fp("a"), "d1"), key("laptop", "web"): want(fp("a"), "d1")}, + map[string]sdkdelivery.Have{key("anchor", "web"): have(fp("a"), "d1", sdkdelivery.PhaseRunning)}, + map[string]bool{"anchor": true, "laptop": false}}, + } { + got := currentNodes(c.wants, c.haves) + for node, want := range c.current { + if got[node] != want { + t.Errorf("%s: %s is current %v, want %v", c.what, node, got[node], want) + } + } + } + if got := currentNodes(nil, nil); len(got) != 0 { + t.Errorf("no want names a node, yet %v", got) + } +} + +// **The bus's machine is behind while its have does not name the composition carrying the user list** (ADR 0539 §15): the +// digest of the bus module's composition written now against the one its have names. +func TestTheBusMachineIsBehindUntilItsHaveNamesTheCompositionCarryingTheUserList(t *testing.T) { + have := sdkdelivery.Have{Composed: "d2", Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK} + for _, c := range []struct { + what string + digest string + err error + have sdkdelivery.Have + had bool + behind bool + }{ + {"its have names the composition written now", "d2", nil, have, true, false}, + {"its have names an older composition (a new user list)", "d3", nil, have, true, true}, + {"it has no have of the module yet", "d2", nil, sdkdelivery.Have{}, false, true}, + {"the composition cannot be read: not known is behind", "", errors.New("no composition"), have, true, true}, + } { + if got := listBehind(c.digest, c.err, c.have, c.had); got != c.behind { + t.Errorf("%s: behind %v, want %v", c.what, got, c.behind) + } + } +} diff --git a/cmd/mesh-controller/helpers_kept_test.go b/cmd/mesh-controller/helpers_kept_test.go index 0efe6a8b..90a126d4 100644 --- a/cmd/mesh-controller/helpers_kept_test.go +++ b/cmd/mesh-controller/helpers_kept_test.go @@ -48,3 +48,37 @@ func aContainer(name string) catalogue.Manifest { return catalogue.Manifest{Module: name, Version: "1", Resources: []map[string]any{{"id": name, "type": "container", "image": name}}} } + +// pushedBy composes a machine's declaration as the compositions are made, so its own secrets are made, and answers its digest. +// Nothing records a send: there is none (ADR 0539 §5). +func pushedBy(t *testing.T, open *stores, node string) string { + t.Helper() + ctx := t.Context() + plan, settings, err := resolveNode(ctx, open, node) + if err != nil { + t.Fatal(err) + } + declared, err := declarationFor(ctx, open, node, plan, settings) + if err != nil { + t.Fatal(err) + } + body, err := declared.Body() + if err != nil { + t.Fatal(err) + } + return digestOf(body) +} + +// pushedAndApplied is pushedBy, and the machine reporting it applied that declaration. +func pushedAndApplied(t *testing.T, open *stores, node string) { + t.Helper() + digest := pushedBy(t, open, node) + record, err := open.inventory.NodeByName(t.Context(), node) + if err != nil { + t.Fatal(err) + } + if _, err := open.inventory.RecordDoing(t.Context(), record.ID, inventory.Doing{ + Outcome: inventory.OutcomeApplied, Declared: digest}); err != nil { + t.Fatal(err) + } +} diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 1b04411d..b6b73bdb 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -94,13 +94,6 @@ var plainWordings = map[string]func(conditions.Observation) words{ "the mesh until it does. A machine asleep is normal.", m, m), Resolved: "Tools on " + m + " answer again"} }), - kindAwaitingPush: worded(func(o conditions.Observation) words { - m := machineOr(o, "a machine") - return words{Headline: m + " waits for new instructions", - Explanation: fmt.Sprintf("%s's new instructions go out with the next push. Nothing is broken meanwhile.", - conditions.Capital(m)), - Resolved: m + " got its new instructions"} - }), "declaration-refused": worded(func(o conditions.Observation) words { if o.Scope == conditions.ScopeMesh { return words{Headline: "No machine can get new instructions", diff --git a/cmd/mesh-controller/probes.go b/cmd/mesh-controller/probes.go index 37d40602..50f11afa 100644 --- a/cmd/mesh-controller/probes.go +++ b/cmd/mesh-controller/probes.go @@ -34,13 +34,6 @@ import ( // wrong now as observations, or an error when it could not tell — never "nothing" for "could not // look" (ADR 0227 rule 4). -// awaitingPushBound is how long a machine may have something only its next push makes — an own -// secret of a module assigned since its last push — before that is said (novox/hq issue 275). -var awaitingPushBound = 30 * time.Minute - -// kindAwaitingPush is D1's kind for a machine waiting for a push past awaitingPushBound. -const kindAwaitingPush = "awaiting-push" - // probeDeclarations is D1: every machine's declaration composes, and the node-engine's own validator // (mesh-host's `validate`, the package the host runs) takes it. // @@ -80,7 +73,7 @@ func probeDeclarations(ctx context.Context, d *doctor) ([]conditions.Observation if err != nil && !unresolvable(err) { return nil, fmt.Errorf("%s cannot be worked out: %w", n.Name, err) } - var problems, foreseen []string + var problems []string // Each module the next send leaves out of this machine's declaration (novox/hq issue 380), judged from this // resolution as the send judges it: a finding, never a refusal. if err == nil { @@ -89,7 +82,6 @@ func probeDeclarations(ctx context.Context, d *doctor) ([]conditions.Observation if err == nil && gensErr == nil { var declared sendable if declared, _, err = composeQuestion(ctx, open, n.Name, compose.Foreseeing, false); err == nil { - foreseen = declared.foreseen // 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 @@ -119,30 +111,11 @@ func probeDeclarations(ctx context.Context, d *doctor) ([]conditions.Observation Summary: fmt.Sprintf("%s's node-engine would refuse its declaration whole: %d problem(s), the first: %s", n.Name, len(problems), problems[0]), Said: strings.Join(problems, "; ")}) - case len(foreseen) > 0: - since, err := open.inventory.AwaitingSince(ctx, n.Name) - if err != nil { - return nil, err // the store, not the machine: the probe could not run - } - if waited := time.Since(since); waited > awaitingPushBound { - out = append(out, awaitingPush(n.Name, foreseen, since, waited)) - } } } return out, nil } -// awaitingPush is D1's warning for a machine whose declaration composes only once its next push has -// made what it names, past awaitingPushBound (novox/hq issue 275). -func awaitingPush(node string, foreseen []string, since time.Time, waited time.Duration) conditions.Observation { - made := strings.Join(foreseen, ", ") - return conditions.Observation{Scope: conditions.ScopeMachine, ID: node, Token: kindAwaitingPush, - Kind: kindAwaitingPush, Machine: node, Severity: conditions.Warning, - Summary: fmt.Sprintf("%s has waited %s for a push: its declaration composes once the next push makes %s — "+ - "`push %s` sends it", node, waited.Round(time.Minute), made, node), - Said: fmt.Sprintf("waiting since %s; the next push makes %s", since.UTC().Format(time.RFC3339), made)} -} - // probeResolvers is D2: every holder of the mesh's resolver answers each machine's name with its // address for IPv4, and with no address and no error for IPv6 — NODATA, not NXDOMAIN, which musl // takes as final (issue 262). @@ -600,7 +573,7 @@ func (d lastSend) settled(module string, now time.Time) bool { // readDeliveries is every machine's last send, by name, from the records a send and a report keep. func readDeliveries(ctx context.Context, inv *inventory.Inventory) (map[string]lastSend, error) { - reports, err := inv.LastReports(ctx) + reports, err := lastReports(ctx, inv) if err != nil { return nil, err } @@ -1087,7 +1060,7 @@ func probeCoreBuilds(ctx context.Context, d *doctor) ([]conditions.Observation, if err != nil { return nil, err } - reports, err := inv.LastReports(ctx) + reports, err := lastReports(ctx, inv) if err != nil { return nil, err } diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index a4ce877d..4da42308 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -6,6 +6,8 @@ import ( "encoding/hex" "errors" "fmt" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/conditions" "log" "os" @@ -254,6 +256,8 @@ func serve(ctx context.Context) (err error) { // And the modules the catalogue does not hold yet, registered from the builder's first done key of each, woken by // the builder's and the indexer's writes (novox/hq ADR 0501 §6, issue 1851): nothing asks a first build. watchNewModules(ctx, open) + // And a value given by hand replaced when its module's have first runs healthy (novox/hq ADR 0228, ADR 0539 §9). + watchHavesForGiven(ctx, open) // And says what it keeps, and keeps what every module says it keeps (novox/hq ADR 0495, issue 1790): a changed word // sends the machine holding the bus its new user list. if err := announceOwnStates(ctx, server.JetStream()); err != nil { @@ -350,35 +354,53 @@ func brokerListState(ctx context.Context, open *stores, names []string) (busList return busListState{Machine: h.Node}, nil } } - // **What cannot be worked out is behind** (the review of hq issue 353). The list it would be sent cannot be - // composed, so whether it carries the grants a send relies on is not known; it is taken as behind, so a push - // sends that machine first and says why it cannot (`plan` says it too), and a rollout that needs the list - // refuses rather than sending code whose grants the bus may not hold. It said "not behind" until then: a - // send went ahead as if the bus held a list nobody could compose. - list, composeErr := "", error(nil) - plan, _, err := resolveNode(ctx, open, h.Node) + // **What cannot be worked out is behind** (the review of hq issue 353): the composition carrying the user list is not + // written, so whether the bus's machine runs it is not known; taken as behind, never "not behind". + // The user list is part of the bus module's composition on the machine holding it (ADR 0539 §15), so the machine is + // behind while its have does not name the digest of the composition written now. + fps, err := currentFingerprints(ctx, inv) if err != nil { - composeErr = err - } else if list, _, err = busUserList(ctx, inv, plan.Modules); err != nil { - composeErr = err + return busListState{Machine: h.Node, Behind: true, Uncomposed: err}, nil } - sent, err := inv.SentBusUsers(ctx, h.Node) + bus, err := aBus() if err != nil { return busListState{}, err } - return busListState{Machine: h.Node, Behind: listBehind(list, composeErr, sent), Uncomposed: composeErr}, nil + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return busListState{}, err + } + digest, digestErr := composedDigestOf(ctx, js, h.Node, h.Module, fps[h.Module]) + have, had, err := haveOn(ctx, js, h.Node, h.Module) + if err != nil { + return busListState{}, err + } + return busListState{Machine: h.Node, Behind: listBehind(digest, digestErr, have, had), Uncomposed: digestErr}, nil } -// listBehind is whether the bus's machine is behind: the list composed now is not the one last sent, or it -// could not be composed at all — not known is behind, never "not behind". -func listBehind(list string, composeErr error, sentDigest string) bool { - return composeErr != nil || userListBehind(list, sentDigest) +// composedDigestOf is the digest of the composition the controller wrote for a module on a node at a fingerprint. +func composedDigestOf(ctx context.Context, js jetstream.JetStream, node, module, fp string) (string, error) { + digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket) + if err != nil { + return "", err + } + key, err := sdkdelivery.ComposedKey(node, module, fp) + if err != nil { + return "", err + } + e, err := digests.Get(ctx, key) + if err != nil { + return "", fmt.Errorf("the composition %s is not written: %w", key, err) + } + v, err := sdkdelivery.DecodeComposedDigest(e.Value()) + return v.Digest, err } -// userListBehind is whether the user list composed now is not the one last sent, by its digest. An -// empty list composed is never behind: there is nothing for it to carry. -func userListBehind(now, sentDigest string) bool { - return now != "" && digestOf([]byte(now)) != sentDigest +// listBehind is whether the bus's machine is behind: its have does not name the composition written now, or the +// composition could not be read at all. +func listBehind(digest string, digestErr error, have sdkdelivery.Have, had bool) bool { + return digestErr != nil || !had || have.Composed != digest } // digestOf is what the mesh compares to answer "has this machine been sent what it should be". diff --git a/cmd/mesh-controller/rollout.go b/cmd/mesh-controller/rollout.go index a8d16bec..d869ba80 100644 --- a/cmd/mesh-controller/rollout.go +++ b/cmd/mesh-controller/rollout.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "github.com/nats-io/nats.go/jetstream" "os" "strings" "time" @@ -170,7 +171,7 @@ func readinessOf(ctx context.Context, inv *inventory.Inventory) (broker.Readines // The machine that holds the bus seat is the one that would be sent the user list. if m.BusUsers != "" && m.ClaimsSeat("mesh-broker") { state.Holder = n.Name - state.AccountsComposed = wasSentTheUserList(ctx, inv, n.Name) + state.AccountsComposed = wasGivenTheUserList(ctx, n.Name, module) } // A module that never speaks needs no credential, so it is not counted as missing one. if !speaksOnTheBus(m) { @@ -196,14 +197,23 @@ func speaksOnTheBus(m catalogue.Manifest) bool { len(m.DefinesSeats) > 0 || len(m.Uses) > 0 || len(m.Claims) > 0 } -// wasSentTheUserList says whether the machine holding the bus has had a declaration since the user -// list became part of one. +// wasGivenTheUserList says whether the machine holding the bus has a composition of the bus module applied since the +// user list became part of one: its have names a composition at all. // -// Read from what the mesh recorded sending rather than asked of the machine: a machine that is away -// has still been sent it, and this question is about whether the mesh did its part. -func wasSentTheUserList(ctx context.Context, inv *inventory.Inventory, node string) bool { - digest, err := inv.Outstanding(ctx, node) - return err == nil && strings.TrimSpace(digest) != "" +// Read from the node-engine's have rather than asked of the machine: a machine that is away has still been given it, and +// this question is about whether the mesh did its part. +func wasGivenTheUserList(ctx context.Context, node, module string) bool { + bus, err := aBus() + if err != nil { + return false + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) + if err != nil { + return false + } + have, had, err := haveOn(ctx, js, node, module) + return err == nil && had && have.Composed != "" } // notReadyOf is the readiness reasoning, named here so a test can reach it without the command's diff --git a/cmd/mesh-controller/status.go b/cmd/mesh-controller/status.go index c28cbe5b..d2e940f2 100644 --- a/cmd/mesh-controller/status.go +++ b/cmd/mesh-controller/status.go @@ -489,7 +489,7 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) { out.heals = countHeals(heals) } - out.reported, err = inv.LastReports(ctx) + out.reported, err = lastReports(ctx, inv) if err != nil { return answers{}, err } diff --git a/cmd/mesh-controller/status_summary.go b/cmd/mesh-controller/status_summary.go index 455624cf..3ea128f2 100644 --- a/cmd/mesh-controller/status_summary.go +++ b/cmd/mesh-controller/status_summary.go @@ -207,11 +207,6 @@ func (l nudgingListener) Heard(ctx context.Context, report link.Report) (bool, e report.Node, herr) } } - if err == nil && l.open != nil && startedWell(report) { - // Off the report's path: replacing a given value sends the machine, and a report waits for - // nothing it caused (novox/hq ADR 0228). - go replaceGiven(context.WithoutCancel(ctx), l.open, report) - } // And, once the move of the secrets into the vault has started, what is left of this machine's (design 57's V3, // issue 676): a machine asleep when it started is moved when it is next heard from. Off the report's path. if err == nil && l.open != nil && report.Superseded == "" && report.Rekey == nil && report.Node != "" { diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index 9597148d..cdce1875 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -380,7 +380,7 @@ func (w *watchdogs) gatherMachines(ctx context.Context, inv *inventory.Inventory if err != nil { return nil, fmt.Errorf("the machines cannot be read: %w", err) } - reports, err := inv.LastReports(ctx) + reports, err := lastReports(ctx, inv) if err != nil { return nil, fmt.Errorf("what each machine last reported cannot be read: %w", err) } diff --git a/internal/inventory/chain_assignments.go b/internal/inventory/chain_assignments.go index 79e9ece2..1d56aae6 100644 --- a/internal/inventory/chain_assignments.go +++ b/internal/inventory/chain_assignments.go @@ -9,8 +9,6 @@ import ( "strings" "time" - "github.com/jackc/pgx/v5" - "github.com/novox/mesh-controller/internal/chain" ) @@ -190,36 +188,6 @@ func (i *Inventory) chainUnassign(ctx context.Context, c *chain.Chain, node Node return fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name) } -// chainAwaitingSince is AwaitingSince over the bucket: the oldest assignment on the machine made after it was last sent. -func (i *Inventory) chainAwaitingSince(ctx context.Context, name string) (time.Time, error) { - var sentAt *time.Time - var created time.Time - err := i.store.Pool().QueryRow(ctx, `select sent_at, created from node where name = $1`, name).Scan(&sentAt, &created) - if errors.Is(err, pgx.ErrNoRows) { - return time.Time{}, fmt.Errorf("%w: %s", ErrNoSuchNode, name) - } - if err != nil { - return time.Time{}, err - } - rows, err := i.assignments(ctx) - if err != nil { - return time.Time{}, err - } - var oldest time.Time - for _, a := range rows { - if a.Node == name && (sentAt == nil || a.Assigned.After(*sentAt)) && (oldest.IsZero() || a.Assigned.Before(oldest)) { - oldest = a.Assigned - } - } - switch { - case !oldest.IsZero(): - return oldest, nil - case sentAt != nil: - return *sentAt, nil - } - return created, nil -} - // KeepAssignmentPlans writes the provisioner's plan of each assignment into its value on the bus, and a value of its own for // each module only pulled in by another on a machine (issue 765), and ends the value of one that is no longer pulled in. The // plans are derived from what each machine was last sent, not part of the record of what is assigned, so they are read diff --git a/internal/inventory/doing_test.go b/internal/inventory/doing_test.go index 7666fccc..8a932146 100644 --- a/internal/inventory/doing_test.go +++ b/internal/inventory/doing_test.go @@ -158,96 +158,6 @@ func TestWhatANodeSaidGoesWhenTheNodeDoes(t *testing.T) { } } -// "Behind" must mean not running what the mesh would send, not only "failed". -// -// novox/hq ADR 0010 names losing "did my change go out?" as the real risk of replacing a pipeline -// with a comparison. With behind meaning only failed-or-refused, that question was answerable -// exactly for the machines that broke — and for every machine that worked, the answer was silence -// whether the change had gone out or not. -func TestAMachineIsWaitingWhenWhatItWasSentIsNotWhatItShouldBe(t *testing.T) { - inv := ForTest(t) - ctx := t.Context() - anchor, err := inv.AddNode(ctx, "anchor") - if err != nil { - t.Fatal(err) - } - if _, err := inv.AddNode(ctx, "laptop"); err != nil { - t.Fatal(err) - } - - // Never sent anything: waiting, and said differently. Nobody has ever asked it to be - // anything, which is not the same as it being out of date. - waiting, err := inv.Waiting(ctx, map[string]string{"anchor": "aaa", "laptop": "bbb"}) - if err != nil { - t.Fatal(err) - } - if len(waiting) != 2 { - t.Fatalf("machines that were never sent anything are not waiting: %+v", waiting) - } - for _, m := range waiting { - if !m.Never { - t.Fatalf("%s was never sent anything and does not say so: %+v", m.Node, m) - } - } - - // Sent what it should be: not waiting. - 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"}) - if err != nil { - t.Fatal(err) - } - if len(waiting) != 1 || waiting[0].Node != "laptop" { - t.Fatalf("a machine sent exactly what it should be is still waiting: %+v", waiting) - } - - // The declaration changes: waiting again, and no longer "never". - waiting, err = inv.Waiting(ctx, map[string]string{"anchor": "ccc", "laptop": "bbb"}) - if err != nil { - t.Fatal(err) - } - var found bool - for _, m := range waiting { - if m.Node != "anchor" { - continue - } - found = true - if m.Never { - t.Fatal("a machine that has been sent something is reported as never told") - } - if m.Sent != "aaa" { - t.Fatalf("what it was last sent was lost: %+v", m) - } - } - if !found { - t.Fatal("a machine whose declaration changed since it was sent is not waiting") - } -} - -// A machine nobody worked out is not reported as waiting: saying so would invent a comparison. -func TestAMachineWithNothingComputedForItIsNotWaiting(t *testing.T) { - inv := ForTest(t) - ctx := t.Context() - node, err := inv.AddNode(ctx, "unresolvable") - if err != nil { - t.Fatal(err) - } - // 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.RecordSentForTest(ctx, node.ID, "what-it-got-last-time"); err != nil { - t.Fatal(err) - } - waiting, err := inv.Waiting(ctx, map[string]string{}) - if err != nil { - t.Fatal(err) - } - if len(waiting) != 0 { - t.Fatalf("a machine the caller could not work out was reported as waiting: %+v", waiting) - } -} - // **A failure that repeats is told apart from one that just happened.** A node re-applies on a // steady interval and reports each time, so a resource that will never apply arrives as the same // report over and over — one row, replaced, "failed" at a fresh time — and nothing distinguished diff --git a/internal/inventory/fortest.go b/internal/inventory/fortest.go index 9b2175eb..aac487fd 100644 --- a/internal/inventory/fortest.go +++ b/internal/inventory/fortest.go @@ -208,10 +208,3 @@ 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 -} diff --git a/internal/inventory/given.go b/internal/inventory/given.go index fd930fea..81ab99ef 100644 --- a/internal/inventory/given.go +++ b/internal/inventory/given.go @@ -79,18 +79,18 @@ type GivenReplaced struct { } // ReplaceGivenAfterStart replaces every given value on a machine whose module has now started well -// under the mesh on it (novox/hq ADR 0228), and answers what it replaced; the caller sends the +// under the mesh on it (novox/hq ADR 0228), and answers what it replaced; the caller composes the // machines each names. // -// **The signal is the machine's clean report of the declaration it was last sent** — every resource -// applied, nothing failed or refused — **when that declaration was sent after the value was given**, -// so it is the start on the given value that counts, not an older one. On an adopted machine the -// module must also be taken: until then the mesh runs nothing of it, and nothing has started under -// the mesh. Each row is claimed by clearing its mark before it is remade, so two reports arriving -// together replace a value once; a remake that fails puts the mark back, and the next good report -// tries again. -func (i *Inventory) ReplaceGivenAfterStart(ctx context.Context, node, declared string) ([]GivenReplaced, error) { - if declared == "" { +// **The signal is the machine's have of the module being healthy** (ADR 0539 §9): started names, per +// module, when the node-engine wrote a have of it that is running and healthy at its composition. +// A value counts when that is **after it was given**, so it is the start on the given value that +// counts, not an older one. On an adopted machine the module must also be taken: until then the mesh +// runs nothing of it, and nothing has started under the mesh. Each row is claimed by clearing its mark +// before it is remade, so two haves arriving together replace a value once; a remake that fails puts +// the mark back, and the next healthy have tries again. +func (i *Inventory) ReplaceGivenAfterStart(ctx context.Context, node string, started map[string]time.Time) ([]GivenReplaced, error) { + if len(started) == 0 { return nil, nil } // Only a module assigned there, from the assignments bucket. @@ -106,10 +106,10 @@ func (i *Inventory) ReplaceGivenAfterStart(ctx context.Context, node, declared s `select s.node, s.module, s.name, s.replace_after_start from module_secret s join node n on n.id = s.node - where n.name = $1 and n.sent = $2 and n.sent_at > s.replace_after_start + where n.name = $1 and s.origin = 'accepted' and s.replace_after_start is not null and (not n.adopted or exists (select 1 from taken t where t.node = s.node and t.module = s.module)) - order by s.module, s.name`, node, declared) + order by s.module, s.name`, node) if err != nil { return nil, err } @@ -125,7 +125,7 @@ func (i *Inventory) ReplaceGivenAfterStart(ctx context.Context, node, declared s rows.Close() return nil, err } - if !assignedHere[d.module] { + if !assignedHere[d.module] || !started[d.module].After(d.given) { continue } dues = append(dues, d) diff --git a/internal/inventory/given_test.go b/internal/inventory/given_test.go index 41ebb816..cf7d9ebf 100644 --- a/internal/inventory/given_test.go +++ b/internal/inventory/given_test.go @@ -45,17 +45,9 @@ func assign(t *testing.T, inv *Inventory, ctx context.Context, node, module stri } } -// sent records a declaration sent to a machine now, as a push does, and answers its digest. -func sent(t *testing.T, inv *Inventory, ctx context.Context, node, digest string) string { - t.Helper() - n, err := inv.NodeByName(ctx, node) - if err != nil { - t.Fatal(err) - } - if err := inv.RecordSentForTest(ctx, n.ID, digest); err != nil { - t.Fatal(err) - } - return digest +// healthy is a module's have, running and healthy, written a while from now: the first good start the store asks about. +func healthy(module string, from time.Duration) map[string]time.Time { + return map[string]time.Time{module: time.Now().Add(from)} } func originOf(t *testing.T, inv *Inventory, ctx context.Context, node, module, name string) string { @@ -68,22 +60,21 @@ func originOf(t *testing.T, inv *Inventory, ctx context.Context, node, module, n } // **A given value is replaced once, after the first good start on it, and never again** (ADR 0228): -// not on a report of a declaration sent before it was given, not on a report of a declaration the -// mesh has moved past, and not a second time. +// not on a start before it was given, and not a second time. func TestAGivenValueIsReplacedAfterTheFirstGoodStartAndNeverAgain(t *testing.T) { inv, ctx := aModuleWithGivenSecrets(t) - before := sent(t, inv, ctx, "consumer", "declaration-before") + before := healthy("letta", -time.Hour) marked, err := inv.AcceptGivenSecret(ctx, "consumer", "letta", "server-password", "typed-by-a-person") if err != nil || !marked { t.Fatalf("a given value for a secret the mesh may make is marked to be replaced: %v %v", marked, err) } - // The machine's report of what it was sent before the value was given is not a start on it. + // A start before the value was given is not a start on it. if got, err := inv.ReplaceGivenAfterStart(ctx, "consumer", before); err != nil || len(got) != 0 { t.Fatalf("a start before the value was given replaced it: %+v %v", got, err) } - carrying := sent(t, inv, ctx, "consumer", "declaration-carrying-it") - // A report about a declaration other than the one last sent says nothing about this one. + carrying := healthy("letta", time.Minute) + // The same early start again says nothing about this value. if got, err := inv.ReplaceGivenAfterStart(ctx, "consumer", before); err != nil || len(got) != 0 { t.Fatalf("a report of an older declaration replaced it: %+v %v", got, err) } @@ -106,7 +97,7 @@ func TestAGivenValueIsReplacedAfterTheFirstGoodStartAndNeverAgain(t *testing.T) if again, err := inv.ReplaceGivenAfterStart(ctx, "consumer", carrying); err != nil || len(again) != 0 { t.Fatalf("the same start replaced it twice: %+v %v", again, err) } - next := sent(t, inv, ctx, "consumer", "declaration-after") + next := healthy("letta", 2*time.Minute) if again, err := inv.ReplaceGivenAfterStart(ctx, "consumer", next); err != nil || len(again) != 0 { t.Fatalf("a later start replaced the mesh's own value: %+v %v", again, err) } @@ -126,7 +117,7 @@ func TestWhatIsNeverReplacedAfterAStart(t *testing.T) { if err := inv.AcceptSecretForModule(ctx, "consumer", "letta", "broker", "issued-by-the-mesh"); err != nil { t.Fatal(err) } - declared := sent(t, inv, ctx, "consumer", "declaration") + declared := healthy("letta", time.Minute) if got, err := inv.ReplaceGivenAfterStart(ctx, "consumer", declared); err != nil || len(got) != 0 { t.Fatalf("a value that stays as given was replaced: %+v %v", got, err) } @@ -157,14 +148,14 @@ func TestAGivenValueWaitsForTheTakeOnAnAdoptedMachine(t *testing.T) { if _, err := inv.AcceptGivenSecret(ctx, "consumer", "letta", "server-password", "the-found-one"); err != nil { t.Fatal(err) } - held := sent(t, inv, ctx, "consumer", "declaration-holding-it") + held := healthy("letta", time.Minute) if got, err := inv.ReplaceGivenAfterStart(ctx, "consumer", held); err != nil || len(got) != 0 { t.Fatalf("a module held as found, not yet taken, had its given value replaced: %+v %v", got, err) } if err := inv.Take(ctx, "consumer", "letta"); err != nil { t.Fatal(err) } - taken := sent(t, inv, ctx, "consumer", "declaration-taking-it") + taken := healthy("letta", time.Minute) got, err := inv.ReplaceGivenAfterStart(ctx, "consumer", taken) if err != nil || len(got) != 1 { t.Fatalf("the first good start after the take did not replace the given value: %+v %v", got, err) @@ -177,7 +168,7 @@ func TestAGivenValueWaitsForTheTakeOnAnAdoptedMachine(t *testing.T) { if err := inv.Unassign(ctx, "consumer", "letta"); err != nil { t.Fatal(err) } - gone := sent(t, inv, ctx, "consumer", "declaration-without-it") + gone := healthy("letta", time.Minute) if got, err := inv.ReplaceGivenAfterStart(ctx, "consumer", gone); err != nil || len(got) != 0 { t.Fatalf("a module no longer assigned had its given value replaced: %+v %v", got, err) } @@ -196,7 +187,7 @@ func TestADefinitionThatNowSaysOutsideKeepsTheGivenValue(t *testing.T) { if err := inv.RegisterModule(ctx, m, Source{}); err != nil { t.Fatal(err) } - declared := sent(t, inv, ctx, "consumer", "declaration") + declared := healthy("letta", time.Minute) if got, err := inv.ReplaceGivenAfterStart(ctx, "consumer", declared); err != nil || len(got) != 0 { t.Fatalf("a value the definition now says is an outside party's was replaced: %+v %v", got, err) } diff --git a/internal/inventory/migrations/0109-the-node-keeps-no-record-of-a-send.sql b/internal/inventory/migrations/0109-the-node-keeps-no-record-of-a-send.sql index 9ca20b8e..638ff825 100644 --- a/internal/inventory/migrations/0109-the-node-keeps-no-record-of-a-send.sql +++ b/internal/inventory/migrations/0109-the-node-keeps-no-record-of-a-send.sql @@ -3,9 +3,8 @@ -- 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). +-- as 0108. What stays: sent and sent_at (the digest and time of the last send), sent_bus_users and sent_provisioning, which holds the +-- provisioner's plans of each composition (inventory.RecordComposedPlans); 0110 drops the first three. 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; diff --git a/internal/inventory/migrations/0110-the-node-keeps-no-digest-of-a-send.sql b/internal/inventory/migrations/0110-the-node-keeps-no-digest-of-a-send.sql new file mode 100644 index 00000000..77d2a6d6 --- /dev/null +++ b/internal/inventory/migrations/0110-the-node-keeps-no-digest-of-a-send.sql @@ -0,0 +1,11 @@ +-- The node row keeps no digest of a send, nor when it was made (novox/hq ADR 0539 §5, §9, §17, issue 1945). +-- +-- A machine is not sent a declaration, so nothing records one: whether a machine has what is wanted of it is read from its +-- haves beside the assigner's wants, a given value is replaced when a have of its module is first running and healthy, and the +-- bus's user list is part of the bus module's composition, which the bus machine's have names or does not. This drops what +-- the last readers of a send's record read: node.sent and node.sent_at (0014), and node.sent_bus_users (0058). **It is +-- one-way**, as 0108 and 0109. node.sent_provisioning stays: it holds the provisioner's plans of each composition +-- (inventory.RecordComposedPlans). +alter table node drop column if exists sent; +alter table node drop column if exists sent_at; +alter table node drop column if exists sent_bus_users; diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index ed2ca743..b652a798 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -906,16 +906,6 @@ type BusyWait struct { // BusyWaiting is the state of a container left running because it said busy (link.BusyWaiting). const BusyWaiting = "waiting" -// Waiting says whether any container waits because it said busy. -func (d Doing) Waiting() bool { - for _, b := range d.Busy { - if b.State == BusyWaiting { - return true - } - } - return false -} - // FailedResource is one thing a node could not do. type FailedResource struct { ID string `json:"id"` @@ -1143,11 +1133,10 @@ type Reported struct { Outcome string // At is when it last reported; nil if it never has. At *time.Time - // Sent is when the current declaration went to it; nil if nothing ever did. + // Sent is not kept: nothing is sent (ADR 0539 §5). Always nil. Sent *time.Time - // Current is whether the last report names the declaration last sent — the machine has - // acted on the current words, not merely spoken after they were written. False also covers - // a machine that has not said which, which is every host from before reports carried it. + // Current is whether the machine has everything the assigner wants of it, read from its haves beside the wants + // (cmd/mesh-controller lastReports); false here, as this reads neither. Current bool // Declared is the digest of the declaration the last report was about, and ReportedSequence that // declaration's sequence as the report claimed it (zero from an engine that claims none): what a @@ -1158,67 +1147,12 @@ type Reported struct { Busy []BusyWait } -// SentDeclaration is what one send carried to a machine, as a gate keeps it: the declaration's digest and -// its sequence (novox/hq issue 352). A report about this declaration, or about one sequenced after it, is a -// report on what the gate sent — whatever the machine was sent since. -// -// At is when the controller sent it to that machine: the node's own send time, which is where that machine's -// `apply` duration starts. A walk measures its rest from it, not from its own record of the send, made only after -// every machine of the rest was sent (novox/hq issue 582). Zero for a send kept before it was recorded. -type SentDeclaration struct { - Digest string `json:"digest"` - Sequence int64 `json:"sequence,omitempty"` - At time.Time `json:"at,omitzero"` - // Provisioned is, for each module the send handed to the provisioner, the digest of the templates its plan - // named (novox/hq ADR 0349 §4, ADR 0362): the build the machine's provisioner client reports `prov.applied` on. - // The machine has applied the send of such a module only once both it and the client reported it. - Provisioned map[string]string `json:"provisioned,omitempty"` -} - -// ReportsOn says a report is about this send: the declaration itself; one the same machine was sequenced -// after it; or the declaration the machine was sent last (Current), which is this send or a later one — -// sends to a machine are made one after another. A send kept without a sequence is matched by its digest -// and by the last send alone. -func (s SentDeclaration) ReportsOn(r Reported) bool { - if r.Current || (s.Digest != "" && r.Declared == s.Digest) { - return true - } - return s.Sequence > 0 && r.ReportedSequence >= s.Sequence -} - -// SentTo is the declaration a machine was last sent, by name: its digest and sequence, and false when it -// was never sent one. -func (i *Inventory) SentTo(ctx context.Context, name string) (SentDeclaration, bool, error) { - var digest *string - var seq *int64 - var at *time.Time - var plans []byte - err := i.store.Pool().QueryRow(ctx, `select sent, sequence, sent_at, sent_provisioning from node where name = $1`, - name).Scan(&digest, &seq, &at, &plans) - if errors.Is(err, pgx.ErrNoRows) { - return SentDeclaration{}, false, fmt.Errorf("%w: %s", ErrNoSuchNode, name) - } - if err != nil || digest == nil || *digest == "" { - return SentDeclaration{}, false, err - } - s := SentDeclaration{Digest: *digest} - if seq != nil { - s.Sequence = *seq - } - if at != nil { - s.At = at.UTC() - } - if s.Provisioned, err = provisionedBy(plans); err != nil { - return SentDeclaration{}, false, fmt.Errorf("what %s was sent of the provisioner's plans: %w", name, err) - } - return s, true, nil -} - -// LastReports is every machine's last report beside when it was last sent a declaration. +// LastReports is every machine's last report: when, what it did, which declaration it was about and what it holds busy. Sent +// and Current are not kept here: nothing is sent (ADR 0539 §5), and whether a machine has what is wanted of it is read from +// its haves (cmd/mesh-controller lastReports). func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) { rows, err := i.store.Pool().Query(ctx, - `select n.name, coalesce(r.outcome, ''), r.at, n.sent_at, - coalesce(r.declared is not null and r.declared <> '' and r.declared = n.sent, false), + `select n.name, coalesce(r.outcome, ''), r.at, coalesce(r.declared, ''), coalesce(r.reported_sequence, 0), coalesce(r.busy, '[]') from node n left join node_report r on r.node = n.id order by n.name`) @@ -1230,8 +1164,7 @@ func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) { for rows.Next() { var r Reported var busy []byte - if err := rows.Scan(&r.Node, &r.Outcome, &r.At, &r.Sent, &r.Current, &r.Declared, &r.ReportedSequence, - &busy); err != nil { + if err := rows.Scan(&r.Node, &r.Outcome, &r.At, &r.Declared, &r.ReportedSequence, &busy); err != nil { return nil, err } if err := json.Unmarshal(busy, &r.Busy); err != nil { @@ -1242,94 +1175,6 @@ func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) { return out, rows.Err() } -// 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 -// read from: every change a push carries happened after the last push, and an assignment is the one -// that says when. -func (i *Inventory) AwaitingSince(ctx context.Context, name string) (time.Time, error) { - return i.chainAwaitingSince(ctx, name) -} - -// 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. -func (i *Inventory) RecordSentBusUsers(ctx context.Context, name, digest string) error { - _, err := i.store.Pool().Exec(ctx, - `update node set sent_bus_users = $2 where name = $1`, name, digest) - return err -} - -// SentBusUsers is the digest of the bus's user list a machine was last sent, empty for none. -func (i *Inventory) SentBusUsers(ctx context.Context, name string) (string, error) { - var sent string - err := i.store.Pool().QueryRow(ctx, - `select sent_bus_users from node where name = $1`, name).Scan(&sent) - if errors.Is(err, pgx.ErrNoRows) { - return "", nil - } - return sent, err -} - -// Outstanding is the digest of the declaration a machine was last sent, by its name, and empty -// for one that has never been sent anything. -// -// **By name rather than by id**, because the caller is the serving loop and what a node puts in a -// report is its name. Asked of one machine rather than read from Waiting's sweep, because it is -// asked per message: a report names the declaration it is about, and a report about one the mesh -// has already moved past is not acted on (design 25 §3). -func (i *Inventory) Outstanding(ctx context.Context, name string) (string, error) { - var sent string - err := i.store.Pool().QueryRow(ctx, - `select coalesce(sent, '') from node where name = $1`, name).Scan(&sent) - if errors.Is(err, pgx.ErrNoRows) { - // Not an error worth carrying up: a report from a machine the mesh has no record of has - // nothing to be stale against, and whatever is wrong with it is the listener's to say. - return "", nil - } - return sent, err -} - -// Waiting is every machine whose declaration has changed since it was last sent one. -// -// The caller works out what each machine should be now, because only it can — resolution is the -// control plane's and this context holds records. What is answered here is the comparison. -// -// **A machine that has never been sent anything is waiting**, and says so differently: it is not -// out of date, it has never been told, and the remedy is the same push while the situation is not -// the same at all. -func (i *Inventory) Waiting(ctx context.Context, would map[string]string) ([]Machine, error) { - rows, err := i.store.Pool().Query(ctx, - `select name, coalesce(sent, ''), sent_at from node order by name`) - if err != nil { - return nil, err - } - defer rows.Close() - - var out []Machine - for rows.Next() { - var m Machine - var at *time.Time - if err := rows.Scan(&m.Node, &m.Sent, &at); err != nil { - return nil, err - } - m.SentAt = at - wanted, known := would[m.Node] - if !known { - // Nothing was computed for it — it resolves to nothing, or the caller did not ask. - // Silence rather than a guess: saying "waiting" about a machine nobody worked out - // would be inventing a comparison. - continue - } - if m.Sent == wanted { - continue - } - m.Never = m.Sent == "" - out = append(out, m) - } - return out, rows.Err() -} - // Machine is one machine that has not been sent what it should be. type Machine struct { Node string diff --git a/internal/inventory/plans_leave_the_store_test.go b/internal/inventory/plans_leave_the_store_test.go index 971e382a..37a43cc4 100644 --- a/internal/inventory/plans_leave_the_store_test.go +++ b/internal/inventory/plans_leave_the_store_test.go @@ -47,7 +47,7 @@ func TestMigration0108DropsTheTablesOfThePlansMachineryAndKeepsWhatTheController } // **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. +// the epoch, the summary and the generation. The provisioner's plans stay (0110 drops the rest). func TestMigration0109DropsWhatASendCarriedFromTheNodeRow(t *testing.T) { inv, migrate := ForTestBefore(t, 109) columns := func() map[string]bool { @@ -72,9 +72,40 @@ func TestMigration0109DropsWhatASendCarriedFromTheNodeRow(t *testing.T) { t.Errorf("node.%s is still in the store", name) } } - for _, stays := range []string{"sent", "sent_at", "sent_bus_users", "sent_provisioning"} { + for _, stays := range []string{"sent_provisioning"} { if !after[stays] { t.Errorf("node.%s is gone", stays) } } } + +// **Migration 0110 drops the digest and time of a send, and the bus users' digest, from the node row** (ADR 0539 §17, issue +// 1945): the haves answer what a machine has. The provisioner's plans stay. +func TestMigration0110DropsTheDigestOfASendFromTheNodeRow(t *testing.T) { + inv, migrate := ForTestBefore(t, 110) + 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", "sent_at", "sent_bus_users"} + 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) + } + } + if !after["sent_provisioning"] { + t.Errorf("node.sent_provisioning is gone") + } +} diff --git a/internal/link/heard_order_integration_test.go b/internal/link/heard_order_integration_test.go index 5a5b5627..b57ed561 100644 --- a/internal/link/heard_order_integration_test.go +++ b/internal/link/heard_order_integration_test.go @@ -72,9 +72,6 @@ func TestAnOlderReportOfTheSameDeclarationIsRefused(t *testing.T) { func TestAnOrderedAccountIsNotJudgedByTheDigestSent(t *testing.T) { inv, heard, id := anOrderedMachine(t) ctx := context.Background() - 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 { t.Fatal(err) } @@ -93,9 +90,6 @@ func TestAnUnorderedAccountClearsTheOrderAndTheEpoch(t *testing.T) { if _, err := heard.Heard(ctx, ordered("d12", 57, 12, 41, "a")); err != nil { t.Fatal(err) } - 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 { t.Fatal(err) } diff --git a/internal/link/heard_test.go b/internal/link/heard_test.go index b4073c58..3223374d 100644 --- a/internal/link/heard_test.go +++ b/internal/link/heard_test.go @@ -102,9 +102,6 @@ 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.RecordSentForTest(ctx, node.ID, digest); err != nil { - t.Fatal(err) - } if _, err := (link.Enrolment{Inventory: inv}).Heard(ctx, link.Report{ Node: "anchor", Applied: []string{"a", "b"}, Declared: digest, Carried: []int{5432}, }); err != nil { @@ -270,13 +267,10 @@ func TestWhatFiltersAMachineIsKeptFromItsReport(t *testing.T) { func TestAnAccountOfAnOlderDeclarationDoesNotReplaceTheNewer(t *testing.T) { inv := inventory.ForTest(t) ctx := context.Background() - node, err := inv.AddNode(ctx, "home-server") + _, err := inv.AddNode(ctx, "home-server") if err != nil { t.Fatal(err) } - if err := inv.RecordSentForTest(ctx, node.ID, "d2"); err != nil { - t.Fatal(err) - } heard := link.Enrolment{Inventory: inv} if _, err := heard.Heard(ctx, link.Report{Node: "home-server", Declared: "d2", Applied: []string{"a", "b"}, Outward: []string{"eth0"}}); err != nil { @@ -304,7 +298,7 @@ func TestAnAccountOfAnOlderDeclarationDoesNotReplaceTheNewer(t *testing.T) { func TestAnAccountOfTheSentDeclarationReplacesTheKeptOne(t *testing.T) { inv := inventory.ForTest(t) ctx := context.Background() - node, err := inv.AddNode(ctx, "home-server") + _, err := inv.AddNode(ctx, "home-server") if err != nil { t.Fatal(err) } @@ -312,9 +306,6 @@ 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.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 { t.Fatal(err) }