Release 2025.5.0

Make sure to enforce snapshots of features and client information for each connection
so that the feature information can change in the background. This allows for new features
to only be applied to a connection if it completely disconnects and attempts a reconnect.

Updates the feature refresh time to 1 hour from previous cloudflared versions which
refreshed every 6 hours.

Closes TUN-9319

.gitlab-ci.yml

@@ -0,0 +1,131 @@
+stages: [build, release]
+
+default:
+  id_tokens:
+    VAULT_ID_TOKEN:
+      aud: https://vault.cfdata.org
+
+# This before_script is injected into every job that runs on master meaning that if there is no tag the step
+# will succeed but only write "No tag present - Skipping" to the console.
+.check_tag:
+  before_script:
+    - |
+      # Check if there is a Git tag pointing to HEAD
+      echo "Tag found: $(git tag --points-at HEAD | grep .)"
+      if git tag --points-at HEAD | grep .; then
+        echo "Tag found: $(git tag --points-at HEAD | grep .)"
+        export "VERSION=$(git tag --points-at HEAD | grep .)"
+      else
+        echo "No tag present — skipping."
+        exit 0
+      fi
+
+## A set of predefined rules to use on the different jobs
+.default_rules:
+  # Rules to run the job only on the master branch
+  run_on_master:
+    - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH
+      when: always
+    - when: never
+  # Rules to run the job only on branches that are not master. This is needed because for now
+  # we need to keep a similar behavior due to the integration with teamcity, which requires us
+  # to not trigger pipelines on tags and/or merge requests.
+  run_on_branch:
+    - if: $CI_COMMIT_TAG
+      when: never
+    - if: $CI_PIPELINE_SOURCE != "merge_request_event" && $CI_COMMIT_BRANCH != $CI_DEFAULT_BRANCH
+      when: always
+    - when: never
+
+# -----------------------------------------------
+# Stage 1: Build on every PR
+# -----------------------------------------------
+build_cloudflared_macos: &build
+  stage: build
+  rules:
+    - !reference [.default_rules, run_on_branch]
+  tags:
+    - "macstadium-${RUNNER_ARCH}"
+  parallel:
+    matrix:
+      - RUNNER_ARCH: [arm, intel]
+  artifacts:
+    paths:
+      - artifacts/*
+  script:
+    - '[ "${RUNNER_ARCH}" = "arm" ] && export TARGET_ARCH=arm64'
+    - '[ "${RUNNER_ARCH}" = "intel" ] && export TARGET_ARCH=amd64'
+    - ARCH=$(uname -m)
+    - echo ARCH=$ARCH - TARGET_ARCH=$TARGET_ARCH
+    - ./.teamcity/mac/install-cloudflare-go.sh
+    - export PATH="/tmp/go/bin:$PATH"
+    - BUILD_SCRIPT=.teamcity/mac/build.sh
+    - if [[ ! -x ${BUILD_SCRIPT} ]] ; then exit ; fi
+    - set -euo pipefail
+    - echo "Executing ${BUILD_SCRIPT}"
+    - exec ${BUILD_SCRIPT}
+
+# -----------------------------------------------
+# Stage 1: Build and sign only on releases
+# -----------------------------------------------
+build_and_sign_cloudflared_macos:
+  <<: *build
+  rules:
+    - !reference [.default_rules, run_on_master]
+  secrets:
+    APPLE_DEV_CA_CERT:
+      vault: gitlab/cloudflare/tun/cloudflared/_branch/master/apple_dev_ca_cert_v2/data@kv
+      file: false
+    CFD_CODE_SIGN_CERT:
+      vault: gitlab/cloudflare/tun/cloudflared/_branch/master/cfd_code_sign_cert_v2/data@kv
+      file: false
+    CFD_CODE_SIGN_KEY:
+      vault: gitlab/cloudflare/tun/cloudflared/_branch/master/cfd_code_sign_key_v2/data@kv
+      file: false
+    CFD_CODE_SIGN_PASS:
+      vault: gitlab/cloudflare/tun/cloudflared/_branch/master/cfd_code_sign_pass_v2/data@kv
+      file: false
+    CFD_INSTALLER_CERT:
+      vault: gitlab/cloudflare/tun/cloudflared/_branch/master/cfd_installer_cert_v2/data@kv
+      file: false
+    CFD_INSTALLER_KEY:
+      vault: gitlab/cloudflare/tun/cloudflared/_branch/master/cfd_installer_key_v2/data@kv
+      file: false
+    CFD_INSTALLER_PASS:
+      vault: gitlab/cloudflare/tun/cloudflared/_branch/master/cfd_installer_pass_v2/data@kv
+      file: false
+
+# -----------------------------------------------
+# Stage 2: Release to Github after building and signing
+# -----------------------------------------------
+release_cloudflared_macos_to_github:
+  stage: release
+  image: docker-registry.cfdata.org/stash/tun/docker-images/cloudflared-ci/main:6-8616fe631b76-amd64@sha256:96f4fd05e66cec03e0864c1bcf09324c130d4728eef45ee994716da499183614
+  extends: .check_tag
+  dependencies:
+    - build_and_sign_cloudflared_macos
+  rules:
+    - !reference [.default_rules, run_on_master]
+  cache:
+    paths:
+      - .cache/pip
+  variables:
+    PIP_CACHE_DIR: "$CI_PROJECT_DIR/.cache/pip"
+    KV_NAMESPACE: 380e19aa04314648949b6ad841417ebe
+    KV_ACCOUNT: 5ab4e9dfbd435d24068829fda0077963
+  secrets:
+    KV_API_TOKEN:
+      vault: gitlab/cloudflare/tun/cloudflared/_dev/cfd_kv_api_token/data@kv
+      file: false
+    API_KEY:
+      vault: gitlab/cloudflare/tun/cloudflared/_dev/cfd_github_api_key/data@kv
+      file: false
+  script:
+    - python3 --version ; pip --version  # For debugging
+    - python3 -m venv venv
+    - source venv/bin/activate
+    - pip install pynacl==1.4.0 pygithub==1.55
+    - echo $VERSION
+    - echo $TAG_EXISTS
+    - echo "Running release because tag exists."
+    - make macos-release

.teamcity/mac/build.sh

@@ -49,7 +49,7 @@ import_certificate() {
       echo -n -e ${CERTIFICATE_ENV_VAR} | base64 -D > ${CERTIFICATE_FILE_NAME}
       # we set || true  here and for every `security import invoke` because the  "duplicate SecKeychainItemImport" error
       # will cause set -e to exit 1. It is okay we do this because we deliberately handle this error in the lines below.
-      local out=$(security import ${CERTIFICATE_FILE_NAME} -A 2>&1) || true
+      local out=$(security import ${CERTIFICATE_FILE_NAME} -T /usr/bin/pkgbuild -A 2>&1) || true
       local exitcode=$?
       # delete the certificate from disk
       rm -rf ${CERTIFICATE_FILE_NAME}
@@ -68,6 +68,28 @@ import_certificate() {
     fi
 }
 
+create_cloudflared_build_keychain() {
+  # Reusing the private key password as the keychain key
+  local PRIVATE_KEY_PASS=$1
+
+  # Create keychain only if it doesn't already exist
+  if [ ! -f "$HOME/Library/Keychains/cloudflared_build_keychain.keychain-db" ]; then
+    security create-keychain -p "$PRIVATE_KEY_PASS" cloudflared_build_keychain
+  else
+    echo "Keychain already exists: cloudflared_build_keychain"
+  fi
+
+  # Append temp keychain to the user domain
+  security list-keychains -d user -s cloudflared_build_keychain $(security list-keychains -d user | sed s/\"//g)
+
+  # Remove relock timeout
+  security set-keychain-settings cloudflared_build_keychain
+
+  # Unlock keychain so it doesn't require password
+  security unlock-keychain -p "$PRIVATE_KEY_PASS" cloudflared_build_keychain
+
+}
+
 # Imports private keys to the Apple KeyChain
 import_private_keys() {
     local PRIVATE_KEY_NAME=$1
@@ -83,7 +105,7 @@ import_private_keys() {
         echo -n -e ${PRIVATE_KEY_ENV_VAR} | base64 -D > ${PRIVATE_KEY_FILE_NAME}
         # we set || true  here and for every `security import invoke` because the  "duplicate SecKeychainItemImport" error
         # will cause set -e to exit 1. It is okay we do this because we deliberately handle this error in the lines below.
-        local out=$(security import ${PRIVATE_KEY_FILE_NAME} -A -P "${PRIVATE_KEY_PASS}" 2>&1) || true
+        local out=$(security import ${PRIVATE_KEY_FILE_NAME} -k cloudflared_build_keychain -P "$PRIVATE_KEY_PASS" -T /usr/bin/pkgbuild -A -P "${PRIVATE_KEY_PASS}" 2>&1) || true
         local exitcode=$?
         rm -rf ${PRIVATE_KEY_FILE_NAME}
         if [ -n "$out" ]; then
@@ -100,6 +122,9 @@ import_private_keys() {
     fi
 }
 
+# Create temp keychain only for this build
+create_cloudflared_build_keychain "${CFD_CODE_SIGN_PASS}"
+
 # Add Apple Root Developer certificate to the key chain
 import_certificate "Apple Developer CA" "${APPLE_DEV_CA_CERT}" "${APPLE_CA_CERT}"
 
@@ -119,8 +144,8 @@ import_certificate "Developer ID Installer" "${CFD_INSTALLER_CERT}" "${INSTALLER
 if [[ ! -z "$CFD_CODE_SIGN_NAME" ]]; then
   CODE_SIGN_NAME="${CFD_CODE_SIGN_NAME}"
 else
-  if [[ -n "$(security find-certificate -c "Developer ID Application" | cut -d'"' -f 4 -s | grep "Developer ID Application:" | head -1)" ]]; then
-    CODE_SIGN_NAME=$(security find-certificate -c "Developer ID Application" | cut -d'"' -f 4 -s | grep "Developer ID Application:" | head -1)
+  if [[ -n "$(security find-certificate -c "Developer ID Application" cloudflared_build_keychain | cut -d'"' -f 4 -s | grep "Developer ID Application:" | head -1)" ]]; then
+    CODE_SIGN_NAME=$(security find-certificate -c "Developer ID Application" cloudflared_build_keychain | cut -d'"' -f 4 -s | grep "Developer ID Application:" | head -1)
   else
     CODE_SIGN_NAME=""
   fi
@@ -130,8 +155,8 @@ fi
 if [[ ! -z "$CFD_INSTALLER_NAME" ]]; then
   PKG_SIGN_NAME="${CFD_INSTALLER_NAME}"
 else
-  if [[ -n "$(security find-certificate -c "Developer ID Installer" | cut -d'"' -f 4 -s | grep "Developer ID Installer:" | head -1)" ]]; then
-    PKG_SIGN_NAME=$(security find-certificate -c "Developer ID Installer" | cut -d'"' -f 4 -s | grep "Developer ID Installer:" | head -1)
+  if [[ -n "$(security find-certificate -c "Developer ID Installer" cloudflared_build_keychain | cut -d'"' -f 4 -s | grep "Developer ID Installer:" | head -1)" ]]; then
+    PKG_SIGN_NAME=$(security find-certificate -c "Developer ID Installer" cloudflared_build_keychain | cut -d'"' -f 4 -s | grep "Developer ID Installer:" | head -1)
   else
     PKG_SIGN_NAME=""
   fi
@@ -142,9 +167,16 @@ rm -rf "${TARGET_DIRECTORY}"
 export TARGET_OS="darwin"
 GOCACHE="$PWD/../../../../" GOPATH="$PWD/../../../../" CGO_ENABLED=1 make cloudflared
 
+
+# This allows apple tools to use the certificates in the keychain without requiring password input.
+# This command always needs to run after the certificates have been loaded into the keychain
+if [[ ! -z "$CFD_CODE_SIGN_PASS" ]]; then
+  security set-key-partition-list -S apple-tool:,apple: -s -k "${CFD_CODE_SIGN_PASS}" cloudflared_build_keychain
+fi
+
 # sign the cloudflared binary
 if [[ ! -z "$CODE_SIGN_NAME" ]]; then
-  codesign -s "${CODE_SIGN_NAME}" -f -v --timestamp --options runtime ${BINARY_NAME}
+  codesign --keychain $HOME/Library/Keychains/cloudflared_build_keychain.keychain-db -s "${CODE_SIGN_NAME}" -fv --options runtime --timestamp ${BINARY_NAME}
 
  # notarize the binary
  # TODO: TUN-5789
@@ -165,11 +197,13 @@ tar czf "$FILENAME" "${BINARY_NAME}"
 
 # build the installer package
 if [[ ! -z "$PKG_SIGN_NAME" ]]; then
+
   pkgbuild --identifier com.cloudflare.${PRODUCT} \
       --version ${VERSION} \
       --scripts ${ARCH_TARGET_DIRECTORY}/scripts \
       --root ${ARCH_TARGET_DIRECTORY}/contents \
       --install-location /usr/local/bin \
+      --keychain cloudflared_build_keychain \
       --sign "${PKG_SIGN_NAME}" \
       ${PKGNAME}
 
@@ -187,3 +221,8 @@ fi
 # cleanup build directory because this script is not ran within containers,
 # which might lead to future issues in subsequent runs.
 rm -rf "${TARGET_DIRECTORY}"
+
+# cleanup the keychain
+security default-keychain -d user -s login.keychain-db
+security list-keychains -d user -s login.keychain-db
+security delete-keychain cloudflared_build_keychain

Makefile

@@ -237,6 +237,10 @@ github-release:
 	python3 github_release.py --path $(PWD)/built_artifacts --release-version $(VERSION)
 	python3 github_message.py --release-version $(VERSION)
 
+.PHONY: macos-release
+macos-release:
+	python3 github_release.py --path $(PWD)/artifacts/ --release-version $(VERSION)
+
 .PHONY: r2-linux-release
 r2-linux-release:
 	python3 ./release_pkgs.py

RELEASE_NOTES

@@ -1,3 +1,14 @@
+2025.5.0
+- 2025-05-14 TUN-9319: Add dynamic loading of features to connections via ConnectionOptionsSnapshot
+- 2025-05-13 TUN-9322: Add metric for unsupported RPC commands for datagram v3
+- 2025-05-07 TUN-9291: Remove dynamic reloading of features for datagram v3
+
+2025.4.2
+- 2025-04-30 chore: Do not use gitlab merge request pipelines
+- 2025-04-30 DEVTOOLS-16383: Create GitlabCI pipeline to release Mac builds
+- 2025-04-24 TUN-9255: Improve flush on write conditions in http2 tunnel type to match what is done on the edge
+- 2025-04-10 SDLC-3727 - Adding FIPS status to backstage
+
 2025.4.0
 - 2025-04-02 Fix broken links in `cmd/cloudflared/*.go` related to running tunnel as a service
 - 2025-04-02 chore: remove repetitive words

catalog-info.yaml

@@ -13,3 +13,5 @@ spec:
   type: "service"
   lifecycle: "Active"
   owner: "teams/tunnel-teams-routing"
+  cf:
+    FIPS: "required"

client/config.go

@@ -0,0 +1,74 @@
+package client
+
+import (
+	"fmt"
+	"net"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/features"
+	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+)
+
+// Config captures the local client runtime configuration.
+type Config struct {
+	ConnectorID uuid.UUID
+	Version     string
+	Arch        string
+
+	featureSelector features.FeatureSelector
+}
+
+func NewConfig(version string, arch string, featureSelector features.FeatureSelector) (*Config, error) {
+	connectorID, err := uuid.NewRandom()
+	if err != nil {
+		return nil, fmt.Errorf("unable to generate a connector UUID: %w", err)
+	}
+	return &Config{
+		ConnectorID:     connectorID,
+		Version:         version,
+		Arch:            arch,
+		featureSelector: featureSelector,
+	}, nil
+}
+
+// ConnectionOptionsSnapshot is a snapshot of the current client information used to initialize a connection.
+//
+// The FeatureSnapshot is the features that are available for this connection. At the client level they may
+// change, but they will not change within the scope of this struct.
+type ConnectionOptionsSnapshot struct {
+	client              pogs.ClientInfo
+	originLocalIP       net.IP
+	numPreviousAttempts uint8
+	FeatureSnapshot     features.FeatureSnapshot
+}
+
+func (c *Config) ConnectionOptionsSnapshot(originIP net.IP, previousAttempts uint8) *ConnectionOptionsSnapshot {
+	snapshot := c.featureSelector.Snapshot()
+	return &ConnectionOptionsSnapshot{
+		client: pogs.ClientInfo{
+			ClientID: c.ConnectorID[:],
+			Version:  c.Version,
+			Arch:     c.Arch,
+			Features: snapshot.FeaturesList,
+		},
+		originLocalIP:       originIP,
+		numPreviousAttempts: previousAttempts,
+		FeatureSnapshot:     snapshot,
+	}
+}
+
+func (c ConnectionOptionsSnapshot) ConnectionOptions() *pogs.ConnectionOptions {
+	return &pogs.ConnectionOptions{
+		Client:              c.client,
+		OriginLocalIP:       c.originLocalIP,
+		ReplaceExisting:     false,
+		CompressionQuality:  0,
+		NumPreviousAttempts: c.numPreviousAttempts,
+	}
+}
+
+func (c ConnectionOptionsSnapshot) LogFields(event *zerolog.Event) *zerolog.Event {
+	return event.Strs("features", c.client.Features)
+}

client/config_test.go

@@ -0,0 +1,50 @@
+package client
+
+import (
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudflare/cloudflared/features"
+)
+
+func TestGenerateConnectionOptions(t *testing.T) {
+	version := "1234"
+	arch := "linux_amd64"
+	originIP := net.ParseIP("192.168.1.1")
+	var previousAttempts uint8 = 4
+
+	config, err := NewConfig(version, arch, &mockFeatureSelector{})
+	require.NoError(t, err)
+	require.Equal(t, version, config.Version)
+	require.Equal(t, arch, config.Arch)
+
+	// Validate ConnectionOptionsSnapshot fields
+	connOptions := config.ConnectionOptionsSnapshot(originIP, previousAttempts)
+	require.Equal(t, version, connOptions.client.Version)
+	require.Equal(t, arch, connOptions.client.Arch)
+	require.Equal(t, config.ConnectorID[:], connOptions.client.ClientID)
+
+	// Vaidate snapshot feature fields against the connOptions generated
+	snapshot := config.featureSelector.Snapshot()
+	require.Equal(t, features.DatagramV3, snapshot.DatagramVersion)
+	require.Equal(t, features.DatagramV3, connOptions.FeatureSnapshot.DatagramVersion)
+
+	pogsConnOptions := connOptions.ConnectionOptions()
+	require.Equal(t, connOptions.client, pogsConnOptions.Client)
+	require.Equal(t, originIP, pogsConnOptions.OriginLocalIP)
+	require.False(t, pogsConnOptions.ReplaceExisting)
+	require.Equal(t, uint8(0), pogsConnOptions.CompressionQuality)
+	require.Equal(t, previousAttempts, pogsConnOptions.NumPreviousAttempts)
+}
+
+type mockFeatureSelector struct{}
+
+func (m *mockFeatureSelector) Snapshot() features.FeatureSnapshot {
+	return features.FeatureSnapshot{
+		PostQuantum:     features.PostQuantumPrefer,
+		DatagramVersion: features.DatagramV3,
+		FeaturesList:    []string{features.FeaturePostQuantum, features.FeatureDatagramV3_1},
+	}
+}

cmd/cloudflared/tunnel/cmd.go

@@ -15,7 +15,6 @@ import (
 	"github.com/coreos/go-systemd/v22/daemon"
 	"github.com/facebookgo/grace/gracenet"
 	"github.com/getsentry/sentry-go"
-	"github.com/google/uuid"
 	"github.com/mitchellh/go-homedir"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
@@ -446,14 +445,7 @@ func StartServer(
 		log.Err(err).Msg("Couldn't start tunnel")
 		return err
 	}
-	var clientID uuid.UUID
-	if tunnelConfig.NamedTunnel != nil {
-		clientID, err = uuid.FromBytes(tunnelConfig.NamedTunnel.Client.ClientID)
-		if err != nil {
-			// set to nil for classic tunnels
-			clientID = uuid.Nil
-		}
-	}
+	connectorID := tunnelConfig.ClientConfig.ConnectorID
 
 	// Disable ICMP packet routing for quick tunnels
 	if quickTunnelURL != "" {
@@ -471,7 +463,7 @@ func StartServer(
 		c.String("management-hostname"),
 		c.Bool("management-diagnostics"),
 		serviceIP,
-		clientID,
+		connectorID,
 		c.String(cfdflags.ConnectorLabel),
 		logger.ManagementLogger.Log,
 		logger.ManagementLogger,
@@ -503,14 +495,14 @@ func StartServer(
 			sources = append(sources, ipv6.String())
 		}
 
-		readinessServer := metrics.NewReadyServer(clientID, tracker)
+		readinessServer := metrics.NewReadyServer(connectorID, tracker)
 		cliFlags := nonSecretCliFlags(log, c, nonSecretFlagsList)
 		diagnosticHandler := diagnostic.NewDiagnosticHandler(
 			log,
 			0,
 			diagnostic.NewSystemCollectorImpl(buildInfo.CloudflaredVersion),
 			tunnelConfig.NamedTunnel.Credentials.TunnelID,
-			clientID,
+			connectorID,
 			tracker,
 			cliFlags,
 			sources,

cmd/cloudflared/tunnel/config_test.go

@@ -1,15 +0,0 @@
-package tunnel
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/cloudflare/cloudflared/features"
-)
-
-func TestDedup(t *testing.T) {
-	expected := []string{"a", "b"}
-	actual := features.Dedup([]string{"a", "b", "a"})
-	require.ElementsMatch(t, expected, actual)
-}

cmd/cloudflared/tunnel/configuration.go

@@ -10,13 +10,13 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/urfave/cli/v2"
 	"github.com/urfave/cli/v2/altsrc"
 	"golang.org/x/term"
 
+	"github.com/cloudflare/cloudflared/client"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/cliutil"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/flags"
 	"github.com/cloudflare/cloudflared/config"
@@ -125,27 +125,29 @@ func prepareTunnelConfig(
 	observer *connection.Observer,
 	namedTunnel *connection.TunnelProperties,
 ) (*supervisor.TunnelConfig, *orchestration.Config, error) {
-	clientID, err := uuid.NewRandom()
+	transportProtocol := c.String(flags.Protocol)
+	isPostQuantumEnforced := c.Bool(flags.PostQuantum)
+	featureSelector, err := features.NewFeatureSelector(ctx, namedTunnel.Credentials.AccountTag, c.StringSlice(flags.Features), isPostQuantumEnforced, log)
 	if err != nil {
-		return nil, nil, errors.Wrap(err, "can't generate connector UUID")
+		return nil, nil, errors.Wrap(err, "Failed to create feature selector")
 	}
-	log.Info().Msgf("Generated Connector ID: %s", clientID)
+
+	clientConfig, err := client.NewConfig(info.Version(), info.OSArch(), featureSelector)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	log.Info().Msgf("Generated Connector ID: %s", clientConfig.ConnectorID)
+
 	tags, err := NewTagSliceFromCLI(c.StringSlice(flags.Tag))
 	if err != nil {
 		log.Err(err).Msg("Tag parse failure")
 		return nil, nil, errors.Wrap(err, "Tag parse failure")
 	}
-	tags = append(tags, pogs.Tag{Name: "ID", Value: clientID.String()})
+	tags = append(tags, pogs.Tag{Name: "ID", Value: clientConfig.ConnectorID.String()})
 
-	transportProtocol := c.String(flags.Protocol)
-	isPostQuantumEnforced := c.Bool(flags.PostQuantum)
-
-	featureSelector, err := features.NewFeatureSelector(ctx, namedTunnel.Credentials.AccountTag, c.StringSlice("features"), c.Bool("post-quantum"), log)
-	if err != nil {
-		return nil, nil, errors.Wrap(err, "Failed to create feature selector")
-	}
-	clientFeatures := featureSelector.ClientFeatures()
-	pqMode := featureSelector.PostQuantumMode()
+	clientFeatures := featureSelector.Snapshot()
+	pqMode := clientFeatures.PostQuantum
 	if pqMode == features.PostQuantumStrict {
 		// Error if the user tries to force a non-quic transport protocol
 		if transportProtocol != connection.AutoSelectFlag && transportProtocol != connection.QUIC.String() {
@@ -154,12 +156,6 @@ func prepareTunnelConfig(
 		transportProtocol = connection.QUIC.String()
 	}
 
-	namedTunnel.Client = pogs.ClientInfo{
-		ClientID: clientID[:],
-		Features: clientFeatures,
-		Version:  info.Version(),
-		Arch:     info.OSArch(),
-	}
 	cfg := config.GetConfiguration()
 	ingressRules, err := ingress.ParseIngressFromConfigAndCLI(cfg, c, log)
 	if err != nil {
@@ -224,10 +220,8 @@ func prepareTunnelConfig(
 	}
 
 	tunnelConfig := &supervisor.TunnelConfig{
+		ClientConfig:    clientConfig,
 		GracePeriod:     gracePeriod,
-		ReplaceExisting: c.Bool(flags.Force),
-		OSArch:          info.OSArch(),
-		ClientID:        clientID.String(),
 		EdgeAddrs:       c.StringSlice(flags.Edge),
 		Region:          resolvedRegion,
 		EdgeIPVersion:   edgeIPVersion,
@@ -246,7 +240,6 @@ func prepareTunnelConfig(
 		NamedTunnel:                         namedTunnel,
 		ProtocolSelector:                    protocolSelector,
 		EdgeTLSConfigs:                      edgeTLSConfigs,
-		FeatureSelector:                     featureSelector,
 		MaxEdgeAddrRetries:                  uint8(c.Int(flags.MaxEdgeAddrRetries)), // nolint: gosec
 		RPCTimeout:                          c.Duration(flags.RpcTimeout),
 		WriteStreamTimeout:                  c.Duration(flags.WriteStreamTimeout),

connection/connection.go

@@ -26,14 +26,20 @@ const (
 	MaxGracePeriod         = time.Minute * 3
 	MaxConcurrentStreams   = math.MaxUint32
 
-	contentTypeHeader = "content-type"
-	sseContentType    = "text/event-stream"
-	grpcContentType   = "application/grpc"
+	contentTypeHeader      = "content-type"
+	contentLengthHeader    = "content-length"
+	transferEncodingHeader = "transfer-encoding"
+
+	sseContentType     = "text/event-stream"
+	grpcContentType    = "application/grpc"
+	sseJsonContentType = "application/x-ndjson"
+
+	chunkTransferEncoding = "chunked"
 )
 
 var (
 	switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
-	flushableContentTypes = []string{sseContentType, grpcContentType}
+	flushableContentTypes = []string{sseContentType, grpcContentType, sseJsonContentType}
 )
 
 // TunnelConnection represents the connection to the edge.
@@ -51,7 +57,6 @@ type Orchestrator interface {
 
 type TunnelProperties struct {
 	Credentials    Credentials
-	Client         pogs.ClientInfo
 	QuickTunnelUrl string
 }
 
@@ -274,6 +279,22 @@ type ConnectedFuse interface {
 // Helper method to let the caller know what content-types should require a flush on every
 // write to a ResponseWriter.
 func shouldFlush(headers http.Header) bool {
+	// When doing Server Side Events (SSE), some frameworks don't respect the `Content-Type` header.
+	// Therefore, we need to rely on other ways to know whether we should flush on write or not. A good
+	// approach is to assume that responses without `Content-Length` or with `Transfer-Encoding: chunked`
+	// are streams, and therefore, should be flushed right away to the eyeball.
+	// References:
+	// - https://datatracker.ietf.org/doc/html/rfc7230#section-4.1
+	// - https://datatracker.ietf.org/doc/html/rfc9112#section-6.1
+	if contentLength := headers.Get(contentLengthHeader); contentLength == "" {
+		return true
+	}
+	if transferEncoding := headers.Get(transferEncodingHeader); transferEncoding != "" {
+		transferEncoding = strings.ToLower(transferEncoding)
+		if strings.Contains(transferEncoding, chunkTransferEncoding) {
+			return true
+		}
+	}
 	if contentType := headers.Get(contentTypeHeader); contentType != "" {
 		contentType = strings.ToLower(contentType)
 		for _, c := range flushableContentTypes {
@@ -282,7 +303,6 @@ func shouldFlush(headers http.Header) bool {
 			}
 		}
 	}
-
 	return false
 }
 

connection/connection_test.go

@@ -7,10 +7,12 @@ import (
 	"io"
 	"math/big"
 	"net/http"
+	"testing"
 	"time"
 
 	pkgerrors "github.com/pkg/errors"
 	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
 
 	cfdflow "github.com/cloudflare/cloudflared/flow"
 
@@ -209,3 +211,48 @@ func (mcf mockConnectedFuse) Connected() {}
 func (mcf mockConnectedFuse) IsConnected() bool {
 	return true
 }
+
+func TestShouldFlushHeaders(t *testing.T) {
+	tests := []struct {
+		headers     map[string]string
+		shouldFlush bool
+	}{
+		{
+			headers:     map[string]string{contentTypeHeader: "application/json", contentLengthHeader: "1"},
+			shouldFlush: false,
+		},
+		{
+			headers:     map[string]string{contentTypeHeader: "text/html", contentLengthHeader: "1"},
+			shouldFlush: false,
+		},
+		{
+			headers:     map[string]string{contentTypeHeader: "text/event-stream", contentLengthHeader: "1"},
+			shouldFlush: true,
+		},
+		{
+			headers:     map[string]string{contentTypeHeader: "application/grpc", contentLengthHeader: "1"},
+			shouldFlush: true,
+		},
+		{
+			headers:     map[string]string{contentTypeHeader: "application/x-ndjson", contentLengthHeader: "1"},
+			shouldFlush: true,
+		},
+		{
+			headers:     map[string]string{contentTypeHeader: "application/json"},
+			shouldFlush: true,
+		},
+		{
+			headers:     map[string]string{contentTypeHeader: "application/json", contentLengthHeader: "-1", transferEncodingHeader: "chunked"},
+			shouldFlush: true,
+		},
+	}
+
+	for _, test := range tests {
+		headers := http.Header{}
+		for k, v := range test.headers {
+			headers.Add(k, v)
+		}
+
+		require.Equal(t, test.shouldFlush, shouldFlush(headers))
+	}
+}

connection/control.go

@@ -10,7 +10,7 @@ import (
 
 	"github.com/cloudflare/cloudflared/management"
 	"github.com/cloudflare/cloudflared/tunnelrpc"
-	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
 // registerClient derives a named tunnel rpc client that can then be used to register and unregister connections.
@@ -36,7 +36,7 @@ type controlStream struct {
 // ControlStreamHandler registers connections with origintunneld and initiates graceful shutdown.
 type ControlStreamHandler interface {
 	// ServeControlStream handles the control plane of the transport in the current goroutine calling this
-	ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions, tunnelConfigGetter TunnelConfigJSONGetter) error
+	ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *pogs.ConnectionOptions, tunnelConfigGetter TunnelConfigJSONGetter) error
 	// IsStopped tells whether the method above has finished
 	IsStopped() bool
 }
@@ -78,7 +78,7 @@ func NewControlStream(
 func (c *controlStream) ServeControlStream(
 	ctx context.Context,
 	rw io.ReadWriteCloser,
-	connOptions *tunnelpogs.ConnectionOptions,
+	connOptions *pogs.ConnectionOptions,
 	tunnelConfigGetter TunnelConfigJSONGetter,
 ) error {
 	registrationClient := c.registerClientFunc(ctx, rw, c.registerTimeout)

connection/http2.go

@@ -16,10 +16,10 @@ import (
 	"github.com/rs/zerolog"
 	"golang.org/x/net/http2"
 
+	"github.com/cloudflare/cloudflared/client"
 	cfdflow "github.com/cloudflare/cloudflared/flow"
 
 	"github.com/cloudflare/cloudflared/tracing"
-	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
 // note: these constants are exported so we can reuse them in the edge-side code
@@ -39,7 +39,7 @@ type HTTP2Connection struct {
 	conn         net.Conn
 	server       *http2.Server
 	orchestrator Orchestrator
-	connOptions  *tunnelpogs.ConnectionOptions
+	connOptions  *client.ConnectionOptionsSnapshot
 	observer     *Observer
 	connIndex    uint8
 
@@ -54,7 +54,7 @@ type HTTP2Connection struct {
 func NewHTTP2Connection(
 	conn net.Conn,
 	orchestrator Orchestrator,
-	connOptions *tunnelpogs.ConnectionOptions,
+	connOptions *client.ConnectionOptionsSnapshot,
 	observer *Observer,
 	connIndex uint8,
 	controlStreamHandler ControlStreamHandler,
@@ -118,7 +118,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	var requestErr error
 	switch connType {
 	case TypeControlStream:
-		requestErr = c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions, c.orchestrator)
+		requestErr = c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions.ConnectionOptions(), c.orchestrator)
 		if requestErr != nil {
 			c.controlStreamErr = requestErr
 		}

connection/http2_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/http2"
 
+	"github.com/cloudflare/cloudflared/client"
 	"github.com/cloudflare/cloudflared/tracing"
 
 	"github.com/cloudflare/cloudflared/tunnelrpc"
@@ -51,7 +52,7 @@ func newTestHTTP2Connection() (*HTTP2Connection, net.Conn) {
 		cfdConn,
 		// OriginProxy is set in testConfigManager
 		testOrchestrator,
-		&pogs.ConnectionOptions{},
+		&client.ConnectionOptionsSnapshot{},
 		obs,
 		connIndex,
 		controlStream,
@@ -74,7 +75,7 @@ func TestHTTP2ConfigurationSet(t *testing.T) {
 	require.NoError(t, err)
 
 	reqBody := []byte(`{
-"version": 2, 
+"version": 2,
 "config": {"warp-routing": {"enabled": true},  "originRequest" : {"connectTimeout": 10}, "ingress" : [ {"hostname": "test", "service": "https://localhost:8000" } , {"service": "http_status:404"} ]}}
 `)
 	reader := bytes.NewReader(reqBody)

connection/quic_connection.go

@@ -17,12 +17,12 @@ import (
 	"github.com/rs/zerolog"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/cloudflare/cloudflared/client"
 	cfdflow "github.com/cloudflare/cloudflared/flow"
 
 	cfdquic "github.com/cloudflare/cloudflared/quic"
 	"github.com/cloudflare/cloudflared/tracing"
 	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
-	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 	rpcquic "github.com/cloudflare/cloudflared/tunnelrpc/quic"
 )
 
@@ -44,7 +44,7 @@ type quicConnection struct {
 	orchestrator         Orchestrator
 	datagramHandler      DatagramSessionHandler
 	controlStreamHandler ControlStreamHandler
-	connOptions          *tunnelpogs.ConnectionOptions
+	connOptions          *client.ConnectionOptionsSnapshot
 	connIndex            uint8
 
 	rpcTimeout         time.Duration
@@ -60,7 +60,7 @@ func NewTunnelConnection(
 	orchestrator Orchestrator,
 	datagramSessionHandler DatagramSessionHandler,
 	controlStreamHandler ControlStreamHandler,
-	connOptions *pogs.ConnectionOptions,
+	connOptions *client.ConnectionOptionsSnapshot,
 	rpcTimeout time.Duration,
 	streamWriteTimeout time.Duration,
 	gracePeriod time.Duration,
@@ -131,7 +131,7 @@ func (q *quicConnection) Serve(ctx context.Context) error {
 
 // serveControlStream will serve the RPC; blocking until the control plane is done.
 func (q *quicConnection) serveControlStream(ctx context.Context, controlStream quic.Stream) error {
-	return q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions, q.orchestrator)
+	return q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions.ConnectionOptions(), q.orchestrator)
 }
 
 // Close the connection with no errors specified.
@@ -235,7 +235,7 @@ func (q *quicConnection) dispatchRequest(ctx context.Context, stream *rpcquic.Re
 }
 
 // UpdateConfiguration is the RPC method invoked by edge when there is a new configuration
-func (q *quicConnection) UpdateConfiguration(ctx context.Context, version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
+func (q *quicConnection) UpdateConfiguration(ctx context.Context, version int32, config []byte) *pogs.UpdateConfigurationResponse {
 	return q.orchestrator.UpdateConfig(version, config)
 }
 

connection/quic_connection_test.go

@@ -29,6 +29,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"golang.org/x/net/nettest"
 
+	"github.com/cloudflare/cloudflared/client"
 	cfdflow "github.com/cloudflare/cloudflared/flow"
 
 	"github.com/cloudflare/cloudflared/datagramsession"
@@ -843,7 +844,7 @@ func testTunnelConnection(t *testing.T, serverAddr netip.AddrPort, index uint8)
 		&mockOrchestrator{originProxy: &mockOriginProxyWithRequest{}},
 		datagramConn,
 		fakeControlStream{},
-		&pogs.ConnectionOptions{},
+		&client.ConnectionOptionsSnapshot{},
 		15*time.Second,
 		0*time.Second,
 		0*time.Second,

connection/quic_datagram_v3.go

@@ -2,11 +2,11 @@ package connection
 
 import (
 	"context"
-	"fmt"
 	"net"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/pkg/errors"
 	"github.com/quic-go/quic-go"
 	"github.com/rs/zerolog"
 
@@ -16,10 +16,17 @@ import (
 	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
+var (
+	ErrUnsupportedRPCUDPRegistration   = errors.New("datagram v3 does not support RegisterUdpSession RPC")
+	ErrUnsupportedRPCUDPUnregistration = errors.New("datagram v3 does not support UnregisterUdpSession RPC")
+)
+
 type datagramV3Connection struct {
-	conn quic.Connection
+	conn  quic.Connection
+	index uint8
 	// datagramMuxer mux/demux datagrams from quic connection
 	datagramMuxer cfdquic.DatagramConn
+	metrics       cfdquic.Metrics
 	logger        *zerolog.Logger
 }
 
@@ -40,7 +47,9 @@ func NewDatagramV3Connection(ctx context.Context,
 
 	return &datagramV3Connection{
 		conn,
+		index,
 		datagramMuxer,
+		metrics,
 		logger,
 	}
 }
@@ -50,9 +59,11 @@ func (d *datagramV3Connection) Serve(ctx context.Context) error {
 }
 
 func (d *datagramV3Connection) RegisterUdpSession(ctx context.Context, sessionID uuid.UUID, dstIP net.IP, dstPort uint16, closeAfterIdleHint time.Duration, traceContext string) (*pogs.RegisterUdpSessionResponse, error) {
-	return nil, fmt.Errorf("datagram v3 does not support RegisterUdpSession RPC")
+	d.metrics.UnsupportedRemoteCommand(d.index, "register_udp_session")
+	return nil, ErrUnsupportedRPCUDPRegistration
 }
 
 func (d *datagramV3Connection) UnregisterUdpSession(ctx context.Context, sessionID uuid.UUID, message string) error {
-	return fmt.Errorf("datagram v3 does not support UnregisterUdpSession RPC")
+	d.metrics.UnsupportedRemoteCommand(d.index, "unregister_udp_session")
+	return ErrUnsupportedRPCUDPUnregistration
 }

datagramsession/session.go

@@ -84,7 +84,7 @@ func (s *Session) waitForCloseCondition(ctx context.Context, closeAfterIdle time
 	// Closing dstConn cancels read so dstToTransport routine in Serve() can return
 	defer s.dstConn.Close()
 	if closeAfterIdle == 0 {
-		// provide deafult is caller doesn't specify one
+		// provide default is caller doesn't specify one
 		closeAfterIdle = defaultCloseIdleAfter
 	}
 

datagramsession/session_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
 
@@ -54,22 +55,22 @@ func testSessionReturns(t *testing.T, closeBy closeMethod, closeAfterIdle time.D
 		closedByRemote, err := session.Serve(ctx, closeAfterIdle)
 		switch closeBy {
 		case closeByContext:
-			require.Equal(t, context.Canceled, err)
-			require.False(t, closedByRemote)
+			assert.Equal(t, context.Canceled, err)
+			assert.False(t, closedByRemote)
 		case closeByCallingClose:
-			require.Equal(t, localCloseReason, err)
-			require.Equal(t, localCloseReason.byRemote, closedByRemote)
+			assert.Equal(t, localCloseReason, err)
+			assert.Equal(t, localCloseReason.byRemote, closedByRemote)
 		case closeByTimeout:
-			require.Equal(t, SessionIdleErr(closeAfterIdle), err)
-			require.False(t, closedByRemote)
+			assert.Equal(t, SessionIdleErr(closeAfterIdle), err)
+			assert.False(t, closedByRemote)
 		}
 		close(sessionDone)
 	}()
 
 	go func() {
 		n, err := session.transportToDst(payload)
-		require.NoError(t, err)
-		require.Equal(t, len(payload), n)
+		assert.NoError(t, err)
+		assert.Equal(t, len(payload), n)
 	}()
 
 	readBuffer := make([]byte, len(payload)+1)
@@ -84,6 +85,8 @@ func testSessionReturns(t *testing.T, closeBy closeMethod, closeAfterIdle time.D
 		cancel()
 	case closeByCallingClose:
 		session.close(localCloseReason)
+	default:
+		// ignore
 	}
 
 	<-sessionDone
@@ -128,7 +131,7 @@ func testActiveSessionNotClosed(t *testing.T, readFromDst bool, writeToDst bool)
 	ctx, cancel := context.WithCancel(context.Background())
 	errGroup, ctx := errgroup.WithContext(ctx)
 	errGroup.Go(func() error {
-		session.Serve(ctx, closeAfterIdle)
+		_, _ = session.Serve(ctx, closeAfterIdle)
 		if time.Now().Before(startTime.Add(activeTime)) {
 			return fmt.Errorf("session closed while it's still active")
 		}

features/features.go

@@ -1,5 +1,7 @@
 package features
 
+import "slices"
+
 const (
 	FeatureSerializedHeaders = "serialized_headers"
 	FeatureQuickReconnects   = "quick_reconnects"
@@ -8,7 +10,9 @@ const (
 	FeaturePostQuantum       = "postquantum"
 	FeatureQUICSupportEOF    = "support_quic_eof"
 	FeatureManagementLogs    = "management_logs"
-	FeatureDatagramV3        = "support_datagram_v3"
+	FeatureDatagramV3_1      = "support_datagram_v3_1"
+
+	DeprecatedFeatureDatagramV3 = "support_datagram_v3" // Deprecated: TUN-9291
 )
 
 var defaultFeatures = []string{
@@ -19,11 +23,25 @@ var defaultFeatures = []string{
 	FeatureManagementLogs,
 }
 
+// List of features that are no longer in-use.
+var deprecatedFeatures = []string{
+	DeprecatedFeatureDatagramV3,
+}
+
 // Features set by user provided flags
 type staticFeatures struct {
 	PostQuantumMode *PostQuantumMode
 }
 
+type FeatureSnapshot struct {
+	PostQuantum     PostQuantumMode
+	DatagramVersion DatagramVersion
+
+	// We provide the list of features since we need it to send in the ConnectionOptions during connection
+	// registrations.
+	FeaturesList []string
+}
+
 type PostQuantumMode uint8
 
 const (
@@ -40,15 +58,19 @@ const (
 	// DatagramV2 is the currently supported datagram protocol for UDP and ICMP packets
 	DatagramV2 DatagramVersion = FeatureDatagramV2
 	// DatagramV3 is a new datagram protocol for UDP and ICMP packets. It is not backwards compatible with datagram v2.
-	DatagramV3 DatagramVersion = FeatureDatagramV3
+	DatagramV3 DatagramVersion = FeatureDatagramV3_1
 )
 
-// Remove any duplicates from the slice
-func Dedup(slice []string) []string {
+// Remove any duplicate features from the list and remove deprecated features
+func dedupAndRemoveFeatures(features []string) []string {
 	// Convert the slice into a set
-	set := make(map[string]bool, 0)
-	for _, str := range slice {
-		set[str] = true
+	set := map[string]bool{}
+	for _, feature := range features {
+		// Remove deprecated features from the provided list
+		if slices.Contains(deprecatedFeatures, feature) {
+			continue
+		}
+		set[feature] = true
 	}
 
 	// Convert the set back into a slice

features/selector.go

@@ -15,28 +15,31 @@ import (
 
 const (
 	featureSelectorHostname = "cfd-features.argotunnel.com"
-	defaultRefreshFreq      = time.Hour * 6
 	lookupTimeout           = time.Second * 10
+	defaultLookupFreq       = time.Hour
 )
 
 // If the TXT record adds other fields, the umarshal logic will ignore those keys
 // If the TXT record is missing a key, the field will unmarshal to the default Go value
 
 type featuresRecord struct {
-	// support_datagram_v3
-	DatagramV3Percentage int32 `json:"dv3"`
+	DatagramV3Percentage uint32 `json:"dv3_1"`
 
+	// DatagramV3Percentage int32 `json:"dv3"` // Removed in TUN-9291
 	// PostQuantumPercentage int32 `json:"pq"` // Removed in TUN-7970
 }
 
-func NewFeatureSelector(ctx context.Context, accountTag string, cliFeatures []string, pq bool, logger *zerolog.Logger) (*FeatureSelector, error) {
-	return newFeatureSelector(ctx, accountTag, logger, newDNSResolver(), cliFeatures, pq, defaultRefreshFreq)
+func NewFeatureSelector(ctx context.Context, accountTag string, cliFeatures []string, pq bool, logger *zerolog.Logger) (FeatureSelector, error) {
+	return newFeatureSelector(ctx, accountTag, logger, newDNSResolver(), cliFeatures, pq, defaultLookupFreq)
 }
 
-// FeatureSelector determines if this account will try new features. It periodically queries a DNS TXT record
-// to see which features are turned on.
-type FeatureSelector struct {
-	accountHash int32
+type FeatureSelector interface {
+	Snapshot() FeatureSnapshot
+}
+
+// FeatureSelector determines if this account will try new features; loaded once during startup.
+type featureSelector struct {
+	accountHash uint32
 	logger      *zerolog.Logger
 	resolver    resolver
 
@@ -44,11 +47,11 @@ type FeatureSelector struct {
 	cliFeatures    []string
 
 	// lock protects concurrent access to dynamic features
-	lock     sync.RWMutex
-	features featuresRecord
+	lock           sync.RWMutex
+	remoteFeatures featuresRecord
 }
 
-func newFeatureSelector(ctx context.Context, accountTag string, logger *zerolog.Logger, resolver resolver, cliFeatures []string, pq bool, refreshFreq time.Duration) (*FeatureSelector, error) {
+func newFeatureSelector(ctx context.Context, accountTag string, logger *zerolog.Logger, resolver resolver, cliFeatures []string, pq bool, refreshFreq time.Duration) (*featureSelector, error) {
 	// Combine default features and user-provided features
 	var pqMode *PostQuantumMode
 	if pq {
@@ -59,28 +62,40 @@ func newFeatureSelector(ctx context.Context, accountTag string, logger *zerolog.
 	staticFeatures := staticFeatures{
 		PostQuantumMode: pqMode,
 	}
-	selector := &FeatureSelector{
+	selector := &featureSelector{
 		accountHash:    switchThreshold(accountTag),
 		logger:         logger,
 		resolver:       resolver,
 		staticFeatures: staticFeatures,
-		cliFeatures:    Dedup(cliFeatures),
+		cliFeatures:    dedupAndRemoveFeatures(cliFeatures),
 	}
 
+	// Load the remote features
 	if err := selector.refresh(ctx); err != nil {
 		logger.Err(err).Msg("Failed to fetch features, default to disable")
 	}
 
+	// Spin off reloading routine
 	go selector.refreshLoop(ctx, refreshFreq)
 
 	return selector, nil
 }
 
-func (fs *FeatureSelector) accountEnabled(percentage int32) bool {
+func (fs *featureSelector) Snapshot() FeatureSnapshot {
+	fs.lock.RLock()
+	defer fs.lock.RUnlock()
+	return FeatureSnapshot{
+		PostQuantum:     fs.postQuantumMode(),
+		DatagramVersion: fs.datagramVersion(),
+		FeaturesList:    fs.clientFeatures(),
+	}
+}
+
+func (fs *featureSelector) accountEnabled(percentage uint32) bool {
 	return percentage > fs.accountHash
 }
 
-func (fs *FeatureSelector) PostQuantumMode() PostQuantumMode {
+func (fs *featureSelector) postQuantumMode() PostQuantumMode {
 	if fs.staticFeatures.PostQuantumMode != nil {
 		return *fs.staticFeatures.PostQuantumMode
 	}
@@ -88,12 +103,9 @@ func (fs *FeatureSelector) PostQuantumMode() PostQuantumMode {
 	return PostQuantumPrefer
 }
 
-func (fs *FeatureSelector) DatagramVersion() DatagramVersion {
-	fs.lock.RLock()
-	defer fs.lock.RUnlock()
-
+func (fs *featureSelector) datagramVersion() DatagramVersion {
 	// If user provides the feature via the cli, we take it as priority over remote feature evaluation
-	if slices.Contains(fs.cliFeatures, FeatureDatagramV3) {
+	if slices.Contains(fs.cliFeatures, FeatureDatagramV3_1) {
 		return DatagramV3
 	}
 	// If the user specifies DatagramV2, we also take that over remote
@@ -101,36 +113,20 @@ func (fs *FeatureSelector) DatagramVersion() DatagramVersion {
 		return DatagramV2
 	}
 
-	if fs.accountEnabled(fs.features.DatagramV3Percentage) {
+	if fs.accountEnabled(fs.remoteFeatures.DatagramV3Percentage) {
 		return DatagramV3
 	}
+
 	return DatagramV2
 }
 
-// ClientFeatures will return the list of currently available features that cloudflared should provide to the edge.
-//
-// This list is dynamic and can change in-between returns.
-func (fs *FeatureSelector) ClientFeatures() []string {
+// clientFeatures will return the list of currently available features that cloudflared should provide to the edge.
+func (fs *featureSelector) clientFeatures() []string {
 	// Evaluate any remote features along with static feature list to construct the list of features
-	return Dedup(slices.Concat(defaultFeatures, fs.cliFeatures, []string{string(fs.DatagramVersion())}))
+	return dedupAndRemoveFeatures(slices.Concat(defaultFeatures, fs.cliFeatures, []string{string(fs.datagramVersion())}))
 }
 
-func (fs *FeatureSelector) refreshLoop(ctx context.Context, refreshFreq time.Duration) {
-	ticker := time.NewTicker(refreshFreq)
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-			err := fs.refresh(ctx)
-			if err != nil {
-				fs.logger.Err(err).Msg("Failed to refresh feature selector")
-			}
-		}
-	}
-}
-
-func (fs *FeatureSelector) refresh(ctx context.Context) error {
+func (fs *featureSelector) refresh(ctx context.Context) error {
 	record, err := fs.resolver.lookupRecord(ctx)
 	if err != nil {
 		return err
@@ -144,11 +140,26 @@ func (fs *FeatureSelector) refresh(ctx context.Context) error {
 	fs.lock.Lock()
 	defer fs.lock.Unlock()
 
-	fs.features = features
+	fs.remoteFeatures = features
 
 	return nil
 }
 
+func (fs *featureSelector) refreshLoop(ctx context.Context, refreshFreq time.Duration) {
+	ticker := time.NewTicker(refreshFreq)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			err := fs.refresh(ctx)
+			if err != nil {
+				fs.logger.Err(err).Msg("Failed to refresh feature selector")
+			}
+		}
+	}
+}
+
 // resolver represents an object that can look up featuresRecord
 type resolver interface {
 	lookupRecord(ctx context.Context) ([]byte, error)
@@ -180,8 +191,8 @@ func (dr *dnsResolver) lookupRecord(ctx context.Context) ([]byte, error) {
 	return []byte(records[0]), nil
 }
 
-func switchThreshold(accountTag string) int32 {
+func switchThreshold(accountTag string) uint32 {
 	h := fnv.New32a()
 	_, _ = h.Write([]byte(accountTag))
-	return int32(h.Sum32() % 100)
+	return h.Sum32() % 100
 }

features/selector_test.go

@@ -11,21 +11,26 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+const (
+	testAccountTag  = "123456"
+	testAccountHash = 74 // switchThreshold of `accountTag`
+)
+
 func TestUnmarshalFeaturesRecord(t *testing.T) {
 	tests := []struct {
 		record             []byte
-		expectedPercentage int32
+		expectedPercentage uint32
 	}{
 		{
-			record:             []byte(`{"dv3":0}`),
+			record:             []byte(`{"dv3_1":0}`),
 			expectedPercentage: 0,
 		},
 		{
-			record:             []byte(`{"dv3":39}`),
+			record:             []byte(`{"dv3_1":39}`),
 			expectedPercentage: 39,
 		},
 		{
-			record:             []byte(`{"dv3":100}`),
+			record:             []byte(`{"dv3_1":100}`),
 			expectedPercentage: 100,
 		},
 		{
@@ -34,6 +39,9 @@ func TestUnmarshalFeaturesRecord(t *testing.T) {
 		{
 			record: []byte(`{"kyber":768}`), // Unmarshal to default struct if key is not present
 		},
+		{
+			record: []byte(`{"pq": 101,"dv3":100}`), // Expired keys don't unmarshal to anything
+		},
 	}
 
 	for _, test := range tests {
@@ -61,7 +69,7 @@ func TestFeaturePrecedenceEvaluationPostQuantum(t *testing.T) {
 		{
 			name:             "user_specified",
 			cli:              true,
-			expectedFeatures: Dedup(append(defaultFeatures, FeaturePostQuantum)),
+			expectedFeatures: dedupAndRemoveFeatures(append(defaultFeatures, FeaturePostQuantum)),
 			expectedVersion:  PostQuantumStrict,
 		},
 	}
@@ -71,8 +79,9 @@ func TestFeaturePrecedenceEvaluationPostQuantum(t *testing.T) {
 			resolver := &staticResolver{record: featuresRecord{}}
 			selector, err := newFeatureSelector(context.Background(), test.name, &logger, resolver, []string{}, test.cli, time.Second)
 			require.NoError(t, err)
-			require.ElementsMatch(t, test.expectedFeatures, selector.ClientFeatures())
-			require.Equal(t, test.expectedVersion, selector.PostQuantumMode())
+			snapshot := selector.Snapshot()
+			require.ElementsMatch(t, test.expectedFeatures, snapshot.FeaturesList)
+			require.Equal(t, test.expectedVersion, snapshot.PostQuantum)
 		})
 	}
 }
@@ -102,37 +111,10 @@ func TestFeaturePrecedenceEvaluationDatagramVersion(t *testing.T) {
 		},
 		{
 			name:             "user_specified_v3",
-			cli:              []string{FeatureDatagramV3},
+			cli:              []string{FeatureDatagramV3_1},
 			remote:           featuresRecord{},
-			expectedFeatures: Dedup(append(defaultFeatures, FeatureDatagramV3)),
-			expectedVersion:  FeatureDatagramV3,
-		},
-		{
-			name: "remote_specified_v3",
-			cli:  []string{},
-			remote: featuresRecord{
-				DatagramV3Percentage: 100,
-			},
-			expectedFeatures: Dedup(append(defaultFeatures, FeatureDatagramV3)),
-			expectedVersion:  FeatureDatagramV3,
-		},
-		{
-			name: "remote_and_user_specified_v3",
-			cli:  []string{FeatureDatagramV3},
-			remote: featuresRecord{
-				DatagramV3Percentage: 100,
-			},
-			expectedFeatures: Dedup(append(defaultFeatures, FeatureDatagramV3)),
-			expectedVersion:  FeatureDatagramV3,
-		},
-		{
-			name: "remote_v3_and_user_specified_v2",
-			cli:  []string{FeatureDatagramV2},
-			remote: featuresRecord{
-				DatagramV3Percentage: 100,
-			},
-			expectedFeatures: defaultFeatures,
-			expectedVersion:  DatagramV2,
+			expectedFeatures: dedupAndRemoveFeatures(append(defaultFeatures, FeatureDatagramV3_1)),
+			expectedVersion:  FeatureDatagramV3_1,
 		},
 	}
 
@@ -141,58 +123,108 @@ func TestFeaturePrecedenceEvaluationDatagramVersion(t *testing.T) {
 			resolver := &staticResolver{record: test.remote}
 			selector, err := newFeatureSelector(context.Background(), test.name, &logger, resolver, test.cli, false, time.Second)
 			require.NoError(t, err)
-			require.ElementsMatch(t, test.expectedFeatures, selector.ClientFeatures())
-			require.Equal(t, test.expectedVersion, selector.DatagramVersion())
+			snapshot := selector.Snapshot()
+			require.ElementsMatch(t, test.expectedFeatures, snapshot.FeaturesList)
+			require.Equal(t, test.expectedVersion, snapshot.DatagramVersion)
+		})
+	}
+}
+
+func TestDeprecatedFeaturesRemoved(t *testing.T) {
+	logger := zerolog.Nop()
+	tests := []struct {
+		name             string
+		cli              []string
+		remote           featuresRecord
+		expectedFeatures []string
+	}{
+		{
+			name:             "no_removals",
+			cli:              []string{},
+			remote:           featuresRecord{},
+			expectedFeatures: defaultFeatures,
+		},
+		{
+			name:             "support_datagram_v3",
+			cli:              []string{DeprecatedFeatureDatagramV3},
+			remote:           featuresRecord{},
+			expectedFeatures: defaultFeatures,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			resolver := &staticResolver{record: test.remote}
+			selector, err := newFeatureSelector(context.Background(), test.name, &logger, resolver, test.cli, false, time.Second)
+			require.NoError(t, err)
+			snapshot := selector.Snapshot()
+			require.ElementsMatch(t, test.expectedFeatures, snapshot.FeaturesList)
 		})
 	}
 }
 
 func TestRefreshFeaturesRecord(t *testing.T) {
-	// The hash of the accountTag is 82
-	accountTag := t.Name()
-	threshold := switchThreshold(accountTag)
-
-	percentages := []int32{0, 10, 81, 82, 83, 100, 101, 1000}
-	refreshFreq := time.Millisecond * 10
-	selector := newTestSelector(t, percentages, false, refreshFreq)
+	percentages := []uint32{0, 10, testAccountHash - 1, testAccountHash, testAccountHash + 1, 100, 101, 1000}
+	selector := newTestSelector(t, percentages, false, time.Minute)
 
 	// Starting out should default to DatagramV2
-	require.Equal(t, DatagramV2, selector.DatagramVersion())
+	snapshot := selector.Snapshot()
+	require.Equal(t, DatagramV2, snapshot.DatagramVersion)
 
 	for _, percentage := range percentages {
-		if percentage > threshold {
-			require.Equal(t, DatagramV3, selector.DatagramVersion())
+		snapshot = selector.Snapshot()
+		if percentage > testAccountHash {
+			require.Equal(t, DatagramV3, snapshot.DatagramVersion)
 		} else {
-			require.Equal(t, DatagramV2, selector.DatagramVersion())
+			require.Equal(t, DatagramV2, snapshot.DatagramVersion)
 		}
 
-		time.Sleep(refreshFreq + time.Millisecond)
+		// Manually progress the next refresh
+		_ = selector.refresh(context.Background())
 	}
 
-	// Make sure error doesn't override the last fetched features
-	require.Equal(t, DatagramV3, selector.DatagramVersion())
+	// Make sure a resolver error doesn't override the last fetched features
+	snapshot = selector.Snapshot()
+	require.Equal(t, DatagramV3, snapshot.DatagramVersion)
+}
+
+func TestSnapshotIsolation(t *testing.T) {
+	percentages := []uint32{testAccountHash, testAccountHash + 1}
+	selector := newTestSelector(t, percentages, false, time.Minute)
+
+	// Starting out should default to DatagramV2
+	snapshot := selector.Snapshot()
+	require.Equal(t, DatagramV2, snapshot.DatagramVersion)
+
+	// Manually progress the next refresh
+	_ = selector.refresh(context.Background())
+
+	snapshot2 := selector.Snapshot()
+	require.Equal(t, DatagramV3, snapshot2.DatagramVersion)
+	require.NotEqual(t, snapshot.DatagramVersion, snapshot2.DatagramVersion)
 }
 
 func TestStaticFeatures(t *testing.T) {
-	percentages := []int32{0}
+	percentages := []uint32{0}
 	// PostQuantum Enabled from user flag
-	selector := newTestSelector(t, percentages, true, time.Millisecond*10)
-	require.Equal(t, PostQuantumStrict, selector.PostQuantumMode())
+	selector := newTestSelector(t, percentages, true, time.Second)
+	snapshot := selector.Snapshot()
+	require.Equal(t, PostQuantumStrict, snapshot.PostQuantum)
 
 	// PostQuantum Disabled (or not set)
-	selector = newTestSelector(t, percentages, false, time.Millisecond*10)
-	require.Equal(t, PostQuantumPrefer, selector.PostQuantumMode())
+	selector = newTestSelector(t, percentages, false, time.Second)
+	snapshot = selector.Snapshot()
+	require.Equal(t, PostQuantumPrefer, snapshot.PostQuantum)
 }
 
-func newTestSelector(t *testing.T, percentages []int32, pq bool, refreshFreq time.Duration) *FeatureSelector {
-	accountTag := t.Name()
+func newTestSelector(t *testing.T, percentages []uint32, pq bool, refreshFreq time.Duration) *featureSelector {
 	logger := zerolog.Nop()
 
 	resolver := &mockResolver{
 		percentages: percentages,
 	}
 
-	selector, err := newFeatureSelector(context.Background(), accountTag, &logger, resolver, []string{}, pq, refreshFreq)
+	selector, err := newFeatureSelector(context.Background(), testAccountTag, &logger, resolver, []string{}, pq, refreshFreq)
 	require.NoError(t, err)
 
 	return selector
@@ -200,7 +232,7 @@ func newTestSelector(t *testing.T, percentages []int32, pq bool, refreshFreq tim
 
 type mockResolver struct {
 	nextIndex   int
-	percentages []int32
+	percentages []uint32
 }
 
 func (mr *mockResolver) lookupRecord(ctx context.Context) ([]byte, error) {

github_release.py

@@ -61,7 +61,7 @@ def assert_tag_exists(repo, version):
         raise Exception("Tag {} not found".format(version))
 
 
-def get_or_create_release(repo, version, dry_run=False):
+def get_or_create_release(repo, version, dry_run=False, is_draft=False):
     """
     Get a Github Release matching the version tag or create a new one.
     If a conflict occurs on creation, attempt to fetch the Release on last time
@@ -81,8 +81,11 @@ def get_or_create_release(repo, version, dry_run=False):
         return
 
     try:
-        logging.info("Creating release %s", version)
-        return repo.create_git_release(version, version, "")
+        if is_draft:
+            logging.info("Drafting release %s", version)
+        else:
+            logging.info("Creating release %s", version)
+        return repo.create_git_release(version, version, "", is_draft)
     except GithubException as e:
         errors = e.data.get("errors", [])
         if e.status == 422 and any(
@@ -129,6 +132,10 @@ def parse_args():
         "--dry-run", action="store_true", help="Do not create release or upload asset"
     )
 
+    parser.add_argument(
+        "--draft", action="store_true", help="Create a draft release"
+    )
+
     args = parser.parse_args()
     is_valid = True
     if not args.release_version:
@@ -292,7 +299,7 @@ def main():
                 for filename in onlyfiles:
                     binary_path = os.path.join(args.path, filename)
                     assert_asset_version(binary_path, args.release_version)
-                release = get_or_create_release(repo, args.release_version, args.dry_run)
+                release = get_or_create_release(repo, args.release_version, args.dry_run, args.draft)
                 for filename in onlyfiles:
                     binary_path = os.path.join(args.path, filename)
                     upload_asset(release, binary_path, filename, args.release_version, args.kv_account_id, args.namespace_id,

quic/metrics.go

@@ -11,12 +11,15 @@ import (
 )
 
 const (
-	namespace = "quic"
+	namespace                  = "quic"
+	ConnectionIndexMetricLabel = "conn_index"
+	frameTypeMetricLabel       = "frame_type"
+	packetTypeMetricLabel      = "packet_type"
+	reasonMetricLabel          = "reason"
 )
 
 var (
-	clientConnLabels = []string{"conn_index"}
-	clientMetrics    = struct {
+	clientMetrics = struct {
 		totalConnections  prometheus.Counter
 		closedConnections prometheus.Counter
 		maxUDPPayloadSize *prometheus.GaugeVec
@@ -35,7 +38,7 @@ var (
 		congestionState   *prometheus.GaugeVec
 	}{
 		totalConnections: prometheus.NewCounter(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "total_connections",
@@ -43,7 +46,7 @@ var (
 			},
 		),
 		closedConnections: prometheus.NewCounter(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "closed_connections",
@@ -57,70 +60,70 @@ var (
 				Name:      "max_udp_payload",
 				Help:      "Maximum UDP payload size in bytes for a QUIC packet",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 		sentFrames: prometheus.NewCounterVec(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "sent_frames",
 				Help:      "Number of frames that have been sent through a connection",
 			},
-			append(clientConnLabels, "frame_type"),
+			[]string{ConnectionIndexMetricLabel, frameTypeMetricLabel},
 		),
 		sentBytes: prometheus.NewCounterVec(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "sent_bytes",
 				Help:      "Number of bytes that have been sent through a connection",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 		receivedFrames: prometheus.NewCounterVec(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "received_frames",
 				Help:      "Number of frames that have been received through a connection",
 			},
-			append(clientConnLabels, "frame_type"),
+			[]string{ConnectionIndexMetricLabel, frameTypeMetricLabel},
 		),
 		receivedBytes: prometheus.NewCounterVec(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "receive_bytes",
 				Help:      "Number of bytes that have been received through a connection",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 		bufferedPackets: prometheus.NewCounterVec(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "buffered_packets",
 				Help:      "Number of bytes that have been buffered on a connection",
 			},
-			append(clientConnLabels, "packet_type"),
+			[]string{ConnectionIndexMetricLabel, packetTypeMetricLabel},
 		),
 		droppedPackets: prometheus.NewCounterVec(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "dropped_packets",
 				Help:      "Number of bytes that have been dropped on a connection",
 			},
-			append(clientConnLabels, "packet_type", "reason"),
+			[]string{ConnectionIndexMetricLabel, packetTypeMetricLabel, reasonMetricLabel},
 		),
 		lostPackets: prometheus.NewCounterVec(
-			prometheus.CounterOpts{
+			prometheus.CounterOpts{ //nolint:promlinter
 				Namespace: namespace,
 				Subsystem: "client",
 				Name:      "lost_packets",
 				Help:      "Number of packets that have been lost from a connection",
 			},
-			append(clientConnLabels, "reason"),
+			[]string{ConnectionIndexMetricLabel, reasonMetricLabel},
 		),
 		minRTT: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
@@ -129,7 +132,7 @@ var (
 				Name:      "min_rtt",
 				Help:      "Lowest RTT measured on a connection in millisec",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 		latestRTT: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
@@ -138,7 +141,7 @@ var (
 				Name:      "latest_rtt",
 				Help:      "Latest RTT measured on a connection",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 		smoothedRTT: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
@@ -147,7 +150,7 @@ var (
 				Name:      "smoothed_rtt",
 				Help:      "Calculated smoothed RTT measured on a connection in millisec",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 		mtu: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
@@ -156,7 +159,7 @@ var (
 				Name:      "mtu",
 				Help:      "Current maximum transmission unit (MTU) of a connection",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 		congestionWindow: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
@@ -165,7 +168,7 @@ var (
 				Name:      "congestion_window",
 				Help:      "Current congestion window size",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 		congestionState: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
@@ -174,13 +177,13 @@ var (
 				Name:      "congestion_state",
 				Help:      "Current congestion control state. See https://pkg.go.dev/github.com/quic-go/quic-go@v0.45.0/logging#CongestionState for what each value maps to",
 			},
-			clientConnLabels,
+			[]string{ConnectionIndexMetricLabel},
 		),
 	}
 
 	registerClient = sync.Once{}
 
-	packetTooBigDropped = prometheus.NewCounter(prometheus.CounterOpts{
+	packetTooBigDropped = prometheus.NewCounter(prometheus.CounterOpts{ //nolint:promlinter
 		Namespace: namespace,
 		Subsystem: "client",
 		Name:      "packet_too_big_dropped",

quic/v3/metrics.go

@@ -1,83 +1,101 @@
 package v3
 
 import (
+	"fmt"
+
 	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/cloudflare/cloudflared/quic"
 )
 
 const (
 	namespace = "cloudflared"
 	subsystem = "udp"
+
+	commandMetricLabel = "command"
 )
 
 type Metrics interface {
-	IncrementFlows()
-	DecrementFlows()
-	PayloadTooLarge()
-	RetryFlowResponse()
-	MigrateFlow()
+	IncrementFlows(connIndex uint8)
+	DecrementFlows(connIndex uint8)
+	PayloadTooLarge(connIndex uint8)
+	RetryFlowResponse(connIndex uint8)
+	MigrateFlow(connIndex uint8)
+	UnsupportedRemoteCommand(connIndex uint8, command string)
 }
 
 type metrics struct {
-	activeUDPFlows     prometheus.Gauge
-	totalUDPFlows      prometheus.Counter
-	payloadTooLarge    prometheus.Counter
-	retryFlowResponses prometheus.Counter
-	migratedFlows      prometheus.Counter
+	activeUDPFlows            *prometheus.GaugeVec
+	totalUDPFlows             *prometheus.CounterVec
+	payloadTooLarge           *prometheus.CounterVec
+	retryFlowResponses        *prometheus.CounterVec
+	migratedFlows             *prometheus.CounterVec
+	unsupportedRemoteCommands *prometheus.CounterVec
 }
 
-func (m *metrics) IncrementFlows() {
-	m.totalUDPFlows.Inc()
-	m.activeUDPFlows.Inc()
+func (m *metrics) IncrementFlows(connIndex uint8) {
+	m.totalUDPFlows.WithLabelValues(fmt.Sprintf("%d", connIndex)).Inc()
+	m.activeUDPFlows.WithLabelValues(fmt.Sprintf("%d", connIndex)).Inc()
 }
 
-func (m *metrics) DecrementFlows() {
-	m.activeUDPFlows.Dec()
+func (m *metrics) DecrementFlows(connIndex uint8) {
+	m.activeUDPFlows.WithLabelValues(fmt.Sprintf("%d", connIndex)).Dec()
 }
 
-func (m *metrics) PayloadTooLarge() {
-	m.payloadTooLarge.Inc()
+func (m *metrics) PayloadTooLarge(connIndex uint8) {
+	m.payloadTooLarge.WithLabelValues(fmt.Sprintf("%d", connIndex)).Inc()
 }
 
-func (m *metrics) RetryFlowResponse() {
-	m.retryFlowResponses.Inc()
+func (m *metrics) RetryFlowResponse(connIndex uint8) {
+	m.retryFlowResponses.WithLabelValues(fmt.Sprintf("%d", connIndex)).Inc()
 }
 
-func (m *metrics) MigrateFlow() {
-	m.migratedFlows.Inc()
+func (m *metrics) MigrateFlow(connIndex uint8) {
+	m.migratedFlows.WithLabelValues(fmt.Sprintf("%d", connIndex)).Inc()
+}
+
+func (m *metrics) UnsupportedRemoteCommand(connIndex uint8, command string) {
+	m.unsupportedRemoteCommands.WithLabelValues(fmt.Sprintf("%d", connIndex), command).Inc()
 }
 
 func NewMetrics(registerer prometheus.Registerer) Metrics {
 	m := &metrics{
-		activeUDPFlows: prometheus.NewGauge(prometheus.GaugeOpts{
+		activeUDPFlows: prometheus.NewGaugeVec(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{
+		}, []string{quic.ConnectionIndexMetricLabel}),
+		totalUDPFlows: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
 			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{
+		}, []string{quic.ConnectionIndexMetricLabel}),
+		payloadTooLarge: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
 			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{
+		}, []string{quic.ConnectionIndexMetricLabel}),
+		retryFlowResponses: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
 			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{
+		}, []string{quic.ConnectionIndexMetricLabel}),
+		migratedFlows: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
 			Namespace: namespace,
 			Subsystem: subsystem,
 			Name:      "migrated_flows",
 			Help:      "Total count of UDP flows have been migrated across local connections",
-		}),
+		}, []string{quic.ConnectionIndexMetricLabel}),
+		unsupportedRemoteCommands: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "unsupported_remote_command_total",
+			Help:      "Total count of unsupported remote RPC commands for the ",
+		}, []string{quic.ConnectionIndexMetricLabel, commandMetricLabel}),
 	}
 	registerer.MustRegister(
 		m.activeUDPFlows,
@@ -85,6 +103,7 @@ func NewMetrics(registerer prometheus.Registerer) Metrics {
 		m.payloadTooLarge,
 		m.retryFlowResponses,
 		m.migratedFlows,
+		m.unsupportedRemoteCommands,
 	)
 	return m
 }

quic/v3/metrics_test.go

@@ -2,8 +2,9 @@ package v3_test
 
 type noopMetrics struct{}
 
-func (noopMetrics) IncrementFlows()    {}
-func (noopMetrics) DecrementFlows()    {}
-func (noopMetrics) PayloadTooLarge()   {}
-func (noopMetrics) RetryFlowResponse() {}
-func (noopMetrics) MigrateFlow()       {}
+func (noopMetrics) IncrementFlows(connIndex uint8)                           {}
+func (noopMetrics) DecrementFlows(connIndex uint8)                           {}
+func (noopMetrics) PayloadTooLarge(connIndex uint8)                          {}
+func (noopMetrics) RetryFlowResponse(connIndex uint8)                        {}
+func (noopMetrics) MigrateFlow(connIndex uint8)                              {}
+func (noopMetrics) UnsupportedRemoteCommand(connIndex uint8, command string) {}

quic/v3/muxer.go

@@ -264,10 +264,10 @@ func (c *datagramConn) handleSessionRegistrationDatagram(ctx context.Context, da
 		return
 	}
 	log = log.With().Str(logSrcKey, session.LocalAddr().String()).Logger()
-	c.metrics.IncrementFlows()
+	c.metrics.IncrementFlows(c.index)
 	// 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()
+	defer c.metrics.DecrementFlows(c.index)
 
 	// Respond that we are able to process the new session
 	err = c.SendUDPSessionResponse(datagram.RequestID, ResponseOk)
@@ -315,7 +315,7 @@ func (c *datagramConn) handleSessionAlreadyRegistered(requestID RequestID, logge
 	// 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()
+	c.metrics.RetryFlowResponse(c.index)
 	logger.Debug().Msgf("flow registration response retry")
 }
 

quic/v3/muxer_test.go

@@ -781,12 +781,12 @@ func newICMPDatagram(pk *packet.ICMP) []byte {
 
 // 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)
+	cancel(errExpectedContextCanceled)
 	err := <-done
 	if !errors.Is(err, context.Canceled) {
 		t.Fatal(err)
 	}
-	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+	if !errors.Is(context.Cause(ctx), errExpectedContextCanceled) {
 		t.Fatal(err)
 	}
 }

quic/v3/session.go

@@ -27,11 +27,11 @@ const (
 )
 
 // SessionCloseErr indicates that the session's Close method was called.
-var SessionCloseErr error = errors.New("flow was closed directly")
+var SessionCloseErr error = errors.New("flow was closed directly") //nolint:errname
 
 // SessionIdleErr is returned when the session was closed because there was no communication
 // in either direction over the session for the timeout period.
-type SessionIdleErr struct {
+type SessionIdleErr struct { //nolint:errname
 	timeout time.Duration
 }
 
@@ -149,7 +149,8 @@ func (s *session) Migrate(eyeball DatagramConn, ctx context.Context, logger *zer
 	}
 	// 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()
+	connectionIndex := eyeball.ID()
+	s.metrics.MigrateFlow(connectionIndex)
 }
 
 func (s *session) Serve(ctx context.Context) error {
@@ -160,7 +161,7 @@ func (s *session) Serve(ctx context.Context) error {
 		// To perform a zero copy write when passing the datagram to the connection, we prepare the buffer with
 		// the required datagram header information. We can reuse this buffer for this session since the header is the
 		// same for the each read.
-		MarshalPayloadHeaderTo(s.id, readBuffer[:DatagramPayloadHeaderLen])
+		_ = MarshalPayloadHeaderTo(s.id, readBuffer[:DatagramPayloadHeaderLen])
 		for {
 			// Read from the origin UDP socket
 			n, err := s.origin.Read(readBuffer[DatagramPayloadHeaderLen:])
@@ -177,7 +178,8 @@ func (s *session) Serve(ctx context.Context) error {
 				continue
 			}
 			if n > maxDatagramPayloadLen {
-				s.metrics.PayloadTooLarge()
+				connectionIndex := s.ConnectionID()
+				s.metrics.PayloadTooLarge(connectionIndex)
 				s.log.Error().Int(logPacketSizeKey, n).Msg("flow (origin) packet read was too large and was dropped")
 				continue
 			}
@@ -241,7 +243,7 @@ func (s *session) waitForCloseCondition(ctx context.Context, closeAfterIdle time
 	// Closing the session at the end cancels read so Serve() can return
 	defer s.Close()
 	if closeAfterIdle == 0 {
-		// provide deafult is caller doesn't specify one
+		// Provided that the default caller doesn't specify one
 		closeAfterIdle = defaultCloseIdleAfter
 	}
 

quic/v3/session_test.go

@@ -17,7 +17,7 @@ import (
 )
 
 var (
-	expectedContextCanceled = errors.New("expected context canceled")
+	errExpectedContextCanceled = 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"))
@@ -40,7 +40,7 @@ func testSessionWrite(t *testing.T, payload []byte) {
 	serverRead := make(chan []byte, 1)
 	go func() {
 		read := make([]byte, 1500)
-		server.Read(read[:])
+		_, _ = server.Read(read[:])
 		serverRead <- read
 	}()
 	// Create session and write to origin
@@ -110,12 +110,12 @@ func testSessionServe_Origin(t *testing.T, payload []byte) {
 	case data := <-eyeball.recvData:
 		// check received data matches provided from origin
 		expectedData := makePayload(1500)
-		v3.MarshalPayloadHeaderTo(testRequestID, expectedData[:])
+		_ = v3.MarshalPayloadHeaderTo(testRequestID, expectedData[:])
 		copy(expectedData[17:], payload)
 		if !slices.Equal(expectedData[:v3.DatagramPayloadHeaderLen+len(payload)], data) {
 			t.Fatal("expected datagram did not equal expected")
 		}
-		cancel(expectedContextCanceled)
+		cancel(errExpectedContextCanceled)
 	case err := <-ctx.Done():
 		// we expect the payload to return before the context to cancel on the session
 		t.Fatal(err)
@@ -125,7 +125,7 @@ func testSessionServe_Origin(t *testing.T, payload []byte) {
 	if !errors.Is(err, context.Canceled) {
 		t.Fatal(err)
 	}
-	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+	if !errors.Is(context.Cause(ctx), errExpectedContextCanceled) {
 		t.Fatal(err)
 	}
 }
@@ -198,7 +198,7 @@ func TestSessionServe_Migrate(t *testing.T) {
 
 	// Origin sends data
 	payload2 := []byte{0xde}
-	pipe1.Write(payload2)
+	_, _ = pipe1.Write(payload2)
 
 	// Expect write to eyeball2
 	data := <-eyeball2.recvData
@@ -249,13 +249,13 @@ func TestSessionServe_Migrate_CloseContext2(t *testing.T) {
 		t.Fatalf("expected session to still be running")
 	default:
 	}
-	if context.Cause(eyeball1Ctx) != contextCancelErr {
+	if !errors.Is(context.Cause(eyeball1Ctx), contextCancelErr) {
 		t.Fatalf("first eyeball context should be cancelled manually: %+v", context.Cause(eyeball1Ctx))
 	}
 
 	// Origin sends data
 	payload2 := []byte{0xde}
-	pipe1.Write(payload2)
+	_, _ = pipe1.Write(payload2)
 
 	// Expect write to eyeball2
 	data := <-eyeball2.recvData

supervisor/tunnel.go

@@ -17,6 +17,7 @@ import (
 	"github.com/rs/zerolog"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/cloudflare/cloudflared/client"
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/edgediscovery"
 	"github.com/cloudflare/cloudflared/edgediscovery/allregions"
@@ -38,10 +39,8 @@ const (
 )
 
 type TunnelConfig struct {
+	ClientConfig       *client.Config
 	GracePeriod        time.Duration
-	ReplaceExisting    bool
-	OSArch             string
-	ClientID           string
 	CloseConnOnce      *sync.Once // Used to close connectedSignal no more than once
 	EdgeAddrs          []string
 	Region             string
@@ -72,22 +71,13 @@ type TunnelConfig struct {
 	DisableQUICPathMTUDiscovery         bool
 	QUICConnectionLevelFlowControlLimit uint64
 	QUICStreamLevelFlowControlLimit     uint64
-
-	FeatureSelector *features.FeatureSelector
 }
 
-func (c *TunnelConfig) connectionOptions(originLocalAddr string, numPreviousAttempts uint8) *pogs.ConnectionOptions {
+func (c *TunnelConfig) connectionOptions(originLocalAddr string, previousAttempts uint8) *client.ConnectionOptionsSnapshot {
 	// attempt to parse out origin IP, but don't fail since it's informational field
 	host, _, _ := net.SplitHostPort(originLocalAddr)
 	originIP := net.ParseIP(host)
-
-	return &pogs.ConnectionOptions{
-		Client:              c.NamedTunnel.Client,
-		OriginLocalIP:       originIP,
-		ReplaceExisting:     c.ReplaceExisting,
-		CompressionQuality:  0,
-		NumPreviousAttempts: numPreviousAttempts,
-	}
+	return c.ClientConfig.ConnectionOptionsSnapshot(originIP, previousAttempts)
 }
 
 func StartTunnelDaemon(
@@ -463,6 +453,8 @@ func (e *EdgeTunnelServer) serveConnection(
 	case connection.QUIC:
 		// nolint: gosec
 		connOptions := e.config.connectionOptions(addr.UDP.String(), uint8(backoff.Retries()))
+		// nolint: zerologlint
+		connOptions.LogFields(connLog.Logger().Debug().Uint8(connection.LogFieldConnIndex, connIndex)).Msgf("Tunnel connection options")
 		return e.serveQUIC(ctx,
 			addr.UDP.AddrPort(),
 			connLog,
@@ -479,6 +471,8 @@ func (e *EdgeTunnelServer) serveConnection(
 
 		// nolint: gosec
 		connOptions := e.config.connectionOptions(edgeConn.LocalAddr().String(), uint8(backoff.Retries()))
+		// nolint: zerologlint
+		connOptions.LogFields(connLog.Logger().Debug().Uint8(connection.LogFieldConnIndex, connIndex)).Msgf("Tunnel connection options")
 		if err := e.serveHTTP2(
 			ctx,
 			connLog,
@@ -508,11 +502,11 @@ func (e *EdgeTunnelServer) serveHTTP2(
 	ctx context.Context,
 	connLog *ConnAwareLogger,
 	tlsServerConn net.Conn,
-	connOptions *pogs.ConnectionOptions,
+	connOptions *client.ConnectionOptionsSnapshot,
 	controlStreamHandler connection.ControlStreamHandler,
 	connIndex uint8,
 ) error {
-	pqMode := e.config.FeatureSelector.PostQuantumMode()
+	pqMode := connOptions.FeatureSnapshot.PostQuantum
 	if pqMode == features.PostQuantumStrict {
 		return unrecoverableError{errors.New("HTTP/2 transport does not support post-quantum")}
 	}
@@ -550,19 +544,19 @@ func (e *EdgeTunnelServer) serveQUIC(
 	ctx context.Context,
 	edgeAddr netip.AddrPort,
 	connLogger *ConnAwareLogger,
-	connOptions *pogs.ConnectionOptions,
+	connOptions *client.ConnectionOptionsSnapshot,
 	controlStreamHandler connection.ControlStreamHandler,
 	connIndex uint8,
 ) (err error, recoverable bool) {
 	tlsConfig := e.config.EdgeTLSConfigs[connection.QUIC]
 
-	pqMode := e.config.FeatureSelector.PostQuantumMode()
+	pqMode := connOptions.FeatureSnapshot.PostQuantum
 	curvePref, err := curvePreference(pqMode, fips.IsFipsEnabled(), tlsConfig.CurvePreferences)
 	if err != nil {
 		return err, true
 	}
 
-	connLogger.Logger().Info().Msgf("Using %v as curve preferences", curvePref)
+	connLogger.Logger().Info().Msgf("Tunnel connection curve preferences: %v", curvePref)
 
 	tlsConfig.CurvePreferences = curvePref
 
@@ -600,12 +594,12 @@ func (e *EdgeTunnelServer) serveQUIC(
 	if err != nil {
 		connLogger.ConnAwareLogger().Err(err).Msgf("Failed to dial a quic connection")
 
-		e.reportErrorToSentry(err)
+		e.reportErrorToSentry(err, connOptions.FeatureSnapshot.PostQuantum)
 		return err, true
 	}
 
 	var datagramSessionManager connection.DatagramSessionHandler
-	if e.config.FeatureSelector.DatagramVersion() == features.DatagramV3 {
+	if connOptions.FeatureSnapshot.DatagramVersion == features.DatagramV3 {
 		datagramSessionManager = connection.NewDatagramV3Connection(
 			ctx,
 			conn,
@@ -672,7 +666,7 @@ func (e *EdgeTunnelServer) serveQUIC(
 
 // The reportErrorToSentry is an helper function that handles
 // verifies if an error should be reported to Sentry.
-func (e *EdgeTunnelServer) reportErrorToSentry(err error) {
+func (e *EdgeTunnelServer) reportErrorToSentry(err error, pqMode features.PostQuantumMode) {
 	dialErr, ok := err.(*connection.EdgeQuicDialError)
 	if ok {
 		// The TransportError provides an Unwrap function however
@@ -681,7 +675,7 @@ func (e *EdgeTunnelServer) reportErrorToSentry(err error) {
 		if ok &&
 			transportErr.ErrorCode.IsCryptoError() &&
 			fips.IsFipsEnabled() &&
-			e.config.FeatureSelector.PostQuantumMode() == features.PostQuantumStrict {
+			pqMode == features.PostQuantumStrict {
 			// Only report to Sentry when using FIPS, PQ,
 			// and the error is a Crypto error reported by
 			// an EdgeQuicDialError