1pub mod tcp;
8pub mod tor;
9pub mod udp;
10#[cfg(feature = "webrtc-transport")]
11pub mod webrtc;
12pub mod websocket;
13
14#[cfg(feature = "sim-transport")]
15pub mod sim;
16
17#[cfg(any(target_os = "linux", target_os = "macos"))]
18pub mod ethernet;
19
20#[cfg(any(target_os = "linux", feature = "host-ble-transport", test))]
21pub mod ble;
22
23mod handle;
24pub(crate) mod link_negotiation;
25mod packet_channel;
26mod resolve;
27mod stream_io;
28
29#[cfg(test)]
30mod tests;
31
32pub use handle::TransportHandle;
33pub(crate) use packet_channel::PacketFastIngressSink;
34pub use packet_channel::{PacketBuffer, PacketRx, PacketTx, ReceivedPacket, packet_channel};
35pub(crate) use resolve::{resolve_socket_addr, resolve_socket_addrs};
36
37use secp256k1::XOnlyPublicKey;
38use std::fmt;
39use std::net::SocketAddr;
40use std::sync::Arc;
41use std::time::Duration;
42use thiserror::Error;
43
44#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
50pub struct TransportId(u32);
51
52impl TransportId {
53 pub fn new(id: u32) -> Self {
55 Self(id)
56 }
57
58 pub fn as_u32(&self) -> u32 {
60 self.0
61 }
62}
63
64impl fmt::Display for TransportId {
65 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
66 write!(f, "transport:{}", self.0)
67 }
68}
69
70#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
72pub struct LinkId(u64);
73
74impl LinkId {
75 pub fn new(id: u64) -> Self {
77 Self(id)
78 }
79
80 pub fn as_u64(&self) -> u64 {
82 self.0
83 }
84}
85
86impl fmt::Display for LinkId {
87 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
88 write!(f, "link:{}", self.0)
89 }
90}
91
92#[derive(Debug, Error)]
98pub enum TransportError {
99 #[error("transport not started")]
100 NotStarted,
101
102 #[error("transport already started")]
103 AlreadyStarted,
104
105 #[error("transport failed to start: {0}")]
106 StartFailed(String),
107
108 #[error("transport address already in use: {address}")]
109 AddressInUse {
110 address: SocketAddr,
111 #[source]
112 source: std::io::Error,
113 },
114
115 #[error("transport shutdown failed: {0}")]
116 ShutdownFailed(String),
117
118 #[error("link failed: {0}")]
119 LinkFailed(String),
120
121 #[error("send failed: {0}")]
122 SendFailed(String),
123
124 #[error("receive failed: {0}")]
125 RecvFailed(String),
126
127 #[error("invalid transport address: {0}")]
128 InvalidAddress(String),
129
130 #[error("mtu exceeded: packet {packet_size} > mtu {mtu}")]
131 MtuExceeded { packet_size: usize, mtu: u16 },
132
133 #[error("transport timeout")]
134 Timeout,
135
136 #[error("connection refused")]
137 ConnectionRefused,
138
139 #[error("transport not supported: {0}")]
140 NotSupported(String),
141
142 #[error("io error: {0}")]
143 Io(#[from] std::io::Error),
144}
145
146impl TransportError {
147 pub(crate) fn bind_failed(address: SocketAddr, source: std::io::Error) -> Self {
150 if source.kind() == std::io::ErrorKind::AddrInUse {
151 Self::AddressInUse { address, source }
152 } else {
153 Self::StartFailed(format!("bind {address} failed: {source}"))
154 }
155 }
156
157 #[cfg(windows)]
161 pub(crate) fn exclusive_bind_failed(address: SocketAddr, source: std::io::Error) -> Self {
162 if source.raw_os_error() == Some(10_013) {
163 return Self::AddressInUse { address, source };
164 }
165 Self::bind_failed(address, source)
166 }
167
168 pub fn is_local_route_unavailable(&self) -> bool {
171 match self {
172 TransportError::Io(error) => is_local_route_error_kind(error.kind()),
173 TransportError::SendFailed(message) => is_local_route_error_text(message),
174 _ => false,
175 }
176 }
177
178 pub(crate) fn requires_local_route_socket_recovery(&self) -> bool {
182 match self {
183 TransportError::Io(error) => error.kind() == std::io::ErrorKind::AddrNotAvailable,
184 TransportError::SendFailed(message) => is_addr_not_available_error_text(message),
185 _ => false,
186 }
187 }
188}
189
190fn is_local_route_error_kind(kind: std::io::ErrorKind) -> bool {
191 matches!(
192 kind,
193 std::io::ErrorKind::NetworkUnreachable
194 | std::io::ErrorKind::HostUnreachable
195 | std::io::ErrorKind::AddrNotAvailable
196 | std::io::ErrorKind::PermissionDenied
197 )
198}
199
200fn is_local_route_error_text(message: &str) -> bool {
201 let lower = message.to_ascii_lowercase();
202 lower.contains("network is unreachable")
203 || lower.contains("no route to host")
204 || lower.contains("host is unreachable")
205 || lower.contains("can't assign requested address")
206 || lower.contains("cannot assign requested address")
207 || lower.contains("operation not permitted")
208 || lower.contains("permission denied")
209 || lower.contains("os error 51")
210 || lower.contains("os error 65")
211 || lower.contains("os error 49")
212 || lower.contains("os error 1")
213}
214
215fn is_addr_not_available_error_text(message: &str) -> bool {
216 let lower = message.to_ascii_lowercase();
217 lower.contains("address not available")
218 || lower.contains("can't assign requested address")
219 || lower.contains("cannot assign requested address")
220 || lower.contains("os error 49")
221 || lower.contains("os error 99")
222 || lower.contains("os error 10049")
223}
224
225#[derive(Clone, Debug, PartialEq, Eq)]
231pub struct TransportType {
232 pub name: &'static str,
234 pub connection_oriented: bool,
236 pub reliable: bool,
238}
239
240impl TransportType {
241 pub const UDP: TransportType = TransportType {
243 name: "udp",
244 connection_oriented: false,
245 reliable: false,
246 };
247
248 pub const TCP: TransportType = TransportType {
250 name: "tcp",
251 connection_oriented: true,
252 reliable: true,
253 };
254
255 pub const ETHERNET: TransportType = TransportType {
257 name: "ethernet",
258 connection_oriented: false,
259 reliable: false,
260 };
261
262 pub const WIFI: TransportType = TransportType {
264 name: "wifi",
265 connection_oriented: false,
266 reliable: false,
267 };
268
269 pub const TOR: TransportType = TransportType {
271 name: "tor",
272 connection_oriented: true,
273 reliable: true,
274 };
275
276 pub const SERIAL: TransportType = TransportType {
278 name: "serial",
279 connection_oriented: false,
280 reliable: true, };
282
283 pub const BLE: TransportType = TransportType {
285 name: "ble",
286 connection_oriented: true,
287 reliable: true, };
289
290 pub const WEBRTC: TransportType = TransportType {
292 name: "webrtc",
293 connection_oriented: true,
294 reliable: false,
295 };
296
297 pub const WEBSOCKET: TransportType = TransportType {
299 name: "websocket",
300 connection_oriented: true,
301 reliable: true,
302 };
303
304 #[cfg(feature = "sim-transport")]
306 pub const SIM: TransportType = TransportType {
307 name: "sim",
308 connection_oriented: false,
309 reliable: false,
310 };
311
312 pub fn is_connectionless(&self) -> bool {
314 !self.connection_oriented
315 }
316}
317
318impl fmt::Display for TransportType {
319 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
320 write!(f, "{}", self.name)
321 }
322}
323
324#[derive(Clone, Copy, Debug, PartialEq, Eq)]
330pub enum TransportState {
331 Configured,
333 Starting,
335 Up,
337 Down,
339 Failed,
341}
342
343impl TransportState {
344 pub fn is_operational(&self) -> bool {
346 matches!(self, TransportState::Up)
347 }
348
349 pub fn can_start(&self) -> bool {
351 matches!(
352 self,
353 TransportState::Configured | TransportState::Down | TransportState::Failed
354 )
355 }
356
357 pub fn is_terminal(&self) -> bool {
359 matches!(self, TransportState::Failed)
360 }
361}
362
363impl fmt::Display for TransportState {
364 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
365 let s = match self {
366 TransportState::Configured => "configured",
367 TransportState::Starting => "starting",
368 TransportState::Up => "up",
369 TransportState::Down => "down",
370 TransportState::Failed => "failed",
371 };
372 write!(f, "{}", s)
373 }
374}
375
376#[derive(Clone, Copy, Debug, PartialEq, Eq)]
382pub enum LinkState {
383 Connecting,
385 Connected,
387 Disconnected,
389 Failed,
391}
392
393impl LinkState {
394 pub fn is_operational(&self) -> bool {
396 matches!(self, LinkState::Connected)
397 }
398
399 pub fn is_terminal(&self) -> bool {
401 matches!(self, LinkState::Disconnected | LinkState::Failed)
402 }
403}
404
405impl fmt::Display for LinkState {
406 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
407 let s = match self {
408 LinkState::Connecting => "connecting",
409 LinkState::Connected => "connected",
410 LinkState::Disconnected => "disconnected",
411 LinkState::Failed => "failed",
412 };
413 write!(f, "{}", s)
414 }
415}
416
417#[derive(Clone, Copy, Debug, PartialEq, Eq)]
419pub enum LinkDirection {
420 Outbound,
422 Inbound,
424}
425
426impl fmt::Display for LinkDirection {
427 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
428 let s = match self {
429 LinkDirection::Outbound => "outbound",
430 LinkDirection::Inbound => "inbound",
431 };
432 write!(f, "{}", s)
433 }
434}
435
436#[derive(Clone, PartialEq, Eq, Hash)]
449pub struct TransportAddr(Arc<[u8]>);
450
451impl TransportAddr {
452 pub fn new(bytes: Vec<u8>) -> Self {
454 Self(bytes.into())
455 }
456
457 pub fn from_bytes(bytes: &[u8]) -> Self {
459 Self(Arc::from(bytes))
460 }
461
462 pub fn from_string(s: &str) -> Self {
464 Self(Arc::from(s.as_bytes()))
465 }
466
467 pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self {
472 match addr {
473 std::net::SocketAddr::V4(addr) => {
474 let octets = addr.ip().octets();
475 let mut buf = Vec::with_capacity(21);
476 push_decimal_u8(&mut buf, octets[0]);
477 buf.push(b'.');
478 push_decimal_u8(&mut buf, octets[1]);
479 buf.push(b'.');
480 push_decimal_u8(&mut buf, octets[2]);
481 buf.push(b'.');
482 push_decimal_u8(&mut buf, octets[3]);
483 buf.push(b':');
484 push_decimal_u16(&mut buf, addr.port());
485 Self(buf.into())
486 }
487 std::net::SocketAddr::V6(addr) => {
488 use std::io::Write;
489 let mut buf = Vec::with_capacity(56);
490 buf.push(b'[');
491 write!(&mut buf, "{}", addr.ip()).expect("Vec<u8>::write_fmt is infallible");
492 buf.push(b']');
493 buf.push(b':');
494 push_decimal_u16(&mut buf, addr.port());
495 Self(buf.into())
496 }
497 }
498 }
499
500 pub fn as_bytes(&self) -> &[u8] {
502 &self.0
503 }
504
505 pub fn as_str(&self) -> Option<&str> {
507 std::str::from_utf8(&self.0).ok()
508 }
509
510 pub fn len(&self) -> usize {
512 self.0.len()
513 }
514
515 pub fn is_empty(&self) -> bool {
517 self.0.is_empty()
518 }
519}
520
521fn push_decimal_u8(buf: &mut Vec<u8>, value: u8) {
522 push_decimal_u16(buf, value as u16);
523}
524
525fn push_decimal_u16(buf: &mut Vec<u8>, value: u16) {
526 if value >= 10_000 {
527 buf.push(b'0' + (value / 10_000) as u8);
528 push_fixed_4_digits(buf, value % 10_000);
529 } else if value >= 1_000 {
530 push_fixed_4_digits(buf, value);
531 } else if value >= 100 {
532 buf.push(b'0' + (value / 100) as u8);
533 push_fixed_2_digits(buf, value % 100);
534 } else if value >= 10 {
535 push_fixed_2_digits(buf, value);
536 } else {
537 buf.push(b'0' + value as u8);
538 }
539}
540
541fn push_fixed_4_digits(buf: &mut Vec<u8>, value: u16) {
542 buf.push(b'0' + (value / 1_000) as u8);
543 push_fixed_3_digits(buf, value % 1_000);
544}
545
546fn push_fixed_3_digits(buf: &mut Vec<u8>, value: u16) {
547 buf.push(b'0' + (value / 100) as u8);
548 push_fixed_2_digits(buf, value % 100);
549}
550
551fn push_fixed_2_digits(buf: &mut Vec<u8>, value: u16) {
552 buf.push(b'0' + (value / 10) as u8);
553 buf.push(b'0' + (value % 10) as u8);
554}
555
556impl fmt::Debug for TransportAddr {
557 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
558 match self.as_str() {
559 Some(s) => write!(f, "TransportAddr(\"{}\")", s),
560 None => write!(f, "TransportAddr({:?})", self.0),
561 }
562 }
563}
564
565impl fmt::Display for TransportAddr {
566 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
567 match self.as_str() {
569 Some(s) => write!(f, "{}", s),
570 None => {
571 for byte in self.0.iter() {
572 write!(f, "{:02x}", byte)?;
573 }
574 Ok(())
575 }
576 }
577 }
578}
579
580impl From<&str> for TransportAddr {
581 fn from(s: &str) -> Self {
582 Self::from_string(s)
583 }
584}
585
586impl From<String> for TransportAddr {
587 fn from(s: String) -> Self {
588 Self(s.into_bytes().into())
589 }
590}
591
592#[derive(Clone, Debug, Default)]
598pub struct LinkStats {
599 pub packets_sent: u64,
601 pub packets_recv: u64,
603 pub bytes_sent: u64,
605 pub bytes_recv: u64,
607 pub last_recv_ms: u64,
609 rtt_estimate: Option<Duration>,
611 pub loss_rate: f32,
613 pub throughput_estimate: u64,
615}
616
617impl LinkStats {
618 pub fn new() -> Self {
620 Self::default()
621 }
622
623 pub fn record_sent(&mut self, bytes: usize) {
625 self.packets_sent += 1;
626 self.bytes_sent += bytes as u64;
627 }
628
629 pub fn record_sent_batch(&mut self, packets: usize, bytes: usize) {
631 self.packets_sent += packets as u64;
632 self.bytes_sent += bytes as u64;
633 }
634
635 pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) {
637 self.packets_recv += 1;
638 self.bytes_recv += bytes as u64;
639 self.last_recv_ms = timestamp_ms;
640 }
641
642 pub fn rtt_estimate(&self) -> Option<Duration> {
644 self.rtt_estimate
645 }
646
647 pub fn update_rtt(&mut self, rtt: Duration) {
651 match self.rtt_estimate {
652 Some(old_rtt) => {
653 let alpha = 0.2;
654 let new_rtt_nanos = (alpha * rtt.as_nanos() as f64
655 + (1.0 - alpha) * old_rtt.as_nanos() as f64)
656 as u64;
657 self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos));
658 }
659 None => {
660 self.rtt_estimate = Some(rtt);
661 }
662 }
663 }
664
665 pub fn time_since_recv(&self, current_time_ms: u64) -> u64 {
667 if self.last_recv_ms == 0 {
668 return u64::MAX;
669 }
670 current_time_ms.saturating_sub(self.last_recv_ms)
671 }
672
673 pub fn reset(&mut self) {
675 *self = Self::default();
676 }
677}
678
679#[derive(Clone, Debug)]
685pub struct Link {
686 link_id: LinkId,
688 transport_id: TransportId,
690 remote_addr: TransportAddr,
692 direction: LinkDirection,
694 state: LinkState,
696 base_rtt: Duration,
698 stats: LinkStats,
700 created_at: u64,
702}
703
704impl Link {
705 pub fn new(
707 link_id: LinkId,
708 transport_id: TransportId,
709 remote_addr: TransportAddr,
710 direction: LinkDirection,
711 base_rtt: Duration,
712 ) -> Self {
713 Self {
714 link_id,
715 transport_id,
716 remote_addr,
717 direction,
718 state: LinkState::Connecting,
719 base_rtt,
720 stats: LinkStats::new(),
721 created_at: 0,
722 }
723 }
724
725 pub fn new_with_timestamp(
727 link_id: LinkId,
728 transport_id: TransportId,
729 remote_addr: TransportAddr,
730 direction: LinkDirection,
731 base_rtt: Duration,
732 created_at: u64,
733 ) -> Self {
734 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
735 link.created_at = created_at;
736 link
737 }
738
739 pub fn connectionless(
744 link_id: LinkId,
745 transport_id: TransportId,
746 remote_addr: TransportAddr,
747 direction: LinkDirection,
748 base_rtt: Duration,
749 ) -> Self {
750 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
751 link.state = LinkState::Connected;
752 link
753 }
754
755 pub fn link_id(&self) -> LinkId {
757 self.link_id
758 }
759
760 pub fn transport_id(&self) -> TransportId {
762 self.transport_id
763 }
764
765 pub fn remote_addr(&self) -> &TransportAddr {
767 &self.remote_addr
768 }
769
770 pub fn direction(&self) -> LinkDirection {
772 self.direction
773 }
774
775 pub fn state(&self) -> LinkState {
777 self.state
778 }
779
780 pub fn base_rtt(&self) -> Duration {
782 self.base_rtt
783 }
784
785 pub fn stats(&self) -> &LinkStats {
787 &self.stats
788 }
789
790 pub fn stats_mut(&mut self) -> &mut LinkStats {
792 &mut self.stats
793 }
794
795 pub fn created_at(&self) -> u64 {
797 self.created_at
798 }
799
800 pub fn set_created_at(&mut self, timestamp: u64) {
802 self.created_at = timestamp;
803 }
804
805 pub fn set_connected(&mut self) {
807 self.state = LinkState::Connected;
808 }
809
810 pub fn set_disconnected(&mut self) {
812 self.state = LinkState::Disconnected;
813 }
814
815 pub fn set_failed(&mut self) {
817 self.state = LinkState::Failed;
818 }
819
820 pub fn is_operational(&self) -> bool {
822 self.state.is_operational()
823 }
824
825 pub fn is_terminal(&self) -> bool {
827 self.state.is_terminal()
828 }
829
830 pub fn effective_rtt(&self) -> Duration {
832 self.stats.rtt_estimate().unwrap_or(self.base_rtt)
833 }
834
835 pub fn age(&self, current_time_ms: u64) -> u64 {
837 if self.created_at == 0 {
838 return 0;
839 }
840 current_time_ms.saturating_sub(self.created_at)
841 }
842}
843
844#[derive(Clone, Debug)]
850pub struct DiscoveredPeer {
851 pub transport_id: TransportId,
853 pub addr: TransportAddr,
855 pub pubkey_hint: Option<XOnlyPublicKey>,
857}
858
859impl DiscoveredPeer {
860 pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self {
862 Self {
863 transport_id,
864 addr,
865 pubkey_hint: None,
866 }
867 }
868
869 pub fn with_hint(
871 transport_id: TransportId,
872 addr: TransportAddr,
873 pubkey: XOnlyPublicKey,
874 ) -> Self {
875 Self {
876 transport_id,
877 addr,
878 pubkey_hint: Some(pubkey),
879 }
880 }
881}
882
883pub trait Transport {
892 fn transport_id(&self) -> TransportId;
894
895 fn transport_type(&self) -> &TransportType;
897
898 fn state(&self) -> TransportState;
900
901 fn mtu(&self) -> u16;
903
904 fn link_mtu(&self, addr: &TransportAddr) -> u16 {
910 let _ = addr;
911 self.mtu()
912 }
913
914 fn start(&mut self) -> Result<(), TransportError>;
916
917 fn stop(&mut self) -> Result<(), TransportError>;
919
920 fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>;
922
923 fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError>;
925
926 fn auto_connect(&self) -> bool {
929 false
930 }
931
932 fn accept_connections(&self) -> bool {
935 true
936 }
937
938 fn close_connection(&self, _addr: &TransportAddr) {
944 }
946}
947
948#[derive(Clone, Debug, PartialEq, Eq)]
957pub enum ConnectionState {
958 None,
960 Connecting,
962 Connected,
964 Failed(String),
966}
967
968#[derive(Clone, Debug, Default)]
977pub struct TransportCongestion {
978 pub recv_drops: Option<u64>,
981 pub socket_recv_drops: Option<u64>,
984 pub namespace_recv_drops: Option<u64>,
988}