fancy-dns/internal/cluster/harness_test.go

package cluster

import (
	"context"
	"io"
	"log/slog"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/alicebob/miniredis/v2"

	"git.k8s-git-k6n.net/mats/fancy-dns/internal/config"
)

const (
	// testPeerTTL is short enough that a heartbeat lands every 100ms (the
	// maintain loop ticks at peerTTL/2) and a dead peer is reaped in
	// peerTTL*4, and long enough that a loaded machine does not make a live
	// peer look dead.
	testPeerTTL = 200 * time.Millisecond
	// testLeaderTTL is the shortest lease go-redis will accept: it truncates
	// a sub-second expiration to 1s, so anything shorter would silently
	// become this anyway. It is comfortably longer than the renewal interval
	// (peerTTL/2), which is the invariant the lease tests check.
	testLeaderTTL = time.Second
)

// harness owns one miniredis and one validated *config.Config for a test.
type harness struct {
	t     *testing.T
	mini  *miniredis.Miniredis
	cfg   *config.Config
	store *config.Store
}

// newHarness starts a miniredis for this test alone, so tests cannot interfere
// with each other, and builds a configuration that passes Validate.
func newHarness(t *testing.T, tweak ...func(*config.Config)) *harness {
	t.Helper()
	m := miniredis.RunT(t)
	return newHarnessOn(t, m, tweak...)
}

// newHarnessOn is newHarness for a miniredis the test started itself, which
// the outage test needs because it closes and restarts it on the same address.
func newHarnessOn(t *testing.T, m *miniredis.Miniredis, tweak ...func(*config.Config)) *harness {
	t.Helper()
	cfg := config.Default()
	// Validate rejects an enabled blocklist with no sources; these tests are
	// about coordination, not rules, so the blocklist is off.
	cfg.Block.Enabled = false
	cfg.Cluster.Enabled = true
	cfg.Cluster.PeerTTL = config.Duration(testPeerTTL)
	cfg.Cluster.Valkey = config.Valkey{
		Addrs:       []string{m.Addr()},
		Prefix:      testPrefix(t),
		PoolSize:    2,
		DialTimeout: config.Duration(500 * time.Millisecond),
		ReadTimeout: config.Duration(time.Second),
		LeaderTTL:   config.Duration(testLeaderTTL),
	}
	for _, fn := range tweak {
		fn(cfg)
	}
	if err := cfg.Validate(); err != nil {
		t.Fatalf("test config does not pass Validate: %v", err)
	}
	return &harness{t: t, mini: m, cfg: cfg, store: config.NewStore(cfg, "")}
}

// testPrefix is the Valkey key prefix for a test, so a failure names the test
// whose keys are in the dump.
func testPrefix(t *testing.T) string {
	r := strings.NewReplacer("/", "_", " ", "_", "#", "_")
	return "fdtest_" + r.Replace(t.Name())
}

// miniredisNew is miniredis.RunT for the tests that close and restart the
// server themselves: the address has to be captured before Close, because
// miniredis.Addr() panics once the server is gone.
func miniredisNew(t *testing.T) *miniredis.Miniredis {
	t.Helper()
	m := miniredis.NewMiniRedis()
	if err := m.Start(); err != nil {
		t.Fatalf("start miniredis: %v", err)
	}
	t.Cleanup(m.Close)
	return m
}

// keyValue reads a Valkey key through miniredis rather than through a client,
// so a test can assert what the server actually stored.
func keyValue(t *testing.T, m *miniredis.Miniredis, key string) string {
	t.Helper()
	v, err := m.Get(key)
	if err != nil {
		return ""
	}
	return v
}

type node struct {
	*Cluster
	t      *testing.T
	events chan Event
	cancel context.CancelFunc
}

// start builds and starts a Cluster against the harness' miniredis. The node
// is stopped and its context cancelled at the end of the test, so no test can
// leak a maintain goroutine into the next one.
func (h *harness) start(name string, opt func(*Options)) *node {
	h.t.Helper()
	o := Options{
		Config: h.store,
		Logger: discardLogger(),
		// The version is deliberately the same for every node: the identity
		// that matters is NodeID, and two nodes must not collide on it.
		Version:   "test",
		StartedAt: time.Now().Add(-time.Hour).Truncate(time.Second),
		Addr:      name + ".test:8080",
	}
	if opt != nil {
		opt(&o)
	}
	c := New(o)
	ctx, cancel := context.WithCancel(context.Background())
	n := &node{Cluster: c, t: h.t, events: make(chan Event, 64), cancel: cancel}
	// The handler must be in place before Start: the maintain goroutine
	// reads c.opt from another goroutine.
	if c.opt.Handler == nil {
		c.opt.Handler = func(_ context.Context, ev Event) {
			select {
			case n.events <- ev:
			default:
			}
		}
	}
	h.t.Cleanup(c.Stop)
	h.t.Cleanup(cancel)
	c.Start(ctx)
	return n
}

// kill ends this node's maintain session the way a crash does: the context is
// cancelled, so the node publishes no "bye" and its peers must notice through
// the heartbeat going stale instead of through a farewell.
func (n *node) kill() { n.cancel() }

// waitEvent returns the first captured event of the given kind, discarding the
// others (heartbeats from the peer, mostly).
func (n *node) waitEvent(kind EventKind, timeout time.Duration) Event {
	n.t.Helper()
	deadline := time.NewTimer(timeout)
	defer deadline.Stop()
	for {
		select {
		case ev := <-n.events:
			if ev.Kind == kind {
				return ev
			}
		case <-deadline.C:
			n.t.Fatalf("no %q event within %s", kind, timeout)
		}
	}
}

// countEvents drains the captured events and counts those of one kind.
func (n *node) countEvents(kind EventKind) int {
	c := 0
	for {
		select {
		case ev := <-n.events:
			if ev.Kind == kind {
				c++
			}
		default:
			return c
		}
	}
}

// hasPeer reports whether states contains id, without asserting anything
// about which other nodes are in the list.
func hasPeer(states []NodeState, id string) bool {
	for _, s := range states {
		if s.NodeID == id {
			return true
		}
	}
	return false
}

// leaderCount counts the nodes that currently claim the refresh lease.
func leaderCount(nodes ...*node) int {
	n := 0
	for _, nd := range nodes {
		if nd.IsLeader() {
			n++
		}
	}
	return n
}

func discardLogger() *slog.Logger {
	return slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError + 1}))
}

// waitFor polls cond until it holds, failing with what it waited for. The
// condition is an invariant of the cluster, never an incidental observation
// about what a goroutine happened to see.
func waitFor(t *testing.T, what string, timeout time.Duration, cond func() bool) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for {
		if cond() {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("timed out after %s waiting for %s", timeout, what)
		}
		time.Sleep(5 * time.Millisecond)
	}
}

// waitSubscribed blocks until miniredis counts a subscriber on every channel.
// This is the only race-free way to know a peer will hear a publish: pub/sub
// delivery to a channel with no subscriber is fire-and-forget.
func waitSubscribed(t *testing.T, m *miniredis.Miniredis, n *node) {
	t.Helper()
	chans := []string{
		n.channel(EventHeartbeat),
		n.channel(EventConfig),
		n.channel(EventReload),
		n.channel(EventBlocklists),
		n.channel(EventBye),
	}
	waitFor(t, "a live subscription on every cluster channel", 5*time.Second, func() bool {
		counts := m.PubSubNumSub(chans...)
		for _, ch := range chans {
			if counts[ch] < 1 {
				return false
			}
		}
		return true
	})
}

// digests is a race-free source of blocklist digests for one node, so a test
// can change what a node publishes and observe the effect.
type digests struct {
	mu        sync.Mutex
	blocklist string
	rules     int
	cfgDigest string
}

func (d *digests) set(blocklist string, rules int) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.blocklist, d.rules = blocklist, rules
}

func (d *digests) fn() (string, int, time.Time, string) {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.blocklist, d.rules, time.Unix(0, 0), d.cfgDigest
}