Release 2024.11.1

## Summary
Some description...

.github/workflows/semgrep.yml

@@ -1,4 +1,3 @@
-
 on:
   pull_request: {}
   workflow_dispatch: {}
@@ -12,14 +11,14 @@ name: Semgrep config
 jobs:
   semgrep:
     name: semgrep/ci
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     env:
       SEMGREP_APP_TOKEN: ${{ secrets.SEMGREP_APP_TOKEN }}
       SEMGREP_URL: https://cloudflare.semgrep.dev
       SEMGREP_APP_URL: https://cloudflare.semgrep.dev
       SEMGREP_VERSION_CHECK_URL: https://cloudflare.semgrep.dev/api/check-version
     container:
-      image: returntocorp/semgrep
+      image: semgrep/semgrep
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - run: semgrep ci

Makefile

@@ -165,6 +165,18 @@ cover:
 	# Generate the HTML report that can be viewed from the browser in CI.
 	$Q go tool cover -html ".cover/c.out" -o .cover/all.html
 
+.PHONY: fuzz
+fuzz:
+	@go test -fuzz=FuzzIPDecoder -fuzztime=600s ./packet
+	@go test -fuzz=FuzzICMPDecoder -fuzztime=600s ./packet
+	@go test -fuzz=FuzzSessionWrite -fuzztime=600s ./quic/v3
+	@go test -fuzz=FuzzSessionServe -fuzztime=600s ./quic/v3
+	@go test -fuzz=FuzzRegistrationDatagram -fuzztime=600s ./quic/v3
+	@go test -fuzz=FuzzPayloadDatagram -fuzztime=600s ./quic/v3
+	@go test -fuzz=FuzzRegistrationResponseDatagram -fuzztime=600s ./quic/v3
+	@go test -fuzz=FuzzNewIdentity -fuzztime=600s ./tracing
+	@go test -fuzz=FuzzNewAccessValidator -fuzztime=600s ./validation
+
 .PHONY: install-go
 install-go:
 	rm -rf ${CF_GO_PATH}

README.md

@@ -49,7 +49,7 @@ Once installed, you can authenticate `cloudflared` into your Cloudflare account
 
 ## TryCloudflare
 
-Want to test Cloudflare Tunnel before adding a website to Cloudflare? You can do so with TryCloudflare using the documentation [available here](https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/run-tunnel/trycloudflare).
+Want to test Cloudflare Tunnel before adding a website to Cloudflare? You can do so with TryCloudflare using the documentation [available here](https://developers.cloudflare.com/cloudflare-one/connections/connect-networks/do-more-with-tunnels/trycloudflare/).
 
 ## Deprecated versions
 

RELEASE_NOTES

@@ -1,3 +1,13 @@
+2024.11.1
+- 2024-11-18 Add cloudflared tunnel ready command
+- 2024-11-14 Make metrics a requirement for tunnel ready command
+- 2024-11-12 TUN-8701:  Simplify flow registration logs for datagram v3
+- 2024-11-11 add: new go-fuzz targets
+- 2024-11-07 TUN-8701: Add metrics and adjust logs for datagram v3
+- 2024-11-06 TUN-8709: Add session migration for datagram v3
+- 2024-11-04 Fixed 404 in README.md to TryCloudflare
+- 2024-09-24 Update semgrep.yml
+
 2024.11.0
 - 2024-11-05 VULN-66059: remove ssh server tests
 - 2024-11-04 TUN-8700: Add datagram v3 muxer

cmd/cloudflared/tunnel/cmd.go

@@ -139,6 +139,7 @@ func Commands() []*cli.Command {
 		buildVirtualNetworkSubcommand(false),
 		buildRunCommand(),
 		buildListCommand(),
+		buildReadyCommand(),
 		buildInfoCommand(),
 		buildIngressSubcommand(),
 		buildDeleteCommand(),

cmd/cloudflared/tunnel/subcommands.go

@@ -5,6 +5,8 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -397,6 +399,39 @@ func fmtConnections(connections []cfapi.Connection, showRecentlyDisconnected boo
 	return strings.Join(output, ", ")
 }
 
+func buildReadyCommand() *cli.Command {
+	return &cli.Command{
+		Name:               "ready",
+		Action:             cliutil.ConfiguredAction(readyCommand),
+		Usage:              "Call /ready endpoint and return proper exit code",
+		UsageText:          "cloudflared tunnel [tunnel command options] ready [subcommand options]",
+		Description:        "cloudflared tunnel ready will return proper exit code based on the /ready endpoint",
+		Flags:              []cli.Flag{},
+		CustomHelpTemplate: commandHelpTemplate(),
+	}
+}
+
+func readyCommand(c *cli.Context) error {
+	metricsOpts := c.String("metrics")
+	if !c.IsSet("metrics") {
+		return fmt.Errorf("--metrics has to be provided")
+	}
+
+	requestURL := fmt.Sprintf("http://%s/ready", metricsOpts)
+	res, err := http.Get(requestURL)
+	if err != nil {
+		return err
+	}
+	if res.StatusCode != 200 {
+		body, err := io.ReadAll(res.Body)
+		if err != nil {
+			return err
+		}
+		return fmt.Errorf("http://%s/ready endpoint returned status code %d\n%s", metricsOpts, res.StatusCode, body)
+	}
+	return nil
+}
+
 func buildInfoCommand() *cli.Command {
 	return &cli.Command{
 		Name:        "info",

connection/quic_datagram_v3.go

@@ -10,6 +10,7 @@ import (
 	"github.com/quic-go/quic-go"
 	"github.com/rs/zerolog"
 
+	"github.com/cloudflare/cloudflared/management"
 	cfdquic "github.com/cloudflare/cloudflared/quic/v3"
 	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
@@ -24,9 +25,16 @@ type datagramV3Connection struct {
 func NewDatagramV3Connection(ctx context.Context,
 	conn quic.Connection,
 	sessionManager cfdquic.SessionManager,
+	index uint8,
+	metrics cfdquic.Metrics,
 	logger *zerolog.Logger,
 ) DatagramSessionHandler {
-	datagramMuxer := cfdquic.NewDatagramConn(conn, sessionManager, logger)
+	log := logger.
+		With().
+		Int(management.EventTypeKey, int(management.UDP)).
+		Uint8(LogFieldConnIndex, index).
+		Logger()
+	datagramMuxer := cfdquic.NewDatagramConn(conn, sessionManager, index, metrics, &log)
 
 	return &datagramV3Connection{
 		conn,

ingress/origin_udp_proxy.go

@@ -39,7 +39,7 @@ func DialUDPAddrPort(dest netip.AddrPort) (*net.UDPConn, error) {
 	// address as context.
 	udpConn, err := net.DialUDP("udp", nil, addr)
 	if err != nil {
-		return nil, fmt.Errorf("unable to create UDP proxy to origin (%v:%v): %w", dest.Addr(), dest.Port(), err)
+		return nil, fmt.Errorf("unable to dial udp to origin %s: %w", dest, err)
 	}
 
 	return udpConn, nil

packet/decoder_test.go

@@ -254,3 +254,18 @@ func (u *UDP) EncodeLayers() ([]gopacket.SerializableLayer, error) {
 	udpLayer.SetNetworkLayerForChecksum(ipLayers[0].(gopacket.NetworkLayer))
 	return append(ipLayers, &udpLayer), nil
 }
+
+func FuzzIPDecoder(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		ipDecoder := NewIPDecoder()
+		ipDecoder.Decode(RawPacket{Data: data})
+
+	})
+}
+
+func FuzzICMPDecoder(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		icmpDecoder := NewICMPDecoder()
+		icmpDecoder.Decode(RawPacket{Data: data})
+	})
+}

quic/v3/datagram.go

@@ -284,8 +284,6 @@ const (
 	ResponseDestinationUnreachable SessionRegistrationResp = 0x01
 	// Session registration was unable to bind to a local UDP socket.
 	ResponseUnableToBindSocket SessionRegistrationResp = 0x02
-	// Session registration is already bound to another connection.
-	ResponseSessionAlreadyConnected SessionRegistrationResp = 0x03
 	// Session registration failed with an unexpected error but provided a message.
 	ResponseErrorWithMsg SessionRegistrationResp = 0xff
 )

quic/v3/datagram_test.go

@@ -350,3 +350,30 @@ func compareRegistrationDatagrams(t *testing.T, l *v3.UDPSessionRegistrationData
 		l.IdleDurationHint == r.IdleDurationHint &&
 		l.Traced == r.Traced
 }
+
+func FuzzRegistrationDatagram(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		unmarshaled := v3.UDPSessionRegistrationDatagram{}
+		err := unmarshaled.UnmarshalBinary(data)
+		if err == nil {
+			_, _ = unmarshaled.MarshalBinary()
+		}
+	})
+}
+
+func FuzzPayloadDatagram(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		unmarshaled := v3.UDPSessionPayloadDatagram{}
+		_ = unmarshaled.UnmarshalBinary(data)
+	})
+}
+
+func FuzzRegistrationResponseDatagram(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		unmarshaled := v3.UDPSessionRegistrationResponseDatagram{}
+		err := unmarshaled.UnmarshalBinary(data)
+		if err == nil {
+			_, _ = unmarshaled.MarshalBinary()
+		}
+	})
+}

quic/v3/manager.go

@@ -7,20 +7,22 @@ import (
 	"sync"
 
 	"github.com/rs/zerolog"
-
-	"github.com/cloudflare/cloudflared/ingress"
 )
 
 var (
-	ErrSessionNotFound         = errors.New("session not found")
-	ErrSessionBoundToOtherConn = errors.New("session is in use by another connection")
+	// ErrSessionNotFound indicates that a session has not been registered yet for the request id.
+	ErrSessionNotFound = errors.New("flow not found")
+	// ErrSessionBoundToOtherConn is returned when a registration already exists for a different connection.
+	ErrSessionBoundToOtherConn = errors.New("flow is in use by another connection")
+	// ErrSessionAlreadyRegistered is returned when a registration already exists for this connection.
+	ErrSessionAlreadyRegistered = errors.New("flow is already registered for this connection")
 )
 
 type SessionManager interface {
 	// RegisterSession will register a new session if it does not already exist for the request ID.
 	// During new session creation, the session will also bind the UDP socket for the origin.
 	// If the session exists for a different connection, it will return [ErrSessionBoundToOtherConn].
-	RegisterSession(request *UDPSessionRegistrationDatagram, conn DatagramWriter) (Session, error)
+	RegisterSession(request *UDPSessionRegistrationDatagram, conn DatagramConn) (Session, error)
 	// GetSession returns an active session if available for the provided connection.
 	// If the session does not exist, it will return [ErrSessionNotFound]. If the session exists for a different
 	// connection, it will return [ErrSessionBoundToOtherConn].
@@ -33,33 +35,47 @@ type SessionManager interface {
 type DialUDP func(dest netip.AddrPort) (*net.UDPConn, error)
 
 type sessionManager struct {
-	sessions map[RequestID]Session
-	mutex    sync.RWMutex
-	log      *zerolog.Logger
+	sessions     map[RequestID]Session
+	mutex        sync.RWMutex
+	originDialer DialUDP
+	metrics      Metrics
+	log          *zerolog.Logger
 }
 
-func NewSessionManager(log *zerolog.Logger, originDialer DialUDP) SessionManager {
+func NewSessionManager(metrics Metrics, log *zerolog.Logger, originDialer DialUDP) SessionManager {
 	return &sessionManager{
-		sessions: make(map[RequestID]Session),
-		log:      log,
+		sessions:     make(map[RequestID]Session),
+		originDialer: originDialer,
+		metrics:      metrics,
+		log:          log,
 	}
 }
 
-func (s *sessionManager) RegisterSession(request *UDPSessionRegistrationDatagram, conn DatagramWriter) (Session, error) {
+func (s *sessionManager) RegisterSession(request *UDPSessionRegistrationDatagram, conn DatagramConn) (Session, error) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 	// Check to make sure session doesn't already exist for requestID
-	_, exists := s.sessions[request.RequestID]
-	if exists {
+	if session, exists := s.sessions[request.RequestID]; exists {
+		if conn.ID() == session.ConnectionID() {
+			return nil, ErrSessionAlreadyRegistered
+		}
 		return nil, ErrSessionBoundToOtherConn
 	}
 	// Attempt to bind the UDP socket for the new session
-	origin, err := ingress.DialUDPAddrPort(request.Dest)
+	origin, err := s.originDialer(request.Dest)
 	if err != nil {
 		return nil, err
 	}
 	// Create and insert the new session in the map
-	session := NewSession(request.RequestID, request.IdleDurationHint, origin, conn, s.log)
+	session := NewSession(
+		request.RequestID,
+		request.IdleDurationHint,
+		origin,
+		origin.RemoteAddr(),
+		origin.LocalAddr(),
+		conn,
+		s.metrics,
+		s.log)
 	s.sessions[request.RequestID] = session
 	return session, nil
 }

quic/v3/manager_test.go

@@ -15,7 +15,7 @@ import (
 
 func TestRegisterSession(t *testing.T) {
 	log := zerolog.Nop()
-	manager := v3.NewSessionManager(&log, ingress.DialUDPAddrPort)
+	manager := v3.NewSessionManager(&noopMetrics{}, &log, ingress.DialUDPAddrPort)
 
 	request := v3.UDPSessionRegistrationDatagram{
 		RequestID:        testRequestID,
@@ -34,8 +34,14 @@ func TestRegisterSession(t *testing.T) {
 
 	// We shouldn't be able to register another session with the same request id
 	_, err = manager.RegisterSession(&request, &noopEyeball{})
+	if !errors.Is(err, v3.ErrSessionAlreadyRegistered) {
+		t.Fatalf("session is already registered for this connection: %v", err)
+	}
+
+	// We shouldn't be able to register another session with the same request id for a different connection
+	_, err = manager.RegisterSession(&request, &noopEyeball{connID: 1})
 	if !errors.Is(err, v3.ErrSessionBoundToOtherConn) {
-		t.Fatalf("session should not be able to be registered again: %v", err)
+		t.Fatalf("session is already registered for a separate connection: %v", err)
 	}
 
 	// Get session
@@ -65,7 +71,7 @@ func TestRegisterSession(t *testing.T) {
 
 func TestGetSession_Empty(t *testing.T) {
 	log := zerolog.Nop()
-	manager := v3.NewSessionManager(&log, ingress.DialUDPAddrPort)
+	manager := v3.NewSessionManager(&noopMetrics{}, &log, ingress.DialUDPAddrPort)
 
 	_, err := manager.GetSession(testRequestID)
 	if !errors.Is(err, v3.ErrSessionNotFound) {

quic/v3/metrics.go

@@ -0,0 +1,90 @@
+package v3
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	namespace = "cloudflared"
+	subsystem = "udp"
+)
+
+type Metrics interface {
+	IncrementFlows()
+	DecrementFlows()
+	PayloadTooLarge()
+	RetryFlowResponse()
+	MigrateFlow()
+}
+
+type metrics struct {
+	activeUDPFlows     prometheus.Gauge
+	totalUDPFlows      prometheus.Counter
+	payloadTooLarge    prometheus.Counter
+	retryFlowResponses prometheus.Counter
+	migratedFlows      prometheus.Counter
+}
+
+func (m *metrics) IncrementFlows() {
+	m.totalUDPFlows.Inc()
+	m.activeUDPFlows.Inc()
+}
+
+func (m *metrics) DecrementFlows() {
+	m.activeUDPFlows.Dec()
+}
+
+func (m *metrics) PayloadTooLarge() {
+	m.payloadTooLarge.Inc()
+}
+
+func (m *metrics) RetryFlowResponse() {
+	m.retryFlowResponses.Inc()
+}
+
+func (m *metrics) MigrateFlow() {
+	m.migratedFlows.Inc()
+}
+
+func NewMetrics(registerer prometheus.Registerer) Metrics {
+	m := &metrics{
+		activeUDPFlows: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "active_flows",
+			Help:      "Concurrent count of UDP flows that are being proxied to any origin",
+		}),
+		totalUDPFlows: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "total_flows",
+			Help:      "Total count of UDP flows that have been proxied to any origin",
+		}),
+		payloadTooLarge: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "payload_too_large",
+			Help:      "Total count of UDP flows that have had origin payloads that are too large to proxy",
+		}),
+		retryFlowResponses: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "retry_flow_responses",
+			Help:      "Total count of UDP flows that have had to send their registration response more than once",
+		}),
+		migratedFlows: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "migrated_flows",
+			Help:      "Total count of UDP flows have been migrated across local connections",
+		}),
+	}
+	registerer.MustRegister(
+		m.activeUDPFlows,
+		m.totalUDPFlows,
+		m.payloadTooLarge,
+		m.retryFlowResponses,
+		m.migratedFlows,
+	)
+	return m
+}

quic/v3/metrics_test.go

@@ -0,0 +1,9 @@
+package v3_test
+
+type noopMetrics struct{}
+
+func (noopMetrics) IncrementFlows()    {}
+func (noopMetrics) DecrementFlows()    {}
+func (noopMetrics) PayloadTooLarge()   {}
+func (noopMetrics) RetryFlowResponse() {}
+func (noopMetrics) MigrateFlow()       {}

quic/v3/muxer.go

@@ -3,6 +3,7 @@ package v3
 import (
 	"context"
 	"errors"
+	"time"
 
 	"github.com/rs/zerolog"
 )
@@ -11,6 +12,10 @@ const (
 	// Allocating a 16 channel buffer here allows for the writer to be slightly faster than the reader.
 	// This has worked previously well for datagramv2, so we will start with this as well
 	demuxChanCapacity = 16
+
+	logSrcKey      = "src"
+	logDstKey      = "dst"
+	logDurationKey = "durationMS"
 )
 
 // DatagramConn is the bridge that multiplexes writes and reads of datagrams for UDP sessions and ICMP packets to
@@ -19,6 +24,8 @@ type DatagramConn interface {
 	DatagramWriter
 	// Serve provides a server interface to process and handle incoming QUIC datagrams and demux their datagram v3 payloads.
 	Serve(context.Context) error
+	// ID indicates connection index identifier
+	ID() uint8
 }
 
 // DatagramWriter provides the Muxer interface to create proper Datagrams when sending over a connection.
@@ -41,24 +48,32 @@ type QuicConnection interface {
 
 type datagramConn struct {
 	conn           QuicConnection
+	index          uint8
 	sessionManager SessionManager
+	metrics        Metrics
 	logger         *zerolog.Logger
 
 	datagrams  chan []byte
 	readErrors chan error
 }
 
-func NewDatagramConn(conn QuicConnection, sessionManager SessionManager, logger *zerolog.Logger) DatagramConn {
+func NewDatagramConn(conn QuicConnection, sessionManager SessionManager, index uint8, metrics Metrics, logger *zerolog.Logger) DatagramConn {
 	log := logger.With().Uint8("datagramVersion", 3).Logger()
 	return &datagramConn{
 		conn:           conn,
+		index:          index,
 		sessionManager: sessionManager,
+		metrics:        metrics,
 		logger:         &log,
 		datagrams:      make(chan []byte, demuxChanCapacity),
 		readErrors:     make(chan error, 2),
 	}
 }
 
+func (c datagramConn) ID() uint8 {
+	return c.index
+}
+
 func (c *datagramConn) SendUDPSessionDatagram(datagram []byte) error {
 	return c.conn.SendDatagram(datagram)
 }
@@ -135,11 +150,12 @@ func (c *datagramConn) Serve(ctx context.Context) error {
 					c.logger.Err(err).Msgf("unable to unmarshal session registration datagram")
 					return
 				}
+				logger := c.logger.With().Str(logFlowID, reg.RequestID.String()).Logger()
 				// We bind the new session to the quic connection context instead of cloudflared context to allow for the
 				// quic connection to close and close only the sessions bound to it. Closing of cloudflared will also
 				// initiate the close of the quic connection, so we don't have to worry about the application context
 				// in the scope of a session.
-				c.handleSessionRegistrationDatagram(connCtx, reg)
+				c.handleSessionRegistrationDatagram(connCtx, reg, &logger)
 			case UDPSessionPayloadType:
 				payload := &UDPSessionPayloadDatagram{}
 				err := payload.UnmarshalBinary(datagram)
@@ -147,7 +163,8 @@ func (c *datagramConn) Serve(ctx context.Context) error {
 					c.logger.Err(err).Msgf("unable to unmarshal session payload datagram")
 					return
 				}
-				c.handleSessionPayloadDatagram(payload)
+				logger := c.logger.With().Str(logFlowID, payload.RequestID.String()).Logger()
+				c.handleSessionPayloadDatagram(payload, &logger)
 			case UDPSessionRegistrationResponseType:
 				// cloudflared should never expect to receive UDP session responses as it will not initiate new
 				// sessions towards the edge.
@@ -161,67 +178,124 @@ func (c *datagramConn) Serve(ctx context.Context) error {
 }
 
 // This method handles new registrations of a session and the serve loop for the session.
-func (c *datagramConn) handleSessionRegistrationDatagram(ctx context.Context, datagram *UDPSessionRegistrationDatagram) {
+func (c *datagramConn) handleSessionRegistrationDatagram(ctx context.Context, datagram *UDPSessionRegistrationDatagram, logger *zerolog.Logger) {
+	log := logger.With().
+		Str(logFlowID, datagram.RequestID.String()).
+		Str(logDstKey, datagram.Dest.String()).
+		Logger()
 	session, err := c.sessionManager.RegisterSession(datagram, c)
-	if err != nil {
-		c.logger.Err(err).Msgf("session registration failure")
-		c.handleSessionRegistrationFailure(datagram.RequestID, err)
+	switch err {
+	case nil:
+		// Continue as normal
+	case ErrSessionAlreadyRegistered:
+		// Session is already registered and likely the response got lost
+		c.handleSessionAlreadyRegistered(datagram.RequestID, &log)
+		return
+	case ErrSessionBoundToOtherConn:
+		// Session is already registered but to a different connection
+		c.handleSessionMigration(datagram.RequestID, &log)
+		return
+	default:
+		log.Err(err).Msgf("flow registration failure")
+		c.handleSessionRegistrationFailure(datagram.RequestID, &log)
 		return
 	}
+	log = log.With().Str(logSrcKey, session.LocalAddr().String()).Logger()
+	c.metrics.IncrementFlows()
 	// Make sure to eventually remove the session from the session manager when the session is closed
 	defer c.sessionManager.UnregisterSession(session.ID())
+	defer c.metrics.DecrementFlows()
 
 	// Respond that we are able to process the new session
 	err = c.SendUDPSessionResponse(datagram.RequestID, ResponseOk)
 	if err != nil {
-		c.logger.Err(err).Msgf("session registration failure: unable to send session registration response")
+		log.Err(err).Msgf("flow registration failure: unable to send session registration response")
 		return
 	}
 
 	// We bind the context of the session to the [quic.Connection] that initiated the session.
 	// [Session.Serve] is blocking and will continue this go routine till the end of the session lifetime.
+	start := time.Now()
 	err = session.Serve(ctx)
+	elapsedMS := time.Now().Sub(start).Milliseconds()
+	log = log.With().Int64(logDurationKey, elapsedMS).Logger()
 	if err == nil {
 		// We typically don't expect a session to close without some error response. [SessionIdleErr] is the typical
 		// expected error response.
-		c.logger.Warn().Msg("session was closed without explicit close or timeout")
+		log.Warn().Msg("flow closed: no explicit close or timeout elapsed")
 		return
 	}
 	// SessionIdleErr and SessionCloseErr are valid and successful error responses to end a session.
 	if errors.Is(err, SessionIdleErr{}) || errors.Is(err, SessionCloseErr) {
-		c.logger.Debug().Msg(err.Error())
+		log.Debug().Msgf("flow closed: %s", err.Error())
 		return
 	}
 
 	// All other errors should be reported as errors
-	c.logger.Err(err).Msgf("session was closed with an error")
+	log.Err(err).Msgf("flow closed with an error")
 }
 
-func (c *datagramConn) handleSessionRegistrationFailure(requestID RequestID, regErr error) {
-	var errResp SessionRegistrationResp
-	switch regErr {
-	case ErrSessionBoundToOtherConn:
-		errResp = ResponseSessionAlreadyConnected
-	default:
-		errResp = ResponseUnableToBindSocket
-	}
-	err := c.SendUDPSessionResponse(requestID, errResp)
+func (c *datagramConn) handleSessionAlreadyRegistered(requestID RequestID, logger *zerolog.Logger) {
+	// Send another registration response since the session is already active
+	err := c.SendUDPSessionResponse(requestID, ResponseOk)
 	if err != nil {
-		c.logger.Err(err).Msgf("unable to send session registration error response (%d)", errResp)
+		logger.Err(err).Msgf("flow registration failure: unable to send an additional flow registration response")
+		return
+	}
+
+	session, err := c.sessionManager.GetSession(requestID)
+	if err != nil {
+		// If for some reason we can not find the session after attempting to register it, we can just return
+		// instead of trying to reset the idle timer for it.
+		return
+	}
+	// The session is already running in another routine so we want to restart the idle timeout since no proxied
+	// packets have come down yet.
+	session.ResetIdleTimer()
+	c.metrics.RetryFlowResponse()
+	logger.Debug().Msgf("flow registration response retry")
+}
+
+func (c *datagramConn) handleSessionMigration(requestID RequestID, logger *zerolog.Logger) {
+	// We need to migrate the currently running session to this edge connection.
+	session, err := c.sessionManager.GetSession(requestID)
+	if err != nil {
+		// If for some reason we can not find the session after attempting to register it, we can just return
+		// instead of trying to reset the idle timer for it.
+		return
+	}
+
+	// Migrate the session to use this edge connection instead of the currently running one.
+	// We also pass in this connection's logger to override the existing logger for the session.
+	session.Migrate(c, c.logger)
+
+	// Send another registration response since the session is already active
+	err = c.SendUDPSessionResponse(requestID, ResponseOk)
+	if err != nil {
+		logger.Err(err).Msgf("flow registration failure: unable to send an additional flow registration response")
+		return
+	}
+	logger.Debug().Msgf("flow registration migration")
+}
+
+func (c *datagramConn) handleSessionRegistrationFailure(requestID RequestID, logger *zerolog.Logger) {
+	err := c.SendUDPSessionResponse(requestID, ResponseUnableToBindSocket)
+	if err != nil {
+		logger.Err(err).Msgf("unable to send flow registration error response (%d)", ResponseUnableToBindSocket)
 	}
 }
 
 // Handles incoming datagrams that need to be sent to a registered session.
-func (c *datagramConn) handleSessionPayloadDatagram(datagram *UDPSessionPayloadDatagram) {
+func (c *datagramConn) handleSessionPayloadDatagram(datagram *UDPSessionPayloadDatagram, logger *zerolog.Logger) {
 	s, err := c.sessionManager.GetSession(datagram.RequestID)
 	if err != nil {
-		c.logger.Err(err).Msgf("unable to find session")
+		logger.Err(err).Msgf("unable to find flow")
 		return
 	}
 	// We ignore the bytes written to the socket because any partial write must return an error.
 	_, err = s.Write(datagram.Payload)
 	if err != nil {
-		c.logger.Err(err).Msgf("unable to write payload for unavailable session")
+		logger.Err(err).Msgf("unable to write payload for the flow")
 		return
 	}
 }

quic/v3/muxer_test.go

@@ -17,17 +17,19 @@ import (
 	v3 "github.com/cloudflare/cloudflared/quic/v3"
 )
 
-type noopEyeball struct{}
-
-func (noopEyeball) SendUDPSessionDatagram(datagram []byte) error {
-	return nil
+type noopEyeball struct {
+	connID uint8
 }
 
+func (noopEyeball) Serve(ctx context.Context) error              { return nil }
+func (n noopEyeball) ID() uint8                                  { return n.connID }
+func (noopEyeball) SendUDPSessionDatagram(datagram []byte) error { return nil }
 func (noopEyeball) SendUDPSessionResponse(id v3.RequestID, resp v3.SessionRegistrationResp) error {
 	return nil
 }
 
 type mockEyeball struct {
+	connID uint8
 	// datagram sent via SendUDPSessionDatagram
 	recvData chan []byte
 	// responses sent via SendUDPSessionResponse
@@ -39,6 +41,7 @@ type mockEyeball struct {
 
 func newMockEyeball() mockEyeball {
 	return mockEyeball{
+		connID:   0,
 		recvData: make(chan []byte, 1),
 		recvResp: make(chan struct {
 			id   v3.RequestID
@@ -47,6 +50,9 @@ func newMockEyeball() mockEyeball {
 	}
 }
 
+func (mockEyeball) Serve(ctx context.Context) error { return nil }
+func (m *mockEyeball) ID() uint8                    { return m.connID }
+
 func (m *mockEyeball) SendUDPSessionDatagram(datagram []byte) error {
 	b := make([]byte, len(datagram))
 	copy(b, datagram)
@@ -66,7 +72,7 @@ func (m *mockEyeball) SendUDPSessionResponse(id v3.RequestID, resp v3.SessionReg
 
 func TestDatagramConn_New(t *testing.T) {
 	log := zerolog.Nop()
-	conn := v3.NewDatagramConn(newMockQuicConn(), v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+	conn := v3.NewDatagramConn(newMockQuicConn(), v3.NewSessionManager(&noopMetrics{}, &log, ingress.DialUDPAddrPort), 0, &noopMetrics{}, &log)
 	if conn == nil {
 		t.Fatal("expected valid connection")
 	}
@@ -75,7 +81,7 @@ func TestDatagramConn_New(t *testing.T) {
 func TestDatagramConn_SendUDPSessionDatagram(t *testing.T) {
 	log := zerolog.Nop()
 	quic := newMockQuicConn()
-	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&noopMetrics{}, &log, ingress.DialUDPAddrPort), 0, &noopMetrics{}, &log)
 
 	payload := []byte{0xef, 0xef}
 	conn.SendUDPSessionDatagram(payload)
@@ -88,7 +94,7 @@ func TestDatagramConn_SendUDPSessionDatagram(t *testing.T) {
 func TestDatagramConn_SendUDPSessionResponse(t *testing.T) {
 	log := zerolog.Nop()
 	quic := newMockQuicConn()
-	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&noopMetrics{}, &log, ingress.DialUDPAddrPort), 0, &noopMetrics{}, &log)
 
 	conn.SendUDPSessionResponse(testRequestID, v3.ResponseDestinationUnreachable)
 	resp := <-quic.recv
@@ -109,7 +115,7 @@ func TestDatagramConn_SendUDPSessionResponse(t *testing.T) {
 func TestDatagramConnServe_ApplicationClosed(t *testing.T) {
 	log := zerolog.Nop()
 	quic := newMockQuicConn()
-	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&noopMetrics{}, &log, ingress.DialUDPAddrPort), 0, &noopMetrics{}, &log)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 	defer cancel()
@@ -125,7 +131,7 @@ func TestDatagramConnServe_ConnectionClosed(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 	defer cancel()
 	quic.ctx = ctx
-	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&noopMetrics{}, &log, ingress.DialUDPAddrPort), 0, &noopMetrics{}, &log)
 
 	err := conn.Serve(context.Background())
 	if !errors.Is(err, context.DeadlineExceeded) {
@@ -136,7 +142,7 @@ func TestDatagramConnServe_ConnectionClosed(t *testing.T) {
 func TestDatagramConnServe_ReceiveDatagramError(t *testing.T) {
 	log := zerolog.Nop()
 	quic := &mockQuicConnReadError{err: net.ErrClosed}
-	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&noopMetrics{}, &log, ingress.DialUDPAddrPort), 0, &noopMetrics{}, &log)
 
 	err := conn.Serve(context.Background())
 	if !errors.Is(err, net.ErrClosed) {
@@ -171,7 +177,7 @@ func TestDatagramConnServe_ErrorDatagramTypes(t *testing.T) {
 			log := zerolog.New(logOutput)
 			quic := newMockQuicConn()
 			quic.send <- test.input
-			conn := v3.NewDatagramConn(quic, &mockSessionManager{}, &log)
+			conn := v3.NewDatagramConn(quic, &mockSessionManager{}, 0, &noopMetrics{}, &log)
 
 			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
 			defer cancel()
@@ -212,7 +218,7 @@ func TestDatagramConnServe_RegisterSession_SessionManagerError(t *testing.T) {
 	quic := newMockQuicConn()
 	expectedErr := errors.New("unable to register session")
 	sessionManager := mockSessionManager{expectedRegErr: expectedErr}
-	conn := v3.NewDatagramConn(quic, &sessionManager, &log)
+	conn := v3.NewDatagramConn(quic, &sessionManager, 0, &noopMetrics{}, &log)
 
 	// Setup the muxer
 	ctx, cancel := context.WithCancelCause(context.Background())
@@ -234,19 +240,12 @@ func TestDatagramConnServe_RegisterSession_SessionManagerError(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if resp.RequestID != testRequestID && resp.ResponseType != v3.ResponseUnableToBindSocket {
+	if resp.RequestID != testRequestID || resp.ResponseType != v3.ResponseUnableToBindSocket {
 		t.Fatalf("expected registration response failure")
 	}
 
 	// Cancel the muxer Serve context and make sure it closes with the expected error
-	cancel(expectedContextCanceled)
-	err = <-done
-	if !errors.Is(err, context.Canceled) {
-		t.Fatal(err)
-	}
-	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
-		t.Fatal(err)
-	}
+	assertContextClosed(t, ctx, done, cancel)
 }
 
 func TestDatagramConnServe(t *testing.T) {
@@ -254,7 +253,7 @@ func TestDatagramConnServe(t *testing.T) {
 	quic := newMockQuicConn()
 	session := newMockSession()
 	sessionManager := mockSessionManager{session: &session}
-	conn := v3.NewDatagramConn(quic, &sessionManager, &log)
+	conn := v3.NewDatagramConn(quic, &sessionManager, 0, &noopMetrics{}, &log)
 
 	// Setup the muxer
 	ctx, cancel := context.WithCancelCause(context.Background())
@@ -276,7 +275,7 @@ func TestDatagramConnServe(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	if resp.RequestID != testRequestID && resp.ResponseType != v3.ResponseOk {
+	if resp.RequestID != testRequestID || resp.ResponseType != v3.ResponseOk {
 		t.Fatalf("expected registration response ok")
 	}
 
@@ -291,21 +290,160 @@ func TestDatagramConnServe(t *testing.T) {
 	}
 
 	// Cancel the muxer Serve context and make sure it closes with the expected error
-	cancel(expectedContextCanceled)
-	err = <-done
-	if !errors.Is(err, context.Canceled) {
+	assertContextClosed(t, ctx, done, cancel)
+}
+
+func TestDatagramConnServe_RegisterTwice(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	session := newMockSession()
+	sessionManager := mockSessionManager{session: &session}
+	conn := v3.NewDatagramConn(quic, &sessionManager, 0, &noopMetrics{}, &log)
+
+	// Setup the muxer
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(errors.New("other error"))
+	done := make(chan error, 1)
+	go func() {
+		done <- conn.Serve(ctx)
+	}()
+
+	// Send new session registration
+	datagram := newRegisterSessionDatagram(testRequestID)
+	quic.send <- datagram
+
+	// Wait for session registration response with success
+	datagram = <-quic.recv
+	var resp v3.UDPSessionRegistrationResponseDatagram
+	err := resp.UnmarshalBinary(datagram)
+	if err != nil {
 		t.Fatal(err)
 	}
-	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+
+	if resp.RequestID != testRequestID || resp.ResponseType != v3.ResponseOk {
+		t.Fatalf("expected registration response ok")
+	}
+
+	// Set the session manager to return already registered
+	sessionManager.expectedRegErr = v3.ErrSessionAlreadyRegistered
+	// Send the registration again as if we didn't receive it at the edge
+	datagram = newRegisterSessionDatagram(testRequestID)
+	quic.send <- datagram
+
+	// Wait for session registration response with success
+	datagram = <-quic.recv
+	err = resp.UnmarshalBinary(datagram)
+	if err != nil {
 		t.Fatal(err)
 	}
+
+	if resp.RequestID != testRequestID || resp.ResponseType != v3.ResponseOk {
+		t.Fatalf("expected registration response ok")
+	}
+
+	// We expect the session to be served
+	timer := time.NewTimer(15 * time.Second)
+	defer timer.Stop()
+	select {
+	case <-session.served:
+		break
+	case <-timer.C:
+		t.Fatalf("expected session serve to be called")
+	}
+
+	// Cancel the muxer Serve context and make sure it closes with the expected error
+	assertContextClosed(t, ctx, done, cancel)
+}
+
+func TestDatagramConnServe_MigrateConnection(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	session := newMockSession()
+	sessionManager := mockSessionManager{session: &session}
+	conn := v3.NewDatagramConn(quic, &sessionManager, 0, &noopMetrics{}, &log)
+	quic2 := newMockQuicConn()
+	conn2 := v3.NewDatagramConn(quic2, &sessionManager, 1, &noopMetrics{}, &log)
+
+	// Setup the muxer
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(errors.New("other error"))
+	done := make(chan error, 1)
+	go func() {
+		done <- conn.Serve(ctx)
+	}()
+
+	ctx2, cancel2 := context.WithCancelCause(context.Background())
+	defer cancel2(errors.New("other error"))
+	done2 := make(chan error, 1)
+	go func() {
+		done2 <- conn2.Serve(ctx2)
+	}()
+
+	// Send new session registration
+	datagram := newRegisterSessionDatagram(testRequestID)
+	quic.send <- datagram
+
+	// Wait for session registration response with success
+	datagram = <-quic.recv
+	var resp v3.UDPSessionRegistrationResponseDatagram
+	err := resp.UnmarshalBinary(datagram)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if resp.RequestID != testRequestID || resp.ResponseType != v3.ResponseOk {
+		t.Fatalf("expected registration response ok")
+	}
+
+	// Set the session manager to return already registered to another connection
+	sessionManager.expectedRegErr = v3.ErrSessionBoundToOtherConn
+	// Send the registration again as if we didn't receive it at the edge for a new connection
+	datagram = newRegisterSessionDatagram(testRequestID)
+	quic2.send <- datagram
+
+	// Wait for session registration response with success
+	datagram = <-quic2.recv
+	err = resp.UnmarshalBinary(datagram)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if resp.RequestID != testRequestID || resp.ResponseType != v3.ResponseOk {
+		t.Fatalf("expected registration response ok")
+	}
+
+	// We expect the session to be served
+	timer := time.NewTimer(15 * time.Second)
+	defer timer.Stop()
+	select {
+	case <-session.served:
+		break
+	case <-timer.C:
+		t.Fatalf("expected session serve to be called")
+	}
+
+	// Expect session to be migrated
+	select {
+	case id := <-session.migrated:
+		if id != conn2.ID() {
+			t.Fatalf("expected session to be migrated to connection 2")
+		}
+	case <-timer.C:
+		t.Fatalf("expected session migration to be called")
+	}
+
+	// Cancel the muxer Serve context and make sure it closes with the expected error
+	assertContextClosed(t, ctx, done, cancel)
+	// Cancel the second muxer Serve context and make sure it closes with the expected error
+	assertContextClosed(t, ctx2, done2, cancel2)
 }
 
 func TestDatagramConnServe_Payload_GetSessionError(t *testing.T) {
 	log := zerolog.Nop()
 	quic := newMockQuicConn()
+	// mockSessionManager will return the ErrSessionNotFound for any session attempting to be queried by the muxer
 	sessionManager := mockSessionManager{session: nil, expectedGetErr: v3.ErrSessionNotFound}
-	conn := v3.NewDatagramConn(quic, &sessionManager, &log)
+	conn := v3.NewDatagramConn(quic, &sessionManager, 0, &noopMetrics{}, &log)
 
 	// Setup the muxer
 	ctx, cancel := context.WithCancelCause(context.Background())
@@ -319,15 +457,13 @@ func TestDatagramConnServe_Payload_GetSessionError(t *testing.T) {
 	datagram := newSessionPayloadDatagram(testRequestID, []byte{0xef, 0xef})
 	quic.send <- datagram
 
+	// Since the muxer should eventually discard a failed registration request, there is no side-effect
+	// that the registration was failed beyond the muxer accepting the registration request. As such, the
+	// test can only ensure that the quic.send channel was consumed and that the muxer closes normally
+	// afterwards with the expected context cancelled trigger.
+
 	// Cancel the muxer Serve context and make sure it closes with the expected error
-	cancel(expectedContextCanceled)
-	err := <-done
-	if !errors.Is(err, context.Canceled) {
-		t.Fatal(err)
-	}
-	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
-		t.Fatal(err)
-	}
+	assertContextClosed(t, ctx, done, cancel)
 }
 
 func TestDatagramConnServe_Payload(t *testing.T) {
@@ -335,7 +471,7 @@ func TestDatagramConnServe_Payload(t *testing.T) {
 	quic := newMockQuicConn()
 	session := newMockSession()
 	sessionManager := mockSessionManager{session: &session}
-	conn := v3.NewDatagramConn(quic, &sessionManager, &log)
+	conn := v3.NewDatagramConn(quic, &sessionManager, 0, &noopMetrics{}, &log)
 
 	// Setup the muxer
 	ctx, cancel := context.WithCancelCause(context.Background())
@@ -357,14 +493,7 @@ func TestDatagramConnServe_Payload(t *testing.T) {
 	}
 
 	// Cancel the muxer Serve context and make sure it closes with the expected error
-	cancel(expectedContextCanceled)
-	err := <-done
-	if !errors.Is(err, context.Canceled) {
-		t.Fatal(err)
-	}
-	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
-		t.Fatal(err)
-	}
+	assertContextClosed(t, ctx, done, cancel)
 }
 
 func newRegisterSessionDatagram(id v3.RequestID) []byte {
@@ -402,6 +531,18 @@ func newSessionPayloadDatagram(id v3.RequestID, payload []byte) []byte {
 	return datagram
 }
 
+// Cancel the provided context and make sure it closes with the expected cancellation error
+func assertContextClosed(t *testing.T, ctx context.Context, done <-chan error, cancel context.CancelCauseFunc) {
+	cancel(expectedContextCanceled)
+	err := <-done
+	if !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+		t.Fatal(err)
+	}
+}
+
 type mockQuicConn struct {
 	ctx  context.Context
 	send chan []byte
@@ -454,7 +595,7 @@ type mockSessionManager struct {
 	expectedGetErr error
 }
 
-func (m *mockSessionManager) RegisterSession(request *v3.UDPSessionRegistrationDatagram, conn v3.DatagramWriter) (v3.Session, error) {
+func (m *mockSessionManager) RegisterSession(request *v3.UDPSessionRegistrationDatagram, conn v3.DatagramConn) (v3.Session, error) {
 	return m.session, m.expectedRegErr
 }
 
@@ -465,20 +606,25 @@ func (m *mockSessionManager) GetSession(requestID v3.RequestID) (v3.Session, err
 func (m *mockSessionManager) UnregisterSession(requestID v3.RequestID) {}
 
 type mockSession struct {
-	served chan struct{}
-	recv   chan []byte
+	served   chan struct{}
+	migrated chan uint8
+	recv     chan []byte
 }
 
 func newMockSession() mockSession {
 	return mockSession{
-		served: make(chan struct{}),
-		recv:   make(chan []byte, 1),
+		served:   make(chan struct{}),
+		migrated: make(chan uint8, 2),
+		recv:     make(chan []byte, 1),
 	}
 }
 
-func (m *mockSession) ID() v3.RequestID {
-	return testRequestID
-}
+func (m *mockSession) ID() v3.RequestID                                  { return testRequestID }
+func (m *mockSession) RemoteAddr() net.Addr                              { return testOriginAddr }
+func (m *mockSession) LocalAddr() net.Addr                               { return testLocalAddr }
+func (m *mockSession) ConnectionID() uint8                               { return 0 }
+func (m *mockSession) Migrate(conn v3.DatagramConn, log *zerolog.Logger) { m.migrated <- conn.ID() }
+func (m *mockSession) ResetIdleTimer()                                   {}
 
 func (m *mockSession) Serve(ctx context.Context) error {
 	close(m.served)

quic/v3/session.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/rs/zerolog"
@@ -20,10 +21,13 @@ const (
 	// read 1500 bytes from the origin, we limit the amount of bytes to be proxied to less than
 	// this value (maxDatagramPayloadLen).
 	maxOriginUDPPacketSize = 1500
+
+	logFlowID        = "flowID"
+	logPacketSizeKey = "packetSize"
 )
 
 // SessionCloseErr indicates that the session's Close method was called.
-var SessionCloseErr error = errors.New("session was closed")
+var SessionCloseErr error = errors.New("flow was closed directly")
 
 // SessionIdleErr is returned when the session was closed because there was no communication
 // in either direction over the session for the timeout period.
@@ -32,7 +36,7 @@ type SessionIdleErr struct {
 }
 
 func (e SessionIdleErr) Error() string {
-	return fmt.Sprintf("session idle for %v", e.timeout)
+	return fmt.Sprintf("flow was idle for %v", e.timeout)
 }
 
 func (e SessionIdleErr) Is(target error) bool {
@@ -47,6 +51,11 @@ func newSessionIdleErr(timeout time.Duration) error {
 type Session interface {
 	io.WriteCloser
 	ID() RequestID
+	ConnectionID() uint8
+	RemoteAddr() net.Addr
+	LocalAddr() net.Addr
+	ResetIdleTimer()
+	Migrate(eyeball DatagramConn, logger *zerolog.Logger)
 	// Serve starts the event loop for processing UDP packets
 	Serve(ctx context.Context) error
 }
@@ -55,31 +64,75 @@ type session struct {
 	id             RequestID
 	closeAfterIdle time.Duration
 	origin         io.ReadWriteCloser
-	eyeball        DatagramWriter
+	originAddr     net.Addr
+	localAddr      net.Addr
+	eyeball        atomic.Pointer[DatagramConn]
 	// activeAtChan is used to communicate the last read/write time
 	activeAtChan chan time.Time
 	closeChan    chan error
+	metrics      Metrics
 	log          *zerolog.Logger
 }
 
-func NewSession(id RequestID, closeAfterIdle time.Duration, origin io.ReadWriteCloser, eyeball DatagramWriter, log *zerolog.Logger) Session {
-	return &session{
+func NewSession(
+	id RequestID,
+	closeAfterIdle time.Duration,
+	origin io.ReadWriteCloser,
+	originAddr net.Addr,
+	localAddr net.Addr,
+	eyeball DatagramConn,
+	metrics Metrics,
+	log *zerolog.Logger,
+) Session {
+	logger := log.With().Str(logFlowID, id.String()).Logger()
+	session := &session{
 		id:             id,
 		closeAfterIdle: closeAfterIdle,
 		origin:         origin,
-		eyeball:        eyeball,
+		originAddr:     originAddr,
+		localAddr:      localAddr,
+		eyeball:        atomic.Pointer[DatagramConn]{},
 		// activeAtChan has low capacity. It can be full when there are many concurrent read/write. markActive() will
 		// drop instead of blocking because last active time only needs to be an approximation
 		activeAtChan: make(chan time.Time, 1),
 		closeChan:    make(chan error, 1),
-		log:          log,
+		metrics:      metrics,
+		log:          &logger,
 	}
+	session.eyeball.Store(&eyeball)
+	return session
 }
 
 func (s *session) ID() RequestID {
 	return s.id
 }
 
+func (s *session) RemoteAddr() net.Addr {
+	return s.originAddr
+}
+
+func (s *session) LocalAddr() net.Addr {
+	return s.localAddr
+}
+
+func (s *session) ConnectionID() uint8 {
+	eyeball := *(s.eyeball.Load())
+	return eyeball.ID()
+}
+
+func (s *session) Migrate(eyeball DatagramConn, logger *zerolog.Logger) {
+	current := *(s.eyeball.Load())
+	// Only migrate if the connection ids are different.
+	if current.ID() != eyeball.ID() {
+		s.eyeball.Store(&eyeball)
+		log := logger.With().Str(logFlowID, s.id.String()).Logger()
+		s.log = &log
+	}
+	// The session is already running so we want to restart the idle timeout since no proxied packets have come down yet.
+	s.markActive()
+	s.metrics.MigrateFlow()
+}
+
 func (s *session) Serve(ctx context.Context) error {
 	go func() {
 		// QUIC implementation copies data to another buffer before returning https://github.com/quic-go/quic-go/blob/v0.24.0/session.go#L1967-L1975
@@ -92,24 +145,29 @@ func (s *session) Serve(ctx context.Context) error {
 		for {
 			// Read from the origin UDP socket
 			n, err := s.origin.Read(readBuffer[DatagramPayloadHeaderLen:])
-			if errors.Is(err, net.ErrClosed) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
-				s.log.Debug().Msg("Session (origin) connection closed")
-			}
 			if err != nil {
+				if errors.Is(err, io.EOF) ||
+					errors.Is(err, io.ErrUnexpectedEOF) {
+					s.log.Debug().Msgf("flow (origin) connection closed: %v", err)
+				}
 				s.closeChan <- err
 				return
 			}
 			if n < 0 {
-				s.log.Warn().Int("packetSize", n).Msg("Session (origin) packet read was negative and was dropped")
+				s.log.Warn().Int(logPacketSizeKey, n).Msg("flow (origin) packet read was negative and was dropped")
 				continue
 			}
 			if n > maxDatagramPayloadLen {
-				s.log.Error().Int("packetSize", n).Msg("Session (origin) packet read was too large and was dropped")
+				s.metrics.PayloadTooLarge()
+				s.log.Error().Int(logPacketSizeKey, n).Msg("flow (origin) packet read was too large and was dropped")
 				continue
 			}
+			// We need to synchronize on the eyeball in-case that the connection was migrated. This should be rarely a point
+			// of lock contention, as a migration can only happen during startup of a session before traffic flow.
+			eyeball := *(s.eyeball.Load())
 			// Sending a packet to the session does block on the [quic.Connection], however, this is okay because it
 			// will cause back-pressure to the kernel buffer if the writes are not fast enough to the edge.
-			err = s.eyeball.SendUDPSessionDatagram(readBuffer[:DatagramPayloadHeaderLen+n])
+			err = eyeball.SendUDPSessionDatagram(readBuffer[:DatagramPayloadHeaderLen+n])
 			if err != nil {
 				s.closeChan <- err
 				return
@@ -124,12 +182,12 @@ func (s *session) Serve(ctx context.Context) error {
 func (s *session) Write(payload []byte) (n int, err error) {
 	n, err = s.origin.Write(payload)
 	if err != nil {
-		s.log.Err(err).Msg("Failed to write payload to session (remote)")
+		s.log.Err(err).Msg("failed to write payload to flow (remote)")
 		return n, err
 	}
 	// Write must return a non-nil error if it returns n < len(p). https://pkg.go.dev/io#Writer
 	if n < len(payload) {
-		s.log.Err(io.ErrShortWrite).Msg("Failed to write the full payload to session (remote)")
+		s.log.Err(io.ErrShortWrite).Msg("failed to write the full payload to flow (remote)")
 		return n, io.ErrShortWrite
 	}
 	// Mark the session as active since we proxied a packet to the origin.
@@ -137,6 +195,14 @@ func (s *session) Write(payload []byte) (n int, err error) {
 	return n, err
 }
 
+// ResetIdleTimer will restart the current idle timer.
+//
+// This public method is used to allow operators of sessions the ability to extend the session using information that is
+// known external to the session itself.
+func (s *session) ResetIdleTimer() {
+	s.markActive()
+}
+
 // Sends the last active time to the idle checker loop without blocking. activeAtChan will only be full when there
 // are many concurrent read/write. It is fine to lose some precision
 func (s *session) markActive() {

quic/v3/session_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"net"
+	"net/netip"
 	"slices"
 	"sync/atomic"
 	"testing"
@@ -14,11 +15,16 @@ import (
 	v3 "github.com/cloudflare/cloudflared/quic/v3"
 )
 
-var expectedContextCanceled = errors.New("expected context canceled")
+var (
+	expectedContextCanceled = errors.New("expected context canceled")
+
+	testOriginAddr = net.UDPAddrFromAddrPort(netip.MustParseAddrPort("127.0.0.1:0"))
+	testLocalAddr  = net.UDPAddrFromAddrPort(netip.MustParseAddrPort("127.0.0.1:0"))
+)
 
 func TestSessionNew(t *testing.T) {
 	log := zerolog.Nop()
-	session := v3.NewSession(testRequestID, 5*time.Second, nil, &noopEyeball{}, &log)
+	session := v3.NewSession(testRequestID, 5*time.Second, nil, testOriginAddr, testLocalAddr, &noopEyeball{}, &noopMetrics{}, &log)
 	if testRequestID != session.ID() {
 		t.Fatalf("session id doesn't match: %s != %s", testRequestID, session.ID())
 	}
@@ -27,7 +33,7 @@ func TestSessionNew(t *testing.T) {
 func testSessionWrite(t *testing.T, payload []byte) {
 	log := zerolog.Nop()
 	origin := newTestOrigin(makePayload(1280))
-	session := v3.NewSession(testRequestID, 5*time.Second, &origin, &noopEyeball{}, &log)
+	session := v3.NewSession(testRequestID, 5*time.Second, &origin, testOriginAddr, testLocalAddr, &noopEyeball{}, &noopMetrics{}, &log)
 	n, err := session.Write(payload)
 	if err != nil {
 		t.Fatal(err)
@@ -64,7 +70,7 @@ func testSessionServe_Origin(t *testing.T, payload []byte) {
 	log := zerolog.Nop()
 	eyeball := newMockEyeball()
 	origin := newTestOrigin(payload)
-	session := v3.NewSession(testRequestID, 3*time.Second, &origin, &eyeball, &log)
+	session := v3.NewSession(testRequestID, 3*time.Second, &origin, testOriginAddr, testLocalAddr, &eyeball, &noopMetrics{}, &log)
 	defer session.Close()
 
 	ctx, cancel := context.WithCancelCause(context.Background())
@@ -103,7 +109,7 @@ func TestSessionServe_OriginTooLarge(t *testing.T) {
 	eyeball := newMockEyeball()
 	payload := makePayload(1281)
 	origin := newTestOrigin(payload)
-	session := v3.NewSession(testRequestID, 2*time.Second, &origin, &eyeball, &log)
+	session := v3.NewSession(testRequestID, 2*time.Second, &origin, testOriginAddr, testLocalAddr, &eyeball, &noopMetrics{}, &log)
 	defer session.Close()
 
 	done := make(chan error)
@@ -123,10 +129,49 @@ func TestSessionServe_OriginTooLarge(t *testing.T) {
 	}
 }
 
+func TestSessionServe_Migrate(t *testing.T) {
+	log := zerolog.Nop()
+	eyeball := newMockEyeball()
+	pipe1, pipe2 := net.Pipe()
+	session := v3.NewSession(testRequestID, 2*time.Second, pipe2, testOriginAddr, testLocalAddr, &eyeball, &noopMetrics{}, &log)
+	defer session.Close()
+
+	done := make(chan error)
+	go func() {
+		done <- session.Serve(context.Background())
+	}()
+
+	// Migrate the session to a new connection before origin sends data
+	eyeball2 := newMockEyeball()
+	eyeball2.connID = 1
+	session.Migrate(&eyeball2, &log)
+
+	// Origin sends data
+	payload2 := []byte{0xde}
+	pipe1.Write(payload2)
+
+	// Expect write to eyeball2
+	data := <-eyeball2.recvData
+	if len(data) <= 17 || !slices.Equal(payload2, data[17:]) {
+		t.Fatalf("expected data to write to eyeball2 after migration: %+v", data)
+	}
+
+	select {
+	case data := <-eyeball.recvData:
+		t.Fatalf("expected no data to write to eyeball1 after migration: %+v", data)
+	default:
+	}
+
+	err := <-done
+	if !errors.Is(err, v3.SessionIdleErr{}) {
+		t.Error(err)
+	}
+}
+
 func TestSessionClose_Multiple(t *testing.T) {
 	log := zerolog.Nop()
 	origin := newTestOrigin(makePayload(128))
-	session := v3.NewSession(testRequestID, 5*time.Second, &origin, &noopEyeball{}, &log)
+	session := v3.NewSession(testRequestID, 5*time.Second, &origin, testOriginAddr, testLocalAddr, &noopEyeball{}, &noopMetrics{}, &log)
 	err := session.Close()
 	if err != nil {
 		t.Fatal(err)
@@ -145,7 +190,7 @@ func TestSessionServe_IdleTimeout(t *testing.T) {
 	log := zerolog.Nop()
 	origin := newTestIdleOrigin(10 * time.Second) // Make idle time longer than closeAfterIdle
 	closeAfterIdle := 2 * time.Second
-	session := v3.NewSession(testRequestID, closeAfterIdle, &origin, &noopEyeball{}, &log)
+	session := v3.NewSession(testRequestID, closeAfterIdle, &origin, testOriginAddr, testLocalAddr, &noopEyeball{}, &noopMetrics{}, &log)
 	err := session.Serve(context.Background())
 	if !errors.Is(err, v3.SessionIdleErr{}) {
 		t.Fatal(err)
@@ -167,7 +212,7 @@ func TestSessionServe_ParentContextCanceled(t *testing.T) {
 	origin := newTestIdleOrigin(10 * time.Second)
 	closeAfterIdle := 10 * time.Second
 
-	session := v3.NewSession(testRequestID, closeAfterIdle, &origin, &noopEyeball{}, &log)
+	session := v3.NewSession(testRequestID, closeAfterIdle, &origin, testOriginAddr, testLocalAddr, &noopEyeball{}, &noopMetrics{}, &log)
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 	defer cancel()
 	err := session.Serve(ctx)
@@ -188,7 +233,7 @@ func TestSessionServe_ParentContextCanceled(t *testing.T) {
 func TestSessionServe_ReadErrors(t *testing.T) {
 	log := zerolog.Nop()
 	origin := newTestErrOrigin(net.ErrClosed, nil)
-	session := v3.NewSession(testRequestID, 30*time.Second, &origin, &noopEyeball{}, &log)
+	session := v3.NewSession(testRequestID, 30*time.Second, &origin, testOriginAddr, testLocalAddr, &noopEyeball{}, &noopMetrics{}, &log)
 	err := session.Serve(context.Background())
 	if !errors.Is(err, net.ErrClosed) {
 		t.Fatal(err)
@@ -249,7 +294,7 @@ func newTestIdleOrigin(d time.Duration) testIdleOrigin {
 
 func (o *testIdleOrigin) Read(p []byte) (n int, err error) {
 	time.Sleep(o.duration)
-	return 0, nil
+	return -1, nil
 }
 
 func (o *testIdleOrigin) Write(p []byte) (n int, err error) {

supervisor/supervisor.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/quic-go/quic-go"
 	"github.com/rs/zerolog"
 
@@ -82,12 +83,14 @@ func NewSupervisor(config *TunnelConfig, orchestrator *orchestration.Orchestrato
 	edgeAddrHandler := NewIPAddrFallback(config.MaxEdgeAddrRetries)
 	edgeBindAddr := config.EdgeBindAddr
 
-	sessionManager := v3.NewSessionManager(config.Log, ingress.DialUDPAddrPort)
+	datagramMetrics := v3.NewMetrics(prometheus.DefaultRegisterer)
+	sessionManager := v3.NewSessionManager(datagramMetrics, config.Log, ingress.DialUDPAddrPort)
 
 	edgeTunnelServer := EdgeTunnelServer{
 		config:            config,
 		orchestrator:      orchestrator,
 		sessionManager:    sessionManager,
+		datagramMetrics:   datagramMetrics,
 		edgeAddrs:         edgeIPs,
 		edgeAddrHandler:   edgeAddrHandler,
 		edgeBindAddr:      edgeBindAddr,

supervisor/tunnel.go

@@ -176,6 +176,7 @@ type EdgeTunnelServer struct {
 	config            *TunnelConfig
 	orchestrator      *orchestration.Orchestrator
 	sessionManager    v3.SessionManager
+	datagramMetrics   v3.Metrics
 	edgeAddrHandler   EdgeAddrHandler
 	edgeAddrs         *edgediscovery.Edge
 	edgeBindAddr      net.IP
@@ -606,6 +607,8 @@ func (e *EdgeTunnelServer) serveQUIC(
 			ctx,
 			conn,
 			e.sessionManager,
+			connIndex,
+			e.datagramMetrics,
 			connLogger.Logger(),
 		)
 	} else {

tracing/tracing_test.go

@@ -72,3 +72,9 @@ func TestAddingSpansWithNilMap(t *testing.T) {
 	// a panic shouldn't occur
 	tr.AddSpans(nil)
 }
+
+func FuzzNewIdentity(f *testing.F) {
+	f.Fuzz(func(t *testing.T, trace string) {
+		_, _ = NewIdentity(trace)
+	})
+}

validation/validation_test.go

@@ -197,3 +197,10 @@ func createSecureMockServerAndClient(handler http.Handler) (*httptest.Server, *h
 
 	return server, client, nil
 }
+
+func FuzzNewAccessValidator(f *testing.F) {
+	f.Fuzz(func(t *testing.T, domain string, issuer string, applicationAUD string) {
+		ctx := context.Background()
+		_, _ = NewAccessValidator(ctx, domain, issuer, applicationAUD)
+	})
+}