Revert "TUN-10621: Propagate max wait timeout"

This reverts merge request !1859

carrier/websocket.go

@@ -37,7 +37,7 @@ func (ws *Websocket) ServeStream(options *StartOptions, conn io.ReadWriter) erro
 	}
 	defer func() { _ = wsConn.Close() }()
 
-	stream.Pipe(wsConn, conn, stream.DefaultTimeoutAfterFirstClose, ws.log)
+	stream.Pipe(wsConn, conn, ws.log)
 	return nil
 }
 

carrier/websocket_test.go

@@ -5,12 +5,8 @@ import (
 	crand "crypto/rand"
 	"crypto/tls"
 	"crypto/x509"
-	"errors"
 	"fmt"
-	"io"
 	"math/big"
-	"net/http"
-	"net/http/httptest"
 	"testing"
 	"time"
 
@@ -21,7 +17,6 @@ import (
 	"golang.org/x/net/websocket"
 
 	"github.com/cloudflare/cloudflared/hello"
-	"github.com/cloudflare/cloudflared/stream"
 	"github.com/cloudflare/cloudflared/tlsconfig"
 	cfwebsocket "github.com/cloudflare/cloudflared/websocket"
 )
@@ -132,137 +127,3 @@ func TestWebsocketWrapper(t *testing.T) {
 	require.Equal(t, 2, n)
 	require.Equal(t, "bc", string(buf[:n]))
 }
-
-// halfClosePipeEnd is a bidirectional pipe built from two independent io.Pipe
-// pairs. Unlike net.Pipe, closing one direction does not affect the other, which
-// lets us simulate a half-close: the local client can signal EOF on its write
-// side while still reading a delayed response from the remote side.
-type halfClosePipeEnd struct {
-	r *io.PipeReader // data flowing into this end
-	w *io.PipeWriter // data flowing out of this end
-}
-
-func newHalfClosePipe() (local, remote *halfClosePipeEnd) {
-	// Pipe A carries data from local → remote.
-	ar, aw := io.Pipe()
-	// Pipe B carries data from remote → local.
-	br, bw := io.Pipe()
-
-	local = &halfClosePipeEnd{r: br, w: aw}
-	remote = &halfClosePipeEnd{r: ar, w: bw}
-	return local, remote
-}
-
-func (p *halfClosePipeEnd) Read(b []byte) (int, error)  { return p.r.Read(b) }
-func (p *halfClosePipeEnd) Write(b []byte) (int, error) { return p.w.Write(b) }
-
-// CloseWrite signals EOF to the remote reader without closing the read side.
-func (p *halfClosePipeEnd) CloseWrite() error { return p.w.Close() }
-
-// Close shuts down both directions.
-func (p *halfClosePipeEnd) Close() error {
-	_ = p.r.Close()
-	return p.w.Close()
-}
-
-// TestServeStreamWaitsForResponseAfterLocalClose exercises the websocket path
-// It verifies that the pipe does not tear down immediately after the client closes
-// its write side.
-//
-// The setup mirrors the cloudflared access tcp path:
-//
-//	local app -> halfClosePipe -> ServeStream -> mock WS echo server
-//
-// Sequence:
-//  1. Write payload to the mock server through ServeStream.
-//  2. Half-close the write side (CloseWrite) — signals EOF upstream.
-//  3. Sleep for halfCloseWait to make the race window explicit: a buggy
-//     (timeout=0) pipe would already have torn down the connection here.
-//  4. Read the echo — must succeed because the patch keeps the pipe alive.
-func TestServeStreamWaitsForResponseAfterLocalClose(t *testing.T) {
-	t.Parallel()
-
-	const (
-		payload = "half-close-test"
-		// halfCloseWait makes the race window visible: if ServeStream tears
-		// down the connection on CloseWrite the read that follows will fail
-		// immediately, mirroring the 3-second sleep in the cftunnel reference
-		// test. It must be shorter than DefaultTimeoutAfterFirstClose (10s).
-		halfCloseWait = 3 * time.Second
-		// testTimeout is an upper bound for the whole test.
-		testTimeout = stream.DefaultTimeoutAfterFirstClose + 5*time.Second
-	)
-
-	server := websocketServer()
-	defer server.Close()
-
-	localEnd, remoteEnd := newHalfClosePipe()
-
-	log := zerolog.Nop()
-	wsConn := NewWSConnection(&log)
-	options := &StartOptions{
-		OriginURL: "ws://" + server.Listener.Addr().String(),
-	}
-
-	serveErrCh := make(chan error, 1)
-	go func() {
-		serveErrCh <- wsConn.ServeStream(options, remoteEnd)
-	}()
-
-	ctx, cancel := context.WithTimeout(t.Context(), testTimeout)
-	defer cancel()
-
-	// 1. Write the payload. ServeStream forwards it as a WS binary frame.
-	_, err := localEnd.Write([]byte(payload))
-	require.NoError(t, err)
-
-	// 2. Half-close the write side
-	require.NoError(t, localEnd.CloseWrite())
-
-	// 3. Wait to make the race window explicit.
-	time.Sleep(halfCloseWait)
-
-	// 4. Read the echo.
-	got := make([]byte, len(payload))
-	_, err = io.ReadFull(localEnd, got)
-	require.NoError(t, err, "read after half-close failed: pipe was torn down too early")
-	require.Equal(t, payload, string(got))
-
-	// Drain ServeStream.
-	_ = localEnd.Close()
-	_ = remoteEnd.Close()
-
-	select {
-	case err := <-serveErrCh:
-		if err != nil && err != io.EOF && !errors.Is(err, io.ErrClosedPipe) {
-			require.NoError(t, err)
-		}
-	case <-ctx.Done():
-		t.Fatal("ServeStream did not return in time")
-	}
-}
-
-func websocketServer() *httptest.Server {
-	upgrader := gws.Upgrader{
-		ReadBufferSize:  1024,
-		WriteBufferSize: 1024,
-		CheckOrigin:     func(*http.Request) bool { return true },
-	}
-
-	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		conn, err := upgrader.Upgrade(w, r, nil)
-		if err != nil {
-			return
-		}
-		defer func() { _ = conn.Close() }()
-
-		_, msg, err := conn.ReadMessage()
-		if err != nil {
-			return
-		}
-
-		if err := conn.WriteMessage(gws.BinaryMessage, msg); err != nil {
-			return
-		}
-	}))
-}

connection/connection_test.go

@@ -151,7 +151,7 @@ func wsEchoEndpoint(w ResponseWriter, r *http.Request) error {
 	}()
 
 	originConn := &echoPipe{reader: readPipe, writer: writePipe}
-	stream.Pipe(wsConn, originConn, 0, &log)
+	stream.Pipe(wsConn, originConn, &log)
 	cancel()
 	wsConn.Close()
 	return nil
@@ -190,7 +190,7 @@ func wsFlakyEndpoint(w ResponseWriter, r *http.Request) error {
 	rInt, _ := rand.Int(rand.Reader, big.NewInt(50))
 	closedAfter := time.Millisecond * time.Duration(rInt.Int64())
 	originConn := &flakyConn{closeAt: time.Now().Add(closedAfter)}
-	stream.Pipe(wsConn, originConn, 0, &log)
+	stream.Pipe(wsConn, originConn, &log)
 	cancel()
 	wsConn.Close()
 	return nil

ingress/origin_connection.go

@@ -25,9 +25,9 @@ type OriginConnection interface {
 type streamHandlerFunc func(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger)
 
 // DefaultStreamHandler is an implementation of streamHandlerFunc that
-// performs a two-way io.Copy between originConn and remoteConn.
+// performs a two way io.Copy between originConn and remoteConn.
 func DefaultStreamHandler(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger) {
-	stream.Pipe(originConn, remoteConn, stream.DefaultTimeoutAfterFirstClose, log)
+	stream.Pipe(originConn, remoteConn, log)
 }
 
 // tcpConnection is an OriginConnection that directly streams to raw TCP.
@@ -38,7 +38,7 @@ type tcpConnection struct {
 }
 
 func (tc *tcpConnection) Stream(_ context.Context, tunnelConn io.ReadWriter, _ *zerolog.Logger) {
-	stream.Pipe(tunnelConn, tc, stream.DefaultTimeoutAfterFirstClose, tc.logger)
+	stream.Pipe(tunnelConn, tc, tc.logger)
 }
 
 func (tc *tcpConnection) Write(b []byte) (int, error) {

ingress/origin_connection_test.go

@@ -38,7 +38,6 @@ func TestStreamTCPConnection(t *testing.T) {
 	tcpConn := tcpConnection{
 		Conn:         cfdConn,
 		writeTimeout: 30 * time.Second,
-		logger:       TestLogger,
 	}
 
 	eyeballConn, edgeConn := net.Pipe()
@@ -158,7 +157,7 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) {
 			require.NoError(t, err)
 			defer func() { _ = wsForwarderInConn.Close() }()
 
-			stream.Pipe(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, stream.DefaultTimeoutAfterFirstClose, TestLogger)
+			stream.Pipe(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, TestLogger)
 			return nil
 		})
 

ingress/origin_service.go

@@ -99,6 +99,7 @@ type rawTCPService struct {
 	name         string
 	dialer       net.Dialer
 	writeTimeout time.Duration
+	logger       *zerolog.Logger
 }
 
 func (o *rawTCPService) String() string {
@@ -232,12 +233,10 @@ func (o *helloWorld) start(
 	if err != nil {
 		return errors.Wrap(err, "Cannot start Hello World Server")
 	}
-	go func() {
+	go hello.StartHelloWorldServer(log, helloListener, shutdownC)
-		_ = hello.StartHelloWorldServer(log, helloListener, shutdownC)
-	}()
 	o.server = helloListener
 
-	o.url = &url.URL{
+	o.httpService.url = &url.URL{
 		Scheme: "https",
 		Host:   o.server.Addr().String(),
 	}
@@ -357,7 +356,7 @@ func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerol
 		IdleConnTimeout:       cfg.KeepAliveTimeout.Duration,
 		TLSHandshakeTimeout:   cfg.TLSTimeout.Duration,
 		ExpectContinueTimeout: 1 * time.Second,
-		TLSClientConfig:       &tls.Config{RootCAs: originCertPool, InsecureSkipVerify: cfg.NoTLSVerify}, //nolint: gosec
+		TLSClientConfig:       &tls.Config{RootCAs: originCertPool, InsecureSkipVerify: cfg.NoTLSVerify},
 		ForceAttemptHTTP2:     cfg.Http2Origin,
 	}
 	if _, isHelloWorld := service.(*helloWorld); !isHelloWorld && cfg.OriginServerName != "" {
@@ -375,6 +374,7 @@ func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerol
 	// DialContext depends on which kind of origin is being used.
 	dialContext := dialer.DialContext
 	switch service := service.(type) {
+
 	// If this origin is a unix socket, enforce network type "unix".
 	case *unixSocketPath:
 		httpTransport.DialContext = func(ctx context.Context, _, _ string) (net.Conn, error) {

proxy/proxy.go

@@ -256,7 +256,7 @@ func (p *Proxy) proxyHTTPRequest(
 			reader: tr.Body,
 		}
 
-		stream.Pipe(eyeballStream, rwc, time.Second*0, logger)
+		stream.Pipe(eyeballStream, rwc, logger)
 		return nil
 	}
 
@@ -311,7 +311,7 @@ func (p *Proxy) proxyStream(
 
 // proxyTCPStream proxies private network type TCP connections as a stream towards an available origin.
 //
-// This is different from proxyStream because it's not leveraged ingress rule services and uses the
+// This is different than proxyStream because it's not leveraged ingress rule services and uses the
 // originDialer from OriginDialerService.
 func (p *Proxy) proxyTCPStream(
 	tr *tracing.TracedContext,
@@ -344,8 +344,7 @@ func (p *Proxy) proxyTCPStream(
 	connectLatency.Observe(float64(time.Since(start).Milliseconds()))
 	logger.Debug().Msg("proxy stream acknowledged")
 
-	stream.Pipe(tunnelConn, originConn, stream.DefaultTimeoutAfterFirstClose, logger)
+	stream.Pipe(tunnelConn, originConn, logger)
-
 	return nil
 }
 

proxy/proxy_test.go

@@ -223,9 +223,9 @@ func testProxyWebsocket(proxy connection.OriginProxy) func(t *testing.T) {
 			}
 			if ctx.Err() == context.DeadlineExceeded {
 				t.Errorf("Test timed out")
-				_ = readPipe.Close()
+				readPipe.Close()
-				_ = writePipe.Close()
+				writePipe.Close()
-				_ = responseWriter.Close()
+				responseWriter.Close()
 			}
 			return nil
 		})
@@ -647,7 +647,7 @@ func TestConnections(t *testing.T) {
 				ingressServiceScheme: "tcp://",
 				originService: func(t *testing.T, ln net.Listener) {
 					// closing the listener created by the test.
-					_ = ln.Close()
+					ln.Close()
 				},
 				eyeballResponseWriter: newTCPRespWriter(replayer),
 				eyeballRequestBody:    newTCPRequestBody([]byte("test2")),
@@ -756,8 +756,6 @@ func newTCPRequestBody(data []byte) *requestBody {
 	pr, pw := io.Pipe()
 	go func() {
 		_, _ = pw.Write(data)
-		// Close the write side once the payload has been sent.
-		_ = pw.Close()
 	}()
 	return &requestBody{
 		pr: pr,
@@ -803,8 +801,8 @@ func (p *pipedRequestBody) roundtrip(addr string) []byte {
 	if err != nil {
 		panic(err)
 	}
-	defer func() { _ = conn.Close() }()
+	defer conn.Close()
-	defer func() { _ = resp.Body.Close() }()
+	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusSwitchingProtocols {
 		panic(fmt.Errorf("resp returned status code: %d", resp.StatusCode))
@@ -951,7 +949,7 @@ func runEchoTCPService(t *testing.T, l net.Listener) {
 			if err != nil {
 				panic(err)
 			}
-			defer func() { _ = conn.Close() }()
+			defer conn.Close()
 
 			for {
 				buf := make([]byte, 1024)
@@ -989,7 +987,7 @@ func runEchoWSService(t *testing.T, l net.Listener) {
 			t.Log(err)
 			return
 		}
-		defer func() { _ = conn.Close() }()
+		defer conn.Close()
 
 		for {
 			messageType, p, err := conn.ReadMessage()

stream/stream.go

@@ -2,7 +2,6 @@ package stream
 
 import (
 	"encoding/hex"
-	"errors"
 	"fmt"
 	"io"
 	"runtime/debug"
@@ -10,17 +9,12 @@ import (
 	"time"
 
 	"github.com/getsentry/sentry-go"
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 
 	"github.com/cloudflare/cloudflared/cfio"
 )
 
-// DefaultTimeoutAfterFirstClose controls the upper bound of how long we wait for the second stream to finish.
-// We use bidirectional streams for our communication. Since the read and write sides can be closed independently,
-// we must have a way to close the second stream once the first one finishes. We don't want to wait indefinitely,
-// since we want to prevent misbehaving clients from blocking cloudflared.
-const DefaultTimeoutAfterFirstClose = time.Second * 10
-
 type Stream interface {
 	Reader
 	WriterCloser
@@ -43,7 +37,7 @@ type nopCloseWriterAdapter struct {
 	io.ReadWriter
 }
 
-func noopCloseWriter(stream io.ReadWriter) *nopCloseWriterAdapter {
+func NopCloseWriterAdapter(stream io.ReadWriter) *nopCloseWriterAdapter {
 	return &nopCloseWriterAdapter{stream}
 }
 
@@ -78,7 +72,7 @@ func (s *bidirectionalStreamStatus) wait(maxWaitForSecondStream time.Duration) e
 
 		select {
 		case <-timer.C:
-			return fmt.Errorf("timeout waiting for second stream to finish %s", maxWaitForSecondStream)
+			return fmt.Errorf("timeout waiting for second stream to finish")
 		case <-s.doneChan:
 			return nil
 		}
@@ -91,19 +85,15 @@ func (s *bidirectionalStreamStatus) isAnyDone() bool {
 }
 
 // Pipe copies copy data to & from provided io.ReadWriters.
-func Pipe(tunnelConn, originConn io.ReadWriter, timeoutAfterFirstClose time.Duration, log *zerolog.Logger) {
+func Pipe(tunnelConn, originConn io.ReadWriter, log *zerolog.Logger) {
-	if err := PipeBidirectional(noopCloseWriter(tunnelConn), noopCloseWriter(originConn), timeoutAfterFirstClose, log); err != nil {
+	_ = PipeBidirectional(NopCloseWriterAdapter(tunnelConn), NopCloseWriterAdapter(originConn), 0, log)
-		log.Warn().Err(err).Msg("Failed to pipe bidirectional stream")
-	}
 }
 
-// PipeBidirectional copies data between two unidirectional streams. It is a special case of Pipe that accepts streams
+// PipeBidirectional copies data to two unidirectional streams. It is a special case of Pipe where it receives a concept that allows for Read and Write side to be closed independently.
-// whose read and write sides can be closed independently. The main difference is that when piping data from a reader
+// The main difference is that when piping data from a reader to a writer, if EOF is read, then this implementation propagates the EOF signal to the destination/writer by closing the write side of the
-// to a writer, if EOF is read, this implementation propagates the EOF signal to the destination by closing the write
+// Bidirectional Stream.
-// side of the bidirectional stream.
+// Finally, depending on once EOF is ready from one of the provided streams, the other direction of streaming data will have a configured time period to also finish, otherwise,
-// Finally, once EOF is received from one of the provided streams, the other direction has a configured grace period to
+// the method will return immediately  with a timeout error. It is however, the responsibility of the caller to close the associated streams in both ends in order to free all the resources/go-routines.
-// finish; otherwise, the method returns a timeout error. It is, however, the responsibility of the caller to close
-// the associated streams at both ends in order to free all resources and goroutines.
 func PipeBidirectional(downstream, upstream Stream, maxWaitForSecondStream time.Duration, log *zerolog.Logger) error {
 	status := newBiStreamStatus()
 
@@ -111,7 +101,7 @@ func PipeBidirectional(downstream, upstream Stream, maxWaitForSecondStream time.
 	go unidirectionalStream(upstream, downstream, "downstream->upstream", status, log)
 
 	if err := status.wait(maxWaitForSecondStream); err != nil {
-		return fmt.Errorf("unable to wait for both streams while proxying: %w", err)
+		return errors.Wrap(err, "unable to wait for both streams while proxying")
 	}
 
 	return nil

stream/stream_test.go

@@ -30,8 +30,8 @@ func TestPipeBidirectionalFinishOneSideTimeout(t *testing.T) {
 
 func TestPipeBidirectionalClosingWriteBothSidesAlsoExists(t *testing.T) {
 	fun := func(upstream, downstream *mockedStream) {
-		_ = downstream.CloseWrite()
+		downstream.CloseWrite()
-		_ = upstream.CloseWrite()
+		upstream.CloseWrite()
 
 		downstream.writeToReader("abc")
 		upstream.writeToReader("abc")
@@ -42,7 +42,7 @@ func TestPipeBidirectionalClosingWriteBothSidesAlsoExists(t *testing.T) {
 
 func TestPipeBidirectionalClosingWriteSingleSideAlsoExists(t *testing.T) {
 	fun := func(upstream, downstream *mockedStream) {
-		_ = downstream.CloseWrite()
+		downstream.CloseWrite()
 
 		downstream.writeToReader("abc")
 		upstream.writeToReader("abc")
@@ -51,46 +51,6 @@ func TestPipeBidirectionalClosingWriteSingleSideAlsoExists(t *testing.T) {
 	testPipeBidirectionalUnblocking(t, fun, time.Millisecond*200, true)
 }
 
-// TestPipeBidirectionalReturnsWhenBothSidesFinish verifies that
-// PipeBidirectional returns as soon as both stream directions finish, without
-// waiting for the full timeout grace period to expire. This guards against a
-// regression where the second-stream wait would block for the whole timeout
-// even when the result is already available.
-func TestPipeBidirectionalReturnsWhenBothSidesFinish(t *testing.T) {
-	t.Parallel()
-
-	const (
-		timeout     = time.Second * 5
-		maxWallTime = time.Millisecond * 500
-	)
-
-	logger := zerolog.Nop()
-	downstream := newMockedStream()
-	upstream := newMockedStream()
-
-	resultCh := make(chan error, 1)
-	go func() {
-		resultCh <- PipeBidirectional(downstream, upstream, timeout, &logger)
-	}()
-
-	// Close both reader sides so both stream directions reach EOF promptly.
-	downstream.closeReader()
-	upstream.closeReader()
-
-	start := time.Now()
-	select {
-	case err := <-resultCh:
-		elapsed := time.Since(start)
-		require.NoError(t, err)
-		require.Less(t, elapsed, maxWallTime,
-			"PipeBidirectional should return as soon as both streams finish, not after the full %s timeout (took %s)",
-			timeout, elapsed,
-		)
-	case <-time.After(timeout):
-		require.Fail(t, "PipeBidirectional did not return before the timeout expired")
-	}
-}
-
 func testPipeBidirectionalUnblocking(t *testing.T, afterFun func(*mockedStream, *mockedStream), timeout time.Duration, expectTimeout bool) {
 	logger := zerolog.Nop()
 
@@ -107,9 +67,9 @@ func testPipeBidirectionalUnblocking(t *testing.T, afterFun func(*mockedStream,
 	select {
 	case err := <-resultCh:
 		if expectTimeout {
-			require.Error(t, err)
+			require.NotNil(t, err)
 		} else {
-			require.NoError(t, err)
+			require.Nil(t, err)
 		}
 
 	case <-time.After(timeout * 2):