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.
This commit is contained in:
2026-10-27 21:01:16 +01:00
parent 0184e2a5b8
commit 45439fd742
7 changed files with 147 additions and 44 deletions
+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}
}
+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.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
+2
View File
@@ -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=
+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) }
+75 -35
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.
@@ -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)
+60
View File
@@ -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
}
+2 -7
View File
@@ -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
}