reverseproxy: Keep streams attached when writer detachment fails (#8139)

.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 }}

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
 
 

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 {

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))
+					}
+				}
+			}
+		})
+	}
+}