#![cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::Arc;
use std::time::Duration;
use anyhow::{anyhow, bail, ensure, Context, Result};
use bytes::Bytes;
use moq_transport::coding::TrackNamespace;
use moq_transport::serve::{self, TrackReaderMode};
use tokio::sync::{Mutex, RwLock};
use super::{MoqMessageHandler, MoqTerminalHandler, NativeMoQState};
#[cfg(any(feature = "test-harness", feature = "moq-carrier-test"))]
pub mod test_relay;
const MOQ_SETUP_TIMEOUT: Duration = Duration::from_secs(10);
const MOQ_SUBSCRIBE_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
const MOQ_SUBSCRIBE_RETRY_DELAY: Duration = Duration::from_millis(250);
const MOQ_SUBSCRIBE_MAX_RETRY_DELAY: Duration = Duration::from_secs(8);
const MOQ_MAX_OBJECT_PAYLOAD_BYTES: usize = 4 * 1024 * 1024;
const MOQ_MAX_ACCESS_TOKEN_BYTES: usize = 16 * 1024;
const MOQ_MAX_BUFFERED_OBJECTS: usize = crate::iroh_carrier::DEFAULT_PACKET_QUEUE_CAPACITY;
const MOQ_MAX_BUFFERED_OBJECT_BYTES: usize = 8 * 1024 * 1024;
const MOQ_OBJECT_BACKPRESSURE_TIMEOUT: Duration = Duration::from_secs(2);
type Draft14TrackWriters = HashMap<String, Draft14TrackWriter>;
struct Draft14TrackWriter {
writer: serve::SubgroupWriter,
retained_object_bytes: VecDeque<usize>,
retained_bytes: usize,
}
impl Draft14TrackWriter {
fn new(writer: serve::SubgroupWriter) -> Self {
Self {
writer,
retained_object_bytes: VecDeque::new(),
retained_bytes: 0,
}
}
fn refresh_retention(&mut self) {
let retained_objects = self.writer.len();
while self.retained_object_bytes.len() > retained_objects {
if let Some(size) = self.retained_object_bytes.pop_front() {
self.retained_bytes = self.retained_bytes.saturating_sub(size);
}
}
}
async fn write_bounded(
&mut self,
payload: Bytes,
max_objects: usize,
max_bytes: usize,
timeout: Duration,
) -> Result<()> {
ensure!(max_objects > 0, "MoQ object bound must be non-zero");
ensure!(
payload.len() <= max_bytes,
"MoQ object exceeds the retained-byte bound"
);
let payload_len = payload.len();
tokio::time::timeout(timeout, async {
loop {
self.refresh_retention();
if self.retained_object_bytes.len() < max_objects
&& self.retained_bytes.saturating_add(payload_len) <= max_bytes
{
self.writer
.write(payload.clone())
.context("failed to write MoQ Draft 14 object")?;
self.retained_object_bytes.push_back(payload_len);
self.retained_bytes = self.retained_bytes.saturating_add(payload_len);
return Ok(());
}
tokio::time::sleep(Duration::from_millis(1)).await;
}
})
.await
.map_err(|_| {
anyhow!(
"MoQ Draft 14 object backpressure exceeded {} ms",
timeout.as_millis()
)
})?
}
}
#[derive(Clone)]
pub struct NativeMoQSession {
local_node_id: String,
remote_node_id: String,
relay_url: String,
access_token: Option<String>,
carrier_session_id: Option<String>,
state: Arc<AtomicU8>,
started: Arc<AtomicBool>,
endpoint_client: Arc<Mutex<Option<moq_native_ietf::quic::Client>>>,
outbound_tracks: Arc<Mutex<Draft14TrackWriters>>,
inbound_subscription_open: Arc<AtomicBool>,
inbound_carrier_readiness: Arc<AtomicU8>,
outbound_carrier_readiness_sent: Arc<AtomicBool>,
tasks: Arc<Mutex<Vec<tokio::task::JoinHandle<()>>>>,
message_handler: Arc<RwLock<Option<MoqMessageHandler>>>,
terminal_handler: Arc<RwLock<Option<MoqTerminalHandler>>>,
pending_messages: Arc<Mutex<Vec<Bytes>>>,
}
impl NativeMoQSession {
pub async fn new(
local_node_id: &str,
remote_node_id: &str,
config: &crate::client::MoQConfig,
) -> Result<Self> {
Self::new_base(local_node_id, remote_node_id, config).await
}
pub async fn new_packet_carrier(
local_node_id: &str,
remote_node_id: &str,
config: &crate::client::MoQConfig,
carrier_session_id: &str,
) -> Result<Self> {
let mut session = Self::new_base(local_node_id, remote_node_id, config).await?;
ensure!(
!carrier_session_id.trim().is_empty()
&& carrier_session_id
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'-'),
"MoQ carrier session id must be a non-empty URL-safe identifier"
);
session.carrier_session_id = Some(carrier_session_id.to_string());
Ok(session)
}
async fn new_base(
local_node_id: &str,
remote_node_id: &str,
config: &crate::client::MoQConfig,
) -> Result<Self> {
let relay_url = config.relay_url.trim();
ensure!(!relay_url.is_empty(), "moq relay_url is required");
ensure!(
relay_url.starts_with("https://"),
"moq relay_url must use https/WebTransport"
);
ensure!(
!relay_url.contains('#'),
"moq relay_url must not contain a fragment"
);
ensure!(
!relay_url_has_jwt_query(relay_url),
"moq relay_url must not contain a jwt query parameter; use access_token"
);
let access_token = config
.access_token
.as_deref()
.map(str::trim)
.map(str::to_string);
if let Some(token) = access_token.as_deref() {
ensure!(!token.is_empty(), "moq access_token must not be empty");
ensure!(
token.len() <= MOQ_MAX_ACCESS_TOKEN_BYTES,
"moq access_token exceeds the maximum supported size"
);
ensure!(
token
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')),
"moq access_token must be a URL-safe JWT"
);
}
Ok(Self {
local_node_id: local_node_id.to_string(),
remote_node_id: remote_node_id.to_string(),
relay_url: relay_url.to_string(),
access_token,
carrier_session_id: None,
state: Arc::new(AtomicU8::new(0)),
started: Arc::new(AtomicBool::new(false)),
endpoint_client: Arc::new(Mutex::new(None)),
outbound_tracks: Arc::new(Mutex::new(HashMap::new())),
inbound_subscription_open: Arc::new(AtomicBool::new(false)),
inbound_carrier_readiness: Arc::new(AtomicU8::new(0)),
outbound_carrier_readiness_sent: Arc::new(AtomicBool::new(false)),
tasks: Arc::new(Mutex::new(Vec::new())),
message_handler: Arc::new(RwLock::new(None)),
terminal_handler: Arc::new(RwLock::new(None)),
pending_messages: Arc::new(Mutex::new(Vec::new())),
})
}
pub fn state(&self) -> NativeMoQState {
decode_state(self.state.load(Ordering::SeqCst))
}
#[cfg(test)]
pub(crate) fn force_state_for_test(&self, state: NativeMoQState) {
self.state.store(encode_state(state), Ordering::SeqCst);
}
pub async fn start(&self) -> Result<()> {
if self.started.swap(true, Ordering::SeqCst) {
return Ok(());
}
self.state.store(1, Ordering::SeqCst);
if self.relay_url.contains(".invalid") {
self.state.store(3, Ordering::SeqCst);
bail!(
"moq relay is unreachable: {}",
relay_diagnostic_label(&self.relay_url)
);
}
let setup = tokio::time::timeout(MOQ_SETUP_TIMEOUT, self.connect_draft14()).await;
let (client, session, publisher, subscriber) = match setup {
Ok(Ok(connected)) => connected,
Ok(Err(error)) => {
self.state.store(3, Ordering::SeqCst);
return Err(error);
}
Err(_) => {
self.state.store(3, Ordering::SeqCst);
bail!(
"MoQ Draft 14 setup timed out after {} ms",
MOQ_SETUP_TIMEOUT.as_millis()
);
}
};
*self.endpoint_client.lock().await = Some(client);
self.state.store(2, Ordering::SeqCst);
self.spawn_session_driver(session).await;
self.spawn_publisher(publisher).await;
self.spawn_subscription(
subscriber,
self.direct_inbound_namespace(),
self.direct_inbound_track_name(),
)
.await;
Ok(())
}
pub async fn send(&self, data: &[u8]) -> Result<()> {
ensure!(
self.state() == NativeMoQState::Connected,
"native moq session is not started"
);
ensure!(
data.len() <= MOQ_MAX_OBJECT_PAYLOAD_BYTES,
"native moq object payload too large: {}",
data.len()
);
let track_name = self.direct_outbound_track_name();
{
let mut tracks = self.outbound_tracks.lock().await;
let Some(track) = tracks.get_mut(&track_name) else {
bail!("native moq track is not subscribed yet; caller must fall back to iroh");
};
track
.write_bounded(
Bytes::copy_from_slice(data),
MOQ_MAX_BUFFERED_OBJECTS,
MOQ_MAX_BUFFERED_OBJECT_BYTES,
MOQ_OBJECT_BACKPRESSURE_TIMEOUT,
)
.await?;
}
if self.carrier_session_id.is_some() {
tokio::task::yield_now().await;
}
Ok(())
}
pub async fn is_peer_data_subscribed(&self) -> bool {
self.outbound_tracks
.lock()
.await
.contains_key(&self.direct_outbound_track_name())
}
pub async fn is_peer_data_bidirectionally_ready(&self) -> bool {
let directed_tracks_ready = self.inbound_subscription_open.load(Ordering::Acquire)
&& self.is_peer_data_subscribed().await;
if self.carrier_session_id.is_none() {
return directed_tracks_ready;
}
directed_tracks_ready
&& crate::iroh_carrier::carrier_readiness_is_complete(
self.inbound_carrier_readiness.load(Ordering::Acquire),
)
&& self.outbound_carrier_readiness_sent.load(Ordering::Acquire)
}
pub async fn peer_data_readiness_diagnostic(&self) -> String {
format!(
"state={:?} inbound_subscription_open={} outbound_subscribed={} inbound_readiness_mask={:#04b} outbound_readiness_sent={}",
self.state(),
self.inbound_subscription_open.load(Ordering::Acquire),
self.is_peer_data_subscribed().await,
self.inbound_carrier_readiness.load(Ordering::Acquire),
self.outbound_carrier_readiness_sent.load(Ordering::Acquire),
)
}
pub async fn wait_for_peer_data_subscription(&self, timeout: Duration) -> bool {
tokio::time::timeout(timeout, async {
loop {
if self.is_peer_data_subscribed().await {
return true;
}
if self.state() != NativeMoQState::Connected {
return false;
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await
.unwrap_or(false)
}
pub async fn wait_for_peer_data_bidirectional_readiness(&self, timeout: Duration) -> bool {
if self.carrier_session_id.is_none() {
return self.wait_for_peer_data_subscription(timeout).await;
}
tokio::time::timeout(timeout, async {
loop {
if self.is_peer_data_subscribed().await {
let mut sent = true;
for (index, object) in crate::iroh_carrier::carrier_readiness_objects()
.into_iter()
.enumerate()
{
if let Err(error) = self.send(&object).await {
if std::env::var_os("OPENRTC_ADMISSION_VERBOSE").is_some() {
eprintln!(
"[NativeMoQ] Draft 14 packet readiness object send failed index={index} bytes={} error={error:#}",
object.len(),
);
}
sent = false;
break;
}
tokio::task::yield_now().await;
}
if sent {
self.outbound_carrier_readiness_sent
.store(true, Ordering::Release);
}
}
if self.is_peer_data_bidirectionally_ready().await {
return true;
}
if self.state() != NativeMoQState::Connected {
return false;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
})
.await
.unwrap_or(false)
}
pub async fn close_gracefully(&self) {
self.state.store(4, Ordering::SeqCst);
self.inbound_subscription_open
.store(false, Ordering::Release);
self.inbound_carrier_readiness.store(0, Ordering::Release);
self.outbound_carrier_readiness_sent
.store(false, Ordering::Release);
self.outbound_tracks.lock().await.clear();
let tasks = self.tasks.lock().await.drain(..).collect::<Vec<_>>();
for task in &tasks {
task.abort();
}
for task in tasks {
let _ = task.await;
}
self.endpoint_client.lock().await.take();
}
pub fn close(&self) {
self.state.store(4, Ordering::SeqCst);
let session = self.clone();
tokio::spawn(async move {
session.close_gracefully().await;
});
}
pub async fn set_message_handler(&self, handler: MoqMessageHandler) {
*self.message_handler.write().await = Some(handler.clone());
let pending = {
let mut guard = self.pending_messages.lock().await;
std::mem::take(&mut *guard)
};
for message in pending {
handler(message).await;
}
}
pub async fn set_terminal_handler(&self, handler: MoqTerminalHandler) {
*self.terminal_handler.write().await = Some(handler);
}
#[cfg(feature = "test-harness")]
pub async fn force_failure_for_test(&self) -> bool {
if !mark_failed_unless_closed(&self.state) {
return false;
}
self.inbound_subscription_open
.store(false, Ordering::Release);
self.outbound_tracks.lock().await.clear();
let tasks = self.tasks.lock().await.drain(..).collect::<Vec<_>>();
for task in &tasks {
task.abort();
}
for task in tasks {
let _ = task.await;
}
self.endpoint_client.lock().await.take();
self.notify_terminal_failure().await;
true
}
async fn connect_draft14(
&self,
) -> Result<(
moq_native_ietf::quic::Client,
moq_transport::session::Session,
moq_transport::session::Publisher,
moq_transport::session::Subscriber,
)> {
let connection_url = relay_connection_url(&self.relay_url, self.access_token.as_deref());
let url = connection_url
.parse::<url::Url>()
.context("invalid MoQ relay URL")?;
let tls = moq_native_ietf::tls::Args {
disable_verify: relay_url_disables_certificate_verification(&self.relay_url),
..Default::default()
}
.load()
.context("failed to configure MoQ TLS")?;
let quic = moq_native_ietf::quic::Args {
tls: moq_native_ietf::tls::Args {
disable_verify: relay_url_disables_certificate_verification(&self.relay_url),
..Default::default()
},
..Default::default()
}
.load()
.or_else(|_| {
moq_native_ietf::quic::Config::new(
"0.0.0.0:0".parse().expect("valid fallback bind address"),
None,
tls,
)
})
.context("failed to configure MoQ QUIC")?;
let endpoint = moq_native_ietf::quic::Endpoint::new(quic)
.context("failed to create MoQ Draft 14 endpoint")?;
let client = endpoint.client;
let relay_label = relay_diagnostic_label(&self.relay_url);
let (webtransport, _, transport) = client.connect(&url, None).await.map_err(|error| {
anyhow!(
"failed to connect MoQ Draft 14 relay {relay_label}: {}",
redact_access_token(&error.to_string(), self.access_token.as_deref())
)
})?;
let (session, publisher, subscriber) =
moq_transport::session::Session::connect(webtransport, None, transport)
.await
.context("MoQ Draft 14 SETUP exchange failed")?;
Ok((client, session, publisher, subscriber))
}
async fn spawn_session_driver(&self, session: moq_transport::session::Session) {
let state = self.state.clone();
let terminal_handler = self.terminal_handler.clone();
self.push_task(tokio::spawn(async move {
if let Err(error) = session.run().await {
eprintln!("[NativeMoQ] Draft 14 session failed: {error:#}");
}
if mark_failed_unless_closed(&state) {
notify_terminal_failure(&terminal_handler).await;
}
}))
.await;
}
async fn spawn_publisher(&self, mut publisher: moq_transport::session::Publisher) {
let namespace = TrackNamespace::from_utf8_path(&self.direct_outbound_namespace());
let (_tracks_writer, mut requests, tracks_reader) = serve::Tracks::new(namespace).produce();
let allowed_direct = self.direct_outbound_track_name();
let outbound_tracks = self.outbound_tracks.clone();
let state = self.state.clone();
let terminal_handler = self.terminal_handler.clone();
self.push_task(tokio::spawn(async move {
let announce = publisher.announce(tracks_reader);
tokio::pin!(announce);
loop {
tokio::select! {
result = &mut announce => {
match result {
Ok(()) => {}
Err(error) => eprintln!("[NativeMoQ] Draft 14 announce failed: {error:#}"),
}
if mark_failed_unless_closed(&state) {
notify_terminal_failure(&terminal_handler).await;
}
return;
}
requested = requests.next() => {
let Some(track) = requested else {
if mark_failed_unless_closed(&state) {
notify_terminal_failure(&terminal_handler).await;
}
return;
};
let name = track.name.clone();
if name != allowed_direct {
let _ = track.close(serve::ServeError::NotFound);
continue;
}
let writer = track
.subgroups()
.and_then(|mut groups| groups.append(0));
match writer {
Ok(writer) => {
outbound_tracks
.lock()
.await
.insert(name, Draft14TrackWriter::new(writer));
}
Err(error) => {
eprintln!("[NativeMoQ] failed to serve Draft 14 track: {error}");
}
}
}
}
}
}))
.await;
}
async fn spawn_subscription(
&self,
subscriber: moq_transport::session::Subscriber,
namespace: TrackNamespace,
track_name: String,
) {
let state = self.state.clone();
let message_handler = self.message_handler.clone();
let pending_messages = self.pending_messages.clone();
let inbound_subscription_open = self.inbound_subscription_open.clone();
let inbound_carrier_readiness = self.inbound_carrier_readiness.clone();
let packet_carrier = self.carrier_session_id.is_some();
self.push_task(tokio::spawn(async move {
let mut consecutive_failures = 0u8;
while decode_state(state.load(Ordering::SeqCst)) == NativeMoQState::Connected {
let (track_writer, track_reader) =
serve::Track::new(namespace.clone(), track_name.clone()).produce();
let mut subscriber = subscriber.clone();
match tokio::time::timeout(
MOQ_SUBSCRIBE_ATTEMPT_TIMEOUT,
subscriber.subscribe_open(track_writer),
)
.await
{
Ok(Ok(subscription)) => {
consecutive_failures = 0;
if packet_carrier && std::env::var_os("OPENRTC_ADMISSION_VERBOSE").is_some()
{
eprintln!(
"[NativeMoQ] Draft 14 packet subscription accepted namespace={} track={track_name}",
namespace.to_string(),
);
}
inbound_subscription_open.store(true, Ordering::Release);
let result = receive_track(
track_reader,
message_handler.clone(),
pending_messages.clone(),
packet_carrier.then(|| inbound_carrier_readiness.clone()),
)
.await;
inbound_subscription_open.store(false, Ordering::Release);
drop(subscription);
if let Err(error) = result {
eprintln!(
"[NativeMoQ] Draft 14 subscription {track_name} ended: {error:#}"
);
}
}
Ok(Err(error)) => {
consecutive_failures = consecutive_failures.saturating_add(1);
if consecutive_failures == 1
|| consecutive_failures.is_power_of_two()
{
eprintln!(
"[NativeMoQ] Draft 14 subscription {track_name} not ready: {error}; attempt={consecutive_failures}"
);
}
}
Err(_) => {
consecutive_failures = consecutive_failures.saturating_add(1);
if consecutive_failures == 1
|| consecutive_failures.is_power_of_two()
{
eprintln!(
"[NativeMoQ] Draft 14 subscription {track_name} acknowledgement timed out after {}ms; attempt={consecutive_failures}",
MOQ_SUBSCRIBE_ATTEMPT_TIMEOUT.as_millis(),
);
}
}
}
if decode_state(state.load(Ordering::SeqCst)) != NativeMoQState::Connected {
return;
}
tokio::time::sleep(moq_subscribe_retry_delay(consecutive_failures)).await;
}
}))
.await;
}
async fn push_task(&self, task: tokio::task::JoinHandle<()>) {
self.tasks.lock().await.push(task);
}
#[cfg(feature = "test-harness")]
async fn notify_terminal_failure(&self) {
notify_terminal_failure(&self.terminal_handler).await;
}
fn direct_outbound_track_name(&self) -> String {
match self.carrier_session_id.as_deref() {
Some(_) => "iroh-packets".to_string(),
None => format!("data / {} ", self.remote_node_id),
}
}
fn direct_outbound_namespace(&self) -> String {
match self.carrier_session_id.as_deref() {
Some(carrier_session_id) => iroh_carrier_moq_namespace(
&self.local_node_id,
&self.remote_node_id,
carrier_session_id,
&self.local_node_id,
),
None => moq_track_namespace(&self.local_node_id, &self.remote_node_id),
}
}
fn direct_inbound_track_name(&self) -> String {
match self.carrier_session_id.as_deref() {
Some(_) => "iroh-packets".to_string(),
None => format!("data / {} ", self.local_node_id),
}
}
fn direct_inbound_namespace(&self) -> TrackNamespace {
let namespace = match self.carrier_session_id.as_deref() {
Some(carrier_session_id) => iroh_carrier_moq_namespace(
&self.local_node_id,
&self.remote_node_id,
carrier_session_id,
&self.remote_node_id,
),
None => moq_track_namespace(&self.remote_node_id, &self.local_node_id),
};
TrackNamespace::from_utf8_path(&namespace)
}
}
fn moq_subscribe_retry_delay(consecutive_failures: u8) -> Duration {
if consecutive_failures == 0 {
return MOQ_SUBSCRIBE_RETRY_DELAY;
}
let multiplier = 1u32 << consecutive_failures.saturating_sub(1).min(5);
MOQ_SUBSCRIBE_RETRY_DELAY
.saturating_mul(multiplier)
.min(MOQ_SUBSCRIBE_MAX_RETRY_DELAY)
}
fn moq_track_namespace(publisher_node_id: &str, subscriber_node_id: &str) -> String {
format!("{}/{}", publisher_node_id.trim(), subscriber_node_id.trim())
}
fn iroh_carrier_moq_namespace(
local_node_id: &str,
remote_node_id: &str,
carrier_session_id: &str,
publisher_node_id: &str,
) -> String {
let (first, second) = if local_node_id <= remote_node_id {
(local_node_id, remote_node_id)
} else {
(remote_node_id, local_node_id)
};
format!(
"openrtc/iroh-carrier/moq/{first}/{second}/{carrier_session_id}/from/{}",
publisher_node_id.trim(),
)
}
async fn receive_track(
track: serve::TrackReader,
message_handler: Arc<RwLock<Option<MoqMessageHandler>>>,
pending_messages: Arc<Mutex<Vec<Bytes>>>,
inbound_carrier_readiness: Option<Arc<AtomicU8>>,
) -> Result<()> {
match track.mode().await? {
TrackReaderMode::Subgroups(mut groups) => {
while let Some(mut group) = groups.next().await? {
while let Some(payload) = group.read_next().await? {
validate_and_dispatch_moq_payload(
&message_handler,
&pending_messages,
payload,
inbound_carrier_readiness.as_ref(),
)
.await?;
}
}
}
TrackReaderMode::Datagrams(mut datagrams) => {
ensure!(
inbound_carrier_readiness.is_none(),
"MoQ packet carrier requires a reliable Draft 14 subgroup stream"
);
let mut first_payload = true;
while let Some(datagram) = datagrams.read().await? {
if first_payload
&& inbound_carrier_readiness.is_some()
&& std::env::var_os("OPENRTC_ADMISSION_VERBOSE").is_some()
{
first_payload = false;
eprintln!(
"[NativeMoQ] Draft 14 application subscription received first datagram bytes={}",
datagram.payload.len(),
);
}
validate_and_dispatch_moq_payload(
&message_handler,
&pending_messages,
datagram.payload,
inbound_carrier_readiness.as_ref(),
)
.await?;
}
}
_ => bail!("MoQ Draft 14 peer selected an unsupported track mode"),
}
Ok(())
}
async fn validate_and_dispatch_moq_payload(
message_handler: &Arc<RwLock<Option<MoqMessageHandler>>>,
pending_messages: &Arc<Mutex<Vec<Bytes>>>,
payload: Bytes,
inbound_carrier_readiness: Option<&Arc<AtomicU8>>,
) -> Result<()> {
ensure!(
payload.len() <= MOQ_MAX_OBJECT_PAYLOAD_BYTES,
"native moq object payload too large: {}",
payload.len()
);
if let Some(readiness) = inbound_carrier_readiness {
let observed = crate::iroh_carrier::observe_carrier_readiness_object(0, &payload);
if observed != 0 {
readiness.fetch_or(observed, Ordering::AcqRel);
return Ok(());
}
}
dispatch_or_buffer_message(message_handler, pending_messages, payload).await
}
async fn dispatch_or_buffer_message(
message_handler: &Arc<RwLock<Option<MoqMessageHandler>>>,
pending_messages: &Arc<Mutex<Vec<Bytes>>>,
payload: Bytes,
) -> Result<()> {
let handler_guard = message_handler.read().await;
if let Some(handler) = handler_guard.clone() {
drop(handler_guard);
handler(payload).await;
} else {
let mut pending = pending_messages.lock().await;
let pending_bytes = pending.iter().map(Bytes::len).sum::<usize>();
ensure!(
pending.len() < MOQ_MAX_BUFFERED_OBJECTS
&& pending_bytes.saturating_add(payload.len()) <= MOQ_MAX_BUFFERED_OBJECT_BYTES,
"native moq pending-handler queue exceeded {} objects or {} bytes",
MOQ_MAX_BUFFERED_OBJECTS,
MOQ_MAX_BUFFERED_OBJECT_BYTES,
);
pending.push(payload);
}
Ok(())
}
fn mark_failed_unless_closed(state: &AtomicU8) -> bool {
loop {
let current = state.load(Ordering::SeqCst);
match decode_state(current) {
NativeMoQState::Closed | NativeMoQState::Failed => return false,
NativeMoQState::Idle | NativeMoQState::Connecting | NativeMoQState::Connected => {
if state
.compare_exchange(current, 3, Ordering::SeqCst, Ordering::SeqCst)
.is_ok()
{
return true;
}
}
}
}
}
async fn notify_terminal_failure(terminal_handler: &Arc<RwLock<Option<MoqTerminalHandler>>>) {
if let Some(handler) = terminal_handler.read().await.clone() {
handler().await;
}
}
#[cfg(test)]
fn encode_state(state: NativeMoQState) -> u8 {
match state {
NativeMoQState::Idle => 0,
NativeMoQState::Connecting => 1,
NativeMoQState::Connected => 2,
NativeMoQState::Failed => 3,
NativeMoQState::Closed => 4,
}
}
fn decode_state(state: u8) -> NativeMoQState {
match state {
1 => NativeMoQState::Connecting,
2 => NativeMoQState::Connected,
3 => NativeMoQState::Failed,
4 => NativeMoQState::Closed,
_ => NativeMoQState::Idle,
}
}
fn relay_url_uses_loopback_host(relay_url: &str) -> bool {
relay_url.starts_with("https://localhost")
|| relay_url.starts_with("https://127.")
|| relay_url.starts_with("https://[::1]")
}
fn relay_url_disables_certificate_verification(relay_url: &str) -> bool {
if relay_url_uses_loopback_host(relay_url) {
return true;
}
#[cfg(feature = "moq-carrier-test")]
{
let private_test_host = url::Url::parse(relay_url)
.ok()
.and_then(|url| url.host_str()?.parse::<std::net::IpAddr>().ok())
.is_some_and(|address| match address {
std::net::IpAddr::V4(address) => address.is_private() || address.is_link_local(),
std::net::IpAddr::V6(address) => {
address.is_unique_local() || address.is_unicast_link_local()
}
});
if private_test_host {
return true;
}
}
false
}
fn relay_url_has_jwt_query(relay_url: &str) -> bool {
relay_url
.split_once('?')
.map(|(_, query)| {
query
.split('&')
.map(|pair| pair.split_once('=').map_or(pair, |(key, _)| key))
.any(|key| key.eq_ignore_ascii_case("jwt"))
})
.unwrap_or(false)
}
fn relay_connection_url(relay_url: &str, access_token: Option<&str>) -> String {
let Some(token) = access_token else {
return relay_url.to_string();
};
let separator = if relay_url.contains('?') { '&' } else { '?' };
format!("{relay_url}{separator}jwt={token}")
}
fn relay_diagnostic_label(relay_url: &str) -> &str {
relay_url
.split_once('?')
.map_or(relay_url, |(base, _)| base)
}
fn redact_access_token(value: &str, access_token: Option<&str>) -> String {
let Some(token) = access_token else {
return value.to_string();
};
value.replace(token, "[redacted]")
}
#[cfg(test)]
mod tests {
use super::*;
use crate::client::MoQConfig;
fn make_config(relay_url: &str) -> MoQConfig {
MoQConfig {
relay_url: relay_url.to_string(),
access_token: None,
}
}
#[tokio::test]
async fn invalid_relay_url_reaches_failed() {
let session =
NativeMoQSession::new("local", "remote", &make_config("https://relay.invalid/moq"))
.await
.unwrap();
assert!(session.start().await.is_err());
assert_eq!(session.state(), NativeMoQState::Failed);
}
#[tokio::test]
async fn send_requires_connected_and_subscribed_route() {
let session = NativeMoQSession::new(
"local",
"remote",
&make_config("https://relay.example.com/moq"),
)
.await
.unwrap();
let error = session.send(b"hello").await.unwrap_err();
assert!(error.to_string().contains("not started"));
session.force_state_for_test(NativeMoQState::Connected);
let error = session.send(b"hello").await.unwrap_err();
assert!(error.to_string().contains("not subscribed"));
}
#[tokio::test]
async fn handler_handoff_queue_is_bounded() {
let handler = Arc::new(RwLock::new(None));
let pending = Arc::new(Mutex::new(Vec::new()));
for _ in 0..MOQ_MAX_BUFFERED_OBJECTS {
dispatch_or_buffer_message(&handler, &pending, Bytes::from_static(b"x"))
.await
.expect("queue within bound");
}
let error = dispatch_or_buffer_message(&handler, &pending, Bytes::from_static(b"x"))
.await
.expect_err("queue must fail closed at its object bound");
assert!(error.to_string().contains("pending-handler queue exceeded"));
assert_eq!(pending.lock().await.len(), MOQ_MAX_BUFFERED_OBJECTS);
}
#[tokio::test]
async fn draft14_writer_applies_backpressure_through_the_published_api() {
let (track_writer, track_reader) = serve::Track::new(
TrackNamespace::from_utf8_path("openrtc/test"),
"objects".to_owned(),
)
.produce();
let mut subgroup_writer = track_writer.subgroups().expect("subgroup track");
let writer = subgroup_writer.append(0).expect("first subgroup");
let mut writer = Draft14TrackWriter::new(writer);
let TrackReaderMode::Subgroups(mut subgroup_reader) =
track_reader.mode().await.expect("reader mode")
else {
panic!("expected subgroup reader");
};
let mut reader = subgroup_reader
.next()
.await
.expect("subgroup read")
.expect("first subgroup");
writer
.write_bounded(Bytes::from_static(b"one"), 1, 16, Duration::from_secs(1))
.await
.expect("first object");
let error = writer
.write_bounded(Bytes::from_static(b"two"), 1, 16, Duration::from_millis(10))
.await
.expect_err("the retained first object must apply backpressure");
assert!(error.to_string().contains("backpressure exceeded"));
assert_eq!(
reader.read_next().await.expect("object read"),
Some(Bytes::from_static(b"one"))
);
writer
.write_bounded(Bytes::from_static(b"two"), 1, 16, Duration::from_secs(1))
.await
.expect("capacity released after consumption");
assert_eq!(
reader.read_next().await.expect("object read"),
Some(Bytes::from_static(b"two"))
);
}
#[tokio::test]
async fn close_reaches_closed_state() {
let session = NativeMoQSession::new(
"local",
"remote",
&make_config("https://relay.example.com/moq"),
)
.await
.unwrap();
session.close_gracefully().await;
assert_eq!(session.state(), NativeMoQState::Closed);
}
#[test]
fn subscription_retry_backoff_is_bounded_and_resets_after_success() {
assert_eq!(moq_subscribe_retry_delay(0), Duration::from_millis(250));
assert_eq!(moq_subscribe_retry_delay(1), Duration::from_millis(250));
assert_eq!(moq_subscribe_retry_delay(2), Duration::from_millis(500));
assert_eq!(moq_subscribe_retry_delay(6), Duration::from_secs(8));
assert_eq!(moq_subscribe_retry_delay(u8::MAX), Duration::from_secs(8));
}
#[test]
fn direct_namespaces_are_unique_per_remote_peer_and_reciprocal() {
assert_ne!(
moq_track_namespace("local", "peer-a"),
moq_track_namespace("local", "peer-b"),
);
assert_eq!(moq_track_namespace("local", "remote"), "local/remote",);
assert_eq!(moq_track_namespace("remote", "local"), "remote/local",);
assert_eq!(
iroh_carrier_moq_namespace("remote", "local", "session-1", "remote"),
"openrtc/iroh-carrier/moq/local/remote/session-1/from/remote",
);
assert_eq!(
iroh_carrier_moq_namespace("local", "remote", "session-1", "local"),
"openrtc/iroh-carrier/moq/local/remote/session-1/from/local",
);
}
#[cfg(feature = "test-harness")]
#[tokio::test]
async fn draft14_local_relay_carries_bilateral_native_payloads() {
let probe = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
let port = probe.local_addr().unwrap().port();
drop(probe);
let relay = tokio::spawn(async move {
let _ = test_relay::run(port).await;
});
tokio::time::sleep(Duration::from_millis(100)).await;
let config = make_config(&format!("https://127.0.0.1:{port}/moq"));
let left = NativeMoQSession::new("left", "right", &config)
.await
.unwrap();
let right = NativeMoQSession::new("right", "left", &config)
.await
.unwrap();
left.start().await.unwrap();
right.start().await.unwrap();
let (left_ready, right_ready) = tokio::join!(
left.wait_for_peer_data_bidirectional_readiness(Duration::from_secs(5)),
right.wait_for_peer_data_bidirectional_readiness(Duration::from_secs(5)),
);
assert!(left_ready && right_ready);
assert!(left.is_peer_data_bidirectionally_ready().await);
assert!(right.is_peer_data_bidirectionally_ready().await);
let (left_tx, left_rx) = tokio::sync::oneshot::channel::<Bytes>();
let left_tx = Arc::new(Mutex::new(Some(left_tx)));
left.set_message_handler(Arc::new(move |payload| {
let left_tx = left_tx.clone();
Box::pin(async move {
if let Some(tx) = left_tx.lock().await.take() {
let _ = tx.send(payload);
}
})
}))
.await;
let (right_tx, right_rx) = tokio::sync::oneshot::channel::<Bytes>();
let right_tx = Arc::new(Mutex::new(Some(right_tx)));
right
.set_message_handler(Arc::new(move |payload| {
let right_tx = right_tx.clone();
Box::pin(async move {
if let Some(tx) = right_tx.lock().await.take() {
let _ = tx.send(payload);
}
})
}))
.await;
left.send(b"left-to-right").await.unwrap();
right.send(b"right-to-left").await.unwrap();
assert_eq!(
tokio::time::timeout(Duration::from_secs(3), right_rx)
.await
.unwrap()
.unwrap(),
Bytes::from_static(b"left-to-right")
);
assert_eq!(
tokio::time::timeout(Duration::from_secs(3), left_rx)
.await
.unwrap()
.unwrap(),
Bytes::from_static(b"right-to-left")
);
left.close_gracefully().await;
right.close_gracefully().await;
relay.abort();
}
#[cfg(feature = "test-harness")]
#[tokio::test]
async fn draft14_local_relay_carries_bilateral_carrier_object_streams() {
let probe = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
let port = probe.local_addr().unwrap().port();
drop(probe);
let relay = tokio::spawn(async move {
let _ = test_relay::run(port).await;
});
tokio::time::sleep(Duration::from_millis(100)).await;
let config = make_config(&format!("https://127.0.0.1:{port}/moq"));
let left = NativeMoQSession::new_packet_carrier(
"left",
"right",
&config,
"00112233445566778899aabbccddeeff",
)
.await
.unwrap();
let right = NativeMoQSession::new_packet_carrier(
"right",
"left",
&config,
"00112233445566778899aabbccddeeff",
)
.await
.unwrap();
left.start().await.unwrap();
right.start().await.unwrap();
let (left_ready, right_ready) = tokio::join!(
left.wait_for_peer_data_bidirectional_readiness(Duration::from_secs(5)),
right.wait_for_peer_data_bidirectional_readiness(Duration::from_secs(5)),
);
assert!(left_ready && right_ready);
assert!(left.is_peer_data_bidirectionally_ready().await);
assert!(right.is_peer_data_bidirectionally_ready().await);
let (left_tx, mut left_rx) = tokio::sync::mpsc::unbounded_channel::<Bytes>();
left.set_message_handler(Arc::new(move |payload| {
let left_tx = left_tx.clone();
Box::pin(async move {
let _ = left_tx.send(payload);
})
}))
.await;
let (right_tx, mut right_rx) = tokio::sync::mpsc::unbounded_channel::<Bytes>();
right
.set_message_handler(Arc::new(move |payload| {
let right_tx = right_tx.clone();
Box::pin(async move {
let _ = right_tx.send(payload);
})
}))
.await;
left.send(b"left-carrier-object").await.unwrap();
right.send(b"right-carrier-object").await.unwrap();
assert_eq!(
tokio::time::timeout(Duration::from_secs(3), right_rx.recv())
.await
.unwrap()
.unwrap(),
Bytes::from_static(b"left-carrier-object")
);
assert_eq!(
tokio::time::timeout(Duration::from_secs(3), left_rx.recv())
.await
.unwrap()
.unwrap(),
Bytes::from_static(b"right-carrier-object")
);
const BURST_OBJECTS: u64 = 512;
tokio::time::timeout(Duration::from_secs(3), async {
for sequence in 0..BURST_OBJECTS {
let mut payload = vec![0x5a; 900];
payload[..8].copy_from_slice(&sequence.to_be_bytes());
left.send(&payload).await.unwrap();
}
})
.await
.expect("carrier burst was throttled below the required media rate");
tokio::time::timeout(Duration::from_secs(5), async {
for expected in 0..BURST_OBJECTS {
let payload = right_rx.recv().await.expect("carrier burst ended early");
assert_eq!(payload.len(), 900);
assert_eq!(&payload[..8], &expected.to_be_bytes());
assert!(payload[8..].iter().all(|byte| *byte == 0x5a));
}
})
.await
.expect("carrier burst did not drain through the Draft 14 relay");
left.close_gracefully().await;
right.close_gracefully().await;
relay.abort();
}
#[cfg(feature = "test-harness")]
#[tokio::test]
async fn draft14_packet_carrier_waits_for_late_bidirectional_publication() {
let probe = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
let port = probe.local_addr().unwrap().port();
drop(probe);
let relay = tokio::spawn(async move {
let _ = test_relay::run(port).await;
});
tokio::time::sleep(Duration::from_millis(100)).await;
let config = make_config(&format!("https://127.0.0.1:{port}/moq"));
let left = NativeMoQSession::new_packet_carrier(
"left",
"right",
&config,
"00112233445566778899aabbccddeeff",
)
.await
.unwrap();
left.start().await.unwrap();
tokio::time::sleep(Duration::from_millis(350)).await;
assert!(!left.is_peer_data_bidirectionally_ready().await);
let right = NativeMoQSession::new_packet_carrier(
"right",
"left",
&config,
"00112233445566778899aabbccddeeff",
)
.await
.unwrap();
right.start().await.unwrap();
let (left_ready, right_ready) = tokio::join!(
left.wait_for_peer_data_bidirectional_readiness(Duration::from_secs(5)),
right.wait_for_peer_data_bidirectional_readiness(Duration::from_secs(5)),
);
assert!(left_ready && right_ready);
assert!(left.is_peer_data_bidirectionally_ready().await);
assert!(right.is_peer_data_bidirectionally_ready().await);
left.close_gracefully().await;
right.close_gracefully().await;
relay.abort();
}
#[cfg(feature = "test-harness")]
#[tokio::test]
async fn draft14_replacement_releases_the_previous_node_namespace() {
let probe = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
let port = probe.local_addr().unwrap().port();
drop(probe);
let relay = tokio::spawn(async move {
let _ = test_relay::run(port).await;
});
tokio::time::sleep(Duration::from_millis(100)).await;
let config = make_config(&format!("https://127.0.0.1:{port}/moq"));
let first_left = NativeMoQSession::new("left", "right", &config)
.await
.unwrap();
let right = NativeMoQSession::new("right", "left", &config)
.await
.unwrap();
first_left.start().await.unwrap();
right.start().await.unwrap();
assert!(
first_left
.wait_for_peer_data_subscription(Duration::from_secs(5))
.await
);
assert!(
right
.wait_for_peer_data_subscription(Duration::from_secs(5))
.await
);
first_left.close_gracefully().await;
let replacement_left = NativeMoQSession::new("left", "right", &config)
.await
.unwrap();
replacement_left.start().await.unwrap();
assert!(
replacement_left
.wait_for_peer_data_subscription(Duration::from_secs(5))
.await,
"replacement publisher never became data-ready"
);
assert!(
right
.wait_for_peer_data_subscription(Duration::from_secs(5))
.await,
"existing peer did not resubscribe to the replacement publisher"
);
let (received_tx, received_rx) = tokio::sync::oneshot::channel::<Bytes>();
let received_tx = Arc::new(Mutex::new(Some(received_tx)));
right
.set_message_handler(Arc::new(move |payload| {
let received_tx = received_tx.clone();
Box::pin(async move {
if let Some(tx) = received_tx.lock().await.take() {
let _ = tx.send(payload);
}
})
}))
.await;
replacement_left.send(b"replacement-payload").await.unwrap();
assert_eq!(
tokio::time::timeout(Duration::from_secs(3), received_rx)
.await
.unwrap()
.unwrap(),
Bytes::from_static(b"replacement-payload")
);
replacement_left.close_gracefully().await;
right.close_gracefully().await;
relay.abort();
}
#[test]
fn access_token_is_connection_only_and_redacted() {
let token = "header.payload.signature";
let relay_url = "https://relay.example.com/moq?existing=value";
assert_eq!(
relay_connection_url(relay_url, Some(token)),
format!("{relay_url}&jwt={token}")
);
assert_eq!(
relay_diagnostic_label(relay_url),
"https://relay.example.com/moq"
);
assert_eq!(
redact_access_token(
&format!("failed to connect https://relay.example.com/moq?jwt={token}"),
Some(token),
),
"failed to connect https://relay.example.com/moq?jwt=[redacted]"
);
}
#[tokio::test]
async fn inline_jwt_and_non_url_safe_tokens_are_rejected() {
let inline = make_config("https://relay.example.com/moq?jwt=secret");
let error = NativeMoQSession::new("local", "remote", &inline)
.await
.err()
.expect("inline JWT must be rejected");
assert!(error.to_string().contains("use access_token"));
let invalid = MoQConfig {
relay_url: "https://relay.example.com/moq".to_string(),
access_token: Some("not a jwt&leak".to_string()),
};
let error = NativeMoQSession::new("local", "remote", &invalid)
.await
.err()
.expect("unsafe token must be rejected");
assert!(error.to_string().contains("URL-safe JWT"));
}
}