MTP: propagate cancel via mtp-rs cancel tokens (M2)

- `WriteOperationState.backend_cancel` (`Arc<AtomicBool>`) is flipped together with `intent` by `cancel_write_operation` and `cancel_all_write_operations`, so any cancel path stops the wire activity (not just the loop above).
- `Volume` trait grows `list_directory_with_cancel` / `delete_with_cancel` defaults that drop the flag — only MTP overrides today; local, SMB, in-memory inherit and ignore.
- `MtpVolume` wraps the flag as a fresh `mtp_rs::CancelToken` via `CancelToken::from_arc(Arc::clone(...))` — shared atomic, no second polling task — and threads it through to `MtpConnectionManager::list_directory_with_cancel`, `list_directory_with_progress_and_cancel`, and `delete_object_with_cancel`. The recursive child-delete loop also checks the token between iterations.
- `MtpConnectionManager` calls `storage.list_objects_with_cancel` / `storage.list_objects_stream_with_cancel` / `storage.delete_with_cancel` in `mtp-rs`, so the per-handle `GetObjectInfo` USB loop bails at the next roundtrip boundary instead of running all 950 entries on a `/DCIM/Camera`-style folder.
- `map_mtp_error` now maps `mtp_rs::Error::Cancelled` to the typed `MtpConnectionError::Cancelled` variant (was wrapped in `Other` with a string match upstream — typed all the way through now).
- `delete.rs` (scan walker + per-leaf delete + dir cleanup) calls the cancel-aware Volume variants with `Some(&state.backend_cancel)`.
- Tests: 5 new state-machine tests pin that the intent-flip adapter sets `backend_cancel`. 2 new integration tests against a `CancellingVolume` mimicking MTP's per-handle loop pin promptness (bail before all entries processed, total wall-clock under 3 s).
- Docs: `mtp/CLAUDE.md` documents the per-roundtrip cancel point, the wiring, the rationale vs PTP `CancelTransaction (0x4001)`, and the hardware caveat. `volume/CLAUDE.md` documents the new trait variants and which backends opt in.
- `mtp-rs` dep switched to local path while 0.15.0 is unpublished.

Author: vdavid

Cargo.lock

@@ -161,7 +161,8 @@ mime_guess = "2"
 
 [target.'cfg(any(target_os = "macos", target_os = "linux"))'.dependencies]
 # MTP (Android device) support via pure Rust implementation
-mtp-rs = "0.14.0"
+# Path dep until mtp-rs 0.15.0 is published. Switch back to a crates.io version once released.
+mtp-rs = { path = "../../../../../../../mtp-rs" }
 # USB hotplug detection for MTP device watcher
 nusb = "0.2.3"
 bytes = "1"

apps/desktop/src-tauri/Cargo.toml

@@ -55,6 +55,33 @@ Optional methods default to `Err(VolumeError::NotSupported)` or `false`, so new
     3. The `refresh_listing` Tauri command (`commands/file_system/listing.rs`) — short-circuits the post-transfer redundant `list_directory` re-read entirely when the volume is keeping the cache fresh via `notify_mutation`. Without this, a 1k-entry MTP folder paid ~17 s + USB session collision after every transfer outcome, wedging the next user op.
   Default `false` so a new backend without a real watcher won't accidentally claim freshness. **Freshness contract**: a `true` result does NOT mean the cache is byte-perfect with the device right now. Every backend has a debounce or settling window between a real change and the cache reflecting it: local FS ≈ 10 ms (FSEvents coalesce), SMB 200 ms (watcher debounce; > 50 events/dir triggers a `FullRefresh`), MTP 500 ms (event debouncer plus per-device polling; many cameras emit no events at all, so on those `true` means only "the device is reachable"). Callers must treat the result as "fresh as our most recent observation" — the same guarantee a `list_directory` call gives. The MTP and SMB checks are volume-level, not path-level: when the gate flips true, every path on that volume becomes oracle-eligible.
 
+## Cancel-aware variants
+
+`list_directory_with_cancel(path, on_progress, cancel)` and
+`delete_with_cancel(path, cancel)` accept an
+`Option<&Arc<AtomicBool>>` that backends interpret as a cooperative cancel
+flag. Default impls delegate to the non-cancel `list_directory` / `delete`,
+dropping the flag — so adding a new backend doesn't have to implement them
+unless its operations are interruptible at a meaningful boundary.
+
+- `MtpVolume` overrides both. The flag wraps a fresh `mtp_rs::CancelToken` via
+  `CancelToken::from_arc(Arc::clone(...))` (shared atomic, no polling task) and
+  threads through to mtp-rs's `list_objects_with_cancel` /
+  `delete_with_cancel`. That bails the per-handle `GetObjectInfo` loop within
+  one USB roundtrip's latency.
+- `LocalPosixVolume`, `SmbVolume`, and `InMemoryVolume` inherit the default
+  (ignore the flag). Local listings are effectively atomic; SMB cancel
+  propagation is a follow-up.
+
+The write-op layer hands `Some(&state.backend_cancel)` (a clone of the same
+`Arc<AtomicBool>` that `cancel_write_operation` flips when intent leaves
+`Running`). Volumes that ignore the flag are unaffected; volumes that consume
+it stop their wire activity, not just the loop above.
+
+See `apps/desktop/src-tauri/src/mtp/CLAUDE.md` § "Cancel propagation" for the
+MTP-specific wiring and the rationale for "between-roundtrip" cancel vs PTP
+`CancelTransaction`.
+
 ## Building a new volume
 
 Adding a new backend (say, FTP, WebDAV, S3, or a new device protocol) is a matter of implementing the `Volume` trait and opting into the capability flags that make sense for your backend. The checklist below walks the path in the order you'd hit each concern.

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

@@ -363,6 +363,29 @@ pub trait Volume: Send + Sync {
         on_progress: Option<&'a (dyn Fn(usize) + Sync)>,
     ) -> Pin<Box<dyn Future<Output = Result<Vec<FileEntry>, VolumeError>> + Send + 'a>>;
 
+    /// Cancel-aware version of [`list_directory`](Self::list_directory).
+    ///
+    /// `cancel`, when `Some`, is consulted by backends that issue many small
+    /// USB or network roundtrips inside one listing (currently MTP — a 950-entry
+    /// folder is 950 `GetObjectInfo` calls). When the flag flips to `true`,
+    /// the backend bails between roundtrips with `VolumeError::Cancelled`
+    /// instead of running to completion.
+    ///
+    /// Local and in-memory backends ignore the flag (their listings are
+    /// effectively atomic from the caller's perspective). SMB ignores it
+    /// today — adding SMB cancel propagation is a follow-up.
+    ///
+    /// Default impl delegates to `list_directory`, dropping the flag.
+    fn list_directory_with_cancel<'a>(
+        &'a self,
+        path: &'a Path,
+        on_progress: Option<&'a (dyn Fn(usize) + Sync)>,
+        cancel: Option<&'a std::sync::Arc<AtomicBool>>,
+    ) -> Pin<Box<dyn Future<Output = Result<Vec<FileEntry>, VolumeError>> + Send + 'a>> {
+        let _ = cancel;
+        self.list_directory(path, on_progress)
+    }
+
     /// Gets metadata for a single path (relative to volume root).
     fn get_metadata<'a>(
         &'a self,
@@ -434,6 +457,23 @@ pub trait Volume: Send + Sync {
         Box::pin(async { Err(VolumeError::NotSupported) })
     }
 
+    /// Cancel-aware version of [`delete`](Self::delete).
+    ///
+    /// MTP overrides this to thread the cancel flag through to mtp-rs's
+    /// `delete_with_cancel`, which bails before issuing the `DeleteObject` PTP
+    /// request when the flag is set. For non-empty directories the MTP
+    /// implementation also checks the flag between recursive child deletes.
+    ///
+    /// Default impl delegates to `delete`, dropping the flag.
+    fn delete_with_cancel<'a>(
+        &'a self,
+        path: &'a Path,
+        cancel: Option<&'a std::sync::Arc<AtomicBool>>,
+    ) -> Pin<Box<dyn Future<Output = Result<(), VolumeError>> + Send + 'a>> {
+        let _ = cancel;
+        self.delete(path)
+    }
+
     /// Renames/moves a file or directory within this volume.
     ///
     /// Both source and destination paths are relative to the volume root.

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

@@ -106,29 +106,50 @@ impl Volume for MtpVolume {
         &'a self,
         path: &'a Path,
         on_progress: Option<&'a (dyn Fn(usize) + Sync)>,
+    ) -> Pin<Box<dyn Future<Output = Result<Vec<FileEntry>, VolumeError>> + Send + 'a>> {
+        self.list_directory_with_cancel(path, on_progress, None)
+    }
+
+    fn list_directory_with_cancel<'a>(
+        &'a self,
+        path: &'a Path,
+        on_progress: Option<&'a (dyn Fn(usize) + Sync)>,
+        cancel: Option<&'a std::sync::Arc<std::sync::atomic::AtomicBool>>,
     ) -> Pin<Box<dyn Future<Output = Result<Vec<FileEntry>, VolumeError>> + Send + 'a>> {
         Box::pin(async move {
             #[cfg(test)]
             test_hooks::bump_list_directory_call_count();
 
             let mtp_path = self.to_mtp_path(path);
 
+            // Build a mtp_rs CancelToken that shares the caller's Arc<AtomicBool>.
+            // No second polling task: flipping the original atomic flips the token.
+            let cancel_token = cancel.map(|c| mtp_rs::CancelToken::from_arc(std::sync::Arc::clone(c)));
+            let cancel_ref = cancel_token.as_ref();
+
             debug!(
-                "MtpVolume::list_directory: device={}, storage={}, input_path={}, mtp_path={}",
+                "MtpVolume::list_directory: device={}, storage={}, input_path={}, mtp_path={}, cancel={}",
                 self.device_id,
                 self.storage_id,
                 path.display(),
-                mtp_path
+                mtp_path,
+                cancel_ref.is_some()
             );
 
             let start = std::time::Instant::now();
             let result = if let Some(on_progress) = on_progress {
                 connection_manager()
-                    .list_directory_with_progress(&self.device_id, self.storage_id, &mtp_path, on_progress)
+                    .list_directory_with_progress_and_cancel(
+                        &self.device_id,
+                        self.storage_id,
+                        &mtp_path,
+                        on_progress,
+                        cancel_ref,
+                    )
                     .await
             } else {
                 connection_manager()
-                    .list_directory(&self.device_id, self.storage_id, &mtp_path)
+                    .list_directory_with_cancel(&self.device_id, self.storage_id, &mtp_path, cancel_ref)
                     .await
             };
 
@@ -322,11 +343,22 @@ impl Volume for MtpVolume {
     }
 
     fn delete<'a>(&'a self, path: &'a Path) -> Pin<Box<dyn Future<Output = Result<(), VolumeError>> + Send + 'a>> {
+        self.delete_with_cancel(path, None)
+    }
+
+    fn delete_with_cancel<'a>(
+        &'a self,
+        path: &'a Path,
+        cancel: Option<&'a std::sync::Arc<std::sync::atomic::AtomicBool>>,
+    ) -> Pin<Box<dyn Future<Output = Result<(), VolumeError>> + Send + 'a>> {
         Box::pin(async move {
             let mtp_path = self.to_mtp_path(path);
 
+            let cancel_token = cancel.map(|c| mtp_rs::CancelToken::from_arc(std::sync::Arc::clone(c)));
+            let cancel_ref = cancel_token.as_ref();
+
             connection_manager()
-                .delete_object(&self.device_id, self.storage_id, &mtp_path)
+                .delete_object_with_cancel(&self.device_id, self.storage_id, &mtp_path, cancel_ref)
                 .await
                 .map_err(map_mtp_error)?;
 

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

@@ -318,7 +318,7 @@ async fn scan_volume_recursive(
                 cached
             }
             None => volume
-                .list_directory(path, Some(&on_progress))
+                .list_directory_with_cancel(path, Some(&on_progress), Some(&state.backend_cancel))
                 .await
                 .map_err(|e| map_volume_error(&path.display().to_string(), e))?,
         };
@@ -682,7 +682,7 @@ pub(super) async fn delete_volume_files_with_progress_inner(
         }
 
         volume
-            .delete(&entry.path)
+            .delete_with_cancel(&entry.path, Some(&state.backend_cancel))
             .await
             .map_err(|e| map_volume_error(&entry.path.display().to_string(), e))?;
 
@@ -732,7 +732,9 @@ pub(super) async fn delete_volume_files_with_progress_inner(
         }
 
         // Best-effort directory removal (may fail if not empty due to partial delete)
-        let _ = volume.delete(&entry.path).await;
+        let _ = volume
+            .delete_with_cancel(&entry.path, Some(&state.backend_cancel))
+            .await;
     }
 
     // Emit completion

apps/desktop/src-tauri/src/file_system/write_operations/delete.rs

@@ -384,3 +384,5 @@ mod tests;
 mod transaction_integration_test;
 #[cfg(test)]
 mod validation_integration_test;
+#[cfg(test)]
+mod volume_cancel_tests;

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

@@ -75,6 +75,20 @@ pub struct WriteOperationState {
     /// at every `write-progress` emit site, so every emitter (local copy/delete,
     /// volume copy/move, MTP, SMB) reports rates and ETA uniformly.
     pub estimator: std::sync::Mutex<EtaEstimator>,
+    /// Cooperative cancel flag for in-flight backend I/O. Flipped whenever the
+    /// op transitions out of `Running` (via `cancel_write_operation`,
+    /// `cancel_all_write_operations`, or `cancel_all_write_operations_with_rollback`).
+    /// MTP volume ops thread this into `list_objects_with_cancel` /
+    /// `delete_with_cancel` so an in-flight `GetObjectInfo` loop bails at the
+    /// next per-handle USB boundary (≈one roundtrip) instead of running the
+    /// full 950-photo `/DCIM/Camera` listing to completion. Non-MTP backends
+    /// ignore it for now.
+    ///
+    /// Kept as a raw `Arc<AtomicBool>` (not the mtp-rs `CancelToken` type) so
+    /// this module doesn't pull mtp-rs onto non-MTP platforms; the MTP wiring
+    /// layer (`mtp::connection`) builds a fresh `mtp_rs::CancelToken` from
+    /// this flag at the entry point of each MTP-aware call.
+    pub backend_cancel: Arc<AtomicBool>,
 }
 
 impl WriteOperationState {
@@ -87,6 +101,7 @@ impl WriteOperationState {
             progress_interval,
             conflict_resolution_tx: std::sync::Mutex::new(None),
             estimator: std::sync::Mutex::new(EtaEstimator::new()),
+            backend_cancel: Arc::new(AtomicBool::new(false)),
         }
     }
 
@@ -312,6 +327,9 @@ pub fn cancel_write_operation(operation_id: &str, rollback: bool) {
         }
 
         state.intent.store(target as u8, Ordering::Relaxed);
+        // Any transition out of `Running` should also stop in-flight backend
+        // I/O (per-handle MTP loops, etc.) — not just the loop above it.
+        state.backend_cancel.store(true, Ordering::Release);
         // Drop the conflict resolution sender to unblock any waiting receiver
         let _ = state.conflict_resolution_tx.lock().unwrap().take();
     }
@@ -329,6 +347,7 @@ pub fn cancel_all_write_operations() {
             if current != OperationIntent::Stopped {
                 log::info!("cancel_all_write_operations: stopping op={id}");
                 state.intent.store(OperationIntent::Stopped as u8, Ordering::Relaxed);
+                state.backend_cancel.store(true, Ordering::Release);
                 // Drop the conflict resolution sender to unblock any waiting receiver
                 let _ = state.conflict_resolution_tx.lock().unwrap().take();
             }
@@ -694,6 +713,67 @@ mod tests {
         uninstall_state(&id);
     }
 
+    // ---- backend_cancel flag flipping (M2 wedge fix) ------------------------
+
+    #[test]
+    fn backend_cancel_starts_unset_on_fresh_state() {
+        let state = WriteOperationState::new(Duration::from_millis(50));
+        assert!(!state.backend_cancel.load(Ordering::Acquire));
+    }
+
+    #[test]
+    fn cancel_write_operation_flips_backend_cancel_to_stopped() {
+        let id = unique_id("cancel-flips-backend-stopped");
+        let state = install_state(&id, OperationIntent::Running);
+        assert!(!state.backend_cancel.load(Ordering::Acquire));
+        cancel_write_operation(&id, false);
+        assert!(
+            state.backend_cancel.load(Ordering::Acquire),
+            "cancel → Stopped must also flip backend_cancel so in-flight USB ops bail"
+        );
+        uninstall_state(&id);
+    }
+
+    #[test]
+    fn cancel_write_operation_flips_backend_cancel_to_rolling_back() {
+        let id = unique_id("cancel-flips-backend-rb");
+        let state = install_state(&id, OperationIntent::Running);
+        cancel_write_operation(&id, true);
+        assert!(
+            state.backend_cancel.load(Ordering::Acquire),
+            "cancel → RollingBack must also flip backend_cancel — the user wants the wire activity stopped, even though we're going to delete created files"
+        );
+        uninstall_state(&id);
+    }
+
+    #[test]
+    fn cancel_all_write_operations_flips_backend_cancel() {
+        let id = unique_id("cancel-all-flips-backend");
+        let state = install_state(&id, OperationIntent::Running);
+        cancel_all_write_operations();
+        assert!(
+            state.backend_cancel.load(Ordering::Acquire),
+            "cancel_all must flip backend_cancel so teardown also stops the wire activity"
+        );
+        uninstall_state(&id);
+    }
+
+    #[test]
+    fn cancel_stopped_is_noop_for_backend_cancel_too() {
+        // Stopped → anything is terminal, so backend_cancel state must not
+        // change either. This guards against a subtle regression where the
+        // flag flip happens before the validity check.
+        let id = unique_id("cancel-stopped-noop");
+        let state = install_state(&id, OperationIntent::Stopped);
+        state.backend_cancel.store(false, Ordering::Release);
+        cancel_write_operation(&id, true);
+        assert!(
+            !state.backend_cancel.load(Ordering::Acquire),
+            "Stopped is terminal: invalid transition must not flip backend_cancel"
+        );
+        uninstall_state(&id);
+    }
+
     #[test]
     fn cancel_unknown_operation_is_a_silent_noop() {
         // No installed state; must not panic, must not affect anything.