Add AggregateTimeout, don't aggregate with permit

src/error.rs

@@ -133,6 +133,9 @@ pub(crate) enum UploadError {
     #[error("Response timeout")]
     Timeout(#[from] crate::stream::TimeoutError),
 
+    #[error("Client took too long to send request")]
+    AggregateTimeout,
+
     #[error("Failed external validation")]
     ExternalValidation,
 }
@@ -171,7 +174,8 @@ impl ResponseError for Error {
                 ))
                 | UploadError::Repo(crate::repo::RepoError::AlreadyClaimed)
                 | UploadError::UnsupportedProcessExtension
-                | UploadError::SilentVideoDisabled,
+                | UploadError::SilentVideoDisabled
+                | UploadError::AggregateTimeout,
             ) => StatusCode::BAD_REQUEST,
             Some(UploadError::Magick(e)) if e.is_client_error() => StatusCode::BAD_REQUEST,
             Some(UploadError::Ffmpeg(e)) if e.is_client_error() => StatusCode::BAD_REQUEST,

src/ingest.rs

@@ -63,9 +63,11 @@ where
     R: FullRepo + 'static,
     S: Store,
 {
-    let permit = crate::PROCESS_SEMAPHORE.acquire().await;
+    let bytes = tokio::time::timeout(Duration::from_secs(60), aggregate(stream))
+        .await
+        .map_err(|_| UploadError::AggregateTimeout)??;
 
-    let bytes = aggregate(stream).await?;
+    let permit = crate::PROCESS_SEMAPHORE.acquire().await;
 
     tracing::trace!("Validating bytes");
     let (input_type, validated_reader) =