Release 2023.6.1

I deliberately kept this as an unregistertimeout because that was the
intent. In the future we could change this to a UDPConnConfig if we want
to pass multiple values here.

The idea of this PR is simply to add a configurable unregister UDP
timeout.

Makefile

@@ -140,6 +140,7 @@ container:
 generate-docker-version:
 	echo latest $(VERSION) > versions
 
+
 .PHONY: test
 test: vet
 ifndef CI
@@ -147,9 +148,18 @@ ifndef CI
 else
 	@mkdir -p .cover
 	go test -v -mod=vendor -race $(LDFLAGS) -coverprofile=".cover/c.out" ./...
-	go tool cover -html ".cover/c.out" -o .cover/all.html
 endif
 
+.PHONY: cover
+cover:
+	@echo ""
+	@echo "=====> Total test coverage: <====="
+	@echo ""
+	# Print the overall coverage here for quick access.
+	$Q go tool cover -func ".cover/c.out" | grep "total:" | awk '{print $$3}'
+	# 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: test-ssh-server
 test-ssh-server:
 	docker-compose -f ssh_server_tests/docker-compose.yml up

RELEASE_NOTES

@@ -1,3 +1,13 @@
+2023.6.1
+- 2023-06-19 TUN-7480: Added a timeout for unregisterUDP.
+- 2023-06-16 TUN-7477: Add UDP/TCP session metrics
+- 2023-06-14 TUN-7468: Increase the limit of incoming streams
+
+2023.6.0
+- 2023-06-15 TUN-7471: Fixes cloudflared not closing the quic stream on unregister UDP session
+- 2023-06-09 TUN-7463: Add default ingress rule if no ingress rules are provided when updating the configuration
+- 2023-05-31 TUN-7447: Add a cover build to report code coverage
+
 2023.5.1
 - 2023-05-16 TUN-7424: Add CORS headers to host_details responses
 - 2023-05-11 TUN-7421: Add *.cloudflare.com to permitted Origins for management WebSocket requests

cfsetup.yaml

@@ -29,6 +29,12 @@ buster: &buster
       - export GOARCH=amd64
       - export FIPS=true
       - make cloudflared
+  cover: 
+    build_dir: *build_dir
+    builddeps: *build_deps
+    pre-cache: *build_pre_cache
+    post-cache: 
+     - make cover
   # except FIPS (handled in github-fips-release-pkgs) and macos (handled in github-release-macos-amd64)
   github-release-pkgs:
     build_dir: *build_dir

cmd/cloudflared/tunnel/cmd.go

@@ -79,6 +79,9 @@ const (
 	// hostKeyPath is the path of the dir to save SSH host keys too
 	hostKeyPath = "host-key-path"
 
+	// udpUnregisterSessionTimeout is how long we wait before we stop trying to unregister a UDP session from the edge
+	udpUnregisterSessionTimeoutFlag = "udp-unregister-session-timeout"
+
 	// uiFlag is to enable launching cloudflared in interactive UI mode
 	uiFlag = "ui"
 
@@ -399,7 +402,7 @@ func StartServer(
 		}
 	}
 
-	localRules := []ingress.Rule{}
+	internalRules := []ingress.Rule{}
 	if features.Contains(features.FeatureManagementLogs) {
 		serviceIP := c.String("service-op-ip")
 		if edgeAddrs, err := edgediscovery.ResolveEdge(log, tunnelConfig.Region, tunnelConfig.EdgeIPVersion); err == nil {
@@ -416,9 +419,9 @@ func StartServer(
 			logger.ManagementLogger.Log,
 			logger.ManagementLogger,
 		)
-		localRules = []ingress.Rule{ingress.NewManagementRule(mgmt)}
+		internalRules = []ingress.Rule{ingress.NewManagementRule(mgmt)}
 	}
-	orchestrator, err := orchestration.NewOrchestrator(ctx, orchestratorConfig, tunnelConfig.Tags, localRules, tunnelConfig.Log)
+	orchestrator, err := orchestration.NewOrchestrator(ctx, orchestratorConfig, tunnelConfig.Tags, internalRules, tunnelConfig.Log)
 	if err != nil {
 		return err
 	}
@@ -683,6 +686,11 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 			Value:  4,
 			Hidden: true,
 		}),
+		altsrc.NewDurationFlag(&cli.DurationFlag{
+			Name:   udpUnregisterSessionTimeoutFlag,
+			Value:  5 * time.Second,
+			Hidden: true,
+		}),
 		altsrc.NewStringFlag(&cli.StringFlag{
 			Name:  connectorLabelFlag,
 			Usage: "Use this option to give a meaningful label to a specific connector. When a tunnel starts up, a connector id unique to the tunnel is generated. This is a uuid. To make it easier to identify a connector, we will use the hostname of the machine the tunnel is running on along with the connector ID. This option exists if one wants to have more control over what their individual connectors are called.",

cmd/cloudflared/tunnel/configuration.go

@@ -231,14 +231,15 @@ func prepareTunnelConfig(
 		Observer:        observer,
 		ReportedVersion: info.Version(),
 		// Note TUN-3758 , we use Int because UInt is not supported with altsrc
-		Retries:            uint(c.Int("retries")),
-		RunFromTerminal:    isRunningFromTerminal(),
-		NamedTunnel:        namedTunnel,
-		ProtocolSelector:   protocolSelector,
-		EdgeTLSConfigs:     edgeTLSConfigs,
-		NeedPQ:             needPQ,
-		PQKexIdx:           pqKexIdx,
-		MaxEdgeAddrRetries: uint8(c.Int("max-edge-addr-retries")),
+		Retries:                     uint(c.Int("retries")),
+		RunFromTerminal:             isRunningFromTerminal(),
+		NamedTunnel:                 namedTunnel,
+		ProtocolSelector:            protocolSelector,
+		EdgeTLSConfigs:              edgeTLSConfigs,
+		NeedPQ:                      needPQ,
+		PQKexIdx:                    pqKexIdx,
+		MaxEdgeAddrRetries:          uint8(c.Int("max-edge-addr-retries")),
+		UDPUnregisterSessionTimeout: c.Duration(udpUnregisterSessionTimeoutFlag),
 	}
 	packetConfig, err := newPacketConfig(c, log)
 	if err != nil {

connection/quic.go

@@ -63,6 +63,8 @@ type QUICConnection struct {
 	controlStreamHandler ControlStreamHandler
 	connOptions          *tunnelpogs.ConnectionOptions
 	connIndex            uint8
+
+	udpUnregisterTimeout time.Duration
 }
 
 // NewQUICConnection returns a new instance of QUICConnection.
@@ -78,6 +80,7 @@ func NewQUICConnection(
 	controlStreamHandler ControlStreamHandler,
 	logger *zerolog.Logger,
 	packetRouterConfig *ingress.GlobalRouterConfig,
+	udpUnregisterTimeout time.Duration,
 ) (*QUICConnection, error) {
 	udpConn, err := createUDPConnForConnIndex(connIndex, localAddr, logger)
 	if err != nil {
@@ -112,6 +115,7 @@ func NewQUICConnection(
 		controlStreamHandler: controlStreamHandler,
 		connOptions:          connOptions,
 		connIndex:            connIndex,
+		udpUnregisterTimeout: udpUnregisterTimeout,
 	}, nil
 }
 
@@ -357,7 +361,7 @@ func (q *QUICConnection) serveUDPSession(session *datagramsession.Session, close
 // closeUDPSession first unregisters the session from session manager, then it tries to unregister from edge
 func (q *QUICConnection) closeUDPSession(ctx context.Context, sessionID uuid.UUID, message string) {
 	q.sessionManager.UnregisterSession(ctx, sessionID, message, false)
-	stream, err := q.session.OpenStream()
+	quicStream, err := q.session.OpenStream()
 	if err != nil {
 		// Log this at debug because this is not an error if session was closed due to lost connection
 		// with edge
@@ -367,7 +371,10 @@ func (q *QUICConnection) closeUDPSession(ctx context.Context, sessionID uuid.UUI
 			Msgf("Failed to open quic stream to unregister udp session with edge")
 		return
 	}
-	rpcClientStream, err := quicpogs.NewRPCClientStream(ctx, stream, q.logger)
+
+	stream := quicpogs.NewSafeStreamCloser(quicStream)
+	defer stream.Close()
+	rpcClientStream, err := quicpogs.NewRPCClientStream(ctx, stream, q.udpUnregisterTimeout, q.logger)
 	if err != nil {
 		// Log this at debug because this is not an error if session was closed due to lost connection
 		// with edge
@@ -375,6 +382,8 @@ func (q *QUICConnection) closeUDPSession(ctx context.Context, sessionID uuid.UUI
 			Msgf("Failed to open rpc stream to unregister udp session with edge")
 		return
 	}
+	defer rpcClientStream.Close()
+
 	if err := rpcClientStream.UnregisterUdpSession(ctx, sessionID, message); err != nil {
 		q.logger.Err(err).Str("sessionID", sessionID.String()).
 			Msgf("Failed to unregister udp session with edge")

connection/quic_test.go

@@ -725,6 +725,7 @@ func testQUICConnection(udpListenerAddr net.Addr, t *testing.T, index uint8) *QU
 		fakeControlStream{},
 		&log,
 		nil,
+		5*time.Second,
 	)
 	require.NoError(t, err)
 	return qc

datagramsession/manager.go

@@ -118,6 +118,7 @@ func (m *manager) registerSession(ctx context.Context, registration *registerSes
 	session := m.newSession(registration.sessionID, registration.originProxy)
 	m.sessions[registration.sessionID] = session
 	registration.resultChan <- session
+	incrementUDPSessions()
 }
 
 func (m *manager) newSession(id uuid.UUID, dstConn io.ReadWriteCloser) *Session {
@@ -163,6 +164,7 @@ func (m *manager) unregisterSession(unregistration *unregisterSessionEvent) {
 	if ok {
 		delete(m.sessions, unregistration.sessionID)
 		session.close(unregistration.err)
+		decrementUDPActiveSessions()
 	}
 }
 

datagramsession/metrics.go

@@ -0,0 +1,40 @@
+package datagramsession
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	namespace = "cloudflared"
+)
+
+var (
+	activeUDPSessions = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: "udp",
+		Name:      "active_sessions",
+		Help:      "Concurrent count of UDP sessions that are being proxied to any origin",
+	})
+	totalUDPSessions = prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Subsystem: "udp",
+		Name:      "total_sessions",
+		Help:      "Total count of UDP sessions that have been proxied to any origin",
+	})
+)
+
+func init() {
+	prometheus.MustRegister(
+		activeUDPSessions,
+		totalUDPSessions,
+	)
+}
+
+func incrementUDPSessions() {
+	totalUDPSessions.Inc()
+	activeUDPSessions.Inc()
+}
+
+func decrementUDPActiveSessions() {
+	activeUDPSessions.Dec()
+}

ingress/ingress.go

@@ -44,7 +44,7 @@ func (ing Ingress) FindMatchingRule(hostname, path string) (*Rule, int) {
 	if err == nil {
 		hostname = host
 	}
-	for i, rule := range ing.LocalRules {
+	for i, rule := range ing.InternalRules {
 		if rule.Matches(hostname, path) {
 			// Local rule matches return a negative rule index to distiguish local rules from user-defined rules in logs
 			// Full range would be [-1 .. )
@@ -77,7 +77,7 @@ func matchHost(ruleHost, reqHost string) bool {
 // Ingress maps eyeball requests to origins.
 type Ingress struct {
 	// Set of ingress rules that are not added to remote config, e.g. management
-	LocalRules []Rule
+	InternalRules []Rule
 	// Rules that are provided by the user from remote or local configuration
 	Rules    []Rule              `json:"ingress"`
 	Defaults OriginRequestConfig `json:"originRequest"`
@@ -110,16 +110,18 @@ func ParseIngressFromConfigAndCLI(conf *config.Configuration, c *cli.Context, lo
 	//   --bastion for ssh bastion service
 	ingressRules, err = parseCLIIngress(c, false)
 	if errors.Is(err, ErrNoIngressRulesCLI) {
-		// Only log a warning if the tunnel is not a remotely managed tunnel and the config
-		// will be loaded after connecting.
+		// If no token is provided, the probability of NOT being a remotely managed tunnel is higher.
+		// So, we should warn the user that no ingress rules were found, because remote configuration will most likely not exist.
 		if !c.IsSet("token") {
 			log.Warn().Msgf(ErrNoIngressRulesCLI.Error())
 		}
 		return newDefaultOrigin(c, log), nil
 	}
+
 	if err != nil {
 		return Ingress{}, err
 	}
+
 	return ingressRules, nil
 }
 
@@ -148,14 +150,10 @@ func parseCLIIngress(c *cli.Context, allowURLFromArgs bool) (Ingress, error) {
 // newDefaultOrigin always returns a 503 response code to help indicate that there are no ingress
 // rules setup, but the tunnel is reachable.
 func newDefaultOrigin(c *cli.Context, log *zerolog.Logger) Ingress {
-	noRulesService := newDefaultStatusCode(log)
+	defaultRule := GetDefaultIngressRules(log)
 	defaults := originRequestFromSingleRule(c)
 	ingress := Ingress{
-		Rules: []Rule{
-			{
-				Service: &noRulesService,
-			},
-		},
+		Rules:    defaultRule,
 		Defaults: defaults,
 	}
 	return ingress
@@ -219,6 +217,17 @@ func (ing Ingress) CatchAll() *Rule {
 	return &ing.Rules[len(ing.Rules)-1]
 }
 
+// Gets the default ingress rule that will be return 503 status
+// code for all incoming requests.
+func GetDefaultIngressRules(log *zerolog.Logger) []Rule {
+	noRulesService := newDefaultStatusCode(log)
+	return []Rule{
+		{
+			Service: &noRulesService,
+		},
+	}
+}
+
 func validateAccessConfiguration(cfg *config.AccessConfig) error {
 	if !cfg.Required {
 		return nil

orchestration/orchestrator.go

@@ -28,8 +28,8 @@ type Orchestrator struct {
 	lock sync.RWMutex
 	// Underlying value is proxy.Proxy, can be read without the lock, but still needs the lock to update
 	proxy atomic.Value
-	// Set of local ingress rules defined at cloudflared startup (separate from user-defined ingress rules)
-	localRules         []ingress.Rule
+	// Set of internal ingress rules defined at cloudflared startup (separate from user-defined ingress rules)
+	internalRules      []ingress.Rule
 	warpRoutingEnabled atomic.Bool
 	config             *Config
 	tags               []tunnelpogs.Tag
@@ -41,13 +41,13 @@ type Orchestrator struct {
 	proxyShutdownC chan<- struct{}
 }
 
-func NewOrchestrator(ctx context.Context, config *Config, tags []tunnelpogs.Tag, localRules []ingress.Rule, log *zerolog.Logger) (*Orchestrator, error) {
+func NewOrchestrator(ctx context.Context, config *Config, tags []tunnelpogs.Tag, internalRules []ingress.Rule, log *zerolog.Logger) (*Orchestrator, error) {
 	o := &Orchestrator{
 		// Lowest possible version, any remote configuration will have version higher than this
 		// Starting at -1 allows a configuration migration (local to remote) to override the current configuration as it
 		// will start at version 0.
 		currentVersion: -1,
-		localRules:     localRules,
+		internalRules:  internalRules,
 		config:         config,
 		tags:           tags,
 		log:            log,
@@ -116,8 +116,13 @@ func (o *Orchestrator) updateIngress(ingressRules ingress.Ingress, warpRouting i
 	default:
 	}
 
-	// Assign the local ingress rules to the parsed ingress
-	ingressRules.LocalRules = o.localRules
+	// Assign the internal ingress rules to the parsed ingress
+	ingressRules.InternalRules = o.internalRules
+
+	// Check if ingress rules are empty, and add the default route if so.
+	if ingressRules.IsEmpty() {
+		ingressRules.Rules = ingress.GetDefaultIngressRules(o.log)
+	}
 
 	// Start new proxy before closing the ones from last version.
 	// The upside is we don't need to restart proxy from last version, which can fail

orchestration/orchestrator_test.go

@@ -91,15 +91,15 @@ func TestUpdateConfiguration(t *testing.T) {
         "enabled": true,
         "connectTimeout": 10
     }
-}	
+}
 `)
 
 	updateWithValidation(t, orchestrator, 2, configJSONV2)
 	configV2 := orchestrator.config
-	// Validate local ingress rules
-	require.Equal(t, "management.argotunnel.com", configV2.Ingress.LocalRules[0].Hostname)
-	require.True(t, configV2.Ingress.LocalRules[0].Matches("management.argotunnel.com", "/ping"))
-	require.Equal(t, "management", configV2.Ingress.LocalRules[0].Service.String())
+	// Validate internal ingress rules
+	require.Equal(t, "management.argotunnel.com", configV2.Ingress.InternalRules[0].Hostname)
+	require.True(t, configV2.Ingress.InternalRules[0].Matches("management.argotunnel.com", "/ping"))
+	require.Equal(t, "management", configV2.Ingress.InternalRules[0].Service.String())
 	// Validate ingress rule 0
 	require.Equal(t, "jira.tunnel.org", configV2.Ingress.Rules[0].Hostname)
 	require.True(t, configV2.Ingress.Rules[0].Matches("jira.tunnel.org", "/login"))
@@ -145,7 +145,7 @@ func TestUpdateConfiguration(t *testing.T) {
 {
 	"originRequest":
 }
-	
+
 `)
 
 	resp = orchestrator.UpdateConfig(3, invalidJSON)
@@ -165,7 +165,7 @@ func TestUpdateConfiguration(t *testing.T) {
     "warp-routing": {
         "enabled": false
     }
-}	
+}
 `)
 	updateWithValidation(t, orchestrator, 10, configJSONV10)
 	configV10 := orchestrator.config
@@ -204,12 +204,34 @@ func TestUpdateConfiguration_FromMigration(t *testing.T) {
     "warp-routing": {
         "enabled": true
     }
-}	
+}
 `)
 	updateWithValidation(t, orchestrator, 0, configJSONV2)
 	require.Len(t, orchestrator.config.Ingress.Rules, 1)
 }
 
+// Validates that the default ingress rule will be set if there is no rule provided from the remote.
+func TestUpdateConfiguration_WithoutIngressRule(t *testing.T) {
+	initConfig := &Config{
+		Ingress: &ingress.Ingress{},
+	}
+	orchestrator, err := NewOrchestrator(context.Background(), initConfig, testTags, []ingress.Rule{}, &testLogger)
+	require.NoError(t, err)
+	initOriginProxy, err := orchestrator.GetOriginProxy()
+	require.NoError(t, err)
+	require.Implements(t, (*connection.OriginProxy)(nil), initOriginProxy)
+
+	// We need to create an empty RemoteConfigJSON because that will get unmarshalled to a RemoteConfig
+	emptyConfig := &ingress.RemoteConfigJSON{}
+	configBytes, err := json.Marshal(emptyConfig)
+	if err != nil {
+		require.FailNow(t, "The RemoteConfigJSON shouldn't fail while being marshalled")
+	}
+
+	updateWithValidation(t, orchestrator, 0, configBytes)
+	require.Len(t, orchestrator.config.Ingress.Rules, 1)
+}
+
 // TestConcurrentUpdateAndRead makes sure orchestrator can receive updates and return origin proxy concurrently
 func TestConcurrentUpdateAndRead(t *testing.T) {
 	const (

proxy/metrics.go

@@ -43,6 +43,22 @@ var (
 			Help:      "Count of error proxying to origin",
 		},
 	)
+	activeTCPSessions = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: connection.MetricsNamespace,
+			Subsystem: "tcp",
+			Name:      "active_sessions",
+			Help:      "Concurrent count of TCP sessions that are being proxied to any origin",
+		},
+	)
+	totalTCPSessions = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: connection.MetricsNamespace,
+			Subsystem: "tcp",
+			Name:      "total_sessions",
+			Help:      "Total count of TCP sessions that have been proxied to any origin",
+		},
+	)
 )
 
 func init() {
@@ -51,6 +67,8 @@ func init() {
 		concurrentRequests,
 		responseByCode,
 		requestErrors,
+		activeTCPSessions,
+		totalTCPSessions,
 	)
 }
 
@@ -62,3 +80,14 @@ func incrementRequests() {
 func decrementConcurrentRequests() {
 	concurrentRequests.Dec()
 }
+
+func incrementTCPRequests() {
+	incrementRequests()
+	totalTCPSessions.Inc()
+	activeTCPSessions.Inc()
+}
+
+func decrementTCPConcurrentRequests() {
+	decrementConcurrentRequests()
+	activeTCPSessions.Dec()
+}

proxy/proxy.go

@@ -158,8 +158,8 @@ func (p *Proxy) ProxyTCP(
 	rwa connection.ReadWriteAcker,
 	req *connection.TCPRequest,
 ) error {
-	incrementRequests()
-	defer decrementConcurrentRequests()
+	incrementTCPRequests()
+	defer decrementTCPConcurrentRequests()
 
 	if p.warpRouting == nil {
 		err := errors.New(`cloudflared received a request from WARP client, but your configuration has disabled ingress from WARP clients. To enable this, set "warp-routing:\n\t enabled: true" in your config.yaml`)

quic/quic_protocol.go

@@ -39,6 +39,9 @@ const (
 	HandshakeIdleTimeout = 5 * time.Second
 	MaxIdleTimeout       = 5 * time.Second
 	MaxIdlePingPeriod    = 1 * time.Second
+
+	// MaxIncomingStreams is 2^60, which is the maximum supported value by Quic-Go
+	MaxIncomingStreams = 1 << 60
 )
 
 // RequestServerStream is a stream to serve requests
@@ -226,9 +229,12 @@ func writeSignature(stream io.Writer, signature ProtocolSignature) error {
 type RPCClientStream struct {
 	client    tunnelpogs.CloudflaredServer_PogsClient
 	transport rpc.Transport
+
+	// Time we wait for the server to respond to a request before we close the connection.
+	rpcUnregisterUDPSessionDeadline time.Duration
 }
 
-func NewRPCClientStream(ctx context.Context, stream io.ReadWriteCloser, logger *zerolog.Logger) (*RPCClientStream, error) {
+func NewRPCClientStream(ctx context.Context, stream io.ReadWriteCloser, rpcUnregisterUDPSessionDeadline time.Duration, logger *zerolog.Logger) (*RPCClientStream, error) {
 	n, err := stream.Write(RPCStreamProtocolSignature[:])
 	if err != nil {
 		return nil, err
@@ -242,8 +248,9 @@ func NewRPCClientStream(ctx context.Context, stream io.ReadWriteCloser, logger *
 		tunnelrpc.ConnLog(logger),
 	)
 	return &RPCClientStream{
-		client:    tunnelpogs.NewCloudflaredServer_PogsClient(conn.Bootstrap(ctx), conn),
-		transport: transport,
+		client:                          tunnelpogs.NewCloudflaredServer_PogsClient(conn.Bootstrap(ctx), conn),
+		transport:                       transport,
+		rpcUnregisterUDPSessionDeadline: rpcUnregisterUDPSessionDeadline,
 	}, nil
 }
 
@@ -252,6 +259,8 @@ func (rcs *RPCClientStream) RegisterUdpSession(ctx context.Context, sessionID uu
 }
 
 func (rcs *RPCClientStream) UnregisterUdpSession(ctx context.Context, sessionID uuid.UUID, message string) error {
+	ctx, cancel := context.WithTimeout(ctx, rcs.rpcUnregisterUDPSessionDeadline)
+	defer cancel()
 	return rcs.client.UnregisterUdpSession(ctx, sessionID, message)
 }
 

quic/quic_protocol_test.go

@@ -109,6 +109,63 @@ func TestConnectResponseMeta(t *testing.T) {
 	}
 }
 
+func TestUnregisterUdpSession(t *testing.T) {
+	unregisterMessage := "closed by eyeball"
+
+	var tests = []struct {
+		name             string
+		sessionRPCServer mockSessionRPCServer
+		timeout          time.Duration
+	}{
+
+		{
+			name: "UnregisterUdpSessionTimesout if the RPC server does not respond",
+			sessionRPCServer: mockSessionRPCServer{
+				sessionID:         uuid.New(),
+				dstIP:             net.IP{172, 16, 0, 1},
+				dstPort:           8000,
+				closeIdleAfter:    testCloseIdleAfterHint,
+				unregisterMessage: unregisterMessage,
+				traceContext:      "1241ce3ecdefc68854e8514e69ba42ca:b38f1bf5eae406f3:0:1",
+			},
+			// very very low value so we trigger the timeout every time.
+			timeout: time.Nanosecond * 1,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			logger := zerolog.Nop()
+			clientStream, serverStream := newMockRPCStreams()
+			sessionRegisteredChan := make(chan struct{})
+			go func() {
+				protocol, err := DetermineProtocol(serverStream)
+				assert.NoError(t, err)
+				rpcServerStream, err := NewRPCServerStream(serverStream, protocol)
+				assert.NoError(t, err)
+				err = rpcServerStream.Serve(test.sessionRPCServer, nil, &logger)
+				assert.NoError(t, err)
+
+				serverStream.Close()
+				close(sessionRegisteredChan)
+			}()
+
+			rpcClientStream, err := NewRPCClientStream(context.Background(), clientStream, test.timeout, &logger)
+			assert.NoError(t, err)
+
+			reg, err := rpcClientStream.RegisterUdpSession(context.Background(), test.sessionRPCServer.sessionID, test.sessionRPCServer.dstIP, test.sessionRPCServer.dstPort, testCloseIdleAfterHint, test.sessionRPCServer.traceContext)
+			assert.NoError(t, err)
+			assert.NoError(t, reg.Err)
+
+			assert.Error(t, rpcClientStream.UnregisterUdpSession(context.Background(), test.sessionRPCServer.sessionID, unregisterMessage))
+
+			rpcClientStream.Close()
+			<-sessionRegisteredChan
+		})
+	}
+
+}
+
 func TestRegisterUdpSession(t *testing.T) {
 	unregisterMessage := "closed by eyeball"
 
@@ -157,7 +214,7 @@ func TestRegisterUdpSession(t *testing.T) {
 				close(sessionRegisteredChan)
 			}()
 
-			rpcClientStream, err := NewRPCClientStream(context.Background(), clientStream, &logger)
+			rpcClientStream, err := NewRPCClientStream(context.Background(), clientStream, 5*time.Second, &logger)
 			assert.NoError(t, err)
 
 			reg, err := rpcClientStream.RegisterUdpSession(context.Background(), test.sessionRPCServer.sessionID, test.sessionRPCServer.dstIP, test.sessionRPCServer.dstPort, testCloseIdleAfterHint, test.sessionRPCServer.traceContext)
@@ -208,7 +265,7 @@ func TestManageConfiguration(t *testing.T) {
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	rpcClientStream, err := NewRPCClientStream(ctx, clientStream, &logger)
+	rpcClientStream, err := NewRPCClientStream(ctx, clientStream, 5*time.Second, &logger)
 	assert.NoError(t, err)
 
 	result, err := rpcClientStream.UpdateConfiguration(ctx, version, config)

supervisor/tunnel.go

@@ -68,6 +68,8 @@ type TunnelConfig struct {
 	ProtocolSelector connection.ProtocolSelector
 	EdgeTLSConfigs   map[connection.Protocol]*tls.Config
 	PacketConfig     *ingress.GlobalRouterConfig
+
+	UDPUnregisterSessionTimeout time.Duration
 }
 
 func (c *TunnelConfig) registrationOptions(connectionID uint8, OriginLocalIP string, uuid uuid.UUID) *tunnelpogs.RegistrationOptions {
@@ -597,8 +599,8 @@ func (e *EdgeTunnelServer) serveQUIC(
 		HandshakeIdleTimeout:  quicpogs.HandshakeIdleTimeout,
 		MaxIdleTimeout:        quicpogs.MaxIdleTimeout,
 		KeepAlivePeriod:       quicpogs.MaxIdlePingPeriod,
-		MaxIncomingStreams:    connection.MaxConcurrentStreams,
-		MaxIncomingUniStreams: connection.MaxConcurrentStreams,
+		MaxIncomingStreams:    quicpogs.MaxIncomingStreams,
+		MaxIncomingUniStreams: quicpogs.MaxIncomingStreams,
 		EnableDatagrams:       true,
 		MaxDatagramFrameSize:  quicpogs.MaxDatagramFrameSize,
 		Tracer:                quicpogs.NewClientTracer(connLogger.Logger(), connIndex),
@@ -615,7 +617,9 @@ func (e *EdgeTunnelServer) serveQUIC(
 		connOptions,
 		controlStreamHandler,
 		connLogger.Logger(),
-		e.config.PacketConfig)
+		e.config.PacketConfig,
+		e.config.UDPUnregisterSessionTimeout,
+	)
 	if err != nil {
 		if e.config.NeedPQ {
 			handlePQTunnelError(err, e.config)