Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS
Mirrored from https://github.com/Slazee/caddy.git
Synced Nov 10, 23:45 UTC
reverseproxy: send every close message before closing tunnel connections (#8371)
Closing one side of a tunnel ends its copy loops, which close the other side. closeConnectionsLocked sent each close message and closed that connection before moving on, in map order, so when the client side came first the upstream was often closed before its close message went out. Send every close message in the batch first, then close the connections. The second close then regularly finds the connection already closed. Treat net.ErrClosed as expected instead of returning it, so reloads with open websockets no longer log "use of closed network connection" at ERROR level. The upstream close message test goes back to running the real cleanup, which it can now do reliably.
Chris Oliver committed
Nov 8, 2026 at 05:05 UTC
553b62c3f3c306046c62221b48f78ad53fc4198a
2 files changed
+66
-14
modules/caddyhttp/reverseproxy/streaming.go
+15
-4
@@ -619,19 +619,30 @@ func (ts *tunnelTracker) closeAttachedConnections() error {
619
// tracked connection that match selects. The caller must hold mu. Every
620
// graceful close shares one absolute deadline: the loop is serial and holds
621
// the lock, so per-connection timeouts would add up across stalled peers.
622
+// All close messages are sent before any connection is closed, because
623
+// closing one side of a tunnel makes its copy loops close the other side,
624
+// which would lose a close message still waiting to be sent there.
625
// The first error is returned.
626
func (ts *tunnelTracker) closeConnectionsLocked(deadline time.Time, match func(openConnection) bool) error {
624
- var err error
627
+ var batch []openConnection
628
for _, oc := range ts.connections {
626
- if !match(oc) {
627
- continue
629
+ if match(oc) {
630
+ batch = append(batch, oc)
631
}
632
+ }
633
+
634
+ var err error
635
+ for _, oc := range batch {
636
if oc.gracefulClose != nil {
637
if gracefulErr := oc.gracefulClose(deadline); gracefulErr != nil && err == nil {
638
err = gracefulErr
639
}
640
}
634
- if closeErr := oc.conn.Close(); closeErr != nil && err == nil {
641
+ }
642
+ for _, oc := range batch {
643
+ // The other side of a tunnel may already have closed this one, which
644
+ // is the expected outcome rather than an error worth logging.
645
+ if closeErr := oc.conn.Close(); closeErr != nil && !errors.Is(closeErr, net.ErrClosed) && err == nil {
646
err = closeErr
647
}
648
}
modules/caddyhttp/reverseproxy/streaming_test.go
+51
-10
@@ -822,17 +822,11 @@ func TestHandlerUpgradeBoundsUpstreamCloseMessage(t *testing.T) {
822
t.Fatalf("backend read %q, %v; want %q", hello, err, "hello")
823
}
824
825
- // Send the close messages without closing anything. Cleanup closes the
826
- // connections in map order, and closing the client side first can tear
827
- // down the tunnel before the upstream gets its close message.
828
- deadline := time.Now().Add(5 * time.Second)
829
- h.tunnelTracker.mu.Lock()
830
- for _, oc := range h.tunnelTracker.connections {
831
- if err := oc.gracefulClose(deadline); err != nil {
832
- t.Errorf("graceful close = %v", err)
833
- }
825
+ // Closing the client side may tear down the tunnel before cleanup closes
826
+ // the upstream itself; that must not surface as a cleanup error.
827
+ if err := h.tunnelTracker.cleanupAttachedConnections(); err != nil {
828
+ t.Errorf("cleanup = %v, want nil", err)
829
}
835
- h.tunnelTracker.mu.Unlock()
830
frame := make([]byte, 2)
831
if _, err := io.ReadFull(srv, frame); err != nil {
832
t.Fatalf("backend did not receive a close message: %v", err)
@@ -1163,3 +1157,50 @@ func (c *closeTrackingBody) Close() error {
1157
c.closed = true
1158
return nil
1159
}
1160
+
1161
+func TestTunnelTrackerSendsCloseMessagesBeforeClosing(t *testing.T) {
1162
+ // Closing one side of a tunnel tears down the other, so every close
1163
+ // message in a batch must go out before any connection is closed.
1164
+ for _, tc := range []struct {
1165
+ name string
1166
+ detached bool
1167
+ close func(*tunnelTracker) error
1168
+ }{
1169
+ {name: "attached", close: (*tunnelTracker).cleanupAttachedConnections},
1170
+ {name: "detached", detached: true, close: func(ts *tunnelTracker) error {
1171
+ // closeConnectionsForUpstream only acts on a stopped tracker.
1172
+ ts.stopped = true
1173
+ return ts.closeConnectionsForUpstream("upstream", time.Now().Add(time.Second))
1174
+ }},
1175
+ } {
1176
+ t.Run(tc.name, func(t *testing.T) {
1177
+ ts := newTunnelTracker(caddy.Log(), 0)
1178
+ var events []string
1179
+ for _, side := range []string{"client", "upstream"} {
1180
+ ts.registerConnection(closeRecorder{side: side, events: &events}, func(time.Time) error {
1181
+ events = append(events, "close message to "+side)
1182
+ return nil
1183
+ }, tc.detached, "upstream")
1184
+ }
1185
+
1186
+ if err := tc.close(ts); err != nil {
1187
+ t.Fatal(err)
1188
+ }
1189
+ if len(events) != 4 || !strings.HasPrefix(events[0], "close message") || !strings.HasPrefix(events[1], "close message") {
1190
+ t.Errorf("events = %q, want both close messages before any connection closes", events)
1191
+ }
1192
+ })
1193
+ }
1194
+}
1195
+
1196
+// closeRecorder records when it is closed.
1197
+type closeRecorder struct {
1198
+ nopReadWriteCloser
1199
+ side string
1200
+ events *[]string
1201
+}
1202
+
1203
+func (c closeRecorder) Close() error {
1204
+ *c.events = append(*c.events, "closed "+c.side)
1205
+ return nil
1206
+}