server/upload_path: Switch to AsyncBufRead

server/src/api/v1/upload_path.rs

@@ -61,7 +61,7 @@ enum ChunkData {
     Bytes(Bytes),
 
     /// A stream with a user-claimed hash and size that are potentially incorrect.
-    Stream(Box<dyn AsyncRead + Send + Unpin + 'static>, Hash, usize),
+    Stream(Box<dyn AsyncBufRead + Send + Unpin + 'static>, Hash, usize),
 }
 
 /// Result of a chunk upload.
@@ -185,7 +185,7 @@ async fn upload_path_dedup(
     username: Option<String>,
     cache: cache::Model,
     upload_info: UploadPathNarInfo,
-    stream: impl AsyncRead + Unpin,
+    stream: impl AsyncBufRead + Unpin,
     database: &DatabaseConnection,
     state: &State,
     existing_nar: NarGuard,
@@ -265,7 +265,7 @@ async fn upload_path_new(
     username: Option<String>,
     cache: cache::Model,
     upload_info: UploadPathNarInfo,
-    stream: impl AsyncRead + Send + Unpin + 'static,
+    stream: impl AsyncBufRead + Send + Unpin + 'static,
     database: &DatabaseConnection,
     state: &State,
 ) -> ServerResult<Json<UploadPathResult>> {
@@ -283,7 +283,7 @@ async fn upload_path_new_chunked(
     username: Option<String>,
     cache: cache::Model,
     upload_info: UploadPathNarInfo,
-    stream: impl AsyncRead + Send + Unpin + 'static,
+    stream: impl AsyncBufRead + Send + Unpin + 'static,
     database: &DatabaseConnection,
     state: &State,
 ) -> ServerResult<Json<UploadPathResult>> {
@@ -474,7 +474,7 @@ async fn upload_path_new_unchunked(
     username: Option<String>,
     cache: cache::Model,
     upload_info: UploadPathNarInfo,
-    stream: impl AsyncRead + Send + Unpin + 'static,
+    stream: impl AsyncBufRead + Send + Unpin + 'static,
     database: &DatabaseConnection,
     state: &State,
 ) -> ServerResult<Json<UploadPathResult>> {
@@ -587,7 +587,7 @@ async fn upload_chunk(
     {
         // There's an existing chunk matching the hash
         if require_proof_of_possession && !data.is_hash_trusted() {
-            let stream = data.into_async_read();
+            let stream = data.into_async_buf_read();
 
             let (mut stream, nar_compute) = HashReader::new(stream, Sha256::new());
             tokio::io::copy(&mut stream, &mut tokio::io::sink())
@@ -669,7 +669,7 @@ async fn upload_chunk(
 
     // Compress and stream to the storage backend
     let compressor = get_compressor_fn(compression_type, compression_level);
-    let mut stream = CompressionStream::new(data.into_async_read(), compressor);
+    let mut stream = CompressionStream::new(data.into_async_buf_read(), compressor);
 
     backend
         .upload_file(key, stream.stream())
@@ -773,8 +773,8 @@ impl ChunkData {
         matches!(self, ChunkData::Bytes(_))
     }
 
-    /// Turns the data into a stream.
-    fn into_async_read(self) -> Box<dyn AsyncRead + Unpin + Send> {
+    /// Turns the data into an AsyncBufRead.
+    fn into_async_buf_read(self) -> Box<dyn AsyncBufRead + Unpin + Send> {
         match self {
             Self::Bytes(bytes) => Box::new(Cursor::new(bytes)),
             Self::Stream(stream, _, _) => stream,

server/src/compression.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use digest::Output as DigestOutput;
 use sha2::{Digest, Sha256};
-use tokio::io::{AsyncRead, BufReader};
+use tokio::io::{AsyncBufRead, AsyncRead, BufReader};
 use tokio::sync::OnceCell;
 
 use attic::io::HashReader;
@@ -40,7 +40,7 @@ impl CompressionStream {
     /// Creates a new compression stream.
     pub fn new<R>(stream: R, compressor: CompressorFn<BufReader<HashReader<R, Sha256>>>) -> Self
     where
-        R: AsyncRead + Unpin + Send + 'static,
+        R: AsyncBufRead + Unpin + Send + 'static,
     {
         // compute NAR hash and size
         let (stream, nar_compute) = HashReader::new(stream, Sha256::new());