Files
mesh-host/internal/agent/agent.go
T
jschoubben 1cdb4d1a29 Say a module found running on its own have again at the agent's start, so its health is this engine's and not an older engine's unknown
xorg's have on g14 and shanks kept the unknown an engine before 958ccbd wrote: the
baseline believed the have and never handed the module to the session, so no look
rewrote its health, and the assigner held every module requiring x11-display back.
2026-10-25 15:27:23 +01:00

831 lines
27 KiB
Go

// Package agent is the node agent (novox/hq ADR 0429 §4, ADR 0539): the node-engine's own process that follows the
// node's wants and applies one module at a time.
//
// **It watches, it never polls.** A want `want.<node>.<module>` names a fingerprint and the digest of the one composition
// the assigner allows the node to apply for it; the composition itself waits under `<node>.<module>.<fingerprint>` in
// the node's bucket `composed-<node>`. When both are there and the composition's digest is the want's, and what runs
// differs, the agent applies that module's composition and writes its have with the fingerprint and the digest. A want
// `removed` removes that module's own resources. Nothing is applied for a composition no want names, and a failed apply
// is retried by the next change of either key, never by a clock (research 063 §9.1).
package agent
import (
"bytes"
"context"
"crypto/ed25519"
"encoding/json"
"fmt"
"sort"
"strings"
"sync"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// Update is one change of a key in a watched bucket. Caught marks the end of the keys that were there when the watch
// began; every Update after it is a change.
type Update struct {
Key string
Value []byte
Revision uint64
Deleted bool
Caught bool
}
// Watcher follows one bucket: the keys matching the pattern now, a Caught update, then every change, until ctx ends.
type Watcher interface {
Watch(ctx context.Context, pattern string) (<-chan Update, error)
}
// NodeModule is the scope of the node's own composition, `<node>.node`.
const NodeModule = "node"
// Have is what a module's have key says, as the agent reads and writes it.
type Have struct {
Module string
FP string
Composed string
// Phase is "running", "failed", "held" or "removed"; empty reads as running.
Phase string
Detail string
}
// Haves is the node's have bucket as the agent uses it.
type Haves interface {
// Read is what the node-engine last wrote for a module, and whether it wrote anything.
Read(ctx context.Context, module string) (Have, bool)
// Write writes a module's have where it differs from the bucket's.
Write(ctx context.Context, h Have)
// Removed writes the module's have with the phase removed, once, if a key of it exists.
Removed(ctx context.Context, module string)
}
// Result is how one apply or removal came out.
type Result struct {
Failed bool
Detail string
}
// Backup is the answer of the backup asked before a module moves.
type Backup struct {
Done bool
Words string
}
// Env is everything the agent needs of the node.
type Env struct {
Node string
Wants Watcher
Composed Watcher
Haves Haves
// Apply applies one module's composition (scope NodeModule for the node's own), given the compositions the node's
// other modules run so a path, unit or group they hold is not removed or taken. It is called with one module at a
// time.
Apply func(ctx context.Context, c delivery.Composed, others []delivery.Composed) Result
// Remove removes the module's own resources and nothing of another's.
Remove func(ctx context.Context, module string, others []delivery.Composed) Result
// Backup backs a module up before a new composition of it is applied; nil backs up nothing.
Backup func(ctx context.Context, module string) Backup
// Kept is the resources the node last applied, by module, from the declaration it kept before the switch (ADR 0539
// §9), as JSON; the node's own are under NodeModule. Nil when the node kept none. A module the agent does not hold yet
// still runs from it.
Kept func() map[string][]json.RawMessage
// KeptFilled is the paths the kept declaration names as filled by the provisioner, by the module whose file is at
// each (the node's own under NodeModule). A module that still runs from the kept declaration keeps its filled files
// filled: without them the filling refused each one as "not named as filled" (issue 1945). Nil names none.
KeptFilled func() map[string][]string
// Trusted is the public keys a composition may be signed by: the controller's and the operator's (ADR 0539 §3). A
// composition unsigned, signed by another key, or changed after it was signed is refused and nothing of it is
// applied. Nil trusts no key, so nothing is applied.
Trusted func() []ed25519.PublicKey
// Recorded says the engine holds a record of every resource of the composition, so it was applied and not removed
// since; nil says yes. A composition the kept declaration or a have says runs is applied again when it says no.
Recorded func(ctx context.Context, c delivery.Composed) bool
// Revisions are a composition key's revisions the bucket keeps, oldest first; nil reads none. The composition a want
// names is taken from them when a newer one replaced it under the same key (ADR 0539 clause 1).
Revisions func(ctx context.Context, key string) ([]Update, error)
Say func(string)
}
// Agent follows the wants.
type Agent struct {
env Env
mu sync.Mutex
wants map[string]wantNow // by module
comps map[string]compNow // by composed key
running map[string]delivery.Have
// applied is the composition each module was last applied from: a revision a want named from the key's history is
// not the key's latest value, and what it holds is still held out of the others' applies.
applied map[string]compNow
// attempt is the last apply of a module that failed or is held, with the revisions it was made from.
attempt map[string]attempt
checked map[string]bool
// prior is a module the node ran before this agent started, so it has something to back up.
prior map[string]bool
said map[string]string
retry chan []string
}
type wantNow struct {
Want delivery.Want
Node bool // the node's own want
Present bool
Rev uint64
}
type compNow struct {
Raw []byte
C delivery.Composed
Dig string
Rev uint64
Err error
}
type attempt struct {
FP, Composed string
WantRev, CompRev uint64
Held bool
AnsweredRev uint64
}
// New is an agent for a node.
func New(env Env) *Agent {
if env.Say == nil {
env.Say = func(string) {}
}
return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{},
attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{},
retry: make(chan []string, 8)}
}
// Run follows the wants until ctx ends. The keys there when it starts are taken first, the node's own composition
// before any module's, then each change as it comes.
func (a *Agent) Run(ctx context.Context) error {
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)
}
comps, err := a.env.Composed.Watch(ctx, a.env.Node+".>")
if err != nil {
return fmt.Errorf("the agent cannot watch its compositions: %w", err)
}
var wantsCaught, compsCaught bool
dirty := map[string]bool{}
for {
select {
case <-ctx.Done():
return nil
case modules := <-a.retry:
// Applied again though the keys did not change: a container that waited because it said busy (ADR 0360
// §11) is replaced once it is idle.
a.mu.Lock()
for _, m := range modules {
delete(a.running, m)
delete(a.attempt, m)
dirty[m] = true
}
a.mu.Unlock()
case u, ok := <-wants:
if !ok {
return fmt.Errorf("the watch of the wants ended")
}
if u.Caught {
wantsCaught = true
} else if m := a.takeWant(u); m != "" {
dirty[m] = true
}
case u, ok := <-comps:
if !ok {
return fmt.Errorf("the watch of the compositions ended")
}
if u.Caught {
compsCaught = true
} else if m := a.takeComposition(u); m != "" {
dirty[m] = true
// The node's own may wait for this one to read (unreadableModules): looked at again with it.
dirty[NodeModule] = true
}
}
// Nothing is applied until what was there at the start is all in: a module's composition may arrive after its
// want, and the node's own composition comes before any module's.
if !(wantsCaught && compsCaught) || len(dirty) == 0 {
continue
}
a.settle(ctx, dirty)
dirty = map[string]bool{}
}
}
// Retry asks for the modules to be applied again from the keys they have, though nothing changed: what ADR 0360 §11
// needs for a container that waits because it said busy. It never blocks; a retry asked while eight wait is dropped, and
// asked again by its caller's next look.
func (a *Agent) Retry(modules ...string) {
select {
case a.retry <- modules:
default:
}
}
// Running is the compositions the node runs now: each module's that the agent applied or found already running, and
// the node's own. What the node judges its health by and says of itself is built from them.
func (a *Agent) Running() []delivery.Composed {
a.mu.Lock()
defer a.mu.Unlock()
mods := make([]string, 0, len(a.running))
for m := range a.running {
mods = append(mods, m)
}
sort.Strings(mods)
var out []delivery.Composed
for _, m := range mods {
h := a.running[m]
if h.Phase == delivery.PhaseRemoved {
continue
}
if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed {
out = append(out, c.C)
} else if c, ok := a.applied[m]; ok && c.Dig == h.Composed {
out = append(out, c.C)
}
}
return out
}
// takeWant takes a change of a want key and returns the module it concerns, or "" for a key that is not a want.
func (a *Agent) takeWant(u Update) string {
a.mu.Lock()
defer a.mu.Unlock()
if _, err := delivery.ParseWantNodeKey(u.Key); err == nil {
w := wantNow{Node: true, Rev: u.Revision}
if !u.Deleted {
nw, err := delivery.DecodeNodeWant(u.Value)
if err != nil {
a.sayOnce("want:node", fmt.Sprintf("the node's own want is not one: %v", err))
return ""
}
w.Present, w.Want = true, delivery.Want{FP: "", Composed: nw.Composed}
}
a.wants[NodeModule] = w
return NodeModule
}
node, module, err := delivery.ParseWantKey(u.Key)
if err != nil || node != a.env.Node {
return ""
}
w := wantNow{Rev: u.Revision}
if !u.Deleted {
want, err := delivery.DecodeWant(u.Value)
if err != nil {
a.sayOnce("want:"+module, fmt.Sprintf("the want of %s is not one: %v", module, err))
return ""
}
w.Present, w.Want = true, want
}
a.wants[module] = w
return module
}
// takeComposition takes a change of a composition key and returns the module it concerns.
func (a *Agent) takeComposition(u Update) string {
node, module, _, isNode, err := delivery.ParseComposedKey(u.Key)
if err != nil || node != a.env.Node {
return ""
}
a.mu.Lock()
defer a.mu.Unlock()
if isNode {
module = NodeModule
}
if u.Deleted {
delete(a.comps, u.Key)
return module
}
now := compNow{Raw: u.Value, Rev: u.Revision}
c, err := delivery.DecodeComposed(u.Value)
if err == nil {
var trusted []ed25519.PublicKey
if a.env.Trusted != nil {
trusted = a.env.Trusted()
}
// The signature before anything of the composition is used: its digest is what the want names.
now.C = c
now.Dig, err = delivery.VerifyComposed(c, trusted)
if err != nil {
now.C = delivery.Composed{}
a.sayOnce("verify:"+u.Key, fmt.Sprintf("the composition %s is refused and nothing of it is applied: %v", u.Key, err))
err = fmt.Errorf("refused: %w", err)
}
} else {
a.sayOnce("decode:"+u.Key, fmt.Sprintf("the composition %s cannot be read, and nothing of it is applied: %v", u.Key, err))
}
now.Err = err
a.comps[u.Key] = now
return module
}
func (a *Agent) sayOnce(key, line string) {
if a.said[key] == line {
return
}
a.said[key] = line
a.env.Say(line)
}
// settle applies what the dirty modules' wants allow: the node's own first, then each module in name order.
func (a *Agent) settle(ctx context.Context, dirty map[string]bool) {
order := make([]string, 0, len(dirty))
for m := range dirty {
order = append(order, m)
}
sort.Slice(order, func(i, j int) bool {
if (order[i] == NodeModule) != (order[j] == NodeModule) {
return order[i] == NodeModule
}
return order[i] < order[j]
})
for _, m := range order {
if ctx.Err() != nil {
return
}
a.settleOne(ctx, m)
}
}
func (a *Agent) compositionFor(module, fp string) (compNow, bool) {
var key string
var err error
if module == NodeModule {
key, err = delivery.ComposedNodeKey(a.env.Node)
} else {
key, err = delivery.ComposedKey(a.env.Node, module, fp)
}
if err != nil {
return compNow{}, false
}
c, ok := a.comps[key]
return c, ok
}
// revisionNamed is the revision of a module's composition key whose digest is the one given, signed by a trusted key,
// from the revisions the bucket keeps; false when none is.
func (a *Agent) revisionNamed(ctx context.Context, module, fp, digest string) (compNow, bool) {
if a.env.Revisions == nil {
return compNow{}, false
}
var key string
var err error
if module == NodeModule {
key, err = delivery.ComposedNodeKey(a.env.Node)
} else {
key, err = delivery.ComposedKey(a.env.Node, module, fp)
}
if err != nil {
return compNow{}, false
}
revs, err := a.env.Revisions(ctx, key)
if err != nil {
a.mu.Lock()
a.sayOnce("history:"+key, fmt.Sprintf("the revisions of %s cannot be read: %v", key, err))
a.mu.Unlock()
return compNow{}, false
}
var trusted []ed25519.PublicKey
if a.env.Trusted != nil {
trusted = a.env.Trusted()
}
for i := len(revs) - 1; i >= 0; i-- {
u := revs[i]
if u.Deleted || u.Key != key {
continue
}
c, err := delivery.DecodeComposed(u.Value)
if err != nil {
continue
}
dig, err := delivery.VerifyComposed(c, trusted)
if err != nil || dig != digest {
continue
}
return compNow{Raw: u.Value, C: c, Dig: dig, Rev: u.Revision}, true
}
return compNow{}, false
}
// others are what the node's other modules run now, so what they hold is not taken: the compositions the agent runs and,
// for every module the agent does not hold yet, the resources the kept declaration has for it (ADR 0429 §4).
func (a *Agent) others(except string) []delivery.Composed {
var out []delivery.Composed
mods := make([]string, 0, len(a.running))
for m := range a.running {
if m != except {
mods = append(mods, m)
}
}
sort.Strings(mods)
for _, m := range mods {
h := a.running[m]
if h.Phase == delivery.PhaseRemoved {
continue
}
if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed {
out = append(out, c.C)
} else if c, ok := a.applied[m]; ok && c.Dig == h.Composed {
out = append(out, c.C)
}
}
if a.env.Kept != nil {
kept := a.env.Kept()
names := make([]string, 0, len(kept))
for m := range kept {
if _, held := a.running[m]; !held && m != except {
names = append(names, m)
}
}
sort.Strings(names)
for _, m := range names {
c := delivery.Composed{Node: a.env.Node, Module: m, Resources: kept[m]}
if a.env.KeptFilled != nil {
c.Filled = a.env.KeptFilled()[m]
}
if m == NodeModule {
c.Module = ""
}
out = append(out, c)
}
}
return out
}
// unreadableModules are the keys of the module compositions in the node's bucket that cannot be read or are refused, of a
// module the assigner has not removed: for each module only the composition its want names, else its newest, since a
// stale one nobody applies (written before a rule the SDK now holds) must not hold the node's own back. Called holding
// a.mu.
func (a *Agent) unreadableModules() []string {
type pick struct {
key string
c compNow
}
current := map[string]pick{}
for key, c := range a.comps {
_, m, fp, isNode, err := delivery.ParseComposedKey(key)
if err != nil || isNode {
continue
}
w := a.wants[m]
if w.Present && w.Want.Removed {
continue
}
if w.Present && w.Want.FP != "" {
if w.Want.FP == fp {
current[m] = pick{key, c}
}
continue
}
if prev, ok := current[m]; !ok || c.Rev > prev.c.Rev {
current[m] = pick{key, c}
}
}
var out []string
for _, p := range current {
if p.c.Err != nil {
out = append(out, p.key)
}
}
sort.Strings(out)
return out
}
// heldOut is what an apply or a removal holds out: others, and every module whose own composition the node's bucket holds
// and the agent does not run yet. A module that left the node's own composition for one of its own keeps its resources
// until that one is applied, so the node's own apply never removes them in between (issue 1945, live on 2026-10-25:
// shanks lost systemd-resolved's guard while its composition waited for a credential). The composition its want names,
// else its newest. Not part of Everything: what has not been applied is not what runs.
func (a *Agent) heldOut(except string) []delivery.Composed {
out := a.others(except)
held := map[string]bool{}
for _, c := range out {
held[c.Module] = true
}
waiting := map[string]compNow{}
for key, c := range a.comps {
_, m, fp, isNode, err := delivery.ParseComposedKey(key)
if err != nil || isNode || c.Err != nil || m == except || held[m] {
continue
}
if h, ok := a.running[m]; ok && h.Phase != delivery.PhaseRemoved {
continue
}
if w := a.wants[m]; w.Present && w.Want.Removed {
continue // a module the assigner removed holds nothing back
}
if w := a.wants[m]; w.Present && !w.Want.Removed && w.Want.FP == fp {
waiting[m] = c
continue
}
if w := a.wants[m]; w.Present && !w.Want.Removed && waiting[m].Rev != 0 && waiting[m].C.FP == w.Want.FP {
continue
}
if prev, ok := waiting[m]; !ok || c.Rev > prev.Rev {
waiting[m] = c
}
}
names := make([]string, 0, len(waiting))
for m := range waiting {
names = append(names, m)
}
sort.Strings(names)
for _, m := range names {
out = append(out, waiting[m].C)
}
return out
}
// Everything is what the node runs: the compositions the agent runs, and the kept declaration's resources for each module
// the agent does not hold yet. What the node judges its health by, and what the filling, the take act and the judgement
// of directories read, is built from it.
func (a *Agent) Everything() []delivery.Composed {
a.mu.Lock()
defer a.mu.Unlock()
return a.others("\x00")
}
// HoldsNode says whether the agent runs the node's own composition, so the node's parts are its and not the kept
// declaration's.
func (a *Agent) HoldsNode() bool {
a.mu.Lock()
defer a.mu.Unlock()
h, ok := a.running[NodeModule]
return ok && h.Phase != delivery.PhaseRemoved
}
func (a *Agent) settleOne(ctx context.Context, module string) {
a.mu.Lock()
w := a.wants[module]
a.mu.Unlock()
if !w.Present {
return
}
if w.Want.Removed {
a.remove(ctx, module, w)
return
}
a.mu.Lock()
comp, have := a.compositionFor(module, w.Want.FP)
a.mu.Unlock()
if (!have || comp.Err != nil || comp.Dig != w.Want.Composed) && w.Want.Composed != "" {
// The composition the want names may be an older revision of the key, replaced by a newer one at the same
// fingerprint (a want held there keeps its digest): taken from the revisions the bucket keeps (ADR 0539 clause 1).
if old, found := a.revisionNamed(ctx, module, w.Want.FP, w.Want.Composed); found {
comp, have = old, true
}
}
if !have || comp.Err != nil {
return // waits for the composition; a change of its key settles the module again
}
if comp.Dig != w.Want.Composed {
// A composition whose digest no want names is not applied.
a.mu.Lock()
a.sayOnce("digest:"+module, fmt.Sprintf("the composition of %s at %s has digest %s, and its want names %s; nothing is applied until they agree",
module, w.Want.FP, comp.Dig, w.Want.Composed))
a.mu.Unlock()
return
}
a.mu.Lock()
if !a.checked[module] {
a.checked[module] = true
a.mu.Unlock()
if a.baseline(ctx, module, w, comp) {
return
}
} else {
a.mu.Unlock()
}
a.mu.Lock()
if r, ok := a.running[module]; ok && r.FP == w.Want.FP && r.Composed == w.Want.Composed && r.Phase != delivery.PhaseRemoved {
a.mu.Unlock()
return
}
att, tried := a.attempt[module]
same := tried && att.FP == w.Want.FP && att.Composed == w.Want.Composed
if same && !att.Held && att.WantRev == w.Rev && att.CompRev == comp.Rev {
a.mu.Unlock()
return // failed already for exactly these keys: retried by the next change of either
}
if same && att.Held && att.AnsweredRev >= w.Rev && att.CompRev == comp.Rev {
a.mu.Unlock()
return // held for a failed backup, and no answer since
}
if module == NodeModule {
// Not while a module's composition cannot be read or is refused: what it carries may be what left the node's own,
// and the node's own apply would remove it, held out by nothing (issue 1945, the review's gap 2).
if bad := a.unreadableModules(); len(bad) > 0 {
a.sayOnce("node-waits:"+strings.Join(bad, ","), fmt.Sprintf("the node's own composition is not applied while "+
"%s cannot be read or is refused: what it carries would be removed", strings.Join(bad, ", ")))
a.mu.Unlock()
return
}
}
others := a.heldOut(module)
a.mu.Unlock()
a.apply(ctx, module, w, comp, others)
}
// baseline decides whether what runs already is what the want names, once per module at the agent's start: the node's
// own have (what this node-engine last applied), else the declaration it kept before the switch (ADR 0539 §9). True when
// it is, with the have written; nothing is applied.
func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp compNow) bool {
// What a have or the kept declaration says runs is believed only when the engine still holds a record of each of its
// resources: a resource removed since (a node's own apply took it, issue 1945) is made again by an apply.
if a.env.Recorded != nil && !a.env.Recorded(ctx, comp.C) {
if module != NodeModule {
if _, ok := a.env.Haves.Read(ctx, module); ok {
a.mu.Lock()
a.prior[module] = true
a.mu.Unlock()
}
}
if a.env.Kept != nil {
if _, ok := a.env.Kept()[module]; ok {
a.mu.Lock()
a.prior[module] = true
a.mu.Unlock()
}
}
a.mu.Lock()
a.sayOnce("unrecorded:"+module+":"+comp.Dig, fmt.Sprintf("%s's composition %s is applied again: this engine holds "+
"no record of some of its resources", module, short(comp.Dig)))
a.mu.Unlock()
return false
}
if module != NodeModule {
if h, ok := a.env.Haves.Read(ctx, module); ok {
a.mu.Lock()
a.prior[module] = true
a.mu.Unlock()
if h.Composed != "" {
// This node-engine wrote it: what it says is what runs, and the kept declaration says nothing more. A
// failed apply or a different composition is applied again.
if h.FP == w.Want.FP && h.Composed == w.Want.Composed &&
(h.Phase == "" || h.Phase == string(delivery.PhaseRunning)) {
a.mu.Lock()
a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning}
a.mu.Unlock()
// Said again, so the session holds the module and its key carries the health this engine judges,
// not the one an older engine left there (xorg's unknown on g14 and shanks, 2026-10-25).
a.env.Haves.Write(ctx, Have{Module: module, FP: h.FP, Composed: h.Composed,
Phase: string(delivery.PhaseRunning), Detail: h.Detail})
return true
}
return false
}
}
}
if a.env.Kept == nil {
return false
}
kept, ok := a.env.Kept()[module]
if ok {
a.mu.Lock()
a.prior[module] = true
a.mu.Unlock()
}
if !ok || !SameResources(kept, comp.C.Resources) {
return false
}
a.mu.Lock()
a.running[module] = delivery.Have{FP: w.Want.FP, Composed: w.Want.Composed, Phase: delivery.PhaseRunning}
a.mu.Unlock()
a.env.Haves.Write(ctx, Have{Module: module, FP: w.Want.FP, Composed: w.Want.Composed, Phase: string(delivery.PhaseRunning)})
return true
}
func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNow, others []delivery.Composed) {
fp, dig := w.Want.FP, w.Want.Composed
a.mu.Lock()
_, ranBefore := a.running[module]
ranBefore = ranBefore || a.prior[module]
a.mu.Unlock()
// A data module is backed up before a new composition of it moves (ADR 0429 §14), unless the want says to go on
// without one (ADR 0539 §16).
if comp.C.Data && ranBefore && a.env.Backup != nil && w.Want.Backup != delivery.BackupSkip {
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseBackingUp)})
if got := a.env.Backup(ctx, module); !got.Done {
a.env.Say(fmt.Sprintf("%s is held: its backup before %s failed: %s", module, fp, got.Words))
a.mu.Lock()
a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev, Held: true, AnsweredRev: w.Rev}
prev := a.running[module]
a.mu.Unlock()
// It keeps running what it has: the have says what it holds and why it waits.
a.env.Haves.Write(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseHeld),
Detail: got.Words})
return
}
}
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)})
res := a.env.Apply(ctx, comp.C, others)
a.mu.Lock()
if res.Failed {
a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev}
a.mu.Unlock()
what := fmt.Sprintf("%s at %s", module, fp)
if module == NodeModule {
what = "the node's own composition"
}
a.env.Say(fmt.Sprintf("applying %s failed: %s", what, res.Detail))
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: res.Detail})
return
}
delete(a.attempt, module)
a.running[module] = delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning}
a.applied[module] = comp
a.mu.Unlock()
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail})
}
func (a *Agent) remove(ctx context.Context, module string, w wantNow) {
a.mu.Lock()
att := a.attempt[module]
r, ran := a.running[module]
if att.FP == "removed" && att.WantRev == w.Rev {
a.mu.Unlock()
return
}
if ran && r.Phase == delivery.PhaseRemoved {
a.mu.Unlock()
return
}
others := a.heldOut(module)
a.mu.Unlock()
res := a.env.Remove(ctx, module, others)
a.mu.Lock()
if res.Failed {
a.attempt[module] = attempt{FP: "removed", WantRev: w.Rev}
a.mu.Unlock()
a.env.Say(fmt.Sprintf("removing %s failed: %s", module, res.Detail))
return
}
delete(a.attempt, module)
a.running[module] = delivery.Have{Phase: delivery.PhaseRemoved}
a.mu.Unlock()
a.env.Haves.Removed(ctx, module)
}
// SameResources says whether two lists of resources hold the same resources, each by its id and its canonical JSON.
func SameResources(a, b []json.RawMessage) bool {
ma, ok := canonByID(a)
if !ok {
return false
}
mb, ok := canonByID(b)
if !ok || len(ma) != len(mb) {
return false
}
for id, x := range ma {
if y, there := mb[id]; !there || x != y {
return false
}
}
return true
}
func canonByID(rs []json.RawMessage) (map[string]string, bool) {
out := map[string]string{}
for _, raw := range rs {
var head struct {
ID string `json:"id"`
}
if json.Unmarshal(raw, &head) != nil || head.ID == "" {
return nil, false
}
canon, err := Canonical(raw)
if err != nil {
return nil, false
}
out[head.ID] = string(canon)
}
return out, true
}
// Canonical is raw with its object keys sorted and its spacing dropped, its numbers kept as written.
func Canonical(raw json.RawMessage) (json.RawMessage, error) {
var v any
d := json.NewDecoder(bytes.NewReader(raw))
d.UseNumber()
if err := d.Decode(&v); err != nil {
return nil, err
}
return json.Marshal(v)
}
func short(s string) string {
if i := strings.IndexByte(s, ':'); i >= 0 {
s = s[i+1:]
}
if len(s) > 12 {
s = s[:12]
}
return s
}