pub mod tcp;
pub mod tor;
pub mod udp;
#[cfg(feature = "webrtc-transport")]
pub mod webrtc;
#[cfg(feature = "sim-transport")]
pub mod sim;
#[cfg(any(target_os = "linux", target_os = "macos"))]
pub mod ethernet;
#[cfg(target_os = "linux")]
pub mod ble;
mod handle;
mod packet_channel;
#[cfg(test)]
mod tests;
pub use handle::TransportHandle;
pub(crate) use packet_channel::received_timestamp_ms;
pub use packet_channel::{PacketRx, PacketTx, ReceivedPacket, packet_channel};
use secp256k1::XOnlyPublicKey;
use std::fmt;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct TransportId(u32);
impl TransportId {
pub fn new(id: u32) -> Self {
Self(id)
}
pub fn as_u32(&self) -> u32 {
self.0
}
}
impl fmt::Display for TransportId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "transport:{}", self.0)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct LinkId(u64);
impl LinkId {
pub fn new(id: u64) -> Self {
Self(id)
}
pub fn as_u64(&self) -> u64 {
self.0
}
}
impl fmt::Display for LinkId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "link:{}", self.0)
}
}
#[derive(Debug, Error)]
pub enum TransportError {
#[error("transport not started")]
NotStarted,
#[error("transport already started")]
AlreadyStarted,
#[error("transport failed to start: {0}")]
StartFailed(String),
#[error("transport shutdown failed: {0}")]
ShutdownFailed(String),
#[error("link failed: {0}")]
LinkFailed(String),
#[error("send failed: {0}")]
SendFailed(String),
#[error("receive failed: {0}")]
RecvFailed(String),
#[error("invalid transport address: {0}")]
InvalidAddress(String),
#[error("mtu exceeded: packet {packet_size} > mtu {mtu}")]
MtuExceeded { packet_size: usize, mtu: u16 },
#[error("transport timeout")]
Timeout,
#[error("connection refused")]
ConnectionRefused,
#[error("transport not supported: {0}")]
NotSupported(String),
#[error("io error: {0}")]
Io(#[from] std::io::Error),
}
impl TransportError {
pub fn is_local_route_unavailable(&self) -> bool {
match self {
TransportError::Io(error) => is_local_route_error_kind(error.kind()),
TransportError::SendFailed(message) => is_local_route_error_text(message),
_ => false,
}
}
}
fn is_local_route_error_kind(kind: std::io::ErrorKind) -> bool {
matches!(
kind,
std::io::ErrorKind::NetworkUnreachable
| std::io::ErrorKind::HostUnreachable
| std::io::ErrorKind::AddrNotAvailable
)
}
fn is_local_route_error_text(message: &str) -> bool {
let lower = message.to_ascii_lowercase();
lower.contains("network is unreachable")
|| lower.contains("no route to host")
|| lower.contains("host is unreachable")
|| lower.contains("can't assign requested address")
|| lower.contains("cannot assign requested address")
|| lower.contains("os error 51")
|| lower.contains("os error 65")
|| lower.contains("os error 49")
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TransportType {
pub name: &'static str,
pub connection_oriented: bool,
pub reliable: bool,
}
impl TransportType {
pub const UDP: TransportType = TransportType {
name: "udp",
connection_oriented: false,
reliable: false,
};
pub const TCP: TransportType = TransportType {
name: "tcp",
connection_oriented: true,
reliable: true,
};
pub const ETHERNET: TransportType = TransportType {
name: "ethernet",
connection_oriented: false,
reliable: false,
};
pub const WIFI: TransportType = TransportType {
name: "wifi",
connection_oriented: false,
reliable: false,
};
pub const TOR: TransportType = TransportType {
name: "tor",
connection_oriented: true,
reliable: true,
};
pub const SERIAL: TransportType = TransportType {
name: "serial",
connection_oriented: false,
reliable: true, };
pub const BLE: TransportType = TransportType {
name: "ble",
connection_oriented: true,
reliable: true, };
pub const WEBRTC: TransportType = TransportType {
name: "webrtc",
connection_oriented: true,
reliable: false,
};
#[cfg(feature = "sim-transport")]
pub const SIM: TransportType = TransportType {
name: "sim",
connection_oriented: false,
reliable: false,
};
pub fn is_connectionless(&self) -> bool {
!self.connection_oriented
}
}
impl fmt::Display for TransportType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}", self.name)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TransportState {
Configured,
Starting,
Up,
Down,
Failed,
}
impl TransportState {
pub fn is_operational(&self) -> bool {
matches!(self, TransportState::Up)
}
pub fn can_start(&self) -> bool {
matches!(
self,
TransportState::Configured | TransportState::Down | TransportState::Failed
)
}
pub fn is_terminal(&self) -> bool {
matches!(self, TransportState::Failed)
}
}
impl fmt::Display for TransportState {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let s = match self {
TransportState::Configured => "configured",
TransportState::Starting => "starting",
TransportState::Up => "up",
TransportState::Down => "down",
TransportState::Failed => "failed",
};
write!(f, "{}", s)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LinkState {
Connecting,
Connected,
Disconnected,
Failed,
}
impl LinkState {
pub fn is_operational(&self) -> bool {
matches!(self, LinkState::Connected)
}
pub fn is_terminal(&self) -> bool {
matches!(self, LinkState::Disconnected | LinkState::Failed)
}
}
impl fmt::Display for LinkState {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let s = match self {
LinkState::Connecting => "connecting",
LinkState::Connected => "connected",
LinkState::Disconnected => "disconnected",
LinkState::Failed => "failed",
};
write!(f, "{}", s)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum LinkDirection {
Outbound,
Inbound,
}
impl fmt::Display for LinkDirection {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let s = match self {
LinkDirection::Outbound => "outbound",
LinkDirection::Inbound => "inbound",
};
write!(f, "{}", s)
}
}
#[derive(Clone, PartialEq, Eq, Hash)]
pub struct TransportAddr(Arc<[u8]>);
impl TransportAddr {
pub fn new(bytes: Vec<u8>) -> Self {
Self(bytes.into())
}
pub fn from_bytes(bytes: &[u8]) -> Self {
Self(Arc::from(bytes))
}
pub fn from_string(s: &str) -> Self {
Self(Arc::from(s.as_bytes()))
}
pub fn from_socket_addr(addr: std::net::SocketAddr) -> Self {
use std::io::Write;
let mut buf = Vec::with_capacity(56);
write!(&mut buf, "{addr}").expect("Vec<u8>::write_fmt is infallible");
Self(buf.into())
}
pub fn as_bytes(&self) -> &[u8] {
&self.0
}
pub fn as_str(&self) -> Option<&str> {
std::str::from_utf8(&self.0).ok()
}
pub fn len(&self) -> usize {
self.0.len()
}
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
}
impl fmt::Debug for TransportAddr {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.as_str() {
Some(s) => write!(f, "TransportAddr(\"{}\")", s),
None => write!(f, "TransportAddr({:?})", self.0),
}
}
}
impl fmt::Display for TransportAddr {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.as_str() {
Some(s) => write!(f, "{}", s),
None => {
for byte in self.0.iter() {
write!(f, "{:02x}", byte)?;
}
Ok(())
}
}
}
}
impl From<&str> for TransportAddr {
fn from(s: &str) -> Self {
Self::from_string(s)
}
}
impl From<String> for TransportAddr {
fn from(s: String) -> Self {
Self(s.into_bytes().into())
}
}
#[derive(Clone, Debug, Default)]
pub struct LinkStats {
pub packets_sent: u64,
pub packets_recv: u64,
pub bytes_sent: u64,
pub bytes_recv: u64,
pub last_recv_ms: u64,
rtt_estimate: Option<Duration>,
pub loss_rate: f32,
pub throughput_estimate: u64,
}
impl LinkStats {
pub fn new() -> Self {
Self::default()
}
pub fn record_sent(&mut self, bytes: usize) {
self.packets_sent += 1;
self.bytes_sent += bytes as u64;
}
pub fn record_sent_batch(&mut self, packets: usize, bytes: usize) {
self.packets_sent += packets as u64;
self.bytes_sent += bytes as u64;
}
pub fn record_recv(&mut self, bytes: usize, timestamp_ms: u64) {
self.packets_recv += 1;
self.bytes_recv += bytes as u64;
self.last_recv_ms = timestamp_ms;
}
pub fn rtt_estimate(&self) -> Option<Duration> {
self.rtt_estimate
}
pub fn update_rtt(&mut self, rtt: Duration) {
match self.rtt_estimate {
Some(old_rtt) => {
let alpha = 0.2;
let new_rtt_nanos = (alpha * rtt.as_nanos() as f64
+ (1.0 - alpha) * old_rtt.as_nanos() as f64)
as u64;
self.rtt_estimate = Some(Duration::from_nanos(new_rtt_nanos));
}
None => {
self.rtt_estimate = Some(rtt);
}
}
}
pub fn time_since_recv(&self, current_time_ms: u64) -> u64 {
if self.last_recv_ms == 0 {
return u64::MAX;
}
current_time_ms.saturating_sub(self.last_recv_ms)
}
pub fn reset(&mut self) {
*self = Self::default();
}
}
#[derive(Clone, Debug)]
pub struct Link {
link_id: LinkId,
transport_id: TransportId,
remote_addr: TransportAddr,
direction: LinkDirection,
state: LinkState,
base_rtt: Duration,
stats: LinkStats,
created_at: u64,
}
impl Link {
pub fn new(
link_id: LinkId,
transport_id: TransportId,
remote_addr: TransportAddr,
direction: LinkDirection,
base_rtt: Duration,
) -> Self {
Self {
link_id,
transport_id,
remote_addr,
direction,
state: LinkState::Connecting,
base_rtt,
stats: LinkStats::new(),
created_at: 0,
}
}
pub fn new_with_timestamp(
link_id: LinkId,
transport_id: TransportId,
remote_addr: TransportAddr,
direction: LinkDirection,
base_rtt: Duration,
created_at: u64,
) -> Self {
let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
link.created_at = created_at;
link
}
pub fn connectionless(
link_id: LinkId,
transport_id: TransportId,
remote_addr: TransportAddr,
direction: LinkDirection,
base_rtt: Duration,
) -> Self {
let mut link = Self::new(link_id, transport_id, remote_addr, direction, base_rtt);
link.state = LinkState::Connected;
link
}
pub fn link_id(&self) -> LinkId {
self.link_id
}
pub fn transport_id(&self) -> TransportId {
self.transport_id
}
pub fn remote_addr(&self) -> &TransportAddr {
&self.remote_addr
}
pub fn direction(&self) -> LinkDirection {
self.direction
}
pub fn state(&self) -> LinkState {
self.state
}
pub fn base_rtt(&self) -> Duration {
self.base_rtt
}
pub fn stats(&self) -> &LinkStats {
&self.stats
}
pub fn stats_mut(&mut self) -> &mut LinkStats {
&mut self.stats
}
pub fn created_at(&self) -> u64 {
self.created_at
}
pub fn set_created_at(&mut self, timestamp: u64) {
self.created_at = timestamp;
}
pub fn set_connected(&mut self) {
self.state = LinkState::Connected;
}
pub fn set_disconnected(&mut self) {
self.state = LinkState::Disconnected;
}
pub fn set_failed(&mut self) {
self.state = LinkState::Failed;
}
pub fn is_operational(&self) -> bool {
self.state.is_operational()
}
pub fn is_terminal(&self) -> bool {
self.state.is_terminal()
}
pub fn effective_rtt(&self) -> Duration {
self.stats.rtt_estimate().unwrap_or(self.base_rtt)
}
pub fn age(&self, current_time_ms: u64) -> u64 {
if self.created_at == 0 {
return 0;
}
current_time_ms.saturating_sub(self.created_at)
}
}
#[derive(Clone, Debug)]
pub struct DiscoveredPeer {
pub transport_id: TransportId,
pub addr: TransportAddr,
pub pubkey_hint: Option<XOnlyPublicKey>,
}
impl DiscoveredPeer {
pub fn new(transport_id: TransportId, addr: TransportAddr) -> Self {
Self {
transport_id,
addr,
pubkey_hint: None,
}
}
pub fn with_hint(
transport_id: TransportId,
addr: TransportAddr,
pubkey: XOnlyPublicKey,
) -> Self {
Self {
transport_id,
addr,
pubkey_hint: Some(pubkey),
}
}
}
pub trait Transport {
fn transport_id(&self) -> TransportId;
fn transport_type(&self) -> &TransportType;
fn state(&self) -> TransportState;
fn mtu(&self) -> u16;
fn link_mtu(&self, addr: &TransportAddr) -> u16 {
let _ = addr;
self.mtu()
}
fn start(&mut self) -> Result<(), TransportError>;
fn stop(&mut self) -> Result<(), TransportError>;
fn send(&self, addr: &TransportAddr, data: &[u8]) -> Result<(), TransportError>;
fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError>;
fn auto_connect(&self) -> bool {
false
}
fn accept_connections(&self) -> bool {
true
}
fn close_connection(&self, _addr: &TransportAddr) {
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ConnectionState {
None,
Connecting,
Connected,
Failed(String),
}
#[derive(Clone, Debug, Default)]
pub struct TransportCongestion {
pub recv_drops: Option<u64>,
}
pub(crate) async fn resolve_socket_addr(
addr: &TransportAddr,
) -> Result<SocketAddr, TransportError> {
resolve_socket_addrs(addr)
.await?
.into_iter()
.next()
.ok_or_else(|| {
TransportError::InvalidAddress(format!(
"DNS resolution returned no addresses for {}",
addr.as_str().unwrap_or("<non-utf8>")
))
})
}
pub(crate) async fn resolve_socket_addrs(
addr: &TransportAddr,
) -> Result<Vec<SocketAddr>, TransportError> {
let s = addr
.as_str()
.ok_or_else(|| TransportError::InvalidAddress("not valid UTF-8".into()))?;
if let Ok(sock_addr) = s.parse::<SocketAddr>() {
return Ok(vec![sock_addr]);
}
let addrs = tokio::net::lookup_host(s)
.await
.map_err(|e| {
TransportError::InvalidAddress(format!("DNS resolution failed for {}: {}", s, e))
})?
.collect::<Vec<_>>();
if addrs.is_empty() {
return Err(TransportError::InvalidAddress(format!(
"DNS resolution returned no addresses for {}",
s
)));
}
Ok(addrs)
}