mesh-host/internal/selfcompose/entries_test.go

package selfcompose

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"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/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/agent"
	"github.com/novox/mesh-host/internal/buckets"
	"github.com/novox/mesh-host/internal/buckets/bucketstest"
	"github.com/novox/mesh-host/internal/compositions"
)

// Design 47 G §421, §683, §698 (the lead's answers A-47, A-50): the node-engine reads the entry its want names by the
// fingerprint, through its own calls of the seat mesh-catalog's verb entry, and reads neither the indexer's states nor the
// builder's; a want whose entry the catalog half holds none of is held, its have saying entry-missing and naming the
// fingerprint; the entries and wants it composes from are kept in its composition state, written before the composition that
// uses them and read at its start before anything is asked; a want removed takes its kept want and entry away.
func composeFromEntries(t *testing.T) {
	b := newFakeBus()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64)
	ghost := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64)
	var manifest map[string]any
	if err := json.Unmarshal([]byte(lemurs), &manifest); err != nil {
		t.Fatal(err)
	}
	entry := meshcatalog.CatalogEntry{Module: "lemurs", Fp: fp, Manifest: manifest, Repository: "novox/x", Commit: "c",
		ManifestDigest: meshcatalogclient.ManifestDigest([]byte(lemurs))}
	var mu sync.Mutex
	var asked []string
	ask := func(_ context.Context, seat, verb string, args any) (json.RawMessage, error) {
		in := args.(meshcatalog.EntryInput)
		mu.Lock()
		asked = append(asked, in.Fingerprint)
		mu.Unlock()
		if seat != "mesh-catalog" || verb != "entry" {
			t.Errorf("asked %s.%s", seat, verb)
		}
		if in.Fingerprint != fp {
			return nil, errors.New("no entry of fingerprint " + in.Fingerprint + " is kept")
		}
		return json.Marshal(meshcatalog.EntryAnswer{Entry: entry})
	}
	state := &fakeComposed{kv: map[string]ComposedEntry{}}
	var held map[string]string
	env := b.env()
	env.Entry, env.ComposedState = EntryCall(ask), state
	refused := make(chan struct{}, 1) // fed at each refusal the composer says, so a wait follows what it says, not a pause
	env.Refused = func(_ string, h map[string]string) {
		mu.Lock()
		held = h
		mu.Unlock()
		select {
		case refused <- struct{}{}:
		default:
		}
	}
	told := map[string]bool{}
	env.EntryNotFound = func(fp string, notFound bool) { mu.Lock(); told[fp] = notFound; mu.Unlock() }
	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: "ghost"}}})
	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 module, w := range map[string]meshwants.WantsByWantNodeModuleValue{"lemurs": {Fp: fp, Self: true}, "ghost": {Fp: ghost, Self: true}} {
		raw, _ := bucketstest.Want(w)
		b.put(t, buckets.Wants, "want.n1."+module, raw)
	}
	for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} {
		b.send(bucket, agent.Update{Caught: true})
	}
	lemursKey, _ := compositions.Key("n1", "lemurs", fp)
	handed := map[string]bool{}
	for !handed[lemursKey] {
		handed[next(t, out).Key] = true
	}
	// Neither the indexer's states nor the builder's are watched or read.
	b.mu.Lock()
	for _, r := range b.reads {
		if r[0] == buckets.Modules || r[0] == buckets.Manifests || r[0] == buckets.Verdicts || strings.HasSuffix(r[0], "_builds") {
			t.Errorf("the node-engine read %s of %s", r[1], r[0])
		}
	}
	b.mu.Unlock()
	// A want whose entry the half holds none of is held, saying entry-missing and the fingerprint; the rest is composed.
	// The ghost's entry is asked first and answered not found after: wait until the refusal says entry-missing.
	var why string
	for deadline := time.After(10 * time.Second); !strings.HasPrefix(why, agent.EntryMissing); {
		mu.Lock()
		why = held["ghost"]
		mu.Unlock()
		if strings.HasPrefix(why, agent.EntryMissing) {
			break
		}
		select {
		case <-refused:
		case <-deadline:
			t.Fatalf("the ghost's reason is still %q after 10 s, want %s", why, agent.EntryMissing)
		}
	}
	if !strings.HasPrefix(why, agent.EntryMissing) || !strings.Contains(why, ghost) || held["lemurs"] != "" {
		t.Errorf("held %q, want ghost held with %s naming %s", held, agent.EntryMissing, ghost)
	}
	// And the finding entry-not-found is told for the fingerprint the half holds none of, and for no other (design 47 G §614, A-117).
	mu.Lock()
	nf, nfOther := told[ghost], told[fp]
	mu.Unlock()
	if !nf || nfOther {
		t.Errorf("entry-not-found told %v for the ghost and %v for the entry held", nf, nfOther)
	}
	// An entry asked again by hand is asked again in the shadow's cache too (lead answer A-319): the shadow holds the ghost as not
	// found, the half then holds it, and the node's own composer asked again takes it in on the shadow as well.
	var late atomic.Bool
	shadowEnv := b.env()
	shadowEnv.Entry = EntryCall(func(ctx context.Context, seat, verb string, args any) (json.RawMessage, error) {
		if in := args.(meshcatalog.EntryInput); in.Fingerprint == ghost && late.Load() {
			return json.Marshal(meshcatalog.EntryAnswer{Entry: meshcatalog.CatalogEntry{Module: "ghost", Fp: ghost}})
		}
		return ask(ctx, seat, verb, args)
	})
	shadow := NewShadow(shadowEnv, nil)
	shadow.c.entries.Need(ctx, ghost)
	for deadline := time.Now().Add(10 * time.Second); shadow.c.entries.Rows() == nil || shadow.c.entries.Rows()[0].State != meshcatalogclient.EntryNotFound; {
		if time.Now().After(deadline) {
			t.Fatal("the shadow did not hold the ghost as not found")
		}
		runtime.Gosched()
	}
	late.Store(true)
	c.AlsoEntryAgain = shadow.EntryAgain
	if err := c.EntryAgain(ctx, ghost); err != nil {
		t.Fatal(err)
	}
	if _, state := shadow.c.entries.Need(ctx, ghost); state != meshcatalogclient.EntryAnswered {
		t.Errorf("the shadow holds the ghost's entry as %s after entry-again, want answered", state)
	}
	// Each entry and want it composes from is kept in the composition state before the composition that uses them.
	state.mu.Lock()
	_, hasEntry := state.kv[entryKey(fp)]
	_, hasWant := state.kv["want.lemurs"]
	state.mu.Unlock()
	if !hasEntry || !hasWant {
		t.Errorf("the composition state holds entry %v and want %v of lemurs: %v", hasEntry, hasWant, state.written())
	}
	// The seat's first-views lists them.
	_, entries, wants := c.FirstViews()
	if len(entries) < 2 || len(wants) != 2 {
		t.Errorf("first-views lists %v and %v", entries, wants)
	}
	// A restart reads what is kept before it asks anything: lemurs's entry is not asked again.
	mu.Lock()
	asked = nil
	mu.Unlock()
	env2 := b.env()
	env2.Entry, env2.ComposedState = EntryCall(ask), state
	// The node's own record, stated by the node-engine of itself in memory (A-120).
	env2.NodeRecord = func() (nodeengine.NodeRecord, bool) {
		return nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}, true
	}
	c2 := New(env2)
	out2, err := c2.Watch(ctx, "n1.>")
	if err != nil {
		t.Fatal(err)
	}
	// And composes the module it kept a want and an entry of before any input is read, nothing of the others (design 47 G §683, A-117).
	if u := next(t, out2); u.Key != lemursKey || u.Deleted {
		t.Errorf("the early pass handed %q, want the composition %s of lemurs", u.Key, lemursKey)
	}
	c2.mu.Lock()
	kept, keptWant := c2.keptEnt[fp], c2.wants["lemurs"].Fp
	c2.mu.Unlock()
	if kept.entry.Module != "lemurs" || keptWant != fp {
		t.Errorf("a restart read entry %q and want %q from the state", kept.entry.Module, keptWant)
	}
	// A want removed takes its kept want and entry away.
	raw, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Removed: true})
	b.put(t, buckets.Wants, "want.n1.lemurs", raw)
	for range 20 {
		state.mu.Lock()
		_, hasWant = state.kv["want.lemurs"]
		_, hasEntry = state.kv[entryKey(fp)]
		state.mu.Unlock()
		if !hasWant && !hasEntry {
			break
		}
		next(t, out)
	}
	if hasWant || hasEntry {
		t.Errorf("a removed want left its kept want %v and entry %v", hasWant, hasEntry)
	}
}

// takeModule lays one module of the catalog half's picture as the composer's own lay of an entry does (layModules): the tests
// that compose from a picture they set read no indexer.
func (c *Composer) takeModule(u agent.Update) {
	name, err := meshcatalogclient.ParseModulesKey(u.Key)
	if err != nil {
		return
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	m, err := meshcatalogclient.DecodeModule(u.Value)
	if u.Deleted || err != nil {
		delete(c.modules, name)
		return
	}
	c.modules[name] = m
}

// catalogHalf is the catalog half's entries a test composes from, by fingerprint, answered through the seat mesh-catalog's verbs
// entry and module (design 47 G §421, A-312): the node-engine reads no indexer.
type catalogHalf map[string]meshcatalog.CatalogEntry

// add holds the entry of module at fp, made of the manifest body.
func (h catalogHalf) add(t *testing.T, module, fp, body string) {
	t.Helper()
	var m map[string]any
	if err := json.Unmarshal([]byte(body), &m); err != nil {
		t.Fatal(err)
	}
	h[fp] = meshcatalog.CatalogEntry{Module: module, Fp: fp, Manifest: m, Repository: "novox/" + module, Commit: "c", ManifestDigest: meshcatalogclient.ManifestDigest([]byte(body))}
}

func (h catalogHalf) call() meshcatalogclient.Call {
	return EntryCall(func(_ context.Context, _, verb string, args any) (json.RawMessage, error) {
		switch in := args.(type) {
		case meshcatalog.ModuleInput:
			for _, e := range h {
				if e.Module == in.Module {
					return json.Marshal(meshcatalog.ModuleAnswer{Module: meshcatalog.ModuleAnswerModule{Module: e.Module, Repository: e.Repository, Commit: e.Commit, Fp: e.Fp}})
				}
			}
			return nil, fmt.Errorf("the catalog holds no module %s", in.Module)
		case meshcatalog.EntryInput:
			if e, ok := h[in.Fingerprint]; ok {
				return json.Marshal(meshcatalog.EntryAnswer{Entry: e})
			}
			return nil, errors.New("no entry of fingerprint " + in.Fingerprint + " is kept")
		}
		return nil, fmt.Errorf("asked %s of %T", verb, args)
	})
}

// holdEntry makes the composer hold the entry of module at fp, as its call of the verb entry would have answered it (§421), for the
// tests that set the rest of its picture directly.
func (c *Composer) holdEntry(t *testing.T, module, fp, body string) {
	t.Helper()
	half := catalogHalf{}
	half.add(t, module, fp, body)
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.keptEnt == nil {
		c.keptEnt = map[string]keptEntry{}
	}
	c.keptEnt[fp] = keptEntry{entry: half[fp]}
}