feat: phase 8 serviceadapter crd rollout (32/33 tasks)

internal/proxy/streaming_test.go

@@ -6,7 +6,9 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"runtime"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -473,3 +475,129 @@ func TestNoFullBuffering(t *testing.T) {
 		t.Errorf("expected to read %d bytes, got %d", len(largeData), totalRead)
 	}
 }
+
+// TestClientDisconnectCancelsUpstream verifies that when a client closes mid-stream,
+// the upstream request context is cancelled immediately and no goroutines are leaked.
+func TestClientDisconnectCancelsUpstream(t *testing.T) {
+	contextCancelledAt := time.Time{}
+	contextCancelledMu := sync.Mutex{}
+	upstreamRequestedAt := time.Time{}
+
+	upstreamServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		upstreamRequestedAt = time.Now()
+
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+
+		rc := http.NewResponseController(w)
+		// Send events until context is cancelled
+		for i := 0; i < 100; i++ {
+			select {
+			case <-r.Context().Done():
+				contextCancelledMu.Lock()
+				contextCancelledAt = time.Now()
+				contextCancelledMu.Unlock()
+				return
+			default:
+			}
+
+			fmt.Fprintf(w, "data: event%d\n\n", i)
+			if err := rc.Flush(); err != nil {
+				contextCancelledMu.Lock()
+				contextCancelledAt = time.Now()
+				contextCancelledMu.Unlock()
+				return
+			}
+			time.Sleep(50 * time.Millisecond)
+		}
+	}))
+	defer upstreamServer.Close()
+
+	upstreamAddr := strings.TrimPrefix(upstreamServer.URL, "http://")
+
+	cfg := &config.Config{
+		Routes: map[string]*config.Route{
+			"disconnect-route": {
+				Name: "disconnect-route",
+				Upstream: config.Upstream{
+					Address:        upstreamAddr,
+					ConnectTimeout: 5 * time.Second,
+					ReadTimeout:    30 * time.Second,
+					WriteTimeout:   30 * time.Second,
+					MaxBodySize:    1024 * 1024,
+					AuthRequired:   false,
+				},
+			},
+		},
+	}
+
+	handler := New(cfg)
+	defer handler.Close()
+
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	// Baseline goroutine count
+	baselineGoroutines := runtime.NumGoroutine()
+
+	// Make a request with a custom HTTP client that allows us to close the connection
+	client := &http.Client{
+		Timeout: 30 * time.Second,
+	}
+
+	req, err := http.NewRequest("GET", server.URL+"/disconnect", nil)
+	if err != nil {
+		t.Fatalf("request creation failed: %v", err)
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatalf("request failed: %v", err)
+	}
+	defer resp.Body.Close()
+
+	// Read a few events
+	reader := bufio.NewReader(resp.Body)
+	for i := 0; i < 2; i++ {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			t.Fatalf("read failed: %v", err)
+		}
+		if !strings.Contains(line, "data:") {
+			i-- // skip non-data lines
+		}
+	}
+
+	// Close the response body (simulating client disconnect)
+	resp.Body.Close()
+
+	// Wait a bit for cancellation to propagate
+	time.Sleep(200 * time.Millisecond)
+
+	// Verify context was cancelled
+	contextCancelledMu.Lock()
+	cancelled := !contextCancelledAt.IsZero()
+	cancelDelay := time.Duration(0)
+	if cancelled {
+		cancelDelay = contextCancelledAt.Sub(upstreamRequestedAt)
+	}
+	contextCancelledMu.Unlock()
+
+	if !cancelled {
+		t.Errorf("expected upstream context to be cancelled, but it was not")
+	}
+
+	// Verify cancellation happened quickly (within 1s)
+	if cancelDelay > 1*time.Second {
+		t.Errorf("context cancellation took %.2fs (expected < 1s)", cancelDelay.Seconds())
+	}
+
+	// Wait a bit for goroutines to clean up
+	time.Sleep(100 * time.Millisecond)
+
+	// Check for goroutine leaks
+	finalGoroutines := runtime.NumGoroutine()
+	if finalGoroutines > baselineGoroutines+5 {
+		t.Errorf("possible goroutine leak: baseline=%d, final=%d", baselineGoroutines, finalGoroutines)
+	}
+}