fix(core,fuse): keep queued writes durable and their retries their own

- A gap-filled revision blob is recorded as complete once filled, so a
  retry after a lost upload answer recognises its own bytes on Drive
  instead of forking a conflict copy (B228).
- Commits that accept a change into the queue run at synchronous=FULL,
  so an op `close()` reported as done survives a power cut (B229).
- A queued create's adoption window starts at its first send, not when
  it was queued, so a create queued offline no longer takes another
  device's empty node of the same name. A send record that does not
  parse is kept instead of overwritten (B230).

crates/pdfs-core/src/db/mod.rs

@@ -179,6 +179,45 @@ impl Drop for Reader<'_> {
     }
 }
 
+/// The write connection with `synchronous=FULL`, back to NORMAL on drop.
+///
+/// NORMAL in WAL mode keeps a commit across a crash of the process but not
+/// across a power loss: the WAL is synced only at checkpoints. That is fine for
+/// metadata the next listing re-fetches, and not for the queue. A staged blob
+/// is synced before its op is queued, and the op row is what makes it
+/// reachable; losing that row loses a write `close()` already reported as
+/// done. A FULL commit syncs the WAL, which covers every NORMAL commit before
+/// it as well.
+pub(super) struct Durable<'a>(parking_lot::MutexGuard<'a, Connection>);
+
+impl<'a> Durable<'a> {
+    pub(super) fn new(conn: parking_lot::MutexGuard<'a, Connection>) -> Result<Self> {
+        conn.pragma_update(None, "synchronous", "FULL")?;
+        Ok(Self(conn))
+    }
+}
+
+impl std::ops::Deref for Durable<'_> {
+    type Target = Connection;
+
+    fn deref(&self) -> &Connection {
+        &self.0
+    }
+}
+
+impl std::ops::DerefMut for Durable<'_> {
+    fn deref_mut(&mut self) -> &mut Connection {
+        &mut self.0
+    }
+}
+
+impl Drop for Durable<'_> {
+    fn drop(&mut self) {
+        // A failure leaves the writer at FULL: slower, never less safe.
+        let _ = self.0.pragma_update(None, "synchronous", "NORMAL");
+    }
+}
+
 /// Open a second connection to the same file for reads only.
 ///
 /// `SQLITE_OPEN_READ_ONLY` plus `query_only` is belt and braces: the flag stops
@@ -270,6 +309,12 @@ impl Db {
         Ok(db)
     }
 
+    /// The write connection for a commit that accepts a user's change into
+    /// the queue. See [`Durable`].
+    pub(super) fn durable(&self) -> Result<Durable<'_>> {
+        Durable::new(self.conn.lock())
+    }
+
     /// A connection for a `SELECT`-only method.
     ///
     /// Reads see the last committed state, which is the same thing they saw
@@ -297,7 +342,8 @@ impl Db {
     fn open_configured(path: &Path) -> Result<Connection> {
         let conn = Connection::open(path)?;
         // WAL: readers never block the single writer. NORMAL sync is the
-        // standard durability/throughput tradeoff for WAL.
+        // standard durability/throughput tradeoff for WAL; the queue commits
+        // that need more raise it to FULL for themselves (`Durable`).
         conn.pragma_update(None, "journal_mode", "WAL")?;
         conn.pragma_update(None, "synchronous", "NORMAL")?;
         conn.pragma_update(None, "foreign_keys", "ON")?;

crates/pdfs-core/src/db/ops.rs

@@ -279,6 +279,45 @@ fn claimable() -> String {
 /// drain replaces it with the uid the server assigns.
 pub const LOCAL_VOLUME: &str = "local";
 
+/// One place a queued create was sent, with when it first was.
+struct SentTarget {
+    parent: String,
+    name: String,
+    at_ms: Option<i64>,
+}
+
+/// The `sent_to` column: `[[parent, name, at_ms], …]`, or `[[parent, name], …]`
+/// as written before the time was kept. `None` when it does not parse.
+fn parse_sent_to(json: Option<&str>) -> Option<Vec<SentTarget>> {
+    #[derive(Deserialize)]
+    #[serde(untagged)]
+    enum Entry {
+        At(String, String, i64),
+        Bare(String, String),
+    }
+    let Some(json) = json else {
+        return Some(Vec::new());
+    };
+    let entries: Vec<Entry> = serde_json::from_str(json).ok()?;
+    Some(
+        entries
+            .into_iter()
+            .map(|entry| match entry {
+                Entry::At(parent, name, at) => SentTarget {
+                    parent,
+                    name,
+                    at_ms: Some(at),
+                },
+                Entry::Bare(parent, name) => SentTarget {
+                    parent,
+                    name,
+                    at_ms: None,
+                },
+            })
+            .collect(),
+    )
+}
+
 /// The uid a node made on this machine goes by until Drive gives it one: its
 /// row's local id under [`LOCAL_VOLUME`].
 ///
@@ -541,7 +580,7 @@ impl Db {
     /// land in the order they were made (`docs/BUGS.md` B141). Should it fail,
     /// [`Db::record_op_failure`] or [`Db::release_op_claim`] drops it.
     pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option<String>)> {
-        let mut conn = self.conn.lock();
+        let mut conn = self.durable()?;
         let tx = conn.transaction()?;
         let superseded: Option<String> = if op_supersedes(&op.kind) {
             let lid = node_lid(&tx, &op.uid)?;
@@ -598,7 +637,7 @@ impl Db {
         name: &str,
         created_at: i64,
     ) -> Result<(i64, Vec<String>)> {
-        let mut conn = self.conn.lock();
+        let mut conn = self.durable()?;
         let tx = conn.transaction()?;
         let blobs = drop_doomed_ops(&tx, uid)?;
         tx.execute(
@@ -628,7 +667,7 @@ impl Db {
         blob_path: &str,
         meta_json: &str,
     ) -> Result<Option<AttachedBlob>> {
-        let conn = self.conn.lock();
+        let conn = self.durable()?;
         let lid = node_lid(&conn, uid)?;
         let existing: Option<(i64, Option<String>)> = conn
             .query_row(
@@ -691,7 +730,7 @@ impl Db {
         rewrite: impl FnOnce(&str) -> Option<String>,
     ) -> Result<CreateRetired> {
         let (local, real) = (landing.local, landing.real);
-        let mut conn = self.conn.lock();
+        let mut conn = self.durable()?;
         let tx = conn.transaction()?;
         let row: Option<CreateRow> = tx
             .query_row(
@@ -808,12 +847,19 @@ impl Db {
         Ok(())
     }
 
-    /// Record that queued create `id` is being sent to `parent` under `name`.
+    /// Record that queued create `id` is being sent to `parent` under `name`
+    /// at `at_ms`.
     ///
     /// A rename while the op waits rewrites its target, but the file a lost
     /// answer left stays where it was sent. The retry and the trash look
-    /// there too (`docs/BUGS.md` B184).
-    pub fn note_create_target(&self, id: i64, parent: &str, name: &str) -> Result<()> {
+    /// there too (`docs/BUGS.md` B184). A target sent before keeps its first
+    /// time.
+    ///
+    /// A record that does not parse is left as it is rather than replaced
+    /// with this one target: the earlier ones are what the duplicate checks
+    /// look for. Without them a lost answer forks a conflict copy, which
+    /// loses nothing.
+    pub fn note_create_target(&self, id: i64, parent: &str, name: &str, at_ms: i64) -> Result<()> {
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
         let sent: Option<String> = tx
@@ -822,15 +868,26 @@ impl Db {
             })
             .optional()?
             .flatten();
-        let mut targets: Vec<(String, String)> = sent
-            .and_then(|json| serde_json::from_str(&json).ok())
-            .unwrap_or_default();
-        let target = (parent.to_string(), name.to_string());
-        if !targets.contains(&target) {
-            targets.push(target);
+        let Some(mut targets) = parse_sent_to(sent.as_deref()) else {
+            tracing::warn!(
+                id,
+                "a queued create's send record does not parse; keeping it"
+            );
+            return Ok(());
+        };
+        if !targets.iter().any(|t| t.parent == parent && t.name == name) {
+            targets.push(SentTarget {
+                parent: parent.to_string(),
+                name: name.to_string(),
+                at_ms: Some(at_ms),
+            });
+            let json: Vec<(&str, &str, i64)> = targets
+                .iter()
+                .map(|t| (t.parent.as_str(), t.name.as_str(), t.at_ms.unwrap_or(0)))
+                .collect();
             tx.execute(
                 "UPDATE pending_op SET sent_to = ?2 WHERE id = ?1",
-                params![id, serde_json::to_string(&targets)?],
+                params![id, serde_json::to_string(&json)?],
             )?;
         }
         tx.commit()?;
@@ -840,6 +897,24 @@ impl Db {
     /// The parent and name queued op `id` was sent under as a create, in the
     /// order it was ([`Db::note_create_target`]).
     pub fn create_targets(&self, id: i64) -> Result<Vec<(String, String)>> {
+        Ok(self
+            .sent_targets(id)?
+            .into_iter()
+            .map(|t| (t.parent, t.name))
+            .collect())
+    }
+
+    /// When queued create `id` was first sent, if that was recorded. Rows
+    /// written before the time was kept have none.
+    pub fn create_first_sent(&self, id: i64) -> Result<Option<i64>> {
+        Ok(self
+            .sent_targets(id)?
+            .into_iter()
+            .filter_map(|t| t.at_ms.filter(|&at| at > 0))
+            .min())
+    }
+
+    fn sent_targets(&self, id: i64) -> Result<Vec<SentTarget>> {
         let sent: Option<String> = self
             .conn
             .lock()
@@ -848,9 +923,7 @@ impl Db {
             })
             .optional()?
             .flatten();
-        Ok(sent
-            .and_then(|json| serde_json::from_str(&json).ok())
-            .unwrap_or_default())
+        Ok(parse_sent_to(sent.as_deref()).unwrap_or_default())
     }
 
     /// The SHA-1 of each blob queued op `id` sent as a create
@@ -898,6 +971,27 @@ impl Db {
         Ok(n > 0)
     }
 
+    /// Replace the sidecar of op `id`, but only while it still holds the blob
+    /// at `blob_path`.
+    ///
+    /// The drain fills an incomplete blob in place and then records it as
+    /// whole, so a retry after a lost answer can recognise its own bytes on the
+    /// server. A write that closed meanwhile has pointed the row at a newer
+    /// blob with its own sidecar; that one must not be overwritten. Returns
+    /// false when the row is gone or holds another blob.
+    pub fn update_op_meta_for_blob(
+        &self,
+        id: i64,
+        blob_path: &str,
+        meta_json: &str,
+    ) -> Result<bool> {
+        let n = self.conn.lock().execute(
+            "UPDATE pending_op SET meta_json = ?3 WHERE id = ?1 AND blob_path = ?2",
+            params![id, blob_path, meta_json],
+        )?;
+        Ok(n > 0)
+    }
+
     /// Point a queued create at a new parent and name, for a node renamed or
     /// moved before it ever reached the server.
     ///
@@ -911,7 +1005,7 @@ impl Db {
     /// name it no longer wants, such as one held by a queued change. A parked
     /// create stays parked.
     pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result<bool> {
-        let conn = self.conn.lock();
+        let conn = self.durable()?;
         let lid = node_lid(&conn, uid)?;
         let parent_lid = node_lid(&conn, parent_uid)?;
         let n = conn.execute(

crates/pdfs-core/src/db/tests.rs

@@ -2352,6 +2352,100 @@ fn refuses_a_schema_newer_than_the_running_build() {
     let _ = std::fs::remove_file(path);
 }
 
+/// Queueing a write syncs it to disk, and leaves the writer at NORMAL after.
+///
+/// The op row is what makes a staged blob reachable once `close()` has
+/// reported success, so its commit must survive a power loss. Everything else
+/// the writer commits is metadata the next listing re-fetches, and pays only
+/// NORMAL's cost.
+#[test]
+fn queueing_a_write_commits_at_full_sync_and_restores_normal() {
+    let path = std::env::temp_dir().join(format!("pdfs-db-sync-{}.db", std::process::id()));
+    let _ = std::fs::remove_file(&path);
+    let db = Db::open(&path).unwrap();
+    let level = |db: &Db| -> i64 {
+        db.conn
+            .lock()
+            .query_row("PRAGMA synchronous", [], |r| r.get(0))
+            .unwrap()
+    };
+    // 1 = NORMAL, 2 = FULL.
+    assert_eq!(level(&db), 1);
+    {
+        let durable = db.durable().unwrap();
+        let during: i64 = durable
+            .query_row("PRAGMA synchronous", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(during, 2);
+    }
+    assert_eq!(level(&db), 1);
+
+    db.enqueue_op(&PendingOp {
+        id: 0,
+        kind: OP_TRASH.to_string(),
+        uid: uid("a").to_string(),
+        parent_uid: None,
+        name: Some("a".to_string()),
+        blob_path: None,
+        meta_json: None,
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    })
+    .unwrap();
+    assert_eq!(level(&db), 1);
+    assert_eq!(db.pending_ops().unwrap().len(), 1);
+    drop(db);
+    for suffix in ["", "-wal", "-shm"] {
+        let _ = std::fs::remove_file(format!("{}{suffix}", path.display()));
+    }
+}
+
+/// A gap-filled blob is recorded as whole only while the op still holds it.
+///
+/// A write that closes during the drain points the row at a newer blob with
+/// its own sidecar. Overwriting that sidecar with the older blob's would
+/// describe bytes the row no longer holds.
+#[test]
+fn marking_a_blob_whole_leaves_a_newer_blob_alone() {
+    let db = Db::open_in_memory().unwrap();
+    let op = PendingOp {
+        id: 0,
+        kind: OP_REVISION.to_string(),
+        uid: uid("a").to_string(),
+        parent_uid: None,
+        name: None,
+        blob_path: Some("/staging/first".to_string()),
+        meta_json: Some(r#"{"complete":false}"#.to_string()),
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    };
+    db.enqueue_op(&op).unwrap();
+    let id = db.pending_ops().unwrap()[0].id;
+
+    assert!(
+        db.update_op_meta_for_blob(id, "/staging/first", r#"{"complete":true}"#)
+            .unwrap()
+    );
+    assert_eq!(
+        db.pending_ops().unwrap()[0].meta_json.as_deref(),
+        Some(r#"{"complete":true}"#)
+    );
+
+    assert!(
+        !db.update_op_meta_for_blob(id, "/staging/older", r#"{"stale":true}"#)
+            .unwrap(),
+        "the row holds another blob"
+    );
+    assert_eq!(
+        db.pending_ops().unwrap()[0].meta_json.as_deref(),
+        Some(r#"{"complete":true}"#)
+    );
+}
+
 /// A queued write whose baseline is restamped must keep everything else.
 ///
 /// The restamp happens after our *own* upload seals a new revision under a
@@ -4081,15 +4175,15 @@ fn what_a_create_sent_outlives_a_newer_blob_its_trash_and_a_restart() {
             .unwrap();
         assert!(db.create_sent(id).unwrap().is_empty());
         assert!(db.create_targets(id).unwrap().is_empty());
-        db.note_create_target(id, &parent, "c").unwrap();
-        db.note_create_target(id, &parent, "c").unwrap();
+        db.note_create_target(id, &parent, "c", 10).unwrap();
+        db.note_create_target(id, &parent, "c", 20).unwrap();
         db.note_create_sent(id, "aaaa").unwrap();
         db.note_create_sent(id, "aaaa").unwrap();
         db.record_op_failure(id, "no answer", 0).unwrap();
         db.attach_blob_to_create("local~sent", "/staging/second", "{}")
             .unwrap()
             .expect("the create is still queued");
-        db.note_create_target(id, &parent, "d").unwrap();
+        db.note_create_target(id, &parent, "d", 30).unwrap();
         db.note_create_sent(id, "bbbb").unwrap();
         db.record_op_failure(id, "no answer", 0).unwrap();
         db.delete_ops_for_uid("local~sent").unwrap();
@@ -4112,6 +4206,73 @@ fn what_a_create_sent_outlives_a_newer_blob_its_trash_and_a_restart() {
     let _ = std::fs::remove_file(path);
 }
 
+/// A queued create knows when it first went out, which is where the window
+/// for adopting a node its lost answer left starts. Records written before
+/// the time was kept still list their targets, and one that does not parse is
+/// not replaced by the next target.
+#[test]
+fn a_create_remembers_when_it_was_first_sent() {
+    let db = Db::open_in_memory().unwrap();
+    let parent = uid("parent").to_string();
+    let (id, _) = db
+        .enqueue_op(&PendingOp {
+            id: 0,
+            kind: OP_CREATE.to_string(),
+            uid: "local~first".to_string(),
+            parent_uid: Some(parent.clone()),
+            name: Some("a.txt".to_string()),
+            blob_path: None,
+            meta_json: None,
+            created_at: 1,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        })
+        .unwrap();
+    assert_eq!(db.create_first_sent(id).unwrap(), None, "never sent");
+    db.note_create_target(id, &parent, "a.txt", 500).unwrap();
+    db.note_create_target(id, &parent, "a.txt", 900).unwrap();
+    db.note_create_target(id, &parent, "b.txt", 700).unwrap();
+    assert_eq!(db.create_first_sent(id).unwrap(), Some(500));
+
+    let set = |json: &str| {
+        db.conn
+            .lock()
+            .execute(
+                "UPDATE pending_op SET sent_to = ?2 WHERE id = ?1",
+                rusqlite::params![id, json],
+            )
+            .unwrap();
+    };
+    set(&format!(r#"[["{parent}","old.txt"]]"#));
+    assert_eq!(
+        db.create_targets(id).unwrap(),
+        vec![(parent.clone(), "old.txt".to_string())]
+    );
+    assert_eq!(db.create_first_sent(id).unwrap(), None);
+    db.note_create_target(id, &parent, "a.txt", 900).unwrap();
+    assert_eq!(
+        db.create_targets(id).unwrap().len(),
+        2,
+        "the old target stays"
+    );
+    assert_eq!(db.create_first_sent(id).unwrap(), Some(900));
+
+    set("not json");
+    db.note_create_target(id, &parent, "c.txt", 1000).unwrap();
+    let raw: String = db
+        .conn
+        .lock()
+        .query_row("SELECT sent_to FROM pending_op WHERE id = ?1", [id], |r| {
+            r.get(0)
+        })
+        .unwrap();
+    assert_eq!(
+        raw, "not json",
+        "an unreadable record is kept, not replaced"
+    );
+}
+
 /// A write after a create failed clears its attempts, so the next try goes
 /// out at once. The create was still sent: deleting the file then left no
 /// trash, and the file a lost answer made stayed on Drive under the name it
@@ -4135,7 +4296,7 @@ fn a_create_rewritten_after_it_failed_still_leaves_a_trash_when_deleted() {
             next_attempt_at: 0,
         })
         .unwrap();
-    db.note_create_target(id, &parent, "a.txt").unwrap();
+    db.note_create_target(id, &parent, "a.txt", 10).unwrap();
     db.note_create_sent(id, "aaaa").unwrap();
     db.record_op_failure(id, "no answer", 0).unwrap();
     db.attach_blob_to_create("local~sent", "/staging/second", "{}")