From 09a825ce5b5942e34cbc9c6d5804c808587eb30d Mon Sep 17 00:00:00 2001
From: Hampus <hampus@fluxer.app>
Date: Fri, 24 Jul 2026 23:17:21 +0200
Subject: [PATCH] fix(media-proxy): bound queued transforms (#1458)

---
 fluxer_media_proxy/src/coalescer.rs       |  1 +
 fluxer_media_proxy/src/server.rs          |  9 +++++++++
 fluxer_media_proxy/src/timed_semaphore.rs | 12 +++++++++++-
 3 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/fluxer_media_proxy/src/coalescer.rs b/fluxer_media_proxy/src/coalescer.rs
index beb29594d..088bb0af5 100644
--- a/fluxer_media_proxy/src/coalescer.rs
+++ b/fluxer_media_proxy/src/coalescer.rs
@@ -77,6 +77,7 @@ impl ByteCoalescer {
             self.in_flight.lock().remove(&key);
             result
         } else {
+            drop(work);
             crate::metrics::GLOBAL
                 .coalescer_waiter
                 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
diff --git a/fluxer_media_proxy/src/server.rs b/fluxer_media_proxy/src/server.rs
index 77a179d79..75501c1dd 100644
--- a/fluxer_media_proxy/src/server.rs
+++ b/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,
         })
diff --git a/fluxer_media_proxy/src/timed_semaphore.rs b/fluxer_media_proxy/src/timed_semaphore.rs
index c0268fe84..69d6e2525 100644
--- a/fluxer_media_proxy/src/timed_semaphore.rs
+++ b/fluxer_media_proxy/src/timed_semaphore.rs
@@ -2,7 +2,7 @@
 
 use std::time::Instant;
 use thiserror::Error;
-use tokio::sync::{OwnedSemaphorePermit, Semaphore};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, TryAcquireError};
 
 #[derive(Debug, Error, Eq, PartialEq)]
 pub enum TimedSemaphoreError {
@@ -41,6 +41,16 @@ impl TimedSemaphore {
             acquire.await.map_err(|_| TimedSemaphoreError::Closed)
         }
     }
+
+    pub fn try_wait(&self) -> Result<OwnedSemaphorePermit, TimedSemaphoreError> {
+        self.inner
+            .clone()
+            .try_acquire_owned()
+            .map_err(|error| match error {
+                TryAcquireError::NoPermits => TimedSemaphoreError::RequestTimeout,
+                TryAcquireError::Closed => TimedSemaphoreError::Closed,
+            })
+    }
 }
 
 #[cfg(test)]