platform/projection/manager_test.go

package projection

import (
	"fmt"
	"path/filepath"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

// TestSnapshotHolder_SwapRace storms the holder: 100 readers Get in tight loops
// while a rotator Reloads a fresh snapshot every 50ms. It asserts the holder's
// safety + liveness invariants under concurrency (run with -race):
//
//   - No panic / SIGBUS — completion is the proof the drain grace never unmaps a
//     reader while an in-flight Get is reading it.
//   - No torn reads — every hit decodes to a valid value (SKU matches, price > 0).
//   - No gap — since every snapshot version contains every key, a Get must never
//     miss; misses would mean the holder briefly exposed an empty/nil reader.
func TestSnapshotHolder_SwapRace(t *testing.T) {
	dir := t.TempDir()

	const versions = 4
	const skuCount = 200
	paths := make([]string, versions)
	for v := 0; v < versions; v++ {
		w := NewSnapshotWriter[tProj]()
		for i := 0; i < skuCount; i++ {
			sku := fmt.Sprintf("SKU-%03d", i)
			// price encodes the version (always > 0) so a torn read is detectable.
			if err := w.Append(sku, tProj{ID: "sku:" + sku, SKU: sku, PriceIncVat: int64((v + 1) * 100)}); err != nil {
				t.Fatal(err)
			}
		}
		p, err := w.Write(dir, fmt.Sprintf("snap-%d.bin", v))
		if err != nil {
			t.Fatal(err)
		}
		paths[v] = p
	}

	h := NewSnapshotHolder[tProj](100 * time.Millisecond)
	if err := h.Open(paths[0]); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = h.Close() })

	stop := make(chan struct{})
	var reads, misses int64
	var torn int32
	var wg sync.WaitGroup

	for r := 0; r < 100; r++ {
		wg.Add(1)
		go func(seed int) {
			defer wg.Done()
			i := seed
			for {
				select {
				case <-stop:
					return
				default:
				}
				sku := fmt.Sprintf("SKU-%03d", i%skuCount)
				if p, ok := h.Get(sku); ok {
					if p.SKU != sku || p.PriceIncVat <= 0 {
						atomic.StoreInt32(&torn, 1)
						return
					}
					atomic.AddInt64(&reads, 1)
				} else {
					atomic.AddInt64(&misses, 1)
				}
				i++
			}
		}(r)
	}

	wg.Add(1)
	go func() {
		defer wg.Done()
		tick := time.NewTicker(50 * time.Millisecond)
		defer tick.Stop()
		v := 0
		for {
			select {
			case <-stop:
				return
			case <-tick.C:
				v = (v + 1) % versions
				if err := h.Reload(paths[v]); err != nil {
					atomic.StoreInt32(&torn, 1)
					return
				}
			}
		}
	}()

	time.Sleep(1500 * time.Millisecond)
	close(stop)
	wg.Wait()

	if atomic.LoadInt32(&torn) != 0 {
		t.Fatal("torn/garbage read or reload error during swap storm")
	}
	if m := atomic.LoadInt64(&misses); m != 0 {
		t.Fatalf("got %d misses — holder exposed an empty reader mid-swap (every version has every key)", m)
	}
	if atomic.LoadInt64(&reads) == 0 {
		t.Fatal("no successful reads — test did not exercise the holder")
	}
	t.Logf("SwapRace OK: %d reads across %d rotations, 0 torn, 0 miss", atomic.LoadInt64(&reads), versions)
}

// fakeCloser counts how many times it's been called. Used to assert the
// drain-grace semantics of SnapshotHolder.
type fakeCloser struct{ hits int }

func (f *fakeCloser) Close() error { f.hits++; return nil }

// withFakeCloser swaps SnapshotReader's closer for our counter.
func withFakeCloser(t *testing.T, path string, fc *fakeCloser) *SnapshotReader[tProj] {
	t.Helper()
	r, err := ReadSnapshotFile[tProj](path)
	if err != nil {
		t.Fatal(err)
	}
	// replace real closer with our counter; the test asserts only via fc
	r.raw.closer = fc.Close
	return r
}

// TestSnapshotHolderReloadDrainsOld — Set swaps the current reader; the
// previous reader is queued. Before grace: still Get-able. After grace:
// reaped and Close fires.
func TestSnapshotHolderReloadDrainsOld(t *testing.T) {
	dir := t.TempDir()

	// 1) Build a valid first snapshot (keys A-1 / A-2).
	w := NewSnapshotWriter[tProj]()
	if err := appendP(w, makeProjection("A-1", 100)); err != nil {
		t.Fatal(err)
	}
	if err := appendP(w, makeProjection("A-2", 200)); err != nil {
		t.Fatal(err)
	}
	if _, err := w.Write(dir, "first.bin"); err != nil {
		t.Fatal(err)
	}
	firstCloser := &fakeCloser{}
	first := withFakeCloser(t, filepath.Join(dir, "first.bin"), firstCloser)
	defer first.Close()

	// 2) Build a second snapshot (key B-1).
	w2 := NewSnapshotWriter[tProj]()
	if err := appendP(w2, makeProjection("B-1", 300)); err != nil {
		t.Fatal(err)
	}
	if _, err := w2.Write(dir, "second.bin"); err != nil {
		t.Fatal(err)
	}
	secondCloser := &fakeCloser{}
	second := withFakeCloser(t, filepath.Join(dir, "second.bin"), secondCloser)
	defer second.Close()

	h := NewSnapshotHolder[tProj](50 * time.Millisecond)
	defer h.Close()
	h.Set(first)
	if h.Len() != 2 {
		t.Errorf("Len after first Set: got %d want 2", h.Len())
	}
	if _, ok := h.Get("A-1"); !ok {
		t.Error("A-1 not found in initial state")
	}

	// 3) Swap to second — first is now queued.
	h.Set(second)
	if h.Len() != 1 {
		t.Errorf("Len after second Set: got %d want 1", h.Len())
	}
	if firstCloser.hits != 0 {
		t.Errorf("firstCloser hits: got %d want 0 (still in drain grace)", firstCloser.hits)
	}
	if h.DrainCount() != 1 {
		t.Errorf("DrainCount: got %d want 1", h.DrainCount())
	}

	// 4) Wait past grace, force a reap tick.
	time.Sleep(80 * time.Millisecond)
	h.reapNow()
	if firstCloser.hits != 1 {
		t.Errorf("firstCloser hits: got %d want 1 (should have reaped)", firstCloser.hits)
	}
	if h.DrainCount() != 0 {
		t.Errorf("DrainCount after reap: got %d want 0", h.DrainCount())
	}
}

// TestSnapshotHolderEmptyGet — holder with no current serves empty.
func TestSnapshotHolderEmptyGet(t *testing.T) {
	h := NewSnapshotHolder[tProj](100 * time.Millisecond)
	defer h.Close()
	if _, ok := h.Get("anything"); ok {
		t.Error("empty holder should return false")
	}
	if h.Len() != 0 {
		t.Errorf("Len: got %d want 0", h.Len())
	}
}

// TestSnapshotHolderCloseImmediatelyReaps — Close with pending drains
// should close them now (no grace wait).
func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) {
	dir := t.TempDir()

	w := NewSnapshotWriter[tProj]()
	if err := appendP(w, makeProjection("X-1", 1)); err != nil {
		t.Fatal(err)
	}
	if _, err := w.Write(dir, "x.bin"); err != nil {
		t.Fatal(err)
	}
	fc1 := &fakeCloser{}
	r1 := withFakeCloser(t, filepath.Join(dir, "x.bin"), fc1)

	w2 := NewSnapshotWriter[tProj]()
	if err := appendP(w2, makeProjection("Y-1", 2)); err != nil {
		t.Fatal(err)
	}
	if _, err := w2.Write(dir, "y.bin"); err != nil {
		t.Fatal(err)
	}
	fc2 := &fakeCloser{}
	r2 := withFakeCloser(t, filepath.Join(dir, "y.bin"), fc2)

	h := NewSnapshotHolder[tProj](10 * time.Second) // long grace so reap wouldn't happen on its own
	h.Set(r1)
	h.Set(r2)

	if fc1.hits != 0 || fc2.hits != 0 {
		t.Errorf("premature Close hits: fc1=%d fc2=%d", fc1.hits, fc2.hits)
	}
	if err := h.Close(); err != nil {
		t.Errorf("Close: %v", err)
	}
	if fc1.hits != 1 || fc2.hits != 1 {
		t.Errorf("after Close: fc1=%d fc2=%d want 1 each", fc1.hits, fc2.hits)
	}
}

// TestSnapshotHolderIdentityReload — Open the same path twice; current reader
// is replaced; DrainCount returns 1 (old queued).
func TestSnapshotHolderIdentityReload(t *testing.T) {
	dir := t.TempDir()
	w := NewSnapshotWriter[tProj]()
	if err := appendP(w, makeProjection("ID-1", 1)); err != nil {
		t.Fatal(err)
	}
	if _, err := w.Write(dir, "id.bin"); err != nil {
		t.Fatal(err)
	}

	h := NewSnapshotHolder[tProj](10 * time.Second)
	defer h.Close()
	if err := h.Reload(filepath.Join(dir, "id.bin")); err != nil {
		t.Fatal(err)
	}
	if err := h.Reload(filepath.Join(dir, "id.bin")); err != nil {
		t.Fatal(err)
	}
	if h.DrainCount() != 1 {
		t.Errorf("DrainCount after two reloads of same path: got %d want 1", h.DrainCount())
	}
	if _, ok := h.Get("ID-1"); !ok {
		t.Error("current reader did not surface ID-1")
	}
}