fix(sync): let an object a store can't hand over stop only its own log

changelog/v0.1.0-beta4.txt

@@ -33,6 +33,7 @@ SYNC
 - Sync no longer hangs on a store that keeps answering with the same page. It stops reading and tries again at the next sync.
 - Sync stops reading an answer larger than it can be, from a Flow server, a WebDAV folder, the live channel or a device on your network, so a broken or hostile store can't use up memory.
 - A damaged or forged object from one device no longer fails the whole sync. Reading that device's changes pauses there, and the other devices' changes still arrive.
+- The same holds for an object or a file in a store's folder that is too large or that the store refuses: only that device's changes pause. A store that is out of reach, refuses your login or asks to slow down still fails the sync.
 - The same holds when a new device starts from a snapshot or a profile starts syncing: a damaged or forged object no longer fails the sync, and the profile is caught up once it can be read.
 - A snapshot or profile name from another device that carries an out-of-range number no longer stops a synced profile from being read.
 - A malformed roster entry from another device on the local network is refused, so the group stays readable on this computer.

src-tauri/src/sync2/http.rs

@@ -18,7 +18,7 @@ pub async fn read_body(mut response: Response, limit: usize) -> Result<Vec<u8>>
     let mut body = Vec::new();
     while let Some(chunk) = response.chunk().await.map_err(StoreError::Network)? {
         if body.len() + chunk.len() > limit {
-            return Err(StoreError::Malformed);
+            return Err(StoreError::TooLarge);
         }
         body.extend_from_slice(&chunk);
     }

src-tauri/src/sync2/server.rs

@@ -51,6 +51,8 @@ pub enum StoreError {
     Local(#[source] sqlx::Error),
     #[error("the server sent an invalid response")]
     Malformed,
+    #[error("the answer is larger than it may be")]
+    TooLarge,
     #[error(transparent)]
     Protocol(#[from] super::Error),
 }
@@ -65,6 +67,29 @@ impl StoreError {
 
 pub type Result<T> = std::result::Result<T, StoreError>;
 
+impl StoreError {
+    /// Whether a read failed for the one item it asked for, and not because the store is out of
+    /// reach, wants a login or asks this device to slow down.
+    fn is_about_one_item(&self) -> bool {
+        match self {
+            Self::TooLarge => true,
+            Self::Refused { status, .. } => {
+                (400..500).contains(status) && !matches!(status, 401 | 407 | 408 | 429)
+            }
+            _ => false,
+        }
+    }
+}
+
+/// What `read` found, with an item that can't be read counted as absent, so it stops only its own
+/// log (spec/store.md, Contain one bad item).
+pub fn one_item(read: Result<Option<Vec<u8>>>) -> Result<Option<Vec<u8>>> {
+    match read {
+        Err(error) if error.is_about_one_item() => Ok(None),
+        read => read,
+    }
+}
+
 /// Whether an append landed, or lost to another device's append at the same head.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Appended {
@@ -429,8 +454,10 @@ impl Store for ServerStore {
     }
     /// The server names a wrap by epoch alone: only one entry can start an epoch there.
     async fn key(&self, epoch: u32, _kc: &str, name: &str) -> Result<Option<Vec<u8>>> {
-        self.get_bytes(&format!("v1/keys/{epoch:08x}/{name}.k"))
-            .await
+        one_item(
+            self.get_bytes(&format!("v1/keys/{epoch:08x}/{name}.k"))
+                .await,
+        )
     }
     async fn put_object(&self, path: &str, bytes: Vec<u8>) -> Result<()> {
         let response = self
@@ -443,7 +470,7 @@ impl Store for ServerStore {
         decode_empty(response).await
     }
     async fn get_object(&self, path: &str) -> Result<Option<Vec<u8>>> {
-        self.get_bytes(&format!("v1/objects/{path}")).await
+        one_item(self.get_bytes(&format!("v1/objects/{path}")).await)
     }
     async fn delete_object(&self, path: &str) -> Result<bool> {
         let response = self

src-tauri/src/sync2/webdav/mod.rs

@@ -14,13 +14,14 @@ use serde_json::{Value, json};
 use tokio::sync::Mutex;
 
 pub use self::client::{WebDav, WebDavLogin};
+use super::envelope::MAX_OBJECT_SIZE;
 use super::forks::winning_chain;
 use super::mailbox::Mailbox;
 use super::paths::{chain_path, log_path, roster_path, wrap_path};
 use super::read_budget::ReadBudget;
 use super::roster::{Entry, entry_hash, parse_entry};
 use super::sealed::random_bytes;
-use super::server::{Appended, ChangedObject, Result, StoreError};
+use super::server::{Appended, ChangedObject, Result, StoreError, one_item};
 use super::store::{Store, wraps_epoch};
 use crate::sync::canonical::to_canonical_json;
 
@@ -69,16 +70,18 @@ impl WebDavStore {
             if name.strip_suffix(".r").is_none() || held.entries.contains_key(&path) {
                 continue;
             }
-            let Some(bytes) = self.client.get(&path).await? else {
-                continue;
+            let bytes = match self.client.get(&path).await {
+                Ok(None) => continue,
+                read => one_item(read)?,
             };
-            held.bytes += bytes.len();
+            // One that can't be read counts too, or each read would fetch it again without end.
+            held.bytes += bytes.as_ref().map_or(MAX_OBJECT_SIZE, Vec::len);
             if held.entries.len() >= ROSTER_FILES || held.bytes > ROSTER_BYTES {
                 return Err(StoreError::Malformed);
             }
             // A file that isn't an entry is ignored, as a losing fork candidate would be.
-            let entry = serde_json::from_slice::<Value>(&bytes)
-                .ok()
+            let entry = bytes
+                .and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok())
                 .and_then(|value| parse_entry(&value).ok());
             held.entries.insert(path, entry);
         }
@@ -222,15 +225,15 @@ impl Store for WebDavStore {
     }
 
     async fn key(&self, epoch: u32, kc: &str, name: &str) -> Result<Option<Vec<u8>>> {
-        self.client.get(&wrap_path(epoch, kc, name)).await
+        one_item(self.client.get(&wrap_path(epoch, kc, name)).await)
     }
 
     async fn put_object(&self, path: &str, bytes: Vec<u8>) -> Result<()> {
         self.client.put(path, bytes).await
     }
 
     async fn get_object(&self, path: &str) -> Result<Option<Vec<u8>>> {
-        self.client.get(path).await
+        one_item(self.client.get(path).await)
     }
 
     async fn delete_object(&self, path: &str) -> Result<bool> {

src-tauri/tests/sync2_server.rs

@@ -537,9 +537,43 @@ async fn answers_past_their_limit_are_refused() {
         store.get_object("fits").await.unwrap().unwrap().len(),
         4 << 20
     );
-    assert!(matches!(
-        store.get_object("big").await,
-        Err(StoreError::Malformed)
-    ));
-    assert!(matches!(store.changes(0).await, Err(StoreError::Malformed)));
+    assert!(matches!(store.get_object("big").await, Ok(None)));
+    assert!(store.changes(0).await.is_err());
+}
+
+#[tokio::test]
+async fn only_a_refusal_of_the_one_object_stops_only_that_object() {
+    let (server, _) = server(INSTANCE).await;
+    for status in [400, 403, 410, 413, 408, 429, 500, 503, 507] {
+        for route in [
+            format!("/v1/objects/{status}"),
+            format!("/v1/keys/00000000/{status}.k"),
+        ] {
+            Mock::given(method("GET"))
+                .and(path(route))
+                .respond_with(ResponseTemplate::new(status))
+                .mount(&server)
+                .await;
+        }
+    }
+    let store = ServerStore::connect(&format!("{}/", server.uri()), member(), None, None)
+        .await
+        .unwrap();
+
+    for status in [400, 403, 410, 413] {
+        let name = status.to_string();
+        assert!(
+            matches!(store.get_object(&name).await, Ok(None)),
+            "{status}"
+        );
+        assert!(
+            matches!(store.key(0, "x", &name).await, Ok(None)),
+            "{status}"
+        );
+    }
+    for status in [408, 429, 500, 503, 507] {
+        let name = status.to_string();
+        assert!(store.get_object(&name).await.is_err(), "{status}");
+        assert!(store.key(0, "x", &name).await.is_err(), "{status}");
+    }
 }

src-tauri/tests/sync2_webdav.rs

@@ -466,9 +466,9 @@ async fn answers_past_their_limit_are_refused() {
     let dav = webdav(&server.uri(), PASSWORD).await;
     assert!(matches!(
         dav.get("dev/x.o").await,
-        Err(StoreError::Malformed)
+        Err(StoreError::TooLarge)
     ));
-    assert!(matches!(dav.list("dev/").await, Err(StoreError::Malformed)));
+    assert!(matches!(dav.list("dev/").await, Err(StoreError::TooLarge)));
 }
 
 /// Answers each `PROPFIND` with a page from `listing`, given the path asked for, and each `GET`
@@ -557,12 +557,14 @@ fn many_roster_files(path: &str) -> String {
 
 #[tokio::test(flavor = "multi_thread")]
 async fn a_roster_folder_too_large_to_hold_is_refused() {
-    let url = scripted_webdav(many_roster_files, " ".repeat(4 << 20)).await;
-    let store = store_for(&url, &random_bytes()).await;
-    assert!(matches!(
-        store.roster_after(-1).await,
-        Err(StoreError::Malformed)
-    ));
+    for size in [4 << 20, (4 << 20) + 1] {
+        let url = scripted_webdav(many_roster_files, " ".repeat(size)).await;
+        let store = store_for(&url, &random_bytes()).await;
+        assert!(matches!(
+            store.roster_after(-1).await,
+            Err(StoreError::Malformed)
+        ));
+    }
 }
 
 /// A member's snapshot can place a log at its last possible `seq`; listing past it must hold.
@@ -886,3 +888,104 @@ async fn a_leave_calls_for_no_new_login() {
         .unwrap();
     assert!(!login_due(&pool_b).await.unwrap());
 }
+
+/// One more object than `device` wrote, past the size any object may have, at the end of its log.
+async fn too_large_at_the_end(url: &str, device: &MemberSecrets, wrote: i64) {
+    let seq = u64::try_from(wrote + 1).unwrap();
+    webdav(url, PASSWORD)
+        .await
+        .put(
+            &log_path(&hex::encode(device.did), seq),
+            vec![0; (4 << 20) + 1],
+        )
+        .await
+        .unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn an_object_past_the_size_limit_stops_only_its_own_log() {
+    let g = group().await;
+    let (url, _) = webdav_server().await;
+    onto_webdav(&g, &url).await;
+    let pool_a = device_pool(&g).await;
+    let mut a = Session::open(&pool_a, store_for(&url, &g.a.did).await, &g.a)
+        .await
+        .unwrap();
+    watch(&pool_a, "aaaaaaaaaaa").await;
+    a.push(&pool_a, exported(&pool_a, "app-a").await)
+        .await
+        .unwrap();
+    too_large_at_the_end(&url, &g.a, a.record().own_seq).await;
+
+    let pool_b = device_pool(&g).await;
+    let mut b = Session::open(&pool_b, store_for(&url, &g.b.did).await, &g.b)
+        .await
+        .unwrap();
+    let pulled = b.pull(&pool_b).await.unwrap();
+
+    let sender = hex::encode(g.a.did);
+    assert!(pulled.batches.contains_key(&sender));
+    b.finish(&pool_b, pulled).await.unwrap();
+    let read = streams(&pool_b).await.unwrap();
+    assert_eq!(read[&sender].seq, a.record().own_seq.unsigned_abs());
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn a_roster_file_past_the_size_limit_is_ignored() {
+    let g = group().await;
+    let (url, _) = webdav_server().await;
+    onto_webdav(&g, &url).await;
+    webdav(&url, PASSWORD)
+        .await
+        .put(
+            "roster/ffffffffffffffff-ffffffffffffffff.r",
+            vec![b' '; (4 << 20) + 1],
+        )
+        .await
+        .unwrap();
+    let pool = device_pool(&g).await;
+
+    let session = Session::open(&pool, store_for(&url, &g.a.did).await, &g.a)
+        .await
+        .unwrap();
+
+    assert_eq!(session.entries().len(), 2);
+}
+
+#[tokio::test]
+async fn only_a_refusal_of_the_one_object_stops_only_that_object() {
+    let answer = |status: u16| async move {
+        let server = MockServer::start().await;
+        Mock::given(method("GET"))
+            .respond_with(ResponseTemplate::new(status))
+            .mount(&server)
+            .await;
+        let store = store_for(&format!("{}/", server.uri()), &random_bytes()).await;
+        let read = store.get_object("dev/x/log/o").await;
+        let wrap = store.key(0, "kc", "name").await;
+        (read, wrap)
+    };
+    for status in [400, 403, 410, 413, 451] {
+        let (read, wrap) = answer(status).await;
+        assert!(matches!(read, Ok(None)), "{status}");
+        assert!(matches!(wrap, Ok(None)), "{status}");
+    }
+    for status in [401, 407, 408, 429, 500, 502, 503, 507] {
+        let (read, wrap) = answer(status).await;
+        assert!(read.is_err(), "{status}");
+        assert!(wrap.is_err(), "{status}");
+    }
+}
+
+#[tokio::test]
+async fn an_unreachable_folder_still_fails_the_read() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    drop(listener);
+    let store = store_for(&format!("http://127.0.0.1:{port}/"), &random_bytes()).await;
+
+    assert!(matches!(
+        store.get_object("dev/x/log/o").await,
+        Err(StoreError::Network(_))
+    ));
+}