refactor(uring)!: derive worker and steering identity from the socket

quest/dev/release.md

@@ -87,7 +87,6 @@ Public API: none beyond the required quests. Wire: none.
 
 ## Required
 
-- [uring identity](/quest/main/uring-identity.md) - bind sockets, connections, workers, and steering identity together
 - [Binding audio tests](/quest/next/binding-audio-tests.md) - every binding proves the audio config it exposes
 - [Decode format](/quest/next/ffi-decode-format.md) - the C-only decode knob reaches every uniffi binding
 - [JSON mutate](/quest/next/json-mutate.md) - Rust and JS share the closure edit

quest/main/README.md

@@ -88,7 +88,6 @@ do not add another media abstraction or a renderer crate during stabilization.
 
 ## Quests
 
-- [uring identity](/quest/main/uring-identity.md) - sockets and connections carry their worker and steering identity
 - [GPU conversion and NVENC](/quest/main/video-gpu-encode.md) - convert, resize and
   encode imported frames without CPU pixel transfers or fallback
 - [Audio configuration](/quest/main/audio-config.md) - PCM layout, codec settings, and subscription policy have distinct contracts

rs/moq-relay/src/uring.rs

@@ -23,12 +23,6 @@ use anyhow::Context as _;
 
 use crate::{auth, cluster, shutdown};
 
-/// One member's bound socket and its slot in the steered group.
-struct Member {
-	shard: moq_sock::shard::Shard,
-	socket: std::net::UdpSocket,
-}
-
 /// A stop signal a worker parks on, wakeable from the shared runtime.
 #[derive(Default)]
 struct Stop {
@@ -83,7 +77,7 @@ struct Serve {
 /// worker that quietly died. Nothing is served until [`serve`](Self::serve)
 /// spawns the threads.
 pub struct Workers {
-	members: Vec<Member>,
+	members: Vec<moq_sock::shard::Socket>,
 	addr: SocketAddr,
 	server: moq_uring::quic::server::Config,
 	/// What the workers serve, fingerprinted once, for `/certificate.sha256`.
@@ -199,10 +193,7 @@ impl Workers {
 			.context("failed to complete the reuseport group")?;
 		let mut members = Vec::with_capacity(count as usize);
 		while let Some(member) = group.member().context("failed to clone a reuseport member")? {
-			members.push(Member {
-				shard: member.shard(),
-				socket: member.into_inner(),
-			});
+			members.push(member);
 		}
 		// Whatever the first member bound, which is the requested address unless
 		// it asked for an ephemeral port.
@@ -325,7 +316,7 @@ impl Workers {
 		};
 
 		for member in std::mem::take(&mut self.members) {
-			let index = member.shard.index();
+			let index = member.shard().index();
 			let core = cores.get(index as usize % cores.len().max(1)).copied();
 			let stop = Arc::new(Stop::default());
 			let (ready_tx, ready_rx) = std::sync::mpsc::channel();
@@ -478,7 +469,7 @@ fn join(threads: Vec<(u16, std::thread::JoinHandle<()>)>) {
 
 /// Everything one worker thread is handed at spawn.
 struct Spawn {
-	member: Member,
+	member: moq_sock::shard::Socket,
 	/// The core to pin to, or `None` when pinning is off or unavailable.
 	core: Option<moq_sock::cpu::CoreId>,
 	/// This worker's counter set, shared with whoever scrapes `/metrics`.
@@ -503,7 +494,7 @@ struct Spawn {
 /// alone would not notice one, since the parent holds a sender of its own and
 /// the channel therefore never closes.
 fn run_worker(spawn: Spawn) {
-	let index = spawn.member.shard.index();
+	let index = spawn.member.shard().index();
 	let stop = spawn.stop.clone();
 	let failures = spawn.failures.clone();
 
@@ -537,7 +528,7 @@ fn serve_worker(spawn: Spawn) -> bool {
 		ready,
 		failures,
 	} = spawn;
-	let index = member.shard.index();
+	let index = member.shard().index();
 	if let Some(core) = core {
 		if moq_sock::cpu::pin(core) {
 			tracing::debug!(index, core = core.id(), "pinned io_uring QUIC worker");
@@ -550,18 +541,15 @@ fn serve_worker(spawn: Spawn) -> bool {
 		let mut config = moq_uring::Config::default();
 		config.metrics = metrics;
 		let worker = moq_uring::Worker::new(config).context("io_uring setup failed")?;
-		let handle = worker.handle();
-		let socket = handle
-			.udp(member.socket, udp)
+		// The member carries its slot in the steered group, so adopting it is
+		// what makes the endpoint issue connection ids that steer back here.
+		let socket = worker
+			.handle()
+			.udp(member, udp)
 			.context("failed to adopt the worker socket")?;
-		let endpoint = moq_uring::quic::Endpoint::new(
-			&handle,
-			socket,
-			moq_uring::quic::endpoint::Config::default()
-				.with_server(server)
-				.with_shard(member.shard),
-		)
-		.context("failed to build the QUIC endpoint")?;
+		let endpoint =
+			moq_uring::quic::Endpoint::new(socket, moq_uring::quic::endpoint::Config::default().with_server(server))
+				.context("failed to build the QUIC endpoint")?;
 		Ok((worker, endpoint))
 	})();
 
@@ -638,7 +626,7 @@ async fn serve_connection(
 	let mut alpn = conn.protocol().map(str::to_owned);
 	let (transport, url) = match conn.protocol() {
 		Some("h3") => {
-			let request = moq_uring::quic::web::Request::accept(&handle, conn)
+			let request = moq_uring::quic::web::Request::accept(conn)
 				.await
 				.context("WebTransport handshake failed")?;
 			let url = request.url().clone();

rs/moq-relay/tests/runtime_uring.rs

@@ -377,7 +377,7 @@ async fn uring_workers_write_qlog_traces() {
 	// rather than a connection that only ever exchanged Initials.
 	let url: url::Url = format!("moql://127.0.0.1:{port}/qlog").parse().expect("parse url");
 	let origin = moq_tokio::origin::spawn();
-	let mut broadcast = origin.create_broadcast("test").expect("create broadcast");
+	let broadcast = origin.create_broadcast("test").expect("create broadcast");
 	broadcast.announce(Default::default()).expect("create broadcast");
 	let track = broadcast.create_track("video", None).expect("create track");
 	let mut group = track.append_group().expect("append group");

rs/moq-uring/README.md

@@ -48,10 +48,14 @@ the UDP sockets bound through it.
   thread that spawned the worker, or read the worker's own with
   `Handle::metrics`. `moq-relay` publishes them at `/metrics` on its internal
   listener.
-- **Steering**: an endpoint whose socket sits in a `moq-sock` steered
-  `SO_REUSEPORT` group sets `endpoint::Config::shard`, and every issued
-  connection id leads with the group's steering byte, so the kernel keeps a
-  connection (and a cluster dial's responses) on the worker that owns it.
+- **Identity**: the socket names its worker. `Handle::udp` adopts a lone
+  `UdpSocket` or a member of a completed `moq-sock` steered `SO_REUSEPORT`
+  group (`udp::Bound`), and a `quic::Endpoint` built on it runs its demux and
+  every connection driver on that worker, whichever handle built it. A member
+  brings its slot along, so every issued connection id leads with the group's
+  steering byte and the kernel keeps a connection (and a cluster dial's
+  responses) on the worker that owns it. An endpoint on a dropped worker is
+  refused.
 
 Requires **Linux 6.12**; `Worker::new` refuses older kernels with a legible
 error rather than degrading (note that default container seccomp policies

rs/moq-uring/benches/echo_noq.rs

@@ -52,8 +52,8 @@ mod linux {
 		let socket = handle
 			.udp(UdpSocket::bind("127.0.0.1:0").expect("bind"), udp_config)
 			.expect("socket");
-		let endpoint = quic::Endpoint::new(&handle, socket, quic::endpoint::Config::default().with_server(server))
-			.expect("endpoint");
+		let endpoint =
+			quic::Endpoint::new(socket, quic::endpoint::Config::default().with_server(server)).expect("endpoint");
 		let addr = endpoint.local_addr();
 		// Queue the first iteration while the worker finishes driving the client
 		// handshake. A zero-capacity channel would block this thread before it can
@@ -87,7 +87,7 @@ mod linux {
 		let mut session = worker
 			.block_on(async {
 				let conn = endpoint.accept().await.expect("accept");
-				quic::web::Request::accept(&handle, conn)
+				quic::web::Request::accept(conn)
 					.await
 					.expect("handshake")
 					.ok()

rs/moq-uring/benches/session_lite.rs

@@ -125,7 +125,7 @@ mod linux {
 
 			let server_handle = handle.clone();
 			handle.spawn(async move {
-				let conn = quic::server::accept(&server_handle, server_sock, &server_config)
+				let conn = quic::server::accept(server_sock, &server_config)
 					.await
 					.expect("quic accept");
 				let (session, driver) = moq_net::Server::new()
@@ -140,9 +140,7 @@ mod linux {
 			// Establish and subscribe once; iterations measure steady state.
 			let (_session, mut sub) = worker
 				.block_on(async {
-					let conn = quic::client::connect(&handle, client_sock, &dial)
-						.await
-						.expect("quic connect");
+					let conn = quic::client::connect(client_sock, &dial).await.expect("quic connect");
 					let (session, driver) = moq_net::Client::new()
 						.with_subscriber(sub_origin.clone())
 						.connect_lite(std::time::Instant::now(), quic::web::Session::raw(conn))

rs/moq-uring/src/lib.rs

@@ -16,9 +16,13 @@
 //! serves many connections on one socket (demuxed by connection id, dials
 //! included), each a [`quic::Connection`] implementing the transport traits,
 //! so `moq_net::Client::connect_lite` and `Server::accept_lite` run real
-//! moq-lite sessions on the worker. [`Handle::run`] supplies time and schedules driver wakeups;
-//! callers run the returned drivers with [`Handle::spawn`]. The stack underneath is enabled by
-//! the `noq` feature; a build without it leaves the module out.
+//! moq-lite sessions on the worker. The socket is the identity: whatever is
+//! built on it runs on the worker that adopted it, and a socket adopted as a
+//! member of a steered reuseport group ([`udp::Bound`]) issues connection ids
+//! that steer back to it. [`Handle::run`] supplies time and schedules driver
+//! wakeups; callers run the returned drivers with [`Handle::spawn`]. The stack
+//! underneath is enabled by the `noq` feature; a build without it leaves the
+//! module out.
 //!
 //! [`metrics::Metrics`] is how the worker's own health leaves its thread:
 //! relaxed counters for the buffer pools, the batching mechanisms, the ring,

rs/moq-uring/src/quic/client.rs

@@ -3,7 +3,7 @@
 use std::net::SocketAddr;
 
 use super::{Connection, Error, Identity};
-use crate::{Handle, udp};
+use crate::udp;
 
 /// Where to dial, as whom, and who to trust.
 #[derive(Clone, Debug)]
@@ -52,10 +52,10 @@ impl Config {
 ///
 /// Shorthand for a dial-only [`Endpoint`](super::Endpoint) and one
 /// [`connect`](super::Endpoint::connect) through it. The connection's driver
-/// runs as a task on the worker behind `handle`, so the returned
+/// runs as a task on the worker that adopted `socket`, so the returned
 /// [`Connection`] just works: hand it to `moq_net::Client::connect_lite` or
 /// use the stream API directly.
-pub async fn connect(handle: &Handle, socket: udp::Socket, config: &Config) -> Result<Connection, Error> {
-	let endpoint = super::Endpoint::new(handle, socket, super::endpoint::Config::default())?;
+pub async fn connect(socket: udp::Socket, config: &Config) -> Result<Connection, Error> {
+	let endpoint = super::Endpoint::new(socket, super::endpoint::Config::default())?;
 	endpoint.connect(config).await
 }

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

@@ -13,12 +13,15 @@
 //! a migrating client stays routable; an Initial for an unsupported version
 //! gets a version negotiation packet back.
 //!
-//! An endpoint whose socket is a member of a steered `SO_REUSEPORT` group
-//! (one worker per core on one port) sets [`Config::shard`], and every id it
-//! issues then carries the [`moq_sock::shard::cid_prefix`] steering byte, so
-//! the kernel keeps delivering a connection's packets to the worker that owns
-//! it. Dials through the endpoint carry it too, which is what steers a
-//! cluster peer's responses back to the dialing worker.
+//! The socket is the endpoint's identity. Its worker is where the demux and
+//! every connection driver run, whichever thread built the endpoint. A socket
+//! adopted as a member of a steered `SO_REUSEPORT` group (one worker per core
+//! on one port; see [`udp::Bound`](crate::udp::Bound)) brings its slot along,
+//! and every id the endpoint issues then carries the
+//! [`moq_sock::shard::cid_prefix`] steering byte, so the kernel keeps
+//! delivering a connection's packets to the worker that owns it. Dials
+//! through the endpoint carry it too, which is what steers a cluster peer's
+//! responses back to the dialing worker.
 
 use super::server;
 
@@ -44,23 +47,13 @@ pub struct Config {
 	/// peer retransmits and lands once the application drains the backlog or
 	/// a handshake fails.
 	pub backlog: usize,
-
-	/// This socket's slot in a steered `SO_REUSEPORT` group, if it is in one.
-	///
-	/// Every connection id the endpoint issues then leads with the slot's
-	/// [`cid_prefix`](moq_sock::shard::cid_prefix) byte, which is what the
-	/// group's filter selects on. Set it if and only if the socket was bound
-	/// into a group with this shard; a lone socket leaves it `None` and keeps
-	/// the whole id random.
-	pub shard: Option<moq_sock::shard::Shard>,
 }
 
 impl Default for Config {
 	fn default() -> Self {
 		Self {
 			server: None,
 			backlog: 1024,
-			shard: None,
 		}
 	}
 }
@@ -71,16 +64,10 @@ impl Config {
 		self.server = Some(server);
 		self
 	}
-
-	/// Issue connection ids steering to `shard`'s slot of a reuseport group.
-	pub fn with_shard(mut self, shard: moq_sock::shard::Shard) -> Self {
-		self.shard = Some(shard);
-		self
-	}
 }
 
 /// A fresh [`CID_LEN`]-byte connection id, leading with the steering prefix
-/// when the endpoint sits in a reuseport group.
+/// when the endpoint's socket is a member of a reuseport group.
 pub(crate) fn cid(shard: Option<moq_sock::shard::Shard>) -> [u8; CID_LEN] {
 	let mut cid: [u8; CID_LEN] = rand::random();
 	if let Some(shard) = shard {

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

@@ -11,7 +11,8 @@ use rustc_hash::FxHashMap;
 
 use super::super::{Error, SEGMENT};
 use super::endpoint;
-use crate::{Handle, udp};
+use crate::udp;
+use crate::worker::Owner;
 
 /// The state shared by every handle and the driver, single-threaded behind
 /// `Rc<RefCell>`.
@@ -28,6 +29,9 @@ const TRAIN_SEGMENTS: usize = 63;
 pub(crate) struct Inner {
 	pub(crate) conn: RefCell<noq_proto::Connection>,
 	pub(crate) state: RefCell<State>,
+	/// The worker driving this connection, which anything layered on it
+	/// (the WebTransport handshake, say) runs on too.
+	pub(crate) owner: Owner,
 }
 
 pub(crate) struct State {
@@ -285,6 +289,11 @@ impl Connection {
 		self.shared.close_code(code, reason);
 	}
 
+	/// The worker driving this connection.
+	pub(crate) fn owner(&self) -> &Owner {
+		&self.shared.owner
+	}
+
 	/// The peer's certificate chain in DER, leaf first, or `None` if it
 	/// presented none.
 	///
@@ -320,7 +329,7 @@ impl Clone for Connection {
 /// [kicks](Inner::kick) the driver. The caller spawns the future and reclaims
 /// the connection's bookkeeping once it resolves.
 pub(crate) fn launch(
-	handle: &Handle,
+	owner: &Owner,
 	socket: Rc<udp::Socket>,
 	endpoint: Weak<endpoint::Inner>,
 	key: ConnectionHandle,
@@ -329,14 +338,15 @@ pub(crate) fn launch(
 	let shared = Rc::new(Inner {
 		conn: RefCell::new(conn),
 		state: RefCell::new(State::new()),
+		owner: owner.clone(),
 	});
 
 	let mut driver = Driver {
 		shared: shared.clone(),
 		socket,
 		endpoint,
 		key,
-		deadline: crate::Timer::new(handle),
+		deadline: owner.timer(),
 		scratch: Vec::with_capacity(TRAIN_SEGMENTS * SEGMENT),
 		blocked: false,
 	};