From 7562b259e7f4e1e2c159bf9885f54bd6944f7fc9 Mon Sep 17 00:00:00 2001 From: wan9chi Date: Sun, 8 Nov 2026 23:08:54 +0800 Subject: [PATCH 1/2] feat(cache): trace where a task's time goes Add debug spans and events, with durations and sizes as fields, at the points that explain a task's time: the scheduler's permit wait, globbed input hashing, local and remote cache lookups (fetch, each entry's check, archive download), inferred input validation, spawn and exit, post-run fingerprinting, output archiving, local recording, restoring, and remote uploads (prepare, send, the OIDC token request, and the final wait). The remote cache client logs each response's status and length without the request URL. Per-file fingerprinting and validation are at trace level. The leaf span carries the task and its command, and an upload keeps its task's span. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + crates/vt/src/session/cache/archive.rs | 6 ++ crates/vt/src/session/cache/mod.rs | 59 ++++++++++++++-- crates/vt/src/session/cache/remote.rs | 70 ++++++++++++++++--- crates/vt/src/session/execute/cache_update.rs | 14 +++- crates/vt/src/session/execute/fingerprint.rs | 27 ++++++- crates/vt/src/session/execute/mod.rs | 16 +++++ crates/vt/src/session/execute/scheduler.rs | 31 ++++++-- crates/vt_remote_cache/Cargo.toml | 1 + .../vt_remote_cache/src/auth/github_oidc.rs | 9 ++- crates/vt_remote_cache/src/lib.rs | 52 ++++++++++++-- 11 files changed, 251 insertions(+), 35 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cf6f60cf3..8e8ccd7b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5303,6 +5303,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tokio", + "tracing", "url", "vt_path", "vt_str", diff --git a/crates/vt/src/session/cache/archive.rs b/crates/vt/src/session/cache/archive.rs index accf96796..480647299 100644 --- a/crates/vt/src/session/cache/archive.rs +++ b/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 { + 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. /// diff --git a/crates/vt/src/session/cache/mod.rs b/crates/vt/src/session/cache/mod.rs index 179d94dda..e76ef7aad 100644 --- a/crates/vt/src/session/cache/mod.rs +++ b/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,7 +368,8 @@ 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, @@ -375,8 +377,17 @@ impl ExecutionCache { &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> { + 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(()) } } diff --git a/crates/vt/src/session/cache/remote.rs b/crates/vt/src/session/cache/remote.rs index 7afdac876..fa39efe1d 100644 --- a/crates/vt/src/session/cache/remote.rs +++ b/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 { + 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> + 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>, ) { 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(); } } diff --git a/crates/vt/src/session/execute/cache_update.rs b/crates/vt/src/session/execute/cache_update.rs index f44791a34..aeb1ac3c3 100644 --- a/crates/vt/src/session/execute/cache_update.rs +++ b/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)) } diff --git a/crates/vt/src/session/execute/fingerprint.rs b/crates/vt/src/session/execute/fingerprint.rs index 35c0cdb6f..9525a2cae 100644 --- a/crates/vt/src/session/execute/fingerprint.rs +++ b/crates/vt/src/session/execute/fingerprint.rs @@ -9,6 +9,7 @@ use std::{ fs::File, io::{self, BufRead}, sync::Arc, + time::Instant, }; use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; @@ -167,6 +168,7 @@ impl PostRunFingerprint { tracked_envs: BTreeMap>, tracked_env_queries: BTreeMap>, ) -> anyhow::Result { + let start = Instant::now(); let inferred_inputs = inferred_path_reads .par_iter() .filter(|(path, _)| { @@ -175,6 +177,7 @@ impl PostRunFingerprint { .map(|(relative_path, path_read)| { let full_path = Arc::::from(base_dir.join(relative_path)); let mut fingerprint = fingerprint_path(&full_path, *path_read)?; + tracing::trace!(path = relative_path.as_str(), "fingerprinted an inferred input"); if let PathFingerprint::Folder(Some(listing)) = &mut fingerprint { if !rebuildable.is_empty() { let DirListing { entries, ignored_entries } = listing; @@ -198,6 +201,12 @@ impl PostRunFingerprint { Ok((relative_path.clone(), InferredInput { fingerprint, may_be_missing })) }) .collect::>>()?; + tracing::debug!( + reads = inferred_path_reads.len(), + inputs = inferred_inputs.len(), + elapsed = ?start.elapsed(), + "created the post-run fingerprint" + ); Ok(Self { inferred_inputs, tracked_envs, tracked_env_queries }) } @@ -220,6 +229,7 @@ impl PostRunFingerprint { input_negative_globs: &PathGlobSet<'_>, unfiltered_envs: &FxHashMap>, Arc>, ) -> anyhow::Result> { + let start = Instant::now(); let input_mismatch = self.inferred_inputs.par_iter().find_map_any(|(input_relative_path, input)| { let InferredInput { fingerprint: path_fingerprint, may_be_missing } = input; @@ -243,9 +253,14 @@ impl PostRunFingerprint { current.ignored_entries.clone_from(&stored.ignored_entries); leave_out_excluded_entries(current, input_relative_path, input_negative_globs); } - if path_fingerprint == ¤t_path_fingerprint - || (*may_be_missing && current_path_fingerprint == PathFingerprint::NotFound) - { + let unchanged = path_fingerprint == ¤t_path_fingerprint + || (*may_be_missing && current_path_fingerprint == PathFingerprint::NotFound); + tracing::trace!( + path = input_relative_path.as_str(), + unchanged, + "validated an inferred input" + ); + if unchanged { None } else { let (kind, entry_name) = @@ -263,6 +278,12 @@ impl PostRunFingerprint { Some(Ok(PostRunMismatch::Input { kind, path })) } }); + tracing::debug!( + inputs = self.inferred_inputs.len(), + valid = input_mismatch.is_none(), + elapsed = ?start.elapsed(), + "validated the inferred inputs" + ); if let Some(result) = input_mismatch { return result.map(Some); } diff --git a/crates/vt/src/session/execute/mod.rs b/crates/vt/src/session/execute/mod.rs index 3d2e4ab47..2d3b91b42 100644 --- a/crates/vt/src/session/execute/mod.rs +++ b/crates/vt/src/session/execute/mod.rs @@ -459,6 +459,7 @@ async fn run( ) .await .map_err(|err| Report::failed(ExecutionError::Spawn(err)))?; + tracing::debug!(fspy = fspy_enabled, elapsed = ?start.elapsed(), "spawned the task process"); // 6. Drain the pipes and wait for exit, concurrently with the IPC driver. // The driver must be polled during pipe drain — otherwise a tool doing @@ -561,6 +562,7 @@ async fn lookup_cache<'a>( // Compute globbed inputs from positive globs at execution time. // Globs are already workspace-root-relative (resolved at task graph stage). + let start = Instant::now(); let globbed_inputs = compute_globbed_inputs( workspace_root, &cache_metadata.input_config.positive_globs, @@ -569,6 +571,11 @@ async fn lookup_cache<'a>( .map_err(|err| { Report::failed(ExecutionError::Cache { kind: CacheErrorKind::Lookup, source: err }) })?; + tracing::debug!( + files = globbed_inputs.len(), + elapsed = ?start.elapsed(), + "hashed the globbed inputs" + ); match cache .try_hit(cache_metadata, &globbed_inputs, workspace_root, cache_dir, cancel_token) @@ -634,6 +641,7 @@ async fn run_child( stop_accepting: Option<&StopAccepting>, fast_fail_token: CancellationToken, ) -> Result { + let start = Instant::now(); let pipe_result: Result<(), ExecutionError> = if let Some(sinks) = sinks { let stdout = child.stdout.take().expect("SpawnStdio::Piped yields a stdout pipe"); let stderr = child.stderr.take().expect("SpawnStdio::Piped yields a stderr pipe"); @@ -663,5 +671,13 @@ async fn run_child( if let Some(stop_accepting) = stop_accepting { stop_accepting.signal(); } + if let Ok(outcome) = &wait_result { + tracing::debug!( + success = outcome.exit_status.success(), + exit_code = outcome.exit_status.code(), + elapsed = ?start.elapsed(), + "the task process exited" + ); + } wait_result } diff --git a/crates/vt/src/session/execute/scheduler.rs b/crates/vt/src/session/execute/scheduler.rs index fb605a4a4..9dcc7b588 100644 --- a/crates/vt/src/session/execute/scheduler.rs +++ b/crates/vt/src/session/execute/scheduler.rs @@ -2,7 +2,7 @@ //! (dependency order, concurrency limits, fast-fail), and hands each leaf to //! [`execute_spawn`] which owns *how* a single spawn runs. -use std::{cell::RefCell, ffi::OsStr, io::Write as _, num::NonZeroUsize, sync::Arc}; +use std::{cell::RefCell, ffi::OsStr, io::Write as _, num::NonZeroUsize, sync::Arc, time::Instant}; use futures_util::{FutureExt, StreamExt, future::LocalBoxFuture, stream::FuturesUnordered}; use petgraph::Direction; @@ -83,7 +83,11 @@ impl ExecutionContext<'_> { /// (killing in-flight child processes). Ctrl-C cancels the `cancel_token`. /// Either cancellation causes this method to close the semaphore, drain /// remaining futures, and return. - #[tracing::instrument(level = "debug", skip_all)] + #[tracing::instrument( + level = "debug", + skip_all, + fields(tasks = graph.graph.node_count(), concurrency_limit = graph.concurrency_limit) + )] async fn execute_expanded_graph(&self, graph: &ExecutionGraph) { if graph.graph.node_count() == 0 { return; @@ -144,9 +148,15 @@ impl ExecutionContext<'_> { ) -> LocalBoxFuture<'a, ExecutionNodeIndex> { let sem = semaphore.clone(); async move { + let start = Instant::now(); if let Ok(_permit) = sem.acquire_owned().await && !self.cancelled() { + tracing::debug!( + task = %graph.graph[node_ix].task_display, + waited = ?start.elapsed(), + "acquired a concurrency permit" + ); self.execute_node(graph, node_ix).await; } node_ix @@ -183,11 +193,21 @@ impl ExecutionContext<'_> { /// from the graph reporter and delegates to the appropriate execution /// method. On failure (non-zero exit or infrastructure error), cancels the /// `fast_fail_token`. - #[tracing::instrument(level = "debug", skip_all)] - async fn execute_leaf(&self, display: &ExecutionItemDisplay, leaf_kind: &LeafExecutionKind) { + #[tracing::instrument( + level = "debug", + skip_all, + fields(task = %item_display.task_display, command = %item_display.command) + )] + async fn execute_leaf( + &self, + item_display: &ExecutionItemDisplay, + leaf_kind: &LeafExecutionKind, + ) { + let start = Instant::now(); // Borrow the reporter briefly to create the leaf reporter, then drop // the RefCell guard before any `.await` point. - let mut leaf_reporter = self.reporter.borrow_mut().new_leaf_execution(display, leaf_kind); + let mut leaf_reporter = + self.reporter.borrow_mut().new_leaf_execution(item_display, leaf_kind); let failed = match leaf_kind { LeafExecutionKind::InProcess(in_process_execution) => { @@ -226,6 +246,7 @@ impl ExecutionContext<'_> { } } }; + tracing::debug!(failed, elapsed = ?start.elapsed(), "finished the task item"); if failed { self.fast_fail_token.cancel(); } diff --git a/crates/vt_remote_cache/Cargo.toml b/crates/vt_remote_cache/Cargo.toml index 13d7e8b26..607665be6 100644 --- a/crates/vt_remote_cache/Cargo.toml +++ b/crates/vt_remote_cache/Cargo.toml @@ -24,6 +24,7 @@ serde_bytes = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["fs", "sync"] } +tracing = { workspace = true } url = { workspace = true } vt_path = { workspace = true } vt_str = { workspace = true } diff --git a/crates/vt_remote_cache/src/auth/github_oidc.rs b/crates/vt_remote_cache/src/auth/github_oidc.rs index b705ee1b1..243c62b8c 100644 --- a/crates/vt_remote_cache/src/auth/github_oidc.rs +++ b/crates/vt_remote_cache/src/auth/github_oidc.rs @@ -3,7 +3,7 @@ use std::{ fmt, sync::Arc, - time::{Duration, SystemTime}, + time::{Duration, Instant, SystemTime}, }; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; @@ -164,6 +164,7 @@ impl TokenRequest { } async fn send(&self, http: &reqwest::Client) -> Result { + let start = Instant::now(); let network_error = |err: reqwest::Error| TokenError::Network(err.without_url()); let response = http .get(self.url.clone()) @@ -174,6 +175,12 @@ impl TokenRequest { .await .map_err(network_error)?; let status = response.status(); + // Neither the request URL nor the tokens are logged. + tracing::debug!( + status = status.as_u16(), + elapsed = ?start.elapsed(), + "GitHub Actions OIDC token responded" + ); if status != StatusCode::OK { return Err(TokenError::Status(status)); } diff --git a/crates/vt_remote_cache/src/lib.rs b/crates/vt_remote_cache/src/lib.rs index 72f402ee8..5987090f8 100644 --- a/crates/vt_remote_cache/src/lib.rs +++ b/crates/vt_remote_cache/src/lib.rs @@ -5,7 +5,10 @@ pub mod auth; #[cfg(test)] mod test_server; -use std::{sync::Arc, time::Duration}; +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; use bytes::Bytes; use reqwest::{ @@ -192,7 +195,9 @@ impl Client { /// Returns an error if the credentials can't be obtained, the request /// fails, the server responds with a status other than 200 or 404, or the /// body of a 200 response isn't an exact or fallback match. + #[tracing::instrument(level = "debug", skip_all)] pub async fn fetch(&self, key: &[u8], secondary_key: &[u8]) -> Result, Error> { + let start = Instant::now(); let headers = self.auth_headers(Operation::Fetch).await?; let body = encode_cbor(&FetchRequest { key, secondary_key }); let response = self @@ -204,6 +209,7 @@ impl Client { .send() .await .map_err(network_error)?; + log_response(&response, start); if response.status() == StatusCode::NOT_FOUND { // Read the body so the connection can be reused. It doesn't matter // if that fails, because the status alone is the answer. @@ -211,6 +217,7 @@ impl Client { return Ok(None); } let body = check_status(response).await?.bytes().await.map_err(network_error)?; + tracing::debug!(bytes = body.len(), elapsed = ?start.elapsed(), "remote cache fetch received"); decode_fetched(&body).map(Some) } @@ -222,11 +229,14 @@ impl Client { /// /// Returns an error if the credentials can't be obtained, the request /// fails, or the server responds with a status other than 200. + #[tracing::instrument(level = "debug", skip_all)] pub async fn download(&self, blob_id: &str) -> Result { + let start = Instant::now(); let mut url = self.blob_url.clone(); url.path_segments_mut().map_err(|()| Error::InvalidEndpoint(None))?.push(blob_id); let headers = self.auth_headers(Operation::Download).await?; let response = self.http.get(url).headers(headers).send().await.map_err(network_error)?; + log_response(&response, start); Ok(Download { response: check_status(response).await? }) } @@ -239,6 +249,7 @@ impl Client { /// Returns an error if the credentials can't be obtained, the blob file /// can't be opened, the request fails, or the server responds with a /// status other than 200. + #[tracing::instrument(level = "debug", skip_all)] pub async fn store( &self, key: &[u8], @@ -246,12 +257,24 @@ impl Client { value: &[u8], blob: Option<&AbsolutePath>, ) -> Result<(), Error> { + let start = Instant::now(); let headers = self.auth_headers(Operation::Store).await?; + let auth = start.elapsed(); let metadata = StoreMetadata { key, secondary_key, value }; let mut form = Form::new().part("metadata", metadata_part(&metadata)); - if let Some(blob) = blob { - form = form.part("blob", blob_part(blob).await?); - } + let blob_bytes = if let Some(blob) = blob { + let (part, length) = blob_part(blob).await?; + form = form.part("blob", part); + Some(length) + } else { + None + }; + tracing::debug!( + value_bytes = value.len(), + blob_bytes, + ?auth, + "sending a remote cache store" + ); let response = self .http .post(self.store_url.clone()) @@ -260,6 +283,7 @@ impl Client { .send() .await .map_err(network_error)?; + log_response(&response, start); // The response's blob ID isn't needed. Read the body anyway, so the // connection can be reused. check_status(response).await?.bytes().await.map_err(network_error)?; @@ -267,6 +291,18 @@ impl Client { } } +/// Log the status and length of `response`, and the time since `start`, when +/// the operation started. The request URL is left out, since the endpoint may +/// contain credentials. +fn log_response(response: &Response, start: Instant) { + tracing::debug!( + status = response.status().as_u16(), + content_length = response.content_length(), + elapsed = ?start.elapsed(), + "remote cache responded" + ); +} + /// Only HTTP 200 counts as success. For other statuses, the error includes the /// message in the response body. async fn check_status(response: Response) -> Result { @@ -322,12 +358,14 @@ fn metadata_part(metadata: &StoreMetadata<'_>) -> Part { Part::bytes(encode_cbor(metadata)).mime_str("application/cbor").expect("valid MIME type") } -async fn blob_part(path: &AbsolutePath) -> Result { +/// The blob part streamed from the file at `path`, and its length. +async fn blob_part(path: &AbsolutePath) -> Result<(Part, u64), Error> { let file = tokio::fs::File::open(path).await.map_err(Error::ReadBlob)?; let length = file.metadata().await.map_err(Error::ReadBlob)?.len(); - Ok(Part::stream_with_length(file, length) + let part = Part::stream_with_length(file, length) .mime_str("application/octet-stream") - .expect("valid MIME type")) + .expect("valid MIME type"); + Ok((part, length)) } #[cfg(test)] From 42bda187be3a6d7bfe6b4decca7ba91ba1a4c0d0 Mon Sep 17 00:00:00 2001 From: wan9chi Date: Mon, 9 Nov 2026 17:03:40 +0800 Subject: [PATCH 2/2] fix(cache): keep per-input traces under their task Rayon's threads don't run in the calling thread's span, so the trace events for each inferred input that the post-run fingerprint and its validation emit had no task or command. They now name the span the work was started in as their parent. Co-Authored-By: Claude Opus 5.5 --- crates/vt/src/session/execute/fingerprint.rs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/crates/vt/src/session/execute/fingerprint.rs b/crates/vt/src/session/execute/fingerprint.rs index 9525a2cae..baec46628 100644 --- a/crates/vt/src/session/execute/fingerprint.rs +++ b/crates/vt/src/session/execute/fingerprint.rs @@ -169,6 +169,9 @@ impl PostRunFingerprint { tracked_env_queries: BTreeMap>, ) -> anyhow::Result { let start = Instant::now(); + // Rayon's threads don't run in this thread's span, so the per-input + // events name it as their parent, which keeps them under the task. + let span = tracing::Span::current(); let inferred_inputs = inferred_path_reads .par_iter() .filter(|(path, _)| { @@ -177,7 +180,11 @@ impl PostRunFingerprint { .map(|(relative_path, path_read)| { let full_path = Arc::::from(base_dir.join(relative_path)); let mut fingerprint = fingerprint_path(&full_path, *path_read)?; - tracing::trace!(path = relative_path.as_str(), "fingerprinted an inferred input"); + tracing::trace!( + parent: &span, + path = relative_path.as_str(), + "fingerprinted an inferred input" + ); if let PathFingerprint::Folder(Some(listing)) = &mut fingerprint { if !rebuildable.is_empty() { let DirListing { entries, ignored_entries } = listing; @@ -230,6 +237,8 @@ impl PostRunFingerprint { unfiltered_envs: &FxHashMap>, Arc>, ) -> anyhow::Result> { let start = Instant::now(); + // As in `create`, the per-input events name this span as their parent. + let span = tracing::Span::current(); let input_mismatch = self.inferred_inputs.par_iter().find_map_any(|(input_relative_path, input)| { let InferredInput { fingerprint: path_fingerprint, may_be_missing } = input; @@ -256,6 +265,7 @@ impl PostRunFingerprint { let unchanged = path_fingerprint == ¤t_path_fingerprint || (*may_be_missing && current_path_fingerprint == PathFingerprint::NotFound); tracing::trace!( + parent: &span, path = input_relative_path.as_str(), unchanged, "validated an inferred input"