jschoubben
7365890298
H8 found it: a module the node could not compose yet got a failed have naming no fingerprint, which counted as run inside the node's own composition, so the next composition took it from its manifest past its build.
851 lines
27 KiB
Go
851 lines
27 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 (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
|
|
"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/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 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). 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 (delivery.Have.Sealed), 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.
|
|
Held func(module string) []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
|
|
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()
|
|
var r struct {
|
|
Value json.RawMessage `json:"value"`
|
|
}
|
|
var v any
|
|
if u.Deleted || json.Unmarshal(u.Value, &r) != nil || len(r.Value) == 0 || json.Unmarshal(r.Value, &v) != nil {
|
|
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.
|
|
inOwn := c.inOwnOf(c.placements[c.env.Node])
|
|
r := reader{c: c}
|
|
c.mu.Unlock()
|
|
// Composed from the build its want names (frombus.Scope.KeptAt, issue 1126).
|
|
for _, module := range behind {
|
|
kept[module] = fps[module]
|
|
}
|
|
|
|
compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn}, fps)
|
|
if err != nil {
|
|
c.refused(fmt.Sprintf("%v", err))
|
|
c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err))
|
|
return nil
|
|
}
|
|
c.refused("")
|
|
c.sayOnce("")
|
|
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)
|
|
}
|
|
}
|
|
|
|
// runningBox is the box the node already runs that opens to the same value as box, or box: what the agent runs, and
|
|
// what each wanted module's have says it runs sealed (Held). Read before c.mu is taken, since the agent's lock is taken
|
|
// to read what runs.
|
|
func (c *Composer) runningBox(box string) string {
|
|
if c.env.Unseal == nil || (c.env.Running == nil && c.env.Held == nil) {
|
|
return box
|
|
}
|
|
value, err := c.env.Unseal(box)
|
|
if err != nil {
|
|
return box
|
|
}
|
|
var held []string
|
|
if c.env.Running != nil {
|
|
for _, comp := range c.env.Running() {
|
|
held = append(held, agent.SealedIn(comp)...)
|
|
}
|
|
}
|
|
if c.env.Held != nil {
|
|
c.mu.Lock()
|
|
// The node's own key among them (agent.NodeModule): its core boxes, as its last run said them.
|
|
modules := make([]string, 0, len(c.wants))
|
|
for m := range c.wants {
|
|
modules = append(modules, m)
|
|
}
|
|
c.mu.Unlock()
|
|
sort.Strings(modules)
|
|
for _, m := range modules {
|
|
held = append(held, c.env.Held(m)...)
|
|
}
|
|
}
|
|
for _, h := range held {
|
|
if h == box {
|
|
continue
|
|
}
|
|
if v, err := c.env.Unseal(h); err == nil && bytes.Equal(v, value) {
|
|
return h
|
|
}
|
|
}
|
|
return box
|
|
}
|
|
|
|
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
|
|
}
|
|
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
|
|
}
|
|
raw, err := r.c.env.Get(ctx, delivery.BuilderBuildsBucket, key)
|
|
r.c.env.Say(fmt.Sprintf("DEBUG build %s: %d bytes err=%v", key, len(raw), err))
|
|
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
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
body, err := r.c.env.Fetch(ctx, 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) {
|
|
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
|
|
}
|
|
var v struct {
|
|
Value json.RawMessage `json:"value"`
|
|
}
|
|
var value any
|
|
if raw == nil || json.Unmarshal(raw, &v) != nil || len(v.Value) == 0 || json.Unmarshal(v.Value, &value) != nil {
|
|
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
|
|
}
|
|
raw, err := r.c.env.Own(ctx, map[string]any{})
|
|
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)
|
|
}
|
|
ids := make([]string, len(a.Own))
|
|
fresh := make([]string, len(a.Own))
|
|
for i, o := range a.Own {
|
|
ids[i] = o.Module + "/" + o.Name
|
|
if !r.c.keeps(ids[i], o.Generation, a.Key) {
|
|
fresh[i] = r.c.runningBox(string(o.Sealed))
|
|
}
|
|
}
|
|
r.c.mu.Lock()
|
|
defer r.c.mu.Unlock()
|
|
out := make([]compose.OwnSecretRow, 0, len(a.Own))
|
|
for i, o := range a.Own {
|
|
id := ids[i]
|
|
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: fresh[i]}
|
|
if kept.sealed == "" {
|
|
kept.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})
|
|
}
|
|
raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask})
|
|
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)
|
|
}
|
|
ids := make([]string, len(a.Credentials))
|
|
fresh := make([]string, len(a.Credentials))
|
|
for i, c := range a.Credentials {
|
|
ids[i] = c.Role + "|" + c.Name + "|" + c.Provision + "|" + c.Consumer + "|" + c.ConsumerModule + "|" + c.Local
|
|
if !r.c.keeps(ids[i], c.Generation, a.Key) {
|
|
fresh[i] = r.c.runningBox(string(c.Sealed))
|
|
}
|
|
}
|
|
r.c.mu.Lock()
|
|
defer r.c.mu.Unlock()
|
|
var creds []compose.Credential
|
|
var issued []compose.Issued
|
|
for i, c := range a.Credentials {
|
|
// The value first had is kept while its generation stands, so sealing it again moves no composition; and the
|
|
// first had is the box the node runs when it opens to the same value (runningBox).
|
|
id := ids[i]
|
|
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: fresh[i]}
|
|
if kept.sealed == "" {
|
|
kept.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
|
|
}
|
|
|
|
// keeps says the composer holds id's sealed value at this generation and key already.
|
|
func (c *Composer) keeps(id string, generation int, key string) bool {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
kept, ok := c.sealedOwn[id]
|
|
return ok && kept.generation == generation && kept.key == key
|
|
}
|