TUN-5118: Quic connection now detects duplicate connections similar to http2

Author: sudarshan-reddy

connection/quic.go

@@ -57,7 +57,11 @@ func NewQUICConnection(
 		return nil, errors.Wrap(err, "failed to open a registration stream")
 	}
 
-	go controlStreamHandler.ServeControlStream(ctx, registrationStream, connOptions)
+	err = controlStreamHandler.ServeControlStream(ctx, registrationStream, connOptions)
+	if err != nil {
+		// Not wrapping error here to be consistent with the http2 message.
+		return nil, err
+	}
 
 	return &QUICConnection{
 		session:   session,
@@ -74,6 +78,10 @@ func (q *QUICConnection) Serve(ctx context.Context) error {
 	for {
 		stream, err := q.session.AcceptStream(ctx)
 		if err != nil {
+			// context.Canceled is usually a user ctrl+c. We don't want to log an error here as it's intentional.
+			if errors.Is(err, context.Canceled) {
+				return nil
+			}
 			return errors.Wrap(err, "failed to accept QUIC stream")
 		}
 		go func() {

connection/quic_test.go

@@ -16,7 +16,6 @@ import (
 	"os"
 	"sync"
 	"testing"
-	"time"
 
 	"github.com/gobwas/ws/wsutil"
 	"github.com/lucas-clemente/quic-go"
@@ -27,6 +26,7 @@ import (
 
 	quicpogs "github.com/cloudflare/cloudflared/quic"
 	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
 // TestQUICServer tests if a quic server accepts and responds to a quic client with the acceptance protocol.
@@ -158,21 +158,7 @@ func TestQUICServer(t *testing.T) {
 				)
 			}()
 
-			rpcClientFactory := mockRPCClientFactory{
-				registered:   make(chan struct{}),
-				unregistered: make(chan struct{}),
-			}
-
-			obs := NewObserver(&log, &log, false)
-			controlStream := NewControlStream(
-				obs,
-				mockConnectedFuse{},
-				&NamedTunnelConfig{},
-				1,
-				rpcClientFactory.newMockRPCClient,
-				nil,
-				1*time.Second,
-			)
+			controlStream := fakeControlStream{}
 
 			qC, err := NewQUICConnection(
 				ctx,
@@ -188,17 +174,20 @@ func TestQUICServer(t *testing.T) {
 			go qC.Serve(ctx)
 
 			wg.Wait()
-
-			select {
-			case <-rpcClientFactory.registered:
-				break //ok
-			case <-time.Tick(time.Second):
-				t.Fatal("timeout out waiting for registration")
-			}
 			cancel()
 		})
 	}
+}
 
+type fakeControlStream struct {
+	ControlStreamHandler
+}
+
+func (fakeControlStream) ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions) error {
+	return nil
+}
+func (fakeControlStream) IsStopped() bool {
+	return true
 }
 
 func quicServer(

origin/tunnel.go

@@ -30,6 +30,8 @@ const (
 	dialTimeout              = 15 * time.Second
 	FeatureSerializedHeaders = "serialized_headers"
 	FeatureQuickReconnects   = "quick_reconnects"
+	quicHandshakeIdleTimeout = 5 * time.Second
+	quicMaxIdleTimeout       = 15 * time.Second
 )
 
 type rpcName string
@@ -271,12 +273,73 @@ func ServeTunnel(
 	}()
 
 	defer config.Observer.SendDisconnect(connIndex)
+	err, recoverable = serveTunnel(
+		ctx,
+		connLog,
+		credentialManager,
+		config,
+		addr,
+		connIndex,
+		fuse,
+		backoff,
+		cloudflaredUUID,
+		reconnectCh,
+		protocol,
+		gracefulShutdownC,
+	)
+
+	if err != nil {
+		switch err := err.(type) {
+		case connection.DupConnRegisterTunnelError:
+			connLog.Err(err).Msg("Unable to establish connection.")
+			// don't retry this connection anymore, let supervisor pick a new address
+			return err, false
+		case connection.ServerRegisterTunnelError:
+			connLog.Err(err).Msg("Register tunnel error from server side")
+			// Don't send registration error return from server to Sentry. They are
+			// logged on server side
+			if incidents := config.IncidentLookup.ActiveIncidents(); len(incidents) > 0 {
+				connLog.Error().Msg(activeIncidentsMsg(incidents))
+			}
+			return err.Cause, !err.Permanent
+		case ReconnectSignal:
+			connLog.Info().
+				Uint8(connection.LogFieldConnIndex, connIndex).
+				Msgf("Restarting connection due to reconnect signal in %s", err.Delay)
+			err.DelayBeforeReconnect()
+			return err, true
+		default:
+			if err == context.Canceled {
+				connLog.Debug().Err(err).Msgf("Serve tunnel error")
+				return err, false
+			}
+			connLog.Err(err).Msgf("Serve tunnel error")
+			_, permanent := err.(unrecoverableError)
+			return err, !permanent
+		}
+	}
+	return nil, false
+}
+
+func serveTunnel(
+	ctx context.Context,
+	connLog *zerolog.Logger,
+	credentialManager *reconnectCredentialManager,
+	config *TunnelConfig,
+	addr *allregions.EdgeAddr,
+	connIndex uint8,
+	fuse *h2mux.BooleanFuse,
+	backoff *protocolFallback,
+	cloudflaredUUID uuid.UUID,
+	reconnectCh chan ReconnectSignal,
+	protocol connection.Protocol,
+	gracefulShutdownC <-chan struct{},
+) (err error, recoverable bool) {
 
 	connectedFuse := &connectedFuse{
 		fuse:    fuse,
 		backoff: backoff,
 	}
-
 	controlStream := connection.NewControlStream(
 		config.Observer,
 		connectedFuse,
@@ -287,7 +350,8 @@ func ServeTunnel(
 		config.ConnectionConfig.GracePeriod,
 	)
 
-	if protocol == connection.QUIC {
+	switch protocol {
+	case connection.QUIC:
 		connOptions := config.ConnectionOptions(addr.UDP.String(), uint8(backoff.Retries()))
 		return ServeQUIC(ctx,
 			addr.UDP,
@@ -297,17 +361,16 @@ func ServeTunnel(
 			connectedFuse,
 			reconnectCh,
 			gracefulShutdownC)
-	}
 
-	edgeConn, err := edgediscovery.DialEdge(ctx, dialTimeout, config.EdgeTLSConfigs[protocol], addr.TCP)
-	if err != nil {
-		connLog.Err(err).Msg("Unable to establish connection with Cloudflare edge")
-		return err, true
-	}
+	case connection.HTTP2:
+		edgeConn, err := edgediscovery.DialEdge(ctx, dialTimeout, config.EdgeTLSConfigs[protocol], addr.TCP)
+		if err != nil {
+			connLog.Err(err).Msg("Unable to establish connection with Cloudflare edge")
+			return err, true
+		}
 
-	if protocol == connection.HTTP2 {
 		connOptions := config.ConnectionOptions(edgeConn.LocalAddr().String(), uint8(backoff.Retries()))
-		err = ServeHTTP2(
+		if err := ServeHTTP2(
 			ctx,
 			connLog,
 			config,
@@ -317,9 +380,18 @@ func ServeTunnel(
 			connIndex,
 			gracefulShutdownC,
 			reconnectCh,
-		)
-	} else {
-		err = ServeH2mux(
+		); err != nil {
+			return err, false
+		}
+
+	default:
+		edgeConn, err := edgediscovery.DialEdge(ctx, dialTimeout, config.EdgeTLSConfigs[protocol], addr.TCP)
+		if err != nil {
+			connLog.Err(err).Msg("Unable to establish connection with Cloudflare edge")
+			return err, true
+		}
+
+		if err := ServeH2mux(
 			ctx,
 			connLog,
 			credentialManager,
@@ -330,40 +402,11 @@ func ServeTunnel(
 			cloudflaredUUID,
 			reconnectCh,
 			gracefulShutdownC,
-		)
-	}
-
-	if err != nil {
-		switch err := err.(type) {
-		case connection.DupConnRegisterTunnelError:
-			connLog.Err(err).Msg("Unable to establish connection.")
-			// don't retry this connection anymore, let supervisor pick a new address
+		); err != nil {
 			return err, false
-		case connection.ServerRegisterTunnelError:
-			connLog.Err(err).Msg("Register tunnel error from server side")
-			// Don't send registration error return from server to Sentry. They are
-			// logged on server side
-			if incidents := config.IncidentLookup.ActiveIncidents(); len(incidents) > 0 {
-				connLog.Error().Msg(activeIncidentsMsg(incidents))
-			}
-			return err.Cause, !err.Permanent
-		case ReconnectSignal:
-			connLog.Info().
-				Uint8(connection.LogFieldConnIndex, connIndex).
-				Msgf("Restarting connection due to reconnect signal in %s", err.Delay)
-			err.DelayBeforeReconnect()
-			return err, true
-		default:
-			if err == context.Canceled {
-				connLog.Debug().Err(err).Msgf("Serve tunnel error")
-				return err, false
-			}
-			connLog.Err(err).Msgf("Serve tunnel error")
-			_, permanent := err.(unrecoverableError)
-			return err, !permanent
 		}
 	}
-	return nil, false
+	return
 }
 
 type unrecoverableError struct {
@@ -472,7 +515,8 @@ func ServeQUIC(
 ) (err error, recoverable bool) {
 	tlsConfig := config.EdgeTLSConfigs[connection.QUIC]
 	quicConfig := &quic.Config{
-		HandshakeIdleTimeout: time.Second * 10,
+		HandshakeIdleTimeout: quicHandshakeIdleTimeout,
+		MaxIdleTimeout:       quicMaxIdleTimeout,
 		KeepAlive:            true,
 	}
 	for {
@@ -511,7 +555,6 @@ func ServeQUIC(
 			if err == nil {
 				return nil, false
 			}
-			config.Log.Info().Msg("Reconnecting with the same udp conn")
 		}
 	}
 }