getsentry · Swatinem · Jan 9, 2023 · Dec 30, 2022
diff --git a/crates/symbolicator-service/src/cache.rs b/crates/symbolicator-service/src/cache.rs
@@ -192,7 +192,7 @@ impl CacheError {
 
 /// An entry in a cache, containing either `Ok(T)` or an error denoting the reason why an
 /// object could not be fetched or is otherwise unusable.
-pub type CacheEntry<T> = Result<T, CacheError>;
+pub type CacheEntry<T = ()> = Result<T, CacheError>;
 
 /// Parses a [`CacheEntry`] from a [`ByteView`](ByteView).
 pub fn cache_entry_from_bytes(bytes: ByteView<'static>) -> CacheEntry<ByteView<'static>> {

diff --git a/crates/symbolicator-service/src/services/cficaches.rs b/crates/symbolicator-service/src/services/cficaches.rs
@@ -196,7 +196,7 @@ impl CfiCacheActor {
 /// The source file is probably an executable or so, the resulting file is in the format of
 /// [`CfiCache`].
 #[tracing::instrument(skip_all)]
-fn write_cficache(file: &mut File, object_handle: &ObjectHandle) -> CacheEntry<()> {
+fn write_cficache(file: &mut File, object_handle: &ObjectHandle) -> CacheEntry {
     object_handle.configure_scope();
 
     tracing::debug!("Converting cficache for {}", object_handle.cache_key);

diff --git a/crates/symbolicator-service/src/services/derived.rs b/crates/symbolicator-service/src/services/derived.rs
@@ -65,8 +65,9 @@ where
                 | CacheError::PermissionDenied(_)
                 | CacheError::Timeout(_)
                 | CacheError::DownloadError(_) => {
-                    // FIXME(swatinem): all the download errors so far lead to `None`
-                    // previously. we should probably decide on a better solution here.
+                    // NOTE: all download related errors are already exposed as the candidates
+                    // `ObjectDownloadInfo`. It is not necessary to duplicate that into the
+                    // `ObjectUseInfo`.
                     ObjectUseInfo::None
                 }
                 _ => ObjectUseInfo::Malformed,

diff --git a/crates/symbolicator-service/src/services/download/filesystem.rs b/crates/symbolicator-service/src/services/download/filesystem.rs
@@ -9,7 +9,7 @@ use tokio::fs;
 
 use symbolicator_sources::FilesystemRemoteFile;
 
-use super::{DownloadError, DownloadStatus};
+use crate::cache::{CacheEntry, CacheError};
 
 /// Downloader implementation that supports the filesystem source.
 #[derive(Debug)]
@@ -25,16 +25,17 @@ impl FilesystemDownloader {
         &self,
         file_source: FilesystemRemoteFile,
         dest: &Path,
-    ) -> Result<DownloadStatus<()>, DownloadError> {
+    ) -> CacheEntry {
         // All file I/O in this function is blocking!
         let abspath = file_source.path();
         tracing::debug!("Fetching debug file from {:?}", abspath);
-        match fs::copy(abspath, dest).await {
-            Ok(_) => Ok(DownloadStatus::Completed(())),
-            Err(e) => match e.kind() {
-                io::ErrorKind::NotFound => Ok(DownloadStatus::NotFound),
-                _ => Err(DownloadError::Io(e)),
-            },
-        }
+
+        fs::copy(abspath, dest)
+            .await
+            .map(|_| ())
+            .map_err(|e| match e.kind() {
+                io::ErrorKind::NotFound => CacheError::NotFound,
+                _ => e.into(),
+            })
     }
 }
diff --git a/crates/symbolicator-service/src/services/download/gcs.rs b/crates/symbolicator-service/src/services/download/gcs.rs
@@ -2,17 +2,12 @@
 
 use std::num::NonZeroUsize;
 use std::path::Path;
-use std::sync::Arc;
-
-use futures::prelude::*;
-use parking_lot::Mutex;
-use reqwest::{header, Client, StatusCode};
+use std::sync::{Arc, Mutex};
 
 use symbolicator_sources::{GcsRemoteFile, GcsSourceKey, RemoteFile};
 
-use crate::utils::gcs::{self, request_new_token, GcsError, GcsToken};
-
-use super::{content_length_timeout, DownloadError, DownloadStatus};
+use crate::cache::CacheEntry;
+use crate::utils::gcs::{self, GcsError, GcsToken};
 
 /// An LRU cache for GCS OAuth tokens.
 type GcsTokenCache = lru::LruCache<Arc<GcsSourceKey>, Arc<GcsToken>>;
@@ -28,7 +23,7 @@ pub struct GcsDownloader {
 
 impl GcsDownloader {
     pub fn new(
-        client: Client,
+        client: reqwest::Client,
         connect_timeout: std::time::Duration,
         streaming_timeout: std::time::Duration,
         token_capacity: NonZeroUsize,
@@ -46,33 +41,30 @@ impl GcsDownloader {
     /// If the cache contains a valid token, then this token is returned. Otherwise, a new token is
     /// requested from GCS and stored in the cache.
     async fn get_token(&self, source_key: &Arc<GcsSourceKey>) -> Result<Arc<GcsToken>, GcsError> {
-        if let Some(token) = self.token_cache.lock().get(source_key) {
+        if let Some(token) = self.token_cache.lock().unwrap().get(source_key) {
             if !token.is_expired() {
                 metric!(counter("source.gcs.token.cached") += 1);
                 return Ok(token.clone());
             }
         }
 
         let source_key = source_key.clone();
-        let token = request_new_token(&self.client, &source_key).await?;
+        let token = gcs::request_new_token(&self.client, &source_key).await?;
         metric!(counter("source.gcs.token.requests") += 1);
         let token = Arc::new(token);
-        self.token_cache.lock().put(source_key, token.clone());
+        self.token_cache
+            .lock()
+            .unwrap()
+            .put(source_key, token.clone());
         Ok(token)
     }
 
     /// Downloads a source hosted on GCS.
-    ///
-    /// # Directly thrown errors
-    /// - [`GcsError::InvalidUrl`]
-    /// - [`DownloadError::Reqwest`]
-    /// - [`DownloadError::Rejected`]
-    /// - [`DownloadError::Canceled`]
     pub async fn download_source(
         &self,
         file_source: GcsRemoteFile,
         destination: &Path,
-    ) -> Result<DownloadStatus<()>, DownloadError> {
+    ) -> CacheEntry {
         let key = file_source.key();
         let bucket = file_source.source.bucket.clone();
         tracing::debug!("Fetching from GCS: {} (from {})", &key, bucket);
@@ -85,66 +77,16 @@ impl GcsDownloader {
         let request = self
             .client
             .get(url.clone())
-            .header("authorization", token.bearer_token())
-            .send();
-        let request = tokio::time::timeout(self.connect_timeout, request);
-        let request = super::measure_download_time(source.source_metric_key(), request);
-
-        let response = request
-            .await
-            .map_err(|_| DownloadError::Canceled)? // Timeout
-            .map_err(|e| {
-                tracing::debug!(
-                    "Skipping response from GCS {} (from {}): {}",
-                    &key,
-                    &bucket,
-                    &e
-                );
-                DownloadError::Reqwest(e)
-            })?;
-
-        if response.status().is_success() {
-            tracing::trace!("Success hitting GCS {} (from {})", &key, bucket);
-
-            let content_length = response
-                .headers()
-                .get(header::CONTENT_LENGTH)
-                .and_then(|hv| hv.to_str().ok())
-                .and_then(|s| s.parse::<i64>().ok());
-
-            let timeout =
-                content_length.map(|cl| content_length_timeout(cl, self.streaming_timeout));
-            let stream = response.bytes_stream().map_err(DownloadError::Reqwest);
-
-            super::download_stream(&source, stream, destination, timeout).await
-        } else if matches!(
-            response.status(),
-            StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED
-        ) {
-            tracing::debug!(
-                "Insufficient permissions to download from GCS {} (from {})",
-                &key,
-                &bucket,
-            );
-            Ok(DownloadStatus::PermissionDenied)
-        // If it's a client error, chances are either it's a 404 or it's permission-related.
-        } else if response.status().is_client_error() {
-            tracing::debug!(
-                "Unexpected client error status code from GCS {} (from {}): {}",
-                &key,
-                &bucket,
-                response.status()
-            );
-            Ok(DownloadStatus::NotFound)
-        } else {
-            tracing::debug!(
-                "Unexpected status code from GCS {} (from {}): {}",
-                &key,
-                &bucket,
-                response.status()
-            );
-            Err(DownloadError::Rejected(response.status()))
-        }
+            .header("authorization", token.bearer_token());
+
+        super::download_reqwest(
+            &source,
+            request,
+            self.connect_timeout,
+            self.streaming_timeout,
+            destination,
+        )
+        .await
     }
 }
 
@@ -157,8 +99,10 @@ mod tests {
         SourceLocation,
     };
 
+    use crate::cache::CacheError;
     use crate::test;
 
+    use reqwest::Client;
     use sha1::{Digest as _, Sha1};
 
     fn gcs_source(source_key: GcsSourceKey) -> Arc<GcsSourceConfig> {
@@ -190,12 +134,9 @@ mod tests {
         let source_location = SourceLocation::new("e5/14c9464eed3be5943a2c61d9241fad/executable");
         let file_source = GcsRemoteFile::new(source, source_location);
 
-        let download_status = downloader
-            .download_source(file_source, &target_path)
-            .await
-            .unwrap();
+        let download_status = downloader.download_source(file_source, &target_path).await;
 
-        assert!(matches!(download_status, DownloadStatus::Completed(_)));
+        assert!(download_status.is_ok());
         assert!(target_path.exists());
 
         let hash = Sha1::digest(std::fs::read(target_path).unwrap());
@@ -221,12 +162,9 @@ mod tests {
         let source_location = SourceLocation::new("does/not/exist");
         let file_source = GcsRemoteFile::new(source, source_location);
 
-        let download_status = downloader
-            .download_source(file_source, &target_path)
-            .await
-            .unwrap();
+        let download_status = downloader.download_source(file_source, &target_path).await;
 
-        assert!(matches!(download_status, DownloadStatus::NotFound));
+        assert_eq!(download_status, Err(CacheError::NotFound));
         assert!(!target_path.exists());
     }
 

diff --git a/crates/symbolicator-service/src/services/download/http.rs b/crates/symbolicator-service/src/services/download/http.rs
@@ -3,13 +3,13 @@
 use std::path::Path;
 use std::time::Duration;
 
-use anyhow::Result;
-use futures::prelude::*;
-use reqwest::{header, Client, StatusCode};
+use reqwest::{header, Client};
 
 use symbolicator_sources::{HttpRemoteFile, RemoteFile};
 
-use super::{content_length_timeout, DownloadError, DownloadStatus, USER_AGENT};
+use crate::cache::{CacheEntry, CacheError};
+
+use super::USER_AGENT;
 
 /// Downloader implementation that supports the HTTP source.
 #[derive(Debug)]
@@ -29,20 +29,12 @@ impl HttpDownloader {
     }
 
     /// Downloads a source hosted on an HTTP server.
-    ///
-    /// # Directly thrown errors
-    /// - [`DownloadError::Reqwest`]
-    /// - [`DownloadError::Rejected`]
-    /// - [`DownloadError::Canceled`]
     pub async fn download_source(
         &self,
         file_source: HttpRemoteFile,
         destination: &Path,
-    ) -> Result<DownloadStatus<()>, DownloadError> {
-        let download_url = match file_source.url() {
-            Ok(x) => x,
-            Err(_) => return Ok(DownloadStatus::NotFound),
-        };
+    ) -> CacheEntry {
+        let download_url = file_source.url().map_err(|_| CacheError::NotFound)?;
 
         tracing::debug!("Fetching debug file from {}", download_url);
         let mut builder = self.client.get(download_url.clone());
@@ -53,55 +45,16 @@ impl HttpDownloader {
             }
         }
         let source = RemoteFile::from(file_source);
-        let request = builder.header(header::USER_AGENT, USER_AGENT).send();
-        let request = tokio::time::timeout(self.connect_timeout, request);
-        let request = super::measure_download_time(source.source_metric_key(), request);
-
-        let response = request
-            .await
-            .map_err(|_| DownloadError::Canceled)? // Timeout
-            .map_err(|e| {
-                tracing::debug!("Skipping response from {}: {}", download_url, e);
-                DownloadError::Reqwest(e)
-            })?;
-
-        if response.status().is_success() {
-            tracing::trace!("Success hitting {}", download_url);
-
-            let content_length = response
-                .headers()
-                .get(header::CONTENT_LENGTH)
-                .and_then(|hv| hv.to_str().ok())
-                .and_then(|s| s.parse::<i64>().ok());
-
-            let timeout =
-                content_length.map(|cl| content_length_timeout(cl, self.streaming_timeout));
-
-            let stream = response.bytes_stream().map_err(DownloadError::Reqwest);
-
-            super::download_stream(&source, stream, destination, timeout).await
-        } else if matches!(
-            response.status(),
-            StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED
-        ) {
-            tracing::debug!("Insufficient permissions to download from {}", download_url);
-            Ok(DownloadStatus::PermissionDenied)
-        // If it's a client error, chances are either it's a 404 or it's permission-related.
-        } else if response.status().is_client_error() {
-            tracing::debug!(
-                "Unexpected client error status code from {}: {}",
-                download_url,
-                response.status()
-            );
-            Ok(DownloadStatus::NotFound)
-        } else {
-            tracing::debug!(
-                "Unexpected status code from {}: {}",
-                download_url,
-                response.status()
-            );
-            Err(DownloadError::Rejected(response.status()))
-        }
+        let request = builder.header(header::USER_AGENT, USER_AGENT);
+
+        super::download_reqwest(
+            &source,
+            request,
+            self.connect_timeout,
+            self.streaming_timeout,
+            destination,
+        )
+        .await
     }
 }
 
@@ -133,9 +86,9 @@ mod tests {
             Duration::from_secs(30),
             Duration::from_secs(30),
         );
-        let download_status = downloader.download_source(file_source, dest).await.unwrap();
+        let download_status = downloader.download_source(file_source, dest).await;
 
-        assert!(matches!(download_status, DownloadStatus::Completed(_)));
+        assert!(download_status.is_ok());
 
         let content = std::fs::read_to_string(dest).unwrap();
         assert_eq!(content, "hello world\n");
@@ -161,8 +114,8 @@ mod tests {
             Duration::from_secs(30),
             Duration::from_secs(30),
         );
-        let download_status = downloader.download_source(file_source, dest).await.unwrap();
+        let download_status = downloader.download_source(file_source, dest).await;
 
-        assert!(matches!(download_status, DownloadStatus::NotFound));
+        assert_eq!(download_status, Err(CacheError::NotFound));
     }
 }