feat(core,fuse): queued ops name their node and folder by local id

Schema 39 adds pending_op.lid and parent_lid, backfilled from nodes. Every
op queued or retargeted gets both, and a create gets its row before it
is queued. The claim reads a parent's uid through parent_lid, and a
landing folder rewrites the ops made inside it by local id instead of by
its placeholder string. Ops queued against a row Drive listed follow it
when it is merged into the landing one.

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

@@ -11,7 +11,7 @@ use super::Db;
 use crate::Result;
 
 /// Current schema version. Bump on every forward migration added below.
-pub(super) const SCHEMA_VERSION: i64 = 38;
+pub(super) const SCHEMA_VERSION: i64 = 39;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -379,6 +379,22 @@ impl Db {
                 tx.execute_batch(MIGRATION_V38)?;
             }
         }
+        if current < 39 {
+            // Same guards as V26-V38.
+            let has_column: bool = tx.query_row(
+                "SELECT COUNT(*) FROM pragma_table_info('pending_op') WHERE name = 'lid'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            let has_ops: bool = tx.query_row(
+                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'pending_op'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            if has_ops && !has_column {
+                tx.execute_batch(MIGRATION_V39)?;
+            }
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -1186,3 +1202,19 @@ CREATE INDEX idx_pending_op_claimed
 const MIGRATION_V38: &str = "
 ALTER TABLE pending_op ADD COLUMN issue TEXT;
 ";
+
+/// Schema v39: a queued op names its node and its parent by local id too.
+///
+/// A create made inside a folder that had not landed named the folder by its
+/// placeholder uid, and every landing had to find and rewrite those strings.
+/// The local id stays the same when the folder lands. An op whose node has no
+/// row keeps `NULL` and is matched by uid, as before.
+const MIGRATION_V39: &str = "
+ALTER TABLE pending_op ADD COLUMN lid INTEGER;
+ALTER TABLE pending_op ADD COLUMN parent_lid INTEGER;
+UPDATE pending_op
+   SET lid = (SELECT lid FROM nodes WHERE nodes.uid = pending_op.uid),
+       parent_lid = (SELECT lid FROM nodes WHERE nodes.uid = pending_op.parent_uid);
+CREATE INDEX pending_op_lid ON pending_op(lid);
+CREATE INDEX pending_op_parent_lid ON pending_op(parent_lid);
+";

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

@@ -760,9 +760,20 @@ pub(super) fn adopt_placeholder_row_tx(
             |row| Ok((row.get(0)?, row.get(1)?)),
         )
         .optional()?;
+    // Ops queued before schema 39 have no `parent_lid`, nor do ops made
+    // under a placeholder that never had a row.
+    tx.execute(
+        "UPDATE pending_op SET parent_uid = ?2 WHERE parent_lid IS NULL AND parent_uid = ?1",
+        params![local, real],
+    )?;
     let Some((lid, json)) = row else {
         return Ok(());
     };
+    // The ops made inside the folder name it by its local id, which stays.
+    tx.execute(
+        "UPDATE pending_op SET parent_uid = ?2 WHERE parent_lid = ?1",
+        params![lid, real],
+    )?;
     let known: Option<(i64, Option<String>)> = tx
         .query_row(
             "SELECT lid, node_json FROM nodes WHERE uid = ?1",
@@ -771,6 +782,15 @@ pub(super) fn adopt_placeholder_row_tx(
         )
         .optional()?;
     if let Some((known_lid, _)) = known {
+        // Ops queued against the row Drive listed now name the landing one.
+        tx.execute(
+            "UPDATE pending_op SET lid = ?2 WHERE lid = ?1",
+            params![known_lid, lid],
+        )?;
+        tx.execute(
+            "UPDATE pending_op SET parent_lid = ?2 WHERE parent_lid = ?1",
+            params![known_lid, lid],
+        )?;
         tx.execute("DELETE FROM nodes WHERE lid = ?1", params![known_lid])?;
         tx.execute("DELETE FROM nodes_fts WHERE rowid = ?1", params![known_lid])?;
     }

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

@@ -175,6 +175,26 @@ pub(super) fn wake_ops_waiting_for_tx(tx: &rusqlite::Transaction<'_>, name: &str
     Ok(())
 }
 
+/// Point queued op `id` at its node and its parent by local id, from the uids
+/// it names (schema 39). A uid with no row leaves `NULL`, and the op is matched
+/// by uid, as before.
+fn fill_op_lids(conn: &rusqlite::Connection, id: i64) -> Result<()> {
+    conn.execute(
+        "UPDATE pending_op
+         SET lid = (SELECT lid FROM nodes WHERE uid = pending_op.uid),
+             parent_lid = (SELECT lid FROM nodes WHERE uid = pending_op.parent_uid)
+         WHERE id = ?1",
+        params![id],
+    )?;
+    Ok(())
+}
+
+/// The uid a queued op's parent has now: the one its row holds, which a
+/// landing folder has already moved to its real uid, or the one the op was
+/// queued with when the parent has no row.
+const PARENT_NOW: &str =
+    "COALESCE((SELECT uid FROM nodes WHERE lid = pending_op.parent_lid), parent_uid)";
+
 /// The volume id given to a node that exists only on this machine, so far. A
 /// real [`NodeUid`] is `{volume}~{link}`, so a placeholder is `local~<uuid>` and
 /// round-trips through the same `Display`/parse path as any other uid.
@@ -434,6 +454,7 @@ impl Db {
             ],
         )?;
         let id = tx.last_insert_rowid();
+        fill_op_lids(&tx, id)?;
         tx.commit()?;
         Ok((id, superseded))
     }
@@ -460,6 +481,7 @@ impl Db {
             params![OP_TRASH, uid, name, created_at],
         )?;
         let id = tx.last_insert_rowid();
+        fill_op_lids(&tx, id)?;
         tx.commit()?;
         Ok((id, blobs))
     }
@@ -575,6 +597,7 @@ impl Db {
                 params![OP_TRASH, real, landing.landed.1, now],
             )?;
             let trash = tx.last_insert_rowid();
+            fill_op_lids(&tx, trash)?;
             tx.commit()?;
             return Ok(CreateRetired::Withdrawn { trash });
         }
@@ -590,7 +613,8 @@ impl Db {
             Some((_, meta)) => {
                 tx.execute(
                     "UPDATE pending_op
-                     SET kind = ?2, uid = ?3, parent_uid = NULL, name = NULL, meta_json = ?4,
+                     SET kind = ?2, uid = ?3, parent_uid = NULL, parent_lid = NULL, name = NULL,
+                         meta_json = ?4,
                          attempts = 0, last_error = NULL, next_attempt_at = 0
                      WHERE id = ?1",
                     params![id, OP_REVISION, real, meta],
@@ -621,11 +645,8 @@ impl Db {
                  VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                 params![OP_RENAME, real, parent, name, meta, created_at],
             )?;
+            fill_op_lids(&tx, tx.last_insert_rowid())?;
         }
-        tx.execute(
-            "UPDATE pending_op SET parent_uid = ?2 WHERE parent_uid = ?1",
-            params![local, real],
-        )?;
         adopt_placeholder_row_tx(&tx, local, real)?;
         tx.commit()?;
         Ok(match newer {
@@ -672,7 +693,9 @@ impl Db {
     pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result<bool> {
         let conn = self.conn.lock();
         let n = conn.execute(
-            "UPDATE pending_op SET parent_uid = ?2, name = ?3,
+            "UPDATE pending_op
+             SET parent_uid = ?2, name = ?3,
+               parent_lid = (SELECT lid FROM nodes WHERE uid = ?2),
                next_attempt_at = CASE WHEN next_attempt_at < ?6 THEN 0 ELSE next_attempt_at END
              WHERE uid = ?1 AND kind IN (?4, ?5)",
             params![uid, parent_uid, name, OP_CREATE, OP_MKDIR, PARK_UNTIL],
@@ -751,17 +774,12 @@ impl Db {
         Ok(blobs)
     }
 
-    /// Rewrite every queued op that points at a placeholder parent, once that
-    /// parent has drained and has a real uid. Also moves the node rows whose
-    /// parent column still names the placeholder, so listings keep resolving,
-    /// and readdresses the placeholder's own row.
+    /// Readdress a placeholder that has drained to its real uid: its row, the
+    /// rows below it, and the queued ops made inside it
+    /// ([`adopt_placeholder_row_tx`]).
     pub fn remap_local_uid(&self, local: &str, real: &str) -> Result<()> {
         let mut conn = self.conn.lock();
         let tx = conn.transaction()?;
-        tx.execute(
-            "UPDATE pending_op SET parent_uid = ?2 WHERE parent_uid = ?1",
-            params![local, real],
-        )?;
         adopt_placeholder_row_tx(&tx, local, real)?;
         tx.commit()?;
         Ok(())
@@ -782,9 +800,10 @@ impl Db {
     /// materialized and rejected in Rust:
     ///
     /// * `next_attempt_at <= now` — the backoff.
-    /// * a `local~` parent is skipped. A node created inside a folder that was
-    ///   itself created offline cannot be sent anywhere until that folder is
-    ///   real. Ops replay in queue order so the parent normally drains first;
+    /// * a parent that has not landed is skipped, read through `parent_lid`
+    ///   from the parent's row ([`PARENT_NOW`]). A node created inside a folder
+    ///   that was itself created offline cannot be sent anywhere until that
+    ///   folder is real. Ops replay in queue order so the parent normally drains first;
     ///   this matters when the parent is backing off, where the child must wait
     ///   rather than burn its own retries.
     ///
@@ -795,11 +814,11 @@ impl Db {
     pub fn next_due_op(&self, now: i64) -> Result<Option<PendingOp>> {
         let conn = self.read();
         let mut stmt = conn.prepare_cached(&format!(
-            "SELECT id, kind, uid, parent_uid, name, blob_path, meta_json, created_at,
+            "SELECT id, kind, uid, {PARENT_NOW}, name, blob_path, meta_json, created_at,
                     attempts, last_error, next_attempt_at
              FROM pending_op
              WHERE next_attempt_at <= ?1
-               AND (parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~')
+               AND (parent_uid IS NULL OR substr({PARENT_NOW}, 1, {n}) <> '{v}~')
              ORDER BY id LIMIT 1",
             v = LOCAL_VOLUME,
             n = LOCAL_VOLUME.len() + 1,
@@ -849,12 +868,12 @@ impl Db {
         let tx = conn.transaction()?;
         let op = {
             let mut stmt = tx.prepare_cached(&format!(
-                "SELECT id, kind, uid, parent_uid, name, blob_path, meta_json, created_at,
+                "SELECT id, kind, uid, {PARENT_NOW}, name, blob_path, meta_json, created_at,
                         attempts, last_error, next_attempt_at
                  FROM pending_op
                  WHERE next_attempt_at <= ?1
                    AND claimed_at = 0
-                   AND (parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~')
+                   AND (parent_uid IS NULL OR substr({PARENT_NOW}, 1, {n}) <> '{v}~')
                    AND uid NOT IN (SELECT uid FROM pending_op WHERE claimed_at <> 0)
                  ORDER BY id LIMIT 1",
                 v = LOCAL_VOLUME,
@@ -932,7 +951,7 @@ impl Db {
                 &format!(
                     "SELECT MIN(next_attempt_at) FROM pending_op \
                      WHERE claimed_at = 0 \
-                       AND (parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~')",
+                       AND (parent_uid IS NULL OR substr({PARENT_NOW}, 1, {n}) <> '{v}~')",
                     v = LOCAL_VOLUME,
                     n = LOCAL_VOLUME.len() + 1,
                 ),

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

@@ -3321,6 +3321,104 @@ fn a_landed_folder_keeps_a_row_until_the_server_copy_replaces_it() {
     assert_eq!(found[0].path, "New folder/inside.txt");
 }
 
+/// A queued create names its folder by local id, so it is sent into the folder
+/// as soon as the folder's row has its real uid, whatever the op's own copy
+/// of the folder's uid still says.
+#[test]
+fn a_create_waits_for_its_folder_by_local_id() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    let local = NodeUid::new(VolumeId::from("local"), LinkId::from("dir"));
+    let mut placeholder = folder("dir", Some("root"), "New folder");
+    placeholder.uid = local.clone();
+    db.upsert_node(&placeholder).unwrap();
+    let dir = db.lid_of(&local.to_string()).unwrap().unwrap();
+    let (id, _) = db
+        .enqueue_op(&PendingOp {
+            id: 0,
+            kind: OP_CREATE.to_string(),
+            uid: "local~child".to_string(),
+            parent_uid: Some(local.to_string()),
+            name: Some("inside.txt".to_string()),
+            blob_path: None,
+            meta_json: None,
+            created_at: 1,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        })
+        .unwrap();
+    let parent_lid: Option<i64> = db
+        .conn
+        .lock()
+        .query_row(
+            "SELECT parent_lid FROM pending_op WHERE id = ?1",
+            [id],
+            |r| r.get(0),
+        )
+        .unwrap();
+    assert_eq!(parent_lid, Some(dir));
+    assert!(
+        db.claim_next_due_op(1).unwrap().is_none(),
+        "nothing to send into a folder that has not landed"
+    );
+
+    db.conn
+        .lock()
+        .execute(
+            "UPDATE nodes SET uid = ?2 WHERE lid = ?1",
+            rusqlite::params![dir, uid("dir").to_string()],
+        )
+        .unwrap();
+
+    let op = db.claim_next_due_op(1).unwrap().expect("the create is due");
+    assert_eq!(op.id, id);
+    assert_eq!(op.parent_uid, Some(uid("dir").to_string()));
+}
+
+/// A rename queued against the row Drive listed for a folder still landing
+/// follows the folder when that row is merged into the landing one (B140).
+#[test]
+fn an_op_on_a_listed_twin_follows_the_folder_that_lands() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    let local = NodeUid::new(VolumeId::from("local"), LinkId::from("dir"));
+    let mut placeholder = folder("dir", Some("root"), "New folder");
+    placeholder.uid = local.clone();
+    db.upsert_node(&placeholder).unwrap();
+    db.upsert_node(&folder("real", Some("root"), "New folder"))
+        .unwrap();
+    let kept = db.lid_of(&local.to_string()).unwrap().unwrap();
+    let (rename, _) = db
+        .enqueue_op(&PendingOp {
+            id: 0,
+            kind: OP_RENAME.to_string(),
+            uid: uid("real").to_string(),
+            parent_uid: Some(uid("root").to_string()),
+            name: Some("Renamed".to_string()),
+            blob_path: None,
+            meta_json: None,
+            created_at: 1,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        })
+        .unwrap();
+
+    db.remap_local_uid(&local.to_string(), &uid("real").to_string())
+        .unwrap();
+
+    let lid: Option<i64> = db
+        .conn
+        .lock()
+        .query_row("SELECT lid FROM pending_op WHERE id = ?1", [rename], |r| {
+            r.get(0)
+        })
+        .unwrap();
+    assert_eq!(lid, Some(kept));
+    assert_eq!(db.lid_of(&uid("real").to_string()).unwrap(), Some(kept));
+}
+
 /// B148: a folder pinned before it landed is still pinned once it has.
 #[test]
 fn a_folder_pinned_before_it_landed_stays_pinned() {
@@ -3726,6 +3824,23 @@ fn a_2_8_database_migrates_with_every_node_and_op_intact() {
     let queued = db.pending_ops().unwrap();
     let inner = queued.iter().find(|op| op.uid == "local~inner").unwrap();
     assert_eq!(inner.parent_uid.as_deref(), Some("local~dir"));
+    let lids: (Option<i64>, Option<i64>) = db
+        .conn
+        .lock()
+        .query_row(
+            "SELECT lid, parent_lid FROM pending_op WHERE uid = 'local~inner'",
+            [],
+            |r| Ok((r.get(0)?, r.get(1)?)),
+        )
+        .unwrap();
+    assert_eq!(
+        lids,
+        (
+            db.lid_of("local~inner").unwrap(),
+            db.lid_of("local~dir").unwrap()
+        ),
+        "the migration names the op's node and parent by local id"
+    );
     let parked = db.parked_create_ops().unwrap();
     assert_eq!(parked.len(), 1);
     assert_eq!(parked[0].name.as_deref(), Some("movie.part"));

crates/pdfs-fuse/src/lib.rs

@@ -3627,8 +3627,8 @@ impl Core {
     /// path returns the one the server minted.
     ///
     /// The parent may itself be a placeholder — `mkdir -p` offline, or `cp -r` of
-    /// a tree. That is fine: the op records the parent it was made under, and the
-    /// parent's own drain rewrites it to the real uid before this op can run.
+    /// a tree. That is fine: the op records the parent it was made under by local
+    /// id, and is not sent until the parent's row has the uid it landed as.
     fn queue_local_node(
         &self,
         parent_uid: &NodeUid,
@@ -3638,6 +3638,13 @@ impl Core {
     ) -> Result<Node, Errno> {
         self.require_uid_writable(parent_uid)?;
         let uid = mint_local_uid();
+        let node = local_node(uid.clone(), parent_uid.clone(), name.to_string(), is_dir);
+        // The row the tree would give it anyway, now, so the op is queued with
+        // the node's local id.
+        self.db.lids_for(&[&node]).map_err(|e| {
+            error!(%parent_uid, name, error = %e, "giving a local node its row failed");
+            Errno::EIO
+        })?;
         let op = PendingOp {
             id: 0,
             kind: if is_dir { OP_MKDIR } else { OP_CREATE }.to_string(),
@@ -3659,12 +3666,7 @@ impl Core {
             Errno::EIO
         })?;
         debug!(%uid, %parent_uid, name, is_dir, "created node offline; queued");
-        Ok(local_node(
-            uid,
-            parent_uid.clone(),
-            name.to_string(),
-            is_dir,
-        ))
+        Ok(node)
     }
 
     /// Queue giving a node a new parent and/or name, and apply it to the tree