The fold's watch of every have key ('>') was refused (permissions violation on
CONSUMER.CREATE KV_node-engine_have), and every self-composing agent stopped
after the seed. The composer now watches '*.node' (records) and '<node>.>' (its
own haves), as buscompose grants a node; the shadow watches '<node>.>' too. The
selfcompose tests check every watch and get against PermissionsFor(KindNode),
and the agent's integration tests run the agent as the node under that grant.
806 lines
25 KiB
Go
806 lines
25 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"
|
|
"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
|
|
}
|
|
|
|
// 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
|
|
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{}, 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 and is not removed (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 {
|
|
in = h.Phase != delivery.PhaseRemoved && !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()
|
|
sort.Strings(behind)
|
|
for _, module := range behind {
|
|
digest, err := c.builtManifest(ctx, module, fps[module])
|
|
if err != nil {
|
|
why := fmt.Sprintf("%s is wanted at %s, whose manifest cannot be read, so the node does not compose: %v", module, fps[module], err)
|
|
c.refused(why)
|
|
c.sayOnce(why)
|
|
return nil
|
|
}
|
|
kept[module] = digest
|
|
}
|
|
|
|
compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, frombus.Scope{Node: c.env.Node, Kept: 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)
|
|
}
|
|
}
|
|
|
|
// builtManifest is the digest of the manifest a module's build at a fingerprint was made from (the builder's build key):
|
|
// what a node composes a module from when its want names another build than the trunk's.
|
|
func (c *Composer) builtManifest(ctx context.Context, module, fp string) (string, error) {
|
|
key, err := delivery.BuilderBuildKey(module, fp)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
raw, err := c.env.Get(ctx, delivery.BuilderBuildsBucket, key)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if raw == nil {
|
|
return "", fmt.Errorf("the builder holds no build %s", key)
|
|
}
|
|
b, err := delivery.DecodeBuilderBuild(raw)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if b.Manifest == "" {
|
|
return "", fmt.Errorf("the build %s names no manifest", key)
|
|
}
|
|
return b.Manifest, nil
|
|
}
|
|
|
|
// 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) 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
|
|
}
|