Bugfix: Prevent OOM crash from unbounded indexing buffers

When I first ran it, the app consumed 500+ GB of RAM and took down my entire system. Oops :D

The trigger: toggling Full Disk Access caused FSEvents to replay millions of events through a pipeline with zero backpressure Then we ralized every buffer was unbounded. Oops.

Solution:
- Bound the watcher→event loop channel (100K), writer channel (100K sync_channel with backpressure), reconciler buffer (500K), affected_paths (50K), and pending_rescans (1K). Overflow triggers full rescan — safe because the index is a disposable cache.
- Abort journal replay after 1M events and fall back to full scan, catching the FDA-toggle scenario early.
- Add a memory watchdog (`mach_task_info`, macOS only) that warns at 8 GB and stops all indexing at 16 GB, emitting a frontend event. No-op stub on Linux.
- Worst-case peak memory is now ~350 MB during a full scan with heavy concurrent filesystem activity, down from effectively unlimited.

Author: vdavid

apps/desktop/src-tauri/src/file_system/volume/local_posix.rs

@@ -287,7 +287,7 @@ impl VolumeWatcher for LocalPosixWatcher {
         &self,
         root: &Path,
         since_when: u64,
-        event_sender: mpsc::UnboundedSender<FsChangeEvent>,
+        event_sender: mpsc::Sender<FsChangeEvent>,
     ) -> Result<DriveWatcher, WatcherError> {
         DriveWatcher::start(root, since_when, event_sender)
     }

apps/desktop/src-tauri/src/file_system/volume/mod.rs

@@ -142,7 +142,7 @@ pub trait VolumeWatcher: Send + Sync {
         &self,
         root: &Path,
         since_when: u64,
-        event_sender: mpsc::UnboundedSender<FsChangeEvent>,
+        event_sender: mpsc::Sender<FsChangeEvent>,
     ) -> Result<DriveWatcher, WatcherError>;
 }
 

apps/desktop/src-tauri/src/indexing/CLAUDE.md

@@ -11,12 +11,13 @@ Full design: `docs/specs/drive-indexing/plan.md`
 - **mod.rs** -- Public API: `init()`, `start_indexing()`, `stop_indexing()`, `clear_index()`, `enrich_entries_with_index()`. `IndexManager` coordinates all subsystems, owns a `PathResolver` (LRU-cached path→ID mapping) for IPC commands. Global read-only store for enrichment. Enrichment uses an integer-keyed fast path: resolve parent dir once → batch-fetch child dir stats by ID → match by name. Falls back to individual path resolution for edge cases.
 - **store.rs** -- SQLite schema v2 (integer-keyed entries, dir_stats by entry_id, meta), platform_case collation, read queries, DB open/migrate. Schema version check: mismatch triggers drop+rebuild. Both path-keyed (backward compat) and integer-keyed APIs.
 - **path_resolver.rs** -- `PathResolver`: resolves filesystem paths to integer entry IDs via component-by-component walk with full-path LRU cache (50K entries). Case-aware `CacheKey` on macOS (NFD + case fold). Prefix-based invalidation for deletes/renames.
-- **writer.rs** -- Single writer thread, owns the write connection, processes `WriteMessage` channel (unbounded mpsc). Priority: `UpdateDirStats` before `InsertEntries`. `Flush` variant + async `flush()` method let callers wait for all prior writes to commit. Has both integer-keyed variants (`InsertEntriesV2`, `UpsertEntryV2`, `DeleteEntryById`, `DeleteSubtreeById`, `PropagateDeltaById`) and path-keyed backward-compat variants. The integer-keyed delete/subtree-delete handlers auto-propagate negative deltas via the `parent_id` chain (same pattern as the path-keyed variants). `propagate_delta_by_id` walks the parent chain using `get_parent_id` lookups.
+- **memory_watchdog.rs** -- Background task monitoring resident memory via `mach_task_info` (macOS). Warns at 8 GB, stops indexing at 16 GB, emits `index-memory-warning` event to frontend. No-op stub on non-macOS. Started from `start_indexing()`.
+- **writer.rs** -- Single writer thread, owns the write connection, processes `WriteMessage` channel (bounded `sync_channel`, 100K capacity, backpressure via blocking). Priority: `UpdateDirStats` before `InsertEntries`. `Flush` variant + async `flush()` method let callers wait for all prior writes to commit. Has both integer-keyed variants (`InsertEntriesV2`, `UpsertEntryV2`, `DeleteEntryById`, `DeleteSubtreeById`, `PropagateDeltaById`) and path-keyed backward-compat variants. The integer-keyed delete/subtree-delete handlers auto-propagate negative deltas via the `parent_id` chain (same pattern as the path-keyed variants). `propagate_delta_by_id` walks the parent chain using `get_parent_id` lookups.
 - **scanner.rs** -- jwalk-based parallel directory walker. `scan_volume()` for full scan, `scan_subtree()` for micro-scans. Uses `ScanContext` (from store.rs) to assign integer IDs and parent IDs during the walk: maintains a `HashMap<PathBuf, i64>` mapping directory paths to assigned IDs. The scan root is mapped to `ROOT_ID` (1). Sends `InsertEntriesV2(Vec<EntryRow>)` batches to the writer. Platform-specific exclusion filters (macOS system paths, Linux virtual filesystems). Physical sizes (`st_blocks * 512`).
 - **micro_scan.rs** -- `MicroScanManager`: bounded task pool (default 3 concurrent), priority queue (`UserSelected` > `CurrentDir`), deduplication, cancellation. Skips after full scan completes.
 - **aggregator.rs** -- Dir stats computation. Bottom-up after full scan (O(N) single pass), per-subtree after micro-scan, incremental delta propagation up ancestor chain for watcher events.
 - **watcher.rs** -- Drive-level filesystem watcher. macOS: FSEvents via `cmdr-fsevent-stream` with event IDs and `sinceWhen` replay. Linux: `notify` crate (inotify backend) with recursive watching and synthetic event counter. Other platforms: stub. `supports_event_replay()` lets callers branch on whether journal replay is available.
-- **reconciler.rs** -- Buffers FSEvents during scan, replays after scan completes using event IDs to skip stale events. Processes live events for file creates/removes/modifies using integer-keyed write messages (`UpsertEntryV2`, `DeleteEntryById`, `DeleteSubtreeById`, `PropagateDeltaById`). Resolves filesystem paths to entry IDs via `store::resolve_path()` using a read connection passed by callers. Key functions (`process_fs_event`, `emit_dir_updated`) are `pub(super)` so `mod.rs` can call them directly during cold-start replay.
+- **reconciler.rs** -- Buffers FSEvents during scan (capped at 500K events; overflow sets `buffer_overflow` flag forcing full rescan), replays after scan completes using event IDs to skip stale events. Processes live events for file creates/removes/modifies using integer-keyed write messages (`UpsertEntryV2`, `DeleteEntryById`, `DeleteSubtreeById`, `PropagateDeltaById`). Resolves filesystem paths to entry IDs via `store::resolve_path()` using a read connection passed by callers. Key functions (`process_fs_event`, `emit_dir_updated`) are `pub(super)` so `mod.rs` can call them directly during cold-start replay.
 - **firmlinks.rs** -- Parses `/usr/share/firmlinks`, builds prefix map, normalizes paths. Converts `/System/Volumes/Data/Users/foo` to `/Users/foo`.
 - **verifier.rs** -- Placeholder for per-navigation background readdir diff (future milestone).
 
@@ -57,7 +58,7 @@ Enrichment (every get_file_range call):
 
 ### Single-writer architecture
 
-All writes go through a dedicated `std::thread` via an unbounded mpsc channel. The writer thread owns the write connection and processes messages in order, prioritizing `UpdateDirStats` over `InsertEntries` for responsive micro-scan results.
+All writes go through a dedicated `std::thread` via a bounded `sync_channel` (100K capacity). When the channel is full, senders block (backpressure). The writer thread owns the write connection and processes messages in order, prioritizing `UpdateDirStats` over `InsertEntries` for responsive micro-scan results.
 
 Reads happen on separate WAL connections (any thread). The global read-only store (`GLOBAL_INDEX_STORE`) provides enrichment without passing `AppHandle` through the listing pipeline.
 
@@ -107,6 +108,8 @@ Key test files are alongside each module (test functions within `#[cfg(test)]` b
 
 **Subtree aggregation uses scoped queries**: `scoped_get_children_stats_by_id` and `scoped_get_child_dir_ids` in `aggregator.rs` use recursive CTEs scoped to the target subtree, not full-table scans. This keeps subtree aggregation O(subtree_size) regardless of total DB size.
 
+**Bounded buffers prevent OOM**: All buffers have capacity limits. Reconciler buffer: 500K events (overflow triggers full rescan). Writer channel: 100K messages (bounded `sync_channel`, backpressure). Replay `affected_paths`: 50K entries (overflow emits full refresh). Replay `pending_rescans`: 1K entries (overflow triggers full rescan). Replay event count: 1M events max (overflow falls back to full scan). Memory watchdog: warns at 8 GB, stops indexing at 16 GB. The index is a disposable cache, so dropping events and rescanning is always safe.
+
 **Disposable cache pattern**: The index DB is a cache, not a source of truth. Any corruption or error triggers delete+rebuild. No user-facing errors for DB issues.
 
 **cmdr-fsevent-stream fork (macOS only)**: Our fork of `fsevent-stream` (v0.3.0) provides direct access to FSEvents event IDs, `sinceWhen` replay, and `MustScanSubDirs` flags. Only used on macOS. On Linux, the `notify` crate (inotify backend) provides recursive directory watching with `RecursiveMode::Recursive`.

apps/desktop/src-tauri/src/indexing/memory_watchdog.rs

@@ -0,0 +1,163 @@
+//! Memory watchdog: monitors the app's resident memory and takes action
+//! at safety thresholds to prevent unbounded memory growth.
+//!
+//! - 8 GB: logs a warning.
+//! - 16 GB: stops all indexing and emits a user-visible event.
+//!
+//! On non-macOS platforms this is a no-op stub (platform memory queries
+//! differ and can be added later).
+
+/// 8 GB in bytes.
+#[cfg(target_os = "macos")]
+const WARN_THRESHOLD: u64 = 8 * 1024 * 1024 * 1024;
+
+/// 16 GB in bytes.
+#[cfg(target_os = "macos")]
+const STOP_THRESHOLD: u64 = 16 * 1024 * 1024 * 1024;
+
+/// How often the watchdog checks memory (seconds).
+#[cfg(target_os = "macos")]
+const CHECK_INTERVAL_SECS: u64 = 5;
+
+/// Start the memory watchdog as a fire-and-forget background task.
+///
+/// On macOS, spawns a task that checks resident memory every 5 seconds
+/// using `mach_task_info`. Runs until the app is stopped or indexing is
+/// halted due to excessive memory usage. On other platforms, this is a no-op.
+#[cfg(target_os = "macos")]
+pub fn start(app: tauri::AppHandle) {
+    tauri::async_runtime::spawn(async move {
+        run_watchdog(app).await;
+    });
+}
+
+#[cfg(not(target_os = "macos"))]
+pub fn start(_app: tauri::AppHandle) {
+    // No-op on non-macOS platforms
+}
+
+#[cfg(target_os = "macos")]
+async fn run_watchdog(app: tauri::AppHandle) {
+    use std::time::Duration;
+
+    let mut interval = tokio::time::interval(Duration::from_secs(CHECK_INTERVAL_SECS));
+    let mut warned = false;
+
+    loop {
+        interval.tick().await;
+
+        let resident_bytes = match get_resident_memory() {
+            Some(b) => b,
+            None => continue,
+        };
+
+        if resident_bytes >= STOP_THRESHOLD {
+            log::error!(
+                "Memory watchdog: resident memory is {} GB, exceeding {} GB safety limit. \
+                 Stopping all indexing to prevent a system crash.",
+                resident_bytes / (1024 * 1024 * 1024),
+                STOP_THRESHOLD / (1024 * 1024 * 1024),
+            );
+
+            // Emit user-visible event
+            use tauri::Emitter;
+            let _ = app.emit(
+                "index-memory-warning",
+                serde_json::json!({
+                    "resident_gb": resident_bytes / (1024 * 1024 * 1024),
+                    "action": "stopped_indexing",
+                }),
+            );
+
+            // Stop indexing
+            if let Err(e) = super::stop_indexing(&app) {
+                log::error!("Memory watchdog: stop_indexing failed: {e}");
+            }
+            return;
+        }
+
+        if resident_bytes >= WARN_THRESHOLD && !warned {
+            warned = true;
+            log::warn!(
+                "Memory watchdog: resident memory is {} MB ({} GB threshold approaching). \
+                 Indexing continues but the system may be under pressure.",
+                resident_bytes / (1024 * 1024),
+                STOP_THRESHOLD / (1024 * 1024 * 1024),
+            );
+        }
+
+        // Reset warning flag if memory drops back below the threshold
+        if resident_bytes < WARN_THRESHOLD && warned {
+            warned = false;
+        }
+    }
+}
+
+/// Query the current task's resident memory using `mach_task_basic_info`.
+///
+/// Uses raw FFI because the `libc` crate doesn't expose `MACH_TASK_BASIC_INFO`.
+#[cfg(target_os = "macos")]
+fn get_resident_memory() -> Option<u64> {
+    // Mach task info constants (from <mach/task_info.h>)
+    const MACH_TASK_BASIC_INFO: u32 = 20;
+
+    #[repr(C)]
+    struct MachTaskBasicInfo {
+        virtual_size: u64,
+        resident_size: u64,
+        resident_size_max: u64,
+        user_time_seconds: i32,
+        user_time_microseconds: i32,
+        system_time_seconds: i32,
+        system_time_microseconds: i32,
+        policy: i32,
+        suspend_count: i32,
+    }
+
+    let info_count = (size_of::<MachTaskBasicInfo>() / size_of::<libc::c_int>()) as u32;
+
+    #[allow(deprecated, reason = "mach_task_self is deprecated in libc but works fine")]
+    unsafe {
+        let mut info: MachTaskBasicInfo = std::mem::zeroed();
+        let mut count = info_count;
+        let result = libc::task_info(
+            libc::mach_task_self(),
+            MACH_TASK_BASIC_INFO,
+            &mut info as *mut MachTaskBasicInfo as *mut i32,
+            &mut count,
+        );
+        if result == 0 {
+            Some(info.resident_size)
+        } else {
+            log::debug!("Memory watchdog: task_info failed with code {result}");
+            None
+        }
+    }
+}
+
+// ── Tests ────────────────────────────────────────────────────────────
+
+#[cfg(test)]
+mod tests {
+    #[cfg(target_os = "macos")]
+    use super::*;
+
+    #[cfg(target_os = "macos")]
+    #[test]
+    fn get_resident_memory_returns_positive_value() {
+        let mem = get_resident_memory();
+        assert!(mem.is_some(), "should be able to query resident memory");
+        assert!(mem.unwrap() > 0, "resident memory should be positive");
+    }
+
+    #[cfg(target_os = "macos")]
+    #[test]
+    fn thresholds_are_ordered() {
+        const {
+            assert!(
+                WARN_THRESHOLD < STOP_THRESHOLD,
+                "warn threshold must be below stop threshold"
+            )
+        };
+    }
+}