Like T3 Code's routes: the host reports the private addresses it listens on, marking those on Tailscale's interface, in Welcome and again when they change. The client saves them with the connection, tries every route fastest first, skips one that answers with another device's key, and while on a slower route keeps trying the faster ones every minute and moves as soon as one works. Pings notice a route that went away. Pairing codes carry every address, and pairing the same device again adds a route instead of a second environment. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Showing 9 changed files with 771 additions and 96 deletions.
| @@ -1,26 +1,48 @@ | ||
| 1 | 1 | //! The client side: a connection to another device's app on its own thread, |
| 2 | 2 | //! reconnecting when it drops. Messages from the host reach the workspace as |
| 3 | 3 | //! [`ClientEvent`]s. |
| 4 | + //! | |
| 5 | + //! A host can have several routes. The client connects over the first one, | |
| 6 | + //! fastest first, that answers with the right key, and while it's on a | |
| 7 | + //! slower one it keeps trying the faster ones and moves as soon as one works. | |
| 4 | 8 | |
| 5 | 9 | use std::{ |
| 6 | 10 | io::ErrorKind, |
| 7 | 11 | net::{TcpStream, ToSocketAddrs}, |
| 8 | 12 | sync::{ |
| 9 | - Arc, | |
| 13 | + Arc, Mutex, | |
| 10 | 14 | atomic::{AtomicBool, Ordering}, |
| 11 | 15 | mpsc, |
| 12 | 16 | }, |
| 13 | 17 | thread, |
| 14 | - time::Duration, | |
| 18 | + time::{Duration, Instant}, | |
| 15 | 19 | }; |
| 16 | 20 | |
| 17 | 21 | use tungstenite::{Message, WebSocket}; |
| 18 | 22 | |
| 19 | - use super::{ClientMessage, PROTOCOL, ServerMessage, with_port}; | |
| 23 | + use super::{ClientMessage, PROTOCOL, Route, RouteKind, ServerMessage}; | |
| 20 | 24 | |
| 25 | + /// How long to wait for a route that goes over the internet or a VPN. | |
| 21 | 26 | const CONNECT_TIMEOUT: Duration = Duration::from_secs(6); |
| 27 | + /// A LAN address answers quickly when it's there; away from home it isn't. | |
| 28 | + const LAN_TIMEOUT: Duration = Duration::from_secs(2); | |
| 22 | 29 | const POLL: Duration = Duration::from_millis(40); |
| 23 | 30 | |
| 31 | + /// How the connection checks on itself. | |
| 32 | + #[derive(Clone, Copy)] | |
| 33 | + struct Timing { | |
| 34 | + /// How often to ping the host, which answers even when idle. | |
| 35 | + ping: Duration, | |
| 36 | + /// Hearing nothing this long means the route is gone, as when this | |
| 37 | + /// device left the network it went over. | |
| 38 | + silence: Duration, | |
| 39 | + /// How often to try the faster routes while on a slower one. | |
| 40 | + probe: Duration, | |
| 41 | + } | |
| 42 | + | |
| 43 | + const TIMING: Timing = | |
| 44 | + Timing { ping: Duration::from_secs(15), silence: Duration::from_secs(45), probe: Duration::from_secs(60) }; | |
| 45 | + | |
| 24 | 46 | #[derive(Clone, Debug, PartialEq)] |
| 25 | 47 | pub enum ConnectionStatus { |
| 26 | 48 | Connecting, |
| @@ -33,12 +55,15 @@ pub enum ConnectionStatus { | ||
| 33 | 55 | |
| 34 | 56 | pub enum ClientEvent { |
| 35 | 57 | Status(ConnectionStatus), |
| 58 | + /// The route the connection goes over, sent before each `Connected`. | |
| 59 | + Route(Route), | |
| 36 | 60 | Message(ServerMessage), |
| 37 | 61 | } |
| 38 | 62 | |
| 39 | 63 | pub struct RemoteClient { |
| 40 | 64 | outbox: mpsc::Sender<String>, |
| 41 | 65 | stop: Arc<AtomicBool>, |
| 66 | + routes: Arc<Mutex<Vec<Route>>>, | |
| 42 | 67 | pub events: smol::channel::Receiver<ClientEvent>, |
| 43 | 68 | } |
| 44 | 69 | |
| @@ -59,27 +84,32 @@ impl RemoteClient { | ||
| 59 | 84 | ClientSender(self.outbox.clone()) |
| 60 | 85 | } |
| 61 | 86 | |
| 62 | - /// Connects to `address` (`host` or `host:port`) and keeps reconnecting | |
| 63 | - /// until dropped. | |
| 64 | - pub fn connect(address: &str, key: String, name: String) -> Self { | |
| 65 | - let address = with_port(address); | |
| 87 | + /// Connects over `routes`, tried in order, and keeps reconnecting until | |
| 88 | + /// dropped. | |
| 89 | + pub fn connect(routes: Vec<Route>, key: String, name: String) -> Self { | |
| 90 | + Self::connect_timed(routes, key, name, TIMING) | |
| 91 | + } | |
| 92 | + | |
| 93 | + fn connect_timed(routes: Vec<Route>, key: String, name: String, timing: Timing) -> Self { | |
| 66 | 94 | let stop = Arc::new(AtomicBool::new(false)); |
| 95 | + let routes = Arc::new(Mutex::new(routes)); | |
| 67 | 96 | let (outbox, inbox) = mpsc::channel::<String>(); |
| 68 | 97 | let (events_tx, events) = smol::channel::unbounded(); |
| 69 | 98 | { |
| 70 | 99 | let stop = stop.clone(); |
| 100 | + let routes = routes.clone(); | |
| 71 | 101 | let _ = thread::Builder::new().name("remote-client".into()).spawn(move || { |
| 102 | + let hello = Hello { key, name, routes }; | |
| 72 | 103 | let mut backoff = Duration::from_secs(2); |
| 73 | 104 | while !stop.load(Ordering::Relaxed) { |
| 74 | 105 | let _ = events_tx.send_blocking(ClientEvent::Status(ConnectionStatus::Connecting)); |
| 75 | - match open(&address, &key, &name) { | |
| 76 | - Ok((mut socket, host)) => { | |
| 106 | + match open_any(&hello.routes(), &hello) { | |
| 107 | + Ok(link) => { | |
| 77 | 108 | backoff = Duration::from_secs(2); |
| 78 | 109 | // Whatever was asked while offline is stale now. |
| 79 | 110 | while inbox.try_recv().is_ok() {} |
| 80 | - let _ = events_tx.send_blocking(ClientEvent::Status(ConnectionStatus::Connected(host))); | |
| 81 | - let reason = run(&mut socket, &inbox, &events_tx, &stop); | |
| 82 | - let _ = socket.close(None); | |
| 111 | + announce(&events_tx, &link); | |
| 112 | + let reason = run(link, &hello, &inbox, &events_tx, &stop, timing); | |
| 83 | 113 | if stop.load(Ordering::Relaxed) { |
| 84 | 114 | break; |
| 85 | 115 | } |
| @@ -103,7 +133,13 @@ impl RemoteClient { | ||
| 103 | 133 | } |
| 104 | 134 | }); |
| 105 | 135 | } |
| 106 | - Self { outbox, stop, events } | |
| 136 | + Self { outbox, stop, routes, events } | |
| 137 | + } | |
| 138 | + | |
| 139 | + /// Changes the routes without dropping the connection; the next try, | |
| 140 | + /// and the next look for a faster route, use them. | |
| 141 | + pub fn set_routes(&self, routes: Vec<Route>) { | |
| 142 | + *self.routes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) = routes; | |
| 107 | 143 | } |
| 108 | 144 | |
| 109 | 145 | pub fn send(&self, message: &ClientMessage) { |
| @@ -124,25 +160,81 @@ enum Failure { | ||
| 124 | 160 | Other(String), |
| 125 | 161 | } |
| 126 | 162 | |
| 163 | + /// What every try says, and where it may go. | |
| 164 | + #[derive(Clone)] | |
| 165 | + struct Hello { | |
| 166 | + key: String, | |
| 167 | + name: String, | |
| 168 | + routes: Arc<Mutex<Vec<Route>>>, | |
| 169 | + } | |
| 170 | + | |
| 171 | + impl Hello { | |
| 172 | + fn routes(&self) -> Vec<Route> { | |
| 173 | + self.routes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).clone() | |
| 174 | + } | |
| 175 | + } | |
| 176 | + | |
| 177 | + /// A connection that said hello and was welcomed. | |
| 178 | + struct Link { | |
| 179 | + socket: WebSocket<TcpStream>, | |
| 180 | + host: String, | |
| 181 | + route: Route, | |
| 182 | + /// Where the host says it listens. | |
| 183 | + routes: Vec<Route>, | |
| 184 | + } | |
| 185 | + | |
| 186 | + /// Tells the workspace about a new connection. | |
| 187 | + fn announce(events: &smol::channel::Sender<ClientEvent>, link: &Link) { | |
| 188 | + let _ = events.send_blocking(ClientEvent::Route(link.route.clone())); | |
| 189 | + let _ = events.send_blocking(ClientEvent::Message(ServerMessage::Routes { routes: link.routes.clone() })); | |
| 190 | + let _ = events.send_blocking(ClientEvent::Status(ConnectionStatus::Connected(link.host.clone()))); | |
| 191 | + } | |
| 192 | + | |
| 193 | + /// Tries each route in turn. It's denied only when every route that | |
| 194 | + /// answered denied it: an address on another network can be another | |
| 195 | + /// device's. | |
| 196 | + fn open_any(routes: &[Route], hello: &Hello) -> Result<Link, Failure> { | |
| 197 | + let mut denied = None; | |
| 198 | + let mut failures = Vec::new(); | |
| 199 | + for route in routes { | |
| 200 | + match open(route, hello) { | |
| 201 | + Ok(link) => return Ok(link), | |
| 202 | + Err(Failure::Denied(reason)) => denied = denied.or(Some(reason)), | |
| 203 | + Err(Failure::Other(reason)) if routes.len() == 1 => failures.push(reason), | |
| 204 | + Err(Failure::Other(reason)) => failures.push(format!("{}: {reason}", route.kind.label())), | |
| 205 | + } | |
| 206 | + } | |
| 207 | + match denied { | |
| 208 | + Some(reason) if failures.is_empty() => Err(Failure::Denied(reason)), | |
| 209 | + _ if routes.is_empty() => Err(Failure::Other("no address to connect to".into())), | |
| 210 | + _ => Err(Failure::Other(failures.join("; "))), | |
| 211 | + } | |
| 212 | + } | |
| 213 | + | |
| 127 | 214 | /// Connects, says hello, and waits for the welcome. |
| 128 | - fn open(address: &str, key: &str, name: &str) -> Result<(WebSocket<TcpStream>, String), Failure> { | |
| 215 | + fn open(route: &Route, hello: &Hello) -> Result<Link, Failure> { | |
| 129 | 216 | let other = |err: &dyn std::fmt::Display| Failure::Other(err.to_string()); |
| 217 | + let address = &route.address; | |
| 130 | 218 | let target = address.to_socket_addrs().map_err(|err| other(&err))?.next().ok_or_else(|| other(&"no such host"))?; |
| 131 | - let stream = TcpStream::connect_timeout(&target, CONNECT_TIMEOUT).map_err(|err| other(&err))?; | |
| 219 | + let timeout = match route.kind { | |
| 220 | + RouteKind::Loopback | RouteKind::Lan => LAN_TIMEOUT, | |
| 221 | + _ => CONNECT_TIMEOUT, | |
| 222 | + }; | |
| 223 | + let stream = TcpStream::connect_timeout(&target, timeout).map_err(|err| other(&err))?; | |
| 132 | 224 | let _ = stream.set_nodelay(true); |
| 133 | 225 | stream.set_read_timeout(Some(Duration::from_secs(15))).map_err(|err| other(&err))?; |
| 134 | 226 | let (mut socket, _) = tungstenite::client(format!("ws://{address}/"), stream).map_err(|err| other(&err))?; |
| 135 | - let hello = ClientMessage::Hello { key: key.to_string(), name: name.to_string(), protocol: PROTOCOL }; | |
| 227 | + let message = ClientMessage::Hello { key: hello.key.clone(), name: hello.name.clone(), protocol: PROTOCOL }; | |
| 136 | 228 | socket |
| 137 | - .send(Message::text(serde_json::to_string(&hello).unwrap_or_default())) | |
| 229 | + .send(Message::text(serde_json::to_string(&message).unwrap_or_default())) | |
| 138 | 230 | .map_err(|err| other(&err))?; |
| 139 | 231 | loop { |
| 140 | 232 | match socket.read().map_err(|err| other(&err))? { |
| 141 | 233 | Message::Text(text) => { |
| 142 | 234 | return match serde_json::from_str::<ServerMessage>(text.as_str()) { |
| 143 | - Ok(ServerMessage::Welcome { host, .. }) => { | |
| 235 | + Ok(ServerMessage::Welcome { host, routes, .. }) => { | |
| 144 | 236 | let _ = socket.get_mut().set_read_timeout(Some(POLL)); |
| 145 | - Ok((socket, host)) | |
| 237 | + Ok(Link { socket, host, route: route.clone(), routes }) | |
| 146 | 238 | } |
| 147 | 239 | Ok(ServerMessage::Denied { reason }) => Err(Failure::Denied(reason)), |
| 148 | 240 | _ => Err(other(&"the host answered something unexpected")), |
| @@ -154,33 +246,140 @@ fn open(address: &str, key: &str, name: &str) -> Result<(WebSocket<TcpStream>, S | ||
| 154 | 246 | } |
| 155 | 247 | } |
| 156 | 248 | |
| 157 | - /// Trades messages until the connection drops; returns why it did. | |
| 249 | + /// Trades messages until the connection drops; returns why it did. Moves | |
| 250 | + /// to a faster route when one answers. | |
| 158 | 251 | fn run( |
| 159 | - socket: &mut WebSocket<TcpStream>, | |
| 252 | + mut link: Link, | |
| 253 | + hello: &Hello, | |
| 160 | 254 | inbox: &mpsc::Receiver<String>, |
| 161 | 255 | events: &smol::channel::Sender<ClientEvent>, |
| 162 | 256 | stop: &AtomicBool, |
| 257 | + timing: Timing, | |
| 163 | 258 | ) -> String { |
| 164 | - while !stop.load(Ordering::Relaxed) { | |
| 165 | - while let Ok(text) = inbox.try_recv() { | |
| 166 | - if let Err(err) = socket.send(Message::text(text)) { | |
| 167 | - return err.to_string(); | |
| 259 | + let (mut pinged, mut heard, mut probed) = (Instant::now(), Instant::now(), Instant::now()); | |
| 260 | + let mut probe: Option<mpsc::Receiver<Link>> = None; | |
| 261 | + let reason = 'connection: { | |
| 262 | + while !stop.load(Ordering::Relaxed) { | |
| 263 | + while let Ok(text) = inbox.try_recv() { | |
| 264 | + if let Err(err) = link.socket.send(Message::text(text)) { | |
| 265 | + break 'connection err.to_string(); | |
| 266 | + } | |
| 168 | 267 | } |
| 169 | - } | |
| 170 | - match socket.read() { | |
| 171 | - Ok(Message::Text(text)) => match serde_json::from_str::<ServerMessage>(text.as_str()) { | |
| 268 | + if pinged.elapsed() >= timing.ping { | |
| 269 | + pinged = Instant::now(); | |
| 270 | + if let Err(err) = link.socket.send(Message::Ping(Default::default())) { | |
| 271 | + break 'connection err.to_string(); | |
| 272 | + } | |
| 273 | + } | |
| 274 | + if heard.elapsed() > timing.silence { | |
| 275 | + break 'connection "The host stopped answering.".into(); | |
| 276 | + } | |
| 277 | + if probe.is_none() && probed.elapsed() >= timing.probe { | |
| 278 | + probed = Instant::now(); | |
| 279 | + let faster: Vec<Route> = | |
| 280 | + hello.routes().into_iter().take_while(|route| route.address != link.route.address).collect(); | |
| 281 | + if !faster.is_empty() { | |
| 282 | + let (found, receiver) = mpsc::channel(); | |
| 283 | + let hello = hello.clone(); | |
| 284 | + let _ = thread::Builder::new().name("remote-probe".into()).spawn(move || { | |
| 285 | + if let Some(link) = faster.iter().find_map(|route| open(route, &hello).ok()) { | |
| 286 | + let _ = found.send(link); | |
| 287 | + } | |
| 288 | + }); | |
| 289 | + probe = Some(receiver); | |
| 290 | + } | |
| 291 | + } | |
| 292 | + if let Some(receiver) = &probe { | |
| 293 | + match receiver.try_recv() { | |
| 294 | + Ok(faster) => { | |
| 295 | + let _ = link.socket.close(None); | |
| 296 | + let _ = link.socket.flush(); | |
| 297 | + link = faster; | |
| 298 | + heard = Instant::now(); | |
| 299 | + probe = None; | |
| 300 | + announce(events, &link); | |
| 301 | + } | |
| 302 | + Err(mpsc::TryRecvError::Disconnected) => probe = None, | |
| 303 | + Err(mpsc::TryRecvError::Empty) => {} | |
| 304 | + } | |
| 305 | + } | |
| 306 | + match link.socket.read() { | |
| 172 | 307 | Ok(message) => { |
| 173 | - if events.send_blocking(ClientEvent::Message(message)).is_err() { | |
| 174 | - return "closed".into(); | |
| 308 | + heard = Instant::now(); | |
| 309 | + if let Message::Text(text) = message { | |
| 310 | + match serde_json::from_str::<ServerMessage>(text.as_str()) { | |
| 311 | + Ok(message) => { | |
| 312 | + if events.send_blocking(ClientEvent::Message(message)).is_err() { | |
| 313 | + break 'connection "closed".into(); | |
| 314 | + } | |
| 315 | + } | |
| 316 | + Err(err) => eprintln!("vectrum-code: unreadable message from host: {err}"), | |
| 317 | + } | |
| 318 | + } else if let Message::Close(_) = message { | |
| 319 | + break 'connection "The host closed the connection.".into(); | |
| 175 | 320 | } |
| 176 | 321 | } |
| 177 | - Err(err) => eprintln!("vectrum-code: unreadable message from host: {err}"), | |
| 178 | - }, | |
| 179 | - Ok(Message::Close(_)) => return "The host closed the connection.".into(), | |
| 180 | - Ok(_) => {} | |
| 181 | - Err(tungstenite::Error::Io(err)) if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {} | |
| 182 | - Err(err) => return err.to_string(), | |
| 322 | + Err(tungstenite::Error::Io(err)) if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {} | |
| 323 | + Err(err) => break 'connection err.to_string(), | |
| 324 | + } | |
| 183 | 325 | } |
| 326 | + "closed".into() | |
| 327 | + }; | |
| 328 | + let _ = link.socket.close(None); | |
| 329 | + reason | |
| 330 | + } | |
| 331 | + | |
| 332 | + #[cfg(test)] | |
| 333 | + mod tests { | |
| 334 | + use super::*; | |
| 335 | + use crate::remote::{new_key, server::RemoteServer}; | |
| 336 | + | |
| 337 | + fn next(client: &RemoteClient) -> Option<ClientEvent> { | |
| 338 | + smol::block_on(async { | |
| 339 | + smol::future::or(async { client.events.recv().await.ok() }, async { | |
| 340 | + smol::Timer::after(Duration::from_secs(10)).await; | |
| 341 | + None | |
| 342 | + }) | |
| 343 | + .await | |
| 344 | + }) | |
| 345 | + } | |
| 346 | + | |
| 347 | + /// The route the next connection goes over. | |
| 348 | + fn connected_route(client: &RemoteClient) -> String { | |
| 349 | + let mut route = None; | |
| 350 | + loop { | |
| 351 | + match next(client) { | |
| 352 | + Some(ClientEvent::Route(found)) => route = Some(found.address), | |
| 353 | + Some(ClientEvent::Status(ConnectionStatus::Connected(_))) => return route.expect("no route before connecting"), | |
| 354 | + Some(ClientEvent::Status(ConnectionStatus::Denied(reason))) => panic!("denied: {reason}"), | |
| 355 | + Some(_) => {} | |
| 356 | + None => panic!("didn't connect"), | |
| 357 | + } | |
| 358 | + } | |
| 359 | + } | |
| 360 | + | |
| 361 | + #[::core::prelude::v1::test] | |
| 362 | + fn moves_to_a_faster_route() { | |
| 363 | + let port = 48_000 + (std::process::id() % 90) as u16; | |
| 364 | + let key = new_key(); | |
| 365 | + let _server = RemoteServer::start(port, key.clone(), "test-host".into()).unwrap(); | |
| 366 | + // Nothing listens on the LAN route at first, as away from home. | |
| 367 | + let gone = Route { address: format!("127.0.0.1:{}", port + 100), kind: RouteKind::Lan, learned: true }; | |
| 368 | + let tailnet = Route { address: format!("127.0.0.1:{port}"), kind: RouteKind::Tailnet, learned: true }; | |
| 369 | + let timing = Timing { ping: Duration::from_millis(200), silence: Duration::from_secs(5), probe: Duration::from_millis(300) }; | |
| 370 | + let client = RemoteClient::connect_timed(vec![gone, tailnet.clone()], key.clone(), "test-client".into(), timing); | |
| 371 | + assert_eq!(connected_route(&client), tailnet.address); | |
| 372 | + | |
| 373 | + // Back home: a faster route answers, and the client moves to it. | |
| 374 | + let home = Route { address: format!("127.0.0.2:{port}"), kind: RouteKind::Lan, learned: true }; | |
| 375 | + client.set_routes(vec![home.clone(), tailnet.clone()]); | |
| 376 | + assert_eq!(connected_route(&client), home.address); | |
| 377 | + | |
| 378 | + // A route that's another device's (a wrong key) is skipped, not fatal. | |
| 379 | + let other_port = port + 200; | |
| 380 | + let _other = RemoteServer::start(other_port, new_key(), "someone-else".into()).unwrap(); | |
| 381 | + let theirs = Route { address: format!("127.0.0.1:{other_port}"), kind: RouteKind::Lan, learned: true }; | |
| 382 | + let client = RemoteClient::connect_timed(vec![theirs, tailnet.clone()], key, "test-client".into(), timing); | |
| 383 | + assert_eq!(connected_route(&client), tailnet.address); | |
| 184 | 384 | } |
| 185 | - "closed".into() | |
| 186 | 385 | } |
| @@ -20,6 +20,13 @@ | ||
| 20 | 20 | //! with [`ClientMessage::Device`], gets [`ServerMessage::DeviceReply`], and |
| 21 | 21 | //! follows [`ServerMessage::DeviceState`]; the video and touches go through |
| 22 | 22 | //! the host's [`device_proxy`] on the same port. |
| 23 | + //! | |
| 24 | + //! A host can be reached more than one way, like T3 Code's routes: over the | |
| 25 | + //! LAN at home and over Tailscale or another VPN away from it. The host | |
| 26 | + //! reports the [`Route`]s it listens on in [`ServerMessage::Welcome`] and | |
| 27 | + //! again in [`ServerMessage::Routes`] when they change; the client saves | |
| 28 | + //! them with the connection, tries them fastest first, and moves back to a | |
| 29 | + //! faster one as soon as it answers. | |
| 23 | 30 | |
| 24 | 31 | pub mod client; |
| 25 | 32 | pub mod device_proxy; |
| @@ -227,7 +234,14 @@ pub enum ClientMessage { | ||
| 227 | 234 | #[derive(Clone, Debug, Serialize, Deserialize)] |
| 228 | 235 | #[serde(tag = "type", rename_all = "snake_case")] |
| 229 | 236 | pub enum ServerMessage { |
| 230 | - Welcome { host: String, version: String, protocol: u32 }, | |
| 237 | + Welcome { | |
| 238 | + host: String, | |
| 239 | + version: String, | |
| 240 | + protocol: u32, | |
| 241 | + /// Where else the host listens. Older hosts don't say. | |
| 242 | + #[serde(default)] | |
| 243 | + routes: Vec<Route>, | |
| 244 | + }, | |
| 231 | 245 | Denied { reason: String }, |
| 232 | 246 | State { projects: Vec<ProjectInfo>, threads: Vec<ThreadSummary> }, |
| 233 | 247 | /// A subscribed thread changed: its state, and its transcript from item |
| @@ -254,8 +268,132 @@ pub enum ServerMessage { | ||
| 254 | 268 | DeviceOpened { thread: String, device: crate::device::DeviceSummary }, |
| 255 | 269 | /// An agent closed a thread's device, or all of them without one. |
| 256 | 270 | DeviceClosed { thread: String, device: Option<String> }, |
| 271 | + /// The host's addresses changed, as when it joined another Wi-Fi. | |
| 272 | + /// Older hosts don't send it. | |
| 273 | + Routes { routes: Vec<Route> }, | |
| 274 | + } | |
| 275 | + | |
| 276 | + /// What kind of network a route goes over, which says how fast it likely is. | |
| 277 | + #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] | |
| 278 | + #[serde(rename_all = "snake_case")] | |
| 279 | + pub enum RouteKind { | |
| 280 | + Loopback, | |
| 281 | + #[default] | |
| 282 | + Lan, | |
| 283 | + Tailnet, | |
| 284 | + /// Another VPN, or Tailscale's range on a host that didn't say so: | |
| 285 | + /// Cloudflare WARP and others share it. | |
| 286 | + Vpn, | |
| 287 | + /// A name or public address. | |
| 288 | + Other, | |
| 289 | + } | |
| 290 | + | |
| 291 | + impl RouteKind { | |
| 292 | + /// Guesses from an address alone, for addresses the user typed. | |
| 293 | + pub fn of(address: &str) -> Self { | |
| 294 | + let address = with_port(address); | |
| 295 | + if let Ok(socket) = address.parse::<std::net::SocketAddr>() { | |
| 296 | + return match socket.ip() { | |
| 297 | + std::net::IpAddr::V4(ip) => kind_of_v4(ip).unwrap_or(Self::Other), | |
| 298 | + std::net::IpAddr::V6(ip) if ip.is_loopback() => Self::Loopback, | |
| 299 | + std::net::IpAddr::V6(ip) if is_tailscale_v6(ip) => Self::Tailnet, | |
| 300 | + std::net::IpAddr::V6(_) => Self::Other, | |
| 301 | + }; | |
| 302 | + } | |
| 303 | + let name = address.rsplit_once(':').map_or(address.as_str(), |(name, _)| name).to_ascii_lowercase(); | |
| 304 | + if name == "localhost" { | |
| 305 | + Self::Loopback | |
| 306 | + } else if name.ends_with(".ts.net") { | |
| 307 | + Self::Tailnet | |
| 308 | + } else if name.ends_with(".local") || !name.contains('.') { | |
| 309 | + Self::Lan | |
| 310 | + } else { | |
| 311 | + Self::Other | |
| 312 | + } | |
| 313 | + } | |
| 314 | + | |
| 315 | + pub fn label(self) -> &'static str { | |
| 316 | + match self { | |
| 317 | + Self::Loopback => "This device", | |
| 318 | + Self::Lan => "LAN", | |
| 319 | + Self::Tailnet => "Tailscale", | |
| 320 | + Self::Vpn => "VPN", | |
| 321 | + Self::Other => "Direct", | |
| 322 | + } | |
| 323 | + } | |
| 324 | + | |
| 325 | + /// Faster first. | |
| 326 | + fn rank(self) -> u8 { | |
| 327 | + match self { | |
| 328 | + Self::Loopback => 0, | |
| 329 | + Self::Lan => 1, | |
| 330 | + Self::Tailnet | Self::Vpn => 2, | |
| 331 | + Self::Other => 3, | |
| 332 | + } | |
| 333 | + } | |
| 334 | + } | |
| 335 | + | |
| 336 | + /// One way to reach a host: `host:port`, and the network it goes over. | |
| 337 | + #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] | |
| 338 | + #[serde(default)] | |
| 339 | + pub struct Route { | |
| 340 | + pub address: String, | |
| 341 | + pub kind: RouteKind, | |
| 342 | + /// Reported by the host rather than saved by the user, so the next | |
| 343 | + /// report replaces it. | |
| 344 | + #[serde(skip_serializing_if = "std::ops::Not::not")] | |
| 345 | + pub learned: bool, | |
| 257 | 346 | } |
| 258 | 347 | |
| 348 | + impl Route { | |
| 349 | + pub fn guess(address: &str) -> Self { | |
| 350 | + Self { address: with_port(address), kind: RouteKind::of(address), learned: false } | |
| 351 | + } | |
| 352 | + | |
| 353 | + /// `LAN · 192.168.2.10:47823`. | |
| 354 | + pub fn describe(&self) -> String { | |
| 355 | + format!("{} · {}", self.kind.label(), self.address) | |
| 356 | + } | |
| 357 | + } | |
| 358 | + | |
| 359 | + /// Every route to a host, fastest first: the address the user saved and | |
| 360 | + /// the routes it reported. One the host reported keeps the host's kind, | |
| 361 | + /// which knows Tailscale from other VPNs. | |
| 362 | + pub fn ordered_routes(address: &str, routes: &[Route]) -> Vec<Route> { | |
| 363 | + let mut all: Vec<Route> = Vec::new(); | |
| 364 | + for route in routes.iter().cloned().chain((!address.trim().is_empty()).then(|| Route::guess(address))) { | |
| 365 | + let route = Route { address: with_port(&route.address), ..route }; | |
| 366 | + if route.address.is_empty() || all.iter().any(|known| known.address == route.address) { | |
| 367 | + continue; | |
| 368 | + } | |
| 369 | + all.push(route); | |
| 370 | + } | |
| 371 | + // The user's address goes first among equals. | |
| 372 | + let saved = with_port(address); | |
| 373 | + all.sort_by_key(|route| (route.kind.rank(), route.address != saved)); | |
| 374 | + all | |
| 375 | + } | |
| 376 | + | |
| 377 | + /// The routes saved after the host reports `reported`: the user's own stay, | |
| 378 | + /// the learned ones are replaced. `None` when nothing changed. | |
| 379 | + pub fn merge_routes(saved: &[Route], reported: &[Route]) -> Option<Vec<Route>> { | |
| 380 | + let mut next: Vec<Route> = saved.iter().filter(|route| !route.learned).cloned().collect(); | |
| 381 | + for route in reported { | |
| 382 | + // A loopback address names whichever device opens it. | |
| 383 | + if route.kind == RouteKind::Loopback || route.address.trim().is_empty() { | |
| 384 | + continue; | |
| 385 | + } | |
| 386 | + let address = with_port(&route.address); | |
| 387 | + match next.iter_mut().find(|known| known.address == address) { | |
| 388 | + // A saved route the host reports takes the host's word for its kind. | |
| 389 | + Some(known) => known.kind = route.kind, | |
| 390 | + None => next.push(Route { address, kind: route.kind, learned: true }), | |
| 391 | + } | |
| 392 | + } | |
| 393 | + (next != saved).then_some(next) | |
| 394 | + } | |
| 395 | + | |
| 396 | + | |
| 259 | 397 | /// A new random key: 24 bytes, hex. |
| 260 | 398 | pub fn new_key() -> String { |
| 261 | 399 | let mut bytes = [0u8; 24]; |
| @@ -275,19 +413,21 @@ pub fn keys_match(a: &str, b: &str) -> bool { | ||
| 275 | 413 | a.len() == b.len() && a.bytes().zip(b.bytes()).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0 |
| 276 | 414 | } |
| 277 | 415 | |
| 278 | - /// A pairing code bundles a host's address and key: `192.168.2.10:47823/<key>`. | |
| 279 | - pub fn pairing_code(address: &str, key: &str) -> String { | |
| 280 | - format!("{address}/{key}") | |
| 416 | + /// A pairing code bundles a host's addresses and key, fastest first: | |
| 417 | + /// `192.168.2.10:47823,100.101.4.2:47823/<key>`. | |
| 418 | + pub fn pairing_code(addresses: &[String], key: &str) -> String { | |
| 419 | + format!("{}/{key}", addresses.join(",")) | |
| 281 | 420 | } |
| 282 | 421 | |
| 283 | - /// Splits a pairing code, or an address alone, into address and key. | |
| 284 | - pub fn parse_pairing(code: &str) -> (String, Option<String>) { | |
| 422 | + /// Splits a pairing code, or an address alone, into addresses and key. | |
| 423 | + pub fn parse_pairing(code: &str) -> (Vec<String>, Option<String>) { | |
| 285 | 424 | let code = code.trim().trim_start_matches("ws://").trim_start_matches("vectrum://"); |
| 286 | - match code.split_once('/') { | |
| 287 | - Some((address, key)) if !key.trim().is_empty() => (address.to_string(), Some(key.trim().to_string())), | |
| 288 | - Some((address, _)) => (address.to_string(), None), | |
| 289 | - None => (code.to_string(), None), | |
| 290 | - } | |
| 425 | + let (addresses, key) = match code.split_once('/') { | |
| 426 | + Some((addresses, key)) => (addresses, Some(key.trim().to_string()).filter(|key| !key.is_empty())), | |
| 427 | + None => (code, None), | |
| 428 | + }; | |
| 429 | + let addresses = addresses.split(',').map(str::trim).filter(|address| !address.is_empty()).map(str::to_string).collect(); | |
| 430 | + (addresses, key) | |
| 291 | 431 | } |
| 292 | 432 | |
| 293 | 433 | /// `host` or `host:port`, with the default port filled in. |
| @@ -301,11 +441,107 @@ pub fn with_port(address: &str) -> String { | ||
| 301 | 441 | if has_port { address.to_string() } else { format!("{address}:{DEFAULT_PORT}") } |
| 302 | 442 | } |
| 303 | 443 | |
| 304 | - /// This machine's IPv4 addresses on the network, for showing where to connect. | |
| 305 | - pub fn local_addresses() -> Vec<String> { | |
| 306 | - let output = std::process::Command::new("hostname").arg("-I").stderr(std::process::Stdio::null()).output(); | |
| 307 | - let text = output.map(|output| String::from_utf8_lossy(&output.stdout).to_string()).unwrap_or_default(); | |
| 308 | - text.split_whitespace().filter(|address| address.parse::<std::net::Ipv4Addr>().is_ok()).map(str::to_string).collect() | |
| 444 | + /// Where this machine listens on `port`, LAN first: its private IPv4 | |
| 445 | + /// addresses, with those on Tailscale's interface marked as such. Public | |
| 446 | + /// addresses are left out, since connections aren't encrypted, and so are | |
| 447 | + /// container and VM networks only this machine reaches. | |
| 448 | + pub fn local_routes(port: u16) -> Vec<Route> { | |
| 449 | + routes_from(&interface_addresses(), port) | |
| 450 | + } | |
| 451 | + | |
| 452 | + fn routes_from(interfaces: &[(String, std::net::IpAddr)], port: u16) -> Vec<Route> { | |
| 453 | + use std::net::IpAddr; | |
| 454 | + // `tailscale0` on Linux; macOS names every tunnel `utunN`, so there it's | |
| 455 | + // the one that also has an address in Tailscale's IPv6 range. | |
| 456 | + let tailscale = |name: &str| { | |
| 457 | + name.to_ascii_lowercase().starts_with("tailscale") | |
| 458 | + || interfaces.iter().any(|(other, ip)| other == name && matches!(ip, IpAddr::V6(ip) if is_tailscale_v6(*ip))) | |
| 459 | + }; | |
| 460 | + const VIRTUAL: [&str; 14] = [ | |
| 461 | + "docker", "br-", "veth", "virbr", "vmnet", "vboxnet", "vEthernet", "podman", "cni", "flannel", "cali", "lxcbr", | |
| 462 | + "lxdbr", "waydroid", | |
| 463 | + ]; | |
| 464 | + let mut routes: Vec<Route> = Vec::new(); | |
| 465 | + for (name, ip) in interfaces { | |
| 466 | + let IpAddr::V4(ip) = ip else { continue }; | |
| 467 | + // macOS's `bridge100` and up are its VMs' networks. | |
| 468 | + let mac_bridge = name.strip_prefix("bridge").and_then(|n| n.parse::<u32>().ok()).is_some_and(|n| n >= 100); | |
| 469 | + if VIRTUAL.iter().any(|prefix| name.starts_with(prefix)) || mac_bridge { | |
| 470 | + continue; | |
| 471 | + } | |
| 472 | + let kind = match kind_of_v4(*ip) { | |
| 473 | + Some(RouteKind::Lan) => RouteKind::Lan, | |
| 474 | + Some(RouteKind::Vpn) if tailscale(name) => RouteKind::Tailnet, | |
| 475 | + Some(RouteKind::Vpn) => RouteKind::Vpn, | |
| 476 | + _ => continue, | |
| 477 | + }; | |
| 478 | + let address = format!("{ip}:{port}"); | |
| 479 | + if !routes.iter().any(|route| route.address == address) { | |
| 480 | + routes.push(Route { address, kind, learned: false }); | |
| 481 | + } | |
| 482 | + } | |
| 483 | + routes.sort_by_key(|route| route.kind.rank()); | |
| 484 | + routes | |
| 485 | + } | |
| 486 | + | |
| 487 | + /// `Lan` for private ranges, `Vpn` for the shared 100.64.0.0/10 that | |
| 488 | + /// Tailscale and other VPNs use, `Loopback`, or `None` for anything else. | |
| 489 | + fn kind_of_v4(ip: std::net::Ipv4Addr) -> Option<RouteKind> { | |
| 490 | + let [a, b, ..] = ip.octets(); | |
| 491 | + if ip.is_loopback() { | |
| 492 | + Some(RouteKind::Loopback) | |
| 493 | + } else if ip.is_private() { | |
| 494 | + Some(RouteKind::Lan) | |
| 495 | + } else if a == 100 && (64..128).contains(&b) { | |
| 496 | + Some(RouteKind::Vpn) | |
| 497 | + } else { | |
| 498 | + None | |
| 499 | + } | |
| 500 | + } | |
| 501 | + | |
| 502 | + /// Tailscale's IPv6 range, fd7a:115c:a1e0::/48. | |
| 503 | + fn is_tailscale_v6(ip: std::net::Ipv6Addr) -> bool { | |
| 504 | + let [a, b, c, ..] = ip.segments(); | |
| 505 | + (a, b, c) == (0xfd7a, 0x115c, 0xa1e0) | |
| 506 | + } | |
| 507 | + | |
| 508 | + /// Each network interface's name and addresses. | |
| 509 | + #[cfg(unix)] | |
| 510 | + fn interface_addresses() -> Vec<(String, std::net::IpAddr)> { | |
| 511 | + use std::net::{Ipv4Addr, Ipv6Addr}; | |
| 512 | + let mut list = Vec::new(); | |
| 513 | + let mut head: *mut libc::ifaddrs = std::ptr::null_mut(); | |
| 514 | + if unsafe { libc::getifaddrs(&mut head) } != 0 { | |
| 515 | + return list; | |
| 516 | + } | |
| 517 | + let mut cursor = head; | |
| 518 | + while !cursor.is_null() { | |
| 519 | + // The list stays valid until it's freed below. | |
| 520 | + let entry = unsafe { &*cursor }; | |
| 521 | + cursor = entry.ifa_next; | |
| 522 | + if entry.ifa_addr.is_null() || entry.ifa_name.is_null() { | |
| 523 | + continue; | |
| 524 | + } | |
| 525 | + let name = unsafe { std::ffi::CStr::from_ptr(entry.ifa_name) }.to_string_lossy().into_owned(); | |
| 526 | + match i32::from(unsafe { (*entry.ifa_addr).sa_family }) { | |
| 527 | + libc::AF_INET => { | |
| 528 | + let address = unsafe { &*(entry.ifa_addr as *const libc::sockaddr_in) }; | |
| 529 | + list.push((name, Ipv4Addr::from(u32::from_be(address.sin_addr.s_addr)).into())); | |
| 530 | + } | |
| 531 | + libc::AF_INET6 => { | |
| 532 | + let address = unsafe { &*(entry.ifa_addr as *const libc::sockaddr_in6) }; | |
| 533 | + list.push((name, Ipv6Addr::from(address.sin6_addr.s6_addr).into())); | |
| 534 | + } | |
| 535 | + _ => {} | |
| 536 | + } | |
| 537 | + } | |
| 538 | + unsafe { libc::freeifaddrs(head) }; | |
| 539 | + list | |
| 540 | + } | |
| 541 | + | |
| 542 | + #[cfg(not(unix))] | |
| 543 | + fn interface_addresses() -> Vec<(String, std::net::IpAddr)> { | |
| 544 | + Vec::new() | |
| 309 | 545 | } |
| 310 | 546 | |
| 311 | 547 | #[cfg(test)] |
| @@ -314,14 +550,74 @@ mod tests { | ||
| 314 | 550 | |
| 315 | 551 | #[::core::prelude::v1::test] |
| 316 | 552 | fn pairing_codes() { |
| 317 | - assert_eq!(parse_pairing("192.168.2.10:47823/abc"), ("192.168.2.10:47823".into(), Some("abc".into()))); | |
| 318 | - assert_eq!(parse_pairing(" 10.0.0.2 "), ("10.0.0.2".into(), None)); | |
| 553 | + assert_eq!(parse_pairing("192.168.2.10:47823/abc"), (vec!["192.168.2.10:47823".into()], Some("abc".into()))); | |
| 554 | + assert_eq!(parse_pairing(" 10.0.0.2 "), (vec!["10.0.0.2".into()], None)); | |
| 555 | + let code = pairing_code(&["192.168.2.10:47823".into(), "100.101.4.2:47823".into()], "abc"); | |
| 556 | + assert_eq!(parse_pairing(&code), (vec!["192.168.2.10:47823".into(), "100.101.4.2:47823".into()], Some("abc".into()))); | |
| 319 | 557 | assert_eq!(with_port("10.0.0.2"), format!("10.0.0.2:{DEFAULT_PORT}")); |
| 320 | 558 | assert_eq!(with_port("10.0.0.2:9000"), "10.0.0.2:9000"); |
| 321 | 559 | assert_eq!(with_port("box.local"), format!("box.local:{DEFAULT_PORT}")); |
| 322 | 560 | assert_eq!(with_port("[::1]:9000"), "[::1]:9000"); |
| 323 | 561 | } |
| 324 | 562 | |
| 563 | + #[::core::prelude::v1::test] | |
| 564 | + fn routes() { | |
| 565 | + use std::net::IpAddr; | |
| 566 | + let interfaces: Vec<(String, IpAddr)> = [ | |
| 567 | + ("lo", "127.0.0.1"), | |
| 568 | + ("tailscale0", "100.101.4.2"), | |
| 569 | + ("wlan0", "192.168.2.10"), | |
| 570 | + ("wlan0", "fe80::1"), | |
| 571 | + ("docker0", "172.17.0.1"), | |
| 572 | + ("warp0", "100.96.0.3"), | |
| 573 | + ("eth9", "8.8.8.8"), | |
| 574 | + ("utun4", "100.80.1.1"), | |
| 575 | + ("utun4", "fd7a:115c:a1e0::1"), | |
| 576 | + ] | |
| 577 | + .into_iter() | |
| 578 | + .map(|(name, ip)| (name.to_string(), ip.parse().unwrap())) | |
| 579 | + .collect(); | |
| 580 | + let found = routes_from(&interfaces, 9000); | |
| 581 | + let found: Vec<(&str, RouteKind)> = found.iter().map(|route| (route.address.as_str(), route.kind)).collect(); | |
| 582 | + assert_eq!( | |
| 583 | + found, | |
| 584 | + [ | |
| 585 | + ("192.168.2.10:9000", RouteKind::Lan), | |
| 586 | + ("100.101.4.2:9000", RouteKind::Tailnet), | |
| 587 | + ("100.96.0.3:9000", RouteKind::Vpn), | |
| 588 | + ("100.80.1.1:9000", RouteKind::Tailnet), | |
| 589 | + ] | |
| 590 | + ); | |
| 591 | + | |
| 592 | + assert_eq!(RouteKind::of("192.168.2.10"), RouteKind::Lan); | |
| 593 | + assert_eq!(RouteKind::of("100.101.4.2:47823"), RouteKind::Vpn); | |
| 594 | + assert_eq!(RouteKind::of("box.tail1234.ts.net"), RouteKind::Tailnet); | |
| 595 | + assert_eq!(RouteKind::of("box.local"), RouteKind::Lan); | |
| 596 | + assert_eq!(RouteKind::of("example.com"), RouteKind::Other); | |
| 597 | + assert_eq!(RouteKind::of("127.0.0.1:9000"), RouteKind::Loopback); | |
| 598 | + | |
| 599 | + // Paired over Tailscale; the host then reports its LAN address too. | |
| 600 | + let reported = [ | |
| 601 | + Route { address: "192.168.2.10:47823".into(), kind: RouteKind::Lan, learned: false }, | |
| 602 | + Route { address: "100.101.4.2:47823".into(), kind: RouteKind::Tailnet, learned: false }, | |
| 603 | + Route { address: "127.0.0.1:47823".into(), kind: RouteKind::Loopback, learned: false }, | |
| 604 | + ]; | |
| 605 | + let saved = merge_routes(&[], &reported).unwrap(); | |
| 606 | + assert_eq!(saved.len(), 2); | |
| 607 | + assert!(saved.iter().all(|route| route.learned)); | |
| 608 | + assert_eq!(merge_routes(&saved, &reported), None); | |
| 609 | + let order: Vec<String> = ordered_routes("100.101.4.2", &saved).into_iter().map(|route| route.describe()).collect(); | |
| 610 | + assert_eq!(order, ["LAN · 192.168.2.10:47823", "Tailscale · 100.101.4.2:47823"]); | |
| 611 | + | |
| 612 | + // The host joined another Wi-Fi: its old LAN address goes, the user's own route stays. | |
| 613 | + let mut saved = saved; | |
| 614 | + saved.push(Route::guess("10.0.0.5")); | |
| 615 | + let moved = [Route { address: "192.168.7.20:47823".into(), kind: RouteKind::Lan, learned: false }]; | |
| 616 | + let next = merge_routes(&saved, &moved).unwrap(); | |
| 617 | + let next: Vec<&str> = next.iter().map(|route| route.address.as_str()).collect(); | |
| 618 | + assert_eq!(next, ["10.0.0.5:47823", "192.168.7.20:47823"]); | |
| 619 | + } | |
| 620 | + | |
| 325 | 621 | #[::core::prelude::v1::test] |
| 326 | 622 | fn keys() { |
| 327 | 623 | let key = new_key(); |
| @@ -409,9 +705,15 @@ mod tests { | ||
| 409 | 705 | }) |
| 410 | 706 | } |
| 411 | 707 | |
| 412 | - let client = RemoteClient::connect(&format!("127.0.0.1:{port}"), key, "test-client".into()); | |
| 708 | + let client = RemoteClient::connect(vec![Route::guess(&format!("127.0.0.1:{port}"))], key, "test-client".into()); | |
| 413 | 709 | assert!(matches!(wait(&client.events), Some(ClientEvent::Status(ConnectionStatus::Connecting)))); |
| 414 | 710 | match wait(&client.events) { |
| 711 | + Some(ClientEvent::Route(route)) => assert_eq!(route.address, format!("127.0.0.1:{port}")), | |
| 712 | + _ => panic!("didn't say which route"), | |
| 713 | + } | |
| 714 | + assert!(matches!(wait(&client.events), Some(ClientEvent::Message(ServerMessage::Routes { .. })))); | |
| 715 | + match wait(&client.events) { | |
| 716 | + | |
| 415 | 717 | Some(ClientEvent::Status(ConnectionStatus::Connected(host))) => assert_eq!(host, "test-host"), |
| 416 | 718 | _ => panic!("didn't connect"), |
| 417 | 719 | } |
| @@ -435,7 +737,7 @@ mod tests { | ||
| 435 | 737 | _ => panic!("client didn't get the reply"), |
| 436 | 738 | } |
| 437 | 739 | |
| 438 | - let intruder = RemoteClient::connect(&format!("127.0.0.1:{port}"), "wrong".into(), "intruder".into()); | |
| 740 | + let intruder = RemoteClient::connect(vec![Route::guess(&format!("127.0.0.1:{port}"))], "wrong".into(), "intruder".into()); | |
| 439 | 741 | assert!(matches!(wait(&intruder.events), Some(ClientEvent::Status(ConnectionStatus::Connecting)))); |
| 440 | 742 | assert!(matches!(wait(&intruder.events), Some(ClientEvent::Status(ConnectionStatus::Denied(_))))); |
| 441 | 743 | } |
| @@ -12,17 +12,22 @@ use std::{ | ||
| 12 | 12 | mpsc, |
| 13 | 13 | }, |
| 14 | 14 | thread, |
| 15 | - time::Duration, | |
| 15 | + time::{Duration, Instant}, | |
| 16 | 16 | }; |
| 17 | 17 | |
| 18 | 18 | use tungstenite::{Message, WebSocket}; |
| 19 | 19 | |
| 20 | - use super::{ClientMessage, PROTOCOL, ServerMessage, keys_match}; | |
| 20 | + use super::{ClientMessage, PROTOCOL, Route, ServerMessage, keys_match, local_routes}; | |
| 21 | 21 | |
| 22 | 22 | /// How long a connection may take to say hello. |
| 23 | 23 | const HELLO_TIMEOUT: Duration = Duration::from_secs(10); |
| 24 | 24 | /// How often a connection's thread looks for messages to send. |
| 25 | 25 | const POLL: Duration = Duration::from_millis(40); |
| 26 | + /// How often the listener looks for changed addresses. | |
| 27 | + const RESCAN: Duration = Duration::from_secs(30); | |
| 28 | + /// A client that pings and then goes quiet this long went away, as when its | |
| 29 | + /// network did. | |
| 30 | + const SILENCE: Duration = Duration::from_secs(60); | |
| 26 | 31 | |
| 27 | 32 | pub enum ServerEvent { |
| 28 | 33 | Connected { conn: u64, name: String, address: String }, |
| @@ -42,6 +47,8 @@ pub struct RemoteServer { | ||
| 42 | 47 | pub port: u16, |
| 43 | 48 | stop: Arc<AtomicBool>, |
| 44 | 49 | outboxes: Arc<Mutex<HashMap<u64, mpsc::Sender<String>>>>, |
| 50 | + /// Where this device listens now. | |
| 51 | + routes: Arc<Mutex<Vec<Route>>>, | |
| 45 | 52 | pub events: smol::channel::Receiver<ServerEvent>, |
| 46 | 53 | } |
| 47 | 54 | |
| @@ -54,18 +61,35 @@ impl RemoteServer { | ||
| 54 | 61 | let outboxes: Arc<Mutex<HashMap<u64, mpsc::Sender<String>>>> = Arc::default(); |
| 55 | 62 | let (events_tx, events) = smol::channel::unbounded(); |
| 56 | 63 | let next_conn = Arc::new(AtomicU64::new(1)); |
| 64 | + let routes = Arc::new(Mutex::new(local_routes(port))); | |
| 57 | 65 | { |
| 58 | 66 | let stop = stop.clone(); |
| 59 | 67 | let outboxes = outboxes.clone(); |
| 68 | + let routes = routes.clone(); | |
| 60 | 69 | thread::Builder::new().name("remote-listener".into()).spawn(move || { |
| 70 | + let mut scanned = Instant::now(); | |
| 61 | 71 | while !stop.load(Ordering::Relaxed) { |
| 72 | + if scanned.elapsed() >= RESCAN { | |
| 73 | + scanned = Instant::now(); | |
| 74 | + let now = local_routes(port); | |
| 75 | + let changed = { | |
| 76 | + let mut routes = routes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); | |
| 77 | + let changed = *routes != now; | |
| 78 | + *routes = now.clone(); | |
| 79 | + changed | |
| 80 | + }; | |
| 81 | + if changed { | |
| 82 | + broadcast_to(&outboxes, &ServerMessage::Routes { routes: now }); | |
| 83 | + } | |
| 84 | + } | |
| 62 | 85 | match listener.accept() { |
| 63 | 86 | Ok((stream, address)) => { |
| 64 | 87 | let conn = next_conn.fetch_add(1, Ordering::Relaxed); |
| 65 | - let (stop, outboxes, events, key, host) = | |
| 66 | - (stop.clone(), outboxes.clone(), events_tx.clone(), key.clone(), host.clone()); | |
| 88 | + let (stop, outboxes, events, key, host, routes) = | |
| 89 | + (stop.clone(), outboxes.clone(), events_tx.clone(), key.clone(), host.clone(), routes.clone()); | |
| 67 | 90 | let _ = thread::Builder::new().name("remote-connection".into()).spawn(move || { |
| 68 | - serve(conn, stream, address.ip().to_string(), &key, &host, &stop, &outboxes, &events); | |
| 91 | + let routes = routes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).clone(); | |
| 92 | + serve(conn, stream, address.ip().to_string(), &key, &host, routes, &stop, &outboxes, &events); | |
| 69 | 93 | outboxes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).remove(&conn); |
| 70 | 94 | }); |
| 71 | 95 | } |
| @@ -75,7 +99,12 @@ impl RemoteServer { | ||
| 75 | 99 | } |
| 76 | 100 | })?; |
| 77 | 101 | } |
| 78 | - Ok(Self { port, stop, outboxes, events }) | |
| 102 | + Ok(Self { port, stop, outboxes, routes, events }) | |
| 103 | + } | |
| 104 | + | |
| 105 | + /// Where this device listens now, LAN first. | |
| 106 | + pub fn routes(&self) -> Vec<Route> { | |
| 107 | + self.routes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).clone() | |
| 79 | 108 | } |
| 80 | 109 | |
| 81 | 110 | pub fn send(&self, conn: u64, message: &ServerMessage) { |
| @@ -88,10 +117,7 @@ impl RemoteServer { | ||
| 88 | 117 | } |
| 89 | 118 | |
| 90 | 119 | pub fn broadcast(&self, message: &ServerMessage) { |
| 91 | - let Ok(text) = serde_json::to_string(message) else { return }; | |
| 92 | - for outbox in self.outboxes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).values() { | |
| 93 | - let _ = outbox.send(text.clone()); | |
| 94 | - } | |
| 120 | + broadcast_to(&self.outboxes, message); | |
| 95 | 121 | } |
| 96 | 122 | |
| 97 | 123 | /// Drops a client's connection. |
| @@ -101,6 +127,13 @@ impl RemoteServer { | ||
| 101 | 127 | } |
| 102 | 128 | } |
| 103 | 129 | |
| 130 | + fn broadcast_to(outboxes: &Mutex<HashMap<u64, mpsc::Sender<String>>>, message: &ServerMessage) { | |
| 131 | + let Ok(text) = serde_json::to_string(message) else { return }; | |
| 132 | + for outbox in outboxes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).values() { | |
| 133 | + let _ = outbox.send(text.clone()); | |
| 134 | + } | |
| 135 | + } | |
| 136 | + | |
| 104 | 137 | impl Drop for RemoteServer { |
| 105 | 138 | fn drop(&mut self) { |
| 106 | 139 | self.stop.store(true, Ordering::Relaxed); |
| @@ -130,6 +163,7 @@ fn serve( | ||
| 130 | 163 | address: String, |
| 131 | 164 | key: &str, |
| 132 | 165 | host: &str, |
| 166 | + routes: Vec<Route>, | |
| 133 | 167 | stop: &AtomicBool, |
| 134 | 168 | outboxes: &Mutex<HashMap<u64, mpsc::Sender<String>>>, |
| 135 | 169 | events: &smol::channel::Sender<ServerEvent>, |
| @@ -157,7 +191,8 @@ fn serve( | ||
| 157 | 191 | return; |
| 158 | 192 | } |
| 159 | 193 | }; |
| 160 | - let welcome = ServerMessage::Welcome { host: host.to_string(), version: env!("CARGO_PKG_VERSION").into(), protocol: PROTOCOL }; | |
| 194 | + let welcome = | |
| 195 | + ServerMessage::Welcome { host: host.to_string(), version: env!("CARGO_PKG_VERSION").into(), protocol: PROTOCOL, routes }; | |
| 161 | 196 | if socket.send(Message::text(serde_json::to_string(&welcome).unwrap_or_default())).is_err() { |
| 162 | 197 | return; |
| 163 | 198 | } |
| @@ -165,8 +200,14 @@ fn serve( | ||
| 165 | 200 | outboxes.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).insert(conn, outbox); |
| 166 | 201 | let _ = events.send_blocking(ServerEvent::Connected { conn, name, address }); |
| 167 | 202 | let _ = socket.get_mut().set_read_timeout(Some(POLL)); |
| 203 | + // Clients that ping are dropped once they go quiet; older ones don't ping. | |
| 204 | + let mut pinged = false; | |
| 205 | + let mut heard = Instant::now(); | |
| 168 | 206 | |
| 169 | 207 | 'connection: while !stop.load(Ordering::Relaxed) { |
| 208 | + if pinged && heard.elapsed() > SILENCE { | |
| 209 | + break; | |
| 210 | + } | |
| 170 | 211 | loop { |
| 171 | 212 | match inbox.try_recv() { |
| 172 | 213 | Ok(text) => { |
| @@ -178,13 +219,19 @@ fn serve( | ||
| 178 | 219 | Err(mpsc::TryRecvError::Disconnected) => break 'connection, |
| 179 | 220 | } |
| 180 | 221 | } |
| 181 | - match socket.read() { | |
| 222 | + let read = socket.read(); | |
| 223 | + if read.is_ok() { | |
| 224 | + heard = Instant::now(); | |
| 225 | + } | |
| 226 | + match read { | |
| 182 | 227 | Ok(Message::Text(text)) => { |
| 183 | 228 | if let Ok(message) = serde_json::from_str::<ClientMessage>(text.as_str()) { |
| 184 | 229 | let _ = events.send_blocking(ServerEvent::Message { conn, message }); |
| 185 | 230 | } |
| 186 | 231 | } |
| 187 | 232 | Ok(Message::Close(_)) => break, |
| 233 | + // Its pong goes out with the next read or send. | |
| 234 | + Ok(Message::Ping(_)) => pinged = true, | |
| 188 | 235 | Ok(_) => {} |
| 189 | 236 | Err(tungstenite::Error::Io(err)) if matches!(err.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => {} |
| 190 | 237 | Err(_) => break, |
| @@ -342,12 +342,21 @@ pub struct RemoteHost { | ||
| 342 | 342 | /// Whether the user was asked, on first connecting, to give the device |
| 343 | 343 | /// this one's environment settings. |
| 344 | 344 | pub settings_offered: bool, |
| 345 | + /// Other ways to reach it: those it reported, such as its Tailscale | |
| 346 | + /// address when paired over Wi-Fi, and others the user paired over. | |
| 347 | + #[serde(skip_serializing_if = "Vec::is_empty")] | |
| 348 | + pub routes: Vec<crate::remote::Route>, | |
| 345 | 349 | } |
| 346 | 350 | |
| 347 | 351 | impl RemoteHost { |
| 348 | 352 | pub fn label(&self) -> String { |
| 349 | 353 | if self.name.trim().is_empty() { self.address.clone() } else { self.name.trim().to_string() } |
| 350 | 354 | } |
| 355 | + | |
| 356 | + /// Every way to reach it, fastest first. | |
| 357 | + pub fn ordered_routes(&self) -> Vec<crate::remote::Route> { | |
| 358 | + crate::remote::ordered_routes(&self.address, &self.routes) | |
| 359 | + } | |
| 351 | 360 | } |
| 352 | 361 | |
| 353 | 362 | /// A usage hub the Usage page also reads, such as a CLIProxyAPI server. |
| @@ -28,7 +28,8 @@ pub struct RemoteDevices { | ||
| 28 | 28 | /// The device's name, shown where this machine would say "This machine". |
| 29 | 29 | pub name: String, |
| 30 | 30 | sender: ClientSender, |
| 31 | - hub: HubTarget, | |
| 31 | + /// Follows the route the connection goes over. | |
| 32 | + hub: Mutex<HubTarget>, | |
| 32 | 33 | pending: Mutex<HashMap<u64, smol::channel::Sender<Result<Value, String>>>>, |
| 33 | 34 | next: AtomicU64, |
| 34 | 35 | } |
| @@ -38,7 +39,12 @@ impl RemoteDevices { | ||
| 38 | 39 | /// proxy answers on too. |
| 39 | 40 | pub fn new(env: String, name: String, sender: ClientSender, address: &str, key: String) -> Self { |
| 40 | 41 | let hub = HubTarget { address: crate::remote::with_port(address), prefix: device_proxy::PREFIX.into(), key: Some(key) }; |
| 41 | - Self { env, name, sender, hub, pending: Mutex::default(), next: AtomicU64::new(1) } | |
| 42 | + Self { env, name, sender, hub: Mutex::new(hub), pending: Mutex::default(), next: AtomicU64::new(1) } | |
| 43 | + } | |
| 44 | + | |
| 45 | + /// The connection moved to another route; new streams go over it too. | |
| 46 | + pub fn set_address(&self, address: &str) { | |
| 47 | + self.hub.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).address = crate::remote::with_port(address); | |
| 42 | 48 | } |
| 43 | 49 | |
| 44 | 50 | /// Asks the device to send its device state now and on every change. |
| @@ -108,7 +114,7 @@ impl Backend { | ||
| 108 | 114 | pub fn hub(&self) -> Option<HubTarget> { |
| 109 | 115 | match self { |
| 110 | 116 | Backend::Local => device::service().hub_origin().map(|origin| HubTarget::local(&origin)), |
| 111 | - Backend::Remote(remote) => Some(remote.hub.clone()), | |
| 117 | + Backend::Remote(remote) => Some(remote.hub.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).clone()), | |
| 112 | 118 | } |
| 113 | 119 | } |
| 114 | 120 | |
| @@ -62,16 +62,48 @@ impl SettingsView { | ||
| 62 | 62 | fn add_connection(&mut self, window: &mut Window, cx: &mut Context<Self>) { |
| 63 | 63 | let state = &self.connections; |
| 64 | 64 | let value = |input: &Entity<InputState>| input.read(cx).value().trim().to_string(); |
| 65 | - let (address, code_key) = remote::parse_pairing(&value(&state.connect_address)); | |
| 65 | + let (addresses, code_key) = remote::parse_pairing(&value(&state.connect_address)); | |
| 66 | 66 | let key = Some(value(&state.connect_key)).filter(|key| !key.is_empty()).or(code_key); |
| 67 | - let (Some(key), false) = (key, address.is_empty()) else { | |
| 67 | + let (Some(key), Some((address, others))) = (key, addresses.split_first()) else { | |
| 68 | 68 | window.push_notification(Notification::error("Enter the other device's pairing code, or its address and key."), cx); |
| 69 | 69 | return; |
| 70 | 70 | }; |
| 71 | + let address = address.clone(); | |
| 72 | + // The code's other addresses, until the device reports its own. | |
| 73 | + let learned: Vec<remote::Route> = | |
| 74 | + others.iter().map(|other| remote::Route { learned: true, ..remote::Route::guess(other) }).collect(); | |
| 71 | 75 | let name = value(&state.connect_name); |
| 72 | - let ids: Vec<String> = settings::get(|settings| settings.remotes.iter().map(|host| host.id.clone()).collect()); | |
| 73 | - let id = (1..).map(|n| format!("remote-{n}")).find(|id| !ids.contains(id)).unwrap_or_default(); | |
| 74 | - settings::update(|settings| settings.remotes.push(RemoteHost { id, name, address, key, enabled: true, settings_offered: false })); | |
| 76 | + let known = settings::get(|settings| settings.remotes.iter().find(|host| host.key == key).map(RemoteHost::label)); | |
| 77 | + if let Some(label) = known { | |
| 78 | + // The same device paired over another address: another route to it. | |
| 79 | + settings::update(|settings| { | |
| 80 | + let Some(host) = settings.remotes.iter_mut().find(|host| host.key == key) else { return }; | |
| 81 | + let route = remote::Route::guess(&address); | |
| 82 | + if !host.ordered_routes().iter().any(|known| known.address == route.address) { | |
| 83 | + host.routes.push(route); | |
| 84 | + } | |
| 85 | + for route in learned { | |
| 86 | + if !host.ordered_routes().iter().any(|known| known.address == route.address) { | |
| 87 | + host.routes.push(route); | |
| 88 | + } | |
| 89 | + } | |
| 90 | + }); | |
| 91 | + window.push_notification(Notification::success(format!("Added another route to {label}.")), cx); | |
| 92 | + } else { | |
| 93 | + let ids: Vec<String> = settings::get(|settings| settings.remotes.iter().map(|host| host.id.clone()).collect()); | |
| 94 | + let id = (1..).map(|n| format!("remote-{n}")).find(|id| !ids.contains(id)).unwrap_or_default(); | |
| 95 | + settings::update(|settings| { | |
| 96 | + settings.remotes.push(RemoteHost { | |
| 97 | + id, | |
| 98 | + name, | |
| 99 | + address, | |
| 100 | + key, | |
| 101 | + enabled: true, | |
| 102 | + settings_offered: false, | |
| 103 | + routes: learned, | |
| 104 | + }) | |
| 105 | + }); | |
| 106 | + } | |
| 75 | 107 | for input in [&state.connect_name, &state.connect_address, &state.connect_key] { |
| 76 | 108 | input.update(cx, |input, cx| input.set_value("", window, cx)); |
| 77 | 109 | } |
| @@ -89,12 +121,11 @@ impl SettingsView { | ||
| 89 | 121 | let saved = settings::load(); |
| 90 | 122 | let view = cx.weak_entity(); |
| 91 | 123 | let overview = self.workspace.upgrade().map(|workspace| workspace.read(cx).remote_overview()); |
| 92 | - let (serving, error, peers, statuses) = match overview { | |
| 93 | - Some(overview) => (overview.serving, overview.error, overview.peers, overview.environments), | |
| 94 | - None => (None, None, Vec::new(), Vec::new()), | |
| 124 | + let (serving, routes, error, peers, statuses) = match overview { | |
| 125 | + Some(overview) => (overview.serving, overview.routes, overview.error, overview.peers, overview.environments), | |
| 126 | + None => (None, Vec::new(), None, Vec::new(), Vec::new()), | |
| 95 | 127 | }; |
| 96 | - let addresses: Vec<String> = | |
| 97 | - remote::local_addresses().into_iter().map(|address| format!("{address}:{}", saved.remote_port)).collect(); | |
| 128 | + let addresses: Vec<String> = routes.iter().map(|route| route.address.clone()).collect(); | |
| 98 | 129 | let row = |title: &'static str| SettingRow::new(title).page_width(WIDE_WIDTH); |
| 99 | 130 | |
| 100 | 131 | let mut machine = Section::new("This machine") |
| @@ -113,9 +144,12 @@ impl SettingsView { | ||
| 113 | 144 | (Some(error), _) => (Led::Failed, error.clone()), |
| 114 | 145 | (None, Some(port)) => ( |
| 115 | 146 | Led::Live, |
| 116 | - match addresses.as_slice() { | |
| 147 | + match routes.as_slice() { | |
| 117 | 148 | [] => format!("Listening on port {port}."), |
| 118 | - addresses => format!("Listening at {}.", addresses.join(", ")), | |
| 149 | + routes => format!( | |
| 150 | + "Listening at {}.", | |
| 151 | + routes.iter().map(|route| format!("{} ({})", route.address, route.kind.label())).collect::<Vec<_>>().join(", ") | |
| 152 | + ), | |
| 119 | 153 | }, |
| 120 | 154 | ), |
| 121 | 155 | (None, None) => (Led::Waiting, "Starting…".to_string()), |
| @@ -191,11 +225,13 @@ impl SettingsView { | ||
| 191 | 225 | ), |
| 192 | 226 | ), |
| 193 | 227 | ); |
| 194 | - if let Some(address) = addresses.first() { | |
| 195 | - let code = remote::pairing_code(address, &key); | |
| 228 | + if !addresses.is_empty() { | |
| 229 | + let code = remote::pairing_code(&addresses, &key); | |
| 196 | 230 | machine = machine.row( |
| 197 | 231 | row("Pairing code") |
| 198 | - .description("Paste it into Settings → Connections on the other device: the address and key in one.") | |
| 232 | + .description( | |
| 233 | + "Paste it into Settings → Connections on the other device: every address and the key in one. The other device then reaches this one over the LAN, Tailscale or another VPN, whichever works.", | |
| 234 | + ) | |
| 199 | 235 | .control( |
| 200 | 236 | MetalButton::new("copy-pairing").compact().icon(IconName::Copy).label("Copy pairing code").on_click( |
| 201 | 237 | move |_, window, cx| { |
| @@ -240,7 +276,10 @@ impl SettingsView { | ||
| 240 | 276 | environments = environments.row(empty_row("No saved remote environments", p)); |
| 241 | 277 | } |
| 242 | 278 | for host in &saved.remotes { |
| 243 | - let status = statuses.iter().find(|(id, _)| *id == host.id).map(|(_, status)| status.clone()); | |
| 279 | + let (status, route) = statuses | |
| 280 | + .iter() | |
| 281 | + .find(|(id, ..)| *id == host.id) | |
| 282 | + .map_or((None, None), |(_, status, route)| (Some(status.clone()), route.clone())); | |
| 244 | 283 | let (led, text) = match (&status, host.enabled) { |
| 245 | 284 | (_, false) => (Led::Off, "Off".to_string()), |
| 246 | 285 | (Some(ConnectionStatus::Connected(name)), _) => (Led::Live, format!("Connected to {name}")), |
| @@ -255,7 +294,7 @@ impl SettingsView { | ||
| 255 | 294 | SettingRow::new(host.label()) |
| 256 | 295 | .page_width(WIDE_WIDTH) |
| 257 | 296 | .title_suffix(metal::led(SharedString::from(format!("remote-led-{}", host.id)), led, connecting, p)) |
| 258 | - .description(format!("{} · {text}", remote::with_port(&host.address))) | |
| 297 | + .description(environment_description(host, route.as_ref().filter(|_| matches!(status, Some(ConnectionStatus::Connected(_)))), &text)) | |
| 259 | 298 | .control( |
| 260 | 299 | h_flex() |
| 261 | 300 | .gap(px(8.)) |
| @@ -334,3 +373,23 @@ impl SettingsView { | ||
| 334 | 373 | vec![machine.render(cx), environments.render(cx)] |
| 335 | 374 | } |
| 336 | 375 | } |
| 376 | + | |
| 377 | + /// Where a saved environment is reached: the route in use while connected, | |
| 378 | + /// and the others it can fall back on. | |
| 379 | + fn environment_description(host: &RemoteHost, active: Option<&remote::Route>, status: &str) -> String { | |
| 380 | + let routes = host.ordered_routes(); | |
| 381 | + let shown = active.or(routes.first()); | |
| 382 | + let mut text = match shown { | |
| 383 | + Some(route) => format!("{} · {status}", route.describe()), | |
| 384 | + None => status.to_string(), | |
| 385 | + }; | |
| 386 | + let others: Vec<String> = routes | |
| 387 | + .iter() | |
| 388 | + .filter(|route| shown.is_none_or(|shown| shown.address != route.address)) | |
| 389 | + .map(|route| format!("{} {}", route.kind.label(), route.address)) | |
| 390 | + .collect(); | |
| 391 | + if !others.is_empty() { | |
| 392 | + text.push_str(&format!(" · also {}", others.join(", "))); | |
| 393 | + } | |
| 394 | + text | |
| 395 | + } | |
| @@ -164,7 +164,7 @@ impl Workspace { | ||
| 164 | 164 | env.to_string(), |
| 165 | 165 | environment.host.label(), |
| 166 | 166 | environment.client.sender(), |
| 167 | - &environment.host.address, | |
| 167 | + environment.route.as_ref().map_or(&environment.host.address, |route| &route.address), | |
| 168 | 168 | environment.host.key.clone(), |
| 169 | 169 | )); |
| 170 | 170 | link.watch(); |
| @@ -172,6 +172,13 @@ impl Workspace { | ||
| 172 | 172 | Some(link) |
| 173 | 173 | } |
| 174 | 174 | |
| 175 | + /// A connected device's connection moved to another route. | |
| 176 | + pub(super) fn move_devices_link(&self, env: &str, address: &str) { | |
| 177 | + if let Some(link) = self.devices.links.get(env) { | |
| 178 | + link.set_address(address); | |
| 179 | + } | |
| 180 | + } | |
| 181 | + | |
| 175 | 182 | /// A connected device went away: its devices with it. |
| 176 | 183 | pub(super) fn forget_devices_link(&mut self, env: &str, cx: &mut Context<Self>) { |
| 177 | 184 | self.devices.links.remove(env); |
| @@ -588,7 +588,7 @@ impl Workspace { | ||
| 588 | 588 | let item = PaletteItem::new("Environments", label.clone(), Lucide::Laptop); |
| 589 | 589 | items.push(match &environment.status { |
| 590 | 590 | ConnectionStatus::Connected(_) => item |
| 591 | - .description(environment.host.address.clone()) | |
| 591 | + .description(environment.route.as_ref().map_or_else(|| environment.host.address.clone(), crate::remote::Route::describe)) | |
| 592 | 592 | .submenu(PaletteView::ProjectSources(Some(environment.host.id.clone()))), |
| 593 | 593 | ConnectionStatus::Connecting => item.description("Connecting…").hint(format!("{label} isn't connected yet.")), |
| 594 | 594 | ConnectionStatus::Failed(_) | ConnectionStatus::Denied(_) => { |
| @@ -19,7 +19,7 @@ use super::{Project, ThreadSettings, Workspace, hostname, initials, tint_for}; | ||
| 19 | 19 | use crate::claude::PermissionMode; |
| 20 | 20 | use crate::git::GitInfo; |
| 21 | 21 | use crate::remote::{ |
| 22 | - ClientMessage, ProjectInfo, ServerMessage, ThreadAction, ThreadSummary, | |
| 22 | + ClientMessage, ProjectInfo, Route, ServerMessage, ThreadAction, ThreadSummary, | |
| 23 | 23 | client::{ClientEvent, ConnectionStatus, RemoteClient}, |
| 24 | 24 | new_key, |
| 25 | 25 | server::{Peer, RemoteServer, ServerEvent}, |
| @@ -42,6 +42,8 @@ pub(super) const FLUSH: Duration = Duration::from_millis(150); | ||
| 42 | 42 | pub(super) struct Environment { |
| 43 | 43 | pub(super) host: RemoteHost, |
| 44 | 44 | pub(super) status: ConnectionStatus, |
| 45 | + /// The route the connection goes over, once it connected. | |
| 46 | + pub(super) route: Option<Route>, | |
| 45 | 47 | pub(super) client: Rc<RemoteClient>, |
| 46 | 48 | /// Its environment settings, once it shared them. |
| 47 | 49 | pub(super) settings: Option<Settings>, |
| @@ -78,10 +80,13 @@ pub(super) struct RemoteState { | ||
| 78 | 80 | pub struct RemoteOverview { |
| 79 | 81 | /// The port this device listens on, when it does. |
| 80 | 82 | pub serving: Option<u16>, |
| 83 | + /// Where it listens, while it does. | |
| 84 | + pub routes: Vec<Route>, | |
| 81 | 85 | pub error: Option<String>, |
| 82 | 86 | pub peers: Vec<Peer>, |
| 83 | - /// Each connection to another device, by its id in the settings. | |
| 84 | - pub environments: Vec<(String, ConnectionStatus)>, | |
| 87 | + /// Each connection to another device, by its id in the settings, and | |
| 88 | + /// the route it goes over. | |
| 89 | + pub environments: Vec<(String, ConnectionStatus, Option<Route>)>, | |
| 85 | 90 | } |
| 86 | 91 | |
| 87 | 92 | /// An environment the settings view edits: this device or a connected one. |
| @@ -156,7 +161,7 @@ impl Workspace { | ||
| 156 | 161 | } |
| 157 | 162 | |
| 158 | 163 | fn connect_environment(&mut self, host: RemoteHost, window: &mut Window, cx: &mut Context<Self>) { |
| 159 | - let client = Rc::new(RemoteClient::connect(&host.address, host.key.clone(), hostname())); | |
| 164 | + let client = Rc::new(RemoteClient::connect(host.ordered_routes(), host.key.clone(), hostname())); | |
| 160 | 165 | let events = client.events.clone(); |
| 161 | 166 | let id = host.id.clone(); |
| 162 | 167 | let task = cx.spawn_in(window, async move |this, cx| { |
| @@ -170,6 +175,7 @@ impl Workspace { | ||
| 170 | 175 | self.remote.environments.push(Environment { |
| 171 | 176 | host, |
| 172 | 177 | status: ConnectionStatus::Connecting, |
| 178 | + route: None, | |
| 173 | 179 | client, |
| 174 | 180 | settings: None, |
| 175 | 181 | _events: task, |
| @@ -214,9 +220,15 @@ impl Workspace { | ||
| 214 | 220 | pub fn remote_overview(&self) -> RemoteOverview { |
| 215 | 221 | RemoteOverview { |
| 216 | 222 | serving: self.remote.server.as_ref().map(|server| server.port), |
| 223 | + routes: self.remote.server.as_ref().map(RemoteServer::routes).unwrap_or_default(), | |
| 217 | 224 | error: self.remote.server_error.clone(), |
| 218 | 225 | peers: self.remote.peers.clone(), |
| 219 | - environments: self.remote.environments.iter().map(|env| (env.host.id.clone(), env.status.clone())).collect(), | |
| 226 | + environments: self | |
| 227 | + .remote | |
| 228 | + .environments | |
| 229 | + .iter() | |
| 230 | + .map(|env| (env.host.id.clone(), env.status.clone(), env.route.clone())) | |
| 231 | + .collect(), | |
| 220 | 232 | } |
| 221 | 233 | } |
| 222 | 234 | |
| @@ -782,6 +794,19 @@ impl Workspace { | ||
| 782 | 794 | let ix = self.ensure_remote_project(env, path, None); |
| 783 | 795 | self.new_thread(ix, window, cx); |
| 784 | 796 | } |
| 797 | + ClientEvent::Route(route) => { | |
| 798 | + let Some(environment) = self.remote.environments.iter_mut().find(|candidate| candidate.host.id == env) else { | |
| 799 | + return; | |
| 800 | + }; | |
| 801 | + let moved = matches!(environment.status, ConnectionStatus::Connected(_)); | |
| 802 | + environment.route = Some(route.clone()); | |
| 803 | + if moved { | |
| 804 | + // Its shells ran on the connection it left. | |
| 805 | + self.stop_remote_terminals(env, "The connection moved to a faster route; this terminal's shell ended.", cx); | |
| 806 | + } | |
| 807 | + self.move_devices_link(env, &route.address); | |
| 808 | + } | |
| 809 | + ClientEvent::Message(ServerMessage::Routes { routes }) => self.learn_routes(env, &routes), | |
| 785 | 810 | ClientEvent::Message(ServerMessage::Welcome { .. } | ServerMessage::Denied { .. }) => {} |
| 786 | 811 | ClientEvent::Message(ServerMessage::DeviceReply { request, ok, error }) => { |
| 787 | 812 | if let Some(link) = self.devices_link(env) { |
| @@ -869,6 +894,27 @@ impl Workspace { | ||
| 869 | 894 | } |
| 870 | 895 | |
| 871 | 896 | /// Stops a device's terminals, whose shells are out of reach now. |
| 897 | + /// Saves where a connected device says it listens, so this device can | |
| 898 | + /// reach it another way when the current one goes, as when one of them | |
| 899 | + /// leaves the Wi-Fi they paired over and only Tailscale is left. | |
| 900 | + fn learn_routes(&mut self, env: &str, reported: &[Route]) { | |
| 901 | + // Older hosts don't report any, and a host with no network has none | |
| 902 | + // worth forgetting the others for. | |
| 903 | + if reported.is_empty() { | |
| 904 | + return; | |
| 905 | + } | |
| 906 | + let Some(environment) = self.remote.environments.iter_mut().find(|candidate| candidate.host.id == env) else { return }; | |
| 907 | + let Some(routes) = crate::remote::merge_routes(&environment.host.routes, reported) else { return }; | |
| 908 | + // Kept in step with the settings, so the change doesn't reconnect. | |
| 909 | + environment.host.routes = routes.clone(); | |
| 910 | + environment.client.set_routes(environment.host.ordered_routes()); | |
| 911 | + settings::update(|settings| { | |
| 912 | + if let Some(host) = settings.remotes.iter_mut().find(|host| host.id == env) { | |
| 913 | + host.routes = routes; | |
| 914 | + } | |
| 915 | + }); | |
| 916 | + } | |
| 917 | + | |
| 872 | 918 | fn stop_remote_terminals(&mut self, env: &str, reason: &str, cx: &mut Context<Self>) { |
| 873 | 919 | let ids: Vec<(String, u64)> = self.remote.remote_terminals.keys().filter(|(owner, _)| owner == env).cloned().collect(); |
| 874 | 920 | for id in ids { |