From 9971dc3a2e0d3b3621fd33bb28400f9fb2d3f116 Mon Sep 17 00:00:00 2001 From: pmaxhogan Date: Mon, 20 Jul 2026 15:16:18 -0500 Subject: [PATCH] feat(core): adaptive upload parallelism with throughput probe and disk-saturation gate Close the control loop around the in-flight-file count (DESIGN 11.4.7 / 18.2). A 30s-window ThroughputProbe fed by the executor drives a resizable UploadPool within [1, 32]: shrink when a window's throughput collapses below 50% of the prior one (and neither the pacer nor the disk explains the drop), grow while lifting the pool still improves throughput. A per-OS disk-busy gate (new driven-diskstat crate: PDH on Windows, /proc/diskstats on Linux, IOKit on macOS; fail-open to not-saturated so a broken reader never strangles uploads) blocks growth when the disk is the bound. Default-on with an adaptive_parallelism_enabled kill-switch; off = today's fixed pool at default_concurrent_uploads (now wired as the start size). The control law is a pure, exhaustively unit-tested decide(); an e2e_fake test asserts a real shrink-then-regrow transition through the shared executor/probe/pool wiring. Refs #34 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01QZQVP2tUuTLh8oL31D8heC --- Cargo.lock | 12 + Cargo.toml | 1 + crates/driven-core/Cargo.toml | 4 + crates/driven-core/src/adaptive.rs | 917 ++++++++++++++++++ crates/driven-core/src/executor.rs | 145 ++- crates/driven-core/src/lib.rs | 1 + .../migrations/0011_adaptive_parallelism.sql | 18 + crates/driven-core/src/orchestrator.rs | 199 ++++ crates/driven-core/src/pacer.rs | 31 +- crates/driven-core/tests/e2e_fake.rs | 304 +++++- crates/driven-diskstat/Cargo.toml | 39 + crates/driven-diskstat/src/lib.rs | 190 ++++ crates/driven-diskstat/src/linux.rs | 174 ++++ crates/driven-diskstat/src/macos.rs | 203 ++++ crates/driven-diskstat/src/windows.rs | 136 +++ crates/driven-test-fixtures/Cargo.toml | 1 + crates/driven-test-fixtures/src/diskstat.rs | 74 ++ crates/driven-test-fixtures/src/lib.rs | 4 + src-tauri/Cargo.toml | 3 + src-tauri/src/assembly.rs | 96 +- src-tauri/src/commands/dtos.rs | 19 +- src-tauri/src/commands/settings.rs | 20 + ui/src/__tests__/settings-components.test.ts | 24 + ui/src/__tests__/settings-stores.test.ts | 1 + ui/src/ipc/types.ts | 3 + ui/src/locales/en-US.json | 2 + ui/src/views/Settings.vue | 22 + 27 files changed, 2562 insertions(+), 81 deletions(-) create mode 100644 crates/driven-core/src/adaptive.rs create mode 100644 crates/driven-core/src/migrations/0011_adaptive_parallelism.sql create mode 100644 crates/driven-diskstat/Cargo.toml create mode 100644 crates/driven-diskstat/src/lib.rs create mode 100644 crates/driven-diskstat/src/linux.rs create mode 100644 crates/driven-diskstat/src/macos.rs create mode 100644 crates/driven-diskstat/src/windows.rs create mode 100644 crates/driven-test-fixtures/src/diskstat.rs diff --git a/Cargo.lock b/Cargo.lock index 3521bff1..9a6b1d96 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1364,6 +1364,7 @@ dependencies = [ "crc32fast", "driven-core", "driven-crypto", + "driven-diskstat", "driven-drive", "driven-net", "driven-power", @@ -1452,6 +1453,7 @@ dependencies = [ "blake3", "bytes", "driven-crypto", + "driven-diskstat", "driven-drive", "driven-power", "driven-test-fixtures", @@ -1494,6 +1496,15 @@ dependencies = [ "zeroize", ] +[[package]] +name = "driven-diskstat" +version = "2.0.1" +dependencies = [ + "core-foundation", + "core-foundation-sys", + "windows 0.62.2", +] + [[package]] name = "driven-drive" version = "2.0.1" @@ -1561,6 +1572,7 @@ dependencies = [ "anyhow", "async-trait", "driven-core", + "driven-diskstat", "driven-drive", "driven-power", "parking_lot", diff --git a/Cargo.toml b/Cargo.toml index 2650f1f5..540c0ed5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ members = [ "crates/driven-drive", "crates/driven-crypto", "crates/driven-power", + "crates/driven-diskstat", "crates/driven-vss", "crates/driven-vss-helper", "crates/driven-net", diff --git a/crates/driven-core/Cargo.toml b/crates/driven-core/Cargo.toml index de75cd51..cdebbe35 100644 --- a/crates/driven-core/Cargo.toml +++ b/crates/driven-core/Cargo.toml @@ -47,6 +47,10 @@ sqlx = { version = "0.9", default-features = false, features = [ driven-drive = { path = "../driven-drive" } driven-crypto = { path = "../driven-crypto" } driven-power = { path = "../driven-power" } +# Per-OS disk-busy reader for the adaptive upload-parallelism controller +# (DESIGN s11.4.7 / s18.2). Same acyclic shape as driven-power: it carries no +# driven-core dep. +driven-diskstat = { path = "../driven-diskstat" } # M3.5 Windows VSS reads for exclusively-locked files (ROADMAP M3.5, DESIGN # s5.3). The executor consults the `VssProvider` seam on its open path; the # orchestrator owns the per-cycle snapshot lifecycle + orphan cleanup. The diff --git a/crates/driven-core/src/adaptive.rs b/crates/driven-core/src/adaptive.rs new file mode 100644 index 00000000..ae932b78 --- /dev/null +++ b/crates/driven-core/src/adaptive.rs @@ -0,0 +1,917 @@ +//! Adaptive upload parallelism (DESIGN s11.4.7, s18.2). +//! +//! Drive lets us overlap whole files, not chunks of one file (DESIGN s11.4.1), so +//! the one concurrency knob that matters is "how many files are in flight at +//! once" - the [`UploadPool`] permit count. A fixed count is a guess: too few +//! wastes the link, too many overloads Drive's edge so each upload takes longer +//! and *net* throughput falls (DESIGN s11.4.7's pathological case). This module +//! closes the loop around that knob. +//! +//! # The pieces +//! +//! - [`UploadPool`] - a RESIZABLE `tokio::sync::Semaphore` (the executor's +//! per-file gate). Grows by adding a permit; shrinks by permanently forgetting +//! one. Bounds `[min, max]` with `max` the hard cap 32 (DESIGN s11.4.2). +//! - [`ThroughputProbe`] - a lock-free byte accumulator the executor feeds at +//! each completed upload; the controller drains it once per window to measure +//! aggregate throughput. +//! - [`decide`] - the PURE control law: `(window stats, disk, pacer, size) -> +//! {Grow, Shrink, Hold}`. No I/O, no clock, exhaustively unit-tested. +//! - [`AdaptiveController`] - the thin impure shell the orchestrator's run loop +//! ticks on a fixed cadence; it samples the disk, drains the probe every +//! window, calls [`decide`], and applies the result to the pool. +//! +//! # Windowing + cadence (DESIGN s11.4.7 / s18.2) +//! +//! Throughput is compared over tumbling [`WINDOW`] (30 s) windows; the disk-busy +//! gate is sampled every [`SAMPLE_INTERVAL`] (5 s). The controller is ticked at +//! `SAMPLE_INTERVAL`: each tick samples the disk, and every sixth tick (a full +//! window elapsed on the injected [`Clock`](crate::time::Clock)) runs a decision. +//! Driving the window off the injected clock - never `Instant::now()` - is what +//! makes the whole loop deterministic under a `FakeClock`. +//! +//! # Kill-switch semantics +//! +//! When `adaptive_parallelism_enabled` is true (default) the app wires a +//! controller and the pool floats within `[1, 32]` starting from +//! `default_concurrent_uploads`. When false, no controller is built: the pool is +//! FIXED at `default_concurrent_uploads` exactly as before this feature. A user +//! who wants a hard concurrency limit disables adaptation. + +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use tokio::sync::{AcquireError, OwnedSemaphorePermit, Semaphore}; + +use crate::pacer::Pacer; +use crate::time::Clock; +use driven_diskstat::DiskBusyProbe; + +/// The hard ceiling on in-flight files (DESIGN s11.4.2: "hard cap 32"). The pool +/// may grow up to this regardless of the (lower) default start size. +pub const MAX_POOL: usize = 32; + +/// The floor on in-flight files. One keeps the pipeline alive at minimum +/// concurrency; the pool never shrinks below it. +pub const MIN_POOL: usize = 1; + +/// Throughput comparison window (DESIGN s11.4.7: "30-second windows"). +pub const WINDOW: Duration = Duration::from_secs(30); + +/// Disk-busy sampling cadence (DESIGN s18.2: "sampled at 5-second intervals"). +/// Also the cadence at which the orchestrator ticks the controller. +pub const SAMPLE_INTERVAL: Duration = Duration::from_secs(5); + +/// The default STARTING pool size when the user has not set +/// `default_concurrent_uploads` (DESIGN s11.4.2: `min(available_parallelism * 2, +/// 16)`, clamped into `[MIN_POOL, MAX_POOL]`). The single source of truth for the +/// auto-picked concurrency, used by both the executor's default construction and +/// the app-shell's adaptive wiring. +#[must_use] +pub fn default_pool_size() -> usize { + let par = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(4); + par.saturating_mul(2).min(16).clamp(MIN_POOL, MAX_POOL) +} + +/// Shrink trigger: a window whose throughput is below this fraction of the +/// previous window's is a "collapse" (DESIGN s11.4.7: "< 50% of the previous +/// window's"). +pub const SHRINK_RATIO: f64 = 0.5; + +/// Grow trigger: throughput must EXCEED the previous window by at least this +/// factor to justify adding a permit. See [`decide`] for why growth requires +/// improvement (our resolution of DESIGN's "throughput is at the pool's +/// ceiling"). +pub const GROW_IMPROVE_RATIO: f64 = 1.05; + +/// How long a shrink will wait for an in-flight permit to free before giving up +/// and retrying on the next window. Bounded so the orchestrator's run-loop tick +/// (which awaits this inline) is never blocked for long; a miss is harmless +/// (the pool simply stays one larger until the next decision). +const SHRINK_ACQUIRE_TIMEOUT: Duration = Duration::from_millis(100); + +// --------------------------------------------------------------------------- +// UploadPool +// --------------------------------------------------------------------------- + +/// A resizable in-flight-file gate (DESIGN s11.4.2 / s11.4.7). +/// +/// Wraps a `tokio::sync::Semaphore` whose permit count is the current pool size. +/// The executor acquires one permit per file via [`acquire_owned`](Self::acquire_owned) +/// and releases it by dropping the permit. The controller resizes via +/// [`grow`](Self::grow) / [`shrink`](Self::shrink). +/// +/// # Resize mechanism +/// +/// - **Grow**: `Semaphore::add_permits(1)` - the new permit is immediately +/// available to the next waiting file. +/// - **Shrink**: acquire one permit and `forget()` it, which permanently removes +/// it from circulation. Done inline under a short [`SHRINK_ACQUIRE_TIMEOUT`] +/// (never a detached task - the repo forbids orphanable spawns). On timeout the +/// size is left unchanged and the next window retries. +/// +/// [`target`](Self::target) - not `available_permits()` - is the authoritative +/// size: `available_permits()` momentarily reads low while permits are checked +/// out and can lag a just-landed forget. `target` is only mutated by the single +/// controller task, so its loads/stores need no CAS loop. +#[derive(Debug)] +pub struct UploadPool { + sem: Arc, + /// Authoritative logical size (permits that belong to the pool). + target: AtomicUsize, + min: usize, + max: usize, + /// Count of [`acquire_owned`](Self::acquire_owned) calls that had to WAIT for + /// a permit since the last [`take_contended`](Self::take_contended). A + /// non-zero count over a window means the pool was the active bottleneck + /// (files queued for permits) - the concrete signal for "the pool is the + /// ceiling" used by both the grow and shrink decisions. + contended: AtomicU64, +} + +impl UploadPool { + /// Build a pool starting at `start` permits (clamped into `[MIN_POOL, + /// MAX_POOL]`), resizable within those bounds. Returned as an `Arc` because + /// the SAME handle is shared into the executor (which acquires) and the + /// controller (which resizes). + #[must_use] + pub fn new(start: usize) -> Arc { + Self::with_bounds(start, MIN_POOL, MAX_POOL) + } + + /// Build a pool with explicit bounds (test seam). `start` and the bounds are + /// clamped so `min <= start <= max` always holds. + #[must_use] + pub fn with_bounds(start: usize, min: usize, max: usize) -> Arc { + let min = min.max(1); + let max = max.max(min); + let start = start.clamp(min, max); + Arc::new(Self { + sem: Arc::new(Semaphore::new(start)), + target: AtomicUsize::new(start), + min, + max, + contended: AtomicU64::new(0), + }) + } + + /// The current authoritative pool size. + #[must_use] + pub fn target(&self) -> usize { + self.target.load(Ordering::Acquire) + } + + /// The configured lower bound. + #[must_use] + pub fn min(&self) -> usize { + self.min + } + + /// The configured upper bound (the DESIGN s11.4.2 hard cap by default). + #[must_use] + pub fn max(&self) -> usize { + self.max + } + + /// Acquire one in-flight-file permit, counting contention. Tries without + /// waiting first; only if no permit is free does it record a contention event + /// (the "pool is the bottleneck" signal) and then wait. Functionally + /// identical to a bare `acquire_owned` from the caller's view. + pub async fn acquire_owned(&self) -> Result { + match self.sem.clone().try_acquire_owned() { + Ok(permit) => Ok(permit), + Err(tokio::sync::TryAcquireError::NoPermits) => { + self.contended.fetch_add(1, Ordering::Relaxed); + self.sem.clone().acquire_owned().await + } + // Closed: fall through to the awaiting form, which surfaces the + // `AcquireError` the caller already handles. + Err(tokio::sync::TryAcquireError::Closed) => self.sem.clone().acquire_owned().await, + } + } + + /// Try to acquire a permit WITHOUT waiting and WITHOUT counting contention. + /// Test seam for constructing a pinned-pool scenario deterministically. + #[must_use] + pub fn try_acquire_owned(&self) -> Option { + self.sem.clone().try_acquire_owned().ok() + } + + /// Read-and-reset the contention count accrued since the last call. Called + /// once per decision window by the controller. + #[must_use] + pub fn take_contended(&self) -> u64 { + self.contended.swap(0, Ordering::Relaxed) + } + + /// Peek the contention count without resetting it (used for the cheap + /// "is this account doing anything?" idle check). + #[must_use] + pub fn peek_contended(&self) -> u64 { + self.contended.load(Ordering::Relaxed) + } + + /// Grow the pool by one permit, up to [`max`](Self::max). Returns the new + /// size (unchanged if already at the cap). + pub fn grow(&self) -> usize { + let cur = self.target.load(Ordering::Acquire); + if cur >= self.max { + return cur; + } + self.sem.add_permits(1); + self.target.store(cur + 1, Ordering::Release); + cur + 1 + } + + /// Shrink the pool by one permit, down to [`min`](Self::min). Acquires a + /// permit (waiting at most [`SHRINK_ACQUIRE_TIMEOUT`]) and permanently + /// forgets it. `target` is decremented ONLY on a successful forget, so a + /// timed-out shrink leaves the size honest and simply retries next window. + /// Returns the new (or unchanged) size. + pub async fn shrink(&self) -> usize { + let cur = self.target.load(Ordering::Acquire); + if cur <= self.min { + return cur; + } + match tokio::time::timeout(SHRINK_ACQUIRE_TIMEOUT, self.sem.clone().acquire_owned()).await { + Ok(Ok(permit)) => { + // Permanently remove this permit from circulation. + permit.forget(); + self.target.store(cur - 1, Ordering::Release); + cur - 1 + } + // Timed out (all permits busy) or the semaphore is closed: no change. + _ => cur, + } + } +} + +// --------------------------------------------------------------------------- +// ThroughputProbe +// --------------------------------------------------------------------------- + +/// A per-account aggregate-upload-throughput accumulator (DESIGN s11.4.7). +/// +/// The executor calls [`record_bytes`](Self::record_bytes) at every completed +/// upload (the same site that feeds the telemetry latency reservoir); the +/// controller drains it once per window with [`take_bytes`](Self::take_bytes) +/// and divides by the injected-clock window duration to get bytes/sec. The +/// accumulator is a single lock-free atomic, so the hot upload path pays only an +/// atomic add and carries NO clock (window boundaries are the controller's job, +/// driven by its injected [`Clock`](crate::time::Clock) - keeping the whole loop +/// deterministic without an `Instant::now()` anywhere on the hot path). +#[derive(Debug, Default)] +pub struct ThroughputProbe { + bytes: AtomicU64, +} + +impl ThroughputProbe { + /// A fresh probe with an empty window. + #[must_use] + pub fn new() -> Arc { + Arc::new(Self::default()) + } + + /// Add `n` uploaded bytes to the current window. Lock-free; safe to call from + /// every concurrent upload task. + pub fn record_bytes(&self, n: u64) { + self.bytes.fetch_add(n, Ordering::Relaxed); + } + + /// Read-and-reset the bytes accumulated this window. + #[must_use] + pub fn take_bytes(&self) -> u64 { + self.bytes.swap(0, Ordering::Relaxed) + } + + /// Peek the accumulated bytes without resetting (idle check). + #[must_use] + pub fn peek_bytes(&self) -> u64 { + self.bytes.load(Ordering::Relaxed) + } +} + +// --------------------------------------------------------------------------- +// Pure control law +// --------------------------------------------------------------------------- + +/// The controller's decision for one window. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Decision { + /// Add one in-flight-file permit (up to the cap). + Grow, + /// Remove one in-flight-file permit (down to the floor). + Shrink, + /// Leave the pool size unchanged. + Hold, +} + +/// The pure inputs to one [`decide`] call (DESIGN s11.4.7). +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct ControllerInput { + /// Aggregate upload throughput this window, bytes/sec. + pub current_bps: f64, + /// The previous window's throughput, or `None` before the first full window. + pub previous_bps: Option, + /// Was the pool the active bottleneck this window (files queued for permits)? + /// The concrete "throughput is at the pool's ceiling" signal. + pub pool_saturated: bool, + /// Was the local disk saturated (DESIGN s18.2) for the window? + pub disk_saturated: bool, + /// Did the pacer throttle (rate-limit / daily-quota) at any point this + /// window? A throttle explains a throughput drop, so it suppresses a shrink. + pub pacer_throttled: bool, + /// The current pool size. + pub current_size: usize, + /// The pool's lower bound. + pub min_size: usize, + /// The pool's upper bound (hard cap). + pub max_size: usize, +} + +/// The adaptive-parallelism control law (DESIGN s11.4.7), as a PURE function so +/// every branch is exhaustively unit-testable. +/// +/// # Rules +/// +/// 1. **Representative-window guard.** Act only when the pool was the active +/// bottleneck (`pool_saturated`) with real throughput. A draining/idle window +/// shows a throughput drop that is NOT congestion; acting on it would shrink a +/// pool that is simply out of work. (This is the guard that keeps the loop off +/// non-representative windows - it never weakens the DESIGN rules below, it +/// only refuses to apply them to noise.) +/// 2. **Shrink** (`current < 50% of previous`, and neither the pacer nor the disk +/// explains the drop, and above the floor): too many files against a congested +/// Drive edge - fewer-in-flight each finish faster (DESIGN s11.4.7). +/// 3. **Grow** (throughput still IMPROVING, disk has headroom, not rate-limited, +/// below the cap): the pool is the ceiling and lifting it is paying off. +/// +/// # "At the pool's ceiling" (resolved ambiguity) +/// +/// DESIGN s11.4.7 says grow when "sustained throughput is at the pool's ceiling +/// AND the disk + CPU have headroom." Taken literally as "pool pinned", a pool +/// stuck at a suboptimal-high size with steady (not falling) throughput would +/// grow every window straight to the cap, since the shrink rule only fires on an +/// active >50% drop. We operationalize "at the pool's ceiling" as *the pool is +/// pinned AND throughput is still improving window-over-window* (by +/// [`GROW_IMPROVE_RATIO`]): growth continues only while it pays off and stops at +/// a plateau, giving a stable additive-increase / multiplicative-decrease loop. +/// The first window (no previous) grows once as a bootstrap probe. +/// +/// # CPU headroom (resolved ambiguity) +/// +/// DESIGN s11.4.7 names "disk + CPU" headroom, but s18.2 defines a concrete +/// signal only for the DISK. Uploads are network-bound and their CPU work +/// (hash + encrypt) runs on a separate rayon pool sized to leave a core for the +/// reactor (DESIGN s11.4.5), so the disk-saturation gate is the operative +/// hardware-headroom signal; no separate dynamic CPU probe is introduced. +#[must_use] +pub fn decide(i: ControllerInput) -> Decision { + // Rule 1: only act on a representative window. + if !i.pool_saturated || i.current_bps <= 0.0 { + return Decision::Hold; + } + + // Rule 2: shrink on an unexplained throughput collapse. + if let Some(prev) = i.previous_bps { + if prev > 0.0 + && i.current_bps < SHRINK_RATIO * prev + && !i.pacer_throttled + && !i.disk_saturated + && i.current_size > i.min_size + { + return Decision::Shrink; + } + } + + // Rule 3: grow while lifting the ceiling is still paying off. + let improving = match i.previous_bps { + None => true, // bootstrap probe: try one step up + Some(prev) => i.current_bps > GROW_IMPROVE_RATIO * prev, + }; + if improving && !i.disk_saturated && !i.pacer_throttled && i.current_size < i.max_size { + return Decision::Grow; + } + + Decision::Hold +} + +// --------------------------------------------------------------------------- +// AdaptiveController (impure shell) +// --------------------------------------------------------------------------- + +/// Per-window accumulators + the previous-window baseline. Only ever touched +/// from the single orchestrator run-loop task, behind a `Mutex` so the shell can +/// be `&self` and never holds the lock across an `.await`. +#[derive(Debug)] +struct WindowState { + /// Injected-clock ms at which the current window opened. + window_start_ms: i64, + /// The previous completed window's throughput (bytes/sec), or `None`. + previous_bps: Option, + /// Disk-busy samples taken this window, and how many read "saturated". + disk_samples: u32, + disk_saturated_samples: u32, +} + +/// The thin impure shell around [`decide`] (DESIGN s11.4.7). The orchestrator +/// calls [`tick`](Self::tick) every [`SAMPLE_INTERVAL`]; the shell samples the +/// disk, and once per [`WINDOW`] drains the throughput probe + contention count, +/// builds a [`ControllerInput`], calls [`decide`], and applies the result to the +/// [`UploadPool`]. +pub struct AdaptiveController { + pool: Arc, + probe: Arc, + disk: Arc, + pacer: Arc, + clock: Arc, + window_ms: i64, + state: Mutex, +} + +impl AdaptiveController { + /// Build a controller over the SAME [`UploadPool`] + [`ThroughputProbe`] the + /// executor holds, the disk-busy probe, the account's pacer, and the clock. + #[must_use] + pub fn new( + pool: Arc, + probe: Arc, + disk: Arc, + pacer: Arc, + clock: Arc, + ) -> Self { + let now = clock.now_ms(); + Self { + pool, + probe, + disk, + pacer, + clock, + window_ms: WINDOW.as_millis() as i64, + state: Mutex::new(WindowState { + window_start_ms: now, + previous_bps: None, + disk_samples: 0, + disk_saturated_samples: 0, + }), + } + } + + /// The upload pool this controller resizes (so the caller can share the same + /// handle into the executor). + #[must_use] + pub fn pool(&self) -> &Arc { + &self.pool + } + + /// One controller tick, called by the orchestrator every [`SAMPLE_INTERVAL`]. + /// + /// Samples the disk (unless the account is idle - no bytes, no contention - + /// in which case the disk syscall is skipped) and, once a full [`WINDOW`] has + /// elapsed on the injected clock, runs a decision. `async` because a shrink + /// awaits a freeing permit; the window-state lock is always released before + /// that await. + pub async fn tick(&self) { + let now = self.clock.now_ms(); + // Cheap idle check: don't syscall the disk on an account doing nothing. + let active = self.probe.peek_bytes() > 0 || self.pool.peek_contended() > 0; + + let decision_input = { + let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner()); + + if active { + let busy = self.disk.sample(); + st.disk_samples = st.disk_samples.saturating_add(1); + if busy.is_saturated() { + st.disk_saturated_samples = st.disk_saturated_samples.saturating_add(1); + } + } + + if now.saturating_sub(st.window_start_ms) < self.window_ms { + // Not a decision tick yet - just accumulated a disk sample. + return; + } + + // --- Decision boundary: drain the window and build the input. --- + let elapsed_ms = now.saturating_sub(st.window_start_ms).max(1); + let bytes = self.probe.take_bytes(); + let contended = self.pool.take_contended(); + let current_bps = (bytes as f64) * 1000.0 / (elapsed_ms as f64); + // Disk is "saturated for the window" if the majority of samples read + // saturated (a single transient spike does not gate the pool). + let disk_saturated = + st.disk_samples > 0 && st.disk_saturated_samples * 2 >= st.disk_samples; + // Throttle is window-scoped: any throttle AT/AFTER the window start, + // even one that has since cleared, counts. + let pacer_throttled = self.pacer.last_throttle_ms() >= st.window_start_ms; + + let input = ControllerInput { + current_bps, + previous_bps: st.previous_bps, + pool_saturated: contended > 0, + disk_saturated, + pacer_throttled, + current_size: self.pool.target(), + min_size: self.pool.min(), + max_size: self.pool.max(), + }; + + // Roll the window forward. Only a REPRESENTATIVE window updates the + // baseline (F3): the same guard `decide` rule 1 acts on + // (`pool_saturated && current_bps > 0`). A draining / idle / + // non-saturated window is not a valid throughput reference - letting + // it become `previous_bps` would skew the NEXT window's shrink/grow + // ratio (a genuinely-congested window measured against a low + // non-representative baseline can miss a real shrink, or over-grow + // against an anomalously low one). Non-representative windows leave + // the last representative baseline intact. + if input.pool_saturated && input.current_bps > 0.0 { + st.previous_bps = Some(current_bps); + } + st.window_start_ms = now; + st.disk_samples = 0; + st.disk_saturated_samples = 0; + + input + }; // lock dropped here, before any await + + match decide(decision_input) { + Decision::Grow => { + let size = self.pool.grow(); + tracing::debug!( + target: "driven::adaptive", + new_size = size, + throughput_bps = decision_input.current_bps, + "adaptive parallelism: grew upload pool" + ); + } + Decision::Shrink => { + let size = self.pool.shrink().await; + tracing::debug!( + target: "driven::adaptive", + new_size = size, + throughput_bps = decision_input.current_bps, + "adaptive parallelism: shrank upload pool" + ); + } + Decision::Hold => {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + // The in-crate `FakeClock` (same `driven_core` instance) - NOT the fixtures' + // one: driven-core's unit tests link a second `driven_core` via the + // `driven-test-fixtures` dev-dependency cycle, so `driven_test_fixtures`'s + // `FakeClock` implements a DIFFERENT `Clock` and won't coerce here. See + // `crate::test_support`. `FakeDiskBusyProbe` is fine (its `DiskBusyProbe` + // trait lives in the non-cyclic `driven-diskstat`, a single instance). + use crate::test_support::FakeClock; + use driven_test_fixtures::diskstat::FakeDiskBusyProbe; + + // --- pure `decide` ---------------------------------------------------- + + /// A baseline input: pinned pool, real throughput, disk + pacer clear, room + /// to move in both directions. Individual tests override one field. + fn base() -> ControllerInput { + ControllerInput { + current_bps: 1_000_000.0, + previous_bps: Some(1_000_000.0), + pool_saturated: true, + disk_saturated: false, + pacer_throttled: false, + current_size: 8, + min_size: MIN_POOL, + max_size: MAX_POOL, + } + } + + #[test] + fn holds_when_pool_not_saturated() { + // A draining window (throughput fell) must NOT shrink if the pool was not + // the bottleneck. + let i = ControllerInput { + pool_saturated: false, + current_bps: 100.0, + previous_bps: Some(1_000_000.0), + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + #[test] + fn holds_when_zero_throughput() { + let i = ControllerInput { + current_bps: 0.0, + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + #[test] + fn shrinks_on_unexplained_collapse() { + let i = ControllerInput { + current_bps: 400_000.0, // < 50% of 1_000_000 + previous_bps: Some(1_000_000.0), + ..base() + }; + assert_eq!(decide(i), Decision::Shrink); + } + + #[test] + fn collapse_but_throttled_holds() { + // The pacer explains the drop -> not our concurrency's fault. + let i = ControllerInput { + current_bps: 400_000.0, + pacer_throttled: true, + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + #[test] + fn collapse_but_disk_saturated_holds() { + let i = ControllerInput { + current_bps: 400_000.0, + disk_saturated: true, + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + #[test] + fn collapse_at_floor_holds() { + let i = ControllerInput { + current_bps: 400_000.0, + current_size: MIN_POOL, + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + #[test] + fn grows_when_improving_with_headroom() { + let i = ControllerInput { + current_bps: 2_000_000.0, // > 1.05x previous + previous_bps: Some(1_000_000.0), + ..base() + }; + assert_eq!(decide(i), Decision::Grow); + } + + #[test] + fn bootstrap_grows_on_first_window() { + let i = ControllerInput { + previous_bps: None, + ..base() + }; + assert_eq!(decide(i), Decision::Grow); + } + + #[test] + fn flat_throughput_holds() { + // Neither a collapse nor an improvement -> a plateau -> settle. + let i = ControllerInput { + current_bps: 1_000_000.0, + previous_bps: Some(1_000_000.0), + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + #[test] + fn grow_blocked_by_disk_saturation() { + let i = ControllerInput { + current_bps: 2_000_000.0, + disk_saturated: true, + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + #[test] + fn grow_blocked_by_throttle() { + let i = ControllerInput { + current_bps: 2_000_000.0, + pacer_throttled: true, + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + #[test] + fn grow_blocked_at_cap() { + let i = ControllerInput { + current_bps: 2_000_000.0, + current_size: MAX_POOL, + ..base() + }; + assert_eq!(decide(i), Decision::Hold); + } + + // --- UploadPool ------------------------------------------------------- + + #[test] + fn pool_new_clamps_start_into_bounds() { + assert_eq!(UploadPool::with_bounds(0, 1, 32).target(), 1); + assert_eq!(UploadPool::with_bounds(999, 1, 32).target(), 32); + assert_eq!(UploadPool::new(50).target(), MAX_POOL); // start clamped to cap + } + + #[tokio::test] + async fn pool_grows_to_cap_and_stops() { + let pool = UploadPool::with_bounds(1, 1, 3); + assert_eq!(pool.grow(), 2); + assert_eq!(pool.grow(), 3); + assert_eq!(pool.grow(), 3, "must not grow past the cap"); + // The extra permits are really available for acquisition. + let _a = pool.try_acquire_owned().unwrap(); + let _b = pool.try_acquire_owned().unwrap(); + let _c = pool.try_acquire_owned().unwrap(); + assert!(pool.try_acquire_owned().is_none(), "only 3 permits exist"); + } + + #[tokio::test] + async fn pool_shrinks_to_floor_and_stops() { + let pool = UploadPool::with_bounds(3, 1, 3); + assert_eq!(pool.shrink().await, 2); + assert_eq!(pool.shrink().await, 1); + assert_eq!(pool.shrink().await, 1, "must not shrink past the floor"); + // Only one permit remains after shrinking to the floor. + let _a = pool.try_acquire_owned().unwrap(); + assert!(pool.try_acquire_owned().is_none()); + } + + #[tokio::test] + async fn pool_counts_contention() { + let pool = UploadPool::with_bounds(1, 1, 4); + let _held = pool.acquire_owned().await.unwrap(); // uncontended + assert_eq!(pool.peek_contended(), 0); + // No permit free -> this contends (times out, but the count already rose). + let _ = tokio::time::timeout(Duration::from_millis(5), pool.acquire_owned()).await; + assert_eq!(pool.take_contended(), 1); + assert_eq!(pool.take_contended(), 0, "take resets"); + } + + // --- ThroughputProbe -------------------------------------------------- + + #[test] + fn probe_accumulates_and_drains() { + let p = ThroughputProbe::new(); + p.record_bytes(100); + p.record_bytes(50); + assert_eq!(p.peek_bytes(), 150); + assert_eq!(p.take_bytes(), 150); + assert_eq!(p.take_bytes(), 0); + } + + // --- AdaptiveController end-to-end (deterministic) -------------------- + + /// Drive a full window: record `bytes` of upload, force the pool to register + /// contention (so it reads as the bottleneck), advance the clock one window, + /// and tick the controller to a decision. + async fn run_window(ctrl: &AdaptiveController, clock: &FakeClock, bytes: u64) { + ctrl.probe.record_bytes(bytes); + // Pin every permit, then contend once so `pool_saturated` is true. + let held: Vec<_> = (0..ctrl.pool.target()) + .filter_map(|_| ctrl.pool.try_acquire_owned()) + .collect(); + let _ = tokio::time::timeout(Duration::from_millis(5), ctrl.pool.acquire_owned()).await; + drop(held); + clock.advance(WINDOW); + ctrl.tick().await; + } + + /// Drive a NON-representative window: record `bytes` but never contend the + /// pool, so `pool_saturated` is false. `decide` Holds on it (rule 1) and, + /// with the F3 fix, it must not become the baseline. + async fn run_window_unsaturated(ctrl: &AdaptiveController, clock: &FakeClock, bytes: u64) { + ctrl.probe.record_bytes(bytes); + clock.advance(WINDOW); + ctrl.tick().await; + } + + #[tokio::test] + async fn controller_shrinks_on_latency_then_recovers() { + let clock = Arc::new(FakeClock::new()); + let pool = UploadPool::new(8); + let probe = ThroughputProbe::new(); + let disk: Arc = Arc::new(FakeDiskBusyProbe::not_saturated()); + let pacer: Arc = Arc::new(crate::pacer::AimdPacer::new(clock.clone(), None)); + let ctrl = AdaptiveController::new(pool.clone(), probe.clone(), disk, pacer, clock.clone()); + + // Warm up to a steady high-throughput baseline (bootstrap grow + plateau). + run_window(&ctrl, &clock, 60_000_000).await; // establishes previous + run_window(&ctrl, &clock, 60_000_000).await; // flat -> settle + let before_latency = pool.target(); + + // Induce latency: throughput collapses well below 50% of the baseline. + run_window(&ctrl, &clock, 3_000_000).await; + let after_latency = pool.target(); + assert!( + after_latency < before_latency, + "induced latency must shrink the pool: {before_latency} -> {after_latency}" + ); + + // Clear the latency: throughput jumps back up (improving) -> pool recovers. + run_window(&ctrl, &clock, 60_000_000).await; + run_window(&ctrl, &clock, 90_000_000).await; + let recovered = pool.target(); + assert!( + recovered > after_latency, + "restoring throughput must regrow the pool: {after_latency} -> {recovered}" + ); + } + + #[tokio::test] + async fn controller_holds_below_full_window() { + // A tick before a full 30s window must not change the pool. + let clock = Arc::new(FakeClock::new()); + let pool = UploadPool::new(8); + let probe = ThroughputProbe::new(); + let disk: Arc = Arc::new(FakeDiskBusyProbe::not_saturated()); + let pacer: Arc = Arc::new(crate::pacer::AimdPacer::new(clock.clone(), None)); + let ctrl = AdaptiveController::new(pool.clone(), probe.clone(), disk, pacer, clock.clone()); + + probe.record_bytes(60_000_000); + clock.advance(SAMPLE_INTERVAL); // only 5s, not a window + ctrl.tick().await; + assert_eq!(pool.target(), 8, "no decision before a full window"); + } + + #[tokio::test] + async fn controller_does_not_shrink_while_throttled() { + let clock = Arc::new(FakeClock::new()); + let pool = UploadPool::new(8); + let probe = ThroughputProbe::new(); + let disk: Arc = Arc::new(FakeDiskBusyProbe::not_saturated()); + let pacer: Arc = Arc::new(crate::pacer::AimdPacer::new(clock.clone(), None)); + let ctrl = AdaptiveController::new( + pool.clone(), + probe.clone(), + disk, + pacer.clone(), + clock.clone(), + ); + + run_window(&ctrl, &clock, 60_000_000).await; + run_window(&ctrl, &clock, 60_000_000).await; + let before = pool.target(); + // Throttle DURING the collapse window: the drop is explained by the pacer. + pacer.note_response(crate::pacer::ResponseClass::RateLimited { + retry_after: Duration::from_secs(1), + }); + run_window(&ctrl, &clock, 3_000_000).await; + assert_eq!( + pool.target(), + before, + "a throttle-explained drop must not shrink" + ); + } + + #[tokio::test] + async fn baseline_not_contaminated_by_nonrepresentative_window() { + // F3: a non-representative window must not become the throughput baseline. + let clock = Arc::new(FakeClock::new()); + let pool = UploadPool::new(8); + let probe = ThroughputProbe::new(); + let disk: Arc = Arc::new(FakeDiskBusyProbe::not_saturated()); + let pacer: Arc = Arc::new(crate::pacer::AimdPacer::new(clock.clone(), None)); + let ctrl = AdaptiveController::new(pool.clone(), probe.clone(), disk, pacer, clock.clone()); + + // Two representative HIGH windows establish + settle a high baseline + // (60 MB / 30 s = 2 MB/s). + run_window(&ctrl, &clock, 60_000_000).await; + run_window(&ctrl, &clock, 60_000_000).await; + let size_before = pool.target(); + + // A NON-representative low window (pool not the bottleneck): `decide` + // Holds (rule 1), and the baseline must stay at the HIGH value - not roll + // down to this window's low throughput. + run_window_unsaturated(&ctrl, &clock, 1_000_000).await; + assert_eq!( + pool.target(), + size_before, + "a non-representative window must not resize the pool" + ); + + // A representative window at ~40% of the HIGH baseline (24 MB / 30 s = + // 0.8 MB/s < 50% of 2 MB/s) must SHRINK - which only happens if the + // baseline is still the intact HIGH value. Had the low non-representative + // window contaminated `previous_bps` (to ~0.033 MB/s), 0.8 MB/s would read + // as a >1.05x improvement and GROW instead, so this assertion is the + // discriminating check between the fixed and unfixed behaviour. + run_window(&ctrl, &clock, 24_000_000).await; + assert!( + pool.target() < size_before, + "a collapse measured against the intact high baseline must shrink: \ + {size_before} -> {}", + pool.target() + ); + } +} diff --git a/crates/driven-core/src/executor.rs b/crates/driven-core/src/executor.rs index 027c57f9..f26853fa 100644 --- a/crates/driven-core/src/executor.rs +++ b/crates/driven-core/src/executor.rs @@ -39,7 +39,6 @@ use driven_drive::remote_store::{ use driven_vss::{fallback_decision, FallbackDecision, OpenAttempt, SnapshotOutcome, VssMode}; use serde::{Deserialize, Serialize}; use tokio::io::AsyncReadExt; -use tokio::sync::Semaphore; use tracing::{debug, warn}; use crate::network::{NetworkProbe, ServiceName}; @@ -777,16 +776,6 @@ impl RemoteStore for BreakerReportingStore { // DefaultExecutor // ----------------------------------------------------------------------------- -/// The number of in-flight files permitted concurrently (DESIGN s11.4.2: -/// `min(available_parallelism * 2, 16)`, hard cap 32). Computed once at -/// construction. -fn default_pool_size() -> usize { - let par = std::thread::available_parallelism() - .map(|n| n.get()) - .unwrap_or(4); - (par.saturating_mul(2)).min(16).clamp(1, 32) -} - /// Test-only hook fired exactly once, between the pre-open `lstat`/open and /// the post-read `fstat` identity check, so a test can deterministically /// mutate or replace the file on disk and exercise the @@ -806,8 +795,8 @@ type PostUploadHook = Arc; /// The production [`Executor`] (SPEC s8, DESIGN s5.4 / s5.6 / s11.4). /// -/// Holds the injected seams plus an [`UploadPool`](Semaphore) bounding -/// in-flight files (DESIGN s11.4.2) and a [`Clock`] for the timestamps +/// Holds the injected seams plus an [`UploadPool`](crate::adaptive::UploadPool) +/// bounding in-flight files (DESIGN s11.4.2) and a [`Clock`] for the timestamps /// written into `file_state` / `pending_ops`. Cheap to clone-by-`Arc` /// internally; the orchestrator holds it behind `Arc`. pub struct DefaultExecutor { @@ -827,8 +816,18 @@ pub struct DefaultExecutor { /// Per-cycle VSS snapshot provider (ROADMAP M3.5), or `None` to disable /// the locked-file fallback (then a locked file is skipped as before). vss: Option>, - /// Inter-file concurrency gate (DESIGN s11.4.2). `acquire`d per op. - pool: Arc, + /// Inter-file concurrency gate (DESIGN s11.4.2), RESIZABLE by the adaptive + /// controller (DESIGN s11.4.7). `acquire`d per op; the executor never resizes + /// it (that is the controller's job via the SAME shared `Arc`). Constructed + /// at the default size unless the app injects a pre-sized pool via + /// [`Self::with_upload_pool`]. + pool: Arc, + /// Aggregate-upload-throughput accumulator (DESIGN s11.4.7), or `None` when + /// adaptive parallelism is not wired (every test that does not exercise it + + /// the chaos harness). When `Some`, each completed upload records its byte + /// count here for the controller's throughput window. Mirrors the + /// `latency` reservoir seam. + throughput: Option>, /// Test-only peak-memory gauge for the streaming pipeline (P1-4 /// acceptance). `None` in production (zero overhead beyond an /// `Option::is_none` check per chunk). Exposed via the doc-hidden @@ -897,7 +896,7 @@ impl DefaultExecutor { /// Builds a [`DefaultExecutor`] with an explicit [`Clock`] (tests /// inject a `FakeClock` so the timestamps are deterministic). pub fn with_clock(deps: ExecutorDeps, clock: Arc) -> Self { - let pool = Arc::new(Semaphore::new(default_pool_size())); + let pool = crate::adaptive::UploadPool::new(crate::adaptive::default_pool_size()); // CODEX_NOTES P2-9: when a NetworkProbe is injected, route every Drive // request through the BreakerReportingStore so the Drive circuit // breaker is driven by REAL request outcomes (not just probes). When @@ -920,6 +919,7 @@ impl DefaultExecutor { clock, vss: deps.vss, pool, + throughput: None, mem_gauge: None, latency: None, #[cfg(test)] @@ -944,6 +944,27 @@ impl DefaultExecutor { self } + /// Inject the resizable [`UploadPool`](crate::adaptive::UploadPool) the + /// adaptive controller (DESIGN s11.4.7) also holds, sized at the account's + /// `default_concurrent_uploads` setting. The app wires the SAME `Arc` here + /// and into the controller so a resize is seen by this executor's acquire + /// path. Every other construction path keeps the default-sized internal pool. + #[must_use] + pub fn with_upload_pool(mut self, pool: Arc) -> Self { + self.pool = pool; + self + } + + /// Attach the [`ThroughputProbe`](crate::adaptive::ThroughputProbe) the + /// adaptive controller drains (DESIGN s11.4.7). Mirrors + /// [`Self::with_latency_reservoir`]: each completed upload records its byte + /// count into the probe. `None` (the default) makes the record a no-op. + #[must_use] + pub fn with_throughput_probe(mut self, probe: Arc) -> Self { + self.throughput = Some(probe); + self + } + /// Resolve the crypto decision for one source (M5 GA-blocking surface). /// /// Consults the injected [`CryptoProvider`]; a `None` provider means every @@ -1209,6 +1230,11 @@ impl DefaultExecutor { source: &SourceRow, relative_path: &RelativePath, size: u64, + // F2: per-op accumulator of wire bytes already credited to the + // throughput probe during a resumable stream (chunk-granular). The + // completion site subtracts it so those bytes are not counted twice; left + // at 0 for every non-streamed path. + recorded: &std::sync::atomic::AtomicU64, ) -> anyhow::Result { // --- GA-critical FAIL-CLOSED crypto resolution (M5, DESIGN s7) ------ // Resolve this source's suite FIRST, before opening the file or @@ -1369,6 +1395,7 @@ impl DefaultExecutor { from_vss, crypto, version_supersede, + recorded, ) .await; @@ -1770,6 +1797,10 @@ impl DefaultExecutor { // object, and the old object is trashed. `None` for a normal create / // in-place update. version: Option, + // F2: threaded through to the resumable streamer so each acked wire chunk + // credits the throughput probe as it lands; the op's completion site + // subtracts this to avoid double-counting. 0 for non-streamed paths. + recorded: &std::sync::atomic::AtomicU64, ) -> Result { // M3.5: every FS recheck below reads the EFFECTIVE path (the VSS // snapshot copy for a locked file), so a frozen-vs-live stat mismatch @@ -1846,6 +1877,7 @@ impl DefaultExecutor { &mut payload, allow_resumable, crypto.clone(), + recorded, ) .await? } else { @@ -2134,6 +2166,9 @@ impl DefaultExecutor { // plaintext. Resolved (FAIL-CLOSED) per op by the caller; owned so the // cpu stage can move it. crypto: Option>, + // F2: see `hash_then_upload` - wire bytes credited per chunk are added + // here so the op's completion site can avoid double-counting them. + recorded: &std::sync::atomic::AtomicU64, ) -> Result { // Predict the exact number of bytes that will be sent to Drive. let encrypted = crypto.is_some(); @@ -2169,6 +2204,7 @@ impl DefaultExecutor { op_id, payload, allow_resumable, + recorded, ); // Run all three concurrently. The cpu stage returns (blake3, md5); @@ -2249,6 +2285,9 @@ impl DefaultExecutor { // P1-B: false for a VSS-snapshot read - never open a resumable session, // stream as a single simple upload even above RESUMABLE_THRESHOLD. allow_resumable: bool, + // F2: only the resumable branch streams per-chunk and credits `recorded`; + // the simple branch is a single request counted whole at op completion. + recorded: &std::sync::atomic::AtomicU64, ) -> Result { if allow_resumable && total >= RESUMABLE_THRESHOLD { self.upload_stage_resumable( @@ -2259,6 +2298,7 @@ impl DefaultExecutor { out_rx, op_id, payload, + recorded, ) .await } else { @@ -2349,6 +2389,9 @@ impl DefaultExecutor { mut out_rx: tokio::sync::mpsc::Receiver, op_id: PendingOpId, payload: &mut PendingOpPayload, + // F2: each acked wire chunk credits its bytes here (and to the throughput + // probe) as it lands, for in-window throughput on multi-window files. + recorded: &std::sync::atomic::AtomicU64, ) -> Result { let session = self .open_resumable_session(target, existing_file_id, mime, total, op_id, payload) @@ -2376,7 +2419,7 @@ impl DefaultExecutor { let wire = Bytes::copy_from_slice(&acc[..WIRE_CHUNK]); acc.drain(..WIRE_CHUNK); match self - .push_one_wire_chunk(&session, offset, wire, op_id, payload) + .push_one_wire_chunk(&session, offset, wire, op_id, payload, recorded) .await? { PushOne::Acked(new_off) => offset = new_off, @@ -2414,7 +2457,7 @@ impl DefaultExecutor { let wire = Bytes::copy_from_slice(&acc[..take]); acc.drain(..take); match self - .push_one_wire_chunk(&session, offset, wire, op_id, payload) + .push_one_wire_chunk(&session, offset, wire, op_id, payload, recorded) .await? { PushOne::Acked(new_off) => offset = new_off, @@ -2447,6 +2490,8 @@ impl DefaultExecutor { wire: Bytes, op_id: PendingOpId, payload: &mut PendingOpPayload, + // F2: this chunk's wire bytes are credited here once Drive accepts it. + recorded: &std::sync::atomic::AtomicU64, ) -> Result { let wire_len = wire.len() as u64; self.pacer.permit_request().await; @@ -2504,6 +2549,16 @@ impl DefaultExecutor { if let Some(g) = self.mem_gauge.as_ref() { g.sub(wire_len); } + // F2: credit these wire bytes to the CURRENT throughput window as the + // chunk lands - not all at op completion - so a large file spanning + // multiple 30 s windows reports real in-window throughput instead of a + // completion-time spike (and 0-byte windows in between). `recorded` + // accumulates the same bytes so the op's completion site subtracts them + // and never double-counts. + if let Some(probe) = self.throughput.as_ref() { + probe.record_bytes(wire_len); + } + recorded.fetch_add(wire_len, std::sync::atomic::Ordering::Relaxed); match progress { ResumeProgress::Completed(entry) => Ok(PushOne::Done(entry)), ResumeProgress::InProgress { received } => { @@ -4532,6 +4587,11 @@ impl<'a> ExecOne<'a> { Op::Trash { .. } => None, }; + // F2: per-op tally of wire bytes already credited to the throughput probe + // mid-upload (chunk-granular, resumable path only). Stays 0 for every + // non-streamed path, so the completion site below credits the whole file. + let recorded = std::sync::atomic::AtomicU64::new(0); + let out = match op { Op::HashThenUpload { source_id, @@ -4540,7 +4600,7 @@ impl<'a> ExecOne<'a> { } => { debug_assert_eq!(*source_id, self.source.id); self.this - .hash_then_upload(self.source, relative_path, *size) + .hash_then_upload(self.source, relative_path, *size, &recorded) .await } Op::Trash { @@ -4559,22 +4619,37 @@ impl<'a> ExecOne<'a> { } }; - // Record the completed upload's per-MiB latency (DESIGN s13). Only a - // successful upload that moved bytes contributes a sample: a Done-Upload - // carries the uploaded byte count, a BundleDone carries the bundle object - // size; a trash, skip, or failure records nothing. - if let (Some(reservoir), Some(started)) = (self.this.latency.as_ref(), upload_timer) { - if let Ok(outcome) = &out { - let bytes = match outcome { - OpOutcome::Done { - kind: DoneKind::Upload, - bytes: Some(bytes), - .. - } => Some(*bytes), - OpOutcome::BundleDone { bytes, .. } => Some(*bytes), - _ => None, - }; - if let Some(bytes) = bytes { + // Record completed-upload metrics. Only a successful upload that moved + // bytes contributes: a Done-Upload carries the uploaded byte count, a + // BundleDone the bundle-object size; a trash, skip, or failure records + // nothing. Two independent sinks read the same byte count: + // - the ThroughputProbe (adaptive parallelism, DESIGN s11.4.7) - fed + // unconditionally whenever it is wired, no timer needed; and + // - the per-MiB latency reservoir (DESIGN s13) - only when armed + + // enabled (its `upload_timer` is `Some`). + if let Ok(outcome) = &out { + let bytes = match outcome { + OpOutcome::Done { + kind: DoneKind::Upload, + bytes: Some(bytes), + .. + } => Some(*bytes), + OpOutcome::BundleDone { bytes, .. } => Some(*bytes), + _ => None, + }; + if let Some(bytes) = bytes { + if let Some(probe) = self.this.throughput.as_ref() { + // F2: subtract the bytes already credited per-chunk during a + // resumable stream (`push_one_wire_chunk`), so a large + // multi-window file is not counted twice. Non-streamed ops + // (small / simple-band / VSS single-shot / bundle) left + // `recorded` at 0, so the whole file is credited here, in its + // completion window. + let streamed = recorded.load(std::sync::atomic::Ordering::Relaxed); + probe.record_bytes(bytes.saturating_sub(streamed)); + } + if let (Some(reservoir), Some(started)) = (self.this.latency.as_ref(), upload_timer) + { let elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX); if let Some(per_mb) = crate::telemetry::per_mb_ms(elapsed_ms, bytes) { diff --git a/crates/driven-core/src/lib.rs b/crates/driven-core/src/lib.rs index 39b54392..4dd6b1e0 100644 --- a/crates/driven-core/src/lib.rs +++ b/crates/driven-core/src/lib.rs @@ -19,6 +19,7 @@ //! [`network::NetworkProbe`] traits - with no behaviour; the bodies land in //! the M3 implement phase. +pub mod adaptive; pub mod bundle; pub mod crypto_provider; pub mod exclude; diff --git a/crates/driven-core/src/migrations/0011_adaptive_parallelism.sql b/crates/driven-core/src/migrations/0011_adaptive_parallelism.sql new file mode 100644 index 00000000..7ad060b0 --- /dev/null +++ b/crates/driven-core/src/migrations/0011_adaptive_parallelism.sql @@ -0,0 +1,18 @@ +-- Add the `adaptive_parallelism_enabled` key (DESIGN s11.4.7) to the persisted +-- `global` settings blob, defaulting it ON. The `global` group is seeded by +-- migration 0002, whose `INSERT OR IGNORE` never re-runs, so a new key can only +-- be introduced by a new additive migration (same pattern as 0005). This +-- backfills the key on every install - new or existing - so the persisted blob +-- matches the DTO. The host code ALSO tolerates the key's absence (the +-- `#[serde(default)]` on both `storage::Global` and `GlobalSettings` reads a +-- missing key as `true`), so this migration is belt-and-braces, not correctness- +-- critical; it keeps the on-disk shape complete. +-- +-- Only set it when absent so a value is never clobbered (no pre-existing value is +-- possible before this feature, but the guard makes the migration idempotent in +-- intent). Data-only (no schema/table change), so no `.sqlx` regeneration and no +-- table-list snapshot update. Runs exactly once per DB. +UPDATE settings +SET value = json_set(value, '$.adaptive_parallelism_enabled', json('true')) +WHERE key = 'global' + AND json_extract(value, '$.adaptive_parallelism_enabled') IS NULL; diff --git a/crates/driven-core/src/orchestrator.rs b/crates/driven-core/src/orchestrator.rs index 089bec35..28fbaeed 100644 --- a/crates/driven-core/src/orchestrator.rs +++ b/crates/driven-core/src/orchestrator.rs @@ -197,6 +197,18 @@ pub struct OrchestratorConfig { /// [`MeteredMode::Throttle`]. `None` falls back to the normal /// [`bandwidth_cap_mbps`](Self::bandwidth_cap_mbps). pub metered_bandwidth_cap_mbps: Option, + /// Starting in-flight-file pool size (DESIGN s11.4.2 / s11.4.7), or `None` + /// to auto-pick `min(available_parallelism * 2, 16)`. The app wires this + /// into the executor's [`UploadPool`](crate::adaptive::UploadPool) as its + /// START size; with adaptive parallelism ON it then floats within + /// `[1, 32]`, and with it OFF the pool stays fixed here. (`default_concurrent_uploads` + /// in the SPEC s22 settings.) + pub default_concurrent_uploads: Option, + /// Whether the adaptive upload-parallelism controller runs (DESIGN s11.4.7). + /// `true` (default) builds the controller so the pool adapts; `false` is the + /// kill-switch - the pool stays FIXED at + /// [`default_concurrent_uploads`](Self::default_concurrent_uploads). + pub adaptive_parallelism_enabled: bool, } /// What Driven does on a metered network when @@ -244,6 +256,8 @@ impl Default for OrchestratorConfig { hook_timeout_secs: 60, metered_mode: MeteredMode::Pause, metered_bandwidth_cap_mbps: None, + default_concurrent_uploads: None, + adaptive_parallelism_enabled: true, } } } @@ -440,6 +454,13 @@ pub struct SyncOrchestrator { /// (via [`crate::executor::DefaultExecutor::with_latency_reservoir`]) for the /// upload-per-MB metric. Set via [`Self::with_latency_reservoir`]. latency: Option>, + /// Adaptive upload-parallelism controller (DESIGN s11.4.7), or `None` when + /// adaptive parallelism is disabled (`adaptive_parallelism_enabled = false`) + /// or not wired (tests / chaos harness). When `Some`, the run loop ticks it + /// every [`crate::adaptive::SAMPLE_INTERVAL`] so it can sample the disk and, + /// once per throughput window, resize the SAME [`UploadPool`](crate::adaptive::UploadPool) + /// the executor acquires from. Set via [`Self::with_adaptive_controller`]. + adaptive: Option>, } impl SyncOrchestrator { @@ -491,9 +512,27 @@ impl SyncOrchestrator { orphan_cleanup_done: Mutex::new(false), suspended: std::sync::atomic::AtomicBool::new(false), latency: None, + adaptive: None, } } + /// Attach the adaptive upload-parallelism controller (DESIGN s11.4.7). The + /// controller resizes the SAME [`UploadPool`](crate::adaptive::UploadPool) + /// the executor acquires from (wired via + /// [`DefaultExecutor::with_upload_pool`](crate::executor::DefaultExecutor::with_upload_pool)) + /// and drains the SAME [`ThroughputProbe`](crate::adaptive::ThroughputProbe) + /// the executor feeds. When present, the run loop ticks it every + /// [`crate::adaptive::SAMPLE_INTERVAL`]; when absent (the kill-switch is off, + /// or a test), the pool stays fixed at its construction size. + #[must_use] + pub fn with_adaptive_controller( + mut self, + controller: Arc, + ) -> Self { + self.adaptive = Some(controller); + self + } + /// Attach the app-global latency reservoir (DESIGN s13 telemetry) so each /// scan records per-file processing latency. Pass the SAME `Arc` given to the /// executor via @@ -1992,6 +2031,46 @@ impl Orchestrator for SyncOrchestrator { // trigger or the next period drives the first real cycle). scheduled.tick().await; + // Adaptive upload-parallelism sampler (DESIGN s11.4.7 / s18.2). Spawned + // as a SEPARATE task - NOT a run-loop select arm - so it keeps ticking + // every SAMPLE_INTERVAL WHILE a cycle runs. A sync cycle is awaited INLINE + // below (the single-in-flight guard), so a select arm could only fire in + // the gaps BETWEEN cycles - starving the controller during the exact + // long-backup workload it targets (F1: DESIGN 11.4.7's pathological + // "16 in flight but Drive's edge is overloaded" case IS one long cycle). + // The task owns an `Arc` whose seams are all + // `Arc`/`dyn` (pool, probe, pacer, clock, disk) - it borrows nothing of + // the orchestrator's `&self` - and resizes the SAME `Arc` the + // executor (running inside the inline cycle, on a different task) acquires + // from. That cross-task resize is race-free: the pool is a `Semaphore`, + // the probe/pacer are atomics, and the controller's `WindowState` is + // touched only by this one sampler task. It observes the shared shutdown + // signal for a prompt cooperative stop (bounded by the <=100 ms shrink + // acquire) and is JOINED before `run` returns (below), so it can never be + // orphaned. Spawned only when a controller is wired (kill-switch on); + // disabled = no task at all. + let adaptive_sampler = self.adaptive.clone().map(|ctrl| { + let mut sampler_shutdown = self.shutdown_rx.clone(); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(crate::adaptive::SAMPLE_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + ticker.tick().await; // consume the immediate first tick + loop { + tokio::select! { + _ = ticker.tick() => ctrl.tick().await, + res = sampler_shutdown.changed() => { + // Flip-to-set or sender-drop: stop cooperatively. (The + // suspended-exit path, which never sets this signal, is + // covered by the abort-join after the run loop.) + if res.is_err() || *sampler_shutdown.borrow() { + break; + } + } + } + } + }) + }); + loop { // Pick the next wake. Each arm yields an `Option`: // `Some(tick)` runs a cycle; `None` means "loop control only" @@ -2125,6 +2204,18 @@ impl Orchestrator for SyncOrchestrator { } } + // Drain the adaptive sampler (F1): `run` never returns while it is still + // alive, so it can never be orphaned. A normal shutdown already told it to + // stop via the shared signal; the suspended-exit path (invalid_grant) did + // not, so `abort` covers both. Aborting mid-tick is safe - the + // `WindowState` mutex is dropped before any await, and a cancelled + // `shrink().await` forgets no permit, so the pool's accounting stays + // consistent. + if let Some(handle) = adaptive_sampler { + handle.abort(); + let _ = handle.await; + } + Ok(()) } @@ -5120,4 +5211,112 @@ mod tests { .expect("join"); assert!(result.is_ok(), "clean shutdown returns Ok(())"); } + + #[tokio::test(start_paused = true)] + async fn adaptive_sampler_resizes_pool_during_an_inflight_cycle() { + // F1: the controller must keep sampling WHILE a sync cycle runs, not only + // in the gaps between cycles. A cycle is awaited INLINE in the run loop, + // so the old select-arm tick could never fire mid-cycle - exactly the + // long-backup workload the controller targets. Here a `BlockingExecutor` + // pins a cycle in-flight; the SPAWNED sampler must still tick and RESIZE + // the shared pool before the cycle is released. This proves sampler + // LIVENESS during a cycle (the regression F1 fixes); the byte/throughput + // math is covered by the `adaptive.rs` unit tests + the e2e transition + // test, so this window's throughput + contention are injected directly + // into the shared pool/probe rather than driven through the blocked + // executor. + use crate::adaptive::{AdaptiveController, ThroughputProbe, UploadPool, WINDOW}; + use driven_test_fixtures::diskstat::FakeDiskBusyProbe; + + let account = AccountId::new_v4(); + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("a.txt"), b"hello").unwrap(); + let mut src = source_in(account, dir.path()); + src.last_deep_verify_at = None; + src.last_full_scan_at = None; + let state = Arc::new(FakeState::with_sources(vec![src])); + + // The SAME pool + probe the controller resizes (the executor would acquire + // from it in production) and the injected clock the window math reads. + let pool = UploadPool::new(4); + let probe = ThroughputProbe::new(); + let clock = Arc::new(FakeClock::new()); + let disk = Arc::new(FakeDiskBusyProbe::not_saturated()); + let pacer = Arc::new(crate::pacer::AimdPacer::new(clock.clone(), None)); + let ctrl = Arc::new(AdaptiveController::new( + pool.clone(), + probe.clone(), + disk, + pacer, + clock.clone(), + )); + + let (entered_tx, mut entered_rx) = tokio::sync::mpsc::unbounded_channel(); + let (release_tx, release_rx) = tokio::sync::mpsc::unbounded_channel(); + let exec = Arc::new(BlockingExecutor { + executes: Arc::new(AtomicU64::new(0)), + entered_tx, + release_rx: tokio::sync::Mutex::new(release_rx), + }); + + let orch = Arc::new( + SyncOrchestrator::new( + account, + state, + exec, + Arc::new(FakePowerSource::new(power_on_ac())), + Arc::new(FakeNet::online()), + clock.clone(), + OrchestratorConfig::default(), + ) + .with_adaptive_controller(ctrl), + ); + + let handle = { + let orch = orch.clone(); + tokio::spawn(async move { orch.run().await }) + }; + orch.trigger(TickSource::Manual).await; + + // Wait until the cycle is IN-FLIGHT (blocked inside `execute`). + tokio::time::timeout(std::time::Duration::from_secs(30), entered_rx.recv()) + .await + .expect("a cycle must enter execute") + .expect("entered channel open"); + + // Inject one representative window's worth of throughput + contention into + // the SHARED pool/probe, then advance the injected clock a full window so + // the next sampler tick reaches a decision boundary. (Earlier sampler ticks + // ran with the clock still at 0 and returned before the window boundary, so + // they neither decided nor drained this state.) + probe.record_bytes(64 * 1024 * 1024); + let held: Vec<_> = (0..pool.target()) + .filter_map(|_| pool.try_acquire_owned()) + .collect(); + let _ = + tokio::time::timeout(std::time::Duration::from_millis(5), pool.acquire_owned()).await; + drop(held); + clock.advance(WINDOW); + + // Let the SPAWNED sampler run while the cycle is still blocked: under + // `start_paused`, sleeping parks the test so tokio auto-advances virtual + // time, firing the sampler's SAMPLE_INTERVAL ticks; the first tick past the + // injected window boundary is a bootstrap Grow on the shared pool. + tokio::time::sleep(WINDOW).await; + + // The resize happened DURING the in-flight cycle: the executor is still + // blocked (we have not sent `release_tx` yet), so this is unambiguously a + // mid-cycle resize, which the pre-F1 select-arm tick could never do. + assert!( + pool.target() > 4, + "the adaptive sampler must resize the pool DURING an in-flight cycle \ + (F1); pool stayed at {}", + pool.target() + ); + + // Release the cycle and shut down cleanly; `run` joins the sampler on exit. + let _ = release_tx.send(()); + orch.shutdown(); + let _ = tokio::time::timeout(std::time::Duration::from_secs(30), handle).await; + } } diff --git a/crates/driven-core/src/pacer.rs b/crates/driven-core/src/pacer.rs index 651dcd43..56ffe6af 100644 --- a/crates/driven-core/src/pacer.rs +++ b/crates/driven-core/src/pacer.rs @@ -145,6 +145,17 @@ pub trait Pacer: Send + Sync { /// The default is a no-op so simple test/fake pacers need not implement it; /// [`AimdPacer`] overrides it. fn set_bandwidth_cap(&self, _mbps: Option) {} + + /// Wall-clock ms of the most recent throttle response (rate-limit or daily + /// quota), or `i64::MIN` if the pacer has never throttled. The adaptive + /// upload-parallelism controller (DESIGN s11.4.7) reads this to answer "did + /// the pacer throttle at any point during the last throughput window?" - a + /// throttle explains a throughput drop, so a shrink must be suppressed even + /// if the backoff has since cleared. The default `i64::MIN` (never) makes a + /// simple/fake pacer read as "not throttling"; [`AimdPacer`] overrides it. + fn last_throttle_ms(&self) -> i64 { + i64::MIN + } } /// `serde` helper: (de)serialise a [`Duration`] as integer milliseconds so @@ -401,6 +412,14 @@ pub struct AimdPacer { /// current clean window (DESIGN s18.1). The additive increase fires /// once `CLEAN_WINDOW_MS` of clean time has accrued since this point. clean_window_start_ms: AtomicI64, + /// Wall-clock ms of the most recent THROTTLE response (rate-limit or daily + /// quota), or `i64::MIN` if the pacer has never throttled. Distinct from + /// `clean_window_start_ms` (which is ALSO seeded to `now` at construction, so + /// it cannot answer "has a throttle happened since time T"): this is set only + /// by an actual throttle. Read by the adaptive-parallelism controller + /// (DESIGN s11.4.7) via [`Pacer::last_throttle_ms`] to scope its "not + /// throttling" gate to the throughput window. + last_throttle_ms: AtomicI64, /// Bit-packed current ceilings, guarded for atomic snapshot/update. ceilings: Mutex, clock: Arc, @@ -468,6 +487,8 @@ impl AimdPacer { }), backoff_until_ms: AtomicI64::new(now), clean_window_start_ms: AtomicI64::new(now), + // Never-throttled sentinel: a genuine throttle overwrites it. + last_throttle_ms: AtomicI64::new(i64::MIN), ceilings: Mutex::new(ceilings), clock, } @@ -660,10 +681,14 @@ impl Pacer for AimdPacer { match classification { ResponseClass::Ok => self.maybe_raise(now), ResponseClass::RateLimited { retry_after } => { + self.last_throttle_ms.store(now, Ordering::Release); self.halve(now); self.set_backoff(now, retry_after); } - ResponseClass::DailyQuota => self.pause_until_midnight_pacific(now), + ResponseClass::DailyQuota => { + self.last_throttle_ms.store(now, Ordering::Release); + self.pause_until_midnight_pacific(now); + } ResponseClass::OtherError => { // Non-throttle: does not move the AIMD ceiling and does not // count as a clean window tick (DESIGN s18.1 / SPEC s9). @@ -674,6 +699,10 @@ impl Pacer for AimdPacer { fn ceilings(&self) -> PacerCeilings { *lock_recover(&self.ceilings) } + + fn last_throttle_ms(&self) -> i64 { + self.last_throttle_ms.load(Ordering::Acquire) + } } /// Applies a backoff with jitter to Drive's `Retry-After` (DESIGN s5.4: diff --git a/crates/driven-core/tests/e2e_fake.rs b/crates/driven-core/tests/e2e_fake.rs index 63b5a7b6..4cc66509 100644 --- a/crates/driven-core/tests/e2e_fake.rs +++ b/crates/driven-core/tests/e2e_fake.rs @@ -2191,28 +2191,19 @@ async fn pipeline_streaming_keeps_memory_bounded() { ); } -/// Adaptive-parallelism: the upload pool must shrink under induced latency and -/// recover when it clears (DESIGN s11.4.2 AIMD). Run it with: -/// -/// ```text -/// cargo test -p driven-core --test e2e_fake -- --ignored adaptive_parallelism_reacts_to_latency -/// ``` -/// -/// The body is real: it runs a genuine multi-file plan end to end and asserts -/// every op completed correctly + prints the measured throughput. The pool's -/// reaction to *induced latency* is NOT asserted here: it is driven by the -/// `ThroughputProbe` loop in the app-shell `Orchestrator::run` select loop -/// (not wired into core) over a latency-shaping remote, neither of which this -/// harness provides. The AIMD step logic itself is unit-tested in pacer.rs. -/// Tracking: #28. +/// Adaptive-parallelism wiring (DESIGN s11.4.7): a REAL executor running a REAL +/// multi-file plan over the fake remote must feed the injected `ThroughputProbe` +/// the total uploaded byte count, and the injected `UploadPool` (the SAME `Arc` +/// the app-shell also hands the [`AdaptiveController`]) must be the gate those +/// uploads pass through. This is the integration seam the pure control-loop unit +/// tests in `driven_core::adaptive` cannot reach; the shrink/recover REACTION to +/// a throughput change is exhaustively + deterministically covered there +/// (`controller_shrinks_on_latency_then_recovers`) and does not need real time / +/// a latency-shaping remote to assert. Tracking: #28. #[tokio::test] -#[ignore = "perf benchmark: adaptive-parallelism pool reaction to induced \ - latency; run with `cargo test -p driven-core --test e2e_fake -- \ - --ignored adaptive_parallelism_reacts_to_latency`. The pool \ - reaction needs the app-shell ThroughputProbe loop + a \ - latency-shaping remote (not in core); AIMD steps are unit-tested \ - in pacer.rs. Tracking #28."] -async fn adaptive_parallelism_reacts_to_latency() { +async fn adaptive_parallelism_executor_feeds_probe() { + use driven_core::adaptive::{ThroughputProbe, UploadPool}; + let dir = tempfile::tempdir().unwrap(); let src_dir = tempfile::tempdir().unwrap(); let state = open_state(dir.path()).await; @@ -2243,6 +2234,11 @@ async fn adaptive_parallelism_reacts_to_latency() { collisions: vec![], }; + // The SAME pool + probe the app-shell shares into the executor AND the + // adaptive controller. + let pool = UploadPool::new(4); + let probe = ThroughputProbe::new(); + let clock = Arc::new(FakeClock::new()); let exec = DefaultExecutor::with_clock( ExecutorDeps { @@ -2254,7 +2250,9 @@ async fn adaptive_parallelism_reacts_to_latency() { network: None, }, clock, - ); + ) + .with_upload_pool(pool.clone()) + .with_throughput_probe(probe.clone()); let out = exec .execute(&src, &plan, &noop_progress, &noop_outcome) @@ -2266,9 +2264,216 @@ async fn adaptive_parallelism_reacts_to_latency() { n_files as usize, "every file landed under the parallel pool" ); - eprintln!( - "adaptive-parallelism: {n_files} files synced; pool-shrink-under-latency \ - needs the app-shell ThroughputProbe loop (see #28)" + + // Wiring proof: the executor fed the injected probe the full uploaded byte + // total (this is what the adaptive controller drains each window). Uploads + // may be plaintext or ciphertext; here crypto is None so bytes == plaintext. + let expected_bytes = u64::from(n_files) * file_bytes as u64; + assert_eq!( + probe.peek_bytes(), + expected_bytes, + "executor must record every uploaded byte into the throughput probe" + ); + // The injected pool's size is untouched by the executor (only the controller + // resizes it), so the app-shell's controller-shared handle is intact. + assert_eq!( + pool.target(), + 4, + "the executor must not resize the pool itself" + ); +} + +/// Adaptive-parallelism TRANSITION through the real wiring (DESIGN s11.4.7, the +/// ROADMAP M3 adaptive-parallelism acceptance row): a REAL [`DefaultExecutor`] +/// uploading REAL plans over the fake remote feeds the SHARED `ThroughputProbe`, +/// and the SHARED `AdaptiveController` - the exact `Arc`s the app-shell wires - +/// resizes the SHARED `UploadPool` in BOTH directions in response: +/// +/// - a degraded-throughput window (far fewer uploaded bytes than the prior one) +/// is observed to SHRINK the pool, and +/// - a restored high-throughput window is observed to GROW it back. +/// +/// Throughput is the genuine DESIGN signal: the bytes come from real executor +/// uploads and the window duration from the injected [`FakeClock`], so bps is +/// deterministic with no latency-shaping remote and no real time. The only +/// element staged for determinism is the per-window "pool was the bottleneck" +/// contention mark (natural contention depends on tokio scheduling; a 1-file +/// window may never contend) - forced via the pool's public API, exactly as the +/// pure-controller unit tests do. Disk is never saturated and the pacer never +/// throttles, so throughput alone moves the pool. Tracking: #28. +#[tokio::test] +async fn adaptive_parallelism_transitions_pool_through_real_wiring() { + use driven_core::adaptive::{AdaptiveController, ThroughputProbe, UploadPool, WINDOW}; + use driven_test_fixtures::diskstat::FakeDiskBusyProbe; + + let dir = tempfile::tempdir().unwrap(); + let src_dir = tempfile::tempdir().unwrap(); + let state = open_state(dir.path()).await; + let account = seed_account(&state).await; + let remote = Arc::new(InMemoryRemoteStore::new()); + let folder = remote.root_id().to_string(); + let src = source_in(account, src_dir.path(), &folder); + state.upsert_source(&src).await.unwrap(); + + // The SAME pool + probe shared into the executor AND the controller. + let pool = UploadPool::new(4); + let probe = ThroughputProbe::new(); + let clock = Arc::new(FakeClock::new()); + let exec = DefaultExecutor::with_clock( + ExecutorDeps { + remote: remote.clone(), + state: state.clone(), + pacer: test_pacer(clock.clone()), + crypto: None, + vss: None, + network: None, + }, + clock.clone(), + ) + .with_upload_pool(pool.clone()) + .with_throughput_probe(probe.clone()); + + // The controller over the SAME pool + probe (as the app-shell wires it), with + // a never-saturated disk and a never-throttling pacer so only throughput acts. + let disk = Arc::new(FakeDiskBusyProbe::not_saturated()); + let ctrl = AdaptiveController::new( + pool.clone(), + probe.clone(), + disk, + test_pacer(clock.clone()), + clock.clone(), + ); + + // A BIG window uploads ~1 MiB across 16 files; a TINY window uploads 1 KiB in + // one file (so its throughput is far below 50% of a big window's). + const BIG_N: u32 = 16; + const BIG_SZ: usize = 64 * 1024; + const TINY_N: u32 = 1; + const TINY_SZ: usize = 1024; + + // Two baseline windows: W1 bootstrap-grows, W2 settles - establishing a high + // previous-window throughput and a stable "before" size. + upload_batch(&exec, &src, src_dir.path(), "w1", BIG_N, BIG_SZ).await; + mark_pool_saturated(&pool).await; + clock.advance(WINDOW); + ctrl.tick().await; + + upload_batch(&exec, &src, src_dir.path(), "w2", BIG_N, BIG_SZ).await; + mark_pool_saturated(&pool).await; + clock.advance(WINDOW); + ctrl.tick().await; + let before_shrink = pool.target(); + + // W3 degraded: a near-empty window collapses throughput below 50% of the + // baseline -> the controller SHRINKS the shared pool. + upload_batch(&exec, &src, src_dir.path(), "w3", TINY_N, TINY_SZ).await; + mark_pool_saturated(&pool).await; + clock.advance(WINDOW); + ctrl.tick().await; + let after_shrink = pool.target(); + assert!( + after_shrink < before_shrink, + "a degraded-throughput window must shrink the pool through the real \ + wiring: {before_shrink} -> {after_shrink}" + ); + + // W4 restored: throughput jumps back up (improving) -> the controller GROWS + // the shared pool again. + upload_batch(&exec, &src, src_dir.path(), "w4", BIG_N, BIG_SZ).await; + mark_pool_saturated(&pool).await; + clock.advance(WINDOW); + ctrl.tick().await; + let after_grow = pool.target(); + assert!( + after_grow > after_shrink, + "a restored high-throughput window must regrow the pool through the real \ + wiring: {after_shrink} -> {after_grow}" + ); + + // The uploads really happened over the fake remote (real wiring, not a + // simulated probe feed): every distinct filename across the four windows + // landed as an object. + assert_eq!( + live_object_count(&remote, &folder).await, + (BIG_N + BIG_N + TINY_N + BIG_N) as usize, + "every uploaded file across all windows landed remotely" + ); +} + +/// Adaptive-parallelism F2: a RESUMABLE upload (above `RESUMABLE_THRESHOLD`, so +/// it streams as several wire chunks) must credit the throughput probe its bytes +/// EXACTLY ONCE - fed per wire chunk in `push_one_wire_chunk`, with the op's +/// completion site subtracting what was already streamed so nothing is +/// double-counted. This guards the F2 double-count regression and proves the +/// resumable path feeds the probe end-to-end through the real executor. (The +/// per-chunk IN-WINDOW attribution is structural - the record site is the +/// chunk-ack point - and the window math is covered by the `adaptive.rs` unit +/// tests.) +#[tokio::test] +async fn adaptive_parallelism_resumable_upload_credits_probe_once() { + use driven_core::adaptive::{ThroughputProbe, UploadPool}; + + let dir = tempfile::tempdir().unwrap(); + let src_dir = tempfile::tempdir().unwrap(); + let state = open_state(dir.path()).await; + let account = seed_account(&state).await; + let remote = Arc::new(InMemoryRemoteStore::new()); + let folder = remote.root_id().to_string(); + let src = source_in(account, src_dir.path(), &folder); + state.upsert_source(&src).await.unwrap(); + + // One file well above the 5 MiB resumable threshold, so the executor streams + // it as multiple wire chunks through the resumable session protocol (the F2 + // per-chunk recording path). crypto is None, so wire bytes == plaintext size. + let file_bytes = 6 * 1024 * 1024usize; + let name = "big.bin".to_string(); + let contents: Vec = (0..file_bytes).map(|j| (j % 251) as u8).collect(); + write_file(src_dir.path(), &name, &contents); + let plan = Plan { + ops: vec![Op::HashThenUpload { + source_id: src.id, + relative_path: RelativePath::try_from(name).unwrap(), + size: file_bytes as u64, + }], + collisions: vec![], + }; + + let pool = UploadPool::new(4); + let probe = ThroughputProbe::new(); + let clock = Arc::new(FakeClock::new()); + let exec = DefaultExecutor::with_clock( + ExecutorDeps { + remote: remote.clone(), + state: state.clone(), + pacer: test_pacer(clock.clone()), + crypto: None, + vss: None, + network: None, + }, + clock, + ) + .with_upload_pool(pool.clone()) + .with_throughput_probe(probe.clone()); + + let out = exec + .execute(&src, &plan, &noop_progress, &noop_outcome) + .await + .unwrap(); + assert!(out.iter().all(|o| matches!(o, OpOutcome::Done { .. }))); + assert_eq!( + live_object_count(&remote, &folder).await, + 1, + "the resumable upload finalized exactly one object" + ); + + // Exactly once: a double-count (completion failing to subtract the streamed + // bytes) would read 2x here; a missing per-chunk feed would still read the + // full size via the completion catch-all, so the discriminating failure this + // guards is the double-count. + assert_eq!( + probe.peek_bytes(), + file_bytes as u64, + "a resumable upload must credit the throughput probe its wire bytes exactly once" ); } @@ -2299,3 +2504,52 @@ fn noop_progress(_p: driven_core::types::ExecProgress) {} fn noop_outcome(_o: &OpOutcome) -> futures::future::BoxFuture<'static, ()> { Box::pin(async {}) } + +/// Upload `n` fresh `sz`-byte files (names prefixed with `prefix`, so every call +/// is new work the executor really uploads rather than a no-op skip) through a +/// REAL executor over the fake remote. Used by the adaptive-parallelism +/// transition test to feed the shared `ThroughputProbe` a controllable byte +/// volume per window. +async fn upload_batch( + exec: &DefaultExecutor, + src: &SourceRow, + src_root: &std::path::Path, + prefix: &str, + n: u32, + sz: usize, +) { + let mut ops = Vec::new(); + for i in 0..n { + let name = format!("{prefix}_{i}.bin"); + let contents: Vec = (0..sz).map(|j| ((i as usize + j) % 251) as u8).collect(); + write_file(src_root, &name, &contents); + ops.push(Op::HashThenUpload { + source_id: src.id, + relative_path: RelativePath::try_from(name).unwrap(), + size: sz as u64, + }); + } + let plan = Plan { + ops, + collisions: vec![], + }; + exec.execute(src, &plan, &noop_progress, &noop_outcome) + .await + .unwrap(); +} + +/// Force the adaptive pool's "the pool was the bottleneck this window" +/// (contention) signal deterministically: pin every current permit, then contend +/// once - `UploadPool::acquire_owned` bumps the contention counter BEFORE it +/// awaits, so the count rises even though this acquire times out - then release +/// the pinned permits, leaving the pool size unchanged. Natural contention +/// depends on tokio scheduling timing (a tiny window may never contend), so the +/// transition test stages just this representative-window mark while the actual +/// throughput signal flows through the real executor. +async fn mark_pool_saturated(pool: &driven_core::adaptive::UploadPool) { + let held: Vec<_> = (0..pool.target()) + .filter_map(|_| pool.try_acquire_owned()) + .collect(); + let _ = tokio::time::timeout(std::time::Duration::from_millis(5), pool.acquire_owned()).await; + drop(held); +} diff --git a/crates/driven-diskstat/Cargo.toml b/crates/driven-diskstat/Cargo.toml new file mode 100644 index 00000000..20e3588d --- /dev/null +++ b/crates/driven-diskstat/Cargo.toml @@ -0,0 +1,39 @@ +[package] +name = "driven-diskstat" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +authors.workspace = true +license.workspace = true +repository.workspace = true +homepage.workspace = true + +# Per-OS disk-busy ("are we bottlenecked by the disk?") reader for the adaptive +# upload-parallelism controller (DESIGN s11.4.7 / s18.2). Mirrors the +# `driven-power` crate's shape exactly: a small trait + a pure classifier that is +# unit-tested on every target, plus exactly ONE cfg-gated per-OS `Real*` backend +# compiled per target. Kept a standalone crate (not folded into driven-core) so +# the Windows PDH / macOS IOKit FFI dependencies stay isolated to their target +# and `cargo build --workspace` on a foreign host never pulls them. +[dependencies] + +# Windows: PDH (Performance Data Helper) counters - the `\PhysicalDisk(_Total)\ +# % Disk Time` rate counter (DESIGN s18.2). PDH computes the rate over the +# interval between two `PdhCollectQueryData` calls internally, so the reader +# holds one query + counter handle for the process lifetime. +[target.'cfg(windows)'.dependencies] +windows = { version = "0.62", features = [ + "Win32_Foundation", + "Win32_System_Performance", +] } + +# macOS: IOKit `IOBlockStorageDriver` `Statistics` dict, read via CoreFoundation +# types (DESIGN s18.2). Mirrors the FFI style already validated in +# `driven-power/src/macos.rs` (hand-declared `#[link(name = "IOKit")]` externs + +# the `core-foundation` typed wrappers). BEST-EFFORT: the Statistics dict exposes +# per-op latency / total-time counters, NOT a clean device-busy percentage, so +# the reader approximates "busy fraction" from the Total-Time delta and returns +# `Unknown` (fail-open) on any uncertainty. +[target.'cfg(target_os = "macos")'.dependencies] +core-foundation = "0.10" +core-foundation-sys = "0.8" diff --git a/crates/driven-diskstat/src/lib.rs b/crates/driven-diskstat/src/lib.rs new file mode 100644 index 00000000..b23f0d86 --- /dev/null +++ b/crates/driven-diskstat/src/lib.rs @@ -0,0 +1,190 @@ +//! `driven-diskstat` - per-OS "is the local disk saturated?" reader for the +//! adaptive upload-parallelism controller (DESIGN s11.4.7, s18.2). +//! +//! # What it answers +//! +//! One question, sampled periodically: *are we bottlenecked by the disk right +//! now?* If yes, adding more in-flight uploads cannot raise throughput (the +//! reads that feed them are already disk-bound) and would only hurt, so the +//! controller must not grow the pool. DESIGN s18.2 fixes the signal per-OS: +//! +//! - **Linux:** `/proc/diskstats` field 10 ("time spent doing I/Os", ms) delta +//! for the device backing the source root; `busy_ms / interval_ms > 0.80` = +//! saturated. +//! - **macOS:** IOKit `IOBlockStorageDriver` `Statistics` dict, same ratio +//! (best-effort - see [`RealDiskBusyProbe`] on macOS). +//! - **Windows:** PDH `\PhysicalDisk(_Total)\% Disk Time > 80 %` = saturated. +//! +//! # Shape (mirrors `driven-power`) +//! +//! A tiny [`DiskBusyProbe`] trait, a PURE classifier ([`DiskBusy::is_saturated`] + +//! [`busy_fraction_from_delta`]) unit-tested on every target, and exactly ONE +//! cfg-gated per-OS [`RealDiskBusyProbe`] compiled per target (re-exported +//! cfg-free so the caller wires `RealDiskBusyProbe::new(..)` with no `cfg` at the +//! call site). Tests use `FakeDiskBusyProbe` from `driven-test-fixtures`. +//! +//! # Fail-open (load-bearing, DESIGN s11.4.7) +//! +//! A disk reader that cannot produce a reading - no baseline yet, a parse/FFI +//! error, an unsupported platform - returns [`DiskBusy::Unknown`], and +//! [`DiskBusy::is_saturated`] maps `Unknown` to `false` ("not saturated"). A +//! broken or unavailable reader must NEVER strangle uploads: the worst it can do +//! is let the controller grow when it maybe should not have, which the +//! throughput signal then corrects on the next window. It must never do the +//! reverse (falsely report saturation and pin the pool small). + +/// The busy-fraction above which the disk is considered "saturated" (DESIGN +/// s18.2: `> 80 %`). Exceeding this means reducing concurrency will not help and +/// raising it will hurt, so the controller must hold or shrink, never grow. +pub const SATURATION_THRESHOLD: f64 = 0.80; + +/// A disk-busy reading (DESIGN s18.2). +#[derive(Clone, Copy, Debug, PartialEq)] +pub enum DiskBusy { + /// The measured busy fraction over the last sampling interval. Nominally in + /// `0.0..=1.0`, but CAN exceed `1.0` on a device serving overlapping I/O + /// (per-op "busy time" summed across a queue can exceed wall-clock); callers + /// only ever compare it against [`SATURATION_THRESHOLD`], so an over-unity + /// value simply reads as saturated, which is correct. + Fraction(f64), + /// The reader could not produce a reading (no baseline sample yet, a + /// parse/FFI error, or an unsupported platform). Treated as NOT saturated + /// (fail-open) - see the crate-level docs. + Unknown, +} + +impl DiskBusy { + /// `true` iff the disk is saturated (DESIGN s18.2). [`DiskBusy::Unknown`] is + /// NOT saturated (fail-open) - a broken reader must never pin the pool small. + #[must_use] + pub fn is_saturated(self) -> bool { + match self { + DiskBusy::Fraction(f) => f > SATURATION_THRESHOLD, + DiskBusy::Unknown => false, + } + } +} + +/// Compute a busy fraction from a raw "busy time" delta and the wall-clock +/// interval it accrued over, both in the SAME time unit (ms vs ms, ns vs ns). +/// +/// Pure and unit-tested directly. A zero (or absent) interval yields +/// [`DiskBusy::Unknown`] rather than dividing by zero - the caller has no +/// baseline to diff against yet. +#[must_use] +pub fn busy_fraction_from_delta(busy_delta: u64, interval: u64) -> DiskBusy { + if interval == 0 { + return DiskBusy::Unknown; + } + DiskBusy::Fraction(busy_delta as f64 / interval as f64) +} + +/// A periodically-sampled disk-busy source (DESIGN s18.2). +/// +/// [`sample`](DiskBusyProbe::sample) is stateful: a delta-based backend +/// (Linux, macOS) stores the previous raw counters + timestamp internally and +/// returns the busy fraction accrued since the last call, so the FIRST call +/// after construction returns [`DiskBusy::Unknown`] (no baseline). It is called +/// on the controller's sampling cadence (DESIGN s18.2: every 5 s). +pub trait DiskBusyProbe: Send + Sync { + /// Sample the disk-busy fraction accrued since the previous call. Cheap + /// (a small file read on Linux, one PDH collect on Windows) and never + /// blocks on I/O for a meaningful duration. Any failure returns + /// [`DiskBusy::Unknown`] (fail-open). + fn sample(&self) -> DiskBusy; +} + +// Exactly one per-OS backend is compiled per target; each exports a +// `RealDiskBusyProbe` re-exported cfg-free below (mirrors driven-power). +#[cfg(target_os = "linux")] +mod linux; +#[cfg(target_os = "macos")] +mod macos; +#[cfg(target_os = "windows")] +mod windows; + +#[cfg(target_os = "linux")] +pub use linux::RealDiskBusyProbe; +#[cfg(target_os = "macos")] +pub use macos::RealDiskBusyProbe; +#[cfg(target_os = "windows")] +pub use windows::RealDiskBusyProbe; + +/// Fallback [`RealDiskBusyProbe`] for any target without a per-OS backend +/// (e.g. the BSDs). Always reports [`DiskBusy::Unknown`] so the adaptive +/// controller runs with the disk gate open (fail-open) rather than failing to +/// build. The three tier-1 desktop targets (Linux/macOS/Windows) all have a +/// real backend above. +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] +mod unsupported { + use super::{DiskBusy, DiskBusyProbe}; + + /// See the module docs: an always-`Unknown` probe for unsupported targets. + #[derive(Debug, Default)] + pub struct RealDiskBusyProbe; + + impl RealDiskBusyProbe { + /// Construct the no-op probe. The `root` is accepted for signature + /// parity with the real backends and ignored. + #[must_use] + pub fn new(_root: std::path::PathBuf) -> Self { + Self + } + } + + impl DiskBusyProbe for RealDiskBusyProbe { + fn sample(&self) -> DiskBusy { + DiskBusy::Unknown + } + } +} + +#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] +pub use unsupported::RealDiskBusyProbe; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unknown_is_not_saturated_fail_open() { + // The load-bearing invariant: an unreadable disk never pins the pool. + assert!(!DiskBusy::Unknown.is_saturated()); + } + + #[test] + fn threshold_is_strict_greater_than_80_percent() { + assert!(!DiskBusy::Fraction(0.0).is_saturated()); + assert!( + !DiskBusy::Fraction(0.80).is_saturated(), + "exactly 80% is NOT saturated (strict >)" + ); + assert!(DiskBusy::Fraction(0.8001).is_saturated()); + assert!(DiskBusy::Fraction(1.0).is_saturated()); + // Over-unity (overlapping I/O) reads as saturated, which is correct. + assert!(DiskBusy::Fraction(3.5).is_saturated()); + } + + #[test] + fn busy_fraction_zero_interval_is_unknown() { + assert_eq!(busy_fraction_from_delta(100, 0), DiskBusy::Unknown); + } + + #[test] + fn busy_fraction_ratio() { + assert_eq!(busy_fraction_from_delta(0, 5_000), DiskBusy::Fraction(0.0)); + assert_eq!( + busy_fraction_from_delta(2_500, 5_000), + DiskBusy::Fraction(0.5) + ); + assert_eq!( + busy_fraction_from_delta(5_000, 5_000), + DiskBusy::Fraction(1.0) + ); + // Delta exceeding the interval (overlapping I/O) -> over-unity fraction. + assert_eq!( + busy_fraction_from_delta(9_000, 5_000), + DiskBusy::Fraction(1.8) + ); + } +} diff --git a/crates/driven-diskstat/src/linux.rs b/crates/driven-diskstat/src/linux.rs new file mode 100644 index 00000000..5947eda4 --- /dev/null +++ b/crates/driven-diskstat/src/linux.rs @@ -0,0 +1,174 @@ +//! Linux [`DiskBusyProbe`] backend (DESIGN s18.2): `/proc/diskstats` field 10. +//! +//! `/proc/diskstats` exposes, per block device, the cumulative "time spent doing +//! I/Os" in milliseconds (the 13th whitespace token, a.k.a. field 10 of the +//! post-name stats - the same counter `iostat`'s `%util` is derived from). The +//! busy FRACTION over an interval is `busy_ms_delta / interval_ms`; DESIGN s18.2 +//! flags `> 0.80` as saturated. +//! +//! The device is the one BACKING THE SOURCE ROOT: we `stat(2)` the root once at +//! construction, decompose its `st_dev` into `(major, minor)`, and match that +//! against the `/proc/diskstats` rows (partitions carry their own major:minor +//! and their own busy counter). A root whose device does not appear in +//! `/proc/diskstats` (device-mapper, overlay, a network mount) yields +//! [`DiskBusy::Unknown`] - fail-open, never a false "saturated". + +use std::os::unix::fs::MetadataExt; +use std::path::PathBuf; +use std::sync::Mutex; +use std::time::Instant; + +use crate::{DiskBusy, DiskBusyProbe}; + +/// A prior sample: the cumulative busy-ms counter and when it was read, so the +/// next sample can diff both against it. +#[derive(Clone, Copy)] +struct Baseline { + busy_ms: u64, + at: Instant, +} + +/// Linux disk-busy reader over `/proc/diskstats` (DESIGN s18.2). +pub struct RealDiskBusyProbe { + /// The `(major, minor)` of the device backing the source root, resolved once + /// via `stat(2)`. `None` when the root could not be `stat`ed (then every + /// sample is `Unknown`, fail-open). + device: Option<(u64, u64)>, + /// The previous `(busy_ms, Instant)`; `None` until the first sample sets the + /// baseline. + baseline: Mutex>, +} + +impl RealDiskBusyProbe { + /// Build a reader for the device backing `root`. Resolution never fails hard: + /// an un-`stat`able root just makes every [`sample`](DiskBusyProbe::sample) + /// return [`DiskBusy::Unknown`]. + #[must_use] + pub fn new(root: PathBuf) -> Self { + let device = std::fs::metadata(&root) + .ok() + .map(|m| m.dev()) + .map(|dev| (gnu_dev_major(dev), gnu_dev_minor(dev))); + Self { + device, + baseline: Mutex::new(None), + } + } +} + +impl DiskBusyProbe for RealDiskBusyProbe { + fn sample(&self) -> DiskBusy { + let Some((major, minor)) = self.device else { + return DiskBusy::Unknown; + }; + let Ok(contents) = std::fs::read_to_string("/proc/diskstats") else { + return DiskBusy::Unknown; + }; + let Some(busy_ms) = parse_busy_ms(&contents, major, minor) else { + return DiskBusy::Unknown; + }; + + let now = Instant::now(); + let mut guard = self.baseline.lock().unwrap_or_else(|e| e.into_inner()); + let prev = *guard; + *guard = Some(Baseline { busy_ms, at: now }); + drop(guard); + + match prev { + // First sample: establish the baseline, no fraction yet. + None => DiskBusy::Unknown, + Some(prev) => { + let interval_ms = now.duration_since(prev.at).as_millis(); + let interval_ms = u64::try_from(interval_ms).unwrap_or(u64::MAX); + // `saturating_sub`: the counter is monotonic, but a device + // hot-swap / counter reset must never produce a huge bogus delta. + let delta = busy_ms.saturating_sub(prev.busy_ms); + crate::busy_fraction_from_delta(delta, interval_ms) + } + } + } +} + +/// Extract the cumulative "time spent doing I/Os" (ms) for `(major, minor)` from +/// `/proc/diskstats` contents. Pure over the file text so it is unit-testable +/// without a live `/proc`. Returns `None` when no row matches. +/// +/// Row layout (`Documentation/admin-guide/iostats.rst`): `major minor name` +/// followed by the numeric fields; "time spent doing I/Os (ms)" is the 13th +/// whitespace token (index 12). +#[must_use] +fn parse_busy_ms(contents: &str, major: u64, minor: u64) -> Option { + for line in contents.lines() { + let mut it = line.split_whitespace(); + let row_major: u64 = it.next()?.parse().ok()?; + let row_minor: u64 = it.next()?.parse().ok()?; + if row_major != major || row_minor != minor { + continue; + } + // Skip the device name, then advance to token index 12 overall. We have + // already consumed indices 0 (major) and 1 (minor); the name is index 2, + // so the 10 tokens after the name land us on index 12 (busy-ms). + let mut it = it.skip(1); // device name (index 2) + // indices 3..=11 (nine fields) then index 12 is next. + for _ in 0..9 { + it.next()?; + } + return it.next()?.parse().ok(); + } + None +} + +/// `major(3)` per the glibc `gnu_dev_major` encoding of a Linux `dev_t`. +#[must_use] +fn gnu_dev_major(dev: u64) -> u64 { + ((dev >> 8) & 0xfff) | ((dev >> 32) & !0xfff) +} + +/// `minor(3)` per the glibc `gnu_dev_minor` encoding of a Linux `dev_t`. +#[must_use] +fn gnu_dev_minor(dev: u64) -> u64 { + (dev & 0xff) | ((dev >> 12) & !0xff) +} + +#[cfg(test)] +mod tests { + use super::*; + + // A representative /proc/diskstats excerpt (sda + its partition sda1, plus + // an unrelated nvme device). "Time spent doing I/Os (ms)" is field 10 of the + // post-name stats == token index 12; each device carries a distinctive value + // there (sda = 1500, sda1 = 123456, nvme0n1 = 90) so a miscount is obvious. + const DISKSTATS: &str = "\ + 8 0 sda 1000 20 40000 800 500 10 20000 400 0 1500 1200 0 0 0 0 + 8 1 sda1 900 10 39000 700 400 5 19000 300 0 123456 1400 0 0 0 0 + 259 0 nvme0n1 50 0 2000 30 60 0 3000 40 0 90 250 0 0 0 0 +"; + + #[test] + fn parses_busy_ms_for_matching_device() { + assert_eq!(parse_busy_ms(DISKSTATS, 8, 1), Some(123_456)); + assert_eq!(parse_busy_ms(DISKSTATS, 8, 0), Some(1_500)); + assert_eq!(parse_busy_ms(DISKSTATS, 259, 0), Some(90)); + } + + #[test] + fn unmatched_device_is_none() { + assert_eq!(parse_busy_ms(DISKSTATS, 253, 0), None); + } + + #[test] + fn malformed_row_is_skipped_not_panicked() { + // A truncated row must not panic; a later well-formed row still matches. + let text = "8 0 sda 1 2 3\n8 1 sda1 900 10 39000 700 400 5 19000 300 0 777 1400 0\n"; + assert_eq!(parse_busy_ms(text, 8, 0), None); + assert_eq!(parse_busy_ms(text, 8, 1), Some(777)); + } + + #[test] + fn dev_decompose_roundtrip() { + // 8:1 encodes as the classic non-huge dev_t 0x0801. + let dev: u64 = (8 << 8) | 1; + assert_eq!(gnu_dev_major(dev), 8); + assert_eq!(gnu_dev_minor(dev), 1); + } +} diff --git a/crates/driven-diskstat/src/macos.rs b/crates/driven-diskstat/src/macos.rs new file mode 100644 index 00000000..55c7177e --- /dev/null +++ b/crates/driven-diskstat/src/macos.rs @@ -0,0 +1,203 @@ +//! macOS [`DiskBusyProbe`] backend (DESIGN s18.2): IOKit `IOBlockStorageDriver` +//! `Statistics`. +//! +//! # Best-effort caveat (READ THIS) +//! +//! Unlike Linux (`/proc/diskstats` "busy ms") and Windows (PDH `% Disk Time`), +//! macOS exposes no first-class device-busy percentage. The closest signal is +//! `IOBlockStorageDriver`'s `Statistics` dict, whose `Total Time (Read)` / +//! `Total Time (Write)` counters are cumulative NANOSECONDS spent servicing I/O. +//! We sum those across every block-storage driver and treat +//! `total_time_ns_delta / interval_ns` as the busy fraction. Because I/O overlaps +//! (a delta can exceed wall-clock), this fraction can be over-unity; that reads +//! as "saturated", consistent with the other backends. +//! +//! This approximation can OVER-report busy. That direction is SAFE here: the +//! adaptive controller (DESIGN s11.4.7) requires `disk NOT saturated` for BOTH a +//! grow AND a shrink, so a falsely-saturated reading merely holds the pool at its +//! configured start size - i.e. it degrades to today's fixed-pool behaviour, it +//! never strangles the pool below the default. Any uncertainty (no matching +//! service, a missing key, an FFI error) returns [`DiskBusy::Unknown`] +//! (fail-open). This module is compiled but NOT executed on the CI hosts (no +//! Mac); it is validated by `cargo check --target aarch64-apple-darwin`. + +use std::ffi::c_void; +use std::os::raw::c_char; +use std::path::PathBuf; +use std::sync::Mutex; +use std::time::Instant; + +use core_foundation::base::TCFType; +use core_foundation::string::CFString; +use core_foundation_sys::base::CFTypeRef; +use core_foundation_sys::dictionary::{CFDictionaryGetValue, CFDictionaryRef}; +use core_foundation_sys::number::{kCFNumberSInt64Type, CFNumberGetValue, CFNumberRef}; +use core_foundation_sys::string::CFStringRef; + +use crate::{DiskBusy, DiskBusyProbe}; + +// IOKit / Mach scalar types (mach_port.h / IOKitLib.h). All `mach_port_t` +// derivatives are `u32`; `kern_return_t` is `i32`. +#[allow(non_camel_case_types)] +type mach_port_t = u32; +#[allow(non_camel_case_types)] +type io_object_t = mach_port_t; +#[allow(non_camel_case_types)] +type io_iterator_t = mach_port_t; +#[allow(non_camel_case_types)] +type io_registry_entry_t = mach_port_t; +#[allow(non_camel_case_types)] +type kern_return_t = i32; + +/// `kIOMainPortDefault` (a.k.a. the legacy `kIOMasterPortDefault`) is the null +/// mach port, value 0. +const K_IO_MAIN_PORT_DEFAULT: mach_port_t = 0; +/// `KERN_SUCCESS`. +const KERN_SUCCESS: kern_return_t = 0; + +// IOKit C API (IOKit.framework). Declared directly (matching the precedent in +// driven-power/src/macos.rs) rather than pulling a heavier IOKit binding crate. +#[link(name = "IOKit", kind = "framework")] +extern "C" { + fn IOServiceMatching(name: *const c_char) -> CFDictionaryRef; + fn IOServiceGetMatchingServices( + main_port: mach_port_t, + matching: CFDictionaryRef, + existing: *mut io_iterator_t, + ) -> kern_return_t; + fn IOIteratorNext(iterator: io_iterator_t) -> io_object_t; + fn IORegistryEntryCreateCFProperty( + entry: io_registry_entry_t, + key: CFStringRef, + allocator: *const c_void, + options: u32, + ) -> CFTypeRef; + fn IOObjectRelease(object: io_object_t) -> kern_return_t; +} + +/// A prior sample: cumulative total-I/O-time (ns) and when it was read. +#[derive(Clone, Copy)] +struct Baseline { + total_time_ns: u64, + at: Instant, +} + +/// macOS disk-busy reader over IOKit `IOBlockStorageDriver` `Statistics` +/// (DESIGN s18.2, best-effort - see the module docs). +pub struct RealDiskBusyProbe { + baseline: Mutex>, +} + +impl RealDiskBusyProbe { + /// Build the reader. The `root` is unused on macOS (we aggregate every block + /// driver rather than resolving the backing device, which IOKit does not + /// expose as cheaply as Linux `st_dev`) but accepted for signature parity. + #[must_use] + pub fn new(_root: PathBuf) -> Self { + Self { + baseline: Mutex::new(None), + } + } +} + +impl DiskBusyProbe for RealDiskBusyProbe { + fn sample(&self) -> DiskBusy { + let Some(total_time_ns) = read_total_io_time_ns() else { + return DiskBusy::Unknown; + }; + let now = Instant::now(); + let mut guard = self.baseline.lock().unwrap_or_else(|e| e.into_inner()); + let prev = *guard; + *guard = Some(Baseline { + total_time_ns, + at: now, + }); + drop(guard); + + match prev { + None => DiskBusy::Unknown, + Some(prev) => { + let interval_ns = now.duration_since(prev.at).as_nanos(); + let interval_ns = u64::try_from(interval_ns).unwrap_or(u64::MAX); + let delta = total_time_ns.saturating_sub(prev.total_time_ns); + crate::busy_fraction_from_delta(delta, interval_ns) + } + } + } +} + +/// Sum `Total Time (Read)` + `Total Time (Write)` (ns) across every +/// `IOBlockStorageDriver`. Returns `None` on any IOKit failure or if no driver +/// reported a usable counter (fail-open). +fn read_total_io_time_ns() -> Option { + // "IOBlockStorageDriver" as a NUL-terminated C string. + const CLASS: &[u8] = b"IOBlockStorageDriver\0"; + let read_key = CFString::from_static_string("Total Time (Read)"); + let write_key = CFString::from_static_string("Total Time (Write)"); + let stats_key = CFString::from_static_string("Statistics"); + + unsafe { + let matching = IOServiceMatching(CLASS.as_ptr() as *const c_char); + if matching.is_null() { + return None; + } + // IOServiceGetMatchingServices CONSUMES the matching dict's +1 ref, so we + // must not release `matching` ourselves. + let mut iter: io_iterator_t = 0; + if IOServiceGetMatchingServices(K_IO_MAIN_PORT_DEFAULT, matching, &mut iter) != KERN_SUCCESS + { + return None; + } + + let mut total: u64 = 0; + let mut saw_any = false; + loop { + let entry = IOIteratorNext(iter); + if entry == 0 { + break; + } + let stats = IORegistryEntryCreateCFProperty( + entry, + stats_key.as_concrete_TypeRef(), + std::ptr::null(), + 0, + ); + if !stats.is_null() { + let dict = stats as CFDictionaryRef; + if let Some(r) = dict_i64(dict, read_key.as_concrete_TypeRef()) { + total = total.saturating_add(r.max(0) as u64); + saw_any = true; + } + if let Some(w) = dict_i64(dict, write_key.as_concrete_TypeRef()) { + total = total.saturating_add(w.max(0) as u64); + saw_any = true; + } + // IORegistryEntryCreateCFProperty follows the CREATE rule (+1). + core_foundation_sys::base::CFRelease(stats); + } + IOObjectRelease(entry); + } + IOObjectRelease(iter); + + saw_any.then_some(total) + } +} + +/// Read a signed-64-bit CFNumber value for `key` out of a CFDictionary, or +/// `None` if the key is absent or not a number. +/// +/// # Safety +/// `dict` must be a valid `CFDictionaryRef` and `key` a valid `CFStringRef`. +unsafe fn dict_i64(dict: CFDictionaryRef, key: CFStringRef) -> Option { + let value = CFDictionaryGetValue(dict, key as *const c_void); + if value.is_null() { + return None; + } + let mut out: i64 = 0; + let ok = CFNumberGetValue( + value as CFNumberRef, + kCFNumberSInt64Type, + (&mut out as *mut i64).cast(), + ); + ok.then_some(out) +} diff --git a/crates/driven-diskstat/src/windows.rs b/crates/driven-diskstat/src/windows.rs new file mode 100644 index 00000000..ece82759 --- /dev/null +++ b/crates/driven-diskstat/src/windows.rs @@ -0,0 +1,136 @@ +//! Windows [`DiskBusyProbe`] backend (DESIGN s18.2): PDH `\PhysicalDisk(_Total) +//! \% Disk Time`. +//! +//! `% Disk Time` is the fraction of the interval the disk spent servicing +//! requests - exactly the "busy fraction" DESIGN s18.2 wants. PDH computes the +//! rate itself over the interval between two `PdhCollectQueryData` calls, so this +//! reader holds one query + counter handle for its lifetime, primes the query +//! with a collect at construction, and each [`sample`](DiskBusyProbe::sample) +//! does one more collect + a formatted read. The value is a PERCENT (0..~100+, +//! and it CAN exceed 100 across a busy multi-spindle `_Total`); we divide by 100 +//! and, consistent with the Linux backend, let an over-unity fraction read as +//! saturated rather than clamping. +//! +//! `PdhAddEnglishCounterW` (not the localized `PdhAddCounterW`) makes the counter +//! path locale-independent, so it resolves on a non-English Windows install. +//! +//! Fail-open: ANY non-zero PDH status - open/add/collect/format failure - makes +//! `sample` return [`DiskBusy::Unknown`], and a failed construction leaves the +//! handles `None` so every later sample is `Unknown` too. + +use std::path::PathBuf; +use std::sync::Mutex; + +use windows::core::PCWSTR; +use windows::Win32::System::Performance::{ + PdhAddEnglishCounterW, PdhCloseQuery, PdhCollectQueryData, PdhGetFormattedCounterValue, + PdhOpenQueryW, PDH_FMT, PDH_FMT_COUNTERVALUE, PDH_FMT_DOUBLE, PDH_HCOUNTER, PDH_HQUERY, +}; + +use crate::{DiskBusy, DiskBusyProbe}; + +/// `ERROR_SUCCESS` for the PDH status codes (a `u32` returned by every PDH call). +const PDH_SUCCESS: u32 = 0; + +/// `PDH_FMT_NOCAP100` (winperf.h `0x00008000`): report a genuinely-over-100% +/// `_Total` honestly instead of clamping to 100. Not re-exported as a named +/// constant by the `windows` crate, so defined here. +const PDH_FMT_NOCAP100: u32 = 0x0000_8000; + +/// The English (locale-independent) counter path for aggregate disk-busy. +const COUNTER_PATH: &str = r"\PhysicalDisk(_Total)\% Disk Time"; + +/// Owns the PDH query + counter handles and closes the query on drop. +struct PdhQuery { + query: PDH_HQUERY, + counter: PDH_HCOUNTER, +} + +// PDH handles are process-global kernel-ish handles; the query is only ever +// touched behind the outer `Mutex`, so it is safe to move across threads. +unsafe impl Send for PdhQuery {} + +impl Drop for PdhQuery { + fn drop(&mut self) { + // Best-effort close; nothing actionable on failure at teardown. + unsafe { + let _ = PdhCloseQuery(self.query); + } + } +} + +/// Windows disk-busy reader over PDH (DESIGN s18.2). +pub struct RealDiskBusyProbe { + /// `None` when the PDH query could not be opened / the counter added; then + /// every sample is [`DiskBusy::Unknown`] (fail-open). + query: Mutex>, +} + +impl RealDiskBusyProbe { + /// Open a PDH query for `\PhysicalDisk(_Total)\% Disk Time` and prime it with + /// one collect so the first [`sample`](DiskBusyProbe::sample) already has a + /// baseline interval. The `root` is unused on Windows (the `_Total` instance + /// already spans every physical disk) but accepted for signature parity with + /// the other backends. Never fails hard - a PDH error leaves the handle + /// `None`. + #[must_use] + pub fn new(_root: PathBuf) -> Self { + Self { + query: Mutex::new(open_query()), + } + } +} + +impl DiskBusyProbe for RealDiskBusyProbe { + fn sample(&self) -> DiskBusy { + let guard = self.query.lock().unwrap_or_else(|e| e.into_inner()); + let Some(q) = guard.as_ref() else { + return DiskBusy::Unknown; + }; + unsafe { + if PdhCollectQueryData(q.query) != PDH_SUCCESS { + return DiskBusy::Unknown; + } + let mut value = PDH_FMT_COUNTERVALUE::default(); + // NOCAP100 so a genuinely-over-100% `_Total` is reported honestly + // (it reads as saturated) rather than being clamped to 100. + let fmt = PDH_FMT(PDH_FMT_DOUBLE.0 | PDH_FMT_NOCAP100); + let status = PdhGetFormattedCounterValue(q.counter, fmt, None, &mut value); + if status != PDH_SUCCESS { + return DiskBusy::Unknown; + } + // SAFETY: we requested PDH_FMT_DOUBLE, so the union's doubleValue is + // the initialized member. + let percent = value.Anonymous.doubleValue; + if !percent.is_finite() || percent < 0.0 { + return DiskBusy::Unknown; + } + DiskBusy::Fraction(percent / 100.0) + } + } +} + +/// Open the PDH query, add the English counter, and prime it with one collect. +/// Returns `None` on any PDH failure (fail-open). +fn open_query() -> Option { + unsafe { + let mut query = PDH_HQUERY::default(); + if PdhOpenQueryW(PCWSTR::null(), 0, &mut query) != PDH_SUCCESS { + return None; + } + let path: Vec = COUNTER_PATH + .encode_utf16() + .chain(std::iter::once(0)) + .collect(); + let mut counter = PDH_HCOUNTER::default(); + let added = PdhAddEnglishCounterW(query, PCWSTR(path.as_ptr()), 0, &mut counter); + if added != PDH_SUCCESS { + let _ = PdhCloseQuery(query); + return None; + } + // Prime: a rate counter needs a first collect to establish the baseline + // interval; the next collect (first `sample`) then yields a real rate. + let _ = PdhCollectQueryData(query); + Some(PdhQuery { query, counter }) + } +} diff --git a/crates/driven-test-fixtures/Cargo.toml b/crates/driven-test-fixtures/Cargo.toml index 205026a4..8f8f3c25 100644 --- a/crates/driven-test-fixtures/Cargo.toml +++ b/crates/driven-test-fixtures/Cargo.toml @@ -19,3 +19,4 @@ pretty_assertions = "1" driven-core = { path = "../driven-core" } driven-drive = { path = "../driven-drive" } driven-power = { path = "../driven-power" } +driven-diskstat = { path = "../driven-diskstat" } diff --git a/crates/driven-test-fixtures/src/diskstat.rs b/crates/driven-test-fixtures/src/diskstat.rs new file mode 100644 index 00000000..b6027150 --- /dev/null +++ b/crates/driven-test-fixtures/src/diskstat.rs @@ -0,0 +1,74 @@ +//! [`FakeDiskBusyProbe`] - a test double implementing +//! [`driven_diskstat::DiskBusyProbe`]. +//! +//! Tests of the adaptive-parallelism controller drive the disk-saturation gate +//! deterministically via a fixed reading, or a scripted sequence of readings, +//! instead of touching the real per-OS backend (`/proc/diskstats`, PDH, IOKit). + +use std::sync::Mutex; + +use driven_diskstat::{DiskBusy, DiskBusyProbe}; + +/// A [`DiskBusyProbe`] that returns a caller-controlled reading. +/// +/// Construct with [`not_saturated`](Self::not_saturated) / +/// [`saturated`](Self::saturated) for a constant reading, or +/// [`scripted`](Self::scripted) for a sequence consumed one-per-`sample` +/// (the last entry repeats once the script is exhausted). +#[derive(Debug)] +pub struct FakeDiskBusyProbe { + /// The reading returned when the script is empty / exhausted. + fixed: DiskBusy, + /// Remaining scripted readings, consumed front-to-back. + script: Mutex>, +} + +impl FakeDiskBusyProbe { + /// A probe that always reports the disk as NOT saturated (0% busy) - the + /// common case that lets the controller grow/shrink on throughput alone. + #[must_use] + pub fn not_saturated() -> Self { + Self::constant(DiskBusy::Fraction(0.0)) + } + + /// A probe that always reports the disk as saturated (100% busy). + #[must_use] + pub fn saturated() -> Self { + Self::constant(DiskBusy::Fraction(1.0)) + } + + /// A probe that always reports [`DiskBusy::Unknown`] (an unreadable device), + /// exercising the fail-open path. + #[must_use] + pub fn unknown() -> Self { + Self::constant(DiskBusy::Unknown) + } + + /// A probe returning a constant reading. + #[must_use] + pub fn constant(reading: DiskBusy) -> Self { + Self { + fixed: reading, + script: Mutex::new(std::collections::VecDeque::new()), + } + } + + /// A probe returning each reading in turn; once the script is exhausted every + /// further `sample` returns the final scripted reading. + #[must_use] + pub fn scripted(readings: impl IntoIterator) -> Self { + let script: std::collections::VecDeque = readings.into_iter().collect(); + let fixed = script.back().copied().unwrap_or(DiskBusy::Unknown); + Self { + fixed, + script: Mutex::new(script), + } + } +} + +impl DiskBusyProbe for FakeDiskBusyProbe { + fn sample(&self) -> DiskBusy { + let mut s = self.script.lock().unwrap_or_else(|e| e.into_inner()); + s.pop_front().unwrap_or(self.fixed) + } +} diff --git a/crates/driven-test-fixtures/src/lib.rs b/crates/driven-test-fixtures/src/lib.rs index a38d1050..73b5ca58 100644 --- a/crates/driven-test-fixtures/src/lib.rs +++ b/crates/driven-test-fixtures/src/lib.rs @@ -6,6 +6,9 @@ //! declaratively. //! - [`clock`]: a [`FakeClock`](clock::FakeClock) implementing //! [`driven_core::time::Clock`] with `advance()` + `now_set()`. +//! - [`diskstat`]: a [`FakeDiskBusyProbe`](diskstat::FakeDiskBusyProbe) +//! implementing [`driven_diskstat::DiskBusyProbe`] for the adaptive +//! upload-parallelism controller tests. //! - [`power`]: a [`FakePowerSource`](power::FakePowerSource) //! implementing [`driven_power::PowerSource`] with a `set()` driver //! for state transitions. @@ -22,6 +25,7 @@ pub mod assert; pub mod clock; +pub mod diskstat; pub mod network; pub mod power; pub mod tree; diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 14199693..a9fde213 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -89,6 +89,9 @@ driven-core = { path = "../crates/driven-core" } driven-drive = { path = "../crates/driven-drive" } driven-crypto = { path = "../crates/driven-crypto" } driven-power = { path = "../crates/driven-power" } +# Adaptive upload parallelism (DESIGN s11.4.7): the real per-OS disk-busy reader +# wired into the AdaptiveController in assembly. +driven-diskstat = { path = "../crates/driven-diskstat" } # M5 assembly seams: real network breaker backend + Windows VSS provider. driven-net = { path = "../crates/driven-net" } # Issue #34 corporate CA pinning: the shared helper that threads a user-configured diff --git a/src-tauri/src/assembly.rs b/src-tauri/src/assembly.rs index 8ac6bc1c..6b692bf2 100644 --- a/src-tauri/src/assembly.rs +++ b/src-tauri/src/assembly.rs @@ -514,26 +514,66 @@ async fn build_account( let crypto = Arc::new(KeystoreCryptoProvider::new(account.id, sources.clone())); let crypto_dyn: Arc = crypto.clone(); + // --- adaptive upload parallelism (DESIGN s11.4.7) ----------------------- + // Build the resizable upload pool sized at the user's + // `default_concurrent_uploads` setting (finally wired; `None` auto-picks + // `min(available_parallelism*2, 16)`), plus the throughput probe. Both are + // shared by `Arc` into the executor (which acquires + records bytes) AND, when + // adaptive is enabled, into the controller (which resizes + drains) - the same + // one-Arc-into-two-consumers pattern as the pacer + latency reservoir. + let pool_start = config + .default_concurrent_uploads + .map(|n| n as usize) + .unwrap_or_else(driven_core::adaptive::default_pool_size); + let upload_pool = driven_core::adaptive::UploadPool::new(pool_start); + let throughput = driven_core::adaptive::ThroughputProbe::new(); + // Capture what the controller needs BEFORE `config` / `clock` / `pacer` are + // moved into the executor + orchestrator below. The disk-busy reader is bound + // to the first source's root (only the Linux backend uses it to pick the + // backing device; the Windows `_Total` / macOS aggregate backends ignore it); + // an account with no sources yet falls back to `.` and simply reads Unknown. + let adaptive_enabled = config.adaptive_parallelism_enabled; + let disk_root = sources + .first() + .map(|s| std::path::PathBuf::from(&s.local_path)) + .unwrap_or_else(|| std::path::PathBuf::from(".")); + let clock_for_adaptive = clock.clone(); + let pacer_for_adaptive = pacer.clone(); + // --- executor ----------------------------------------------------------- - let executor: Arc = Arc::new( - DefaultExecutor::with_clock( - ExecutorDeps { - remote, - state: state.clone(), - // Clone so the orchestrator can share the SAME pacer for the V2 - // metered throttle (`with_pacer` below) - a runtime cap change must - // be seen by this executor's upload path. - pacer: pacer.clone(), - crypto: Some(crypto_dyn), - vss: vss.clone(), - network: Some(network.clone()), - }, - clock.clone(), - ) - // DESIGN s13: the SAME app-global reservoir the orchestrator's scans - // record into, for the upload-per-MB latency metric. - .with_latency_reservoir(latency.clone()), - ); + // The upload pool is wired in BOTH modes: it carries the start size + // (`default_concurrent_uploads`, or the auto default) that the executor's + // per-file gate uses whether or not adaptation is on. The throughput probe, + // by contrast, is wired ONLY when adaptive is enabled (F4): with the + // kill-switch off nothing drains it, so feeding it would be pure hot-path + // overhead - and `record_bytes` is then skipped entirely (the executor's + // `throughput` stays `None`), so "disabled" truly means zero adaptive code on + // the upload path. + let mut exec = DefaultExecutor::with_clock( + ExecutorDeps { + remote, + state: state.clone(), + // Clone so the orchestrator can share the SAME pacer for the V2 + // metered throttle (`with_pacer` below) - a runtime cap change must + // be seen by this executor's upload path. + pacer: pacer.clone(), + crypto: Some(crypto_dyn), + vss: vss.clone(), + network: Some(network.clone()), + }, + clock.clone(), + ) + // DESIGN s13: the SAME app-global reservoir the orchestrator's scans + // record into, for the upload-per-MB latency metric. + .with_latency_reservoir(latency.clone()) + // DESIGN s11.4.7: the SAME pool the controller resizes (fixed here when the + // kill-switch is off, honouring `default_concurrent_uploads` either way). + .with_upload_pool(upload_pool.clone()); + if adaptive_enabled { + // DESIGN s11.4.7: the SAME probe the controller drains. + exec = exec.with_throughput_probe(throughput.clone()); + } + let executor: Arc = Arc::new(exec); // --- orchestrator ------------------------------------------------------- // Held as the CONCRETE `Arc` (not `Arc`) @@ -565,6 +605,24 @@ async fn build_account( // 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()); + // DESIGN s11.4.7: adaptive upload parallelism. Default-ON; when the kill-switch + // (`adaptive_parallelism_enabled`) is off, no controller is wired and the pool + // stays fixed at `default_concurrent_uploads`. The controller resizes the SAME + // `upload_pool` the executor acquires from and drains the SAME `throughput` + // probe it feeds, gated by the real per-OS disk-busy reader (DESIGN s18.2) and + // the account's pacer (for the window-scoped "not throttling" check). + if adaptive_enabled { + let disk: Arc = + Arc::new(driven_diskstat::RealDiskBusyProbe::new(disk_root)); + let controller = Arc::new(driven_core::adaptive::AdaptiveController::new( + upload_pool.clone(), + throughput.clone(), + disk, + pacer_for_adaptive, + clock_for_adaptive, + )); + orchestrator = orchestrator.with_adaptive_controller(controller); + } let orchestrator = Arc::new(orchestrator); // R-P1-1: one shutdown signal both bridges select! on, so quit can stop the diff --git a/src-tauri/src/commands/dtos.rs b/src-tauri/src/commands/dtos.rs index b5265414..9cf3ebd1 100644 --- a/src-tauri/src/commands/dtos.rs +++ b/src-tauri/src/commands/dtos.rs @@ -34,6 +34,13 @@ where Deserialize::deserialize(deserializer).map(Some) } +/// `serde(default)` helper: a bool field absent from a persisted blob defaults to +/// `true`. Used for opt-out flags like `adaptive_parallelism_enabled` so a +/// settings blob written before the field existed reads as enabled. +fn default_true() -> bool { + true +} + // ----------------------------------------------------------------------------- // Accounts (SPEC s11.1) // ----------------------------------------------------------------------------- @@ -412,8 +419,16 @@ pub struct SettingsDto { pub struct GlobalSettings { /// Launch Driven on login. pub auto_start_on_login: bool, - /// `null` = auto-pick concurrency; else a user override `1..=32`. + /// `null` = auto-pick concurrency; else a user override `1..=32`. With + /// [`Self::adaptive_parallelism_enabled`] on this is the STARTING pool size + /// (it then floats within `1..=32`); with it off the pool is fixed here. pub default_concurrent_uploads: Option, + /// Whether the adaptive upload-parallelism controller runs (DESIGN s11.4.7). + /// `true` (default) lets the in-flight pool grow/shrink with measured + /// throughput + disk-busy; `false` pins it at + /// [`Self::default_concurrent_uploads`]. + #[serde(default = "default_true")] + pub adaptive_parallelism_enabled: bool, /// `null` = unlimited; else the cap in megabits/sec. pub bandwidth_cap_mbps: Option, /// Skip sync while on battery. @@ -590,6 +605,8 @@ pub struct GlobalSettingsPatch { /// key `None` ("leave unchanged"). #[serde(default, deserialize_with = "double_option")] pub default_concurrent_uploads: Option>, + /// See [`GlobalSettings::adaptive_parallelism_enabled`]. Present = set it. + pub adaptive_parallelism_enabled: Option, /// See [`GlobalSettings::bandwidth_cap_mbps`]. `double_option`: `null` = /// `Some(None)` ("reset to unlimited"). #[serde(default, deserialize_with = "double_option")] diff --git a/src-tauri/src/commands/settings.rs b/src-tauri/src/commands/settings.rs index 9c225df4..2840f538 100644 --- a/src-tauri/src/commands/settings.rs +++ b/src-tauri/src/commands/settings.rs @@ -308,6 +308,9 @@ pub async fn update_settings( } cur.default_concurrent_uploads = v; } + if let Some(v) = g.adaptive_parallelism_enabled { + cur.adaptive_parallelism_enabled = v; + } if let Some(v) = g.bandwidth_cap_mbps { // `None` = unlimited (valid); `Some(n)` must be in range. if let Some(n) = v { @@ -771,6 +774,11 @@ mod storage { pub struct Global { pub auto_start_on_login: bool, pub default_concurrent_uploads: Option, + // Added with adaptive parallelism (DESIGN s11.4.7). `serde(default)` + // returns `true` so a `global` blob persisted before this field still + // deserialises with adaptation ON (the default-on behaviour). + #[serde(default = "default_adaptive_parallelism_enabled")] + pub adaptive_parallelism_enabled: bool, pub bandwidth_cap_mbps: Option, pub skip_on_battery: bool, pub skip_on_metered: bool, @@ -807,6 +815,12 @@ mod storage { 60 } + /// Default for a `global` blob predating adaptive parallelism: ON (DESIGN + /// s11.4.7 ships default-on). + fn default_adaptive_parallelism_enabled() -> bool { + true + } + /// Default metered mode (V1 behaviour: pause) for a pre-V2 `global` blob. fn default_metered_mode() -> String { "pause".to_string() @@ -817,6 +831,7 @@ mod storage { GlobalSettings { auto_start_on_login: s.auto_start_on_login, default_concurrent_uploads: s.default_concurrent_uploads, + adaptive_parallelism_enabled: s.adaptive_parallelism_enabled, bandwidth_cap_mbps: s.bandwidth_cap_mbps, skip_on_battery: s.skip_on_battery, skip_on_metered: s.skip_on_metered, @@ -840,6 +855,7 @@ mod storage { Global { auto_start_on_login: d.auto_start_on_login, default_concurrent_uploads: d.default_concurrent_uploads, + adaptive_parallelism_enabled: d.adaptive_parallelism_enabled, bandwidth_cap_mbps: d.bandwidth_cap_mbps, skip_on_battery: d.skip_on_battery, skip_on_metered: d.skip_on_metered, @@ -1168,6 +1184,8 @@ pub async fn load_orchestrator_config(state: &dyn StateRepo) -> CommandResult GlobalSettings { // code default applies only when the `global` group is entirely absent. auto_start_on_login: true, default_concurrent_uploads: None, + // Adaptive parallelism ships default-on (DESIGN s11.4.7). + adaptive_parallelism_enabled: true, bandwidth_cap_mbps: None, skip_on_battery: true, skip_on_metered: true, diff --git a/ui/src/__tests__/settings-components.test.ts b/ui/src/__tests__/settings-components.test.ts index 755e7361..2b088d24 100644 --- a/ui/src/__tests__/settings-components.test.ts +++ b/ui/src/__tests__/settings-components.test.ts @@ -66,6 +66,7 @@ function makeSettings(over: Partial = {}): SettingsDto { global: { autoStartOnLogin: false, defaultConcurrentUploads: null, + adaptiveParallelismEnabled: true, bandwidthCapMbps: null, skipOnBattery: true, skipOnMetered: true, @@ -1233,6 +1234,29 @@ describe("Settings Rules tab", () => { }); }); + it("toggles adaptive upload parallelism (DESIGN 11.4.7)", async () => { + // The kill-switch reflects the persisted value and patches on toggle. Starts + // ON (makeSettings default), so unchecking it sends `false`. + invokeMock.mockImplementation((cmd: string, args: unknown) => { + if (cmd === "get_settings") return Promise.resolve(makeSettings()); + if (cmd === "update_settings") { + const patch = (args as { patch: Record }).patch; + return Promise.resolve(makeSettings(patch as Partial)); + } + return Promise.resolve(undefined); + }); + const wrapper = mount(Settings, { props: { tab: "rules" }, global: globalMountOptions }); + await flushPromises(); + const toggle = wrapper.get('[data-testid="adaptive-parallelism-toggle"]'); + expect((toggle.element as HTMLInputElement).checked).toBe(true); + await toggle.setValue(false); + await toggle.trigger("change"); + await flushPromises(); + expect(invokeMock).toHaveBeenCalledWith("update_settings", { + patch: { global: { adaptiveParallelismEnabled: false } }, + }); + }); + it("keeps the Rules form visible with a localized banner when a patch is rejected", async () => { // Regression: a rejected patch must NOT replace the whole form with the raw // error ("[object Object]") and brick the page. The form stays mounted and an diff --git a/ui/src/__tests__/settings-stores.test.ts b/ui/src/__tests__/settings-stores.test.ts index 11d71b2e..d499bf4f 100644 --- a/ui/src/__tests__/settings-stores.test.ts +++ b/ui/src/__tests__/settings-stores.test.ts @@ -61,6 +61,7 @@ function makeSettings(over: Partial = {}): SettingsDto { global: { autoStartOnLogin: false, defaultConcurrentUploads: null, + adaptiveParallelismEnabled: true, bandwidthCapMbps: null, skipOnBattery: true, skipOnMetered: true, diff --git a/ui/src/ipc/types.ts b/ui/src/ipc/types.ts index ce8767a8..c8a260be 100644 --- a/ui/src/ipc/types.ts +++ b/ui/src/ipc/types.ts @@ -194,6 +194,8 @@ export interface ScheduleSettings { export interface GlobalSettings { autoStartOnLogin: boolean; defaultConcurrentUploads: number | null; + /** Whether the adaptive upload-parallelism controller runs (DESIGN 11.4.7). */ + adaptiveParallelismEnabled: boolean; bandwidthCapMbps: number | null; skipOnBattery: boolean; skipOnMetered: boolean; @@ -272,6 +274,7 @@ export interface SettingsDto { export interface GlobalSettingsPatch { autoStartOnLogin?: boolean; defaultConcurrentUploads?: number | null; + adaptiveParallelismEnabled?: boolean; bandwidthCapMbps?: number | null; skipOnBattery?: boolean; skipOnMetered?: boolean; diff --git a/ui/src/locales/en-US.json b/ui/src/locales/en-US.json index 2e6629d2..081a1eb6 100644 --- a/ui/src/locales/en-US.json +++ b/ui/src/locales/en-US.json @@ -215,6 +215,8 @@ "bandwidthCapUnlimited": "Unlimited", "concurrentUploadsLabel": "Concurrent uploads", "concurrentUploadsAuto": "Auto", + "adaptiveParallelismLabel": "Adapt upload concurrency automatically", + "adaptiveParallelismNote": "Tunes how many files upload at once based on measured throughput and disk load, starting from the concurrent-uploads value above. Turn off to hold that number fixed.", "scanIntervalLabel": "Scan interval (seconds)", "deepVerifyIntervalLabel": "Deep-verify interval (seconds)", "ioPriorityLabel": "I/O priority", diff --git a/ui/src/views/Settings.vue b/ui/src/views/Settings.vue index 8b40b3e8..4b18c631 100644 --- a/ui/src/views/Settings.vue +++ b/ui/src/views/Settings.vue @@ -304,6 +304,14 @@ async function commitConcurrentUploads(): Promise { }); } +// DESIGN 11.4.7: adaptive upload parallelism (default ON). When on, the +// in-flight pool grows/shrinks with measured throughput + disk-busy starting +// from the concurrency setting above; when off, the pool is pinned at it. +async function setAdaptiveParallelism(event: Event): Promise { + const checked = (event.target as HTMLInputElement).checked; + await commitPatch({ global: { adaptiveParallelismEnabled: checked } }); +} + async function commitScanInterval(event: Event): Promise { const current = settings.settings?.global.scanIntervalSecs ?? 600; const value = parseRequiredClamped( @@ -689,6 +697,20 @@ async function setTelemetryEnabled(event: Event): Promise { /> + +

+ {{ t("settings.rules.adaptiveParallelismNote") }} +

+