jschoubben
653ff61662
A new step, internal/ownresync, after each shadow answer: for each own secret masked as another value, while the node applies the controller's compositions and runs the one its want names, it asks the vault's move-in with replace_running: true and the running value sealed to the call's key, once per value, and says what it asked and what the vault answered. The answer written to the node's have key carries values_differ. Pins the group's mesh-sdk head 1062b33 (go.sum is filled once that head is pushed).
989 lines
38 KiB
Go
989 lines
38 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ed25519"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
|
"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/link"
|
|
"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
|
|
}
|
|
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, nil
|
|
}
|
|
|
|
// 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 {
|
|
got, ok := h.Read(ctx, module)
|
|
return got, ok, 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,
|
|
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.
|
|
if said, err := session.AskSeat(ctx, "node-backup", "backed-up", map[string]any{"module": module}, 30*time.Second); 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 := session.AskSeat(ctx, "node-backup", "now", map[string]any{"module": module}, 30*time.Second); err != nil {
|
|
return agent.Backup{Words: err.Error()}
|
|
}
|
|
return awaitBackup(ctx, events, session.Membership.Node, module, asked, backupWithin)
|
|
}
|
|
}
|
|
|
|
// 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: "the node agent was stopped while the backup ran"}
|
|
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 "<id> (<module>)".
|
|
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
|
|
}
|
|
|
|
// operatorKeysFile is the file beside the node's state that holds the operator's public keys, one ed25519 key in base64 a
|
|
// line (a line that starts with # is a note). The switch's last declaration writes it as a file resource, beside the
|
|
// controller's key the node was enrolled with (novox/hq ADR 0539 §3, §11); a node adopted later has it at its adoption.
|
|
const operatorKeysFile = "operator-keys"
|
|
|
|
// trustedKeys are the keys a composition may be signed by: the controller's, which the node was enrolled with, and the
|
|
// operator's, read from operatorKeysFile each time it is asked, so a key that arrives while the agent runs is trusted at
|
|
// the next composition. A file that cannot be read trusts the controller's alone, and says so.
|
|
func trustedKeys(opts options, controller ed25519.PublicKey) func() []ed25519.PublicKey {
|
|
return func() []ed25519.PublicKey {
|
|
keys := []ed25519.PublicKey{}
|
|
if len(controller) == ed25519.PublicKeySize {
|
|
keys = append(keys, controller)
|
|
}
|
|
raw, err := os.ReadFile(filepath.Join(filepath.Dir(opts.state), operatorKeysFile))
|
|
if err != nil {
|
|
return keys
|
|
}
|
|
for _, line := range strings.Split(string(raw), "\n") {
|
|
line = strings.TrimSpace(line)
|
|
if line == "" || strings.HasPrefix(line, "#") {
|
|
continue
|
|
}
|
|
key, err := base64.StdEncoding.DecodeString(line)
|
|
if err != nil || len(key) != ed25519.PublicKeySize {
|
|
fmt.Fprintf(os.Stderr, "mesh-host: a line of %s is not an ed25519 public key in base64, and is not trusted\n", operatorKeysFile)
|
|
continue
|
|
}
|
|
keys = append(keys, ed25519.PublicKey(key))
|
|
}
|
|
return keys
|
|
}
|
|
}
|
|
|
|
// 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.
|
|
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() {
|
|
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, 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
|
|
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, written alone from the newest composition of the bus module (ADR 0547 §8): on every
|
|
// node, and a node whose compositions carry no list writes nothing.
|
|
users := userListFollower(opts, session, composedBucket, trustedKeys(opts, signer), 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()
|
|
}
|
|
})
|
|
})
|
|
// A vault key heard (or heard again changed) looks at the list again, so its hashes are handed (ADR 0547 §9).
|
|
vaultKey.following(users.Again)
|
|
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),
|
|
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
|
|
},
|
|
Trusted: trustedKeys(opts, signer),
|
|
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, say) },
|
|
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, 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, sched, say)
|
|
done()
|
|
afterApply()
|
|
sayFacts(ctx, ag)
|
|
return res
|
|
},
|
|
}
|
|
// The node composes its own modules once the catalogue places it (novox/hq ADR 0547 §2, ADR 0549, issue 1126): its
|
|
// placement key is read at the agent's start; until then it applies the controller's compositions as before.
|
|
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 now.Everything()
|
|
}
|
|
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.
|
|
held: func(module string) (delivery.Seals, []string) {
|
|
if module == agent.NodeModule {
|
|
got, _ := session.ReadNodeHave(linked)
|
|
return got.SealedAt, got.Sealed
|
|
}
|
|
got, ok := session.ReadHave(linked, module)
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
return got.SealedAt, got.Sealed
|
|
},
|
|
refused: func(why string) {
|
|
if now := own.get(); now != nil && run.Err() == nil {
|
|
now.NotComposed(run, why)
|
|
}
|
|
},
|
|
}
|
|
// And its answer to every placement candidate, placed or not, applying nothing (novox/hq ADR 0559 §4): written into
|
|
// its own key, `<node>.node`, under candidates.
|
|
shadowHooks := hooks
|
|
shadowHooks.refused = nil
|
|
shadowEnv := composerEnv(session, node, say, shadowHooks)
|
|
// And, after each answer, what runs of an own secret the answer says the vault holds another value of, handed to
|
|
// the vault while the node applies the controller's compositions (issue 1126, cause C4; ownresync).
|
|
composes := &composesNow{}
|
|
resync := ownResync(session, node, say, shadowEnv, composes)
|
|
shadow := selfcompose.NewShadow(shadowEnv,
|
|
func(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
|
|
session.NoteCandidates(ctx, delivery.WithValuesDiffer(answers))
|
|
go resync.take(ctx, answers)
|
|
})
|
|
go shadow.Keep(run, link.Backoff) // this run's alone: a switch begins a run, and its shadow, again
|
|
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 {
|
|
composes.set(delivery.ComposesOwn)
|
|
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
|
|
// placement key deleted ends this run, the way back (issue 1126, the seed's revert).
|
|
go endWhenUnplaced(run, session, node, stopRun, say)
|
|
} else {
|
|
composes.set(delivery.ComposesController)
|
|
go resync.again(run)
|
|
session.NoteComposes(linked, delivery.ComposesController, why)
|
|
// And once the catalogue places it, the agent begins again composing its own: the placement key appearing ends
|
|
// this run, so a node does not wait for a restart to follow the seed of its placement (ADR 0549 §4).
|
|
go endWhenPlaced(run, session, node, stopRun, say)
|
|
}
|
|
ag = agent.New(env)
|
|
own.set(ag)
|
|
current.set(ag)
|
|
err = ag.Run(run)
|
|
stopRun()
|
|
cancel()
|
|
current.set(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.
|
|
func applyComposition(ctx context.Context, opts options, c delivery.Composed, others []delivery.Composed,
|
|
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)}
|
|
got := applyModuleHeld(ctx, opts, d, &scope, sched, say)
|
|
if got.Refused != "" {
|
|
return failedWith(got.Refused)
|
|
}
|
|
results := apply.ModuleResults(d, got.Outcome, got.Err)
|
|
if r, ok := results[name]; ok && r.Failed {
|
|
return failedWith(r.Detail)
|
|
}
|
|
if got.Err != nil {
|
|
return failedWith(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,
|
|
sched *apply.Scheduler, say link.Announce) agent.Result {
|
|
scope := apply.Scope{Module: module, Others: resourcesOf(others)}
|
|
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
|
|
}
|
|
|
|
// keptDeclaration is the declaration the node kept before the switch, verified with the mesh's key when it is given.
|
|
func keptDeclaration(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) {
|
|
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)
|
|
}
|
|
|
|
// selfComposer is the node's own composer when the catalogue places the node (its key `placement.<node>` in the
|
|
// indexer's state), and nil, said once, while it does not or the key cannot be read.
|
|
// 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(string)
|
|
held func(module string) (delivery.Seals, []string)
|
|
}
|
|
|
|
func selfComposer(ctx context.Context, session *link.Session, node string, say func(string), hooks composerHooks) (*selfcompose.Composer, string) {
|
|
key, err := delivery.PlacementKey(node)
|
|
if err != nil {
|
|
say("this node applies the controller's compositions: " + err.Error())
|
|
return nil, err.Error()
|
|
}
|
|
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"
|
|
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(env), why
|
|
}
|
|
|
|
// 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) {
|
|
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}
|
|
}
|
|
|
|
// endWhenUnplaced calls stop once this node's placement key is deleted from the indexer's placement (after the watch
|
|
// caught up, or gone already), so the agent begins again and applies the controller's compositions, as a node the
|
|
// catalogue does not place does from its start: the way back (issue 1126, the seed's revert). A watch that cannot be
|
|
// made is said; the node then returns at its next start.
|
|
func endWhenUnplaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) {
|
|
key, err := delivery.PlacementKey(node)
|
|
if err != nil {
|
|
return
|
|
}
|
|
updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key)
|
|
if err != nil {
|
|
say("whether the catalogue still places this node is not followed (" + err.Error() + "): it returns to the controller's compositions at its next start")
|
|
return
|
|
}
|
|
placed := false
|
|
for u := range updates {
|
|
switch {
|
|
case u.Caught && placed:
|
|
continue
|
|
case u.Caught || u.Deleted || len(u.Value) == 0:
|
|
say("the catalogue no longer places this node (" + key + "): the node agent begins again, applying the controller's compositions")
|
|
stop()
|
|
return
|
|
default:
|
|
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
|
|
// caught up, or one already there), so the agent begins again and composes its own. A watch that cannot be made is said;
|
|
// the node then switches at its next start.
|
|
func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) {
|
|
key, err := delivery.PlacementKey(node)
|
|
if err != nil {
|
|
return
|
|
}
|
|
updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key)
|
|
if err != nil {
|
|
say("whether the catalogue places this node is not followed (" + err.Error() + "): it composes its own from its next start")
|
|
return
|
|
}
|
|
for u := range updates {
|
|
if !u.Caught && !u.Deleted && len(u.Value) > 0 {
|
|
say("the catalogue places this node now (" + key + "): the node agent begins again, composing its own modules")
|
|
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")
|
|
}
|
|
}
|