Release 2025.10.0

* chore: Fix upload of RPM repo file during double signing

This commit fixes a variable that was supposed to hold the path of the repo file, but instead was being overwritten with the repo file handle

.ci/github.gitlab-ci.yml

@@ -0,0 +1,17 @@
+include:
+  - local: .ci/commons.gitlab-ci.yml
+
+######################################
+### Sync master branch with Github ###
+######################################
+push-github:
+  stage: sync
+  rules:
+    - !reference [.default-rules, run-on-master]
+  script:
+    - ./.ci/scripts/github-push.sh
+  secrets:
+    CLOUDFLARED_DEPLOY_SSH_KEY:
+      vault: gitlab/cloudflare/tun/cloudflared/_branch/master/cloudflared_github_ssh/data@kv
+      file: false
+  cache: {}

.ci/linux.gitlab-ci.yml

@@ -16,6 +16,7 @@ include:
       <<: *golang_inputs
       jobPrefix: linux-build
       GOLANG_MAKE_TARGET: ci-build
+      imageVersion: "3308-283bdf9@sha256:fcd83570c91565a72eab132c38e0f589a481e2f3d4f3779f9f9a93eb555fee4a"
 
   ########################
   ### Linux FIPS Build ###
@@ -25,6 +26,8 @@ include:
       <<: *golang_inputs
       jobPrefix: linux-fips-build
       GOLANG_MAKE_TARGET: ci-fips-build
+      imageVersion: "3308-283bdf9@sha256:fcd83570c91565a72eab132c38e0f589a481e2f3d4f3779f9f9a93eb555fee4a"
+
 
   #################
   ### Unit Tests ##
@@ -35,6 +38,8 @@ include:
       stage: test
       jobPrefix: test
       GOLANG_MAKE_TARGET: ci-test
+      imageVersion: "3308-283bdf9@sha256:fcd83570c91565a72eab132c38e0f589a481e2f3d4f3779f9f9a93eb555fee4a"
+
 
   ######################
   ### Unit Tests FIPS ##
@@ -45,6 +50,8 @@ include:
       stage: test
       jobPrefix: test-fips
       GOLANG_MAKE_TARGET: ci-fips-test
+      imageVersion: "3308-283bdf9@sha256:fcd83570c91565a72eab132c38e0f589a481e2f3d4f3779f9f9a93eb555fee4a"
+
 
   #################
   ### Vuln Check ##
@@ -56,6 +63,8 @@ include:
       stage: validate
       jobPrefix: vulncheck
       GOLANG_MAKE_TARGET: vulncheck
+      imageVersion: "3308-283bdf9@sha256:fcd83570c91565a72eab132c38e0f589a481e2f3d4f3779f9f9a93eb555fee4a"
+
 
 #################################
 ### Run Linux Component Tests ###

.ci/scripts/github-push.sh

@@ -0,0 +1,31 @@
+#!/bin/bash
+set -e -o pipefail
+
+BRANCH="master"
+TMP_PATH="$PWD/tmp"
+PRIVATE_KEY_PATH="$TMP_PATH/github-deploy-key"
+PUBLIC_KEY_GITHUB_PATH="$TMP_PATH/github.pub"
+
+mkdir -p $TMP_PATH
+
+# Setup Private Key
+echo "$CLOUDFLARED_DEPLOY_SSH_KEY" > $PRIVATE_KEY_PATH
+chmod 400 $PRIVATE_KEY_PATH
+
+# Download GitHub Public Key for KnownHostsFile
+ssh-keyscan -t ed25519 github.com > $PUBLIC_KEY_GITHUB_PATH
+
+# Setup git ssh command with the right configurations
+export GIT_SSH_COMMAND="ssh -o UserKnownHostsFile=$PUBLIC_KEY_GITHUB_PATH -o IdentitiesOnly=yes -i $PRIVATE_KEY_PATH"
+
+# Add GitHub as a new remote
+git remote add github git@github.com:cloudflare/cloudflared.git || true
+
+# GitLab doesn't pull branch references, instead it creates a new one on each pipeline.
+# Therefore, we need to manually fetch the reference to then push it to GitHub.
+git fetch origin $BRANCH:$BRANCH
+git push -u github $BRANCH
+
+if TAG="$(git describe --tags --exact-match 2>/dev/null)"; then
+  git push -u github "$TAG"
+fi

.gitlab-ci.yml

@@ -7,7 +7,7 @@ default:
     VAULT_ID_TOKEN:
       aud: https://vault.cfdata.org
 
-stages: [pre-build, build, validate, test, package, release]
+stages: [sync, pre-build, build, validate, test, package, release]
 
 include:
   #####################################################
@@ -15,6 +15,11 @@ include:
   #####################################################
   - local: .ci/commons.gitlab-ci.yml
 
+  #####################################################
+  ########### Sync Repository with Github #############
+  #####################################################
+  - local: .ci/github.gitlab-ci.yml
+
   #####################################################
   ############# Build or Fetch CI Image ###############
   #####################################################

Makefile

@@ -36,7 +36,7 @@ ifdef PACKAGE_MANAGER
 	VERSION_FLAGS := $(VERSION_FLAGS) -X "github.com/cloudflare/cloudflared/cmd/cloudflared/updater.BuiltForPackageManager=$(PACKAGE_MANAGER)"
 endif
 
-ifdef CONTAINER_BUILD 
+ifdef CONTAINER_BUILD
 	VERSION_FLAGS := $(VERSION_FLAGS) -X "github.com/cloudflare/cloudflared/metrics.Runtime=virtual"
 endif
 
@@ -119,7 +119,7 @@ ifneq ($(TARGET_ARM), )
 	ARM_COMMAND := GOARM=$(TARGET_ARM)
 endif
 
-ifeq ($(TARGET_ARM), 7) 
+ifeq ($(TARGET_ARM), 7)
 	PACKAGE_ARCH := armhf
 else
 	PACKAGE_ARCH := $(TARGET_ARCH)
@@ -182,7 +182,7 @@ fuzz:
 	@go test -fuzz=FuzzIPDecoder -fuzztime=600s ./packet
 	@go test -fuzz=FuzzICMPDecoder -fuzztime=600s ./packet
 	@go test -fuzz=FuzzSessionWrite -fuzztime=600s ./quic/v3
-	@go test -fuzz=FuzzSessionServe -fuzztime=600s ./quic/v3
+	@go test -fuzz=FuzzSessionRead -fuzztime=600s ./quic/v3
 	@go test -fuzz=FuzzRegistrationDatagram -fuzztime=600s ./quic/v3
 	@go test -fuzz=FuzzPayloadDatagram -fuzztime=600s ./quic/v3
 	@go test -fuzz=FuzzRegistrationResponseDatagram -fuzztime=600s ./quic/v3
@@ -246,6 +246,11 @@ gitlab-release:
 r2-linux-release:
 	python3 ./release_pkgs.py
 
+.PHONY: r2-next-linux-release
+# Publishes to a separate R2 repository during GPG key rollover, using dual-key signing.
+r2-next-linux-release:
+	python3 ./release_pkgs.py --upload-repo-file
+
 .PHONY: capnp
 capnp:
 	which capnp  # https://capnproto.org/install.html

RELEASE_NOTES

@@ -1,3 +1,17 @@
+2025.10.0
+- 2025-10-14 chore: Fix upload of RPM repo file during double signing
+- 2025-10-13 TUN-9882: Bump datagram v3 write channel capacity
+- 2025-10-10 chore: Fix import of GPG keys when two keys are provided
+- 2025-10-10 chore: Fix parameter order when uploading RPM .repo file to R2
+- 2025-10-10 TUN-9883: Add new datagram v3 feature flag
+- 2025-10-09 chore: Force usage of go-boring 1.24
+- 2025-10-08 TUN-9882: Improve metrics for datagram v3
+- 2025-10-07 GRC-16749: Add fedramp tags to catalog
+- 2025-10-07 TUN-9882: Add buffers for UDP and ICMP datagrams in datagram v3
+- 2025-10-07 TUN-9882: Add write deadline for UDP origin writes
+- 2025-09-29 TUN-9776: Support signing Debian packages with two keys for rollover
+- 2025-09-22 TUN-9800: Add pipeline to sync between gitlab and github repos
+
 2025.9.1
 - 2025-09-22 TUN-9855: Create script to ignore vulnerabilities from govuln check
 - 2025-09-19 TUN-9852: Remove fmt.Println from cloudflared access command

catalog-info.yaml

@@ -14,4 +14,7 @@ spec:
   lifecycle: "Active"
   owner: "teams/tunnel-teams-routing"
   cf:
+    compliance:
+      fedramp-high: "pending"
+      fedramp-moderate: "yes"
     FIPS: "required"

cfsetup.yaml

@@ -211,7 +211,7 @@ bookworm: &bookworm
       - make github-release
   r2-linux-release:
     build_dir: *build_dir
-    builddeps:
+    builddeps: &r2-linux-release-deps
       - *pinned_go
       - build-essential
       - fakeroot
@@ -231,4 +231,13 @@ bookworm: &bookworm
       - pip install pynacl==1.4.0 pygithub==1.55 boto3==1.22.9 python-gnupg==0.4.9
       - make r2-linux-release
 
+  r2-next-linux-release:
+    build_dir: *build_dir
+    builddeps: *r2-linux-release-deps
+    post-cache:
+      - python3 -m venv env
+      - . env/bin/activate
+      - pip install pynacl==1.4.0 pygithub==1.55 boto3==1.22.9 python-gnupg==0.4.9
+      - make r2-next-linux-release
+
 trixie: *bookworm

client/config_test.go

@@ -45,6 +45,6 @@ func (m *mockFeatureSelector) Snapshot() features.FeatureSnapshot {
 	return features.FeatureSnapshot{
 		PostQuantum:     features.PostQuantumPrefer,
 		DatagramVersion: features.DatagramV3,
-		FeaturesList:    []string{features.FeaturePostQuantum, features.FeatureDatagramV3_1},
+		FeaturesList:    []string{features.FeaturePostQuantum, features.FeatureDatagramV3_2},
 	}
 }

features/features.go

@@ -10,9 +10,10 @@ const (
 	FeaturePostQuantum       = "postquantum"
 	FeatureQUICSupportEOF    = "support_quic_eof"
 	FeatureManagementLogs    = "management_logs"
-	FeatureDatagramV3_1      = "support_datagram_v3_1"
+	FeatureDatagramV3_2      = "support_datagram_v3_2"
 
-	DeprecatedFeatureDatagramV3 = "support_datagram_v3" // Deprecated: TUN-9291
+	DeprecatedFeatureDatagramV3   = "support_datagram_v3"   // Deprecated: TUN-9291
+	DeprecatedFeatureDatagramV3_1 = "support_datagram_v3_1" // Deprecated: TUN-9883
 )
 
 var defaultFeatures = []string{
@@ -26,6 +27,7 @@ var defaultFeatures = []string{
 // List of features that are no longer in-use.
 var deprecatedFeatures = []string{
 	DeprecatedFeatureDatagramV3,
+	DeprecatedFeatureDatagramV3_1,
 }
 
 // Features set by user provided flags
@@ -58,7 +60,7 @@ 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_1
+	DatagramV3 DatagramVersion = FeatureDatagramV3_2
 )
 
 // Remove any duplicate features from the list and remove deprecated features

features/selector.go

@@ -23,9 +23,10 @@ const (
 // If the TXT record is missing a key, the field will unmarshal to the default Go value
 
 type featuresRecord struct {
-	DatagramV3Percentage uint32 `json:"dv3_1"`
+	DatagramV3Percentage uint32 `json:"dv3_2"`
 
 	// DatagramV3Percentage int32 `json:"dv3"` // Removed in TUN-9291
+	// DatagramV3Percentage uint32 `json:"dv3_1"` // Removed in TUN-9883
 	// PostQuantumPercentage int32 `json:"pq"` // Removed in TUN-7970
 }
 
@@ -105,7 +106,7 @@ func (fs *featureSelector) postQuantumMode() PostQuantumMode {
 
 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_1) {
+	if slices.Contains(fs.cliFeatures, FeatureDatagramV3_2) {
 		return DatagramV3
 	}
 	// If the user specifies DatagramV2, we also take that over remote

features/selector_test.go

@@ -22,15 +22,15 @@ func TestUnmarshalFeaturesRecord(t *testing.T) {
 		expectedPercentage uint32
 	}{
 		{
-			record:             []byte(`{"dv3_1":0}`),
+			record:             []byte(`{"dv3_2":0}`),
 			expectedPercentage: 0,
 		},
 		{
-			record:             []byte(`{"dv3_1":39}`),
+			record:             []byte(`{"dv3_2":39}`),
 			expectedPercentage: 39,
 		},
 		{
-			record:             []byte(`{"dv3_1":100}`),
+			record:             []byte(`{"dv3_2":100}`),
 			expectedPercentage: 100,
 		},
 		{
@@ -40,7 +40,7 @@ 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
+			record: []byte(`{"pq": 101,"dv3":100,"dv3_1":100}`), // Expired keys don't unmarshal to anything
 		},
 	}
 
@@ -111,10 +111,10 @@ func TestFeaturePrecedenceEvaluationDatagramVersion(t *testing.T) {
 		},
 		{
 			name:             "user_specified_v3",
-			cli:              []string{FeatureDatagramV3_1},
+			cli:              []string{FeatureDatagramV3_2},
 			remote:           featuresRecord{},
-			expectedFeatures: dedupAndRemoveFeatures(append(defaultFeatures, FeatureDatagramV3_1)),
-			expectedVersion:  FeatureDatagramV3_1,
+			expectedFeatures: dedupAndRemoveFeatures(append(defaultFeatures, FeatureDatagramV3_2)),
+			expectedVersion:  FeatureDatagramV3_2,
 		},
 	}
 
@@ -150,6 +150,12 @@ func TestDeprecatedFeaturesRemoved(t *testing.T) {
 			remote:           featuresRecord{},
 			expectedFeatures: defaultFeatures,
 		},
+		{
+			name:             "support_datagram_v3_1",
+			cli:              []string{DeprecatedFeatureDatagramV3_1},
+			remote:           featuresRecord{},
+			expectedFeatures: defaultFeatures,
+		},
 	}
 
 	for _, test := range tests {

ingress/origin_dialer.go

@@ -11,6 +11,8 @@ import (
 	"github.com/rs/zerolog"
 )
 
+const writeDeadlineUDP = 200 * time.Millisecond
+
 // OriginTCPDialer provides a TCP dial operation to a requested address.
 type OriginTCPDialer interface {
 	DialTCP(ctx context.Context, addr netip.AddrPort) (net.Conn, error)
@@ -141,6 +143,21 @@ func (d *Dialer) DialUDP(dest netip.AddrPort) (net.Conn, error) {
 	if err != nil {
 		return nil, fmt.Errorf("unable to dial udp to origin %s: %w", dest, err)
 	}
-
-	return conn, nil
+	return &writeDeadlineConn{
+		Conn: conn,
+	}, nil
+}
+
+// writeDeadlineConn is a wrapper around a net.Conn that sets a write deadline of 200ms.
+// This is to prevent the socket from blocking on the write operation if it were to occur. However,
+// we typically never expect this to occur except under high load or kernel issues.
+type writeDeadlineConn struct {
+	net.Conn
+}
+
+func (w *writeDeadlineConn) Write(b []byte) (int, error) {
+	if err := w.SetWriteDeadline(time.Now().Add(writeDeadlineUDP)); err != nil {
+		return 0, err
+	}
+	return w.Conn.Write(b)
 }

quic/v3/datagram_test.go

@@ -1,6 +1,7 @@
 package v3_test
 
 import (
+	"crypto/rand"
 	"encoding/binary"
 	"errors"
 	"net/netip"
@@ -14,12 +15,18 @@ import (
 
 func makePayload(size int) []byte {
 	payload := make([]byte, size)
-	for i := range len(payload) {
-		payload[i] = 0xfc
-	}
+	_, _ = rand.Read(payload)
 	return payload
 }
 
+func makePayloads(size int, count int) [][]byte {
+	payloads := make([][]byte, count)
+	for i := range payloads {
+		payloads[i] = makePayload(size)
+	}
+	return payloads
+}
+
 func TestSessionRegistration_MarshalUnmarshal(t *testing.T) {
 	payload := makePayload(1280)
 	tests := []*v3.UDPSessionRegistrationDatagram{

quic/v3/metrics.go

@@ -9,28 +9,59 @@ import (
 )
 
 const (
-	namespace = "cloudflared"
-	subsystem = "udp"
+	namespace      = "cloudflared"
+	subsystem_udp  = "udp"
+	subsystem_icmp = "icmp"
 
 	commandMetricLabel = "command"
+	reasonMetricLabel  = "reason"
 )
 
+type DroppedReason int
+
+const (
+	DroppedWriteFailed DroppedReason = iota
+	DroppedWriteDeadlineExceeded
+	DroppedWriteFull
+	DroppedWriteFlowUnknown
+	DroppedReadFailed
+	// Origin payloads that are too large to proxy.
+	DroppedReadTooLarge
+)
+
+var droppedReason = map[DroppedReason]string{
+	DroppedWriteFailed:           "write_failed",
+	DroppedWriteDeadlineExceeded: "write_deadline_exceeded",
+	DroppedWriteFull:             "write_full",
+	DroppedWriteFlowUnknown:      "write_flow_unknown",
+	DroppedReadFailed:            "read_failed",
+	DroppedReadTooLarge:          "read_too_large",
+}
+
+func (dr DroppedReason) String() string {
+	return droppedReason[dr]
+}
+
 type Metrics interface {
 	IncrementFlows(connIndex uint8)
 	DecrementFlows(connIndex uint8)
-	PayloadTooLarge(connIndex uint8)
+	FailedFlow(connIndex uint8)
 	RetryFlowResponse(connIndex uint8)
 	MigrateFlow(connIndex uint8)
 	UnsupportedRemoteCommand(connIndex uint8, command string)
+	DroppedUDPDatagram(connIndex uint8, reason DroppedReason)
+	DroppedICMPPackets(connIndex uint8, reason DroppedReason)
 }
 
 type metrics struct {
 	activeUDPFlows            *prometheus.GaugeVec
 	totalUDPFlows             *prometheus.CounterVec
-	payloadTooLarge           *prometheus.CounterVec
 	retryFlowResponses        *prometheus.CounterVec
 	migratedFlows             *prometheus.CounterVec
 	unsupportedRemoteCommands *prometheus.CounterVec
+	droppedUDPDatagrams       *prometheus.CounterVec
+	droppedICMPPackets        *prometheus.CounterVec
+	failedFlows               *prometheus.CounterVec
 }
 
 func (m *metrics) IncrementFlows(connIndex uint8) {
@@ -42,8 +73,8 @@ func (m *metrics) DecrementFlows(connIndex uint8) {
 	m.activeUDPFlows.WithLabelValues(fmt.Sprintf("%d", connIndex)).Dec()
 }
 
-func (m *metrics) PayloadTooLarge(connIndex uint8) {
-	m.payloadTooLarge.WithLabelValues(fmt.Sprintf("%d", connIndex)).Inc()
+func (m *metrics) FailedFlow(connIndex uint8) {
+	m.failedFlows.WithLabelValues(fmt.Sprintf("%d", connIndex)).Inc()
 }
 
 func (m *metrics) RetryFlowResponse(connIndex uint8) {
@@ -58,52 +89,74 @@ func (m *metrics) UnsupportedRemoteCommand(connIndex uint8, command string) {
 	m.unsupportedRemoteCommands.WithLabelValues(fmt.Sprintf("%d", connIndex), command).Inc()
 }
 
+func (m *metrics) DroppedUDPDatagram(connIndex uint8, reason DroppedReason) {
+	m.droppedUDPDatagrams.WithLabelValues(fmt.Sprintf("%d", connIndex), reason.String()).Inc()
+}
+
+func (m *metrics) DroppedICMPPackets(connIndex uint8, reason DroppedReason) {
+	m.droppedICMPPackets.WithLabelValues(fmt.Sprintf("%d", connIndex), reason.String()).Inc()
+}
+
 func NewMetrics(registerer prometheus.Registerer) Metrics {
 	m := &metrics{
 		activeUDPFlows: prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: namespace,
-			Subsystem: subsystem,
+			Subsystem: subsystem_udp,
 			Name:      "active_flows",
 			Help:      "Concurrent count of UDP flows that are being proxied to any origin",
 		}, []string{quic.ConnectionIndexMetricLabel}),
 		totalUDPFlows: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
 			Namespace: namespace,
-			Subsystem: subsystem,
+			Subsystem: subsystem_udp,
 			Name:      "total_flows",
 			Help:      "Total count of UDP flows that have been proxied to any origin",
 		}, []string{quic.ConnectionIndexMetricLabel}),
-		payloadTooLarge: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
+		failedFlows: 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",
+			Subsystem: subsystem_udp,
+			Name:      "failed_flows",
+			Help:      "Total count of flows that errored and closed",
 		}, []string{quic.ConnectionIndexMetricLabel}),
 		retryFlowResponses: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
 			Namespace: namespace,
-			Subsystem: subsystem,
+			Subsystem: subsystem_udp,
 			Name:      "retry_flow_responses",
 			Help:      "Total count of UDP flows that have had to send their registration response more than once",
 		}, []string{quic.ConnectionIndexMetricLabel}),
 		migratedFlows: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
 			Namespace: namespace,
-			Subsystem: subsystem,
+			Subsystem: subsystem_udp,
 			Name:      "migrated_flows",
 			Help:      "Total count of UDP flows have been migrated across local connections",
 		}, []string{quic.ConnectionIndexMetricLabel}),
-		unsupportedRemoteCommands: prometheus.NewCounterVec(prometheus.CounterOpts{
+		unsupportedRemoteCommands: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
 			Namespace: namespace,
-			Subsystem: subsystem,
+			Subsystem: subsystem_udp,
 			Name:      "unsupported_remote_command_total",
-			Help:      "Total count of unsupported remote RPC commands for the ",
+			Help:      "Total count of unsupported remote RPC commands called",
 		}, []string{quic.ConnectionIndexMetricLabel, commandMetricLabel}),
+		droppedUDPDatagrams: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
+			Namespace: namespace,
+			Subsystem: subsystem_udp,
+			Name:      "dropped_datagrams",
+			Help:      "Total count of UDP dropped datagrams",
+		}, []string{quic.ConnectionIndexMetricLabel, reasonMetricLabel}),
+		droppedICMPPackets: prometheus.NewCounterVec(prometheus.CounterOpts{ //nolint:promlinter
+			Namespace: namespace,
+			Subsystem: subsystem_icmp,
+			Name:      "dropped_packets",
+			Help:      "Total count of ICMP dropped datagrams",
+		}, []string{quic.ConnectionIndexMetricLabel, reasonMetricLabel}),
 	}
 	registerer.MustRegister(
 		m.activeUDPFlows,
 		m.totalUDPFlows,
-		m.payloadTooLarge,
+		m.failedFlows,
 		m.retryFlowResponses,
 		m.migratedFlows,
 		m.unsupportedRemoteCommands,
+		m.droppedUDPDatagrams,
+		m.droppedICMPPackets,
 	)
 	return m
 }

quic/v3/metrics_test.go

@@ -1,10 +1,15 @@
 package v3_test
 
+import v3 "github.com/cloudflare/cloudflared/quic/v3"
+
 type noopMetrics struct{}
 
-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) {}
+func (noopMetrics) IncrementFlows(connIndex uint8)                              {}
+func (noopMetrics) DecrementFlows(connIndex uint8)                              {}
+func (noopMetrics) FailedFlow(connIndex uint8)                                  {}
+func (noopMetrics) PayloadTooLarge(connIndex uint8)                             {}
+func (noopMetrics) RetryFlowResponse(connIndex uint8)                           {}
+func (noopMetrics) MigrateFlow(connIndex uint8)                                 {}
+func (noopMetrics) UnsupportedRemoteCommand(connIndex uint8, command string)    {}
+func (noopMetrics) DroppedUDPDatagram(connIndex uint8, reason v3.DroppedReason) {}
+func (noopMetrics) DroppedICMPPackets(connIndex uint8, reason v3.DroppedReason) {}

quic/v3/muxer.go

@@ -17,6 +17,9 @@ const (
 	// Allocating a 16 channel buffer here allows for the writer to be slightly faster than the reader.
 	// This has worked previously well for datagramv2, so we will start with this as well
 	demuxChanCapacity = 16
+	// This provides a small buffer for the PacketRouter to poll ICMP packets from the QUIC connection
+	// before writing them to the origin.
+	icmpDatagramChanCapacity = 128
 
 	logSrcKey      = "src"
 	logDstKey      = "dst"
@@ -59,14 +62,15 @@ type QuicConnection interface {
 }
 
 type datagramConn struct {
-	conn           QuicConnection
-	index          uint8
-	sessionManager SessionManager
-	icmpRouter     ingress.ICMPRouter
-	metrics        Metrics
-	logger         *zerolog.Logger
-	datagrams      chan []byte
-	readErrors     chan error
+	conn             QuicConnection
+	index            uint8
+	sessionManager   SessionManager
+	icmpRouter       ingress.ICMPRouter
+	metrics          Metrics
+	logger           *zerolog.Logger
+	datagrams        chan []byte
+	icmpDatagramChan chan *ICMPDatagram
+	readErrors       chan error
 
 	icmpEncoderPool sync.Pool // a pool of *packet.Encoder
 	icmpDecoderPool sync.Pool
@@ -75,14 +79,15 @@ type datagramConn struct {
 func NewDatagramConn(conn QuicConnection, sessionManager SessionManager, icmpRouter ingress.ICMPRouter, index uint8, metrics Metrics, logger *zerolog.Logger) DatagramConn {
 	log := logger.With().Uint8("datagramVersion", 3).Logger()
 	return &datagramConn{
-		conn:           conn,
-		index:          index,
-		sessionManager: sessionManager,
-		icmpRouter:     icmpRouter,
-		metrics:        metrics,
-		logger:         &log,
-		datagrams:      make(chan []byte, demuxChanCapacity),
-		readErrors:     make(chan error, 2),
+		conn:             conn,
+		index:            index,
+		sessionManager:   sessionManager,
+		icmpRouter:       icmpRouter,
+		metrics:          metrics,
+		logger:           &log,
+		datagrams:        make(chan []byte, demuxChanCapacity),
+		icmpDatagramChan: make(chan *ICMPDatagram, icmpDatagramChanCapacity),
+		readErrors:       make(chan error, 2),
 		icmpEncoderPool: sync.Pool{
 			New: func() any {
 				return packet.NewEncoder()
@@ -168,6 +173,9 @@ func (c *datagramConn) Serve(ctx context.Context) error {
 	readCtx, cancel := context.WithCancel(connCtx)
 	defer cancel()
 	go c.pollDatagrams(readCtx)
+	// Processing ICMP datagrams also monitors the reader context since the ICMP datagrams from the reader are the input
+	// for the routine.
+	go c.processICMPDatagrams(readCtx)
 	for {
 		// We make sure to monitor the context of cloudflared and the underlying connection to return if any errors occur.
 		var datagram []byte
@@ -181,58 +189,59 @@ func (c *datagramConn) Serve(ctx context.Context) error {
 		// Monitor for any hard errors from reading the connection
 		case err := <-c.readErrors:
 			return err
-		// Otherwise, wait and dequeue datagrams as they come in
+		// Wait and dequeue datagrams as they come in
 		case d := <-c.datagrams:
 			datagram = d
 		}
 
 		// Each incoming datagram will be processed in a new go routine to handle the demuxing and action associated.
-		go func() {
-			typ, err := ParseDatagramType(datagram)
+		typ, err := ParseDatagramType(datagram)
+		if err != nil {
+			c.logger.Err(err).Msgf("unable to parse datagram type: %d", typ)
+			continue
+		}
+		switch typ {
+		case UDPSessionRegistrationType:
+			reg := &UDPSessionRegistrationDatagram{}
+			err := reg.UnmarshalBinary(datagram)
 			if err != nil {
-				c.logger.Err(err).Msgf("unable to parse datagram type: %d", typ)
-				return
+				c.logger.Err(err).Msgf("unable to unmarshal session registration datagram")
+				continue
 			}
-			switch typ {
-			case UDPSessionRegistrationType:
-				reg := &UDPSessionRegistrationDatagram{}
-				err := reg.UnmarshalBinary(datagram)
-				if err != nil {
-					c.logger.Err(err).Msgf("unable to unmarshal session registration datagram")
-					return
-				}
-				logger := c.logger.With().Str(logFlowID, reg.RequestID.String()).Logger()
-				// We bind the new session to the quic connection context instead of cloudflared context to allow for the
-				// quic connection to close and close only the sessions bound to it. Closing of cloudflared will also
-				// initiate the close of the quic connection, so we don't have to worry about the application context
-				// in the scope of a session.
-				c.handleSessionRegistrationDatagram(connCtx, reg, &logger)
-			case UDPSessionPayloadType:
-				payload := &UDPSessionPayloadDatagram{}
-				err := payload.UnmarshalBinary(datagram)
-				if err != nil {
-					c.logger.Err(err).Msgf("unable to unmarshal session payload datagram")
-					return
-				}
-				logger := c.logger.With().Str(logFlowID, payload.RequestID.String()).Logger()
-				c.handleSessionPayloadDatagram(payload, &logger)
-			case ICMPType:
-				packet := &ICMPDatagram{}
-				err := packet.UnmarshalBinary(datagram)
-				if err != nil {
-					c.logger.Err(err).Msgf("unable to unmarshal icmp datagram")
-					return
-				}
-				c.handleICMPPacket(packet)
-			case UDPSessionRegistrationResponseType:
-				// cloudflared should never expect to receive UDP session responses as it will not initiate new
-				// sessions towards the edge.
-				c.logger.Error().Msgf("unexpected datagram type received: %d", UDPSessionRegistrationResponseType)
-				return
-			default:
-				c.logger.Error().Msgf("unknown datagram type received: %d", typ)
+			logger := c.logger.With().Str(logFlowID, reg.RequestID.String()).Logger()
+			// We bind the new session to the quic connection context instead of cloudflared context to allow for the
+			// quic connection to close and close only the sessions bound to it. Closing of cloudflared will also
+			// initiate the close of the quic connection, so we don't have to worry about the application context
+			// in the scope of a session.
+			//
+			// Additionally, we spin out the registration into a separate go routine to handle the Serve'ing of the
+			// session in a separate routine from the demuxer.
+			go c.handleSessionRegistrationDatagram(connCtx, reg, &logger)
+		case UDPSessionPayloadType:
+			payload := &UDPSessionPayloadDatagram{}
+			err := payload.UnmarshalBinary(datagram)
+			if err != nil {
+				c.logger.Err(err).Msgf("unable to unmarshal session payload datagram")
+				continue
 			}
-		}()
+			logger := c.logger.With().Str(logFlowID, payload.RequestID.String()).Logger()
+			c.handleSessionPayloadDatagram(payload, &logger)
+		case ICMPType:
+			packet := &ICMPDatagram{}
+			err := packet.UnmarshalBinary(datagram)
+			if err != nil {
+				c.logger.Err(err).Msgf("unable to unmarshal icmp datagram")
+				continue
+			}
+			c.handleICMPPacket(packet)
+		case UDPSessionRegistrationResponseType:
+			// cloudflared should never expect to receive UDP session responses as it will not initiate new
+			// sessions towards the edge.
+			c.logger.Error().Msgf("unexpected datagram type received: %d", UDPSessionRegistrationResponseType)
+			continue
+		default:
+			c.logger.Error().Msgf("unknown datagram type received: %d", typ)
+		}
 	}
 }
 
@@ -243,24 +252,21 @@ func (c *datagramConn) handleSessionRegistrationDatagram(ctx context.Context, da
 		Str(logDstKey, datagram.Dest.String()).
 		Logger()
 	session, err := c.sessionManager.RegisterSession(datagram, c)
-	switch err {
-	case nil:
-		// Continue as normal
-	case ErrSessionAlreadyRegistered:
-		// Session is already registered and likely the response got lost
-		c.handleSessionAlreadyRegistered(datagram.RequestID, &log)
-		return
-	case ErrSessionBoundToOtherConn:
-		// Session is already registered but to a different connection
-		c.handleSessionMigration(datagram.RequestID, &log)
-		return
-	case ErrSessionRegistrationRateLimited:
-		// There are too many concurrent sessions so we return an error to force a retry later
-		c.handleSessionRegistrationRateLimited(datagram, &log)
-		return
-	default:
-		log.Err(err).Msg("flow registration failure")
-		c.handleSessionRegistrationFailure(datagram.RequestID, &log)
+	if err != nil {
+		switch err {
+		case ErrSessionAlreadyRegistered:
+			// Session is already registered and likely the response got lost
+			c.handleSessionAlreadyRegistered(datagram.RequestID, &log)
+		case ErrSessionBoundToOtherConn:
+			// Session is already registered but to a different connection
+			c.handleSessionMigration(datagram.RequestID, &log)
+		case ErrSessionRegistrationRateLimited:
+			// There are too many concurrent sessions so we return an error to force a retry later
+			c.handleSessionRegistrationRateLimited(datagram, &log)
+		default:
+			log.Err(err).Msg("flow registration failure")
+			c.handleSessionRegistrationFailure(datagram.RequestID, &log)
+		}
 		return
 	}
 	log = log.With().Str(logSrcKey, session.LocalAddr().String()).Logger()
@@ -362,30 +368,54 @@ func (c *datagramConn) handleSessionRegistrationRateLimited(datagram *UDPSession
 func (c *datagramConn) handleSessionPayloadDatagram(datagram *UDPSessionPayloadDatagram, logger *zerolog.Logger) {
 	s, err := c.sessionManager.GetSession(datagram.RequestID)
 	if err != nil {
+		c.metrics.DroppedUDPDatagram(c.index, DroppedWriteFlowUnknown)
 		logger.Err(err).Msgf("unable to find flow")
 		return
 	}
-	// We ignore the bytes written to the socket because any partial write must return an error.
-	_, err = s.Write(datagram.Payload)
-	if err != nil {
-		logger.Err(err).Msgf("unable to write payload for the flow")
-		return
-	}
+	s.Write(datagram.Payload)
 }
 
-// Handles incoming ICMP datagrams.
+// Handles incoming ICMP datagrams into a serialized channel to be handled by a single consumer.
 func (c *datagramConn) handleICMPPacket(datagram *ICMPDatagram) {
 	if c.icmpRouter == nil {
 		// ICMPRouter is disabled so we drop the current packet and ignore all incoming ICMP packets
 		return
 	}
+	select {
+	case c.icmpDatagramChan <- datagram:
+	default:
+		// If the ICMP datagram channel is full, drop any additional incoming.
+		c.metrics.DroppedICMPPackets(c.index, DroppedWriteFull)
+		c.logger.Warn().Msg("failed to write icmp packet to origin: dropped")
+	}
+}
 
+// Consumes from the ICMP datagram channel to write out the ICMP requests to an origin.
+func (c *datagramConn) processICMPDatagrams(ctx context.Context) {
+	if c.icmpRouter == nil {
+		// ICMPRouter is disabled so we ignore all incoming ICMP packets
+		return
+	}
+
+	for {
+		select {
+		// If the provided context is closed we want to exit the write loop
+		case <-ctx.Done():
+			return
+		case datagram := <-c.icmpDatagramChan:
+			c.writeICMPPacket(datagram)
+		}
+	}
+}
+
+func (c *datagramConn) writeICMPPacket(datagram *ICMPDatagram) {
 	// Decode the provided ICMPDatagram as an ICMP packet
 	rawPacket := packet.RawPacket{Data: datagram.Payload}
 	cachedDecoder := c.icmpDecoderPool.Get()
 	defer c.icmpDecoderPool.Put(cachedDecoder)
 	decoder, ok := cachedDecoder.(*packet.ICMPDecoder)
 	if !ok {
+		c.metrics.DroppedICMPPackets(c.index, DroppedWriteFailed)
 		c.logger.Error().Msg("Could not get ICMPDecoder from the pool. Dropping packet")
 		return
 	}
@@ -393,6 +423,7 @@ func (c *datagramConn) handleICMPPacket(datagram *ICMPDatagram) {
 	icmp, err := decoder.Decode(rawPacket)
 
 	if err != nil {
+		c.metrics.DroppedICMPPackets(c.index, DroppedWriteFailed)
 		c.logger.Err(err).Msgf("unable to marshal icmp packet")
 		return
 	}
@@ -400,6 +431,7 @@ func (c *datagramConn) handleICMPPacket(datagram *ICMPDatagram) {
 	// If the ICMP packet's TTL is expired, we won't send it to the origin and immediately return a TTL Exceeded Message
 	if icmp.TTL <= 1 {
 		if err := c.SendICMPTTLExceed(icmp, rawPacket); err != nil {
+			c.metrics.DroppedICMPPackets(c.index, DroppedWriteFailed)
 			c.logger.Err(err).Msg("failed to return ICMP TTL exceed error")
 		}
 		return
@@ -411,6 +443,7 @@ func (c *datagramConn) handleICMPPacket(datagram *ICMPDatagram) {
 	// connection context which will have no tracing information available.
 	err = c.icmpRouter.Request(c.conn.Context(), icmp, newPacketResponder(c, c.index))
 	if err != nil {
+		c.metrics.DroppedICMPPackets(c.index, DroppedWriteFailed)
 		c.logger.Err(err).
 			Str(logSrcKey, icmp.Src.String()).
 			Str(logDstKey, icmp.Dst.String()).

quic/v3/muxer_test.go

@@ -13,6 +13,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/fortytw2/leaktest"
 	"github.com/google/gopacket/layers"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
@@ -92,7 +93,7 @@ func TestDatagramConn_New(t *testing.T) {
 		DefaultDialer:   testDefaultDialer,
 		TCPWriteTimeout: 0,
 	}, &log)
-	conn := v3.NewDatagramConn(newMockQuicConn(), v3.NewSessionManager(&noopMetrics{}, &log, originDialerService, cfdflow.NewLimiter(0)), &noopICMPRouter{}, 0, &noopMetrics{}, &log)
+	conn := v3.NewDatagramConn(newMockQuicConn(t.Context()), v3.NewSessionManager(&noopMetrics{}, &log, originDialerService, cfdflow.NewLimiter(0)), &noopICMPRouter{}, 0, &noopMetrics{}, &log)
 	if conn == nil {
 		t.Fatal("expected valid connection")
 	}
@@ -104,7 +105,9 @@ func TestDatagramConn_SendUDPSessionDatagram(t *testing.T) {
 		DefaultDialer:   testDefaultDialer,
 		TCPWriteTimeout: 0,
 	}, &log)
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&noopMetrics{}, &log, originDialerService, cfdflow.NewLimiter(0)), &noopICMPRouter{}, 0, &noopMetrics{}, &log)
 
 	payload := []byte{0xef, 0xef}
@@ -123,7 +126,9 @@ func TestDatagramConn_SendUDPSessionResponse(t *testing.T) {
 		DefaultDialer:   testDefaultDialer,
 		TCPWriteTimeout: 0,
 	}, &log)
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&noopMetrics{}, &log, originDialerService, cfdflow.NewLimiter(0)), &noopICMPRouter{}, 0, &noopMetrics{}, &log)
 
 	err := conn.SendUDPSessionResponse(testRequestID, v3.ResponseDestinationUnreachable)
@@ -149,7 +154,9 @@ func TestDatagramConnServe_ApplicationClosed(t *testing.T) {
 		DefaultDialer:   testDefaultDialer,
 		TCPWriteTimeout: 0,
 	}, &log)
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&noopMetrics{}, &log, originDialerService, cfdflow.NewLimiter(0)), &noopICMPRouter{}, 0, &noopMetrics{}, &log)
 
 	ctx, cancel := context.WithTimeout(t.Context(), 1*time.Second)
@@ -166,7 +173,9 @@ func TestDatagramConnServe_ConnectionClosed(t *testing.T) {
 		DefaultDialer:   testDefaultDialer,
 		TCPWriteTimeout: 0,
 	}, &log)
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	ctx, cancel := context.WithTimeout(t.Context(), 1*time.Second)
 	defer cancel()
 	quic.ctx = ctx
@@ -195,15 +204,17 @@ func TestDatagramConnServe_ReceiveDatagramError(t *testing.T) {
 
 func TestDatagramConnServe_SessionRegistrationRateLimit(t *testing.T) {
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	sessionManager := &mockSessionManager{
 		expectedRegErr: v3.ErrSessionRegistrationRateLimited,
 	}
 	conn := v3.NewDatagramConn(quic, sessionManager, &noopICMPRouter{}, 0, &noopMetrics{}, &log)
 
 	// Setup the muxer
-	ctx, cancel := context.WithCancel(t.Context())
-	defer cancel()
+	ctx, cancel := context.WithCancelCause(t.Context())
+	defer cancel(context.Canceled)
 	done := make(chan error, 1)
 	go func() {
 		done <- conn.Serve(ctx)
@@ -223,9 +234,12 @@ func TestDatagramConnServe_SessionRegistrationRateLimit(t *testing.T) {
 
 	require.EqualValues(t, testRequestID, resp.RequestID)
 	require.EqualValues(t, v3.ResponseTooManyActiveFlows, resp.ResponseType)
+
+	assertContextClosed(t, ctx, done, cancel)
 }
 
 func TestDatagramConnServe_ErrorDatagramTypes(t *testing.T) {
+	defer leaktest.Check(t)()
 	for _, test := range []struct {
 		name     string
 		input    []byte
@@ -250,7 +264,9 @@ func TestDatagramConnServe_ErrorDatagramTypes(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			logOutput := new(LockedBuffer)
 			log := zerolog.New(logOutput)
-			quic := newMockQuicConn()
+			connCtx, connCancel := context.WithCancelCause(t.Context())
+			defer connCancel(context.Canceled)
+			quic := newMockQuicConn(connCtx)
 			quic.send <- test.input
 			conn := v3.NewDatagramConn(quic, &mockSessionManager{}, &noopICMPRouter{}, 0, &noopMetrics{}, &log)
 
@@ -289,8 +305,11 @@ func (b *LockedBuffer) String() string {
 }
 
 func TestDatagramConnServe_RegisterSession_SessionManagerError(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	expectedErr := errors.New("unable to register session")
 	sessionManager := mockSessionManager{expectedRegErr: expectedErr}
 	conn := v3.NewDatagramConn(quic, &sessionManager, &noopICMPRouter{}, 0, &noopMetrics{}, &log)
@@ -324,8 +343,11 @@ func TestDatagramConnServe_RegisterSession_SessionManagerError(t *testing.T) {
 }
 
 func TestDatagramConnServe(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	session := newMockSession()
 	sessionManager := mockSessionManager{session: &session}
 	conn := v3.NewDatagramConn(quic, &sessionManager, &noopICMPRouter{}, 0, &noopMetrics{}, &log)
@@ -372,8 +394,11 @@ func TestDatagramConnServe(t *testing.T) {
 // instances causes inteference resulting in multiple different raw packets being decoded
 // as the same decoded packet.
 func TestDatagramConnServeDecodeMultipleICMPInParallel(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	session := newMockSession()
 	sessionManager := mockSessionManager{session: &session}
 	router := newMockICMPRouter()
@@ -413,10 +438,14 @@ func TestDatagramConnServeDecodeMultipleICMPInParallel(t *testing.T) {
 	wg := sync.WaitGroup{}
 	var receivedPackets []*packet.ICMP
 	go func() {
-		for ctx.Err() == nil {
-			icmpPacket := <-router.recv
-			receivedPackets = append(receivedPackets, icmpPacket)
-			wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case icmpPacket := <-router.recv:
+				receivedPackets = append(receivedPackets, icmpPacket)
+				wg.Done()
+			}
 		}
 	}()
 
@@ -452,8 +481,11 @@ func TestDatagramConnServeDecodeMultipleICMPInParallel(t *testing.T) {
 }
 
 func TestDatagramConnServe_RegisterTwice(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	session := newMockSession()
 	sessionManager := mockSessionManager{session: &session}
 	conn := v3.NewDatagramConn(quic, &sessionManager, &noopICMPRouter{}, 0, &noopMetrics{}, &log)
@@ -514,12 +546,17 @@ func TestDatagramConnServe_RegisterTwice(t *testing.T) {
 }
 
 func TestDatagramConnServe_MigrateConnection(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	session := newMockSession()
 	sessionManager := mockSessionManager{session: &session}
 	conn := v3.NewDatagramConn(quic, &sessionManager, &noopICMPRouter{}, 0, &noopMetrics{}, &log)
-	quic2 := newMockQuicConn()
+	conn2Ctx, conn2Cancel := context.WithCancelCause(t.Context())
+	defer conn2Cancel(context.Canceled)
+	quic2 := newMockQuicConn(conn2Ctx)
 	conn2 := v3.NewDatagramConn(quic2, &sessionManager, &noopICMPRouter{}, 1, &noopMetrics{}, &log)
 
 	// Setup the muxer
@@ -597,8 +634,11 @@ func TestDatagramConnServe_MigrateConnection(t *testing.T) {
 }
 
 func TestDatagramConnServe_Payload_GetSessionError(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	// mockSessionManager will return the ErrSessionNotFound for any session attempting to be queried by the muxer
 	sessionManager := mockSessionManager{session: nil, expectedGetErr: v3.ErrSessionNotFound}
 	conn := v3.NewDatagramConn(quic, &sessionManager, &noopICMPRouter{}, 0, &noopMetrics{}, &log)
@@ -624,9 +664,12 @@ func TestDatagramConnServe_Payload_GetSessionError(t *testing.T) {
 	assertContextClosed(t, ctx, done, cancel)
 }
 
-func TestDatagramConnServe_Payload(t *testing.T) {
+func TestDatagramConnServe_Payloads(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	session := newMockSession()
 	sessionManager := mockSessionManager{session: &session}
 	conn := v3.NewDatagramConn(quic, &sessionManager, &noopICMPRouter{}, 0, &noopMetrics{}, &log)
@@ -639,15 +682,26 @@ func TestDatagramConnServe_Payload(t *testing.T) {
 		done <- conn.Serve(ctx)
 	}()
 
-	// Send new session registration
-	expectedPayload := []byte{0xef, 0xef}
-	datagram := newSessionPayloadDatagram(testRequestID, expectedPayload)
-	quic.send <- datagram
+	// Send session payloads
+	expectedPayloads := makePayloads(256, 16)
+	go func() {
+		for _, payload := range expectedPayloads {
+			datagram := newSessionPayloadDatagram(testRequestID, payload)
+			quic.send <- datagram
+		}
+	}()
 
-	// Session should receive the payload
-	payload := <-session.recv
-	if !slices.Equal(expectedPayload, payload) {
-		t.Fatalf("expected session receieve the payload sent via the muxer")
+	// Session should receive the payloads (in-order)
+	for i, payload := range expectedPayloads {
+		select {
+		case recv := <-session.recv:
+			if !slices.Equal(recv, payload) {
+				t.Fatalf("expected session receieve the payload[%d] sent via the muxer: (%x) (%x)", i, recv[:16], payload[:16])
+			}
+		case err := <-ctx.Done():
+			// we expect the payload to return before the context to cancel on the session
+			t.Fatal(err)
+		}
 	}
 
 	// Cancel the muxer Serve context and make sure it closes with the expected error
@@ -655,8 +709,11 @@ func TestDatagramConnServe_Payload(t *testing.T) {
 }
 
 func TestDatagramConnServe_ICMPDatagram_TTLDecremented(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	router := newMockICMPRouter()
 	conn := v3.NewDatagramConn(quic, &mockSessionManager{}, router, 0, &noopMetrics{}, &log)
 
@@ -701,8 +758,11 @@ func TestDatagramConnServe_ICMPDatagram_TTLDecremented(t *testing.T) {
 }
 
 func TestDatagramConnServe_ICMPDatagram_TTLExceeded(t *testing.T) {
+	defer leaktest.Check(t)()
 	log := zerolog.Nop()
-	quic := newMockQuicConn()
+	connCtx, connCancel := context.WithCancelCause(t.Context())
+	defer connCancel(context.Canceled)
+	quic := newMockQuicConn(connCtx)
 	router := newMockICMPRouter()
 	conn := v3.NewDatagramConn(quic, &mockSessionManager{}, router, 0, &noopMetrics{}, &log)
 
@@ -821,9 +881,9 @@ type mockQuicConn struct {
 	recv chan []byte
 }
 
-func newMockQuicConn() *mockQuicConn {
+func newMockQuicConn(ctx context.Context) *mockQuicConn {
 	return &mockQuicConn{
-		ctx:  context.Background(),
+		ctx:  ctx,
 		send: make(chan []byte, 1),
 		recv: make(chan []byte, 1),
 	}
@@ -841,7 +901,12 @@ func (m *mockQuicConn) SendDatagram(payload []byte) error {
 }
 
 func (m *mockQuicConn) ReceiveDatagram(_ context.Context) ([]byte, error) {
-	return <-m.send, nil
+	select {
+	case <-m.ctx.Done():
+		return nil, m.ctx.Err()
+	case b := <-m.send:
+		return b, nil
+	}
 }
 
 type mockQuicConnReadError struct {
@@ -905,11 +970,10 @@ func (m *mockSession) Serve(ctx context.Context) error {
 	return v3.SessionCloseErr
 }
 
-func (m *mockSession) Write(payload []byte) (n int, err error) {
+func (m *mockSession) Write(payload []byte) {
 	b := make([]byte, len(payload))
 	copy(b, payload)
 	m.recv <- b
-	return len(b), nil
 }
 
 func (m *mockSession) Close() error {

quic/v3/session.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"os"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -22,6 +23,11 @@ const (
 	// this value (maxDatagramPayloadLen).
 	maxOriginUDPPacketSize = 1500
 
+	// The maximum amount of datagrams a session will queue up before it begins dropping datagrams.
+	// This channel buffer is small because we assume that the dedicated writer to the origin is typically
+	// fast enought to keep the channel empty.
+	writeChanCapacity = 512
+
 	logFlowID        = "flowID"
 	logPacketSizeKey = "packetSize"
 )
@@ -49,7 +55,7 @@ func newSessionIdleErr(timeout time.Duration) error {
 }
 
 type Session interface {
-	io.WriteCloser
+	io.Closer
 	ID() RequestID
 	ConnectionID() uint8
 	RemoteAddr() net.Addr
@@ -58,6 +64,7 @@ type Session interface {
 	Migrate(eyeball DatagramConn, ctx context.Context, logger *zerolog.Logger)
 	// Serve starts the event loop for processing UDP packets
 	Serve(ctx context.Context) error
+	Write(payload []byte)
 }
 
 type session struct {
@@ -67,12 +74,18 @@ type session struct {
 	originAddr     net.Addr
 	localAddr      net.Addr
 	eyeball        atomic.Pointer[DatagramConn]
+	writeChan      chan []byte
 	// activeAtChan is used to communicate the last read/write time
 	activeAtChan chan time.Time
-	closeChan    chan error
-	contextChan  chan context.Context
-	metrics      Metrics
-	log          *zerolog.Logger
+	errChan      chan error
+	// The close channel signal only exists for the write loop because the read loop is always waiting on a read
+	// from the UDP socket to the origin. To close the read loop we close the socket.
+	// Additionally, we can't close the writeChan to indicate that writes are complete because the producer (edge)
+	// side may still be trying to write to this session.
+	closeWrite  chan struct{}
+	contextChan chan context.Context
+	metrics     Metrics
+	log         *zerolog.Logger
 
 	// A special close function that we wrap with sync.Once to make sure it is only called once
 	closeFn func() error
@@ -89,10 +102,12 @@ func NewSession(
 	log *zerolog.Logger,
 ) Session {
 	logger := log.With().Str(logFlowID, id.String()).Logger()
-	// closeChan has two slots to allow for both writers (the closeFn and the Serve routine) to both be able to
-	// write to the channel without blocking since there is only ever one value read from the closeChan by the
+	writeChan := make(chan []byte, writeChanCapacity)
+	// errChan has three slots to allow for all writers (the closeFn, the read loop and the write loop) to
+	// write to the channel without blocking since there is only ever one value read from the errChan by the
 	// waitForCloseCondition.
-	closeChan := make(chan error, 2)
+	errChan := make(chan error, 3)
+	closeWrite := make(chan struct{})
 	session := &session{
 		id:             id,
 		closeAfterIdle: closeAfterIdle,
@@ -100,10 +115,12 @@ func NewSession(
 		originAddr:     originAddr,
 		localAddr:      localAddr,
 		eyeball:        atomic.Pointer[DatagramConn]{},
+		writeChan:      writeChan,
 		// activeAtChan has low capacity. It can be full when there are many concurrent read/write. markActive() will
 		// drop instead of blocking because last active time only needs to be an approximation
 		activeAtChan: make(chan time.Time, 1),
-		closeChan:    closeChan,
+		errChan:      errChan,
+		closeWrite:   closeWrite,
 		// contextChan is an unbounded channel to help enforce one active migration of a session at a time.
 		contextChan: make(chan context.Context),
 		metrics:     metrics,
@@ -111,9 +128,12 @@ func NewSession(
 		closeFn: sync.OnceValue(func() error {
 			// We don't want to block on sending to the close channel if it is already full
 			select {
-			case closeChan <- SessionCloseErr:
+			case errChan <- SessionCloseErr:
 			default:
 			}
+			// Indicate to the write loop that the session is now closed
+			close(closeWrite)
+			// Close the socket directly to unblock the read loop and cause it to also end
 			return origin.Close()
 		}),
 	}
@@ -154,66 +174,115 @@ func (s *session) Migrate(eyeball DatagramConn, ctx context.Context, logger *zer
 }
 
 func (s *session) Serve(ctx context.Context) error {
-	go func() {
-		// QUIC implementation copies data to another buffer before returning https://github.com/quic-go/quic-go/blob/v0.24.0/session.go#L1967-L1975
-		// This makes it safe to share readBuffer between iterations
-		readBuffer := [maxOriginUDPPacketSize + DatagramPayloadHeaderLen]byte{}
-		// 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])
-		for {
-			// Read from the origin UDP socket
-			n, err := s.origin.Read(readBuffer[DatagramPayloadHeaderLen:])
-			if err != nil {
-				if errors.Is(err, io.EOF) ||
-					errors.Is(err, io.ErrUnexpectedEOF) {
-					s.log.Debug().Msgf("flow (origin) connection closed: %v", err)
-				}
-				s.closeChan <- err
-				return
-			}
-			if n < 0 {
-				s.log.Warn().Int(logPacketSizeKey, n).Msg("flow (origin) packet read was negative and was dropped")
-				continue
-			}
-			if n > maxDatagramPayloadLen {
-				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
-			}
-			// We need to synchronize on the eyeball in-case that the connection was migrated. This should be rarely a point
-			// of lock contention, as a migration can only happen during startup of a session before traffic flow.
-			eyeball := *(s.eyeball.Load())
-			// Sending a packet to the session does block on the [quic.Connection], however, this is okay because it
-			// will cause back-pressure to the kernel buffer if the writes are not fast enough to the edge.
-			err = eyeball.SendUDPSessionDatagram(readBuffer[:DatagramPayloadHeaderLen+n])
-			if err != nil {
-				s.closeChan <- err
-				return
-			}
-			// Mark the session as active since we proxied a valid packet from the origin.
-			s.markActive()
-		}
-	}()
+	go s.writeLoop()
+	go s.readLoop()
 	return s.waitForCloseCondition(ctx, s.closeAfterIdle)
 }
 
-func (s *session) Write(payload []byte) (n int, err error) {
-	n, err = s.origin.Write(payload)
-	if err != nil {
-		s.log.Err(err).Msg("failed to write payload to flow (remote)")
-		return n, err
+// Read datagrams from the origin and write them to the connection.
+func (s *session) readLoop() {
+	// QUIC implementation copies data to another buffer before returning https://github.com/quic-go/quic-go/blob/v0.24.0/session.go#L1967-L1975
+	// This makes it safe to share readBuffer between iterations
+	readBuffer := [maxOriginUDPPacketSize + DatagramPayloadHeaderLen]byte{}
+	// 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])
+	for {
+		// Read from the origin UDP socket
+		n, err := s.origin.Read(readBuffer[DatagramPayloadHeaderLen:])
+		if err != nil {
+			if isConnectionClosed(err) {
+				s.log.Debug().Msgf("flow (read) connection closed: %v", err)
+			}
+			s.closeSession(err)
+			return
+		}
+		if n < 0 {
+			s.metrics.DroppedUDPDatagram(s.ConnectionID(), DroppedReadFailed)
+			s.log.Warn().Int(logPacketSizeKey, n).Msg("flow (origin) packet read was negative and was dropped")
+			continue
+		}
+		if n > maxDatagramPayloadLen {
+			s.metrics.DroppedUDPDatagram(s.ConnectionID(), DroppedReadTooLarge)
+			s.log.Error().Int(logPacketSizeKey, n).Msg("flow (origin) packet read was too large and was dropped")
+			continue
+		}
+		// We need to synchronize on the eyeball in-case that the connection was migrated. This should be rarely a point
+		// of lock contention, as a migration can only happen during startup of a session before traffic flow.
+		eyeball := *(s.eyeball.Load())
+		// Sending a packet to the session does block on the [quic.Connection], however, this is okay because it
+		// will cause back-pressure to the kernel buffer if the writes are not fast enough to the edge.
+		err = eyeball.SendUDPSessionDatagram(readBuffer[:DatagramPayloadHeaderLen+n])
+		if err != nil {
+			s.closeSession(err)
+			return
+		}
+		// Mark the session as active since we proxied a valid packet from the origin.
+		s.markActive()
 	}
-	// Write must return a non-nil error if it returns n < len(p). https://pkg.go.dev/io#Writer
-	if n < len(payload) {
-		s.log.Err(io.ErrShortWrite).Msg("failed to write the full payload to flow (remote)")
-		return n, io.ErrShortWrite
+}
+
+func (s *session) Write(payload []byte) {
+	select {
+	case s.writeChan <- payload:
+	default:
+		s.metrics.DroppedUDPDatagram(s.ConnectionID(), DroppedWriteFull)
+		s.log.Error().Msg("failed to write flow payload to origin: dropped")
+	}
+}
+
+// Read datagrams from the write channel to the origin.
+func (s *session) writeLoop() {
+	for {
+		select {
+		case <-s.closeWrite:
+			// When the closeWrite channel is closed, we will no longer write to the origin and end this
+			// goroutine since the session is now closed.
+			return
+		case payload := <-s.writeChan:
+			n, err := s.origin.Write(payload)
+			if err != nil {
+				// Check if this is a write deadline exceeded to the connection
+				if errors.Is(err, os.ErrDeadlineExceeded) {
+					s.metrics.DroppedUDPDatagram(s.ConnectionID(), DroppedWriteDeadlineExceeded)
+					s.log.Warn().Err(err).Msg("flow (write) deadline exceeded: dropping packet")
+					continue
+				}
+				if isConnectionClosed(err) {
+					s.log.Debug().Msgf("flow (write) connection closed: %v", err)
+				}
+				s.log.Err(err).Msg("failed to write flow payload to origin")
+				s.closeSession(err)
+				// If we fail to write to the origin socket, we need to end the writer and close the session
+				return
+			}
+			// Write must return a non-nil error if it returns n < len(p). https://pkg.go.dev/io#Writer
+			if n < len(payload) {
+				s.metrics.DroppedUDPDatagram(s.ConnectionID(), DroppedWriteFailed)
+				s.log.Err(io.ErrShortWrite).Msg("failed to write the full flow payload to origin")
+				continue
+			}
+			// Mark the session as active since we successfully proxied a packet to the origin.
+			s.markActive()
+		}
+	}
+}
+
+func isConnectionClosed(err error) bool {
+	return errors.Is(err, net.ErrClosed) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF)
+}
+
+// Send an error to the error channel to report that an error has either happened on the tunnel or origin side of the
+// proxied connection.
+func (s *session) closeSession(err error) {
+	select {
+	case s.errChan <- err:
+	default:
+		// In the case that the errChan is already full, we will skip over it and return as to not block
+		// the caller because we should start cleaning up the session.
+		s.log.Warn().Msg("error channel was full")
 	}
-	// Mark the session as active since we proxied a packet to the origin.
-	s.markActive()
-	return n, err
 }
 
 // ResetIdleTimer will restart the current idle timer.
@@ -240,7 +309,8 @@ func (s *session) Close() error {
 
 func (s *session) waitForCloseCondition(ctx context.Context, closeAfterIdle time.Duration) error {
 	connCtx := ctx
-	// Closing the session at the end cancels read so Serve() can return
+	// Closing the session at the end cancels read so Serve() can return, additionally, it closes the
+	// closeWrite channel which indicates to the write loop to return.
 	defer s.Close()
 	if closeAfterIdle == 0 {
 		// Provided that the default caller doesn't specify one
@@ -260,7 +330,10 @@ func (s *session) waitForCloseCondition(ctx context.Context, closeAfterIdle time
 			// still be active on the existing connection.
 			connCtx = newContext
 			continue
-		case reason := <-s.closeChan:
+		case reason := <-s.errChan:
+			// Any error returned here is from the read or write loops indicating that it can no longer process datagrams
+			// and as such the session needs to close.
+			s.metrics.FailedFlow(s.ConnectionID())
 			return reason
 		case <-checkIdleTimer.C:
 			// The check idle timer will only return after an idle period since the last active

quic/v3/session_fuzz_test.go

@@ -4,20 +4,24 @@ import (
 	"testing"
 )
 
-// FuzzSessionWrite verifies that we don't run into any panics when writing variable sized payloads to the origin.
+// FuzzSessionWrite verifies that we don't run into any panics when writing a single variable sized payload to the origin.
 func FuzzSessionWrite(f *testing.F) {
-	f.Fuzz(func(t *testing.T, b []byte) {
-		testSessionWrite(t, b)
-	})
-}
-
-// FuzzSessionServe verifies that we don't run into any panics when reading variable sized payloads from the origin.
-func FuzzSessionServe(f *testing.F) {
 	f.Fuzz(func(t *testing.T, b []byte) {
 		// The origin transport read is bound to 1280 bytes
 		if len(b) > 1280 {
 			b = b[:1280]
 		}
-		testSessionServe_Origin(t, b)
+		testSessionWrite(t, [][]byte{b})
+	})
+}
+
+// FuzzSessionRead verifies that we don't run into any panics when reading a single variable sized payload from the origin.
+func FuzzSessionRead(f *testing.F) {
+	f.Fuzz(func(t *testing.T, b []byte) {
+		// The origin transport read is bound to 1280 bytes
+		if len(b) > 1280 {
+			b = b[:1280]
+		}
+		testSessionRead(t, [][]byte{b})
 	})
 }

quic/v3/session_test.go

@@ -31,60 +31,61 @@ func TestSessionNew(t *testing.T) {
 	}
 }
 
-func testSessionWrite(t *testing.T, payload []byte) {
+func testSessionWrite(t *testing.T, payloads [][]byte) {
 	log := zerolog.Nop()
 	origin, server := net.Pipe()
 	defer origin.Close()
 	defer server.Close()
-	// Start origin server read
-	serverRead := make(chan []byte, 1)
+	// Start origin server reads
+	serverRead := make(chan []byte, len(payloads))
 	go func() {
-		read := make([]byte, 1500)
-		_, _ = server.Read(read[:])
-		serverRead <- read
+		for range len(payloads) {
+			buf := make([]byte, 1500)
+			_, _ = server.Read(buf[:])
+			serverRead <- buf
+		}
+		close(serverRead)
 	}()
-	// Create session and write to origin
+
+	// Create a session
 	session := v3.NewSession(testRequestID, 5*time.Second, origin, testOriginAddr, testLocalAddr, &noopEyeball{}, &noopMetrics{}, &log)
-	n, err := session.Write(payload)
 	defer session.Close()
-	if err != nil {
-		t.Fatal(err)
-	}
-	if n != len(payload) {
-		t.Fatal("unable to write the whole payload")
+	// Start the Serve to begin the writeLoop
+	ctx, cancel := context.WithCancelCause(t.Context())
+	defer cancel(context.Canceled)
+	done := make(chan error)
+	go func() {
+		done <- session.Serve(ctx)
+	}()
+	// Write the payloads to the session
+	for _, payload := range payloads {
+		session.Write(payload)
 	}
 
-	read := <-serverRead
-	if !slices.Equal(payload, read[:len(payload)]) {
-		t.Fatal("payload provided from origin and read value are not the same")
+	// Read from the origin to ensure the payloads were received (in-order)
+	for i, payload := range payloads {
+		read := <-serverRead
+		if !slices.Equal(payload, read[:len(payload)]) {
+			t.Fatalf("payload[%d] provided from origin and read value are not the same (%x) and (%x)", i, payload[:16], read[:16])
+		}
+	}
+	_, more := <-serverRead
+	if more {
+		t.Fatalf("expected the session to have all of the origin payloads received: %d", len(serverRead))
+	}
+
+	assertContextClosed(t, ctx, done, cancel)
+}
+
+func TestSessionWrite(t *testing.T) {
+	defer leaktest.Check(t)()
+	for i := range 1280 {
+		payloads := makePayloads(i, 16)
+		testSessionWrite(t, payloads)
 	}
 }
 
-func TestSessionWrite_Max(t *testing.T) {
-	defer leaktest.Check(t)()
-	payload := makePayload(1280)
-	testSessionWrite(t, payload)
-}
-
-func TestSessionWrite_Min(t *testing.T) {
-	defer leaktest.Check(t)()
-	payload := makePayload(0)
-	testSessionWrite(t, payload)
-}
-
-func TestSessionServe_OriginMax(t *testing.T) {
-	defer leaktest.Check(t)()
-	payload := makePayload(1280)
-	testSessionServe_Origin(t, payload)
-}
-
-func TestSessionServe_OriginMin(t *testing.T) {
-	defer leaktest.Check(t)()
-	payload := makePayload(0)
-	testSessionServe_Origin(t, payload)
-}
-
-func testSessionServe_Origin(t *testing.T, payload []byte) {
+func testSessionRead(t *testing.T, payloads [][]byte) {
 	log := zerolog.Nop()
 	origin, server := net.Pipe()
 	defer origin.Close()
@@ -100,37 +101,42 @@ func testSessionServe_Origin(t *testing.T, payload []byte) {
 		done <- session.Serve(ctx)
 	}()
 
-	// Write from the origin server
-	_, err := server.Write(payload)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	select {
-	case data := <-eyeball.recvData:
-		// check received data matches provided from origin
-		expectedData := makePayload(1500)
-		_ = 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")
+	// Write from the origin server to the eyeball
+	go func() {
+		for _, payload := range payloads {
+			_, _ = server.Write(payload)
+		}
+	}()
+
+	// Read from the eyeball to ensure the payloads were received (in-order)
+	for i, payload := range payloads {
+		select {
+		case data := <-eyeball.recvData:
+			// check received data matches provided from origin
+			expectedData := makePayload(1500)
+			_ = v3.MarshalPayloadHeaderTo(testRequestID, expectedData[:])
+			copy(expectedData[17:], payload)
+			if !slices.Equal(expectedData[:v3.DatagramPayloadHeaderLen+len(payload)], data) {
+				t.Fatalf("expected datagram[%d] did not equal expected", i)
+			}
+		case err := <-ctx.Done():
+			// we expect the payload to return before the context to cancel on the session
+			t.Fatal(err)
 		}
-		cancel(errExpectedContextCanceled)
-	case err := <-ctx.Done():
-		// we expect the payload to return before the context to cancel on the session
-		t.Fatal(err)
 	}
 
-	err = <-done
-	if !errors.Is(err, context.Canceled) {
-		t.Fatal(err)
-	}
-	if !errors.Is(context.Cause(ctx), errExpectedContextCanceled) {
-		t.Fatal(err)
+	assertContextClosed(t, ctx, done, cancel)
+}
+
+func TestSessionRead(t *testing.T) {
+	defer leaktest.Check(t)()
+	for i := range 1280 {
+		payloads := makePayloads(i, 16)
+		testSessionRead(t, payloads)
 	}
 }
 
-func TestSessionServe_OriginTooLarge(t *testing.T) {
+func TestSessionRead_OriginTooLarge(t *testing.T) {
 	defer leaktest.Check(t)()
 	log := zerolog.Nop()
 	eyeball := newMockEyeball()
@@ -317,6 +323,8 @@ func TestSessionServe_IdleTimeout(t *testing.T) {
 	closeAfterIdle := 2 * time.Second
 	session := v3.NewSession(testRequestID, closeAfterIdle, origin, testOriginAddr, testLocalAddr, &noopEyeball{}, &noopMetrics{}, &log)
 	err := session.Serve(t.Context())
+
+	// Session should idle timeout if no reads or writes occur
 	if !errors.Is(err, v3.SessionIdleErr{}) {
 		t.Fatal(err)
 	}

release_pkgs.py

@@ -50,7 +50,7 @@ class PkgUploader:
             config=config,
         )
 
-        print(f"uploading asset: {filename} to {upload_file_path} in bucket{self.bucket_name}...")
+        print(f"uploading asset: {filename} to {upload_file_path} in bucket {self.bucket_name}...")
         try:
             r2.upload_file(filename, self.bucket_name, upload_file_path)
         except ClientError as e:
@@ -133,16 +133,19 @@ class PkgCreator:
     """
 
     def create_repo_file(self, file_path, binary_name, baseurl, gpgkey_url):
-        with open(os.path.join(file_path, binary_name + '.repo'), "w+") as repo_file:
-            repo_file.write(f"[{binary_name}-stable]")
-            repo_file.write(f"{binary_name}-stable")
-            repo_file.write(f"baseurl={baseurl}/rpm")
-            repo_file.write("enabled=1")
-            repo_file.write("type=rpm")
-            repo_file.write("gpgcheck=1")
-            repo_file.write(f"gpgkey={gpgkey_url}")
+        repo_file_path = os.path.join(file_path, binary_name + '.repo')
+        with open(repo_file_path, "w+") as repo_file:
+            repo_file.write(f"[{binary_name}-stable]\n")
+            repo_file.write(f"name={binary_name}-stable\n")
+            repo_file.write(f"baseurl={baseurl}/rpm\n")
+            repo_file.write("enabled=1\n")
+            repo_file.write("type=rpm\n")
+            repo_file.write("gpgcheck=1\n")
+            repo_file.write(f"gpgkey={gpgkey_url}\n")
+        return repo_file_path
 
-    def _sign_rpms(self, file_path):
+
+    def _sign_rpms(self, file_path, gpg_key_name):
         p = Popen(["rpm", "--define", f"_gpg_name {gpg_key_name}", "--addsign", file_path], stdout=PIPE, stderr=PIPE)
         out, err = p.communicate()
         if p.returncode != 0:
@@ -150,7 +153,7 @@ class PkgCreator:
             raise
 
     def _sign_repomd(self):
-        p = Popen(["gpg", "--batch", "--detach-sign", "--armor", "./rpm/repodata/repomd.xml"], stdout=PIPE, stderr=PIPE)
+        p = Popen(["gpg", "--batch", "--yes", "--detach-sign", "--armor", "./rpm/repodata/repomd.xml"], stdout=PIPE, stderr=PIPE)
         out, err = p.communicate()
         if p.returncode != 0:
             print(f"sign repomd result => {out}, {err}")
@@ -176,7 +179,7 @@ class PkgCreator:
                         old_path = os.path.join(root, file)
                         new_path = os.path.join(new_dir, file)
                         shutil.copyfile(old_path, new_path)
-                        self._sign_rpms(new_path)
+                        self._sign_rpms(new_path, gpg_key_name)
 
     """
         imports gpg keys into the system so reprepro and createrepo can use it to sign packages.
@@ -186,11 +189,34 @@ class PkgCreator:
     def import_gpg_keys(self, private_key, public_key):
         gpg = gnupg.GPG()
         private_key = base64.b64decode(private_key)
-        gpg.import_keys(private_key)
+        import_result = gpg.import_keys(private_key)
+        if not import_result.fingerprints:
+            raise Exception("Failed to import private key")
+
         public_key = base64.b64decode(public_key)
         gpg.import_keys(public_key)
+
+        imported_fingerprint = import_result.fingerprints[0]
         data = gpg.list_keys(secret=True)
-        return (data[0]["fingerprint"], data[0]["uids"][0])
+
+        # Find the specific key we just imported by comparing fingerprints
+        for key in data:
+            if key["fingerprint"] == imported_fingerprint:
+                return (key["fingerprint"], key["uids"][0])
+
+        raise Exception(f"Could not find imported key with fingerprint {imported_fingerprint}")
+
+    def import_multiple_gpg_keys(self, primary_private_key, primary_public_key, secondary_private_key=None, secondary_public_key=None):
+        """
+        Import one or two GPG keypairs. Returns a list of (fingerprint, uid) with the primary first.
+        """
+        results = []
+        if primary_private_key and primary_public_key:
+            results.append(self.import_gpg_keys(primary_private_key, primary_public_key))
+        if secondary_private_key and secondary_public_key:
+            # Ensure secondary is imported and appended
+            results.append(self.import_gpg_keys(secondary_private_key, secondary_public_key))
+        return results
 
     """
         basically rpm --import <key_file>
@@ -247,11 +273,13 @@ def upload_from_directories(pkg_uploader, directory, release, binary):
 """
 
 
-def create_deb_packaging(pkg_creator, pkg_uploader, releases, gpg_key_id, binary_name, archs, package_component,
+def create_deb_packaging(pkg_creator, pkg_uploader, releases, primary_gpg_key_id, secondary_gpg_key_id, binary_name, archs, package_component,
                          release_version):
     # set configuration for package creation.
     print(f"initialising configuration for {binary_name} , {archs}")
     Path("./conf").mkdir(parents=True, exist_ok=True)
+    # If in rollover mode (secondary provided), tell reprepro to sign with both keys.
+    sign_with_ids = primary_gpg_key_id if not secondary_gpg_key_id else f"{primary_gpg_key_id} {secondary_gpg_key_id}"
     pkg_creator.create_distribution_conf(
         "./conf/distributions",
         binary_name,
@@ -260,7 +288,7 @@ def create_deb_packaging(pkg_creator, pkg_uploader, releases, gpg_key_id, binary
         archs,
         package_component,
         f"apt repository for {binary_name}",
-        gpg_key_id)
+        sign_with_ids)
 
     # create deb pkgs
     for release in releases:
@@ -287,15 +315,19 @@ def create_rpm_packaging(
         gpg_key_name,
         base_url,
         gpg_key_url,
+        upload_repo_file=False,
 ):
     print(f"creating rpm pkgs...")
     pkg_creator.create_rpm_pkgs(artifacts_path, gpg_key_name)
-    pkg_creator.create_repo_file(artifacts_path, binary_name, base_url, gpg_key_url)
+    repo_file = pkg_creator.create_repo_file(artifacts_path, binary_name, base_url, gpg_key_url)
+
+    print("Uploading repo file")
+    pkg_uploader.upload_pkg_to_r2(repo_file, binary_name + ".repo")
 
     print("uploading latest to r2...")
     upload_from_directories(pkg_uploader, "rpm", None, binary_name)
 
-    if release_version:
+    if upload_repo_file:
         print(f"uploading versioned release {release_version} to r2...")
         upload_from_directories(pkg_uploader, "rpm", release_version, binary_name)
 
@@ -336,11 +368,23 @@ def parse_args():
             signing packages"
     )
 
+    # Optional secondary keypair for key rollover
+    parser.add_argument(
+        "--gpg-private-key-2", default=os.environ.get("LINUX_SIGNING_PRIVATE_KEY_2"), help="Secondary GPG private key for rollover"
+    )
+    parser.add_argument(
+        "--gpg-public-key-2", default=os.environ.get("LINUX_SIGNING_PUBLIC_KEY_2"), help="Secondary GPG public key for rollover"
+    )
+
     parser.add_argument(
         "--gpg-public-key-url", default=os.environ.get("GPG_PUBLIC_KEY_URL"), help="GPG public key url that\
             downloaders can use to verify signing"
     )
 
+    parser.add_argument(
+        "--gpg-public-key-url-2", default=os.environ.get("GPG_PUBLIC_KEY_URL_2"), help="Secondary GPG public key url for rollover"
+    )
+
     parser.add_argument(
         "--pkg-upload-url", default=os.environ.get("PKG_URL"), help="URL to be used by downloaders"
     )
@@ -355,6 +399,10 @@ def parse_args():
             it is the caller's responsiblity to ensure that these debs are already present in a directory. This script\
             will not build binaries or create their debs."
     )
+
+    parser.add_argument(
+        "--upload-repo-file", action='store_true', help="Upload RPM repo file to R2"
+    )
     args = parser.parse_args()
 
     return args
@@ -368,13 +416,36 @@ if __name__ == "__main__":
         exit(1)
 
     pkg_creator = PkgCreator()
-    (gpg_key_id, gpg_key_name) = pkg_creator.import_gpg_keys(args.gpg_private_key, args.gpg_public_key)
+    # Import one or two keypairs; primary first
+    key_results = pkg_creator.import_multiple_gpg_keys(
+        args.gpg_private_key,
+        args.gpg_public_key,
+        args.gpg_private_key_2,
+        args.gpg_public_key_2,
+    )
+    if not key_results or len(key_results) == 0:
+        raise SystemExit("No GPG keys were provided for signing")
+    primary_gpg_key_id, primary_gpg_key_name = key_results[0]
+    secondary_gpg_key_id = None
+    secondary_gpg_key_name = None
+    if len(key_results) > 1:
+        secondary_gpg_key_id, secondary_gpg_key_name = key_results[1]
+    # Import RPM public keys (one or two)
     pkg_creator.import_rpm_key(args.gpg_public_key)
 
     pkg_uploader = PkgUploader(args.account, args.bucket, args.id, args.secret)
-    print(f"signing with gpg_key: {gpg_key_id}")
-    create_deb_packaging(pkg_creator, pkg_uploader, args.deb_based_releases, gpg_key_id, args.binary, args.archs,
-                         "main", args.release_tag)
+    print(f"signing with primary gpg_key: {primary_gpg_key_id} and secondary gpg_key: {secondary_gpg_key_id}")
+    create_deb_packaging(
+        pkg_creator,
+        pkg_uploader,
+        args.deb_based_releases,
+        primary_gpg_key_id,
+        secondary_gpg_key_id,
+        args.binary,
+        args.archs,
+        "main",
+        args.release_tag,
+    )
 
     create_rpm_packaging(
         pkg_creator,
@@ -382,7 +453,8 @@ if __name__ == "__main__":
         "./built_artifacts",
         args.release_tag,
         args.binary,
-        gpg_key_name,
-        args.gpg_public_key_url,
+        primary_gpg_key_name,
         args.pkg_upload_url,
+        args.gpg_public_key_url,
+        args.upload_repo_file,
     )