mesh-host/internal/selfcompose/data_test.go

package selfcompose

import (
	"context"
	"errors"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/novox/mesh-host/internal/compose/frombus"
	"git.novox.be/novox/mesh-sdk/go/delivery"

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

// fakeData is the engine's state data as a map, counting the writes and deletes made.
type fakeData struct {
	mu      sync.Mutex
	kv      map[string][]byte
	puts    []string
	deletes []string
	fail    error
}

func (f *fakeData) Get(_ context.Context, key string) ([]byte, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.kv[key], nil
}

func (f *fakeData) Put(_ context.Context, key string, value []byte) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.fail != nil {
		return f.fail
	}
	f.kv[key] = value
	f.puts = append(f.puts, key)
	return nil
}

func (f *fakeData) 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 *fakeData) Keys(_ context.Context, pattern string) ([]string, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	var out []string
	for k := range f.kv {
		if matches(pattern, k) {
			out = append(out, k)
		}
	}
	return out, nil
}

// ADR 0577 clause 1 (and its check in clause 5): after a pass that composed something new, the engine keeps each placed
// module's data items in its state data, key `<node>.<module>`, with its composition's digest: written when the module's
// items change, not written when they do not, and deleted when the module leaves the node; a key of another node is never
// touched, a pass that handed nothing new reads nothing, and items that could not be worked out or written are tried
// again with no new composition, after the composer's growing bound (the loop's owed pass).
func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
	item := func(module, id, path string) catalogue.PlacedDataItem {
		return catalogue.PlacedDataItem{Module: module, ID: id, Class: "valuable", Path: path, Measure: catalogue.MeasureWalk}
	}
	store := &fakeData{kv: map[string][]byte{
		// Left by a module that went while the engine was down, and another node's key.
		"n1.gone": []byte(`{"v":1}`), "n2.web": []byte(`{"v":1}`)}}
	var items map[string][]catalogue.PlacedDataItem
	var asked int
	var itemsErr error
	var heldNow map[string]string
	setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, map[string]string, error) {
		asked++
		return items, nil, heldNow, itemsErr
	})
	var said []string
	env := Env{Node: "n1", Data: store, Say: func(s string) { said = append(said, s) }}
	// The engine's grant lets it read its own node's keys, as the listing at a run's first pass does, and no other's.
	if !grantAllows(t, "n1", delivery.EngineDataBucket, "n1.*") || grantAllows(t, "n1", delivery.EngineDataBucket, "n2.*") {
		t.Error("the node's grant reads other keys of the engine's state data than its own")
	}
	c := New(env)
	webKey, _ := delivery.ComposedKey("n1", "web", delivery.SourcePrefix2+strings.Repeat("a", 64))
	nodeKey, _ := delivery.ComposedNodeKey("n1")
	for _, step := range []struct {
		name    string
		due     bool
		items   map[string][]catalogue.PlacedDataItem
		digests map[string]string // composed key to digest handed
		fail    error
		refused error             // what working out the items says
		held    map[string]string // modules held for a refused manifest, with why (ADR 0578 §3)
		owed    time.Duration     // the bound the loop waits before trying again; 0 when nothing is owed
		puts    []string
		deletes []string
		asked   int
		// want is each key's items and digest after the step.
		want map[string]delivery.EngineData
	}{
		{name: "the run's first pass writes each module's items and deletes what left", due: true,
			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}, "bare": nil},
			digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"},
			puts:    []string{"n1.oldie", "n1.web"}, deletes: []string{"n1.gone"}, asked: 1,
			want: map[string]delivery.EngineData{
				"n1.web":   {Module: "web", Digest: "sha256:w1", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")}},
				"n1.oldie": {Module: "oldie", Digest: "sha256:n1", Items: []catalogue.PlacedDataItem{item("oldie", "x", "/srv/oldie")}}}},
		{name: "a pass that handed nothing new reads nothing", due: false, asked: 1},
		{name: "a pass that changes no module's items writes nothing", due: true,
			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
			digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"}, asked: 2},
		{name: "items that cannot be worked out are tried again after the bound", due: true, refused: errors.New("no answer"),
			digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 3, owed: LaterWait(0)},
		{name: "a write that fails is tried again with no new composition, after a longer bound", due: false, fail: errors.New("the bus said no"),
			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
			digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 4, owed: LaterWait(1)},
		{name: "a module whose composition changed its items is written again", due: false,
			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
			digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 5,
			want: map[string]delivery.EngineData{
				"n1.web": {Module: "web", Digest: "sha256:w2", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")}}}},
		{name: "a module that left the node is deleted", due: true,
			items:   map[string][]catalogue.PlacedDataItem{"oldie": {item("oldie", "x", "/srv/oldie")}},
			digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 6},
		// ADR 0578 §3: one refused manifest never ends the read; its module is named, its key stands, the rest is kept.
		{name: "a module held for a refused manifest is named, its key stands and every other module's items are written", due: true,
			items:   map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}},
			held:    map[string]string{"oldie": "the manifest of oldie is refused: a bar item is no icon"},
			digests: map[string]string{webKey: "sha256:w3", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 7,
			want: map[string]delivery.EngineData{
				"n1.web": {Module: "web", Digest: "sha256:w3", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")}}}},
	} {
		t.Run(step.name, func(t *testing.T) {
			store.mu.Lock()
			store.puts, store.deletes, store.fail = nil, nil, step.fail
			store.mu.Unlock()
			items, itemsErr, heldNow, said = step.items, step.refused, step.held, nil
			c.paid() // the pass before composed: nothing owed
			c.mu.Lock()
			c.handed = step.digests
			c.dataDue = c.dataDue || step.due
			c.mu.Unlock()
			c.publishData(context.Background())
			slices.Sort(store.puts)
			if !slices.Equal(store.puts, step.puts) || !slices.Equal(store.deletes, step.deletes) || asked != step.asked {
				t.Fatalf("wrote %v and deleted %v, items asked %d times; want %v, %v, %d", store.puts, store.deletes, asked,
					step.puts, step.deletes, step.asked)
			}
			if wait, _ := c.waitNow(); c.owed.Load() != (step.owed > 0) || wait != step.owed {
				t.Errorf("owed a pass: %v after %s; want after %s", c.owed.Load(), wait, step.owed)
			}
			for key, want := range step.want {
				got, err := delivery.DecodeEngineData(store.kv[key])
				if err != nil || got.Node != "n1" || got.Module != want.Module || got.Digest != want.Digest ||
					!slices.Equal(got.Items, want.Items) {
					t.Errorf("%s holds %+v (%v), want %+v", key, got, err, want)
				}
			}
			for module, why := range step.held {
				if !slices.ContainsFunc(said, func(l string) bool { return strings.Contains(l, module) && strings.Contains(l, why) }) {
					t.Errorf("the held %s is not named with why in %q", module, said)
				}
				if store.kv["n1."+module] == nil {
					t.Errorf("the key of the held %s was deleted", module)
				}
			}
			if store.kv["n2.web"] == nil {
				t.Error("another node's key was deleted")
			}
		})
	}
	// ADR 0565 §6 (1953): the node's key of the engine's state packages is written after a pass that handed something new
	// when the packages its compositions declare differ, and only then: a composition adding a package writes it once,
	// one that changes no package writes nothing; a write refused is owed after the growing bound; the verb's write reads
	// the key again, so a key lost on the bus is written back. Another node's key is never touched.
	pkgStore := &fakeData{kv: map[string][]byte{"n2": []byte(`{"node":"n2","packages":[]}`)}}
	pc := New(Env{Node: "n1", Packages: pkgStore})
	web := func(pkgs ...string) delivery.Composed {
		comp := delivery.Composed{Node: "n1", Module: "web"}
		for _, p := range pkgs {
			comp.Resources = append(comp.Resources, []byte(`{"id":"`+p+`","type":"package","package":"`+p+`"}`))
		}
		comp.Resources = append(comp.Resources, []byte(`{"id":"conf","type":"file","path":"/etc/web.conf"}`))
		return packagesOf(comp)
	}
	for _, step := range []struct {
		name  string
		comp  delivery.Composed
		fail  error
		puts  int
		owed  bool
		holds string
	}{
		{name: "the run's first pass writes the key", comp: web("curl"), puts: 1,
			holds: `{"node":"n1","packages":[{"package":"curl","modules":["web"],"absent":false}]}`},
		{name: "a pass that changes no package writes nothing", comp: web("curl"), puts: 1},
		{name: "a refused write is owed after the bound", comp: web("curl", "jq"), fail: errors.New("no grant yet"), puts: 1, owed: true},
		{name: "a composition adding a package writes the key once", comp: web("curl", "jq"), puts: 2,
			holds: `{"node":"n1","packages":[{"package":"curl","modules":["web"],"absent":false},{"package":"jq","modules":["web"],"absent":false}]}`},
	} {
		pkgStore.mu.Lock()
		pkgStore.fail = step.fail
		pkgStore.mu.Unlock()
		pc.paid()
		pc.mu.Lock()
		pc.pkgs, pc.pkgsComposed, pc.pkgsDue = map[string]delivery.Composed{webKey: step.comp}, true, true
		pc.mu.Unlock()
		pc.publishPackages(context.Background())
		if len(pkgStore.puts) != step.puts || pc.owed.Load() != step.owed ||
			(step.holds != "" && string(pkgStore.kv["n1"]) != step.holds) || string(pkgStore.kv["n2"]) != `{"node":"n2","packages":[]}` {
			t.Errorf("%s: %d writes, owed %v, n1 holds %s", step.name, len(pkgStore.puts), pc.owed.Load(), pkgStore.kv["n1"])
		}
	}
	delete(pkgStore.kv, "n1")
	if wrote, err := pc.WritePackages(context.Background()); !wrote || err != nil || pkgStore.kv["n1"] == nil {
		t.Errorf("the verb's write of a lost key wrote %v (%v)", wrote, err)
	}
	if wrote, _ := pc.WritePackages(context.Background()); wrote {
		t.Error("the verb's write of a key holding what was composed wrote it again")
	}

	// The items are worked out from what the pass read: a done build is read over the bus once, not again for the data
	// items after the pass (a module in a running build is read again, since it may land).
	b := newFakeBus()
	done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: delivery.ManifestDigest([]byte(lemurs))})
	running, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: "building"})
	fp := delivery.SourcePrefix2 + strings.Repeat("b", 64)
	doneKey, _ := delivery.BuilderBuildKey("web", fp)
	runningKey, _ := delivery.BuilderBuildKey("pg", fp)
	b.kv[delivery.BuilderBuildsBucket+"/"+doneKey], b.kv[delivery.BuilderBuildsBucket+"/"+runningKey] = done, running
	r := reader{c: New(b.env())}
	for range 3 {
		if got, err := r.Build(context.Background(), "web", fp); err != nil || got.State != delivery.BuildDone {
			t.Fatalf("the done build: %+v, %v", got, err)
		}
		if got, err := r.Build(context.Background(), "pg", fp); err != nil || got.State != "building" {
			t.Fatalf("the running build: %+v, %v", got, err)
		}
	}
	reads := map[string]int{}
	for _, read := range b.reads {
		reads[read[1]]++
	}
	if reads[doneKey] != 1 || reads[runningKey] != 3 {
		t.Errorf("the builds were read %v times over the bus; want the done one once, the running one each time", reads)
	}
	// A done key the builder claims again (a gone step, issue 1937) and lands done again is read again, not served kept.
	again, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: delivery.ManifestDigest([]byte("{}"))})
	b.mu.Lock()
	b.kv[delivery.BuilderBuildsBucket+"/"+doneKey] = again
	b.mu.Unlock()
	r.c.takeBuildDone(agent.Update{Key: doneKey, Value: running})
	if got, err := r.Build(context.Background(), "web", fp); err != nil || got.Manifest != delivery.ManifestDigest([]byte("{}")) {
		t.Errorf("a done build written again is served as %+v (%v), not read again", got, err)
	}
}