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"
	"git.novox.be/novox/mesh-sdk/go/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
	setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, error) {
		asked++
		return items, nil, itemsErr
	})
	// 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{Node: "n1", Data: store})
	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
		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},
	} {
		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 = step.items, step.refused
			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)
				}
			}
			if store.kv["n2.web"] == nil {
				t.Error("another node's key was deleted")
			}
		})
	}
	// 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)
	}
}