mesh/delivery merged: source.main moved to novox/mesh-host@c3a2ffde6755
mesh/check being carried to 6502ba6ad6e6 on c3a2ffde6755 | from f6c0520f26b9
653 lines
27 KiB
Go
653 lines
27 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ed25519"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats-server/v2/server"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/novox/mesh-host/internal/agent"
|
|
"github.com/novox/mesh-host/internal/link"
|
|
"github.com/novox/mesh-host/internal/selfcompose"
|
|
)
|
|
|
|
// The node agent whole, against a real bus with JetStream started in the test: the wants bucket, the node's composition
|
|
// bucket and the have bucket, a session on the bus, and the real apply of a module's composition into a directory of the
|
|
// test. R6-A AC1, AC5 and AC9 (novox/hq ADR 0539).
|
|
|
|
const itNode = "it1"
|
|
|
|
type itBed struct {
|
|
t *testing.T
|
|
conn *nats.Conn
|
|
js nats.JetStreamContext
|
|
// nodeConn and nodeJS are the node's own, under its real grant (buscompose.PermissionsFor, KindNode): the agent runs
|
|
// on them, so a watch or a read the grant does not allow fails here as it fails on the bus (issue 1126, 2026-10-27
|
|
// 09:21: a watch of every have key stopped every node agent). conn is the test's, unrestricted, for setting up.
|
|
nodeConn *nats.Conn
|
|
nodeJS nats.JetStreamContext
|
|
wants nats.KeyValue
|
|
composed nats.KeyValue
|
|
haves nats.KeyValue
|
|
dir string
|
|
opts options
|
|
session *link.Session
|
|
controll ed25519.PrivateKey
|
|
operator ed25519.PrivateKey
|
|
said []string
|
|
// refusals are the bus's asynchronous errors on the node's connection: each permissions violation it met.
|
|
refusals []string
|
|
mu sync.Mutex
|
|
}
|
|
|
|
// aServer is an embedded bus with JetStream and two users: the test's, unrestricted, and the node's, granted exactly what
|
|
// buscompose.PermissionsFor gives a node (KindNode); it answers a connection of each.
|
|
func aServer(t *testing.T, principal buscompose.Principal, refused func(error)) (admin, node *nats.Conn) {
|
|
t.Helper()
|
|
perms, err := buscompose.PermissionsFor(principal)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
granted := &server.Permissions{Publish: &server.SubjectPermission{Allow: perms.Publish, Deny: perms.PublishDeny},
|
|
Subscribe: &server.SubjectPermission{Allow: perms.Subscribe}}
|
|
if perms.AllowResponses {
|
|
granted.Response = &server.ResponsePermission{MaxMsgs: 1, Expires: buscompose.ResponseTTL}
|
|
}
|
|
nodeUser := buscompose.Principal{Kind: buscompose.KindNode, Node: itNode}.Username()
|
|
srv, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true,
|
|
Users: []*server.User{{Username: "test", Password: "test"}, {Username: nodeUser, Password: "node", Permissions: granted}}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
go srv.Start()
|
|
if !srv.ReadyForConnections(10 * time.Second) {
|
|
t.Fatal("the embedded server did not start")
|
|
}
|
|
t.Cleanup(srv.Shutdown)
|
|
if admin, err = nats.Connect(srv.ClientURL(), nats.UserInfo("test", "test")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(admin.Close)
|
|
if node, err = nats.Connect(srv.ClientURL(), nats.UserInfo(nodeUser, "node"), nats.CustomInboxPrefix("_INBOX."+nodeUser),
|
|
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refused(err) })); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(node.Close)
|
|
return admin, node
|
|
}
|
|
|
|
func newItBed(t *testing.T) *itBed {
|
|
t.Helper()
|
|
return newItBedAs(t, buscompose.Principal{Kind: buscompose.KindNode, Node: itNode})
|
|
}
|
|
|
|
// newItBedAs is newItBed with the node's connection under the grant of principal, a node's (KindNode).
|
|
func newItBedAs(t *testing.T, principal buscompose.Principal) *itBed {
|
|
t.Helper()
|
|
// Not skipped off Arch: a package manager that answers is put on PATH, so the apply of files runs on any build node.
|
|
bin := t.TempDir()
|
|
if err := os.WriteFile(filepath.Join(bin, "pacman"), []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH"))
|
|
was := builtFor
|
|
builtFor = "arch"
|
|
t.Cleanup(func() { builtFor = was })
|
|
|
|
b := &itBed{t: t, dir: t.TempDir()}
|
|
b.conn, b.nodeConn = aServer(t, principal, func(err error) {
|
|
b.mu.Lock()
|
|
b.refusals = append(b.refusals, err.Error())
|
|
b.mu.Unlock()
|
|
})
|
|
var err error
|
|
if b.js, err = b.conn.JetStream(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if b.nodeJS, err = b.nodeConn.JetStream(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
composedBucket, _ := delivery.ComposedBucketFor(itNode)
|
|
for _, c := range []struct {
|
|
name string
|
|
into *nats.KeyValue
|
|
history uint8
|
|
}{{delivery.AssignerWantsBucket, &b.wants, 8}, {composedBucket, &b.composed, delivery.ComposedHistory},
|
|
{delivery.HaveBucket, &b.haves, delivery.HaveHistory}} {
|
|
kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: c.name, History: c.history})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
*c.into = kv
|
|
}
|
|
b.opts = options{state: filepath.Join(b.dir, "state", "state.json"), timeout: 2 * time.Second}
|
|
if err := os.MkdirAll(filepath.Dir(b.opts.state), 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, b.controll, _ = ed25519.GenerateKey(nil)
|
|
_, b.operator, _ = ed25519.GenerateKey(nil)
|
|
return b
|
|
}
|
|
|
|
// start runs the node agent under a session on the bus, trusting the controller's key, and the operator's when asked.
|
|
func (b *itBed) start(trustOperator bool) (stop func()) {
|
|
b.t.Helper()
|
|
if trustOperator {
|
|
pub := b.operator.Public().(ed25519.PublicKey)
|
|
if err := os.WriteFile(filepath.Join(filepath.Dir(b.opts.state), operatorKeysFile),
|
|
[]byte("# the operator\n"+base64.StdEncoding.EncodeToString(pub)+"\n"), 0o600); err != nil {
|
|
b.t.Fatal(err)
|
|
}
|
|
}
|
|
b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
|
|
Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }}
|
|
b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode})
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
nodeAgent(ctx, b.opts, b.controll.Public().(ed25519.PublicKey), b.session, nil, b.say, func() {},
|
|
&agentNow{}, func(context.Context, *agent.Agent) {})
|
|
}()
|
|
return func() { cancel(); <-done }
|
|
}
|
|
|
|
func (b *itBed) say(line string) {
|
|
b.mu.Lock()
|
|
b.said = append(b.said, line)
|
|
b.mu.Unlock()
|
|
b.t.Log(line)
|
|
}
|
|
|
|
func fpIt(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) }
|
|
|
|
func fpDigest(c string) string { return "sha256:" + strings.Repeat(c, 64) }
|
|
|
|
// compose writes a module's signed composition of one file into the node's bucket and returns its digest.
|
|
func (b *itBed) compose(module, fp, content string, data bool, signer ed25519.PrivateKey) string {
|
|
b.t.Helper()
|
|
path := filepath.Join(b.dir, module+".conf")
|
|
res, _ := json.Marshal(map[string]any{"id": module + ".conf", "type": "file", "path": path, "content": content, "module": module})
|
|
c := delivery.Composed{Node: itNode, Module: module, FP: fp, Data: data, Resources: []json.RawMessage{res}}
|
|
signed, err := delivery.SignComposed(c, signer)
|
|
if err != nil {
|
|
b.t.Fatal(err)
|
|
}
|
|
raw, err := delivery.EncodeComposed(signed)
|
|
if err != nil {
|
|
b.t.Fatal(err)
|
|
}
|
|
dig, _ := delivery.ComposedDigestOf(signed)
|
|
key, _ := delivery.ComposedKey(itNode, module, fp)
|
|
if _, err := b.composed.Put(key, raw); err != nil {
|
|
b.t.Fatal(err)
|
|
}
|
|
return dig
|
|
}
|
|
|
|
func (b *itBed) want(module, fp, digest, backup string) {
|
|
b.t.Helper()
|
|
raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Composed: digest, Backup: backup})
|
|
if err != nil {
|
|
b.t.Fatal(err)
|
|
}
|
|
key, _ := delivery.WantKey(itNode, module)
|
|
if _, err := b.wants.Put(key, raw); err != nil {
|
|
b.t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func (b *itBed) file(module string) string {
|
|
got, _ := os.ReadFile(filepath.Join(b.dir, module+".conf"))
|
|
return string(got)
|
|
}
|
|
|
|
func (b *itBed) have(module string) delivery.Have {
|
|
key, _ := delivery.HaveKey(itNode, module)
|
|
e, err := b.haves.Get(key)
|
|
if err != nil {
|
|
return delivery.Have{}
|
|
}
|
|
h, _ := delivery.DecodeHave(e.Value())
|
|
return h
|
|
}
|
|
|
|
func within(t *testing.T, d time.Duration, what string, ok func() bool) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(d)
|
|
for !ok() {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal(what)
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// R6-A AC1: given a running agent on a bus, when a want naming a digest and that composition are written, the module is
|
|
// applied within one second and its have names the fingerprint and the digest; a composition whose digest no want names
|
|
// is not applied.
|
|
func TestNatsAWantAndItsCompositionAreAppliedWithinASecond(t *testing.T) {
|
|
b := newItBed(t)
|
|
stop := b.start(false)
|
|
defer stop()
|
|
time.Sleep(300 * time.Millisecond) // the watches are up
|
|
|
|
fp := fpIt("a")
|
|
dig := b.compose("web", fp, "one\n", false, b.controll)
|
|
time.Sleep(300 * time.Millisecond)
|
|
if b.file("web") != "" {
|
|
t.Fatal("a composition no want names was applied")
|
|
}
|
|
began := time.Now()
|
|
b.want("web", fp, dig, "")
|
|
within(t, time.Second, "the module was not applied within one second of its want", func() bool { return b.file("web") == "one\n" })
|
|
t.Logf("applied %s after the want", time.Since(began))
|
|
within(t, 2*time.Second, "the have does not name the fingerprint and the digest", func() bool {
|
|
h := b.have("web")
|
|
return h.FP == fp && h.Composed == dig && h.Phase == delivery.PhaseRunning
|
|
})
|
|
|
|
// A want naming a digest the composition at its fingerprint does not have applies nothing.
|
|
fp2 := fpIt("b")
|
|
b.compose("web", fp2, "two\n", false, b.controll)
|
|
b.want("web", fp2, "sha256:"+strings.Repeat("0", 64), "")
|
|
time.Sleep(300 * time.Millisecond)
|
|
if b.file("web") != "one\n" {
|
|
t.Fatalf("a want naming another digest changed the module: %q", b.file("web"))
|
|
}
|
|
}
|
|
|
|
// R6-A AC5: a data module that ran before asks its backup before any resource changes; a backup that fails holds the
|
|
// module (nothing applied, the have says held with the backup's words); when its want then says skip, it is applied.
|
|
func TestNatsADataModuleAsksItsBackupFirstAndIsHeldWhenItFails(t *testing.T) {
|
|
b := newItBed(t)
|
|
var mu sync.Mutex
|
|
var asked []string
|
|
var onDiskWhenAsked string
|
|
if _, err := b.conn.Subscribe("mesh.seat.node-backup.tool.now."+itNode, func(m *nats.Msg) {
|
|
mu.Lock()
|
|
asked = append(asked, "now")
|
|
onDiskWhenAsked = b.file("db")
|
|
mu.Unlock()
|
|
_ = m.Respond([]byte(`{"result":{"started":"db"}}`))
|
|
// The holder says the backup ended on its event.
|
|
ev, _ := json.Marshal(map[string]any{"node": itNode, "module": "db", "ok": false, "at": time.Now().UTC(),
|
|
"error": "restic: the repository is locked"})
|
|
_ = b.conn.Publish(backedUpEvent, ev)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stop := b.start(false)
|
|
defer stop()
|
|
time.Sleep(300 * time.Millisecond)
|
|
|
|
fp1, fp2 := fpIt("a"), fpIt("b")
|
|
b.want("db", fp1, b.compose("db", fp1, "v1\n", true, b.controll), "")
|
|
within(t, 3*time.Second, "the first composition was not applied", func() bool { return b.file("db") == "v1\n" })
|
|
mu.Lock()
|
|
if len(asked) != 0 {
|
|
t.Fatalf("a module that never ran was backed up: %v", asked)
|
|
}
|
|
mu.Unlock()
|
|
|
|
dig2 := b.compose("db", fp2, "v2\n", true, b.controll)
|
|
b.want("db", fp2, dig2, "")
|
|
within(t, 3*time.Second, "the module was not held behind the failed backup", func() bool {
|
|
h := b.have("db")
|
|
return h.Phase == delivery.PhaseHeld && strings.Contains(h.Detail, "repository is locked")
|
|
})
|
|
mu.Lock()
|
|
if len(asked) != 1 || onDiskWhenAsked != "v1\n" {
|
|
t.Fatalf("the backup was asked %v, with %q on disk: it is asked before any resource changes", asked, onDiskWhenAsked)
|
|
}
|
|
mu.Unlock()
|
|
if b.file("db") != "v1\n" {
|
|
t.Fatalf("something was applied behind a failed backup: %q", b.file("db"))
|
|
}
|
|
if h := b.have("db"); h.FP != fp1 {
|
|
t.Fatalf("a held module should still name what it runs: %+v", h)
|
|
}
|
|
|
|
b.want("db", fp2, dig2, delivery.BackupSkip)
|
|
within(t, 3*time.Second, "the module was not applied once its want said skip", func() bool { return b.file("db") == "v2\n" })
|
|
within(t, 2*time.Second, "the have does not say running at the new fingerprint", func() bool {
|
|
h := b.have("db")
|
|
return h.Phase == delivery.PhaseRunning && h.FP == fp2 && h.Composed == dig2
|
|
})
|
|
}
|
|
|
|
// R6-A AC9, the hand path end to end: with no controller running, a composition signed by the operator's key and a want
|
|
// naming it are applied and the have written; the same signed by a key nobody trusts is not.
|
|
func TestNatsTheHandPathAppliesACompositionTheOperatorSigned(t *testing.T) {
|
|
b := newItBed(t)
|
|
stop := b.start(true)
|
|
defer stop()
|
|
time.Sleep(300 * time.Millisecond)
|
|
|
|
_, stranger, _ := ed25519.GenerateKey(nil)
|
|
fpS := fpIt("c")
|
|
b.want("evil", fpS, b.compose("evil", fpS, "no\n", false, stranger), "")
|
|
time.Sleep(300 * time.Millisecond)
|
|
if b.file("evil") != "" {
|
|
t.Fatal("a composition signed by a key nobody trusts was applied")
|
|
}
|
|
|
|
fp := fpIt("d")
|
|
dig := b.compose("rescue", fp, "back\n", false, b.operator)
|
|
b.want("rescue", fp, dig, "")
|
|
within(t, 2*time.Second, "the operator's composition was not applied", func() bool { return b.file("rescue") == "back\n" })
|
|
within(t, 2*time.Second, "the have was not written", func() bool {
|
|
h := b.have("rescue")
|
|
return h.FP == fp && h.Composed == dig && h.Phase == delivery.PhaseRunning
|
|
})
|
|
}
|
|
|
|
// Issue 1126 (ADR 0547 §2, the milestone's H8): the node agent started whole against a real bus, once the catalogue
|
|
// places its node (the placement written after it started), composes a module from its want, the indexer's manifest, its record and its mesh-config value, and applies it,
|
|
// with no key in the controller's compositions; a changed config value is composed again on its event and applied as a
|
|
// revision of what runs. And a catalogue pull request's placement candidate equal to its placement is answered "same"
|
|
// in the node's own key, and nothing is applied for it (ADR 0559 §4, subtask 10 of the milestone).
|
|
//
|
|
// And (issue 1126, ADR 0560) it reads, through its own session under the grant the grants part composes for it
|
|
// (buscompose.Users, so it carries ServedReads), the build of a placed module that declares one and the one config value
|
|
// a provider on another node serves from (a mail domain); under a grant without that served key the bus refuses the
|
|
// read and the node composes nothing.
|
|
func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
|
|
users, err := buscompose.Users(buscompose.Records{Nodes: []string{itNode, "it2"},
|
|
Assigned: map[string][]buscompose.Declared{"it2": {{Module: "mailer", Served: []string{"domain"}}}}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var principal buscompose.Principal
|
|
for _, u := range users {
|
|
if u.Kind == buscompose.KindNode && u.Node == itNode {
|
|
principal = u
|
|
}
|
|
}
|
|
if len(principal.ServedReads) != 1 {
|
|
t.Fatalf("the grants part's principal of %s reads %v of the served keys", itNode, principal.ServedReads)
|
|
}
|
|
t.Run("without the served key the bus refuses the read", func(t *testing.T) {
|
|
b := newItBed(t)
|
|
stop, _ := h8(t, b, false)
|
|
defer stop()
|
|
// 15 s, above link.historyWithin (10 s): the bus's refusal of a publish reaches only the connection's error
|
|
// handler (nats.go fails no request on it), so the refused read ends at its own bound, then the node says it.
|
|
within(t, 15*time.Second, "the bus did not refuse the served key's read, or the node did not say it", func() bool {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
return strings.Contains(strings.Join(b.refusals, "\n"), "Permissions Violation for Publish to \"$JS.API.CONSUMER.CREATE.KV_"+
|
|
selfcompose.ResolvedBucket) && strings.Contains(strings.Join(b.refusals, "\n"), ".it2.mailer.domain") &&
|
|
strings.Contains(strings.Join(b.said, "\n"), "what mailer serves on it2 (domain)")
|
|
})
|
|
if b.file("web") != "" {
|
|
t.Errorf("composed and applied under a grant without the served key: %q", b.file("web"))
|
|
}
|
|
})
|
|
b := newItBedAs(t, principal)
|
|
stop, st := h8(t, b, true)
|
|
defer stop()
|
|
placement, resolved, candidates := st.placement, st.resolved, st.candidates
|
|
hkey, place, fp := st.hkey, st.place, st.fp
|
|
within(t, 5*time.Second, "the node did not compose and apply its own module once placed", func() bool { return b.file("web") == "greeting=bonjour\n" })
|
|
h8rest(t, b, resolved, candidates, hkey, place, fp)
|
|
h8back(t, b, placement, resolved, candidates, hkey)
|
|
}
|
|
|
|
// h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is compared with
|
|
// the controller's compositions at its wants (here none: refused, naming web), and the placement key deleted returns the agent to them without a
|
|
// restart, said in the journal and the node's have, keeping what runs.
|
|
func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValue, hkey string) {
|
|
t.Helper()
|
|
node := func() delivery.HaveNode {
|
|
e, err := b.haves.Get(hkey)
|
|
if err != nil {
|
|
return delivery.HaveNode{}
|
|
}
|
|
n, _ := delivery.DecodeHaveNode(e.Value())
|
|
return n
|
|
}
|
|
if n := node(); n.Composes != delivery.ComposesOwn {
|
|
t.Errorf("a node composing its own says it composes %q", n.Composes)
|
|
}
|
|
craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "revert/x",
|
|
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 2, Head: "h2"}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ckey, _ := delivery.PlacementCandidateKey("revert/x", itNode)
|
|
if _, err := candidates.Put(ckey, craw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// The controller holds no composition at web's want (the node composed its own): returning would leave web with
|
|
// nothing to apply, and the answer says so rather than "same" (review of the shadow, round 1).
|
|
within(t, 10*time.Second, "the node did not refuse returning to compositions the controller does not hold", func() bool {
|
|
a := node().Candidates["revert/x"]
|
|
return a.Head == "h2" && a.Returns == "" && a.Verdict == delivery.CandidateRefused &&
|
|
strings.Contains(a.Refused, "web: the controller holds no composition")
|
|
})
|
|
ran := b.have("web").Composed
|
|
pkey, _ := delivery.PlacementKey(itNode)
|
|
// A pass in flight at the switch (issue 1126, 2026-10-28: the last pass of the own run, failing on its run's end,
|
|
// marked the next agent's modules failed): a config value changed starts one, and the key goes at once.
|
|
if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := placement.Delete(pkey); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stopWatch := make(chan struct{})
|
|
marked := make(chan Have1126, 1)
|
|
go func() {
|
|
for {
|
|
select {
|
|
case <-stopWatch:
|
|
return
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
if h := b.have("web"); h.Phase == delivery.PhaseFailed {
|
|
select {
|
|
case marked <- Have1126{h.Phase, h.Detail}:
|
|
default:
|
|
}
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
defer close(stopWatch)
|
|
within(t, 15*time.Second, "the node did not return to the controller's compositions when its placement went", func() bool {
|
|
b.mu.Lock()
|
|
said := strings.Join(b.said, "\n")
|
|
b.mu.Unlock()
|
|
return strings.Contains(said, "the catalogue no longer places this node") && node().Composes == delivery.ComposesController
|
|
})
|
|
time.Sleep(3 * time.Second) // what a pass in flight would mark lands within it
|
|
select {
|
|
case h := <-marked:
|
|
t.Errorf("a pass in flight at the switch marked web failed: %+v", h)
|
|
default:
|
|
}
|
|
if h := b.have("web"); h.Composed != ran || h.Phase == delivery.PhaseFailed || b.file("web") != "greeting=salut\n" {
|
|
t.Errorf("returning to the controller's compositions moved what runs: %+v, %q", h, b.file("web"))
|
|
}
|
|
}
|
|
|
|
// Have1126 is what H8's way back watches of web's have.
|
|
type Have1126 struct {
|
|
Phase delivery.Phase
|
|
Detail string
|
|
}
|
|
|
|
// h8state is what h8 set up that the rest of H8 writes to.
|
|
type h8state struct {
|
|
placement, resolved, candidates nats.KeyValue
|
|
hkey string
|
|
place []byte
|
|
fp string
|
|
}
|
|
|
|
// h8 sets H8's bus up and starts the agent, then places the node. With checkBuild, the placed module's build key is
|
|
// first one that does not decode, which the node must read through its session (and say), then a done one.
|
|
func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) {
|
|
t.Helper()
|
|
kvOf := func(bucket string) nats.KeyValue {
|
|
kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 5})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return kv
|
|
}
|
|
placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket)
|
|
resolved := kvOf(selfcompose.ResolvedBucket)
|
|
builds := kvOf(delivery.BuilderBuildsBucket)
|
|
candidates := kvOf(delivery.PlacementCandidatesBucket)
|
|
// The controller's stream the node's facts go into, so the node's key carries the facts it says.
|
|
if _, err := b.js.AddStream(&nats.StreamConfig{Name: "CONTROL", Subjects: []string{link.ReportSubject(itNode)}}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
path := filepath.Join(b.dir, "web.conf")
|
|
man := `{"module":"web","version":"1","config":{"greeting":{"type":"string","description":"what the file says","default":"hello"}},` +
|
|
`"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0644","content":"greeting=${config:greeting}\n"}],` +
|
|
`"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}`
|
|
digest := delivery.ManifestDigest([]byte(man))
|
|
mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "web", JSON: man})
|
|
mkey, _ := delivery.ManifestsKey(digest)
|
|
fp := fpIt("a")
|
|
modv, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40)})
|
|
place, _ := delivery.EncodePlacement(delivery.Placement{Node: itNode, Modules: []delivery.PlacedModule{{Module: "web"}}})
|
|
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: itNode, Capabilities: []string{"service-manager"}}})
|
|
pkey, _ := delivery.PlacementKey(itNode)
|
|
hkey, _ := delivery.HaveNodeKey(itNode)
|
|
// A provider on it2 serving from its own config value: a mail domain, which it1's composition settles.
|
|
mailer := `{"module":"mailer","version":"1","config":{"domain":{"type":"string","description":"the mail domain","required":true}},` +
|
|
`"provides":[{"name":"smtp","scope":"mesh"}],"serves":{"smtp":{"port":587,"domain":"${config:domain}"}}}`
|
|
mdigest := delivery.ManifestDigest([]byte(mailer))
|
|
mlraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "mailer", JSON: mailer})
|
|
mlkey, _ := delivery.ManifestsKey(mdigest)
|
|
mlmod, _ := delivery.EncodeModule(delivery.Module{FP2: fpIt("b"), Manifest: mdigest, Repository: "novox/mailer", Commit: strings.Repeat("2", 40)})
|
|
place2, _ := delivery.EncodePlacement(delivery.Placement{Node: "it2", Modules: []delivery.PlacedModule{{Module: "mailer"}}})
|
|
rec2, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "it2", Capabilities: []string{"service-manager"}}})
|
|
pkey2, _ := delivery.PlacementKey("it2")
|
|
hkey2, _ := delivery.HaveNodeKey("it2")
|
|
buildKey, _ := delivery.BuilderBuildKey("web", fp)
|
|
done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: digest,
|
|
Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://web/files/blobs/" + fpDigest("f"), Digest: fpDigest("f")}}})
|
|
firstBuild := done
|
|
if checkBuild {
|
|
firstBuild = []byte(`{"v":1,"state":"no such state"}`)
|
|
}
|
|
for _, w := range []struct {
|
|
kv nats.KeyValue
|
|
key string
|
|
value []byte
|
|
}{{manifests, mkey, mraw}, {modules, "web", modv}, {b.haves, hkey, rec},
|
|
{resolved, itNode + ".web.greeting", []byte(`{"value":"bonjour","source":"module"}`)},
|
|
{manifests, mlkey, mlraw}, {modules, "mailer", mlmod}, {b.haves, hkey2, rec2}, {placement, pkey2, place2},
|
|
{resolved, "it2.mailer.domain", []byte(`{"value":"example.org","source":"module"}`)}, {builds, buildKey, firstBuild}} {
|
|
if _, err := w.kv.Put(w.key, w.value); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Self: true})
|
|
wkey, _ := delivery.WantKey(itNode, "web")
|
|
if _, err := b.wants.Put(wkey, want); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
stop := b.start(false)
|
|
// Not placed yet: the agent applies the controller's compositions, of which there are none. The catalogue's seed then
|
|
// places the node, and the agent begins again composing its own (ADR 0549 §4), with no restart of the engine.
|
|
time.Sleep(300 * time.Millisecond)
|
|
if b.file("web") != "" {
|
|
t.Fatal("a node the catalogue does not place composed its own")
|
|
}
|
|
if _, err := placement.Put(pkey, place); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if checkBuild {
|
|
// The build of the placed module is read through the node's session: one that does not decode is said, and nothing
|
|
// is applied; a done one, then a config event, and the node composes.
|
|
within(t, 5*time.Second, "the node did not read web's build key through its session", func() bool {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
return strings.Contains(strings.Join(b.said, "\n"), "the builder's "+buildKey)
|
|
})
|
|
if b.file("web") != "" {
|
|
t.Fatal("composed with a build key that does not decode")
|
|
}
|
|
if _, err := builds.Put(buildKey, done); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"bonjour","source":"module"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
return stop, h8state{placement: placement, resolved: resolved, candidates: candidates, hkey: hkey, place: place, fp: fp}
|
|
}
|
|
|
|
// h8rest is H8 once the node composes its own: the have, a config revision, and the candidate answered "same".
|
|
func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey string, place []byte, fp string) {
|
|
t.Helper()
|
|
within(t, 2*time.Second, "the have does not name the build", func() bool {
|
|
h := b.have("web")
|
|
return h.FP == fp && h.Composed != "" && h.Phase == delivery.PhaseRunning
|
|
})
|
|
if keys, _ := b.composed.Keys(); len(keys) != 0 {
|
|
t.Errorf("the controller's compositions hold %v; the node composed its own", keys)
|
|
}
|
|
first := b.have("web").Composed
|
|
if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
within(t, 3*time.Second, "a changed config value was not composed and applied again", func() bool { return b.file("web") == "greeting=salut\n" })
|
|
within(t, 2*time.Second, "the have does not name the revision", func() bool {
|
|
h := b.have("web")
|
|
return h.FP == fp && h.Composed != first && h.Phase == delivery.PhaseRunning
|
|
})
|
|
|
|
// The node says its facts (its record, as the test wrote it), then a candidate equal to its placement is written: the
|
|
// node answers "same" at the candidate's head and revision in its own key, and applies nothing.
|
|
b.session.SayFacts(context.Background(), link.Report{Profile: map[string]any{"capabilities": []any{
|
|
map[string]any{"name": "service-manager", "present": true}}}})
|
|
second := b.have("web").Composed
|
|
same, _ := delivery.DecodePlacement(place)
|
|
craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x",
|
|
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1"}, Placement: &same})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode)
|
|
crev, err := candidates.Put(ckey, craw)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var answer delivery.CandidateAnswer
|
|
within(t, 10*time.Second, "the node did not answer its candidate", func() bool {
|
|
e, err := b.haves.Get(hkey)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
n, err := delivery.DecodeHaveNode(e.Value())
|
|
answer = n.Candidates["feat/x"]
|
|
return err == nil && answer.Head == "h1" && n.Record != nil
|
|
})
|
|
if answer.Verdict != delivery.CandidateSame || answer.Revision != crev || len(answer.Same) != 1 || answer.Same[0] != "web" {
|
|
t.Errorf("the answer to a candidate equal to the placement: %+v", answer)
|
|
}
|
|
if h := b.have("web"); h.Composed != second || b.file("web") != "greeting=salut\n" {
|
|
t.Errorf("a candidate moved what runs: %+v, %q", h, b.file("web"))
|
|
}
|
|
}
|