mbx_cache_core/
lib.rs

1//! Protocol and storage primitives for mbx build caches.
2//!
3//! This crate contains the types shared by cache clients, the task-scoped
4//! cache agent, and remote cache implementations. Protocol records are
5//! serialized with [`canonical_json`] before hashing; changing their shape is
6//! therefore a wire-format change, not merely an implementation detail.
7//!
8//! Most consumers start with [`CacheDigest`] and the local stores
9//! [`LocalCas`] and [`LocalActionCache`]. Remote clients use
10//! [`RemoteCacheClient`], while mbx's compiler shim communicates with a
11//! [`CacheAgent`] using [`AgentRequest`] and [`AgentResponse`].
12//!
13//! ```
14//! use mbx_cache_core::{CacheDigest, canonical_json};
15//! use serde::Serialize;
16//!
17//! #[derive(Serialize)]
18//! struct Key<'a> {
19//!     compiler: &'a str,
20//!     source: CacheDigest,
21//! }
22//!
23//! let source = CacheDigest::blake3(b"fn main() {}\n");
24//! let bytes = canonical_json(&Key { compiler: "rustc", source })?;
25//! let action = CacheDigest::blake3(&bytes);
26//! assert_eq!(action.algorithm, "blake3");
27//! # Ok::<(), eyre::Report>(())
28//! ```
29#![deny(missing_docs)]
30
31use eyre::{Result, bail, eyre};
32use log::warn;
33use reqwest::header::HeaderValue;
34use serde::{Deserialize, Serialize};
35use std::path::{Path, PathBuf};
36use std::time::{Duration, Instant};
37use url::Url;
38
39mod agent;
40mod client;
41mod local;
42mod path_mapping;
43mod remote_http;
44mod remote_s3;
45mod sigv4;
46mod uploads;
47
48pub use agent::{
49    AGENT_PROTOCOL_VERSION, AgentEvent, AgentEventObserver, AgentRemoteCache, AgentRequest,
50    AgentResponse, AgentStats, CacheAgent, CompilerStats, FileDigestCache, FileDigestScope,
51    FileIdentity, NoFileDigestCache, RecordedFileDigest, RestoreStats, is_task_identity,
52    task_manifest_actions,
53};
54pub use client::BlockingAgentClient;
55pub use local::{LocalActionCache, LocalCas};
56pub use mbx_cache_protocol::{
57    ACTION_PROMISE_MEDIA_TYPE, ACTION_RESULT_BATCH_MEDIA_TYPE, ACTION_RESULT_MEDIA_TYPE,
58    ActionPrediction, ActionPromiseCompletion, ActionPromiseJoin, ActionPromiseState,
59    ActionResult as RemoteActionResult, BLOB_MEDIA_TYPE, BLOB_PACK_BLOBS_HEADER,
60    BLOB_PACK_BYTES_HEADER, BLOB_PACK_HEADER_BYTES, BLOB_PACK_MAGIC, BLOB_PACK_MEDIA_TYPE,
61    BLOB_PACK_RECEIPT_MEDIA_TYPE, CLIENT_METADATA_MEDIA_TYPE, Capabilities, CapabilityFeatures,
62    CapabilityLimits, CapabilityProtocol, CcMetadata, DIGEST_LIST_MEDIA_TYPE, DIRECTORY_MEDIA_TYPE,
63    Digest as CacheDigest, DigestAlgorithm, Directory as CacheDirectory,
64    DirectoryNode as CacheDirectoryNode, FileNode as CacheFileNode, MAX_ACTION_PREDICTION_PAYLOAD,
65    MAX_ACTION_PROMISE_CLAIM_BYTES, NAMESPACE_HEADER, PROTOCOL_HEADER, PROTOCOL_VERSION,
66    RustcMetadata, SymlinkNode as CacheSymlinkNode, TASK_ACTION_MANIFEST_MEDIA_TYPE,
67    TaskActionManifest,
68};
69pub use path_mapping::{
70    PathMapping, PathNormalizationError, normalize_mapped_path, normalize_resolved_mapped_path,
71    resolve_path_mappings,
72};
73use remote_http::HttpRemoteCache;
74#[cfg(feature = "fuzzing")]
75#[doc(hidden)]
76pub use remote_http::fuzz_decode_blob_pack;
77pub(crate) use remote_http::{BlobPackLimits, blob_pack_chunk};
78use remote_s3::S3RemoteCache;
79pub use remote_s3::{S3ConditionalWrites, S3RemoteCacheConfig};
80pub use sigv4::S3Credentials;
81/// Cap the JSON bodies a remote cache can hand back. Blob downloads are bounded
82/// by the size their digest promises, but action results and manifests carry no
83/// such claim, so without an explicit ceiling a hostile or broken server can
84/// stream until this process runs out of memory -- for manifests, long before
85/// `validate_task_manifest` ever sees the payload. The bound matches the agent's
86/// own request ceiling so both ends of the protocol refuse the same magnitude.
87const MAX_REMOTE_JSON_BYTES: u64 = 16 * 1024 * 1024;
88/// Ceiling on the opaque part of an entity tag this client will carry back.
89///
90/// A tag is only ever echoed into `If-Match`, so its length is bounded to keep
91/// a server from choosing how large a request header this client sends.
92const MAX_ETAG_BYTES: usize = 256;
93// Match the server's default maximum while retaining a client-side ceiling
94// when the remote advertises or names something larger.
95const MAX_REMOTE_BLOB_BYTES: u64 = 5 * 1024 * 1024 * 1024;
96const MAX_STAGED_BLOB_PACK_BYTES: u64 = 256 * 1024 * 1024;
97const MAX_STAGED_BLOB_PACK_ITEMS: usize = 2 * 1024;
98const BLOB_PACK_TIMEOUT_BYTES_PER_UNIT: u64 = MAX_STAGED_BLOB_PACK_BYTES / 4;
99const BLOB_PACK_TIMEOUT_ITEMS_PER_UNIT: usize = MAX_STAGED_BLOB_PACK_ITEMS / 4;
100/// Bytes read from one pack member before the next chunk is yielded.
101const PACK_STREAM_CHUNK_BYTES: usize = 64 * 1024;
102/// Actions looked up in one batched request.
103///
104/// A prefetch wants its first results back while the rest are still being
105/// answered, so a batch stays small enough to keep the download pipeline fed
106/// rather than as large as a server would accept.
107const MAX_ACTION_BATCH_ITEMS: usize = 256;
108/// Ceiling on one batched action-result response, scaled by what was asked for.
109const MAX_ACTION_RESULT_BYTES: u64 = 64 * 1024;
110
111/// Serialize a protocol object using the JSON Canonicalization Scheme.
112///
113/// Action digests are computed from these bytes, so callers must not use
114/// serde's struct field order as part of the wire contract.
115pub fn canonical_json(value: &impl Serialize) -> Result<Vec<u8>> {
116    Ok(mbx_cache_protocol::canonical_json(value)?)
117}
118
119#[derive(
120    Debug,
121    Clone,
122    Copy,
123    Serialize,
124    Deserialize,
125    Default,
126    strum::EnumString,
127    strum::Display,
128    PartialEq,
129    Eq,
130)]
131#[serde(rename_all = "kebab-case")]
132#[strum(serialize_all = "kebab-case")]
133/// Operations permitted against a configured remote cache.
134pub enum RemoteCacheMode {
135    /// Permit reads from and writes to the remote cache.
136    #[default]
137    ReadWrite,
138    /// Permit reads but never publish new objects.
139    ReadOnly,
140    /// Publish objects but never satisfy lookups from the remote cache.
141    WriteOnly,
142}
143
144impl RemoteCacheMode {
145    /// Whether this mode permits remote cache reads.
146    pub fn reads(self) -> bool {
147        matches!(self, Self::ReadWrite | Self::ReadOnly)
148    }
149
150    /// Whether this mode permits remote cache writes.
151    pub fn writes(self) -> bool {
152        matches!(self, Self::ReadWrite | Self::WriteOnly)
153    }
154}
155
156/// Connection, authentication, and retry settings for [`RemoteCacheClient`].
157pub struct RemoteCacheConfig {
158    /// Base URL of the remote cache service.
159    pub base_url: Url,
160    /// Server-side namespace used to isolate cache objects.
161    pub namespace: String,
162    /// Static bearer token, if configured directly.
163    pub token: Option<String>,
164    /// File containing a bearer token that may be refreshed externally.
165    pub token_file: Option<PathBuf>,
166    /// Audience used when obtaining an OIDC token from the CI environment.
167    pub oidc_audience: Option<String>,
168    /// Maximum time allowed to establish a connection.
169    pub connect_timeout: Duration,
170    /// Maximum time without response progress for ordinary requests.
171    pub read_timeout: Duration,
172    /// Deadline for one blob download, spanning every retry attempt and the
173    /// backoff between them rather than bounding a single attempt.
174    ///
175    /// A single stalled attempt is already bounded by `connect_timeout` and
176    /// `read_timeout`, so this budget exists to cap the total wall-clock one
177    /// logical download may spend: exhausting it fails the download even when
178    /// retries remain. Size it for the largest artifact worth waiting on, not
179    /// for one attempt at it.
180    pub download_timeout: Duration,
181    /// Number of attempts after the initial request for retryable failures.
182    pub retries: i64,
183}
184
185/// Backing data for a blob upload.
186pub enum BlobSource {
187    /// Bytes held in memory.
188    Bytes(Vec<u8>),
189    /// A temporary file whose lifetime is owned by the upload.
190    File(tempfile::NamedTempFile),
191    /// A persistent file at the given path.
192    Path(PathBuf),
193}
194
195/// A digest paired with the data to upload under that digest.
196pub struct BlobUpload {
197    /// Expected digest and length of the source data.
198    pub digest: CacheDigest,
199    /// Data source read by [`RemoteCacheClient::put_blob`].
200    pub source: BlobSource,
201}
202
203/// Task action-manifest bytes returned with their concurrency token.
204pub struct RemoteActionManifest {
205    /// Raw canonical manifest JSON.
206    pub bytes: Vec<u8>,
207    /// Entity tag used for conditional manifest replacement.
208    pub etag: String,
209}
210
211/// A verified set of remote CAS objects downloaded through blob-pack streams.
212pub struct RemoteBlobPack {
213    _directory: tempfile::TempDir,
214    /// Verified blobs paired with paths in this pack's temporary directory.
215    pub blobs: Vec<(CacheDigest, PathBuf)>,
216    /// Number of HTTP pack requests needed to retrieve the requested set.
217    pub requests: u64,
218    /// Unique digests requested from the remote service.
219    pub requested: Vec<CacheDigest>,
220    /// Number of verified blob frames received.
221    pub blob_count: u64,
222    /// Total unframed blob payload bytes received.
223    pub payload_bytes: u64,
224    /// Total bytes received including framing.
225    pub framed_bytes: u64,
226}
227
228/// What a server did with the blobs in an uploaded pack.
229#[derive(Debug, Clone, Copy, Deserialize)]
230pub struct BlobPackReceipt {
231    /// Blobs this request added to the remote cache.
232    #[serde(default)]
233    pub created: u64,
234    /// Blobs the remote cache already held.
235    #[serde(default)]
236    pub existing: u64,
237}
238
239#[derive(Debug, Clone, Copy, PartialEq, Eq)]
240/// Result of a conditional task-manifest write.
241pub enum ManifestPutOutcome {
242    /// The manifest was stored.
243    Stored,
244    /// The supplied entity-tag precondition did not match.
245    PreconditionFailed,
246}
247
248/// A client for a remote mbx cache.
249///
250/// The client validates digests, response sizes, media types, and redirects at
251/// the protocol boundary. It is safe to share between asynchronous tasks.
252///
253/// Which service is on the other end is not part of this type's contract. A
254/// cache server speaks the full protocol; an object store answers the same
255/// lookups without the extensions built on top of it, and reports that by
256/// declining them rather than by failing.
257pub struct RemoteCacheClient {
258    backend: Backend,
259}
260
261/// The service a [`RemoteCacheClient`] talks to.
262///
263/// Backends are a closed set, so they dispatch through an enum rather than a
264/// trait: an `async fn` in a trait is not object-safe, and boxing every call to
265/// work around that would buy nothing here.
266enum Backend {
267    Http(HttpRemoteCache),
268    S3(S3RemoteCache),
269}
270
271impl RemoteCacheClient {
272    /// Construct a client and validate its URL and authentication settings.
273    pub fn new(config: RemoteCacheConfig) -> Result<Self> {
274        Ok(Self {
275            backend: Backend::Http(HttpRemoteCache::new(config)?),
276        })
277    }
278
279    /// Construct a client backed directly by an S3-compatible object store.
280    ///
281    /// The store answers the same lookups a cache server does, without the
282    /// extensions built on top of the protocol. See [`S3RemoteCacheConfig`].
283    pub fn new_s3(config: S3RemoteCacheConfig) -> Result<Self> {
284        Ok(Self {
285            backend: Backend::S3(S3RemoteCache::new(config)?),
286        })
287    }
288
289    /// Connect to the service, authenticate, and negotiate protocol capabilities.
290    ///
291    /// This performs no cache reads or writes. It is intended for diagnostics
292    /// that need to distinguish a valid client configuration from a reachable,
293    /// compatible remote cache.
294    pub async fn check_connection(&self) -> Result<()> {
295        match &self.backend {
296            Backend::Http(client) => client.check_connection().await,
297            Backend::S3(store) => store.check_connection().await,
298        }
299    }
300
301    /// Download verified CAS objects using the server's negotiated blob-pack extension.
302    ///
303    /// `None` means the service does not support blob packs. Objects omitted by a
304    /// supported server are absent from `blobs`, so callers can retry them through
305    /// the ordinary single-blob endpoint.
306    pub async fn get_blob_pack(
307        &self,
308        digests: &[CacheDigest],
309        staging_dir: &Path,
310    ) -> Result<Option<RemoteBlobPack>> {
311        match &self.backend {
312            Backend::Http(client) => client.get_blob_pack(digests, staging_dir).await,
313            Backend::S3(store) => store.get_blob_pack(digests, staging_dir).await,
314        }
315    }
316
317    pub(crate) async fn get_blob_pack_with_limit(
318        &self,
319        digests: &[CacheDigest],
320        staging_dir: &Path,
321        max_bytes: u64,
322    ) -> Result<Option<RemoteBlobPack>> {
323        match &self.backend {
324            Backend::Http(client) => {
325                client
326                    .get_blob_pack_with_limit(digests, staging_dir, max_bytes)
327                    .await
328            }
329            Backend::S3(store) => {
330                store
331                    .get_blob_pack_with_limit(digests, staging_dir, max_bytes)
332                    .await
333            }
334        }
335    }
336
337    /// Fetch and validate an action-result record, returning `None` on a miss.
338    pub async fn get_action_result(
339        &self,
340        action: &CacheDigest,
341    ) -> Result<Option<RemoteActionResult>> {
342        match &self.backend {
343            Backend::Http(client) => client.get_action_result(action).await,
344            Backend::S3(store) => store.get_action_result(action).await,
345        }
346    }
347
348    /// How many actions one batched lookup may ask about.
349    ///
350    /// `None` means the service does not answer batched lookups.
351    pub(crate) async fn action_batch_limit(&self) -> Result<Option<usize>> {
352        match &self.backend {
353            Backend::Http(client) => client.action_batch_limit().await,
354            Backend::S3(store) => store.action_batch_limit().await,
355        }
356    }
357
358    /// Look up several action results in one request.
359    ///
360    /// `None` means the service does not answer batched lookups, leaving the
361    /// caller to ask for each action individually. The response carries only the
362    /// results the service holds, in no particular order, so each record is bound
363    /// to its request by the action digest it names rather than by position.
364    pub async fn get_action_results(
365        &self,
366        actions: &[CacheDigest],
367    ) -> Result<Option<Vec<RemoteActionResult>>> {
368        match &self.backend {
369            Backend::Http(client) => client.get_action_results(actions).await,
370            Backend::S3(store) => store.get_action_results(actions).await,
371        }
372    }
373
374    /// Canonically serialize and store an action-result record.
375    pub async fn put_action_result(&self, result: &RemoteActionResult) -> Result<()> {
376        match &self.backend {
377            Backend::Http(client) => client.put_action_result(result).await,
378            Backend::S3(store) => store.put_action_result(result).await,
379        }
380    }
381
382    /// Atomically join or claim a server-wide compilation promise.
383    ///
384    /// `None` means this backend does not support ephemeral coordination.
385    pub async fn join_action_promise(
386        &self,
387        invocation: &CacheDigest,
388        adapter: &str,
389    ) -> Result<Option<ActionPromiseState>> {
390        match &self.backend {
391            Backend::Http(client) => client.join_action_promise(invocation, adapter).await,
392            Backend::S3(_) => Ok(None),
393        }
394    }
395
396    /// Complete a claimed promise after its action result has been published.
397    ///
398    /// `false` means this backend does not support ephemeral coordination.
399    pub async fn complete_action_promise(
400        &self,
401        invocation: &CacheDigest,
402        completion: &ActionPromiseCompletion,
403    ) -> Result<bool> {
404        match &self.backend {
405            Backend::Http(client) => client.complete_action_promise(invocation, completion).await,
406            Backend::S3(_) => Ok(false),
407        }
408    }
409
410    /// Fetch a task action manifest and the entity tag needed to update it.
411    pub async fn get_action_manifest(
412        &self,
413        key: &CacheDigest,
414    ) -> Result<Option<RemoteActionManifest>> {
415        match &self.backend {
416            Backend::Http(client) => client.get_action_manifest(key).await,
417            Backend::S3(store) => store.get_action_manifest(key).await,
418        }
419    }
420
421    /// Store a task action manifest, optionally requiring an entity-tag match.
422    pub async fn put_action_manifest(
423        &self,
424        key: &CacheDigest,
425        bytes: &[u8],
426        expected_etag: Option<&str>,
427    ) -> Result<ManifestPutOutcome> {
428        match &self.backend {
429            Backend::Http(client) => client.put_action_manifest(key, bytes, expected_etag).await,
430            Backend::S3(store) => store.put_action_manifest(key, bytes, expected_etag).await,
431        }
432    }
433
434    /// Download a small blob into memory and verify its digest.
435    pub async fn get_blob(
436        &self,
437        digest: &CacheDigest,
438        media_type: &'static str,
439    ) -> Result<Vec<u8>> {
440        match &self.backend {
441            Backend::Http(client) => client.get_blob(digest, media_type).await,
442            Backend::S3(store) => store.get_blob(digest, media_type).await,
443        }
444    }
445
446    /// Download a blob to a temporary file and verify its digest.
447    pub async fn get_blob_file(
448        &self,
449        digest: &CacheDigest,
450        staging_dir: &Path,
451    ) -> Result<tempfile::NamedTempFile> {
452        match &self.backend {
453            Backend::Http(client) => client.get_blob_file(digest, staging_dir).await,
454            Backend::S3(store) => store.get_blob_file(digest, staging_dir).await,
455        }
456    }
457
458    /// How many blobs, and how many payload bytes, one uploaded pack may carry.
459    ///
460    /// `None` means the service does not accept packed uploads.
461    pub(crate) async fn blob_pack_upload_limits(&self) -> Result<Option<BlobPackLimits>> {
462        match &self.backend {
463            Backend::Http(client) => client.blob_pack_upload_limits().await,
464            Backend::S3(store) => store.blob_pack_upload_limits().await,
465        }
466    }
467
468    /// Upload several content-addressed blobs in one framed request.
469    ///
470    /// `None` means the service does not accept packed uploads, leaving the
471    /// caller to send each blob on its own. A rejected pack is reported as an
472    /// error and publishes nothing the caller may rely on: a server verifies
473    /// each frame as it arrives, so an accepted prefix may exist, but every blob
474    /// is content-addressed and storing one twice is not an error.
475    pub async fn put_blob_pack(&self, uploads: &[BlobUpload]) -> Result<Option<BlobPackReceipt>> {
476        match &self.backend {
477            Backend::Http(client) => client.put_blob_pack(uploads).await,
478            Backend::S3(store) => store.put_blob_pack(uploads).await,
479        }
480    }
481
482    /// Verify and upload a content-addressed blob.
483    pub async fn put_blob(&self, upload: &BlobUpload) -> Result<()> {
484        match &self.backend {
485            Backend::Http(client) => client.put_blob(upload).await,
486            Backend::S3(store) => store.put_blob(upload).await,
487        }
488    }
489}
490
491/// Buffer a JSON response body, refusing to grow past [`MAX_REMOTE_JSON_BYTES`].
492///
493/// A declared `Content-Length` is rejected up front so an oversized body costs
494/// nothing to refuse; the streaming check then covers servers that understate or
495/// omit it.
496async fn read_bounded_json(response: reqwest::Response, what: &str) -> Result<Vec<u8>> {
497    read_json_within(response, what, MAX_REMOTE_JSON_BYTES).await
498}
499
500async fn read_json_within(response: reqwest::Response, what: &str, limit: u64) -> Result<Vec<u8>> {
501    if let Some(length) = response.content_length()
502        && length > limit
503    {
504        bail!("remote cache {what} declared {length} bytes, over the {limit} byte limit");
505    }
506    let mut response = response;
507    let mut bytes = Vec::new();
508    while let Some(chunk) = response.chunk().await? {
509        if bytes.len() as u64 + chunk.len() as u64 > limit {
510            bail!("remote cache {what} exceeded the {limit} byte limit");
511        }
512        bytes.extend_from_slice(&chunk);
513    }
514    Ok(bytes)
515}
516/// Read the entity tag that a later conditional update has to send back.
517///
518/// The tag is carried through opaquely, and nothing may infer content from it.
519/// RFC 9110 section 8.8.3.3 requires an intermediary that re-encodes a response
520/// to vary the strong tag along with it, and proxies do: Caddy appends the
521/// content coding, so a manifest served through compression arrives tagged
522/// `"<hash>-zstd"`. What the body actually is gets established by the caller
523/// comparing it against canonical JSON, not by the shape of this header.
524fn parse_strong_etag(value: Option<&HeaderValue>) -> Result<String> {
525    let value = value
526        .and_then(|value| value.to_str().ok())
527        .ok_or_else(|| eyre!("remote action manifest response is missing an ETag"))?;
528    if value.starts_with("W/") {
529        // If-Match rejects a weak validator, so an update could not tell that it
530        // was overwriting a manifest someone else had published.
531        bail!("remote action manifest response has a weak ETag");
532    }
533    let etag = value
534        .strip_prefix('"')
535        .and_then(|value| value.strip_suffix('"'))
536        .filter(|value| is_entity_tag(value))
537        .ok_or_else(|| eyre!("remote action manifest response has an invalid ETag"))?;
538    Ok(etag.to_owned())
539}
540
541fn quoted_etag(etag: &str) -> Result<HeaderValue> {
542    if !is_entity_tag(etag) {
543        bail!("invalid remote action manifest ETag");
544    }
545    Ok(HeaderValue::from_str(&format!("\"{etag}\""))?)
546}
547
548/// Whether this is the opaque body of a strong entity tag (RFC 9110 `etagc`).
549///
550/// `HeaderValue::to_str` has already ruled out anything but visible ASCII, so
551/// the double quote that would end the tag early is all that is left to reject.
552fn is_entity_tag(value: &str) -> bool {
553    !value.is_empty()
554        && value.len() <= MAX_ETAG_BYTES
555        && value.bytes().all(|byte| matches!(byte, 0x21 | 0x23..=0x7e))
556}
557fn retry_delays(retries: i64) -> impl Iterator<Item = Duration> {
558    [200u64, 1_000, 4_000, 15_000]
559        .into_iter()
560        .chain(std::iter::repeat(15_000))
561        .map(Duration::from_millis)
562        .map(|duration| {
563            let factor = 0.5 + rand::random::<f64>() * 0.5;
564            Duration::from_secs_f64(duration.as_secs_f64() * factor)
565        })
566        .take(retries.max(0) as usize)
567}
568
569/// hyper-util exposes DNS failures in the error chain as a `dns error` source,
570/// but reqwest intentionally erases the concrete connector type. Match that
571/// stable connector error label rather than platform-specific resolver text.
572fn is_dns_error(error: &(dyn std::error::Error + 'static)) -> bool {
573    let mut current = Some(error);
574    while let Some(source) = current {
575        if source.to_string() == "dns error" {
576            return true;
577        }
578        current = source.source();
579    }
580    false
581}
582
583/// A failure a backend has identified as worth retrying.
584///
585/// [`is_transient`] recognizes the statuses that are transient for any HTTP
586/// service. A backend that knows one of its own -- S3 answers `409` while a
587/// concurrent conditional write to the same key is in flight, and asks that it
588/// be retried -- attaches this instead of teaching that function about it.
589#[derive(Debug)]
590pub(crate) struct TransientRequest(pub(crate) &'static str);
591
592impl std::fmt::Display for TransientRequest {
593    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
594        formatter.write_str(self.0)
595    }
596}
597
598impl std::error::Error for TransientRequest {}
599
600fn is_transient(error: &eyre::Report) -> bool {
601    // An unavailable hostname is a deterministic configuration error. reqwest
602    // categorizes it as a connect error, but retrying only delays the diagnosis.
603    if is_dns_error(error.as_ref()) {
604        return false;
605    }
606    error.chain().any(|source| {
607        if source.downcast_ref::<TransientRequest>().is_some() {
608            return true;
609        }
610        let Some(error) = source.downcast_ref::<reqwest::Error>() else {
611            return false;
612        };
613        if error.is_timeout() || error.is_connect() || error.is_body() {
614            return true;
615        }
616        error.status().is_some_and(|status| {
617            let status = status.as_u16();
618            status == 408 || status == 429 || (500..600).contains(&status)
619        })
620    })
621}
622
623async fn retry_async<F, Fut, T>(verb: &str, url: &Url, retries: i64, mut operation: F) -> Result<T>
624where
625    F: FnMut() -> Fut,
626    Fut: std::future::Future<Output = Result<T>>,
627{
628    let mut delays = retry_delays(retries);
629    let mut attempt = 1;
630    loop {
631        let started_at = Instant::now();
632        match operation().await {
633            Ok(value) => return Ok(value),
634            Err(error) if is_transient(&error) => {
635                let Some(delay) = delays.next() else {
636                    return Err(error);
637                };
638                warn!(
639                    "HTTP {verb} {url} attempt {attempt} failed after {:?} (transient): {error}; retrying in {delay:?}",
640                    started_at.elapsed()
641                );
642                tokio::time::sleep(delay).await;
643                attempt += 1;
644            }
645            Err(error) => return Err(error),
646        }
647    }
648}
mbx_cache_core/lib.rs

mbx_cache_core/
lib.rs
