diff --git a/crates/driven-core/src/executor.rs b/crates/driven-core/src/executor.rs index febf2266..bd0e4b97 100644 --- a/crates/driven-core/src/executor.rs +++ b/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>, + /// 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>, #[cfg(test)] mid_upload_hook: Option, #[cfg(test)] @@ -893,6 +899,7 @@ impl DefaultExecutor { vss: deps.vss, pool, mem_gauge: None, + latency: None, #[cfg(test)] mid_upload_hook: None, #[cfg(test)] @@ -900,6 +907,21 @@ impl DefaultExecutor { } } + /// 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, + ) -> 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 { + // 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. diff --git a/crates/driven-core/src/lib.rs b/crates/driven-core/src/lib.rs index 07e4409a..39b54392 100644 --- a/crates/driven-core/src/lib.rs +++ b/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; diff --git a/crates/driven-core/src/orchestrator.rs b/crates/driven-core/src/orchestrator.rs index 04094112..dcd36f1f 100644 --- a/crates/driven-core/src/orchestrator.rs +++ b/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>, } 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, + ) -> Self { + self.latency = Some(reservoir); + self + } + /// Attach the per-cycle Windows VSS snapshot provider (ROADMAP M3.5). /// /// Pass the SAME `Arc` 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 diff --git a/crates/driven-core/src/scanner.rs b/crates/driven-core/src/scanner.rs index 4a347f6b..33a792fd 100644 --- a/crates/driven-core/src/scanner.rs +++ b/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 { + 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 { 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 diff --git a/crates/driven-core/src/telemetry.rs b/crates/driven-core/src/telemetry.rs new file mode 100644 index 00000000..493af35e --- /dev/null +++ b/crates/driven-core/src/telemetry.rs @@ -0,0 +1,355 @@ +//! In-memory latency reservoirs for the anonymous telemetry ping (DESIGN s13, +//! SPEC s16). +//! +//! DESIGN s13 lists "latency histograms (p50, p95) for scan and +//! upload-per-file" in the telemetry payload, but V1 shipped the wire keys +//! (`latency_p50_p95_ms.{scan,upload_per_mb}`) as ALWAYS-EMPTY arrays because +//! nothing captured per-op durations. This module is that capture: a small, +//! allocation-cheap, in-memory sampler the hot paths (the scanner's per-file +//! loop, the executor's per-upload op) feed at op completion, and the telemetry +//! ping reads at report-build time. +//! +//! SHAPE (mirrors the [`crate::executor::MemGauge`] instrumentation seam): a +//! single [`LatencyReservoir`] is created once per app, shared as an `Arc` into +//! every account's executor + orchestrator, and held on the app's telemetry +//! runtime. It is NEVER persisted - a restart starts empty (latency is a +//! best-effort signal, not durable state). +//! +//! PRIVACY / CONSENT (load-bearing, SPEC s16): capture is gated on the +//! telemetry-enabled pref. When telemetry is OFF, every `record_*` call is a +//! cheap no-op AND [`LatencyReservoir::set_enabled(false)`] drops any samples +//! already captured, so opting out cannot leave latency data lingering. A +//! latency sample is a bare millisecond duration - it carries no path, name, or +//! content, so it is privacy-safe by construction. +//! +//! WINDOWING: the reservoir is a bounded ring buffer per metric (most-recent +//! [`RESERVOIR_CAP`] samples), and [`LatencyReservoir::reset`] clears it. The +//! ping path takes a read-only [`LatencyReservoir::snapshot`] when it BUILDS the +//! payload, and resets ONLY after a SUCCESSFUL send - so a dropped/aborted ping +//! re-uses the same window's samples on the next attempt (matching how the +//! event-count aggregates re-send an un-checkpointed window). + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Mutex; + +/// Max samples retained per metric. A few thousand is plenty for a stable p50/ +/// p95 over a 24h window while staying tiny in memory (`u64` * cap = ~32 KiB per +/// metric) and allocation-free on the hot path once warmed (the ring overwrites +/// in place). Most-recent-wins: once full, new samples overwrite the oldest. +const RESERVOIR_CAP: usize = 4096; + +/// Bytes in one mebibyte - the normalizer for the upload-per-MB metric. Binary +/// MiB matches the byte-oriented accounting used elsewhere in the payload +/// (`bytes_uploaded` is raw bytes). +pub const BYTES_PER_MB: u64 = 1 << 20; + +/// A single metric's bounded ring buffer of millisecond samples. +#[derive(Debug, Default)] +struct Reservoir { + /// The retained samples (at most [`RESERVOIR_CAP`]). + samples: Vec, + /// Next overwrite index once at capacity (ring cursor). + next: usize, +} + +impl Reservoir { + /// Record one sample, overwriting the oldest once at capacity. + fn record(&mut self, v: u64) { + if self.samples.len() < RESERVOIR_CAP { + self.samples.push(v); + } else { + self.samples[self.next] = v; + self.next += 1; + if self.next >= RESERVOIR_CAP { + self.next = 0; + } + } + } + + /// Drop every sample (window reset). + fn clear(&mut self) { + self.samples.clear(); + self.next = 0; + } + + /// `[p50, p95]` (nearest-rank) over the current samples, or an EMPTY vec + /// when there are none - so the wire keeps emitting empty arrays until real + /// data exists (the pre-existing V1 behaviour + what the Worker tolerates). + fn percentiles(&self) -> Vec { + percentiles_p50_p95(&self.samples) + } +} + +/// Compute `[p50, p95]` by the nearest-rank method, or an empty vec when +/// `samples` is empty. Split out (pure, over a slice) so the percentile math is +/// unit-tested directly against the edge cases (empty / single / even / odd). +#[must_use] +fn percentiles_p50_p95(samples: &[u64]) -> Vec { + if samples.is_empty() { + return Vec::new(); + } + let mut sorted = samples.to_vec(); + sorted.sort_unstable(); + vec![nearest_rank(&sorted, 50), nearest_rank(&sorted, 95)] +} + +/// Nearest-rank percentile of a NON-EMPTY sorted slice: `rank = ceil(p/100 * n)` +/// (1-indexed), clamped to `[1, n]`, returning `sorted[rank - 1]`. For `n == 1` +/// every percentile is the single sample; for the max percentile it lands on the +/// last element. +#[must_use] +fn nearest_rank(sorted: &[u64], p: u32) -> u64 { + debug_assert!( + !sorted.is_empty(), + "nearest_rank requires a non-empty slice" + ); + let n = sorted.len() as u64; + // ceil(p * n / 100) via integer arithmetic. + let rank = (u64::from(p) * n).div_ceil(100); + let idx = rank.clamp(1, n) as usize - 1; + sorted[idx] +} + +/// The `[p50, p95]` pairs for both latency metrics, as read at ping-build time. +/// Each vec is either empty (no samples this window) or exactly `[p50, p95]`. +/// Mirrors the wire shape of the telemetry payload's `latency_p50_p95_ms`. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct LatencyPercentiles { + /// `[p50, p95]` per-file scan-processing latency in ms (empty when none). + pub scan: Vec, + /// `[p50, p95]` upload latency normalized per MiB in ms (empty when none). + pub upload_per_mb: Vec, +} + +/// App-global latency sampler shared (as an `Arc`) into every account's executor +/// and orchestrator and held on the telemetry runtime. Cheap to `record_*` into +/// from the hot paths; snapshotted + reset by the telemetry ping. +#[derive(Debug)] +pub struct LatencyReservoir { + /// Consent gate (mirrors the telemetry-enabled pref). When false, `record_*` + /// is a no-op and the buffers are kept empty. + enabled: AtomicBool, + /// Per-file scan-processing durations (ms). + scan: Mutex, + /// Per-upload latency normalized per MiB (ms). + upload_per_mb: Mutex, +} + +impl Default for LatencyReservoir { + /// A default-ON reservoir (telemetry is DEFAULT ON, SPEC s16). Boot replaces + /// this with one initialized from the persisted `telemetry.enabled` pref + /// before any capture happens; the default is only the pre-install placeholder + /// and the no-orchestrator (quiesced) app state. + fn default() -> Self { + Self::new(true) + } +} + +impl LatencyReservoir { + /// Create a reservoir with the initial consent state (set from the persisted + /// `telemetry.enabled` pref at boot, BEFORE any executor/scanner can capture, + /// so a user who opted out gets no startup capture window). + #[must_use] + pub fn new(enabled: bool) -> Self { + Self { + enabled: AtomicBool::new(enabled), + scan: Mutex::new(Reservoir::default()), + upload_per_mb: Mutex::new(Reservoir::default()), + } + } + + /// Whether capture is currently enabled. + #[must_use] + pub fn is_enabled(&self) -> bool { + self.enabled.load(Ordering::Relaxed) + } + + /// Flip the consent gate. Turning it OFF also DROPS any samples already + /// captured (SPEC s16: opting out must not leave latency data lingering); + /// turning it back ON starts from an empty window. + pub fn set_enabled(&self, enabled: bool) { + self.enabled.store(enabled, Ordering::Relaxed); + if !enabled { + self.lock_scan().clear(); + self.lock_upload().clear(); + } + } + + /// Record one per-file scan-processing duration (ms). No-op when disabled. + pub fn record_scan_ms(&self, ms: u64) { + if !self.is_enabled() { + return; + } + self.lock_scan().record(ms); + } + + /// Record one upload's latency normalized per MiB (ms). No-op when disabled. + /// The caller computes the per-MiB figure from the op's wall-clock duration + /// and its byte size (see [`per_mb_ms`]). + pub fn record_upload_per_mb_ms(&self, ms: u64) { + if !self.is_enabled() { + return; + } + self.lock_upload().record(ms); + } + + /// Read-only `[p50, p95]` for each metric (empty when no samples). Does NOT + /// reset - the caller resets via [`Self::reset`] only after a SUCCESSFUL + /// send, so a dropped ping re-uses the same window. + #[must_use] + pub fn snapshot(&self) -> LatencyPercentiles { + LatencyPercentiles { + scan: self.lock_scan().percentiles(), + upload_per_mb: self.lock_upload().percentiles(), + } + } + + /// Clear both reservoirs (called after a successful telemetry send so the + /// next reporting window starts fresh). + pub fn reset(&self) { + self.lock_scan().clear(); + self.lock_upload().clear(); + } + + fn lock_scan(&self) -> std::sync::MutexGuard<'_, Reservoir> { + self.scan + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + fn lock_upload(&self) -> std::sync::MutexGuard<'_, Reservoir> { + self.upload_per_mb + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +/// Normalize an upload op's wall-clock duration to milliseconds-per-MiB. +/// Returns `None` when `bytes == 0` (nothing to normalize against - a zero-byte +/// object's per-MB latency is meaningless). Uses `u128` intermediate arithmetic +/// so a large `elapsed_ms * BYTES_PER_MB` cannot overflow. +#[must_use] +pub fn per_mb_ms(elapsed_ms: u64, bytes: u64) -> Option { + if bytes == 0 { + return None; + } + let per_mb = u128::from(elapsed_ms) * u128::from(BYTES_PER_MB) / u128::from(bytes); + Some(per_mb.min(u128::from(u64::MAX)) as u64) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn percentiles_empty_is_empty() { + // No samples -> empty vec (the wire keeps emitting empty arrays). + assert!(percentiles_p50_p95(&[]).is_empty()); + } + + #[test] + fn percentiles_single_sample_repeats() { + // One sample: p50 == p95 == that sample. + assert_eq!(percentiles_p50_p95(&[42]), vec![42, 42]); + } + + #[test] + fn percentiles_odd_count() { + // n = 5 sorted [1,2,3,4,5]: p50 rank ceil(2.5)=3 -> idx2 -> 3; + // p95 rank ceil(4.75)=5 -> idx4 -> 5. + assert_eq!(percentiles_p50_p95(&[5, 3, 1, 4, 2]), vec![3, 5]); + } + + #[test] + fn percentiles_even_count() { + // n = 4 sorted [10,20,30,40]: p50 rank ceil(2.0)=2 -> idx1 -> 20; + // p95 rank ceil(3.8)=4 -> idx3 -> 40. + assert_eq!(percentiles_p50_p95(&[40, 10, 30, 20]), vec![20, 40]); + } + + #[test] + fn percentiles_large_uniform() { + // 1..=100: p50 -> 50, p95 -> 95 (nearest-rank on a dense range). + let samples: Vec = (1..=100).collect(); + assert_eq!(percentiles_p50_p95(&samples), vec![50, 95]); + } + + #[test] + fn record_no_op_when_disabled() { + // SPEC s16: no capture when telemetry is off. + let r = LatencyReservoir::new(false); + r.record_scan_ms(5); + r.record_upload_per_mb_ms(7); + let snap = r.snapshot(); + assert!(snap.scan.is_empty()); + assert!(snap.upload_per_mb.is_empty()); + } + + #[test] + fn record_and_snapshot_when_enabled() { + let r = LatencyReservoir::new(true); + for v in [10u64, 20, 30] { + r.record_scan_ms(v); + } + r.record_upload_per_mb_ms(100); + let snap = r.snapshot(); + // [10,20,30]: p50 rank ceil(1.5)=2 -> idx1 -> 20; p95 rank ceil(2.85)=3 -> idx2 -> 30. + assert_eq!(snap.scan, vec![20, 30]); + assert_eq!(snap.upload_per_mb, vec![100, 100]); + // A read-only snapshot does NOT drain: a second snapshot is identical. + assert_eq!(r.snapshot().scan, vec![20, 30]); + } + + #[test] + fn reset_clears_the_window() { + let r = LatencyReservoir::new(true); + r.record_scan_ms(1); + r.record_upload_per_mb_ms(2); + assert!(!r.snapshot().scan.is_empty()); + r.reset(); + let snap = r.snapshot(); + assert!(snap.scan.is_empty()); + assert!(snap.upload_per_mb.is_empty()); + } + + #[test] + fn disable_drops_captured_samples() { + // Turning capture off must drop anything already captured (consent). + let r = LatencyReservoir::new(true); + r.record_scan_ms(9); + assert!(!r.snapshot().scan.is_empty()); + r.set_enabled(false); + assert!(r.snapshot().scan.is_empty()); + // Re-enabling starts from an empty window. + r.set_enabled(true); + assert!(r.snapshot().scan.is_empty()); + r.record_scan_ms(3); + assert_eq!(r.snapshot().scan, vec![3, 3]); + } + + #[test] + fn ring_buffer_caps_at_capacity() { + // More than CAP samples: only the most-recent CAP are retained. Push + // CAP zeros then CAP hundreds; the window should be all hundreds. + let r = LatencyReservoir::new(true); + for _ in 0..RESERVOIR_CAP { + r.record_scan_ms(0); + } + for _ in 0..RESERVOIR_CAP { + r.record_scan_ms(100); + } + assert_eq!(r.snapshot().scan, vec![100, 100]); + } + + #[test] + fn per_mb_ms_normalizes_and_guards_zero() { + // 1 MiB in 200 ms -> 200 ms/MiB. + assert_eq!(per_mb_ms(200, BYTES_PER_MB), Some(200)); + // 2 MiB in 200 ms -> 100 ms/MiB. + assert_eq!(per_mb_ms(200, 2 * BYTES_PER_MB), Some(100)); + // Half a MiB in 50 ms -> 100 ms/MiB. + assert_eq!(per_mb_ms(50, BYTES_PER_MB / 2), Some(100)); + // Zero bytes -> no sample (nothing to normalize against). + assert_eq!(per_mb_ms(200, 0), None); + } +} diff --git a/src-tauri/src/app_state.rs b/src-tauri/src/app_state.rs index f3572195..8895728c 100644 --- a/src-tauri/src/app_state.rs +++ b/src-tauri/src/app_state.rs @@ -401,6 +401,14 @@ pub struct TelemetryRuntime { /// the cancel flag already set and aborts. A `tokio::Mutex` so it can be held /// across the awaited send. Shared via [`AppState::telemetry_send_gate`]. send_gate: Arc>, + /// DESIGN s13: the app-global latency sampler shared into every account's + /// executor + orchestrator (the SAME `Arc`), read at ping-build time for the + /// scan / upload-per-MB percentiles. Default-ON; boot replaces it via + /// [`AppState::install_telemetry_latency`] with one initialized from the + /// persisted `telemetry.enabled` pref BEFORE any capture, and the enable/ + /// disable toggle flips it in lockstep with the pref + /// (`telemetry::apply_enabled_change`). Shared via [`AppState::telemetry_latency`]. + latency: Arc, } /// M8 (P2-3): max number of TERMINAL restore-job records retained for late @@ -854,6 +862,27 @@ impl AppState { Arc::clone(&self.telemetry.send_gate) } + /// DESIGN s13: the app-global latency reservoir. The ping task snapshots it at + /// build time (and resets it after a successful send); the enable/disable + /// toggle flips its capture gate via `telemetry::apply_enabled_change`. Returns + /// a cloned `Arc` - the SAME instance every account's executor + orchestrator + /// records into. + #[must_use] + pub fn telemetry_latency(&self) -> Arc { + Arc::clone(&self.telemetry.latency) + } + + /// DESIGN s13: install the boot-built latency reservoir (initialized from the + /// persisted `telemetry.enabled` pref) so [`AppState`] shares the SAME `Arc` + /// the executors + orchestrators were wired with in assembly. Called once + /// before `.manage(..)`, replacing the default-ON placeholder. + pub fn install_telemetry_latency( + &mut self, + latency: Arc, + ) { + self.telemetry.latency = latency; + } + /// M9b: signal the periodic-ping task to stop and TAKE its handle so the /// caller can await it (the app-quit drain). Returns `None` if no task is /// tracked (never spawned / already drained). Mirrors `shutdown_updater_task`. diff --git a/src-tauri/src/assembly.rs b/src-tauri/src/assembly.rs index bc2b24db..a577c2d8 100644 --- a/src-tauri/src/assembly.rs +++ b/src-tauri/src/assembly.rs @@ -120,6 +120,17 @@ pub async fn build_and_spawn( // account. A picker-minted folder id is therefore visible to the uploader. let fake_remote_stores: FakeRemoteStores = Arc::new(std::sync::Mutex::new(HashMap::new())); + // DESIGN s13 telemetry: the app-global latency reservoir, created ONCE with + // its capture gate initialized from the persisted `telemetry.enabled` pref + // (BEFORE any executor / scanner is wired, so an opted-out user gets no + // startup capture window). The SAME `Arc` is threaded into every account's + // executor + orchestrator below and installed on the AppState so the ping + // task reads the samples both write. + let telemetry_enabled = crate::telemetry::read_enabled(state.as_ref()).await; + let latency = Arc::new(driven_core::telemetry::LatencyReservoir::new( + telemetry_enabled, + )); + let mut handles: HashMap = HashMap::new(); for account in &accounts { @@ -162,6 +173,7 @@ pub async fn build_and_spawn( use_fake, &fake_remote_stores, vss_helper.as_ref(), + &latency, ) .await { @@ -199,7 +211,11 @@ pub async fn build_and_spawn( } } - let app_state = AppState::new(state, handles, remote_mode, fake_remote_stores); + let mut app_state = AppState::new(state, handles, remote_mode, fake_remote_stores); + // DESIGN s13: share the SAME reservoir the executors + orchestrators record + // into with the AppState so the ping task snapshots those samples (replaces + // the default-ON placeholder). + app_state.install_telemetry_latency(latency); // Issue #25: install the broker manager so the quit sweep can shut it down // and `get_vss_helper_status` can report truthful liveness. if let Some(manager) = vss_helper { @@ -315,6 +331,10 @@ pub async fn spawn_account( // hot-added account's BrokeredVssProvider shares the one launch / pipe. let vss_helper = app_state.vss_helper_manager(); + // DESIGN s13: a hot-added account records into the SAME app-global latency + // reservoir the running AppState + ping task already share. + let latency = app_state.telemetry_latency(); + match build_account( app, &state, @@ -323,6 +343,7 @@ pub async fn spawn_account( use_fake, &fake_remote_stores, vss_helper.as_ref(), + &latency, ) .await? { @@ -380,6 +401,7 @@ enum RemoteOutcome { /// Build + spawn ONE account's orchestrator over the real seams. Returns a /// [`BuildOutcome`] (spawned, or needs-reauth) or an error that the caller /// logs + skips. +#[allow(clippy::too_many_arguments)] async fn build_account( app: &AppHandle, state: &Arc, @@ -388,6 +410,7 @@ async fn build_account( use_fake: bool, fake_remote_stores: &FakeRemoteStores, vss_helper: Option<&Arc>, + latency: &Arc, ) -> anyhow::Result { let clock: Arc = Arc::new(SystemClock); @@ -482,20 +505,25 @@ async fn build_account( let crypto_dyn: Arc = crypto.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(), - )); + 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()), + ); // --- orchestrator ------------------------------------------------------- // Held as the CONCRETE `Arc` (not `Arc`) @@ -524,6 +552,9 @@ async fn build_account( // Share the executor's pacer so the V2 metered throttle (DESIGN s17) can // lower / lift its bandwidth cap as the network goes on / off metered. orchestrator = orchestrator.with_pacer(pacer); + // 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()); 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/settings.rs b/src-tauri/src/commands/settings.rs index cc5be55f..cf6c0851 100644 --- a/src-tauri/src/commands/settings.rs +++ b/src-tauri/src/commands/settings.rs @@ -424,7 +424,9 @@ pub async fn update_settings( if let Some(v) = t.enabled { let cancel = state.telemetry_cancel(); let gate = state.telemetry_send_gate(); - crate::telemetry::apply_enabled_change(repo, &cancel, &gate, v).await?; + let latency = state.telemetry_latency(); + crate::telemetry::apply_enabled_change(repo, &cancel, &gate, latency.as_ref(), v) + .await?; } } diff --git a/src-tauri/src/telemetry.rs b/src-tauri/src/telemetry.rs index 87c77106..04467474 100644 --- a/src-tauri/src/telemetry.rs +++ b/src-tauri/src/telemetry.rs @@ -169,6 +169,21 @@ async fn read_prefs(state: &dyn StateRepo) -> CommandResult { }) } +/// Read the persisted `telemetry.enabled` pref (DEFAULT ON per SPEC s16). Public +/// so the assembly can initialize the app-global latency reservoir's capture gate +/// BEFORE any executor / scanner is wired, so a user who opted out gets no startup +/// capture window. A read error degrades to `true` (default ON) - the same +/// default the ping path applies. +pub async fn read_enabled(state: &dyn StateRepo) -> bool { + match read_prefs(state).await { + Ok(p) => p.enabled, + Err(e) => { + tracing::debug!(target: TARGET, error = %e, "telemetry: could not read enabled pref for reservoir init; defaulting ON"); + true + } + } +} + /// Persist `telemetry.last_sent_at` (Unix ms) after a SUCCESSFUL send. /// /// R3-P1-1 (CONSENT INTEGRITY): this is an ATOMIC, COMMUTING field-level patch @@ -478,6 +493,7 @@ fn build_payload( channel: String, os_version: Option, aggregate: driven_core::state::TelemetryAggregate, + latency: LatencyP50P95, ) -> TelemetryPayload { let errors_by_class = aggregate.errors_by_class.into_iter().collect(); TelemetryPayload { @@ -500,9 +516,20 @@ fn build_payload( // window or not), not a count. update_applied: aggregate.update_applied > 0, }, - // No per-op latency is recorded in durable state in V1; emit empty - // arrays (the keys are present) rather than fabricating percentiles. - latency_p50_p95_ms: LatencyP50P95::default(), + // DESIGN s13: the [p50, p95] scan + upload-per-MB latencies snapshotted + // from the app-global reservoir at build time (empty arrays when no + // samples were captured this window - the reservoir yields empty vecs, + // preserving the original wire shape). + latency_p50_p95_ms: latency, + } +} + +impl From for LatencyP50P95 { + fn from(p: driven_core::telemetry::LatencyPercentiles) -> Self { + LatencyP50P95 { + scan: p.scan, + upload_per_mb: p.upload_per_mb, + } } } @@ -590,6 +617,7 @@ async fn maybe_send_once( sink: &dyn TelemetrySink, cancel: Option<&std::sync::atomic::AtomicBool>, gate: Option<&tokio::sync::Mutex<()>>, + latency: Option<&driven_core::telemetry::LatencyReservoir>, ) -> bool { use std::sync::atomic::Ordering; @@ -629,7 +657,21 @@ async fn maybe_send_once( } }; let os_version = coarse_os_version(); - let payload = build_payload(install_id, now_ms, version, channel, os_version, aggregate); + // DESIGN s13: a READ-ONLY snapshot of the latency percentiles for this + // window. NOT reset here - only after a SUCCESSFUL send below, so a dropped + // or aborted ping re-uses the same window's samples on the next attempt + // (mirroring how the event-count aggregates re-send an un-checkpointed + // window keyed on `last_sent_at`). + let latency_pcts: LatencyP50P95 = latency.map(|r| r.snapshot().into()).unwrap_or_default(); + let payload = build_payload( + install_id, + now_ms, + version, + channel, + os_version, + aggregate, + latency_pcts, + ); // R3-P1-2 (SEND-ADMISSION GATE): acquire the shared gate, then do the final // cancel/pref re-check AND the network send WHILE HOLDING IT. The disable path @@ -679,6 +721,14 @@ async fn maybe_send_once( if let Err(e) = write_last_sent_at(state, now_ms).await { tracing::debug!(target: TARGET, error = %e, "telemetry: could not record last_sent_at"); } + // DESIGN s13: reset the latency window ONLY after a successful send, + // so the next ping's percentiles cover only new samples. A dropped + // send leaves the reservoir intact to re-send next tick. (A handful + // of samples captured between the snapshot above and this reset are + // dropped - an acceptable best-effort loss for a latency signal.) + if let Some(r) = latency { + r.reset(); + } } Err(e) => { tracing::info!(target: TARGET, error = %e, "telemetry ping failed (best-effort, ignored)"); @@ -752,6 +802,7 @@ async fn ping_once(app: &AppHandle, sink: &dyn TelemetrySink) { // disable path. let cancel = state.telemetry_cancel(); let gate = state.telemetry_send_gate(); + let latency = state.telemetry_latency(); let _ = maybe_send_once( state.state().as_ref(), version, @@ -759,6 +810,7 @@ async fn ping_once(app: &AppHandle, sink: &dyn TelemetrySink) { sink, Some(&cancel), Some(&gate), + Some(latency.as_ref()), ) .await; } @@ -798,12 +850,18 @@ pub async fn apply_enabled_change( state: &dyn StateRepo, cancel: &std::sync::atomic::AtomicBool, gate: &tokio::sync::Mutex<()>, + latency: &driven_core::telemetry::LatencyReservoir, enabled: bool, ) -> CommandResult<()> { use std::sync::atomic::Ordering; // P1-2: flip the cancel flag BEFORE anything else when disabling, so an // in-flight ping's under-gate re-check observes the cancellation. cancel.store(!enabled, Ordering::SeqCst); + // DESIGN s13 / SPEC s16 (consent): flip the latency-capture gate in lockstep. + // Turning telemetry OFF stops new samples AND drops any already captured, so + // opting out never leaves latency data lingering; turning it back ON starts a + // fresh window. + latency.set_enabled(enabled); // R3-P1-2: coordinate with the send-admission gate. Acquiring it serializes // this disable against an in-flight send's admission+send section: we cannot // proceed until any in-progress send releases the gate, and by then cancel is @@ -829,7 +887,15 @@ pub async fn set_telemetry_enabled( ) -> CommandResult { let cancel = state.telemetry_cancel(); let gate = state.telemetry_send_gate(); - apply_enabled_change(state.state().as_ref(), &cancel, &gate, enabled).await?; + let latency = state.telemetry_latency(); + apply_enabled_change( + state.state().as_ref(), + &cancel, + &gate, + latency.as_ref(), + enabled, + ) + .await?; Ok(enabled) } @@ -922,6 +988,7 @@ mod tests { "dev".to_string(), Some("11.26200".to_string()), aggregate, + LatencyP50P95::default(), ); assert_eq!(p.install_id, "00000000-0000-4000-8000-000000000000"); assert_eq!(p.version, "0.1.0"); @@ -984,6 +1051,7 @@ mod tests { "stable".to_string(), None, driven_core::state::TelemetryAggregate::default(), + LatencyP50P95::default(), ); let json = serde_json::to_value(&p).unwrap(); let obj = json.as_object().unwrap(); @@ -1068,6 +1136,7 @@ mod tests { &sink, None, None, + None, ) .await; @@ -1094,6 +1163,7 @@ mod tests { &sink, None, None, + None, ) .await; @@ -1121,6 +1191,7 @@ mod tests { &sink, None, None, + None, ) .await; assert!( @@ -1168,6 +1239,7 @@ mod tests { &sink, Some(&cancel), None, + None, ) .await; assert!(!attempted, "a cancelled ping is not attempted"); @@ -1199,6 +1271,7 @@ mod tests { &sink, None, None, + None, ) .await; assert!(!attempted, "disabled pref aborts the send"); @@ -1226,9 +1299,15 @@ mod tests { // Step 2: the disable lands (the user opted out) while a ping is mid-flight. let cancel = std::sync::atomic::AtomicBool::new(false); let gate = tokio::sync::Mutex::new(()); - apply_enabled_change(&repo, &cancel, &gate, false) - .await - .unwrap(); + apply_enabled_change( + &repo, + &cancel, + &gate, + &driven_core::telemetry::LatencyReservoir::new(true), + false, + ) + .await + .unwrap(); // Step 3: the (already-in-flight) ping commits its delta checkpoint. This is // the write that, under the old RMW, would resurrect enabled=true. @@ -1289,6 +1368,7 @@ mod tests { sink_c.as_ref(), Some(cancel_c.as_ref()), Some(gate_c.as_ref()), + None, ) .await }); @@ -1387,7 +1467,7 @@ mod tests { let sink = RecordingSink::default(); // First ping at `now` sends the 1 upload, records last_sent_at = now. - assert!(maybe_send_once(&repo, "0.1.0".to_string(), now, &sink, None, None).await); + assert!(maybe_send_once(&repo, "0.1.0".to_string(), now, &sink, None, None, None).await); { let sent = sink.sent.lock().unwrap_or_else(|e| e.into_inner()); assert_eq!(sent.len(), 1); @@ -1402,7 +1482,7 @@ mod tests { // Second ping at the SAME instant (a restart) - the delta window // (last_sent, now] is empty -> 0 uploads (NOT 1 again). - assert!(maybe_send_once(&repo, "0.1.0".to_string(), now, &sink, None, None).await); + assert!(maybe_send_once(&repo, "0.1.0".to_string(), now, &sink, None, None, None).await); { let sent = sink.sent.lock().unwrap_or_else(|e| e.into_inner()); assert_eq!(sent.len(), 2); @@ -1458,9 +1538,15 @@ mod tests { let cancel = std::sync::atomic::AtomicBool::new(false); let gate = tokio::sync::Mutex::new(()); - apply_enabled_change(&repo, &cancel, &gate, false) - .await - .unwrap(); + apply_enabled_change( + &repo, + &cancel, + &gate, + &driven_core::telemetry::LatencyReservoir::new(true), + false, + ) + .await + .unwrap(); assert!( cancel.load(Ordering::SeqCst), "disabling trips the cancel flag (in-flight ping aborts)" @@ -1471,9 +1557,15 @@ mod tests { ); // Re-enabling re-arms (clears) the flag. - apply_enabled_change(&repo, &cancel, &gate, true) - .await - .unwrap(); + apply_enabled_change( + &repo, + &cancel, + &gate, + &driven_core::telemetry::LatencyReservoir::new(true), + true, + ) + .await + .unwrap(); assert!( !cancel.load(Ordering::SeqCst), "enabling clears the cancel flag" @@ -1498,12 +1590,24 @@ mod tests { // Toggle enabled off then on via the shared path. let cancel = std::sync::atomic::AtomicBool::new(false); let gate = tokio::sync::Mutex::new(()); - apply_enabled_change(&repo, &cancel, &gate, false) - .await - .unwrap(); - apply_enabled_change(&repo, &cancel, &gate, true) - .await - .unwrap(); + apply_enabled_change( + &repo, + &cancel, + &gate, + &driven_core::telemetry::LatencyReservoir::new(true), + false, + ) + .await + .unwrap(); + apply_enabled_change( + &repo, + &cancel, + &gate, + &driven_core::telemetry::LatencyReservoir::new(true), + true, + ) + .await + .unwrap(); assert_eq!( read_prefs(&repo).await.unwrap().last_sent_at, @@ -1567,4 +1671,151 @@ mod tests { ); cleanup(dir); } + + #[test] + fn build_payload_carries_the_captured_latency_percentiles() { + // DESIGN s13: the drained [p50, p95] pairs flow into the wire payload + // (replacing the old hardcoded empty default). + let latency = LatencyP50P95 { + scan: vec![3, 12], + upload_per_mb: vec![40, 110], + }; + let p = build_payload( + "00000000-0000-4000-8000-000000000000".to_string(), + 1_700_000_000_000, + "0.1.0".to_string(), + "stable".to_string(), + None, + driven_core::state::TelemetryAggregate::default(), + latency, + ); + assert_eq!(p.latency_p50_p95_ms.scan, vec![3, 12]); + assert_eq!(p.latency_p50_p95_ms.upload_per_mb, vec![40, 110]); + } + + #[tokio::test] + async fn successful_send_snapshots_then_resets_the_latency_window() { + // DESIGN s13: an enabled ping snapshots the reservoir into the payload + // and, on a SUCCESSFUL send, resets it so the next window starts fresh. + let (repo, dir) = temp_repo().await; + let reservoir = driven_core::telemetry::LatencyReservoir::new(true); + reservoir.record_scan_ms(10); + reservoir.record_scan_ms(20); + reservoir.record_upload_per_mb_ms(100); + + let sink = RecordingSink::default(); + let attempted = maybe_send_once( + &repo, + "0.1.0".to_string(), + 1_700_000_000_000, + &sink, + None, + None, + Some(&reservoir), + ) + .await; + assert!(attempted); + // The ping carried the captured percentiles ([10,20]: p50 rank 1 -> 10, + // p95 rank 2 -> 20; the single upload sample repeats). + let sent = sink.sent.lock().unwrap_or_else(|e| e.into_inner()); + assert_eq!(sent[0].latency_p50_p95_ms.scan, vec![10, 20]); + assert_eq!(sent[0].latency_p50_p95_ms.upload_per_mb, vec![100, 100]); + drop(sent); + // ...and the reservoir was reset after the successful send. + let snap = reservoir.snapshot(); + assert!(snap.scan.is_empty(), "window reset after a successful send"); + assert!(snap.upload_per_mb.is_empty()); + cleanup(dir); + } + + #[tokio::test] + async fn failed_send_keeps_the_latency_window_for_retry() { + // DESIGN s13: a dropped send must NOT reset the reservoir, so the same + // samples re-send on the next tick (mirroring the event-count delta reuse). + let (repo, dir) = temp_repo().await; + let reservoir = driven_core::telemetry::LatencyReservoir::new(true); + reservoir.record_scan_ms(7); + + let sink = FailingSink; + let attempted = maybe_send_once( + &repo, + "0.1.0".to_string(), + 1_700_000_000_000, + &sink, + None, + None, + Some(&reservoir), + ) + .await; + assert!(attempted, "a failed send is still an attempt"); + assert_eq!( + reservoir.snapshot().scan, + vec![7, 7], + "a failed send leaves the window intact for the next tick" + ); + cleanup(dir); + } + + #[tokio::test] + async fn disabled_ping_does_not_touch_the_reservoir() { + // SPEC s16: a disabled ping makes no send AND does not reset the window + // (nothing is captured while off anyway, but the guard must hold). + let (repo, dir) = temp_repo().await; + write_enabled(&repo, false).await.unwrap(); + // A reservoir that (defensively) still holds a stray sample. + let reservoir = driven_core::telemetry::LatencyReservoir::new(true); + reservoir.record_scan_ms(5); + + let sink = RecordingSink::default(); + let attempted = maybe_send_once( + &repo, + "0.1.0".to_string(), + 1_700_000_000_000, + &sink, + None, + None, + Some(&reservoir), + ) + .await; + assert!(!attempted, "disabled telemetry sends nothing"); + assert_eq!( + sink.calls.load(Ordering::SeqCst), + 0, + "no network call when disabled" + ); + assert_eq!( + reservoir.snapshot().scan, + vec![5, 5], + "a disabled ping does not reset the reservoir" + ); + cleanup(dir); + } + + #[tokio::test] + async fn apply_enabled_change_off_clears_the_reservoir() { + // DESIGN s13 / SPEC s16 (consent): disabling telemetry through the shared + // path drops any captured latency samples; re-enabling starts fresh. + let (repo, dir) = temp_repo().await; + let cancel = std::sync::atomic::AtomicBool::new(false); + let gate = tokio::sync::Mutex::new(()); + let reservoir = driven_core::telemetry::LatencyReservoir::new(true); + reservoir.record_scan_ms(9); + reservoir.record_upload_per_mb_ms(50); + assert!(!reservoir.snapshot().scan.is_empty()); + + apply_enabled_change(&repo, &cancel, &gate, &reservoir, false) + .await + .unwrap(); + assert!(!reservoir.is_enabled(), "capture gate flipped off"); + let snap = reservoir.snapshot(); + assert!(snap.scan.is_empty(), "disable drops captured samples"); + assert!(snap.upload_per_mb.is_empty()); + + // Re-enable re-arms capture. + apply_enabled_change(&repo, &cancel, &gate, &reservoir, true) + .await + .unwrap(); + assert!(reservoir.is_enabled(), "capture gate flipped back on"); + cleanup(dir); + } } diff --git a/telemetry-worker/README.md b/telemetry-worker/README.md new file mode 100644 index 00000000..639d79ca --- /dev/null +++ b/telemetry-worker/README.md @@ -0,0 +1,109 @@ +# Driven telemetry Worker + +The server side of Driven's opt-out anonymous telemetry (DESIGN s13, SPEC s16). A +small Cloudflare Worker that: + +- **Ingests** the usage ping the desktop client POSTs on startup + every 24h + (`POST /telemetry/v1/ping`), validates it strictly (public endpoint - never + trust the client), and writes it to an Analytics Engine dataset. +- **Serves a gated latency rollup** (`GET /telemetry/v1/stats/latency`) reading + the per-day scan / upload-per-MB percentiles back out via the Analytics Engine + SQL API. + +It is deployed to `driven.maxhogan.dev/telemetry/*` (the Worker route takes +precedence over the CF Pages site for that prefix) and auto-deploys via +`.github/workflows/deploy-telemetry.yml` on any change under `telemetry-worker/`. + +## Toolchain + +Its own toolchain, NOT part of the cargo workspace or the `ui/` build. + +```sh +pnpm install +pnpm run typecheck # tsc --noEmit +pnpm run lint # eslint +pnpm test # vitest (handler unit tests, mocked AE + fetch) +pnpm run deploy # wrangler deploy (CI does this on merge) +``` + +## Analytics Engine dataset layout (`driven_telemetry`) + +One data point per ping (`writePing`): + +| column | value | +|---|---| +| `index1` | `install_id` (anonymous UUID v4 sampling key) | +| `blob1..6` | `os`, `arch`, `channel`, `version`, `os_version` (`""` if absent), `errors_by_class` JSON | +| `double1..6` | `files_uploaded`, `bytes_uploaded`, `deep_verify_runs`, `update_applied` (0/1), `total_errors`, `ts` (epoch ms) | +| `double7..10` | `scan_p50`, `scan_p95`, `upload_per_mb_p50`, `upload_per_mb_p95` (ms) | + +The 4 latency doubles (DESIGN s13) are **appended** so the original columns keep +their positions. When the client had no samples for a metric this window (its +array is empty), the pair is written as the sentinel **`-1`** so the rollup query +can tell "no samples" apart from a legitimate `0 ms` (a sub-millisecond per-file +scan rounds to 0). + +## `GET /telemetry/v1/stats/latency` + +Per-day aggregates of the client-reported percentiles. **Authenticated** - it +exposes aggregate telemetry, so it is never served open. + +- **Auth:** `Authorization: Bearer `. A missing/wrong token is `401`. +- **Query:** `?days=N` - lookback window, default `7`, clamped to `[1, 90]`. +- **Contract:** + +``` +GET /telemetry/v1/stats/latency?days=7 +Authorization: Bearer + +200 OK +{ + "days": 7, + "metrics": { + "scan": [ { "day": "2026-07-14", "avg_p50_ms": 3, "avg_p95_ms": 12, "max_p95_ms": 40, "samples": 9 } ], + "upload_per_mb": [ { "day": "2026-07-14", "avg_p50_ms": 50, "avg_p95_ms": 120, "max_p95_ms": 300, "samples": 4 } ] + } +} +``` + +Per metric, per UTC day: `avg_p50_ms` (mean of the pinged p50s), `avg_p95_ms` +(mean of the pinged p95s), `max_p95_ms` (worst pinged p95), and `samples` (number +of pings that reported the metric). Empty-latency pings (the `-1` sentinel) are +excluded per metric via `WHERE >= 0`, so a real `0 ms` still counts. The +two metrics are queried separately (each filters its own sentinel column) via the +Analytics Engine SQL API. + +Status codes: `200` success; `401` missing/wrong bearer; `405` non-GET; +`502` upstream AE SQL query failed; `503` the endpoint is not configured +(a required secret is missing - see below). + +### Note on the documented path + +The task refers to this as `GET /stats/latency`. The Worker route only serves the +`/telemetry/*` prefix (`wrangler.jsonc`), so the real path is +`/telemetry/v1/stats/latency`. + +## Required secrets / vars (set post-deploy) + +The ingest path needs none of these; the `/stats/latency` READ path needs all +three. Until they are set, the endpoint returns `503 stats_not_configured` (it +never falls through to an unauthenticated or broken read). AE **reads** go through +the SQL HTTP API (the write binding cannot read), which needs a Cloudflare API +token - hence `CF_API_TOKEN`. + +| name | kind | purpose | +|---|---|---| +| `QUERY_TOKEN` | secret | Bearer token gating `/stats/latency`. Generate a random value. | +| `CF_API_TOKEN` | secret | Cloudflare API token with **Account Analytics: Read** on the Driven account, used to call the AE SQL API. | +| `CF_ACCOUNT_ID` | var (optional) | Account id for the SQL API URL. Defaults to the Driven account (`9c20c14daa20466a2d761a47162f719a`) when unset. | + +```sh +npx wrangler secret put QUERY_TOKEN +npx wrangler secret put CF_API_TOKEN +# optional (defaults to the Driven account): +npx wrangler secret put CF_ACCOUNT_ID # or set as a [vars] entry +``` + +> The `/stats/latency` endpoint's SQL was validated in unit tests against a mocked +> `fetch`; the live query (day-grouping function, response shape) should be +> smoke-checked against the real Analytics Engine SQL API once the secrets are set. diff --git a/telemetry-worker/src/index.ts b/telemetry-worker/src/index.ts index dd09672f..dbb4206b 100644 --- a/telemetry-worker/src/index.ts +++ b/telemetry-worker/src/index.ts @@ -20,15 +20,47 @@ // against a mocked AE binding) now; the live deploy + e2e telemetry validation // happen at M10. See design/CODEX_NOTES.md "## M9b - telemetry". -/// The Worker environment bindings (wrangler.jsonc). `TELEMETRY` is the Analytics -/// Engine dataset the validated ping is written to. +/// The Worker environment bindings (wrangler.jsonc + secrets). `TELEMETRY` is the +/// Analytics Engine dataset the validated ping is WRITTEN to (the binding). +/// +/// The `/stats/latency` rollup READ path (DESIGN s13) cannot read the AE dataset +/// through the write binding - AE reads go through the SQL HTTP API - so it needs: +/// - `QUERY_TOKEN`: the bearer secret gating the endpoint (it exposes aggregate +/// telemetry, so it is NEVER served unauthenticated). Unset => the endpoint 503s. +/// - `CF_API_TOKEN`: a Cloudflare API token with Account Analytics read, used to +/// call the AE SQL API. Unset => the endpoint 503s. +/// - `CF_ACCOUNT_ID`: the account id for the SQL API URL; defaults to the Driven +/// account when unset. +/// All three are wrangler secrets/vars set post-deploy (see the worker README); +/// they are optional in the type so the ingest path keeps working without them. export interface Env { TELEMETRY: AnalyticsEngineDataset; + QUERY_TOKEN?: string; + CF_API_TOKEN?: string; + CF_ACCOUNT_ID?: string; } -/// The only path this Worker serves (SPEC s16). Anything else is 404. +/// The ingest path (SPEC s16): POST a ping here. const PING_PATH = "/telemetry/v1/ping"; +/// The latency-rollup read path (DESIGN s13): GET per-day aggregates here. Scoped +/// under `/telemetry/*` because that is the only prefix the Worker route serves +/// (wrangler.jsonc); the documented `GET /stats/latency` maps here. +const STATS_LATENCY_PATH = "/telemetry/v1/stats/latency"; + +/// The Analytics Engine dataset name (matches wrangler.jsonc `dataset`). Used in +/// the SQL API `FROM` clause for the rollup query. +const DATASET = "driven_telemetry"; + +/// The Driven Cloudflare account id (wrangler.jsonc `account_id`), the default +/// target for the AE SQL API when `CF_ACCOUNT_ID` is not set. +const DEFAULT_ACCOUNT_ID = "9c20c14daa20466a2d761a47162f719a"; + +/// `/stats/latency` `days` window: default and hard cap (SPEC/DESIGN s13). The +/// query looks back `days` days; the value is clamped to `[1, 90]`. +const DEFAULT_STATS_DAYS = 7; +const MAX_STATS_DAYS = 90; + /// Max accepted request body size (bytes). The real ping is well under 4 KB; this /// cap rejects a hostile / malformed oversized body before parsing (SPEC s16 /// "cap body size"). @@ -341,14 +373,33 @@ function totalErrors(errors: Record): number { return sum; } -/// Write one validated ping to Analytics Engine (SPEC s16). The dataset schema: +/// The value written to a latency percentile double when the client reported NO +/// samples for that metric this window (its array is empty). A NEGATIVE sentinel +/// so the rollup query can distinguish "no samples" (`< 0`) from a LEGIT `0 ms` +/// (a sub-millisecond per-file scan rounds to 0) - the query filters `>= 0`. +const NO_LATENCY = -1; + +/// Extract `[p50, p95]` from a client latency array, mapping an empty (or +/// malformed short) array to the [`NO_LATENCY`] sentinel pair (DESIGN s13). +function latencyPair(arr: number[]): [number, number] { + if (arr.length >= 2) return [arr[0], arr[1]]; + return [NO_LATENCY, NO_LATENCY]; +} + +/// Write one validated ping to Analytics Engine (SPEC s16, DESIGN s13). Schema: /// - indexes: [install_id] (the sampling/grouping key - anonymous) /// - blobs: [os, arch, channel, version, os_version, errors_by_class JSON] /// (low-card dims; os_version is "" when the client did not send one) /// - doubles: [files_uploaded, bytes_uploaded, deep_verify_runs, update_applied, -/// total_errors, ts] (the numeric measures; update_applied is 0/1) +/// total_errors, ts, // double1..double6 +/// scan_p50, scan_p95, upload_per_mb_p50, upload_per_mb_p95] +/// // double7..double10 +/// The 4 latency doubles are appended (never reordered) so existing columns keep +/// their positions; an absent metric writes the NO_LATENCY (-1) sentinel. /// Writes are non-blocking (no await / waitUntil needed per the CF docs). export function writePing(env: Env, p: PingPayload): void { + const [scanP50, scanP95] = latencyPair(p.latency_p50_p95_ms.scan); + const [upP50, upP95] = latencyPair(p.latency_p50_p95_ms.upload_per_mb); env.TELEMETRY.writeDataPoint({ indexes: [p.install_id], blobs: [ @@ -370,6 +421,12 @@ export function writePing(env: Env, p: PingPayload): void { p.events_24h.update_applied ? 1 : 0, totalErrors(p.events_24h.errors_by_class), p.ts, + // DESIGN s13: the client-reported [p50, p95] latency percentiles (ms), or + // the NO_LATENCY sentinel when the client had no samples this window. + scanP50, + scanP95, + upP50, + upP95, ], }); } @@ -441,17 +498,162 @@ async function readBodyCapped( return { ok: true, text }; } -/// The Worker request handler (SPEC s16). Pure-ish: takes `request` + `env`, so a -/// unit test drives it with a mocked AE binding (no live runtime, no network). +// -------------------------------------------------------------------------- +// LATENCY ROLLUP (DESIGN s13): a gated, read-only per-day aggregate of the +// client-reported scan / upload-per-MB percentiles, over the Analytics Engine +// SQL API. NEVER served unauthenticated (it exposes aggregate telemetry). +// -------------------------------------------------------------------------- + +/// One metric's per-day aggregate row returned by `GET /stats/latency`. +interface LatencyDayRow { + /// The UTC day, `YYYY-MM-DD`. + day: string; + /// Mean of the pinged p50s that day (ms). + avg_p50_ms: number; + /// Mean of the pinged p95s that day (ms). + avg_p95_ms: number; + /// Worst pinged p95 that day (ms). + max_p95_ms: number; + /// Number of pings that reported this metric that day. + samples: number; +} + +/// Constant-time-ish string equality (avoids leaking the token via early-exit +/// timing on a per-char compare). Length difference is not hidden - fine for a +/// random bearer token. +function safeEqual(a: string, b: string): boolean { + if (a.length !== b.length) return false; + let mismatch = 0; + for (let i = 0; i < a.length; i++) { + mismatch |= a.charCodeAt(i) ^ b.charCodeAt(i); + } + return mismatch === 0; +} + +/// Clamp the `days` query param to a validated integer in `[1, MAX_STATS_DAYS]`, +/// defaulting to [`DEFAULT_STATS_DAYS`] when absent / non-numeric. Interpolated +/// into the SQL string, so it MUST be a bounded integer (there are no bind params +/// on the AE SQL API). +function clampStatsDays(raw: string | null): number { + // Absent / blank -> default (note `Number(null)` and `Number("")` are 0, NOT + // NaN, so these must be handled before the numeric parse). + if (raw === null || raw.trim() === "") return DEFAULT_STATS_DAYS; + const n = Number(raw); + if (!Number.isFinite(n)) return DEFAULT_STATS_DAYS; + const i = Math.trunc(n); + if (i < 1) return 1; + if (i > MAX_STATS_DAYS) return MAX_STATS_DAYS; + return i; +} + +/// Query one latency metric's per-day aggregates over the AE SQL API. `p50Col` / +/// `p95Col` are the AE double column names for this metric (e.g. `double7` / +/// `double8`). Rows carrying the NO_LATENCY sentinel (`< 0`, an empty-latency +/// ping) are excluded via `WHERE p50 >= 0`, so a legit `0 ms` still counts. The +/// response is the CF `{ meta, data }` JSON (NOT ndjson); each `data[]` row's +/// numeric columns arrive as strings, so they are coerced with `Number`. +async function queryLatencyMetric( + env: Env, + accountId: string, + days: number, + p50Col: string, + p95Col: string, +): Promise { + // `days` is a validated integer (clampStatsDays) and the column names are + // internal constants, so this interpolation carries no injection surface. + const sql = + `SELECT toDate(timestamp) AS day, ` + + `AVG(${p50Col}) AS avg_p50, ` + + `AVG(${p95Col}) AS avg_p95, ` + + `MAX(${p95Col}) AS max_p95, ` + + `COUNT() AS samples ` + + `FROM ${DATASET} ` + + `WHERE timestamp > NOW() - INTERVAL '${days}' DAY AND ${p50Col} >= 0 ` + + `GROUP BY day ORDER BY day`; + + const resp = await fetch( + `https://api.cloudflare.com/client/v4/accounts/${accountId}/analytics_engine/sql`, + { + method: "POST", + headers: { Authorization: `Bearer ${env.CF_API_TOKEN}` }, + body: sql, + }, + ); + if (!resp.ok) { + throw new Error(`analytics_engine sql query failed: ${resp.status}`); + } + const body = (await resp.json()) as { data?: Array> }; + const rows = Array.isArray(body.data) ? body.data : []; + return rows.map((r) => ({ + day: String(r.day), + avg_p50_ms: Number(r.avg_p50), + avg_p95_ms: Number(r.avg_p95), + max_p95_ms: Number(r.max_p95), + samples: Number(r.samples), + })); +} + +/// Handle `GET /stats/latency?days=N` (DESIGN s13). AUTH: a `Bearer QUERY_TOKEN` +/// header. Misconfiguration (no `QUERY_TOKEN` or no `CF_API_TOKEN`) is a 503 - +/// the endpoint is NEVER served open. Returns per-day aggregates for both metrics. +export async function handleStatsLatency(request: Request, env: Env): Promise { + // Misconfigured => 503 (never fall through to an unauthenticated / broken read). + if (!env.QUERY_TOKEN || !env.CF_API_TOKEN) { + return json(503, { error: "stats_not_configured" }); + } + // Bearer auth against the QUERY_TOKEN secret. + const auth = request.headers.get("authorization"); + if (!auth || !safeEqual(auth, `Bearer ${env.QUERY_TOKEN}`)) { + return new Response(JSON.stringify({ error: "unauthorized" }), { + status: 401, + headers: { "content-type": "application/json", "www-authenticate": "Bearer" }, + }); + } + + const url = new URL(request.url); + const days = clampStatsDays(url.searchParams.get("days")); + const accountId = env.CF_ACCOUNT_ID ?? DEFAULT_ACCOUNT_ID; + + try { + // Two queries (one per metric) so each filters its OWN sentinel column - a + // shared WHERE could not exclude an empty-scan ping without also dropping its + // (present) upload sample, and vice versa. Runs them concurrently. + const [scan, uploadPerMb] = await Promise.all([ + queryLatencyMetric(env, accountId, days, "double7", "double8"), + queryLatencyMetric(env, accountId, days, "double9", "double10"), + ]); + return json(200, { days, metrics: { scan, upload_per_mb: uploadPerMb } }); + } catch { + // Never echo the query / token; a generic upstream-failure signal. + console.error("telemetry: stats/latency AE query failed"); + return json(502, { error: "query_failed" }); + } +} + +/// The Worker request handler (SPEC s16, DESIGN s13). Pure-ish: takes `request` + +/// `env`, so a unit test drives it with a mocked AE binding + mocked `fetch` (no +/// live runtime, no network). /// /// Contract: /// - POST /telemetry/v1/ping with a valid JSON body -> write to AE, 204. /// - POST /telemetry/v1/ping with a malformed body / oversized body -> 400. +/// - GET /telemetry/v1/stats/latency (Bearer QUERY_TOKEN) -> per-day rollup, 200. /// - the right path but the wrong method -> 405. /// - any other path -> 404. export async function handle(request: Request, env: Env): Promise { const url = new URL(request.url); + // DESIGN s13: the gated latency-rollup read path. GET only (405 otherwise). + if (url.pathname === STATS_LATENCY_PATH) { + if (request.method !== "GET") { + return new Response(JSON.stringify({ error: "method_not_allowed" }), { + status: 405, + headers: { "content-type": "application/json", allow: "GET" }, + }); + } + return handleStatsLatency(request, env); + } + // Only the ping path exists; everything else is 404 (the CF Pages site serves // the root + /updates; this Worker owns only /telemetry/*). if (url.pathname !== PING_PATH) { diff --git a/telemetry-worker/test/handler.test.ts b/telemetry-worker/test/handler.test.ts index c0fc747e..12ab9e84 100644 --- a/telemetry-worker/test/handler.test.ts +++ b/telemetry-worker/test/handler.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, vi } from "vitest"; +import { describe, it, expect, vi, afterEach } from "vitest"; import { handle, validatePing, writePing, type Env } from "../src/index"; @@ -560,3 +560,189 @@ describe("telemetry worker handler", () => { if (!r.ok) expect(r.reason).toBe("ts"); }); }); + +// -------------------------------------------------------------------------- +// DESIGN s13: latency percentile doubles + the gated /stats/latency rollup. +// -------------------------------------------------------------------------- + +/// The AE double indices (0-based) the 4 latency percentiles occupy - appended +/// after the original 6 measures (files, bytes, deep_verify, update_applied, +/// total_errors, ts). +const SCAN_P50 = 6; +const SCAN_P95 = 7; +const UP_P50 = 8; +const UP_P95 = 9; + +describe("telemetry worker latency doubles (DESIGN s13)", () => { + it("writes the client-reported [p50, p95] latency doubles", async () => { + const { env, writes } = mockEnv(); + const p = validPayload(); + p.latency_p50_p95_ms = { scan: [3, 12], upload_per_mb: [40, 110] }; + const res = await handle(postPing(JSON.stringify(p)), env); + expect(res.status).toBe(204); + const dp = writes[0] as { doubles: number[] }; + // The original 6 measures keep their positions. + expect(dp.doubles[5]).toBe(1_700_000_000_000); // ts unchanged at double6 + expect(dp.doubles[SCAN_P50]).toBe(3); + expect(dp.doubles[SCAN_P95]).toBe(12); + expect(dp.doubles[UP_P50]).toBe(40); + expect(dp.doubles[UP_P95]).toBe(110); + }); + + it("writes the -1 sentinel for a metric with no samples (empty array)", async () => { + const { env, writes } = mockEnv(); + const p = validPayload(); + // Scan has data; upload had no completed uploads this window (empty array). + p.latency_p50_p95_ms = { scan: [0, 5], upload_per_mb: [] }; + const res = await handle(postPing(JSON.stringify(p)), env); + expect(res.status).toBe(204); + const dp = writes[0] as { doubles: number[] }; + // A legit 0 ms p50 is preserved (NOT turned into the sentinel). + expect(dp.doubles[SCAN_P50]).toBe(0); + expect(dp.doubles[SCAN_P95]).toBe(5); + // The empty upload metric -> -1 sentinel (distinguishable from a real 0). + expect(dp.doubles[UP_P50]).toBe(-1); + expect(dp.doubles[UP_P95]).toBe(-1); + }); + + it("writes both sentinels when V1-style empty latency arrays arrive", async () => { + const { env, writes } = mockEnv(); + // The default validPayload() carries empty latency arrays (the V1 wire shape). + const res = await handle(postPing(JSON.stringify(validPayload())), env); + expect(res.status).toBe(204); + const dp = writes[0] as { doubles: number[] }; + expect(dp.doubles[SCAN_P50]).toBe(-1); + expect(dp.doubles[SCAN_P95]).toBe(-1); + expect(dp.doubles[UP_P50]).toBe(-1); + expect(dp.doubles[UP_P95]).toBe(-1); + }); +}); + +describe("telemetry worker GET /stats/latency (DESIGN s13)", () => { + const STATS_URL = "https://driven.maxhogan.dev/telemetry/v1/stats/latency"; + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + /// A configured stats Env (both secrets present) plus a fetch stub returning + /// the given per-metric AE `{ data }` rows for the two metric queries in order. + function statsEnv(): Env { + return { + TELEMETRY: { writeDataPoint: () => undefined }, + QUERY_TOKEN: "s3cret", + CF_API_TOKEN: "cf-token", + CF_ACCOUNT_ID: "acct-123", + } as unknown as Env; + } + + function authed(): Request { + return new Request(STATS_URL, { + method: "GET", + headers: { authorization: "Bearer s3cret" }, + }); + } + + it("503s when QUERY_TOKEN / CF_API_TOKEN are not configured", async () => { + const env = { TELEMETRY: { writeDataPoint: () => undefined } } as unknown as Env; + const res = await handle(authed(), env); + expect(res.status).toBe(503); + expect(await res.json()).toMatchObject({ error: "stats_not_configured" }); + }); + + it("401s without a bearer token", async () => { + const res = await handle(new Request(STATS_URL, { method: "GET" }), statsEnv()); + expect(res.status).toBe(401); + expect(res.headers.get("www-authenticate")).toBe("Bearer"); + }); + + it("401s with the wrong bearer token", async () => { + const req = new Request(STATS_URL, { + method: "GET", + headers: { authorization: "Bearer nope" }, + }); + const res = await handle(req, statsEnv()); + expect(res.status).toBe(401); + }); + + it("405s on the wrong method for the stats path", async () => { + const res = await handle(new Request(STATS_URL, { method: "POST" }), statsEnv()); + expect(res.status).toBe(405); + expect(res.headers.get("allow")).toBe("GET"); + }); + + it("returns per-day aggregates for both metrics on a valid authed request", async () => { + const sqls: string[] = []; + const fetchMock = vi.fn(async (_url: string, init: { body: string }) => { + sqls.push(init.body); + // First call (scan) then second (upload); return distinct rows. + const isScan = init.body.includes("double7"); + const data = isScan + ? [{ day: "2026-07-14", avg_p50: "3", avg_p95: "12", max_p95: "40", samples: "9" }] + : [{ day: "2026-07-14", avg_p50: "50", avg_p95: "120", max_p95: "300", samples: "4" }]; + return new Response(JSON.stringify({ meta: [], data }), { status: 200 }); + }); + vi.stubGlobal("fetch", fetchMock); + + const res = await handle(authed(), statsEnv()); + expect(res.status).toBe(200); + const body = (await res.json()) as { + days: number; + metrics: { scan: unknown[]; upload_per_mb: unknown[] }; + }; + expect(body.days).toBe(7); // default window + expect(body.metrics.scan).toEqual([ + { day: "2026-07-14", avg_p50_ms: 3, avg_p95_ms: 12, max_p95_ms: 40, samples: 9 }, + ]); + expect(body.metrics.upload_per_mb).toEqual([ + { day: "2026-07-14", avg_p50_ms: 50, avg_p95_ms: 120, max_p95_ms: 300, samples: 4 }, + ]); + // Two queries issued, one per metric, each filtering its own sentinel column. + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(sqls.some((s) => s.includes("double7") && s.includes("double7 >= 0"))).toBe(true); + expect(sqls.some((s) => s.includes("double9") && s.includes("double9 >= 0"))).toBe(true); + // The default 7-day window is in the SQL. + expect(sqls.every((s) => s.includes("INTERVAL '7' DAY"))).toBe(true); + }); + + it("clamps the days param to [1, 90] and defaults to 7", async () => { + const seen: string[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (_url: string, init: { body: string }) => { + seen.push(init.body); + return new Response(JSON.stringify({ meta: [], data: [] }), { status: 200 }); + }), + ); + const call = async (q: string) => + handle( + new Request(`${STATS_URL}?days=${q}`, { + method: "GET", + headers: { authorization: "Bearer s3cret" }, + }), + statsEnv(), + ); + + await call("500"); // over the cap -> 90 + await call("0"); // under the floor -> 1 + await call("abc"); // non-numeric -> default 7 + await call("30"); // valid pass-through + + expect(seen.some((s) => s.includes("INTERVAL '90' DAY"))).toBe(true); + expect(seen.some((s) => s.includes("INTERVAL '1' DAY"))).toBe(true); + expect(seen.some((s) => s.includes("INTERVAL '7' DAY"))).toBe(true); + expect(seen.some((s) => s.includes("INTERVAL '30' DAY"))).toBe(true); + }); + + it("502s (not 200) when the AE SQL query fails upstream", async () => { + vi.stubGlobal( + "fetch", + vi.fn(async () => new Response("nope", { status: 403 })), + ); + const errSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + const res = await handle(authed(), statsEnv()); + expect(res.status).toBe(502); + expect(await res.json()).toMatchObject({ error: "query_failed" }); + errSpy.mockRestore(); + }); +});