diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 8296b75..e57b079 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -883,7 +883,9 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com 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, + // And a moved module's templates archive, in the artifact store, as every archive is fetched (issue 1126). + Fetch: apply.Fetch, Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held} } diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 06bc6c8..2d5a7d9 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -29,9 +30,14 @@ import ( const itNode = "it1" type itBed struct { - t *testing.T - conn *nats.Conn - js nats.JetStreamContext + t *testing.T + conn *nats.Conn + js nats.JetStreamContext + // nodeConn and nodeJS are the node's own, under its real grant (buscompose.PermissionsFor, KindNode): the agent runs + // on them, so a watch or a read the grant does not allow fails here as it fails on the bus (issue 1126, 2026-10-27 + // 09:21: a watch of every have key stopped every node agent). conn is the test's, unrestricted, for setting up. + nodeConn *nats.Conn + nodeJS nats.JetStreamContext wants nats.KeyValue composed nats.KeyValue haves nats.KeyValue @@ -41,12 +47,27 @@ type itBed struct { controll ed25519.PrivateKey operator ed25519.PrivateKey said []string + // refusals are the bus's asynchronous errors on the node's connection: each permissions violation it met. + refusals []string mu sync.Mutex } -func aServer(t *testing.T) *nats.Conn { +// aServer is an embedded bus with JetStream and two users: the test's, unrestricted, and the node's, granted exactly what +// buscompose.PermissionsFor gives a node (KindNode); it answers a connection of each. +func aServer(t *testing.T, principal buscompose.Principal, refused func(error)) (admin, node *nats.Conn) { t.Helper() - srv, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) + perms, err := buscompose.PermissionsFor(principal) + if err != nil { + t.Fatal(err) + } + granted := &server.Permissions{Publish: &server.SubjectPermission{Allow: perms.Publish, Deny: perms.PublishDeny}, + Subscribe: &server.SubjectPermission{Allow: perms.Subscribe}} + if perms.AllowResponses { + granted.Response = &server.ResponsePermission{MaxMsgs: 1, Expires: buscompose.ResponseTTL} + } + nodeUser := buscompose.Principal{Kind: buscompose.KindNode, Node: itNode}.Username() + srv, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true, + Users: []*server.User{{Username: "test", Password: "test"}, {Username: nodeUser, Password: "node", Permissions: granted}}}) if err != nil { t.Fatal(err) } @@ -55,15 +76,25 @@ func aServer(t *testing.T) *nats.Conn { t.Fatal("the embedded server did not start") } t.Cleanup(srv.Shutdown) - conn, err := nats.Connect(srv.ClientURL()) - if err != nil { + if admin, err = nats.Connect(srv.ClientURL(), nats.UserInfo("test", "test")); err != nil { t.Fatal(err) } - t.Cleanup(conn.Close) - return conn + t.Cleanup(admin.Close) + if node, err = nats.Connect(srv.ClientURL(), nats.UserInfo(nodeUser, "node"), nats.CustomInboxPrefix("_INBOX."+nodeUser), + nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused(err) })); err != nil { + t.Fatal(err) + } + t.Cleanup(node.Close) + return admin, node } func newItBed(t *testing.T) *itBed { + t.Helper() + return newItBedAs(t, buscompose.Principal{Kind: buscompose.KindNode, Node: itNode}) +} + +// newItBedAs is newItBed with the node's connection under the grant of principal, a node's (KindNode). +func newItBedAs(t *testing.T, principal buscompose.Principal) *itBed { t.Helper() // Not skipped off Arch: a package manager that answers is put on PATH, so the apply of files runs on any build node. bin := t.TempDir() @@ -75,11 +106,19 @@ func newItBed(t *testing.T) *itBed { builtFor = "arch" t.Cleanup(func() { builtFor = was }) - b := &itBed{t: t, conn: aServer(t), dir: t.TempDir()} + b := &itBed{t: t, dir: t.TempDir()} + b.conn, b.nodeConn = aServer(t, principal, func(err error) { + b.mu.Lock() + b.refusals = append(b.refusals, err.Error()) + b.mu.Unlock() + }) var err error if b.js, err = b.conn.JetStream(); err != nil { t.Fatal(err) } + if b.nodeJS, err = b.nodeConn.JetStream(); err != nil { + t.Fatal(err) + } composedBucket, _ := delivery.ComposedBucketFor(itNode) for _, c := range []struct { name string @@ -114,7 +153,7 @@ func (b *itBed) start(trustOperator bool) (stop func()) { } b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second, Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }} - b.session.Attach(context.Background(), link.OverNATS{Conn: b.conn, JS: b.js, Node: itNode}) + b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode}) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { @@ -134,6 +173,8 @@ func (b *itBed) say(line string) { func fpIt(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) } +func fpDigest(c string) string { return "sha256:" + strings.Repeat(c, 64) } + // compose writes a module's signed composition of one file into the node's bucket and returns its digest. func (b *itBed) compose(module, fp, content string, data bool, signer ed25519.PrivateKey) string { b.t.Helper() @@ -318,8 +359,65 @@ func TestNatsTheHandPathAppliesACompositionTheOperatorSigned(t *testing.T) { // 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. And a catalogue pull request's placement candidate equal to its placement is answered "same" // in the node's own key, and nothing is applied for it (ADR 0559 §4, subtask 10 of the milestone). +// +// And (issue 1126, ADR 0560) it reads, through its own session under the grant the grants part composes for it +// (buscompose.Users, so it carries ServedReads), the build of a placed module that declares one and the one config value +// a provider on another node serves from (a mail domain); under a grant without that served key the bus refuses the +// read and the node composes nothing. func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { - b := newItBed(t) + users, err := buscompose.Users(buscompose.Records{Nodes: []string{itNode, "it2"}, + Assigned: map[string][]buscompose.Declared{"it2": {{Module: "mailer", Served: []string{"domain"}}}}}) + if err != nil { + t.Fatal(err) + } + var principal buscompose.Principal + for _, u := range users { + if u.Kind == buscompose.KindNode && u.Node == itNode { + principal = u + } + } + if len(principal.ServedReads) != 1 { + t.Fatalf("the grants part's principal of %s reads %v of the served keys", itNode, principal.ServedReads) + } + t.Run("without the served key the bus refuses the read", func(t *testing.T) { + b := newItBed(t) + stop, _ := h8(t, b, false) + defer stop() + // 15 s, above link.historyWithin (10 s): the bus's refusal of a publish reaches only the connection's error + // handler (nats.go fails no request on it), so the refused read ends at its own bound, then the node says it. + within(t, 15*time.Second, "the bus did not refuse the served key's read, or the node did not say it", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + return strings.Contains(strings.Join(b.refusals, "\n"), "Permissions Violation for Publish to \"$JS.API.CONSUMER.CREATE.KV_"+ + selfcompose.ResolvedBucket) && strings.Contains(strings.Join(b.refusals, "\n"), ".it2.mailer.domain") && + strings.Contains(strings.Join(b.said, "\n"), "what mailer serves on it2 (domain)") + }) + if b.file("web") != "" { + t.Errorf("composed and applied under a grant without the served key: %q", b.file("web")) + } + }) + b := newItBedAs(t, principal) + stop, st := h8(t, b, true) + defer stop() + placement, resolved, candidates := st.placement, st.resolved, st.candidates + hkey, place, fp := st.hkey, st.place, st.fp + 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" }) + _ = placement + h8rest(t, b, resolved, candidates, hkey, place, fp) +} + +// h8state is what h8 set up that the rest of H8 writes to. +type h8state struct { + placement, resolved, candidates nats.KeyValue + hkey string + place []byte + fp string +} + +// h8 sets H8's bus up and starts the agent, then places the node. With checkBuild, the placed module's build key is +// first one that does not decode, which the node must read through its session (and say), then a done one. +func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { + t.Helper() kvOf := func(bucket string) nats.KeyValue { kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 5}) if err != nil { @@ -329,7 +427,7 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { } placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket) resolved := kvOf(selfcompose.ResolvedBucket) - kvOf(delivery.BuilderBuildsBucket) + builds := kvOf(delivery.BuilderBuildsBucket) candidates := kvOf(delivery.PlacementCandidatesBucket) // The controller's stream the node's facts go into, so the node's key carries the facts it says. if _, err := b.js.AddStream(&nats.StreamConfig{Name: "CONTROL", Subjects: []string{link.ReportSubject(itNode)}}); err != nil { @@ -337,7 +435,8 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { } 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"}]}` + `"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0644","content":"greeting=${config:greeting}\n"}],` + + `"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}` digest := delivery.ManifestDigest([]byte(man)) mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "web", JSON: man}) mkey, _ := delivery.ManifestsKey(digest) @@ -347,12 +446,32 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: itNode, Capabilities: []string{"service-manager"}}}) pkey, _ := delivery.PlacementKey(itNode) hkey, _ := delivery.HaveNodeKey(itNode) + // A provider on it2 serving from its own config value: a mail domain, which it1's composition settles. + mailer := `{"module":"mailer","version":"1","config":{"domain":{"type":"string","description":"the mail domain","required":true}},` + + `"provides":[{"name":"smtp","scope":"mesh"}],"serves":{"smtp":{"port":587,"domain":"${config:domain}"}}}` + mdigest := delivery.ManifestDigest([]byte(mailer)) + mlraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "mailer", JSON: mailer}) + mlkey, _ := delivery.ManifestsKey(mdigest) + mlmod, _ := delivery.EncodeModule(delivery.Module{FP2: fpIt("b"), Manifest: mdigest, Repository: "novox/mailer", Commit: strings.Repeat("2", 40)}) + place2, _ := delivery.EncodePlacement(delivery.Placement{Node: "it2", Modules: []delivery.PlacedModule{{Module: "mailer"}}}) + rec2, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "it2", Capabilities: []string{"service-manager"}}}) + pkey2, _ := delivery.PlacementKey("it2") + hkey2, _ := delivery.HaveNodeKey("it2") + buildKey, _ := delivery.BuilderBuildKey("web", fp) + done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: digest, + Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://web/files/blobs/" + fpDigest("f"), Digest: fpDigest("f")}}}) + firstBuild := done + if checkBuild { + firstBuild = []byte(`{"v":1,"state":"no such state"}`) + } 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"}`)}} { + {resolved, itNode + ".web.greeting", []byte(`{"value":"bonjour"}`)}, + {manifests, mlkey, mlraw}, {modules, "mailer", mlmod}, {b.haves, hkey2, rec2}, {placement, pkey2, place2}, + {resolved, "it2.mailer.domain", []byte(`{"value":"example.org"}`)}, {builds, buildKey, firstBuild}} { if _, err := w.kv.Put(w.key, w.value); err != nil { t.Fatal(err) } @@ -364,7 +483,6 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { } 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) @@ -374,7 +492,30 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { 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" }) + if checkBuild { + // The build of the placed module is read through the node's session: one that does not decode is said, and nothing + // is applied; a done one, then a config event, and the node composes. + within(t, 5*time.Second, "the node did not read web's build key through its session", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + return strings.Contains(strings.Join(b.said, "\n"), "the builder's "+buildKey) + }) + if b.file("web") != "" { + t.Fatal("composed with a build key that does not decode") + } + if _, err := builds.Put(buildKey, done); err != nil { + t.Fatal(err) + } + if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"bonjour"}`)); err != nil { + t.Fatal(err) + } + } + return stop, h8state{placement: placement, resolved: resolved, candidates: candidates, hkey: hkey, place: place, fp: fp} +} + +// h8rest is H8 once the node composes its own: the have, a config revision, and the candidate answered "same". +func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey string, place []byte, fp string) { + t.Helper() 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 diff --git a/go.mod b/go.mod index c5812c9..ed71e1e 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.20261027075026-05b678310366 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027201213-7a8e9f970648 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 ae78f0e..07b9c16 100644 --- a/go.sum +++ b/go.sum @@ -12,6 +12,12 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERg git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027075026-05b678310366 h1:B7oQ82ar2McqZ9OrbJfyvG+50Ol6fBUoK3eYE2uTPvo= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027075026-05b678310366/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027195758-cebc6519d5b0 h1:9B7B38e3RtjbtX1uGP9dRgU9VY1/J2yAfnb5Ajoq4/s= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027195758-cebc6519d5b0/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027200317-538ffbaa085b h1:LuBK5dtBgm978krhj41N2MKA07Yv4zetRPOdHRTB6L8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027200317-538ffbaa085b/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027201213-7a8e9f970648 h1:XoMsgg9kSgjQWa86NZM8h8OpxJ2kovxcMS2C0i6WSVU= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027201213-7a8e9f970648/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/apply/archive.go b/internal/apply/archive.go index 13c9bd3..fef2f12 100644 --- a/internal/apply/archive.go +++ b/internal/apply/archive.go @@ -290,3 +290,7 @@ func unpack(body []byte, into string) (int, error) { } } } + +// Fetch is an archive at source as the host fetches every archive it unpacks, at most maxArchive bytes: what the node's +// own composer reads a moved module's templates archive with (issue 1126). Its digest is the caller's to check. +func Fetch(ctx context.Context, source string) ([]byte, error) { return fetch(ctx, source) } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 24f190f..a39ccf0 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -13,6 +13,8 @@ package selfcompose import ( "bytes" "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "fmt" "sort" @@ -58,6 +60,9 @@ type Env struct { // Held is the sealed values a module's have says it runs (delivery.Have.Sealed), kept by the agent on the bus: what // runs after a restart, before the agent applied anything, and after the switch once the kept declaration is old. Held func(module string) []string + // Fetch reads an archive at a URL in the artifact store: a moved module's templates archive, whose texts its built + // manifest carries (issue 1126). Read once per digest. Nil reads none, and a moved module does not compose. + Fetch func(ctx context.Context, url string) ([]byte, error) } // VaultSecretChanged is the vault's event that a secret changed: a name and its generation, never a value. @@ -77,6 +82,7 @@ type Composer struct { modules map[string]delivery.Module values map[string]map[string]any // by module, this node's manifests map[string][]byte // by digest, read once + archives map[string][]byte // templates archives by digest, read once handed map[string]string // composed key to the digest handed // sealedOwn is each own secret's sealed value as first handed, by module/name and generation: a value sealed again // is other bytes, so the one first had is kept while the generation stands, and the composition stays the same. @@ -94,7 +100,7 @@ func New(env Env) *Composer { } return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{}, - manifests: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}} + manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}} } // ownSealed is one own secret's sealed value at a generation and the key it is sealed to. @@ -117,8 +123,11 @@ func (c *Composer) inputs() []input { return []input{ {delivery.AssignerWantsBucket, "want." + node + ".>", (*Composer).takeWant, nil}, {delivery.PlacementBucket, ">", (*Composer).takePlacement, nil}, - // Every node's record, and this node's haves: what it runs inside its own composition. - {delivery.HaveBucket, ">", nil, (*Composer).takeHave}, + // Every node's record, and this node's own haves (what it runs inside its own composition): two watches, each + // within the node's grant (buscompose: readGrants of the have bucket at `*.node`, NodeKeysGrants at `.>`). + // A watch of every key is refused by the bus (issue 1126, 2026-10-27 09:21: the fold's `>` stopped every agent). + {delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, (*Composer).takeRecordChanged}, + {delivery.HaveBucket, node + ".>", nil, (*Composer).takeModuleHave}, {delivery.ModulesBucket, ">", (*Composer).takeModule, nil}, {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, } @@ -310,18 +319,14 @@ func (c *Composer) takeRecord(u agent.Update) { c.records[node] = *h.Record } -// takeHave takes a have key: a node's record (always composed again), or one of this node's modules' haves, which -// changes what is composed only when whether it runs inside the node's own composition changes. -func (c *Composer) takeHave(u agent.Update) bool { - if isNodeKey(u.Key) { - c.takeRecord(u) - return true - } - return c.takeModuleHave(u) +// takeRecordChanged takes a node's record, which is always composed again. +func (c *Composer) takeRecordChanged(u agent.Update) bool { + c.takeRecord(u) + return true } // takeModuleHave takes one of this node's modules' haves, and says whether it changed which it runs inside its own -// composition: one whose have names no fingerprint of the new form and is not removed (nats on novox: a build-... +// composition: one whose have names no fingerprint of the new form, is not removed, and runs something (nats on novox: a build-... // id; mesh-wireguard: none). Read from the haves, which outlast a restart, never from what the agent has applied // this run: a fresh agent composes before it applies anything (review of the fold, round 1). func (c *Composer) takeModuleHave(u agent.Update) bool { @@ -332,7 +337,11 @@ func (c *Composer) takeModuleHave(u agent.Update) bool { in := false if !u.Deleted { if h, err := delivery.DecodeHave(u.Value); err == nil { - in = h.Phase != delivery.PhaseRemoved && !strings.HasPrefix(h.FP, delivery.SourcePrefix2) + // A have that names nothing it runs, failed (the node could not compose a module it never ran: NotComposed), is + // no module run inside the node's own composition: taken as one, it was composed from its manifest past its + // build (issue 1126, review round 1 of the builds). + runsNothing := h.Phase == delivery.PhaseFailed && h.FP == "" && h.Composed == "" + in = h.Phase != delivery.PhaseRemoved && !runsNothing && !strings.HasPrefix(h.FP, delivery.SourcePrefix2) } } c.mu.Lock() @@ -422,19 +431,12 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { inOwn := c.inOwnOf(c.placements[c.env.Node]) r := reader{c: c} c.mu.Unlock() - sort.Strings(behind) + // Composed from the build its want names (frombus.Scope.KeptAt, issue 1126). 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 + kept[module] = fps[module] } - compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, frombus.Scope{Node: c.env.Node, Kept: kept, InOwn: inOwn}, fps) + compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn}, fps) if err != nil { c.refused(fmt.Sprintf("%v", err)) c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err)) @@ -549,30 +551,6 @@ func (c *Composer) sayOnce(line string) { } } -// builtManifest is the digest of the manifest a module's build at a fingerprint was made from (the builder's build key): -// what a node composes a module from when its want names another build than the trunk's. -func (c *Composer) builtManifest(ctx context.Context, module, fp string) (string, error) { - key, err := delivery.BuilderBuildKey(module, fp) - if err != nil { - return "", err - } - raw, err := c.env.Get(ctx, delivery.BuilderBuildsBucket, key) - if err != nil { - return "", err - } - if raw == nil { - return "", fmt.Errorf("the builder holds no build %s", key) - } - b, err := delivery.DecodeBuilderBuild(raw) - if err != nil { - return "", err - } - if b.Manifest == "" { - return "", fmt.Errorf("the build %s names no manifest", key) - } - return b.Manifest, nil -} - // Policy is a module's upgrade policy as its manifest at the indexer's trunk derives it (catalogue.DerivedUpgrade), for // the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read. The manifest is // read within ctx, the agent's (review of L2, round 1, finding 5), so a stopping agent stops the read. @@ -656,6 +634,71 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { return []byte(m.JSON), nil } +func (r reader) Build(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error) { + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return nil, err + } + raw, err := r.c.env.Get(ctx, delivery.BuilderBuildsBucket, key) + if err != nil || raw == nil { + return nil, err + } + b, err := delivery.DecodeBuilderBuild(raw) + if err != nil { + return nil, fmt.Errorf("the builder's %s: %w", key, err) + } + return &b, nil +} + +// Archive reads a templates archive once per digest (an archive is the whole of its digest), checked against it. +func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error) { + r.c.mu.Lock() + body, ok := r.c.archives[digest] + r.c.mu.Unlock() + if ok { + return body, nil + } + if r.c.env.Fetch == nil { + return nil, fmt.Errorf("this node reads no archive (%s)", url) + } + body, err := r.c.env.Fetch(ctx, url) + if err != nil { + return nil, err + } + sum := sha256.Sum256(body) + if got := "sha256:" + hex.EncodeToString(sum[:]); got != digest { + return nil, fmt.Errorf("%s was recorded as %s and what arrived is %s", url, digest, got) + } + r.c.mu.Lock() + r.c.archives[digest] = body + r.c.mu.Unlock() + return body, nil +} + +// Served reads, of another node's module, the keys named of mesh-config's resolved values, one key each: the node is +// granted those keys alone (buscompose Principal.ServedReads, issue 1126). +func (r reader) Served(ctx context.Context, node, module string, keys []string) (map[string]any, error) { + out := map[string]any{} + for _, k := range keys { + raw, err := r.c.env.Get(ctx, ResolvedBucket, node+"."+module+"."+k) + if err != nil { + return nil, err + } + var v struct { + Value json.RawMessage `json:"value"` + } + var value any + if raw == nil || json.Unmarshal(raw, &v) != nil || len(v.Value) == 0 || json.Unmarshal(v.Value, &value) != nil { + continue + } + out[k] = value + } + if len(out) == 0 { + return nil, nil + } + return out, nil +} + func (r reader) Values(_ context.Context, node, module string) (map[string]any, error) { if node != r.c.env.Node { return nil, nil // a node reads its own slice alone (ADR 0547 §2) diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 3175f53..afe26b4 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -2,13 +2,17 @@ package selfcompose import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "fmt" + "slices" "strings" "sync" "testing" "time" + "git.novox.be/novox/mesh-sdk/go/buscompose" "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" @@ -18,41 +22,95 @@ import ( 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. +// fakeBus is the buckets the composer watches, each watch a channel that gets the updates a test writes to its bucket +// whose key its pattern matches, as the bus's own watches do, and the keys Get reads. It keeps every watch and get +// made, so a test checks each against the node's grant. type fakeBus struct { - mu sync.Mutex - chans map[string]chan agent.Update - kv map[string][]byte // bucket/key - said []string + mu sync.Mutex + watches []patternWatch + kv map[string][]byte // bucket/key + said []string + reads [][2]string // bucket, pattern or key: every watch and get } 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] + return &fakeBus{kv: map[string][]byte{}} } func (b *fakeBus) env() Env { return Env{Node: "n1", - Watch: func(_ context.Context, bucket, _ string) (<-chan agent.Update, error) { return b.ch(bucket), nil }, + Watch: func(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) { + b.mu.Lock() + defer b.mu.Unlock() + w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)} + b.watches = append(b.watches, w) + b.reads = append(b.reads, [2]string{bucket, pattern}) + return w.ch, nil + }, Get: func(_ context.Context, bucket, key string) ([]byte, error) { b.mu.Lock() defer b.mu.Unlock() + b.reads = append(b.reads, [2]string{bucket, key}) return b.kv[bucket+"/"+key], nil }, Say: func(s string) { b.mu.Lock(); b.said = append(b.said, s); b.mu.Unlock() }} } +// send gives u to every watch of bucket whose pattern matches its key; Caught to every watch of bucket. +func (b *fakeBus) send(bucket string, u agent.Update) { + b.mu.Lock() + ws := append([]patternWatch(nil), b.watches...) + b.mu.Unlock() + for _, w := range ws { + if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) { + w.ch <- u + } + } +} + func (b *fakeBus) put(t *testing.T, bucket, key string, value []byte) { t.Helper() - b.ch(bucket) <- agent.Update{Key: key, Value: value} + b.send(bucket, agent.Update{Key: key, Value: value}) +} + +// grantAllows says whether the node's grant (buscompose.PermissionsFor, KindNode) lets it watch or read pattern of +// bucket: the consumer a watch or a get makes (`$JS.API.CONSUMER.CREATE.KV_..$KV..`) +// and the stream's information, matched as the bus matches a permission against a subject. +func grantAllows(t *testing.T, node, bucket, pattern string) bool { + t.Helper() + perms, err := buscompose.PermissionsFor(buscompose.Principal{Kind: buscompose.KindNode, Node: node}) + if err != nil { + t.Fatal(err) + } + allowed := func(subject string) bool { + for _, deny := range perms.PublishDeny { + if subjectMatches(deny, subject) { + return false + } + } + for _, p := range perms.Publish { + if subjectMatches(p, subject) { + return true + } + } + return false + } + stream := "KV_" + bucket + return allowed("$JS.API.STREAM.INFO."+stream) && allowed("$JS.API.CONSUMER.CREATE."+stream+".c1.$KV."+bucket+"."+pattern) +} + +// subjectMatches is the bus's match of a permission against a subject: `*` any one token, `>` one or more to the end. +func subjectMatches(permission, subject string) bool { + p, s := strings.Split(permission, "."), strings.Split(subject, ".") + for i, tok := range p { + if tok == ">" { + return len(s) > i + } + if i >= len(s) || (tok != "*" && tok != s[i]) { + return false + } + } + return len(p) == len(s) } func next(t *testing.T, out <-chan agent.Update) agent.Update { @@ -135,7 +193,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { 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} + b.send(bucket, agent.Update{Caught: true}) } got := map[string]bool{} var own delivery.Composed @@ -163,6 +221,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { if !strings.Contains(strings.Join(ownText, " "), `"id":"oldie.service"`) { t.Errorf("the node's own composition, composed before anything was applied, lacks oldie: %v", ownText) } + // Issue 1126, 2026-10-27 09:21: every watch and get the composer makes is one the node's grant allows; a watch of + // every have key (`>`) was refused by the bus and stopped every node agent. + b.mu.Lock() + for _, r := range b.reads { + if !grantAllows(t, "n1", r[0], r[1]) { + t.Errorf("the composer reads %s of %s, which the node's grant does not allow", r[1], r[0]) + } + } + b.mu.Unlock() lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp) nodeKey, _ := delivery.ComposedNodeKey("n1") if !got[lemursKey] || !got[nodeKey] || len(got) != 2 { @@ -424,6 +491,21 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 4 }) + // Every watch and get the shadow makes is one the node's grant allows (issue 1126, 2026-10-27 09:21). + sb.mu.Lock() + for _, w := range sb.watches { + if !grantAllows(t, "n1", w.bucket, w.pattern) { + t.Errorf("the shadow watches %s of %s, which the node's grant does not allow", w.pattern, w.bucket) + } + } + sb.mu.Unlock() + readOnly.mu.Lock() + for _, r := range readOnly.reads { + if !grantAllows(t, "n1", r[0], r[1]) { + t.Errorf("the shadow reads %s of %s, which the node's grant does not allow", r[1], r[0]) + } + } + readOnly.mu.Unlock() for group, want := range map[string]func(delivery.CandidateAnswer) bool{ "same": func(a delivery.CandidateAnswer) bool { return a.Verdict == delivery.CandidateSame && a.Head == "h-same" && a.Revision == 3 && len(a.Same) == 2 @@ -477,6 +559,54 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { if fmt.Sprint(gotFps) != fmt.Sprint(map[string]string{"lemurs": fp, "keeper": other}) { t.Errorf("shadow builds %v; want lemurs at what runs, keeper at the trunk's, nats at none (folded)", gotFps) } + + // Issue 1126, the seed's refusals: what a node composes from its builds and reads of a provider elsewhere. A build + // is the builder's key; a templates archive is fetched once per digest and refused when it is not that digest; another + // node's served key is read alone, as the node's grant allows it (buscompose Principal.ServedReads). + built, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: digest}) + buildKey, _ := delivery.BuilderBuildKey("lemurs", fp) + b.mu.Lock() + b.kv[delivery.BuilderBuildsBucket+"/"+buildKey] = built + b.kv[ResolvedBucket+"/n2.mailer.domain"] = []byte(`{"value":"example.org"}`) + b.reads = nil + b.mu.Unlock() + fetched := 0 + archive := []byte("an archive") + sum := sha256.Sum256(archive) + env.Fetch = func(context.Context, string) ([]byte, error) { fetched++; return archive, nil } + r := reader{c: New(env)} + if got, err := r.Build(ctx, "lemurs", fp); err != nil || got == nil || got.Manifest != digest { + t.Errorf("the build of lemurs: %+v, %v", got, err) + } + if got, err := r.Build(ctx, "lemurs", other); err != nil || got != nil { + t.Errorf("no build at another fingerprint: %+v, %v", got, err) + } + for range 2 { + if got, err := r.Archive(ctx, "http://store/v2/x/files/blobs/d", "sha256:"+hex.EncodeToString(sum[:])); err != nil || string(got) != "an archive" { + t.Errorf("the archive: %q, %v", got, err) + } + } + if _, err := r.Archive(ctx, "http://store/v2/x/files/blobs/e", "sha256:"+strings.Repeat("0", 64)); err == nil || fetched != 2 { + t.Errorf("fetched %d times (want once per digest), and an archive of another digest: %v", fetched, err) + } + if got, err := r.Served(ctx, "n2", "mailer", []string{"domain"}); err != nil || fmt.Sprint(got) != "map[domain:example.org]" { + t.Errorf("what n2's mailer serves: %v, %v", got, err) + } + served, err := buscompose.PermissionsFor(buscompose.Principal{Kind: buscompose.KindNode, Node: "n1", ServedReads: []string{"n2.mailer.domain"}}) + if err != nil { + t.Fatal(err) + } + b.mu.Lock() + for _, read := range b.reads { + ok := grantAllows(t, "n1", read[0], read[1]) + if read[0] == ResolvedBucket { + ok = slices.Contains(served.Publish, "$JS.API.CONSUMER.CREATE.KV_"+ResolvedBucket+".*.$KV."+ResolvedBucket+"."+read[1]) + } + if !ok { + t.Errorf("the composer reads %s of %s, which the node's grant does not allow", read[1], read[0]) + } + } + b.mu.Unlock() } // patternBus is a bus whose every watch gets the updates of its bucket matching its pattern, as the bus's own watches @@ -640,3 +770,46 @@ func (r fixedReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow func (r fixedReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) { return nil, nil, nil } +func (r fixedReader) Build(context.Context, string, string) (*delivery.BuilderBuild, error) { + return nil, nil +} +func (r fixedReader) Archive(_ context.Context, url, _ string) ([]byte, error) { + return nil, fmt.Errorf("no archive %s", url) +} +func (r fixedReader) Served(context.Context, string, string, []string) (map[string]any, error) { + return nil, nil +} + +// TestAHaveSaysWhetherItsModuleRunsInsideTheNodesOwn feeds takeModuleHave a module's have and reads whether the node +// counts the module as run inside its own composition (issue 1126, review round 2): a failed have that names nothing +// it runs (NotComposed before the module ever ran) is not, so the next composition does not take the module from its +// manifest past its build. +func TestAHaveSaysWhetherItsModuleRunsInsideTheNodesOwn(t *testing.T) { + for _, row := range []struct { + name string + have delivery.Have + in bool + }{ + {"failed, naming nothing it runs", delivery.Have{Phase: delivery.PhaseFailed, Health: delivery.HealthFailed, + Detail: "the node cannot compose it, and keeps running what it ran: no build"}, false}, + {"failed at a build fingerprint", delivery.Have{FP: "build-0123abcd", Composed: "sha256:" + strings.Repeat("c", 64), Phase: delivery.PhaseFailed, + Health: delivery.HealthFailed, Detail: "the node cannot compose it, and keeps running what it ran: no build"}, true}, + {"running with no fingerprint (mesh-wireguard)", delivery.Have{Phase: delivery.PhaseRunning, Health: delivery.HealthOK}, true}, + } { + t.Run(row.name, func(t *testing.T) { + c := New(Env{Node: "n1"}) + key, err := delivery.HaveKey("n1", "web") + if err != nil { + t.Fatal(err) + } + value, err := delivery.EncodeHave(row.have) + if err != nil { + t.Fatal(err) + } + c.takeModuleHave(agent.Update{Key: key, Value: value}) + if c.inOwn["web"] != row.in { + t.Errorf("web counted inside the node's own composition: %v, want %v", c.inOwn["web"], row.in) + } + }) + } +} diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index e03cc51..6ebc4f1 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -95,7 +95,7 @@ func (s *Shadow) Run(ctx context.Context) error { {delivery.ModulesBucket, ">", (*Composer).takeModule, nil}, {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, // What this node runs moved: an apply ends in its module's have. - {delivery.HaveBucket, node + ".*", nil, nil}, + {delivery.HaveBucket, node + ".>", nil, nil}, } type tagged struct { in int @@ -323,12 +323,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid if mod, ok := modules[module]; ok && mod.FP2 == fp { continue } - digest, err := c.builtManifest(ctx, module, fp) - if err != nil { - return delivery.CandidateAnswer{Head: head, Revision: rev, Verdict: delivery.CandidateRefused, - Refused: fmt.Sprintf("%s runs at %s, whose manifest cannot be read: %v", module, fp, err)}, true - } - kept[module] = digest + kept[module] = fp // composed from that build (frombus.Scope.KeptAt, issue 1126) } // The boxes it runs, by what each module's have and its own key say after a restart (Held), are reused for the // modules it runs: a value sealed again is no difference. @@ -339,7 +334,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid c.mu.Lock() c.wants = held c.mu.Unlock() - a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, Kept: kept, Placements: given, InOwn: inOwn}, fps, running, head, rev) + a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, KeptAt: kept, Placements: given, InOwn: inOwn}, fps, running, head, rev) return a, true }