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::received_timestamp_ms;
30pub use packet_channel::{PacketRx, PacketTx, ReceivedPacket, packet_channel};
31
32use secp256k1::XOnlyPublicKey;
33use std::fmt;
34use std::net::SocketAddr;
35use std::sync::Arc;
36use std::time::Duration;
37use thiserror::Error;
38
39#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
45pub struct TransportId(u32);
46
47impl TransportId {
48 pub fn new(id: u32) -> Self {
50 Self(id)
51 }
52
53 pub fn as_u32(&self) -> u32 {
55 self.0
56 }
57}
58
59impl fmt::Display for TransportId {
60 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
61 write!(f, "transport:{}", self.0)
62 }
63}
64
65#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
67pub struct LinkId(u64);
68
69impl LinkId {
70 pub fn new(id: u64) -> Self {
72 Self(id)
73 }
74
75 pub fn as_u64(&self) -> u64 {
77 self.0
78 }
79}
80
81impl fmt::Display for LinkId {
82 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
83 write!(f, "link:{}", self.0)
84 }
85}
86
87#[derive(Debug, Error)]
93pub enum TransportError {
94 #[error("transport not started")]
95 NotStarted,
96
97 #[error("transport already started")]
98 AlreadyStarted,
99
100 #[error("transport failed to start: {0}")]
101 StartFailed(String),
102
103 #[error("transport shutdown failed: {0}")]
104 ShutdownFailed(String),
105
106 #[error("link failed: {0}")]
107 LinkFailed(String),
108
109 #[error("send failed: {0}")]
110 SendFailed(String),
111
112 #[error("receive failed: {0}")]
113 RecvFailed(String),
114
115 #[error("invalid transport address: {0}")]
116 InvalidAddress(String),
117
118 #[error("mtu exceeded: packet {packet_size} > mtu {mtu}")]
119 MtuExceeded { packet_size: usize, mtu: u16 },
120
121 #[error("transport timeout")]
122 Timeout,
123
124 #[error("connection refused")]
125 ConnectionRefused,
126
127 #[error("transport not supported: {0}")]
128 NotSupported(String),
129
130 #[error("io error: {0}")]
131 Io(#[from] std::io::Error),
132}
133
134impl TransportError {
135 pub fn is_local_route_unavailable(&self) -> bool {
138 match self {
139 TransportError::Io(error) => is_local_route_error_kind(error.kind()),
140 TransportError::SendFailed(message) => is_local_route_error_text(message),
141 _ => false,
142 }
143 }
144}
145
146fn is_local_route_error_kind(kind: std::io::ErrorKind) -> bool {
147 matches!(
148 kind,
149 std::io::ErrorKind::NetworkUnreachable
150 | std::io::ErrorKind::HostUnreachable
151 | std::io::ErrorKind::AddrNotAvailable
152 )
153}
154
155fn is_local_route_error_text(message: &str) -> bool {
156 let lower = message.to_ascii_lowercase();
157 lower.contains("network is unreachable")
158 || lower.contains("no route to host")
159 || lower.contains("host is unreachable")
160 || lower.contains("can't assign requested address")
161 || lower.contains("cannot assign requested address")
162 || lower.contains("os error 51")
163 || lower.contains("os error 65")
164 || lower.contains("os error 49")
165}
166
167#[derive(Clone, Debug, PartialEq, Eq)]
173pub struct TransportType {
174 pub name: &'static str,
176 pub connection_oriented: bool,
178 pub reliable: bool,
180}
181
182impl TransportType {
183 pub const UDP: TransportType = TransportType {
185 name: "udp",
186 connection_oriented: false,
187 reliable: false,
188 };
189
190 pub const TCP: TransportType = TransportType {
192 name: "tcp",
193 connection_oriented: true,
194 reliable: true,
195 };
196
197 pub const ETHERNET: TransportType = TransportType {
199 name: "ethernet",
200 connection_oriented: false,
201 reliable: false,
202 };
203
204 pub const WIFI: TransportType = TransportType {
206 name: "wifi",
207 connection_oriented: false,
208 reliable: false,
209 };
210
211 pub const TOR: TransportType = TransportType {
213 name: "tor",
214 connection_oriented: true,
215 reliable: true,
216 };
217
218 pub const SERIAL: TransportType = TransportType {
220 name: "serial",
221 connection_oriented: false,
222 reliable: true, };
224
225 pub const BLE: TransportType = TransportType {
227 name: "ble",
228 connection_oriented: true,
229 reliable: true, };
231
232 pub const WEBRTC: TransportType = TransportType {
234 name: "webrtc",
235 connection_oriented: true,
236 reliable: false,
237 };
238
239 #[cfg(feature = "sim-transport")]
241 pub const SIM: TransportType = TransportType {
242 name: "sim",
243 connection_oriented: false,
244 reliable: false,
245 };
246
247 pub fn is_connectionless(&self) -> bool {
249 !self.connection_oriented
250 }
251}
252
253impl fmt::Display for TransportType {
254 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
255 write!(f, "{}", self.name)
256 }
257}
258
259#[derive(Clone, Copy, Debug, PartialEq, Eq)]
265pub enum TransportState {
266 Configured,
268 Starting,
270 Up,
272 Down,
274 Failed,
276}
277
278impl TransportState {
279 pub fn is_operational(&self) -> bool {
281 matches!(self, TransportState::Up)
282 }
283
284 pub fn can_start(&self) -> bool {
286 matches!(
287 self,
288 TransportState::Configured | TransportState::Down | TransportState::Failed
289 )
290 }
291
292 pub fn is_terminal(&self) -> bool {
294 matches!(self, TransportState::Failed)
295 }
296}
297
298impl fmt::Display for TransportState {
299 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
300 let s = match self {
301 TransportState::Configured => "configured",
302 TransportState::Starting => "starting",
303 TransportState::Up => "up",
304 TransportState::Down => "down",
305 TransportState::Failed => "failed",
306 };
307 write!(f, "{}", s)
308 }
309}
310
311#[derive(Clone, Copy, Debug, PartialEq, Eq)]
317pub enum LinkState {
318 Connecting,
320 Connected,
322 Disconnected,
324 Failed,
326}
327
328impl LinkState {
329 pub fn is_operational(&self) -> bool {
331 matches!(self, LinkState::Connected)
332 }
333
334 pub fn is_terminal(&self) -> bool {
336 matches!(self, LinkState::Disconnected | LinkState::Failed)
337 }
338}
339
340impl fmt::Display for LinkState {
341 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
342 let s = match self {
343 LinkState::Connecting => "connecting",
344 LinkState::Connected => "connected",
345 LinkState::Disconnected => "disconnected",
346 LinkState::Failed => "failed",
347 };
348 write!(f, "{}", s)
349 }
350}
351
352#[derive(Clone, Copy, Debug, PartialEq, Eq)]
354pub enum LinkDirection {
355 Outbound,
357 Inbound,
359}
360
361impl fmt::Display for LinkDirection {
362 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
363 let s = match self {
364 LinkDirection::Outbound => "outbound",
365 LinkDirection::Inbound => "inbound",
366 };
367 write!(f, "{}", s)
368 }
369}
370
371#[derive(Clone, PartialEq, Eq, Hash)]
384pub struct TransportAddr(Arc<[u8]>);
385
386impl TransportAddr {
387 pub fn new(bytes: Vec<u8>) -> Self {
389 Self(bytes.into())
390 }
391
392 pub fn from_bytes(bytes: &[u8]) -> Self {
394 Self(Arc::from(bytes))
395 }
396
397 pub fn from_string(s: &str) -> Self {
399 Self(Arc::from(s.as_bytes()))
400 }
401
402 pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self {
414 use std::io::Write;
415 let mut buf = Vec::with_capacity(56);
419 write!(&mut buf, "{addr}").expect("Vec<u8>::write_fmt is infallible");
423 Self(buf.into())
424 }
425
426 pub fn as_bytes(&self) -> &[u8] {
428 &self.0
429 }
430
431 pub fn as_str(&self) -> Option<&str> {
433 std::str::from_utf8(&self.0).ok()
434 }
435
436 pub fn len(&self) -> usize {
438 self.0.len()
439 }
440
441 pub fn is_empty(&self) -> bool {
443 self.0.is_empty()
444 }
445}
446
447impl fmt::Debug for TransportAddr {
448 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
449 match self.as_str() {
450 Some(s) => write!(f, "TransportAddr(\"{}\")", s),
451 None => write!(f, "TransportAddr({:?})", self.0),
452 }
453 }
454}
455
456impl fmt::Display for TransportAddr {
457 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
458 match self.as_str() {
460 Some(s) => write!(f, "{}", s),
461 None => {
462 for byte in self.0.iter() {
463 write!(f, "{:02x}", byte)?;
464 }
465 Ok(())
466 }
467 }
468 }
469}
470
471impl From<&str> for TransportAddr {
472 fn from(s: &str) -> Self {
473 Self::from_string(s)
474 }
475}
476
477impl From<String> for TransportAddr {
478 fn from(s: String) -> Self {
479 Self(s.into_bytes().into())
480 }
481}
482
483#[derive(Clone, Debug, Default)]
489pub struct LinkStats {
490 pub packets_sent: u64,
492 pub packets_recv: u64,
494 pub bytes_sent: u64,
496 pub bytes_recv: u64,
498 pub last_recv_ms: u64,
500 rtt_estimate: Option<Duration>,
502 pub loss_rate: f32,
504 pub throughput_estimate: u64,
506}
507
508impl LinkStats {
509 pub fn new() -> Self {
511 Self::default()
512 }
513
514 pub fn record_sent(&mut self, bytes: usize) {
516 self.packets_sent += 1;
517 self.bytes_sent += bytes as u64;
518 }
519
520 pub fn record_sent_batch(&mut self, packets: usize, bytes: usize) {
522 self.packets_sent += packets as u64;
523 self.bytes_sent += bytes as u64;
524 }
525
526 pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) {
528 self.packets_recv += 1;
529 self.bytes_recv += bytes as u64;
530 self.last_recv_ms = timestamp_ms;
531 }
532
533 pub fn rtt_estimate(&self) -> Option<Duration> {
535 self.rtt_estimate
536 }
537
538 pub fn update_rtt(&mut self, rtt: Duration) {
542 match self.rtt_estimate {
543 Some(old_rtt) => {
544 let alpha = 0.2;
545 let new_rtt_nanos = (alpha * rtt.as_nanos() as f64
546 + (1.0 - alpha) * old_rtt.as_nanos() as f64)
547 as u64;
548 self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos));
549 }
550 None => {
551 self.rtt_estimate = Some(rtt);
552 }
553 }
554 }
555
556 pub fn time_since_recv(&self, current_time_ms: u64) -> u64 {
558 if self.last_recv_ms == 0 {
559 return u64::MAX;
560 }
561 current_time_ms.saturating_sub(self.last_recv_ms)
562 }
563
564 pub fn reset(&mut self) {
566 *self = Self::default();
567 }
568}
569
570#[derive(Clone, Debug)]
576pub struct Link {
577 link_id: LinkId,
579 transport_id: TransportId,
581 remote_addr: TransportAddr,
583 direction: LinkDirection,
585 state: LinkState,
587 base_rtt: Duration,
589 stats: LinkStats,
591 created_at: u64,
593}
594
595impl Link {
596 pub fn new(
598 link_id: LinkId,
599 transport_id: TransportId,
600 remote_addr: TransportAddr,
601 direction: LinkDirection,
602 base_rtt: Duration,
603 ) -> Self {
604 Self {
605 link_id,
606 transport_id,
607 remote_addr,
608 direction,
609 state: LinkState::Connecting,
610 base_rtt,
611 stats: LinkStats::new(),
612 created_at: 0,
613 }
614 }
615
616 pub fn new_with_timestamp(
618 link_id: LinkId,
619 transport_id: TransportId,
620 remote_addr: TransportAddr,
621 direction: LinkDirection,
622 base_rtt: Duration,
623 created_at: u64,
624 ) -> Self {
625 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
626 link.created_at = created_at;
627 link
628 }
629
630 pub fn connectionless(
635 link_id: LinkId,
636 transport_id: TransportId,
637 remote_addr: TransportAddr,
638 direction: LinkDirection,
639 base_rtt: Duration,
640 ) -> Self {
641 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
642 link.state = LinkState::Connected;
643 link
644 }
645
646 pub fn link_id(&self) -> LinkId {
648 self.link_id
649 }
650
651 pub fn transport_id(&self) -> TransportId {
653 self.transport_id
654 }
655
656 pub fn remote_addr(&self) -> &TransportAddr {
658 &self.remote_addr
659 }
660
661 pub fn direction(&self) -> LinkDirection {
663 self.direction
664 }
665
666 pub fn state(&self) -> LinkState {
668 self.state
669 }
670
671 pub fn base_rtt(&self) -> Duration {
673 self.base_rtt
674 }
675
676 pub fn stats(&self) -> &LinkStats {
678 &self.stats
679 }
680
681 pub fn stats_mut(&mut self) -> &mut LinkStats {
683 &mut self.stats
684 }
685
686 pub fn created_at(&self) -> u64 {
688 self.created_at
689 }
690
691 pub fn set_created_at(&mut self, timestamp: u64) {
693 self.created_at = timestamp;
694 }
695
696 pub fn set_connected(&mut self) {
698 self.state = LinkState::Connected;
699 }
700
701 pub fn set_disconnected(&mut self) {
703 self.state = LinkState::Disconnected;
704 }
705
706 pub fn set_failed(&mut self) {
708 self.state = LinkState::Failed;
709 }
710
711 pub fn is_operational(&self) -> bool {
713 self.state.is_operational()
714 }
715
716 pub fn is_terminal(&self) -> bool {
718 self.state.is_terminal()
719 }
720
721 pub fn effective_rtt(&self) -> Duration {
723 self.stats.rtt_estimate().unwrap_or(self.base_rtt)
724 }
725
726 pub fn age(&self, current_time_ms: u64) -> u64 {
728 if self.created_at == 0 {
729 return 0;
730 }
731 current_time_ms.saturating_sub(self.created_at)
732 }
733}
734
735#[derive(Clone, Debug)]
741pub struct DiscoveredPeer {
742 pub transport_id: TransportId,
744 pub addr: TransportAddr,
746 pub pubkey_hint: Option<XOnlyPublicKey>,
748}
749
750impl DiscoveredPeer {
751 pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self {
753 Self {
754 transport_id,
755 addr,
756 pubkey_hint: None,
757 }
758 }
759
760 pub fn with_hint(
762 transport_id: TransportId,
763 addr: TransportAddr,
764 pubkey: XOnlyPublicKey,
765 ) -> Self {
766 Self {
767 transport_id,
768 addr,
769 pubkey_hint: Some(pubkey),
770 }
771 }
772}
773
774pub trait Transport {
783 fn transport_id(&self) -> TransportId;
785
786 fn transport_type(&self) -> &TransportType;
788
789 fn state(&self) -> TransportState;
791
792 fn mtu(&self) -> u16;
794
795 fn link_mtu(&self, addr: &TransportAddr) -> u16 {
801 let _ = addr;
802 self.mtu()
803 }
804
805 fn start(&mut self) -> Result<(), TransportError>;
807
808 fn stop(&mut self) -> Result<(), TransportError>;
810
811 fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>;
813
814 fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError>;
816
817 fn auto_connect(&self) -> bool {
820 false
821 }
822
823 fn accept_connections(&self) -> bool {
826 true
827 }
828
829 fn close_connection(&self, _addr: &TransportAddr) {
835 }
837}
838
839#[derive(Clone, Debug, PartialEq, Eq)]
848pub enum ConnectionState {
849 None,
851 Connecting,
853 Connected,
855 Failed(String),
857}
858
859#[derive(Clone, Debug, Default)]
868pub struct TransportCongestion {
869 pub recv_drops: Option<u64>,
872}
873
874pub(crate) async fn resolve_socket_addr(
885 addr: &TransportAddr,
886) -> Result<SocketAddr, TransportError> {
887 resolve_socket_addrs(addr)
888 .await?
889 .into_iter()
890 .next()
891 .ok_or_else(|| {
892 TransportError::InvalidAddress(format!(
893 "DNS resolution returned no addresses for {}",
894 addr.as_str().unwrap_or("<non-utf8>")
895 ))
896 })
897}
898
899pub(crate) async fn resolve_socket_addrs(
906 addr: &TransportAddr,
907) -> Result<Vec<SocketAddr>, TransportError> {
908 let s = addr
909 .as_str()
910 .ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?;
911
912 if let Ok(sock_addr) = s.parse::<SocketAddr>() {
914 return Ok(vec![sock_addr]);
915 }
916
917 let addrs = tokio::net::lookup_host(s)
919 .await
920 .map_err(|e| {
921 TransportError::InvalidAddress(format!("DNS resolution failed for {}: {}", s, e))
922 })?
923 .collect::<Vec<_>>();
924 if addrs.is_empty() {
925 return Err(TransportError::InvalidAddress(format!(
926 "DNS resolution returned no addresses for {}",
927 s
928 )));
929 }
930 Ok(addrs)
931}