1pub mod nostr_relay;
8pub mod tcp;
9pub mod tor;
10pub mod udp;
11#[cfg(feature = "webrtc-transport")]
12pub mod webrtc;
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(target_os = "linux")]
21pub mod ble;
22
23mod handle;
24pub(crate) mod link_negotiation;
25mod packet_channel;
26
27#[cfg(test)]
28mod tests;
29
30pub use handle::TransportHandle;
31pub(crate) use packet_channel::PacketFastIngressSink;
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 address already in use: {address}")]
106 AddressInUse {
107 address: SocketAddr,
108 #[source]
109 source: std::io::Error,
110 },
111
112 #[error("transport shutdown failed: {0}")]
113 ShutdownFailed(String),
114
115 #[error("link failed: {0}")]
116 LinkFailed(String),
117
118 #[error("send failed: {0}")]
119 SendFailed(String),
120
121 #[error("receive failed: {0}")]
122 RecvFailed(String),
123
124 #[error("invalid transport address: {0}")]
125 InvalidAddress(String),
126
127 #[error("mtu exceeded: packet {packet_size} > mtu {mtu}")]
128 MtuExceeded { packet_size: usize, mtu: u16 },
129
130 #[error("transport timeout")]
131 Timeout,
132
133 #[error("connection refused")]
134 ConnectionRefused,
135
136 #[error("transport not supported: {0}")]
137 NotSupported(String),
138
139 #[error("io error: {0}")]
140 Io(#[from] std::io::Error),
141}
142
143impl TransportError {
144 pub(crate) fn bind_failed(address: SocketAddr, source: std::io::Error) -> Self {
147 if source.kind() == std::io::ErrorKind::AddrInUse {
148 Self::AddressInUse { address, source }
149 } else {
150 Self::StartFailed(format!("bind {address} failed: {source}"))
151 }
152 }
153
154 #[cfg(windows)]
158 pub(crate) fn exclusive_bind_failed(address: SocketAddr, source: std::io::Error) -> Self {
159 if source.raw_os_error() == Some(10_013) {
160 return Self::AddressInUse { address, source };
161 }
162 Self::bind_failed(address, source)
163 }
164
165 pub fn is_local_route_unavailable(&self) -> bool {
168 match self {
169 TransportError::Io(error) => is_local_route_error_kind(error.kind()),
170 TransportError::SendFailed(message) => is_local_route_error_text(message),
171 _ => false,
172 }
173 }
174}
175
176fn is_local_route_error_kind(kind: std::io::ErrorKind) -> bool {
177 matches!(
178 kind,
179 std::io::ErrorKind::NetworkUnreachable
180 | std::io::ErrorKind::HostUnreachable
181 | std::io::ErrorKind::AddrNotAvailable
182 | std::io::ErrorKind::PermissionDenied
183 )
184}
185
186fn is_local_route_error_text(message: &str) -> bool {
187 let lower = message.to_ascii_lowercase();
188 lower.contains("network is unreachable")
189 || lower.contains("no route to host")
190 || lower.contains("host is unreachable")
191 || lower.contains("can't assign requested address")
192 || lower.contains("cannot assign requested address")
193 || lower.contains("operation not permitted")
194 || lower.contains("permission denied")
195 || lower.contains("os error 51")
196 || lower.contains("os error 65")
197 || lower.contains("os error 49")
198 || lower.contains("os error 1")
199}
200
201#[derive(Clone, Debug, PartialEq, Eq)]
207pub struct TransportType {
208 pub name: &'static str,
210 pub connection_oriented: bool,
212 pub reliable: bool,
214}
215
216impl TransportType {
217 pub const UDP: TransportType = TransportType {
219 name: "udp",
220 connection_oriented: false,
221 reliable: false,
222 };
223
224 pub const TCP: TransportType = TransportType {
226 name: "tcp",
227 connection_oriented: true,
228 reliable: true,
229 };
230
231 pub const ETHERNET: TransportType = TransportType {
233 name: "ethernet",
234 connection_oriented: false,
235 reliable: false,
236 };
237
238 pub const WIFI: TransportType = TransportType {
240 name: "wifi",
241 connection_oriented: false,
242 reliable: false,
243 };
244
245 pub const TOR: TransportType = TransportType {
247 name: "tor",
248 connection_oriented: true,
249 reliable: true,
250 };
251
252 pub const SERIAL: TransportType = TransportType {
254 name: "serial",
255 connection_oriented: false,
256 reliable: true, };
258
259 pub const BLE: TransportType = TransportType {
261 name: "ble",
262 connection_oriented: true,
263 reliable: true, };
265
266 pub const WEBRTC: TransportType = TransportType {
268 name: "webrtc",
269 connection_oriented: true,
270 reliable: false,
271 };
272
273 pub const NOSTR_RELAY: TransportType = TransportType {
275 name: "nostr_relay",
276 connection_oriented: false,
277 reliable: false,
278 };
279
280 #[cfg(feature = "sim-transport")]
282 pub const SIM: TransportType = TransportType {
283 name: "sim",
284 connection_oriented: false,
285 reliable: false,
286 };
287
288 pub fn is_connectionless(&self) -> bool {
290 !self.connection_oriented
291 }
292}
293
294impl fmt::Display for TransportType {
295 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
296 write!(f, "{}", self.name)
297 }
298}
299
300#[derive(Clone, Copy, Debug, PartialEq, Eq)]
306pub enum TransportState {
307 Configured,
309 Starting,
311 Up,
313 Down,
315 Failed,
317}
318
319impl TransportState {
320 pub fn is_operational(&self) -> bool {
322 matches!(self, TransportState::Up)
323 }
324
325 pub fn can_start(&self) -> bool {
327 matches!(
328 self,
329 TransportState::Configured | TransportState::Down | TransportState::Failed
330 )
331 }
332
333 pub fn is_terminal(&self) -> bool {
335 matches!(self, TransportState::Failed)
336 }
337}
338
339impl fmt::Display for TransportState {
340 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
341 let s = match self {
342 TransportState::Configured => "configured",
343 TransportState::Starting => "starting",
344 TransportState::Up => "up",
345 TransportState::Down => "down",
346 TransportState::Failed => "failed",
347 };
348 write!(f, "{}", s)
349 }
350}
351
352#[derive(Clone, Copy, Debug, PartialEq, Eq)]
358pub enum LinkState {
359 Connecting,
361 Connected,
363 Disconnected,
365 Failed,
367}
368
369impl LinkState {
370 pub fn is_operational(&self) -> bool {
372 matches!(self, LinkState::Connected)
373 }
374
375 pub fn is_terminal(&self) -> bool {
377 matches!(self, LinkState::Disconnected | LinkState::Failed)
378 }
379}
380
381impl fmt::Display for LinkState {
382 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
383 let s = match self {
384 LinkState::Connecting => "connecting",
385 LinkState::Connected => "connected",
386 LinkState::Disconnected => "disconnected",
387 LinkState::Failed => "failed",
388 };
389 write!(f, "{}", s)
390 }
391}
392
393#[derive(Clone, Copy, Debug, PartialEq, Eq)]
395pub enum LinkDirection {
396 Outbound,
398 Inbound,
400}
401
402impl fmt::Display for LinkDirection {
403 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
404 let s = match self {
405 LinkDirection::Outbound => "outbound",
406 LinkDirection::Inbound => "inbound",
407 };
408 write!(f, "{}", s)
409 }
410}
411
412#[derive(Clone, PartialEq, Eq, Hash)]
425pub struct TransportAddr(Arc<[u8]>);
426
427impl TransportAddr {
428 pub fn new(bytes: Vec<u8>) -> Self {
430 Self(bytes.into())
431 }
432
433 pub fn from_bytes(bytes: &[u8]) -> Self {
435 Self(Arc::from(bytes))
436 }
437
438 pub fn from_string(s: &str) -> Self {
440 Self(Arc::from(s.as_bytes()))
441 }
442
443 pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self {
448 match addr {
449 std::net::SocketAddr::V4(addr) => {
450 let octets = addr.ip().octets();
451 let mut buf = Vec::with_capacity(21);
452 push_decimal_u8(&mut buf, octets[0]);
453 buf.push(b'.');
454 push_decimal_u8(&mut buf, octets[1]);
455 buf.push(b'.');
456 push_decimal_u8(&mut buf, octets[2]);
457 buf.push(b'.');
458 push_decimal_u8(&mut buf, octets[3]);
459 buf.push(b':');
460 push_decimal_u16(&mut buf, addr.port());
461 Self(buf.into())
462 }
463 std::net::SocketAddr::V6(addr) => {
464 use std::io::Write;
465 let mut buf = Vec::with_capacity(56);
466 buf.push(b'[');
467 write!(&mut buf, "{}", addr.ip()).expect("Vec<u8>::write_fmt is infallible");
468 buf.push(b']');
469 buf.push(b':');
470 push_decimal_u16(&mut buf, addr.port());
471 Self(buf.into())
472 }
473 }
474 }
475
476 pub fn as_bytes(&self) -> &[u8] {
478 &self.0
479 }
480
481 pub fn as_str(&self) -> Option<&str> {
483 std::str::from_utf8(&self.0).ok()
484 }
485
486 pub fn len(&self) -> usize {
488 self.0.len()
489 }
490
491 pub fn is_empty(&self) -> bool {
493 self.0.is_empty()
494 }
495}
496
497fn push_decimal_u8(buf: &mut Vec<u8>, value: u8) {
498 push_decimal_u16(buf, value as u16);
499}
500
501fn push_decimal_u16(buf: &mut Vec<u8>, value: u16) {
502 if value >= 10_000 {
503 buf.push(b'0' + (value / 10_000) as u8);
504 push_fixed_4_digits(buf, value % 10_000);
505 } else if value >= 1_000 {
506 push_fixed_4_digits(buf, value);
507 } else if value >= 100 {
508 buf.push(b'0' + (value / 100) as u8);
509 push_fixed_2_digits(buf, value % 100);
510 } else if value >= 10 {
511 push_fixed_2_digits(buf, value);
512 } else {
513 buf.push(b'0' + value as u8);
514 }
515}
516
517fn push_fixed_4_digits(buf: &mut Vec<u8>, value: u16) {
518 buf.push(b'0' + (value / 1_000) as u8);
519 push_fixed_3_digits(buf, value % 1_000);
520}
521
522fn push_fixed_3_digits(buf: &mut Vec<u8>, value: u16) {
523 buf.push(b'0' + (value / 100) as u8);
524 push_fixed_2_digits(buf, value % 100);
525}
526
527fn push_fixed_2_digits(buf: &mut Vec<u8>, value: u16) {
528 buf.push(b'0' + (value / 10) as u8);
529 buf.push(b'0' + (value % 10) as u8);
530}
531
532impl fmt::Debug for TransportAddr {
533 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
534 match self.as_str() {
535 Some(s) => write!(f, "TransportAddr(\"{}\")", s),
536 None => write!(f, "TransportAddr({:?})", self.0),
537 }
538 }
539}
540
541impl fmt::Display for TransportAddr {
542 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
543 match self.as_str() {
545 Some(s) => write!(f, "{}", s),
546 None => {
547 for byte in self.0.iter() {
548 write!(f, "{:02x}", byte)?;
549 }
550 Ok(())
551 }
552 }
553 }
554}
555
556impl From<&str> for TransportAddr {
557 fn from(s: &str) -> Self {
558 Self::from_string(s)
559 }
560}
561
562impl From<String> for TransportAddr {
563 fn from(s: String) -> Self {
564 Self(s.into_bytes().into())
565 }
566}
567
568#[derive(Clone, Debug, Default)]
574pub struct LinkStats {
575 pub packets_sent: u64,
577 pub packets_recv: u64,
579 pub bytes_sent: u64,
581 pub bytes_recv: u64,
583 pub last_recv_ms: u64,
585 rtt_estimate: Option<Duration>,
587 pub loss_rate: f32,
589 pub throughput_estimate: u64,
591}
592
593impl LinkStats {
594 pub fn new() -> Self {
596 Self::default()
597 }
598
599 pub fn record_sent(&mut self, bytes: usize) {
601 self.packets_sent += 1;
602 self.bytes_sent += bytes as u64;
603 }
604
605 pub fn record_sent_batch(&mut self, packets: usize, bytes: usize) {
607 self.packets_sent += packets as u64;
608 self.bytes_sent += bytes as u64;
609 }
610
611 pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) {
613 self.packets_recv += 1;
614 self.bytes_recv += bytes as u64;
615 self.last_recv_ms = timestamp_ms;
616 }
617
618 pub fn rtt_estimate(&self) -> Option<Duration> {
620 self.rtt_estimate
621 }
622
623 pub fn update_rtt(&mut self, rtt: Duration) {
627 match self.rtt_estimate {
628 Some(old_rtt) => {
629 let alpha = 0.2;
630 let new_rtt_nanos = (alpha * rtt.as_nanos() as f64
631 + (1.0 - alpha) * old_rtt.as_nanos() as f64)
632 as u64;
633 self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos));
634 }
635 None => {
636 self.rtt_estimate = Some(rtt);
637 }
638 }
639 }
640
641 pub fn time_since_recv(&self, current_time_ms: u64) -> u64 {
643 if self.last_recv_ms == 0 {
644 return u64::MAX;
645 }
646 current_time_ms.saturating_sub(self.last_recv_ms)
647 }
648
649 pub fn reset(&mut self) {
651 *self = Self::default();
652 }
653}
654
655#[derive(Clone, Debug)]
661pub struct Link {
662 link_id: LinkId,
664 transport_id: TransportId,
666 remote_addr: TransportAddr,
668 direction: LinkDirection,
670 state: LinkState,
672 base_rtt: Duration,
674 stats: LinkStats,
676 created_at: u64,
678}
679
680impl Link {
681 pub fn new(
683 link_id: LinkId,
684 transport_id: TransportId,
685 remote_addr: TransportAddr,
686 direction: LinkDirection,
687 base_rtt: Duration,
688 ) -> Self {
689 Self {
690 link_id,
691 transport_id,
692 remote_addr,
693 direction,
694 state: LinkState::Connecting,
695 base_rtt,
696 stats: LinkStats::new(),
697 created_at: 0,
698 }
699 }
700
701 pub fn new_with_timestamp(
703 link_id: LinkId,
704 transport_id: TransportId,
705 remote_addr: TransportAddr,
706 direction: LinkDirection,
707 base_rtt: Duration,
708 created_at: u64,
709 ) -> Self {
710 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
711 link.created_at = created_at;
712 link
713 }
714
715 pub fn connectionless(
720 link_id: LinkId,
721 transport_id: TransportId,
722 remote_addr: TransportAddr,
723 direction: LinkDirection,
724 base_rtt: Duration,
725 ) -> Self {
726 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
727 link.state = LinkState::Connected;
728 link
729 }
730
731 pub fn link_id(&self) -> LinkId {
733 self.link_id
734 }
735
736 pub fn transport_id(&self) -> TransportId {
738 self.transport_id
739 }
740
741 pub fn remote_addr(&self) -> &TransportAddr {
743 &self.remote_addr
744 }
745
746 pub fn direction(&self) -> LinkDirection {
748 self.direction
749 }
750
751 pub fn state(&self) -> LinkState {
753 self.state
754 }
755
756 pub fn base_rtt(&self) -> Duration {
758 self.base_rtt
759 }
760
761 pub fn stats(&self) -> &LinkStats {
763 &self.stats
764 }
765
766 pub fn stats_mut(&mut self) -> &mut LinkStats {
768 &mut self.stats
769 }
770
771 pub fn created_at(&self) -> u64 {
773 self.created_at
774 }
775
776 pub fn set_created_at(&mut self, timestamp: u64) {
778 self.created_at = timestamp;
779 }
780
781 pub fn set_connected(&mut self) {
783 self.state = LinkState::Connected;
784 }
785
786 pub fn set_disconnected(&mut self) {
788 self.state = LinkState::Disconnected;
789 }
790
791 pub fn set_failed(&mut self) {
793 self.state = LinkState::Failed;
794 }
795
796 pub fn is_operational(&self) -> bool {
798 self.state.is_operational()
799 }
800
801 pub fn is_terminal(&self) -> bool {
803 self.state.is_terminal()
804 }
805
806 pub fn effective_rtt(&self) -> Duration {
808 self.stats.rtt_estimate().unwrap_or(self.base_rtt)
809 }
810
811 pub fn age(&self, current_time_ms: u64) -> u64 {
813 if self.created_at == 0 {
814 return 0;
815 }
816 current_time_ms.saturating_sub(self.created_at)
817 }
818}
819
820#[derive(Clone, Debug)]
826pub struct DiscoveredPeer {
827 pub transport_id: TransportId,
829 pub addr: TransportAddr,
831 pub pubkey_hint: Option<XOnlyPublicKey>,
833}
834
835impl DiscoveredPeer {
836 pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self {
838 Self {
839 transport_id,
840 addr,
841 pubkey_hint: None,
842 }
843 }
844
845 pub fn with_hint(
847 transport_id: TransportId,
848 addr: TransportAddr,
849 pubkey: XOnlyPublicKey,
850 ) -> Self {
851 Self {
852 transport_id,
853 addr,
854 pubkey_hint: Some(pubkey),
855 }
856 }
857}
858
859pub trait Transport {
868 fn transport_id(&self) -> TransportId;
870
871 fn transport_type(&self) -> &TransportType;
873
874 fn state(&self) -> TransportState;
876
877 fn mtu(&self) -> u16;
879
880 fn link_mtu(&self, addr: &TransportAddr) -> u16 {
886 let _ = addr;
887 self.mtu()
888 }
889
890 fn start(&mut self) -> Result<(), TransportError>;
892
893 fn stop(&mut self) -> Result<(), TransportError>;
895
896 fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>;
898
899 fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError>;
901
902 fn auto_connect(&self) -> bool {
905 false
906 }
907
908 fn accept_connections(&self) -> bool {
911 true
912 }
913
914 fn close_connection(&self, _addr: &TransportAddr) {
920 }
922}
923
924#[derive(Clone, Debug, PartialEq, Eq)]
933pub enum ConnectionState {
934 None,
936 Connecting,
938 Connected,
940 Failed(String),
942}
943
944#[derive(Clone, Debug, Default)]
953pub struct TransportCongestion {
954 pub recv_drops: Option<u64>,
957 pub socket_recv_drops: Option<u64>,
960 pub namespace_recv_drops: Option<u64>,
964}
965
966pub(crate) async fn resolve_socket_addr(
977 addr: &TransportAddr,
978) -> Result<SocketAddr, TransportError> {
979 resolve_socket_addrs(addr)
980 .await?
981 .into_iter()
982 .next()
983 .ok_or_else(|| {
984 TransportError::InvalidAddress(format!(
985 "DNS resolution returned no addresses for {}",
986 addr.as_str().unwrap_or("<non-utf8>")
987 ))
988 })
989}
990
991pub(crate) async fn resolve_socket_addrs(
998 addr: &TransportAddr,
999) -> Result<Vec<SocketAddr>, TransportError> {
1000 let s = addr
1001 .as_str()
1002 .ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?;
1003
1004 if let Ok(sock_addr) = s.parse::<SocketAddr>() {
1006 return Ok(vec![sock_addr]);
1007 }
1008
1009 let addrs = tokio::net::lookup_host(s)
1011 .await
1012 .map_err(|e| {
1013 TransportError::InvalidAddress(format!("DNS resolution failed for {}: {}", s, e))
1014 })?
1015 .collect::<Vec<_>>();
1016 if addrs.is_empty() {
1017 return Err(TransportError::InvalidAddress(format!(
1018 "DNS resolution returned no addresses for {}",
1019 s
1020 )));
1021 }
1022 Ok(addrs)
1023}