diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index d71c8c5..1f0b801 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -544,7 +544,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess session.NoteCandidates(ctx, answers) }) go shadow.Keep(linked, link.Backoff) - if composer, why := selfComposer(linked, session, node, say, hooks); composer != nil { + composer, why := selfComposer(linked, session, node, say, hooks) + if linked.Err() != nil { + // The link ended before the node knew whose compositions to apply: it chose neither, and waits for the next. + stopRun() + cancel() + continue + } + if composer != nil { env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy session.NoteComposes(linked, delivery.ComposesOwn, why) // And once the catalogue no longer places it, the agent begins again on the controller's compositions: the @@ -854,18 +861,19 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f say("this node applies the controller's compositions: " + err.Error()) return nil, err.Error() } - got, err := session.KVHistory(ctx, delivery.PlacementBucket, key) - if err != nil || len(got) == 0 || got[len(got)-1].Deleted { + env := composerEnv(session, node, say, hooks) + placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say) + if err != nil { + return nil, "its placement could not be read before the link ended: " + err.Error() + } + if !placed { why := "the catalogue does not place it" - if err != nil { - why = "its placement cannot be read: " + err.Error() - } say("this node applies the controller's compositions: " + why) return nil, why } why := "the catalogue places it (" + key + ")" say("this node composes its own modules: " + why) - return selfcompose.New(composerEnv(session, node, say, hooks)), why + return selfcompose.New(env), why } // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer @@ -922,6 +930,9 @@ func endWhenUnplaced(ctx context.Context, session *link.Session, node string, st placed = true } } + if ctx.Err() == nil { + say("the watch of " + key + " ended while the link stays: whether the catalogue still places this node is followed again at the node agent's next start") + } } // endWhenPlaced calls stop once this node's placement key appears in the indexer's placement (a value after the watch @@ -944,4 +955,7 @@ func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop return } } + if ctx.Err() == nil { + say("the watch of " + key + " ended while the link stays: whether the catalogue places this node is followed again at the node agent's next start") + } } diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index d3a352f..7d152c1 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -406,8 +406,8 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { h8back(t, b, placement, candidates, hkey) } -// h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is answered as -// returning to the controller's compositions, and the placement key deleted returns the agent to them without a +// h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is compared with +// the controller's compositions at its wants (here none: refused, naming web), and the placement key deleted returns the agent to them without a // restart, said in the journal and the node's have, keeping what runs. func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey string) { t.Helper() @@ -431,10 +431,12 @@ func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey st if _, err := candidates.Put(ckey, craw); err != nil { t.Fatal(err) } - within(t, 10*time.Second, "the node did not answer a candidate deleting its placement as returning", func() bool { + // The controller holds no composition at web's want (the node composed its own): returning would leave web with + // nothing to apply, and the answer says so rather than "same" (review of the shadow, round 1). + within(t, 10*time.Second, "the node did not refuse returning to compositions the controller does not hold", func() bool { a := node().Candidates["revert/x"] - return a.Head == "h2" && a.Returns == delivery.ReturnsToController && a.Verdict == delivery.CandidateSame && - len(a.Removed) == 0 + return a.Head == "h2" && a.Returns == "" && a.Verdict == delivery.CandidateRefused && + strings.Contains(a.Refused, "web: the controller holds no composition") }) ran := b.have("web").Composed pkey, _ := delivery.PlacementKey(itNode) diff --git a/go.mod b/go.mod index 678e26f..412919b 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index fd262e6..17fbe1e 100644 --- a/go.sum +++ b/go.sum @@ -22,6 +22,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263 h1:JZV8C+WR git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027221705-2fd5ffdd6263/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466 h1:xzbkI9dOkN/Iv78AS/fOwhjkgBEQOYtpbHzNYwb4eu0= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024 h1:MnmyspVoaCldYmm4Ut1O39RDR9WpiL4TfFgDEnY+QRg= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 37afe58..8f7420e 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -16,6 +16,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "sort" "strings" @@ -27,6 +28,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go" + "github.com/novox/mesh-host/internal/agent" ) @@ -866,3 +869,38 @@ func (c *Composer) keeps(id string, generation int, key string) bool { kept, ok := c.sealedOwn[id] return ok && kept.generation == generation && kept.key == key } + +// Placed reads whether the catalogue places node now, for the node agent choosing whose compositions it applies at +// each start: true when its placement key holds a value, false only when a read succeeds and finds no key or a delete +// marker. A read that fails chooses neither (review of the shadow, round 1: a timeout or a bus not ready at a restart +// flipped a self-composing node to the controller's compositions for a whole run): it is tried again after wait(try), +// each distinct failure said once, until ctx ends, which is then its error. +func Placed(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string, + wait func(try int) time.Duration, say func(string)) (bool, error) { + key, err := delivery.PlacementKey(node) + if err != nil { + return false, err + } + said := "" + for try := 1; ; try++ { + raw, err := get(ctx, delivery.PlacementBucket, key) + if errors.Is(err, nats.ErrBucketNotFound) { + return false, nil // the indexer made no placement state yet: nothing places it, read and known + } + if err == nil { + return len(raw) > 0, nil + } + if ctx.Err() != nil { + return false, ctx.Err() + } + if why := err.Error(); why != said { + say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err)) + said = why + } + select { + case <-ctx.Done(): + return false, ctx.Err() + case <-time.After(wait(try)): + } + } +} diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index b225bcc..7f5ed4e 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -16,6 +16,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/compose" "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go" "github.com/novox/mesh-host/internal/agent" ) @@ -440,6 +441,18 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } return out } + // The controller's compositions at n1's wants, in n1's composition bucket: what a head deleting its placement + // returns it to (the seed's revert); here the ones it runs. + composedBucket, _ := delivery.ComposedBucketFor("n1") + for key, c := range running { + raw, err := delivery.EncodeComposed(c) + if err != nil { + t.Fatal(err) + } + readOnly.mu.Lock() + readOnly.kv[composedBucket+"/"+key] = raw + readOnly.mu.Unlock() + } answered := make(chan map[string]delivery.CandidateAnswer, 8) sh := NewShadow(senv, func(_ context.Context, a map[string]delivery.CandidateAnswer) { answered <- a }) sctx, scancel := context.WithCancel(ctx) @@ -450,7 +463,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond }) time.Sleep(30 * time.Millisecond) sb.missing(delivery.PlacementCandidatesBucket, false) - sb.made(6) + sb.made(7) rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) tp, _ := delivery.EncodePlacement(trunk) sb.put(delivery.PlacementBucket, "placement.n1", tp, 1) @@ -459,6 +472,21 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { raw, _ := delivery.EncodeModule(m) sb.put(delivery.ModulesBucket, name, raw, 1) } + wantLemurs, _ := delivery.WantKey("n1", "lemurs") + var lemursDigest string + for _, c := range running { + if c.Module == "lemurs" { + lemursDigest, _ = delivery.ComposedDigestOf(c) + } + } + wantRaw, err := delivery.EncodeWant(delivery.Want{FP: fp, Composed: lemursDigest}) + if err != nil { + t.Fatal(err) + } + sb.put(delivery.AssignerWantsBucket, wantLemurs, wantRaw, 1) + wantNode, _ := delivery.WantNodeKey("n1") + nodeRaw, _ := delivery.EncodeWant(delivery.Want{}) + sb.put(delivery.AssignerWantsBucket, wantNode, nodeRaw, 1) candidate := func(group string, rev uint64, modules ...string) { p := trunk p.Modules = nil @@ -531,7 +559,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { }, "deletes": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateSame && a.Returns == delivery.ReturnsToController && a.Revision == 7 && - len(a.Removed) == 0 + len(a.Removed) == 0 && fmt.Sprint(a.Same) == "[lemurs node]" }, } { if !want(answers[group]) { @@ -562,6 +590,81 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { return !ok && len(a) == 3 }) + // Whose compositions a node applies at a start (review of the shadow, round 1): only a read that succeeds chooses; a + // failing read is tried again, said once, and chooses neither, even when the link ends. + for name, row := range map[string]struct { + answers []error // per try; nil then the value + value []byte + end bool + placed bool + tries int + quiet bool // read and known: nothing to say + }{ + "placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3}, + "no key": {placed: false, tries: 1}, + "no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true}, + "failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"), + fmt.Errorf("jetstream not ready")}, end: true, tries: 3}, + } { + lctx, lcancel := context.WithCancel(ctx) + tries := 0 + var said []string + get := func(context.Context, string, string) ([]byte, error) { + tries++ + if tries <= len(row.answers) { + if row.end && tries == len(row.answers) { + lcancel() + } + return nil, row.answers[tries-1] + } + return row.value, nil + } + placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) }) + lcancel() + if placed != row.placed || tries != row.tries || (err != nil) != row.end || (len(row.answers) > 0 && !row.quiet) != (len(said) == 1) { + t.Errorf("%s: placed %v, err %v after %d reads, said %q", name, placed, err, tries, said) + } + } + + // The way back, row by row (review of the shadow, round 1): the controller's compositions at n1's wants compared + // with what runs; one differing is answered so; one the controller holds none of refuses, without returning. + var runsList []delivery.Composed + for _, c := range running { + runsList = append(runsList, c) + } + lemursKey, _ = delivery.ComposedKey("n1", "lemurs", fp) + changed := running[lemursKey] + changed.Resources = []json.RawMessage{json.RawMessage(`{"id":"lemurs.service","type":"service","module":"lemurs","unit":"lemurs.service","state":"stopped","boot":"disabled"}`)} + changedRaw, _ := delivery.EncodeComposed(changed) + for name, row := range map[string]struct { + lemurs []byte + check func(delivery.CandidateAnswer) bool + }{ + "the controller's differs": {changedRaw, func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateDiffers && a.Returns == delivery.ReturnsToController && + len(a.Differs) == 1 && a.Differs[0].Module == "lemurs" && fmt.Sprint(a.Same) == "[node]" + }}, + "the controller holds none": {nil, func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateRefused && a.Returns == "" && strings.Contains(a.Refused, "lemurs: the controller holds no composition") + }}, + } { + get := func(_ context.Context, bucket, key string) ([]byte, error) { + if key == lemursKey { + return row.lemurs, nil + } + readOnly.mu.Lock() + defer readOnly.mu.Unlock() + return readOnly.kv[bucket+"/"+key], nil + } + a := returning(ctx, get, "n1", map[string]delivery.Want{"lemurs": {FP: fp}, agent.NodeModule: {}}, runsList, "h", 1) + if !row.check(a) { + t.Errorf("%s: %+v", name, a) + } + if _, err := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": a}}); err != nil { + t.Errorf("%s: the answer is not one a have holds: %v", name, err) + } + } + // Issue 1126, the seed's review: a module n1 runs inside its own composition (nats, at a build of the old form) is // composed there again, not at the trunk's build as one added; lemurs at the build it runs, keeper (added) at the // trunk's. diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index 1a776c8..4a8fea0 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "sort" + "strings" "sync" "time" @@ -31,6 +32,9 @@ type Shadow struct { mu sync.Mutex candidates map[string]map[string]candidateKey // by group, by node answers map[string]delivery.CandidateAnswer + // wants are the node's wants by module (agent.NodeModule for its own): what a head deleting its placement returns + // it to, the controller's compositions at them. + wants map[string]delivery.Want } // candidateKey is one candidate key as last read: its value and revision. @@ -42,7 +46,7 @@ type candidateKey struct { // NewShadow is the shadow composer of one node; answer writes the node's answers into its own have key. func NewShadow(env Env, answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)) *Shadow { return &Shadow{c: New(env), answer: answer, candidates: map[string]map[string]candidateKey{}, - answers: map[string]delivery.CandidateAnswer{}} + answers: map[string]delivery.CandidateAnswer{}, wants: map[string]delivery.Want{}} } // Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch @@ -84,6 +88,7 @@ func (s *Shadow) Run(ctx context.Context) error { defer cancel() s.mu.Lock() s.candidates = map[string]map[string]candidateKey{} + s.wants = map[string]delivery.Want{} s.c = New(s.c.env) s.mu.Unlock() c := s.c @@ -96,6 +101,8 @@ func (s *Shadow) Run(ctx context.Context) error { {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, // What this node runs moved: an apply ends in its module's have. {delivery.HaveBucket, node + ".>", nil, nil}, + // Its wants: a head deleting its placement returns it to the controller's compositions at them. + {delivery.AssignerWantsBucket, "want." + node + ".>", nil, nil}, } type tagged struct { in int @@ -142,6 +149,8 @@ func (s *Shadow) Run(ctx context.Context) error { case t.in == 5: c.takeModuleHave(t.u) dirty = !isNodeKey(t.u.Key) || dirty + case t.in == 6: + dirty = s.takeWant(t.u) || dirty default: ins[t.in].take(c, t.u) dirty = true @@ -293,10 +302,15 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid } if p, ok := given[node]; ok && p == nil { // The head deletes this node's placement file: the node then returns to the controller's compositions, as one the - // catalogue does not place (endWhenUnplaced), and keeps what runs; nothing is removed (issue 1126, the seed's - // revert, whose check read "would stop running node"). - return delivery.CandidateAnswer{Head: head, Revision: rev, Returns: delivery.ReturnsToController, - Verdict: delivery.CandidateSame}, true + // catalogue does not place (endWhenUnplaced). Those at its wants are compared with what runs (issue 1126, the + // seed's revert, whose check read "would stop running node"). + s.mu.Lock() + wants := make(map[string]delivery.Want, len(s.wants)) + for m, w := range s.wants { + wants[m] = w + } + s.mu.Unlock() + return returning(ctx, c.env.Get, node, wants, running, head, rev), true } placement := given[node] @@ -362,3 +376,88 @@ func sameAnswer(a, b delivery.CandidateAnswer) bool { y, errB := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": b}}) return errA == nil && errB == nil && string(x) == string(y) } + +// takeWant takes one of the node's wants, and says whether it changed. +func (s *Shadow) takeWant(u agent.Update) bool { + module := "" + var w delivery.Want + if _, err := delivery.ParseWantNodeKey(u.Key); err == nil { + module = agent.NodeModule + } else if node, m, err := delivery.ParseWantKey(u.Key); err == nil && node == s.c.env.Node { + module = m + if !u.Deleted { + if w, err = delivery.DecodeWant(u.Value); err != nil || w.Removed || w.FP == "" { + u.Deleted = true + } + } + } else { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + old, had := s.wants[module] + if u.Deleted { + delete(s.wants, module) + return had + } + s.wants[module] = w + return !had || old.FP != w.FP +} + +// returning is the node's answer to a head deleting its placement: it returns to the controller's compositions, each +// read at the fingerprint its want names in the node's composition bucket and compared with what runs +// (delivery.CompareComposed). A want the controller holds no composition for, or one that cannot be read, refuses the +// answer, naming each: returning then would leave that module with nothing to apply. +func returning(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string, + wants map[string]delivery.Want, running []delivery.Composed, head string, rev uint64) delivery.CandidateAnswer { + a := delivery.CandidateAnswer{Head: head, Revision: rev} + bucket, err := delivery.ComposedBucketFor(node) + if err != nil { + a.Refused, a.Verdict = err.Error(), delivery.CandidateRefused + return a + } + modules := make([]string, 0, len(wants)) + for m := range wants { + modules = append(modules, m) + } + sort.Strings(modules) + comps := map[string]delivery.Composed{} + var missing []string + for _, m := range modules { + var key string + if m == agent.NodeModule { + key, err = delivery.ComposedNodeKey(node) + } else { + key, err = delivery.ComposedKey(node, m, wants[m].FP) + } + if err != nil { + missing = append(missing, m+": "+err.Error()) + continue + } + raw, err := get(ctx, bucket, key) + switch { + case err != nil: + missing = append(missing, fmt.Sprintf("%s: the controller's composition %s cannot be read: %v", m, key, err)) + continue + case raw == nil: + missing = append(missing, fmt.Sprintf("%s: the controller holds no composition %s, which its want names", m, key)) + continue + } + comp, err := delivery.DecodeComposed(raw) + if err != nil { + missing = append(missing, fmt.Sprintf("%s: the controller's composition %s: %v", m, key, err)) + continue + } + comps[key] = comp + } + if len(missing) > 0 { + a.Refused = "returning to the controller's compositions, " + strings.Join(missing, "; ") + a.Verdict = delivery.CandidateRefused + return a + } + cmp := delivery.CompareComposed(comps, running) + a.Returns = delivery.ReturnsToController + a.Same, a.Differs, a.Added, a.Removed = cmp.Same, cmp.Differs, cmp.Added, cmp.Removed + a.Verdict = delivery.VerdictOf(a) + return a +}