perf(core): index queued ops by kind and name (schema 43)

The name checks the drain makes before each create, rename and trash,
and the wake of ops waiting for a name, read the whole queue, which made
a large offline batch quadratic to drain. Schema 43 adds an index on
pending_op(kind, name) (B238).

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

@@ -11,7 +11,7 @@ use super::Db;
 use crate::Result;
 
 /// Current schema version. Bump on every forward migration added below.
-pub(super) const SCHEMA_VERSION: i64 = 42;
+pub(super) const SCHEMA_VERSION: i64 = 43;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -440,6 +440,16 @@ impl Db {
                 tx.execute_batch(MIGRATION_V42)?;
             }
         }
+        if current < 43 {
+            let has_ops: bool = tx.query_row(
+                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'pending_op'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            if has_ops {
+                tx.execute_batch(MIGRATION_V43)?;
+            }
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -1330,3 +1340,14 @@ const MIGRATION_V42: &str = "
 ALTER TABLE pending_op ADD COLUMN sent_sha1 TEXT;
 ALTER TABLE pending_op ADD COLUMN sent_to TEXT;
 ";
+
+/// Schema v43: queued ops by kind and name.
+///
+/// Before each create, rename and trash the drain asks whether a queued trash
+/// or rename holds the name it wants, and a released name wakes the ops
+/// waiting for it. Each of those scanned the whole queue, so a large offline
+/// batch made draining it quadratic. The kind leads, so a lookup of every
+/// queued rename uses the index too.
+const MIGRATION_V43: &str = "
+CREATE INDEX IF NOT EXISTS pending_op_kind_name ON pending_op(kind, name);
+";

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

@@ -7531,6 +7531,30 @@ fn a_refused_change_names_its_node_by_local_id() {
     );
 }
 
+/// The name checks the drain makes before each op look the name up in an
+/// index. Scanning the queue instead made a large offline batch quadratic.
+#[test]
+fn queue_name_checks_use_an_index() {
+    let db = Db::open_in_memory().unwrap();
+    let conn = db.conn.lock();
+    let plan = |sql: &str| -> String {
+        let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap();
+        stmt.query_map([], |r| r.get::<_, String>(3))
+            .unwrap()
+            .map(|row| row.unwrap())
+            .collect::<Vec<_>>()
+            .join("; ")
+    };
+    for sql in [
+        "SELECT 1 FROM pending_op WHERE kind = 'trash' AND name = 'a'",
+        "SELECT 1 FROM pending_op WHERE name = 'a' AND kind IN ('create', 'mkdir', 'rename')",
+        "SELECT 1 FROM pending_op WHERE kind = 'rename'",
+    ] {
+        let plan = plan(sql);
+        assert!(plan.contains("pending_op_kind_name"), "{sql}: {plan}");
+    }
+}
+
 /// A reconnect moves up the ops that failed for want of a link. One Drive
 /// refused, for quota or access, keeps its backoff: the link coming back
 /// changes nothing about it. "Retry all" still moves both.