fix(sync): keep reading when the store refuses an upload

changelog/v0.1.0-beta4.txt

@@ -46,6 +46,7 @@ SYNC
 - 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.
+- 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

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 {

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<S: Store> HybridStore<S> {
 }
 
 impl<S> HybridStore<S> {
+    /// 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<String, u64>,
@@ -162,14 +172,16 @@ impl<S: Store + Sync> Store for HybridStore<S> {
 
     async fn put_object(&self, path: &str, bytes: Vec<u8>) -> 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,
         }

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<T> = std::result::Result<T, StoreError>;
 
 /// Whether an append landed, or lost to another device's append at the same head.

src-tauri/src/sync2/service.rs

@@ -128,9 +128,11 @@ async fn run(app: &AppHandle) -> Result<Vec<Received>> {
         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<Vec<Received>> {
     {
         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<EngineError>) -> 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<EngineError>,
 ) -> Result<Option<Vec<Received>>> {
     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,18 +210,18 @@ 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?
         && kept_count(pool, link).await? == 0
         && !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))
 }

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<Vec<u8>> {
     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<Option<String>>,
+}
+
+impl Store for &RefusingPath<'_> {
+    async fn roster_after(&self, after: i64) -> Result<Vec<Entry>> {
+        self.store.roster_after(after).await
+    }
+
+    async fn append_roster(
+        &self,
+        head: &str,
+        entry: &Entry,
+        wraps: &BTreeMap<String, String>,
+        chain: Option<&str>,
+    ) -> Result<Appended> {
+        self.store.append_roster(head, entry, wraps, chain).await
+    }
+
+    async fn key(&self, epoch: u32, kc: &str, name: &str) -> Result<Option<Vec<u8>>> {
+        self.store.key(epoch, kc, name).await
+    }
+
+    async fn put_object(&self, path: &str, bytes: Vec<u8>) -> 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<Option<Vec<u8>>> {
+        self.store.get_object(path).await
+    }
+
+    async fn delete_object(&self, path: &str) -> Result<bool> {
+        self.store.delete_object(path).await
+    }
+
+    async fn changes(&self, cursor: u64) -> Result<(Vec<ChangedObject>, 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());
+}