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") } }