refactor(core,fuse): a node made here goes by its local id until it lands (schema 41)

A file or folder made on the mount used to get a minted `local~<ms>-<n>`
placeholder. It now gets `local~<lid>`, a function of the row it is given
first (`Db::add_local_row`), so the stand-in never changes and always leads
back to the row, and through it to the real uid once the node lands.
`Db::lid_of` reads it without a lookup.

Migration V41 rewrites the placeholders still stored to their row's
stand-in: the row and its node_json, its children, the ops on it or into
it (uid, parent, sidecar uid, a rename's original parent), and its pins.

Step 6a of the plan for 3.0.0 phase 3.

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 = 40;
+pub(super) const SCHEMA_VERSION: i64 = 41;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -411,6 +411,19 @@ impl Db {
                 tx.execute_batch(MIGRATION_V40)?;
             }
         }
+        if current < 41 {
+            // Needs both tables; on a database that never had them there is
+            // nothing to rewrite.
+            let tables: i64 = tx.query_row(
+                "SELECT COUNT(*) FROM sqlite_master
+                 WHERE type = 'table' AND name IN ('nodes', 'pending_op', 'pins')",
+                [],
+                |row| row.get(0),
+            )?;
+            if tables == 3 {
+                tx.execute_batch(MIGRATION_V41)?;
+            }
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -1245,3 +1258,47 @@ 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);
 ";
+
+/// Schema v41: a node made on this machine goes by `local~<lid>` until it
+/// lands, a function of its row, where it used to get a `local~<ms>-<n>`
+/// placeholder. Every placeholder still stored is rewritten to its row's
+/// stand-in: the row itself and its `node_json`, its children, the ops on it
+/// or into it, and its pins. A placeholder op with no row keeps its uid and
+/// drains as before.
+const MIGRATION_V41: &str = "
+CREATE TEMP TABLE local_v41 AS
+  SELECT uid AS old, 'local~' || lid AS new, lid FROM nodes
+   WHERE uid LIKE 'local~%' AND uid <> 'local~' || lid;
+UPDATE pending_op SET uid = (SELECT new FROM local_v41 WHERE old = pending_op.uid)
+ WHERE uid IN (SELECT old FROM local_v41);
+UPDATE pending_op SET parent_uid = (SELECT new FROM local_v41 WHERE old = pending_op.parent_uid)
+ WHERE parent_uid IN (SELECT old FROM local_v41);
+UPDATE pending_op
+   SET meta_json = json_set(meta_json, '$.uid',
+         (SELECT new FROM local_v41 WHERE old = json_extract(pending_op.meta_json, '$.uid')))
+ WHERE json_valid(meta_json)
+   AND json_extract(meta_json, '$.uid') IN (SELECT old FROM local_v41);
+UPDATE pending_op
+   SET meta_json = json_set(meta_json, '$.original_parent_uid',
+         (SELECT new FROM local_v41
+           WHERE old = json_extract(pending_op.meta_json, '$.original_parent_uid')))
+ WHERE json_valid(meta_json)
+   AND json_extract(meta_json, '$.original_parent_uid') IN (SELECT old FROM local_v41);
+UPDATE pins SET uid = (SELECT new FROM local_v41 WHERE old = pins.uid)
+ WHERE uid IN (SELECT old FROM local_v41);
+UPDATE nodes
+   SET node_json = json_set(node_json, '$.parent_uid.link_id',
+         (SELECT CAST(lid AS TEXT) FROM local_v41 WHERE old = nodes.parent_uid))
+ WHERE parent_uid IN (SELECT old FROM local_v41)
+   AND json_valid(node_json)
+   AND json_extract(node_json, '$.parent_uid.volume_id') = 'local';
+UPDATE nodes SET parent_uid = (SELECT new FROM local_v41 WHERE old = nodes.parent_uid)
+ WHERE parent_uid IN (SELECT old FROM local_v41);
+UPDATE nodes
+   SET node_json = json_set(node_json, '$.uid.link_id', CAST(lid AS TEXT))
+ WHERE lid IN (SELECT lid FROM local_v41)
+   AND json_valid(node_json)
+   AND json_extract(node_json, '$.uid.volume_id') = 'local';
+UPDATE nodes SET uid = 'local~' || lid WHERE lid IN (SELECT lid FROM local_v41);
+DROP TABLE local_v41;
+";

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

@@ -58,7 +58,7 @@ pub use nodes::{PublishedSharedRoot, SearchHit, StoredNode};
 pub use ops::{
     AttachedBlob, CreateLanding, CreateRetired, FAILING_ATTEMPTS, LOCAL_VOLUME, OP_CREATE,
     OP_MKDIR, OP_RENAME, OP_REVISION, OP_TRASH, PARK_EXPIRY_MS, PARK_UNTIL, PendingCounts,
-    PendingOp, RenameMeta, op_supersedes,
+    PendingOp, RenameMeta, local_lid, local_uid, op_supersedes,
 };
 pub use photos::{StoredPhoto, StoredPlace, THUMB_HAVE, THUMB_NONE, THUMB_UNKNOWN, TimelineRow};
 pub use pins::PinRow;

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

@@ -6,7 +6,7 @@ use std::collections::HashSet;
 use rusqlite::{OptionalExtension, Transaction, params};
 
 use super::Db;
-use super::ops::{OP_CREATE, OP_MKDIR, OP_RENAME, wake_ops_waiting_for_tx};
+use super::ops::{OP_CREATE, OP_MKDIR, OP_RENAME, local_lid, local_uid, wake_ops_waiting_for_tx};
 use crate::{Access, Result};
 use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
 use proton_drive_rs::{Node, NodeKind};
@@ -593,14 +593,21 @@ impl Db {
 
     /// The local id of the node stored under `uid`, or `None` when it has no
     /// row. It stays the same for as long as the row exists, including when a
-    /// create lands and the row takes the uid Drive gave it.
+    /// create lands and the row takes the uid Drive gave it: a node made on
+    /// this machine is found by its stand-in ([`local_uid`]) before and after.
     pub fn lid_of(&self, uid: &str) -> Result<Option<i64>> {
         let conn = self.read();
-        Ok(conn
-            .query_row("SELECT lid FROM nodes WHERE uid = ?1", params![uid], |r| {
+        let row = match local_lid(uid) {
+            Some(lid) => {
+                conn.query_row("SELECT lid FROM nodes WHERE lid = ?1", params![lid], |r| {
+                    r.get(0)
+                })
+            }
+            None => conn.query_row("SELECT lid FROM nodes WHERE uid = ?1", params![uid], |r| {
                 r.get(0)
-            })
-            .optional()?)
+            }),
+        };
+        Ok(row.optional()?)
     }
 
     /// The local id of each node, in order, giving a row to each that has none.
@@ -644,6 +651,41 @@ impl Db {
         Ok(lids)
     }
 
+    /// Give a node being made on this machine its row under `parent_uid`, and
+    /// return its local id. The row goes by [`local_uid`] of that id until
+    /// the node lands.
+    ///
+    /// Like [`Db::lids_for`], only a stub: the caller upserts the node.
+    pub fn add_local_row(
+        &self,
+        parent_uid: &str,
+        name: &str,
+        is_dir: bool,
+        mtime: i64,
+    ) -> Result<i64> {
+        let mut conn = self.conn.lock();
+        let tx = conn.transaction()?;
+        tx.execute(
+            "INSERT INTO nodes (uid, parent_uid, name, is_dir, mtime, trashed, parent_lid)
+             VALUES (NULL, ?1, ?2, ?3, ?4, 0,
+                     COALESCE(?5, (SELECT lid FROM nodes WHERE uid = ?1)))",
+            params![
+                parent_uid,
+                name,
+                is_dir as i64,
+                mtime,
+                local_lid(parent_uid)
+            ],
+        )?;
+        let lid = tx.last_insert_rowid();
+        tx.execute(
+            "UPDATE nodes SET uid = ?2 WHERE lid = ?1",
+            params![lid, local_uid(lid)],
+        )?;
+        tx.commit()?;
+        Ok(lid)
+    }
+
     /// Read the persisted incremental-sync cursor (a `DriveEventId`), if any.
     /// The daemon resumes from this on restart instead of reseeding to the
     /// server head, so changes made while unmounted are still applied (P2).

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

@@ -205,13 +205,33 @@ 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
+/// real [`NodeUid`] is `{volume}~{link}`, so a placeholder is `local~<lid>` and
 /// round-trips through the same `Display`/parse path as any other uid.
 ///
 /// Nothing bearing this volume may be handed to the API — it would 404. The
 /// drain replaces it with the uid the server assigns.
 pub const LOCAL_VOLUME: &str = "local";
 
+/// The uid a node made on this machine goes by until Drive gives it one: its
+/// row's local id under [`LOCAL_VOLUME`].
+///
+/// A function of the row, so it never changes: whatever still holds it once
+/// the node has landed finds the node, and its real uid, through
+/// [`local_lid`].
+pub fn local_uid(lid: i64) -> String {
+    format!("{LOCAL_VOLUME}~{lid}")
+}
+
+/// The local id a [`local_uid`] stands for, or `None` for any other uid,
+/// including a placeholder minted before schema 41.
+pub fn local_lid(uid: &str) -> Option<i64> {
+    let digits = uid.strip_prefix(LOCAL_VOLUME)?.strip_prefix('~')?;
+    if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) {
+        return None;
+    }
+    digits.parse().ok()
+}
+
 /// A `next_attempt_at` far enough in the future that [`Db::next_due_op`] never
 /// selects the op (ms since epoch, ≈ year 2223). Used to *park* a queued create
 /// for a transient file — a browser's in-flight `*.crdownload`/`*.part`, an

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

@@ -3897,15 +3897,25 @@ fn migration_v36_gives_every_node_its_rowid_as_lid() {
     assert_eq!(rowids.len(), 3);
 
     let db = Db::open(&path).unwrap();
+    // The placeholder goes by its row's stand-in from V41 on.
+    let stand_in = |uid: &str, rowid: i64| match uid == local.to_string() {
+        true => local_uid(rowid),
+        false => uid.to_string(),
+    };
     for (uid, rowid) in &rowids {
-        assert_eq!(db.lid_of(uid).unwrap(), Some(*rowid), "{uid}");
+        let uid = stand_in(uid, *rowid);
+        assert_eq!(db.lid_of(&uid).unwrap(), Some(*rowid), "{uid}");
     }
     let found = db.search("report", 10).unwrap();
     assert_eq!(found.len(), 1, "the search index still finds the node");
     assert_eq!(found[0].node.uid, uid("kept"));
     let ops = db.pending_ops().unwrap();
     assert_eq!(ops.len(), 1);
-    assert_eq!(ops[0].uid, local.to_string());
+    let (_, dir) = rowids
+        .iter()
+        .find(|(uid, _)| *uid == local.to_string())
+        .unwrap();
+    assert_eq!(ops[0].uid, local_uid(*dir));
     assert_eq!(ops[0].attempts, 1);
     assert_eq!(ops[0].last_error.as_deref(), Some("network unreachable"));
 
@@ -3916,6 +3926,93 @@ fn migration_v36_gives_every_node_its_rowid_as_lid() {
     let _ = std::fs::remove_file(&path);
 }
 
+#[test]
+fn migration_v41_gives_every_placeholder_its_rows_stand_in() {
+    let path = std::env::temp_dir().join(format!(
+        "pdfs-db-v40-fixture-{}-{}.db",
+        std::process::id(),
+        now_test_id()
+    ));
+    let _ = std::fs::remove_file(&path);
+    let (dir, inner) = {
+        let db = Db::open(&path).unwrap();
+        db.upsert_node(&folder("root", None, "My Files")).unwrap();
+        let old_dir = NodeUid::new(VolumeId::from("local"), LinkId::from("1700000000000-0"));
+        let old_inner = NodeUid::new(VolumeId::from("local"), LinkId::from("1700000000000-1"));
+        let mut dir = folder("x", Some("root"), "Offline folder");
+        dir.uid = old_dir.clone();
+        db.upsert_node(&dir).unwrap();
+        let mut inner = file("x", "root", "inner.txt", 0);
+        inner.uid = old_inner.clone();
+        inner.parent_uid = Some(old_dir.clone());
+        db.upsert_node(&inner).unwrap();
+        let op = |kind: &str, uid: &NodeUid, parent: &str, meta: Option<String>| PendingOp {
+            id: 0,
+            kind: kind.to_string(),
+            uid: uid.to_string(),
+            parent_uid: Some(parent.to_string()),
+            name: Some("n".to_string()),
+            blob_path: None,
+            meta_json: meta,
+            created_at: 1,
+            attempts: 0,
+            last_error: None,
+            next_attempt_at: 0,
+        };
+        db.enqueue_op(&op(OP_MKDIR, &old_dir, &uid("root").to_string(), None))
+            .unwrap();
+        let staged = json!({"uid": old_inner.to_string(), "len": 0}).to_string();
+        db.enqueue_op(&op(
+            OP_CREATE,
+            &old_inner,
+            &old_dir.to_string(),
+            Some(staged),
+        ))
+        .unwrap();
+        db.pin_add(&old_dir.to_string(), "/Offline folder", true)
+            .unwrap();
+        let lids = (
+            db.lid_of(&old_dir.to_string()).unwrap().unwrap(),
+            db.lid_of(&old_inner.to_string()).unwrap().unwrap(),
+        );
+        drop(db);
+        rusqlite::Connection::open(&path)
+            .unwrap()
+            .execute(
+                "UPDATE sync_state SET value = '40' WHERE key = 'schema_version'",
+                [],
+            )
+            .unwrap();
+        lids
+    };
+
+    let db = Db::open(&path).unwrap();
+    let (dir_uid, inner_uid) = (local_uid(dir), local_uid(inner));
+    assert_eq!(db.uid_of_lid(dir).unwrap(), Some(dir_uid.clone()));
+    assert_eq!(db.uid_of_lid(inner).unwrap(), Some(inner_uid.clone()));
+    let stored = db.node_by_uid(&inner_uid).unwrap().unwrap();
+    assert_eq!(
+        stored.uid.to_string(),
+        inner_uid,
+        "node_json follows the row"
+    );
+    assert_eq!(
+        stored.parent_uid.map(|p| p.to_string()),
+        Some(dir_uid.clone())
+    );
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops[0].uid, dir_uid);
+    assert_eq!(ops[1].uid, inner_uid);
+    assert_eq!(ops[1].parent_uid, Some(dir_uid.clone()));
+    let staged: serde_json::Value =
+        serde_json::from_str(ops[1].meta_json.as_deref().unwrap()).unwrap();
+    assert_eq!(staged["uid"], inner_uid.as_str());
+    let pins: Vec<String> = db.pin_list().unwrap().into_iter().map(|p| p.uid).collect();
+    assert_eq!(pins, vec![dir_uid]);
+    drop(db);
+    let _ = std::fs::remove_file(&path);
+}
+
 /// A database as 2.8.2 wrote it (`sqlite3 .dump`, schema 35), with queued ops
 /// in every state the drain leaves them in: due, parked, failing, deferred for
 /// access, and claimed by a worker when the daemon stopped. The creates below
@@ -3935,36 +4032,45 @@ fn a_2_8_database_migrates_with_every_node_and_op_intact() {
     let (nodes, ops) = migrates_intact(&path);
     assert_eq!((nodes, ops), (8, 6));
     let db = Db::open(&path).unwrap();
+    let lid_named = |name: &str| -> i64 {
+        db.conn
+            .lock()
+            .query_row("SELECT lid FROM nodes WHERE name = ?1", [name], |r| {
+                r.get(0)
+            })
+            .unwrap()
+    };
+    let (dir, inner_lid) = (lid_named("Offline folder"), lid_named("inner.txt"));
     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 inner = queued
+        .iter()
+        .find(|op| op.uid == local_uid(inner_lid))
+        .unwrap();
+    assert_eq!(inner.parent_uid, Some(local_uid(dir)));
     let lids: (Option<i64>, Option<i64>) = db
         .conn
         .lock()
         .query_row(
-            "SELECT lid, parent_lid FROM pending_op WHERE uid = 'local~inner'",
-            [],
+            "SELECT lid, parent_lid FROM pending_op WHERE id = ?1",
+            [inner.id],
             |r| Ok((r.get(0)?, r.get(1)?)),
         )
         .unwrap();
     assert_eq!(
         lids,
-        (
-            db.lid_of("local~inner").unwrap(),
-            db.lid_of("local~dir").unwrap()
-        ),
+        (Some(inner_lid), Some(dir)),
         "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'",
-            [],
+            "SELECT parent_lid FROM nodes WHERE lid = ?1",
+            [inner_lid],
             |r| r.get(0),
         )
         .unwrap();
-    assert_eq!(parent_lid, db.lid_of("local~dir").unwrap());
+    assert_eq!(parent_lid, Some(dir));
     let parked = db.parked_create_ops().unwrap();
     assert_eq!(parked.len(), 1);
     assert_eq!(parked[0].name.as_deref(), Some("movie.part"));
@@ -3997,7 +4103,7 @@ fn a_copy_of_a_real_database_migrates_with_every_node_and_op_intact() {
 /// Opens the database at `path` and checks that the migration kept every node
 /// on its row id and every queued op as it was. Returns how many of each.
 fn migrates_intact(path: &std::path::Path) -> (usize, usize) {
-    let (version, nodes, ops, sealed) = {
+    let (version, nodes, mut ops, sealed) = {
         let conn = rusqlite::Connection::open(path).unwrap();
         let version: String = conn
             .query_row(
@@ -4017,6 +4123,33 @@ fn migrates_intact(path: &std::path::Path) -> (usize, usize) {
         let sealed = rows(&conn, "SELECT * FROM own_sealed_rev ORDER BY uid");
         (version, nodes, ops, sealed)
     };
+    // A node made on this machine goes by its row's stand-in from schema 41
+    // on, in its ops as everywhere else.
+    let stand_ins: Vec<(String, String)> = nodes
+        .iter()
+        .filter(|(uid, rowid)| uid.starts_with("local~") && *uid != local_uid(*rowid))
+        .map(|(uid, rowid)| (uid.clone(), local_uid(*rowid)))
+        .collect();
+    let nodes: Vec<(String, i64)> = nodes
+        .into_iter()
+        .map(
+            |(uid, rowid)| match stand_ins.iter().any(|(old, _)| *old == uid) {
+                true => (local_uid(rowid), rowid),
+                false => (uid, rowid),
+            },
+        )
+        .collect();
+    for value in ops.iter_mut().flatten() {
+        if let rusqlite::types::Value::Text(text) = value {
+            for (old, new) in &stand_ins {
+                if text == old {
+                    *text = new.clone();
+                } else {
+                    *text = text.replace(&format!("\"{old}\""), &format!("\"{new}\""));
+                }
+            }
+        }
+    }
     let db = Db::open(path).unwrap();
     {
         let conn = rusqlite::Connection::open(path).unwrap();