diff --git a/cmd/mesh-controller/data_integration_test.go b/cmd/mesh-controller/data_integration_test.go index 68247082..d121a20b 100644 --- a/cmd/mesh-controller/data_integration_test.go +++ b/cmd/mesh-controller/data_integration_test.go @@ -80,17 +80,18 @@ func TestNatsUnassigningIrreplaceableDataRetiresItAndAnEmptyReplacementIsUrgent( if _, err := assign(ctx, open, "laptop", "house"); err != nil { t.Fatal(err) } - // Sent, and applied: a holder is asked only once it has been (novox/hq issue 275). - for _, node := range []string{"laptop", "anchor"} { - pushedAndApplied(t, open, node) - } - conn := onATestBus(t) js, err := broker.Dial(testbus.URL(t)) if err != nil { t.Fatal(err) } t.Cleanup(js.Close) + // Applied, and running: a holder is asked only once its module's have says it runs (novox/hq issue 275, ADR 0539 §9). + servingBus.Store(js) + t.Cleanup(func() { servingBus.Store(nil) }) + for _, run := range [][2]string{{"laptop", "keeper"}, {"laptop", "house"}, {"anchor", "keeper"}} { + putRunningHave(t, js, run[0], run[1]) + } laptop, anchor := &fakeHolder{}, &fakeHolder{} laptop.serve(t, conn, "laptop") anchor.serve(t, conn, "anchor") diff --git a/cmd/mesh-controller/haves_test.go b/cmd/mesh-controller/haves_test.go index d4322eb9..19893d7e 100644 --- a/cmd/mesh-controller/haves_test.go +++ b/cmd/mesh-controller/haves_test.go @@ -133,3 +133,41 @@ func TestWithTheBusOutOfReachNoNodeIsSaidCurrentOrNot(t *testing.T) { } } } + +// **A holder is asked once its module's have says it runs** (novox/hq ADR 0539 §9, issue 275): assigned and not applied yet, it +// is not there to answer; a machine that has said nothing holds nothing the mesh put there. +func TestAHolderIsAskedOnlyOnceItsModuleRuns(t *testing.T) { + have := func(phase sdkdelivery.Phase) sdkdelivery.Have { + return sdkdelivery.Have{Phase: phase, Health: sdkdelivery.HealthOK} + } + key := func(node, module string) string { + k, err := sdkdelivery.HaveKey(node, module) + if err != nil { + t.Fatal(err) + } + return k + } + nodeKey, err := sdkdelivery.HaveNodeKey("anchor") + if err != nil { + t.Fatal(err) + } + got := runningModules(map[string]sdkdelivery.Have{ + key("anchor", "keeper"): have(sdkdelivery.PhaseRunning), + key("anchor", "house"): have(sdkdelivery.PhaseApplying), + key("laptop", "house"): have(sdkdelivery.PhaseFailed), + nodeKey: have(sdkdelivery.PhaseRunning), + }) + now := time.Now() + if !got["anchor"].settled("keeper", now) { + t.Error("a running module's holder is not asked") + } + if got["anchor"].settled("house", now) { + t.Error("a module still applying is asked") + } + if got["laptop"].settled("house", now) || got["ace"].settled("keeper", now) { + t.Error("a failed module, or a machine that has said nothing, is asked") + } + if got["anchor"].settled("node", now) { + t.Error("the node's own have is taken for a module") + } +} diff --git a/cmd/mesh-controller/helpers_kept_integration_test.go b/cmd/mesh-controller/helpers_kept_integration_test.go index 92a3fb1e..a398f2b5 100644 --- a/cmd/mesh-controller/helpers_kept_integration_test.go +++ b/cmd/mesh-controller/helpers_kept_integration_test.go @@ -4,6 +4,9 @@ package main import ( "context" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" + "strings" "testing" "time" @@ -78,3 +81,29 @@ func (k planKeysOf) Build(_ context.Context, module, fp string) (builderkeys.Key } return builderkeys.Key{Module: module, Fingerprint: fp}, nil } + +// putRunningHave writes a node-engine's have of a module: running and healthy, at a fingerprint and composition of its own. +func putRunningHave(t *testing.T, js *broker.JetStream, node, module string) { + t.Helper() + ctx := t.Context() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + kv, err := api.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket}) + if err != nil { + t.Fatal(err) + } + key, err := sdkdelivery.HaveKey(node, module) + if err != nil { + t.Fatal(err) + } + value, err := sdkdelivery.EncodeHave(sdkdelivery.Have{FP: sdkdelivery.SourcePrefix2 + strings.Repeat("a", 64), + Composed: "sha256:" + strings.Repeat("b", 64), Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK}) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, value); err != nil { + t.Fatal(err) + } +} diff --git a/cmd/mesh-controller/helpers_kept_test.go b/cmd/mesh-controller/helpers_kept_test.go index 90a126d4..f196680b 100644 --- a/cmd/mesh-controller/helpers_kept_test.go +++ b/cmd/mesh-controller/helpers_kept_test.go @@ -68,17 +68,3 @@ func pushedBy(t *testing.T, open *stores, node string) string { } 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/probes.go b/cmd/mesh-controller/probes.go index 50f11afa..2eb138f0 100644 --- a/cmd/mesh-controller/probes.go +++ b/cmd/mesh-controller/probes.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "net" "regexp" "slices" @@ -540,48 +541,53 @@ func holdingNeedsAnAnswer(s catalogue.Seat) bool { return false } -// reportGrace is how long a machine is given, after it was sent a declaration, to apply it and report -// before what the declaration carries is expected to answer (novox/hq issue 275). -var reportGrace = 10 * time.Minute - -// lastSend is what a machine was last sent, as far as the self-check needs it: which modules the send -// carried, when, and whether the machine has reported acting on it. +// lastSend is what a machine runs, as far as the self-check needs it: the modules whose have on it is running (novox/hq ADR +// 0539 §9). It keeps its old name, which said what was sent. type lastSend struct { - // carried is the modules its last send carried; nil when that was not kept (a send from before - // it was, or one by hand), and then every module is taken as carried — as before this existed. - carried map[string]string - // sent is when its current declaration went to it; nil if nothing ever did. - sent *time.Time - // current says its last report names the declaration last sent. - current bool + // running are the modules the node-engine says it runs there. + running map[string]bool } -// settled says whether a module's holder on this machine can be asked to answer now: the machine was -// sent a declaration carrying it, and has reported acting on that declaration or had reportGrace to. -// A machine never sent anything holds nothing the mesh put there. -func (d lastSend) settled(module string, now time.Time) bool { - if d.sent == nil { - return false - } - if d.carried != nil { - if _, in := d.carried[module]; !in { - return false +// settled says whether a module's holder on this machine can be asked to answer now: the node-engine's have of the module is +// running. A module assigned and not applied yet is not there to answer, and a machine that has said nothing holds nothing the +// mesh put there. now is unused: the have says it, with no grace to wait out. +func (d lastSend) settled(module string, _ time.Time) bool { return d.running[module] } + +// runningModules reads from the haves which modules run on which node. Pure. +func runningModules(haves map[string]sdkdelivery.Have) map[string]lastSend { + out := map[string]lastSend{} + for key, h := range haves { + node, module, isNode, err := sdkdelivery.ParseHaveKey(key) + if err != nil || isNode || h.Phase != sdkdelivery.PhaseRunning { + continue } + d := out[node] + if d.running == nil { + d.running = map[string]bool{} + } + d.running[module] = true + out[node] = d } - return d.current || now.Sub(*d.sent) > reportGrace + return out } -// readDeliveries is every machine's last send, by name, from the records a send and a report keep. +// readDeliveries is every machine's running modules, by name, from the node-engines' haves. A process with no bus reads none: +// nothing is then taken as running, never everything. func readDeliveries(ctx context.Context, inv *inventory.Inventory) (map[string]lastSend, error) { - reports, err := lastReports(ctx, inv) + bus, err := aBus() + if err != nil { + return map[string]lastSend{}, nil // empty-on-error: no bus to read the haves from, so no module is taken as running + } + defer bus.Close() + js, err := jetstream.New(bus.Conn()) if err != nil { return nil, err } - out := map[string]lastSend{} - for _, r := range reports { - out[r.Node] = lastSend{sent: r.Sent, current: r.Current} + _, haves, err := readWantsAndHaves(ctx, js) + if err != nil { + return nil, err } - return out, nil + return runningModules(haves), nil } // heardMachines is every machine within its heartbeat's bound, as the watchdogs last saw.