1914 lines
89 KiB
Go
1914 lines
89 KiB
Go
package selfcompose
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
|
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
|
"git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
|
|
"git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
|
|
"git.novox.be/novox/mesh-contracts/placement/go/seats/meshwants"
|
|
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
|
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
|
"git.novox.be/novox/mesh-sdk/go/declaration"
|
|
"github.com/novox/mesh-host/internal/buckets"
|
|
"github.com/novox/mesh-host/internal/buckets/bucketstest"
|
|
"github.com/novox/mesh-host/internal/compositions"
|
|
"os"
|
|
"reflect"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go/overlay"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/novox/mesh-host/internal/compose"
|
|
"github.com/novox/mesh-host/internal/compose/frombus"
|
|
"github.com/novox/mesh-host/internal/nodegrant"
|
|
|
|
"github.com/novox/mesh-host/internal/agent"
|
|
)
|
|
|
|
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 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
|
|
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{kv: map[string][]byte{}}
|
|
}
|
|
|
|
func (b *fakeBus) env() Env {
|
|
return Env{Node: "n1",
|
|
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.send(bucket, agent.Update{Key: key, Value: value})
|
|
}
|
|
|
|
// grantAllows says whether the node's grant (the grants part's, nodegrant.For) 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 := nodegrant.For(node)
|
|
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 {
|
|
t.Helper()
|
|
select {
|
|
case u := <-out:
|
|
return u
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("no update from the composer")
|
|
return agent.Update{}
|
|
}
|
|
}
|
|
|
|
func quiet(t *testing.T, out <-chan agent.Update) {
|
|
t.Helper()
|
|
select {
|
|
case u := <-out:
|
|
t.Fatalf("the composer handed %s, and nothing changed", u.Key)
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
// Issue 1126 (ADR 0547 §2): the node composes its wanted modules and its own composition from its inputs on the bus once
|
|
// every input was read, hands each, then Caught; an input that changes nothing composed hands nothing; a module no want
|
|
// names any more is taken back; a node that cannot compose says so once and hands nothing.
|
|
func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|
b := newFakeBus()
|
|
fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
|
|
digest := meshcatalogclient.ManifestDigest([]byte(lemurs))
|
|
man, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "lemurs", Manifest: lemurs})
|
|
key, _ := meshcatalogclient.ManifestsKey(digest)
|
|
b.kv[buckets.Manifests+"/"+key] = man
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
env := b.env()
|
|
// The vault's own: keeper's password sealed to n1's key, sealed again (other bytes) at every ask.
|
|
var asks, generation int
|
|
sealedAt := map[string]int{} // each box handed, by the generation it opens to
|
|
env.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
asks++
|
|
sealedAt[fmt.Sprintf(`{"ask":%d}`, asks)] = generation + 1
|
|
return json.RawMessage(fmt.Sprintf(`{"node":"n1","key":"sk","own":[{"module":"keeper","name":"password","origin":"made",`+
|
|
`"generation":%d,"sealed":{"ask":%d}}],"handed":{"keeper":false}}`, generation+1, asks)), nil
|
|
}
|
|
secretChanged := make(chan []byte, 1)
|
|
env.Hear = func(subject string) (<-chan []byte, func(), error) {
|
|
if subject != VaultSecretChanged {
|
|
t.Errorf("heard %s", subject)
|
|
}
|
|
return secretChanged, func() {}, nil
|
|
}
|
|
var composed atomic.Int64 // passes made: each reads what runs once
|
|
env.Running = func() []compositions.Composed { composed.Add(1); return nil }
|
|
c := New(env)
|
|
out, err := c.Watch(ctx, "n1.>")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// oldie is placed and no want names it (a module at a build of the old form, nats on novox): the node runs it inside
|
|
// its own composition, and composes it there again, even before the agent has applied anything (Running empty).
|
|
oldie := `{"module":"oldie","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"oldie.service","state":"running","boot":"enabled"}]}`
|
|
odigest := meshcatalogclient.ManifestDigest([]byte(oldie))
|
|
oman, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "oldie", Manifest: oldie})
|
|
okey, _ := meshcatalogclient.ManifestsKey(odigest)
|
|
b.kv[buckets.Manifests+"/"+okey] = oman
|
|
omod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64), Manifest: odigest, Repository: "novox/o", Commit: "c"})
|
|
place, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}})
|
|
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
|
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
|
want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true})
|
|
nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
|
|
b.put(t, buckets.Placement, "placement.n1", place)
|
|
b.put(t, buckets.Have, "n1.node", have)
|
|
oldHave, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Build: "build-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
|
b.put(t, buckets.Have, "n1.oldie", oldHave)
|
|
b.put(t, buckets.Modules, "lemurs", mod)
|
|
b.put(t, buckets.Modules, "oldie", omod)
|
|
b.put(t, buckets.Wants, "want.n1.lemurs", want)
|
|
b.put(t, buckets.Wants, "want.n1.node", nodeWant)
|
|
for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
|
b.send(bucket, agent.Update{Caught: true})
|
|
}
|
|
got := map[string]bool{}
|
|
var own compositions.Composed
|
|
for {
|
|
u := next(t, out)
|
|
if u.Caught {
|
|
break
|
|
}
|
|
c, err := compositions.Decode(u.Value)
|
|
if err != nil {
|
|
t.Fatalf("%s: %v", u.Key, err)
|
|
}
|
|
if c.Node != "n1" {
|
|
t.Errorf("%s composed for %q", u.Key, c.Node)
|
|
}
|
|
if c.Module == "" {
|
|
own = c
|
|
}
|
|
got[u.Key] = true
|
|
}
|
|
var ownText []string
|
|
for _, r := range own.Resources {
|
|
ownText = append(ownText, string(r))
|
|
}
|
|
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()
|
|
// Issue 774, 2026-10-31: the verb compose makes a pass now and says how it went; a composer that runs no watch
|
|
// makes none, and says so rather than waiting.
|
|
// The pass made at the start is less than askedWithin old for a moment: the ask waits that out first.
|
|
setFor(t, &askedWithin, 200*time.Millisecond)
|
|
time.Sleep(250 * time.Millisecond)
|
|
madeBefore := composed.Load()
|
|
cctx, ccancel := context.WithTimeout(ctx, 5*time.Second)
|
|
nowPass, err := c.ComposeNow(cctx)
|
|
ccancel()
|
|
if err != nil || nowPass.At.IsZero() || nowPass.Refused != "" || !nowPass.Again.IsZero() || composed.Load() == madeBefore {
|
|
t.Errorf("compose now: %+v (%v), %d pass(es) made; want one pass, composed", nowPass, err, composed.Load()-madeBefore)
|
|
}
|
|
// Asked again at once, the last pass is the answer: a caller in a loop drives no passes back to back (review round 1).
|
|
madeBefore = composed.Load()
|
|
if again, err := c.ComposeNow(ctx); err != nil || !again.Earlier || !again.At.Equal(nowPass.At) || composed.Load() != madeBefore {
|
|
t.Errorf("compose now asked again at once: %+v (%v), %d pass(es) made; want the last pass, none made", again, err, composed.Load()-madeBefore)
|
|
}
|
|
// A composer whose loop runs not refuses at once, not after the verb's whole bound.
|
|
idle, icancel := context.WithTimeout(ctx, 5*time.Second)
|
|
if _, err := New(env).ComposeNow(idle); err == nil || !strings.Contains(err.Error(), "runs no loop") || idle.Err() != nil {
|
|
t.Errorf("a composer that watches nothing answered compose now: %v", err)
|
|
}
|
|
icancel()
|
|
lemursKey, _ := compositions.Key("n1", "lemurs", fp)
|
|
nodeKey, _ := compositions.NodeKey("n1")
|
|
if !got[lemursKey] || !got[nodeKey] || len(got) != 2 {
|
|
b.mu.Lock()
|
|
t.Logf("said: %v", b.said)
|
|
b.mu.Unlock()
|
|
t.Fatalf("handed %v, want %s and %s", got, lemursKey, nodeKey)
|
|
}
|
|
// ADR 0601 §1: the module composed today at the build it runs, as the agent compares it with what runs. The same inputs give
|
|
// the digest handed; a changed input (here its manifest's service, as a changed setting would change a file) another; a
|
|
// build that cannot be composed (no done build or manifest of it: held) gives none.
|
|
c.mu.Lock()
|
|
handedDig := c.handed[lemursKey]
|
|
c.mu.Unlock()
|
|
if dig, ok := c.ComposedToday(ctx, "lemurs", fp); !ok || dig != handedDig {
|
|
t.Errorf("composed today at the build it runs: %s %v, want %s", dig, ok, handedDig)
|
|
}
|
|
if dig, ok := c.ComposedToday(ctx, "lemurs", meshcatalogclient.SourcePrefix2+strings.Repeat("9", 64)); ok {
|
|
t.Errorf("a build that cannot be composed was composed today: %s", dig)
|
|
}
|
|
chgJSON := strings.Replace(lemurs, `"state":"running"`, `"state":"stopped"`, 1)
|
|
chgDigest := meshcatalogclient.ManifestDigest([]byte(chgJSON))
|
|
chgMan, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "lemurs", Manifest: chgJSON})
|
|
chgKey, _ := meshcatalogclient.ManifestsKey(chgDigest)
|
|
b.mu.Lock()
|
|
b.kv[buckets.Manifests+"/"+chgKey] = chgMan
|
|
b.mu.Unlock()
|
|
chgMod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: chgDigest, Repository: "novox/x", Commit: "c"})
|
|
c.takeModule(agent.Update{Key: "lemurs", Value: chgMod})
|
|
if dig, ok := c.ComposedToday(ctx, "lemurs", fp); !ok || dig == handedDig {
|
|
t.Errorf("composed today with a changed input: %s %v, want another digest than %s", dig, ok, handedDig)
|
|
}
|
|
c.takeModule(agent.Update{Key: "lemurs", Value: mod})
|
|
|
|
// Issue 1126, the switch made safe: a bus not answering for a moment (a reload of its accounts) is tried again and
|
|
// refuses nothing; a pass whose run ended refuses nothing either.
|
|
setFor(t, &composeAgain, []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond})
|
|
fresh := func(get func(context.Context, string, string) ([]byte, error)) (*Composer, *[]string) {
|
|
var refusals []string
|
|
e := b.env()
|
|
e.Get = get
|
|
e.Refused = func(why string, held map[string]string) {
|
|
for _, m := range heldModules(held) {
|
|
why += "held " + m + ": " + held[m]
|
|
}
|
|
refusals = append(refusals, why)
|
|
}
|
|
c2 := New(e)
|
|
c.mu.Lock()
|
|
for k, v := range c.wants {
|
|
c2.wants[k] = v
|
|
}
|
|
for k, v := range c.placements {
|
|
c2.placements[k] = v
|
|
}
|
|
for k, v := range c.records {
|
|
c2.records[k] = v
|
|
}
|
|
for k, v := range c.modules {
|
|
c2.modules[k] = v
|
|
}
|
|
c.mu.Unlock()
|
|
return c2, &refusals
|
|
}
|
|
setFor(t, &passSaysEvery, 5*time.Millisecond)
|
|
for name, row := range map[string]struct {
|
|
failures int // reads failing before the bus answers
|
|
cancel bool
|
|
composes bool
|
|
fail error // what each failing read says; JetStream not enabled when nil
|
|
slow time.Duration // each failing read takes this long
|
|
// again: the refusal says when the node composes again, and a pass is owed (issue 774, 2026-10-31: ace's
|
|
// route-proxy said only "the vault's own could not be asked ... context deadline exceeded").
|
|
again bool
|
|
}{
|
|
"the bus paused for one read": {failures: 1, composes: true},
|
|
"the bus paused past every try": {failures: 100, again: true},
|
|
"the run ended mid-pass": {failures: 100, cancel: true},
|
|
"the vault did not answer in time": {failures: 100, again: true, slow: 30 * time.Millisecond,
|
|
fail: fmt.Errorf("the vault's own could not be asked: no answer from mesh.seat.mesh-vault.call.own.node~n1: %w", context.DeadlineExceeded)},
|
|
"a refusal": {failures: 100, fail: errors.New("nats says ${dir:mesh-state}, and nats declares no directory")},
|
|
} {
|
|
b.mu.Lock()
|
|
saidBefore := len(b.said)
|
|
b.mu.Unlock()
|
|
failed := 0
|
|
pctx, pcancel := context.WithCancel(ctx)
|
|
c2, refusals := fresh(func(gctx context.Context, bucket, key string) ([]byte, error) {
|
|
if failed < row.failures {
|
|
failed++
|
|
if row.cancel {
|
|
pcancel()
|
|
return nil, fmt.Errorf("the vault's own could not be asked: %w", context.Canceled)
|
|
}
|
|
time.Sleep(row.slow)
|
|
if row.fail != nil {
|
|
return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, row.fail)
|
|
}
|
|
return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, nats.ErrJetStreamNotEnabled)
|
|
}
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
return b.kv[bucket+"/"+key], nil
|
|
})
|
|
handed := c2.compose(pctx)
|
|
pcancel()
|
|
tries := 1 + len(composeAgain) // a bus paused for a moment is tried again at once; a timeout or a refusal is not
|
|
if row.fail != nil {
|
|
tries = 1
|
|
}
|
|
switch {
|
|
case row.composes && (len(handed) == 0 || len(*refusals) != 0):
|
|
t.Errorf("%s: handed %d, refusals %q; want composed and nothing refused", name, len(handed), *refusals)
|
|
case !row.composes && !row.cancel && (len(*refusals) != 1 || failed != tries):
|
|
t.Errorf("%s: refusals %q after %d reads; want one refusal after %d tries", name, *refusals, failed, tries)
|
|
case row.cancel && len(*refusals) != 0:
|
|
t.Errorf("%s: a pass whose run ended refused %q", name, *refusals)
|
|
}
|
|
if !row.composes && !row.cancel && len(*refusals) > 0 {
|
|
// The retry pass says the same try the refusal named (review round 1: it said try 2 for try 1).
|
|
says := strings.Contains((*refusals)[0], "; the node composes again at ") && strings.HasSuffix((*refusals)[0], "(try 1)") &&
|
|
strings.Contains(c2.againLine(), "(try 1)")
|
|
if says != row.again || c2.owed.Load() != row.again || (c2.last.Again.IsZero() == row.again) || c2.last.Refused != (*refusals)[0] {
|
|
t.Errorf("%s: refused %q, owes a pass %v, last pass %+v; want the time it composes again said and owed: %v",
|
|
name, (*refusals)[0], c2.owed.Load(), c2.last, row.again)
|
|
}
|
|
}
|
|
// A pass that runs on says so, and what it waits on (issue 774, 2026-10-31: ace was silent for five minutes).
|
|
if row.slow > 0 {
|
|
b.mu.Lock()
|
|
long := false
|
|
for _, line := range b.said[saidBefore:] {
|
|
long = long || strings.Contains(line, "the node's composition has run for ") && strings.Contains(line, " and waits on ")
|
|
}
|
|
b.mu.Unlock()
|
|
if !long {
|
|
t.Errorf("%s: a pass that ran on said nothing of what it waits on", name)
|
|
}
|
|
}
|
|
}
|
|
if !Transient(fmt.Errorf("the manifest of avahi: the revisions of x in mesh-delivery-indexer_manifests: nats: jetstream not enabled")) ||
|
|
!Transient(fmt.Errorf("x: %w", nats.ErrNoResponders)) || Transient(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) ||
|
|
Transient(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) {
|
|
t.Error("Transient misjudges a bus that does not answer, or a refusal")
|
|
}
|
|
// Issue 1126, the applying path (review of the seed): one module's own refusal holds that module alone. On novox,
|
|
// de-spiegel's unset route names (its manifest at the indexer's commit e25e1beb, captured) refused the whole node and
|
|
// marked all its modules not composed. Now lemurs and proxy are handed; de-spiegel is held alone (issue 1960: a hold
|
|
// never spreads to proxy, which composes without its routes, nor to its smtp provider): what was handed of de-spiegel
|
|
// stays handed, so the node keeps running it, and the agent is told it alone, in its config form's words.
|
|
spiegel, err := os.ReadFile("testdata/de-spiegel-e25e1beb.module.json")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
proxy := `{"module":"proxy","version":"1","capabilities":["systemd"],"provides":[{"name":"route","scope":"mesh"},{"name":"smtp","scope":"mesh"}],` +
|
|
`"resources":[{"id":"service","type":"service","unit":"proxy.service","state":"running","boot":"enabled"},` +
|
|
`{"id":"pkg","type":"package","package":"haproxy"}]}`
|
|
c3, told := fresh(env.Get)
|
|
for m, raw := range map[string]string{"de-spiegel": string(spiegel), "proxy": proxy} {
|
|
dig := meshcatalogclient.ManifestDigest([]byte(raw))
|
|
enc, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: m, Manifest: raw})
|
|
k, _ := meshcatalogclient.ManifestsKey(dig)
|
|
b.mu.Lock()
|
|
b.kv[buckets.Manifests+"/"+k] = enc
|
|
b.mu.Unlock()
|
|
mfp := meshcatalogclient.SourcePrefix2 + strings.Repeat(map[string]string{"de-spiegel": "d", "proxy": "e"}[m], 64)
|
|
c3.modules[m] = meshcatalog.ModulesByModuleValue{Fp2: mfp, Manifest: dig, Repository: "novox/" + m, Commit: "c"}
|
|
c3.wants[m] = meshwants.WantsByWantNodeModuleValue{Fp: mfp, Self: true}
|
|
}
|
|
p3 := c3.placements["n1"]
|
|
p3.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p3.Modules...), meshmoduleassigner.PlacedModule{Module: "proxy"}, meshmoduleassigner.PlacedModule{Module: "de-spiegel"})
|
|
c3.placements["n1"] = p3
|
|
r3 := c3.records["n1"]
|
|
r3.Capabilities = []string{"systemd", "container-runtime"}
|
|
c3.records["n1"] = r3
|
|
spiegelKey, _ := compositions.Key("n1", "de-spiegel", c3.wants["de-spiegel"].Fp)
|
|
proxyKey, _ := compositions.Key("n1", "proxy", c3.wants["proxy"].Fp)
|
|
c3.handed[spiegelKey] = "sha256:" + strings.Repeat("5", 64) // what the node runs of it, handed before
|
|
handed3 := map[string]agent.Update{}
|
|
for _, u := range c3.compose(ctx) {
|
|
handed3[u.Key] = u
|
|
}
|
|
lemursKey3, _ := compositions.Key("n1", "lemurs", c3.wants["lemurs"].Fp)
|
|
if u, ok := handed3[lemursKey3]; !ok || u.Deleted {
|
|
t.Errorf("lemurs, which composes, is not handed beside de-spiegel's refusal: %v (told %q)", handed3, *told)
|
|
}
|
|
// Issue 1960: a hold never spreads. proxy is handed, composed without held de-spiegel's routes, which do not compose.
|
|
if u, ok := handed3[proxyKey]; !ok || u.Deleted {
|
|
t.Errorf("proxy is not handed beside held de-spiegel: %v (told %q)", handed3, *told)
|
|
}
|
|
if u, ok := handed3[spiegelKey]; ok {
|
|
t.Errorf("held de-spiegel is handed %+v, want what was handed of it kept", u)
|
|
}
|
|
// ADR 0565 §6 (1953): the pass keeps the packages each handed composition declares, with their module, for the node's
|
|
// key of the engine's state packages.
|
|
if d, err := c3.DeclaredPackages(); err != nil || len(d.Packages) != 1 || d.Packages[0].Package != "haproxy" ||
|
|
strings.Join(d.Packages[0].Modules, ",") != "proxy" || d.Node != "n1" {
|
|
t.Errorf("the node's packages after the pass are %+v (%v)", d, err)
|
|
}
|
|
if len(*told) != 1 || !strings.HasPrefix((*told)[0], "held de-spiegel: ") || !strings.Contains((*told)[0], "${config:praktijk-name}") ||
|
|
!strings.Contains((*told)[0], "without what it contributes to route") || strings.Contains((*told)[0], "held proxy") ||
|
|
strings.Contains((*told)[0], "held lemurs") {
|
|
t.Errorf("the agent was told %q, want de-spiegel held alone, in its config form's words, without its routes", *told)
|
|
}
|
|
// Issue 774, 2026-10-30 19:13: a manifest the mesh refuses (power's bar item at mesh-catalog ca8d6f33, captured, with
|
|
// attributes its bar places by none of) refused every node's composition and failed all its modules. Now power is
|
|
// held alone, naming why, and lemurs is handed.
|
|
power, err := os.ReadFile("testdata/power-ca8d6f33.module.json")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
c4, told4 := fresh(env.Get)
|
|
powerDig := meshcatalogclient.ManifestDigest(power)
|
|
powerEnc, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "power", Manifest: string(power)})
|
|
powerKey, _ := meshcatalogclient.ManifestsKey(powerDig)
|
|
b.mu.Lock()
|
|
b.kv[buckets.Manifests+"/"+powerKey] = powerEnc
|
|
b.mu.Unlock()
|
|
powerFP := meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64)
|
|
c4.modules["power"] = meshcatalog.ModulesByModuleValue{Fp2: powerFP, Manifest: powerDig, Repository: "novox/mesh-catalog", Commit: "ca8d6f33",
|
|
// The indexer's verdict on the captured manifest (ADR 0578): the engine judges nothing, it takes this.
|
|
Verdict: meshcatalogclient.VerdictRefused, Problems: []string{`bar item "shows" is none of them`}}
|
|
c4.wants["power"] = meshwants.WantsByWantNodeModuleValue{Fp: powerFP, Self: true}
|
|
p4 := c4.placements["n1"]
|
|
p4.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p4.Modules...), meshmoduleassigner.PlacedModule{Module: "power"})
|
|
c4.placements["n1"] = p4
|
|
handed4 := map[string]agent.Update{}
|
|
for _, u := range c4.compose(ctx) {
|
|
handed4[u.Key] = u
|
|
}
|
|
if u, ok := handed4[lemursKey3]; !ok || u.Deleted {
|
|
t.Errorf("lemurs is not handed beside power's refused manifest: %v (told %q)", handed4, *told4)
|
|
}
|
|
if len(*told4) != 1 || !strings.HasPrefix((*told4)[0], "held power: ") || !strings.Contains((*told4)[0], `"shows" is none of them`) {
|
|
t.Errorf("the agent was told %q, want power held alone, naming its item", *told4)
|
|
}
|
|
// Review of issue 774 (the kept read): mesh-wireguard, which the mesh provides and which never has a fingerprint, is
|
|
// folded in the node's own and composed from the manifest mesh-sdk provides (overlay.ProvidedManifest), not from
|
|
// what runs nor the kept declaration: with nothing running (the kept read retired, Running empty) the node's own
|
|
// still carries its resources, so retiring the read removes nothing of it.
|
|
c5, told5 := fresh(env.Get)
|
|
p5 := c5.placements["n1"]
|
|
p5.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p5.Modules...), meshmoduleassigner.PlacedModule{Module: overlay.Name})
|
|
p5.Decided.Endpoint = "n1.example:51820"
|
|
c5.placements["n1"] = p5
|
|
c5.inOwn[overlay.Name] = true
|
|
nodeKey5, _ := compositions.NodeKey("n1")
|
|
var own5 string
|
|
for _, u := range c5.compose(ctx) {
|
|
if u.Key == nodeKey5 {
|
|
own5 = string(u.Value)
|
|
}
|
|
}
|
|
if len(*told5) != 0 || !strings.Contains(own5, overlay.Name+".") {
|
|
t.Errorf("the node's own with %s folded and nothing running: told %q, carries %s: %.300s", overlay.Name, *told5,
|
|
overlay.Name, own5)
|
|
}
|
|
|
|
// A read the bus did not answer, timed out or not, is tried again later; the run's own end and a refusal are not.
|
|
if !BusFailed(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) || !BusFailed(fmt.Errorf("x: %w", nats.ErrNoResponders)) ||
|
|
!BusFailed(fmt.Errorf("the have of web: %w", nats.ErrTimeout)) || BusFailed(fmt.Errorf("x: %w", context.Canceled)) ||
|
|
BusFailed(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) || BusFailed(nil) {
|
|
t.Error("BusFailed misjudges a read the bus did not answer, the run's end, or a refusal")
|
|
}
|
|
|
|
// Issue 1126, C2: a config value is laid as the controller lays it (catalogue.LaidValue), from the mesh's, the node's
|
|
// or the module's own and never from a default or an unset one, both of this node's and of what another node serves:
|
|
// a default laid blanked Element's links on ace.
|
|
for _, row := range []struct {
|
|
source string
|
|
laid bool
|
|
}{{"mesh", true}, {"node", true}, {"module", true}, {"default", false}, {"unset", false}, {"", false}} {
|
|
probe := New(env)
|
|
value := []byte(`{"value":["x"],"source":"` + row.source + `"}`)
|
|
probe.takeValue(agent.Update{Key: "n1.lemurs.links", Value: value})
|
|
_, laid := probe.values["lemurs"]["links"]
|
|
b.mu.Lock()
|
|
b.kv[ResolvedBucket+"/n2.lemurs.links"] = value
|
|
b.mu.Unlock()
|
|
served, err := reader{c: probe}.Served(ctx, "n2", "lemurs", []string{"links"})
|
|
if laid != row.laid || err != nil || (served["links"] != nil) != row.laid {
|
|
t.Errorf("a value from %q: laid %v, served %v (%v), want laid %v", row.source, laid, served, err, row.laid)
|
|
}
|
|
}
|
|
|
|
// A config value of another module, and a value changing nothing composed, hand nothing.
|
|
b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1,"source":"node"}`))
|
|
quiet(t, out)
|
|
|
|
// Issue 774, 2026-10-30 19:30: node-tools' want moved to a build not yet done, and the node stayed refused after it
|
|
// landed until another input moved. A done build of a module the node wants, at its want's or the trunk's build,
|
|
// makes a pass, and so does one of a module placed here at its trunk's build though no want names it (novox/hq ADR 0593
|
|
// §6: the composition state holds it at its newest done build before its want moves); a build not done, of another
|
|
// build or of a module neither wanted nor placed here, makes none.
|
|
passes := func() int64 { return composed.Load() }
|
|
for _, row := range []struct {
|
|
name, module, fp, state string
|
|
pass bool
|
|
}{
|
|
{"lemurs's wanted build, not done", "lemurs", fp, "building", false},
|
|
{"lemurs's wanted build lands", "lemurs", fp, builds.BuildDone, true},
|
|
{"another build of lemurs", "lemurs", meshcatalogclient.SourcePrefix2 + strings.Repeat("e", 64), builds.BuildDone, false},
|
|
{"a placed module no want names, at its trunk's build", "oldie", meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64), builds.BuildDone, true},
|
|
{"a module neither wanted nor placed here", "ghost", meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64), builds.BuildDone, false},
|
|
} {
|
|
before := passes()
|
|
key, _ := builds.BuilderBuildKey(row.module, row.fp)
|
|
build := builds.BuilderBuild{State: row.state}
|
|
if row.state == builds.BuildDone {
|
|
build.Manifest = digest
|
|
}
|
|
raw, err := builds.EncodeBuilderBuild(build)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.put(t, builds.BuilderBuildsBucket, key, raw)
|
|
quiet(t, out)
|
|
if got := passes() > before; got != row.pass {
|
|
t.Errorf("%s: a pass made %v, want %v", row.name, got, row.pass)
|
|
}
|
|
}
|
|
// Issue 774, 2026-10-31: every put under want.n1.> made a pass, and each pass asked the vault for the node's own
|
|
// secrets. A key written again with what it held, or a want whose words moved and not its build, makes none.
|
|
worded, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true, Words: "lemurs is held at aaaaaaaaaaaa"})
|
|
elsewhere, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: meshcatalogclient.SourcePrefix2 + strings.Repeat("e", 64), Self: true})
|
|
// Research 072 row 37: the placement converted to one key per module and node moves nothing composed; a write in
|
|
// keys makes its pass at its node key, the commit, and not at a module key written before it nor at the mirror.
|
|
placeBase := meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}}}
|
|
placeKeys := meshmoduleassignerclient.NewKeys()
|
|
keyRows := func(after meshmoduleassigner.Placement, what string) []struct {
|
|
name, bucket, key string
|
|
value []byte
|
|
pass bool
|
|
} {
|
|
writes, err := placeKeys.Writes("n1", &after)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var rows []struct {
|
|
name, bucket, key string
|
|
value []byte
|
|
pass bool
|
|
}
|
|
for i, w := range writes {
|
|
_, _ = placeKeys.Put(w.Key, w.Value, uint64(i+1), w.Value == nil)
|
|
rows = append(rows, struct {
|
|
name, bucket, key string
|
|
value []byte
|
|
pass bool
|
|
}{what + ": " + w.Key, buckets.Placement, w.Key, w.Value, w.Commit && what != "converted"})
|
|
}
|
|
return rows
|
|
}
|
|
_, _ = placeKeys.Put("placement.n1", place, 1, false)
|
|
lemursMoved := placeBase
|
|
lemursMoved.Modules = []meshmoduleassigner.PlacedModule{{Module: "lemurs", Values: map[string]any{"greeting": "hoi"}}, {Module: "oldie"}}
|
|
var placementRows []struct {
|
|
name, bucket, key string
|
|
value []byte
|
|
pass bool
|
|
}
|
|
for _, step := range []struct {
|
|
what string
|
|
p meshmoduleassigner.Placement
|
|
}{{"converted", placeBase}, {"lemurs moved", lemursMoved}, {"lemurs back", placeBase}} {
|
|
placementRows = append(placementRows, keyRows(step.p, step.what)...)
|
|
}
|
|
for _, row := range append([]struct {
|
|
name, bucket, key string
|
|
value []byte
|
|
pass bool
|
|
}{
|
|
{"lemurs's want again", buckets.Wants, "want.n1.lemurs", want, false},
|
|
{"the node's own want again", buckets.Wants, "want.n1.node", nodeWant, false},
|
|
{"lemurs's want with other words", buckets.Wants, "want.n1.lemurs", worded, false},
|
|
{"the placement again", buckets.Placement, "placement.n1", place, false},
|
|
{"lemurs's module again", buckets.Modules, "lemurs", mod, false},
|
|
{"lemurs's want at another build", buckets.Wants, "want.n1.lemurs", elsewhere, true},
|
|
{"lemurs's want back at its build", buckets.Wants, "want.n1.lemurs", want, true},
|
|
}, placementRows...) {
|
|
before := passes()
|
|
b.put(t, row.bucket, row.key, row.value)
|
|
if !row.pass {
|
|
quiet(t, out)
|
|
} else {
|
|
for done := false; !done; {
|
|
select {
|
|
case <-out:
|
|
case <-time.After(200 * time.Millisecond):
|
|
done = true
|
|
}
|
|
}
|
|
}
|
|
if got := passes() > before; got != row.pass {
|
|
t.Errorf("%s: a pass made %v, want %v", row.name, got, row.pass)
|
|
}
|
|
}
|
|
// And today's sequence whole: tool runs at its old build; its want and the trunk move to a build not yet done, and
|
|
// the node holds tool, keeping what runs, naming why; the build lands, and with no other change tool is handed at it.
|
|
{
|
|
tb := newFakeBus()
|
|
tool := `{"module":"tool","version":"1","capabilities":["systemd"],"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]},` +
|
|
`"resources":[{"id":"service","type":"service","unit":"tool.service","state":"running","boot":"enabled"}]}`
|
|
tdigest := meshcatalogclient.ManifestDigest([]byte(tool))
|
|
tman, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "tool", Manifest: tool})
|
|
tkey, _ := meshcatalogclient.ManifestsKey(tdigest)
|
|
tb.kv[buckets.Manifests+"/"+tkey] = tman
|
|
old, next2 := meshcatalogclient.SourcePrefix2+strings.Repeat("1", 64), meshcatalogclient.SourcePrefix2+strings.Repeat("2", 64)
|
|
tenv := tb.env()
|
|
var held []string
|
|
var hmu sync.Mutex
|
|
tenv.Refused = func(why string, h map[string]string) {
|
|
hmu.Lock()
|
|
defer hmu.Unlock()
|
|
held = append(held, why+fmt.Sprint(h))
|
|
}
|
|
tenv.Running = func() []compositions.Composed {
|
|
return []compositions.Composed{{Node: "n1", Module: "tool", FP: old, Resources: []json.RawMessage{
|
|
json.RawMessage(`{"id":"service","module":"tool","type":"service","unit":"tool.service","state":"running","boot":"enabled"}`)}}}
|
|
}
|
|
tc := New(tenv)
|
|
tout, err := tc.Watch(ctx, "n1.>")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
tplace, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []meshmoduleassigner.PlacedModule{{Module: "tool"}}})
|
|
tmod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: next2, Manifest: tdigest, Repository: "novox/t", Commit: "c"})
|
|
twant, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: next2, Self: true})
|
|
building, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: "building"})
|
|
bkey, _ := builds.BuilderBuildKey("tool", next2)
|
|
tb.kv[builds.BuilderBuildsBucket+"/"+bkey] = building
|
|
tb.put(t, buckets.Placement, "placement.n1", tplace)
|
|
tb.put(t, buckets.Have, "n1.node", have)
|
|
tb.put(t, buckets.Modules, "tool", tmod)
|
|
tb.put(t, buckets.Wants, "want.n1.tool", twant)
|
|
for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules,
|
|
ResolvedBucket, builds.BuilderBuildsBucket} {
|
|
tb.send(bucket, agent.Update{Caught: true})
|
|
}
|
|
toolKey, _ := compositions.Key("n1", "tool", next2)
|
|
for u := next(t, tout); !u.Caught; u = next(t, tout) {
|
|
if u.Key == toolKey {
|
|
t.Errorf("tool is handed at a build not yet done")
|
|
}
|
|
}
|
|
hmu.Lock()
|
|
if len(held) == 0 || !strings.Contains(held[len(held)-1], "no done build yet") {
|
|
t.Errorf("the node was told %q, want tool held for its build not yet done", held)
|
|
}
|
|
hmu.Unlock()
|
|
done, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: tdigest,
|
|
Artifacts: []builds.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://tool/files/blobs/sha256:" + strings.Repeat("9", 64),
|
|
Digest: "sha256:" + strings.Repeat("9", 64)}}})
|
|
tb.mu.Lock()
|
|
tb.kv[builds.BuilderBuildsBucket+"/"+bkey] = done
|
|
tb.mu.Unlock()
|
|
tb.put(t, builds.BuilderBuildsBucket, bkey, done)
|
|
if u := next(t, tout); u.Key != toolKey || u.Deleted {
|
|
t.Errorf("the build landed and the node handed %s, want tool at its new build", u.Key)
|
|
}
|
|
}
|
|
|
|
// A want removed takes its composition back.
|
|
removed, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Removed: true})
|
|
b.put(t, buckets.Wants, "want.n1.lemurs", removed)
|
|
if u := next(t, out); u.Key != lemursKey || !u.Deleted {
|
|
t.Fatalf("a removed want handed %+v", u)
|
|
}
|
|
|
|
// A want of a module whose manifest cannot be read: said once, nothing handed, what was handed stands.
|
|
gone := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64)
|
|
missing, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: gone, Self: true})
|
|
b.put(t, buckets.Wants, "want.n1.lemurs", missing)
|
|
quiet(t, out)
|
|
b.put(t, buckets.Wants, "want.n1.lemurs", missing)
|
|
quiet(t, out)
|
|
b.mu.Lock()
|
|
n := 0
|
|
for _, s := range b.said {
|
|
if strings.Contains(s, "lemurs is wanted at "+gone) {
|
|
n++
|
|
}
|
|
}
|
|
b.mu.Unlock()
|
|
if n != 1 {
|
|
t.Errorf("the fault was said %d times, want once: %v", n, b.said)
|
|
}
|
|
|
|
// An own secret comes sealed from the vault into the composition; sealed again at the same generation it is the value
|
|
// first had, so nothing is handed again; a new generation is handed.
|
|
b.put(t, buckets.Wants, "want.n1.lemurs", want)
|
|
if u := next(t, out); u.Key != lemursKey {
|
|
t.Fatalf("lemurs wanted back handed %s", u.Key)
|
|
}
|
|
keeper := `{"module":"keeper","version":"1","own-secrets":{"password":"/var/lib/mesh/keeper/password"},` +
|
|
`"resources":[{"id":"state","type":"directory","path":"/var/lib/mesh/keeper","mode":"0700"}]}`
|
|
kdigest := meshcatalogclient.ManifestDigest([]byte(keeper))
|
|
kman, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "keeper", Manifest: keeper})
|
|
kkey, _ := meshcatalogclient.ManifestsKey(kdigest)
|
|
b.mu.Lock()
|
|
b.kv[buckets.Manifests+"/"+kkey] = kman
|
|
b.mu.Unlock()
|
|
kmod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"})
|
|
both, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}, {Module: "keeper"}}})
|
|
b.put(t, buckets.Modules, "keeper", kmod)
|
|
// n1 is held in keys per module and node since the rows above, which a legacy key no longer moves.
|
|
bothPlaced, _ := meshmoduleassignerclient.DecodePlacement(both)
|
|
bothWrites, err := placeKeys.Writes("n1", &bothPlaced)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, w := range bothWrites {
|
|
if w.Value != nil {
|
|
b.put(t, buckets.Placement, w.Key, w.Value)
|
|
}
|
|
}
|
|
b.put(t, buckets.Wants, "want.n1.keeper", want)
|
|
keeperKey, _ := compositions.Key("n1", "keeper", fp)
|
|
var handed string
|
|
for handed == "" {
|
|
if u := next(t, out); u.Key == keeperKey {
|
|
handed = string(u.Value)
|
|
}
|
|
}
|
|
if !strings.Contains(handed, `{\"ask\":`) {
|
|
t.Fatalf("keeper's composition does not carry the sealed value: %s", handed)
|
|
}
|
|
before := handed
|
|
b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":2,"source":"node"}`))
|
|
quiet(t, out)
|
|
b.mu.Lock()
|
|
generation++
|
|
b.mu.Unlock()
|
|
// The vault says the secret changed: the node composes again on that word, with no other input moving.
|
|
secretChanged <- []byte(`{"name":"own/n1/keeper/password","generation":2}`)
|
|
rotated := next(t, out)
|
|
if rotated.Key != keeperKey {
|
|
t.Fatalf("a new generation handed %s, want keeper again", rotated.Key)
|
|
}
|
|
// Issue 774: what the vault judged handed in its own answer is what the node's composition hands, for n1 alone.
|
|
b.mu.Lock()
|
|
asked := asks
|
|
b.mu.Unlock()
|
|
if h := (reader{c: c}).Handed(ctx, "n1"); asked == 0 || !reflect.DeepEqual(h, map[string]bool{"keeper": false}) {
|
|
t.Errorf("the vault's handed verdicts as the composition reads them, after %d asks: %v", asked, h)
|
|
}
|
|
if h := (reader{c: c}).Handed(ctx, "n2"); h != nil {
|
|
t.Errorf("another node's verdicts: %v", h)
|
|
}
|
|
|
|
// Review of L2 (round 2): rotated, then the node-engine restarts. The agent has applied nothing yet, so the box the
|
|
// node runs comes from keeper's have (Held); the vault seals the value again, other bytes, and the box the have holds
|
|
// is kept, since it opens to the same value: the composition does not move. The box from before the rotation opens
|
|
// to the old value, so the vault's is taken.
|
|
// Issue 1126, C1: a box is kept only at the slot that holds it (the same field of the same resource of the same
|
|
// module). A box another module holds (a provider's grant of the value its consumer holds as a secret, on one node)
|
|
// or another resource of keeper holds is never taken, though it opens to the same value; a have written before slots
|
|
// (a list) is reused within its own module alone.
|
|
boxOf := func(raw string) (string, string) {
|
|
c, err := compositions.Decode([]byte(raw))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sealed := compositions.SealsIn(c)
|
|
if len(sealed) != 1 {
|
|
t.Fatalf("keeper's composition carries %v", sealed)
|
|
}
|
|
for slot, box := range sealed {
|
|
return slot, box
|
|
}
|
|
return "", ""
|
|
}
|
|
slot, rotatedBox := boxOf(string(rotated.Value))
|
|
_, beforeBox := boxOf(before)
|
|
for _, row := range []struct {
|
|
name, in, held string
|
|
slots compositions.Seals
|
|
kept bool
|
|
fails bool // the have's read the bus did not answer: no composition, made again later (2026-10-28)
|
|
}{
|
|
{"the have holds the rotated value at its slot", "keeper", rotatedBox, compositions.Seals{slot: rotatedBox}, true, false},
|
|
{"the have holds the value before the rotation", "keeper", beforeBox, compositions.Seals{slot: beforeBox}, false, false},
|
|
{"a have written before slots holds the rotated value", "keeper", rotatedBox, nil, true, false},
|
|
{"another module holds the same value", "lemurs", rotatedBox, compositions.Seals{"lemurs.grant-x#sealed": rotatedBox}, false, false},
|
|
{"another resource of keeper holds the same value", "keeper", rotatedBox, compositions.Seals{"keeper.other#sealed": rotatedBox}, false, false},
|
|
{"the bus does not answer for the have", "keeper", rotatedBox, nil, false, true},
|
|
} {
|
|
restarted := env
|
|
restarted.Running = func() []compositions.Composed { return nil }
|
|
restarted.Held = func(module string) (compositions.Seals, []string, error) {
|
|
if module != row.in {
|
|
return nil, nil, nil
|
|
}
|
|
if row.fails {
|
|
return nil, nil, fmt.Errorf("the have of %s: %w", module, nats.ErrTimeout)
|
|
}
|
|
if row.slots == nil {
|
|
return nil, []string{row.held}, nil
|
|
}
|
|
return row.slots, nil, nil
|
|
}
|
|
restarted.Unseal = func(box string) ([]byte, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if g, ok := sealedAt[box]; ok {
|
|
return []byte(fmt.Sprintf("password-%d", g)), nil
|
|
}
|
|
return nil, fmt.Errorf("not sealed to n1: %s", box)
|
|
}
|
|
again := New(restarted)
|
|
c.mu.Lock()
|
|
for k, v := range c.wants {
|
|
again.wants[k] = v
|
|
}
|
|
for k, v := range c.placements {
|
|
again.placements[k] = v
|
|
}
|
|
for k, v := range c.records {
|
|
again.records[k] = v
|
|
}
|
|
for k, v := range c.modules {
|
|
again.modules[k] = v
|
|
}
|
|
c.mu.Unlock()
|
|
var composed string
|
|
for _, u := range again.compose(ctx) {
|
|
if u.Key == keeperKey {
|
|
composed = string(u.Value)
|
|
}
|
|
}
|
|
if row.fails {
|
|
if composed != "" || !again.owed.Load() {
|
|
t.Errorf("%s: composed %v and owes a pass again %v; want nothing handed and a pass owed", row.name, composed != "", again.owed.Load())
|
|
}
|
|
// Then the catalogue no longer places the node: it composes nothing of its own, and owes no pass either, so
|
|
// no timer is armed again (rev-seed-same-3, round 1).
|
|
again.mu.Lock()
|
|
clear(again.placements)
|
|
again.mu.Unlock()
|
|
if again.compose(ctx); again.owed.Load() {
|
|
t.Errorf("%s: an unplaced node still owes a pass", row.name)
|
|
}
|
|
continue
|
|
}
|
|
if composed == "" {
|
|
t.Fatalf("%s: keeper was not composed", row.name)
|
|
}
|
|
if _, box := boxOf(composed); (box == row.held) != row.kept {
|
|
t.Errorf("%s: composed with %s (held %s), kept %v", row.name, box, row.held, row.kept)
|
|
}
|
|
}
|
|
// And the node's own key (review of L2, round 3, finding 1): a core box its own composition carries is held there, and
|
|
// kept after a restart as a module's is, so the node's own composition does not move either.
|
|
ownComp := compositions.Composed{Node: "n1", Resources: []json.RawMessage{json.RawMessage(`{"id":"core-box","sealed":"{\"ask\":999}"}`)}}
|
|
b.mu.Lock()
|
|
sealedAt[`{"ask":999}`] = generation + 1
|
|
b.mu.Unlock()
|
|
coreEnv := env
|
|
coreEnv.Unseal = func(box string) ([]byte, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if g, ok := sealedAt[box]; ok {
|
|
return []byte(fmt.Sprintf("password-%d", g)), nil
|
|
}
|
|
return nil, fmt.Errorf("not sealed to n1: %s", box)
|
|
}
|
|
coreEnv.Held = func(module string) (compositions.Seals, []string, error) {
|
|
if module == agent.NodeModule {
|
|
return compositions.Seals{"core-box#sealed": rotatedBox}, nil, nil
|
|
}
|
|
return nil, nil, nil
|
|
}
|
|
if got := compositions.SealsIn(New(coreEnv).keepRunning(ownComp, nil))["core-box#sealed"]; got != rotatedBox {
|
|
t.Errorf("the node's own core box after a restart is %s, want the one its key holds %s", got, rotatedBox)
|
|
}
|
|
// Issue 1126, C1, the seed-7 probe: each entry of a file's secrets (catalogue's intoFile, the shape captured from it
|
|
// with the module a composition names) is a slot of its own: the box that runs at that entry is kept where it opens
|
|
// to the same value; one at another entry is not, though it opens to the same.
|
|
b.mu.Lock()
|
|
sealedAt["box-again"] = sealedAt[rotatedBox]
|
|
sealedAt["box-other"] = sealedAt[rotatedBox] + 100
|
|
b.mu.Unlock()
|
|
ini := func(db, key string) compositions.Composed {
|
|
return compositions.Composed{Node: "n1", Module: "grafana", Resources: []json.RawMessage{json.RawMessage(
|
|
`{"content":"[database]\npassword = ${secret:postgres-database}\n[security]\nsecret_key = ${secret:secret-key}\n",` +
|
|
`"id":"grafana.ini","mode":"0600","module":"grafana","path":"/etc/grafana/grafana.ini","secrets":{"postgres-database":` +
|
|
strconv.Quote(db) + `,"secret-key":` + strconv.Quote(key) + `},"type":"file"}`)}}
|
|
}
|
|
for _, row := range []struct {
|
|
name string
|
|
held compositions.Seals
|
|
want string
|
|
}{
|
|
{"the box that runs at its entry", compositions.Seals{"grafana.ini#secrets.postgres-database": rotatedBox}, rotatedBox},
|
|
{"a box at another entry", compositions.Seals{"grafana.ini#secrets.secret-key": rotatedBox}, "box-again"},
|
|
} {
|
|
secretsEnv := coreEnv
|
|
secretsEnv.Held = func(module string) (compositions.Seals, []string, error) {
|
|
if module == "grafana" {
|
|
return row.held, nil, nil
|
|
}
|
|
return nil, nil, nil
|
|
}
|
|
got := compositions.SealsIn(New(secretsEnv).keepRunning(ini("box-again", "box-other"), nil))
|
|
if got["grafana.ini#secrets.postgres-database"] != row.want || got["grafana.ini#secrets.secret-key"] != "box-other" {
|
|
t.Errorf("%s: the secrets kept are %v", row.name, got)
|
|
}
|
|
}
|
|
|
|
// The pair credentials asked (ADR 0550 §5): the keys go to the vault's own with their role, a consumer's comes back as
|
|
// a credential held, a provider's as the grant's copy, each sealed to n1's key, the value first had kept.
|
|
var sent map[string]any
|
|
pairEnv := b.env()
|
|
pairEnv.Own = func(_ context.Context, args map[string]any) (json.RawMessage, error) {
|
|
sent = args
|
|
return json.RawMessage(`{"node":"n1","key":"sk","credentials":[` +
|
|
`{"role":"consumer","provision":"pg","consumer":"n1","consumer_module":"web","provider":"n2","name":"pair/n2/pg/n1/web","generation":1,"sealed":{"c":1}},` +
|
|
`{"role":"provider","provision":"s3","consumer":"n3","consumer_module":"files","provider":"n1","name":"provision/n1/s3","generation":2,"sealed":{"p":1}}]}`), nil
|
|
}
|
|
// The node runs the consumer's credential already, sealed otherwise (the controller sealed it, or the vault at
|
|
// another ask): that box opens to the same value, so it is kept, and the composition does not move.
|
|
pairEnv.Running = func() []compositions.Composed {
|
|
return []compositions.Composed{{Module: "web", Resources: []json.RawMessage{json.RawMessage(`{"id":"web.secret-pg","sealed":"{\"run\":1}"}`)}}}
|
|
}
|
|
opened := map[string]string{`{"c":1}`: "pg-pw", `{"run":1}`: "pg-pw", `{"p":1}`: "s3-pw"}
|
|
pairEnv.Unseal = func(box string) ([]byte, error) {
|
|
if v, ok := opened[box]; ok {
|
|
return []byte(v), nil
|
|
}
|
|
return nil, fmt.Errorf("not sealed to n1: %s", box)
|
|
}
|
|
var refusals []string
|
|
pairEnv.Refused = func(why string, held map[string]string) { refusals = append(refusals, why) }
|
|
pc := New(pairEnv)
|
|
consumer := []compose.CredentialKey{{Provision: "pg", Consumer: "n1", ConsumerModule: "web", Provider: "n2"}}
|
|
provider := []compose.CredentialKey{{Provision: "s3", Consumer: "n3", ConsumerModule: "files", Provider: "n1"}}
|
|
creds, issued, err := reader{c: pc}.Credentials(ctx, "n1", consumer, provider)
|
|
if err != nil || len(creds) != 1 || len(issued) != 1 {
|
|
t.Fatalf("credentials %+v issued %+v, %v", creds, issued, err)
|
|
}
|
|
if creds[0].State != compose.CredentialHeld || creds[0].Provider != "n2" || creds[0].Sealed != `{"c":1}` ||
|
|
issued[0].Consumer != "n3" || issued[0].ForProvider != `{"p":1}` {
|
|
t.Errorf("credentials %+v issued %+v", creds, issued)
|
|
}
|
|
// What runs at the credential's slot is kept, after the composition (keepRunning, issue 1126, C1).
|
|
webComp := compositions.Composed{Node: "n1", Module: "web", Resources: []json.RawMessage{json.RawMessage(`{"id":"web.secret-pg","sealed":"{\"c\":1}"}`)}}
|
|
if got := compositions.SealsIn(pc.keepRunning(webComp, pairEnv.Running()))["web.secret-pg#sealed"]; got != `{"run":1}` {
|
|
t.Errorf("web's credential composed as %s, want the box that runs", got)
|
|
}
|
|
// A node that cannot compose tells the agent once, and once again when it composes (the agent's haves say it).
|
|
pc.refused("a stand-in for web's password", nil)
|
|
pc.refused("a stand-in for web's password", nil)
|
|
pc.refused("", nil)
|
|
if len(refusals) != 2 || refusals[0] == "" || refusals[1] != "" {
|
|
t.Errorf("the agent was told %q", refusals)
|
|
}
|
|
if asked, _ := json.Marshal(sent["credentials"]); !strings.Contains(string(asked), `"role":"consumer"`) || !strings.Contains(string(asked), `"role":"provider"`) {
|
|
t.Errorf("the vault was asked %s", asked)
|
|
}
|
|
// ADR 0559 §4: the node answers placement candidates in shadow, placed or not, applying nothing and handing no
|
|
// composition: the same; a module the group adds, its secret not yet issued; a module removed; a placement it cannot
|
|
// compose; a group whose keys went is no longer answered. What runs is never composed from a candidate: the
|
|
// Composer does not read the state.
|
|
for _, in := range New(env).inputs() {
|
|
if in.bucket == buckets.PlacementCandidates {
|
|
t.Fatal("the node's own composer reads placement candidates")
|
|
}
|
|
}
|
|
sb := newPatternBus()
|
|
trunk := meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}}
|
|
mods := map[string]meshcatalog.ModulesByModuleValue{"lemurs": {Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"},
|
|
"keeper": {Fp2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"}}
|
|
readOnly := &fakeBus{kv: map[string][]byte{buckets.Manifests + "/" + key: man, buckets.Manifests + "/" + kkey: kman}}
|
|
senv := readOnly.env()
|
|
senv.Watch = sb.watch
|
|
var ownFails atomic.Bool // the vault's own on a bus that does not answer (2026-10-28, ace's answer to seed-7)
|
|
senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
|
|
if ownFails.Load() {
|
|
return nil, fmt.Errorf("the vault's own could not be asked: %w", nats.ErrNoResponders)
|
|
}
|
|
return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here
|
|
}
|
|
// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
|
|
running, _, _, err := frombus.Compose(ctx, fixedReader{placements: map[string]meshmoduleassigner.Placement{"n1": trunk},
|
|
records: map[string]nodeengine.NodeRecord{"n1": {Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}},
|
|
modules: mods, manifests: map[string][]byte{digest: []byte(lemurs), kdigest: []byte(keeper)}}, frombus.Mesh{}, frombus.Scope{Node: "n1"}, map[string]string{"lemurs": fp})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
senv.Running = func() []compositions.Composed {
|
|
var out []compositions.Composed
|
|
for _, c := range running {
|
|
out = append(out, c)
|
|
}
|
|
return out
|
|
}
|
|
// The controller's compositions at n1's wants, in n1's composition bucket: what a head deleting its placement
|
|
// returns it to (the seed's revert); here the ones it runs.
|
|
composedBucket, _ := compositions.BucketFor("n1")
|
|
for key, c := range running {
|
|
raw, err := compositions.Encode(c)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
readOnly.mu.Lock()
|
|
readOnly.kv[composedBucket+"/"+key] = raw
|
|
readOnly.mu.Unlock()
|
|
}
|
|
answered := make(chan map[string]nodeengine.CandidateAnswer, 8)
|
|
sh := NewShadow(senv, func(_ context.Context, a map[string]nodeengine.CandidateAnswer) { answered <- a })
|
|
sctx, scancel := context.WithCancel(ctx)
|
|
defer scancel()
|
|
// Review of L2b, round 1, finding 2: the candidates' state is not there when the node starts (a delivery before the
|
|
// indexer made it): the shadow tries again, and answers once it appears.
|
|
sb.missing(buckets.PlacementCandidates, true)
|
|
kept := make(chan struct{})
|
|
go func() { sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond }); close(kept) }()
|
|
time.Sleep(30 * time.Millisecond)
|
|
sb.missing(buckets.PlacementCandidates, false)
|
|
sb.made(6)
|
|
rec, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
|
tp, _ := meshmoduleassignerclient.EncodePlacement(trunk)
|
|
sb.put(buckets.Placement, "placement.n1", tp, 1)
|
|
sb.put(buckets.Have, "n1.node", rec, 1)
|
|
for name, m := range mods {
|
|
raw, _ := meshcatalogclient.EncodeModule(m)
|
|
sb.put(buckets.Modules, name, raw, 1)
|
|
}
|
|
wantLemurs := meshwants.WantsByWantNodeModuleKey("n1", "lemurs")
|
|
var lemursDigest string
|
|
for _, c := range running {
|
|
if c.Module == "lemurs" {
|
|
lemursDigest, _ = compositions.DigestOf(c)
|
|
}
|
|
}
|
|
wantRaw, err := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Composed: lemursDigest})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sb.put(buckets.Wants, wantLemurs, wantRaw, 1)
|
|
wantNode := meshwants.WantsByWantNodeNodeKey("n1")
|
|
nodeRaw, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{})
|
|
sb.put(buckets.Wants, wantNode, nodeRaw, 1)
|
|
candidate := func(group string, rev uint64, modules ...string) {
|
|
p := trunk
|
|
p.Modules = nil
|
|
for _, m := range modules {
|
|
p.Modules = append(p.Modules, meshmoduleassigner.PlacedModule{Module: m})
|
|
}
|
|
raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: group, Placement: &p})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
k := meshmoduleassigner.CandidatesByProposalNodeKey(group, "n1")
|
|
sb.put(buckets.PlacementCandidates, k, raw, rev)
|
|
}
|
|
candidate("same", 3, "lemurs")
|
|
candidate("adds", 4, "lemurs", "keeper")
|
|
candidate("drops", 5)
|
|
candidate("ghost", 6, "lemurs", "ghost")
|
|
// A head deleting n1's placement file (the seed's revert, mesh-catalog #526): n1 returns to the controller's
|
|
// compositions, answered as such, not as every module removed.
|
|
if raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "deletes"}); err != nil {
|
|
t.Fatal(err)
|
|
} else {
|
|
k := meshmoduleassigner.CandidatesByProposalNodeKey("deletes", "n1")
|
|
sb.put(buckets.PlacementCandidates, k, raw, 7)
|
|
}
|
|
// A proposal that changes another node's key and reaches n1's composition (ADR 0567 §3): n1's candidate holds no
|
|
// placement of its own change (Reached, nil), and n1 answers against its current placement, never as one whose key
|
|
// the proposal takes away.
|
|
if raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "reach", Reached: true}); err != nil {
|
|
t.Fatal(err)
|
|
} else {
|
|
k := meshmoduleassigner.CandidatesByProposalNodeKey("reach", "n1")
|
|
sb.put(buckets.PlacementCandidates, k, raw, 8)
|
|
}
|
|
sb.caught()
|
|
var answers map[string]nodeengine.CandidateAnswer
|
|
waitFor := func(why string, ok func(map[string]nodeengine.CandidateAnswer) bool) {
|
|
t.Helper()
|
|
deadline := time.After(10 * time.Second)
|
|
for answers == nil || !ok(answers) {
|
|
select {
|
|
case answers = <-answered:
|
|
case <-deadline:
|
|
t.Fatalf("%s: last answers %+v", why, answers)
|
|
}
|
|
}
|
|
}
|
|
waitFor("every group answered", func(a map[string]nodeengine.CandidateAnswer) bool { return len(a) == 6 })
|
|
// 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(nodeengine.CandidateAnswer) bool{
|
|
"same": func(a nodeengine.CandidateAnswer) bool {
|
|
return a.Verdict == nodeengine.CandidateAnswerVerdictSame && a.Proposal == "same" && a.Revision == 3 && len(a.Same) == 2
|
|
},
|
|
"adds": func(a nodeengine.CandidateAnswer) bool {
|
|
return a.Verdict == nodeengine.CandidateAnswerVerdictDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" &&
|
|
len(a.Added[0].Unissued) == 1 && len(a.Refusals) == 0
|
|
},
|
|
"drops": func(a nodeengine.CandidateAnswer) bool {
|
|
return a.Verdict == nodeengine.CandidateAnswerVerdictDiffers && len(a.Removed) == 1 && a.Removed[0] == "lemurs"
|
|
},
|
|
"ghost": func(a nodeengine.CandidateAnswer) bool {
|
|
return a.Verdict == nodeengine.CandidateAnswerVerdictRefused && a.Refused != ""
|
|
},
|
|
"reach": func(a nodeengine.CandidateAnswer) bool {
|
|
return a.Verdict == nodeengine.CandidateAnswerVerdictSame && a.Returns == "" && a.Revision == 8 && len(a.Removed) == 0 &&
|
|
fmt.Sprint(a.Same) == "[lemurs node]"
|
|
},
|
|
// Issue 1126, A7: a candidate deleting the node's placement is refused, never answered as a return to the
|
|
// controller's compositions: no composition made elsewhere is applied, so the node would compose nothing.
|
|
"deletes": func(a nodeengine.CandidateAnswer) bool {
|
|
return a.Verdict == nodeengine.CandidateAnswerVerdictRefused && a.Returns == "" && a.Revision == 7 &&
|
|
strings.Contains(a.Refused, "composes nothing")
|
|
},
|
|
} {
|
|
if !want(answers[group]) {
|
|
t.Errorf("%s: %+v", group, answers[group])
|
|
}
|
|
}
|
|
// The node's own key changing (its answers written) composes nothing again; a group's key deleted drops its answer.
|
|
sb.put(buckets.Have, "n1.node", rec, 2)
|
|
select {
|
|
case a := <-answered:
|
|
t.Errorf("answers handed again for nothing changed: %+v", a)
|
|
case <-time.After(100 * time.Millisecond):
|
|
}
|
|
// Issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s): a candidate revision is
|
|
// answered once. A module's have whose health or words alone move (each look rewrites them), and a placement, a
|
|
// module or a config value written again with the value it held, answer nothing again; a new revision answers its
|
|
// group alone, and what runs moving answers every group once.
|
|
answering := func() int {
|
|
readOnly.mu.Lock()
|
|
defer readOnly.mu.Unlock()
|
|
n := 0
|
|
for _, l := range readOnly.said {
|
|
if strings.HasPrefix(l, "answering placement candidate ") {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
drain := func() {
|
|
for {
|
|
select {
|
|
case answers = <-answered:
|
|
case <-time.After(150 * time.Millisecond):
|
|
return
|
|
}
|
|
}
|
|
}
|
|
lemursHave := func(detail string, rev uint64) {
|
|
raw, err := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: lemursDigest, Phase: nodeengineclient.PhaseRunning,
|
|
Health: nodeengineclient.HealthFailed, Detail: detail})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
hk, _ := nodeengineclient.HaveKey("n1", "lemurs")
|
|
sb.put(buckets.Have, hk, raw, rev)
|
|
}
|
|
lemursHave("its tool lemurs_health: no link to the bus is open", 10)
|
|
drain()
|
|
settled := answering()
|
|
lemursHave("its tool lemurs_health: no answer within 10s", 11)
|
|
lemursHave("its tool lemurs_health: no answer within 20s", 12)
|
|
sb.put(buckets.Placement, "placement.n1", tp, 2)
|
|
lraw, _ := meshcatalogclient.EncodeModule(mods["lemurs"])
|
|
sb.put(buckets.Modules, "lemurs", lraw, 2)
|
|
drain()
|
|
if got := answering(); got != settled {
|
|
t.Errorf("answered %d more time(s) with no candidate revision and no input of the node changed", got-settled)
|
|
}
|
|
candidate("same", 8, "lemurs")
|
|
waitFor("the new revision answered", func(a map[string]nodeengine.CandidateAnswer) bool { return a["same"].Revision == 8 })
|
|
drain()
|
|
if got := answering(); got != settled+1 {
|
|
t.Errorf("a new revision of one group answered %d group(s), want 1", got-settled)
|
|
}
|
|
// Issue 1126, 2026-10-28 (seed-9's probe): each answer says which input woke it, so a loop is told from inputs moving.
|
|
lastSaid := func(prefix string) string {
|
|
readOnly.mu.Lock()
|
|
defer readOnly.mu.Unlock()
|
|
for i := len(readOnly.said) - 1; i >= 0; i-- {
|
|
if strings.HasPrefix(readOnly.said[i], prefix) {
|
|
return readOnly.said[i]
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
if got := lastSaid("answering placement candidate same "); got != "answering placement candidate same (revision 8; woken by the candidate)" {
|
|
t.Errorf("a new revision's answer said %q", got)
|
|
}
|
|
settled = answering()
|
|
hk, _ := nodeengineclient.HaveKey("n1", "lemurs")
|
|
moved, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Composed: "sha256:" + strings.Repeat("e", 64), Phase: nodeengineclient.PhaseRunning})
|
|
sb.put(buckets.Have, hk, moved, 13)
|
|
drain()
|
|
if got := answering(); got != settled+len(answers) {
|
|
t.Errorf("what runs moved, and %d answer(s) were composed again for %d group(s)", got-settled, len(answers))
|
|
}
|
|
if got := lastSaid("answering placement candidate drops "); got != "answering placement candidate drops (revision 5; woken by haves)" {
|
|
t.Errorf("an answer woken by what runs moving said %q", got)
|
|
}
|
|
// A read the bus did not answer is no answer to keep: the group is answered again after a bound, though nothing moved.
|
|
setFor(t, &laterAgain, []time.Duration{20 * time.Millisecond})
|
|
ownFails.Store(true)
|
|
candidate("adds", 9, "lemurs", "keeper")
|
|
waitFor("the answer met the vault not answering", func(a map[string]nodeengine.CandidateAnswer) bool {
|
|
return a["adds"].Revision == 9 && strings.Contains(a["adds"].Refused, "no responders")
|
|
})
|
|
ownFails.Store(false)
|
|
waitFor("answered again once the bus answers, with nothing else moved", func(a map[string]nodeengine.CandidateAnswer) bool {
|
|
return a["adds"].Revision == 9 && a["adds"].Verdict == nodeengine.CandidateAnswerVerdictDiffers
|
|
})
|
|
drain()
|
|
if got := lastSaid("answering placement candidate adds "); got != "answering placement candidate adds (revision 9; woken by a retry after a read the bus did not answer)" {
|
|
t.Errorf("an answer composed again after the bus did not answer said %q", got)
|
|
}
|
|
k := meshmoduleassigner.CandidatesByProposalNodeKey("same", "n1")
|
|
sb.del(buckets.PlacementCandidates, k)
|
|
waitFor("the deleted group's answer stays", func(a map[string]nodeengine.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 5 })
|
|
|
|
// Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches
|
|
// again. The new run reads afresh, so that group is no longer answered.
|
|
sb.missing(buckets.PlacementCandidates, true)
|
|
sb.end(buckets.Modules)
|
|
dropsKey := meshmoduleassigner.CandidatesByProposalNodeKey("drops", "n1")
|
|
sb.del(buckets.PlacementCandidates, dropsKey)
|
|
time.Sleep(30 * time.Millisecond)
|
|
sb.missing(buckets.PlacementCandidates, false)
|
|
waitFor("a group deleted while the shadow was down is still answered", func(a map[string]nodeengine.CandidateAnswer) bool {
|
|
_, ok := a["drops"]
|
|
return !ok && len(a) == 4
|
|
})
|
|
|
|
// Issue 1126, A7 AC1: a placement that cannot be read composes nothing and says why; the node never applies a
|
|
// composition made elsewhere instead. (A placement read without the node's key composes nothing either: above.)
|
|
{
|
|
refused := newFakeBus()
|
|
env := refused.env()
|
|
watch := env.Watch
|
|
env.Watch = func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) {
|
|
if bucket == buckets.Placement {
|
|
return nil, fmt.Errorf("nats: Permissions Violation for Subscription to %s", bucket)
|
|
}
|
|
return watch(ctx, bucket, pattern)
|
|
}
|
|
out, err := New(env).Watch(ctx, "")
|
|
if out != nil || err == nil || !strings.Contains(err.Error(), buckets.Placement) || !strings.Contains(err.Error(), "cannot be watched") {
|
|
t.Errorf("an unreadable placement: composed %v, said %v", out != nil, err)
|
|
}
|
|
}
|
|
|
|
// Issue 1126, the shadow shows its work: one input not read after the wait is said, naming it, in the journal and in
|
|
// a refused answer to each group known; once read, the group is answered for real.
|
|
setFor(t, ¬ReadyAfter, 30*time.Millisecond)
|
|
nb := newPatternBus()
|
|
nenv := readOnly.env()
|
|
nenv.Watch = nb.watch
|
|
nenv.Running = senv.Running
|
|
var nsaid []string
|
|
var nmu sync.Mutex
|
|
nenv.Say = func(l string) { nmu.Lock(); nsaid = append(nsaid, l); nmu.Unlock() }
|
|
nanswered := make(chan map[string]nodeengine.CandidateAnswer, 8)
|
|
nsh := NewShadow(nenv, func(_ context.Context, a map[string]nodeengine.CandidateAnswer) { nanswered <- a })
|
|
nctx, ncancel := context.WithCancel(ctx)
|
|
defer ncancel()
|
|
go func() { _ = nsh.Run(nctx) }()
|
|
nb.made(6)
|
|
nb.put(buckets.Placement, "placement.n1", tp, 1)
|
|
nb.put(buckets.Have, "n1.node", rec, 1)
|
|
for name, m := range mods {
|
|
raw, _ := meshcatalogclient.EncodeModule(m)
|
|
nb.put(buckets.Modules, name, raw, 1)
|
|
}
|
|
sameRaw, _ := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "same", Placement: &trunk})
|
|
sameKey := meshmoduleassigner.CandidatesByProposalNodeKey("same", "n1")
|
|
nb.put(buckets.PlacementCandidates, sameKey, sameRaw, 3)
|
|
// Issue 1126, T3: an older group (a closed pull request's candidate, rev 2) is answered after the newest.
|
|
oldRaw, _ := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "older", Placement: &trunk})
|
|
oldKey := meshmoduleassigner.CandidatesByProposalNodeKey("older", "n1")
|
|
nb.put(buckets.PlacementCandidates, oldKey, oldRaw, 2)
|
|
for _, bucket := range []string{buckets.PlacementCandidates, buckets.Placement, buckets.Have, buckets.Modules, builds.BuilderBuildsBucket} {
|
|
nb.send(bucket, agent.Update{Caught: true})
|
|
}
|
|
select {
|
|
case a := <-nanswered:
|
|
if g := a["same"]; g.Verdict != nodeengine.CandidateAnswerVerdictRefused || !strings.Contains(g.Refused, ResolvedBucket) || g.Proposal != "same" {
|
|
t.Errorf("a shadow waiting for its values answered %+v", g)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("a shadow not ready after its wait answered nothing")
|
|
}
|
|
nmu.Lock()
|
|
if !strings.Contains(strings.Join(nsaid, "\n"), "this node answers no placement candidate yet: the node has not read "+ResolvedBucket) {
|
|
t.Errorf("a shadow not ready said %q", nsaid)
|
|
}
|
|
nmu.Unlock()
|
|
// The first shadow ends before slowAfter is set: it reads it while it answers (a race the race detector names).
|
|
scancel()
|
|
<-kept
|
|
setFor(t, &slowAfter, time.Nanosecond) // every answer is slow: said, and noted in words that do not move with the time
|
|
nb.send(ResolvedBucket, agent.Update{Caught: true})
|
|
select {
|
|
case a := <-nanswered:
|
|
if g := a["same"]; g.Verdict != nodeengine.CandidateAnswerVerdictSame || !strings.Contains(g.Note, "this answer took over") {
|
|
t.Errorf("the shadow, once ready, answered %+v", g)
|
|
}
|
|
// The newest first, handed at once: the older group still holds the answer it had.
|
|
if g := a["older"]; g.Verdict != nodeengine.CandidateAnswerVerdictRefused {
|
|
t.Errorf("the older group was answered before the newest was handed: %+v", g)
|
|
}
|
|
if g := (<-nanswered)["older"]; g.Verdict != nodeengine.CandidateAnswerVerdictSame {
|
|
t.Errorf("the older group, answered next: %+v", g)
|
|
}
|
|
// Issue 1126, T2: an answer says when it starts and when it ends (and, every minute between, what it waits on).
|
|
nmu.Lock()
|
|
said := strings.Join(nsaid, "\n")
|
|
for _, line := range []string{"the answer to same took", "answering placement candidate same (revision 3; woken by its first answer since the shadow began)",
|
|
"answered placement candidate same in", "answering placement candidate older (revision 2;"} {
|
|
if !strings.Contains(said, line) {
|
|
t.Errorf("%q was not said: %q", line, nsaid)
|
|
}
|
|
}
|
|
if strings.Index(said, "answering placement candidate same") > strings.Index(said, "answering placement candidate older") {
|
|
t.Errorf("the older group was answered first: %q", nsaid)
|
|
}
|
|
nmu.Unlock()
|
|
done := nsh.c.waiting("the vault's own secrets")
|
|
if got := nsh.c.waitingOn(); got != "the vault's own secrets" {
|
|
t.Errorf("a running answer would say it waits on %q", got)
|
|
}
|
|
done()
|
|
case <-time.After(10 * time.Second):
|
|
t.Fatal("the shadow, once ready, answered nothing")
|
|
}
|
|
|
|
// Issue 1126, probe #532 on shanks: what runs is the composition the have names, never the copy of the declaration
|
|
// kept before the switch (no fp, no digest): a module composed at the digest its have says runs is the same.
|
|
{
|
|
probe := nodeengine.CandidateAnswer{Proposal: "pro", Revision: 68, Verdict: nodeengine.CandidateAnswerVerdictDiffers, Same: []string{"adwaita"},
|
|
Differs: []nodeengine.CandidateDiffers{
|
|
{Module: "xterm", Composed: "sha256:3c0a31df64283c0678b1cdbeff8ac030609cab65cc7e47faaf714da743d9340e", Why: "the fingerprint differs"},
|
|
{Module: "zsh", Composed: "sha256:6093f31056d39eaf604410b7289ec6b5fe602b19c8547283562fcfc078c4844f", Resource: "zsh.bundle-tools", Why: "the resource differs"},
|
|
{Module: "brother", Composed: "sha256:0702b13991288da72d230460624e949d3c1d7d833eac2ec2134073c82e5a61cf",
|
|
Running: "sha256:c11c14af2738e3fcdb04da4004bbe84728bdefaf8c986d76601ffbe502435871", Resource: "brother.needs-broker-device", Why: "the resource differs"}}}
|
|
got := asHavesSay(probe, map[string]string{
|
|
"xterm": "sha256:3c0a31df64283c0678b1cdbeff8ac030609cab65cc7e47faaf714da743d9340e",
|
|
"zsh": "sha256:6093f31056d39eaf604410b7289ec6b5fe602b19c8547283562fcfc078c4844f",
|
|
"brother": "sha256:c11c14af2738e3fcdb04da4004bbe84728bdefaf8c986d76601ffbe502435871"})
|
|
if fmt.Sprint(got.Same) != "[adwaita xterm zsh]" || len(got.Differs) != 1 || got.Differs[0].Module != "brother" || got.Verdict != nodeengine.CandidateAnswerVerdictDiffers {
|
|
t.Errorf("shanks' probe answered %+v", got)
|
|
}
|
|
if one := asHavesSay(nodeengine.CandidateAnswer{Proposal: "pro", Revision: 1, Verdict: nodeengine.CandidateAnswerVerdictDiffers, Differs: probe.Differs[:1]},
|
|
map[string]string{"xterm": probe.Differs[0].Composed}); one.Verdict != nodeengine.CandidateAnswerVerdictSame {
|
|
t.Errorf("a module the same as its have says is answered %+v", one)
|
|
}
|
|
}
|
|
|
|
// Issue 1126, the seed's review: a module n1 runs inside its own composition (nats, at a build of the old form) is
|
|
// composed there again, not at the trunk's build as one added; lemurs at the build it runs, keeper (added) at the
|
|
// trunk's.
|
|
inOwnRunning := []compositions.Composed{{Node: "n1", Module: "lemurs", FP: fp}}
|
|
other := "src2:" + strings.Repeat("9", 64)
|
|
gotFps := shadowBuilds(inOwnRunning, meshmoduleassigner.Placement{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "nats"}, {Module: "keeper"}}},
|
|
map[string]meshcatalog.ModulesByModuleValue{"lemurs": {Fp2: other}, "nats": {Fp2: other}, "keeper": {Fp2: other}}, []string{"nats"})
|
|
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, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: digest})
|
|
buildKey, _ := builds.BuilderBuildKey("lemurs", fp)
|
|
b.mu.Lock()
|
|
b.kv[builds.BuilderBuildsBucket+"/"+buildKey] = built
|
|
b.kv[ResolvedBucket+"/n2.mailer.domain"] = []byte(`{"value":"example.org","source":"module"}`)
|
|
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 := nodegrant.For("n1", "n2.mailer.domain")
|
|
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()
|
|
|
|
// L3 s7 (ADR 0550 §2): the node asks the vault for its own certificate only when a module placed here asks for one
|
|
// and its record holds none from an authority it trusts; once per serving key.
|
|
certBus := newFakeBus()
|
|
certAsks := 0
|
|
certEnv := certBus.env()
|
|
certEnv.Certify = func(context.Context) error { certAsks++; return nil }
|
|
certEnv.VaultAuthority = func() string { return "" }
|
|
cc := New(certEnv)
|
|
withCert := `{"module":"web","version":"1","certificate":{"into":"/etc/web/tls.crt"}}`
|
|
cdigest := meshcatalogclient.ManifestDigest([]byte(withCert))
|
|
cc.manifests[digest] = []byte(lemurs)
|
|
cc.manifests[cdigest] = []byte(withCert)
|
|
cc.modules = map[string]meshcatalog.ModulesByModuleValue{"lemurs": {Manifest: digest}, "web": {Manifest: cdigest}}
|
|
cc.records["n1"] = nodeengine.NodeRecord{Node: "n1", ServingKey: base64.StdEncoding.EncodeToString(make([]byte, 32))}
|
|
onlyLemurs := meshmoduleassigner.Placement{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}}
|
|
cc.certifyIfWanted(context.Background(), reader{c: cc}, onlyLemurs)
|
|
if certAsks != 0 {
|
|
t.Errorf("no module asks for a certificate, and the vault was asked %d times", certAsks)
|
|
}
|
|
withWeb := meshmoduleassigner.Placement{Node: "n1", Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "web"}}}
|
|
cc.certifyIfWanted(context.Background(), reader{c: cc}, withWeb)
|
|
cc.certifyIfWanted(context.Background(), reader{c: cc}, withWeb)
|
|
if certAsks != 1 {
|
|
t.Errorf("a module asks for a certificate: the vault was asked %d times, want once", certAsks)
|
|
}
|
|
}
|
|
|
|
// patternBus is a bus whose every watch gets the updates of its bucket matching its pattern, as the bus's own watches
|
|
// do: two watches of one bucket each see what matches theirs.
|
|
type patternBus struct {
|
|
mu sync.Mutex
|
|
watches []patternWatch
|
|
absent map[string]bool
|
|
// kv is each bucket's keys now; once replay is set, a new watch gets those matching it and then Caught, as a watch
|
|
// of the bus's own does.
|
|
kv map[string]map[string]agent.Update
|
|
replay bool
|
|
}
|
|
|
|
// end ends every watch of a bucket, as a watch the bus drops, and from then on each new watch replays what is there.
|
|
func (b *patternBus) end(bucket string) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.replay = true
|
|
kept := b.watches[:0]
|
|
for _, w := range b.watches {
|
|
if w.bucket == bucket {
|
|
close(w.ch)
|
|
continue
|
|
}
|
|
kept = append(kept, w)
|
|
}
|
|
b.watches = kept
|
|
}
|
|
|
|
// missing makes a bucket's watch fail, as one the bus holds no state for, or work again.
|
|
func (b *patternBus) missing(bucket string, gone bool) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if b.absent == nil {
|
|
b.absent = map[string]bool{}
|
|
}
|
|
b.absent[bucket] = gone
|
|
}
|
|
|
|
type patternWatch struct {
|
|
bucket, pattern string
|
|
ch chan agent.Update
|
|
}
|
|
|
|
func newPatternBus() *patternBus { return &patternBus{} }
|
|
|
|
func (b *patternBus) watch(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if b.absent[bucket] {
|
|
return nil, fmt.Errorf("nats: bucket not found")
|
|
}
|
|
w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)}
|
|
b.watches = append(b.watches, w)
|
|
if b.replay {
|
|
for key, u := range b.kv[bucket] {
|
|
if matches(pattern, key) {
|
|
w.ch <- u
|
|
}
|
|
}
|
|
w.ch <- agent.Update{Caught: true}
|
|
}
|
|
return w.ch, nil
|
|
}
|
|
|
|
func matches(pattern, key string) bool {
|
|
p, k := strings.Split(pattern, "."), strings.Split(key, ".")
|
|
for i, tok := range p {
|
|
if tok == ">" {
|
|
return len(k) > i
|
|
}
|
|
if i >= len(k) || (tok != "*" && tok != k[i]) {
|
|
return false
|
|
}
|
|
}
|
|
return len(p) == len(k)
|
|
}
|
|
|
|
func (b *patternBus) send(bucket string, u agent.Update) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if !u.Caught {
|
|
if b.kv == nil {
|
|
b.kv = map[string]map[string]agent.Update{}
|
|
}
|
|
if b.kv[bucket] == nil {
|
|
b.kv[bucket] = map[string]agent.Update{}
|
|
}
|
|
if u.Deleted {
|
|
delete(b.kv[bucket], u.Key)
|
|
} else {
|
|
b.kv[bucket][u.Key] = u
|
|
}
|
|
}
|
|
for _, w := range b.watches {
|
|
if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) {
|
|
select {
|
|
case w.ch <- u:
|
|
default: // a watch of a run that ended reads no more
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (b *patternBus) put(bucket, key string, value []byte, rev uint64) {
|
|
b.send(bucket, agent.Update{Key: key, Value: value, Revision: rev})
|
|
}
|
|
|
|
func (b *patternBus) del(bucket, key string) { b.send(bucket, agent.Update{Key: key, Deleted: true}) }
|
|
|
|
// made waits until n watches are made, so nothing put is lost.
|
|
func (b *patternBus) made(n int) {
|
|
for {
|
|
b.mu.Lock()
|
|
got := len(b.watches)
|
|
b.mu.Unlock()
|
|
if got >= n {
|
|
return
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// caught says every watch caught up.
|
|
func (b *patternBus) caught() {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
for _, w := range b.watches {
|
|
w.ch <- agent.Update{Caught: true}
|
|
}
|
|
}
|
|
|
|
// fixedReader is frombus's Reader over values held, for composing what a node runs.
|
|
type fixedReader struct {
|
|
placements map[string]meshmoduleassigner.Placement
|
|
records map[string]nodeengine.NodeRecord
|
|
modules map[string]meshcatalog.ModulesByModuleValue
|
|
manifests map[string][]byte
|
|
}
|
|
|
|
func (r fixedReader) Placements(context.Context) (map[string]meshmoduleassigner.Placement, error) {
|
|
return r.placements, nil
|
|
}
|
|
func (r fixedReader) Records(context.Context) (map[string]nodeengine.NodeRecord, error) {
|
|
return r.records, nil
|
|
}
|
|
func (r fixedReader) Modules(context.Context) (map[string]meshcatalog.ModulesByModuleValue, error) {
|
|
return r.modules, nil
|
|
}
|
|
func (r fixedReader) Manifest(_ context.Context, d string) ([]byte, error) {
|
|
if raw, ok := r.manifests[d]; ok {
|
|
return raw, nil
|
|
}
|
|
return nil, fmt.Errorf("no manifest %s", d)
|
|
}
|
|
func (r fixedReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil }
|
|
func (r fixedReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
|
|
return nil, nil
|
|
}
|
|
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) (*builds.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 nodeengine.HaveByNodeModuleValue
|
|
in bool
|
|
// heldToday says the shadow counts the module held today (issue 1126: a seed holding it too holds it as today).
|
|
heldToday bool
|
|
}{
|
|
{"failed, naming nothing it runs", nodeengine.HaveByNodeModuleValue{Phase: nodeengineclient.PhaseFailed, Health: nodeengineclient.HealthFailed,
|
|
Detail: "the node cannot compose it, and keeps running what it ran: no build"}, false, true},
|
|
{"failed at a build fingerprint", nodeengine.HaveByNodeModuleValue{Fp: "build-0123abcd", Composed: "sha256:" + strings.Repeat("c", 64), Phase: nodeengineclient.PhaseFailed,
|
|
Health: nodeengineclient.HealthFailed, Detail: "the node cannot compose it, and keeps running what it ran: no build"}, true, true},
|
|
// The live have of novox's de-spiegel (2026-10-28): a build of the old form, named by `build`, and no `fp`.
|
|
{"running a build of the old form", nodeengine.HaveByNodeModuleValue{Build: "build-1790368817085312877", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}, true, true},
|
|
{"running a build of the new form", nodeengine.HaveByNodeModuleValue{Fp: meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64), Composed: "sha256:" + strings.Repeat("c", 64),
|
|
Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}, false, false},
|
|
{"running with no fingerprint (mesh-wireguard)", nodeengine.HaveByNodeModuleValue{Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}, true, false},
|
|
{"held today", nodeengine.HaveByNodeModuleValue{Fp: "build-0123abcd", Composed: "sha256:" + strings.Repeat("c", 64), Phase: nodeengineclient.PhaseHeld,
|
|
Health: nodeengineclient.HealthWaits, Detail: "held: a required config value is unset"}, true, true},
|
|
// novox's postgres (2026-10-31), its have written before mesh-host #189: held by a backup the node agent's stop
|
|
// interrupted and naming nothing; counted in, it was composed from its manifest and refused for its secret reader.
|
|
{"held by an interrupted backup, naming nothing", nodeengine.HaveByNodeModuleValue{Phase: nodeengineclient.PhaseHeld, Health: nodeengineclient.HealthWaits,
|
|
Detail: agent.BackupInterrupted}, false, true},
|
|
} {
|
|
t.Run(row.name, func(t *testing.T) {
|
|
c := New(Env{Node: "n1"})
|
|
key, err := nodeengineclient.HaveKey("n1", "web")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
value, err := nodeengineclient.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)
|
|
}
|
|
s := &Shadow{c: c, ran: map[string]string{}}
|
|
s.takeRan(agent.Update{Key: key, Value: value})
|
|
if _, got := s.heldToday()["web"]; got != row.heldToday {
|
|
t.Errorf("web held today: %v, want %v", got, row.heldToday)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// setFor sets a package global (a wait the composer or the shadow reads) for one test, and puts back what it held when
|
|
// the test ends, so no later test runs on another test's waits.
|
|
func setFor[T any](t *testing.T, p *T, v T) {
|
|
t.Helper()
|
|
old := *p
|
|
*p = v
|
|
t.Cleanup(func() { *p = old })
|
|
}
|
|
|
|
type gate struct{ open atomic.Bool }
|
|
|
|
func (g *gate) MayAct(...string) bool { return g.open.Load() }
|
|
|
|
// ADR 0636 §1: a node that has not read the seat list composes nothing new of a module whose seats are not bootstrapped,
|
|
// says it holds the module (so its have says why) and hands it at the read, which the reader says as an event; what
|
|
// already runs with the same composition is not held.
|
|
func TestAnUnreadSeatListHoldsWhatIsNewAndHandsItAtTheRead(t *testing.T) {
|
|
b := newFakeBus()
|
|
fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
|
|
digest := meshcatalogclient.ManifestDigest([]byte(lemurs))
|
|
man, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "lemurs", Manifest: lemurs})
|
|
key, _ := meshcatalogclient.ManifestsKey(digest)
|
|
b.kv[buckets.Manifests+"/"+key] = man
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
env := b.env()
|
|
g := &gate{}
|
|
changed := make(chan struct{}, 1)
|
|
env.SeatList, env.SeatListChanged = g, changed
|
|
var mu sync.Mutex
|
|
heldBy := map[string]string{}
|
|
env.Refused = func(_ string, held map[string]string) { mu.Lock(); heldBy = held; mu.Unlock() }
|
|
c := New(env)
|
|
out, err := c.Watch(ctx, "n1.>")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
place, _ := meshmoduleassignerclient.EncodePlacement(meshmoduleassigner.Placement{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}})
|
|
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
|
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
|
want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true})
|
|
nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
|
|
b.put(t, buckets.Placement, "placement.n1", place)
|
|
b.put(t, buckets.Have, "n1.node", have)
|
|
b.put(t, buckets.Modules, "lemurs", mod)
|
|
b.put(t, buckets.Wants, "want.n1.lemurs", want)
|
|
b.put(t, buckets.Wants, "want.n1.node", nodeWant)
|
|
for _, bucket := range []string{buckets.Wants, buckets.Placement, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
|
b.send(bucket, agent.Update{Caught: true})
|
|
}
|
|
handed := map[string]bool{}
|
|
for {
|
|
u := next(t, out)
|
|
if u.Caught {
|
|
break
|
|
}
|
|
handed[u.Key] = true
|
|
}
|
|
mu.Lock()
|
|
why := heldBy["lemurs"]
|
|
mu.Unlock()
|
|
if why != WaitsForTheSeatList {
|
|
t.Errorf("lemurs is held with %q, want %q", why, WaitsForTheSeatList)
|
|
}
|
|
for k := range handed {
|
|
if strings.Contains(k, "lemurs") {
|
|
t.Errorf("the node handed %s before it read the seat list", k)
|
|
}
|
|
}
|
|
// The node's own composition folds modules and accounts it cannot judge before the read: held whole (the ruling of
|
|
// ADR 0636 §1), under the scope agent.NodeModule.
|
|
mu.Lock()
|
|
ownWhy := heldBy[agent.NodeModule]
|
|
mu.Unlock()
|
|
if ownWhy != WaitsForTheSeatListOwn || handed["n1.node"] {
|
|
t.Errorf("the node's own composition is held with %q and handed=%v, want %q and not handed: %v", ownWhy, handed["n1.node"], WaitsForTheSeatListOwn, handed)
|
|
}
|
|
// The reader reads the list and says so: the module and the node's own are handed, with no timer.
|
|
g.open.Store(true)
|
|
changed <- struct{}{}
|
|
seen := map[string]bool{}
|
|
for !(seen["lemurs"] && seen["node"]) {
|
|
u := next(t, out)
|
|
if u.Deleted {
|
|
continue
|
|
}
|
|
seen["lemurs"] = seen["lemurs"] || strings.Contains(u.Key, "lemurs")
|
|
seen["node"] = seen["node"] || u.Key == "n1.node"
|
|
}
|
|
mu.Lock()
|
|
still := heldBy["lemurs"]
|
|
mu.Unlock()
|
|
if still != "" {
|
|
t.Errorf("lemurs is still held after the read: %q", still)
|
|
}
|
|
}
|
|
|
|
// ADR 0636 §1, the cycle of the node's own composition: while the list is unread the node's own composition is held, but
|
|
// a node whose new own composition must start the holder of the seat mesh-catalog composes of it, whatever the list says,
|
|
// the holder's resources and its account and nothing else, over what the node's own composition ran before; and the rest
|
|
// of the new one is handed once the list was read.
|
|
func TestAnUnreadSeatListLetsThroughOfTheOwnCompositionOnlyWhatTheHolderOfTheCatalogNeeds(t *testing.T) {
|
|
catalog := `{"module":"catalog","version":"1","claims":[{"name":"` + meshcatalog.Seat + `"}]}`
|
|
other := `{"module":"other","version":"1","claims":[{"name":"node-files"}]}`
|
|
c := New(newFakeBus().env())
|
|
c.env.SeatList = &gate{}
|
|
c.modules = map[string]meshcatalog.ModulesByModuleValue{"catalog": {Manifest: "dc"}, "other": {Manifest: "do"}}
|
|
c.manifests = map[string][]byte{"dc": []byte(catalog), "do": []byte(other)}
|
|
res := func(id string) json.RawMessage {
|
|
return json.RawMessage(`{"id":"` + id + `","type":"directory","path":"/x/` + id + `"}`)
|
|
}
|
|
ran := compositions.Composed{Node: "n1", Resources: []json.RawMessage{res("base.dir")},
|
|
Own: compositions.NodeOwn{Identities: []declaration.Identity{{Module: "base", Name: "mesh-base", UID: 1}}}}
|
|
next := compositions.Composed{Node: "n1", Resources: []json.RawMessage{res("base.dir"), res("catalog.dir"), res("other.dir")},
|
|
Own: compositions.NodeOwn{Identities: []declaration.Identity{{Module: "base", Name: "mesh-base", UID: 1},
|
|
{Module: "catalog", Name: "mesh-catalog", UID: 2}, {Module: "other", Name: "mesh-other", UID: 3}}}}
|
|
comps := map[string]compositions.Composed{"n1.node": next}
|
|
held := c.holdUntilListed(context.Background(), reader{c: c}, comps, nil, []compositions.Composed{ran})
|
|
got, ok := comps["n1.node"]
|
|
if !ok {
|
|
t.Fatalf("the node's own composition is held whole, so the holder of the seat mesh-catalog is never started and the list never read")
|
|
}
|
|
var ids []string
|
|
for _, raw := range got.Resources {
|
|
var h struct{ ID string }
|
|
_ = json.Unmarshal(raw, &h)
|
|
ids = append(ids, h.ID)
|
|
}
|
|
var names []string
|
|
for _, i := range got.Own.Identities {
|
|
names = append(names, i.Name)
|
|
}
|
|
if !slices.Equal(ids, []string{"base.dir", "catalog.dir"}) || !slices.Equal(names, []string{"mesh-base", "mesh-catalog"}) {
|
|
t.Errorf("the own composition lets through resources %v and accounts %v, want what ran and the holder's alone", ids, names)
|
|
}
|
|
if held[agent.NodeModule] != WaitsForTheSeatListOwnPart {
|
|
t.Errorf("the rest of the own composition is held with %q, want %q", held[agent.NodeModule], WaitsForTheSeatListOwnPart)
|
|
}
|
|
// Nothing ran: a fresh node composes the holder's part alone.
|
|
comps = map[string]compositions.Composed{"n1.node": next}
|
|
c.holdUntilListed(context.Background(), reader{c: c}, comps, nil, nil)
|
|
if got := comps["n1.node"]; len(got.Resources) != 1 || len(got.Own.Identities) != 1 || got.Own.Identities[0].Name != "mesh-catalog" {
|
|
t.Errorf("a fresh node lets through %d resources and %d accounts, want the holder's one each", len(got.Resources), len(got.Own.Identities))
|
|
}
|
|
}
|
|
|
|
// ADR 0636 §1: a reader the node could not start is no reader that gates nothing: the node holds like one that waits.
|
|
func TestACompositionHoldsWhenTheSeatListReaderCouldNotStart(t *testing.T) {
|
|
gate := Unread{}
|
|
c := New(newFakeBus().env())
|
|
comps := map[string]compositions.Composed{"n1.node": {Node: "n1"}}
|
|
c.env.SeatList = gate
|
|
held := c.holdUntilListed(context.Background(), reader{c: c}, comps, nil, nil)
|
|
if _, ok := held[agent.NodeModule]; !ok || len(comps) != 0 {
|
|
t.Errorf("held %v, comps %v: a reader that did not start must hold the node's own composition", held, comps)
|
|
}
|
|
}
|