chore: migrate to object_store

Move to the more modern and maintained object_store
library to access generic buckets.

We should be able also to do streaming of files now,
and proceed by implementing paginated results for listings
if we want.

Fixes #1.

Cargo.toml

@@ -21,14 +21,14 @@ edition = "2024"
 
 [dependencies]
 anyhow = "1.0"
+bytes = "1.10"
+futures = "0.3"
 human-size = "0.4"
 lazy_static = "1.4"
 log = "0.4"
 rocket = "0.5"
 rocket_dyn_templates = { version = "0.2.0", features = ["tera"] }
-rust-s3 = { version = "0.35", default-features = false, features = [
-    "tokio-rustls-tls",
-] }
+object_store = { version = "0.12", features = ["aws"] }
 serde = "1.0"
 tempfile = "3.20"
 
@@ -38,7 +38,7 @@ futures = "0.3"
 libc = "0.2"
 regex = "1.11"
 reqwest = "0.12"
-rstest = "0.25"
+rstest = "0.26"
 scraper = "0.23"
 test-log = "0.2"
 testcontainers = "0.24"

src/main.rs

@@ -6,16 +6,19 @@ mod sizes;
 
 use {
     anyhow::Result,
+    bytes::Bytes,
+    futures::{StreamExt, stream::BoxStream},
     lazy_static::lazy_static,
+    object_store::path::Path,
     rocket::{
-        State,
+        Request, State,
         fairing::AdHoc,
         figment::{
             Profile,
             providers::{Env, Format as _, Toml},
         },
         http::uri::Origin,
-        response::{Redirect, Responder},
+        response::{self, Redirect, Responder, stream::ByteStream},
         serde::Serialize,
     },
     rocket_dyn_templates::{Template, context},
@@ -23,17 +26,22 @@ use {
     std::path::PathBuf,
 };
 
-#[derive(Responder)]
 enum FileView {
-    #[response(content_type = "text/html")]
     Folder(Template),
-
-    #[response(content_type = "application/octet-stream")]
-    File(Vec<u8>),
-
+    File(ByteStream<BoxStream<'static, Bytes>>),
     Redirect(Redirect),
 }
 
+impl<'r> Responder<'r, 'r> for FileView {
+    fn respond_to(self, req: &'r Request<'_>) -> response::Result<'r> {
+        match self {
+            Self::Folder(template) => template.respond_to(req),
+            Self::File(stream) => stream.respond_to(req),
+            Self::Redirect(redirect) => redirect.respond_to(req),
+        }
+    }
+}
+
 #[derive(Serialize)]
 struct FileViewItem {
     path: String,
@@ -68,7 +76,7 @@ async fn index(
        We try first to retrieve list an object as a file. If we fail,
        we fallback to retrieving the equivalent folder.
     */
-    if let Ok(result) = s3_serve_file(&path, &state).await {
+    if let Ok(result) = s3_serve_file(path.clone(), &state).await {
         Ok(result)
     } else {
         // We need to redirect to a path ending with a slash as
@@ -91,28 +99,41 @@ async fn index(
     }
 }
 
-async fn s3_serve_file(path: &PathBuf, settings: &Settings) -> Result<FileView, Error> {
+async fn s3_serve_file(path: PathBuf, settings: &Settings) -> Result<FileView, Error> {
     let is_root_prefix = path.as_os_str().is_empty();
     if is_root_prefix {
-        return Err(Error::NotFound("Root prefix is not a file".into()));
+        return Err(Error::UnknownError("root prefix is not a file".into()));
     }
 
-    // FIXME: this can be big, we should use streaming,
-    // not loading in memory!
-    let response = settings
+    let s3_path = object_store::path::Path::from(path.to_str().unwrap());
+
+    let object_stream = settings
         .s3_bucket
-        .get_object(format!("{}", path.display()))
+        .get(&s3_path)
         .await
-        .map_err(|_| Error::UnknownError("Unable to connect to S3 bucket".into()))?;
+        .map_err(|e| match e {
+            object_store::Error::NotFound { path: _, source: _ } => {
+                Error::UnknownError(e.to_string())
+            }
+            _ => Error::UnknownError(e.to_string()),
+        })?
+        .into_stream();
 
-    match response.status_code() {
-        200 | 204 => {
-            let bytes = response.bytes().to_vec();
-            Ok(FileView::File(bytes))
-        }
-        404 => Err(Error::NotFound("Object not found".into())),
-        _ => Err(Error::UnknownError("Unknown S3 error".into())),
+    let stream = object_stream
+        .map(move |chunk| match chunk {
+            Ok(bytes) => bytes,
+            Err(err) => {
+                log::error!("connection error while reading {}: {}", path.display(), err);
+                Bytes::new() // Forces end of stream
             }
+        })
+        .boxed();
+
+    // TODO: unfortunately Rocket does not have a ByteStream with a Result per chunk,
+    // meaning that if there is a failure in the middle of the stream, the best we can do is...
+    // nothing? Panic? All options are bad.
+
+    Ok(FileView::File(ByteStream::from(stream)))
 }
 
 async fn s3_fileview(path: &PathBuf, settings: &Settings) -> Result<Vec<FileViewItem>, Error> {
@@ -122,59 +143,37 @@ async fn s3_fileview(path: &PathBuf, settings: &Settings) -> Result<Vec<FileView
         - files will be under the 'contents' property
     */
 
-    let parent = path.parent();
-    let s3_folder_path = match parent {
-        Some(_) => format!("{}/", path.display()),
-        None => "".into(),
-    };
-
+    let s3_folder_path = path.to_str().map(|p| Path::from(p));
     let s3_objects = settings
         .s3_bucket
-        .list(s3_folder_path, Some("/".into()))
+        .list_with_delimiter(s3_folder_path.as_ref())
         .await
-        .map_err(|_| Error::NotFound("Object not found".into()))?;
+        .map_err(|_| Error::NotFound("object not found".into()))?;
 
-    let objects = s3_objects
-        .iter()
-        .flat_map(|list| -> Vec<Option<FileViewItem>> {
-            let prefix = if let Some(p) = &list.prefix {
-                p.as_str()
-            } else {
-                ""
-            };
-
-            let folders = list.common_prefixes.iter().flatten().map(|dir| {
-                let path = dir.prefix.strip_prefix(&prefix);
-                path.map(|path| FileViewItem {
-                    path: path.to_owned(),
+    let folders = s3_objects.common_prefixes.into_iter().map(|dir| {
+        let dirname = dir.parts().last().unwrap();
+        FileViewItem {
+            path: dirname.as_ref().into(),
             size_bytes: 0,
             size: "[DIR]".to_owned(),
             last_modification: String::default(),
-                })
+        }
     });
 
-            let files = list.contents.iter().map(|obj| {
-                let path = obj.key.strip_prefix(&prefix);
-                path.map(|path| FileViewItem {
-                    path: path.to_owned(),
+    let files = s3_objects.objects.into_iter().map(|obj| FileViewItem {
+        path: obj.location.filename().unwrap().into(),
         size_bytes: obj.size,
         size: sizes::bytes_to_human(obj.size),
-                    last_modification: obj.last_modified.clone(),
-                })
+        last_modification: obj.last_modified.to_rfc3339(),
     });
 
-            folders.chain(files).collect()
-        })
-        .flatten()
-        .collect();
-
-    Ok(objects)
+    Ok(folders.chain(files).collect())
 }
 
 lazy_static! {
     // Workaround for https://github.com/SergioBenitez/Rocket/issues/1792
     static ref EMPTY_DIR: tempfile::TempDir = tempfile::tempdir()
-        .expect("Unable to create an empty temporary folder, is the whole FS read-only?");
+        .expect("unable to create an empty temporary folder, is the whole FS read-only?");
 }
 
 #[rocket::launch]

src/settings.rs

@@ -1,16 +1,20 @@
 // SPDX-FileCopyrightText: © Matteo Settenvini <matteo.settenvini@montecristosoftware.eu>
 // SPDX-License-Identifier: EUPL-1.2
 
-use {anyhow::anyhow, rocket::serde::Deserialize, serde::de::Error};
+use {
+    object_store::{ObjectStore, aws},
+    rocket::serde::Deserialize,
+    serde::de::Error,
+};
 
 #[derive(Deserialize)]
 #[serde(crate = "rocket::serde")]
 pub struct Settings {
     #[serde(deserialize_with = "deserialize_s3_bucket")]
-    pub s3_bucket: Box<s3::Bucket>,
+    pub s3_bucket: Box<dyn ObjectStore>,
 }
 
-fn deserialize_s3_bucket<'de, D>(deserializer: D) -> Result<Box<s3::Bucket>, D::Error>
+fn deserialize_s3_bucket<'de, D>(deserializer: D) -> Result<Box<dyn ObjectStore>, D::Error>
 where
     D: serde::Deserializer<'de>,
 {
@@ -31,37 +35,26 @@ pub struct S3Config {
     pub secret_access_key: String,
 }
 
-impl TryInto<Box<s3::Bucket>> for S3Config {
+impl TryInto<Box<dyn ObjectStore>> for S3Config {
     type Error = anyhow::Error;
 
-    fn try_into(self) -> Result<Box<s3::Bucket>, Self::Error> {
-        let region = s3::Region::Custom {
-            region: self.region,
-            endpoint: self.endpoint,
-        };
-
-        let credentials = s3::creds::Credentials::new(
-            Some(&self.access_key_id),
-            Some(&self.secret_access_key),
-            None,
-            None,
-            None,
-        )?;
+    fn try_into(self) -> Result<Box<dyn ObjectStore>, Self::Error> {
+        // TODO: support object stores other than than AWS
+        let object_store = aws::AmazonS3Builder::new()
+            .with_region(self.region)
+            .with_endpoint(&self.endpoint)
+            .with_bucket_name(&self.name)
+            .with_access_key_id(self.access_key_id)
+            .with_secret_access_key(self.secret_access_key)
+            .with_virtual_hosted_style_request(!self.path_style)
+            .build()?;
 
         log::info!(
             "Serving contents from bucket {} at {}",
-            &self.name,
-            region.endpoint()
+            self.endpoint,
+            self.name,
         );
 
-        let bucket = s3::Bucket::new(&self.name, region, credentials).map_err(|e| anyhow!(e));
-        if self.path_style {
-            bucket.map(|mut b| {
-                b.set_path_style();
-                b
-            })
-        } else {
-            bucket
-        }
+        Ok(Box::new(object_store))
     }
 }

tests/common/mod.rs

@@ -5,6 +5,7 @@ mod minio;
 
 use {
     anyhow::{Result, anyhow},
+    object_store::ObjectStore,
     reqwest::Url,
     std::{ptr::null_mut, str::FromStr},
     testcontainers::{ContainerAsync, runners::AsyncRunner},
@@ -13,7 +14,7 @@ use {
 
 pub struct Test {
     pub base_url: Url,
-    pub bucket: Box<s3::Bucket>,
+    pub bucket: Box<dyn ObjectStore>,
     pub serves3: tokio::process::Child,
     pub _minio: ContainerAsync<minio::MinIO>,
 }