mesh-host/internal/selfcompose/composedstate_test.go

package selfcompose

import (
	"context"
	"encoding/json"
	"errors"
	"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
	"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
	"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
	"github.com/novox/mesh-host/internal/buckets"
	"github.com/novox/mesh-host/internal/buckets/bucketstest"
	engcomp "github.com/novox/mesh-host/internal/compositions"
	"maps"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/novox/mesh-host/internal/agent"
)

// fakeComposed is a node's composition state: its keys, and every write and deletion made.
type fakeComposed struct {
	mu      sync.Mutex
	kv      map[string]ComposedEntry
	rev     uint64
	puts    []string
	deletes []string
}

func (f *fakeComposed) Entries(context.Context) (map[string]ComposedEntry, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := map[string]ComposedEntry{}
	for k, v := range f.kv {
		out[k] = v
	}
	return out, nil
}

func (f *fakeComposed) Put(_ context.Context, key string, value []byte) (uint64, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.rev++
	f.kv[key] = ComposedEntry{Value: value, Revision: f.rev}
	f.puts = append(f.puts, key)
	return f.rev, nil
}

func (f *fakeComposed) Delete(_ context.Context, key string) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	delete(f.kv, key)
	f.deletes = append(f.deletes, key)
	return nil
}

func (f *fakeComposed) written() []string {
	f.mu.Lock()
	defer f.mu.Unlock()
	var out []string
	for _, key := range f.puts {
		// The entries and wants the node keeps are its composition state's too (§683), not compositions.
		if !strings.HasPrefix(key, keptEntryPrefix) && !strings.HasPrefix(key, keptWantPrefix) {
			out = append(out, key)
		}
	}
	return out
}

// keep holds in the state, as an earlier run of the node left them, the entry of fp and the want of module at it (§683): the
// composer then composes at its start from what it kept, and asks the catalog half of nothing first.
func (f *fakeComposed) keep(t *testing.T, half catalogHalf, module, fp string) {
	t.Helper()
	entry, _ := json.Marshal(meshcatalog.EntryAnswer{Entry: half[fp]})
	want, _ := json.Marshal(keptWant{V: 1, Since: time.Now(), Want: meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true}})
	f.mu.Lock()
	defer f.mu.Unlock()
	f.kv[entryKey(fp)] = ComposedEntry{Value: entry, Revision: 1}
	f.kv[wantStateKey(module)] = ComposedEntry{Value: want, Revision: 1}
}

// novox/hq ADR 0593 §6: the node publishes each composition it makes in its composition state, a record never applied:
// the wanted module and the node's own at the first pass, from the entry and want it kept; nothing again for a pass that
// changes nothing; a failed build of a placed module said refused by its build-ended event. And a module's keys are kept while among the five it wrote last or named
// by a want, and deleted otherwise.
func TestTheNodePublishesItsCompositions(t *testing.T) {
	t.Run("the verb compose is answered while a pass runs", composeAnsweredWhileAPassRuns)
	b := newFakeBus()
	fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
	ofp := meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64)
	digest := meshcatalogclient.ManifestDigest([]byte(lemurs))
	oldie := `{"module":"oldie","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"oldie.service","state":"running","boot":"enabled"}]}`
	odigest := meshcatalogclient.ManifestDigest([]byte(oldie))
	brokenJSON := strings.ReplaceAll(oldie, "oldie", "broken")
	waitingJSON := strings.ReplaceAll(oldie, "oldie", "waiting")
	bdigest, wdigest := meshcatalogclient.ManifestDigest([]byte(brokenJSON)), meshcatalogclient.ManifestDigest([]byte(waitingJSON))
	for d, json := range map[string]string{digest: lemurs, odigest: oldie, bdigest: brokenJSON, wdigest: waitingJSON} {
		man, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: map[string]string{digest: "lemurs", odigest: "oldie", bdigest: "broken", wdigest: "waiting"}[d], Manifest: json})
		key, _ := meshcatalogclient.ManifestsKey(d)
		b.kv[buckets.Manifests+"/"+key] = man
	}
	state := &fakeComposed{kv: map[string]ComposedEntry{}}
	env := b.env()
	env.ComposedState = state
	ofp0 := meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64)
	half := catalogHalf{}
	half.add(t, "lemurs", fp, lemurs)
	half.add(t, "oldie", ofp0, oldie)
	env.Entry = half.call()
	state.keep(t, half, "lemurs", fp)
	// Design 47, *What each step says*: each module composition written says composed, with its fingerprint and digest.
	var saidMu sync.Mutex
	var saidComposed []string
	env.SayComposed = func(module, fingerprint, digest string) {
		saidMu.Lock()
		defer saidMu.Unlock()
		saidComposed = append(saidComposed, module+" "+fingerprint+" "+digest)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	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: "oldie"}, {Module: "broken"}, {Module: "waiting"}}})
	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"})
	omod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: ofp, Manifest: odigest, Repository: "novox/o", 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.Modules, "oldie", omod)
	bfp := meshcatalogclient.SourcePrefix2 + strings.Repeat("d", 64)
	wfp := meshcatalogclient.SourcePrefix2 + strings.Repeat("e", 64)
	for m, f := range map[string]string{"broken": bfp, "waiting": wfp} {
		raw, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: f, Manifest: map[string]string{"broken": bdigest, "waiting": wdigest}[m], Repository: "novox/o", Commit: "c"})
		b.put(t, buckets.Modules, m, raw)
	}
	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, ResolvedBucket} {
		b.send(bucket, agent.Update{Caught: true})
	}
	for u := next(t, out); !u.Caught; u = next(t, out) {
	}
	lemursKey, _ := engcomp.Key("n1", "lemurs", fp)
	waitFor := func(what string, ok func([]string) bool) []string {
		t.Helper()
		deadline := time.Now().Add(5 * time.Second)
		for time.Now().Before(deadline) {
			if got := state.written(); ok(got) {
				return got
			}
			time.Sleep(10 * time.Millisecond)
		}
		b.mu.Lock()
		t.Logf("said: %q", b.said)
		b.mu.Unlock()
		t.Fatalf("%s: written %v", what, state.written())
		return nil
	}
	first := waitFor("the first pass", func(got []string) bool { return len(got) >= 2 })
	if !slices.Equal(first, []string{lemursKey, "n1.node"}) {
		t.Errorf("the first pass wrote %v, want the wanted module's key and the node's own", first)
	}
	if raw := state.kv[lemursKey].Value; raw != nil {
		comp, err := engcomp.Decode(raw)
		if err != nil || comp.Module != "lemurs" || comp.FP != fp || len(comp.Resources) == 0 {
			t.Errorf("the published composition of lemurs: %+v, %v", comp, err)
		}
	}
	if comp, err := engcomp.Decode(state.kv[lemursKey].Value); err == nil {
		dig, _ := engcomp.DigestOf(comp)
		saidMu.Lock()
		if len(saidComposed) != 1 || saidComposed[0] != "lemurs "+fp+" "+dig || dig == "" {
			t.Errorf("composed said %v, want once for lemurs at %s with its digest %q (the node's own says none)", saidComposed, fp, dig)
		}
		saidMu.Unlock()
	}
	// Design 47 G §199, §556 (test 56): what the nodes say they run is acted on once the first read of the stream ended; the
	// newest by `at` stands, an older one and the same again change nothing.
	at := time.Now()
	c.NoteRunning(nodeengine.Running{Node: "n2", Module: "lemurs", Fingerprint: "a", Manifest: "sha256:a", At: at})
	if got := c.heardNow(); got != nil {
		t.Errorf("heard %v before the first read ended", got)
	}
	c.RunningRead()
	c.NoteRunning(nodeengine.Running{Node: "n2", Module: "lemurs", Fingerprint: "old", At: at.Add(-time.Hour)})
	c.NoteRunning(nodeengine.Running{Node: "n3", Module: "lemurs", Stopped: true, At: at})
	if got := c.heardNow(); len(got) != 2 || got["n2"]["lemurs"].Fingerprint != "a" || !got["n3"]["lemurs"].Stopped {
		t.Errorf("heard %v, want n2's lemurs at a (not the older) and n3's stopped", got)
	}
	c.RunningLost()
	if got := c.heardNow(); got != nil {
		t.Errorf("heard %v with the stream lost", got)
	}
	// A pass that changes nothing writes nothing.
	setFor(t, &askedWithin, 0)
	if _, err := c.ComposeNow(ctx); err != nil {
		t.Fatal(err)
	}
	time.Sleep(50 * time.Millisecond)
	if got := state.written(); len(got) != 2 {
		t.Errorf("a pass that changed nothing wrote %v", got[2:])
	}
	// What is kept: the five keys of a module written last, and one a want names; the rest is deleted.
	keep := New(b.env())
	keep.pubKept = map[string]ComposedEntry{}
	var keys []string
	for i, ch := range "123456" {
		key, _ := engcomp.Key("n1", "lemurs", meshcatalogclient.SourcePrefix2+strings.Repeat(string(ch), 64))
		keys = append(keys, key)
		keep.pubKept[key] = ComposedEntry{Revision: uint64(10 + i)}
	}
	keep.pubKept["n1.node"] = ComposedEntry{Revision: 1}
	if past := keep.pastKeeping(); !slices.Equal(past, []string{keys[0]}) {
		t.Errorf("past keeping: %v, want the oldest of six, %s", past, keys[0])
	}
	keep.wants = map[string]meshwants.WantsByWantNodeModuleValue{"lemurs": {Fp: meshcatalogclient.SourcePrefix2 + strings.Repeat("1", 64)}}
	if past := keep.pastKeeping(); len(past) != 0 {
		t.Errorf("past keeping with a want naming the oldest: %v, want none", past)
	}

	brokenKey, _ := engcomp.Key("n1", "broken", bfp)
	// Design 47 G A-293: reading the catalog half by fingerprint, a placed module's newest build that ended is the last trunk
	// build.ended heard of it; a failed one is said refused with the event's words once the first read ended, an end naming a
	// branch moves nothing, a module with no end (queued or running) stays unpublished, and a later done end, at any
	// fingerprint, deletes the refused key.
	byEnv := b.env()
	byEnv.Entry = EntryCall(func(context.Context, string, string, any) (json.RawMessage, error) {
		return nil, errors.New("nothing is asked")
	})
	byState := &fakeComposed{kv: map[string]ComposedEntry{}}
	byEnv.ComposedState = byState
	by := New(byEnv)
	byAt := &published{fps: map[string]string{}, placed: []string{"broken", "waiting", "branched"}}
	words := `refused kind: package of language "go"`
	by.NoteBuildEnded(builds.BuildEnded{Module: "broken", Fingerprint: bfp, Outcome: builds.BuildFailed, Error: words})
	by.NoteBuildEnded(builds.BuildEnded{Module: "branched", Fingerprint: wfp, Outcome: builds.BuildFailed, Error: "no", Branch: "feature"})
	for _, step := range []struct {
		name  string
		do    func()
		holds map[string]string // by key, the refusal it says
	}{
		{"before the first read ended", func() {}, map[string]string{}},
		{"once the first read ended", by.BuildEndsRead, map[string]string{brokenKey: words}},
		{"a later done end", func() {
			by.NoteBuildEnded(builds.BuildEnded{Module: "broken", Fingerprint: wfp, Outcome: builds.BuildDone})
		}, map[string]string{}},
	} {
		step.do()
		if _, _, err := by.writeComposed(ctx, byAt); err != nil {
			t.Fatalf("%s: %v", step.name, err)
		}
		byState.mu.Lock()
		got := map[string]string{}
		for k, v := range byState.kv {
			comp, _ := engcomp.Decode(v.Value)
			got[k] = comp.Refused
		}
		byState.mu.Unlock()
		if !maps.Equal(got, step.holds) {
			t.Errorf("%s: the state holds %v, want %v", step.name, got, step.holds)
		}
	}
}