Transfer: Honest ETA when files outnumber bytes

- New `EtaEstimator` in `write_operations/eta.rs` tracks bytes/s and files/s independently with time-weighted EWMA (τ ≈ 3 s) and combines via `max(ETA_bytes, ETA_files)`. The operation can't finish before either axis is done, so the larger remaining time is reality.
- Fixes the small-file-tail bug: deleting 5 GB / 174k files showed `~0 s remaining` for 20 s once the size bar saturated. ETA now tracks whichever axis still has work.
- `WriteProgressEvent` gains `bytesPerSecond`, `filesPerSecond`, `etaSeconds` — populated by `WriteOperationState::enrich_progress` at every emit site (local copy/delete/trash/scan/move + volume copy/move). One estimator per operation, lives in state.
- Pure EWMA, no overall-average anchor — adapts to network changes within ~3τ instead of being pulled toward historical numbers. Warm-up gate (≥ 2 samples AND ≥ 800 ms) kills early-readout flicker.
- `TransferProgressDialog` ripped out its blended-speed math; renders both speeds side by side ("27.7 MB/s · 1,234 files/s"). 25%/tick display low-pass on ETA prevents flicker without dampening real changes.
- 10 unit tests covering byte-heavy, file-heavy, big-then-small-tail pathology, mid-op slowdown convergence, brief stall + resume, phase transitions, and rollback. All write_operations tests (161) still green.

Author: vdavid

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

@@ -3502,7 +3502,6 @@ mod tests {
         use crate::file_system::write_operations::{
             CollectorEventSink, VolumeCopyConfig, WriteOperationState, copy_volumes_with_progress,
         };
-        use std::sync::atomic::AtomicU8;
         use std::time::{Duration, Instant};
 
         // Content scheme: `blake3(b"cmdr-fix8-" || index_le) .as_bytes() repeated 320 times`
@@ -3564,11 +3563,7 @@ mod tests {
             local_dir.path().to_path_buf(),
         ));
 
-        let state = Arc::new(WriteOperationState {
-            intent: Arc::new(AtomicU8::new(0)),
-            progress_interval: Duration::from_millis(200),
-            conflict_resolution_tx: std::sync::Mutex::new(None),
-        });
+        let state = Arc::new(WriteOperationState::new(Duration::from_millis(200)));
         let events = CollectorEventSink::new();
         let config = VolumeCopyConfig::default();
 
@@ -3777,7 +3772,6 @@ mod tests {
         use crate::file_system::write_operations::{
             CollectorEventSink, VolumeCopyConfig, WriteOperationState, copy_volumes_with_progress,
         };
-        use std::sync::atomic::AtomicU8;
         use std::time::{Duration, Instant};
 
         let _ = env_logger::try_init();
@@ -3889,11 +3883,7 @@ mod tests {
                 "dest",
                 local_dir.path().to_path_buf(),
             ));
-            let state = Arc::new(WriteOperationState {
-                intent: Arc::new(AtomicU8::new(0)),
-                progress_interval: Duration::from_millis(200),
-                conflict_resolution_tx: std::sync::Mutex::new(None),
-            });
+            let state = Arc::new(WriteOperationState::new(Duration::from_millis(200)));
             let events = CollectorEventSink::new();
             let config = VolumeCopyConfig::default();
 

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

@@ -21,6 +21,7 @@ network mounts, cross-filesystem moves, and name/path length limits.
 | `copy.rs` | `copy_files_with_progress`: scan → disk space check → per-file copy via `copy_single_item`. `CopyTransaction` for rollback. |
 | `move_op.rs` | Same-fs: `fs::rename`. Cross-fs: copy to `.cmdr-staging-<uuid>`, atomic rename, delete sources. |
 | `delete.rs` | Scan, delete files first, then directories in reverse/deepest-first order. Not rollbackable. Also contains `delete_volume_files_with_progress` for non-local volumes (MTP): scans via `volume.list_directory()`, deletes via `volume.delete()` per item. |
+| `eta.rs` | `EtaEstimator` — time-weighted EWMA per axis (bytes, files), τ ≈ 3 s. Combines via `max(ETA_bytes, ETA_files)`. One per `WriteOperationState`, fed by `state.enrich_progress` at every `write-progress` emit site. See [ETA + throughput](#eta--throughput) below. |
 | `trash.rs` | `move_to_trash_sync()` (macOS: ObjC `trashItemAtURL`; Linux: `trash` crate; reused by `commands/rename.rs`) and `trash_files_with_progress()` (batch trash with per-item progress, cancellation, partial failure). Uses `symlink_metadata()` for existence checks (handles dangling symlinks). |
 | `copy_strategy.rs` | Strategy selection per file: network FS → chunked copy; overwrite → temp+rename; macOS → `copyfile(3)`; Linux → `copy_file_range(2)`. |
 | `macos_copy.rs` | FFI to macOS `copyfile(3)`. Preserves xattrs, ACLs, resource forks, Finder metadata. Supports APFS `clonefile`. |
@@ -58,6 +59,24 @@ Frontend
   → state removed from both caches
 ```
 
+## ETA + throughput
+
+Rates and ETA are computed in the backend (`eta.rs`) and shipped on every `WriteProgressEvent` as `bytes_per_second`, `files_per_second`, and `eta_seconds`. The frontend renders these directly — no client-side math, no sample buffer.
+
+**Why backend, not frontend:** one place to test, one set of fields exposed on the wire, identical behavior across copy/move/delete/MTP/SMB/local. Putting the math in Svelte couples the estimator to dialog lifecycle and makes any future client (CLI, menu bar app) reinvent it.
+
+**Why two axes, not one:** the bug we hit in May 2026 was a delete of 5.4 GB / 174k files where the size bar saturated in the first second (a few large files) and the byte-based ETA collapsed to ~0 s while 165k small files were still streaming through. The estimator now tracks bytes/sec and files/sec independently and reports `eta = max(ETA_bytes, ETA_files)`. The operation can't finish before either axis is done, so the larger one is reality. When one axis has zero remaining work, its ETA is `0` and the other axis dominates naturally — no branching needed.
+
+**EWMA, not blended overall:** `α = 1 - exp(-Δt / τ)` with `τ = 3 s` (see `EWMA_TAU_SECS`). Pure exponential decay, no "overall average" anchor — if the network drops mid-operation, the EWMA converges to the new rate within a few τ instead of being pulled back toward historical numbers. Time-weighted means the response is the same whether progress events arrive every 50 ms or every 500 ms.
+
+**Warm-up:** the estimator returns `None` for ETA until it has ≥ 2 samples in the current phase AND ≥ 800 ms elapsed (`MIN_SAMPLES_FOR_ETA`, `MIN_ELAPSED_FOR_ETA`). This kills the early "200 ms in, rate = 50 MB/s → ETA = 0 s" footgun. Rates are populated as soon as we have the first delta; only the ETA is gated.
+
+**Phase transitions reset:** `update()` reseeds on every `phase` change. Without this, the counters' reset (scanning → copying both restart from 0) would feed a negative delta into the EWMA. Rollback is treated as a forward phase toward target `(0, 0)` — the estimator subtracts the new counters from the previous ones and ETA = current value / decay rate.
+
+**Wiring:** every `write-progress` emit site calls `state.emit_progress_via_app(app, event)` (for the AppHandle-direct path: copy/delete/trash/scan/move) or `state.emit_progress_via_sink(events, event)` (for the `OperationEventSink` path: volume copy/move). Both methods call `enrich_progress` internally, so no caller has to remember. The `bytes_per_second: None, files_per_second: None, eta_seconds: None` placeholders in the struct literals get overwritten before the event reaches the FE.
+
+**Frontend display:** `TransferProgressDialog.svelte` stores the three fields in local `$state` and renders both speeds side by side ("27.7 MB/s · 1,234 files/s"). A tiny low-pass on the displayed ETA (25% gap-closure per tick) prevents flicker without dampening real changes. The display ETA also resets to `null` on phase transitions to re-warm with the backend.
+
 ## Key patterns and gotchas
 
 **All blocking work in `spawn_blocking`.** Never call blocking I/O on the async executor.

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

@@ -156,8 +156,8 @@ pub(super) fn copy_files_with_progress(
     let mut created_dirs: HashSet<PathBuf> = HashSet::new();
 
     // Emit initial copying phase event (important when reusing cached scan - no scanning events were emitted)
-    let _ = app.emit(
-        "write-progress",
+    state.emit_progress_via_app(
+        app,
         WriteProgressEvent {
             operation_id: operation_id.to_string(),
             operation_type: WriteOperationType::Copy,
@@ -167,6 +167,10 @@ pub(super) fn copy_files_with_progress(
             files_total: scan_result.file_count,
             bytes_done: 0,
             bytes_total: scan_result.total_bytes,
+
+            bytes_per_second: None,
+            files_per_second: None,
+            eta_seconds: None,
         },
     );
     update_operation_status(
@@ -365,8 +369,6 @@ pub(super) fn copy_single_item(
     apply_to_all_resolution: &mut Option<ConflictResolution>,
     created_dirs: &mut HashSet<PathBuf>,
 ) -> Result<(), WriteOperationError> {
-    use tauri::Emitter;
-
     // Check cancellation
     if is_cancelled(&state.intent) {
         log::debug!(
@@ -620,8 +622,8 @@ pub(super) fn copy_single_item(
                     effective_bytes_done,
                     bytes_total
                 );
-                let _ = app.emit(
-                    "write-progress",
+                state.emit_progress_via_app(
+                    app,
                     WriteProgressEvent {
                         operation_id: operation_id.to_string(),
                         operation_type,
@@ -631,6 +633,10 @@ pub(super) fn copy_single_item(
                         files_total,
                         bytes_done: effective_bytes_done,
                         bytes_total,
+
+                        bytes_per_second: None,
+                        files_per_second: None,
+                        eta_seconds: None,
                     },
                 );
                 update_operation_status(
@@ -669,8 +675,8 @@ pub(super) fn copy_single_item(
                 *bytes_done,
                 bytes_total
             );
-            let _ = app.emit(
-                "write-progress",
+            state.emit_progress_via_app(
+                app,
                 WriteProgressEvent {
                     operation_id: operation_id.to_string(),
                     operation_type,
@@ -680,6 +686,10 @@ pub(super) fn copy_single_item(
                     files_total,
                     bytes_done: *bytes_done,
                     bytes_total,
+
+                    bytes_per_second: None,
+                    files_per_second: None,
+                    eta_seconds: None,
                 },
             );
             update_operation_status(
@@ -727,15 +737,13 @@ fn rollback_with_progress(
     files_total: usize,
     bytes_total: u64,
 ) -> bool {
-    use tauri::Emitter;
-
     let files_to_delete = transaction.created_files.len();
     let mut files_deleted = 0usize;
     let mut last_progress_time = Instant::now();
 
     // Emit initial rollback phase event (same values as cancellation point)
-    let _ = app.emit(
-        "write-progress",
+    state.emit_progress_via_app(
+        app,
         WriteProgressEvent {
             operation_id: operation_id.to_string(),
             operation_type,
@@ -745,6 +753,10 @@ fn rollback_with_progress(
             files_total,
             bytes_done: bytes_at_cancel,
             bytes_total,
+
+            bytes_per_second: None,
+            files_per_second: None,
+            eta_seconds: None,
         },
     );
     update_operation_status(
@@ -788,8 +800,8 @@ fn rollback_with_progress(
                 .file_name()
                 .map(|n| n.to_string_lossy().to_string())
                 .unwrap_or_default();
-            let _ = app.emit(
-                "write-progress",
+            state.emit_progress_via_app(
+                app,
                 WriteProgressEvent {
                     operation_id: operation_id.to_string(),
                     operation_type,
@@ -799,6 +811,10 @@ fn rollback_with_progress(
                     files_total,
                     bytes_done: remaining_bytes,
                     bytes_total,
+
+                    bytes_per_second: None,
+                    files_per_second: None,
+                    eta_seconds: None,
                 },
             );
             update_operation_status(

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

@@ -101,8 +101,8 @@ pub(super) fn delete_files_with_progress(
         // Emit progress
         if last_progress_time.elapsed() >= state.progress_interval {
             let current_file = file_info.path.file_name().map(|n| n.to_string_lossy().to_string());
-            let _ = app.emit(
-                "write-progress",
+            state.emit_progress_via_app(
+                app,
                 WriteProgressEvent {
                     operation_id: operation_id.to_string(),
                     operation_type: WriteOperationType::Delete,
@@ -112,6 +112,10 @@ pub(super) fn delete_files_with_progress(
                     files_total: scan_result.file_count,
                     bytes_done,
                     bytes_total: scan_result.total_bytes,
+
+                    bytes_per_second: None,
+                    files_per_second: None,
+                    eta_seconds: None,
                 },
             );
             update_operation_status(
@@ -194,8 +198,6 @@ async fn scan_volume_recursive(
     operation_id: &str,
     last_progress_time: &mut Instant,
 ) -> Result<(), WriteOperationError> {
-    use tauri::Emitter;
-
     if super::state::is_cancelled(&state.intent) {
         return Err(WriteOperationError::Cancelled {
             message: "Operation cancelled by user".to_string(),
@@ -261,8 +263,8 @@ async fn scan_volume_recursive(
     if last_progress_time.elapsed() >= state.progress_interval {
         let file_count = entries.iter().filter(|e| !e.is_dir).count();
         let current_file = path.file_name().map(|n| n.to_string_lossy().to_string());
-        let _ = app.emit(
-            "write-progress",
+        state.emit_progress_via_app(
+            app,
             WriteProgressEvent {
                 operation_id: operation_id.to_string(),
                 operation_type: WriteOperationType::Delete,
@@ -272,6 +274,10 @@ async fn scan_volume_recursive(
                 files_total: 0,
                 bytes_done: *total_bytes,
                 bytes_total: 0,
+
+                bytes_per_second: None,
+                files_per_second: None,
+                eta_seconds: None,
             },
         );
         update_operation_status(
@@ -344,8 +350,8 @@ pub(super) async fn delete_volume_files_with_progress(
     let file_count = entries.iter().filter(|e| !e.is_dir).count();
 
     // Emit final scan progress
-    let _ = app.emit(
-        "write-progress",
+    state.emit_progress_via_app(
+        app,
         WriteProgressEvent {
             operation_id: operation_id.to_string(),
             operation_type: WriteOperationType::Delete,
@@ -355,6 +361,10 @@ pub(super) async fn delete_volume_files_with_progress(
             files_total: file_count,
             bytes_done: total_bytes,
             bytes_total: total_bytes,
+
+            bytes_per_second: None,
+            files_per_second: None,
+            eta_seconds: None,
         },
     );
 
@@ -408,8 +418,8 @@ pub(super) async fn delete_volume_files_with_progress(
 
         if last_progress_time.elapsed() >= state.progress_interval {
             let current_file = entry.path.file_name().map(|n| n.to_string_lossy().to_string());
-            let _ = app.emit(
-                "write-progress",
+            state.emit_progress_via_app(
+                app,
                 WriteProgressEvent {
                     operation_id: operation_id.to_string(),
                     operation_type: WriteOperationType::Delete,
@@ -419,6 +429,10 @@ pub(super) async fn delete_volume_files_with_progress(
                     files_total: file_count,
                     bytes_done,
                     bytes_total: total_bytes,
+
+                    bytes_per_second: None,
+                    files_per_second: None,
+                    eta_seconds: None,
                 },
             );
             update_operation_status(