mesh-host/cmd/mesh-host/nodeagent_integration_test.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/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
	wants    nats.KeyValue
	composed nats.KeyValue
	haves    nats.KeyValue
	dir      string
	opts     options
	session  *link.Session
	controll ed25519.PrivateKey
	operator ed25519.PrivateKey
	said     []string
	mu       sync.Mutex
}

func aServer(t *testing.T) *nats.Conn {
	t.Helper()
	srv, err := server.NewServer(&server.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	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)
	conn, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(conn.Close)
	return conn
}

func newItBed(t *testing.T) *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, conn: aServer(t), dir: t.TempDir()}
	var err error
	if b.js, err = b.conn.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}
	b.session.Attach(context.Background(), link.OverNATS{Conn: b.conn, JS: b.js, 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) }

// 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.
func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
	b := newItBed(t)
	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)
	kvOf(delivery.BuilderBuildsBucket)
	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"}]}`
	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)
	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"}`)}} {
		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)
	defer stop()
	// 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)
	}
	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" })
	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"}`)); 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
	})
}