novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests 1 Releases Wiki Activity

Merge pull request 'Node agent: compose its own modules from their builds and providers' served keys under its own grant (issue 1126, group fix/1126-node-creds)' (#153) from fix/1126-node-creds into main

Browse Source
This commit was merged in pull request #153.
This commit is contained in:
mesh-tools
2026-10-27 20:32:37 +00:00
parent 50c229f905 6a4718b712
commit bb15aab50e
8 changed files with 456 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-host/nodeagent.go
+3 -1
View File
@@ -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}
}
cmd/mesh-host/nodeagent_integration_test.go
+158 -17
View File
@@ -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
go.mod
+1 -1
View File
@@ -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
go.sum
+6
View File
@@ -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=
internal/apply/archive.go
+4
View File
@@ -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) }
internal/selfcompose/selfcompose.go
+90 -47
View File
@@ -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 `<node>.>`).
// A watch of every key is refused by the bus (issue 1126, 2026-10-27 09:21: the fold's `>` stopped every agent).
{delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil, (*Composer).takeRecordChanged},
{delivery.HaveBucket, node + ".>", nil, (*Composer).takeModuleHave},
{delivery.ModulesBucket, ">", (*Composer).takeModule, nil},
{ResolvedBucket, node + ".>", (*Composer).takeValue, nil},
}
@@ -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)
internal/selfcompose/selfcompose_test.go
+191 -18
View File
@@ -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_<bucket>.<name>.$KV.<bucket>.<pattern>`)
// 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)
}
})
}
}
internal/selfcompose/shadow.go
+3 -8
View File
@@ -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
}