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;
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
179fn is_local_route_error_kind(kind: std::io::ErrorKind) -> bool {
180 matches!(
181 kind,
182 std::io::ErrorKind::NetworkUnreachable
183 | std::io::ErrorKind::HostUnreachable
184 | std::io::ErrorKind::AddrNotAvailable
185 | std::io::ErrorKind::PermissionDenied
186 )
187}
188
189fn is_local_route_error_text(message: &str) -> bool {
190 let lower = message.to_ascii_lowercase();
191 lower.contains("network is unreachable")
192 || lower.contains("no route to host")
193 || lower.contains("host is unreachable")
194 || lower.contains("can't assign requested address")
195 || lower.contains("cannot assign requested address")
196 || lower.contains("operation not permitted")
197 || lower.contains("permission denied")
198 || lower.contains("os error 51")
199 || lower.contains("os error 65")
200 || lower.contains("os error 49")
201 || lower.contains("os error 1")
202}
203
204#[derive(Clone, Debug, PartialEq, Eq)]
210pub struct TransportType {
211 pub name: &'static str,
213 pub connection_oriented: bool,
215 pub reliable: bool,
217}
218
219impl TransportType {
220 pub const UDP: TransportType = TransportType {
222 name: "udp",
223 connection_oriented: false,
224 reliable: false,
225 };
226
227 pub const TCP: TransportType = TransportType {
229 name: "tcp",
230 connection_oriented: true,
231 reliable: true,
232 };
233
234 pub const ETHERNET: TransportType = TransportType {
236 name: "ethernet",
237 connection_oriented: false,
238 reliable: false,
239 };
240
241 pub const WIFI: TransportType = TransportType {
243 name: "wifi",
244 connection_oriented: false,
245 reliable: false,
246 };
247
248 pub const TOR: TransportType = TransportType {
250 name: "tor",
251 connection_oriented: true,
252 reliable: true,
253 };
254
255 pub const SERIAL: TransportType = TransportType {
257 name: "serial",
258 connection_oriented: false,
259 reliable: true, };
261
262 pub const BLE: TransportType = TransportType {
264 name: "ble",
265 connection_oriented: true,
266 reliable: true, };
268
269 pub const WEBRTC: TransportType = TransportType {
271 name: "webrtc",
272 connection_oriented: true,
273 reliable: false,
274 };
275
276 pub const NOSTR_RELAY: TransportType = TransportType {
278 name: "nostr_relay",
279 connection_oriented: false,
280 reliable: false,
281 };
282
283 #[cfg(feature = "sim-transport")]
285 pub const SIM: TransportType = TransportType {
286 name: "sim",
287 connection_oriented: false,
288 reliable: false,
289 };
290
291 pub fn is_connectionless(&self) -> bool {
293 !self.connection_oriented
294 }
295}
296
297impl fmt::Display for TransportType {
298 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
299 write!(f, "{}", self.name)
300 }
301}
302
303#[derive(Clone, Copy, Debug, PartialEq, Eq)]
309pub enum TransportState {
310 Configured,
312 Starting,
314 Up,
316 Down,
318 Failed,
320}
321
322impl TransportState {
323 pub fn is_operational(&self) -> bool {
325 matches!(self, TransportState::Up)
326 }
327
328 pub fn can_start(&self) -> bool {
330 matches!(
331 self,
332 TransportState::Configured | TransportState::Down | TransportState::Failed
333 )
334 }
335
336 pub fn is_terminal(&self) -> bool {
338 matches!(self, TransportState::Failed)
339 }
340}
341
342impl fmt::Display for TransportState {
343 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
344 let s = match self {
345 TransportState::Configured => "configured",
346 TransportState::Starting => "starting",
347 TransportState::Up => "up",
348 TransportState::Down => "down",
349 TransportState::Failed => "failed",
350 };
351 write!(f, "{}", s)
352 }
353}
354
355#[derive(Clone, Copy, Debug, PartialEq, Eq)]
361pub enum LinkState {
362 Connecting,
364 Connected,
366 Disconnected,
368 Failed,
370}
371
372impl LinkState {
373 pub fn is_operational(&self) -> bool {
375 matches!(self, LinkState::Connected)
376 }
377
378 pub fn is_terminal(&self) -> bool {
380 matches!(self, LinkState::Disconnected | LinkState::Failed)
381 }
382}
383
384impl fmt::Display for LinkState {
385 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
386 let s = match self {
387 LinkState::Connecting => "connecting",
388 LinkState::Connected => "connected",
389 LinkState::Disconnected => "disconnected",
390 LinkState::Failed => "failed",
391 };
392 write!(f, "{}", s)
393 }
394}
395
396#[derive(Clone, Copy, Debug, PartialEq, Eq)]
398pub enum LinkDirection {
399 Outbound,
401 Inbound,
403}
404
405impl fmt::Display for LinkDirection {
406 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
407 let s = match self {
408 LinkDirection::Outbound => "outbound",
409 LinkDirection::Inbound => "inbound",
410 };
411 write!(f, "{}", s)
412 }
413}
414
415#[derive(Clone, PartialEq, Eq, Hash)]
428pub struct TransportAddr(Arc<[u8]>);
429
430impl TransportAddr {
431 pub fn new(bytes: Vec<u8>) -> Self {
433 Self(bytes.into())
434 }
435
436 pub fn from_bytes(bytes: &[u8]) -> Self {
438 Self(Arc::from(bytes))
439 }
440
441 pub fn from_string(s: &str) -> Self {
443 Self(Arc::from(s.as_bytes()))
444 }
445
446 pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self {
451 match addr {
452 std::net::SocketAddr::V4(addr) => {
453 let octets = addr.ip().octets();
454 let mut buf = Vec::with_capacity(21);
455 push_decimal_u8(&mut buf, octets[0]);
456 buf.push(b'.');
457 push_decimal_u8(&mut buf, octets[1]);
458 buf.push(b'.');
459 push_decimal_u8(&mut buf, octets[2]);
460 buf.push(b'.');
461 push_decimal_u8(&mut buf, octets[3]);
462 buf.push(b':');
463 push_decimal_u16(&mut buf, addr.port());
464 Self(buf.into())
465 }
466 std::net::SocketAddr::V6(addr) => {
467 use std::io::Write;
468 let mut buf = Vec::with_capacity(56);
469 buf.push(b'[');
470 write!(&mut buf, "{}", addr.ip()).expect("Vec<u8>::write_fmt is infallible");
471 buf.push(b']');
472 buf.push(b':');
473 push_decimal_u16(&mut buf, addr.port());
474 Self(buf.into())
475 }
476 }
477 }
478
479 pub fn as_bytes(&self) -> &[u8] {
481 &self.0
482 }
483
484 pub fn as_str(&self) -> Option<&str> {
486 std::str::from_utf8(&self.0).ok()
487 }
488
489 pub fn len(&self) -> usize {
491 self.0.len()
492 }
493
494 pub fn is_empty(&self) -> bool {
496 self.0.is_empty()
497 }
498}
499
500fn push_decimal_u8(buf: &mut Vec<u8>, value: u8) {
501 push_decimal_u16(buf, value as u16);
502}
503
504fn push_decimal_u16(buf: &mut Vec<u8>, value: u16) {
505 if value >= 10_000 {
506 buf.push(b'0' + (value / 10_000) as u8);
507 push_fixed_4_digits(buf, value % 10_000);
508 } else if value >= 1_000 {
509 push_fixed_4_digits(buf, value);
510 } else if value >= 100 {
511 buf.push(b'0' + (value / 100) as u8);
512 push_fixed_2_digits(buf, value % 100);
513 } else if value >= 10 {
514 push_fixed_2_digits(buf, value);
515 } else {
516 buf.push(b'0' + value as u8);
517 }
518}
519
520fn push_fixed_4_digits(buf: &mut Vec<u8>, value: u16) {
521 buf.push(b'0' + (value / 1_000) as u8);
522 push_fixed_3_digits(buf, value % 1_000);
523}
524
525fn push_fixed_3_digits(buf: &mut Vec<u8>, value: u16) {
526 buf.push(b'0' + (value / 100) as u8);
527 push_fixed_2_digits(buf, value % 100);
528}
529
530fn push_fixed_2_digits(buf: &mut Vec<u8>, value: u16) {
531 buf.push(b'0' + (value / 10) as u8);
532 buf.push(b'0' + (value % 10) as u8);
533}
534
535impl fmt::Debug for TransportAddr {
536 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
537 match self.as_str() {
538 Some(s) => write!(f, "TransportAddr(\"{}\")", s),
539 None => write!(f, "TransportAddr({:?})", self.0),
540 }
541 }
542}
543
544impl fmt::Display for TransportAddr {
545 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
546 match self.as_str() {
548 Some(s) => write!(f, "{}", s),
549 None => {
550 for byte in self.0.iter() {
551 write!(f, "{:02x}", byte)?;
552 }
553 Ok(())
554 }
555 }
556 }
557}
558
559impl From<&str> for TransportAddr {
560 fn from(s: &str) -> Self {
561 Self::from_string(s)
562 }
563}
564
565impl From<String> for TransportAddr {
566 fn from(s: String) -> Self {
567 Self(s.into_bytes().into())
568 }
569}
570
571#[derive(Clone, Debug, Default)]
577pub struct LinkStats {
578 pub packets_sent: u64,
580 pub packets_recv: u64,
582 pub bytes_sent: u64,
584 pub bytes_recv: u64,
586 pub last_recv_ms: u64,
588 rtt_estimate: Option<Duration>,
590 pub loss_rate: f32,
592 pub throughput_estimate: u64,
594}
595
596impl LinkStats {
597 pub fn new() -> Self {
599 Self::default()
600 }
601
602 pub fn record_sent(&mut self, bytes: usize) {
604 self.packets_sent += 1;
605 self.bytes_sent += bytes as u64;
606 }
607
608 pub fn record_sent_batch(&mut self, packets: usize, bytes: usize) {
610 self.packets_sent += packets as u64;
611 self.bytes_sent += bytes as u64;
612 }
613
614 pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) {
616 self.packets_recv += 1;
617 self.bytes_recv += bytes as u64;
618 self.last_recv_ms = timestamp_ms;
619 }
620
621 pub fn rtt_estimate(&self) -> Option<Duration> {
623 self.rtt_estimate
624 }
625
626 pub fn update_rtt(&mut self, rtt: Duration) {
630 match self.rtt_estimate {
631 Some(old_rtt) => {
632 let alpha = 0.2;
633 let new_rtt_nanos = (alpha * rtt.as_nanos() as f64
634 + (1.0 - alpha) * old_rtt.as_nanos() as f64)
635 as u64;
636 self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos));
637 }
638 None => {
639 self.rtt_estimate = Some(rtt);
640 }
641 }
642 }
643
644 pub fn time_since_recv(&self, current_time_ms: u64) -> u64 {
646 if self.last_recv_ms == 0 {
647 return u64::MAX;
648 }
649 current_time_ms.saturating_sub(self.last_recv_ms)
650 }
651
652 pub fn reset(&mut self) {
654 *self = Self::default();
655 }
656}
657
658#[derive(Clone, Debug)]
664pub struct Link {
665 link_id: LinkId,
667 transport_id: TransportId,
669 remote_addr: TransportAddr,
671 direction: LinkDirection,
673 state: LinkState,
675 base_rtt: Duration,
677 stats: LinkStats,
679 created_at: u64,
681}
682
683impl Link {
684 pub fn new(
686 link_id: LinkId,
687 transport_id: TransportId,
688 remote_addr: TransportAddr,
689 direction: LinkDirection,
690 base_rtt: Duration,
691 ) -> Self {
692 Self {
693 link_id,
694 transport_id,
695 remote_addr,
696 direction,
697 state: LinkState::Connecting,
698 base_rtt,
699 stats: LinkStats::new(),
700 created_at: 0,
701 }
702 }
703
704 pub fn new_with_timestamp(
706 link_id: LinkId,
707 transport_id: TransportId,
708 remote_addr: TransportAddr,
709 direction: LinkDirection,
710 base_rtt: Duration,
711 created_at: u64,
712 ) -> Self {
713 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
714 link.created_at = created_at;
715 link
716 }
717
718 pub fn connectionless(
723 link_id: LinkId,
724 transport_id: TransportId,
725 remote_addr: TransportAddr,
726 direction: LinkDirection,
727 base_rtt: Duration,
728 ) -> Self {
729 let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
730 link.state = LinkState::Connected;
731 link
732 }
733
734 pub fn link_id(&self) -> LinkId {
736 self.link_id
737 }
738
739 pub fn transport_id(&self) -> TransportId {
741 self.transport_id
742 }
743
744 pub fn remote_addr(&self) -> &TransportAddr {
746 &self.remote_addr
747 }
748
749 pub fn direction(&self) -> LinkDirection {
751 self.direction
752 }
753
754 pub fn state(&self) -> LinkState {
756 self.state
757 }
758
759 pub fn base_rtt(&self) -> Duration {
761 self.base_rtt
762 }
763
764 pub fn stats(&self) -> &LinkStats {
766 &self.stats
767 }
768
769 pub fn stats_mut(&mut self) -> &mut LinkStats {
771 &mut self.stats
772 }
773
774 pub fn created_at(&self) -> u64 {
776 self.created_at
777 }
778
779 pub fn set_created_at(&mut self, timestamp: u64) {
781 self.created_at = timestamp;
782 }
783
784 pub fn set_connected(&mut self) {
786 self.state = LinkState::Connected;
787 }
788
789 pub fn set_disconnected(&mut self) {
791 self.state = LinkState::Disconnected;
792 }
793
794 pub fn set_failed(&mut self) {
796 self.state = LinkState::Failed;
797 }
798
799 pub fn is_operational(&self) -> bool {
801 self.state.is_operational()
802 }
803
804 pub fn is_terminal(&self) -> bool {
806 self.state.is_terminal()
807 }
808
809 pub fn effective_rtt(&self) -> Duration {
811 self.stats.rtt_estimate().unwrap_or(self.base_rtt)
812 }
813
814 pub fn age(&self, current_time_ms: u64) -> u64 {
816 if self.created_at == 0 {
817 return 0;
818 }
819 current_time_ms.saturating_sub(self.created_at)
820 }
821}
822
823#[derive(Clone, Debug)]
829pub struct DiscoveredPeer {
830 pub transport_id: TransportId,
832 pub addr: TransportAddr,
834 pub pubkey_hint: Option<XOnlyPublicKey>,
836}
837
838impl DiscoveredPeer {
839 pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self {
841 Self {
842 transport_id,
843 addr,
844 pubkey_hint: None,
845 }
846 }
847
848 pub fn with_hint(
850 transport_id: TransportId,
851 addr: TransportAddr,
852 pubkey: XOnlyPublicKey,
853 ) -> Self {
854 Self {
855 transport_id,
856 addr,
857 pubkey_hint: Some(pubkey),
858 }
859 }
860}
861
862pub trait Transport {
871 fn transport_id(&self) -> TransportId;
873
874 fn transport_type(&self) -> &TransportType;
876
877 fn state(&self) -> TransportState;
879
880 fn mtu(&self) -> u16;
882
883 fn link_mtu(&self, addr: &TransportAddr) -> u16 {
889 let _ = addr;
890 self.mtu()
891 }
892
893 fn start(&mut self) -> Result<(), TransportError>;
895
896 fn stop(&mut self) -> Result<(), TransportError>;
898
899 fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>;
901
902 fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError>;
904
905 fn auto_connect(&self) -> bool {
908 false
909 }
910
911 fn accept_connections(&self) -> bool {
914 true
915 }
916
917 fn close_connection(&self, _addr: &TransportAddr) {
923 }
925}
926
927#[derive(Clone, Debug, PartialEq, Eq)]
936pub enum ConnectionState {
937 None,
939 Connecting,
941 Connected,
943 Failed(String),
945}
946
947#[derive(Clone, Debug, Default)]
956pub struct TransportCongestion {
957 pub recv_drops: Option<u64>,
960 pub socket_recv_drops: Option<u64>,
963 pub namespace_recv_drops: Option<u64>,
967}