pansy/internal/api/sse_deadline_test.go

package api

import (
	"bufio"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
)

// bareEngine is a gin engine with NO middleware: the narrowest possible host for
// openEventStream, and what the #78/#87 tests were originally written against.
// It is not what production runs — the middleware stack in New wraps the
// ResponseWriter, and that difference is the whole subject of the third test.
func bareEngine() *gin.Engine {
	gin.SetMode(gin.TestMode)
	return gin.New()
}

// streamFrames spins up a real http.Server around r with the given WriteTimeout
// and an SSE route that emits `frames` data frames, one every `tick`, then
// returns. It reports how many frames the client actually received and any read
// error — the only vantage point from which the deadline failures in #78/#87 are
// visible, since the writes themselves return nil when the bytes are dropped.
func streamFrames(t *testing.T, r *gin.Engine, serverWriteTimeout, tick time.Duration, frames int) (int, error) {
	t.Helper()
	r.GET("/stream", func(c *gin.Context) {
		s := openEventStream(c)
		for i := 0; i < frames; i++ {
			time.Sleep(tick)
			s.send(chatEvent{Error: "frame"})
		}
	})

	srv := httptest.NewUnstartedServer(r)
	srv.Config.WriteTimeout = serverWriteTimeout
	srv.Start()
	defer srv.Close()

	resp, err := srv.Client().Get(srv.URL + "/stream")
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	defer resp.Body.Close()

	got := 0
	sc := bufio.NewScanner(resp.Body)
	for sc.Scan() {
		if strings.HasPrefix(sc.Text(), "data: ") {
			got++
		}
	}
	return got, sc.Err()
}

// TestEventStreamOutlivesServerWriteTimeout is the regression test for #78.
//
// http.Server.WriteTimeout is an ABSOLUTE deadline measured from when the
// request header was read — not an idle timeout — so a streaming response is cut
// once it passes, however recently the handler wrote. pansy sets it to 30s while
// an agent turn may run for minutes. openEventStream must override it.
//
// This has to be asserted from the CLIENT side because the failure cannot be
// observed from the handler: writes made after the deadline return err == nil
// and their bytes are silently discarded. A test that checked the return of
// io.WriteString would pass against the bug.
func TestEventStreamOutlivesServerWriteTimeout(t *testing.T) {
	// sseWriteTimeout stays at its 30s default here, so the per-frame refresh
	// keeps the stream alive with a huge margin — CI slowness only ever makes
	// this pass more surely. The server's 300ms WriteTimeout is the thing being
	// overridden; frames straddle it (300ms/600ms/900ms).
	got, err := streamFrames(t, bareEngine(), 300*time.Millisecond, 300*time.Millisecond, 3)
	if err != nil {
		t.Errorf("client read error after %d/3 frames: %v", got, err)
	}
	if got != 3 {
		t.Errorf("client received %d frames, want 3 — the stream was cut at the server WriteTimeout", got)
	}
}

// TestEventStreamRefreshesDeadlinePerFrame guards the #87 fix specifically: the
// write deadline is refreshed on EVERY frame, not set once.
//
// A set-once deadline is a plausible "simplification" and it reintroduces the
// unbounded-block risk the per-frame refresh exists to prevent — yet it would
// sail through the test above, whose whole run is far under sseWriteTimeout. So
// shrink sseWriteTimeout below the stream's total duration and send frames whose
// gap stays comfortably under it: per-frame refresh delivers them all, while a
// deadline set once at open would expire mid-stream and cut it short.
func TestEventStreamRefreshesDeadlinePerFrame(t *testing.T) {
	orig := sseWriteTimeout
	sseWriteTimeout = 400 * time.Millisecond
	t.Cleanup(func() { sseWriteTimeout = orig })

	// The server WriteTimeout is generous (5s), so it isn't the limiter — the
	// per-frame sseWriteTimeout is. 8 frames at a 100ms tick span 800ms, well past
	// the 400ms deadline, but each 100ms gap is a 4× margin under it.
	got, err := streamFrames(t, bareEngine(), 5*time.Second, 100*time.Millisecond, 8)
	if err != nil {
		t.Errorf("client read error after %d/8 frames: %v", got, err)
	}
	if got != 8 {
		t.Errorf("client received %d frames, want 8 — a set-once deadline would cut the stream at ~%v; the refresh must be per-frame",
			got, sseWriteTimeout)
	}
}

// TestEventStreamOutlivesWriteTimeoutBehindMiddleware is #78 again, through the
// production middleware stack — which is where it was still broken.
//
// The two tests above passed while the deployed instance cut every agent turn
// at exactly 30s with "The connection dropped partway through." The difference
// is the middleware: the logging middleware in New replaces c.Writer with a
// wrapper that embeds the gin.ResponseWriter INTERFACE, which has no Unwrap. An
// http.ResponseController built from the handler's c.Writer unwraps layer by
// layer looking for SetWriteDeadline, can't see past that wrapper, and returns
// ErrNotSupported — putting the stream back on the server's absolute
// WriteTimeout. The first write past it fails, which cancels the request
// context and closes the socket under the client mid-frame.
//
// So: the same scenario as the first test, hosted on the engine New builds, in
// the order cmd/pansy runs it. Any future middleware that wraps the writer, or a
// reorder that puts one ahead of the controller capture, fails here.
func TestEventStreamOutlivesWriteTimeoutBehindMiddleware(t *testing.T) {
	got, err := streamFrames(t, authEngine(t, localCfg()), 300*time.Millisecond, 300*time.Millisecond, 3)
	if err != nil {
		t.Errorf("client read error after %d/3 frames: %v", got, err)
	}
	if got != 3 {
		t.Errorf("client received %d frames, want 3 — the stream was cut at the server WriteTimeout; the deadline override is not reaching the socket through the middleware stack", got)
	}
}

// TestResponseControllerReachesTheSocketBehindMiddleware pins the mechanism the
// test above depends on, for every handler that extends a deadline — the scan
// upload extends both (seed_packet.go), and its calls were failing just as
// silently, with the errors discarded.
func TestResponseControllerReachesTheSocketBehindMiddleware(t *testing.T) {
	r := authEngine(t, localCfg())
	var readErr, writeErr error
	r.GET("/deadlines", func(c *gin.Context) {
		rc := responseController(c)
		readErr = rc.SetReadDeadline(time.Now().Add(time.Minute))
		writeErr = rc.SetWriteDeadline(time.Now().Add(time.Minute))
		c.Status(http.StatusNoContent)
	})
	srv := httptest.NewServer(r)
	defer srv.Close()

	resp, err := srv.Client().Get(srv.URL + "/deadlines")
	if err != nil {
		t.Fatalf("get: %v", err)
	}
	resp.Body.Close()
	if readErr != nil {
		t.Errorf("SetReadDeadline through the production middleware: %v", readErr)
	}
	if writeErr != nil {
		t.Errorf("SetWriteDeadline through the production middleware: %v", writeErr)
	}
}