fix(media-proxy): bound queued transforms (#1458)

fluxer_media_proxy/src/server.rs

@@ -49,6 +49,7 @@ struct AppState {
     nsfw_client: reqwest::Client,
     transform_cache: Arc<Cache>,
     coalescer: Arc<ByteCoalescer>,
+    native_transform_admissions: TimedSemaphore,
     native_transforms: TimedSemaphore,
 }
 
@@ -75,6 +76,7 @@ pub async fn run(cfg: Config) -> anyhow::Result<()> {
             cfg.transform_cache_ttl_ms,
         )),
         coalescer: Arc::new(ByteCoalescer::new()),
+        native_transform_admissions: TimedSemaphore::new(transform_admission_capacity(&cfg)),
         native_transforms: TimedSemaphore::new(cfg.max_native_transforms),
         cfg,
     });
@@ -2396,6 +2398,7 @@ async fn run_transform(
     options: media_process::ImageOptions,
 ) -> anyhow::Result<media_process::ProcessedMedia> {
     let deadline = deadline_instant(options.deadline_ms);
+    let _admission = app.native_transform_admissions.try_wait()?;
     let wait_start = metrics::now_ms();
     let _permit = app.native_transforms.wait_until(deadline).await?;
     let waited = (metrics::now_ms() - wait_start).max(0) as u64;
@@ -2420,6 +2423,7 @@ async fn run_video_transform(
     deadline_ms: Option<i64>,
 ) -> anyhow::Result<media_process::ProcessedMedia> {
     let deadline = deadline_instant(deadline_ms);
+    let _admission = app.native_transform_admissions.try_wait()?;
     let wait_start = metrics::now_ms();
     let _permit = app.native_transforms.wait_until(deadline).await?;
     let waited = (metrics::now_ms() - wait_start).max(0) as u64;
@@ -2467,6 +2471,10 @@ fn transform_error_is_timeout(error: &anyhow::Error) -> bool {
             == Some(&crate::timed_semaphore::TimedSemaphoreError::RequestTimeout)
 }
 
+fn transform_admission_capacity(cfg: &Config) -> usize {
+    cfg.max_native_transforms + cfg.worker_queue_capacity
+}
+
 fn media_response(
     method: Method,
     data: Bytes,
@@ -3116,6 +3124,7 @@ mod tests {
                 cfg.transform_cache_ttl_ms,
             )),
             coalescer: Arc::new(ByteCoalescer::new()),
+            native_transform_admissions: TimedSemaphore::new(transform_admission_capacity(&cfg)),
             native_transforms: TimedSemaphore::new(cfg.max_native_transforms),
             cfg,
         })