diff --git a/Cargo.lock b/Cargo.lock
index ba7494e..4959bda 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3518,7 +3518,7 @@ checksum = "35fb2e5f958ec131621fdd531e9fc186ed768cbe395337403ae56c17a74c68ec"
 
 [[package]]
 name = "pdfs-cli"
-version = "1.6.0"
+version = "1.7.0"
 dependencies = [
  "anyhow",
  "clap 4.6.4",
@@ -3533,7 +3533,7 @@ dependencies = [
 
 [[package]]
 name = "pdfs-core"
-version = "1.6.0"
+version = "1.7.0"
 dependencies = [
  "async-trait",
  "directories",
@@ -3554,7 +3554,7 @@ dependencies = [
 
 [[package]]
 name = "pdfs-fuse"
-version = "1.6.0"
+version = "1.7.0"
 dependencies = [
  "fuser",
  "futures",
@@ -3571,7 +3571,7 @@ dependencies = [
 
 [[package]]
 name = "pdfs-gui"
-version = "1.6.0"
+version = "1.7.0"
 dependencies = [
  "anyhow",
  "async-channel",
diff --git a/Cargo.toml b/Cargo.toml
index 73e796a..6161973 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -3,7 +3,7 @@ resolver = "2"
 members = ["crates/pdfs-core", "crates/pdfs-fuse", "crates/pdfs-cli", "crates/pdfs-gui"]
 
 [workspace.package]
-version = "1.6.0"
+version = "1.7.0"
 edition = "2024"
 license = "MIT"
 authors = ["Nils Pukropp <contact@narl.io>"]
diff --git a/crates/pdfs-core/src/db/activity.rs b/crates/pdfs-core/src/db/activity.rs
index 4a77ee1..3998598 100644
--- a/crates/pdfs-core/src/db/activity.rs
+++ b/crates/pdfs-core/src/db/activity.rs
@@ -33,7 +33,7 @@ impl Db {
     /// whose stored `kind` no longer parses (written by an older build) are
     /// skipped rather than failing the whole read.
     pub fn activity_list(&self, limit: usize) -> Result<Vec<ActivityEntry>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT time, kind, target, detail, ok FROM activity
              ORDER BY id DESC LIMIT ?1",
diff --git a/crates/pdfs-core/src/db/albums.rs b/crates/pdfs-core/src/db/albums.rs
index 21bd232..69ae088 100644
--- a/crates/pdfs-core/src/db/albums.rs
+++ b/crates/pdfs-core/src/db/albums.rs
@@ -76,7 +76,7 @@ impl Db {
 
     /// The persisted album listing, in stored order.
     pub fn albums_list(&self) -> Result<Vec<StoredAlbum>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT uid, name, photo_count, cover_uid, last_activity, shared
              FROM albums ORDER BY seq",
@@ -98,7 +98,7 @@ impl Db {
 
     /// Number of albums in the persisted listing.
     pub fn albums_count(&self) -> Result<usize> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let n: i64 = conn.query_row("SELECT COUNT(*) FROM albums", [], |r| r.get(0))?;
         Ok(n.max(0) as usize)
     }
@@ -164,7 +164,7 @@ impl Db {
         offset: usize,
         limit: usize,
     ) -> Result<Vec<StoredPhoto>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT uid, capture_time, name, ratio, thumb_state, kind FROM album_photos
              WHERE album_uid = ?1 ORDER BY seq LIMIT ?2 OFFSET ?3",
@@ -190,7 +190,7 @@ impl Db {
 
     /// How many photos of `album_uid` are persisted.
     pub fn album_photos_count(&self, album_uid: &str) -> Result<usize> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let n: i64 = conn.query_row(
             "SELECT COUNT(*) FROM album_photos WHERE album_uid = ?1",
             [album_uid],
@@ -207,7 +207,7 @@ impl Db {
         if uids.is_empty() {
             return Ok(Vec::new());
         }
-        let conn = self.conn.lock();
+        let conn = self.read();
         let placeholders = vec!["?"; uids.len()].join(",");
         let mut stmt = conn.prepare(&format!(
             "SELECT uid, capture_time, name, ratio, thumb_state, kind FROM album_photos
diff --git a/crates/pdfs-core/src/db/cache.rs b/crates/pdfs-core/src/db/cache.rs
index 8c585bf..f71761a 100644
--- a/crates/pdfs-core/src/db/cache.rs
+++ b/crates/pdfs-core/src/db/cache.rs
@@ -99,7 +99,7 @@ impl Db {
     /// because `enforce_block_budget` runs on every cached block — once per
     /// 4 MiB of every cold read — while holding the shared connection.
     pub fn cache_total_bytes(&self, kind: &str) -> Result<u64> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let total: i64 = conn.query_row(
             "SELECT COALESCE(SUM(size_bytes), 0) FROM cache_entries WHERE kind = ?1",
             params![kind],
@@ -119,7 +119,7 @@ impl Db {
         kind: &str,
         limit: usize,
     ) -> Result<Vec<(String, u64)>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare_cached(
             "SELECT cache_key, size_bytes FROM cache_entries
              WHERE kind = ?1 ORDER BY last_accessed ASC LIMIT ?2",
@@ -139,7 +139,7 @@ impl Db {
     /// (`clear_unpinned`); incremental eviction wants
     /// [`cache_eviction_candidates`](Self::cache_eviction_candidates).
     pub fn cache_entries_by_kind(&self, kind: &str) -> Result<Vec<(String, u64)>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT cache_key, size_bytes FROM cache_entries
              WHERE kind = ?1 ORDER BY last_accessed ASC",
diff --git a/crates/pdfs-core/src/db/devices.rs b/crates/pdfs-core/src/db/devices.rs
index e17a52e..2d7c242 100644
--- a/crates/pdfs-core/src/db/devices.rs
+++ b/crates/pdfs-core/src/db/devices.rs
@@ -15,7 +15,7 @@ pub struct StoredDevice {
 
 impl Db {
     pub fn device_get(&self) -> Result<Option<StoredDevice>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         conn.query_row(
             "SELECT uid, share_id, root_uid, name, created FROM device LIMIT 1",
             [],
diff --git a/crates/pdfs-core/src/db/local.rs b/crates/pdfs-core/src/db/local.rs
index 2e2419e..f5f4ae2 100644
--- a/crates/pdfs-core/src/db/local.rs
+++ b/crates/pdfs-core/src/db/local.rs
@@ -118,7 +118,7 @@ impl Db {
     /// has never been built. The daemon uses this to decide whether a fresh mount
     /// needs an immediate rescan or can serve the existing index.
     pub fn local_indexed_at(&self) -> Result<Option<i64>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         Ok(conn
             .query_row(
                 "SELECT value FROM sync_state WHERE key = 'local_indexed_at'",
@@ -138,7 +138,7 @@ impl Db {
         if query.is_empty() {
             return Ok(Vec::new());
         }
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt;
         let rows = if query.chars().count() < TRIGRAM_MIN {
             let pat = format!("%{}%", like_escape(query));
@@ -177,7 +177,7 @@ impl Db {
         if query.is_empty() || limit == 0 {
             return Ok(Vec::new());
         }
-        let conn = self.conn.lock();
+        let conn = self.read();
         let terms = candidate_trigrams(query);
         let lane_limit = limit.div_ceil(2);
         let mut rows: Vec<LocalFileHit> = if terms.is_empty() {
diff --git a/crates/pdfs-core/src/db/migrations.rs b/crates/pdfs-core/src/db/migrations.rs
index 116b601..1baebb7 100644
--- a/crates/pdfs-core/src/db/migrations.rs
+++ b/crates/pdfs-core/src/db/migrations.rs
@@ -9,7 +9,7 @@ use super::Db;
 use crate::Result;
 
 /// Current schema version. Bump on every forward migration added below.
-pub(super) const SCHEMA_VERSION: i64 = 24;
+pub(super) const SCHEMA_VERSION: i64 = 25;
 
 impl Db {
     pub(super) fn migrate(&self) -> Result<()> {
@@ -149,6 +149,18 @@ impl Db {
         if current < 24 {
             tx.execute_batch(MIGRATION_V24)?;
         }
+        if current < 25 {
+            // Same guard as V21: an old fixture may not have `nodes` at all,
+            // and there is no path to materialise without it.
+            let has_nodes: bool = tx.query_row(
+                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'nodes'",
+                [],
+                |row| row.get::<_, i64>(0),
+            )? > 0;
+            if has_nodes {
+                tx.execute_batch(MIGRATION_V25)?;
+            }
+        }
         tx.execute(
             "INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
              ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -670,3 +682,46 @@ CREATE TABLE IF NOT EXISTS own_sealed_rev (
   sealed_at   INTEGER NOT NULL
 );
 ";
+
+/// V25: `nodes.path` — a node's mountpoint-relative path, stored rather than
+/// walked.
+///
+/// `path_of` was a recursive CTE per node, and search ran it once per candidate
+/// (up to a thousand per keystroke) on top of the one the write path already ran
+/// per descendant of every folder upsert. The value only changes when a node's
+/// name or parent changes, which is exactly when the write path is already
+/// walking the subtree — so it belongs in a column.
+///
+/// The backfill is the same walk V21 used for `nodes_fts`, minus the trashed
+/// filter: a trashed node stays out of the index but still has a path. Roots get
+/// `''` (their name is the mount), and a node whose parent is merely uncached
+/// keeps its own name — both matching what `path_of` returned at runtime.
+///
+/// Materialised into a temp table rather than written by a correlated subquery,
+/// so the recursive walk runs once instead of once per row.
+const MIGRATION_V25: &str = "
+ALTER TABLE nodes ADD COLUMN path TEXT;
+
+CREATE TEMP TABLE node_paths (uid TEXT PRIMARY KEY, path TEXT NOT NULL);
+
+INSERT INTO node_paths (uid, path)
+WITH RECURSIVE paths(uid, path, depth) AS (
+  SELECT n.uid, CASE WHEN n.parent_uid IS NULL THEN '' ELSE n.name END, 0
+    FROM nodes n
+   WHERE n.parent_uid IS NULL
+      OR NOT EXISTS (SELECT 1 FROM nodes p WHERE p.uid = n.parent_uid)
+  UNION ALL
+  SELECT n.uid,
+         CASE WHEN paths.path = '' THEN n.name ELSE paths.path || '/' || n.name END,
+         paths.depth + 1
+    FROM nodes n JOIN paths ON n.parent_uid = paths.uid
+   WHERE paths.depth < 256
+)
+SELECT uid, path FROM paths;
+
+UPDATE nodes
+   SET path = (SELECT p.path FROM node_paths p WHERE p.uid = nodes.uid)
+ WHERE EXISTS (SELECT 1 FROM node_paths p WHERE p.uid = nodes.uid);
+
+DROP TABLE node_paths;
+";
diff --git a/crates/pdfs-core/src/db/mod.rs b/crates/pdfs-core/src/db/mod.rs
index e5a4099..4b3d86d 100644
--- a/crates/pdfs-core/src/db/mod.rs
+++ b/crates/pdfs-core/src/db/mod.rs
@@ -2,9 +2,17 @@
 //! inode bookkeeping, full-text search, content-cache LRU tracking, and pins.
 //!
 //! Only the daemon (`pdfs-fuse`) opens this for writes; the GUI and CLI reach
-//! the same data through the control socket. The connection is wrapped in a
-//! `Mutex` because the FUSE callbacks are synchronous and already serialize
-//! behind the `State` lock, so a connection pool would be overkill.
+//! the same data through the control socket. There is one write connection
+//! behind a `Mutex` — SQLite allows exactly one writer, and the single-instance
+//! `flock` makes sure it is ours — and a small pool of read-only connections
+//! behind [`Db::read`], which every `SELECT`-only method uses.
+//!
+//! The pool is what makes WAL worth having. The original justification for a
+//! single connection was that "FUSE callbacks already serialize behind the
+//! `State` lock"; that stopped being true when reads moved onto an 11-thread
+//! worker pool, and a lookup then queued behind whatever listing happened to be
+//! committing. Under WAL a reader never blocks the writer and vice versa, so the
+//! only thing left serialising them was this process's own mutex.
 //!
 //! This module is the P0 foundation: it opens the database, enables WAL, and
 //! applies the forward-only schema migrations. Write-through of nodes, the
@@ -88,6 +96,10 @@ const WAL_AUTOCHECKPOINT_PAGES: i64 = 500;
 /// the inner `Mutex<Connection>`.
 pub struct Db {
     conn: Mutex<Connection>,
+    /// Read-only connections for `SELECT`-only methods. `None` for an in-memory
+    /// database, where a second connection would open a different, empty one —
+    /// those fall back to the write connection.
+    readers: Option<ReadPool>,
     /// The single-writer lock, held open for as long as this handle lives.
     ///
     /// Never read; dropping the `File` is what releases the `flock`, so it has
@@ -96,6 +108,101 @@ pub struct Db {
     _single_writer: Option<File>,
 }
 
+/// Idle read-only connections, opened on demand and kept for reuse.
+///
+/// Not a bounded pool: the number in flight is bounded by the number of threads
+/// that can be inside a read at once (the FUSE worker lanes plus the control
+/// handlers), and blocking one of those on a permit would reintroduce exactly
+/// the queueing this exists to remove. Only the *idle* set is capped, so a burst
+/// does not leave a connection per thread parked forever.
+struct ReadPool {
+    path: PathBuf,
+    idle: Mutex<Vec<Connection>>,
+}
+
+/// How many idle read connections to keep. Enough that the steady state never
+/// reopens, small enough to be unremarkable.
+const MAX_IDLE_READERS: usize = 4;
+
+impl ReadPool {
+    fn take(&self) -> Option<Connection> {
+        if let Some(conn) = self.idle.lock().pop() {
+            return Some(conn);
+        }
+        match open_read_only(&self.path) {
+            Ok(conn) => Some(conn),
+            Err(error) => {
+                // Falling back to the write connection is correct, just slower,
+                // so this is a warning and not a failed read.
+                tracing::warn!(%error, "could not open a read-only connection");
+                None
+            }
+        }
+    }
+
+    fn put(&self, conn: Connection) {
+        let mut idle = self.idle.lock();
+        if idle.len() < MAX_IDLE_READERS {
+            idle.push(conn);
+        }
+    }
+}
+
+/// A connection to read through: one borrowed from [`ReadPool`], or the write
+/// connection when there is no pool (in-memory) or opening one failed.
+///
+/// Derefs to `Connection`, so a method converts by swapping `self.conn.lock()`
+/// for `self.read()` and nothing else.
+pub(super) struct Reader<'a> {
+    pooled: Option<Connection>,
+    writer: Option<parking_lot::MutexGuard<'a, Connection>>,
+    pool: Option<&'a ReadPool>,
+}
+
+impl std::ops::Deref for Reader<'_> {
+    type Target = Connection;
+
+    fn deref(&self) -> &Connection {
+        self.pooled
+            .as_ref()
+            .or(self.writer.as_deref())
+            .expect("a Reader always holds one of the two")
+    }
+}
+
+impl Drop for Reader<'_> {
+    fn drop(&mut self) {
+        if let (Some(conn), Some(pool)) = (self.pooled.take(), self.pool) {
+            pool.put(conn);
+        }
+    }
+}
+
+/// Open a second connection to the same file for reads only.
+///
+/// `SQLITE_OPEN_READ_ONLY` plus `query_only` is belt and braces: the flag stops
+/// a routing mistake at the file, the pragma stops it at the statement. The rest
+/// mirror [`Db::open_configured`] — a read connection has its own page cache,
+/// its own temp store and its own busy timeout, and gets nothing from the
+/// writer's.
+///
+/// `journal_mode` is deliberately absent: it is a property of the database file,
+/// which the writer already set to WAL, and a read-only connection cannot change
+/// it anyway.
+fn open_read_only(path: &Path) -> Result<Connection> {
+    use rusqlite::OpenFlags;
+    let conn = Connection::open_with_flags(
+        path,
+        OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
+    )?;
+    conn.busy_timeout(BUSY_TIMEOUT)?;
+    conn.pragma_update(None, "cache_size", -CACHE_SIZE_KIB)?;
+    conn.pragma_update(None, "temp_store", "MEMORY")?;
+    conn.pragma_update(None, "mmap_size", MMAP_SIZE)?;
+    conn.pragma_update(None, "query_only", true)?;
+    Ok(conn)
+}
+
 impl Db {
     /// Open (creating if absent) the database at `path`, enable WAL, and bring
     /// the schema up to [`SCHEMA_VERSION`].
@@ -125,10 +232,39 @@ impl Db {
         };
         Ok(Self {
             conn: Mutex::new(conn),
+            // No pool for `:memory:`. A second connection to that name is a
+            // second, empty database rather than another view of this one, so
+            // every routed read would find a schema that does not exist.
+            readers: (!is_in_memory(path)).then(|| ReadPool {
+                path: path.to_path_buf(),
+                idle: Mutex::new(Vec::new()),
+            }),
             _single_writer: single_writer,
         })
     }
 
+    /// A connection for a `SELECT`-only method.
+    ///
+    /// Reads see the last committed state, which is the same thing they saw
+    /// through the write connection: every method here commits before it
+    /// returns, and nothing holds a transaction open across calls.
+    pub(super) fn read(&self) -> Reader<'_> {
+        if let Some(pool) = &self.readers
+            && let Some(conn) = pool.take()
+        {
+            return Reader {
+                pooled: Some(conn),
+                writer: None,
+                pool: Some(pool),
+            };
+        }
+        Reader {
+            pooled: None,
+            writer: Some(self.conn.lock()),
+            pool: None,
+        }
+    }
+
     /// Open the file, apply every PRAGMA, and migrate. Split out of
     /// [`open`](Self::open) so the corruption path can retry it verbatim.
     fn open_configured(path: &Path) -> Result<Connection> {
@@ -177,6 +313,7 @@ impl Db {
 
         let db = Self {
             conn: Mutex::new(conn),
+            readers: None,
             _single_writer: None,
         };
         db.migrate()?;
@@ -200,6 +337,7 @@ impl Db {
         conn.pragma_update(None, "foreign_keys", "ON")?;
         let db = Self {
             conn: Mutex::new(conn),
+            readers: None,
             _single_writer: None,
         };
         db.migrate()?;
diff --git a/crates/pdfs-core/src/db/mounts.rs b/crates/pdfs-core/src/db/mounts.rs
index 6031c37..8860067 100644
--- a/crates/pdfs-core/src/db/mounts.rs
+++ b/crates/pdfs-core/src/db/mounts.rs
@@ -74,7 +74,7 @@ impl Db {
 
     /// Every local location, with device fields joined from `sync_folder`.
     pub fn mount_list(&self) -> Result<Vec<MountSpec>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT
                  m.id,
diff --git a/crates/pdfs-core/src/db/nodes.rs b/crates/pdfs-core/src/db/nodes.rs
index a7fb182..a669c0f 100644
--- a/crates/pdfs-core/src/db/nodes.rs
+++ b/crates/pdfs-core/src/db/nodes.rs
@@ -10,7 +10,9 @@ use crate::{Access, Result};
 use proton_drive_rs::proton_sdk::ids::{LinkId, NodeUid, VolumeId};
 use proton_drive_rs::{Node, NodeKind};
 
-use super::utils::{TRIGRAM_MIN, collect_pairs, like_escape, pair, path_of};
+use super::utils::{
+    HitRow, TRIGRAM_MIN, collect_hits, hit_row, join_path, like_escape, path_of, walk_path_of,
+};
 
 pub struct StoredNode {
     pub node: Node,
@@ -59,7 +61,7 @@ impl Db {
     /// flag. Synthetic shared listings use this to serve the last completed
     /// snapshot while offline, including after a live event expired its TTL.
     pub fn visible_children(&self, parent: &NodeUid) -> Result<Vec<Node>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT node_json FROM nodes
               WHERE parent_uid = ?1 AND trashed = 0 AND node_json IS NOT NULL
@@ -285,7 +287,7 @@ impl Db {
     /// Check if a folder node has any non-trashed children in the database.
     pub fn has_children(&self, parent_uid: &NodeUid) -> Result<bool> {
         let uid_str = parent_uid.to_string();
-        let conn = self.conn.lock();
+        let conn = self.read();
         let count: i64 = conn.query_row(
             "SELECT COUNT(*) FROM nodes WHERE parent_uid = ?1 AND trashed = 0",
             params![uid_str],
@@ -294,6 +296,25 @@ impl Db {
         Ok(count > 0)
     }
 
+    /// A node's mountpoint-relative path, or `None` when it is not cached.
+    ///
+    /// One indexed read of the stored `path`. Callers that need to relate many
+    /// nodes to a handful of roots — search decorating each hit with its
+    /// mountpoint — resolve the roots once through this and do the rest as
+    /// string arithmetic, instead of a `path_relative_to` walk per pair.
+    pub fn node_path(&self, uid: &str) -> Result<Option<String>> {
+        let conn = self.read();
+        let exists: Option<i64> = conn
+            .query_row("SELECT 1 FROM nodes WHERE uid = ?1", params![uid], |r| {
+                r.get(0)
+            })
+            .optional()?;
+        if exists.is_none() {
+            return Ok(None);
+        }
+        Ok(Some(path_of(&conn, uid)?))
+    }
+
     /// Resolve `uid` relative to an ancestor node.
     ///
     /// This is used when a remote subtree has its own local sync mount: the
@@ -302,7 +323,7 @@ impl Db {
     /// chain keeps callers from accidentally joining a Drive-wide path onto the
     /// wrong mountpoint. The ancestor itself resolves to the empty path.
     pub fn path_relative_to(&self, ancestor_uid: &str, uid: &str) -> Result<Option<String>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "WITH RECURSIVE anc(uid, parent_uid, name, depth) AS (
                SELECT uid, parent_uid, name, 0 FROM nodes WHERE uid = ?1
@@ -338,15 +359,15 @@ impl Db {
         if query.is_empty() {
             return Ok(Vec::new());
         }
-        let conn = self.conn.lock();
-        let rows: Vec<(String, String)> = if query.chars().count() < TRIGRAM_MIN {
+        let conn = self.read();
+        let rows: Vec<HitRow> = if query.chars().count() < TRIGRAM_MIN {
             let pat = format!("%{}%", like_escape(query));
             let mut stmt = conn.prepare(
-                "SELECT node_json, uid FROM nodes
+                "SELECT node_json, uid, path FROM nodes
                  WHERE name LIKE ?1 ESCAPE '\\' AND trashed = 0 AND node_json IS NOT NULL
                  ORDER BY name LIMIT ?2",
             )?;
-            collect_pairs(stmt.query_map(params![pat, limit as i64], pair)?)?
+            collect_hits(stmt.query_map(params![pat, limit as i64], hit_row)?)?
         } else {
             // Escape double quotes and quote each term, then combine with AND so
             // all terms must match but can appear in any order or position.
@@ -356,18 +377,21 @@ impl Db {
                 .collect::<Vec<_>>()
                 .join(" AND ");
             let mut stmt = conn.prepare(
-                "SELECT n.node_json, n.uid
+                "SELECT n.node_json, n.uid, n.path
                  FROM nodes_fts f JOIN nodes n ON n.rowid = f.rowid
                  WHERE f.name MATCH ?1 AND n.trashed = 0 AND n.node_json IS NOT NULL
                  ORDER BY f.rank LIMIT ?2",
             )?;
-            collect_pairs(stmt.query_map(params![phrase, limit as i64], pair)?)?
+            collect_hits(stmt.query_map(params![phrase, limit as i64], hit_row)?)?
         };
 
         let mut hits = Vec::with_capacity(rows.len());
-        for (json, uid) in rows {
+        for (json, uid, path) in rows {
             let node: Node = serde_json::from_str(&json)?;
-            let path = path_of(&conn, &uid)?;
+            let path = match path {
+                Some(path) => path,
+                None => walk_path_of(&conn, &uid)?,
+            };
             hits.push(SearchHit { node, path });
         }
         Ok(hits)
@@ -382,26 +406,26 @@ impl Db {
         if query.is_empty() || limit == 0 {
             return Ok(Vec::new());
         }
-        let conn = self.conn.lock();
+        let conn = self.read();
         let terms = candidate_trigrams(query);
         let lane_limit = limit.div_ceil(2);
-        let mut rows: Vec<(String, String)> = if terms.is_empty() {
+        let mut rows: Vec<HitRow> = if terms.is_empty() {
             let pat = format!("%{}%", like_escape(query));
             let mut stmt = conn.prepare(
-                "SELECT node_json, uid FROM nodes
+                "SELECT node_json, uid, path FROM nodes
                  WHERE name LIKE ?1 ESCAPE '\\' AND trashed = 0 AND node_json IS NOT NULL
                  ORDER BY name LIMIT ?2",
             )?;
-            collect_pairs(stmt.query_map(params![pat, limit as i64], pair)?)?
+            collect_hits(stmt.query_map(params![pat, limit as i64], hit_row)?)?
         } else {
             let expression = terms.join(" OR ");
             let mut stmt = conn.prepare(
-                "SELECT n.node_json, n.uid
+                "SELECT n.node_json, n.uid, n.path
                    FROM nodes_fts f JOIN nodes n ON n.rowid = f.rowid
                   WHERE nodes_fts MATCH ?1 AND n.trashed = 0 AND n.node_json IS NOT NULL
                   ORDER BY f.rank LIMIT ?2",
             )?;
-            collect_pairs(stmt.query_map(params![expression, lane_limit as i64], pair)?)?
+            collect_hits(stmt.query_map(params![expression, lane_limit as i64], hit_row)?)?
         };
         // Trigram rank does not know that a name *starts* with the query, and a
         // short substitution can destroy every trigram outright (`vedio` vs
@@ -411,7 +435,7 @@ impl Db {
             && let Some(first) = query.chars().next()
         {
             let mut stmt = conn.prepare(
-                "SELECT node_json, uid FROM nodes
+                "SELECT node_json, uid, path FROM nodes
                  WHERE name COLLATE NOCASE LIKE ?1 ESCAPE '\\'
                    AND trashed = 0 AND node_json IS NOT NULL
                  ORDER BY name COLLATE NOCASE LIMIT ?2",
@@ -421,23 +445,26 @@ impl Db {
                 format!("{}%", like_escape(query)),
                 format!("{}%", like_escape(&first.to_string())),
             ] {
-                extra.extend(collect_pairs(
-                    stmt.query_map(params![pattern, lane_limit as i64], pair)?,
+                extra.extend(collect_hits(
+                    stmt.query_map(params![pattern, lane_limit as i64], hit_row)?,
                 )?);
             }
             let trigram = std::mem::take(&mut rows);
             for row in extra.into_iter().chain(trigram) {
-                if !rows.iter().any(|(_, uid)| uid == &row.1) {
+                if !rows.iter().any(|(_, uid, _)| uid == &row.1) {
                     rows.push(row);
                 }
             }
             rows.truncate(limit);
         }
         rows.into_iter()
-            .map(|(json, uid)| {
+            .map(|(json, uid, path)| {
                 Ok(SearchHit {
                     node: serde_json::from_str(&json)?,
-                    path: path_of(&conn, &uid)?,
+                    path: match path {
+                        Some(path) => path,
+                        None => walk_path_of(&conn, &uid)?,
+                    },
                 })
             })
             .collect()
@@ -457,7 +484,7 @@ impl Db {
 
     /// Load every persisted node for cold-start hydration of the `State` maps.
     pub fn load_all(&self) -> Result<Vec<StoredNode>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt =
             conn.prepare("SELECT node_json, listed FROM nodes WHERE node_json IS NOT NULL")?;
         let rows = stmt.query_map([], |row| {
@@ -478,7 +505,7 @@ impl Db {
     /// when the API is unreachable, so the mount can still serve the cached tree
     /// (offline.md Phase 1).
     pub fn node_by_uid(&self, uid: &str) -> Result<Option<Node>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let json: Option<String> = conn
             .query_row(
                 "SELECT node_json FROM nodes WHERE uid = ?1 AND node_json IS NOT NULL",
@@ -496,7 +523,7 @@ impl Db {
     /// The daemon resumes from this on restart instead of reseeding to the
     /// server head, so changes made while unmounted are still applied (P2).
     pub fn children_if_listed(&self, parent: &NodeUid) -> Result<Option<Vec<Node>>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let listed: Option<i64> = conn
             .query_row(
                 "SELECT listed FROM nodes WHERE uid = ?1",
@@ -529,13 +556,57 @@ impl Db {
     // in sync on every store/read/evict, so it is authoritative for eviction.
 }
 
+/// What a node's row looked like before this upsert, for the fields that decide
+/// whether the subtree below it has to be touched at all.
+struct PriorRow {
+    parent_uid: Option<String>,
+    name: String,
+    path: Option<String>,
+    trashed: bool,
+}
+
 fn upsert_node_tx(tx: &Transaction<'_>, node: &Node) -> Result<()> {
     let json = serde_json::to_string(node)?;
     let uid = node.uid.to_string();
+    let parent_uid = node.parent_uid.as_ref().map(|u| u.to_string());
+
+    let prior: Option<PriorRow> = tx
+        .query_row(
+            "SELECT parent_uid, name, path, trashed FROM nodes WHERE uid = ?1",
+            params![uid],
+            |row| {
+                Ok(PriorRow {
+                    parent_uid: row.get(0)?,
+                    name: row.get(1)?,
+                    path: row.get(2)?,
+                    trashed: row.get::<_, i64>(3)? != 0,
+                })
+            },
+        )
+        .optional()?;
+
+    // A node's path is its parent's path plus its own name — one indexed lookup,
+    // where resolving it from scratch is a recursive walk to the root. A node
+    // whose parent is not cached (a device folder's root lives in `device`, never
+    // in `nodes`) starts a path of its own, which is what the walk did too.
+    let path = match &parent_uid {
+        None => String::new(),
+        Some(parent) => {
+            let parent_path: Option<Option<String>> = tx
+                .query_row(
+                    "SELECT path FROM nodes WHERE uid = ?1",
+                    params![parent],
+                    |row| row.get(0),
+                )
+                .optional()?;
+            join_path(parent_path.flatten().as_deref().unwrap_or(""), &node.name)
+        }
+    };
+
     tx.execute(
         "INSERT INTO nodes
-           (uid, parent_uid, name, is_dir, size, mtime, trashed, node_json)
-         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
+           (uid, parent_uid, name, is_dir, size, mtime, trashed, node_json, path)
+         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
          ON CONFLICT(uid) DO UPDATE SET
            parent_uid = excluded.parent_uid,
            name       = excluded.name,
@@ -543,65 +614,116 @@ fn upsert_node_tx(tx: &Transaction<'_>, node: &Node) -> Result<()> {
            size       = excluded.size,
            mtime      = excluded.mtime,
            trashed    = excluded.trashed,
-           node_json  = excluded.node_json",
+           node_json  = excluded.node_json,
+           path       = excluded.path",
         params![
             uid,
-            node.parent_uid.as_ref().map(|u| u.to_string()),
+            parent_uid,
             node.name,
             node.is_folder() as i64,
             node_size(node),
             node.modification_time,
             node.trashed as i64,
             json,
+            path,
         ],
     )?;
-    // FTS5 has no UPSERT. Folder moves also change every descendant path, so
-    // refresh the whole affected subtree while keeping rowid deletes indexed.
     let rowid: i64 = tx.query_row(
         "SELECT rowid FROM nodes WHERE uid = ?1",
         params![uid],
         |row| row.get(0),
     )?;
-    let mut affected = vec![rowid];
-    if node.is_folder() {
-        // The depth cap is a liveness guard, not a policy: `UNION ALL` over a
-        // parent cycle (a's parent is b, b's parent is a — corrupt data, but the
-        // API can hand it to us) never terminates, and this runs inside the
-        // write transaction that holds the daemon's only SQLite connection.
-        // Real Drive trees are nowhere near this deep.
-        let mut stmt = tx.prepare(
-            "WITH RECURSIVE descendants(rowid, uid, depth) AS (
-               SELECT rowid, uid, 0 FROM nodes WHERE parent_uid = ?1
-               UNION ALL
-               SELECT n.rowid, n.uid, d.depth + 1 FROM nodes n
-                 JOIN descendants d ON n.parent_uid = d.uid
-                WHERE d.depth < 256
-             ) SELECT rowid FROM descendants",
-        )?;
-        for descendant in stmt.query_map(params![uid], |row| row.get(0))? {
-            affected.push(descendant?);
-        }
-    }
-    for affected_rowid in affected {
+
+    // FTS5 has no UPSERT, so the node's own row is always replaced.
+    let indexable = node_is_indexable_tx(tx, &uid)?;
+    tx.execute("DELETE FROM nodes_fts WHERE rowid = ?1", params![rowid])?;
+    if indexable {
         tx.execute(
-            "DELETE FROM nodes_fts WHERE rowid = ?1",
-            params![affected_rowid],
+            "INSERT INTO nodes_fts (rowid, name, path) VALUES (?1, ?2, ?3)",
+            params![rowid, node.name, path],
         )?;
-        let indexed: Option<(String, String)> = tx
-            .query_row(
-                "SELECT name, uid FROM nodes WHERE rowid = ?1",
-                params![affected_rowid],
-                |row| Ok((row.get(0)?, row.get(1)?)),
-            )
-            .optional()?;
-        if let Some((name, indexed_uid)) = indexed
-            && node_is_indexable_tx(tx, &indexed_uid)?
-        {
-            let path = path_of(tx, &indexed_uid)?;
-            tx.execute(
-                "INSERT INTO nodes_fts (rowid, name, path) VALUES (?1, ?2, ?3)",
-                params![affected_rowid, name, path],
-            )?;
+    }
+
+    // Everything below only matters when the subtree's paths or reachability
+    // actually moved. Re-listing a folder upserts it unchanged, and paying for a
+    // full descendant walk on each of those was most of what a large tree spent
+    // its time on.
+    let subtree_moved = prior.is_none_or(|prior| {
+        prior.parent_uid != parent_uid
+            || prior.name != node.name
+            || prior.path.as_deref() != Some(path.as_str())
+            || prior.trashed != node.trashed
+    });
+    if node.is_folder() && subtree_moved {
+        reindex_subtree_tx(tx, &uid, &path, indexable)?;
+    }
+    Ok(())
+}
+
+/// Rewrite every descendant's stored path and search-index row after `folder`
+/// moved, was renamed, or changed reachability.
+///
+/// One recursive walk carries both the new path and whether anything on the way
+/// down is trashed, so each descendant costs two trivial statements instead of a
+/// `path_of` walk plus an ancestor walk of its own. `folder_indexable` is the
+/// answer `node_is_indexable_tx` gave for the folder itself: what is above the
+/// subtree is the same for every node in it.
+///
+/// The depth cap is a liveness guard, not a policy: `UNION ALL` over a parent
+/// cycle (a's parent is b, b's parent is a — corrupt data, but the API can hand
+/// it to us) never terminates, and this runs inside the write transaction that
+/// holds the daemon's only SQLite connection. Real Drive trees are nowhere near
+/// this deep.
+fn reindex_subtree_tx(
+    tx: &Transaction<'_>,
+    folder: &str,
+    folder_path: &str,
+    folder_indexable: bool,
+) -> Result<()> {
+    let descendants: Vec<(i64, String, bool)> = {
+        let mut stmt = tx.prepare(
+            "WITH RECURSIVE sub(rowid, uid, name, path, blocked, depth) AS (
+               SELECT n.rowid, n.uid, n.name,
+                      CASE WHEN ?2 = '' THEN n.name ELSE ?2 || '/' || n.name END,
+                      n.trashed, 0
+                 FROM nodes n WHERE n.parent_uid = ?1
+               UNION ALL
+               SELECT n.rowid, n.uid, n.name,
+                      sub.path || '/' || n.name,
+                      MAX(sub.blocked, n.trashed), sub.depth + 1
+                 FROM nodes n JOIN sub ON n.parent_uid = sub.uid
+                WHERE sub.depth < 256
+             )
+             SELECT rowid, path, blocked FROM sub",
+        )?;
+        let rows = stmt.query_map(params![folder, folder_path], |row| {
+            Ok((
+                row.get::<_, i64>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, i64>(2)? != 0,
+            ))
+        })?;
+        let mut out = Vec::new();
+        for row in rows {
+            out.push(row?);
+        }
+        out
+    };
+    if descendants.is_empty() {
+        return Ok(());
+    }
+
+    let mut set_path = tx.prepare("UPDATE nodes SET path = ?2 WHERE rowid = ?1")?;
+    let mut unindex = tx.prepare("DELETE FROM nodes_fts WHERE rowid = ?1")?;
+    let mut index = tx.prepare(
+        "INSERT INTO nodes_fts (rowid, name, path)
+         SELECT rowid, name, ?2 FROM nodes WHERE rowid = ?1",
+    )?;
+    for (rowid, path, blocked) in descendants {
+        set_path.execute(params![rowid, path])?;
+        unindex.execute(params![rowid])?;
+        if folder_indexable && !blocked {
+            index.execute(params![rowid, path])?;
         }
     }
     Ok(())
diff --git a/crates/pdfs-core/src/db/ops.rs b/crates/pdfs-core/src/db/ops.rs
index 7b23efc..041e244 100644
--- a/crates/pdfs-core/src/db/ops.rs
+++ b/crates/pdfs-core/src/db/ops.rs
@@ -157,7 +157,7 @@ pub struct AttachedBlob {
 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<bool> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         conn.query_row(
             "SELECT EXISTS(
                 SELECT 1 FROM pending_op WHERE uid = ?1 AND kind = ?2
@@ -170,7 +170,7 @@ impl Db {
 
     /// 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.conn.lock();
+        let conn = self.read();
         conn.query_row(
             "SELECT meta_json FROM pending_op WHERE uid = ?1 AND kind = ?2",
             params![uid, kind],
@@ -369,7 +369,7 @@ impl Db {
 
     /// Check if a queued create or mkdir op exists for `uid`.
     pub fn has_create_op(&self, uid: &str) -> Result<bool> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         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],
@@ -385,7 +385,7 @@ impl Db {
     /// 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.conn.lock();
+        let conn = self.read();
         let count: i64 = conn.query_row(
             "SELECT COUNT(*) FROM pending_op WHERE uid = ?1",
             params![uid],
@@ -481,7 +481,7 @@ impl Db {
     /// 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.conn.lock();
+        let conn = self.read();
         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
@@ -517,7 +517,7 @@ impl Db {
     /// debounced or backed-off op becomes eligible rather than waiting the full
     /// idle-poll interval.
     pub fn earliest_due_at(&self) -> Result<Option<i64>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let ts: Option<i64> = conn
             .query_row(
                 &format!(
@@ -537,7 +537,7 @@ impl Db {
     /// 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.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT id, kind, uid, parent_uid, name, blob_path, meta_json, created_at,
                     attempts, last_error, next_attempt_at
@@ -570,7 +570,7 @@ impl Db {
     /// 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.conn.lock();
+        let conn = self.read();
         let uploads = conn.query_row(
             "SELECT COUNT(*) FROM pending_op WHERE kind IN (?1, ?2)",
             params![OP_REVISION, OP_CREATE],
@@ -659,7 +659,7 @@ impl Db {
     /// `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.conn.lock();
+        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))?;
@@ -700,7 +700,7 @@ impl Db {
     /// 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.conn.lock();
+        let conn = self.read();
         Ok(conn
             .query_row(
                 "SELECT revision_id FROM own_sealed_rev
diff --git a/crates/pdfs-core/src/db/photos.rs b/crates/pdfs-core/src/db/photos.rs
index 77ddc31..56bf140 100644
--- a/crates/pdfs-core/src/db/photos.rs
+++ b/crates/pdfs-core/src/db/photos.rs
@@ -115,7 +115,7 @@ impl Db {
         range: Option<(i64, i64)>,
         favorites: bool,
     ) -> Result<Vec<StoredPhoto>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         // Built up so any combination of the optional filters is one indexed
         // query rather than a statement per case.
         let mut sql = String::from(
@@ -174,7 +174,7 @@ impl Db {
         &self,
         kind: Option<crate::control::PhotoKind>,
     ) -> Result<Vec<(i32, i32, usize)>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let (filter, binds): (&str, Vec<i64>) = match kind {
             Some(k) => (" WHERE kind = ?1", vec![k.as_i64()]),
             None => ("", Vec::new()),
@@ -201,7 +201,7 @@ impl Db {
     /// Per-tab counts for the Photos page subtitle: `(photos, videos, raw)`.
     pub fn photos_counts(&self) -> Result<(usize, usize, usize)> {
         use crate::control::PhotoKind;
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare("SELECT kind, COUNT(*) FROM photos GROUP BY kind")?;
         let rows = stmt.query_map([], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)))?;
         let (mut photos, mut videos, mut raw) = (0usize, 0usize, 0usize);
@@ -223,7 +223,7 @@ impl Db {
         if uids.is_empty() {
             return Ok(Vec::new());
         }
-        let conn = self.conn.lock();
+        let conn = self.read();
         let placeholders = vec!["?"; uids.len()].join(",");
         let mut stmt = conn.prepare(&format!(
             "SELECT uid, capture_time, name, ratio, thumb_state, kind, favorite FROM photos
@@ -273,7 +273,7 @@ impl Db {
 
     /// Number of photos in the persisted timeline.
     pub fn photos_count(&self) -> Result<usize> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let n: i64 = conn.query_row("SELECT COUNT(*) FROM photos", [], |r| r.get(0))?;
         Ok(n.max(0) as usize)
     }
diff --git a/crates/pdfs-core/src/db/pins.rs b/crates/pdfs-core/src/db/pins.rs
index a89f046..0b201db 100644
--- a/crates/pdfs-core/src/db/pins.rs
+++ b/crates/pdfs-core/src/db/pins.rs
@@ -40,7 +40,7 @@ impl Db {
 
     /// Every directly-pinned entry, ordered by uid.
     pub fn pin_list(&self) -> Result<Vec<PinRow>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         // `recursive` is pin policy; the node's kind comes from `nodes`, and is
         // NULL for a pin whose node was never cached.
         let mut stmt = conn.prepare(
@@ -73,7 +73,7 @@ impl Db {
     /// terminates — while holding the daemon's only SQLite connection, on a path
     /// that runs per cached read.
     pub fn is_pinned(&self, uid: &str) -> Result<bool> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let direct: Option<i64> = conn
             .query_row("SELECT 1 FROM pins WHERE uid = ?1", params![uid], |r| {
                 r.get(0)
@@ -107,7 +107,7 @@ impl Db {
     /// every descendant of a recursively-pinned folder. Used to build the
     /// eviction-exempt set (the budget enforcer hashes these into cache keys).
     pub fn pinned_uids(&self) -> Result<Vec<String>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "WITH RECURSIVE sub(uid) AS (
                SELECT uid FROM pins WHERE recursive = 1
@@ -129,7 +129,7 @@ impl Db {
     /// Every descendant uid of `folder` (all depths), via a `parent_uid` CTE.
     /// Used to evict a recursively-pinned subtree's cached blobs on unpin.
     pub fn descendants(&self, folder: &str) -> Result<Vec<String>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "WITH RECURSIVE sub(uid) AS (
                SELECT uid FROM nodes WHERE parent_uid = ?1
diff --git a/crates/pdfs-core/src/db/share_access.rs b/crates/pdfs-core/src/db/share_access.rs
index 8a47bd4..d8a17aa 100644
--- a/crates/pdfs-core/src/db/share_access.rs
+++ b/crates/pdfs-core/src/db/share_access.rs
@@ -16,7 +16,7 @@ impl Db {
     /// remains the restart/offline authority and every explicit role change is
     /// written through before this cache is updated.
     pub fn all_share_access(&self) -> Result<HashMap<NodeUid, Access>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare("SELECT root_uid, access FROM share_access")?;
         let rows = stmt.query_map([], |row| {
             Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
@@ -42,7 +42,7 @@ impl Db {
 
     /// Read the effective access recorded for a shared-tree root.
     pub fn share_access(&self, uid: &NodeUid) -> Result<Option<Access>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let value = conn
             .query_row(
                 "SELECT access FROM share_access WHERE root_uid = ?1",
@@ -96,7 +96,7 @@ impl Db {
     /// share root. `None` means the node itself is not persisted, which callers
     /// must not treat as owned: stale handles need to fail closed.
     pub fn effective_node_access(&self, uid: &NodeUid) -> Result<Option<Access>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let exists: bool = conn.query_row(
             "SELECT EXISTS(SELECT 1 FROM nodes WHERE uid = ?1)",
             [uid.to_string()],
diff --git a/crates/pdfs-core/src/db/state.rs b/crates/pdfs-core/src/db/state.rs
index d2a39fd..9742b4d 100644
--- a/crates/pdfs-core/src/db/state.rs
+++ b/crates/pdfs-core/src/db/state.rs
@@ -8,7 +8,7 @@ use crate::Result;
 
 impl Db {
     pub fn get_event_cursor(&self) -> Result<Option<String>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let v = conn
             .query_row(
                 "SELECT value FROM sync_state WHERE key = 'event_cursor'",
@@ -32,7 +32,7 @@ impl Db {
 
     /// Read a `sync_state` value as a string.
     pub fn state_str(&self, key: &str) -> Result<Option<String>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let v = conn
             .query_row(
                 "SELECT value FROM sync_state WHERE key = ?1",
@@ -66,7 +66,7 @@ impl Db {
     /// dropping it would leave a file that exists nowhere but this machine with
     /// nothing left to create it. (Writes to a node that is itself still queued
     pub fn state_i64(&self, key: &str) -> Result<Option<i64>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let v: Option<String> = conn
             .query_row(
                 "SELECT value FROM sync_state WHERE key = ?1",
diff --git a/crates/pdfs-core/src/db/sync.rs b/crates/pdfs-core/src/db/sync.rs
index 1aff0e1..55feab2 100644
--- a/crates/pdfs-core/src/db/sync.rs
+++ b/crates/pdfs-core/src/db/sync.rs
@@ -57,7 +57,7 @@ impl Db {
 
     /// Every synced folder, oldest first.
     pub fn sync_folder_list(&self) -> Result<Vec<StoredSyncFolder>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT id, local_path, remote_uid, remote_share_id, mode, pending_mode, state, last_sync
              FROM sync_folder ORDER BY id",
@@ -83,7 +83,7 @@ impl Db {
 
     /// Look up one synced folder by id.
     pub fn sync_folder_get(&self, id: i64) -> Result<Option<StoredSyncFolder>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         conn.query_row(
             "SELECT id, local_path, remote_uid, remote_share_id, mode, pending_mode, state, last_sync
              FROM sync_folder WHERE id = ?1",
@@ -166,7 +166,7 @@ impl Db {
 
     /// The whole per-file sync baseline for a folder, keyed by relative path.
     pub fn sync_entries(&self, folder_id: i64) -> Result<HashMap<String, StoredSyncEntry>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT rel_path, remote_uid, local_mtime, local_size, remote_rev, remote_hash
              FROM sync_entry WHERE folder_id = ?1",
@@ -196,7 +196,7 @@ impl Db {
     /// `sync_folder` and descendants come from the last successful
     /// `sync_entry` baseline.
     pub fn mirror_contains_uid(&self, uid: &str) -> Result<bool> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         conn.query_row(
             "SELECT EXISTS(
                  SELECT 1
diff --git a/crates/pdfs-core/src/db/tests.rs b/crates/pdfs-core/src/db/tests.rs
index 38ac293..55caac7 100644
--- a/crates/pdfs-core/src/db/tests.rs
+++ b/crates/pdfs-core/src/db/tests.rs
@@ -1509,6 +1509,7 @@ fn migration_v17_adds_share_access_to_a_v16_fixture() {
              DROP TABLE albums;
              DROP INDEX idx_photos_favorite;
              ALTER TABLE photos DROP COLUMN favorite;
+             ALTER TABLE nodes DROP COLUMN path;
              UPDATE sync_state SET value = '16' WHERE key = 'schema_version';",
         )
         .unwrap();
@@ -3282,3 +3283,265 @@ fn clear_state_prefix_takes_only_the_prefixed_keys() {
         "the listing's own stamp is not a per-album one"
     );
 }
+
+// --- `nodes.path`, and the subtree work it replaces (DB3) -------------------
+
+/// Read the stored path column directly: the point of these tests is that the
+/// column itself is right, not that `path_of` can still fall back to a walk.
+fn stored_path(db: &Db, link: &str) -> Option<String> {
+    use rusqlite::OptionalExtension;
+    db.with_conn(|conn| {
+        Ok(conn
+            .query_row(
+                "SELECT path FROM nodes WHERE uid = ?1",
+                [uid(link).to_string()],
+                |row| row.get::<_, Option<String>>(0),
+            )
+            .optional()?
+            .flatten())
+    })
+    .unwrap()
+}
+
+#[test]
+fn a_node_stores_its_path_when_it_is_written() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&folder("docs", Some("root"), "Documents"))
+        .unwrap();
+    db.upsert_node(&file("f1", "docs", "report.pdf", 1))
+        .unwrap();
+
+    // The root's name is the mount, so it contributes nothing.
+    assert_eq!(stored_path(&db, "root").as_deref(), Some(""));
+    assert_eq!(stored_path(&db, "docs").as_deref(), Some("Documents"));
+    assert_eq!(
+        stored_path(&db, "f1").as_deref(),
+        Some("Documents/report.pdf")
+    );
+}
+
+#[test]
+fn a_node_below_an_uncached_parent_starts_its_own_path() {
+    // A device folder's root lives in `device`/`sync_folder` and never gets a
+    // `nodes` row, so its children are the top of the cached chain.
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("orphan", Some("never-cached"), "Work"))
+        .unwrap();
+    db.upsert_node(&file("f1", "orphan", "notes.txt", 1))
+        .unwrap();
+
+    assert_eq!(stored_path(&db, "orphan").as_deref(), Some("Work"));
+    assert_eq!(stored_path(&db, "f1").as_deref(), Some("Work/notes.txt"));
+}
+
+#[test]
+fn renaming_a_folder_rewrites_every_descendant_path() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&folder("a", Some("root"), "Before"))
+        .unwrap();
+    db.upsert_node(&folder("b", Some("a"), "Inner")).unwrap();
+    db.upsert_node(&file("f1", "b", "deep.txt", 1)).unwrap();
+    assert_eq!(
+        stored_path(&db, "f1").as_deref(),
+        Some("Before/Inner/deep.txt")
+    );
+
+    db.upsert_node(&folder("a", Some("root"), "After")).unwrap();
+
+    assert_eq!(stored_path(&db, "a").as_deref(), Some("After"));
+    assert_eq!(stored_path(&db, "b").as_deref(), Some("After/Inner"));
+    assert_eq!(
+        stored_path(&db, "f1").as_deref(),
+        Some("After/Inner/deep.txt")
+    );
+    let hit = &db.search("deep.txt", 10).unwrap()[0];
+    assert_eq!(
+        hit.path, "After/Inner/deep.txt",
+        "the search index must carry the rewritten path too"
+    );
+}
+
+#[test]
+fn moving_a_folder_rewrites_every_descendant_path() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&folder("old", Some("root"), "Old")).unwrap();
+    db.upsert_node(&folder("new", Some("root"), "New")).unwrap();
+    db.upsert_node(&folder("moved", Some("old"), "Moved"))
+        .unwrap();
+    db.upsert_node(&file("f1", "moved", "deep.txt", 1)).unwrap();
+
+    db.upsert_node(&folder("moved", Some("new"), "Moved"))
+        .unwrap();
+
+    assert_eq!(stored_path(&db, "moved").as_deref(), Some("New/Moved"));
+    assert_eq!(
+        stored_path(&db, "f1").as_deref(),
+        Some("New/Moved/deep.txt")
+    );
+}
+
+#[test]
+fn trashing_a_folder_drops_its_subtree_from_the_index() {
+    let db = Db::open_in_memory().unwrap();
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&folder("a", Some("root"), "Holder"))
+        .unwrap();
+    db.upsert_node(&file("f1", "a", "hidden.txt", 1)).unwrap();
+    assert_eq!(db.search("hidden", 10).unwrap().len(), 1);
+
+    let mut trashed = folder("a", Some("root"), "Holder");
+    trashed.trashed = true;
+    db.upsert_node(&trashed).unwrap();
+
+    assert_eq!(
+        db.search("hidden", 10).unwrap().len(),
+        0,
+        "a file under a trashed folder is not reachable and must not be findable"
+    );
+
+    // …and comes back when the folder does.
+    db.upsert_node(&folder("a", Some("root"), "Holder"))
+        .unwrap();
+    assert_eq!(db.search("hidden", 10).unwrap().len(), 1);
+}
+
+/// The re-listing case, which is most of what the write path actually does: a
+/// folder upserted with the same name and parent must not touch its subtree.
+///
+/// Written as a ratio between two subtree sizes, so it fails on the scaling
+/// regression rather than on a slow machine. Before this, every re-listing paid
+/// a `path_of` walk plus an ancestor walk per descendant.
+#[test]
+fn re_listing_a_folder_does_not_scale_with_its_subtree() {
+    fn elapsed_for(children: usize) -> std::time::Duration {
+        let db = Db::open_in_memory().unwrap();
+        db.upsert_node(&folder("root", None, "My Files")).unwrap();
+        db.upsert_node(&folder("big", Some("root"), "Big")).unwrap();
+        let kids: Vec<Node> = (0..children)
+            .map(|i| file(&format!("k{i}"), "big", &format!("child-{i}.txt"), 1))
+            .collect();
+        db.upsert_nodes(&kids).unwrap();
+
+        let unchanged = folder("big", Some("root"), "Big");
+        let start = std::time::Instant::now();
+        for _ in 0..20 {
+            db.upsert_node(&unchanged).unwrap();
+        }
+        start.elapsed()
+    }
+
+    let small = elapsed_for(100);
+    let large = elapsed_for(2_000);
+    assert!(
+        large < small * 8,
+        "re-listing a folder scaled with its subtree: {small:?} for 100 children, \
+         {large:?} for 2000 (20x the descendants)"
+    );
+}
+
+#[test]
+fn migration_v25_backfills_paths_for_an_existing_tree() {
+    let path = std::env::temp_dir().join(format!(
+        "pdfs-db-v24-fixture-{}-{}.db",
+        std::process::id(),
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    ));
+    {
+        let db = Db::open(&path).unwrap();
+        db.upsert_node(&folder("root", None, "My Files")).unwrap();
+        db.upsert_node(&folder("docs", Some("root"), "Documents"))
+            .unwrap();
+        db.upsert_node(&file("f1", "docs", "report.pdf", 1))
+            .unwrap();
+        db.upsert_node(&folder("orphan", Some("never-cached"), "Work"))
+            .unwrap();
+        db.upsert_node(&file("f2", "orphan", "notes.txt", 1))
+            .unwrap();
+    }
+    {
+        // Put the file back in the state a released V24 database was in.
+        let conn = rusqlite::Connection::open(&path).unwrap();
+        conn.execute_batch(
+            "ALTER TABLE nodes DROP COLUMN path;
+             UPDATE sync_state SET value = '24' WHERE key = 'schema_version';",
+        )
+        .unwrap();
+    }
+
+    let db = Db::open(&path).unwrap();
+    assert_eq!(stored_path(&db, "root").as_deref(), Some(""));
+    assert_eq!(stored_path(&db, "docs").as_deref(), Some("Documents"));
+    assert_eq!(
+        stored_path(&db, "f1").as_deref(),
+        Some("Documents/report.pdf"),
+        "an existing tree must come out of the migration with the same paths \
+         the walk used to produce"
+    );
+    // A subtree whose top has an uncached parent is ordinary data, not an error.
+    assert_eq!(stored_path(&db, "orphan").as_deref(), Some("Work"));
+    assert_eq!(stored_path(&db, "f2").as_deref(), Some("Work/notes.txt"));
+
+    let _ = std::fs::remove_file(&path);
+    let _ = std::fs::remove_file(format!("{}.lock", path.display()));
+}
+
+// --- read-only connection pool (DB2) ---------------------------------------
+
+/// A read must not queue behind whatever happens to be writing.
+///
+/// The write connection is held for the whole measurement — which is what a
+/// large listing's commit does to it — while a reader asks for a search. Before
+/// the pool this was the same mutex, so the read waited for the writer;
+/// `busy_timeout` never entered into it, because the contention was this
+/// process's own lock rather than SQLite's.
+#[test]
+fn a_read_does_not_wait_for_the_write_connection() {
+    let path = std::env::temp_dir().join(format!(
+        "pdfs-db-readpool-{}-{}.db",
+        std::process::id(),
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos()
+    ));
+    let db = std::sync::Arc::new(Db::open(&path).unwrap());
+    db.upsert_node(&folder("root", None, "My Files")).unwrap();
+    db.upsert_node(&file("f1", "root", "findme.txt", 1))
+        .unwrap();
+
+    let held = std::time::Duration::from_millis(500);
+    let (tx, rx) = std::sync::mpsc::channel();
+    let writer = {
+        let db = db.clone();
+        std::thread::spawn(move || {
+            db.with_conn(|_conn| {
+                tx.send(()).unwrap();
+                std::thread::sleep(held);
+                Ok(())
+            })
+            .unwrap();
+        })
+    };
+    rx.recv().unwrap();
+
+    let start = std::time::Instant::now();
+    let hits = db.search("findme", 10).unwrap();
+    let waited = start.elapsed();
+    writer.join().unwrap();
+
+    assert_eq!(hits.len(), 1, "the read must still see committed data");
+    assert!(
+        waited < held / 2,
+        "the read waited {waited:?} for a write connection held {held:?}"
+    );
+
+    drop(db);
+    let _ = std::fs::remove_file(&path);
+    let _ = std::fs::remove_file(format!("{}.lock", path.display()));
+}
diff --git a/crates/pdfs-core/src/db/trash.rs b/crates/pdfs-core/src/db/trash.rs
index 1ff305f..583aeb9 100644
--- a/crates/pdfs-core/src/db/trash.rs
+++ b/crates/pdfs-core/src/db/trash.rs
@@ -39,7 +39,7 @@ impl Db {
     /// The persisted trash listing, folders first then by name — the order the
     /// Trash page shows it in.
     pub fn trash_list(&self) -> Result<Vec<StoredTrash>> {
-        let conn = self.conn.lock();
+        let conn = self.read();
         let mut stmt = conn.prepare(
             "SELECT uid, name, is_dir, size, mtime FROM trash
              ORDER BY is_dir DESC, name COLLATE NOCASE",
diff --git a/crates/pdfs-core/src/db/utils.rs b/crates/pdfs-core/src/db/utils.rs
index 154e711..c8500b5 100644
--- a/crates/pdfs-core/src/db/utils.rs
+++ b/crates/pdfs-core/src/db/utils.rs
@@ -1,6 +1,6 @@
 //! Small query helpers shared across the table modules.
 
-use rusqlite::{Connection, params};
+use rusqlite::{Connection, OptionalExtension, params};
 
 use crate::Result;
 
@@ -8,14 +8,19 @@ use crate::Result;
 /// grams), so short queries fall back to a `LIKE` scan over `nodes.name`.
 pub(super) const TRIGRAM_MIN: usize = 3;
 
-pub(super) fn pair(row: &rusqlite::Row) -> rusqlite::Result<(String, String)> {
-    Ok((row.get(0)?, row.get(1)?))
+/// One raw search row: the node's JSON, its uid, and its stored path — `None`
+/// only for a row written before the `path` column existed, which
+/// [`path_of`] repairs by walking.
+pub(super) type HitRow = (String, String, Option<String>);
+
+pub(super) fn hit_row(row: &rusqlite::Row) -> rusqlite::Result<HitRow> {
+    Ok((row.get(0)?, row.get(1)?, row.get(2)?))
 }
 
-/// Drain a `query_map` of [`pair`] rows into a `Vec`, propagating row errors.
-pub(super) fn collect_pairs(
-    rows: impl Iterator<Item = rusqlite::Result<(String, String)>>,
-) -> Result<Vec<(String, String)>> {
+/// Drain a `query_map` of [`hit_row`] rows into a `Vec`, propagating row errors.
+pub(super) fn collect_hits(
+    rows: impl Iterator<Item = rusqlite::Result<HitRow>>,
+) -> Result<Vec<HitRow>> {
     let mut out = Vec::new();
     for r in rows {
         out.push(r?);
@@ -37,16 +42,42 @@ pub(super) fn like_escape(s: &str) -> String {
 /// directly and so never passed through the index's cycle check.
 pub(super) const MAX_PATH_DEPTH: usize = 256;
 
-/// Resolve a node's mountpoint-relative path by walking `parent_uid` to the
-/// root via a recursive CTE. The root (the node with no parent) is excluded, so
-/// a top-level file `report.pdf` yields `"report.pdf"`, not `"My Files/report.pdf"`.
+/// Append `name` to a parent's stored path. An empty parent path is the root,
+/// whose name is the mount and so contributes nothing.
+pub(super) fn join_path(parent_path: &str, name: &str) -> String {
+    if parent_path.is_empty() {
+        name.to_string()
+    } else {
+        format!("{parent_path}/{name}")
+    }
+}
+
+/// Resolve a node's mountpoint-relative path. The root (the node with no
+/// parent) is excluded, so a top-level file `report.pdf` yields
+/// `"report.pdf"`, not `"My Files/report.pdf"`.
+///
+/// Reads the stored `nodes.path`, which the write path maintains (see
+/// `upsert_node_tx`). Falls back to [`walk_path_of`] for a row that predates the
+/// column or was never written through — a search serving a keystroke used to
+/// run that walk once per candidate, which is the reason the column exists.
+pub(super) fn path_of(conn: &Connection, uid: &str) -> Result<String> {
+    let stored: Option<Option<String>> = conn
+        .query_row("SELECT path FROM nodes WHERE uid = ?1", params![uid], |r| {
+            r.get(0)
+        })
+        .optional()?;
+    match stored.flatten() {
+        Some(path) => Ok(path),
+        None => walk_path_of(conn, uid),
+    }
+}
+
+/// Resolve a path by walking `parent_uid` to the root via a recursive CTE.
 ///
 /// A cycle in `parent_uid` is corrupt data the API can still hand us, and
 /// `UNION ALL` over one never terminates — the walk is therefore depth-capped
-/// and returns the truncated path rather than hanging the caller. That caller is
-/// usually a search serving a keystroke, holding the daemon's only SQLite
-/// connection.
-pub(super) fn path_of(conn: &Connection, uid: &str) -> Result<String> {
+/// and returns the truncated path rather than hanging the caller.
+pub(super) fn walk_path_of(conn: &Connection, uid: &str) -> Result<String> {
     let mut stmt = conn.prepare(&format!(
         "WITH RECURSIVE anc(uid, parent_uid, name, depth) AS (
            SELECT uid, parent_uid, name, 0 FROM nodes WHERE uid = ?1
diff --git a/crates/pdfs-fuse/src/devices.rs b/crates/pdfs-fuse/src/devices.rs
index 48c0ca9..ee89fe7 100644
--- a/crates/pdfs-fuse/src/devices.rs
+++ b/crates/pdfs-fuse/src/devices.rs
@@ -591,19 +591,8 @@ impl Core {
     pub(crate) fn fork_state(&self) -> Core {
         let mut fork = self.clone();
         fork.primary = false;
-        let share_access = self.state.lock().share_access.clone();
-        fork.state = Arc::new(Mutex::new(State {
-            entries: HashMap::new(),
-            by_uid: HashMap::new(),
-            children: HashMap::new(),
-            next_ino: 2,
-            active_writes: HashMap::new(),
-            handles: HashMap::new(),
-            next_fh: 1,
-            access_changes: std::collections::HashSet::new(),
-            share_access,
-            db: self.db.clone(),
-        }));
+        let share_access = self.state().share_access.clone();
+        fork.state = Arc::new(Mutex::new(State::new(self.db.clone(), share_access, 2)));
         // A fresh inode space needs a fresh notification channel: this fork's
         // session is the only one that knows these inodes, so it must be the one
         // notified about them. Filled in by `spawn_ondemand_mount`.
diff --git a/crates/pdfs-fuse/src/drain.rs b/crates/pdfs-fuse/src/drain.rs
index 5217726..26444cd 100644
--- a/crates/pdfs-fuse/src/drain.rs
+++ b/crates/pdfs-fuse/src/drain.rs
@@ -861,7 +861,7 @@ impl Core {
     /// root yet, which is not where the drain runs; the drain must not panic
     /// over it regardless, since that would stop the queue for good.
     pub(crate) fn root_uid(&self) -> Option<NodeUid> {
-        let st = self.state.lock();
+        let st = self.state();
         st.entries.get(&ROOT_INO).map(|e| e.uid.clone())
     }
 
diff --git a/crates/pdfs-fuse/src/filesystem.rs b/crates/pdfs-fuse/src/filesystem.rs
index 5fe28e5..990b657 100644
--- a/crates/pdfs-fuse/src/filesystem.rs
+++ b/crates/pdfs-fuse/src/filesystem.rs
@@ -94,13 +94,13 @@ impl Filesystem for ProtonFs {
     }
 
     fn forget(&self, _req: &Request, ino: INodeNo, nlookup: u64) {
-        let mut st = self.core.state.lock();
+        let mut st = self.core.state();
         st.forget_lookup(ino.0, nlookup);
     }
 
     fn getattr(&self, _req: &Request, ino: INodeNo, _fh: Option<FileHandle>, reply: ReplyAttr) {
         let (attr, provisional_in) = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             match st.entries.get(&ino.0) {
                 Some(e) => {
                     // A file still carrying the cheap enumeration's placeholder
@@ -146,7 +146,7 @@ impl Filesystem for ProtonFs {
             // Re-read: the upgrade writes through `state`, and on timeout or
             // failure this is simply the provisional attr we already had.
             let attr = {
-                let st = fs.core.state.lock();
+                let st = fs.core.state();
                 match st.entries.get(&ino) {
                     Some(e) => fs.attr(ino, &e.node, e.access),
                     // Forgotten while we waited.
@@ -218,7 +218,7 @@ impl Filesystem for ProtonFs {
         // has always had between a write and a read of the same handle; a mostly
         // local read is not worth reasoning about that for.
         let handle = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             st.active_writes.get(&ino.0).map(|h| {
                 (
                     h.file.clone(),
@@ -247,7 +247,7 @@ impl Filesystem for ProtonFs {
             return;
         }
         let (uid, mtime, fsize, is_video) = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             match st.entries.get(&ino.0) {
                 Some(e) if e.node.is_file() => {
                     let media_type = match &e.node.kind {
@@ -347,7 +347,7 @@ impl Filesystem for ProtonFs {
         // Stage the bytes straight into the scratch file (no base download): only
         // the untouched remainder is pulled from the remote, and only at commit.
         let file = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             match st.handles.get(&fh).and_then(|&i| st.active_writes.get(&i)) {
                 Some(aw) => aw.file.clone(),
                 None => {
@@ -371,7 +371,7 @@ impl Filesystem for ProtonFs {
             return;
         }
         let new_len = {
-            let mut st = self.core.state.lock();
+            let mut st = self.core.state();
             let Some(aw) = st
                 .handles
                 .get(&fh)
@@ -428,7 +428,7 @@ impl Filesystem for ProtonFs {
         if let Some(size) = size {
             let fh = fh.map(|f| f.0).filter(|&f| f != 0);
             let handled = {
-                let mut st = self.core.state.lock();
+                let mut st = self.core.state();
                 match st.active_writes.get_mut(&ino.0) {
                     Some(aw) => {
                         if size < aw.len {
@@ -471,7 +471,7 @@ impl Filesystem for ProtonFs {
                 });
                 return;
             }
-            self.core.state.lock().set_size(ino.0, size);
+            self.core.state().set_size(ino.0, size);
         }
         self.reply_attr(ino.0, reply);
     }
@@ -496,7 +496,7 @@ impl Filesystem for ProtonFs {
         let is_punch_hole = (mode & 0x02) != 0; // FALLOC_FL_PUNCH_HOLE
 
         let res = {
-            let mut st = self.core.state.lock();
+            let mut st = self.core.state();
             match st
                 .handles
                 .get(&_fh.0)
@@ -590,7 +590,7 @@ impl Filesystem for ProtonFs {
         reply: ReplyEmpty,
     ) {
         let (handle, unlinked_uid) = {
-            let mut st = self.core.state.lock();
+            let mut st = self.core.state();
             let unlinked_uid = release_unlinked_entry(&mut st, _ino.0);
             let ino = st.handles.remove(&fh.0);
             let h = ino.and_then(|i| {
@@ -736,7 +736,7 @@ impl Filesystem for ProtonFs {
     }
 
     fn access(&self, _req: &Request, ino: INodeNo, mask: AccessFlags, reply: ReplyEmpty) {
-        let state = self.core.state.lock();
+        let state = self.core.state();
         let Some(entry) = state.entries.get(&ino.0) else {
             reply.error(Errno::ENOENT);
             return;
@@ -921,7 +921,7 @@ impl ProtonFs {
     /// Build the filesystem rooted at `root` (the user's My Files folder).
     pub(super) fn new(core: Core, root: Node) -> Self {
         {
-            let mut st = core.state.lock();
+            let mut st = core.state();
             if let Err(e) = st.db.upsert_node(&root) {
                 warn!(uid = %root.uid, error = %e, "db upsert root failed");
             }
@@ -963,7 +963,7 @@ impl ProtonFs {
             return;
         }
         let hit = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             st.children.get(&parent).and_then(|kids| {
                 kids.iter().copied().find_map(|ino| {
                     st.entries
@@ -1013,7 +1013,7 @@ impl ProtonFs {
             // Re-read: on timeout or failure this is the provisional attr again,
             // which is no worse than what this path used to always send.
             let attr = {
-                let st = fs.core.state.lock();
+                let st = fs.core.state();
                 st.entries
                     .get(&ino)
                     .map_or(attr, |e| fs.attr(ino, &e.node, e.access))
@@ -1032,7 +1032,7 @@ impl ProtonFs {
             reply.error(e);
             return;
         }
-        let st = self.core.state.lock();
+        let st = self.core.state();
         let parent = st.entries.get(&ino).map_or(ROOT_INO, |e| e.parent);
 
         // "." and ".." occupy offsets 0 and 1; real children follow.
@@ -1099,7 +1099,7 @@ impl ProtonFs {
     fn serve_open(&self, ino: INodeNo, flags: OpenFlags, reply: ReplyOpen) {
         let write_requested = flags.acc_mode() != OpenAccMode::O_RDONLY;
         let (uid, base_mtime, base_size, base_revision_id) = {
-            let mut st = self.core.state.lock();
+            let mut st = self.core.state();
             match st.entries.get_mut(&ino.0) {
                 Some(e) if e.node.is_file() => {
                     if write_requested && !e.writable() {
@@ -1144,7 +1144,7 @@ impl ProtonFs {
         // safely by WriteHandle, so fail the open rather than risk corruption.
         let pending_base = self.core.pending.lock().get(&uid).cloned();
         if pending_base.as_ref().is_some_and(|p| !p.meta.complete) {
-            if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
+            if let Some(entry) = self.core.state().entries.get_mut(&ino.0) {
                 entry.open_count = entry.open_count.saturating_sub(1);
             }
             error!(%uid, "refusing write over incomplete queued revision");
@@ -1154,7 +1154,7 @@ impl ProtonFs {
         let (file, path) = match self.core.cache.create_scratch() {
             Ok(x) => x,
             Err(e) => {
-                if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
+                if let Some(entry) = self.core.state().entries.get_mut(&ino.0) {
                     entry.open_count = entry.open_count.saturating_sub(1);
                 }
                 error!(%uid, error = %e, "create scratch file failed");
@@ -1168,7 +1168,7 @@ impl ProtonFs {
             // a byte-for-byte copy of a multi-GB pending revision is the kind of
             // work `open` must not do (`ContentCache::clone_or_copy`).
             if let Err(e) = ContentCache::clone_or_copy(&pending.path, &path) {
-                if let Some(entry) = self.core.state.lock().entries.get_mut(&ino.0) {
+                if let Some(entry) = self.core.state().entries.get_mut(&ino.0) {
                     entry.open_count = entry.open_count.saturating_sub(1);
                 }
                 let _ = std::fs::remove_file(&path);
@@ -1179,7 +1179,7 @@ impl ProtonFs {
             }
             initial_written.add(0, pending.meta.len);
         }
-        let mut st = self.core.state.lock();
+        let mut st = self.core.state();
         let fh = st.next_fh;
         st.next_fh += 1;
         let aw = st.active_writes.entry(ino.0).or_insert_with(|| {
@@ -1209,7 +1209,7 @@ impl ProtonFs {
         // Everything the durability sidecar needs, copied out so no lock is held
         // across the sync.
         let handle = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             let ino = st.handles.get(&fh.0).copied();
             ino.and_then(|i| st.active_writes.get(&i)).map(|aw| {
                 (
@@ -1301,7 +1301,7 @@ impl ProtonFs {
             return;
         }
         let parent_uid = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             match st.entries.get(&parent) {
                 Some(e) => e.uid.clone(),
                 None => {
@@ -1383,7 +1383,7 @@ impl ProtonFs {
                 return;
             }
         };
-        let mut st = self.core.state.lock();
+        let mut st = self.core.state();
         let ino = st.intern(parent, node);
         if let Some(entry) = st.entries.get_mut(&ino) {
             entry.open_count = entry.open_count.saturating_add(1);
@@ -1438,7 +1438,7 @@ impl ProtonFs {
             return;
         }
         let parent_uid = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             match st.entries.get(&parent) {
                 Some(e) => e.uid.clone(),
                 None => {
@@ -1488,7 +1488,7 @@ impl ProtonFs {
                 }
             }
         };
-        let mut st = self.core.state.lock();
+        let mut st = self.core.state();
         let local = is_local_uid(&node.uid);
         let uid = node.uid.clone();
         let ino = st.intern(parent, node);
@@ -1520,7 +1520,7 @@ impl ProtonFs {
             return;
         }
         if let Ok((ino, _)) = self.core.lookup_child(parent, name_str) {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             if let Some(entry) = st.entries.get(&ino)
                 && entry.node.is_folder()
             {
@@ -1539,7 +1539,7 @@ impl ProtonFs {
         }
         if let Ok((ino, _)) = self.core.lookup_child(parent, name_str) {
             {
-                let st = self.core.state.lock();
+                let st = self.core.state();
                 if st
                     .entries
                     .get(&ino)
@@ -1556,7 +1556,7 @@ impl ProtonFs {
                 reply.error(e);
                 return;
             }
-            if self.core.state.lock().has_children(ino) {
+            if self.core.state().has_children(ino) {
                 reply.error(Errno::ENOTEMPTY);
                 return;
             }
@@ -1637,7 +1637,7 @@ impl ProtonFs {
         };
         // Ancestor cycle check: moving a directory into itself or one of its descendants is forbidden by POSIX (EINVAL).
         {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             if let Some(entry) = st.entries.get(&ino)
                 && entry.node.is_folder()
                 && st.is_ancestor_of(ino, newparent)
@@ -1656,7 +1656,7 @@ impl ProtonFs {
             return;
         }
         let (old_parent_uid, new_parent_uid) = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             match (st.entries.get(&parent), st.entries.get(&newparent)) {
                 (Some(old), Some(new)) => (old.uid.clone(), new.uid.clone()),
                 _ => {
@@ -1697,7 +1697,7 @@ impl ProtonFs {
             // check has to happen before anything is trashed: getting it wrong
             // turns a refusal into the destruction of the destination.
             let (src_dir, dst_dir) = {
-                let st = self.core.state.lock();
+                let st = self.core.state();
                 let is_dir = |i: &u64| {
                     st.entries
                         .get(i)
@@ -1760,12 +1760,9 @@ impl ProtonFs {
                 newname,
             ) {
                 Ok(true) => {
-                    self.core.state.lock().rename_in_place(
-                        ino,
-                        newparent,
-                        &new_parent_uid,
-                        newname,
-                    );
+                    self.core
+                        .state()
+                        .rename_in_place(ino, newparent, &new_parent_uid, newname);
                     // Finalize: a transient scratch file (its create parked, bytes
                     // held back) renamed to a finished name is the moment the
                     // completed file is meant to reach Drive. Un-park the create so
@@ -1906,7 +1903,7 @@ impl ProtonFs {
             if let Err(e) = moved {
                 error!(%uid, attempts, error = %e, "move after rename failed");
                 // The rename half landed in the source directory.
-                let mut state = self.core.state.lock();
+                let mut state = self.core.state();
                 let old_parent_uid = state
                     .entries
                     .get(&parent)
@@ -1932,7 +1929,7 @@ impl ProtonFs {
     /// Answer with an inode's current attributes, or `ENOENT` if it went away
     /// while the caller was doing something slower.
     fn reply_attr(&self, ino: u64, reply: ReplyAttr) {
-        let st = self.core.state.lock();
+        let st = self.core.state();
         match st.entries.get(&ino) {
             Some(e) => {
                 let attr = self.attr(ino, &e.node, e.access);
@@ -1969,7 +1966,7 @@ impl ProtonFs {
                 reply.error(error);
                 return;
             }
-            self.core.state.lock().forget_or_unlink(&uid);
+            self.core.state().forget_or_unlink(&uid);
             debug!(%uid, name, "deleted a node that had not been created remotely yet");
             reply.ok();
             return;
@@ -2003,7 +2000,7 @@ impl ProtonFs {
         // Only remove the namespace entry after Drive accepted the mutation.
         // Otherwise an EIO response still made the path disappear locally.
         self.core.hidden.lock().insert(uid.clone());
-        self.core.state.lock().forget_or_unlink(&uid);
+        self.core.state().forget_or_unlink(&uid);
         if !open_now {
             if let Err(error) = self.core.discard_queued_ops(&uid) {
                 warn!(%uid, ?error, "remote trash landed but queued-op cleanup failed");
@@ -2067,7 +2064,7 @@ impl ProtonFs {
     /// request for an unadvertised name, so nothing becomes unreachable.
     fn serve_listxattr(&self, ino: INodeNo, size: u32, reply: ReplyXattr) {
         let thumbnailable = {
-            let st = self.core.state.lock();
+            let st = self.core.state();
             match st.entries.get(&ino.0) {
                 Some(e) => match &e.node.kind {
                     NodeKind::File { media_type, .. } => {
diff --git a/crates/pdfs-fuse/src/lib.rs b/crates/pdfs-fuse/src/lib.rs
index 155036d..b4afe94 100644
--- a/crates/pdfs-fuse/src/lib.rs
+++ b/crates/pdfs-fuse/src/lib.rs
@@ -106,7 +106,7 @@ use mount::{
     SecondaryInsertRejection, SecondaryMount, clear_stale_mount, fuse_connection_id, spawn_session,
 };
 use reads::{BlockFlight, BlockRing, PREFETCH_BUDGET, Prefetch, ReaderSlot, STREAM_BYPASS_MIN};
-use state::{Entry, Intervals, PendingRevision, State, WriteHandle};
+use state::{Entry, Intervals, PendingRevision, State, StateGuard, WriteHandle, lock_state};
 use tracing::{debug, error, info, warn};
 use transfers::{CountingWriter, JobGuard, OwnedCountingReader, TransferRegistry};
 use r#virtual::{
@@ -322,6 +322,45 @@ struct VirtualRootPlan {
     visible: bool,
 }
 
+/// The sync folders a search hit can be opened through, as
+/// (Drive-relative root path, local directory) pairs.
+///
+/// Built once per search by [`Core::search_roots`]; resolving a hit against it
+/// is string arithmetic over the paths the index already carries, with no
+/// further queries.
+struct SearchRoots {
+    roots: Vec<(String, PathBuf)>,
+}
+
+impl SearchRoots {
+    /// The absolute local path for a hit at Drive-relative `path`, through the
+    /// most specific root covering it. A node can sit below several configured
+    /// roots, so the one leaving the shortest descendant path wins.
+    fn resolve(&self, path: &str) -> Option<String> {
+        self.roots
+            .iter()
+            .filter_map(|(root, local)| {
+                let relative = relative_to(root, path)?;
+                Some((relative.split('/').count(), local.join(relative)))
+            })
+            .min_by_key(|(depth, _)| *depth)
+            .map(|(_, path)| path.to_string_lossy().into_owned())
+    }
+}
+
+/// `path` expressed relative to `root`, or `None` if it is not below it. The
+/// root itself resolves to the empty path; an empty root is the mount itself
+/// and covers everything.
+fn relative_to<'a>(root: &str, path: &'a str) -> Option<&'a str> {
+    if root.is_empty() {
+        return Some(path);
+    }
+    if path == root {
+        return Some("");
+    }
+    path.strip_prefix(root)?.strip_prefix('/')
+}
+
 #[derive(Clone)]
 struct Core {
     client: ProtonDriveClient,
@@ -835,10 +874,20 @@ impl Core {
     /// `pending_op` row records a uid, not the session that queued it. Applying
     /// to all of them is correct because a uid is unique across mounts — at most
     /// one state has an entry for it, and the rest are no-ops.
+    /// Take this mount's inode lock.
+    ///
+    /// The guard is what applies the mutation's write-throughs, after releasing
+    /// the lock — see [`StateGuard`]. Every caller takes it this way; nothing
+    /// locks `self.state` directly, or the writes it queues would be applied
+    /// only when some later caller happened to take the guard.
+    fn state(&self) -> StateGuard<'_> {
+        StateGuard::new(self.state.lock(), &self.db)
+    }
+
     fn for_each_state(&self, mut apply: impl FnMut(&mut State)) {
         for (_, state, notifier, _) in self.states.live() {
             let changed = {
-                let mut state = state.lock();
+                let mut state = lock_state(&state, &self.db);
                 apply(&mut state);
                 state.take_access_changes()
             };
@@ -858,7 +907,7 @@ impl Core {
         for (_, state, notifier, _) in self.states.live() {
             let mut batch = NotifyBatch::default();
             {
-                let mut state = state.lock();
+                let mut state = lock_state(&state, &self.db);
                 apply(&mut state, &mut batch);
             }
             batch.flush(notifier.get());
@@ -869,12 +918,12 @@ impl Core {
     /// lock is released before notifying the kernel because notifier callbacks
     /// may synchronously provoke more filesystem work.
     fn flush_access_changes(&self) {
-        let changed = self.state.lock().take_access_changes();
+        let changed = self.state().take_access_changes();
         notify_access_changes(self.notifier.get(), &changed);
     }
 
     fn require_writable(&self, ino: u64) -> Result<(), Errno> {
-        self.state.lock().require_writable(ino)
+        self.state().require_writable(ino)
     }
 
     /// Mutation admission point. Persisted and every resident authority must
@@ -888,7 +937,7 @@ impl Core {
     fn require_uid_writable(&self, uid: &NodeUid) -> Result<(), Errno> {
         let mut live_access = Vec::new();
         {
-            let state = self.state.lock();
+            let state = self.state();
             if let Some(access) = state.access_by_uid(uid) {
                 live_access.push(access);
             }
@@ -1255,7 +1304,7 @@ impl Core {
         // Their node rows remain persisted as drain-time access authority, but
         // must not reappear in the mounted namespace after a restart.
         let hidden = self.hidden.lock().clone();
-        let mut st = self.state.lock();
+        let mut st = self.state();
 
         // Pass 1: assign a stable inode to every uid (root is already mapped).
         for sn in &stored {
@@ -1588,7 +1637,7 @@ impl Core {
             match self.rt.block_on(self.client.get_my_files_folder()) {
                 Ok(root) => {
                     {
-                        let mut st = self.state.lock();
+                        let mut st = self.state();
                         if let Some(e) = st.entries.get_mut(&ROOT_INO) {
                             e.node = root.clone();
                         }
@@ -1617,7 +1666,7 @@ impl Core {
     /// Lets a handler decide between answering inline and handing off to a
     /// worker, at the cost of one uncontended map lookup.
     fn children_cached(&self, ino: u64) -> bool {
-        self.state.lock().children.contains_key(&ino)
+        self.state().children.contains_key(&ino)
     }
 
     /// Re-apply the optimistic size of any queued write to `nodes`.
@@ -1693,7 +1742,7 @@ impl Core {
     ) -> Result<(), Errno> {
         let _publication = self.shared_publication.lock();
         let plan = self.prepare_virtual_root(&snapshot.real_names())?;
-        let mut st = self.state.lock();
+        let mut st = self.state();
         // State methods already establish the state -> DB lock order. No path
         // holds the DB connection while acquiring state, including shared event
         // publication, so keeping state locked across this one transaction
@@ -1705,7 +1754,7 @@ impl Core {
     }
 
     fn resident_root_listing_snapshot(&self, ino: u64) -> Result<RootListingSnapshot, Errno> {
-        RootListingSnapshot::capture(&self.state.lock(), ino).ok_or(Errno::EAGAIN)
+        RootListingSnapshot::capture(&self.state(), ino).ok_or(Errno::EAGAIN)
     }
 
     fn shared_folder_freshness_key(uid: &NodeUid) -> String {
@@ -1743,7 +1792,7 @@ impl Core {
 
     fn ensure_shared_children(&self, ino: u64) -> Result<(), Errno> {
         let online = self.online.load(Ordering::Relaxed);
-        let resident = self.state.lock().children.contains_key(&ino);
+        let resident = self.state().children.contains_key(&ino);
         match shared_listing_plan(
             resident,
             online,
@@ -1760,7 +1809,7 @@ impl Core {
                         error!(%error, "loading persisted shared roots failed");
                         Errno::EIO
                     })?;
-                let mut st = self.state.lock();
+                let mut st = self.state();
                 if st.children.contains_key(&ino) {
                     return Ok(());
                 }
@@ -1817,7 +1866,7 @@ impl Core {
             Ok(removed) => removed,
             Err(error) => {
                 error!(%error, "publishing shared-root listing failed");
-                let mut st = self.state.lock();
+                let mut st = self.state();
                 for uid in &accepted.uids {
                     st.downgrade_shared_subtree(uid);
                 }
@@ -1834,7 +1883,7 @@ impl Core {
             .iter()
             .map(|root| (root.node.uid.clone(), root.access))
             .collect();
-        let mut st = self.state.lock();
+        let mut st = self.state();
         for uid in &removed {
             st.hide_shared_root(uid);
         }
@@ -1863,7 +1912,7 @@ impl Core {
     /// held so concurrent metadata reads aren't blocked behind a fetch.
     fn ensure_children(&self, ino: u64) -> Result<(), Errno> {
         let (folder_uid, cached) = {
-            let st = self.state.lock();
+            let st = self.state();
             match st.entries.get(&ino) {
                 Some(e) => (e.uid.clone(), st.children.contains_key(&ino)),
                 None => return Err(Errno::ENOENT),
@@ -1906,7 +1955,7 @@ impl Core {
                 // sealed, which a queued write is ahead of (B11).
                 self.stamp_pending_sizes(&mut nodes);
                 let hidden = self.hidden.lock().clone();
-                let mut st = self.state.lock();
+                let mut st = self.state();
                 if st.children.contains_key(&ino) {
                     return Ok(());
                 }
@@ -2014,7 +2063,7 @@ impl Core {
                 })
                 .map(|node| node.uid.clone())
                 .collect();
-            let mut st = self.state.lock();
+            let mut st = self.state();
             for uid in &removed {
                 st.hide_foreign_subtree(uid);
             }
@@ -2034,7 +2083,7 @@ impl Core {
             return Ok(());
         }
 
-        let mut st = self.state.lock();
+        let mut st = self.state();
         // Lost the race? Another thread already populated it.
         if st.children.contains_key(&ino) {
             return Ok(());
@@ -2098,7 +2147,7 @@ impl Core {
             )
         };
         let (key, mut missing): (u64, Vec<NodeUid>) = {
-            let st = self.state.lock();
+            let st = self.state();
             match st.children.get(&parent) {
                 // The listing is resident: batch the whole folder under its
                 // inode, so the rest of an `ls -l` rides along on this fetch.
@@ -2249,7 +2298,7 @@ impl Core {
     /// A node that vanished counts as resolved; there is nothing left to wait
     /// for, and its caller will find the `ENOENT` for itself.
     fn size_is_real(&self, ino: u64) -> bool {
-        let st = self.state.lock();
+        let st = self.state();
         st.entries.get(&ino).is_none_or(|e| {
             !matches!(
                 &e.node.kind,
@@ -2281,7 +2330,7 @@ impl Core {
             // (B11).
             core.stamp_pending_sizes(&mut nodes);
             let mut changed: Vec<u64> = Vec::new();
-            let mut st = core.state.lock();
+            let mut st = core.state();
             for node in nodes {
                 // Only adopt the size. Re-interning wholesale would also adopt a
                 // name or parent that a rename/move may have changed locally
@@ -2332,7 +2381,7 @@ impl Core {
     /// the parent's listing is cached first.
     fn lookup_child(&self, parent: u64, name: &str) -> Result<(u64, NodeUid), Errno> {
         self.ensure_children(parent)?;
-        let st = self.state.lock();
+        let st = self.state();
         st.children
             .get(&parent)
             .and_then(|kids| {
@@ -2352,7 +2401,7 @@ impl Core {
     fn resolve_path(&self, rel: &Path) -> Result<(u64, NodeUid), Errno> {
         let mut ino = ROOT_INO;
         let mut uid = {
-            let st = self.state.lock();
+            let st = self.state();
             st.entries
                 .get(&ROOT_INO)
                 .map(|e| e.uid.clone())
@@ -2400,7 +2449,7 @@ impl Core {
     }
 
     fn source_parent_uid(&self, ino: u64, rel: &Path) -> CoreResult<NodeUid> {
-        let state = self.state.lock();
+        let state = self.state();
         let entry = state
             .entries
             .get(&ino)
@@ -2758,14 +2807,19 @@ impl Core {
         // Reflect the write in the tree straight away: `ls` must show the new
         // size and mtime even though the remote still holds the old revision.
         let now = now_secs();
-        let recorded = {
-            let mut st = self.state.lock();
-            // The only record that the file is as long as the caller was told.
-            // If it cannot be written the write did not fully take, and saying
-            // so beats acknowledging a `close(2)` and then serving the old size
-            // — the bytes themselves are already staged and queued either way.
+        let updated = {
+            let mut st = self.state();
             st.record_pending_write(ino, len, now)
         };
+        // The only record that the file is as long as the caller was told. If it
+        // cannot be written the write did not fully take, and saying so beats
+        // acknowledging a `close(2)` and then serving the old size — the bytes
+        // themselves are already staged and queued either way. Written out here
+        // rather than inside `State` so the commit does not hold the inode lock.
+        let recorded = match updated {
+            Some(node) => self.db.upsert_node(&node),
+            None => Ok(()),
+        };
         // Cached blobs and open readers describe the superseded revision. Reads
         // come from the staged blob until the op drains, so just drop them.
         // Done even when the row above failed: the staged blob is the file's
@@ -2803,7 +2857,7 @@ impl Core {
     ///   base, so it is the drain that has to fetch it.
     fn queue_truncate(&self, ino: u64, size: u64) -> Result<(), Errno> {
         let (uid, base_mtime, base_size, base_revision_id) = {
-            let st = self.state.lock();
+            let st = self.state();
             match st.entries.get(&ino) {
                 Some(e) if e.node.is_file() => (
                     e.uid.clone(),
@@ -3108,7 +3162,7 @@ impl Core {
                     "cannot queue write; bytes kept in staging"
                 );
                 let name = {
-                    let st = self.state.lock();
+                    let st = self.state();
                     st.entries
                         .get(&ino)
                         .map(|e| e.node.name.clone())
@@ -3136,7 +3190,7 @@ impl Core {
                     "write rescue was only partially published; bytes retained"
                 );
                 let name = {
-                    let st = self.state.lock();
+                    let st = self.state();
                     st.entries
                         .get(&ino)
                         .map(|entry| entry.node.name.clone())
@@ -3223,7 +3277,7 @@ impl Core {
     fn pin(&self, rel: &Path) -> CoreResult<String> {
         let (ino, uid) = self.resolve(rel)?;
         let (name, is_folder, mtime, size) = {
-            let st = self.state.lock();
+            let st = self.state();
             let e = st
                 .entries
                 .get(&ino)
@@ -3266,7 +3320,7 @@ impl Core {
         while let Some(dir) = stack.pop() {
             self.ensure_children(dir)
                 .map_err(|e| self.errno_error(e, "enumerate"))?;
-            let st = self.state.lock();
+            let st = self.state();
             if let Some(kids) = st.children.get(&dir) {
                 for &k in kids {
                     if let Some(e) = st.entries.get(&k) {
@@ -3307,7 +3361,7 @@ impl Core {
     /// `Ok(None)` when the node is not a file or has no thumbnail of that type.
     fn thumbnail(&self, ino: u64, ttype: ThumbnailType) -> Result<Option<Vec<u8>>, Errno> {
         let (uid, mtime) = {
-            let st = self.state.lock();
+            let st = self.state();
             match st.entries.get(&ino) {
                 Some(e) if e.node.is_file() => (e.uid.clone(), e.node.modification_time),
                 Some(_) => return Ok(None),
@@ -3356,7 +3410,7 @@ impl Core {
     fn unpin(&self, rel: &Path) -> CoreResult<String> {
         let (ino, uid) = self.resolve(rel)?;
         let (name, is_folder) = {
-            let st = self.state.lock();
+            let st = self.state();
             st.entries
                 .get(&ino)
                 .map(|e| (e.node.name.clone(), e.node.is_folder()))
@@ -3397,7 +3451,7 @@ impl Core {
         // Snapshot the listing, then drop the lock before touching the on-disk
         // pin registry so a slow disk read doesn't block FUSE metadata ops.
         let rows: Vec<(String, bool, u64, i64, NodeUid)> = {
-            let st = self.state.lock();
+            let st = self.state();
             st.children
                 .get(&ino)
                 .map(|kids| {
@@ -3436,31 +3490,47 @@ impl Core {
             .collect())
     }
 
-    /// Return an absolute path through the most specific local location
-    /// covering `uid`. A node may be below several configured roots, so prefer
-    /// the root leaving the shortest descendant path.
+    /// Every uid that is pinned right now, as one set.
+    ///
+    /// `cache.is_pinned` is a direct lookup plus an ancestor walk; asking it per
+    /// candidate was a thousand walks per keystroke. The recursive expansion is
+    /// the same query, run once.
+    fn pinned_set(&self) -> HashSet<String> {
+        self.db
+            .pinned_uids()
+            .unwrap_or_default()
+            .into_iter()
+            .collect()
+    }
+
+    /// The local locations a search result can be opened through, resolved once
+    /// per search rather than once per hit.
     ///
     /// An on-demand folder only answers while its mount is live — without the
     /// mount its `local_path` is an ordinary empty directory. A mirror folder
     /// needs no mount: its files are real local copies the sync loop pushes
     /// back, so it is a valid open target whenever it is configured.
-    fn mounted_search_path(&self, uid: &str) -> Option<String> {
+    ///
+    /// This used to be a per-hit call that re-read the sync-folder list and ran
+    /// an ancestor walk per folder, against a candidate pool of up to a
+    /// thousand: the whole cost of a keystroke was in here.
+    fn search_roots(&self) -> SearchRoots {
         let live_mounts = self.mounts.lock();
-        let folders = self.db.sync_folder_list().ok()?;
-        folders
-            .into_iter()
-            .filter(|folder| match folder.mode.as_str() {
-                "ondemand" => live_mounts.contains_key(&folder.id),
-                "mirror" => true,
-                _ => false,
-            })
-            .filter_map(|folder| {
-                let relative = self.db.path_relative_to(&folder.remote_uid, uid).ok()??;
-                let depth = relative.split('/').count();
-                Some((depth, PathBuf::from(folder.local_path).join(relative)))
-            })
-            .min_by_key(|(depth, _)| *depth)
-            .map(|(_, path)| path.to_string_lossy().into_owned())
+        let folders = self.db.sync_folder_list().unwrap_or_default();
+        SearchRoots {
+            roots: folders
+                .into_iter()
+                .filter(|folder| match folder.mode.as_str() {
+                    "ondemand" => live_mounts.contains_key(&folder.id),
+                    "mirror" => true,
+                    _ => false,
+                })
+                .filter_map(|folder| {
+                    let root = self.db.node_path(&folder.remote_uid).ok().flatten()?;
+                    Some((root, PathBuf::from(folder.local_path)))
+                })
+                .collect(),
+        }
     }
 
     /// Full-text search node names against the local SQLite index, mapping each
@@ -3471,17 +3541,19 @@ impl Core {
             .db
             .search(query, limit)
             .map_err(|e| CoreError::from_api(&e, "search"))?;
+        let roots = self.search_roots();
+        let pinned = self.pinned_set();
         Ok(hits
             .into_iter()
             .map(|h| SearchHit {
                 name: h.node.name.clone(),
-                path: h.path,
                 is_dir: h.node.is_folder(),
                 size: node_size(&h.node),
                 modified: h.node.modification_time,
-                pinned: self.cache.is_pinned(&h.node.uid),
+                pinned: pinned.contains(&h.node.uid.to_string()),
                 uid: h.node.uid.to_string(),
-                mounted_path: self.mounted_search_path(&h.node.uid.to_string()),
+                mounted_path: roots.resolve(&h.path),
+                path: h.path,
                 score: 0,
             })
             .collect())
@@ -3525,6 +3597,8 @@ impl Core {
         // SQLite boundary to `limit` would recreate the old empty-filter bug.
         let candidate_limit = limit.saturating_mul(10).clamp(100, 1_000);
         let mut drive_hits = if filters.sources.contains(&SearchSource::Drive) {
+            let roots = self.search_roots();
+            let pinned = self.pinned_set();
             self.db
                 .search_candidates(query, candidate_limit)
                 .map_err(|e| CoreError::from_api(&e, "search candidates"))?
@@ -3534,18 +3608,19 @@ impl Core {
                     let parent = Path::new(&hit.path)
                         .parent()
                         .map_or_else(String::new, |path| path.display().to_string());
-                    let pinned = self.cache.is_pinned(&hit.node.uid);
+                    let uid = hit.node.uid.to_string();
+                    let is_pinned = pinned.contains(&uid);
                     let score = relevance_score(query, &hit.node.name, &parent)?
-                        + if pinned { 250 } else { 0 };
+                        + if is_pinned { 250 } else { 0 };
                     Some(SearchHit {
                         name: hit.node.name.clone(),
-                        path: hit.path,
                         is_dir: hit.node.is_folder(),
                         size: node_size(&hit.node),
                         modified: hit.node.modification_time,
-                        pinned,
-                        uid: hit.node.uid.to_string(),
-                        mounted_path: self.mounted_search_path(&hit.node.uid.to_string()),
+                        pinned: is_pinned,
+                        uid,
+                        mounted_path: roots.resolve(&hit.path),
+                        path: hit.path,
                         score,
                     })
                 })
@@ -3621,7 +3696,7 @@ impl Core {
     fn open_file(&self, rel: &Path) -> CoreResult<PathBuf> {
         let (ino, uid) = self.resolve(rel)?;
         let (name, mtime, size) = {
-            let st = self.state.lock();
+            let st = self.state();
             let e = st
                 .entries
                 .get(&ino)
@@ -3746,7 +3821,7 @@ impl Core {
     fn invalidate_parent_listing(&self, rel: &Path) {
         let parent = rel.parent().unwrap_or_else(|| Path::new(""));
         if let Ok((pino, _)) = self.resolve_path(parent) {
-            self.state.lock().invalidate_listing(pino);
+            self.state().invalidate_listing(pino);
         }
     }
 
@@ -3805,7 +3880,7 @@ impl Core {
             .map(|(_, n)| n)
             .unwrap_or_default();
         self.invalidate_parent_listing(rel);
-        self.state.lock().invalidate_listing(pino);
+        self.state().invalidate_listing(pino);
         Ok(name)
     }
 
@@ -4096,7 +4171,7 @@ impl Core {
     /// [`CtlRequest::Refresh`] with a [`RefreshScope::Dir`] scope.
     fn refresh_dir(&self, rel: &Path) -> CoreResult<()> {
         let (ino, _uid) = self.resolve(rel)?;
-        self.state.lock().invalidate_listing(ino);
+        self.state().invalidate_listing(ino);
         Ok(())
     }
 
@@ -4262,7 +4337,7 @@ impl Core {
         let node = self
             .fetch_node(&new_uid)
             .map_err(|e| self.errno_error(e, "fetch node"))?;
-        let mut st = self.state.lock();
+        let mut st = self.state();
         let ino = st.intern(pino, node);
         if let Some(kids) = st.children.get_mut(&pino)
             && !kids.contains(&ino)
@@ -5935,7 +6010,7 @@ mod tests {
         let mut persisted_virtual = virtual_node(root_uid.clone(), "Shared with me".into(), 0);
         persisted_virtual.trashed = true;
         state
-            .db
+            .flushed_db()
             .publish_virtual_root(super::SHARED_WITH_ME_NAME, &persisted_virtual)
             .unwrap();
         let mut indexed_descendant = node_helper_in_volume(
@@ -5946,7 +6021,7 @@ mod tests {
             false,
         );
         indexed_descendant.parent_uid = Some(shared_with_me_uid());
-        state.db.upsert_node(&indexed_descendant).unwrap();
+        state.flushed_db().upsert_node(&indexed_descendant).unwrap();
         assert!(
             state
                 .db
@@ -5972,7 +6047,7 @@ mod tests {
             node: stale_node.clone(),
             visible: true,
         };
-        let db = state.db.clone();
+        let db = state.flushed_db();
         let error = publish_virtual_root_in_listing(
             &db,
             &mut state,
@@ -6048,14 +6123,17 @@ mod tests {
             false,
         );
         descendant.parent_uid = Some(shared_with_me_uid());
-        state.db.upsert_node(&descendant).unwrap();
+        state.flushed_db().upsert_node(&descendant).unwrap();
         // A node whose parent row is not cached — the synthetic root is not one
         // until this publication commits — is indexed under the path it can
         // actually be resolved to, which is its bare name. Publication reindexes
         // the subtree beneath the synthetic ancestor, so that path is what the
         // rollback has to restore.
         let path_before = {
-            let hits = state.db.search("AtomicRollbackFindable", 10).unwrap();
+            let hits = state
+                .flushed_db()
+                .search("AtomicRollbackFindable", 10)
+                .unwrap();
             assert_eq!(hits.len(), 1, "the descendant starts out searchable");
             hits[0].path.clone()
         };
@@ -6064,7 +6142,7 @@ mod tests {
         // The pin is the transaction's final statement. Failing it proves the
         // earlier access, node, and descendant FTS work rolls back with it.
         state
-            .db
+            .flushed_db()
             .with_conn(|conn| {
                 conn.execute_batch(
                     "CREATE TRIGGER reject_virtual_root_pin
@@ -6082,7 +6160,7 @@ mod tests {
         let by_uid_before = state.by_uid.len();
         let next_ino_before = state.next_ino;
         let access_before = state.share_access.clone();
-        let db = state.db.clone();
+        let db = state.flushed_db();
         let result = publish_virtual_root_in_listing(
             &db,
             &mut state,
@@ -6110,7 +6188,10 @@ mod tests {
         );
 
         assert_eq!(
-            state.db.state_str(super::SHARED_WITH_ME_NAME).unwrap(),
+            state
+                .flushed_db()
+                .state_str(super::SHARED_WITH_ME_NAME)
+                .unwrap(),
             None
         );
         assert!(
@@ -6120,8 +6201,17 @@ mod tests {
                 .unwrap()
                 .is_none()
         );
-        assert_eq!(state.db.share_access(&shared_with_me_uid()).unwrap(), None);
-        let hits = state.db.search("AtomicRollbackFindable", 10).unwrap();
+        assert_eq!(
+            state
+                .flushed_db()
+                .share_access(&shared_with_me_uid())
+                .unwrap(),
+            None
+        );
+        let hits = state
+            .flushed_db()
+            .search("AtomicRollbackFindable", 10)
+            .unwrap();
         assert_eq!(hits.len(), 1);
         assert_eq!(
             hits[0].path, path_before,
@@ -6589,7 +6679,7 @@ mod tests {
         let shared = state.intern(root, node_helper("shared", "root", "shared", true));
         state.entries.get_mut(&shared).unwrap().access = Access::Viewer;
         let uid = state.entries[&shared].uid.clone();
-        let db = state.db.clone();
+        let db = state.flushed_db();
         let before = db.pending_ops().unwrap().len();
 
         let preserved = Cell::new(false);
@@ -6629,7 +6719,7 @@ mod tests {
         let root = state.intern(0, node_helper("root", "none", "root", true));
         let node = state.intern(root, node_helper("shared", "root", "shared", true));
         let uid = state.entries[&node].uid.clone();
-        let db = state.db.clone();
+        let db = state.flushed_db();
 
         db.set_share_access(&uid, Access::Viewer).unwrap();
         assert_eq!(
@@ -6713,7 +6803,7 @@ mod tests {
         let mut virtual_node = node_helper("sharedwithme", "root", "Shared with me", true);
         virtual_node.uid = shared_with_me_uid();
         state
-            .db
+            .flushed_db()
             .set_share_access(&virtual_node.uid, Access::Viewer)
             .unwrap();
         state
@@ -6957,18 +7047,7 @@ mod tests {
         std::fs::create_dir_all(&dir_path).unwrap();
         let db = pdfs_core::db::Db::open(&dir_path.join("cache.db")).unwrap();
         let share_access = db.all_share_access().unwrap();
-        let st = crate::state::State {
-            entries: std::collections::HashMap::new(),
-            by_uid: std::collections::HashMap::new(),
-            children: std::collections::HashMap::new(),
-            next_ino: 1,
-            active_writes: std::collections::HashMap::new(),
-            handles: std::collections::HashMap::new(),
-            next_fh: 1,
-            access_changes: std::collections::HashSet::new(),
-            share_access,
-            db: std::sync::Arc::new(db),
-        };
+        let st = crate::state::State::new(std::sync::Arc::new(db), share_access, 1);
         (st, TestDir(dir_path))
     }
 
@@ -7008,7 +7087,7 @@ mod tests {
         // The typed DB operation atomically replaces the old revision with the
         // one intent that must survive the open handle's final release.
         state
-            .db
+            .flushed_db()
             .enqueue_op(&PendingOp {
                 id: 0,
                 kind: OP_REVISION.to_string(),
@@ -7024,45 +7103,64 @@ mod tests {
             })
             .unwrap();
         let (_, blobs) = state
-            .db
+            .flushed_db()
             .replace_ops_with_trash(&uid.to_string(), "dirty.txt", 2)
             .unwrap();
         assert!(blobs.is_empty());
-        let queued_counts = state.db.pending_op_counts().unwrap();
+        let queued_counts = state.flushed_db().pending_op_counts().unwrap();
 
         state.unlink_mem(&uid);
         assert!(state.entries[&ino].unlinked);
         assert!(!state.children[&parent].contains(&ino));
-        assert!(state.db.node_by_uid(&uid.to_string()).unwrap().is_some());
+        assert!(
+            state
+                .flushed_db()
+                .node_by_uid(&uid.to_string())
+                .unwrap()
+                .is_some()
+        );
 
         let released = release_unlinked_entry(&mut state, ino).unwrap();
         assert_eq!(released, uid);
         assert!(!state.by_uid.contains_key(&uid));
         assert!(release_must_retain_queued_trash(&state.db, &uid).unwrap());
         assert!(
-            state.db.node_by_uid(&uid.to_string()).unwrap().is_some(),
+            state
+                .flushed_db()
+                .node_by_uid(&uid.to_string())
+                .unwrap()
+                .is_some(),
             "last release keeps persisted drain authority"
         );
-        let pending = state.db.pending_ops().unwrap();
+        let pending = state.flushed_db().pending_ops().unwrap();
         assert_eq!(pending.len(), 1, "release must not add a revision");
         assert_eq!(pending[0].kind, OP_TRASH);
-        let after_release = state.db.pending_op_counts().unwrap();
+        let after_release = state.flushed_db().pending_op_counts().unwrap();
         assert_eq!(after_release.uploads, 0);
         assert_eq!(after_release.changes, 1);
         assert_eq!(after_release.uploads, queued_counts.uploads);
         assert_eq!(after_release.changes, queued_counts.changes);
 
-        state.db.complete_trash_op(pending[0].id, &uid).unwrap();
-        assert!(state.db.pending_ops().unwrap().is_empty());
-        assert!(state.db.node_by_uid(&uid.to_string()).unwrap().is_none());
+        state
+            .flushed_db()
+            .complete_trash_op(pending[0].id, &uid)
+            .unwrap();
+        assert!(state.flushed_db().pending_ops().unwrap().is_empty());
+        assert!(
+            state
+                .flushed_db()
+                .node_by_uid(&uid.to_string())
+                .unwrap()
+                .is_none()
+        );
     }
 
     #[test]
     fn queued_trash_lookup_failure_conservatively_retains_unlinked_state() {
-        let (state, _dir) = state_test_helper();
+        let (mut state, _dir) = state_test_helper();
         let uid = NodeUid::new(VolumeId::from("vol"), LinkId::from("uncertain-trash"));
         state
-            .db
+            .flushed_db()
             .with_conn(|conn| {
                 conn.execute_batch("DROP TABLE pending_op")?;
                 Ok(())
@@ -7113,3 +7211,72 @@ mod tests {
         }
     }
 }
+
+#[cfg(test)]
+mod search_root_tests {
+    use super::*;
+
+    fn roots(pairs: &[(&str, &str)]) -> SearchRoots {
+        SearchRoots {
+            roots: pairs
+                .iter()
+                .map(|(root, local)| ((*root).to_string(), PathBuf::from(local)))
+                .collect(),
+        }
+    }
+
+    #[test]
+    fn a_hit_below_a_root_resolves_through_it() {
+        let roots = roots(&[("Videos/Anime", "/home/u/anime")]);
+        assert_eq!(
+            roots.resolve("Videos/Anime/s01/e01.mkv").as_deref(),
+            Some("/home/u/anime/s01/e01.mkv")
+        );
+    }
+
+    #[test]
+    fn a_hit_outside_every_root_resolves_nowhere() {
+        let roots = roots(&[("Videos/Anime", "/home/u/anime")]);
+        assert_eq!(roots.resolve("Documents/tax.pdf"), None);
+    }
+
+    #[test]
+    fn a_sibling_sharing_a_name_prefix_is_not_below_the_root() {
+        // `Videos/Anime2` starts with `Videos/Anime` as a string and is a
+        // different folder; only a `/` boundary makes it a descendant.
+        let roots = roots(&[("Videos/Anime", "/home/u/anime")]);
+        assert_eq!(roots.resolve("Videos/Anime2/e01.mkv"), None);
+    }
+
+    #[test]
+    fn the_most_specific_root_wins() {
+        let roots = roots(&[
+            ("Videos", "/home/u/videos"),
+            ("Videos/Anime", "/home/u/anime"),
+        ]);
+        assert_eq!(
+            roots.resolve("Videos/Anime/e01.mkv").as_deref(),
+            Some("/home/u/anime/e01.mkv"),
+            "the root leaving the shortest descendant path is the one the user \
+             configured for this content"
+        );
+    }
+
+    #[test]
+    fn a_root_covering_the_whole_drive_resolves_everything() {
+        let roots = roots(&[("", "/home/u/drive")]);
+        assert_eq!(
+            roots.resolve("Documents/tax.pdf").as_deref(),
+            Some("/home/u/drive/Documents/tax.pdf")
+        );
+    }
+
+    #[test]
+    fn the_root_folder_itself_resolves_to_its_local_directory() {
+        let roots = roots(&[("Videos/Anime", "/home/u/anime")]);
+        assert_eq!(
+            roots.resolve("Videos/Anime").as_deref(),
+            Some("/home/u/anime/")
+        );
+    }
+}
diff --git a/crates/pdfs-fuse/src/mount.rs b/crates/pdfs-fuse/src/mount.rs
index 302941c..9b942c5 100644
--- a/crates/pdfs-fuse/src/mount.rs
+++ b/crates/pdfs-fuse/src/mount.rs
@@ -360,18 +360,7 @@ pub fn mount(
         maintenance: Arc::new(Mutex::new(Default::default())),
         primary_root_uid: root.uid.clone(),
         primary: true,
-        state: Arc::new(Mutex::new(State {
-            entries: HashMap::new(),
-            by_uid: HashMap::new(),
-            children: HashMap::new(),
-            next_ino: 2,
-            active_writes: HashMap::new(),
-            handles: HashMap::new(),
-            next_fh: 1,
-            access_changes: HashSet::new(),
-            share_access,
-            db: db.clone(),
-        })),
+        state: Arc::new(Mutex::new(State::new(db.clone(), share_access, 2))),
         cache: Arc::new(cache),
         readers: Arc::new(Mutex::new(HashMap::new())),
         block_ring: Arc::new(Mutex::new(BlockRing::default())),
diff --git a/crates/pdfs-fuse/src/state.rs b/crates/pdfs-fuse/src/state.rs
index f2fe96d..2602ccb 100644
--- a/crates/pdfs-fuse/src/state.rs
+++ b/crates/pdfs-fuse/src/state.rs
@@ -166,13 +166,163 @@ pub(crate) struct State {
     /// on every explicit role/tombstone change. Descendant inheritance reads
     /// this map instead of issuing one SQLite query per interned inode.
     pub(crate) share_access: HashMap<NodeUid, Access>,
-    /// Unified SQLite metadata cache. Every map mutation below writes through to
-    /// it inside the `State` lock so the DB stays the authoritative copy across
-    /// restarts (see plan.md P1).
+    /// Unified SQLite metadata cache. Read from here (`has_children`) but never
+    /// written: mutations go to [`State::outbox`] and are applied after the lock
+    /// is released. See [`StateGuard`].
     pub(crate) db: Arc<Db>,
+    /// Write-throughs this mutation owes SQLite, applied by [`StateGuard`] once
+    /// the inode lock is released.
+    ///
+    /// The DB stays the authoritative copy across restarts, so every map
+    /// mutation still writes through — it just no longer does so with the
+    /// whole-mount lock held. A single listing's write-through is one
+    /// transaction over up to a few thousand rows, and every other FUSE callback
+    /// used to wait behind it for no reason: the maps were already updated
+    /// before the first row was written.
+    outbox: Vec<DbWrite>,
+}
+
+/// A write-through queued by a `State` mutation, applied after the lock drops.
+///
+/// Deliberately a small closed set rather than a boxed closure: these run
+/// outside the lock, and what they can touch should be readable here.
+pub(crate) enum DbWrite {
+    /// One node or a whole listing. Both go through `upsert_nodes`, which is a
+    /// single transaction either way, so there is nothing to gain by splitting
+    /// them — and a `Node` is large enough that a variant holding one inline
+    /// would set the size of every other.
+    Upsert(Vec<Node>),
+    Delete(NodeUid),
+    SetListed(NodeUid, bool),
+    SetShareAccess(NodeUid, Access),
+}
+
+impl DbWrite {
+    fn apply(self, db: &Db) {
+        let outcome = match &self {
+            Self::Upsert(nodes) => db.upsert_nodes(nodes),
+            Self::Delete(uid) => db.delete_node(uid),
+            Self::SetListed(uid, listed) => db.set_listed(uid, *listed),
+            Self::SetShareAccess(uid, access) => db.set_share_access(uid, *access),
+        };
+        if let Err(error) = outcome {
+            let what = match &self {
+                Self::Upsert(_) => "upsert_nodes",
+                Self::Delete(_) => "delete_node",
+                Self::SetListed(..) => "set_listed",
+                Self::SetShareAccess(..) => "set_share_access",
+            };
+            warn!(%error, write = what, "db write-through failed");
+        }
+    }
+}
+
+/// The inode lock, held for the duration of a mutation, plus the write-throughs
+/// that mutation owed the database.
+///
+/// Dropping it releases the lock *first* and only then writes, which is the
+/// whole point: interning a five-thousand-child listing commits one transaction
+/// — several fsyncs' worth — and nothing about that needs the maps frozen. The
+/// write still happens synchronously on this thread before the caller proceeds,
+/// so a read that follows a mutation still sees it.
+pub(crate) struct StateGuard<'a> {
+    guard: Option<parking_lot::MutexGuard<'a, State>>,
+    db: &'a Arc<Db>,
+}
+
+/// Take a mount's inode lock through a [`StateGuard`], for the loops that walk
+/// every live mount rather than this `Core`'s own.
+pub(crate) fn lock_state<'a>(
+    state: &'a parking_lot::Mutex<State>,
+    db: &'a Arc<Db>,
+) -> StateGuard<'a> {
+    StateGuard::new(state.lock(), db)
+}
+
+impl<'a> StateGuard<'a> {
+    pub(crate) fn new(guard: parking_lot::MutexGuard<'a, State>, db: &'a Arc<Db>) -> Self {
+        Self {
+            guard: Some(guard),
+            db,
+        }
+    }
+}
+
+impl std::ops::Deref for StateGuard<'_> {
+    type Target = State;
+
+    fn deref(&self) -> &State {
+        self.guard.as_ref().expect("held until drop")
+    }
+}
+
+impl std::ops::DerefMut for StateGuard<'_> {
+    fn deref_mut(&mut self) -> &mut State {
+        self.guard.as_mut().expect("held until drop")
+    }
+}
+
+impl Drop for StateGuard<'_> {
+    fn drop(&mut self) {
+        let Some(mut guard) = self.guard.take() else {
+            return;
+        };
+        let writes = std::mem::take(&mut guard.outbox);
+        // The lock goes before the writes do. That ordering is the feature.
+        drop(guard);
+        for write in writes {
+            write.apply(self.db);
+        }
+    }
 }
 
 impl State {
+    /// A fresh, empty inode space over `db`.
+    ///
+    /// `next_ino` differs by caller: a real mount hands out [`crate::ROOT_INO`]
+    /// to its root and starts allocating above it, while a bare state under test
+    /// has no root and starts at 1.
+    /// Apply the write-throughs this state owes SQLite, here and now.
+    ///
+    /// Production code never calls this: [`StateGuard`] releases the inode lock
+    /// first and then applies them, which is the whole point. A test driving a
+    /// bare `State` has no guard, so it calls this where the daemon would have
+    /// dropped one.
+    #[cfg(test)]
+    pub(crate) fn flush_outbox(&mut self) {
+        let db = self.db.clone();
+        for write in std::mem::take(&mut self.outbox) {
+            write.apply(&db);
+        }
+    }
+
+    /// The database, with everything this state owes it already written.
+    ///
+    /// What a test means by `st.db` is almost always "the database as of now",
+    /// which in the daemon is the state after the guard dropped. Handing back an
+    /// owned `Arc` rather than a borrow keeps it usable inside an `assert_eq!`
+    /// that also touches `st`.
+    #[cfg(test)]
+    pub(crate) fn flushed_db(&mut self) -> Arc<Db> {
+        self.flush_outbox();
+        self.db.clone()
+    }
+
+    pub(crate) fn new(db: Arc<Db>, share_access: HashMap<NodeUid, Access>, next_ino: u64) -> Self {
+        Self {
+            entries: HashMap::new(),
+            by_uid: HashMap::new(),
+            children: HashMap::new(),
+            next_ino,
+            active_writes: HashMap::new(),
+            handles: HashMap::new(),
+            next_fh: 1,
+            access_changes: HashSet::new(),
+            share_access,
+            db,
+            outbox: Vec::new(),
+        }
+    }
     /// Whether `uid` names an owned node currently reachable from this mount's
     /// root. Resident entries can outlive their dentries for open-handle and
     /// access-revocation semantics, so `by_uid` membership alone is not proof
@@ -278,11 +428,9 @@ impl State {
                 || stored.unwrap_or(Access::Viewer),
                 |membership| access_for(membership.role_exact(), parent_access, true),
             );
-            if let Err(error) = self.db.set_share_access(&node.uid, access) {
-                warn!(uid = %node.uid, ?access, %error, "db set_share_access failed");
-            } else {
-                self.share_access.insert(node.uid.clone(), access);
-            }
+            self.outbox
+                .push(DbWrite::SetShareAccess(node.uid.clone(), access));
+            self.share_access.insert(node.uid.clone(), access);
             return access;
         }
         parent_access
@@ -313,11 +461,9 @@ impl State {
             None => (None, Access::Unknown),
         };
         let access = access_for(role, parent_access, true);
-        if let Err(error) = self.db.set_share_access(uid, access) {
-            warn!(%uid, ?access, %error, "db set observed share access failed");
-        } else {
-            self.share_access.insert(uid.clone(), access);
-        }
+        self.outbox
+            .push(DbWrite::SetShareAccess(uid.clone(), access));
+        self.share_access.insert(uid.clone(), access);
         let Some(root) = root else {
             return Vec::new();
         };
@@ -602,18 +748,14 @@ impl State {
             })
             .collect();
         for (uid, access) in discovered {
-            if let Err(error) = self.db.set_share_access(&uid, access) {
-                warn!(%uid, ?access, %error, "db set hydrated share access failed");
-            } else {
-                self.share_access.insert(uid, access);
-            }
+            self.outbox
+                .push(DbWrite::SetShareAccess(uid.clone(), access));
+            self.share_access.insert(uid, access);
         }
     }
 
     pub(crate) fn intern(&mut self, parent: u64, node: Node) -> u64 {
-        if let Err(e) = self.db.upsert_node(&node) {
-            warn!(uid = %node.uid, error = %e, "db upsert_node failed");
-        }
+        self.outbox.push(DbWrite::Upsert(vec![node.clone()]));
         self.intern_mem(parent, node)
     }
 
@@ -685,9 +827,7 @@ impl State {
     /// This is what keeps `ls` on a large folder quick: one commit for the
     /// listing, rather than one autocommit — and one fsync — per child.
     pub(crate) fn intern_batch(&mut self, parent: u64, nodes: Vec<Node>) -> Vec<u64> {
-        if let Err(e) = self.db.upsert_nodes(&nodes) {
-            warn!(error = %e, "db upsert_nodes failed");
-        }
+        self.outbox.push(DbWrite::Upsert(nodes.clone()));
         nodes
             .into_iter()
             .map(|node| self.intern_mem(parent, node))
@@ -769,9 +909,7 @@ impl State {
     /// `(parent_ino, name)` when the node was known, so the caller can notify
     /// the kernel.
     pub(crate) fn forget(&mut self, uid: &NodeUid) -> Option<(u64, String)> {
-        if let Err(e) = self.db.delete_node(uid) {
-            warn!(%uid, error = %e, "db delete_node failed");
-        }
+        self.outbox.push(DbWrite::Delete(uid.clone()));
         self.forget_mem(uid)
     }
 
@@ -834,9 +972,7 @@ impl State {
                 kids.push(ino);
             }
         }
-        if let Err(e) = self.db.upsert_node(&node) {
-            warn!(uid = %node.uid, error = %e, "db upsert_node failed for a queued rename");
-        }
+        self.outbox.push(DbWrite::Upsert(vec![node]));
         self.recompute_descendant_access(ino);
     }
 
@@ -851,9 +987,7 @@ impl State {
         self.children.remove(&ino);
         if let Some(e) = self.entries.get(&ino) {
             let uid = e.uid.clone();
-            if let Err(err) = self.db.set_listed(&uid, false) {
-                warn!(%uid, error = %err, "db set_listed(false) failed");
-            }
+            self.outbox.push(DbWrite::SetListed(uid, false));
         }
     }
 
@@ -920,24 +1054,15 @@ impl State {
     /// file is as long as the caller was told it is. Without it a restart serves
     /// the stale size and the file reads as truncated — or empty — while its
     /// bytes sit safely in staging (offline.md Phase 3).
-    /// Failure is reported rather than logged: dropping the row silently means
+    /// Returns the node to persist rather than queueing it in the outbox, and is
+    /// the one mutation here whose write-through the caller must do itself:
+    /// failure has to be reported, not logged. Dropping the row silently means
     /// the daemon acknowledged a write and then serves the old size for it, so
     /// the caller has to be able to turn it into a failed `release`.
-    pub(crate) fn record_pending_write(
-        &mut self,
-        ino: u64,
-        size: u64,
-        mtime: i64,
-    ) -> pdfs_core::Result<()> {
+    pub(crate) fn record_pending_write(&mut self, ino: u64, size: u64, mtime: i64) -> Option<Node> {
         self.set_size(ino, size);
         self.touch_mtime(ino, mtime);
-        if let Some(e) = self.entries.get(&ino) {
-            self.db.upsert_node(&e.node).map_err(|err| {
-                warn!(uid = %e.uid, error = %err, "db upsert_node failed for a queued write");
-                err
-            })?;
-        }
-        Ok(())
+        self.entries.get(&ino).map(|e| e.node.clone())
     }
 }
 
@@ -1018,18 +1143,7 @@ mod tests {
         let dir = TempDir::new();
         let db = Db::open(&dir.0.join("cache.db")).unwrap();
         let share_access = db.all_share_access().unwrap();
-        let st = State {
-            entries: HashMap::new(),
-            by_uid: HashMap::new(),
-            children: HashMap::new(),
-            next_ino: 1,
-            active_writes: HashMap::new(),
-            handles: HashMap::new(),
-            next_fh: 1,
-            access_changes: HashSet::new(),
-            share_access,
-            db: Arc::new(db),
-        };
+        let st = State::new(Arc::new(db), share_access, 1);
         (st, dir)
     }
 
@@ -1042,8 +1156,8 @@ mod tests {
         let f = st.intern(src, node("f", "src", "f.txt", false));
         st.children.insert(src, vec![f]);
         st.children.insert(dst, vec![]);
-        st.db.set_listed(&uid("src"), true).unwrap();
-        st.db.set_listed(&uid("dst"), true).unwrap();
+        st.flushed_db().set_listed(&uid("src"), true).unwrap();
+        st.flushed_db().set_listed(&uid("dst"), true).unwrap();
         (st, dir, src, dst)
     }
 
@@ -1057,11 +1171,17 @@ mod tests {
         st.relocate(f, src, dst, &uid("dst"), "f.txt");
 
         assert!(
-            st.db.children_if_listed(&uid("dst")).unwrap().is_none(),
+            st.flushed_db()
+                .children_if_listed(&uid("dst"))
+                .unwrap()
+                .is_none(),
             "destination re-enumerates: it gained a child whose row was just deleted"
         );
         assert!(
-            st.db.children_if_listed(&uid("src")).unwrap().is_none(),
+            st.flushed_db()
+                .children_if_listed(&uid("src"))
+                .unwrap()
+                .is_none(),
             "source re-enumerates: its listing predates the move, and a stale \
              `listed` flag would hide every later remote change under it"
         );
@@ -1084,12 +1204,41 @@ mod tests {
         let (mut st, _dir, src, _dst) = two_folders();
         let f = st.by_uid[&uid("f")];
         st.relocate(f, src, src, &uid("src"), "renamed.txt");
-        assert!(st.db.children_if_listed(&uid("src")).unwrap().is_none());
+        assert!(
+            st.flushed_db()
+                .children_if_listed(&uid("src"))
+                .unwrap()
+                .is_none()
+        );
         assert!(!st.children.contains_key(&src));
         assert_eq!(st.by_uid.get(&uid("f")), Some(&f), "the inode is stable");
         assert_eq!(st.entries[&f].node.name, "renamed.txt");
     }
 
+    /// DB1: a mutation's write-through must not happen with the inode lock
+    /// held. Interning a listing commits a transaction over every child, and
+    /// every other FUSE callback used to wait behind that for no reason — the
+    /// maps it needs were updated before the first row was written.
+    #[test]
+    fn a_mutation_writes_through_only_after_the_lock_is_released() {
+        let (st, _dir) = state();
+        let db = st.db.clone();
+        let root = uid("root").to_string();
+        let state = parking_lot::Mutex::new(st);
+        {
+            let mut guard = lock_state(&state, &db);
+            guard.intern(0, node("root", "none", "My Files", true));
+            assert!(
+                db.node_by_uid(&root).unwrap().is_none(),
+                "the row must not be written while the lock is held"
+            );
+        }
+        assert!(
+            db.node_by_uid(&root).unwrap().is_some(),
+            "…and must be written by the time the guard is gone"
+        );
+    }
+
     /// The failure mode the fix exists for, stated directly: forgetting the node
     /// prunes it from the resident listing but leaves the DB claiming the source
     /// is fully enumerated. This pins *why* `relocate` cannot just call `forget`.
@@ -1097,7 +1246,7 @@ mod tests {
     fn forget_alone_leaves_the_source_claiming_a_complete_listing() {
         let (mut st, _dir, _src, _dst) = two_folders();
         st.forget(&uid("f"));
-        let listed = st.db.children_if_listed(&uid("src")).unwrap();
+        let listed = st.flushed_db().children_if_listed(&uid("src")).unwrap();
         assert!(
             listed.is_some(),
             "forget does not clear the flag — which is exactly why relocate must"
@@ -1146,7 +1295,7 @@ mod tests {
         assert!(!st.has_children(folder), "cleared memory children");
 
         // DB children
-        st.db
+        st.flushed_db()
             .upsert_node(&node("file2_uid", "dir_uid", "file2.txt", false))
             .unwrap();
         assert!(
@@ -1172,7 +1321,7 @@ mod tests {
         assert_eq!(st.entries[&shared_ino].access, Access::Editor);
         assert_eq!(st.entries[&child].access, Access::Editor);
         assert_eq!(
-            st.db.share_access(&uid("shared")).unwrap(),
+            st.flushed_db().share_access(&uid("shared")).unwrap(),
             Some(Access::Editor)
         );
     }
@@ -1184,8 +1333,8 @@ mod tests {
         let virtual_uid = NodeUid::new(VolumeId::from("virtual"), LinkId::from("sharedwithme"));
         let mut virtual_root = node("placeholder", "root", "Shared with me", true);
         virtual_root.uid = virtual_uid.clone();
-        st.db.upsert_node(&virtual_root).unwrap();
-        st.db
+        st.flushed_db().upsert_node(&virtual_root).unwrap();
+        st.flushed_db()
             .set_share_access(&virtual_uid, Access::Viewer)
             .unwrap();
         let virtual_ino = st.intern_published_share_root(owned_root, virtual_root, Access::Viewer);
@@ -1193,15 +1342,17 @@ mod tests {
         let mut shared = node("shared", "placeholder", "Retained", true);
         shared.parent_uid = Some(virtual_uid.clone());
         shared.membership = Some(membership(6));
-        st.db.upsert_node(&shared).unwrap();
-        st.db.set_share_access(&shared.uid, Access::Editor).unwrap();
+        st.flushed_db().upsert_node(&shared).unwrap();
+        st.flushed_db()
+            .set_share_access(&shared.uid, Access::Editor)
+            .unwrap();
         let shared_ino =
             st.intern_published_share_root(virtual_ino, shared.clone(), Access::Editor);
         let child = st.intern(
             shared_ino,
             node("shared-child", "shared", "pending.txt", false),
         );
-        st.db
+        st.flushed_db()
             .enqueue_op(&pdfs_core::db::PendingOp {
                 id: 0,
                 kind: pdfs_core::db::OP_RENAME.to_string(),
@@ -1219,20 +1370,20 @@ mod tests {
         assert_eq!(st.entries[&shared_ino].access, Access::Editor);
         assert_eq!(st.entries[&child].access, Access::Editor);
 
-        st.db
+        st.flushed_db()
             .publish_shared_roots(&virtual_uid, std::slice::from_ref(&shared.uid), &[])
             .unwrap();
-        let snapshot = st.db.visible_children(&virtual_uid).unwrap();
+        let snapshot = st.flushed_db().visible_children(&virtual_uid).unwrap();
         assert_eq!(snapshot.len(), 1);
         st.intern_published_share_root(virtual_ino, snapshot[0].clone(), Access::Viewer);
 
         assert_eq!(
-            st.db.share_access(&shared.uid).unwrap(),
+            st.flushed_db().share_access(&shared.uid).unwrap(),
             Some(Access::Viewer)
         );
         assert_eq!(st.entries[&shared_ino].access, Access::Viewer);
         assert_eq!(st.entries[&child].access, Access::Viewer);
-        assert_eq!(st.db.pending_ops().unwrap().len(), 1);
+        assert_eq!(st.flushed_db().pending_ops().unwrap().len(), 1);
     }
 
     #[test]
@@ -1243,12 +1394,12 @@ mod tests {
         let child_node = node("offline-child", "offline-share", "nested folder", true);
         let leaf_node = node("offline-leaf", "offline-child", "child.txt", false);
         for node in [&root_node, &shared_node, &child_node, &leaf_node] {
-            st.db.upsert_node(node).unwrap();
+            st.flushed_db().upsert_node(node).unwrap();
         }
-        st.db
+        st.flushed_db()
             .set_share_access(&uid("offline-share"), Access::Viewer)
             .unwrap();
-        st.share_access = st.db.all_share_access().unwrap();
+        st.share_access = st.flushed_db().all_share_access().unwrap();
 
         // Reconstruct in deliberately non-topological order, matching the
         // database hydration path that first allocates every inode and only
@@ -1306,7 +1457,7 @@ mod tests {
     /// `ORPHAN_INO`. `nodes[0]` becomes the mount root.
     fn hydrate_from(st: &mut State, nodes: &[Node]) -> Vec<u64> {
         for node in nodes {
-            st.db.upsert_node(node).unwrap();
+            st.flushed_db().upsert_node(node).unwrap();
         }
         let inos: Vec<u64> = nodes
             .iter()
@@ -1404,10 +1555,10 @@ mod tests {
     #[test]
     fn persisted_authority_outranks_the_own_volume_fail_open() {
         let (mut st, _dir) = state();
-        st.db
+        st.flushed_db()
             .set_share_access(&uid("device-root"), Access::Viewer)
             .unwrap();
-        st.share_access = st.db.all_share_access().unwrap();
+        st.share_access = st.flushed_db().all_share_access().unwrap();
         let nodes = [
             node("root", "none", "My Files", true),
             node("documents", "device-root", "Documents", true),
@@ -1441,7 +1592,7 @@ mod tests {
             assert_eq!(st.require_writable(ino).unwrap_err().code(), libc::EACCES);
         }
         assert_eq!(
-            st.db.share_access(&uid("shared")).unwrap(),
+            st.flushed_db().share_access(&uid("shared")).unwrap(),
             Some(Access::Viewer)
         );
     }
@@ -1451,8 +1602,10 @@ mod tests {
         for persisted in [Access::Viewer, Access::Editor] {
             let (mut st, _dir) = state();
             let owned_root = st.intern(0, node("root", "none", "My Files", true));
-            st.db.set_share_access(&uid("shared"), persisted).unwrap();
-            st.share_access = st.db.all_share_access().unwrap();
+            st.flushed_db()
+                .set_share_access(&uid("shared"), persisted)
+                .unwrap();
+            st.share_access = st.flushed_db().all_share_access().unwrap();
             let shared = st.intern(
                 owned_root,
                 node("shared", "root", "Shared without membership", true),
@@ -1464,7 +1617,10 @@ mod tests {
 
             assert_eq!(st.entries[&shared].access, persisted);
             assert_eq!(st.entries[&child].access, persisted);
-            assert_eq!(st.db.share_access(&uid("shared")).unwrap(), Some(persisted));
+            assert_eq!(
+                st.flushed_db().share_access(&uid("shared")).unwrap(),
+                Some(persisted)
+            );
         }
     }
 
@@ -1472,7 +1628,7 @@ mod tests {
     fn explicit_missing_share_role_fails_closed_and_marks_kernel_attrs() {
         let (mut st, _dir) = state();
         let owned_root = st.intern(0, node("root", "none", "My Files", true));
-        st.db
+        st.flushed_db()
             .set_share_access(&uid("shared"), Access::Editor)
             .unwrap();
         let shared = st.intern(owned_root, node("shared", "root", "Shared", true));
@@ -1484,7 +1640,7 @@ mod tests {
         assert_eq!(st.entries[&shared].access, Access::Viewer);
         assert_eq!(st.entries[&child].access, Access::Viewer);
         assert_eq!(
-            st.db.share_access(&uid("shared")).unwrap(),
+            st.flushed_db().share_access(&uid("shared")).unwrap(),
             Some(Access::Viewer)
         );
         assert!(changed.contains(&shared));
@@ -1501,7 +1657,7 @@ mod tests {
         let virtual_uid = NodeUid::new(VolumeId::from("virtual"), LinkId::from("sharedwithme"));
         let mut virtual_node = node("sharedwithme", "root", "Shared with me", true);
         virtual_node.uid = virtual_uid.clone();
-        st.db
+        st.flushed_db()
             .set_share_access(&virtual_uid, Access::Viewer)
             .unwrap();
         st.share_access.insert(virtual_uid.clone(), Access::Viewer);
@@ -1568,7 +1724,7 @@ mod tests {
         let child = st.intern(shared, node("child", "shared", "folder", true));
         let leaf = st.intern(child, node("leaf", "child", "leaf.txt", false));
 
-        st.db
+        st.flushed_db()
             .set_share_access(&uid("shared"), Access::Viewer)
             .unwrap();
         let changed = st.downgrade_shared_subtree(&uid("shared"));
@@ -1582,7 +1738,7 @@ mod tests {
         assert_eq!(st.entries[&leaf].access, Access::Viewer);
         assert_eq!(st.require_writable(leaf).unwrap_err().code(), libc::EACCES);
         assert_eq!(
-            st.db.effective_node_access(&uid("leaf")).unwrap(),
+            st.flushed_db().effective_node_access(&uid("leaf")).unwrap(),
             Some(Access::Viewer)
         );
 
@@ -1638,7 +1794,7 @@ mod tests {
         let shared = st.intern(owned_root, shared_node);
         let shared_child = st.intern(shared, node("child", "shared", "child.txt", false));
 
-        st.db.downgrade_all_share_access().unwrap();
+        st.flushed_db().downgrade_all_share_access().unwrap();
         let changed = st.downgrade_known_shared_access();
 
         assert!(changed.contains(&shared));
@@ -1662,7 +1818,7 @@ mod tests {
             let parent_link = st.entries[&parent].uid.link_id.to_string();
             parent = st.intern(parent, node(&link, &parent_link, &link, true));
         }
-        st.db.downgrade_all_share_access().unwrap();
+        st.flushed_db().downgrade_all_share_access().unwrap();
 
         let changed = st.downgrade_known_shared_access();
 
diff --git a/crates/pdfs-fuse/src/upload.rs b/crates/pdfs-fuse/src/upload.rs
index 44614a9..f1bb350 100644
--- a/crates/pdfs-fuse/src/upload.rs
+++ b/crates/pdfs-fuse/src/upload.rs
@@ -171,7 +171,7 @@ impl Core {
                     stats.bytes += size;
                     match self.fetch_node(&uid) {
                         Ok(node) => {
-                            let mut st = self.state.lock();
+                            let mut st = self.state();
                             let ino = st.intern(parent_ino, node);
                             if let Some(kids) = st.children.get_mut(&parent_ino)
                                 && !kids.contains(&ino)
@@ -216,7 +216,7 @@ impl Core {
         self.ensure_children(pino)
             .map_err(|e| self.errno_error(e, "enumerate"))?;
         let existing = {
-            let st = self.state.lock();
+            let st = self.state();
             st.children.get(&pino).and_then(|kids| {
                 kids.iter().find_map(|ino| {
                     st.entries
@@ -243,7 +243,7 @@ impl Core {
         let node = self
             .fetch_node(&new_uid)
             .map_err(|e| self.errno_error(e, "fetch node"))?;
-        let mut st = self.state.lock();
+        let mut st = self.state();
         let ino = st.intern(pino, node);
         if let Some(kids) = st.children.get_mut(&pino)
             && !kids.contains(&ino)
diff --git a/packaging/PKGBUILD b/packaging/PKGBUILD
index 08c8b2c..272d742 100644
--- a/packaging/PKGBUILD
+++ b/packaging/PKGBUILD
@@ -9,7 +9,7 @@
 # Build + install:   cd packaging && makepkg -fi
 
 pkgname=proton-drive-linux
-pkgver=1.6.0
+pkgver=1.7.0
 pkgrel=1
 pkgdesc="Proton Drive client for Linux: FUSE files-on-demand mount, CLI, GTK4 app + tray"
 arch=('x86_64')