The reviewer showed how76e214astill lets the 07:54 outage happen at the re-landed engine's first start on novox. applied/ is empty there, so loadApplied loads nothing. The kept declaration retires, because every placed module has a have. Nothing then carries the node's own composition. The guard misses nats, which has no want there, so its nats.conf is guessed as nats's and removed. - At the start, when the store has no node's own composition, the node's own runs as the kept declaration has it, and the store keeps it now. The kept declaration retires only once the store holds `node`. - What this engine applied runs until loadHad reads the composition a differing have names. The bucket no longer keeps the one novox's have names, so the stored one stays. - In the node's own apply, a resource of another module goes only when that module's want says removed (Scope.RemovedByWant). A module the node's own composition carried has no want, so Placed never named it. Cost: the node's own apply cannot drop one of its own dotted resources without such a want. It is refused whole and said, never removed silently. Tests, with no new test functions: - A first-start case in the agent's Everything test (empty store, then a start with the declaration retired). It fails on76e214a. - Two rows in the guard's table: no want refuses; a want saying removed removes. - A case for the store in the authority test.
1679 lines
62 KiB
Go
1679 lines
62 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) — except an apply that failed only
|
|
// because the bus did not answer for a moment (Result.Transient: no responders while the bus reloads its accounts, a
|
|
// reconnect): that one is retried with a bounded backoff (Env.Backoff), and at once when the engine is told the bus is
|
|
// back (RetryTransient), since no key will change for it.
|
|
package agent
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/ed25519"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/declaration"
|
|
"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", "waiting-yes" or "removed"; empty reads as running.
|
|
Phase string
|
|
Detail string
|
|
// Judgement and Waits are, in phase waiting-yes, the judgement of the move and the digest of the composition that
|
|
// waits for a person's yes (ADR 0547 §3).
|
|
Judgement string
|
|
Waits string
|
|
// Sealed is, on a node composing its own, each value the composition it names carries sealed to the node's key, by
|
|
// slot (delivery.Have.SealedAt): kept in the have so the composer reuses them after a restart (ADR 0547 §3), each at
|
|
// its own slot alone (issue 1126, C1).
|
|
Sealed delivery.Seals
|
|
}
|
|
|
|
// 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
|
|
// Transient says the failure was the bus not answering for a moment (no responders, a timeout, a reconnect, a
|
|
// JetStream API error while the bus reloads), not the module's: the apply is tried again without a new revision.
|
|
Transient 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. A module
|
|
// whose have names a composition the bucket keeps runs from that composition, never from the kept declaration
|
|
// (issue 924): the kept declaration is older than every composition applied since the switch.
|
|
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
|
|
// Inactive says a resource the composition declares running is not active on the machine now (a unit stopped, a
|
|
// container not running), read from the machine and not from any record (ADR 0568 §1). A have or a kept declaration
|
|
// that says the module runs is believed at the agent's start only when this says no; nil says no, and an unreadable
|
|
// machine says no (unknown is not inactive).
|
|
Inactive 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)
|
|
// Settled is told a module's apply or removal that succeeded, once what runs holds it (Everything), outside the
|
|
// agent's lock: the declaration the node applies now names what that composition names, its filled files among
|
|
// them. The node-engine's filling tells the provisioner's client then to fill again what it refused (mesh-host
|
|
// internal/filling Knock; ace, 2026-10-25 21:17). Nil tells nothing.
|
|
Settled func(module string)
|
|
// RunsChanged is told, outside the agent's lock, that what runs (Everything) changed other than by an apply or a
|
|
// removal: once at the start, when the haves and the compositions they name are read, and each time a module is
|
|
// taken as running from its have or the kept declaration. Like Settled, the filling then tells the provisioner's
|
|
// client to fill again what it refused (issue 924, the second meeting: ace, g14 and shanks on 2026-10-29 at 00:09
|
|
// UTC refused power's three new drop-ins from an engine start on, as no module settled after it). Nil tells nothing.
|
|
RunsChanged func(why string)
|
|
// Backoff is how long to wait before the try-th retry (1, 2, …) of an apply that failed transiently. Nil retries a
|
|
// transient failure only on RetryTransient or a change of either key.
|
|
Backoff func(try int) time.Duration
|
|
Say func(string)
|
|
// SelfComposed says the compositions are the node's own (novox/hq ADR 0547 §2, issue 1126): Composed hands what the
|
|
// node composed itself, unsigned, and a want names the build alone. The composition composed for the want's build is
|
|
// the one applied, when it differs from what runs.
|
|
SelfComposed bool
|
|
// Policy is a module's upgrade policy as its manifest derives it (catalogue.DerivedUpgrade); with SelfComposed, a
|
|
// move of a module of policy record that recreates a container or moves its data waits for a person's yes in its
|
|
// want (ADR 0547 §3). Nil judges no move.
|
|
Policy func(ctx context.Context, module string) string
|
|
// HoldsData says whether a module's manifest declares data (catalogue.Manifest.DataItems), and whether its manifest
|
|
// could be read: what a composition's Data is, for what runs of a module from the kept declaration, which carries no
|
|
// Data of its own. A composition says Data, so without it a data module from the kept declaration is judged to move
|
|
// its data against its first composition after a restart (mongodb and mssql on novox, issue 924). Nil reads none.
|
|
HoldsData func(ctx context.Context, module string) (data, known bool)
|
|
// Applied keeps every composition the agent applied, by module (the node's own under NodeModule), beside the engine's
|
|
// state on the machine: read at the start before anything is watched, so what runs is known before the node composes
|
|
// its own for the first time, whatever the bucket and the kept declaration still hold (issue 774: on 2026-10-29 the
|
|
// bus node, its kept declaration retired and its own composition in no bucket, composed its own as if its folded
|
|
// modules ran nothing and removed the bus server and four held modules). Nil keeps none.
|
|
Applied AppliedStore
|
|
}
|
|
|
|
// AppliedStore is where the engine keeps the compositions it applied (Env.Applied): each module's last, as its bytes.
|
|
type AppliedStore interface {
|
|
// Load is every composition kept, by module; a store that cannot be read says so and loads none.
|
|
Load() (map[string][]byte, error)
|
|
// Keep keeps the composition the module was applied from, in place of the one before.
|
|
Keep(module string, raw []byte) error
|
|
// Drop forgets the module's, once it is removed.
|
|
Drop(module string) error
|
|
}
|
|
|
|
// transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has
|
|
// not answered for long enough that it is no longer a moment, and the failure is loud while the retries go on.
|
|
const transientFor = 4
|
|
|
|
// 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
|
|
// had is, for each module whose have names a composition the bucket keeps (its latest value or a revision), that
|
|
// composition, read once at the start (loadHad): what runs of a module the agent has not taken yet, ahead of the kept
|
|
// declaration (issue 924).
|
|
had map[string]compNow
|
|
// keptData is, for each module of the kept declaration whose manifest could be read, whether it declares data
|
|
// (Env.HoldsData): the Data of what runs of it from the kept declaration.
|
|
keptData map[string]bool
|
|
// 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
|
|
// notComposed is the modules whose have says the node could not compose them: why, and the have before (NotComposed).
|
|
notComposed map[string]notComposedMark
|
|
// heldAlone is the modules the composer holds alone (NotComposedEach): what runs of each in the node's own
|
|
// composition is held out of the node's own apply until a composition of its own is handed (heldOut).
|
|
heldAlone map[string]bool
|
|
retry chan []string
|
|
// reconcile carries the judge's edges: a resource found not active (ADR 0568 §2).
|
|
reconcile chan []string
|
|
// tries counts the transient failures in a row of each module; again carries a retry whose wait ran out, and
|
|
// transient asks for every transiently failed module at once. stopped is closed when Run returns.
|
|
tries map[string]int
|
|
gen uint64
|
|
again chan againAsk
|
|
transient chan struct{}
|
|
stopped chan struct{}
|
|
}
|
|
|
|
type againAsk struct {
|
|
module string
|
|
gen uint64
|
|
}
|
|
|
|
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
|
|
// Transient marks a failure of the bus, retried by the clock; Gen names the retry that is due for it.
|
|
Transient bool
|
|
Gen 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{}, had: map[string]compNow{}, keptData: map[string]bool{},
|
|
attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, notComposed: map[string]notComposedMark{},
|
|
retry: make(chan []string, 8), reconcile: make(chan []string, 64), tries: map[string]int{}, again: make(chan againAsk, 64),
|
|
transient: make(chan struct{}, 1), stopped: make(chan struct{})}
|
|
}
|
|
|
|
// 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 {
|
|
defer close(a.stopped)
|
|
// What this engine last applied, before any composition is watched: the node composing its own reads what runs
|
|
// (Everything) from its first pass on, and that pass comes before the haves are read below (issue 774).
|
|
a.loadApplied(ctx)
|
|
wants, err := a.env.Wants.Watch(ctx, "want."+a.env.Node+".>")
|
|
if err != nil {
|
|
return fmt.Errorf("the agent cannot watch its wants: %w", err)
|
|
}
|
|
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, hadRead 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 modules := <-a.reconcile:
|
|
a.mu.Lock()
|
|
for _, m := range modules {
|
|
if w := a.wants[m]; !w.Present || w.Want.Removed || a.attempt[m].Held {
|
|
continue
|
|
}
|
|
delete(a.running, m)
|
|
delete(a.attempt, m)
|
|
dirty[m] = true
|
|
}
|
|
a.mu.Unlock()
|
|
case ask := <-a.again:
|
|
// The wait of a transiently failed apply ran out: tried again from the keys it has, if nothing tried it since.
|
|
a.mu.Lock()
|
|
if att, ok := a.attempt[ask.module]; ok && att.Transient && att.Gen == ask.gen {
|
|
delete(a.attempt, ask.module)
|
|
dirty[ask.module] = true
|
|
}
|
|
a.mu.Unlock()
|
|
case <-a.transient:
|
|
a.mu.Lock()
|
|
for m, att := range a.attempt {
|
|
if att.Transient {
|
|
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) {
|
|
continue
|
|
}
|
|
if !hadRead {
|
|
// Before the first settle: what runs of each module with a have, from the composition it names, so the
|
|
// filling judges by it at once and not by the kept declaration until some module settles (issue 924).
|
|
hadRead = true
|
|
a.loadHad(ctx)
|
|
a.runsChanged("the haves were read at the start")
|
|
}
|
|
if 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:
|
|
}
|
|
}
|
|
|
|
// Reconcile asks for the modules to be applied again because a resource each declares running was found not active, once
|
|
// per edge (the judge's look, ADR 0568 §2). It is Retry, but a module whose apply is held (a failed backup) is left
|
|
// alone, as is one the node wants no more (ADR 0568 §4). It never blocks.
|
|
func (a *Agent) Reconcile(modules ...string) {
|
|
select {
|
|
case a.reconcile <- modules:
|
|
default:
|
|
}
|
|
}
|
|
|
|
// RetryTransient asks for every module whose last apply failed transiently to be applied again now: the engine says it
|
|
// when it knows the bus answers again (after it wrote the bus's user list and the bus reloaded its accounts). It never
|
|
// blocks.
|
|
func (a *Agent) RetryTransient() {
|
|
select {
|
|
case a.transient <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
// notComposedMark is why a module's have says the node could not compose it, and the have it said before.
|
|
type notComposedMark struct {
|
|
why string
|
|
prior Have
|
|
}
|
|
|
|
// NotComposedDetail opens the detail of a have NotComposed marks: the module keeps running what its build and digest
|
|
// name, and only the node's composition of it failed.
|
|
const NotComposedDetail = "the node cannot compose it, and keeps running what it ran: "
|
|
|
|
// composerMark says h is a mark NotComposed wrote: failed, its detail opening with NotComposedDetail.
|
|
func composerMark(h Have) bool {
|
|
return h.Phase == string(delivery.PhaseFailed) && strings.HasPrefix(h.Detail, NotComposedDetail)
|
|
}
|
|
|
|
// HavesReadErr is what a Haves may answer besides Read: a have absent (false, nil) told from one that could not be read
|
|
// (false, the error).
|
|
type HavesReadErr interface {
|
|
ReadErr(ctx context.Context, module string) (Have, bool, error)
|
|
}
|
|
|
|
// readHave reads a module's have: found, absent, or unread. A read that fails is tried again after a short wait while
|
|
// the bus does not answer (a reload pauses it for a few seconds, issue 1126), and is unread when it still fails. A
|
|
// Haves that cannot tell an absent key from a failed read says absent.
|
|
func (a *Agent) readHave(ctx context.Context, module string) (h Have, found, unread bool) {
|
|
re, tells := a.env.Haves.(HavesReadErr)
|
|
if !tells {
|
|
h, found = a.env.Haves.Read(ctx, module)
|
|
return h, found, false
|
|
}
|
|
for try := 0; ; try++ {
|
|
h, found, err := re.ReadErr(ctx, module)
|
|
if err == nil {
|
|
return h, found, false
|
|
}
|
|
if try == 2 || ctx.Err() != nil {
|
|
return Have{}, false, true
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return Have{}, false, true
|
|
case <-time.After(time.Duration(1<<try) * readAgain):
|
|
}
|
|
}
|
|
}
|
|
|
|
// readAgain is the first wait before a have is read again.
|
|
var readAgain = time.Second
|
|
|
|
// NotComposed says the node composing its own could not compose its modules (ADR 0547 §2: a stand-in refused, an
|
|
// input that cannot be read): each module with a want gets its have failed, naming why, while it keeps running what it
|
|
// ran, so the failure is seen where every have is read and not only in the journal. The have keeps what runs: the
|
|
// fingerprint, the digest and the sealed values the agent runs, or, before it ran anything since its start, those its
|
|
// last have said (a refusal at the first compose after a start). Called with "" once the node composes again: each
|
|
// module so marked has its have back as it was, unless an apply wrote it since.
|
|
func (a *Agent) NotComposed(ctx context.Context, why string) { a.NotComposedEach(ctx, why, nil) }
|
|
|
|
// NotComposedEach is NotComposed for the node refused whole (why), or, with why empty, for the modules the node holds
|
|
// alone (held, each with its own why: refused alone, the rest composed and applied, issue 1126): only those are marked,
|
|
// each naming its own why, and every other module marked before has its have back. Both empty, every mark is cleared.
|
|
func (a *Agent) NotComposedEach(ctx context.Context, why string, held map[string]string) {
|
|
a.mu.Lock()
|
|
if why == "" {
|
|
a.heldAlone = make(map[string]bool, len(held))
|
|
for m := range held {
|
|
a.heldAlone[m] = true
|
|
}
|
|
}
|
|
whyOf := map[string]string{} // the mark each module is to carry; a marked module not in it is cleared
|
|
for module, w := range a.wants {
|
|
if !w.Present || w.Want.Removed || module == NodeModule {
|
|
continue
|
|
}
|
|
if why != "" {
|
|
whyOf[module] = why
|
|
} else if h, ok := held[module]; ok && h != "" {
|
|
whyOf[module] = h
|
|
}
|
|
}
|
|
var modules []string
|
|
for module, w := range whyOf {
|
|
if a.notComposed[module].why != w {
|
|
modules = append(modules, module)
|
|
}
|
|
}
|
|
for module := range a.notComposed {
|
|
if _, still := whyOf[module]; !still && why == "" { // the node refused whole marks; it clears none
|
|
modules = append(modules, module)
|
|
}
|
|
}
|
|
a.mu.Unlock()
|
|
sort.Strings(modules)
|
|
for _, module := range modules {
|
|
why := whyOf[module]
|
|
if ctx.Err() != nil {
|
|
return // the run ended: what it would mark or clear is the next run's to say (issue 1126)
|
|
}
|
|
a.mu.Lock()
|
|
mark, marked := a.notComposed[module]
|
|
r, ran := a.running[module]
|
|
a.mu.Unlock()
|
|
if why == "" {
|
|
a.mu.Lock()
|
|
delete(a.notComposed, module)
|
|
a.mu.Unlock()
|
|
back := mark.prior
|
|
if ran {
|
|
back = Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(delivery.PhaseRunning)}
|
|
if r.Phase != "" {
|
|
back.Phase = string(r.Phase)
|
|
}
|
|
}
|
|
if back.Composed == "" {
|
|
continue // nothing said what ran: nothing to say back
|
|
}
|
|
a.writeHave(ctx, back)
|
|
continue
|
|
}
|
|
if !marked {
|
|
if ran {
|
|
mark.prior = Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(r.Phase)}
|
|
if mark.prior.Phase == "" {
|
|
mark.prior.Phase = string(delivery.PhaseRunning)
|
|
}
|
|
} else if h, found, unread := a.readHave(ctx, module); found {
|
|
mark.prior = h // what its last have said runs: this agent has run nothing of it since its start
|
|
} else if unread {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
// What its have names cannot be read (the bus paused, issue 1126): a mark written now would drop the build
|
|
// and digest it names. Nothing is written; the next refusal marks it once the have reads.
|
|
a.mu.Lock()
|
|
a.sayOnce("unmarked:"+module, fmt.Sprintf("%s's have cannot be read, so it is not marked: the node cannot compose it: %s", module, why))
|
|
a.mu.Unlock()
|
|
continue
|
|
}
|
|
}
|
|
// A have that already says another cause (an apply that failed, a backup held, an apply under way) keeps it: a
|
|
// composer's mark over it would hide that cause, and baseline would take the mark for what runs (review of the
|
|
// switch, round 1).
|
|
if p := mark.prior.Phase; !marked && p != "" && p != string(delivery.PhaseRunning) && !composerMark(mark.prior) {
|
|
a.mu.Lock()
|
|
a.sayOnce("unmarked-cause:"+module, fmt.Sprintf("%s is not marked: its have says %s (%s), and the node cannot compose it: %s",
|
|
module, p, mark.prior.Detail, why))
|
|
a.mu.Unlock()
|
|
continue
|
|
}
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
mark.why = why
|
|
a.mu.Lock()
|
|
a.notComposed[module] = mark
|
|
a.mu.Unlock()
|
|
a.writeHave(ctx, Have{Module: module, FP: mark.prior.FP, Composed: mark.prior.Composed, Sealed: mark.prior.Sealed,
|
|
Phase: string(delivery.PhaseFailed), Detail: NotComposedDetail + why})
|
|
}
|
|
}
|
|
|
|
// writeHave writes a module's have (the node's own key for NodeModule); on a node composing its own, with the sealed
|
|
// values of the composition it names when the caller gave none (Have.Sealed): those of the composition the agent applied or holds at that digest, else
|
|
// those the module's last have said at that digest.
|
|
//
|
|
// A writer that names no composition (one written from the agent's running entry before this agent ran anything of the
|
|
// module since its start: held by a failed backup, waiting for a yes, not composed) keeps what the module's last have
|
|
// said runs, its fingerprint, digest and sealed values, rather than a have naming nothing: what runs is not lost, and
|
|
// the boxes the composer reuses after a restart stay (review of L2, rounds 2 and 3).
|
|
func (a *Agent) writeHave(ctx context.Context, h Have) {
|
|
if h.Composed == "" && h.Module != NodeModule {
|
|
if last, ok := a.env.Haves.Read(ctx, h.Module); ok && last.Composed != "" {
|
|
h.FP, h.Composed = last.FP, last.Composed
|
|
if h.Sealed == nil {
|
|
h.Sealed = last.Sealed
|
|
}
|
|
}
|
|
}
|
|
// The node's own composition too: its core boxes are sealed again by the vault at a restart, and kept in its key the
|
|
// composer reuses them, so the restart applies nothing again (review of L2, round 3, finding 1).
|
|
if a.env.SelfComposed && h.Sealed == nil && h.Composed != "" {
|
|
h.Sealed = a.sealedOf(ctx, h.Module, h.FP, h.Composed)
|
|
}
|
|
a.env.Haves.Write(ctx, h)
|
|
}
|
|
|
|
func (a *Agent) sealedOf(ctx context.Context, module, fp, composed string) delivery.Seals {
|
|
a.mu.Lock()
|
|
if c, ok := a.applied[module]; ok && c.Err == nil && c.Dig == composed {
|
|
a.mu.Unlock()
|
|
return delivery.SealsIn(c.C)
|
|
}
|
|
if c, ok := a.compositionFor(module, fp); ok && c.Err == nil && c.Dig == composed {
|
|
a.mu.Unlock()
|
|
return delivery.SealsIn(c.C)
|
|
}
|
|
a.mu.Unlock()
|
|
if h, ok := a.env.Haves.Read(ctx, module); ok && h.Composed == composed {
|
|
return h.Sealed
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// 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 && a.env.SelfComposed {
|
|
// The node composed it itself: nothing crossed the bus, so nothing is signed (ADR 0547 §2).
|
|
now.C = c
|
|
now.Dig, err = delivery.ComposedDigestOf(c)
|
|
if err != nil {
|
|
now.C = delivery.Composed{}
|
|
}
|
|
} else 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
|
|
}
|
|
var dig string
|
|
if a.env.SelfComposed {
|
|
dig, err = delivery.ComposedDigestOf(c) // the node's own: unsigned (ADR 0547 §2)
|
|
} else {
|
|
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)
|
|
} else if c, ok := a.had[m]; ok && c.Dig == h.Composed {
|
|
out = append(out, c.C) // a revision its have named, replaced since under the same key
|
|
}
|
|
}
|
|
hadNames := make([]string, 0, len(a.had))
|
|
for m := range a.had {
|
|
if _, held := a.running[m]; !held && m != except {
|
|
hadNames = append(hadNames, m)
|
|
}
|
|
}
|
|
sort.Strings(hadNames)
|
|
for _, m := range hadNames {
|
|
out = append(out, a.had[m].C)
|
|
}
|
|
if a.env.Kept != nil {
|
|
kept := a.env.Kept()
|
|
names := make([]string, 0, len(kept))
|
|
for m := range kept {
|
|
_, held := a.running[m]
|
|
_, had := a.had[m]
|
|
if !held && !had && m != except {
|
|
names = append(names, m)
|
|
}
|
|
}
|
|
sort.Strings(names)
|
|
for _, m := range names {
|
|
c, _ := a.keptComposed(m, kept[m])
|
|
if m == NodeModule {
|
|
c.Module = ""
|
|
}
|
|
out = append(out, c)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// loadHad reads, once at the start, the have of every module with a want or in the kept declaration and, where it names
|
|
// a composition, that composition: the bucket's value at its key when its digest is the have's, else the revision the
|
|
// bucket keeps with that digest. A have naming none, or one the bucket no longer keeps, leaves the module to the kept
|
|
// declaration, said once (issue 924).
|
|
func (a *Agent) loadHad(ctx context.Context) {
|
|
a.mu.Lock()
|
|
seen := map[string]bool{}
|
|
for m, w := range a.wants {
|
|
if m != NodeModule && w.Present && !w.Want.Removed {
|
|
seen[m] = true
|
|
}
|
|
}
|
|
if a.env.Kept != nil {
|
|
for m := range a.env.Kept() {
|
|
if m != NodeModule {
|
|
seen[m] = true
|
|
}
|
|
}
|
|
}
|
|
a.mu.Unlock()
|
|
names := make([]string, 0, len(seen))
|
|
for m := range seen {
|
|
names = append(names, m)
|
|
}
|
|
sort.Strings(names)
|
|
a.readKeptData(ctx, names...)
|
|
for _, m := range names {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
h, ok := a.env.Haves.Read(ctx, m)
|
|
if !ok || h.Composed == "" || !haveRuns(h) {
|
|
continue
|
|
}
|
|
a.mu.Lock()
|
|
if c, kept := a.had[m]; kept && c.Dig == h.Composed {
|
|
a.mu.Unlock()
|
|
continue // what this engine applied and kept itself (loadApplied)
|
|
}
|
|
c, found := a.compositionFor(m, h.FP)
|
|
a.mu.Unlock()
|
|
if !found || c.Err != nil || c.Dig != h.Composed {
|
|
c, found = a.revisionNamed(ctx, m, h.FP, h.Composed)
|
|
}
|
|
a.mu.Lock()
|
|
if found {
|
|
a.had[m] = c
|
|
} else if _, applied := a.had[m]; applied {
|
|
// Kept as this engine applied it (loadApplied), never composed as nothing.
|
|
} else if _, kept := a.keptHas(m); kept {
|
|
a.sayOnce("had:"+m, fmt.Sprintf("%s's have names the composition %s, which the bucket no longer keeps: it runs "+
|
|
"from the kept declaration until it is applied or taken", m, short(h.Composed)))
|
|
}
|
|
a.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
// loadApplied takes, once at the start, what this engine last applied of each module as what runs of it, ahead of the
|
|
// bucket and the kept declaration (Env.Applied, issue 774): it is what the machine holds, whatever the bus still keeps.
|
|
// One whose have names another composition runs as applied until loadHad reads that one from the bucket; one whose
|
|
// have cannot be read, or that has none, is kept as applied, so a module is never composed as running nothing.
|
|
func (a *Agent) loadApplied(ctx context.Context) {
|
|
if a.env.Applied == nil {
|
|
return
|
|
}
|
|
kept, err := a.env.Applied.Load()
|
|
if err != nil {
|
|
a.env.Say(fmt.Sprintf("the compositions this engine applied cannot be read, so what runs is read from the haves alone: %v", err))
|
|
}
|
|
// At an engine's first start nothing is kept yet: the node's own runs as the kept declaration has it, which the store
|
|
// keeps now, so the node never composes its own as if its folded modules ran nothing once the declaration retires
|
|
// (issue 774, review of the re-land, round 1). Taken whatever digest the node's have names: the composition that names
|
|
// is in no bucket, and the declaration is the one record of what runs on disk.
|
|
seeded := false
|
|
if _, own := kept[NodeModule]; !own && err == nil && a.env.Kept != nil {
|
|
if rs, ok := a.env.Kept()[NodeModule]; ok && len(rs) > 0 {
|
|
c, _ := a.keptComposed(NodeModule, rs)
|
|
c.Module = ""
|
|
raw, eerr := delivery.EncodeComposed(c)
|
|
if eerr == nil {
|
|
eerr = a.env.Applied.Keep(NodeModule, raw)
|
|
}
|
|
if eerr != nil {
|
|
a.env.Say(fmt.Sprintf("the node's own as the kept declaration has it cannot be kept: %v", eerr))
|
|
} else {
|
|
if kept == nil {
|
|
kept = map[string][]byte{}
|
|
}
|
|
kept[NodeModule], seeded = raw, true
|
|
a.env.Say("the node's own runs as the kept declaration has it, kept now as what this engine runs")
|
|
}
|
|
}
|
|
}
|
|
names := make([]string, 0, len(kept))
|
|
for m := range kept {
|
|
names = append(names, m)
|
|
}
|
|
sort.Strings(names)
|
|
for _, m := range names {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
c, err := delivery.DecodeComposed(kept[m])
|
|
var dig string
|
|
if err == nil {
|
|
dig, err = delivery.ComposedDigestOf(c)
|
|
}
|
|
if err != nil {
|
|
a.env.Say(fmt.Sprintf("the composition this engine kept of %s cannot be read, so what runs of it is read from its have: %v", m, err))
|
|
continue
|
|
}
|
|
h, found, unread := a.readHave(ctx, m)
|
|
if seeded && m == NodeModule {
|
|
found, unread = true, false // what the kept declaration has runs, whatever the have names
|
|
} else if found && h.Composed != "" && h.Composed != dig && haveRuns(h) {
|
|
// The composition the have names replaces this once loadHad finds it in the bucket; until then, and when
|
|
// the bucket keeps it no more, what this engine applied runs (issue 774, review of the re-land, round 1).
|
|
a.env.Say(fmt.Sprintf("%s's have names the composition %s, not %s this engine applied last: it runs as applied "+
|
|
"until the one its have names is read", m, short(h.Composed), short(dig)))
|
|
}
|
|
if m == NodeModule {
|
|
c.Module, c.FP = "", ""
|
|
}
|
|
a.mu.Lock()
|
|
a.had[m] = compNow{Raw: kept[m], C: c, Dig: dig}
|
|
a.mu.Unlock()
|
|
switch {
|
|
case unread:
|
|
a.env.Say(fmt.Sprintf("%s's have cannot be read: it runs as this engine applied it, %s", m, short(dig)))
|
|
case !found:
|
|
a.env.Say(fmt.Sprintf("%s has no have: it runs as this engine applied it, %s", m, short(dig)))
|
|
}
|
|
}
|
|
}
|
|
|
|
// keepApplied keeps the composition a module was applied from (Env.Applied); one that cannot be kept is said, and the
|
|
// apply stands.
|
|
func (a *Agent) keepApplied(module string, comp compNow) {
|
|
if a.env.Applied == nil {
|
|
return
|
|
}
|
|
raw := comp.Raw
|
|
if len(raw) == 0 {
|
|
var err error
|
|
if raw, err = delivery.EncodeComposed(comp.C); err != nil {
|
|
a.env.Say(fmt.Sprintf("the composition %s was applied from cannot be kept: %v", module, err))
|
|
return
|
|
}
|
|
}
|
|
if err := a.env.Applied.Keep(module, raw); err != nil {
|
|
a.env.Say(fmt.Sprintf("the composition %s was applied from cannot be kept: %v", module, err))
|
|
}
|
|
}
|
|
|
|
// haveRuns says a have names what runs, in the phases baseline takes as running: running (or empty), held for a failed
|
|
// backup, or a composer's mark over what still runs. A have backing up, applying, checking, failed or waiting names a
|
|
// composition not applied, or not yet (issue 924, review of 031e106).
|
|
func haveRuns(h Have) bool {
|
|
switch delivery.Phase(h.Phase) {
|
|
case "", delivery.PhaseRunning, delivery.PhaseHeld:
|
|
return true
|
|
}
|
|
return composerMark(h)
|
|
}
|
|
|
|
// keptHas says whether the kept declaration has resources for the module. Called holding a.mu or not: Kept takes no lock
|
|
// of the agent's.
|
|
func (a *Agent) keptHas(module string) ([]json.RawMessage, bool) {
|
|
if a.env.Kept == nil {
|
|
return nil, false
|
|
}
|
|
rs, ok := a.env.Kept()[module]
|
|
return rs, ok
|
|
}
|
|
|
|
// runsChanged tells RunsChanged, outside the agent's lock.
|
|
func (a *Agent) runsChanged(why string) {
|
|
if a.env.RunsChanged != nil {
|
|
a.env.RunsChanged(why)
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
held[m] = true
|
|
}
|
|
// A module the composer holds alone keeps what runs of it in the node's own composition, until a composition of its
|
|
// own is handed: held out as one of its own, so the node's own apply never removes it (issue 1126, review of the
|
|
// apply, round 1: de-spiegel leaving the fold with its names unset; issue 1945).
|
|
if len(a.heldAlone) == 0 {
|
|
return out
|
|
}
|
|
var own *delivery.Composed
|
|
for _, c := range a.others("\x00") {
|
|
if c.Module == "" {
|
|
own = &c
|
|
break
|
|
}
|
|
}
|
|
alone := make([]string, 0, len(a.heldAlone))
|
|
for m := range a.heldAlone {
|
|
if !held[m] && m != except && own != nil {
|
|
alone = append(alone, m)
|
|
}
|
|
}
|
|
sort.Strings(alone)
|
|
for _, m := range alone {
|
|
var rs []json.RawMessage
|
|
for _, raw := range own.Resources {
|
|
var head struct {
|
|
ID string `json:"id"`
|
|
Module string `json:"module"`
|
|
}
|
|
if json.Unmarshal(raw, &head) != nil {
|
|
continue
|
|
}
|
|
owner := head.Module
|
|
if owner == "" {
|
|
owner, _ = declaration.ModuleGuess(head.ID)
|
|
}
|
|
if owner == m {
|
|
rs = append(rs, raw)
|
|
}
|
|
}
|
|
if len(rs) > 0 {
|
|
// Held out, never judged: its Data from its manifest where known, as every other composition says it.
|
|
out = append(out, delivery.Composed{Node: a.env.Node, Module: m, Resources: rs, Data: a.keptData[m]})
|
|
}
|
|
}
|
|
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")
|
|
}
|
|
|
|
// Placed is the modules a want still places on the node (present and not removed), held ones included, but except:
|
|
// what no apply but the module's own may remove (apply.Scope.Placed, issue 774). Called without a.mu held.
|
|
func (a *Agent) Placed(except string) map[string]bool {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
out := map[string]bool{}
|
|
for m, w := range a.wants {
|
|
if m != NodeModule && m != except && w.Present && !w.Want.Removed {
|
|
out[m] = true
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// RemovedByWant is the modules whose want says removed (apply.Scope.RemovedByWant, issue 774): in the node's own apply,
|
|
// only their resources may go. Called without a.mu held.
|
|
func (a *Agent) RemovedByWant() map[string]bool {
|
|
a.mu.Lock()
|
|
defer a.mu.Unlock()
|
|
out := map[string]bool{}
|
|
for m, w := range a.wants {
|
|
if m != NodeModule && w.Present && w.Want.Removed {
|
|
out[m] = true
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// 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 a.env.SelfComposed {
|
|
if !have || comp.Err != nil {
|
|
return // waits for the node's own composition of the want's build
|
|
}
|
|
// The composition the node composed for the want's build is the one it applies (ADR 0547 §2).
|
|
w.Want.Composed = comp.Dig
|
|
if a.waitsForYes(ctx, module, w, comp) {
|
|
return
|
|
}
|
|
} else 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.Phase == string(delivery.PhaseHeld) {
|
|
// Held for a failed backup, and the operator's answer is still to come: the hold is kept, nothing is
|
|
// backed up again or applied, and the next revision of the want is the answer (ADR 0568 §4).
|
|
a.mu.Lock()
|
|
a.attempt[module] = attempt{FP: w.Want.FP, Composed: w.Want.Composed, WantRev: w.Rev, CompRev: comp.Rev, Held: true,
|
|
AnsweredRev: w.Rev}
|
|
if h.Composed != "" {
|
|
a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning}
|
|
}
|
|
a.mu.Unlock()
|
|
a.runsChanged(module + " was taken as held")
|
|
return true
|
|
}
|
|
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.
|
|
// A have a composer marked failed (NotComposed) names what still runs: taken as running, so a mark this
|
|
// agent did not make clears once what it names is what the want names (issue 1126, the stale marks after
|
|
// the seed's revert).
|
|
marked := composerMark(h)
|
|
if marked {
|
|
h.Detail = ""
|
|
}
|
|
if h.FP == w.Want.FP && h.Composed == w.Want.Composed &&
|
|
(h.Phase == "" || h.Phase == string(delivery.PhaseRunning) || marked) {
|
|
if a.inactiveAtStart(ctx, module, comp) {
|
|
return false
|
|
}
|
|
a.mu.Lock()
|
|
a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning}
|
|
a.mu.Unlock()
|
|
a.runsChanged(module + " was taken from its have")
|
|
// 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.writeHave(ctx, Have{Module: module, FP: h.FP, Composed: h.Composed,
|
|
Phase: string(delivery.PhaseRunning), Detail: h.Detail, Sealed: h.Sealed})
|
|
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
|
|
}
|
|
if a.inactiveAtStart(ctx, module, comp) {
|
|
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.runsChanged(module + " was taken from the kept declaration")
|
|
a.writeHave(ctx, Have{Module: module, FP: w.Want.FP, Composed: w.Want.Composed, Phase: string(delivery.PhaseRunning)})
|
|
return true
|
|
}
|
|
|
|
// inactiveAtStart says the machine shows a resource the module declares running not active, so what a have or the kept
|
|
// declaration says runs is not believed and the module is applied once, as for a changed want (ADR 0568 §1).
|
|
func (a *Agent) inactiveAtStart(ctx context.Context, module string, comp compNow) bool {
|
|
if a.env.Inactive == nil || !a.env.Inactive(ctx, comp.C) {
|
|
return false
|
|
}
|
|
a.mu.Lock()
|
|
a.prior[module] = true
|
|
a.sayOnce("inactive:"+module+":"+comp.Dig, fmt.Sprintf("%s is applied again: a resource it declares running is not "+
|
|
"active on this machine", module))
|
|
a.mu.Unlock()
|
|
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.writeHave(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.writeHave(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseHeld),
|
|
Detail: got.Words})
|
|
return
|
|
}
|
|
}
|
|
a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)})
|
|
res := a.env.Apply(ctx, comp.C, others)
|
|
a.mu.Lock()
|
|
what := fmt.Sprintf("%s at %s", module, fp)
|
|
if module == NodeModule {
|
|
what = "the node's own composition"
|
|
}
|
|
if res.Failed && res.Transient {
|
|
a.tries[module]++
|
|
try := a.tries[module]
|
|
a.gen++
|
|
gen := a.gen
|
|
a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev, Transient: true, Gen: gen}
|
|
a.mu.Unlock()
|
|
phase, when := delivery.PhaseApplying, "when the bus answers again"
|
|
if a.env.Backoff != nil {
|
|
wait := a.env.Backoff(try)
|
|
when = "in " + wait.String()
|
|
time.AfterFunc(wait, func() {
|
|
select {
|
|
case a.again <- againAsk{module, gen}:
|
|
case <-a.stopped:
|
|
case <-ctx.Done():
|
|
}
|
|
})
|
|
}
|
|
if try > transientFor {
|
|
phase = delivery.PhaseFailed // still retried, and said as failed: no longer a moment
|
|
}
|
|
a.env.Say(fmt.Sprintf("applying %s failed because the bus did not answer (try %d), retried %s: %s", what, try, when, res.Detail))
|
|
a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(phase),
|
|
Detail: fmt.Sprintf("retrying %s (try %d): %s", when, try, res.Detail)})
|
|
return
|
|
}
|
|
delete(a.tries, module)
|
|
if res.Failed {
|
|
a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev}
|
|
a.mu.Unlock()
|
|
a.env.Say(fmt.Sprintf("applying %s failed: %s", what, res.Detail))
|
|
a.writeHave(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: res.Detail})
|
|
return
|
|
}
|
|
delete(a.attempt, module)
|
|
delete(a.notComposed, module) // the apply says what it runs now
|
|
a.running[module] = delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning}
|
|
a.applied[module] = comp
|
|
a.mu.Unlock()
|
|
a.keepApplied(module, comp)
|
|
if a.env.Settled != nil {
|
|
a.env.Settled(module)
|
|
}
|
|
a.writeHave(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)
|
|
delete(a.had, module)
|
|
a.running[module] = delivery.Have{Phase: delivery.PhaseRemoved}
|
|
a.mu.Unlock()
|
|
if a.env.Applied != nil {
|
|
if err := a.env.Applied.Drop(module); err != nil {
|
|
a.env.Say(fmt.Sprintf("the composition %s was applied from cannot be forgotten now it is removed: %v", module, err))
|
|
}
|
|
}
|
|
if a.env.Settled != nil {
|
|
a.env.Settled(module)
|
|
}
|
|
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
|
|
}
|
|
|
|
// waitsForYes judges a move the node composed for a module of policy record (ADR 0547 §3, ADR 0377 §3): one that
|
|
// recreates a container or moves the module's data, or that cannot be judged, is not applied while the want names no yes
|
|
// for this composition; the have says phase waiting-yes with the judgement, and the assigner asks a person. A composition
|
|
// equal to what runs is no move. True when the module waits.
|
|
func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp compNow) bool {
|
|
if module == NodeModule || a.env.Policy == nil || a.env.Policy(ctx, module) != catalogue.PolicyRecord || w.Want.Yes == comp.Dig {
|
|
return false
|
|
}
|
|
a.readKeptData(ctx, module) // the manifest may read now though it did not at the start
|
|
a.mu.Lock()
|
|
r, ran := a.running[module]
|
|
if ran && r.Composed == comp.Dig && r.Phase != delivery.PhaseRemoved {
|
|
a.mu.Unlock()
|
|
return false
|
|
}
|
|
from, known := a.runningComposition(module)
|
|
prior := ran || a.prior[module]
|
|
a.mu.Unlock()
|
|
if !ran && !known {
|
|
// Nothing of it run since this agent's start, and no kept declaration: its last have says what runs (review of L2,
|
|
// round 3). The same composition is no move; another one is a move whose source cannot be read, so it waits for
|
|
// a yes rather than being taken as the module's first composition.
|
|
if h, ok := a.env.Haves.Read(ctx, module); ok && h.Composed != "" {
|
|
if h.Composed == comp.Dig {
|
|
return false
|
|
}
|
|
prior = true
|
|
}
|
|
}
|
|
var j delivery.Judged
|
|
switch {
|
|
case known:
|
|
j = delivery.JudgeMove(&from, comp.C)
|
|
case prior:
|
|
j = delivery.Judged{Judgement: delivery.JudgedCannot, Words: "what " + module + " runs on this node cannot be read"}
|
|
default:
|
|
j = delivery.JudgeMove(nil, comp.C)
|
|
}
|
|
switch j.Judgement {
|
|
case delivery.JudgedRecreates, delivery.JudgedMovesData, delivery.JudgedCannot:
|
|
default:
|
|
return false
|
|
}
|
|
a.mu.Lock()
|
|
a.sayOnce("yes:"+module+":"+comp.Dig, fmt.Sprintf("%s at %s waits for a person's yes: %s", module, w.Want.FP, j.Words))
|
|
prev := a.running[module]
|
|
a.mu.Unlock()
|
|
a.writeHave(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseWaitingYes),
|
|
Detail: j.Words, Judgement: string(j.Judgement), Waits: comp.Dig})
|
|
return true
|
|
}
|
|
|
|
// runningComposition is what the module runs now: the composition last applied, else its resources in the declaration
|
|
// the node kept from before the switch. Called with a.mu held.
|
|
//
|
|
// The composition its have named at the start comes before the kept declaration (issue 924). From the kept declaration,
|
|
// what runs carries Data as its manifest declares it; a module whose manifest could not be read cannot be judged against
|
|
// it (false), and that is said, rather than taking it to hold no data and judging a false move of its data.
|
|
func (a *Agent) runningComposition(module string) (delivery.Composed, bool) {
|
|
if c, ok := a.applied[module]; ok && c.Err == nil {
|
|
return c.C, true
|
|
}
|
|
if c, ok := a.had[module]; ok {
|
|
return c.C, true
|
|
}
|
|
if rs, ok := a.keptHas(module); ok {
|
|
c, known := a.keptComposed(module, rs)
|
|
if !known {
|
|
a.sayOnce("kept-data:"+module, fmt.Sprintf("%s runs from the kept declaration, and whether it holds data cannot be "+
|
|
"read from its manifest: a move of it cannot be judged", module))
|
|
}
|
|
return c, known
|
|
}
|
|
return delivery.Composed{}, false
|
|
}
|
|
|
|
// keptComposed is what runs of a module from the kept declaration: its resources there, the paths it names as filled,
|
|
// and Data as the module's manifest declares it; false when the manifest could not be read, so Data is not known. Called
|
|
// with a.mu held.
|
|
func (a *Agent) keptComposed(module string, rs []json.RawMessage) (delivery.Composed, bool) {
|
|
c := delivery.Composed{Node: a.env.Node, Module: module, Resources: rs}
|
|
if a.env.KeptFilled != nil {
|
|
c.Filled = a.env.KeptFilled()[module]
|
|
}
|
|
data, known := a.keptData[module]
|
|
c.Data = data
|
|
return c, known || module == NodeModule
|
|
}
|
|
|
|
// readKeptData reads, outside the agent's lock, whether each module of the kept declaration not known yet declares data
|
|
// (Env.HoldsData), and keeps the answers it could read.
|
|
func (a *Agent) readKeptData(ctx context.Context, modules ...string) {
|
|
if a.env.HoldsData == nil {
|
|
return
|
|
}
|
|
for _, m := range modules {
|
|
if m == NodeModule {
|
|
continue
|
|
}
|
|
a.mu.Lock()
|
|
_, done := a.keptData[m]
|
|
a.mu.Unlock()
|
|
if _, kept := a.keptHas(m); done || !kept {
|
|
continue
|
|
}
|
|
if data, known := a.env.HoldsData(ctx, m); known {
|
|
a.mu.Lock()
|
|
a.keptData[m] = data
|
|
a.mu.Unlock()
|
|
}
|
|
}
|
|
}
|