nvrl/proton-drive-linux
1
0 Fork 0
mirror of https://github.com/narrrl/proton-drive-linux.git synced 2026-11-04 16:56:59 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(perf) read-only connection pool, States doesn't write to db, search optimized

Browse Source
This commit is contained in:
nvrl committed 2026-08-16 00:33:27 +02:00
1 parent e48c4bb4b2
commit 35418eea3f
28 files changed
+1320 -413
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/pdfs-core/src/db/activity.rs
+1 -1
View File
@@ -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",
crates/pdfs-core/src/db/albums.rs
+5 -5
View File
@@ -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
crates/pdfs-core/src/db/cache.rs
+3 -3
View File
@@ -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",
crates/pdfs-core/src/db/devices.rs
+1 -1
View File
@@ -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",
[],
crates/pdfs-core/src/db/local.rs
+3 -3
View File
@@ -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() {
crates/pdfs-core/src/db/migrations.rs
+56 -1
View File
@@ -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;
";
crates/pdfs-core/src/db/mod.rs
+141 -3
View File
@@ -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()?;
crates/pdfs-core/src/db/mounts.rs
+1 -1
View File
@@ -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,
crates/pdfs-core/src/db/nodes.rs
+193 -71
View File
@@ -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(())
crates/pdfs-core/src/db/ops.rs
+10 -10
View File
@@ -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
crates/pdfs-core/src/db/photos.rs
+5 -5
View File
@@ -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)
}
crates/pdfs-core/src/db/pins.rs
+4 -4
View File
@@ -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
crates/pdfs-core/src/db/share_access.rs
+3 -3
View File
@@ -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()],
crates/pdfs-core/src/db/state.rs
+3 -3
View File
@@ -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",
crates/pdfs-core/src/db/sync.rs
+4 -4
View File
@@ -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
crates/pdfs-core/src/db/tests.rs
+263
View File
@@ -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()));
}
crates/pdfs-core/src/db/trash.rs
+1 -1
View File
@@ -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",
crates/pdfs-core/src/db/utils.rs
+45 -14
View File
@@ -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