felhom.eu/hub/internal/intent/hub_test.go

package intent

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

// --- white-box helpers (same package) ---

func (h *Hub) waiterCount(customerID string) int {
	h.mu.Lock()
	defer h.mu.Unlock()
	return len(h.waiters[customerID])
}

func (h *Hub) totalWaiters() int {
	h.mu.Lock()
	defer h.mu.Unlock()
	n := 0
	for _, ws := range h.waiters {
		n += len(ws)
	}
	return n
}

// waitUntilRegistered blocks (briefly) until n waiters are registered for customerID, so tests
// don't race the goroutine that calls Wait. Fails the test on timeout.
func waitUntilRegistered(t *testing.T, h *Hub, customerID string, n int) {
	t.Helper()
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		if h.waiterCount(customerID) >= n {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("timed out waiting for %d waiter(s) on %q (have %d)", n, customerID, h.waiterCount(customerID))
}

// asyncWait runs Wait in a goroutine and returns a channel that yields its result.
func asyncWait(h *Hub, ctx context.Context, customerID string, lastSeen uint64, maxHold time.Duration) <-chan uint64 {
	out := make(chan uint64, 1)
	go func() { out <- h.Wait(ctx, customerID, lastSeen, maxHold) }()
	return out
}

func recvWithin(t *testing.T, ch <-chan uint64, d time.Duration) uint64 {
	t.Helper()
	select {
	case g := <-ch:
		return g
	case <-time.After(d):
		t.Fatalf("Wait did not return within %s", d)
		return 0
	}
}

// --- tests ---

func TestBump_WakesRegisteredWaiter(t *testing.T) {
	h := New()
	ch := asyncWait(h, context.Background(), "c1", 0, time.Second)
	waitUntilRegistered(t, h, "c1", 1)
	h.Bump("c1")
	if g := recvWithin(t, ch, time.Second); g != 1 {
		t.Fatalf("expected gen 1 after one bump, got %d", g)
	}
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked after bump: %d", h.totalWaiters())
	}
}

// TestWait_CoalescesBurstToLatestGen is the load-bearing property (§11 Group A) with its
// RED-PROOF anchor: N rapid bumps during one hold produce ONE completion carrying the LATEST
// generation (a counter, not a per-bump queue).
//
// RED-PROOF: change Bump to complete one waiter per bump / re-register (a queue), or make the
// completion carry the FIRST bump's value instead of finish() re-reading the latest — this
// assertion (gen == N) then fails (it sees 1). Revert to restore green.
func TestWait_CoalescesBurstToLatestGen(t *testing.T) {
	h := New()
	ch := asyncWait(h, context.Background(), "c1", 0, 2*time.Second)
	waitUntilRegistered(t, h, "c1", 1)

	const n = 5
	for i := 0; i < n; i++ {
		h.Bump("c1")
	}

	if g := recvWithin(t, ch, time.Second); g != n {
		t.Fatalf("expected coalesced completion with latest gen %d, got %d", n, g)
	}
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked after burst: %d", h.totalWaiters())
	}
	if got := h.Generation("c1"); got != n {
		t.Fatalf("generation counter should be %d, got %d", n, got)
	}
}

// TestWait_RaceCloser_BumpBeforeWaitNotLost is the second load-bearing property with its
// RED-PROOF anchor: a bump that lands BEFORE the wait registers must not be lost — the wait
// returns immediately with the advanced generation instead of blocking to the timeout.
//
// RED-PROOF: remove the `cur != lastSeen` pre-register check in Wait — the wait then blocks the
// full maxHold and this sub-100ms assertion fails. Revert to restore green.
func TestWait_RaceCloser_BumpBeforeWaitNotLost(t *testing.T) {
	h := New()
	h.Bump("c1") // gen -> 1, no waiter registered yet

	start := time.Now()
	g := h.Wait(context.Background(), "c1", 0, 2*time.Second) // lastSeen 0 != 1 → immediate
	elapsed := time.Since(start)

	if g != 1 {
		t.Fatalf("expected immediate gen 1, got %d", g)
	}
	if elapsed > 100*time.Millisecond {
		t.Fatalf("race-closer should return immediately, took %s", elapsed)
	}
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked on race-closer path: %d", h.totalWaiters())
	}
}

func TestWait_TimeoutReturnsCurrentGenNoLeak(t *testing.T) {
	h := New()
	// Fresh customer, no bumps: times out returning 0.
	start := time.Now()
	if g := h.Wait(context.Background(), "c1", 0, 40*time.Millisecond); g != 0 {
		t.Fatalf("expected 0 on timeout, got %d", g)
	}
	if el := time.Since(start); el < 30*time.Millisecond {
		t.Fatalf("returned before maxHold elapsed (%s)", el)
	}
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked after timeout: %d", h.totalWaiters())
	}

	// Existing generation: a same-gen wait times out returning that (unchanged) gen.
	h.Bump("c2")
	h.Bump("c2") // gen -> 2
	if g := h.Wait(context.Background(), "c2", 2, 40*time.Millisecond); g != 2 {
		t.Fatalf("expected unchanged gen 2 on timeout, got %d", g)
	}
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked after same-gen timeout: %d", h.totalWaiters())
	}
}

func TestWait_CtxCancelReturnsPromptly(t *testing.T) {
	h := New()
	ctx, cancel := context.WithCancel(context.Background())
	ch := asyncWait(h, ctx, "c1", 0, 10*time.Second)
	waitUntilRegistered(t, h, "c1", 1)

	cancel()
	if g := recvWithin(t, ch, time.Second); g != 0 {
		t.Fatalf("expected 0 on ctx cancel, got %d", g)
	}
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked after ctx cancel: %d", h.totalWaiters())
	}
}

func TestClose_WakesAllWaitersWithCurrentGen(t *testing.T) {
	h := New()
	h.Bump("c1") // gen c1 -> 1
	ch1 := asyncWait(h, context.Background(), "c1", 1, 10*time.Second)
	ch2 := asyncWait(h, context.Background(), "c2", 0, 10*time.Second)
	waitUntilRegistered(t, h, "c1", 1)
	waitUntilRegistered(t, h, "c2", 1)

	h.Close()

	if g := recvWithin(t, ch1, time.Second); g != 1 {
		t.Fatalf("c1 should return current gen 1 on close, got %d", g)
	}
	if g := recvWithin(t, ch2, time.Second); g != 0 {
		t.Fatalf("c2 should return current gen 0 on close, got %d", g)
	}
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked after close: %d", h.totalWaiters())
	}
	// Post-close waits return immediately.
	start := time.Now()
	_ = h.Wait(context.Background(), "c3", 0, 10*time.Second)
	if el := time.Since(start); el > 100*time.Millisecond {
		t.Fatalf("post-close Wait should return immediately, took %s", el)
	}
	// Post-close bumps are no-ops (must not panic on the closed channel).
	h.Bump("c1")
}

func TestConcurrentWaitsSameCustomerBothComplete(t *testing.T) {
	h := New()
	ch1 := asyncWait(h, context.Background(), "c1", 0, 2*time.Second)
	ch2 := asyncWait(h, context.Background(), "c1", 0, 2*time.Second)
	waitUntilRegistered(t, h, "c1", 2)

	h.Bump("c1")

	if g := recvWithin(t, ch1, time.Second); g != 1 {
		t.Fatalf("waiter 1 expected gen 1, got %d", g)
	}
	if g := recvWithin(t, ch2, time.Second); g != 1 {
		t.Fatalf("waiter 2 expected gen 1, got %d", g)
	}
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked: %d", h.totalWaiters())
	}
}

// TestBump_EmptyCustomerIsNoop guards the global-floor path (which iterates customers) and any
// stray empty id.
func TestBump_EmptyCustomerIsNoop(t *testing.T) {
	h := New()
	h.Bump("")
	if h.Generation("") != 0 {
		t.Fatalf("empty-customer bump must be a no-op")
	}
}

// TestRaceUnderStress runs Wait/Bump concurrently to shake out data races (run with -race).
func TestRaceUnderStress(t *testing.T) {
	h := New()
	var wg sync.WaitGroup
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
			defer cancel()
			_ = h.Wait(ctx, "c1", 0, 80*time.Millisecond)
		}()
	}
	for i := 0; i < 50; i++ {
		h.Bump("c1")
	}
	wg.Wait()
	if h.totalWaiters() != 0 {
		t.Fatalf("waiter leaked under stress: %d", h.totalWaiters())
	}
}