feat(telemetry): capture latency percentiles and add rollup query endpoint

crates/driven-core/src/executor.rs

@@ -815,6 +815,12 @@ pub struct DefaultExecutor {
     /// the file size - the one qualitative pipeline contract that IS
     /// deterministically measurable against the instantaneous fake.
     mem_gauge: Option<Arc<MemGauge>>,
+    /// App-global latency sampler (DESIGN s13 telemetry), or `None` when
+    /// telemetry latency capture is not wired (every test + the chaos harness).
+    /// When `Some`, each completed upload op records its wall-clock latency
+    /// normalized per MiB via [`crate::telemetry::per_mb_ms`]; the reservoir's
+    /// own enable gate makes the record a no-op when telemetry is off.
+    latency: Option<Arc<crate::telemetry::LatencyReservoir>>,
     #[cfg(test)]
     mid_upload_hook: Option<MidUploadHook>,
     #[cfg(test)]
@@ -893,13 +899,29 @@ impl DefaultExecutor {
             vss: deps.vss,
             pool,
             mem_gauge: None,
+            latency: None,
             #[cfg(test)]
             mid_upload_hook: None,
             #[cfg(test)]
             post_upload_hook: None,
         }
     }
 
+    /// Attach the app-global latency reservoir (DESIGN s13 telemetry) so each
+    /// completed upload op records its per-MiB latency. Mirrors
+    /// [`Self::with_mem_gauge`]: a builder over the always-present `Option`
+    /// field, so no test / e2e struct-literal site changes. Production wires it
+    /// from the assembly; every other construction path leaves it `None` (zero
+    /// overhead).
+    #[must_use]
+    pub fn with_latency_reservoir(
+        mut self,
+        reservoir: Arc<crate::telemetry::LatencyReservoir>,
+    ) -> Self {
+        self.latency = Some(reservoir);
+        self
+    }
+
     /// Resolve the crypto decision for one source (M5 GA-blocking surface).
     ///
     /// Consults the injected [`CryptoProvider`]; a `None` provider means every
@@ -4467,6 +4489,20 @@ impl<'a> ExecOne<'a> {
         permit: tokio::sync::OwnedSemaphorePermit,
         on_outcome: &OutcomeSink<'_>,
     ) -> anyhow::Result<OpOutcome> {
+        // Telemetry (DESIGN s13): time upload ops so a completed upload records
+        // its per-MiB latency. Only armed for the upload op kinds and only when a
+        // reservoir is wired + enabled (a trash op is not an "upload-per-MB"
+        // sample). Zero cost otherwise (not even an `Instant::now`).
+        let upload_timer = match op {
+            Op::HashThenUpload { .. } | Op::UploadBundle { .. } => self
+                .this
+                .latency
+                .as_ref()
+                .filter(|r| r.is_enabled())
+                .map(|_| std::time::Instant::now()),
+            Op::Trash { .. } => None,
+        };
+
         let out = match op {
             Op::HashThenUpload {
                 source_id,
@@ -4493,6 +4529,32 @@ impl<'a> ExecOne<'a> {
                 self.this.bundle_upload(self.source, members).await
             }
         };
+
+        // 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 {
+                    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) {
+                        reservoir.record_upload_per_mb_ms(per_mb);
+                    }
+                }
+            }
+        }
+
         // Stream the per-op activity for a produced outcome (the op's durable
         // file-state commit already happened inside hash_then_upload / trash_op).
         // A hard error (Err) carries no OpOutcome and is handled by the caller.

crates/driven-core/src/lib.rs

@@ -30,6 +30,7 @@ pub mod pacer;
 pub mod planner;
 pub mod scanner;
 pub mod state;
+pub mod telemetry;
 pub mod time;
 pub mod types;
 pub mod watcher;

crates/driven-core/src/orchestrator.rs

@@ -433,6 +433,13 @@ pub struct SyncOrchestrator {
     /// token. An atomic so the cycle path can set it and the run loop can read it
     /// without holding a lock across the select.
     suspended: std::sync::atomic::AtomicBool,
+    /// App-global latency sampler (DESIGN s13 telemetry), or `None` when
+    /// telemetry latency capture is not wired (tests / the chaos harness).
+    /// Threaded into [`crate::scanner::scan_with_latency`] so each scan records
+    /// per-file processing latency; the SAME `Arc` is also given to the executor
+    /// (via [`crate::executor::DefaultExecutor::with_latency_reservoir`]) for the
+    /// upload-per-MB metric. Set via [`Self::with_latency_reservoir`].
+    latency: Option<Arc<crate::telemetry::LatencyReservoir>>,
 }
 
 impl SyncOrchestrator {
@@ -483,9 +490,25 @@ impl SyncOrchestrator {
             vss_create_ledger: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
             orphan_cleanup_done: Mutex::new(false),
             suspended: std::sync::atomic::AtomicBool::new(false),
+            latency: None,
         }
     }
 
+    /// 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
+    /// [`DefaultExecutor::with_latency_reservoir`](crate::executor::DefaultExecutor::with_latency_reservoir)
+    /// so both metrics feed one reservoir. Without this, scans capture nothing
+    /// (the tests / chaos harness path).
+    #[must_use]
+    pub fn with_latency_reservoir(
+        mut self,
+        reservoir: Arc<crate::telemetry::LatencyReservoir>,
+    ) -> Self {
+        self.latency = Some(reservoir);
+        self
+    }
+
     /// Attach the per-cycle Windows VSS snapshot provider (ROADMAP M3.5).
     ///
     /// Pass the SAME `Arc<dyn VssProvider>` that was threaded into the
@@ -1294,7 +1317,13 @@ impl SyncOrchestrator {
             scanned: 0,
         })
         .await;
-        let scan = crate::scanner::scan(source, self.state.as_ref(), mode).await?;
+        let scan = crate::scanner::scan_with_latency(
+            source,
+            self.state.as_ref(),
+            mode,
+            self.latency.as_deref(),
+        )
+        .await?;
 
         // DESIGN s5.5: flag still-present-but-now-excluded paths so the UI can
         // surface them; never a trash. Non-fatal - a flag write failure must

crates/driven-core/src/scanner.rs

@@ -156,10 +156,29 @@ fn has_alternate_data_streams(path: &Path) -> bool {
 /// Pure aside from local filesystem reads and the `state` load; emits no
 /// ops and mutates no state - the planner (SPEC s7) and executor (SPEC s8)
 /// own those side effects.
+///
+/// Thin wrapper over [`scan_with_latency`] with no telemetry capture; the
+/// existing tests + callers that do not thread a reservoir use this.
 pub async fn scan(
     source: &SourceRow,
     state: &dyn StateRepo,
     mode: ScanMode,
+) -> anyhow::Result<ScanResult> {
+    scan_with_latency(source, state, mode, None).await
+}
+
+/// [`scan`] with an optional [`LatencyReservoir`](crate::telemetry::LatencyReservoir)
+/// for per-file scan-processing latency capture (DESIGN s13 telemetry). When
+/// `latency` is `Some` AND capture is enabled, each fully-processed file records
+/// its stat-through-change-detection wall-clock time (the dominant cost is the
+/// BLAKE3 re-hash on a deep-verify pass); when `None` there is zero overhead.
+/// The orchestrator passes its shared reservoir here; every other caller passes
+/// `None`.
+pub async fn scan_with_latency(
+    source: &SourceRow,
+    state: &dyn StateRepo,
+    mode: ScanMode,
+    latency: Option<&crate::telemetry::LatencyReservoir>,
 ) -> anyhow::Result<ScanResult> {
     let known = state
         .load_source_file_state(source.id)
@@ -292,6 +311,16 @@ pub async fn scan(
             continue;
         }
 
+        // Telemetry (DESIGN s13): time this file's stat-through-change-detection
+        // processing. Only armed when a reservoir is threaded in AND capture is
+        // enabled, so a scan with no telemetry pays nothing (not even the
+        // `Instant::now`). Recorded at the end of the iteration for a
+        // fully-processed file; the cheap early-`continue` skips below (cloud-only
+        // placeholder, NFC collision) intentionally record nothing.
+        let file_timer = latency
+            .filter(|r| r.is_enabled())
+            .map(|_| std::time::Instant::now());
+
         let meta = match entry.metadata() {
             Ok(m) => m,
             Err(err) => {
@@ -388,6 +417,14 @@ pub async fn scan(
                 mtime_ns,
             });
         }
+
+        // Telemetry (DESIGN s13): record this file's per-file scan-processing
+        // latency. `latency` is `Some` iff a reservoir was armed above; the
+        // `record_scan_ms` call re-checks the enable gate (a no-op if telemetry
+        // was disabled mid-scan).
+        if let (Some(res), Some(started)) = (latency, file_timer) {
+            res.record_scan_ms(u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX));
+        }
     }
 
     // Split the known-but-not-seen paths into genuine deletions vs