diff --git a/changelog/v0.1.0-beta4.txt b/changelog/v0.1.0-beta4.txt index 6a4050a3..d1184e6d 100644 --- a/changelog/v0.1.0-beta4.txt +++ b/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 diff --git a/src-tauri/src/sync2/engine.rs b/src-tauri/src/sync2/engine.rs index c10ad4e5..135549dd 100644 --- a/src-tauri/src/sync2/engine.rs +++ b/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 = 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; diff --git a/src-tauri/tests/sync2_engine.rs b/src-tauri/tests/sync2_engine.rs index 336a6ccf..6a9b4555 100644 --- a/src-tauri/tests/sync2_engine.rs +++ b/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> { + self.0.roster_after(after).await + } + async fn append_roster( + &self, + head: &str, + entry: &Entry, + wraps: &BTreeMap, + chain: Option<&str>, + ) -> Result { + self.0.append_roster(head, entry, wraps, chain).await + } + async fn key(&self, epoch: u32, kc: &str, name: &str) -> Result>> { + self.0.key(epoch, kc, name).await + } + async fn put_object(&self, path: &str, bytes: Vec) -> Result<()> { + self.0.put_object(path, bytes).await + } + async fn get_object(&self, path: &str) -> Result>> { + self.0.get_object(path).await + } + async fn delete_object(&self, path: &str) -> Result { + self.0.delete_object(path).await + } + async fn changes(&self, cursor: u64) -> Result<(Vec, 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) { + 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);