1pub mod tcp;
8pub mod tor;
9pub mod udp;
10#[cfg(feature = "webrtc-transport")]
11pub mod webrtc;
12
13#[cfg(feature = "sim-transport")]
14pub mod sim;
15
16#[cfg(any(target_os = "linux", target_os = "macos"))]
17pub mod ethernet;
18
19#[cfg(target_os = "linux")]
20pub mod ble;
21
22mod handle;
23mod packet_channel;
24
25#[cfg(test)]
26mod tests;
27
28pub use handle::TransportHandle;
29#[cfg(any(target_os = "linux", target_os = "macos"))]
30pub(crate) use packet_channel::received_timestamp_ms;
31pub use packet_channel::{PacketBuffer, PacketRx, PacketTx, ReceivedPacket, packet_channel};
32
33use secp256k1::XOnlyPublicKey;
34use std::fmt;
35use std::net::SocketAddr;
36use std::sync::Arc;
37use std::time::Duration;
38use thiserror::Error;
39
40#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
46pub struct TransportId(u32);
47
48impl TransportId {
49 pub fn new(id: u32) -> Self {
51 Self(id)
52 }
53
54 pub fn as_u32(&self) -> u32 {
56 self.0
57 }
58}
59
60impl fmt::Display for TransportId {
61 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
62 write!(f, "transport:{}", self.0)
63 }
64}
65
66#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
68pub struct LinkId(u64);
69
70impl LinkId {
71 pub fn new(id: u64) -> Self {
73 Self(id)
74 }
75
76 pub fn as_u64(&self) -> u64 {
78 self.0
79 }
80}
81
82impl fmt::Display for LinkId {
83 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
84 write!(f, "link:{}", self.0)
85 }
86}
87
88#[derive(Debug, Error)]
94pub enum TransportError {
95 #[error("transport not started")]
96 NotStarted,
97
98 #[error("transport already started")]
99 AlreadyStarted,
100
101 #[error("transport failed to start: {0}")]
102 StartFailed(String),
103
104 #[error("transport shutdown failed: {0}")]
105 ShutdownFailed(String),
106
107 #[error("link failed: {0}")]
108 LinkFailed(String),
109
110 #[error("send failed: {0}")]
111 SendFailed(String),
112
113 #[error("receive failed: {0}")]
114 RecvFailed(String),
115
116 #[error("invalid transport address: {0}")]
117 InvalidAddress(String),
118
119 #[error("mtu exceeded: packet {packet_size} > mtu {mtu}")]
120 MtuExceeded { packet_size: usize, mtu: u16 },
121
122 #[error("transport timeout")]
123 Timeout,
124
125 #[error("connection refused")]
126 ConnectionRefused,
127
128 #[error("transport not supported: {0}")]
129 NotSupported(String),
130
131 #[error("io error: {0}")]
132 Io(#[from] std::io::Error),
133}
134
135impl TransportError {
136 pub fn is_local_route_unavailable(&self) -> bool {
139 match self {
140 TransportError::Io(error) => is_local_route_error_kind(error.kind()),
141 TransportError::SendFailed(message) => is_local_route_error_text(message),
142 _ => false,
143 }
144 }
145}
146
147fn is_local_route_error_kind(kind: std::io::ErrorKind) -> bool {
148 matches!(
149 kind,
150 std::io::ErrorKind::NetworkUnreachable
151 | std::io::ErrorKind::HostUnreachable
152 | std::io::ErrorKind::AddrNotAvailable
153 | std::io::ErrorKind::PermissionDenied
154 )
155}
156
157fn is_local_route_error_text(message: &str) -> bool {
158 let lower = message.to_ascii_lowercase();
159 lower.contains("network is unreachable")
160 || lower.contains("no route to host")
161 || lower.contains("host is unreachable")
162 || lower.contains("can't assign requested address")
163 || lower.contains("cannot assign requested address")
164 || lower.contains("operation not permitted")
165 || lower.contains("permission denied")
166 || lower.contains("os error 51")
167 || lower.contains("os error 65")
168 || lower.contains("os error 49")
169 || lower.contains("os error 1")
170}
171
172#[derive(Clone, Debug, PartialEq, Eq)]
178pub struct TransportType {
179 pub name: &'static str,
181 pub connection_oriented: bool,
183 pub reliable: bool,
185}
186
187impl TransportType {
188 pub const UDP: TransportType = TransportType {
190 name: "udp",
191 connection_oriented: false,
192 reliable: false,
193 };
194
195 pub const TCP: TransportType = TransportType {
197 name: "tcp",
198 connection_oriented: true,
199 reliable: true,
200 };
201
202 pub const ETHERNET: TransportType = TransportType {
204 name: "ethernet",
205 connection_oriented: false,
206 reliable: false,
207 };
208
209 pub const WIFI: TransportType = TransportType {
211 name: "wifi",
212 connection_oriented: false,
213 reliable: false,
214 };
215
216 pub const TOR: TransportType = TransportType {
218 name: "tor",
219 connection_oriented: true,
220 reliable: true,
221 };
222
223 pub const SERIAL: TransportType = TransportType {
225 name: "serial",
226 connection_oriented: false,
227 reliable: true, };
229
230 pub const BLE: TransportType = TransportType {
232 name: "ble",
233 connection_oriented: true,
234 reliable: true, };
236
237 pub const WEBRTC: TransportType = TransportType {
239 name: "webrtc",
240 connection_oriented: true,
241 reliable: false,
242 };
243
244 #[cfg(feature = "sim-transport")]
246 pub const SIM: TransportType = TransportType {
247 name: "sim",
248 connection_oriented: false,
249 reliable: false,
250 };
251
252 pub fn is_connectionless(&self) -> bool {
254 !self.connection_oriented
255 }
256}
257
258impl fmt::Display for TransportType {
259 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
260 write!(f, "{}", self.name)
261 }
262}
263
264#[derive(Clone, Copy, Debug, PartialEq, Eq)]
270pub enum TransportState {
271 Configured,
273 Starting,
275 Up,
277 Down,
279 Failed,
281}
282
283impl TransportState {
284 pub fn is_operational(&self) -> bool {
286 matches!(self, TransportState::Up)
287 }
288
289 pub fn can_start(&self) -> bool {
291 matches!(
292 self,
293 TransportState::Configured | TransportState::Down | TransportState::Failed
294 )
295 }
296
297 pub fn is_terminal(&self) -> bool {
299 matches!(self, TransportState::Failed)
300 }
301}
302
303impl fmt::Display for TransportState {
304 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
305 let s = match self {
306 TransportState::Configured => "configured",
307 TransportState::Starting => "starting",
308 TransportState::Up => "up",
309 TransportState::Down => "down",
310 TransportState::Failed => "failed",
311 };
312 write!(f, "{}", s)
313 }
314}
315
316#[derive(Clone, Copy, Debug, PartialEq, Eq)]
322pub enum LinkState {
323 Connecting,
325 Connected,
327 Disconnected,
329 Failed,
331}
332
333impl LinkState {
334 pub fn is_operational(&self) -> bool {
336 matches!(self, LinkState::Connected)
337 }
338
339 pub fn is_terminal(&self) -> bool {
341 matches!(self, LinkState::Disconnected | LinkState::Failed)
342 }
343}
344
345impl fmt::Display for LinkState {
346 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
347 let s = match self {
348 LinkState::Connecting => "connecting",
349 LinkState::Connected => "connected",
350 LinkState::Disconnected => "disconnected",
351 LinkState::Failed => "failed",
352 };
353 write!(f, "{}", s)
354 }
355}
356
357#[derive(Clone, Copy, Debug, PartialEq, Eq)]
359pub enum LinkDirection {
360 Outbound,
362 Inbound,
364}
365
366impl fmt::Display for LinkDirection {
367 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
368 let s = match self {
369 LinkDirection::Outbound => "outbound",
370 LinkDirection::Inbound => "inbound",
371 };
372 write!(f, "{}", s)
373 }
374}
375
376#[derive(Clone, PartialEq, Eq, Hash)]
389pub struct TransportAddr(Arc<[u8]>);
390
391impl TransportAddr {
392 pub fn new(bytes: Vec<u8>) -> Self {
394 Self(bytes.into())
395 }
396
397 pub fn from_bytes(bytes: &[u8]) -> Self {
399 Self(Arc::from(bytes))
400 }
401
402 pub fn from_string(s: &str) -> Self {
404 Self(Arc::from(s.as_bytes()))
405 }
406
407 pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self {
419 use std::io::Write;
420 let mut buf = Vec::with_capacity(56);
424 write!(&mut buf, "{addr}").expect("Vec<u8>::write_fmt is infallible");
428 Self(buf.into())
429 }
430
431 pub fn as_bytes(&self) -> &[u8] {
433 &self.0
434 }
435
436 pub fn as_str(&self) -> Option<&str> {
438 std::str::from_utf8(&self.0).ok()
439 }
440
441 pub fn len(&self) -> usize {
443 self.0.len()
444 }
445
446 pub fn is_empty(&self) -> bool {
448 self.0.is_empty()
449 }
450}
451
452impl fmt::Debug for TransportAddr {
453 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
454 match self.as_str() {
455 Some(s) => write!(f, "TransportAddr(\"{}\")", s),
456 None => write!(f, "TransportAddr({:?})", self.0),
457 }
458 }
459}
460
461impl fmt::Display for TransportAddr {
462 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
463 match self.as_str() {
465 Some(s) => write!(f, "{}", s),
466 None => {
467 for byte in self.0.iter() {
468 write!(f, "{:02x}", byte)?;
469 }
470 Ok(())
471 }
472 }
473 }
474}
475
476impl From<&str> for TransportAddr {
477 fn from(s: &str) -> Self {
478 Self::from_string(s)
479 }
480}
481
482impl From<String> for TransportAddr {
483 fn from(s: String) -> Self {
484 Self(s.into_bytes().into())
485 }
486}
487
488#[derive(Clone, Debug, Default)]
494pub struct LinkStats {
495 pub packets_sent: u64,
497 pub packets_recv: u64,
499 pub bytes_sent: u64,
501 pub bytes_recv: u64,
503 pub last_recv_ms: u64,
505 rtt_estimate: Option<Duration>,
507 pub loss_rate: f32,
509 pub throughput_estimate: u64,
511}
512
513impl LinkStats {
514 pub fn new() -> Self {
516 Self::default()
517 }
518
519 pub fn record_sent(&mut self, bytes: usize) {
521 self.packets_sent += 1;
522 self.bytes_sent += bytes as u64;
523 }
524
525 pub fn record_sent_batch(&mut self, packets: usize, bytes: usize) {
527 self.packets_sent += packets as u64;
528 self.bytes_sent += bytes as u64;
529 }
530
531 pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) {
533 self.packets_recv += 1;
534 self.bytes_recv += bytes as u64;
535 self.last_recv_ms = timestamp_ms;
536 }
537
538 pub fn rtt_estimate(&self) -> Option<Duration> {
540 self.rtt_estimate
541 }
542
543 pub fn update_rtt(&mut self, rtt: Duration) {
547 match self.rtt_estimate {
548 Some(old_rtt) => {
549 let alpha = 0.2;
550 let new_rtt_nanos = (alpha * rtt.as_nanos() as f64
551 + (1.0 - alpha) * old_rtt.as_nanos() as f64)
552 as u64;
553 self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos));
554 }
555 None => {
556 self.rtt_estimate = Some(rtt);
557 }
558 }
559 }
560
561 pub fn time_since_recv(&self, current_time_ms: u64) -> u64 {
563 if self.last_recv_ms == 0 {
564 return u64::MAX;
565 }
566 current_time_ms.saturating_sub(self.last_recv_ms)
567 }
568
569 pub fn reset(&mut self) {
571 *self = Self::default();
572 }
573}
574
575#[derive(Clone, Debug)]
581pub struct Link {
582 link_id: LinkId,
584 transport_id: TransportId,
586 remote_addr: TransportAddr,
588 direction: LinkDirection,
590 state: LinkState,
592 base_rtt: Duration,
594 stats: LinkStats,
596 created_at: u64,
598}
599
600impl Link {
601 pub fn new(
603 link_id: LinkId,
604 transport_id: TransportId,
605 remote_addr: TransportAddr,
606 direction: LinkDirection,
607 base_rtt: Duration,
608 ) -> Self {
609 Self {
610 link_id,
611 transport_id,
612 remote_addr,
613 direction,
614 state: LinkState::Connecting,
615 base_rtt,
616 stats: LinkStats::new(),
617 created_at: 0,
618 }
619 }
620
621 pub fn new_with_timestamp(
623 link_id: LinkId,
624 transport_id: TransportId,
625 remote_addr: TransportAddr,
626 direction: LinkDirection,
627 base_rtt: Duration,
628 created_at: u64,
629 ) -> Self {
630 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
631 link.created_at = created_at;
632 link
633 }
634
635 pub fn connectionless(
640 link_id: LinkId,
641 transport_id: TransportId,
642 remote_addr: TransportAddr,
643 direction: LinkDirection,
644 base_rtt: Duration,
645 ) -> Self {
646 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
647 link.state = LinkState::Connected;
648 link
649 }
650
651 pub fn link_id(&self) -> LinkId {
653 self.link_id
654 }
655
656 pub fn transport_id(&self) -> TransportId {
658 self.transport_id
659 }
660
661 pub fn remote_addr(&self) -> &TransportAddr {
663 &self.remote_addr
664 }
665
666 pub fn direction(&self) -> LinkDirection {
668 self.direction
669 }
670
671 pub fn state(&self) -> LinkState {
673 self.state
674 }
675
676 pub fn base_rtt(&self) -> Duration {
678 self.base_rtt
679 }
680
681 pub fn stats(&self) -> &LinkStats {
683 &self.stats
684 }
685
686 pub fn stats_mut(&mut self) -> &mut LinkStats {
688 &mut self.stats
689 }
690
691 pub fn created_at(&self) -> u64 {
693 self.created_at
694 }
695
696 pub fn set_created_at(&mut self, timestamp: u64) {
698 self.created_at = timestamp;
699 }
700
701 pub fn set_connected(&mut self) {
703 self.state = LinkState::Connected;
704 }
705
706 pub fn set_disconnected(&mut self) {
708 self.state = LinkState::Disconnected;
709 }
710
711 pub fn set_failed(&mut self) {
713 self.state = LinkState::Failed;
714 }
715
716 pub fn is_operational(&self) -> bool {
718 self.state.is_operational()
719 }
720
721 pub fn is_terminal(&self) -> bool {
723 self.state.is_terminal()
724 }
725
726 pub fn effective_rtt(&self) -> Duration {
728 self.stats.rtt_estimate().unwrap_or(self.base_rtt)
729 }
730
731 pub fn age(&self, current_time_ms: u64) -> u64 {
733 if self.created_at == 0 {
734 return 0;
735 }
736 current_time_ms.saturating_sub(self.created_at)
737 }
738}
739
740#[derive(Clone, Debug)]
746pub struct DiscoveredPeer {
747 pub transport_id: TransportId,
749 pub addr: TransportAddr,
751 pub pubkey_hint: Option<XOnlyPublicKey>,
753}
754
755impl DiscoveredPeer {
756 pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self {
758 Self {
759 transport_id,
760 addr,
761 pubkey_hint: None,
762 }
763 }
764
765 pub fn with_hint(
767 transport_id: TransportId,
768 addr: TransportAddr,
769 pubkey: XOnlyPublicKey,
770 ) -> Self {
771 Self {
772 transport_id,
773 addr,
774 pubkey_hint: Some(pubkey),
775 }
776 }
777}
778
779pub trait Transport {
788 fn transport_id(&self) -> TransportId;
790
791 fn transport_type(&self) -> &TransportType;
793
794 fn state(&self) -> TransportState;
796
797 fn mtu(&self) -> u16;
799
800 fn link_mtu(&self, addr: &TransportAddr) -> u16 {
806 let _ = addr;
807 self.mtu()
808 }
809
810 fn start(&mut self) -> Result<(), TransportError>;
812
813 fn stop(&mut self) -> Result<(), TransportError>;
815
816 fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>;
818
819 fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError>;
821
822 fn auto_connect(&self) -> bool {
825 false
826 }
827
828 fn accept_connections(&self) -> bool {
831 true
832 }
833
834 fn close_connection(&self, _addr: &TransportAddr) {
840 }
842}
843
844#[derive(Clone, Debug, PartialEq, Eq)]
853pub enum ConnectionState {
854 None,
856 Connecting,
858 Connected,
860 Failed(String),
862}
863
864#[derive(Clone, Debug, Default)]
873pub struct TransportCongestion {
874 pub recv_drops: Option<u64>,
877 pub socket_recv_drops: Option<u64>,
880 pub namespace_recv_drops: Option<u64>,
884}
885
886pub(crate) async fn resolve_socket_addr(
897 addr: &TransportAddr,
898) -> Result<SocketAddr, TransportError> {
899 resolve_socket_addrs(addr)
900 .await?
901 .into_iter()
902 .next()
903 .ok_or_else(|| {
904 TransportError::InvalidAddress(format!(
905 "DNS resolution returned no addresses for {}",
906 addr.as_str().unwrap_or("<non-utf8>")
907 ))
908 })
909}
910
911pub(crate) async fn resolve_socket_addrs(
918 addr: &TransportAddr,
919) -> Result<Vec<SocketAddr>, TransportError> {
920 let s = addr
921 .as_str()
922 .ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?;
923
924 if let Ok(sock_addr) = s.parse::<SocketAddr>() {
926 return Ok(vec![sock_addr]);
927 }
928
929 let addrs = tokio::net::lookup_host(s)
931 .await
932 .map_err(|e| {
933 TransportError::InvalidAddress(format!("DNS resolution failed for {}: {}", s, e))
934 })?
935 .collect::<Vec<_>>();
936 if addrs.is_empty() {
937 return Err(TransportError::InvalidAddress(format!(
938 "DNS resolution returned no addresses for {}",
939 s
940 )));
941 }
942 Ok(addrs)
943}