Skip to content

feat(tokio): accept_with and connect_with hand an established session to moq-net's policy #8085

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kixelated merged 4 commits into from
Nov 7, 2026
Merged

feat(tokio): accept_with and connect_with hand an established session to moq-net's policy #8085

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8a11da0
chore(quest): claim quest/m1/session-policy/tokio-handoff.md
kixelated Nov 6, 2026
aacc073
refactor(tokio)!: accept_with and connect_with hand an established qm…
kixelated Nov 6, 2026
85d7543
chore(quest): finish tokio-handoff
kixelated Nov 6, 2026
2785505
fix(tokio): a WebRTC accept counts toward the drain while establishing
kixelated Nov 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions quest/m1/session-policy/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,10 @@ where moq-webrtc should plug its externally established transport in. The
maintainer asked what that means for io_uring and how much of moq-tokio belongs
in moq-net, and decided:

- ✅ **Move the policy first**, then give moq-tokio thin handoffs; the
[WebRTC rename](/quest/m2/webrtc-rename.md) waits on
[moq-tokio becomes a thin shell](/quest/m1/session-policy/tokio-handoff.md)
(#7804 closed unmerged).
- ✅ **Move the policy first**, then give moq-tokio thin handoffs
(`Listener::accept_with`, `Client::connect_with`), which the
[WebRTC rename](/quest/m2/webrtc-rename.md) builds on (#7804 closed
unmerged).
Not chosen: a moq-tokio handoff now with the move later; moq-webrtc calling
moq-net directly with its own policy.
- ✅ **Reconnect is in scope**: the reconnect, backoff and GOAWAY candidate
Expand Down Expand Up @@ -66,4 +66,3 @@ Research (2026-11-04, `origin/main` at e6be13d5f8):
## Required

- [The io_uring relay path uses moq-net's policy](/quest/m1/session-policy/uring.md) - deletes `uring.rs`'s copy
- [moq-tokio becomes a thin shell](/quest/m1/session-policy/tokio-handoff.md) - sockets, TLS, accept loops and spawning, plus the transport handoffs WebRTC needs
36 changes: 0 additions & 36 deletions quest/m1/session-policy/tokio-handoff.md
View file
Open in desktop

This file was deleted.

6 changes: 1 addition & 5 deletions quest/m2/webrtc-rename.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,17 +11,13 @@ Decided 2026-10-30 with the maintainer (planning interview); see the [epic](/que
Milestone: ✅ m1 (moved from m2, maintainer, 2026-11-03). Not chosen: keep it in m2, or send [moq-rtc on rind](/quest/m1/rind/rtc.md) back to m2. [Vendor and trim the WebRTC stack](/quest/m1/webrtc-vendor.md) requires this, and m1's rind epic requires the vendoring, so this sits at the m2 top level. It stays a child of the m3 [WebRTC epic](/quest/m3/webrtc/README.md).

- Move `rs/moq-rtc` to `rs/moq-webrtc` and merge `rs/moq-ice` in as a module. Keep the gateway's `server` feature; add a feature for the data-channel transport so WHIP/WHEP-only builds skip SCTP.
- Move moq-tokio's `webrtc` module (`rs/moq-tokio/src/webrtc.rs`, host gathering, `.local` resolution) into moq-webrtc, built on moq-tokio's public session types. Its connect and accept go through moq-tokio's `connect_with`/`accept_with` handoffs, which run moq-net's session policy over the established data channel.
- Move moq-tokio's `webrtc` module (`rs/moq-tokio/src/webrtc.rs`, host gathering, `.local` resolution) into moq-webrtc, built on moq-tokio's public session types. Its connect and accept go through moq-tokio's `connect_with`/`accept_with` handoffs, which run moq-net's session policy over the established data channel. The handoffs take the qmux session and bound only the MoQ handshake from the call; bounding ICE and the channel coming up (moq-tokio's wrappers use the listen and connect timeouts today) moves with the transport, so moq-webrtc needs its own establishment bound.

@pullfrog pullfrog Bot Nov 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The note records the establishment bound that moves out with ICE, but accept_webrtc also does two things before the handoff that a Listener handle cannot express: it holds a moq_net::server::Active guard across establishment so a drain waits for a peer still coming up (the round-1 fix, pinned by a_drain_waits_for_an_establishing_peer), and it refuses a data-channel protocol the listener's Versions does not speak. Listener exposes neither its Activity nor its Versions, and accept_with counts the connection only from the handoff and never checks link.alpn, so the rename either plumbs both in from the app's server::Config/listen::Config or silently loses those guarantees. Recording that here (or folding the checks into accept_with) keeps the follow-up from regressing them.

Technical details 
# WebRTC handoff leaves the listener-identity checks behind

## Affected sites
- `quest/m2/webrtc-rename.md:14` — the new note covers only the establishment deadline.
- `rs/moq-tokio/src/server.rs:861` — `Listener::accept_with` enters `Activity` only at the handoff, and copies `link.alpn` through unchanged.
- `rs/moq-tokio/src/server.rs:887` — `_establishing` guard, on the private `Server::activity`.
- `rs/moq-tokio/src/server.rs:888` — `versions.alpns()` gate, on the private `Server::versions`.

## Required outcome
- Either the rename quest records that moq-webrtc also needs the listener's `Activity`
  (to hold `Active` across ICE/channel setup) and its `Versions` (to refuse a non-MoQ
  data-channel protocol), or `accept_with` owns both checks so a consumer cannot drop them.

## Suggested approach (optional)
- Add `Listener::activity()` / `Listener::versions()` accessors, or have `accept_with`
  validate `link.alpn` against the server's versions and accept a pre-acquired `Active`
  for the establishment window.
- The same gap applies to the `connect_with` side: `Client::versions` and `Client::policy`
  are `pub(crate)`, so a crate handed a `Client` cannot re-run `connect_webrtc`'s ALPN gate.

- Boundary decided 2026-11-04 (maintainer, from [moq-dev/moq#7804](https://github.com/moq-dev/moq/pull/7804)): ✅ move session policy into moq-net first ([Session policy lives in moq-net](/quest/m1/session-policy/README.md)), then this rename uses the thin handoffs. Not chosen: a moq-tokio handoff over today's policy, or moq-webrtc calling moq-net with its own policy. Removing SCTP stays with [Vendor and trim the WebRTC stack](/quest/m1/webrtc-vendor.md).
- Update moq-cli's `rtc` subcommand, the bindings or examples that use the transport, `test/p2p`, and every doc and quest naming the old crates.
- Publishing: the last `moq-rtc`/`moq-ice` releases stay on crates.io; `doc/setup/upgrade.md` says where things moved. Version numbers are the maintainer's.

Public API: breaking (crate rename; `moq_tokio::webrtc` moves). Wire: none.

## Required

- [moq-tokio becomes a thin shell](/quest/m1/session-policy/tokio-handoff.md) - the `accept_with`/`connect_with` handoffs moq-webrtc uses

## Related

- [moq-rtc REMB and TIAS](/rs/moq-rtc/src/server/whip.rs) - #7373 added the capped answer in moq-rtc; carry it through the move
71 changes: 44 additions & 27 deletions rs/moq-tokio/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -322,40 +322,57 @@ impl Client {
Some(self.with_subscriber(origin).connect(url))
}

/// Run a MoQ client session over the data channel an [`crate::webrtc::Offerer`] opened.
/// Run a MoQ client session over a qmux session established outside this client, such
/// as over a WebRTC data channel.
///
/// Waits for ICE and the channel to come up, bounded by the connect timeout. A
/// peer link cannot be redialed, so there is no reconnect: the session ends with
/// the peer connection.
#[cfg(feature = "webrtc")]
pub async fn connect_webrtc(&self, peer: &crate::webrtc::Offerer) -> crate::Result<moq_net::Session> {
/// The connect timeout bounds the MoQ handshake from this call. A transport handed
/// over cannot be redialed, so there is no reconnect: the session ends with it.
#[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))]
pub async fn connect_with(&self, session: qmux::Session) -> crate::Result<moq_net::Session> {
let timeout = self.policy.timeout;
let connect = connect_session(&self.moq, session);
crate::runtime::scope(async move {
let protocol = peer.protocol();
if !self.versions.alpns().contains(&protocol) {
return Err(crate::webrtc::Error::Protocol(protocol.to_string()).into());
match timeout.is_zero() {
true => Ok(connect.await?),
false => Ok(kio::clock::timeout(timeout, connect)
.await
.map_err(|_| Error::ConnectTimeout(timeout))??),
}
})
.await
}

let attempt = async {
let opened = peer.channel().await?;
let config = opened.config;
let session = qmux::Session::connect(opened.channel, config)
.await
.map(crate::spawn_qmux)
.map_err(crate::webrtc::Error::from)?;
Ok::<_, Error>(connect_session(&self.moq, session).await?)
};
/// Run a MoQ client session over the data channel an [`crate::webrtc::Offerer`] opened.
///
/// The connect timeout bounds ICE and the channel coming up, then
/// [`Self::connect_with`] bounds the MoQ handshake. A peer link cannot be redialed, so
/// there is no reconnect: the session ends with the peer connection.
#[cfg(feature = "webrtc")]
pub async fn connect_webrtc(&self, peer: &crate::webrtc::Offerer) -> crate::Result<moq_net::Session> {
let protocol = peer.protocol();
if !self.versions.alpns().contains(&protocol) {
return Err(crate::webrtc::Error::Protocol(protocol.to_string()).into());
}

let timeout = self.policy.timeout;
let session = match timeout.is_zero() {
true => attempt.await?,
false => kio::clock::timeout(timeout, attempt)
let establish = async {
let opened = peer.channel().await?;
let session = qmux::Session::connect(opened.channel, opened.config).await?;
Ok::<_, crate::webrtc::Error>(crate::spawn_qmux(session))
};
let timeout = self.policy.timeout;
let session = crate::runtime::scope(async move {
match timeout.is_zero() {
true => Ok(establish.await?),
false => kio::clock::timeout(timeout, establish)
.await
.map_err(|_| Error::ConnectTimeout(timeout))??,
};
tracing::info!(version = %session.version(), transport = %crate::Transport::WebRtc, "connected");
Ok(session)
.map_err(|_| Error::ConnectTimeout(timeout))?
.map_err(Error::from),
}
})
.await
.await?;
let session = self.connect_with(session).await?;
tracing::info!(version = %session.version(), transport = %crate::Transport::WebRtc, "connected");
Ok(session)
}

/// Dial one attempt of a [`Connection`]'s policy and complete the MoQ handshake.
Expand Down
Loading
Loading