From 76e214a8b93550b8e7f56d17881853ae8b877642 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:16:38 +0100 Subject: [PATCH] Keep every composition the engine applied on the machine, so a module whose composition the bus no longer keeps runs as applied and is never composed as nothing (issue 774) On novox on 2026-10-29 at 07:54 UTC the kept declaration was retired and the node's own composition was in no bucket. The composer's first pass reads what runs before the agent reads any have (the agent waits for the composer's caught, sent after that pass), so the node's own was composed with its folded modules running nothing, and the bus server and four held modules were removed. - Quick fix: read the node's own have in loadHad too. It still waits for the composer's first pass, and still finds nothing when the bucket no longer keeps the composition the have names. - Thorough fix, chosen: the engine keeps each composition it applied (the node's own too) in `applied/` beside its state, 0600. The agent reads it at Run's start, before any watch, as what runs ahead of the bucket and the kept declaration. A have naming another running composition wins. An apply keeps; a removal forgets. Test: a case in TestEverythingIsWhatTheAgentRunsAndWhatIsKeptForTheRest (no new test function). It fails on L3's engine 4f9227a (0 compositions at the first own composition) and passes here. --- cmd/mesh-host/applied_store.go | 97 ++++++++++++++++++++++++++++++ cmd/mesh-host/nodeagent.go | 5 +- internal/agent/agent.go | 102 +++++++++++++++++++++++++++++++ internal/agent/agent_test.go | 106 +++++++++++++++++++++++++++++++++ 4 files changed, 309 insertions(+), 1 deletion(-) create mode 100644 cmd/mesh-host/applied_store.go diff --git a/cmd/mesh-host/applied_store.go b/cmd/mesh-host/applied_store.go new file mode 100644 index 0000000..60df951 --- /dev/null +++ b/cmd/mesh-host/applied_store.go @@ -0,0 +1,97 @@ +package main + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "regexp" + "strings" +) + +// appliedStore keeps every composition the agent applied, one file per module (the node's own as `node.json`), in the +// directory `applied` beside the node's state (agent.Env.Applied, issue 774). It is the machine's own record of what runs: +// read at the agent's start before anything is watched, so the node never composes its own as if a module it runs ran +// nothing because the bucket no longer keeps that composition or the kept declaration is retired. Files are 0600 in a +// 0700 directory: a composition carries values sealed to this machine. +type appliedStore struct{ dir string } + +func appliedStoreFor(statePath string) appliedStore { + return appliedStore{dir: filepath.Join(filepath.Dir(statePath), "applied")} +} + +// appliedName is a module name the store takes as a file name: a catalogue module's or the node's own, never a path. +var appliedName = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]*$`) + +func (s appliedStore) path(module string) (string, error) { + if !appliedName.MatchString(module) || strings.Contains(module, "..") { + return "", fmt.Errorf("%q is not a module name the applied store takes", module) + } + return filepath.Join(s.dir, module+".json"), nil +} + +func (s appliedStore) Load() (map[string][]byte, error) { + entries, err := os.ReadDir(s.dir) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, err + } + out := map[string][]byte{} + var errs []error + for _, e := range entries { + module, ok := strings.CutSuffix(e.Name(), ".json") + if !ok || e.IsDir() || !appliedName.MatchString(module) { + continue + } + raw, err := os.ReadFile(filepath.Join(s.dir, e.Name())) + if err != nil { + errs = append(errs, err) + continue + } + out[module] = raw + } + return out, errors.Join(errs...) +} + +func (s appliedStore) Keep(module string, raw []byte) error { + path, err := s.path(module) + if err != nil { + return err + } + if was, err := os.ReadFile(path); err == nil && string(was) == string(raw) { + return nil + } + if err := os.MkdirAll(s.dir, 0o700); err != nil { + return err + } + tmp, err := os.CreateTemp(s.dir, "."+module+".*") + if err != nil { + return err + } + defer os.Remove(tmp.Name()) + if _, err := tmp.Write(raw); err != nil { + tmp.Close() + return err + } + if err := tmp.Sync(); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmp.Name(), path) +} + +func (s appliedStore) Drop(module string) error { + path, err := s.path(module) + if err != nil { + return err + } + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + return nil +} diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 0d22773..07fc79c 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -494,7 +494,10 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus // The same kept declaration's filled paths, so a module still running from it keeps its files filled. KeptFilled: func() map[string][]string { _, filled := kept(); return filled }, Recorded: recordedIn(opts.state, say), - Inactive: inactiveIn(&liveness.Exec{Run: apply.ExecRunner, System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))), + // What this engine applied, kept on the machine and read at the start, so what runs is known before the node + // composes its own (issue 774). + Applied: appliedStoreFor(opts.state), + Inactive: inactiveIn(&liveness.Exec{Run: apply.ExecRunner, System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))), Revisions: func(ctx context.Context, key string) ([]agent.Update, error) { got, err := session.KVHistory(ctx, composedBucket, key) if err != nil { diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 8525b4b..0b47ede 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -156,6 +156,22 @@ type Env struct { // Data of its own. A composition says Data, so without it a data module from the kept declaration is judged to move // its data against its first composition after a restart (mongodb and mssql on novox, issue 924). Nil reads none. HoldsData func(ctx context.Context, module string) (data, known bool) + // Applied keeps every composition the agent applied, by module (the node's own under NodeModule), beside the engine's + // state on the machine: read at the start before anything is watched, so what runs is known before the node composes + // its own for the first time, whatever the bucket and the kept declaration still hold (issue 774: on 2026-10-29 the + // bus node, its kept declaration retired and its own composition in no bucket, composed its own as if its folded + // modules ran nothing and removed the bus server and four held modules). Nil keeps none. + Applied AppliedStore +} + +// AppliedStore is where the engine keeps the compositions it applied (Env.Applied): each module's last, as its bytes. +type AppliedStore interface { + // Load is every composition kept, by module; a store that cannot be read says so and loads none. + Load() (map[string][]byte, error) + // Keep keeps the composition the module was applied from, in place of the one before. + Keep(module string, raw []byte) error + // Drop forgets the module's, once it is removed. + Drop(module string) error } // transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has @@ -248,6 +264,9 @@ func New(env Env) *Agent { // before any module's, then each change as it comes. func (a *Agent) Run(ctx context.Context) error { defer close(a.stopped) + // What this engine last applied, before any composition is watched: the node composing its own reads what runs + // (Everything) from its first pass on, and that pass comes before the haves are read below (issue 774). + a.loadApplied(ctx) wants, err := a.env.Wants.Watch(ctx, "want."+a.env.Node+".>") if err != nil { return fmt.Errorf("the agent cannot watch its wants: %w", err) @@ -861,6 +880,10 @@ func (a *Agent) loadHad(ctx context.Context) { continue } a.mu.Lock() + if c, kept := a.had[m]; kept && c.Dig == h.Composed { + a.mu.Unlock() + continue // what this engine applied and kept itself (loadApplied) + } c, found := a.compositionFor(m, h.FP) a.mu.Unlock() if !found || c.Err != nil || c.Dig != h.Composed { @@ -869,6 +892,8 @@ func (a *Agent) loadHad(ctx context.Context) { a.mu.Lock() if found { a.had[m] = c + } else if _, applied := a.had[m]; applied { + // Kept as this engine applied it (loadApplied), never composed as nothing. } else if _, kept := a.keptHas(m); kept { a.sayOnce("had:"+m, fmt.Sprintf("%s's have names the composition %s, which the bucket no longer keeps: it runs "+ "from the kept declaration until it is applied or taken", m, short(h.Composed))) @@ -877,6 +902,76 @@ func (a *Agent) loadHad(ctx context.Context) { } } +// loadApplied takes, once at the start, what this engine last applied of each module as what runs of it, ahead of the +// bucket and the kept declaration (Env.Applied, issue 774): it is what the machine holds, whatever the bus still keeps. +// One whose have names another composition that runs (applied since by another engine) is left to the have; one whose +// have cannot be read, or that has none, is kept as applied, so a module is never composed as running nothing. +func (a *Agent) loadApplied(ctx context.Context) { + if a.env.Applied == nil { + return + } + kept, err := a.env.Applied.Load() + if err != nil { + a.env.Say(fmt.Sprintf("the compositions this engine applied cannot be read, so what runs is read from the haves alone: %v", err)) + } + names := make([]string, 0, len(kept)) + for m := range kept { + names = append(names, m) + } + sort.Strings(names) + for _, m := range names { + if ctx.Err() != nil { + return + } + c, err := delivery.DecodeComposed(kept[m]) + var dig string + if err == nil { + dig, err = delivery.ComposedDigestOf(c) + } + if err != nil { + a.env.Say(fmt.Sprintf("the composition this engine kept of %s cannot be read, so what runs of it is read from its have: %v", m, err)) + continue + } + h, found, unread := a.readHave(ctx, m) + if found && h.Composed != "" && h.Composed != dig && haveRuns(h) { + a.env.Say(fmt.Sprintf("%s's have names the composition %s, not %s this engine applied last: what runs of it is read from its have", + m, short(h.Composed), short(dig))) + continue + } + if m == NodeModule { + c.Module, c.FP = "", "" + } + a.mu.Lock() + a.had[m] = compNow{Raw: kept[m], C: c, Dig: dig} + a.mu.Unlock() + switch { + case unread: + a.env.Say(fmt.Sprintf("%s's have cannot be read: it runs as this engine applied it, %s", m, short(dig))) + case !found: + a.env.Say(fmt.Sprintf("%s has no have: it runs as this engine applied it, %s", m, short(dig))) + } + } +} + +// keepApplied keeps the composition a module was applied from (Env.Applied); one that cannot be kept is said, and the +// apply stands. +func (a *Agent) keepApplied(module string, comp compNow) { + if a.env.Applied == nil { + return + } + raw := comp.Raw + if len(raw) == 0 { + var err error + if raw, err = delivery.EncodeComposed(comp.C); err != nil { + a.env.Say(fmt.Sprintf("the composition %s was applied from cannot be kept: %v", module, err)) + return + } + } + if err := a.env.Applied.Keep(module, raw); err != nil { + a.env.Say(fmt.Sprintf("the composition %s was applied from cannot be kept: %v", module, err)) + } +} + // haveRuns says a have names what runs, in the phases baseline takes as running: running (or empty), held for a failed // backup, or a composer's mark over what still runs. A have backing up, applying, checking, failed or waiting names a // composition not applied, or not yet (issue 924, review of 031e106). @@ -1327,6 +1422,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo a.running[module] = delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning} a.applied[module] = comp a.mu.Unlock() + a.keepApplied(module, comp) if a.env.Settled != nil { a.env.Settled(module) } @@ -1356,8 +1452,14 @@ func (a *Agent) remove(ctx context.Context, module string, w wantNow) { return } delete(a.attempt, module) + delete(a.had, module) a.running[module] = delivery.Have{Phase: delivery.PhaseRemoved} a.mu.Unlock() + if a.env.Applied != nil { + if err := a.env.Applied.Drop(module); err != nil { + a.env.Say(fmt.Sprintf("the composition %s was applied from cannot be forgotten now it is removed: %v", module, err)) + } + } if a.env.Settled != nil { a.env.Settled(module) } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index a73faa2..4aa7d9f 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -769,6 +769,112 @@ func TestEverythingIsWhatTheAgentRunsAndWhatIsKeptForTheRest(t *testing.T) { if a.HoldsNode() { t.Fatal("the agent holds no node composition") } + + // Issue 774 (novox, 2026-10-29 07:54 UTC): the kept declaration retired and the node's own composition in no bucket, a + // fresh engine's first own composition was composed from what runs with the node's own missing, so its folded modules + // (the bus server among them) were composed as nothing and removed. What this engine applied is kept on the machine + // and is what runs from before the first composition is watched; an apply keeps, a removal forgets. + t.Run("what this engine applied is what runs before the node composes its own", func(t *testing.T) { + b := newBed(t) + b.env.SelfComposed, b.env.Trusted, b.kept = true, nil, nil + store := &memApplied{kept: map[string][]byte{}} + b.env.Applied = store + // A folded module's resource in the node's own composition: its id names the module, as the controller folds it. + natsConf, _ := json.Marshal(map[string]any{"id": "nats.conf", "type": "file", "path": "/tmp/r6/nats/conf", "content": "listen 4222"}) + own := delivery.Composed{Node: node, Resources: []json.RawMessage{natsConf}} + ownRaw, err := delivery.EncodeComposed(own) + if err != nil { + t.Fatal(err) + } + ownDig, _ := delivery.ComposedDigestOf(own) + web := delivery.Composed{Node: node, Module: "web", FP: fpOf("a"), Resources: []json.RawMessage{file("web", "x", "1")}} + webRaw, _ := delivery.EncodeComposed(web) + webDig, _ := delivery.ComposedDigestOf(web) + store.kept[NodeModule], store.kept["web"] = ownRaw, webRaw + b.haves.Write(context.Background(), Have{Module: NodeModule, Composed: ownDig}) + b.haves.Write(context.Background(), Have{Module: "web", FP: fpOf("a"), Composed: webDig, Phase: string(delivery.PhaseRunning)}) + // The composer's first pass reads what runs when its watch begins, before the agent reads any have. + firstCh := make(chan []delivery.Composed, 1) + comps := b.comps + var ag *Agent + b.env.Composed = watchFunc(func(ctx context.Context, p string) (<-chan Update, error) { + firstCh <- ag.Everything() + return comps.Watch(ctx, p) + }) + ag = New(b.env) + b.ag = ag + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { defer close(done); _ = ag.Run(ctx) }() + defer func() { cancel(); <-done }() + var first []delivery.Composed + select { + case first = <-firstCh: + case <-time.After(5 * time.Second): + t.Fatal("the composer was not watched") + } + var ranOwn, ranWeb bool + for _, c := range first { + ranOwn = ranOwn || (c.Module == "" && len(c.Resources) == 1 && strings.Contains(string(c.Resources[0]), "4222")) + ranWeb = ranWeb || c.Module == "web" + } + if !ranOwn || !ranWeb { + t.Fatalf("what runs at the node's first own composition: own %v, web %v (%d compositions); want both as this engine applied them", + ranOwn, ranWeb, len(first)) + } + // An apply keeps the composition it applied; a removal forgets it. + web2 := delivery.Composed{Node: node, Module: "web", FP: fpOf("b"), Resources: []json.RawMessage{file("web", "x", "2")}} + ck, _ := delivery.ComposedKey(node, "web", fpOf("b")) + raw2, _ := delivery.EncodeComposed(web2) + comps.put(ck, raw2) + wk, _ := delivery.WantKey(node, "web") + wraw, _ := delivery.EncodeWant(delivery.Want{FP: fpOf("b"), Self: true}) + b.wants.put(wk, wraw) + eventually(t, "the applied composition was not kept", func() bool { return string(store.get("web")) == string(raw2) }) + rraw, _ := delivery.EncodeWant(delivery.Want{Removed: true}) + b.wants.put(wk, rraw) + eventually(t, "the removed module's composition was not forgotten", func() bool { return store.get("web") == nil }) + }) +} + +// watchFunc is a Watcher from a function. +type watchFunc func(ctx context.Context, pattern string) (<-chan Update, error) + +func (f watchFunc) Watch(ctx context.Context, pattern string) (<-chan Update, error) { + return f(ctx, pattern) +} + +// memApplied is an AppliedStore in memory. +type memApplied struct { + mu sync.Mutex + kept map[string][]byte +} + +func (m *memApplied) Load() (map[string][]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + out := map[string][]byte{} + for k, v := range m.kept { + out[k] = v + } + return out, nil +} +func (m *memApplied) Keep(module string, raw []byte) error { + m.mu.Lock() + defer m.mu.Unlock() + m.kept[module] = raw + return nil +} +func (m *memApplied) Drop(module string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.kept, module) + return nil +} +func (m *memApplied) get(module string) []byte { + m.mu.Lock() + defer m.mu.Unlock() + return m.kept[module] } // A failed apply is applied again after a restart even when the kept declaration holds the composition: once this