feat(cache): trace where a task's time goes

crates/vt/src/session/cache/archive.rs

@@ -73,6 +73,12 @@ pub fn create_output_archive(
     Ok(())
 }
 
+/// The size of the file at `path`, such as an archive, or `None` if it can't
+/// be read. Only for logging.
+pub fn file_size(path: &AbsolutePath) -> Option<u64> {
+    std::fs::metadata(path.as_path()).ok().map(|metadata| metadata.len())
+}
+
 /// Extract a tar.zst archive, restoring files, folders and symbolic links
 /// relative to workspace root.
 ///

crates/vt/src/session/cache/mod.rs

@@ -11,7 +11,7 @@ use std::{
     fs::File,
     io::Write,
     sync::{Arc, OnceLock},
-    time::Duration,
+    time::{Duration, Instant},
 };
 
 // Re-export display functions for convenience
@@ -31,6 +31,7 @@ use vt_plan::{
     cache_metadata::{CacheMetadata, ExecutionCacheKey, SpawnFingerprint},
     remote_cache::{RemoteCacheAccess, ResolvedRemoteCacheConfig},
 };
+use vt_remote_cache::Fetched;
 use vt_str::Str;
 use wincode::{
     SchemaRead, SchemaReadOwned, SchemaWrite,
@@ -367,16 +368,26 @@ impl ExecutionCache {
         // Part of the key, so the same as when the entry was created.
         let input_negative_globs = PathGlobSet::new(&cache_metadata.input_config.negative_globs)?;
 
-        let local_miss = match self
+        let start = Instant::now();
+        let local_lookup = self
             .try_hit_local(
                 cache_metadata,
                 &cache_key,
                 globbed_inputs,
                 &input_negative_globs,
                 workspace_root,
             )
-            .await?
-        {
+            .await?;
+        tracing::debug!(
+            outcome = match &local_lookup {
+                Ok(_) => "hit",
+                Err(CacheMiss::NotFound) => "not_found",
+                Err(_) => "mismatch",
+            },
+            elapsed = ?start.elapsed(),
+            "looked up the local cache"
+        );
+        let local_miss = match local_lookup {
             Ok(value) => return Ok(Ok(CacheHit { value, source: CacheHitSource::Local })),
             Err(miss) => miss,
         };
@@ -393,7 +404,8 @@ impl ExecutionCache {
             ) => remote_config,
             None => return Ok(Err(local_miss)),
         };
-        let remote_miss = match self
+        let start = Instant::now();
+        let remote_lookup = self
             .try_hit_remote(
                 remote_config,
                 cache_metadata,
@@ -404,8 +416,18 @@ impl ExecutionCache {
                 cache_dir,
                 cancel_token,
             )
-            .await?
-        {
+            .await?;
+        tracing::debug!(
+            outcome = match &remote_lookup {
+                Ok(_) => "hit",
+                Err(CacheMiss::NotFound) => "not_found",
+                Err(CacheMiss::FingerprintMismatch(_)) => "mismatch",
+                Err(CacheMiss::RemoteReadFailed(_)) => "read_failed",
+            },
+            elapsed = ?start.elapsed(),
+            "looked up the remote cache"
+        );
+        let remote_miss = match remote_lookup {
             Ok(value) => return Ok(Ok(CacheHit { value, source: CacheHitSource::Remote })),
             Err(miss) => miss,
         };
@@ -473,10 +495,24 @@ impl ExecutionCache {
         cache_dir: &AbsolutePath,
         cancel_token: &CancellationToken,
     ) -> anyhow::Result<Result<CacheEntryValue, CacheMiss>> {
+        let start = Instant::now();
         let fetched = self
             .remote_clients
             .fetch(remote_config, cache_key, &cache_metadata.execution_cache_key, cancel_token)
             .await;
+        let (outcome, entries) = match &fetched {
+            Ok(Some(Fetched::Exact { entries })) => ("exact", entries.len()),
+            Ok(Some(Fetched::Fallback { .. })) => ("fallback", 0),
+            Ok(None) => ("not_found", 0),
+            Err(_) => ("failed", 0),
+        };
+        tracing::debug!(
+            outcome,
+            entries,
+            error = fetched.as_ref().err().map(tracing::field::debug),
+            elapsed = ?start.elapsed(),
+            "fetched from the remote cache"
+        );
         let validate = |cache_value: &CacheEntryValue| {
             cache_value.validate(
                 &cache_metadata.unfiltered_envs,
@@ -555,7 +591,9 @@ impl ExecutionCache {
 
         let cache_key = CacheEntryKey::from_metadata(cache_metadata);
 
+        let start = Instant::now();
         self.record(&cache_key, execution_cache_key, &cache_value, cache_dir).await?;
+        tracing::debug!(elapsed = ?start.elapsed(), "recorded the cache entry locally");
 
         let upload_error = Arc::new(OnceLock::new());
         let remote_config = match &cache_metadata.remote_cache {
@@ -611,6 +649,7 @@ impl ExecutionCache {
         workspace_root: &AbsolutePath,
         cache_dir: &AbsolutePath,
     ) -> anyhow::Result<()> {
+        let start = Instant::now();
         if let Some(archive_name) = &hit.value.output_archive {
             let archive_path = cache_dir.join(archive_name.as_str());
             if let Err(err) = archive::extract_output_archive(workspace_root, &archive_path) {
@@ -620,6 +659,11 @@ impl ExecutionCache {
                 }
                 return Err(err.context("failed to extract the output archive"));
             }
+            tracing::debug!(
+                bytes = archive::file_size(&archive_path),
+                elapsed = ?start.elapsed(),
+                "extracted the output archive"
+            );
         }
 
         if hit.source == CacheHitSource::Remote {
@@ -633,6 +677,7 @@ impl ExecutionCache {
                 tracing::warn!(?err, "failed to record a remote cache hit locally");
             }
         }
+        tracing::debug!(source = ?hit.source, elapsed = ?start.elapsed(), "restored the outputs");
         Ok(())
     }
 }

crates/vt/src/session/cache/remote.rs

@@ -17,12 +17,14 @@ use std::{
     fs::File,
     io::{self, Read as _, Write as _},
     sync::{Arc, Mutex, OnceLock, PoisonError},
+    time::Instant,
 };
 
 use bytes::Bytes;
 use rustc_hash::FxHashMap;
 use tokio::sync::mpsc;
 use tokio_util::{sync::CancellationToken, task::TaskTracker};
+use tracing::Instrument as _;
 use vt_path::AbsolutePath;
 use vt_plan::{
     cache_metadata::ExecutionCacheKey,
@@ -142,8 +144,18 @@ pub(super) fn resolve(
         None => return Err(CacheMiss::NotFound),
     };
     let mut newest_miss = None;
-    for entry in entries {
-        match check_entry(entry, &mut validate) {
+    for (index, entry) in entries.into_iter().enumerate() {
+        let start = Instant::now();
+        let value_bytes = entry.value.len();
+        let checked = check_entry(entry, &mut validate);
+        tracing::debug!(
+            index,
+            value_bytes,
+            valid = checked.is_ok(),
+            elapsed = ?start.elapsed(),
+            "checked a remote cache entry"
+        );
+        match checked {
             Ok(restore) => return Ok(restore),
             Err(miss) => {
                 newest_miss.get_or_insert(miss);
@@ -238,6 +250,7 @@ impl RemoteClients {
         cache_dir: &AbsolutePath,
         cancel_token: &CancellationToken,
     ) -> Result<Str, ReadError> {
+        let start = Instant::now();
         let client = self.client(remote_config).map_err(ReadError::Download)?;
         let archive_name = vt_str::format!("{}.tar.zst", uuid::Uuid::new_v4());
         let archive_path = cache_dir.join(archive_name.as_str());
@@ -251,6 +264,12 @@ impl RemoteClients {
             // Best-effort cleanup: the file may not have been created.
             let _ = std::fs::remove_file(temp_path.as_path());
         }
+        tracing::debug!(
+            ok = result.is_ok(),
+            bytes = result.as_ref().ok().and_then(|&()| archive::file_size(&archive_path)),
+            elapsed = ?start.elapsed(),
+            "downloaded an output archive"
+        );
         result.map(|()| archive_name)
     }
 
@@ -268,11 +287,18 @@ impl RemoteClients {
         cache_value: &CacheEntryValue,
         cache_dir: &AbsolutePath,
     ) -> Result<impl Future<Output = Result<(), UploadError>> + Send + use<>, UploadError> {
+        let start = Instant::now();
         let client = self.client(remote_config)?;
         let key = encode_key(cache_key)?;
         let secondary_key = encode_key(execution_cache_key)?;
         let value = encode_value(cache_value)?;
         let archive = cache_value.output_archive.as_ref().map(|name| cache_dir.join(name.as_str()));
+        tracing::debug!(
+            value_bytes = value.len(),
+            archive = archive.is_some(),
+            elapsed = ?start.elapsed(),
+            "prepared a remote cache upload"
+        );
         Ok(async move {
             client.store(&key, &secondary_key, &value, archive.as_deref()).await?;
             Ok(())
@@ -311,16 +337,28 @@ impl RemoteUploads {
         error: Arc<OnceLock<UploadError>>,
     ) {
         let cancel = self.cancel.clone();
-        self.tracker.spawn(async move {
-            // A cancelled upload sets its error itself, instead of being
-            // aborted, so every cancelled upload has one.
-            let result =
-                cancel.run_until_cancelled(upload).await.unwrap_or(Err(UploadError::Interrupted));
-            if let Err(err) = result {
-                tracing::debug!(?err, "remote cache upload failed");
-                let _ = error.set(err);
+        // The upload outlives its task, so it carries the task's span.
+        self.tracker.spawn(
+            async move {
+                let start = Instant::now();
+                // A cancelled upload sets its error itself, instead of being
+                // aborted, so every cancelled upload has one.
+                let result = cancel
+                    .run_until_cancelled(upload)
+                    .await
+                    .unwrap_or(Err(UploadError::Interrupted));
+                match result {
+                    Ok(()) => {
+                        tracing::debug!(elapsed = ?start.elapsed(), "remote cache upload finished");
+                    }
+                    Err(err) => {
+                        tracing::debug!(?err, elapsed = ?start.elapsed(), "remote cache upload failed");
+                        let _ = error.set(err);
+                    }
+                }
             }
-        });
+            .in_current_span(),
+        );
     }
 
     /// The number of uploads still running.
@@ -331,6 +369,8 @@ impl RemoteUploads {
     /// Wait for all uploads to finish. If `interrupt_token` is cancelled
     /// first, or already was, cancel them and wait for them to stop.
     pub(super) async fn wait(&self, interrupt_token: &CancellationToken) {
+        let start = Instant::now();
+        let pending = self.tracker.len();
         self.tracker.close();
         tokio::select! {
             biased;
@@ -340,6 +380,14 @@ impl RemoteUploads {
                 self.tracker.wait().await;
             }
         }
+        if pending > 0 {
+            tracing::debug!(
+                pending,
+                interrupted = interrupt_token.is_cancelled(),
+                elapsed = ?start.elapsed(),
+                "waited for remote cache uploads"
+            );
+        }
         self.tracker.reopen();
     }
 }

crates/vt/src/session/execute/cache_update.rs

@@ -1,7 +1,11 @@
 //! Post-run cache update: decide whether a finished spawn may be cached and,
 //! if so, store its fingerprint, captured output, and output archive.
 
-use std::{collections::BTreeMap, sync::Arc, time::Duration};
+use std::{
+    collections::BTreeMap,
+    sync::Arc,
+    time::{Duration, Instant},
+};
 
 use rustc_hash::FxHashSet;
 use tokio_util::sync::CancellationToken;
@@ -460,7 +464,15 @@ fn collect_and_archive_outputs(
     let archive_name: Str = vt_str::format!("{}.tar.zst", uuid::Uuid::new_v4());
     let archive_path = cache_dir.join(archive_name.as_str());
 
+    let start = Instant::now();
     archive::create_output_archive(workspace_root, &sorted_files, &output_folders, &archive_path)?;
+    tracing::debug!(
+        files = sorted_files.len(),
+        folders = output_folders.len(),
+        bytes = archive::file_size(&archive_path),
+        elapsed = ?start.elapsed(),
+        "archived the outputs"
+    );
 
     Ok(Some(archive_name))
 }