use super::route_adapter::{
accepted_native_route_event_generation, is_independent_route_adapter,
native_optional_route_generation_matches,
};
use super::*;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
use crate::native_moq_policy::{
is_data_ready as native_moq_is_data_ready,
is_repairable_state as native_moq_is_repairable_state, setup_status_report,
should_close_before_restart as native_moq_should_close_before_restart,
should_preserve_active_moq_on_iroh_send,
should_reuse_existing_setup as native_moq_should_reuse_existing_setup,
successful_send_status_report,
};
#[cfg(not(target_arch = "wasm32"))]
use crate::native_send_policy::{peer_send_order, NativePeerSendOrder};
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
use crate::native_webrtc_policy::{
admission_trigger_defers_to_remote_initiator, inbound_signal_can_bootstrap_responder,
native_signal_send_failure_is_terminal_for_attempt, native_webrtc_session_is_active,
should_ignore_stale_inbound_signal_recovery,
should_immediately_retire_active_webrtc_for_reason, NativeWebRTCAttemptReservation,
NativeWebRTCBrowserSignalTrigger, NativeWebRTCNativeTrigger, NativeWebRTCRecoveryOptions,
NativeWebRTCRecoveryTrigger,
};
use crate::route_policy::KnownRoute;
#[cfg(not(target_arch = "wasm32"))]
type NativeMainSendFuture<'a> =
std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), String>> + Send + 'a>>;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
const NATIVE_WEBRTC_ROUTE_PROOF_TYPE: &str = "#openrtc-webrtc-route-proof";
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
const NATIVE_MOQ_ROUTE_PROOF_TYPE: &str = "#openrtc-moq-route-proof";
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
#[derive(Debug, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct NativeMoQRouteProofFrame {
#[serde(rename = "type")]
frame_type: String,
version: u8,
role: String,
probe_id: String,
sender_id: String,
recipient_id: String,
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
fn parse_native_moq_route_proof(payload: &[u8]) -> Option<NativeMoQRouteProofFrame> {
const MAX_ROUTE_PROOF_BYTES: usize = 512;
if payload.is_empty() || payload.len() > MAX_ROUTE_PROOF_BYTES {
return None;
}
let frame: NativeMoQRouteProofFrame = serde_json::from_slice(payload).ok()?;
let valid_probe_id = (8..=64).contains(&frame.probe_id.len())
&& frame
.probe_id
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'));
let valid_peer_ids = !frame.sender_id.is_empty()
&& frame.sender_id.len() <= 128
&& !frame.recipient_id.is_empty()
&& frame.recipient_id.len() <= 128;
if frame.frame_type != NATIVE_MOQ_ROUTE_PROOF_TYPE
|| frame.version != 1
|| !matches!(frame.role.as_str(), "probe" | "ack")
|| !valid_probe_id
|| !valid_peer_ids
{
return None;
}
Some(frame)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[derive(Debug, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "camelCase")]
struct NativeWebRtcRouteProofFrame {
#[serde(rename = "type")]
frame_type: String,
version: u8,
role: String,
probe_id: String,
negotiation_id: String,
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn parse_native_webrtc_route_proof(payload: &[u8]) -> Option<NativeWebRtcRouteProofFrame> {
const MAX_ROUTE_PROOF_BYTES: usize = 2_048;
const MAX_ROUTE_PROOF_ID_BYTES: usize = 256;
if payload.is_empty() || payload.len() > MAX_ROUTE_PROOF_BYTES {
return None;
}
let frame: NativeWebRtcRouteProofFrame = serde_json::from_slice(payload).ok()?;
if frame.frame_type != NATIVE_WEBRTC_ROUTE_PROOF_TYPE
|| frame.version != 1
|| !matches!(frame.role.as_str(), "probe" | "ack")
|| frame.probe_id.is_empty()
|| frame.probe_id.len() > MAX_ROUTE_PROOF_ID_BYTES
|| frame.negotiation_id.is_empty()
|| frame.negotiation_id.len() > MAX_ROUTE_PROOF_ID_BYTES
{
return None;
}
Some(frame)
}
#[cfg(not(target_arch = "wasm32"))]
fn native_control_stream_rank(stream_id: iroh::endpoint::StreamId) -> u64 {
let initiator_rank = match stream_id.initiator() {
iroh::endpoint::Side::Client => 0,
iroh::endpoint::Side::Server => 1,
};
stream_id.index().saturating_mul(2) + initiator_rank
}
#[cfg(not(target_arch = "wasm32"))]
fn should_replace_native_control_stream(
existing: NativeControlStreamOwner,
incoming: NativeControlStreamOwner,
) -> bool {
existing.transport_stable_id != incoming.transport_stable_id
|| incoming.stream_rank < existing.stream_rank
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn retire_losing_native_control_candidate(
send: iroh::endpoint::SendStream,
recv: iroh::endpoint::RecvStream,
) -> std::io::Result<()> {
let delivery = crate::application_crypto_streams::PeerSendStream::plain(send)
.finish_and_wait_for_peer(std::time::Duration::from_secs(2))
.await;
drop(recv);
delivery
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn retire_replaced_native_control_send(
send: Arc<tokio::sync::Mutex<iroh::endpoint::SendStream>>,
) -> std::io::Result<()> {
let mut send = send.lock().await;
send.finish()
.map_err(|error| std::io::Error::other(format!("finish replaced stream: {error}")))?;
match tokio::time::timeout(std::time::Duration::from_secs(2), send.stopped()).await {
Ok(Ok(_)) => Ok(()),
Ok(Err(error)) => Err(std::io::Error::other(format!(
"wait for peer acknowledgement: {error}"
))),
Err(_) => Err(std::io::Error::new(
std::io::ErrorKind::TimedOut,
"timed out waiting for peer acknowledgement",
)),
}
}
#[cfg(not(target_arch = "wasm32"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum NativeBleSwitchAdmission {
Ready,
Wait,
Reject,
}
#[cfg(not(target_arch = "wasm32"))]
const NATIVE_BLE_MAX_UPGRADE_ATTEMPTS: u8 = 3;
#[cfg(not(target_arch = "wasm32"))]
const NATIVE_BLE_RETRY_BASE_DELAY_MS: u64 = 1_000;
#[cfg(not(target_arch = "wasm32"))]
const NATIVE_BLE_SWITCH_RESPONSE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
#[cfg(not(target_arch = "wasm32"))]
fn native_ble_retry_delay(attempt: u8, jitter_seed: u64) -> Option<std::time::Duration> {
if attempt >= NATIVE_BLE_MAX_UPGRADE_ATTEMPTS {
return None;
}
let exponent = u32::from(attempt.saturating_sub(1));
let base_ms = NATIVE_BLE_RETRY_BASE_DELAY_MS.saturating_mul(2_u64.saturating_pow(exponent));
let jitter_percent = i64::try_from(jitter_seed % 41).unwrap_or(0) - 20;
let jitter_ms = i128::from(base_ms) * i128::from(jitter_percent) / 100;
let delay_ms = (i128::from(base_ms) + jitter_ms).max(1) as u64;
Some(std::time::Duration::from_millis(delay_ms))
}
#[cfg(not(target_arch = "wasm32"))]
fn native_ble_attempt_state_after_recovery_wake(
current: crate::client::NativeBleUpgradeAttemptState,
gate_active: bool,
) -> Option<crate::client::NativeBleUpgradeAttemptState> {
gate_active.then_some(current)
}
#[cfg(not(target_arch = "wasm32"))]
fn reserve_native_ble_attempt_state(
current: Option<crate::client::NativeBleUpgradeAttemptState>,
generation: crate::client::NativePeerDataGeneration,
) -> (crate::client::NativeBleUpgradeAttemptState, Option<u8>) {
let mut state = current
.filter(|state| state.generation == generation)
.unwrap_or(crate::client::NativeBleUpgradeAttemptState {
generation,
attempts: 0,
});
if state.attempts >= NATIVE_BLE_MAX_UPGRADE_ATTEMPTS {
return (state, None);
}
state.attempts = state.attempts.saturating_add(1);
let attempt = state.attempts;
(state, Some(attempt))
}
#[cfg(not(target_arch = "wasm32"))]
fn classify_native_ble_switch_admission(
block: Option<&(bool, String)>,
) -> NativeBleSwitchAdmission {
match block {
None => NativeBleSwitchAdmission::Ready,
Some((true, _)) => NativeBleSwitchAdmission::Reject,
Some((false, _)) => NativeBleSwitchAdmission::Wait,
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct NativeMoqUpgradeOptions {
force_restart_when_not_ready: bool,
allow_on_optimal_iroh_path: bool,
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
impl NativeMoqUpgradeOptions {
const OPPORTUNISTIC: Self = Self {
force_restart_when_not_ready: false,
allow_on_optimal_iroh_path: false,
};
const OPPORTUNISTIC_REPAIR: Self = Self {
force_restart_when_not_ready: true,
allow_on_optimal_iroh_path: false,
};
const EXPLICIT: Self = Self {
force_restart_when_not_ready: false,
allow_on_optimal_iroh_path: true,
};
const EXPLICIT_REPAIR: Self = Self {
force_restart_when_not_ready: true,
allow_on_optimal_iroh_path: true,
};
}
#[cfg(not(target_arch = "wasm32"))]
fn iroh_path_allows_opportunistic_optional_route(path: IrohPathKind) -> bool {
crate::route_policy::normalize_route(path.transport_label())
.map(|route| {
let descriptor = route.descriptor();
descriptor.base_protocol == crate::transport_label::IROH
&& descriptor.locality == crate::route_policy::RouteLocality::Relay
})
.unwrap_or(false)
}
impl Client {
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn refresh_native_optional_route_generation(&self, connection_id: &str) {
let Some(record) = self
.connection_manager
.get_by_connection_id(connection_id)
.await
else {
self.native_optional_route_generations
.write()
.await
.remove(connection_id);
#[cfg(feature = "transport-moq")]
self.native_moq_route_proofs
.write()
.await
.remove(connection_id);
#[cfg(feature = "transport-webrtc")]
self.native_webrtc_route_proofs
.write()
.await
.remove(connection_id);
return;
};
let current_webrtc_instance = {
#[cfg(feature = "transport-webrtc")]
{
self.native_webrtc_sessions
.read()
.await
.get(connection_id)
.map(|session| Arc::as_ptr(session) as usize)
}
#[cfg(not(feature = "transport-webrtc"))]
{
None
}
};
let current_moq_instance = {
#[cfg(feature = "transport-moq")]
{
self.native_moq_sessions
.read()
.await
.get(connection_id)
.map(|session| Arc::as_ptr(session) as usize)
}
#[cfg(not(feature = "transport-moq"))]
{
None
}
};
let mut installed = self.native_optional_route_generations.write().await;
let Some(generations) = installed.get_mut(connection_id) else {
return;
};
for (route, current_instance) in [
(KnownRoute::WebRtc, current_webrtc_instance),
(KnownRoute::Moq, current_moq_instance),
] {
let Some(mut generation) = generations.get(route) else {
continue;
};
let still_installed = current_instance == Some(generation.route_instance_id)
&& record.transport_stable_id == Some(generation.transport_stable_id)
&& record.transport_generation == generation.transport_generation;
if still_installed {
generation.route_generation = record.route_generation;
generations.set(route, Some(generation));
} else {
generations.set(route, None);
}
}
if generations.is_empty() {
installed.remove(connection_id);
}
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn install_native_optional_route_generation(
&self,
connection_id: &str,
route: KnownRoute,
route_instance_id: usize,
) -> Option<crate::client::NativeOptionalRouteGeneration> {
if !is_independent_route_adapter(route) {
return None;
}
let record = self
.connection_manager
.get_by_connection_id(connection_id)
.await?;
if !matches!(
record.state,
crate::connection_manager::ConnectionState::Connecting
| crate::connection_manager::ConnectionState::Connected
) {
return None;
}
let transport_stable_id = record.transport_stable_id?;
let generation = crate::client::NativeOptionalRouteGeneration {
transport_stable_id,
transport_generation: record.transport_generation,
route_generation: record.route_generation,
route_instance_id,
};
let mut installed = self.native_optional_route_generations.write().await;
let routes = installed.entry(connection_id.to_string()).or_default();
if !routes.set(route, Some(generation)) {
return None;
}
Some(generation)
}
#[cfg(not(target_arch = "wasm32"))]
async fn clear_native_optional_route_generation_if_current(
&self,
connection_id: &str,
route: KnownRoute,
route_instance_id: usize,
) {
let cleared = {
let mut installed = self.native_optional_route_generations.write().await;
let Some(generations) = installed.get_mut(connection_id) else {
return;
};
let cleared = generations
.get(route)
.is_some_and(|generation| generation.route_instance_id == route_instance_id);
if cleared {
generations.set(route, None);
}
if generations.is_empty() {
installed.remove(connection_id);
}
cleared
};
#[cfg(all(not(feature = "transport-moq"), not(feature = "transport-webrtc")))]
let _ = cleared;
#[cfg(feature = "transport-moq")]
if cleared && route == KnownRoute::Moq {
let mut proofs = self.native_moq_route_proofs.write().await;
if proofs
.get(connection_id)
.is_some_and(|proof| proof.route_instance_id == route_instance_id)
{
proofs.remove(connection_id);
}
}
#[cfg(feature = "transport-webrtc")]
if cleared && route == KnownRoute::WebRtc {
let mut proofs = self.native_webrtc_route_proofs.write().await;
if proofs
.get(connection_id)
.is_some_and(|proof| proof.route_instance_id == route_instance_id)
{
proofs.remove(connection_id);
}
}
}
#[cfg(not(target_arch = "wasm32"))]
async fn native_optional_route_generation_for_instance(
&self,
connection_id: &str,
route: KnownRoute,
route_instance_id: usize,
) -> Option<crate::client::NativeOptionalRouteGeneration> {
self.native_optional_route_generations
.read()
.await
.get(connection_id)
.and_then(|generations| generations.get(route))
.filter(|generation| generation.route_instance_id == route_instance_id)
}
#[cfg(not(target_arch = "wasm32"))]
async fn native_optional_route_generation_is_current(
&self,
connection_id: &str,
route: KnownRoute,
route_instance_id: usize,
) -> bool {
self.current_native_optional_route_event_generation(connection_id, route, route_instance_id)
.await
.is_some()
}
#[cfg(not(target_arch = "wasm32"))]
async fn current_native_optional_route_event_generation(
&self,
connection_id: &str,
route: KnownRoute,
route_instance_id: usize,
) -> Option<crate::client::NativePeerDataGeneration> {
let record = self
.connection_manager
.get_by_connection_id(connection_id)
.await;
let expected = self
.native_optional_route_generation_for_instance(connection_id, route, route_instance_id)
.await;
accepted_native_route_event_generation(record.as_ref(), expected, route_instance_id)
}
#[cfg(not(target_arch = "wasm32"))]
async fn report_current_native_optional_route_status_locked(
&self,
connection_id: &str,
route: KnownRoute,
route_instance_id: usize,
active_transport: &str,
parallel_transport: Option<&str>,
settled: bool,
) -> Option<bool> {
let normalized_active = crate::transport_label::normalize(active_transport)?.to_string();
let normalized_parallel = parallel_transport
.and_then(crate::transport_label::normalize)
.filter(|value| *value != normalized_active.as_str())
.map(str::to_string);
let mut installed = self.native_optional_route_generations.write().await;
let expected = installed
.get(connection_id)
.and_then(|generations| generations.get(route))
.filter(|generation| generation.route_instance_id == route_instance_id)?;
let record = self
.connection_manager
.get_by_connection_id(connection_id)
.await?;
if !native_optional_route_generation_matches(
Some(&record),
Some(expected),
route_instance_id,
) {
return None;
}
let transport_changed = record.active_transport != normalized_active
|| record.parallel_transport != normalized_parallel;
let snapshot = self
.connection_manager
.report_transport_status_if_current(
connection_id,
expected.transport_stable_id,
expected.transport_generation,
expected.route_generation,
normalized_active,
normalized_parallel,
)
.await?;
if let Some(generations) = installed.get_mut(connection_id) {
generations.sync_route_generation(
expected.transport_stable_id,
expected.transport_generation,
snapshot.active_route_generation,
);
}
if settled
&& self
.connection_manager
.set_settled_if_current(
connection_id,
expected.transport_stable_id,
expected.transport_generation,
snapshot.active_route_generation,
true,
)
.await
.is_none()
{
return None;
}
Some(transport_changed)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn report_current_native_webrtc_route_status(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
active_transport: &str,
parallel_transport: Option<&str>,
settled: bool,
) -> bool {
let sessions = self.native_webrtc_sessions.read().await;
if source.state() != crate::transport::NativeWebRTCState::Connected
|| !sessions
.get(connection_id)
.is_some_and(|current| Arc::ptr_eq(current, source))
{
return false;
}
let changed = self
.report_current_native_optional_route_status_locked(
connection_id,
KnownRoute::WebRtc,
Arc::as_ptr(source) as usize,
active_transport,
parallel_transport,
settled,
)
.await;
drop(sessions);
if changed == Some(true) {
self.emit_current_native_connection_state(connection_id)
.await;
}
changed.is_some()
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn report_current_native_moq_route_status(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeMoQSession>,
active_transport: &str,
parallel_transport: Option<&str>,
settled: bool,
) -> bool {
let sessions = self.native_moq_sessions.read().await;
if !matches!(
source.state(),
crate::transport::NativeMoQState::Connecting
| crate::transport::NativeMoQState::Connected
) || !sessions
.get(connection_id)
.is_some_and(|current| Arc::ptr_eq(current, source))
{
return false;
}
let changed = self
.report_current_native_optional_route_status_locked(
connection_id,
KnownRoute::Moq,
Arc::as_ptr(source) as usize,
active_transport,
parallel_transport,
settled,
)
.await;
drop(sessions);
if changed == Some(true) {
self.emit_current_native_connection_state(connection_id)
.await;
}
changed.is_some()
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) const NATIVE_MOQ_SEND_READY_RETRY_WINDOW_MS: u64 = 6_000;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) const NATIVE_MOQ_SEND_READY_RETRY_DELAY_MS: u64 = 250;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) const NATIVE_MOQ_SEND_REPAIR_RETRY_MIN_INTERVAL_MS: u64 = 1_000;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) const NATIVE_MOQ_SEND_MAX_REPAIR_REQUESTS: u32 = 3;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) const NATIVE_MOQ_DATA_READY_TIMEOUT_MS: u64 = 5_000;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) const NATIVE_MOQ_DATA_READY_STABLE_POLL_MS: u64 = 125;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) const NATIVE_MOQ_DATA_READY_STABLE_SAMPLES: u32 = 8;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_ADMISSION_FAILURE_BACKOFF_MS: i64 =
crate::runtime_policy::NATIVE_WEBRTC_ADMISSION_FAILURE_BACKOFF_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_CONNECT_TIMEOUT_BACKOFF_MS: i64 =
crate::runtime_policy::NATIVE_WEBRTC_CONNECT_TIMEOUT_BACKOFF_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS: u64 =
crate::runtime_policy::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_DIRECT_QUIC_SUPPRESSION_MS: i64 =
crate::runtime_policy::NATIVE_WEBRTC_DIRECT_QUIC_SUPPRESSION_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS: u32 =
crate::runtime_policy::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_SIGNAL_SEND_TIMEOUT_MS: u64 =
crate::runtime_policy::NATIVE_WEBRTC_SIGNAL_SEND_TIMEOUT_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_SIGNAL_SEND_RETRY_DELAY_MS: u64 =
crate::runtime_policy::NATIVE_WEBRTC_SIGNAL_SEND_RETRY_DELAY_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_CONNECT_TIMEOUT_MS: u64 =
crate::runtime_policy::NATIVE_WEBRTC_CONNECT_TIMEOUT_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) const NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS: u64 =
crate::runtime_policy::NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn native_webrtc_heartbeat_config() -> crate::heartbeat::HeartbeatConfig {
crate::heartbeat::HeartbeatConfig {
tick_interval: std::time::Duration::from_secs(
crate::runtime_policy::native_webrtc_heartbeat_tick_secs(),
),
suspect_after: std::time::Duration::from_secs(
crate::runtime_policy::native_webrtc_heartbeat_suspect_after_secs(),
),
stale_after: std::time::Duration::from_secs(
crate::runtime_policy::native_webrtc_heartbeat_stale_after_secs(),
),
send_timeout: std::time::Duration::from_secs(
crate::runtime_policy::native_webrtc_heartbeat_send_timeout_secs(),
),
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn current_native_webrtc_route_identity_matches(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
) -> bool {
let route_instance_id = Arc::as_ptr(source) as usize;
let is_current_session = self
.native_webrtc_sessions
.read()
.await
.get(connection_id)
.is_some_and(|current| Arc::ptr_eq(current, source));
is_current_session
&& self
.native_optional_route_generation_is_current(
connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn current_native_webrtc_route_matches(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
) -> bool {
native_webrtc_session_is_active(source.state())
&& self
.current_native_webrtc_route_identity_matches(connection_id, source)
.await
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn install_native_webrtc_route_proof(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
) -> Option<crate::client::NativeWebRTCRouteProofState> {
if !self
.current_native_webrtc_route_identity_matches(connection_id, source)
.await
{
return None;
}
let route_instance_id = Arc::as_ptr(source) as usize;
let state = crate::client::NativeWebRTCRouteProofState {
route_instance_id,
probe_id: format!("native-webrtc-{}", uuid::Uuid::new_v4().simple()),
negotiation_id: source.negotiation_id(),
retry_active: false,
proven: false,
};
self.native_webrtc_route_proofs
.write()
.await
.insert(connection_id.to_string(), state.clone());
Some(state)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn current_native_webrtc_route_is_proven(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
) -> bool {
let route_instance_id = Arc::as_ptr(source) as usize;
self.native_webrtc_route_proofs
.read()
.await
.get(connection_id)
.is_some_and(|proof| {
proof.route_instance_id == route_instance_id
&& proof.negotiation_id == source.negotiation_id()
&& proof.proven
})
&& self
.current_native_webrtc_route_matches(connection_id, source)
.await
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn mark_current_native_webrtc_route_proven(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
reason: &str,
) -> bool {
if !self
.current_native_webrtc_route_matches(connection_id, source)
.await
{
return false;
}
let route_instance_id = Arc::as_ptr(source) as usize;
let marked = {
let mut proofs = self.native_webrtc_route_proofs.write().await;
let Some(proof) = proofs.get_mut(connection_id) else {
return false;
};
if proof.route_instance_id != route_instance_id
|| proof.negotiation_id != source.negotiation_id()
{
return false;
}
let changed = !proof.proven;
proof.proven = true;
proof.retry_active = false;
changed
};
if !self
.current_native_webrtc_route_matches(connection_id, source)
.await
{
return false;
}
let parallel_transport = self
.resolve_parallel_iroh_transport_label(connection_id, remote_node_id)
.await;
let reported = self
.report_current_native_webrtc_route_status(
connection_id,
source,
crate::transport_label::WEBRTC,
Some(¶llel_transport),
true,
)
.await;
if marked && reported {
println!(
"[NativeWebRTC][route-proof] proven connection_id={} route_instance_id={} reason={}",
connection_id, route_instance_id, reason,
);
}
reported
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn request_current_native_webrtc_route_proof(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
) -> anyhow::Result<bool> {
if !self
.current_native_webrtc_route_matches(connection_id, source)
.await
{
return Ok(false);
}
let route_instance_id = Arc::as_ptr(source) as usize;
let negotiation_id = source.negotiation_id();
let should_start = {
let mut proofs = self.native_webrtc_route_proofs.write().await;
let Some(proof) = proofs.get_mut(connection_id) else {
return Ok(false);
};
if proof.route_instance_id != route_instance_id {
return Ok(false);
}
if proof.negotiation_id != negotiation_id {
proof.negotiation_id = negotiation_id;
proof.probe_id = format!("native-webrtc-{}", uuid::Uuid::new_v4().simple());
proof.proven = false;
proof.retry_active = false;
}
if proof.proven || proof.retry_active {
false
} else {
proof.retry_active = true;
true
}
};
if !should_start {
return Ok(false);
}
let client = self.clone();
let connection_id = connection_id.to_string();
let source = source.clone();
tokio::spawn(async move {
const MAX_ATTEMPTS: usize = 8;
const RETRY_DELAY_MS: u64 = 750;
for attempt in 1..=MAX_ATTEMPTS {
if !client
.current_native_webrtc_route_matches(&connection_id, &source)
.await
|| client
.current_native_webrtc_route_is_proven(&connection_id, &source)
.await
{
break;
}
if let Err(error) = client
.send_current_native_webrtc_route_proof_once(&connection_id, &source)
.await
{
println!(
"[NativeWebRTC][route-proof] probe failed connection_id={} attempt={} error={:#}",
connection_id, attempt, error,
);
}
if attempt < MAX_ATTEMPTS {
tokio::time::sleep(std::time::Duration::from_millis(RETRY_DELAY_MS)).await;
}
}
let mut proofs = client.native_webrtc_route_proofs.write().await;
if let Some(proof) = proofs.get_mut(&connection_id) {
if proof.route_instance_id == Arc::as_ptr(&source) as usize {
proof.retry_active = false;
}
}
});
Ok(true)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn send_current_native_webrtc_route_proof_once(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
) -> anyhow::Result<bool> {
if source.state() != crate::transport::NativeWebRTCState::Connected {
return Ok(false);
}
let route_instance_id = Arc::as_ptr(source) as usize;
let proof = self
.native_webrtc_route_proofs
.read()
.await
.get(connection_id)
.filter(|proof| proof.route_instance_id == route_instance_id && !proof.proven)
.cloned();
let Some(proof) = proof else {
return Ok(false);
};
let frame = NativeWebRtcRouteProofFrame {
frame_type: NATIVE_WEBRTC_ROUTE_PROOF_TYPE.to_string(),
version: 1,
role: "probe".to_string(),
probe_id: proof.probe_id,
negotiation_id: proof.negotiation_id,
};
let encoded = serde_json::to_vec(&frame)?;
let protected = self.protect_outbound_application_payload(connection_id, &encoded)?;
if !self
.current_native_webrtc_route_matches(connection_id, source)
.await
{
return Ok(false);
}
source.send(&protected).await?;
Ok(true)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn current_native_moq_route_identity_matches(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeMoQSession>,
) -> bool {
let route_instance_id = Arc::as_ptr(source) as usize;
let is_current_session = self
.native_moq_sessions
.read()
.await
.get(connection_id)
.is_some_and(|current| Arc::ptr_eq(current, source));
is_current_session
&& self
.native_optional_route_generation_is_current(
connection_id,
KnownRoute::Moq,
route_instance_id,
)
.await
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn current_native_moq_route_matches(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeMoQSession>,
) -> bool {
matches!(
source.state(),
crate::transport::NativeMoQState::Connecting
| crate::transport::NativeMoQState::Connected
) && self
.current_native_moq_route_identity_matches(connection_id, source)
.await
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn install_native_moq_route_proof(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeMoQSession>,
local_node_id: &str,
remote_node_id: &str,
) -> crate::client::NativeMoQRouteProofState {
let route_instance_id = Arc::as_ptr(source) as usize;
let state = crate::client::NativeMoQRouteProofState {
route_instance_id,
probe_id: format!("native-moq-{}", uuid::Uuid::new_v4().simple()),
local_node_id: local_node_id.to_string(),
remote_node_id: remote_node_id.to_string(),
retry_active: false,
proven: false,
};
self.native_moq_route_proofs
.write()
.await
.insert(connection_id.to_string(), state.clone());
state
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn current_native_moq_route_is_proven(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeMoQSession>,
) -> bool {
let route_instance_id = Arc::as_ptr(source) as usize;
self.native_moq_route_proofs
.read()
.await
.get(connection_id)
.is_some_and(|proof| proof.route_instance_id == route_instance_id && proof.proven)
&& self
.current_native_moq_route_matches(connection_id, source)
.await
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn mark_current_native_moq_route_proven(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
source: &Arc<crate::transport::NativeMoQSession>,
reason: &str,
) -> bool {
if !self
.current_native_moq_route_matches(connection_id, source)
.await
{
return false;
}
let route_instance_id = Arc::as_ptr(source) as usize;
let marked = {
let mut proofs = self.native_moq_route_proofs.write().await;
let Some(proof) = proofs.get_mut(connection_id) else {
return false;
};
if proof.route_instance_id != route_instance_id {
return false;
}
let changed = !proof.proven;
proof.proven = true;
proof.retry_active = false;
changed
};
if !self
.current_native_moq_route_matches(connection_id, source)
.await
{
return false;
}
let parallel_transport = self
.resolve_parallel_iroh_transport_label(connection_id, remote_node_id)
.await;
let report = successful_send_status_report(parallel_transport.as_str());
let reported = self
.report_current_native_moq_route_status(
connection_id,
source,
report.active_transport,
report.parallel_transport,
true,
)
.await;
if marked && reported {
println!(
"[NativeMoQ][route-proof] proven connection_id={} route_instance_id={} reason={}",
connection_id, route_instance_id, reason,
);
}
reported
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn request_current_native_moq_route_proof(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeMoQSession>,
) -> anyhow::Result<bool> {
if !self
.current_native_moq_route_matches(connection_id, source)
.await
{
return Ok(false);
}
let route_instance_id = Arc::as_ptr(source) as usize;
let should_start = {
let mut proofs = self.native_moq_route_proofs.write().await;
let Some(proof) = proofs.get_mut(connection_id) else {
return Ok(false);
};
if proof.route_instance_id != route_instance_id || proof.proven || proof.retry_active {
false
} else {
proof.retry_active = true;
true
}
};
if !should_start {
return Ok(false);
}
let client = self.clone();
let connection_id = connection_id.to_string();
let source = source.clone();
tokio::spawn(async move {
const MAX_ATTEMPTS: usize = 8;
const RETRY_DELAY_MS: u64 = 750;
for attempt in 1..=MAX_ATTEMPTS {
if !client
.current_native_moq_route_matches(&connection_id, &source)
.await
|| client
.current_native_moq_route_is_proven(&connection_id, &source)
.await
{
break;
}
if let Err(error) = client
.send_current_native_moq_route_proof_once(&connection_id, &source)
.await
{
println!(
"[NativeMoQ][route-proof] probe failed connection_id={} attempt={} error={:#}",
connection_id, attempt, error,
);
}
if attempt < MAX_ATTEMPTS {
tokio::time::sleep(std::time::Duration::from_millis(RETRY_DELAY_MS)).await;
}
}
let mut proofs = client.native_moq_route_proofs.write().await;
if let Some(proof) = proofs.get_mut(&connection_id) {
if proof.route_instance_id == Arc::as_ptr(&source) as usize {
proof.retry_active = false;
}
}
});
Ok(true)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn send_current_native_moq_route_proof_once(
&self,
connection_id: &str,
source: &Arc<crate::transport::NativeMoQSession>,
) -> anyhow::Result<bool> {
if !self
.wait_for_stable_native_moq_data_ready(connection_id, source)
.await
{
return Ok(false);
}
let route_instance_id = Arc::as_ptr(source) as usize;
let proof = self
.native_moq_route_proofs
.read()
.await
.get(connection_id)
.filter(|proof| proof.route_instance_id == route_instance_id && !proof.proven)
.cloned();
let Some(proof) = proof else {
return Ok(false);
};
let frame = NativeMoQRouteProofFrame {
frame_type: NATIVE_MOQ_ROUTE_PROOF_TYPE.to_string(),
version: 1,
role: "probe".to_string(),
probe_id: proof.probe_id,
sender_id: proof.local_node_id,
recipient_id: proof.remote_node_id,
};
let encoded = serde_json::to_vec(&frame)?;
let protected = self.protect_outbound_direct_moq_payload(connection_id, &encoded)?;
if !self
.current_native_moq_route_matches(connection_id, source)
.await
{
return Ok(false);
}
source.send(&protected).await?;
Ok(true)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) async fn handle_current_native_moq_terminal_failure(
&self,
connection_id: &str,
remote_node_id: &str,
source: Arc<crate::transport::NativeMoQSession>,
) -> crate::client::NativeOptionalRouteTerminalDisposition {
use crate::client::NativeOptionalRouteTerminalDisposition::{Handled, Stale};
if !self
.current_native_moq_route_identity_matches(connection_id, &source)
.await
{
println!(
"[NativeMoQ] stale terminal callback ignored connection_id={} remote_node_id={}",
connection_id, remote_node_id,
);
return Stale;
}
let route_instance_id = Arc::as_ptr(&source) as usize;
let Some(expected) = self
.current_native_optional_route_event_generation(
connection_id,
KnownRoute::Moq,
route_instance_id,
)
.await
else {
return Stale;
};
let removed = {
let mut sessions = self.native_moq_sessions.write().await;
if sessions
.get(connection_id)
.is_some_and(|current| Arc::ptr_eq(current, &source))
{
sessions.remove(connection_id);
true
} else {
false
}
};
if !removed {
return Stale;
}
self.clear_native_optional_route_generation_if_current(
connection_id,
KnownRoute::Moq,
route_instance_id,
)
.await;
drop(source);
let fallback_transport = self
.resolve_parallel_iroh_transport_label(connection_id, Some(remote_node_id))
.await;
let demoted = self
.report_transport_status_for_generation(
connection_id,
&fallback_transport,
None,
expected.transport_stable_id,
expected.transport_generation,
expected.route_generation,
)
.await
.is_some();
if !demoted {
return Handled {
recovery_started: false,
};
}
let recovery_started = match self
.maybe_start_native_moq_upgrade_with_options(
connection_id,
Some(remote_node_id),
NativeMoqUpgradeOptions::OPPORTUNISTIC_REPAIR,
)
.await
{
Ok(restarted) => restarted,
Err(error) => {
println!(
"[NativeMoQ] recovery setup failed connection_id={} remote_node_id={} error={:#}",
connection_id, remote_node_id, error,
);
false
}
};
Handled { recovery_started }
}
#[cfg(all(
not(target_arch = "wasm32"),
any(feature = "transport-webrtc", feature = "transport-moq")
))]
pub(crate) async fn handle_native_transport_peer_data(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
transport: &str,
generation: crate::client::NativePeerDataGeneration,
payload: bytes::Bytes,
) -> anyhow::Result<()> {
let payload = if transport == "moq" {
self.open_inbound_direct_moq_payload(connection_id, &payload)?
} else {
self.open_inbound_application_payload(connection_id, &payload)?
};
self.emit_native_peer_data(crate::client::NativePeerDataEvent {
connection_id: connection_id.to_string(),
remote_node_id: remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned),
transport: transport.to_string(),
transport_stable_id: generation.transport_stable_id,
transport_generation: generation.transport_generation,
route_generation: generation.route_generation,
payload,
});
Ok(())
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn handle_current_native_webrtc_peer_data(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
source: &Arc<crate::transport::NativeWebRTCDataChannel>,
payload: bytes::Bytes,
) -> anyhow::Result<bool> {
let is_current = self
.current_native_webrtc_route_matches(connection_id, source)
.await;
if !is_current || source.state() != crate::transport::NativeWebRTCState::Connected {
println!(
"[NativeWebRTC] stale peer-data ignored connection_id={} remote_node_id={:?} negotiation_id={}",
connection_id,
remote_node_id,
source.negotiation_id(),
);
return Ok(false);
}
let route_instance_id = Arc::as_ptr(source) as usize;
let Some(generation) = self
.current_native_optional_route_event_generation(
connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await
else {
return Ok(false);
};
if !self
.current_native_webrtc_route_matches(connection_id, source)
.await
{
return Ok(false);
}
let payload = self.open_inbound_application_payload(connection_id, &payload)?;
if let Some(frame) = parse_native_webrtc_route_proof(&payload) {
if frame.role == "probe" {
let ack = NativeWebRtcRouteProofFrame {
frame_type: NATIVE_WEBRTC_ROUTE_PROOF_TYPE.to_string(),
version: 1,
role: "ack".to_string(),
probe_id: frame.probe_id,
negotiation_id: frame.negotiation_id,
};
let ack = serde_json::to_vec(&ack)?;
let protected = self.protect_outbound_application_payload(connection_id, &ack)?;
if !self
.current_native_webrtc_route_matches(connection_id, source)
.await
{
return Ok(false);
}
source.send(&protected).await?;
} else {
let route_instance_id = Arc::as_ptr(source) as usize;
let matching = self
.native_webrtc_route_proofs
.read()
.await
.get(connection_id)
.is_some_and(|proof| {
proof.route_instance_id == route_instance_id
&& proof.probe_id == frame.probe_id
&& proof.negotiation_id == frame.negotiation_id
});
if matching {
let _ = self
.mark_current_native_webrtc_route_proven(
connection_id,
remote_node_id,
source,
"matching-protected-ack",
)
.await;
}
}
return Ok(true);
}
self.emit_native_peer_data(crate::client::NativePeerDataEvent {
connection_id: connection_id.to_string(),
remote_node_id: remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned),
transport: crate::transport_label::WEBRTC.to_string(),
transport_stable_id: generation.transport_stable_id,
transport_generation: generation.transport_generation,
route_generation: generation.route_generation,
payload,
});
Ok(true)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) async fn handle_current_native_moq_peer_data(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
source: &Arc<crate::transport::NativeMoQSession>,
payload: bytes::Bytes,
) -> anyhow::Result<bool> {
let is_current = self
.current_native_moq_route_matches(connection_id, source)
.await;
if !is_current || source.state() != crate::transport::NativeMoQState::Connected {
println!(
"[NativeMoQ] stale peer-data ignored connection_id={} remote_node_id={:?}",
connection_id, remote_node_id,
);
return Ok(false);
}
let route_instance_id = Arc::as_ptr(source) as usize;
let Some(generation) = self
.current_native_optional_route_event_generation(
connection_id,
KnownRoute::Moq,
route_instance_id,
)
.await
else {
return Ok(false);
};
if !self
.current_native_moq_route_matches(connection_id, source)
.await
{
return Ok(false);
}
let route_proof = self
.open_inbound_direct_moq_route_proof(connection_id, &payload)
.ok()
.and_then(|payload| parse_native_moq_route_proof(&payload));
if let Some(frame) = route_proof {
let local_node_id = self.current_node_id().await;
let expected_remote_node_id = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty());
if local_node_id.as_deref() != Some(frame.recipient_id.as_str())
|| expected_remote_node_id != Some(frame.sender_id.as_str())
{
return Ok(true);
}
if frame.role == "probe" {
let ack = NativeMoQRouteProofFrame {
frame_type: NATIVE_MOQ_ROUTE_PROOF_TYPE.to_string(),
version: 1,
role: "ack".to_string(),
probe_id: frame.probe_id,
sender_id: frame.recipient_id,
recipient_id: frame.sender_id,
};
let encoded = serde_json::to_vec(&ack)?;
let protected =
self.protect_outbound_direct_moq_payload(connection_id, &encoded)?;
if !self
.current_native_moq_route_matches(connection_id, source)
.await
{
return Ok(false);
}
source.send(&protected).await?;
} else {
let route_instance_id = Arc::as_ptr(source) as usize;
let matching = self
.native_moq_route_proofs
.read()
.await
.get(connection_id)
.is_some_and(|proof| {
proof.route_instance_id == route_instance_id
&& proof.probe_id == frame.probe_id
&& proof.local_node_id == frame.recipient_id
&& proof.remote_node_id == frame.sender_id
});
if matching {
let _ = self
.mark_current_native_moq_route_proven(
connection_id,
remote_node_id,
source,
"matching-protected-ack",
)
.await;
}
}
return Ok(true);
}
let payload = self.open_inbound_direct_moq_payload(connection_id, &payload)?;
self.emit_native_peer_data(crate::client::NativePeerDataEvent {
connection_id: connection_id.to_string(),
remote_node_id: remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned),
transport: crate::transport_label::MOQ.to_string(),
transport_stable_id: generation.transport_stable_id,
transport_generation: generation.transport_generation,
route_generation: generation.route_generation,
payload,
});
Ok(true)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn create_native_webrtc_negotiation_id(connection_id: &str) -> String {
format!(
"native-{}-{}",
connection_id.replace(':', "-"),
now_millis_i64()
)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn native_webrtc_session_for_connection(
&self,
connection_id: &str,
) -> Option<Arc<crate::transport::NativeWebRTCDataChannel>> {
self.native_webrtc_sessions
.read()
.await
.get(connection_id)
.cloned()
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn suppress_native_webrtc_restarts(
&self,
connection_id: &str,
duration_ms: i64,
reason: impl Into<String>,
) {
if connection_id.trim().is_empty() || duration_ms <= 0 {
return;
}
let reason = reason.into();
let until_ms = now_millis_i64().saturating_add(duration_ms);
self.native_webrtc_suppressions.write().await.insert(
connection_id.to_string(),
crate::client::NativeWebRTCSuppression {
until_ms,
reason: reason.clone(),
},
);
println!(
"[NativeWebRTC] suppressing restart connection_id={} duration_ms={} until_ms={} reason={}",
connection_id,
duration_ms,
until_ms,
reason,
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn active_native_webrtc_suppression(
&self,
connection_id: &str,
) -> Option<crate::client::NativeWebRTCSuppression> {
let now = now_millis_i64();
let mut suppressions = self.native_webrtc_suppressions.write().await;
let suppression = suppressions.get(connection_id).cloned();
match suppression {
Some(value) if value.until_ms > now => Some(value),
Some(_) => {
suppressions.remove(connection_id);
None
}
None => None,
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
const NATIVE_WEBRTC_ATTEMPT_IN_FLIGHT_GRACE_MS: i64 =
crate::runtime_policy::NATIVE_WEBRTC_ATTEMPT_IN_FLIGHT_GRACE_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn active_native_webrtc_attempt(
&self,
connection_id: &str,
) -> Option<crate::client::NativeWebRTCAttemptInFlight> {
let now = now_millis_i64();
let mut attempts = self.native_webrtc_attempts_in_flight.write().await;
match attempts.get(connection_id).cloned() {
Some(value) if value.expires_at_ms > now => Some(value),
Some(_) => {
attempts.remove(connection_id);
None
}
None => None,
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn record_native_webrtc_attempt_started(
&self,
connection_id: &str,
attempt_id: &str,
negotiation_id: &str,
transport_stable_id: Option<u64>,
preserve_on_direct_path: bool,
) {
let now = now_millis_i64();
let expires_at_ms = now
.saturating_add(Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_MS as i64)
.saturating_add(Self::NATIVE_WEBRTC_ATTEMPT_IN_FLIGHT_GRACE_MS);
self.native_webrtc_attempts_in_flight.write().await.insert(
connection_id.to_string(),
crate::client::NativeWebRTCAttemptInFlight {
attempt_id: attempt_id.to_string(),
negotiation_id: negotiation_id.to_string(),
transport_stable_id,
started_at_ms: now,
expires_at_ms,
preserve_on_direct_path,
last_skip_log_ms: 0,
},
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
const NATIVE_WEBRTC_SKIP_LOG_INTERVAL_MS: i64 =
crate::runtime_policy::NATIVE_WEBRTC_SKIP_LOG_INTERVAL_MS;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn should_log_native_webrtc_skip(&self, connection_id: &str) -> bool {
let now = now_millis_i64();
let mut attempts = self.native_webrtc_attempts_in_flight.write().await;
let Some(entry) = attempts.get_mut(connection_id) else {
return true;
};
if entry.last_skip_log_ms == 0
|| now.saturating_sub(entry.last_skip_log_ms)
>= Self::NATIVE_WEBRTC_SKIP_LOG_INTERVAL_MS
{
entry.last_skip_log_ms = now;
return true;
}
false
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn clear_native_webrtc_attempt_if_current(
&self,
connection_id: &str,
attempt_id: &str,
) {
let mut attempts = self.native_webrtc_attempts_in_flight.write().await;
if let Some(existing) = attempts.get(connection_id) {
if existing.attempt_id == attempt_id {
attempts.remove(connection_id);
}
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn clear_native_webrtc_attempt(&self, connection_id: &str) {
self.native_webrtc_attempts_in_flight
.write()
.await
.remove(connection_id);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn clear_native_webrtc_suppression(
&self,
connection_id: &str,
reason_prefix: Option<&str>,
) -> bool {
let mut suppressions = self.native_webrtc_suppressions.write().await;
match suppressions.get(connection_id) {
Some(value)
if reason_prefix.map(|prefix| value.reason.starts_with(prefix)) != Some(false) =>
{
suppressions.remove(connection_id);
true
}
_ => false,
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn reset_native_webrtc_attempt_budget(
&self,
connection_id: &str,
reason: &str,
) {
let removed = self
.native_webrtc_attempt_counts
.write()
.await
.remove(connection_id);
let cleared_suppression = self
.clear_native_webrtc_suppression(connection_id, Some("upgrade-attempts-exhausted"))
.await;
if removed.is_some() || cleared_suppression {
println!(
"[NativeWebRTC] reset attempt budget connection_id={} reason={} had_counter={} cleared_exhausted_suppression={}",
connection_id,
reason,
removed.is_some(),
cleared_suppression,
);
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn close_native_webrtc_session_for_replacement(
&self,
connection_id: &str,
reason: &str,
) -> bool {
let session = self
.native_webrtc_session_for_connection(connection_id)
.await;
let Some(session) = session else {
self.evict_native_signal_stream(connection_id, reason).await;
return false;
};
session.close();
let removed = self
.remove_native_webrtc_session_if_current(connection_id, &session, reason)
.await;
self.evict_native_signal_stream(connection_id, reason).await;
if removed {
self.finalize_deferred_managed_retirement_if_needed(connection_id)
.await;
}
removed
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn retire_replaced_peer_connection(
&self,
record: crate::connection_manager::ConnectionRecord,
reason: &str,
) {
self.forget_session_connection(&record.connection_id);
self.cancel_native_webrtc_retry(&record.connection_id).await;
self.reset_native_webrtc_attempt_budget(&record.connection_id, reason)
.await;
let _ = self
.clear_native_webrtc_suppression(&record.connection_id, None)
.await;
let _ = self
.close_native_webrtc_session_for_replacement(&record.connection_id, reason)
.await;
if let Some(endpoint_str) = record.endpoint_id.as_deref().or(record.node_id.as_deref()) {
if let Ok(endpoint_id) = endpoint_str.parse::<iroh::EndpointId>() {
if let Some(connection) = self.get_connection(endpoint_id).await {
connection.close(0u8.into(), reason.as_bytes());
}
}
}
self.retire_managed_connection_now(&record.connection_id, Some(reason.to_string()))
.await;
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn accept_replacement_peer(
&self,
connection_id: &str,
authoritative_device_id: Option<&str>,
reason: &str,
) {
let phase7_ctx = self.correlation_for_connection(connection_id).await;
crate::clog!(
"[NativeWebRTC][replacement-peer]",
&phase7_ctx,
"begin reason={} authoritative_device_id={}",
reason,
authoritative_device_id.unwrap_or("<none>")
);
let current_record = self
.connection_manager
.get_by_connection_id(connection_id)
.await;
if let Some(device_id) = authoritative_device_id
.map(str::trim)
.filter(|value| !value.is_empty())
{
let _ = self
.connection_manager
.set_device_id(connection_id, device_id.to_string())
.await;
}
let mut stale_records =
std::collections::HashMap::<String, crate::connection_manager::ConnectionRecord>::new();
let authoritative_device_id = authoritative_device_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
let mut candidate_device_ids = std::collections::HashSet::<String>::new();
if let Some(device_id) = authoritative_device_id.as_deref() {
candidate_device_ids.insert(device_id.to_string());
}
if let Some(record) = current_record.as_ref() {
if let Some(device_id) = record.device_id.as_deref() {
candidate_device_ids.insert(device_id.to_string());
}
if let Some(device_id_hint) = record.device_id_hint.as_deref() {
candidate_device_ids.insert(device_id_hint.to_string());
}
}
for device_id in candidate_device_ids {
for record in self.connection_manager.get_by_device_id(&device_id).await {
if record.connection_id != connection_id {
stale_records.insert(record.connection_id.clone(), record);
}
}
}
if stale_records.is_empty() {
let current_hint = current_record.as_ref().and_then(|record| {
authoritative_device_id
.clone()
.or_else(|| record.device_id.clone())
.or_else(|| record.device_id_hint.clone())
});
if let Some(current_hint) = current_hint {
for record in self.connection_manager.list_active().await {
if record.connection_id == connection_id {
continue;
}
let matches_device_hint = record.device_id.as_deref()
== Some(current_hint.as_str())
|| record.device_id_hint.as_deref() == Some(current_hint.as_str());
if matches_device_hint {
stale_records.insert(record.connection_id.clone(), record);
}
}
}
}
for (_, record) in stale_records {
println!(
"[NativeWebRTC] replacement peer admitted: retiring stale connection_id={} replaced_by={} reason={}",
record.connection_id,
connection_id,
reason,
);
self.retire_replaced_peer_connection(record, reason).await;
}
self.cancel_native_webrtc_retry(connection_id).await;
self.reset_native_webrtc_attempt_budget(connection_id, reason)
.await;
let _ = self
.clear_native_webrtc_suppression(connection_id, None)
.await;
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn request_native_webrtc_recovery(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
trigger: NativeWebRTCRecoveryTrigger,
options: NativeWebRTCRecoveryOptions<'_>,
) -> anyhow::Result<()> {
if connection_id.trim().is_empty() || !self.is_webrtc_transport_enabled().await {
println!(
"[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason={}",
connection_id,
trigger.as_str(),
if connection_id.trim().is_empty() {
"empty-connection-id"
} else {
"transport-disabled"
},
);
return Ok(());
}
let preferred_negotiation_id = options
.preferred_negotiation_id
.map(str::trim)
.filter(|value| !value.is_empty());
let Some(connection_record) = self
.connection_manager
.get_by_connection_id(connection_id)
.await
else {
println!(
"[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=missing-connection-record remote_node_id={:?} preferred_negotiation_id={}",
connection_id,
trigger.as_str(),
remote_node_id,
preferred_negotiation_id.unwrap_or("none"),
);
return Ok(());
};
let resolved_remote_node_id = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.or_else(|| connection_record.endpoint_id.clone())
.or_else(|| connection_record.node_id.clone());
let local_node_id = self.current_node_id().await;
if admission_trigger_defers_to_remote_initiator(
trigger,
local_node_id.as_deref(),
resolved_remote_node_id.as_deref(),
preferred_negotiation_id,
) {
println!(
"[NativeWebRTC][lifecycle] request deferred connection_id={} trigger={} reason=deterministic-responder local_node_id={} remote_node_id={}",
connection_id,
trigger.as_str(),
local_node_id.as_deref().unwrap_or("none"),
resolved_remote_node_id.as_deref().unwrap_or("none"),
);
return Ok(());
}
if self.session_registry_active() {
match self.session_admission(connection_id) {
crate::session_token::SessionAdmission::Pending => {
println!(
"[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=admission-pending preferred_negotiation_id={}",
connection_id,
trigger.as_str(),
preferred_negotiation_id.unwrap_or("none"),
);
return Ok(());
}
crate::session_token::SessionAdmission::Rejected { reason } => {
println!(
"[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=admission-rejected admission_reason={} preferred_negotiation_id={}",
connection_id,
trigger.as_str(),
reason,
preferred_negotiation_id.unwrap_or("none"),
);
return Ok(());
}
crate::session_token::SessionAdmission::Accepted { .. } => {}
}
}
if crate::native_webrtc_policy::native_webrtc_suppression_blocks_recovery(
preferred_negotiation_id,
) {
if let Some(suppression) = self.active_native_webrtc_suppression(connection_id).await {
println!(
"[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=restart-suppressed suppression_reason={} remaining_ms={} preferred_negotiation_id={} force_restart={}",
connection_id,
trigger.as_str(),
suppression.reason,
suppression.until_ms.saturating_sub(now_millis_i64()),
preferred_negotiation_id.unwrap_or("none"),
options.force_restart,
);
return Ok(());
}
} else if let Some(suppression) = self.active_native_webrtc_suppression(connection_id).await
{
println!(
"[NativeWebRTC][lifecycle] request continuing through suppression connection_id={} trigger={} suppression_reason={} remaining_ms={} preferred_negotiation_id={} force_restart={}",
connection_id,
trigger.as_str(),
suppression.reason,
suppression.until_ms.saturating_sub(now_millis_i64()),
preferred_negotiation_id.unwrap_or("none"),
options.force_restart,
);
}
if let Some(existing) = self
.native_webrtc_session_for_connection(connection_id)
.await
{
let active_negotiation_id = existing.negotiation_id();
let has_active_attempt = self
.active_native_webrtc_attempt(connection_id)
.await
.is_some();
if should_ignore_stale_inbound_signal_recovery(
trigger,
existing.state(),
active_negotiation_id.as_str(),
preferred_negotiation_id,
has_active_attempt,
) {
println!(
"[NativeWebRTC] stale inbound signal ignored connection_id={} trigger={} active_negotiation_id={} signal_negotiation_id={} has_active_attempt={}",
connection_id,
trigger.as_str(),
active_negotiation_id,
preferred_negotiation_id.unwrap_or("none"),
has_active_attempt,
);
return Ok(());
}
}
if let Some(in_flight) = self.active_native_webrtc_attempt(connection_id).await {
let transport_generation_replaced =
in_flight.transport_stable_id != connection_record.transport_stable_id;
let matches_in_flight = preferred_negotiation_id
.map(|value| value == in_flight.negotiation_id)
.unwrap_or(false);
let in_flight_age_ms = now_millis_i64().saturating_sub(in_flight.started_at_ms);
let is_preempting_trigger =
trigger.can_preempt_in_flight_attempt(preferred_negotiation_id, matches_in_flight);
if transport_generation_replaced {
println!(
"[NativeWebRTC][lifecycle] preempting stale transport-generation attempt connection_id={} trigger={} attempt_id={} in_flight_transport_stable_id={:?} current_transport_stable_id={:?}",
connection_id,
trigger.as_str(),
in_flight.attempt_id,
in_flight.transport_stable_id,
connection_record.transport_stable_id,
);
self.clear_native_webrtc_attempt(connection_id).await;
} else if options.force_restart {
let stale_in_flight =
in_flight_age_ms >= Self::NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS as i64;
if is_preempting_trigger || stale_in_flight {
println!(
"[NativeWebRTC][lifecycle] force-restart preempting in-flight attempt connection_id={} trigger={} coordinator={:?} in_flight_negotiation_id={} preferred_negotiation_id={} matches_in_flight={} stale_in_flight={} age_ms={}",
connection_id,
trigger.as_str(),
trigger.coordinator(),
in_flight.negotiation_id,
preferred_negotiation_id.unwrap_or("none"),
matches_in_flight,
stale_in_flight,
in_flight_age_ms,
);
self.clear_native_webrtc_attempt(connection_id).await;
} else {
println!(
"[NativeWebRTC][lifecycle] force-restart deferred to active attempt connection_id={} trigger={} coordinator={:?} in_flight_negotiation_id={} preferred_negotiation_id={} age_ms={}",
connection_id,
trigger.as_str(),
trigger.coordinator(),
in_flight.negotiation_id,
preferred_negotiation_id.unwrap_or("none"),
in_flight_age_ms,
);
return Ok(());
}
} else {
if !is_preempting_trigger && !matches_in_flight {
if self.should_log_native_webrtc_skip(connection_id).await {
println!(
"[NativeWebRTC] upgrade skipped: attempt in flight connection_id={} trigger={} coordinator={:?} force_restart={} in_flight_negotiation_id={} preferred_negotiation_id={} age_ms={}",
connection_id,
trigger.as_str(),
trigger.coordinator(),
options.force_restart,
in_flight.negotiation_id,
preferred_negotiation_id.unwrap_or("none"),
in_flight_age_ms,
);
}
return Ok(());
}
if is_preempting_trigger {
println!(
"[NativeWebRTC][lifecycle] preempting in-flight attempt connection_id={} trigger={} coordinator={:?} in_flight_negotiation_id={} preferred_negotiation_id={} age_ms={}",
connection_id,
trigger.as_str(),
trigger.coordinator(),
in_flight.negotiation_id,
preferred_negotiation_id.unwrap_or("none"),
in_flight_age_ms,
);
self.clear_native_webrtc_attempt(connection_id).await;
}
}
}
if trigger.skips_when_session_active() && !options.force_restart {
if let Some(existing) = self
.native_webrtc_session_for_connection(connection_id)
.await
{
let negotiation_mismatch = preferred_negotiation_id
.map(|value| value != existing.negotiation_id().as_str())
.unwrap_or(false);
if !negotiation_mismatch
&& matches!(
existing.state(),
crate::transport::NativeWebRTCState::Connecting
| crate::transport::NativeWebRTCState::Connected
)
{
println!(
"[NativeWebRTC][lifecycle] request skipped connection_id={} trigger={} reason=active-session state={:?} negotiation_id={} preferred_negotiation_id={} force_restart={}",
connection_id,
trigger.as_str(),
existing.state(),
existing.negotiation_id(),
preferred_negotiation_id.unwrap_or("none"),
options.force_restart,
);
return Ok(());
}
}
}
println!(
"[NativeWebRTC] recovery request connection_id={} remote_node_id={:?} trigger={} force_restart={} preferred_negotiation_id={}",
connection_id,
remote_node_id,
trigger.as_str(),
options.force_restart,
preferred_negotiation_id.unwrap_or("none"),
);
let role_override = options.role_override.or(match trigger {
NativeWebRTCRecoveryTrigger::BrowserSignal(
NativeWebRTCBrowserSignalTrigger::InboundSdp,
) => Some(crate::transport::NativeWebRTCRole::Responder),
_ => None,
});
self.maybe_start_native_webrtc_upgrade(
connection_id,
remote_node_id,
options.force_restart,
preferred_negotiation_id,
role_override,
)
.await
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) fn reserve_native_webrtc_attempt_slot(
current: u32,
) -> NativeWebRTCAttemptReservation {
crate::native_webrtc_policy::reserve_attempt_slot(
current,
Self::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS,
)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn native_webrtc_upgrade_fast_path_skip_reason(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
force_restart: bool,
preferred_negotiation_id: Option<&str>,
) -> Option<String> {
if self
.connection_manager
.get_by_connection_id(connection_id)
.await
.is_none()
{
return Some("missing connection record".to_string());
}
if self.session_registry_active() {
match self.session_admission(connection_id) {
crate::session_token::SessionAdmission::Pending => {
return Some("session admission pending".to_string());
}
crate::session_token::SessionAdmission::Rejected { reason } => {
return Some(format!("session admission rejected: {}", reason));
}
crate::session_token::SessionAdmission::Accepted { .. } => {}
}
}
if crate::native_webrtc_policy::native_webrtc_suppression_blocks_recovery(
preferred_negotiation_id,
) {
if let Some(suppression) = self.active_native_webrtc_suppression(connection_id).await {
return Some(format!(
"restart suppressed reason={} remaining_ms={}",
suppression.reason,
suppression.until_ms.saturating_sub(now_millis_i64())
));
}
} else if let Some(suppression) = self.active_native_webrtc_suppression(connection_id).await
{
println!(
"[NativeWebRTC][lifecycle] fast-path continuing through suppression connection_id={} suppression_reason={} remaining_ms={} preferred_negotiation_id={}",
connection_id,
suppression.reason,
suppression.until_ms.saturating_sub(now_millis_i64()),
preferred_negotiation_id.unwrap_or("none"),
);
}
if preferred_negotiation_id.is_none() {
let attempt_exhausted = {
let counts = self.native_webrtc_attempt_counts.read().await;
counts.get(connection_id).copied().unwrap_or(0)
> Self::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS
};
if attempt_exhausted {
return Some("attempt budget exhausted".to_string());
}
}
if preferred_negotiation_id.is_none() {
if let Some(remote_node_id) = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
{
let primary_iroh_path = self.iroh_path_kind(remote_node_id).await;
if !iroh_path_allows_opportunistic_optional_route(primary_iroh_path) {
return Some(format!(
"optimal iroh path is primary path={}",
primary_iroh_path.transport_label()
));
}
}
}
if let Some(existing) = self
.native_webrtc_session_for_connection(connection_id)
.await
{
let state = existing.state();
let negotiation_mismatch = preferred_negotiation_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value != existing.negotiation_id().as_str())
.unwrap_or(false);
let is_stale = state == crate::transport::NativeWebRTCState::Failed
|| state == crate::transport::NativeWebRTCState::Closed
|| (force_restart
&& matches!(
state,
crate::transport::NativeWebRTCState::Connecting
| crate::transport::NativeWebRTCState::Connected
))
|| negotiation_mismatch;
if !is_stale {
return Some(format!(
"existing session active state={:?} negotiation_id={}",
state,
existing.negotiation_id()
));
}
}
None
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn acquire_native_webrtc_start_gate(
&self,
connection_id: &str,
_remote_node_id: &str,
_force_restart: bool,
_preferred_negotiation_id: Option<&str>,
) -> Arc<tokio::sync::Notify> {
loop {
let wait_for = {
let mut starts = self.native_webrtc_start_gates.lock().await;
if let Some(existing) = starts.get(connection_id) {
Some(existing.clone())
} else {
let gate = Arc::new(tokio::sync::Notify::new());
starts.insert(connection_id.to_string(), gate.clone());
return gate;
}
};
wait_for
.expect("existing start gate must be present")
.notified()
.await;
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn release_native_webrtc_start_gate(
&self,
connection_id: &str,
gate: Arc<tokio::sync::Notify>,
) {
let removed = {
let mut starts = self.native_webrtc_start_gates.lock().await;
match starts.get(connection_id) {
Some(current) if Arc::ptr_eq(current, &gate) => {
starts.remove(connection_id);
true
}
_ => false,
}
};
if removed {
gate.notify_waiters();
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn cancel_native_webrtc_retry(&self, connection_id: &str) {
self.native_webrtc_retry_deadlines
.lock()
.await
.remove(connection_id);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
fn schedule_native_webrtc_retry(
&self,
connection_id: &str,
remote_node_id: &str,
expected_base: crate::client::NativePeerDataGeneration,
delay_ms: u64,
reason_prefix: &'static str,
) {
if connection_id.trim().is_empty() || remote_node_id.trim().is_empty() {
return;
}
let connection_id = connection_id.to_string();
let remote_node_id = remote_node_id.to_string();
let client = self.clone();
let due_at_ms = now_millis_i64().saturating_add(delay_ms.min(i64::MAX as u64) as i64);
let should_schedule = {
let deadlines = self.native_webrtc_retry_deadlines.clone();
let connection_id = connection_id.clone();
async move {
let now = now_millis_i64();
let mut deadlines = deadlines.lock().await;
if deadlines
.get(&connection_id)
.copied()
.is_some_and(|existing| existing > now)
{
return false;
}
deadlines.insert(connection_id, due_at_ms);
true
}
};
tokio::spawn(async move {
if !should_schedule.await {
return;
}
println!(
"[NativeWebRTC] scheduling retry connection_id={} remote_node_id={} delay_ms={} reason={}",
connection_id,
remote_node_id,
delay_ms,
reason_prefix,
);
tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
let should_run = {
let mut deadlines = client.native_webrtc_retry_deadlines.lock().await;
match deadlines.get(&connection_id).copied() {
Some(current_due_at_ms) if current_due_at_ms == due_at_ms => {
deadlines.remove(&connection_id);
true
}
_ => false,
}
};
if !should_run {
return;
}
let base_generation_is_current = client
.current_native_peer_data_generation(
&connection_id,
Some(expected_base.transport_stable_id),
)
.await
== Some(expected_base);
if !base_generation_is_current
|| client
.native_webrtc_session_for_connection(&connection_id)
.await
.is_some_and(|session| native_webrtc_session_is_active(session.state()))
{
return;
}
let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
return;
};
if let Err(error) = client
.ensure_connected_with_timeout(endpoint_id, std::time::Duration::from_secs(8))
.await
{
println!(
"[NativeWebRTC] scheduled retry skipped: iroh redial unavailable connection_id={} remote_node_id={} reason={} error={}",
connection_id,
remote_node_id,
reason_prefix,
error,
);
return;
}
let base_generation_is_still_current = client
.current_native_peer_data_generation(
&connection_id,
Some(expected_base.transport_stable_id),
)
.await
== Some(expected_base);
if !base_generation_is_still_current {
return;
}
if matches!(
client.session_admission(&connection_id),
crate::session_token::SessionAdmission::Rejected { .. }
) {
return;
}
let _ = client
.clear_native_webrtc_suppression(&connection_id, Some(reason_prefix))
.await;
if let Err(error) = client
.request_native_webrtc_recovery(
&connection_id,
Some(remote_node_id.as_str()),
NativeWebRTCRecoveryTrigger::Native(NativeWebRTCNativeTrigger::ScheduledRetry),
NativeWebRTCRecoveryOptions {
force_restart: true,
preferred_negotiation_id: None,
role_override: None,
},
)
.await
{
println!(
"[NativeWebRTC] scheduled retry failed connection_id={} remote_node_id={} error={}",
connection_id,
remote_node_id,
error,
);
}
});
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn schedule_native_webrtc_retry_after_route_retirement(
&self,
connection_id: &str,
remote_node_id: &str,
retired_route: crate::client::NativeOptionalRouteGeneration,
delay_ms: u64,
reason_prefix: &'static str,
) {
let Some(expected_base) = self
.current_native_peer_data_generation(
connection_id,
Some(retired_route.transport_stable_id),
)
.await
else {
return;
};
self.schedule_native_webrtc_retry(
connection_id,
remote_node_id,
expected_base,
delay_ms,
reason_prefix,
);
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) fn send_native_main_frame<'a>(
&'a self,
connection_id: &'a str,
endpoint_id: iroh::EndpointId,
signal_type: &'a str,
frame_buf: &'a [u8],
) -> NativeMainSendFuture<'a> {
Box::pin(self.send_native_main_frame_inner(
connection_id,
endpoint_id,
signal_type,
frame_buf,
))
}
#[cfg(not(target_arch = "wasm32"))]
async fn send_native_main_frame_inner(
&self,
connection_id: &str,
endpoint_id: iroh::EndpointId,
signal_type: &str,
frame_buf: &[u8],
) -> Result<(), String> {
use tokio::io::AsyncWriteExt;
let current_transport_stable_id = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64)
.ok_or_else(|| format!("missing Iroh connection for endpoint {}", endpoint_id))?;
let cached = {
let map = self.native_control_streams.lock().await;
map.get(connection_id).cloned()
};
if let Some(entry) = cached {
if entry.endpoint_id != endpoint_id
|| entry.owner.transport_stable_id != current_transport_stable_id
{
self.evict_native_signal_stream_if_owner(
connection_id,
entry.owner,
"cached-owner-stale",
)
.await;
} else {
let mut send_guard = entry.send.lock().await;
match send_guard.write_all(frame_buf).await {
Ok(_) => {
if let Err(flush_err) = send_guard.flush().await {
println!(
"[SIGNAL-STREAM] flush-failed-on-cached connection_id={} signal_type={} error={}",
connection_id, signal_type, flush_err,
);
drop(send_guard);
self.evict_native_signal_stream_if_owner(
connection_id,
entry.owner,
"cached-flush-failed",
)
.await;
} else {
println!(
"[SIGNAL-STREAM] wrote-on-cached connection_id={} signal_type={} bytes={}",
connection_id,
signal_type,
frame_buf.len(),
);
return Ok(());
}
}
Err(err) => {
println!(
"[SIGNAL-STREAM] write-failed-on-cached connection_id={} signal_type={} error={}",
connection_id, signal_type, err,
);
drop(send_guard);
self.evict_native_signal_stream_if_owner(
connection_id,
entry.owner,
"cached-write-failed",
)
.await;
}
}
}
}
let (transport_stable_id, mut send, recv) = self
.open_bi_internal_with_transport_stable_id(endpoint_id)
.await
.map_err(|e| format!("open_bi via iroh failed: {}", e))?;
println!(
"[SIGNAL-STREAM] opened-persistent connection_id={} endpoint_id={} signal_type={} stream_id={}",
connection_id,
endpoint_id,
signal_type,
send.id().index(),
);
let label = b"main";
let label_len = (label.len() as u32).to_be_bytes();
let mut opening_buf = Vec::with_capacity(1 + 4 + label.len() + frame_buf.len());
opening_buf.push(0x00);
opening_buf.extend_from_slice(&label_len);
opening_buf.extend_from_slice(label);
opening_buf.extend_from_slice(frame_buf);
send.write_all(&opening_buf)
.await
.map_err(|e| format!("initial write failed: {}", e))?;
send.flush()
.await
.map_err(|e| format!("initial flush failed: {}", e))?;
println!(
"[SIGNAL-STREAM] wrote-on-fresh connection_id={} signal_type={} bytes={}",
connection_id,
signal_type,
opening_buf.len(),
);
if !self
.install_native_control_stream(
connection_id,
endpoint_id,
transport_stable_id,
send,
recv,
"opened",
)
.await
{
return Err(format!(
"fresh native-main stream lost ownership race connection_id={} transport_stable_id={}",
connection_id, transport_stable_id
));
}
Ok(())
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn install_native_control_stream(
&self,
connection_id: &str,
endpoint_id: iroh::EndpointId,
transport_stable_id: u64,
send: iroh::endpoint::SendStream,
recv: iroh::endpoint::RecvStream,
source: &str,
) -> bool {
let stream_id = send.id();
let owner = NativeControlStreamOwner {
transport_stable_id,
stream_rank: native_control_stream_rank(stream_id),
};
let current_transport_stable_id = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64);
if current_transport_stable_id != Some(transport_stable_id) {
println!(
"[OpenRTC][control] stale stream ignored connection_id={} endpoint_id={} transport_stable_id={} current_transport_stable_id={:?} stream_id={} source={}",
connection_id,
endpoint_id,
transport_stable_id,
current_transport_stable_id,
stream_id,
source,
);
return false;
}
let mut candidate_send = Some(send);
let (installed, replaced_entry) = {
let mut streams = self.native_control_streams.lock().await;
let replace = streams
.get(connection_id)
.is_none_or(|existing| should_replace_native_control_stream(existing.owner, owner));
if replace {
let replaced = streams.insert(
connection_id.to_string(),
NativeControlStreamEntry {
owner,
endpoint_id,
send: Arc::new(tokio::sync::Mutex::new(
candidate_send
.take()
.expect("native control candidate must be available"),
)),
},
);
(true, replaced)
} else {
(false, None)
}
};
if !installed {
let delivery_result = retire_losing_native_control_candidate(
candidate_send
.take()
.expect("losing native control candidate must be available"),
recv,
)
.await;
println!(
"[OpenRTC][control] duplicate stream retired connection_id={} endpoint_id={} transport_stable_id={} stream_id={} stream_rank={} source={} verdict_delivery={}",
connection_id,
endpoint_id,
transport_stable_id,
stream_id,
owner.stream_rank,
source,
if delivery_result.is_ok() { "acknowledged" } else { "unconfirmed" },
);
if let Err(error) = delivery_result {
eprintln!(
"[OpenRTC][control] duplicate stream delivery wait ended connection_id={} endpoint_id={} transport_stable_id={} stream_id={} source={} error={}",
connection_id,
endpoint_id,
transport_stable_id,
stream_id,
source,
error,
);
}
return false;
}
let replaced_owner = replaced_entry.as_ref().map(|entry| entry.owner);
println!(
"[OpenRTC][control] stream installed connection_id={} endpoint_id={} transport_stable_id={} stream_id={} stream_rank={} source={} replaced_owner={:?}",
connection_id,
endpoint_id,
transport_stable_id,
stream_id,
owner.stream_rank,
source,
replaced_owner,
);
if let Some(replaced) = replaced_entry {
let connection_id = connection_id.to_string();
tokio::spawn(async move {
let delivery_result = retire_replaced_native_control_send(replaced.send).await;
println!(
"[OpenRTC][control] replaced stream retired connection_id={} endpoint_id={} transport_stable_id={} stream_rank={} verdict_delivery={}",
connection_id,
replaced.endpoint_id,
replaced.owner.transport_stable_id,
replaced.owner.stream_rank,
if delivery_result.is_ok() { "acknowledged" } else { "unconfirmed" },
);
if let Err(error) = delivery_result {
eprintln!(
"[OpenRTC][control] replaced stream delivery wait ended connection_id={} endpoint_id={} transport_stable_id={} stream_rank={} error={}",
connection_id,
replaced.endpoint_id,
replaced.owner.transport_stable_id,
replaced.owner.stream_rank,
error,
);
}
});
}
self.spawn_native_signal_stream_tasks(connection_id, endpoint_id, owner, recv);
true
}
#[cfg(not(target_arch = "wasm32"))]
fn spawn_native_signal_stream_tasks(
&self,
connection_id: &str,
endpoint_id: iroh::EndpointId,
owner: NativeControlStreamOwner,
recv: iroh::endpoint::RecvStream,
) {
let (frame_tx, frame_rx) =
tokio::sync::mpsc::channel::<crate::native_protocol::ParsedMainFrame>(32);
let connection_id_owned = connection_id.to_string();
let endpoint_id_for_log = endpoint_id.to_string();
tokio::spawn(Self::read_signal_stream_loop(
connection_id_owned.clone(),
endpoint_id_for_log.clone(),
recv,
frame_tx,
));
let client = self.clone();
tokio::spawn(async move {
client
.dispatch_signal_stream_frames(
connection_id_owned,
endpoint_id_for_log,
owner,
frame_rx,
)
.await;
});
}
#[cfg(not(target_arch = "wasm32"))]
async fn dispatch_signal_stream_frames(
&self,
connection_id: String,
endpoint_id_for_log: String,
owner: NativeControlStreamOwner,
mut frame_rx: tokio::sync::mpsc::Receiver<crate::native_protocol::ParsedMainFrame>,
) {
use crate::native_protocol::ParsedMainFrame;
while let Some(parsed) = frame_rx.recv().await {
if !self
.native_signal_stream_is_current(&connection_id, owner)
.await
{
println!(
"[SIGNAL-STREAM] stale dispatcher stopped connection_id={} transport_stable_id={} stream_rank={}",
connection_id, owner.transport_stable_id, owner.stream_rank,
);
break;
}
match parsed {
ParsedMainFrame::TypeScriptJson(json) => {
self.maybe_handle_typescript_json_frame(
&connection_id,
Some(&endpoint_id_for_log),
&json,
)
.await;
}
ParsedMainFrame::TypeScriptHandshake(handshake) => {
if let crate::session_token::SessionAdmission::Rejected { reason } =
self.session_admission(&connection_id)
{
println!(
"[SIGNAL-STREAM] handshake-frame ignored connection_id={} endpoint_id={} reason=admission-rejected detail={}",
connection_id, endpoint_id_for_log, reason,
);
continue;
}
println!(
"[SIGNAL-STREAM] handshake-frame dispatched connection_id={} endpoint_id={} action={}",
connection_id,
endpoint_id_for_log,
handshake.action.as_deref().unwrap_or("<none>"),
);
if let Some(device_id) = handshake.claimed_device_id.as_deref() {
let _ = self
.bind_session_admission_authoritative_device_id(
&connection_id,
device_id,
)
.await;
}
self.maybe_handle_typescript_handshake_capabilities(
&connection_id,
Some(&endpoint_id_for_log),
&handshake,
)
.await;
}
ParsedMainFrame::NativeMessage(message) => {
let serialized = match serde_json::to_vec(&message) {
Ok(serialized) => serialized,
Err(error) => {
println!(
"[SIGNAL-STREAM] native-message rejected connection_id={} endpoint_id={} reason=serialize-failed error={}",
connection_id, endpoint_id_for_log, error,
);
continue;
}
};
if let Err(error) = self
.inspect_incoming_native_main_frame_for_transport(
&connection_id,
Some(&endpoint_id_for_log),
None,
Some(owner.transport_stable_id),
&serialized,
)
.await
{
println!(
"[SIGNAL-STREAM] native-message rejected connection_id={} endpoint_id={} channel={} error={}",
connection_id, endpoint_id_for_log, message.channel, error,
);
}
}
ParsedMainFrame::Opaque => {
println!(
"[SIGNAL-STREAM] frame skipped connection_id={} endpoint_id={} reason=unsupported-on-signal-stream kind=opaque",
connection_id, endpoint_id_for_log,
);
}
}
}
println!(
"[SIGNAL-STREAM] dispatcher-exited connection_id={}",
connection_id,
);
#[cfg(feature = "transport-webrtc")]
self.cleanup_native_webrtc_session_after_signal_stream_end(
&connection_id,
owner,
"signal-stream-ended",
)
.await;
#[cfg(not(feature = "transport-webrtc"))]
self.evict_native_signal_stream_if_owner(&connection_id, owner, "control-stream-ended")
.await;
}
#[cfg(not(target_arch = "wasm32"))]
async fn read_native_main_payload_frame(
recv: &mut iroh::endpoint::RecvStream,
frame_len_buf: [u8; 4],
first_payload_byte: Option<u8>,
) -> Result<Option<crate::native_protocol::ParsedMainFrame>, String> {
let frame_len = u32::from_be_bytes(frame_len_buf) as usize;
if frame_len == 0 {
return Ok(None);
}
if frame_len > 4 * 1024 * 1024 {
return Err(format!("frame-too-large: {}", frame_len));
}
let mut frame = Vec::with_capacity(frame_len);
if let Some(first) = first_payload_byte {
frame.push(first);
}
if frame.len() < frame_len {
let remaining = frame_len - frame.len();
let mut rest = vec![0u8; remaining];
recv.read_exact(&mut rest)
.await
.map_err(|error| format!("frame-read-failed: {}", error))?;
frame.extend_from_slice(&rest);
}
Ok(Some(crate::native_protocol::parse_main_frame(&frame)))
}
#[cfg(not(target_arch = "wasm32"))]
async fn read_native_main_json_frame(
recv: &mut iroh::endpoint::RecvStream,
) -> Result<Option<crate::native_protocol::ParsedMainFrame>, String> {
let mut protocol_byte = [0u8; 1];
match recv.read_exact(&mut protocol_byte).await {
Ok(_) => {}
Err(error) => {
return Err(format!("protocol-byte-read-failed: {}", error));
}
}
if protocol_byte[0] != 0x00 {
return Err(format!("unexpected-protocol-byte: {}", protocol_byte[0]));
}
let mut label_len_buf = [0u8; 4];
recv.read_exact(&mut label_len_buf)
.await
.map_err(|error| format!("label-length-read-failed: {}", error))?;
let label_len = u32::from_be_bytes(label_len_buf) as usize;
if label_len == 0 || label_len > 256 {
let frame_len_buf = [
protocol_byte[0],
label_len_buf[0],
label_len_buf[1],
label_len_buf[2],
];
return Self::read_native_main_payload_frame(
recv,
frame_len_buf,
Some(label_len_buf[3]),
)
.await;
}
let mut label_buf = vec![0u8; label_len];
recv.read_exact(&mut label_buf)
.await
.map_err(|error| format!("label-read-failed: {}", error))?;
if String::from_utf8_lossy(&label_buf) != "main" {
return Err(format!(
"unexpected-label: {}",
String::from_utf8_lossy(&label_buf)
));
}
let mut frame_len_buf = [0u8; 4];
recv.read_exact(&mut frame_len_buf)
.await
.map_err(|error| format!("frame-length-read-failed: {}", error))?;
Self::read_native_main_payload_frame(recv, frame_len_buf, None).await
}
#[cfg(not(target_arch = "wasm32"))]
async fn read_signal_stream_loop(
connection_id: String,
endpoint_id_for_log: String,
mut recv: iroh::endpoint::RecvStream,
frame_tx: tokio::sync::mpsc::Sender<crate::native_protocol::ParsedMainFrame>,
) {
println!(
"[SIGNAL-STREAM] reader-spawned connection_id={} endpoint_id={}",
connection_id, endpoint_id_for_log,
);
loop {
let parsed = match Self::read_native_main_json_frame(&mut recv).await {
Ok(Some(parsed)) => parsed,
Ok(None) => {
println!(
"[SIGNAL-STREAM] reader-empty-frame connection_id={} endpoint_id={}",
connection_id, endpoint_id_for_log,
);
continue;
}
Err(err) => {
println!(
"[SIGNAL-STREAM] reader-closed connection_id={} endpoint_id={} reason=native-main-frame-read-failed error={}",
connection_id, endpoint_id_for_log, err,
);
return;
}
};
let frame_type = match &parsed {
crate::native_protocol::ParsedMainFrame::TypeScriptJson(json) => json
.get("type")
.and_then(|v| v.as_str())
.unwrap_or("<none>"),
crate::native_protocol::ParsedMainFrame::TypeScriptHandshake(_) => "handshake",
crate::native_protocol::ParsedMainFrame::NativeMessage(_) => "native-message",
crate::native_protocol::ParsedMainFrame::Opaque => "opaque",
};
println!(
"[SIGNAL-STREAM] reader-frame connection_id={} endpoint_id={} frame_type={}",
connection_id, endpoint_id_for_log, frame_type,
);
if frame_tx.send(parsed).await.is_err() {
println!(
"[SIGNAL-STREAM] reader-closed connection_id={} endpoint_id={} reason=dispatcher-dropped",
connection_id, endpoint_id_for_log,
);
return;
}
}
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn evict_native_signal_stream(&self, connection_id: &str, reason: &str) {
let removed = {
let mut map = self.native_control_streams.lock().await;
map.remove(connection_id)
};
if let Some(entry) = removed {
if self.native_admission_stream_is_persistent(connection_id) {
self.invalidate_native_persistent_route_proofs_for_transport(
connection_id,
entry.owner.transport_stable_id,
);
}
println!(
"[SIGNAL-STREAM] evicted connection_id={} reason={}",
connection_id, reason,
);
}
}
#[cfg(not(target_arch = "wasm32"))]
async fn native_signal_stream_is_current(
&self,
connection_id: &str,
expected_owner: NativeControlStreamOwner,
) -> bool {
self.native_control_streams
.lock()
.await
.get(connection_id)
.is_some_and(|entry| entry.owner == expected_owner)
}
#[cfg(not(target_arch = "wasm32"))]
async fn evict_native_signal_stream_if_owner(
&self,
connection_id: &str,
expected_owner: NativeControlStreamOwner,
reason: &str,
) -> bool {
let removed = {
let mut map = self.native_control_streams.lock().await;
match map.get(connection_id) {
Some(entry) if entry.owner == expected_owner => {
map.remove(connection_id);
true
}
_ => false,
}
};
if removed {
if self.native_admission_stream_is_persistent(connection_id) {
self.invalidate_native_persistent_route_proofs_for_transport(
connection_id,
expected_owner.transport_stable_id,
);
}
println!(
"[SIGNAL-STREAM] evicted-current connection_id={} transport_stable_id={} stream_rank={} reason={}",
connection_id,
expected_owner.transport_stable_id,
expected_owner.stream_rank,
reason,
);
} else {
println!(
"[SIGNAL-STREAM] stale-eviction-ignored connection_id={} transport_stable_id={} stream_rank={} reason={}",
connection_id,
expected_owner.transport_stable_id,
expected_owner.stream_rank,
reason,
);
}
removed
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn retire_stale_native_main_route(
&self,
connection_id: &str,
current_transport_stable_id: u64,
) -> bool {
let mut retired = false;
let stale_owner = self
.native_control_streams
.lock()
.await
.get(connection_id)
.map(|entry| entry.owner)
.filter(|owner| owner.transport_stable_id != current_transport_stable_id);
if let Some(owner) = stale_owner {
retired |= self
.evict_native_signal_stream_if_owner(
connection_id,
owner,
"transport-generation-replaced",
)
.await;
}
retired |= self.invalidate_native_main_route_proofs_except_transport(
connection_id,
current_transport_stable_id,
);
if retired {
self.clear_connection_application_crypto_confirmation(connection_id);
println!(
"[OpenRTC][KEY-AGREEMENT] replacement requires fresh confirmation connection_id={} transport_stable_id={}",
connection_id, current_transport_stable_id,
);
}
retired
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn cleanup_native_webrtc_session_after_signal_stream_end(
&self,
connection_id: &str,
owner: NativeControlStreamOwner,
reason: &str,
) {
if !self
.evict_native_signal_stream_if_owner(connection_id, owner, reason)
.await
{
return;
}
let state = {
let sessions = self.native_webrtc_sessions.read().await;
sessions.get(connection_id).map(|session| session.state())
};
if let Some(state) = state {
println!(
"[SIGNAL-STREAM] session-retained-after-stream-end connection_id={} webrtc_state={:?} reason={}",
connection_id,
state,
reason,
);
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn remove_native_webrtc_session_if_current(
&self,
connection_id: &str,
expected_session: &Arc<crate::transport::NativeWebRTCDataChannel>,
reason: &str,
) -> bool {
let route_instance_id = Arc::as_ptr(expected_session) as usize;
let expected_generation = self
.native_optional_route_generation_for_instance(
connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await;
let removed = {
let mut sessions = self.native_webrtc_sessions.write().await;
match sessions.get(connection_id) {
Some(current) if Arc::ptr_eq(current, expected_session) => {
sessions.remove(connection_id);
true
}
_ => false,
}
};
if removed {
self.clear_native_optional_route_generation_if_current(
connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await;
let demoted = if let Some(expected_generation) = expected_generation {
let fallback_transport = self
.resolve_parallel_iroh_transport_label(connection_id, None)
.await;
self.report_transport_status_for_generation(
connection_id,
&fallback_transport,
None,
expected_generation.transport_stable_id,
expected_generation.transport_generation,
expected_generation.route_generation,
)
.await
.is_some()
} else {
false
};
println!(
"[NativeWebRTC] removed session connection_id={} reason={} fallback_reported={}",
connection_id, reason, demoted,
);
expected_session.close();
}
removed
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn retire_native_webrtc_watcher_session_if_current(
&self,
connection_id: &str,
expected_session: &Arc<crate::transport::NativeWebRTCDataChannel>,
expected_attempt_id: &str,
reason: &str,
) -> bool {
expected_session.close();
self.clear_native_webrtc_attempt_if_current(connection_id, expected_attempt_id)
.await;
let removed = self
.remove_native_webrtc_session_if_current(connection_id, expected_session, reason)
.await;
if removed {
self.finalize_deferred_managed_retirement_if_needed(connection_id)
.await;
}
removed
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn close_connecting_native_webrtc_session(
&self,
connection_id: &str,
reason: &str,
) -> bool {
let session = {
let sessions = self.native_webrtc_sessions.read().await;
sessions.get(connection_id).cloned()
};
let Some(session) = session else {
return false;
};
if session.state() != crate::transport::NativeWebRTCState::Connecting {
return false;
}
if let Some(attempt) = self.active_native_webrtc_attempt(connection_id).await {
if attempt.preserve_on_direct_path && attempt.negotiation_id == session.negotiation_id()
{
println!(
"[NativeWebRTC] keeping explicit connecting session through direct path connection_id={} reason={} negotiation_id={}",
connection_id,
reason,
attempt.negotiation_id,
);
return false;
}
}
println!(
"[NativeWebRTC] closing connecting session connection_id={} reason={}",
connection_id, reason,
);
session.close();
let removed = self
.remove_native_webrtc_session_if_current(connection_id, &session, reason)
.await;
if removed {
self.finalize_deferred_managed_retirement_if_needed(connection_id)
.await;
}
removed
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn dedupe_native_webrtc_session_for_remote(
&self,
connection_id: &str,
remote_node_id: &str,
force_restart: bool,
) -> bool {
let related_records = self.connection_manager.get_by_node_id(remote_node_id).await;
for record in related_records {
if record.connection_id == connection_id {
continue;
}
let session = {
let sessions = self.native_webrtc_sessions.read().await;
sessions.get(&record.connection_id).cloned()
};
let Some(session) = session else {
continue;
};
match session.state() {
crate::transport::NativeWebRTCState::Connected => {
println!(
"[NativeWebRTC] upgrade skipped: remote node already has connected session remote_node_id={} connection_id={} existing_connection_id={}",
remote_node_id,
connection_id,
record.connection_id,
);
return true;
}
crate::transport::NativeWebRTCState::Connecting if !force_restart => {
println!(
"[NativeWebRTC] upgrade skipped: remote node already has connecting session remote_node_id={} connection_id={} existing_connection_id={}",
remote_node_id,
connection_id,
record.connection_id,
);
return true;
}
crate::transport::NativeWebRTCState::Connecting => {
println!(
"[NativeWebRTC] closing duplicate connecting session remote_node_id={} connection_id={} existing_connection_id={}",
remote_node_id,
connection_id,
record.connection_id,
);
session.close();
if self
.remove_native_webrtc_session_if_current(
&record.connection_id,
&session,
"duplicate-connecting-session",
)
.await
{
self.finalize_deferred_managed_retirement_if_needed(&record.connection_id)
.await;
}
}
_ => {}
}
}
false
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn defer_managed_retirement_while_webrtc_active(
&self,
connection_id: &str,
reason: Option<String>,
) -> bool {
if should_immediately_retire_active_webrtc_for_reason(reason.as_deref()) {
self.deferred_managed_retirements
.write()
.await
.remove(connection_id);
println!(
"[pluto-rtc][managed-retire] not deferring cleanup for terminal reason connection_id={} reason={:?}",
connection_id,
reason,
);
return false;
}
let Some(session) = self
.native_webrtc_session_for_connection(connection_id)
.await
else {
self.deferred_managed_retirements
.write()
.await
.remove(connection_id);
return false;
};
let state = session.state();
if !matches!(
state,
crate::transport::NativeWebRTCState::Connecting
| crate::transport::NativeWebRTCState::Connected
) {
self.deferred_managed_retirements
.write()
.await
.remove(connection_id);
return false;
}
let mut deferred = self.deferred_managed_retirements.write().await;
deferred
.entry(connection_id.to_string())
.and_modify(|existing| {
if existing.is_none() {
*existing = reason.clone();
}
})
.or_insert(reason.clone());
println!(
"[pluto-rtc][managed-retire] deferring cleanup until native webrtc session finishes connection_id={} webrtc_state={:?} reason={:?}",
connection_id,
state,
reason,
);
true
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub async fn is_managed_retirement_deferred(&self, connection_id: &str) -> bool {
self.deferred_managed_retirements
.read()
.await
.contains_key(connection_id)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn finalize_deferred_managed_retirement_if_needed(
&self,
connection_id: &str,
) -> bool {
let reason = self
.deferred_managed_retirements
.write()
.await
.remove(connection_id);
let Some(reason) = reason else {
return false;
};
println!(
"[pluto-rtc][managed-retire] finalizing deferred cleanup after native webrtc session ended connection_id={} reason={:?}",
connection_id,
reason,
);
self.retire_managed_connection_now(connection_id, reason)
.await;
true
}
#[cfg(all(
not(target_arch = "wasm32"),
any(feature = "transport-webrtc", feature = "transport-moq")
))]
async fn resolve_parallel_iroh_transport_label(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
) -> String {
if let Some(record) = self
.connection_manager
.get_by_connection_id(connection_id)
.await
{
if let Some(parallel) = record.parallel_transport.as_deref() {
if crate::transport_label::is_iroh_base(parallel) {
return parallel.to_string();
}
}
if crate::transport_label::is_iroh_base(record.active_transport.as_str()) {
return record.active_transport;
}
}
#[cfg(not(target_arch = "wasm32"))]
{
let path_lookup_id = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(connection_id);
return match self.iroh_path_kind(path_lookup_id).await {
kind => kind.transport_label().to_string(),
};
}
#[cfg(target_arch = "wasm32")]
{
let _ = remote_node_id;
"iroh".to_string()
}
}
pub async fn update_transport_config(
&self,
transport_config: TransportConfig,
) -> anyhow::Result<()> {
if let Some(moq) = transport_config.moq.as_ref() {
anyhow::ensure!(
!moq.relay_url.trim().is_empty(),
"transports.moq.relayUrl is required because OpenRTC has no anonymous MoQ relay default"
);
}
#[cfg(not(target_arch = "wasm32"))]
let ble_available = self
.native_custom_transport_kinds
.read()
.await
.values()
.any(|kind| matches!(kind, IrohPathKind::Ble));
#[cfg(target_arch = "wasm32")]
let ble_available = false;
#[cfg(not(target_arch = "wasm32"))]
let was_ble_enabled = ble_available
&& self
.transport_config
.read()
.await
.ble
.as_ref()
.is_some_and(|config| config.enabled);
let transport_config =
transport_config.sanitize_for_runtime_with_ble_available(ble_available);
#[cfg(not(target_arch = "wasm32"))]
let will_enable_ble = transport_config
.ble
.as_ref()
.is_some_and(|config| config.enabled);
#[cfg(not(target_arch = "wasm32"))]
if self.iroh_endpoint.read().await.is_some() {
let current = self.transport_config.read().await;
anyhow::ensure!(
!current.endpoint_rebind_required(&transport_config),
"irohRelayOnly, irohRelayTransportPolicy, and irohLan must be configured before Iroh endpoint initialization"
);
}
let mut guard = self.transport_config.write().await;
*guard = transport_config;
drop(guard);
#[cfg(not(target_arch = "wasm32"))]
if !was_ble_enabled && will_enable_ble {
self.wake_native_ble_recovery("capability-enabled").await;
}
Ok(())
}
pub async fn transport_config(&self) -> TransportConfig {
self.transport_config.read().await.clone()
}
pub async fn is_webrtc_transport_enabled(&self) -> bool {
self.transport_config.read().await.webrtc.is_some()
}
pub async fn is_moq_transport_enabled(&self) -> bool {
self.transport_config.read().await.moq.is_some()
}
#[cfg(not(target_arch = "wasm32"))]
pub async fn is_ble_transport_enabled(&self) -> bool {
self.transport_config
.read()
.await
.ble
.as_ref()
.is_some_and(|config| config.enabled)
&& self
.native_transport_upgrade_available(IrohPathKind::Ble)
.await
}
#[cfg(target_arch = "wasm32")]
pub async fn is_ble_transport_enabled(&self) -> bool {
false
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn peer_has_application_crypto_key(&self, id: &str) -> bool {
self.resolve_transport_connection_ids(id)
.await
.iter()
.any(|connection_id| {
self.application_crypto_key_for_connection(Some(connection_id))
.is_some()
})
}
#[cfg(all(
not(target_arch = "wasm32"),
any(feature = "transport-webrtc", feature = "transport-moq")
))]
pub(crate) async fn resolve_transport_connection_ids(&self, id: &str) -> Vec<String> {
let mut connection_ids = Vec::new();
let exact_id = id.trim();
if !exact_id.is_empty() {
connection_ids.push(exact_id.to_string());
}
connection_ids.extend(self.resolve_peer_connection_ids(id).await);
let records = self.resolve_peer_connection_records(id).await;
connection_ids.extend(records.into_iter().map(|record| record.connection_id));
if connection_ids.is_empty() {
if let Some(record) = self.best_connection_record_for_peer(id).await {
connection_ids.push(record.connection_id);
}
}
if connection_ids.len() > 1 {
let mut unique = std::collections::HashSet::with_capacity(connection_ids.len());
connection_ids.retain(|connection_id| unique.insert(connection_id.clone()));
}
connection_ids
}
pub async fn native_webrtc_state_for_peer(
&self,
id: &str,
) -> Option<(String, crate::transport::NativeWebRTCState)> {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_webrtc_sessions.read().await;
sessions.get(&connection_id).cloned()
};
if let Some(session) = session {
return Some((connection_id, session.state()));
}
}
return None;
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
{
let _ = id;
None
}
}
#[cfg(all(
not(target_arch = "wasm32"),
feature = "transport-webrtc",
feature = "test-harness"
))]
pub async fn force_native_webrtc_session_failure_for_test(&self, id: &str) -> bool {
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_webrtc_sessions.read().await;
sessions.get(&connection_id).cloned()
};
if let Some(session) = session {
if self
.current_native_webrtc_route_identity_matches(&connection_id, &session)
.await
{
let route_instance_id = Arc::as_ptr(&session) as usize;
let retired_route = self
.native_optional_route_generation_for_instance(
&connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await;
let remote_node_id = self
.connection_manager
.get_by_connection_id(&connection_id)
.await
.and_then(|record| record.endpoint_id.or(record.node_id));
let removed = self
.remove_native_webrtc_session_if_current(
&connection_id,
&session,
"forced-test-failure",
)
.await;
if removed {
self.finalize_deferred_managed_retirement_if_needed(&connection_id)
.await;
if let (Some(retired_route), Some(remote_node_id)) =
(retired_route, remote_node_id)
{
self.schedule_native_webrtc_retry_after_route_retirement(
&connection_id,
&remote_node_id,
retired_route,
Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
"forced-test-failure",
)
.await;
}
}
return removed;
}
}
}
false
}
#[cfg(all(
not(target_arch = "wasm32"),
not(feature = "transport-webrtc"),
feature = "test-harness"
))]
pub async fn force_native_webrtc_session_failure_for_test(&self, id: &str) -> bool {
let _ = id;
false
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn report_native_webrtc_candidate_transport(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
) {
let Some(session) = self
.native_webrtc_session_for_connection(connection_id)
.await
else {
return;
};
self.report_native_webrtc_candidate_transport_for_session(
connection_id,
remote_node_id,
&session,
)
.await;
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn report_native_webrtc_candidate_transport_for_session(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
session: &Arc<crate::transport::NativeWebRTCDataChannel>,
) {
if !self
.current_native_webrtc_route_matches(connection_id, session)
.await
{
return;
}
let webrtc_transport = session
.selected_ice_pair_summary()
.await
.map(|pair| pair.transport_label())
.unwrap_or(crate::transport_label::WEBRTC);
let fallback_transport = self
.resolve_parallel_iroh_transport_label(connection_id, remote_node_id)
.await;
if !self
.current_native_webrtc_route_matches(connection_id, session)
.await
{
return;
}
let preserve_proven_active = self
.connection_manager
.get_by_connection_id(connection_id)
.await
.map(|record| crate::transport_label::is_webrtc(record.active_transport.as_str()))
.unwrap_or(false)
&& self
.current_native_webrtc_route_is_proven(connection_id, session)
.await;
if preserve_proven_active {
self.report_current_native_webrtc_route_status(
connection_id,
session,
webrtc_transport,
Some(fallback_transport.as_str()),
false,
)
.await;
} else {
self.report_current_native_webrtc_route_status(
connection_id,
session,
fallback_transport.as_str(),
Some(webrtc_transport),
false,
)
.await;
}
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
pub(crate) async fn report_native_webrtc_candidate_transport(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
) {
let _ = (connection_id, remote_node_id);
}
pub async fn native_moq_state_for_peer(
&self,
id: &str,
) -> Option<(String, crate::transport::NativeMoQState)> {
self.native_moq_state_detail_for_peer(id)
.await
.map(|detail| (detail.0, detail.1))
}
pub async fn native_moq_state_detail_for_peer(
&self,
id: &str,
) -> Option<(String, crate::transport::NativeMoQState, bool)> {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
{
let mut first_seen = None;
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_moq_sessions.read().await;
sessions.get(&connection_id).cloned()
};
if let Some(session) = session {
if !self
.current_native_moq_route_matches(&connection_id, &session)
.await
{
continue;
}
let state = session.state();
let subscribed = session.is_peer_data_subscribed().await;
if !self
.current_native_moq_route_matches(&connection_id, &session)
.await
{
continue;
}
let data_ready = self
.application_crypto_key_for_connection(Some(&connection_id))
.is_some()
&& native_moq_is_data_ready(state, subscribed)
&& self
.current_native_moq_route_is_proven(&connection_id, &session)
.await;
if data_ready {
return Some((connection_id, state, true));
}
if state == crate::transport::NativeMoQState::Connected {
first_seen.get_or_insert((connection_id, state, false));
} else {
first_seen.get_or_insert((connection_id, state, false));
}
}
}
return first_seen;
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
{
let _ = id;
None
}
}
pub async fn native_moq_data_ready_for_peer(&self, id: &str) -> bool {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
{
for connection_id in self.resolve_transport_connection_ids(id).await {
if self
.application_crypto_key_for_connection(Some(&connection_id))
.is_none()
{
continue;
}
let session = {
let sessions = self.native_moq_sessions.read().await;
sessions.get(&connection_id).cloned()
};
if let Some(session) = session {
if !self
.current_native_moq_route_matches(&connection_id, &session)
.await
{
continue;
}
let subscribed = session.is_peer_data_subscribed().await;
if self
.current_native_moq_route_matches(&connection_id, &session)
.await
&& native_moq_is_data_ready(session.state(), subscribed)
&& self
.current_native_moq_route_is_proven(&connection_id, &session)
.await
{
return true;
}
}
}
return false;
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
{
let _ = id;
false
}
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn native_active_independent_route_ready(&self, connection_id: &str) -> bool {
#[cfg(not(any(feature = "transport-webrtc", feature = "transport-moq")))]
{
let _ = connection_id;
return false;
}
#[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
let Some(record) = self
.connection_manager
.get_by_connection_id(connection_id)
.await
else {
return false;
};
#[cfg(feature = "transport-webrtc")]
if crate::transport_label::is_webrtc(record.active_transport.as_str()) {
if self
.application_crypto_key_for_connection(Some(connection_id))
.is_none()
{
return false;
}
let session = self
.native_webrtc_sessions
.read()
.await
.get(connection_id)
.cloned();
return match session {
Some(session)
if session.state() == crate::transport::NativeWebRTCState::Connected =>
{
self.current_native_webrtc_route_is_proven(connection_id, &session)
.await
}
_ => false,
};
}
#[cfg(feature = "transport-moq")]
if crate::transport_label::normalize(record.active_transport.as_str())
== Some(crate::transport_label::MOQ)
{
return self.native_moq_data_ready_for_peer(connection_id).await;
}
false
}
pub async fn try_send_peer_over_webrtc(
&self,
id: &str,
payload: &[u8],
) -> anyhow::Result<bool> {
if payload.is_empty() {
return Ok(true);
}
if !self.is_webrtc_transport_enabled().await {
return Ok(false);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_webrtc_sessions.read().await;
sessions.get(&connection_id).cloned()
};
let Some(session) = session else {
continue;
};
if session.state() != crate::transport::NativeWebRTCState::Connected
|| !self
.current_native_webrtc_route_matches(&connection_id, &session)
.await
{
continue;
}
if !self
.current_native_webrtc_route_is_proven(&connection_id, &session)
.await
{
let _ = self
.request_current_native_webrtc_route_proof(&connection_id, &session)
.await;
continue;
}
if session.send(payload).await.is_ok() {
if !self
.current_native_webrtc_route_matches(&connection_id, &session)
.await
{
continue;
}
return Ok(true);
}
}
return Ok(false);
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
{
let _ = id;
Ok(false)
}
}
pub async fn get_connected_webrtc_session_for_peer(
&self,
id: &str,
) -> Option<Arc<crate::transport::NativeWebRTCDataChannel>> {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_webrtc_sessions.read().await;
sessions.get(&connection_id).cloned()
};
let Some(session) = session else {
continue;
};
if session.state() == crate::transport::NativeWebRTCState::Connected
&& self
.current_native_webrtc_route_matches(&connection_id, &session)
.await
{
return Some(session);
}
}
return None;
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
{
let _ = id;
None
}
}
pub async fn get_webrtc_session_for_peer(
&self,
id: &str,
) -> Option<Arc<crate::transport::NativeWebRTCDataChannel>> {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_webrtc_sessions.read().await;
sessions.get(&connection_id).cloned()
};
if let Some(session) = session {
if native_webrtc_session_is_active(session.state())
&& self
.current_native_webrtc_route_matches(&connection_id, &session)
.await
{
return Some(session);
}
}
}
return None;
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
{
let _ = id;
None
}
}
pub async fn try_send_peer_over_moq(&self, id: &str, payload: &[u8]) -> anyhow::Result<bool> {
if payload.is_empty() {
return Ok(true);
}
if !self.is_moq_transport_enabled().await {
return Ok(false);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
{
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_moq_sessions.read().await;
sessions.get(&connection_id).cloned()
};
let Some(session) = session else {
continue;
};
if session.state() != crate::transport::NativeMoQState::Connected
|| !self
.current_native_moq_route_matches(&connection_id, &session)
.await
{
continue;
}
if !self
.current_native_moq_route_is_proven(&connection_id, &session)
.await
{
let _ = self
.request_current_native_moq_route_proof(&connection_id, &session)
.await;
continue;
}
if !self
.wait_for_stable_native_moq_data_ready(&connection_id, &session)
.await
{
continue;
}
if session.send(payload).await.is_ok() {
if !self
.current_native_moq_route_matches(&connection_id, &session)
.await
{
continue;
}
let parallel_transport = self
.resolve_parallel_iroh_transport_label(&connection_id, None)
.await;
if !self
.current_native_moq_route_matches(&connection_id, &session)
.await
{
continue;
}
let report = successful_send_status_report(parallel_transport.as_str());
if !self
.report_current_native_moq_route_status(
&connection_id,
&session,
report.active_transport,
report.parallel_transport,
false,
)
.await
{
continue;
}
return Ok(true);
}
}
return Ok(false);
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
{
let _ = id;
Ok(false)
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn wait_for_stable_native_moq_data_ready(
&self,
connection_id: &str,
session: &Arc<crate::transport::NativeMoQSession>,
) -> bool {
let deadline = tokio::time::Instant::now()
+ std::time::Duration::from_millis(Self::NATIVE_MOQ_DATA_READY_TIMEOUT_MS);
let poll_delay =
std::time::Duration::from_millis(Self::NATIVE_MOQ_DATA_READY_STABLE_POLL_MS);
let mut ready_samples = 0u32;
loop {
if session.state() != crate::transport::NativeMoQState::Connected
|| !self
.current_native_moq_route_matches(connection_id, session)
.await
{
return false;
}
let subscribed = session.is_peer_data_subscribed().await;
if !self
.current_native_moq_route_matches(connection_id, session)
.await
{
return false;
}
if subscribed {
ready_samples = ready_samples.saturating_add(1);
if ready_samples >= Self::NATIVE_MOQ_DATA_READY_STABLE_SAMPLES {
return true;
}
} else {
ready_samples = 0;
}
let now = tokio::time::Instant::now();
if now >= deadline {
return false;
}
tokio::time::sleep(std::cmp::min(poll_delay, deadline - now)).await;
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn native_moq_has_repairable_session_for_peer(&self, id: &str) -> bool {
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_moq_sessions.read().await;
sessions.get(&connection_id).cloned()
};
let Some(session) = session else {
continue;
};
if native_moq_is_repairable_state(session.state()) {
return true;
}
}
false
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn native_moq_state_summary_for_peer(&self, id: &str) -> String {
let mut parts = Vec::new();
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_moq_sessions.read().await;
sessions.get(&connection_id).cloned()
};
if let Some(session) = session {
let subscribed = session.is_peer_data_subscribed().await;
parts.push(format!(
"connection_id={} state={:?} data_ready={}",
connection_id,
session.state(),
native_moq_is_data_ready(session.state(), subscribed)
));
}
}
if parts.is_empty() {
"no native MoQ session".to_string()
} else {
parts.join("; ")
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub async fn send_peer_over_moq(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
if data.is_empty() {
return Ok(());
}
if !self.peer_has_application_crypto_key(id).await {
anyhow::bail!(
"MoQ peer payloads require application crypto; no key is installed for peer {id}"
);
}
let protected = self.protect_outbound_peer_payload(id, data).await?;
if self.try_send_peer_over_moq(id, &protected).await? {
return Ok(());
}
let deadline = tokio::time::Instant::now()
+ std::time::Duration::from_millis(Self::NATIVE_MOQ_SEND_READY_RETRY_WINDOW_MS);
let mut requested_repair = false;
let mut repair_requests = 0u32;
let mut last_repair_requested_at: Option<tokio::time::Instant> = None;
loop {
let now = tokio::time::Instant::now();
if now >= deadline {
break;
}
let repair_retry_elapsed = last_repair_requested_at
.map(|instant| {
now.duration_since(instant)
>= std::time::Duration::from_millis(
Self::NATIVE_MOQ_SEND_REPAIR_RETRY_MIN_INTERVAL_MS,
)
})
.unwrap_or(true);
let repairable_session = if requested_repair && repair_retry_elapsed {
self.native_moq_has_repairable_session_for_peer(id).await
} else {
false
};
let should_request_repair = (!requested_repair || repairable_session)
&& repair_requests < Self::NATIVE_MOQ_SEND_MAX_REPAIR_REQUESTS
&& repair_retry_elapsed;
if should_request_repair {
repair_requests += 1;
last_repair_requested_at = Some(now);
let repair_started = self
.request_moq_repair(id, None, Some("explicit-moq-send"))
.await?;
requested_repair = requested_repair || repair_started;
}
if !requested_repair {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(
Self::NATIVE_MOQ_SEND_READY_RETRY_DELAY_MS,
))
.await;
if self.try_send_peer_over_moq(id, &protected).await? {
return Ok(());
}
}
let state = self.native_moq_state_summary_for_peer(id).await;
anyhow::bail!("MoQ transport is not data-ready for peer {id}: {state}");
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
pub async fn send_peer_over_moq(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
let _ = (id, data);
anyhow::bail!("MoQ transport is not available in this native build");
}
pub(crate) async fn maybe_handle_typescript_handshake_capabilities(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
handshake: &crate::native_protocol::TypeScriptHandshake,
) {
let remote_supports_webrtc = handshake
.capabilities
.as_ref()
.and_then(|caps| caps.webrtc)
.unwrap_or(false);
let remote_supports_moq = handshake
.capabilities
.as_ref()
.and_then(|caps| caps.moq)
.unwrap_or(false);
let remote_supports_ble = handshake
.capabilities
.as_ref()
.and_then(|caps| caps.ble)
.unwrap_or(false);
#[cfg(not(target_arch = "wasm32"))]
{
let mut capabilities = self.native_peer_transport_capabilities.write().await;
let peer_capabilities = capabilities.entry(connection_id.to_string()).or_default();
if remote_supports_ble {
peer_capabilities.insert(IrohPathKind::Ble);
} else {
peer_capabilities.remove(&IrohPathKind::Ble);
if peer_capabilities.is_empty() {
capabilities.remove(connection_id);
}
}
}
#[cfg(not(target_arch = "wasm32"))]
let application_crypto_was_confirmed =
self.connection_application_crypto_is_confirmed(connection_id);
#[cfg(target_arch = "wasm32")]
let application_crypto_was_confirmed = false;
#[cfg(not(target_arch = "wasm32"))]
let local_application_key_agreement_public_key = self
.maybe_negotiate_typescript_application_crypto(connection_id, remote_node_id, handshake)
.await;
#[cfg(target_arch = "wasm32")]
let local_application_key_agreement_public_key = None;
let local_webrtc_enabled = self.is_webrtc_transport_enabled().await;
let local_moq_enabled = self.is_moq_transport_enabled().await;
let local_ble_enabled = self.is_ble_transport_enabled().await;
#[cfg(not(target_arch = "wasm32"))]
let received_application_key = handshake.application_key_agreement_public_key.is_some();
#[cfg(target_arch = "wasm32")]
let received_application_key = false;
if let Some(reply_action) = typescript_application_key_reply_action(
handshake.action.as_deref(),
received_application_key,
local_application_key_agreement_public_key.is_some(),
application_crypto_was_confirmed,
) {
let public_key = local_application_key_agreement_public_key
.or_else(|| self.connection_key_agreement_public_key(connection_id));
if let Err(error) = self
.send_typescript_handshake_update(
connection_id,
"application-key-confirmation",
public_key,
reply_action,
)
.await
{
println!(
"[OpenRTC][capability] key acknowledgement send failed connection_id={} remote_node_id={:?} error={}",
connection_id, remote_node_id, error,
);
}
}
println!(
"[OpenRTC][transport] TS handshake capabilities connection_id={} remote_node_id={:?} remote_webrtc={} local_webrtc={} remote_moq={} local_moq={} remote_ble={} local_ble={}",
connection_id,
remote_node_id,
remote_supports_webrtc,
local_webrtc_enabled,
remote_supports_moq,
local_moq_enabled,
remote_supports_ble,
local_ble_enabled,
);
let should_send_capability_update = should_send_typescript_capability_update(
local_webrtc_enabled,
local_moq_enabled,
local_ble_enabled,
local_application_key_agreement_public_key.is_some() && !received_application_key,
handshake.action.as_deref(),
);
if should_send_capability_update {
if let Err(error) = self
.send_typescript_capability_update(
connection_id,
if local_webrtc_enabled || local_moq_enabled || local_ble_enabled {
"local-capability"
} else {
"application-key-agreement"
},
local_application_key_agreement_public_key,
)
.await
{
println!(
"[OpenRTC][capability] update send failed connection_id={} remote_node_id={:?} error={}",
connection_id,
remote_node_id,
error,
);
}
} else if handshake.action.as_deref() == Some("capability-update") {
println!(
"[OpenRTC][capability] update received without echo connection_id={} remote_node_id={:?}",
connection_id, remote_node_id,
);
}
if remote_supports_webrtc && local_webrtc_enabled {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
let local_node_id = self.current_node_id().await;
let remote_node_id_str = remote_node_id.map(str::trim).filter(|v| !v.is_empty());
let should_initiate = local_node_id
.as_deref()
.zip(remote_node_id_str)
.map(|(local, remote)| local > remote)
.unwrap_or(false);
if should_initiate {
let _ = self
.request_native_webrtc_recovery(
connection_id,
remote_node_id,
NativeWebRTCRecoveryTrigger::BrowserSignal(
NativeWebRTCBrowserSignalTrigger::TypeScriptHandshake,
),
NativeWebRTCRecoveryOptions {
force_restart: false,
preferred_negotiation_id: None,
role_override: None,
},
)
.await;
} else {
println!(
"[NativeWebRTC] handshake-triggered upgrade skipped (responder) connection_id={} local_node_id={:?} remote_node_id={:?}",
connection_id,
local_node_id,
remote_node_id,
);
}
}
}
if remote_supports_moq && local_moq_enabled {
let _ = self
.maybe_start_native_moq_upgrade(connection_id, remote_node_id)
.await;
}
#[cfg(not(target_arch = "wasm32"))]
if remote_supports_ble && local_ble_enabled {
let _ = self
.maybe_start_native_ble_upgrade(connection_id, remote_node_id)
.await;
}
}
pub async fn handle_typescript_webrtc_signal_frame(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
frame: &serde_json::Value,
) -> bool {
if parse_webrtc_signal_message(frame).is_none() {
return false;
}
self.maybe_handle_typescript_json_frame(connection_id, remote_node_id, frame)
.await;
true
}
#[cfg(not(target_arch = "wasm32"))]
async fn send_native_ble_upgrade_control(
&self,
connection_id: &str,
action: &str,
upgrade_id: &str,
) -> anyhow::Result<()> {
let frame = serde_json::json!({
"type": "transport-upgrade",
"transport": "ble",
"action": action,
"upgradeId": upgrade_id,
});
self.send_typescript_handshake_over_native_main(connection_id, &frame)
.await
}
#[cfg(not(target_arch = "wasm32"))]
async fn wait_for_native_ble_switch_admission(
&self,
connection_id: &str,
endpoint_id: iroh::EndpointId,
expected_generation: crate::client::NativePeerDataGeneration,
) -> anyhow::Result<()> {
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
loop {
let current_transport_stable_id = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64);
if current_transport_stable_id != Some(expected_generation.transport_stable_id) {
anyhow::bail!(
"BLE switch request lost its base transport generation: expected={} current={:?}",
expected_generation.transport_stable_id,
current_transport_stable_id,
);
}
if self
.current_native_peer_data_generation(
connection_id,
Some(expected_generation.transport_stable_id),
)
.await
!= Some(expected_generation)
{
anyhow::bail!(
"BLE switch request lost its logical generation: expected={:?}",
expected_generation,
);
}
let block = self.session_admission_block_reason_for_transport(
connection_id,
Some(expected_generation.transport_stable_id),
);
match classify_native_ble_switch_admission(block.as_ref()) {
NativeBleSwitchAdmission::Ready => return Ok(()),
NativeBleSwitchAdmission::Reject => {
anyhow::bail!(
"BLE switch request admission rejected: {}",
block
.as_ref()
.map(|(_, reason)| reason.as_str())
.unwrap_or("unknown"),
);
}
NativeBleSwitchAdmission::Wait => {}
}
if tokio::time::Instant::now() >= deadline {
anyhow::bail!(
"timed out waiting for BLE switch admission: {}",
block
.as_ref()
.map(|(_, reason)| reason.as_str())
.unwrap_or("unknown"),
);
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
}
#[cfg(not(target_arch = "wasm32"))]
async fn reserve_native_ble_upgrade_attempt(
&self,
connection_id: &str,
generation: crate::client::NativePeerDataGeneration,
) -> Option<u8> {
let mut attempts = self.native_ble_upgrade_attempts.lock().await;
let (state, attempt) =
reserve_native_ble_attempt_state(attempts.get(connection_id).copied(), generation);
attempts.insert(connection_id.to_string(), state);
attempt
}
#[cfg(not(target_arch = "wasm32"))]
async fn current_native_ble_upgrade_attempt(
&self,
connection_id: &str,
generation: crate::client::NativePeerDataGeneration,
) -> Option<u8> {
self.native_ble_upgrade_attempts
.lock()
.await
.get(connection_id)
.filter(|state| state.generation == generation)
.map(|state| state.attempts)
}
#[cfg(not(target_arch = "wasm32"))]
async fn clear_native_ble_upgrade_attempts(
&self,
connection_id: &str,
generation: crate::client::NativePeerDataGeneration,
) {
let mut attempts = self.native_ble_upgrade_attempts.lock().await;
if attempts
.get(connection_id)
.is_some_and(|state| state.generation == generation)
{
attempts.remove(connection_id);
}
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn wake_native_ble_recovery(&self, reason: &'static str) -> usize {
if !self.is_ble_transport_enabled().await {
return 0;
}
let active_gates = self
.native_transport_upgrade_gates
.lock()
.await
.keys()
.filter(|(_, kind)| *kind == IrohPathKind::Ble)
.map(|(connection_id, _)| connection_id.clone())
.collect::<std::collections::HashSet<_>>();
let reset_count = {
let mut attempts = self.native_ble_upgrade_attempts.lock().await;
let before = attempts.len();
attempts.retain(|connection_id, state| {
native_ble_attempt_state_after_recovery_wake(
*state,
active_gates.contains(connection_id),
)
.is_some()
});
before.saturating_sub(attempts.len())
};
let ble_capable_connections = self
.native_peer_transport_capabilities
.read()
.await
.iter()
.filter(|(_, capabilities)| capabilities.contains(&IrohPathKind::Ble))
.map(|(connection_id, _)| connection_id.clone())
.collect::<std::collections::HashSet<_>>();
let mut eligible_records = 0usize;
for record in self.connection_manager.list_all().await {
if !ble_capable_connections.contains(&record.connection_id)
|| !matches!(
record.state,
crate::connection_manager::ConnectionState::Connecting
| crate::connection_manager::ConnectionState::Connected
)
{
continue;
}
let Some(remote_node_id) = record
.endpoint_id
.as_deref()
.or(record.node_id.as_deref())
.map(str::trim)
.filter(|value| !value.is_empty())
else {
continue;
};
eligible_records = eligible_records.saturating_add(1);
if let Err(error) = self
.maybe_start_native_ble_upgrade(&record.connection_id, Some(remote_node_id))
.await
{
eprintln!(
"[OpenRTC][BLE] recovery wake failed connection_id={} remote_node_id={} reason={} error={:#}",
record.connection_id, remote_node_id, reason, error,
);
}
}
println!(
"[OpenRTC][BLE] recovery wake reason={} reset_attempt_budgets={} eligible_records={}",
reason, reset_count, eligible_records,
);
eligible_records
}
#[cfg(not(target_arch = "wasm32"))]
fn schedule_native_ble_upgrade_retry(
&self,
connection_id: String,
remote_node_id: String,
expected_generation: crate::client::NativePeerDataGeneration,
failed_attempt: u8,
reason: &'static str,
) {
let Some(delay) =
native_ble_retry_delay(failed_attempt, uuid::Uuid::new_v4().as_u128() as u64)
else {
println!(
"[OpenRTC][BLE] retry exhausted connection_id={} remote_node_id={} attempts={} reason={}",
connection_id, remote_node_id, failed_attempt, reason,
);
return;
};
let client = self.clone();
tokio::spawn(async move {
tokio::time::sleep(delay).await;
let attempt_is_current = client
.current_native_ble_upgrade_attempt(&connection_id, expected_generation)
.await
== Some(failed_attempt);
let generation_is_current = client
.current_native_peer_data_generation(
&connection_id,
Some(expected_generation.transport_stable_id),
)
.await
== Some(expected_generation);
if !attempt_is_current || !generation_is_current {
return;
}
println!(
"[OpenRTC][BLE] retrying upgrade connection_id={} remote_node_id={} next_attempt={} reason={}",
connection_id,
remote_node_id,
failed_attempt.saturating_add(1),
reason,
);
if let Err(error) = client
.maybe_start_native_ble_upgrade(&connection_id, Some(&remote_node_id))
.await
{
eprintln!(
"[OpenRTC][BLE] retry setup failed connection_id={} remote_node_id={} error={}",
connection_id, remote_node_id, error,
);
}
});
}
#[cfg(not(target_arch = "wasm32"))]
async fn accept_native_ble_switch_request(
&self,
connection_id: &str,
remote_node_id: &str,
upgrade_id: &str,
) -> anyhow::Result<bool> {
let Some(local_node_id) = self.current_node_id().await else {
return Ok(false);
};
let current_path = self.iroh_path_kind(remote_node_id).await;
if !should_accept_native_ble_switch_request(&local_node_id, remote_node_id, current_path) {
return Ok(false);
}
let remote_endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
let provider = self
.native_transport_upgrade_provider(IrohPathKind::Ble)
.await
.ok_or_else(|| anyhow::anyhow!("BLE upgrade provider is unavailable"))?;
let replacement_transport_id = provider.transport_id();
let native_node = self
.iroh_node
.read()
.await
.as_ref()
.cloned()
.ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
{
let mut gates = self.native_transport_upgrade_gates.lock().await;
match gates.get(&gate_key) {
Some(expected) if expected != upgrade_id => return Ok(false),
Some(_) => {}
None => {
gates.insert(gate_key.clone(), upgrade_id.to_string());
}
}
}
let authorization_expires_at = native_node
.authorize_inbound_replacement(
remote_endpoint_id,
replacement_transport_id,
std::time::Duration::from_secs(45),
)
.await;
if let Err(error) = self
.send_native_ble_upgrade_control(connection_id, "switch-ready", upgrade_id)
.await
{
remove_native_transport_upgrade_gate(
&self.native_transport_upgrade_gates,
&gate_key,
upgrade_id,
)
.await;
native_node
.revoke_inbound_replacement_if_current(
remote_endpoint_id,
replacement_transport_id,
authorization_expires_at,
)
.await;
return Err(error);
}
println!(
"[OpenRTC][BLE] switch request accepted connection_id={} remote_node_id={} upgrade_id={}",
connection_id, remote_node_id, upgrade_id,
);
let gates = self.native_transport_upgrade_gates.clone();
let responder_upgrade_id = upgrade_id.to_string();
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(45)).await;
native_node
.revoke_inbound_replacement_if_current(
remote_endpoint_id,
replacement_transport_id,
authorization_expires_at,
)
.await;
remove_native_transport_upgrade_gate(&gates, &gate_key, &responder_upgrade_id).await;
});
Ok(true)
}
#[cfg(not(target_arch = "wasm32"))]
pub(super) async fn maybe_start_native_ble_upgrade(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
) -> anyhow::Result<()> {
let remote_node_id = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.ok_or_else(|| anyhow::anyhow!("BLE upgrade requires a remote node id"))?;
let local_node_id = self
.current_node_id()
.await
.ok_or_else(|| anyhow::anyhow!("BLE upgrade requires an initialized native node"))?;
let current_path = self.iroh_path_kind(remote_node_id).await;
if !should_initiate_native_ble_upgrade(local_node_id.as_str(), remote_node_id, current_path)
{
println!(
"[OpenRTC][BLE] upgrade decision connection_id={} local_node_id={} remote_node_id={} current_path={:?} role=responder-or-better-path",
connection_id, local_node_id, remote_node_id, current_path,
);
return Ok(());
}
println!(
"[OpenRTC][BLE] upgrade decision connection_id={} local_node_id={} remote_node_id={} current_path={:?} role=initiator",
connection_id, local_node_id, remote_node_id, current_path,
);
let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
let transport_stable_id = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64)
.ok_or_else(|| anyhow::anyhow!("BLE upgrade requires a live base transport"))?;
let expected_generation = self
.current_native_peer_data_generation(connection_id, Some(transport_stable_id))
.await
.ok_or_else(|| anyhow::anyhow!("BLE upgrade requires a current peer generation"))?;
let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
let upgrade_id = uuid::Uuid::new_v4().to_string();
{
let mut gates = self.native_transport_upgrade_gates.lock().await;
if gates.contains_key(&gate_key) {
return Ok(());
}
gates.insert(gate_key.clone(), upgrade_id.clone());
}
let Some(attempt) = self
.reserve_native_ble_upgrade_attempt(connection_id, expected_generation)
.await
else {
remove_native_transport_upgrade_gate(
&self.native_transport_upgrade_gates,
&gate_key,
&upgrade_id,
)
.await;
println!(
"[OpenRTC][BLE] upgrade skipped connection_id={} remote_node_id={} reason=attempts-exhausted",
connection_id, remote_node_id,
);
return Ok(());
};
let result = async {
let provider = self
.native_transport_upgrade_provider(IrohPathKind::Ble)
.await
.ok_or_else(|| anyhow::anyhow!("BLE upgrade provider is unavailable"))?;
let _ = provider.prepare_endpoint_addr(endpoint_id).await?;
self.send_native_ble_upgrade_control(connection_id, "switch-request", &upgrade_id)
.await?;
println!(
"[OpenRTC][BLE] prepared upgrade connection_id={} remote_node_id={} upgrade_id={}",
connection_id, remote_node_id, upgrade_id
);
Ok::<(), anyhow::Error>(())
}
.await;
if result.is_err() {
let removed = remove_native_transport_upgrade_gate(
&self.native_transport_upgrade_gates,
&gate_key,
&upgrade_id,
)
.await;
if removed {
self.schedule_native_ble_upgrade_retry(
connection_id.to_string(),
remote_node_id.to_string(),
expected_generation,
attempt,
"setup-failed",
);
}
} else {
let client = self.clone();
let gates = self.native_transport_upgrade_gates.clone();
let connection_id = connection_id.to_string();
let remote_node_id = remote_node_id.to_string();
tokio::spawn(async move {
tokio::time::sleep(NATIVE_BLE_SWITCH_RESPONSE_TIMEOUT).await;
if remove_native_transport_upgrade_gate(&gates, &gate_key, &upgrade_id).await {
client.schedule_native_ble_upgrade_retry(
connection_id,
remote_node_id,
expected_generation,
attempt,
"switch-response-timeout",
);
}
});
}
result
}
#[cfg(not(target_arch = "wasm32"))]
async fn handle_native_ble_upgrade_control(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
frame: &serde_json::Value,
) -> bool {
if frame.get("type").and_then(|value| value.as_str()) != Some("transport-upgrade")
|| frame.get("transport").and_then(|value| value.as_str()) != Some("ble")
{
return false;
}
let Some(remote_node_id) = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
else {
return true;
};
let action = frame
.get("action")
.and_then(|value| value.as_str())
.unwrap_or_default();
let upgrade_id = frame
.get("upgradeId")
.and_then(|value| value.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or("legacy");
if !self.is_ble_transport_enabled().await {
return true;
}
let Some(local_node_id) = self.current_node_id().await else {
return true;
};
let current_path = self.iroh_path_kind(remote_node_id).await;
println!(
"[OpenRTC][BLE] control received connection_id={} remote_node_id={} action={} upgrade_id={} current_path={:?}",
connection_id, remote_node_id, action, upgrade_id, current_path,
);
match action {
"switch-request" => {
if !should_accept_native_ble_switch_request(
&local_node_id,
remote_node_id,
current_path,
) {
return true;
}
let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
return true;
};
let Some(transport_stable_id) = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64)
else {
return true;
};
let Some(expected_generation) = self
.current_native_peer_data_generation(connection_id, Some(transport_stable_id))
.await
else {
return true;
};
let admission_block = self.session_admission_block_reason_for_transport(
connection_id,
Some(transport_stable_id),
);
match classify_native_ble_switch_admission(admission_block.as_ref()) {
NativeBleSwitchAdmission::Reject => {
println!(
"[OpenRTC][BLE] switch request rejected connection_id={} remote_node_id={} upgrade_id={} reason={}",
connection_id,
remote_node_id,
upgrade_id,
admission_block
.as_ref()
.map(|(_, reason)| reason.as_str())
.unwrap_or("unknown"),
);
return true;
}
NativeBleSwitchAdmission::Wait => {
let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
{
let mut gates = self.native_transport_upgrade_gates.lock().await;
match gates.get(&gate_key) {
Some(expected) if expected != upgrade_id => return true,
Some(_) => return true,
None => {
gates.insert(gate_key.clone(), upgrade_id.to_string());
}
}
}
println!(
"[OpenRTC][BLE] switch request deferred connection_id={} remote_node_id={} upgrade_id={} transport_stable_id={} reason={}",
connection_id,
remote_node_id,
upgrade_id,
transport_stable_id,
admission_block
.as_ref()
.map(|(_, reason)| reason.as_str())
.unwrap_or("unknown"),
);
let client = self.clone();
let connection_id = connection_id.to_string();
let remote_node_id = remote_node_id.to_string();
let upgrade_id = upgrade_id.to_string();
tokio::spawn(async move {
let result = client
.wait_for_native_ble_switch_admission(
&connection_id,
endpoint_id,
expected_generation,
)
.await;
if let Err(error) = result {
remove_native_transport_upgrade_gate(
&client.native_transport_upgrade_gates,
&(connection_id.clone(), IrohPathKind::Ble),
&upgrade_id,
)
.await;
eprintln!(
"[OpenRTC][BLE] deferred switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
connection_id, remote_node_id, upgrade_id, error,
);
return;
}
if let Err(error) = client
.accept_native_ble_switch_request(
&connection_id,
&remote_node_id,
&upgrade_id,
)
.await
{
eprintln!(
"[OpenRTC][BLE] deferred switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
connection_id, remote_node_id, upgrade_id, error,
);
}
});
return true;
}
NativeBleSwitchAdmission::Ready => {}
}
if let Err(error) = self
.accept_native_ble_switch_request(connection_id, remote_node_id, upgrade_id)
.await
{
eprintln!(
"[OpenRTC][BLE] switch request failed connection_id={} remote_node_id={} upgrade_id={} error={}",
connection_id, remote_node_id, upgrade_id, error,
);
}
}
"switch-ready" => {
let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
let expected = self
.native_transport_upgrade_gates
.lock()
.await
.get(&gate_key)
.is_some_and(|expected| expected == upgrade_id);
if !expected {
println!(
"[OpenRTC][BLE] switch ready ignored connection_id={} remote_node_id={} upgrade_id={} reason=generation-mismatch",
connection_id, remote_node_id, upgrade_id,
);
return true;
}
println!(
"[OpenRTC][BLE] switch ready accepted connection_id={} remote_node_id={} upgrade_id={}",
connection_id, remote_node_id, upgrade_id,
);
let Ok(endpoint_id) = remote_node_id.parse::<iroh::EndpointId>() else {
return true;
};
let Some(transport_stable_id) = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64)
else {
return true;
};
let Some(expected_generation) = self
.current_native_peer_data_generation(connection_id, Some(transport_stable_id))
.await
else {
return true;
};
let client = self.clone();
let connection_id = connection_id.to_string();
let remote = remote_node_id.to_string();
let upgrade_id = upgrade_id.to_string();
let attempt = self
.current_native_ble_upgrade_attempt(connection_id.as_str(), expected_generation)
.await;
tokio::spawn(async move {
let gate_key = (connection_id.clone(), IrohPathKind::Ble);
if let Err(error) = client
.wait_for_native_ble_switch_admission(
&connection_id,
endpoint_id,
expected_generation,
)
.await
{
let removed = remove_native_transport_upgrade_gate(
&client.native_transport_upgrade_gates,
&gate_key,
&upgrade_id,
)
.await;
if removed {
if let Some(attempt) = attempt {
client.schedule_native_ble_upgrade_retry(
connection_id.clone(),
remote.clone(),
expected_generation,
attempt,
"admission-wait-failed",
);
}
}
eprintln!(
"[OpenRTC][BLE] switch ready admission failed connection_id={} remote_node_id={} upgrade_id={} error={}",
connection_id, remote, upgrade_id, error,
);
return;
}
let result = client
.complete_native_ble_upgrade(
&connection_id,
&remote,
&upgrade_id,
expected_generation,
)
.await;
let removed = remove_native_transport_upgrade_gate(
&client.native_transport_upgrade_gates,
&gate_key,
&upgrade_id,
)
.await;
match result {
Ok(()) => {
client
.clear_native_ble_upgrade_attempts(
&connection_id,
expected_generation,
)
.await;
}
Err(error) => {
if removed {
if let Some(attempt) = attempt {
client.schedule_native_ble_upgrade_retry(
connection_id.clone(),
remote.clone(),
expected_generation,
attempt,
"replacement-failed",
);
}
}
eprintln!(
"[OpenRTC][BLE] upgrade failed connection_id={} remote_node_id={} upgrade_id={} error={}",
connection_id, remote, upgrade_id, error
);
}
}
});
}
_ => {}
}
true
}
#[cfg(not(target_arch = "wasm32"))]
async fn complete_native_ble_upgrade(
&self,
connection_id: &str,
remote_node_id: &str,
upgrade_id: &str,
expected_generation: crate::client::NativePeerDataGeneration,
) -> anyhow::Result<()> {
use futures::StreamExt as _;
let endpoint_id = remote_node_id.parse::<iroh::EndpointId>()?;
let provider = self
.native_transport_upgrade_provider(IrohPathKind::Ble)
.await
.ok_or_else(|| anyhow::anyhow!("BLE upgrade provider is unavailable"))?;
println!(
"[OpenRTC][BLE] replacement connect starting connection_id={} remote_node_id={} upgrade_id={}",
connection_id, remote_node_id, upgrade_id,
);
let endpoint_addr = provider.prepare_endpoint_addr(endpoint_id).await?;
let mut events = {
let node = self.iroh_node.read().await;
let node = node
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Iroh node is unavailable"))?;
node.connect_replacement_addr(endpoint_id, endpoint_addr, provider.transport_id())
};
let replacement_transport_stable_id = tokio::time::timeout(
std::time::Duration::from_secs(30),
async {
while let Some(event) = events.next().await {
match event {
crate::native_node::ReplacementConnectEvent::Connected {
transport_stable_id,
} => {
println!(
"[OpenRTC][BLE] replacement transport connected connection_id={} remote_node_id={} upgrade_id={} transport_stable_id={}",
connection_id, remote_node_id, upgrade_id, transport_stable_id,
);
return Ok(transport_stable_id);
}
crate::native_node::ReplacementConnectEvent::Closed { error } => {
anyhow::bail!(
error.unwrap_or_else(|| "replacement closed".to_string())
);
}
}
}
anyhow::bail!("replacement event stream ended")
},
)
.await
.map_err(|_| anyhow::anyhow!("timed out waiting for BLE replacement"))??;
let current_transport_stable_id = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64);
if current_transport_stable_id != Some(replacement_transport_stable_id) {
anyhow::bail!(
"BLE replacement completion is stale before dial finalization: expected replacement={} current={:?}",
replacement_transport_stable_id,
current_transport_stable_id,
);
}
let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
if !self
.native_transport_upgrade_gates
.lock()
.await
.get(&gate_key)
.is_some_and(|current| current == upgrade_id)
{
anyhow::bail!(
"BLE replacement completion belongs to a retired upgrade: upgrade_id={}",
upgrade_id,
);
}
let committed = self
.connection_manager
.install_transport_replacement_if_generation_current(
connection_id,
expected_generation.transport_stable_id,
expected_generation.transport_generation,
expected_generation.route_generation,
replacement_transport_stable_id,
Some("ble-upgrade".to_string()),
)
.await
.ok_or_else(|| {
anyhow::anyhow!(
"BLE replacement completion is stale before dial finalization: expected_base={:?} replacement={}",
expected_generation,
replacement_transport_stable_id,
)
})?;
self.retire_stale_native_main_route(connection_id, replacement_transport_stable_id)
.await;
let finalized_transport_stable_id = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64);
if finalized_transport_stable_id != Some(replacement_transport_stable_id)
|| committed.transport_stable_id != Some(replacement_transport_stable_id)
|| committed.transport_generation
!= expected_generation.transport_generation.saturating_add(1)
{
anyhow::bail!(
"BLE replacement completion became stale before admission migration: expected_base={:?} current_physical={:?} committed_stable_id={:?} committed_generation={}",
expected_generation,
finalized_transport_stable_id,
committed.transport_stable_id,
committed.transport_generation,
);
}
self.migrate_remote_admission_proof_for_authorized_ble_upgrade(
connection_id,
endpoint_id,
upgrade_id,
)
.await;
let known_device_id = self
.known_device_ids_by_node
.read()
.ok()
.and_then(|known| known.get(remote_node_id).cloned());
if let Some(device_id) = known_device_id {
let _ = self
.mark_trusted_user_device_connection_admitted(connection_id, &device_id)
.await;
}
let path = self.iroh_path_kind(remote_node_id).await;
if path != IrohPathKind::Ble {
anyhow::bail!("replacement connected without selecting BLE: {path:?}");
}
println!(
"[OpenRTC][BLE] upgrade complete connection_id={} remote_node_id={} upgrade_id={}",
connection_id, remote_node_id, upgrade_id
);
Ok(())
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn migrate_remote_admission_proof_for_authorized_ble_upgrade(
&self,
connection_id: &str,
endpoint_id: iroh::EndpointId,
upgrade_id: &str,
) -> bool {
let gate_key = (connection_id.to_string(), IrohPathKind::Ble);
let authorized = self
.native_transport_upgrade_gates
.lock()
.await
.get(&gate_key)
.is_some_and(|expected| expected == upgrade_id);
if !authorized || self.iroh_path_kind(&endpoint_id.to_string()).await != IrohPathKind::Ble {
return false;
}
if !matches!(
self.session_admission(connection_id),
crate::session_token::SessionAdmission::Accepted {
mechanism: crate::session_token::SessionAdmissionMechanism::SessionToken,
..
}
) {
return false;
}
let Some(stable_id) = self
.get_connection(endpoint_id)
.await
.map(|connection| connection.stable_id() as u64)
else {
return false;
};
let migrated = self
.remote_session_admission_proofs
.write()
.ok()
.and_then(|mut proofs| {
let proof = proofs.get_mut(connection_id)?;
let previous_stable_id = proof.transport_stable_id;
proof.transport_stable_id = stable_id;
Some(previous_stable_id)
});
if let Some(previous_stable_id) = migrated {
println!(
"[OpenRTC][BLE][session-admission] migrated remote proof connection_id={} endpoint_id={} upgrade_id={} previous_stable_id={} stable_id={}",
connection_id, endpoint_id, upgrade_id, previous_stable_id, stable_id
);
true
} else {
false
}
}
#[cfg(not(target_arch = "wasm32"))]
pub(crate) async fn migrate_remote_admission_proof_for_pending_ble_upgrade(
&self,
connection_id: &str,
endpoint_id: iroh::EndpointId,
) -> bool {
let upgrade_id = self
.native_transport_upgrade_gates
.lock()
.await
.get(&(connection_id.to_string(), IrohPathKind::Ble))
.cloned();
let Some(upgrade_id) = upgrade_id else {
return false;
};
self.migrate_remote_admission_proof_for_authorized_ble_upgrade(
connection_id,
endpoint_id,
&upgrade_id,
)
.await
}
pub(crate) async fn maybe_handle_typescript_json_frame(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
frame: &serde_json::Value,
) {
#[cfg(not(target_arch = "wasm32"))]
if self
.handle_native_ble_upgrade_control(connection_id, remote_node_id, frame)
.await
{
return;
}
let frame_type = frame
.get("type")
.and_then(|value| value.as_str())
.unwrap_or("<missing>");
let content_transport = frame
.get("content")
.and_then(|value| value.get("transport"))
.and_then(|value| value.as_str())
.unwrap_or("<missing>");
let signal = match parse_webrtc_signal_message(frame) {
Some(value) => value,
None => {
if frame_type == "#pluto-signal" {
println!(
"[NativeWebRTC] inbound signal envelope ignored connection_id={} remote_node_id={:?} reason=unsupported-or-malformed transport={} has_content={} has_sdp={} has_candidate={} negotiation_id={}",
connection_id,
remote_node_id,
content_transport,
frame.get("content").is_some(),
frame
.get("content")
.and_then(|value| value.get("sdp"))
.is_some(),
frame
.get("content")
.and_then(|value| value.get("candidate"))
.is_some(),
frame
.get("content")
.and_then(|value| value.get("negotiationId"))
.and_then(|value| value.as_str())
.unwrap_or("legacy")
);
}
return;
}
};
println!(
"[NativeWebRTC] inbound signal envelope detected connection_id={} remote_node_id={:?} signal_type={} negotiation_id={}",
connection_id,
remote_node_id,
signal.signal_type,
signal.negotiation_id.as_deref().unwrap_or("legacy"),
);
println!(
"[NativeWebRTC] inbound signal dispatch begin connection_id={} remote_node_id={:?} signal_type={} negotiation_id={} has_sdp={} has_candidate={}",
connection_id,
remote_node_id,
signal.signal_type,
signal.negotiation_id.as_deref().unwrap_or("legacy"),
signal.sdp.is_some(),
signal.candidate.is_some(),
);
if !self.is_webrtc_transport_enabled().await {
println!(
"[NativeWebRTC] inbound signal dispatch skipped connection_id={} remote_node_id={:?} reason=webrtc-disabled signal_type={} negotiation_id={}",
connection_id,
remote_node_id,
signal.signal_type,
signal.negotiation_id.as_deref().unwrap_or("legacy"),
);
return;
}
if signal.signal_type == "renegotiate" {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
if let crate::session_token::SessionAdmission::Rejected { reason } =
self.session_admission(connection_id)
{
println!(
"[NativeWebRTC] renegotiate ignored: session admission rejected connection_id={} remote_node_id={:?} reason={}",
connection_id, remote_node_id, reason,
);
return;
}
let active_attempt = self.active_native_webrtc_attempt(connection_id).await;
if let Some(session) = self
.native_webrtc_session_for_connection(connection_id)
.await
{
let existing_state = session.state();
let signal_negotiation_id = signal
.negotiation_id
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty());
let negotiation_mismatch = signal_negotiation_id
.map(|value| value != session.negotiation_id().as_str())
.unwrap_or(false);
if existing_state == crate::transport::NativeWebRTCState::Connected {
println!(
"[NativeWebRTC] renegotiate ignored: connected session owns settled generation connection_id={} remote_node_id={:?} active_negotiation_id={} signal_negotiation_id={} negotiation_mismatch={} has_active_attempt={}",
connection_id,
remote_node_id,
session.negotiation_id(),
signal_negotiation_id.unwrap_or("none"),
negotiation_mismatch,
active_attempt.is_some(),
);
return;
}
if existing_state == crate::transport::NativeWebRTCState::Connecting {
if !negotiation_mismatch {
println!(
"[NativeWebRTC] renegotiate ignored: same negotiation still connecting connection_id={} remote_node_id={:?} active_negotiation_id={}",
connection_id,
remote_node_id,
session.negotiation_id(),
);
return;
}
println!(
"[NativeWebRTC] renegotiate replacing Connecting session with fresh generation connection_id={} remote_node_id={:?}",
connection_id,
remote_node_id,
);
}
}
}
println!(
"[NativeWebRTC] renegotiate requested: restarting session connection_id={} remote_node_id={:?}",
connection_id,
remote_node_id,
);
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
self.reset_native_webrtc_attempt_budget(connection_id, "renegotiate")
.await;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
self.clear_native_webrtc_attempt(connection_id).await;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
if let Some(session) = self
.native_webrtc_session_for_connection(connection_id)
.await
{
session.close();
if self
.remove_native_webrtc_session_if_current(
connection_id,
&session,
"renegotiate-restart",
)
.await
{
self.finalize_deferred_managed_retirement_if_needed(connection_id)
.await;
}
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
if let Err(error) = self
.request_native_webrtc_recovery(
connection_id,
remote_node_id,
NativeWebRTCRecoveryTrigger::BrowserSignal(
NativeWebRTCBrowserSignalTrigger::InboundRenegotiate,
),
NativeWebRTCRecoveryOptions {
force_restart: true,
preferred_negotiation_id: signal.negotiation_id.as_deref(),
role_override: None,
},
)
.await
{
println!(
"[NativeWebRTC] renegotiate: failed to restart session connection_id={} error={}",
connection_id,
error,
);
}
}
return;
}
if signal.signal_type == "sdp" {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
if let Some(existing) = self
.native_webrtc_session_for_connection(connection_id)
.await
{
let signal_negotiation_id = signal
.negotiation_id
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty());
let negotiation_mismatch = signal_negotiation_id
.map(|value| value != existing.negotiation_id().as_str())
.unwrap_or(false);
let is_offer = signal.sdp.as_ref().is_some_and(|value| {
value
.get("type")
.and_then(serde_json::Value::as_str)
.map(|value| value == "offer")
.unwrap_or(false)
});
if negotiation_mismatch
&& is_offer
&& native_webrtc_session_is_active(existing.state())
{
println!(
"[NativeWebRTC] mismatched SDP offer ignored: explicit renegotiate required connection_id={} remote_node_id={:?} active_negotiation_id={} signal_negotiation_id={} state={:?}",
connection_id,
remote_node_id,
existing.negotiation_id(),
signal_negotiation_id.unwrap_or("none"),
existing.state(),
);
return;
}
}
if let Err(error) = self
.request_native_webrtc_recovery(
connection_id,
remote_node_id,
NativeWebRTCRecoveryTrigger::BrowserSignal(
NativeWebRTCBrowserSignalTrigger::InboundSdp,
),
NativeWebRTCRecoveryOptions {
force_restart: false,
preferred_negotiation_id: signal.negotiation_id.as_deref(),
role_override: None,
},
)
.await
{
println!(
"[NativeWebRTC] signal ignored: failed to initialize upgrade session connection_id={} remote_node_id={:?} error={}",
connection_id,
remote_node_id,
error,
);
return;
}
}
} else {
let mut has_session_for_signal = self
.get_webrtc_session_for_peer(connection_id)
.await
.is_some();
if !has_session_for_signal {
if let Some(remote_node_id) = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
{
has_session_for_signal = self
.get_webrtc_session_for_peer(remote_node_id)
.await
.is_some();
}
}
if !has_session_for_signal {
return;
}
}
if let Err(error) = self
.forward_native_webrtc_signal(connection_id, remote_node_id, signal)
.await
{
println!(
"[NativeWebRTC] signal ignored: forward failed connection_id={} remote_node_id={:?} error={}",
connection_id,
remote_node_id,
error,
);
} else {
println!(
"[NativeWebRTC] inbound signal dispatched to session connection_id={} remote_node_id={:?}",
connection_id,
remote_node_id,
);
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub(crate) async fn maybe_start_native_webrtc_upgrade(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
force_restart: bool,
preferred_negotiation_id: Option<&str>,
role_override: Option<crate::transport::NativeWebRTCRole>,
) -> anyhow::Result<()> {
if connection_id.trim().is_empty() {
return Ok(());
}
let remote_node_id = match remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
{
Some(value) => value.to_string(),
None => {
let fallback = self
.connection_manager
.get_by_connection_id(connection_id)
.await
.and_then(|record| record.endpoint_id.or(record.node_id))
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty());
match fallback {
Some(value) => value,
None => {
println!("[NativeWebRTC] upgrade skipped: no remote_node_id for connection_id={}", connection_id);
return Ok(());
}
}
}
};
let webrtc_config = {
let guard = self.transport_config.read().await;
match guard.webrtc.clone() {
Some(value) => value,
None => {
println!("[NativeWebRTC] upgrade skipped: webrtc config is None for connection_id={}", connection_id);
return Ok(());
}
}
};
let local_node_id = match self.current_node_id().await {
Some(value) => value,
None => {
println!(
"[NativeWebRTC] upgrade skipped: no local_node_id for connection_id={}",
connection_id
);
return Ok(());
}
};
let preferred_negotiation_id = preferred_negotiation_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
let _ = (force_restart, preferred_negotiation_id, role_override);
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
if let Some(reason) = self
.native_webrtc_upgrade_fast_path_skip_reason(
connection_id,
Some(&remote_node_id),
force_restart,
preferred_negotiation_id.as_deref(),
)
.await
{
println!(
"[NativeWebRTC] upgrade skipped: fast-path guard connection_id={} remote_node_id={} reason={}",
connection_id,
remote_node_id,
reason,
);
return Ok(());
}
self.cancel_native_webrtc_retry(connection_id).await;
let start_gate = self
.acquire_native_webrtc_start_gate(
connection_id,
&remote_node_id,
force_restart,
preferred_negotiation_id.as_deref(),
)
.await;
let result = async {
if let crate::session_token::SessionAdmission::Rejected { reason } =
self.session_admission(connection_id)
{
self.suppress_native_webrtc_restarts(
connection_id,
Self::NATIVE_WEBRTC_ADMISSION_FAILURE_BACKOFF_MS,
format!("session-admission-rejected: {}", reason),
)
.await;
println!(
"[NativeWebRTC] upgrade skipped: session admission rejected connection_id={} remote_node_id={} reason={}",
connection_id,
remote_node_id,
reason,
);
return Ok(());
}
if self.session_registry_active()
&& matches!(
self.session_admission(connection_id),
crate::session_token::SessionAdmission::Pending
)
{
println!(
"[NativeWebRTC] upgrade deferred: session admission pending connection_id={} remote_node_id={}",
connection_id,
remote_node_id,
);
return Ok(());
}
if crate::native_webrtc_policy::native_webrtc_suppression_blocks_recovery(
preferred_negotiation_id.as_deref(),
) {
if let Some(suppression) =
self.active_native_webrtc_suppression(connection_id).await
{
println!(
"[NativeWebRTC] upgrade skipped: restart suppressed connection_id={} remote_node_id={} reason={} remaining_ms={}",
connection_id,
remote_node_id,
suppression.reason,
suppression.until_ms.saturating_sub(now_millis_i64()),
);
return Ok(());
}
} else if let Some(suppression) =
self.active_native_webrtc_suppression(connection_id).await
{
println!(
"[NativeWebRTC][lifecycle] start continuing through suppression connection_id={} remote_node_id={} reason={} remaining_ms={} preferred_negotiation_id={}",
connection_id,
remote_node_id,
suppression.reason,
suppression.until_ms.saturating_sub(now_millis_i64()),
preferred_negotiation_id.as_deref().unwrap_or("none"),
);
}
let Some(record) = self
.connection_manager
.get_by_connection_id(connection_id)
.await
else {
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=connection-record-retired force_restart={} preferred_negotiation_id={}",
connection_id,
remote_node_id,
force_restart,
preferred_negotiation_id.as_deref().unwrap_or("none"),
);
return Ok(());
};
if !matches!(
record.state,
crate::connection_manager::ConnectionState::Pending
| crate::connection_manager::ConnectionState::Connecting
| crate::connection_manager::ConnectionState::Connected
) {
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=connection-record-terminal state={:?} force_restart={} preferred_negotiation_id={}",
connection_id,
remote_node_id,
record.state,
force_restart,
preferred_negotiation_id.as_deref().unwrap_or("none"),
);
return Ok(());
}
let endpoint_text = record
.endpoint_id
.as_deref()
.or(record.node_id.as_deref())
.unwrap_or(remote_node_id.as_str());
let Ok(endpoint_id) = endpoint_text.parse::<iroh::EndpointId>() else {
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=invalid-control-endpoint endpoint={}",
connection_id, remote_node_id, endpoint_text,
);
return Ok(());
};
let Some(control_connection) = self.get_connection(endpoint_id).await else {
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=missing-live-iroh-control-path endpoint_id={}",
connection_id, remote_node_id, endpoint_id,
);
return Ok(());
};
let control_stable_id = control_connection.stable_id() as u64;
if record
.transport_stable_id
.is_some_and(|expected| expected != control_stable_id)
{
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=stale-iroh-control-generation expected_stable_id={:?} active_stable_id={}",
connection_id,
remote_node_id,
record.transport_stable_id,
control_stable_id,
);
return Ok(());
}
{
let sessions = self.native_webrtc_sessions.read().await;
if let Some(existing) = sessions.get(connection_id) {
let state = existing.state();
let negotiation_mismatch = preferred_negotiation_id
.as_deref()
.map(|value| value != existing.negotiation_id().as_str())
.unwrap_or(false);
let is_stale = state == crate::transport::NativeWebRTCState::Failed
|| state == crate::transport::NativeWebRTCState::Closed
|| (force_restart
&& matches!(
state,
crate::transport::NativeWebRTCState::Connecting
| crate::transport::NativeWebRTCState::Connected
))
|| negotiation_mismatch;
if !is_stale {
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=existing-session-active state={:?} force_restart={} negotiation_id={} preferred_negotiation_id={}",
connection_id,
remote_node_id,
state,
force_restart,
existing.negotiation_id(),
preferred_negotiation_id.as_deref().unwrap_or("none"),
);
return Ok(());
}
println!(
"[NativeWebRTC] replacing stale session connection_id={} prior_state={:?} force_restart={} negotiation_mismatch={}",
connection_id,
state,
force_restart,
negotiation_mismatch,
);
let stale = existing.clone();
drop(sessions);
stale.close();
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
let _ = self
.remove_native_webrtc_session_if_current(
connection_id,
&stale,
"replace-stale-session",
)
.await;
}
}
if self
.dedupe_native_webrtc_session_for_remote(
connection_id,
&remote_node_id,
force_restart,
)
.await
{
return Ok(());
}
if preferred_negotiation_id.is_none() {
let primary_iroh_path = self.iroh_path_kind(&remote_node_id).await;
if !iroh_path_allows_opportunistic_optional_route(primary_iroh_path) {
let primary_label = primary_iroh_path.transport_label();
self.suppress_native_webrtc_restarts(
connection_id,
Self::NATIVE_WEBRTC_DIRECT_QUIC_SUPPRESSION_MS,
primary_label,
)
.await;
let _ = self
.close_connecting_native_webrtc_session(connection_id, primary_label)
.await;
println!(
"[NativeWebRTC] upgrade skipped: optimal iroh path is primary connection_id={} remote_node_id={} path={}",
connection_id,
remote_node_id,
primary_label,
);
return Ok(());
}
}
if preferred_negotiation_id.is_some() {
let removed = self
.native_webrtc_attempt_counts
.write()
.await
.remove(connection_id);
if removed.is_some() {
println!(
"[NativeWebRTC][lifecycle] reset attempt budget for preferred negotiation connection_id={} preferred_negotiation_id={}",
connection_id,
preferred_negotiation_id.as_deref().unwrap_or("none"),
);
}
}
let attempt_reservation = {
let mut counts = self.native_webrtc_attempt_counts.write().await;
let attempt = counts.entry(connection_id.to_string()).or_insert(0);
let reservation = Self::reserve_native_webrtc_attempt_slot(*attempt);
match reservation {
NativeWebRTCAttemptReservation::Start(next) => {
*attempt = next;
}
NativeWebRTCAttemptReservation::JustExhausted(exhausted) => {
*attempt = exhausted;
}
NativeWebRTCAttemptReservation::AlreadyExhausted => {}
}
reservation
};
let current_attempt = match attempt_reservation {
NativeWebRTCAttemptReservation::Start(attempt) => attempt,
NativeWebRTCAttemptReservation::JustExhausted(attempt) => {
self.suppress_native_webrtc_restarts(
connection_id,
Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_BACKOFF_MS,
"upgrade-attempts-exhausted",
)
.await;
println!(
"[NativeWebRTC] upgrade exhausted: max attempts reached connection_id={} remote_node_id={} attempts={}",
connection_id,
remote_node_id,
attempt,
);
return Ok(());
}
NativeWebRTCAttemptReservation::AlreadyExhausted => {
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=attempt-budget-already-exhausted force_restart={} preferred_negotiation_id={}",
connection_id,
remote_node_id,
force_restart,
preferred_negotiation_id.as_deref().unwrap_or("none"),
);
return Ok(());
}
};
println!(
"[NativeWebRTC] starting upgrade attempt={}/{} connection_id={} local_node_id={} remote_node_id={} force_restart={} preferred_negotiation_id={} role_override={:?}",
current_attempt,
Self::NATIVE_WEBRTC_MAX_UPGRADE_ATTEMPTS,
connection_id,
local_node_id,
remote_node_id,
force_restart,
preferred_negotiation_id.as_deref().unwrap_or("none"),
role_override,
);
if self
.should_proactively_notify_browser_peer(connection_id, &remote_node_id)
.await
{
if let Err(error) = self
.send_typescript_capability_update(
connection_id,
"native-upgrade-start",
self.connection_key_agreement_public_key(connection_id),
)
.await
{
println!(
"[NativeWebRTC] proactive capability-update send failed connection_id={} remote_node_id={} reason=native-upgrade-start error={}",
connection_id,
remote_node_id,
error,
);
}
}
let negotiation_id = preferred_negotiation_id
.clone()
.unwrap_or_else(|| Self::create_native_webrtc_negotiation_id(connection_id));
let attempt_id = uuid::Uuid::new_v4().to_string();
let attempt_transport_stable_id = self
.connection_manager
.get_by_connection_id(connection_id)
.await
.and_then(|record| record.transport_stable_id);
let source_session = Arc::new(tokio::sync::OnceCell::<
std::sync::Weak<crate::transport::NativeWebRTCDataChannel>,
>::new());
let signal_sender: crate::transport::NativeWebRTCSignalSender = {
let client = self.clone();
let connection_id = connection_id.to_string();
let remote_node_id = remote_node_id.clone();
let source_session = source_session.clone();
let attempt_id = attempt_id.clone();
Arc::new(move |frame: serde_json::Value| {
let client = client.clone();
let connection_id = connection_id.clone();
let remote_node_id = remote_node_id.clone();
let source_session = source_session.clone();
let attempt_id = attempt_id.clone();
Box::pin(async move {
let signal_type = frame
.get("content")
.and_then(|value| value.get("type"))
.and_then(|value| value.as_str())
.unwrap_or("unknown");
let frame_buf = match frame_typescript_native_main_json(&frame) {
Ok(frame_buf) => frame_buf,
Err(error) => {
println!(
"[NativeWebRTC] signal drop: serialization failed connection_id={} signal_type={} error={}",
connection_id,
signal_type,
error,
);
return;
}
};
let started = std::time::Instant::now();
let retry_delay = std::time::Duration::from_millis(
Self::NATIVE_WEBRTC_SIGNAL_SEND_RETRY_DELAY_MS,
);
let send_timeout = std::time::Duration::from_millis(
Self::NATIVE_WEBRTC_SIGNAL_SEND_TIMEOUT_MS,
);
let mut attempts = 0usize;
let mut endpoint_for_logs: Option<String> = None;
let last_failure = loop {
attempts += 1;
if client
.connection_manager
.get_by_connection_id(&connection_id)
.await
.is_none()
{
println!(
"[NativeWebRTC] signal send aborted: connection retired connection_id={} signal_type={} attempts={}",
connection_id,
signal_type,
attempts,
);
return;
}
if client.session_registry_active() {
if let crate::session_token::SessionAdmission::Rejected { reason } =
client.session_admission(&connection_id)
{
println!(
"[NativeWebRTC] signal send aborted: session admission rejected connection_id={} signal_type={} reason={} attempts={}",
connection_id,
signal_type,
reason,
attempts,
);
return;
}
}
let Some(session) = source_session
.get()
.and_then(std::sync::Weak::upgrade)
else {
println!(
"[NativeWebRTC] signal send aborted: source session missing connection_id={} signal_type={} attempts={}",
connection_id,
signal_type,
attempts,
);
return;
};
if !client
.current_native_webrtc_route_matches(&connection_id, &session)
.await
{
println!(
"[NativeWebRTC] signal send aborted: stale route generation connection_id={} signal_type={} attempts={}",
connection_id, signal_type, attempts,
);
return;
}
match session.state() {
crate::transport::NativeWebRTCState::Failed
| crate::transport::NativeWebRTCState::Closed => {
println!(
"[NativeWebRTC] signal send aborted: session ended connection_id={} signal_type={} state={:?} attempts={}",
connection_id,
signal_type,
session.state(),
attempts,
);
return;
}
_ => {}
}
let endpoint_str = client
.connection_manager
.get_by_connection_id(&connection_id)
.await
.and_then(|record| {
record
.endpoint_id
.as_ref()
.or(record.node_id.as_ref())
.map(|value| value.trim().to_string())
})
.filter(|value| !value.is_empty())
.or_else(|| {
Some(remote_node_id.trim().to_string())
.filter(|value| !value.is_empty())
});
let Some(endpoint_str) = endpoint_str else {
if started.elapsed() >= send_timeout {
break "missing endpoint mapping";
}
tokio::time::sleep(retry_delay).await;
continue;
};
endpoint_for_logs = Some(endpoint_str.clone());
let Ok(endpoint_id) = endpoint_str.parse::<iroh::EndpointId>() else {
println!(
"[NativeWebRTC] signal drop: invalid endpoint_id connection_id={} endpoint={} signal_type={}",
connection_id,
endpoint_str,
signal_type,
);
return;
};
match client
.send_native_main_frame(
&connection_id,
endpoint_id,
&signal_type,
&frame_buf,
)
.await
{
Ok(_) => {
if !client
.current_native_webrtc_route_matches(
&connection_id,
&session,
)
.await
{
println!(
"[NativeWebRTC] signal send completion ignored: route replaced connection_id={} signal_type={} attempts={}",
connection_id, signal_type, attempts,
);
return;
}
if attempts > 1 {
println!(
"[NativeWebRTC] signal send delayed until signaling path recovered connection_id={} endpoint_id={} signal_type={} attempts={} elapsed_ms={}",
connection_id,
endpoint_id,
signal_type,
attempts,
started.elapsed().as_millis(),
);
}
println!(
"[NativeWebRTC] signal sent connection_id={} endpoint_id={} signal_type={}",
connection_id, endpoint_id, signal_type,
);
return;
}
Err(err) => {
if !client
.current_native_webrtc_route_matches(
&connection_id,
&session,
)
.await
{
println!(
"[NativeWebRTC] signal send failure ignored: route replaced connection_id={} signal_type={} attempts={}",
connection_id, signal_type, attempts,
);
return;
}
if native_signal_send_failure_is_terminal_for_attempt(&err) {
let retired_route = client
.native_optional_route_generation_for_instance(
&connection_id,
KnownRoute::WebRtc,
Arc::as_ptr(&session) as usize,
)
.await;
println!(
"[NativeWebRTC] signal send aborted: control path unavailable connection_id={} endpoint_id={} signal_type={} attempts={} error={}",
connection_id,
endpoint_id,
signal_type,
attempts,
err,
);
session.close();
client
.clear_native_webrtc_attempt_if_current(
&connection_id,
&attempt_id,
)
.await;
if client
.remove_native_webrtc_session_if_current(
&connection_id,
&session,
"signal-control-path-unavailable",
)
.await
{
client
.finalize_deferred_managed_retirement_if_needed(
&connection_id,
)
.await;
}
if let Some(retired_route) = retired_route {
client
.schedule_native_webrtc_retry_after_route_retirement(
&connection_id,
&remote_node_id,
retired_route,
Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
"signal-control-path-unavailable",
)
.await;
}
return;
}
if started.elapsed() >= send_timeout {
println!(
"[SIGNAL-STREAM] send failed (terminal) connection_id={} signal_type={} error={}",
connection_id, signal_type, err,
);
break "signal-stream write failed";
}
println!(
"[SIGNAL-STREAM] send failed (retry) connection_id={} signal_type={} attempts={} error={}",
connection_id, signal_type, attempts, err,
);
tokio::time::sleep(retry_delay).await;
continue;
}
}
};
println!(
"[NativeWebRTC] signal send timed out waiting for signaling path connection_id={} endpoint={} signal_type={} attempts={} timeout_ms={} last_failure={}",
connection_id,
endpoint_for_logs.unwrap_or_else(|| "unknown".to_string()),
signal_type,
attempts,
Self::NATIVE_WEBRTC_SIGNAL_SEND_TIMEOUT_MS,
last_failure,
);
})
})
};
self.record_native_webrtc_attempt_started(
connection_id,
&attempt_id,
&negotiation_id,
attempt_transport_stable_id,
preferred_negotiation_id.is_some(),
)
.await;
let session = Arc::new(
crate::transport::NativeWebRTCDataChannel::new_with_role(
&local_node_id,
&remote_node_id,
&webrtc_config,
signal_sender,
&negotiation_id,
role_override.unwrap_or(crate::transport::NativeWebRTCRole::Auto),
)
.await?,
);
source_session
.set(Arc::downgrade(&session))
.map_err(|_| anyhow::anyhow!("native WebRTC source session already installed"))?;
{
let client_ref = self.clone();
let connection_id_ref = connection_id.to_string();
let remote_node_id_ref = remote_node_id.clone();
let source_session = Arc::downgrade(&session);
session
.set_message_handler(Arc::new(move |bytes| {
let client_ref = client_ref.clone();
let connection_id_ref = connection_id_ref.clone();
let remote_node_id_ref = remote_node_id_ref.clone();
let source_session = source_session.clone();
Box::pin(async move {
let Some(source_session) = source_session.upgrade() else {
return;
};
if let Err(error) = client_ref
.handle_current_native_webrtc_peer_data(
&connection_id_ref,
Some(&remote_node_id_ref),
&source_session,
bytes,
)
.await
{
println!(
"[NativeWebRTC] peer-data receive failed connection_id={} remote_node_id={} error={}",
connection_id_ref, remote_node_id_ref, error,
);
}
})
}))
.await;
}
let (heartbeat_health_tx, _heartbeat_health_rx) =
tokio::sync::mpsc::channel::<crate::heartbeat::HealthTransition>(8);
session
.with_heartbeat(
"webrtc".to_string(),
Self::native_webrtc_heartbeat_config(),
heartbeat_health_tx,
)
.await;
self.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), session.clone());
if self
.install_native_optional_route_generation(
connection_id,
KnownRoute::WebRtc,
Arc::as_ptr(&session) as usize,
)
.await
.is_none()
{
self.native_webrtc_sessions.write().await.remove(connection_id);
self.clear_native_webrtc_attempt_if_current(connection_id, &attempt_id)
.await;
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=missing-current-route-generation",
connection_id, remote_node_id,
);
return Ok(());
}
if self
.install_native_webrtc_route_proof(connection_id, &session)
.await
.is_none()
{
let _ = self
.remove_native_webrtc_session_if_current(
connection_id,
&session,
"missing-route-proof-fence",
)
.await;
self.clear_native_webrtc_attempt_if_current(connection_id, &attempt_id)
.await;
println!(
"[NativeWebRTC][lifecycle] start skipped connection_id={} remote_node_id={} reason=missing-route-proof-fence",
connection_id, remote_node_id,
);
return Ok(());
}
if !self
.current_native_webrtc_route_identity_matches(connection_id, &session)
.await
{
let _ = self
.remove_native_webrtc_session_if_current(
connection_id,
&session,
"stale-before-start",
)
.await;
return Ok(());
}
if let Err(error) = session.start().await {
self.clear_native_webrtc_attempt_if_current(connection_id, &attempt_id)
.await;
if self
.remove_native_webrtc_session_if_current(
connection_id,
&session,
"start-failed",
)
.await
{
self.finalize_deferred_managed_retirement_if_needed(connection_id)
.await;
}
return Err(error);
}
if !self
.current_native_webrtc_route_matches(connection_id, &session)
.await
{
let _ = self
.remove_native_webrtc_session_if_current(
connection_id,
&session,
"stale-after-start",
)
.await;
return Ok(());
}
println!(
"[NativeWebRTC] upgrade started connection_id={} state={:?}",
connection_id,
session.state(),
);
{
let session_ref = session.clone();
let connection_id_ref = connection_id.to_string();
let remote_node_id_ref = remote_node_id.clone();
let negotiation_id_ref = negotiation_id.clone();
let attempt_id_ref = attempt_id.clone();
let client_ref = self.clone();
tokio::spawn(async move {
let no_signal_deadline_ticks =
(Self::NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS / 250) as usize;
let mut connected = false;
let mut aborted_no_signal = false;
for tick in 0..(Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_MS / 250) as usize {
if !client_ref
.current_native_webrtc_route_matches(
&connection_id_ref,
&session_ref,
)
.await
{
client_ref
.retire_native_webrtc_watcher_session_if_current(
&connection_id_ref,
&session_ref,
&attempt_id_ref,
"watcher-generation-replaced",
)
.await;
return;
}
match session_ref.state() {
crate::transport::NativeWebRTCState::Connected => {
connected = true;
break;
}
crate::transport::NativeWebRTCState::Failed
| crate::transport::NativeWebRTCState::Closed => {
client_ref
.clear_native_webrtc_attempt_if_current(
&connection_id_ref,
&attempt_id_ref,
)
.await;
return;
}
_ => {}
}
let missing_remote_description = tick == no_signal_deadline_ticks
&& !session_ref.has_remote_description().await;
if !client_ref
.current_native_webrtc_route_matches(
&connection_id_ref,
&session_ref,
)
.await
{
client_ref
.retire_native_webrtc_watcher_session_if_current(
&connection_id_ref,
&session_ref,
&attempt_id_ref,
"watcher-generation-replaced-before-timeout",
)
.await;
return;
}
if missing_remote_description {
println!(
"[NativeWebRTC] upgrade aborted: no remote SDP after {}ms connection_id={} negotiation_id={}",
Self::NATIVE_WEBRTC_NO_SIGNAL_ABORT_MS,
connection_id_ref,
negotiation_id_ref,
);
aborted_no_signal = true;
break;
}
tokio::time::sleep(tokio::time::Duration::from_millis(250)).await;
}
if aborted_no_signal {
let retired_route = client_ref
.native_optional_route_generation_for_instance(
&connection_id_ref,
KnownRoute::WebRtc,
Arc::as_ptr(&session_ref) as usize,
)
.await;
session_ref.close();
client_ref
.clear_native_webrtc_attempt_if_current(
&connection_id_ref,
&attempt_id_ref,
)
.await;
if client_ref
.remove_native_webrtc_session_if_current(
&connection_id_ref,
&session_ref,
"upgrade-no-signal",
)
.await
{
client_ref
.finalize_deferred_managed_retirement_if_needed(
&connection_id_ref,
)
.await;
}
if let Some(retired_route) = retired_route {
client_ref
.schedule_native_webrtc_retry_after_route_retirement(
&connection_id_ref,
&remote_node_id_ref,
retired_route,
Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
"upgrade-no-signal",
)
.await;
}
return;
}
if !connected {
let retired_route = client_ref
.native_optional_route_generation_for_instance(
&connection_id_ref,
KnownRoute::WebRtc,
Arc::as_ptr(&session_ref) as usize,
)
.await;
client_ref
.clear_native_webrtc_attempt_if_current(
&connection_id_ref,
&attempt_id_ref,
)
.await;
client_ref
.suppress_native_webrtc_restarts(
&connection_id_ref,
Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_BACKOFF_MS,
"upgrade-timeout",
)
.await;
println!(
"[NativeWebRTC] upgrade timeout: closing stuck Connecting session connection_id={}",
connection_id_ref,
);
session_ref.close();
if client_ref
.remove_native_webrtc_session_if_current(
&connection_id_ref,
&session_ref,
"upgrade-timeout",
)
.await
{
client_ref
.finalize_deferred_managed_retirement_if_needed(
&connection_id_ref,
)
.await;
}
if let Some(retired_route) = retired_route {
client_ref
.schedule_native_webrtc_retry_after_route_retirement(
&connection_id_ref,
&remote_node_id_ref,
retired_route,
Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
"upgrade-timeout",
)
.await;
}
return;
}
let _ = client_ref
.clear_native_webrtc_suppression(&connection_id_ref, None)
.await;
client_ref
.native_webrtc_attempt_counts
.write()
.await
.remove(&connection_id_ref);
client_ref
.clear_native_webrtc_attempt_if_current(
&connection_id_ref,
&attempt_id_ref,
)
.await;
let _ = client_ref
.request_current_native_webrtc_route_proof(
&connection_id_ref,
&session_ref,
)
.await;
client_ref
.report_native_webrtc_candidate_transport_for_session(
&connection_id_ref,
Some(remote_node_id_ref.as_str()),
&session_ref,
)
.await;
loop {
if session_ref.state()
!= crate::transport::NativeWebRTCState::Connected
|| !client_ref
.current_native_webrtc_route_matches(
&connection_id_ref,
&session_ref,
)
.await
{
break;
}
if !client_ref.is_app_backgrounded() {
client_ref
.report_native_webrtc_candidate_transport_for_session(
&connection_id_ref,
None,
&session_ref,
)
.await;
}
let interval = if client_ref.is_app_backgrounded() {
tokio::time::Duration::from_secs(30)
} else {
tokio::time::Duration::from_secs(5)
};
tokio::time::sleep(interval).await;
}
let terminal_failure = matches!(
session_ref.state(),
crate::transport::NativeWebRTCState::Failed
| crate::transport::NativeWebRTCState::Closed
);
let retired_route = if terminal_failure {
client_ref
.native_optional_route_generation_for_instance(
&connection_id_ref,
KnownRoute::WebRtc,
Arc::as_ptr(&session_ref) as usize,
)
.await
} else {
None
};
if client_ref
.remove_native_webrtc_session_if_current(
&connection_id_ref,
&session_ref,
"session-ended",
)
.await
{
client_ref
.finalize_deferred_managed_retirement_if_needed(&connection_id_ref)
.await;
if let Some(retired_route) = retired_route {
client_ref
.schedule_native_webrtc_retry_after_route_retirement(
&connection_id_ref,
&remote_node_id_ref,
retired_route,
Self::NATIVE_WEBRTC_CONNECT_TIMEOUT_RETRY_DELAY_MS,
"session-ended",
)
.await;
}
}
});
}
Ok(())
}
.await;
self.release_native_webrtc_start_gate(connection_id, start_gate)
.await;
return result;
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
{
let _ = (local_node_id, remote_node_id, webrtc_config);
Ok(())
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) async fn maybe_start_native_moq_upgrade(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
) -> anyhow::Result<bool> {
self.maybe_start_native_moq_upgrade_with_options(
connection_id,
remote_node_id,
NativeMoqUpgradeOptions::OPPORTUNISTIC,
)
.await
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
pub(crate) async fn maybe_start_native_moq_upgrade(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
) -> anyhow::Result<bool> {
let _ = (connection_id, remote_node_id);
Ok(false)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn maybe_start_native_moq_upgrade_with_options(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
options: NativeMoqUpgradeOptions,
) -> anyhow::Result<bool> {
let Some((start_gate, start_guard)) = self
.native_route_start_gates
.acquire(connection_id, KnownRoute::Moq)
.await
else {
return Ok(false);
};
let result = self
.maybe_start_native_moq_upgrade_with_options_unlocked(
connection_id,
remote_node_id,
options,
)
.await;
self.native_route_start_gates
.release(connection_id, KnownRoute::Moq, start_gate, start_guard)
.await;
result
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
async fn maybe_start_native_moq_upgrade_with_options_unlocked(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
options: NativeMoqUpgradeOptions,
) -> anyhow::Result<bool> {
if connection_id.trim().is_empty() {
return Ok(false);
}
let remote_node_id = match remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
{
Some(value) => value,
None => return Ok(false),
};
let moq_config = {
let guard = self.transport_config.read().await;
match guard.moq.clone() {
Some(value) => value,
None => return Ok(false),
}
};
let local_node_id = match self.current_node_id().await {
Some(value) => value,
None => return Ok(false),
};
if !options.allow_on_optimal_iroh_path {
let primary_iroh_path = self.iroh_path_kind(remote_node_id).await;
if !iroh_path_allows_opportunistic_optional_route(primary_iroh_path) {
println!(
"[NativeMoQ] opportunistic upgrade skipped: optimal iroh path is primary connection_id={} remote_node_id={} path={}",
connection_id,
remote_node_id,
primary_iroh_path.transport_label(),
);
return Ok(false);
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
{
enum ExistingMoQAction {
Close(Arc<crate::transport::NativeMoQSession>),
Reannounce(Arc<crate::transport::NativeMoQSession>),
Reuse(Arc<crate::transport::NativeMoQSession>),
}
let session_to_close = {
let existing = {
let sessions = self.native_moq_sessions.read().await;
sessions.get(connection_id).cloned()
};
if let Some(existing) = existing {
let state = existing.state();
if !self
.current_native_moq_route_matches(connection_id, &existing)
.await
{
Some(ExistingMoQAction::Close(existing.clone()))
} else {
let subscribed = existing.is_peer_data_subscribed().await;
if !self
.current_native_moq_route_matches(connection_id, &existing)
.await
{
Some(ExistingMoQAction::Close(existing.clone()))
} else {
let data_ready = native_moq_is_data_ready(state, subscribed);
let force_replace = options.force_restart_when_not_ready && !data_ready;
if force_replace || native_moq_should_close_before_restart(state) {
Some(ExistingMoQAction::Close(existing.clone()))
} else if state == crate::transport::NativeMoQState::Connected
&& !data_ready
{
Some(ExistingMoQAction::Reannounce(existing.clone()))
} else if native_moq_should_reuse_existing_setup(state) {
Some(ExistingMoQAction::Reuse(existing.clone()))
} else {
None
}
}
}
} else {
None
}
};
if let Some(action) = session_to_close {
match action {
ExistingMoQAction::Close(existing) => {
existing.close_gracefully().await;
let mut sessions = self.native_moq_sessions.write().await;
if sessions
.get(connection_id)
.map(|current| Arc::ptr_eq(current, &existing))
.unwrap_or(false)
{
sessions.remove(connection_id);
}
drop(sessions);
self.clear_native_optional_route_generation_if_current(
connection_id,
KnownRoute::Moq,
Arc::as_ptr(&existing) as usize,
)
.await;
}
ExistingMoQAction::Reannounce(existing) => {
if !self
.current_native_moq_route_matches(connection_id, &existing)
.await
{
return Ok(false);
}
let parallel_transport = self
.resolve_parallel_iroh_transport_label(connection_id, None)
.await;
if !self
.current_native_moq_route_matches(connection_id, &existing)
.await
{
return Ok(false);
}
let report = setup_status_report(parallel_transport.as_str());
if !self
.report_current_native_moq_route_status(
connection_id,
&existing,
report.active_transport,
report.parallel_transport,
false,
)
.await
{
return Ok(false);
}
if let Err(error) = self
.send_typescript_capability_update(
connection_id,
"native-moq-ready",
self.connection_key_agreement_public_key(connection_id),
)
.await
{
println!(
"[NativeMoQ] capability-update resend failed connection_id={} remote_node_id={} reason=native-moq-ready error={}",
connection_id, remote_node_id, error,
);
}
let _ = self
.request_current_native_moq_route_proof(connection_id, &existing)
.await;
return Ok(true);
}
ExistingMoQAction::Reuse(existing) => {
let _ = self
.request_current_native_moq_route_proof(connection_id, &existing)
.await;
return Ok(true);
}
}
}
let session = Arc::new(
crate::transport::NativeMoQSession::new(
&local_node_id,
remote_node_id,
&moq_config,
)
.await?,
);
{
let client_ref = self.clone();
let connection_id_ref = connection_id.to_string();
let remote_node_id_ref = remote_node_id.to_string();
let source_session = Arc::downgrade(&session);
session
.set_message_handler(Arc::new(move |bytes| {
let client_ref = client_ref.clone();
let connection_id_ref = connection_id_ref.clone();
let remote_node_id_ref = remote_node_id_ref.clone();
let source_session = source_session.clone();
Box::pin(async move {
let Some(source_session) = source_session.upgrade() else {
return;
};
if let Err(error) = client_ref
.handle_current_native_moq_peer_data(
&connection_id_ref,
Some(&remote_node_id_ref),
&source_session,
bytes,
)
.await
{
println!(
"[NativeMoQ] peer-data receive failed connection_id={} remote_node_id={} error={}",
connection_id_ref, remote_node_id_ref, error,
);
}
})
}))
.await;
}
{
let client_ref = self.clone();
let connection_id_ref = connection_id.to_string();
let remote_node_id_ref = remote_node_id.to_string();
let source_session = Arc::downgrade(&session);
session
.set_terminal_handler(Arc::new(move || {
let client_ref = client_ref.clone();
let connection_id_ref = connection_id_ref.clone();
let remote_node_id_ref = remote_node_id_ref.clone();
let source_session = source_session.clone();
let runtime = tokio::runtime::Handle::current();
tokio::task::spawn_blocking(move || {
runtime.block_on(async move {
let Some(source_session) = source_session.upgrade() else {
return;
};
let _ = client_ref
.handle_current_native_moq_terminal_failure(
&connection_id_ref,
&remote_node_id_ref,
source_session,
)
.await;
});
});
Box::pin(async {})
}))
.await;
}
self.native_moq_sessions
.write()
.await
.insert(connection_id.to_string(), session.clone());
if self
.install_native_optional_route_generation(
connection_id,
KnownRoute::Moq,
Arc::as_ptr(&session) as usize,
)
.await
.is_none()
{
self.native_moq_sessions.write().await.remove(connection_id);
println!(
"[NativeMoQ][lifecycle] setup skipped connection_id={} remote_node_id={} reason=missing-current-route-generation",
connection_id, remote_node_id,
);
return Ok(false);
}
self.install_native_moq_route_proof(
connection_id,
&session,
&local_node_id,
remote_node_id,
)
.await;
if !self
.current_native_moq_route_identity_matches(connection_id, &session)
.await
{
session.close_gracefully().await;
return Ok(false);
}
if let Err(error) = session.start().await {
println!(
"[NativeMoQ] setup failed connection_id={} local_node_id={} remote_node_id={} error={:#}",
connection_id, local_node_id, remote_node_id, error,
);
let mut sessions = self.native_moq_sessions.write().await;
if sessions
.get(connection_id)
.is_some_and(|current| Arc::ptr_eq(current, &session))
{
sessions.remove(connection_id);
}
drop(sessions);
self.clear_native_optional_route_generation_if_current(
connection_id,
KnownRoute::Moq,
Arc::as_ptr(&session) as usize,
)
.await;
return Ok(true);
}
if !self
.current_native_moq_route_matches(connection_id, &session)
.await
{
session.close_gracefully().await;
return Ok(false);
}
let parallel_transport = self
.resolve_parallel_iroh_transport_label(connection_id, None)
.await;
if !self
.current_native_moq_route_matches(connection_id, &session)
.await
{
return Ok(false);
}
let report = setup_status_report(parallel_transport.as_str());
if !self
.report_current_native_moq_route_status(
connection_id,
&session,
report.active_transport,
report.parallel_transport,
false,
)
.await
{
return Ok(false);
}
if let Err(error) = self
.send_typescript_capability_update(
connection_id,
"native-moq-ready",
self.connection_key_agreement_public_key(connection_id),
)
.await
{
println!(
"[NativeMoQ] capability-update send failed connection_id={} remote_node_id={} reason=native-moq-ready error={}",
connection_id, remote_node_id, error,
);
}
let _ = self
.request_current_native_moq_route_proof(connection_id, &session)
.await;
if !self
.current_native_moq_route_matches(connection_id, &session)
.await
{
return Ok(false);
}
println!(
"[NativeMoQ] setup connected connection_id={} local_node_id={} remote_node_id={} active={} parallel=moq",
connection_id, local_node_id, remote_node_id, parallel_transport,
);
return Ok(true);
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
{
let _ = (local_node_id, remote_node_id, moq_config);
Ok(false)
}
}
async fn forward_native_webrtc_signal(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
signal: crate::transport::WebRTCSignalMessage,
) -> anyhow::Result<()> {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
let mut candidate_connection_ids = vec![connection_id.to_string()];
if let Some(remote_node_id) = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
{
candidate_connection_ids
.extend(self.resolve_transport_connection_ids(remote_node_id).await);
}
if candidate_connection_ids.len() > 1 {
let mut seen =
std::collections::HashSet::with_capacity(candidate_connection_ids.len());
candidate_connection_ids.retain(|candidate| seen.insert(candidate.clone()));
}
let resolved_session = {
let sessions = self.native_webrtc_sessions.read().await;
let mut resolved: Option<(String, Arc<crate::transport::NativeWebRTCDataChannel>)> =
None;
for candidate_connection_id in candidate_connection_ids.iter() {
if let Some(session) = sessions.get(candidate_connection_id) {
resolved = Some((candidate_connection_id.clone(), session.clone()));
break;
}
}
resolved
};
if let Some((resolved_connection_id, session)) = resolved_session {
if resolved_connection_id != connection_id {
println!(
"[NativeWebRTC] signal routing alias resolved incoming_connection_id={} resolved_connection_id={}",
connection_id,
resolved_connection_id,
);
}
let initial_state = session.state();
if !self
.current_native_webrtc_route_matches(&resolved_connection_id, &session)
.await
{
anyhow::bail!(
"native WebRTC signal rejected for retired session connection_id={} state={:?}",
resolved_connection_id,
initial_state,
);
}
let was_connected =
session.state() == crate::transport::NativeWebRTCState::Connected;
let signal_type = signal.signal_type.clone();
let result = session.handle_signal(signal).await;
if !self
.current_native_webrtc_route_matches(&resolved_connection_id, &session)
.await
{
println!(
"[NativeWebRTC] signal completion ignored: route replaced connection_id={} signal_type={} negotiation_id={}",
resolved_connection_id,
signal_type,
session.negotiation_id(),
);
return Ok(());
}
result?;
println!(
"[NativeWebRTC] signal received connection_id={} signal_type={} state={:?}",
resolved_connection_id,
signal_type,
session.state(),
);
if !was_connected
&& session.state() == crate::transport::NativeWebRTCState::Connected
{
println!(
"[NativeWebRTC] state=Connected connection_id={} route_state=candidate",
resolved_connection_id,
);
self.report_native_webrtc_candidate_transport_for_session(
&resolved_connection_id,
remote_node_id,
&session,
)
.await;
}
} else {
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
{
let should_bootstrap = inbound_signal_can_bootstrap_responder(
signal.signal_type.as_str(),
signal
.sdp
.as_ref()
.and_then(|sdp| sdp.get("type").and_then(|value| value.as_str())),
);
if should_bootstrap {
let role_override = if signal.signal_type == "sdp"
&& signal.sdp.as_ref().is_some_and(|sdp| {
sdp.get("type")
.and_then(|value| value.as_str())
.map(|value| value == "offer")
.unwrap_or(false)
}) {
Some(crate::transport::NativeWebRTCRole::Responder)
} else {
None
};
let _ = self
.maybe_start_native_webrtc_upgrade(
connection_id,
remote_node_id,
false,
signal.negotiation_id.as_deref(),
role_override,
)
.await;
let sessions = self.native_webrtc_sessions.read().await;
let session = sessions.get(connection_id).cloned();
drop(sessions);
if let Some(session) = session {
if !self
.current_native_webrtc_route_matches(connection_id, &session)
.await
{
return Ok(());
}
let was_connected =
session.state() == crate::transport::NativeWebRTCState::Connected;
let signal_type = signal.signal_type.clone();
let result = session.handle_signal(signal).await;
if !self
.current_native_webrtc_route_matches(connection_id, &session)
.await
{
return Ok(());
}
result?;
println!(
"[NativeWebRTC] signal received connection_id={} signal_type={} state={:?}",
connection_id,
signal_type,
session.state(),
);
if !was_connected
&& session.state() == crate::transport::NativeWebRTCState::Connected
{
println!(
"[NativeWebRTC] state=Connected connection_id={} route_state=candidate",
connection_id,
);
self.report_native_webrtc_candidate_transport_for_session(
connection_id,
remote_node_id,
&session,
)
.await;
}
return Ok(());
}
}
}
println!(
"[NativeWebRTC] signal ignored: no native session found incoming_connection_id={} remote_node_id={:?}",
connection_id,
remote_node_id,
);
}
return Ok(());
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
{
let _ = (connection_id, remote_node_id, signal);
Ok(())
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
pub async fn request_webrtc_upgrade(
&self,
id: &str,
remote_node_id: Option<&str>,
reason: Option<&str>,
role_override: Option<&str>,
force_restart: bool,
) -> anyhow::Result<bool> {
if !self.is_webrtc_transport_enabled().await {
println!(
"[NativeWebRTC] explicit request ignored id={} remote_node_id={} force_restart={} reason={} cause=transport-disabled",
id,
remote_node_id.unwrap_or("none"),
force_restart,
reason.unwrap_or("none"),
);
return Ok(false);
}
let reason_label = reason.unwrap_or("none");
let role_override = match role_override
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase())
.as_deref()
{
Some("auto") => Some(crate::transport::NativeWebRTCRole::Auto),
Some("initiator") => Some(crate::transport::NativeWebRTCRole::Initiator),
Some("responder") => Some(crate::transport::NativeWebRTCRole::Responder),
Some(other) => anyhow::bail!("invalid WebRTC role override: {}", other),
None => None,
};
let explicit_remote_node_id = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
let mut requested = false;
let resolved_connection_ids = self.resolve_transport_connection_ids(id).await;
println!(
"[NativeWebRTC] explicit request resolving id={} remote_node_id={} force_restart={} role_override={:?} reason={} candidates={}",
id,
explicit_remote_node_id.as_deref().unwrap_or("none"),
force_restart,
role_override,
reason_label,
resolved_connection_ids.join(","),
);
for connection_id in resolved_connection_ids {
let Some(record) = self
.connection_manager
.get_by_connection_id(&connection_id)
.await
else {
println!(
"[NativeWebRTC] explicit request skipped connection_id={} reason=missing-managed-record",
connection_id,
);
continue;
};
let remote_node_id = explicit_remote_node_id
.clone()
.or_else(|| record.endpoint_id.clone())
.or_else(|| record.node_id.clone());
let Some(remote_node_id) = remote_node_id
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
else {
println!(
"[NativeWebRTC] explicit request skipped connection_id={} reason=missing-remote-node-id",
record.connection_id,
);
continue;
};
let explicit_negotiation_id =
Self::create_native_webrtc_negotiation_id(&record.connection_id);
if let Some(reason) = self
.native_webrtc_upgrade_fast_path_skip_reason(
&record.connection_id,
Some(&remote_node_id),
force_restart,
Some(&explicit_negotiation_id),
)
.await
{
println!(
"[NativeWebRTC] explicit request skipped connection_id={} remote_node_id={} reason={}",
record.connection_id,
remote_node_id,
reason,
);
continue;
}
println!(
"[NativeWebRTC] explicit request starting connection_id={} remote_node_id={} force_restart={} role_override={:?} negotiation_id={} reason={}",
record.connection_id,
remote_node_id,
force_restart,
role_override,
explicit_negotiation_id,
reason_label,
);
self.request_native_webrtc_recovery(
&record.connection_id,
Some(&remote_node_id),
NativeWebRTCRecoveryTrigger::Native(NativeWebRTCNativeTrigger::ExplicitRequest),
NativeWebRTCRecoveryOptions {
force_restart,
preferred_negotiation_id: Some(&explicit_negotiation_id),
role_override,
},
)
.await?;
requested = true;
}
println!(
"[NativeWebRTC] explicit request complete id={} remote_node_id={} requested={} reason={}",
id,
explicit_remote_node_id.as_deref().unwrap_or("none"),
requested,
reason_label,
);
Ok(requested)
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-webrtc")))]
pub async fn request_webrtc_upgrade(
&self,
id: &str,
remote_node_id: Option<&str>,
reason: Option<&str>,
role_override: Option<&str>,
force_restart: bool,
) -> anyhow::Result<bool> {
let _ = (id, remote_node_id, reason, role_override, force_restart);
Ok(false)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub async fn request_moq_upgrade(
&self,
id: &str,
remote_node_id: Option<&str>,
reason: Option<&str>,
) -> anyhow::Result<bool> {
if !self.is_moq_transport_enabled().await {
return Ok(false);
}
let _ = reason; let explicit_remote_node_id = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
let mut requested = false;
for connection_id in self.resolve_transport_connection_ids(id).await {
let Some(record) = self
.connection_manager
.get_by_connection_id(&connection_id)
.await
else {
continue;
};
let remote_node_id = explicit_remote_node_id
.clone()
.or_else(|| record.endpoint_id.clone())
.or_else(|| record.node_id.clone());
let Some(remote_node_id) = remote_node_id
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
else {
continue;
};
requested = self
.maybe_start_native_moq_upgrade_with_options(
&record.connection_id,
Some(&remote_node_id),
NativeMoqUpgradeOptions::EXPLICIT,
)
.await?
|| requested;
}
Ok(requested)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
pub(crate) async fn request_moq_repair(
&self,
id: &str,
remote_node_id: Option<&str>,
reason: Option<&str>,
) -> anyhow::Result<bool> {
if !self.is_moq_transport_enabled().await {
return Ok(false);
}
let _ = reason;
let explicit_remote_node_id = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
let mut requested = false;
for connection_id in self.resolve_transport_connection_ids(id).await {
let Some(record) = self
.connection_manager
.get_by_connection_id(&connection_id)
.await
else {
continue;
};
let remote_node_id = explicit_remote_node_id
.clone()
.or_else(|| record.endpoint_id.clone())
.or_else(|| record.node_id.clone());
let Some(remote_node_id) = remote_node_id
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
else {
continue;
};
requested = self
.maybe_start_native_moq_upgrade_with_options(
&record.connection_id,
Some(&remote_node_id),
NativeMoqUpgradeOptions::EXPLICIT_REPAIR,
)
.await?
|| requested;
}
Ok(requested)
}
#[cfg(all(
not(target_arch = "wasm32"),
feature = "transport-moq",
feature = "test-harness"
))]
pub async fn force_native_moq_session_failure_for_test(&self, id: &str) -> bool {
for connection_id in self.resolve_transport_connection_ids(id).await {
let session = {
let sessions = self.native_moq_sessions.read().await;
sessions.get(&connection_id).cloned()
};
if let Some(session) = session {
if self
.current_native_moq_route_identity_matches(&connection_id, &session)
.await
{
return session.force_failure_for_test().await;
}
}
}
false
}
#[cfg(all(
not(target_arch = "wasm32"),
not(feature = "transport-moq"),
feature = "test-harness"
))]
pub async fn force_native_moq_session_failure_for_test(&self, id: &str) -> bool {
let _ = id;
false
}
#[cfg(any(target_arch = "wasm32", not(feature = "transport-moq")))]
pub async fn request_moq_upgrade(
&self,
id: &str,
remote_node_id: Option<&str>,
reason: Option<&str>,
) -> anyhow::Result<bool> {
let _ = (id, remote_node_id, reason);
Ok(false)
}
}
impl Client {
pub async fn send_peer(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
if data.is_empty() {
return Ok(());
}
if data.len() > crate::native_protocol::MAX_PEER_MESSAGE_BYTES {
anyhow::bail!(
"peer message exceeds {}-byte limit; use a named stream for bulk data",
crate::native_protocol::MAX_PEER_MESSAGE_BYTES,
);
}
let protected = self.protect_outbound_peer_payload(id, data).await?;
match self.native_peer_send_order(id).await {
NativePeerSendOrder::OptionalTransportsFirst => {
if self.try_send_peer_over_webrtc(id, &protected).await? {
return Ok(());
}
if self.try_send_peer_over_moq(id, &protected).await? {
return Ok(());
}
self.send_peer_over_iroh(id, &protected).await
}
NativePeerSendOrder::IrohFirstAllowOptionalFallback => {
match self.send_peer_over_iroh(id, &protected).await {
Ok(()) => Ok(()),
Err(iroh_error) => {
if self.try_send_peer_over_webrtc(id, &protected).await? {
return Ok(());
}
if self.try_send_peer_over_moq(id, &protected).await? {
return Ok(());
}
Err(iroh_error)
}
}
}
}
}
async fn protect_outbound_peer_payload(
&self,
id: &str,
data: &[u8],
) -> anyhow::Result<Vec<u8>> {
let mut connection_ids = Vec::new();
if let Some(record) = self.connection_manager.get_by_connection_id(id).await {
connection_ids.push(record.connection_id);
} else if let Some(record) = self.connection_manager.best_connection_for_peer(id).await {
connection_ids.push(record.connection_id);
}
for connection_id in &connection_ids {
if self
.connection_application_crypto_keys
.read()
.ok()
.and_then(|keys| keys.get(connection_id).copied())
.is_some()
{
return self.protect_outbound_application_payload(connection_id, data);
}
}
if self.connection_ids_requiring_application_crypto(&connection_ids) {
anyhow::bail!("application crypto required for peer send but no key is installed");
}
Ok(data.to_vec())
}
#[cfg(not(target_arch = "wasm32"))]
async fn native_peer_send_order(&self, id: &str) -> NativePeerSendOrder {
let record = if let Some(record) = self.connection_manager.get_by_connection_id(id).await {
Some(record)
} else {
self.connection_manager.best_connection_for_peer(id).await
};
let active_transport = record.as_ref().map(|value| value.active_transport.as_str());
let iroh_lookup_id = record
.as_ref()
.and_then(|value| value.node_id.as_deref())
.or_else(|| {
record
.as_ref()
.and_then(|value| value.endpoint_id.as_deref())
})
.unwrap_or(id);
let iroh_transport = self.iroh_path_kind(iroh_lookup_id).await.transport_label();
peer_send_order(active_transport, Some(iroh_transport))
}
#[cfg(target_arch = "wasm32")]
async fn native_peer_send_order(&self, _id: &str) -> NativePeerSendOrder {
NativePeerSendOrder::OptionalTransportsFirst
}
}
#[cfg(not(target_arch = "wasm32"))]
fn frame_iroh_peer_application_payload(data: &[u8]) -> anyhow::Result<Vec<u8>> {
let frame_len = 1usize
.checked_add(data.len())
.and_then(|len| u32::try_from(len).ok())
.ok_or_else(|| anyhow::anyhow!("peer application frame is too large"))?;
let mut frame = Vec::with_capacity(4 + frame_len as usize);
frame.extend_from_slice(&frame_len.to_be_bytes());
frame.push(0x00);
frame.extend_from_slice(data);
Ok(frame)
}
#[cfg(not(target_arch = "wasm32"))]
fn frame_typescript_native_main_json(message: &serde_json::Value) -> anyhow::Result<Vec<u8>> {
let serialized = serde_json::to_vec(message)?;
let frame_len = serialized
.len()
.checked_add(1)
.and_then(|len| u32::try_from(len).ok())
.ok_or_else(|| anyhow::anyhow!("TypeScript native-main frame is too large"))?;
let mut frame = Vec::with_capacity(4 + frame_len as usize);
frame.extend_from_slice(&frame_len.to_be_bytes());
frame.push(0x00);
frame.extend_from_slice(&serialized);
Ok(frame)
}
#[cfg(not(target_arch = "wasm32"))]
impl Client {
pub async fn send_peer_over_iroh(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
let record = if let Some(r) = self.connection_manager.get_by_connection_id(id).await {
r
} else if let Some(r) = self.connection_manager.best_connection_for_peer(id).await {
r
} else {
anyhow::bail!("send_peer_over_iroh: no connection record for id={}", id);
};
let frame = frame_iroh_peer_application_payload(data)?;
self.send_peer_application_frame(&record.connection_id, &frame, Some(5_000))
.await?;
let transport_label = self.iroh_path_kind(id).await.transport_label();
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
if should_preserve_active_moq_on_iroh_send(
&record.active_transport,
self.native_moq_data_ready_for_peer(&record.connection_id)
.await,
) {
return Ok(());
}
if self
.get_connected_webrtc_session_for_peer(&record.connection_id)
.await
.is_some()
{
self.report_native_webrtc_candidate_transport(&record.connection_id, None)
.await;
} else {
if let Some(transport_stable_id) = record.transport_stable_id {
let _ = self
.report_transport_status_for_generation(
&record.connection_id,
transport_label,
None,
transport_stable_id,
record.transport_generation,
record.route_generation,
)
.await;
}
}
Ok(())
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn should_proactively_notify_browser_peer(
&self,
connection_id: &str,
remote_node_id: &str,
) -> bool {
let scopes = self.connection_manager.get_scopes(connection_id).await;
if scopes.iter().any(|scope| {
super::admission_impl::session_scope_uses_transient_peer_identity(scope.as_str())
}) {
return true;
}
self.connection_manager
.get_by_connection_id(connection_id)
.await
.and_then(|record| record.device_id)
.map(|device_id| device_id.trim().eq(remote_node_id))
.unwrap_or(false)
}
#[cfg(not(target_arch = "wasm32"))]
async fn maybe_negotiate_typescript_application_crypto(
&self,
connection_id: &str,
remote_node_id: Option<&str>,
handshake: &crate::native_protocol::TypeScriptHandshake,
) -> Option<[u8; crate::key_agreement::KEY_AGREEMENT_PUBLIC_KEY_BYTES]> {
let remote_public = handshake.application_key_agreement_public_key?;
let remote_node_id = remote_node_id
.map(str::trim)
.filter(|value| !value.is_empty())?;
let local_node_id = self.current_node_id().await?;
let key_agreement = self
.get_or_create_connection_key_agreement(connection_id)
.ok()?;
let key = key_agreement
.derive_application_crypto_key(&remote_public, &local_node_id, remote_node_id)
.ok()?;
let key_changed =
self.application_crypto_key_for_connection(Some(connection_id)) != Some(key);
self.set_connection_application_crypto_required(connection_id);
self.set_connection_application_crypto_key(connection_id, key);
if matches!(handshake.action.as_deref(), Some("response" | "ack")) {
self.confirm_connection_application_crypto(connection_id);
}
println!(
"[OpenRTC][KEY-AGREEMENT] connection_id={} action={} key_changed={} confirmed={} key_fingerprint={} local_public_fingerprint={} remote_public_fingerprint={} local_node_id={} remote_node_id={}",
connection_id,
handshake.action.as_deref().unwrap_or("hello"),
key_changed,
self.connection_application_crypto_is_confirmed(connection_id),
crate::key_agreement::key_agreement_fingerprint(&key),
crate::key_agreement::key_agreement_fingerprint(&key_agreement.public_key_bytes()),
crate::key_agreement::key_agreement_fingerprint(&remote_public),
local_node_id,
remote_node_id,
);
key_changed.then(|| key_agreement.public_key_bytes())
}
pub(crate) async fn send_typescript_capability_update(
&self,
connection_id: &str,
reason: &str,
application_key_agreement_public_key: Option<
[u8; crate::key_agreement::KEY_AGREEMENT_PUBLIC_KEY_BYTES],
>,
) -> anyhow::Result<()> {
self.send_typescript_handshake_update(
connection_id,
reason,
application_key_agreement_public_key,
"capability-update",
)
.await
}
async fn send_typescript_handshake_update(
&self,
connection_id: &str,
reason: &str,
application_key_agreement_public_key: Option<
[u8; crate::key_agreement::KEY_AGREEMENT_PUBLIC_KEY_BYTES],
>,
action: &str,
) -> anyhow::Result<()> {
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
let local_webrtc_enabled = self.is_webrtc_transport_enabled().await;
let local_moq_enabled = self.is_moq_transport_enabled().await;
let local_ble_enabled = self.is_ble_transport_enabled().await;
let application_key_agreement_public_key =
application_key_agreement_public_key.or_else(|| {
self.get_or_create_connection_key_agreement(connection_id)
.ok()
.map(|agreement| agreement.public_key_bytes())
});
let mut capability_frame = serde_json::json!({
"type": "handshake",
"action": action,
"capabilities": {
"webrtc": local_webrtc_enabled,
"moq": local_moq_enabled,
"ble": local_ble_enabled,
}
});
if let Some(public_key) = application_key_agreement_public_key {
capability_frame["capabilities"]["applicationKeyAgreement"] =
serde_json::Value::Bool(true);
capability_frame["applicationKeyAgreement"] = serde_json::json!({
"algorithm": crate::key_agreement::KEY_AGREEMENT_ALGORITHM,
"publicKey": URL_SAFE_NO_PAD.encode(public_key),
});
}
println!(
"[OpenRTC][capability] sending update connection_id={} reason={} action={} webrtc={} moq={} ble={} application_key_agreement={}",
connection_id,
reason,
action,
local_webrtc_enabled,
local_moq_enabled,
local_ble_enabled,
application_key_agreement_public_key.is_some(),
);
self.send_typescript_handshake_over_native_main(connection_id, &capability_frame)
.await
}
async fn send_typescript_handshake_over_native_main(
&self,
id: &str,
message: &serde_json::Value,
) -> anyhow::Result<()> {
let record = if let Some(r) = self.connection_manager.get_by_connection_id(id).await {
r
} else if let Some(r) = self.connection_manager.best_connection_for_peer(id).await {
r
} else {
anyhow::bail!(
"send_typescript_handshake_over_native_main: no connection record for id={}",
id
);
};
let endpoint_str = record
.endpoint_id
.as_deref()
.or(record.node_id.as_deref())
.ok_or_else(|| {
anyhow::anyhow!(
"send_typescript_handshake_over_native_main: missing endpoint for connection_id={}",
record.connection_id
)
})?;
let endpoint_id = endpoint_str.parse::<iroh::EndpointId>().map_err(|e| {
anyhow::anyhow!(
"send_typescript_handshake_over_native_main: invalid endpoint_id={} error={}",
endpoint_str,
e
)
})?;
let frame = frame_typescript_native_main_json(message)?;
self.send_native_main_frame(
id,
endpoint_id,
message
.get("type")
.and_then(|value| value.as_str())
.unwrap_or("control"),
&frame,
)
.await
.map_err(anyhow::Error::msg)
}
}
#[cfg(target_arch = "wasm32")]
impl Client {
pub async fn send_peer_over_iroh(&self, id: &str, data: &[u8]) -> anyhow::Result<()> {
let _ = (id, data);
Err(anyhow::anyhow!(
"send_peer_over_iroh: not available on WASM — use TypeScript Connection.sendTyped()"
))
}
}
fn parse_webrtc_signal_message(
frame: &serde_json::Value,
) -> Option<crate::transport::WebRTCSignalMessage> {
if frame.get("type").and_then(|value| value.as_str()) != Some("#pluto-signal") {
return None;
}
let content = frame.get("content")?;
let transport = content
.get("transport")
.and_then(|value| value.as_str())
.unwrap_or("webrtc");
if transport != "webrtc" {
return None;
}
let signal_type = content
.get("type")
.and_then(|value| value.as_str())
.map(ToOwned::to_owned)
.unwrap_or_else(|| "candidate".to_string());
Some(crate::transport::WebRTCSignalMessage {
transport: transport.to_string(),
signal_type,
negotiation_id: content
.get("negotiationId")
.and_then(|value| value.as_str())
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned),
sdp: content.get("sdp").cloned(),
candidate: content.get("candidate").cloned(),
})
}
fn should_send_typescript_capability_update(
local_webrtc_enabled: bool,
local_moq_enabled: bool,
local_ble_enabled: bool,
has_application_key_agreement_public_key: bool,
incoming_action: Option<&str>,
) -> bool {
match incoming_action {
Some("capability-update" | "response" | "ack") => return false,
_ => {}
}
has_application_key_agreement_public_key
|| local_webrtc_enabled
|| local_moq_enabled
|| local_ble_enabled
}
fn typescript_application_key_reply_action(
incoming_action: Option<&str>,
has_application_key: bool,
_application_key_changed: bool,
_application_crypto_was_confirmed: bool,
) -> Option<&'static str> {
if !has_application_key {
return None;
}
match incoming_action {
Some("ack") => None,
Some("response") => Some("ack"),
_ => Some("response"),
}
}
#[cfg(not(target_arch = "wasm32"))]
fn should_initiate_native_ble_upgrade(
local_node_id: &str,
remote_node_id: &str,
current_path: IrohPathKind,
) -> bool {
local_node_id > remote_node_id
&& !matches!(
current_path,
IrohPathKind::DirectQuic | IrohPathKind::DirectLan | IrohPathKind::Ble
)
}
#[cfg(not(target_arch = "wasm32"))]
fn should_accept_native_ble_switch_request(
local_node_id: &str,
remote_node_id: &str,
current_path: IrohPathKind,
) -> bool {
local_node_id < remote_node_id
&& matches!(current_path, IrohPathKind::Relay | IrohPathKind::Unknown)
}
#[cfg(not(target_arch = "wasm32"))]
async fn remove_native_transport_upgrade_gate(
gates: &std::sync::Arc<
tokio::sync::Mutex<std::collections::HashMap<(String, IrohPathKind), String>>,
>,
key: &(String, IrohPathKind),
upgrade_id: &str,
) -> bool {
let mut gates = gates.lock().await;
if gates.get(key).is_some_and(|current| current == upgrade_id) {
gates.remove(key);
true
} else {
false
}
}
#[cfg(test)]
mod tests {
#[cfg(not(target_arch = "wasm32"))]
use super::{
classify_native_ble_switch_admission, frame_iroh_peer_application_payload,
frame_typescript_native_main_json, iroh_path_allows_opportunistic_optional_route,
native_ble_attempt_state_after_recovery_wake, native_ble_retry_delay,
native_control_stream_rank, remove_native_transport_upgrade_gate,
reserve_native_ble_attempt_state, should_accept_native_ble_switch_request,
should_initiate_native_ble_upgrade, should_replace_native_control_stream,
NativeBleSwitchAdmission, NATIVE_BLE_MAX_UPGRADE_ATTEMPTS,
};
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
use super::{
parse_native_moq_route_proof, NativeMoqUpgradeOptions, NATIVE_MOQ_ROUTE_PROOF_TYPE,
};
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
use super::{parse_native_webrtc_route_proof, NATIVE_WEBRTC_ROUTE_PROOF_TYPE};
use super::{
should_send_typescript_capability_update, typescript_application_key_reply_action,
};
#[cfg(not(target_arch = "wasm32"))]
use crate::client::NativeControlStreamOwner;
#[cfg(not(target_arch = "wasm32"))]
use crate::route_policy::KnownRoute;
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn native_webrtc_signals_use_the_typescript_native_main_wire_discriminator() {
let signal = serde_json::json!({
"type": "#pluto-signal",
"content": {
"transport": "webrtc",
"type": "sdp",
"negotiationId": "native-native-1",
"sdp": { "type": "offer", "sdp": "v=0\r\n" }
}
});
let framed = frame_typescript_native_main_json(&signal).expect("frame signal");
let declared_len = u32::from_be_bytes(framed[..4].try_into().expect("length prefix"));
assert_eq!(declared_len as usize, framed.len() - 4);
match crate::native_protocol::parse_main_frame(&framed[4..]) {
crate::native_protocol::ParsedMainFrame::TypeScriptJson(parsed) => {
assert_eq!(parsed, signal);
}
other => panic!("expected TypeScriptJson, got {other:?}"),
}
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[test]
fn native_webrtc_route_proof_parser_accepts_only_bounded_wire_frames() {
let valid = serde_json::json!({
"type": NATIVE_WEBRTC_ROUTE_PROOF_TYPE,
"version": 1,
"role": "probe",
"probeId": "probe-1",
"negotiationId": "negotiation-1",
});
let parsed =
parse_native_webrtc_route_proof(&serde_json::to_vec(&valid).expect("valid proof JSON"))
.expect("valid proof");
assert_eq!(parsed.role, "probe");
assert_eq!(parsed.probe_id, "probe-1");
for invalid in [
serde_json::json!({
"type": NATIVE_WEBRTC_ROUTE_PROOF_TYPE,
"version": 2,
"role": "probe",
"probeId": "probe-1",
"negotiationId": "negotiation-1",
}),
serde_json::json!({
"type": NATIVE_WEBRTC_ROUTE_PROOF_TYPE,
"version": 1,
"role": "unexpected",
"probeId": "probe-1",
"negotiationId": "negotiation-1",
}),
serde_json::json!({
"type": NATIVE_WEBRTC_ROUTE_PROOF_TYPE,
"version": 1,
"role": "ack",
"probeId": "x".repeat(257),
"negotiationId": "negotiation-1",
}),
] {
assert!(parse_native_webrtc_route_proof(
&serde_json::to_vec(&invalid).expect("invalid proof JSON"),
)
.is_none());
}
assert!(parse_native_webrtc_route_proof(&vec![b'x'; 2_049]).is_none());
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
async fn native_webrtc_route_proof_fixture(
connection_id: &str,
) -> (
crate::client::Client,
std::sync::Arc<crate::transport::NativeWebRTCDataChannel>,
) {
let client = crate::client::Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
client
.update_transport_config(crate::client::TransportConfig {
webrtc: Some(crate::client::WebRTCConfig::default()),
..crate::client::TransportConfig::default()
})
.await
.expect("WebRTC transport config");
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some("remote-node".to_string()),
Some("remote-device".to_string()),
Some("remote-node".to_string()),
)
.await;
client
.connection_manager
.set_connected_with_transport(
connection_id,
Some("remote-node".to_string()),
Some(101),
Some(crate::transport_label::IROH_RELAY.to_string()),
)
.await;
client
.connection_manager
.set_health(
connection_id,
crate::connection_manager::ConnectionHealth::Stale,
)
.await;
client.set_connection_application_crypto_key(
connection_id,
[23u8; crate::application_crypto::APPLICATION_KEY_BYTES],
);
let session = std::sync::Arc::new(
crate::transport::NativeWebRTCDataChannel::new(
"local-node",
"remote-node",
&crate::client::WebRTCConfig::default(),
std::sync::Arc::new(|_| Box::pin(async move {})),
"native-webrtc-proof-negotiation",
)
.await
.expect("native WebRTC session"),
);
session.force_state_for_test(crate::transport::NativeWebRTCState::Connected);
let route_instance_id = std::sync::Arc::as_ptr(&session) as usize;
client
.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), session.clone());
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await
.expect("current WebRTC route generation");
client
.install_native_webrtc_route_proof(connection_id, &session)
.await
.expect("current WebRTC route proof");
(client, session)
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test]
async fn native_webrtc_watcher_retires_session_owned_by_replaced_transport_generation() {
let connection_id = "conn-webrtc-watcher-replacement";
let (client, session) = native_webrtc_route_proof_fixture(connection_id).await;
session.force_state_for_test(crate::transport::NativeWebRTCState::Connecting);
client
.native_webrtc_attempts_in_flight
.write()
.await
.insert(
connection_id.to_string(),
crate::client::NativeWebRTCAttemptInFlight {
attempt_id: "watcher-attempt".to_string(),
negotiation_id: session.negotiation_id(),
transport_stable_id: Some(101),
started_at_ms: super::now_millis_i64(),
expires_at_ms: i64::MAX,
preserve_on_direct_path: true,
last_skip_log_ms: 0,
},
);
client
.connection_manager
.set_connected_with_transport(
connection_id,
Some("remote-node".to_string()),
Some(202),
Some(crate::transport_label::IROH_RELAY.to_string()),
)
.await;
assert!(
!client
.current_native_webrtc_route_matches(connection_id, &session)
.await
);
assert!(
client
.retire_native_webrtc_watcher_session_if_current(
connection_id,
&session,
"watcher-attempt",
"test-generation-replaced",
)
.await
);
assert_eq!(session.state(), crate::transport::NativeWebRTCState::Closed);
assert!(client
.native_webrtc_session_for_connection(connection_id)
.await
.is_none());
assert!(client
.active_native_webrtc_attempt(connection_id)
.await
.is_none());
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test]
async fn native_webrtc_matching_protected_ack_promotes_current_route() {
let connection_id = "conn-webrtc-proof-ack";
let (client, session) = native_webrtc_route_proof_fixture(connection_id).await;
let proof = client
.native_webrtc_route_proofs
.read()
.await
.get(connection_id)
.cloned()
.expect("installed route proof");
let frame = super::NativeWebRtcRouteProofFrame {
frame_type: super::NATIVE_WEBRTC_ROUTE_PROOF_TYPE.to_string(),
version: 1,
role: "ack".to_string(),
probe_id: proof.probe_id,
negotiation_id: proof.negotiation_id,
};
let encoded = serde_json::to_vec(&frame).expect("proof frame");
let protected = client
.protect_outbound_application_payload(connection_id, &encoded)
.expect("protected proof frame");
assert!(client
.handle_current_native_webrtc_peer_data(
connection_id,
Some("remote-node"),
&session,
bytes::Bytes::from(protected),
)
.await
.expect("matching ACK should be consumed"));
assert!(
client
.current_native_webrtc_route_is_proven(connection_id, &session)
.await
);
assert_eq!(
client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("connection record")
.active_transport,
crate::transport_label::WEBRTC
);
assert_eq!(
client
.connection_manager
.peer_snapshot(connection_id)
.await
.expect("peer snapshot")
.health,
crate::connection_manager::ConnectionHealth::Healthy,
"a current protected route proof must restore managed readiness",
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test]
async fn native_webrtc_receiver_ack_never_promotes_without_local_proof() {
let connection_id = "conn-webrtc-proof-receiver";
let (client, session) = native_webrtc_route_proof_fixture(connection_id).await;
let frame = super::NativeWebRtcRouteProofFrame {
frame_type: super::NATIVE_WEBRTC_ROUTE_PROOF_TYPE.to_string(),
version: 1,
role: "probe".to_string(),
probe_id: "remote-probe".to_string(),
negotiation_id: "remote-negotiation".to_string(),
};
let encoded = serde_json::to_vec(&frame).expect("proof frame");
let protected = client
.protect_outbound_application_payload(connection_id, &encoded)
.expect("protected proof frame");
let _ = client
.handle_current_native_webrtc_peer_data(
connection_id,
Some("remote-node"),
&session,
bytes::Bytes::from(protected),
)
.await;
assert!(
!client
.current_native_webrtc_route_is_proven(connection_id, &session)
.await
);
assert!(crate::transport_label::is_iroh_base(
client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("connection record")
.active_transport
.as_str()
));
assert_eq!(
client
.connection_manager
.peer_snapshot(connection_id)
.await
.expect("peer snapshot")
.health,
crate::connection_manager::ConnectionHealth::Stale,
"a receiver-only probe must not heal readiness without local proof",
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test]
async fn native_webrtc_arbitrary_send_requires_proof_and_keeps_iroh_fallback() {
let connection_id = "conn-webrtc-proof-send";
let (client, session) = native_webrtc_route_proof_fixture(connection_id).await;
assert!(!client
.try_send_peer_over_webrtc(connection_id, b"arbitrary payload")
.await
.expect("unproven route should fall back"));
assert!(
!client
.current_native_webrtc_route_is_proven(connection_id, &session)
.await
);
assert!(crate::transport_label::is_iroh_base(
client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("connection record")
.active_transport
.as_str()
));
assert_eq!(
client
.connection_manager
.peer_snapshot(connection_id)
.await
.expect("peer snapshot")
.health,
crate::connection_manager::ConnectionHealth::Stale,
"an unproven route must not heal managed readiness",
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test]
async fn native_webrtc_stale_route_ack_cannot_promote_a_replacement() {
let connection_id = "conn-webrtc-proof-stale";
let (client, stale) = native_webrtc_route_proof_fixture(connection_id).await;
let proof = client
.native_webrtc_route_proofs
.read()
.await
.get(connection_id)
.cloned()
.expect("installed stale route proof");
let replacement = std::sync::Arc::new(
crate::transport::NativeWebRTCDataChannel::new(
"local-node",
"remote-node",
&crate::client::WebRTCConfig::default(),
std::sync::Arc::new(|_| Box::pin(async move {})),
"native-webrtc-replacement",
)
.await
.expect("replacement WebRTC session"),
);
replacement.force_state_for_test(crate::transport::NativeWebRTCState::Connected);
let replacement_instance_id = std::sync::Arc::as_ptr(&replacement) as usize;
client
.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), replacement);
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::WebRtc,
replacement_instance_id,
)
.await
.expect("replacement route generation");
let frame = super::NativeWebRtcRouteProofFrame {
frame_type: super::NATIVE_WEBRTC_ROUTE_PROOF_TYPE.to_string(),
version: 1,
role: "ack".to_string(),
probe_id: proof.probe_id,
negotiation_id: proof.negotiation_id,
};
let encoded = serde_json::to_vec(&frame).expect("proof frame");
let protected = client
.protect_outbound_application_payload(connection_id, &encoded)
.expect("protected proof frame");
assert!(!client
.handle_current_native_webrtc_peer_data(
connection_id,
Some("remote-node"),
&stale,
bytes::Bytes::from(protected),
)
.await
.expect("stale ACK should be ignored"));
assert!(crate::transport_label::is_iroh_base(
client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("connection record")
.active_transport
.as_str()
));
assert_eq!(
client
.connection_manager
.peer_snapshot(connection_id)
.await
.expect("peer snapshot")
.health,
crate::connection_manager::ConnectionHealth::Stale,
"a stale route proof must not heal the replacement generation",
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test(start_paused = true)]
async fn native_webrtc_route_proof_has_one_bounded_retry_owner() {
let connection_id = "conn-webrtc-proof-retry-owner";
let (client, session) = native_webrtc_route_proof_fixture(connection_id).await;
assert!(client
.request_current_native_webrtc_route_proof(connection_id, &session)
.await
.expect("first proof request"));
assert!(!client
.request_current_native_webrtc_route_proof(connection_id, &session)
.await
.expect("duplicate proof request"));
assert!(client
.native_webrtc_route_proofs
.read()
.await
.get(connection_id)
.is_some_and(|proof| proof.retry_active));
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
#[test]
fn native_moq_route_proof_parser_matches_the_browser_wire_contract() {
let valid = serde_json::json!({
"type": NATIVE_MOQ_ROUTE_PROOF_TYPE,
"version": 1,
"role": "ack",
"probeId": "probe_12345678",
"senderId": "peer-node",
"recipientId": "local-node",
});
let parsed =
parse_native_moq_route_proof(&serde_json::to_vec(&valid).expect("valid proof JSON"))
.expect("valid proof");
assert_eq!(parsed.role, "ack");
assert_eq!(parsed.probe_id, "probe_12345678");
assert_eq!(parsed.sender_id, "peer-node");
assert_eq!(parsed.recipient_id, "local-node");
for invalid in [
serde_json::json!({
"type": NATIVE_MOQ_ROUTE_PROOF_TYPE,
"version": 2,
"role": "probe",
"probeId": "probe_12345678",
"senderId": "peer-node",
"recipientId": "local-node",
}),
serde_json::json!({
"type": NATIVE_MOQ_ROUTE_PROOF_TYPE,
"version": 1,
"role": "unexpected",
"probeId": "probe_12345678",
"senderId": "peer-node",
"recipientId": "local-node",
}),
serde_json::json!({
"type": NATIVE_MOQ_ROUTE_PROOF_TYPE,
"version": 1,
"role": "probe",
"probeId": "short",
"senderId": "peer-node",
"recipientId": "local-node",
}),
serde_json::json!({
"type": NATIVE_MOQ_ROUTE_PROOF_TYPE,
"version": 1,
"role": "probe",
"probeId": "invalid probe id",
"senderId": "peer-node",
"recipientId": "local-node",
}),
serde_json::json!({
"type": NATIVE_MOQ_ROUTE_PROOF_TYPE,
"version": 1,
"role": "probe",
"probeId": "probe_12345678",
"senderId": "",
"recipientId": "local-node",
}),
] {
assert!(parse_native_moq_route_proof(
&serde_json::to_vec(&invalid).expect("invalid proof JSON"),
)
.is_none());
}
assert!(parse_native_moq_route_proof(&vec![b'x'; 513]).is_none());
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
#[tokio::test]
async fn native_moq_current_route_proof_restores_managed_readiness() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "conn-moq-proof-settles";
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some("remote-node".to_string()),
Some("remote-device".to_string()),
Some("remote-node".to_string()),
)
.await;
client
.connection_manager
.set_connected_with_transport(
connection_id,
Some("remote-node".to_string()),
Some(102),
Some(crate::transport_label::IROH_RELAY.to_string()),
)
.await;
client
.connection_manager
.set_health(
connection_id,
crate::connection_manager::ConnectionHealth::Stale,
)
.await;
let session = std::sync::Arc::new(
crate::transport::NativeMoQSession::new(
"local-node",
"remote-node",
&crate::client::MoQConfig {
relay_url: "https://relay.invalid/moq".to_string(),
..crate::client::MoQConfig::default()
},
)
.await
.expect("native MoQ session"),
);
session.force_state_for_test(crate::transport::NativeMoQState::Connected);
let route_instance_id = std::sync::Arc::as_ptr(&session) as usize;
client
.native_moq_sessions
.write()
.await
.insert(connection_id.to_string(), session.clone());
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::Moq,
route_instance_id,
)
.await
.expect("current MoQ route generation");
client
.install_native_moq_route_proof(connection_id, &session, "local-node", "remote-node")
.await;
assert!(
client
.mark_current_native_moq_route_proven(
connection_id,
Some("remote-node"),
&session,
"matching-protected-ack",
)
.await
);
assert_eq!(
client
.connection_manager
.peer_snapshot(connection_id)
.await
.expect("peer snapshot")
.health,
crate::connection_manager::ConnectionHealth::Healthy,
);
}
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn optional_route_generation_fails_closed_without_manager_record() {
let expected = crate::client::NativeOptionalRouteGeneration {
transport_stable_id: 7,
transport_generation: 2,
route_generation: 3,
route_instance_id: 11,
};
assert!(!super::native_optional_route_generation_matches(
None,
Some(expected),
expected.route_instance_id,
));
}
#[cfg(not(target_arch = "wasm32"))]
use super::Client;
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn closed_webrtc_session_rejects_delayed_signals_from_competing_generations() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "conn-closed-signal-storm";
let session = std::sync::Arc::new(
crate::transport::NativeWebRTCDataChannel::new(
"node-z",
"node-a",
&crate::client::WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
},
std::sync::Arc::new(|_| Box::pin(async move {})),
"negotiation-retired",
)
.await
.expect("native webrtc session"),
);
session.close();
client
.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), session.clone());
for negotiation_id in ["negotiation-old", "negotiation-racing"] {
for _ in 0..2 {
let error = client
.forward_native_webrtc_signal(
connection_id,
Some("node-a"),
crate::transport::WebRTCSignalMessage {
transport: "webrtc".to_string(),
signal_type: "candidate".to_string(),
negotiation_id: Some(negotiation_id.to_string()),
sdp: None,
candidate: Some(serde_json::json!({
"candidate": "candidate:delayed",
})),
},
)
.await
.expect_err("retired sessions must reject delayed ICE");
assert!(
error.to_string().contains("retired session"),
"unexpected rejection: {error}",
);
}
}
assert!(
client
.get_webrtc_session_for_peer(connection_id)
.await
.is_none(),
"a delayed signal storm must not expose a retired route to product handlers",
);
let retained = client
.native_webrtc_sessions
.read()
.await
.get(connection_id)
.cloned()
.expect("retired generation remains fenced until its owner replaces it");
assert!(std::sync::Arc::ptr_eq(&retained, &session));
assert_eq!(
retained.state(),
crate::transport::NativeWebRTCState::Closed,
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn optional_route_fence_tracks_post_promotion_route_generation() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "conn-post-promotion-generation";
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some("node-a".to_string()),
None,
Some("node-a".to_string()),
)
.await;
client
.connection_manager
.set_connected_with_transport(
connection_id,
None,
Some(73),
Some(crate::transport_label::IROH_RELAY.to_string()),
)
.await;
let session = std::sync::Arc::new(
crate::transport::NativeWebRTCDataChannel::new(
"node-z",
"node-a",
&crate::client::WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
},
std::sync::Arc::new(|_| Box::pin(async move {})),
"negotiation-post-promotion",
)
.await
.expect("native webrtc session"),
);
let route_instance_id = std::sync::Arc::as_ptr(&session) as usize;
client
.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), session);
let installed = client
.install_native_optional_route_generation(
connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await
.expect("current transport must install a route fence");
assert_eq!(installed.route_generation, 0);
client
.report_transport_status(
connection_id,
crate::transport_label::WEBRTC,
Some(crate::transport_label::IROH_RELAY),
)
.await
.expect("promotion should retain the peer");
let record = client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("promoted connection record");
let refreshed = client
.native_optional_route_generation_for_instance(
connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await
.expect("promotion must refresh the exact installed route fence");
assert_eq!(record.route_generation, 1);
assert_eq!(refreshed.route_generation, record.route_generation);
assert!(!super::native_optional_route_generation_matches(
Some(&record),
Some(installed),
route_instance_id,
));
assert!(super::native_optional_route_generation_matches(
Some(&record),
Some(refreshed),
route_instance_id,
));
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn idle_webrtc_route_can_be_fenced_before_start_without_being_data_ready() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "conn-idle-webrtc-fence";
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some("node-a".to_string()),
None,
Some("node-a".to_string()),
)
.await;
client
.connection_manager
.set_connected_with_transport(
connection_id,
None,
Some(81),
Some(crate::transport_label::IROH_RELAY.to_string()),
)
.await;
let session = std::sync::Arc::new(
crate::transport::NativeWebRTCDataChannel::new(
"node-z",
"node-a",
&crate::client::WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
},
std::sync::Arc::new(|_| Box::pin(async move {})),
"negotiation-idle-fence",
)
.await
.expect("native webrtc session"),
);
let route_instance_id = std::sync::Arc::as_ptr(&session) as usize;
client
.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), session.clone());
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::WebRtc,
route_instance_id,
)
.await
.expect("idle route should install its generation fence");
assert!(
client
.current_native_webrtc_route_identity_matches(connection_id, &session)
.await
);
assert!(
!client
.current_native_webrtc_route_matches(connection_id, &session)
.await
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
#[tokio::test]
async fn idle_moq_route_can_be_fenced_before_start_without_being_data_ready() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "conn-idle-moq-fence";
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some("node-a".to_string()),
None,
Some("node-a".to_string()),
)
.await;
client
.connection_manager
.set_connected_with_transport(
connection_id,
None,
Some(82),
Some(crate::transport_label::IROH_RELAY.to_string()),
)
.await;
let session = std::sync::Arc::new(
crate::transport::NativeMoQSession::new(
"node-z",
"node-a",
&crate::client::MoQConfig {
relay_url: "https://relay.invalid/moq".to_string(),
..crate::client::MoQConfig::default()
},
)
.await
.expect("native moq session"),
);
let route_instance_id = std::sync::Arc::as_ptr(&session) as usize;
client
.native_moq_sessions
.write()
.await
.insert(connection_id.to_string(), session.clone());
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::Moq,
route_instance_id,
)
.await
.expect("idle route should install its generation fence");
assert!(
client
.current_native_moq_route_identity_matches(connection_id, &session)
.await
);
assert!(
!client
.current_native_moq_route_matches(connection_id, &session)
.await
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
#[tokio::test]
async fn stale_webrtc_route_cannot_publish_status_over_replacement() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "conn-webrtc-status-fence";
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some("node-a".to_string()),
None,
Some("node-a".to_string()),
)
.await;
client
.connection_manager
.set_connected_with_transport(
connection_id,
None,
Some(91),
Some(crate::transport_label::IROH_RELAY.to_string()),
)
.await;
let make_session = |negotiation_id: &'static str| async move {
let session = std::sync::Arc::new(
crate::transport::NativeWebRTCDataChannel::new(
"node-z",
"node-a",
&crate::client::WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
},
std::sync::Arc::new(|_| Box::pin(async move {})),
negotiation_id,
)
.await
.expect("native webrtc session"),
);
session.force_state_for_test(crate::transport::NativeWebRTCState::Connected);
session
};
let stale = make_session("negotiation-stale").await;
client
.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), stale.clone());
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::WebRtc,
std::sync::Arc::as_ptr(&stale) as usize,
)
.await
.expect("stale route generation");
let current = make_session("negotiation-current").await;
client
.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), current.clone());
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::WebRtc,
std::sync::Arc::as_ptr(¤t) as usize,
)
.await
.expect("replacement route generation");
assert!(
!client
.report_current_native_webrtc_route_status(
connection_id,
&stale,
crate::transport_label::WEBRTC,
Some(crate::transport_label::IROH_RELAY),
false,
)
.await
);
assert!(crate::transport_label::is_iroh_base(
client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("connection record")
.active_transport
.as_str(),
));
assert!(
client
.report_current_native_webrtc_route_status(
connection_id,
¤t,
crate::transport_label::WEBRTC,
Some(crate::transport_label::IROH_RELAY),
false,
)
.await
);
let record = client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("promoted record");
assert_eq!(record.active_transport, crate::transport_label::WEBRTC);
assert_eq!(record.route_generation, 1);
assert_eq!(
client
.native_optional_route_generation_for_instance(
connection_id,
KnownRoute::WebRtc,
std::sync::Arc::as_ptr(¤t) as usize,
)
.await
.expect("refreshed route generation")
.route_generation,
1,
);
}
#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
#[tokio::test]
async fn stale_moq_route_cannot_publish_status_over_replacement() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "conn-moq-status-fence";
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some("node-a".to_string()),
None,
Some("node-a".to_string()),
)
.await;
client
.connection_manager
.set_connected_with_transport(
connection_id,
None,
Some(92),
Some(crate::transport_label::IROH_RELAY.to_string()),
)
.await;
let make_session = || async {
let session = std::sync::Arc::new(
crate::transport::NativeMoQSession::new(
"node-z",
"node-a",
&crate::client::MoQConfig {
relay_url: "https://relay.invalid/moq".to_string(),
..crate::client::MoQConfig::default()
},
)
.await
.expect("native moq session"),
);
session.force_state_for_test(crate::transport::NativeMoQState::Connected);
session
};
let stale = make_session().await;
client
.native_moq_sessions
.write()
.await
.insert(connection_id.to_string(), stale.clone());
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::Moq,
std::sync::Arc::as_ptr(&stale) as usize,
)
.await
.expect("stale route generation");
let current = make_session().await;
client
.native_moq_sessions
.write()
.await
.insert(connection_id.to_string(), current.clone());
client
.install_native_optional_route_generation(
connection_id,
KnownRoute::Moq,
std::sync::Arc::as_ptr(¤t) as usize,
)
.await
.expect("replacement route generation");
assert!(
!client
.report_current_native_moq_route_status(
connection_id,
&stale,
crate::transport_label::MOQ,
Some(crate::transport_label::IROH_RELAY),
false,
)
.await
);
assert!(crate::transport_label::is_iroh_base(
client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("connection record")
.active_transport
.as_str(),
));
assert!(
client
.report_current_native_moq_route_status(
connection_id,
¤t,
crate::transport_label::MOQ,
Some(crate::transport_label::IROH_RELAY),
false,
)
.await
);
let record = client
.connection_manager
.get_by_connection_id(connection_id)
.await
.expect("promoted record");
assert_eq!(record.active_transport, crate::transport_label::MOQ);
assert_eq!(record.route_generation, 1);
assert_eq!(
client
.native_optional_route_generation_for_instance(
connection_id,
KnownRoute::Moq,
std::sync::Arc::as_ptr(¤t) as usize,
)
.await
.expect("refreshed route generation")
.route_generation,
1,
);
}
#[cfg(not(target_arch = "wasm32"))]
#[tokio::test]
async fn repeated_typescript_application_key_is_not_reciprocated_twice() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
*client.node_id.write().await = Some("local-node".to_string());
let remote_agreement =
crate::key_agreement::EphemeralKeyAgreement::generate().expect("remote agreement");
let handshake = crate::native_protocol::TypeScriptHandshake {
action: Some("capability-update".to_string()),
session_token: None,
session_token_payload: None,
claimed_device_id: None,
capabilities: Some(crate::native_protocol::TypeScriptHandshakeCapabilities {
webrtc: Some(true),
moq: Some(false),
ble: Some(false),
application_key_agreement: Some(true),
}),
application_key_agreement_public_key: Some(remote_agreement.public_key_bytes()),
};
assert!(client
.maybe_negotiate_typescript_application_crypto(
"connection-1",
Some("remote-node"),
&handshake,
)
.await
.is_some());
assert!(client
.maybe_negotiate_typescript_application_crypto(
"connection-1",
Some("remote-node"),
&handshake,
)
.await
.is_none());
assert!(client.connection_requires_application_crypto_confirmation("connection-1"));
assert!(!client.connection_application_crypto_is_confirmed("connection-1"));
assert_eq!(
client.session_admission_block_reason("connection-1"),
Some((false, "application-crypto-confirmation-pending".to_string())),
);
let mut response = handshake.clone();
response.action = Some("response".to_string());
assert!(client
.maybe_negotiate_typescript_application_crypto(
"connection-1",
Some("remote-node"),
&response,
)
.await
.is_none());
assert!(client.connection_application_crypto_is_confirmed("connection-1"));
assert_eq!(client.session_admission_block_reason("connection-1"), None);
client.set_connection_application_crypto_key("connection-1", [9u8; 32]);
assert!(
!client.connection_application_crypto_is_confirmed("connection-1"),
"rotating the key must retire confirmation for the prior key",
);
let mut ack = handshake;
ack.action = Some("ack".to_string());
assert!(client
.maybe_negotiate_typescript_application_crypto(
"connection-1",
Some("remote-node"),
&ack,
)
.await
.is_some());
assert!(client.connection_application_crypto_is_confirmed("connection-1"));
}
#[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_webrtc_does_not_start_without_live_iroh_control_generation() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "stale-manager-record";
let remote_node_id = "5d388fcaaa61efa536d78051e060928f6c1fff0983a32f6b532176b12328c3ac";
*client.node_id.write().await = Some("local-node".to_string());
client
.update_transport_config(crate::client::TransportConfig {
webrtc: Some(crate::client::WebRTCConfig::default()),
..crate::client::TransportConfig::default()
})
.await
.expect("transport config");
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some(remote_node_id.to_string()),
Some("remote-device".to_string()),
Some(remote_node_id.to_string()),
)
.await;
client
.connection_manager
.set_connected(connection_id, Some(remote_node_id.to_string()))
.await;
client
.maybe_start_native_webrtc_upgrade(
connection_id,
Some(remote_node_id),
true,
None,
None,
)
.await
.expect("stale start is ignored");
assert!(
client
.native_webrtc_session_for_connection(connection_id)
.await
.is_none(),
"a manager record without its live Iroh generation must not create a WebRTC session",
);
assert!(
client
.active_native_webrtc_attempt(connection_id)
.await
.is_none(),
"a rejected start must not leave an in-flight negotiation owner",
);
}
#[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_webrtc_fast_path_allows_forced_recovery_of_active_session() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "local-node-remote-node";
let remote_node_id = "remote-node";
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some(remote_node_id.to_string()),
Some("remote-device".to_string()),
Some(remote_node_id.to_string()),
)
.await;
client
.connection_manager
.set_connected(connection_id, Some(remote_node_id.to_string()))
.await;
let (tx, _rx) = tokio::sync::mpsc::unbounded_channel::<serde_json::Value>();
let sender: crate::transport::NativeWebRTCSignalSender =
std::sync::Arc::new(move |signal| {
let tx = tx.clone();
Box::pin(async move {
let _ = tx.send(signal);
})
});
let session = std::sync::Arc::new(
crate::transport::NativeWebRTCDataChannel::new_with_role(
"local-node",
remote_node_id,
&crate::client::WebRTCConfig {
ice_servers: vec![],
privacy_mode: false,
lan_mode: false,
},
sender,
"native-local-node-remote-node",
crate::transport::NativeWebRTCRole::Responder,
)
.await
.expect("native WebRTC session"),
);
session.start().await.expect("session start");
assert_eq!(
session.state(),
crate::transport::NativeWebRTCState::Connecting
);
client
.native_webrtc_sessions
.write()
.await
.insert(connection_id.to_string(), session.clone());
let negotiation_id = session.negotiation_id();
let unforced_skip = client
.native_webrtc_upgrade_fast_path_skip_reason(
connection_id,
Some(remote_node_id),
false,
Some(&negotiation_id),
)
.await;
assert!(
unforced_skip
.as_deref()
.unwrap_or("")
.contains("existing session active"),
"unforced explicit request should skip an active session, got {unforced_skip:?}",
);
let forced_skip = client
.native_webrtc_upgrade_fast_path_skip_reason(
connection_id,
Some(remote_node_id),
true,
Some(&negotiation_id),
)
.await;
assert!(
forced_skip.is_none(),
"forceRestart must allow explicit recovery to replace an active native WebRTC session, got {forced_skip:?}",
);
session.close();
}
#[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_webrtc_force_restart_preempts_matching_in_flight_attempt() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "local-node-remote-node";
let remote_node_id = "remote-node";
let negotiation_id = "native-local-node-remote-node";
client
.update_transport_config(crate::client::TransportConfig {
webrtc: Some(crate::client::WebRTCConfig::default()),
..crate::client::TransportConfig::default()
})
.await
.expect("transport config");
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some(remote_node_id.to_string()),
Some("remote-device".to_string()),
Some(remote_node_id.to_string()),
)
.await;
client
.connection_manager
.set_connected(connection_id, Some(remote_node_id.to_string()))
.await;
client
.native_webrtc_attempts_in_flight
.write()
.await
.insert(
connection_id.to_string(),
crate::client::NativeWebRTCAttemptInFlight {
attempt_id: "old-attempt".to_string(),
negotiation_id: negotiation_id.to_string(),
transport_stable_id: None,
started_at_ms: 1,
expires_at_ms: i64::MAX,
preserve_on_direct_path: true,
last_skip_log_ms: 0,
},
);
client
.request_native_webrtc_recovery(
connection_id,
Some(remote_node_id),
crate::native_webrtc_policy::NativeWebRTCRecoveryTrigger::Native(
crate::native_webrtc_policy::NativeWebRTCNativeTrigger::ExplicitRequest,
),
crate::native_webrtc_policy::NativeWebRTCRecoveryOptions {
force_restart: true,
preferred_negotiation_id: Some(negotiation_id),
role_override: Some(crate::transport::NativeWebRTCRole::Initiator),
},
)
.await
.expect("forced recovery request");
assert!(
client
.active_native_webrtc_attempt(connection_id)
.await
.is_none(),
"forceRestart must preempt a stale in-flight native WebRTC attempt even when the deterministic negotiation id matches",
);
}
#[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_webrtc_force_restart_preserves_recent_matching_attempt() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "local-node-remote-node";
let remote_node_id = "remote-node";
let negotiation_id = "native-local-node-remote-node";
client
.update_transport_config(crate::client::TransportConfig {
webrtc: Some(crate::client::WebRTCConfig::default()),
..crate::client::TransportConfig::default()
})
.await
.expect("transport config");
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some(remote_node_id.to_string()),
Some("remote-device".to_string()),
Some(remote_node_id.to_string()),
)
.await;
client
.connection_manager
.set_connected(connection_id, Some(remote_node_id.to_string()))
.await;
client
.native_webrtc_attempts_in_flight
.write()
.await
.insert(
connection_id.to_string(),
crate::client::NativeWebRTCAttemptInFlight {
attempt_id: "current-attempt".to_string(),
negotiation_id: negotiation_id.to_string(),
transport_stable_id: None,
started_at_ms: super::now_millis_i64(),
expires_at_ms: i64::MAX,
preserve_on_direct_path: true,
last_skip_log_ms: 0,
},
);
client
.request_native_webrtc_recovery(
connection_id,
Some(remote_node_id),
crate::native_webrtc_policy::NativeWebRTCRecoveryTrigger::Native(
crate::native_webrtc_policy::NativeWebRTCNativeTrigger::ExplicitRequest,
),
crate::native_webrtc_policy::NativeWebRTCRecoveryOptions {
force_restart: true,
preferred_negotiation_id: Some(negotiation_id),
role_override: Some(crate::transport::NativeWebRTCRole::Initiator),
},
)
.await
.expect("forced recovery request");
assert!(
client
.active_native_webrtc_attempt(connection_id)
.await
.is_some(),
"forceRestart must not preempt a recent in-flight attempt without a replacement generation",
);
}
#[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn native_webrtc_replacement_transport_preempts_recent_in_flight_attempt() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "local-node-remote-node";
let remote_node_id = "remote-node";
let negotiation_id = "native-local-node-remote-node";
client
.update_transport_config(crate::client::TransportConfig {
webrtc: Some(crate::client::WebRTCConfig::default()),
..crate::client::TransportConfig::default()
})
.await
.expect("transport config");
client
.connection_manager
.upsert_pending(
connection_id.to_string(),
Some(remote_node_id.to_string()),
Some("remote-device".to_string()),
Some(remote_node_id.to_string()),
)
.await;
client
.connection_manager
.set_connected_with_transport(
connection_id,
Some(remote_node_id.to_string()),
Some(202),
Some("iroh-relay".to_string()),
)
.await;
client
.native_webrtc_attempts_in_flight
.write()
.await
.insert(
connection_id.to_string(),
crate::client::NativeWebRTCAttemptInFlight {
attempt_id: "stale-transport-attempt".to_string(),
negotiation_id: negotiation_id.to_string(),
transport_stable_id: Some(101),
started_at_ms: super::now_millis_i64(),
expires_at_ms: i64::MAX,
preserve_on_direct_path: true,
last_skip_log_ms: 0,
},
);
client
.request_native_webrtc_recovery(
connection_id,
Some(remote_node_id),
crate::native_webrtc_policy::NativeWebRTCRecoveryTrigger::Native(
crate::native_webrtc_policy::NativeWebRTCNativeTrigger::ExplicitRequest,
),
crate::native_webrtc_policy::NativeWebRTCRecoveryOptions {
force_restart: true,
preferred_negotiation_id: Some(negotiation_id),
role_override: Some(crate::transport::NativeWebRTCRole::Initiator),
},
)
.await
.expect("replacement recovery request");
assert_ne!(
client
.active_native_webrtc_attempt(connection_id)
.await
.map(|attempt| attempt.attempt_id),
Some("stale-transport-attempt".to_string()),
"a replacement base transport must never defer to an attempt owned by the retired transport generation",
);
}
#[cfg(all(feature = "transport-webrtc", not(target_arch = "wasm32")))]
#[tokio::test]
async fn native_webrtc_attempt_clear_is_fenced_by_unique_owner() {
let client = Client::new_with_app_tag(
crate::test_constants::TEST_PROJECT_ID.to_string(),
"test-app".to_string(),
Box::new(|| None),
);
let connection_id = "local-node-remote-node";
let negotiation_id = "native-local-node-remote-node";
client
.native_webrtc_attempts_in_flight
.write()
.await
.insert(
connection_id.to_string(),
crate::client::NativeWebRTCAttemptInFlight {
attempt_id: "replacement-attempt".to_string(),
negotiation_id: negotiation_id.to_string(),
transport_stable_id: Some(202),
started_at_ms: super::now_millis_i64(),
expires_at_ms: i64::MAX,
preserve_on_direct_path: true,
last_skip_log_ms: 0,
},
);
client
.clear_native_webrtc_attempt_if_current(connection_id, "retired-attempt")
.await;
assert_eq!(
client
.active_native_webrtc_attempt(connection_id)
.await
.map(|attempt| attempt.attempt_id),
Some("replacement-attempt".to_string()),
"a delayed callback from a retired attempt must not clear its replacement even when negotiation ids are reused",
);
}
#[test]
fn sends_typescript_capability_update_for_application_key_without_webrtc() {
assert!(should_send_typescript_capability_update(
false,
false,
false,
true,
Some("hello")
));
}
#[test]
fn application_key_reply_protocol_is_advertise_response_ack() {
assert_eq!(
typescript_application_key_reply_action(Some("capability-update"), true, true, false,),
Some("response"),
);
assert_eq!(
typescript_application_key_reply_action(Some("hello"), true, true, false),
Some("response"),
);
assert_eq!(
typescript_application_key_reply_action(Some("response"), true, false, false),
Some("ack"),
);
assert_eq!(
typescript_application_key_reply_action(Some("ack"), true, true, false),
None,
);
assert_eq!(
typescript_application_key_reply_action(Some("capability-update"), false, false, false,),
None,
);
assert_eq!(
typescript_application_key_reply_action(Some("capability-update"), true, false, true,),
Some("response"),
);
assert_eq!(
typescript_application_key_reply_action(Some("response"), true, false, true),
Some("ack"),
);
}
#[test]
fn advertises_local_transport_before_remote_capability_is_known() {
assert!(should_send_typescript_capability_update(
true,
false,
false,
false,
Some("hello")
));
assert!(!should_send_typescript_capability_update(
false,
true,
false,
false,
Some("response")
));
assert!(should_send_typescript_capability_update(
false, false, true, false, None
));
assert!(!should_send_typescript_capability_update(
false,
false,
false,
false,
Some("hello")
));
assert!(!should_send_typescript_capability_update(
false,
false,
true,
false,
Some("capability-update")
));
assert!(!should_send_typescript_capability_update(
true,
false,
false,
true,
Some("capability-update")
));
assert!(!should_send_typescript_capability_update(
true,
true,
true,
true,
Some("ack")
));
}
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn native_ble_upgrade_has_one_initiator_and_preserves_better_paths() {
assert!(should_initiate_native_ble_upgrade(
"node-z",
"node-a",
crate::client::IrohPathKind::Relay,
));
assert!(should_send_typescript_capability_update(
true,
true,
true,
false,
Some("hello")
));
assert!(!should_initiate_native_ble_upgrade(
"node-a",
"node-z",
crate::client::IrohPathKind::Relay,
));
for path in [
crate::client::IrohPathKind::DirectQuic,
crate::client::IrohPathKind::DirectLan,
crate::client::IrohPathKind::Ble,
] {
assert!(!should_initiate_native_ble_upgrade(
"node-z", "node-a", path,
));
}
assert!(should_initiate_native_ble_upgrade(
"node-z",
"node-a",
crate::client::IrohPathKind::Unknown,
));
assert!(should_accept_native_ble_switch_request(
"node-a",
"node-z",
crate::client::IrohPathKind::Relay,
));
assert!(!should_accept_native_ble_switch_request(
"node-z",
"node-a",
crate::client::IrohPathKind::Relay,
));
assert!(!should_accept_native_ble_switch_request(
"node-a",
"node-z",
crate::client::IrohPathKind::Ble,
));
}
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn opportunistic_optional_routes_only_start_when_iroh_is_not_direct() {
assert!(iroh_path_allows_opportunistic_optional_route(
crate::client::IrohPathKind::Relay,
));
for path in [
crate::client::IrohPathKind::Unknown,
crate::client::IrohPathKind::DirectQuic,
crate::client::IrohPathKind::DirectLan,
crate::client::IrohPathKind::Ble,
] {
assert!(!iroh_path_allows_opportunistic_optional_route(path));
}
#[cfg(feature = "transport-moq")]
{
assert!(!NativeMoqUpgradeOptions::OPPORTUNISTIC.allow_on_optimal_iroh_path);
assert!(NativeMoqUpgradeOptions::EXPLICIT.allow_on_optimal_iroh_path);
assert!(NativeMoqUpgradeOptions::EXPLICIT_REPAIR.allow_on_optimal_iroh_path);
}
}
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn native_ble_switch_waits_for_pending_admission_without_accepting_rejection() {
let pending = (false, "native-main-route-pending".to_string());
let rejected = (true, "session-token-rejected".to_string());
assert_eq!(
classify_native_ble_switch_admission(None),
NativeBleSwitchAdmission::Ready,
);
assert_eq!(
classify_native_ble_switch_admission(Some(&pending)),
NativeBleSwitchAdmission::Wait,
);
assert_eq!(
classify_native_ble_switch_admission(Some(&rejected)),
NativeBleSwitchAdmission::Reject,
);
}
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn native_ble_retry_is_bounded_jittered_and_reset_by_typed_recovery_wakes() {
let generation_one = crate::client::NativePeerDataGeneration {
transport_stable_id: 10,
transport_generation: 2,
route_generation: 1,
};
let generation_two = crate::client::NativePeerDataGeneration {
transport_stable_id: 11,
transport_generation: 3,
route_generation: 0,
};
let (state, attempt) = reserve_native_ble_attempt_state(None, generation_one);
assert_eq!(attempt, Some(1));
assert_eq!(native_ble_retry_delay(1, 0).unwrap().as_millis(), 800);
assert_eq!(native_ble_retry_delay(1, 40).unwrap().as_millis(), 1_200);
let (state, attempt) = reserve_native_ble_attempt_state(Some(state), generation_one);
assert_eq!(attempt, Some(2));
assert_eq!(native_ble_retry_delay(2, 0).unwrap().as_millis(), 1_600);
assert_eq!(native_ble_retry_delay(2, 40).unwrap().as_millis(), 2_400);
let (state, attempt) = reserve_native_ble_attempt_state(Some(state), generation_one);
assert_eq!(attempt, Some(3));
assert_eq!(native_ble_retry_delay(3, 20), None);
let (exhausted, attempt) = reserve_native_ble_attempt_state(Some(state), generation_one);
assert_eq!(attempt, None);
assert_eq!(exhausted.attempts, NATIVE_BLE_MAX_UPGRADE_ATTEMPTS);
assert_eq!(
native_ble_attempt_state_after_recovery_wake(exhausted, true),
Some(exhausted),
);
let reset_after_wake = native_ble_attempt_state_after_recovery_wake(exhausted, false);
let (reset, attempt) = reserve_native_ble_attempt_state(reset_after_wake, generation_one);
assert_eq!(attempt, Some(1));
assert_eq!(reset.generation, generation_one);
let (reset, attempt) = reserve_native_ble_attempt_state(Some(exhausted), generation_two);
assert_eq!(attempt, Some(1));
assert_eq!(reset.generation, generation_two);
}
#[cfg(not(target_arch = "wasm32"))]
#[tokio::test]
async fn native_ble_upgrade_gate_cleanup_is_generation_fenced() {
let gates =
std::sync::Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::from([
(
("connection-1".to_string(), crate::client::IrohPathKind::Ble),
"upgrade-new".to_string(),
),
])));
let key = ("connection-1".to_string(), crate::client::IrohPathKind::Ble);
assert!(!remove_native_transport_upgrade_gate(&gates, &key, "upgrade-old").await);
assert_eq!(
gates.lock().await.get(&key).map(String::as_str),
Some("upgrade-new")
);
assert!(remove_native_transport_upgrade_gate(&gates, &key, "upgrade-new").await);
assert!(!gates.lock().await.contains_key(&key));
}
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn native_control_stream_rank_converges_on_same_quic_stream() {
let client_stream =
iroh::endpoint::StreamId::new(iroh::endpoint::Side::Client, iroh::endpoint::Dir::Bi, 0);
let server_stream =
iroh::endpoint::StreamId::new(iroh::endpoint::Side::Server, iroh::endpoint::Dir::Bi, 0);
assert!(
native_control_stream_rank(client_stream) < native_control_stream_rank(server_stream)
);
}
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn native_control_stream_replacement_is_generation_fenced() {
let generation_one = NativeControlStreamOwner {
transport_stable_id: 10,
stream_rank: 4,
};
let same_generation_better_stream = NativeControlStreamOwner {
transport_stable_id: 10,
stream_rank: 2,
};
let same_generation_worse_stream = NativeControlStreamOwner {
transport_stable_id: 10,
stream_rank: 6,
};
let generation_two = NativeControlStreamOwner {
transport_stable_id: 11,
stream_rank: 8,
};
assert!(should_replace_native_control_stream(
generation_one,
same_generation_better_stream,
));
assert!(!should_replace_native_control_stream(
generation_one,
same_generation_worse_stream,
));
assert!(should_replace_native_control_stream(
generation_one,
generation_two,
));
}
#[cfg(not(target_arch = "wasm32"))]
#[test]
fn iroh_peer_payload_uses_application_stream_framing_not_native_main() {
let frame = frame_iroh_peer_application_payload(b"hello").unwrap();
assert_eq!(u32::from_be_bytes(frame[..4].try_into().unwrap()), 6);
assert_eq!(frame[4], 0x00);
assert_eq!(&frame[5..], b"hello");
assert_ne!(&frame[5..], b"main");
}
}