jschoubben
4af6c274b8
On ace, a node-engine left on its bootstrapped seat list could not find the artifact store's seat, refused avahi's archive reference and with it the whole pass, so the seat-list hold never ran and every module said "reaches no artifact store". While the list is unread, a placed module whose seats are not all bootstrapped is now held as it runs before anything of its build is read (frombus.Scope.Held), its reason names the bootstrapped seats the node holds, and a pass refused whole says the unread list first (issue 1966). Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2214 lines
103 KiB
Go
2214 lines
103 KiB
Go
package selfcompose
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient"
|
|
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
|
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
|
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
|
"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/contracts/go/seats/meshwants"
|
|
"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"
|
|
"maps"
|
|
"os"
|
|
"reflect"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go/overlay"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
|
"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 := assignedValues(t, meshmoduleassigner.Assignments{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.putAssigned(t, 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.Assignments, 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)
|
|
}
|
|
// Issue 1966 (novox, 2026-11-09): every pass names each wanted module it composed nothing of, or composed from the indexer's
|
|
// manifest for want of a done build, with why, and a want it did not take; said once; the agent's word that it waits
|
|
// hands a composition the node believes handed again.
|
|
tfp := meshcatalogclient.SourcePrefix2 + strings.Repeat("7", 64)
|
|
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)
|
|
b.mu.Lock()
|
|
b.kv[buckets.Manifests+"/"+tkey] = tman
|
|
b.mu.Unlock()
|
|
tmod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: tfp, Manifest: tdigest, Repository: "novox/t", Commit: "c"})
|
|
c.takeModule(agent.Update{Key: "tool", Value: tmod, Revision: 9})
|
|
for module, w := range map[string]meshwants.WantsByWantNodeModuleValue{"tool": {Fp: tfp, Self: true}, "ghost": {Fp: fp, Self: true}, "nofp": {Self: true}} {
|
|
raw, _ := bucketstest.Want(w)
|
|
c.takeWant(agent.Update{Key: "want.n1." + module, Value: raw})
|
|
}
|
|
c.mu.Lock()
|
|
placed := c.assignments["n1"]
|
|
placed.Modules = append(slices.Clone(placed.Modules), meshmoduleassigner.PlacedModule{Module: "tool"})
|
|
c.assignments["n1"] = placed
|
|
c.mu.Unlock()
|
|
said := func(sub string) (n int) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
for _, l := range b.said {
|
|
if strings.Contains(l, sub) {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
c.compose(ctx)
|
|
why := map[string]string{"tool": c.WhyUnmade("tool"), "ghost": c.WhyUnmade("ghost"), "nofp": c.WhyUnmade("nofp"), "lemurs": c.WhyUnmade("lemurs")}
|
|
if !strings.Contains(why["tool"], "the builder holds no done build of it at "+tfp) || !strings.Contains(why["ghost"], "places no ghost") ||
|
|
!strings.Contains(why["nofp"], "want not taken: its want names no fingerprint") || why["lemurs"] != "" {
|
|
t.Errorf("the pass says it composed nothing of %q", why)
|
|
}
|
|
// An update of another module moves the picture's revision: the same reasons are not said again.
|
|
c.takeModule(agent.Update{Key: "lemurs", Value: mod, Revision: 10})
|
|
c.compose(ctx)
|
|
if n := said("the node composes nothing of ghost"); n != 1 {
|
|
t.Errorf("the journal says why ghost is not composed %d times, want once", n)
|
|
}
|
|
// The state of this step is taken away again, and the pass that takes back what it handed is made here.
|
|
for _, module := range []string{"tool", "ghost", "nofp"} {
|
|
c.takeWant(agent.Update{Key: "want.n1." + module, Deleted: true})
|
|
}
|
|
c.takeModule(agent.Update{Key: "tool", Deleted: true})
|
|
c.mu.Lock()
|
|
placed.Modules = placed.Modules[:len(placed.Modules)-1]
|
|
c.assignments["n1"] = placed
|
|
c.mu.Unlock()
|
|
c.compose(ctx)
|
|
// The agent waits for what the node believes it handed: handed again; for what it never handed, nothing is.
|
|
c.Waiting("ghost", fp)
|
|
select {
|
|
case u := <-out:
|
|
t.Errorf("handed %s for a wait of a composition never handed", u.Key)
|
|
default:
|
|
}
|
|
c.Waiting("lemurs", fp)
|
|
if u := next(t, out); !strings.Contains(u.Key, "lemurs") {
|
|
t.Errorf("handed %s, want lemurs's composition again", u.Key)
|
|
}
|
|
// 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.assignments {
|
|
c2.assignments[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.assignments["n1"]
|
|
p3.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p3.Modules...), meshmoduleassigner.PlacedModule{Module: "proxy"}, meshmoduleassigner.PlacedModule{Module: "de-spiegel"})
|
|
c3.assignments["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.assignments["n1"]
|
|
p4.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p4.Modules...), meshmoduleassigner.PlacedModule{Module: "power"})
|
|
c4.assignments["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)
|
|
}
|
|
// Issue 1921: the composition verb's held says what holds power and the build it would be composed at, as the input a
|
|
// check's composer reads says it.
|
|
if why := c4.LastHeld()["power"]; !strings.Contains(why, `"shows" is none of them`) ||
|
|
!strings.Contains(why, "held by its manifest's verdict, at the build "+powerFP) {
|
|
t.Errorf("the last pass held power for %q, want its verdict and its build named", why)
|
|
}
|
|
// 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.assignments["n1"]
|
|
p5.Modules = append(append([]meshmoduleassigner.PlacedModule{}, p5.Modules...), meshmoduleassigner.PlacedModule{Module: overlay.Name})
|
|
p5.Decided.Endpoint = "n1.example:51820"
|
|
c5.assignments["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 assignments kept as one key per module and node move nothing composed when written again; a
|
|
// write in keys makes its pass at its node key, the commit, and not at a module key written before it.
|
|
placeBase := meshmoduleassigner.Assignments{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.Assignments, 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.Assignments, w.Key, w.Value, w.Commit})
|
|
}
|
|
return rows
|
|
}
|
|
for k, v := range place {
|
|
_, _ = placeKeys.Put(k, v, 1, false)
|
|
}
|
|
lemursMoved := placeBase
|
|
lemursMoved.Modules = []meshmoduleassigner.PlacedModule{{Module: "lemurs", Values: map[string]any{"greeting": "hoi"}}, {Module: "oldie"}}
|
|
var assignmentRows []struct {
|
|
name, bucket, key string
|
|
value []byte
|
|
pass bool
|
|
}
|
|
for _, step := range []struct {
|
|
what string
|
|
p meshmoduleassigner.Assignments
|
|
}{{"lemurs moved", lemursMoved}, {"lemurs back", placeBase}} {
|
|
assignmentRows = append(assignmentRows, 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 assignment again", buckets.Assignments, "n1.node", place["n1.node"], 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},
|
|
}, assignmentRows...) {
|
|
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 := assignedValues(t, meshmoduleassigner.Assignments{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.putAssigned(t, 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.Assignments, 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 := assignedValues(t, meshmoduleassigner.Assignments{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, which the writes below move.
|
|
bothPlaced, _ := meshmoduleassignerclient.AssignmentsOf(both)
|
|
bothNode := bothPlaced["n1"]
|
|
bothWrites, err := placeKeys.Writes("n1", &bothNode)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, w := range bothWrites {
|
|
if w.Value != nil {
|
|
b.put(t, buckets.Assignments, 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.assignments {
|
|
again.assignments[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.assignments)
|
|
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 assignment 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 assignment 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.ProposalCandidates {
|
|
t.Fatal("the node's own composer reads assignment candidates")
|
|
}
|
|
}
|
|
sb := newPatternBus()
|
|
trunk := meshmoduleassigner.Assignments{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{assignments: map[string]meshmoduleassigner.Assignments{"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 assignment
|
|
// 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.ProposalCandidates, 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.ProposalCandidates, false)
|
|
sb.made(6)
|
|
rec, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
|
tp := assignedValues(t, trunk)
|
|
sb.putAssigned(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, Assignments: &p})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
k := meshmoduleassigner.CandidatesByProposalNodeKey(group, "n1")
|
|
sb.put(buckets.ProposalCandidates, 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 assignment 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.ProposalCandidates, k, raw, 7)
|
|
}
|
|
// A proposal that changes another node's key and reaches n1's composition (ADR 0567 §3): n1's candidate holds no
|
|
// assignment of its own change (Reached, nil), and n1 answers against its current assignment, 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.ProposalCandidates, 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 assignment 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 assignment, 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 assignment 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.putAssigned(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 assignment candidate same "); got != "answering assignment 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 assignment candidate drops "); got != "answering assignment 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 assignment candidate adds "); got != "answering assignment 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.ProposalCandidates, 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.ProposalCandidates, true)
|
|
sb.end(buckets.Modules)
|
|
dropsKey := meshmoduleassigner.CandidatesByProposalNodeKey("drops", "n1")
|
|
sb.del(buckets.ProposalCandidates, dropsKey)
|
|
time.Sleep(30 * time.Millisecond)
|
|
sb.missing(buckets.ProposalCandidates, 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 assignment that cannot be read composes nothing and says why; the node never applies a
|
|
// composition made elsewhere instead. (A assignment 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.Assignments {
|
|
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.Assignments) || !strings.Contains(err.Error(), "cannot be watched") {
|
|
t.Errorf("an unreadable assignment: 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.putAssigned(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", Assignments: &trunk})
|
|
sameKey := meshmoduleassigner.CandidatesByProposalNodeKey("same", "n1")
|
|
nb.put(buckets.ProposalCandidates, 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", Assignments: &trunk})
|
|
oldKey := meshmoduleassigner.CandidatesByProposalNodeKey("older", "n1")
|
|
nb.put(buckets.ProposalCandidates, oldKey, oldRaw, 2)
|
|
for _, bucket := range []string{buckets.ProposalCandidates, buckets.Assignments, 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 assignment 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 assignment candidate same (revision 3; woken by its first answer since the shadow began)",
|
|
"answered assignment candidate same in", "answering assignment candidate older (revision 2;"} {
|
|
if !strings.Contains(said, line) {
|
|
t.Errorf("%q was not said: %q", line, nsaid)
|
|
}
|
|
}
|
|
if strings.Index(said, "answering assignment candidate same") > strings.Index(said, "answering assignment 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.Assignments{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.Assignments{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.Assignments{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 {
|
|
assignments map[string]meshmoduleassigner.Assignments
|
|
records map[string]nodeengine.NodeRecord
|
|
modules map[string]meshcatalog.ModulesByModuleValue
|
|
manifests map[string][]byte
|
|
}
|
|
|
|
func (r fixedReader) Assignments(context.Context) (map[string]meshmoduleassigner.Assignments, error) {
|
|
return r.assignments, 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() }
|
|
|
|
// namedGate is a gate that says the seats it holds, as the reader of the seat list does (seatlist.Reader.Status).
|
|
type namedGate struct {
|
|
*gate
|
|
seats []string
|
|
}
|
|
|
|
func (g *namedGate) Status() seatlist.Status {
|
|
st := seatlist.Status{Heard: g.open.Load()}
|
|
for _, s := range g.seats {
|
|
st.Held = append(st.Held, seatlist.Held{Seat: s, Source: seatlist.SourceBootstrap})
|
|
}
|
|
return st
|
|
}
|
|
|
|
// 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
|
|
// Issue 1966: avahi's archive is in the artifact store, whose seat the bootstrap does not hold, so before the read the
|
|
// node reaches no store; avahi is held for the list before its build is read, and the pass is not refused for it.
|
|
avahi := `{"module":"avahi","version":"1","capabilities":["systemd"],"resources":[{"id":"tools","type":"archive",` +
|
|
`"path":"/opt/avahi","source":"` + catalogue.ArtifactStoreScheme + `avahi/tools/blobs/sha256:` + strings.Repeat("d", 64) + `"}]}`
|
|
avahiDigest := meshcatalogclient.ManifestDigest([]byte(avahi))
|
|
avahiMan, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "avahi", Manifest: avahi})
|
|
avahiKey, _ := meshcatalogclient.ManifestsKey(avahiDigest)
|
|
b.kv[buckets.Manifests+"/"+avahiKey] = avahiMan
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
env := b.env()
|
|
g := &namedGate{gate: &gate{}, seats: []string{"mesh-catalog", "node-engine"}}
|
|
changed := make(chan struct{}, 1)
|
|
env.SeatList, env.SeatListChanged = g, changed
|
|
var mu sync.Mutex
|
|
heldBy := map[string]string{}
|
|
var refusedWhole []string
|
|
env.Refused = func(why string, held map[string]string) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
heldBy = held
|
|
if why != "" {
|
|
refusedWhole = append(refusedWhole, why)
|
|
}
|
|
}
|
|
c := New(env)
|
|
out, err := c.Watch(ctx, "n1.>")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "avahi"}}})
|
|
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.putAssigned(t, place)
|
|
b.put(t, buckets.Have, "n1.node", have)
|
|
b.put(t, buckets.Modules, "lemurs", mod)
|
|
avahiMod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: avahiDigest, Repository: "novox/x", Commit: "c"})
|
|
b.put(t, buckets.Modules, "avahi", avahiMod)
|
|
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.Assignments, 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
|
|
}
|
|
// The reason names the bootstrapped list the node holds (issue 1966: ace's modules said "reaches no artifact store").
|
|
waits := fmt.Sprintf(WaitsForTheSeatListHeld, 2, "mesh-catalog, node-engine") + composesNothingNewOfIt
|
|
mu.Lock()
|
|
why, avahiWhy, whole := heldBy["lemurs"], heldBy["avahi"], slices.Clone(refusedWhole)
|
|
mu.Unlock()
|
|
if why != waits || avahiWhy != waits || len(whole) != 0 {
|
|
t.Errorf("lemurs is held with %q, avahi with %q, the pass refused %q; want both %q and no refusal", why, avahiWhy, whole, waits)
|
|
}
|
|
// A pass refused whole while the list is unread says the list first (issue 1966).
|
|
refusing := b.env()
|
|
refusing.SeatList = g
|
|
refusing.Get = func(context.Context, string, string) ([]byte, error) {
|
|
return nil, errors.New("nats says ${dir:mesh-state}, and nats declares no directory")
|
|
}
|
|
var refusal string
|
|
refusing.Refused = func(w string, _ map[string]string) { refusal = w }
|
|
c2 := New(refusing)
|
|
c.mu.Lock()
|
|
maps.Copy(c2.wants, c.wants)
|
|
maps.Copy(c2.assignments, c.assignments)
|
|
maps.Copy(c2.records, c.records)
|
|
maps.Copy(c2.modules, c.modules)
|
|
c.mu.Unlock()
|
|
c2.compose(ctx)
|
|
if line := fmt.Sprintf(WaitsForTheSeatListHeld, 2, "mesh-catalog, node-engine") + "; "; !strings.HasPrefix(refusal, line) {
|
|
t.Errorf("a pass refused while the list is unread says %q, want it to begin %q", refusal, line)
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
|
|
func TestAHeldOwnCompositionThatWasHandedIsNotDropped(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
|
|
passes := make(chan map[string]string, 8)
|
|
env.Refused = func(_ string, held map[string]string) { passes <- maps.Clone(held) }
|
|
c := New(env)
|
|
out, err := c.Watch(ctx, "n1.>")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
place := assignedValues(t, meshmoduleassigner.Assignments{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.putAssigned(t, 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.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
|
b.send(bucket, agent.Update{Caught: true})
|
|
}
|
|
// The list is read: the node's own composition is handed. Then the reader cannot act again (it restarted and has
|
|
// not read): the composition is held, and what was handed of it stays, as a held module's does.
|
|
g.open.Store(true)
|
|
changed <- struct{}{}
|
|
for u := next(t, out); !(u.Key == "n1.node" && !u.Deleted); u = next(t, out) {
|
|
}
|
|
g.open.Store(false)
|
|
changed <- struct{}{}
|
|
for h := range passes {
|
|
if h[agent.NodeModule] == WaitsForTheSeatListOwn {
|
|
break
|
|
}
|
|
}
|
|
deadline := time.After(100 * time.Millisecond)
|
|
for {
|
|
select {
|
|
case u := <-out:
|
|
if u.Key == "n1.node" && u.Deleted {
|
|
t.Fatalf("the held own composition that was handed is deleted, and the node drops what it runs")
|
|
}
|
|
case <-deadline:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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, the whole case: a node that runs two modules inside its own composition (a build of the old form each),
|
|
// one of them the holder of the seat mesh-catalog, and reads no seat list. Its own composition, handed before the read,
|
|
// carries the holder; the read, said as an event, hands the rest, the other module's resources, with no timer.
|
|
func TestTheHolderOfTheCatalogIsStartedFromTheOwnCompositionBeforeTheListIsReadAndTheRestAfter(t *testing.T) {
|
|
catalog := `{"module":"catalog","version":"1","claims":[{"name":"` + meshcatalog.Seat + `"}],"capabilities":["systemd"],` +
|
|
`"resources":[{"id":"service","type":"service","unit":"catalog.service","state":"running","boot":"enabled"}]}`
|
|
other := `{"module":"other","version":"1","capabilities":["systemd"],` +
|
|
`"resources":[{"id":"service","type":"service","unit":"other.service","state":"running","boot":"enabled"},` +
|
|
`{"id":"extra","type":"service","unit":"extra.service","state":"running","boot":"enabled"}]}`
|
|
b := newFakeBus()
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
env := b.env()
|
|
g := &gate{}
|
|
changed := make(chan struct{}, 1)
|
|
env.SeatList, env.SeatListChanged = g, changed
|
|
env.Running = func() []compositions.Composed {
|
|
return []compositions.Composed{{Node: "n1", Resources: []json.RawMessage{json.RawMessage(`{"id":"other.service","type":"service","unit":"other.service","state":"running","boot":"enabled"}`)}}}
|
|
}
|
|
c := New(env)
|
|
out, err := c.Watch(ctx, "n1.>")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
mods := map[string]string{"catalog": catalog, "other": other}
|
|
var placed []meshmoduleassigner.PlacedModule
|
|
for _, m := range []string{"catalog", "other"} {
|
|
digest := meshcatalogclient.ManifestDigest([]byte(mods[m]))
|
|
man, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: m, Manifest: mods[m]})
|
|
mk, _ := meshcatalogclient.ManifestsKey(digest)
|
|
b.kv[buckets.Manifests+"/"+mk] = man
|
|
rec, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64), Manifest: digest, Repository: "novox/x", Commit: "c"})
|
|
b.put(t, buckets.Modules, m, rec)
|
|
hk, _ := nodeengineclient.HaveKey("n1", m)
|
|
hv, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Build: "build-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
|
b.put(t, buckets.Have, hk, hv)
|
|
placed = append(placed, meshmoduleassigner.PlacedModule{Module: m})
|
|
}
|
|
place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: placed})
|
|
have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
|
nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true})
|
|
b.putAssigned(t, place)
|
|
b.put(t, buckets.Have, "n1.node", have)
|
|
b.put(t, buckets.Wants, "want.n1.node", nodeWant)
|
|
for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} {
|
|
b.send(bucket, agent.Update{Caught: true})
|
|
}
|
|
servicesIn := func(u agent.Update) []string {
|
|
var comp compositions.Composed
|
|
if err := json.Unmarshal(u.Value, &comp); err != nil {
|
|
t.Fatalf("the own composition does not decode: %v", err)
|
|
}
|
|
var ids []string
|
|
for _, raw := range comp.Resources {
|
|
var h struct{ ID string }
|
|
_ = json.Unmarshal(raw, &h)
|
|
if strings.HasPrefix(h.ID, "catalog.") || strings.HasPrefix(h.ID, "other.") {
|
|
ids = append(ids, h.ID)
|
|
}
|
|
}
|
|
return ids
|
|
}
|
|
var first agent.Update
|
|
for first.Key == "" {
|
|
if u := next(t, out); u.Key == "n1.node" && !u.Deleted {
|
|
first = u
|
|
}
|
|
}
|
|
if got := servicesIn(first); !slices.Equal(got, []string{"other.service", "catalog.service"}) {
|
|
t.Fatalf("before the read the own composition carries services %v, want what ran and the holder's alone", got)
|
|
}
|
|
g.open.Store(true)
|
|
changed <- struct{}{}
|
|
for {
|
|
u := next(t, out)
|
|
if u.Key == "n1.node" && !u.Deleted {
|
|
if got := servicesIn(u); !slices.Equal(got, []string{"catalog.service", "other.service", "other.extra"}) {
|
|
t.Fatalf("after the read the own composition carries services %v, want what ran, the rest of the module and the holder", got)
|
|
}
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
}
|
|
|
|
// assignedValues is every key the assigner writes for p: the node's, `<node>.node`, and each module's, `<node>.<module>`.
|
|
func assignedValues(t *testing.T, p meshmoduleassigner.Assignments) map[string][]byte {
|
|
t.Helper()
|
|
vals, err := meshmoduleassignerclient.AssignmentsValues(p)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return vals
|
|
}
|
|
|
|
// putAssigned writes the keys of one node's assignments into the assignments state.
|
|
func (b *fakeBus) putAssigned(t *testing.T, vals map[string][]byte) {
|
|
t.Helper()
|
|
for _, k := range slices.Sorted(maps.Keys(vals)) {
|
|
b.put(t, buckets.Assignments, k, vals[k])
|
|
}
|
|
}
|
|
|
|
// putAssigned sends the keys of one node's assignments, each at the revision.
|
|
func (b *patternBus) putAssigned(vals map[string][]byte, rev uint64) {
|
|
for _, k := range slices.Sorted(maps.Keys(vals)) {
|
|
b.put(buckets.Assignments, k, vals[k], rev)
|
|
}
|
|
}
|