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;
29pub(crate) use packet_channel::PacketFastIngressSink;
30#[cfg(target_os = "linux")]
31pub(crate) use packet_channel::received_timestamp_ms;
32pub use packet_channel::{PacketBuffer, PacketRx, PacketTx, ReceivedPacket, packet_channel};
33
34use secp256k1::XOnlyPublicKey;
35use std::fmt;
36use std::net::SocketAddr;
37use std::sync::Arc;
38use std::time::Duration;
39use thiserror::Error;
40
41#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
47pub struct TransportId(u32);
48
49impl TransportId {
50 pub fn new(id: u32) -> Self {
52 Self(id)
53 }
54
55 pub fn as_u32(&self) -> u32 {
57 self.0
58 }
59}
60
61impl fmt::Display for TransportId {
62 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
63 write!(f, "transport:{}", self.0)
64 }
65}
66
67#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
69pub struct LinkId(u64);
70
71impl LinkId {
72 pub fn new(id: u64) -> Self {
74 Self(id)
75 }
76
77 pub fn as_u64(&self) -> u64 {
79 self.0
80 }
81}
82
83impl fmt::Display for LinkId {
84 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
85 write!(f, "link:{}", self.0)
86 }
87}
88
89#[derive(Debug, Error)]
95pub enum TransportError {
96 #[error("transport not started")]
97 NotStarted,
98
99 #[error("transport already started")]
100 AlreadyStarted,
101
102 #[error("transport failed to start: {0}")]
103 StartFailed(String),
104
105 #[error("transport shutdown failed: {0}")]
106 ShutdownFailed(String),
107
108 #[error("link failed: {0}")]
109 LinkFailed(String),
110
111 #[error("send failed: {0}")]
112 SendFailed(String),
113
114 #[error("receive failed: {0}")]
115 RecvFailed(String),
116
117 #[error("invalid transport address: {0}")]
118 InvalidAddress(String),
119
120 #[error("mtu exceeded: packet {packet_size} > mtu {mtu}")]
121 MtuExceeded { packet_size: usize, mtu: u16 },
122
123 #[error("transport timeout")]
124 Timeout,
125
126 #[error("connection refused")]
127 ConnectionRefused,
128
129 #[error("transport not supported: {0}")]
130 NotSupported(String),
131
132 #[error("io error: {0}")]
133 Io(#[from] std::io::Error),
134}
135
136impl TransportError {
137 pub fn is_local_route_unavailable(&self) -> bool {
140 match self {
141 TransportError::Io(error) => is_local_route_error_kind(error.kind()),
142 TransportError::SendFailed(message) => is_local_route_error_text(message),
143 _ => false,
144 }
145 }
146}
147
148fn is_local_route_error_kind(kind: std::io::ErrorKind) -> bool {
149 matches!(
150 kind,
151 std::io::ErrorKind::NetworkUnreachable
152 | std::io::ErrorKind::HostUnreachable
153 | std::io::ErrorKind::AddrNotAvailable
154 | std::io::ErrorKind::PermissionDenied
155 )
156}
157
158fn is_local_route_error_text(message: &str) -> bool {
159 let lower = message.to_ascii_lowercase();
160 lower.contains("network is unreachable")
161 || lower.contains("no route to host")
162 || lower.contains("host is unreachable")
163 || lower.contains("can't assign requested address")
164 || lower.contains("cannot assign requested address")
165 || lower.contains("operation not permitted")
166 || lower.contains("permission denied")
167 || lower.contains("os error 51")
168 || lower.contains("os error 65")
169 || lower.contains("os error 49")
170 || lower.contains("os error 1")
171}
172
173#[derive(Clone, Debug, PartialEq, Eq)]
179pub struct TransportType {
180 pub name: &'static str,
182 pub connection_oriented: bool,
184 pub reliable: bool,
186}
187
188impl TransportType {
189 pub const UDP: TransportType = TransportType {
191 name: "udp",
192 connection_oriented: false,
193 reliable: false,
194 };
195
196 pub const TCP: TransportType = TransportType {
198 name: "tcp",
199 connection_oriented: true,
200 reliable: true,
201 };
202
203 pub const ETHERNET: TransportType = TransportType {
205 name: "ethernet",
206 connection_oriented: false,
207 reliable: false,
208 };
209
210 pub const WIFI: TransportType = TransportType {
212 name: "wifi",
213 connection_oriented: false,
214 reliable: false,
215 };
216
217 pub const TOR: TransportType = TransportType {
219 name: "tor",
220 connection_oriented: true,
221 reliable: true,
222 };
223
224 pub const SERIAL: TransportType = TransportType {
226 name: "serial",
227 connection_oriented: false,
228 reliable: true, };
230
231 pub const BLE: TransportType = TransportType {
233 name: "ble",
234 connection_oriented: true,
235 reliable: true, };
237
238 pub const WEBRTC: TransportType = TransportType {
240 name: "webrtc",
241 connection_oriented: true,
242 reliable: false,
243 };
244
245 #[cfg(feature = "sim-transport")]
247 pub const SIM: TransportType = TransportType {
248 name: "sim",
249 connection_oriented: false,
250 reliable: false,
251 };
252
253 pub fn is_connectionless(&self) -> bool {
255 !self.connection_oriented
256 }
257}
258
259impl fmt::Display for TransportType {
260 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
261 write!(f, "{}", self.name)
262 }
263}
264
265#[derive(Clone, Copy, Debug, PartialEq, Eq)]
271pub enum TransportState {
272 Configured,
274 Starting,
276 Up,
278 Down,
280 Failed,
282}
283
284impl TransportState {
285 pub fn is_operational(&self) -> bool {
287 matches!(self, TransportState::Up)
288 }
289
290 pub fn can_start(&self) -> bool {
292 matches!(
293 self,
294 TransportState::Configured | TransportState::Down | TransportState::Failed
295 )
296 }
297
298 pub fn is_terminal(&self) -> bool {
300 matches!(self, TransportState::Failed)
301 }
302}
303
304impl fmt::Display for TransportState {
305 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
306 let s = match self {
307 TransportState::Configured => "configured",
308 TransportState::Starting => "starting",
309 TransportState::Up => "up",
310 TransportState::Down => "down",
311 TransportState::Failed => "failed",
312 };
313 write!(f, "{}", s)
314 }
315}
316
317#[derive(Clone, Copy, Debug, PartialEq, Eq)]
323pub enum LinkState {
324 Connecting,
326 Connected,
328 Disconnected,
330 Failed,
332}
333
334impl LinkState {
335 pub fn is_operational(&self) -> bool {
337 matches!(self, LinkState::Connected)
338 }
339
340 pub fn is_terminal(&self) -> bool {
342 matches!(self, LinkState::Disconnected | LinkState::Failed)
343 }
344}
345
346impl fmt::Display for LinkState {
347 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
348 let s = match self {
349 LinkState::Connecting => "connecting",
350 LinkState::Connected => "connected",
351 LinkState::Disconnected => "disconnected",
352 LinkState::Failed => "failed",
353 };
354 write!(f, "{}", s)
355 }
356}
357
358#[derive(Clone, Copy, Debug, PartialEq, Eq)]
360pub enum LinkDirection {
361 Outbound,
363 Inbound,
365}
366
367impl fmt::Display for LinkDirection {
368 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
369 let s = match self {
370 LinkDirection::Outbound => "outbound",
371 LinkDirection::Inbound => "inbound",
372 };
373 write!(f, "{}", s)
374 }
375}
376
377#[derive(Clone, PartialEq, Eq, Hash)]
390pub struct TransportAddr(Arc<[u8]>);
391
392impl TransportAddr {
393 pub fn new(bytes: Vec<u8>) -> Self {
395 Self(bytes.into())
396 }
397
398 pub fn from_bytes(bytes: &[u8]) -> Self {
400 Self(Arc::from(bytes))
401 }
402
403 pub fn from_string(s: &str) -> Self {
405 Self(Arc::from(s.as_bytes()))
406 }
407
408 pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self {
413 match addr {
414 std::net::SocketAddr::V4(addr) => {
415 let octets = addr.ip().octets();
416 let mut buf = Vec::with_capacity(21);
417 push_decimal_u8(&mut buf, octets[0]);
418 buf.push(b'.');
419 push_decimal_u8(&mut buf, octets[1]);
420 buf.push(b'.');
421 push_decimal_u8(&mut buf, octets[2]);
422 buf.push(b'.');
423 push_decimal_u8(&mut buf, octets[3]);
424 buf.push(b':');
425 push_decimal_u16(&mut buf, addr.port());
426 Self(buf.into())
427 }
428 std::net::SocketAddr::V6(addr) => {
429 use std::io::Write;
430 let mut buf = Vec::with_capacity(56);
431 buf.push(b'[');
432 write!(&mut buf, "{}", addr.ip()).expect("Vec<u8>::write_fmt is infallible");
433 buf.push(b']');
434 buf.push(b':');
435 push_decimal_u16(&mut buf, addr.port());
436 Self(buf.into())
437 }
438 }
439 }
440
441 pub fn as_bytes(&self) -> &[u8] {
443 &self.0
444 }
445
446 pub fn as_str(&self) -> Option<&str> {
448 std::str::from_utf8(&self.0).ok()
449 }
450
451 pub fn len(&self) -> usize {
453 self.0.len()
454 }
455
456 pub fn is_empty(&self) -> bool {
458 self.0.is_empty()
459 }
460}
461
462fn push_decimal_u8(buf: &mut Vec<u8>, value: u8) {
463 push_decimal_u16(buf, value as u16);
464}
465
466fn push_decimal_u16(buf: &mut Vec<u8>, value: u16) {
467 if value >= 10_000 {
468 buf.push(b'0' + (value / 10_000) as u8);
469 push_fixed_4_digits(buf, value % 10_000);
470 } else if value >= 1_000 {
471 push_fixed_4_digits(buf, value);
472 } else if value >= 100 {
473 buf.push(b'0' + (value / 100) as u8);
474 push_fixed_2_digits(buf, value % 100);
475 } else if value >= 10 {
476 push_fixed_2_digits(buf, value);
477 } else {
478 buf.push(b'0' + value as u8);
479 }
480}
481
482fn push_fixed_4_digits(buf: &mut Vec<u8>, value: u16) {
483 buf.push(b'0' + (value / 1_000) as u8);
484 push_fixed_3_digits(buf, value % 1_000);
485}
486
487fn push_fixed_3_digits(buf: &mut Vec<u8>, value: u16) {
488 buf.push(b'0' + (value / 100) as u8);
489 push_fixed_2_digits(buf, value % 100);
490}
491
492fn push_fixed_2_digits(buf: &mut Vec<u8>, value: u16) {
493 buf.push(b'0' + (value / 10) as u8);
494 buf.push(b'0' + (value % 10) as u8);
495}
496
497impl fmt::Debug for TransportAddr {
498 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
499 match self.as_str() {
500 Some(s) => write!(f, "TransportAddr(\"{}\")", s),
501 None => write!(f, "TransportAddr({:?})", self.0),
502 }
503 }
504}
505
506impl fmt::Display for TransportAddr {
507 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
508 match self.as_str() {
510 Some(s) => write!(f, "{}", s),
511 None => {
512 for byte in self.0.iter() {
513 write!(f, "{:02x}", byte)?;
514 }
515 Ok(())
516 }
517 }
518 }
519}
520
521impl From<&str> for TransportAddr {
522 fn from(s: &str) -> Self {
523 Self::from_string(s)
524 }
525}
526
527impl From<String> for TransportAddr {
528 fn from(s: String) -> Self {
529 Self(s.into_bytes().into())
530 }
531}
532
533#[derive(Clone, Debug, Default)]
539pub struct LinkStats {
540 pub packets_sent: u64,
542 pub packets_recv: u64,
544 pub bytes_sent: u64,
546 pub bytes_recv: u64,
548 pub last_recv_ms: u64,
550 rtt_estimate: Option<Duration>,
552 pub loss_rate: f32,
554 pub throughput_estimate: u64,
556}
557
558impl LinkStats {
559 pub fn new() -> Self {
561 Self::default()
562 }
563
564 pub fn record_sent(&mut self, bytes: usize) {
566 self.packets_sent += 1;
567 self.bytes_sent += bytes as u64;
568 }
569
570 pub fn record_sent_batch(&mut self, packets: usize, bytes: usize) {
572 self.packets_sent += packets as u64;
573 self.bytes_sent += bytes as u64;
574 }
575
576 pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) {
578 self.packets_recv += 1;
579 self.bytes_recv += bytes as u64;
580 self.last_recv_ms = timestamp_ms;
581 }
582
583 pub fn rtt_estimate(&self) -> Option<Duration> {
585 self.rtt_estimate
586 }
587
588 pub fn update_rtt(&mut self, rtt: Duration) {
592 match self.rtt_estimate {
593 Some(old_rtt) => {
594 let alpha = 0.2;
595 let new_rtt_nanos = (alpha * rtt.as_nanos() as f64
596 + (1.0 - alpha) * old_rtt.as_nanos() as f64)
597 as u64;
598 self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos));
599 }
600 None => {
601 self.rtt_estimate = Some(rtt);
602 }
603 }
604 }
605
606 pub fn time_since_recv(&self, current_time_ms: u64) -> u64 {
608 if self.last_recv_ms == 0 {
609 return u64::MAX;
610 }
611 current_time_ms.saturating_sub(self.last_recv_ms)
612 }
613
614 pub fn reset(&mut self) {
616 *self = Self::default();
617 }
618}
619
620#[derive(Clone, Debug)]
626pub struct Link {
627 link_id: LinkId,
629 transport_id: TransportId,
631 remote_addr: TransportAddr,
633 direction: LinkDirection,
635 state: LinkState,
637 base_rtt: Duration,
639 stats: LinkStats,
641 created_at: u64,
643}
644
645impl Link {
646 pub fn new(
648 link_id: LinkId,
649 transport_id: TransportId,
650 remote_addr: TransportAddr,
651 direction: LinkDirection,
652 base_rtt: Duration,
653 ) -> Self {
654 Self {
655 link_id,
656 transport_id,
657 remote_addr,
658 direction,
659 state: LinkState::Connecting,
660 base_rtt,
661 stats: LinkStats::new(),
662 created_at: 0,
663 }
664 }
665
666 pub fn new_with_timestamp(
668 link_id: LinkId,
669 transport_id: TransportId,
670 remote_addr: TransportAddr,
671 direction: LinkDirection,
672 base_rtt: Duration,
673 created_at: u64,
674 ) -> Self {
675 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
676 link.created_at = created_at;
677 link
678 }
679
680 pub fn connectionless(
685 link_id: LinkId,
686 transport_id: TransportId,
687 remote_addr: TransportAddr,
688 direction: LinkDirection,
689 base_rtt: Duration,
690 ) -> Self {
691 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
692 link.state = LinkState::Connected;
693 link
694 }
695
696 pub fn link_id(&self) -> LinkId {
698 self.link_id
699 }
700
701 pub fn transport_id(&self) -> TransportId {
703 self.transport_id
704 }
705
706 pub fn remote_addr(&self) -> &TransportAddr {
708 &self.remote_addr
709 }
710
711 pub fn direction(&self) -> LinkDirection {
713 self.direction
714 }
715
716 pub fn state(&self) -> LinkState {
718 self.state
719 }
720
721 pub fn base_rtt(&self) -> Duration {
723 self.base_rtt
724 }
725
726 pub fn stats(&self) -> &LinkStats {
728 &self.stats
729 }
730
731 pub fn stats_mut(&mut self) -> &mut LinkStats {
733 &mut self.stats
734 }
735
736 pub fn created_at(&self) -> u64 {
738 self.created_at
739 }
740
741 pub fn set_created_at(&mut self, timestamp: u64) {
743 self.created_at = timestamp;
744 }
745
746 pub fn set_connected(&mut self) {
748 self.state = LinkState::Connected;
749 }
750
751 pub fn set_disconnected(&mut self) {
753 self.state = LinkState::Disconnected;
754 }
755
756 pub fn set_failed(&mut self) {
758 self.state = LinkState::Failed;
759 }
760
761 pub fn is_operational(&self) -> bool {
763 self.state.is_operational()
764 }
765
766 pub fn is_terminal(&self) -> bool {
768 self.state.is_terminal()
769 }
770
771 pub fn effective_rtt(&self) -> Duration {
773 self.stats.rtt_estimate().unwrap_or(self.base_rtt)
774 }
775
776 pub fn age(&self, current_time_ms: u64) -> u64 {
778 if self.created_at == 0 {
779 return 0;
780 }
781 current_time_ms.saturating_sub(self.created_at)
782 }
783}
784
785#[derive(Clone, Debug)]
791pub struct DiscoveredPeer {
792 pub transport_id: TransportId,
794 pub addr: TransportAddr,
796 pub pubkey_hint: Option<XOnlyPublicKey>,
798}
799
800impl DiscoveredPeer {
801 pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self {
803 Self {
804 transport_id,
805 addr,
806 pubkey_hint: None,
807 }
808 }
809
810 pub fn with_hint(
812 transport_id: TransportId,
813 addr: TransportAddr,
814 pubkey: XOnlyPublicKey,
815 ) -> Self {
816 Self {
817 transport_id,
818 addr,
819 pubkey_hint: Some(pubkey),
820 }
821 }
822}
823
824pub trait Transport {
833 fn transport_id(&self) -> TransportId;
835
836 fn transport_type(&self) -> &TransportType;
838
839 fn state(&self) -> TransportState;
841
842 fn mtu(&self) -> u16;
844
845 fn link_mtu(&self, addr: &TransportAddr) -> u16 {
851 let _ = addr;
852 self.mtu()
853 }
854
855 fn start(&mut self) -> Result<(), TransportError>;
857
858 fn stop(&mut self) -> Result<(), TransportError>;
860
861 fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>;
863
864 fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError>;
866
867 fn auto_connect(&self) -> bool {
870 false
871 }
872
873 fn accept_connections(&self) -> bool {
876 true
877 }
878
879 fn close_connection(&self, _addr: &TransportAddr) {
885 }
887}
888
889#[derive(Clone, Debug, PartialEq, Eq)]
898pub enum ConnectionState {
899 None,
901 Connecting,
903 Connected,
905 Failed(String),
907}
908
909#[derive(Clone, Debug, Default)]
918pub struct TransportCongestion {
919 pub recv_drops: Option<u64>,
922 pub socket_recv_drops: Option<u64>,
925 pub namespace_recv_drops: Option<u64>,
929}
930
931pub(crate) async fn resolve_socket_addr(
942 addr: &TransportAddr,
943) -> Result<SocketAddr, TransportError> {
944 resolve_socket_addrs(addr)
945 .await?
946 .into_iter()
947 .next()
948 .ok_or_else(|| {
949 TransportError::InvalidAddress(format!(
950 "DNS resolution returned no addresses for {}",
951 addr.as_str().unwrap_or("<non-utf8>")
952 ))
953 })
954}
955
956pub(crate) async fn resolve_socket_addrs(
963 addr: &TransportAddr,
964) -> Result<Vec<SocketAddr>, TransportError> {
965 let s = addr
966 .as_str()
967 .ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?;
968
969 if let Ok(sock_addr) = s.parse::<SocketAddr>() {
971 return Ok(vec![sock_addr]);
972 }
973
974 let addrs = tokio::net::lookup_host(s)
976 .await
977 .map_err(|e| {
978 TransportError::InvalidAddress(format!("DNS resolution failed for {}: {}", s, e))
979 })?
980 .collect::<Vec<_>>();
981 if addrs.is_empty() {
982 return Err(TransportError::InvalidAddress(format!(
983 "DNS resolution returned no addresses for {}",
984 s
985 )));
986 }
987 Ok(addrs)
988}