mirror of
https://github.com/narrrl/proton-drive-linux.git
synced 2026-11-03 16:41:26 +01:00
added local file search to prompt
This commit is contained in:
1 parent
8f28f5fd22
commit
dacacf51bb
12 files changed
+1742
-953
No files matched your search
Generated
+59
@@ -468,6 +468,16 @@ dependencies = [
|
||||
"cipher 0.5.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "bstr"
|
||||
version = "1.12.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5cee35f73844aa3014bb606320a6c1f010249dbdf43342fe54b5a4f6a8ed4b79"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "buffer-redux"
|
||||
version = "1.1.0"
|
||||
@@ -804,6 +814,25 @@ version = "0.1.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fd121741cf3eb82c08dd3023eb55bf2665e5f60ec20f89760cf836ae4562e6a0"
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-deque"
|
||||
version = "0.8.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb"
|
||||
dependencies = [
|
||||
"crossbeam-epoch",
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-epoch"
|
||||
version = "0.9.20"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-utils"
|
||||
version = "0.8.21"
|
||||
@@ -1689,6 +1718,19 @@ dependencies = [
|
||||
"system-deps",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "globset"
|
||||
version = "0.4.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "52dfc19153a48bde0cbd630453615c8151bce3a5adfac7a0aebfbf0a1e1f57e3"
|
||||
dependencies = [
|
||||
"aho-corasick",
|
||||
"bstr",
|
||||
"log",
|
||||
"regex-automata",
|
||||
"regex-syntax",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "gobject-sys"
|
||||
version = "0.20.10"
|
||||
@@ -2152,6 +2194,22 @@ dependencies = [
|
||||
"icu_properties",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ignore"
|
||||
version = "0.4.28"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2adf14691c72bcfc1058740436a35bdd3ae9c07d1a941ef00b749e9ea16aefa7"
|
||||
dependencies = [
|
||||
"crossbeam-deque",
|
||||
"globset",
|
||||
"log",
|
||||
"memchr",
|
||||
"regex-automata",
|
||||
"same-file",
|
||||
"walkdir",
|
||||
"winapi-util",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "indexmap"
|
||||
version = "2.14.0"
|
||||
@@ -2858,6 +2916,7 @@ name = "pdfs-core"
|
||||
version = "0.2.5"
|
||||
dependencies = [
|
||||
"directories",
|
||||
"ignore",
|
||||
"keyring",
|
||||
"proton-drive-rs",
|
||||
"proton-sdk",
|
||||
|
||||
@@ -25,6 +25,7 @@ anyhow = "1"
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
||||
directories = "5"
|
||||
ignore = "0.4"
|
||||
sha2 = "0.10"
|
||||
rusqlite = { version = "0.32", features = ["bundled"] }
|
||||
keyring = { version = "3", features = ["sync-secret-service", "crypto-rust"] }
|
||||
|
||||
@@ -251,7 +251,9 @@ fn cmd_daemon(mountpoint: Option<PathBuf>) -> Result<()> {
|
||||
std::thread::sleep(std::time::Duration::from_secs(2));
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!(error = %e, "mount failed; retrying in 5s");
|
||||
// `{:#}` so the whole anyhow chain lands in the log — the top
|
||||
// context alone ("mount failed") never says *why*.
|
||||
tracing::error!(error = format!("{e:#}"), "mount failed; retrying in 5s");
|
||||
std::thread::sleep(std::time::Duration::from_secs(5));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,6 +16,7 @@ serde_json.workspace = true
|
||||
thiserror.workspace = true
|
||||
tracing.workspace = true
|
||||
directories.workspace = true
|
||||
ignore.workspace = true
|
||||
sha2.workspace = true
|
||||
rusqlite.workspace = true
|
||||
keyring.workspace = true
|
||||
@@ -178,15 +178,13 @@ fn register_refresh_handler(session: &ProtonApiSession, mailbox_password: String
|
||||
password_mode,
|
||||
mailbox_password: mailbox_password.clone(),
|
||||
};
|
||||
if let Ok(json) = serde_json::to_string(&stored) {
|
||||
if let Ok(entry) = keyring_entry() {
|
||||
if let Ok(json) = serde_json::to_string(&stored)
|
||||
&& let Ok(entry) = keyring_entry() {
|
||||
if let Err(e) = entry.set_password(&json) {
|
||||
tracing::warn!(error = %e, "failed to auto-persist refreshed tokens in keyring");
|
||||
} else {
|
||||
tracing::info!("successfully auto-persisted refreshed tokens in keyring");
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -142,6 +142,12 @@ impl AppDirs {
|
||||
self.state_dir().join("tray.sock")
|
||||
}
|
||||
|
||||
/// The user's home directory — the root the daemon indexes for local
|
||||
/// (non-Drive) file search. `None` if it cannot be resolved.
|
||||
pub fn home_dir(&self) -> Option<PathBuf> {
|
||||
directories::UserDirs::new().map(|u| u.home_dir().to_path_buf())
|
||||
}
|
||||
|
||||
/// Default mountpoint when the user does not pass one explicitly.
|
||||
pub fn default_mountpoint(&self) -> PathBuf {
|
||||
directories::UserDirs::new()
|
||||
|
||||
@@ -63,6 +63,10 @@ pub enum Request {
|
||||
/// Full-text search node names against the daemon's local metadata index.
|
||||
/// `limit` caps the number of hits returned. Replies with [`Response::SearchResults`].
|
||||
Search { query: String, limit: usize },
|
||||
/// Search the daemon's index of *local* (non-Drive) files on this machine.
|
||||
/// Independent of [`Request::Search`] so a front-end can fire both at once and
|
||||
/// render whichever lands first. Replies with [`Response::LocalResults`].
|
||||
SearchLocal { query: String, limit: usize },
|
||||
/// Rename a file or folder. `path` is mountpoint-relative; `new_name` is a
|
||||
/// single path component (no separators). Replies with [`Response::Ok`].
|
||||
Rename { path: String, new_name: String },
|
||||
@@ -162,6 +166,20 @@ pub struct SearchHit {
|
||||
pub uid: String,
|
||||
}
|
||||
|
||||
/// One hit in a [`Request::SearchLocal`] result: a file on this machine, outside
|
||||
/// Proton Drive. Unlike [`SearchHit`] there is no uid or pin state — the file is
|
||||
/// already local, so the front-end opens `path` directly.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
pub struct LocalHit {
|
||||
pub name: String,
|
||||
/// Absolute path on this machine.
|
||||
pub path: String,
|
||||
pub is_dir: bool,
|
||||
pub size: u64,
|
||||
/// Modification time, epoch seconds.
|
||||
pub modified: i64,
|
||||
}
|
||||
|
||||
/// One photo in a [`Request::PhotosTimeline`] page.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||
pub struct PhotoItem {
|
||||
@@ -206,6 +224,10 @@ pub enum Response {
|
||||
FilePath { path: String },
|
||||
/// Full-text search results (reply to [`Request::Search`]).
|
||||
SearchResults { hits: Vec<SearchHit> },
|
||||
/// Local-file search results (reply to [`Request::SearchLocal`]). `indexing`
|
||||
/// is true while a scan of the machine is still running, so a front-end can
|
||||
/// say "still indexing" instead of "no matches" on a cold first launch.
|
||||
LocalResults { hits: Vec<LocalHit>, indexing: bool },
|
||||
/// A snapshot of in-flight transfers (reply to [`Request::GetQueueStatus`]).
|
||||
Transfers { items: Vec<TransferItem> },
|
||||
/// The request failed.
|
||||
|
||||
+253
-3
@@ -18,9 +18,10 @@ use proton_drive_rs::{Node, NodeKind};
|
||||
use rusqlite::{Connection, OptionalExtension, params};
|
||||
|
||||
use crate::Result;
|
||||
use crate::localindex::LocalEntry;
|
||||
|
||||
/// Current schema version. Bump on every forward migration added below.
|
||||
const SCHEMA_VERSION: i64 = 5;
|
||||
const SCHEMA_VERSION: i64 = 6;
|
||||
|
||||
/// Below this length the trigram tokenizer indexes nothing (it needs 3-char
|
||||
/// grams), so short queries fall back to a `LIKE` scan over `nodes.name`.
|
||||
@@ -119,6 +120,9 @@ impl Db {
|
||||
if current < 5 {
|
||||
tx.execute_batch(MIGRATION_V5)?;
|
||||
}
|
||||
if current < 6 {
|
||||
tx.execute_batch(MIGRATION_V6)?;
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT INTO sync_state (key, value) VALUES ('schema_version', ?1)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
@@ -526,6 +530,161 @@ impl Db {
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
// ---- local (non-Drive) file index (schema v6) ----
|
||||
|
||||
/// Open a new local-index scan generation. Rows written under it survive
|
||||
/// [`local_finish_scan`](Self::local_finish_scan); rows still carrying an
|
||||
/// older generation are pruned there as "no longer on disk".
|
||||
pub fn local_begin_scan(&self) -> Result<i64> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let current: i64 = conn
|
||||
.query_row(
|
||||
"SELECT value FROM sync_state WHERE key = 'local_scan_gen'",
|
||||
[],
|
||||
|row| row.get::<_, String>(0),
|
||||
)
|
||||
.optional()?
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(0);
|
||||
let next = current + 1;
|
||||
conn.execute(
|
||||
"INSERT INTO sync_state (key, value) VALUES ('local_scan_gen', ?1)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
[next.to_string()],
|
||||
)?;
|
||||
Ok(next)
|
||||
}
|
||||
|
||||
/// Write one batch of walked entries under scan generation `generation`. The
|
||||
/// FTS index is *not* touched here — it is rebuilt once in
|
||||
/// [`local_finish_scan`](Self::local_finish_scan).
|
||||
pub fn local_upsert_batch(&self, generation: i64, entries: &[LocalEntry]) -> Result<()> {
|
||||
let mut conn = self.conn.lock().unwrap();
|
||||
let tx = conn.transaction()?;
|
||||
{
|
||||
let mut stmt = tx.prepare_cached(
|
||||
"INSERT INTO local_files (path, name, is_dir, size, mtime, scan_gen)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
|
||||
ON CONFLICT(path) DO UPDATE SET
|
||||
name = excluded.name,
|
||||
is_dir = excluded.is_dir,
|
||||
size = excluded.size,
|
||||
mtime = excluded.mtime,
|
||||
scan_gen = excluded.scan_gen",
|
||||
)?;
|
||||
for e in entries {
|
||||
stmt.execute(params![
|
||||
e.path,
|
||||
e.name,
|
||||
e.is_dir as i64,
|
||||
e.size,
|
||||
e.mtime,
|
||||
generation
|
||||
])?;
|
||||
}
|
||||
}
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Close scan generation `generation`: drop every row an older scan wrote
|
||||
/// (those paths are gone from disk), rebuild the FTS index over what remains,
|
||||
/// and stamp the completion time. Returns the number of indexed entries.
|
||||
pub fn local_finish_scan(&self, generation: i64, finished_at: i64) -> Result<i64> {
|
||||
let mut conn = self.conn.lock().unwrap();
|
||||
let tx = conn.transaction()?;
|
||||
tx.execute(
|
||||
"DELETE FROM local_files WHERE scan_gen != ?1",
|
||||
params![generation],
|
||||
)?;
|
||||
tx.execute_batch("INSERT INTO local_fts(local_fts) VALUES('rebuild');")?;
|
||||
tx.execute(
|
||||
"INSERT INTO sync_state (key, value) VALUES ('local_indexed_at', ?1)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
[finished_at.to_string()],
|
||||
)?;
|
||||
let count: i64 = tx.query_row("SELECT COUNT(*) FROM local_files", [], |r| r.get(0))?;
|
||||
tx.commit()?;
|
||||
Ok(count)
|
||||
}
|
||||
|
||||
/// When the last local scan completed (epoch seconds), or `None` if the index
|
||||
/// 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().unwrap();
|
||||
Ok(conn
|
||||
.query_row(
|
||||
"SELECT value FROM sync_state WHERE key = 'local_indexed_at'",
|
||||
[],
|
||||
|row| row.get::<_, String>(0),
|
||||
)
|
||||
.optional()?
|
||||
.and_then(|v| v.parse().ok()))
|
||||
}
|
||||
|
||||
/// Substring search over indexed local file names, newest-modified first
|
||||
/// within a relevance tier. Mirrors [`search`](Self::search): the trigram
|
||||
/// index handles queries of at least [`TRIGRAM_MIN`] chars, shorter ones fall
|
||||
/// back to a `LIKE` scan.
|
||||
pub fn search_local(&self, query: &str, limit: usize) -> Result<Vec<LocalFileHit>> {
|
||||
let query = query.trim();
|
||||
if query.is_empty() {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let mut stmt;
|
||||
let rows = if query.chars().count() < TRIGRAM_MIN {
|
||||
let pat = format!("%{}%", like_escape(query));
|
||||
stmt = conn.prepare(
|
||||
"SELECT path, name, is_dir, size, mtime FROM local_files
|
||||
WHERE name LIKE ?1 ESCAPE '\\'
|
||||
ORDER BY mtime DESC LIMIT ?2",
|
||||
)?;
|
||||
stmt.query_map(params![pat, limit as i64], local_hit)?
|
||||
} else {
|
||||
let phrase = query
|
||||
.split_whitespace()
|
||||
.map(|word| format!("\"{}\"", word.replace('"', "\"\"")))
|
||||
.collect::<Vec<_>>()
|
||||
.join(" AND ");
|
||||
stmt = conn.prepare(
|
||||
"SELECT f.path, f.name, f.is_dir, f.size, f.mtime
|
||||
FROM local_fts x JOIN local_files f ON f.id = x.rowid
|
||||
WHERE x.name MATCH ?1
|
||||
ORDER BY x.rank LIMIT ?2",
|
||||
)?;
|
||||
stmt.query_map(params![phrase, limit as i64], local_hit)?
|
||||
};
|
||||
let mut out = Vec::new();
|
||||
for r in rows {
|
||||
out.push(r?);
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
}
|
||||
|
||||
/// One hit from [`Db::search_local`]: an indexed file on the machine itself, not
|
||||
/// in Drive. `path` is absolute.
|
||||
pub struct LocalFileHit {
|
||||
pub path: String,
|
||||
pub name: String,
|
||||
pub is_dir: bool,
|
||||
pub size: i64,
|
||||
pub mtime: i64,
|
||||
}
|
||||
|
||||
/// Row mapper for [`Db::search_local`]'s two query paths, which select the same
|
||||
/// columns in the same order.
|
||||
fn local_hit(row: &rusqlite::Row) -> rusqlite::Result<LocalFileHit> {
|
||||
Ok(LocalFileHit {
|
||||
path: row.get(0)?,
|
||||
name: row.get(1)?,
|
||||
is_dir: row.get::<_, i64>(2)? != 0,
|
||||
size: row.get(3)?,
|
||||
mtime: row.get(4)?,
|
||||
})
|
||||
}
|
||||
|
||||
/// Row mapper for the `(node_json, uid)` pairs both search paths return.
|
||||
@@ -664,6 +823,32 @@ CREATE TABLE pins (
|
||||
);
|
||||
";
|
||||
|
||||
/// Schema v6: the index of *local* (non-Drive) files, so the launcher prompt can
|
||||
/// search the machine alongside Drive. `local_files` is keyed by absolute path;
|
||||
/// `scan_gen` stamps the scan that last saw a row, so a rescan prunes vanished
|
||||
/// files with one `DELETE` instead of diffing.
|
||||
///
|
||||
/// `local_fts` is an *external-content* FTS5 index over `local_files.name`: the
|
||||
/// text lives once in the base table and the index is rebuilt in bulk at the end
|
||||
/// of a scan (`INSERT INTO local_fts(local_fts) VALUES('rebuild')`). That is far
|
||||
/// cheaper than the delete-then-insert-per-row dance `nodes_fts` needs, because
|
||||
/// a scan rewrites most rows at once rather than trickling single updates.
|
||||
const MIGRATION_V6: &str = "
|
||||
CREATE TABLE local_files (
|
||||
id INTEGER PRIMARY KEY,
|
||||
path TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
is_dir INTEGER NOT NULL,
|
||||
size INTEGER NOT NULL DEFAULT 0,
|
||||
mtime INTEGER NOT NULL DEFAULT 0,
|
||||
scan_gen INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
CREATE VIRTUAL TABLE local_fts USING fts5(
|
||||
name, content='local_files', content_rowid='id', tokenize='trigram'
|
||||
);
|
||||
";
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -788,6 +973,62 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
fn local(path: &str, name: &str, is_dir: bool) -> LocalEntry {
|
||||
LocalEntry {
|
||||
path: path.into(),
|
||||
name: name.into(),
|
||||
is_dir,
|
||||
size: 10,
|
||||
mtime: 5,
|
||||
}
|
||||
}
|
||||
|
||||
/// A local scan is searchable by substring, and a *later* scan prunes the
|
||||
/// paths it no longer sees — including out of the FTS index, so a deleted
|
||||
/// file cannot keep surfacing in the prompt.
|
||||
#[test]
|
||||
fn local_scan_indexes_then_prunes_stale_paths() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
|
||||
let gen1 = db.local_begin_scan().unwrap();
|
||||
db.local_upsert_batch(
|
||||
gen1,
|
||||
&[
|
||||
local("/home/u/docs/report.pdf", "report.pdf", false),
|
||||
local("/home/u/docs/notes.md", "notes.md", false),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(db.local_finish_scan(gen1, 1_000).unwrap(), 2);
|
||||
|
||||
// Trigram index gives substring (not just prefix) matches.
|
||||
let hits = db.search_local("port", 10).unwrap();
|
||||
assert_eq!(hits.len(), 1);
|
||||
assert_eq!(hits[0].path, "/home/u/docs/report.pdf");
|
||||
assert!(!hits[0].is_dir);
|
||||
assert_eq!(db.local_indexed_at().unwrap(), Some(1_000));
|
||||
|
||||
// Second scan sees only notes.md → report.pdf is gone from disk.
|
||||
let gen2 = db.local_begin_scan().unwrap();
|
||||
db.local_upsert_batch(gen2, &[local("/home/u/docs/notes.md", "notes.md", false)])
|
||||
.unwrap();
|
||||
assert_eq!(db.local_finish_scan(gen2, 2_000).unwrap(), 1);
|
||||
assert!(db.search_local("report", 10).unwrap().is_empty());
|
||||
assert_eq!(db.search_local("notes", 10).unwrap().len(), 1);
|
||||
}
|
||||
|
||||
/// Queries below the trigram minimum still match, via the `LIKE` fallback.
|
||||
#[test]
|
||||
fn local_search_short_query_like_fallback() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
let g = db.local_begin_scan().unwrap();
|
||||
db.local_upsert_batch(g, &[local("/home/u/a.txt", "a.txt", false)])
|
||||
.unwrap();
|
||||
db.local_finish_scan(g, 1).unwrap();
|
||||
assert_eq!(db.search_local("a", 10).unwrap().len(), 1);
|
||||
assert!(db.search_local("", 10).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_cursor_roundtrip() {
|
||||
let db = Db::open_in_memory().unwrap();
|
||||
@@ -809,7 +1050,13 @@ mod tests {
|
||||
db.upsert_node(&file("f1", "docs", "report.pdf", 1))
|
||||
.unwrap();
|
||||
db.upsert_node(&file("f2", "root", "notes.txt", 2)).unwrap();
|
||||
db.upsert_node(&file("f3", "root", "Rampage Open Air 2026 - order 166765244.pdf", 3)).unwrap();
|
||||
db.upsert_node(&file(
|
||||
"f3",
|
||||
"root",
|
||||
"Rampage Open Air 2026 - order 166765244.pdf",
|
||||
3,
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
// Substring match (trigram), not just prefix.
|
||||
let hits = db.search("port", 10).unwrap();
|
||||
@@ -825,7 +1072,10 @@ mod tests {
|
||||
// Multi-term FTS5 query matching (out of order, separated terms)
|
||||
let hits = db.search("rampage 2026", 10).unwrap();
|
||||
assert_eq!(hits.len(), 1);
|
||||
assert_eq!(hits[0].node.name, "Rampage Open Air 2026 - order 166765244.pdf");
|
||||
assert_eq!(
|
||||
hits[0].node.name,
|
||||
"Rampage Open Air 2026 - order 166765244.pdf"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -8,6 +8,7 @@ pub mod config;
|
||||
pub mod control;
|
||||
pub mod db;
|
||||
pub mod error;
|
||||
pub mod localindex;
|
||||
pub mod service;
|
||||
|
||||
pub use error::{Error, Result};
|
||||
|
||||
@@ -0,0 +1,224 @@
|
||||
//! Filesystem walker behind the daemon's index of *local* (non-Drive) files.
|
||||
//!
|
||||
//! The launcher prompt searches Proton Drive and the machine's own files side by
|
||||
//! side. Drive names come from the `nodes` FTS index; local names come from the
|
||||
//! `local_files` index this module feeds. The daemon runs [`scan`] on a
|
||||
//! background thread — never on a FUSE or control-socket thread — and streams
|
||||
//! batches into [`crate::db::Db::local_upsert_batch`].
|
||||
//!
|
||||
//! Walking is deliberately shallow on cost: we stat each entry once (the walker
|
||||
//! already has the `DirEntry` metadata) and skip the directories that dominate a
|
||||
//! home directory's inode count without ever being interesting to search
|
||||
//! (`node_modules`, `target`, caches, VCS internals). The Drive mountpoint is
|
||||
//! always excluded — walking it would fault in every remote node through FUSE.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::UNIX_EPOCH;
|
||||
|
||||
use ignore::{WalkBuilder, WalkState};
|
||||
|
||||
/// Directory names skipped anywhere in the tree. These hold build artefacts,
|
||||
/// dependency trees, and caches: high inode count, no search value.
|
||||
const SKIP_DIRS: &[&str] = &[
|
||||
"node_modules",
|
||||
"target",
|
||||
"__pycache__",
|
||||
"venv",
|
||||
".venv",
|
||||
"vendor",
|
||||
"dist-newstyle",
|
||||
"Trash",
|
||||
];
|
||||
|
||||
/// Hard cap on indexed entries, so a pathological home directory cannot grow the
|
||||
/// database without bound. Reaching it stops the walk early.
|
||||
const MAX_ENTRIES: usize = 400_000;
|
||||
|
||||
/// Rows handed to the sink at a time. Large enough that the per-transaction cost
|
||||
/// amortises, small enough that the writer lock is released often (FUSE
|
||||
/// write-through shares the same connection).
|
||||
const BATCH: usize = 2_000;
|
||||
|
||||
/// One indexed local file or directory.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct LocalEntry {
|
||||
/// Absolute path.
|
||||
pub path: String,
|
||||
/// Final path component.
|
||||
pub name: String,
|
||||
pub is_dir: bool,
|
||||
/// Size in bytes (0 for directories).
|
||||
pub size: i64,
|
||||
/// Modification time, epoch seconds.
|
||||
pub mtime: i64,
|
||||
}
|
||||
|
||||
/// Walk `roots` in parallel, calling `sink` with batches of entries.
|
||||
///
|
||||
/// Hidden files and anything under `excludes` (the Drive mountpoint, our own
|
||||
/// state/cache dirs) are skipped, as are the [`SKIP_DIRS`] names at any depth.
|
||||
/// Symlinks are not followed, so a link loop cannot hang the scan. Returns the
|
||||
/// number of entries handed to `sink`.
|
||||
///
|
||||
/// `sink` is called from the walker's worker threads, serialized by an internal
|
||||
/// mutex; it must not block for long.
|
||||
pub fn scan(
|
||||
roots: &[PathBuf],
|
||||
excludes: &[PathBuf],
|
||||
sink: impl FnMut(Vec<LocalEntry>) + Send,
|
||||
) -> usize {
|
||||
let Some((first, rest)) = roots.split_first() else {
|
||||
return 0;
|
||||
};
|
||||
|
||||
let mut builder = WalkBuilder::new(first);
|
||||
for root in rest {
|
||||
builder.add(root);
|
||||
}
|
||||
let excludes: Vec<PathBuf> = excludes.to_vec();
|
||||
builder
|
||||
// `standard_filters` would honour .gitignore/.ignore files: a source
|
||||
// tree's ignored-but-present files are still files the user may want to
|
||||
// find, so only the hidden filter stays on.
|
||||
.standard_filters(false)
|
||||
.hidden(true)
|
||||
.follow_links(false)
|
||||
.same_file_system(true)
|
||||
.threads(std::thread::available_parallelism().map_or(4, |n| n.get().min(8)))
|
||||
.filter_entry(move |entry| {
|
||||
let path = entry.path();
|
||||
if excludes.iter().any(|e| path.starts_with(e)) {
|
||||
return false;
|
||||
}
|
||||
!matches!(entry.file_name().to_str(), Some(name) if SKIP_DIRS.contains(&name))
|
||||
});
|
||||
|
||||
let state = std::sync::Mutex::new(SinkState {
|
||||
buf: Vec::with_capacity(BATCH),
|
||||
total: 0,
|
||||
sink,
|
||||
});
|
||||
|
||||
builder.build_parallel().run(|| {
|
||||
Box::new(|result| {
|
||||
let Ok(entry) = result else {
|
||||
return WalkState::Continue;
|
||||
};
|
||||
// Depth 0 is a root itself; indexing it adds nothing to search.
|
||||
if entry.depth() == 0 {
|
||||
return WalkState::Continue;
|
||||
}
|
||||
let Some(local) = to_entry(&entry) else {
|
||||
return WalkState::Continue;
|
||||
};
|
||||
|
||||
let mut state = state.lock().unwrap();
|
||||
state.push(local);
|
||||
if state.total >= MAX_ENTRIES {
|
||||
return WalkState::Quit;
|
||||
}
|
||||
WalkState::Continue
|
||||
})
|
||||
});
|
||||
|
||||
let mut state = state.into_inner().unwrap();
|
||||
state.flush();
|
||||
state.total
|
||||
}
|
||||
|
||||
/// Batching buffer shared by the walker's worker threads.
|
||||
struct SinkState<F: FnMut(Vec<LocalEntry>)> {
|
||||
buf: Vec<LocalEntry>,
|
||||
total: usize,
|
||||
sink: F,
|
||||
}
|
||||
|
||||
impl<F: FnMut(Vec<LocalEntry>)> SinkState<F> {
|
||||
fn push(&mut self, entry: LocalEntry) {
|
||||
self.buf.push(entry);
|
||||
self.total += 1;
|
||||
if self.buf.len() >= BATCH {
|
||||
self.flush();
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self) {
|
||||
if !self.buf.is_empty() {
|
||||
let batch = std::mem::take(&mut self.buf);
|
||||
(self.sink)(batch);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert a walker entry into a [`LocalEntry`], reusing the metadata the walker
|
||||
/// already fetched. Entries with non-UTF-8 paths are dropped: the index (and the
|
||||
/// JSON wire format) is UTF-8 only.
|
||||
fn to_entry(entry: &ignore::DirEntry) -> Option<LocalEntry> {
|
||||
let path = entry.path().to_str()?.to_string();
|
||||
let name = entry.file_name().to_str()?.to_string();
|
||||
let meta = entry.metadata().ok()?;
|
||||
let is_dir = meta.is_dir();
|
||||
let mtime = meta
|
||||
.modified()
|
||||
.ok()
|
||||
.and_then(|t| t.duration_since(UNIX_EPOCH).ok())
|
||||
.map_or(0, |d| d.as_secs() as i64);
|
||||
Some(LocalEntry {
|
||||
path,
|
||||
name,
|
||||
is_dir,
|
||||
size: if is_dir { 0 } else { meta.len() as i64 },
|
||||
mtime,
|
||||
})
|
||||
}
|
||||
|
||||
/// The paths a scan should never descend into: the Drive mountpoint (walking it
|
||||
/// would fault every remote node in through FUSE) plus our own state and cache
|
||||
/// dirs, which hold blobs no user searches for by name.
|
||||
pub fn default_excludes(mountpoint: &Path, state_dir: &Path, cache_dir: &Path) -> Vec<PathBuf> {
|
||||
vec![
|
||||
mountpoint.to_path_buf(),
|
||||
state_dir.to_path_buf(),
|
||||
cache_dir.to_path_buf(),
|
||||
]
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// A scan indexes plain files, skips the excluded subtree and the junk dirs,
|
||||
/// and reports names/paths the search index can match on.
|
||||
#[test]
|
||||
fn scan_indexes_files_and_honours_excludes() {
|
||||
let tmp = std::env::temp_dir().join(format!("pdfs-scan-{}", std::process::id()));
|
||||
let _ = std::fs::remove_dir_all(&tmp);
|
||||
std::fs::create_dir_all(tmp.join("docs")).unwrap();
|
||||
std::fs::create_dir_all(tmp.join("node_modules/pkg")).unwrap();
|
||||
std::fs::create_dir_all(tmp.join("mnt")).unwrap();
|
||||
std::fs::write(tmp.join("docs/report.pdf"), b"x").unwrap();
|
||||
std::fs::write(tmp.join("node_modules/pkg/index.js"), b"x").unwrap();
|
||||
std::fs::write(tmp.join("mnt/remote.txt"), b"x").unwrap();
|
||||
|
||||
let mut got = Vec::new();
|
||||
let n = scan(std::slice::from_ref(&tmp), &[tmp.join("mnt")], |batch| {
|
||||
got.extend(batch)
|
||||
});
|
||||
|
||||
let names: Vec<&str> = got.iter().map(|e| e.name.as_str()).collect();
|
||||
assert_eq!(n, got.len());
|
||||
assert!(names.contains(&"report.pdf"));
|
||||
assert!(names.contains(&"docs"));
|
||||
// Junk dir pruned, mountpoint-style exclude pruned.
|
||||
assert!(!names.contains(&"index.js"));
|
||||
assert!(!names.contains(&"node_modules"));
|
||||
assert!(!names.contains(&"remote.txt"));
|
||||
|
||||
let report = got.iter().find(|e| e.name == "report.pdf").unwrap();
|
||||
assert!(!report.is_dir);
|
||||
assert_eq!(report.size, 1);
|
||||
assert!(report.path.ends_with("docs/report.pdf"));
|
||||
|
||||
std::fs::remove_dir_all(&tmp).unwrap();
|
||||
}
|
||||
}
|
||||
+121
-1
@@ -38,6 +38,7 @@ use std::io::{BufRead, BufReader, Write as _};
|
||||
use std::os::unix::fs::FileExt;
|
||||
use std::os::unix::net::{UnixListener, UnixStream};
|
||||
use std::path::{Component, Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
||||
|
||||
@@ -51,10 +52,11 @@ use fuser::{
|
||||
use pdfs_core::cache::{BLOCK_SIZE, ContentCache};
|
||||
use pdfs_core::config::AppDirs;
|
||||
use pdfs_core::control::{
|
||||
DirEntry, PhotoItem, Request as CtlRequest, Response as CtlResponse, SearchHit,
|
||||
DirEntry, LocalHit, PhotoItem, Request as CtlRequest, Response as CtlResponse, SearchHit,
|
||||
TransferDirection,
|
||||
};
|
||||
use pdfs_core::db::{Db, StoredNode};
|
||||
use pdfs_core::localindex;
|
||||
use proton_drive_rs::proton_sdk::error::ProtonError;
|
||||
use proton_drive_rs::proton_sdk::ids::{DriveEventId, LinkId, NodeUid, VolumeId};
|
||||
use proton_drive_rs::{
|
||||
@@ -78,6 +80,15 @@ const POLL_INTERVAL: Duration = Duration::from_secs(10);
|
||||
/// in [`Core`] and serve every page from memory until it goes stale.
|
||||
const TIMELINE_TTL: Duration = Duration::from_secs(60);
|
||||
|
||||
/// How stale the local-file index may get before the background scanner rebuilds
|
||||
/// it. A rescan is a full walk of `$HOME`, so this trades index freshness against
|
||||
/// disk churn; the scanner also always rebuilds once per daemon start when the
|
||||
/// index is older than this.
|
||||
const LOCAL_INDEX_TTL: Duration = Duration::from_secs(15 * 60);
|
||||
|
||||
/// How often the scanner thread wakes to check whether the index went stale.
|
||||
const LOCAL_INDEX_CHECK: Duration = Duration::from_secs(60);
|
||||
|
||||
/// The FUSE root inode is always 1.
|
||||
const ROOT_INO: u64 = 1;
|
||||
|
||||
@@ -301,6 +312,9 @@ struct Core {
|
||||
/// In-flight upload/download progress, served to `GetQueueStatus`. Shared
|
||||
/// across the FUSE session and the control-socket task.
|
||||
transfers: Arc<TransferRegistry>,
|
||||
/// True while the background scanner is rebuilding the local-file index, so
|
||||
/// `SearchLocal` can tell a front-end "still indexing" apart from "no match".
|
||||
indexing: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
/// The whole photos timeline plus when it was fetched, for TTL freshness.
|
||||
@@ -1001,6 +1015,26 @@ impl Core {
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Search the index of files on this machine (outside Drive), built by the
|
||||
/// background scanner in [`run_local_index`]. Pure local lookup, never hits
|
||||
/// the network — and never touches the FUSE mount, which the scanner excludes.
|
||||
fn search_local(&self, query: &str, limit: usize) -> Result<Vec<LocalHit>, String> {
|
||||
let hits = self
|
||||
.db
|
||||
.search_local(query, limit)
|
||||
.map_err(|e| format!("local search: {e}"))?;
|
||||
Ok(hits
|
||||
.into_iter()
|
||||
.map(|h| LocalHit {
|
||||
name: h.name,
|
||||
path: h.path,
|
||||
is_dir: h.is_dir,
|
||||
size: h.size.max(0) as u64,
|
||||
modified: h.mtime,
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// A page of the photos timeline (newest first), with the page's thumbnails
|
||||
/// fetched into the cache and their paths attached. `Ok(None)` when the
|
||||
/// account has no photos volume.
|
||||
@@ -2293,6 +2327,73 @@ async fn run_event_sync(
|
||||
}
|
||||
}
|
||||
|
||||
/// Keep the local-file index fresh for the launcher prompt's "This computer"
|
||||
/// results. Rebuilds the index whenever it is older than [`LOCAL_INDEX_TTL`],
|
||||
/// then sleeps; runs on its own thread for the life of the daemon.
|
||||
///
|
||||
/// The walk is the one part of the daemon that touches the wider filesystem, so
|
||||
/// it is deliberately kept off every hot path: it never runs on a FUSE or
|
||||
/// control-socket thread, and it excludes the mountpoint (walking it would fault
|
||||
/// every remote node in through FUSE, defeating on-demand hydration).
|
||||
fn run_local_index(db: Arc<Db>, indexing: Arc<AtomicBool>, mountpoint: PathBuf) {
|
||||
loop {
|
||||
let age = db.local_indexed_at().ok().flatten();
|
||||
let stale =
|
||||
age.is_none_or(|at| now_secs().saturating_sub(at) >= LOCAL_INDEX_TTL.as_secs() as i64);
|
||||
if stale {
|
||||
scan_local_once(&db, &indexing, &mountpoint);
|
||||
}
|
||||
std::thread::sleep(LOCAL_INDEX_CHECK);
|
||||
}
|
||||
}
|
||||
|
||||
/// Walk `$HOME` once and replace the local-file index with what it finds.
|
||||
/// Batches stream straight into SQLite, so peak memory is one batch — not the
|
||||
/// whole home directory.
|
||||
fn scan_local_once(db: &Db, indexing: &AtomicBool, mountpoint: &Path) {
|
||||
let dirs = match AppDirs::new() {
|
||||
Ok(d) => d,
|
||||
Err(e) => {
|
||||
warn!(error = %e, "local index: cannot resolve app dirs");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let Some(home) = dirs.home_dir() else {
|
||||
warn!("local index: cannot resolve home directory");
|
||||
return;
|
||||
};
|
||||
let generation = match db.local_begin_scan() {
|
||||
Ok(g) => g,
|
||||
Err(e) => {
|
||||
warn!(error = %e, "local index: cannot open scan generation");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let excludes = localindex::default_excludes(mountpoint, &dirs.state_dir(), &dirs.cache_dir());
|
||||
indexing.store(true, Ordering::Relaxed);
|
||||
let started = Instant::now();
|
||||
|
||||
let walked = localindex::scan(&[home], &excludes, |batch| {
|
||||
if let Err(e) = db.local_upsert_batch(generation, &batch) {
|
||||
warn!(error = %e, "local index: batch write failed");
|
||||
}
|
||||
});
|
||||
|
||||
// Prune what this scan did not see and rebuild the FTS index over the rest,
|
||||
// even if some batches failed — a partial index still beats none.
|
||||
match db.local_finish_scan(generation, now_secs()) {
|
||||
Ok(indexed) => info!(
|
||||
walked,
|
||||
indexed,
|
||||
elapsed_ms = started.elapsed().as_millis() as u64,
|
||||
"local index rebuilt"
|
||||
),
|
||||
Err(e) => warn!(error = %e, "local index: finish failed"),
|
||||
}
|
||||
indexing.store(false, Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Turn a CLI-supplied path into a mountpoint-relative path. An absolute path
|
||||
/// must live under `mountpoint`; a relative path is taken as already relative to
|
||||
/// the mount root.
|
||||
@@ -2436,6 +2537,13 @@ fn handle_control_conn(core: &Core, username: &str, mountpoint: &Path, stream: U
|
||||
Ok(hits) => CtlResponse::SearchResults { hits },
|
||||
Err(e) => CtlResponse::Error { message: e },
|
||||
},
|
||||
Ok(CtlRequest::SearchLocal { query, limit }) => match core.search_local(&query, limit) {
|
||||
Ok(hits) => CtlResponse::LocalResults {
|
||||
hits,
|
||||
indexing: core.indexing.load(Ordering::Relaxed),
|
||||
},
|
||||
Err(e) => CtlResponse::Error { message: e },
|
||||
},
|
||||
Ok(CtlRequest::Rename { path, new_name }) => match rel_to_mount(mountpoint, &path) {
|
||||
Ok(rel) => match core.rename(&rel, &new_name) {
|
||||
Ok(name) => CtlResponse::Ok {
|
||||
@@ -2629,8 +2737,20 @@ pub fn mount(
|
||||
db,
|
||||
timeline: Arc::new(Mutex::new(None)),
|
||||
transfers: TransferRegistry::new(),
|
||||
indexing: Arc::new(AtomicBool::new(false)),
|
||||
};
|
||||
|
||||
// Keep the launcher prompt's "This computer" index fresh. Its own thread:
|
||||
// the walk is I/O-heavy and must never sit in front of a FUSE callback.
|
||||
{
|
||||
let db = core.db.clone();
|
||||
let indexing = core.indexing.clone();
|
||||
let mountpoint = mountpoint.to_path_buf();
|
||||
std::thread::Builder::new()
|
||||
.name("pdfs-localindex".into())
|
||||
.spawn(move || run_local_index(db, indexing, mountpoint))?;
|
||||
}
|
||||
|
||||
let mut config = Config::default();
|
||||
config.mount_options = vec![
|
||||
MountOption::FSName("protondrive".to_string()),
|
||||
|
||||
+1049
-944
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user