silo-server/internal/worker/reconciler_test.go

package worker

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

// TestSyncNowSerializesSnapshotCapture guards the SyncNow ordering contract:
// snapshot capture and reconciliation run under one lock, so a request-path
// sync (playback start/stop) can never interleave with the periodic tick and
// commit an older session snapshot after a newer one.
func TestSyncNowSerializesSnapshotCapture(t *testing.T) {
	var inflight atomic.Int32
	var overlapped atomic.Bool
	provider := func() []SessionSync {
		if inflight.Add(1) > 1 {
			overlapped.Store(true)
		}
		time.Sleep(2 * time.Millisecond)
		inflight.Add(-1)
		return nil
	}

	// No pool is needed: an empty snapshot with no node name returns before
	// any database work, keeping the test focused on the locking contract.
	r := NewReconciler(nil, "", provider)

	var wg sync.WaitGroup
	for range 8 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := r.SyncNow(context.Background()); err != nil {
				t.Errorf("SyncNow: %v", err)
			}
		}()
	}
	wg.Wait()

	if overlapped.Load() {
		t.Fatal("concurrent SyncNow calls captured snapshots concurrently; capture and reconcile must be serialized")
	}
}

// TestSyncNowCoalescesPendingPass guards the follow-up contract: a SyncNow
// call that arrives while a sync is in flight returns immediately, and the
// running owner re-captures a fresh snapshot afterwards — so a stop that lands
// mid-sync is still reflected without waiting for the periodic tick.
func TestSyncNowCoalescesPendingPass(t *testing.T) {
	captures := make(chan struct{}, 16)
	release := make(chan struct{})
	first := true
	provider := func() []SessionSync {
		captures <- struct{}{}
		if first {
			first = false
			<-release // hold the first sync mid-flight
		}
		return nil
	}
	r := NewReconciler(nil, "", provider)

	ownerDone := make(chan error, 1)
	go func() { ownerDone <- r.SyncNow(context.Background()) }()
	<-captures // owner is now blocked inside its snapshot capture

	// A second sync while the first is in flight must not block.
	done := make(chan struct{})
	go func() {
		_ = r.SyncNow(context.Background())
		close(done)
	}()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("SyncNow blocked behind an in-flight sync; it must coalesce and return")
	}

	close(release)
	if err := <-ownerDone; err != nil {
		t.Fatalf("owner SyncNow: %v", err)
	}
	select {
	case <-captures: // the owner's follow-up pass with a fresh snapshot
	default:
		t.Fatal("no follow-up snapshot capture ran; the coalesced sync was lost")
	}
}