1#[cfg(test)]
6use super::state_signaling_impl::should_present_session_token_on_connected_transport;
7use super::*;
8
9#[cfg(test)]
10pub(crate) fn should_auto_present_session_token(
11 has_extracted_token: bool,
12 has_current_remote_admission_proof: bool,
13) -> bool {
14 should_present_session_token_on_connected_transport(
15 has_extracted_token,
16 has_current_remote_admission_proof,
17 )
18}
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21#[cfg(not(target_arch = "wasm32"))]
22pub(crate) struct SessionAdmissionPresentationDecision {
23 pub(crate) should_present: bool,
24 pub(crate) request_reciprocal: bool,
25}
26
27#[cfg(not(target_arch = "wasm32"))]
35pub(crate) fn session_admission_presentation_decision(
36 has_extracted_token: bool,
37 has_current_remote_admission_proof: bool,
38 has_current_inbound_admission_proof: bool,
39 peer_requested_reciprocal: bool,
40 is_designated_initiator: bool,
41) -> SessionAdmissionPresentationDecision {
42 if !has_extracted_token {
43 return SessionAdmissionPresentationDecision {
44 should_present: false,
45 request_reciprocal: false,
46 };
47 }
48
49 let bilateral_proof_is_current =
57 has_current_remote_admission_proof && has_current_inbound_admission_proof;
58 if bilateral_proof_is_current {
59 return SessionAdmissionPresentationDecision {
60 should_present: false,
61 request_reciprocal: false,
62 };
63 }
64
65 if !has_current_remote_admission_proof && !has_current_inbound_admission_proof {
73 return SessionAdmissionPresentationDecision {
74 should_present: is_designated_initiator,
75 request_reciprocal: is_designated_initiator,
76 };
77 }
78
79 if peer_requested_reciprocal {
80 return SessionAdmissionPresentationDecision {
81 should_present: true,
82 request_reciprocal: false,
83 };
84 }
85
86 let request_reciprocal = has_current_remote_admission_proof && is_designated_initiator;
91 SessionAdmissionPresentationDecision {
92 should_present: !has_current_remote_admission_proof || request_reciprocal,
93 request_reciprocal,
94 }
95}
96
97#[cfg(not(target_arch = "wasm32"))]
111pub(crate) fn persistent_managed_actor_may_own_transport(
112 current_scopes: &[String],
113 admitted_scope: Option<&str>,
114 presentation_in_flight: bool,
115) -> bool {
116 if presentation_in_flight {
117 return false;
118 }
119
120 current_scopes
121 .iter()
122 .map(String::as_str)
123 .chain(admitted_scope)
124 .map(str::trim)
125 .filter(|scope| !scope.is_empty())
126 .all(|scope| scope == super::core_impl::PERSISTENT_MANAGED_ADMISSION_SCOPE)
127}
128
129#[cfg(not(target_arch = "wasm32"))]
130#[derive(Debug, Clone, Copy, PartialEq, Eq)]
131pub(crate) enum ActiveConnectionProbeDecision {
132 Healthy,
133 PreserveLiveTransport,
134 Failed,
135}
136
137#[cfg(not(target_arch = "wasm32"))]
138#[derive(Debug, Clone, Copy, PartialEq, Eq)]
139enum FailedProbeRetirementOutcome {
140 Retired,
141 DeferredAdmissionInFlight,
142 CancelledCurrentGenerationActivity,
143 GenerationChanged,
144}
145
146const ACTIVE_CONNECTION_PROBE_TIMEOUT_MS: i64 = 4_000;
147
148#[cfg(all(
156 not(target_arch = "wasm32"),
157 openrtc_iroh_retire_remote_connections_api
158))]
159pub(crate) async fn retire_stale_endpoint_connections(
160 endpoint: &iroh::Endpoint,
161 endpoint_id: iroh::EndpointId,
162 current_transport: &iroh::endpoint::Connection,
163 reason: &[u8],
164) -> usize {
165 endpoint
166 .retire_remote_connections_except(endpoint_id, current_transport.stable_id(), reason)
167 .await
168}
169
170pub(crate) fn generation_activity_is_recent(
171 last_inbound_activity_age: Option<std::time::Duration>,
172 probe_interval_ms: i64,
173) -> bool {
174 let activity_window_ms = probe_interval_ms
175 .max(0)
176 .saturating_add(ACTIVE_CONNECTION_PROBE_TIMEOUT_MS) as u64;
177 last_inbound_activity_age
178 .is_some_and(|age| age <= std::time::Duration::from_millis(activity_window_ms))
179}
180
181#[cfg(not(target_arch = "wasm32"))]
182impl Client {
183 pub(crate) async fn wake_native_auto_connect_owner(&self) -> bool {
187 if self.external_auto_connect_is_active().await {
188 return self.wake_native_external_auto_connect().await;
189 }
190
191 let provider_actor_active = self
192 .auto_connect_loop_key
193 .lock()
194 .unwrap_or_else(|poisoned| poisoned.into_inner())
195 .is_some();
196 if provider_actor_active {
197 self.native_auto_connect_wake.notify_one();
198 }
199 provider_actor_active
200 }
201
202 pub(crate) fn record_native_transport_protocol_activity(
208 &self,
209 connection_id: &str,
210 transport_stable_id: u64,
211 ) {
212 if let Ok(mut activity) = self.native_transport_protocol_activity.write() {
213 activity.insert(
214 connection_id.to_string(),
215 super::NativeTransportProtocolActivity {
216 transport_stable_id,
217 observed_at: std::time::Instant::now(),
218 },
219 );
220 }
221 }
222
223 pub(crate) fn native_transport_protocol_activity_age(
224 &self,
225 connection_id: &str,
226 expected_transport_stable_id: u64,
227 ) -> Option<std::time::Duration> {
228 let observed_at = self
229 .native_transport_protocol_activity
230 .read()
231 .ok()
232 .and_then(|activity| activity.get(connection_id).copied())
233 .filter(|activity| activity.transport_stable_id == expected_transport_stable_id)?
234 .observed_at;
235 Some(std::time::Instant::now().saturating_duration_since(observed_at))
236 }
237
238 pub(crate) fn forget_native_transport_protocol_activity_for_transport(
239 &self,
240 connection_id: &str,
241 expected_transport_stable_id: u64,
242 ) -> bool {
243 self.native_transport_protocol_activity
244 .write()
245 .ok()
246 .is_some_and(|mut activity| {
247 if activity
248 .get(connection_id)
249 .is_some_and(|entry| entry.transport_stable_id == expected_transport_stable_id)
250 {
251 activity.remove(connection_id);
252 true
253 } else {
254 false
255 }
256 })
257 }
258
259 pub(crate) fn forget_native_transport_protocol_activity(&self, connection_id: &str) {
260 if let Ok(mut activity) = self.native_transport_protocol_activity.write() {
261 activity.remove(connection_id);
262 }
263 }
264
265 async fn retire_transport_generation_after_failed_probe(
273 &self,
274 connection_id: &str,
275 endpoint_id: iroh::EndpointId,
276 expected_transport_stable_id: u64,
277 activity_window_ms: i64,
278 reason: &str,
279 ) -> FailedProbeRetirementOutcome {
280 let Some(retirement_guard) = self
281 .session_token_registry
282 .try_begin_transport_retirement(connection_id)
283 else {
284 return FailedProbeRetirementOutcome::DeferredAdmissionInFlight;
285 };
286
287 let current_transport_stable_id = self
288 .get_connection(endpoint_id)
289 .await
290 .map(|connection| crate::transport_generation::for_connection(&connection));
291 if current_transport_stable_id != Some(expected_transport_stable_id) {
292 drop(retirement_guard);
293 return FailedProbeRetirementOutcome::GenerationChanged;
294 }
295
296 if generation_activity_is_recent(
297 self.native_transport_protocol_activity_age(
298 connection_id,
299 expected_transport_stable_id,
300 ),
301 activity_window_ms,
302 ) {
303 drop(retirement_guard);
304 return FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity;
305 }
306
307 let retired = self
308 .disconnect_transport_generation_with_reason(
309 endpoint_id,
310 expected_transport_stable_id,
311 reason,
312 )
313 .await
314 .unwrap_or(false);
315 drop(retirement_guard);
316
317 if retired {
318 FailedProbeRetirementOutcome::Retired
319 } else {
320 FailedProbeRetirementOutcome::GenerationChanged
321 }
322 }
323}
324
325#[cfg(target_arch = "wasm32")]
326impl Client {
327 pub(crate) fn native_transport_protocol_activity_age(
328 &self,
329 _connection_id: &str,
330 _expected_transport_stable_id: u64,
331 ) -> Option<std::time::Duration> {
332 None
333 }
334}
335
336#[cfg(not(target_arch = "wasm32"))]
337pub(crate) fn active_connection_probe_decision(
338 active_probe_healthy: bool,
339 passive_transport_alive: bool,
340) -> ActiveConnectionProbeDecision {
341 if active_probe_healthy {
342 ActiveConnectionProbeDecision::Healthy
343 } else if passive_transport_alive {
344 ActiveConnectionProbeDecision::PreserveLiveTransport
345 } else {
346 ActiveConnectionProbeDecision::Failed
347 }
348}
349
350#[cfg(not(target_arch = "wasm32"))]
351pub(crate) fn active_connection_probe_decision_with_generation_activity(
352 active_probe_healthy: bool,
353 passive_transport_alive: bool,
354 recent_generation_activity: bool,
355) -> ActiveConnectionProbeDecision {
356 if recent_generation_activity {
357 ActiveConnectionProbeDecision::Healthy
358 } else {
359 active_connection_probe_decision(active_probe_healthy, passive_transport_alive)
360 }
361}
362
363#[cfg(not(target_arch = "wasm32"))]
364pub(crate) fn next_active_connection_probe_failure_count(
365 decision: ActiveConnectionProbeDecision,
366 current_failures: u8,
367) -> u8 {
368 if matches!(decision, ActiveConnectionProbeDecision::Healthy) {
369 0
370 } else {
371 current_failures.saturating_add(1).min(10)
372 }
373}
374
375#[cfg(not(target_arch = "wasm32"))]
376pub(crate) fn active_connection_probe_is_healthy(
377 decision: ActiveConnectionProbeDecision,
378 failure_count: u8,
379 failure_threshold: u8,
380) -> bool {
381 matches!(decision, ActiveConnectionProbeDecision::Healthy)
382 || (matches!(
383 decision,
384 ActiveConnectionProbeDecision::PreserveLiveTransport
385 ) && failure_count < failure_threshold)
386}
387
388#[derive(Debug, Clone, Copy, PartialEq, Eq)]
394pub(crate) enum SessionTokenPresentationFailureAction {
395 RetryOnNextTransport,
396 RejectAndDisconnect,
397}
398
399#[derive(Debug, Clone, Copy, PartialEq, Eq)]
400#[cfg(not(target_arch = "wasm32"))]
401pub(crate) enum RetryableSessionAdmissionFailureDecision {
402 PreserveCurrentTransport,
403 RetireCurrentTransport,
404 ReplacementAlreadyWon,
405}
406
407pub(crate) fn superseded_desired_peer_was_withdrawn(
412 actor_is_current: bool,
413 current_revision: u64,
414 attempt_revision: u64,
415 peer_is_still_desired: bool,
416) -> bool {
417 actor_is_current && current_revision > attempt_revision && !peer_is_still_desired
418}
419
420#[cfg(not(target_arch = "wasm32"))]
428pub(crate) fn retryable_session_admission_failure_decision(
429 failure_count: u8,
430 failure_threshold: u8,
431 attempted_transport_stable_id: Option<u64>,
432 current_transport_stable_id: Option<u64>,
433 remote_runtime_instance_changed: bool,
434) -> RetryableSessionAdmissionFailureDecision {
435 if attempted_transport_stable_id != current_transport_stable_id
436 || attempted_transport_stable_id.is_none()
437 {
438 return RetryableSessionAdmissionFailureDecision::ReplacementAlreadyWon;
439 }
440
441 if !remote_runtime_instance_changed && failure_count < failure_threshold.max(1) {
449 RetryableSessionAdmissionFailureDecision::PreserveCurrentTransport
450 } else {
451 RetryableSessionAdmissionFailureDecision::RetireCurrentTransport
452 }
453}
454
455pub(crate) fn session_token_presentation_failure_action(
456 error: &str,
457) -> SessionTokenPresentationFailureAction {
458 let error = error.to_ascii_lowercase();
459 if error.contains("connection lost")
460 || error.contains("[session-token-response:timeout]")
461 || error.contains("stream finished early")
462 || error.contains("peer stream ended before")
463 || error.contains("connection closed")
464 || error.contains("stream reset")
465 || error.contains("stream stopped")
466 || error.contains("failed to open bi-stream")
467 || error.contains("failed to write session-token presentation")
468 || error.contains("missing connection for session-token presentation")
469 || error.contains("[session-token-presentation:in-flight]")
470 || error.contains("authoritative-device-binding-pending")
471 || error.contains(crate::session_token::PEER_ADMISSION_LEASE_EXPIRED)
472 {
473 SessionTokenPresentationFailureAction::RetryOnNextTransport
474 } else {
475 SessionTokenPresentationFailureAction::RejectAndDisconnect
476 }
477}
478
479#[cfg(not(target_arch = "wasm32"))]
480pub(crate) fn session_token_presentation_was_duplicate(error: &str) -> bool {
481 error
482 .to_ascii_lowercase()
483 .contains("[session-token-presentation:in-flight]")
484}
485
486fn remote_excludes_local_device(excluded_peers: &[String], local_device_id: &str) -> bool {
487 let local_device_id = local_device_id.trim();
488 !local_device_id.is_empty()
489 && excluded_peers
490 .iter()
491 .any(|peer| peer.trim().eq_ignore_ascii_case(local_device_id))
492}
493
494#[cfg(not(target_arch = "wasm32"))]
495fn observe_remote_runtime_instance(
496 last_known: &mut std::collections::HashMap<String, String>,
497 device_id: &str,
498 runtime_instance_id: Option<&str>,
499) -> bool {
500 let Some(runtime_instance_id) = runtime_instance_id
501 .map(str::trim)
502 .filter(|value| !value.is_empty() && *value != device_id)
503 else {
504 return false;
505 };
506 let key = format!("{device_id}::runtime-instance");
507 last_known
508 .insert(key, runtime_instance_id.to_string())
509 .is_some_and(|previous| previous != runtime_instance_id)
510}
511
512#[cfg(not(target_arch = "wasm32"))]
513fn runtime_instance_change_requires_replacement(
514 runtime_instance_changed: bool,
515 active_transport_responsive: bool,
516) -> bool {
517 runtime_instance_changed && !active_transport_responsive
518}
519
520#[cfg(not(target_arch = "wasm32"))]
521fn observe_remote_ticket(
522 last_known: &mut std::collections::HashMap<String, String>,
523 device_id: &str,
524 ticket: &str,
525) -> bool {
526 let ticket = ticket.trim();
527 if ticket.is_empty() {
528 return false;
529 }
530 let key = format!("{device_id}::ticket");
531 let fingerprint = crate::session_token::token_fingerprint(ticket);
532 last_known
533 .insert(key, fingerprint.clone())
534 .is_some_and(|previous| previous != fingerprint)
535}
536
537#[cfg(not(target_arch = "wasm32"))]
538fn reconcile_remote_ticket_retry_state(
539 last_known: &mut std::collections::HashMap<String, String>,
540 device_id: &str,
541 ticket: &str,
542 last_attempt_at: &mut std::collections::HashMap<String, i64>,
543 failure_count: &mut std::collections::HashMap<String, u8>,
544 failure_backoff_until: &mut std::collections::HashMap<String, i64>,
545) -> bool {
546 if !observe_remote_ticket(last_known, device_id, ticket) {
547 return false;
548 }
549 last_attempt_at.remove(device_id);
550 clear_auto_connect_failure_state(device_id, failure_count, failure_backoff_until);
551 true
552}
553
554#[derive(Debug, Clone, PartialEq, Eq, serde::Deserialize)]
555#[serde(rename_all = "camelCase")]
556pub(crate) struct BrowserDesiredPeer {
557 pub(crate) device_id: String,
558 #[serde(default)]
559 pub(crate) node_id: Option<String>,
560 #[serde(default)]
561 pub(crate) ticket: Option<String>,
562 #[serde(default)]
563 pub(crate) online: bool,
564 #[serde(default)]
565 pub(crate) session_id: Option<String>,
566 #[serde(default)]
567 pub(crate) excluded_peers: Vec<String>,
568 #[serde(default)]
569 pub(crate) topology_role: Option<SparseTopologyRole>,
570 #[serde(default)]
571 pub(crate) topology_revision: Option<u64>,
572}
573
574#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Deserialize)]
575#[serde(rename_all = "lowercase")]
576pub(crate) enum SparseTopologyRole {
577 Active,
578 Backup,
579}
580
581#[derive(Debug, Default)]
582struct BrowserDesiredPeerMailbox {
583 revision: u64,
584 peers: std::collections::BTreeMap<String, BrowserDesiredPeer>,
585 pending_topology_revision: Option<u64>,
586 pending_ready_active_peers: std::collections::HashSet<String>,
587 settled_active: std::collections::BTreeMap<String, BrowserDesiredPeer>,
588 withdrawn: std::collections::BTreeMap<String, Option<String>>,
597}
598
599impl BrowserDesiredPeerMailbox {
600 #[cfg(test)]
601 fn accept(&mut self, revision: u64, peers: Vec<BrowserDesiredPeer>) -> bool {
602 self.accept_with_withdrawn(revision, peers).is_some()
603 }
604
605 fn accept_with_withdrawn(
611 &mut self,
612 revision: u64,
613 peers: Vec<BrowserDesiredPeer>,
614 ) -> Option<Vec<BrowserDesiredPeer>> {
615 self.accept_update(revision, peers, None)
616 }
617
618 fn accept_with_observation_expired(
622 &mut self,
623 revision: u64,
624 peers: Vec<BrowserDesiredPeer>,
625 expired_device_id: &str,
626 ) -> Option<Vec<BrowserDesiredPeer>> {
627 let expired_device_id = expired_device_id.trim();
628 if expired_device_id.is_empty() {
629 return None;
630 }
631 self.accept_update(revision, peers, Some(expired_device_id))
632 }
633
634 fn accept_update(
635 &mut self,
636 revision: u64,
637 peers: Vec<BrowserDesiredPeer>,
638 observation_expired_device_id: Option<&str>,
639 ) -> Option<Vec<BrowserDesiredPeer>> {
640 if revision <= self.revision {
641 return None;
642 }
643 let mut peers = peers
644 .into_iter()
645 .filter_map(|mut peer| {
646 peer.device_id = peer.device_id.trim().to_string();
647 if peer.device_id.is_empty() {
648 return None;
649 }
650 peer.node_id = peer
651 .node_id
652 .take()
653 .map(|value| value.trim().to_string())
654 .filter(|value| !value.is_empty());
655 peer.ticket = peer
656 .ticket
657 .take()
658 .map(|value| value.trim().to_string())
659 .filter(|value| !value.is_empty());
660 peer.session_id = peer
661 .session_id
662 .take()
663 .map(|value| value.trim().to_string())
664 .filter(|value| !value.is_empty());
665 peer.excluded_peers = peer
666 .excluded_peers
667 .into_iter()
668 .map(|value| value.trim().to_ascii_lowercase())
669 .filter(|value| !value.is_empty())
670 .collect();
671 Some((peer.device_id.clone(), peer))
672 })
673 .collect::<std::collections::BTreeMap<_, _>>();
674 let topology_revision = peers
675 .values()
676 .filter_map(|peer| peer.topology_revision)
677 .max();
678 if topology_revision == self.pending_topology_revision {
679 self.pending_ready_active_peers.retain(|device_id| {
680 let Some(previous) = self.peers.get(device_id) else {
681 return false;
682 };
683 peers.get(device_id).is_some_and(|current| {
684 current.topology_role == Some(SparseTopologyRole::Active)
685 && browser_peer_evidence(current) == browser_peer_evidence(previous)
686 })
687 });
688 } else {
689 self.pending_ready_active_peers.clear();
690 if topology_revision.is_none() {
691 self.settled_active.clear();
692 } else if !self.settled_active.is_empty() {
693 peers.retain(|_, peer| peer.topology_role != Some(SparseTopologyRole::Backup));
698 for (device_id, settled) in self.settled_active.iter().take(4) {
699 let still_authorized =
700 peers.get(device_id).is_some_and(|current| current.online);
701 let is_current_active = peers.get(device_id).is_some_and(|current| {
702 current.topology_role == Some(SparseTopologyRole::Active)
703 });
704 if still_authorized && !is_current_active {
705 let mut overlap = settled.clone();
706 overlap.topology_role = Some(SparseTopologyRole::Backup);
707 overlap.topology_revision = topology_revision;
708 peers.insert(device_id.clone(), overlap);
709 }
710 }
711 }
712 }
713 let active_ids = peers
714 .values()
715 .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Active))
716 .map(|peer| peer.device_id.clone())
717 .collect::<std::collections::HashSet<_>>();
718 if !active_ids.is_empty() && active_ids.is_subset(&self.pending_ready_active_peers) {
719 peers.retain(|_, peer| peer.topology_role != Some(SparseTopologyRole::Backup));
720 }
721 let withdrawn = self
722 .peers
723 .values()
724 .filter(|peer| {
725 peer.online
726 && observation_expired_device_id != Some(peer.device_id.as_str())
727 && peers
728 .get(&peer.device_id)
729 .is_none_or(|current| !current.online)
730 })
731 .cloned()
732 .collect::<Vec<_>>();
733 for peer in &withdrawn {
734 self.withdrawn
735 .insert(peer.device_id.clone(), peer.node_id.clone());
736 }
737 for peer in peers.values() {
738 if peer.online {
739 self.withdrawn.remove(&peer.device_id);
740 }
741 }
742 self.revision = revision;
743 self.pending_topology_revision = topology_revision;
744 self.peers = peers;
745 Some(withdrawn)
746 }
747
748 fn acknowledge_topology_peer_ready(
753 &mut self,
754 device_id: &str,
755 topology_revision: Option<u64>,
756 ) -> Vec<BrowserDesiredPeer> {
757 let Some(topology_revision) = topology_revision else {
758 return Vec::new();
759 };
760 if self.pending_topology_revision != Some(topology_revision)
761 || !self.peers.get(device_id).is_some_and(|peer| {
762 peer.topology_revision == Some(topology_revision)
763 && peer.topology_role == Some(SparseTopologyRole::Active)
764 })
765 {
766 return Vec::new();
767 }
768 self.pending_ready_active_peers
769 .insert(device_id.to_string());
770 let active_ids = self
771 .peers
772 .values()
773 .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Active))
774 .map(|peer| peer.device_id.clone())
775 .collect::<std::collections::HashSet<_>>();
776 if active_ids.is_empty() || !active_ids.is_subset(&self.pending_ready_active_peers) {
777 return Vec::new();
778 }
779 self.settled_active = self
780 .peers
781 .iter()
782 .filter(|(_, peer)| peer.topology_role == Some(SparseTopologyRole::Active))
783 .map(|(device_id, peer)| (device_id.clone(), peer.clone()))
784 .take(4)
785 .collect();
786 let backups = self
787 .peers
788 .values()
789 .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
790 .cloned()
791 .collect::<Vec<_>>();
792 for peer in &backups {
793 self.peers.remove(&peer.device_id);
794 self.withdrawn
795 .insert(peer.device_id.clone(), peer.node_id.clone());
796 }
797 backups
798 }
799
800 fn observe_topology_peer_not_ready(&mut self, device_id: &str, topology_revision: Option<u64>) {
801 if self.pending_topology_revision == topology_revision
802 && self.peers.get(device_id).is_some_and(|peer| {
803 peer.topology_revision == topology_revision
804 && peer.topology_role == Some(SparseTopologyRole::Active)
805 })
806 {
807 self.pending_ready_active_peers.remove(device_id);
808 }
809 }
810
811 fn device_is_withdrawn(&self, device_id: &str) -> bool {
812 self.withdrawn.contains_key(device_id.trim())
813 }
814
815 fn withdrawn_device_for_node(&self, node_id: &str) -> Option<String> {
816 let node_id = node_id.trim();
817 if node_id.is_empty() {
818 return None;
819 }
820 self.withdrawn
821 .iter()
822 .find_map(|(device_id, withdrawn_node_id)| {
823 (withdrawn_node_id.as_deref() == Some(node_id)).then(|| device_id.clone())
824 })
825 }
826
827 fn current_online_device_for_node(&self, node_id: &str) -> Option<String> {
828 let node_id = node_id.trim();
829 if node_id.is_empty() {
830 return None;
831 }
832 self.peers.values().find_map(|peer| {
833 if !peer.online {
834 return None;
835 }
836 let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
837 let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
838 let payload = token_suffix
839 .and_then(|suffix| crate::session_token::decode_payload(iroh_ticket, suffix))?;
840 if payload.scope.as_str() != "user-device" {
841 return None;
842 }
843 parse_endpoint_ticket(iroh_ticket)
844 .ok()
845 .map(|address| address.id.to_string())
846 });
847 let ticket_node_id = ticket_node_id.as_deref()?;
848 let resolved_node_id = match peer.node_id.as_deref() {
849 Some(directory) if directory != ticket_node_id => return None,
850 Some(directory) => directory,
851 None => ticket_node_id,
852 };
853 (resolved_node_id == node_id).then(|| peer.device_id.clone())
854 })
855 }
856
857 fn current_online_assignment_matches(
858 &self,
859 device_id: &str,
860 node_id: &str,
861 ticket: &str,
862 evidence: &str,
863 ) -> bool {
864 self.peers.get(device_id.trim()).is_some_and(|peer| {
865 peer.online
866 && peer.ticket.as_deref() == Some(ticket.trim())
867 && browser_peer_evidence(peer) == evidence
868 && self.current_online_device_for_node(node_id).as_deref()
869 == Some(peer.device_id.as_str())
870 })
871 }
872}
873
874#[cfg(not(target_arch = "wasm32"))]
875#[derive(Debug, Default)]
876pub(crate) struct NativeExternalAutoConnectActorState {
877 active: bool,
878 generation: u64,
879 local_device_id: String,
880 mailbox: BrowserDesiredPeerMailbox,
881 peer_evidence: std::collections::HashMap<String, String>,
882 failure_count: std::collections::HashMap<String, u8>,
883 retry_not_before_ms: std::collections::HashMap<String, i64>,
884 non_initiator_not_before_ms: std::collections::HashMap<String, i64>,
885 scheduled_deadline_ms: Option<i64>,
886 schedule_epoch: u64,
887 reconcile_running: bool,
888 wake_pending: bool,
889 recovery_device_ids: std::collections::HashSet<String>,
892 recovery_suppressed_until_ms: std::collections::HashMap<String, i64>,
895}
896
897#[cfg(target_arch = "wasm32")]
898struct BrowserAutoConnectActorState {
899 active: bool,
900 client: std::sync::Weak<Client>,
901 generation: u64,
902 local_device_id: String,
903 mailbox: BrowserDesiredPeerMailbox,
904 peer_evidence: std::collections::HashMap<String, String>,
905 failure_count: std::collections::HashMap<String, u8>,
906 retry_not_before_ms: std::collections::HashMap<String, i64>,
907 non_initiator_not_before_ms: std::collections::HashMap<String, i64>,
908 scheduled_deadline_ms: Option<i64>,
909 schedule_epoch: u64,
910 reconcile_running: bool,
911 wake_pending: bool,
912}
913
914#[cfg(target_arch = "wasm32")]
915thread_local! {
916 static BROWSER_AUTO_CONNECT_ACTORS: std::cell::RefCell<
917 std::collections::HashMap<usize, std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>>
918 > = std::cell::RefCell::new(std::collections::HashMap::new());
919}
920
921fn browser_auto_connect_failure_backoff_ms(failure_count: u8) -> i64 {
922 500_i64
923 .saturating_mul(1_i64 << (failure_count as u32).min(4))
924 .min(8_000)
925}
926
927fn browser_peer_evidence(peer: &BrowserDesiredPeer) -> String {
928 format!(
929 "{}|{}|{}",
930 peer.node_id.as_deref().unwrap_or_default(),
931 peer.ticket
932 .as_deref()
933 .map(crate::session_token::token_fingerprint)
934 .unwrap_or_default(),
935 peer.session_id.as_deref().unwrap_or_default(),
936 )
937}
938
939fn desired_peer_evidence_is_current(
940 mailbox: &BrowserDesiredPeerMailbox,
941 peer: &BrowserDesiredPeer,
942) -> bool {
943 mailbox
948 .peers
949 .get(&peer.device_id)
950 .is_some_and(|current| browser_peer_evidence(current) == browser_peer_evidence(peer))
951}
952
953#[cfg(target_arch = "wasm32")]
954fn browser_actor_key(client: &Arc<Client>) -> usize {
955 Arc::as_ptr(client) as usize
956}
957
958impl Client {
959 #[cfg(not(target_arch = "wasm32"))]
960 pub(crate) async fn native_external_auto_connect_owner_is_current(
961 &self,
962 user_id: &str,
963 local_device_id: &str,
964 ) -> bool {
965 let expected = (
966 user_id.trim().to_string(),
967 local_device_id.trim().to_string(),
968 );
969 let state = self.external_desired_peer_actor.lock().await;
970 let key = self
971 .auto_connect_loop_key
972 .lock()
973 .unwrap_or_else(|poisoned| poisoned.into_inner());
974 key.as_ref() == Some(&expected)
975 && state.active
976 && state.local_device_id == expected.1
977 && self
978 .auto_connect_generation
979 .load(std::sync::atomic::Ordering::SeqCst)
980 == state.generation
981 }
982
983 pub(crate) async fn withdrawn_desired_device_for_node(
991 &self,
992 remote_node_id: &str,
993 ) -> Option<String> {
994 #[cfg(not(target_arch = "wasm32"))]
995 {
996 let state = self.external_desired_peer_actor.lock().await;
997 if !state.active
998 || self
999 .auto_connect_generation
1000 .load(std::sync::atomic::Ordering::SeqCst)
1001 != state.generation
1002 {
1003 return None;
1004 }
1005 state.mailbox.withdrawn_device_for_node(remote_node_id)
1006 }
1007 #[cfg(target_arch = "wasm32")]
1008 {
1009 let actor_key = self as *const Client as usize;
1010 BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
1011 let actor = actors.borrow().get(&actor_key).cloned()?;
1012 let state = actor.borrow();
1013 browser_actor_is_current(&state)
1014 .then(|| state.mailbox.withdrawn_device_for_node(remote_node_id))
1015 .flatten()
1016 })
1017 }
1018 }
1019
1020 pub(crate) async fn desired_device_is_withdrawn(&self, device_id: &str) -> bool {
1025 #[cfg(not(target_arch = "wasm32"))]
1026 {
1027 let state = self.external_desired_peer_actor.lock().await;
1028 state.active
1029 && self
1030 .auto_connect_generation
1031 .load(std::sync::atomic::Ordering::SeqCst)
1032 == state.generation
1033 && state.mailbox.device_is_withdrawn(device_id)
1034 }
1035 #[cfg(target_arch = "wasm32")]
1036 {
1037 let actor_key = self as *const Client as usize;
1038 BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
1039 actors
1040 .borrow()
1041 .get(&actor_key)
1042 .cloned()
1043 .is_some_and(|actor| {
1044 let state = actor.borrow();
1045 browser_actor_is_current(&state)
1046 && state.mailbox.device_is_withdrawn(device_id)
1047 })
1048 })
1049 }
1050 }
1051
1052 #[cfg(target_arch = "wasm32")]
1053 pub(crate) fn current_desired_user_device_for_node_now(
1054 &self,
1055 remote_node_id: &str,
1056 ) -> Option<String> {
1057 let actor_key = self as *const Client as usize;
1058 BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
1059 let actor = actors.borrow().get(&actor_key).cloned()?;
1060 let state = actor.borrow();
1061 browser_actor_is_current(&state)
1062 .then(|| state.mailbox.current_online_device_for_node(remote_node_id))
1063 .flatten()
1064 })
1065 }
1066
1067 #[cfg(target_arch = "wasm32")]
1068 pub(crate) async fn current_desired_user_device_for_node(
1069 &self,
1070 remote_node_id: &str,
1071 ) -> Option<String> {
1072 self.current_desired_user_device_for_node_now(remote_node_id)
1073 }
1074
1075 pub(crate) async fn current_desired_user_device_assignment_matches(
1076 &self,
1077 device_id: &str,
1078 remote_node_id: &str,
1079 ticket: &str,
1080 evidence: &str,
1081 ) -> bool {
1082 #[cfg(not(target_arch = "wasm32"))]
1083 {
1084 let state = self.external_desired_peer_actor.lock().await;
1085 state.active
1086 && self
1087 .auto_connect_generation
1088 .load(std::sync::atomic::Ordering::SeqCst)
1089 == state.generation
1090 && state.mailbox.current_online_assignment_matches(
1091 device_id,
1092 remote_node_id,
1093 ticket,
1094 evidence,
1095 )
1096 }
1097 #[cfg(target_arch = "wasm32")]
1098 {
1099 let actor_key = self as *const Client as usize;
1100 BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
1101 actors
1102 .borrow()
1103 .get(&actor_key)
1104 .cloned()
1105 .is_some_and(|actor| {
1106 let state = actor.borrow();
1107 browser_actor_is_current(&state)
1108 && state.mailbox.current_online_assignment_matches(
1109 device_id,
1110 remote_node_id,
1111 ticket,
1112 evidence,
1113 )
1114 })
1115 })
1116 }
1117 }
1118}
1119
1120#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1121enum BrowserDesiredPeerDecision {
1122 ObserveExisting,
1123 WaitUntil(i64),
1124 Dial,
1125}
1126
1127fn browser_desired_peer_decision(
1128 local_node_id: &str,
1129 remote_node_id: &str,
1130 transport_alive: bool,
1131 now_ms: i64,
1132 non_initiator_not_before_ms: Option<i64>,
1133) -> BrowserDesiredPeerDecision {
1134 if transport_alive {
1135 return BrowserDesiredPeerDecision::ObserveExisting;
1136 }
1137 if local_node_id > remote_node_id {
1138 return BrowserDesiredPeerDecision::Dial;
1139 }
1140
1141 let not_before = non_initiator_not_before_ms.unwrap_or_else(|| now_ms.saturating_add(3_000));
1142 if now_ms >= not_before {
1143 BrowserDesiredPeerDecision::Dial
1144 } else {
1145 BrowserDesiredPeerDecision::WaitUntil(not_before)
1146 }
1147}
1148
1149fn should_replace_browser_retry_deadline(current: Option<i64>, next: Option<i64>) -> bool {
1150 match (current, next) {
1151 (None, None) => false,
1152 (Some(_), None) | (None, Some(_)) => true,
1153 (Some(current), Some(next)) => next < current,
1154 }
1155}
1156
1157fn browser_reconcile_deadlines(
1158 outcome: BrowserPeerReconcileOutcome,
1159 failure_retry_deadline: Option<i64>,
1160) -> (Option<i64>, Option<i64>) {
1161 match outcome {
1162 BrowserPeerReconcileOutcome::Retry => (failure_retry_deadline, None),
1163 BrowserPeerReconcileOutcome::WakeAt(deadline) => (Some(deadline), None),
1164 BrowserPeerReconcileOutcome::WaitUntil(deadline) => (None, Some(deadline)),
1165 BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
1166 (None, None)
1167 }
1168 }
1169}
1170
1171#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1172enum BrowserPeerReconcileOutcome {
1173 Stable,
1174 Ineligible,
1175 Retry,
1176 WakeAt(i64),
1177 WaitUntil(i64),
1178}
1179
1180#[cfg(target_arch = "wasm32")]
1181fn browser_actor_is_current(actor: &BrowserAutoConnectActorState) -> bool {
1182 actor.active
1183 && actor.client.upgrade().is_some_and(|client| {
1184 client
1185 .auto_connect_generation
1186 .load(std::sync::atomic::Ordering::SeqCst)
1187 == actor.generation
1188 })
1189}
1190
1191#[cfg(target_arch = "wasm32")]
1192fn browser_attempt_is_current(
1193 actor: &BrowserAutoConnectActorState,
1194 _revision: u64,
1195 peer: &BrowserDesiredPeer,
1196) -> bool {
1197 browser_actor_is_current(actor) && desired_peer_evidence_is_current(&actor.mailbox, peer)
1198}
1199
1200#[cfg(target_arch = "wasm32")]
1201fn deactivate_browser_actor(actor: &mut BrowserAutoConnectActorState) {
1202 actor.active = false;
1203 actor.mailbox.peers.clear();
1204 actor.peer_evidence.clear();
1205 actor.failure_count.clear();
1206 actor.retry_not_before_ms.clear();
1207 actor.non_initiator_not_before_ms.clear();
1208 actor.scheduled_deadline_ms = None;
1209 actor.schedule_epoch = actor.schedule_epoch.saturating_add(1);
1210 actor.wake_pending = false;
1211}
1212
1213#[cfg(target_arch = "wasm32")]
1214fn schedule_browser_auto_connect_retry(
1215 actor: &std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
1216) {
1217 let (next_deadline, generation) = {
1218 let state = actor.borrow();
1219 if !browser_actor_is_current(&state) {
1220 return;
1221 }
1222 let next = state
1223 .retry_not_before_ms
1224 .values()
1225 .chain(state.non_initiator_not_before_ms.values())
1226 .copied()
1227 .min();
1228 (next, state.generation)
1229 };
1230
1231 let schedule = {
1232 let mut state = actor.borrow_mut();
1233 if !should_replace_browser_retry_deadline(state.scheduled_deadline_ms, next_deadline) {
1234 return;
1235 }
1236 state.schedule_epoch = state.schedule_epoch.saturating_add(1);
1237 state.scheduled_deadline_ms = next_deadline;
1238 next_deadline.map(|deadline| (deadline, state.schedule_epoch))
1239 };
1240
1241 let Some((deadline, epoch)) = schedule else {
1242 return;
1243 };
1244 let actor = actor.clone();
1245 wasm_bindgen_futures::spawn_local(async move {
1246 let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
1247 gloo_timers::future::sleep(std::time::Duration::from_millis(delay_ms)).await;
1248 let should_wake = {
1249 let mut state = actor.borrow_mut();
1250 if !state.active
1251 || state.generation != generation
1252 || state.schedule_epoch != epoch
1253 || !browser_actor_is_current(&state)
1254 {
1255 false
1256 } else {
1257 state.scheduled_deadline_ms = None;
1258 true
1259 }
1260 };
1261 if should_wake {
1262 wake_browser_auto_connect_actor(actor);
1263 }
1264 });
1265}
1266
1267#[cfg(target_arch = "wasm32")]
1268fn wake_browser_auto_connect_actor(
1269 actor: std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
1270) {
1271 let should_spawn = {
1272 let mut state = actor.borrow_mut();
1273 if !browser_actor_is_current(&state) {
1274 return;
1275 }
1276 if state.reconcile_running {
1277 state.wake_pending = true;
1278 false
1279 } else {
1280 state.reconcile_running = true;
1281 true
1282 }
1283 };
1284 if !should_spawn {
1285 return;
1286 }
1287
1288 wasm_bindgen_futures::spawn_local(async move {
1289 loop {
1290 run_browser_auto_connect_pass(actor.clone()).await;
1291 let run_again = {
1292 let mut state = actor.borrow_mut();
1293 if !browser_actor_is_current(&state) {
1294 state.reconcile_running = false;
1295 false
1296 } else if state.wake_pending {
1297 state.wake_pending = false;
1298 true
1299 } else {
1300 state.reconcile_running = false;
1301 false
1302 }
1303 };
1304 if !run_again {
1305 schedule_browser_auto_connect_retry(&actor);
1306 break;
1307 }
1308 }
1309 });
1310}
1311
1312#[cfg(target_arch = "wasm32")]
1313async fn reconcile_browser_desired_peer(
1314 actor: &std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
1315 client: &Arc<Client>,
1316 generation: u64,
1317 revision: u64,
1318 local_device_id: &str,
1319 peer: &BrowserDesiredPeer,
1320 non_initiator_not_before_ms: Option<i64>,
1321) -> BrowserPeerReconcileOutcome {
1322 if client
1323 .auto_connect_generation
1324 .load(std::sync::atomic::Ordering::SeqCst)
1325 != generation
1326 {
1327 return BrowserPeerReconcileOutcome::Ineligible;
1328 }
1329 if client.is_app_execution_suspended() {
1330 return BrowserPeerReconcileOutcome::Ineligible;
1331 }
1332
1333 let parsed_ticket = peer.ticket.as_deref().and_then(|ticket| {
1334 let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
1335 parse_endpoint_ticket(iroh_ticket).ok()
1336 });
1337 let parsed_node_id = parsed_ticket.as_ref().map(|address| address.id.to_string());
1338 let binding_node_id = peer.node_id.as_deref().or(parsed_node_id.as_deref());
1339 if let Some(node_id) = binding_node_id {
1340 client.bind_node_device_id(node_id, &peer.device_id).await;
1341 }
1342
1343 if !actor
1344 .try_borrow()
1345 .is_ok_and(|state| browser_attempt_is_current(&state, revision, peer))
1346 {
1347 return BrowserPeerReconcileOutcome::Ineligible;
1348 }
1349 if !peer.online || peer.device_id == local_device_id {
1350 return BrowserPeerReconcileOutcome::Ineligible;
1351 }
1352 if remote_excludes_local_device(&peer.excluded_peers, local_device_id) {
1353 return BrowserPeerReconcileOutcome::Ineligible;
1359 }
1360 if client.is_auto_connect_excluded(&peer.device_id) {
1361 return BrowserPeerReconcileOutcome::Ineligible;
1362 }
1363
1364 let Some(ticket) = peer.ticket.as_deref() else {
1365 return BrowserPeerReconcileOutcome::Ineligible;
1366 };
1367 let Some(endpoint_addr) = parsed_ticket else {
1368 return BrowserPeerReconcileOutcome::Ineligible;
1369 };
1370 let endpoint_id = endpoint_addr.id;
1371 let remote_node_id = endpoint_id.to_string();
1372 if peer
1373 .node_id
1374 .as_deref()
1375 .is_some_and(|node_id| node_id != remote_node_id)
1376 {
1377 return BrowserPeerReconcileOutcome::Ineligible;
1378 }
1379
1380 let Some(local_node_id) = client.current_node_id().await else {
1381 return BrowserPeerReconcileOutcome::WakeAt(now_millis_i64().saturating_add(250));
1382 };
1383 if local_node_id == remote_node_id {
1384 return BrowserPeerReconcileOutcome::Ineligible;
1385 }
1386
1387 client
1388 .reconcile_authoritative_device_node(&peer.device_id, &remote_node_id)
1389 .await;
1390 if !actor
1391 .try_borrow()
1392 .is_ok_and(|state| browser_attempt_is_current(&state, revision, peer))
1393 {
1394 return BrowserPeerReconcileOutcome::Ineligible;
1395 }
1396
1397 let transport_alive = client.is_connection_transport_alive(endpoint_id).await;
1398 match browser_desired_peer_decision(
1399 &local_node_id,
1400 &remote_node_id,
1401 transport_alive,
1402 now_millis_i64(),
1403 non_initiator_not_before_ms,
1404 ) {
1405 BrowserDesiredPeerDecision::WaitUntil(deadline) => {
1406 return BrowserPeerReconcileOutcome::WaitUntil(deadline);
1407 }
1408 BrowserDesiredPeerDecision::ObserveExisting | BrowserDesiredPeerDecision::Dial => {}
1409 }
1410
1411 let desired_peer_evidence = browser_peer_evidence(peer);
1412 let result = client
1413 .connect_desired_device(&peer.device_id, ticket, &desired_peer_evidence)
1414 .await;
1415 let (attempt_is_current, peer_was_withdrawn) = actor
1416 .try_borrow()
1417 .map(|state| {
1418 let actor_is_current = browser_actor_is_current(&state);
1419 (
1420 actor_is_current && browser_attempt_is_current(&state, revision, peer),
1421 superseded_desired_peer_was_withdrawn(
1422 actor_is_current,
1423 state.mailbox.revision,
1424 revision,
1425 state.mailbox.peers.contains_key(&peer.device_id),
1426 ),
1427 )
1428 })
1429 .unwrap_or((false, false));
1430 if !attempt_is_current {
1431 if peer_was_withdrawn {
1432 client
1433 .retire_withdrawn_desired_peer_connections(&peer.device_id, peer.node_id.as_deref())
1434 .await;
1435 }
1436 return BrowserPeerReconcileOutcome::Ineligible;
1437 }
1438 if client.is_auto_connect_excluded(&peer.device_id) {
1439 return BrowserPeerReconcileOutcome::Ineligible;
1443 }
1444
1445 let Ok(result) = result else {
1446 return BrowserPeerReconcileOutcome::Retry;
1447 };
1448 if matches!(result.state.as_str(), "failed" | "closed")
1449 || !client.is_connection_transport_alive(endpoint_id).await
1450 {
1451 return BrowserPeerReconcileOutcome::Retry;
1452 }
1453
1454 let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
1455 if let Some(token) = token_suffix.and_then(|suffix| {
1456 crate::session_token::decode_payload(iroh_ticket, suffix).map(|payload| payload.token)
1457 }) {
1458 let has_proof = client
1459 .has_current_remote_session_admission_proof(&result.connection_id, endpoint_id, &token)
1460 .await;
1461 if !has_proof && result.approved_scope.as_deref().is_none_or(str::is_empty) {
1465 return BrowserPeerReconcileOutcome::Retry;
1466 }
1467 }
1468
1469 BrowserPeerReconcileOutcome::Stable
1470}
1471
1472#[cfg(target_arch = "wasm32")]
1473async fn run_browser_auto_connect_pass(
1474 actor: std::rc::Rc<std::cell::RefCell<BrowserAutoConnectActorState>>,
1475) {
1476 let (client, generation, revision, local_device_id, peers) = {
1477 let state = actor.borrow();
1478 if !browser_actor_is_current(&state) {
1479 return;
1480 }
1481 let Some(client) = state.client.upgrade() else {
1482 return;
1483 };
1484 (
1485 client,
1486 state.generation,
1487 state.mailbox.revision,
1488 state.local_device_id.clone(),
1489 state.mailbox.peers.values().cloned().collect::<Vec<_>>(),
1490 )
1491 };
1492
1493 for peer in peers {
1494 let (retry_not_before, non_initiator_not_before) = {
1495 let state = actor.borrow();
1496 if !browser_attempt_is_current(&state, revision, &peer) {
1497 continue;
1498 }
1499 (
1500 state.retry_not_before_ms.get(&peer.device_id).copied(),
1501 state
1502 .non_initiator_not_before_ms
1503 .get(&peer.device_id)
1504 .copied(),
1505 )
1506 };
1507 let now = now_millis_i64();
1508 if retry_not_before.is_some_and(|deadline| now < deadline) {
1509 continue;
1510 }
1511
1512 let outcome = reconcile_browser_desired_peer(
1513 &actor,
1514 &client,
1515 generation,
1516 revision,
1517 &local_device_id,
1518 &peer,
1519 non_initiator_not_before,
1520 )
1521 .await;
1522 let backups_to_retire = {
1523 let mut state = actor.borrow_mut();
1524 if !browser_attempt_is_current(&state, revision, &peer) {
1525 continue;
1526 }
1527 let backups = if outcome == BrowserPeerReconcileOutcome::Stable {
1528 state
1529 .mailbox
1530 .acknowledge_topology_peer_ready(&peer.device_id, peer.topology_revision)
1531 } else {
1532 state
1533 .mailbox
1534 .observe_topology_peer_not_ready(&peer.device_id, peer.topology_revision);
1535 Vec::new()
1536 };
1537 match outcome {
1538 BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
1539 state.failure_count.remove(&peer.device_id);
1540 state.retry_not_before_ms.remove(&peer.device_id);
1541 state.non_initiator_not_before_ms.remove(&peer.device_id);
1542 }
1543 BrowserPeerReconcileOutcome::Retry => {
1544 let count = state
1545 .failure_count
1546 .entry(peer.device_id.clone())
1547 .or_insert(0);
1548 *count = count.saturating_add(1).min(10);
1549 let deadline = now_millis_i64()
1550 .saturating_add(browser_auto_connect_failure_backoff_ms(*count));
1551 let (retry_deadline, non_initiator_deadline) =
1552 browser_reconcile_deadlines(outcome, Some(deadline));
1553 match retry_deadline {
1554 Some(deadline) => {
1555 state
1556 .retry_not_before_ms
1557 .insert(peer.device_id.clone(), deadline);
1558 }
1559 None => {
1560 state.retry_not_before_ms.remove(&peer.device_id);
1561 }
1562 }
1563 match non_initiator_deadline {
1564 Some(deadline) => {
1565 state
1566 .non_initiator_not_before_ms
1567 .insert(peer.device_id.clone(), deadline);
1568 }
1569 None => {
1570 state.non_initiator_not_before_ms.remove(&peer.device_id);
1571 }
1572 }
1573 }
1574 BrowserPeerReconcileOutcome::WakeAt(deadline) => {
1575 state
1576 .retry_not_before_ms
1577 .insert(peer.device_id.clone(), deadline);
1578 state.non_initiator_not_before_ms.remove(&peer.device_id);
1579 }
1580 BrowserPeerReconcileOutcome::WaitUntil(deadline) => {
1581 state.retry_not_before_ms.remove(&peer.device_id);
1582 state
1583 .non_initiator_not_before_ms
1584 .insert(peer.device_id.clone(), deadline);
1585 }
1586 }
1587 backups
1588 };
1589 for backup in backups_to_retire {
1590 client
1591 .retire_withdrawn_desired_peer_connections(
1592 &backup.device_id,
1593 backup.node_id.as_deref(),
1594 )
1595 .await;
1596 }
1597 }
1598}
1599
1600#[cfg(not(target_arch = "wasm32"))]
1601async fn native_external_attempt_is_current(
1602 client: &Arc<Client>,
1603 generation: u64,
1604 _revision: u64,
1605 peer: &BrowserDesiredPeer,
1606) -> bool {
1607 if client
1608 .auto_connect_generation
1609 .load(std::sync::atomic::Ordering::SeqCst)
1610 != generation
1611 {
1612 return false;
1613 }
1614 let state = client.external_desired_peer_actor.lock().await;
1615 state.active
1616 && state.generation == generation
1617 && desired_peer_evidence_is_current(&state.mailbox, peer)
1618}
1619
1620#[cfg(not(target_arch = "wasm32"))]
1621async fn reconcile_native_external_desired_peer(
1622 client: &Arc<Client>,
1623 generation: u64,
1624 revision: u64,
1625 local_device_id: &str,
1626 peer: &BrowserDesiredPeer,
1627 non_initiator_not_before_ms: Option<i64>,
1628) -> BrowserPeerReconcileOutcome {
1629 if !native_external_attempt_is_current(client, generation, revision, peer).await
1630 || client.is_app_execution_suspended()
1631 {
1632 return BrowserPeerReconcileOutcome::Ineligible;
1633 }
1634
1635 let parsed_ticket = peer.ticket.as_deref().and_then(|ticket| {
1636 let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
1637 parse_endpoint_ticket(iroh_ticket).ok()
1638 });
1639 let parsed_node_id = parsed_ticket.as_ref().map(|address| address.id.to_string());
1640 let binding_node_id = peer.node_id.as_deref().or(parsed_node_id.as_deref());
1641 if let Some(node_id) = binding_node_id {
1642 client.bind_node_device_id(node_id, &peer.device_id).await;
1643 }
1644
1645 if !peer.online || peer.device_id == local_device_id {
1646 return BrowserPeerReconcileOutcome::Ineligible;
1647 }
1648 if remote_excludes_local_device(&peer.excluded_peers, local_device_id) {
1649 return BrowserPeerReconcileOutcome::Ineligible;
1653 }
1654 if client.is_auto_connect_excluded(&peer.device_id) {
1655 return BrowserPeerReconcileOutcome::Ineligible;
1656 }
1657
1658 let Some(ticket) = peer.ticket.as_deref() else {
1659 return BrowserPeerReconcileOutcome::Ineligible;
1660 };
1661 let Some(endpoint_addr) = parsed_ticket else {
1662 return BrowserPeerReconcileOutcome::Ineligible;
1663 };
1664 let endpoint_id = endpoint_addr.id;
1665 let remote_node_id = endpoint_id.to_string();
1666 if peer
1667 .node_id
1668 .as_deref()
1669 .is_some_and(|node_id| node_id != remote_node_id)
1670 {
1671 return BrowserPeerReconcileOutcome::Ineligible;
1672 }
1673
1674 let Some(local_node_id) = client.current_node_id().await else {
1675 return BrowserPeerReconcileOutcome::WakeAt(now_millis_i64().saturating_add(250));
1676 };
1677 if local_node_id == remote_node_id {
1678 return BrowserPeerReconcileOutcome::Ineligible;
1679 }
1680
1681 client
1682 .reconcile_authoritative_device_node(&peer.device_id, &remote_node_id)
1683 .await;
1684 if !native_external_attempt_is_current(client, generation, revision, peer).await {
1685 return BrowserPeerReconcileOutcome::Ineligible;
1686 }
1687
1688 let transport_alive = client.is_connection_transport_alive(endpoint_id).await;
1689 match browser_desired_peer_decision(
1690 &local_node_id,
1691 &remote_node_id,
1692 transport_alive,
1693 now_millis_i64(),
1694 non_initiator_not_before_ms,
1695 ) {
1696 BrowserDesiredPeerDecision::WaitUntil(deadline) => {
1697 return BrowserPeerReconcileOutcome::WaitUntil(deadline);
1698 }
1699 BrowserDesiredPeerDecision::ObserveExisting | BrowserDesiredPeerDecision::Dial => {}
1700 }
1701
1702 let desired_peer_evidence = browser_peer_evidence(peer);
1703 let result = client
1704 .connect_desired_device(&peer.device_id, ticket, &desired_peer_evidence)
1705 .await;
1706 let transport_stable_id = client
1707 .get_connection(endpoint_id)
1708 .await
1709 .map(|connection| crate::transport_generation::for_connection(&connection));
1710 if let Some(transport_stable_id) = transport_stable_id {
1711 let connection_id = Client::deterministic_connection_id(&local_node_id, &remote_node_id);
1712 let _ = client
1713 .maybe_start_offline_proof_for_transport(
1714 &connection_id,
1715 endpoint_id,
1716 transport_stable_id,
1717 )
1718 .await;
1719 }
1720 if !native_external_attempt_is_current(client, generation, revision, peer).await {
1721 let peer_was_withdrawn = {
1722 let state = client.external_desired_peer_actor.lock().await;
1723 let actor_is_current = state.active && state.generation == generation;
1724 superseded_desired_peer_was_withdrawn(
1725 actor_is_current,
1726 state.mailbox.revision,
1727 revision,
1728 state.mailbox.peers.contains_key(&peer.device_id),
1729 )
1730 };
1731 if peer_was_withdrawn {
1732 client
1733 .retire_withdrawn_desired_peer_connections(&peer.device_id, peer.node_id.as_deref())
1734 .await;
1735 }
1736 return BrowserPeerReconcileOutcome::Ineligible;
1737 }
1738 if client.is_auto_connect_excluded(&peer.device_id) {
1739 return BrowserPeerReconcileOutcome::Ineligible;
1743 }
1744
1745 let Ok(result) = result else {
1746 return BrowserPeerReconcileOutcome::Retry;
1747 };
1748 if matches!(result.state.as_str(), "failed" | "closed")
1749 || !client.is_connection_transport_alive(endpoint_id).await
1750 {
1751 return BrowserPeerReconcileOutcome::Retry;
1752 }
1753
1754 let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket);
1755 if let Some(token) = token_suffix.and_then(|suffix| {
1756 crate::session_token::decode_payload(iroh_ticket, suffix).map(|payload| payload.token)
1757 }) {
1758 let has_proof = client
1759 .has_current_remote_session_admission_proof(&result.connection_id, endpoint_id, &token)
1760 .await;
1761 if !has_proof && result.approved_scope.as_deref().is_none_or(str::is_empty) {
1762 return BrowserPeerReconcileOutcome::Retry;
1763 }
1764 }
1765
1766 BrowserPeerReconcileOutcome::Stable
1767}
1768
1769#[cfg(not(target_arch = "wasm32"))]
1770async fn run_native_external_auto_connect_pass(client: Arc<Client>) {
1771 let (generation, revision, local_device_id, peers, recovery_scoped) = {
1772 let mut state = client.external_desired_peer_actor.lock().await;
1773 if !state.active
1774 || client
1775 .auto_connect_generation
1776 .load(std::sync::atomic::Ordering::SeqCst)
1777 != state.generation
1778 {
1779 return;
1780 }
1781 let recovery_device_ids = std::mem::take(&mut state.recovery_device_ids);
1782 let recovery_scoped = !recovery_device_ids.is_empty();
1783 let peers = state
1784 .mailbox
1785 .peers
1786 .values()
1787 .filter(|peer| {
1788 recovery_device_ids.is_empty() || recovery_device_ids.contains(&peer.device_id)
1789 })
1790 .cloned()
1791 .collect::<Vec<_>>();
1792 (
1793 state.generation,
1794 state.mailbox.revision,
1795 state.local_device_id.clone(),
1796 peers,
1797 recovery_scoped,
1798 )
1799 };
1800
1801 for peer in peers {
1802 let (retry_not_before, non_initiator_not_before) = {
1803 let state = client.external_desired_peer_actor.lock().await;
1804 if !state.active
1805 || state.generation != generation
1806 || state.mailbox.revision != revision
1807 || !state
1808 .mailbox
1809 .peers
1810 .get(&peer.device_id)
1811 .is_some_and(|current| {
1812 browser_peer_evidence(current) == browser_peer_evidence(&peer)
1813 })
1814 {
1815 continue;
1816 }
1817 (
1818 state.retry_not_before_ms.get(&peer.device_id).copied(),
1819 state
1820 .non_initiator_not_before_ms
1821 .get(&peer.device_id)
1822 .copied(),
1823 )
1824 };
1825 let now = now_millis_i64();
1826 if retry_not_before.is_some_and(|deadline| now < deadline) {
1827 continue;
1828 }
1829
1830 let outcome = reconcile_native_external_desired_peer(
1831 &client,
1832 generation,
1833 revision,
1834 &local_device_id,
1835 &peer,
1836 non_initiator_not_before,
1837 )
1838 .await;
1839 if recovery_scoped {
1840 eprintln!(
1841 "[pluto-rtc][auto-connect][recovery-pass] device_id={} outcome={:?}",
1842 peer.device_id, outcome,
1843 );
1844 }
1845 let backups_to_retire = {
1846 let mut state = client.external_desired_peer_actor.lock().await;
1847 if !state.active
1848 || state.generation != generation
1849 || state.mailbox.revision != revision
1850 || !state
1851 .mailbox
1852 .peers
1853 .get(&peer.device_id)
1854 .is_some_and(|current| {
1855 browser_peer_evidence(current) == browser_peer_evidence(&peer)
1856 })
1857 {
1858 continue;
1859 }
1860 let backups = if outcome == BrowserPeerReconcileOutcome::Stable {
1861 state
1862 .mailbox
1863 .acknowledge_topology_peer_ready(&peer.device_id, peer.topology_revision)
1864 } else {
1865 state
1866 .mailbox
1867 .observe_topology_peer_not_ready(&peer.device_id, peer.topology_revision);
1868 Vec::new()
1869 };
1870 match outcome {
1871 BrowserPeerReconcileOutcome::Stable | BrowserPeerReconcileOutcome::Ineligible => {
1872 state.failure_count.remove(&peer.device_id);
1873 state.retry_not_before_ms.remove(&peer.device_id);
1874 state.non_initiator_not_before_ms.remove(&peer.device_id);
1875 }
1876 BrowserPeerReconcileOutcome::Retry => {
1877 let count = state
1878 .failure_count
1879 .entry(peer.device_id.clone())
1880 .or_insert(0);
1881 *count = count.saturating_add(1).min(10);
1882 let deadline = now_millis_i64()
1883 .saturating_add(browser_auto_connect_failure_backoff_ms(*count));
1884 let (retry_deadline, non_initiator_deadline) =
1885 browser_reconcile_deadlines(outcome, Some(deadline));
1886 match retry_deadline {
1887 Some(deadline) => {
1888 state
1889 .retry_not_before_ms
1890 .insert(peer.device_id.clone(), deadline);
1891 }
1892 None => {
1893 state.retry_not_before_ms.remove(&peer.device_id);
1894 }
1895 }
1896 match non_initiator_deadline {
1897 Some(deadline) => {
1898 state
1899 .non_initiator_not_before_ms
1900 .insert(peer.device_id.clone(), deadline);
1901 }
1902 None => {
1903 state.non_initiator_not_before_ms.remove(&peer.device_id);
1904 }
1905 }
1906 if recovery_scoped {
1912 state.recovery_device_ids.insert(peer.device_id.clone());
1913 }
1914 }
1915 BrowserPeerReconcileOutcome::WakeAt(deadline) => {
1916 state
1917 .retry_not_before_ms
1918 .insert(peer.device_id.clone(), deadline);
1919 state.non_initiator_not_before_ms.remove(&peer.device_id);
1920 if recovery_scoped {
1921 state.recovery_device_ids.insert(peer.device_id.clone());
1922 }
1923 }
1924 BrowserPeerReconcileOutcome::WaitUntil(deadline) => {
1925 state.retry_not_before_ms.remove(&peer.device_id);
1926 state
1927 .non_initiator_not_before_ms
1928 .insert(peer.device_id.clone(), deadline);
1929 if recovery_scoped {
1930 state.recovery_device_ids.insert(peer.device_id.clone());
1931 }
1932 }
1933 }
1934 backups
1935 };
1936 for backup in backups_to_retire {
1937 client
1938 .retire_withdrawn_desired_peer_connections(
1939 &backup.device_id,
1940 backup.node_id.as_deref(),
1941 )
1942 .await;
1943 }
1944 }
1945}
1946
1947#[cfg(not(target_arch = "wasm32"))]
1948async fn schedule_native_external_auto_connect_retry(client: Arc<Client>) {
1949 let schedule = {
1950 let mut state = client.external_desired_peer_actor.lock().await;
1951 if !state.active
1952 || client
1953 .auto_connect_generation
1954 .load(std::sync::atomic::Ordering::SeqCst)
1955 != state.generation
1956 {
1957 return;
1958 }
1959 let next = state
1960 .retry_not_before_ms
1961 .values()
1962 .chain(state.non_initiator_not_before_ms.values())
1963 .copied()
1964 .min();
1965 if !should_replace_browser_retry_deadline(state.scheduled_deadline_ms, next) {
1966 return;
1967 }
1968 state.schedule_epoch = state.schedule_epoch.saturating_add(1);
1969 state.scheduled_deadline_ms = next;
1970 next.map(|deadline| (deadline, state.schedule_epoch, state.generation))
1971 };
1972 let Some((deadline, epoch, generation)) = schedule else {
1973 return;
1974 };
1975 tokio::spawn(async move {
1976 let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
1977 tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
1978 let should_wake = {
1979 let mut state = client.external_desired_peer_actor.lock().await;
1980 if !state.active
1981 || state.generation != generation
1982 || state.schedule_epoch != epoch
1983 || client
1984 .auto_connect_generation
1985 .load(std::sync::atomic::Ordering::SeqCst)
1986 != generation
1987 {
1988 false
1989 } else {
1990 state.scheduled_deadline_ms = None;
1991 true
1992 }
1993 };
1994 if should_wake {
1995 let _ = wake_native_external_auto_connect_boxed(client).await;
1996 }
1997 });
1998}
1999
2000#[cfg(not(target_arch = "wasm32"))]
2001fn wake_native_external_auto_connect_boxed(
2002 client: Arc<Client>,
2003) -> std::pin::Pin<Box<dyn std::future::Future<Output = bool> + Send>> {
2004 Box::pin(async move {
2005 let should_spawn = {
2006 let mut state = client.external_desired_peer_actor.lock().await;
2007 if !state.active
2008 || client
2009 .auto_connect_generation
2010 .load(std::sync::atomic::Ordering::SeqCst)
2011 != state.generation
2012 {
2013 return false;
2014 }
2015 if state.reconcile_running {
2016 state.wake_pending = true;
2017 false
2018 } else {
2019 state.reconcile_running = true;
2020 true
2021 }
2022 };
2023 if !should_spawn {
2024 return true;
2025 }
2026 tokio::spawn(async move {
2027 loop {
2028 run_native_external_auto_connect_pass(client.clone()).await;
2029 let run_again = {
2030 let mut state = client.external_desired_peer_actor.lock().await;
2031 if !state.active
2032 || client
2033 .auto_connect_generation
2034 .load(std::sync::atomic::Ordering::SeqCst)
2035 != state.generation
2036 {
2037 state.reconcile_running = false;
2038 false
2039 } else if state.wake_pending {
2040 state.wake_pending = false;
2041 true
2042 } else {
2043 state.reconcile_running = false;
2044 false
2045 }
2046 };
2047 if !run_again {
2048 schedule_native_external_auto_connect_retry(client.clone()).await;
2049 break;
2050 }
2051 }
2052 });
2053 true
2054 })
2055}
2056
2057#[cfg(not(target_arch = "wasm32"))]
2058impl Client {
2059 #[doc(hidden)]
2065 pub async fn external_desired_peer_debug_snapshot(&self) -> serde_json::Value {
2066 let state = self.external_desired_peer_actor.lock().await;
2067 let mut ready_active_peers = state
2068 .mailbox
2069 .pending_ready_active_peers
2070 .iter()
2071 .cloned()
2072 .collect::<Vec<_>>();
2073 ready_active_peers.sort();
2074 let peers = state
2075 .mailbox
2076 .peers
2077 .values()
2078 .map(|peer| {
2079 let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
2080 let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
2081 parse_endpoint_ticket(iroh_ticket)
2082 .ok()
2083 .map(|address| address.id.to_string())
2084 });
2085 serde_json::json!({
2086 "deviceId": peer.device_id,
2087 "nodeId": peer.node_id,
2088 "ticketNodeId": ticket_node_id,
2089 "ticketFingerprint": peer.ticket.as_deref()
2090 .map(crate::session_token::token_fingerprint),
2091 "online": peer.online,
2092 "sessionId": peer.session_id,
2093 })
2094 })
2095 .collect::<Vec<_>>();
2096 serde_json::json!({
2097 "active": state.active,
2098 "generation": state.generation,
2099 "revision": state.mailbox.revision,
2100 "reconcileRunning": state.reconcile_running,
2101 "wakePending": state.wake_pending,
2102 "scheduledDeadlineMs": state.scheduled_deadline_ms,
2103 "lastFailureCounts": state.failure_count,
2104 "retryNotBeforeMs": state.retry_not_before_ms,
2105 "topologyRevision": state.mailbox.pending_topology_revision,
2106 "readyActivePeers": ready_active_peers,
2107 "backupPeerCount": state.mailbox.peers.values()
2108 .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
2109 .count(),
2110 "peers": peers,
2111 })
2112 }
2113
2114 pub async fn start_external_auto_connect(
2115 self: &Arc<Self>,
2116 user_id: String,
2117 local_device_id: String,
2118 ) -> anyhow::Result<()> {
2119 let user_id = user_id.trim().to_string();
2120 let local_device_id = local_device_id.trim().to_string();
2121 if user_id.is_empty() || local_device_id.is_empty() {
2122 return Err(anyhow::anyhow!(
2123 "external auto-connect requires non-empty user and local device ids"
2124 ));
2125 }
2126 let key = (user_id.clone(), local_device_id.clone());
2127 let duplicate_start = {
2128 let state = self.external_desired_peer_actor.lock().await;
2129 let guard = self
2130 .auto_connect_loop_key
2131 .lock()
2132 .unwrap_or_else(|poisoned| poisoned.into_inner());
2133 guard.as_ref() == Some(&key)
2134 && state.active
2135 && state.local_device_id == local_device_id
2136 && self
2137 .auto_connect_generation
2138 .load(std::sync::atomic::Ordering::SeqCst)
2139 == state.generation
2140 };
2141 if duplicate_start {
2142 self.wake_native_external_auto_connect().await;
2143 return Ok(());
2144 }
2145 let generation = {
2146 let mut guard = self
2147 .auto_connect_loop_key
2148 .lock()
2149 .unwrap_or_else(|poisoned| poisoned.into_inner());
2150 self.known_device_ids_by_node
2151 .write()
2152 .unwrap_or_else(|poisoned| poisoned.into_inner())
2153 .clear();
2154 self.native_route_repair_credentials
2155 .write()
2156 .unwrap_or_else(|poisoned| poisoned.into_inner())
2157 .clear();
2158 self.scoped_route_repair_credentials
2159 .write()
2160 .unwrap_or_else(|poisoned| poisoned.into_inner())
2161 .clear();
2162 *guard = Some(key);
2163 self.auto_connect_generation
2164 .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
2165 + 1
2166 };
2167 let mut state = self.external_desired_peer_actor.lock().await;
2168 *state = NativeExternalAutoConnectActorState {
2169 active: true,
2170 generation,
2171 local_device_id,
2172 ..NativeExternalAutoConnectActorState::default()
2173 };
2174 Ok(())
2175 }
2176
2177 pub async fn submit_external_desired_peers(
2178 self: &Arc<Self>,
2179 revision: u64,
2180 peers_json: &str,
2181 ) -> anyhow::Result<bool> {
2182 self.submit_external_desired_peer_update(revision, peers_json, None, false)
2183 .await
2184 }
2185
2186 pub async fn withdraw_external_desired_peers(
2191 self: &Arc<Self>,
2192 revision: u64,
2193 peers_json: &str,
2194 ) -> anyhow::Result<bool> {
2195 self.submit_external_desired_peer_update(revision, peers_json, None, true)
2196 .await
2197 }
2198
2199 #[cfg(feature = "transport-lan")]
2200 pub(crate) async fn submit_external_desired_peer_observation_expired(
2201 self: &Arc<Self>,
2202 revision: u64,
2203 peers_json: &str,
2204 expired_device_id: &str,
2205 ) -> anyhow::Result<bool> {
2206 let expired_device_id = expired_device_id.trim();
2207 if expired_device_id.is_empty() {
2208 return Err(anyhow::anyhow!("expired offline device id is required"));
2209 }
2210 self.submit_external_desired_peer_update(revision, peers_json, Some(expired_device_id), false)
2211 .await
2212 }
2213
2214 async fn submit_external_desired_peer_update(
2215 self: &Arc<Self>,
2216 revision: u64,
2217 peers_json: &str,
2218 observation_expired_device_id: Option<&str>,
2219 withdrawing: bool,
2220 ) -> anyhow::Result<bool> {
2221 if peers_json.len() > 256 * 1024 {
2222 return Err(anyhow::anyhow!(
2223 "external desired-peer payload exceeds 256 KiB"
2224 ));
2225 }
2226 let peers: Vec<BrowserDesiredPeer> = serde_json::from_str(peers_json)
2227 .map_err(|error| anyhow::anyhow!("invalid external desired-peer payload: {error}"))?;
2228 if peers.len() > 100 {
2229 return Err(anyhow::anyhow!(
2230 "external desired-peer payload exceeds 100 peers"
2231 ));
2232 }
2233 if observation_expired_device_id
2234 .is_some_and(|expired| peers.iter().any(|peer| peer.device_id.trim() == expired))
2235 {
2236 return Err(anyhow::anyhow!(
2237 "expired offline device remains in desired-peer input"
2238 ));
2239 }
2240
2241 let (bindings, withdrawn) = {
2242 let mut state = self.external_desired_peer_actor.lock().await;
2243 if !state.active {
2244 if withdrawing {
2245 return Ok(false);
2246 }
2247 return Err(anyhow::anyhow!("external auto-connect is not started"));
2248 }
2249 let accepted = match observation_expired_device_id {
2250 Some(expired) => state
2251 .mailbox
2252 .accept_with_observation_expired(revision, peers, expired),
2253 None => state.mailbox.accept_with_withdrawn(revision, peers),
2254 };
2255 let Some(withdrawn) = accepted else {
2256 return Ok(false);
2257 };
2258 let desired_ids = state
2259 .mailbox
2260 .peers
2261 .keys()
2262 .cloned()
2263 .collect::<std::collections::HashSet<_>>();
2264 state.peer_evidence.retain(|id, _| desired_ids.contains(id));
2265 state.failure_count.retain(|id, _| desired_ids.contains(id));
2266 state
2267 .retry_not_before_ms
2268 .retain(|id, _| desired_ids.contains(id));
2269 state
2270 .non_initiator_not_before_ms
2271 .retain(|id, _| desired_ids.contains(id));
2272
2273 let evidence = state
2274 .mailbox
2275 .peers
2276 .iter()
2277 .map(|(id, peer)| (id.clone(), browser_peer_evidence(peer)))
2278 .collect::<Vec<_>>();
2279 for (id, evidence) in evidence {
2280 let changed = state
2281 .peer_evidence
2282 .insert(id.clone(), evidence.clone())
2283 .is_some_and(|previous| previous != evidence);
2284 if changed {
2285 state.failure_count.remove(&id);
2286 state.retry_not_before_ms.remove(&id);
2287 state.non_initiator_not_before_ms.remove(&id);
2288 }
2289 }
2290 state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2291 state.scheduled_deadline_ms = None;
2292 let bindings = state
2293 .mailbox
2294 .peers
2295 .values()
2296 .filter(|peer| peer.online)
2297 .filter_map(|peer| {
2298 let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
2299 let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
2300 parse_endpoint_ticket(iroh_ticket)
2301 .ok()
2302 .map(|address| address.id.to_string())
2303 });
2304 let node_id = match (peer.node_id.as_deref(), ticket_node_id.as_deref()) {
2305 (Some(directory), Some(ticket)) if directory != ticket => return None,
2306 (Some(directory), _) => directory.to_string(),
2307 (None, Some(ticket)) => ticket.to_string(),
2308 (None, None) => return None,
2309 };
2310 Some((peer.device_id.clone(), node_id))
2311 })
2312 .collect::<Vec<_>>();
2313 (bindings, withdrawn)
2314 };
2315 for peer in withdrawn {
2316 let still_withdrawn = {
2317 let state = self.external_desired_peer_actor.lock().await;
2318 state.active
2319 && state.mailbox.revision == revision
2320 && state
2321 .mailbox
2322 .peers
2323 .get(&peer.device_id)
2324 .is_none_or(|current| !current.online)
2325 };
2326 if still_withdrawn {
2327 self.retire_withdrawn_desired_peer_connections(
2328 &peer.device_id,
2329 peer.node_id.as_deref(),
2330 )
2331 .await;
2332 }
2333 }
2334 for (device_id, node_id) in bindings {
2335 self.observe_authoritative_device_node(&device_id, &node_id);
2336 }
2337 self.wake_native_external_auto_connect().await;
2338 Ok(true)
2339 }
2340
2341 pub async fn wake_native_external_auto_connect(&self) -> bool {
2342 {
2345 let mut state = self.external_desired_peer_actor.lock().await;
2346 state.recovery_device_ids.clear();
2347 state.recovery_suppressed_until_ms.clear();
2348 }
2349 wake_native_external_auto_connect_boxed(Arc::new(self.clone())).await
2350 }
2351
2352 pub(crate) async fn wake_native_external_auto_connect_for_replaced_route(
2363 &self,
2364 remote_node_id: &str,
2365 ) -> bool {
2366 let remote_node_id = remote_node_id.trim();
2367 let Some(remote_device_id) = self
2368 .known_device_ids_by_node
2369 .read()
2370 .unwrap_or_else(|poisoned| poisoned.into_inner())
2371 .get(remote_node_id)
2372 .cloned()
2373 else {
2374 eprintln!(
2375 "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} action=skip reason=device-binding-missing",
2376 remote_node_id,
2377 );
2378 return false;
2379 };
2380
2381 {
2382 let mut state = self.external_desired_peer_actor.lock().await;
2383 if !state.active
2384 || self
2385 .auto_connect_generation
2386 .load(std::sync::atomic::Ordering::SeqCst)
2387 != state.generation
2388 || !state.mailbox.peers.contains_key(&remote_device_id)
2389 {
2390 eprintln!(
2391 "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} device_id={} action=skip reason=actor-or-peer-inactive",
2392 remote_node_id, remote_device_id,
2393 );
2394 return false;
2395 }
2396
2397 state.recovery_device_ids.insert(remote_device_id.clone());
2398 state.recovery_suppressed_until_ms.remove(&remote_device_id);
2399 state.retry_not_before_ms.remove(&remote_device_id);
2400 state.non_initiator_not_before_ms.remove(&remote_device_id);
2401 state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2402 state.scheduled_deadline_ms = None;
2403 }
2404
2405 eprintln!(
2406 "[pluto-rtc][auto-connect][replacement-route] remote_node_id={} device_id={} action=wake",
2407 remote_node_id, remote_device_id,
2408 );
2409 wake_native_external_auto_connect_boxed(Arc::new(self.clone())).await
2410 }
2411
2412 pub(crate) async fn schedule_native_external_auto_connect_recovery(
2419 &self,
2420 remote_node_id: &str,
2421 ) -> bool {
2422 const RECOVERY_COALESCE_MS: i64 = 1_000;
2428 const RECOVERY_DUPLICATE_SUPPRESSION_MS: i64 = MANAGED_SETTLE_DEADLINE_MS;
2434 let remote_node_id = remote_node_id.trim();
2435 let Some(remote_device_id) = self
2436 .known_device_ids_by_node
2437 .read()
2438 .unwrap_or_else(|poisoned| poisoned.into_inner())
2439 .get(remote_node_id)
2440 .cloned()
2441 else {
2442 return false;
2446 };
2447 let schedule = {
2448 let mut state = self.external_desired_peer_actor.lock().await;
2449 if !state.active
2450 || self
2451 .auto_connect_generation
2452 .load(std::sync::atomic::Ordering::SeqCst)
2453 != state.generation
2454 {
2455 return false;
2456 }
2457
2458 let now = now_millis_i64();
2459 if state
2460 .recovery_suppressed_until_ms
2461 .get(&remote_device_id)
2462 .is_some_and(|deadline| *deadline > now)
2463 {
2464 return true;
2465 }
2466 state.recovery_device_ids.insert(remote_device_id.clone());
2467 state.recovery_suppressed_until_ms.insert(
2468 remote_device_id,
2469 now.saturating_add(RECOVERY_DUPLICATE_SUPPRESSION_MS),
2470 );
2471
2472 let deadline = now.saturating_add(RECOVERY_COALESCE_MS);
2473 if state
2474 .scheduled_deadline_ms
2475 .is_some_and(|scheduled| scheduled <= deadline)
2476 {
2477 return true;
2478 }
2479 state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2480 state.scheduled_deadline_ms = Some(deadline);
2481 (deadline, state.schedule_epoch, state.generation)
2482 };
2483
2484 let client = Arc::new(self.clone());
2485 tokio::spawn(async move {
2486 let (deadline, epoch, generation) = schedule;
2487 let delay_ms = deadline.saturating_sub(now_millis_i64()).max(0) as u64;
2488 tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
2489 let should_wake = {
2490 let mut state = client.external_desired_peer_actor.lock().await;
2491 if !state.active
2492 || state.generation != generation
2493 || state.schedule_epoch != epoch
2494 || client
2495 .auto_connect_generation
2496 .load(std::sync::atomic::Ordering::SeqCst)
2497 != generation
2498 {
2499 false
2500 } else {
2501 state.scheduled_deadline_ms = None;
2502 true
2503 }
2504 };
2505 if should_wake {
2506 let _ = wake_native_external_auto_connect_boxed(client).await;
2507 }
2508 });
2509 true
2510 }
2511
2512 pub(crate) async fn external_auto_connect_is_active(&self) -> bool {
2513 self.external_desired_peer_actor.lock().await.active
2514 }
2515
2516 pub async fn stop_external_auto_connect(&self) {
2517 let mut state = self.external_desired_peer_actor.lock().await;
2518 state.active = false;
2519 state.mailbox.peers.clear();
2520 state.peer_evidence.clear();
2521 state.failure_count.clear();
2522 state.retry_not_before_ms.clear();
2523 state.non_initiator_not_before_ms.clear();
2524 state.scheduled_deadline_ms = None;
2525 state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2526 state.wake_pending = false;
2527 state.recovery_device_ids.clear();
2528 state.recovery_suppressed_until_ms.clear();
2529 }
2530}
2531
2532#[cfg(target_arch = "wasm32")]
2533impl Client {
2534 pub(crate) fn start_browser_auto_connect(
2535 self: &Arc<Self>,
2536 user_id: String,
2537 local_device_id: String,
2538 ) -> anyhow::Result<()> {
2539 let user_id = user_id.trim().to_string();
2540 let local_device_id = local_device_id.trim().to_string();
2541 if user_id.is_empty() || local_device_id.is_empty() {
2542 return Err(anyhow::anyhow!(
2543 "browser auto-connect requires non-empty user and local device ids"
2544 ));
2545 }
2546
2547 let key = (user_id, local_device_id.clone());
2548 let actor_key = browser_actor_key(self);
2549 let duplicate_actor = BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
2550 actors.borrow().get(&actor_key).cloned().filter(|actor| {
2551 let state = actor.borrow();
2552 browser_actor_is_current(&state) && state.local_device_id == local_device_id
2553 })
2554 });
2555 {
2556 let guard = self
2557 .auto_connect_loop_key
2558 .lock()
2559 .unwrap_or_else(|poisoned| poisoned.into_inner());
2560 if guard.as_ref() == Some(&key) {
2561 if let Some(actor) = duplicate_actor {
2562 wake_browser_auto_connect_actor(actor);
2563 return Ok(());
2564 }
2565 }
2566 }
2567
2568 let generation = {
2569 let mut guard = self
2570 .auto_connect_loop_key
2571 .lock()
2572 .unwrap_or_else(|poisoned| poisoned.into_inner());
2573 self.known_device_ids_by_node
2574 .write()
2575 .unwrap_or_else(|poisoned| poisoned.into_inner())
2576 .clear();
2577 *guard = Some(key);
2578 self.auto_connect_generation
2579 .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
2580 + 1
2581 };
2582
2583 let actor = std::rc::Rc::new(std::cell::RefCell::new(BrowserAutoConnectActorState {
2584 active: true,
2585 client: Arc::downgrade(self),
2586 generation,
2587 local_device_id,
2588 mailbox: BrowserDesiredPeerMailbox::default(),
2589 peer_evidence: std::collections::HashMap::new(),
2590 failure_count: std::collections::HashMap::new(),
2591 retry_not_before_ms: std::collections::HashMap::new(),
2592 non_initiator_not_before_ms: std::collections::HashMap::new(),
2593 scheduled_deadline_ms: None,
2594 schedule_epoch: 0,
2595 reconcile_running: false,
2596 wake_pending: false,
2597 }));
2598 BROWSER_AUTO_CONNECT_ACTORS.with(|actors| {
2599 if let Some(previous) = actors.borrow_mut().insert(actor_key, actor) {
2600 deactivate_browser_actor(&mut previous.borrow_mut());
2601 }
2602 });
2603 Ok(())
2604 }
2605
2606 pub(crate) fn submit_browser_desired_peers(
2607 self: &Arc<Self>,
2608 revision: u64,
2609 peers_json: &str,
2610 ) -> anyhow::Result<bool> {
2611 let peers: Vec<BrowserDesiredPeer> = serde_json::from_str(peers_json)
2612 .map_err(|error| anyhow::anyhow!("invalid browser desired-peer payload: {error}"))?;
2613 let actor = BROWSER_AUTO_CONNECT_ACTORS
2614 .with(|actors| actors.borrow().get(&browser_actor_key(self)).cloned());
2615 let Some(actor) = actor else {
2616 return Err(anyhow::anyhow!("browser auto-connect is not started"));
2617 };
2618
2619 let (accepted, current_bindings, withdrawn) = {
2620 let mut state = actor.borrow_mut();
2621 if !browser_actor_is_current(&state) {
2622 (false, Vec::new(), Vec::new())
2623 } else if let Some(withdrawn) = state.mailbox.accept_with_withdrawn(revision, peers) {
2624 let desired_ids = state
2625 .mailbox
2626 .peers
2627 .keys()
2628 .cloned()
2629 .collect::<std::collections::HashSet<_>>();
2630 state
2631 .peer_evidence
2632 .retain(|device_id, _| desired_ids.contains(device_id));
2633 state
2634 .failure_count
2635 .retain(|device_id, _| desired_ids.contains(device_id));
2636 state
2637 .retry_not_before_ms
2638 .retain(|device_id, _| desired_ids.contains(device_id));
2639 state
2640 .non_initiator_not_before_ms
2641 .retain(|device_id, _| desired_ids.contains(device_id));
2642
2643 let evidence = state
2644 .mailbox
2645 .peers
2646 .iter()
2647 .map(|(device_id, peer)| (device_id.clone(), browser_peer_evidence(peer)))
2648 .collect::<Vec<_>>();
2649 for (device_id, evidence) in evidence {
2650 let changed = state
2651 .peer_evidence
2652 .insert(device_id.clone(), evidence.clone())
2653 .is_some_and(|previous| previous != evidence);
2654 if changed {
2655 state.failure_count.remove(&device_id);
2656 state.retry_not_before_ms.remove(&device_id);
2657 state.non_initiator_not_before_ms.remove(&device_id);
2658 }
2659 }
2660 state.schedule_epoch = state.schedule_epoch.saturating_add(1);
2661 state.scheduled_deadline_ms = None;
2662 let bindings = state
2663 .mailbox
2664 .peers
2665 .values()
2666 .filter(|peer| peer.online)
2667 .filter_map(|peer| {
2668 let ticket_node_id = peer.ticket.as_deref().and_then(|ticket| {
2669 let (iroh_ticket, _) = crate::session_token::split_ticket(ticket);
2670 parse_endpoint_ticket(iroh_ticket)
2671 .ok()
2672 .map(|address| address.id.to_string())
2673 });
2674 let node_id = match (peer.node_id.as_deref(), ticket_node_id.as_deref()) {
2675 (Some(directory_node_id), Some(ticket_node_id))
2676 if directory_node_id != ticket_node_id =>
2677 {
2678 return None;
2679 }
2680 (Some(directory_node_id), _) => directory_node_id.to_string(),
2681 (None, Some(ticket_node_id)) => ticket_node_id.to_string(),
2682 (None, None) => return None,
2683 };
2684 Some((peer.device_id.clone(), node_id))
2685 })
2686 .collect::<Vec<_>>();
2687 (true, bindings, withdrawn)
2688 } else {
2689 (false, Vec::new(), Vec::new())
2690 }
2691 };
2692 if accepted {
2693 for (device_id, node_id) in current_bindings {
2697 self.observe_authoritative_device_node(&device_id, &node_id);
2698 }
2699 if !withdrawn.is_empty() {
2700 let client = Arc::clone(self);
2701 let actor_for_retirement = actor.clone();
2702 wasm_bindgen_futures::spawn_local(async move {
2703 for peer in withdrawn {
2704 let still_withdrawn = {
2705 let state = actor_for_retirement.borrow();
2706 browser_actor_is_current(&state)
2707 && state.mailbox.revision == revision
2708 && state
2709 .mailbox
2710 .peers
2711 .get(&peer.device_id)
2712 .is_none_or(|current| !current.online)
2713 };
2714 if still_withdrawn {
2715 client
2716 .retire_withdrawn_desired_peer_connections(
2717 &peer.device_id,
2718 peer.node_id.as_deref(),
2719 )
2720 .await;
2721 }
2722 }
2723 });
2724 }
2725 wake_browser_auto_connect_actor(actor);
2726 }
2727 Ok(accepted)
2728 }
2729
2730 pub(crate) fn wake_browser_auto_connect(self: &Arc<Self>) -> bool {
2731 let actor = BROWSER_AUTO_CONNECT_ACTORS
2732 .with(|actors| actors.borrow().get(&browser_actor_key(self)).cloned());
2733 let Some(actor) = actor else {
2734 return false;
2735 };
2736 if !browser_actor_is_current(&actor.borrow()) {
2737 return false;
2738 }
2739 wake_browser_auto_connect_actor(actor);
2740 true
2741 }
2742
2743 pub(crate) fn stop_browser_auto_connect(self: &Arc<Self>) {
2744 let actor = BROWSER_AUTO_CONNECT_ACTORS
2745 .with(|actors| actors.borrow_mut().remove(&browser_actor_key(self)));
2746 if let Some(actor) = actor {
2747 deactivate_browser_actor(&mut actor.borrow_mut());
2748 }
2749 }
2750}
2751
2752#[cfg(test)]
2753mod tests {
2754 use super::*;
2755
2756 #[cfg(not(target_arch = "wasm32"))]
2757 #[tokio::test]
2758 async fn auto_connect_disabled_during_reconcile_preserves_manual_transport() {
2759 let client = Arc::new(Client::new_with_app_tag(
2760 crate::test_constants::TEST_PROJECT_ID.to_string(),
2761 "exclusion-race".to_string(),
2762 Box::new(|| None),
2763 ));
2764 let remote = Client::new_with_app_tag(
2765 crate::test_constants::TEST_PROJECT_ID.to_string(),
2766 "exclusion-race".to_string(),
2767 Box::new(|| None),
2768 );
2769 for owner in [&*client, &remote] {
2770 let endpoint = iroh::Endpoint::builder(iroh::endpoint::presets::N0)
2771 .alpns(vec![crate::native_node::PlutoniumProtocol::ALPN.to_vec()])
2772 .relay_mode(iroh::RelayMode::Disabled)
2773 .bind_addr("127.0.0.1:0".parse::<std::net::SocketAddr>().unwrap())
2774 .unwrap()
2775 .bind()
2776 .await
2777 .unwrap();
2778 owner.adopt_endpoint(endpoint).await;
2779 }
2780 let ticket = remote.endpoint_ticket().await.unwrap();
2781 let remote_node_id = remote.current_node_id().await.unwrap();
2782 let endpoint_id = remote_node_id.parse().unwrap();
2783 let connected = client
2784 .connect_device(Some("remote"), &ticket)
2785 .await
2786 .unwrap();
2787 let original = client.get_connection(endpoint_id).await.unwrap();
2788 let original_stable_id = crate::transport_generation::for_connection(&original);
2789 let peer = BrowserDesiredPeer {
2790 device_id: "remote".to_string(),
2791 node_id: Some(remote_node_id),
2792 ticket: Some(ticket),
2793 online: true,
2794 session_id: Some("runtime-1".to_string()),
2795 excluded_peers: Vec::new(),
2796 topology_role: None,
2797 topology_revision: None,
2798 };
2799 let generation = client
2800 .auto_connect_generation
2801 .load(std::sync::atomic::Ordering::SeqCst);
2802 {
2803 let mut actor = client.external_desired_peer_actor.lock().await;
2804 actor.active = true;
2805 actor.generation = generation;
2806 actor.local_device_id = "local".to_string();
2807 actor
2808 .mailbox
2809 .accept_with_withdrawn(1, vec![peer.clone()])
2810 .unwrap();
2811 }
2812 let gate = client.managed_connect_gate(&connected.connection_id).await;
2816 let guard = gate.lock().await;
2817 let reconcile =
2818 reconcile_native_external_desired_peer(&client, generation, 1, "local", &peer, None);
2819 tokio::pin!(reconcile);
2820 assert!(
2821 tokio::time::timeout(std::time::Duration::from_millis(100), &mut reconcile)
2822 .await
2823 .is_err()
2824 );
2825 client.set_auto_connect_excluded("remote", true);
2826 drop(guard);
2827 assert_eq!(
2828 tokio::time::timeout(std::time::Duration::from_secs(5), reconcile)
2829 .await
2830 .unwrap(),
2831 BrowserPeerReconcileOutcome::Ineligible,
2832 );
2833 assert!(client.is_auto_connect_excluded("remote"));
2834 assert!(
2835 original.close_reason().is_none(),
2836 "auto-connect policy must not close a manual transport"
2837 );
2838 let current = client
2839 .get_connection(endpoint_id)
2840 .await
2841 .expect("manual connection survives");
2842 assert_eq!(
2843 crate::transport_generation::for_connection(¤t),
2844 original_stable_id
2845 );
2846 assert!(client.is_connection_transport_alive(endpoint_id).await);
2847 }
2848
2849 fn replacement_snapshot(last_reconnect_reason: Option<&str>) -> ConnectionHealthView {
2850 ConnectionHealthView {
2851 connection_id: "conn-1".to_string(),
2852 device_id: Some("device-1".to_string()),
2853 device_id_hint: Some("device-1".to_string()),
2854 node_id: Some("node-1".to_string()),
2855 active_transport_stable_id: None,
2856 transport_generation: 2,
2857 route_generation: 0,
2858 status: ManagedHealth::AwaitingReplacement,
2859 settled_ready: false,
2860 readiness_state: ReadinessState::AwaitingReplacement,
2861 replacement_pending: true,
2862 last_lifecycle_transition_at_ms: 1_000,
2863 readiness_reason: "missing-active-transport".to_string(),
2864 transition_count: 1,
2865 connecting_transition_count: 0,
2866 replacement_count: 1,
2867 retire_count: 0,
2868 last_disconnect_reason: None,
2869 last_reconnect_reason: last_reconnect_reason.map(str::to_string),
2870 }
2871 }
2872
2873 #[test]
2874 fn runtime_instance_change_is_observed_only_after_the_initial_generation() {
2875 let mut last_known = std::collections::HashMap::new();
2876
2877 assert!(!observe_remote_runtime_instance(
2878 &mut last_known,
2879 "device-1",
2880 Some("runtime-1"),
2881 ));
2882 assert!(!observe_remote_runtime_instance(
2883 &mut last_known,
2884 "device-1",
2885 Some("runtime-1"),
2886 ));
2887 assert!(observe_remote_runtime_instance(
2888 &mut last_known,
2889 "device-1",
2890 Some("runtime-2"),
2891 ));
2892 }
2893
2894 #[cfg(not(target_arch = "wasm32"))]
2895 #[test]
2896 fn delayed_runtime_instance_update_preserves_a_responsive_replacement() {
2897 assert!(!runtime_instance_change_requires_replacement(true, true));
2898 }
2899
2900 #[cfg(not(target_arch = "wasm32"))]
2901 #[test]
2902 fn runtime_instance_update_replaces_an_unresponsive_transport() {
2903 assert!(runtime_instance_change_requires_replacement(true, false));
2904 assert!(!runtime_instance_change_requires_replacement(false, false));
2905 }
2906
2907 #[test]
2908 fn ticket_change_clears_transient_retry_without_retaining_ticket_material() {
2909 let mut last_known = std::collections::HashMap::new();
2910 let mut last_attempt_at = std::collections::HashMap::new();
2911 let mut failure_count = std::collections::HashMap::new();
2912 let mut failure_backoff_until = std::collections::HashMap::new();
2913
2914 assert!(!reconcile_remote_ticket_retry_state(
2915 &mut last_known,
2916 "device-1",
2917 "ticket-1.token-1",
2918 &mut last_attempt_at,
2919 &mut failure_count,
2920 &mut failure_backoff_until,
2921 ));
2922 last_attempt_at.insert("device-1".to_string(), 100);
2923 failure_count.insert("device-1".to_string(), 4);
2924 failure_backoff_until.insert("device-1".to_string(), 8_100);
2925
2926 assert!(!reconcile_remote_ticket_retry_state(
2927 &mut last_known,
2928 "device-1",
2929 "ticket-1.token-1",
2930 &mut last_attempt_at,
2931 &mut failure_count,
2932 &mut failure_backoff_until,
2933 ));
2934 assert_eq!(last_attempt_at.get("device-1"), Some(&100));
2935 assert_eq!(failure_count.get("device-1"), Some(&4));
2936 assert_eq!(failure_backoff_until.get("device-1"), Some(&8_100));
2937
2938 assert!(reconcile_remote_ticket_retry_state(
2939 &mut last_known,
2940 "device-1",
2941 "ticket-1.token-2",
2942 &mut last_attempt_at,
2943 &mut failure_count,
2944 &mut failure_backoff_until,
2945 ));
2946 assert!(!last_attempt_at.contains_key("device-1"));
2947 assert!(!failure_count.contains_key("device-1"));
2948 assert!(!failure_backoff_until.contains_key("device-1"));
2949
2950 let stored = last_known
2951 .get("device-1::ticket")
2952 .expect("ticket fingerprint");
2953 assert_eq!(stored.len(), 16);
2954 assert!(!stored.contains("token-2"));
2955 }
2956
2957 #[test]
2958 fn replacement_window_suppresses_auto_connect_for_active_replacement_churn() {
2959 let snapshot = replacement_snapshot(Some("incoming-replacement-churn"));
2960
2961 assert!(should_suppress_auto_connect_for_replacement_window(
2962 &snapshot,
2963 snapshot.last_lifecycle_transition_at_ms + 1_000,
2964 false,
2965 ));
2966 assert!(
2967 !should_suppress_auto_connect_for_replacement_window(
2968 &snapshot,
2969 snapshot.last_lifecycle_transition_at_ms + 1_000,
2970 true,
2971 ),
2972 "replacement suppression must not bypass positive liveness checks for an owned transport"
2973 );
2974 }
2975
2976 #[test]
2977 fn transient_disconnect_replacement_window_allows_auto_connect_redial() {
2978 let snapshot = replacement_snapshot(Some("transient-disconnect"));
2979
2980 assert!(!should_suppress_auto_connect_for_replacement_window(
2981 &snapshot,
2982 snapshot.last_lifecycle_transition_at_ms + 1_000,
2983 false,
2984 ));
2985 }
2986
2987 #[test]
2988 fn expired_replacement_window_allows_auto_connect_redial() {
2989 let snapshot = replacement_snapshot(Some("incoming-replacement-churn"));
2990
2991 assert!(!should_suppress_auto_connect_for_replacement_window(
2992 &snapshot,
2993 snapshot.last_lifecycle_transition_at_ms + MANAGED_SETTLE_DEADLINE_MS,
2994 false,
2995 ));
2996 }
2997
2998 #[test]
2999 fn remote_exclusion_matches_canonical_device_id_case_insensitively() {
3000 let excluded_peers = vec![" DEVICE-LOCAL ".to_string()];
3001
3002 assert!(remote_excludes_local_device(
3003 &excluded_peers,
3004 "device-local"
3005 ));
3006 assert!(!remote_excludes_local_device(
3007 &excluded_peers,
3008 "device-other"
3009 ));
3010 assert!(!remote_excludes_local_device(&excluded_peers, ""));
3011 }
3012
3013 #[test]
3014 fn browser_desired_peer_mailbox_fences_stale_revisions_and_replaces_state() {
3015 let mut mailbox = BrowserDesiredPeerMailbox::default();
3016 let peer = BrowserDesiredPeer {
3017 device_id: " device-1 ".to_string(),
3018 node_id: Some(" node-1 ".to_string()),
3019 ticket: Some(" ticket-1 ".to_string()),
3020 online: true,
3021 session_id: Some(" runtime-1 ".to_string()),
3022 excluded_peers: vec![" DEVICE-LOCAL ".to_string()],
3023 topology_role: None,
3024 topology_revision: None,
3025 };
3026
3027 assert!(mailbox.accept_with_withdrawn(2, vec![peer]).is_some());
3028 assert!(mailbox.accept_with_withdrawn(1, Vec::new()).is_none());
3029 assert!(mailbox.accept_with_withdrawn(2, Vec::new()).is_none());
3030 assert_eq!(mailbox.revision, 2);
3031 let stored = mailbox.peers.get("device-1").expect("desired peer");
3032 assert_eq!(stored.node_id.as_deref(), Some("node-1"));
3033 assert_eq!(stored.excluded_peers, vec!["device-local"]);
3034
3035 let withdrawn = mailbox
3036 .accept_with_withdrawn(3, Vec::new())
3037 .expect("newer revision is accepted");
3038 assert_eq!(withdrawn.len(), 1);
3039 assert_eq!(withdrawn[0].device_id, "device-1");
3040 assert!(mailbox.peers.is_empty());
3041 assert!(mailbox.device_is_withdrawn("device-1"));
3042 assert_eq!(
3043 mailbox.withdrawn_device_for_node("node-1").as_deref(),
3044 Some("device-1")
3045 );
3046
3047 let mut restored = withdrawn[0].clone();
3048 restored.online = true;
3049 assert!(mailbox
3050 .accept_with_withdrawn(4, vec![restored])
3051 .expect("restored revision is accepted")
3052 .is_empty());
3053 assert!(!mailbox.device_is_withdrawn("device-1"));
3054 assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
3055 }
3056
3057 #[test]
3058 fn current_user_device_assignment_requires_exact_online_runtime_evidence() {
3059 let endpoint_id = iroh::SecretKey::generate().public();
3060 let node_id = endpoint_id.to_string();
3061 let iroh_ticket =
3062 iroh_tickets::endpoint::EndpointTicket::new(iroh::EndpointAddr::new(endpoint_id))
3063 .to_string();
3064 let ticket =
3065 crate::session_token::build_ticket(&iroh_ticket, "user-device-token", "user-device", 0);
3066 let peer = BrowserDesiredPeer {
3067 device_id: "device-1".to_string(),
3068 node_id: Some(node_id.clone()),
3069 ticket: Some(ticket.clone()),
3070 online: true,
3071 session_id: Some("runtime-1".to_string()),
3072 excluded_peers: Vec::new(),
3073 topology_role: None,
3074 topology_revision: None,
3075 };
3076 let evidence = browser_peer_evidence(&peer);
3077 let mut mailbox = BrowserDesiredPeerMailbox::default();
3078 assert!(mailbox.accept(1, vec![peer.clone()]));
3079 assert_eq!(
3080 mailbox.current_online_device_for_node(&node_id).as_deref(),
3081 Some("device-1")
3082 );
3083 assert!(
3084 mailbox.current_online_assignment_matches("device-1", &node_id, &ticket, &evidence,)
3085 );
3086 assert!(!mailbox.current_online_assignment_matches(
3087 "device-1",
3088 &node_id,
3089 "stale-ticket",
3090 &evidence,
3091 ));
3092
3093 let mut replacement = peer.clone();
3094 replacement.session_id = Some("runtime-2".to_string());
3095 assert!(mailbox.accept(2, vec![replacement.clone()]));
3096 assert!(
3097 !mailbox.current_online_assignment_matches("device-1", &node_id, &ticket, &evidence,)
3098 );
3099 let replacement_evidence = browser_peer_evidence(&replacement);
3100 assert!(mailbox.current_online_assignment_matches(
3101 "device-1",
3102 &node_id,
3103 &ticket,
3104 &replacement_evidence,
3105 ));
3106
3107 replacement.online = false;
3108 assert!(mailbox.accept(3, vec![replacement]));
3109 assert_eq!(mailbox.current_online_device_for_node(&node_id), None);
3110 assert!(!mailbox.current_online_assignment_matches(
3111 "device-1",
3112 &node_id,
3113 &ticket,
3114 &replacement_evidence,
3115 ));
3116 }
3117
3118 #[test]
3119 fn browser_desired_peer_mailbox_withdraws_live_route_when_device_goes_offline() {
3120 let mut mailbox = BrowserDesiredPeerMailbox::default();
3121 let online = BrowserDesiredPeer {
3122 device_id: "device-1".to_string(),
3123 node_id: Some("node-1".to_string()),
3124 ticket: Some("ticket-1".to_string()),
3125 online: true,
3126 session_id: Some("runtime-1".to_string()),
3127 excluded_peers: Vec::new(),
3128 topology_role: None,
3129 topology_revision: None,
3130 };
3131 assert!(mailbox
3132 .accept_with_withdrawn(1, vec![online.clone()])
3133 .expect("initial revision is accepted")
3134 .is_empty());
3135
3136 let mut offline = online.clone();
3137 offline.online = false;
3138 let withdrawn = mailbox
3139 .accept_with_withdrawn(2, vec![offline])
3140 .expect("offline revision is accepted");
3141 assert_eq!(withdrawn, vec![online]);
3142 assert!(mailbox.peers.contains_key("device-1"));
3143 assert!(!mailbox.peers["device-1"].online);
3144 assert!(mailbox.device_is_withdrawn("device-1"));
3145 assert_eq!(
3146 mailbox.withdrawn_device_for_node("node-1").as_deref(),
3147 Some("device-1")
3148 );
3149
3150 assert!(mailbox
3151 .accept_with_withdrawn(3, vec![mailbox.peers["device-1"].clone()])
3152 .expect("repeated offline revision is accepted")
3153 .is_empty());
3154 }
3155
3156 #[test]
3157 fn offline_boot_is_open() {
3158 let mut mailbox = BrowserDesiredPeerMailbox::default();
3159 let offline = BrowserDesiredPeer {
3160 device_id: "device-1".to_string(),
3161 node_id: Some("node-1".to_string()),
3162 ticket: Some("ticket-1".to_string()),
3163 online: false,
3164 session_id: None,
3165 excluded_peers: Vec::new(),
3166 topology_role: None,
3167 topology_revision: None,
3168 };
3169
3170 assert!(mailbox
3171 .accept_with_withdrawn(1, vec![offline])
3172 .expect("initial offline roster is accepted")
3173 .is_empty());
3174 assert!(!mailbox.device_is_withdrawn("device-1"));
3175 assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
3176 }
3177
3178 #[test]
3179 fn candidate_observation_expiry_removes_dial_intent_without_withdrawal() {
3180 let mut mailbox = BrowserDesiredPeerMailbox::default();
3181 let peer = BrowserDesiredPeer {
3182 device_id: "device-1".to_string(),
3183 node_id: Some("node-1".to_string()),
3184 ticket: Some("ticket-1".to_string()),
3185 online: true,
3186 session_id: None,
3187 excluded_peers: Vec::new(),
3188 topology_role: None,
3189 topology_revision: None,
3190 };
3191 assert!(mailbox
3192 .accept_with_withdrawn(1, vec![peer.clone()])
3193 .unwrap()
3194 .is_empty());
3195 assert!(mailbox
3196 .accept_with_observation_expired(2, Vec::new(), "device-1")
3197 .expect("newer observation revision")
3198 .is_empty());
3199 assert!(mailbox.peers.is_empty());
3200 assert!(!mailbox.device_is_withdrawn("device-1"));
3201 assert_eq!(mailbox.withdrawn_device_for_node("node-1"), None);
3202 assert!(mailbox
3203 .accept_with_observation_expired(2, Vec::new(), "device-1")
3204 .is_none());
3205
3206 assert!(mailbox
3207 .accept_with_withdrawn(3, vec![peer])
3208 .expect("rediscovery revision")
3209 .is_empty());
3210 assert!(mailbox.peers.contains_key("device-1"));
3211 assert!(!mailbox.device_is_withdrawn("device-1"));
3212 }
3213
3214 #[test]
3215 fn topology_backups_retire_only_after_current_active_edges_are_ready() {
3216 let peer = |device_id: &str, role: SparseTopologyRole, topology_revision: u64| {
3217 BrowserDesiredPeer {
3218 device_id: device_id.to_string(),
3219 node_id: Some(format!("node-{device_id}")),
3220 ticket: Some(format!("ticket-{device_id}")),
3221 online: true,
3222 session_id: Some(format!("runtime-{device_id}")),
3223 excluded_peers: Vec::new(),
3224 topology_role: Some(role),
3225 topology_revision: Some(topology_revision),
3226 }
3227 };
3228 let active_a = peer("active-a", SparseTopologyRole::Active, 7);
3229 let active_b = peer("active-b", SparseTopologyRole::Active, 7);
3230 let backup = peer("backup", SparseTopologyRole::Backup, 7);
3231 let membership_only = |mut peer: BrowserDesiredPeer| {
3232 peer.node_id = None;
3233 peer.ticket = None;
3234 peer.topology_role = None;
3235 peer.topology_revision = None;
3236 peer
3237 };
3238 let mut mailbox = BrowserDesiredPeerMailbox::default();
3239 assert!(mailbox
3240 .accept_with_withdrawn(10, vec![active_a.clone(), active_b.clone(), backup.clone()],)
3241 .expect("initial topology")
3242 .is_empty());
3243
3244 assert!(mailbox
3245 .acknowledge_topology_peer_ready("active-a", Some(7))
3246 .is_empty());
3247 assert!(mailbox.peers.contains_key("backup"));
3248 mailbox.observe_topology_peer_not_ready("active-a", Some(7));
3249 assert!(mailbox
3250 .acknowledge_topology_peer_ready("active-b", Some(7))
3251 .is_empty());
3252 assert!(mailbox.peers.contains_key("backup"));
3253 assert!(mailbox
3254 .acknowledge_topology_peer_ready("active-b", Some(6))
3255 .is_empty());
3256 assert!(mailbox.peers.contains_key("backup"));
3257
3258 assert_eq!(
3259 mailbox.acknowledge_topology_peer_ready("active-a", Some(7)),
3260 vec![backup.clone()],
3261 );
3262 assert!(!mailbox.peers.contains_key("backup"));
3263 assert!(mailbox.device_is_withdrawn("backup"));
3264
3265 assert!(mailbox
3268 .accept_with_withdrawn(11, vec![active_a.clone(), active_b.clone(), backup.clone()],)
3269 .expect("same topology revision")
3270 .is_empty());
3271 assert!(!mailbox.peers.contains_key("backup"));
3272
3273 let next_active = peer("active-next", SparseTopologyRole::Active, 8);
3276 let next_backup = peer("backup-next", SparseTopologyRole::Backup, 8);
3277 mailbox
3278 .accept_with_withdrawn(
3279 12,
3280 vec![
3281 next_active.clone(),
3282 next_backup.clone(),
3283 membership_only(active_a.clone()),
3284 membership_only(active_b.clone()),
3285 ],
3286 )
3287 .expect("next topology revision");
3288 assert!(mailbox
3289 .acknowledge_topology_peer_ready("active-a", Some(7))
3290 .is_empty());
3291 assert!(!mailbox.peers.contains_key("backup-next"));
3292 assert_eq!(
3293 mailbox.peers["active-a"].topology_role,
3294 Some(SparseTopologyRole::Backup),
3295 );
3296
3297 let latest_active = peer("active-latest", SparseTopologyRole::Active, 9);
3301 let speculative_overlap = peer("active-next", SparseTopologyRole::Backup, 9);
3302 mailbox
3303 .accept_with_withdrawn(
3304 13,
3305 vec![
3306 latest_active.clone(),
3307 speculative_overlap,
3308 membership_only(active_a.clone()),
3309 membership_only(active_b.clone()),
3310 membership_only(next_active.clone()),
3311 ],
3312 )
3313 .expect("rapid topology revision");
3314 assert_eq!(
3315 mailbox
3316 .peers
3317 .values()
3318 .filter(|peer| peer.topology_role == Some(SparseTopologyRole::Backup))
3319 .map(|peer| peer.device_id.as_str())
3320 .collect::<std::collections::BTreeSet<_>>(),
3321 std::collections::BTreeSet::from(["active-a", "active-b"]),
3322 );
3323 assert_eq!(
3324 mailbox.acknowledge_topology_peer_ready("active-latest", Some(9)),
3325 vec![active_a.clone(), active_b.clone()]
3326 .into_iter()
3327 .map(|mut peer| {
3328 peer.topology_role = Some(SparseTopologyRole::Backup);
3329 peer.topology_revision = Some(9);
3330 peer
3331 })
3332 .collect::<Vec<_>>(),
3333 );
3334
3335 let mut restarted = BrowserDesiredPeerMailbox::default();
3337 restarted
3338 .accept_with_withdrawn(1, vec![next_active, next_backup])
3339 .expect("restarted topology");
3340 assert!(restarted.peers.contains_key("backup-next"));
3341 }
3342
3343 #[test]
3344 fn browser_peer_evidence_fingerprints_ticket_instead_of_retaining_it() {
3345 let peer = BrowserDesiredPeer {
3346 device_id: "device-1".to_string(),
3347 node_id: Some("node-1".to_string()),
3348 ticket: Some("iroh-ticket.sensitive-token-material".to_string()),
3349 online: true,
3350 session_id: Some("runtime-1".to_string()),
3351 excluded_peers: Vec::new(),
3352 topology_role: None,
3353 topology_revision: None,
3354 };
3355
3356 let evidence = browser_peer_evidence(&peer);
3357 assert!(evidence.starts_with("node-1|"));
3358 assert!(!evidence.contains("sensitive-token-material"));
3359 }
3360
3361 #[test]
3362 fn desired_peer_attempt_survives_unrelated_mailbox_revision() {
3363 let mut mailbox = BrowserDesiredPeerMailbox::default();
3364 let tracked = BrowserDesiredPeer {
3365 device_id: "device-tracked".to_string(),
3366 node_id: Some("node-tracked".to_string()),
3367 ticket: Some("ticket-tracked".to_string()),
3368 online: true,
3369 session_id: Some("runtime-tracked".to_string()),
3370 excluded_peers: Vec::new(),
3371 topology_role: None,
3372 topology_revision: None,
3373 };
3374 let unrelated = BrowserDesiredPeer {
3375 device_id: "device-unrelated".to_string(),
3376 node_id: Some("node-unrelated".to_string()),
3377 ticket: Some("ticket-unrelated".to_string()),
3378 online: true,
3379 session_id: Some("runtime-unrelated".to_string()),
3380 excluded_peers: Vec::new(),
3381 topology_role: None,
3382 topology_revision: None,
3383 };
3384
3385 assert!(mailbox.accept(1, vec![tracked.clone()]));
3386 assert!(desired_peer_evidence_is_current(&mailbox, &tracked));
3387
3388 assert!(mailbox.accept(2, vec![tracked.clone(), unrelated.clone()]));
3389 assert!(desired_peer_evidence_is_current(&mailbox, &tracked));
3390
3391 let mut changed = tracked.clone();
3392 changed.session_id = Some("runtime-replacement".to_string());
3393 assert!(mailbox.accept(3, vec![changed, unrelated.clone()]));
3394 assert!(!desired_peer_evidence_is_current(&mailbox, &tracked));
3395
3396 assert!(mailbox.accept(4, vec![unrelated]));
3397 assert!(!desired_peer_evidence_is_current(&mailbox, &tracked));
3398 }
3399
3400 #[test]
3401 fn browser_desired_peer_tie_break_has_one_initial_dialer_and_bounded_takeover() {
3402 assert_eq!(
3403 browser_desired_peer_decision("node-z", "node-a", false, 1_000, None),
3404 BrowserDesiredPeerDecision::Dial
3405 );
3406 assert_eq!(
3407 browser_desired_peer_decision("node-a", "node-z", false, 1_000, None),
3408 BrowserDesiredPeerDecision::WaitUntil(4_000)
3409 );
3410 assert_eq!(
3411 browser_desired_peer_decision("node-a", "node-z", false, 4_000, Some(4_000)),
3412 BrowserDesiredPeerDecision::Dial
3413 );
3414 assert_eq!(
3415 browser_desired_peer_decision("node-a", "node-z", true, 1_000, None),
3416 BrowserDesiredPeerDecision::ObserveExisting
3417 );
3418 }
3419
3420 #[test]
3421 fn browser_retry_deadline_only_replaces_with_earlier_work_or_cancellation() {
3422 assert!(!should_replace_browser_retry_deadline(None, None));
3423 assert!(should_replace_browser_retry_deadline(None, Some(5_000)));
3424 assert!(!should_replace_browser_retry_deadline(
3425 Some(5_000),
3426 Some(6_000)
3427 ));
3428 assert!(should_replace_browser_retry_deadline(
3429 Some(5_000),
3430 Some(4_000)
3431 ));
3432 assert!(should_replace_browser_retry_deadline(Some(5_000), None));
3433 }
3434
3435 #[test]
3436 fn browser_failure_backoff_is_bounded_for_recovery() {
3437 assert_eq!(browser_auto_connect_failure_backoff_ms(1), 1_000);
3438 assert_eq!(browser_auto_connect_failure_backoff_ms(2), 2_000);
3439 assert_eq!(browser_auto_connect_failure_backoff_ms(10), 8_000);
3440 }
3441
3442 #[test]
3443 fn reconcile_deadlines_replace_expired_non_initiator_wait_with_failure_backoff() {
3444 let (retry_deadline, non_initiator_deadline) =
3445 browser_reconcile_deadlines(BrowserPeerReconcileOutcome::Retry, Some(9_000));
3446 assert_eq!(retry_deadline, Some(9_000));
3447 assert_eq!(non_initiator_deadline, None);
3448
3449 let (retry_deadline, non_initiator_deadline) =
3450 browser_reconcile_deadlines(BrowserPeerReconcileOutcome::WaitUntil(4_000), None);
3451 assert_eq!(retry_deadline, None);
3452 assert_eq!(non_initiator_deadline, Some(4_000));
3453 }
3454
3455 #[cfg(not(target_arch = "wasm32"))]
3456 #[tokio::test]
3457 async fn duplicate_external_auto_connect_preserves_gateway_mailbox_generation() {
3458 let client = std::sync::Arc::new(Client::new_with_app_tag(
3459 crate::test_constants::TEST_PROJECT_ID.to_string(),
3460 "test-app".to_string(),
3461 Box::new(|| None),
3462 ));
3463 client
3464 .start_external_auto_connect("test-user".to_string(), "local-device".to_string())
3465 .await
3466 .expect("external actor starts");
3467 assert!(client
3468 .submit_external_desired_peers(7, "[]")
3469 .await
3470 .expect("gateway revision is accepted"));
3471
3472 let before = client.external_desired_peer_debug_snapshot().await;
3473 client
3474 .start_external_auto_connect("test-user".to_string(), "local-device".to_string())
3475 .await
3476 .expect("duplicate start is idempotent");
3477 let after = client.external_desired_peer_debug_snapshot().await;
3478
3479 assert_eq!(after["generation"], before["generation"]);
3480 assert_eq!(after["revision"], serde_json::json!(7));
3481 assert_eq!(after["active"], serde_json::json!(true));
3482 }
3483}
3484
3485#[cfg(not(target_arch = "wasm32"))]
3486fn should_suppress_auto_connect_for_replacement_window(
3487 snapshot: &ConnectionHealthView,
3488 now_ms: i64,
3489 transport_connected: bool,
3490) -> bool {
3491 if !snapshot.replacement_pending || transport_connected {
3492 return false;
3493 }
3494
3495 if now_ms.saturating_sub(snapshot.last_lifecycle_transition_at_ms) >= MANAGED_SETTLE_DEADLINE_MS
3496 {
3497 return false;
3498 }
3499
3500 !matches!(
3504 snapshot.last_reconnect_reason.as_deref(),
3505 Some("transient-disconnect")
3506 )
3507}
3508
3509impl Client {
3510 async fn retire_desired_peer_connections(
3511 &self,
3512 remote_device_id: &str,
3513 remote_node_id: Option<&str>,
3514 reason: &str,
3515 ) {
3516 let mut records = self.resolve_peer_connection_records(remote_device_id).await;
3517 if records.is_empty() {
3518 if let Some(node_id) = remote_node_id
3519 .map(str::trim)
3520 .filter(|value| !value.is_empty())
3521 {
3522 records = self.resolve_peer_connection_records(node_id).await;
3523 }
3524 }
3525
3526 for record in records {
3527 let endpoint_id = record
3528 .endpoint_id
3529 .as_deref()
3530 .or(record.node_id.as_deref())
3531 .and_then(|value| value.trim().parse::<iroh::EndpointId>().ok());
3532 let disconnected = if let Some(endpoint_id) = endpoint_id {
3533 self.disconnect_with_reason(endpoint_id, reason)
3534 .await
3535 .is_ok()
3536 } else {
3537 false
3538 };
3539 if !disconnected {
3540 self.retire_managed_connection(&record.connection_id, Some(reason.to_string()))
3541 .await;
3542 }
3543 eprintln!(
3544 "[pluto-rtc][auto-connect] retired undesired connection remote_device_id={} connection_id={} reason={} transport_closed={}",
3545 remote_device_id, record.connection_id, reason, disconnected
3546 );
3547 }
3548 }
3549
3550 async fn retire_withdrawn_desired_peer_connections(
3551 &self,
3552 remote_device_id: &str,
3553 remote_node_id: Option<&str>,
3554 ) {
3555 self.retire_desired_peer_connections(
3556 remote_device_id,
3557 remote_node_id,
3558 crate::lifecycle_reason::REASON_DESIRED_PEER_WITHDRAWN,
3559 )
3560 .await;
3561 }
3562
3563 #[cfg(not(target_arch = "wasm32"))]
3564 pub(crate) async fn retire_offline_desired_route(
3565 &self,
3566 remote_device_id: &str,
3567 remote_node_id: Option<&str>,
3568 ) {
3569 self.retire_withdrawn_desired_peer_connections(remote_device_id, remote_node_id)
3570 .await;
3571 }
3572
3573 pub async fn is_connection_transport_alive(&self, endpoint_id: iroh::EndpointId) -> bool {
3574 if !self.is_connected(endpoint_id).await {
3575 return false;
3576 }
3577
3578 #[cfg(not(target_arch = "wasm32"))]
3579 {
3580 let Some(connection) = self.get_connection(endpoint_id).await else {
3581 return false;
3582 };
3583
3584 return connection.close_reason().is_none();
3585 }
3586
3587 #[cfg(target_arch = "wasm32")]
3588 {
3589 let Some(connection) = self.get_connection(endpoint_id).await else {
3590 return false;
3591 };
3592
3593 connection.close_reason().is_none()
3594 }
3595 }
3596
3597 #[cfg(not(target_arch = "wasm32"))]
3598 pub(crate) async fn validate_active_connection(
3599 &self,
3600 endpoint_id: iroh::EndpointId,
3601 ) -> Option<crate::heartbeat::IrohConnectionProbe> {
3602 if !self.is_connection_transport_alive(endpoint_id).await {
3603 return None;
3604 }
3605 let node = self.iroh_node.read().await.as_ref().cloned()?;
3606 node.probe_connection(endpoint_id, std::time::Duration::from_secs(4))
3607 .await
3608 }
3609
3610 #[cfg(not(target_arch = "wasm32"))]
3611 pub(crate) async fn should_force_network_change_after_connect_failures(
3612 &self,
3613 remote_device_id: &str,
3614 ) -> bool {
3615 for record in self.connection_manager.list_active().await {
3616 if !matches!(
3617 record.state,
3618 crate::connection_manager::ConnectionState::Connected
3619 ) {
3620 continue;
3621 }
3622
3623 let Some(record_device_id) = record
3624 .device_id
3625 .as_deref()
3626 .or(record.device_id_hint.as_deref())
3627 .map(str::trim)
3628 .filter(|value| !value.is_empty())
3629 else {
3630 continue;
3631 };
3632
3633 if record_device_id == remote_device_id {
3634 continue;
3635 }
3636
3637 eprintln!(
3638 "[pluto-rtc][auto-connect] suppress network-change recovery remote_device_id={} because healthy peer remains connected peer_device_id={} connection_id={}",
3639 remote_device_id,
3640 record_device_id,
3641 record.connection_id
3642 );
3643 return false;
3644 }
3645
3646 true
3647 }
3648
3649 #[cfg(not(target_arch = "wasm32"))]
3650 pub(crate) async fn try_auto_connect_device(
3651 self: &Arc<Self>,
3652 user_id: &str,
3653 device: crate::signaling::Device,
3654 local_device_id: &str,
3655 last_attempt_at: &mut std::collections::HashMap<String, i64>,
3656 failure_count: &mut std::collections::HashMap<String, u8>,
3657 failure_backoff_until: &mut std::collections::HashMap<String, i64>,
3658 retry_throttle_ms: i64,
3659 non_initiator_wait_started_at: &mut std::collections::HashMap<String, i64>,
3660 non_initiator_escalation_count: &mut std::collections::HashMap<String, u8>,
3661 initiator_grace_ms: i64,
3662 skip_log_at: &mut std::collections::HashMap<String, i64>,
3663 connected_health_checked_at: &mut std::collections::HashMap<String, i64>,
3664 connected_health_state: &mut std::collections::HashMap<String, bool>,
3665 connected_health_failures: &mut std::collections::HashMap<String, u8>,
3666 connected_health_probe_ms: i64,
3667 connected_health_failure_threshold: u8,
3668 last_presence_republish_at_ms: &mut i64,
3669 presence_republish_interval_ms: i64,
3670 last_network_change_at_ms: &mut i64,
3671 network_change_recovery_interval_ms: i64,
3672 network_change_failure_threshold: u8,
3673 last_known_node_id: &mut std::collections::HashMap<String, String>,
3674 ) {
3675 if self.is_app_execution_suspended() {
3676 return;
3677 }
3678
3679 #[cfg(target_arch = "wasm32")]
3680 let _ = (
3681 user_id,
3682 last_presence_republish_at_ms,
3683 presence_republish_interval_ms,
3684 last_network_change_at_ms,
3685 network_change_recovery_interval_ms,
3686 network_change_failure_threshold,
3687 );
3688
3689 let remote_device_id = device.device_id.clone();
3690 let now = now_millis_i64();
3691 let skip_log_interval_ms = 2_000;
3692
3693 let mut log_skip = |reason: &str| {
3694 let is_noise_reason = matches!(
3695 reason,
3696 "already-connected"
3697 | "tie-break-not-initiator-waiting"
3698 | "retry-throttle"
3699 | "self-ticket"
3700 | "auto-connect-excluded"
3701 | "remote-excluded-local-device"
3702 );
3703 if is_noise_reason && !auto_connect_verbose() {
3704 return;
3705 }
3706
3707 let skip_key = format!("{}::{}", remote_device_id, reason);
3708 let should_log = match skip_log_at.get(&skip_key) {
3709 Some(last_at) => now.saturating_sub(*last_at) >= skip_log_interval_ms,
3710 None => true,
3711 };
3712 if !should_log {
3713 return;
3714 }
3715 skip_log_at.insert(skip_key, now);
3716
3717 #[cfg(not(target_arch = "wasm32"))]
3718 eprintln!(
3719 "[pluto-rtc][auto-connect] skip remote_device_id={} local_device_id={} reason={}",
3720 remote_device_id, local_device_id, reason
3721 );
3722 #[cfg(target_arch = "wasm32")]
3723 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
3724 "[pluto-rtc][auto-connect] skip remote_device_id={} local_device_id={} reason={}",
3725 remote_device_id, local_device_id, reason
3726 )));
3727 };
3728
3729 if !device.online {
3730 last_attempt_at.remove(&remote_device_id);
3731 non_initiator_wait_started_at.remove(&remote_device_id);
3732 non_initiator_escalation_count.remove(&remote_device_id);
3733 connected_health_checked_at.remove(&remote_device_id);
3734 connected_health_state.remove(&remote_device_id);
3735 connected_health_failures.remove(&remote_device_id);
3736 clear_auto_connect_failure_state(
3737 &remote_device_id,
3738 failure_count,
3739 failure_backoff_until,
3740 );
3741 return;
3742 }
3743
3744 if remote_device_id == local_device_id {
3745 last_attempt_at.remove(&remote_device_id);
3746 non_initiator_wait_started_at.remove(&remote_device_id);
3747 non_initiator_escalation_count.remove(&remote_device_id);
3748 connected_health_checked_at.remove(&remote_device_id);
3749 connected_health_state.remove(&remote_device_id);
3750 connected_health_failures.remove(&remote_device_id);
3751 clear_auto_connect_failure_state(
3752 &remote_device_id,
3753 failure_count,
3754 failure_backoff_until,
3755 );
3756 return;
3757 }
3758
3759 if remote_excludes_local_device(&device.excluded_peers, local_device_id) {
3760 log_skip("remote-excluded-local-device");
3761 return;
3762 }
3763
3764 if self.is_auto_connect_peer_excluded(&remote_device_id, device.node_id.as_deref()) {
3765 log_skip("auto-connect-excluded");
3766 return;
3767 }
3768
3769 let provider_ticket = match device.ticket.as_deref() {
3770 Some(ticket) if !ticket.trim().is_empty() => ticket,
3771 _ => {
3772 log_skip("missing-ticket");
3773 return;
3774 }
3775 };
3776
3777 let (provider_iroh_ticket, _) = crate::session_token::split_ticket(provider_ticket.trim());
3778 let provider_endpoint_addr = match parse_endpoint_ticket(provider_iroh_ticket) {
3779 Ok(addr) => addr,
3780 Err(error) => {
3781 log_skip("invalid-ticket");
3782 eprintln!(
3783 "[pluto-rtc][auto-connect] invalid ticket remote_device_id={} error={}",
3784 remote_device_id, error
3785 );
3786 return;
3787 }
3788 };
3789 let preferred_ticket = self.preferred_scoped_route_ticket(
3790 provider_endpoint_addr.id.to_string().as_str(),
3791 provider_iroh_ticket,
3792 );
3793 let ticket_str = preferred_ticket.as_deref().unwrap_or(provider_ticket);
3794
3795 let (iroh_ticket, token_suffix) = crate::session_token::split_ticket(ticket_str.trim());
3798 let extracted_token_payload = token_suffix
3799 .and_then(|suffix| crate::session_token::decode_payload(iroh_ticket, suffix));
3800 let extracted_token = extracted_token_payload
3801 .as_ref()
3802 .map(|payload| payload.token.clone());
3803 let extracted_token_suffix = token_suffix
3804 .filter(|_| extracted_token_payload.is_some())
3805 .map(ToOwned::to_owned);
3806
3807 let endpoint_addr = provider_endpoint_addr;
3808 #[cfg(feature = "transport-lan")]
3809 if self.transport_config.read().await.network_policy
3810 == crate::client::NetworkPolicy::LocalOnly
3811 && !crate::local_discovery::endpoint_addr_is_local_only(&endpoint_addr, &[])
3812 {
3813 log_skip("local-only-address-rejected");
3814 return;
3815 }
3816 let endpoint_id = endpoint_addr.id;
3817 let node_id_str = endpoint_id.to_string();
3818 #[cfg(not(target_arch = "wasm32"))]
3819 if let Some(payload) = extracted_token_payload
3820 .as_ref()
3821 .filter(|payload| payload.scope.as_str() == "user-device")
3822 {
3823 self.update_native_route_repair_credential(
3824 &node_id_str,
3825 Some(&remote_device_id),
3826 extracted_token.as_deref(),
3827 extracted_token_suffix.as_deref(),
3828 &payload.scope,
3829 );
3830 }
3831 let runtime_instance_changed = observe_remote_runtime_instance(
3832 last_known_node_id,
3833 &remote_device_id,
3834 device.session_id.as_deref(),
3835 );
3836 let ticket_changed = reconcile_remote_ticket_retry_state(
3837 last_known_node_id,
3838 &remote_device_id,
3839 provider_ticket,
3840 last_attempt_at,
3841 failure_count,
3842 failure_backoff_until,
3843 );
3844
3845 if ticket_changed {
3846 eprintln!(
3847 "[pluto-rtc][auto-connect] peer ticket changed remote_device_id={} node_id={} - clearing stale retry deadline",
3848 remote_device_id, node_id_str
3849 );
3850 }
3851
3852 {
3856 let identity_changed = last_known_node_id
3857 .get(&remote_device_id)
3858 .map(|prev| prev.as_str() != node_id_str)
3859 .unwrap_or(false);
3860 last_known_node_id.insert(remote_device_id.clone(), node_id_str.clone());
3861 if identity_changed {
3862 #[cfg(not(target_arch = "wasm32"))]
3863 eprintln!(
3864 "[pluto-rtc][auto-connect] peer identity changed remote_device_id={} new_node_id={} — resetting state for fast reconnect",
3865 remote_device_id, node_id_str
3866 );
3867 #[cfg(target_arch = "wasm32")]
3868 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
3869 "[pluto-rtc][auto-connect] peer identity changed remote_device_id={} new_node_id={} — resetting state for fast reconnect",
3870 remote_device_id, node_id_str
3871 )));
3872 last_attempt_at.remove(&remote_device_id);
3873 non_initiator_wait_started_at.remove(&remote_device_id);
3874 non_initiator_escalation_count.remove(&remote_device_id);
3875 connected_health_checked_at.remove(&remote_device_id);
3876 connected_health_state.remove(&remote_device_id);
3877 connected_health_failures.remove(&remote_device_id);
3878 clear_auto_connect_failure_state(
3879 &remote_device_id,
3880 failure_count,
3881 failure_backoff_until,
3882 );
3883 }
3884 }
3885
3886 let local_node_id = match self.current_node_id().await {
3887 Some(id) => id,
3888 None => {
3889 log_skip("local-node-not-ready");
3890 return;
3891 }
3892 };
3893
3894 if local_node_id == node_id_str {
3895 log_skip("self-ticket");
3896 return;
3897 }
3898
3899 if runtime_instance_changed {
3900 let current_transport = self.get_connection(endpoint_id).await;
3907 let current_transport_stable_id = current_transport
3908 .as_ref()
3909 .map(crate::transport_generation::for_connection);
3910 #[cfg(openrtc_iroh_retire_remote_connections_api)]
3911 if let Some(current_transport) = current_transport.as_ref() {
3912 let transport_stable_id =
3913 crate::transport_generation::for_connection(current_transport);
3914 let endpoint = { self.iroh_endpoint.read().await.clone() };
3923 if let Some(endpoint) = endpoint {
3924 let retired = retire_stale_endpoint_connections(
3925 &endpoint,
3926 endpoint_id,
3927 current_transport,
3928 crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT
3929 .as_bytes(),
3930 )
3931 .await;
3932 if retired > 0 {
3933 eprintln!(
3934 "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - retired {} stale shared-endpoint protocol connection(s), preserved transport_stable_id={}",
3935 remote_device_id, node_id_str, retired, transport_stable_id
3936 );
3937 }
3938 }
3939 }
3940 let active_transport_responsive = if current_transport_stable_id.is_some() {
3941 self.validate_active_connection(endpoint_id)
3942 .await
3943 .is_some_and(|probe| probe.responsive)
3944 } else {
3945 false
3946 };
3947 if runtime_instance_change_requires_replacement(
3948 runtime_instance_changed,
3949 active_transport_responsive,
3950 ) {
3951 if let Some(transport_stable_id) = current_transport_stable_id {
3952 match self
3953 .retire_transport_generation_after_failed_probe(
3954 &Self::deterministic_connection_id(&local_node_id, &node_id_str),
3955 endpoint_id,
3956 transport_stable_id,
3957 connected_health_probe_ms,
3958 crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
3959 )
3960 .await
3961 {
3962 FailedProbeRetirementOutcome::Retired => {
3963 eprintln!(
3964 "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - replaced unresponsive physical transport",
3965 remote_device_id, node_id_str
3966 );
3967 }
3968 FailedProbeRetirementOutcome::DeferredAdmissionInFlight => {
3969 eprintln!(
3970 "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - deferred replacement while admission is in flight",
3971 remote_device_id, node_id_str
3972 );
3973 }
3974 FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity => {
3975 eprintln!(
3976 "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - preserved generation with recent protocol activity",
3977 remote_device_id, node_id_str
3978 );
3979 }
3980 FailedProbeRetirementOutcome::GenerationChanged => {
3981 eprintln!(
3982 "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - ignored stale generation",
3983 remote_device_id, node_id_str
3984 );
3985 }
3986 }
3987 }
3988 } else {
3989 eprintln!(
3990 "[pluto-rtc][auto-connect] peer runtime instance changed remote_device_id={} node_id={} - preserving responsive physical transport",
3991 remote_device_id, node_id_str
3992 );
3993 }
3994 last_attempt_at.remove(&remote_device_id);
3995 non_initiator_wait_started_at.remove(&remote_device_id);
3996 non_initiator_escalation_count.remove(&remote_device_id);
3997 connected_health_checked_at.remove(&remote_device_id);
3998 connected_health_state.remove(&remote_device_id);
3999 connected_health_failures.remove(&remote_device_id);
4000 clear_auto_connect_failure_state(
4001 &remote_device_id,
4002 failure_count,
4003 failure_backoff_until,
4004 );
4005 }
4006
4007 if let Some(discovered_node_id) = device.node_id.as_deref() {
4008 if discovered_node_id != node_id_str {
4009 log_skip("ticket-node-mismatch-discovery");
4010 eprintln!(
4011 "[pluto-rtc][auto-connect] rejected inconsistent discovery identity remote_device_id={} discovered_node_id={} ticket_node_id={}",
4012 remote_device_id, discovered_node_id, node_id_str
4013 );
4014 return;
4015 }
4016 }
4017
4018 self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
4025 .await;
4026
4027 #[cfg(feature = "transport-lan")]
4028 if self.is_peer_locally_reachable(&node_id_str).await {
4029 eprintln!(
4030 "[pluto-rtc][auto-connect] LAN peer reachable remote_device_id={} node_id={}",
4031 remote_device_id, node_id_str
4032 );
4033 }
4034
4035 let connection_id = Self::deterministic_connection_id(&local_node_id, &node_id_str);
4036
4037 let managed_health_snapshot = self.managed_connection_health(&connection_id).await;
4038
4039 #[cfg(not(target_arch = "wasm32"))]
4040 {
4041 if let Some(existing) = self
4042 .connection_manager
4043 .get_by_connection_id(&connection_id)
4044 .await
4045 {
4046 let existing_device_id = existing.device_id.as_deref().map(str::trim);
4047 if existing_device_id != Some(remote_device_id.as_str()) {
4048 self.connection_manager
4049 .set_device_id(&connection_id, remote_device_id.clone())
4050 .await;
4051 eprintln!(
4052 "[pluto-rtc][auto-connect] backfilled device_id hint connection_id={} node_id={} device_id={}",
4053 connection_id, node_id_str, remote_device_id
4054 );
4055 }
4056 }
4057 }
4058
4059 #[cfg(not(target_arch = "wasm32"))]
4064 let transport_connected = self.is_connection_transport_alive(endpoint_id).await;
4065 #[cfg(target_arch = "wasm32")]
4066 let transport_connected = self.is_connected(endpoint_id).await;
4067 if let Some(snapshot) = managed_health_snapshot.as_ref() {
4068 if should_suppress_auto_connect_for_replacement_window(
4073 &snapshot,
4074 now,
4075 transport_connected,
4076 ) {
4077 log_skip("replacement-pending");
4078 return;
4079 }
4080 }
4081 #[cfg(target_arch = "wasm32")]
4082 if !transport_connected {
4083 if let Some(existing) = self
4084 .connection_manager
4085 .get_by_connection_id(&connection_id)
4086 .await
4087 {
4088 if matches!(
4089 existing.state,
4090 crate::connection_manager::ConnectionState::Connected
4091 ) {
4092 web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
4093 "[pluto-rtc][auto-connect][wasm] transport mismatch remote_device_id={} remote_node_id={} connection_id={} record_stable_id={:?} record_generation={} state={:?} status_reason={:?}",
4094 remote_device_id,
4095 node_id_str,
4096 connection_id,
4097 existing.transport_stable_id,
4098 existing.transport_generation,
4099 existing.state,
4100 existing.status_reason
4101 )));
4102 }
4103 }
4104 }
4105
4106 let esc_count = non_initiator_escalation_count
4114 .get(&remote_device_id)
4115 .copied()
4116 .unwrap_or(0);
4117 let effective_grace_ms = if local_node_id <= node_id_str {
4118 non_initiator_escalation_grace_ms(esc_count)
4119 } else {
4120 initiator_grace_ms
4121 };
4122 let waited_ms = if local_node_id <= node_id_str {
4123 let wait_started_at = non_initiator_wait_started_at
4124 .entry(remote_device_id.clone())
4125 .or_insert(now);
4126 now.saturating_sub(*wait_started_at)
4127 } else {
4128 0
4129 };
4130 match auto_connect_tie_break_decision(
4131 &local_node_id,
4132 &node_id_str,
4133 transport_connected,
4134 waited_ms,
4135 effective_grace_ms,
4136 ) {
4137 AutoConnectTieBreakDecision::ObserveConnectedTransport => {
4138 non_initiator_wait_started_at.remove(&remote_device_id);
4139 non_initiator_escalation_count.remove(&remote_device_id);
4140 }
4141 AutoConnectTieBreakDecision::WaitForInitiator => {
4142 if waited_ms < 1_000 {
4143 log_skip("tie-break-not-initiator-waiting");
4144 }
4145 return;
4146 }
4147 AutoConnectTieBreakDecision::ActAsInitiator => {
4148 if local_node_id <= node_id_str {
4149 let new_esc_count = esc_count.saturating_add(1);
4151 non_initiator_escalation_count.insert(remote_device_id.clone(), new_esc_count);
4152 #[cfg(not(target_arch = "wasm32"))]
4153 eprintln!(
4154 "[pluto-rtc][auto-connect] non-initiator escalated after {}ms grace remote_device_id={} local_device_id={} escalation_count={}",
4155 waited_ms, remote_device_id, local_device_id, new_esc_count
4156 );
4157 #[cfg(target_arch = "wasm32")]
4158 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
4159 "[pluto-rtc][auto-connect] non-initiator escalated after {}ms grace remote_device_id={} local_device_id={} escalation_count={}",
4160 waited_ms, remote_device_id, local_device_id, new_esc_count
4161 )));
4162 }
4163 non_initiator_wait_started_at.remove(&remote_device_id);
4164 }
4165 }
4166
4167 self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
4168 .await;
4169
4170 if transport_connected {
4171 let current_scopes = self.connection_manager.get_scopes(&connection_id).await;
4172 let admitted_scope = match self.session_admission(&connection_id) {
4173 crate::session_token::SessionAdmission::Accepted {
4174 scope: Some(scope), ..
4175 } => Some(scope.into_inner()),
4176 _ => None,
4177 };
4178 let presentation_in_flight = self
4179 .session_token_registry
4180 .is_session_token_presentation_in_flight(&connection_id);
4181 if !persistent_managed_actor_may_own_transport(
4182 ¤t_scopes,
4183 admitted_scope.as_deref(),
4184 presentation_in_flight,
4185 ) {
4186 log_skip(if presentation_in_flight {
4187 "explicit-admission-presentation-owns-current-transport"
4188 } else {
4189 "non-managed-scope-owns-current-transport"
4190 });
4191 return;
4192 }
4193
4194 connected_health_checked_at
4195 .entry(remote_device_id.clone())
4196 .or_insert(now);
4197 connected_health_state
4198 .entry(remote_device_id.clone())
4199 .or_insert(true);
4200
4201 let should_probe = match connected_health_checked_at.get(&remote_device_id) {
4202 Some(last_probe_at) => {
4203 now.saturating_sub(*last_probe_at) >= connected_health_probe_ms
4204 }
4205 None => false,
4206 };
4207 let mut healthy = connected_health_state
4208 .get(&remote_device_id)
4209 .copied()
4210 .unwrap_or(true);
4211 let mut probed_transport_stable_id = None;
4212
4213 if should_probe {
4214 let active_probe = self.validate_active_connection(endpoint_id).await;
4215 probed_transport_stable_id =
4216 active_probe.as_ref().map(|probe| probe.transport_stable_id);
4217 let active_probe_healthy =
4218 active_probe.as_ref().is_some_and(|probe| probe.responsive);
4219 let passive_transport_alive = self.is_connection_transport_alive(endpoint_id).await;
4220 let recent_generation_activity = active_probe.as_ref().is_some_and(|probe| {
4221 generation_activity_is_recent(
4222 probe.last_inbound_activity_age,
4223 connected_health_probe_ms,
4224 ) || generation_activity_is_recent(
4225 self.native_transport_protocol_activity_age(
4226 &connection_id,
4227 probe.transport_stable_id,
4228 ),
4229 connected_health_probe_ms,
4230 )
4231 });
4232 let probe_decision = active_connection_probe_decision_with_generation_activity(
4233 active_probe_healthy,
4234 passive_transport_alive,
4235 recent_generation_activity,
4236 );
4237 let current_failures = connected_health_failures
4238 .get(&remote_device_id)
4239 .copied()
4240 .unwrap_or(0);
4241 let failures =
4242 next_active_connection_probe_failure_count(probe_decision, current_failures);
4243 healthy = active_connection_probe_is_healthy(
4244 probe_decision,
4245 failures,
4246 connected_health_failure_threshold,
4247 );
4248 connected_health_checked_at.insert(remote_device_id.clone(), now);
4249 connected_health_state.insert(remote_device_id.clone(), healthy);
4250 if matches!(probe_decision, ActiveConnectionProbeDecision::Healthy) {
4251 connected_health_failures.remove(&remote_device_id);
4252 let _ = self
4258 .confirm_managed_connection_readiness(&connection_id)
4259 .await;
4260 } else {
4261 connected_health_failures.insert(remote_device_id.clone(), failures);
4262 if matches!(
4263 probe_decision,
4264 ActiveConnectionProbeDecision::PreserveLiveTransport
4265 ) && healthy
4266 {
4267 log_skip("active-probe-missed-preserving-live-transport");
4268 }
4269 }
4270 }
4271
4272 if !healthy {
4273 if !should_probe {
4274 log_skip("awaiting-current-generation-health-probe");
4275 return;
4276 }
4277 if self.is_app_execution_suspended() {
4278 log_skip("backgrounded-after-health-probe");
4279 return;
4280 }
4281 #[cfg(not(target_arch = "wasm32"))]
4282 {
4283 let failures = connected_health_failures
4284 .get(&remote_device_id)
4285 .copied()
4286 .unwrap_or(1);
4287 if failures < connected_health_failure_threshold {
4288 log_skip("active-connection-health-probe-failed");
4289 return;
4290 }
4291 let Some(transport_stable_id) = probed_transport_stable_id else {
4292 log_skip("active-probe-generation-disappeared");
4293 connected_health_checked_at.remove(&remote_device_id);
4294 connected_health_state.remove(&remote_device_id);
4295 connected_health_failures.remove(&remote_device_id);
4296 return;
4297 };
4298 match self
4299 .retire_transport_generation_after_failed_probe(
4300 &connection_id,
4301 endpoint_id,
4302 transport_stable_id,
4303 connected_health_probe_ms,
4304 crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
4305 )
4306 .await
4307 {
4308 FailedProbeRetirementOutcome::Retired => {
4309 log_skip("stale-active-connection-reconnect");
4310 }
4311 FailedProbeRetirementOutcome::DeferredAdmissionInFlight => {
4312 log_skip("active-probe-retirement-deferred-admission-in-flight");
4313 connected_health_checked_at.insert(remote_device_id.clone(), now);
4314 connected_health_state.insert(remote_device_id.clone(), true);
4315 connected_health_failures.remove(&remote_device_id);
4316 return;
4317 }
4318 FailedProbeRetirementOutcome::CancelledCurrentGenerationActivity => {
4319 log_skip(
4320 "active-probe-retirement-cancelled-by-current-generation-activity",
4321 );
4322 connected_health_checked_at.insert(remote_device_id.clone(), now);
4323 connected_health_state.insert(remote_device_id.clone(), true);
4324 connected_health_failures.remove(&remote_device_id);
4325 let _ = self
4326 .confirm_managed_connection_readiness(&connection_id)
4327 .await;
4328 return;
4329 }
4330 FailedProbeRetirementOutcome::GenerationChanged => {
4331 log_skip("active-probe-generation-disappeared");
4332 connected_health_checked_at.remove(&remote_device_id);
4333 connected_health_state.remove(&remote_device_id);
4334 connected_health_failures.remove(&remote_device_id);
4335 return;
4336 }
4337 }
4338 self.maybe_refresh_gateway_presence_for_auto_connect(
4339 user_id,
4340 local_device_id,
4341 "stale-active-connection-reconnect",
4342 last_presence_republish_at_ms,
4343 presence_republish_interval_ms,
4344 )
4345 .await;
4346 self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
4347 .await;
4348 connected_health_checked_at.remove(&remote_device_id);
4349 connected_health_state.remove(&remote_device_id);
4350 connected_health_failures.remove(&remote_device_id);
4351 if failures >= network_change_failure_threshold {
4352 self.maybe_force_network_change_for_auto_connect(
4353 user_id,
4354 local_device_id,
4355 "stale-active-connection",
4356 last_network_change_at_ms,
4357 network_change_recovery_interval_ms,
4358 last_presence_republish_at_ms,
4359 presence_republish_interval_ms,
4360 )
4361 .await;
4362 }
4363 }
4364 #[cfg(target_arch = "wasm32")]
4365 {
4366 let failures = connected_health_failures
4367 .get(&remote_device_id)
4368 .copied()
4369 .unwrap_or(1);
4370 if failures < connected_health_failure_threshold {
4371 eprintln!(
4372 "[pluto-rtc][auto-connect][health] remote_device_id={} local_device_id={} remote_node_id={} healthy=false failures={} threshold={} action=wait",
4373 remote_device_id,
4374 local_device_id,
4375 node_id_str,
4376 failures,
4377 connected_health_failure_threshold
4378 );
4379 log_skip("active-connection-health-probe-failed");
4380 return;
4381 }
4382 eprintln!(
4383 "[pluto-rtc][auto-connect][health] remote_device_id={} local_device_id={} remote_node_id={} healthy=false failures={} threshold={} action=disconnect-and-reconnect",
4384 remote_device_id,
4385 local_device_id,
4386 node_id_str,
4387 failures,
4388 connected_health_failure_threshold
4389 );
4390 log_skip("stale-active-connection-reconnect");
4391 let _ = self
4392 .disconnect_with_reason(
4393 endpoint_id,
4394 crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
4395 )
4396 .await;
4397 connected_health_checked_at.remove(&remote_device_id);
4398 connected_health_state.remove(&remote_device_id);
4399 connected_health_failures.remove(&remote_device_id);
4400 }
4401 } else {
4402 if self.is_auto_connect_peer_excluded(&remote_device_id, Some(&node_id_str)) {
4410 log_skip("auto-connect-excluded-after-health-probe");
4411 let _ = self
4412 .disconnect_with_reason(
4413 endpoint_id,
4414 crate::lifecycle_reason::REASON_MANUAL_DISCONNECT,
4415 )
4416 .await;
4417 return;
4418 }
4419 self.connection_manager
4420 .upsert_pending(
4421 connection_id.clone(),
4422 Some(node_id_str.clone()),
4423 Some(remote_device_id.clone()),
4424 Some(node_id_str.clone()),
4425 )
4426 .await;
4427 let current_transport_stable_id = self
4428 .get_connection(endpoint_id)
4429 .await
4430 .map(|connection| crate::transport_generation::for_connection(&connection));
4431 self.commit_current_native_transport_record(
4432 &connection_id,
4433 &node_id_str,
4434 Some(remote_device_id.clone()),
4435 current_transport_stable_id,
4436 Some("health-check".to_string()),
4437 )
4438 .await;
4439 #[cfg(target_arch = "wasm32")]
4440 {
4441 let active_transport =
4442 self.iroh_path_kind(&node_id_str).await.transport_label();
4443 let _ = self
4444 .report_transport_status_for_current_generation(
4445 &connection_id,
4446 active_transport,
4447 None,
4448 )
4449 .await;
4450 }
4451
4452 #[cfg(not(target_arch = "wasm32"))]
4459 let has_remote_admission_proof =
4460 current_transport_stable_id.is_some_and(|transport_stable_id| {
4461 self.remote_session_token_admitted_for_transport(
4462 &connection_id,
4463 transport_stable_id,
4464 )
4465 });
4466 #[cfg(target_arch = "wasm32")]
4467 let has_remote_admission_proof = if let Some(token) = extracted_token.as_deref() {
4468 self.has_current_remote_session_admission_proof(
4469 &connection_id,
4470 endpoint_id,
4471 token,
4472 )
4473 .await
4474 } else {
4475 false
4476 };
4477 #[cfg(not(target_arch = "wasm32"))]
4478 let has_current_inbound_admission_proof =
4479 current_transport_stable_id.is_some_and(|transport_stable_id| {
4480 self.inbound_session_token_admitted_for_transport(
4481 &connection_id,
4482 transport_stable_id,
4483 )
4484 });
4485 #[cfg(not(target_arch = "wasm32"))]
4486 let peer_requested_reciprocal =
4487 self.has_pending_reciprocal_session_admission_request(&connection_id);
4488 #[cfg(not(target_arch = "wasm32"))]
4489 if has_remote_admission_proof
4490 && has_current_inbound_admission_proof
4491 && peer_requested_reciprocal
4492 {
4493 self.clear_reciprocal_session_admission_request(&connection_id);
4494 }
4495 #[cfg(not(target_arch = "wasm32"))]
4496 let presentation_decision = session_admission_presentation_decision(
4497 extracted_token.is_some(),
4498 has_remote_admission_proof,
4499 has_current_inbound_admission_proof,
4500 peer_requested_reciprocal,
4501 local_node_id > node_id_str,
4502 );
4503 #[cfg(target_arch = "wasm32")]
4504 let presentation_decision = session_admission_presentation_decision(
4505 extracted_token.is_some(),
4506 has_remote_admission_proof,
4507 true,
4508 false,
4509 true,
4510 );
4511 if presentation_decision.should_present {
4512 if let Some(backoff_until) =
4513 failure_backoff_until.get(&remote_device_id).copied()
4514 {
4515 if now < backoff_until {
4516 log_skip("session-admission-backoff");
4517 return;
4518 }
4519 failure_backoff_until.remove(&remote_device_id);
4520 }
4521
4522 let token = extracted_token
4523 .as_ref()
4524 .expect("representation requires an extracted token");
4525 eprintln!(
4526 "[pluto-rtc][auto-connect][session-admission] re-presenting token on live transport connection_id={} remote_device_id={} remote_node_id={} token_fp={}",
4527 connection_id,
4528 remote_device_id,
4529 node_id_str,
4530 super::core_impl::log_fingerprint(token.as_str())
4531 );
4532 #[cfg(not(target_arch = "wasm32"))]
4533 let presentation_result = self
4534 .present_and_accept_session_token_for_route_repair(
4535 endpoint_id,
4536 &connection_id,
4537 token,
4538 extracted_token_suffix.as_deref(),
4539 Some(remote_device_id.clone()),
4540 Some(local_device_id.to_string()),
4541 has_remote_admission_proof,
4542 presentation_decision.request_reciprocal,
4543 )
4544 .await;
4545 #[cfg(target_arch = "wasm32")]
4546 let presentation_result = self
4547 .present_and_accept_session_token_with_local_claim(
4548 endpoint_id,
4549 &connection_id,
4550 token,
4551 extracted_token_suffix.as_deref(),
4552 Some(remote_device_id.clone()),
4553 Some(local_device_id.to_string()),
4554 )
4555 .await;
4556 match presentation_result {
4557 Ok(approval_scope) => {
4558 clear_auto_connect_failure_state(
4559 &remote_device_id,
4560 failure_count,
4561 failure_backoff_until,
4562 );
4563 eprintln!(
4564 "[pluto-rtc][auto-connect][session-admission] live transport admitted connection_id={} remote_device_id={} remote_node_id={} scope={} token_fp={}",
4565 connection_id,
4566 remote_device_id,
4567 node_id_str,
4568 approval_scope,
4569 super::core_impl::log_fingerprint(token.as_str())
4570 );
4571 }
4572 Err(error) => {
4573 let active_transport_stable_id =
4574 self.get_connection(endpoint_id).await.map(|connection| {
4575 crate::transport_generation::for_connection(&connection)
4576 });
4577 if current_transport_stable_id != active_transport_stable_id {
4578 eprintln!(
4579 "[pluto-rtc][auto-connect][session-admission] stale presentation failure ignored because replacement already owns endpoint connection_id={} remote_device_id={} attempted_transport_stable_id={:?} active_transport_stable_id={:?}",
4580 connection_id,
4581 remote_device_id,
4582 current_transport_stable_id,
4583 active_transport_stable_id
4584 );
4585 return;
4586 }
4587
4588 #[cfg(not(target_arch = "wasm32"))]
4589 let current_generation_is_routable = active_transport_stable_id
4590 .is_some_and(|transport_stable_id| {
4591 self.native_application_stream_admitted_for_transport(
4592 &connection_id,
4593 transport_stable_id,
4594 )
4595 });
4596 #[cfg(target_arch = "wasm32")]
4597 let current_generation_is_routable = false;
4598
4599 if current_generation_is_routable {
4600 clear_auto_connect_failure_state(
4601 &remote_device_id,
4602 failure_count,
4603 failure_backoff_until,
4604 );
4605 connected_health_failures.remove(&remote_device_id);
4606 let _ = self
4607 .confirm_managed_connection_readiness(&connection_id)
4608 .await;
4609 eprintln!(
4610 "[pluto-rtc][auto-connect][session-admission] ignored losing presentation failure because another exchange made the generation routable connection_id={} remote_device_id={} transport_stable_id={:?} error={}",
4611 connection_id,
4612 remote_device_id,
4613 active_transport_stable_id,
4614 error,
4615 );
4616 return;
4617 }
4618
4619 if session_token_presentation_was_duplicate(&error) {
4625 eprintln!(
4626 "[pluto-rtc][auto-connect][session-admission] ignored duplicate presentation owned by canonical exchange connection_id={} remote_device_id={} transport_stable_id={:?}",
4627 connection_id,
4628 remote_device_id,
4629 active_transport_stable_id,
4630 );
4631 return;
4632 }
4633
4634 if self
4635 .session_token_registry
4636 .is_session_token_presentation_in_flight(&connection_id)
4637 {
4638 eprintln!(
4639 "[pluto-rtc][auto-connect][session-admission] ignored duplicate presentation while canonical exchange remains in flight connection_id={} remote_device_id={} transport_stable_id={:?}",
4640 connection_id,
4641 remote_device_id,
4642 active_transport_stable_id,
4643 );
4644 return;
4645 }
4646
4647 if session_token_presentation_failure_action(&error)
4648 == SessionTokenPresentationFailureAction::RejectAndDisconnect
4649 {
4650 let rejection_reason =
4651 format!("session-token-presentation-failed: {}", error);
4652 self.reject_session_connection(
4653 &connection_id,
4654 rejection_reason.as_str(),
4655 );
4656 self.connection_manager
4657 .set_failed(&connection_id, Some(rejection_reason.clone()))
4658 .await;
4659 let _ = self
4660 .disconnect_with_reason(
4661 endpoint_id,
4662 crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
4663 )
4664 .await;
4665 self.forget_session_connection(&connection_id);
4666 register_auto_connect_admission_rejection(
4667 &remote_device_id,
4668 now_millis_i64(),
4669 failure_count,
4670 failure_backoff_until,
4671 );
4672 eprintln!(
4673 "[pluto-rtc][auto-connect][session-admission] live transport admission rejected connection_id={} remote_device_id={} remote_node_id={} token_fp={} error={}",
4674 connection_id,
4675 remote_device_id,
4676 node_id_str,
4677 super::core_impl::log_fingerprint(token.as_str()),
4678 error
4679 );
4680 return;
4681 }
4682
4683 register_auto_connect_failure(
4684 &remote_device_id,
4685 now_millis_i64(),
4686 failure_count,
4687 failure_backoff_until,
4688 );
4689 let failures =
4690 failure_count.get(&remote_device_id).copied().unwrap_or(1);
4691 match retryable_session_admission_failure_decision(
4692 failures,
4693 connected_health_failure_threshold,
4694 current_transport_stable_id,
4695 active_transport_stable_id,
4696 runtime_instance_changed,
4697 ) {
4698 RetryableSessionAdmissionFailureDecision::PreserveCurrentTransport => {
4699 eprintln!(
4700 "[pluto-rtc][auto-connect][session-admission] transient presentation failure; preserving current generation for one bounded retry connection_id={} remote_device_id={} remote_node_id={} transport_stable_id={:?} failures={} threshold={} token_fp={} error={}",
4701 connection_id,
4702 remote_device_id,
4703 node_id_str,
4704 current_transport_stable_id,
4705 failures,
4706 connected_health_failure_threshold,
4707 super::core_impl::log_fingerprint(token.as_str()),
4708 error
4709 );
4710 }
4711 RetryableSessionAdmissionFailureDecision::RetireCurrentTransport => {
4712 eprintln!(
4713 "[pluto-rtc][auto-connect][session-admission] repeated transient presentation failure; retiring unresponsive generation for canonical reconnect connection_id={} remote_device_id={} remote_node_id={} transport_stable_id={:?} failures={} threshold={} token_fp={} error={}",
4714 connection_id,
4715 remote_device_id,
4716 node_id_str,
4717 current_transport_stable_id,
4718 failures,
4719 connected_health_failure_threshold,
4720 super::core_impl::log_fingerprint(token.as_str()),
4721 error
4722 );
4723 let _ = self
4724 .disconnect_with_reason(
4725 endpoint_id,
4726 crate::lifecycle_reason::REASON_STALE_ACTIVE_CONNECTION_RECONNECT,
4727 )
4728 .await;
4729 connected_health_checked_at.remove(&remote_device_id);
4730 connected_health_state.remove(&remote_device_id);
4731 connected_health_failures.remove(&remote_device_id);
4732 }
4733 RetryableSessionAdmissionFailureDecision::ReplacementAlreadyWon => {
4734 eprintln!(
4735 "[pluto-rtc][auto-connect][session-admission] transient failure belongs to retired generation; replacement preserved connection_id={} remote_device_id={} attempted_transport_stable_id={:?} active_transport_stable_id={:?}",
4736 connection_id,
4737 remote_device_id,
4738 current_transport_stable_id,
4739 active_transport_stable_id
4740 );
4741 }
4742 }
4743 return;
4744 }
4745 }
4746 }
4747
4748 #[cfg(not(target_arch = "wasm32"))]
4749 {
4750 if self
4751 .promote_known_native_user_device_connection(&connection_id, &node_id_str)
4752 .await
4753 .is_none()
4754 {
4755 let _ = self
4756 .confirm_managed_connection_readiness(&connection_id)
4757 .await;
4758 }
4759 }
4766 #[cfg(target_arch = "wasm32")]
4767 {
4768 let _ = self
4769 .confirm_managed_connection_readiness(&connection_id)
4770 .await;
4771 self.emit_current_wasm_connection_state(&connection_id)
4772 .await;
4773 }
4774 clear_auto_connect_failure_state(
4775 &remote_device_id,
4776 failure_count,
4777 failure_backoff_until,
4778 );
4779 non_initiator_escalation_count.remove(&remote_device_id);
4780 log_skip("already-connected");
4781 return;
4782 }
4783 } else {
4784 connected_health_checked_at.remove(&remote_device_id);
4785 connected_health_state.remove(&remote_device_id);
4786 connected_health_failures.remove(&remote_device_id);
4787 }
4788
4789 if let Some(backoff_until) = failure_backoff_until.get(&remote_device_id).copied() {
4790 if now < backoff_until {
4791 log_skip("failure-backoff");
4792 return;
4793 }
4794 failure_backoff_until.remove(&remote_device_id);
4795 }
4796
4797 let last = last_attempt_at.get(&remote_device_id).cloned().unwrap_or(0);
4798 if now - last < retry_throttle_ms {
4799 log_skip("retry-throttle");
4800 return;
4801 }
4802 last_attempt_at.insert(remote_device_id.clone(), now);
4803 let connect_mode = "ticket-endpoint-addr";
4804
4805 #[cfg(not(target_arch = "wasm32"))]
4806 {
4807 self.connection_manager
4808 .upsert_pending(
4809 connection_id.clone(),
4810 Some(node_id_str.clone()),
4811 Some(remote_device_id.clone()),
4812 Some(node_id_str.clone()),
4813 )
4814 .await;
4815 self.connection_manager.set_connecting(&connection_id).await;
4816 let dial_attempt_fence = self
4817 .connection_manager
4818 .get_by_connection_id(&connection_id)
4819 .await
4820 .map(|record| (record.transport_generation, record.transport_stable_id));
4821
4822 eprintln!(
4823 "[pluto-rtc][auto-connect] attempt remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} mode={}",
4824 remote_device_id,
4825 local_device_id,
4826 local_node_id,
4827 node_id_str,
4828 connect_mode
4829 );
4830
4831 eprintln!("[pluto-rtc] auto-connecting to {}", remote_device_id);
4832 let result = self.ensure_connected_addr(endpoint_id, endpoint_addr).await;
4833
4834 match result {
4835 Ok(()) => {
4836 if self.is_auto_connect_peer_excluded(&remote_device_id, Some(&node_id_str)) {
4837 log_skip("auto-connect-excluded-after-dial");
4838 let _ = self
4839 .disconnect_with_reason(
4840 endpoint_id,
4841 crate::lifecycle_reason::REASON_MANUAL_DISCONNECT,
4842 )
4843 .await;
4844 self.connection_manager
4845 .set_closed(
4846 &connection_id,
4847 Some(crate::lifecycle_reason::REASON_MANUAL_DISCONNECT.to_string()),
4848 )
4849 .await;
4850 return;
4851 }
4852 let active_probe = self.validate_active_connection(endpoint_id).await;
4853 let active_probe_stable =
4854 active_probe.as_ref().is_some_and(|probe| probe.responsive);
4855 let transport_alive = self.is_connection_transport_alive(endpoint_id).await;
4856 let probe_decision =
4857 active_connection_probe_decision(active_probe_stable, transport_alive);
4858 let stable = matches!(
4859 probe_decision,
4860 ActiveConnectionProbeDecision::Healthy
4861 | ActiveConnectionProbeDecision::PreserveLiveTransport
4862 );
4863 if !stable {
4864 if self.is_app_execution_suspended() {
4865 log_skip("backgrounded-after-stability-probe");
4866 return;
4867 }
4868 self.connection_manager
4869 .set_failed(
4870 &connection_id,
4871 Some("connection failed stability verification".to_string()),
4872 )
4873 .await;
4874 if let Some(transport_stable_id) =
4875 active_probe.as_ref().map(|probe| probe.transport_stable_id)
4876 {
4877 let _ = self
4878 .disconnect_transport_generation_with_reason(
4879 endpoint_id,
4880 transport_stable_id,
4881 crate::lifecycle_reason::REASON_CONNECTION_FAILED_STABILITY,
4882 )
4883 .await;
4884 }
4885 self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
4886 .await;
4887 register_auto_connect_failure(
4888 &remote_device_id,
4889 now_millis_i64(),
4890 failure_count,
4891 failure_backoff_until,
4892 );
4893 let current_failures =
4894 failure_count.get(&remote_device_id).copied().unwrap_or(0);
4895 if current_failures >= network_change_failure_threshold
4896 && self
4897 .should_force_network_change_after_connect_failures(
4898 &remote_device_id,
4899 )
4900 .await
4901 {
4902 self.maybe_force_network_change_for_auto_connect(
4903 user_id,
4904 local_device_id,
4905 "unstable-connection",
4906 last_network_change_at_ms,
4907 network_change_recovery_interval_ms,
4908 last_presence_republish_at_ms,
4909 presence_republish_interval_ms,
4910 )
4911 .await;
4912 }
4913 eprintln!(
4914 "[pluto-rtc][auto-connect] unstable connection suppressed remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} connection_id={}",
4915 remote_device_id,
4916 local_device_id,
4917 local_node_id,
4918 node_id_str,
4919 connection_id
4920 );
4921 return;
4922 }
4923 if matches!(
4924 probe_decision,
4925 ActiveConnectionProbeDecision::PreserveLiveTransport
4926 ) {
4927 log_skip("stability-probe-missed-preserving-live-transport");
4928 }
4929 let current_transport_stable_id = self
4930 .get_connection(endpoint_id)
4931 .await
4932 .map(|connection| crate::transport_generation::for_connection(&connection));
4933 self.commit_current_native_transport_record(
4934 &connection_id,
4935 &node_id_str,
4936 Some(remote_device_id.clone()),
4937 current_transport_stable_id,
4938 Some(connect_mode.to_string()),
4939 )
4940 .await;
4941
4942 if let Some(ref token) = extracted_token {
4943 let bilateral_user_device = extracted_token_payload
4944 .as_ref()
4945 .is_some_and(|payload| payload.scope.as_str() == "user-device");
4946 let has_remote_admission_proof =
4947 current_transport_stable_id.is_some_and(|transport_stable_id| {
4948 self.remote_session_token_admitted_for_transport(
4949 &connection_id,
4950 transport_stable_id,
4951 )
4952 });
4953 let has_current_inbound_admission_proof = current_transport_stable_id
4954 .is_some_and(|transport_stable_id| {
4955 self.inbound_session_token_admitted_for_transport(
4956 &connection_id,
4957 transport_stable_id,
4958 )
4959 });
4960 let peer_requested_reciprocal =
4961 self.has_pending_reciprocal_session_admission_request(&connection_id);
4962 let presentation_decision = if bilateral_user_device {
4963 session_admission_presentation_decision(
4964 true,
4965 has_remote_admission_proof,
4966 has_current_inbound_admission_proof,
4967 peer_requested_reciprocal,
4968 local_node_id > node_id_str,
4969 )
4970 } else {
4971 SessionAdmissionPresentationDecision {
4972 should_present: !has_remote_admission_proof,
4973 request_reciprocal: false,
4974 }
4975 };
4976 if !presentation_decision.should_present {
4977 eprintln!(
4978 "[pluto-rtc][auto-connect][session-admission] awaiting elected presenter on fresh transport connection_id={} remote_device_id={} remote_node_id={} local_node_id={}",
4979 connection_id,
4980 remote_device_id,
4981 node_id_str,
4982 local_node_id,
4983 );
4984 return;
4985 }
4986 match self
4987 .present_and_accept_session_token_for_route_repair(
4988 endpoint_id,
4989 &connection_id,
4990 token,
4991 extracted_token_suffix.as_deref(),
4992 Some(remote_device_id.clone()),
4993 Some(local_device_id.to_string()),
4994 has_remote_admission_proof,
4995 presentation_decision.request_reciprocal,
4996 )
4997 .await
4998 {
4999 Ok(approval_scope) => {
5000 eprintln!(
5001 "[pluto-rtc][auto-connect] session-token presented and approved connection_id={} scope={} device_id={} token_fp={} local_admission=accepted",
5002 connection_id,
5003 approval_scope,
5004 remote_device_id,
5005 super::core_impl::log_fingerprint(token.as_str())
5006 );
5007 }
5008 Err(error) => {
5009 if session_token_presentation_failure_action(&error)
5010 == SessionTokenPresentationFailureAction::RetryOnNextTransport
5011 {
5012 register_auto_connect_failure(
5013 &remote_device_id,
5014 now_millis_i64(),
5015 failure_count,
5016 failure_backoff_until,
5017 );
5018 eprintln!(
5019 "[pluto-rtc][auto-connect] transient session-token presentation failure; preserving peer session for canonical transport retry connection_id={} device_id={} token_fp={} error={}",
5020 connection_id,
5021 remote_device_id,
5022 super::core_impl::log_fingerprint(token.as_str()),
5023 error
5024 );
5025 return;
5026 }
5027 let rejection_reason =
5028 format!("session-token-presentation-failed: {}", error);
5029 self.reject_session_connection(
5030 &connection_id,
5031 rejection_reason.as_str(),
5032 );
5033 self.connection_manager
5034 .set_failed(&connection_id, Some(rejection_reason.clone()))
5035 .await;
5036 let _ = self
5037 .disconnect_with_reason(
5038 endpoint_id,
5039 crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
5040 )
5041 .await;
5042 self.reconcile_authoritative_device_node(
5043 &remote_device_id,
5044 &node_id_str,
5045 )
5046 .await;
5047 self.forget_session_connection(&connection_id);
5052 register_auto_connect_admission_rejection(
5053 &remote_device_id,
5054 now_millis_i64(),
5055 failure_count,
5056 failure_backoff_until,
5057 );
5058 eprintln!(
5059 "[pluto-rtc][auto-connect] session-token presentation failed connection_id={} device_id={} token_fp={} error={}",
5060 connection_id,
5061 remote_device_id,
5062 super::core_impl::log_fingerprint(token.as_str()),
5063 error
5064 );
5065 return;
5066 }
5067 }
5068 }
5069
5070 let _ = self
5071 .confirm_managed_connection_readiness(&connection_id)
5072 .await;
5073 if let Some(outgoing_conn) = self.get_connection(endpoint_id).await {
5078 self.spawn_iroh_path_watcher(
5082 &connection_id,
5083 &node_id_str,
5084 &outgoing_conn,
5085 false,
5086 );
5087 }
5088 clear_auto_connect_failure_state(
5089 &remote_device_id,
5090 failure_count,
5091 failure_backoff_until,
5092 );
5093 non_initiator_escalation_count.remove(&remote_device_id);
5094 eprintln!(
5095 "[pluto-rtc][auto-connect] connected remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} connection_id={}",
5096 remote_device_id,
5097 local_device_id,
5098 local_node_id,
5099 node_id_str,
5100 connection_id
5101 );
5102 }
5103 Err(e) => {
5104 if let Some((transport_generation, transport_stable_id)) = dial_attempt_fence {
5105 let failed = self
5106 .connection_manager
5107 .set_failed_if_dial_attempt_current(
5108 &connection_id,
5109 transport_generation,
5110 transport_stable_id,
5111 Some(
5112 crate::lifecycle_reason::REASON_AUTO_CONNECT_FAILURE
5113 .to_string(),
5114 ),
5115 )
5116 .await;
5117 if failed.is_none() {
5118 eprintln!(
5119 "[pluto-rtc][auto-connect] ignored stale dial failure connection_id={} remote_device_id={} remote_node_id={} expected_transport_generation={} expected_transport_stable_id={:?}",
5120 connection_id,
5121 remote_device_id,
5122 node_id_str,
5123 transport_generation,
5124 transport_stable_id,
5125 );
5126 return;
5131 }
5132 }
5133 register_auto_connect_failure(
5134 &remote_device_id,
5135 now_millis_i64(),
5136 failure_count,
5137 failure_backoff_until,
5138 );
5139 let current_failures =
5140 failure_count.get(&remote_device_id).copied().unwrap_or(0);
5141 if current_failures >= network_change_failure_threshold
5142 && self
5143 .should_force_network_change_after_connect_failures(&remote_device_id)
5144 .await
5145 {
5146 let _ = self
5147 .disconnect_with_reason(
5148 endpoint_id,
5149 crate::lifecycle_reason::REASON_NETWORK_CHANGE_RECONNECT,
5150 )
5151 .await;
5152 self.reconcile_authoritative_device_node(&remote_device_id, &node_id_str)
5153 .await;
5154 self.maybe_force_network_change_for_auto_connect(
5155 user_id,
5156 local_device_id,
5157 "repeated-connect-failures",
5158 last_network_change_at_ms,
5159 network_change_recovery_interval_ms,
5160 last_presence_republish_at_ms,
5161 presence_republish_interval_ms,
5162 )
5163 .await;
5164 }
5165 self.maybe_refresh_gateway_presence_for_auto_connect(
5166 user_id,
5167 local_device_id,
5168 "connect-failed",
5169 last_presence_republish_at_ms,
5170 presence_republish_interval_ms,
5171 )
5172 .await;
5173 eprintln!(
5174 "[pluto-rtc] auto-connect failed to {}: {}",
5175 remote_device_id, e
5176 );
5177 }
5178 }
5179 }
5180
5181 #[cfg(target_arch = "wasm32")]
5182 {
5183 self.connection_manager
5184 .upsert_pending(
5185 connection_id.clone(),
5186 Some(node_id_str.clone()),
5187 Some(remote_device_id.clone()),
5188 Some(node_id_str.clone()),
5189 )
5190 .await;
5191 self.connection_manager
5195 .set_device_id(&connection_id, remote_device_id.clone())
5196 .await;
5197 self.connection_manager.set_connecting(&connection_id).await;
5198 self.emit_current_wasm_connection_state(&connection_id)
5199 .await;
5200 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5201 "[pluto-rtc][auto-connect] attempt remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} mode={}",
5202 remote_device_id,
5203 local_device_id,
5204 local_node_id,
5205 node_id_str,
5206 connect_mode
5207 )));
5208 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5209 "[pluto-rtc] auto-connecting to {}",
5210 remote_device_id
5211 )));
5212 let result = self.ensure_connected_addr(endpoint_id, endpoint_addr).await;
5213
5214 match result {
5215 Ok(()) => {
5216 let active_probe_stable = self.validate_active_connection(endpoint_id).await;
5217 let transport_alive = self.is_connection_transport_alive(endpoint_id).await;
5218 let probe_decision =
5219 active_connection_probe_decision(active_probe_stable, transport_alive);
5220 let stable = matches!(
5221 probe_decision,
5222 ActiveConnectionProbeDecision::Healthy
5223 | ActiveConnectionProbeDecision::PreserveLiveTransport
5224 );
5225 if !stable {
5226 if self.is_app_execution_suspended() {
5227 log_skip("backgrounded-after-stability-probe");
5228 return;
5229 }
5230 self.connection_manager
5231 .upsert_pending(
5232 connection_id.clone(),
5233 Some(node_id_str.clone()),
5234 Some(remote_device_id.clone()),
5235 Some(node_id_str.clone()),
5236 )
5237 .await;
5238 self.connection_manager
5239 .set_failed(
5240 &connection_id,
5241 Some("connection failed stability verification".to_string()),
5242 )
5243 .await;
5244 self.emit_current_wasm_connection_state(&connection_id)
5245 .await;
5246 register_auto_connect_failure(
5247 &remote_device_id,
5248 now_millis_i64(),
5249 failure_count,
5250 failure_backoff_until,
5251 );
5252 let _ = self
5253 .disconnect_with_reason(
5254 endpoint_id,
5255 crate::lifecycle_reason::REASON_AUTO_CONNECT_FAILURE,
5256 )
5257 .await;
5258 web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
5259 "[pluto-rtc][auto-connect] unstable connection suppressed remote_device_id={} local_device_id={} local_node_id={} remote_node_id={}",
5260 remote_device_id,
5261 local_device_id,
5262 local_node_id,
5263 node_id_str
5264 )));
5265 return;
5266 }
5267 if matches!(
5268 probe_decision,
5269 ActiveConnectionProbeDecision::PreserveLiveTransport
5270 ) {
5271 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5272 "[pluto-rtc][auto-connect] stability probe missed but transport remains alive; preserving connection remote_device_id={} local_device_id={} local_node_id={} remote_node_id={} connection_id={}",
5273 remote_device_id,
5274 local_device_id,
5275 local_node_id,
5276 node_id_str,
5277 connection_id
5278 )));
5279 }
5280
5281 self.connection_manager
5282 .upsert_pending(
5283 connection_id.clone(),
5284 Some(node_id_str.clone()),
5285 Some(remote_device_id.clone()),
5286 Some(node_id_str.clone()),
5287 )
5288 .await;
5289 let pending_record = self
5290 .connection_manager
5291 .get_by_connection_id(&connection_id)
5292 .await;
5293 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5294 "[pluto-rtc][auto-connect][wasm] upsert_pending remote_device_id={} remote_node_id={} connection_id={} pending_record_exists={} record={:?}",
5295 remote_device_id,
5296 node_id_str,
5297 connection_id,
5298 pending_record.is_some(),
5299 pending_record,
5300 )));
5301 let transport_stable_id = self
5302 .get_connection(endpoint_id)
5303 .await
5304 .map(|connection| crate::transport_generation::for_connection(&connection));
5305 self.commit_current_native_transport_record(
5306 &connection_id,
5307 &node_id_str,
5308 Some(remote_device_id.clone()),
5309 transport_stable_id,
5310 Some(connect_mode.to_string()),
5311 )
5312 .await;
5313 let active_transport =
5314 self.iroh_path_kind(&node_id_str).await.transport_label();
5315 let _ = self
5316 .report_transport_status_for_current_generation(
5317 &connection_id,
5318 active_transport,
5319 None,
5320 )
5321 .await;
5322
5323 if let Some(ref token) = extracted_token {
5324 match self
5325 .present_and_accept_session_token(
5326 endpoint_id,
5327 &connection_id,
5328 token,
5329 extracted_token_suffix.as_deref(),
5330 Some(remote_device_id.clone()),
5331 )
5332 .await
5333 {
5334 Ok(approval_scope) => {
5335 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5336 "[pluto-rtc][auto-connect][wasm] session-token presented and approved connection_id={} scope={} device_id={} token_fp={} local_admission=accepted",
5337 connection_id,
5338 approval_scope,
5339 remote_device_id,
5340 super::core_impl::log_fingerprint(token.as_str())
5341 )));
5342 }
5343 Err(error) => {
5344 if session_token_presentation_failure_action(&error)
5345 == SessionTokenPresentationFailureAction::RetryOnNextTransport
5346 {
5347 register_auto_connect_failure(
5348 &remote_device_id,
5349 now_millis_i64(),
5350 failure_count,
5351 failure_backoff_until,
5352 );
5353 web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
5354 "[pluto-rtc][auto-connect][wasm] transient session-token presentation failure; preserving peer session for canonical transport retry connection_id={} device_id={} token_fp={} error={}",
5355 connection_id,
5356 remote_device_id,
5357 super::core_impl::log_fingerprint(token.as_str()),
5358 error
5359 )));
5360 return;
5361 }
5362 let rejection_reason =
5363 format!("session-token-presentation-failed: {}", error);
5364 self.reject_session_connection(
5365 &connection_id,
5366 rejection_reason.as_str(),
5367 );
5368 self.connection_manager
5369 .set_failed(&connection_id, Some(rejection_reason))
5370 .await;
5371 self.emit_current_wasm_connection_state(&connection_id)
5372 .await;
5373 let _ = self
5374 .disconnect_with_reason(
5375 endpoint_id,
5376 crate::lifecycle_reason::REASON_SESSION_ADMISSION_REJECTED,
5377 )
5378 .await;
5379 self.forget_session_connection(&connection_id);
5383 register_auto_connect_admission_rejection(
5384 &remote_device_id,
5385 now_millis_i64(),
5386 failure_count,
5387 failure_backoff_until,
5388 );
5389 web_sys::console::warn_1(&wasm_bindgen::JsValue::from_str(&format!(
5390 "[pluto-rtc][auto-connect][wasm] session-token presentation failed connection_id={} device_id={} token_fp={} error={}",
5391 connection_id,
5392 remote_device_id,
5393 super::core_impl::log_fingerprint(token.as_str()),
5394 error
5395 )));
5396 return;
5397 }
5398 }
5399 }
5400
5401 let _ = self
5402 .confirm_managed_connection_readiness(&connection_id)
5403 .await;
5404 let connected_record = self
5405 .connection_manager
5406 .get_by_connection_id(&connection_id)
5407 .await;
5408 let record_count = self.connection_manager.list_all().await.len();
5409 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5410 "[pluto-rtc][auto-connect][wasm] set_connected remote_device_id={} remote_node_id={} connection_id={} connected_record_exists={} total_records={} record={:?}",
5411 remote_device_id,
5412 node_id_str,
5413 connection_id,
5414 connected_record.is_some(),
5415 record_count,
5416 connected_record,
5417 )));
5418 self.emit_current_wasm_connection_state(&connection_id)
5419 .await;
5420 clear_auto_connect_failure_state(
5421 &remote_device_id,
5422 failure_count,
5423 failure_backoff_until,
5424 );
5425 non_initiator_escalation_count.remove(&remote_device_id);
5426 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5427 "[pluto-rtc][auto-connect] connected remote_device_id={} local_device_id={} local_node_id={} remote_node_id={}",
5428 remote_device_id,
5429 local_device_id,
5430 local_node_id,
5431 node_id_str
5432 )));
5433 }
5434 Err(e) => {
5435 let err = e.to_string();
5436 self.connection_manager
5437 .upsert_pending(
5438 connection_id.clone(),
5439 Some(node_id_str.clone()),
5440 Some(remote_device_id.clone()),
5441 Some(node_id_str.clone()),
5442 )
5443 .await;
5444 self.connection_manager
5445 .set_failed(&connection_id, Some(err.clone()))
5446 .await;
5447 self.emit_current_wasm_connection_state(&connection_id)
5448 .await;
5449 register_auto_connect_failure(
5450 &remote_device_id,
5451 now_millis_i64(),
5452 failure_count,
5453 failure_backoff_until,
5454 );
5455 web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
5456 "[pluto-rtc] auto-connect failed to {}: {}",
5457 remote_device_id, err
5458 )));
5459 }
5460 }
5461 }
5462 }
5463
5464 #[cfg(not(target_arch = "wasm32"))]
5465 async fn run_auto_connect_snapshot(
5466 self: &Arc<Self>,
5467 user_id: &str,
5468 local_device_id: &str,
5469 last_attempt_at: &mut std::collections::HashMap<String, i64>,
5470 failure_count: &mut std::collections::HashMap<String, u8>,
5471 failure_backoff_until: &mut std::collections::HashMap<String, i64>,
5472 retry_throttle_ms: i64,
5473 non_initiator_wait_started_at: &mut std::collections::HashMap<String, i64>,
5474 non_initiator_escalation_count: &mut std::collections::HashMap<String, u8>,
5475 initiator_grace_ms: i64,
5476 skip_log_at: &mut std::collections::HashMap<String, i64>,
5477 connected_health_checked_at: &mut std::collections::HashMap<String, i64>,
5478 connected_health_state: &mut std::collections::HashMap<String, bool>,
5479 connected_health_failures: &mut std::collections::HashMap<String, u8>,
5480 connected_health_probe_ms: i64,
5481 connected_health_failure_threshold: u8,
5482 last_presence_republish_at_ms: &mut i64,
5483 presence_republish_interval_ms: i64,
5484 last_network_change_at_ms: &mut i64,
5485 network_change_recovery_interval_ms: i64,
5486 network_change_failure_threshold: u8,
5487 last_snapshot_signature: &mut Option<(usize, usize, usize)>,
5488 last_snapshot_log_at: &mut i64,
5489 last_known_node_id: &mut std::collections::HashMap<String, String>,
5490 known_devices: &mut std::collections::HashMap<String, crate::signaling::Device>,
5491 ) {
5492 if self.is_app_execution_suspended() {
5493 return;
5494 }
5495
5496 let now = now_millis_i64();
5497 match self.search_devices(user_id).await {
5498 Ok(devices) => {
5499 let total = devices.len();
5500 let online = devices.iter().filter(|d| d.online).count();
5501 let remote_online = devices
5502 .iter()
5503 .filter(|d| d.online && d.device_id != local_device_id)
5504 .count();
5505 let signature = (total, online, remote_online);
5506 let changed = last_snapshot_signature
5507 .map(|prev| prev != signature)
5508 .unwrap_or(true);
5509 let should_log = if auto_connect_verbose() {
5510 now.saturating_sub(*last_snapshot_log_at) >= 2_000
5511 } else {
5512 changed || now.saturating_sub(*last_snapshot_log_at) >= 30_000
5513 };
5514 if should_log {
5515 *last_snapshot_log_at = now;
5516 *last_snapshot_signature = Some(signature);
5517 #[cfg(not(target_arch = "wasm32"))]
5518 eprintln!(
5519 "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} total_devices={} online_devices={} remote_online_devices={}",
5520 user_id, local_device_id, total, online, remote_online
5521 );
5522 #[cfg(target_arch = "wasm32")]
5523 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5524 "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} total_devices={} online_devices={} remote_online_devices={}",
5525 user_id, local_device_id, total, online, remote_online
5526 )));
5527 }
5528
5529 let known_device_ids: std::collections::HashSet<String> =
5530 devices.iter().map(|d| d.device_id.clone()).collect();
5531
5532 for device in devices {
5533 known_devices.insert(device.device_id.clone(), device.clone());
5534 self.try_auto_connect_device(
5535 user_id,
5536 device,
5537 local_device_id,
5538 last_attempt_at,
5539 failure_count,
5540 failure_backoff_until,
5541 retry_throttle_ms,
5542 non_initiator_wait_started_at,
5543 non_initiator_escalation_count,
5544 initiator_grace_ms,
5545 skip_log_at,
5546 connected_health_checked_at,
5547 connected_health_state,
5548 connected_health_failures,
5549 connected_health_probe_ms,
5550 connected_health_failure_threshold,
5551 last_presence_republish_at_ms,
5552 presence_republish_interval_ms,
5553 last_network_change_at_ms,
5554 network_change_recovery_interval_ms,
5555 network_change_failure_threshold,
5556 last_known_node_id,
5557 )
5558 .await;
5559 }
5560
5561 last_attempt_at.retain(|k, _| known_device_ids.contains(k));
5565 failure_count.retain(|k, _| known_device_ids.contains(k));
5566 failure_backoff_until.retain(|k, _| known_device_ids.contains(k));
5567 non_initiator_wait_started_at.retain(|k, _| known_device_ids.contains(k));
5568 non_initiator_escalation_count.retain(|k, _| known_device_ids.contains(k));
5569 connected_health_checked_at.retain(|k, _| known_device_ids.contains(k));
5570 connected_health_state.retain(|k, _| known_device_ids.contains(k));
5571 connected_health_failures.retain(|k, _| known_device_ids.contains(k));
5572 last_known_node_id.retain(|key, _| {
5573 let device_id = key
5574 .strip_suffix("::runtime-instance")
5575 .or_else(|| key.strip_suffix("::ticket"))
5576 .unwrap_or(key.as_str());
5577 known_device_ids.contains(device_id)
5578 });
5579 skip_log_at.retain(|k, _| {
5581 k.split_once("::")
5582 .map_or(false, |(id, _)| known_device_ids.contains(id))
5583 });
5584 }
5585 Err(error) => {
5586 let min_interval = if auto_connect_verbose() {
5587 2_000
5588 } else {
5589 30_000
5590 };
5591 if now.saturating_sub(*last_snapshot_log_at) >= min_interval {
5592 *last_snapshot_log_at = now;
5593 #[cfg(not(target_arch = "wasm32"))]
5594 eprintln!(
5595 "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} search_devices_error={}",
5596 user_id, local_device_id, error
5597 );
5598 #[cfg(target_arch = "wasm32")]
5599 web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
5600 "[pluto-rtc][auto-connect][snapshot] user_id={} local_device_id={} search_devices_error={}",
5601 user_id, local_device_id, error
5602 )));
5603 }
5604 }
5605 }
5606 }
5607
5608 #[cfg(not(target_arch = "wasm32"))]
5609 pub(crate) async fn auto_connect_loop(
5610 self: Arc<Self>,
5611 user_id: String,
5612 local_device_id: String,
5613 generation: u64,
5614 ) {
5615 use futures::StreamExt;
5616 let mut last_attempt_at: std::collections::HashMap<String, i64> =
5617 std::collections::HashMap::new();
5618 let mut non_initiator_wait_started_at: std::collections::HashMap<String, i64> =
5619 std::collections::HashMap::new();
5620 let mut non_initiator_escalation_count: std::collections::HashMap<String, u8> =
5621 std::collections::HashMap::new();
5622 let mut skip_log_at: std::collections::HashMap<String, i64> =
5623 std::collections::HashMap::new();
5624 let mut connected_health_checked_at: std::collections::HashMap<String, i64> =
5625 std::collections::HashMap::new();
5626 let mut connected_health_state: std::collections::HashMap<String, bool> =
5627 std::collections::HashMap::new();
5628 let mut connected_health_failures: std::collections::HashMap<String, u8> =
5629 std::collections::HashMap::new();
5630 let mut failure_count: std::collections::HashMap<String, u8> =
5631 std::collections::HashMap::new();
5632 let mut failure_backoff_until: std::collections::HashMap<String, i64> =
5633 std::collections::HashMap::new();
5634 let mut last_known_node_id: std::collections::HashMap<String, String> =
5635 std::collections::HashMap::new();
5636 let mut known_devices: std::collections::HashMap<String, crate::signaling::Device> =
5641 std::collections::HashMap::new();
5642 let mut last_snapshot_signature: Option<(usize, usize, usize)> = None;
5643 let mut last_snapshot_log_at = 0_i64;
5644 let mut last_presence_republish_at_ms = 0_i64;
5645 let retry_throttle_ms = 1_000;
5646 let rescan_interval_ms_fg: u64 = 300_000;
5650 let rescan_interval_ms_bg: u64 = 300_000;
5651 let initiator_grace_ms = 3_000;
5652 let connected_health_probe_ms_fg = 5_000_i64;
5655 let connected_health_probe_ms_bg = i64::MAX;
5656 let connected_health_failure_threshold = 2;
5657 let presence_republish_interval_ms_fg = 10_000_i64;
5660 let presence_republish_interval_ms_bg = i64::MAX;
5661 let mut last_network_change_at_ms = 0_i64;
5662 let network_change_recovery_interval_ms_fg = 15_000_i64;
5664 let network_change_recovery_interval_ms_bg = i64::MAX;
5665 let network_change_failure_threshold = auto_connect_network_change_failure_threshold();
5666
5667 loop {
5668 if !self.is_auto_connect_generation_current(generation) {
5669 #[cfg(not(target_arch = "wasm32"))]
5670 eprintln!(
5671 "[pluto-rtc][auto-connect] stopping stale loop user_id={} local_device_id={} generation={}",
5672 user_id, local_device_id, generation
5673 );
5674 #[cfg(target_arch = "wasm32")]
5675 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5676 "[pluto-rtc][auto-connect] stopping stale loop user_id={} local_device_id={} generation={}",
5677 user_id, local_device_id, generation
5678 )));
5679 break;
5680 }
5681
5682 if !self.is_app_execution_suspended() {
5685 self.run_auto_connect_snapshot(
5686 &user_id,
5687 &local_device_id,
5688 &mut last_attempt_at,
5689 &mut failure_count,
5690 &mut failure_backoff_until,
5691 retry_throttle_ms,
5692 &mut non_initiator_wait_started_at,
5693 &mut non_initiator_escalation_count,
5694 initiator_grace_ms,
5695 &mut skip_log_at,
5696 &mut connected_health_checked_at,
5697 &mut connected_health_state,
5698 &mut connected_health_failures,
5699 connected_health_probe_ms_fg,
5700 connected_health_failure_threshold,
5701 &mut last_presence_republish_at_ms,
5702 presence_republish_interval_ms_fg,
5703 &mut last_network_change_at_ms,
5704 network_change_recovery_interval_ms_fg,
5705 network_change_failure_threshold,
5706 &mut last_snapshot_signature,
5707 &mut last_snapshot_log_at,
5708 &mut last_known_node_id,
5709 &mut known_devices,
5710 )
5711 .await;
5712 }
5713
5714 #[allow(unused_mut)]
5715 let mut stream = match self.subscribe_devices(&user_id).await {
5716 Ok(s) => {
5717 #[cfg(not(target_arch = "wasm32"))]
5718 eprintln!(
5719 "[pluto-rtc][auto-connect] subscribed user_id={} local_device_id={}",
5720 user_id, local_device_id
5721 );
5722 #[cfg(target_arch = "wasm32")]
5723 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
5724 "[pluto-rtc][auto-connect] subscribed user_id={} local_device_id={}",
5725 user_id, local_device_id
5726 )));
5727 s
5728 }
5729 Err(e) => {
5730 #[cfg(not(target_arch = "wasm32"))]
5731 eprintln!(
5732 "[pluto-rtc][auto-connect] subscribe failed user_id={} local_device_id={} error={}",
5733 user_id, local_device_id, e
5734 );
5735 #[cfg(target_arch = "wasm32")]
5736 web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(&format!(
5737 "[pluto-rtc][auto-connect] subscribe failed user_id={} local_device_id={} error={}",
5738 user_id, local_device_id, e
5739 )));
5740
5741 #[cfg(not(target_arch = "wasm32"))]
5742 tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
5743 #[cfg(target_arch = "wasm32")]
5744 gloo_timers::future::sleep(std::time::Duration::from_millis(1500)).await;
5745 continue;
5746 }
5747 };
5748
5749 #[cfg(not(target_arch = "wasm32"))]
5750 {
5751 let mut last_bg_state = self.is_app_execution_suspended();
5752 let mut rescan_interval =
5753 tokio::time::interval(std::time::Duration::from_millis(if last_bg_state {
5754 rescan_interval_ms_bg
5755 } else {
5756 rescan_interval_ms_fg
5757 }));
5758 let mut health_interval = tokio::time::interval(std::time::Duration::from_millis(
5759 connected_health_probe_ms_fg as u64,
5760 ));
5761 health_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
5762 loop {
5763 if !self.is_auto_connect_generation_current(generation) {
5764 break;
5765 }
5766 let bg = self.is_app_execution_suspended();
5770 if bg != last_bg_state {
5771 last_bg_state = bg;
5772 rescan_interval =
5773 tokio::time::interval(std::time::Duration::from_millis(if bg {
5774 rescan_interval_ms_bg
5775 } else {
5776 rescan_interval_ms_fg
5777 }));
5778 }
5779 let connected_health_probe_ms = if bg {
5780 connected_health_probe_ms_bg
5781 } else {
5782 connected_health_probe_ms_fg
5783 };
5784 let presence_republish_interval_ms = if bg {
5785 presence_republish_interval_ms_bg
5786 } else {
5787 presence_republish_interval_ms_fg
5788 };
5789 let network_change_recovery_interval_ms = if bg {
5790 network_change_recovery_interval_ms_bg
5791 } else {
5792 network_change_recovery_interval_ms_fg
5793 };
5794 let grace_wakeup_delay_ms = next_non_initiator_wakeup_delay_ms(
5800 &non_initiator_wait_started_at,
5801 &non_initiator_escalation_count,
5802 now_millis_i64(),
5803 );
5804 let grace_wakeup = tokio::time::sleep(std::time::Duration::from_millis(
5805 grace_wakeup_delay_ms.unwrap_or(rescan_interval_ms_fg),
5806 ));
5807 tokio::pin!(grace_wakeup);
5808 tokio::select! {
5809 _ = self.native_auto_connect_wake.notified() => {
5810 if self.is_app_execution_suspended() {
5811 continue;
5812 }
5813 self.run_auto_connect_snapshot(
5818 &user_id,
5819 &local_device_id,
5820 &mut last_attempt_at,
5821 &mut failure_count,
5822 &mut failure_backoff_until,
5823 retry_throttle_ms,
5824 &mut non_initiator_wait_started_at,
5825 &mut non_initiator_escalation_count,
5826 initiator_grace_ms,
5827 &mut skip_log_at,
5828 &mut connected_health_checked_at,
5829 &mut connected_health_state,
5830 &mut connected_health_failures,
5831 connected_health_probe_ms,
5832 connected_health_failure_threshold,
5833 &mut last_presence_republish_at_ms,
5834 presence_republish_interval_ms,
5835 &mut last_network_change_at_ms,
5836 network_change_recovery_interval_ms,
5837 network_change_failure_threshold,
5838 &mut last_snapshot_signature,
5839 &mut last_snapshot_log_at,
5840 &mut last_known_node_id,
5841 &mut known_devices,
5842 ).await;
5843 }
5844 _ = rescan_interval.tick() => {
5845 if self.is_app_execution_suspended() {
5846 continue;
5847 }
5848 self.run_auto_connect_snapshot(
5849 &user_id,
5850 &local_device_id,
5851 &mut last_attempt_at,
5852 &mut failure_count,
5853 &mut failure_backoff_until,
5854 retry_throttle_ms,
5855 &mut non_initiator_wait_started_at,
5856 &mut non_initiator_escalation_count,
5857 initiator_grace_ms,
5858 &mut skip_log_at,
5859 &mut connected_health_checked_at,
5860 &mut connected_health_state,
5861 &mut connected_health_failures,
5862 connected_health_probe_ms,
5863 connected_health_failure_threshold,
5864 &mut last_presence_republish_at_ms,
5865 presence_republish_interval_ms,
5866 &mut last_network_change_at_ms,
5867 network_change_recovery_interval_ms,
5868 network_change_failure_threshold,
5869 &mut last_snapshot_signature,
5870 &mut last_snapshot_log_at,
5871 &mut last_known_node_id,
5872 &mut known_devices,
5873 ).await;
5874 }
5875 _ = health_interval.tick() => {
5876 if self.is_app_execution_suspended() {
5877 continue;
5878 }
5879
5880 let cached_devices: Vec<crate::signaling::Device> =
5887 known_devices.values().cloned().collect();
5888 for mut device in cached_devices {
5889 if !device.online {
5890 let Some(ticket) = device.ticket.as_deref() else {
5891 continue;
5892 };
5893 let (iroh_ticket, _) =
5894 crate::session_token::split_ticket(ticket.trim());
5895 let Ok(endpoint_addr) = parse_endpoint_ticket(iroh_ticket) else {
5896 continue;
5897 };
5898 if !self.is_connected(endpoint_addr.id).await {
5899 continue;
5900 }
5901 device.online = true;
5902 }
5903 self.try_auto_connect_device(
5904 &user_id,
5905 device,
5906 &local_device_id,
5907 &mut last_attempt_at,
5908 &mut failure_count,
5909 &mut failure_backoff_until,
5910 retry_throttle_ms,
5911 &mut non_initiator_wait_started_at,
5912 &mut non_initiator_escalation_count,
5913 initiator_grace_ms,
5914 &mut skip_log_at,
5915 &mut connected_health_checked_at,
5916 &mut connected_health_state,
5917 &mut connected_health_failures,
5918 connected_health_probe_ms_fg,
5919 connected_health_failure_threshold,
5920 &mut last_presence_republish_at_ms,
5921 presence_republish_interval_ms_fg,
5922 &mut last_network_change_at_ms,
5923 network_change_recovery_interval_ms_fg,
5924 network_change_failure_threshold,
5925 &mut last_known_node_id,
5926 )
5927 .await;
5928 }
5929 }
5930 _ = &mut grace_wakeup, if grace_wakeup_delay_ms.is_some() => {
5931 if self.is_app_execution_suspended() {
5932 continue;
5933 }
5934 self.run_auto_connect_snapshot(
5935 &user_id,
5936 &local_device_id,
5937 &mut last_attempt_at,
5938 &mut failure_count,
5939 &mut failure_backoff_until,
5940 retry_throttle_ms,
5941 &mut non_initiator_wait_started_at,
5942 &mut non_initiator_escalation_count,
5943 initiator_grace_ms,
5944 &mut skip_log_at,
5945 &mut connected_health_checked_at,
5946 &mut connected_health_state,
5947 &mut connected_health_failures,
5948 connected_health_probe_ms,
5949 connected_health_failure_threshold,
5950 &mut last_presence_republish_at_ms,
5951 presence_republish_interval_ms,
5952 &mut last_network_change_at_ms,
5953 network_change_recovery_interval_ms,
5954 network_change_failure_threshold,
5955 &mut last_snapshot_signature,
5956 &mut last_snapshot_log_at,
5957 &mut last_known_node_id,
5958 &mut known_devices,
5959 ).await;
5960 }
5961 result = stream.next() => {
5962 let Some(result) = result else { break; };
5963 let events = match result {
5964 Ok(e) => e,
5965 Err(error) => {
5966 eprintln!(
5967 "[pluto-rtc][auto-connect] discovery subscription ended; recreating owner-managed stream: {}",
5968 error
5969 );
5970 break;
5971 }
5972 };
5973
5974 for event in events {
5975 match event {
5976 crate::signaling::DeviceEvent::Added { device }
5977 | crate::signaling::DeviceEvent::Modified { device } => {
5978 known_devices
5979 .insert(device.device_id.clone(), device.clone());
5980 self.try_auto_connect_device(
5981 &user_id,
5982 device,
5983 &local_device_id,
5984 &mut last_attempt_at,
5985 &mut failure_count,
5986 &mut failure_backoff_until,
5987 retry_throttle_ms,
5988 &mut non_initiator_wait_started_at,
5989 &mut non_initiator_escalation_count,
5990 initiator_grace_ms,
5991 &mut skip_log_at,
5992 &mut connected_health_checked_at,
5993 &mut connected_health_state,
5994 &mut connected_health_failures,
5995 connected_health_probe_ms_fg,
5996 connected_health_failure_threshold,
5997 &mut last_presence_republish_at_ms,
5998 presence_republish_interval_ms_fg,
5999 &mut last_network_change_at_ms,
6000 network_change_recovery_interval_ms_fg,
6001 network_change_failure_threshold,
6002 &mut last_known_node_id,
6003 )
6004 .await;
6005 }
6006 crate::signaling::DeviceEvent::Removed { device_id } => {
6007 known_devices.remove(&device_id);
6008 last_attempt_at.remove(&device_id);
6009 non_initiator_wait_started_at.remove(&device_id);
6010 non_initiator_escalation_count.remove(&device_id);
6011 skip_log_at.retain(|key, _| !key.starts_with(&format!("{}::", device_id)));
6012 connected_health_checked_at.remove(&device_id);
6013 connected_health_state.remove(&device_id);
6014 connected_health_failures.remove(&device_id);
6015 last_known_node_id
6016 .remove(&format!("{device_id}::runtime-instance"));
6017 last_known_node_id
6018 .remove(&format!("{device_id}::ticket"));
6019 clear_auto_connect_failure_state(
6020 &device_id,
6021 &mut failure_count,
6022 &mut failure_backoff_until,
6023 );
6024
6025 for record in self.connection_manager.get_by_device_id(&device_id).await {
6026 if matches!(
6027 record.state,
6028 crate::connection_manager::ConnectionState::Pending
6029 | crate::connection_manager::ConnectionState::Connecting
6030 ) {
6031 self.connection_manager
6032 .set_failed(
6033 &record.connection_id,
6034 Some("device removed from signaling snapshot".to_string()),
6035 )
6036 .await;
6037 }
6038 }
6039 }
6040 }
6041 }
6042 }
6043 }
6044 }
6045 }
6046
6047 #[cfg(target_arch = "wasm32")]
6048 {
6049 let mut stream = stream.fuse();
6050
6051 loop {
6052 if !self.is_auto_connect_generation_current(generation) {
6053 break;
6054 }
6055 let bg = self.is_app_execution_suspended();
6056 let connected_health_probe_ms = if bg {
6057 connected_health_probe_ms_bg
6058 } else {
6059 connected_health_probe_ms_fg
6060 };
6061 let presence_republish_interval_ms = if bg {
6062 presence_republish_interval_ms_bg
6063 } else {
6064 presence_republish_interval_ms_fg
6065 };
6066 let network_change_recovery_interval_ms = if bg {
6067 network_change_recovery_interval_ms_bg
6068 } else {
6069 network_change_recovery_interval_ms_fg
6070 };
6071 let rescan_ms = if bg {
6072 rescan_interval_ms_bg
6073 } else {
6074 rescan_interval_ms_fg
6075 };
6076 let sleep =
6077 gloo_timers::future::sleep(std::time::Duration::from_millis(rescan_ms))
6078 .fuse();
6079 futures::pin_mut!(sleep);
6080 futures::select! {
6081 _ = sleep => {
6082 if self.is_app_execution_suspended() {
6083 continue;
6084 }
6085 self.run_auto_connect_snapshot(
6086 &user_id,
6087 &local_device_id,
6088 &mut last_attempt_at,
6089 &mut failure_count,
6090 &mut failure_backoff_until,
6091 retry_throttle_ms,
6092 &mut non_initiator_wait_started_at,
6093 &mut non_initiator_escalation_count,
6094 initiator_grace_ms,
6095 &mut skip_log_at,
6096 &mut connected_health_checked_at,
6097 &mut connected_health_state,
6098 &mut connected_health_failures,
6099 connected_health_probe_ms,
6100 connected_health_failure_threshold,
6101 &mut last_presence_republish_at_ms,
6102 presence_republish_interval_ms,
6103 &mut last_network_change_at_ms,
6104 network_change_recovery_interval_ms,
6105 network_change_failure_threshold,
6106 &mut last_snapshot_signature,
6107 &mut last_snapshot_log_at,
6108 &mut last_known_node_id,
6109 &mut known_devices,
6110 ).await;
6111 }
6112 result = stream.next() => {
6113 let Some(result) = result else { break; };
6114 let events = match result {
6115 Ok(e) => e,
6116 Err(error) => {
6117 web_sys::console::error_1(&wasm_bindgen::JsValue::from_str(
6118 &format!(
6119 "[pluto-rtc][auto-connect] discovery subscription ended; recreating owner-managed stream: {}",
6120 error
6121 ),
6122 ));
6123 break;
6124 }
6125 };
6126
6127 for event in events {
6128 match event {
6129 crate::signaling::DeviceEvent::Added { device }
6130 | crate::signaling::DeviceEvent::Modified { device } => {
6131 self.try_auto_connect_device(
6133 &user_id,
6134 device,
6135 &local_device_id,
6136 &mut last_attempt_at,
6137 &mut failure_count,
6138 &mut failure_backoff_until,
6139 retry_throttle_ms,
6140 &mut non_initiator_wait_started_at,
6141 &mut non_initiator_escalation_count,
6142 initiator_grace_ms,
6143 &mut skip_log_at,
6144 &mut connected_health_checked_at,
6145 &mut connected_health_state,
6146 &mut connected_health_failures,
6147 connected_health_probe_ms_fg,
6148 connected_health_failure_threshold,
6149 &mut last_presence_republish_at_ms,
6150 presence_republish_interval_ms_fg,
6151 &mut last_network_change_at_ms,
6152 network_change_recovery_interval_ms_fg,
6153 network_change_failure_threshold,
6154 &mut last_known_node_id,
6155 )
6156 .await;
6157 }
6158 crate::signaling::DeviceEvent::Removed { device_id } => {
6159 last_attempt_at.remove(&device_id);
6160 non_initiator_wait_started_at.remove(&device_id);
6161 non_initiator_escalation_count.remove(&device_id);
6162 skip_log_at.retain(|key, _| !key.starts_with(&format!("{}::", device_id)));
6163 connected_health_checked_at.remove(&device_id);
6164 connected_health_state.remove(&device_id);
6165 connected_health_failures.remove(&device_id);
6166 last_known_node_id
6167 .remove(&format!("{device_id}::runtime-instance"));
6168 last_known_node_id
6169 .remove(&format!("{device_id}::ticket"));
6170 clear_auto_connect_failure_state(
6171 &device_id,
6172 &mut failure_count,
6173 &mut failure_backoff_until,
6174 );
6175
6176 for record in self.connection_manager.get_by_device_id(&device_id).await {
6177 if matches!(
6178 record.state,
6179 crate::connection_manager::ConnectionState::Pending
6180 | crate::connection_manager::ConnectionState::Connecting
6181 ) {
6182 self.connection_manager
6183 .set_failed(
6184 &record.connection_id,
6185 Some("device removed from signaling snapshot".to_string()),
6186 )
6187 .await;
6188 }
6189 }
6190 }
6191 }
6192 }
6193 }
6194 }
6195 }
6196 }
6197
6198 #[cfg(not(target_arch = "wasm32"))]
6199 eprintln!(
6200 "[pluto-rtc][auto-connect] device stream ended user_id={} local_device_id={}, resubscribing",
6201 user_id, local_device_id
6202 );
6203 #[cfg(target_arch = "wasm32")]
6204 web_sys::console::log_1(&wasm_bindgen::JsValue::from_str(&format!(
6205 "[pluto-rtc][auto-connect] device stream ended user_id={} local_device_id={}, resubscribing",
6206 user_id, local_device_id
6207 )));
6208
6209 #[cfg(not(target_arch = "wasm32"))]
6210 tokio::time::sleep(std::time::Duration::from_millis(400)).await;
6211 #[cfg(target_arch = "wasm32")]
6212 gloo_timers::future::sleep(std::time::Duration::from_millis(400)).await;
6213 }
6214 }
6215}