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

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)

Browse Source
This commit is contained in:
jschoubben
2026-10-24 18:09:38 +02:00
parent ad0aabc2a9
commit 0e85230769
6 changed files with 150 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/adopting_test.go
+70 -1
View File
@@ -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")
}
}
cmd/mesh-controller/adoption.go
+22 -16
View File
@@ -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
}
cmd/mesh-controller/haves.go
+21 -7
View File
@@ -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
}
cmd/mesh-controller/haves_test.go
+33
View File
@@ -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)
}
}
}
cmd/mesh-controller/take_preview_test.go
+1 -1
View File
@@ -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)
}
}
internal/inventory/nodes.go
+3
View File
@@ -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).