package main import ( "context" "crypto/ed25519" "encoding/json" "errors" "fmt" "os" "path/filepath" "sort" "strings" "sync" "sync/atomic" "time" "git.novox.be/novox/mesh-sdk/go/declaration" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/overlay" "github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/compose/frombus" "github.com/novox/mesh-host/internal/link" "github.com/novox/mesh-host/internal/liveness" "github.com/novox/mesh-host/internal/selfcompose" "github.com/novox/mesh-host/internal/store" ) // The node agent's wiring (novox/hq ADR 0429 §4, ADR 0539 §9): the buckets it watches, the haves it writes, the backup // it asks and the apply of one module's composition. // bucketWatch is a bucket of the bus as the agent watches it, through the session's link. type bucketWatch struct { session *link.Session bucket string } func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Update, error) { in, err := w.session.KVWatch(ctx, w.bucket, pattern) if err != nil { return nil, err } return w.updates(ctx, in), nil } // WatchAwait is Watch for a state that its holder makes when it starts (the placement seat's): a bucket not on the bus // yet is waited for on the bus's word that it was created (link.Session.KVWatchAwait), and waiting is told why. func (w bucketWatch) WatchAwait(ctx context.Context, pattern string, waiting func(error)) (<-chan agent.Update, error) { in, err := w.session.KVWatchAwait(ctx, w.bucket, pattern, waiting) if err != nil { return nil, err } return w.updates(ctx, in), nil } func (w bucketWatch) updates(ctx context.Context, in <-chan link.KVUpdate) <-chan agent.Update { out := make(chan agent.Update, 64) go func() { defer close(out) for u := range in { select { case out <- agent.Update{Key: u.Key, Value: u.Value, Revision: u.Revision, Deleted: u.Deleted, Caught: u.Caught}: case <-ctx.Done(): return } } }() return out } // sessionHaves writes the haves the agent says, through the session. type sessionHaves struct{ session *link.Session } func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool) { if module == agent.NodeModule { // The node's own key: the digest of its own composition it runs, and what that carries sealed. got, ok := h.session.ReadNodeHave(ctx) if !ok || got.Composed == "" { return agent.Have{}, false } return agent.Have{Module: module, Composed: got.Composed, Sealed: got.SealedAt}, true } got, ok := h.session.ReadHave(ctx, module) if !ok { return agent.Have{}, false } return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, Sealed: got.SealedAt}, true } // ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr). func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, bool, error) { if module == agent.NodeModule { // The node's own key, its read the bus did not answer told as such (issue 1126, 2026-10-28). got, ok, err := h.session.ReadNodeHaveErr(ctx) if !ok || err != nil || got.Composed == "" { return agent.Have{}, false, err } return agent.Have{Module: module, Composed: got.Composed, Sealed: got.SealedAt}, true, nil } got, ok, err := h.session.ReadHaveErr(ctx, module) if !ok || err != nil { return agent.Have{}, false, err } return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail, Sealed: got.SealedAt}, true, nil } func (h sessionHaves) Write(ctx context.Context, x agent.Have) { if x.Module == agent.NodeModule { // The node's own key holds its facts and the digest of the composition it runs. if x.Phase == "" || x.Phase == string(delivery.PhaseRunning) { h.session.NoteNode(ctx, x.Composed, x.Sealed) } return } phase := x.Phase if phase == string(delivery.PhaseRunning) { phase = "" } h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase, Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail), Judgement: x.Judgement, Waits: x.Waits, BuildAlone: x.BuildAlone, Kept: x.Kept, Sealed: x.Sealed}) } func (h sessionHaves) Removed(ctx context.Context, module string) { h.session.NoteRemoved(ctx, module, "") } // shorten keeps a have's detail within what a have value holds: its head, which says what failed, and its tail, which is // the cause (an error wrapped five times names the bus's "no responders" only at its end: 2026-10-25 23:04, mesh-vault on // the bus node said only which object it read). func shorten(s string) string { const most, head = 300, 100 if len(s) <= most { return s } const gap = " … " tail := s[len(s)-(most-head-len(gap)):] return strings.ToValidUTF8(s[:head], "") + gap + strings.ToValidUTF8(tail, "") } // failedWith is a failed apply that says why, transient when the words are the bus not answering for a moment // (link.TransientWords): the agent then tries it again without a new revision. func failedWith(why string) agent.Result { return agent.Result{Failed: true, Transient: link.TransientWords(why), Detail: shorten(why)} } // backupWithin is how long a backup is given from its ask to say it is done (ADR 0429 §14). var backupWithin = 10 * time.Minute // backedUpEvent is where the holder of node-backup says a backup taken by hand ended, also one that could not start. const backedUpEvent = "mesh.mod.restic.event.backed-up" // backupNode asks the holder of node-backup on this node to back a module up now, and waits for its event to say how it // went: the holder answers `now` when it has started, and says `backed-up` when the backup ends. func backupNode(session *link.Session) func(ctx context.Context, module string) agent.Backup { return func(ctx context.Context, module string) agent.Backup { // Heard before the ask, so an event that comes at once is not lost. events, stop, err := session.Hear(backedUpEvent) if err != nil { return agent.Backup{Words: "the end of the backup cannot be heard: " + err.Error()} } defer stop() // A module this machine backs nothing of up moves without a backup: never held for one that cannot run (issue // 1945). An answer that cannot be had or read asks for the backup as before. ask := func(verb string) func(context.Context) (json.RawMessage, error) { return func(ctx context.Context) (json.RawMessage, error) { return session.AskSeat(ctx, "node-backup", verb, map[string]any{"module": module}, 30*time.Second) } } if said, err := askUntilHeard(ctx, ask("backed-up"), backupAskFor, link.Backoff); err == nil { if nothing, read := declaresNothingToBackUp(said, module); read && nothing { return agent.Backup{Done: true, Words: module + " declares nothing to back up on this machine"} } } asked := time.Now() if _, err := askUntilHeard(ctx, ask("now"), backupAskFor, link.Backoff); err != nil { return agent.Backup{Words: err.Error()} } return awaitBackup(ctx, events, session.Membership.Node, module, asked, backupWithin) } } // backupAskFor is how long the backup's ask is asked again while it is known never to have reached the holder of // node-backup: nothing on this machine answers it (the tool runner restarting, its subscriptions not made yet) or no // link is open. The sweep of 2026-10-30 met it on ace's redis, postgres and restic (14:03), step-ca (13:55) and // mesh-vault on novox (15:22): each was held on one unanswered ask. Past the bound the backup failed, said with how // often and how long it was asked: the module is held, its have says so, and the assigner asks the operator. var backupAskFor = 3 * time.Minute // askUntilHeard asks, and asks again after wait(try) while the answer says the ask was never delivered (link.ErrNotAsked), // for at most within: an ask that may have reached its holder (a timeout, a refusal in words) is never asked twice, so a // backup is never started twice. func askUntilHeard(ctx context.Context, ask func(context.Context) (json.RawMessage, error), within time.Duration, wait func(int) time.Duration) (json.RawMessage, error) { began := time.Now() for try := 1; ; try++ { said, err := ask(ctx) if err == nil || !errors.Is(err, link.ErrNotAsked) { return said, err } pause := wait(try) if time.Since(began)+pause > within { return nil, fmt.Errorf("asked %d times over %s and never heard: %w", try, time.Since(began).Round(time.Second), err) } select { case <-ctx.Done(): return nil, fmt.Errorf("the node agent was stopped while it asked again: %w", err) case <-time.After(pause): } } } // awaitBackup waits for the event that says the backup of a module on this node ended, within the bound: done when it // says ok, held on its words when it does not, and held when it says nothing within the bound. An event of another node // or module, or from before the ask, is not the answer. func awaitBackup(ctx context.Context, events <-chan []byte, node, module string, since time.Time, within time.Duration) agent.Backup { bound := time.NewTimer(within) defer bound.Stop() for { select { case <-ctx.Done(): return agent.Backup{Words: agent.BackupInterrupted, Interrupted: true} case <-bound.C: return agent.Backup{Words: fmt.Sprintf("the backup did not say it was done within %s", within)} case raw := <-events: var e struct { Node string `json:"node"` Module string `json:"module"` OK bool `json:"ok"` At time.Time `json:"at"` Error string `json:"error"` } if json.Unmarshal(raw, &e) != nil || e.Node != node || e.Module != module || e.At.Before(since.Add(-time.Second)) { continue } if e.OK { return agent.Backup{Done: true} } words := "the backup failed" if e.Error != "" { words += ": " + e.Error } return agent.Backup{Words: words} } } } // declarationOf is the declaration the compositions make together: their resources, and the node's own parts when the // node's composition is among them. It is what the node judges and says its health by, and what one module's apply is // made of (with a single composition). // // **A resource the node's own composition shares with a module's is the module's** (issue 1945): from 2026-10-25 12:43 // the controller put modules it knew no fingerprint of the new form for into the node's own composition while their // wants named their own, and the declaration joined of both declared each such file twice, which the filling refuses // ("2 resources declare X") on every node. The node's own copy is left out, and which were left out is said once. func declarationOf(cs ...delivery.Composed) (*declaration.Declaration, error) { d := &declaration.Declaration{Version: declaration.Version} var modules []delivery.Composed for _, c := range cs { if c.Module != "" { modules = append(modules, c) } } if len(modules) > 0 { joined := make([]delivery.Composed, 0, len(cs)) for _, c := range cs { if c.Module == "" { var dropped []string c, dropped = ownWithout(c, modules) sayDroppedOnce(dropped) } joined = append(joined, c) } cs = joined } for _, c := range cs { for _, raw := range c.Resources { r, err := declaration.ParseResource(raw) if err != nil { return nil, fmt.Errorf("a resource of %q: %w", c.Module, err) } d.Resources = append(d.Resources, r) } d.Filled = append(d.Filled, c.Filled...) if c.Module == "" { d.Adoption = c.Own.Adoption d.Identities = append(d.Identities, c.Own.Identities...) d.Groups = append(d.Groups, c.Own.Groups...) d.Libraries = append(d.Libraries, c.Own.Libraries...) d.Provisioner = c.Own.Provisioner } } return d, nil } // ownWithout is the node's own composition without the resources a module's composition among modules also declares: // the same kind and id. A resource that shares only its target (a unit or container of the same name, another id) is // not the module's, and stays: dropped, its record would be an orphan the apply removes, stopping the module's unit. It // answers what it left out, as " ()". func ownWithout(own delivery.Composed, modules []delivery.Composed) (delivery.Composed, []string) { byID := map[string]string{} for _, c := range modules { if c.Module == "" { continue } for _, raw := range c.Resources { r, err := declaration.ParseResource(raw) if err != nil { continue } byID[string(r.Kind())+"\x00"+r.Identity()] = c.Module } } if len(byID) == 0 { return own, nil } var dropped []string kept := make([]json.RawMessage, 0, len(own.Resources)) for _, raw := range own.Resources { r, err := declaration.ParseResource(raw) if err != nil { kept = append(kept, raw) // refused where it is parsed, with its words continue } module, dup := byID[string(r.Kind())+"\x00"+r.Identity()] if dup { dropped = append(dropped, fmt.Sprintf("%s (%s)", r.Identity(), module)) continue } kept = append(kept, raw) } if len(dropped) == 0 { return own, nil } own.Resources = kept return own, dropped } // droppedSaid is every set of the node's own resources left out that was said, so each is said once. var droppedSaid sync.Map // sayDroppedOnce says once which of the node's own resources were left out because a module's composition declares them. func sayDroppedOnce(dropped []string) { if len(dropped) == 0 { return } sort.Strings(dropped) key := strings.Join(dropped, ", ") if _, said := droppedSaid.LoadOrStore(key, true); said { return } fmt.Fprintf(os.Stderr, "mesh-host: the node's own composition also declares %d resource(s) a module's composition "+ "declares, and they are left out of it: %s (issue 1945)\n", len(dropped), key) } func scopeName(c delivery.Composed) string { if c.Module == "" { return agent.NodeModule } return c.Module } func resourcesOf(cs []delivery.Composed) []declaration.Resource { var out []declaration.Resource for _, c := range cs { for _, raw := range c.Resources { if r, err := declaration.ParseResource(raw); err == nil { out = append(out, r) } } } return out } // keptOnce reads the kept declaration once and splits it by module: its resources (the node's own are those that say // none, ADR 0539 §9) and its filled paths. Once the read is retired (keptRetired) nothing is read. func keptOnce(opts options, signer ed25519.PublicKey) func() (map[string][]json.RawMessage, map[string][]string) { var once sync.Once var by map[string][]json.RawMessage var filled map[string][]string return func() (map[string][]json.RawMessage, map[string][]string) { once.Do(func() { if keptRetired.Load() { return } raw, err := store.LoadDeclared(store.DeclaredPath(opts.state), signer) if err != nil { return } by, filled = splitKept(raw) }) return by, filled } } // splitKept is a kept declaration's resources by the module they say (the node's own under agent.NodeModule) and its // filled paths by the module whose file is at each path; nil for a declaration that does not read. func splitKept(raw []byte) (map[string][]json.RawMessage, map[string][]string) { var kept struct { Resources []json.RawMessage `json:"resources"` Filled []string `json:"filled_files"` } if json.Unmarshal(raw, &kept) != nil { return nil, nil } isFilled := make(map[string]bool, len(kept.Filled)) for _, p := range kept.Filled { isFilled[p] = true } by, filled := map[string][]json.RawMessage{}, map[string][]string{} named := map[string]bool{} for _, r := range kept.Resources { var head struct { Module string `json:"module"` Type string `json:"type"` Path string `json:"path"` } _ = json.Unmarshal(r, &head) name := head.Module if name == "" { name = agent.NodeModule } by[name] = append(by[name], r) if head.Type == string(declaration.TypeFile) && isFilled[head.Path] && !named[name+"\x00"+head.Path] { named[name+"\x00"+head.Path] = true filled[name] = append(filled[name], head.Path) } } return by, filled } // nodeAgent runs the node agent under the session's links for as long as ctx lasts: it waits for a link, follows the // node's wants and compositions on it, and begins again with the next link. A watch it cannot make (a bucket this node's // account may not read) is said, and tried again with a growing wait, as the link is. func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trust *meshTrust, session *link.Session, sched *apply.Scheduler, say link.Announce, afterApply func(), current *agentNow, sayFacts func(ctx context.Context, ag *agent.Agent)) { wait := 2 * time.Second for ctx.Err() == nil { linked, cancel, err := session.WhenLinked(ctx) if err != nil { return } began := time.Now() node := session.Membership.Node composedBucket, err := delivery.ComposedBucketFor(node) if err != nil { cancel() say("the node agent cannot name this node's composition bucket: " + err.Error()) return } var ag *agent.Agent // The kept release-5 declaration is no longer read once every module the seat's placement puts here has a have of its // own (ADR 0550 §2: its signature is the controller's, trusted no more after the lease write). // Nor before this engine keeps the node's own as it runs (applied/node.json): until then the kept declaration is the one // record of what the node's own composition carries, its folded modules among them (issue 774, review of the // re-land, round 1: novox's first start would compose its own with nats running nothing). if retired, why := placedAllHave(linked, session, node); retired && !keptRetired.Load() { if !appliedStoreFor(opts.state).holds(agent.NodeModule) { say("the kept declaration is still read: " + why + ", but this engine keeps no node's own composition yet") } else if !keptRetired.Swap(true) { say("the kept declaration is no longer read: " + why) } } kept := keptOnce(opts, signer) // The mesh's own certificate the declaration kept from before carries, kept for the record before this machine // composes its own (ADR 0555 §3): without it a module that asks for a certificate would be stood in for. if byModule, _ := kept(); byModule != nil { for module, resources := range byModule { if err := keepAuthority(opts.state, delivery.Composed{Module: module, Resources: resources}); err != nil { say("the mesh's certificate the kept declaration carries could not be kept for this machine's record: " + err.Error()) } } } // The bus's user list, the grants part's alone (ADR 0547 §8, clause 11; issue 1126), written into the file the // bus module this node runs declares, as its agent holds what runs (its own composition, or the declaration kept // before the switch): never read from the node's composition bucket, which the controller wrote. On every node; // a node that runs no list writes nothing. Looked at again whenever what runs changed (RunsChanged, Settled). users := userListFollower(opts, session, func() []delivery.Composed { if now := current.get(); now != nil { return now.Everything() } return nil }, say, func() { // This engine wrote the list, so the bus reloads its accounts and answers no JetStream request for a few // seconds: what failed in them is tried again once that is over, not at its next wait. time.AfterFunc(reloadSettles, func() { if now := current.get(); now != nil { now.RetryTransient() } }) }) go func() { if err := users.Run(linked); err != nil && linked.Err() == nil { say("the bus's user list is not followed on this link: " + err.Error()) } }() env := agent.Env{ Node: node, Wants: bucketWatch{session, delivery.AssignerWantsBucket}, Composed: bucketWatch{session, composedBucket}, Haves: sessionHaves{session}, Kept: func() map[string][]json.RawMessage { by, _ := kept(); return by }, // 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), // 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: liveness.Runner(opts.run()), System: builtFor}, heldBy(apply.WindowsIn(filepath.Dir(opts.state)))), Held: func() map[string]string { if c := composerNow.Load(); c != nil { return c.LastHeld() } return nil }, Revisions: func(ctx context.Context, key string) ([]agent.Update, error) { got, err := session.KVHistory(ctx, composedBucket, key) if err != nil { return nil, err } out := make([]agent.Update, 0, len(got)) for _, u := range got { out = append(out, agent.Update{Key: u.Key, Value: u.Value, Revision: u.Revision, Deleted: u.Deleted}) } return out, nil }, Backup: backupNode(session), // What the filling refused before this apply may be named now: the provisioner's client is told to fill // again (ace, 2026-10-25 21:17). Settled: func(module string) { knockFilling(module+" settled", say) judgeWhatRunsNow(opts.state, signer) users.RunsChanged() }, // And once the haves are read at the start, and whenever a module is taken as running without an apply, so // a refusal from the engine's start does not wait for some module to settle (issue 924). RunsChanged: func(why string) { knockFilling(why, say) judgeWhatRunsNow(opts.state, signer) users.RunsChanged() }, Say: say, // A failure of the bus not answering for a moment is tried again: 2 s, 5 s, 15 s, 60 s, then every 5 min. Backoff: link.Backoff, Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result { actx, done := applyContext(session.WithLink(ctx)) // While the grants part's list is followed, the controller's list in the bus module's composition is not // written: the apply keeps the list on the machine (ADR 0547 clause 8, never both). held, herr := users.Hold(c) var res agent.Result if herr != nil { res = failedWith(herr.Error()) } else { res = applyComposition(actx, opts, held, others, ag.Placed(scopeName(c)), ag.RemovedByWant(), sched, say) } done() // The mesh's own certificate it carries is kept for this machine's record (ADR 0555 §3). if !res.Failed { if err := keepAuthority(opts.state, c); err != nil { say("the mesh's certificate in " + c.Module + "'s composition could not be kept for this machine's record: " + err.Error()) } } // An apply of the bus module may write an older list than the newest composition holds: looked at again. if module, _ := users.State(); module != "" && module == c.Module { users.Again() } afterApply() sayFacts(ctx, ag) return res }, Remove: func(ctx context.Context, module string, others []delivery.Composed) agent.Result { actx, done := applyContext(session.WithLink(ctx)) res := removeComposition(actx, opts, module, others, ag.Placed(module), sched, say) done() afterApply() sayFacts(ctx, ag) return res }, } // The node composes its own modules (novox/hq ADR 0547 §2, ADR 0549, issue 1126). run, stopRun := context.WithCancel(linked) // What the node runs and how it opens a sealed value, for the composer to keep the boxes it runs; and the agent // told when the node cannot compose, so each wanted module's have says it. // The agent of this run alone: its composer tells it, never the agent of a run begun since (issue 1126: a pass in // flight at the switch marked the next agent's modules failed). own := &agentNow{} hooks := composerHooks{ running: func() []delivery.Composed { if now := own.get(); now != nil { return runsAsWritten(now.Everything(), opts.state) } return nil }, unseal: sealOpener(opts.state), // And what each module's have says it runs sealed: after a restart the agent has applied nothing yet. // A have the bus did not answer for is said as such, never as one holding no seals (issue 1126, 2026-10-28). held: func(module string) (delivery.Seals, []string, error) { if module == agent.NodeModule { got, _, err := session.ReadNodeHaveErr(linked) return got.SealedAt, got.Sealed, err } got, ok, err := session.ReadHaveErr(linked, module) if !ok { return nil, nil, err } return got.SealedAt, got.Sealed, nil }, refused: func(why string, held map[string]string) { if now := own.get(); now != nil && run.Err() == nil { now.NotComposedEach(run, why, held) // a module held alone marks that module, never the others (issue 1126) } }, // And what it keeps of a module as it runs, for a manifest that does not say it yet: on the module's have (ADR 0601 §4). kept: func(kept map[string][]string) { if now := own.get(); now != nil && run.Err() == nil { now.KeptNow(run, kept) } }, // And the vault's authority beside the controller's, and this node's certificate asked of the vault when a // module here asks for one (ADR 0550 §2). vaultAuthority: trust.VaultAuthority, certify: func(ctx context.Context) error { return certifyThisNode(ctx, opts.state, node, trust, session.AskVault, func() { sayFacts(ctx, ag) }) }, } // And its answer to every placement candidate, placed or not, applying nothing (novox/hq ADR 0559 §4): written into // its own key, `.node`, under candidates. shadowHooks := hooks shadowHooks.refused, shadowHooks.kept, shadowHooks.certify = nil, nil, nil shadowEnv := composerEnv(session, node, say, shadowHooks) shadow := selfcompose.NewShadow(shadowEnv, func(ctx context.Context, answers map[string]delivery.CandidateAnswer) { session.NoteCandidates(ctx, delivery.WithValuesDiffer(answers)) }) go shadow.Keep(run, link.Backoff) // this run's alone: a switch begins a run, and its shadow, again shadowNow.Store(shadow) // what the verb compose-candidate composes a check's candidate with // Its own composer, always (issue 1126, A7): a composition made elsewhere (the controller's, in the node's // composition bucket) is never applied. While the seat's placement holds no key for this node it composes // nothing and keeps running what it runs; a placement it cannot watch stops the agent, said, and it begins again. composer := selfComposer(session, node, hooks, say) machinesNow.Store(composer) // the machines a failing network part points at, by the placements (issue 774) composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does session.NoteComposes(linked, delivery.ComposesOwn, "it composes its own modules, and applies no composition made elsewhere") ag = agent.New(env) own.set(ag) current.set(ag) err = ag.Run(run) stopRun() cancel() current.set(nil) composerNow.Store(nil) shadowNow.Store(nil) if ctx.Err() != nil { return } if time.Since(began) > 30*time.Second { wait = 2 * time.Second } if err != nil { say(fmt.Sprintf("the node agent stopped: %v — beginning again in %s", err, wait)) } select { case <-ctx.Done(): return case <-time.After(wait): } if wait *= 2; wait > 2*time.Minute { wait = 2 * time.Minute } } } // reloadSettles is how long after this engine wrote the bus's user list the bus is taken to have reloaded its accounts: // the bus module's entrypoint looks at the list every 5 s and then signals the server, whose reload of the accounts // answers no JetStream request for about three seconds (on the bus node, 2026-10-25). var reloadSettles = 10 * time.Second // agentNow is the agent running now, which the node's facts are built from. type agentNow struct { mu sync.Mutex ag *agent.Agent } func (c *agentNow) set(a *agent.Agent) { c.mu.Lock() c.ag = a c.mu.Unlock() } func (c *agentNow) get() *agent.Agent { c.mu.Lock() defer c.mu.Unlock() return c.ag } // applyComposition applies one composition: a module's, or the node's own. // // placed is the modules a want still places here besides this one: none of their resources is removed by this apply, // which is refused instead (issue 774). // // removed is the modules whose want says removed: in the node's own apply, a resource of any other module goes only when // its module is among them (issue 774). func applyComposition(ctx context.Context, opts options, c delivery.Composed, others []delivery.Composed, placed, removed map[string]bool, sched *apply.Scheduler, say link.Announce) agent.Result { name := scopeName(c) if c.Module == "" { // What another module's composition declares is that module's, never the node's own apply's (issue 1945). var dropped []string c, dropped = ownWithout(c, others) sayDroppedOnce(dropped) } d, err := declarationOf(c) if err != nil { return failedWith(err.Error()) } if err := refuseEmptyWrites(d); err != nil { return failedWith(err.Error()) } scope := apply.Scope{Module: name, Others: resourcesOf(others), Placed: placed} if name == agent.NodeModule { scope.RemovedByWant = removed if scope.RemovedByWant == nil { scope.RemovedByWant = map[string]bool{} } } got := applyModuleHeld(ctx, opts, d, &scope, sched, say) if got.Refused != "" { return failedWith(got.Refused) } // An image that could not be pulled from a store is said with its holder, for the engine to try the apply again once the // holder has settled (design 51, ENG-8). failedOn := func(why string) agent.Result { res := failedWith(why) res.Pulls = pullsOf(d, got.Err) return res } results := apply.ModuleResults(d, got.Outcome, got.Err) if r, ok := results[name]; ok && r.Failed { return failedOn(r.Detail) } if got.Err != nil { return failedOn(got.Err.Error()) } if got.Probe != nil { return failedWith(got.Probe.partial().Error()) } return agent.Result{} } // removeComposition removes one module's own resources and nothing of another's. func removeComposition(ctx context.Context, opts options, module string, others []delivery.Composed, placed map[string]bool, sched *apply.Scheduler, say link.Announce) agent.Result { scope := apply.Scope{Module: module, Others: resourcesOf(others), Placed: placed} got := applyModuleHeld(ctx, opts, &declaration.Declaration{Version: declaration.Version}, &scope, sched, say) switch { case got.Refused != "": return agent.Result{Failed: true, Detail: shorten(got.Refused)} case got.Err != nil: return agent.Result{Failed: true, Detail: shorten(got.Err.Error())} } return agent.Result{} } // busyModules are the modules whose last apply left a container running because it said busy. func busyModules() []string { busyMu.Lock() defer busyMu.Unlock() var out []string for m, w := range busyBy { if w && m != "" { out = append(out, m) } } sort.Strings(out) return out } // machinesNow is the composer whose placements name the mesh's machines by address, for the health report's targets // (writeHealthReport); nil on a node that does not compose its own, whose parts then name no target. var machinesNow atomic.Pointer[selfcompose.Composer] // composerNow is the composer of the run that composes the node's own now, nil while the node applies the controller's // compositions or no run is going: what the verb compose asks a pass of (issue 774). var composerNow atomic.Pointer[selfcompose.Composer] // shadowNow is the shadow of the run now, which answers placement candidates and the verb compose-candidate; nil // between runs. var shadowNow atomic.Pointer[selfcompose.Shadow] // keptRetired says the kept declaration is read no more: every module the seat's placement puts on this node has a have of // its own (placedAllHave), so what runs is the agent's alone (ADR 0550 §2, design 47's L3 s9). var keptRetired atomic.Bool // errKeptRetired is keptDeclaration's answer once the read is retired. var errKeptRetired = errors.New("the kept declaration is read no more: every module placed on this node has a have of its own") // keptDeclaration is the declaration the node kept before the switch, verified with the mesh's key when it is given; // once the read is retired, nothing is read. func keptDeclaration(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) { if keptRetired.Load() { return nil, errKeptRetired } var raw []byte if len(signer) > 0 { var err error if raw, err = store.LoadDeclared(store.DeclaredPath(statePath), signer); err != nil { return nil, err } } else { kept, err := store.ReadDeclared(store.DeclaredPath(statePath)) if err != nil { return nil, err } raw = kept.Declaration } return declaration.Parse(raw) } // whatRuns is what runs on this node as the declaration of it (novox/hq ADR 0539 §9): the compositions the agent runs // and, for every module the agent does not hold yet, what the kept declaration has for it, with the node's own parts // (adoption, identities, groups, libraries) from the node's composition once the agent holds it and from the kept // declaration until then. What the node judges its health by, and what the filling, the take act and the judgement of // directories read. func whatRuns(everything []delivery.Composed, holdsNode bool, kept func() (*declaration.Declaration, error)) (*declaration.Declaration, error) { d, err := declarationOf(everything...) if err != nil { return nil, err } if holdsNode { return d, nil } old, err := kept() if err != nil { if len(everything) > 0 { return d, nil } return nil, err } d.Adoption, d.Identities, d.Groups, d.Libraries = old.Adoption, old.Identities, old.Groups, old.Libraries d.Provisioner, d.Builds = old.Provisioner, old.Builds d.Filled = append(d.Filled, old.Filled...) return d, nil } // runsHere says what runs on this node; the running serve sets it to the agent's view, and a command run by hand reads // the kept declaration. var runsHere struct { mu sync.Mutex fn func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) } func setRunsHere(fn func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error)) { runsHere.mu.Lock() runsHere.fn = fn runsHere.mu.Unlock() } // declaredNow is the declaration of what runs on this node now. func declaredNow(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) { runsHere.mu.Lock() fn := runsHere.fn runsHere.mu.Unlock() if fn != nil { return fn(statePath, signer) } return keptDeclaration(statePath, signer) } // runsFromAgent is what runs now as the agent in hand sees it. func runsFromAgent(current *agentNow) func(string, ed25519.PublicKey) (*declaration.Declaration, error) { return func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) { kept := func() (*declaration.Declaration, error) { return keptDeclaration(statePath, signer) } ag := current.get() if ag == nil { return kept() } return whatRuns(ag.Everything(), ag.HoldsNode(), kept) } } // refuseEmptyWrites refuses a declaration that names a file the provisioner fills (owned at runtime, no content, no // bytes, no digest, no sealed value) without saying it is filled: applied, the file would be written empty over what the // provisioner wrote (issue 1945, live on 2026-10-25 at 01:22 on ace, g14 and shanks). The apply fails, naming the file, // and nothing is written. func refuseEmptyWrites(d *declaration.Declaration) error { filled := make(map[string]bool, len(d.Filled)) for _, p := range d.Filled { filled[p] = true } var refused []string for _, r := range d.Resources { f, ok := r.(*declaration.File) if !ok || f.OwnerAtRuntime != declaration.RuntimeOwned || filled[f.Path] { continue } if f.Content == "" && f.Bytes == "" && f.Digest == "" && f.Sealed == "" && f.VaultSealed == nil { refused = append(refused, fmt.Sprintf("%s (%s)", f.ID, f.Path)) } } if len(refused) > 0 { return fmt.Errorf("the composition names %s with no content and does not say the provisioner fills it, so it "+ "would be written empty; nothing was applied", strings.Join(refused, ", ")) } return nil } // recordedIn says the engine's applied state holds a record of every resource of a composition: what the agent believes // of a have or the kept declaration only when it does, so a resource removed since is made again (issue 1945). A state // that cannot be read says yes: the agent then believes what it was told, as before. func recordedIn(statePath string, say func(string)) func(ctx context.Context, c delivery.Composed) bool { return func(_ context.Context, c delivery.Composed) bool { known, err := store.Load(statePath) if err != nil { if say != nil { say(fmt.Sprintf("the applied state cannot be read, so what runs is believed as told: %v", err)) } return true } filled := make(map[string]bool, len(c.Filled)) for _, p := range c.Filled { filled[p] = true } for _, raw := range c.Resources { r, err := declaration.ParseResource(raw) if err != nil { return true } // A file the provisioner client fills is left to it, and has no record until it writes it. if f, ok := r.(*declaration.File); ok && filled[f.Path] { continue } if _, ok := known.Find(r.Identity()); !ok { return false } } return true } } // declaresNothingToBackUp reads node-backup.backed-up's answer for a module: true when this machine backs nothing of it // up (no runs, no paths), and whether the answer could be read at all. func declaresNothingToBackUp(said json.RawMessage, module string) (nothing bool, read bool) { var list []struct { Module string `json:"module"` Runs int `json:"runs"` Paths []string `json:"paths"` } if err := json.Unmarshal(said, &list); err != nil { return false, false } for _, m := range list { if m.Module == module && (m.Runs > 0 || len(m.Paths) > 0) { return false, true } } return true, true } // applyBound is how long one apply or removal may take: long enough for the slowest image pull and backup step, short // enough that a step that hangs cannot hold the node agent for ever. const applyBound = 30 * time.Minute // applyContext is the context an apply or a removal runs in: the link's values, never its end, and a bound of its own. // The link's context ends when the bus goes, and the bus may go because of the apply itself (it replaces the bus's own // container), which must then finish: a container removed and never made again took the mesh down (issue 1945, live on // 2026-10-25 at 11:23). applyBound keeps a hung step from blocking the agent for ever. func applyContext(ctx context.Context) (context.Context, context.CancelFunc) { return context.WithTimeout(context.WithoutCancel(ctx), applyBound) } // composerHooks are what the composer reads of the agent and the machine (selfcompose.Env Running, Unseal, Refused). type composerHooks struct { running func() []delivery.Composed unseal func(string) ([]byte, error) refused func(why string, held map[string]string) kept func(kept map[string][]string) held func(module string) (delivery.Seals, []string, error) // vaultAuthority and certify are selfcompose.Env's VaultAuthority and Certify. vaultAuthority func() string certify func(ctx context.Context) error } // selfComposer is the node's own composer: it composes what the seat's placement places on the node, and nothing while // the placement holds no key for it (issue 1126). func selfComposer(session *link.Session, node string, hooks composerHooks, say func(string)) *selfcompose.Composer { env := composerEnv(session, node, say, hooks) // And it keeps each placed module's data items in its own state data, which the node's disk-load watches (novox/hq // ADR 0577 clause 1); the shadow composes for candidates and keeps none. env.Data = link.EngineData{Session: session} // And the packages its compositions declare in its own key of the state packages, which pacman reads (novox/hq ADR // 0565 §6). env.Packages = link.EnginePackages{Session: session} // And every composition it makes in its composition state composed-, a record the provisioner, the assigner // and the view read and nothing applies (novox/hq ADR 0593 §6). env.ComposedState = link.EngineComposed{Session: session, Node: node} return selfcompose.New(env) } // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer // to placement candidates (selfcompose.Shadow) alike. func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { return selfcompose.Env{Node: node, Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) { if bucket == delivery.PlacementBucket || bucket == delivery.PlacementCandidatesBucket { // The seat's states are made when their holder starts: a node-engine that starts first waits for the // bus's word that the bucket was created, and the shadow and the composer say they wait. return bucketWatch{session, bucket}.WatchAwait(ctx, pattern, func(err error) { say("the seat's state " + bucket + " is not on the bus yet (" + err.Error() + "): it is watched when the bus says it was created") }) } return bucketWatch{session, bucket}.Watch(ctx, pattern) }, Get: func(ctx context.Context, bucket, key string) ([]byte, error) { h, err := session.KVHistory(ctx, bucket, key) if err != nil || len(h) == 0 || h[len(h)-1].Deleted { return nil, err } return h[len(h)-1].Value, nil }, Mesh: frombus.Mesh{InternalSuffix: overlay.Suffix()}, Say: say, // Its modules' own secrets, sealed by the vault to this machine's key (ADR 0550 §5). Own: func(ctx context.Context, args map[string]any) (json.RawMessage, error) { return session.AskVault(ctx, "own", args) }, // And the vault's word that a secret changed, on which it composes again (ADR 0547 §2). Hear: session.Hear, // And a moved module's templates archive, in the artifact store, as every archive is fetched (issue 1126). Fetch: apply.Fetch, Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held, Kept: hooks.kept, VaultAuthority: hooks.vaultAuthority, Certify: hooks.certify} }