diff --git a/crates/turborepo-lib/src/run/mod.rs b/crates/turborepo-lib/src/run/mod.rs index 95d309fcab88c..d171568d2f311 100644 --- a/crates/turborepo-lib/src/run/mod.rs +++ b/crates/turborepo-lib/src/run/mod.rs @@ -987,6 +987,7 @@ impl Run { ) -> Option<( turborepo_repository::toolchain::CompileCacheEndpoint, tokio::sync::broadcast::Sender<()>, + std::sync::Arc, )> { if !self.opts.future_flags.experimental_cargo_sccache { return None; @@ -1075,13 +1076,14 @@ impl Run { server_port: turborepo_sccache_proxy::derive_server_port(&self.repo_root), }; let shutdown = server.shutdown_handle(); + let stats = server.stats(); info!("sccache compile cache proxy listening on {}", endpoint.url); tokio::spawn(async move { if let Err(err) = server.run().await { error!("sccache compile cache proxy error: {err}"); } }); - Some((endpoint, shutdown)) + Some((endpoint, shutdown, stats)) } async fn cleanup_proxy( @@ -1288,10 +1290,11 @@ impl Run { drop(_setup_span); let sccache_proxy = self.start_sccache_proxy_if_needed().await; - let (compile_cache_endpoint, sccache_shutdown) = match sccache_proxy { - Some((endpoint, shutdown)) => (Some(endpoint), Some(shutdown)), - None => (None, None), - }; + let (compile_cache_endpoint, sccache_shutdown, incremental_cache_stats) = + match sccache_proxy { + Some((endpoint, shutdown, stats)) => (Some(endpoint), Some(shutdown), Some(stats)), + None => (None, None, None), + }; let mut visitor = Visitor::new( self.pkg_dep_graph.clone(), @@ -1365,6 +1368,18 @@ impl Run { self.processes.stop().await; } + // Snapshot the incremental-cache traffic now that every task (and + // its tools) has finished. `None` when the proxy never started — + // the summary line only appears for runs that attempted + // incremental caching. + let incremental_cache = incremental_cache_stats.map(|stats| { + let snapshot = stats.snapshot(); + turborepo_run_summary::IncrementalCacheSummary { + hits: snapshot.hits, + misses: snapshot.misses, + } + }); + visitor .finish( exit_code, @@ -1374,6 +1389,7 @@ impl Run { &self.env_at_execution_start, &self.scm, self.opts.scope_opts.pkg_inference_root.as_deref(), + incremental_cache, ) .await?; diff --git a/crates/turborepo-lib/src/task_graph/visitor/mod.rs b/crates/turborepo-lib/src/task_graph/visitor/mod.rs index 0248ecbb54b0b..228a9cbe72208 100644 --- a/crates/turborepo-lib/src/task_graph/visitor/mod.rs +++ b/crates/turborepo-lib/src/task_graph/visitor/mod.rs @@ -1015,6 +1015,7 @@ impl<'a> Visitor<'a> { env_at_execution_start, scm, ))] + #[allow(clippy::too_many_arguments)] pub(crate) async fn finish( self, exit_code: i32, @@ -1024,6 +1025,7 @@ impl<'a> Visitor<'a> { env_at_execution_start: &EnvironmentVariableMap, scm: &SCM, pkg_inference_root: Option<&AnchoredSystemPath>, + incremental_cache: Option, ) -> Result<(), Error> { let Self { package_graph, @@ -1077,6 +1079,7 @@ impl<'a> Visitor<'a> { scm, is_watch, Some(task_hasher.external_deps_hash_cache()), + incremental_cache, ) .await?) } diff --git a/crates/turborepo-run-summary/src/execution.rs b/crates/turborepo-run-summary/src/execution.rs index 3833a371c3f79..da4d5b6db7845 100644 --- a/crates/turborepo-run-summary/src/execution.rs +++ b/crates/turborepo-run-summary/src/execution.rs @@ -45,6 +45,36 @@ pub struct ExecutionSummary<'a> { pub exit_code: i32, } +/// Totals for reuse below the task boundary: work units a tool running +/// inside a task fetched from (hits) or had to rebuild (misses) via the +/// incremental cache. Toolchain-agnostic — for Rust this is sccache +/// compile units served through the Remote Cache; other toolchains can +/// contribute the same shape. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct IncrementalCacheSummary { + pub hits: u64, + pub misses: u64, +} + +/// The summary footer's "Incremental cache" line: reuse below the task +/// boundary. `None` — and therefore absent from the output — unless the +/// run actually exchanged work units: a repo that never engaged +/// incremental caching should not see the line at all. +fn incremental_cache_line( + ui: ColorConfig, + incremental_cache: Option, +) -> Option<(&'static str, String)> { + let incremental = incremental_cache.filter(|s| s.hits + s.misses > 0)?; + Some(( + "Incremental cache", + format!( + "{}, {} misses", + color!(ui, BOLD, "{} hits", incremental.hits), + incremental.misses, + ), + )) +} + impl<'a> ExecutionSummary<'a> { pub fn new( command: String, @@ -77,6 +107,7 @@ impl<'a> ExecutionSummary<'a> { ui: ColorConfig, path: AbsoluteSystemPathBuf, failed_tasks: Vec<&T>, + incremental_cache: Option, ) { let maybe_full_turbo = if self.cached == self.attempted && self.attempted > 0 { match std::env::var("TERM_PROGRAM").as_deref() { @@ -115,6 +146,10 @@ impl<'a> ExecutionSummary<'a> { ), ]; + if let Some(line) = incremental_cache_line(ui, incremental_cache) { + line_data.insert(2, line); + } + if path.exists() { line_data.push(("Summary", path.to_string())); } @@ -474,6 +509,39 @@ mod test { use super::*; + #[test] + fn test_incremental_cache_line_conditional() { + // strip_ansi: label/value text is what users see uncolored. + let ui = ColorConfig::new(true); + + // No incremental caching attempted: no line. + assert!(incremental_cache_line(ui, None).is_none()); + + // Attempted, but no work units exchanged (e.g. every task was a + // task-cache hit and no tool ever ran): still no line. + assert!( + incremental_cache_line(ui, Some(IncrementalCacheSummary { hits: 0, misses: 0 })) + .is_none() + ); + + // Real traffic renders, misses included even when zero. + let (label, value) = incremental_cache_line( + ui, + Some(IncrementalCacheSummary { + hits: 407, + misses: 12, + }), + ) + .unwrap(); + assert_eq!(label, "Incremental cache"); + assert_eq!(value, "407 hits, 12 misses"); + + let (_, value) = + incremental_cache_line(ui, Some(IncrementalCacheSummary { hits: 0, misses: 3 })) + .unwrap(); + assert_eq!(value, "0 hits, 3 misses"); + } + #[tokio::test] async fn test_multiple_tasks() -> Result<(), Box> { let summary = ExecutionTracker::new(); diff --git a/crates/turborepo-run-summary/src/lib.rs b/crates/turborepo-run-summary/src/lib.rs index 01fbadfcfc64f..5f2b067c2c8c7 100644 --- a/crates/turborepo-run-summary/src/lib.rs +++ b/crates/turborepo-run-summary/src/lib.rs @@ -14,7 +14,8 @@ mod tracker; pub use duration::TurboDuration; pub use execution::{ - ExecutionSummary, ExecutionTracker, SummaryState, TaskState, TaskSummaryInfo, TaskTracker, + ExecutionSummary, ExecutionTracker, IncrementalCacheSummary, SummaryState, TaskState, + TaskSummaryInfo, TaskTracker, }; pub use global_hash::{GlobalEnvConfiguration, GlobalEnvVarSummary, GlobalHashSummary}; pub use observability::Handle as ObservabilityHandle; diff --git a/crates/turborepo-run-summary/src/tracker.rs b/crates/turborepo-run-summary/src/tracker.rs index e058f4950f750..7de85ccf37b8a 100644 --- a/crates/turborepo-run-summary/src/tracker.rs +++ b/crates/turborepo-run-summary/src/tracker.rs @@ -26,7 +26,7 @@ use turborepo_ui::{BOLD, BOLD_CYAN, ColorConfig, GREY, color, cprintln, cwriteln use crate::{ GlobalHashSummary, SCMState, TaskTracker, - execution::{ExecutionSummary, ExecutionTracker, TaskState}, + execution::{ExecutionSummary, ExecutionTracker, IncrementalCacheSummary, TaskState}, observability::Handle as ObservabilityHandle, task::{SinglePackageTaskSummary, TaskSummary}, task_factory::TaskSummaryFactory, @@ -235,6 +235,7 @@ impl RunTracker { scm: &SCM, is_watch: bool, external_deps_hashes: Option<&HashMap>, + incremental_cache: Option, ) -> Result<(), Error> where E: EngineInfo + Sync, @@ -274,7 +275,7 @@ impl RunTracker { ); let path = repo_root.join_components(&[".turbo", "runs", "dummy.json"]); - execution.print(ui, path, failed_tasks.iter().collect()); + execution.print(ui, path, failed_tasks.iter().collect(), incremental_cache); } return Ok(()); @@ -307,7 +308,14 @@ impl RunTracker { .await?; run_summary - .finish(end_time, exit_code, pkg_dep_graph, ui, is_watch) + .finish( + end_time, + exit_code, + pkg_dep_graph, + ui, + is_watch, + incremental_cache, + ) .await } @@ -398,6 +406,7 @@ impl<'a> RunSummary<'a> { pkg_dep_graph: &PackageGraph, ui: ColorConfig, is_watch: bool, + incremental_cache: Option, ) -> Result<(), Error> { // Handle observability shutdown before the dry run check to ensure graceful // cleanup even when metrics are not being emitted. @@ -428,7 +437,7 @@ impl<'a> RunSummary<'a> { if !is_watch && let Some(execution) = &self.execution { let path = self.get_path(); let failed_tasks = self.get_failed_tasks(); - execution.print(ui, path, failed_tasks); + execution.print(ui, path, failed_tasks, incremental_cache); } Ok(()) diff --git a/crates/turborepo-sccache-proxy/src/lib.rs b/crates/turborepo-sccache-proxy/src/lib.rs index 4c28d015bd957..404a3a4fa2805 100644 --- a/crates/turborepo-sccache-proxy/src/lib.rs +++ b/crates/turborepo-sccache-proxy/src/lib.rs @@ -129,11 +129,65 @@ fn artifact_id_for_key(key: &str) -> String { hex::encode(hasher.finalize()) } +/// Live counters for the compile-unit traffic the proxy serves, feeding the +/// run summary's "Incremental cache" line. Object-granular: one GET hit is +/// one reused work unit, one GET miss is one unit the tool rebuilt (and +/// usually stored afterward). Health-check traffic (`.sccache_check`) is +/// excluded — it says nothing about reuse. +#[derive(Debug, Default)] +pub struct IncrementalCacheStats { + hits: std::sync::atomic::AtomicU64, + misses: std::sync::atomic::AtomicU64, + stores: std::sync::atomic::AtomicU64, +} + +/// A point-in-time copy of [`IncrementalCacheStats`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct IncrementalCacheSnapshot { + pub hits: u64, + pub misses: u64, + pub stores: u64, +} + +impl IncrementalCacheStats { + pub fn snapshot(&self) -> IncrementalCacheSnapshot { + use std::sync::atomic::Ordering; + IncrementalCacheSnapshot { + hits: self.hits.load(Ordering::Relaxed), + misses: self.misses.load(Ordering::Relaxed), + stores: self.stores.load(Ordering::Relaxed), + } + } + + fn record_hit(&self) { + self.hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + + fn record_miss(&self) { + self.misses + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + + fn record_store(&self) { + self.stores + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } +} + +/// The storage self-check object sccache reads and writes at server +/// startup. Infrastructure traffic, not work-unit reuse. +const SCCACHE_HEALTH_CHECK_KEY: &str = ".sccache_check"; + +fn is_health_check(key: &str) -> bool { + key == SCCACHE_HEALTH_CHECK_KEY +} + struct ProxyState { client: APIClient, auth: APIAuth, /// Expected `Authorization` header value: `Bearer {token}`. expected_authorization: String, + stats: Arc, } impl ProxyState { @@ -159,6 +213,7 @@ pub struct SccacheProxyServer { router: Router, port: u16, shutdown_tx: tokio::sync::broadcast::Sender<()>, + stats: Arc, } impl SccacheProxyServer { @@ -178,10 +233,12 @@ impl SccacheProxyServer { .map_err(|source| Error::Bind { port, source })? .port(); + let stats = Arc::new(IncrementalCacheStats::default()); let state = Arc::new(ProxyState { client, auth, expected_authorization: format!("Bearer {token}"), + stats: stats.clone(), }); // A single fallback dispatcher rather than per-method routes: the // webdav surface opendal (sccache's storage client) speaks includes @@ -196,9 +253,16 @@ impl SccacheProxyServer { router, port, shutdown_tx, + stats, }) } + /// Live counters for the traffic this proxy serves. Snapshot at run end + /// for the run summary's "Incremental cache" line. + pub fn stats(&self) -> Arc { + self.stats.clone() + } + pub fn port(&self) -> u16 { self.port } @@ -340,10 +404,16 @@ async fn handle_get(state: Arc, key: String) -> Response { { Ok(Some(response)) => { debug!("sccache proxy hit for key {key}"); + if !is_health_check(&key) { + state.stats.record_hit(); + } Body::from_stream(response.bytes_stream()).into_response() } Ok(None) => { debug!("sccache proxy miss for key {key}"); + if !is_health_check(&key) { + state.stats.record_miss(); + } StatusCode::NOT_FOUND.into_response() } Err(err) => { @@ -391,6 +461,9 @@ async fn handle_put(state: Arc, key: String, body: bytes::Bytes) -> { Ok(()) => { debug!("sccache proxy stored key {key} ({len} bytes)"); + if !is_health_check(&key) { + state.stats.record_store(); + } StatusCode::OK.into_response() } Err(err) => { @@ -496,6 +569,7 @@ mod tests { let bearer = "opendal-test-token"; let (server, port) = start_proxy(backend_port, bearer).await; let shutdown = server.shutdown_handle(); + let stats = server.stats(); let proxy = tokio::spawn(server.run()); let builder = opendal::services::Webdav::default() @@ -520,6 +594,16 @@ mod tests { let read = op.read(key).await.expect("cache read through opendal"); assert_eq!(read.to_bytes().as_ref(), b"object bytes"); + // The stats feeding the run summary's "Incremental cache" line + // count real work-unit traffic and exclude the health-check probe. + let snapshot = stats.snapshot(); + assert_eq!(snapshot.hits, 1, "one successful cache read"); + assert_eq!(snapshot.stores, 1, "one cache write (probe excluded)"); + assert!( + snapshot.misses >= 1, + "the read-before-write must count at least one miss, got {snapshot:?}" + ); + let _ = shutdown.send(()); let _ = proxy.await; backend.abort(); diff --git a/crates/turborepo/ARCHITECTURE.md b/crates/turborepo/ARCHITECTURE.md index 8006c746baccd..7bd803764d19c 100644 --- a/crates/turborepo/ARCHITECTURE.md +++ b/crates/turborepo/ARCHITECTURE.md @@ -304,7 +304,11 @@ whether anything changed; Cargo decides how and in what order to build.** pays off, while local development is served by cargo's own incremental compilation — which the injected `CARGO_INCREMENTAL=0` would disable. Lifecycle: started in `Run::execute_visitor` before the visitor, - shut down fire-and-forget after it. + shut down fire-and-forget after it. The proxy counts the work-unit + traffic it serves (hits/misses/stores, health-check probe excluded) + and the run summary footer reports it as a toolchain-agnostic + "Incremental cache" line — reuse below the task boundary — shown only + when the run actually exchanged work units. A `--filter` that names a crate while support is disabled gets an error hint pointing at the flag. Released turbo versions hard-error on unknown