1#[cfg(all(not(target_arch = "wasm32"), feature = "native-dns-socket-initializer"))]
2pub use iroh::dns::{
3 NativeSocketProtocol as NativeDnsSocketProtocol, NativeSocketRef as NativeDnsSocketRef,
4};
5#[cfg(all(not(target_arch = "wasm32"), feature = "native-dns-socket-initializer"))]
6type NativeDnsSocketInitializer = Arc<
7 dyn for<'a> Fn(NativeDnsSocketRef<'a>, NativeDnsSocketProtocol) -> std::io::Result<()>
8 + Send
9 + Sync,
10>;
11use super::coordination::room::{GatewayRequiredRoomBackend, RoomBackend};
12use crate::signaling::{GatewayRequiredSignalingBackend, SignalingBackend};
13#[cfg(all(not(target_arch = "wasm32"), feature = "native-udp-socket-initializer"))]
14pub use iroh::endpoint::{NativeUdpIpFamily, NativeUdpSocketRef};
15#[cfg(all(
16 not(target_arch = "wasm32"),
17 feature = "native-relay-tcp-socket-initializer"
18))]
19pub use iroh_relay::client::NativeTcpSocketRef;
20#[cfg(all(
21 not(target_arch = "wasm32"),
22 feature = "native-relay-tcp-socket-initializer"
23))]
24type NativeRelayTcpSocketInitializer =
25 Arc<dyn for<'a> Fn(NativeTcpSocketRef<'a>) -> std::io::Result<()> + Send + Sync>;
26use iroh::Endpoint;
27#[cfg(all(not(target_arch = "wasm32"), feature = "native-udp-socket-initializer"))]
28type NativeUdpSocketInitializer = Arc<
29 dyn for<'a> Fn(NativeUdpSocketRef<'a>, NativeUdpIpFamily) -> std::io::Result<()> + Send + Sync,
30>;
31
32use iroh_tickets::endpoint::EndpointTicket;
33use std::collections::{HashMap, HashSet};
34use std::str::FromStr;
35#[cfg(any(
36 not(target_arch = "wasm32"),
37 feature = "transport-webrtc",
38 feature = "transport-moq"
39))]
40use std::sync::atomic::Ordering;
41use std::sync::atomic::{AtomicBool, AtomicU64};
42use std::sync::{Arc, Mutex, RwLock as StdRwLock};
43use tokio::sync::RwLock;
44
45#[cfg(not(target_arch = "wasm32"))]
46use crate::client::scope_classifier::{default_classifier, ScopeClassifier};
47
48#[cfg(target_arch = "wasm32")]
49fn wasm_init_log(stage: &str) {
50 let now = js_sys::Date::now();
51 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
52 "[OPENRTC][WASM-INIT] ts_ms={:.0} stage={}",
53 now, stage
54 )));
55}
56
57#[cfg(target_arch = "wasm32")]
58use crate::wasm_node::{AcceptEvent, ConnectEvent, IrohWasmNode};
59
60#[cfg(not(target_arch = "wasm32"))]
61use crate::native_node::{AcceptEvent, ConnectEvent, IncomingStream, IrohNativeNode};
62
63#[cfg(not(target_arch = "wasm32"))]
64#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
65#[serde(rename_all = "camelCase")]
66pub struct EndpointHandle {
67 pub node_id: String,
68 pub node_addr: String,
69}
70
71#[cfg(not(target_arch = "wasm32"))]
72pub struct BiStream {
73 pub send: crate::application_crypto_streams::PeerSendStream,
74 pub recv: crate::application_crypto_streams::PeerRecvStream,
75 pub id: String,
76}
77
78#[cfg(not(target_arch = "wasm32"))]
82pub enum BiStreamDisposition {
83 Consumed,
84 Forward {
85 send: iroh::endpoint::SendStream,
86 recv: iroh::endpoint::RecvStream,
87 recv_prefix: Vec<u8>,
92 },
93}
94
95#[cfg(not(target_arch = "wasm32"))]
96#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
97#[serde(rename_all = "camelCase")]
98pub struct NativePeerDataEvent {
99 pub connection_id: String,
100 pub remote_node_id: Option<String>,
101 pub transport: String,
102 pub transport_stable_id: u64,
103 pub transport_generation: u64,
104 pub route_generation: u64,
105 pub payload: Vec<u8>,
106}
107
108#[cfg(not(target_arch = "wasm32"))]
109#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
110#[serde(rename_all = "camelCase")]
111pub(crate) struct NativePeerDataGeneration {
112 pub transport_stable_id: u64,
113 pub transport_generation: u64,
114 pub route_generation: u64,
115}
116
117#[cfg(target_arch = "wasm32")]
118#[derive(Debug, Clone, Copy, PartialEq, Eq)]
119pub(crate) struct WasmPeerDataGeneration {
120 pub transport_stable_id: u64,
121 pub transport_generation: u64,
122 pub route_generation: u64,
123}
124
125#[cfg(not(target_arch = "wasm32"))]
126#[derive(Debug, Clone, Copy, PartialEq, Eq)]
127pub(crate) struct NativeBleUpgradeAttemptState {
128 pub generation: NativePeerDataGeneration,
129 pub attempts: u8,
130}
131
132#[cfg(not(target_arch = "wasm32"))]
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub(crate) struct NativeTransportUpgradeGate {
135 pub upgrade_id: String,
136 pub generation: NativePeerDataGeneration,
137 pub policy_epoch: u64,
138 pub peer_policy_epoch: u64,
139}
140
141#[cfg(target_arch = "wasm32")]
142#[derive(Debug, Clone, PartialEq, Eq)]
143pub(crate) struct WasmTransportUpgradeGate {
144 pub upgrade_id: String,
145 pub generation: WasmPeerDataGeneration,
149 pub policy_epoch: u64,
150 pub peer_policy_epoch: u64,
151}
152
153#[derive(Debug, Clone, Copy, PartialEq, Eq)]
158#[cfg(any(
159 test,
160 all(
161 target_arch = "wasm32",
162 any(feature = "transport-webrtc", feature = "transport-moq")
163 )
164))]
165pub(crate) enum WasmCarrierUpgradeReservation {
166 Reserved,
167 CurrentReplay,
168 Busy,
169}
170
171#[derive(Debug, Clone, Copy, PartialEq, Eq)]
173#[cfg(any(
174 test,
175 all(
176 target_arch = "wasm32",
177 any(feature = "transport-webrtc", feature = "transport-moq")
178 )
179))]
180pub(crate) enum WasmCarrierUpgradeFence {
181 Current(u64),
182 Retired,
183 Superseded,
184 PolicyChanged,
185 BaseGenerationChanged,
186}
187
188#[cfg(any(
192 test,
193 all(
194 target_arch = "wasm32",
195 any(feature = "transport-webrtc", feature = "transport-moq")
196 )
197))]
198pub(crate) fn wasm_peer_carrier_failure_code(failure_code: Option<&str>) -> &'static str {
199 match failure_code {
200 Some("carrier-base-generation-stale") => "carrier-base-generation-stale",
201 Some("carrier-base-not-ready") => "carrier-base-not-ready",
202 _ => "peer-rejected-carrier",
203 }
204}
205
206#[cfg(any(
210 test,
211 all(
212 target_arch = "wasm32",
213 any(feature = "transport-webrtc", feature = "transport-moq")
214 )
215))]
216pub(crate) fn should_rearm_wasm_carrier_event_retry(
217 failure_code: &str,
218 initiator: bool,
219 completion_started: bool,
220) -> bool {
221 initiator && !completion_started && failure_code == "carrier-base-not-ready"
222}
223
224#[cfg(any(
228 test,
229 all(
230 target_arch = "wasm32",
231 any(feature = "transport-webrtc", feature = "transport-moq")
232 )
233))]
234pub(crate) fn classify_wasm_carrier_upgrade_reservation(
235 current: Option<(&str, (u64, u64, u64), u64)>,
236 competing_kind_current: bool,
237 upgrade_id: &str,
238 generation: (u64, u64, u64),
239 policy_epoch: u64,
240) -> WasmCarrierUpgradeReservation {
241 if competing_kind_current {
242 return WasmCarrierUpgradeReservation::Busy;
243 }
244 match current {
245 Some((current_id, current_generation, current_policy_epoch))
246 if current_generation == generation
247 && current_id == upgrade_id
248 && current_policy_epoch == policy_epoch =>
249 {
250 WasmCarrierUpgradeReservation::CurrentReplay
251 }
252 Some((_, current_generation, _)) if current_generation == generation => {
253 WasmCarrierUpgradeReservation::Busy
254 }
255 _ => WasmCarrierUpgradeReservation::Reserved,
256 }
257}
258
259#[cfg(any(
260 test,
261 all(
262 target_arch = "wasm32",
263 any(feature = "transport-webrtc", feature = "transport-moq")
264 )
265))]
266pub(crate) fn classify_wasm_carrier_upgrade_fence(
267 current: Option<(&str, (u64, u64, u64), u64)>,
268 upgrade_id: &str,
269 generation: (u64, u64, u64),
270 policy_epoch: u64,
271) -> WasmCarrierUpgradeFence {
272 match current {
273 None => WasmCarrierUpgradeFence::Retired,
274 Some((_, current_generation, _)) if current_generation != generation => {
275 WasmCarrierUpgradeFence::BaseGenerationChanged
276 }
277 Some((current_id, _, _)) if current_id != upgrade_id => WasmCarrierUpgradeFence::Superseded,
278 Some((_, _, current_policy_epoch)) if current_policy_epoch != policy_epoch => {
279 WasmCarrierUpgradeFence::PolicyChanged
280 }
281 Some((_, _, current_policy_epoch)) => {
282 WasmCarrierUpgradeFence::Current(current_policy_epoch)
283 }
284 }
285}
286
287#[cfg(any(test, feature = "iroh-carrier-core"))]
291pub(crate) fn should_bump_remote_carrier_peer_policy_epoch(
292 capabilities_were_known: bool,
293 capabilities_changed: bool,
294 matching_first_attempt: bool,
295) -> bool {
296 capabilities_changed && (capabilities_were_known || !matching_first_attempt)
297}
298
299#[cfg(test)]
300mod wasm_carrier_upgrade_gate_tests {
301 use super::*;
302
303 const BASE: (u64, u64, u64) = (7, 11, 13);
304
305 #[test]
306 fn delayed_proof_duplicate_replay_keeps_the_original_attempt_fenced() {
307 let current = Some(("attempt-a", BASE, 23));
308
309 assert_eq!(
310 classify_wasm_carrier_upgrade_reservation(current, false, "attempt-a", BASE, 23),
311 WasmCarrierUpgradeReservation::CurrentReplay,
312 );
313 assert_eq!(
314 classify_wasm_carrier_upgrade_reservation(current, false, "attempt-b", BASE, 23),
315 WasmCarrierUpgradeReservation::Busy,
316 );
317 assert_eq!(
318 classify_wasm_carrier_upgrade_fence(current, "attempt-a", BASE, 23),
319 WasmCarrierUpgradeFence::Current(23),
320 );
321 }
322
323 #[test]
324 fn another_carrier_cannot_overtake_a_proving_browser_generation() {
325 assert_eq!(
326 classify_wasm_carrier_upgrade_reservation(None, true, "webrtc-attempt", BASE, 23,),
327 WasmCarrierUpgradeReservation::Busy,
328 );
329 }
330
331 #[test]
332 fn remote_capability_epoch_preserves_only_a_matching_first_observation() {
333 assert!(!should_bump_remote_carrier_peer_policy_epoch(
334 false, true, true,
335 ));
336 assert!(should_bump_remote_carrier_peer_policy_epoch(
337 false, true, false,
338 ));
339 assert!(should_bump_remote_carrier_peer_policy_epoch(
340 true, true, true,
341 ));
342 assert!(!should_bump_remote_carrier_peer_policy_epoch(
343 true, false, false,
344 ));
345 }
346
347 #[test]
348 fn peer_base_readiness_failure_remains_recoverable() {
349 assert_eq!(
350 wasm_peer_carrier_failure_code(Some("carrier-base-not-ready")),
351 "carrier-base-not-ready",
352 );
353 assert_eq!(
354 wasm_peer_carrier_failure_code(Some("carrier-base-generation-stale")),
355 "carrier-base-generation-stale",
356 );
357 assert_eq!(
358 wasm_peer_carrier_failure_code(Some("carrier-authorization-rejected")),
359 "peer-rejected-carrier",
360 );
361 assert!(should_rearm_wasm_carrier_event_retry(
362 "carrier-base-not-ready",
363 true,
364 false,
365 ));
366 assert!(!should_rearm_wasm_carrier_event_retry(
367 "peer-rejected-carrier",
368 true,
369 false,
370 ));
371 assert!(!should_rearm_wasm_carrier_event_retry(
372 "carrier-base-not-ready",
373 false,
374 false,
375 ));
376 assert!(!should_rearm_wasm_carrier_event_retry(
377 "carrier-base-not-ready",
378 true,
379 true,
380 ));
381 }
382
383 #[test]
384 fn carrier_fence_distinguishes_lifecycle_retirement_from_policy_change() {
385 assert_eq!(
386 classify_wasm_carrier_upgrade_fence(None, "attempt-a", BASE, 23),
387 WasmCarrierUpgradeFence::Retired,
388 );
389 assert_eq!(
390 classify_wasm_carrier_upgrade_fence(
391 Some(("attempt-a", BASE, 23)),
392 "attempt-a",
393 BASE,
394 24,
395 ),
396 WasmCarrierUpgradeFence::PolicyChanged,
397 );
398 assert_eq!(
399 classify_wasm_carrier_upgrade_fence(
400 Some(("attempt-b", BASE, 23)),
401 "attempt-a",
402 BASE,
403 23,
404 ),
405 WasmCarrierUpgradeFence::Superseded,
406 );
407 assert_eq!(
408 classify_wasm_carrier_upgrade_fence(
409 Some(("attempt-a", (17, 19, 23), 23)),
410 "attempt-a",
411 BASE,
412 23,
413 ),
414 WasmCarrierUpgradeFence::BaseGenerationChanged,
415 );
416 }
417}
418
419#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
420pub(crate) struct NativeWebRtcCarrierAttempt {
421 pub upgrade_id: String,
422 pub bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
423 pub generation: NativePeerDataGeneration,
424 pub remote_endpoint_id: iroh::EndpointId,
425 pub channel: Arc<crate::transport::WebRtcDataChannel>,
426 pub pump: tokio::sync::Mutex<Option<crate::native_webrtc_carrier::NativeWebRtcCarrierSession>>,
427 pub completion_started: std::sync::atomic::AtomicBool,
430 pub ice_refresh: tokio::sync::Mutex<crate::iroh_carrier_bootstrap::WebRtcIceRefreshGate>,
431 pub ice_refresh_baseline:
432 tokio::sync::Mutex<Option<crate::packet_carrier_transport::WebRtcIceRefreshPathLoan>>,
433}
434
435#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
436pub(crate) struct NativeMoqCarrierAttempt {
437 pub upgrade_id: String,
438 pub bootstrap: crate::iroh_carrier_bootstrap::CarrierBootstrapFrame,
439 pub generation: NativePeerDataGeneration,
440 pub remote_endpoint_id: iroh::EndpointId,
441 pub session: Arc<crate::transport::NativeMoQSession>,
442 pub pump: tokio::sync::Mutex<Option<crate::native_moq_carrier::NativeMoqCarrierSession>>,
443}
444
445#[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
446impl std::fmt::Debug for NativeMoqCarrierAttempt {
447 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
448 formatter
449 .debug_struct("NativeMoqCarrierAttempt")
450 .field("upgrade_id", &self.upgrade_id)
451 .field("generation", &self.generation)
452 .field("remote_endpoint_id", &self.remote_endpoint_id)
453 .finish_non_exhaustive()
454 }
455}
456
457#[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
458impl std::fmt::Debug for NativeWebRtcCarrierAttempt {
459 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
460 formatter
461 .debug_struct("NativeWebRtcCarrierAttempt")
462 .field("upgrade_id", &self.upgrade_id)
463 .field("generation", &self.generation)
464 .field("remote_endpoint_id", &self.remote_endpoint_id)
465 .finish_non_exhaustive()
466 }
467}
468
469#[cfg(not(target_arch = "wasm32"))]
473#[derive(Debug, Clone, Copy, PartialEq, Eq)]
474pub(crate) struct NativeControlStreamOwner {
475 pub transport_stable_id: u64,
476 pub stream_rank: u64,
477}
478
479#[cfg(not(target_arch = "wasm32"))]
480#[derive(Clone)]
481pub(crate) struct NativeControlStreamEntry {
482 pub owner: NativeControlStreamOwner,
483 pub endpoint_id: iroh::EndpointId,
484 pub send: Arc<tokio::sync::Mutex<iroh::endpoint::SendStream>>,
485}
486
487#[cfg(not(target_arch = "wasm32"))]
496#[derive(Debug, Clone, Copy)]
497pub(crate) struct NativeTransportProtocolActivity {
498 pub transport_stable_id: u64,
499 pub observed_at: std::time::Instant,
500}
501
502#[cfg(not(target_arch = "wasm32"))]
503#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
504pub(crate) struct NativeAdmissionStreamContracts {
505 pub inbound: Option<crate::native_protocol::TokenStreamContract>,
506 pub outbound: Option<crate::native_protocol::TokenStreamContract>,
507}
508
509#[cfg(not(target_arch = "wasm32"))]
510#[derive(Debug, Clone)]
511pub(crate) struct CachedManagedScopeTicket {
512 pub scope: crate::session_token::GrantScope,
513 pub token: String,
514 pub max_connections: u32,
515 pub compound_ticket: String,
516 pub iroh_ticket: String,
517}
518
519#[cfg(not(target_arch = "wasm32"))]
523#[derive(Debug)]
524pub(crate) struct ManagedScopeTicketCache {
525 tickets: HashMap<String, CachedManagedScopeTicket>,
526 pub(crate) persistent_load_allowed: bool,
527}
528
529#[cfg(not(target_arch = "wasm32"))]
530impl Default for ManagedScopeTicketCache {
531 fn default() -> Self {
532 Self {
533 tickets: HashMap::new(),
534 persistent_load_allowed: true,
535 }
536 }
537}
538
539#[cfg(not(target_arch = "wasm32"))]
540impl std::ops::Deref for ManagedScopeTicketCache {
541 type Target = HashMap<String, CachedManagedScopeTicket>;
542 fn deref(&self) -> &Self::Target {
543 &self.tickets
544 }
545}
546
547#[cfg(not(target_arch = "wasm32"))]
548impl std::ops::DerefMut for ManagedScopeTicketCache {
549 fn deref_mut(&mut self) -> &mut Self::Target {
550 &mut self.tickets
551 }
552}
553
554#[cfg(not(target_arch = "wasm32"))]
555#[derive(Debug, Clone)]
556pub(crate) struct NativeRouteRepairCredential {
557 pub token: String,
558 pub token_payload: String,
559 pub scope: crate::session_token::GrantScope,
561 pub authoritative_device_id: String,
562}
563
564#[derive(Debug, Clone, PartialEq, Eq)]
571pub(crate) struct ScopedRouteRepairCredential {
572 pub token: String,
573 pub scope: String,
574 pub max_connections: u32,
575 pub expires_at_ms: Option<u64>,
576 pub authoritative_device_id: String,
577}
578
579#[allow(dead_code)]
584#[derive(Clone)]
585pub struct Client {
586 app_tag: String,
587 #[cfg(target_arch = "wasm32")]
588 pub(crate) wasm_connection_state_owner_id: String,
589 broadcasts: crate::broadcast::Broadcasts,
590 signaling: Arc<dyn SignalingBackend>,
591 pub room: Arc<dyn RoomBackend>,
592 node_id: Arc<RwLock<Option<String>>>,
593 pub iroh_endpoint: Arc<RwLock<Option<Endpoint>>>,
594 #[cfg(target_arch = "wasm32")]
595 pub iroh_node: Arc<RwLock<Option<IrohWasmNode>>>,
596 #[cfg(not(target_arch = "wasm32"))]
597 pub iroh_node: Arc<RwLock<Option<IrohNativeNode>>>,
598 #[cfg(not(target_arch = "wasm32"))]
599 native_application_streams: async_channel::Sender<IncomingStream>,
600 #[cfg(not(target_arch = "wasm32"))]
601 native_application_streams_receiver: async_channel::Receiver<IncomingStream>,
602 pub connection_manager: Arc<crate::connection_manager::ConnectionManager>,
603 #[cfg(not(target_arch = "wasm32"))]
604 native_device_identity: Arc<RwLock<Option<crate::native_device::NativeDeviceIdentity>>>,
605 #[cfg(not(target_arch = "wasm32"))]
606 native_device_base_dir: Arc<RwLock<Option<std::path::PathBuf>>>,
607 #[cfg(not(target_arch = "wasm32"))]
608 native_device_identity_init_guard: Arc<tokio::sync::Mutex<()>>,
609 #[cfg(not(target_arch = "wasm32"))]
610 native_device_updates:
611 tokio::sync::broadcast::Sender<crate::native_device::NativeDeviceIdentity>,
612 #[cfg(not(target_arch = "wasm32"))]
613 native_connection_state_updates: tokio::sync::broadcast::Sender<StateSnapshot>,
614 #[cfg(not(target_arch = "wasm32"))]
615 native_peer_data_updates: tokio::sync::broadcast::Sender<NativePeerDataEvent>,
616 auto_connect_loop_key: Arc<Mutex<Option<(String, String)>>>,
617 auto_connect_generation: Arc<AtomicU64>,
618 #[cfg(not(target_arch = "wasm32"))]
622 native_auto_connect_wake: Arc<tokio::sync::Notify>,
623 #[cfg(not(target_arch = "wasm32"))]
626 external_desired_peer_actor:
627 Arc<tokio::sync::Mutex<auto_connect_impl::NativeExternalAutoConnectActorState>>,
628 auto_connect_exclusion_owner: Arc<Mutex<()>>,
632 auto_connect_excluded: Arc<Mutex<HashSet<String>>>,
633 auto_connect_peer_requested_excluded: Arc<Mutex<HashSet<String>>>,
640 auto_connect_excluded_node_aliases: Arc<Mutex<HashMap<String, HashSet<String>>>>,
644 pub session_token_registry: Arc<crate::session_token::SessionTokenRegistry>,
648 #[cfg(not(target_arch = "wasm32"))]
652 inbound_session_admission_transport_ids: Arc<StdRwLock<HashMap<String, u64>>>,
653 #[cfg(not(target_arch = "wasm32"))]
656 offline_admission_requirements:
657 Arc<StdRwLock<HashMap<String, crate::offline::OfflineCandidateHandoff>>>,
658 #[cfg(not(target_arch = "wasm32"))]
660 offline_admission_transport_ids: Arc<StdRwLock<HashMap<String, u64>>>,
661 #[cfg(not(target_arch = "wasm32"))]
664 pub(crate) offline_runtime:
665 Arc<tokio::sync::Mutex<Option<crate::offline::InstalledOfflineRuntime>>>,
666 #[cfg(not(target_arch = "wasm32"))]
669 pub(crate) offline_proof_attempts: Arc<tokio::sync::Mutex<HashMap<String, u64>>>,
670 #[cfg(not(target_arch = "wasm32"))]
674 native_admission_stream_contracts:
675 Arc<StdRwLock<HashMap<String, NativeAdmissionStreamContracts>>>,
676 remote_session_admission_proofs: Arc<StdRwLock<RemoteSessionAdmissionProofs>>,
680 pending_inline_reciprocal_admissions:
685 Arc<StdRwLock<HashMap<String, PendingInlineReciprocalAdmission>>>,
686 outbound_application_security_epoch_fingerprints: Arc<StdRwLock<HashMap<String, String>>>,
691 #[cfg(not(target_arch = "wasm32"))]
695 pending_reciprocal_session_admission_requests: Arc<StdRwLock<HashSet<String>>>,
696 #[cfg(not(target_arch = "wasm32"))]
700 native_route_repair_credentials: Arc<StdRwLock<HashMap<String, NativeRouteRepairCredential>>>,
701 scoped_route_repair_credentials: Arc<StdRwLock<HashMap<String, ScopedRouteRepairCredential>>>,
705 connection_application_crypto_keys:
706 Arc<StdRwLock<HashMap<String, [u8; crate::application_crypto::APPLICATION_KEY_BYTES]>>>,
707 connection_application_crypto_state: Arc<StdRwLock<()>>,
712 connection_application_crypto_required: Arc<StdRwLock<HashSet<String>>>,
713 #[cfg(not(target_arch = "wasm32"))]
717 trusted_user_device_application_crypto_required: Arc<AtomicBool>,
718 connection_application_crypto_confirmed: Arc<
722 StdRwLock<
723 HashMap<
724 String,
725 crate::client::application_crypto_impl::ConnectionApplicationCryptoConfirmation,
726 >,
727 >,
728 >,
729 #[cfg(not(target_arch = "wasm32"))]
736 connection_application_route_updates: Arc<tokio::sync::Notify>,
737 connection_application_crypto_outbound_sequences: Arc<StdRwLock<HashMap<String, u64>>>,
738 connection_application_key_agreements:
739 Arc<StdRwLock<HashMap<String, crate::key_agreement::KeyAgreement>>>,
740 #[cfg(not(target_arch = "wasm32"))]
749 managed_scope_tickets: Arc<StdRwLock<ManagedScopeTicketCache>>,
750 known_endpoint_addrs: Arc<RwLock<HashMap<String, iroh::EndpointAddr>>>,
756 known_device_ids_by_node: Arc<StdRwLock<HashMap<String, String>>>,
761 known_device_endpoint_revision: tokio::sync::watch::Sender<u64>,
765 sparse_fanout: Arc<tokio::sync::Mutex<crate::sparse_fanout::SparseFanoutState>>,
769 app_backgrounded: Arc<AtomicBool>,
773 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
774 webrtc_availability: Arc<Mutex<crate::iroh_carrier_bootstrap::WebRtcAvailability>>,
775 background_execution_allowed: Arc<AtomicBool>,
776 #[cfg(target_arch = "wasm32")]
777 wasm_accept_bridge_started: Arc<std::sync::atomic::AtomicBool>,
778 #[cfg(target_arch = "wasm32")]
779 last_emitted_connection_states:
780 Arc<Mutex<std::collections::HashMap<String, WasmConnectionStateFingerprint>>>,
781 #[cfg(target_arch = "wasm32")]
782 last_empty_peer_sessions_warning_ms: Arc<AtomicU64>,
783 iroh_init_guard: Arc<tokio::sync::Mutex<()>>,
788 #[cfg(not(target_arch = "wasm32"))]
791 native_ip_exclusions: Arc<RwLock<Option<Vec<ipnet::IpNet>>>>,
792 #[cfg(all(not(target_arch = "wasm32"), feature = "native-udp-socket-initializer"))]
793 native_udp_socket_initializer: Option<NativeUdpSocketInitializer>,
794 #[cfg(all(
795 not(target_arch = "wasm32"),
796 feature = "native-relay-tcp-socket-initializer"
797 ))]
798 native_relay_tcp_socket_initializer: Option<NativeRelayTcpSocketInitializer>,
799 #[cfg(all(not(target_arch = "wasm32"), feature = "native-dns-socket-initializer"))]
800 native_dns_socket_initializer: Option<NativeDnsSocketInitializer>,
801 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
805 iroh_packet_carriers: Arc<
806 RwLock<
807 HashMap<
808 crate::iroh_carrier_kind::IrohCarrierKind,
809 Arc<crate::packet_carrier_transport::PacketCarrierAddressProvider>,
810 >,
811 >,
812 >,
813 managed_connect_gates: Arc<tokio::sync::Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
820 #[cfg(not(target_arch = "wasm32"))]
824 iroh_path_watcher_stable_ids: Arc<Mutex<HashSet<u64>>>,
825 #[cfg(not(target_arch = "wasm32"))]
831 native_custom_transport_kinds: Arc<RwLock<HashMap<u64, IrohPathKind>>>,
832 #[cfg(not(target_arch = "wasm32"))]
836 native_transport_upgrade_providers:
837 Arc<RwLock<HashMap<IrohPathKind, Arc<dyn UpgradeProvider>>>>,
838 #[cfg(target_arch = "wasm32")]
842 wasm_transport_upgrade_gates:
843 Arc<tokio::sync::Mutex<HashMap<(String, IrohPathKind), WasmTransportUpgradeGate>>>,
844 #[cfg(not(target_arch = "wasm32"))]
847 native_transport_upgrade_gates:
848 Arc<tokio::sync::Mutex<HashMap<(String, IrohPathKind), NativeTransportUpgradeGate>>>,
849 #[cfg(not(target_arch = "wasm32"))]
853 native_ble_upgrade_attempts:
854 Arc<tokio::sync::Mutex<HashMap<String, NativeBleUpgradeAttemptState>>>,
855 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
856 native_webrtc_carrier_attempts: Arc<RwLock<HashMap<String, Arc<NativeWebRtcCarrierAttempt>>>>,
857 #[cfg(all(
862 not(target_arch = "wasm32"),
863 feature = "iroh-carrier-core",
864 any(feature = "test-harness", feature = "testing-endpoints")
865 ))]
866 native_iroh_carrier_debug_events: Arc<StdRwLock<std::collections::VecDeque<serde_json::Value>>>,
867 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
868 native_moq_carrier_attempts: Arc<RwLock<HashMap<String, Arc<NativeMoqCarrierAttempt>>>>,
869 #[cfg(not(target_arch = "wasm32"))]
873 native_peer_transport_capabilities:
874 Arc<RwLock<HashMap<String, HashSet<NativePeerTransportCapability>>>>,
875 pub(crate) presence_loop_tx:
876 Arc<Mutex<Option<tokio::sync::mpsc::Sender<crate::presence::PresenceCommand>>>>,
877 #[cfg(not(target_arch = "wasm32"))]
881 pub(crate) project_id: String,
882 #[cfg(not(target_arch = "wasm32"))]
883 pub(crate) token_provider: Arc<dyn Fn() -> Option<String> + Send + Sync>,
884 transport_config: Arc<RwLock<TransportConfig>>,
885 iroh_carrier_policy_epoch: Arc<AtomicU64>,
888 iroh_carrier_peer_policy_epochs: Arc<StdRwLock<HashMap<String, u64>>>,
892 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
893 pub(crate) local_discovery_registry: crate::local_discovery::LocalDiscoveryRegistry,
894 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
895 mdns_address_lookup: Arc<RwLock<Option<iroh_mdns_address_lookup::MdnsAddressLookup>>>,
896 #[cfg(not(target_arch = "wasm32"))]
902 pub(crate) native_control_streams:
903 Arc<tokio::sync::Mutex<HashMap<String, NativeControlStreamEntry>>>,
904 #[cfg(not(target_arch = "wasm32"))]
908 native_transport_protocol_activity:
909 Arc<StdRwLock<HashMap<String, NativeTransportProtocolActivity>>>,
910 #[cfg(all(not(target_arch = "wasm32"), feature = "experimental-scoped-actor"))]
914 scoped_connection_actor_registry:
915 Arc<RwLock<Option<Arc<crate::client::scoped_connection_actor::ConnectionActors>>>>,
916 #[cfg(not(target_arch = "wasm32"))]
922 auth_readiness: Arc<crate::client::auth_readiness::AuthReadinessStore>,
923 #[cfg(not(target_arch = "wasm32"))]
926 scope_classifier: Arc<RwLock<Arc<dyn ScopeClassifier>>>,
927}
928
929#[derive(Debug, Clone, PartialEq, Eq)]
930struct RemoteSessionAdmissionProof {
931 transport_stable_id: u64,
932 token_fingerprint: String,
933 approval_scope: String,
934}
935
936#[derive(Default)]
940struct RemoteSessionAdmissionProofs {
941 connections: HashMap<String, (String, HashMap<String, RemoteSessionAdmissionProof>)>,
942}
943
944impl RemoteSessionAdmissionProofs {
945 fn new() -> Self {
946 Self::default()
947 }
948 fn get(&self, connection: &str) -> Option<&RemoteSessionAdmissionProof> {
949 let (current, proofs) = self.connections.get(connection)?;
950 proofs.get(current)
951 }
952 fn get_mut(&mut self, connection: &str) -> Option<&mut RemoteSessionAdmissionProof> {
953 let (current, proofs) = self.connections.get_mut(connection)?;
954 proofs.get_mut(current)
955 }
956 fn get_for_token(
957 &self,
958 connection: &str,
959 fingerprint: &str,
960 ) -> Option<&RemoteSessionAdmissionProof> {
961 self.connections
962 .get(connection)?
963 .1
964 .values()
965 .find(|proof| proof.token_fingerprint == fingerprint)
966 }
967 fn insert(&mut self, connection: String, proof: RemoteSessionAdmissionProof) {
968 let scope = proof.approval_scope.clone();
969 let (current, proofs) = self.connections.entry(connection).or_default();
970 *current = scope.clone();
971 proofs.insert(scope, proof);
972 }
973 fn contains_key(&self, connection: &str) -> bool {
974 self.connections.contains_key(connection)
975 }
976 fn remove(&mut self, connection: &str) -> Option<RemoteSessionAdmissionProof> {
977 let (current, mut proofs) = self.connections.remove(connection)?;
978 proofs.remove(¤t)
979 }
980 fn clear(&mut self) {
981 self.connections.clear();
982 }
983 fn remove_scope(&mut self, scope: &str) {
984 self.connections.retain(|_, (current, proofs)| {
985 proofs.remove(scope);
986 if !proofs.contains_key(current) {
987 *current = proofs.keys().min().cloned().unwrap_or_default();
988 }
989 !proofs.is_empty()
990 });
991 }
992 fn remove_token(&mut self, fingerprint: &str) {
993 self.connections.retain(|_, (current, proofs)| {
994 proofs.retain(|_, proof| proof.token_fingerprint != fingerprint);
995 if !proofs.contains_key(current) {
996 *current = proofs.keys().min().cloned().unwrap_or_default();
997 }
998 !proofs.is_empty()
999 });
1000 }
1001}
1002
1003#[derive(Debug, Clone)]
1004struct PendingInlineReciprocalAdmission {
1005 connection_id: String,
1006 remote_node_id: String,
1007 local_device_id: String,
1008 remote_device_id: String,
1009 transport_stable_id: u64,
1010 stream_instance_id: String,
1011 presentation_id: String,
1012 token: String,
1013 token_fingerprint: String,
1014 expected_scope: String,
1015 inbound_admission_fingerprint: String,
1016 inbound_admission_epoch: u64,
1017}
1018
1019fn now_millis_i64() -> i64 {
1020 crate::coordination::now_millis_u64().min(i64::MAX as u64) as i64
1021}
1022
1023#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1033pub enum IrohPathKind {
1034 DirectQuic,
1036 DirectLan,
1038 Relay,
1040 Ble,
1042 WebRtc,
1044 Moq,
1046 Unknown,
1048}
1049
1050#[cfg(not(target_arch = "wasm32"))]
1051#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
1052pub(crate) enum NativePeerTransportCapability {
1053 WebRtc,
1054 Moq,
1055 Ble,
1056}
1057
1058#[cfg(not(target_arch = "wasm32"))]
1065#[async_trait::async_trait]
1066pub trait UpgradeProvider: std::fmt::Debug + Send + Sync {
1067 fn kind(&self) -> IrohPathKind;
1068 fn transport_id(&self) -> u64;
1069 async fn prepare_endpoint_addr(
1070 &self,
1071 endpoint_id: iroh::EndpointId,
1072 ) -> anyhow::Result<iroh::EndpointAddr>;
1073}
1074
1075impl IrohPathKind {
1076 pub fn is_relay_path(self) -> bool {
1077 matches!(self, Self::Relay)
1078 }
1079
1080 pub fn transport_label(self) -> &'static str {
1081 match self {
1082 Self::DirectQuic => crate::transport_label::IROH_QUIC,
1083 Self::DirectLan => crate::transport_label::IROH_LAN,
1084 Self::Relay => crate::transport_label::IROH_RELAY,
1085 Self::Ble => crate::transport_label::BLE,
1086 Self::WebRtc => crate::transport_label::WEBRTC,
1087 Self::Moq => crate::transport_label::MOQ,
1088 Self::Unknown => crate::transport_label::IROH,
1089 }
1090 }
1091}
1092
1093#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
1095#[serde(rename_all = "camelCase")]
1096pub struct IrohLanConfig {
1097 #[serde(default = "default_lan_enabled")]
1098 pub enabled: bool,
1099 #[serde(default = "default_lan_advertise")]
1101 pub advertise: bool,
1102}
1103
1104fn default_lan_enabled() -> bool {
1105 true
1106}
1107
1108fn default_lan_advertise() -> bool {
1109 true
1110}
1111
1112impl Default for IrohLanConfig {
1113 fn default() -> Self {
1114 Self {
1115 enabled: true,
1116 advertise: true,
1117 }
1118 }
1119}
1120
1121#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Default)]
1126#[serde(rename_all = "camelCase")]
1127pub struct BleConfig {
1128 #[serde(default)]
1129 pub enabled: bool,
1130 #[serde(default, skip_serializing_if = "Option::is_none")]
1131 pub connect_timeout_ms: Option<u64>,
1132}
1133
1134pub(crate) const MANAGED_SETTLE_DEADLINE_MS: i64 =
1139 crate::runtime_policy::MANAGED_SETTLE_DEADLINE_MS;
1140
1141#[cfg(not(target_arch = "wasm32"))]
1149pub(crate) const SESSION_ADMISSION_TIMEOUT_MS: u64 =
1150 crate::runtime_policy::SESSION_ADMISSION_TIMEOUT_MS;
1151
1152#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
1153#[serde(rename_all = "camelCase")]
1154pub enum ConnectionStatus {
1155 Disconnected,
1156 Connecting,
1157 Connected,
1158 Failed,
1159 Closed,
1160 Online,
1161}
1162
1163#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
1164#[serde(rename_all = "camelCase")]
1165pub enum DevicePresenceStatus {
1166 Online,
1167 Idle,
1168 Offline,
1169}
1170
1171#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
1172#[serde(rename_all = "camelCase")]
1173pub enum ReadinessState {
1174 Connecting,
1175 TransportOnly,
1176 Settling,
1177 Routable,
1178 AwaitingReplacement,
1179 Closed,
1180 Failed,
1181}
1182
1183#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
1184#[serde(rename_all = "camelCase")]
1185pub struct LatencySnapshot {
1186 pub webrtc: Option<u64>,
1187 pub iroh: Option<u64>,
1188 pub iroh_lan: Option<u64>,
1189 pub iroh_relay: Option<u64>,
1190 pub ble: Option<u64>,
1191 pub moq: Option<u64>,
1192}
1193
1194impl LatencySnapshot {
1195 fn set(&mut self, transport: &str, latency_ms: u64) {
1196 match transport.trim().to_ascii_lowercase().as_str() {
1197 "webrtc" => self.webrtc = Some(latency_ms),
1198 "iroh-lan" => self.iroh_lan = Some(latency_ms),
1199 "iroh-relay" => self.iroh_relay = Some(latency_ms),
1200 "iroh" | "iroh-quic" => self.iroh = Some(latency_ms),
1201 "ble" => self.ble = Some(latency_ms),
1202 "moq" => self.moq = Some(latency_ms),
1203 _ => {}
1204 }
1205 }
1206
1207 fn get(&self, transport: &str) -> Option<u64> {
1208 match transport.trim().to_ascii_lowercase().as_str() {
1209 "webrtc" => self.webrtc,
1210 "iroh-lan" => self.iroh_lan,
1211 "iroh-relay" => self.iroh_relay,
1212 "iroh" | "iroh-quic" => self.iroh,
1213 "ble" => self.ble,
1214 "moq" => self.moq,
1215 _ => None,
1216 }
1217 }
1218}
1219
1220#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1221#[serde(rename_all = "camelCase")]
1222pub struct DeviceStatusSnapshot {
1223 #[serde(flatten)]
1224 pub device: crate::signaling::Device,
1225 pub presence_status: DevicePresenceStatus,
1226 pub presence_updated_at: Option<i64>,
1227 pub presence_expires_at: Option<i64>,
1228 pub connectable: bool,
1229 pub connection_status: ConnectionStatus,
1230 pub settled_ready: bool,
1231 pub readiness_state: ReadinessState,
1232 pub readiness_reason: String,
1233 pub peer_health: crate::connection_manager::ConnectionHealth,
1234 pub peer_id: Option<String>,
1235 pub scopes: Vec<String>,
1236 pub connection_id: Option<String>,
1237 pub device_id_hint: Option<String>,
1238 pub active_transport_stable_id: Option<u64>,
1239 pub transport_generation: u64,
1240 pub route_generation: u64,
1241 pub active_transport: String,
1242 pub parallel_transport: Option<String>,
1243 #[serde(default)]
1244 pub latency_ms: Option<u64>,
1245 #[serde(default)]
1246 pub latency_by_transport: LatencySnapshot,
1247}
1248
1249#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1250#[serde(rename_all = "camelCase")]
1251pub struct PeerSessionSnapshot {
1252 pub peer_id: String,
1253 pub device_id: Option<String>,
1254 pub device_id_hint: Option<String>,
1255 pub node_id: Option<String>,
1256 pub active_connection_id: Option<String>,
1257 pub candidate_connection_ids: Vec<String>,
1258 pub status: crate::connection_manager::ConnectionState,
1259 pub health: crate::connection_manager::ConnectionHealth,
1260 pub settled_ready: bool,
1261 #[serde(default)]
1262 pub logical_session_terminal: bool,
1263 pub readiness_state: ReadinessState,
1264 pub active_transport_stable_id: Option<u64>,
1265 pub transport_generation: u64,
1266 pub route_generation: u64,
1267 pub active_transport: String,
1268 pub parallel_transport: Option<String>,
1269 pub replacement_pending: bool,
1270 pub last_lifecycle_transition_at_ms: i64,
1271 pub readiness_reason: String,
1272 pub transition_count: u64,
1273 pub connecting_transition_count: u64,
1274 pub replacement_count: u64,
1275 pub retire_count: u64,
1276 pub last_disconnect_reason: Option<String>,
1277 pub last_reconnect_reason: Option<String>,
1278 pub scopes: Vec<String>,
1279 pub last_seen_at_ms: i64,
1280 pub error: Option<String>,
1281}
1282
1283#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1284#[serde(rename_all = "camelCase")]
1285pub struct ManagedConnectResult {
1286 pub connection_id: String,
1287 pub device_id: Option<String>,
1288 pub device_id_hint: Option<String>,
1289 pub remote_node_id: String,
1290 pub state: String,
1291 #[serde(default, skip_serializing_if = "Option::is_none")]
1292 pub approved_scope: Option<String>,
1293}
1294
1295#[cfg(not(target_arch = "wasm32"))]
1296#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1297#[serde(rename_all = "camelCase")]
1298pub struct ConnectionAdoption {
1299 pub connection_id: String,
1300 pub node_id: String,
1301 pub device_id: Option<String>,
1302 pub transport_generation: u64,
1303 pub status_reason: Option<String>,
1304 pub main_stream_ready: bool,
1305}
1306
1307#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
1308#[serde(rename_all = "camelCase")]
1309pub enum ManagedHealth {
1310 Healthy,
1311 AwaitingReplacement,
1312 Dead,
1313}
1314
1315#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1316#[serde(rename_all = "camelCase")]
1317pub struct ConnectionHealthView {
1318 pub connection_id: String,
1319 pub device_id: Option<String>,
1320 pub device_id_hint: Option<String>,
1321 pub node_id: Option<String>,
1322 pub active_transport_stable_id: Option<u64>,
1323 pub transport_generation: u64,
1324 pub route_generation: u64,
1325 pub status: ManagedHealth,
1326 pub settled_ready: bool,
1327 pub readiness_state: ReadinessState,
1328 pub replacement_pending: bool,
1329 pub last_lifecycle_transition_at_ms: i64,
1330 pub readiness_reason: String,
1331 pub transition_count: u64,
1332 pub connecting_transition_count: u64,
1333 pub replacement_count: u64,
1334 pub retire_count: u64,
1335 pub last_disconnect_reason: Option<String>,
1336 pub last_reconnect_reason: Option<String>,
1337}
1338
1339#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
1340#[serde(rename_all = "camelCase")]
1341pub struct StateSnapshot {
1342 pub connection_id: String,
1343 pub device_id: Option<String>,
1344 pub device_id_hint: Option<String>,
1345 pub remote_node_id: Option<String>,
1346 pub state: String,
1347 pub transport_state: String,
1348 pub protocol_state: String,
1349 pub routable: bool,
1350 #[serde(default)]
1351 pub logical_session_terminal: bool,
1352 pub readiness_state: ReadinessState,
1353 pub readiness_reason: String,
1354 pub transport_generation: u64,
1355 pub route_generation: u64,
1356 pub active_transport_stable_id: Option<u64>,
1357 pub active_transport: String,
1358 pub parallel_transport: Option<String>,
1359 #[serde(default)]
1361 pub latency_ms: Option<u64>,
1362 #[serde(default)]
1363 pub carrier: Option<String>,
1364 #[serde(default)]
1365 pub metrics_sampled_at_ms: Option<i64>,
1366 #[serde(default)]
1368 pub moq_object_quantities: Option<crate::iroh_carrier_kind::MoqObjectQuantities>,
1369 pub replacement_in_progress: bool,
1370 pub last_lifecycle_transition_at_ms: i64,
1371 pub transition_count: u64,
1372 pub connecting_transition_count: u64,
1373 pub replacement_count: u64,
1374 pub retire_count: u64,
1375 pub last_disconnect_reason: Option<String>,
1376 pub last_reconnect_reason: Option<String>,
1377 #[serde(default)]
1383 pub scopes: Vec<String>,
1384 pub error: Option<String>,
1385 pub created_at: i64,
1386 pub updated_at: i64,
1387}
1388
1389#[cfg(target_arch = "wasm32")]
1390#[derive(Debug, Clone, PartialEq, Eq)]
1391struct WasmConnectionStateFingerprint {
1392 connection_id: String,
1393 device_id: Option<String>,
1394 device_id_hint: Option<String>,
1395 remote_node_id: Option<String>,
1396 state: String,
1397 transport_state: String,
1398 protocol_state: String,
1399 routable: bool,
1400 logical_session_terminal: bool,
1401 transport_generation: u64,
1402 route_generation: u64,
1403 active_transport_stable_id: Option<u64>,
1404 active_transport: String,
1405 parallel_transport: Option<String>,
1406 replacement_in_progress: bool,
1407 transition_count: u64,
1408 connecting_transition_count: u64,
1409 replacement_count: u64,
1410 retire_count: u64,
1411 last_disconnect_reason: Option<String>,
1412 last_reconnect_reason: Option<String>,
1413 scopes: Vec<String>,
1414 error: Option<String>,
1415}
1416
1417#[cfg(target_arch = "wasm32")]
1418impl From<&StateSnapshot> for WasmConnectionStateFingerprint {
1419 fn from(snapshot: &StateSnapshot) -> Self {
1420 Self {
1421 connection_id: snapshot.connection_id.clone(),
1422 device_id: snapshot.device_id.clone(),
1423 device_id_hint: snapshot.device_id_hint.clone(),
1424 remote_node_id: snapshot.remote_node_id.clone(),
1425 state: snapshot.state.clone(),
1426 transport_state: snapshot.transport_state.clone(),
1427 protocol_state: snapshot.protocol_state.clone(),
1428 routable: snapshot.routable,
1429 logical_session_terminal: snapshot.logical_session_terminal,
1430 transport_generation: snapshot.transport_generation,
1431 route_generation: snapshot.route_generation,
1432 active_transport_stable_id: snapshot.active_transport_stable_id,
1433 active_transport: snapshot.active_transport.clone(),
1434 parallel_transport: snapshot.parallel_transport.clone(),
1435 replacement_in_progress: snapshot.replacement_in_progress,
1436 transition_count: snapshot.transition_count,
1437 connecting_transition_count: snapshot.connecting_transition_count,
1438 replacement_count: snapshot.replacement_count,
1439 retire_count: snapshot.retire_count,
1440 last_disconnect_reason: snapshot.last_disconnect_reason.clone(),
1441 last_reconnect_reason: snapshot.last_reconnect_reason.clone(),
1442 scopes: snapshot.scopes.clone(),
1443 error: snapshot.error.clone(),
1444 }
1445 }
1446}
1447
1448#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
1449#[serde(rename_all = "camelCase", tag = "kind")]
1450pub enum BridgeAction {
1451 Healthy,
1452 AwaitReplacement {
1453 reason: String,
1454 },
1455 Rebind {
1456 remote_node_id: Option<String>,
1457 transport_generation: u64,
1458 },
1459 Retire {
1460 reason: String,
1461 },
1462}
1463
1464#[cfg(not(target_arch = "wasm32"))]
1465#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1466enum DuplicateClosedHandling {
1467 RebindKeptTransport,
1468 RetireClosedTransport,
1469}
1470
1471#[cfg(not(target_arch = "wasm32"))]
1472#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1473pub(crate) enum IncomingTransportCloseResolution {
1474 PreservedKeptTransport,
1475 RetiredClosedTransport,
1476}
1477
1478#[cfg(not(target_arch = "wasm32"))]
1479#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1480enum AutoConnectTieBreakDecision {
1481 ObserveConnectedTransport,
1482 WaitForInitiator,
1483 ActAsInitiator,
1484}
1485
1486#[cfg(not(target_arch = "wasm32"))]
1487fn duplicate_closed_handling(
1488 close_reason_debug: &str,
1489 kept_transport_alive: bool,
1490 kept_transport_healthy: bool,
1491 kept_stable_id: u64,
1492 closed_stable_id: u64,
1493) -> DuplicateClosedHandling {
1494 let _ = close_reason_debug;
1495 let _ = kept_transport_healthy;
1496 if kept_stable_id != closed_stable_id && kept_transport_alive {
1497 DuplicateClosedHandling::RebindKeptTransport
1498 } else {
1499 DuplicateClosedHandling::RetireClosedTransport
1500 }
1501}
1502
1503#[cfg(not(target_arch = "wasm32"))]
1504fn auto_connect_tie_break_decision(
1505 local_node_id: &str,
1506 remote_node_id: &str,
1507 transport_connected: bool,
1508 waited_ms: i64,
1509 initiator_grace_ms: i64,
1510) -> AutoConnectTieBreakDecision {
1511 if transport_connected {
1512 return AutoConnectTieBreakDecision::ObserveConnectedTransport;
1513 }
1514 if local_node_id <= remote_node_id {
1515 if waited_ms >= initiator_grace_ms {
1518 AutoConnectTieBreakDecision::ActAsInitiator
1519 } else {
1520 AutoConnectTieBreakDecision::WaitForInitiator
1521 }
1522 } else {
1523 AutoConnectTieBreakDecision::ActAsInitiator
1524 }
1525}
1526
1527fn normalize_lookup_id(value: Option<&str>) -> Option<String> {
1528 let trimmed = value?.trim();
1529 if trimmed.is_empty() {
1530 None
1531 } else {
1532 Some(trimmed.to_ascii_lowercase())
1533 }
1534}
1535
1536fn peer_snapshot_lookup_aliases(peer: &crate::connection_manager::PeerSnapshot) -> Vec<String> {
1537 [
1538 normalize_lookup_id(peer.device_id.as_deref()),
1539 normalize_lookup_id(peer.device_id_hint.as_deref()),
1540 normalize_lookup_id(peer.node_id.as_deref()),
1541 normalize_lookup_id(Some(peer.peer_id.as_str())),
1542 ]
1543 .into_iter()
1544 .flatten()
1545 .collect()
1546}
1547
1548fn peer_snapshot_matches_any_alias(
1549 peer: &crate::connection_manager::PeerSnapshot,
1550 aliases: &std::collections::HashSet<String>,
1551) -> bool {
1552 peer_snapshot_lookup_aliases(peer)
1553 .into_iter()
1554 .any(|alias| aliases.contains(&alias))
1555}
1556
1557fn preferred_peer_snapshot(
1558 current: Option<crate::connection_manager::PeerSnapshot>,
1559 candidate: crate::connection_manager::PeerSnapshot,
1560) -> crate::connection_manager::PeerSnapshot {
1561 match current {
1562 Some(existing) if !prefer_device_status_peer_snapshot(&candidate, &existing) => existing,
1563 _ => candidate,
1564 }
1565}
1566
1567fn snapshot_connection_status(
1568 device: &crate::signaling::Device,
1569 peer: Option<&crate::connection_manager::PeerSnapshot>,
1570) -> ConnectionStatus {
1571 match peer.map(|value| &value.status) {
1572 Some(crate::connection_manager::ConnectionState::Pending)
1573 | Some(crate::connection_manager::ConnectionState::Connecting) => {
1574 ConnectionStatus::Connecting
1575 }
1576 Some(crate::connection_manager::ConnectionState::Connected)
1577 if peer.map(peer_snapshot_settled_ready).unwrap_or(false) =>
1578 {
1579 ConnectionStatus::Connected
1580 }
1581 Some(crate::connection_manager::ConnectionState::Connected) => ConnectionStatus::Connecting,
1582 Some(crate::connection_manager::ConnectionState::Failed) => ConnectionStatus::Failed,
1583 Some(crate::connection_manager::ConnectionState::Closed)
1584 | Some(crate::connection_manager::ConnectionState::Closing) => ConnectionStatus::Closed,
1585 None if device.online => ConnectionStatus::Online,
1586 None => ConnectionStatus::Disconnected,
1587 }
1588}
1589
1590fn snapshot_settled_ready(
1591 peer: Option<&crate::connection_manager::PeerSnapshot>,
1592 connection_status: &ConnectionStatus,
1593) -> bool {
1594 matches!(connection_status, ConnectionStatus::Connected)
1595 && peer.map(peer_snapshot_settled_ready).unwrap_or(false)
1596}
1597
1598fn device_presence_updated_at(device: &crate::signaling::Device) -> Option<i64> {
1599 [
1600 device.last_seen_at.as_ref(),
1601 device.updated_at.as_ref(),
1602 device.created_at.as_ref(),
1603 ]
1604 .into_iter()
1605 .flatten()
1606 .filter_map(|value| crate::presence_policy::parse_millis(Some(value)))
1607 .max()
1608}
1609
1610fn device_presence_expires_at(device: &crate::signaling::Device) -> Option<i64> {
1611 crate::presence_policy::parse_millis(device.expires_at.as_ref())
1612}
1613
1614fn device_presence_status(
1615 device: &crate::signaling::Device,
1616 presence_updated_at: Option<i64>,
1617 now_ms: i64,
1618) -> DevicePresenceStatus {
1619 if !device.online {
1620 return DevicePresenceStatus::Offline;
1621 }
1622
1623 if let Some(expires_at) = device_presence_expires_at(device) {
1626 return if expires_at > now_ms {
1627 DevicePresenceStatus::Online
1628 } else {
1629 DevicePresenceStatus::Idle
1630 };
1631 }
1632
1633 if presence_updated_at
1634 .map(|updated_at| {
1635 updated_at.saturating_add(crate::presence_policy::DEVICE_STALE_HEARTBEAT_MS) <= now_ms
1636 })
1637 .unwrap_or(false)
1638 {
1639 DevicePresenceStatus::Idle
1640 } else {
1641 DevicePresenceStatus::Online
1642 }
1643}
1644
1645fn device_connectable(
1646 device: &crate::signaling::Device,
1647 presence_status: &DevicePresenceStatus,
1648) -> bool {
1649 matches!(presence_status, DevicePresenceStatus::Online)
1650 && device
1651 .ticket
1652 .as_deref()
1653 .map(str::trim)
1654 .is_some_and(|value| !value.is_empty())
1655}
1656
1657fn peer_snapshot_settled_ready(peer: &crate::connection_manager::PeerSnapshot) -> bool {
1658 matches!(peer_readiness_state(peer), ReadinessState::Routable)
1659}
1660
1661fn peer_has_live_transport(peer: &crate::connection_manager::PeerSnapshot) -> bool {
1662 !peer.connection_ids.is_empty() && peer.active_transport_stable_id.is_some()
1663}
1664
1665fn peer_is_replacement_handoff(peer: &crate::connection_manager::PeerSnapshot) -> bool {
1666 peer.active_transport_stable_id.is_none()
1667 && peer.error.as_deref() == Some(crate::lifecycle_reason::REASON_REPLACEMENT_IN_PROGRESS)
1668}
1669
1670fn peer_readiness_state(peer: &crate::connection_manager::PeerSnapshot) -> ReadinessState {
1671 match peer.status {
1672 crate::connection_manager::ConnectionState::Pending
1673 | crate::connection_manager::ConnectionState::Connecting => ReadinessState::Connecting,
1674 crate::connection_manager::ConnectionState::Connected => {
1675 let connection_ids_present = !peer.connection_ids.is_empty();
1679 let healthy_with_connections = connection_ids_present
1680 && matches!(
1681 peer.health,
1682 crate::connection_manager::ConnectionHealth::Healthy
1683 );
1684 if healthy_with_connections {
1685 ReadinessState::Routable
1686 } else if peer_is_replacement_handoff(peer) {
1687 ReadinessState::AwaitingReplacement
1688 } else if !peer_has_live_transport(peer) {
1689 ReadinessState::TransportOnly
1690 } else if now_millis_i64().saturating_sub(peer.last_lifecycle_transition_at_ms)
1691 >= MANAGED_SETTLE_DEADLINE_MS
1692 {
1693 ReadinessState::AwaitingReplacement
1694 } else {
1695 ReadinessState::Settling
1696 }
1697 }
1698 crate::connection_manager::ConnectionState::Closing
1699 | crate::connection_manager::ConnectionState::Closed => ReadinessState::Closed,
1700 crate::connection_manager::ConnectionState::Failed => ReadinessState::Failed,
1701 }
1702}
1703
1704fn peer_readiness_reason(peer: &crate::connection_manager::PeerSnapshot) -> String {
1705 match peer_readiness_state(peer) {
1706 ReadinessState::Connecting => "waiting-for-transport".to_string(),
1707 ReadinessState::TransportOnly => {
1708 "transport-attached-awaiting-readiness-confirmation".to_string()
1709 }
1710 ReadinessState::Settling => "transport-connected-health-check-pending".to_string(),
1711 ReadinessState::Routable => "transport-healthy".to_string(),
1712 ReadinessState::AwaitingReplacement => {
1713 if peer.active_transport_stable_id.is_none() {
1714 "missing-active-transport".to_string()
1715 } else {
1716 "readiness-confirmation-timed-out".to_string()
1717 }
1718 }
1719 ReadinessState::Closed => "transport-closed".to_string(),
1720 ReadinessState::Failed => "connection-failed".to_string(),
1721 }
1722}
1723
1724fn connection_readiness_state(
1725 record: &crate::connection_manager::ConnectionRecord,
1726 peer_snapshot: Option<&crate::connection_manager::PeerSnapshot>,
1727) -> ReadinessState {
1728 match record.state {
1729 crate::connection_manager::ConnectionState::Pending
1730 | crate::connection_manager::ConnectionState::Connecting => ReadinessState::Connecting,
1731 crate::connection_manager::ConnectionState::Connected => {
1732 if peer_snapshot
1733 .map(peer_snapshot_settled_ready)
1734 .unwrap_or(false)
1735 {
1736 ReadinessState::Routable
1737 } else if record.transport_stable_id.is_none() {
1738 ReadinessState::AwaitingReplacement
1739 } else if now_millis_i64().saturating_sub(record.last_transport_change_at_ms)
1740 >= MANAGED_SETTLE_DEADLINE_MS
1741 {
1742 ReadinessState::AwaitingReplacement
1743 } else {
1744 ReadinessState::Settling
1745 }
1746 }
1747 crate::connection_manager::ConnectionState::Closing
1748 | crate::connection_manager::ConnectionState::Closed => ReadinessState::Closed,
1749 crate::connection_manager::ConnectionState::Failed => ReadinessState::Failed,
1750 }
1751}
1752
1753fn connection_readiness_reason(
1754 record: &crate::connection_manager::ConnectionRecord,
1755 peer_snapshot: Option<&crate::connection_manager::PeerSnapshot>,
1756) -> String {
1757 match connection_readiness_state(record, peer_snapshot) {
1758 ReadinessState::Connecting => "waiting-for-transport".to_string(),
1759 ReadinessState::TransportOnly => {
1760 "transport-attached-awaiting-readiness-confirmation".to_string()
1761 }
1762 ReadinessState::Settling => "bounded-health-confirmation-pending".to_string(),
1763 ReadinessState::Routable => "transport-healthy".to_string(),
1764 ReadinessState::AwaitingReplacement => {
1765 if record.transport_stable_id.is_none() {
1766 "missing-active-transport".to_string()
1767 } else {
1768 "readiness-confirmation-timed-out".to_string()
1769 }
1770 }
1771 ReadinessState::Closed => "transport-closed".to_string(),
1772 ReadinessState::Failed => "connection-failed".to_string(),
1773 }
1774}
1775
1776fn public_peer_state(
1777 peer: &crate::connection_manager::PeerSnapshot,
1778) -> crate::connection_manager::ConnectionState {
1779 if matches!(
1780 peer.status,
1781 crate::connection_manager::ConnectionState::Connected
1782 ) && !peer_snapshot_settled_ready(peer)
1783 {
1784 crate::connection_manager::ConnectionState::Connecting
1785 } else {
1786 peer.status.clone()
1787 }
1788}
1789
1790fn peer_session_snapshot_from_peer(
1791 peer: crate::connection_manager::PeerSnapshot,
1792) -> PeerSessionSnapshot {
1793 let status = public_peer_state(&peer);
1794 let active_connection_id = peer.connection_ids.first().cloned();
1795 let candidate_connection_ids = if peer.connection_ids.len() > 1 {
1796 peer.connection_ids[1..].to_vec()
1797 } else {
1798 Vec::new()
1799 };
1800 let settled_ready = peer_snapshot_settled_ready(&peer);
1801 let readiness_state = peer_readiness_state(&peer);
1802 let readiness_reason = peer_readiness_reason(&peer);
1803 let crate::connection_manager::PeerSnapshot {
1804 peer_id,
1805 device_id,
1806 device_id_hint,
1807 node_id,
1808 active_transport_stable_id,
1809 active_transport_generation,
1810 active_route_generation,
1811 active_transport,
1812 parallel_transport,
1813 last_lifecycle_transition_at_ms,
1814 health,
1815 transition_count,
1816 connecting_transition_count,
1817 replacement_count,
1818 retire_count,
1819 last_disconnect_reason,
1820 last_reconnect_reason,
1821 logical_session_terminal,
1822 scopes,
1823 last_seen_at_ms,
1824 error,
1825 ..
1826 } = peer;
1827
1828 PeerSessionSnapshot {
1829 peer_id,
1830 device_id,
1831 device_id_hint,
1832 node_id,
1833 active_connection_id,
1834 candidate_connection_ids,
1835 status,
1836 health,
1837 settled_ready,
1838 logical_session_terminal,
1839 readiness_state: readiness_state.clone(),
1840 active_transport_stable_id,
1841 transport_generation: active_transport_generation,
1842 route_generation: active_route_generation,
1843 active_transport,
1844 parallel_transport,
1845 replacement_pending: matches!(readiness_state, ReadinessState::AwaitingReplacement),
1846 last_lifecycle_transition_at_ms,
1847 readiness_reason,
1848 transition_count,
1849 connecting_transition_count,
1850 replacement_count,
1851 retire_count,
1852 last_disconnect_reason,
1853 last_reconnect_reason,
1854 scopes,
1855 last_seen_at_ms,
1856 error,
1857 }
1858}
1859
1860fn connection_state_snapshot_from_parts(
1861 record: &crate::connection_manager::ConnectionRecord,
1862 peer_snapshot: Option<&crate::connection_manager::PeerSnapshot>,
1863) -> StateSnapshot {
1864 let settled_ready = peer_snapshot
1865 .map(peer_snapshot_settled_ready)
1866 .unwrap_or(false);
1867 let readiness_state = connection_readiness_state(record, peer_snapshot);
1868 let in_place_reconnect_close = matches!(
1885 record.state,
1886 crate::connection_manager::ConnectionState::Closing
1887 | crate::connection_manager::ConnectionState::Closed
1888 ) && crate::lifecycle_reason::LifecycleReasonCode::from_text(
1889 record.status_reason.as_deref(),
1890 )
1891 .map(|code| {
1892 matches!(
1893 code,
1894 crate::lifecycle_reason::LifecycleReasonCode::ReplacementInProgress
1895 ) || code.is_transient()
1896 })
1897 .unwrap_or(false);
1898 let state = match record.state {
1899 crate::connection_manager::ConnectionState::Pending => "connecting",
1900 crate::connection_manager::ConnectionState::Connecting => "connecting",
1901 crate::connection_manager::ConnectionState::Connected if settled_ready => "connected",
1902 crate::connection_manager::ConnectionState::Connected => "connecting",
1903 crate::connection_manager::ConnectionState::Failed => "failed",
1904 crate::connection_manager::ConnectionState::Closing
1905 | crate::connection_manager::ConnectionState::Closed
1906 if in_place_reconnect_close =>
1907 {
1908 "connecting"
1909 }
1910 crate::connection_manager::ConnectionState::Closing
1911 | crate::connection_manager::ConnectionState::Closed => "closed",
1912 };
1913 let transport_state = match record.state {
1914 crate::connection_manager::ConnectionState::Pending
1915 | crate::connection_manager::ConnectionState::Connecting => "connecting",
1916 crate::connection_manager::ConnectionState::Connected => "connected",
1917 crate::connection_manager::ConnectionState::Closing
1918 | crate::connection_manager::ConnectionState::Closed
1919 if in_place_reconnect_close =>
1920 {
1921 "connecting"
1922 }
1923 crate::connection_manager::ConnectionState::Failed
1924 | crate::connection_manager::ConnectionState::Closing
1925 | crate::connection_manager::ConnectionState::Closed => "closed",
1926 };
1927 let protocol_state = match record.state {
1928 crate::connection_manager::ConnectionState::Connected if settled_ready => "routable",
1929 crate::connection_manager::ConnectionState::Connected => "transport-only",
1930 crate::connection_manager::ConnectionState::Pending
1931 | crate::connection_manager::ConnectionState::Connecting => "connecting",
1932 crate::connection_manager::ConnectionState::Closing
1933 | crate::connection_manager::ConnectionState::Closed
1934 if in_place_reconnect_close =>
1935 {
1936 "connecting"
1937 }
1938 crate::connection_manager::ConnectionState::Failed
1939 | crate::connection_manager::ConnectionState::Closing
1940 | crate::connection_manager::ConnectionState::Closed => "closed",
1941 };
1942
1943 StateSnapshot {
1944 connection_id: record.connection_id.clone(),
1945 device_id: record
1946 .device_id
1947 .clone()
1948 .or_else(|| peer_snapshot.and_then(|snapshot| snapshot.device_id.clone())),
1949 device_id_hint: record
1950 .device_id_hint
1951 .clone()
1952 .or_else(|| peer_snapshot.and_then(|snapshot| snapshot.device_id_hint.clone())),
1953 remote_node_id: record.node_id.clone(),
1954 state: state.to_string(),
1955 transport_state: transport_state.to_string(),
1956 protocol_state: protocol_state.to_string(),
1957 routable: settled_ready,
1958 logical_session_terminal: record.logical_session_terminal()
1961 && peer_snapshot
1962 .map(|peer| {
1963 peer.logical_session_terminal
1964 && peer.connection_ids.first() == Some(&record.connection_id)
1965 && peer.active_transport_generation == record.transport_generation
1966 && peer.active_route_generation == record.route_generation
1967 && peer.active_transport_stable_id == record.transport_stable_id
1968 })
1969 .unwrap_or(false),
1970 readiness_state: readiness_state.clone(),
1971 readiness_reason: connection_readiness_reason(record, peer_snapshot),
1972 transport_generation: record.transport_generation,
1973 route_generation: record.route_generation,
1974 active_transport_stable_id: record.transport_stable_id,
1975 active_transport: record.active_transport.clone(),
1976 parallel_transport: record.parallel_transport.clone(),
1977 latency_ms: None,
1978 carrier: None,
1979 metrics_sampled_at_ms: None,
1980 moq_object_quantities: None,
1981 replacement_in_progress: matches!(readiness_state, ReadinessState::AwaitingReplacement),
1982 last_lifecycle_transition_at_ms: record
1983 .last_state_change_at_ms
1984 .max(record.last_transport_change_at_ms)
1985 .max(record.last_route_change_at_ms),
1986 transition_count: record.transition_count,
1987 connecting_transition_count: record.connecting_transition_count,
1988 replacement_count: record.replacement_count,
1989 retire_count: record.retire_count,
1990 last_disconnect_reason: record.last_disconnect_reason.clone(),
1991 last_reconnect_reason: record.last_reconnect_reason.clone(),
1992 scopes: peer_snapshot
1993 .map(|snapshot| snapshot.scopes.clone())
1994 .unwrap_or_default(),
1995 error: record.status_reason.clone(),
1996 created_at: record.created_at_ms,
1997 updated_at: record.updated_at_ms,
1998 }
1999}
2000
2001fn merge_device_status_snapshots(
2002 devices: Vec<crate::signaling::Device>,
2003 peers: Vec<crate::connection_manager::PeerSnapshot>,
2004) -> Vec<DeviceStatusSnapshot> {
2005 let mut peer_by_device: std::collections::HashMap<
2006 String,
2007 crate::connection_manager::PeerSnapshot,
2008 > = std::collections::HashMap::new();
2009
2010 for peer in peers {
2011 for alias in peer_snapshot_lookup_aliases(&peer) {
2012 let replace = match peer_by_device.get(&alias) {
2013 Some(existing) => prefer_device_status_peer_snapshot(&peer, existing),
2014 None => true,
2015 };
2016 if replace {
2017 peer_by_device.insert(alias, peer.clone());
2018 }
2019 }
2020 }
2021
2022 devices
2023 .into_iter()
2024 .map(|device| {
2025 let authoritative_node_id = normalize_lookup_id(device.node_id.as_deref());
2026 let lookup_aliases: Vec<String> = [
2027 normalize_lookup_id(Some(device.device_id.as_str())),
2028 authoritative_node_id.clone(),
2029 ]
2030 .into_iter()
2031 .flatten()
2032 .collect();
2033 let peer = lookup_aliases
2034 .iter()
2035 .filter_map(|key| peer_by_device.get(key).cloned())
2036 .filter(|candidate| {
2037 let candidate_node_id = normalize_lookup_id(candidate.node_id.as_deref());
2038 match (&authoritative_node_id, candidate_node_id) {
2039 (Some(expected), Some(actual)) => expected == &actual,
2040 _ => true,
2041 }
2042 })
2043 .fold(None, |current, candidate| {
2044 Some(preferred_peer_snapshot(current, candidate))
2045 });
2046 let connection_status = snapshot_connection_status(&device, peer.as_ref());
2047 let presence_updated_at = device_presence_updated_at(&device);
2048 let presence_expires_at = device_presence_expires_at(&device);
2049 let presence_status =
2050 device_presence_status(&device, presence_updated_at, now_millis_i64());
2051 let connectable = device_connectable(&device, &presence_status);
2052 let readiness_state = peer.as_ref().map(peer_readiness_state).unwrap_or_else(|| {
2053 if device.online {
2054 ReadinessState::Connecting
2055 } else {
2056 ReadinessState::Closed
2057 }
2058 });
2059 DeviceStatusSnapshot {
2060 presence_status,
2061 presence_updated_at,
2062 presence_expires_at,
2063 connectable,
2064 settled_ready: snapshot_settled_ready(peer.as_ref(), &connection_status),
2065 readiness_state: readiness_state.clone(),
2066 readiness_reason: peer.as_ref().map(peer_readiness_reason).unwrap_or_else(|| {
2067 if device.online {
2068 "device-online-awaiting-runtime-session".to_string()
2069 } else {
2070 "device-offline".to_string()
2071 }
2072 }),
2073 connection_status,
2074 peer_health: peer
2075 .as_ref()
2076 .map(|value| value.health.clone())
2077 .unwrap_or_else(|| {
2078 if device.online {
2079 crate::connection_manager::ConnectionHealth::Healthy
2080 } else {
2081 crate::connection_manager::ConnectionHealth::Unknown
2082 }
2083 }),
2084 peer_id: peer.as_ref().map(|value| value.peer_id.clone()),
2085 scopes: peer
2086 .as_ref()
2087 .map(|value| value.scopes.clone())
2088 .unwrap_or_default(),
2089 connection_id: peer
2090 .as_ref()
2091 .and_then(|value| value.connection_ids.first().cloned()),
2092 device_id_hint: peer.as_ref().and_then(|value| value.device_id_hint.clone()),
2093 active_transport_stable_id: peer
2094 .as_ref()
2095 .and_then(|value| value.active_transport_stable_id),
2096 transport_generation: peer
2097 .as_ref()
2098 .map(|value| value.active_transport_generation)
2099 .unwrap_or(0),
2100 route_generation: peer
2101 .as_ref()
2102 .map(|value| value.active_route_generation)
2103 .unwrap_or(0),
2104 active_transport: peer
2105 .as_ref()
2106 .map(|value| value.active_transport.clone())
2107 .unwrap_or_else(|| "iroh".to_string()),
2108 parallel_transport: peer
2109 .as_ref()
2110 .and_then(|value| value.parallel_transport.clone()),
2111 latency_ms: None,
2112 latency_by_transport: LatencySnapshot::default(),
2113 device,
2114 }
2115 })
2116 .collect()
2117}
2118
2119fn peer_snapshot_recency_ms(peer: &crate::connection_manager::PeerSnapshot) -> i64 {
2120 peer.last_seen_at_ms
2121 .max(peer.last_lifecycle_transition_at_ms)
2122}
2123
2124fn prefer_device_status_peer_snapshot(
2130 candidate: &crate::connection_manager::PeerSnapshot,
2131 existing: &crate::connection_manager::PeerSnapshot,
2132) -> bool {
2133 use crate::connection_manager::ConnectionState;
2134 let c = &candidate.status;
2135 let e = &existing.status;
2136 let c_fail = matches!(c, ConnectionState::Failed);
2137 let c_conn = matches!(c, ConnectionState::Connected);
2138 let e_fail = matches!(e, ConnectionState::Failed);
2139 let e_conn = matches!(e, ConnectionState::Connected);
2140
2141 let c_live = c_conn && peer_has_live_transport(candidate);
2147 let e_live = e_conn && peer_has_live_transport(existing);
2148 if c_live && !e_live {
2149 return true;
2150 }
2151 if !c_live && e_live {
2152 return false;
2153 }
2154
2155 if c_fail && e_conn {
2156 return peer_snapshot_recency_ms(candidate) >= peer_snapshot_recency_ms(existing);
2157 }
2158 if c_conn && e_fail {
2159 return peer_snapshot_recency_ms(candidate) > peer_snapshot_recency_ms(existing);
2160 }
2161 device_status_peer_priority(candidate) > device_status_peer_priority(existing)
2162}
2163
2164fn device_status_peer_priority(
2165 peer: &crate::connection_manager::PeerSnapshot,
2166) -> (u8, u8, u8, u8, i64, u64, u64) {
2167 let readiness_rank = match peer_readiness_state(peer) {
2168 ReadinessState::Routable => 5,
2169 ReadinessState::Settling => 4,
2170 ReadinessState::TransportOnly => 3,
2171 ReadinessState::Connecting => 2,
2172 ReadinessState::AwaitingReplacement => 1,
2173 ReadinessState::Closed | ReadinessState::Failed => 0,
2174 };
2175
2176 let status_rank = match peer.status {
2177 crate::connection_manager::ConnectionState::Connected => 3,
2178 crate::connection_manager::ConnectionState::Connecting
2179 | crate::connection_manager::ConnectionState::Pending => 2,
2180 crate::connection_manager::ConnectionState::Closing
2181 | crate::connection_manager::ConnectionState::Closed => 1,
2182 crate::connection_manager::ConnectionState::Failed => 0,
2183 };
2184
2185 let health_rank = match peer.health {
2186 crate::connection_manager::ConnectionHealth::Healthy => 3,
2187 crate::connection_manager::ConnectionHealth::Suspect => 2,
2188 crate::connection_manager::ConnectionHealth::Unknown => 1,
2189 crate::connection_manager::ConnectionHealth::Stale => 0,
2190 };
2191
2192 (
2193 readiness_rank,
2194 status_rank,
2195 health_rank,
2196 u8::from(peer_has_live_transport(peer)),
2197 peer.last_seen_at_ms,
2198 peer.transition_count,
2199 peer.replacement_count,
2200 )
2201}
2202
2203#[cfg(not(target_arch = "wasm32"))]
2204fn auto_connect_verbose() -> bool {
2205 std::env::var("OPENRTC_AUTOCONNECT_VERBOSE")
2206 .map(|value| {
2207 let normalized = value.trim().to_ascii_lowercase();
2208 normalized == "1" || normalized == "true" || normalized == "yes"
2209 })
2210 .unwrap_or(false)
2211}
2212
2213#[cfg(not(target_arch = "wasm32"))]
2214fn parse_env_bool(value: &str) -> Option<bool> {
2215 match value.trim().to_ascii_lowercase().as_str() {
2216 "1" | "true" | "yes" | "on" => Some(true),
2217 "0" | "false" | "no" | "off" => Some(false),
2218 _ => None,
2219 }
2220}
2221
2222#[cfg(not(target_arch = "wasm32"))]
2223fn env_bool(name: &str) -> Option<bool> {
2224 std::env::var(name)
2225 .ok()
2226 .and_then(|value| parse_env_bool(&value))
2227}
2228
2229#[cfg(not(target_arch = "wasm32"))]
2230fn should_use_relay_only_mode() -> bool {
2231 env_bool("PLUTO_IROH_RELAY_ONLY").unwrap_or(false)
2232}
2233
2234#[cfg(not(target_arch = "wasm32"))]
2235fn should_disable_ipv6() -> bool {
2236 resolve_disable_ipv6(
2237 env_bool("PLUTO_IROH_DISABLE_IPV6"),
2238 env_bool("PLUTO_IROH_ENABLE_IPV6"),
2239 )
2240}
2241
2242#[cfg(not(target_arch = "wasm32"))]
2243fn resolve_disable_ipv6(disable: Option<bool>, enable: Option<bool>) -> bool {
2244 disable
2245 .or_else(|| enable.map(|value| !value))
2246 .unwrap_or(false)
2247}
2248
2249#[cfg(not(target_arch = "wasm32"))]
2250fn apply_native_network_preferences(
2251 mut builder: iroh::endpoint::Builder,
2252 relay_only: bool,
2253 relay_transport_policy: RelayPolicy,
2254) -> anyhow::Result<iroh::endpoint::Builder> {
2255 #[cfg(openrtc_iroh_relay_transport_policy_api)]
2256 {
2257 let policy = match relay_transport_policy {
2258 RelayPolicy::Auto => iroh::RelayTransportPolicy::Auto,
2259 RelayPolicy::QuicRequired => iroh::RelayTransportPolicy::QuicRequired,
2260 RelayPolicy::WebsocketRequired => iroh::RelayTransportPolicy::WebsocketRequired,
2261 };
2262 builder = builder.relay_transport_policy(policy);
2263 }
2264 #[cfg(not(openrtc_iroh_relay_transport_policy_api))]
2265 {
2266 match relay_transport_policy {
2267 RelayPolicy::QuicRequired => {
2268 anyhow::bail!(
2269 "requested relay carriage quicRequired is unsupported by this upstream-Iroh build; refusing to initialize"
2270 );
2271 }
2272 RelayPolicy::Auto => {
2273 eprintln!(
2274 "[pluto-rtc][native] requested relay carriage auto; effective carriage is websocket fallback because this upstream-Iroh build has no QUIC relay policy API"
2275 );
2276 }
2277 RelayPolicy::WebsocketRequired => {}
2278 }
2279 }
2280
2281 builder = builder.transport_config(crate::native_node::iroh_config());
2298
2299 if relay_only {
2300 return Ok(builder.clear_ip_transports());
2302 }
2303
2304 if should_disable_ipv6() {
2305 builder = builder.clear_ip_transports();
2308 builder = builder
2309 .bind_addr("0.0.0.0:0")
2310 .map_err(|e| anyhow::anyhow!("failed to bind IPv4 transport: {}", e))?;
2311 }
2312
2313 Ok(builder)
2314}
2315
2316#[cfg(not(target_arch = "wasm32"))]
2317fn auto_connect_failure_backoff_ms(failure_count: u8) -> i64 {
2318 match failure_count {
2319 0 => 0,
2320 _ => {
2321 let exp = 500_i64.saturating_mul(1_i64 << (failure_count as u32).min(4));
2325 exp.min(8_000)
2326 }
2327 }
2328}
2329
2330#[cfg(not(target_arch = "wasm32"))]
2331fn auto_connect_admission_rejection_backoff_ms(failure_count: u8) -> i64 {
2332 match failure_count {
2333 0 => 0,
2334 _ => {
2335 let exp = 1_000_i64.saturating_mul(1_i64 << (failure_count as u32).min(5));
2339 exp.min(30_000)
2340 }
2341 }
2342}
2343
2344#[cfg(not(target_arch = "wasm32"))]
2345fn auto_connect_network_change_failure_threshold() -> u8 {
2346 std::env::var("OPENRTC_NETWORK_CHANGE_FAILURE_THRESHOLD")
2347 .ok()
2348 .and_then(|value| value.parse::<u8>().ok())
2349 .map(|value| value.clamp(1, 6))
2350 .unwrap_or(2)
2351}
2352
2353#[cfg(not(target_arch = "wasm32"))]
2357fn non_initiator_escalation_grace_ms(escalation_count: u8) -> i64 {
2358 let base_ms = 2_000_i64;
2359 let exp_ms = base_ms.saturating_mul(1_i64 << (escalation_count as u32).min(3));
2360 let capped_ms = exp_ms.min(16_000);
2361 let jitter_pct = ((escalation_count as i64 * 37 + 7) % 41) - 20; let jitter_ms = capped_ms * jitter_pct / 100;
2365 (capped_ms + jitter_ms).max(1_000)
2366}
2367
2368#[cfg(not(target_arch = "wasm32"))]
2369fn next_non_initiator_wakeup_delay_ms(
2370 wait_started_at: &std::collections::HashMap<String, i64>,
2371 escalation_count: &std::collections::HashMap<String, u8>,
2372 now_ms: i64,
2373) -> Option<u64> {
2374 wait_started_at
2375 .iter()
2376 .map(|(device_id, started_at)| {
2377 let grace_ms = non_initiator_escalation_grace_ms(
2378 escalation_count.get(device_id).copied().unwrap_or(0),
2379 );
2380 started_at
2381 .saturating_add(grace_ms)
2382 .saturating_sub(now_ms)
2383 .max(1) as u64
2384 })
2385 .min()
2386}
2387
2388#[cfg(not(target_arch = "wasm32"))]
2389fn register_auto_connect_failure(
2390 remote_device_id: &str,
2391 now_ms: i64,
2392 failure_count: &mut std::collections::HashMap<String, u8>,
2393 failure_backoff_until: &mut std::collections::HashMap<String, i64>,
2394) {
2395 let count = failure_count
2396 .entry(remote_device_id.to_string())
2397 .or_insert(0);
2398 *count = count.saturating_add(1).min(10);
2399 let backoff_ms = auto_connect_failure_backoff_ms(*count);
2400 failure_backoff_until.insert(
2401 remote_device_id.to_string(),
2402 now_ms.saturating_add(backoff_ms),
2403 );
2404}
2405
2406#[cfg(not(target_arch = "wasm32"))]
2407fn register_auto_connect_admission_rejection(
2408 remote_device_id: &str,
2409 now_ms: i64,
2410 failure_count: &mut std::collections::HashMap<String, u8>,
2411 failure_backoff_until: &mut std::collections::HashMap<String, i64>,
2412) {
2413 let count = failure_count
2414 .entry(remote_device_id.to_string())
2415 .or_insert(0);
2416 *count = count.saturating_add(1).min(10);
2417 let backoff_ms = auto_connect_admission_rejection_backoff_ms(*count);
2418 failure_backoff_until.insert(
2419 remote_device_id.to_string(),
2420 now_ms.saturating_add(backoff_ms),
2421 );
2422}
2423
2424#[cfg(not(target_arch = "wasm32"))]
2425fn clear_auto_connect_failure_state(
2426 remote_device_id: &str,
2427 failure_count: &mut std::collections::HashMap<String, u8>,
2428 failure_backoff_until: &mut std::collections::HashMap<String, i64>,
2429) {
2430 failure_count.remove(remote_device_id);
2431 failure_backoff_until.remove(remote_device_id);
2432}
2433
2434#[cfg(target_arch = "wasm32")]
2435fn emit_wasm_connection_state_event(runtime_owner_id: &str, snapshot: &StateSnapshot) {
2436 use wasm_bindgen::JsValue;
2437
2438 let Some(window) = web_sys::window() else {
2439 return;
2440 };
2441
2442 let detail = js_sys::Object::new();
2443 let _ = js_sys::Reflect::set(
2444 &detail,
2445 &JsValue::from_str("runtimeOwnerId"),
2446 &JsValue::from_str(runtime_owner_id),
2447 );
2448 let _ = js_sys::Reflect::set(
2449 &detail,
2450 &JsValue::from_str("connectionId"),
2451 &JsValue::from_str(snapshot.connection_id.as_str()),
2452 );
2453 let _ = js_sys::Reflect::set(
2454 &detail,
2455 &JsValue::from_str("state"),
2456 &JsValue::from_str(snapshot.state.as_str()),
2457 );
2458 let _ = js_sys::Reflect::set(
2459 &detail,
2460 &JsValue::from_str("createdAt"),
2461 &JsValue::from_f64(snapshot.created_at as f64),
2462 );
2463 let _ = js_sys::Reflect::set(
2464 &detail,
2465 &JsValue::from_str("updatedAt"),
2466 &JsValue::from_f64(snapshot.updated_at as f64),
2467 );
2468 let _ = js_sys::Reflect::set(
2469 &detail,
2470 &JsValue::from_str("transportState"),
2471 &JsValue::from_str(snapshot.transport_state.as_str()),
2472 );
2473 let _ = js_sys::Reflect::set(
2474 &detail,
2475 &JsValue::from_str("protocolState"),
2476 &JsValue::from_str(snapshot.protocol_state.as_str()),
2477 );
2478 let _ = js_sys::Reflect::set(
2479 &detail,
2480 &JsValue::from_str("routable"),
2481 &JsValue::from_bool(snapshot.routable),
2482 );
2483 let _ = js_sys::Reflect::set(
2484 &detail,
2485 &JsValue::from_str("logicalSessionTerminal"),
2486 &JsValue::from_bool(snapshot.logical_session_terminal),
2487 );
2488 let _ = js_sys::Reflect::set(
2489 &detail,
2490 &JsValue::from_str("transportGeneration"),
2491 &JsValue::from_f64(snapshot.transport_generation as f64),
2492 );
2493 let _ = js_sys::Reflect::set(
2494 &detail,
2495 &JsValue::from_str("routeGeneration"),
2496 &JsValue::from_f64(snapshot.route_generation as f64),
2497 );
2498 let _ = js_sys::Reflect::set(
2499 &detail,
2500 &JsValue::from_str("activeTransportStableId"),
2501 &snapshot
2502 .active_transport_stable_id
2503 .map(|value| JsValue::from_f64(value as f64))
2504 .unwrap_or(JsValue::NULL),
2505 );
2506 let _ = js_sys::Reflect::set(
2507 &detail,
2508 &JsValue::from_str("activeTransport"),
2509 &JsValue::from_str(snapshot.active_transport.as_str()),
2510 );
2511 let _ = js_sys::Reflect::set(
2512 &detail,
2513 &JsValue::from_str("parallelTransport"),
2514 &snapshot
2515 .parallel_transport
2516 .as_deref()
2517 .map(JsValue::from_str)
2518 .unwrap_or(JsValue::NULL),
2519 );
2520 let _ = js_sys::Reflect::set(
2521 &detail,
2522 &JsValue::from_str("replacementInProgress"),
2523 &JsValue::from_bool(snapshot.replacement_in_progress),
2524 );
2525 let _ = js_sys::Reflect::set(
2526 &detail,
2527 &JsValue::from_str("transitionCount"),
2528 &JsValue::from_f64(snapshot.transition_count as f64),
2529 );
2530 let _ = js_sys::Reflect::set(
2531 &detail,
2532 &JsValue::from_str("connectingTransitionCount"),
2533 &JsValue::from_f64(snapshot.connecting_transition_count as f64),
2534 );
2535 let _ = js_sys::Reflect::set(
2536 &detail,
2537 &JsValue::from_str("replacementCount"),
2538 &JsValue::from_f64(snapshot.replacement_count as f64),
2539 );
2540 let _ = js_sys::Reflect::set(
2541 &detail,
2542 &JsValue::from_str("retireCount"),
2543 &JsValue::from_f64(snapshot.retire_count as f64),
2544 );
2545
2546 if let Some(device_id) = snapshot.device_id.as_deref() {
2547 let _ = js_sys::Reflect::set(
2548 &detail,
2549 &JsValue::from_str("deviceId"),
2550 &JsValue::from_str(device_id),
2551 );
2552 }
2553 if let Some(device_id_hint) = snapshot.device_id_hint.as_deref() {
2554 let _ = js_sys::Reflect::set(
2555 &detail,
2556 &JsValue::from_str("deviceIdHint"),
2557 &JsValue::from_str(device_id_hint),
2558 );
2559 }
2560 if let Some(remote_node_id) = snapshot.remote_node_id.as_deref() {
2561 let _ = js_sys::Reflect::set(
2562 &detail,
2563 &JsValue::from_str("remoteNodeId"),
2564 &JsValue::from_str(remote_node_id),
2565 );
2566 }
2567 if let Some(error) = snapshot.error.as_deref() {
2568 let _ = js_sys::Reflect::set(
2569 &detail,
2570 &JsValue::from_str("error"),
2571 &JsValue::from_str(error),
2572 );
2573 }
2574 if let Some(reason) = snapshot.last_disconnect_reason.as_deref() {
2575 let _ = js_sys::Reflect::set(
2576 &detail,
2577 &JsValue::from_str("lastDisconnectReason"),
2578 &JsValue::from_str(reason),
2579 );
2580 }
2581 if let Some(reason) = snapshot.last_reconnect_reason.as_deref() {
2582 let _ = js_sys::Reflect::set(
2583 &detail,
2584 &JsValue::from_str("lastReconnectReason"),
2585 &JsValue::from_str(reason),
2586 );
2587 }
2588
2589 let init = web_sys::CustomEventInit::new();
2590 init.set_detail(&detail.into());
2591 if let Ok(event) =
2592 web_sys::CustomEvent::new_with_event_init_dict("connection-state-changed", &init)
2593 {
2594 let _ = window.dispatch_event(&event);
2595 }
2596}
2597
2598fn parse_endpoint_ticket(ticket: &str) -> anyhow::Result<iroh::EndpointAddr> {
2599 let trimmed = ticket.trim();
2600 if trimmed.is_empty() {
2601 return Err(anyhow::anyhow!("endpoint ticket is required"));
2602 }
2603
2604 let parsed = EndpointTicket::from_str(trimmed)
2605 .map_err(|e| anyhow::anyhow!("invalid endpoint ticket: {}", e))?;
2606 Ok(parsed.endpoint_addr().clone())
2607}
2608
2609#[derive(Debug, Clone, serde::Serialize)]
2610#[serde(rename_all = "camelCase")]
2611pub struct RuntimeStatus {
2612 pub ready: bool,
2613 pub node_id: Option<String>,
2614 pub transport: TransportStatus,
2615}
2616
2617#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
2618#[serde(rename_all = "kebab-case")]
2619pub enum CapabilityMaturity {
2620 Stable,
2621 Preview,
2622 SupportOnly,
2623 Unavailable,
2624}
2625
2626#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
2627#[serde(rename_all = "camelCase")]
2628pub struct ProductCapabilityMaturity {
2629 pub bounded_rooms: CapabilityMaturity,
2630 pub service_nodes: CapabilityMaturity,
2631 pub offline_edge: CapabilityMaturity,
2632 pub broadcast: CapabilityMaturity,
2633}
2634
2635impl ProductCapabilityMaturity {
2636 pub(crate) fn for_current_target() -> Self {
2637 Self {
2638 bounded_rooms: CapabilityMaturity::Stable,
2639 service_nodes: CapabilityMaturity::Preview,
2640 offline_edge: if cfg!(all(not(target_arch = "wasm32"), feature = "transport-lan")) {
2641 CapabilityMaturity::Preview
2642 } else {
2643 CapabilityMaturity::SupportOnly
2644 },
2645 broadcast: CapabilityMaturity::Preview,
2646 }
2647 }
2648}
2649
2650#[derive(Debug, Clone, serde::Serialize)]
2651#[serde(rename_all = "camelCase")]
2652pub struct FeatureStatus {
2653 pub compiled: bool,
2654 pub enabled: bool,
2655}
2656
2657impl FeatureStatus {
2658 pub(crate) fn new(compiled: bool, enabled: bool) -> Self {
2659 Self {
2660 compiled,
2661 enabled: compiled && enabled,
2662 }
2663 }
2664}
2665
2666#[derive(Debug, Clone, serde::Serialize)]
2667#[serde(rename_all = "camelCase")]
2668pub struct TransportStatus {
2669 pub iroh_quic: FeatureStatus,
2670 pub iroh_lan: FeatureStatus,
2671 pub web_rtc: FeatureStatus,
2672 pub moq: FeatureStatus,
2673 pub ble: FeatureStatus,
2674 pub iroh_relay_only: bool,
2675 #[serde(skip_serializing_if = "Option::is_none")]
2676 pub iroh_relay_transport_policy: Option<RelayPolicy>,
2677 pub iroh_relay: RelayStatus,
2678 pub route_priority: Vec<crate::route_policy::KnownRoute>,
2679 pub optimize_for: crate::route_policy::RouteOptimization,
2680 pub network_policy: NetworkPolicy,
2681}
2682
2683#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
2684#[serde(rename_all = "camelCase")]
2685pub enum IrohRelayProvider {
2686 VendoredIroh,
2688 UpstreamIroh,
2690}
2691
2692#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
2693#[serde(rename_all = "camelCase")]
2694pub enum RelayCarriage {
2695 Quic,
2696 Websocket,
2697 QuicWithWebsocketFallback,
2698 WebsocketFallback,
2699 Unsupported,
2700}
2701
2702#[derive(Debug, Clone, serde::Serialize)]
2703#[serde(rename_all = "camelCase")]
2704pub struct RelayStatus {
2705 pub provider: IrohRelayProvider,
2706 pub requested_carriage: RelayPolicy,
2707 pub effective_carriage: RelayCarriage,
2708}
2709
2710impl RelayStatus {
2711 pub(crate) fn from_policy(
2712 requested_carriage: RelayPolicy,
2713 provider: IrohRelayProvider,
2714 ) -> Self {
2715 let effective_carriage = match requested_carriage {
2716 RelayPolicy::WebsocketRequired => RelayCarriage::Websocket,
2717 RelayPolicy::Auto => {
2718 #[cfg(target_arch = "wasm32")]
2719 {
2720 RelayCarriage::Websocket
2721 }
2722 #[cfg(all(not(target_arch = "wasm32"), openrtc_iroh_relay_transport_policy_api))]
2723 {
2724 RelayCarriage::QuicWithWebsocketFallback
2725 }
2726 #[cfg(all(
2727 not(target_arch = "wasm32"),
2728 not(openrtc_iroh_relay_transport_policy_api)
2729 ))]
2730 {
2731 RelayCarriage::WebsocketFallback
2732 }
2733 }
2734 RelayPolicy::QuicRequired => {
2735 #[cfg(target_arch = "wasm32")]
2736 {
2737 RelayCarriage::Unsupported
2738 }
2739 #[cfg(all(not(target_arch = "wasm32"), openrtc_iroh_relay_transport_policy_api))]
2740 {
2741 RelayCarriage::Quic
2742 }
2743 #[cfg(all(
2744 not(target_arch = "wasm32"),
2745 not(openrtc_iroh_relay_transport_policy_api)
2746 ))]
2747 {
2748 RelayCarriage::Unsupported
2749 }
2750 }
2751 };
2752
2753 Self {
2754 provider,
2755 requested_carriage,
2756 effective_carriage,
2757 }
2758 }
2759}
2760
2761#[cfg(openrtc_iroh_relay_transport_policy_api)]
2762pub(crate) const fn current_iroh_relay_provider() -> IrohRelayProvider {
2763 IrohRelayProvider::VendoredIroh
2764}
2765
2766#[cfg(not(openrtc_iroh_relay_transport_policy_api))]
2767pub(crate) const fn current_iroh_relay_provider() -> IrohRelayProvider {
2768 IrohRelayProvider::UpstreamIroh
2769}
2770
2771impl TransportStatus {
2772 pub(crate) fn from_config_with_ble_available(
2773 config: &TransportConfig,
2774 ble_available: bool,
2775 relay_provider: IrohRelayProvider,
2776 ) -> Self {
2777 let lan_compiled = cfg!(feature = "transport-lan");
2778 let lan_enabled = config
2779 .iroh_lan
2780 .as_ref()
2781 .map(|lan| lan.enabled)
2782 .unwrap_or(false);
2783 let webrtc_compiled = cfg!(feature = "transport-webrtc");
2784 let webrtc_enabled = config.webrtc.is_some();
2785 let moq_compiled = cfg!(feature = "transport-moq");
2786 let moq_enabled = config.moq.is_some();
2787 let ble_compiled = ble_available;
2788 let ble_enabled = config.ble.as_ref().map(|ble| ble.enabled).unwrap_or(false);
2789
2790 Self {
2791 iroh_quic: FeatureStatus::new(true, true),
2792 iroh_lan: FeatureStatus::new(lan_compiled, lan_enabled),
2793 web_rtc: FeatureStatus::new(webrtc_compiled, webrtc_enabled),
2794 moq: FeatureStatus::new(moq_compiled, moq_enabled),
2795 ble: FeatureStatus::new(ble_compiled, ble_enabled),
2796 iroh_relay_only: config.iroh_relay_only,
2797 iroh_relay_transport_policy: config.iroh_relay_transport_policy.clone(),
2798 iroh_relay: RelayStatus::from_policy(
2799 config
2800 .iroh_relay_transport_policy
2801 .clone()
2802 .unwrap_or(RelayPolicy::WebsocketRequired),
2803 relay_provider,
2804 ),
2805 route_priority: config.route_priority.clone(),
2806 optimize_for: config.optimize_for,
2807 network_policy: config.network_policy,
2808 }
2809 }
2810}
2811
2812#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
2813#[serde(rename_all = "camelCase")]
2814pub struct IceServerConfig {
2815 #[serde(default)]
2816 pub urls: Vec<String>,
2817 #[serde(default, skip_serializing_if = "Option::is_none")]
2818 pub username: Option<String>,
2819 #[serde(default, skip_serializing_if = "Option::is_none")]
2820 pub credential: Option<String>,
2821}
2822
2823#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
2824#[serde(rename_all = "camelCase")]
2825pub struct WebRTCConfig {
2826 #[serde(default)]
2827 pub ice_servers: Vec<IceServerConfig>,
2828 #[serde(default)]
2829 pub privacy_mode: bool,
2830 #[serde(default)]
2831 pub lan_mode: bool,
2832}
2833
2834impl Default for WebRTCConfig {
2835 fn default() -> Self {
2836 Self {
2837 ice_servers: Vec::new(),
2838 privacy_mode: false,
2839 lan_mode: false,
2840 }
2841 }
2842}
2843
2844#[derive(Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
2845#[serde(rename_all = "camelCase")]
2846pub struct MoQConfig {
2847 #[serde(default)]
2848 pub relay_url: String,
2849 #[serde(default, skip_serializing)]
2850 pub access_token: Option<String>,
2851}
2852
2853impl std::fmt::Debug for MoQConfig {
2854 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2855 let relay_label = self
2856 .relay_url
2857 .split(['?', '#'])
2858 .next()
2859 .unwrap_or("invalid-moq-relay-url");
2860 formatter
2861 .debug_struct("MoQConfig")
2862 .field("relay_url", &relay_label)
2863 .field(
2864 "access_token",
2865 &self.access_token.as_ref().map(|_| "[redacted]"),
2866 )
2867 .finish()
2868 }
2869}
2870
2871impl Default for MoQConfig {
2872 fn default() -> Self {
2873 Self {
2874 relay_url: String::new(),
2875 access_token: None,
2876 }
2877 }
2878}
2879
2880#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
2881#[serde(rename_all = "camelCase")]
2882pub enum RelayPolicy {
2883 Auto,
2884 QuicRequired,
2885 WebsocketRequired,
2886}
2887
2888#[derive(Debug, Clone, Copy, Default, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
2895#[serde(rename_all = "kebab-case")]
2896pub enum NetworkPolicy {
2897 #[default]
2898 Managed,
2899 LocalOnly,
2900}
2901
2902#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
2903#[serde(rename_all = "camelCase")]
2904pub struct TransportConfig {
2905 #[serde(default)]
2906 pub network_policy: NetworkPolicy,
2907 #[serde(default = "default_true")]
2910 pub relay: bool,
2911 #[serde(default)]
2913 pub privacy_mode: bool,
2914 #[serde(default)]
2917 pub iroh_relay_only: bool,
2918 #[serde(default, skip_serializing_if = "Option::is_none")]
2919 pub iroh_relay_transport_policy: Option<RelayPolicy>,
2920 #[serde(default, skip_serializing_if = "Option::is_none")]
2921 pub iroh_lan: Option<IrohLanConfig>,
2922 #[serde(default, skip_serializing_if = "Option::is_none")]
2923 pub webrtc: Option<WebRTCConfig>,
2924 #[serde(default, skip_serializing_if = "Option::is_none")]
2925 pub moq: Option<MoQConfig>,
2926 #[serde(default, skip_serializing_if = "Option::is_none")]
2927 pub ble: Option<BleConfig>,
2928 #[serde(default = "default_route_priority")]
2931 pub route_priority: Vec<crate::route_policy::KnownRoute>,
2932 #[serde(default)]
2933 pub optimize_for: crate::route_policy::RouteOptimization,
2934}
2935
2936fn default_route_priority() -> Vec<crate::route_policy::KnownRoute> {
2937 crate::route_policy::DEFAULT_ROUTE_PRIORITY.to_vec()
2938}
2939
2940fn default_true() -> bool {
2941 true
2942}
2943
2944impl Default for TransportConfig {
2945 fn default() -> Self {
2946 Self {
2947 network_policy: NetworkPolicy::Managed,
2948 relay: true,
2949 privacy_mode: false,
2950 iroh_relay_only: false,
2951 iroh_relay_transport_policy: None,
2952 iroh_lan: Some(IrohLanConfig::default()),
2953 webrtc: None,
2954 moq: None,
2955 ble: None,
2956 route_priority: default_route_priority(),
2957 optimize_for: crate::route_policy::RouteOptimization::Balanced,
2958 }
2959 }
2960}
2961
2962impl TransportConfig {
2963 #[cfg(any(test, feature = "transport-webrtc", feature = "transport-moq"))]
2967 pub(crate) fn ranked_iroh_carriers(
2968 &self,
2969 supports_webrtc: bool,
2970 supports_moq: bool,
2971 ) -> Vec<crate::route_policy::KnownRoute> {
2972 crate::route_policy::rank_iroh_carriers(
2973 &self.route_priority,
2974 supports_webrtc,
2975 supports_moq,
2976 self.privacy_mode,
2977 self.webrtc
2978 .as_ref()
2979 .is_some_and(|webrtc| webrtc.privacy_mode),
2980 )
2981 }
2982
2983 #[cfg(not(target_arch = "wasm32"))]
2987 pub(crate) fn endpoint_rebind_required(&self, next: &Self) -> bool {
2988 self.network_policy != next.network_policy
2989 || self.relay != next.relay
2990 || self.iroh_relay_only != next.iroh_relay_only
2991 || self.iroh_relay_transport_policy != next.iroh_relay_transport_policy
2992 || self.iroh_lan != next.iroh_lan
2993 }
2994
2995 pub fn sanitize_for_runtime(self) -> Self {
2996 self.sanitize_for_runtime_with_ble_available(false)
2997 }
2998
2999 fn sanitize_for_client_construction(self) -> Self {
3005 #[cfg(not(target_arch = "wasm32"))]
3006 let deferred_ble_requested = self.ble.as_ref().is_some_and(|config| config.enabled);
3007 #[cfg(target_arch = "wasm32")]
3008 let deferred_ble_requested = false;
3009 self.sanitize_for_runtime_with_ble_available(deferred_ble_requested)
3010 }
3011
3012 pub(crate) fn sanitize_for_runtime_with_ble_available(mut self, ble_available: bool) -> Self {
3013 let mut seen_routes = std::collections::HashSet::new();
3014 self.route_priority
3015 .retain(|route| seen_routes.insert(*route));
3016 if self.route_priority.is_empty() {
3017 self.route_priority = default_route_priority();
3018 }
3019
3020 if self.network_policy == NetworkPolicy::LocalOnly {
3021 self.relay = false;
3022 self.privacy_mode = false;
3023 self.iroh_relay_only = false;
3024 self.iroh_relay_transport_policy = None;
3025 self.iroh_lan = Some(IrohLanConfig {
3026 enabled: true,
3027 advertise: self
3028 .iroh_lan
3029 .as_ref()
3030 .map(|lan| lan.advertise)
3031 .unwrap_or(true),
3032 });
3033 self.webrtc = None;
3034 self.moq = None;
3035 self.route_priority.retain(|route| {
3036 matches!(
3037 route,
3038 crate::route_policy::KnownRoute::IrohLan
3039 | crate::route_policy::KnownRoute::IrohQuic
3040 | crate::route_policy::KnownRoute::Ble
3041 )
3042 });
3043 if self.route_priority.is_empty() {
3044 self.route_priority = vec![
3045 crate::route_policy::KnownRoute::IrohLan,
3046 crate::route_policy::KnownRoute::IrohQuic,
3047 crate::route_policy::KnownRoute::Ble,
3048 ];
3049 }
3050 }
3051
3052 if self.privacy_mode && self.relay {
3053 self.iroh_relay_only = true;
3054 self.iroh_lan = None;
3055 self.ble = None;
3056 self.route_priority.retain(|route| {
3057 !matches!(
3058 route,
3059 crate::route_policy::KnownRoute::IrohLan
3060 | crate::route_policy::KnownRoute::IrohQuic
3061 | crate::route_policy::KnownRoute::Ble
3062 )
3063 });
3064 if let Some(webrtc) = self.webrtc.as_mut() {
3065 webrtc.privacy_mode = true;
3066 webrtc.lan_mode = false;
3067 for server in &mut webrtc.ice_servers {
3068 server.urls.retain(|url| {
3069 let url = url.trim().to_ascii_lowercase();
3070 url.starts_with("turn:") || url.starts_with("turns:")
3071 });
3072 }
3073 webrtc.ice_servers.retain(|server| !server.urls.is_empty());
3074 }
3075 }
3076
3077 if !self.relay {
3078 self.privacy_mode = false;
3079 self.iroh_relay_only = false;
3080 self.iroh_relay_transport_policy = None;
3081 self.moq = None;
3082 self.route_priority.retain(|route| {
3083 !matches!(
3084 route,
3085 crate::route_policy::KnownRoute::IrohRelay
3086 | crate::route_policy::KnownRoute::Moq
3087 )
3088 });
3089 if self.route_priority.is_empty() {
3090 self.route_priority = vec![
3091 crate::route_policy::KnownRoute::IrohLan,
3092 crate::route_policy::KnownRoute::IrohQuic,
3093 crate::route_policy::KnownRoute::WebRtc,
3094 crate::route_policy::KnownRoute::Ble,
3095 ];
3096 }
3097 if let Some(webrtc) = self.webrtc.as_mut() {
3098 webrtc.privacy_mode = false;
3099 for server in &mut webrtc.ice_servers {
3100 server.urls.retain(|url| {
3101 let url = url.trim().to_ascii_lowercase();
3102 !url.starts_with("turn:") && !url.starts_with("turns:")
3103 });
3104 }
3105 webrtc.ice_servers.retain(|server| !server.urls.is_empty());
3106 }
3107 }
3108
3109 if self.ble.as_ref().map(|ble| ble.enabled).unwrap_or(false) && !ble_available {
3110 self.ble = None;
3111 }
3112
3113 self
3114 }
3115}
3116
3117pub struct ClientBuilder {
3118 #[cfg(not(target_arch = "wasm32"))]
3119 native_ip_exclusions: Option<Vec<ipnet::IpNet>>,
3120 #[cfg(all(not(target_arch = "wasm32"), feature = "native-udp-socket-initializer"))]
3121 native_udp_socket_initializer: Option<NativeUdpSocketInitializer>,
3122 #[cfg(all(
3123 not(target_arch = "wasm32"),
3124 feature = "native-relay-tcp-socket-initializer"
3125 ))]
3126 native_relay_tcp_socket_initializer: Option<NativeRelayTcpSocketInitializer>,
3127 #[cfg(all(not(target_arch = "wasm32"), feature = "native-dns-socket-initializer"))]
3128 native_dns_socket_initializer: Option<NativeDnsSocketInitializer>,
3129 #[cfg(not(target_arch = "wasm32"))]
3130 project_id: String,
3131 app_tag: String,
3132 transport_config: TransportConfig,
3133 #[cfg(not(target_arch = "wasm32"))]
3134 token_provider: Arc<dyn Fn() -> Option<String> + Send + Sync>,
3135 signaling: Option<Arc<dyn SignalingBackend>>,
3136 room: Option<Arc<dyn RoomBackend>>,
3137}
3138
3139impl ClientBuilder {
3140 pub fn new_provider_neutral(
3141 app_tag: String,
3142 identity_credential_provider: Box<dyn Fn() -> Option<String> + Send + Sync>,
3143 ) -> Self {
3144 #[cfg(target_arch = "wasm32")]
3145 let _ = identity_credential_provider;
3146 Self {
3147 #[cfg(not(target_arch = "wasm32"))]
3151 project_id: String::new(),
3152 app_tag,
3153 transport_config: TransportConfig::default(),
3154 #[cfg(not(target_arch = "wasm32"))]
3155 native_ip_exclusions: None,
3156 #[cfg(all(not(target_arch = "wasm32"), feature = "native-udp-socket-initializer"))]
3157 native_udp_socket_initializer: None,
3158 #[cfg(all(
3159 not(target_arch = "wasm32"),
3160 feature = "native-relay-tcp-socket-initializer"
3161 ))]
3162 native_relay_tcp_socket_initializer: None,
3163 #[cfg(all(not(target_arch = "wasm32"), feature = "native-dns-socket-initializer"))]
3164 native_dns_socket_initializer: None,
3165 #[cfg(not(target_arch = "wasm32"))]
3166 token_provider: Arc::from(identity_credential_provider),
3167 signaling: None,
3168 room: None,
3169 }
3170 }
3171
3172 #[cfg(test)]
3175 pub(crate) fn new_for_test(
3176 project_id: String,
3177 api_key: String,
3178 token_provider: Box<dyn Fn() -> Option<String> + Send + Sync>,
3179 ) -> Self {
3180 Self::new_with_app_tag(
3181 project_id,
3182 crate::app_tag_from_api_key(&api_key),
3183 token_provider,
3184 )
3185 }
3186
3187 #[cfg(test)]
3188 pub fn new_with_app_tag(
3189 project_id: String,
3190 app_tag: String,
3191 token_provider: Box<dyn Fn() -> Option<String> + Send + Sync>,
3192 ) -> Self {
3193 Self {
3194 #[cfg(not(target_arch = "wasm32"))]
3195 project_id,
3196 app_tag,
3197 transport_config: TransportConfig::default(),
3198 #[cfg(not(target_arch = "wasm32"))]
3199 native_ip_exclusions: None,
3200 #[cfg(all(not(target_arch = "wasm32"), feature = "native-udp-socket-initializer"))]
3201 native_udp_socket_initializer: None,
3202 #[cfg(all(
3203 not(target_arch = "wasm32"),
3204 feature = "native-relay-tcp-socket-initializer"
3205 ))]
3206 native_relay_tcp_socket_initializer: None,
3207 #[cfg(all(not(target_arch = "wasm32"), feature = "native-dns-socket-initializer"))]
3208 native_dns_socket_initializer: None,
3209 #[cfg(not(target_arch = "wasm32"))]
3210 token_provider: Arc::from(token_provider),
3211 signaling: None,
3212 room: None,
3213 }
3214 }
3215
3216 #[cfg(not(target_arch = "wasm32"))]
3223 pub fn native_ip_exclusions(mut self, networks: Vec<ipnet::IpNet>) -> Self {
3224 self.native_ip_exclusions = Some(networks);
3225 self
3226 }
3227
3228 #[cfg(all(not(target_arch = "wasm32"), feature = "native-udp-socket-initializer"))]
3234 pub fn native_udp_socket_initializer(
3235 mut self,
3236 initialize: impl for<'a> Fn(NativeUdpSocketRef<'a>, NativeUdpIpFamily) -> std::io::Result<()>
3237 + Send
3238 + Sync
3239 + 'static,
3240 ) -> Self {
3241 self.native_udp_socket_initializer = Some(Arc::new(initialize));
3242 self
3243 }
3244
3245 #[cfg(all(
3251 not(target_arch = "wasm32"),
3252 feature = "native-relay-tcp-socket-initializer"
3253 ))]
3254 pub fn native_relay_tcp_socket_initializer(
3255 mut self,
3256 initialize: impl for<'a> Fn(NativeTcpSocketRef<'a>) -> std::io::Result<()>
3257 + Send
3258 + Sync
3259 + 'static,
3260 ) -> Self {
3261 self.native_relay_tcp_socket_initializer = Some(Arc::new(initialize));
3262 self
3263 }
3264
3265 #[cfg(all(not(target_arch = "wasm32"), feature = "native-dns-socket-initializer"))]
3270 pub fn native_dns_socket_initializer(
3271 mut self,
3272 initialize: impl for<'a> Fn(NativeDnsSocketRef<'a>, NativeDnsSocketProtocol) -> std::io::Result<()>
3273 + Send
3274 + Sync
3275 + 'static,
3276 ) -> Self {
3277 self.native_dns_socket_initializer = Some(Arc::new(initialize));
3278 self
3279 }
3280
3281 pub fn transport_config(mut self, transport_config: TransportConfig) -> Self {
3282 self.transport_config = transport_config.sanitize_for_client_construction();
3283 self
3284 }
3285
3286 pub fn signaling_backend(mut self, signaling: Arc<dyn SignalingBackend>) -> Self {
3287 self.signaling = Some(signaling);
3288 self
3289 }
3290
3291 pub fn room_backend(mut self, room: Arc<dyn RoomBackend>) -> Self {
3292 self.room = Some(room);
3293 self
3294 }
3295
3296 pub fn build(self) -> Client {
3297 let app_backgrounded = Arc::new(AtomicBool::new(false));
3298
3299 let room = self
3300 .room
3301 .unwrap_or_else(|| Arc::new(GatewayRequiredRoomBackend) as Arc<dyn RoomBackend>);
3302
3303 let signaling = self.signaling.unwrap_or_else(|| {
3304 Arc::new(GatewayRequiredSignalingBackend) as Arc<dyn SignalingBackend>
3305 });
3306
3307 #[cfg(not(target_arch = "wasm32"))]
3308 let (native_device_updates, _) = tokio::sync::broadcast::channel(32);
3309 #[cfg(not(target_arch = "wasm32"))]
3310 let (native_connection_state_updates, _) = tokio::sync::broadcast::channel(64);
3311 #[cfg(not(target_arch = "wasm32"))]
3312 let (native_peer_data_updates, _) = tokio::sync::broadcast::channel(256);
3313 #[cfg(not(target_arch = "wasm32"))]
3314 let (native_application_streams, native_application_streams_receiver) =
3315 async_channel::bounded(256);
3316
3317 Client {
3318 app_tag: self.app_tag,
3319 #[cfg(target_arch = "wasm32")]
3320 wasm_connection_state_owner_id: crate::session_token::generate_nonce(),
3321 broadcasts: crate::broadcast::Broadcasts::default(),
3322 signaling,
3323 room,
3324 node_id: Arc::new(RwLock::new(None)),
3325 iroh_endpoint: Arc::new(RwLock::new(None)),
3326 #[cfg(target_arch = "wasm32")]
3327 iroh_node: Arc::new(RwLock::new(None)),
3328 #[cfg(not(target_arch = "wasm32"))]
3329 iroh_node: Arc::new(RwLock::new(None)),
3330 #[cfg(not(target_arch = "wasm32"))]
3331 native_application_streams,
3332 #[cfg(not(target_arch = "wasm32"))]
3333 native_application_streams_receiver,
3334 connection_manager: Arc::new(crate::connection_manager::ConnectionManager::new()),
3335 session_token_registry: Arc::new(crate::session_token::SessionTokenRegistry::new()),
3336 #[cfg(not(target_arch = "wasm32"))]
3337 inbound_session_admission_transport_ids: Arc::new(StdRwLock::new(HashMap::new())),
3338 #[cfg(not(target_arch = "wasm32"))]
3339 offline_admission_requirements: Arc::new(StdRwLock::new(HashMap::new())),
3340 #[cfg(not(target_arch = "wasm32"))]
3341 offline_admission_transport_ids: Arc::new(StdRwLock::new(HashMap::new())),
3342 #[cfg(not(target_arch = "wasm32"))]
3343 offline_runtime: Arc::new(tokio::sync::Mutex::new(None)),
3344 #[cfg(not(target_arch = "wasm32"))]
3345 offline_proof_attempts: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
3346 #[cfg(not(target_arch = "wasm32"))]
3347 native_admission_stream_contracts: Arc::new(StdRwLock::new(HashMap::new())),
3348 remote_session_admission_proofs: Arc::new(StdRwLock::new(RemoteSessionAdmissionProofs::new())),
3349 pending_inline_reciprocal_admissions: Arc::new(StdRwLock::new(HashMap::new())),
3350 outbound_application_security_epoch_fingerprints: Arc::new(StdRwLock::new(
3351 HashMap::new(),
3352 )),
3353 #[cfg(not(target_arch = "wasm32"))]
3354 pending_reciprocal_session_admission_requests: Arc::new(StdRwLock::new(HashSet::new())),
3355 #[cfg(not(target_arch = "wasm32"))]
3356 native_route_repair_credentials: Arc::new(StdRwLock::new(HashMap::new())),
3357 scoped_route_repair_credentials: Arc::new(StdRwLock::new(HashMap::new())),
3358 connection_application_crypto_keys:
3359 crate::client::application_crypto_impl::new_connection_application_crypto_key_map(),
3360 connection_application_crypto_state: Arc::new(StdRwLock::new(())),
3361 connection_application_crypto_required:
3362 crate::client::application_crypto_impl::new_connection_application_crypto_required_set(),
3363 #[cfg(not(target_arch = "wasm32"))]
3364 trusted_user_device_application_crypto_required: Arc::new(AtomicBool::new(false)),
3365 connection_application_crypto_confirmed:
3366 crate::client::application_crypto_impl::new_connection_application_crypto_confirmed_map(),
3367 #[cfg(not(target_arch = "wasm32"))]
3368 connection_application_route_updates: Arc::new(tokio::sync::Notify::new()),
3369 connection_application_crypto_outbound_sequences:
3370 crate::client::application_crypto_impl::new_connection_application_crypto_outbound_sequences(),
3371 connection_application_key_agreements:
3372 crate::client::application_crypto_impl::new_connection_application_key_agreement_map(),
3373 #[cfg(not(target_arch = "wasm32"))]
3374 managed_scope_tickets: Arc::new(StdRwLock::new(ManagedScopeTicketCache::default())),
3375 known_endpoint_addrs: Arc::new(RwLock::new(HashMap::new())),
3376 known_device_ids_by_node: Arc::new(StdRwLock::new(HashMap::new())),
3377 known_device_endpoint_revision: tokio::sync::watch::channel(0).0,
3378 sparse_fanout: Arc::new(tokio::sync::Mutex::new(
3379 crate::sparse_fanout::SparseFanoutState::default(),
3380 )),
3381 #[cfg(not(target_arch = "wasm32"))]
3382 native_device_identity: Arc::new(RwLock::new(None)),
3383 #[cfg(not(target_arch = "wasm32"))]
3384 native_device_base_dir: Arc::new(RwLock::new(None)),
3385 #[cfg(not(target_arch = "wasm32"))]
3386 native_device_identity_init_guard: Arc::new(tokio::sync::Mutex::new(())),
3387 #[cfg(not(target_arch = "wasm32"))]
3388 native_device_updates,
3389 #[cfg(not(target_arch = "wasm32"))]
3390 native_connection_state_updates,
3391 #[cfg(not(target_arch = "wasm32"))]
3392 native_peer_data_updates,
3393 auto_connect_loop_key: Arc::new(Mutex::new(None)),
3394 auto_connect_generation: Arc::new(AtomicU64::new(0)),
3395 #[cfg(not(target_arch = "wasm32"))]
3396 native_auto_connect_wake: Arc::new(tokio::sync::Notify::new()),
3397 #[cfg(not(target_arch = "wasm32"))]
3398 external_desired_peer_actor: Arc::new(tokio::sync::Mutex::new(
3399 auto_connect_impl::NativeExternalAutoConnectActorState::default(),
3400 )),
3401 auto_connect_exclusion_owner: Arc::new(Mutex::new(())),
3402 auto_connect_excluded: Arc::new(Mutex::new(HashSet::new())),
3403 auto_connect_peer_requested_excluded: Arc::new(Mutex::new(HashSet::new())),
3404 auto_connect_excluded_node_aliases: Arc::new(Mutex::new(HashMap::new())),
3405 app_backgrounded,
3406 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
3407 webrtc_availability: Arc::new(Mutex::new(Default::default())),
3408 background_execution_allowed: Arc::new(AtomicBool::new(false)),
3409 #[cfg(target_arch = "wasm32")]
3410 wasm_accept_bridge_started: Arc::new(std::sync::atomic::AtomicBool::new(false)),
3411 #[cfg(target_arch = "wasm32")]
3412 last_emitted_connection_states: Arc::new(Mutex::new(std::collections::HashMap::new())),
3413 #[cfg(target_arch = "wasm32")]
3414 last_empty_peer_sessions_warning_ms: Arc::new(AtomicU64::new(0)),
3415 iroh_init_guard: Arc::new(tokio::sync::Mutex::new(())),
3416 #[cfg(not(target_arch = "wasm32"))]
3417 native_ip_exclusions: Arc::new(RwLock::new(self.native_ip_exclusions)),
3418 #[cfg(all(not(target_arch = "wasm32"), feature = "native-udp-socket-initializer"))]
3419 native_udp_socket_initializer: self.native_udp_socket_initializer,
3420 #[cfg(all(not(target_arch = "wasm32"), feature = "native-relay-tcp-socket-initializer"))]
3421 native_relay_tcp_socket_initializer: self.native_relay_tcp_socket_initializer,
3422 #[cfg(all(not(target_arch = "wasm32"), feature = "native-dns-socket-initializer"))]
3423 native_dns_socket_initializer: self.native_dns_socket_initializer,
3424 #[cfg(any(feature = "transport-webrtc", feature = "transport-moq"))]
3425 iroh_packet_carriers: Arc::new(RwLock::new(HashMap::new())),
3426 managed_connect_gates: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
3427 #[cfg(not(target_arch = "wasm32"))]
3428 iroh_path_watcher_stable_ids: Arc::new(Mutex::new(HashSet::new())),
3429 #[cfg(not(target_arch = "wasm32"))]
3430 native_custom_transport_kinds: Arc::new(RwLock::new(HashMap::new())),
3431 #[cfg(not(target_arch = "wasm32"))]
3432 native_transport_upgrade_providers: Arc::new(RwLock::new(HashMap::new())),
3433 #[cfg(target_arch = "wasm32")]
3434 wasm_transport_upgrade_gates: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
3435 #[cfg(not(target_arch = "wasm32"))]
3436 native_transport_upgrade_gates: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
3437 #[cfg(not(target_arch = "wasm32"))]
3438 native_ble_upgrade_attempts: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
3439 #[cfg(not(target_arch = "wasm32"))]
3440 native_peer_transport_capabilities: Arc::new(RwLock::new(HashMap::new())),
3441 presence_loop_tx: Arc::new(Mutex::new(None)),
3442 #[cfg(not(target_arch = "wasm32"))]
3443 project_id: self.project_id,
3444 #[cfg(not(target_arch = "wasm32"))]
3445 token_provider: self.token_provider,
3446 transport_config: Arc::new(RwLock::new(
3447 self.transport_config.sanitize_for_client_construction(),
3448 )),
3449 iroh_carrier_policy_epoch: Arc::new(AtomicU64::new(0)),
3450 iroh_carrier_peer_policy_epochs: Arc::new(StdRwLock::new(HashMap::new())),
3451 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
3452 local_discovery_registry: crate::local_discovery::LocalDiscoveryRegistry::new(),
3453 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-lan"))]
3454 mdns_address_lookup: Arc::new(RwLock::new(None)),
3455 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-webrtc"))]
3456 native_webrtc_carrier_attempts: Arc::new(RwLock::new(HashMap::new())),
3457 #[cfg(all(
3458 not(target_arch = "wasm32"),
3459 feature = "iroh-carrier-core",
3460 any(feature = "test-harness", feature = "testing-endpoints")
3461 ))]
3462 native_iroh_carrier_debug_events: Arc::new(StdRwLock::new(
3463 std::collections::VecDeque::new(),
3464 )),
3465 #[cfg(all(not(target_arch = "wasm32"), feature = "transport-moq"))]
3466 native_moq_carrier_attempts: Arc::new(RwLock::new(HashMap::new())),
3467 #[cfg(not(target_arch = "wasm32"))]
3468 native_control_streams: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
3469 #[cfg(not(target_arch = "wasm32"))]
3470 native_transport_protocol_activity: Arc::new(StdRwLock::new(HashMap::new())),
3471 #[cfg(all(not(target_arch = "wasm32"), feature = "experimental-scoped-actor"))]
3472 scoped_connection_actor_registry: Arc::new(RwLock::new(None)),
3473 #[cfg(not(target_arch = "wasm32"))]
3474 auth_readiness: Arc::new(crate::client::auth_readiness::AuthReadinessStore::new()),
3475 #[cfg(not(target_arch = "wasm32"))]
3476 scope_classifier: Arc::new(RwLock::new(default_classifier())),
3477 }
3478 }
3479}
3480
3481mod admission_impl;
3489mod application_crypto_impl;
3490#[cfg(not(target_arch = "wasm32"))]
3491pub mod auth_readiness;
3492mod transfer_peer_demand;
3493pub use transfer_peer_demand::{
3494 TransferPeerDemandPurpose, TransferPeerDemandRequest, TransferPeerDemandState,
3495 TransferPeerDemandStatus,
3496};
3497mod auto_connect_impl;
3498mod core_impl;
3499mod diagnostics_impl;
3500#[cfg(not(target_arch = "wasm32"))]
3501pub mod correlation;
3502#[cfg(all(not(target_arch = "wasm32"), feature = "experimental-scoped-actor"))]
3503mod drive_grant_actor;
3504#[cfg(all(
3505 test,
3506 not(target_arch = "wasm32"),
3507 feature = "experimental-scoped-actor"
3508))]
3509mod drive_grant_actor_tests;
3510#[cfg(not(target_arch = "wasm32"))]
3511mod native_ip_exclusions;
3512#[cfg(all(
3513 test,
3514 not(target_arch = "wasm32"),
3515 any(
3516 feature = "native-udp-socket-initializer",
3517 feature = "native-relay-tcp-socket-initializer",
3518 feature = "native-dns-socket-initializer"
3519 )
3520))]
3521#[path = "client/native_socket_initializer_tests.rs"]
3522mod native_socket_initializer_tests;
3523#[cfg(not(target_arch = "wasm32"))]
3524pub mod scope_classifier;
3525#[cfg(all(not(target_arch = "wasm32"), feature = "experimental-scoped-actor"))]
3526pub mod scoped_connection_actor;
3527mod sparse_fanout_impl;
3528mod state_signaling_impl;
3529#[cfg(not(target_arch = "wasm32"))]
3530mod transport_upgrade_impl;
3531#[cfg(all(
3532 target_arch = "wasm32",
3533 any(feature = "transport-webrtc", feature = "transport-moq")
3534))]
3535mod wasm_carrier_impl;
3536#[cfg(all(not(target_arch = "wasm32"), test))]
3537pub(crate) use transport_upgrade_impl::{
3538 retire_losing_native_control_candidate, retire_replaced_native_control_send,
3539};
3540
3541impl Client {
3542 pub fn broadcasts(&self) -> crate::broadcast::Broadcasts {
3545 self.broadcasts.clone()
3546 }
3547}
3548
3549#[cfg(any(
3550 not(target_arch = "wasm32"),
3551 feature = "transport-webrtc",
3552 feature = "transport-moq"
3553))]
3554impl Client {
3555 pub(crate) fn current_iroh_carrier_policy_epoch(&self) -> u64 {
3556 self.iroh_carrier_policy_epoch.load(Ordering::Acquire)
3557 }
3558
3559 #[cfg(feature = "iroh-carrier-core")]
3560 pub(crate) fn bump_iroh_carrier_policy_epoch(&self) -> u64 {
3561 self.iroh_carrier_policy_epoch
3562 .fetch_add(1, Ordering::AcqRel)
3563 .saturating_add(1)
3564 }
3565
3566 #[cfg(feature = "iroh-carrier-core")]
3567 pub(crate) fn iroh_carrier_policy_epoch_is_current(&self, policy_epoch: u64) -> bool {
3568 self.current_iroh_carrier_policy_epoch() == policy_epoch
3569 }
3570
3571 pub(crate) fn current_iroh_carrier_peer_policy_epoch(&self, connection_id: &str) -> u64 {
3572 self.iroh_carrier_peer_policy_epochs
3573 .read()
3574 .ok()
3575 .and_then(|epochs| epochs.get(connection_id).copied())
3576 .unwrap_or_default()
3577 }
3578
3579 #[cfg(feature = "iroh-carrier-core")]
3580 pub(crate) fn bump_iroh_carrier_peer_policy_epoch(&self, connection_id: &str) -> u64 {
3581 let Ok(mut epochs) = self.iroh_carrier_peer_policy_epochs.write() else {
3582 return self.current_iroh_carrier_peer_policy_epoch(connection_id);
3583 };
3584 let next = epochs
3585 .get(connection_id)
3586 .copied()
3587 .unwrap_or_default()
3588 .saturating_add(1);
3589 epochs.insert(connection_id.to_string(), next);
3590 next
3591 }
3592
3593 #[cfg(feature = "iroh-carrier-core")]
3594 pub(crate) fn forget_iroh_carrier_peer_policy_epoch(&self, connection_id: &str) {
3595 if let Ok(mut epochs) = self.iroh_carrier_peer_policy_epochs.write() {
3596 epochs.remove(connection_id);
3597 }
3598 }
3599}
3600
3601#[cfg(target_arch = "wasm32")]
3606impl Client {
3607 pub(crate) async fn set_relay(&self, enabled: bool) -> anyhow::Result<()> {
3608 anyhow::ensure!(
3609 self.iroh_endpoint.read().await.is_none(),
3610 "relay must be configured before Iroh endpoint initialization",
3611 );
3612 let mut config = self.transport_config.read().await.clone();
3613 config.relay = enabled;
3614 *self.transport_config.write().await = config.sanitize_for_runtime();
3615 Ok(())
3616 }
3617
3618 pub async fn send_peer(&self, _id: &str, _data: &[u8]) -> anyhow::Result<()> {
3619 Err(anyhow::anyhow!(
3620 "send_peer: use TypeScript Connection.sendTyped() for transport-aware sending in browser environments"
3621 ))
3622 }
3623}
3624
3625#[cfg(not(target_arch = "wasm32"))]
3626impl Client {
3627 pub fn subscribe_native_peer_data(
3628 &self,
3629 ) -> tokio::sync::broadcast::Receiver<NativePeerDataEvent> {
3630 self.native_peer_data_updates.subscribe()
3631 }
3632
3633 #[cfg_attr(
3634 not(any(feature = "transport-webrtc", feature = "transport-moq")),
3635 allow(dead_code)
3636 )]
3637 pub(crate) fn emit_native_peer_data(&self, event: NativePeerDataEvent) {
3638 let _ = self.native_peer_data_updates.send(event);
3639 }
3640}
3641
3642#[cfg(test)]
3643mod tests;
3644
3645#[cfg(all(test, not(target_arch = "wasm32")))]
3646mod sparse_iroh_lane_tests;
3647
3648#[cfg(all(test, not(target_arch = "wasm32")))]
3649mod relay_diagnostic_truth_tests {
3650 use super::*;
3651
3652 #[test]
3653 fn auto_reports_requested_and_effective_relay_carriage() {
3654 let diagnostic = RelayStatus::from_policy(RelayPolicy::Auto, current_iroh_relay_provider());
3655
3656 assert_eq!(diagnostic.requested_carriage, RelayPolicy::Auto);
3657 let serialized = serde_json::to_value(&diagnostic).expect("relay diagnostic serializes");
3658 assert_eq!(serialized["requestedCarriage"], "auto");
3659 #[cfg(openrtc_iroh_relay_transport_policy_api)]
3660 assert_eq!(
3661 diagnostic.effective_carriage,
3662 RelayCarriage::QuicWithWebsocketFallback
3663 );
3664 #[cfg(not(openrtc_iroh_relay_transport_policy_api))]
3665 assert_eq!(
3666 diagnostic.effective_carriage,
3667 RelayCarriage::WebsocketFallback
3668 );
3669 }
3670
3671 #[cfg(not(openrtc_iroh_relay_transport_policy_api))]
3672 #[test]
3673 fn upstream_quic_required_fails_closed_before_endpoint_bind() {
3674 let result = apply_native_network_preferences(
3675 iroh::Endpoint::builder(iroh::endpoint::presets::N0),
3676 false,
3677 RelayPolicy::QuicRequired,
3678 );
3679
3680 assert!(result.is_err());
3681 assert!(result
3682 .err()
3683 .expect("unsupported QUIC policy should return an error")
3684 .to_string()
3685 .contains("unsupported by this upstream-Iroh build"));
3686 }
3687
3688 #[cfg(openrtc_iroh_relay_transport_policy_api)]
3689 #[test]
3690 fn vendored_quic_required_reports_quic_and_keeps_endpoint_policy_available() {
3691 let diagnostic =
3692 RelayStatus::from_policy(RelayPolicy::QuicRequired, current_iroh_relay_provider());
3693
3694 assert_eq!(diagnostic.effective_carriage, RelayCarriage::Quic);
3695 assert_eq!(diagnostic.provider, IrohRelayProvider::VendoredIroh);
3696 }
3697}
3698
3699#[cfg(all(test, native))]
3700mod managed_scope_persistence_tests;