The node composes its own modules once the catalogue places it, and writes its record (issue 1126) #149
@@ -0,0 +1,39 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// authorityPath is where this engine keeps the mesh's own certificate as it last applied it (public): beside the
|
||||
// certificate issued for the machine's name, read into its record (novox/hq ADR 0555 §3).
|
||||
func authorityPath(state string) string { return filepath.Join(filepath.Dir(state), "authority.crt") }
|
||||
|
||||
// keepAuthority keeps the mesh's own certificate a composition carries (the resource a module's certificate.authority
|
||||
// asks for, catalogue.AuthorityID), so the record says the authority this machine holds. A composition carrying none
|
||||
// changes nothing.
|
||||
func keepAuthority(state string, c delivery.Composed) error {
|
||||
for _, raw := range c.Resources {
|
||||
var r struct {
|
||||
ID string `json:"id"`
|
||||
Content string `json:"content"`
|
||||
}
|
||||
if json.Unmarshal(raw, &r) != nil || r.Content == "" || !strings.HasSuffix(r.ID, "."+catalogue.AuthorityID()) {
|
||||
continue
|
||||
}
|
||||
path := authorityPath(state)
|
||||
if was, err := os.ReadFile(path); err == nil && string(was) == r.Content {
|
||||
return nil
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, []byte(r.Content), 0o644)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// Issue 1126 (ADR 0555 §3): the mesh's own certificate a composition carries is kept for the machine's record, and the
|
||||
// record's keys read it; a composition carrying none changes nothing.
|
||||
func TestTheAppliedAuthorityIsKeptForTheRecord(t *testing.T) {
|
||||
state := filepath.Join(t.TempDir(), "state", "state.json")
|
||||
res := func(id, content string) json.RawMessage {
|
||||
raw, _ := json.Marshal(map[string]any{"id": id, "type": "file", "path": "/etc/x", "content": content})
|
||||
return raw
|
||||
}
|
||||
if err := keepAuthority(state, delivery.Composed{Resources: []json.RawMessage{res("web.conf", "x")}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := os.Stat(authorityPath(state)); !os.IsNotExist(err) {
|
||||
t.Fatalf("a composition with no authority kept one: %v", err)
|
||||
}
|
||||
if err := keepAuthority(state, delivery.Composed{Resources: []json.RawMessage{res("ca-trust.certificate-authority", "CA-1")}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := recordKeys(state)(); got.Authority != "CA-1" {
|
||||
t.Errorf("the record's keys read the authority %q", got.Authority)
|
||||
}
|
||||
}
|
||||
@@ -1115,6 +1115,8 @@ func runLink(ctx context.Context, opts options) error {
|
||||
proof := &proving{statePath: opts.state, version: runningVersion(), say: say}
|
||||
queue := &link.Session{
|
||||
Membership: membership,
|
||||
// And the public halves and certificate its record says (novox/hq ADR 0555 §3), read at each facts report.
|
||||
Keys: recordKeys(opts.state),
|
||||
// And the first have this build writes is what proves it to the launcher (ADR 0539 §9, ADR 0227 §8).
|
||||
HaveWritten: proof.haveWritten,
|
||||
// And the operator's hand-over of a directory used as found, carried by the controller from the
|
||||
@@ -2131,3 +2133,28 @@ func newlyMissing(last map[string]bool, st liveness.Statement) ([]string, map[st
|
||||
}
|
||||
return missingModules(fresh), now
|
||||
}
|
||||
|
||||
// recordKeys reads the public halves of the node's keys and the certificate the mesh issued for its internal name from
|
||||
// its state directory (novox/hq ADR 0555 §3): what the node's record says of them. A key that cannot be read is left
|
||||
// empty, never invented; the certificate is there only after the first declaration that carried it.
|
||||
func recordKeys(state string) func() link.RecordKeys {
|
||||
return func() link.RecordKeys {
|
||||
var k link.RecordKeys
|
||||
if id, err := identity.Load(identity.Path(state)); err == nil {
|
||||
k.TunnelKey = id.Overlay.Public
|
||||
}
|
||||
if sk, err := identity.LoadSealingKey(identity.SealingKeyPath(state)); err == nil {
|
||||
k.SealingKey = sk.Public
|
||||
}
|
||||
if sv, err := identity.LoadServingKey(identity.ServingKeyPath(state)); err == nil {
|
||||
k.ServingKey = sv.Public
|
||||
}
|
||||
if cert, err := os.ReadFile(identity.CertificatePath(state)); err == nil {
|
||||
k.Certificate = string(cert)
|
||||
}
|
||||
if authority, err := os.ReadFile(authorityPath(state)); err == nil {
|
||||
k.Authority = string(authority)
|
||||
}
|
||||
return k
|
||||
}
|
||||
}
|
||||
|
||||
+128
-3
@@ -13,11 +13,14 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
||||
"git.novox.be/novox/mesh-sdk/go/declaration"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/overlay"
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
"github.com/novox/mesh-host/internal/apply"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
"github.com/novox/mesh-host/internal/selfcompose"
|
||||
"github.com/novox/mesh-host/internal/store"
|
||||
)
|
||||
|
||||
@@ -57,7 +60,8 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool
|
||||
if !ok {
|
||||
return agent.Have{}, false
|
||||
}
|
||||
return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail}, true
|
||||
return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail,
|
||||
Sealed: got.Sealed}, true
|
||||
}
|
||||
|
||||
func (h sessionHaves) Write(ctx context.Context, x agent.Have) {
|
||||
@@ -73,7 +77,8 @@ func (h sessionHaves) Write(ctx context.Context, x agent.Have) {
|
||||
phase = ""
|
||||
}
|
||||
h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase,
|
||||
Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail)})
|
||||
Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail), Judgement: x.Judgement, Waits: x.Waits,
|
||||
Sealed: x.Sealed})
|
||||
}
|
||||
|
||||
func (h sessionHaves) Removed(ctx context.Context, module string) {
|
||||
@@ -401,6 +406,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
||||
}
|
||||
var ag *agent.Agent
|
||||
kept := keptOnce(opts, signer)
|
||||
// The mesh's own certificate the declaration kept from before carries, kept for the record before this machine
|
||||
// composes its own (ADR 0555 §3): without it a module that asks for a certificate would be stood in for.
|
||||
if byModule, _ := kept(); byModule != nil {
|
||||
for module, resources := range byModule {
|
||||
if err := keepAuthority(opts.state, delivery.Composed{Module: module, Resources: resources}); err != nil {
|
||||
say("the mesh's certificate the kept declaration carries could not be kept for this machine's record: " + err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
// The bus's user list, written alone from the newest composition of the bus module (ADR 0547 §8): on every
|
||||
// node, and a node whose compositions carry no list writes nothing.
|
||||
users := userListFollower(opts, session, composedBucket, trustedKeys(opts, signer), say, func() {
|
||||
@@ -459,6 +473,12 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
||||
res = applyComposition(actx, opts, held, others, sched, say)
|
||||
}
|
||||
done()
|
||||
// The mesh's own certificate it carries is kept for this machine's record (ADR 0555 §3).
|
||||
if !res.Failed {
|
||||
if err := keepAuthority(opts.state, c); err != nil {
|
||||
say("the mesh's certificate in " + c.Module + "'s composition could not be kept for this machine's record: " + err.Error())
|
||||
}
|
||||
}
|
||||
// An apply of the bus module may write an older list than the newest composition holds: looked at again.
|
||||
if module, _ := users.State(); module != "" && module == c.Module {
|
||||
users.Again()
|
||||
@@ -476,9 +496,44 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
||||
return res
|
||||
},
|
||||
}
|
||||
// The node composes its own modules once the catalogue places it (novox/hq ADR 0547 §2, ADR 0549, issue 1126): its
|
||||
// placement key is read at the agent's start; until then it applies the controller's compositions as before.
|
||||
run, stopRun := context.WithCancel(linked)
|
||||
// What the node runs and how it opens a sealed value, for the composer to keep the boxes it runs; and the agent
|
||||
// told when the node cannot compose, so each wanted module's have says it.
|
||||
hooks := composerHooks{
|
||||
running: func() []delivery.Composed {
|
||||
if now := current.get(); now != nil {
|
||||
return now.Everything()
|
||||
}
|
||||
return nil
|
||||
},
|
||||
unseal: sealOpener(opts.state),
|
||||
// And what each module's have says it runs sealed: after a restart the agent has applied nothing yet.
|
||||
held: func(module string) []string {
|
||||
got, ok := session.ReadHave(linked, module)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return got.Sealed
|
||||
},
|
||||
refused: func(why string) {
|
||||
if now := current.get(); now != nil {
|
||||
now.NotComposed(linked, why)
|
||||
}
|
||||
},
|
||||
}
|
||||
if composer := selfComposer(linked, session, node, say, hooks); composer != nil {
|
||||
env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy
|
||||
} else {
|
||||
// And once the catalogue places it, the agent begins again composing its own: the placement key appearing ends
|
||||
// this run, so a node does not wait for a restart to follow the seed of its placement (ADR 0549 §4).
|
||||
go endWhenPlaced(run, session, node, stopRun, say)
|
||||
}
|
||||
ag = agent.New(env)
|
||||
current.set(ag)
|
||||
err = ag.Run(linked)
|
||||
err = ag.Run(run)
|
||||
stopRun()
|
||||
cancel()
|
||||
current.set(nil)
|
||||
if ctx.Err() != nil {
|
||||
@@ -756,3 +811,73 @@ const applyBound = 30 * time.Minute
|
||||
func applyContext(ctx context.Context) (context.Context, context.CancelFunc) {
|
||||
return context.WithTimeout(context.WithoutCancel(ctx), applyBound)
|
||||
}
|
||||
|
||||
// selfComposer is the node's own composer when the catalogue places the node (its key `placement.<node>` in the
|
||||
// indexer's state), and nil, said once, while it does not or the key cannot be read.
|
||||
// composerHooks are what the composer reads of the agent and the machine (selfcompose.Env Running, Unseal, Refused).
|
||||
type composerHooks struct {
|
||||
running func() []delivery.Composed
|
||||
unseal func(string) ([]byte, error)
|
||||
refused func(string)
|
||||
held func(module string) []string
|
||||
}
|
||||
|
||||
func selfComposer(ctx context.Context, session *link.Session, node string, say func(string), hooks composerHooks) *selfcompose.Composer {
|
||||
key, err := delivery.PlacementKey(node)
|
||||
if err != nil {
|
||||
say("this node applies the controller's compositions: " + err.Error())
|
||||
return nil
|
||||
}
|
||||
got, err := session.KVHistory(ctx, delivery.PlacementBucket, key)
|
||||
if err != nil || len(got) == 0 || got[len(got)-1].Deleted {
|
||||
why := "the catalogue does not place it yet"
|
||||
if err != nil {
|
||||
why = "its placement cannot be read: " + err.Error()
|
||||
}
|
||||
say("this node applies the controller's compositions: " + why)
|
||||
return nil
|
||||
}
|
||||
say("this node composes its own modules: the catalogue places it (" + key + ")")
|
||||
return selfcompose.New(selfcompose.Env{Node: node,
|
||||
Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) {
|
||||
return bucketWatch{session, bucket}.Watch(ctx, pattern)
|
||||
},
|
||||
Get: func(ctx context.Context, bucket, key string) ([]byte, error) {
|
||||
h, err := session.KVHistory(ctx, bucket, key)
|
||||
if err != nil || len(h) == 0 || h[len(h)-1].Deleted {
|
||||
return nil, err
|
||||
}
|
||||
return h[len(h)-1].Value, nil
|
||||
},
|
||||
Mesh: frombus.Mesh{InternalSuffix: overlay.Suffix()},
|
||||
Say: say,
|
||||
// Its modules' own secrets, sealed by the vault to this machine's key (ADR 0550 §5).
|
||||
Own: func(ctx context.Context, args map[string]any) (json.RawMessage, error) {
|
||||
return session.AskVault(ctx, "own", args)
|
||||
},
|
||||
// And the vault's word that a secret changed, on which it composes again (ADR 0547 §2).
|
||||
Hear: session.Hear,
|
||||
Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held})
|
||||
}
|
||||
|
||||
// endWhenPlaced calls stop once this node's placement key appears in the indexer's placement (a value after the watch
|
||||
// caught up, or one already there), so the agent begins again and composes its own. A watch that cannot be made is said;
|
||||
// the node then switches at its next start.
|
||||
func endWhenPlaced(ctx context.Context, session *link.Session, node string, stop func(), say link.Announce) {
|
||||
key, err := delivery.PlacementKey(node)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
updates, err := bucketWatch{session, delivery.PlacementBucket}.Watch(ctx, key)
|
||||
if err != nil {
|
||||
say("whether the catalogue places this node is not followed (" + err.Error() + "): it composes its own from its next start")
|
||||
return
|
||||
}
|
||||
for u := range updates {
|
||||
if !u.Caught && !u.Deleted && len(u.Value) > 0 {
|
||||
say("the catalogue places this node now (" + key + "): the node agent begins again, composing its own modules")
|
||||
stop()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
"github.com/novox/mesh-host/internal/selfcompose"
|
||||
)
|
||||
|
||||
// The node agent whole, against a real bus with JetStream started in the test: the wants bucket, the node's composition
|
||||
@@ -311,3 +312,77 @@ func TestNatsTheHandPathAppliesACompositionTheOperatorSigned(t *testing.T) {
|
||||
return h.FP == fp && h.Composed == dig && h.Phase == delivery.PhaseRunning
|
||||
})
|
||||
}
|
||||
|
||||
// Issue 1126 (ADR 0547 §2, the milestone's H8): the node agent started whole against a real bus, once the catalogue
|
||||
// places its node (the placement written after it started), composes a module from its want, the indexer's manifest, its record and its mesh-config value, and applies it,
|
||||
// with no key in the controller's compositions; a changed config value is composed again on its event and applied as a
|
||||
// revision of what runs.
|
||||
func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
|
||||
b := newItBed(t)
|
||||
kvOf := func(bucket string) nats.KeyValue {
|
||||
kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 5})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return kv
|
||||
}
|
||||
placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket)
|
||||
resolved := kvOf(selfcompose.ResolvedBucket)
|
||||
kvOf(delivery.BuilderBuildsBucket)
|
||||
path := filepath.Join(b.dir, "web.conf")
|
||||
man := `{"module":"web","version":"1","config":{"greeting":{"type":"string","description":"what the file says","default":"hello"}},` +
|
||||
`"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0644","content":"greeting=${config:greeting}\n"}]}`
|
||||
digest := delivery.ManifestDigest([]byte(man))
|
||||
mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "web", JSON: man})
|
||||
mkey, _ := delivery.ManifestsKey(digest)
|
||||
fp := fpIt("a")
|
||||
modv, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40)})
|
||||
place, _ := delivery.EncodePlacement(delivery.Placement{Node: itNode, Modules: []delivery.PlacedModule{{Module: "web"}}})
|
||||
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: itNode, Capabilities: []string{"service-manager"}}})
|
||||
pkey, _ := delivery.PlacementKey(itNode)
|
||||
hkey, _ := delivery.HaveNodeKey(itNode)
|
||||
for _, w := range []struct {
|
||||
kv nats.KeyValue
|
||||
key string
|
||||
value []byte
|
||||
}{{manifests, mkey, mraw}, {modules, "web", modv}, {b.haves, hkey, rec},
|
||||
{resolved, itNode + ".web.greeting", []byte(`{"value":"bonjour"}`)}} {
|
||||
if _, err := w.kv.Put(w.key, w.value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
|
||||
wkey, _ := delivery.WantKey(itNode, "web")
|
||||
if _, err := b.wants.Put(wkey, want); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
stop := b.start(false)
|
||||
defer stop()
|
||||
// Not placed yet: the agent applies the controller's compositions, of which there are none. The catalogue's seed then
|
||||
// places the node, and the agent begins again composing its own (ADR 0549 §4), with no restart of the engine.
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
if b.file("web") != "" {
|
||||
t.Fatal("a node the catalogue does not place composed its own")
|
||||
}
|
||||
if _, err := placement.Put(pkey, place); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
within(t, 5*time.Second, "the node did not compose and apply its own module once placed", func() bool { return b.file("web") == "greeting=bonjour\n" })
|
||||
within(t, 2*time.Second, "the have does not name the build", func() bool {
|
||||
h := b.have("web")
|
||||
return h.FP == fp && h.Composed != "" && h.Phase == delivery.PhaseRunning
|
||||
})
|
||||
if keys, _ := b.composed.Keys(); len(keys) != 0 {
|
||||
t.Errorf("the controller's compositions hold %v; the node composed its own", keys)
|
||||
}
|
||||
first := b.have("web").Composed
|
||||
if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
within(t, 3*time.Second, "a changed config value was not composed and applied again", func() bool { return b.file("web") == "greeting=salut\n" })
|
||||
within(t, 2*time.Second, "the have does not name the revision", func() bool {
|
||||
h := b.have("web")
|
||||
return h.FP == fp && h.Composed != first && h.Phase == delivery.PhaseRunning
|
||||
})
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134601-17b7cc480d6d
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
golang.org/x/crypto v0.57.0
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134601-17b7cc480d6d h1:7F/dFerm01wvuGLPgJ67/VfdcvDexc+BQEE3qzOrlu4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026134601-17b7cc480d6d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
+249
-14
@@ -23,6 +23,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
@@ -49,9 +50,16 @@ type Have struct {
|
||||
Module string
|
||||
FP string
|
||||
Composed string
|
||||
// Phase is "running", "failed", "held" or "removed"; empty reads as running.
|
||||
// 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
|
||||
// (delivery.Have.Sealed): kept in the have so the composer reuses them after a restart (ADR 0547 §3).
|
||||
Sealed []string
|
||||
}
|
||||
|
||||
// Haves is the node's have bucket as the agent uses it.
|
||||
@@ -120,6 +128,14 @@ type Env struct {
|
||||
// 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(module string) string
|
||||
}
|
||||
|
||||
// transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has
|
||||
@@ -143,7 +159,9 @@ type Agent struct {
|
||||
// prior is a module the node ran before this agent started, so it has something to back up.
|
||||
prior map[string]bool
|
||||
said map[string]string
|
||||
retry chan []string
|
||||
// notComposed is the modules whose have says the node could not compose them: why, and the have before (NotComposed).
|
||||
notComposed map[string]notComposedMark
|
||||
retry 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
|
||||
@@ -189,7 +207,7 @@ func New(env Env) *Agent {
|
||||
env.Say = func(string) {}
|
||||
}
|
||||
return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{}, applied: map[string]compNow{},
|
||||
attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{},
|
||||
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), tries: map[string]int{}, again: make(chan againAsk, 64),
|
||||
transient: make(chan struct{}, 1), stopped: make(chan struct{})}
|
||||
}
|
||||
@@ -290,6 +308,141 @@ func (a *Agent) RetryTransient() {
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// 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.mu.Lock()
|
||||
var modules []string
|
||||
if why != "" {
|
||||
for module, w := range a.wants {
|
||||
if w.Present && !w.Want.Removed && module != NodeModule && a.notComposed[module].why != why {
|
||||
modules = append(modules, module)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
for module := range a.notComposed {
|
||||
modules = append(modules, module)
|
||||
}
|
||||
}
|
||||
a.mu.Unlock()
|
||||
sort.Strings(modules)
|
||||
for _, module := range modules {
|
||||
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, ok := a.env.Haves.Read(ctx, module); ok {
|
||||
mark.prior = h // what its last have said runs: this agent has run nothing of it since its start
|
||||
}
|
||||
}
|
||||
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: "the node cannot compose it, and keeps running what it ran: " + why})
|
||||
}
|
||||
}
|
||||
|
||||
// writeHave writes a module's have; 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
|
||||
}
|
||||
}
|
||||
}
|
||||
if a.env.SelfComposed && h.Sealed == nil && h.Module != NodeModule && 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) []string {
|
||||
a.mu.Lock()
|
||||
if c, ok := a.applied[module]; ok && c.Err == nil && c.Dig == composed {
|
||||
a.mu.Unlock()
|
||||
return SealedIn(c.C)
|
||||
}
|
||||
if c, ok := a.compositionFor(module, fp); ok && c.Err == nil && c.Dig == composed {
|
||||
a.mu.Unlock()
|
||||
return SealedIn(c.C)
|
||||
}
|
||||
a.mu.Unlock()
|
||||
if h, ok := a.env.Haves.Read(ctx, module); ok && h.Composed == composed {
|
||||
return h.Sealed
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// SealedIn is each value a composition carries sealed to its node's key: a file's `sealed` and a vault box
|
||||
// (catalogue.VaultSealedField), as their JSON text, in the order of its resources.
|
||||
func SealedIn(c delivery.Composed) []string {
|
||||
var out []string
|
||||
for _, raw := range c.Resources {
|
||||
var r map[string]json.RawMessage
|
||||
if json.Unmarshal(raw, &r) != nil {
|
||||
continue
|
||||
}
|
||||
for _, field := range []string{"sealed", catalogue.VaultSealedField} {
|
||||
f, ok := r[field]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
var text string
|
||||
if json.Unmarshal(f, &text) != nil {
|
||||
text = string(f) // a vault box is carried as its JSON
|
||||
}
|
||||
if text != "" {
|
||||
out = append(out, text)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// 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 {
|
||||
@@ -366,7 +519,14 @@ func (a *Agent) takeComposition(u Update) string {
|
||||
}
|
||||
now := compNow{Raw: u.Value, Rev: u.Revision}
|
||||
c, err := delivery.DecodeComposed(u.Value)
|
||||
if err == nil {
|
||||
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()
|
||||
@@ -636,7 +796,16 @@ func (a *Agent) settleOne(ctx context.Context, module string) {
|
||||
a.mu.Lock()
|
||||
comp, have := a.compositionFor(module, w.Want.FP)
|
||||
a.mu.Unlock()
|
||||
if (!have || comp.Err != nil || comp.Dig != w.Want.Composed) && w.Want.Composed != "" {
|
||||
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 {
|
||||
@@ -736,8 +905,8 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
|
||||
a.mu.Unlock()
|
||||
// Said again, so the session holds the module and its key carries the health this engine judges,
|
||||
// not the one an older engine left there (xorg's unknown on g14 and shanks, 2026-10-25).
|
||||
a.env.Haves.Write(ctx, Have{Module: module, FP: h.FP, Composed: h.Composed,
|
||||
Phase: string(delivery.PhaseRunning), Detail: h.Detail})
|
||||
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
|
||||
@@ -759,7 +928,7 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
|
||||
a.mu.Lock()
|
||||
a.running[module] = delivery.Have{FP: w.Want.FP, Composed: w.Want.Composed, Phase: delivery.PhaseRunning}
|
||||
a.mu.Unlock()
|
||||
a.env.Haves.Write(ctx, Have{Module: module, FP: w.Want.FP, Composed: w.Want.Composed, Phase: string(delivery.PhaseRunning)})
|
||||
a.writeHave(ctx, Have{Module: module, FP: w.Want.FP, Composed: w.Want.Composed, Phase: string(delivery.PhaseRunning)})
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -773,7 +942,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
|
||||
// A data module is backed up before a new composition of it moves (ADR 0429 §14), unless the want says to go on
|
||||
// without one (ADR 0539 §16).
|
||||
if comp.C.Data && ranBefore && a.env.Backup != nil && w.Want.Backup != delivery.BackupSkip {
|
||||
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseBackingUp)})
|
||||
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()
|
||||
@@ -781,12 +950,12 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
|
||||
prev := a.running[module]
|
||||
a.mu.Unlock()
|
||||
// It keeps running what it has: the have says what it holds and why it waits.
|
||||
a.env.Haves.Write(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseHeld),
|
||||
a.writeHave(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseHeld),
|
||||
Detail: got.Words})
|
||||
return
|
||||
}
|
||||
}
|
||||
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)})
|
||||
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)
|
||||
@@ -816,7 +985,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
|
||||
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.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(phase),
|
||||
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
|
||||
}
|
||||
@@ -825,17 +994,18 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
|
||||
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.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: 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()
|
||||
if a.env.Settled != nil {
|
||||
a.env.Settled(module)
|
||||
}
|
||||
a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail})
|
||||
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) {
|
||||
@@ -925,3 +1095,68 @@ func short(s string) string {
|
||||
}
|
||||
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(module) != catalogue.PolicyRecord || w.Want.Yes == comp.Dig {
|
||||
return false
|
||||
}
|
||||
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.
|
||||
func (a *Agent) runningComposition(module string) (delivery.Composed, bool) {
|
||||
if c, ok := a.applied[module]; ok && c.Err == nil {
|
||||
return c.C, true
|
||||
}
|
||||
if a.env.Kept != nil {
|
||||
if rs, ok := a.env.Kept()[module]; ok {
|
||||
return delivery.Composed{Node: a.env.Node, Module: module, Resources: rs}, true
|
||||
}
|
||||
}
|
||||
return delivery.Composed{}, false
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
@@ -326,6 +327,7 @@ func TestStartingOnWhatTheKeptDeclarationHoldsAppliesNothing(t *testing.T) {
|
||||
|
||||
// What the node-engine itself last applied is what runs after a restart, not the kept declaration.
|
||||
func TestStartingOnItsOwnHaveAppliesNothing(t *testing.T) {
|
||||
t.Run("says the have again", testStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines)
|
||||
b := newBed(t)
|
||||
fp := fpOf("a")
|
||||
ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1"))
|
||||
@@ -345,7 +347,7 @@ func TestStartingOnItsOwnHaveAppliesNothing(t *testing.T) {
|
||||
// it and writes the health this engine judges. Not said, its key kept the health an older engine wrote: xorg's have on
|
||||
// g14 and shanks said unknown from before the engine said a module with nothing to judge ok, and the assigner held every
|
||||
// module requiring x11-display back for it (2026-10-25).
|
||||
func TestStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines(t *testing.T) {
|
||||
func testStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines(t *testing.T) {
|
||||
b := newBed(t)
|
||||
fp := fpOf("a")
|
||||
ck, craw, dig := composition(t, "xorg", fp, false, file("xorg", "x", "1"))
|
||||
@@ -545,6 +547,7 @@ func TestABackupDoneIsFollowedByTheApplyAndANewModuleIsNotBackedUp(t *testing.T)
|
||||
|
||||
// R6-A AC4 at the agent: a removed want removes that module only, and its have is written removed.
|
||||
func TestARemovedWantRemovesThatModuleOnly(t *testing.T) {
|
||||
t.Run("its composition holds nothing back", testARemovedModulesCompositionHoldsNothingBack)
|
||||
b := newBed(t)
|
||||
stop := b.run()
|
||||
defer stop()
|
||||
@@ -598,6 +601,7 @@ func TestTheNodesOwnCompositionComesBeforeAnyModules(t *testing.T) {
|
||||
// R6-A AC9: a composition signed by either trusted key is applied; one signed by another key, one changed after it was
|
||||
// signed and one unsigned are refused and nothing of them is applied (ADR 0539 §3).
|
||||
func TestOnlyACompositionSignedByATrustedKeyIsApplied(t *testing.T) {
|
||||
t.Run("with no trusted key", testWithNoTrustedKeyNothingIsApplied)
|
||||
b := newBed(t)
|
||||
stop := b.run()
|
||||
defer stop()
|
||||
@@ -648,7 +652,7 @@ func TestOnlyACompositionSignedByATrustedKeyIsApplied(t *testing.T) {
|
||||
}
|
||||
|
||||
// With no key trusted, nothing is applied: a node that does not know its signers fails closed.
|
||||
func TestWithNoTrustedKeyNothingIsApplied(t *testing.T) {
|
||||
func testWithNoTrustedKeyNothingIsApplied(t *testing.T) {
|
||||
b := newBed(t)
|
||||
b.env.Trusted = nil
|
||||
stop := b.run()
|
||||
@@ -787,7 +791,7 @@ func TestTheNodesOwnApplyHoldsOutAModuleThatMovedToItsOwnComposition(t *testing.
|
||||
|
||||
// A module the assigner removed holds nothing back, its compositions still in the bucket or not: its resources are its
|
||||
// removal's to take (issue 1945).
|
||||
func TestARemovedModulesCompositionHoldsNothingBack(t *testing.T) {
|
||||
func testARemovedModulesCompositionHoldsNothingBack(t *testing.T) {
|
||||
b := newBed(t)
|
||||
fp := fpOf("b")
|
||||
rk, rraw, _ := composition(t, "gone", fp, false, file("gone", "x", "1"))
|
||||
@@ -952,3 +956,133 @@ func TestAModuleRunningFromTheKeptDeclarationKeepsItsFilledPaths(t *testing.T) {
|
||||
t.Fatalf("what runs from the kept declaration: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Issue 1126 (ADR 0547 §2, §3): with the node's own compositions, unsigned, and wants naming the build alone, the
|
||||
// composition composed for the want's build is applied, and only when it differs from what runs; for a module of
|
||||
// policy record a move that recreates a container waits in phase waiting-yes with its judgement until the want names
|
||||
// the yes for that composition, and is then applied.
|
||||
func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(t *testing.T) {
|
||||
b := newBed(t)
|
||||
b.env.SelfComposed, b.env.Trusted = true, nil
|
||||
b.env.Policy = func(m string) string {
|
||||
if m == "db" || m == "rec" {
|
||||
return catalogue.PolicyRecord
|
||||
}
|
||||
return catalogue.PolicyRoll
|
||||
}
|
||||
stop := b.run()
|
||||
defer stop()
|
||||
unsigned := func(module, fp string, rs ...json.RawMessage) (string, []byte, string) {
|
||||
c := delivery.Composed{Node: node, Module: module, FP: fp, Resources: rs}
|
||||
key, _ := delivery.ComposedKey(node, module, fp)
|
||||
raw, err := delivery.EncodeComposed(c)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dig, _ := delivery.ComposedDigestOf(c)
|
||||
return key, raw, dig
|
||||
}
|
||||
want := func(module, fp, yes string) {
|
||||
key, _ := delivery.WantKey(node, module)
|
||||
raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Yes: yes, Self: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.wants.put(key, raw)
|
||||
}
|
||||
|
||||
fp := fpOf("a")
|
||||
pw, _ := json.Marshal(map[string]any{"id": "web.pw", "type": "file", "path": "/tmp/r6/web/pw", "sealed": "1", "module": "web"})
|
||||
ck, craw, dig := unsigned("web", fp, file("web", "x", "1"), pw)
|
||||
b.comps.put(ck, craw)
|
||||
want("web", fp, "")
|
||||
eventually(t, "the node's own composition was not applied", func() bool { return len(b.appliedList()) == 1 })
|
||||
eventually(t, "the have does not name the composed digest", func() bool { return b.haves.last("web").Composed == dig })
|
||||
// The same composition handed again is no move.
|
||||
b.comps.put(ck, craw)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
// The node cannot compose (a stand-in refused): each wanted module's have says so, failed, naming why, and still
|
||||
// names what runs; composed again, the have is what it runs. web was applied by this agent; idle was not since the
|
||||
// agent's start (the switch and every delivery of the node-engine restart it), so what runs is what its have said.
|
||||
idleFP, idleDig := fpOf("c"), "sha256:"+strings.Repeat("9", 64)
|
||||
b.haves.Write(context.Background(), Have{Module: "idle", FP: idleFP, Composed: idleDig, Phase: string(delivery.PhaseRunning),
|
||||
Sealed: []string{`{"box":"idle"}`}})
|
||||
want("idle", idleFP, "")
|
||||
eventually(t, "idle's want was not taken", func() bool {
|
||||
b.ag.mu.Lock()
|
||||
defer b.ag.mu.Unlock()
|
||||
return b.ag.wants["idle"].Present
|
||||
})
|
||||
if h := b.haves.last("web"); len(h.Sealed) != 1 || h.Sealed[0] != "1" {
|
||||
t.Errorf("web's running have does not carry its sealed value: %+v", h)
|
||||
}
|
||||
b.ag.NotComposed(context.Background(), "a stand-in for web's password")
|
||||
for _, row := range []struct {
|
||||
module, fp, composed, sealed string
|
||||
}{{"web", fp, dig, "1"}, {"idle", idleFP, idleDig, `{"box":"idle"}`}} {
|
||||
h := b.haves.last(row.module)
|
||||
if h.Phase != string(delivery.PhaseFailed) || h.FP != row.fp || h.Composed != row.composed ||
|
||||
len(h.Sealed) != 1 || h.Sealed[0] != row.sealed || !strings.Contains(h.Detail, "web's password") {
|
||||
t.Errorf("%s's have while the node cannot compose: %+v", row.module, h)
|
||||
}
|
||||
}
|
||||
b.ag.NotComposed(context.Background(), "")
|
||||
for _, row := range []struct{ module, composed string }{{"web", dig}, {"idle", idleDig}} {
|
||||
if h := b.haves.last(row.module); h.Phase != string(delivery.PhaseRunning) || h.Composed != row.composed || len(h.Sealed) != 1 {
|
||||
t.Errorf("%s's have once the node composes again: %+v", row.module, h)
|
||||
}
|
||||
}
|
||||
// A revision at the same build (a config value changed) is applied.
|
||||
_, craw2, dig2 := unsigned("web", fp, file("web", "x", "2"))
|
||||
b.comps.put(ck, craw2)
|
||||
eventually(t, "the revision was not applied", func() bool { return len(b.appliedList()) == 2 && b.haves.last("web").Composed == dig2 })
|
||||
|
||||
// A record module: its first composition is no move; a new image recreates its container and waits for a yes.
|
||||
container := func(image string) json.RawMessage {
|
||||
module := strings.SplitN(image, "@", 2)[0]
|
||||
raw, _ := json.Marshal(map[string]any{"id": module + ".server", "type": "container", "module": module, "name": module, "image": image})
|
||||
return raw
|
||||
}
|
||||
dk, draw, _ := unsigned("db", fp, container("db@sha256:"+strings.Repeat("1", 64)))
|
||||
b.comps.put(dk, draw)
|
||||
want("db", fp, "")
|
||||
eventually(t, "the record module's first composition was not applied", func() bool { return len(b.appliedList()) == 3 })
|
||||
fpB := fpOf("b")
|
||||
dk2, draw2, ddig2 := unsigned("db", fpB, container("db@sha256:"+strings.Repeat("2", 64)))
|
||||
// rec is a record module this agent has run nothing of since its start (the switch and every delivery of the
|
||||
// node-engine restart it): its have alone says what runs. Its move waits for a yes as db's does, and its have keeps
|
||||
// what runs (fingerprint, digest, sealed values) rather than naming nothing (review of L2, round 3).
|
||||
recFP, recDig := fpOf("d"), "sha256:"+strings.Repeat("8", 64)
|
||||
b.haves.Write(context.Background(), Have{Module: "rec", FP: recFP, Composed: recDig, Phase: string(delivery.PhaseRunning),
|
||||
Sealed: []string{`{"box":"rec"}`}})
|
||||
rk, rraw, rdig := unsigned("rec", fpB, container("rec@sha256:"+strings.Repeat("3", 64)))
|
||||
b.comps.put(dk2, draw2)
|
||||
b.comps.put(rk, rraw)
|
||||
want("db", fpB, "")
|
||||
want("rec", fpB, "")
|
||||
applied := len(b.appliedList())
|
||||
for _, row := range []struct {
|
||||
module, fp, composed, waits string
|
||||
sealed []string
|
||||
}{{"db", fp, "", ddig2, nil}, {"rec", recFP, recDig, rdig, []string{`{"box":"rec"}`}}} {
|
||||
eventually(t, row.module+"'s move did not wait for a yes", func() bool {
|
||||
h := b.haves.last(row.module)
|
||||
return h.Phase == string(delivery.PhaseWaitingYes) && h.Waits == row.waits
|
||||
})
|
||||
h := b.haves.last(row.module)
|
||||
if h.FP != row.fp || (row.composed != "" && h.Composed != row.composed) || h.Composed == "" ||
|
||||
(row.sealed != nil && !slices.Equal(h.Sealed, row.sealed)) {
|
||||
t.Errorf("%s's have waiting for a yes does not keep what runs: %+v", row.module, h)
|
||||
}
|
||||
}
|
||||
if got := b.appliedList(); len(got) != applied {
|
||||
t.Fatalf("a move waiting for a yes was applied: %v", got)
|
||||
}
|
||||
want("db", fpB, ddig2)
|
||||
want("rec", fpB, rdig)
|
||||
eventually(t, "the yes did not apply the move", func() bool {
|
||||
h, r := b.haves.last("db"), b.haves.last("rec")
|
||||
return len(b.appliedList()) == applied+2 && h.Phase == string(delivery.PhaseRunning) && h.FP == fpB &&
|
||||
r.Phase == string(delivery.PhaseRunning) && r.FP == fpB && r.Composed == rdig
|
||||
})
|
||||
}
|
||||
|
||||
+24
-4
@@ -189,8 +189,7 @@ func (q *Session) noteApply(r Report) bool {
|
||||
if r.Health != nil {
|
||||
s.takeLook(r.Health)
|
||||
}
|
||||
n := nodeFacts(r)
|
||||
s.node = &n
|
||||
s.node = q.nodeFactsWithRecord(r, s.node)
|
||||
return true
|
||||
}
|
||||
s.mu.Lock()
|
||||
@@ -242,11 +241,28 @@ func (q *Session) noteApply(r Report) bool {
|
||||
if s.owed() {
|
||||
q.oweHealth()
|
||||
}
|
||||
n := nodeFacts(r)
|
||||
s.node = &n
|
||||
s.node = q.nodeFactsWithRecord(r, s.node)
|
||||
return true
|
||||
}
|
||||
|
||||
// nodeFactsWithRecord is the node's own facts of a report with its record (ADR 0555 §3): a report that says nothing of
|
||||
// the machine leaves the record last made standing, and one without a health statement keeps the contract last stated.
|
||||
func (q *Session) nodeFactsWithRecord(r Report, before *delivery.HaveNode) *delivery.HaveNode {
|
||||
n := nodeFacts(r)
|
||||
if q.Keys != nil {
|
||||
n.Record = recordOf(q.Membership.Node, r, q.Keys())
|
||||
}
|
||||
if before != nil && before.Record != nil {
|
||||
switch {
|
||||
case n.Record == nil:
|
||||
n.Record = before.Record
|
||||
case n.Record.Contract == 0:
|
||||
n.Record.Contract = before.Record.Contract
|
||||
}
|
||||
}
|
||||
return &n
|
||||
}
|
||||
|
||||
// noteHealth takes a look's health, and says whether any module's changed.
|
||||
func (q *Session) noteHealth(h Health) bool {
|
||||
s := &q.haves
|
||||
@@ -308,12 +324,16 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
|
||||
f := s.modules[module]
|
||||
h := delivery.Have{V: delivery.Version, FP: f.FP, Composed: f.Composed, Build: f.Build,
|
||||
Phase: delivery.PhaseRunning, Health: delivery.HealthUnknown, Detail: f.Detail}
|
||||
if delivery.Phase(f.Phase) == delivery.PhaseWaitingYes {
|
||||
h.Judgement, h.Waits = delivery.Judgement(f.Judgement), f.Waits
|
||||
}
|
||||
switch {
|
||||
case f.Phase != "":
|
||||
h.Phase = delivery.Phase(f.Phase)
|
||||
case f.Failed:
|
||||
h.Phase = delivery.PhaseFailed
|
||||
}
|
||||
h.Sealed = f.Sealed
|
||||
h.Health = s.healthOf(module, f)
|
||||
if w, ok := s.userList[module]; ok {
|
||||
h.UserList = &w
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
@@ -259,6 +260,43 @@ func TestTheReportIsSentFirstAndTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
|
||||
if !delivery.SameHaveNode(got, want) {
|
||||
t.Errorf("the node key holds %+v, want %+v", got, want)
|
||||
}
|
||||
|
||||
// 1126 (ADR 0555 §3): with its keys, the node key carries the node's record: the profile's present capabilities and
|
||||
// networks, the outward links, the health's contract and the public keys; a report that changes none of it writes
|
||||
// nothing, a changed fact writes the key once, and a report without a profile or health leaves them standing.
|
||||
keys := RecordKeys{TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk", Certificate: "cert"}
|
||||
q.Keys = func() RecordKeys { return keys }
|
||||
rep.Profile = map[string]any{"capabilities": []any{map[string]any{"name": "systemd", "present": true},
|
||||
map[string]any{"name": "docker", "present": false}, map[string]any{"name": "apt", "present": true}},
|
||||
"machine": map[string]any{"networks": []any{map[string]any{"link": "eth0", "network": "192.0.2.0/24"}}}}
|
||||
rep.Outward = []string{"eth0"}
|
||||
rep.Health = &Health{Contract: NodeKeyContract}
|
||||
q.SayFacts(context.Background(), rep)
|
||||
if got, err = delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || got.Record == nil {
|
||||
t.Fatalf("the node key holds no record: %+v, %v", got, err)
|
||||
}
|
||||
wantRecord := delivery.NodeRecord{V: delivery.Version, Node: "n1", TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk",
|
||||
Certificate: "cert", Capabilities: []string{"apt", "systemd"}, OutwardLinks: []string{"eth0"}, NetworksSaid: true,
|
||||
Networks: []catalogue.AttachedNetwork{{Link: "eth0", Network: "192.0.2.0/24"}}, Contract: NodeKeyContract}
|
||||
if !delivery.SameNodeRecord(*got.Record, wantRecord) {
|
||||
t.Errorf("record = %+v, want %+v", *got.Record, wantRecord)
|
||||
}
|
||||
at := l.kv["n1.node"].rev
|
||||
q.SayFacts(context.Background(), rep)
|
||||
if l.kv["n1.node"].rev != at {
|
||||
t.Errorf("a report changing nothing wrote the node key again")
|
||||
}
|
||||
keys.Certificate = "cert-2"
|
||||
q.SayFacts(context.Background(), rep)
|
||||
if got, _ = delivery.DecodeHaveNode(l.kv["n1.node"].value); l.kv["n1.node"].rev != at+1 || got.Record.Certificate != "cert-2" {
|
||||
t.Errorf("a new certificate: revision %d (was %d), record %+v", l.kv["n1.node"].rev, at, got.Record)
|
||||
}
|
||||
bare := rep
|
||||
bare.Health = nil
|
||||
q.SayFacts(context.Background(), bare)
|
||||
if got, _ = delivery.DecodeHaveNode(l.kv["n1.node"].value); got.Record.Contract != NodeKeyContract {
|
||||
t.Errorf("a report without a health statement dropped the contract: %+v", got.Record)
|
||||
}
|
||||
}
|
||||
|
||||
// 1132/AC7: a write the bucket refuses is said in the log; the report went out, and the next apply writes again.
|
||||
|
||||
@@ -743,4 +743,9 @@ type ModuleFact struct {
|
||||
Phase string
|
||||
Failed bool
|
||||
Detail string
|
||||
// Judgement and Waits are, in phase waiting-yes, the judgement of the move and the composition waiting (ADR 0547 §3).
|
||||
Judgement string
|
||||
Waits string
|
||||
// Sealed is the values the composition it names carries sealed to this node's key (delivery.Have.Sealed).
|
||||
Sealed []string
|
||||
}
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"sort"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The node's record (novox/hq ADR 0555 §3, issue 1126): what the machine observes of itself, written in its own have key
|
||||
// `<node>.node` (delivery.HaveNode.Record) with its facts, so every node and the parts that compose read it from the bus
|
||||
// and hear its change as the key's. Each value is the one the controller took from this node's enrolment and reports
|
||||
// until now (its profile's present capabilities and networks, its outward links, its health statement's contract, and
|
||||
// the public halves of the keys it enrolled with), read here from the same sources, never from the controller.
|
||||
|
||||
// RecordKeys are the public halves the node holds and the certificate the mesh issued for its internal name, as read
|
||||
// from its state directory at the time of a pass: the certificate arrives after enrolment, a rekey changes the tunnel key.
|
||||
type RecordKeys struct {
|
||||
TunnelKey string
|
||||
SealingKey string
|
||||
ServingKey string
|
||||
Certificate string
|
||||
// Authority is the mesh's own certificate as this machine last applied it (public).
|
||||
Authority string
|
||||
}
|
||||
|
||||
// recordOf is the node's record from one facts report and its keys; nil for a report that carries no profile (one that
|
||||
// says nothing of the machine, such as a rekey), so the record last written stands.
|
||||
func recordOf(node string, r Report, keys RecordKeys) *delivery.NodeRecord {
|
||||
if r.Profile == nil {
|
||||
return nil
|
||||
}
|
||||
var p struct {
|
||||
Capabilities []struct {
|
||||
Name string `json:"name"`
|
||||
Present bool `json:"present"`
|
||||
} `json:"capabilities"`
|
||||
Machine *struct {
|
||||
Networks []catalogue.AttachedNetwork `json:"networks"`
|
||||
} `json:"machine"`
|
||||
}
|
||||
raw, err := json.Marshal(r.Profile)
|
||||
if err != nil || json.Unmarshal(raw, &p) != nil {
|
||||
return nil
|
||||
}
|
||||
rec := &delivery.NodeRecord{V: delivery.Version, Node: node, TunnelKey: keys.TunnelKey, SealingKey: keys.SealingKey,
|
||||
ServingKey: keys.ServingKey, Certificate: keys.Certificate, Authority: keys.Authority}
|
||||
for _, c := range p.Capabilities {
|
||||
if c.Present && c.Name != "" {
|
||||
rec.Capabilities = append(rec.Capabilities, c.Name)
|
||||
}
|
||||
}
|
||||
sort.Strings(rec.Capabilities)
|
||||
if p.Machine != nil {
|
||||
rec.NetworksSaid = true
|
||||
rec.Networks = p.Machine.Networks
|
||||
}
|
||||
if len(r.Outward) > 0 {
|
||||
rec.OutwardLinks = append([]string(nil), r.Outward...)
|
||||
}
|
||||
if r.Health != nil {
|
||||
rec.Contract = r.Health.Contract
|
||||
}
|
||||
return rec
|
||||
}
|
||||
@@ -47,8 +47,11 @@ type Session struct {
|
||||
// MoveSecret moves one value sealed to this machine into the vault, asked by the controller during the move of
|
||||
// the secrets (design 57's V3, issue 676). Nil answers every such ask with a refusal.
|
||||
MoveSecret MoveSecretFunc
|
||||
Say Announce
|
||||
Timeout time.Duration
|
||||
// Keys reads the node's public keys and its certificate for its record (ADR 0555 §3), at each facts report; nil writes
|
||||
// no record.
|
||||
Keys func() RecordKeys
|
||||
Say Announce
|
||||
Timeout time.Duration
|
||||
|
||||
once sync.Once
|
||||
|
||||
|
||||
@@ -0,0 +1,745 @@
|
||||
// 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
|
||||
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{}, 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)
|
||||
}
|
||||
|
||||
func (c *Composer) inputs() []input {
|
||||
node := c.env.Node
|
||||
return []input{
|
||||
{delivery.AssignerWantsBucket, "want." + node + ".>", (*Composer).takeWant},
|
||||
{delivery.PlacementBucket, ">", (*Composer).takePlacement},
|
||||
{delivery.HaveBucket, "*." + delivery.HaveNodeWord, (*Composer).takeRecord},
|
||||
{delivery.ModulesBucket, ">", (*Composer).takeModule},
|
||||
{ResolvedBucket, node + ".>", (*Composer).takeValue},
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
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
|
||||
}
|
||||
|
||||
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]
|
||||
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}, 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()
|
||||
modules := make([]string, 0, len(c.wants))
|
||||
for m := range c.wants {
|
||||
if m != agent.NodeModule {
|
||||
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.
|
||||
func (c *Composer) Policy(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(context.Background(), 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
|
||||
}
|
||||
@@ -0,0 +1,316 @@
|
||||
package selfcompose
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
)
|
||||
|
||||
const lemurs = `{"module":"lemurs","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"lemurs.service","state":"running","boot":"enabled"}]}`
|
||||
|
||||
// fakeBus is the buckets the composer watches, each a channel a test writes updates to, and the keys Get reads.
|
||||
type fakeBus struct {
|
||||
mu sync.Mutex
|
||||
chans map[string]chan agent.Update
|
||||
kv map[string][]byte // bucket/key
|
||||
said []string
|
||||
}
|
||||
|
||||
func newFakeBus() *fakeBus {
|
||||
return &fakeBus{chans: map[string]chan agent.Update{}, kv: map[string][]byte{}}
|
||||
}
|
||||
|
||||
func (b *fakeBus) ch(bucket string) chan agent.Update {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.chans[bucket] == nil {
|
||||
b.chans[bucket] = make(chan agent.Update, 16)
|
||||
}
|
||||
return b.chans[bucket]
|
||||
}
|
||||
|
||||
func (b *fakeBus) env() Env {
|
||||
return Env{Node: "n1",
|
||||
Watch: func(_ context.Context, bucket, _ string) (<-chan agent.Update, error) { return b.ch(bucket), nil },
|
||||
Get: func(_ context.Context, bucket, key string) ([]byte, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.kv[bucket+"/"+key], nil
|
||||
},
|
||||
Say: func(s string) { b.mu.Lock(); b.said = append(b.said, s); b.mu.Unlock() }}
|
||||
}
|
||||
|
||||
func (b *fakeBus) put(t *testing.T, bucket, key string, value []byte) {
|
||||
t.Helper()
|
||||
b.ch(bucket) <- agent.Update{Key: key, Value: value}
|
||||
}
|
||||
|
||||
func next(t *testing.T, out <-chan agent.Update) agent.Update {
|
||||
t.Helper()
|
||||
select {
|
||||
case u := <-out:
|
||||
return u
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("no update from the composer")
|
||||
return agent.Update{}
|
||||
}
|
||||
}
|
||||
|
||||
func quiet(t *testing.T, out <-chan agent.Update) {
|
||||
t.Helper()
|
||||
select {
|
||||
case u := <-out:
|
||||
t.Fatalf("the composer handed %s, and nothing changed", u.Key)
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
|
||||
// Issue 1126 (ADR 0547 §2): the node composes its wanted modules and its own composition from its inputs on the bus once
|
||||
// every input was read, hands each, then Caught; an input that changes nothing composed hands nothing; a module no want
|
||||
// names any more is taken back; a node that cannot compose says so once and hands nothing.
|
||||
func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
b := newFakeBus()
|
||||
fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
|
||||
digest := delivery.ManifestDigest([]byte(lemurs))
|
||||
man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: lemurs})
|
||||
key, _ := delivery.ManifestsKey(digest)
|
||||
b.kv[delivery.ManifestsBucket+"/"+key] = man
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
env := b.env()
|
||||
// The vault's own: keeper's password sealed to n1's key, sealed again (other bytes) at every ask.
|
||||
var asks, generation int
|
||||
sealedAt := map[string]int{} // each box handed, by the generation it opens to
|
||||
env.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
asks++
|
||||
sealedAt[fmt.Sprintf(`{"ask":%d}`, asks)] = generation + 1
|
||||
return json.RawMessage(fmt.Sprintf(`{"node":"n1","key":"sk","own":[{"module":"keeper","name":"password","origin":"made",`+
|
||||
`"generation":%d,"sealed":{"ask":%d}}]}`, generation+1, asks)), nil
|
||||
}
|
||||
secretChanged := make(chan []byte, 1)
|
||||
env.Hear = func(subject string) (<-chan []byte, func(), error) {
|
||||
if subject != VaultSecretChanged {
|
||||
t.Errorf("heard %s", subject)
|
||||
}
|
||||
return secretChanged, func() {}, nil
|
||||
}
|
||||
c := New(env)
|
||||
out, err := c.Watch(ctx, "n1.>")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []delivery.PlacedModule{{Module: "lemurs"}}})
|
||||
have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
||||
want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
|
||||
nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
|
||||
b.put(t, delivery.PlacementBucket, "placement.n1", place)
|
||||
b.put(t, delivery.HaveBucket, "n1.node", have)
|
||||
b.put(t, delivery.ModulesBucket, "lemurs", mod)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
|
||||
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
|
||||
b.ch(bucket) <- agent.Update{Caught: true}
|
||||
}
|
||||
got := map[string]bool{}
|
||||
for {
|
||||
u := next(t, out)
|
||||
if u.Caught {
|
||||
break
|
||||
}
|
||||
c, err := delivery.DecodeComposed(u.Value)
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", u.Key, err)
|
||||
}
|
||||
if c.Node != "n1" {
|
||||
t.Errorf("%s composed for %q", u.Key, c.Node)
|
||||
}
|
||||
got[u.Key] = true
|
||||
}
|
||||
lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp)
|
||||
nodeKey, _ := delivery.ComposedNodeKey("n1")
|
||||
if !got[lemursKey] || !got[nodeKey] || len(got) != 2 {
|
||||
b.mu.Lock()
|
||||
t.Logf("said: %v", b.said)
|
||||
b.mu.Unlock()
|
||||
t.Fatalf("handed %v, want %s and %s", got, lemursKey, nodeKey)
|
||||
}
|
||||
|
||||
// A config value of another module, and a value changing nothing composed, hand nothing.
|
||||
b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1}`))
|
||||
quiet(t, out)
|
||||
|
||||
// A want removed takes its composition back.
|
||||
removed, _ := delivery.EncodeWant(delivery.Want{Removed: true})
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", removed)
|
||||
if u := next(t, out); u.Key != lemursKey || !u.Deleted {
|
||||
t.Fatalf("a removed want handed %+v", u)
|
||||
}
|
||||
|
||||
// A want of a module whose manifest cannot be read: said once, nothing handed, what was handed stands.
|
||||
gone := delivery.SourcePrefix2 + strings.Repeat("b", 64)
|
||||
missing, _ := delivery.EncodeWant(delivery.Want{FP: gone, Self: true})
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing)
|
||||
quiet(t, out)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing)
|
||||
quiet(t, out)
|
||||
b.mu.Lock()
|
||||
n := 0
|
||||
for _, s := range b.said {
|
||||
if strings.Contains(s, "lemurs is wanted at") {
|
||||
n++
|
||||
}
|
||||
}
|
||||
b.mu.Unlock()
|
||||
if n != 1 {
|
||||
t.Errorf("the fault was said %d times, want once: %v", n, b.said)
|
||||
}
|
||||
|
||||
// An own secret comes sealed from the vault into the composition; sealed again at the same generation it is the value
|
||||
// first had, so nothing is handed again; a new generation is handed.
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
|
||||
if u := next(t, out); u.Key != lemursKey {
|
||||
t.Fatalf("lemurs wanted back handed %s", u.Key)
|
||||
}
|
||||
keeper := `{"module":"keeper","version":"1","own-secrets":{"password":"/var/lib/mesh/keeper/password"},` +
|
||||
`"resources":[{"id":"state","type":"directory","path":"/var/lib/mesh/keeper","mode":"0700"}]}`
|
||||
kdigest := delivery.ManifestDigest([]byte(keeper))
|
||||
kman, _ := delivery.EncodeManifest(delivery.Manifest{Module: "keeper", JSON: keeper})
|
||||
kkey, _ := delivery.ManifestsKey(kdigest)
|
||||
b.mu.Lock()
|
||||
b.kv[delivery.ManifestsBucket+"/"+kkey] = kman
|
||||
b.mu.Unlock()
|
||||
kmod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"})
|
||||
both, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
||||
Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "keeper"}}})
|
||||
b.put(t, delivery.ModulesBucket, "keeper", kmod)
|
||||
b.put(t, delivery.PlacementBucket, "placement.n1", both)
|
||||
b.put(t, delivery.AssignerWantsBucket, "want.n1.keeper", want)
|
||||
keeperKey, _ := delivery.ComposedKey("n1", "keeper", fp)
|
||||
var handed string
|
||||
for handed == "" {
|
||||
if u := next(t, out); u.Key == keeperKey {
|
||||
handed = string(u.Value)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(handed, `{\"ask\":`) {
|
||||
t.Fatalf("keeper's composition does not carry the sealed value: %s", handed)
|
||||
}
|
||||
before := handed
|
||||
b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":2}`))
|
||||
quiet(t, out)
|
||||
b.mu.Lock()
|
||||
generation++
|
||||
b.mu.Unlock()
|
||||
// The vault says the secret changed: the node composes again on that word, with no other input moving.
|
||||
secretChanged <- []byte(`{"name":"own/n1/keeper/password","generation":2}`)
|
||||
rotated := next(t, out)
|
||||
if rotated.Key != keeperKey {
|
||||
t.Fatalf("a new generation handed %s, want keeper again", rotated.Key)
|
||||
}
|
||||
|
||||
// Review of L2 (round 2): rotated, then the node-engine restarts. The agent has applied nothing yet, so the box the
|
||||
// node runs comes from keeper's have (Held); the vault seals the value again, other bytes, and the box the have holds
|
||||
// is kept, since it opens to the same value: the composition does not move. The box from before the rotation opens
|
||||
// to the old value, so the vault's is taken.
|
||||
boxOf := func(raw string) string {
|
||||
c, err := delivery.DecodeComposed([]byte(raw))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sealed := agent.SealedIn(c)
|
||||
if len(sealed) != 1 {
|
||||
t.Fatalf("keeper's composition carries %v", sealed)
|
||||
}
|
||||
return sealed[0]
|
||||
}
|
||||
for _, row := range []struct {
|
||||
name, held string
|
||||
kept bool
|
||||
}{{"the have holds the rotated value", boxOf(string(rotated.Value)), true}, {"the have holds the value before the rotation", boxOf(before), false}} {
|
||||
restarted := env
|
||||
restarted.Running = func() []delivery.Composed { return nil }
|
||||
restarted.Held = func(module string) []string {
|
||||
if module == "keeper" {
|
||||
return []string{row.held}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
restarted.Unseal = func(box string) ([]byte, error) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if g, ok := sealedAt[box]; ok {
|
||||
return []byte(fmt.Sprintf("password-%d", g)), nil
|
||||
}
|
||||
return nil, fmt.Errorf("not sealed to n1: %s", box)
|
||||
}
|
||||
again := New(restarted)
|
||||
again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}}
|
||||
rows, err := reader{c: again}.OwnSecrets(ctx, "n1")
|
||||
if err != nil || len(rows) != 1 {
|
||||
t.Fatalf("%s: own secrets %+v, %v", row.name, rows, err)
|
||||
}
|
||||
if kept := rows[0].Sealed == row.held; kept != row.kept {
|
||||
t.Errorf("%s: composed with %s (held %s), kept %v, want %v", row.name, rows[0].Sealed, row.held, kept, row.kept)
|
||||
}
|
||||
}
|
||||
|
||||
// The pair credentials asked (ADR 0550 §5): the keys go to the vault's own with their role, a consumer's comes back as
|
||||
// a credential held, a provider's as the grant's copy, each sealed to n1's key, the value first had kept.
|
||||
var sent map[string]any
|
||||
pairEnv := b.env()
|
||||
pairEnv.Own = func(_ context.Context, args map[string]any) (json.RawMessage, error) {
|
||||
sent = args
|
||||
return json.RawMessage(`{"node":"n1","key":"sk","credentials":[` +
|
||||
`{"role":"consumer","provision":"pg","consumer":"n1","consumer_module":"web","provider":"n2","name":"pair/n2/pg/n1/web","generation":1,"sealed":{"c":1}},` +
|
||||
`{"role":"provider","provision":"s3","consumer":"n3","consumer_module":"files","provider":"n1","name":"provision/n1/s3","generation":2,"sealed":{"p":1}}]}`), nil
|
||||
}
|
||||
// The node runs the consumer's credential already, sealed otherwise (the controller sealed it, or the vault at
|
||||
// another ask): that box opens to the same value, so it is kept, and the composition does not move.
|
||||
pairEnv.Running = func() []delivery.Composed {
|
||||
return []delivery.Composed{{Module: "web", Resources: []json.RawMessage{json.RawMessage(`{"id":"web.secret-pg","sealed":"{\"run\":1}"}`)}}}
|
||||
}
|
||||
opened := map[string]string{`{"c":1}`: "pg-pw", `{"run":1}`: "pg-pw", `{"p":1}`: "s3-pw"}
|
||||
pairEnv.Unseal = func(box string) ([]byte, error) {
|
||||
if v, ok := opened[box]; ok {
|
||||
return []byte(v), nil
|
||||
}
|
||||
return nil, fmt.Errorf("not sealed to n1: %s", box)
|
||||
}
|
||||
var refusals []string
|
||||
pairEnv.Refused = func(why string) { refusals = append(refusals, why) }
|
||||
pc := New(pairEnv)
|
||||
consumer := []compose.CredentialKey{{Provision: "pg", Consumer: "n1", ConsumerModule: "web", Provider: "n2"}}
|
||||
provider := []compose.CredentialKey{{Provision: "s3", Consumer: "n3", ConsumerModule: "files", Provider: "n1"}}
|
||||
creds, issued, err := reader{c: pc}.Credentials(ctx, "n1", consumer, provider)
|
||||
if err != nil || len(creds) != 1 || len(issued) != 1 {
|
||||
t.Fatalf("credentials %+v issued %+v, %v", creds, issued, err)
|
||||
}
|
||||
if creds[0].State != compose.CredentialHeld || creds[0].Provider != "n2" || creds[0].Sealed != `{"run":1}` ||
|
||||
issued[0].Consumer != "n3" || issued[0].ForProvider != `{"p":1}` {
|
||||
t.Errorf("credentials %+v issued %+v", creds, issued)
|
||||
}
|
||||
// A node that cannot compose tells the agent once, and once again when it composes (the agent's haves say it).
|
||||
pc.refused("a stand-in for web's password")
|
||||
pc.refused("a stand-in for web's password")
|
||||
pc.refused("")
|
||||
if len(refusals) != 2 || refusals[0] == "" || refusals[1] != "" {
|
||||
t.Errorf("the agent was told %q", refusals)
|
||||
}
|
||||
if asked, _ := json.Marshal(sent["credentials"]); !strings.Contains(string(asked), `"role":"consumer"`) || !strings.Contains(string(asked), `"role":"provider"`) {
|
||||
t.Errorf("the vault was asked %s", asked)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user