chore: add env-gated SSE read timing diagnostics

internal/httpx/stream_timing_test.go

@@ -0,0 +1,62 @@
+package httpx
+
+import (
+	"context"
+	"io"
+	"os"
+	"testing"
+	"time"
+)
+
+// TestStreamTiming measures whether the httpx stack (tls-client + proxy)
+// delivers a slow-drip HTTP response progressively or in one burst.
+//
+//	httpbin drip: delay=2s, then numbytes=8 over duration=4s.
+//
+// Progressive: TTFB ~2s, done ~6s. Buffered: TTFB ~6s, all at once.
+func TestStreamTiming(t *testing.T) {
+	if os.Getenv("STREAM_TIMING") == "" {
+		t.Skip("set STREAM_TIMING=1 to run")
+	}
+	proxy := os.Getenv("STREAM_PROXY")
+	s, err := NewSession(Options{ProxyURL: proxy, Timeout: 30 * time.Second})
+	if err != nil {
+		t.Fatalf("session: %v", err)
+	}
+	defer s.Close()
+
+	url := "https://httpbin.org/drip?duration=4&numbytes=8&delay=2"
+	start := time.Now()
+	rc, status, err := s.Stream(context.Background(), Request{Method: "GET", URL: url})
+	if err != nil {
+		t.Fatalf("stream: %v", err)
+	}
+	defer rc.Close()
+	t.Logf("proxy=%q status=%d headers in %v", proxy, status, time.Since(start).Round(time.Millisecond))
+
+	buf := make([]byte, 256)
+	var readAt []time.Duration
+	for {
+		n, rerr := rc.Read(buf)
+		if n > 0 {
+			readAt = append(readAt, time.Since(start))
+		}
+		if rerr == io.EOF {
+			break
+		}
+		if rerr != nil {
+			t.Logf("read err after %d reads: %v", len(readAt), rerr)
+			break
+		}
+	}
+	total := time.Since(start)
+	if len(readAt) == 0 {
+		t.Fatalf("no bytes read")
+	}
+	t.Logf("first byte=%v last byte=%v reads=%d", readAt[0].Round(time.Millisecond), total.Round(time.Millisecond), len(readAt))
+	for i, ts := range readAt {
+		if i < 6 || i >= len(readAt)-3 {
+			t.Logf("  read[%d] @ %v", i, ts.Round(time.Millisecond))
+		}
+	}
+}

internal/upstream/client.go

@@ -9,7 +9,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"html"
+	"io"
 	"log/slog"
+	"os"
 	"regexp"
 	"strconv"
 	"strings"
@@ -673,6 +675,13 @@ func (c *Client) chatOnce(
 	}
 	defer rc.Close()
 
+	if os.Getenv("DSFREE_SSE_TIMING") != "" {
+		start := time.Now()
+		rc = &sseTimingReader{rc: rc, start: start, logf: func(n int) {
+			c.log.Info("sse read", "at_ms", time.Since(start).Milliseconds(), "n", n)
+		}}
+	}
+
 	return readSSE(rc, func(event string, lines []string) error {
 		events, err := translateEvent(event, lines)
 		if err != nil {
@@ -687,6 +696,24 @@ func (c *Client) chatOnce(
 	})
 }
 
+// sseTimingReader (diagnosis only, gated by DSFREE_SSE_TIMING) logs when bytes
+// arrive from the upstream stream.
+type sseTimingReader struct {
+	rc    io.ReadCloser
+	start time.Time
+	logf  func(n int)
+}
+
+func (r *sseTimingReader) Read(p []byte) (int, error) {
+	n, err := r.rc.Read(p)
+	if n > 0 {
+		r.logf(n)
+	}
+	return n, err
+}
+
+func (r *sseTimingReader) Close() error { return r.rc.Close() }
+
 // chatConfig fetches (and caches) the page-embedded nonce for a model.
 func (c *Client) chatConfig(
 	ctx context.Context,