@marcel91200 / caddy Mirror · read-only / commits / 7ef5c60e
Fork 0

Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS

Mirrored from https://github.com/Slazee/caddy.git

Synced Nov 11, 03:30 UTC

reverseproxy: Optionally detach stream (websockets) from config lifecycle (#7649)

* reverseproxy: Optionally detach stream (websockets) from config lifecycle * lint * Improved logging facilities * Adjustments from Weidi's review * record bytes read and written for response writers unless detached * simplify streaming handling * correctly close detached streams * make handleUpgradeTunnel a standalone func * fix tests * fix tests * change the log level if hijacking without writing a status code first * only clean up connections when stopped * Move type and const down to the bottom * Rename to tunnelTracker, reflow some comments * Add note about capturing h * Rename to stream_detached * More comments * reverseproxy: Preserve half-close behavior after rebase Keep master half-close propagation and its unit coverage in the refactored tunnel lifecycle. Adapt response finalization tests to the selected upstream argument. AI-assisted-by: Codex (GPT-6) * reverseproxy: Fix detached stream cleanup and recorder chains Snapshot the tracker registry before closing connections, and unregister empty trackers outside their mutex. Mark trackers stopped at cleanup entry so upstream removal closes detached streams even with stream_close_delay, while attached streams retain their delay. Detach every supported response writer in the middleware chain so inner recorders do not retain state or count detached traffic. Add regressions for all four failures and an access-log/encode/intercept reload integration test. AI-assisted-by: Codex (GPT-6) --------- Co-authored-by: WeidiDeng <weidi_deng@icloud.com>

Francis Lavoie committed Oct 4, 2026 at 07:45 UTC 7ef5c60ed0f1f0a5d87f9f8db4a290dbb50e3e48
15 files changed +2262 -163
.github/workflows/ci.yml
+3 -1
@@ -132,6 +132,8 @@ jobs:
132 - name: Run tests
133 # id: step_test
134 # continue-on-error: true
135 + env:
136 + GODEBUG: http2xconnect=1
137 run: |
138 # (go test -coverprofile=cover-profile.out -race ./... 2>&1) > test-results/test-result.out
139 go test -coverprofile="cover-profile.out" -short -race ./...
@@ -191,7 +193,7 @@ jobs:
193 retries=3
194 exit_code=0
195 while ((retries > 0)); do
194 - CGO_ENABLED=0 go test -p 1 ./...
196 + GODEBUG=http2xconnect=1 CGO_ENABLED=0 go test -p 1 ./...
197 exit_code=$?
198 if ((exit_code == 0)); then
199 break
caddytest/integration/reverseproxy_extended_connect_test.go new
+328
@@ -0,0 +1,328 @@
1 +package integration
2 +
3 +import (
4 + "bytes"
5 + "crypto/tls"
6 + "errors"
7 + "fmt"
8 + "io"
9 + "net"
10 + "net/http"
11 + "strings"
12 + "testing"
13 + "time"
14 +
15 + "golang.org/x/net/http2"
16 + "golang.org/x/net/http2/hpack"
17 +
18 + "github.com/caddyserver/caddy/v2/caddytest"
19 +)
20 +
21 +var errExtendedConnectUnsupportedByPeer = errors.New("peer did not advertise RFC 8441 extended CONNECT support")
22 +
23 +func TestReverseProxyExtendedConnectOverH2(t *testing.T) {
24 + tester := caddytest.NewTester(t)
25 + backend := newWebsocketUpgradeEchoBackend(t)
26 + defer backend.Close()
27 +
28 + tester.InitServer(fmt.Sprintf(`
29 +{
30 + admin localhost:2999
31 + http_port 9080
32 + https_port 9443
33 + grace_period 1ns
34 + skip_install_trust
35 + servers :9443 {
36 + protocols h2
37 + }
38 +}
39 +
40 +https://localhost:9443 {
41 + reverse_proxy %s
42 +}
43 +`, backend.addr), "caddyfile")
44 +
45 + const payload = "extended-connect-echo\n"
46 + if err := assertExtendedConnectH2Echo("localhost:9443", payload); err != nil {
47 + if errors.Is(err, errExtendedConnectUnsupportedByPeer) {
48 + t.Skipf("skipping extended CONNECT integration test: %v", err)
49 + }
50 + t.Fatalf("extended connect h2 echo failed: %v", err)
51 + }
52 +}
53 +
54 +func assertExtendedConnectH2Echo(addr, payload string) error {
55 + conn, err := tlsDialH2(addr)
56 + if err != nil {
57 + return fmt.Errorf("dialing h2 tls: %w", err)
58 + }
59 + defer conn.Close()
60 +
61 + if err := conn.SetDeadline(time.Now().Add(5 * time.Second)); err != nil {
62 + return fmt.Errorf("setting deadline: %w", err)
63 + }
64 +
65 + fr := http2.NewFramer(conn, conn)
66 +
67 + if _, err := conn.Write([]byte(http2.ClientPreface)); err != nil {
68 + return fmt.Errorf("writing client preface: %w", err)
69 + }
70 + if err := fr.WriteSettings(http2.Setting{ID: http2.SettingEnableConnectProtocol, Val: 1}); err != nil {
71 + return fmt.Errorf("writing client settings: %w", err)
72 + }
73 +
74 + supported, err := waitForServerSettings(fr)
75 + if err != nil {
76 + return err
77 + }
78 + if !supported {
79 + return errExtendedConnectUnsupportedByPeer
80 + }
81 + if err := waitForSettingsAck(fr); err != nil {
82 + return err
83 + }
84 +
85 + if err := writeExtendedConnectHeaders(fr, addr); err != nil {
86 + return err
87 + }
88 +
89 + status, err := readResponseStatus(fr, 1)
90 + if err != nil {
91 + return err
92 + }
93 + if status != "200" {
94 + return fmt.Errorf("unexpected extended connect status: got=%s want=200", status)
95 + }
96 +
97 + if err := fr.WriteData(1, false, []byte(payload)); err != nil {
98 + return fmt.Errorf("writing stream data: %w", err)
99 + }
100 +
101 + echo, err := readStreamData(fr, 1, len(payload))
102 + if err != nil {
103 + return err
104 + }
105 + if echo != payload {
106 + return fmt.Errorf("unexpected echoed payload: got=%q want=%q", echo, payload)
107 + }
108 +
109 + _ = fr.WriteRSTStream(1, http2.ErrCodeNo)
110 + return nil
111 +}
112 +
113 +func tlsDialH2(addr string) (net.Conn, error) {
114 + var lastErr error
115 + for i := 0; i < 30; i++ {
116 + dialer := &net.Dialer{Timeout: 2 * time.Second}
117 + conn, err := tls.DialWithDialer(dialer, "tcp", addr, &tls.Config{
118 + ServerName: "localhost",
119 + InsecureSkipVerify: true,
120 + NextProtos: []string{"h2"},
121 + })
122 + if err == nil {
123 + return conn, nil
124 + }
125 + lastErr = err
126 + time.Sleep(100 * time.Millisecond)
127 + }
128 + return nil, lastErr
129 +}
130 +
131 +func waitForServerSettings(fr *http2.Framer) (bool, error) {
132 + for {
133 + frame, err := fr.ReadFrame()
134 + if err != nil {
135 + return false, fmt.Errorf("reading frame before connect: %w", err)
136 + }
137 + settings, ok := frame.(*http2.SettingsFrame)
138 + if !ok {
139 + continue
140 + }
141 + if settings.IsAck() {
142 + continue
143 + }
144 +
145 + supported := false
146 + if err := settings.ForeachSetting(func(s http2.Setting) error {
147 + if s.ID == http2.SettingEnableConnectProtocol && s.Val == 1 {
148 + supported = true
149 + }
150 + return nil
151 + }); err != nil {
152 + return false, fmt.Errorf("reading server settings: %w", err)
153 + }
154 +
155 + if err := fr.WriteSettingsAck(); err != nil {
156 + return false, fmt.Errorf("writing settings ack: %w", err)
157 + }
158 + return supported, nil
159 + }
160 +}
161 +
162 +func waitForSettingsAck(fr *http2.Framer) error {
163 + for {
164 + frame, err := fr.ReadFrame()
165 + if err != nil {
166 + return fmt.Errorf("reading settings ack: %w", err)
167 + }
168 + settings, ok := frame.(*http2.SettingsFrame)
169 + if ok && settings.IsAck() {
170 + return nil
171 + }
172 + }
173 +}
174 +
175 +func writeExtendedConnectHeaders(fr *http2.Framer, addr string) error {
176 + var hb bytes.Buffer
177 + enc := hpack.NewEncoder(&hb)
178 + for _, hf := range []hpack.HeaderField{
179 + {Name: ":method", Value: "CONNECT"},
180 + {Name: ":scheme", Value: "https"},
181 + {Name: ":authority", Value: addr},
182 + {Name: ":path", Value: "/upgrade"},
183 + {Name: ":protocol", Value: "websocket"},
184 + } {
185 + if err := enc.WriteField(hf); err != nil {
186 + return fmt.Errorf("encoding request headers: %w", err)
187 + }
188 + }
189 +
190 + if err := fr.WriteHeaders(http2.HeadersFrameParam{
191 + StreamID: 1,
192 + BlockFragment: hb.Bytes(),
193 + EndHeaders: true,
194 + EndStream: false,
195 + }); err != nil {
196 + return fmt.Errorf("writing extended connect headers: %w", err)
197 + }
198 + return nil
199 +}
200 +
201 +func readResponseStatus(fr *http2.Framer, streamID uint32) (string, error) {
202 + var block bytes.Buffer
203 +
204 + for {
205 + frame, err := fr.ReadFrame()
206 + if err != nil {
207 + return "", fmt.Errorf("reading response headers: %w", err)
208 + }
209 + if rst, ok := frame.(*http2.RSTStreamFrame); ok && rst.StreamID == streamID {
210 + return "", fmt.Errorf("stream reset before response headers: %s", rst.ErrCode)
211 + }
212 +
213 + h, ok := frame.(*http2.HeadersFrame)
214 + if !ok || h.StreamID != streamID {
215 + continue
216 + }
217 +
218 + if _, err := block.Write(h.HeaderBlockFragment()); err != nil {
219 + return "", fmt.Errorf("buffering response header fragment: %w", err)
220 + }
221 + for !h.HeadersEnded() {
222 + next, err := fr.ReadFrame()
223 + if err != nil {
224 + return "", fmt.Errorf("reading continuation frame: %w", err)
225 + }
226 + c, ok := next.(*http2.ContinuationFrame)
227 + if !ok || c.StreamID != streamID {
228 + continue
229 + }
230 + if _, err := block.Write(c.HeaderBlockFragment()); err != nil {
231 + return "", fmt.Errorf("buffering continuation fragment: %w", err)
232 + }
233 + if c.HeadersEnded() {
234 + break
235 + }
236 + }
237 + break
238 + }
239 +
240 + var status string
241 + dec := hpack.NewDecoder(4096, func(f hpack.HeaderField) {
242 + if f.Name == ":status" {
243 + status = f.Value
244 + }
245 + })
246 + if _, err := dec.Write(block.Bytes()); err != nil {
247 + return "", fmt.Errorf("decoding response header block: %w", err)
248 + }
249 + if status == "" {
250 + return "", fmt.Errorf("missing :status in response headers")
251 + }
252 + return status, nil
253 +}
254 +
255 +func readStreamData(fr *http2.Framer, streamID uint32, n int) (string, error) {
256 + buf := make([]byte, 0, n)
257 + for len(buf) < n {
258 + frame, err := fr.ReadFrame()
259 + if err != nil {
260 + return "", fmt.Errorf("reading stream data: %w", err)
261 + }
262 + d, ok := frame.(*http2.DataFrame)
263 + if !ok || d.StreamID != streamID {
264 + continue
265 + }
266 + buf = append(buf, d.Data()...)
267 + }
268 + return string(buf[:n]), nil
269 +}
270 +
271 +type websocketUpgradeEchoBackend struct {
272 + addr string
273 + ln net.Listener
274 + server *http.Server
275 +}
276 +
277 +func newWebsocketUpgradeEchoBackend(t *testing.T) *websocketUpgradeEchoBackend {
278 + t.Helper()
279 +
280 + backend := &websocketUpgradeEchoBackend{}
281 + backend.server = &http.Server{
282 + Handler: http.HandlerFunc(backend.serveHTTP),
283 + }
284 +
285 + ln, err := net.Listen("tcp", "127.0.0.1:0")
286 + if err != nil {
287 + t.Fatalf("listening for websocket backend: %v", err)
288 + }
289 + backend.ln = ln
290 + backend.addr = ln.Addr().String()
291 +
292 + go func() {
293 + _ = backend.server.Serve(ln)
294 + }()
295 +
296 + return backend
297 +}
298 +
299 +func (b *websocketUpgradeEchoBackend) serveHTTP(w http.ResponseWriter, r *http.Request) {
300 + if !strings.EqualFold(r.Header.Get("Connection"), "Upgrade") || !strings.EqualFold(r.Header.Get("Upgrade"), "websocket") {
301 + http.Error(w, "upgrade required", http.StatusUpgradeRequired)
302 + return
303 + }
304 +
305 + hijacker, ok := w.(http.Hijacker)
306 + if !ok {
307 + http.Error(w, "hijacking not supported", http.StatusInternalServerError)
308 + return
309 + }
310 +
311 + conn, rw, err := hijacker.Hijack()
312 + if err != nil {
313 + return
314 + }
315 +
316 + _, _ = rw.WriteString("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n")
317 + _ = rw.Flush()
318 +
319 + go func() {
320 + defer conn.Close()
321 + _, _ = io.Copy(conn, conn)
322 + }()
323 +}
324 +
325 +func (b *websocketUpgradeEchoBackend) Close() {
326 + _ = b.server.Close()
327 + _ = b.ln.Close()
328 +}
caddytest/integration/reverseproxy_upgrade_handlers_test.go new
+175
@@ -0,0 +1,175 @@
1 +package integration
2 +
3 +import (
4 + "bufio"
5 + "fmt"
6 + "io"
7 + "net"
8 + "net/textproto"
9 + "strings"
10 + "testing"
11 + "time"
12 +
13 + "github.com/caddyserver/caddy/v2/caddytest"
14 +)
15 +
16 +func TestReverseProxyUpgradeWithEncode(t *testing.T) {
17 + tester := caddytest.NewTester(t)
18 + backend := newUpgradeEchoBackend(t)
19 + defer backend.Close()
20 +
21 + tester.InitServer(fmt.Sprintf(`
22 +{
23 + admin localhost:2999
24 + http_port 9080
25 + https_port 9443
26 + grace_period 1ns
27 + skip_install_trust
28 +}
29 +
30 +localhost:9080 {
31 + route {
32 + encode gzip
33 + reverse_proxy %s
34 + }
35 +}
36 +`, backend.addr), "caddyfile")
37 +
38 + client := newUpgradedStreamClientWithHeaders(t, map[string]string{
39 + "Accept-Encoding": "gzip",
40 + })
41 + defer client.Close()
42 +
43 + if err := client.echo("encode-upgrade\n"); err != nil {
44 + t.Fatalf("upgraded stream echo through encode failed: %v", err)
45 + }
46 +}
47 +
48 +func TestReverseProxyUpgradeWithInterceptHandleResponse(t *testing.T) {
49 + tester := caddytest.NewTester(t)
50 + backend := newUpgradeEchoBackend(t)
51 + defer backend.Close()
52 +
53 + tester.InitServer(fmt.Sprintf(`
54 +{
55 + admin localhost:2999
56 + http_port 9080
57 + https_port 9443
58 + grace_period 1ns
59 + skip_install_trust
60 +}
61 +
62 +localhost:9080 {
63 + route {
64 + intercept {
65 + @upgrade status 101
66 + handle_response @upgrade {
67 + respond "should-not-run"
68 + }
69 + }
70 + reverse_proxy %s
71 + }
72 +}
73 +`, backend.addr), "caddyfile")
74 +
75 + client := newUpgradedStreamClientWithHeaders(t, nil)
76 + defer client.Close()
77 +
78 + if err := client.echo("intercept-upgrade\n"); err != nil {
79 + t.Fatalf("upgraded stream echo through intercept failed: %v", err)
80 + }
81 +}
82 +
83 +func TestReverseProxyDetachedUpgradeWithAccessLogAndIntercept(t *testing.T) {
84 + tester := caddytest.NewTester(t)
85 + backend := newUpgradeEchoBackend(t)
86 + defer backend.Close()
87 + config := func(marker string) string {
88 + return fmt.Sprintf(`
89 +{
90 + admin localhost:2999
91 + http_port 9080
92 + https_port 9443
93 + grace_period 1ns
94 + skip_install_trust
95 +}
96 +localhost:9080 {
97 + log {
98 + output discard
99 + }
100 + route {
101 + header X-Reload %s
102 + encode gzip
103 + intercept {
104 + @upgrade status 101
105 + handle_response @upgrade {
106 + respond "should-not-run"
107 + }
108 + }
109 + reverse_proxy %s {
110 + stream_detached
111 + }
112 + }
113 +}
114 +`, marker, backend.addr)
115 + }
116 + tester.InitServer(config("before"), "caddyfile")
117 + client := newUpgradedStreamClientWithHeaders(t, map[string]string{"Accept-Encoding": "gzip"})
118 + defer client.Close()
119 + if err := client.echo("before-reload\n"); err != nil {
120 + t.Fatalf("detached stream echo before reload failed: %v", err)
121 + }
122 + tester.InitServer(config("after"), "caddyfile")
123 + if err := client.echo("after-reload\n"); err != nil {
124 + t.Fatalf("detached stream echo after reload failed: %v", err)
125 + }
126 +}
127 +
128 +func newUpgradedStreamClientWithHeaders(t *testing.T, extraHeaders map[string]string) *upgradedStreamClient {
129 + t.Helper()
130 +
131 + conn, err := net.DialTimeout("tcp", "127.0.0.1:9080", 5*time.Second)
132 + if err != nil {
133 + t.Fatalf("dialing caddy: %v", err)
134 + }
135 +
136 + requestLines := []string{
137 + "GET /upgrade HTTP/1.1",
138 + "Host: localhost:9080",
139 + "Connection: Upgrade",
140 + "Upgrade: stress-stream",
141 + }
142 + for k, v := range extraHeaders {
143 + requestLines = append(requestLines, k+": "+v)
144 + }
145 + requestLines = append(requestLines, "", "")
146 +
147 + if _, err := io.WriteString(conn, strings.Join(requestLines, "\r\n")); err != nil {
148 + _ = conn.Close()
149 + t.Fatalf("writing upgrade request: %v", err)
150 + }
151 +
152 + reader := bufio.NewReader(conn)
153 + tproto := textproto.NewReader(reader)
154 + statusLine, err := tproto.ReadLine()
155 + if err != nil {
156 + _ = conn.Close()
157 + t.Fatalf("reading upgrade status line: %v", err)
158 + }
159 + if !strings.Contains(statusLine, "101") {
160 + _ = conn.Close()
161 + t.Fatalf("unexpected upgrade status: %s", statusLine)
162 + }
163 +
164 + headers, err := tproto.ReadMIMEHeader()
165 + if err != nil {
166 + _ = conn.Close()
167 + t.Fatalf("reading upgrade headers: %v", err)
168 + }
169 + if !strings.EqualFold(headers.Get("Connection"), "Upgrade") {
170 + _ = conn.Close()
171 + t.Fatalf("unexpected upgrade response headers: %v", headers)
172 + }
173 +
174 + return &upgradedStreamClient{conn: conn, reader: reader}
175 +}
caddytest/integration/stream_reload_stress_test.go new
+504
@@ -0,0 +1,504 @@
1 +package integration
2 +
3 +import (
4 + "bufio"
5 + "bytes"
6 + "fmt"
7 + "io"
8 + "net"
9 + "net/http"
10 + "net/textproto"
11 + "os"
12 + "runtime"
13 + "runtime/debug"
14 + "runtime/pprof"
15 + "strconv"
16 + "strings"
17 + "sync"
18 + "testing"
19 + "time"
20 +
21 + "github.com/caddyserver/caddy/v2/caddytest"
22 +)
23 +
24 +const (
25 + defaultStressStreamCount = 1
26 + defaultStressReloadCount = 1
27 + defaultStressCloseDelay = 500 * time.Millisecond
28 +)
29 +
30 +func TestReverseProxyReloadStressUpgradedStreamsHeapProfiles(t *testing.T) {
31 + tester := caddytest.NewTester(t).WithDefaultOverrides(caddytest.Config{
32 + LoadRequestTimeout: 30 * time.Second,
33 + TestRequestTimeout: 30 * time.Second,
34 + })
35 +
36 + backend := newUpgradeEchoBackend(t)
37 + defer backend.Close()
38 +
39 + // Three scenarios, each sequential so they don't share Caddy state:
40 + //
41 + // legacy – no delay, close on reload immediately (old default)
42 + // close_delay – stream_close_delay, the old "keep-alive workaround"
43 + // detached – stream_detached, the new explicit detached flag
44 + //
45 + // Reloads are spread across time and interleaved with echo-checks so
46 + // stream health is exercised at each reload boundary, not only at the end.
47 + legacy := runReloadStress(t, tester, backend.addr, "legacy", false, 0)
48 + closeDelay := runReloadStress(t, tester, backend.addr, "close_delay", false, stressCloseDelay(t))
49 + detached := runReloadStress(t, tester, backend.addr, "detached", true, 0)
50 +
51 + if legacy.aliveAfterReloads != 0 {
52 + t.Fatalf("legacy mode left %d upgraded streams alive after reloads", legacy.aliveAfterReloads)
53 + }
54 + if closeDelay.aliveBeforeDelayExpiry == 0 {
55 + t.Fatalf("close_delay mode: all streams closed before delay expired (expected them alive)")
56 + }
57 + if closeDelay.aliveAfterReloads != 0 {
58 + t.Fatalf("close_delay mode left %d upgraded streams alive after delay expiry", closeDelay.aliveAfterReloads)
59 + }
60 + if detached.aliveAfterReloads != detached.streamCount {
61 + t.Fatalf("detached mode kept %d/%d upgraded streams alive after reloads", detached.aliveAfterReloads, detached.streamCount)
62 + }
63 +
64 + t.Logf("legacy heap: before=%s mid=%s after=%s delta(before→after)=%s objects(before=%d after=%d) handler_frames(before=%d after=%d)",
65 + formatBytes(legacy.beforeReload.HeapInuse),
66 + formatBytes(legacy.midReload.HeapInuse),
67 + formatBytes(legacy.afterReload.HeapInuse),
68 + formatBytesDiff(legacy.beforeReload.HeapInuse, legacy.afterReload.HeapInuse),
69 + legacy.beforeReload.HeapObjects, legacy.afterReload.HeapObjects,
70 + legacy.beforeReload.handlerFrames, legacy.afterReload.handlerFrames,
71 + )
72 + t.Logf("close_delay heap: before=%s mid=%s after=%s delta(before→after)=%s objects(before=%d after=%d) handler_frames(before=%d after=%d)",
73 + formatBytes(closeDelay.beforeReload.HeapInuse),
74 + formatBytes(closeDelay.midReload.HeapInuse),
75 + formatBytes(closeDelay.afterReload.HeapInuse),
76 + formatBytesDiff(closeDelay.beforeReload.HeapInuse, closeDelay.afterReload.HeapInuse),
77 + closeDelay.beforeReload.HeapObjects, closeDelay.afterReload.HeapObjects,
78 + closeDelay.beforeReload.handlerFrames, closeDelay.afterReload.handlerFrames,
79 + )
80 + t.Logf("detached heap: before=%s mid=%s after=%s delta(before→after)=%s objects(before=%d after=%d) handler_frames(before=%d after=%d)",
81 + formatBytes(detached.beforeReload.HeapInuse),
82 + formatBytes(detached.midReload.HeapInuse),
83 + formatBytes(detached.afterReload.HeapInuse),
84 + formatBytesDiff(detached.beforeReload.HeapInuse, detached.afterReload.HeapInuse),
85 + detached.beforeReload.HeapObjects, detached.afterReload.HeapObjects,
86 + detached.beforeReload.handlerFrames, detached.afterReload.handlerFrames,
87 + )
88 +}
89 +
90 +type stressRunResult struct {
91 + streamCount int
92 + aliveAfterReloads int
93 + aliveBeforeDelayExpiry int // only meaningful for close_delay mode
94 + beforeReload heapSnapshot
95 + midReload heapSnapshot // after all reloads, before delay expiry clean-up
96 + afterReload heapSnapshot // after all streams have been fully cleaned up
97 +}
98 +
99 +type heapSnapshot struct {
100 + HeapInuse uint64
101 + HeapObjects uint64
102 + handlerFrames int
103 + profileBytes int
104 +}
105 +
106 +// runReloadStress opens streamCount upgraded streams, then performs reloadCount
107 +// config reloads spread over time. An echo check is performed every 6 reloads so
108 +// stream health is exercised at each reload boundary rather than only at the end.
109 +// closeDelay mirrors the stream_close_delay config option; pass 0 to disable.
110 +func runReloadStress(t *testing.T, tester *caddytest.Tester, backendAddr, mode string, detach bool, closeDelay time.Duration) stressRunResult {
111 + t.Helper()
112 +
113 + const echoEvery = 6 // perform an echo check every N reloads
114 +
115 + streamCount := envIntOrDefault(t, "CADDY_STRESS_STREAM_COUNT", defaultStressStreamCount)
116 + reloadCount := envIntOrDefault(t, "CADDY_STRESS_RELOAD_COUNT", defaultStressReloadCount)
117 +
118 + tester.InitServer(reloadStressConfig(backendAddr, detach, closeDelay, 0), "caddyfile")
119 +
120 + clients := make([]*upgradedStreamClient, 0, streamCount)
121 + for i := 0; i < streamCount; i++ {
122 + client := newUpgradedStreamClient(t)
123 + clients = append(clients, client)
124 + if err := client.echo(fmt.Sprintf("%s-warmup-%02d\n", mode, i)); err != nil {
125 + closeClients(clients)
126 + t.Fatalf("warmup echo failed in %s mode: %v", mode, err)
127 + }
128 + }
129 + defer closeClients(clients)
130 +
131 + before := captureHeapSnapshot(t)
132 +
133 + // Reloads are spread across time; between batches of echoEvery reloads we
134 + // pause briefly and measure stream health so the snapshot reflects real-world
135 + // reload cadence rather than a tight loop.
136 + for i := 1; i <= reloadCount; i++ {
137 + loadCaddyfileConfig(t, reloadStressConfig(backendAddr, detach, closeDelay, i))
138 +
139 + // Small pause after each reload to let connection teardown propagate.
140 + time.Sleep(50 * time.Millisecond)
141 +
142 + if i%echoEvery == 0 {
143 + alive := countAliveStreams(clients)
144 + t.Logf("%s mode: %d/%d streams alive after reload %d", mode, alive, streamCount, i)
145 +
146 + // In detached mode, every stream must survive every reload (upstream unchanged).
147 + if detach {
148 + for j, client := range clients {
149 + if err := client.echo(fmt.Sprintf("%s-mid-%02d-%02d\n", mode, i, j)); err != nil {
150 + t.Fatalf("detached mode stream %d died at reload %d: %v", j, i, err)
151 + }
152 + }
153 + }
154 + }
155 + }
156 +
157 + // mid snapshot: after all reloads but before any close_delay timer has fired
158 + // (the delay is long enough to still be running at this point).
159 + mid := captureHeapSnapshot(t)
160 +
161 + // For legacy mode: the reloads close streams immediately; wait for that to complete.
162 + // For close_delay mode: streams are still alive here; wait for the delay to fire.
163 + // For detached mode: streams survive indefinitely; no wait needed.
164 + var aliveBeforeDelayExpiry int
165 + aliveAfterReloads := countAliveStreams(clients)
166 + switch {
167 + case detach:
168 + // nothing to wait for
169 + case closeDelay > 0:
170 + // streams should still be alive at this point (delay hasn't expired)
171 + aliveBeforeDelayExpiry = aliveAfterReloads
172 + t.Logf("%s mode: %d/%d streams alive before close_delay expires; waiting %v for cleanup",
173 + mode, aliveBeforeDelayExpiry, streamCount, closeDelay)
174 + time.Sleep(closeDelay + 200*time.Millisecond)
175 + aliveAfterReloads = countAliveStreams(clients)
176 + default:
177 + deadline := time.Now().Add(2 * time.Second)
178 + for aliveAfterReloads > 0 && time.Now().Before(deadline) {
179 + time.Sleep(50 * time.Millisecond)
180 + aliveAfterReloads = countAliveStreams(clients)
181 + }
182 + }
183 +
184 + after := captureHeapSnapshot(t)
185 + t.Logf("%s mode heap profile size: before=%dB mid=%dB after=%dB objects(before=%d mid=%d after=%d)",
186 + mode,
187 + before.profileBytes, mid.profileBytes, after.profileBytes,
188 + before.HeapObjects, mid.HeapObjects, after.HeapObjects,
189 + )
190 +
191 + return stressRunResult{
192 + streamCount: streamCount,
193 + aliveAfterReloads: aliveAfterReloads,
194 + aliveBeforeDelayExpiry: aliveBeforeDelayExpiry,
195 + beforeReload: before,
196 + midReload: mid,
197 + afterReload: after,
198 + }
199 +}
200 +
201 +func envIntOrDefault(t *testing.T, key string, def int) int {
202 + t.Helper()
203 + raw := strings.TrimSpace(os.Getenv(key))
204 + if raw == "" {
205 + return def
206 + }
207 + v, err := strconv.Atoi(raw)
208 + if err != nil || v <= 0 {
209 + t.Fatalf("invalid %s=%q: must be a positive integer", key, raw)
210 + }
211 + return v
212 +}
213 +
214 +func stressCloseDelay(t *testing.T) time.Duration {
215 + t.Helper()
216 +
217 + const key = "CADDY_STRESS_CLOSE_DELAY"
218 + raw := strings.TrimSpace(os.Getenv(key))
219 + if raw == "" {
220 + return defaultStressCloseDelay
221 + }
222 + v, err := time.ParseDuration(raw)
223 + if err != nil || v <= 0 {
224 + t.Fatalf("invalid %s=%q: must be a positive duration", key, raw)
225 + }
226 + return v
227 +}
228 +
229 +func loadCaddyfileConfig(t *testing.T, rawConfig string) {
230 + t.Helper()
231 +
232 + client := &http.Client{Timeout: 30 * time.Second}
233 + req, err := http.NewRequest(http.MethodPost, "http://localhost:2999/load", strings.NewReader(rawConfig))
234 + if err != nil {
235 + t.Fatalf("creating load request: %v", err)
236 + }
237 + req.Header.Set("Content-Type", "text/caddyfile")
238 +
239 + resp, err := client.Do(req)
240 + if err != nil {
241 + t.Fatalf("loading config: %v", err)
242 + }
243 + defer resp.Body.Close()
244 +
245 + body, err := io.ReadAll(resp.Body)
246 + if err != nil {
247 + t.Fatalf("reading load response: %v", err)
248 + }
249 + if resp.StatusCode != http.StatusOK {
250 + t.Fatalf("loading config failed: status=%d body=%s", resp.StatusCode, body)
251 + }
252 +}
253 +
254 +func reloadStressConfig(backendAddr string, detach bool, closeDelay time.Duration, revision int) string {
255 + var directives string
256 + if detach {
257 + directives += "\n\t\tstream_detached"
258 + }
259 + if closeDelay > 0 {
260 + directives += fmt.Sprintf("\n\t\tstream_close_delay %s", closeDelay)
261 + }
262 +
263 + return fmt.Sprintf(`
264 +{
265 + admin localhost:2999
266 + http_port 9080
267 + https_port 9443
268 + grace_period 1ns
269 + skip_install_trust
270 +}
271 +
272 +localhost:9080 {
273 + reverse_proxy %s {
274 + header_up X-Reload-Revision %d%s
275 + }
276 +}
277 +`, backendAddr, revision, directives)
278 +}
279 +
280 +func captureHeapSnapshot(t *testing.T) heapSnapshot {
281 + t.Helper()
282 +
283 + runtime.GC()
284 + debug.FreeOSMemory()
285 +
286 + var mem runtime.MemStats
287 + runtime.ReadMemStats(&mem)
288 +
289 + var buf bytes.Buffer
290 + if err := pprof.Lookup("heap").WriteTo(&buf, 1); err != nil {
291 + t.Fatalf("capturing heap profile: %v", err)
292 + }
293 + profile := buf.String()
294 +
295 + return heapSnapshot{
296 + HeapInuse: mem.HeapInuse,
297 + HeapObjects: mem.HeapObjects,
298 + handlerFrames: strings.Count(profile, "modules/caddyhttp/reverseproxy.(*Handler)"),
299 + profileBytes: buf.Len(),
300 + }
301 +}
302 +
303 +func countAliveStreams(clients []*upgradedStreamClient) int {
304 + alive := 0
305 + for index, client := range clients {
306 + if err := client.echo(fmt.Sprintf("alive-check-%02d\n", index)); err == nil {
307 + alive++
308 + }
309 + }
310 + return alive
311 +}
312 +
313 +func closeClients(clients []*upgradedStreamClient) {
314 + for _, client := range clients {
315 + if client != nil {
316 + _ = client.Close()
317 + }
318 + }
319 +}
320 +
321 +func formatBytes(value uint64) string {
322 + const unit = 1024
323 + if value < unit {
324 + return fmt.Sprintf("%d B", value)
325 + }
326 + div, exp := uint64(unit), 0
327 + for n := value / unit; n >= unit; n /= unit {
328 + div *= unit
329 + exp++
330 + }
331 + return fmt.Sprintf("%.1f %ciB", float64(value)/float64(div), "KMGTPE"[exp])
332 +}
333 +
334 +func formatBytesDiff(before, after uint64) string {
335 + if after >= before {
336 + return "+" + formatBytes(after-before)
337 + }
338 + return "-" + formatBytes(before-after)
339 +}
340 +
341 +type upgradedStreamClient struct {
342 + conn net.Conn
343 + reader *bufio.Reader
344 + mu sync.Mutex
345 +}
346 +
347 +func newUpgradedStreamClient(t *testing.T) *upgradedStreamClient {
348 + t.Helper()
349 +
350 + conn, err := net.DialTimeout("tcp", "127.0.0.1:9080", 5*time.Second)
351 + if err != nil {
352 + t.Fatalf("dialing caddy: %v", err)
353 + }
354 +
355 + request := strings.Join([]string{
356 + "GET /upgrade HTTP/1.1",
357 + "Host: localhost:9080",
358 + "Connection: Upgrade",
359 + "Upgrade: stress-stream",
360 + "",
361 + "",
362 + }, "\r\n")
363 + if _, err := io.WriteString(conn, request); err != nil {
364 + _ = conn.Close()
365 + t.Fatalf("writing upgrade request: %v", err)
366 + }
367 +
368 + reader := bufio.NewReader(conn)
369 + tproto := textproto.NewReader(reader)
370 + statusLine, err := tproto.ReadLine()
371 + if err != nil {
372 + _ = conn.Close()
373 + t.Fatalf("reading upgrade status line: %v", err)
374 + }
375 + if !strings.Contains(statusLine, "101") {
376 + _ = conn.Close()
377 + t.Fatalf("unexpected upgrade status: %s", statusLine)
378 + }
379 +
380 + headers, err := tproto.ReadMIMEHeader()
381 + if err != nil {
382 + _ = conn.Close()
383 + t.Fatalf("reading upgrade headers: %v", err)
384 + }
385 + if !strings.EqualFold(headers.Get("Connection"), "Upgrade") {
386 + _ = conn.Close()
387 + t.Fatalf("unexpected upgrade response headers: %v", headers)
388 + }
389 +
390 + return &upgradedStreamClient{conn: conn, reader: reader}
391 +}
392 +
393 +func (c *upgradedStreamClient) echo(payload string) error {
394 + c.mu.Lock()
395 + defer c.mu.Unlock()
396 +
397 + deadline := time.Now().Add(1 * time.Second)
398 + if err := c.conn.SetWriteDeadline(deadline); err != nil {
399 + return err
400 + }
401 + if _, err := io.WriteString(c.conn, payload); err != nil {
402 + return err
403 + }
404 + if err := c.conn.SetReadDeadline(deadline); err != nil {
405 + return err
406 + }
407 +
408 + buf := make([]byte, len(payload))
409 + if _, err := io.ReadFull(c.reader, buf); err != nil {
410 + return err
411 + }
412 + if string(buf) != payload {
413 + return fmt.Errorf("unexpected echoed payload: got %q want %q", string(buf), payload)
414 + }
415 + return nil
416 +}
417 +
418 +func (c *upgradedStreamClient) Close() error {
419 + c.mu.Lock()
420 + defer c.mu.Unlock()
421 + return c.conn.Close()
422 +}
423 +
424 +type upgradeEchoBackend struct {
425 + addr string
426 + ln net.Listener
427 + mu sync.Mutex
428 + conns map[net.Conn]struct{}
429 + server *http.Server
430 +}
431 +
432 +func newUpgradeEchoBackend(t *testing.T) *upgradeEchoBackend {
433 + t.Helper()
434 +
435 + backend := &upgradeEchoBackend{conns: make(map[net.Conn]struct{})}
436 + backend.server = &http.Server{
437 + Handler: http.HandlerFunc(backend.serveHTTP),
438 + }
439 +
440 + ln, err := net.Listen("tcp", "127.0.0.1:0")
441 + if err != nil {
442 + t.Fatalf("listening for backend: %v", err)
443 + }
444 + backend.ln = ln
445 + backend.addr = ln.Addr().String()
446 +
447 + go func() {
448 + _ = backend.server.Serve(ln)
449 + }()
450 +
451 + return backend
452 +}
453 +
454 +func (b *upgradeEchoBackend) serveHTTP(w http.ResponseWriter, r *http.Request) {
455 + if !strings.EqualFold(r.Header.Get("Connection"), "Upgrade") || !strings.EqualFold(r.Header.Get("Upgrade"), "stress-stream") {
456 + http.Error(w, "upgrade required", http.StatusUpgradeRequired)
457 + return
458 + }
459 +
460 + hijacker, ok := w.(http.Hijacker)
461 + if !ok {
462 + http.Error(w, "hijacking not supported", http.StatusInternalServerError)
463 + return
464 + }
465 +
466 + conn, rw, err := hijacker.Hijack()
467 + if err != nil {
468 + return
469 + }
470 +
471 + b.trackConn(conn)
472 + _, _ = rw.WriteString("HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: stress-stream\r\n\r\n")
473 + _ = rw.Flush()
474 +
475 + go func() {
476 + defer b.untrackConn(conn)
477 + defer conn.Close()
478 + _, _ = io.Copy(conn, conn)
479 + }()
480 +}
481 +
482 +func (b *upgradeEchoBackend) trackConn(conn net.Conn) {
483 + b.mu.Lock()
484 + b.conns[conn] = struct{}{}
485 + b.mu.Unlock()
486 +}
487 +
488 +func (b *upgradeEchoBackend) untrackConn(conn net.Conn) {
489 + b.mu.Lock()
490 + delete(b.conns, conn)
491 + b.mu.Unlock()
492 +}
493 +
494 +func (b *upgradeEchoBackend) Close() {
495 + _ = b.server.Close()
496 + _ = b.ln.Close()
497 +
498 + b.mu.Lock()
499 + defer b.mu.Unlock()
500 + for conn := range b.conns {
501 + _ = conn.Close()
502 + }
503 + clear(b.conns)
504 +}
modules/caddyhttp/responsewriter.go
+62 -4
@@ -21,6 +21,8 @@ import (
21 "io"
22 "net"
23 "net/http"
24 +
25 + "github.com/caddyserver/caddy/v2"
26 )
27
28 // ResponseWriterWrapper wraps an underlying ResponseWriter and
@@ -70,6 +72,8 @@ type responseRecorder struct {
72 size int
73 wroteHeader bool
74 stream bool
75 + hijacked bool
76 + detached bool
77
78 readSize *int
79 }
@@ -144,7 +148,8 @@ func NewResponseRecorder(w http.ResponseWriter, buf *bytes.Buffer, shouldBuffer
148
149 // WriteHeader writes the headers with statusCode to the wrapped
150 // ResponseWriter unless the response is to be buffered instead.
147 -// 1xx responses are never buffered.
151 +// 1xx responses are never buffered, except 101 which is treated
152 +// as a final upgrade response.
153 func (rr *responseRecorder) WriteHeader(statusCode int) {
154 if rr.wroteHeader {
155 return
@@ -161,12 +166,12 @@ func (rr *responseRecorder) WriteHeader(statusCode int) {
166 rr.stream = !rr.shouldBuffer(rr.statusCode, rr.ResponseWriterWrapper.Header())
167 }
168
164 - // 1xx responses aren't final; just informational
165 - if statusCode < 100 || statusCode > 199 {
169 + // 1xx responses except 101 aren't final; just informational
170 + if statusCode < 100 || statusCode > 199 || statusCode == http.StatusSwitchingProtocols {
171 rr.wroteHeader = true
172 }
173
169 - // if informational or not buffered, immediately write header
174 + // if 1xx or not buffered, immediately write header
175 if rr.stream || (100 <= statusCode && statusCode <= 199) {
176 rr.ResponseWriterWrapper.WriteHeader(statusCode)
177 }
@@ -222,7 +227,18 @@ func (rr *responseRecorder) Buffered() bool {
227 return !rr.stream
228 }
229
230 +func (rr *responseRecorder) DetachAfterHijack(detached bool) bool {
231 + if rr.hijacked {
232 + return false
233 + }
234 + rr.detached = detached
235 + return true
236 +}
237 +
238 func (rr *responseRecorder) WriteResponse() error {
239 + if rr.hijacked {
240 + return nil
241 + }
242 if rr.statusCode == 0 {
243 // could happen if no handlers actually wrote anything,
244 // and this prevents a panic; status must be > 0
@@ -253,11 +269,25 @@ func (rr *responseRecorder) setReadSize(size *int) {
269 }
270
271 func (rr *responseRecorder) Hijack() (net.Conn, *bufio.ReadWriter, error) {
272 + if !rr.wroteHeader {
273 + // hijacking without writing status code first works as long as
274 + // subsequent writes follows http1.1 wire format, but it will
275 + // show up with a status code of 0 in the access log and bytes
276 + // written will include response headers. Response headers won't
277 + // be present in the log if not set on the response writer.
278 + caddy.Log().Warn("hijacking without writing status code first")
279 + }
280 //nolint:bodyclose
281 conn, brw, err := http.NewResponseController(rr.ResponseWriterWrapper).Hijack()
282 if err != nil {
283 return nil, nil, err
284 }
285 + rr.hijacked = true
286 + rr.stream = true
287 + rr.wroteHeader = true
288 + if rr.detached {
289 + return conn, brw, nil
290 + }
291 // Per http documentation, returned bufio.Writer is empty, but bufio.Read maybe not
292 conn = &hijackedConn{conn, rr}
293 brw.Writer.Reset(conn)
@@ -311,6 +341,33 @@ func (hc *hijackedConn) ReadFrom(r io.Reader) (int64, error) {
341 return n, err
342 }
343
344 +// DetachResponseWriterAfterHijack configures w and its wrapped response
345 +// writers to detach when hijacked. It returns true if at least one writer
346 +// supports detachment and all such writers accept it. When detached,
347 +// bytes read or written stats will not be recorded for the hijacked
348 +// connection, and it's safe to use the connection after HTTP middleware returns.
349 +func DetachResponseWriterAfterHijack(w http.ResponseWriter, detached bool) bool {
350 + found := false
351 + for w != nil {
352 + if detacher, ok := w.(interface{ DetachAfterHijack(bool) bool }); ok {
353 + if !detacher.DetachAfterHijack(detached) {
354 + return false
355 + }
356 + found = true
357 + }
358 + unwrapper, ok := w.(interface{ Unwrap() http.ResponseWriter })
359 + if !ok {
360 + return found
361 + }
362 + next := unwrapper.Unwrap()
363 + if next == w {
364 + return found
365 + }
366 + w = next
367 + }
368 + return found
369 +}
370 +
371 // ResponseRecorder is a http.ResponseWriter that records
372 // responses instead of writing them to the client. See
373 // docs for NewResponseRecorder for proper usage.
@@ -319,6 +376,7 @@ type ResponseRecorder interface {
376 Status() int
377 Buffer() *bytes.Buffer
378 Buffered() bool
379 + DetachAfterHijack(bool) bool
380 Size() int
381 WriteResponse() error
382 }
modules/caddyhttp/responsewriter_test.go
+138
@@ -1,11 +1,14 @@
1 package caddyhttp
2
3 import (
4 + "bufio"
5 "bytes"
6 "io"
7 + "net"
8 "net/http"
9 "strings"
10 "testing"
11 + "time"
12 )
13
14 type responseWriterSpy interface {
@@ -44,6 +47,50 @@ func (rf *readFromRespWriter) ReadFrom(r io.Reader) (int64, error) {
47
48 func (rf *readFromRespWriter) CalledReadFrom() bool { return rf.called }
49
50 +type hijackRespWriter struct {
51 + baseRespWriter
52 + header http.Header
53 + status int
54 + conn net.Conn
55 +}
56 +
57 +func newHijackRespWriter() *hijackRespWriter {
58 + return &hijackRespWriter{
59 + header: make(http.Header),
60 + conn: stubConn{},
61 + }
62 +}
63 +
64 +func (hrw *hijackRespWriter) Header() http.Header {
65 + return hrw.header
66 +}
67 +
68 +func (hrw *hijackRespWriter) WriteHeader(statusCode int) {
69 + hrw.status = statusCode
70 +}
71 +
72 +func (hrw *hijackRespWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
73 + br := bufio.NewReader(hrw.conn)
74 + bw := bufio.NewWriter(hrw.conn)
75 + return hrw.conn, bufio.NewReadWriter(br, bw), nil
76 +}
77 +
78 +type stubConn struct{}
79 +
80 +func (stubConn) Read(_ []byte) (int, error) { return 0, io.EOF }
81 +func (stubConn) Write(p []byte) (int, error) { return len(p), nil }
82 +func (stubConn) Close() error { return nil }
83 +func (stubConn) LocalAddr() net.Addr { return stubAddr("local") }
84 +func (stubConn) RemoteAddr() net.Addr { return stubAddr("remote") }
85 +func (stubConn) SetDeadline(time.Time) error { return nil }
86 +func (stubConn) SetReadDeadline(time.Time) error { return nil }
87 +func (stubConn) SetWriteDeadline(time.Time) error { return nil }
88 +
89 +type stubAddr string
90 +
91 +func (a stubAddr) Network() string { return "tcp" }
92 +func (a stubAddr) String() string { return string(a) }
93 +
94 func TestResponseWriterWrapperReadFrom(t *testing.T) {
95 tests := map[string]struct {
96 responseWriter responseWriterSpy
@@ -169,3 +216,94 @@ func TestResponseRecorderReadFrom(t *testing.T) {
216 })
217 }
218 }
219 +
220 +func TestResponseRecorderSwitchingProtocolsIsHijackAware(t *testing.T) {
221 + w := newHijackRespWriter()
222 + var buf bytes.Buffer
223 +
224 + rr := NewResponseRecorder(w, &buf, func(status int, header http.Header) bool {
225 + return true
226 + })
227 + rr.WriteHeader(http.StatusSwitchingProtocols)
228 +
229 + if rr.Status() != http.StatusSwitchingProtocols {
230 + t.Fatalf("status = %d, want %d", rr.Status(), http.StatusSwitchingProtocols)
231 + }
232 + if w.status != http.StatusSwitchingProtocols {
233 + t.Fatalf("underlying status = %d, want %d", w.status, http.StatusSwitchingProtocols)
234 + }
235 +
236 + hj, ok := rr.(http.Hijacker)
237 + if !ok {
238 + t.Fatal("response recorder does not implement http.Hijacker")
239 + }
240 + conn, _, err := hj.Hijack()
241 + if err != nil {
242 + t.Fatalf("Hijack() error = %v", err)
243 + }
244 + defer conn.Close()
245 +
246 + if rr.Buffered() {
247 + t.Fatal("hijacked response should not remain buffered")
248 + }
249 + if rr.DetachAfterHijack(true) {
250 + t.Fatal("response recorder should report hijacked state by returning false")
251 + }
252 + if DetachResponseWriterAfterHijack(rr, true) {
253 + t.Fatal("DetachResponseWriterAfterHijack() should report false after hijack")
254 + }
255 + if err := rr.WriteResponse(); err != nil {
256 + t.Fatalf("WriteResponse() after hijack returned error: %v", err)
257 + }
258 + if rr.Size() != 0 {
259 + t.Fatalf("size = %d, want 0 after hijack handshake", rr.Size())
260 + }
261 + if got := w.Written(); got != "" {
262 + t.Fatalf("unexpected buffered body write after hijack: %q", got)
263 + }
264 +}
265 +
266 +func TestResponseRecorderNestedDetach(t *testing.T) {
267 + for _, detached := range []bool{false, true} {
268 + name := "attached"
269 + if detached {
270 + name = "detached"
271 + }
272 + t.Run(name, func(t *testing.T) {
273 + inner := NewResponseRecorder(newHijackRespWriter(), nil, nil)
274 + // Include an intervening middleware wrapper in the recorder chain.
275 + outer := NewResponseRecorder(&ResponseWriterWrapper{ResponseWriter: inner}, nil, nil)
276 + outer.WriteHeader(http.StatusSwitchingProtocols)
277 + if !DetachResponseWriterAfterHijack(outer, detached) {
278 + t.Fatal("detach configuration failed")
279 + }
280 + conn, brw, err := http.NewResponseController(outer).Hijack()
281 + if err != nil {
282 + t.Fatal(err)
283 + }
284 + defer conn.Close()
285 + const payload = "stream payload"
286 + if _, err := conn.Write([]byte(payload)); err != nil {
287 + t.Fatal(err)
288 + }
289 + if _, err := brw.WriteString(payload); err != nil {
290 + t.Fatal(err)
291 + }
292 + if err := brw.Flush(); err != nil {
293 + t.Fatal(err)
294 + }
295 + want := 2 * len(payload)
296 + if detached {
297 + want = 0
298 + }
299 + for name, rr := range map[string]ResponseRecorder{"inner": inner, "outer": outer} {
300 + if rr.Size() != want {
301 + t.Errorf("%s recorder size = %d, want %d", name, rr.Size(), want)
302 + }
303 + }
304 + if DetachResponseWriterAfterHijack(outer, true) {
305 + t.Error("detach should fail after hijack")
306 + }
307 + })
308 + }
309 +}
modules/caddyhttp/reverseproxy/caddyfile.go
+49
@@ -99,6 +99,12 @@ func parseCaddyfile(h httpcaddyfile.Helper) (caddyhttp.MiddlewareHandler, error)
99 // stream_buffer_size <size>
100 // stream_timeout <duration>
101 // stream_close_delay <duration>
102 +// stream_detached
103 +// stream_logs {
104 +// level <debug|info|warn|error>
105 +// logger_name <name|access>
106 +// skip_handshake
107 +// }
108 // verbose_logs
109 //
110 // # reporting
@@ -715,6 +721,49 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
721 h.StreamCloseDelay = caddy.Duration(dur)
722 }
723
724 + case "stream_detached":
725 + if d.NextArg() {
726 + return d.ArgErr()
727 + }
728 + h.StreamDetached = true
729 +
730 + case "stream_logs":
731 + if d.NextArg() {
732 + return d.ArgErr()
733 + }
734 + if h.StreamLogs == nil {
735 + h.StreamLogs = new(StreamLogs)
736 + }
737 +
738 + nesting := d.Nesting()
739 + for d.NextBlock(nesting) {
740 + switch d.Val() {
741 + case "level":
742 + if !d.NextArg() {
743 + return d.ArgErr()
744 + }
745 + h.StreamLogs.Level = d.Val()
746 + if d.NextArg() {
747 + return d.ArgErr()
748 + }
749 + case "logger_name":
750 + if !d.NextArg() {
751 + return d.ArgErr()
752 + }
753 + h.StreamLogs.LoggerName = d.Val()
754 + if d.NextArg() {
755 + return d.ArgErr()
756 + }
757 + case "skip_handshake":
758 + if d.NextArg() {
759 + return d.ArgErr()
760 + }
761 + h.StreamLogs.SkipHandshake = true
762 + default:
763 + return d.Errf("unrecognized stream_logs option: %s", d.Val())
764 + }
765 + }
766 +
767 case "trusted_proxies":
768 for d.NextArg() {
769 if d.Val() == "private_ranges" {
modules/caddyhttp/reverseproxy/copyresponse.go
+1 -1
@@ -80,7 +80,7 @@ func (h CopyResponseHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request
80 hrc.isFinalized = true
81
82 // write the response
83 - return hrc.handler.finalizeResponse(rw, req, hrc.response, repl, hrc.start, hrc.logger)
83 + return hrc.handler.finalizeResponse(rw, req, hrc.response, repl, hrc.start, hrc.logger, hrc.upstreamAddr)
84 }
85
86 // CopyResponseHeadersHandler is a special HTTP handler which may
modules/caddyhttp/reverseproxy/extended_connect_test.go new
+146
@@ -0,0 +1,146 @@
1 +package reverseproxy
2 +
3 +import (
4 + "io"
5 + "net/http"
6 + "net/http/httptest"
7 + "strings"
8 + "sync"
9 + "testing"
10 +
11 + "go.uber.org/zap"
12 +
13 + "github.com/caddyserver/caddy/v2/modules/caddyhttp"
14 +)
15 +
16 +type extendedConnectCapture struct {
17 + method string
18 + headers http.Header
19 + body []byte
20 + extendedBodyPresent bool
21 + extendedConnectBody []byte
22 +}
23 +
24 +type extendedConnectCaptureTransport struct {
25 + mu sync.Mutex
26 + capture extendedConnectCapture
27 +}
28 +
29 +func (tr *extendedConnectCaptureTransport) RoundTrip(req *http.Request) (*http.Response, error) {
30 + body, err := io.ReadAll(req.Body)
31 + if err != nil {
32 + return nil, err
33 + }
34 +
35 + c := extendedConnectCapture{
36 + method: req.Method,
37 + headers: req.Header.Clone(),
38 + body: body,
39 + }
40 + if rc, ok := caddyhttp.GetVar(req.Context(), "extended_connect_websocket_body").(io.ReadCloser); ok {
41 + c.extendedBodyPresent = true
42 + c.extendedConnectBody, err = io.ReadAll(rc)
43 + if err != nil {
44 + return nil, err
45 + }
46 + _ = rc.Close()
47 + }
48 +
49 + tr.mu.Lock()
50 + tr.capture = c
51 + tr.mu.Unlock()
52 +
53 + return &http.Response{
54 + StatusCode: http.StatusOK,
55 + Header: make(http.Header),
56 + Body: io.NopCloser(strings.NewReader("ok")),
57 + Request: req,
58 + }, nil
59 +}
60 +
61 +func (tr *extendedConnectCaptureTransport) Snapshot() extendedConnectCapture {
62 + tr.mu.Lock()
63 + defer tr.mu.Unlock()
64 + return tr.capture
65 +}
66 +
67 +func TestServeHTTPRewritesExtendedConnectWebsocketRequest(t *testing.T) {
68 + tests := []struct {
69 + name string
70 + protoMajor int
71 + proto string
72 + headers map[string]string
73 + }{
74 + {
75 + name: "h2 extended connect",
76 + protoMajor: 2,
77 + proto: "HTTP/2.0",
78 + headers: map[string]string{
79 + ":protocol": "websocket",
80 + },
81 + },
82 + {
83 + name: "h3 extended connect",
84 + protoMajor: 3,
85 + proto: "websocket",
86 + headers: map[string]string{},
87 + },
88 + }
89 +
90 + for _, tc := range tests {
91 + t.Run(tc.name, func(t *testing.T) {
92 + const payload = "extended-connect-body"
93 +
94 + transport := new(extendedConnectCaptureTransport)
95 + h := &Handler{
96 + logger: zap.NewNop(),
97 + Transport: transport,
98 + Upstreams: UpstreamPool{
99 + &Upstream{Host: new(Host), Dial: "127.0.0.1:8443"},
100 + },
101 + LoadBalancing: &LoadBalancing{
102 + SelectionPolicy: &RoundRobinSelection{},
103 + },
104 + }
105 +
106 + req := httptest.NewRequest(http.MethodConnect, "http://example.test/upgrade", strings.NewReader(payload))
107 + req.ProtoMajor = tc.protoMajor
108 + req.Proto = tc.proto
109 + for key, value := range tc.headers {
110 + req.Header.Set(key, value)
111 + }
112 + req = prepareTestRequest(req)
113 +
114 + rr := httptest.NewRecorder()
115 + err := h.ServeHTTP(rr, req, caddyhttp.HandlerFunc(func(w http.ResponseWriter, r *http.Request) error {
116 + return nil
117 + }))
118 + if err != nil {
119 + t.Fatalf("ServeHTTP() error = %v", err)
120 + }
121 +
122 + captured := transport.Snapshot()
123 + if captured.method != http.MethodGet {
124 + t.Fatalf("upstream method = %s, want %s", captured.method, http.MethodGet)
125 + }
126 + if got := captured.headers.Get("Upgrade"); !strings.EqualFold(got, "websocket") {
127 + t.Fatalf("Upgrade header = %q, want websocket", got)
128 + }
129 + if got := captured.headers.Get("Connection"); !strings.EqualFold(got, "Upgrade") {
130 + t.Fatalf("Connection header = %q, want Upgrade", got)
131 + }
132 + if got := captured.headers.Get(":protocol"); got != "" {
133 + t.Fatalf(":protocol header should be removed, got %q", got)
134 + }
135 + if len(captured.body) != 0 {
136 + t.Fatalf("upstream request body length = %d, want 0", len(captured.body))
137 + }
138 + if !captured.extendedBodyPresent {
139 + t.Fatal("extended_connect_websocket_body variable missing from request context")
140 + }
141 + if string(captured.extendedConnectBody) != payload {
142 + t.Fatalf("extended_connect_websocket_body = %q, want %q", string(captured.extendedConnectBody), payload)
143 + }
144 + })
145 + }
146 +}
modules/caddyhttp/reverseproxy/hopheaders_test.go
+4 -4
@@ -45,7 +45,7 @@ func TestFinalizeResponse_101_StripsHopByHopHeaders(t *testing.T) {
45 repl := caddy.NewReplacer()
46 rw := httptest.NewRecorder()
47
48 - err := h.finalizeResponse(rw, req, res, repl, fakeStart, caddy.Log())
48 + err := h.finalizeResponse(rw, req, res, repl, fakeStart, caddy.Log(), "")
49 if err != nil {
50 t.Logf("finalizeResponse returned error (expected, no real conn): %v", err)
51 }
@@ -93,7 +93,7 @@ func TestFinalizeResponse_101_StripsConnectionNamedHeaders(t *testing.T) {
93 repl := caddy.NewReplacer()
94 rw := httptest.NewRecorder()
95
96 - err := h.finalizeResponse(rw, req, res, repl, fakeStart, caddy.Log())
96 + err := h.finalizeResponse(rw, req, res, repl, fakeStart, caddy.Log(), "")
97 if err != nil {
98 t.Logf("finalizeResponse returned error (no real connection): %v", err)
99 }
@@ -133,7 +133,7 @@ func TestFinalizeResponse_200_StillStripsHopByHop(t *testing.T) {
133 repl := caddy.NewReplacer()
134 rw := httptest.NewRecorder()
135
136 - err := h.finalizeResponse(rw, req, res, repl, fakeStart, caddy.Log())
136 + err := h.finalizeResponse(rw, req, res, repl, fakeStart, caddy.Log(), "")
137 if err != nil {
138 t.Fatalf("finalizeResponse returned error: %v", err)
139 }
@@ -170,7 +170,7 @@ func TestFinalizeResponse_101_NoUpgradeRequest(t *testing.T) {
170 repl := caddy.NewReplacer()
171 rw := httptest.NewRecorder()
172
173 - err := h.finalizeResponse(rw, req, res, repl, fakeStart, caddy.Log())
173 + err := h.finalizeResponse(rw, req, res, repl, fakeStart, caddy.Log(), "")
174 if err != nil {
175 t.Fatalf("finalizeResponse returned error: %v", err)
176 }
modules/caddyhttp/reverseproxy/metrics.go
+79
@@ -16,6 +16,10 @@ import (
16 var reverseProxyMetrics = struct {
17 once sync.Once
18 upstreamsHealthy *prometheus.GaugeVec
19 + streamsActive *prometheus.GaugeVec
20 + streamsTotal *prometheus.CounterVec
21 + streamDuration *prometheus.HistogramVec
22 + streamBytes *prometheus.CounterVec
23 logger *zap.Logger
24 }{}
25
@@ -23,6 +27,8 @@ func initReverseProxyMetrics(handler *Handler, registry *prometheus.Registry) {
27 const ns, sub = "caddy", "reverse_proxy"
28
29 upstreamsLabels := []string{"upstream"}
30 + streamResultLabels := []string{"upstream", "result"}
31 + streamBytesLabels := []string{"upstream", "direction"}
32 reverseProxyMetrics.once.Do(func() {
33 reverseProxyMetrics.upstreamsHealthy = prometheus.NewGaugeVec(prometheus.GaugeOpts{
34 Namespace: ns,
@@ -30,6 +36,31 @@ func initReverseProxyMetrics(handler *Handler, registry *prometheus.Registry) {
36 Name: "upstreams_healthy",
37 Help: "Health status of reverse proxy upstreams.",
38 }, upstreamsLabels)
39 + reverseProxyMetrics.streamsActive = prometheus.NewGaugeVec(prometheus.GaugeOpts{
40 + Namespace: ns,
41 + Subsystem: sub,
42 + Name: "streams_active",
43 + Help: "Number of currently active upgraded reverse proxy streams.",
44 + }, upstreamsLabels)
45 + reverseProxyMetrics.streamsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
46 + Namespace: ns,
47 + Subsystem: sub,
48 + Name: "streams_total",
49 + Help: "Total number of upgraded reverse proxy streams by close result.",
50 + }, streamResultLabels)
51 + reverseProxyMetrics.streamDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
52 + Namespace: ns,
53 + Subsystem: sub,
54 + Name: "stream_duration_seconds",
55 + Help: "Duration of upgraded reverse proxy streams by close result.",
56 + Buckets: prometheus.DefBuckets,
57 + }, streamResultLabels)
58 + reverseProxyMetrics.streamBytes = prometheus.NewCounterVec(prometheus.CounterOpts{
59 + Namespace: ns,
60 + Subsystem: sub,
61 + Name: "stream_bytes_total",
62 + Help: "Total bytes proxied across upgraded reverse proxy streams.",
63 + }, streamBytesLabels)
64 })
65
66 // duplicate registration could happen if multiple sites with reverse proxy are configured; so ignore the error because
@@ -42,10 +73,58 @@ func initReverseProxyMetrics(handler *Handler, registry *prometheus.Registry) {
73 }) {
74 panic(err)
75 }
76 + if err := registry.Register(reverseProxyMetrics.streamsActive); err != nil &&
77 + !errors.Is(err, prometheus.AlreadyRegisteredError{
78 + ExistingCollector: reverseProxyMetrics.streamsActive,
79 + NewCollector: reverseProxyMetrics.streamsActive,
80 + }) {
81 + panic(err)
82 + }
83 + if err := registry.Register(reverseProxyMetrics.streamsTotal); err != nil &&
84 + !errors.Is(err, prometheus.AlreadyRegisteredError{
85 + ExistingCollector: reverseProxyMetrics.streamsTotal,
86 + NewCollector: reverseProxyMetrics.streamsTotal,
87 + }) {
88 + panic(err)
89 + }
90 + if err := registry.Register(reverseProxyMetrics.streamDuration); err != nil &&
91 + !errors.Is(err, prometheus.AlreadyRegisteredError{
92 + ExistingCollector: reverseProxyMetrics.streamDuration,
93 + NewCollector: reverseProxyMetrics.streamDuration,
94 + }) {
95 + panic(err)
96 + }
97 + if err := registry.Register(reverseProxyMetrics.streamBytes); err != nil &&
98 + !errors.Is(err, prometheus.AlreadyRegisteredError{
99 + ExistingCollector: reverseProxyMetrics.streamBytes,
100 + NewCollector: reverseProxyMetrics.streamBytes,
101 + }) {
102 + panic(err)
103 + }
104
105 reverseProxyMetrics.logger = handler.logger.Named("reverse_proxy.metrics")
106 }
107
108 +func trackActiveStream(upstream string) func(result string, duration time.Duration, toBackend, fromBackend int64) {
109 + labels := prometheus.Labels{"upstream": upstream}
110 + reverseProxyMetrics.streamsActive.With(labels).Inc()
111 +
112 + var once sync.Once
113 + return func(result string, duration time.Duration, toBackend, fromBackend int64) {
114 + once.Do(func() {
115 + reverseProxyMetrics.streamsActive.With(labels).Dec()
116 + reverseProxyMetrics.streamsTotal.WithLabelValues(upstream, result).Inc()
117 + reverseProxyMetrics.streamDuration.WithLabelValues(upstream, result).Observe(duration.Seconds())
118 + if toBackend > 0 {
119 + reverseProxyMetrics.streamBytes.WithLabelValues(upstream, "to_upstream").Add(float64(toBackend))
120 + }
121 + if fromBackend > 0 {
122 + reverseProxyMetrics.streamBytes.WithLabelValues(upstream, "from_upstream").Add(float64(fromBackend))
123 + }
124 + })
125 + }
126 +}
127 +
128 type metricsUpstreamsHealthyUpdater struct {
129 handler *Handler
130 }
modules/caddyhttp/reverseproxy/metrics_test.go new
+67
@@ -0,0 +1,67 @@
1 +package reverseproxy
2 +
3 +import (
4 + "testing"
5 + "time"
6 +
7 + "github.com/prometheus/client_golang/prometheus"
8 + "github.com/prometheus/client_golang/prometheus/testutil"
9 +)
10 +
11 +func TestTrackActiveStreamRecordsLifecycleAndBytes(t *testing.T) {
12 + const upstream = "127.0.0.1:7443"
13 +
14 + // Use fresh metric vectors for deterministic assertions in this unit test.
15 + reverseProxyMetrics.streamsActive = prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"upstream"})
16 + reverseProxyMetrics.streamsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"upstream", "result"})
17 + reverseProxyMetrics.streamDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{}, []string{"upstream", "result"})
18 + reverseProxyMetrics.streamBytes = prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"upstream", "direction"})
19 +
20 + finish := trackActiveStream(upstream)
21 +
22 + if got := testutil.ToFloat64(reverseProxyMetrics.streamsActive.WithLabelValues(upstream)); got != 1 {
23 + t.Fatalf("active streams = %v, want 1", got)
24 + }
25 +
26 + finish("closed", 150*time.Millisecond, 1234, 4321)
27 +
28 + if got := testutil.ToFloat64(reverseProxyMetrics.streamsActive.WithLabelValues(upstream)); got != 0 {
29 + t.Fatalf("active streams = %v, want 0", got)
30 + }
31 + if got := testutil.ToFloat64(reverseProxyMetrics.streamsTotal.WithLabelValues(upstream, "closed")); got != 1 {
32 + t.Fatalf("streams_total closed = %v, want 1", got)
33 + }
34 + if got := testutil.ToFloat64(reverseProxyMetrics.streamBytes.WithLabelValues(upstream, "to_upstream")); got != 1234 {
35 + t.Fatalf("bytes to_upstream = %v, want 1234", got)
36 + }
37 + if got := testutil.ToFloat64(reverseProxyMetrics.streamBytes.WithLabelValues(upstream, "from_upstream")); got != 4321 {
38 + t.Fatalf("bytes from_upstream = %v, want 4321", got)
39 + }
40 +
41 + // A second finish call should be ignored by the once guard.
42 + finish("error", 1*time.Second, 111, 222)
43 + if got := testutil.ToFloat64(reverseProxyMetrics.streamsTotal.WithLabelValues(upstream, "error")); got != 0 {
44 + t.Fatalf("streams_total error = %v, want 0", got)
45 + }
46 +}
47 +
48 +func TestTrackActiveStreamDoesNotCountZeroBytes(t *testing.T) {
49 + const upstream = "127.0.0.1:9000"
50 +
51 + reverseProxyMetrics.streamsActive = prometheus.NewGaugeVec(prometheus.GaugeOpts{}, []string{"upstream"})
52 + reverseProxyMetrics.streamsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"upstream", "result"})
53 + reverseProxyMetrics.streamDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{}, []string{"upstream", "result"})
54 + reverseProxyMetrics.streamBytes = prometheus.NewCounterVec(prometheus.CounterOpts{}, []string{"upstream", "direction"})
55 +
56 + trackActiveStream(upstream)("timeout", 250*time.Millisecond, 0, 0)
57 +
58 + if got := testutil.ToFloat64(reverseProxyMetrics.streamBytes.WithLabelValues(upstream, "to_upstream")); got != 0 {
59 + t.Fatalf("bytes to_upstream = %v, want 0", got)
60 + }
61 + if got := testutil.ToFloat64(reverseProxyMetrics.streamBytes.WithLabelValues(upstream, "from_upstream")); got != 0 {
62 + t.Fatalf("bytes from_upstream = %v, want 0", got)
63 + }
64 + if got := testutil.ToFloat64(reverseProxyMetrics.streamsTotal.WithLabelValues(upstream, "timeout")); got != 1 {
65 + t.Fatalf("streams_total timeout = %v, want 1", got)
66 + }
67 +}
modules/caddyhttp/reverseproxy/reverseproxy.go
+156 -19
@@ -199,6 +199,22 @@ type Handler struct {
199 // by the previous config closing. Default: no delay.
200 StreamCloseDelay caddy.Duration `json:"stream_close_delay,omitempty"`
201
202 + // If true, upgraded connections such as WebSockets are detached from
203 + // the handler and retained across config reloads when their upstream
204 + // still exists in the new config. Connections using upstreams that are
205 + // removed are closed during cleanup. By default this is false, preserving
206 + // legacy behavior where upgraded connections are closed on reload
207 + // (optionally delayed by stream_close_delay).
208 + // Only http1.1 websocket connections are affected, websockets for h2/h3
209 + // are not affected. If true, bytes transferred for http1.1 in the access
210 + // logs will be zero but those stats can be found in the stream logs for
211 + // http1/2/3 regardless if this is enabled.
212 + StreamDetached bool `json:"stream_detached,omitempty"`
213 +
214 + // Controls logging behavior for upgraded stream lifecycle events.
215 + // If omitted, defaults are used (level=DEBUG, logger_name="http.handlers.reverse_proxy.stream").
216 + StreamLogs *StreamLogs `json:"stream_logs,omitempty"`
217 +
218 // If configured, rewrites the copy of the upstream request.
219 // Allows changing the request method and URI (path and query).
220 // Since the rewrite is applied to the copy, it does not persist
@@ -258,14 +274,16 @@ type Handler struct {
274 // Holds the handle_response Caddyfile tokens while adapting
275 handleResponseSegments []*caddyfile.Dispenser
276
261 - // Stores upgraded requests (hijacked connections) for proper cleanup
262 - connections map[io.ReadWriteCloser]openConnection
263 - connectionsCloseTimer *time.Timer
264 - connectionsMu *sync.Mutex
277 + // Tracks hijacked/upgraded connections (WebSocket etc.) so they can be
278 + // closed when their upstream is removed from the config.
279 + tunnelTracker *tunnelTracker
280
281 ctx caddy.Context
282 logger *zap.Logger
283 events *caddyevents.App
284 +
285 + streamLogLevel zapcore.Level
286 + streamLogLoggerName string
287 }
288
289 // CaddyModule returns the Caddy module information.
@@ -285,8 +303,25 @@ func (h *Handler) Provision(ctx caddy.Context) error {
303 h.events = eventAppIface.(*caddyevents.App)
304 h.ctx = ctx
305 h.logger = ctx.Logger()
288 - h.connections = make(map[io.ReadWriteCloser]openConnection)
289 - h.connectionsMu = new(sync.Mutex)
306 + h.tunnelTracker = newTunnelTracker(h.logger, time.Duration(h.StreamCloseDelay))
307 + h.streamLogLevel = defaultStreamLogLevel
308 + h.streamLogLoggerName = defaultStreamLoggerName
309 + if h.StreamLogs != nil {
310 + if h.StreamLogs.Level != "" {
311 + lvl, err := zapcore.ParseLevel(strings.ToLower(strings.TrimSpace(h.StreamLogs.Level)))
312 + if err != nil {
313 + return fmt.Errorf("invalid stream_logs.level %q: %w", h.StreamLogs.Level, err)
314 + }
315 + h.streamLogLevel = lvl
316 + }
317 + if name := strings.TrimSpace(h.StreamLogs.LoggerName); name != "" {
318 + h.streamLogLoggerName = name
319 + }
320 + }
321 +
322 + if h.StreamDetached {
323 + registerDetachedTunnelTrackers(h.tunnelTracker)
324 + }
325
326 // warn about unsafe buffering config
327 if h.RequestBuffers == -1 || h.ResponseBuffers == -1 {
@@ -456,15 +491,91 @@ func (h *Handler) Provision(ctx caddy.Context) error {
491 return nil
492 }
493
494 +func (h Handler) streamLogsSkipHandshake() bool {
495 + return h.StreamLogs != nil && h.StreamLogs.SkipHandshake
496 +}
497 +
498 +func (h Handler) streamLoggerForRequest(req *http.Request) *zap.Logger {
499 + name := strings.TrimSpace(h.streamLogLoggerName)
500 + if name == "" {
501 + name = defaultStreamLoggerName
502 + }
503 +
504 + if name == streamLoggerNameUseAccess {
505 + logger := caddy.Log().Named(defaultAccessLoggerBase)
506 + names := caddyhttp.GetVar(req.Context(), caddyhttp.AccessLoggerNameVarKey)
507 + namesSlice, ok := names.([]any)
508 + if !ok {
509 + return logger
510 + }
511 + for _, v := range namesSlice {
512 + name, ok := v.(string)
513 + if !ok {
514 + continue
515 + }
516 + if name == "" {
517 + return logger
518 + }
519 + return logger.Named(name)
520 + }
521 + return logger
522 + }
523 +
524 + return caddy.Log().Named(name)
525 +}
526 +
527 +var (
528 + detachedTunnelTrackers = make(map[*tunnelTracker]struct{})
529 + detachedTunnelTrackersMu sync.Mutex
530 +)
531 +
532 +func registerDetachedTunnelTrackers(ts *tunnelTracker) {
533 + detachedTunnelTrackersMu.Lock()
534 + defer detachedTunnelTrackersMu.Unlock()
535 + detachedTunnelTrackers[ts] = struct{}{}
536 +}
537 +
538 +func notifyDetachedTunnelTrackersOfUpstreamRemoval(upstream string, self *tunnelTracker) error {
539 + detachedTunnelTrackersMu.Lock()
540 + trackers := make([]*tunnelTracker, 0, len(detachedTunnelTrackers))
541 + for tunnel := range detachedTunnelTrackers {
542 + trackers = append(trackers, tunnel)
543 + }
544 + detachedTunnelTrackersMu.Unlock()
545 +
546 + // Connection deletion unregisters empty trackers. Do not hold the
547 + // registry lock while acquiring tracker locks or closing connections.
548 + var err error
549 + for _, tunnel := range trackers {
550 + if closeErr := tunnel.closeConnectionsForUpstream(upstream); closeErr != nil && tunnel == self && err == nil {
551 + err = closeErr
552 + }
553 + }
554 + return err
555 +}
556 +
557 +func unregisterDetachedTunnelTrackers(ts *tunnelTracker) {
558 + detachedTunnelTrackersMu.Lock()
559 + defer detachedTunnelTrackersMu.Unlock()
560 + delete(detachedTunnelTrackers, ts)
561 +}
562 +
563 // Cleanup cleans up the resources made by h.
564 func (h *Handler) Cleanup() error {
461 - err := h.cleanupConnections()
462 -
463 - // remove hosts from our config from the pool
565 + // even if StreamDetached is true, extended connect websockets may still be running
566 + err := h.tunnelTracker.cleanupAttachedConnections()
567 for _, upstream := range h.Upstreams {
465 - _, _ = hosts.Delete(upstream.String())
568 + // hosts.Delete returns deleted=true when the ref count reaches zero,
569 + // meaning no other active config references this upstream. In that
570 + // case close any tunnels proxying to it; otherwise let them survive
571 + // to their natural end since the upstream is still in use.
572 + deleted, _ := hosts.Delete(upstream.String())
573 + if deleted {
574 + if closeErr := notifyDetachedTunnelTrackersOfUpstreamRemoval(upstream.String(), h.tunnelTracker); closeErr != nil && err == nil {
575 + err = closeErr
576 + }
577 + }
578 }
467 -
579 return err
580 }
581
@@ -1218,10 +1329,11 @@ func (h *Handler) reverseProxy(rw http.ResponseWriter, req *http.Request, origRe
1329 // we use the original request here, so that any routes from 'next'
1330 // see the original request rather than the proxy cloned request.
1331 hrc := &handleResponseContext{
1221 - handler: h,
1222 - response: res,
1223 - start: start,
1224 - logger: logger,
1332 + handler: h,
1333 + response: res,
1334 + start: start,
1335 + logger: logger,
1336 + upstreamAddr: di.Upstream.String(),
1337 }
1338 ctx := origReq.Context()
1339 ctx = context.WithValue(ctx, proxyHandleResponseContextCtxKey, hrc)
@@ -1251,7 +1363,7 @@ func (h *Handler) reverseProxy(rw http.ResponseWriter, req *http.Request, origRe
1363 }
1364
1365 // copy the response body and headers back to the upstream client
1254 - return h.finalizeResponse(rw, req, res, repl, start, logger)
1366 + return h.finalizeResponse(rw, req, res, repl, start, logger, di.Upstream.String())
1367 }
1368
1369 // finalizeResponse prepares and copies the response.
@@ -1262,6 +1374,7 @@ func (h *Handler) finalizeResponse(
1374 repl *caddy.Replacer,
1375 start time.Time,
1376 logger *zap.Logger,
1377 + upstreamAddr string,
1378 ) error {
1379 responseWasIncremental := caddyhttp.IsIncremental(res.Header)
1380
@@ -1283,9 +1396,7 @@ func (h *Handler) finalizeResponse(
1396
1397 // deal with 101 Switching Protocols responses: (WebSocket, h2c, etc)
1398 if res.StatusCode == http.StatusSwitchingProtocols {
1286 - var wg sync.WaitGroup
1287 - h.handleUpgradeResponse(logger, &wg, rw, req, res)
1288 - wg.Wait()
1399 + h.handleUpgradeResponse(logger, rw, req, res, upstreamAddr)
1400 return nil
1401 }
1402
@@ -2047,6 +2158,22 @@ func (brc bodyReadCloser) Close() error {
2158 return nil
2159 }
2160
2161 +// StreamLogs controls logging for upgraded stream lifecycle events.
2162 +type StreamLogs struct {
2163 + // The minimum level at which stream lifecycle events are logged.
2164 + // Supported values are debug, info, warn, and error. Default: debug.
2165 + Level string `json:"level,omitempty"`
2166 +
2167 + // Logger name for stream lifecycle logs. Default: "http.handlers.reverse_proxy.stream".
2168 + // Special value "access" uses the access logger namespace and, if set,
2169 + // respects the first value in access_logger_names/log_name for the request.
2170 + LoggerName string `json:"logger_name,omitempty"`
2171 +
2172 + // If true, suppresses the access log entry normally emitted when an
2173 + // upgraded stream handshake completes and the request unwinds.
2174 + SkipHandshake bool `json:"skip_handshake,omitempty"`
2175 +}
2176 +
2177 // bufPool is used for buffering requests and responses.
2178 var bufPool = sync.Pool{
2179 New: func() any {
@@ -2079,6 +2206,9 @@ type handleResponseContext struct {
2206 // i.e. copied and closed, to make sure that it doesn't
2207 // happen twice.
2208 isFinalized bool
2209 +
2210 + // upstreamAddr is the selected upstream address for this request.
2211 + upstreamAddr string
2212 }
2213
2214 // proxyHandleResponseContextCtxKey is the context key for the active proxy handler
@@ -2089,6 +2219,13 @@ const proxyHandleResponseContextCtxKey caddy.CtxKey = "reverse_proxy_handle_resp
2219 // errNoUpstream occurs when there are no upstream available.
2220 var errNoUpstream = fmt.Errorf("no upstreams available")
2221
2222 +const (
2223 + defaultStreamLogLevel = zapcore.DebugLevel
2224 + defaultStreamLoggerName = "http.handlers.reverse_proxy.stream"
2225 + streamLoggerNameUseAccess = "access"
2226 + defaultAccessLoggerBase = "http.log.access"
2227 +)
2228 +
2229 // Interface guards
2230 var (
2231 _ caddy.Provisioner = (*Handler)(nil)
modules/caddyhttp/reverseproxy/streaming.go
+296 -124
@@ -26,6 +26,7 @@ import (
26 "io"
27 weakrand "math/rand/v2"
28 "mime"
29 + "net"
30 "net/http"
31 "sync"
32 "time"
@@ -35,15 +36,16 @@ import (
36 "go.uber.org/zap/zapcore"
37 "golang.org/x/net/http/httpguts"
38
39 + "github.com/caddyserver/caddy/v2"
40 "github.com/caddyserver/caddy/v2/modules/caddyhttp"
41 )
42
41 -type h2ReadWriteCloser struct {
43 +type extendedConnectReadWriteCloser struct {
44 io.ReadCloser
45 http.ResponseWriter
46 }
47
46 -func (rwc h2ReadWriteCloser) Write(p []byte) (n int, err error) {
48 +func (rwc extendedConnectReadWriteCloser) Write(p []byte) (n int, err error) {
49 n, err = rwc.ResponseWriter.Write(p)
50 if err != nil {
51 return 0, err
@@ -57,7 +59,7 @@ func (rwc h2ReadWriteCloser) Write(p []byte) (n int, err error) {
59 return n, nil
60 }
61
60 -func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup, rw http.ResponseWriter, req *http.Request, res *http.Response) {
62 +func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWriter, req *http.Request, res *http.Response, upstreamAddr string) {
63 reqUpType := upgradeType(req.Header)
64 resUpType := upgradeType(res.Header)
65
@@ -90,13 +92,37 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup,
92 copyHeader(rw.Header(), res.Header)
93 normalizeWebsocketHeaders(rw.Header())
94
95 + // Capture all h fields needed by the tunnel now, so that the Handler (h)
96 + // is not referenced after this function returns (for HTTP/1.1 hijacked
97 + // connections the tunnel runs in a detached goroutine).
98 + tunnel := h.tunnelTracker
99 + bufferSize := h.StreamBufferSize
100 + streamTimeout := time.Duration(h.StreamTimeout)
101 +
102 + if h.StreamDetached {
103 + // the return value should be true as it's not hijacked yet,
104 + // but some middleware may wrap response writers incorrectly
105 + if !caddyhttp.DetachResponseWriterAfterHijack(rw, true) {
106 + if c := logger.Check(zap.DebugLevel, "detaching connection failed"); c != nil {
107 + c.Write(zap.String("tip", "check if your response writers have an Unwrap method or if already hijacked"))
108 + }
109 + }
110 + }
111 +
112 var (
94 - conn io.ReadWriteCloser
95 - brw *bufio.ReadWriter
113 + conn io.ReadWriteCloser
114 + brw *bufio.ReadWriter
115 + detached = h.StreamDetached
116 )
97 - // websocket over http2 or http3 if extended connect is enabled, assuming backend doesn't support this, the request will be modified to http1.1 upgrade
98 - // TODO: once we can reliably detect backend support this, it can be removed for those backends
117 + // websocket over http2 or http3 if extended connect is enabled,
118 + // assuming backend doesn't support this, the request will be
119 + // modified to http1.1 upgrade
120 + // TODO: once we can reliably detect backend support this, it can
121 + // be removed for those backends
122 if body, ok := caddyhttp.GetVar(req.Context(), "extended_connect_websocket_body").(io.ReadCloser); ok {
123 + // websocket over extended connect can't be detached. rw and req.Body
124 + // are only valid while the handler goroutine is running
125 + detached = false
126 req.Body = body
127 rw.Header().Del("Upgrade")
128 rw.Header().Del("Connection")
@@ -104,18 +130,18 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup,
130 rw.WriteHeader(http.StatusOK)
131
132 if c := logger.Check(zap.DebugLevel, "upgrading connection"); c != nil {
107 - c.Write(zap.Int("http_version", 2))
133 + c.Write(zap.Int("http_version", req.ProtoMajor))
134 }
135
136 //nolint:bodyclose
137 flushErr := http.NewResponseController(rw).Flush()
138 if flushErr != nil {
113 - if c := h.logger.Check(zap.ErrorLevel, "failed to flush http2 websocket response"); c != nil {
139 + if c := h.logger.Check(zap.ErrorLevel, "failed to flush extended_connect websocket response"); c != nil {
140 c.Write(zap.Error(flushErr))
141 }
142 return
143 }
118 - conn = h2ReadWriteCloser{req.Body, rw}
144 + conn = extendedConnectReadWriteCloser{req.Body, rw}
145 // bufio is not needed, use minimal buffer
146 brw = bufio.NewReadWriter(bufio.NewReaderSize(conn, 1), bufio.NewWriterSize(conn, 1))
147 } else {
@@ -143,27 +169,6 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup,
169 }
170 }
171
146 - // adopted from https://github.com/golang/go/commit/8bcf2834afdf6a1f7937390903a41518715ef6f5
147 - backConnCloseCh := make(chan struct{})
148 - go func() {
149 - // Ensure that the cancellation of a request closes the backend.
150 - // See issue https://golang.org/issue/35559.
151 - select {
152 - case <-req.Context().Done():
153 - case <-backConnCloseCh:
154 - }
155 - backConn.Close()
156 - }()
157 - defer close(backConnCloseCh)
158 -
159 - start := time.Now()
160 - defer func() {
161 - conn.Close()
162 - if c := logger.Check(zapcore.DebugLevel, "connection closed"); c != nil {
163 - c.Write(zap.Duration("duration", time.Since(start)))
164 - }
165 - }()
166 -
172 if err := brw.Flush(); err != nil {
173 if c := logger.Check(zapcore.DebugLevel, "response flush"); c != nil {
174 c.Write(zap.Error(err))
@@ -184,13 +189,12 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup,
189 }
190 }
191
187 - // Ensure the hijacked client connection, and the new connection established
188 - // with the backend, are both closed in the event of a server shutdown. This
189 - // is done by registering them. We also try to gracefully close connections
190 - // we recognize as websockets.
191 - // We need to make sure the client connection messages (i.e. to upstream)
192 - // are masked, so we need to know whether the connection is considered the
193 - // server or the client side of the proxy.
192 + // Register both connections with the tunnel tracker. We also try to
193 + // gracefully close connections we recognize as websockets. We need to make
194 + // sure the client connection messages (i.e. to upstream) are masked, so we
195 + // need to know whether the connection is considered the server or the
196 + // client side of the proxy. Note that gracefulClose must not capture h,
197 + // since the tunnel may outlive the handler instance.
198 gracefulClose := func(conn io.ReadWriteCloser, isClient bool) func() error {
199 if isWebsocket(req) {
200 return func() error {
@@ -199,34 +203,100 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup,
203 }
204 return nil
205 }
202 - deleteFrontConn := h.registerConnection(conn, gracefulClose(conn, false))
203 - deleteBackConn := h.registerConnection(backConn, gracefulClose(backConn, true))
204 - defer deleteFrontConn()
206 + deleteFrontConn := tunnel.registerConnection(conn, gracefulClose(conn, false), detached, upstreamAddr)
207 + deleteBackConn := tunnel.registerConnection(backConn, gracefulClose(backConn, true), detached, upstreamAddr)
208 + if h.streamLogsSkipHandshake() {
209 + caddyhttp.SetVar(req.Context(), caddyhttp.LogSkipVar, true)
210 + }
211 + repl := req.Context().Value(caddy.ReplacerCtxKey).(*caddy.Replacer)
212 + repl.Set("http.reverse_proxy.upgraded", true)
213 + streamUUID, _ := repl.GetString("http.request.uuid")
214 + streamFields := makeStreamLogFields(streamUUID)
215 + streamLogger := h.streamLoggerForRequest(req)
216 + streamLevel := h.streamLogLevel
217 + finishMetrics := trackActiveStream(upstreamAddr)
218 +
219 + start := time.Now()
220 +
221 + if !detached {
222 + handleUpgradeTunnel(
223 + streamLogger,
224 + streamLevel,
225 + conn,
226 + backConn,
227 + deleteFrontConn,
228 + deleteBackConn,
229 + bufferSize,
230 + streamTimeout,
231 + start,
232 + finishMetrics,
233 + streamFields,
234 + )
235 + } else {
236 + // start a new goroutine
237 + go handleUpgradeTunnel(
238 + streamLogger,
239 + streamLevel,
240 + conn,
241 + backConn,
242 + deleteFrontConn,
243 + deleteBackConn,
244 + bufferSize,
245 + streamTimeout,
246 + start,
247 + finishMetrics,
248 + streamFields,
249 + )
250 + }
251 +}
252 +
253 +// handleUpgradeTunnel returns when transfer is done.
254 +func handleUpgradeTunnel(
255 + streamLogger *zap.Logger,
256 + streamLevel zapcore.Level,
257 + conn io.ReadWriteCloser,
258 + backConn io.ReadWriteCloser,
259 + deleteFrontConn func(),
260 + deleteBackConn func(),
261 + bufferSize int,
262 + streamTimeout time.Duration,
263 + start time.Time,
264 + finishMetrics func(result string, duration time.Duration, toBackend int64, fromBackend int64),
265 + streamFields []zap.Field,
266 +) {
267 defer deleteBackConn()
268 + defer deleteFrontConn()
269 + var (
270 + wg sync.WaitGroup
271 + toBackend int64
272 + fromBackend int64
273 + result string
274 + )
275
276 + // when a stream timeout is encountered, no error will be read from errc
277 + // a buffer size of 2 will allow both the read and write goroutines to
278 + // send the error and exit
279 + // see: https://github.com/caddyserver/caddy/issues/7418
280 + errc := make(chan error, 2)
281 spc := switchProtocolCopier{
282 user: conn,
283 backend: backConn,
210 - wg: wg,
211 - bufferSize: h.StreamBufferSize,
284 + wg: &wg,
285 + bufferSize: bufferSize,
286 + sent: &toBackend,
287 + received: &fromBackend,
288 }
289 + wg.Add(2)
290
214 - // setup the timeout if requested
291 var timeoutc <-chan time.Time
216 - if h.StreamTimeout > 0 {
217 - timer := time.NewTimer(time.Duration(h.StreamTimeout))
292 + if streamTimeout > 0 {
293 + timer := time.NewTimer(streamTimeout)
294 defer timer.Stop()
295 timeoutc = timer.C
296 }
297
222 - // when a stream timeout is encountered, no error will be read from errc
223 - // a buffer size of 2 will allow both the read and write goroutines to send the error and exit
224 - // see: https://github.com/caddyserver/caddy/issues/7418
225 - errc := make(chan error, 2)
226 - wg.Add(2)
298 go spc.copyToBackend(errc)
299 go spc.copyFromBackend(errc)
229 -
300 // Wait for both directions to finish. A clean EOF in one direction is a
301 // half-close of that direction only, not the end of the tunnel: the copier
302 // propagates it to the destination with CloseWrite and reports nil, so the
@@ -248,22 +318,62 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup,
318 //
319 // See https://github.com/caddyserver/caddy/issues/8026, and the same class
320 // fixed upstream in net/http/httputil (https://go.dev/issue/35892).
321 + result = "closed"
322 +copyLoop:
323 for range 2 {
324 select {
325 case err := <-errc:
326 if err != nil {
255 - if c := logger.Check(zapcore.DebugLevel, "streaming error"); c != nil {
327 + result = classifyStreamResult(err)
328 + if c := streamLogger.Check(streamLevel, "streaming error"); c != nil && !errors.Is(err, errCopyDone) {
329 c.Write(zap.Error(err))
330 }
258 - return
331 + break copyLoop
332 }
260 - case time := <-timeoutc:
261 - if c := logger.Check(zapcore.DebugLevel, "stream timed out"); c != nil {
262 - c.Write(zap.Time("timeout", time))
333 + case t := <-timeoutc:
334 + result = "timeout"
335 + if c := streamLogger.Check(streamLevel, "stream timed out"); c != nil {
336 + c.Write(zap.Time("timeout", t))
337 }
264 - return
338 + break copyLoop
339 }
340 }
341 +
342 + // Close both ends to unblock the still-running copy goroutine,
343 + // then wait for it so byte counts are final before metrics/logging.
344 + conn.Close()
345 + backConn.Close()
346 + wg.Wait()
347 +
348 + finishMetrics(result, time.Since(start), toBackend, fromBackend)
349 + if c := streamLogger.Check(streamLevel, "connection closed"); c != nil {
350 + fields := append([]zap.Field{}, streamFields...)
351 + fields = append(fields,
352 + zap.Duration("duration", time.Since(start)),
353 + zap.Int64("bytes_to_backend", toBackend),
354 + zap.Int64("bytes_from_backend", fromBackend),
355 + )
356 + c.Write(fields...)
357 + }
358 +}
359 +
360 +func classifyStreamResult(err error) string {
361 + if err == nil ||
362 + errors.Is(err, errCopyDone) ||
363 + errors.Is(err, io.EOF) ||
364 + errors.Is(err, net.ErrClosed) ||
365 + errors.Is(err, context.Canceled) {
366 + return "closed"
367 + }
368 + return "error"
369 +}
370 +
371 +func makeStreamLogFields(streamUUID string) []zap.Field {
372 + fields := make([]zap.Field, 0, 1)
373 + if streamUUID != "" {
374 + fields = append(fields, zap.String("uuid", streamUUID))
375 + }
376 + return fields
377 }
378
379 // flushInterval returns the p.FlushInterval value, conditionally
@@ -409,83 +519,118 @@ func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte, logger *za
519 }
520 }
521
412 -// registerConnection holds onto conn so it can be closed in the event
413 -// of a server shutdown. This is useful because hijacked connections or
414 -// connections dialed to backends don't close when server is shut down.
415 -// The caller should call the returned delete() function when the
416 -// connection is done to remove it from memory.
417 -func (h *Handler) registerConnection(conn io.ReadWriteCloser, gracefulClose func() error) (del func()) {
418 - h.connectionsMu.Lock()
419 - h.connections[conn] = openConnection{conn, gracefulClose}
420 - h.connectionsMu.Unlock()
522 +// openConnection maps an open connection to an optional function for graceful
523 +// close and records which upstream address the connection is proxying to.
524 +// Also tracks whether the connection is detached, which means it should only be
525 +// closed when the upstream is removed from the config, not on every reload.
526 +type openConnection struct {
527 + conn io.ReadWriteCloser
528 + gracefulClose func() error
529 + detached bool
530 + upstream string
531 +}
532 +
533 +// tunnelTracker tracks hijacked/upgraded connections for selective cleanup.
534 +// This exists to detach the lifecycle of streaming connections from the proxy
535 +// Handler and config, since we typically want them to survive past config reloads.
536 +// It also allows for selective connection cleanup based on their attachment status.
537 +type tunnelTracker struct {
538 + connections map[io.ReadWriteCloser]openConnection
539 + closeTimer *time.Timer
540 + closeDelay time.Duration
541 + stopped bool
542 + mu sync.Mutex
543 + logger *zap.Logger
544 +}
545 +
546 +func newTunnelTracker(logger *zap.Logger, closeDelay time.Duration) *tunnelTracker {
547 + return &tunnelTracker{
548 + connections: make(map[io.ReadWriteCloser]openConnection),
549 + closeDelay: closeDelay,
550 + logger: logger,
551 + }
552 +}
553 +
554 +// registerConnection stores conn in the tracking map. The caller must invoke
555 +// the returned del func when the connection is done.
556 +func (ts *tunnelTracker) registerConnection(conn io.ReadWriteCloser, gracefulClose func() error, detached bool, upstream string) (del func()) {
557 + ts.mu.Lock()
558 + ts.connections[conn] = openConnection{conn, gracefulClose, detached, upstream}
559 + ts.mu.Unlock()
560 return func() {
422 - h.connectionsMu.Lock()
423 - delete(h.connections, conn)
424 - // if there is no connection left before the connections close timer fires
425 - if len(h.connections) == 0 && h.connectionsCloseTimer != nil {
426 - // we release the timer that holds the reference to Handler
427 - if (*h.connectionsCloseTimer).Stop() {
428 - h.logger.Debug("stopped streaming connections close timer - all connections are already closed")
561 + ts.mu.Lock()
562 + delete(ts.connections, conn)
563 + empty := len(ts.connections) == 0 && ts.stopped
564 + if empty {
565 + if ts.closeTimer != nil {
566 + if ts.closeTimer.Stop() {
567 + ts.logger.Debug("stopped streaming connections close timer - all connections are already closed")
568 + }
569 + ts.closeTimer = nil
570 }
430 - h.connectionsCloseTimer = nil
571 }
432 - h.connectionsMu.Unlock()
572 + ts.mu.Unlock()
573 + if empty {
574 + unregisterDetachedTunnelTrackers(ts)
575 + }
576 }
577 }
578
436 -// closeConnections immediately closes all hijacked connections (both to client and backend).
437 -func (h *Handler) closeConnections() error {
579 +// closeAttachedConnections closes all tracked attached connections.
580 +func (ts *tunnelTracker) closeAttachedConnections() error {
581 var err error
439 - h.connectionsMu.Lock()
440 - defer h.connectionsMu.Unlock()
441 -
442 - for _, oc := range h.connections {
582 + ts.mu.Lock()
583 + defer ts.mu.Unlock()
584 + for _, oc := range ts.connections {
585 + // detached connections are only closed when the upstream is gone from the config
586 + if oc.detached {
587 + continue
588 + }
589 if oc.gracefulClose != nil {
444 - // this is potentially blocking while we have the lock on the connections
445 - // map, but that should be OK since the server has in theory shut down
446 - // and we are no longer using the connections map
447 - gracefulErr := oc.gracefulClose()
448 - if gracefulErr != nil && err == nil {
590 + if gracefulErr := oc.gracefulClose(); gracefulErr != nil && err == nil {
591 err = gracefulErr
592 }
593 }
452 - closeErr := oc.conn.Close()
453 - if closeErr != nil && err == nil {
594 + if closeErr := oc.conn.Close(); closeErr != nil && err == nil {
595 err = closeErr
596 }
597 }
598 return err
599 }
600
460 -// cleanupConnections closes hijacked connections.
461 -// Depending on the value of StreamCloseDelay it does that either immediately
462 -// or sets up a timer that will do that later.
463 -func (h *Handler) cleanupConnections() error {
464 - if h.StreamCloseDelay == 0 {
465 - return h.closeConnections()
466 - }
467 -
468 - h.connectionsMu.Lock()
469 - defer h.connectionsMu.Unlock()
470 - // the handler is shut down, no new connection can appear,
471 - // so we can skip setting up the timer when there are no connections
472 - if len(h.connections) > 0 {
473 - delay := time.Duration(h.StreamCloseDelay)
474 - h.connectionsCloseTimer = time.AfterFunc(delay, func() {
475 - if c := h.logger.Check(zapcore.DebugLevel, "closing streaming connections after delay"); c != nil {
476 - c.Write(zap.Duration("delay", delay))
477 - }
478 - err := h.closeConnections()
479 - if err != nil {
480 - if c := h.logger.Check(zapcore.ErrorLevel, "failed to closed connections after delay"); c != nil {
481 - c.Write(
482 - zap.Error(err),
483 - zap.Duration("delay", delay),
484 - )
485 - }
486 - }
487 - })
601 +// cleanupAttachedConnections closes upgraded attached connections.
602 +// Depending on closeDelay it does that either immediately or after a timer.
603 +func (ts *tunnelTracker) cleanupAttachedConnections() error {
604 + ts.mu.Lock()
605 + // Mark the tracker stopped before upstream-removal notifications, even
606 + // when attached connections will be closed after a delay.
607 + ts.stopped = true
608 + if len(ts.connections) == 0 {
609 + ts.mu.Unlock()
610 + unregisterDetachedTunnelTrackers(ts)
611 + return nil
612 + }
613 + if ts.closeDelay == 0 {
614 + ts.mu.Unlock()
615 + return ts.closeAttachedConnections()
616 }
617 +
618 + defer ts.mu.Unlock()
619 + delay := ts.closeDelay
620 + ts.closeTimer = time.AfterFunc(delay, func() {
621 + if c := ts.logger.Check(zapcore.DebugLevel, "closing streaming connections after delay"); c != nil {
622 + c.Write(zap.Duration("delay", delay))
623 + }
624 + err := ts.closeAttachedConnections()
625 + if err != nil {
626 + if c := ts.logger.Check(zapcore.ErrorLevel, "failed to close connections after delay"); c != nil {
627 + c.Write(
628 + zap.Error(err),
629 + zap.Duration("delay", delay),
630 + )
631 + }
632 + }
633 + })
634 return nil
635 }
636
@@ -601,11 +746,29 @@ func isWebsocket(r *http.Request) bool {
746 httpguts.HeaderValuesContainsToken(r.Header["Upgrade"], "websocket")
747 }
748
604 -// openConnection maps an open connection to
605 -// an optional function for graceful close.
606 -type openConnection struct {
607 - conn io.ReadWriteCloser
608 - gracefulClose func() error
749 +// closeConnectionsForUpstream closes detached connections to the given
750 +// upstream on stopped trackers. Attached connections keep their close delay.
751 +func (ts *tunnelTracker) closeConnectionsForUpstream(addr string) error {
752 + var err error
753 + ts.mu.Lock()
754 + defer ts.mu.Unlock()
755 + if !ts.stopped {
756 + return nil
757 + }
758 + for _, oc := range ts.connections {
759 + if !oc.detached || oc.upstream != addr {
760 + continue
761 + }
762 + if oc.gracefulClose != nil {
763 + if gracefulErr := oc.gracefulClose(); gracefulErr != nil && err == nil {
764 + err = gracefulErr
765 + }
766 + }
767 + if closeErr := oc.conn.Close(); closeErr != nil && err == nil {
768 + err = closeErr
769 + }
770 + }
771 + return err
772 }
773
774 type maxLatencyWriter struct {
@@ -676,6 +839,11 @@ type switchProtocolCopier struct {
839 user, backend io.ReadWriteCloser
840 wg *sync.WaitGroup
841 bufferSize int
842 + // sent and received accumulate byte counts for each direction.
843 + // They are written before wg.Done() and read after wg.Wait(), so no
844 + // additional synchronization is needed beyond the WaitGroup barrier.
845 + sent *int64 // bytes copied to backend; must be non-nil
846 + received *int64 // bytes copied from backend; must be non-nil
847 }
848
849 // errCopyDone is reported by a direction that ended at a clean EOF on a
@@ -685,7 +853,9 @@ var errCopyDone = errors.New("hijacked connection copy complete")
853
854 func (c switchProtocolCopier) copyFromBackend(errc chan<- error) {
855 defer c.wg.Done()
688 - if _, err := io.CopyBuffer(c.user, c.backend, c.buffer()); err != nil {
856 + n, err := io.CopyBuffer(c.user, c.backend, c.buffer())
857 + *c.received = n
858 + if err != nil {
859 errc <- err
860 return
861 }
@@ -701,7 +871,9 @@ func (c switchProtocolCopier) copyFromBackend(errc chan<- error) {
871
872 func (c switchProtocolCopier) copyToBackend(errc chan<- error) {
873 defer c.wg.Done()
704 - if _, err := io.CopyBuffer(c.backend, c.user, c.buffer()); err != nil {
874 + n, err := io.CopyBuffer(c.backend, c.user, c.buffer())
875 + *c.sent = n
876 + if err != nil {
877 errc <- err
878 return
879 }
modules/caddyhttp/reverseproxy/streaming_test.go
+254 -10
@@ -16,7 +16,10 @@ import (
16 "testing"
17 "time"
18
19 + "github.com/prometheus/client_golang/prometheus"
20 +
21 "github.com/caddyserver/caddy/v2"
22 + "github.com/caddyserver/caddy/v2/caddyconfig/caddyfile"
23 "github.com/caddyserver/caddy/v2/modules/caddyhttp"
24 )
25
@@ -50,12 +53,15 @@ func TestSwitchProtocolCopierBufferSize(t *testing.T) {
53 var wg sync.WaitGroup
54 var errc = make(chan error, 1)
55 var dst bytes.Buffer
56 + var sent, received int64
57
58 copier := switchProtocolCopier{
59 user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
60 backend: nopReadWriteCloser{Writer: &dst},
61 wg: &wg,
62 bufferSize: 7,
63 + sent: &sent,
64 + received: &received,
65 }
66
67 buf := copier.buffer()
@@ -67,9 +73,9 @@ func TestSwitchProtocolCopierBufferSize(t *testing.T) {
73 go copier.copyToBackend(errc)
74 wg.Wait()
75
70 - // the destination cannot be half-closed, so a clean copy reports errCopyDone
76 + // The destination cannot be half-closed, so a clean copy reports errCopyDone.
77 if err := <-errc; !errors.Is(err, errCopyDone) {
72 - t.Fatalf("copyToBackend() error = %v, want %v", err, errCopyDone)
78 + t.Fatalf("copyToBackend() error = %v", err)
79 }
80 if got := dst.String(); got != "hello" {
81 t.Fatalf("copied data = %q, want %q", got, "hello")
@@ -105,11 +111,14 @@ func TestSwitchProtocolCopierHalfClose(t *testing.T) {
111 } {
112 t.Run(tc.name, func(t *testing.T) {
113 var wg sync.WaitGroup
114 + var sent, received int64
115 errc := make(chan error, 1)
116 copier := switchProtocolCopier{
110 - user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
111 - backend: tc.dst,
112 - wg: &wg,
117 + user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
118 + backend: tc.dst,
119 + wg: &wg,
120 + sent: &sent,
121 + received: &received,
122 }
123
124 wg.Add(1)
@@ -130,14 +139,17 @@ func TestSwitchProtocolCopierHalfClose(t *testing.T) {
139 // half-close, so the tunnel is still torn down immediately.
140 func TestSwitchProtocolCopierErrorIsNotHalfClose(t *testing.T) {
141 var wg sync.WaitGroup
142 + var sent, received int64
143 errc := make(chan error, 1)
144 wantErr := errors.New("write failed")
145 dst := &closeWriteRecorder{writeErr: wantErr}
146
147 copier := switchProtocolCopier{
138 - user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
139 - backend: dst,
140 - wg: &wg,
148 + user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
149 + backend: dst,
150 + wg: &wg,
151 + sent: &sent,
152 + received: &received,
153 }
154
155 wg.Add(1)
@@ -167,6 +179,149 @@ type nopReadWriteCloser struct {
179
180 func (nopReadWriteCloser) Close() error { return nil }
181
182 +type trackingReadWriteCloser struct {
183 + closed chan struct{}
184 + one sync.Once
185 +}
186 +
187 +func newTrackingReadWriteCloser() *trackingReadWriteCloser {
188 + return &trackingReadWriteCloser{closed: make(chan struct{})}
189 +}
190 +
191 +func (c *trackingReadWriteCloser) Read(_ []byte) (int, error) { return 0, io.EOF }
192 +func (c *trackingReadWriteCloser) Write(p []byte) (int, error) { return len(p), nil }
193 +func (c *trackingReadWriteCloser) Close() error {
194 + c.one.Do(func() {
195 + close(c.closed)
196 + })
197 + return nil
198 +}
199 +
200 +func (c *trackingReadWriteCloser) isClosed() bool {
201 + select {
202 + case <-c.closed:
203 + return true
204 + default:
205 + return false
206 + }
207 +}
208 +
209 +func TestHandlerCleanupLegacyModeClosesAllConnections(t *testing.T) {
210 + ts := newTunnelTracker(caddy.Log(), 0)
211 + connA := newTrackingReadWriteCloser()
212 + connB := newTrackingReadWriteCloser()
213 + ts.registerConnection(connA, nil, false, "a")
214 + ts.registerConnection(connB, nil, false, "b")
215 +
216 + h := &Handler{
217 + tunnelTracker: ts,
218 + StreamDetached: false,
219 + }
220 +
221 + if err := h.Cleanup(); err != nil {
222 + t.Fatalf("cleanup failed: %v", err)
223 + }
224 + if !connA.isClosed() || !connB.isClosed() {
225 + t.Fatalf("legacy cleanup should close all upgraded connections")
226 + }
227 +}
228 +
229 +func TestHandlerCleanupLegacyModeHonorsDelay(t *testing.T) {
230 + ts := newTunnelTracker(caddy.Log(), 40*time.Millisecond)
231 + conn := newTrackingReadWriteCloser()
232 + ts.registerConnection(conn, nil, false, "a")
233 +
234 + h := &Handler{
235 + tunnelTracker: ts,
236 + StreamDetached: false,
237 + }
238 +
239 + if err := h.Cleanup(); err != nil {
240 + t.Fatalf("cleanup failed: %v", err)
241 + }
242 + if conn.isClosed() {
243 + t.Fatal("connection should not close immediately when stream_close_delay is set")
244 + }
245 +
246 + select {
247 + case <-conn.closed:
248 + case <-time.After(500 * time.Millisecond):
249 + t.Fatal("connection did not close after stream_close_delay elapsed")
250 + }
251 +}
252 +
253 +func TestHandlerCleanupDetachedModeClosesOnlyRemovedUpstreams(t *testing.T) {
254 + const upstreamA = "upstream-a"
255 + const upstreamB = "upstream-b"
256 +
257 + // Simulate old+new configs both referencing upstreamA (refcount 2),
258 + // while upstreamB is only referenced by the old config (refcount 1).
259 + hosts.LoadOrStore(upstreamA, struct{}{})
260 + hosts.LoadOrStore(upstreamA, struct{}{})
261 + hosts.LoadOrStore(upstreamB, struct{}{})
262 + t.Cleanup(func() {
263 + _, _ = hosts.Delete(upstreamA)
264 + _, _ = hosts.Delete(upstreamA)
265 + _, _ = hosts.Delete(upstreamB)
266 + })
267 +
268 + ts := newTunnelTracker(caddy.Log(), 0)
269 + registerDetachedTunnelTrackers(ts)
270 + connA := newTrackingReadWriteCloser()
271 + connB := newTrackingReadWriteCloser()
272 + ts.registerConnection(connA, nil, true, upstreamA)
273 + ts.registerConnection(connB, nil, true, upstreamB)
274 +
275 + h := &Handler{
276 + tunnelTracker: ts,
277 + StreamDetached: true,
278 + Upstreams: UpstreamPool{
279 + &Upstream{Dial: upstreamA},
280 + &Upstream{Dial: upstreamB},
281 + },
282 + }
283 +
284 + if err := h.Cleanup(); err != nil {
285 + t.Fatalf("cleanup failed: %v", err)
286 + }
287 +
288 + if connA.isClosed() {
289 + t.Fatal("connection for detached upstream should remain open")
290 + }
291 + if !connB.isClosed() {
292 + t.Fatal("connection for removed upstream should be closed")
293 + }
294 +}
295 +
296 +func TestHandlerUnmarshalCaddyfileStreamLogsBlock(t *testing.T) {
297 + d := caddyfile.NewTestDispenser(`
298 + reverse_proxy localhost:9000 {
299 + stream_logs {
300 + level info
301 + logger_name access
302 + skip_handshake
303 + }
304 + }
305 + `)
306 +
307 + var h Handler
308 + if err := h.UnmarshalCaddyfile(d); err != nil {
309 + t.Fatalf("UnmarshalCaddyfile() error = %v", err)
310 + }
311 + if h.StreamLogs == nil {
312 + t.Fatal("expected stream_logs to be configured")
313 + }
314 + if h.StreamLogs.Level != "info" {
315 + t.Fatalf("expected stream_logs.level=info, got %q", h.StreamLogs.Level)
316 + }
317 + if h.StreamLogs.LoggerName != "access" {
318 + t.Fatalf("expected stream_logs.logger_name=access, got %q", h.StreamLogs.LoggerName)
319 + }
320 + if !h.StreamLogs.SkipHandshake {
321 + t.Fatal("expected stream_logs.skip_handshake=true")
322 + }
323 +}
324 +
325 // closeWriteRecorder is a destination that supports closing only its write
326 // half, and records whether that happened.
327 type closeWriteRecorder struct {
@@ -312,8 +467,8 @@ func TestHandlerUpgradedStreamHalfClose(t *testing.T) {
467 Host: new(Host),
468 Dial: backend.Listener.Addr().String(),
469 })
315 - h.connections = make(map[io.ReadWriteCloser]openConnection)
316 - h.connectionsMu = new(sync.Mutex)
470 + h.tunnelTracker = newTunnelTracker(caddy.Log(), 0)
471 + initReverseProxyMetrics(h, prometheus.NewRegistry())
472
473 frontend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
474 // Isolate half-close propagation from request cancellation which has
@@ -398,3 +553,92 @@ func TestFlushIntervalIncremental(t *testing.T) {
553 })
554 }
555 }
556 +
557 +func TestHandlerCleanupDetachedRemovalWithDelay(t *testing.T) {
558 + const upstream = "detached-removal-with-delay"
559 + hosts.LoadOrStore(upstream, new(Host))
560 + t.Cleanup(func() { _, _ = hosts.Delete(upstream) })
561 + ts := newTunnelTracker(caddy.Log(), time.Hour)
562 + registerDetachedTunnelTrackers(ts)
563 + t.Cleanup(func() { unregisterDetachedTunnelTrackers(ts) })
564 + detached := newTrackingReadWriteCloser()
565 + attached := newTrackingReadWriteCloser()
566 + deleteDetached := ts.registerConnection(detached, nil, true, upstream)
567 + deleteAttached := ts.registerConnection(attached, nil, false, upstream)
568 + t.Cleanup(deleteDetached)
569 + t.Cleanup(deleteAttached)
570 + h := &Handler{tunnelTracker: ts, StreamDetached: true, Upstreams: UpstreamPool{&Upstream{Dial: upstream}}}
571 + if err := h.Cleanup(); err != nil {
572 + t.Fatal(err)
573 + }
574 + if !detached.isClosed() {
575 + t.Error("detached stream remains open after upstream removal")
576 + }
577 + if attached.isClosed() {
578 + t.Error("attached stream closed before stream_close_delay elapsed")
579 + }
580 +}
581 +
582 +func TestHandlerCleanupEmptyDetachedTracker(t *testing.T) {
583 + for _, delay := range []time.Duration{0, time.Hour} {
584 + t.Run(delay.String(), func(t *testing.T) {
585 + ts := newTunnelTracker(caddy.Log(), delay)
586 + registerDetachedTunnelTrackers(ts)
587 + t.Cleanup(func() { unregisterDetachedTunnelTrackers(ts) })
588 + // Cover a tracker whose last connection closed before handler cleanup.
589 + conn := newTrackingReadWriteCloser()
590 + ts.registerConnection(conn, nil, true, "upstream")()
591 + if err := (&Handler{tunnelTracker: ts, StreamDetached: true}).Cleanup(); err != nil {
592 + t.Fatal(err)
593 + }
594 + detachedTunnelTrackersMu.Lock()
595 + _, retained := detachedTunnelTrackers[ts]
596 + detachedTunnelTrackersMu.Unlock()
597 + if retained {
598 + t.Error("empty tracker remains globally registered after cleanup")
599 + }
600 + })
601 + }
602 +}
603 +
604 +func TestDetachedTrackerRemovalAllowsConcurrentConnectionDeletion(t *testing.T) {
605 + ts := newTunnelTracker(caddy.Log(), 0)
606 + registerDetachedTunnelTrackers(ts)
607 + t.Cleanup(func() { unregisterDetachedTunnelTrackers(ts) })
608 + // An attached stream on another stopped handler can finish while an
609 + // upstream-removal notification closes a detached stream. Its deletion
610 + // must be able to acquire the registry lock and release its tracker lock.
611 + other := newTunnelTracker(caddy.Log(), 0)
612 + otherConn := newTrackingReadWriteCloser()
613 + deleteOther := other.registerConnection(otherConn, nil, false, "other")
614 + if err := other.cleanupAttachedConnections(); err != nil {
615 + t.Fatal(err)
616 + }
617 + deleted := make(chan struct{})
618 + conn := newTrackingReadWriteCloser()
619 + del := ts.registerConnection(conn, func() error {
620 + go func() {
621 + deleteOther()
622 + close(deleted)
623 + }()
624 + select {
625 + case <-deleted:
626 + case <-time.After(time.Second):
627 + t.Error("upstream removal blocked another connection's deletion")
628 + }
629 + return nil
630 + }, true, "upstream")
631 + t.Cleanup(del)
632 + if err := ts.cleanupAttachedConnections(); err != nil {
633 + t.Fatal(err)
634 + }
635 + if err := notifyDetachedTunnelTrackersOfUpstreamRemoval("upstream", ts); err != nil {
636 + t.Fatal(err)
637 + }
638 + // On a regression the notification must release its lock before waiting
639 + // for deletion, so the failing test does not leave a blocked goroutine.
640 + <-deleted
641 + if !conn.isClosed() {
642 + t.Error("removed upstream connection was not closed")
643 + }
644 +}