use anyhow::{anyhow, Context, Result};
use async_trait::async_trait;
use bytes::Bytes;
use futures_util::FutureExt;
use quinn::{Connection, RecvStream, SendStream};
use rust_socketio::asynchronous::{Client, ClientBuilder};
use rust_socketio::{Event as SioEvent, Payload, TransportType};
use serde_json::{json, Value};
use std::net::{IpAddr, SocketAddr};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, Mutex, Notify};
use webrtc::api::interceptor_registry::register_default_interceptors;
use webrtc::api::media_engine::MediaEngine;
use webrtc::api::setting_engine::SettingEngine;
use webrtc::api::APIBuilder;
use webrtc::data::data_channel::DataChannel as RawDataChannel;
use webrtc::data_channel::RTCDataChannel;
use webrtc::ice_transport::ice_candidate::{RTCIceCandidate, RTCIceCandidateInit};
use webrtc::ice_transport::ice_candidate_type::RTCIceCandidateType;
use webrtc::ice_transport::ice_server::RTCIceServer;
use webrtc::interceptor::registry::Registry;
use webrtc::peer_connection::configuration::RTCConfiguration;
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
use webrtc::peer_connection::policy::ice_transport_policy::RTCIceTransportPolicy;
use webrtc::peer_connection::sdp::sdp_type::RTCSdpType;
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
use webrtc::peer_connection::signaling_state::RTCSignalingState;
use webrtc::peer_connection::RTCPeerConnection;
pub const MAX_DC_PAYLOAD: usize = 60 * 1024;
const HIGH_WATER: usize = 4 * 1024 * 1024;
const LOW_WATER: usize = 1024 * 1024;
pub const WATCHDOG_SECS: u64 = 15;
pub const STALL_MS_DEFAULT: u64 = 6_000;
pub fn stall_ms() -> u64 {
std::env::var("FILAMENT_STALL_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(STALL_MS_DEFAULT)
}
pub const ESTABLISH_GRACE_MS_DEFAULT: u64 = 45_000;
pub fn establish_grace_ms() -> u64 {
std::env::var("FILAMENT_ESTABLISH_GRACE_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(ESTABLISH_GRACE_MS_DEFAULT)
.max(stall_ms())
}
pub const DIRECT_STREAMS_DEFAULT: usize = 1;
pub fn direct_streams() -> usize {
let from_env = std::env::var("FILAMENT_DIRECT_STREAMS")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n >= 1);
if let Some(n) = from_env {
return n;
}
DIRECT_STREAMS_DEFAULT
}
pub const SIGNALING_SILENCE_MS_DEFAULT: u64 = 15_000;
pub fn signaling_silence_ms() -> u64 {
std::env::var("FILAMENT_SIGNALING_SILENCE_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(SIGNALING_SILENCE_MS_DEFAULT)
}
pub fn warm_standby_override() -> Option<bool> {
match std::env::var("FILAMENT_WARM_STANDBY").ok().as_deref() {
Some("1") | Some("true") | Some("on") => Some(true),
Some("0") | Some("false") | Some("off") => Some(false),
_ => None,
}
}
pub fn upgrade_prober_enabled() -> bool {
match std::env::var("FILAMENT_UPGRADE_PROBE").ok().as_deref() {
Some("0") | Some("false") | Some("off") => false,
_ => true,
}
}
pub const UPGRADE_FIRST_MS_DEFAULT: u64 = 5_000;
pub fn upgrade_first_ms() -> u64 {
std::env::var("FILAMENT_UPGRADE_FIRST_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(UPGRADE_FIRST_MS_DEFAULT)
}
pub const UPGRADE_STEADY_MS_DEFAULT: u64 = 25_000;
pub fn upgrade_steady_ms() -> u64 {
std::env::var("FILAMENT_UPGRADE_STEADY_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(UPGRADE_STEADY_MS_DEFAULT)
}
pub const UPGRADE_VERIFY_MS_DEFAULT: u64 = 2_500;
pub fn upgrade_verify_ms() -> u64 {
std::env::var("FILAMENT_UPGRADE_VERIFY_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(UPGRADE_VERIFY_MS_DEFAULT)
}
pub const UPGRADE_VERIFY_IDLE_MS_DEFAULT: u64 = 1_200;
pub fn upgrade_verify_idle_ms() -> u64 {
std::env::var("FILAMENT_UPGRADE_VERIFY_IDLE_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(UPGRADE_VERIFY_IDLE_MS_DEFAULT)
}
#[derive(Debug)]
pub enum Ev {
Welcome(Value),
PeerJoined(Value),
PeerLeft(Value),
Signal(Value),
PairCode(Value),
PairOk(Value),
PairMatched(Value),
#[allow(dead_code)] PairUsed(Value),
PairError(Value),
KnownPeer(Value),
#[allow(dead_code)] KnownPeerLeft(Value),
Synced(Value),
SignalingAlive,
ChannelReady(String, Arc<dyn Transport>),
DirectReady(String, Arc<dyn Transport>, &'static str),
DirectUpgradeReady(String, Arc<dyn Transport>, &'static str),
DirectWorkersReady(String, Vec<Arc<dyn Transport>>),
Control(String, Value),
Chunk(String, u32, Option<u64>, Bytes),
PcState(String, String),
#[allow(dead_code)] TransferDone(String),
TransferFailed { id: String, err: String },
MaybeComplete(String, u32),
Stuck(String, u32),
TransferStalled(String, u64),
GraceExpired(String, u32),
#[allow(dead_code)] SignalingDown(String),
DropLink(String),
Interrupted,
StdinLine(String),
}
impl std::fmt::Debug for dyn Transport {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Transport")
}
}
#[async_trait]
pub trait Transport: Send + Sync {
async fn send_control(&self, msg: &Value) -> Result<()>;
async fn send_frame(&self, sid: u32, offset: u64, payload: &[u8]) -> Result<()>;
async fn flush(&self) -> Result<()>;
async fn drain_finish(&self) -> Result<()> {
self.flush().await
}
fn max_payload(&self) -> usize;
fn sid_answerer(&self) -> bool {
false
}
fn idle_ms(&self) -> u64 {
u64::MAX
}
fn is_alive(&self) -> bool {
true
}
fn supports_datagrams(&self) -> bool {
false
}
fn is_dead(&self) -> bool {
false
}
fn as_any(&self) -> &dyn std::any::Any;
fn force_close(&self) {}
async fn open_stream(&self) -> Option<(SendStream, RecvStream, quinn::Connection)> {
None
}
fn send_datagram(&self, _packet: &[u8]) -> Result<()> {
anyhow::bail!("transport has no L3 datagram support")
}
async fn recv_datagram(&self) -> Result<bytes::Bytes> {
anyhow::bail!("transport has no L3 datagram support")
}
fn max_datagram_size(&self) -> Option<usize> {
None
}
fn channel_binding(&self) -> Option<Vec<u8>> {
None
}
fn remote_addr(&self) -> Option<std::net::SocketAddr> {
None
}
fn rtt_ms(&self) -> Option<u64> {
None
}
fn local_ip(&self) -> Option<std::net::IpAddr> {
None
}
fn has_flowed(&self) -> bool {
true
}
}
const READ_BUF: usize = 1 << 20;
fn now_ms() -> u64 {
use std::sync::OnceLock;
static EPOCH: OnceLock<std::time::Instant> = OnceLock::new();
EPOCH
.get_or_init(std::time::Instant::now)
.elapsed()
.as_millis() as u64
}
pub struct DataChannelTransport {
raw: Arc<RawDataChannel>,
drained: Arc<Notify>,
dead: Arc<std::sync::atomic::AtomicBool>, last_activity: Arc<std::sync::atomic::AtomicU64>,
first_data: Arc<std::sync::atomic::AtomicBool>,
answerer: bool,
}
impl DataChannelTransport {
fn is_dead(&self) -> bool {
self.dead.load(std::sync::atomic::Ordering::Relaxed)
}
}
#[async_trait]
impl Transport for DataChannelTransport {
async fn send_control(&self, msg: &Value) -> Result<()> {
if self.is_dead() {
return Err(anyhow!("channel closed"));
}
self.raw
.write_data_channel(&Bytes::from(msg.to_string()), true)
.await?;
Ok(())
}
async fn send_frame(&self, sid: u32, _offset: u64, payload: &[u8]) -> Result<()> {
let mut framed = Vec::with_capacity(4 + payload.len());
framed.extend_from_slice(&sid.to_be_bytes());
framed.extend_from_slice(payload);
loop {
if self.is_dead() {
return Err(anyhow!("channel closed"));
}
if self.raw.buffered_amount() <= HIGH_WATER {
break;
}
let notified = self.drained.notified();
if self.raw.buffered_amount() <= HIGH_WATER {
break;
}
notified.await;
}
self.raw
.write_data_channel(&Bytes::from(framed), false)
.await?;
self.last_activity
.store(now_ms(), std::sync::atomic::Ordering::Relaxed);
self.first_data
.store(true, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
async fn flush(&self) -> Result<()> {
while self.raw.buffered_amount() > 0 {
if self.is_dead() {
return Err(anyhow!("channel closed while flushing"));
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
Ok(())
}
fn max_payload(&self) -> usize {
MAX_DC_PAYLOAD
}
fn idle_ms(&self) -> u64 {
if self.is_dead() {
return u64::MAX;
}
now_ms().saturating_sub(self.last_activity.load(std::sync::atomic::Ordering::Relaxed))
}
fn has_flowed(&self) -> bool {
self.first_data.load(std::sync::atomic::Ordering::Relaxed)
}
fn is_alive(&self) -> bool {
!self.is_dead()
}
fn sid_answerer(&self) -> bool {
self.answerer
}
fn as_any(&self) -> &dyn std::any::Any {
self
}
}
const DNS_RACE_MS_DEFAULT: u64 = 700;
fn dns_race_ms() -> u64 {
std::env::var("FILAMENT_DNS_RACE_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(DNS_RACE_MS_DEFAULT)
}
const DNS_TIMEOUT_MS_DEFAULT: u64 = 9_000;
fn dns_timeout_ms() -> u64 {
std::env::var("FILAMENT_DNS_TIMEOUT_MS")
.ok()
.and_then(|v| v.parse::<u64>().ok())
.filter(|n| *n > 0)
.unwrap_or(DNS_TIMEOUT_MS_DEFAULT)
}
fn dns_cache_path() -> PathBuf {
crate::platform::Paths::config_path("signaling-dns.json")
}
pub(crate) fn host_port_from_url(url: &str) -> Option<(String, u16)> {
let (scheme, rest) = url.split_once("://")?;
let authority = rest.split(['/', '?', '#']).next().unwrap_or(rest);
let hostport = authority.rsplit('@').next().unwrap_or(authority);
let default_port = match scheme.to_ascii_lowercase().as_str() {
"https" | "wss" => 443,
"http" | "ws" => 80,
_ => return None,
};
if let Some(hp) = hostport.strip_prefix('[') {
let (h, after) = hp.split_once(']')?;
let port = after
.strip_prefix(':')
.and_then(|p| p.parse::<u16>().ok())
.unwrap_or(default_port);
return Some((h.to_string(), port));
}
match hostport.rsplit_once(':') {
Some((h, p)) if !h.is_empty() => {
let port = p.parse::<u16>().unwrap_or(default_port);
Some((h.to_string(), port))
}
_ => Some((hostport.to_string(), default_port)),
}
}
pub(crate) fn cached_addrs_from_doc(doc: &Value, host: &str, port: u16) -> Vec<SocketAddr> {
doc.get(host)
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str())
.filter_map(|s| s.parse::<IpAddr>().ok())
.map(|ip| SocketAddr::new(ip, port))
.collect()
})
.unwrap_or_default()
}
fn read_cached_addrs(host: &str, port: u16) -> Vec<SocketAddr> {
let Ok(text) = std::fs::read_to_string(dns_cache_path()) else {
return Vec::new();
};
let Ok(doc) = serde_json::from_str::<Value>(&text) else {
return Vec::new();
};
cached_addrs_from_doc(&doc, host, port)
}
fn write_cached_addrs(host: &str, addrs: &[SocketAddr]) {
if addrs.is_empty() {
return;
}
let path = dns_cache_path();
if let Some(dir) = path.parent() {
let _ = std::fs::create_dir_all(dir);
}
let mut doc = std::fs::read_to_string(&path)
.ok()
.and_then(|t| serde_json::from_str::<Value>(&t).ok())
.filter(|v| v.is_object())
.unwrap_or_else(|| json!({}));
let ips: Vec<String> = addrs.iter().map(|a| a.ip().to_string()).collect();
doc[host] = json!(ips);
let _ = std::fs::write(&path, doc.to_string());
}
pub(crate) async fn resolve_warm(host: &str, port: u16) -> Vec<SocketAddr> {
if host.parse::<IpAddr>().is_ok() {
return Vec::new();
}
let cached = read_cached_addrs(host, port);
let host_owned = format!("{host}:{port}");
let host_for_cache = host.to_string();
let fresh = tokio::spawn(async move {
let lookup = tokio::net::lookup_host(host_owned);
match tokio::time::timeout(Duration::from_millis(dns_timeout_ms()), lookup).await {
Ok(Ok(it)) => {
let addrs: Vec<SocketAddr> = it.collect();
if !addrs.is_empty() {
write_cached_addrs(&host_for_cache, &addrs);
}
addrs
}
_ => Vec::new(),
}
});
let result = if cached.is_empty() {
fresh.await.unwrap_or_default()
} else {
match tokio::time::timeout(Duration::from_millis(dns_race_ms()), fresh).await {
Ok(Ok(addrs)) if !addrs.is_empty() => addrs,
_ => cached,
}
};
prefer_v4(result)
}
fn prefer_v4(addrs: Vec<SocketAddr>) -> Vec<SocketAddr> {
let routable: Vec<SocketAddr> = addrs.iter().copied().filter(|a| source_ip_for(*a).is_some()).collect();
let mut result = if routable.is_empty() { addrs } else { routable };
result.sort_by_key(|a| a.is_ipv6());
result
}
#[derive(Clone)]
pub struct ServerConfig {
pub ice_servers: Vec<RTCIceServer>,
pub chunk_size: usize,
}
pub fn is_stun_only(s: &RTCIceServer) -> bool {
!s.urls.is_empty()
&& s.urls.iter().all(|u| {
let u = u.trim().to_ascii_lowercase();
u.starts_with("stun:") || u.starts_with("stuns:")
})
}
pub async fn fetch_config(server: &str) -> Result<ServerConfig> {
let body: Value = http_get_json(&format!("{server}/api/config")).await?;
let mut ice_servers = Vec::new();
if let Some(arr) = body["iceServers"].as_array() {
for s in arr {
let urls: Vec<String> = match &s["urls"] {
Value::String(u) => vec![u.clone()],
Value::Array(us) => us
.iter()
.filter_map(|u| u.as_str().map(|x| x.to_string()))
.collect(),
_ => continue,
};
ice_servers.push(RTCIceServer {
urls,
username: s["username"].as_str().unwrap_or_default().to_string(),
credential: s["credential"].as_str().unwrap_or_default().to_string(),
});
}
}
let chunk_size = body["chunkSize"].as_u64().unwrap_or(64 * 1024) as usize;
Ok(ServerConfig {
ice_servers,
chunk_size: chunk_size.min(MAX_DC_PAYLOAD),
})
}
pub async fn fetch_auto_room(server: &str) -> Result<String> {
let body: Value = http_get_json(&format!("{server}/api/room")).await?;
body["room"]
.as_str()
.map(|s| s.to_string())
.ok_or_else(|| anyhow!("no room in /api/room response"))
}
pub(crate) async fn http_get_json(url: &str) -> Result<Value> {
let host_port = host_port_from_url(url);
let overrides = match &host_port {
Some((host, port)) => {
let addrs = resolve_warm(host, *port).await;
(!addrs.is_empty()).then(|| (host.clone(), addrs))
}
None => None,
};
let mut last = None;
for attempt in 0..3 {
if attempt > 0 {
tokio::time::sleep(std::time::Duration::from_millis(700)).await;
}
let mut builder =
reqwest::Client::builder().timeout(std::time::Duration::from_secs(10));
if let Some((host, addrs)) = &overrides {
builder = builder.resolve_to_addrs(host, addrs);
}
let client = builder.build();
let Ok(client) = client else { continue };
match client.get(url).send().await {
Ok(resp) if resp.status().is_success() => match resp.json().await {
Ok(v) => return Ok(v),
Err(e) => last = Some(anyhow!(e)),
},
Ok(resp) => last = Some(anyhow!("GET {url} -> {}", resp.status())),
Err(e) => last = Some(anyhow!(e)),
}
}
Err(last.unwrap_or_else(|| anyhow!("GET {url} failed"))).with_context(|| format!("GET {url}"))
}
pub async fn connect_signaling(server: &str, tx: mpsc::UnboundedSender<Ev>) -> Result<Client> {
if let Some((host, port)) = host_port_from_url(server) {
let _ = resolve_warm(&host, port).await;
}
let fwd = |variant: fn(Value) -> Ev, tx: mpsc::UnboundedSender<Ev>| {
move |payload: Payload, _c: Client| {
let tx = tx.clone();
let v = match payload {
Payload::Text(mut vals) if !vals.is_empty() => Some(vals.remove(0)),
_ => None,
};
async move {
if let Some(v) = v {
let _ = tx.send(variant(v));
}
}
.boxed()
}
};
let down = {
let tx = tx.clone();
move |reason: &'static str| {
let tx = tx.clone();
move |_p: Payload, _c: Client| {
let tx = tx.clone();
async move {
let _ = tx.send(Ev::SignalingDown(reason.to_string()));
}
.boxed()
}
}
};
let sio = ClientBuilder::new(server)
.transport_type(TransportType::Websocket)
.reconnect(false)
.on(SioEvent::Connect, |_p: Payload, _c: Client| async {}.boxed())
.on(SioEvent::Close, down("close"))
.on(SioEvent::Error, down("error"))
.on("welcome", fwd(Ev::Welcome, tx.clone()))
.on("peer-joined", fwd(Ev::PeerJoined, tx.clone()))
.on("peer-left", fwd(Ev::PeerLeft, tx.clone()))
.on("signal", fwd(Ev::Signal, tx.clone()))
.on("pair-code", fwd(Ev::PairCode, tx.clone()))
.on("pair-ok", fwd(Ev::PairOk, tx.clone()))
.on("pair-matched", fwd(Ev::PairMatched, tx.clone()))
.on("pair-used", fwd(Ev::PairUsed, tx.clone()))
.on("pair-error", fwd(Ev::PairError, tx.clone()))
.on("known-peer", fwd(Ev::KnownPeer, tx.clone()))
.on("known-peer-left", fwd(Ev::KnownPeerLeft, tx.clone()))
.on("synced", fwd(Ev::Synced, tx.clone()))
.connect()
.await
.with_context(|| format!("socket.io connect to {server}"))?;
Ok(sio)
}
pub async fn reconnect_signaling(server: &str, tx: mpsc::UnboundedSender<Ev>) -> Result<Client> {
connect_signaling(server, tx).await
}
pub async fn heartbeat(sio: &Client, payload: Value, tx: mpsc::UnboundedSender<Ev>) {
let _ = sio
.emit_with_ack(
"sync",
payload,
std::time::Duration::from_secs(5),
move |_p: Payload, _c: Client| {
let tx = tx.clone();
async move {
let _ = tx.send(Ev::SignalingAlive);
}
.boxed()
},
)
.await;
}
pub async fn subscribe_with_ack(sio: &Client, channels: Vec<String>, tx: mpsc::UnboundedSender<Ev>) {
let _ = sio
.emit_with_ack(
"subscribe",
json!({ "channels": channels }),
std::time::Duration::from_secs(5),
move |payload: Payload, _c: Client| {
let tx = tx.clone();
async move {
if let Payload::Text(vals) = payload {
for p in roster_from_ack(&vals) {
let _ = tx.send(Ev::KnownPeer(p));
}
}
}
.boxed()
},
)
.await;
}
fn roster_from_ack(vals: &[Value]) -> Vec<Value> {
vals.iter()
.filter_map(|v| v["peers"].as_array())
.flatten()
.cloned()
.collect()
}
pub fn polite_role(my_uid: &str, peer_uid: Option<&str>, my_id: &str, peer_id: &str) -> bool {
match peer_uid {
Some(p) if p != my_uid => my_uid > p,
_ => my_id > peer_id,
}
}
pub struct Peer {
pub id: String,
pub polite: bool,
pub pc: Arc<RTCPeerConnection>,
state: Mutex<PeerSignalState>,
sio: Client,
closed: Arc<std::sync::atomic::AtomicBool>,
fps: Mutex<(Option<String>, Option<String>)>,
}
fn sdp_fingerprint(sdp: &str) -> Option<String> {
sdp.lines()
.find_map(|l| l.strip_prefix("a=fingerprint:"))
.map(|v| v.trim().to_uppercase())
}
struct PeerSignalState {
pending_candidates: Vec<RTCIceCandidateInit>,
has_remote: bool,
}
#[derive(Debug)]
pub enum SignalOutcome {
Handled,
Glare(Value),
}
impl Peer {
pub async fn connect(
peer_id: String,
polite: bool,
ice_servers: Vec<RTCIceServer>,
relay_only: bool,
sio: Client,
tx: mpsc::UnboundedSender<Ev>,
generation: u32,
) -> Result<Arc<Peer>> {
let mut m = MediaEngine::default();
m.register_default_codecs()?;
let mut registry = Registry::new();
registry = register_default_interceptors(registry, &mut m)?;
let mut se = SettingEngine::default();
#[cfg(feature = "test-hooks")]
if std::env::var("FILAMENT_DIRECT_LOOPBACK_ONLY").map(|v| v == "1").unwrap_or(false) {
se.set_ip_filter(Box::new(|ip: std::net::IpAddr| ip.is_loopback()));
se.set_ice_multicast_dns_mode(webrtc::ice::mdns::MulticastDnsMode::Disabled);
}
se.detach_data_channels(); let api = APIBuilder::new()
.with_media_engine(m)
.with_interceptor_registry(registry)
.with_setting_engine(se)
.build();
let pc = Arc::new(
api.new_peer_connection(RTCConfiguration {
ice_servers,
ice_transport_policy: if relay_only {
RTCIceTransportPolicy::Relay
} else {
RTCIceTransportPolicy::All
},
..Default::default()
})
.await?,
);
pc.on_ice_gathering_state_change(Box::new(move |s| {
crate::ui::trace(&format!("[ice-gathering] state={}", s));
Box::pin(async {})
}));
pc.on_ice_connection_state_change(Box::new(move |s| {
crate::ui::trace(&format!("[ice-connection] state={}", s));
Box::pin(async {})
}));
let closed = Arc::new(std::sync::atomic::AtomicBool::new(false));
{
let sio = sio.clone();
let to = peer_id.clone();
pc.on_ice_candidate(Box::new(move |c: Option<RTCIceCandidate>| {
let sio = sio.clone();
let to = to.clone();
Box::pin(async move {
if let Some(ref c) = c {
let typ = c.typ.to_string();
let addr = c.address.clone();
let port = c.port;
crate::ui::trace(&format!("[ice-candidate] type={} addr={} port={}", typ, addr, port));
if let Ok(init) = c.to_json() {
let _ = sio
.emit(
"signal",
json!({ "to": to, "data": { "type": "candidate", "candidate": init } }),
)
.await;
}
} else {
crate::ui::trace(&format!("[ice-candidate] gathering complete (null candidate)"));
}
})
}));
}
{
let tx = tx.clone();
let closed = closed.clone();
let pid = peer_id.clone();
pc.on_peer_connection_state_change(Box::new(move |s| {
crate::ui::trace(&format!("[pc-state] {} -> {}", pid, s));
if !closed.load(std::sync::atomic::Ordering::Relaxed) {
let _ = tx.send(Ev::PcState(pid.clone(), s.to_string()));
}
Box::pin(async {})
}));
}
if !polite {
let dc = pc.create_data_channel("filament", None).await?;
wire_channel(peer_id.clone(), dc, tx.clone(), closed.clone(), polite).await;
let offer = pc.create_offer(None).await?;
pc.set_local_description(offer).await?;
let ld = pc
.local_description()
.await
.ok_or_else(|| anyhow!("no local description"))?;
sio.emit(
"signal",
json!({ "to": peer_id, "data": { "type": "description", "description": advertise_max_message_size(&ld) } }),
)
.await
.ok();
} else {
let tx = tx.clone();
let closed = closed.clone();
let pid = peer_id.clone();
pc.on_data_channel(Box::new(move |dc: Arc<RTCDataChannel>| {
let tx = tx.clone();
let closed = closed.clone();
let pid = pid.clone();
Box::pin(async move {
wire_channel(pid, dc, tx, closed, polite).await;
})
}));
}
{
let pc = pc.clone();
let tx = tx.clone();
let pid = peer_id.clone();
let closed = closed.clone();
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(WATCHDOG_SECS)).await;
if !closed.load(std::sync::atomic::Ordering::Relaxed)
&& pc.connection_state() != RTCPeerConnectionState::Connected
{
let _ = tx.send(Ev::Stuck(pid, generation));
}
});
}
let peer = Arc::new(Peer {
id: peer_id,
polite,
pc,
state: Mutex::new(PeerSignalState {
pending_candidates: Vec::new(),
has_remote: false,
}),
sio,
closed,
fps: Mutex::new((None, None)),
});
if let Some(ld) = peer.pc.local_description().await {
peer.fps.lock().await.0 = sdp_fingerprint(&ld.sdp);
}
Ok(peer)
}
pub fn is_connected(&self) -> bool {
self.pc.connection_state() == RTCPeerConnectionState::Connected
}
pub async fn restart_ice(&self) {
let _ = self.pc.restart_ice().await;
if let Ok(offer) = self.pc.create_offer(None).await {
if self.pc.set_local_description(offer).await.is_ok() {
if let Some(ld) = self.pc.local_description().await {
let _ = self
.sio
.emit(
"signal",
json!({ "to": self.id, "data": { "type": "description", "description": advertise_max_message_size(&ld) } }),
)
.await;
}
}
}
}
pub fn mark_closed(&self) {
self.closed.store(true, std::sync::atomic::Ordering::Relaxed);
}
pub async fn close(&self) {
self.mark_closed();
let _ = self.pc.close().await;
}
pub async fn handle_signal(&self, data: Value) -> Result<SignalOutcome> {
match data["type"].as_str() {
Some("description") => {
let desc: RTCSessionDescription =
serde_json::from_value(data["description"].clone())
.context("parse remote description")?;
let is_offer = desc.sdp_type.to_string() == "offer";
let state = self.pc.signaling_state();
if desc.sdp_type == RTCSdpType::Offer && state != RTCSignalingState::Stable {
if !self.polite {
return Ok(SignalOutcome::Handled); }
return Ok(SignalOutcome::Glare(data));
}
if desc.sdp_type == RTCSdpType::Answer
&& state != RTCSignalingState::HaveLocalOffer
{
return Ok(SignalOutcome::Handled);
}
self.fps.lock().await.1 = sdp_fingerprint(&desc.sdp);
self.pc.set_remote_description(desc).await?;
let pending = {
let mut st = self.state.lock().await;
st.has_remote = true;
std::mem::take(&mut st.pending_candidates)
};
for c in pending {
if let Err(e) = self.pc.add_ice_candidate(c).await {
crate::ui::trace(&format!("filament: queued candidate failed: {e}"));
}
}
if is_offer {
let answer = self.pc.create_answer(None).await?;
self.pc.set_local_description(answer).await?;
let ld = self
.pc
.local_description()
.await
.ok_or_else(|| anyhow!("no local description"))?;
self.fps.lock().await.0 = sdp_fingerprint(&ld.sdp);
self.sio
.emit(
"signal",
json!({ "to": self.id, "data": { "type": "description", "description": advertise_max_message_size(&ld) } }),
)
.await
.ok();
}
}
Some("candidate") => {
let init: RTCIceCandidateInit =
serde_json::from_value(data["candidate"].clone()).context("parse candidate")?;
let buffered = {
let mut st = self.state.lock().await;
if st.has_remote {
false
} else {
st.pending_candidates.push(init.clone());
true
}
};
if !buffered {
if let Err(e) = self.pc.add_ice_candidate(init).await {
crate::ui::trace(&format!("filament: addIceCandidate failed: {e}"));
}
}
}
_ => {}
}
Ok(SignalOutcome::Handled)
}
pub async fn fingerprints(&self) -> Option<(String, String)> {
{
let mut f = self.fps.lock().await;
if f.0.is_none() {
if let Some(ld) = self.pc.local_description().await {
f.0 = sdp_fingerprint(&ld.sdp);
}
}
if f.1.is_none() {
if let Some(rd) = self.pc.remote_description().await {
f.1 = sdp_fingerprint(&rd.sdp);
}
}
if let (Some(a), Some(b)) = (&f.0, &f.1) {
return Some((a.clone(), b.clone()));
}
}
None
}
pub async fn route(&self) -> Option<String> {
let pair = self
.pc
.sctp()
.transport()
.ice_transport()
.get_selected_candidate_pair()
.await?;
if pair.local.typ == RTCIceCandidateType::Relay
|| pair.remote.typ == RTCIceCandidateType::Relay
{
return Some("relay".into());
}
let same_host =
pair.local.address == pair.remote.address || is_own_addr(&pair.remote.address);
if same_host {
return Some("direct over lo".into());
}
let iface = resolve_iface_name(&pair.local.address);
if is_tailscale_addr(&pair.local.address) || is_tailscale_addr(&pair.remote.address) {
return Some(format!("direct over {iface}"));
}
let both_private = is_private_addr(&pair.local.address) && is_private_addr(&pair.remote.address);
if both_private {
return Some(format!("direct over {iface}"));
}
Some(format!("direct over {iface}"))
}
pub async fn path_detail(&self) -> Option<PathPair> {
let pair = self
.pc
.sctp()
.transport()
.ice_transport()
.get_selected_candidate_pair()
.await?;
let typ = |t: RTCIceCandidateType| {
match t {
RTCIceCandidateType::Host => "host",
RTCIceCandidateType::Srflx => "srflx",
RTCIceCandidateType::Prflx => "prflx",
RTCIceCandidateType::Relay => "relay",
_ => "?",
}
.to_string()
};
Some(PathPair {
local_ip: pair.local.address.clone(),
local_typ: typ(pair.local.typ),
remote_ip: pair.remote.address.clone(),
remote_port: pair.remote.port,
remote_typ: typ(pair.remote.typ),
relayed: pair.local.typ == RTCIceCandidateType::Relay
|| pair.remote.typ == RTCIceCandidateType::Relay,
})
}
}
#[derive(Debug, Clone)]
pub struct PathPair {
pub local_ip: String,
pub local_typ: String,
pub remote_ip: String,
pub remote_port: u16,
pub remote_typ: String,
pub relayed: bool,
}
#[derive(Debug, Clone, Default)]
pub struct PathInfo {
pub iface: Option<String>,
pub vpn: bool,
pub class: Option<String>,
pub local: Option<String>,
pub remote: Option<String>,
pub cand: Option<String>,
pub relay: bool,
}
impl PathInfo {
pub fn to_json(&self) -> Value {
let mut m = serde_json::Map::new();
if let Some(i) = &self.iface {
m.insert("iface".into(), json!(i));
m.insert("vpn".into(), json!(self.vpn));
}
if let Some(c) = &self.class {
m.insert("class".into(), json!(c));
}
if let Some(l) = &self.local {
m.insert("local".into(), json!(l));
}
if let Some(r) = &self.remote {
m.insert("remote".into(), json!(r));
}
if let Some(c) = &self.cand {
m.insert("cand".into(), json!(c));
}
m.insert("relay".into(), json!(self.relay));
Value::Object(m)
}
}
pub async fn describe_path(t: &dyn Transport, peer: Option<&Peer>) -> PathInfo {
let mut info = PathInfo::default();
if let Some(ra) = t.remote_addr() {
info.remote = Some(ra.to_string());
info.class = Some(crate::doctor::ip_class(ra.ip()));
if let Some(la) = t.local_ip().or_else(|| source_ip_for(ra)) {
info.local = Some(la.to_string());
if let Some((n, v)) = crate::doctor::iface_for_ip(la) {
info.iface = Some(n);
info.vpn = v;
}
}
} else if let Some(p) = peer {
if let Some(pp) = p.path_detail().await {
info.remote = Some(format!("{}:{}", pp.remote_ip, pp.remote_port));
info.local = Some(pp.local_ip.clone());
info.cand = Some(format!("{}\u{2194}{}", pp.local_typ, pp.remote_typ));
info.relay = pp.relayed;
if let Ok(rip) = pp.remote_ip.parse::<std::net::IpAddr>() {
info.class = Some(crate::doctor::ip_class(rip));
}
if let Ok(lip) = pp.local_ip.parse::<std::net::IpAddr>() {
if let Some((n, v)) = crate::doctor::iface_for_ip(lip) {
info.iface = Some(n);
info.vpn = v;
}
}
}
}
info
}
pub const ADVERTISED_MAX_MESSAGE: u32 = 262144;
fn advertise_max_message_size(desc: &RTCSessionDescription) -> Value {
let mut sdp = desc.sdp.clone();
if !sdp.contains("max-message-size") && sdp.contains("m=application") {
if !sdp.ends_with('\n') {
sdp.push_str("\r\n");
}
sdp.push_str(&format!("a=max-message-size:{ADVERTISED_MAX_MESSAGE}\r\n"));
}
json!({ "type": desc.sdp_type.to_string(), "sdp": sdp })
}
pub fn is_own_addr(addr: &str) -> bool {
let Ok(ip) = addr.parse::<std::net::IpAddr>() else {
return false;
};
let bind = if ip.is_ipv4() { "0.0.0.0:0" } else { "[::]:0" };
std::net::UdpSocket::bind(bind)
.and_then(|s| {
s.connect((ip, 9))?;
s.local_addr()
})
.map(|la| la.ip() == ip)
.unwrap_or(false)
}
pub fn source_ip_for(remote: std::net::SocketAddr) -> Option<std::net::IpAddr> {
let bind = if remote.is_ipv4() { "0.0.0.0:0" } else { "[::]:0" };
std::net::UdpSocket::bind(bind)
.and_then(|s| {
s.connect(remote)?;
s.local_addr()
})
.ok()
.map(|la| la.ip())
.filter(|ip| !ip.is_unspecified())
}
pub fn is_private_addr(addr: &str) -> bool {
match addr.parse::<std::net::IpAddr>() {
Ok(std::net::IpAddr::V4(v4)) => {
v4.is_loopback()
|| v4.is_private()
|| v4.is_link_local()
|| (v4.octets()[0] == 100 && (v4.octets()[1] & 0xc0) == 64)
}
Ok(std::net::IpAddr::V6(v6)) => {
v6.is_loopback()
|| (v6.segments()[0] & 0xfe00) == 0xfc00 || (v6.segments()[0] & 0xffc0) == 0xfe80 }
Err(_) => false,
}
}
pub fn is_tailscale_addr(addr: &str) -> bool {
match addr.parse::<std::net::IpAddr>() {
Ok(std::net::IpAddr::V4(v4)) => {
v4.octets()[0] == 100 && (v4.octets()[1] & 0xc0) == 64
}
_ => false,
}
}
fn resolve_iface_name(addr: &str) -> String {
if let Ok(ip) = addr.parse::<std::net::IpAddr>() {
for iface in crate::interact::enumerate_interfaces() {
if iface.ips.iter().any(|i| *i == ip) {
return iface.name;
}
}
}
"?".to_string()
}
async fn wire_channel(
peer_id: String,
dc: Arc<RTCDataChannel>,
tx: mpsc::UnboundedSender<Ev>,
closed: Arc<std::sync::atomic::AtomicBool>,
polite: bool,
) {
let dc2 = dc.clone();
dc.on_open(Box::new(move || {
let dc2 = dc2.clone();
let tx = tx.clone();
let closed = closed.clone();
Box::pin(async move {
let raw = match dc2.detach().await {
Ok(raw) => raw,
Err(e) => {
crate::ui::trace(&format!("filament: data channel detach failed: {e}"));
return;
}
};
let drained = Arc::new(Notify::new());
let dead = Arc::new(std::sync::atomic::AtomicBool::new(false));
let last_activity = Arc::new(std::sync::atomic::AtomicU64::new(now_ms()));
let first_data = Arc::new(std::sync::atomic::AtomicBool::new(false));
raw.set_buffered_amount_low_threshold(LOW_WATER);
{
let drained = drained.clone();
raw.on_buffered_amount_low(Box::new(move || {
let drained = drained.clone();
Box::pin(async move {
drained.notify_waiters();
})
}));
}
{
let raw = raw.clone();
let tx = tx.clone();
let closed = closed.clone();
let dead = dead.clone();
let drained = drained.clone();
let last_activity = last_activity.clone();
let first_data = first_data.clone();
let peer_id = peer_id.clone();
tokio::spawn(async move {
let mut buf = vec![0u8; READ_BUF];
loop {
match raw.read_data_channel(&mut buf).await {
Ok((0, _)) | Err(_) => {
dead.store(true, std::sync::atomic::Ordering::Relaxed);
drained.notify_waiters(); break;
}
Ok((n, true)) => {
if !closed.load(std::sync::atomic::Ordering::Relaxed) {
if let Ok(v) = serde_json::from_slice::<Value>(&buf[..n]) {
let _ = tx.send(Ev::Control(peer_id.clone(), v));
}
}
}
Ok((n, false)) => {
if n >= 4 && !closed.load(std::sync::atomic::Ordering::Relaxed) {
last_activity
.store(now_ms(), std::sync::atomic::Ordering::Relaxed);
first_data
.store(true, std::sync::atomic::Ordering::Relaxed);
let sid = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]);
let _ = tx.send(Ev::Chunk(
peer_id.clone(),
sid,
None,
Bytes::copy_from_slice(&buf[4..n]),
));
}
}
}
}
});
}
if !closed.load(std::sync::atomic::Ordering::Relaxed) {
{
let raw_k = raw.clone();
let dead_k = dead.clone();
let act_k = last_activity.clone();
tokio::spawn(async move {
let frame = Bytes::from(json!({ "type": "ping", "reason": "keepalive" }).to_string());
loop {
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
if dead_k.load(std::sync::atomic::Ordering::Relaxed) {
break;
}
if raw_k.write_data_channel(&frame, true).await.is_err() {
break;
}
act_k.store(now_ms(), std::sync::atomic::Ordering::Relaxed);
}
});
}
let transport: Arc<dyn Transport> =
Arc::new(DataChannelTransport { raw, drained, dead, last_activity, first_data, answerer: polite });
let _ = tx.send(Ev::ChannelReady(peer_id.clone(), transport));
}
})
}));
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn polite_role_prefers_uid_then_sid() {
assert!(polite_role("uid-z", Some("uid-a"), "s1", "s9"));
assert!(!polite_role("uid-a", Some("uid-z"), "s9", "s1"));
assert!(polite_role("u", None, "s9", "s1"));
assert!(!polite_role("u", None, "s1", "s9"));
assert!(polite_role("u", Some("u"), "s9", "s1"));
}
#[test]
fn roster_from_ack_extracts_present_peers() {
let ack = json!({
"ok": true, "n": 1,
"peers": [
{"id": "sidA", "name": "dovm", "uid": "cli-r-1", "channel": "ch"},
{"id": "sidB", "name": "dovm2", "uid": "cli-r-2", "channel": "ch"}
]
});
let out = roster_from_ack(&[ack]);
assert_eq!(out.len(), 2);
assert_eq!(out[0]["id"], "sidA");
assert_eq!(out[1]["id"], "sidB");
assert_eq!(out[0]["channel"], "ch");
}
#[test]
fn host_port_parses_scheme_defaults_and_explicit() {
assert_eq!(
host_port_from_url("https://api.filament.autumated.com"),
Some(("api.filament.autumated.com".to_string(), 443))
);
assert_eq!(
host_port_from_url("http://example.com/api/config?x=1"),
Some(("example.com".to_string(), 80))
);
assert_eq!(
host_port_from_url("https://example.com:8443/api/config"),
Some(("example.com".to_string(), 8443))
);
assert_eq!(
host_port_from_url("wss://relay.example.com"),
Some(("relay.example.com".to_string(), 443))
);
assert_eq!(
host_port_from_url("https://user:pass@host.example/path"),
Some(("host.example".to_string(), 443))
);
assert_eq!(
host_port_from_url("https://[2606:4700::1]:443/api"),
Some(("2606:4700::1".to_string(), 443))
);
assert_eq!(
host_port_from_url("http://[::1]"),
Some(("::1".to_string(), 80))
);
assert_eq!(host_port_from_url("ftp://example.com"), None);
assert_eq!(host_port_from_url("not a url"), None);
}
#[test]
fn cached_addrs_select_host_and_apply_port() {
let doc = json!({
"api.filament.autumated.com": ["104.21.73.85", "172.67.142.8"],
"other.example": ["10.0.0.1"]
});
let mut got = cached_addrs_from_doc(&doc, "api.filament.autumated.com", 443);
got.sort();
assert_eq!(got.len(), 2);
assert!(got.contains(&"104.21.73.85:443".parse::<SocketAddr>().unwrap()));
assert!(got.contains(&"172.67.142.8:443".parse::<SocketAddr>().unwrap()));
let other = cached_addrs_from_doc(&doc, "other.example", 8443);
assert_eq!(other, vec!["10.0.0.1:8443".parse::<SocketAddr>().unwrap()]);
}
#[test]
fn cached_addrs_robust_to_missing_or_malformed() {
assert!(cached_addrs_from_doc(&json!({}), "nope.example", 443).is_empty());
assert!(cached_addrs_from_doc(&json!({"h": "oops"}), "h", 443).is_empty());
let doc = json!({"h": ["not-an-ip", "192.0.2.7", 42, null]});
assert_eq!(
cached_addrs_from_doc(&doc, "h", 443),
vec!["192.0.2.7:443".parse::<SocketAddr>().unwrap()]
);
}
#[test]
fn roster_from_ack_robust_to_empty_or_missing() {
assert!(roster_from_ack(&[json!({"ok": true, "n": 1, "peers": []})]).is_empty());
assert!(roster_from_ack(&[json!({"ok": false, "n": 0})]).is_empty());
assert!(roster_from_ack(&[json!({"peers": "oops"})]).is_empty());
assert!(roster_from_ack(&[]).is_empty());
}
}