Stream oversized ETag responses

Avoid fully buffering responses that exceed the ETag cache entry budget when Content-Length is unknown. Preserve the consumed prefix with the remaining response stream so callers still receive the complete body.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: c4eabbcd-abb9-4c5d-a1f5-c98618623c5b

Author: joshfree

pkg/http/transport/etag.go

@@ -166,27 +166,27 @@ func (t *ETagTransport) RoundTrip(req *http.Request) (*http.Response, error) {
 			return resp, nil
 		}
 
-		// Skip caching bodies that exceed the per-entry byte budget. When the
-		// length is known up front, avoid buffering the body at all.
 		maxEntry := t.maxEntryBytes()
-		if resp.ContentLength > int64(maxEntry) {
-			t.remove(key)
-			return resp, nil
-		}
-
-		body, readErr := io.ReadAll(resp.Body)
-		resp.Body.Close()
+		limited := io.LimitReader(resp.Body, int64(maxEntry)+1)
+		body, readErr := io.ReadAll(limited)
 		if readErr != nil {
+			resp.Body.Close()
 			return nil, readErr
 		}
-		resp.Body = io.NopCloser(bytes.NewReader(body))
-		resp.ContentLength = int64(len(body))
 
 		if len(body) > maxEntry {
 			t.remove(key)
+			resp.Body = struct {
+				io.Reader
+				io.Closer
+			}{io.MultiReader(bytes.NewReader(body), resp.Body), resp.Body}
 			return resp, nil
 		}
 
+		resp.Body.Close()
+		resp.Body = io.NopCloser(bytes.NewReader(body))
+		resp.ContentLength = int64(len(body))
+
 		t.add(key, etagEntry{
 			etag:   etag,
 			status: resp.StatusCode,

pkg/http/transport/etag_test.go

@@ -1,6 +1,7 @@
 package transport
 
 import (
+	"bytes"
 	"context"
 	"io"
 	"net/http"
@@ -14,6 +15,27 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
+type countingReadCloser struct {
+	reader    io.Reader
+	bytesRead int
+}
+
+func (c *countingReadCloser) Read(p []byte) (int, error) {
+	n, err := c.reader.Read(p)
+	c.bytesRead += n
+	return n, err
+}
+
+func (c *countingReadCloser) Close() error {
+	return nil
+}
+
 // TestETagTransport_ServesCachedBodyOn304 verifies the core conditional-request
 // flow: the first GET carries no If-None-Match and is cached with its ETag; the
 // second GET sends the cached ETag and, on a 304 Not Modified, is served the
@@ -236,6 +258,43 @@ func TestETagTransport_SkipsBodiesOverEntryByteBudget(t *testing.T) {
 	assert.Empty(t, lastIfNoneMatch, "an oversized response must not be cached or revalidated")
 }
 
+func TestETagTransport_StreamsOversizedUnknownLengthBody(t *testing.T) {
+	t.Parallel()
+
+	const maxEntry = 16
+	body := []byte("0123456789abcdef-streamed-remainder")
+	stream := &countingReadCloser{reader: bytes.NewReader(body)}
+	transport := roundTripFunc(func(req *http.Request) (*http.Response, error) {
+		assert.Empty(t, req.Header.Get(headers.IfNoneMatchHeader), "an oversized response must not be cached or revalidated")
+		return &http.Response{
+			StatusCode:    http.StatusOK,
+			Status:        "200 OK",
+			Header:        http.Header{headers.ETagHeader: []string{`"streamed"`}},
+			Body:          stream,
+			ContentLength: -1,
+			Request:       req,
+		}, nil
+	})
+	rt := &ETagTransport{Transport: transport, MaxEntryBytes: maxEntry}
+
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, "https://example.test/stream", nil)
+	require.NoError(t, err)
+	resp, err := rt.RoundTrip(req)
+	require.NoError(t, err)
+	require.LessOrEqual(t, stream.bytesRead, maxEntry+1, "RoundTrip must not read the entire oversized body before returning")
+
+	data, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.NoError(t, resp.Body.Close())
+	assert.Equal(t, body, data, "caller receives the buffered prefix plus the streamed remainder")
+
+	stream = &countingReadCloser{reader: bytes.NewReader(body)}
+	resp, err = rt.RoundTrip(req)
+	require.NoError(t, err)
+	_, _ = io.Copy(io.Discard, resp.Body)
+	require.NoError(t, resp.Body.Close())
+}
+
 // TestETagTransport_EvictsByTotalByteBudget verifies that inserting a second
 // entry that pushes the cache over its total-byte budget evicts the
 // least-recently-used entry, which is then re-fetched in full.