feat(core): node rows name their parent by local id

Schema 40 adds nodes.parent_lid, backfilled from parent_uid. An upsert
and a new row's stub keep it in step, a folder's new row links the rows
stored before it, and a row merged into a landing folder hands its
children over. A restart places each node under its parent's row as it
is now, so a child whose stored node still names a landed folder's
placeholder lands under the folder's real uid.

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 = 39;
+pub(super) const SCHEMA_VERSION: i64 = 40;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -395,6 +395,22 @@ impl Db {
                 tx.execute_batch(MIGRATION_V39)?;
             }
         }
+        if current < 40 {
+            // Same guards as V26-V39.
+            let has_column: bool = tx.query_row(
+                "SELECT COUNT(*) FROM pragma_table_info('nodes') WHERE name = 'parent_lid'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            let has_nodes: bool = tx.query_row(
+                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'nodes'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            if has_nodes && !has_column {
+                tx.execute_batch(MIGRATION_V40)?;
+            }
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -1218,3 +1234,14 @@ UPDATE pending_op
 CREATE INDEX pending_op_lid ON pending_op(lid);
 CREATE INDEX pending_op_parent_lid ON pending_op(parent_lid);
 ";
+
+/// Schema v40: a node row names its parent by local id too.
+///
+/// The local id stays when a folder made here lands and its row takes the uid
+/// Drive gave it, so a child linked by it needs no rewrite. `NULL` while the
+/// parent has no row, as `parent_uid` then names nothing either.
+const MIGRATION_V40: &str = "
+ALTER TABLE nodes ADD COLUMN parent_lid INTEGER;
+UPDATE nodes SET parent_lid = (SELECT p.lid FROM nodes p WHERE p.uid = nodes.parent_uid);
+CREATE INDEX idx_nodes_parent_lid ON nodes(parent_lid);
+";

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

@@ -529,19 +529,30 @@ impl Db {
     }
 
     /// Load every persisted node for cold-start hydration of the `State` maps.
+    ///
+    /// Each node names its parent as the parent's row has it now, found by
+    /// local id: a folder that landed has its new uid there, while the node
+    /// stored below it may still hold the placeholder.
     pub fn load_all(&self) -> Result<Vec<StoredNode>> {
         let conn = self.read();
-        let mut stmt =
-            conn.prepare("SELECT node_json, listed, lid FROM nodes WHERE node_json IS NOT NULL")?;
+        let mut stmt = conn.prepare(
+            "SELECT n.node_json, n.listed, n.lid, p.uid
+             FROM nodes n LEFT JOIN nodes p ON p.lid = n.parent_lid
+             WHERE n.node_json IS NOT NULL",
+        )?;
         let rows = stmt.query_map([], |row| {
             let json: String = row.get(0)?;
             let listed: i64 = row.get(1)?;
-            Ok((json, listed != 0, row.get(2)?))
+            let parent: Option<String> = row.get(3)?;
+            Ok((json, listed != 0, row.get(2)?, parent))
         })?;
         let mut out = Vec::new();
         for row in rows {
-            let (json, listed, lid) = row?;
-            let node: Node = serde_json::from_str(&json)?;
+            let (json, listed, lid, parent) = row?;
+            let mut node: Node = serde_json::from_str(&json)?;
+            if let Some(parent) = parent.as_deref().and_then(parse_node_uid) {
+                node.parent_uid = Some(parent);
+            }
             out.push(StoredNode { node, listed, lid });
         }
         Ok(out)
@@ -602,8 +613,8 @@ impl Db {
         {
             let mut known = tx.prepare_cached("SELECT lid FROM nodes WHERE uid = ?1")?;
             let mut stub = tx.prepare_cached(
-                "INSERT INTO nodes (uid, parent_uid, name, is_dir, mtime, trashed)
-                 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
+                "INSERT INTO nodes (uid, parent_uid, name, is_dir, mtime, trashed, parent_lid)
+                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, (SELECT lid FROM nodes WHERE uid = ?2))",
             )?;
             for node in nodes {
                 let uid = node.uid.to_string();
@@ -618,7 +629,9 @@ impl Db {
                             node.modification_time,
                             node.trashed as i64,
                         ])?;
-                        tx.last_insert_rowid()
+                        let lid = tx.last_insert_rowid();
+                        adopt_children_tx(&tx, &uid, lid)?;
+                        lid
                     }
                 };
                 lids.push(lid);
@@ -791,6 +804,10 @@ pub(super) fn adopt_placeholder_row_tx(
             "UPDATE pending_op SET parent_lid = ?2 WHERE parent_lid = ?1",
             params![known_lid, lid],
         )?;
+        tx.execute(
+            "UPDATE nodes 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])?;
     }
@@ -807,6 +824,16 @@ pub(super) fn adopt_placeholder_row_tx(
     upsert_node_tx(tx, &node)
 }
 
+/// Link the rows that name `uid` as their parent to its new row `lid`: they
+/// were stored before it, or under a row of the same uid since dropped.
+fn adopt_children_tx(tx: &Transaction<'_>, uid: &str, lid: i64) -> Result<()> {
+    tx.execute(
+        "UPDATE nodes SET parent_lid = ?2 WHERE parent_uid = ?1 AND parent_lid IS NOT ?2",
+        params![uid, lid],
+    )?;
+    Ok(())
+}
+
 fn upsert_node_tx(tx: &Transaction<'_>, node: &Node) -> Result<()> {
     let json = serde_json::to_string(node)?;
     let uid = node.uid.to_string();
@@ -831,26 +858,31 @@ fn upsert_node_tx(tx: &Transaction<'_>, node: &Node) -> Result<()> {
     // where resolving it from scratch is a recursive walk to the root. A node
     // whose parent is not cached (a device folder's root lives in `device`, never
     // in `nodes`) starts a path of its own, which is what the walk did too.
-    let path = match &parent_uid {
-        None => String::new(),
+    let (parent_lid, path) = match &parent_uid {
+        None => (None, String::new()),
         Some(parent) => {
-            let parent_path: Option<Option<String>> = tx
+            let parent_row: Option<(i64, Option<String>)> = tx
                 .query_row(
-                    "SELECT path FROM nodes WHERE uid = ?1",
+                    "SELECT lid, path FROM nodes WHERE uid = ?1",
                     params![parent],
-                    |row| row.get(0),
+                    |row| Ok((row.get(0)?, row.get(1)?)),
                 )
                 .optional()?;
-            join_path(parent_path.flatten().as_deref().unwrap_or(""), &node.name)
+            let (lid, parent_path) = parent_row.unzip();
+            (
+                lid,
+                join_path(parent_path.flatten().as_deref().unwrap_or(""), &node.name),
+            )
         }
     };
 
     tx.execute(
         "INSERT INTO nodes
-           (uid, parent_uid, name, is_dir, size, mtime, trashed, node_json, path)
-         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
+           (uid, parent_uid, name, is_dir, size, mtime, trashed, node_json, path, parent_lid)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
          ON CONFLICT(uid) DO UPDATE SET
            parent_uid = excluded.parent_uid,
+           parent_lid = excluded.parent_lid,
            name       = excluded.name,
            is_dir     = excluded.is_dir,
            size       = excluded.size,
@@ -868,6 +900,7 @@ fn upsert_node_tx(tx: &Transaction<'_>, node: &Node) -> Result<()> {
             node.trashed as i64,
             json,
             path,
+            parent_lid,
         ],
     )?;
     let rowid: i64 = tx.query_row(
@@ -875,6 +908,9 @@ fn upsert_node_tx(tx: &Transaction<'_>, node: &Node) -> Result<()> {
         params![uid],
         |row| row.get(0),
     )?;
+    if prior.is_none() {
+        adopt_children_tx(tx, &uid, rowid)?;
+    }
 
     // FTS5 has no UPSERT, so the node's own row is always replaced.
     let indexable = node_is_indexable_tx(tx, &uid)?;

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

@@ -3376,6 +3376,67 @@ fn a_create_waits_for_its_folder_by_local_id() {
     assert_eq!(op.parent_uid, Some(uid("dir").to_string()));
 }
 
+fn parent_lid(db: &Db, uid: &NodeUid) -> Option<i64> {
+    db.conn
+        .lock()
+        .query_row(
+            "SELECT parent_lid FROM nodes WHERE uid = ?1",
+            [uid.to_string()],
+            |r| r.get(0),
+        )
+        .unwrap()
+}
+
+/// A node stored before its folder, as an event can bring it, is linked to
+/// the folder by local id once the folder's row arrives.
+#[test]
+fn a_node_stored_before_its_folder_is_linked_when_the_folder_arrives() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&file("child", "dir", "inside.txt", 1))
+        .unwrap();
+    assert_eq!(parent_lid(&db, &uid("child")), None);
+
+    db.upsert_node(&folder("dir", Some("root"), "Folder"))
+        .unwrap();
+
+    assert_eq!(
+        parent_lid(&db, &uid("child")),
+        db.lid_of(&uid("dir").to_string()).unwrap()
+    );
+    assert_eq!(
+        parent_lid(&db, &uid("dir")),
+        db.lid_of(&uid("root").to_string()).unwrap()
+    );
+}
+
+/// A restart puts a node below the folder its row links to by local id, under
+/// the uid that row has now, whatever the node's own copy of it says.
+#[test]
+fn a_restart_finds_a_node_under_its_landed_folder() {
+    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 mut child = file("child", "root", "inside.txt", 1);
+    child.parent_uid = Some(local.clone());
+    db.upsert_node(&child).unwrap();
+    let dir = db.lid_of(&local.to_string()).unwrap().unwrap();
+    db.conn
+        .lock()
+        .execute(
+            "UPDATE nodes SET uid = ?2 WHERE lid = ?1",
+            rusqlite::params![dir, uid("dir").to_string()],
+        )
+        .unwrap();
+
+    let stored = db.load_all().unwrap();
+    let child = stored.iter().find(|s| s.node.uid == uid("child")).unwrap();
+    assert_eq!(child.node.parent_uid, Some(uid("dir")));
+}
+
 /// 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]
@@ -3841,6 +3902,16 @@ fn a_2_8_database_migrates_with_every_node_and_op_intact() {
         ),
         "the migration names the op's node and parent by local id"
     );
+    let parent_lid: Option<i64> = db
+        .conn
+        .lock()
+        .query_row(
+            "SELECT parent_lid FROM nodes WHERE uid = 'local~inner'",
+            [],
+            |r| r.get(0),
+        )
+        .unwrap();
+    assert_eq!(parent_lid, db.lid_of("local~dir").unwrap());
     let parked = db.parked_create_ops().unwrap();
     assert_eq!(parked.len(), 1);
     assert_eq!(parked[0].name.as_deref(), Some("movie.part"));