Listing: Remove last 5 block_on calls from file_system module

Zero `block_on` remaining in the entire `file_system/` production code.

- `streaming.rs`: replaced `std::thread::spawn` + `block_on` with `tokio::spawn` + async polling loop
- `operations.rs`: `list_directory_start_with_volume` now async, direct `.await`
- `caching.rs`: `notify_full_refresh` now async, spawned via `tokio::spawn`
- `watcher.rs`: `handle_directory_change` now async, spawned from FSEvents callbacks
- `smb_watcher.rs`: `stat_via_volume` and `process_event_batch` now async, direct `.await`
- Tests updated: removed `spawn_blocking` wrappers, direct `.await`

Author: vdavid

apps/desktop/src-tauri/src/commands/file_system/listing.rs

@@ -15,9 +15,7 @@ use crate::file_system::{
 use std::path::{Path, PathBuf};
 use tokio::time::Duration;
 
-use crate::commands::util::{
-    IpcError, TimedOut, blocking_result_with_timeout, blocking_with_timeout, blocking_with_timeout_flag,
-};
+use crate::commands::util::{IpcError, TimedOut, blocking_with_timeout};
 use crate::file_system::validation::{MAX_NAME_BYTES, MAX_PATH_BYTES};
 
 use super::expand_tilde;
@@ -75,11 +73,19 @@ pub async fn list_directory_start(
     let expanded_path = expand_tilde(&path);
     let path_buf = PathBuf::from(&expanded_path);
     let dir_sort_mode = directory_sort_mode.unwrap_or_default();
-    blocking_result_with_timeout(Duration::from_secs(2), move || {
-        ops_list_directory_start_with_volume("root", &path_buf, include_hidden, sort_by, sort_order, dir_sort_mode)
-            .map_err(|e| format!("Failed to start directory listing '{}': {}", path, e))
-    })
+    match tokio::time::timeout(
+        Duration::from_secs(2),
+        ops_list_directory_start_with_volume("root", &path_buf, include_hidden, sort_by, sort_order, dir_sort_mode),
+    )
     .await
+    {
+        Ok(Ok(result)) => Ok(result),
+        Ok(Err(e)) => Err(IpcError::from_err(format!(
+            "Failed to start directory listing '{}': {}",
+            path, e
+        ))),
+        Err(_) => Err(IpcError::timeout()),
+    }
 }
 
 /// Returns immediately; reads in background.
@@ -220,10 +226,12 @@ pub fn list_directory_end(listing_id: String) {
 /// Used after write operations (move) when the file watcher may not fire promptly.
 #[tauri::command]
 pub async fn refresh_listing(listing_id: String) -> TimedOut<()> {
-    blocking_with_timeout_flag(Duration::from_secs(2), (), move || {
-        crate::file_system::watcher::handle_directory_change(&listing_id);
+    let timed_out = tokio::time::timeout(Duration::from_secs(2), async {
+        crate::file_system::watcher::handle_directory_change(&listing_id).await;
     })
     .await
+    .is_err();
+    TimedOut { data: (), timed_out }
 }
 
 /// Returns total file/dir counts and sizes, plus selection stats if `selected_indices` is given.

apps/desktop/src-tauri/src/file_system/listing/CLAUDE.md

@@ -61,7 +61,7 @@ Frontend                          Backend
 ## Key decisions
 
 **Decision**: Streaming with background task, not chunked IPC
-**Why**: Chunked approach required multiple IPC calls, complex state tracking. Streaming spawns `tokio::task::spawn_blocking()`, emits events. Frontend stays responsive—Tab works, ESC cancels.
+**Why**: Chunked approach required multiple IPC calls, complex state tracking. Streaming spawns a `tokio::spawn` async task, emits events. Frontend stays responsive—Tab works, ESC cancels via `tokio::select!`-style polling.
 
 **Decision**: Cancellation via `AtomicBool` checked per-entry
 **Why**: Network folders iterate slowly (seconds per entry). Checking on each iteration ensures responsive cancellation. ESC → cancel within ~100ms.
@@ -89,7 +89,7 @@ Frontend                          Backend
 **Why**: SMB and MTP volumes don't use FSEvents (`supports_watching() == false`), so they never get a `WatchedDirectory` entry. With the sequence on the watcher, `increment_sequence` returned `None` and `directory-diff` events were never emitted for those volumes. Moving the `AtomicU64` to `CachedListing` makes it work for all volume types. The FSEvents watcher path also uses this same counter now.
 
 **Decision**: `ListingEventSink` trait decouples streaming from Tauri (same pattern as `OperationEventSink` in write_operations)
-**Why**: `read_directory_with_progress` needs to emit events, but `tauri::AppHandle` can't be created in tests. The trait allows `CollectorListingEventSink` to capture events for assertions. `Arc<dyn ListingEventSink>` is used (not `&dyn`) because the sink is cloned into `std::thread::spawn` for progress callbacks.
+**Why**: `read_directory_with_progress` needs to emit events, but `tauri::AppHandle` can't be created in tests. The trait allows `CollectorListingEventSink` to capture events for assertions. `Arc<dyn ListingEventSink>` is used (not `&dyn`) because the sink is cloned into `tokio::spawn` for progress callbacks.
 
 **Decision**: File watcher starts AFTER listing complete
 **Why**: Watcher diffs rely on cached entries. Starting before cache is populated would miss initial state.
@@ -108,12 +108,11 @@ Frontend                          Backend
 
 ## Gotchas
 
-**Gotcha**: Listing code uses `Handle::current().block_on(volume.method())` for async Volume calls
-**Why**: The listing pipeline (`streaming.rs`, `caching.rs`, `operations.rs`) and the FSEvents watcher (`watcher.rs`)
-call Volume methods from sync contexts (`spawn_blocking` closures, watcher callbacks). These `block_on` bridges are
-intentional and correct — the threads are OS threads with no entered runtime. They're NOT the same anti-pattern as the
-old Volume-internal `block_on` bridges (which were eliminated by the async refactor). These could be migrated to fully
-async in a future listing pipeline refactor, but they work correctly as-is.
+**Gotcha**: Watcher callbacks (FSEvents) run on OS threads, not the tokio runtime
+**Why**: The FSEvents debouncer callback is called from an OS thread. Functions like `handle_directory_change` and
+`notify_directory_changed(FullRefresh)` are async, so callers in watcher callbacks use `tokio::spawn` to dispatch
+the async work onto the runtime. The incremental watcher path (`handle_directory_change_incremental`) remains sync
+since it only does cache lookups and `stat` calls via `get_single_entry`.
 
 ### Cache helpers (caching.rs)
 

apps/desktop/src-tauri/src/file_system/listing/caching.rs

@@ -272,16 +272,20 @@ pub fn notify_directory_changed(volume_id: &str, parent_path: &Path, change: Dir
                 // path is the share root, but the user may be browsing a subdirectory.
                 // Refresh all listings on this volume instead.
                 let volume_listings = find_listings_on_volume(volume_id);
-                for (lid, path, sort_by, sort_order, dir_sort_mode) in &volume_listings {
-                    notify_full_refresh(
-                        &app,
+                for (lid, path, sort_by, sort_order, dir_sort_mode) in volume_listings {
+                    let app = app.clone();
+                    let volume_id = volume_id.to_string();
+                    tokio::spawn(notify_full_refresh(
+                        app,
                         volume_id,
                         path,
-                        &[(lid.clone(), *sort_by, *sort_order, *dir_sort_mode)],
-                    );
+                        vec![(lid, sort_by, sort_order, dir_sort_mode)],
+                    ));
                 }
             } else {
-                notify_full_refresh(&app, volume_id, parent_path, &listings);
+                let volume_id = volume_id.to_string();
+                let parent_path = parent_path.to_path_buf();
+                tokio::spawn(notify_full_refresh(app, volume_id, parent_path, listings));
             }
         }
     }
@@ -390,25 +394,25 @@ fn notify_modified(app: &tauri::AppHandle, listing_id: &str, entry: FileEntry) {
 }
 
 /// Re-reads a directory via the Volume trait, computes a diff, and emits it.
-fn notify_full_refresh(
-    app: &tauri::AppHandle,
-    volume_id: &str,
-    parent_path: &Path,
-    listings: &[(String, SortColumn, SortOrder, DirectorySortMode)],
+async fn notify_full_refresh(
+    app: tauri::AppHandle,
+    volume_id: String,
+    parent_path: PathBuf,
+    listings: Vec<(String, SortColumn, SortOrder, DirectorySortMode)>,
 ) {
     use crate::file_system::listing::sorting::sort_entries;
     use crate::file_system::watcher::{DirectoryDiff, compute_diff};
     use tauri::Emitter;
 
-    let vol = match crate::file_system::get_volume_manager().get(volume_id) {
+    let vol = match crate::file_system::get_volume_manager().get(&volume_id) {
         Some(v) => v,
         None => {
             log::warn!("notify_directory_changed: volume `{}` not found", volume_id);
             return;
         }
     };
 
-    let mut new_entries = match tokio::runtime::Handle::current().block_on(vol.list_directory(parent_path)) {
+    let mut new_entries = match vol.list_directory(&parent_path).await {
         Ok(entries) => entries,
         Err(e) => {
             log::warn!(
@@ -422,7 +426,7 @@ fn notify_full_refresh(
 
     crate::indexing::enrich_entries_with_index(&mut new_entries);
 
-    for (listing_id, sort_by, sort_order, dir_sort_mode) in listings {
+    for (listing_id, sort_by, sort_order, dir_sort_mode) in &listings {
         // Re-sort to match this listing's sort params
         let mut sorted = new_entries.clone();
         sort_entries(&mut sorted, *sort_by, *sort_order, *dir_sort_mode);

apps/desktop/src-tauri/src/file_system/listing/operations.rs

@@ -47,7 +47,7 @@ pub struct ListingStartResult {
 ///
 /// Reads the directory once, caches it, and returns listing ID + total count.
 /// Frontend then fetches visible ranges on demand via `get_file_range`.
-pub fn list_directory_start(path: &Path, include_hidden: bool) -> Result<ListingStartResult, std::io::Error> {
+pub async fn list_directory_start(path: &Path, include_hidden: bool) -> Result<ListingStartResult, std::io::Error> {
     list_directory_start_with_volume(
         "root",
         path,
@@ -56,12 +56,13 @@ pub fn list_directory_start(path: &Path, include_hidden: bool) -> Result<Listing
         SortOrder::Ascending,
         DirectorySortMode::LikeFiles,
     )
+    .await
 }
 
 /// Starts a new directory listing using a specific volume.
 ///
 /// This is the internal implementation that supports multi-volume access.
-pub fn list_directory_start_with_volume(
+pub async fn list_directory_start_with_volume(
     volume_id: &str,
     path: &Path,
     include_hidden: bool,
@@ -82,8 +83,9 @@ pub fn list_directory_start_with_volume(
     })?;
 
     // Use the Volume trait to list the directory
-    let all_entries = tokio::runtime::Handle::current()
-        .block_on(volume.list_directory(path))
+    let all_entries = volume
+        .list_directory(path)
+        .await
         .map_err(|e| std::io::Error::other(e.to_string()))?;
     benchmark::log_event_value("volume.list_directory COMPLETE, entries", all_entries.len());
 

apps/desktop/src-tauri/src/file_system/listing/operations_test.rs

@@ -329,22 +329,15 @@ async fn test_list_directory_start_with_volume_caches_entries() {
 
     get_volume_manager().register(&volume_id, volume);
 
-    // Run on a blocking thread because the function internally uses
-    // `Handle::current().block_on()` which panics if called from an async context.
-    let vid = volume_id.clone();
-    let dp = dir_path.clone();
-    let result = tokio::task::spawn_blocking(move || {
-        super::list_directory_start_with_volume(
-            &vid,
-            &dp,
-            true,
-            SortColumn::Name,
-            SortOrder::Ascending,
-            DirectorySortMode::LikeFiles,
-        )
-    })
-    .await
-    .unwrap();
+    let result = super::list_directory_start_with_volume(
+        &volume_id,
+        &dir_path,
+        true,
+        SortColumn::Name,
+        SortOrder::Ascending,
+        DirectorySortMode::LikeFiles,
+    )
+    .await;
 
     assert!(
         result.is_ok(),
@@ -382,18 +375,15 @@ async fn test_list_directory_start_with_volume_unknown_volume() {
     use crate::file_system::listing::sorting::{DirectorySortMode, SortColumn, SortOrder};
     use std::path::PathBuf;
 
-    let result = tokio::task::spawn_blocking(move || {
-        super::list_directory_start_with_volume(
-            "nonexistent-volume-id",
-            &PathBuf::from("/some/path"),
-            true,
-            SortColumn::Name,
-            SortOrder::Ascending,
-            DirectorySortMode::LikeFiles,
-        )
-    })
-    .await
-    .unwrap();
+    let result = super::list_directory_start_with_volume(
+        "nonexistent-volume-id",
+        &PathBuf::from("/some/path"),
+        true,
+        SortColumn::Name,
+        SortOrder::Ascending,
+        DirectorySortMode::LikeFiles,
+    )
+    .await;
 
     assert!(result.is_err());
     assert!(result.unwrap_err().to_string().contains("not found"));

apps/desktop/src-tauri/src/file_system/listing/streaming.rs

@@ -7,7 +7,6 @@ use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::Path;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::mpsc;
 use std::sync::{Arc, LazyLock, RwLock};
 use std::time::Duration;
 
@@ -322,24 +321,21 @@ pub async fn list_directory_start_streaming(
 
     // Spawn background task
     tokio::spawn(async move {
-        // Clone again for use after spawn_blocking
+        // Clone again for use after the listing call
         let listing_id_for_cleanup = listing_id_for_spawn.clone();
         let events_for_error = Arc::clone(&events);
 
-        // Run blocking I/O on dedicated thread pool
-        let result = tokio::task::spawn_blocking(move || {
-            read_directory_with_progress(
-                &events,
-                &listing_id_for_spawn,
-                &state,
-                &volume_id_owned,
-                &path_owned,
-                include_hidden,
-                sort_by,
-                sort_order,
-                dir_sort_mode,
-            )
-        })
+        let result = read_directory_with_progress(
+            &events,
+            &listing_id_for_spawn,
+            &state,
+            &volume_id_owned,
+            &path_owned,
+            include_hidden,
+            sort_by,
+            sort_order,
+            dir_sort_mode,
+        )
         .await;
 
         // Clean up streaming state
@@ -350,21 +346,12 @@ pub async fn list_directory_start_streaming(
         // Handle task result
         match result {
             Err(e) => {
-                // Task panicked or was cancelled
-                events_for_error.emit_error(
-                    &listing_id_for_cleanup,
-                    "Something went wrong while reading this folder".to_string(),
-                    None,
-                );
-                log::error!("Listing task panicked: {}", e);
-            }
-            Ok(Err(e)) => {
                 // Function returned an error (volume not found, permission denied, I/O, etc.)
                 let mut friendly = friendly_error_from_volume_error(&e, &path_for_error);
                 enrich_with_provider(&mut friendly, &path_for_error);
                 events_for_error.emit_error(&listing_id_for_cleanup, e.to_string(), Some(friendly));
             }
-            Ok(Ok(())) => {
+            Ok(()) => {
                 // Success - read_directory_with_progress already emitted listing-complete
             }
         }
@@ -379,13 +366,14 @@ pub async fn list_directory_start_streaming(
 
 /// Reads a directory with progress reporting.
 ///
-/// Runs on a blocking thread pool and emits progress events.
+/// Async implementation that spawns the Volume I/O in a background task
+/// and uses `tokio::select!` with a cancellation polling loop for responsive ESC handling.
 /// Uses the Volume abstraction to support both local filesystem and MTP devices.
 #[allow(
     clippy::too_many_arguments,
     reason = "Streaming operation requires many state parameters"
 )]
-pub(crate) fn read_directory_with_progress(
+pub(crate) async fn read_directory_with_progress(
     events: &Arc<dyn ListingEventSink>,
     listing_id: &str,
     state: &Arc<StreamingListingState>,
@@ -420,47 +408,36 @@ pub(crate) fn read_directory_with_progress(
         .get(volume_id)
         .ok_or_else(|| VolumeError::NotFound(format!("Volume not found: {}", volume_id)))?;
 
-    // Read directory entries via Volume abstraction
-    // Use polling-based cancellation to remain responsive even when filesystem I/O blocks
-    // (for example, on slow/stuck network drives like SMB mounts)
+    // Read directory entries via Volume abstraction.
+    // Spawn the listing as a tokio task and use select! with a cancellation poll loop
+    // to remain responsive even when filesystem I/O blocks (slow/stuck network drives).
     let read_start = std::time::Instant::now();
-    let path_for_thread = path.to_path_buf();
-    let (tx, rx) = mpsc::channel();
+    let path_for_task = path.to_path_buf();
     let events_for_progress = Arc::clone(events);
     let listing_id_for_progress = listing_id.to_string();
 
-    // Capture the Tokio runtime handle so the spawned thread can access it.
-    // This is needed for MTP volumes, which use `Handle::block_on` internally.
-    let runtime_handle = tokio::runtime::Handle::current();
-
-    std::thread::spawn(move || {
-        // Enter the Tokio runtime context so `Handle::current()` works inside volumes
-        let _guard = runtime_handle.enter();
-
+    let listing_task = tokio::spawn(async move {
         let on_progress = |loaded_count: usize| {
             events_for_progress.emit_progress(&listing_id_for_progress, loaded_count);
         };
-        let result = runtime_handle.block_on(volume.list_directory_with_progress(&path_for_thread, &on_progress));
-        let _ = tx.send(result);
+        volume.list_directory_with_progress(&path_for_task, &on_progress).await
     });
 
-    // Poll for results, checking cancellation between polls
+    // Poll for cancellation while waiting for the listing task
     let entries_result = loop {
         if state.cancelled.load(Ordering::Relaxed) {
             benchmark::log_event("read_directory_with_progress CANCELLED (during read_dir polling)");
+            listing_task.abort();
             events.emit_cancelled(listing_id);
             return Ok(());
         }
 
-        match rx.recv_timeout(CANCELLATION_POLL_INTERVAL) {
-            Ok(result) => break result,
-            Err(mpsc::RecvTimeoutError::Timeout) => continue,
-            Err(mpsc::RecvTimeoutError::Disconnected) => {
-                return Err(VolumeError::IoError {
-                    message: "Directory listing thread terminated unexpectedly".into(),
-                    raw_os_error: None,
-                });
-            }
+        tokio::time::sleep(CANCELLATION_POLL_INTERVAL).await;
+        if listing_task.is_finished() {
+            break listing_task.await.map_err(|e| VolumeError::IoError {
+                message: format!("Directory listing task failed: {}", e),
+                raw_os_error: None,
+            })?;
         }
     };