From 2523b6644b12f79eaca69e1b4ed15e6e6e4ff762 Mon Sep 17 00:00:00 2001
From: asonix <asonix@asonix.dog>
Date: Mon, 4 Dec 2023 15:36:23 -0600
Subject: [PATCH] Add AggregateTimeout, don't aggregate with permit

---
 src/error.rs  | 8 ++++++--
 src/ingest.rs | 6 ++++--
 2 files changed, 10 insertions(+), 4 deletions(-)

diff --git a/src/error.rs b/src/error.rs
index e16a4a2..649a6d6 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -150,6 +150,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")]
     FailedExternalValidation,
 }
@@ -183,7 +186,7 @@ impl UploadError {
             Self::PushJob(_) => ErrorCode::PUSH_JOB,
             Self::Range => ErrorCode::RANGE_NOT_SATISFIABLE,
             Self::Limit(_) => ErrorCode::VALIDATE_FILE_SIZE,
-            Self::Timeout(_) => ErrorCode::STREAM_TOO_SLOW,
+            Self::Timeout(_) | Self::AggregateTimeout => ErrorCode::STREAM_TOO_SLOW,
             Self::FailedExternalValidation => ErrorCode::FAILED_EXTERNAL_VALIDATION,
         }
     }
@@ -233,7 +236,8 @@ impl ResponseError for Error {
                 | UploadError::Validation(_)
                 | UploadError::UnsupportedProcessExtension
                 | UploadError::ReadOnly
-                | UploadError::FailedExternalValidation,
+                | UploadError::FailedExternalValidation
+                | 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,
diff --git a/src/ingest.rs b/src/ingest.rs
index 3b85952..ffdad03 100644
--- a/src/ingest.rs
+++ b/src/ingest.rs
@@ -55,9 +55,11 @@ async fn process_ingest<S>(
 where
     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?;
 
     let prescribed = Validations {
         image: &media.image,