Review of L2: a node that cannot compose says so in each wanted module's have, failed, while it keeps what it runs (H2); the composer keeps the box the node runs when it opens to the vault's value, so neither the switch nor a restart moves a composition (H1); pin mesh-sdk 3968849, whose node asks the vault only for a module declaring a secret (H8's failure) (issue 1126)
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: unit: ..test: ✔ a transient empty list for four passes retires nothing; five stable passes retire, data kept (290.408814ms) | 58de6…
mesh/delivery closed: a newer head of the same pull request
mesh/check failed: unit: ..test: ✔ a transient empty list for four passes retires nothing; five stable passes retire, data kept (290.408814ms) | 58de6…
This commit is contained in:
@@ -497,7 +497,23 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
||||
// 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)
|
||||
if composer := selfComposer(linked, session, node, say); composer != nil {
|
||||
// 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),
|
||||
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
|
||||
@@ -788,7 +804,14 @@ func applyContext(ctx context.Context) (context.Context, context.CancelFunc) {
|
||||
|
||||
// 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.
|
||||
func selfComposer(ctx context.Context, session *link.Session, node string, say func(string)) *selfcompose.Composer {
|
||||
// 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)
|
||||
}
|
||||
|
||||
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())
|
||||
@@ -822,7 +845,8 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f
|
||||
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})
|
||||
Hear: session.Hear,
|
||||
Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused})
|
||||
}
|
||||
|
||||
// endWhenPlaced calls stop once this node's placement key appears in the indexer's placement (a value after the watch
|
||||
|
||||
@@ -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.20261026183837-9713c3023a18
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026190832-3968849b1b36
|
||||
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
|
||||
|
||||
@@ -18,6 +18,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026182519-53c2def1a520 h1:twiziR88
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026182519-53c2def1a520/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026183837-9713c3023a18 h1:HWNPfnT9YT9O6xHPLy2y6Q+LYT8XRjvLnZw9O61NETk=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026183837-9713c3023a18/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026190832-3968849b1b36 h1:aBkmFUq/zAVdR3PzfgjjC9AxSymNl3eVC7UEZv9tZ0Q=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026190832-3968849b1b36/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=
|
||||
|
||||
+43
-2
@@ -156,7 +156,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, and why (NotComposed).
|
||||
notComposed map[string]string
|
||||
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
|
||||
@@ -202,7 +204,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]string{},
|
||||
retry: make(chan []string, 8), tries: map[string]int{}, again: make(chan againAsk, 64),
|
||||
transient: make(chan struct{}, 1), stopped: make(chan struct{})}
|
||||
}
|
||||
@@ -303,6 +305,44 @@ func (a *Agent) RetryTransient() {
|
||||
}
|
||||
}
|
||||
|
||||
// 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. Called with "" once the node
|
||||
// composes again: each module so marked has its have back as what it runs, unless an apply wrote it since.
|
||||
func (a *Agent) NotComposed(ctx context.Context, why string) {
|
||||
a.mu.Lock()
|
||||
var write []Have
|
||||
if why != "" {
|
||||
for module, w := range a.wants {
|
||||
if !w.Present || w.Want.Removed || a.notComposed[module] == why {
|
||||
continue
|
||||
}
|
||||
a.notComposed[module] = why
|
||||
r := a.running[module]
|
||||
write = append(write, Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(delivery.PhaseFailed),
|
||||
Detail: "the node cannot compose it, and keeps running what it ran: " + why})
|
||||
}
|
||||
} else {
|
||||
for module := range a.notComposed {
|
||||
delete(a.notComposed, module)
|
||||
r, ok := a.running[module]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
phase := r.Phase
|
||||
if phase == "" {
|
||||
phase = delivery.PhaseRunning
|
||||
}
|
||||
write = append(write, Have{Module: module, FP: r.FP, Composed: r.Composed, Phase: string(phase)})
|
||||
}
|
||||
}
|
||||
a.mu.Unlock()
|
||||
sort.Slice(write, func(i, j int) bool { return write[i].Module < write[j].Module })
|
||||
for _, h := range write {
|
||||
a.env.Haves.Write(ctx, h)
|
||||
}
|
||||
}
|
||||
|
||||
// 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 {
|
||||
@@ -858,6 +898,7 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
|
||||
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()
|
||||
|
||||
@@ -1000,6 +1000,16 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
|
||||
// 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): web's have says so, failed, naming why, while it runs what it ran;
|
||||
// composed again, the have is what it runs.
|
||||
b.ag.NotComposed(context.Background(), "a stand-in for web's password")
|
||||
if h := b.haves.last("web"); h.Phase != string(delivery.PhaseFailed) || h.Composed != dig || !strings.Contains(h.Detail, "web's password") {
|
||||
t.Errorf("web's have while the node cannot compose: %+v", h)
|
||||
}
|
||||
b.ag.NotComposed(context.Background(), "")
|
||||
if h := b.haves.last("web"); h.Phase != string(delivery.PhaseRunning) || h.Composed != dig {
|
||||
t.Errorf("web's have once the node composes again: %+v", 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)
|
||||
|
||||
@@ -11,6 +11,7 @@
|
||||
package selfcompose
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
@@ -45,6 +46,15 @@ type Env struct {
|
||||
// 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)
|
||||
}
|
||||
|
||||
// VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value.
|
||||
@@ -66,6 +76,7 @@ type Composer struct {
|
||||
// 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
|
||||
}
|
||||
@@ -351,7 +362,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
|
||||
for _, module := range behind {
|
||||
digest, err := c.builtManifest(ctx, module, fps[module])
|
||||
if err != nil {
|
||||
c.sayOnce(fmt.Sprintf("%s is wanted at %s, whose manifest cannot be read, so the node does not compose: %v", module, fps[module], err))
|
||||
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
|
||||
@@ -359,9 +372,11 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
|
||||
|
||||
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()
|
||||
@@ -405,6 +420,54 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
|
||||
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: 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.Running == nil || c.env.Unseal == nil {
|
||||
return box
|
||||
}
|
||||
value, err := c.env.Unseal(box)
|
||||
if err != nil {
|
||||
return box
|
||||
}
|
||||
for _, comp := range c.env.Running() {
|
||||
for _, raw := range comp.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 held string
|
||||
if json.Unmarshal(f, &held) != nil {
|
||||
held = string(f) // a vault box is carried as its JSON
|
||||
}
|
||||
if held == "" || held == box {
|
||||
continue
|
||||
}
|
||||
if v, err := c.env.Unseal(held); err == nil && bytes.Equal(v, value) {
|
||||
return held
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return box
|
||||
}
|
||||
|
||||
func (c *Composer) sayOnce(line string) {
|
||||
c.mu.Lock()
|
||||
if c.said == line {
|
||||
@@ -565,14 +628,25 @@ func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecre
|
||||
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 _, o := range a.Own {
|
||||
id := o.Module + "/" + o.Name
|
||||
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: string(o.Sealed)}
|
||||
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,
|
||||
@@ -617,16 +691,28 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv
|
||||
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 _, c := range a.Credentials {
|
||||
// The value first had is kept while its generation stands, so sealing it again moves no composition.
|
||||
id := c.Role + "|" + c.Name + "|" + c.Provision + "|" + c.Consumer + "|" + c.ConsumerModule + "|" + c.Local
|
||||
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: string(c.Sealed)}
|
||||
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
|
||||
@@ -641,3 +727,11 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
@@ -228,6 +228,20 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
`{"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"}}
|
||||
@@ -235,10 +249,17 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
||||
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 != `{"c":1}` ||
|
||||
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