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;
30pub use packet_channel::{PacketBuffer, 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 | std::io::ErrorKind::PermissionDenied
153 )
154}
155
156fn is_local_route_error_text(message: &str) -> bool {
157 let lower = message.to_ascii_lowercase();
158 lower.contains("network is unreachable")
159 || lower.contains("no route to host")
160 || lower.contains("host is unreachable")
161 || lower.contains("can't assign requested address")
162 || lower.contains("cannot assign requested address")
163 || lower.contains("operation not permitted")
164 || lower.contains("permission denied")
165 || lower.contains("os error 51")
166 || lower.contains("os error 65")
167 || lower.contains("os error 49")
168 || lower.contains("os error 1")
169}
170
171#[derive(Clone, Debug, PartialEq, Eq)]
177pub struct TransportType {
178 pub name: &'static str,
180 pub connection_oriented: bool,
182 pub reliable: bool,
184}
185
186impl TransportType {
187 pub const UDP: TransportType = TransportType {
189 name: "udp",
190 connection_oriented: false,
191 reliable: false,
192 };
193
194 pub const TCP: TransportType = TransportType {
196 name: "tcp",
197 connection_oriented: true,
198 reliable: true,
199 };
200
201 pub const ETHERNET: TransportType = TransportType {
203 name: "ethernet",
204 connection_oriented: false,
205 reliable: false,
206 };
207
208 pub const WIFI: TransportType = TransportType {
210 name: "wifi",
211 connection_oriented: false,
212 reliable: false,
213 };
214
215 pub const TOR: TransportType = TransportType {
217 name: "tor",
218 connection_oriented: true,
219 reliable: true,
220 };
221
222 pub const SERIAL: TransportType = TransportType {
224 name: "serial",
225 connection_oriented: false,
226 reliable: true, };
228
229 pub const BLE: TransportType = TransportType {
231 name: "ble",
232 connection_oriented: true,
233 reliable: true, };
235
236 pub const WEBRTC: TransportType = TransportType {
238 name: "webrtc",
239 connection_oriented: true,
240 reliable: false,
241 };
242
243 #[cfg(feature = "sim-transport")]
245 pub const SIM: TransportType = TransportType {
246 name: "sim",
247 connection_oriented: false,
248 reliable: false,
249 };
250
251 pub fn is_connectionless(&self) -> bool {
253 !self.connection_oriented
254 }
255}
256
257impl fmt::Display for TransportType {
258 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
259 write!(f, "{}", self.name)
260 }
261}
262
263#[derive(Clone, Copy, Debug, PartialEq, Eq)]
269pub enum TransportState {
270 Configured,
272 Starting,
274 Up,
276 Down,
278 Failed,
280}
281
282impl TransportState {
283 pub fn is_operational(&self) -> bool {
285 matches!(self, TransportState::Up)
286 }
287
288 pub fn can_start(&self) -> bool {
290 matches!(
291 self,
292 TransportState::Configured | TransportState::Down | TransportState::Failed
293 )
294 }
295
296 pub fn is_terminal(&self) -> bool {
298 matches!(self, TransportState::Failed)
299 }
300}
301
302impl fmt::Display for TransportState {
303 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
304 let s = match self {
305 TransportState::Configured => "configured",
306 TransportState::Starting => "starting",
307 TransportState::Up => "up",
308 TransportState::Down => "down",
309 TransportState::Failed => "failed",
310 };
311 write!(f, "{}", s)
312 }
313}
314
315#[derive(Clone, Copy, Debug, PartialEq, Eq)]
321pub enum LinkState {
322 Connecting,
324 Connected,
326 Disconnected,
328 Failed,
330}
331
332impl LinkState {
333 pub fn is_operational(&self) -> bool {
335 matches!(self, LinkState::Connected)
336 }
337
338 pub fn is_terminal(&self) -> bool {
340 matches!(self, LinkState::Disconnected | LinkState::Failed)
341 }
342}
343
344impl fmt::Display for LinkState {
345 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
346 let s = match self {
347 LinkState::Connecting => "connecting",
348 LinkState::Connected => "connected",
349 LinkState::Disconnected => "disconnected",
350 LinkState::Failed => "failed",
351 };
352 write!(f, "{}", s)
353 }
354}
355
356#[derive(Clone, Copy, Debug, PartialEq, Eq)]
358pub enum LinkDirection {
359 Outbound,
361 Inbound,
363}
364
365impl fmt::Display for LinkDirection {
366 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
367 let s = match self {
368 LinkDirection::Outbound => "outbound",
369 LinkDirection::Inbound => "inbound",
370 };
371 write!(f, "{}", s)
372 }
373}
374
375#[derive(Clone, PartialEq, Eq, Hash)]
388pub struct TransportAddr(Arc<[u8]>);
389
390impl TransportAddr {
391 pub fn new(bytes: Vec<u8>) -> Self {
393 Self(bytes.into())
394 }
395
396 pub fn from_bytes(bytes: &[u8]) -> Self {
398 Self(Arc::from(bytes))
399 }
400
401 pub fn from_string(s: &str) -> Self {
403 Self(Arc::from(s.as_bytes()))
404 }
405
406 pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self {
411 match addr {
412 std::net::SocketAddr::V4(addr) => {
413 let octets = addr.ip().octets();
414 let mut buf = Vec::with_capacity(21);
415 push_decimal_u8(&mut buf, octets[0]);
416 buf.push(b'.');
417 push_decimal_u8(&mut buf, octets[1]);
418 buf.push(b'.');
419 push_decimal_u8(&mut buf, octets[2]);
420 buf.push(b'.');
421 push_decimal_u8(&mut buf, octets[3]);
422 buf.push(b':');
423 push_decimal_u16(&mut buf, addr.port());
424 Self(buf.into())
425 }
426 std::net::SocketAddr::V6(addr) => {
427 use std::io::Write;
428 let mut buf = Vec::with_capacity(56);
429 buf.push(b'[');
430 write!(&mut buf, "{}", addr.ip()).expect("Vec<u8>::write_fmt is infallible");
431 buf.push(b']');
432 buf.push(b':');
433 push_decimal_u16(&mut buf, addr.port());
434 Self(buf.into())
435 }
436 }
437 }
438
439 pub fn as_bytes(&self) -> &[u8] {
441 &self.0
442 }
443
444 pub fn as_str(&self) -> Option<&str> {
446 std::str::from_utf8(&self.0).ok()
447 }
448
449 pub fn len(&self) -> usize {
451 self.0.len()
452 }
453
454 pub fn is_empty(&self) -> bool {
456 self.0.is_empty()
457 }
458}
459
460fn push_decimal_u8(buf: &mut Vec<u8>, value: u8) {
461 push_decimal_u16(buf, value as u16);
462}
463
464fn push_decimal_u16(buf: &mut Vec<u8>, value: u16) {
465 if value >= 10_000 {
466 buf.push(b'0' + (value / 10_000) as u8);
467 push_fixed_4_digits(buf, value % 10_000);
468 } else if value >= 1_000 {
469 push_fixed_4_digits(buf, value);
470 } else if value >= 100 {
471 buf.push(b'0' + (value / 100) as u8);
472 push_fixed_2_digits(buf, value % 100);
473 } else if value >= 10 {
474 push_fixed_2_digits(buf, value);
475 } else {
476 buf.push(b'0' + value as u8);
477 }
478}
479
480fn push_fixed_4_digits(buf: &mut Vec<u8>, value: u16) {
481 buf.push(b'0' + (value / 1_000) as u8);
482 push_fixed_3_digits(buf, value % 1_000);
483}
484
485fn push_fixed_3_digits(buf: &mut Vec<u8>, value: u16) {
486 buf.push(b'0' + (value / 100) as u8);
487 push_fixed_2_digits(buf, value % 100);
488}
489
490fn push_fixed_2_digits(buf: &mut Vec<u8>, value: u16) {
491 buf.push(b'0' + (value / 10) as u8);
492 buf.push(b'0' + (value % 10) as u8);
493}
494
495impl fmt::Debug for TransportAddr {
496 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
497 match self.as_str() {
498 Some(s) => write!(f, "TransportAddr(\"{}\")", s),
499 None => write!(f, "TransportAddr({:?})", self.0),
500 }
501 }
502}
503
504impl fmt::Display for TransportAddr {
505 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
506 match self.as_str() {
508 Some(s) => write!(f, "{}", s),
509 None => {
510 for byte in self.0.iter() {
511 write!(f, "{:02x}", byte)?;
512 }
513 Ok(())
514 }
515 }
516 }
517}
518
519impl From<&str> for TransportAddr {
520 fn from(s: &str) -> Self {
521 Self::from_string(s)
522 }
523}
524
525impl From<String> for TransportAddr {
526 fn from(s: String) -> Self {
527 Self(s.into_bytes().into())
528 }
529}
530
531#[derive(Clone, Debug, Default)]
537pub struct LinkStats {
538 pub packets_sent: u64,
540 pub packets_recv: u64,
542 pub bytes_sent: u64,
544 pub bytes_recv: u64,
546 pub last_recv_ms: u64,
548 rtt_estimate: Option<Duration>,
550 pub loss_rate: f32,
552 pub throughput_estimate: u64,
554}
555
556impl LinkStats {
557 pub fn new() -> Self {
559 Self::default()
560 }
561
562 pub fn record_sent(&mut self, bytes: usize) {
564 self.packets_sent += 1;
565 self.bytes_sent += bytes as u64;
566 }
567
568 pub fn record_sent_batch(&mut self, packets: usize, bytes: usize) {
570 self.packets_sent += packets as u64;
571 self.bytes_sent += bytes as u64;
572 }
573
574 pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) {
576 self.packets_recv += 1;
577 self.bytes_recv += bytes as u64;
578 self.last_recv_ms = timestamp_ms;
579 }
580
581 pub fn rtt_estimate(&self) -> Option<Duration> {
583 self.rtt_estimate
584 }
585
586 pub fn update_rtt(&mut self, rtt: Duration) {
590 match self.rtt_estimate {
591 Some(old_rtt) => {
592 let alpha = 0.2;
593 let new_rtt_nanos = (alpha * rtt.as_nanos() as f64
594 + (1.0 - alpha) * old_rtt.as_nanos() as f64)
595 as u64;
596 self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos));
597 }
598 None => {
599 self.rtt_estimate = Some(rtt);
600 }
601 }
602 }
603
604 pub fn time_since_recv(&self, current_time_ms: u64) -> u64 {
606 if self.last_recv_ms == 0 {
607 return u64::MAX;
608 }
609 current_time_ms.saturating_sub(self.last_recv_ms)
610 }
611
612 pub fn reset(&mut self) {
614 *self = Self::default();
615 }
616}
617
618#[derive(Clone, Debug)]
624pub struct Link {
625 link_id: LinkId,
627 transport_id: TransportId,
629 remote_addr: TransportAddr,
631 direction: LinkDirection,
633 state: LinkState,
635 base_rtt: Duration,
637 stats: LinkStats,
639 created_at: u64,
641}
642
643impl Link {
644 pub fn new(
646 link_id: LinkId,
647 transport_id: TransportId,
648 remote_addr: TransportAddr,
649 direction: LinkDirection,
650 base_rtt: Duration,
651 ) -> Self {
652 Self {
653 link_id,
654 transport_id,
655 remote_addr,
656 direction,
657 state: LinkState::Connecting,
658 base_rtt,
659 stats: LinkStats::new(),
660 created_at: 0,
661 }
662 }
663
664 pub fn new_with_timestamp(
666 link_id: LinkId,
667 transport_id: TransportId,
668 remote_addr: TransportAddr,
669 direction: LinkDirection,
670 base_rtt: Duration,
671 created_at: u64,
672 ) -> Self {
673 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
674 link.created_at = created_at;
675 link
676 }
677
678 pub fn connectionless(
683 link_id: LinkId,
684 transport_id: TransportId,
685 remote_addr: TransportAddr,
686 direction: LinkDirection,
687 base_rtt: Duration,
688 ) -> Self {
689 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
690 link.state = LinkState::Connected;
691 link
692 }
693
694 pub fn link_id(&self) -> LinkId {
696 self.link_id
697 }
698
699 pub fn transport_id(&self) -> TransportId {
701 self.transport_id
702 }
703
704 pub fn remote_addr(&self) -> &TransportAddr {
706 &self.remote_addr
707 }
708
709 pub fn direction(&self) -> LinkDirection {
711 self.direction
712 }
713
714 pub fn state(&self) -> LinkState {
716 self.state
717 }
718
719 pub fn base_rtt(&self) -> Duration {
721 self.base_rtt
722 }
723
724 pub fn stats(&self) -> &LinkStats {
726 &self.stats
727 }
728
729 pub fn stats_mut(&mut self) -> &mut LinkStats {
731 &mut self.stats
732 }
733
734 pub fn created_at(&self) -> u64 {
736 self.created_at
737 }
738
739 pub fn set_created_at(&mut self, timestamp: u64) {
741 self.created_at = timestamp;
742 }
743
744 pub fn set_connected(&mut self) {
746 self.state = LinkState::Connected;
747 }
748
749 pub fn set_disconnected(&mut self) {
751 self.state = LinkState::Disconnected;
752 }
753
754 pub fn set_failed(&mut self) {
756 self.state = LinkState::Failed;
757 }
758
759 pub fn is_operational(&self) -> bool {
761 self.state.is_operational()
762 }
763
764 pub fn is_terminal(&self) -> bool {
766 self.state.is_terminal()
767 }
768
769 pub fn effective_rtt(&self) -> Duration {
771 self.stats.rtt_estimate().unwrap_or(self.base_rtt)
772 }
773
774 pub fn age(&self, current_time_ms: u64) -> u64 {
776 if self.created_at == 0 {
777 return 0;
778 }
779 current_time_ms.saturating_sub(self.created_at)
780 }
781}
782
783#[derive(Clone, Debug)]
789pub struct DiscoveredPeer {
790 pub transport_id: TransportId,
792 pub addr: TransportAddr,
794 pub pubkey_hint: Option<XOnlyPublicKey>,
796}
797
798impl DiscoveredPeer {
799 pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self {
801 Self {
802 transport_id,
803 addr,
804 pubkey_hint: None,
805 }
806 }
807
808 pub fn with_hint(
810 transport_id: TransportId,
811 addr: TransportAddr,
812 pubkey: XOnlyPublicKey,
813 ) -> Self {
814 Self {
815 transport_id,
816 addr,
817 pubkey_hint: Some(pubkey),
818 }
819 }
820}
821
822pub trait Transport {
831 fn transport_id(&self) -> TransportId;
833
834 fn transport_type(&self) -> &TransportType;
836
837 fn state(&self) -> TransportState;
839
840 fn mtu(&self) -> u16;
842
843 fn link_mtu(&self, addr: &TransportAddr) -> u16 {
849 let _ = addr;
850 self.mtu()
851 }
852
853 fn start(&mut self) -> Result<(), TransportError>;
855
856 fn stop(&mut self) -> Result<(), TransportError>;
858
859 fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>;
861
862 fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError>;
864
865 fn auto_connect(&self) -> bool {
868 false
869 }
870
871 fn accept_connections(&self) -> bool {
874 true
875 }
876
877 fn close_connection(&self, _addr: &TransportAddr) {
883 }
885}
886
887#[derive(Clone, Debug, PartialEq, Eq)]
896pub enum ConnectionState {
897 None,
899 Connecting,
901 Connected,
903 Failed(String),
905}
906
907#[derive(Clone, Debug, Default)]
916pub struct TransportCongestion {
917 pub recv_drops: Option<u64>,
920 pub socket_recv_drops: Option<u64>,
923 pub namespace_recv_drops: Option<u64>,
927}
928
929pub(crate) async fn resolve_socket_addr(
940 addr: &TransportAddr,
941) -> Result<SocketAddr, TransportError> {
942 resolve_socket_addrs(addr)
943 .await?
944 .into_iter()
945 .next()
946 .ok_or_else(|| {
947 TransportError::InvalidAddress(format!(
948 "DNS resolution returned no addresses for {}",
949 addr.as_str().unwrap_or("<non-utf8>")
950 ))
951 })
952}
953
954pub(crate) async fn resolve_socket_addrs(
961 addr: &TransportAddr,
962) -> Result<Vec<SocketAddr>, TransportError> {
963 let s = addr
964 .as_str()
965 .ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?;
966
967 if let Ok(sock_addr) = s.parse::<SocketAddr>() {
969 return Ok(vec![sock_addr]);
970 }
971
972 let addrs = tokio::net::lookup_host(s)
974 .await
975 .map_err(|e| {
976 TransportError::InvalidAddress(format!("DNS resolution failed for {}: {}", s, e))
977 })?
978 .collect::<Vec<_>>();
979 if addrs.is_empty() {
980 return Err(TransportError::InvalidAddress(format!(
981 "DNS resolution returned no addresses for {}",
982 s
983 )));
984 }
985 Ok(addrs)
986}