8823050 fix(proxy): handle client-aborted streaming requests · gordon

M internal/adapters/in/http/middleware/logging.go +10 -0


@@ -5,6 +5,7 @@ import (
5	5		"crypto/rand"
6	6		"encoding/hex"
7	7		"encoding/json"
	8	+	"errors"
8	9		"fmt"
9	10		"net"
10	11		"net/http"

@@ -163,6 +164,10 @@ func PanicRecovery(log zerowrap.Logger) func(http.Handler) http.Handler {
163	164		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
164	165		defer func() {
165	166		if err := recover(); err != nil {
	167	+	if isAbortHandlerPanic(err) {
	168	+	panic(http.ErrAbortHandler)
	169	+	}
	170	+
166	171		log.Error().
167	172		Str(zerowrap.FieldLayer, "adapter").
168	173		Str(zerowrap.FieldAdapter, "http").

@@ -183,6 +188,11 @@ func PanicRecovery(log zerowrap.Logger) func(http.Handler) http.Handler {
183	188		}
184	189		}
185	190
	191	+	func isAbortHandlerPanic(v any) bool {
	192	+	err, ok := v.(error)
	193	+	return ok && errors.Is(err, http.ErrAbortHandler)
	194	+	}
	195	+
186	196		// CORS middleware adds permissive CORS headers.
187	197		// NOTE: This is intentionally NOT used on the proxy chain (backends control their own CORS).
188	198		// Available for use on specific endpoints (e.g., registry API) where CORS is needed.


@@ -5,6 +5,7 @@ import (
5	"crypto/rand"	5		"crypto/rand"
6	"encoding/hex"	6		"encoding/hex"
7	"encoding/json"	7		"encoding/json"
		8	+	"errors"
8	"fmt"	9		"fmt"
9	"net"	10		"net"
10	"net/http"	11		"net/http"

@@ -163,6 +164,10 @@ func PanicRecovery(log zerowrap.Logger) func(http.Handler) http.Handler {
163	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {	164		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
164	defer func() {	165		defer func() {
165	if err := recover(); err != nil {	166		if err := recover(); err != nil {
		167	+	if isAbortHandlerPanic(err) {
		168	+	panic(http.ErrAbortHandler)
		169	+	}
		170	+
166	log.Error().	171		log.Error().
167	Str(zerowrap.FieldLayer, "adapter").	172		Str(zerowrap.FieldLayer, "adapter").
168	Str(zerowrap.FieldAdapter, "http").	173		Str(zerowrap.FieldAdapter, "http").

@@ -183,6 +188,11 @@ func PanicRecovery(log zerowrap.Logger) func(http.Handler) http.Handler {
183	}	188		}
184	}	189		}
185		190
		191	+	func isAbortHandlerPanic(v any) bool {
		192	+	err, ok := v.(error)
		193	+	return ok && errors.Is(err, http.ErrAbortHandler)
		194	+	}
		195	+
186	// CORS middleware adds permissive CORS headers.	196		// CORS middleware adds permissive CORS headers.
187	// NOTE: This is intentionally NOT used on the proxy chain (backends control their own CORS).	197		// NOTE: This is intentionally NOT used on the proxy chain (backends control their own CORS).
188	// Available for use on specific endpoints (e.g., registry API) where CORS is needed.	198		// Available for use on specific endpoints (e.g., registry API) where CORS is needed.

A internal/adapters/in/http/middleware/logging_test.go +25 -0

@@ -0,0 +1,25 @@
	1	+	package middleware
	2	+
	3	+	import (
	4	+	"net/http"
	5	+	"net/http/httptest"
	6	+	"testing"
	7	+
	8	+	"github.com/stretchr/testify/assert"
	9	+	)
	10	+
	11	+	func TestPanicRecovery_RethrowsReverseProxyAbort(t *testing.T) {
	12	+	log := testLogger()
	13	+
	14	+	handler := PanicRecovery(log)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	15	+	panic(http.ErrAbortHandler)
	16	+	}))
	17	+
	18	+	req := httptest.NewRequest(http.MethodGet, "/stream", nil)
	19	+	rw := httptest.NewRecorder()
	20	+
	21	+	assert.PanicsWithValue(t, http.ErrAbortHandler, func() {
	22	+	handler.ServeHTTP(rw, req)
	23	+	})
	24	+	assert.NotEqual(t, http.StatusInternalServerError, rw.Code)
	25	+	}

@@ -0,0 +1,25 @@
	1	+	package middleware
	2	+
	3	+	import (
	4	+	"net/http"
	5	+	"net/http/httptest"
	6	+	"testing"
	7	+
	8	+	"github.com/stretchr/testify/assert"
	9	+	)
	10	+
	11	+	func TestPanicRecovery_RethrowsReverseProxyAbort(t *testing.T) {
	12	+	log := testLogger()
	13	+
	14	+	handler := PanicRecovery(log)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	15	+	panic(http.ErrAbortHandler)
	16	+	}))
	17	+
	18	+	req := httptest.NewRequest(http.MethodGet, "/stream", nil)
	19	+	rw := httptest.NewRecorder()
	20	+
	21	+	assert.PanicsWithValue(t, http.ErrAbortHandler, func() {
	22	+	handler.ServeHTTP(rw, req)
	23	+	})
	24	+	assert.NotEqual(t, http.StatusInternalServerError, rw.Code)
	25	+	}

M internal/adapters/in/http/proxy/forwarder.go +63 -21


@@ -2,6 +2,7 @@
2	2		package proxy
3	3
4	4		import (
	5	+	"context"
5	6		"errors"
6	7		"fmt"
7	8		"io"

@@ -102,16 +103,15 @@ func (h Handler) forwardToTarget(w http.ResponseWriter, r http.Request, target
102	103		transport: transport,
103	104		incomingReq: r,
104	105		trustedNets: h.trustedNets,
105		-	errorHandler: func(w http.ResponseWriter, _ *http.Request, proxyErr error) {
106		-	var maxBytesErr *http.MaxBytesError
107		-	if errors.As(proxyErr, &maxBytesErr) {
108		-	log.Warn().Err(proxyErr).Str("target", targetURL.String()).Msg("proxy error: request body too large")
109		-	proxyError(w, "Request Entity Too Large", http.StatusRequestEntityTooLarge)
110		-	return
111		-	}
112		-	log.Error().Err(proxyErr).Str("target", targetURL.String()).Msg("proxy error: connection failed")
113		-	proxyError(w, "Service Unavailable", http.StatusServiceUnavailable)
114		-	},
	106	+	errorHandler: newProxyErrorHandler(
	107	+	log.WithField("target", targetURL.String()),
	108	+	proxyErrorHandlerConfig{
	109	+	requestTooLargeLog: "proxy error: request body too large",
	110	+	clientDisconnectLog: "proxy request canceled by client",
	111	+	upstreamErrorLog: "proxy error: connection failed",
	112	+	upstreamErrorBody: "Service Unavailable",
	113	+	},
	114	+	),
115	115		modifyResponse: modifyResponse(maxResponseSize),
116	116		})
117	117

@@ -152,16 +152,15 @@ func (h Handler) forwardToRegistry(w http.ResponseWriter, r http.Request, regi
152	152		pr.Out.Host = targetURL.Host
153	153		},
154	154		Transport: h.registryTransport,
155		-	ErrorHandler: func(w http.ResponseWriter, _ *http.Request, proxyErr error) {
156		-	var maxBytesErr *http.MaxBytesError
157		-	if errors.As(proxyErr, &maxBytesErr) {
158		-	log.Warn().Err(proxyErr).Int("registry_port", registryPort).Msg("registry proxy error: request body too large")
159		-	proxyError(w, "Request Entity Too Large", http.StatusRequestEntityTooLarge)
160		-	return
161		-	}
162		-	log.Error().Err(proxyErr).Int("registry_port", registryPort).Msg("registry proxy error")
163		-	proxyError(w, "Registry Unavailable", http.StatusServiceUnavailable)
164		-	},
	155	+	ErrorHandler: newProxyErrorHandler(
	156	+	log.WithField("registry_port", registryPort),
	157	+	proxyErrorHandlerConfig{
	158	+	requestTooLargeLog: "registry proxy error: request body too large",
	159	+	clientDisconnectLog: "registry proxy request canceled by client",
	160	+	upstreamErrorLog: "registry proxy error",
	161	+	upstreamErrorBody: "Registry Unavailable",
	162	+	},
	163	+	),
165	164		// Strip browser-oriented security headers from upstream registry responses
166	165		// to avoid conflicts with Gordon's own headers on browser-facing routes.
167	166		ModifyResponse: func(resp *http.Response) error {

@@ -181,6 +180,34 @@ func (h Handler) forwardToRegistry(w http.ResponseWriter, r http.Request, regi
181	180		proxy.ServeHTTP(w, r)
182	181		}
183	182
	183	+	type proxyErrorHandlerConfig struct {
	184	+	requestTooLargeLog string
	185	+	clientDisconnectLog string
	186	+	upstreamErrorLog string
	187	+	upstreamErrorBody string
	188	+	}
	189	+
	190	+	func newProxyErrorHandler(log zerowrap.Logger, cfg proxyErrorHandlerConfig) func(http.ResponseWriter, *http.Request, error) {
	191	+	return func(w http.ResponseWriter, _ *http.Request, proxyErr error) {
	192	+	switch {
	193	+	case isRequestBodyTooLargeError(proxyErr):
	194	+	log.Warn().Err(proxyErr).Msg(cfg.requestTooLargeLog)
	195	+	proxyError(w, "Request Entity Too Large", http.StatusRequestEntityTooLarge)
	196	+	case isClientDisconnectError(proxyErr):
	197	+	log.Debug().Err(proxyErr).Msg(cfg.clientDisconnectLog)
	198	+	clientClosedRequest(w)
	199	+	default:
	200	+	log.Error().Err(proxyErr).Msg(cfg.upstreamErrorLog)
	201	+	proxyError(w, cfg.upstreamErrorBody, http.StatusServiceUnavailable)
	202	+	}
	203	+	}
	204	+	}
	205	+
	206	+	func isRequestBodyTooLargeError(err error) bool {
	207	+	_, ok := errors.AsType[*http.MaxBytesError](err)
	208	+	return ok
	209	+	}
	210	+
184	211		type reverseProxyOptions struct {
185	212		targetURL *url.URL
186	213		hostHeader string

@@ -305,10 +332,25 @@ func (l *limitedReadCloser) Close() error {
305	332		return l.ReadCloser.Close()
306	333		}
307	334
	335	+	const statusClientClosedRequest = 499
	336	+
	337	+	func isClientDisconnectError(err error) bool {
	338	+	return errors.Is(err, context.Canceled) \|\| errors.Is(err, http.ErrAbortHandler)
	339	+	}
	340	+
	341	+	func clientClosedRequest(w http.ResponseWriter) {
	342	+	setProxyGeneratedResponseHeaders(w)
	343	+	w.WriteHeader(statusClientClosedRequest)
	344	+	}
	345	+
308	346		// proxyError writes an error response with security headers appropriate for
309	347		// proxy-generated error pages.
310	348		func proxyError(w http.ResponseWriter, msg string, code int) {
	349	+	setProxyGeneratedResponseHeaders(w)
	350	+	http.Error(w, msg, code)
	351	+	}
	352	+
	353	+	func setProxyGeneratedResponseHeaders(w http.ResponseWriter) {
311	354		w.Header().Set("Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'")
312	355		w.Header().Set("Cache-Control", "no-store")
313		-	http.Error(w, msg, code)
314	356		}

 package proxy
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
 		transport:     transport,
 		incomingReq:   r,
 		trustedNets:   h.trustedNets,
-		errorHandler: func(w http.ResponseWriter, _ *http.Request, proxyErr error) {
+		errorHandler: newProxyErrorHandler(
-			var maxBytesErr *http.MaxBytesError
+			log.WithField("target", targetURL.String()),
-			if errors.As(proxyErr, &maxBytesErr) {
+			proxyErrorHandlerConfig{
-				log.Warn().Err(proxyErr).Str("target", targetURL.String()).Msg("proxy error: request body too large")
+				requestTooLargeLog:  "proxy error: request body too large",
-				proxyError(w, "Request Entity Too Large", http.StatusRequestEntityTooLarge)
+				clientDisconnectLog: "proxy request canceled by client",
-				return
+				upstreamErrorLog:    "proxy error: connection failed",
-			}
+				upstreamErrorBody:   "Service Unavailable",
-			log.Error().Err(proxyErr).Str("target", targetURL.String()).Msg("proxy error: connection failed")
+			},
-			proxyError(w, "Service Unavailable", http.StatusServiceUnavailable)
+		),
-		},
 		modifyResponse: modifyResponse(maxResponseSize),
 	})
 			pr.Out.Host = targetURL.Host
 		},
 		Transport: h.registryTransport,
-		ErrorHandler: func(w http.ResponseWriter, _ *http.Request, proxyErr error) {
+		ErrorHandler: newProxyErrorHandler(
-			var maxBytesErr *http.MaxBytesError
+			log.WithField("registry_port", registryPort),
-			if errors.As(proxyErr, &maxBytesErr) {
+			proxyErrorHandlerConfig{
-				log.Warn().Err(proxyErr).Int("registry_port", registryPort).Msg("registry proxy error: request body too large")
+				requestTooLargeLog:  "registry proxy error: request body too large",
-				proxyError(w, "Request Entity Too Large", http.StatusRequestEntityTooLarge)
+				clientDisconnectLog: "registry proxy request canceled by client",
-				return
+				upstreamErrorLog:    "registry proxy error",
-			}
+				upstreamErrorBody:   "Registry Unavailable",
-			log.Error().Err(proxyErr).Int("registry_port", registryPort).Msg("registry proxy error")
+			},
-			proxyError(w, "Registry Unavailable", http.StatusServiceUnavailable)
+		),
-		},
 		// Strip browser-oriented security headers from upstream registry responses
 		// to avoid conflicts with Gordon's own headers on browser-facing routes.
 		ModifyResponse: func(resp *http.Response) error {
 	proxy.ServeHTTP(w, r)
 }
+type proxyErrorHandlerConfig struct {
+	requestTooLargeLog  string
+	clientDisconnectLog string
+	upstreamErrorLog    string
+	upstreamErrorBody   string
+}
+func newProxyErrorHandler(log zerowrap.Logger, cfg proxyErrorHandlerConfig) func(http.ResponseWriter, *http.Request, error) {
+	return func(w http.ResponseWriter, _ *http.Request, proxyErr error) {
+		switch {
+		case isRequestBodyTooLargeError(proxyErr):
+			log.Warn().Err(proxyErr).Msg(cfg.requestTooLargeLog)
+			proxyError(w, "Request Entity Too Large", http.StatusRequestEntityTooLarge)
+		case isClientDisconnectError(proxyErr):
+			log.Debug().Err(proxyErr).Msg(cfg.clientDisconnectLog)
+			clientClosedRequest(w)
+		default:
+			log.Error().Err(proxyErr).Msg(cfg.upstreamErrorLog)
+			proxyError(w, cfg.upstreamErrorBody, http.StatusServiceUnavailable)
+		}
+	}
+}
+func isRequestBodyTooLargeError(err error) bool {
+	_, ok := errors.AsType[*http.MaxBytesError](err)
+	return ok
+}
 type reverseProxyOptions struct {
 	targetURL      *url.URL
 	hostHeader     string
 	return l.ReadCloser.Close()
 }
+const statusClientClosedRequest = 499
+func isClientDisconnectError(err error) bool {
+	return errors.Is(err, context.Canceled) || errors.Is(err, http.ErrAbortHandler)
+}
+func clientClosedRequest(w http.ResponseWriter) {
+	setProxyGeneratedResponseHeaders(w)
+	w.WriteHeader(statusClientClosedRequest)
+}
 // proxyError writes an error response with security headers appropriate for
 // proxy-generated error pages.
 func proxyError(w http.ResponseWriter, msg string, code int) {
+	setProxyGeneratedResponseHeaders(w)
+	http.Error(w, msg, code)
+}
+func setProxyGeneratedResponseHeaders(w http.ResponseWriter) {
 	w.Header().Set("Content-Security-Policy", "default-src 'none'; frame-ancestors 'none'")
 	w.Header().Set("Cache-Control", "no-store")
-	http.Error(w, msg, code)
 }

M internal/adapters/in/http/proxy/forwarder_test.go +36 -0

@@ -1,6 +1,7 @@
1	1		package proxy
2	2
3	3		import (
	4	+	"context"
4	5		"fmt"
5	6		"net"
6	7		"net/http"

@@ -20,6 +21,12 @@ import (
20	21		"github.com/bnema/gordon/internal/domain"
21	22		)
22	23
	24	+	type roundTripFunc func(http.Request) (http.Response, error)
	25	+
	26	+	func (f roundTripFunc) RoundTrip(r http.Request) (http.Response, error) {
	27	+	return f(r)
	28	+	}
	29	+
23	30		func TestForwardToTarget_H2CEndToEnd(t *testing.T) {
24	31		var h2cProtos http.Protocols
25	32		h2cProtos.SetUnencryptedHTTP2(true)

@@ -185,6 +192,35 @@ func TestForwardToTarget_RouteHostPreservedForManagedTarget(t *testing.T) {
185	192		assert.Equal(t, "app.example.com", receivedForwardedHost)
186	193		}
187	194
	195	+	func TestForwardToTarget_ContextCanceled_Returns499(t *testing.T) {
	196	+	proxySvc := inmocks.NewMockProxyService(t)
	197	+
	198	+	proxySvc.EXPECT().ProxyConfig().Return(in.ProxyServiceConfig{})
	199	+	proxySvc.EXPECT().IsRegistryDomain("app.example.com").Return(false)
	200	+	proxySvc.EXPECT().GetTarget(mock.Anything, "app.example.com").Return(&domain.ProxyTarget{
	201	+	Host: "127.0.0.1",
	202	+	Port: 8080,
	203	+	ContainerID: "c-sse",
	204	+	Scheme: "http",
	205	+	}, nil)
	206	+	proxySvc.EXPECT().TrackInFlight("c-sse").Return(func() {})
	207	+
	208	+	handler := NewHandler(proxySvc, nil, zerowrap.Default())
	209	+	handler.appTransport = roundTripFunc(func(http.Request) (http.Response, error) {
	210	+	return nil, context.Canceled
	211	+	})
	212	+
	213	+	req := httptest.NewRequest(http.MethodGet, "http://app.example.com/api/notifications/sent-events", nil)
	214	+	req.Host = "app.example.com"
	215	+	w := httptest.NewRecorder()
	216	+	handler.ServeHTTP(w, req)
	217	+
	218	+	assert.Equal(t, statusClientClosedRequest, w.Code)
	219	+	assert.Equal(t, "no-store", w.Header().Get("Cache-Control"))
	220	+	assert.Equal(t, "default-src 'none'; frame-ancestors 'none'", w.Header().Get("Content-Security-Policy"))
	221	+	assert.Empty(t, w.Body.String())
	222	+	}
	223	+
188	224		func TestForwardToTarget_BackendDown_Returns503(t *testing.T) {
189	225		proxySvc := inmocks.NewMockProxyService(t)
190	226

@@ -1,6 +1,7 @@
1	package proxy	1		package proxy
2		2
3	import (	3		import (
		4	+	"context"
4	"fmt"	5		"fmt"
5	"net"	6		"net"
6	"net/http"	7		"net/http"

@@ -20,6 +21,12 @@ import (
20	"github.com/bnema/gordon/internal/domain"	21		"github.com/bnema/gordon/internal/domain"
21	)	22		)
22		23
		24	+	type roundTripFunc func(http.Request) (http.Response, error)
		25	+
		26	+	func (f roundTripFunc) RoundTrip(r http.Request) (http.Response, error) {
		27	+	return f(r)
		28	+	}
		29	+
23	func TestForwardToTarget_H2CEndToEnd(t *testing.T) {	30		func TestForwardToTarget_H2CEndToEnd(t *testing.T) {
24	var h2cProtos http.Protocols	31		var h2cProtos http.Protocols
25	h2cProtos.SetUnencryptedHTTP2(true)	32		h2cProtos.SetUnencryptedHTTP2(true)

@@ -185,6 +192,35 @@ func TestForwardToTarget_RouteHostPreservedForManagedTarget(t *testing.T) {
185	assert.Equal(t, "app.example.com", receivedForwardedHost)	192		assert.Equal(t, "app.example.com", receivedForwardedHost)
186	}	193		}
187		194
		195	+	func TestForwardToTarget_ContextCanceled_Returns499(t *testing.T) {
		196	+	proxySvc := inmocks.NewMockProxyService(t)
		197	+
		198	+	proxySvc.EXPECT().ProxyConfig().Return(in.ProxyServiceConfig{})
		199	+	proxySvc.EXPECT().IsRegistryDomain("app.example.com").Return(false)
		200	+	proxySvc.EXPECT().GetTarget(mock.Anything, "app.example.com").Return(&domain.ProxyTarget{
		201	+	Host: "127.0.0.1",
		202	+	Port: 8080,
		203	+	ContainerID: "c-sse",
		204	+	Scheme: "http",
		205	+	}, nil)
		206	+	proxySvc.EXPECT().TrackInFlight("c-sse").Return(func() {})
		207	+
		208	+	handler := NewHandler(proxySvc, nil, zerowrap.Default())
		209	+	handler.appTransport = roundTripFunc(func(http.Request) (http.Response, error) {
		210	+	return nil, context.Canceled
		211	+	})
		212	+
		213	+	req := httptest.NewRequest(http.MethodGet, "http://app.example.com/api/notifications/sent-events", nil)
		214	+	req.Host = "app.example.com"
		215	+	w := httptest.NewRecorder()
		216	+	handler.ServeHTTP(w, req)
		217	+
		218	+	assert.Equal(t, statusClientClosedRequest, w.Code)
		219	+	assert.Equal(t, "no-store", w.Header().Get("Cache-Control"))
		220	+	assert.Equal(t, "default-src 'none'; frame-ancestors 'none'", w.Header().Get("Content-Security-Policy"))
		221	+	assert.Empty(t, w.Body.String())
		222	+	}
		223	+
188	func TestForwardToTarget_BackendDown_Returns503(t *testing.T) {	224		func TestForwardToTarget_BackendDown_Returns503(t *testing.T) {
189	proxySvc := inmocks.NewMockProxyService(t)	225		proxySvc := inmocks.NewMockProxyService(t)
190		226