From 8ad83ebd4316d47fa40d11181eef148f018ec159 Mon Sep 17 00:00:00 2001 From: Francis Lavoie Date: Sun, 4 Oct 2026 12:46:34 -0400 Subject: [PATCH] reverseproxy: Keep streams attached when writer detachment fails (#8139) --- .github/workflows/ci.yml | 2 +- AGENTS.md | 6 + modules/caddyhttp/reverseproxy/streaming.go | 25 ++-- .../caddyhttp/reverseproxy/streaming_test.go | 139 ++++++++++++++++++ 4 files changed, 159 insertions(+), 13 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fb18946ce..5bee0c597 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -248,4 +248,4 @@ jobs: version: latest args: build --single-target --snapshot env: - TAG: ${{ github.head_ref || github.ref_name }} + TAG: ${{ github.event.pull_request.head.sha || github.sha }} diff --git a/AGENTS.md b/AGENTS.md index 894ab3b4e..b455cc33b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -117,6 +117,12 @@ Caddy is built around a **module system** where everything is a module registere Certificate management logic is also treated carefully, and is spread across caddyserver/caddy and caddyserver/certmagic repositories. +## Branch Naming and Module Queries + +- Use lowercase, hyphen-separated branch names without slashes, such as `fix-stream-detach-fallback`. This lets contributors use branch names directly in `go get module@branch` and `xcaddy build branch` with the default Go module proxy settings. +- Go's module-proxy version queries reject slash-containing branch names with `disallowed version string`. Shell quoting or URL-encoding the slash does not solve this. Use the branch's commit SHA instead; Go resolves it to a canonical version or pseudo-version. See [Go's version-query reference](https://go.dev/ref/mod#version-queries) and [version escaping rules](https://go.dev/src/cmd/vendor/golang.org/x/mod/module/module.go). +- `GOPROXY=direct` can resolve slash-containing branches through Git, but do not rely on bypassing the module proxy for normal contributor builds. CI snapshot builds should use the PR head commit SHA so they build the exact revision regardless of branch naming. + ## Quality Gates diff --git a/modules/caddyhttp/reverseproxy/streaming.go b/modules/caddyhttp/reverseproxy/streaming.go index 8f0569cb8..c64edda78 100644 --- a/modules/caddyhttp/reverseproxy/streaming.go +++ b/modules/caddyhttp/reverseproxy/streaming.go @@ -99,20 +99,10 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrit bufferSize := h.StreamBufferSize streamTimeout := time.Duration(h.StreamTimeout) - if h.StreamDetached { - // the return value should be true as it's not hijacked yet, - // but some middleware may wrap response writers incorrectly - if !caddyhttp.DetachResponseWriterAfterHijack(rw, true) { - if c := logger.Check(zap.DebugLevel, "detaching connection failed"); c != nil { - c.Write(zap.String("tip", "check if your response writers have an Unwrap method or if already hijacked")) - } - } - } - var ( conn io.ReadWriteCloser brw *bufio.ReadWriter - detached = h.StreamDetached + detached bool ) // websocket over http2 or http3 if extended connect is enabled, // assuming backend doesn't support this, the request will be @@ -122,7 +112,6 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrit if body, ok := caddyhttp.GetVar(req.Context(), "extended_connect_websocket_body").(io.ReadCloser); ok { // websocket over extended connect can't be detached. rw and req.Body // are only valid while the handler goroutine is running - detached = false req.Body = body rw.Header().Del("Upgrade") rw.Header().Del("Connection") @@ -145,6 +134,18 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrit // bufio is not needed, use minimal buffer brw = bufio.NewReadWriter(bufio.NewReaderSize(conn, 1), bufio.NewWriterSize(conn, 1)) } else { + if h.StreamDetached { + detached = caddyhttp.DetachResponseWriterAfterHijack(rw, true) + if !detached { + // Some outer writers may have accepted detachment before an + // inner writer refused it. Restore attached byte accounting + // and keep the handler running for the lifetime of the tunnel. + caddyhttp.DetachResponseWriterAfterHijack(rw, false) + if c := logger.Check(zap.DebugLevel, "detaching connection failed; keeping stream attached"); c != nil { + c.Write(zap.String("tip", "check if your response writers have an Unwrap method or if already hijacked")) + } + } + } rw.WriteHeader(res.StatusCode) if c := logger.Check(zap.DebugLevel, "upgrading connection"); c != nil { diff --git a/modules/caddyhttp/reverseproxy/streaming_test.go b/modules/caddyhttp/reverseproxy/streaming_test.go index 2b86598e0..fe66e05a8 100644 --- a/modules/caddyhttp/reverseproxy/streaming_test.go +++ b/modules/caddyhttp/reverseproxy/streaming_test.go @@ -642,3 +642,142 @@ func TestDetachedTrackerRemovalAllowsConcurrentConnectionDeletion(t *testing.T) t.Error("removed upstream connection was not closed") } } + +// opaqueUpgradeWriter supports hijacking but hides the recorder from the +// detachment helper, as a middleware wrapper without Unwrap would do. +type opaqueUpgradeWriter struct { + http.ResponseWriter +} + +func (w opaqueUpgradeWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { + return http.NewResponseController(w.ResponseWriter).Hijack() +} + +type rejectingUpgradeWriter struct { + *caddyhttp.ResponseWriterWrapper +} + +func (rejectingUpgradeWriter) DetachAfterHijack(detached bool) bool { + return !detached +} + +type pipeUpgradeWriter struct { + *httptest.ResponseRecorder + conn net.Conn +} + +func (w pipeUpgradeWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { + return w.conn, bufio.NewReadWriter(bufio.NewReader(w.conn), bufio.NewWriter(w.conn)), nil +} + +func TestUpgradeDetachmentRequiresWriterSupport(t *testing.T) { + for _, tc := range []struct { + name string + wrap func(http.ResponseWriter) http.ResponseWriter + detached bool + }{ + {name: "opaque wrapper", wrap: func(w http.ResponseWriter) http.ResponseWriter { return opaqueUpgradeWriter{w} }}, + {name: "inner writer rejects detachment", wrap: func(w http.ResponseWriter) http.ResponseWriter { + rejecting := rejectingUpgradeWriter{&caddyhttp.ResponseWriterWrapper{ResponseWriter: w}} + return caddyhttp.NewResponseRecorder(rejecting, nil, nil) + }}, + {name: "supported recorder", wrap: func(w http.ResponseWriter) http.ResponseWriter { return w }, detached: true}, + } { + t.Run(tc.name, func(t *testing.T) { + front, client := net.Pipe() + back, backend := net.Pipe() + defer front.Close() + defer client.Close() + defer back.Close() + defer backend.Close() + if err := client.SetDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatal(err) + } + echoDone := make(chan struct{}) + go func() { + defer close(echoDone) + _, _ = io.Copy(backend, backend) + }() + h := &Handler{StreamDetached: true, logger: caddy.Log(), tunnelTracker: newTunnelTracker(caddy.Log(), 0)} + initReverseProxyMetrics(h, prometheus.NewRegistry()) + recorder := caddyhttp.NewResponseRecorder(pipeUpgradeWriter{httptest.NewRecorder(), front}, nil, nil) + writer := tc.wrap(recorder) + req := prepareTestRequest(httptest.NewRequest(http.MethodGet, "/upgrade", nil)) + req.Header.Set("Connection", "Upgrade") + req.Header.Set("Upgrade", "test-stream") + res := &http.Response{StatusCode: http.StatusSwitchingProtocols, Header: req.Header.Clone(), Body: back} + handlerDone := make(chan struct{}) + go func() { + defer close(handlerDone) + h.handleUpgradeResponse(caddy.Log(), writer, req, res, "detachment-test") + }() + const payload = "echo" + if _, err := io.WriteString(client, payload); err != nil { + t.Fatal(err) + } + received := make([]byte, len(payload)) + if _, err := io.ReadFull(client, received); err != nil { + t.Fatal(err) + } + if string(received) != payload { + t.Fatalf("echo = %q, want %q", received, payload) + } + h.tunnelTracker.mu.Lock() + for _, conn := range h.tunnelTracker.connections { + if conn.detached != tc.detached { + t.Errorf("tracked detached = %v, want %v", conn.detached, tc.detached) + } + } + h.tunnelTracker.mu.Unlock() + if tc.detached { + select { + case <-handlerDone: + case <-time.After(time.Second): + t.Error("supported detached tunnel kept the handler running") + } + } else { + select { + case <-handlerDone: + t.Error("handler returned while the writer chain remained attached") + case <-time.After(100 * time.Millisecond): + } + } + client.Close() + backend.Close() + select { + case <-handlerDone: + case <-time.After(time.Second): + t.Fatal("handler did not finish after stream close") + } + <-echoDone + // Wait for detached tunnel cleanup too, so no copy goroutine can + // still be updating counters after this subtest finishes. + deadline := time.Now().Add(time.Second) + for { + h.tunnelTracker.mu.Lock() + remaining := len(h.tunnelTracker.connections) + h.tunnelTracker.mu.Unlock() + if remaining == 0 { + break + } + if time.Now().After(deadline) { + t.Fatal("tunnel connections were not unregistered") + } + time.Sleep(time.Millisecond) + } + // Attached byte counters are safe to inspect after the handler has waited + // for both copying goroutines. Partial detachment must also be undone. + if !tc.detached { + recorders := []caddyhttp.ResponseRecorder{recorder} + if outer, ok := writer.(caddyhttp.ResponseRecorder); ok { + recorders = append(recorders, outer) + } + for _, rr := range recorders { + if rr.Size() != len(payload) { + t.Errorf("attached recorder size = %d, want %d", rr.Size(), len(payload)) + } + } + } + }) + } +}