Skip to content

Commit 4e9fde6

Browse files
pmaxhoganclaude
andauthored
feat(telemetry): capture latency percentiles and add rollup query endpoint (#132)
## What Fills in the telemetry latency percentiles that shipped in V1 as always-empty wire keys, end to end: - **Client capture.** A new app-global `LatencyReservoir` in `driven-core` (bounded ring buffer per metric, cap 4096, nearest-rank p50/p95). The scanner records per-file scan-processing latency; the executor records each completed upload op's latency normalized per MiB. One reservoir is shared (`Arc`) into every account's executor + orchestrator (mirroring the existing `with_mem_gauge` seam), wired once in `assembly`. - **Ping build.** `build_payload` now carries the drained percentiles instead of the hardcoded `LatencyP50P95::default()`. The window is snapshotted read-only at build and reset **only after a successful send**, so a dropped/aborted ping re-uses the same window (matching how the event-count deltas re-send an un-checkpointed window). - **Worker ingest.** `writePing` appends the 4 percentiles as AE doubles (`double7..10`), with a `-1` sentinel for an empty metric so the rollup can tell "no samples" from a legitimate `0 ms` (a sub-ms per-file scan rounds to 0). - **Worker rollup query.** New gated `GET /telemetry/v1/stats/latency?days=N` returning per-day aggregates via the Analytics Engine SQL API. ## Why DESIGN s13 lists "latency histograms (p50, p95) for scan and upload-per-file" in the payload, but nothing captured per-op durations, so `latency_p50_p95_ms.{scan,upload_per_mb}` were always empty and the worker had no read surface. This makes the signal real and queryable. ## Consent / privacy Capture is gated on the telemetry-enabled pref: the reservoir's enable flag is initialized from the persisted pref at boot (before any capture) and flipped in lockstep by `apply_enabled_change`. Turning telemetry off drops any captured samples. A latency sample is a bare millisecond duration - no path/name/content. ## Endpoint contract ``` GET /telemetry/v1/stats/latency?days=7 Authorization: Bearer <QUERY_TOKEN> 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 } ] } } ``` `days` defaults to 7, clamped to `[1, 90]`. Per metric, per UTC day: mean of the pinged p50s, mean + max of the pinged p95s, and the count of pings that reported the metric. Empty-latency pings (the `-1` sentinel) are excluded per metric (`WHERE <p50col> >= 0`). Status: `401` missing/wrong bearer, `405` non-GET, `502` upstream AE failure, `503` not configured. ## Where things live - Reservoir: `crates/driven-core/src/telemetry.rs` (new). - Scan capture: `scanner::scan_with_latency` (thin `scan()` wrapper keeps existing callers/tests unchanged). - Upload capture: `ExecOne::run` in `executor.rs`. - Wiring: `src-tauri/src/assembly.rs`; held on `AppState`'s `TelemetryRuntime`. - Ping: `src-tauri/src/telemetry.rs`. Worker: `telemetry-worker/src/index.ts` + `README.md`. ## Operational caveats (action required post-merge) - **Set `QUERY_TOKEN` + `CF_API_TOKEN` as wrangler secrets** (see `telemetry-worker/README.md`). `deploy-telemetry.yml` auto-deploys on merge, so `/stats/latency` ships **503-until-configured** by design; the ingest path needs neither. `CF_ACCOUNT_ID` is optional (defaults to the Driven account). - **The AE SQL query has only been tested against a mocked `fetch`** - `toDate()` day-grouping, the `double7..10` column mapping, and `{meta,data}` parsing have never hit real Analytics Engine. It needs a one-time live smoke-check after the secrets are set. This is the one thing the test suite structurally cannot cover. - **`upload_per_mb` is an op-latency proxy, not pure transfer time:** the timer wraps the whole upload op (hash -> crypto -> pacer gate -> network), so on a throttled/metered link it reflects pacer wait, not Drive throughput. A defensible reading of "upload op"; documented so the percentiles aren't misread. ## Tests - Rust (`driven-core`): reservoir + percentile edge cases (empty / single / even / odd / large-uniform / ring-cap), consent no-op + disable-drops-samples, `per_mb_ms` normalization + zero guard. - Rust (`driven-app`): `build_payload` carries the drained percentiles; ping snapshots then resets on success; a failed send keeps the window; a disabled ping does not touch the reservoir; `apply_enabled_change` clears the reservoir on disable. - Worker (vitest): the 4 latency doubles + `-1` sentinel (legit `0` preserved); `/stats/latency` 503-unconfigured, 401 no/wrong bearer, 405 wrong method, 200 per-day aggregates (mocked AE `fetch`), `days` clamping, 502 on upstream failure. Gates green: `cargo fmt --check`, `clippy --workspace --all-targets -D warnings`, `cargo test -p driven-core`, `cargo check --workspace`; worker `typecheck` + `lint` + `test` (56). ## Issue reference The task said "Closes #5", but `#5` resolves to a **PR**, not an issue (shared numbering - `gh issue view 5` fails), so "Closes #5" would attach to a PR and close nothing. Referencing the V2 backlog tracking issue instead; correct me if a different issue was meant. Refs #34 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent fecac7a commit 4e9fde6

12 files changed

Lines changed: 1341 additions & 47 deletions

File tree

crates/driven-core/src/executor.rs

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -815,6 +815,12 @@ pub struct DefaultExecutor {
815815
/// the file size - the one qualitative pipeline contract that IS
816816
/// deterministically measurable against the instantaneous fake.
817817
mem_gauge: Option<Arc<MemGauge>>,
818+
/// App-global latency sampler (DESIGN s13 telemetry), or `None` when
819+
/// telemetry latency capture is not wired (every test + the chaos harness).
820+
/// When `Some`, each completed upload op records its wall-clock latency
821+
/// normalized per MiB via [`crate::telemetry::per_mb_ms`]; the reservoir's
822+
/// own enable gate makes the record a no-op when telemetry is off.
823+
latency: Option<Arc<crate::telemetry::LatencyReservoir>>,
818824
#[cfg(test)]
819825
mid_upload_hook: Option<MidUploadHook>,
820826
#[cfg(test)]
@@ -893,13 +899,29 @@ impl DefaultExecutor {
893899
vss: deps.vss,
894900
pool,
895901
mem_gauge: None,
902+
latency: None,
896903
#[cfg(test)]
897904
mid_upload_hook: None,
898905
#[cfg(test)]
899906
post_upload_hook: None,
900907
}
901908
}
902909

910+
/// Attach the app-global latency reservoir (DESIGN s13 telemetry) so each
911+
/// completed upload op records its per-MiB latency. Mirrors
912+
/// [`Self::with_mem_gauge`]: a builder over the always-present `Option`
913+
/// field, so no test / e2e struct-literal site changes. Production wires it
914+
/// from the assembly; every other construction path leaves it `None` (zero
915+
/// overhead).
916+
#[must_use]
917+
pub fn with_latency_reservoir(
918+
mut self,
919+
reservoir: Arc<crate::telemetry::LatencyReservoir>,
920+
) -> Self {
921+
self.latency = Some(reservoir);
922+
self
923+
}
924+
903925
/// Resolve the crypto decision for one source (M5 GA-blocking surface).
904926
///
905927
/// Consults the injected [`CryptoProvider`]; a `None` provider means every
@@ -4467,6 +4489,20 @@ impl<'a> ExecOne<'a> {
44674489
permit: tokio::sync::OwnedSemaphorePermit,
44684490
on_outcome: &OutcomeSink<'_>,
44694491
) -> anyhow::Result<OpOutcome> {
4492+
// Telemetry (DESIGN s13): time upload ops so a completed upload records
4493+
// its per-MiB latency. Only armed for the upload op kinds and only when a
4494+
// reservoir is wired + enabled (a trash op is not an "upload-per-MB"
4495+
// sample). Zero cost otherwise (not even an `Instant::now`).
4496+
let upload_timer = match op {
4497+
Op::HashThenUpload { .. } | Op::UploadBundle { .. } => self
4498+
.this
4499+
.latency
4500+
.as_ref()
4501+
.filter(|r| r.is_enabled())
4502+
.map(|_| std::time::Instant::now()),
4503+
Op::Trash { .. } => None,
4504+
};
4505+
44704506
let out = match op {
44714507
Op::HashThenUpload {
44724508
source_id,
@@ -4493,6 +4529,32 @@ impl<'a> ExecOne<'a> {
44934529
self.this.bundle_upload(self.source, members).await
44944530
}
44954531
};
4532+
4533+
// Record the completed upload's per-MiB latency (DESIGN s13). Only a
4534+
// successful upload that moved bytes contributes a sample: a Done-Upload
4535+
// carries the uploaded byte count, a BundleDone carries the bundle object
4536+
// size; a trash, skip, or failure records nothing.
4537+
if let (Some(reservoir), Some(started)) = (self.this.latency.as_ref(), upload_timer) {
4538+
if let Ok(outcome) = &out {
4539+
let bytes = match outcome {
4540+
OpOutcome::Done {
4541+
kind: DoneKind::Upload,
4542+
bytes: Some(bytes),
4543+
..
4544+
} => Some(*bytes),
4545+
OpOutcome::BundleDone { bytes, .. } => Some(*bytes),
4546+
_ => None,
4547+
};
4548+
if let Some(bytes) = bytes {
4549+
let elapsed_ms =
4550+
u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
4551+
if let Some(per_mb) = crate::telemetry::per_mb_ms(elapsed_ms, bytes) {
4552+
reservoir.record_upload_per_mb_ms(per_mb);
4553+
}
4554+
}
4555+
}
4556+
}
4557+
44964558
// Stream the per-op activity for a produced outcome (the op's durable
44974559
// file-state commit already happened inside hash_then_upload / trash_op).
44984560
// A hard error (Err) carries no OpOutcome and is handled by the caller.

crates/driven-core/src/lib.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ pub mod pacer;
3030
pub mod planner;
3131
pub mod scanner;
3232
pub mod state;
33+
pub mod telemetry;
3334
pub mod time;
3435
pub mod types;
3536
pub mod watcher;

crates/driven-core/src/orchestrator.rs

Lines changed: 30 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -433,6 +433,13 @@ pub struct SyncOrchestrator {
433433
/// token. An atomic so the cycle path can set it and the run loop can read it
434434
/// without holding a lock across the select.
435435
suspended: std::sync::atomic::AtomicBool,
436+
/// App-global latency sampler (DESIGN s13 telemetry), or `None` when
437+
/// telemetry latency capture is not wired (tests / the chaos harness).
438+
/// Threaded into [`crate::scanner::scan_with_latency`] so each scan records
439+
/// per-file processing latency; the SAME `Arc` is also given to the executor
440+
/// (via [`crate::executor::DefaultExecutor::with_latency_reservoir`]) for the
441+
/// upload-per-MB metric. Set via [`Self::with_latency_reservoir`].
442+
latency: Option<Arc<crate::telemetry::LatencyReservoir>>,
436443
}
437444

438445
impl SyncOrchestrator {
@@ -483,9 +490,25 @@ impl SyncOrchestrator {
483490
vss_create_ledger: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
484491
orphan_cleanup_done: Mutex::new(false),
485492
suspended: std::sync::atomic::AtomicBool::new(false),
493+
latency: None,
486494
}
487495
}
488496

497+
/// Attach the app-global latency reservoir (DESIGN s13 telemetry) so each
498+
/// scan records per-file processing latency. Pass the SAME `Arc` given to the
499+
/// executor via
500+
/// [`DefaultExecutor::with_latency_reservoir`](crate::executor::DefaultExecutor::with_latency_reservoir)
501+
/// so both metrics feed one reservoir. Without this, scans capture nothing
502+
/// (the tests / chaos harness path).
503+
#[must_use]
504+
pub fn with_latency_reservoir(
505+
mut self,
506+
reservoir: Arc<crate::telemetry::LatencyReservoir>,
507+
) -> Self {
508+
self.latency = Some(reservoir);
509+
self
510+
}
511+
489512
/// Attach the per-cycle Windows VSS snapshot provider (ROADMAP M3.5).
490513
///
491514
/// Pass the SAME `Arc<dyn VssProvider>` that was threaded into the
@@ -1294,7 +1317,13 @@ impl SyncOrchestrator {
12941317
scanned: 0,
12951318
})
12961319
.await;
1297-
let scan = crate::scanner::scan(source, self.state.as_ref(), mode).await?;
1320+
let scan = crate::scanner::scan_with_latency(
1321+
source,
1322+
self.state.as_ref(),
1323+
mode,
1324+
self.latency.as_deref(),
1325+
)
1326+
.await?;
12981327

12991328
// DESIGN s5.5: flag still-present-but-now-excluded paths so the UI can
13001329
// surface them; never a trash. Non-fatal - a flag write failure must

crates/driven-core/src/scanner.rs

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -156,10 +156,29 @@ fn has_alternate_data_streams(path: &Path) -> bool {
156156
/// Pure aside from local filesystem reads and the `state` load; emits no
157157
/// ops and mutates no state - the planner (SPEC s7) and executor (SPEC s8)
158158
/// own those side effects.
159+
///
160+
/// Thin wrapper over [`scan_with_latency`] with no telemetry capture; the
161+
/// existing tests + callers that do not thread a reservoir use this.
159162
pub async fn scan(
160163
source: &SourceRow,
161164
state: &dyn StateRepo,
162165
mode: ScanMode,
166+
) -> anyhow::Result<ScanResult> {
167+
scan_with_latency(source, state, mode, None).await
168+
}
169+
170+
/// [`scan`] with an optional [`LatencyReservoir`](crate::telemetry::LatencyReservoir)
171+
/// for per-file scan-processing latency capture (DESIGN s13 telemetry). When
172+
/// `latency` is `Some` AND capture is enabled, each fully-processed file records
173+
/// its stat-through-change-detection wall-clock time (the dominant cost is the
174+
/// BLAKE3 re-hash on a deep-verify pass); when `None` there is zero overhead.
175+
/// The orchestrator passes its shared reservoir here; every other caller passes
176+
/// `None`.
177+
pub async fn scan_with_latency(
178+
source: &SourceRow,
179+
state: &dyn StateRepo,
180+
mode: ScanMode,
181+
latency: Option<&crate::telemetry::LatencyReservoir>,
163182
) -> anyhow::Result<ScanResult> {
164183
let known = state
165184
.load_source_file_state(source.id)
@@ -292,6 +311,16 @@ pub async fn scan(
292311
continue;
293312
}
294313

314+
// Telemetry (DESIGN s13): time this file's stat-through-change-detection
315+
// processing. Only armed when a reservoir is threaded in AND capture is
316+
// enabled, so a scan with no telemetry pays nothing (not even the
317+
// `Instant::now`). Recorded at the end of the iteration for a
318+
// fully-processed file; the cheap early-`continue` skips below (cloud-only
319+
// placeholder, NFC collision) intentionally record nothing.
320+
let file_timer = latency
321+
.filter(|r| r.is_enabled())
322+
.map(|_| std::time::Instant::now());
323+
295324
let meta = match entry.metadata() {
296325
Ok(m) => m,
297326
Err(err) => {
@@ -388,6 +417,14 @@ pub async fn scan(
388417
mtime_ns,
389418
});
390419
}
420+
421+
// Telemetry (DESIGN s13): record this file's per-file scan-processing
422+
// latency. `latency` is `Some` iff a reservoir was armed above; the
423+
// `record_scan_ms` call re-checks the enable gate (a no-op if telemetry
424+
// was disabled mid-scan).
425+
if let (Some(res), Some(started)) = (latency, file_timer) {
426+
res.record_scan_ms(u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX));
427+
}
391428
}
392429

393430
// Split the known-but-not-seen paths into genuine deletions vs

0 commit comments

Comments
 (0)