diff --git a/changelog/v0.1.0-beta4.txt b/changelog/v0.1.0-beta4.txt index b5c5a9e0..5c0dd848 100644 --- a/changelog/v0.1.0-beta4.txt +++ b/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. diff --git a/src-tauri/src/sync2/http.rs b/src-tauri/src/sync2/http.rs index 76dac961..208ae363 100644 --- a/src-tauri/src/sync2/http.rs +++ b/src-tauri/src/sync2/http.rs @@ -18,7 +18,7 @@ pub async fn read_body(mut response: Response, limit: usize) -> Result> 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); } diff --git a/src-tauri/src/sync2/server.rs b/src-tauri/src/sync2/server.rs index 895d8cb1..59e58c7d 100644 --- a/src-tauri/src/sync2/server.rs +++ b/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 = std::result::Result; +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>>) -> Result>> { + 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>> { - 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) -> Result<()> { let response = self @@ -443,7 +470,7 @@ impl Store for ServerStore { decode_empty(response).await } async fn get_object(&self, path: &str) -> Result>> { - 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 { let response = self diff --git a/src-tauri/src/sync2/webdav/mod.rs b/src-tauri/src/sync2/webdav/mod.rs index df9e5bf2..3ccfb6b2 100644 --- a/src-tauri/src/sync2/webdav/mod.rs +++ b/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::(&bytes) - .ok() + let entry = bytes + .and_then(|bytes| serde_json::from_slice::(&bytes).ok()) .and_then(|value| parse_entry(&value).ok()); held.entries.insert(path, entry); } @@ -222,7 +225,7 @@ impl Store for WebDavStore { } async fn key(&self, epoch: u32, kc: &str, name: &str) -> Result>> { - 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) -> Result<()> { @@ -230,7 +233,7 @@ impl Store for WebDavStore { } async fn get_object(&self, path: &str) -> Result>> { - self.client.get(path).await + one_item(self.client.get(path).await) } async fn delete_object(&self, path: &str) -> Result { diff --git a/src-tauri/tests/sync2_server.rs b/src-tauri/tests/sync2_server.rs index a6c384b9..48700030 100644 --- a/src-tauri/tests/sync2_server.rs +++ b/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}"); + } } diff --git a/src-tauri/tests/sync2_webdav.rs b/src-tauri/tests/sync2_webdav.rs index cc67507e..ef789094 100644 --- a/src-tauri/tests/sync2_webdav.rs +++ b/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(_)) + )); +}