fix(core): never give out a pending_op id twice

A plain rowid is the largest one plus one, so a trash queued in the same
transaction that dropped the newest op took its id. The drain retires the
op it sent by id, so a rename landing deleted the trash queued over it and
the file stayed on Drive (B137). Schema 37 rebuilds pending_op with
AUTOINCREMENT, keeping every id and index.

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

@@ -5,11 +5,13 @@
 //! is already on disk somewhere, so a change here would silently diverge from
 //! what an existing install actually has. Add a new one instead.
 
+use rusqlite::OptionalExtension;
+
 use super::Db;
 use crate::Result;
 
 /// Current schema version. Bump on every forward migration added below.
-pub(super) const SCHEMA_VERSION: i64 = 36;
+pub(super) const SCHEMA_VERSION: i64 = 37;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -348,6 +350,19 @@ impl Db {
                 tx.execute_batch(MIGRATION_V36)?;
             }
         }
+        if current < 37 {
+            // Same guards as V26-V36.
+            let sql: Option<String> = tx
+                .query_row(
+                    "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'pending_op'",
+                    [],
+                    |row| row.get(0),
+                )
+                .optional()?;
+            if sql.is_some_and(|sql| !sql.contains("AUTOINCREMENT")) {
+                tx.execute_batch(MIGRATION_V37)?;
+            }
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -1112,3 +1127,40 @@ CREATE INDEX idx_nodes_parent ON nodes(parent_uid);
 CREATE INDEX idx_nodes_name_nocase ON nodes(name COLLATE NOCASE);
 CREATE INDEX idx_nodes_parent_live ON nodes(parent_uid) WHERE trashed = 0;
 ";
+
+/// Schema v37: a `pending_op` id is never given out twice.
+///
+/// A plain rowid is the largest one plus one, so an op queued right after the
+/// newest one was dropped took its id. The drain retires the op it is
+/// working on by id, and so deleted the new op: a trash queued over a rename
+/// on the wire was lost when the rename landed. `AUTOINCREMENT` never reuses
+/// an id. Ids are kept, as is every index.
+const MIGRATION_V37: &str = "
+CREATE TABLE pending_op_v37 (
+  id                    INTEGER PRIMARY KEY AUTOINCREMENT,
+  kind                  TEXT NOT NULL,
+  uid                   TEXT NOT NULL,
+  blob_path             TEXT,
+  meta_json             TEXT,
+  created_at            INTEGER NOT NULL,
+  attempts              INTEGER NOT NULL DEFAULT 0,
+  last_error            TEXT,
+  next_attempt_at       INTEGER NOT NULL DEFAULT 0,
+  parent_uid            TEXT,
+  name                  TEXT,
+  claimed_at            INTEGER NOT NULL DEFAULT 0,
+  access_deferred_since INTEGER NOT NULL DEFAULT 0
+);
+INSERT INTO pending_op_v37
+  (id, kind, uid, blob_path, meta_json, created_at, attempts, last_error, next_attempt_at,
+   parent_uid, name, claimed_at, access_deferred_since)
+  SELECT id, kind, uid, blob_path, meta_json, created_at, attempts, last_error,
+         next_attempt_at, parent_uid, name, claimed_at, access_deferred_since
+    FROM pending_op;
+DROP TABLE pending_op;
+ALTER TABLE pending_op_v37 RENAME TO pending_op;
+CREATE INDEX pending_op_uid ON pending_op(uid);
+CREATE INDEX pending_op_parent ON pending_op(parent_uid);
+CREATE INDEX idx_pending_op_claimed
+    ON pending_op(uid) WHERE claimed_at <> 0;
+";

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

@@ -3538,6 +3538,39 @@ fn access_deferral_does_not_unpark_a_transient_create() {
     assert_eq!(db.pending_ops().unwrap()[0].next_attempt_at, PARK_UNTIL);
 }
 
+/// The drain retires the op it worked on by id. An id given out again let
+/// a rename that landed delete the trash queued over it.
+#[test]
+fn a_dropped_op_id_is_never_given_out_again() {
+    let db = Db::open_in_memory().unwrap();
+    let rename = |name: &str| PendingOp {
+        id: 0,
+        kind: OP_RENAME.to_string(),
+        uid: uid("node").to_string(),
+        parent_uid: Some(uid("root").to_string()),
+        name: Some(name.to_string()),
+        blob_path: None,
+        meta_json: None,
+        created_at: 1,
+        attempts: 0,
+        last_error: None,
+        next_attempt_at: 0,
+    };
+    let (renamed, _) = db.enqueue_op(&rename("b.txt")).unwrap();
+    let in_flight = db.claim_next_due_op(10).unwrap().unwrap();
+    assert_eq!(in_flight.id, renamed);
+
+    let (trash, _) = db
+        .replace_ops_with_trash(&uid("node").to_string(), "b.txt", 2)
+        .unwrap();
+    assert_ne!(trash, renamed);
+    db.delete_op(in_flight.id).unwrap();
+
+    let ops = db.pending_ops().unwrap();
+    assert_eq!(ops.len(), 1);
+    assert_eq!(ops[0].kind, OP_TRASH);
+}
+
 #[test]
 fn migrate_is_idempotent() {
     let db = Db::open_in_memory().unwrap();