fix(sync): keep reading other devices past one that stopped

changelog/v0.1.0-beta4.txt

@@ -45,6 +45,7 @@ SYNC
 - Linked devices: allowing or declining remote control now takes effect whichever profile is open.
 - Linked devices reach a Flow server or WebDAV folder on your local network directly, even when a system proxy is set.
 - 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.
 - Linked devices sign in to your Flow server about once every 15 minutes instead of before every sync.
 
 MUSIC

src-tauri/src/sync2/engine.rs

@@ -767,8 +767,11 @@ impl<'a, S: Store> Session<'a, S> {
         let mut floors = floors(pool).await?;
         floors.extend(pulled.floors.clone());
         let mut budget = ReadBudget::default();
+        let mut cursor = u64::try_from(pulled.cursor).unwrap_or(0);
+        // A sender that stopped stays stopped for this read, and the others go on. The saved cursor
+        // stays at the first page that held one back, so the next sync lists it again.
+        let mut stopped = BTreeSet::new();
         loop {
-            let cursor = u64::try_from(pulled.cursor).unwrap_or(0);
             let read: BTreeMap<String, u64> = pulled
                 .positions
                 .iter()
@@ -786,7 +789,8 @@ impl<'a, S: Store> Session<'a, S> {
                     parse_log_path(&change.path).map(|(did, seq)| (did, seq, change.path))
                 })
                 .filter(|(did, seq, _)| {
-                    *did != me && pulled.positions.get(did).is_none_or(|p| p.seq < *seq)
+                    let at = pulled.positions.get(did).map_or(0, |p| p.seq);
+                    *did != me && at < *seq && (!stopped.contains(did) || at + 1 == *seq)
                 })
                 .partition(|(did, seq, _)| self.past_its_cut(did, *seq));
             pulled
@@ -811,11 +815,9 @@ impl<'a, S: Store> Session<'a, S> {
                     break;
                 }
             }
-            // A missing object means the store hasn't caught up: read this page again next time.
-            if !waiting.is_empty() {
-                pulled.stalled = true;
-                return Ok(pulled);
-            }
+            // A missing object means the store hasn't caught up: this page is read again next time.
+            stopped.extend(waiting.into_iter().map(|(did, _, _)| did));
+            pulled.stalled |= !stopped.is_empty();
             let advanced = next > cursor
                 || pulled
                     .positions
@@ -824,7 +826,10 @@ impl<'a, S: Store> Session<'a, S> {
             if !advanced {
                 return Ok(pulled);
             }
-            pulled.cursor = i64::try_from(next).map_err(|_| Error::Length)?;
+            cursor = next;
+            if stopped.is_empty() {
+                pulled.cursor = i64::try_from(next).map_err(|_| Error::Length)?;
+            }
             // The next sync goes on from here, and a restored device doesn't choose its link yet.
             if !budget.take(listed) {
                 pulled.stalled = true;

src-tauri/tests/sync2_engine.rs

@@ -249,6 +249,116 @@ async fn a_removed_device_stops_and_a_new_device_reads_the_old_epoch() {
     assert_eq!(watched(&pool_c).await, ["aaaaaaaaaaa"]);
 }
 
+/// Answers `changes` with one object per page, as a server does past its page limit.
+struct OnePerPage<'a>(&'a MemoryStore);
+
+impl Store for OnePerPage<'_> {
+    async fn roster_after(&self, after: i64) -> Result<Vec<Entry>> {
+        self.0.roster_after(after).await
+    }
+    async fn append_roster(
+        &self,
+        head: &str,
+        entry: &Entry,
+        wraps: &BTreeMap<String, String>,
+        chain: Option<&str>,
+    ) -> Result<Appended> {
+        self.0.append_roster(head, entry, wraps, chain).await
+    }
+    async fn key(&self, epoch: u32, kc: &str, name: &str) -> Result<Option<Vec<u8>>> {
+        self.0.key(epoch, kc, name).await
+    }
+    async fn put_object(&self, path: &str, bytes: Vec<u8>) -> Result<()> {
+        self.0.put_object(path, bytes).await
+    }
+    async fn get_object(&self, path: &str) -> Result<Option<Vec<u8>>> {
+        self.0.get_object(path).await
+    }
+    async fn delete_object(&self, path: &str) -> Result<bool> {
+        self.0.delete_object(path).await
+    }
+    async fn changes(&self, cursor: u64) -> Result<(Vec<ChangedObject>, u64)> {
+        let memory = self.0.0.lock().unwrap();
+        let first = memory
+            .objects
+            .iter()
+            .filter(|(_, (at, _))| *at > cursor)
+            .min_by_key(|(_, (at, _))| *at);
+        Ok(first.map_or((Vec::new(), cursor), |(path, (at, bytes))| {
+            let size = bytes.len() as u64;
+            (
+                vec![ChangedObject {
+                    path: path.clone(),
+                    size,
+                }],
+                *at,
+            )
+        }))
+    }
+}
+
+/// C's first object arrived before anything of A's. Returns the path and C's genuine bytes.
+async fn a_tampered_object_before_another_sender(
+    g: &Group,
+    c: &MemberSecrets,
+    pool_a: &SqlitePool,
+    pool_c: &SqlitePool,
+) -> (String, Vec<u8>) {
+    watch(pool_c, "ccccccccccc").await;
+    sync(pool_c, &g.store, c, "app-c").await.unwrap();
+    let path = log_path(&hex::encode(c.did), 1);
+    let genuine = g.store.tamper(&path);
+    watch(pool_a, "aaaaaaaaaaa").await;
+    sync(pool_a, &g.store, &g.a, "app-a").await.unwrap();
+    (path, genuine)
+}
+
+#[tokio::test]
+async fn a_sender_that_stopped_does_not_hide_others_on_later_pages() {
+    let g = group().await;
+    let c = add_device(&g, 0, &g.epoch_key).await;
+    let (pool_a, pool_b, pool_c) = (
+        device_pool(&g).await,
+        device_pool(&g).await,
+        device_pool(&g).await,
+    );
+    let (path, genuine) = a_tampered_object_before_another_sender(&g, &c, &pool_a, &pool_c).await;
+
+    sync(&pool_b, OnePerPage(&g.store), &g.b, "app-b")
+        .await
+        .unwrap();
+    assert_eq!(watched(&pool_b).await, ["aaaaaaaaaaa"]);
+
+    // Same place in the listing: the next sync has to list that page again.
+    g.store.0.lock().unwrap().objects.get_mut(&path).unwrap().1 = genuine;
+    sync(&pool_b, OnePerPage(&g.store), &g.b, "app-b")
+        .await
+        .unwrap();
+    assert_eq!(watched(&pool_b).await, ["aaaaaaaaaaa", "ccccccccccc"]);
+}
+
+#[tokio::test]
+async fn a_log_listed_backwards_across_pages_is_still_read() {
+    let g = group().await;
+    let (pool_a, pool_b) = (device_pool(&g).await, device_pool(&g).await);
+    watch(&pool_a, "aaaaaaaaaaa").await;
+    sync(&pool_a, &g.store, &g.a, "app-a").await.unwrap();
+    watch(&pool_a, "ccccccccccc").await;
+    sync(&pool_a, &g.store, &g.a, "app-a").await.unwrap();
+
+    let first = log_path(&hex::encode(g.a.did), 1);
+    let bytes = g.store.take(&first);
+    g.store.put(&first, bytes);
+
+    sync(&pool_b, OnePerPage(&g.store), &g.b, "app-b")
+        .await
+        .unwrap();
+    sync(&pool_b, OnePerPage(&g.store), &g.b, "app-b")
+        .await
+        .unwrap();
+    assert_eq!(watched(&pool_b).await, ["aaaaaaaaaaa", "ccccccccccc"]);
+}
+
 /// A store whose change cursor never moves: it hands out every object on each call.
 struct Repeating<'a>(&'a MemoryStore);