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