diff --git a/api/middleware/log.go b/api/middleware/log.go index 1a5ef7400..c9cb0720a 100644 --- a/api/middleware/log.go +++ b/api/middleware/log.go @@ -8,14 +8,10 @@ import ( "github.com/gin-gonic/gin" "opencsg.com/csghub-server/api/httpbase" + commonutils "opencsg.com/csghub-server/common/utils/common" "opencsg.com/csghub-server/common/utils/trace" ) -// Status 499 is a non-standard code introduced by nginx to indicate -// "Client Closed Request" — the client disconnected before the server -// finished processing. -const StatusClientClosedRequest = 499 - func Log() gin.HandlerFunc { return func(ctx *gin.Context) { if ctx.Request.URL.Path == "/healthz" { @@ -43,7 +39,7 @@ func Log() gin.HandlerFunc { // If the client disconnected (timeout or explicit cancel), // override the logged status to 499. if ctx.Request.Context().Err() == context.Canceled { - status = StatusClientClosedRequest + status = commonutils.StatusClientClosedRequest } // Derive a non-canceled context for the log call. diff --git a/api/middleware/log_test.go b/api/middleware/log_test.go index 29951ce64..9ff41c789 100644 --- a/api/middleware/log_test.go +++ b/api/middleware/log_test.go @@ -13,6 +13,7 @@ import ( "github.com/gin-gonic/gin" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + commonutils "opencsg.com/csghub-server/common/utils/common" ) // captureLog is a helper that installs a slog JSON handler writing to a buffer @@ -82,7 +83,7 @@ func TestLog_ClientCancelBeforeHandler(t *testing.T) { logs := readLogs() require.Len(t, logs, 1, "must emit a log even when client canceled") - assert.Equal(t, float64(StatusClientClosedRequest), logs[0]["status"], "status should be 499 for client cancel") + assert.Equal(t, float64(commonutils.StatusClientClosedRequest), logs[0]["status"], "status should be 499 for client cancel") assert.Equal(t, "/test", logs[0]["url"]) } @@ -116,7 +117,7 @@ func TestLog_ClientCancelDuringSlowHandler(t *testing.T) { logs := readLogs() require.Len(t, logs, 1, "must emit a log when client cancels during slow handler") - assert.Equal(t, float64(StatusClientClosedRequest), logs[0]["status"], "status should be 499") + assert.Equal(t, float64(commonutils.StatusClientClosedRequest), logs[0]["status"], "status should be 499") } func TestLog_HandlerPanicWithRecovery(t *testing.T) { diff --git a/builder/proxy/reverse_proxy.go b/builder/proxy/reverse_proxy.go index ad4e37e4f..6c3fbb19d 100644 --- a/builder/proxy/reverse_proxy.go +++ b/builder/proxy/reverse_proxy.go @@ -1,12 +1,15 @@ package proxy import ( + "context" + "errors" "log/slog" "net/http" "net/http/httputil" "net/url" "github.com/openai/openai-go/v3" + commonutils "opencsg.com/csghub-server/common/utils/common" "opencsg.com/csghub-server/common/utils/trace" ) @@ -54,6 +57,15 @@ func (rp *reverseProxyImpl) ServeHTTP(w http.ResponseWriter, r *http.Request, ap } }() proxy := httputil.NewSingleHostReverseProxy(rp.target) + proxy.ErrorHandler = func(rw http.ResponseWriter, req *http.Request, err error) { + if errors.Is(err, context.Canceled) { + slog.InfoContext(context.WithoutCancel(req.Context()), "http: proxy request canceled", slog.Any("error", err)) + rw.WriteHeader(commonutils.StatusClientClosedRequest) + return + } + slog.ErrorContext(req.Context(), "http: proxy error", slog.Any("error", err)) + rw.WriteHeader(http.StatusBadGateway) + } proxy.Director = func(req *http.Request) { if len(svcHost) > 0 { slog.Info("update reverse proxy header host", slog.Any("svc-host", svcHost)) diff --git a/builder/proxy/reverse_proxy_test.go b/builder/proxy/reverse_proxy_test.go index d3a37b33a..407088d18 100644 --- a/builder/proxy/reverse_proxy_test.go +++ b/builder/proxy/reverse_proxy_test.go @@ -1,11 +1,13 @@ package proxy import ( + "context" "net/http" "net/http/httptest" "testing" "github.com/stretchr/testify/require" + commonutils "opencsg.com/csghub-server/common/utils/common" ) func TestReverseProxy_AcceptEncodingDefaultGzip(t *testing.T) { @@ -47,6 +49,24 @@ func TestReverseProxy_AcceptEncodingDisabled(t *testing.T) { require.Equal(t, "identity", downstreamAcceptEncoding) } +func TestReverseProxy_ContextCanceledWritesClientClosed(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + rp, err := NewReverseProxy(server.URL) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + req := httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx) + resp := httptest.NewRecorder() + rp.ServeHTTP(resp, req, "", "") + + require.Equal(t, commonutils.StatusClientClosedRequest, resp.Code) +} + // import ( // "bytes" // "io" diff --git a/common/utils/common/http_status.go b/common/utils/common/http_status.go new file mode 100644 index 000000000..0dca01d33 --- /dev/null +++ b/common/utils/common/http_status.go @@ -0,0 +1,5 @@ +package common + +// StatusClientClosedRequest is the non-standard nginx status code for a client +// closing the request before the server finishes processing it. +const StatusClientClosedRequest = 499