jschoubben
caafd9306e
Pins mesh-sdk 05b67831, whose compositions fold a module no new-form build names into the node's own. The shadow composed a placed module it runs inside its own composition (nats, photos) at the trunk's build, as if the candidate added it; shadowBuilds now leaves those to the fold, and both paths pass frombus.InOwn of what runs.
622 lines
24 KiB
Go
622 lines
24 KiB
Go
package selfcompose
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/compose"
|
|
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
|
|
"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 a channel a test writes updates to, and the keys Get reads.
|
|
type fakeBus struct {
|
|
mu sync.Mutex
|
|
chans map[string]chan agent.Update
|
|
kv map[string][]byte // bucket/key
|
|
said []string
|
|
}
|
|
|
|
func newFakeBus() *fakeBus {
|
|
return &fakeBus{chans: map[string]chan agent.Update{}, kv: map[string][]byte{}}
|
|
}
|
|
|
|
func (b *fakeBus) ch(bucket string) chan agent.Update {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if b.chans[bucket] == nil {
|
|
b.chans[bucket] = make(chan agent.Update, 16)
|
|
}
|
|
return b.chans[bucket]
|
|
}
|
|
|
|
func (b *fakeBus) env() Env {
|
|
return Env{Node: "n1",
|
|
Watch: func(_ context.Context, bucket, _ string) (<-chan agent.Update, error) { return b.ch(bucket), nil },
|
|
Get: func(_ context.Context, bucket, key string) ([]byte, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
return b.kv[bucket+"/"+key], nil
|
|
},
|
|
Say: func(s string) { b.mu.Lock(); b.said = append(b.said, s); b.mu.Unlock() }}
|
|
}
|
|
|
|
func (b *fakeBus) put(t *testing.T, bucket, key string, value []byte) {
|
|
t.Helper()
|
|
b.ch(bucket) <- agent.Update{Key: key, Value: value}
|
|
}
|
|
|
|
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 := delivery.SourcePrefix2 + strings.Repeat("a", 64)
|
|
digest := delivery.ManifestDigest([]byte(lemurs))
|
|
man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: lemurs})
|
|
key, _ := delivery.ManifestsKey(digest)
|
|
b.kv[delivery.ManifestsBucket+"/"+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}}]}`, 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
|
|
}
|
|
c := New(env)
|
|
out, err := c.Watch(ctx, "n1.>")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []delivery.PlacedModule{{Module: "lemurs"}}})
|
|
have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
|
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"})
|
|
want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
|
|
nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
|
|
b.put(t, delivery.PlacementBucket, "placement.n1", place)
|
|
b.put(t, delivery.HaveBucket, "n1.node", have)
|
|
b.put(t, delivery.ModulesBucket, "lemurs", mod)
|
|
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want)
|
|
b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
|
|
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} {
|
|
b.ch(bucket) <- agent.Update{Caught: true}
|
|
}
|
|
got := map[string]bool{}
|
|
for {
|
|
u := next(t, out)
|
|
if u.Caught {
|
|
break
|
|
}
|
|
c, err := delivery.DecodeComposed(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)
|
|
}
|
|
got[u.Key] = true
|
|
}
|
|
lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp)
|
|
nodeKey, _ := delivery.ComposedNodeKey("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)
|
|
}
|
|
|
|
// A config value of another module, and a value changing nothing composed, hand nothing.
|
|
b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1}`))
|
|
quiet(t, out)
|
|
|
|
// A want removed takes its composition back.
|
|
removed, _ := delivery.EncodeWant(delivery.Want{Removed: true})
|
|
b.put(t, delivery.AssignerWantsBucket, "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 := delivery.SourcePrefix2 + strings.Repeat("b", 64)
|
|
missing, _ := delivery.EncodeWant(delivery.Want{FP: gone, Self: true})
|
|
b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing)
|
|
quiet(t, out)
|
|
b.put(t, delivery.AssignerWantsBucket, "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") {
|
|
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, delivery.AssignerWantsBucket, "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 := delivery.ManifestDigest([]byte(keeper))
|
|
kman, _ := delivery.EncodeManifest(delivery.Manifest{Module: "keeper", JSON: keeper})
|
|
kkey, _ := delivery.ManifestsKey(kdigest)
|
|
b.mu.Lock()
|
|
b.kv[delivery.ManifestsBucket+"/"+kkey] = kman
|
|
b.mu.Unlock()
|
|
kmod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"})
|
|
both, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
|
|
Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "keeper"}}})
|
|
b.put(t, delivery.ModulesBucket, "keeper", kmod)
|
|
b.put(t, delivery.PlacementBucket, "placement.n1", both)
|
|
b.put(t, delivery.AssignerWantsBucket, "want.n1.keeper", want)
|
|
keeperKey, _ := delivery.ComposedKey("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}`))
|
|
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)
|
|
}
|
|
|
|
// 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.
|
|
boxOf := func(raw string) string {
|
|
c, err := delivery.DecodeComposed([]byte(raw))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sealed := agent.SealedIn(c)
|
|
if len(sealed) != 1 {
|
|
t.Fatalf("keeper's composition carries %v", sealed)
|
|
}
|
|
return sealed[0]
|
|
}
|
|
// 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.
|
|
for _, row := range []struct {
|
|
name, held, in string
|
|
kept bool
|
|
}{{"the have holds the rotated value", boxOf(string(rotated.Value)), "keeper", true},
|
|
{"the have holds the value before the rotation", boxOf(before), "keeper", false},
|
|
{"the node's own key holds the rotated value", boxOf(string(rotated.Value)), agent.NodeModule, true}} {
|
|
restarted := env
|
|
restarted.Running = func() []delivery.Composed { return nil }
|
|
restarted.Held = func(module string) []string {
|
|
if module == row.in {
|
|
return []string{row.held}
|
|
}
|
|
return 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)
|
|
again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}, agent.NodeModule: {}}
|
|
rows, err := reader{c: again}.OwnSecrets(ctx, "n1")
|
|
if err != nil || len(rows) != 1 {
|
|
t.Fatalf("%s: own secrets %+v, %v", row.name, rows, err)
|
|
}
|
|
if kept := rows[0].Sealed == row.held; kept != row.kept {
|
|
t.Errorf("%s: composed with %s (held %s), kept %v, want %v", row.name, rows[0].Sealed, row.held, kept, row.kept)
|
|
}
|
|
}
|
|
|
|
// 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() []delivery.Composed {
|
|
return []delivery.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) { 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 != `{"run":1}` ||
|
|
issued[0].Consumer != "n3" || issued[0].ForProvider != `{"p":1}` {
|
|
t.Errorf("credentials %+v issued %+v", creds, issued)
|
|
}
|
|
// 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")
|
|
pc.refused("a stand-in for web's password")
|
|
pc.refused("")
|
|
if len(refusals) != 2 || refusals[0] == "" || refusals[1] != "" {
|
|
t.Errorf("the agent was told %q", refusals)
|
|
}
|
|
if asked, _ := json.Marshal(sent["credentials"]); !strings.Contains(string(asked), `"role":"consumer"`) || !strings.Contains(string(asked), `"role":"provider"`) {
|
|
t.Errorf("the vault was asked %s", asked)
|
|
}
|
|
// ADR 0559 §4: the node answers placement candidates in shadow, placed or not, applying nothing and handing no
|
|
// composition: the same; a module the group adds, its secret not yet issued; a module removed; a placement it cannot
|
|
// compose; a group whose keys went is no longer answered. What runs is never composed from a candidate: the
|
|
// Composer does not read the state.
|
|
for _, in := range New(env).inputs() {
|
|
if in.bucket == delivery.PlacementCandidatesBucket {
|
|
t.Fatal("the node's own composer reads placement candidates")
|
|
}
|
|
}
|
|
sb := newPatternBus()
|
|
trunk := delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []delivery.PlacedModule{{Module: "lemurs"}}}
|
|
mods := map[string]delivery.Module{"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{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}}
|
|
senv := readOnly.env()
|
|
senv.Watch = sb.watch
|
|
senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
|
|
return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here
|
|
}
|
|
// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
|
|
running, _, err := frombus.Compose(ctx, fixedReader{placements: map[string]delivery.Placement{"n1": trunk},
|
|
records: map[string]delivery.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() []delivery.Composed {
|
|
var out []delivery.Composed
|
|
for _, c := range running {
|
|
out = append(out, c)
|
|
}
|
|
return out
|
|
}
|
|
answered := make(chan map[string]delivery.CandidateAnswer, 8)
|
|
sh := NewShadow(senv, func(_ context.Context, a map[string]delivery.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(delivery.PlacementCandidatesBucket, true)
|
|
go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond })
|
|
time.Sleep(30 * time.Millisecond)
|
|
sb.missing(delivery.PlacementCandidatesBucket, false)
|
|
sb.made(6)
|
|
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
|
|
tp, _ := delivery.EncodePlacement(trunk)
|
|
sb.put(delivery.PlacementBucket, "placement.n1", tp, 1)
|
|
sb.put(delivery.HaveBucket, "n1.node", rec, 1)
|
|
for name, m := range mods {
|
|
raw, _ := delivery.EncodeModule(m)
|
|
sb.put(delivery.ModulesBucket, name, raw, 1)
|
|
}
|
|
candidate := func(group string, rev uint64, modules ...string) {
|
|
p := trunk
|
|
p.Modules = nil
|
|
for _, m := range modules {
|
|
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
|
|
}
|
|
raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: group,
|
|
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-" + group}, Placement: &p})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
k, _ := delivery.PlacementCandidateKey(group, "n1")
|
|
sb.put(delivery.PlacementCandidatesBucket, k, raw, rev)
|
|
}
|
|
candidate("same", 3, "lemurs")
|
|
candidate("adds", 4, "lemurs", "keeper")
|
|
candidate("drops", 5)
|
|
candidate("ghost", 6, "lemurs", "ghost")
|
|
sb.caught()
|
|
var answers map[string]delivery.CandidateAnswer
|
|
waitFor := func(why string, ok func(map[string]delivery.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]delivery.CandidateAnswer) bool { return len(a) == 4 })
|
|
for group, want := range map[string]func(delivery.CandidateAnswer) bool{
|
|
"same": func(a delivery.CandidateAnswer) bool {
|
|
return a.Verdict == delivery.CandidateSame && a.Head == "h-same" && a.Revision == 3 && len(a.Same) == 2
|
|
},
|
|
"adds": func(a delivery.CandidateAnswer) bool {
|
|
return a.Verdict == delivery.CandidateDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" &&
|
|
len(a.Added[0].Unissued) == 1 && len(a.Refusals) == 0
|
|
},
|
|
"drops": func(a delivery.CandidateAnswer) bool {
|
|
return a.Verdict == delivery.CandidateDiffers && len(a.Removed) == 1 && a.Removed[0] == "lemurs"
|
|
},
|
|
"ghost": func(a delivery.CandidateAnswer) bool {
|
|
return a.Verdict == delivery.CandidateRefused && a.Refused != ""
|
|
},
|
|
} {
|
|
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(delivery.HaveBucket, "n1.node", rec, 2)
|
|
select {
|
|
case a := <-answered:
|
|
t.Errorf("answers handed again for nothing changed: %+v", a)
|
|
case <-time.After(100 * time.Millisecond):
|
|
}
|
|
k, _ := delivery.PlacementCandidateKey("same", "n1")
|
|
sb.del(delivery.PlacementCandidatesBucket, k)
|
|
waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 3 })
|
|
|
|
// 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(delivery.PlacementCandidatesBucket, true)
|
|
sb.end(delivery.ModulesBucket)
|
|
dropsKey, _ := delivery.PlacementCandidateKey("drops", "n1")
|
|
sb.del(delivery.PlacementCandidatesBucket, dropsKey)
|
|
time.Sleep(30 * time.Millisecond)
|
|
sb.missing(delivery.PlacementCandidatesBucket, false)
|
|
waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool {
|
|
_, ok := a["drops"]
|
|
return !ok && len(a) == 2
|
|
})
|
|
|
|
// 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 := []delivery.Composed{{Node: "n1", Resources: []json.RawMessage{json.RawMessage(`{"id":"nats.server","type":"service"}`)}},
|
|
{Node: "n1", Module: "lemurs", FP: fp}}
|
|
other := "src2:" + strings.Repeat("9", 64)
|
|
gotFps, gotInOwn := shadowBuilds(inOwnRunning, delivery.Placement{Node: "n1", Modules: []delivery.PlacedModule{{Module: "lemurs"}, {Module: "nats"}, {Module: "keeper"}}},
|
|
map[string]delivery.Module{"lemurs": {FP2: other}, "nats": {FP2: other}, "keeper": {FP2: other}})
|
|
if fmt.Sprint(gotFps) != fmt.Sprint(map[string]string{"lemurs": fp, "keeper": other}) || fmt.Sprint(gotInOwn) != "[nats]" {
|
|
t.Errorf("shadow builds %v, in its own %v; want lemurs at what runs, keeper at the trunk's, nats in its own", gotFps, gotInOwn)
|
|
}
|
|
}
|
|
|
|
// patternBus is a bus whose every watch gets the updates of its bucket matching its pattern, as the bus's own watches
|
|
// do: two watches of one bucket each see what matches theirs.
|
|
type patternBus struct {
|
|
mu sync.Mutex
|
|
watches []patternWatch
|
|
absent map[string]bool
|
|
// kv is each bucket's keys now; once replay is set, a new watch gets those matching it and then Caught, as a watch
|
|
// of the bus's own does.
|
|
kv map[string]map[string]agent.Update
|
|
replay bool
|
|
}
|
|
|
|
// end ends every watch of a bucket, as a watch the bus drops, and from then on each new watch replays what is there.
|
|
func (b *patternBus) end(bucket string) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.replay = true
|
|
kept := b.watches[:0]
|
|
for _, w := range b.watches {
|
|
if w.bucket == bucket {
|
|
close(w.ch)
|
|
continue
|
|
}
|
|
kept = append(kept, w)
|
|
}
|
|
b.watches = kept
|
|
}
|
|
|
|
// missing makes a bucket's watch fail, as one the bus holds no state for, or work again.
|
|
func (b *patternBus) missing(bucket string, gone bool) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if b.absent == nil {
|
|
b.absent = map[string]bool{}
|
|
}
|
|
b.absent[bucket] = gone
|
|
}
|
|
|
|
type patternWatch struct {
|
|
bucket, pattern string
|
|
ch chan agent.Update
|
|
}
|
|
|
|
func newPatternBus() *patternBus { return &patternBus{} }
|
|
|
|
func (b *patternBus) watch(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if b.absent[bucket] {
|
|
return nil, fmt.Errorf("nats: bucket not found")
|
|
}
|
|
w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)}
|
|
b.watches = append(b.watches, w)
|
|
if b.replay {
|
|
for key, u := range b.kv[bucket] {
|
|
if matches(pattern, key) {
|
|
w.ch <- u
|
|
}
|
|
}
|
|
w.ch <- agent.Update{Caught: true}
|
|
}
|
|
return w.ch, nil
|
|
}
|
|
|
|
func matches(pattern, key string) bool {
|
|
p, k := strings.Split(pattern, "."), strings.Split(key, ".")
|
|
for i, tok := range p {
|
|
if tok == ">" {
|
|
return len(k) > i
|
|
}
|
|
if i >= len(k) || (tok != "*" && tok != k[i]) {
|
|
return false
|
|
}
|
|
}
|
|
return len(p) == len(k)
|
|
}
|
|
|
|
func (b *patternBus) send(bucket string, u agent.Update) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
if !u.Caught {
|
|
if b.kv == nil {
|
|
b.kv = map[string]map[string]agent.Update{}
|
|
}
|
|
if b.kv[bucket] == nil {
|
|
b.kv[bucket] = map[string]agent.Update{}
|
|
}
|
|
if u.Deleted {
|
|
delete(b.kv[bucket], u.Key)
|
|
} else {
|
|
b.kv[bucket][u.Key] = u
|
|
}
|
|
}
|
|
for _, w := range b.watches {
|
|
if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) {
|
|
select {
|
|
case w.ch <- u:
|
|
default: // a watch of a run that ended reads no more
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (b *patternBus) put(bucket, key string, value []byte, rev uint64) {
|
|
b.send(bucket, agent.Update{Key: key, Value: value, Revision: rev})
|
|
}
|
|
|
|
func (b *patternBus) del(bucket, key string) { b.send(bucket, agent.Update{Key: key, Deleted: true}) }
|
|
|
|
// made waits until n watches are made, so nothing put is lost.
|
|
func (b *patternBus) made(n int) {
|
|
for {
|
|
b.mu.Lock()
|
|
got := len(b.watches)
|
|
b.mu.Unlock()
|
|
if got >= n {
|
|
return
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// caught says every watch caught up.
|
|
func (b *patternBus) caught() {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
for _, w := range b.watches {
|
|
w.ch <- agent.Update{Caught: true}
|
|
}
|
|
}
|
|
|
|
// fixedReader is frombus's Reader over values held, for composing what a node runs.
|
|
type fixedReader struct {
|
|
placements map[string]delivery.Placement
|
|
records map[string]delivery.NodeRecord
|
|
modules map[string]delivery.Module
|
|
manifests map[string][]byte
|
|
}
|
|
|
|
func (r fixedReader) Placements(context.Context) (map[string]delivery.Placement, error) {
|
|
return r.placements, nil
|
|
}
|
|
func (r fixedReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
|
|
return r.records, nil
|
|
}
|
|
func (r fixedReader) Modules(context.Context) (map[string]delivery.Module, 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
|
|
}
|