feat(uring): carry ECN on the io_uring UDP path

quest/m2/quic/README.md

@@ -45,8 +45,6 @@ This is a transport API change, not a MoQ wire change.
   written down
 - [Deliver the application close before io_uring teardown](/quest/m2/quic/uring-close.md) -
   the peer receives the final close when the client immediately stops its worker
-- [ECN on the io_uring UDP path](/quest/m2/quic/ecn-uring.md) - the ring's
-  sends carry ECT(0) and its receives read the mark, matching `noq-udp`
 - [Measure ECN on the backbone](/quest/m2/quic/ecn-measure.md) - a written
   verdict on marking versus dropping, and whether Linode and OVH keep marks
 - [Per-stream ACK progress](/quest/m2/quic/ack-progress.md) - the fork reports

quest/m2/quic/ecn-measure.md

@@ -30,8 +30,3 @@ repeat it.
   and the tcpdump summaries beside the numbers in the L4S quest's Plan.
   If neither provider preserves the marks, say so there: L4S stays off and
   the marking response is only a lab result.
-
-## Required
-
-- [ECN on the io_uring UDP path](/quest/m2/quic/ecn-uring.md) - the relays
-  under test must mark

quest/m2/quic/ecn.md

@@ -9,8 +9,7 @@ marks fall back to no ECN, and a viewer's session is unaffected.
 
 ## Plan
 
-Classic ECN is already end to end once the
-[io_uring path marks](/quest/m2/quic/ecn-uring.md); this quest is the
+Classic ECN is already end to end on both runtimes; this quest is the
 fork-side half. noq-proto has no ECN knob: `sending_ecn` is hardcoded on
 per path and only an ACK without counts turns it off, so both `off` and
 `ect1` need the fork.
@@ -36,7 +35,5 @@ per path and only an ACK without counts turns it off, so both `off` and
 
 - [Fork noq](/quest/m2/quic/fork.md) - the `Ect1` option and the `off` knob
   live there
-- [ECN on the io_uring UDP path](/quest/m2/quic/ecn-uring.md) - the relays
-  must mark before a response can be measured
 - [Measure ECN on the backbone](/quest/m2/quic/ecn-measure.md) - the
   provider verdict this quest acts on

rs/moq-uring/src/quic/quiche/connection.rs

@@ -840,7 +840,12 @@ impl Driver {
 			// Nothing to send; the buffer returns to the pool on drop.
 			return Poll::Pending;
 		};
-		if let Err(err) = tx.send(filled, to, SEGMENT) {
+		if let Err(err) = tx.send(udp::Transmit {
+			to,
+			len: filled,
+			segment: SEGMENT,
+			ecn: None,
+		}) {
 			return Poll::Ready(Err(Error::Io(err.to_string())));
 		}
 		Poll::Ready(Ok(()))

rs/moq-uring/src/quic/quiche/endpoint.rs

@@ -473,7 +473,12 @@ impl Inner {
 		};
 		match quiche::negotiate_version(&hdr.scid, &hdr.dcid, &mut tx) {
 			Ok(len) => {
-				if let Err(err) = tx.send(len, to, len) {
+				if let Err(err) = tx.send(udp::Transmit {
+					to,
+					len,
+					segment: len,
+					ecn: None,
+				}) {
 					tracing::debug!(%err, "failed to send a version negotiation packet");
 				}
 			}

rs/moq-uring/src/quic/quinn/connection.rs

@@ -793,8 +793,13 @@ impl Driver {
 		tx[..transmit.size].copy_from_slice(&self.scratch[..transmit.size]);
 		// A lone datagram is its own segment size, and the socket's GSO
 		// stride has to match what quinn actually packed.
-		let segment = transmit.segment_size.unwrap_or(transmit.size);
-		if let Err(err) = tx.send(transmit.size, transmit.destination, segment) {
+		let transmit = udp::Transmit {
+			to: transmit.destination,
+			len: transmit.size,
+			segment: transmit.segment_size.unwrap_or(transmit.size),
+			ecn: transmit.ecn.map(super::ecn_from_quinn),
+		};
+		if let Err(err) = tx.send(transmit) {
 			return Poll::Ready(Err(Error::Io(err.to_string())));
 		}
 		// A flush frees datagram-send queue space.

rs/moq-uring/src/quic/quinn/endpoint.rs

@@ -223,6 +223,7 @@ impl Inner {
 	/// Route every datagram in one receive, then kick the connections fed.
 	fn demux(self: &Rc<Self>, packet: &mut udp::Packet) {
 		let from = packet.from();
+		let ecn = packet.ecn().map(super::ecn_to_quinn);
 		// Where the endpoint writes its own answers (version negotiation,
 		// retry, a refusal), reused across the whole receive.
 		let mut buf = Vec::new();
@@ -240,7 +241,7 @@ impl Inner {
 			let event = self.endpoint.borrow_mut().handle(
 				Instant::now(),
 				from.into(),
-				None,
+				ecn,
 				BytesMut::from(&segment[..]),
 				&mut buf,
 			);
@@ -249,7 +250,7 @@ impl Inner {
 				Instant::now(),
 				from,
 				None,
-				None,
+				ecn,
 				BytesMut::from(&segment[..]),
 				&mut buf,
 			);
@@ -373,8 +374,13 @@ impl Inner {
 			return;
 		}
 		tx[..transmit.size].copy_from_slice(&buf[..transmit.size]);
-		let segment = transmit.segment_size.unwrap_or(transmit.size);
-		if let Err(err) = tx.send(transmit.size, transmit.destination, segment) {
+		let transmit = udp::Transmit {
+			to: transmit.destination,
+			len: transmit.size,
+			segment: transmit.segment_size.unwrap_or(transmit.size),
+			ecn: transmit.ecn.map(super::ecn_from_quinn),
+		};
+		if let Err(err) = tx.send(transmit) {
 			tracing::debug!(%err, "failed to send an endpoint response");
 		}
 	}

rs/moq-uring/src/quic/quinn/mod.rs

@@ -31,6 +31,7 @@ use rustls::pki_types::{CertificateDer, PrivateKeyDer};
 #[cfg(feature = "qlog")]
 use super::qlog;
 use super::{Congestion, Error, Identity, SEGMENT, Transport, client, endpoint::CID_LEN, server};
+use crate::udp;
 
 /// Per-stream flow control credit, matching the quiche backend.
 const STREAM_WINDOW: u32 = 4 * 1024 * 1024;
@@ -40,6 +41,22 @@ const CONNECTION_WINDOW: u32 = 16 * 1024 * 1024;
 /// backend's 64.
 const DATAGRAM_WINDOW: usize = 64 * SEGMENT;
 
+fn ecn_to_quinn(ecn: udp::Ecn) -> quinn_proto::EcnCodepoint {
+	match ecn {
+		udp::Ecn::Ect0 => quinn_proto::EcnCodepoint::Ect0,
+		udp::Ecn::Ect1 => quinn_proto::EcnCodepoint::Ect1,
+		udp::Ecn::Ce => quinn_proto::EcnCodepoint::Ce,
+	}
+}
+
+fn ecn_from_quinn(ecn: quinn_proto::EcnCodepoint) -> udp::Ecn {
+	match ecn {
+		quinn_proto::EcnCodepoint::Ect0 => udp::Ecn::Ect0,
+		quinn_proto::EcnCodepoint::Ect1 => udp::Ecn::Ect1,
+		quinn_proto::EcnCodepoint::Ce => udp::Ecn::Ce,
+	}
+}
+
 impl From<quinn_proto::ConnectionError> for Error {
 	fn from(err: quinn_proto::ConnectionError) -> Self {
 		use quinn_proto::ConnectionError;