From 45439fd74246556b4852b29f0f9c768aebe604d6 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 21:01:16 +0100 Subject: [PATCH] Compose each module from its build and read what providers elsewhere serve, pinning mesh-sdk cebc651, so a self-composing node hands its moved modules and settles mailu's domain (issue 1126) The node read a behind module's source manifest digest and composed every module from the indexer's raw manifest; frombus now takes the build itself (Scope.KeptAt, a fingerprint). The reader gives frombus the builder's key, a templates archive fetched as every archive is (apply.Fetch) once per digest and checked against it, and another node's served keys one by one, as the node's grant allows them. --- cmd/mesh-host/nodeagent.go | 4 +- go.mod | 2 +- go.sum | 2 + internal/apply/archive.go | 4 + internal/selfcompose/selfcompose.go | 110 +++++++++++++++-------- internal/selfcompose/selfcompose_test.go | 60 +++++++++++++ internal/selfcompose/shadow.go | 9 +- 7 files changed, 147 insertions(+), 44 deletions(-) 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/go.mod b/go.mod index c5812c9..fe43222 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.20261027195758-cebc6519d5b0 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..736cc33 100644 --- a/go.sum +++ b/go.sum @@ -12,6 +12,8 @@ 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= 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 e060e39..9946c9a 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. @@ -421,19 +427,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)) @@ -548,30 +547,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. @@ -655,6 +630,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 c8b3896..f038c3b 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -2,8 +2,11 @@ package selfcompose import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "fmt" + "slices" "strings" "sync" "testing" @@ -556,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 @@ -719,3 +770,12 @@ 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 +} diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index f84cf88..6ebc4f1 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -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 }