diff --git a/internal/httpx/stream_timing_test.go b/internal/httpx/stream_timing_test.go new file mode 100644 index 0000000..6aebeb1 --- /dev/null +++ b/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)) + } + } +} diff --git a/internal/upstream/client.go b/internal/upstream/client.go index d068935..04abcfd 100644 --- a/internal/upstream/client.go +++ b/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,