Use rust-s3 fork with asyncwrite support

Cargo.lock

@@ -1754,8 +1754,7 @@ dependencies = [
 [[package]]
 name = "rust-s3"
 version = "0.27.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b2f26775d15f43dc848ef0ec65f83de8775549e486c7a3a576652049a7122d32"
+source = "git+https://github.com/asonix/rust-s3?branch=asonix/use-async-write#f61e422d77484288cf32e40896c8b361eb60afac"
 dependencies = [
  "anyhow",
  "async-trait",

Cargo.toml

@@ -33,7 +33,7 @@ opentelemetry = { version = "0.16", features = ["rt-tokio"] }
 opentelemetry-otlp = "0.9"
 pin-project-lite = "0.2.7"
 reqwest = { version = "0.11.5", default-features = false, features = ["stream"], optional = true}
-rust-s3 = { version = "0.27.0", default-features = false, features = ["tokio-rustls-tls"], optional = true }
+rust-s3 = { version = "0.27.0", default-features = false, features = ["tokio-rustls-tls"], optional = true, git = "https://github.com/asonix/rust-s3", branch = "asonix/use-async-write" }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sha2 = "0.9.0"

src/store.rs

@@ -46,7 +46,7 @@ pub(crate) trait Store: Send + Sync + Clone + Debug + 'static {
         writer: &mut Writer,
     ) -> Result<(), std::io::Error>
     where
-        Writer: AsyncWrite + Unpin;
+        Writer: AsyncWrite + Send + Unpin;
 
     async fn len(&self, identifier: &Self::Identifier) -> Result<u64, Self::Error>;
 

src/store/file_store.rs

@@ -108,7 +108,7 @@ impl Store for FileStore {
         writer: &mut Writer,
     ) -> Result<(), std::io::Error>
     where
-        Writer: AsyncWrite + Unpin,
+        Writer: AsyncWrite + Send + Unpin,
     {
         let path = self.path_from_file_id(identifier);
 

src/store/object_store.rs

@@ -1,6 +1,6 @@
 use crate::store::Store;
 use actix_web::web::Bytes;
-use futures_util::stream::{Stream, StreamExt};
+use futures_util::stream::Stream;
 use s3::{
     command::Command, creds::Credentials, request::Reqwest, request_trait::Request, Bucket, Region,
 };
@@ -10,7 +10,7 @@ use std::{
     task::{Context, Poll},
 };
 use storage_path_generator::{Generator, Path};
-use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
+use tokio::io::{AsyncRead, AsyncWrite};
 use uuid::Uuid;
 
 mod object_id;
@@ -109,18 +109,14 @@ impl Store for ObjectStore {
         writer: &mut Writer,
     ) -> Result<(), std::io::Error>
     where
-        Writer: AsyncWrite + Unpin,
+        Writer: AsyncWrite + Send + Unpin,
     {
-        let mut stream = self
-            .to_stream(identifier, None, None)
+        let path = identifier.as_str();
+
+        self.bucket
+            .get_object_stream(path, writer)
             .await
-            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
-
-        while let Some(res) = stream.next().await {
-            let mut bytes = res?;
-
-            writer.write_all_buf(&mut bytes).await?;
-        }
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, Self::Error::from(e)))?;
 
         Ok(())
     }