//! The offline mutation queue (`pending_op`): writes the kernel has accepted but //! the remote has not yet seen. Rebuilt into memory on mount and drained in row //! order, so a child never drains before the parent that gives it a real uid. use rusqlite::{OptionalExtension, params}; use serde::{Deserialize, Serialize}; use super::Db; use crate::Result; /// The `kind` of a [`PendingOp`] that uploads a staged file as a new revision. pub const OP_REVISION: &str = "revision"; /// The `kind` of a [`PendingOp`] that creates a file that so far exists only /// locally, under a `local:` placeholder uid (offline.md Phase 3b). /// /// The written bytes ride along on the same row rather than as a follow-on /// [`OP_REVISION`]: draining the create mints the node's real uid, which would /// leave a separate revision op addressed to a uid that no longer exists. pub const OP_CREATE: &str = "create"; /// The `kind` of a [`PendingOp`] that creates a folder that so far exists only /// locally. Ordering matters: a child's op is queued after its parent's and /// [`Db::pending_ops`] replays by row id, so the parent has a real uid by the /// time the child drains. pub const OP_MKDIR: &str = "mkdir"; /// The `kind` of a [`PendingOp`] that gives a node a new name, a new parent, or /// both — the queued form of `mv` (offline.md Phase 3b). /// /// `parent_uid` and `name` hold the node's *desired end state*, not a delta, so /// a second rename simply replaces the row (see [`Db::enqueue_op`]) and the /// drain can compare them against the remote and skip whichever half already /// matches. The parent may be a `local~` placeholder — moving a file into a /// folder that was itself created offline — in which case it is rewritten by /// that folder's own drain, exactly as for [`OP_CREATE`]. pub const OP_RENAME: &str = "rename"; /// Drain-time authority retained by a queued rename. /// /// The optimistic local move rewrites the node's persisted ancestry to the /// destination immediately. Keeping the original parent here prevents that /// rewrite from erasing the shared tree that admitted the move. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct RenameMeta { pub original_parent_uid: String, } /// The `kind` of a [`PendingOp`] that trashes a node the server knows about /// (offline.md Phase 3b). /// /// A node that only ever existed locally never gets one of these: there is /// nothing to trash remotely, so deleting it just drops its queued ops /// ([`Db::delete_ops_for_uid`]). pub const OP_TRASH: &str = "trash"; /// Whether at most one op of this `kind` may be queued per node, so that a newer /// one replaces the older rather than queueing behind it. /// /// True exactly of the kinds that describe a node's desired *end state* rather /// than a step towards it: the newest revision already contains every earlier /// one's bytes, the newest name is the only name wanted, and a trash subsumes /// anything queued before it. A `create`/`mkdir` is the opposite — it is the one /// thing that will ever make the node exist, so it must never be replaced. pub fn op_supersedes(kind: &str) -> bool { matches!(kind, OP_REVISION | OP_RENAME | OP_TRASH) } /// 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~` 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"; /// 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 /// editor's `*.swp` — so its bytes never upload while it wears that name. A /// rename to the finished name un-parks it (see [`Db::set_create_hold`]), which /// is the only moment the completed file is meant to reach Drive. Without the /// park, every partial and every abandoned temp would upload and, on a /// stall+resume, fork a `(sync-conflict)` copy (docs/BUGS.md B70). pub const PARK_UNTIL: i64 = 8_000_000_000_000; /// A mutation that has been accepted locally but not yet performed against the /// API — the durable half of the write-back queue (offline.md Phase 3). /// /// The daemon answers the FUSE call the moment this row and its staged blob are /// on disk, so a `cp` into the mount runs at disk speed and the upload happens /// behind it. That also makes an offline write succeed rather than EIO: the row /// simply waits for the network. #[derive(Debug, Clone)] pub struct PendingOp { /// Row id, `0` on a value being inserted. pub id: i64, /// See [`OP_REVISION`], [`OP_CREATE`], [`OP_MKDIR`]. pub kind: String, /// Node this op targets. For [`OP_CREATE`]/[`OP_MKDIR`] this is the /// `local~` placeholder the node is known by until it drains. pub uid: String, /// Where the new node goes; only set for [`OP_CREATE`]/[`OP_MKDIR`]. May /// itself be a placeholder when the parent folder is also still queued, in /// which case it is rewritten when the parent drains. pub parent_uid: Option, /// The new node's name; only set for [`OP_CREATE`]/[`OP_MKDIR`]. pub name: Option, /// Staged blob holding the bytes to upload. pub blob_path: Option, /// Serialized [`StagedWrite`](crate::cache::StagedWrite). pub meta_json: Option, /// When the op was queued (ms since epoch). pub created_at: i64, pub attempts: i64, pub last_error: Option, /// Earliest ms at which to retry, for backoff. pub next_attempt_at: i64, } /// How much work the queue owes the server, by whether it carries bytes. #[derive(Debug, Clone, Copy, Default)] pub struct PendingCounts { /// Queued `create`/`revision` ops: files whose content is not on the remote. pub uploads: i64, /// Queued `mkdir`/`rename`/`trash` ops: metadata the remote has not been /// told about. pub changes: i64, } /// The outcome of folding freshly written bytes into a queued create. #[derive(Debug, Clone)] pub struct AttachedBlob { /// Row id of the create the bytes were attached to. pub id: i64, /// Blob the create held before, now orphaned. pub superseded: Option, } 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.conn.lock(); conn.query_row( "SELECT EXISTS( SELECT 1 FROM pending_op WHERE uid = ?1 AND kind = ?2 )", params![uid, kind], |row| row.get(0), ) .map_err(Into::into) } /// Read an existing operation's metadata before a superseding enqueue. pub fn pending_op_meta(&self, uid: &str, kind: &str) -> Result> { let conn = self.conn.lock(); conn.query_row( "SELECT meta_json FROM pending_op WHERE uid = ?1 AND kind = ?2", params![uid, kind], |row| row.get(0), ) .optional() .map(|value| value.flatten()) .map_err(Into::into) } pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option)> { let mut conn = self.conn.lock(); let tx = conn.transaction()?; let superseded: Option = if op_supersedes(&op.kind) { let blob: Option = tx .query_row( "SELECT blob_path FROM pending_op WHERE uid = ?1 AND kind = ?2", params![op.uid, op.kind], |r| r.get(0), ) .optional()? .flatten(); tx.execute( "DELETE FROM pending_op WHERE uid = ?1 AND kind = ?2", params![op.uid, op.kind], )?; blob } else { None }; tx.execute( "INSERT INTO pending_op (kind, uid, parent_uid, name, blob_path, meta_json, created_at, next_attempt_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", params![ op.kind, op.uid, op.parent_uid, op.name, op.blob_path, op.meta_json, op.created_at, op.next_attempt_at ], )?; let id = tx.last_insert_rowid(); tx.commit()?; Ok((id, superseded)) } /// Atomically replace every queued operation for `uid` and its queued /// descendants with one remote trash intent. /// /// Staged blob paths are returned only after the transaction commits. A /// failed trash insert therefore rolls the deletions back and leaves every /// prior operation owning its bytes. pub fn replace_ops_with_trash( &self, uid: &str, name: &str, created_at: i64, ) -> Result<(i64, Vec)> { const SUBTREE: &str = " WITH RECURSIVE doomed(uid) AS ( SELECT ?1 UNION SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid )"; let mut conn = self.conn.lock(); let tx = conn.transaction()?; let blobs: Vec = { let mut stmt = tx.prepare(&format!( "{SUBTREE} SELECT blob_path FROM pending_op WHERE uid IN (SELECT uid FROM doomed) AND blob_path IS NOT NULL" ))?; let rows = stmt.query_map(params![uid], |row| row.get::<_, String>(0))?; rows.collect::, _>>()? }; tx.execute( &format!("{SUBTREE} DELETE FROM pending_op WHERE uid IN (SELECT uid FROM doomed)"), params![uid], )?; tx.execute( "INSERT INTO pending_op (kind, uid, parent_uid, name, blob_path, meta_json, created_at, next_attempt_at) VALUES (?1, ?2, NULL, ?3, NULL, NULL, ?4, 0)", params![OP_TRASH, uid, name, created_at], )?; let id = tx.last_insert_rowid(); tx.commit()?; Ok((id, blobs)) } /// Point a queued create at the bytes that were just written to it, returning /// any blob it previously held so the caller can discard it. /// /// This is what `release` does for a file that only exists locally: the node /// has no uid to hang a revision op on yet, so the bytes ride on the create. /// Repeated writes before the drain simply replace the blob. /// /// Returns `Ok(None)` and touches nothing if the create has already drained — /// the caller must then queue an ordinary revision against the real uid. pub fn attach_blob_to_create( &self, uid: &str, blob_path: &str, meta_json: &str, ) -> Result> { let conn = self.conn.lock(); 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], |r| Ok((r.get(0)?, r.get(1)?)), ) .optional()?; let Some((id, superseded)) = existing else { return Ok(None); }; // Clearing the backoff lets the fresh bytes be tried promptly — except // when the create is *parked* (a transient file, next_attempt_at at the // PARK_UNTIL sentinel). Attaching a new revision to a still-growing // download must not wake it; only the finalize rename does. So preserve a // park and reset an ordinary backoff. conn.execute( "UPDATE pending_op SET blob_path = ?2, meta_json = ?3, attempts = 0, next_attempt_at = CASE WHEN next_attempt_at >= ?4 THEN next_attempt_at ELSE 0 END WHERE id = ?1", params![id, blob_path, meta_json, PARK_UNTIL], )?; Ok(Some(AttachedBlob { id, superseded })) } /// Replace the sidecar of a queued op, for a baseline that has moved under /// it. /// /// Exists for one caller: when *our own* upload seals a new revision while a /// further write to the same node is already queued, that queued write's /// [`StagedWrite::based_on`](crate::cache::StagedWrite::based_on) still /// names the revision we just replaced. Left alone it would read as "another /// device changed this file" and divert the write into a conflict copy over /// nothing — a self-conflict. Restamping it is what keeps a chain of writes /// to one file a chain rather than a pile of copies. /// /// Deliberately narrow: only the sidecar changes, so the blob, the attempt /// count and the backoff all survive. Returns false when no such op is /// 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 n = conn.execute( "UPDATE pending_op SET meta_json = ?3 WHERE uid = ?1 AND kind = ?2", params![uid, kind, 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. /// /// A `local~` uid means nothing to the API, so there is no rename call to /// make — the node is still only a queued intent, and rewriting that intent /// *is* the rename. Returns false when the create has already drained, in /// which case the node has a real uid and the caller must rename it there /// instead (offline.md Phase 3b). pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result { let conn = self.conn.lock(); let n = conn.execute( "UPDATE pending_op SET parent_uid = ?2, name = ?3 WHERE uid = ?1 AND kind IN (?4, ?5)", params![uid, parent_uid, name, OP_CREATE, OP_MKDIR], )?; Ok(n > 0) } /// Park or un-park a queued create/mkdir for `uid`, returning whether a row /// was touched. /// /// Parking sets `next_attempt_at` to [`PARK_UNTIL`] so [`Db::next_due_op`] /// skips it indefinitely; un-parking sets it due now. Used to keep a /// transient file's bytes off Drive until it is renamed to its finished name /// (docs/BUGS.md B70). Un-parking a create that was never parked is harmless: /// it just makes the create due immediately, which for a normal local node it /// already is. pub fn set_create_hold(&self, uid: &str, held: bool) -> Result { let conn = self.conn.lock(); 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], )?; Ok(n > 0) } /// Check if a queued create or mkdir op exists for `uid`. pub fn has_create_op(&self, uid: &str) -> Result { let conn = self.conn.lock(); 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], |row| row.get(0), )?; Ok(count > 0) } /// Check whether *any* op of any kind is queued against `uid`. /// /// Used as a safety interlock by callers that are about to remove a node /// they only know about from a stale snapshot: a queued op means the user's /// bytes are still owed an upload, and dropping it would discard the staged /// blob that holds them (`docs/BUGS.md` B71). pub fn has_any_op(&self, uid: &str) -> Result { let conn = self.conn.lock(); let count: i64 = conn.query_row( "SELECT COUNT(*) FROM pending_op WHERE uid = ?1", params![uid], |row| row.get(0), )?; Ok(count > 0) } /// Drop every op targeting a node **or anything queued beneath it**, /// returning the staged blobs they held so the caller can delete them. /// /// Used when a node that only ever existed locally is deleted: there is /// nothing on the server to trash, so the queued work simply stops being /// wanted. /// /// The descent is what keeps the queue alive. Deleting a folder created /// offline drops the `mkdir` that would have given it a real uid, and any /// op still queued under that placeholder is then unreachable forever: it /// can never be attempted (its parent is a `local~` uid, so `op_is_ready` /// refuses it) and nothing is left to remap it. It would sit in the queue, /// and in the user's pending count, for the life of the database. /// /// Only `create`/`mkdir` ops carry a `parent_uid`, so for a file this /// recursion finds nothing and costs one query. pub fn delete_ops_for_uid(&self, uid: &str) -> Result> { const SUBTREE: &str = " WITH RECURSIVE doomed(uid) AS ( SELECT ?1 UNION SELECT p.uid FROM pending_op p JOIN doomed d ON p.parent_uid = d.uid )"; let mut conn = self.conn.lock(); let tx = conn.transaction()?; let blobs: Vec = { let mut stmt = tx.prepare(&format!( "{SUBTREE} SELECT blob_path FROM pending_op WHERE uid IN (SELECT uid FROM doomed) AND blob_path IS NOT NULL" ))?; let rows = stmt.query_map(params![uid], |r| r.get::<_, String>(0))?; rows.collect::, _>>()? }; tx.execute( &format!("{SUBTREE} DELETE FROM pending_op WHERE uid IN (SELECT uid FROM doomed)"), params![uid], )?; tx.commit()?; 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. 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], )?; tx.execute( "UPDATE nodes SET parent_uid = ?2 WHERE parent_uid = ?1", params![local, real], )?; tx.execute("DELETE FROM nodes WHERE uid = ?1", params![local])?; tx.commit()?; Ok(()) } /// Every queued op, oldest first. The drain worker replays them in this order /// so a file's writes land in the order they were made. /// The oldest queued op that is due and not blocked, or `None`. /// /// This is the drain loop's per-iteration query, and it is deliberately not /// `pending_ops().find(...)`. That variant read every row — including each /// op's full `meta_json` — sorted them, and threw all but one away, once per /// drained op *and* once per retry, while holding the single connection /// mutex that every FUSE `lookup` also needs. A queue of a few thousand ops /// (an ordinary offline `cp -r`) made draining quadratic in the queue length. /// /// Both filters are in SQL so the row is decided by the database rather than /// 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; /// this matters when the parent is backing off, where the child must wait /// rather than burn its own retries. /// /// `id` is `INTEGER PRIMARY KEY`, i.e. the rowid, so `ORDER BY id LIMIT 1` /// walks the table in insertion order and stops at the first match — no /// sort, and no separate index to maintain. The common case (something is /// due) exits on the first row. pub fn next_due_op(&self, now: i64) -> Result> { let conn = self.conn.lock(); let mut stmt = conn.prepare_cached(&format!( "SELECT id, kind, uid, parent_uid, 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}~') ORDER BY id LIMIT 1", v = LOCAL_VOLUME, n = LOCAL_VOLUME.len() + 1, ))?; let op = stmt .query_row(params![now], |r| { Ok(PendingOp { id: r.get(0)?, kind: r.get(1)?, uid: r.get(2)?, parent_uid: r.get(3)?, name: r.get(4)?, blob_path: r.get(5)?, meta_json: r.get(6)?, created_at: r.get(7)?, attempts: r.get(8)?, last_error: r.get(9)?, next_attempt_at: r.get(10)?, }) }) .optional()?; Ok(op) } /// The earliest `next_attempt_at` among all queued ops, or `None` if the /// queue is empty. Used by the drain loop to sleep exactly until the next /// debounced or backed-off op becomes eligible rather than waiting the full /// idle-poll interval. pub fn earliest_due_at(&self) -> Result> { let conn = self.conn.lock(); let ts: Option = conn .query_row( &format!( "SELECT MIN(next_attempt_at) FROM pending_op \ WHERE parent_uid IS NULL OR substr(parent_uid, 1, {n}) <> '{v}~'", v = LOCAL_VOLUME, n = LOCAL_VOLUME.len() + 1, ), [], |r| r.get(0), ) .optional()? .flatten(); Ok(ts) } /// Every queued op, oldest first. For status read-outs and the CLI — the /// drain wants [`next_due_op`](Self::next_due_op) instead. pub fn pending_ops(&self) -> Result> { let conn = self.conn.lock(); let mut stmt = conn.prepare( "SELECT id, kind, uid, parent_uid, name, blob_path, meta_json, created_at, attempts, last_error, next_attempt_at FROM pending_op ORDER BY id", )?; let rows = stmt .query_map([], |r| { Ok(PendingOp { id: r.get(0)?, kind: r.get(1)?, uid: r.get(2)?, parent_uid: r.get(3)?, name: r.get(4)?, blob_path: r.get(5)?, meta_json: r.get(6)?, created_at: r.get(7)?, attempts: r.get(8)?, last_error: r.get(9)?, next_attempt_at: r.get(10)?, }) })? .collect::, _>>()?; Ok(rows) } /// How many queued ops there are of each kind, for `Response::Status`. /// /// Split because "3 uploads queued" has to mean three files whose bytes are /// not on the remote yet. A queued `mkdir`/`rename`/`trash` is also work the /// mount owes the server, but it carries no bytes and reporting it as an /// upload is simply untrue. pub fn pending_op_counts(&self) -> Result { let conn = self.conn.lock(); let uploads = conn.query_row( "SELECT COUNT(*) FROM pending_op WHERE kind IN (?1, ?2)", params![OP_REVISION, OP_CREATE], |r| r.get(0), )?; let changes = conn.query_row( "SELECT COUNT(*) FROM pending_op WHERE kind NOT IN (?1, ?2)", params![OP_REVISION, OP_CREATE], |r| r.get(0), )?; Ok(PendingCounts { uploads, changes }) } /// 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"; /// Drop a queued op, once its upload has actually landed. pub fn delete_op(&self, id: i64) -> Result<()> { let conn = self.conn.lock(); conn.execute("DELETE FROM pending_op WHERE id = ?1", params![id])?; Ok(()) } /// Record a failed attempt and when to next try. Leaves the row in place — /// the staged bytes are still the only copy of the user's write. pub fn record_op_failure(&self, id: i64, error: &str, next_attempt_at: i64) -> Result<()> { let conn = self.conn.lock(); conn.execute( "UPDATE pending_op SET attempts = attempts + 1, last_error = ?2, next_attempt_at = ?3 WHERE id = ?1", params![id, error, next_attempt_at], )?; Ok(()) } /// Defer a due operation without recording a failed remote attempt. /// /// Access checks happen before the drain calls Proton, so a locally denied /// operation has not failed remotely and must not consume a retry. Preserve /// the long-lived transient-create sentinel if this is ever called for one. pub fn defer_op_without_attempt(&self, id: i64, next_attempt_at: i64) -> Result<()> { let conn = self.conn.lock(); conn.execute( "UPDATE pending_op SET next_attempt_at = CASE WHEN next_attempt_at >= ?3 THEN next_attempt_at ELSE ?2 END WHERE id = ?1", params![id, next_attempt_at, PARK_UNTIL], )?; Ok(()) } }