perf: Batch dry-run remote cache existence checks

crates/turborepo-api-client/src/lib.rs

@@ -7,7 +7,7 @@
 #![allow(unused_assignments)]
 #![deny(clippy::all)]
 
-use std::{backtrace::Backtrace, env, future::Future, time::Duration};
+use std::{backtrace::Backtrace, collections::HashMap, env, future::Future, time::Duration};
 #[cfg(feature = "rustls-tls")]
 use std::{io::Cursor, path::Path};
 
@@ -76,6 +76,15 @@ pub trait Client {
     fn make_url(&self, endpoint: &str) -> Result<Url>;
 }
 
+/// Metadata returned for a hit by POST /v8/artifacts.
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ArtifactQueryHit {
+    pub task_duration_ms: u64,
+    pub sha: Option<String>,
+    pub dirty_hash: Option<String>,
+}
+
 pub trait CacheClient {
     fn get_artifact(
         &self,
@@ -291,6 +300,33 @@ impl Client for APIClient {
     }
 }
 
+impl APIClient {
+    /// Query multiple artifacts without downloading their bodies. An invalid or
+    /// unsupported response is an error so callers can fall back to HEAD.
+    pub async fn query_artifacts(
+        &self,
+        hashes: &[String],
+        token: &SecretString,
+        team_id: Option<&str>,
+        team_slug: Option<&str>,
+    ) -> Result<HashMap<String, Option<ArtifactQueryHit>>> {
+        let request = self
+            .api_request(Method::POST, self.make_url("/v8/artifacts")?)
+            .header("User-Agent", self.user_agent.clone())
+            .header("Content-Type", "application/json")
+            .bearer_auth(token.expose())
+            .json(&serde_json::json!({ "hashes": hashes }));
+        let request = Self::add_team_params(request, team_id, team_slug);
+        let response = retry::make_retryable_request(request, retry::RetryStrategy::Timeout)
+            .await?
+            .into_response();
+        if response.status() == StatusCode::FORBIDDEN {
+            return Err(Self::handle_403(response).await);
+        }
+        Ok(response.error_for_status()?.json().await?)
+    }
+}
+
 impl CacheClient for APIClient {
     #[tracing::instrument(skip_all)]
     async fn get_artifact(
@@ -1289,6 +1325,42 @@ mod test {
         assert_eq!(url.as_str(), "https://cache.example/v8/artifacts/abc123");
     }
 
+    #[tokio::test]
+    async fn query_artifacts_sends_one_authenticated_batch_and_parses_metadata()
+    -> anyhow::Result<()> {
+        let server = httpmock::MockServer::start_async().await;
+        let query = server
+            .mock_async(|when, then| {
+                when.method(httpmock::Method::POST)
+                    .path("/v8/artifacts")
+                    .query_param("teamId", "team_123")
+                    .query_param("slug", "my-team")
+                    .header("authorization", "Bearer secret-token")
+                    .json_body(serde_json::json!({"hashes": ["hit", "miss"]}));
+                then.status(200).json_body(serde_json::json!({
+                    "hit": {"size": 12, "taskDurationMs": 456, "sha": "abc", "dirtyHash": "def"},
+                    "miss": null
+                }));
+            })
+            .await;
+        let client = APIClient::new(server.base_url(), None, None, "2.0.0", false)?;
+        let results = client
+            .query_artifacts(
+                &["hit".into(), "miss".into()],
+                &SecretString::new("secret-token".into()),
+                Some("team_123"),
+                Some("my-team"),
+            )
+            .await?;
+        query.assert_calls_async(1).await;
+        let hit = results.get("hit").unwrap().as_ref().unwrap();
+        assert_eq!(hit.task_duration_ms, 456);
+        assert_eq!(hit.sha.as_deref(), Some("abc"));
+        assert_eq!(hit.dirty_hash.as_deref(), Some("def"));
+        assert!(results.get("miss").unwrap().is_none());
+        Ok(())
+    }
+
     #[tokio::test]
     async fn fetch_artifact_does_not_leak_credentials_to_preflight_location() -> anyhow::Result<()>
     {

crates/turborepo-cache/Cargo.toml

@@ -14,6 +14,7 @@ rustls-tls = ["turborepo-api-client/rustls-tls"]
 anyhow = { workspace = true, features = ["backtrace"] }
 filetime = "0.2"
 futures = { workspace = true }
+httpmock = { workspace = true }
 insta = { workspace = true }
 port_scanner = { workspace = true }
 test-case = { workspace = true }

crates/turborepo-cache/src/async_cache.rs

@@ -18,6 +18,7 @@ const WARNING_CUTOFF: u8 = 4;
 pub struct AsyncCache {
     real_cache: Arc<CacheMultiplexer>,
     writer_sender: mpsc::Sender<WorkerRequest>,
+    dry_run_sender: mpsc::UnboundedSender<(String, oneshot::Sender<Option<CacheHitMetadata>>)>,
 }
 
 enum WorkerRequest {
@@ -140,9 +141,32 @@ impl AsyncCache {
             }
         });
 
+        let (dry_run_sender, mut dry_run_receiver) =
+            mpsc::unbounded_channel::<(String, oneshot::Sender<Option<CacheHitMetadata>>)>();
+        let batch_cache = real_cache.clone();
+        tokio::spawn(async move {
+            while let Some(first) = dry_run_receiver.recv().await {
+                let mut pending = vec![first];
+                // Allow concurrently ready dry-run tasks to share a request.
+                let deadline = tokio::time::Instant::now() + std::time::Duration::from_millis(5);
+                while pending.len() < 100 {
+                    match tokio::time::timeout_at(deadline, dry_run_receiver.recv()).await {
+                        Ok(Some(next)) => pending.push(next),
+                        _ => break,
+                    }
+                }
+                let keys: Vec<_> = pending.iter().map(|(key, _)| key.clone()).collect();
+                let hits = batch_cache.batch_exists(&keys).await;
+                for ((_, sender), hit) in pending.into_iter().zip(hits) {
+                    let _ = sender.send(hit);
+                }
+            }
+        });
+
         Ok(AsyncCache {
             real_cache,
             writer_sender,
+            dry_run_sender,
         })
     }
 
@@ -176,6 +200,16 @@ impl AsyncCache {
         self.real_cache.exists(key).await
     }
 
+    /// Only dry runs use the query endpoint; ordinary cache reads retain their
+    /// existing HEAD/GET behavior.
+    pub async fn dry_run_exists(&self, key: &str) -> Result<Option<CacheHitMetadata>, CacheError> {
+        let (sender, receiver) = oneshot::channel();
+        self.dry_run_sender
+            .send((key.to_owned(), sender))
+            .map_err(|_| CacheError::CacheShuttingDown)?;
+        receiver.await.map_err(|_| CacheError::CacheShuttingDown)
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn fetch(
         &self,
@@ -249,6 +283,134 @@ mod tests {
         test_cases::{TestCase, get_test_cases},
     };
 
+    #[tokio::test]
+    async fn dry_run_falls_back_to_head_when_batch_is_unsupported() -> Result<()> {
+        let server = httpmock::MockServer::start_async().await;
+        let query = server
+            .mock_async(|when, then| {
+                when.method(httpmock::Method::POST).path("/v8/artifacts");
+                then.status(404);
+            })
+            .await;
+        let head = server
+            .mock_async(|when, then| {
+                when.method(httpmock::Method::HEAD)
+                    .path("/v8/artifacts/hit");
+                then.status(200)
+                    .header("x-artifact-duration", "42")
+                    .header("x-artifact-sha", "head-sha")
+                    .header("x-artifact-dirty-hash", "head-dirty");
+            })
+            .await;
+        let root = tempdir()?;
+        let root = AbsoluteSystemPathBuf::try_from(root.path())?;
+        let opts = CacheOpts {
+            cache_dir: ".turbo/cache".into(),
+            cache: CacheConfig::remote_only(),
+            workers: 1,
+            remote_cache_opts: None,
+            cache_max_age: None,
+            cache_max_size: None,
+        };
+        let cache = AsyncCache::new(
+            &opts,
+            &root,
+            Some(APIClient::new(
+                server.base_url(),
+                None,
+                None,
+                "2.0.0",
+                false,
+            )?),
+            Some(APIAuth {
+                team_id: None,
+                team_slug: None,
+                token: SecretString::new("test-token".into()),
+            }),
+            None,
+            LazyScmState::resolved(None),
+        )?;
+        assert_eq!(
+            cache.dry_run_exists("hit").await?,
+            Some(CacheHitMetadata {
+                source: CacheSource::Remote,
+                time_saved: 42,
+                sha: Some("head-sha".into()),
+                dirty_hash: Some("head-dirty".into()),
+            })
+        );
+        query.assert_calls_async(1).await;
+        head.assert_calls_async(1).await;
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn dry_run_batch_preserves_remote_metadata_and_misses() -> Result<()> {
+        use turborepo_api_client::{Bytes, CacheClient};
+
+        let port = port_scanner::request_open_port().unwrap();
+        let (ready_tx, ready_rx) = tokio::sync::oneshot::channel();
+        let server = tokio::spawn(start_test_server(port, Some(ready_tx)));
+        tokio::time::timeout(Duration::from_secs(5), ready_rx).await??;
+        let root = tempdir()?;
+        let root = AbsoluteSystemPathBuf::try_from(root.path())?;
+        let opts = CacheOpts {
+            cache_dir: ".turbo/cache".into(),
+            cache: CacheConfig::remote_only(),
+            workers: 1,
+            remote_cache_opts: None,
+            cache_max_age: None,
+            cache_max_size: None,
+        };
+        let client = APIClient::new(
+            format!("http://localhost:{port}"),
+            None,
+            None,
+            "2.0.0",
+            false,
+        )?;
+        let token = SecretString::new("test-token".into());
+        client
+            .put_artifact(
+                "hit",
+                tokio_stream::once(Ok(Bytes::from_static(b"data"))),
+                4,
+                123,
+                None,
+                &token,
+                None,
+                None,
+                Some("abc"),
+                Some("dirty"),
+            )
+            .await?;
+        let cache = AsyncCache::new(
+            &opts,
+            &root,
+            Some(client),
+            Some(APIAuth {
+                team_id: None,
+                team_slug: None,
+                token,
+            }),
+            None,
+            LazyScmState::resolved(None),
+        )?;
+        let (hit, miss) = tokio::join!(cache.dry_run_exists("hit"), cache.dry_run_exists("miss"));
+        assert_eq!(
+            hit?,
+            Some(CacheHitMetadata {
+                source: CacheSource::Remote,
+                time_saved: 123,
+                sha: Some("abc".into()),
+                dirty_hash: Some("dirty".into()),
+            })
+        );
+        assert_eq!(miss?, None);
+        server.abort();
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_async_cache() -> Result<()> {
         let port = port_scanner::request_open_port().unwrap();

crates/turborepo-cache/src/http.rs

@@ -320,6 +320,41 @@ impl HTTPCache {
         Ok(())
     }
 
+    /// Query remote hits in one request; missing or invalid entries are left to
+    /// the caller to resolve with the single-artifact endpoint.
+    pub async fn batch_exists(
+        &self,
+        hashes: &[String],
+    ) -> Result<HashMap<String, Option<CacheHitMetadata>>, CacheError> {
+        let response = self
+            .execute_with_token_refresh("batch", |api_auth| {
+                let client = &self.client;
+                async move {
+                    client
+                        .query_artifacts(
+                            hashes,
+                            &api_auth.token,
+                            api_auth.team_id.as_deref(),
+                            api_auth.team_slug.as_deref(),
+                        )
+                        .await
+                }
+            })
+            .await?;
+        Ok(response
+            .into_iter()
+            .map(|(hash, hit)| {
+                let metadata = hit.map(|hit| CacheHitMetadata {
+                    source: CacheSource::Remote,
+                    time_saved: hit.task_duration_ms,
+                    sha: hit.sha,
+                    dirty_hash: hit.dirty_hash,
+                });
+                (hash, metadata)
+            })
+            .collect())
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn exists(&self, hash: &str) -> Result<Option<CacheHitMetadata>, CacheError> {
         let response = self