diff --git a/changelog/v0.1.0-beta4.txt b/changelog/v0.1.0-beta4.txt index 65199c72..b5c5a9e0 100644 --- a/changelog/v0.1.0-beta4.txt +++ b/changelog/v0.1.0-beta4.txt @@ -47,6 +47,7 @@ SYNC - Linked devices only ever touch files inside the group's own WebDAV folder. - One device's unreadable change no longer keeps other devices' changes from arriving, however many are waiting. - When the store refuses something this computer wrote, Linked devices keep receiving changes, retry it in order and show the error, instead of stopping the sync. +- On the local network, a connection that stays silent is dropped after a few seconds, and it no longer stops this computer from meeting your other devices or answering them. - Linked devices sign in to your Flow server about once every 15 minutes instead of before every sync. MUSIC diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 406426e8..94aeb1fc 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -154,6 +154,7 @@ codegen-units = 16 [dev-dependencies] +tokio = { version = "1", features = ["test-util"] } wiremock = "0.6" # In-process WebDAV server for the FLOW-SYNC/2 WebDAV store tests. dav-server = { version = "0.11", default-features = false, features = ["memfs"] } diff --git a/src-tauri/src/sync2/lan_link.rs b/src-tauri/src/sync2/lan_link.rs index 770fdbb1..4c2f9c60 100644 --- a/src-tauri/src/sync2/lan_link.rs +++ b/src-tauri/src/sync2/lan_link.rs @@ -11,6 +11,7 @@ use serde::{Deserialize, Deserializer, Serialize}; use serde_json::{Value, json}; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::sync::{mpsc, oneshot}; +use tokio::time::timeout; use tokio_tungstenite::tungstenite::protocol::WebSocketConfig; use super::curve::{sign, verify}; @@ -38,6 +39,10 @@ const MAX_FRAME: usize = MAX_OBJECT_SIZE + MAX_OBJECT_SIZE / 16; const NONCE_SIZE: usize = 32; /// Once this side has no more requests, the session closes when the peer stops sending. pub const IDLE: Duration = Duration::from_secs(3); +/// How long a peer that hasn't proved itself gets for each hello message. +pub const HELLO_WITHIN: Duration = Duration::from_secs(3); +/// How long a member gets to answer one request, which can carry a full object. +const ASK_WITHIN: Duration = Duration::from_secs(30); #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "op", rename_all = "lowercase")] @@ -238,9 +243,9 @@ where .send_binary(hello_frame(epoch, &lan_key, 1, &hello)?) .await?; - let reply = socket - .recv_binary() + let reply = timeout(HELLO_WITHIN, socket.recv_binary()) .await + .map_err(|_| EngineError::TimedOut)? .map_err(|_| EngineError::LanUnreachable)?; if frame_epoch(&reply)? != epoch { return Err(EngineError::Unexpected); @@ -289,10 +294,11 @@ where Ok((framed, peer)) } -/// The advertising side: answers a hello under any epoch this device has a key for. -/// The advertising side; returns the peer and the port it listens on, when it said. +/// The advertising side: answers `first`, the peer's first message, under any epoch this device +/// has a key for. Returns the peer and the port it listens on, when it said. pub async fn hello_as_server( mut socket: WsChannel, + first: &[u8], session: &mut Session<'_, T>, device: &MemberSecrets, ) -> Result<(Framed, String, Option)> @@ -300,14 +306,13 @@ where S: AsyncRead + AsyncWrite + Unpin + Send, T: Store, { - let first = socket.recv_binary().await?; - let epoch = frame_epoch(&first)?; + let epoch = frame_epoch(first)?; if epoch > session.roster().epoch { return Err(EngineError::Unexpected); } let epoch_key = session.epoch_key(epoch).await?; let lan_key = purpose_key(&epoch_key, session.group(), epoch, Purpose::Lan); - let hello = open_hello(&first, &lan_key, 1)?; + let hello = open_hello(first, &lan_key, 1)?; let peer = hello["did"] .as_str() .ok_or(EngineError::Unexpected)? @@ -334,7 +339,9 @@ where .send_binary(hello_frame(epoch, &lan_key, 2, &reply)?) .await?; - let finish = socket.recv_binary().await?; + let finish = timeout(HELLO_WITHIN, socket.recv_binary()) + .await + .map_err(|_| EngineError::TimedOut)??; let finish = open_hello(&finish, &lan_key, 3)?; let signed = lan_transcript( LanRole::Client, @@ -528,7 +535,10 @@ impl Link { .send((request, Vec::new(), reply)) .await .map_err(|_| EngineError::LanUnreachable)?; - answer.await.map_err(|_| EngineError::LanUnreachable)? + timeout(ASK_WITHIN, answer) + .await + .map_err(|_| EngineError::TimedOut)? + .map_err(|_| EngineError::LanUnreachable)? } /// Sends a live envelope, which gets no answer. @@ -615,6 +625,27 @@ mod tests { }) } + #[tokio::test(start_paused = true)] + async fn a_request_the_peer_never_answers_ends() { + let (a, _silent) = duplex(1 << 16); + let framed = Framed { + socket: WsChannel::new(WebSocketStream::from_raw_socket(a, Role::Client, None).await), + send_key: vec![1; 32], + receive_key: vec![2; 32], + sent: 0, + received: 0, + }; + let pool = sqlx::SqlitePool::connect_lazy("sqlite::memory:").unwrap(); + let holder = Holder { + store: HeldStore::new(pool), + group: [0; 16], + senders: BTreeMap::new(), + live: Arc::new(|_| {}), + }; + let link = Link::spawn(framed, holder); + assert!(matches!(link.roster(-1).await, Err(EngineError::TimedOut))); + } + #[test] fn any_change_to_a_session_frame_is_refused() { let (frame, receiver) = sealed_frame(); diff --git a/src-tauri/src/sync2/lan_runtime.rs b/src-tauri/src/sync2/lan_runtime.rs index 34f227e8..63ca0efd 100644 --- a/src-tauri/src/sync2/lan_runtime.rs +++ b/src-tauri/src/sync2/lan_runtime.rs @@ -7,6 +7,8 @@ use std::net::IpAddr; use std::sync::Arc; use std::time::Duration; +use futures_util::StreamExt as _; +use futures_util::stream; use mdns_sd::{DaemonEvent, Receiver, ResolvedService, ServiceDaemon, ServiceEvent, ServiceInfo}; use tauri::AppHandle; use tokio::task::JoinSet; @@ -37,6 +39,8 @@ const MEET_EVERY_MS: i64 = 5 * 60 * 1000; const DIAL_EVERY_MS: i64 = 60 * 1000; /// How long a device that leaves a LAN group spends passing its `leave` on. const LEAVE_WITHIN: Duration = Duration::from_secs(10); +/// Members dialed at once in a round, so one that stays silent doesn't hold up the others. +const DIALS_AT_ONCE: usize = 8; /// The group this device meets members of on the LAN, a LAN group or a server group: this /// hour's tag with the ones before and after it, and the group's other devices. @@ -106,44 +110,109 @@ impl Dialer<'_> { } } - /// Meets every member found over mDNS under one of `tags`, every few minutes. - async fn meet_found(&self, found: &mut BTreeMap, tags: &[String; 3]) { - for member in found.values_mut().filter(|f| tags.contains(&f.tag)) { - if now_ms() - member.met_ms < MEET_EVERY_MS { - continue; - } - member.met_ms = now_ms(); - for address in &member.addresses { - if self.meet((address, member.port), &[]).await { - break; + /// Dials a round's members, those found over mDNS first, and then the rest not met by then. + async fn meet_round(&self, round: Round) { + stream::iter(round.found) + .for_each_concurrent(DIALS_AT_ONCE, |(addresses, port)| async move { + for address in &addresses { + if self.meet((address, port), &[]).await { + break; + } } - } - } + }) + .await; + let (seen, since) = (&met(self.app), round.since); + stream::iter(round.kept) + .for_each_concurrent(DIALS_AT_ONCE, |(did, (address, port))| async move { + if seen.get(&did).is_none_or(|at| *at < since) + && !self.meet((&address, port), &[]).await + { + unreached(self.app, &did); + } + }) + .await; } +} - /// Dials the members not met since `since` at the address kept for each, at most once a - /// minute each: mDNS may never find them on a network that drops multicast. - async fn meet_kept( - &self, - members: &BTreeSet, +/// The members one round dials. It runs beside the loop that advertises and browses, so a member +/// that is slow to answer holds up neither. +struct Round { + /// Addresses and port of each member found over mDNS. + found: Vec<(Vec, u16)>, + /// Each member dialed at the address kept for it. + kept: Vec<(String, (String, u16))>, + since: i64, +} + +impl Round { + fn due( + app: &AppHandle, + meeting: &Meeting, + found: &mut BTreeMap, tried: &mut BTreeMap, since: i64, - ) { - let seen = met(self.app); - for (did, (address, port)) in addresses(self.app) { - let now = now_ms(); - if !members.contains(&did) - || seen.get(&did).is_some_and(|at| *at >= since) - || tried.get(&did).is_some_and(|at| now - at < DIAL_EVERY_MS) - { - continue; - } - tried.insert(did.clone(), now); - if !self.meet((&address, port), &[]).await { - unreached(self.app, &did); - } + ) -> Self { + Self { + found: due_found(found, &meeting.tags), + kept: due_kept(app, &meeting.members, tried, since), + since, } } + + fn is_empty(&self) -> bool { + self.found.is_empty() && self.kept.is_empty() + } +} + +fn start_round(rounds: &mut JoinSet<()>, app: &AppHandle, local: &Local, port: u16, round: Round) { + if round.is_empty() { + return; + } + let (app, local) = (app.clone(), local.clone()); + rounds.spawn(async move { + let dialer = Dialer { + app: &app, + local: &local, + port, + }; + dialer.meet_round(round).await; + }); +} + +/// Members found over mDNS under one of `tags` and due, every few minutes. +fn due_found(found: &mut BTreeMap, tags: &[String; 3]) -> Vec<(Vec, u16)> { + let now = now_ms(); + found + .values_mut() + .filter(|member| tags.contains(&member.tag) && now - member.met_ms >= MEET_EVERY_MS) + .map(|member| { + member.met_ms = now; + (member.addresses.clone(), member.port) + }) + .collect() +} + +/// Members not met since `since`, at the address kept for each, at most once a minute each: mDNS +/// may never find them on a network that drops multicast. +fn due_kept( + app: &AppHandle, + members: &BTreeSet, + tried: &mut BTreeMap, + since: i64, +) -> Vec<(String, (String, u16))> { + let (seen, now) = (met(app), now_ms()); + addresses(app) + .into_iter() + .filter(|(did, _)| { + let due = members.contains(did) + && seen.get(did).is_none_or(|at| *at < since) + && tried.get(did).is_none_or(|at| now - at >= DIAL_EVERY_MS); + if due { + tried.insert(did.clone(), now); + } + due + }) + .collect() } /// Meets every member this device can reach, so they pull the `leave` that `session` just @@ -327,11 +396,8 @@ async fn run(app: &AppHandle, daemon: Option<&ServiceDaemon>) -> Result<()> { live_sink(app), )); - let dialer = Dialer { - app, - local: &local, - port, - }; + // Dropped with the run too, and at most one at a time. + let mut rounds = JoinSet::new(); let instance = format!("Flow-{}", hex::encode(random_bytes::<2>())); let mut advertised: Option<(String, String)> = None; let browse = || daemon.and_then(|daemon| daemon.browse(SERVICE_TYPE).ok()); @@ -343,6 +409,8 @@ async fn run(app: &AppHandle, daemon: Option<&ServiceDaemon>) -> Result<()> { let mut timer = tokio::time::interval(CHECK_EVERY); loop { let event = tokio::select! { + // What it met shows on the page, and members it didn't reach may be due again. + Some(_) = rounds.join_next(), if !rounds.is_empty() => None, _ = timer.tick() => { if renewed(&local.device) { break; @@ -392,10 +460,12 @@ async fn run(app: &AppHandle, daemon: Option<&ServiceDaemon>) -> Result<()> { { record(&mut found, &info); } - if let Some(meeting) = &meeting { - dialer.meet_found(&mut found, &meeting.tags).await; + if let Some(meeting) = &meeting + && rounds.is_empty() + { let since = (now_ms() - MEET_EVERY_MS).max(network_changed_at); - dialer.meet_kept(&meeting.members, &mut tried, since).await; + let round = Round::due(app, meeting, &mut found, &mut tried, since); + start_round(&mut rounds, app, &local, port, round); } save_addresses(app, &local.pool, &mut saved).await; follow_nearby(app, &mut near).await; diff --git a/src-tauri/src/sync2/lan_sync.rs b/src-tauri/src/sync2/lan_sync.rs index 7946f085..62030b46 100644 --- a/src-tauri/src/sync2/lan_sync.rs +++ b/src-tauri/src/sync2/lan_sync.rs @@ -2,10 +2,13 @@ //! keys and the objects it lacks from the other, and holds them to serve in turn. use std::collections::BTreeSet; +use std::sync::Arc; use std::time::Duration; use sqlx::SqlitePool; use tokio::net::TcpListener; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; +use tokio::time::timeout; use tokio_tungstenite::{accept_async_with_config, connect_async_with_config}; use super::engine::{EngineError, Result, Session, verify_logged}; @@ -14,7 +17,7 @@ use super::held::HeldStore; use super::kdf::sha256; use super::keys::MemberSecrets; use super::lan_link::{ - Framed, Holder, Link, LiveSink, SESSION_PATH, hello_as_client, hello_as_server, + Framed, HELLO_WITHIN, Holder, Link, LiveSink, SESSION_PATH, hello_as_client, hello_as_server, lan_socket_config, }; use super::paths::parse_log_path; @@ -108,6 +111,13 @@ const CONNECT_WITHIN: Duration = Duration::from_secs(3); /// How many epochs to try a member under, one lower each time (spec/live.md, Session). const DIAL_ATTEMPTS: u32 = 4; +/// Connections that haven't proved themselves yet, beyond which new ones are closed at once, so +/// they can't use up this process's descriptors. +const HELLOS_AT_ONCE: usize = 16; + +/// How soon to accept again after a failure such as running out of descriptors. +const ACCEPT_AGAIN: Duration = Duration::from_millis(100); + /// A session that ended: the member met, how many objects this device took from it, and where /// the member can be reached for the next one, when known. pub struct Met { @@ -202,10 +212,18 @@ pub async fn serve( met: impl Fn(Met) + Clone + Send + 'static, sink: LiveSink, ) { - while let Ok((stream, _)) = listener.accept().await { + let hellos = Arc::new(Semaphore::new(HELLOS_AT_ONCE)); + loop { + let Ok((stream, _)) = listener.accept().await else { + tokio::time::sleep(ACCEPT_AGAIN).await; + continue; + }; + let Ok(hello) = Arc::clone(&hellos).try_acquire_owned() else { + continue; + }; let (pool, device, met, sink) = (pool.clone(), device.clone(), met.clone(), sink.clone()); tokio::spawn(async move { - match answer(stream, &pool, &device, sink).await { + match answer(stream, hello, &pool, &device, sink).await { Ok(session) => met(session), Err(error) => { tracing::debug!(target: "flow::sync2", kind = error.kind(), "LAN session ended"); @@ -217,18 +235,28 @@ pub async fn serve( async fn answer( stream: tokio::net::TcpStream, + hello: OwnedSemaphorePermit, pool: &SqlitePool, device: &MemberSecrets, sink: LiveSink, ) -> Result { let address = stream.peer_addr().ok().map(|peer| peer.ip().to_string()); - let socket = accept_async_with_config(stream, Some(lan_socket_config())) + let first_message = async { + let socket = accept_async_with_config(stream, Some(lan_socket_config())) + .await + .map_err(|_| EngineError::LanUnreachable)?; + let mut socket = WsChannel::new(socket); + let first = socket.recv_binary().await?; + Ok::<_, EngineError>((socket, first)) + }; + let (socket, first) = timeout(HELLO_WITHIN, first_message) .await - .map_err(|_| EngineError::LanUnreachable)?; + .map_err(|_| EngineError::TimedOut)??; + // Opening a session replays the roster, so a connection that says nothing doesn't cost that. let held = HeldStore::new(pool.clone()); let mut session = Session::open_for_lan(pool, held.clone(), device).await?; - let (framed, peer, listening) = - hello_as_server(WsChannel::new(socket), &mut session, device).await?; + let (framed, peer, listening) = hello_as_server(socket, &first, &mut session, device).await?; + drop(hello); let link = Link::spawn(framed, Holder::of(held.clone(), &session, sink)); let taken = pull_from(&held, &link, session.group()).await?; Ok(Met { diff --git a/src-tauri/tests/sync2_lan.rs b/src-tauri/tests/sync2_lan.rs index 726505c8..31afb577 100644 --- a/src-tauri/tests/sync2_lan.rs +++ b/src-tauri/tests/sync2_lan.rs @@ -41,9 +41,10 @@ use flow_desktop_lib::sync2::wrap::wrap_epoch_key; use futures_util::{SinkExt, StreamExt}; use sqlx::SqlitePool; use sync2_common::{Group, LINK, draft, empty_pool, exported, group, head, watch}; -use tokio::net::TcpListener; -use tokio_tungstenite::connect_async; +use tokio::io::AsyncReadExt as _; +use tokio::net::{TcpListener, TcpStream}; use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{accept_async, connect_async}; /// A device that holds the group as pairing leaves it: the roster and the wraps, no objects. async fn holding(g: &Group) -> SqlitePool { @@ -752,3 +753,96 @@ async fn a_dial_tries_lower_epochs_and_stops_after_four_attempts() { assert!(matches!(met.err(), Some(EngineError::LanUnreachable))); assert_eq!(attempts.await.unwrap(), [5, 4, 3, 2]); } + +/// Whether the other side has closed `stream`, once what it sent is read. +async fn closed_within(stream: &mut TcpStream, limit: Duration) -> bool { + let mut bytes = [0; 1024]; + tokio::time::timeout(limit, async { + while !matches!(stream.read(&mut bytes).await, Ok(0) | Err(_)) {} + }) + .await + .is_ok() +} + +#[tokio::test] +async fn connections_that_say_nothing_are_closed_and_members_still_meet() { + let g = group().await; + let (pool_a, pool_b) = (holding(&g).await, holding(&g).await); + let port = listening(&pool_a, &g.a).await; + let mut silent = Vec::new(); + for _ in 0..64 { + silent.push(TcpStream::connect(("127.0.0.1", port)).await.unwrap()); + } + let last = silent.last_mut().unwrap(); + assert!( + closed_within(last, Duration::from_secs(1)).await, + "over the limit, a connection is closed at once" + ); + for stream in &mut silent { + assert!(closed_within(stream, Duration::from_secs(5)).await); + } + meet(&pool_b, &g.b, ("127.0.0.1", port), None, &[], no_live()) + .await + .unwrap(); +} + +#[tokio::test] +async fn a_websocket_that_says_nothing_is_closed() { + let g = group().await; + let pool = holding(&g).await; + let port = listening(&pool, &g.a).await; + let (mut socket, _) = connect_async(format!("ws://127.0.0.1:{port}/flow-sync/2")) + .await + .unwrap(); + let ended = tokio::time::timeout(Duration::from_secs(5), async { + while let Some(Ok(_)) = socket.next().await {} + }) + .await; + assert!(ended.is_ok()); +} + +#[tokio::test] +async fn meeting_a_peer_that_stays_silent_after_connecting_ends() { + let g = group().await; + let pool = holding(&g).await; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let _socket = accept_async(stream).await.unwrap(); + std::future::pending::<()>().await; + }); + let met = tokio::time::timeout( + Duration::from_secs(5), + meet(&pool, &g.b, ("127.0.0.1", port), None, &[], no_live()), + ) + .await; + assert!(matches!(met, Ok(Err(EngineError::TimedOut)))); +} + +/// Run under a low descriptor limit, as a macOS GUI app has by default: +/// `bash -c 'ulimit -n 256; cargo test --test sync2_lan -- --ignored --exact '`. +#[tokio::test] +#[ignore = "needs ulimit -n 256"] +async fn the_service_keeps_accepting_after_running_out_of_descriptors() { + let g = group().await; + let pool = holding(&g).await; + let port = listening(&pool, &g.a).await; + let mut files = Vec::new(); + while let Ok(file) = std::fs::File::open("/dev/null") { + files.push(file); + } + files.truncate(files.len() - 3); + let mut streams = Vec::new(); + for _ in 0..3 { + if let Ok(stream) = TcpStream::connect(("127.0.0.1", port)).await { + streams.push(stream); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + drop(streams); + drop(files); + tokio::time::sleep(Duration::from_millis(200)).await; + let mut again = TcpStream::connect(("127.0.0.1", port)).await.unwrap(); + assert!(closed_within(&mut again, Duration::from_secs(5)).await); +}