feat!: keep the peer's close code over qmux and raw QUIC, on web-transport-trait 0.5

Cargo.toml

@@ -152,8 +152,8 @@ moq-mux = { version = "0.10.6", path = "rs/moq-mux" }
 moq-net = { version = "0.3.5", path = "rs/moq-net" }
 # The MoQ fork of noq (moq-dev/noq). iroh keeps upstream noq, so a build with the
 # iroh feature carries both stacks.
-moq-noq-proto = { version = "1.3.1", default-features = false }
-moq-noq-udp = "1.3.1"
+moq-noq-proto = { version = "2.0", default-features = false }
+moq-noq-udp = "2.0"
 # NVENC bindings, forked from ViliamVadocz/nvidia-video-codec-sdk to dlopen the
 # driver at runtime. Compiles on any platform (macOS included) but only actually
 # used by moq-video on Linux.
@@ -204,7 +204,7 @@ object_store = { version = "0.14", default-features = false }
 percent-encoding = "2"
 pollster = "1.0"
 pulldown-cmark = { version = "0.13", default-features = false }
-qmux = { version = "0.5.1", default-features = false }
+qmux = { version = "0.6", default-features = false }
 rand = "0.10.1"
 # default-features off so each consumer picks its own TLS backend and body
 # features; nothing here wants the default native-tls stack.
@@ -234,14 +234,14 @@ url = "2"
 usage = { package = "usage-rs", version = "6.11.1" }
 uuid = { version = "1.26", features = ["v7"] }
 web-async = { version = "0.1.5", features = ["tracing"] }
-web-transport-iroh = "0.7"
+web-transport-iroh = "0.8"
 # default-features off so the QUIC crypto provider is chosen by moq-tokio's
 # aws-lc-rs / ring features.
 # The WebTransport adapter released with moq-noq, from the same repository.
-web-transport-moq = { version = "1.3.1", default-features = false }
+web-transport-moq = { version = "2.0", default-features = false }
 web-transport-proto = "0.6"
-web-transport-trait = "0.4"
-web-transport-wasm = "0.6"
+web-transport-trait = "0.5"
+web-transport-wasm = "0.7"
 x11rb = { version = "0.14.0", features = ["randr", "xfixes"] }
 zeroize = { version = "1", features = ["derive"] }
 

quest/m1/README.md

@@ -28,7 +28,6 @@ transport, benchmark tooling); worktrees isolate commits, not semantics.
 - [FFI shape](/quest/m1/ffi-shape/README.md) - the bindings mirror Rust's layers: net at the root, then media, json, audio, and video namespaces built from the handle below
 - [Track demand](/quest/m1/track-demand.md) - Rust and JS watch a track's subscribers through `demand()` alone
 - [Session close](/quest/m1/session-close.md) - a graceful session end withdraws announces and waits one second for the ack
-- [Close codes](/quest/m1/close-codes.md) - a client sees the peer's application close code over WebSocket and raw QUIC, like WebTransport
 - [JS active count](/quest/m1/js-active-count.md) - @moq/net speaks MoQ Active Count, so its IETF announce consumers go live without a timer
 - [Page-load marker](/quest/m1/announce-page-load.md) - an announcement stream opened before the first connection waits for its replay before `live`
 - [Empty state](/quest/m1/announce-empty-state.md) - watch, room, and the demo show "no broadcasts" once `live` arrives with nothing announced

quest/m1/quic/qmux.md

@@ -38,8 +38,8 @@ Add the missing wire evidence before release: golden draft-02 vectors,
 bidirectional interoperability against the published `qmux` 0.5.x crate, and
 the TypeScript qmux/WebSocket peer used by `js/net`. Preserve rejection of
 prohibited QUIC frames, params-first setup, record-size validation, close and
-reset semantics (the first recorded close wins, per
-[close codes](/quest/m1/close-codes.md)), keep-alive behavior, and bounded
+reset semantics (the first recorded close wins, which moq-tokio's
+`close_code` test checks end to end), keep-alive behavior, and bounded
 flow-control tests.
 
 The crate lives in the fork's workspace as `moq-noq-qmux`, so the stream

rs/moq-e2ee/tests/support/mock.rs

@@ -118,7 +118,7 @@ impl poll::SendStream for MockSendStream {
 		}
 	}
 
-	fn set_priority(&mut self, _order: u8) {}
+	fn set_priority(&mut self, _order: i32) {}
 
 	fn finish(&mut self) -> Result<(), Self::Error> {
 		if let Some(tx) = self.tx.take() {

rs/moq-net/src/client.rs

@@ -606,7 +606,7 @@ mod tests {
 			Poll::Ready(Ok(buf.len()))
 		}
 
-		fn set_priority(&mut self, _order: u8) {}
+		fn set_priority(&mut self, _order: i32) {}
 
 		fn finish(&mut self) -> Result<(), Self::Error> {
 			Ok(())
@@ -957,7 +957,7 @@ mod tests {
 			self.inner.poll_write(cx, buf)
 		}
 
-		fn set_priority(&mut self, order: u8) {
+		fn set_priority(&mut self, order: i32) {
 			self.inner.set_priority(order);
 		}
 

rs/moq-net/src/coding/writer.rs

@@ -174,7 +174,7 @@ impl<S: crate::transport::poll::SendStream, V: StreamCodes> Writer<S, V> {
 	/// where higher values preempt lower ones. The lite priority queue's rank is the
 	/// opposite (0 = most urgent); rank holders convert via `PriorityHandle::send_order`.
 	pub fn set_priority(&mut self, send_order: u8) {
-		self.stream.as_mut().unwrap().set_priority(send_order);
+		self.stream.as_mut().unwrap().set_priority(send_order.into());
 	}
 
 	/// Cast the writer to a different version, used during version negotiation.
@@ -235,7 +235,7 @@ mod tests {
 			Poll::Ready(Err(self.clone()))
 		}
 
-		fn set_priority(&mut self, _: u8) {}
+		fn set_priority(&mut self, _: i32) {}
 
 		fn finish(&mut self) -> Result<(), Self::Error> {
 			Err(self.clone())

rs/moq-net/src/ietf/adapter.rs

@@ -376,7 +376,7 @@ impl web_transport_trait::poll::SendStream for VirtualSendStream {
 		Poll::Ready(self.push(chunk).map(|()| len))
 	}
 
-	fn set_priority(&mut self, _order: u8) {}
+	fn set_priority(&mut self, _order: i32) {}
 
 	fn finish(&mut self) -> Result<(), Self::Error> {
 		// Flush any remaining buffered data (e.g. if registration never completed).
@@ -419,7 +419,7 @@ impl<S: crate::transport::poll::Session> web_transport_trait::poll::SendStream f
 		}
 	}
 
-	fn set_priority(&mut self, order: u8) {
+	fn set_priority(&mut self, order: i32) {
 		match self {
 			Self::Real(s) => s.set_priority(order),
 			Self::Virtual(s) => s.set_priority(order),

rs/moq-net/src/ietf/publisher.rs

@@ -2112,7 +2112,7 @@ impl<S: crate::transport::poll::Session> TrackServe<S> {
 		let mut stream = std::future::poll_fn(|cx| self.session.poll_open_uni(cx))
 			.await
 			.map_err(Error::from_transport)?;
-		stream.set_priority(priority);
+		stream.set_priority(priority.into());
 
 		let mut writer = Writer::new(stream, self.version);
 		writer.buffer(&ietf::GroupHeader {
@@ -2303,7 +2303,7 @@ impl<S: crate::transport::poll::Session> GroupServe<S> {
 					};
 					self.opened.fetch_add(1, Ordering::Relaxed);
 					let mut stream = stream;
-					stream.set_priority(self.priority);
+					stream.set_priority(self.priority.into());
 
 					let mut writer = Writer::new(stream, self.version);
 					if let Err(err) = writer.buffer(&self.msg) {

rs/moq-net/src/lite/test_transport.rs

@@ -133,7 +133,8 @@ impl poll::SendStream for SinkSend {
 		Poll::Ready(Ok(buf.len()))
 	}
 
-	fn set_priority(&mut self, order: u8) {
+	fn set_priority(&mut self, order: i32) {
+		let order = u8::try_from(order).expect("moq-net sends u8 send orders");
 		self.log.priorities.lock().unwrap().push(order);
 	}
 

rs/moq-net/src/server.rs

@@ -879,7 +879,7 @@ mod tests {
 		) -> std::task::Poll<Result<usize, Self::Error>> {
 			std::task::Poll::Ready(Ok(buf.len()))
 		}
-		fn set_priority(&mut self, _order: u8) {}
+		fn set_priority(&mut self, _order: i32) {}
 		fn finish(&mut self) -> Result<(), Self::Error> {
 			Ok(())
 		}

rs/moq-net/tests/support/mock.rs

@@ -144,7 +144,7 @@ impl poll::SendStream for MockSendStream {
 		)
 	}
 
-	fn set_priority(&mut self, _order: u8) {}
+	fn set_priority(&mut self, _order: i32) {}
 
 	fn finish(&mut self) -> Result<(), Self::Error> {
 		if self.tx.is_some() {

rs/moq-tokio/src/transport.rs

@@ -211,7 +211,7 @@ pub struct SendStream<S: web_transport_trait::SendStream + 'static> {
 	// `Sync` off the containing session types.
 	state: std::sync::Mutex<Option<SendState<S>>>,
 	// Deferred actions, applied when the in-flight operation settles.
-	priority: Option<u8>,
+	priority: Option<i32>,
 	finish: bool,
 	reset: Option<u32>,
 	/// Whether `finish` or `reset` has been observed, so no further writes can
@@ -370,7 +370,7 @@ impl<S: web_transport_trait::SendStream + 'static> wt_poll::SendStream for SendS
 		}
 	}
 
-	fn set_priority(&mut self, order: u8) {
+	fn set_priority(&mut self, order: i32) {
 		match self.state.get_mut().unwrap().as_mut() {
 			Some(SendState::Idle(stream)) => stream.set_priority(order),
 			_ => self.priority = Some(order),
@@ -807,7 +807,7 @@ mod tests {
 	struct FakeSend {
 		writes: Arc<Mutex<Vec<u8>>>,
 		blocked: Arc<AtomicBool>,
-		priorities: Arc<Mutex<Vec<u8>>>,
+		priorities: Arc<Mutex<Vec<i32>>>,
 		finished: Arc<AtomicBool>,
 		resets: Arc<Mutex<Vec<u32>>>,
 		/// The peer never acknowledges the FIN: closed() stays pending forever.
@@ -830,7 +830,7 @@ mod tests {
 			Ok(buf.len())
 		}
 
-		fn set_priority(&mut self, order: u8) {
+		fn set_priority(&mut self, order: i32) {
 			self.priorities.lock().unwrap().push(order);
 		}