NetDesk 1.5.0 — source release
This commit is contained in:
NetDesk committed
commit
228c492a5c
227 files changed
+80714
No files matched your search
@@ -0,0 +1,831 @@
|
||||
//! Relay-first P2P multiplexer with seamless fallback.
|
||||
//!
|
||||
//! [`upgrade_relay`] wraps any [`Transport`] (the WSS relay) and returns another
|
||||
//! `Transport` that the session loop ([`crate::signaling::connect_with`] /
|
||||
//! [`crate::host::run_host_with`]) uses completely unchanged. All the P2P logic
|
||||
//! runs in a background engine task that owns both the relay and, once
|
||||
//! negotiated, the WebRTC DataChannel:
|
||||
//!
|
||||
//! * The relay is LIVE from the first frame and is NEVER torn down while P2P is
|
||||
//! up (warm standby), so the first frame never waits on ICE and a P2P drop is
|
||||
//! invisible to the session.
|
||||
//! * The engine SNIFFS the session `sid` out of the handshake messages flowing
|
||||
//! through it (so neither loop has to hand it over), then drives the
|
||||
//! relay-brokered `rtc.*` handshake ([`crate::rtc::upgrade_offerer`] /
|
||||
//! [`crate::rtc::answer_offer`]).
|
||||
//! * UP edge: when the DataChannel opens AND the nominated ICE pair is DIRECT
|
||||
//! (non-TURN — [`crate::rtc::selected_pair_is_direct`]), media + the
|
||||
//! latency-critical input/keyframe control are routed over the channel; a
|
||||
//! TURN-relayed or unknown pair keeps media on the relay.
|
||||
//! * DOWN edge: if the peer connection fails/closes, media reroutes to the
|
||||
//! still-open relay with no `Ended` and no frame loss.
|
||||
//!
|
||||
//! `rtc.*` and `session.*` ALWAYS ride the relay; only `CHAN_VIDEO` / file
|
||||
//! binary and `i.mouse` / `i.key` / `keyframe` move to P2P when it is direct.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use bytes::Bytes;
|
||||
|
||||
use serde_json::Value;
|
||||
use tokio::sync::mpsc;
|
||||
use webrtc::api::API;
|
||||
use webrtc::data_channel::RTCDataChannel;
|
||||
use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
|
||||
use webrtc::peer_connection::RTCPeerConnection;
|
||||
use webrtc::stats::StatsReportType;
|
||||
|
||||
use crate::error::{NdError, Result};
|
||||
use crate::rtc;
|
||||
use crate::rtc::{RtcIceServerJson, WebDetached};
|
||||
use crate::transport::{install_reassembler, send_fragmented, Lane, Transport, TransportMsg};
|
||||
|
||||
/// A negotiated P2P connection: the native single reliable channel (16 KiB
|
||||
/// app-fragmented) or the web 3-channel topology (media/ctl/move, detached,
|
||||
/// whole frames) used to interop with the browser / Python host.
|
||||
enum P2pConn {
|
||||
Single(Arc<RTCDataChannel>),
|
||||
Web(Arc<WebDetached>),
|
||||
}
|
||||
|
||||
/// High-frequency, latency-critical control that rides the web `ctl` channel
|
||||
/// (the web viewer's send override routes exactly these; other control stays on
|
||||
/// the relay). Cheap prefix checks, no JSON parse on the hot path.
|
||||
fn is_fast_path_text(t: &str) -> bool {
|
||||
t.contains("\"t\":\"i.mouse\"")
|
||||
|| t.contains("\"t\":\"i.key\"")
|
||||
|| t.contains("\"t\":\"keyframe\"")
|
||||
}
|
||||
|
||||
/// Whether the direct-P2P fast path is enabled for this process. DEFAULT-ON:
|
||||
/// the native-peer capability gate keeps P2P to native<->native pairs and every
|
||||
/// other pairing stays on the relay, so default-on cannot break the existing
|
||||
/// web/Python peers. Opt OUT with `NETDESK_P2P=0` (also `false`/`off`/`no`).
|
||||
pub fn p2p_enabled() -> bool {
|
||||
!matches!(
|
||||
std::env::var("NETDESK_P2P").as_deref(),
|
||||
Ok("0") | Ok("false") | Ok("off") | Ok("no") | Ok("OFF") | Ok("FALSE") | Ok("NO")
|
||||
)
|
||||
}
|
||||
|
||||
/// Performance-based path health for the active DIRECT P2P pair.
|
||||
///
|
||||
/// The relay-first state machine already prefers a direct pair and falls back on
|
||||
/// hard failure. This adds the other half of the hybrid: abandon a direct pair
|
||||
/// that is measurably WORSE than a relay hop would be — not just one that
|
||||
/// dropped. It is deliberately conservative (only a *pathologically* high,
|
||||
/// *sustained* RTT trips it; a VPS relay hop is almost always below that), so it
|
||||
/// never fights the normal "direct is faster" case. Feed it the nominated
|
||||
/// candidate pair's RTT (ms) every ~[`HEALTH_INTERVAL`]; `on_rtt` returns true
|
||||
/// once the path has been bad long enough to downgrade.
|
||||
#[derive(Default)]
|
||||
pub struct P2pHealth {
|
||||
bad_streak: u32,
|
||||
}
|
||||
|
||||
/// How often the engine samples the active pair's RTT.
|
||||
pub const HEALTH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(2);
|
||||
/// A direct-pair RTT at/above this (ms) is "bad" — far worse than a relay hop.
|
||||
pub const P2P_RTT_BAD_MS: u32 = 500;
|
||||
/// Consecutive bad samples before downgrading (~10 s at [`HEALTH_INTERVAL`]).
|
||||
pub const P2P_RTT_CONFIRM: u32 = 5;
|
||||
|
||||
impl P2pHealth {
|
||||
/// Record one RTT sample (ms). `0` means "unknown" (not yet measured) and is
|
||||
/// ignored. Returns true when a downgrade to the relay is warranted.
|
||||
pub fn on_rtt(&mut self, rtt_ms: u32) -> bool {
|
||||
if rtt_ms == 0 {
|
||||
return false;
|
||||
}
|
||||
if rtt_ms >= P2P_RTT_BAD_MS {
|
||||
self.bad_streak += 1;
|
||||
} else {
|
||||
self.bad_streak = 0;
|
||||
}
|
||||
self.bad_streak >= P2P_RTT_CONFIRM
|
||||
}
|
||||
}
|
||||
|
||||
/// The nominated ICE candidate-pair RTT in ms, or 0 if unknown/not yet measured.
|
||||
async fn p2p_pair_rtt_ms(pc: &Arc<RTCPeerConnection>) -> u32 {
|
||||
let report = pc.get_stats().await;
|
||||
for v in report.reports.values() {
|
||||
if let StatsReportType::CandidatePair(cp) = v {
|
||||
if cp.nominated && cp.current_round_trip_time > 0.0 {
|
||||
return (cp.current_round_trip_time * 1000.0).round() as u32;
|
||||
}
|
||||
}
|
||||
}
|
||||
0
|
||||
}
|
||||
|
||||
/// Which side of the upgrade handshake this peer drives.
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
|
||||
pub enum P2pRole {
|
||||
/// Viewer — sends `rtc.config.req` + the offer.
|
||||
Offerer,
|
||||
/// Host — answers an inbound `rtc.offer`.
|
||||
Answerer,
|
||||
}
|
||||
|
||||
/// SCTP send-queue depth (bytes) past which a fresh MEDIA (video) frame is
|
||||
/// DROPPED rather than queued — fresh beats late. Only the drop-tolerant Media
|
||||
/// lane drops; Ctl (file bytes / control) and Move are always sent reliably.
|
||||
/// Matches the web host's 256 KiB `_send_bin` threshold.
|
||||
const P2P_MEDIA_CAP: usize = 262144;
|
||||
|
||||
/// A handle to force the session back onto the relay (the plan's "force-relay
|
||||
/// fast path", and the test hook for the DOWN edge). Cloneable; outlives the
|
||||
/// transport move into the session loop.
|
||||
#[derive(Clone)]
|
||||
pub struct P2pControl {
|
||||
down: mpsc::UnboundedSender<()>,
|
||||
}
|
||||
|
||||
impl P2pControl {
|
||||
/// Drop any active P2P path and route all media back over the warm relay.
|
||||
pub fn force_relay(&self) {
|
||||
let _ = self.down.send(());
|
||||
}
|
||||
}
|
||||
|
||||
/// The app-facing transport the session loop uses unchanged — just the mailbox
|
||||
/// into the engine task.
|
||||
pub struct P2pRelay {
|
||||
out: mpsc::UnboundedSender<(TransportMsg, Lane)>,
|
||||
inn: mpsc::UnboundedReceiver<Result<TransportMsg>>,
|
||||
down: mpsc::UnboundedSender<()>,
|
||||
_engine: tokio::task::JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl P2pRelay {
|
||||
/// A control handle to force media back onto the relay.
|
||||
pub fn control(&self) -> P2pControl {
|
||||
P2pControl {
|
||||
down: self.down.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl Transport for P2pRelay {
|
||||
async fn send_binary(&mut self, data: Vec<u8>) -> Result<()> {
|
||||
// No lane hint -> treat as reliable Ctl (never dropped). Callers that
|
||||
// know it is video use send_binary_lane(.., Media).
|
||||
self.out
|
||||
.send((TransportMsg::Binary(data), Lane::Ctl))
|
||||
.map_err(|_| NdError::Connect("p2p engine stopped".into()))
|
||||
}
|
||||
async fn send_binary_lane(&mut self, data: Vec<u8>, lane: Lane) -> Result<()> {
|
||||
self.out
|
||||
.send((TransportMsg::Binary(data), lane))
|
||||
.map_err(|_| NdError::Connect("p2p engine stopped".into()))
|
||||
}
|
||||
async fn send_text(&mut self, text: String) -> Result<()> {
|
||||
self.out
|
||||
.send((TransportMsg::Text(text), Lane::Ctl))
|
||||
.map_err(|_| NdError::Connect("p2p engine stopped".into()))
|
||||
}
|
||||
async fn recv(&mut self) -> Option<Result<TransportMsg>> {
|
||||
self.inn.recv().await
|
||||
}
|
||||
}
|
||||
|
||||
/// Wrap `relay` with transparent P2P upgrade + fallback. `api` builds the peer
|
||||
/// connection(s). The returned transport is a drop-in for the relay.
|
||||
pub fn upgrade_relay(relay: Box<dyn Transport>, role: P2pRole, api: Arc<API>) -> P2pRelay {
|
||||
let (app_out_tx, app_out_rx) = mpsc::unbounded_channel::<(TransportMsg, Lane)>();
|
||||
let (app_in_tx, app_in_rx) = mpsc::unbounded_channel::<Result<TransportMsg>>();
|
||||
let (down_tx, down_rx) = mpsc::unbounded_channel::<()>();
|
||||
let engine = tokio::spawn(engine(
|
||||
relay,
|
||||
role,
|
||||
api,
|
||||
app_out_rx,
|
||||
app_in_tx,
|
||||
down_tx.clone(),
|
||||
down_rx,
|
||||
));
|
||||
P2pRelay {
|
||||
out: app_out_tx,
|
||||
inn: app_in_rx,
|
||||
down: down_tx,
|
||||
_engine: engine,
|
||||
}
|
||||
}
|
||||
|
||||
/// Pull `sid` out of a control-text message, if present and non-empty.
|
||||
fn sniff_sid_text(t: &str) -> Option<String> {
|
||||
serde_json::from_str::<Value>(t)
|
||||
.ok()
|
||||
.and_then(|v| v.get("sid").and_then(|x| x.as_str()).map(|s| s.to_string()))
|
||||
.filter(|s| !s.is_empty())
|
||||
}
|
||||
|
||||
/// Whether this control text rides the P2P channel when it is up. Everything
|
||||
/// does EXCEPT `session.*` (session lifecycle) and `rtc.*` (upgrade signaling),
|
||||
/// which stay on the relay. Keeping ALL app traffic — input, keyframe requests
|
||||
/// AND file.* control — on the one ordered channel alongside the binary data is
|
||||
/// essential: splitting data (P2P) from its `file.done` (relay) lets the text
|
||||
/// race ahead of the slower-fragmented data and truncate the download. Cheap
|
||||
/// substring checks avoid a JSON parse on the hot path (serde emits `t` first).
|
||||
fn route_text_p2p(t: &str) -> bool {
|
||||
!(t.contains("\"t\":\"session.") || t.contains("\"t\":\"rtc."))
|
||||
}
|
||||
|
||||
type DoneTx = mpsc::UnboundedSender<(Arc<RTCPeerConnection>, P2pConn)>;
|
||||
|
||||
fn spawn_offerer(
|
||||
api: Arc<API>,
|
||||
sid: String,
|
||||
rtc_out: mpsc::UnboundedSender<String>,
|
||||
mut rtc_in: mpsc::UnboundedReceiver<Value>,
|
||||
done: DoneTx,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
let send = {
|
||||
let tx = rtc_out.clone();
|
||||
move |s: String| {
|
||||
let _ = tx.send(s);
|
||||
}
|
||||
};
|
||||
match rtc::upgrade_offerer(&api, &sid, None, "netdesk", &send, &mut rtc_in).await {
|
||||
Ok((pc, dc)) => {
|
||||
let _ = done.send((pc, P2pConn::Single(dc)));
|
||||
}
|
||||
Err(e) => tracing::debug!("P2P offerer upgrade abandoned: {e}"),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// WEB-topology offerer (interop with the browser / Python host). Builds its own
|
||||
/// detach-enabled API, negotiates media/ctl/move, and reports a `P2pConn::Web`.
|
||||
fn spawn_offerer_web(
|
||||
sid: String,
|
||||
rtc_out: mpsc::UnboundedSender<String>,
|
||||
mut rtc_in: mpsc::UnboundedReceiver<Value>,
|
||||
done: DoneTx,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
let api = match rtc::build_api_web() {
|
||||
Ok(a) => a,
|
||||
Err(e) => {
|
||||
tracing::debug!("build_api_web: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let send = {
|
||||
let tx = rtc_out.clone();
|
||||
move |s: String| {
|
||||
let _ = tx.send(s);
|
||||
}
|
||||
};
|
||||
match rtc::upgrade_offerer_web(&api, &sid, None, &send, &mut rtc_in).await {
|
||||
Ok((pc, web)) => {
|
||||
let _ = done.send((pc, P2pConn::Web(Arc::new(web))));
|
||||
}
|
||||
Err(e) => tracing::debug!("P2P web offerer abandoned: {e}"),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn spawn_answerer(
|
||||
api: Arc<API>,
|
||||
sid: String,
|
||||
offer_sdp: String,
|
||||
ice: Vec<RtcIceServerJson>,
|
||||
rtc_out: mpsc::UnboundedSender<String>,
|
||||
done: DoneTx,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
let send = {
|
||||
let tx = rtc_out.clone();
|
||||
move |s: String| {
|
||||
let _ = tx.send(s);
|
||||
}
|
||||
};
|
||||
match rtc::answer_offer(&api, &sid, &offer_sdp, ice, &send).await {
|
||||
Ok((pc, dc)) => {
|
||||
let _ = done.send((pc, P2pConn::Single(dc)));
|
||||
}
|
||||
Err(e) => tracing::debug!("P2P answerer abandoned: {e}"),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// WEB-topology answerer (a browser / web viewer connects to this native host).
|
||||
/// Builds its own detach-enabled API and reports a `P2pConn::Web`.
|
||||
fn spawn_answerer_web(
|
||||
sid: String,
|
||||
offer_sdp: String,
|
||||
ice: Vec<RtcIceServerJson>,
|
||||
rtc_out: mpsc::UnboundedSender<String>,
|
||||
done: DoneTx,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
let api = match rtc::build_api_web() {
|
||||
Ok(a) => a,
|
||||
Err(e) => {
|
||||
tracing::debug!("build_api_web: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let send = {
|
||||
let tx = rtc_out.clone();
|
||||
move |s: String| {
|
||||
let _ = tx.send(s);
|
||||
}
|
||||
};
|
||||
match rtc::answer_offer_web(&api, &sid, &offer_sdp, ice, &send).await {
|
||||
Ok((pc, web)) => {
|
||||
let _ = done.send((pc, P2pConn::Web(Arc::new(web))));
|
||||
}
|
||||
Err(e) => tracing::debug!("P2P web answerer abandoned: {e}"),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Start one offerer upgrade attempt (the initial one, or a re-upgrade after a
|
||||
/// drop). Creates a fresh rtc_in channel, spawns the Single or Web offerer, and
|
||||
/// returns the sender the relay loop forwards inbound rtc.* to.
|
||||
fn start_offerer_attempt(
|
||||
web: bool,
|
||||
api: Arc<API>,
|
||||
sid: String,
|
||||
rtc_out: mpsc::UnboundedSender<String>,
|
||||
done: DoneTx,
|
||||
) -> mpsc::UnboundedSender<Value> {
|
||||
let (tx, rx) = mpsc::unbounded_channel::<Value>();
|
||||
if web {
|
||||
spawn_offerer_web(sid, rtc_out, rx, done);
|
||||
} else {
|
||||
spawn_offerer(api, sid, rtc_out, rx, done);
|
||||
}
|
||||
tx
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn engine(
|
||||
mut relay: Box<dyn Transport>,
|
||||
role: P2pRole,
|
||||
api: Arc<API>,
|
||||
mut app_out_rx: mpsc::UnboundedReceiver<(TransportMsg, Lane)>,
|
||||
app_in_tx: mpsc::UnboundedSender<Result<TransportMsg>>,
|
||||
down_tx: mpsc::UnboundedSender<()>,
|
||||
mut down_rx: mpsc::UnboundedReceiver<()>,
|
||||
) {
|
||||
let mut sid: Option<String> = None;
|
||||
let mut media_on_p2p = false;
|
||||
let mut p2p_send: Option<P2pConn> = None;
|
||||
let mut _pc: Option<Arc<RTCPeerConnection>> = None;
|
||||
let mut next_frame_id: u32 = 0;
|
||||
let mut offerer_started = false;
|
||||
let mut answerer_started = false;
|
||||
// Performance health of the active direct pair, and a latch: once a direct
|
||||
// pair is abandoned for being too slow, stay on the relay for the rest of
|
||||
// the session (don't re-upgrade into the same bad path and flap).
|
||||
let mut health = P2pHealth::default();
|
||||
let mut health_down = false;
|
||||
|
||||
// Inbound P2P frames (reassembled) land here; held open for the engine's
|
||||
// lifetime so re-use across the session is clean. Fed by install_reassembler
|
||||
// once a channel is established.
|
||||
let (p2p_in_tx, mut p2p_in_rx) = mpsc::unbounded_channel::<TransportMsg>();
|
||||
// rtc.* the upgrade task wants put on the relay.
|
||||
let (rtc_out_tx, mut rtc_out_rx) = mpsc::unbounded_channel::<String>();
|
||||
// Inbound rtc.* forwarded to the CURRENT offerer upgrade attempt. Re-created
|
||||
// per attempt so a re-upgrade after a drop gets a fresh channel.
|
||||
let mut rtc_in_tx: Option<mpsc::UnboundedSender<Value>> = None;
|
||||
// Negotiated (pc, connection).
|
||||
let (done_tx, mut done_rx) =
|
||||
mpsc::unbounded_channel::<(Arc<RTCPeerConnection>, P2pConn)>();
|
||||
// Re-upgrade (hysteresis): after a P2P drop, retry the upgrade once after a
|
||||
// backoff so a transient drop recovers to P2P instead of staying on the
|
||||
// relay. Default 15s; shorter in tests via NETDESK_P2P_REUP_MS.
|
||||
let reup_ms: u64 = std::env::var("NETDESK_P2P_REUP_MS")
|
||||
.ok()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(15000);
|
||||
let (reup_tx, mut reup_rx) = mpsc::unbounded_channel::<()>();
|
||||
// Offerer topology once decided: Some(false)=native single, Some(true)=web,
|
||||
// None=relay-only. Remembered so a re-upgrade uses the same topology.
|
||||
let mut offerer_mode: Option<bool> = None;
|
||||
// Periodic RTT health probe of the active direct pair.
|
||||
let mut health_tick = tokio::time::interval(HEALTH_INTERVAL);
|
||||
health_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
|
||||
enum Ev {
|
||||
AppOut(Option<(TransportMsg, Lane)>),
|
||||
RelayIn(Option<Result<TransportMsg>>),
|
||||
P2pIn(Option<TransportMsg>),
|
||||
RtcOut(Option<String>),
|
||||
Done(Option<(Arc<RTCPeerConnection>, P2pConn)>),
|
||||
ReUpgrade,
|
||||
Down,
|
||||
Health,
|
||||
}
|
||||
|
||||
loop {
|
||||
let ev = tokio::select! {
|
||||
m = app_out_rx.recv() => Ev::AppOut(m),
|
||||
m = relay.recv() => Ev::RelayIn(m),
|
||||
m = p2p_in_rx.recv() => Ev::P2pIn(m),
|
||||
s = rtc_out_rx.recv() => Ev::RtcOut(s),
|
||||
d = done_rx.recv() => Ev::Done(d),
|
||||
_ = reup_rx.recv() => Ev::ReUpgrade,
|
||||
_ = down_rx.recv() => Ev::Down,
|
||||
_ = health_tick.tick() => Ev::Health,
|
||||
};
|
||||
|
||||
match ev {
|
||||
// ---- outbound from the session loop ----
|
||||
Ev::AppOut(None) => break, // the session dropped the transport
|
||||
Ev::AppOut(Some((msg, lane))) => {
|
||||
if sid.is_none() {
|
||||
if let TransportMsg::Text(t) = &msg {
|
||||
if let Some(s) = sniff_sid_text(t) {
|
||||
sid = Some(s);
|
||||
}
|
||||
}
|
||||
}
|
||||
match msg {
|
||||
TransportMsg::Binary(b) => match (media_on_p2p, p2p_send.as_ref()) {
|
||||
// native single channel: 16 KiB app-fragments; Media is
|
||||
// drop-tolerant past the queue cap, Ctl/Move reliable.
|
||||
(true, Some(P2pConn::Single(dc))) => {
|
||||
let dc = dc.clone();
|
||||
if lane == Lane::Media && dc.buffered_amount().await > P2P_MEDIA_CAP {
|
||||
// drop (fresh beats late)
|
||||
} else {
|
||||
let fid = next_frame_id;
|
||||
next_frame_id = next_frame_id.wrapping_add(1);
|
||||
if send_fragmented(&dc, fid, &b).await.is_err() {
|
||||
let _ = relay.send_binary_lane(b, lane).await;
|
||||
let _ = down_tx.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
// web topology: video (Media) rides the media channel whole;
|
||||
// everything else (e.g. file upload) stays on the relay, like
|
||||
// the web viewer. Media is drop-tolerant past the cap.
|
||||
(true, Some(P2pConn::Web(web))) => {
|
||||
if lane == Lane::Media {
|
||||
if web.media.buffered_amount() <= P2P_MEDIA_CAP
|
||||
&& web
|
||||
.media
|
||||
.write_data_channel(&Bytes::from(b.clone()), false)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
let _ = relay.send_binary_lane(b, lane).await;
|
||||
let _ = down_tx.send(());
|
||||
}
|
||||
} else {
|
||||
let _ = relay.send_binary_lane(b, lane).await;
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
let _ = relay.send_binary_lane(b, lane).await;
|
||||
}
|
||||
},
|
||||
TransportMsg::Text(t) => match (media_on_p2p, p2p_send.as_ref()) {
|
||||
(true, Some(P2pConn::Single(dc))) if route_text_p2p(&t) => {
|
||||
let dc = dc.clone();
|
||||
if dc.send_text(t.clone()).await.is_err() {
|
||||
let _ = relay.send_text(t).await;
|
||||
let _ = down_tx.send(());
|
||||
}
|
||||
}
|
||||
// web: only input / keyframe rides ctl; other control stays
|
||||
// on the relay (the web viewer's send policy).
|
||||
(true, Some(P2pConn::Web(web))) if is_fast_path_text(&t) => {
|
||||
if web
|
||||
.ctl
|
||||
.write_data_channel(&Bytes::from(t.clone().into_bytes()), true)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
let _ = relay.send_text(t).await;
|
||||
let _ = down_tx.send(());
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
let _ = relay.send_text(t).await;
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// ---- inbound from the relay ----
|
||||
Ev::RelayIn(None) => break, // relay closed -> session ends (app recv None)
|
||||
Ev::RelayIn(Some(Err(e))) => {
|
||||
let _ = app_in_tx.send(Err(e));
|
||||
break;
|
||||
}
|
||||
Ev::RelayIn(Some(Ok(m))) => {
|
||||
let mut forward = true;
|
||||
if let TransportMsg::Text(t) = &m {
|
||||
if let Ok(v) = serde_json::from_str::<Value>(t) {
|
||||
let tt = v.get("t").and_then(|x| x.as_str()).unwrap_or("").to_string();
|
||||
if sid.is_none() {
|
||||
if let Some(s) = v.get("sid").and_then(|x| x.as_str()) {
|
||||
if !s.is_empty() {
|
||||
sid = Some(s.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
if tt.starts_with("rtc.") {
|
||||
forward = false; // rtc.* is ours, never the app's
|
||||
match role {
|
||||
P2pRole::Offerer => {
|
||||
if let Some(tx) = &rtc_in_tx {
|
||||
let _ = tx.send(v);
|
||||
}
|
||||
}
|
||||
P2pRole::Answerer => {
|
||||
if tt == "rtc.offer" && !answerer_started {
|
||||
answerer_started = true;
|
||||
let sid_s = sid.clone().unwrap_or_default();
|
||||
let is_native =
|
||||
v.get("native").and_then(|x| x.as_bool()) == Some(true);
|
||||
let sdp = v
|
||||
.get("sdp")
|
||||
.and_then(|x| x.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
let ice: Vec<RtcIceServerJson> = v
|
||||
.get("ice")
|
||||
.and_then(|x| serde_json::from_value(x.clone()).ok())
|
||||
.unwrap_or_default();
|
||||
if is_native {
|
||||
// native single-channel offerer.
|
||||
spawn_answerer(
|
||||
api.clone(),
|
||||
sid_s,
|
||||
sdp,
|
||||
ice,
|
||||
rtc_out_tx.clone(),
|
||||
done_tx.clone(),
|
||||
);
|
||||
} else if std::env::var("NETDESK_P2P_WEB").is_ok() {
|
||||
// web/3-channel offerer (a browser) — answer in
|
||||
// web topology when opted in.
|
||||
spawn_answerer_web(
|
||||
sid_s,
|
||||
sdp,
|
||||
ice,
|
||||
rtc_out_tx.clone(),
|
||||
done_tx.clone(),
|
||||
);
|
||||
} else {
|
||||
// not opted into web interop: decline so the
|
||||
// pairing stays on the relay.
|
||||
let _ = rtc_out_tx.send(
|
||||
format!(r#"{{"t":"rtc.unavailable","sid":"{sid_s}"}}"#),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if role == P2pRole::Offerer
|
||||
&& !offerer_started
|
||||
&& tt == "session.accept"
|
||||
{
|
||||
// A NATIVE host (info.p2p == "native") uses the native
|
||||
// single-channel upgrade. A non-native host (browser /
|
||||
// Python) uses the WEB 3-channel topology, but only when
|
||||
// NETDESK_P2P_WEB is opted in — so the default stays the
|
||||
// safe native-only behavior until the live cross-matrix
|
||||
// is confirmed.
|
||||
let native_host = v
|
||||
.get("info")
|
||||
.and_then(|i| i.get("p2p"))
|
||||
.and_then(|p| p.as_str())
|
||||
== Some("native");
|
||||
let web_ok = std::env::var("NETDESK_P2P_WEB").is_ok();
|
||||
if native_host {
|
||||
offerer_started = true;
|
||||
offerer_mode = Some(false);
|
||||
rtc_in_tx = Some(start_offerer_attempt(
|
||||
false,
|
||||
api.clone(),
|
||||
sid.clone().unwrap_or_default(),
|
||||
rtc_out_tx.clone(),
|
||||
done_tx.clone(),
|
||||
));
|
||||
} else if web_ok {
|
||||
offerer_started = true;
|
||||
offerer_mode = Some(true);
|
||||
rtc_in_tx = Some(start_offerer_attempt(
|
||||
true,
|
||||
api.clone(),
|
||||
sid.clone().unwrap_or_default(),
|
||||
rtc_out_tx.clone(),
|
||||
done_tx.clone(),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if forward {
|
||||
let _ = app_in_tx.send(Ok(m));
|
||||
}
|
||||
}
|
||||
|
||||
// ---- inbound from the P2P channel ----
|
||||
Ev::P2pIn(Some(m)) => {
|
||||
let _ = app_in_tx.send(Ok(m));
|
||||
}
|
||||
Ev::P2pIn(None) => {} // we hold a sender; this does not fire in practice
|
||||
|
||||
// ---- rtc.* the upgrade task wants on the relay ----
|
||||
Ev::RtcOut(Some(s)) => {
|
||||
let _ = relay.send_text(s).await;
|
||||
}
|
||||
Ev::RtcOut(None) => {}
|
||||
|
||||
// ---- UP edge: a channel was negotiated ----
|
||||
Ev::Done(Some((new_pc, conn))) => {
|
||||
let direct = rtc::selected_pair_is_direct(&new_pc).await;
|
||||
match &conn {
|
||||
P2pConn::Single(dc) => {
|
||||
// native: 16 KiB app-fragments reassembled by install_reassembler.
|
||||
install_reassembler(dc, p2p_in_tx.clone());
|
||||
}
|
||||
P2pConn::Web(web) => {
|
||||
// web: whole frames — spawn a detached read-loop per inbound
|
||||
// channel (media = video, ctl = control). Each read is one
|
||||
// complete message (is_string -> Text, else Binary).
|
||||
for raw in [web.media.clone(), web.ctl.clone()] {
|
||||
let tx = p2p_in_tx.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut buf = vec![0u8; 262144];
|
||||
loop {
|
||||
match raw.read_data_channel(&mut buf).await {
|
||||
Ok((0, _)) => break,
|
||||
Ok((n, is_string)) => {
|
||||
let msg = if is_string {
|
||||
match String::from_utf8(buf[..n].to_vec()) {
|
||||
Ok(s) => TransportMsg::Text(s),
|
||||
Err(_) => continue,
|
||||
}
|
||||
} else {
|
||||
TransportMsg::Binary(buf[..n].to_vec())
|
||||
};
|
||||
let _ = tx.send(msg);
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
let dtx = down_tx.clone();
|
||||
new_pc.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
|
||||
let dtx = dtx.clone();
|
||||
Box::pin(async move {
|
||||
if matches!(
|
||||
s,
|
||||
RTCPeerConnectionState::Failed
|
||||
| RTCPeerConnectionState::Disconnected
|
||||
| RTCPeerConnectionState::Closed
|
||||
) {
|
||||
let _ = dtx.send(());
|
||||
}
|
||||
})
|
||||
}));
|
||||
p2p_send = Some(conn);
|
||||
_pc = Some(new_pc);
|
||||
// Only move media off the relay for a DIRECT pair; a TURN-relayed
|
||||
// hop is no better than our own relay.
|
||||
media_on_p2p = direct;
|
||||
tracing::info!(direct, "P2P DataChannel open");
|
||||
if role == P2pRole::Offerer {
|
||||
let _ = app_in_tx.send(Ok(TransportMsg::Text(format!(
|
||||
r#"{{"t":"p2p.state","active":{media_on_p2p},"direct":{direct}}}"#
|
||||
))));
|
||||
}
|
||||
}
|
||||
Ev::Done(None) => {}
|
||||
|
||||
// ---- periodic health probe: abandon a pathologically slow direct
|
||||
// pair for the relay (the "pick the faster path" half of hybrid). ----
|
||||
Ev::Health => {
|
||||
if media_on_p2p && !health_down {
|
||||
if let Some(pc) = &_pc {
|
||||
let rtt = p2p_pair_rtt_ms(pc).await;
|
||||
if health.on_rtt(rtt) {
|
||||
tracing::warn!(
|
||||
rtt_ms = rtt,
|
||||
"P2P direct pair too slow — downgrading to the relay for the session"
|
||||
);
|
||||
health_down = true;
|
||||
let _ = down_tx.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- DOWN edge: pc failed/closed -> fall back to the warm relay ----
|
||||
Ev::Down => {
|
||||
let was_up = media_on_p2p || p2p_send.is_some();
|
||||
if was_up {
|
||||
tracing::info!("P2P dropped — continuing over the relay");
|
||||
}
|
||||
media_on_p2p = false;
|
||||
p2p_send = None;
|
||||
_pc = None;
|
||||
health = P2pHealth::default(); // reset the RTT streak for the next pair
|
||||
if was_up && role == P2pRole::Offerer {
|
||||
let _ = app_in_tx.send(Ok(TransportMsg::Text(
|
||||
r#"{"t":"p2p.state","active":false,"direct":false}"#.to_string(),
|
||||
)));
|
||||
}
|
||||
// Re-upgrade hysteresis: allow a fresh attempt. The answerer
|
||||
// just re-answers the viewer's next rtc.offer; the offerer
|
||||
// schedules one retry after a backoff.
|
||||
if was_up {
|
||||
answerer_started = false;
|
||||
offerer_started = false;
|
||||
rtc_in_tx = None;
|
||||
// Suppress re-upgrade after a health downgrade: the direct
|
||||
// path proved too slow, so stay on the relay (no flapping).
|
||||
if role == P2pRole::Offerer && offerer_mode.is_some() && !health_down {
|
||||
let reup_tx = reup_tx.clone();
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(reup_ms)).await;
|
||||
let _ = reup_tx.send(());
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- re-upgrade timer fired: retry the offerer upgrade once ----
|
||||
Ev::ReUpgrade => {
|
||||
if role == P2pRole::Offerer
|
||||
&& !offerer_started
|
||||
&& p2p_send.is_none()
|
||||
&& sid.is_some()
|
||||
{
|
||||
if let Some(web) = offerer_mode {
|
||||
offerer_started = true;
|
||||
rtc_in_tx = Some(start_offerer_attempt(
|
||||
web,
|
||||
api.clone(),
|
||||
sid.clone().unwrap_or_default(),
|
||||
rtc_out_tx.clone(),
|
||||
done_tx.clone(),
|
||||
));
|
||||
tracing::info!("P2P re-upgrade attempt");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod health_tests {
|
||||
use super::{P2pHealth, P2P_RTT_CONFIRM};
|
||||
|
||||
#[test]
|
||||
fn good_rtt_never_downgrades() {
|
||||
let mut h = P2pHealth::default();
|
||||
for _ in 0..50 {
|
||||
assert!(!h.on_rtt(30), "a healthy direct pair must stay on P2P");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_rtt_is_ignored() {
|
||||
let mut h = P2pHealth::default();
|
||||
for _ in 0..50 {
|
||||
assert!(!h.on_rtt(0), "0 = not-yet-measured, never a downgrade");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sustained_bad_rtt_downgrades_after_confirm() {
|
||||
let mut h = P2pHealth::default();
|
||||
// Must take the full confirm streak — not a single spike.
|
||||
for i in 1..P2P_RTT_CONFIRM {
|
||||
assert!(!h.on_rtt(800), "should not trip on sample {i} (< confirm)");
|
||||
}
|
||||
assert!(h.on_rtt(800), "should downgrade once the streak reaches confirm");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_good_sample_resets_the_streak() {
|
||||
let mut h = P2pHealth::default();
|
||||
for _ in 0..(P2P_RTT_CONFIRM - 1) {
|
||||
assert!(!h.on_rtt(900));
|
||||
}
|
||||
assert!(!h.on_rtt(40), "a healthy sample clears the bad streak");
|
||||
// Now it must take another full streak to trip.
|
||||
for i in 1..P2P_RTT_CONFIRM {
|
||||
assert!(!h.on_rtt(900), "streak restarted; sample {i} must not trip");
|
||||
}
|
||||
assert!(h.on_rtt(900));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user