felhom-controller/controller/internal/report/waiter_test.go

package report

import (
	"context"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/http/httptest"
	"sync/atomic"
	"testing"
	"time"
)

func testWaiter(t *testing.T, handler http.HandlerFunc) (*Waiter, *int32, *httptest.Server) {
	t.Helper()
	srv := httptest.NewServer(handler)
	t.Cleanup(srv.Close)
	var fires int32
	w := NewWaiter(srv.URL, "KEY", func() { atomic.AddInt32(&fires, 1) }, log.New(io.Discard, "", 0))
	// Shrink tunables so the loop spins in milliseconds.
	w.holdCeiling = 2 * time.Second
	w.backoffMin = 10 * time.Millisecond
	w.backoffMax = 40 * time.Millisecond
	w.jitterMin = 3 * time.Millisecond
	w.jitterMax = 6 * time.Millisecond
	return w, &fires, srv
}

// runFor runs the waiter for d, then cancels and waits for Run to return.
func runFor(w *Waiter, d time.Duration) {
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { w.Run(ctx); close(done) }()
	time.Sleep(d)
	cancel()
	<-done
}

// genSequence returns a handler that answers the Nth request with seq[min(N-1,len-1)] (clamped),
// optionally prefixed with heartbeat newlines.
func genSequence(heartbeats int, seq ...uint64) http.HandlerFunc {
	var n int32
	return func(w http.ResponseWriter, r *http.Request) {
		i := int(atomic.AddInt32(&n, 1)) - 1
		if i >= len(seq) {
			i = len(seq) - 1
		}
		fl, _ := w.(http.Flusher)
		for h := 0; h < heartbeats; h++ {
			io.WriteString(w, "\n")
			if fl != nil {
				fl.Flush()
			}
		}
		fmt.Fprintf(w, "{\"gen\":%d}\n", seq[i])
	}
}

func TestWaiter_FiresOnceOnGenChange(t *testing.T) {
	// baseline gen 0, then 1, then steady 1 → exactly one fire.
	w, fires, _ := testWaiter(t, genSequence(0, 0, 1, 1, 1, 1))
	runFor(w, 200*time.Millisecond)
	if got := atomic.LoadInt32(fires); got != 1 {
		t.Fatalf("expected exactly 1 fire on a single gen change, got %d", got)
	}
}

func TestWaiter_TimeoutSameGenNoFire(t *testing.T) {
	// The hub keeps returning the same generation (its hold timed out) → never fires.
	w, fires, _ := testWaiter(t, genSequence(0, 7, 7, 7, 7, 7))
	runFor(w, 150*time.Millisecond)
	if got := atomic.LoadInt32(fires); got != 0 {
		t.Fatalf("same-gen timeouts must not fire, got %d", got)
	}
}

// TestWaiter_FirstObservationRecordsNoFire is the load-bearing property with its RED-PROOF anchor:
// a fresh process whose FIRST wait returns an already-advanced generation records it WITHOUT firing
// (the startup report already covered current state; firing would echo one extra report on every
// restart / config-refresh).
//
// RED-PROOF: drop the `!w.seen` guard in Run (fire whenever gen != lastSeen, with lastSeen starting
// at 0) — the first poll here returns gen 5 != 0 and fires once; this assertion (0 fires) then fails.
// Revert to restore green.
func TestWaiter_FirstObservationRecordsNoFire(t *testing.T) {
	w, fires, _ := testWaiter(t, genSequence(0, 5, 5, 5, 5)) // existing non-zero gen, never changes
	runFor(w, 150*time.Millisecond)
	if got := atomic.LoadInt32(fires); got != 0 {
		t.Fatalf("first observation must record-not-fire; got %d fires", got)
	}
}

func TestWaiter_HeartbeatsToleratedThenFire(t *testing.T) {
	// 3 heartbeat newlines precede each completion; baseline 0 then 1 → one fire, no panic.
	w, fires, _ := testWaiter(t, genSequence(3, 0, 1, 1, 1))
	runFor(w, 200*time.Millisecond)
	if got := atomic.LoadInt32(fires); got != 1 {
		t.Fatalf("heartbeats must be tolerated and the change fire once; got %d", got)
	}
}

func TestWaiter_MalformedCompletionNoFireNoPanic(t *testing.T) {
	w, fires, _ := testWaiter(t, func(w http.ResponseWriter, r *http.Request) {
		io.WriteString(w, "\n\nnot-json\n")
	})
	runFor(w, 120*time.Millisecond)
	if got := atomic.LoadInt32(fires); got != 0 {
		t.Fatalf("a malformed completion must not fire; got %d", got)
	}
}

func TestWaiter_404TreatedAsErrorNoFire(t *testing.T) {
	// A hub that predates the endpoint returns 404 — the Waiter backs off, never fires, keeps looping.
	w, fires, _ := testWaiter(t, func(w http.ResponseWriter, r *http.Request) {
		http.Error(w, "not found", http.StatusNotFound)
	})
	runFor(w, 120*time.Millisecond)
	if got := atomic.LoadInt32(fires); got != 0 {
		t.Fatalf("404 must be an ordinary (no-fire) error; got %d", got)
	}
}

func TestWaiter_SendsBearerAndGenParam(t *testing.T) {
	var sawAuth, sawGen atomic.Value
	sawAuth.Store("")
	sawGen.Store("")
	w, _, _ := testWaiter(t, func(w http.ResponseWriter, r *http.Request) {
		sawAuth.Store(r.Header.Get("Authorization"))
		sawGen.Store(r.URL.Query().Get("gen"))
		fmt.Fprint(w, "{\"gen\":0}\n")
	})
	runFor(w, 80*time.Millisecond)
	if sawAuth.Load().(string) != "Bearer KEY" {
		t.Fatalf("expected Bearer KEY, got %q", sawAuth.Load())
	}
	if sawGen.Load().(string) != "0" {
		t.Fatalf("expected gen=0 on the first poll, got %q", sawGen.Load())
	}
}

func TestWaiter_CtxCancelMidHoldReturnsPromptly(t *testing.T) {
	// The server holds until the client disconnects; cancelling the ctx must return Run promptly.
	w, _, _ := testWaiter(t, func(w http.ResponseWriter, r *http.Request) {
		<-r.Context().Done() // hold until the client goes away
	})
	w.holdCeiling = 10 * time.Second // must NOT be what unblocks us
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { w.Run(ctx); close(done) }()
	time.Sleep(40 * time.Millisecond)
	cancel()
	select {
	case <-done:
	case <-time.After(2 * time.Second):
		t.Fatal("Run did not return promptly after ctx cancel mid-hold")
	}
}