mesh-controller/cmd/mesh-controller/record_mirrors_test.go

package main

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/compose"
)

// The declared-packages mirror composes only when what it composes from changed (issue 1945 note 22: it composed every node
// in full every 10 s, a pass of about 45 s, so the controller on the control node never rested at 130-175 % CPU).

// scaled is a cadence of the mirror with a second taken as a millisecond, so a test's minute is 60 ms.
func scaled() packagesTiming {
	return packagesTiming{safety: packagesSafetyEvery / 1000, gap: packagesGap / 1000}
}

const scaledMinute = 60 * time.Millisecond

// fakeMesh is the keeper's input as a test changes it: each node's assigned modules, and the compositions counted by node.
type fakeMesh struct {
	mu       sync.Mutex
	assigned map[string][]string
	composed atomic.Int64 // node compositions
	synced   atomic.Int64
	lastSync map[string][]byte
	// block, when set, holds the next composition until it is closed; entered is closed when that composition began.
	block   chan struct{}
	entered chan struct{}
}

func newFakeMesh(nodes int) *fakeMesh {
	f := &fakeMesh{assigned: map[string][]string{}}
	for i := 0; i < nodes; i++ {
		f.assigned[fmt.Sprintf("node%d", i)] = []string{"base"}
	}
	return f
}

func (f *fakeMesh) assign(node, module string) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.assigned[node] = append(append([]string(nil), f.assigned[node]...), module)
}

func (f *fakeMesh) keeper() *packagesKeeper {
	read := func(context.Context) (*meshInput, error) {
		f.mu.Lock()
		defer f.mu.Unlock()
		in := compose.Input{Catalogue: map[string]catalogue.Manifest{"base": {Module: "base", Version: "1"}}}
		for _, n := range nodesOfFake(f.assigned) {
			in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: append([]string(nil), f.assigned[n]...)})
		}
		return &meshInput{Input: in}, nil
	}
	composeAll := func(_ context.Context, in *meshInput) (map[string][]byte, error) {
		f.mu.Lock()
		block, entered := f.block, f.entered
		f.block = nil
		f.mu.Unlock()
		if block != nil {
			close(entered)
			<-block
		}
		out := map[string][]byte{}
		for _, n := range in.Nodes {
			f.composed.Add(1)
			out[n.Name] = []byte(fmt.Sprint(n.Assigned))
		}
		return out, nil
	}
	syncAll := func(_ context.Context, want map[string][]byte) error {
		f.mu.Lock()
		f.lastSync = want
		f.mu.Unlock()
		f.synced.Add(1)
		return nil
	}
	return newPackagesKeeper(read, composeAll, syncAll, func() bool { return true })
}

func nodesOfFake(m map[string][]string) []string {
	out := make([]string, 0, len(m))
	for k := range m {
		out = append(out, k)
	}
	for i := range out {
		for j := i + 1; j < len(out); j++ {
			if out[j] < out[i] {
				out[i], out[j] = out[j], out[i]
			}
		}
	}
	return out
}

func runKeeper(t *testing.T, k *packagesKeeper) context.CancelFunc {
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		defer close(done)
		k.run(ctx, scaled(), func(string) {})
	}()
	t.Cleanup(func() {
		cancel()
		<-done
	})
	return cancel
}

func eventually(t *testing.T, what string, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(time.Second)
	for !cond() {
		if time.Now().After(deadline) {
			t.Fatalf("not within a second: %s", what)
		}
		time.Sleep(time.Millisecond)
	}
}

// 1945/AC-mirror-1: given a mesh whose input does not change, when a minute passes after the mirror's first pass, then it
// composes no node (before: a pass every 10 s, each composing every node). The count per minute is said.
func TestTheDeclaredPackagesMirrorComposesNothingWithoutAChange(t *testing.T) {
	const nodes = 5
	f := newFakeMesh(nodes)
	runKeeper(t, f.keeper())
	eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
	start := f.composed.Load()
	time.Sleep(scaledMinute)
	perMinute := f.composed.Load() - start
	t.Logf("node compositions in a minute of an unchanged mesh of %d nodes: %d", nodes, perMinute)
	if perMinute != 0 {
		t.Errorf("an unchanged mesh of %d nodes was composed %d times in a minute; want 0", nodes, perMinute)
	}
}

// 1945/AC-mirror-2: given a pass that is running, when fifty changes arrive during it, then exactly one more pass follows,
// never two at once and never one per change.
func TestTheDeclaredPackagesMirrorPassesOncePerBurst(t *testing.T) {
	f := newFakeMesh(1)
	release := make(chan struct{})
	f.block, f.entered = release, make(chan struct{})
	k := f.keeper()
	runKeeper(t, k)
	<-f.entered
	for i := 0; i < 50; i++ {
		f.assign("node0", fmt.Sprintf("m%d", i))
		k.poke()
	}
	close(release)
	eventually(t, "the pass after the burst", func() bool { return f.synced.Load() >= 2 })
	time.Sleep(2*scaled().gap + 20*time.Millisecond)
	if got := f.synced.Load(); got != 2 {
		t.Errorf("a burst of 50 changes during a pass gave %d passes in all; want 2 (the running one and one after)", got)
	}
}

// 1945/AC-mirror-3: given the mirror at rest, when an input changes and its event arrives, then the change is mirrored.
func TestTheDeclaredPackagesMirrorMirrorsAChange(t *testing.T) {
	f := newFakeMesh(2)
	k := f.keeper()
	runKeeper(t, k)
	eventually(t, "the first pass", func() bool { return f.synced.Load() >= 1 })
	f.assign("node1", "viewer")
	k.poke()
	eventually(t, "the change mirrored", func() bool {
		f.mu.Lock()
		defer f.mu.Unlock()
		return string(f.lastSync["node1"]) == "[base viewer]"
	})
}

// 1945/AC-mirror-4: the digest of the input is the same for the same input and differs when an assignment differs.
func TestTheInputDigestFollowsTheInput(t *testing.T) {
	of := func(assigned ...string) *meshInput {
		return &meshInput{Input: compose.Input{Nodes: []compose.Node{{Name: "anchor", Assigned: assigned}}}}
	}
	a, b, c := inputDigest(of("base")), inputDigest(of("base")), inputDigest(of("base", "viewer"))
	if a == "" || a != b {
		t.Errorf("the same input gave the digests %q and %q", a, b)
	}
	if a == c {
		t.Errorf("an added assignment left the digest at %q", a)
	}
}