@marcel91200 / caddy Mirror · read-only / commits / b3d67c86

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 10, 23:45 UTC

reverseproxy: set a write deadline on websocket close messages (#8309)

* reverseproxy: set a write deadline on websocket close messages Cleanup sends a Close control message to every attached websocket while holding the tunnel tracker lock, during a config reload. If the peer has stopped reading, the tunnel's copy goroutine is blocked writing to it and the close message waits behind that write forever. The reload never finishes and holds the config lock, so every later /load and /config request hangs until Caddy is restarted. Setting a write deadline before the close message unblocks both writes. The connection is closed right after, so the deadline is never cleared. * Add bot disclosure * reverseproxy: make close control timeout a const, harden stalled-peer test * Delete BOT_DISCLOSURE.md Signed-off-by: Francis Lavoie <lavofr@gmail.com> * reverseproxy: bound each batch of websocket close messages by one deadline The close loops are serial and hold the tunnel tracker lock, so a per-connection write timeout let N stalled peers hold a reload for N times the timeout. Compute one absolute deadline per batch instead, and skip the best-effort close frame on connections that cannot set a write deadline rather than risk the hang. Extended CONNECT streams gain SetWriteDeadline through http.ResponseController. For HTTP/1 upgrades, net/http returns the upstream socket inside a body that hides its deadline methods, so the upstream close message was never bounded; record the connection from the request's ClientTrace and write the close message to that socket. Restores the TODO about being the exclusive writer, per review. * reverseproxy: make websocket close tests deterministic Per review, the stalled-peer tests use net.Pipe instead of real TCP: it is unbuffered and lets one write through at a time, so a close message waits behind a stuck copy write without large allocations or sleeps. The batch test checks that graceful closes share one deadline instead of timing cleanup. The HTTP/1 upstream close test now runs a real upgrade through ServeHTTP and the transport, so it fails if the GotConn hook is removed. The choice of upstream close writer moves into upstreamCloseWriter, with a table test for the HTTP/2 and HTTP/3 case. * reverseproxy: record the upstream conn with atomic.Pointer Per review, store the connection from the GotConn hook in an atomic.Pointer[net.Conn] instead of a mutex-guarded holder type. Rename upstreamCloseWriter to upstreamWriteCloser. It returns io.ReadWriteCloser, since both the body and the recorded socket are one, so gracefulClose keeps its original io.ReadWriteCloser parameter. * reverseproxy: share one close deadline across detached trackers Per review, an upstream removal notifies every stopped tracker in turn, and each close loop holds its own lock, so a deadline per tracker let N trackers with stalled peers hold a reload for N times the timeout. Compute one absolute deadline in notifyDetachedTunnelTrackersOfUpstreamRemoval and pass it through to closeConnectionsForUpstream. Add a test with two stopped trackers holding stalled net.Pipe peers that asserts both graceful closes receive the same deadline. * reverseproxy: simplify the websocket close deadline change Fold the two tunnel close loops into closeConnectionsLocked, so the shared-deadline rule and its rationale live in one place. Replace the per-tracker closeWriteTimeout field, which production never varied, with a package variable that tests shorten. Record the upstream socket only for websocket requests, since only their graceful close reads it, and store a copy of the connection rather than the address of the hook's parameter. Use the package's caddy.CtxKey convention for the context key. Share the upgrade tunnel fixture between the half-close test and the upstream close message test.

Chris Oliver committed Nov 6, 2026 at 04:49 UTC b3d67c861d0957f68566298227241a171db6e883

3 files changed +401 -102

modules/caddyhttp/reverseproxy/reverseproxy.go

+13 -1

@@ -557,9 +557,11 @@ func notifyDetachedTunnelTrackersOfUpstreamRemoval(upstream string, self *tunnel

557

558 // Connection deletion unregisters empty trackers. Do not hold the

559 // registry lock while acquiring tracker locks or closing connections.

560 + // One deadline for every tracker; see closeConnectionsLocked.

561 + deadline := time.Now().Add(closeControlWriteTimeout)

562 var err error

563 for _, tunnel := range trackers {

562 - if closeErr := tunnel.closeConnectionsForUpstream(upstream); closeErr != nil && tunnel == self && err == nil {

564 + if closeErr := tunnel.closeConnectionsForUpstream(upstream, deadline); closeErr != nil && tunnel == self && err == nil {

565 err = closeErr

566 }

567 }

@@ -1242,6 +1244,16 @@ func (h *Handler) reverseProxy(rw http.ResponseWriter, req *http.Request, origRe

1244 // http.Transport reads a newer value to dial a new connection when that new

1245 // value is updated by another reverse proxy handler, typically forward_auth.

1246 ctx := context.WithValue(req.Context(), dialInfoCtxKey, di)

1247 + // Record the upstream socket for the websocket close message, since an

1248 + // upgraded HTTP/1 body hides it; see upstreamWriteCloser.

1249 + if isWebsocket(req) {

1250 + upstreamConn := new(atomic.Pointer[net.Conn])

1251 + trace.GotConn = func(info httptrace.GotConnInfo) {

1252 + conn := info.Conn

1253 + upstreamConn.Store(&conn)

1254 + }

1255 + ctx = context.WithValue(ctx, upstreamConnCtxKey, upstreamConn)

1256 + }

1257 req = req.WithContext(httptrace.WithClientTrace(ctx, trace))

1258

1259 // do the round-trip

modules/caddyhttp/reverseproxy/streaming.go

+83 -27

@@ -29,6 +29,7 @@ import (

29 "net"

30 "net/http"

31 "sync"

32 + "sync/atomic"

33 "time"

34 "unsafe"

@@ -59,6 +60,38 @@ func (rwc extendedConnectReadWriteCloser) Write(p []byte) (n int, err error) {

60 return n, nil

61 }

63 +// SetWriteDeadline bounds writes to the extended CONNECT stream, so a peer

64 +// that stopped reading cannot block the close message forever.

65 +func (rwc extendedConnectReadWriteCloser) SetWriteDeadline(deadline time.Time) error {

66 + //nolint:bodyclose

67 + return http.NewResponseController(rwc.ResponseWriter).SetWriteDeadline(deadline)

68 +}

69 +

70 +// upstreamConnCtxKey is the context key for the *atomic.Pointer[net.Conn]

71 +// recording the connection a websocket upgrade's round trip used.

72 +const upstreamConnCtxKey caddy.CtxKey = "reverse_proxy.upstream_conn"

73 +

74 +// upstreamWriteCloser returns where to write the close message to the

75 +// upstream. The write must be bounded by a deadline, but for HTTP/1

76 +// upgrades net/http hands the upstream socket back inside a body type that

77 +// hides its deadline methods and writes straight through to the socket, so

78 +// the close message goes to the socket recorded during the round trip

79 +// instead. An HTTP/1 connection is dedicated to this request; HTTP/2 and

80 +// HTTP/3 share theirs between streams, so their bodies are used as they are.

81 +func upstreamWriteCloser(req *http.Request, res *http.Response, backConn io.ReadWriteCloser) io.ReadWriteCloser {

82 + if res.ProtoMajor != 1 {

83 + return backConn

84 + }

85 + recorded, ok := req.Context().Value(upstreamConnCtxKey).(*atomic.Pointer[net.Conn])

86 + if !ok {

87 + return backConn

88 + }

89 + if uc := recorded.Load(); uc != nil {

90 + return *uc

91 + }

92 + return backConn

93 +}

94 +

95 func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWriter, req *http.Request, res *http.Response, upstreamAddr string) {

96 // Until the tunnel takes ownership, every setup failure must close the

97 // backend body and any hijacked client connection. net/http no longer

@@ -208,16 +241,16 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, rw http.ResponseWrit

241 // need to know whether the connection is considered the server or the

242 // client side of the proxy. Note that gracefulClose must not capture h,

243 // since the tunnel may outlive the handler instance.

211 - gracefulClose := func(conn io.ReadWriteCloser, isClient bool) func() error {

244 + gracefulClose := func(conn io.ReadWriteCloser, isClient bool) func(time.Time) error {

245 if isWebsocket(req) {

213 - return func() error {

214 - return writeCloseControl(conn, isClient)

246 + return func(deadline time.Time) error {

247 + return writeCloseControl(conn, isClient, deadline)

248 }

249 }

250 return nil

251 }

252 deleteFrontConn := tunnel.registerConnection(conn, gracefulClose(conn, false), detached, upstreamAddr)

220 - deleteBackConn := tunnel.registerConnection(backConn, gracefulClose(backConn, true), detached, upstreamAddr)

253 + deleteBackConn := tunnel.registerConnection(backConn, gracefulClose(upstreamWriteCloser(req, res, backConn), true), detached, upstreamAddr)

254 if h.streamLogsSkipHandshake() {

255 caddyhttp.SetVar(req.Context(), caddyhttp.LogSkipVar, true)

256 }

@@ -521,7 +554,7 @@ func (h Handler) copyBuffer(dst io.Writer, src io.Reader, buf []byte, logger *za

554 // closed when the upstream is removed from the config, not on every reload.

555 type openConnection struct {

556 conn io.ReadWriteCloser

524 - gracefulClose func() error

557 + gracefulClose func(deadline time.Time) error

558 detached bool

559 upstream string

560 }

@@ -549,7 +582,7 @@ func newTunnelTracker(logger *zap.Logger, closeDelay time.Duration) *tunnelTrack

582

583 // registerConnection stores conn in the tracking map. The caller must invoke

584 // the returned del func when the connection is done.

552 -func (ts *tunnelTracker) registerConnection(conn io.ReadWriteCloser, gracefulClose func() error, detached bool, upstream string) (del func()) {

585 +func (ts *tunnelTracker) registerConnection(conn io.ReadWriteCloser, gracefulClose func(deadline time.Time) error, detached bool, upstream string) (del func()) {

586 ts.mu.Lock()

587 ts.connections[conn] = openConnection{conn, gracefulClose, detached, upstream}

588 ts.mu.Unlock()

@@ -574,16 +607,27 @@ func (ts *tunnelTracker) registerConnection(conn io.ReadWriteCloser, gracefulClo

607

608 // closeAttachedConnections closes all tracked attached connections.

609 func (ts *tunnelTracker) closeAttachedConnections() error {

577 - var err error

610 ts.mu.Lock()

611 defer ts.mu.Unlock()

612 + // detached connections are only closed when the upstream is gone from the config

613 + return ts.closeConnectionsLocked(time.Now().Add(closeControlWriteTimeout), func(oc openConnection) bool {

614 + return !oc.detached

615 + })

616 +}

617 +

618 +// closeConnectionsLocked sends a close message to, then closes, every

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 +// The first error is returned.

623 +func (ts *tunnelTracker) closeConnectionsLocked(deadline time.Time, match func(openConnection) bool) error {

624 + var err error

625 for _, oc := range ts.connections {

581 - // detached connections are only closed when the upstream is gone from the config

582 - if oc.detached {

626 + if !match(oc) {

627 continue

628 }

629 if oc.gracefulClose != nil {

586 - if gracefulErr := oc.gracefulClose(); gracefulErr != nil && err == nil {

630 + if gracefulErr := oc.gracefulClose(deadline); gracefulErr != nil && err == nil {

631 err = gracefulErr

632 }

633 }

@@ -635,7 +679,24 @@ func (ts *tunnelTracker) cleanupAttachedConnections() error {

679 // from his simple implementation of this I was able to learn from at:

680 // github.com/pascaldekloe/websocket. Further work for handling masking

681 // taken from github.com/gorilla/websocket.

638 -func writeCloseControl(conn io.Writer, isClient bool) error {

682 +//

683 +// The write must finish by deadline: a peer that stopped reading leaves the

684 +// tunnel's copy goroutine stuck writing, and the close message would wait

685 +// behind it forever while the caller holds the tunnel tracker lock. The

686 +// connection is closed right after, so the deadline needs no reset.

687 +// Connections that cannot set a write deadline are skipped instead.

688 +func writeCloseControl(conn io.Writer, isClient bool, deadline time.Time) error {

689 + dc, ok := conn.(interface{ SetWriteDeadline(time.Time) error })

690 + if !ok {

691 + return nil

692 + }

693 + if err := dc.SetWriteDeadline(deadline); err != nil {

694 + if errors.Is(err, http.ErrNotSupported) {

695 + return nil

696 + }

697 + return err

698 + }

699 +

700 // Sources:

701 // https://github.com/pascaldekloe/websocket/blob/32050af67a5d/websocket.go#L119

702 // https://github.com/gorilla/websocket/blob/v1.5.0/conn.go#L413

@@ -685,6 +746,11 @@ func writeCloseControl(conn io.Writer, isClient bool) error {

746 return err

747 }

748

749 +// closeControlWriteTimeout is how long one batch of graceful closes waits

750 +// for peers to accept their close messages before giving up on them. It is

751 +// a variable so tests can shorten it.

752 +var closeControlWriteTimeout = 5 * time.Second

753 +

754 // Copied from https://github.com/gorilla/websocket/blob/v1.5.0/mask.go

755 func maskBytes(key [4]byte, pos int, b []byte) int {

756 // Mask one byte at a time for small buffers.

@@ -744,27 +810,17 @@ func isWebsocket(r *http.Request) bool {

810

811 // closeConnectionsForUpstream closes detached connections to the given

812 // upstream on stopped trackers. Attached connections keep their close delay.

747 -func (ts *tunnelTracker) closeConnectionsForUpstream(addr string) error {

748 - var err error

813 +// The deadline is shared so one upstream removal is bounded across every

814 +// tracker it notifies.

815 +func (ts *tunnelTracker) closeConnectionsForUpstream(addr string, deadline time.Time) error {

816 ts.mu.Lock()

817 defer ts.mu.Unlock()

818 if !ts.stopped {

819 return nil

820 }

754 - for _, oc := range ts.connections {

755 - if !oc.detached || oc.upstream != addr {

756 - continue

757 - }

758 - if oc.gracefulClose != nil {

759 - if gracefulErr := oc.gracefulClose(); gracefulErr != nil && err == nil {

760 - err = gracefulErr

761 - }

762 - }

763 - if closeErr := oc.conn.Close(); closeErr != nil && err == nil {

764 - err = closeErr

765 - }

766 - }

767 - return err

821 + return ts.closeConnectionsLocked(deadline, func(oc openConnection) bool {

822 + return oc.detached && oc.upstream == addr

823 + })

824 }

825

826 type maxLatencyWriter struct {

modules/caddyhttp/reverseproxy/streaming_test.go

+305 -74

@@ -10,9 +10,11 @@ import (

10 "net/http"

11 "net/http/httptest"

12 "net/url"

13 + "os"

14 "runtime"

15 "strings"

16 "sync"

17 + "sync/atomic"

18 "testing"

19 "time"

@@ -441,86 +443,93 @@ func TestHandlerUpgradedStreamHalfClose(t *testing.T) {

443

444 for name, test := range tests {

445 t.Run(name, func(t *testing.T) {

444 - srvc := make(chan *net.TCPConn, 1)

445 -

446 - backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

447 - conn, _, err := http.NewResponseController(w).Hijack()

448 - if err != nil {

449 - t.Errorf("backend hijack failed: %v", err)

450 - return

451 - }

452 - tcp, ok := conn.(*net.TCPConn)

453 - if !ok {

454 - conn.Close()

455 - t.Errorf("backend conn is %T, want *net.TCPConn", conn)

456 - return

457 - }

458 - if _, err := io.WriteString(tcp, "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"); err != nil {

459 - tcp.Close()

460 - t.Errorf("backend upgrade write failed: %v", err)

461 - return

462 - }

463 - srvc <- tcp

464 - }))

465 - defer backend.Close()

446 + _, cli, srv := newUpgradedTunnel(t)

447 + test(t, cli, srv)

448 + })

449 + }

450 +}

451

467 - h := minimalHandler(0, &Upstream{

468 - Host: new(Host),

469 - Dial: backend.Listener.Addr().String(),

470 - })

471 - h.tunnelTracker = newTunnelTracker(caddy.Log(), 0)

472 - initReverseProxyMetrics(h, prometheus.NewRegistry())

452 +// newUpgradedTunnel proxies a websocket upgrade from a raw client connection

453 +// through a Handler to a hijacking backend and returns both ends of the

454 +// tunnel. The Handler's tunnel tracker holds the proxied connections.

455 +func newUpgradedTunnel(t *testing.T) (h *Handler, cli, srv *net.TCPConn) {

456 + t.Helper()

457 + srvc := make(chan *net.TCPConn, 1)

458

474 - frontend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

475 - // Isolate half-close propagation from request cancellation which has

476 - // its own backend-close path.

477 - r = r.WithContext(context.WithoutCancel(r.Context()))

478 - _ = h.ServeHTTP(w, prepareTestRequest(r), caddyhttp.HandlerFunc(func(http.ResponseWriter, *http.Request) error {

479 - return nil

480 - }))

481 - }))

482 - defer frontend.Close()

483 -

484 - frontendURL, err := url.Parse(frontend.URL)

485 - if err != nil {

486 - t.Fatalf("failed to parse frontend URL: %v", err)

487 - }

488 - addr, err := net.ResolveTCPAddr("tcp", frontendURL.Host)

489 - if err != nil {

490 - t.Fatalf("failed to resolve TCP address: %v", err)

491 - }

492 - cli, err := net.DialTCP("tcp", nil, addr)

493 - if err != nil {

494 - t.Fatalf("failed to dial frontend: %v", err)

495 - }

496 - defer cli.Close()

459 + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

460 + conn, _, err := http.NewResponseController(w).Hijack()

461 + if err != nil {

462 + t.Errorf("backend hijack failed: %v", err)

463 + return

464 + }

465 + tcp, ok := conn.(*net.TCPConn)

466 + if !ok {

467 + conn.Close()

468 + t.Errorf("backend conn is %T, want *net.TCPConn", conn)

469 + return

470 + }

471 + if _, err := io.WriteString(tcp, "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"); err != nil {

472 + tcp.Close()

473 + t.Errorf("backend upgrade write failed: %v", err)

474 + return

475 + }

476 + srvc <- tcp

477 + }))

478 + t.Cleanup(backend.Close)

479

498 - req, _ := http.NewRequest(http.MethodGet, frontend.URL, nil)

499 - req.Header.Set("Connection", "Upgrade")

500 - req.Header.Set("Upgrade", "websocket")

501 - if err := req.Write(cli); err != nil {

502 - t.Fatalf("failed to write upgrade request: %v", err)

503 - }

480 + h = minimalHandler(0, &Upstream{

481 + Host: new(Host),

482 + Dial: backend.Listener.Addr().String(),

483 + })

484 + h.tunnelTracker = newTunnelTracker(caddy.Log(), 0)

485 + initReverseProxyMetrics(h, prometheus.NewRegistry())

486 +

487 + frontend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

488 + // Isolate the tunnel from request cancellation, which has its own

489 + // backend-close path.

490 + r = r.WithContext(context.WithoutCancel(r.Context()))

491 + _ = h.ServeHTTP(w, prepareTestRequest(r), caddyhttp.HandlerFunc(func(http.ResponseWriter, *http.Request) error {

492 + return nil

493 + }))

494 + }))

495 + t.Cleanup(frontend.Close)

496 +

497 + frontendURL, err := url.Parse(frontend.URL)

498 + if err != nil {

499 + t.Fatalf("failed to parse frontend URL: %v", err)

500 + }

501 + addr, err := net.ResolveTCPAddr("tcp", frontendURL.Host)

502 + if err != nil {

503 + t.Fatalf("failed to resolve TCP address: %v", err)

504 + }

505 + cli, err = net.DialTCP("tcp", nil, addr)

506 + if err != nil {

507 + t.Fatalf("failed to dial frontend: %v", err)

508 + }

509 + t.Cleanup(func() { cli.Close() })

510

505 - resp, err := http.ReadResponse(bufio.NewReader(cli), &http.Request{Method: http.MethodGet})

506 - if err != nil {

507 - t.Fatalf("failed to read upgrade response: %v", err)

508 - }

509 - if resp.StatusCode != http.StatusSwitchingProtocols {

510 - t.Fatalf("status = %d, want 101", resp.StatusCode)

511 - }

511 + req, _ := http.NewRequest(http.MethodGet, frontend.URL, nil)

512 + req.Header.Set("Connection", "Upgrade")

513 + req.Header.Set("Upgrade", "websocket")

514 + if err := req.Write(cli); err != nil {

515 + t.Fatalf("failed to write upgrade request: %v", err)

516 + }

517

513 - var srv *net.TCPConn

514 - select {

515 - case srv = <-srvc:

516 - case <-time.After(5 * time.Second):

517 - t.Fatal("timed out waiting for the backend connection")

518 - }

519 - defer srv.Close()

518 + resp, err := http.ReadResponse(bufio.NewReader(cli), &http.Request{Method: http.MethodGet})

519 + if err != nil {

520 + t.Fatalf("failed to read upgrade response: %v", err)

521 + }

522 + if resp.StatusCode != http.StatusSwitchingProtocols {

523 + t.Fatalf("status = %d, want 101", resp.StatusCode)

524 + }

525

521 - test(t, cli, srv)

522 - })

526 + select {

527 + case srv = <-srvc:

528 + case <-time.After(10 * time.Second):

529 + t.Fatal("timed out waiting for the backend connection")

530 }

531 + t.Cleanup(func() { srv.Close() })

532 + return h, cli, srv

533 }

534

535 // A response carrying the Incremental header field (RFC 10036) must be

@@ -617,7 +626,7 @@ func TestDetachedTrackerRemovalAllowsConcurrentConnectionDeletion(t *testing.T)

626 }

627 deleted := make(chan struct{})

628 conn := newTrackingReadWriteCloser()

620 - del := ts.registerConnection(conn, func() error {

629 + del := ts.registerConnection(conn, func(time.Time) error {

630 go func() {

631 deleteOther()

632 close(deleted)

@@ -644,6 +653,228 @@ func TestDetachedTrackerRemovalAllowsConcurrentConnectionDeletion(t *testing.T)

653 }

654 }

655

656 +// newStalledWebsocketPeer returns one end of a pipe whose other end never

657 +// reads, with a goroutine stuck writing to it the way a tunnel's copy loop

658 +// would be. net.Pipe is unbuffered and lets only one write through at a

659 +// time, so a close message written to conn waits behind that stuck write,

660 +// just as it does on a socket whose peer stopped reading.

661 +func newStalledWebsocketPeer(t *testing.T) net.Conn {

662 + t.Helper()

663 + conn, peer := net.Pipe()

664 + t.Cleanup(func() {

665 + conn.Close()

666 + peer.Close()

667 + })

668 + go func() { _, _ = conn.Write([]byte("stalled")) }()

669 + return conn

670 +}

671 +

672 +// shortenCloseControlWriteTimeout makes a stalled peer time out quickly for

673 +// the rest of the test.

674 +func shortenCloseControlWriteTimeout(t *testing.T) {

675 + t.Helper()

676 + saved := closeControlWriteTimeout

677 + closeControlWriteTimeout = 10 * time.Millisecond

678 + t.Cleanup(func() { closeControlWriteTimeout = saved })

679 +}

680 +

681 +func TestHandlerCleanupDoesNotBlockOnStalledWebsocketPeer(t *testing.T) {

682 + conn := newStalledWebsocketPeer(t)

683 +

684 + ts := newTunnelTracker(caddy.Log(), 0)

685 + shortenCloseControlWriteTimeout(t)

686 + ts.registerConnection(conn, func(deadline time.Time) error {

687 + return writeCloseControl(conn, false, deadline)

688 + }, false, "upstream")

689 +

690 + cleanupDone := make(chan error, 1)

691 + go func() { cleanupDone <- ts.cleanupAttachedConnections() }()

692 +

693 + select {

694 + case err := <-cleanupDone:

695 + if !errors.Is(err, os.ErrDeadlineExceeded) {

696 + t.Errorf("cleanup = %v, want %v", err, os.ErrDeadlineExceeded)

697 + }

698 + case <-time.After(10 * time.Second):

699 + t.Fatal("cleanup blocked on a websocket peer that stopped reading")

700 + }

701 +}

702 +

703 +func TestHandlerCleanupUsesOneDeadlinePerBatch(t *testing.T) {

704 + // The close loop is serial and holds the tracker lock, so a deadline per

705 + // connection would let every stalled peer add another full timeout.

706 + ts := newTunnelTracker(caddy.Log(), 0)

707 + shortenCloseControlWriteTimeout(t)

708 + var deadlines []time.Time

709 + for range 2 {

710 + conn := newStalledWebsocketPeer(t)

711 + ts.registerConnection(conn, func(deadline time.Time) error {

712 + deadlines = append(deadlines, deadline)

713 + return writeCloseControl(conn, false, deadline)

714 + }, false, "upstream")

715 + }

716 +

717 + _ = ts.cleanupAttachedConnections()

718 + if len(deadlines) != 2 {

719 + t.Fatalf("got %d graceful closes, want 2", len(deadlines))

720 + }

721 + if !deadlines[0].Equal(deadlines[1]) {

722 + t.Errorf("graceful closes got deadlines %v and %v, want one shared deadline", deadlines[0], deadlines[1])

723 + }

724 +}

725 +

726 +func TestUpstreamRemovalUsesOneDeadlineAcrossTrackers(t *testing.T) {

727 + // Removing an upstream notifies every stopped tracker in turn, and each

728 + // close loop holds its own lock, so a deadline per tracker would let

729 + // every tracker with a stalled peer add another full timeout.

730 + const upstream = "upstream-removal-one-deadline"

731 + shortenCloseControlWriteTimeout(t)

732 + var deadlines []time.Time

733 + var trackers []*tunnelTracker

734 + for range 2 {

735 + ts := newTunnelTracker(caddy.Log(), 0)

736 + registerDetachedTunnelTrackers(ts)

737 + t.Cleanup(func() { unregisterDetachedTunnelTrackers(ts) })

738 + conn := newStalledWebsocketPeer(t)

739 + ts.registerConnection(conn, func(deadline time.Time) error {

740 + deadlines = append(deadlines, deadline)

741 + return writeCloseControl(conn, false, deadline)

742 + }, true, upstream)

743 + if err := ts.cleanupAttachedConnections(); err != nil {

744 + t.Fatal(err)

745 + }

746 + trackers = append(trackers, ts)

747 + }

748 +

749 + _ = notifyDetachedTunnelTrackersOfUpstreamRemoval(upstream, trackers[0])

750 + if len(deadlines) != 2 {

751 + t.Fatalf("got %d graceful closes, want 2", len(deadlines))

752 + }

753 + if !deadlines[0].Equal(deadlines[1]) {

754 + t.Errorf("trackers got deadlines %v and %v, want one shared deadline", deadlines[0], deadlines[1])

755 + }

756 +}

757 +

758 +func TestWriteCloseControlSkipsConnectionsWithoutWriteDeadline(t *testing.T) {

759 + // A write that cannot be bounded could block forever behind a stalled

760 + // peer, so the best-effort close message is skipped instead.

761 + var buf bytes.Buffer

762 + if err := writeCloseControl(&buf, false, time.Now().Add(time.Second)); err != nil {

763 + t.Fatal(err)

764 + }

765 + if buf.Len() != 0 {

766 + t.Errorf("close message %q was written to a connection without write deadlines", buf.Bytes())

767 + }

768 +}

769 +

770 +func TestExtendedConnectReadWriteCloserSetWriteDeadline(t *testing.T) {

771 + rw := &writeDeadlineRecorder{ResponseRecorder: httptest.NewRecorder()}

772 + rwc := extendedConnectReadWriteCloser{io.NopCloser(strings.NewReader("")), rw}

773 + deadline := time.Now().Add(time.Second)

774 + if err := rwc.SetWriteDeadline(deadline); err != nil {

775 + t.Fatal(err)

776 + }

777 + if !rw.writeDeadline.Equal(deadline) {

778 + t.Errorf("write deadline = %v, want %v", rw.writeDeadline, deadline)

779 + }

780 +

781 + // Writers without deadline support report ErrNotSupported, which

782 + // writeCloseControl treats as a reason to skip the close message.

783 + plain := extendedConnectReadWriteCloser{io.NopCloser(strings.NewReader("")), httptest.NewRecorder()}

784 + if err := plain.SetWriteDeadline(deadline); !errors.Is(err, http.ErrNotSupported) {

785 + t.Errorf("SetWriteDeadline on a plain recorder = %v, want ErrNotSupported", err)

786 + }

787 +}

788 +

789 +type writeDeadlineRecorder struct {

790 + *httptest.ResponseRecorder

791 + writeDeadline time.Time

792 +}

793 +

794 +func (r *writeDeadlineRecorder) SetWriteDeadline(deadline time.Time) error {

795 + r.writeDeadline = deadline

796 + return nil

797 +}

798 +

799 +func TestHandlerUpgradeBoundsUpstreamCloseMessage(t *testing.T) {

800 + switch runtime.GOOS {

801 + case "plan9", "js", "wasip1":

802 + t.Skipf("not supported on %s", runtime.GOOS)

803 + }

804 +

805 + // net/http hands upgraded HTTP/1 bodies back inside a type that hides the

806 + // socket's deadline methods, so without the connection recorded during

807 + // the round trip the close message to the upstream would be skipped.

808 + h, cli, srv := newUpgradedTunnel(t)

809 + if err := cli.SetDeadline(time.Now().Add(10 * time.Second)); err != nil {

810 + t.Fatal(err)

811 + }

812 + if err := srv.SetDeadline(time.Now().Add(10 * time.Second)); err != nil {

813 + t.Fatal(err)

814 + }

815 +

816 + // Data reaching the backend proves both connections are registered.

817 + if _, err := io.WriteString(cli, "hello"); err != nil {

818 + t.Fatal(err)

819 + }

820 + hello := make([]byte, len("hello"))

821 + if _, err := io.ReadFull(srv, hello); err != nil || string(hello) != "hello" {

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

834 + }

835 + h.tunnelTracker.mu.Unlock()

836 + frame := make([]byte, 2)

837 + if _, err := io.ReadFull(srv, frame); err != nil {

838 + t.Fatalf("backend did not receive a close message: %v", err)

839 + }

840 + if frame[0] != 0x88 || frame[1]&0x80 == 0 {

841 + t.Fatalf("backend received %x, want a masked close frame", frame)

842 + }

843 +}

844 +

845 +func TestUpstreamWriteCloser(t *testing.T) {

846 + // The HTTP/1 path, including the recorded socket, is covered end to end

847 + // by TestHandlerUpgradeBoundsUpstreamCloseMessage. HTTP/2 and HTTP/3 share

848 + // their connection between streams, so the close message must never go

849 + // to a recorded socket for them.

850 + body, _ := net.Pipe()

851 + socket, _ := net.Pipe()

852 + for _, tc := range []struct {

853 + name string

854 + protoMajor int

855 + recorded net.Conn

856 + want io.ReadWriteCloser

857 + }{

858 + {name: "http/1 with recorded socket", protoMajor: 1, recorded: socket, want: socket},

859 + {name: "http/1 without recorded socket", protoMajor: 1, want: body},

860 + {name: "http/2", protoMajor: 2, recorded: socket, want: body},

861 + {name: "http/3", protoMajor: 3, recorded: socket, want: body},

862 + } {

863 + t.Run(tc.name, func(t *testing.T) {

864 + recorded := new(atomic.Pointer[net.Conn])

865 + if tc.recorded != nil {

866 + recorded.Store(&tc.recorded)

867 + }

868 + req := httptest.NewRequest(http.MethodGet, "/", nil)

869 + req = req.WithContext(context.WithValue(req.Context(), upstreamConnCtxKey, recorded))

870 + res := &http.Response{ProtoMajor: tc.protoMajor}

871 + if got := upstreamWriteCloser(req, res, body); got != tc.want {

872 + t.Errorf("upstreamWriteCloser() = %v, want %v", got, tc.want)

873 + }

874 + })

875 + }

876 +}

877 +

878 // opaqueUpgradeWriter supports hijacking but hides the recorder from the

879 // detachment helper, as a middleware wrapper without Unwrap would do.

880 type opaqueUpgradeWriter struct {