From 58de6aa10ac9f78ad5859b926bceee8091549784 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 20:09:40 +0100 Subject: [PATCH] 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) --- cmd/mesh-host/nodeagent.go | 30 ++++++- go.mod | 2 +- go.sum | 2 + internal/agent/agent.go | 45 +++++++++- internal/agent/agent_test.go | 10 +++ internal/selfcompose/selfcompose.go | 110 +++++++++++++++++++++-- internal/selfcompose/selfcompose_test.go | 23 ++++- 7 files changed, 207 insertions(+), 15 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 09aca00..d48e866 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -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.` 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 diff --git a/go.mod b/go.mod index 74c7e2f..4bf1048 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 5f7cea5..be24169 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/internal/agent/agent.go b/internal/agent/agent.go index ea2f8df..5f26b6b 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -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() diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index c111a31..6c23c90 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -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) diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 133ceb0..7cb5e8a 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -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 +} diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index c792cb2..c55a972 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -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) }