fix: preserve HTTP response integrity and streaming delivery

docs/DEPLOYMENT.md

@@ -253,6 +253,15 @@ progress as activity, including small buffered responses. Stalled request or
 response bodies and blocked writes remain bounded; header and keepalive
 timeouts are unchanged.
 
+Source builds also forward unknown-length HTTP response bodies and
+`text/event-stream` events without waiting for the origin to finish. Fixed-size
+non-streaming responses retain normal buffering. If copying an origin response
+fails, the proxy aborts that response instead of emitting a successful final
+chunk: HTTP/1.1 clients can detect truncation. Failure before buffered headers
+are sent can instead appear as a connection error. Declared trailers are
+forwarded only after the body completes successfully. This does not add HTTP
+Upgrade support or change CONNECT tunnels, and is not included in v1.0.1.
+
 The relay's separate `--destination-write-timeout` defaults to `5m`; `0`
 also selects `5m`, rather than disabling the limit. It bounds the completion
 of each TCP destination write, split into chunks of at most 32 KiB. Each

internal/proxy/http.go

@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"errors"
 	"io"
+	"mime"
 	"net"
 	"net/http"
 	"net/textproto"
@@ -290,7 +291,13 @@ func (s *HTTPServer) serveForward(w http.ResponseWriter, r *http.Request) {
 		// writer buffers small chunks and the client sends nothing more.
 		body = &httpResponseActivityReader{Reader: body, conn: conn}
 	}
-	_, _ = io.Copy(w, body)
+	if err := copyHTTPResponse(w, body, response); err != nil {
+		// Headers may already be on the wire. Returning normally would let
+		// net/http finalize a chunked response (or infer a short Content-Length),
+		// hiding a truncated origin body from the client. Abort only this
+		// response; net/http handles the sentinel without logging a stack trace.
+		panic(http.ErrAbortHandler)
+	}
 	for key, values := range response.Trailer {
 		if isHopByHopHeader(key) {
 			continue
@@ -299,6 +306,42 @@ func (s *HTTPServer) serveForward(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+// Unknown-length responses and server-sent events can remain open indefinitely.
+// Flush their headers and each body write so small events don't wait in the
+// HTTP server's response buffer until the origin finishes. Ordinary fixed-size
+// responses retain net/http's buffering and copy optimizations.
+func copyHTTPResponse(w http.ResponseWriter, body io.Reader, response *http.Response) error {
+	contentType, _, _ := mime.ParseMediaType(response.Header.Get("Content-Type"))
+	var destination io.Writer = w
+	if response.ContentLength == -1 || contentType == "text/event-stream" {
+		flush := http.NewResponseController(w).Flush
+		if err := flush(); err != nil {
+			// Serve's native ResponseWriter supports flushing. Preserve Handler
+			// compatibility for embedders whose wrappers don't expose it.
+			if !errors.Is(err, http.ErrNotSupported) {
+				return err
+			}
+		} else {
+			destination = httpResponseFlushWriter{writer: w, flush: flush}
+		}
+	}
+	_, err := io.Copy(destination, body)
+	return err
+}
+
+type httpResponseFlushWriter struct {
+	writer io.Writer
+	flush  func() error
+}
+
+func (w httpResponseFlushWriter) Write(p []byte) (int, error) {
+	n, err := w.writer.Write(p)
+	if n > 0 && err == nil {
+		err = w.flush()
+	}
+	return n, err
+}
+
 type httpResponseActivityReader struct {
 	io.Reader
 	conn *trackedConn

internal/proxy/http_response_copy_test.go

@@ -0,0 +1,84 @@
+package proxy
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"testing/iotest"
+)
+
+func TestCopyHTTPResponseFlushAndErrors(t *testing.T) {
+	copyErr := errors.New("response copy fixture")
+	for _, test := range []struct {
+		name         string
+		length       int64
+		contentType  string
+		hideFlusher  bool
+		flushErrorAt int
+		writeError   bool
+		readError    bool
+		wantFlushes  int
+		wantBody     string
+		wantError    bool
+	}{
+		{name: "unknown length", length: -1, wantFlushes: 2, wantBody: "body"},
+		{name: "event stream", length: 4, contentType: "text/event-stream; charset=utf-8", wantFlushes: 2, wantBody: "body"},
+		{name: "ordinary fixed size", length: 4, contentType: "text/plain", wantBody: "body"},
+		{name: "wrapper without flusher", length: -1, hideFlusher: true, wantBody: "body"},
+		{name: "header flush error", length: -1, flushErrorAt: 1, wantFlushes: 1, wantError: true},
+		{name: "body flush error", length: -1, flushErrorAt: 2, wantFlushes: 2, wantBody: "body", wantError: true},
+		{name: "body write error", length: -1, writeError: true, wantFlushes: 1, wantError: true},
+		{name: "body read error", length: -1, readError: true, wantFlushes: 1, wantError: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			writer := &httpCopyTestWriter{header: make(http.Header), err: copyErr, flushErrorAt: test.flushErrorAt, writeError: test.writeError}
+			var destination http.ResponseWriter = writer
+			if test.hideFlusher {
+				destination = struct{ http.ResponseWriter }{writer}
+			}
+			var source io.Reader = strings.NewReader("body")
+			if test.readError {
+				source = iotest.ErrReader(copyErr)
+			}
+			response := &http.Response{ContentLength: test.length, Header: make(http.Header)}
+			response.Header.Set("Content-Type", test.contentType)
+			err := copyHTTPResponse(destination, source, response)
+			if (err != nil) != test.wantError || (test.wantError && !errors.Is(err, copyErr)) {
+				t.Fatalf("copy error = %v, want fixture error = %v", err, test.wantError)
+			}
+			if writer.flushes != test.wantFlushes || writer.body.String() != test.wantBody {
+				t.Fatalf("flushes/body = %d/%q, want %d/%q", writer.flushes, writer.body.String(), test.wantFlushes, test.wantBody)
+			}
+		})
+	}
+}
+
+type httpCopyTestWriter struct {
+	header       http.Header
+	body         bytes.Buffer
+	err          error
+	flushes      int
+	flushErrorAt int
+	writeError   bool
+}
+
+func (w *httpCopyTestWriter) Header() http.Header { return w.header }
+func (*httpCopyTestWriter) WriteHeader(int)       {}
+
+func (w *httpCopyTestWriter) Write(p []byte) (int, error) {
+	if w.writeError {
+		return 0, w.err
+	}
+	return w.body.Write(p)
+}
+
+func (w *httpCopyTestWriter) FlushError() error {
+	w.flushes++
+	if w.flushes == w.flushErrorAt {
+		return w.err
+	}
+	return nil
+}

internal/proxy/http_response_integrity_test.go

@@ -0,0 +1,119 @@
+package proxy
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestHTTPForwardRejectsTruncatedChunkedResponse(t *testing.T) {
+	for _, size := range []int{32, 64 << 10} {
+		t.Run(fmt.Sprint(size), func(t *testing.T) {
+			payload := bytes.Repeat([]byte("x"), size)
+			origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				conn, rw, err := w.(http.Hijacker).Hijack()
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				defer conn.Close()
+				// Send a complete data chunk but omit the terminal zero chunk.
+				// The body length is unknown, so a proxy that returns normally
+				// after a copy error can falsely turn this into a complete reply.
+				_, _ = fmt.Fprintf(rw, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n%x\r\n", len(payload))
+				_, _ = rw.Write(payload)
+				_, _ = rw.WriteString("\r\n")
+				_ = rw.Flush()
+			}))
+			defer origin.Close()
+			server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()})
+			defer stop(server)
+			client := proxyHTTPClient(t, proxyURL, nil)
+			defer client.CloseIdleConnections()
+			response, err := client.Get(origin.URL)
+			if err != nil {
+				t.Fatalf("read streamed response headers: %v", err)
+			}
+			defer response.Body.Close()
+			got, err := io.ReadAll(response.Body)
+			if !errors.Is(err, io.ErrUnexpectedEOF) || response.StatusCode != http.StatusOK || !bytes.Equal(got, payload) {
+				t.Fatalf("truncated chunked response: status=%d bytes=%d error=%v; want 200, %d bytes and unexpected EOF", response.StatusCode, len(got), err, size)
+			}
+		})
+	}
+}
+
+func TestHTTPForwardStreamsBeforeOriginCompletes(t *testing.T) {
+	for _, kind := range []string{"unknown length", "event stream with known length"} {
+		t.Run(kind, func(t *testing.T) {
+			const first = "data: first\n\n"
+			const last = "data: last\n\n"
+			release := make(chan struct{})
+			origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if kind == "event stream with known length" {
+					w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
+					w.Header().Set("Content-Length", fmt.Sprint(len(first)+len(last)))
+				}
+				_, _ = io.WriteString(w, first)
+				w.(http.Flusher).Flush()
+				select {
+				case <-release:
+					_, _ = io.WriteString(w, last)
+				case <-r.Context().Done():
+				}
+			}))
+			defer origin.Close()
+			server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()})
+			defer stop(server)
+			client := proxyHTTPClient(t, proxyURL, nil)
+			client.Timeout = 2 * time.Second
+			defer client.CloseIdleConnections()
+			response, err := client.Get(origin.URL)
+			if err != nil {
+				t.Fatalf("receive response while origin remains open: %v", err)
+			}
+			defer response.Body.Close()
+			prefix := make([]byte, len(first))
+			if _, err := io.ReadFull(response.Body, prefix); err != nil {
+				t.Fatalf("read first event before releasing origin: %v", err)
+			}
+			if string(prefix) != first {
+				t.Fatalf("first event = %q", prefix)
+			}
+			close(release)
+			rest, err := io.ReadAll(response.Body)
+			if err != nil || string(rest) != last {
+				t.Fatalf("remaining stream = %q, %v", rest, err)
+			}
+		})
+	}
+}
+
+func TestHTTPForwardCompleteChunkedResponseWithTrailers(t *testing.T) {
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Trailer", "X-Body-Complete")
+		_, _ = io.WriteString(w, "first")
+		w.(http.Flusher).Flush()
+		_, _ = io.WriteString(w, "last")
+		w.Header().Set("X-Body-Complete", "yes")
+	}))
+	defer origin.Close()
+	server, proxyURL, stop := startHTTPProxy(t, Config{Dialer: directDialer()})
+	defer stop(server)
+	client := proxyHTTPClient(t, proxyURL, nil)
+	defer client.CloseIdleConnections()
+	response, err := client.Get(origin.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	body, err := io.ReadAll(response.Body)
+	if err != nil || string(body) != "firstlast" || response.Trailer.Get("X-Body-Complete") != "yes" {
+		t.Fatalf("complete response body=%q trailer=%v error=%v", body, response.Trailer, err)
+	}
+}

internal/proxy/http_test.go

@@ -375,12 +375,17 @@ func TestHTTPActiveBodyIdleTimeouts(t *testing.T) {
 		server, proxyURL, stopProxy := startHTTPProxy(t, Config{Dialer: directDialer(), IdleTimeout: idle})
 		defer stopProxy(server)
 		client := proxyHTTPClient(t, proxyURL, nil)
+		started := time.Now()
 		response, err := client.Get(origin.URL + "/stall")
 		if err != nil {
-			t.Fatal(err)
+			// A fixed-size response with no body can remain buffered. Aborting
+			// the failed origin copy may therefore close before headers arrive.
+			if !errors.Is(err, io.EOF) || time.Since(started) > time.Second {
+				t.Fatalf("stalled response did not abort promptly: %v", err)
+			}
+			return
 		}
 		defer response.Body.Close()
-		started := time.Now()
 		_, err = io.ReadAll(response.Body)
 		if err == nil {
 			t.Fatal("truncated stalled response unexpectedly completed successfully")