From 0e85230769c3adbe13b562d38af1f936ba63bdc6 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 24 Oct 2026 18:09:38 +0200 Subject: [PATCH] A node with no want that has reported is current; with the bus out of reach it is not known, and the flip says so; no 'push' hint is left in adoption (release 6, issue 1945 B8) --- cmd/mesh-controller/adopting_test.go | 71 +++++++++++++++++++++++- cmd/mesh-controller/adoption.go | 38 +++++++------ cmd/mesh-controller/haves.go | 28 +++++++--- cmd/mesh-controller/haves_test.go | 33 +++++++++++ cmd/mesh-controller/take_preview_test.go | 2 +- internal/inventory/nodes.go | 3 + 6 files changed, 150 insertions(+), 25 deletions(-) diff --git a/cmd/mesh-controller/adopting_test.go b/cmd/mesh-controller/adopting_test.go index 67a5b104..7d2009e8 100644 --- a/cmd/mesh-controller/adopting_test.go +++ b/cmd/mesh-controller/adopting_test.go @@ -6,6 +6,10 @@ import ( "context" "encoding/json" "errors" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/testbus" "net/http" "reflect" "strings" @@ -50,6 +54,7 @@ func anAdoptedAnchor(t *testing.T) (*stores, *[]string) { t.Fatal(err) } } + onTheBusWithItsHaves(t, open) sent := &[]string{} saved := sendNodes sendNodes = func(_ context.Context, _ *stores, names []string) error { @@ -171,7 +176,7 @@ func TestTakingNamesWhatItReplaces(t *testing.T) { t.Fatal(err) } if !strings.Contains(said, "container hello-web (hello-web.server)") || - !strings.Contains(said, "push anchor") { + !strings.Contains(said, "compositions of anchor are written") { t.Fatalf("taking did not say what it replaces and what to run:\n%s", said) } } @@ -558,3 +563,67 @@ func TestTheFlipIsRefusedOnAnAccountNamingNothingReachable(t *testing.T) { t.Fatal("a refused flip changed something") } } + +// onTheBusWithItsHaves gives the controller a bus that holds, for the anchor, a want and a have of its module that agree: the node +// has everything wanted of it, as read from the haves (novox/hq ADR 0539 §9), so a flip is refused for what it is about and +// not because the bus could not be read. +func onTheBusWithItsHaves(t *testing.T, open *stores) { + t.Helper() + ctx := t.Context() + js, err := broker.Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(js.Close) + servingBus.Store(js) + t.Cleanup(func() { servingBus.Store(nil) }) + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + fp := sdkdelivery.SourcePrefix2 + strings.Repeat("a", 64) + composed := "sha256:" + strings.Repeat("b", 64) + put := func(bucket, key string, value []byte) { + kv, err := api.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket}) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, value); err != nil { + t.Fatal(err) + } + } + wantKey, err := sdkdelivery.WantKey("anchor", "hello-web") + if err != nil { + t.Fatal(err) + } + haveKey, err := sdkdelivery.HaveKey("anchor", "hello-web") + if err != nil { + t.Fatal(err) + } + want, err := sdkdelivery.EncodeWant(sdkdelivery.Want{FP: fp, Composed: composed}) + if err != nil { + t.Fatal(err) + } + have, err := sdkdelivery.EncodeHave(sdkdelivery.Have{FP: fp, Composed: composed, Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK}) + if err != nil { + t.Fatal(err) + } + put(sdkdelivery.AssignerWantsBucket, wantKey, want) + put(sdkdelivery.HaveBucket, haveKey, have) +} + +// **When the bus cannot be read, whether the machine runs what is wanted is not known, and the flip says exactly that** (issue +// 1945): not "push", and not "has not reported". +func TestTheFlipIsRefusedWithTheBusUnreachableSayingItCannotTell(t *testing.T) { + open, sent := anAdoptedAnchor(t) + reportsHolding(t, open, heldContainer, heldFile) + servingBus.Store(nil) // this process has no bus now, and no address to dial + _, err := converge(t.Context(), open, "anchor", true, "", "") + if err == nil || !strings.Contains(err.Error(), "cannot tell whether anchor runs what is wanted: the bus is not reachable") || + strings.Contains(err.Error(), "push") { + t.Fatalf("the flip with the bus out of reach: %v", err) + } + if len(*sent) != 0 { + t.Fatal("a refused flip sent something") + } +} diff --git a/cmd/mesh-controller/adoption.go b/cmd/mesh-controller/adoption.go index 0dd4f50f..b61a124d 100644 --- a/cmd/mesh-controller/adoption.go +++ b/cmd/mesh-controller/adoption.go @@ -276,8 +276,8 @@ func take(ctx context.Context, open *stores, node, module string, opts takeOptio // is still the machine: the same two refusals the flip makes. if age := time.Since(c.reported.At); age > reportFreshFor { return preview, fmt.Errorf("%s last said what it holds %s ago, and a take acts only on "+ - "an account newer than %s: run `push %s --wait 2m`, then preview again", - node, age.Round(time.Second), reportFreshFor, node) + "an account newer than %s: wait for its next report, then preview again", + node, age.Round(time.Second), reportFreshFor) } if opts.Digest == "" { return preview, fmt.Errorf("taking %s on %s acts on the preview you saw: name its digest, "+ @@ -294,9 +294,12 @@ func take(ctx context.Context, open *stores, node, module string, opts takeOptio } said := fmt.Sprintf("%s is taken on %s", module, node) if holds { - said += "; the next push replaces what the node found and holds for it:\n" + preview + said += "; its next composition replaces what the node found and holds for it:\n" + preview } - return said + fmt.Sprintf("\n run `push %s` to cut it over", node), nil + if err := sendNodes(ctx, open, []string{node}); err != nil { + return said + "\n and its compositions could not be written yet: the next act that composes them carries it", err + } + return said + fmt.Sprintf("\n the compositions of %s are written; its wants then cut it over", node), nil } // comparison is everything a take puts beside what the module declares: the machine's account of @@ -433,7 +436,7 @@ func comparisonOf(module string, c comparison, opts takeOptions) (preview string fmt.Fprintf(&b, " settings from %s: %s\n", layer.From, strings.Join(keys, ", ")) } if c.settingsRefused != "" { - fmt.Fprintf(&b, " SETTINGS DO NOT COMPOSE with the module's definition, so the push leaves it out: %s\n", c.settingsRefused) + fmt.Fprintf(&b, " SETTINGS DO NOT COMPOSE with the module's definition, so the composition leaves it out: %s\n", c.settingsRefused) } preview = strings.TrimRight(b.String(), "\n") sum := sha256.Sum256([]byte(preview)) @@ -619,20 +622,23 @@ func converge(ctx context.Context, open *stores, node string, yes bool, digest s if err != nil { return "", err } - current := false + current, unknown := false, false for _, r := range reports { if r.Node == node { - current = r.Current + current, unknown = r.Current, r.CurrentUnknown } } + if unknown { + return "", fmt.Errorf("cannot tell whether %s runs what is wanted: the bus is not reachable", node) + } reported, err := inv.AdoptionOf(ctx, node) if err != nil { return "", err } if !current || reported.At.IsZero() { - return "", fmt.Errorf("%s has not reported on the declaration it was last sent, so what it "+ - "says is reachable may not be the machine as it is: run `push %s --wait 2m` and "+ - "converge once it has applied", node, node) + return "", fmt.Errorf("%s does not run everything that is wanted of it yet, or has not reported, so what it "+ + "says is reachable may not be the machine as it is: converge once its haves match its wants and it has "+ + "reported", node) } assignedWhenPreviewed, err := inv.Assigned(ctx, node) @@ -728,13 +734,13 @@ func converge(ctx context.Context, open *stores, node string, yes bool, digest s if yes && countReachable(reported) == 0 { return preview, fmt.Errorf("%s says nothing is reachable on it, which no machine that is "+ "up ever is: its account looks partial — whatever reads what is listening, or what "+ - "the container runtime publishes, did not answer. Fix that on the machine and run "+ - "`push %s --wait 2m`, then preview again", node, node) + "the container runtime publishes, did not answer. Fix that on the machine, wait for its next report, "+ + "then preview again", node) } if age := time.Since(reported.At); age > reportFreshFor { return preview, fmt.Errorf("%s last said what is reachable on it %s ago, and the flip acts "+ - "only on an account newer than %s: wait for its next report, or run `push %s --wait 2m`, "+ - "then preview again", node, age.Round(time.Second), reportFreshFor, node) + "only on an account newer than %s: wait for its next report, then preview again", node, + age.Round(time.Second), reportFreshFor) } if digest == "" { return preview, fmt.Errorf("converging %s acts on the preview you saw: name its digest, "+ @@ -777,7 +783,7 @@ func converge(ctx context.Context, open *stores, node string, yes bool, digest s said += "; took " + strings.Join(took, ", ") } if err := sendNodes(ctx, open, []string{node}); err != nil { - return said + "\n and it could not be sent: run `push " + node + "`", err + return said + "\n and its compositions could not be written yet: the next act that composes them carries it", err } return said + "\n sent: the host loads the mesh's filter and disables the firewall it found", nil } @@ -1026,7 +1032,7 @@ func adopt(ctx context.Context, open *stores, node string) (string, error) { } said := fmt.Sprintf("%s is adopted; what was taken on it stays taken", node) if err := sendNodes(ctx, open, []string{node}); err != nil { - return said + "\n and it could not be sent: run `push " + node + "`", err + return said + "\n and its compositions could not be written yet: the next act that composes them carries it", err } return said + "\n sent: the host unloads the mesh's filter and enables the firewall it found", nil } diff --git a/cmd/mesh-controller/haves.go b/cmd/mesh-controller/haves.go index 5d1f74d0..1669d39e 100644 --- a/cmd/mesh-controller/haves.go +++ b/cmd/mesh-controller/haves.go @@ -88,8 +88,25 @@ func readWantsAndHaves(ctx context.Context, js jetstream.JetStream) (map[string] return wants, haves, nil } +// applyCurrent sets Current of each report from the nodes' states: a node some want names is current as currentNodes says; a +// node nothing is wanted on that has reported has nothing to apply and is current. Pure. +func applyCurrent(reports []inventory.Reported, current map[string]bool) { + for i := range reports { + c, wanted := current[reports[i].Node] + reports[i].Current = c || !wanted && reports[i].At != nil + reports[i].CurrentUnknown = false + } +} + +// currentUnknown marks every report's Current as not known: the haves could not be read. Pure. +func currentUnknown(reports []inventory.Reported) { + for i := range reports { + reports[i].Current, reports[i].CurrentUnknown = false, true + } +} + // 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. +// has everything the assigner wants of it. A process that cannot reach the bus cannot tell, and says so (CurrentUnknown). func lastReports(ctx context.Context, inv *inventory.Inventory) ([]inventory.Reported, error) { reports, err := inv.LastReports(ctx) if err != nil { @@ -97,7 +114,8 @@ func lastReports(ctx context.Context, inv *inventory.Inventory) ([]inventory.Rep } 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" + currentUnknown(reports) + return reports, nil // empty-on-error: the bus is not reachable, so whether a machine is current is marked not known } defer bus.Close() js, err := jetstream.New(bus.Conn()) @@ -108,10 +126,6 @@ func lastReports(ctx context.Context, inv *inventory.Inventory) ([]inventory.Rep 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 - } + applyCurrent(reports, currentNodes(wants, haves)) return reports, nil } diff --git a/cmd/mesh-controller/haves_test.go b/cmd/mesh-controller/haves_test.go index 9d9e8ac8..d4322eb9 100644 --- a/cmd/mesh-controller/haves_test.go +++ b/cmd/mesh-controller/haves_test.go @@ -1,7 +1,10 @@ package main import ( + "time" + "errors" + "github.com/novox/mesh-controller/internal/inventory" "strings" "testing" @@ -100,3 +103,33 @@ func TestTheBusMachineIsBehindUntilItsHaveNamesTheCompositionCarryingTheUserList } } } + +// **A node nothing is wanted on that has reported is current; one that has not is not** (issue 1945): it has nothing to apply. +func TestANodeWithNoWantThatHasReportedIsCurrent(t *testing.T) { + at := time.Now() + reports := []inventory.Reported{{Node: "anchor", At: &at}, {Node: "laptop"}, {Node: "ace", At: &at}} + applyCurrent(reports, map[string]bool{"ace": false}) + got := map[string]bool{} + for _, r := range reports { + got[r.Node] = r.Current + if r.CurrentUnknown { + t.Errorf("%s: current is unknown with the haves read", r.Node) + } + } + if !got["anchor"] || got["laptop"] || got["ace"] { + t.Errorf("current by node: %v; want anchor (reported, nothing wanted) true, laptop (never reported) false, ace (behind) false", got) + } +} + +// **With the bus out of reach, whether a node is current is not known** (issue 1945): every report says so, and none says "not +// current". +func TestWithTheBusOutOfReachNoNodeIsSaidCurrentOrNot(t *testing.T) { + at := time.Now() + reports := []inventory.Reported{{Node: "anchor", At: &at, Current: true}, {Node: "laptop"}} + currentUnknown(reports) + for _, r := range reports { + if r.Current || !r.CurrentUnknown { + t.Errorf("%s: current %v, unknown %v; want not known", r.Node, r.Current, r.CurrentUnknown) + } + } +} diff --git a/cmd/mesh-controller/take_preview_test.go b/cmd/mesh-controller/take_preview_test.go index c7aac893..4bc24067 100644 --- a/cmd/mesh-controller/take_preview_test.go +++ b/cmd/mesh-controller/take_preview_test.go @@ -137,7 +137,7 @@ func TestTheKeptNetworkAndTheSettingsAreInThePreview(t *testing.T) { } c.settingsRefused = "forge: ports is a { port: machine-port } map" preview, _, _ = comparisonOf("forge", c, takeOptions{Downgrade: true, Replace: map[string]bool{"*": true}}) - if !strings.Contains(preview, "SETTINGS DO NOT COMPOSE with the module's definition, so the push leaves it out: forge: ports") { + if !strings.Contains(preview, "SETTINGS DO NOT COMPOSE with the module's definition, so the composition leaves it out: forge: ports") { t.Errorf("settings that cannot compose are not said:\n%s", preview) } } diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index b652a798..c0fb5d47 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -1138,6 +1138,9 @@ type Reported struct { // 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 + // CurrentUnknown says Current could not be read (the bus holding the haves was not reachable): it is not "not current", + // it is not known. + CurrentUnknown 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 // gate holds against the send it made, rather than against the send made last (novox/hq issue 352).