Files
mesh-host/internal/selfcompose/selfcompose.go
T
jschoubben a7cd0a7807 Node shadow: reuse a running seal only at its own slot, lay values as the controller does, answer the newest candidate first and say how it goes (issue 1126, C1 C2, T1 T2 T3)
A box is reused by slot (resource, field) after the composition, from what runs or the
have's slots; a have written before slots is reused within its module alone. Config
values are laid by mesh-sdk's one rule (never a default). The shadow answers the
newest candidate first and hands each answer as it lands, says when an answer starts,
every minute what it waits on, and when it ends, and fits an answer to the node's key.
Needs mesh-sdk fix/1126-seed-same (16b8d8a1).
2026-10-28 09:41:29 +01:00

960 lines
32 KiB
Go

// Package selfcompose is the node agent's own composer (novox/hq ADR 0547 §2, issue 1126): it composes this node's
// modules and its own composition with mesh-sdk's composer, from what is on the bus, and hands the compositions to the
// agent as the agent once read them from the controller's bucket `composed-<node>`.
//
// **It watches, it never polls** (ADR 0547 §2): its inputs are watches of the node's wants, the indexer's placement and
// modules, every node's own have key (where a node's record rides, ADR 0555 §3) and mesh-config's resolved values of this
// node. A change of any of them marks the node to compose again; one loop drains the mark, so a burst of changes composes
// the node once (composing one node takes 0.8 to 4.4 s, measured 2026-10-26). It hands the agent only the compositions that
// differ from the ones it handed before, and takes back one no want names any more. A composition that cannot be made is
// said, once per reason, and the ones handed before stand: a node keeps running what it ran rather than nothing.
package selfcompose
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"sort"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-host/internal/agent"
)
// ResolvedBucket is mesh-config's state of resolved values, keyed `<node>.<module>.<key>` (mesh-catalog modules/mesh-config
// resolve.go ResolvedKey); a node reads its own slice alone (ADR 0547 §2).
const ResolvedBucket = "mesh-config_resolved"
// Env is what the composer needs of the node.
type Env struct {
Node string
// Watch follows one bucket's keys matching pattern, as the agent's watches do.
Watch func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error)
// Get reads one key's latest value; nil and no error when there is none.
Get func(ctx context.Context, bucket, key string) ([]byte, error)
Mesh frombus.Mesh
Say func(string)
// Own asks the vault's `own` in this node's engine's name, with the arguments given: its modules' own secrets and the
// pair credentials asked, sealed to its key (ADR 0550 §5). Nil asks nothing, and a module with either does not compose.
Own func(ctx context.Context, args map[string]any) (json.RawMessage, error)
// Hear subscribes a subject: here the vault's word that a secret changed (VaultSecretChanged), on which the node
// composes again, so a rotated secret or credential reaches it at once (ADR 0547 §2). Nil hears nothing.
Hear func(subject string) (<-chan []byte, func(), error)
// Refused is told why the node cannot compose its modules, and "" once it composes again, so each wanted module's
// have says it (agent.NotComposed): a refusal is loud where haves are read, not one journal line. Nil tells nobody.
Refused func(why string)
// Running is what the node runs now (agent.Everything), and Unseal opens a value sealed to this machine: a secret
// the vault seals again for each ask is kept as the box the node already runs at the same slot (the same field of
// the same resource of the same module, delivery.KeepSeals) when that box opens to the same value, so neither the
// switch to composing its own nor a restart changes a composition (ADR 0547 §3); a box held at another slot is
// never taken, though it opens to the same value (issue 1126, C1: a provider's grant and its consumer's secret).
// Either nil keeps the vault's box.
Running func() []delivery.Composed
Unseal func(sealed string) ([]byte, error)
// Held is the sealed values a module's have says it runs, kept by the agent on the bus: what runs after a restart,
// before the agent applied anything, and after the switch once the kept declaration is old. By slot
// (delivery.Have.SealedAt); legacy is the list a have written before slots holds (delivery.Have.Sealed), reused
// only within its own module.
Held func(module string) (slots delivery.Seals, legacy []string)
// Fetch reads an archive at a URL in the artifact store: a moved module's templates archive, whose texts its built
// manifest carries (issue 1126). Read once per digest. Nil reads none, and a moved module does not compose.
Fetch func(ctx context.Context, url string) ([]byte, error)
}
// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
const VaultSecretChanged = "mesh.mod.mesh-vault.event.secret-changed"
// Composer composes the node and is the agent's watcher of its compositions (agent.Env.Composed).
type Composer struct {
env Env
mu sync.Mutex
wants map[string]delivery.Want // by module, the node's own under agent.NodeModule
placements map[string]delivery.Placement
records map[string]delivery.NodeRecord
// inOwn is, by module, whether this node's have of it names no fingerprint of the new form (mesh-wireguard, a module
// at a build of the old form): the node runs it inside its own composition, and composes it there again (issue 1126).
inOwn map[string]bool
modules map[string]delivery.Module
values map[string]map[string]any // by module, this node's
manifests map[string][]byte // by digest, read once
archives map[string][]byte // templates archives by digest, read once
handed map[string]string // composed key to the digest handed
// sealedOwn is each own secret's sealed value as first handed, by module/name and generation: a value sealed again
// is other bytes, so the one first had is kept while the generation stands, and the composition stays the same.
sealedOwn map[string]ownSealed
said string
refusal string
// doing is what the composer waits on now (waiting), for a running answer to say (issue 1126, T2); its own lock, so
// it is read while c.mu is held.
doingMu sync.Mutex
doing string
caught bool
rev uint64
}
// New is a composer for one node.
func New(env Env) *Composer {
if env.Say == nil {
env.Say = func(string) {}
}
return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{},
records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}}
}
// ownSealed is one own secret's sealed value at a generation and the key it is sealed to.
type ownSealed struct {
generation int
key string
sealed string
}
// input is one watched bucket and how its updates are taken.
type input struct {
bucket, pattern string
take func(c *Composer, u agent.Update)
// changed, when set, takes the update in take's place and says whether it changed anything composed.
changed func(c *Composer, u agent.Update) bool
}
func (c *Composer) inputs() []input {
node := c.env.Node
return []input{
{delivery.AssignerWantsBucket, "want." + node + ".>", (*Composer).takeWant, nil},
{delivery.PlacementBucket, ">", (*Composer).takePlacement, nil},
// Every node's record, and this node's own haves (what it runs inside its own composition): two watches, each
// within the node's grant (buscompose: readGrants of the have bucket at `*.node`, NodeKeysGrants at `<node>.>`).
// A watch of every key is refused by the bus (issue 1126, 2026-10-27 09:21: the fold's `>` stopped every agent).
{delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, (*Composer).takeRecordChanged},
{delivery.HaveBucket, node + ".>", nil, (*Composer).takeModuleHave},
{delivery.ModulesBucket, ">", (*Composer).takeModule, nil},
{ResolvedBucket, node + ".>", (*Composer).takeValue, nil},
}
}
// Watch starts the composer and answers the compositions it makes: each as a key of `composed-<node>` would be, then a
// Caught update once every input was read and the node composed once, then every change. It ends when ctx ends or an
// input's watch ends.
func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, error) {
ins := c.inputs()
type tagged struct {
in int
u agent.Update
ok bool
}
merged := make(chan tagged, 64)
for i, in := range ins {
ch, err := c.env.Watch(ctx, in.bucket, in.pattern)
if err != nil {
return nil, fmt.Errorf("the node composes from %s, which cannot be watched: %w", in.bucket, err)
}
go func(i int, ch <-chan agent.Update) {
for u := range ch {
select {
case merged <- tagged{in: i, u: u, ok: true}:
case <-ctx.Done():
return
}
}
select {
case merged <- tagged{in: i}:
case <-ctx.Done():
}
}(i, ch)
}
if c.env.Hear != nil {
heard, stop, err := c.env.Hear(VaultSecretChanged)
if err != nil {
return nil, fmt.Errorf("the node composes again when a secret changes, and cannot hear the vault say so: %w", err)
}
go func() {
defer stop()
for {
select {
case <-ctx.Done():
return
case _, ok := <-heard:
if !ok {
return
}
select {
case merged <- tagged{in: -1, ok: true}:
case <-ctx.Done():
return
}
}
}
}()
}
out := make(chan agent.Update, 64)
go func() {
defer close(out)
caught := make([]bool, len(ins))
ready := false
dirty := false
for {
// Drain whatever came, then compose once for all of it.
select {
case <-ctx.Done():
return
case t := <-merged:
if !t.ok {
c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket))
return
}
switch {
case t.in < 0:
dirty = true // a secret changed: its new generation is asked of the vault at the compose
case t.u.Caught:
caught[t.in] = true
case ins[t.in].changed != nil:
dirty = ins[t.in].changed(c, t.u) || dirty
default:
ins[t.in].take(c, t.u)
dirty = true
}
if len(merged) > 0 {
continue
}
}
if !ready {
ready = all(caught)
if !ready {
continue
}
dirty = true
}
if !dirty {
continue
}
dirty = false
for _, u := range c.compose(ctx) {
select {
case out <- u:
case <-ctx.Done():
return
}
}
if !c.caughtSent() {
select {
case out <- agent.Update{Caught: true}:
case <-ctx.Done():
return
}
}
}
}()
return out, nil
}
func all(b []bool) bool {
for _, v := range b {
if !v {
return false
}
}
return true
}
// caughtSent says the Caught update went out, marking it sent the first time it is asked.
func (c *Composer) caughtSent() bool {
c.mu.Lock()
defer c.mu.Unlock()
sent := c.caught
c.caught = true
return sent
}
func (c *Composer) takeWant(u agent.Update) {
c.mu.Lock()
defer c.mu.Unlock()
if _, err := delivery.ParseWantNodeKey(u.Key); err == nil {
if u.Deleted {
delete(c.wants, agent.NodeModule)
} else {
c.wants[agent.NodeModule] = delivery.Want{}
}
return
}
node, module, err := delivery.ParseWantKey(u.Key)
if err != nil || node != c.env.Node {
return
}
w, err := delivery.DecodeWant(u.Value)
if u.Deleted || err != nil || w.Removed || w.FP == "" {
delete(c.wants, module)
return
}
c.wants[module] = w
}
func (c *Composer) takePlacement(u agent.Update) {
node, err := delivery.ParsePlacementKey(u.Key)
if err != nil {
return
}
c.mu.Lock()
defer c.mu.Unlock()
p, err := delivery.DecodePlacement(u.Value)
if u.Deleted || err != nil || p.Node != node {
delete(c.placements, node)
return
}
c.placements[node] = p
}
func (c *Composer) takeRecord(u agent.Update) {
node, _, isNode, err := delivery.ParseHaveKey(u.Key)
if err != nil || !isNode {
return
}
c.mu.Lock()
defer c.mu.Unlock()
h, err := delivery.DecodeHaveNode(u.Value)
if u.Deleted || err != nil || h.Record == nil {
delete(c.records, node)
return
}
c.records[node] = *h.Record
}
// takeRecordChanged takes a node's record, which is always composed again.
func (c *Composer) takeRecordChanged(u agent.Update) bool {
c.takeRecord(u)
return true
}
// takeModuleHave takes one of this node's modules' haves, and says whether it changed which it runs inside its own
// composition: one whose have names no fingerprint of the new form, is not removed, and runs something (nats on novox: a build-...
// id; mesh-wireguard: none). Read from the haves, which outlast a restart, never from what the agent has applied
// this run: a fresh agent composes before it applies anything (review of the fold, round 1).
func (c *Composer) takeModuleHave(u agent.Update) bool {
node, module, isNode, err := delivery.ParseHaveKey(u.Key)
if err != nil || isNode || node != c.env.Node {
return false
}
in := false
if !u.Deleted {
if h, err := delivery.DecodeHave(u.Value); err == nil {
// A have that names nothing it runs, failed (the node could not compose a module it never ran: NotComposed), is
// no module run inside the node's own composition: taken as one, it was composed from its manifest past its
// build (issue 1126, review round 1 of the builds).
runsNothing := h.Phase == delivery.PhaseFailed && h.FP == "" && h.Composed == ""
in = h.Phase != delivery.PhaseRemoved && !runsNothing && !strings.HasPrefix(h.FP, delivery.SourcePrefix2)
}
}
c.mu.Lock()
defer c.mu.Unlock()
if c.inOwn[module] == in {
return false
}
if in {
c.inOwn[module] = true
} else {
delete(c.inOwn, module)
}
return true
}
// inOwnOf is, of the modules p places, those the node runs inside its own composition (c.inOwn), sorted; c.mu held.
func (c *Composer) inOwnOf(p delivery.Placement) []string {
var out []string
for _, m := range p.Modules {
if c.inOwn[m.Module] {
out = append(out, m.Module)
}
}
sort.Strings(out)
return out
}
func (c *Composer) takeModule(u agent.Update) {
name, err := delivery.ParseModulesKey(u.Key)
if err != nil {
return
}
c.mu.Lock()
defer c.mu.Unlock()
m, err := delivery.DecodeModule(u.Value)
if u.Deleted || err != nil {
delete(c.modules, name)
return
}
c.modules[name] = m
}
func (c *Composer) takeValue(u agent.Update) {
rest, ok := strings.CutPrefix(u.Key, c.env.Node+".")
if !ok {
return
}
module, key, ok := strings.Cut(rest, ".")
if !ok {
return
}
c.mu.Lock()
defer c.mu.Unlock()
// Only a value its source lays: a default is the composition's already, as the controller reads them
// (catalogue.LaidValue, issue 1126, C2: Element's links blanked by a laid default).
v, laid := catalogue.LaidValue(u.Value)
if u.Deleted || !laid {
delete(c.values[module], key)
return
}
if c.values[module] == nil {
c.values[module] = map[string]any{}
}
c.values[module][key] = v
}
// compose composes the node once and answers the updates to hand: each composition whose digest differs from the one
// handed, and a deletion for each handed one no want names any more.
func (c *Composer) compose(ctx context.Context) []agent.Update {
c.mu.Lock()
fps := map[string]string{}
kept := map[string]string{}
var behind []string // modules wanted at another build than the trunk's: composed with that build's manifest
for module, w := range c.wants {
if module == agent.NodeModule {
continue
}
fps[module] = w.FP
if m, ok := c.modules[module]; ok && m.FP2 != w.FP {
behind = append(behind, module)
}
}
_, nodeWanted := c.wants[agent.NodeModule]
// What it runs inside its own composition and no want names at a build of the new form stays there, as the
// controller's composed writer folds it (issue 1126); compose folds none a want names.
own, placed := c.placements[c.env.Node]
inOwn := c.inOwnOf(own)
r := reader{c: c}
c.mu.Unlock()
if !placed {
// The catalogue no longer places it: the agent returns to the controller's compositions (endWhenUnplaced), and
// this is no refusal of what it runs (issue 1126, the seed's revert).
c.sayOnce("the catalogue no longer places this node: it composes nothing of its own, and keeps running what it runs")
return nil
}
// Composed from the build its want names (frombus.Scope.KeptAt, issue 1126).
for _, module := range behind {
kept[module] = fps[module]
}
// What it runs: a folded module of an old-form build is carried in its own composition as it runs (issue 1126).
var running []delivery.Composed
if c.env.Running != nil {
running = c.env.Running()
}
scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running}
compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
// A bus that does not answer for a moment (a reload of its accounts pauses it a few seconds) is no refusal: the pass is
// tried again after a growing wait before anything is marked (issue 1126).
for try := 0; err != nil && Transient(err) && ctx.Err() == nil && try < len(composeAgain); try++ {
c.env.Say(fmt.Sprintf("the node's composition met a bus that did not answer, and is tried again in %s: %v", composeAgain[try], err))
select {
case <-ctx.Done():
case <-time.After(composeAgain[try]):
}
if ctx.Err() == nil {
compositions, _, err = frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
}
}
if ctx.Err() != nil {
// The run ended (the agent begins again, on its own compositions or the controller's): what failed with it is
// no refusal, and nothing is marked (issue 1126: a pass in flight at the switch marked the next agent's modules).
return nil
}
if err != nil {
// Said before the agent marks the modules: the marks read each have first, which a paused bus delays.
c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err))
c.refused(fmt.Sprintf("%v", err))
return nil
}
c.refused("")
c.sayOnce("")
// The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs.
for key, comp := range compositions {
compositions[key] = c.keepRunning(comp, running)
}
c.mu.Lock()
defer c.mu.Unlock()
var out []agent.Update
now := map[string]bool{}
keys := make([]string, 0, len(compositions))
for key := range compositions {
keys = append(keys, key)
}
sort.Strings(keys)
for _, key := range keys {
comp := compositions[key]
if comp.Module == "" && !nodeWanted {
continue
}
now[key] = true
dig, err := delivery.ComposedDigestOf(comp)
if err != nil {
c.env.Say(fmt.Sprintf("the composition %s has no digest, and is not handed: %v", key, err))
continue
}
if c.handed[key] == dig {
continue
}
raw, err := delivery.EncodeComposed(comp)
if err != nil {
c.env.Say(fmt.Sprintf("the composition %s cannot be written, and is not handed: %v", key, err))
continue
}
c.rev++
c.handed[key] = dig
out = append(out, agent.Update{Key: key, Value: raw, Revision: c.rev})
}
for key := range c.handed {
if !now[key] {
c.rev++
delete(c.handed, key)
out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev})
}
}
return out
}
// refused tells the agent why the node cannot compose, or "" that it composes again, once per change.
func (c *Composer) refused(why string) {
c.mu.Lock()
same := c.refusal == why
c.refusal = why
c.mu.Unlock()
if !same && c.env.Refused != nil {
c.env.Refused(why)
}
}
// keepRunning is comp with each sealed value the node already runs at the same slot kept, where it opens to the same
// value (delivery.KeepSeals, issue 1126, C1): of what runs (running, the agent's), else of what its module's have says
// it runs (Held). A have written before slots names no slot: its boxes are reused within their own module alone, at a
// slot no box is kept for, and only one opening to the slot's value.
func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Composed) delivery.Composed {
if c.env.Unseal == nil {
return comp
}
module := comp.Module
if module == "" {
module = agent.NodeModule
}
kept := delivery.Seals{}
var legacy []string
if c.env.Held != nil {
slots, old := c.env.Held(module)
for slot, box := range slots {
kept[slot] = box
}
legacy = old
}
for _, r := range running {
if r.Module == comp.Module {
for slot, box := range delivery.SealsIn(r) {
kept[slot] = box
}
}
}
for slot, box := range delivery.SealsIn(comp) {
if _, ok := kept[slot]; ok {
continue
}
for _, h := range legacy {
if same, known := c.sameValue(box, h); known && same {
kept[slot] = h
break
}
}
}
return delivery.KeepSeals(comp, kept, c.sameValue)
}
// waiting notes what the composer waits on until the returned func is called, which restores what it waited on before.
func (c *Composer) waiting(what string) func() {
c.doingMu.Lock()
before := c.doing
c.doing = what
c.doingMu.Unlock()
return func() {
c.doingMu.Lock()
c.doing = before
c.doingMu.Unlock()
}
}
// waitingOn is what the composer waits on now, in words.
func (c *Composer) waitingOn() string {
c.doingMu.Lock()
defer c.doingMu.Unlock()
if c.doing == "" {
return "nothing it reads (composing)"
}
return c.doing
}
func (c *Composer) sayOnce(line string) {
c.mu.Lock()
if c.said == line {
c.mu.Unlock()
return
}
c.said = line
c.mu.Unlock()
if line != "" {
c.env.Say(line)
}
}
// Policy is a module's upgrade policy as its manifest at the indexer's trunk derives it (catalogue.DerivedUpgrade), for
// the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read. The manifest is
// read within ctx, the agent's (review of L2, round 1, finding 5), so a stopping agent stops the read.
func (c *Composer) Policy(ctx context.Context, module string) string {
c.mu.Lock()
m, ok := c.modules[module]
c.mu.Unlock()
if !ok || m.Manifest == "" {
return ""
}
raw, err := reader{c: c}.Manifest(ctx, m.Manifest)
if err != nil {
return ""
}
parsed, err := catalogue.ParseManifestForInputs(raw)
if err != nil {
return ""
}
policy, _, _ := catalogue.DerivedUpgrade(parsed)
return policy
}
// reader is the composer's view of the bus as frombus reads it.
type reader struct{ c *Composer }
func (r reader) Placements(context.Context) (map[string]delivery.Placement, error) {
r.c.mu.Lock()
defer r.c.mu.Unlock()
out := make(map[string]delivery.Placement, len(r.c.placements))
for k, v := range r.c.placements {
out[k] = v
}
return out, nil
}
func (r reader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
r.c.mu.Lock()
defer r.c.mu.Unlock()
out := make(map[string]delivery.NodeRecord, len(r.c.records))
for k, v := range r.c.records {
out[k] = v
}
return out, nil
}
func (r reader) Modules(context.Context) (map[string]delivery.Module, error) {
r.c.mu.Lock()
defer r.c.mu.Unlock()
out := make(map[string]delivery.Module, len(r.c.modules))
for k, v := range r.c.modules {
out[k] = v
}
return out, nil
}
func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) {
r.c.mu.Lock()
raw, ok := r.c.manifests[digest]
r.c.mu.Unlock()
if ok {
return raw, nil
}
key, err := delivery.ManifestsKey(digest)
if err != nil {
return nil, err
}
defer r.c.waiting("the manifest " + digest + " (" + delivery.ManifestsBucket + ")")()
value, err := r.c.env.Get(ctx, delivery.ManifestsBucket, key)
if err != nil {
return nil, err
}
if value == nil {
return nil, fmt.Errorf("the indexer holds no manifest %s", digest)
}
m, err := delivery.DecodeManifest(value)
if err != nil {
return nil, err
}
r.c.mu.Lock()
r.c.manifests[digest] = []byte(m.JSON)
r.c.mu.Unlock()
return []byte(m.JSON), nil
}
func (r reader) Build(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error) {
key, err := delivery.BuilderBuildKey(module, fp)
if err != nil {
return nil, err
}
defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")()
raw, err := r.c.env.Get(ctx, delivery.BuilderBuildsBucket, key)
if err != nil || raw == nil {
return nil, err
}
b, err := delivery.DecodeBuilderBuild(raw)
if err != nil {
return nil, fmt.Errorf("the builder's %s: %w", key, err)
}
return &b, nil
}
// archiveWithin bounds one read of a templates archive in the artifact store.
const archiveWithin = 30 * time.Second
// Archive reads a templates archive once per digest (an archive is the whole of its digest), checked against it.
func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error) {
r.c.mu.Lock()
body, ok := r.c.archives[digest]
r.c.mu.Unlock()
if ok {
return body, nil
}
if r.c.env.Fetch == nil {
return nil, fmt.Errorf("this node reads no archive (%s)", url)
}
// Bounded: a store that does not answer must not hold the composer or the node's answer (issue 1126).
defer r.c.waiting("the archive " + url)()
fetching, cancel := context.WithTimeout(ctx, archiveWithin)
defer cancel()
body, err := r.c.env.Fetch(fetching, url)
if err != nil {
return nil, err
}
sum := sha256.Sum256(body)
if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest {
return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got)
}
r.c.mu.Lock()
r.c.archives[digest] = body
r.c.mu.Unlock()
return body, nil
}
// Served reads, of another node's module, the keys named of mesh-config's resolved values, one key each: the node is
// granted those keys alone (buscompose Principal.ServedReads, issue 1126).
func (r reader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) {
defer r.c.waiting("what " + module + " serves on " + node + " (" + ResolvedBucket + ")")()
out := map[string]any{}
for _, k := range keys {
raw, err := r.c.env.Get(ctx, ResolvedBucket, node+"."+module+"."+k)
if err != nil {
return nil, err
}
value, laid := catalogue.LaidValue(raw)
if raw == nil || !laid {
continue
}
out[k] = value
}
if len(out) == 0 {
return nil, nil
}
return out, nil
}
func (r reader) Values(_ context.Context, node, module string) (map[string]any, error) {
if node != r.c.env.Node {
return nil, nil // a node reads its own slice alone (ADR 0547 §2)
}
r.c.mu.Lock()
defer r.c.mu.Unlock()
if len(r.c.values[module]) == 0 {
return nil, nil
}
out := make(map[string]any, len(r.c.values[module]))
for k, v := range r.c.values[module] {
out[k] = v
}
return out, nil
}
func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecretRow, error) {
if node != r.c.env.Node || r.c.env.Own == nil {
return nil, nil
}
done := r.c.waiting("the vault's own secrets")
raw, err := r.c.env.Own(ctx, map[string]any{})
done()
if err != nil {
return nil, fmt.Errorf("the vault's own: %w", err)
}
var a struct {
Node string `json:"node"`
Key string `json:"key"`
Own []struct {
Module string `json:"module"`
Name string `json:"name"`
Origin string `json:"origin"`
Generation int `json:"generation"`
Sealed json.RawMessage `json:"sealed"`
} `json:"own"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return nil, fmt.Errorf("the vault's own answer is not one: %w", err)
}
if a.Node != node {
return nil, fmt.Errorf("the vault answered %q's own secrets to %s", a.Node, node)
}
r.c.mu.Lock()
defer r.c.mu.Unlock()
out := make([]compose.OwnSecretRow, 0, len(a.Own))
for _, o := range a.Own {
// The box first had is kept while its generation stands, so sealing it again moves no composition; the box the
// node runs at the same slot is kept after the composition (keepRunning).
id := o.Module + "/" + o.Name
kept, ok := r.c.sealedOwn[id]
if !ok || kept.generation != o.Generation || kept.key != a.Key {
kept = ownSealed{generation: o.Generation, key: a.Key, sealed: string(o.Sealed)}
r.c.sealedOwn[id] = kept
}
out = append(out, compose.OwnSecretRow{Node: node, Module: o.Module, Name: o.Name, Origin: o.Origin,
SealedTo: a.Key, Sealed: kept.sealed})
}
return out, nil
}
func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProvider []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
if node != r.c.env.Node || r.c.env.Own == nil {
return nil, nil, nil
}
type asked struct {
Role string `json:"role"`
compose.CredentialKey
}
var ask []asked
for _, k := range asConsumer {
ask = append(ask, asked{Role: "consumer", CredentialKey: k})
}
for _, k := range asProvider {
ask = append(ask, asked{Role: "provider", CredentialKey: k})
}
done := r.c.waiting("the vault's credentials")
raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask})
done()
if err != nil {
return nil, nil, fmt.Errorf("the vault's own: %w", err)
}
var a struct {
Node string `json:"node"`
Key string `json:"key"`
Credentials []struct {
Role string `json:"role"`
compose.CredentialKey
Name string `json:"name"`
Generation int `json:"generation"`
Sealed json.RawMessage `json:"sealed"`
} `json:"credentials"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return nil, nil, fmt.Errorf("the vault's own answer is not one: %w", err)
}
if a.Node != node {
return nil, nil, fmt.Errorf("the vault answered %q's credentials to %s", a.Node, node)
}
r.c.mu.Lock()
defer r.c.mu.Unlock()
var creds []compose.Credential
var issued []compose.Issued
for _, c := range a.Credentials {
// The value first had is kept while its generation stands, so sealing it again moves no composition; the box the
// node runs at the same slot is kept after the composition (keepRunning).
id := c.Role + "|" + c.Name + "|" + c.Provision + "|" + c.Consumer + "|" + c.ConsumerModule + "|" + c.Local
kept, ok := r.c.sealedOwn[id]
if !ok || kept.generation != c.Generation || kept.key != a.Key {
kept = ownSealed{generation: c.Generation, key: a.Key, sealed: string(c.Sealed)}
r.c.sealedOwn[id] = kept
}
k := c.CredentialKey
switch c.Role {
case "consumer":
creds = append(creds, compose.Credential{Provision: k.Provision, Consumer: k.Consumer, ConsumerModule: k.ConsumerModule,
Provider: k.Provider, Local: k.Local, State: compose.CredentialHeld, Sealed: kept.sealed})
case "provider":
issued = append(issued, compose.Issued{Provision: k.Provision, Consumer: k.Consumer, ConsumerModule: k.ConsumerModule,
Local: k.Local, Provider: k.Provider, ForProvider: kept.sealed})
}
}
return creds, issued, nil
}
// Placed reads whether the catalogue places node now, for the node agent choosing whose compositions it applies at
// each start: true when its placement key holds a value, false only when a read succeeds and finds no key or a delete
// marker. A read that fails chooses neither (review of the shadow, round 1: a timeout or a bus not ready at a restart
// flipped a self-composing node to the controller's compositions for a whole run): it is tried again after wait(try),
// each distinct failure said once, until ctx ends, which is then its error.
func Placed(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string,
wait func(try int) time.Duration, say func(string)) (bool, error) {
key, err := delivery.PlacementKey(node)
if err != nil {
return false, err
}
said := ""
for try := 1; ; try++ {
raw, err := get(ctx, delivery.PlacementBucket, key)
if errors.Is(err, nats.ErrBucketNotFound) {
return false, nil // the indexer made no placement state yet: nothing places it, read and known
}
if err == nil {
return len(raw) > 0, nil
}
if ctx.Err() != nil {
return false, ctx.Err()
}
if why := err.Error(); why != said {
say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err))
said = why
}
select {
case <-ctx.Done():
return false, ctx.Err()
case <-time.After(wait(try)):
}
}
}
// composeAgain are the waits before a pass that met a bus not answering is tried again.
var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}
// Transient says err is the bus not answering for a moment, as a reload of its accounts makes it for a few seconds:
// JetStream not enabled, or no responders; each fails at once, so a pass tried again costs only its wait. A read that
// timed out is not one: it already waited longer than such a pause, and is most often a read the bus refuses. Matched
// by value and by words, since the vault's and the reads' errors arrive wrapped in words (issue 1126).
func Transient(err error) bool {
if err == nil {
return false
}
if errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) || errors.Is(err, nats.ErrNoResponders) {
return true
}
words := err.Error()
for _, w := range []string{"jetstream not enabled", "no responders"} {
if strings.Contains(words, w) {
return true
}
}
return false
}