JezzWTF/phokus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(watcher): adaptive filesystem watchdog with zero CPU when idle

Browse Source 
Monitors all registered folders using OS-native events (ReadDirectoryChangesW
on Windows) so new, modified, and deleted files are reflected in the gallery
automatically without manual reindexing.

Key design points:
- Adaptive blocking: recv() when no events pending (zero CPU), switches to
  recv_timeout(earliest_deadline) only when debounce timers are running — wakes
  exactly when the soonest event is ready, no busy-polling
- 500 ms per-path debounce coalesces rapid OS event bursts into one action
- Change detection preserved: build_record skips upsert if file_size + mtime
  unchanged, preventing spurious thumbnail/embedding re-queues and avoiding
  clobbering of existing metadata like thumbnail_path
- Access events (reads) filtered out; only Create/Modify/Remove trigger work
- Deletion path: emits watcher-deleted event with image IDs; frontend removes
  those images from state and clears selectedImage if it was deleted

WatcherHandle stored in app state; add_folder / remove_folder / update_folder_path
commands keep the watched path set in sync with the DB.
This commit is contained in:
LyAhn
2026-06-08 06:50:01 +01:00
parent 9ee5b08c93
commit ae9e806e61
7 changed files with 396 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/Cargo.lock
Generated
+91 -1
View File
@@ -1750,6 +1750,15 @@ dependencies = [
"winapi",
]
[[package]]
name = "fsevent-sys"
version = "4.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2"
dependencies = [
"libc",
]
[[package]]
name = "futf"
version = "0.1.5"
@@ -2932,6 +2941,26 @@ dependencies = [
"cfb",
]
[[package]]
name = "inotify"
version = "0.9.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8069d3ec154eb856955c1c0fbffefbf5f3c40a104ec912d4797314c1801abff"
dependencies = [
"bitflags 1.3.2",
"inotify-sys",
"libc",
]
[[package]]
name = "inotify-sys"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e05c02b5e89bff3b946cedeca278abc628fe811e604f027c45a8aa3cf793d0eb"
dependencies = [
"libc",
]
[[package]]
name = "ipnet"
version = "2.12.0"
@@ -3133,6 +3162,26 @@ dependencies = [
"unicode-segmentation",
]
[[package]]
name = "kqueue"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "273c0752728918e0ac4976f2b275b6fefb9ecd400585dec929419f3844cd87b5"
dependencies = [
"kqueue-sys",
"libc",
]
[[package]]
name = "kqueue-sys"
version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087"
dependencies = [
"bitflags 2.11.0",
"libc",
]
[[package]]
name = "kuchikiki"
version = "0.8.8-speedreader"
@@ -3438,6 +3487,18 @@ dependencies = [
"simd-adler32",
]
[[package]]
name = "mio"
version = "0.8.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c"
dependencies = [
"libc",
"log",
"wasi 0.11.1+wasi-snapshot-preview1",
"windows-sys 0.48.0",
]
[[package]]
name = "mio"
version = "1.2.0"
@@ -3621,6 +3682,25 @@ dependencies = [
"minimal-lexical",
]
[[package]]
name = "notify"
version = "6.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6205bd8bb1e454ad2e27422015fb5e4f2bcc7e08fa8f27058670d208324a4d2d"
dependencies = [
"bitflags 2.11.0",
"crossbeam-channel",
"filetime",
"fsevent-sys",
"inotify",
"kqueue",
"libc",
"log",
"mio 0.8.11",
"walkdir",
"windows-sys 0.48.0",
]
[[package]]
name = "notify-rust"
version = "4.17.0"
@@ -4308,6 +4388,7 @@ dependencies = [
"kamadak-exif",
"log",
"memmap2",
"notify",
"ort",
"r2d2",
"r2d2_sqlite",
@@ -6392,7 +6473,7 @@ checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd"
dependencies = [
"bytes",
"libc",
"mio",
"mio 1.2.0",
"parking_lot",
"pin-project-lite",
"signal-hook-registry",
@@ -7519,6 +7600,15 @@ dependencies = [
"windows-targets 0.42.2",
]
[[package]]
name = "windows-sys"
version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
dependencies = [
"windows-targets 0.48.5",
]
[[package]]
name = "windows-sys"
version = "0.52.0"
src-tauri/Cargo.toml
+1
View File
@@ -51,6 +51,7 @@ ureq = { version = "3.3.0", default-features = false, features = ["native-tls"]
zip = { version = "4.6.1", default-features = false, features = ["deflate"] }
csv = "1"
kamadak-exif = "0.5"
notify = "6"
tauri-plugin-notification = "2"
# ── Dev-mode performance ────────────────────────────────────────────────────
src-tauri/src/commands.rs
+24 -5
View File
@@ -5,7 +5,7 @@ use crate::captioner::{
use crate::db::{self, DbPool, ExploreTagEntry, Folder, FolderJobProgress, ImageRecord, ImageTag};
use crate::embedder;
use crate::hnsw_index;
use crate::indexer;
use crate::indexer::{self, WatcherHandle};
use crate::tagger::{self, TaggerAcceleration, TaggerModelStatus, TaggerRuntimeProbe};
use crate::vector;
use serde::{Deserialize, Serialize};
@@ -202,6 +202,7 @@ pub struct GetImagesByIdsParams {
pub async fn add_folder(
app: AppHandle,
db: State<'_, DbState>,
watcher: State<'_, WatcherHandle>,
path: String,
) -> Result<Folder, String> {
let folder_path = PathBuf::from(&path);
@@ -230,6 +231,7 @@ pub async fn add_folder(
.find(|f| f.id == folder_id)
.ok_or("Folder not found after insert")?;
watcher.add_folder(folder_path.clone(), folder_id);
indexer::index_folder(app, db.inner().clone(), folder_id, folder_path);
Ok(folder)
@@ -250,9 +252,23 @@ pub async fn get_background_job_progress(
}
#[tauri::command]
pub async fn remove_folder(db: State<'_, DbState>, folder_id: i64) -> Result<(), String> {
pub async fn remove_folder(
db: State<'_, DbState>,
watcher: State<'_, WatcherHandle>,
folder_id: i64,
) -> Result<(), String> {
let conn = db.get().map_err(|e| e.to_string())?;
db::delete_folder(&conn, folder_id).map_err(|e| e.to_string())
// Capture the path before deletion so we can unregister the watcher.
let folder_path = db::get_folders(&conn)
.map_err(|e| e.to_string())?
.into_iter()
.find(|f| f.id == folder_id)
.map(|f| PathBuf::from(f.path));
db::delete_folder(&conn, folder_id).map_err(|e| e.to_string())?;
if let Some(path) = folder_path {
watcher.remove_folder(&path);
}
Ok(())
}
#[tauri::command]
@@ -352,6 +368,7 @@ pub async fn rename_folder(
pub async fn update_folder_path(
app: AppHandle,
db: State<'_, DbState>,
watcher: State<'_, WatcherHandle>,
folder_id: i64,
new_path: String,
) -> Result<(), String> {
@@ -363,7 +380,7 @@ pub async fn update_folder_path(
.file_name()
.map(|n| n.to_string_lossy().to_string())
.unwrap_or_else(|| new_path.clone());
{
let old_path = {
let conn = db.get().map_err(|e| e.to_string())?;
// Fetch the old path before updating so image paths can be rewritten.
let old_path = db::get_folders(&conn)
@@ -374,7 +391,9 @@ pub async fn update_folder_path(
.ok_or("Folder not found")?;
db::update_folder_path(&conn, folder_id, &old_path, &new_path, &new_name)
.map_err(|e| e.to_string())?;
}
old_path
};
watcher.update_folder(&PathBuf::from(old_path), new_path_buf.clone(), folder_id);
indexer::index_folder(app, db.inner().clone(), folder_id, new_path_buf);
Ok(())
}
src-tauri/src/db.rs
+38
View File
@@ -1352,6 +1352,44 @@ pub fn update_image_details(
.map_err(Into::into)
}
/// Look up the lightweight indexed-media entry for a single path.
/// Used by the filesystem watcher to run change-detection before upserting.
pub fn get_indexed_entry_by_path(conn: &Connection, path: &str) -> Result<Option<IndexedMediaEntry>> {
let result = conn.query_row(
"SELECT id, path, modified_at, file_size, media_kind FROM images WHERE path = ?1",
[path],
|row| {
Ok(IndexedMediaEntry {
id: row.get(0)?,
path: row.get(1)?,
modified_at: row.get(2)?,
file_size: row.get(3)?,
media_kind: row.get(4)?,
})
},
);
match result {
Ok(entry) => Ok(Some(entry)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
}
}
/// Look up just the image id for a path. Used by the filesystem watcher
/// to find the DB row to delete when a file is removed from disk.
pub fn get_image_id_by_path(conn: &Connection, path: &str) -> Result<Option<i64>> {
let result = conn.query_row(
"SELECT id FROM images WHERE path = ?1",
[path],
|row| row.get(0),
);
match result {
Ok(id) => Ok(Some(id)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
}
}
pub fn get_image_by_id(conn: &Connection, image_id: i64) -> Result<ImageRecord> {
conn.query_row(
"SELECT id, folder_id, path, filename, thumbnail_path, width, height, file_size, created_at, modified_at, taken_at, mime_type,
src-tauri/src/indexer.rs
+220 -1
View File
@@ -9,9 +9,10 @@ use crate::vector;
use anyhow::Result;
use rayon::prelude::*;
use serde::Serialize;
use notify::{RecommendedWatcher, RecursiveMode, Watcher};
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::{Mutex, OnceLock};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant};
use tauri::{AppHandle, Emitter};
use walkdir::WalkDir;
@@ -1282,3 +1283,221 @@ fn mime_for_ext(ext: &str) -> &'static str {
_ => "image/jpeg",
}
}
// ── Filesystem watcher ────────────────────────────────────────────────────────
/// How long to wait after the last event for a path before processing it.
/// Absorbs bursts of OS events that accompany a single logical file write.
const WATCHER_DEBOUNCE: Duration = Duration::from_millis(500);
struct WatcherInner {
watcher: Mutex<RecommendedWatcher>,
/// Maps each watched folder root → its folder_id in the DB.
folder_map: Arc<Mutex<HashMap<PathBuf, i64>>>,
}
/// Shared handle that lets command handlers register and deregister watched
/// directories without touching the debounce thread directly.
#[derive(Clone)]
pub struct WatcherHandle {
inner: Arc<WatcherInner>,
}
impl WatcherHandle {
pub fn add_folder(&self, path: PathBuf, folder_id: i64) {
{
let mut w = self.inner.watcher.lock().unwrap();
if path.is_dir() {
if let Err(e) = w.watch(&path, RecursiveMode::Recursive) {
eprintln!("Watcher: failed to watch {:?}: {}", path, e);
}
}
}
self.inner.folder_map.lock().unwrap().insert(path, folder_id);
}
pub fn remove_folder(&self, path: &Path) {
{
let mut w = self.inner.watcher.lock().unwrap();
let _ = w.unwatch(path);
}
self.inner.folder_map.lock().unwrap().remove(path);
}
pub fn update_folder(&self, old_path: &Path, new_path: PathBuf, folder_id: i64) {
{
let mut w = self.inner.watcher.lock().unwrap();
let _ = w.unwatch(old_path);
if new_path.is_dir() {
if let Err(e) = w.watch(&new_path, RecursiveMode::Recursive) {
eprintln!("Watcher: failed to watch {:?}: {}", new_path, e);
}
}
}
let mut map = self.inner.folder_map.lock().unwrap();
map.remove(old_path);
map.insert(new_path, folder_id);
}
}
/// Start the filesystem watcher. Watches all folders currently in the DB and
/// returns a handle that command handlers can use to add/remove watched paths.
///
/// The debounce loop uses adaptive blocking:
/// - `recv()` when no events are pending — zero CPU
/// - `recv_timeout(earliest_deadline)` when events are pending — wakes exactly
/// when the soonest debounce window expires, no busy-polling
pub fn start_watcher(app: AppHandle, pool: DbPool) -> WatcherHandle {
let (tx, rx) = std::sync::mpsc::channel::<notify::Result<notify::Event>>();
let raw_watcher = notify::recommended_watcher(move |result| {
let _ = tx.send(result);
})
.expect("Failed to create filesystem watcher");
// Seed the folder map from the DB so existing folders are watched on startup.
let folder_map: Arc<Mutex<HashMap<PathBuf, i64>>> = Arc::new(Mutex::new(HashMap::new()));
{
let conn = pool.get().expect("Watcher: failed to get DB connection for init");
let folders = db::get_folders(&conn).unwrap_or_default();
let mut map = folder_map.lock().unwrap();
for f in folders {
map.insert(PathBuf::from(f.path), f.id);
}
}
// Register each known folder with the OS watcher.
let raw_watcher = {
let mut w = raw_watcher;
let map = folder_map.lock().unwrap();
for path in map.keys() {
if path.is_dir() {
if let Err(e) = w.watch(path, RecursiveMode::Recursive) {
eprintln!("Watcher: failed to watch {:?}: {}", path, e);
}
}
}
w
};
let handle = WatcherHandle {
inner: Arc::new(WatcherInner {
watcher: Mutex::new(raw_watcher),
folder_map: Arc::clone(&folder_map),
}),
};
// Spawn the debounce loop on its own thread.
let folder_map_thread = Arc::clone(&folder_map);
std::thread::spawn(move || {
// path → deadline: the earliest instant at which this path should be processed.
let mut pending: HashMap<PathBuf, Instant> = HashMap::new();
loop {
// Adaptive blocking: block forever when idle, wake at the earliest
// deadline when events are queued.
let received = if pending.is_empty() {
match rx.recv() {
Ok(e) => Some(e),
Err(_) => break, // sender dropped — app is shutting down
}
} else {
let earliest = pending.values().copied().min().unwrap(); // safe: non-empty
let timeout = earliest.saturating_duration_since(Instant::now());
match rx.recv_timeout(timeout) {
Ok(e) => Some(e),
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => None,
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
}
};
let now = Instant::now();
// Absorb incoming event — coalesces rapid writes into one deadline.
if let Some(Ok(event)) = received {
use notify::EventKind;
// Skip pure access events (reads); they never change file content.
if !matches!(event.kind, EventKind::Access(_)) {
for path in event.paths {
if is_supported_media(&path) {
pending.insert(path, now + WATCHER_DEBOUNCE);
}
}
}
}
// Process all paths whose debounce window has expired.
let ready: Vec<PathBuf> = pending
.iter()
.filter(|(_, &deadline)| deadline <= now)
.map(|(p, _)| p.clone())
.collect();
for path in ready {
pending.remove(&path);
process_watcher_path(&app, &pool, &folder_map_thread, &path);
}
}
});
handle
}
/// Decide what to do with a path whose debounce window just expired.
/// If the file exists → upsert (with change-detection to avoid clobbering
/// metadata like thumbnail_path). If the file is gone → delete from DB.
fn process_watcher_path(
app: &AppHandle,
pool: &DbPool,
folder_map: &Arc<Mutex<HashMap<PathBuf, i64>>>,
path: &Path,
) {
// Determine which registered folder owns this file.
let folder_id = {
let map = folder_map.lock().unwrap();
map.iter()
.find(|(folder_path, _)| path.starts_with(folder_path.as_path()))
.map(|(_, &id)| id)
};
let Some(folder_id) = folder_id else { return };
let conn = match pool.get() {
Ok(c) => c,
Err(e) => {
eprintln!("Watcher: DB pool error: {}", e);
return;
}
};
if path.exists() {
// File still on disk — upsert if content changed.
let path_str = path.to_string_lossy();
let existing = db::get_indexed_entry_by_path(&conn, &path_str).unwrap_or(None);
let Some(record) = build_record(path, folder_id, existing.as_ref()) else {
return; // file unchanged (same size + mtime) — nothing to do
};
drop(conn); // commit_batch acquires its own connection from the pool
match commit_batch(pool, &[record]) {
Ok(committed) if !committed.is_empty() => {
emit_images(app, &IndexedImagesBatch { folder_id, images: committed });
emit_folder_job_progress(app, pool, &[folder_id], false);
}
Ok(_) => {}
Err(e) => eprintln!("Watcher: commit error for {:?}: {}", path, e),
}
} else {
// File removed from disk — clean up DB row.
let path_str = path.to_string_lossy();
match db::get_image_id_by_path(&conn, &path_str) {
Ok(Some(image_id)) => {
if db::delete_images_by_ids(&conn, &[image_id]).is_ok() {
db::update_folder_count(&conn, folder_id).ok();
let _ = app.emit("watcher-deleted", vec![image_id]);
}
}
Ok(None) => {} // never indexed or already removed
Err(e) => eprintln!("Watcher: lookup error for {:?}: {}", path, e),
}
}
}
src-tauri/src/lib.rs
+3
View File
@@ -74,8 +74,11 @@ pub fn run() {
// indexer::start_caption_worker(app.handle().clone(), pool.clone(), app_dir.clone());
indexer::start_tagging_worker(app.handle().clone(), pool.clone(), app_dir.clone());
let watcher_handle = indexer::start_watcher(app.handle().clone(), pool.clone());
app.manage(pool);
app.manage(media_tools);
app.manage(watcher_handle);
Ok(())
})