diff --git a/crates/pdfs-core/src/db/nodes.rs b/crates/pdfs-core/src/db/nodes.rs index 33f8da4..4de542c 100644 --- a/crates/pdfs-core/src/db/nodes.rs +++ b/crates/pdfs-core/src/db/nodes.rs @@ -596,18 +596,7 @@ impl Db { /// 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> { - let conn = self.read(); - 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) - }), - }; - Ok(row.optional()?) + node_lid(&self.read(), uid) } /// The local id of each node, in order, giving a row to each that has none. @@ -775,6 +764,19 @@ struct PriorRow { trashed: bool, } +/// [`Db::lid_of`] on `conn`, which may be a transaction. +pub(super) fn node_lid(conn: &rusqlite::Connection, uid: &str) -> Result> { + 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) + }), + }; + Ok(row.optional()?) +} + /// Hand a drained placeholder's row, and the rows below it, to the real uid it /// landed as. /// diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs index 10e43cc..22d3a59 100644 --- a/crates/pdfs-core/src/db/ops.rs +++ b/crates/pdfs-core/src/db/ops.rs @@ -8,7 +8,7 @@ use rusqlite::{OptionalExtension, params}; use serde::{Deserialize, Serialize}; use super::Db; -use super::nodes::adopt_placeholder_row_tx; +use super::nodes::{adopt_placeholder_row_tx, node_lid}; use crate::Result; /// The `kind` of a [`PendingOp`] that uploads a staged file as a new revision. @@ -127,9 +127,11 @@ fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result = { let mut stmt = tx.prepare(&format!( "{DOOMED} @@ -137,7 +139,7 @@ fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result(0))?; + let rows = stmt.query_map(params![uid, lid], |r| r.get::<_, String>(0))?; rows.collect::, _>>()? }; tx.execute( @@ -149,7 +151,7 @@ fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result, uid: &str) -> Result, name: &str } /// 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. +/// it names (schema 39), a stand-in ([`local_uid`]) included. 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", + let (uid, parent): (String, Option) = conn.query_row( + "SELECT uid, parent_uid FROM pending_op WHERE id = ?1", params![id], + |r| Ok((r.get(0)?, r.get(1)?)), + )?; + let parent_lid = match parent { + Some(parent) => node_lid(conn, &parent)?, + None => None, + }; + conn.execute( + "UPDATE pending_op SET lid = ?2, parent_lid = ?3 WHERE id = ?1", + params![id, node_lid(conn, &uid)?, parent_lid], )?; Ok(()) } +/// The ops on the node `?1` names, `?L` being its local id +/// ([`Db::lid_of`]): queued under that uid, or under the node's other one, its +/// stand-in before it landed or its real uid after. +const ON_NODE: &str = "(uid = ?1 OR lid = ?L)"; + +/// [`ON_NODE`] with `?L` numbered `n`. +fn on_node(n: usize) -> String { + ON_NODE.replace("?L", &format!("?{n}")) +} + +/// The uid a queued op's node has now: the one its row holds, which a landing +/// create has already moved to its real uid, or the one the op was queued with +/// when the node has no row. +const UID_NOW: &str = "COALESCE((SELECT uid FROM nodes WHERE lid = pending_op.lid), uid)"; + /// 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. @@ -438,11 +461,13 @@ impl Db { /// Whether a specific desired-state operation is still queued for a node. pub fn has_pending_op(&self, uid: &str, kind: &str) -> Result { let conn = self.read(); + let lid = node_lid(&conn, uid)?; conn.query_row( - "SELECT EXISTS( - SELECT 1 FROM pending_op WHERE uid = ?1 AND kind = ?2 - )", - params![uid, kind], + &format!( + "SELECT EXISTS(SELECT 1 FROM pending_op WHERE {} AND kind = ?2)", + on_node(3) + ), + params![uid, kind, lid], |row| row.get(0), ) .map_err(Into::into) @@ -451,9 +476,13 @@ impl Db { /// Read an existing operation's metadata before a superseding enqueue. pub fn pending_op_meta(&self, uid: &str, kind: &str) -> Result> { let conn = self.read(); + let lid = node_lid(&conn, uid)?; conn.query_row( - "SELECT meta_json FROM pending_op WHERE uid = ?1 AND kind = ?2", - params![uid, kind], + &format!( + "SELECT meta_json FROM pending_op WHERE {} AND kind = ?2", + on_node(3) + ), + params![uid, kind, lid], |row| row.get(0), ) .optional() @@ -471,19 +500,22 @@ impl Db { let mut conn = self.conn.lock(); let tx = conn.transaction()?; let superseded: Option = if op_supersedes(&op.kind) { + let lid = node_lid(&tx, &op.uid)?; + let same = format!( + "{} AND kind = ?2 AND (kind <> ?3 OR claimed_at = 0)", + on_node(4) + ); let blob: Option = tx .query_row( - "SELECT blob_path FROM pending_op - WHERE uid = ?1 AND kind = ?2 AND (kind <> ?3 OR claimed_at = 0)", - params![op.uid, op.kind, OP_RENAME], + &format!("SELECT blob_path FROM pending_op WHERE {same}"), + params![op.uid, op.kind, OP_RENAME, lid], |r| r.get(0), ) .optional()? .flatten(); tx.execute( - "DELETE FROM pending_op - WHERE uid = ?1 AND kind = ?2 AND (kind <> ?3 OR claimed_at = 0)", - params![op.uid, op.kind, OP_RENAME], + &format!("DELETE FROM pending_op WHERE {same}"), + params![op.uid, op.kind, OP_RENAME, lid], )?; blob } else { @@ -553,10 +585,14 @@ impl Db { meta_json: &str, ) -> Result> { let conn = self.conn.lock(); + let lid = node_lid(&conn, uid)?; let existing: Option<(i64, Option)> = conn .query_row( - "SELECT id, blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2", - params![uid, OP_CREATE], + &format!( + "SELECT id, blob_path FROM pending_op WHERE {} AND kind = ?2", + on_node(3) + ), + params![uid, OP_CREATE, lid], |r| Ok((r.get(0)?, r.get(1)?)), ) .optional()?; @@ -688,8 +724,8 @@ impl Db { original_name: Some(landing.landed.1.to_string()), })?; tx.execute( - "DELETE FROM pending_op WHERE uid = ?1 AND kind = ?2", - params![real, OP_RENAME], + &format!("DELETE FROM pending_op WHERE {} AND kind = ?2", on_node(3)), + params![real, OP_RENAME, node_lid(&tx, local)?], )?; tx.execute( "INSERT INTO pending_op (kind, uid, parent_uid, name, meta_json, created_at) @@ -722,9 +758,13 @@ impl Db { /// queued, which is the ordinary case. pub fn update_op_meta(&self, uid: &str, kind: &str, meta_json: &str) -> Result { let conn = self.conn.lock(); + let lid = node_lid(&conn, uid)?; let n = conn.execute( - "UPDATE pending_op SET meta_json = ?3 WHERE uid = ?1 AND kind = ?2", - params![uid, kind, meta_json], + &format!( + "UPDATE pending_op SET meta_json = ?3 WHERE {} AND kind = ?2", + on_node(4) + ), + params![uid, kind, meta_json, lid], )?; Ok(n > 0) } @@ -743,13 +783,19 @@ impl Db { /// create stays parked. pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result { let conn = self.conn.lock(); + let lid = node_lid(&conn, uid)?; + let parent_lid = node_lid(&conn, parent_uid)?; let n = conn.execute( - "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], + &format!( + "UPDATE pending_op + SET parent_uid = ?2, name = ?3, parent_lid = ?7, + next_attempt_at = CASE WHEN next_attempt_at < ?6 THEN 0 ELSE next_attempt_at END + WHERE {} AND kind IN (?4, ?5)", + on_node(8) + ), + params![ + uid, parent_uid, name, OP_CREATE, OP_MKDIR, PARK_UNTIL, parent_lid, lid + ], )?; Ok(n > 0) } @@ -765,10 +811,19 @@ impl Db { /// already is. pub fn set_create_hold(&self, uid: &str, held: bool) -> Result { let conn = self.conn.lock(); + let lid = node_lid(&conn, uid)?; let n = conn.execute( - "UPDATE pending_op SET next_attempt_at = ?2 - WHERE uid = ?1 AND kind IN (?3, ?4)", - params![uid, if held { PARK_UNTIL } else { 0 }, OP_CREATE, OP_MKDIR], + &format!( + "UPDATE pending_op SET next_attempt_at = ?2 WHERE {} AND kind IN (?3, ?4)", + on_node(5) + ), + params![ + uid, + if held { PARK_UNTIL } else { 0 }, + OP_CREATE, + OP_MKDIR, + lid + ], )?; Ok(n > 0) } @@ -776,9 +831,13 @@ impl Db { /// Check if a queued create or mkdir op exists for `uid`. pub fn has_create_op(&self, uid: &str) -> Result { let conn = self.read(); + let lid = node_lid(&conn, uid)?; let count: i64 = conn.query_row( - "SELECT COUNT(*) FROM pending_op WHERE uid = ?1 AND kind IN (?2, ?3)", - params![uid, OP_CREATE, OP_MKDIR], + &format!( + "SELECT COUNT(*) FROM pending_op WHERE {} AND kind IN (?2, ?3)", + on_node(4) + ), + params![uid, OP_CREATE, OP_MKDIR, lid], |row| row.get(0), )?; Ok(count > 0) @@ -792,9 +851,10 @@ impl Db { /// blob that holds them (`docs/BUGS.md` B71). pub fn has_any_op(&self, uid: &str) -> Result { let conn = self.read(); + let lid = node_lid(&conn, uid)?; let count: i64 = conn.query_row( - "SELECT COUNT(*) FROM pending_op WHERE uid = ?1", - params![uid], + &format!("SELECT COUNT(*) FROM pending_op WHERE {}", on_node(2)), + params![uid, lid], |row| row.get(0), )?; Ok(count > 0) @@ -825,10 +885,10 @@ impl Db { Ok(blobs) } - /// 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<()> { + /// What [`Db::finish_create`] does to the node rows and pins when a create + /// lands, without the op ([`adopt_placeholder_row_tx`]). + #[cfg(test)] + pub(crate) fn land_placeholder_row(&self, local: &str, real: &str) -> Result<()> { let mut conn = self.conn.lock(); let tx = conn.transaction()?; adopt_placeholder_row_tx(&tx, local, real)?; @@ -865,7 +925,7 @@ impl Db { pub fn next_due_op(&self, now: i64) -> Result> { let conn = self.read(); let mut stmt = conn.prepare_cached(&format!( - "SELECT id, kind, uid, {PARENT_NOW}, name, blob_path, meta_json, created_at, + "SELECT id, kind, {UID_NOW}, {PARENT_NOW}, name, blob_path, meta_json, created_at, attempts, last_error, next_attempt_at FROM pending_op WHERE next_attempt_at <= ?1 @@ -919,13 +979,15 @@ impl Db { let tx = conn.transaction()?; let op = { let mut stmt = tx.prepare_cached(&format!( - "SELECT id, kind, uid, {PARENT_NOW}, name, blob_path, meta_json, created_at, + "SELECT id, kind, {UID_NOW}, {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_NOW}, 1, {n}) <> '{v}~') AND uid NOT IN (SELECT uid FROM pending_op WHERE claimed_at <> 0) + AND (lid IS NULL OR lid NOT IN ( + SELECT lid FROM pending_op WHERE claimed_at <> 0 AND lid IS NOT NULL)) ORDER BY id LIMIT 1", v = LOCAL_VOLUME, n = LOCAL_VOLUME.len() + 1, @@ -1018,11 +1080,11 @@ impl Db { /// drain wants [`next_due_op`](Self::next_due_op) instead. pub fn pending_ops(&self) -> Result> { let conn = self.read(); - let mut stmt = conn.prepare( - "SELECT id, kind, uid, parent_uid, name, blob_path, meta_json, created_at, + let mut stmt = conn.prepare(&format!( + "SELECT id, kind, {UID_NOW}, {PARENT_NOW}, name, blob_path, meta_json, created_at, attempts, last_error, next_attempt_at - FROM pending_op ORDER BY id", - )?; + FROM pending_op ORDER BY id" + ))?; let rows = stmt .query_map([], |r| { Ok(PendingOp { diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs index 5556a0b..580694c 100644 --- a/crates/pdfs-core/src/db/tests.rs +++ b/crates/pdfs-core/src/db/tests.rs @@ -3178,6 +3178,87 @@ fn landing<'a>(local: &'a str, real: &'a str, parent: &'a str, name: &'a str) -> } } +/// A queued op on a node made here, `kind` under `uid`. +fn op_on(kind: &str, uid: &str, parent: Option<&str>) -> PendingOp { + PendingOp { + id: 0, + kind: kind.to_string(), + uid: uid.to_string(), + parent_uid: parent.map(str::to_string), + name: parent.map(|_| "new.txt".to_string()), + blob_path: Some(format!("/staging/{kind}")), + meta_json: Some("{}".to_string()), + created_at: 1, + attempts: 0, + last_error: None, + next_attempt_at: 0, + } +} + +#[test] +fn an_op_queued_under_a_stand_in_after_its_create_landed_goes_to_the_landed_node() { + let db = Db::open_in_memory().unwrap(); + db.upsert_node(&folder("parent", None, "My Files")).unwrap(); + let parent = uid("parent").to_string(); + let lid = db.add_local_row(&parent, "new.txt", false, 1).unwrap(); + let local = local_uid(lid); + let real = uid("real").to_string(); + let (id, _) = db + .enqueue_op(&op_on(OP_CREATE, &local, Some(&parent))) + .unwrap(); + db.claim_next_due_op(1).unwrap().unwrap(); + db.finish_create( + id, + Some("/staging/create"), + &landing(&local, &real, &parent, "new.txt"), + |_| None, + ) + .unwrap(); + + // A handle opened before the create landed still names the stand-in. + db.enqueue_op(&op_on(OP_REVISION, &local, None)).unwrap(); + assert!(db.has_pending_op(&real, OP_REVISION).unwrap()); + assert!(db.has_any_op(&real).unwrap()); + // A second write, made through the real uid, replaces it. + let (_, superseded) = db.enqueue_op(&op_on(OP_REVISION, &real, None)).unwrap(); + assert_eq!(superseded.as_deref(), Some("/staging/revision")); + let ops = db.pending_ops().unwrap(); + assert_eq!(ops.len(), 1); + assert_eq!( + ops[0].uid, real, + "the queue names the node by the uid it has now" + ); + let claimed = db.claim_next_due_op(1).unwrap().unwrap(); + assert_eq!(claimed.uid, real, "the drain sends the real uid"); +} + +#[test] +fn a_node_queued_under_both_its_uids_is_sent_one_op_at_a_time() { + let db = Db::open_in_memory().unwrap(); + db.upsert_node(&folder("parent", None, "My Files")).unwrap(); + let parent = uid("parent").to_string(); + let lid = db.add_local_row(&parent, "new.txt", false, 1).unwrap(); + let local = local_uid(lid); + let real = uid("real").to_string(); + db.land_placeholder_row(&local, &real).unwrap(); + db.enqueue_op(&op_on(OP_RENAME, &real, Some(&parent))) + .unwrap(); + db.enqueue_op(&op_on(OP_REVISION, &local, None)).unwrap(); + + let first = db.claim_next_due_op(1).unwrap().unwrap(); + assert_eq!(first.kind, OP_RENAME); + assert!( + db.claim_next_due_op(1).unwrap().is_none(), + "the write waits for the rename of the same node" + ); + db.delete_op(first.id).unwrap(); + let second = db.claim_next_due_op(1).unwrap().unwrap(); + assert_eq!( + (second.kind.as_str(), second.uid.as_str()), + (OP_REVISION, real.as_str()) + ); +} + #[test] fn a_rename_made_while_its_create_uploads_is_queued_for_the_landed_node() { let db = Db::open_in_memory().unwrap(); @@ -3220,7 +3301,7 @@ fn a_rename_made_while_its_create_uploads_is_queued_for_the_landed_node() { assert_eq!(meta.original_name.as_deref(), Some("new (conflict).txt")); // The folder lands next and the move follows it. - db.remap_local_uid("local~dir", &uid("dir").to_string()) + db.land_placeholder_row("local~dir", &uid("dir").to_string()) .unwrap(); let ops = db.pending_ops().unwrap(); assert_eq!(ops[0].parent_uid, Some(uid("dir").to_string())); @@ -3246,7 +3327,7 @@ fn draining_a_folder_repoints_its_queued_children() { }) .unwrap(); - db.remap_local_uid(local_dir, &real_dir).unwrap(); + db.land_placeholder_row(local_dir, &real_dir).unwrap(); // The child was queued against a folder that did not exist yet. Once the // folder is real, the child must target the server's uid — otherwise the @@ -3466,7 +3547,7 @@ fn an_op_on_a_listed_twin_follows_the_folder_that_lands() { }) .unwrap(); - db.remap_local_uid(&local.to_string(), &uid("real").to_string()) + db.land_placeholder_row(&local.to_string(), &uid("real").to_string()) .unwrap(); let lid: Option = db @@ -3646,7 +3727,7 @@ fn a_landed_placeholder_never_overwrites_the_server_copy() { db.upsert_node(&folder("real", Some("root"), "New folder (1)")) .unwrap(); - db.remap_local_uid(&local.to_string(), &uid("real").to_string()) + db.land_placeholder_row(&local.to_string(), &uid("real").to_string()) .unwrap(); assert!(db.node_by_uid(&local.to_string()).unwrap().is_none()); @@ -3721,7 +3802,7 @@ fn a_landed_placeholder_keeps_its_lid() { db.upsert_node(&listed).unwrap(); } - db.remap_local_uid(&local.to_string(), &real.to_string()) + db.land_placeholder_row(&local.to_string(), &real.to_string()) .unwrap(); assert_eq!(db.lid_of(&local.to_string()).unwrap(), None); @@ -3805,7 +3886,7 @@ fn a_landed_folder_hands_its_children_over_in_the_stored_node_too() { child.parent_uid = Some(local.clone()); db.upsert_node(&child).unwrap(); - db.remap_local_uid(&local.to_string(), &uid("dir").to_string()) + db.land_placeholder_row(&local.to_string(), &uid("dir").to_string()) .unwrap(); let stored = db.node_by_uid(&uid("child").to_string()).unwrap().unwrap(); diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs index cb94744..98b446d 100644 --- a/crates/pdfs-fuse/src/drain.rs +++ b/crates/pdfs-fuse/src/drain.rs @@ -1568,10 +1568,8 @@ impl Core { // B123). let remote = node.clone(); self.stamp_pending_sizes(std::slice::from_mut(&mut node)); - // Repoints queued children and node rows, and readdresses the placeholder row. - // A no-op after `retire_create`, which has already done both. - self.db - .remap_local_uid(&local.to_string(), &real.to_string())?; + // The rows and queued ops already follow the node: `retire_create` + // moved its row to the real uid, and ops find it by local id. self.for_each_state(|st| { if let Some(ino) = st.by_uid.remove(local) { diff --git a/docs/MILESTONE-3.0.0-STATUS.md b/docs/MILESTONE-3.0.0-STATUS.md index a7c8918..4f4dd25 100644 --- a/docs/MILESTONE-3.0.0-STATUS.md +++ b/docs/MILESTONE-3.0.0-STATUS.md @@ -40,11 +40,15 @@ Tracks `docs/MILESTONE-3.0.0.md`. Update it with every commit that moves a phase - [ ] No more placeholders: `nodes.uid` is `NULL` until a node lands. In four parts: - [x] 6a: a node made here goes by `local~`, a function of its row, instead of a minted `local~-`; schema 41 rewrites the old placeholders (nodes, ops, pins). - - [ ] 6b: ops by lid; a landing create rewrites no other op. - - [ ] 6c: a local node's row has `uid NULL`; children and subtrees by `parent_lid`; pins - by lid. + - [x] 6b: an op is found by its node's local id under either uid, the claim keeps one op + per local id on the wire, and the drain sends the uid the row has now; + `remap_local_uid` is gone. + - [ ] 6c: a local node's row has `uid NULL`; children, subtrees and ops' parents by lid, + so a landing create rewrites no other row or op; pins by lid. - [ ] 6d: in memory, `pending`, `hidden` and `creating` by lid; `is_local_uid` checks - become `remote_uid().is_none()`. + become `remote_uid().is_none()`; a write or rename after landing queues under the + stand-in, so `landed_uid`, `follow_landed_create`, `landed_row_uid` and the + `rewrite_op_target` fallback go. - [ ] An executor that runs ops in dependency order, many at once. 3. **Speed (§7)**: run independent ops in parallel, batch trashes and moves; measure 1,000 files and 100 folders on Wi-Fi against LAN (target: within 10 %).