fix(sync): limit how long LAN connections and requests can wait

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

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"] }

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,25 +294,25 @@ 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<S, T>(
     mut socket: WsChannel<S>,
+    first: &[u8],
     session: &mut Session<'_, T>,
     device: &MemberSecrets,
 ) -> Result<(Framed<S>, String, Option<u16>)>
 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();

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<String, Found>, 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<String>,
+/// 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<String>, 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<String, Found>,
         tried: &mut BTreeMap<String, i64>,
         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<String, Found>, tags: &[String; 3]) -> Vec<(Vec<String>, 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<String>,
+    tried: &mut BTreeMap<String, i64>,
+    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;