diff --git a/README.md b/README.md index 2eec5178..59901897 100644 --- a/README.md +++ b/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 diff --git a/crates/driven-core/src/executor.rs b/crates/driven-core/src/executor.rs index aca57308..b87c1791 100644 --- a/crates/driven-core/src/executor.rs +++ b/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,6 +7296,10 @@ async fn cpu_stage( out_tx: tokio::sync::mpsc::Sender, crypto: Option>, 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>, ) -> Result { use md5::{Digest, Md5}; @@ -7299,13 +7307,18 @@ async fn cpu_stage( 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 = (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 diff --git a/crates/driven-core/src/iostat.rs b/crates/driven-core/src/iostat.rs index d0f4b7bf..a61fefbc 100644 --- a/crates/driven-core/src/iostat.rs +++ b/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); } diff --git a/crates/driven-core/src/orchestrator.rs b/crates/driven-core/src/orchestrator.rs index cc8d7b7f..856c2c8e 100644 --- a/crates/driven-core/src/orchestrator.rs +++ b/crates/driven-core/src/orchestrator.rs @@ -471,6 +471,27 @@ pub trait Orchestrator: Send + Sync { /// [`OrchestratorState`] for the tray / Activity dashboard. async fn state(&self) -> OrchestratorState; + /// Non-blocking snapshot of this account's rate pacer (issue #308 + /// bottleneck classifier): `Some(remaining_ms)` when the pacer is + /// presently inside a backoff window (a rate-limit or circuit-breaker + /// trip), `None` when it is clear or no pacer is wired. Delegates to + /// [`Pacer::backoff_remaining_ms`](crate::pacer::Pacer::backoff_remaining_ms); + /// the default `None` covers pacer-less trait objects (tests / the + /// chaos harness), and [`SyncOrchestrator`] overrides it by reading its + /// shared pacer (wired via [`SyncOrchestrator::with_pacer`]). + fn pacer_backoff_remaining_ms(&self) -> Option { + None + } + + /// Short display label for this account's destination ("Google Drive", + /// "S3", "SFTP", "your local folder"), for the bottleneck classifier's + /// "Drive rate-limited" / "S3 rate-limited" sub-line. Defaults to a + /// generic label so a trait object built without + /// [`SyncOrchestrator::with_backend_label`] still reads sensibly. + fn backend_label(&self) -> &'static str { + "your destination" + } + /// Applies a new [`OrchestratorConfig`], taking effect on the next /// cycle (the `Arc>` swap, SPEC s5). async fn reconfigure(&self, config: OrchestratorConfig); @@ -683,6 +704,18 @@ pub struct SyncOrchestrator { /// exactly the lie this feature exists to prevent. Set via /// [`Self::with_restore_probe`]. restore_probe: Option>, + /// Short destination display label (issue #308 bottleneck classifier's + /// "X rate-limited" sub-line). Set via [`Self::with_backend_label`]; + /// defaults to a generic label when unset (tests / chaos harness). + backend_label: &'static str, + /// App-global disk/net/hash byte counters (issue #308 bottleneck + /// classifier, 2026-08-17 follow-up): threaded into + /// [`crate::scanner::scan_with_priority`] so a deep-verify's blake3 pass + /// credits the SAME counters the upload pipeline's cpu stage does. `None` + /// in tests / the chaos harness (deep-verify hashing then credits + /// nothing - the same degradation as [`Self::latency`]). Set via + /// [`Self::with_io_counters`]. + io_counters: Option>, } impl SyncOrchestrator { @@ -744,6 +777,8 @@ impl SyncOrchestrator { latency: None, adaptive: None, restore_probe: None, + backend_label: "your destination", + io_counters: None, } } @@ -869,6 +904,28 @@ impl SyncOrchestrator { self } + /// Set the destination's short display label for the bottleneck + /// classifier (issue #308). Pass a literal matching the account's + /// `BackendKind` ("Google Drive" / "S3" / "SFTP" / "your local folder"); + /// unset accounts (tests, chaos harness) keep the generic default. + #[must_use] + pub fn with_backend_label(mut self, label: &'static str) -> Self { + self.backend_label = label; + self + } + + /// Share the app-global disk/net/hash byte counters (issue #308 + /// bottleneck classifier). Pass the SAME `Arc` the executor was built + /// with (via `DefaultExecutor::with_io_counters`) so a deep-verify + /// re-hash and an upload's cpu stage credit one counter. Without this + /// call `scan_with_priority` runs with no hash instrumentation (the + /// cpu bottleneck signal then always reads zero). + #[must_use] + pub fn with_io_counters(mut self, io_counters: Arc) -> Self { + self.io_counters = Some(io_counters); + self + } + /// Apply the effective bandwidth cap for the current network to the shared /// pacer (V2 metered throttle, DESIGN s17). On a metered network in /// [`MeteredMode::Throttle`] the metered cap applies; otherwise the normal @@ -2276,6 +2333,10 @@ impl SyncOrchestrator { // foreground apps (the same cell the executor's bundle path // reads). self.priority.get(), + // issue #308: the SAME app-global counters the upload + // pipeline credits, so a deep-verify re-hash moves the + // bottleneck classifier's cpu rate too. + self.io_counters.clone(), ) .await? }; @@ -3615,6 +3676,14 @@ impl Orchestrator for SyncOrchestrator { self.state_machine.read().await.clone() } + fn pacer_backoff_remaining_ms(&self) -> Option { + self.pacer.as_ref()?.backoff_remaining_ms() + } + + fn backend_label(&self) -> &'static str { + self.backend_label + } + async fn reconfigure(&self, config: OrchestratorConfig) { // P1-5 (M3.5 codex): thread the (possibly changed) VSS mode to the // attached provider so `vss_mode = never` actually disables snapshots @@ -4869,6 +4938,86 @@ mod tests { } } + #[test] + fn pacer_backoff_and_backend_label_default_when_unwired() { + // issue #308: a trait object built without `with_pacer` / + // `with_backend_label` (the historical construction, and every test + // fake elsewhere in this crate) must read as "clear" with a generic + // label, never panic or silently misreport a backoff. + let account = AccountId::new_v4(); + let dir = tempfile::tempdir().unwrap(); + let src = source_in(account, dir.path()); + let exec = Arc::new(RecordingExecutor::default()); + let (orch, _clock) = build( + account, + vec![src], + exec, + power_on_ac(), + Arc::new(FakeNet::online()), + OrchestratorConfig::default(), + ); + assert_eq!(orch.pacer_backoff_remaining_ms(), None); + assert_eq!(orch.backend_label(), "your destination"); + } + + #[test] + fn pacer_backoff_delegates_to_the_wired_pacer_when_clear() { + // issue #308: a wired-but-not-throttling pacer (`FakePacer`'s default + // `backoff_remaining_ms`, which it does not override) still reads as + // clear through the orchestrator - `with_pacer` alone must not + // fabricate a backoff. + let account = AccountId::new_v4(); + let dir = tempfile::tempdir().unwrap(); + let src = source_in(account, dir.path()); + let exec = Arc::new(RecordingExecutor::default()); + let (orch, _clock) = build( + account, + vec![src], + exec, + power_on_ac(), + Arc::new(FakeNet::online()), + OrchestratorConfig::default(), + ); + let orch = orch + .with_pacer(Arc::new(FakePacer::default())) + .with_backend_label("Drive"); + assert_eq!(orch.pacer_backoff_remaining_ms(), None); + assert_eq!(orch.backend_label(), "Drive"); + } + + #[test] + fn pacer_backoff_reports_the_live_deadline_from_a_throttled_pacer() { + // issue #308: a REAL AimdPacer that has actually throttled reports its + // live remaining-ms through the orchestrator, non-blocking (no + // `.await`, no clock advance needed to observe it). + let account = AccountId::new_v4(); + let dir = tempfile::tempdir().unwrap(); + let src = source_in(account, dir.path()); + let exec = Arc::new(RecordingExecutor::default()); + let (orch, _clock) = build( + account, + vec![src], + exec, + power_on_ac(), + Arc::new(FakeNet::online()), + OrchestratorConfig::default(), + ); + + let pacer_clock: Arc = Arc::new(FakeClock::new()); + let pacer = crate::pacer::AimdPacer::new(pacer_clock, None); + pacer.note_response(crate::pacer::ResponseClass::RateLimited { + retry_after: std::time::Duration::from_secs(5), + }); + let pacer: Arc = Arc::new(pacer); + let orch = orch.with_pacer(pacer).with_backend_label("S3"); + + let remaining = orch + .pacer_backoff_remaining_ms() + .expect("the pacer is mid-backoff"); + assert!(remaining >= 5_000, "remaining_ms = {remaining}"); + assert_eq!(orch.backend_label(), "S3"); + } + #[test] fn effective_cap_throttles_only_when_metered_and_throttle_mode() { let base = OrchestratorConfig { diff --git a/crates/driven-core/src/pacer.rs b/crates/driven-core/src/pacer.rs index 56ffe6af..959b2397 100644 --- a/crates/driven-core/src/pacer.rs +++ b/crates/driven-core/src/pacer.rs @@ -156,6 +156,17 @@ pub trait Pacer: Send + Sync { fn last_throttle_ms(&self) -> i64 { i64::MIN } + + /// Non-blocking snapshot of the CURRENT backoff window (issue #308 + /// bottleneck classifier): `Some(remaining_ms)` when `permit_request` + /// would presently sleep before proceeding, `None` when the pacer is + /// clear. Never sleeps and never mutates - a plain read against the + /// injected clock, safe to poll every tick from outside the upload path. + /// The default `None` makes a simple/fake pacer read as "not backing + /// off"; [`AimdPacer`] overrides it with the real deadline. + fn backoff_remaining_ms(&self) -> Option { + None + } } /// `serde` helper: (de)serialise a [`Duration`] as integer milliseconds so @@ -512,6 +523,21 @@ impl AimdPacer { } } + /// Non-blocking snapshot of the current backoff window (issue #308 + /// bottleneck classifier): `Some(remaining_ms)` when `permit_request` + /// would presently sleep before proceeding, `None` when the pacer is + /// clear. Never sleeps and never mutates - a plain read of the atomic + /// deadline against the injected clock. `backoff_until_ms` itself stays + /// private; this is the read-only surface for callers outside the pacer + /// (the [`Pacer::backoff_remaining_ms`] trait method delegates here so a + /// caller holding only `Arc` can poll it too). + pub fn backoff_remaining_ms(&self) -> Option { + let until = self.backoff_until_ms.load(Ordering::Acquire); + let now = self.clock.now_ms(); + let remaining = until.saturating_sub(now); + (remaining > 0).then_some(remaining) + } + /// Applies the additive increase if a full clean window has accrued /// (DESIGN s18.1: +5 qps, +1/s file-create per 10 clean minutes, capped /// at the hard cap). Called on each clean (`Ok`) response. Multiple @@ -703,6 +729,10 @@ impl Pacer for AimdPacer { fn last_throttle_ms(&self) -> i64 { self.last_throttle_ms.load(Ordering::Acquire) } + + fn backoff_remaining_ms(&self) -> Option { + AimdPacer::backoff_remaining_ms(self) + } } /// Applies a backoff with jitter to Drive's `Retry-After` (DESIGN s5.4: @@ -843,6 +873,30 @@ mod tests { ); } + #[test] + fn backoff_remaining_ms_reports_none_when_clear_and_some_while_throttled() { + let (fake, c) = clock(); + let pacer = AimdPacer::new(c, None); + // No throttle yet: the pacer reads as clear. + assert_eq!(pacer.backoff_remaining_ms(), None); + assert_eq!(Pacer::backoff_remaining_ms(&pacer), None); + + pacer.note_response(ResponseClass::RateLimited { + retry_after: Duration::from_secs(2), + }); + // A non-blocking read (no `.await`, no clock advance) sees the live + // window: at least the 2s floor, never blocking the caller. + let remaining = pacer.backoff_remaining_ms().expect("backoff active"); + assert!(remaining >= 2_000, "remaining_ms = {remaining}"); + // Reachable through the trait object too (the orchestrator only ever + // holds `Arc`). + assert_eq!(Pacer::backoff_remaining_ms(&pacer), Some(remaining)); + + // Advancing the clock past the deadline clears it again. + fake.advance(Duration::from_secs(3)); + assert_eq!(pacer.backoff_remaining_ms(), None); + } + #[tokio::test] async fn repeated_rate_limits_floor_at_one() { let (_fake, c) = clock(); diff --git a/crates/driven-core/src/scanner.rs b/crates/driven-core/src/scanner.rs index bb43a345..f4a7b147 100644 --- a/crates/driven-core/src/scanner.rs +++ b/crates/driven-core/src/scanner.rs @@ -344,6 +344,12 @@ struct WalkCtx { /// walk (spawned and joined inside `build_parallel().visit()`), so a /// one-shot apply with no restore is correct - nothing pooled is demoted. priority: WorkPriority, + /// App-global disk/net/hash byte counters (issue #308 bottleneck + /// classifier), or `None` when not wired (tests / chaos harness). A + /// deep-verify (or coarse-fs-fallback) re-hash credits the file's size + /// here so the bottleneck classifier's cpu rate reflects scan-time + /// hashing, not just the upload pipeline's. + io_counters: Option>, } /// What a worker concluded about ONE file. Deliberately excludes every @@ -665,7 +671,18 @@ fn process_file(ctx: &WalkCtx, entry: &DirEntry, rel: RelativePath) -> FileRecor ); if ctx.mode == ScanMode::DeepVerify || coarse_suspect { let stored_hash = stored.map(|r| r.hash_blake3); - match hash_file(abs, ctx.priority) { + let hash_result = hash_file(abs, ctx.priority); + // issue #308: credit the WHOLE file's bytes on a successful hash, + // win or lose the comparison below - the bottleneck classifier + // cares that hashing happened, not what it concluded. A failed + // read (handled in the `Err` arm) credits nothing, matching the + // "no evidence" treatment the rest of this branch already gives it. + if hash_result.is_ok() { + if let Some(io) = ctx.io_counters.as_ref() { + io.add_hashed(size); + } + } + match hash_result { Ok(hash) if stored_hash != Some(hash) => { let reason = if ctx.mode == ScanMode::DeepVerify { "deep-verify" @@ -718,6 +735,7 @@ pub async fn scan_with_progress( latency, on_progress, WorkPriority::Normal, + None, ) .await } @@ -737,6 +755,11 @@ pub async fn scan_with_priority( latency: Option<&crate::telemetry::LatencyReservoir>, on_progress: Option<&ScanProgressSink<'_>>, priority: WorkPriority, + // issue #308: the app-global disk/net/hash byte counters, so a + // deep-verify (or coarse-fs fallback) re-hash credits the SAME hash + // counter the upload pipeline's cpu stage does. `None` in tests / the + // chaos harness - the walk still runs, it just credits nothing. + io_counters: Option>, ) -> anyhow::Result { let known = state .load_source_file_state(source.id) @@ -889,6 +912,7 @@ pub async fn scan_with_priority( last_scan_end_ns, capture_latency, priority, + io_counters, }); let mut wb = build_walker_with_matcher(&walk_source, Arc::clone(&matcher)); wb.threads(walk_threads()); @@ -2056,6 +2080,51 @@ mod tests { assert!(res.new_or_changed.is_empty(), "{:?}", res.new_or_changed); } + /// issue #308 bottleneck classifier: a deep-verify re-hash credits the + /// app-global hash counter with the WHOLE file's bytes, whether the + /// content turns out unchanged (this test) or changed (the mismatch + /// path re-hashes exactly the same way before it can tell the + /// difference). `scan_with_priority` is called directly (rather than the + /// `scan`/`scan_with_progress` wrappers, which always pass `None`) since + /// only it threads the counters through. + #[tokio::test] + async fn deep_verify_credits_the_hash_counter() { + let dir = tempfile::tempdir().unwrap(); + let root = dir.path(); + let p = root.join("a.txt"); + write(&p, b"hello"); + + let src = source_at(root); + let state = FakeState::default(); + let (size, mtime) = stat_of(&p); + state.put(row( + src.id, + "a.txt", + size, + mtime, + *blake3::hash(b"hello").as_bytes(), + )); + + let io = Arc::new(crate::iostat::IoCounters::default()); + let res = scan_with_priority( + &src, + &state, + ScanMode::DeepVerify, + None, + None, + WorkPriority::Normal, + Some(io.clone()), + ) + .await + .unwrap(); + assert!(res.new_or_changed.is_empty(), "{:?}", res.new_or_changed); + assert_eq!( + io.snapshot().hashed_bytes, + size, + "the whole file was hashed" + ); + } + /// Issue #35 item e: a BUNDLED member - a `file_state` row with /// `drive_file_id = NULL` (its bytes live inside a `.tar.gz` bundle), status /// Synced, and a matching stored hash - must NOT be re-emitted as changed by a diff --git a/src-tauri/src/app_state.rs b/src-tauri/src/app_state.rs index 5bf61c49..a08e3564 100644 --- a/src-tauri/src/app_state.rs +++ b/src-tauri/src/app_state.rs @@ -305,6 +305,9 @@ pub struct AppState { updater: UpdaterRuntime, /// 2026-08-14 follow-up: live disk/network throughput sampling runtime. iostat: IostatRuntime, + /// issue #308 (2026-08-17 follow-up): live bottleneck-classification + /// sampling runtime (the Activity dashboard's Bottleneck stat tile). + bottleneck: BottleneckRuntime, /// The ONE in-flight streaming exclusion preview /// ([`crate::commands::exclusion_stream`]). The exclusion editor re-previews /// on every rule edit, so without a single-slot registry a user tweaking @@ -418,6 +421,25 @@ pub struct IostatRuntime { shutdown: std::sync::Mutex>>, } +/// issue #308 (2026-08-17 follow-up): the live bottleneck-classification +/// runtime held on [`AppState`] - the latest-snapshot hub plus the sampler +/// task's lifecycle slots (mirrors [`IostatRuntime`], which this sampler +/// reads from). Unlike `IostatRuntime` there is nothing to "install" from +/// assembly: the hub reads the app-global IO counters and the accounts map +/// straight off `AppState` each tick, so the default hub is already correct +/// even in the quiesced boot path (it just classifies `NotBackingUp`). +#[derive(Default)] +pub struct BottleneckRuntime { + /// The latest-snapshot hub the sampler pushes into and the + /// `bottleneck_status` command reads. + hub: Arc, + /// The spawned sampler task, behind `Option` so the shutdown drain can + /// TAKE + await it by value; `None` once drained / never spawned. + task: std::sync::Mutex>>, + /// The shutdown signal the sampler `select!`s on. + shutdown: std::sync::Mutex>>, +} + /// M9b (SPEC s16): the anonymous-telemetry runtime state held on [`AppState`]. /// /// `task` + `shutdown` track the single app-wide periodic-ping task so the quit @@ -679,6 +701,7 @@ impl AppState { restore_jobs: std::sync::Mutex::new(HashMap::new()), updater: UpdaterRuntime::default(), iostat: IostatRuntime::default(), + bottleneck: BottleneckRuntime::default(), exclusion_previews: Arc::default(), preview_tree_cache: Arc::default(), telemetry: TelemetryRuntime::default(), @@ -1037,6 +1060,50 @@ impl AppState { .take() } + // --- issue #308: bottleneck-classification runtime ----------------------- + + /// The live bottleneck-classification hub the sampler pushes into and the + /// `bottleneck_status` command reads. Always available (no "install" + /// step needed - it reads `AppState` directly each tick). + pub fn bottleneck_hub(&self) -> Arc { + self.bottleneck.hub.clone() + } + + /// Register the spawned sampler task + its shutdown sender so the app-quit + /// drain can stop + join it with no orphan (mirrors [`Self::set_iostat_task`]). + pub fn set_bottleneck_task(&self, task: JoinHandle<()>, shutdown: watch::Sender) { + *self + .bottleneck + .task + .lock() + .unwrap_or_else(|e| e.into_inner()) = Some(task); + *self + .bottleneck + .shutdown + .lock() + .unwrap_or_else(|e| e.into_inner()) = Some(shutdown); + } + + /// Signal the sampler to stop and TAKE its handle so the quit drain can + /// await it. Mirrors [`Self::shutdown_iostat_task`]. + #[must_use] + pub fn shutdown_bottleneck_task(&self) -> Option> { + if let Some(tx) = self + .bottleneck + .shutdown + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take() + { + let _ = tx.send(true); + } + self.bottleneck + .task + .lock() + .unwrap_or_else(|e| e.into_inner()) + .take() + } + // --- M9b telemetry runtime (SPEC s16) ---------------------------------- /// M9b: register the spawned periodic-ping task + its shutdown sender so the @@ -1995,6 +2062,63 @@ pub(crate) mod tests { let _ = std::fs::remove_dir_all(dir); } + #[test] + fn fake_orchestrator_inherits_the_orchestrator_traits_default_bottleneck_methods() { + // issue #308: `FakeOrchestrator` deliberately does not override + // `pacer_backoff_remaining_ms` / `backend_label` - it exists to + // exercise the `Orchestrator` trait's DEFAULT bodies (a pacer-less / + // label-less trait object must read as clear with a generic label, + // never panic), the same contract the bottleneck sampler leans on + // for any account whose orchestrator was built without those seams. + let orch = FakeOrchestrator::new(); + assert_eq!(orch.pacer_backoff_remaining_ms(), None); + assert_eq!(orch.backend_label(), "your destination"); + } + + #[tokio::test] + async fn bottleneck_runtime_hub_task_and_shutdown_round_trip() { + // issue #308: the bottleneck sampler's runtime bookkeeping. Unlike + // `IostatRuntime` there is no "install" step - the hub is always + // available, defaulting to `NotBackingUp` - so this only needs to + // cover the getter plus the set/shutdown task pair (mirrors + // `set_iostat_task`/`shutdown_iostat_task`'s no-orphan drain). + let (state, dir) = temp_state().await; + let app_state = AppState::new( + state, + HashMap::new(), + RemoteMode::Fake, + default_fake_registry(), + ); + + let hub = app_state.bottleneck_hub(); + assert_eq!( + hub.latest().state, + crate::bottleneck_hub::BottleneckState::NotBackingUp + ); + + // No task registered yet: shutdown is a safe no-op. + assert!(app_state.shutdown_bottleneck_task().is_none()); + + // Register a task that exits promptly on the shutdown signal (the + // real sampler's own shape), then confirm shutdown signals + hands + // back the handle so the quit drain can join it. + let (shutdown_tx, mut shutdown_rx) = watch::channel(false); + let task = tokio::spawn(async move { + let _ = shutdown_rx.changed().await; + }); + app_state.set_bottleneck_task(task, shutdown_tx); + + let handle = app_state + .shutdown_bottleneck_task() + .expect("the just-registered task round-trips"); + handle.await.unwrap(); + + // Taken: a second shutdown call is again a safe no-op. + assert!(app_state.shutdown_bottleneck_task().is_none()); + + let _ = std::fs::remove_dir_all(dir); + } + #[tokio::test] async fn vss_helper_manager_installs_and_shutdown_is_noop() { // Issue #25: AppState owns the least-privilege VSS helper broker manager - diff --git a/src-tauri/src/assembly.rs b/src-tauri/src/assembly.rs index 15957aa5..aa315bbc 100644 --- a/src-tauri/src/assembly.rs +++ b/src-tauri/src/assembly.rs @@ -658,6 +658,13 @@ async fn build_account( // SPEC s22 `io_priority`: the SAME cell the executor reads, so a settings // save reaches the backup threads on the next piece of work. orchestrator = orchestrator.with_priority_cell(priority); + // issue #308 bottleneck classifier: the destination's short display + // label ("Drive rate-limited...") and the SAME app-global disk/net/hash + // counters the executor was built with, so a deep-verify re-hash credits + // the same cpu-rate signal an upload's cpu stage does. + orchestrator = orchestrator + .with_backend_label(bottleneck_backend_label(account.backend_kind)) + .with_io_counters(io_counters.clone()); // DESIGN s13: the SAME reservoir the executor holds, so per-file scan latency // and upload-per-MB latency feed one app-global sampler. orchestrator = orchestrator.with_latency_reservoir(latency.clone()); @@ -819,6 +826,19 @@ pub(crate) fn account_backend(account: &AccountRow) -> driven_backend::AccountBa } } +/// Short destination display label for the bottleneck classifier's "X +/// rate-limited" sub-line (issue #308). Matches the wording the mockup used +/// ("Drive rate-limited...") rather than the longer `BackendKind::id()` wire +/// identifiers. +fn bottleneck_backend_label(kind: driven_remote::BackendKind) -> &'static str { + match kind { + driven_remote::BackendKind::GoogleDrive => "Drive", + driven_remote::BackendKind::S3 => "S3", + driven_remote::BackendKind::LocalFolder => "your local folder", + driven_remote::BackendKind::Sftp => "SFTP", + } +} + /// Emit the `account:needs_reauth` webview banner + raise the OS notification /// for an account that requires re-consent at assembly time (C5-P1-1 / /// C5-P1-2). Mirrors the orchestrator-event bridge's reauth handling for the @@ -1399,13 +1419,42 @@ struct SourceProgressEvent { #[cfg(test)] mod tests { - use super::{classify_bridge_event, BridgeAction}; + use super::{bottleneck_backend_label, classify_bridge_event, BridgeAction}; use driven_core::orchestrator::OrchestratorConfig; use driven_core::state::sqlite::SqliteStateRepo; use driven_core::state::StateRepo; use driven_core::types::{AccountId, ActivityEntry, ExecProgress, OrchestratorEvent, SourceId}; use tokio::sync::broadcast::error::RecvError; + /// issue #308: every `BackendKind` variant maps to a distinct, non-empty + /// display label for the bottleneck classifier's "X rate-limited" + /// sub-line, matching the mockup's short-form wording ("Drive", not + /// "Google Drive"). + #[test] + fn bottleneck_backend_label_covers_every_backend_kind_distinctly() { + use driven_remote::BackendKind; + let labels: Vec<&'static str> = BackendKind::ALL + .iter() + .map(|&kind| bottleneck_backend_label(kind)) + .collect(); + assert_eq!(bottleneck_backend_label(BackendKind::GoogleDrive), "Drive"); + assert_eq!(bottleneck_backend_label(BackendKind::S3), "S3"); + assert_eq!( + bottleneck_backend_label(BackendKind::LocalFolder), + "your local folder" + ); + assert_eq!(bottleneck_backend_label(BackendKind::Sftp), "SFTP"); + for label in &labels { + assert!(!label.is_empty()); + } + let unique: std::collections::HashSet<_> = labels.iter().collect(); + assert_eq!( + unique.len(), + labels.len(), + "every backend gets its own label" + ); + } + /// M7-P1-1: a broadcast `Lagged` MUST classify as an `ActivityReconcile` /// (carrying the dropped count) so the bridge emits `activity:lagged` and the /// webview reconciles the dropped rows from the durable `activity_log` - diff --git a/src-tauri/src/bottleneck_hub.rs b/src-tauri/src/bottleneck_hub.rs new file mode 100644 index 00000000..b9cfad4b --- /dev/null +++ b/src-tauri/src/bottleneck_hub.rs @@ -0,0 +1,728 @@ +//! Live bottleneck classification behind the Activity dashboard's Bottleneck +//! stat tile (issue #308, 2026-08-17 follow-up). +//! +//! Every second the sampler task diffs the SAME app-global disk/net/hash byte +//! counters [`crate::iostat_hub`] diffs for the throughput graphs, polls every +//! account's orchestrator for its current state + rate-pacer backoff, runs +//! the pure [`classify`] heuristic over the combined signals, and pushes the +//! result to the webview as a live `sync:bottleneck` event. The +//! `bottleneck_status` command hydrates the initial paint from the latest +//! snapshot. Idle-suppressed the same way `iostat_hub` is: once the state +//! settles on `NotBackingUp` further identical ticks are recorded but not +//! re-emitted, so a fully idle app costs nothing on the IPC bridge. + +use std::sync::Mutex; +use std::time::Duration; + +use driven_core::time::{Clock, SystemClock}; +use driven_core::types::OrchestratorState; +use serde::Serialize; +use tauri::AppHandle; + +const TARGET: &str = "driven::app::bottleneck"; + +/// Sampling cadence (matches [`crate::iostat_hub::SAMPLE_INTERVAL`]). +pub const SAMPLE_INTERVAL: Duration = Duration::from_secs(1); + +/// Below this per-second rate a pipeline stage reads as "not really moving" - +/// keeps a single stray byte from tipping the classification, and keeps a +/// genuinely idle stage out of the dominance comparison entirely. +const IDLE_FLOOR_BYTES_PER_SEC: u64 = 32 * 1024; + +/// A stage's rate must trail the FASTEST active stage by at least this +/// multiple to be called a clear bottleneck; short of it there is no clear +/// limiter (`Mixed`) - the pipeline is flowing at one shared rate. +const DOMINANCE_RATIO: f64 = 1.5; + +/// Which stage of the backup pipeline is presently limiting throughput (the +/// six states the mockup's Bottleneck tile shows). +#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum BottleneckState { + /// No account is mid-cycle. + NotBackingUp, + /// Local disk reads are the slowest active stage. + Disk, + /// Network wire acceptance is the slowest active stage. + Network, + /// A rate pacer is presently backing off (rate-limit / circuit-breaker + /// trip) on an account that is mid-cycle. + Api, + /// Blake3 hashing is the slowest active stage. + Cpu, + /// Multiple stages are active with no one clearly trailing the others. + Mixed, +} + +/// One classified snapshot - the `sync:bottleneck` event and +/// `bottleneck_status` command payload (camelCase on the wire). +#[derive(Debug, Clone, Serialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct BottleneckSnapshot { + /// Wall-clock ms this snapshot was classified. + pub ts_ms: i64, + pub state: BottleneckState, + /// The saturated stage's rate in bytes/sec, present only when `state` + /// names a rate-bearing stage (Disk/Network/Cpu). + pub rate_bytes_per_sec: Option, + /// The rate-limited destination's short display label ("Drive", "S3", + /// ...), present only when `state == Api`. + pub backend: Option, + /// Ms remaining in the active backoff window, present only when `state + /// == Api`. + pub backoff_remaining_ms: Option, +} + +impl BottleneckSnapshot { + fn not_backing_up(ts_ms: i64) -> Self { + Self { + ts_ms, + state: BottleneckState::NotBackingUp, + rate_bytes_per_sec: None, + backend: None, + backoff_remaining_ms: None, + } + } +} + +/// Raw per-tick signals the pure [`classify`] function reasons over - the +/// unit-test seam, independent of AppState/orchestrator plumbing. +#[derive(Debug, Clone, PartialEq)] +pub struct BottleneckSignals { + /// Whether ANY account is presently mid-cycle: `Scanning` / `Planning` / + /// `Executing` / `Verifying` / `Recovering` / `PowerCheck`, OR backing off + /// (a `Backoff` state, or a mid-cycle account whose rate pacer is + /// presently throttled, both count as "still working, just paced"). + /// `Idle` / `Paused` / `Error` accounts do not count. + pub active_cycle: bool, + /// The most-throttled mid-cycle account's backoff, if any: `(backend + /// label, remaining ms)`. Aggregated across accounts by picking the + /// longest remaining window, so the tile never undersells how long the + /// app will stay paced. + pub backoff: Option<(String, i64)>, + /// Plaintext bytes read from local files, per second. + pub disk_bytes_per_sec: u64, + /// Wire bytes the destination accepted, per second. + pub net_bytes_per_sec: u64, + /// Plaintext bytes blake3-hashed, per second. + pub hash_bytes_per_sec: u64, +} + +/// The pure classification heuristic (issue #308), unit-tested over all six +/// states plus the boundary cases around [`IDLE_FLOOR_BYTES_PER_SEC`] and +/// [`DOMINANCE_RATIO`]. A pure function of one snapshot: calling it twice +/// with the same signals always yields the same state, so any flap-smoothing +/// (debounce, hysteresis) is entirely the caller's problem - here the +/// frontend Pinia store, which is where the spec puts it. +/// +/// 1. No account mid-cycle => [`BottleneckState::NotBackingUp`]. +/// 2. A mid-cycle account's rate pacer is presently backing off (or the +/// account itself is in the orchestrator-level `Backoff` state, a +/// circuit-breaker/rate-limit trip) => [`BottleneckState::Api`]. Checked +/// before the rate comparison because a paced account can otherwise show +/// misleadingly "healthy" disk/net/hash rates between backoff-gated +/// requests. +/// 3. Otherwise compare the three per-second rates (disk read, net wire +/// accepted, blake3-hashed): stages at or below the idle floor are +/// dropped from consideration entirely (a scan phase with no upload +/// traffic yet should not make Network read as "the bottleneck at +/// 0 B/s"). Among the remaining ACTIVE stages, the slowest is the +/// reported bottleneck, provided the fastest active stage clears it by at +/// least [`DOMINANCE_RATIO`] - a pipeline flowing at one shared rate (the +/// common case: every stage paced by the same backpressure) has no clear +/// winner and reports [`BottleneckState::Mixed`] instead. Zero active +/// stages during a mid-cycle account (e.g. the `Planning` phase, which +/// moves no bytes) is also `Mixed` - work is happening, but nothing +/// currently measurable is the limiter. +pub fn classify(signals: &BottleneckSignals) -> (BottleneckState, Option) { + if !signals.active_cycle { + return (BottleneckState::NotBackingUp, None); + } + if signals.backoff.is_some() { + return (BottleneckState::Api, None); + } + + let stages = [ + (BottleneckState::Disk, signals.disk_bytes_per_sec), + (BottleneckState::Network, signals.net_bytes_per_sec), + (BottleneckState::Cpu, signals.hash_bytes_per_sec), + ]; + let active: Vec<(BottleneckState, u64)> = stages + .into_iter() + .filter(|&(_, rate)| rate > IDLE_FLOOR_BYTES_PER_SEC) + .collect(); + + match active.len() { + 0 => (BottleneckState::Mixed, None), + 1 => (active[0].0, Some(active[0].1)), + _ => { + // `active` has >= 2 distinct stages (Disk/Network/Cpu can never + // repeat), so both `min_by_key`/`max` below always find a value. + let (slow_state, slow_rate) = *active.iter().min_by_key(|&&(_, r)| r).unwrap(); + let fastest = active.iter().map(|&(_, r)| r).max().unwrap(); + if (fastest as f64) >= (slow_rate as f64) * DOMINANCE_RATIO { + (slow_state, Some(slow_rate)) + } else { + (BottleneckState::Mixed, None) + } + } + } +} + +/// Whether an [`OrchestratorState`] counts as "this account is mid-cycle" +/// for [`BottleneckSignals::active_cycle`] - everything except the three +/// at-rest states. +fn is_active(state: &OrchestratorState) -> bool { + !matches!( + state, + OrchestratorState::Idle { .. } + | OrchestratorState::Paused { .. } + | OrchestratorState::Error { .. } + ) +} + +/// Bytes/sec from a cumulative-counter delta over `elapsed_ms`. Guards +/// against a zero/negative elapsed (clock oddities) by flooring at 1ms so +/// this can never divide by zero or return a spurious infinite rate. +fn rate_bytes_per_sec(cur: u64, prev: u64, elapsed_ms: i64) -> u64 { + let delta = cur.saturating_sub(prev); + let elapsed = elapsed_ms.max(1) as u128; + ((delta as u128 * 1000) / elapsed) as u64 +} + +/// The latest classified snapshot, held for the `bottleneck_status` command's +/// hydration read. No ring/history - the frontend store owns any windowing +/// it wants for hysteresis. +#[derive(Debug)] +pub struct BottleneckHub { + latest: Mutex, +} + +impl Default for BottleneckHub { + fn default() -> Self { + Self { + latest: Mutex::new(BottleneckSnapshot::not_backing_up(0)), + } + } +} + +impl BottleneckHub { + /// The latest classification, for the hydration command. Always + /// available (defaults to `NotBackingUp` before the first tick). + pub fn latest(&self) -> BottleneckSnapshot { + self.latest + .lock() + .map(|g| g.clone()) + .unwrap_or_else(|e| e.into_inner().clone()) + } + + fn push(&self, snapshot: BottleneckSnapshot) { + if let Ok(mut g) = self.latest.lock() { + *g = snapshot; + } + } +} + +/// One account's contribution to the aggregate signals: whether it is +/// mid-cycle, and (if so) its resolved backoff, if any. Pure - the async +/// orchestrator polling (`state().await`, `pacer_backoff_remaining_ms()`, +/// `backend_label()`) lives in `tick_once`; this is the unit-test seam for +/// the DECISION those reads feed into. +/// +/// The account-level circuit-breaker/rate-limit trip (`Backoff { until }`) +/// carries its own deadline; a mid-cycle account otherwise defers to its +/// rate pacer's live backoff window (`pacer_backoff_remaining_ms`, issue +/// #308's primary signal) - passed in already resolved, since only the +/// caller has an `Arc` to poll. +fn account_signal( + state: &OrchestratorState, + now_ms: i64, + pacer_backoff_remaining_ms: Option, + backend_label: &str, +) -> (bool, Option<(String, i64)>) { + let active = is_active(state); + let remaining_ms = match state { + OrchestratorState::Backoff { until } => Some((*until - now_ms).max(0)), + _ if active => pacer_backoff_remaining_ms, + _ => None, + }; + let backoff = remaining_ms.map(|r| (backend_label.to_string(), r)); + (active, backoff) +} + +/// Fold every account's [`account_signal`] result into the aggregate +/// `(active_cycle, backoff)` pair [`BottleneckSignals`] carries: ANY account +/// active makes the whole app active, and the LONGEST remaining backoff wins +/// (so the tile never undersells how long the app will stay paced). Pure. +fn aggregate_accounts( + per_account: &[(bool, Option<(String, i64)>)], +) -> (bool, Option<(String, i64)>) { + let active_cycle = per_account.iter().any(|(active, _)| *active); + let backoff = per_account + .iter() + .filter_map(|(_, backoff)| backoff.clone()) + .max_by_key(|(_, remaining_ms)| *remaining_ms); + (active_cycle, backoff) +} + +/// Assemble the `sync:bottleneck` / `bottleneck_status` payload from a +/// classification result. Pure: the Api-only fields (`backend`, +/// `backoff_remaining_ms`) are gated on `state` rather than merely on +/// `backoff.is_some()`, so a future classifier bug that returns a non-`Api` +/// state alongside a backoff can never leak the backoff fields onto the +/// wrong tile reading. +fn build_snapshot( + now_ms: i64, + state: BottleneckState, + rate_bytes_per_sec: Option, + backoff: Option<(String, i64)>, +) -> BottleneckSnapshot { + let is_api = state == BottleneckState::Api; + BottleneckSnapshot { + ts_ms: now_ms, + state, + rate_bytes_per_sec, + backend: if is_api { + backoff.as_ref().map(|(b, _)| b.clone()) + } else { + None + }, + backoff_remaining_ms: if is_api { + backoff.as_ref().map(|(_, r)| *r) + } else { + None + }, + } +} + +/// Whether this tick's classification should be broadcast on `sync:bottleneck` +/// (idle suppression, mirrors `iostat_hub::tick`): once a `NotBackingUp` has +/// been emitted, further identical ticks are recorded (so a late +/// `bottleneck_status` hydration is still correct) but not re-broadcast - an +/// idle app costs nothing on the IPC bridge. Pure. +fn should_emit(state: BottleneckState, last_emitted: Option) -> bool { + !(state == BottleneckState::NotBackingUp && last_emitted == Some(state)) +} + +/// One sampler tick: reads the app-global IO counters + every account's +/// orchestrator, classifies, records the snapshot on `hub`, and - unless +/// idle-suppressed - emits `sync:bottleneck`. Thin by design: every actual +/// decision (per-account resolution, aggregation, DTO assembly, the emit +/// gate) lives in a pure helper above with its own unit tests; this function +/// is just the AppHandle/AppState plumbing those helpers cannot see without +/// a real app. +async fn tick_once( + app: &AppHandle, + hub: &BottleneckHub, + prev_io: &mut driven_core::iostat::IoSnapshot, + prev_ms: &mut i64, + last_emitted: &mut Option, +) { + use tauri::Manager; + let Some(state) = app.try_state::() else { + return; + }; + + let now_ms = SystemClock.now_ms(); + let elapsed_ms = now_ms - *prev_ms; + *prev_ms = now_ms; + + let cur_io = state.iostat_hub().counters().snapshot(); + let disk_bytes_per_sec = + rate_bytes_per_sec(cur_io.disk_read_bytes, prev_io.disk_read_bytes, elapsed_ms); + let net_bytes_per_sec = + rate_bytes_per_sec(cur_io.net_wire_bytes, prev_io.net_wire_bytes, elapsed_ms); + let hash_bytes_per_sec = + rate_bytes_per_sec(cur_io.hashed_bytes, prev_io.hashed_bytes, elapsed_ms); + *prev_io = cur_io; + + let mut per_account = Vec::new(); + for (_, handle) in state.accounts() { + let account_state = handle.orchestrator.state().await; + per_account.push(account_signal( + &account_state, + now_ms, + handle.orchestrator.pacer_backoff_remaining_ms(), + handle.orchestrator.backend_label(), + )); + } + let (active_cycle, chosen_backoff) = aggregate_accounts(&per_account); + + let signals = BottleneckSignals { + active_cycle, + backoff: chosen_backoff.clone(), + disk_bytes_per_sec, + net_bytes_per_sec, + hash_bytes_per_sec, + }; + let (bottleneck_state, rate_bytes_per_sec) = classify(&signals); + let snapshot = build_snapshot(now_ms, bottleneck_state, rate_bytes_per_sec, chosen_backoff); + hub.push(snapshot.clone()); + + if should_emit(bottleneck_state, *last_emitted) { + *last_emitted = Some(bottleneck_state); + crate::events::emit_sync_bottleneck(app, snapshot); + } +} + +/// Spawn the sampling loop (updater/telemetry lifecycle pattern; see +/// [`crate::iostat_hub::spawn_sampler`]). Called once from setup after +/// `AppState` is managed. +pub fn spawn_sampler(app: &AppHandle) { + use tauri::Manager; + let Some(state) = app.try_state::() else { + tracing::warn!(target: TARGET, "AppState not managed; bottleneck sampler not started"); + return; + }; + let hub = state.bottleneck_hub(); + let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false); + let app_handle = app.clone(); + + let task = tokio::spawn(async move { + let mut ticker = tokio::time::interval(SAMPLE_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut prev_io = driven_core::iostat::IoSnapshot { + disk_read_bytes: 0, + net_wire_bytes: 0, + hashed_bytes: 0, + }; + let mut prev_ms = SystemClock.now_ms(); + let mut last_emitted: Option = None; + loop { + tokio::select! { + biased; + res = shutdown_rx.changed() => { + match res { + Ok(()) if *shutdown_rx.borrow() => break, + Ok(()) => {} + Err(_) => break, + } + } + _ = ticker.tick() => { + tick_once(&app_handle, &hub, &mut prev_io, &mut prev_ms, &mut last_emitted).await; + } + } + } + tracing::debug!(target: TARGET, "bottleneck sampler exited"); + }); + + state.set_bottleneck_task(task, shutdown_tx); + tracing::info!( + target: TARGET, + interval_ms = SAMPLE_INTERVAL.as_millis() as u64, + "bottleneck sampler started" + ); +} + +#[cfg(test)] +mod tests { + use super::*; + + fn signals( + active_cycle: bool, + backoff: Option<(&str, i64)>, + disk: u64, + net: u64, + hash: u64, + ) -> BottleneckSignals { + BottleneckSignals { + active_cycle, + backoff: backoff.map(|(b, r)| (b.to_string(), r)), + disk_bytes_per_sec: disk, + net_bytes_per_sec: net, + hash_bytes_per_sec: hash, + } + } + + #[test] + fn not_backing_up_when_no_account_is_mid_cycle() { + let (state, rate) = classify(&signals(false, None, 999_999, 999_999, 999_999)); + assert_eq!(state, BottleneckState::NotBackingUp); + assert_eq!(rate, None); + } + + #[test] + fn api_wins_over_any_rate_reading_while_a_pacer_backs_off() { + let (state, rate) = classify(&signals( + true, + Some(("Drive", 8_000)), + 500_000, + 500_000, + 500_000, + )); + assert_eq!(state, BottleneckState::Api); + // Api carries no rate in the classifier's own output - the backend + + // remaining ms come from `signals.backoff`, surfaced by the caller. + assert_eq!(rate, None); + } + + #[test] + fn disk_is_the_clear_bottleneck_when_it_trails_the_others() { + let (state, rate) = classify(&signals(true, None, 100_000, 400_000, 400_000)); + assert_eq!(state, BottleneckState::Disk); + assert_eq!(rate, Some(100_000)); + } + + #[test] + fn network_is_the_clear_bottleneck_when_it_trails_the_others() { + let (state, rate) = classify(&signals(true, None, 400_000, 100_000, 400_000)); + assert_eq!(state, BottleneckState::Network); + assert_eq!(rate, Some(100_000)); + } + + #[test] + fn cpu_is_the_clear_bottleneck_when_it_trails_the_others() { + let (state, rate) = classify(&signals(true, None, 400_000, 400_000, 100_000)); + assert_eq!(state, BottleneckState::Cpu); + assert_eq!(rate, Some(100_000)); + } + + #[test] + fn cpu_alone_active_reads_as_cpu_even_with_no_rival_stage() { + // A deep-verify scan hashing files with nothing queued to upload yet: + // only the hash counter is moving. + let (state, rate) = classify(&signals(true, None, 0, 0, 900_000)); + assert_eq!(state, BottleneckState::Cpu); + assert_eq!(rate, Some(900_000)); + } + + #[test] + fn mixed_when_all_active_stages_run_at_one_shared_rate() { + let (state, rate) = classify(&signals(true, None, 300_000, 310_000, 305_000)); + assert_eq!(state, BottleneckState::Mixed); + assert_eq!(rate, None); + } + + #[test] + fn mixed_when_mid_cycle_but_nothing_measurable_is_moving() { + // e.g. the `Planning` phase: an active cycle, but no bytes moved yet. + let (state, rate) = classify(&signals(true, None, 0, 0, 0)); + assert_eq!(state, BottleneckState::Mixed); + assert_eq!(rate, None); + } + + #[test] + fn a_stage_at_or_under_the_idle_floor_is_dropped_from_consideration() { + // Net sits right at the idle floor - not "active" - so with only + // disk+hash left active and equal, this is Mixed, not Network. + let (state, _) = classify(&signals( + true, + None, + 300_000, + IDLE_FLOOR_BYTES_PER_SEC, + 300_000, + )); + assert_eq!(state, BottleneckState::Mixed); + } + + #[test] + fn dominance_ratio_boundary_is_inclusive() { + // Fastest exactly DOMINANCE_RATIO x the slowest: still a clear call. + let slow = 100_000u64; + let fast = (slow as f64 * DOMINANCE_RATIO) as u64; + let (state, _) = classify(&signals(true, None, slow, fast, fast)); + assert_eq!(state, BottleneckState::Disk); + + // One byte short of the ratio: no longer a clear call. + let (state, _) = classify(&signals(true, None, slow, fast - 1, fast - 1)); + assert_eq!(state, BottleneckState::Mixed); + } + + #[test] + fn classify_is_pure_and_hysteresis_friendly() { + // Calling it twice with the same signals is idempotent - any + // flap-smoothing is entirely the caller's (the frontend store's) job. + let s = signals(true, None, 100_000, 400_000, 400_000); + assert_eq!(classify(&s), classify(&s)); + } + + #[test] + fn rate_bytes_per_sec_floors_a_degenerate_elapsed_at_1ms() { + // A zero/negative elapsed (clock oddity) must never divide by zero + // or panic; it floors at 1ms, producing a large-but-finite rate. + assert_eq!(rate_bytes_per_sec(1_000, 0, 0), 1_000_000); + assert_eq!(rate_bytes_per_sec(1_000, 0, -5), 1_000_000); + assert_eq!(rate_bytes_per_sec(0, 0, 1000), 0); + } + + #[test] + fn is_active_excludes_only_the_three_at_rest_states() { + assert!(!is_active(&OrchestratorState::Idle { last_run_at: None })); + assert!(!is_active(&OrchestratorState::Paused { + reason: driven_core::types::PauseReason::Manual + })); + assert!(is_active(&OrchestratorState::PowerCheck)); + assert!(is_active(&OrchestratorState::Backoff { until: 0 })); + } + + // --- account_signal / aggregate_accounts -------------------------------- + + #[test] + fn account_signal_is_inactive_and_backoff_free_at_rest() { + for state in [ + OrchestratorState::Idle { last_run_at: None }, + OrchestratorState::Paused { + reason: driven_core::types::PauseReason::Manual, + }, + OrchestratorState::Error { + detail: driven_core::types::ErrorDetail::new( + driven_core::types::ErrorCode::AuthInvalidGrant, + "boom", + ), + }, + ] { + // Even a pacer that WOULD report a backoff is ignored at rest - + // an Idle/Paused/Error account cannot be "rate-limited right now". + let (active, backoff) = account_signal(&state, 1_000, Some(5_000), "Drive"); + assert!(!active, "{state:?} must not count as mid-cycle"); + assert_eq!(backoff, None); + } + } + + #[test] + fn account_signal_backoff_prefers_the_orchestrator_level_deadline() { + // A circuit-breaker/rate-limit trip carries its own deadline, + // independent of (and NOT summed with) any pacer reading. + let (active, backoff) = account_signal( + &OrchestratorState::Backoff { until: 9_000 }, + 1_000, + Some(500), + "S3", + ); + assert!(active); + assert_eq!(backoff, Some(("S3".to_string(), 8_000))); + } + + #[test] + fn account_signal_backoff_deadline_never_goes_negative() { + // A stale `until` in the past (clock skew, or the deadline just + // lifted) floors at zero rather than reporting a negative remaining. + let (_, backoff) = account_signal( + &OrchestratorState::Backoff { until: 500 }, + 1_000, + None, + "Drive", + ); + assert_eq!(backoff, Some(("Drive".to_string(), 0))); + } + + #[test] + fn account_signal_mid_cycle_defers_to_the_pacer_reading() { + let (active, backoff) = + account_signal(&OrchestratorState::PowerCheck, 1_000, Some(3_000), "Drive"); + assert!(active); + assert_eq!(backoff, Some(("Drive".to_string(), 3_000))); + + // Mid-cycle but the pacer reports clear: active, no backoff. + let (active, backoff) = + account_signal(&OrchestratorState::PowerCheck, 1_000, None, "Drive"); + assert!(active); + assert_eq!(backoff, None); + } + + #[test] + fn aggregate_accounts_any_active_and_longest_backoff_wins() { + let per_account = vec![ + (false, None), + (true, Some(("Drive".to_string(), 2_000))), + (true, Some(("S3".to_string(), 9_000))), + (true, None), + ]; + let (active_cycle, backoff) = aggregate_accounts(&per_account); + assert!(active_cycle); + assert_eq!( + backoff, + Some(("S3".to_string(), 9_000)), + "the longer window wins" + ); + } + + #[test] + fn aggregate_accounts_empty_or_all_at_rest_is_inactive_with_no_backoff() { + assert_eq!(aggregate_accounts(&[]), (false, None)); + assert_eq!( + aggregate_accounts(&[(false, None), (false, None)]), + (false, None) + ); + } + + // --- build_snapshot ------------------------------------------------------ + + #[test] + fn build_snapshot_gates_backend_and_backoff_fields_to_the_api_state() { + let snap = build_snapshot( + 1_234, + BottleneckState::Api, + None, + Some(("Drive".to_string(), 8_000)), + ); + assert_eq!(snap.ts_ms, 1_234); + assert_eq!(snap.state, BottleneckState::Api); + assert_eq!(snap.rate_bytes_per_sec, None); + assert_eq!(snap.backend, Some("Drive".to_string())); + assert_eq!(snap.backoff_remaining_ms, Some(8_000)); + } + + #[test] + fn build_snapshot_never_leaks_a_backoff_onto_a_non_api_state() { + // Defensive: even if a caller somehow hands in a `backoff` alongside + // a non-Api state, the DTO must not surface it. + let snap = build_snapshot( + 1_234, + BottleneckState::Disk, + Some(100_000), + Some(("Drive".to_string(), 8_000)), + ); + assert_eq!(snap.rate_bytes_per_sec, Some(100_000)); + assert_eq!(snap.backend, None); + assert_eq!(snap.backoff_remaining_ms, None); + } + + #[test] + fn build_snapshot_not_backing_up_carries_no_rate_or_backoff() { + let snap = build_snapshot(0, BottleneckState::NotBackingUp, None, None); + assert_eq!(snap.rate_bytes_per_sec, None); + assert_eq!(snap.backend, None); + assert_eq!(snap.backoff_remaining_ms, None); + } + + // --- should_emit ----------------------------------------------------------- + + #[test] + fn should_emit_suppresses_only_a_repeated_not_backing_up() { + assert!( + should_emit(BottleneckState::NotBackingUp, None), + "first tick always emits" + ); + assert!( + !should_emit( + BottleneckState::NotBackingUp, + Some(BottleneckState::NotBackingUp) + ), + "repeat idle is suppressed" + ); + assert!( + should_emit(BottleneckState::NotBackingUp, Some(BottleneckState::Disk)), + "the transition INTO idle still emits once" + ); + assert!( + should_emit(BottleneckState::Disk, Some(BottleneckState::Disk)), + "a non-idle state always emits, even unchanged" + ); + } + + // --- BottleneckHub --------------------------------------------------------- + + #[test] + fn bottleneck_hub_defaults_to_not_backing_up_and_push_updates_latest() { + let hub = BottleneckHub::default(); + assert_eq!(hub.latest().state, BottleneckState::NotBackingUp); + + let snap = build_snapshot(42, BottleneckState::Cpu, Some(900_000), None); + hub.push(snap.clone()); + assert_eq!(hub.latest(), snap); + + // `latest()` is a peek, not a drain - repeated reads see the same value. + assert_eq!(hub.latest(), snap); + } +} diff --git a/src-tauri/src/commands/sync.rs b/src-tauri/src/commands/sync.rs index bdf1832e..3a3fc6f8 100644 --- a/src-tauri/src/commands/sync.rs +++ b/src-tauri/src/commands/sync.rs @@ -393,6 +393,17 @@ pub async fn io_throughput_series( Ok(state.iostat_hub().series()) } +/// `bottleneck_status()` - the latest live bottleneck classification (issue +/// #308), for the Activity dashboard's Bottleneck stat tile's initial paint; +/// live updates then ride the `sync:bottleneck` event. Always available (the +/// quiesced boot path serves a `NotBackingUp` default). +#[tauri::command] +pub async fn bottleneck_status( + state: State<'_, AppState>, +) -> CommandResult { + Ok(state.bottleneck_hub().latest()) +} + /// `get_sync_status()` - snapshot the aggregate sync state (SPEC s11.3). /// /// Reads each account orchestrator's current [`OrchestratorState`] into the diff --git a/src-tauri/src/events.rs b/src-tauri/src/events.rs index 6124297b..ddb47f5f 100644 --- a/src-tauri/src/events.rs +++ b/src-tauri/src/events.rs @@ -30,6 +30,10 @@ pub const EVENT_SYNC_SOURCE_PROGRESS: &str = "sync:source_progress"; /// One live disk/network throughput sample from the app-global sampler /// (2026-08-14 follow-up; payload is `iostat_hub::IoSample`, camelCase). pub const EVENT_SYNC_IO_THROUGHPUT: &str = "sync:io_throughput"; +/// One live bottleneck classification from the app-global sampler (issue +/// #308, 2026-08-17 follow-up; payload is `bottleneck_hub::BottleneckSnapshot`, +/// camelCase). +pub const EVENT_SYNC_BOTTLENECK: &str = "sync:bottleneck"; /// `activity:new` - a new activity-log entry (payload: `ActivityEntry`, /// SPEC s11.7). /// @@ -210,6 +214,13 @@ pub fn emit_sync_io_throughput(app: &AppHandle, sample: crate::iostat_hub::IoSam let _ = app.emit(EVENT_SYNC_IO_THROUGHPUT, sample); } +/// Broadcast `sync:bottleneck` with one live bottleneck classification +/// (issue #308). Fire-and-forget like the other live-telemetry events: a +/// failed emit is logged by the caller's tracing, never propagated. +pub fn emit_sync_bottleneck(app: &AppHandle, snapshot: crate::bottleneck_hub::BottleneckSnapshot) { + let _ = app.emit(EVENT_SYNC_BOTTLENECK, snapshot); +} + /// Broadcast `activity:new` with the new activity entry (SPEC s11.7). /// /// M7 (activity dashboard): the event bridge calls this on every diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 23bdeecf..67bf6ded 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -22,6 +22,9 @@ rust_i18n::i18n!("locales", fallback = "en-US"); mod apfs_helper; mod app_state; mod assembly; +// issue #308: live bottleneck classification behind the Activity dashboard's +// Bottleneck stat tile (mirrors `iostat_hub`'s sampler lifecycle). +mod bottleneck_hub; // `pub` so the integration tests (`tests/ipc_path_validation.rs`, SPEC s11.6.1) // can exercise the path-validation helpers (`validate_writable_dest`, // `DialogToken`) against the real implementation. @@ -282,6 +285,8 @@ struct ShutdownHandles { telemetry: Option>, /// 2026-08-14 follow-up: the 1 Hz io-throughput sampler, if it was started. iostat: Option>, + /// issue #308: the 1 Hz bottleneck-classification sampler, if it was started. + bottleneck: Option>, } /// Signal every shutdown-able task and TAKE its handle, synchronously. @@ -317,6 +322,8 @@ fn take_shutdown_handles(app: &tauri::AppHandle) -> Option { telemetry: state.shutdown_telemetry_task(), // 2026-08-14 follow-up: the io-throughput sampler, same shape again. iostat: state.shutdown_iostat_task(), + // issue #308: the bottleneck-classification sampler, same shape again. + bottleneck: state.shutdown_bottleneck_task(), }) } @@ -350,6 +357,7 @@ async fn drain_shutdown_handles(handles: ShutdownHandles) { updater, telemetry, iostat, + bottleneck, } = handles; // R3-P1-1: drive ALL per-account shutdowns concurrently. Each @@ -405,6 +413,12 @@ async fn drain_shutdown_handles(handles: ShutdownHandles) { tracing::info!(target: "driven::app", "io throughput sampler drained (no orphan)"); } + // issue #308: drain the bottleneck sampler the same way. + if let Some(handle) = bottleneck { + drain_restore_handle(handle).await; + tracing::info!(target: "driven::app", "bottleneck sampler drained (no orphan)"); + } + // Stop the cosmetic tray syncing-spinner LAST - AFTER every orchestrator // is dropped (so the per-account event bridges' broadcasts are closed and // no further `StateChanged` can drive `apply_state` -> restart the @@ -711,6 +725,9 @@ pub fn run() { // 2026-08-14 follow-up: the live disk/network throughput // sampler behind the Activity dashboard's split graphs. iostat_hub::spawn_sampler(&handle); + // issue #308: the live bottleneck-classification sampler + // behind the Activity dashboard's Bottleneck stat tile. + bottleneck_hub::spawn_sampler(&handle); // M9b R2-P2-3 (SPEC s16): record an `update_applied` activity row // when the running version differs from the last-recorded one, so // the telemetry `update_applied` aggregate is driven by a real @@ -803,6 +820,7 @@ pub fn run() { commands::sync::get_work_queue, commands::sync::cancel_work_item, commands::sync::clear_work_queue, + commands::sync::bottleneck_status, // SPEC s11.1 accounts (M6). commands::accounts::list_accounts, commands::accounts::list_backends, diff --git a/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/empty.png b/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/empty.png index 820f13ed..482ba3d7 100644 Binary files a/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/empty.png and b/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/empty.png differ diff --git a/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/loading.png b/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/loading.png index 004ac0a1..d94a0bca 100644 Binary files a/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/loading.png and b/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/loading.png differ diff --git a/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/populated.png b/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/populated.png index 51bca919..763ef7f1 100644 Binary files a/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/populated.png and b/ui/e2e-visual/__screenshots__/linux/dark/activity.spec.ts/populated.png differ diff --git a/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/empty.png b/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/empty.png index 670fda41..68e4c228 100644 Binary files a/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/empty.png and b/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/empty.png differ diff --git a/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/loading.png b/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/loading.png index 3c8dcf75..3d5ba45d 100644 Binary files a/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/loading.png and b/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/loading.png differ diff --git a/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/populated.png b/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/populated.png index 5e5d73f3..5f2743bf 100644 Binary files a/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/populated.png and b/ui/e2e-visual/__screenshots__/linux/light/activity.spec.ts/populated.png differ diff --git a/ui/src/__tests__/bottleneck-stat-tile.test.ts b/ui/src/__tests__/bottleneck-stat-tile.test.ts new file mode 100644 index 00000000..d871d9a4 --- /dev/null +++ b/ui/src/__tests__/bottleneck-stat-tile.test.ts @@ -0,0 +1,88 @@ +// @vitest-environment jsdom +import { describe, it, expect } from "vitest"; +import { mount } from "@vue/test-utils"; + +import { i18n } from "../i18n"; +import BottleneckStatTile from "../components/BottleneckStatTile.vue"; +import type { BottleneckSnapshot } from "../ipc/types"; + +// BottleneckStatTile tests (issue #308): the tile is a pure render of the +// debounced snapshot prop, so every one of the six states - plus the +// not-yet-hydrated (null) state - is drivable without a backend or store. + +const TILE = '[data-testid="bottleneck-tile"]'; +const VALUE = '[data-testid="bottleneck-value"]'; +const SUB = '[data-testid="bottleneck-sub"]'; + +function snap(state: BottleneckSnapshot["state"], extra: Partial = {}) { + return { + tsMs: 0, + state, + rateBytesPerSec: null, + backend: null, + backoffRemainingMs: null, + ...extra, + }; +} + +function mountTile(snapshot: BottleneckSnapshot | null) { + return mount(BottleneckStatTile, { + props: { snapshot }, + global: { plugins: [i18n] }, + }); +} + +describe("BottleneckStatTile", () => { + it("renders a placeholder before the store has hydrated", () => { + const wrapper = mountTile(null); + expect(wrapper.find(TILE).exists()).toBe(true); + expect(wrapper.find(VALUE).text()).toBe("..."); + expect(wrapper.find(SUB).exists()).toBe(false); + }); + + it("not_backing_up: no sub-line", () => { + const wrapper = mountTile(snap("not_backing_up")); + expect(wrapper.find(VALUE).text()).toBe("Not backing up"); + expect(wrapper.find(SUB).exists()).toBe(false); + }); + + it("disk: names the rate as read-bound", () => { + const wrapper = mountTile(snap("disk", { rateBytesPerSec: 210_000_000 })); + expect(wrapper.find(VALUE).text()).toBe("Disk"); + expect(wrapper.find(SUB).text()).toBe("read-bound · 200.3 MB/s"); + }); + + it("network: names the rate as upload-bound", () => { + const wrapper = mountTile(snap("network", { rateBytesPerSec: 42_000_000 })); + expect(wrapper.find(VALUE).text()).toBe("Network"); + expect(wrapper.find(SUB).text()).toBe("upload-bound · 40.1 MB/s"); + }); + + it("cpu: names the rate as hash-bound", () => { + const wrapper = mountTile(snap("cpu", { rateBytesPerSec: 900_000_000 })); + expect(wrapper.find(VALUE).text()).toBe("CPU"); + expect(wrapper.find(SUB).text()).toBe("hash-bound · 858.3 MB/s"); + }); + + it("mixed: no clear limiter", () => { + const wrapper = mountTile(snap("mixed")); + expect(wrapper.find(VALUE).text()).toBe("Mixed"); + expect(wrapper.find(SUB).text()).toBe("no clear limiter"); + }); + + it("api: names the backend and the remaining backoff in whole seconds", () => { + const wrapper = mountTile(snap("api", { backend: "Drive", backoffRemainingMs: 8_400 })); + expect(wrapper.find(VALUE).text()).toBe("API"); + expect(wrapper.find(SUB).text()).toBe("Drive rate-limited · backing off 8s"); + }); + + it("api: falls back to a generic backend label when the wire omits it", () => { + const wrapper = mountTile(snap("api", { backend: null, backoffRemainingMs: 1_000 })); + expect(wrapper.find(SUB).text()).toBe("the destination rate-limited · backing off 1s"); + }); + + it("a rate-bearing state with no rate yet renders no sub-line rather than a bogus one", () => { + const wrapper = mountTile(snap("disk", { rateBytesPerSec: null })); + expect(wrapper.find(SUB).exists()).toBe(false); + }); +}); diff --git a/ui/src/__tests__/bottleneck-store.test.ts b/ui/src/__tests__/bottleneck-store.test.ts new file mode 100644 index 00000000..1a2b39e0 --- /dev/null +++ b/ui/src/__tests__/bottleneck-store.test.ts @@ -0,0 +1,149 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { createPinia, setActivePinia } from "pinia"; + +// Live bottleneck-classification store tests (issue #308). The seams are the +// `bottleneck_status` seed command and the `sync:bottleneck` live event; +// mocking both drives the whole store - seed, live folding, the debounce + +// hysteresis gate, and the garbled-wire-data coercion - with no backend. + +const invokeMock = vi.fn(); +vi.mock("@tauri-apps/api/core", () => ({ + invoke: (cmd: string, args?: unknown) => invokeMock(cmd, args), +})); + +const handlers: Record void> = {}; +const unlistenMock = vi.fn(); +const listenMock = vi.fn(async (event: string, cb: (e: { payload: unknown }) => void) => { + handlers[event] = (payload: unknown) => cb({ payload }); + return vi.fn(() => { + delete handlers[event]; + unlistenMock(); + }); +}); +vi.mock("@tauri-apps/api/event", () => ({ + listen: (event: string, cb: (e: { payload: unknown }) => void) => listenMock(event, cb), +})); + +import { useBottleneckStore, DEBOUNCE_MS } from "../stores/bottleneck"; +import type { BottleneckSnapshot } from "../ipc/types"; + +function snap( + tsMs: number, + state: BottleneckSnapshot["state"], + extra: Partial = {} +): BottleneckSnapshot { + return { + tsMs, + state, + rateBytesPerSec: null, + backend: null, + backoffRemainingMs: null, + ...extra, + }; +} + +beforeEach(() => { + setActivePinia(createPinia()); + invokeMock.mockReset(); + listenMock.mockClear(); + unlistenMock.mockClear(); + for (const k of Object.keys(handlers)) delete handlers[k]; +}); + +describe("bottleneck store", () => { + it("seeds from bottleneck_status and subscribes to live events", async () => { + invokeMock.mockResolvedValueOnce(snap(1000, "not_backing_up")); + const store = useBottleneckStore(); + await store.start(); + + expect(invokeMock).toHaveBeenCalledWith("bottleneck_status", undefined); + expect(store.displayed?.state).toBe("not_backing_up"); + expect(listenMock).toHaveBeenCalledWith("sync:bottleneck", expect.any(Function)); + }); + + it("adopts the FIRST snapshot immediately (a hydration read needs no debounce)", async () => { + invokeMock.mockResolvedValueOnce(snap(1000, "disk", { rateBytesPerSec: 210_000_000 })); + const store = useBottleneckStore(); + await store.start(); + expect(store.displayed?.state).toBe("disk"); + }); + + it("holds the old state until a new one has been stable for the debounce window", async () => { + invokeMock.mockResolvedValueOnce(snap(0, "not_backing_up")); + const store = useBottleneckStore(); + await store.start(); + expect(store.displayed?.state).toBe("not_backing_up"); + + // A new state arrives, but hasn't held long enough yet. + handlers["sync:bottleneck"](snap(1000, "network", { rateBytesPerSec: 42_000_000 })); + expect(store.displayed?.state).toBe("not_backing_up"); + + handlers["sync:bottleneck"]( + snap(1000 + DEBOUNCE_MS - 1, "network", { rateBytesPerSec: 42_000_000 }) + ); + expect(store.displayed?.state).toBe("not_backing_up"); + + // Now it has held for >= DEBOUNCE_MS since it first appeared. + handlers["sync:bottleneck"]( + snap(1000 + DEBOUNCE_MS, "network", { rateBytesPerSec: 42_000_000 }) + ); + expect(store.displayed?.state).toBe("network"); + }); + + it("flapping between two states never reaches the display", async () => { + invokeMock.mockResolvedValueOnce(snap(0, "mixed")); + const store = useBottleneckStore(); + await store.start(); + + // Alternates every tick for well over the debounce window - each change + // resets the window, so `displayed` never moves off the seeded state. + for (let i = 1; i <= 20; i++) { + handlers["sync:bottleneck"](snap(i * 1000, i % 2 === 0 ? "disk" : "cpu")); + } + expect(store.displayed?.state).toBe("mixed"); + }); + + it("keeps the debounced state's numbers fresh while it holds", async () => { + invokeMock.mockResolvedValueOnce(snap(0, "cpu", { rateBytesPerSec: 100_000 })); + const store = useBottleneckStore(); + await store.start(); + + // Same state, later tick, different rate, still not past the window. + handlers["sync:bottleneck"](snap(1000, "cpu", { rateBytesPerSec: 500_000 })); + // Not yet promoted (started counting from ts=0, the seed). + expect(store.displayed?.rateBytesPerSec).toBe(100_000); + + handlers["sync:bottleneck"](snap(DEBOUNCE_MS, "cpu", { rateBytesPerSec: 900_000 })); + expect(store.displayed?.state).toBe("cpu"); + expect(store.displayed?.rateBytesPerSec).toBe(900_000); + }); + + it("coerces a garbled wire snapshot to safe defaults", async () => { + invokeMock.mockRejectedValueOnce(new Error("ipc down")); + const store = useBottleneckStore(); + await store.start(); + expect(store.displayed).toBeNull(); + + handlers["sync:bottleneck"]({ + tsMs: "nope", + state: "not_a_real_state", + rateBytesPerSec: -5, + backend: 42, + backoffRemainingMs: Number.NaN, + }); + expect(store.displayed).toEqual( + snap(0, "not_backing_up", { rateBytesPerSec: null, backend: null, backoffRemainingMs: null }) + ); + }); + + it("stop() unsubscribes and start() is idempotent", async () => { + invokeMock.mockResolvedValue(snap(0, "not_backing_up")); + const store = useBottleneckStore(); + await store.start(); + await store.start(); + expect(listenMock).toHaveBeenCalledTimes(1); + store.stop(); + expect(unlistenMock).toHaveBeenCalledTimes(1); + expect(handlers["sync:bottleneck"]).toBeUndefined(); + }); +}); diff --git a/ui/src/components/BottleneckStatTile.vue b/ui/src/components/BottleneckStatTile.vue new file mode 100644 index 00000000..b13028f3 --- /dev/null +++ b/ui/src/components/BottleneckStatTile.vue @@ -0,0 +1,105 @@ + + + diff --git a/ui/src/ipc/commands.ts b/ui/src/ipc/commands.ts index 185853fc..921b1cd6 100644 --- a/ui/src/ipc/commands.ts +++ b/ui/src/ipc/commands.ts @@ -18,6 +18,7 @@ import type { ApfsHelperStatus, BackendDto, BackendKindId, + BottleneckSnapshot, CreateS3AccountRequest, CreateLocalFolderAccountRequest, CreateSftpAccountRequest, @@ -490,6 +491,13 @@ export function ioThroughputSeries(): Promise { return invoke("io_throughput_series"); } +/** The latest live bottleneck classification (issue #308), for the Activity + * dashboard's Bottleneck stat tile's initial paint; live updates then ride + * the `sync:bottleneck` event. */ +export function bottleneckStatus(): Promise { + return invoke("bottleneck_status"); +} + /** The persisted integrity-scrub reports, newest first. Omit `sourceId` for * every source interleaved by time. Every field is a COUNT - the shape carries * no paths or remote ids - so this can never surface an encrypted source's diff --git a/ui/src/ipc/events.ts b/ui/src/ipc/events.ts index 6f004e4b..32ce9bd8 100644 --- a/ui/src/ipc/events.ts +++ b/ui/src/ipc/events.ts @@ -9,6 +9,7 @@ import { listen, type UnlistenFn } from "@tauri-apps/api/event"; import type { AccountSyncStatus, ActivityEntry, + BottleneckSnapshot, ExclusionPreviewBatch, ExclusionPreviewDone, ExclusionPreviewError, @@ -85,6 +86,16 @@ export function onQueueChanged(handler: (snapshot: QueueSnapshot) => void): Prom return listen("queue:changed", (e) => handler(e.payload)); } +/** `sync:bottleneck` payload: one live bottleneck classification (issue + * #308, camelCase, the same `BottleneckSnapshot` shape `bottleneck_status` + * returns). Emitted ~1/s while any account is mid-cycle; suppressed while + * fully idle after one trailing `not_backing_up`. */ +export function onSyncBottleneck( + handler: (snapshot: BottleneckSnapshot) => void +): Promise { + return listen("sync:bottleneck", (e) => handler(e.payload)); +} + /** `activity:new` payload: ActivityEntry (SPEC s11.7). The Activity dashboard's * live tail subscribes to this and prepends new entries (deduped by id). */ export function onActivityNew(handler: (entry: ActivityEntry) => void): Promise { diff --git a/ui/src/ipc/types.ts b/ui/src/ipc/types.ts index 73da8be3..65b6a57a 100644 --- a/ui/src/ipc/types.ts +++ b/ui/src/ipc/types.ts @@ -822,6 +822,24 @@ export interface IoThroughputSeriesDto { samples: IoSample[]; } +/** Which stage of the backup pipeline is presently limiting throughput + * (issue #308, mirrors src-tauri `bottleneck_hub::BottleneckState`). */ +export type BottleneckState = "not_backing_up" | "disk" | "network" | "api" | "cpu" | "mixed"; + +/** One live bottleneck classification (issue #308), the `sync:bottleneck` + * event and `bottleneck_status` command payload, camelCase on the wire + * (src-tauri/src/bottleneck_hub.rs `BottleneckSnapshot`). */ +export interface BottleneckSnapshot { + tsMs: number; + state: BottleneckState; + /** The saturated stage's rate in bytes/sec, present only for Disk/Network/Cpu. */ + rateBytesPerSec: number | null; + /** The rate-limited destination's short label, present only for Api. */ + backend: string | null; + /** Ms remaining in the active backoff window, present only for Api. */ + backoffRemainingMs: number | null; +} + /** Mirrors the Rust `OrchestratorState` (driven_core::types). Carried as an * opaque tagged object; the UI reads the discriminant for the status pill. * On the wire it is internally tagged on a snake_case `state` field (e.g. diff --git a/ui/src/locales/en-US.json b/ui/src/locales/en-US.json index bc8f83df..22b6051c 100644 --- a/ui/src/locales/en-US.json +++ b/ui/src/locales/en-US.json @@ -704,7 +704,21 @@ "noFiles": "No files tracked yet", "diskThroughput": "Disk read", "noDiskThroughput": "No recent disk activity", - "diskSparklineLabel": "Disk read throughput over the last {minutes} minutes, peaking at {peak} per second" + "diskSparklineLabel": "Disk read throughput over the last {minutes} minutes, peaking at {peak} per second", + "bottleneck": "Bottleneck", + "bottleneckUnknown": "...", + "bottleneckNotBackingUp": "Not backing up", + "bottleneckDisk": "Disk", + "bottleneckNetwork": "Network", + "bottleneckApi": "API", + "bottleneckCpu": "CPU", + "bottleneckMixed": "Mixed", + "bottleneckDiskSub": "read-bound · {rate}", + "bottleneckNetworkSub": "upload-bound · {rate}", + "bottleneckCpuSub": "hash-bound · {rate}", + "bottleneckApiSub": "{backend} rate-limited · backing off {seconds}s", + "bottleneckApiGenericBackend": "the destination", + "bottleneckMixedSub": "no clear limiter" }, "status": { "synced": "Synced", diff --git a/ui/src/stores/bottleneck.ts b/ui/src/stores/bottleneck.ts new file mode 100644 index 00000000..b2bd76ce --- /dev/null +++ b/ui/src/stores/bottleneck.ts @@ -0,0 +1,144 @@ +import { defineStore } from "pinia"; +import { ref } from "vue"; +import type { UnlistenFn } from "@tauri-apps/api/event"; + +import * as ipc from "../ipc/commands"; +import { onSyncBottleneck } from "../ipc/events"; +import type { BottleneckSnapshot, BottleneckState } from "../ipc/types"; + +/** + * Live bottleneck-classification store (issue #308). The backend classifies + * the limiting pipeline stage once a second; this store seeds from + * `bottleneck_status` and then folds each `sync:bottleneck` event, same + * subscribe()/hydrate() shape as `stores/iostat.ts`. + * + * DEBOUNCE + HYSTERESIS: the backend's classifier is a pure per-tick function + * with no memory (driven-core intentionally leaves flap-smoothing to the + * caller), so a state a few B/s from a threshold can genuinely flip tick to + * tick. Rather than repaint the tile every second, `displayed` only adopts a + * NEW state once that state has been the raw incoming value continuously for + * `DEBOUNCE_MS` (5s) - any state change resets the window, so brief flapping + * between two states never reaches the UI. The window is measured off the + * snapshots' own `tsMs` (backend wall-clock), not `Date.now()`, so this is + * deterministic under test with no fake timers. + */ + +/** How long a new state must hold before the tile adopts it. */ +export const DEBOUNCE_MS = 5000; + +/** Read a finite non-negative number, or null - untrusted wire data. */ +function numOrNull(v: unknown): number | null { + return typeof v === "number" && Number.isFinite(v) && v >= 0 ? v : null; +} + +function num(v: unknown): number { + return typeof v === "number" && Number.isFinite(v) && v >= 0 ? v : 0; +} + +const VALID_STATES: readonly BottleneckState[] = [ + "not_backing_up", + "disk", + "network", + "api", + "cpu", + "mixed", +]; + +/** Coerce one wire snapshot (missing/garbled fields degrade to safe + * defaults rather than throwing, matching `stores/iostat.ts`'s `readSample`). */ +function readSnapshot(s: unknown): BottleneckSnapshot { + const o = (s ?? {}) as Record; + const state = VALID_STATES.includes(o["state"] as BottleneckState) + ? (o["state"] as BottleneckState) + : "not_backing_up"; + return { + tsMs: num(o["tsMs"]), + state, + rateBytesPerSec: numOrNull(o["rateBytesPerSec"]), + backend: typeof o["backend"] === "string" ? (o["backend"] as string) : null, + backoffRemainingMs: numOrNull(o["backoffRemainingMs"]), + }; +} + +export const useBottleneckStore = defineStore("bottleneck", () => { + /** The debounced value the tile renders. Null until the first snapshot + * arrives (hydrate or a live event), so the tile can show its own loading + * state rather than a misleading default. */ + const displayed = ref(null); + + // The raw incoming state being timed for the debounce window - NOT + // reactive (it is an implementation detail of `apply`, not something a + // consumer should render mid-debounce). + let pending: BottleneckSnapshot | null = null; + let pendingSinceMs: number | null = null; + + /** Fold one incoming snapshot (hydrate seed or live event) through the + * debounce/hysteresis gate. */ + function apply(snapshot: BottleneckSnapshot): void { + if (displayed.value === null) { + // Nothing shown yet: adopt immediately (a hydration read is already a + // stable point-in-time read, not a tick that might flap) and seed the + // debounce window so subsequent flapping is judged against it. + displayed.value = snapshot; + pending = snapshot; + pendingSinceMs = snapshot.tsMs; + return; + } + if (pending === null || pending.state !== snapshot.state) { + // A new candidate state: start (or restart) its debounce window. + pending = snapshot; + pendingSinceMs = snapshot.tsMs; + return; + } + // Same candidate state as before: keep its numbers fresh, and promote it + // to `displayed` once it has held for the whole debounce window. + pending = snapshot; + if (pendingSinceMs !== null && snapshot.tsMs - pendingSinceMs >= DEBOUNCE_MS) { + displayed.value = snapshot; + } + } + + // --- lifecycle (the Activity view owns the registration) ------------------ + let unlisten: UnlistenFn | null = null; + let started = false; + + /** Seed from the latest snapshot + subscribe to live updates (idempotent). + * Best-effort: a failed seed still leaves the live stream to populate the + * tile. */ + async function start(): Promise { + if (started) return; + started = true; + try { + apply(readSnapshot(await ipc.bottleneckStatus())); + } catch (e) { + console.error("bottleneck status seed failed", e); + } + try { + const un = await onSyncBottleneck((snapshot) => apply(readSnapshot(snapshot))); + // stop() may have raced ahead while we awaited; honor it. + if (!started) { + un(); + return; + } + unlisten = un; + } catch (e) { + started = false; + console.error("bottleneck subscribe failed", e); + } + } + + /** Stop the live subscription (view unmount). */ + function stop(): void { + started = false; + if (unlisten) { + unlisten(); + unlisten = null; + } + } + + return { + displayed, + start, + stop, + }; +}); diff --git a/ui/src/views/Activity.vue b/ui/src/views/Activity.vue index c8b90db9..bfe363a2 100644 --- a/ui/src/views/Activity.vue +++ b/ui/src/views/Activity.vue @@ -5,10 +5,12 @@ import { useI18n } from "vue-i18n"; import * as ipc from "../ipc/commands"; import { toErrorCode } from "../ipc/errors"; import { flushFrontendLogs } from "../frontendLog"; +import BottleneckStatTile from "../components/BottleneckStatTile.vue"; import FilesUploadedStatTile from "../components/FilesUploadedStatTile.vue"; import DrillHistoryPanel from "../components/DrillHistoryPanel.vue"; import ScrubHistoryPanel from "../components/ScrubHistoryPanel.vue"; import ThroughputStatTile from "../components/ThroughputStatTile.vue"; +import { useBottleneckStore } from "../stores/bottleneck"; import { useIostatStore } from "../stores/iostat"; import { activityEventLabel } from "../stores/activityEventLabel"; import { @@ -33,6 +35,9 @@ const activity = useActivityStore(); // split tiles (probe-fed 1s samples; moves during reconcile-phase recovery, // unlike the activity-log-backed series which only updates on completed rows). const iostat = useIostatStore(); +// issue #308: the live bottleneck-classification store behind the Bottleneck +// tile (debounced 5s with hysteresis so the tile does not flap tick to tick). +const bottleneck = useBottleneckStore(); const sources = useSourcesStore(); const toasts = useToastsStore(); @@ -259,6 +264,7 @@ onMounted(async () => { // (M7-P1-1), so a broadcast-lag burst loses no rows. await activity.subscribeLive(); await iostat.start(); + await bottleneck.start(); try { await Promise.all([ sources.refresh(), @@ -278,6 +284,7 @@ onMounted(async () => { onUnmounted(() => { activity.unsubscribeLive(); iostat.stop(); + bottleneck.stop(); }); @@ -325,16 +332,19 @@ onUnmounted(() => { data-testid="activity-summary-skeleton" aria-hidden="true" > -
+
- +
@@ -369,6 +379,7 @@ onUnmounted(() => { :bucket-ms="SPARKLINE_BUCKET_MS" :files-uploaded="filesUploaded" /> +
{{ t("activity.summary.byStatus") }} diff --git a/ui/test-support/fixtures.ts b/ui/test-support/fixtures.ts index f8e845b0..fe996848 100644 --- a/ui/test-support/fixtures.ts +++ b/ui/test-support/fixtures.ts @@ -38,6 +38,7 @@ import type { VersioningConfig, VssHelperStatus, IoThroughputSeriesDto, + BottleneckSnapshot, } from "../src/ipc/types"; /** The instant every fixture is anchored to: 2026-03-15T12:00:00Z. @@ -403,6 +404,19 @@ export const IO_THROUGHPUT: IoThroughputSeriesDto = { })), }; +/** issue #308: a deterministic "network-bound upload" bottleneck reading for + * the Activity dashboard's Bottleneck stat tile, matching the shape + * `IO_THROUGHPUT` implies (a steadier upload trailing a disk-read burst). The + * store adopts the FIRST snapshot immediately (no debounce on a hydration + * read), so this renders straight away in the visual baselines. */ +export const BOTTLENECK_STATUS: BottleneckSnapshot = { + tsMs: FIXED_NOW, + state: "network", + rateBytesPerSec: 42_000_000, + backend: null, + backoffRemainingMs: null, +}; + /** * Recent restore-drill runs, newest first. Deliberately covers all three * outcomes: a pass, a run that could not restore a file, and an INCONCLUSIVE diff --git a/ui/test-support/mock-backend.ts b/ui/test-support/mock-backend.ts index 825b6362..d4655983 100644 --- a/ui/test-support/mock-backend.ts +++ b/ui/test-support/mock-backend.ts @@ -33,6 +33,7 @@ import { ACTIVITY_SUMMARY, ACTIVITY_THROUGHPUT, IO_THROUGHPUT, + BOTTLENECK_STATUS, APFS_HELPER_STATUS, BACKENDS, DRIVE_FOLDER_LISTING, @@ -219,6 +220,7 @@ export function defaultCommands(): Record { activity_summary: ACTIVITY_SUMMARY, activity_throughput_series: ACTIVITY_THROUGHPUT, io_throughput_series: IO_THROUGHPUT, + bottleneck_status: BOTTLENECK_STATUS, // --- Restore (SPEC s11.5) --- list_remote_tree: REMOTE_TREE,