feat: live bottleneck indicator on the Activity dashboard

README.md

@@ -178,10 +178,12 @@ These move: check each project's current docs before relying on a cell.
   snapshot (Settings > macOS), which does not help with a Full Disk Access
   denial; there is no Linux equivalent.
 - In-app restore browser with full-text file-name search and streaming decrypt.
-- Activity dashboard with a live tail, filterable history, and real-time
+- Activity dashboard with a live tail, filterable history, real-time
   disk-read and network-upload throughput graphs (probe-fed, one-second
   resolution - they move during every phase of a backup, including crash
-  recovery).
+  recovery), and a live Bottleneck tile naming which stage - disk, network,
+  a rate-limited destination, or CPU hashing - is presently the limiting
+  factor (debounced a few seconds so it does not flicker between readings).
 - Rolling local log files covering both the backend and the webview console,
   collected into a one-click diagnostics bundle alongside a redacted summary of
   in-flight upload recovery state and a trailing window of process-memory

crates/driven-core/src/executor.rs

@@ -2746,6 +2746,10 @@ impl DefaultExecutor {
             .map_err(UploadError::from_read)?;
         if let Some(io) = self.io_counters.as_ref() {
             io.add_disk_read(plaintext_len);
+            // issue #308 bottleneck classifier: this buffered path both reads
+            // and blake3-hashes the whole file in one pass, so both counters
+            // move together here.
+            io.add_hashed(plaintext_len);
         }
 
         // --- post-read fstat identity check (SPEC s8 defence #3) -----------
@@ -2862,7 +2866,7 @@ impl DefaultExecutor {
             self.mem_gauge.clone(),
             self.io_counters.clone(),
         );
-        let cpu = cpu_stage(raw_rx, out_tx, crypto, size);
+        let cpu = cpu_stage(raw_rx, out_tx, crypto, size, self.io_counters.clone());
         let uploader = self.upload_stage(
             target,
             existing_file_id,
@@ -7292,20 +7296,29 @@ async fn cpu_stage(
     out_tx: tokio::sync::mpsc::Sender<Bytes>,
     crypto: Option<Arc<dyn SourceCryptoSuite>>,
     size: u64,
+    // issue #308 bottleneck classifier: credited alongside the disk/net
+    // counters the reader/uploader stages already feed, so the "cpu" state
+    // has a real hash-bytes/sec rate to compare against them.
+    io_counters: Option<Arc<crate::iostat::IoCounters>>,
 ) -> Result<CpuOutput, StageError> {
     use md5::{Digest, Md5};
 
     let mut hasher = blake3::Hasher::new();
     let use_rayon = size >= RAYON_HASH_THRESHOLD;
     let mut md5 = Md5::new();
 
-    // Hash a plaintext chunk into blake3, multi-core for big files.
+    // Hash a plaintext chunk into blake3, multi-core for big files. Credits
+    // the hash counter (issue #308) on every chunk regardless of path
+    // (encrypted or not) since both arms below call this closure.
     let hash_chunk = |h: &mut blake3::Hasher, chunk: &[u8]| {
         if use_rayon {
             h.update_rayon(chunk);
         } else {
             h.update(chunk);
         }
+        if let Some(io) = io_counters.as_ref() {
+            io.add_hashed(chunk.len() as u64);
+        }
     };
 
     if let Some(suite) = crypto {
@@ -13265,6 +13278,65 @@ mod tests {
         );
     }
 
+    /// issue #308 bottleneck classifier: a fresh small-file upload (the
+    /// BUFFERED `inline_upload` path, below [`PIPELINE_THRESHOLD`]) credits
+    /// the hash counter with the whole plaintext, alongside the existing
+    /// disk-read credit - both happen in the same `read_hash_encrypt` pass.
+    #[tokio::test]
+    async fn inline_upload_credits_hashed_bytes() {
+        let h = harness().await;
+        let body = vec![7u8; 4096];
+        let (rel, size) = h.write_file("small.bin", &body);
+
+        let io = Arc::new(crate::iostat::IoCounters::default());
+        let exec = h.executor().with_io_counters(io.clone());
+        let out = exec
+            .execute(
+                &h.source,
+                &h.upload_plan(&rel, size),
+                &noop_progress,
+                &noop_outcome,
+            )
+            .await
+            .unwrap();
+        assert!(matches!(out[0], OpOutcome::Done { .. }), "got {:?}", out[0]);
+
+        let snap = io.snapshot();
+        assert_eq!(snap.hashed_bytes, size, "the whole plaintext was hashed");
+        assert_eq!(snap.disk_read_bytes, size, "and read from disk");
+    }
+
+    /// issue #308 bottleneck classifier: a fresh large-file upload (the
+    /// STREAMING `cpu_stage` path, at/above [`PIPELINE_THRESHOLD`]) credits
+    /// the hash counter chunk-by-chunk as it streams, summing to the whole
+    /// plaintext by the time the upload completes.
+    #[tokio::test]
+    async fn stream_upload_credits_hashed_bytes() {
+        let h = harness().await;
+        let size_bytes = (PIPELINE_THRESHOLD + 64 * 1024) as usize;
+        let body: Vec<u8> = (0..size_bytes).map(|i| (i % 251) as u8).collect();
+        let (rel, size) = h.write_file("streamed.bin", &body);
+
+        let io = Arc::new(crate::iostat::IoCounters::default());
+        let exec = h.executor().with_io_counters(io.clone());
+        let out = exec
+            .execute(
+                &h.source,
+                &h.upload_plan(&rel, size),
+                &noop_progress,
+                &noop_outcome,
+            )
+            .await
+            .unwrap();
+        assert!(matches!(out[0], OpOutcome::Done { .. }), "got {:?}", out[0]);
+
+        let snap = io.snapshot();
+        assert_eq!(
+            snap.hashed_bytes, size,
+            "every streamed chunk's bytes were credited to the hash counter"
+        );
+    }
+
     /// [`ResumeAcc`]'s gauge accounting is symmetric across push / partial
     /// drain / clear, the drain clamps at the buffered length, and DROP
     /// refunds whatever is left - the guarantee the error-unwind test relies

crates/driven-core/src/iostat.rs

@@ -21,11 +21,17 @@
 //!   completion for single-request uploads). Each byte is credited exactly
 //!   once; bundle members are covered by their bundle's wire push, never
 //!   double-counted at completion.
+//! - `hashed`: plaintext bytes blake3-hashed (issue #308 bottleneck
+//!   classifier, 2026-08-17 follow-up). Credited from the two hot hashing
+//!   paths - the upload pipeline's cpu stage (streamed and buffered) and the
+//!   scanner's deep-verify re-hash - so the "cpu" bottleneck state has a real
+//!   rate to compare against `disk_read` and `net_wire`. Deliberately its own
+//!   counter rather than folded into `disk_read`: a deep-verify re-hash of an
+//!   already-synced file hashes bytes without any corresponding upload, so
+//!   conflating the two would make a hash-only scan look like disk activity.
 //!
-//! v1 scope notes: the scanner's deep-verify hashing and the restore path do
-//! not credit `disk_read` yet, and bundle ASSEMBLY reads (tar-ing members)
-//! are approximated by the bundle's wire push rather than counted at read
-//! time.
+//! v1 scope notes: bundle ASSEMBLY reads (tar-ing members) are approximated
+//! by the bundle's wire push rather than counted at read time.
 
 use std::sync::atomic::{AtomicU64, Ordering};
 
@@ -36,6 +42,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 pub struct IoCounters {
     disk_read: AtomicU64,
     net_wire: AtomicU64,
+    hashed: AtomicU64,
 }
 
 /// One peek of the cumulative totals.
@@ -45,6 +52,8 @@ pub struct IoSnapshot {
     pub disk_read_bytes: u64,
     /// Total wire bytes accepted by the destination.
     pub net_wire_bytes: u64,
+    /// Total plaintext bytes blake3-hashed (issue #308).
+    pub hashed_bytes: u64,
 }
 
 impl IoCounters {
@@ -58,11 +67,20 @@ impl IoCounters {
         self.net_wire.fetch_add(n, Ordering::Relaxed);
     }
 
-    /// Peek both totals. Never resets - samplers diff consecutive snapshots.
+    /// Credit `n` plaintext bytes blake3-hashed (issue #308 bottleneck
+    /// classifier's cpu signal). A single relaxed atomic add on the same
+    /// buffer the hashing path already owns - zero measurable overhead in
+    /// the hot loop.
+    pub fn add_hashed(&self, n: u64) {
+        self.hashed.fetch_add(n, Ordering::Relaxed);
+    }
+
+    /// Peek all totals. Never resets - samplers diff consecutive snapshots.
     pub fn snapshot(&self) -> IoSnapshot {
         IoSnapshot {
             disk_read_bytes: self.disk_read.load(Ordering::Relaxed),
             net_wire_bytes: self.net_wire.load(Ordering::Relaxed),
+            hashed_bytes: self.hashed.load(Ordering::Relaxed),
         }
     }
 }
@@ -78,15 +96,18 @@ mod tests {
             c.snapshot(),
             IoSnapshot {
                 disk_read_bytes: 0,
-                net_wire_bytes: 0
+                net_wire_bytes: 0,
+                hashed_bytes: 0,
             }
         );
         c.add_disk_read(100);
         c.add_net_wire(40);
         c.add_disk_read(1);
+        c.add_hashed(7);
         let s1 = c.snapshot();
         assert_eq!(s1.disk_read_bytes, 101);
         assert_eq!(s1.net_wire_bytes, 40);
+        assert_eq!(s1.hashed_bytes, 7);
         // Peek-only: a second reader sees the same cumulative totals.
         assert_eq!(c.snapshot(), s1);
     }