From 7ccb32bb64a958057c1e0a3e92c8bcb649ffcbb9 Mon Sep 17 00:00:00 2001 From: Patrick Ecord Date: Sun, 21 Jun 2026 03:22:09 -0500 Subject: [PATCH] =?UTF-8?q?core:=20prefer-direct=20=E2=80=94=20suppress=20?= =?UTF-8?q?the=20relay=20once=20P2P=20covers=20the=20room?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine sent clipboard over every transport at once, so the server always saw the payload even with RTC/LAN up. Now the RTC and LAN transports report their directly-reachable peers (data-channel open / TCP connected) into a shared set, and the RTC transport records who's present from presence chirps. send() skips the SSE relay when every present peer is directly reachable — so once P2P is established the server only ever carries signaling/presence, never content. Falls back to the relay whenever a present peer isn't directly reachable, or before the direct link is up. Also drop empty-text clipboards in the sink (were surfacing as blank "unknown" feed rows). Proven by examples/prefer_direct.rs: two engines link directly on the live server; a send reaches the peer but never appears on the server bus (only presence does). Co-Authored-By: Claude Opus 4.8 --- core/examples/prefer_direct.rs | 49 ++++++++++++++++++++++++++++++++++ core/src/lan.rs | 8 ++++-- core/src/lib.rs | 46 +++++++++++++++++++++++++++---- core/src/rtc.rs | 21 ++++++++++++--- 4 files changed, 114 insertions(+), 10 deletions(-) create mode 100644 core/examples/prefer_direct.rs diff --git a/core/examples/prefer_direct.rs b/core/examples/prefer_direct.rs new file mode 100644 index 0000000..a5d44d9 --- /dev/null +++ b/core/examples/prefer_direct.rs @@ -0,0 +1,49 @@ +//! Proves the relay carries content only until a direct path exists. A and B +//! join a room on the LIVE server and link directly (LAN/RTC). After that, A's +//! send must reach B but must NOT appear on the server bus. +//! cargo run --example prefer_direct -- +//! (run an external SSE listener on the same room to confirm suppression) + +use std::sync::mpsc; +use std::time::Duration; + +use tethercore::{Engine, Message, MessageHandler}; + +struct Collector(mpsc::Sender); +impl MessageHandler for Collector { + fn on_message(&self, m: Message) { + eprintln!(" B.on_message: kind={:?} text={:?} from={:?}", m.kind, m.text, m.from); + let _ = self.0.send(m.text); + } + fn on_status(&self, _c: bool) {} +} + +fn main() { + let server = "https://tether.pecord.io".to_string(); + let room = std::env::args().nth(1).unwrap_or_else(|| "prefer-direct-demo".into()); + + let (tx, rx) = mpsc::channel(); + let a = Engine::new(server.clone(), room.clone(), "dev-a".into(), "macos".into(), "A".into(), "".into()); + let b = Engine::new(server, room, "dev-b".into(), "ios".into(), "B".into(), "".into()); + a.start(Box::new(Collector(mpsc::channel().0))); + b.start(Box::new(Collector(tx))); + + // Let presence + the direct (LAN/RTC) link establish. + eprintln!("waiting ~12s for presence + direct link…"); + std::thread::sleep(Duration::from_secs(12)); + + let secret = "DIRECT-ONLY-SECRET"; + eprintln!("A sends — should go P2P, NOT via the relay"); + a.send(secret.to_string()); + + let mut got = false; + while let Ok(t) = rx.recv_timeout(Duration::from_secs(4)) { + if t == secret { got = true; break; } + } + println!( + "{}", + if got { "✅ B received the secret over the direct path" } else { "❌ B never got the secret" } + ); + a.stop(); + b.stop(); +} diff --git a/core/src/lan.rs b/core/src/lan.rs index afc44b9..63a19aa 100644 --- a/core/src/lan.rs +++ b/core/src/lan.rs @@ -23,7 +23,7 @@ use tokio::net::{TcpListener, TcpStream}; use tokio::runtime::Handle; use tokio::sync::Mutex; -use crate::{Config, Discovered, Message, Peer, Sink, Status, Transport}; +use crate::{Config, DirectSet, Discovered, Message, Peer, Sink, Status, Transport}; const SERVICE: &str = "_tether._tcp.local."; @@ -32,15 +32,17 @@ pub(crate) struct LanTransport { peers: Arc>>, // peer `from` → its write half sink: StdMutex>, discovered: Discovered, // nearby devices (any room), for the engine's nearby() + direct: DirectSet, // LAN-connected peers (a direct path; suppresses relay) } impl LanTransport { - pub(crate) fn new(cfg: Config, discovered: Discovered) -> Self { + pub(crate) fn new(cfg: Config, discovered: Discovered, direct: DirectSet) -> Self { Self { cfg, peers: Arc::new(Mutex::new(HashMap::new())), sink: StdMutex::new(None), discovered, + direct, } } @@ -240,6 +242,7 @@ impl LanTransport { } map.insert(peer_from.clone(), wr); } + self.direct.lock().unwrap().insert(peer_from.clone()); // direct path up eprintln!("tethercore[lan]: peer {peer_from} connected"); while let Ok(Some(line)) = lines.next_line().await { @@ -253,6 +256,7 @@ impl LanTransport { } self.peers.lock().await.remove(&peer_from); + self.direct.lock().unwrap().remove(&peer_from); // relay needed again eprintln!("tethercore[lan]: peer {peer_from} disconnected"); } } diff --git a/core/src/lib.rs b/core/src/lib.rs index eb35b81..283f353 100644 --- a/core/src/lib.rs +++ b/core/src/lib.rs @@ -13,7 +13,7 @@ //! across all transports and de-duplicates inbound so the same clipboard //! arriving over two channels surfaces once. -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -88,6 +88,15 @@ pub struct Receipt { type Receipts = Arc>>; +/// Peers reachable over a direct channel right now (RTC data channel or LAN TCP), +/// by `from`. Written by the RTC/LAN transports, read by `send` to decide whether +/// the relay is still needed. +type DirectSet = Arc>>; +/// Peers currently in the room, by `from`, last-seen via presence chirps. Lets +/// `send` know the full audience so it only suppresses the relay when *every* +/// present peer is directly reachable. +type PresentSet = Arc>>; + /// Canonical room id for a shared identity: sha256(account)[..4] as 8 hex chars. /// Defined once here and exported so every client (Swift/Kotlin/agent) derives /// the SAME room — that's what puts all of one account's devices together across @@ -136,6 +145,8 @@ pub struct Engine { dedup: Arc>, discovered: Discovered, receipts: Receipts, + direct: DirectSet, + present: PresentSet, } #[uniffi::export] @@ -162,13 +173,15 @@ impl Engine { .expect("tokio runtime"); let cfg = Config { server, room, from, source, name, account }; let discovered: Discovered = Arc::new(Mutex::new(HashMap::new())); + let direct: DirectSet = Arc::new(Mutex::new(HashSet::new())); + let present: PresentSet = Arc::new(Mutex::new(HashMap::new())); let transports: Vec> = vec![ Arc::new(SseTransport { cfg: cfg.clone(), http: reqwest::Client::new(), }), - Arc::new(rtc::RtcTransport::new(cfg.clone())), - Arc::new(lan::LanTransport::new(cfg.clone(), discovered.clone())), + Arc::new(rtc::RtcTransport::new(cfg.clone(), direct.clone(), present.clone())), + Arc::new(lan::LanTransport::new(cfg.clone(), discovered.clone(), direct.clone())), ]; Arc::new(Self { cfg, @@ -178,6 +191,8 @@ impl Engine { dedup: Arc::new(Mutex::new(Dedup::default())), discovered, receipts: Arc::new(Mutex::new(Vec::new())), + direct, + present, }) } @@ -232,7 +247,10 @@ impl Engine { } return; } - // Clipboard: de-dup, ack the sender, then deliver. + // Clipboard: ignore empties, de-dup, ack the sender, then deliver. + if m.text.is_empty() { + return; + } if dedup.lock().unwrap().seen(&m) { return; // already delivered via another transport } @@ -267,7 +285,10 @@ impl Engine { } } - /// Publish clipboard text to the room across all transports. + /// Publish clipboard text to the room. Always sent over the direct + /// transports (RTC/LAN); sent over the SSE relay ONLY when a direct path + /// doesn't already reach every present peer — so once P2P is up the server + /// never sees the payload (it stays just signaling + fallback). pub fn send(&self, text: String) { let msg = Message { kind: "clipboard".into(), @@ -279,12 +300,27 @@ impl Engine { room: self.cfg.room.clone(), ts: 0, // server stamps authoritative ts }; + let covered = self.directly_covered(); let rt = self.rt.handle(); for t in &self.transports { + if covered && t.name() == "sse" { + continue; // direct channels reach everyone → keep content off the relay + } t.publish(rt, msg.clone()); } } + /// True when every peer currently present in the room is reachable over a + /// direct channel — i.e. the relay isn't needed for delivery. False if we + /// don't yet know who's present (be safe, relay). + fn directly_covered(&self) -> bool { + let direct = self.direct.lock().unwrap(); + let now = Instant::now(); + let mut present = self.present.lock().unwrap(); + present.retain(|_, t| now.duration_since(*t) < Duration::from_secs(20)); + !present.is_empty() && present.keys().all(|p| direct.contains(p)) + } + /// Stop all transports at their next checkpoint. pub fn stop(&self) { self.running.store(false, Ordering::SeqCst); diff --git a/core/src/rtc.rs b/core/src/rtc.rs index a669649..c2d06ca 100644 --- a/core/src/rtc.rs +++ b/core/src/rtc.rs @@ -13,7 +13,7 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex as StdMutex}; -use std::time::Duration; +use std::time::{Duration, Instant}; use futures_util::StreamExt; use serde_json::{json, Value}; @@ -31,7 +31,7 @@ use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState; use webrtc::peer_connection::sdp::session_description::RTCSessionDescription; use webrtc::peer_connection::RTCPeerConnection; -use crate::{Config, Message, Sink, Status, Transport}; +use crate::{Config, DirectSet, Message, PresentSet, Sink, Status, Transport}; struct Peer { pc: Arc, @@ -46,6 +46,8 @@ pub(crate) struct RtcTransport { peers: Arc>>>, sink: StdMutex>, ice: StdMutex>, // refreshed from /api/turn-cred at start + direct: DirectSet, // peers with an open data channel + present: PresentSet, // peers seen via presence chirps } fn default_ice() -> Vec { @@ -56,7 +58,7 @@ fn default_ice() -> Vec { } impl RtcTransport { - pub(crate) fn new(cfg: Config) -> Self { + pub(crate) fn new(cfg: Config, direct: DirectSet, present: PresentSet) -> Self { // Data-channel-only: a bare media engine, no codecs/interceptors needed. let api = APIBuilder::new() .with_media_engine(MediaEngine::default()) @@ -68,6 +70,8 @@ impl RtcTransport { peers: Arc::new(Mutex::new(HashMap::new())), sink: StdMutex::new(None), ice: StdMutex::new(default_ice()), + direct, + present, } } @@ -188,9 +192,11 @@ impl RtcTransport { // Drop the peer when the connection dies. { let peers = self.peers.clone(); + let direct = self.direct.clone(); let remote = remote.to_owned(); pc.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| { let peers = peers.clone(); + let direct = direct.clone(); let remote = remote.clone(); Box::pin(async move { if matches!( @@ -200,6 +206,7 @@ impl RtcTransport { | RTCPeerConnectionState::Disconnected ) { peers.lock().await.remove(&remote); + direct.lock().unwrap().remove(&remote); // relay needed again } }) })); @@ -243,9 +250,12 @@ impl RtcTransport { async fn wire_channel(self: &Arc, remote: &str, dc: Arc, peer: &Arc) { { let remote_log = remote.to_owned(); + let direct = self.direct.clone(); dc.on_open(Box::new(move || { let remote_log = remote_log.clone(); + let direct = direct.clone(); Box::pin(async move { + direct.lock().unwrap().insert(remote_log.clone()); // now P2P-reachable eprintln!("tethercore[rtc]: data channel open ↔ {remote_log}"); }) })); @@ -405,6 +415,11 @@ impl RtcTransport { } match v["type"].as_str().unwrap_or("") { "presence" => { + // Track who's in the room so send() knows the full audience. + self.present + .lock() + .unwrap() + .insert(from.to_string(), Instant::now()); // Deterministic initiator: the smaller id offers. if self.cfg.from.as_str() < from { self.initiate_offer(from).await;