proton-drive-linux/crates/pdfs-core/src/db/ops.rs

//! 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 std::collections::HashMap;

use rusqlite::{OptionalExtension, params};
use serde::{Deserialize, Serialize};

use super::Db;
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.
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.
///
/// 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`.
///
/// `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 node's name on Drive when the first of its queued renames was made,
    /// kept across the renames that replace the row, like the parent. With it
    /// the drain knows where the node should still be when the op lands.
    /// `None` on a row queued before this field existed.
    #[serde(default)]
    pub original_name: Option<String>,
}

/// The `kind` of a [`PendingOp`] that trashes a node the server knows about.
///
/// 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)
}

/// Delete the ops that removing `uid` takes with it, returning the staged blobs
/// they held.
///
/// That is every op for `uid` or for anything queued beneath it, plus the
/// revisions of files the node tree places beneath it. A revision op carries
/// no `parent_uid`, so the queue alone cannot tell that its file sits inside
/// a removed folder: the upload outlived the folder, failed forever against a
/// trashed node, and kept its blob on disk (`docs/BUGS.md` B102).
///
/// "Queued beneath it" starts from every folder the node tree places there,
/// not only from `uid`. A create in a subfolder that already exists on Drive
/// names that subfolder as its parent, so following the queue from `uid` alone
/// missed it: a parked one kept its blob for good, and a due one would have
/// been made in the root once its folder turned out trashed (`docs/BUGS.md`
/// B116).
///
/// Other ops are taken from the node tree only for revisions. A queued rename
/// that moved a node *into* the folder has to land, or the node would stay
/// where it was on the server instead of going to the trash with the folder.
///
/// A create already on the wire cannot be stopped, so its row is not deleted
/// but turned into a trash of the node it is making, still claimed by the
/// upload. [`Db::finish_create`] points it at the landed node. Until then it
/// holds the name, as any queued trash does ([`Db::has_pending_trash_named`]):
/// a rename over the file sent first found the name taken and landed under a
/// conflict name (`docs/BUGS.md` B129).
///
/// So does a create that was sent before and failed. Its answer may have been
/// lost after Drive made the node, which would then keep the deleted file's
/// name and bytes (`docs/BUGS.md` B151). Such a trash keeps the create's
/// parent and blob, which the drain looks for the node by, and which it
/// discards once the trash is done ([`Db::complete_trash_op`]).
///
/// Sent is where it was sent to ([`Db::note_create_target`]), not only its
/// attempts: a write after the failure clears those for a prompt retry, and
/// the delete then dropped the create whole (`docs/BUGS.md` B191).
fn drop_doomed_ops(tx: &rusqlite::Transaction<'_>, uid: &str) -> Result<Vec<String>> {
    const DOOMED: &str = "
        WITH RECURSIVE
          below(uid, lid) AS (
            SELECT ?1, ?2
            UNION
            SELECT n.uid, n.lid FROM nodes n JOIN below b
              ON n.parent_lid = b.lid OR (n.parent_lid IS NULL AND n.parent_uid = b.uid)
          ),
          queued(uid, lid) AS (
            SELECT ?1, ?2
            UNION
            SELECT p.uid, p.lid FROM pending_op p
            WHERE p.kind IN ('create', 'mkdir')
              AND (p.parent_lid IN (SELECT lid FROM below)
                   OR p.parent_uid IN (SELECT uid FROM below))
            UNION
            SELECT p.uid, p.lid FROM pending_op p JOIN queued q
              ON p.parent_lid = q.lid OR p.parent_uid = q.uid
          ),
          doomed(id) AS (
            SELECT id FROM pending_op
            WHERE uid IN (SELECT uid FROM queued)
               OR lid IN (SELECT lid FROM queued)
               OR (kind = 'revision'
                   AND (uid IN (SELECT uid FROM below) OR lid IN (SELECT lid FROM below)))
          )";
    const SENT: &str = "kind IN ('create', 'mkdir')
        AND (claimed_at <> 0 OR attempts > 0 OR sent_to IS NOT NULL)";
    let lid = node_lid(tx, uid)?;
    let blobs: Vec<String> = {
        let mut stmt = tx.prepare(&format!(
            "{DOOMED}
             SELECT blob_path FROM pending_op
             WHERE id IN (SELECT id FROM doomed) AND blob_path IS NOT NULL
               AND NOT ({SENT})"
        ))?;
        let rows = stmt.query_map(params![uid, lid], |r| r.get::<_, String>(0))?;
        rows.collect::<std::result::Result<Vec<_>, _>>()?
    };
    tx.execute(
        &format!(
            "{DOOMED}
             UPDATE pending_op
             SET kind = 'trash', issue = NULL,
                 attempts = 0, last_error = NULL, next_attempt_at = 0,
                 parent_uid = CASE WHEN {PARENT_NOW} LIKE '{LOCAL_VOLUME}~%' THEN NULL
                                   ELSE {PARENT_NOW} END,
                 parent_lid = CASE WHEN {PARENT_NOW} LIKE '{LOCAL_VOLUME}~%' THEN NULL
                                   ELSE parent_lid END
             WHERE id IN (SELECT id FROM doomed) AND {SENT}"
        ),
        params![uid, lid],
    )?;
    tx.execute(
        &format!(
            "{DOOMED}
             DELETE FROM pending_op
             WHERE id IN (SELECT id FROM doomed)
               AND NOT (kind = 'trash' AND uid LIKE '{LOCAL_VOLUME}~%')"
        ),
        params![uid, lid],
    )?;
    Ok(blobs)
}

/// The `pending_op` rows, `?1` being [`OP_RENAME`], that move a node away from
/// the name `?2`.
const MOVES_FROM: &str = "kind = ?1
     AND json_extract(meta_json, '$.original_name') = ?2
     AND (name <> ?2 OR parent_uid <> json_extract(meta_json, '$.original_parent_uid'))";

/// [`Db::wake_ops_waiting_for`] within a transaction.
pub(super) fn wake_ops_waiting_for_tx(tx: &rusqlite::Transaction<'_>, name: &str) -> Result<()> {
    tx.execute(
        &format!(
            "UPDATE pending_op SET next_attempt_at = 0
             WHERE name = ?2 AND kind IN (?3, ?4, ?1) AND attempts > 0
               AND next_attempt_at < ?5
               AND NOT EXISTS (SELECT 1 FROM pending_op WHERE kind = ?6 AND name = ?2)
               AND NOT EXISTS (SELECT 1 FROM pending_op WHERE {MOVES_FROM})"
        ),
        params![OP_RENAME, name, OP_CREATE, OP_MKDIR, PARK_UNTIL, OP_TRASH],
    )?;
    Ok(())
}

/// Point queued op `id` at its node and its parent by local id, from the uids
/// 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<()> {
    let (uid, parent): (String, Option<String>) = 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}"))
}

/// Delete op `id` if it is a rename with a newer rename of the same node queued
/// behind it, which says where the node goes now. Returns whether it did.
fn drop_replaced_rename(conn: &rusqlite::Connection, id: i64) -> Result<bool> {
    let dropped = conn.execute(
        "DELETE FROM pending_op
         WHERE id = ?1 AND kind = ?2
           AND EXISTS (SELECT 1 FROM pending_op AS newer
                       WHERE (newer.uid = pending_op.uid OR newer.lid = pending_op.lid)
                         AND newer.kind = ?2 AND newer.id > ?1)",
        params![id, OP_RENAME],
    )?;
    Ok(dropped > 0)
}

/// 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.
const PARENT_NOW: &str =
    "COALESCE((SELECT uid FROM nodes WHERE lid = pending_op.parent_lid), parent_uid)";

/// The `WHERE` terms for a row a drain worker may claim once it falls due: not
/// claimed already, no other claimed op on the same node, and not under a
/// folder that has yet to land.
///
/// One fragment for [`Db::claim_next_due_op`] and [`Db::earliest_due_at`]. A
/// row the claim skips but the wait counts is due "now" for as long as it is
/// held, and every idle worker spun on it (`docs/BUGS.md` B200).
fn claimable() -> String {
    format!(
        "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))",
        v = LOCAL_VOLUME,
        n = LOCAL_VOLUME.len() + 1,
    )
}

/// 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~<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
/// 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;

/// How long a create may stay parked before the park is treated as permanent
/// and the bytes are let through anyway (ms).
///
/// The park has exactly one exit: a rename from the transient name to the
/// finished one. Every writer that reaches that rename does so in seconds — an
/// atomic-write scratch file immediately, a download when it completes. A park
/// still standing an hour later belongs to a writer that will never perform
/// that rename: an editor that deletes its swap file instead, a download the
/// user abandoned, or a file that simply *is* named `report.tmp` and is never
/// going to be called anything else.
///
/// Before this existed such a row sat at [`PARK_UNTIL`] for the life of the
/// database, holding the only copy of the file's bytes in `staging/` and
/// counting against the queue while promising progress that could not come. The
/// sweep un-parks it instead of dropping it: bytes the user can see in the mount
/// are bytes the user expects on Drive, and invariant 1 of the drain (a staged
/// blob is dropped only once its op has landed) admits no other answer.
pub const PARK_EXPIRY_MS: i64 = 60 * 60 * 1000;

/// A mutation that has been accepted locally but not yet performed against the
/// API — the durable half of the write-back queue.
///
/// 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~<uuid>` 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<String>,
    /// The new node's name; only set for [`OP_CREATE`]/[`OP_MKDIR`].
    pub name: Option<String>,
    /// Staged blob holding the bytes to upload.
    pub blob_path: Option<String>,
    /// Serialized [`StagedWrite`](crate::cache::StagedWrite).
    pub meta_json: Option<String>,
    /// When the op was queued (ms since epoch).
    pub created_at: i64,
    pub attempts: i64,
    pub last_error: Option<String>,
    /// 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, 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,
    /// Ops parked at [`PARK_UNTIL`]: a transient file's bytes, held back until
    /// the rename that gives it its finished name. They are queued but will
    /// never drain on their own, so counting them as "waiting to upload" would
    /// have the front-end promise progress that is not coming.
    pub parked: i64,
    /// Ops that have failed at least [`FAILING_ATTEMPTS`] times, or that Drive
    /// refused ([`Db::set_op_issue`]). A permanently
    /// failing op never wedges the queue — the backoff sees to that — which is
    /// exactly why it needs surfacing: it retries forever, invisibly.
    pub failing: i64,
    /// The most recent error text from a failing op, for the front-end to show
    /// instead of the bare count.
    pub last_error: Option<String>,
}

/// Failed attempts after which an op is reported as failing rather than merely
/// retrying. Six is where the backoff has reached its ceiling, so an op past it
/// is retrying at the slowest rate it ever will and has been for minutes.
pub const FAILING_ATTEMPTS: i64 = 6;

/// A queued create that has landed, for [`Db::finish_create`].
#[derive(Debug, Clone, Copy)]
pub struct CreateLanding<'a> {
    /// The placeholder the node was known by.
    pub local: &'a str,
    /// The uid Drive gave it.
    pub real: &'a str,
    /// The parent and name the op held when the drain took it.
    pub sent: (&'a str, &'a str),
    /// The parent and name the node has on Drive. They differ from `sent`
    /// when the name was taken or the folder was gone.
    pub landed: (&'a str, &'a str),
}

/// What became of a queued create when it landed, from [`Db::finish_create`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CreateRetired {
    /// Its row is gone and nothing of it is left queued.
    Landed,
    /// A newer write attached to it on the wire and is now queued as a
    /// revision of the real node: its blob and its rewritten sidecar.
    Newer { blob: String, meta: String },
    /// Its row had been dropped while it was on the wire: the node was deleted,
    /// or a rename replaced it. The node Drive just made is unwanted, and a
    /// trash of it is queued under this id, claimed by the caller.
    Withdrawn { trash: i64 },
}

struct CreateRow {
    blob: Option<String>,
    meta: Option<String>,
    parent_uid: Option<String>,
    name: Option<String>,
    created_at: 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<String>,
}

impl Db {
    /// Whether a queued trash still holds `name` somewhere.
    ///
    /// A replacing rename or a delete-then-recreate queues the trash of the old
    /// node ahead of the op that wants its name. With several drain workers the
    /// later op can run first and find the name still taken; this tells it to
    /// wait for the trash rather than land under a conflict name.
    pub fn has_pending_trash_named(&self, name: &str) -> Result<bool> {
        let conn = self.read();
        conn.query_row(
            "SELECT EXISTS(
                SELECT 1 FROM pending_op WHERE kind = ?1 AND name = ?2
             )",
            params![OP_TRASH, name],
            |row| row.get(0),
        )
        .map_err(Into::into)
    }

    /// Whether the trash of a create a delete withdrew after it was sent is
    /// queued for `name` in `parent_uid` ([`drop_doomed_ops`]).
    ///
    /// The drain looks for the node that create may have made by its folder
    /// and name, so a create of that name there waits for the trash: sent
    /// first, it made a node the trash took for the withdrawn one.
    pub fn has_withdrawn_create(&self, parent_uid: &str, name: &str) -> Result<bool> {
        let conn = self.read();
        conn.query_row(
            &format!(
                "SELECT EXISTS(
                    SELECT 1 FROM pending_op
                    WHERE kind = ?1 AND {PARENT_NOW} = ?2 AND name = ?3
                      AND uid LIKE '{LOCAL_VOLUME}~%'
                 )"
            ),
            params![OP_TRASH, parent_uid, name],
            |row| row.get(0),
        )
        .map_err(Into::into)
    }

    /// Whether a queued rename moves a node away from `name`.
    ///
    /// Like a queued trash ([`Db::has_pending_trash_named`]) it frees the name
    /// once it lands, so an op that wants the name can wait for it. A create
    /// renamed while it was on the wire queues one (`Db::finish_create`).
    pub fn has_pending_move_from(&self, name: &str) -> Result<bool> {
        let conn = self.read();
        conn.query_row(
            &format!("SELECT EXISTS(SELECT 1 FROM pending_op WHERE {MOVES_FROM})"),
            params![OP_RENAME, name],
            |row| row.get(0),
        )
        .map_err(Into::into)
    }

    /// Make the ops that failed because `name` was held due at once, now that
    /// what held it let go. Nothing changes while another queued trash or
    /// rename still holds it, or for a parked op.
    pub fn wake_ops_waiting_for(&self, name: &str) -> Result<()> {
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        wake_ops_waiting_for_tx(&tx, name)?;
        tx.commit()?;
        Ok(())
    }

    /// Whether a specific desired-state operation is still queued for a node.
    pub fn has_pending_op(&self, uid: &str, kind: &str) -> Result<bool> {
        let conn = self.read();
        let lid = node_lid(&conn, uid)?;
        conn.query_row(
            &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)
    }

    /// Read an existing operation's metadata before a superseding enqueue.
    pub fn pending_op_meta(&self, uid: &str, kind: &str) -> Result<Option<String>> {
        let conn = self.read();
        let lid = node_lid(&conn, uid)?;
        conn.query_row(
            &format!(
                "SELECT meta_json FROM pending_op WHERE {} AND kind = ?2",
                on_node(3)
            ),
            params![uid, kind, lid],
            |row| row.get(0),
        )
        .optional()
        .map(|value| value.flatten())
        .map_err(Into::into)
    }

    /// Queue `op`, returning its id and the staged blob of the op it replaced.
    ///
    /// A rename a drain worker holds is not replaced but queued behind: the
    /// claim keeps the newer one from being sent until it retires, so the two
    /// land in the order they were made (`docs/BUGS.md` B141). Should it fail,
    /// [`Db::record_op_failure`] or [`Db::release_op_claim`] drops it.
    pub fn enqueue_op(&self, op: &PendingOp) -> Result<(i64, Option<String>)> {
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        let superseded: Option<String> = 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<String> = tx
                .query_row(
                    &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(
                &format!("DELETE FROM pending_op WHERE {same}"),
                params![op.uid, op.kind, OP_RENAME, lid],
            )?;
            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();
        fill_op_lids(&tx, id)?;
        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<String>)> {
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        let blobs = drop_doomed_ops(&tx, 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();
        fill_op_lids(&tx, id)?;
        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<Option<AttachedBlob>> {
        let conn = self.conn.lock();
        let lid = node_lid(&conn, uid)?;
        let existing: Option<(i64, Option<String>)> = conn
            .query_row(
                &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()?;
        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 }))
    }

    /// Retire a queued create that has just landed as `real`, and repoint
    /// everything that named its placeholder, in one transaction.
    ///
    /// The drain uploads the blob the create held when it was picked up. A
    /// write that closes while that upload is on the wire attaches a newer blob
    /// to the *same row* (`attach_blob_to_create`), so deleting the row by id
    /// threw that write away: the server kept the older bytes and nothing was
    /// left queued to replace them. When the row no longer holds `uploaded`, it
    /// becomes a revision of `real` instead, with its sidecar passed through
    /// `rewrite` so it names the real node. Returns that blob and sidecar, or
    /// `None` when the create is simply done.
    ///
    /// A rename while the create is on the wire rewrites the row's target
    /// ([`Db::rewrite_op_target`]), which the create no longer reads. When the
    /// row names another parent or name than `landing.sent`, a rename of `real`
    /// to that target is queued in the same transaction, or the move would be
    /// lost with the row.
    ///
    /// A delete while the create is on the wire cannot stop the upload. It
    /// turns the row into a trash ([`Db::delete_ops_for_uid`]), which is
    /// pointed at `real` here, still claimed by the caller, so the node is not
    /// left on Drive holding the name (`docs/BUGS.md` B129). A row gone
    /// altogether is queued as such a trash. The placeholder row is left to
    /// the delete that withdrew it.
    pub fn finish_create(
        &self,
        id: i64,
        uploaded: Option<&str>,
        landing: &CreateLanding<'_>,
        rewrite: impl FnOnce(&str) -> Option<String>,
    ) -> Result<CreateRetired> {
        let (local, real) = (landing.local, landing.real);
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        let row: Option<CreateRow> = tx
            .query_row(
                &format!(
                    "SELECT blob_path, meta_json, {PARENT_NOW}, name, created_at
                     FROM pending_op WHERE id = ?1 AND kind <> ?2"
                ),
                params![id, OP_TRASH],
                |r| {
                    Ok(CreateRow {
                        blob: r.get(0)?,
                        meta: r.get(1)?,
                        parent_uid: r.get(2)?,
                        name: r.get(3)?,
                        created_at: r.get(4)?,
                    })
                },
            )
            .optional()?;
        let withdrawn = tx.execute(
            "UPDATE pending_op SET uid = ?2, name = ?3 WHERE id = ?1 AND kind = ?4",
            params![id, real, landing.landed.1, OP_TRASH],
        )?;
        if withdrawn > 0 {
            tx.commit()?;
            return Ok(CreateRetired::Withdrawn { trash: id });
        }
        if row.is_none() {
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_or(1, |d| d.as_millis() as i64);
            tx.execute(
                "INSERT INTO pending_op
                   (kind, uid, parent_uid, name, created_at, next_attempt_at, claimed_at)
                 VALUES (?1, ?2, NULL, ?3, ?4, 0, ?4)",
                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 });
        }
        let newer = match &row {
            Some(CreateRow {
                blob: Some(blob),
                meta: Some(meta),
                ..
            }) if uploaded != Some(blob.as_str()) => rewrite(meta).map(|meta| (blob.clone(), meta)),
            _ => None,
        };
        match &newer {
            Some((_, meta)) => {
                tx.execute(
                    "UPDATE pending_op
                     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],
                )?;
            }
            None => {
                tx.execute("DELETE FROM pending_op WHERE id = ?1", params![id])?;
            }
        }
        if let Some(CreateRow {
            parent_uid: Some(parent),
            name: Some(name),
            created_at,
            ..
        }) = &row
            && (parent.as_str(), name.as_str()) != landing.sent
        {
            let meta = serde_json::to_string(&RenameMeta {
                original_parent_uid: landing.landed.0.to_string(),
                original_name: Some(landing.landed.1.to_string()),
            })?;
            tx.execute(
                &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)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![OP_RENAME, real, parent, name, meta, created_at],
            )?;
            fill_op_lids(&tx, tx.last_insert_rowid())?;
        }
        adopt_placeholder_row_tx(&tx, local, real)?;
        tx.commit()?;
        Ok(match newer {
            Some((blob, meta)) => CreateRetired::Newer { blob, meta },
            None => CreateRetired::Landed,
        })
    }

    /// Record that queued create `id` is sending a blob with SHA-1 `sha`.
    ///
    /// Drive may make the file and lose its answer. The op that next finds
    /// the file, the create's retry or the trash a delete made of it, takes
    /// it for its own when it holds any bytes sent so. Kept on the row, so a
    /// newer blob, the turn into a trash and a restart all keep it
    /// (`docs/BUGS.md` B156, B183).
    pub fn note_create_sent(&self, id: i64, sha: &str) -> Result<()> {
        self.conn.lock().execute(
            "UPDATE pending_op
             SET sent_sha1 = CASE
                 WHEN sent_sha1 IS NULL THEN ?2
                 WHEN instr(' ' || sent_sha1 || ' ', ' ' || ?2 || ' ') > 0 THEN sent_sha1
                 ELSE sent_sha1 || ' ' || ?2 END
             WHERE id = ?1",
            params![id, sha],
        )?;
        Ok(())
    }

    /// Record that queued create `id` is being sent to `parent` under `name`.
    ///
    /// A rename while the op waits rewrites its target, but the file a lost
    /// answer left stays where it was sent. The retry and the trash look
    /// there too (`docs/BUGS.md` B184).
    pub fn note_create_target(&self, id: i64, parent: &str, name: &str) -> Result<()> {
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        let sent: Option<String> = tx
            .query_row("SELECT sent_to FROM pending_op WHERE id = ?1", [id], |r| {
                r.get(0)
            })
            .optional()?
            .flatten();
        let mut targets: Vec<(String, String)> = sent
            .and_then(|json| serde_json::from_str(&json).ok())
            .unwrap_or_default();
        let target = (parent.to_string(), name.to_string());
        if !targets.contains(&target) {
            targets.push(target);
            tx.execute(
                "UPDATE pending_op SET sent_to = ?2 WHERE id = ?1",
                params![id, serde_json::to_string(&targets)?],
            )?;
        }
        tx.commit()?;
        Ok(())
    }

    /// The parent and name queued op `id` was sent under as a create, in the
    /// order it was ([`Db::note_create_target`]).
    pub fn create_targets(&self, id: i64) -> Result<Vec<(String, String)>> {
        let sent: Option<String> = self
            .conn
            .lock()
            .query_row("SELECT sent_to FROM pending_op WHERE id = ?1", [id], |r| {
                r.get(0)
            })
            .optional()?
            .flatten();
        Ok(sent
            .and_then(|json| serde_json::from_str(&json).ok())
            .unwrap_or_default())
    }

    /// The SHA-1 of each blob queued op `id` sent as a create
    /// ([`Db::note_create_sent`]).
    pub fn create_sent(&self, id: i64) -> Result<Vec<String>> {
        let sent: Option<String> = self
            .conn
            .lock()
            .query_row(
                "SELECT sent_sha1 FROM pending_op WHERE id = ?1",
                [id],
                |r| r.get(0),
            )
            .optional()?
            .flatten();
        Ok(sent
            .map(|sent| sent.split(' ').map(str::to_string).collect())
            .unwrap_or_default())
    }

    /// 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<bool> {
        let conn = self.conn.lock();
        let lid = node_lid(&conn, uid)?;
        let n = conn.execute(
            &format!(
                "UPDATE pending_op SET meta_json = ?3 WHERE {} AND kind = ?2",
                on_node(4)
            ),
            params![uid, kind, meta_json, lid],
        )?;
        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.
    ///
    /// A create backing off is made due: what it failed on may have been the
    /// name it no longer wants, such as one held by a queued change. A parked
    /// create stays parked.
    pub fn rewrite_op_target(&self, uid: &str, parent_uid: &str, name: &str) -> Result<bool> {
        let conn = self.conn.lock();
        let lid = node_lid(&conn, uid)?;
        let parent_lid = node_lid(&conn, parent_uid)?;
        let n = conn.execute(
            &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)
    }

    /// 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<bool> {
        let conn = self.conn.lock();
        let lid = node_lid(&conn, uid)?;
        let n = conn.execute(
            &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)
    }

    /// Check if a queued create or mkdir op exists for `uid`.
    pub fn has_create_op(&self, uid: &str) -> Result<bool> {
        let conn = self.read();
        let lid = node_lid(&conn, uid)?;
        let count: i64 = conn.query_row(
            &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)
    }

    /// 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<bool> {
        let conn = self.read();
        let lid = node_lid(&conn, uid)?;
        let count: i64 = conn.query_row(
            &format!("SELECT COUNT(*) FROM pending_op WHERE {}", on_node(2)),
            params![uid, lid],
            |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. Revisions of files the
    /// node tree places beneath `uid` go too; see [`drop_doomed_ops`].
    pub fn delete_ops_for_uid(&self, uid: &str) -> Result<Vec<String>> {
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        let blobs = drop_doomed_ops(&tx, uid)?;
        tx.commit()?;
        Ok(blobs)
    }

    /// Drop the ops queued on the node `uid` names, and nothing else, returning
    /// the staged blobs they held. `None`, dropping nothing, while one of them
    /// is on the wire: it may land after the drop.
    ///
    /// For a change the user discards on a node Drive has. Unlike
    /// [`Db::delete_ops_for_uid`], what is queued below the node stays: the
    /// files made in a folder whose rename is discarded still go in it.
    pub fn discard_ops_on_node(&self, uid: &str) -> Result<Option<Vec<String>>> {
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        let lid = node_lid(&tx, uid)?;
        let on = on_node(2);
        let sending: bool = tx.query_row(
            &format!("SELECT EXISTS (SELECT 1 FROM pending_op WHERE {on} AND claimed_at <> 0)"),
            params![uid, lid],
            |r| r.get(0),
        )?;
        if sending {
            return Ok(None);
        }
        let blobs = {
            let mut stmt = tx.prepare(&format!(
                "SELECT blob_path FROM pending_op WHERE {on} AND blob_path IS NOT NULL"
            ))?;
            let rows = stmt.query_map(params![uid, lid], |r| r.get::<_, String>(0))?;
            rows.collect::<std::result::Result<Vec<_>, _>>()?
        };
        tx.execute(
            &format!("DELETE FROM pending_op WHERE {on}"),
            params![uid, lid],
        )?;
        tx.commit()?;
        Ok(Some(blobs))
    }

    /// 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)?;
        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 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.
    ///
    /// `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<Option<PendingOp>> {
        let conn = self.read();
        let mut stmt = conn.prepare_cached(&format!(
            "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 (parent_uid IS NULL OR substr({PARENT_NOW}, 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)
    }

    /// Take the oldest due op for this worker, marking it claimed so no other
    /// drain worker takes it, or `None` when there is nothing to do.
    ///
    /// The drain is several workers over one queue, which is only safe because
    /// the claim is a column: a row is offered to exactly one worker, and the
    /// worker that took it is the only one that may retire, defer or fail it.
    ///
    /// Two exclusions on top of [`next_due_op`](Self::next_due_op)'s:
    ///
    /// * `claimed_at = 0` — not already in flight somewhere else.
    /// * no *other* claimed op shares this `uid`. Ordering only has to hold per
    ///   node, but there it has to hold absolutely: a queued rename and a queued
    ///   revision of one file must land in the order they were made, and two
    ///   workers on one node would also race over its staged blob and its
    ///   pending entry. `ORDER BY id` plus this exclusion gives per-uid serial
    ///   order for free — the older op is claimed first and the newer one is
    ///   invisible until it retires.
    ///
    /// Select and mark are one transaction, so two workers cannot both see the
    /// row as unclaimed.
    pub fn claim_next_due_op(&self, now: i64) -> Result<Option<PendingOp>> {
        let mut conn = self.conn.lock();
        let tx = conn.transaction()?;
        let op = {
            let mut stmt = tx.prepare_cached(&format!(
                "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 {claimable}
                 ORDER BY id LIMIT 1",
                claimable = claimable(),
            ))?;
            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()?
        };
        if let Some(op) = op.as_ref() {
            tx.execute(
                "UPDATE pending_op SET claimed_at = ?2 WHERE id = ?1",
                params![op.id, now.max(1)],
            )?;
        }
        tx.commit()?;
        Ok(op)
    }

    /// Give a claimed op back to the queue, whatever became of the attempt.
    ///
    /// Called on every path out of a drain attempt rather than only the failing
    /// ones: a handler that retires its own row leaves nothing to release (the
    /// `UPDATE` matches nothing), and a handler that returns without retiring
    /// *and* without failing would otherwise leave the row claimed by a worker
    /// that has moved on — invisible to every worker, forever.
    ///
    /// A rename still here with a newer one queued behind it goes instead, as
    /// in [`Db::record_op_failure`]. The newer one may have been queued after
    /// the failure was recorded: kept, the older one sat out its backoff while
    /// the newer one landed, and then put the old name back (`docs/BUGS.md`
    /// B158).
    pub fn release_op_claim(&self, id: i64) -> Result<()> {
        let conn = self.conn.lock();
        if drop_replaced_rename(&conn, id)? {
            return Ok(());
        }
        conn.execute(
            "UPDATE pending_op SET claimed_at = 0 WHERE id = ?1",
            params![id],
        )?;
        Ok(())
    }

    /// Drop every claim in the table.
    ///
    /// A claim describes a worker in *this* process, so one surviving on disk
    /// means the previous run died holding it. The single-writer lock
    /// ([`Db::open`]) is what makes that inference safe: nobody else can be
    /// draining this queue while we are opening it.
    pub fn clear_op_claims(&self) -> Result<usize> {
        let conn = self.conn.lock();
        Ok(conn.execute(
            "UPDATE pending_op SET claimed_at = 0 WHERE claimed_at <> 0",
            [],
        )?)
    }

    /// 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.
    ///
    /// Only rows [`claim_next_due_op`](Self::claim_next_due_op) could take
    /// count. A claimed op, or one held behind a claimed op on the same node,
    /// is not work this worker is waiting for, and counting it would have an
    /// idle worker spin on a row it cannot have.
    pub fn earliest_due_at(&self) -> Result<Option<i64>> {
        let conn = self.read();
        let ts: Option<i64> = conn
            .query_row(
                &format!(
                    "SELECT MIN(next_attempt_at) FROM pending_op WHERE {claimable}",
                    claimable = claimable(),
                ),
                [],
                |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<Vec<PendingOp>> {
        let conn = self.read();
        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"
        ))?;
        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::<std::result::Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    /// Every parked create/mkdir, oldest first, for the park sweep.
    ///
    /// Parked rows are invisible to [`next_due_op`](Self::next_due_op) by
    /// construction, so nothing else in the drain ever looks at them. This is
    /// the one query that does.
    pub fn parked_create_ops(&self) -> Result<Vec<PendingOp>> {
        let conn = self.read();
        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
             WHERE next_attempt_at >= ?1 AND kind IN (?2, ?3)
             ORDER BY created_at"
        ))?;
        let rows = stmt
            .query_map(params![PARK_UNTIL, OP_CREATE, OP_MKDIR], |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::<std::result::Result<Vec<_>, _>>()?;
        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<PendingCounts> {
        let conn = self.read();
        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),
        )?;
        let parked = conn.query_row(
            "SELECT COUNT(*) FROM pending_op WHERE next_attempt_at >= ?1",
            params![PARK_UNTIL],
            |r| r.get(0),
        )?;
        let failing = conn.query_row(
            "SELECT COUNT(*) FROM pending_op WHERE attempts >= ?1 OR issue IS NOT NULL",
            params![FAILING_ATTEMPTS],
            |r| r.get(0),
        )?;
        let last_error = conn
            .query_row(
                "SELECT last_error FROM pending_op
                  WHERE (attempts >= ?1 OR issue IS NOT NULL) AND last_error IS NOT NULL
                  ORDER BY issue IS NULL, attempts DESC LIMIT 1",
                params![FAILING_ATTEMPTS],
                |r| r.get::<_, Option<String>>(0),
            )
            .optional()?
            .flatten();
        Ok(PendingCounts {
            uploads,
            changes,
            parked,
            failing,
            last_error,
        })
    }

    /// 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";

    /// Whether op `id` is still queued. A drain worker that claimed it can find
    /// it gone: the node was deleted, or a newer write superseded it, while the
    /// worker was talking to the server.
    pub fn op_exists(&self, id: i64) -> Result<bool> {
        let conn = self.read();
        let count: i64 = conn.query_row(
            "SELECT COUNT(*) FROM pending_op WHERE id = ?1",
            params![id],
            |row| row.get(0),
        )?;
        Ok(count > 0)
    }

    /// 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(())
    }

    /// Make one backed-off op due now, for the user's "Retry now". Parked ops are
    /// left alone: they wait for a rename, not for time, and un-parking one would
    /// upload a file its writer has not finished. Returns whether a row changed.
    pub fn retry_op_now(&self, id: i64, now: i64) -> Result<bool> {
        let conn = self.conn.lock();
        let changed = conn.execute(
            "UPDATE pending_op SET next_attempt_at = ?2
             WHERE id = ?1 AND next_attempt_at > ?2 AND next_attempt_at < ?3",
            params![id, now, PARK_UNTIL],
        )?;
        Ok(changed > 0)
    }

    /// [`retry_op_now`](Self::retry_op_now) for every op that has failed at
    /// least once. Returns how many rows changed.
    pub fn retry_failed_ops_now(&self, now: i64) -> Result<usize> {
        let conn = self.conn.lock();
        let changed = conn.execute(
            "UPDATE pending_op SET next_attempt_at = ?1
             WHERE attempts > 0 AND next_attempt_at > ?1 AND next_attempt_at < ?2",
            params![now, PARK_UNTIL],
        )?;
        Ok(changed)
    }

    /// 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.
    ///
    /// Clears any access-deferral window: whatever the op was waiting for has
    /// now been reported, so the next uncleared deferral starts its own window
    /// rather than escalating again on the following recheck.
    ///
    /// A rename with a newer one queued behind it ([`Db::enqueue_op`]) is
    /// dropped instead: retried, it could land after the newer one and undo it.
    pub fn record_op_failure(&self, id: i64, error: &str, next_attempt_at: i64) -> Result<()> {
        self.record_failure(id, None, error, next_attempt_at)
    }

    /// [`Db::record_op_failure`] for an attempt at op `id` as a `kind` op. An
    /// op that has become another kind while the attempt ran is left as it
    /// is: a create deleted on the wire is a trash now, and the create's
    /// failure would hold the trash back for the create's backoff
    /// (`docs/BUGS.md` B171).
    pub fn record_attempt_failure(
        &self,
        id: i64,
        kind: &str,
        error: &str,
        next_attempt_at: i64,
    ) -> Result<()> {
        self.record_failure(id, Some(kind), error, next_attempt_at)
    }

    fn record_failure(
        &self,
        id: i64,
        kind: Option<&str>,
        error: &str,
        next_attempt_at: i64,
    ) -> Result<()> {
        let conn = self.conn.lock();
        if drop_replaced_rename(&conn, id)? {
            return Ok(());
        }
        conn.execute(
            "UPDATE pending_op
             SET attempts = attempts + 1, last_error = ?2, next_attempt_at = ?3,
                 access_deferred_since = 0
             WHERE id = ?1 AND (?4 IS NULL OR kind = ?4)",
            params![id, error, next_attempt_at, kind],
        )?;
        Ok(())
    }

    /// Record why Drive refused op `id`, or that the latest failure was not a
    /// refusal (`None`). An op with an issue counts as failing at once.
    pub fn set_op_issue(&self, id: i64, issue: Option<&str>) -> Result<()> {
        self.conn.lock().execute(
            "UPDATE pending_op SET issue = ?2 WHERE id = ?1",
            params![id, issue],
        )?;
        Ok(())
    }

    /// The issue of every op Drive refused, by op id.
    pub fn op_issues(&self) -> Result<HashMap<i64, String>> {
        let conn = self.read();
        let mut stmt = conn.prepare("SELECT id, issue FROM pending_op WHERE issue IS NOT NULL")?;
        let rows = stmt
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
            .collect::<rusqlite::Result<_>>()?;
        Ok(rows)
    }

    /// The issue each node with a refused change has, by local id: what the
    /// file browser shows on the node itself.
    pub fn node_issues(&self) -> Result<HashMap<i64, String>> {
        let conn = self.read();
        let mut stmt = conn.prepare(
            "SELECT COALESCE(o.lid, n.lid), o.issue FROM pending_op o
             LEFT JOIN nodes n ON o.lid IS NULL AND n.uid = o.uid
             WHERE o.issue IS NOT NULL",
        )?;
        let rows = stmt
            .query_map([], |r| Ok((r.get::<_, Option<i64>>(0)?, r.get(1)?)))?
            .filter_map(|row| match row {
                Ok((Some(lid), issue)) => Some(Ok((lid, issue))),
                Ok((None, _)) => None,
                Err(e) => Some(Err(e)),
            })
            .collect::<rusqlite::Result<_>>()?;
        Ok(rows)
    }

    /// Make a create due again if it was renamed or moved
    /// ([`Db::rewrite_op_target`]) while the attempt that just failed was on
    /// the wire, `parent_uid` and `name` being what that attempt sent. What it
    /// failed on may have been the target it no longer has. Likewise if a
    /// write replaced its blob (`blob_path` being the one the attempt read):
    /// the old one is discarded as the new one attaches, and an attempt that
    /// had not opened it yet failed on a file that is gone. Returns whether it
    /// was.
    pub fn retry_if_retargeted(
        &self,
        id: i64,
        parent_uid: Option<&str>,
        name: Option<&str>,
        blob_path: Option<&str>,
    ) -> Result<bool> {
        let conn = self.conn.lock();
        let n = conn.execute(
            &format!(
                "UPDATE pending_op SET next_attempt_at = 0
                 WHERE id = ?1 AND kind IN (?4, ?5) AND next_attempt_at < ?6
                   AND ({PARENT_NOW} IS NOT ?2 OR name IS NOT ?3 OR blob_path IS NOT ?7)"
            ),
            params![
                id, parent_uid, name, OP_CREATE, OP_MKDIR, PARK_UNTIL, blob_path
            ],
        )?;
        Ok(n > 0)
    }

    /// 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(())
    }

    /// Defer an op the local access check refused, and report how long it has
    /// been refused for.
    ///
    /// Identical to [`defer_op_without_attempt`](Self::defer_op_without_attempt)
    /// on the retry clock, plus the one thing that was missing: the first
    /// deferral of a run stamps `access_deferred_since`, and every deferral
    /// returns that stamp. A caller can therefore tell a deferral that is one
    /// recheck old from one that has been repeating unchanged for days, which is
    /// the difference between a permission the user is mid-way through changing
    /// and a queue that will never drain.
    ///
    /// Returns the ms at which the current run of deferrals began — `now` on the
    /// first one.
    pub fn defer_op_for_access(&self, id: i64, now: i64, next_attempt_at: i64) -> Result<i64> {
        let conn = self.conn.lock();
        conn.execute(
            "UPDATE pending_op
             SET next_attempt_at = CASE
                 WHEN next_attempt_at >= ?4 THEN next_attempt_at
                 ELSE ?3
             END,
             access_deferred_since = CASE
                 WHEN access_deferred_since = 0 THEN ?2
                 ELSE access_deferred_since
             END
             WHERE id = ?1",
            params![id, now, next_attempt_at, PARK_UNTIL],
        )?;
        let since = conn
            .query_row(
                "SELECT access_deferred_since FROM pending_op WHERE id = ?1",
                params![id],
                |r| r.get::<_, i64>(0),
            )
            .optional()?;
        // A row that drained out from under this call leaves nothing to measure;
        // reporting `now` says "this run just started", which is the reading that
        // never escalates something that no longer exists.
        Ok(since.filter(|since| *since != 0).unwrap_or(now))
    }

    /// Forget an op's access-deferral window, after an attempt the access check
    /// let through. The op is no longer blocked, so a later deferral is the
    /// start of a new run rather than the continuation of an old one.
    pub fn clear_op_access_deferral(&self, id: i64) -> Result<()> {
        let conn = self.conn.lock();
        conn.execute(
            "UPDATE pending_op SET access_deferred_since = 0
             WHERE id = ?1 AND access_deferred_since <> 0",
            params![id],
        )?;
        Ok(())
    }

    /// Every staged blob the queue still refers to.
    ///
    /// The startup staging reconcile subtracts this from what is actually in
    /// `staging/`: the difference is bytes the kernel was told were written and
    /// that nothing is going to upload.
    pub fn op_blob_paths(&self) -> Result<std::collections::HashSet<String>> {
        let conn = self.read();
        let mut stmt =
            conn.prepare("SELECT blob_path FROM pending_op WHERE blob_path IS NOT NULL")?;
        let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
        let mut out = std::collections::HashSet::new();
        for row in rows {
            out.insert(row?);
        }
        Ok(out)
    }

    /// Record that this daemon sealed `revision_id` on `uid`, and prune entries
    /// past [`OWN_SEALED_TTL_MS`].
    ///
    /// One row per node, overwritten as the node advances. See
    /// [`own_sealed_rev`](Self::own_sealed_rev) for what reads it.
    pub fn set_own_sealed_rev(&self, uid: &str, revision_id: &str, now_ms: i64) -> Result<()> {
        let conn = self.conn.lock();
        conn.execute(
            "INSERT INTO own_sealed_rev (uid, revision_id, sealed_at)
             VALUES (?1, ?2, ?3)
             ON CONFLICT(uid) DO UPDATE SET
               revision_id = excluded.revision_id,
               sealed_at   = excluded.sealed_at",
            params![uid, revision_id, now_ms],
        )?;
        conn.execute(
            "DELETE FROM own_sealed_rev WHERE sealed_at < ?1",
            params![now_ms - OWN_SEALED_TTL_MS],
        )?;
        Ok(())
    }

    /// The last revision of `uid` this daemon sealed itself, if it is still
    /// within [`OWN_SEALED_TTL_MS`].
    ///
    /// The drain asks before treating a moved-on remote as a conflict: a
    /// revision we sealed means one writer stalled and resumed, which
    /// supersedes rather than forking a `(sync-conflict)` copy (docs/BUGS.md
    /// B70 layer B).
    pub fn own_sealed_rev(&self, uid: &str, now_ms: i64) -> Result<Option<String>> {
        let conn = self.read();
        Ok(conn
            .query_row(
                "SELECT revision_id FROM own_sealed_rev
                  WHERE uid = ?1 AND sealed_at >= ?2",
                params![uid, now_ms - OWN_SEALED_TTL_MS],
                |r| r.get(0),
            )
            .optional()?)
    }

    /// Forget what we sealed on `uid` — it has been trashed, so there is no
    /// revision left to chain onto.
    pub fn clear_own_sealed_rev(&self, uid: &str) -> Result<()> {
        let conn = self.conn.lock();
        conn.execute("DELETE FROM own_sealed_rev WHERE uid = ?1", params![uid])?;
        Ok(())
    }
}

/// How long a revision this daemon sealed stays recognisable as its own.
///
/// Bounds a table that would otherwise grow with every write for the life of
/// the install. A week is far longer than any stall→resume — a browser download
/// that pauses and continues, an editor that reopens its handle — and short
/// enough that the table tracks recent work rather than accumulating history.
pub const OWN_SEALED_TTL_MS: i64 = 7 * 24 * 60 * 60 * 1000;