From c71fca026b650fcc1085450d2b3cc3149c5ea0f7 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:05:31 +0200 Subject: [PATCH] fix(sync): keep reading when the store refuses an upload --- changelog/v0.1.0-beta4.txt | 1 + src-tauri/src/sync2/engine.rs | 4 ++ src-tauri/src/sync2/hybrid.rs | 30 ++++++--- src-tauri/src/sync2/server.rs | 8 +++ src-tauri/src/sync2/service.rs | 35 +++++++--- src-tauri/tests/sync2_hybrid.rs | 112 ++++++++++++++++++++++++++++++++ 6 files changed, 172 insertions(+), 18 deletions(-) diff --git a/changelog/v0.1.0-beta4.txt b/changelog/v0.1.0-beta4.txt index 1f308b7d..cbc10f48 100644 --- a/changelog/v0.1.0-beta4.txt +++ b/changelog/v0.1.0-beta4.txt @@ -42,6 +42,7 @@ SYNC - On the local network, an object as large as a sync object can be is now passed between devices whole, instead of being refused for its size. - Linked devices check for devices asking to join every 5 seconds only while the Sync page is on screen, and the check no longer reads the whole device list from the server. - A new device that starts from a snapshot whose data can't be read now starts from an older snapshot instead of failing the sync. +- 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. - 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..07fb664b 100644 --- a/src-tauri/src/sync2/engine.rs +++ b/src-tauri/src/sync2/engine.rs @@ -93,6 +93,10 @@ pub enum EngineError { } impl EngineError { + pub fn refused_write(&self) -> bool { + matches!(self, Self::Store(error) if error.refused_write()) + } + /// A stable name the interface turns into a translated message. pub fn kind(&self) -> &'static str { match self { diff --git a/src-tauri/src/sync2/hybrid.rs b/src-tauri/src/sync2/hybrid.rs index cc2b4105..8eb11508 100644 --- a/src-tauri/src/sync2/hybrid.rs +++ b/src-tauri/src/sync2/hybrid.rs @@ -1,8 +1,8 @@ //! A group on a store that its devices also sync over the LAN (spec/live.md, LAN groups: "Groups //! on a store sync the same way"). Everything read from or written to the store is also held, as //! a LAN group holds it, so it can be served to members on the LAN. While the store is out of -//! reach, reads and listings come from what is held, and new objects wait to be uploaded. Roster -//! appends still need the store, which orders them. +//! reach, reads and listings come from what is held, and new objects wait to be uploaded, as they +//! do while the store refuses one. Roster appends still need the store, which orders them. use std::collections::BTreeMap; @@ -78,6 +78,16 @@ impl HybridStore { } impl HybridStore { + /// Objects go up in the order they were written, since readers stop at a gap in a log. + async fn upload_later(&self, path: &str) -> Result<()> { + sqlx::query("INSERT OR IGNORE INTO sync2_pending (path) VALUES (?)") + .bind(path) + .execute(&self.pool) + .await + .map_err(StoreError::Local)?; + Ok(()) + } + async fn held_above( &self, read: &BTreeMap, @@ -162,14 +172,16 @@ impl Store for HybridStore { async fn put_object(&self, path: &str, bytes: Vec) -> Result<()> { self.held.put_object(path, bytes.clone()).await?; + let waiting: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sync2_pending") + .fetch_one(&self.pool) + .await + .map_err(StoreError::Local)?; + if waiting > 0 { + return self.upload_later(path).await; + } match self.store.put_object(path, bytes).await { - Err(error) if unreachable(&error) => { - sqlx::query("INSERT OR IGNORE INTO sync2_pending (path) VALUES (?)") - .bind(path) - .execute(&self.pool) - .await - .map_err(StoreError::Local)?; - Ok(()) + Err(error) if unreachable(&error) || error.refused_write() => { + self.upload_later(path).await } other => other, } diff --git a/src-tauri/src/sync2/server.rs b/src-tauri/src/sync2/server.rs index e4611298..895d8cb1 100644 --- a/src-tauri/src/sync2/server.rs +++ b/src-tauri/src/sync2/server.rs @@ -55,6 +55,14 @@ pub enum StoreError { Protocol(#[from] super::Error), } +impl StoreError { + /// The store answered and turned a write down. A refused sign-in or a store that lost the + /// group isn't this: those end the run. + pub fn refused_write(&self) -> bool { + matches!(self, Self::Refused { status, code } if *status != 401 && code != "unclaimed") + } +} + pub type Result = std::result::Result; /// Whether an append landed, or lost to another device's append at the same head. diff --git a/src-tauri/src/sync2/service.rs b/src-tauri/src/sync2/service.rs index 80bb1193..5c02de31 100644 --- a/src-tauri/src/sync2/service.rs +++ b/src-tauri/src/sync2/service.rs @@ -128,9 +128,11 @@ async fn run(app: &AppHandle) -> Result> { return Ok(Vec::new()); }; let mut session = open_session(&local, &record).await?; - let Some(received) = exchange(app, &local, &mut session).await? else { + let mut refused = None; + let Some(received) = exchange(app, &local, &mut session, &mut refused).await? else { return Ok(Vec::new()); }; + let uploaded = session.store().upload_pending().await; session.collect_garbage(&local.pool).await?; // A server without join requests (older than flow-sync#41) shouldn't fail the sync. if let Some(server) = session.store().server() @@ -139,7 +141,19 @@ async fn run(app: &AppHandle) -> Result> { { let _ = app.emit(EVENT_JOIN_REQUESTS, requests); } - Ok(received) + uploaded?; + refused.map_or(Ok(received), Err) +} + +/// A store that refuses a write still answers reads, so the run goes on and reports it at the end. +fn note_refused(result: Result<()>, refused: &mut Option) -> Result<()> { + match result { + Err(error) if error.refused_write() => { + *refused = Some(error); + Ok(()) + } + other => other, + } } /// Pushes the running profile's changes, applies everyone else's to it, and keeps the batches of @@ -148,6 +162,7 @@ async fn exchange( app: &AppHandle, local: &Local, session: &mut Session<'_, GroupStore>, + refused: &mut Option, ) -> Result>> { let (pool, data) = (&local.pool, &local.data); if !session.adopt_link(pool, &local.default_profile).await? { @@ -175,10 +190,12 @@ async fn exchange( } else { Vec::new() }; - session.store().upload_pending().await?; - session.push(pool, exported).await?; - name_linked(app, pool, &local.meta.app).await?; - session.push_profiles(pool).await?; + let sent = async { + session.push(pool, exported).await?; + name_linked(app, pool, &local.meta.app).await?; + session.push_profiles(pool).await + }; + note_refused(sent.await, refused)?; let pulled = session.pull(pool).await?; let groups: Vec<_> = pulled @@ -193,7 +210,7 @@ async fn exchange( if let Some(login) = &pulled.login { save_pending_login(login).await?; } - session.finish(pool, pulled).await?; + note_refused(session.finish(pool, pulled).await, refused)?; let received = applied.received(session).await; if let Some(link) = &reading.active && snapshot_due(pool, link).await? @@ -201,10 +218,10 @@ async fn exchange( && !links_to_read(pool).await?.contains(link) { let exported = export_collections(data, &local.meta.app, &Collection::ALL).await?; - session.write_snapshot(pool, exported).await?; + note_refused(session.write_snapshot(pool, exported).await, refused)?; } if snapshot_due(pool, PROFILES_KEY).await? { - session.write_profiles_snapshot(pool).await?; + note_refused(session.write_profiles_snapshot(pool).await, refused)?; } Ok(Some(received)) } diff --git a/src-tauri/tests/sync2_hybrid.rs b/src-tauri/tests/sync2_hybrid.rs index a32c33b8..ea52dccc 100644 --- a/src-tauri/tests/sync2_hybrid.rs +++ b/src-tauri/tests/sync2_hybrid.rs @@ -410,3 +410,115 @@ async fn a_damaged_object_held_from_before_gives_way_to_the_good_one() { async fn held_object(pool: &SqlitePool, path: &str) -> Option> { HeldStore::new(pool.clone()).get_object(path).await.unwrap() } + +/// A server that refuses one path, as it does an object it holds different bytes for. +struct RefusingPath<'a> { + store: &'a MemoryStore, + path: Mutex>, +} + +impl Store for &RefusingPath<'_> { + async fn roster_after(&self, after: i64) -> Result> { + self.store.roster_after(after).await + } + + async fn append_roster( + &self, + head: &str, + entry: &Entry, + wraps: &BTreeMap, + chain: Option<&str>, + ) -> Result { + self.store.append_roster(head, entry, wraps, chain).await + } + + async fn key(&self, epoch: u32, kc: &str, name: &str) -> Result>> { + self.store.key(epoch, kc, name).await + } + + async fn put_object(&self, path: &str, bytes: Vec) -> Result<()> { + if self.path.lock().unwrap().as_deref() == Some(path) { + return Err(StoreError::Refused { + status: 409, + code: "conflict".into(), + }); + } + self.store.put_object(path, bytes).await + } + + async fn get_object(&self, path: &str) -> Result>> { + self.store.get_object(path).await + } + + async fn delete_object(&self, path: &str) -> Result { + self.store.delete_object(path).await + } + + async fn changes(&self, cursor: u64) -> Result<(Vec, u64)> { + self.store.changes(cursor).await + } +} + +#[tokio::test] +async fn a_refused_object_waits_in_order_and_reads_go_on() { + let g = group().await; + let (pool_a, pool_b) = (device_pool(&g).await, device_pool(&g).await); + watch(&pool_b, "bbbbbbbbbbb").await; + let mut b = Session::open(&pool_b, &g.store, &g.b).await.unwrap(); + b.push(&pool_b, exported(&pool_b, "app-b").await) + .await + .unwrap(); + + let me = hex::encode(g.a.did); + let (first, second) = (log_path(&me, 1), log_path(&me, 2)); + let server = RefusingPath { + store: &g.store, + path: Mutex::new(Some(first.clone())), + }; + let store = HybridStore::new(&server, pool_a.clone(), me).await.unwrap(); + let mut a = Session::open(&pool_a, store, &g.a).await.unwrap(); + + watch(&pool_a, "aaaaaaaaaaa").await; + a.push(&pool_a, exported(&pool_a, "app-a").await) + .await + .unwrap(); + let pulled = a.pull(&pool_a).await.unwrap(); + for (sender, batches) in &pulled.batches { + apply_batches(&pool_a, "app-a", sender, batches, now_ms()) + .await + .unwrap(); + } + a.finish(&pool_a, pulled).await.unwrap(); + assert_eq!(watched(&pool_a).await, ["aaaaaaaaaaa", "bbbbbbbbbbb"]); + + assert!(matches!( + a.store().upload_pending().await, + Err(StoreError::Refused { status: 409, .. }) + )); + let waiting = pending(&pool_a).await; + assert!(waiting >= 2, "the ack waits behind the batch"); + let objects = |path: &str| g.store.0.lock().unwrap().objects.contains_key(path); + assert!(!objects(&first)); + assert!(!objects(&second), "nothing goes ahead of what was refused"); + + *server.path.lock().unwrap() = None; + a.store().upload_pending().await.unwrap(); + assert_eq!(pending(&pool_a).await, 0); + assert!(objects(&first) && objects(&second)); + + let pulled = b.pull(&pool_b).await.unwrap(); + assert!(pulled.batches.contains_key(&hex::encode(g.a.did))); +} + +#[test] +fn a_refused_sign_in_or_a_lost_group_is_not_a_refused_write() { + let refused = |status, code: &str| StoreError::Refused { + status, + code: code.into(), + }; + assert!(refused(409, "conflict").refused_write()); + assert!(refused(507, "quota").refused_write()); + assert!(!refused(401, "unauthorized").refused_write()); + assert!(!refused(503, "unclaimed").refused_write()); + assert!(!StoreError::SignInRefused.refused_write()); +}