Merge branch 'main' into bt-redis-rate-limiter-to-async

Port limiter to async from old futures
This has been a long time coming...
2026-11-13 10:23:12 -05:00 · 2026-01-18 10:59:38 -05:00 · 2026-01-18 10:57:49 -05:00 · 2026-01-18 00:56:02 -05:00 · 2026-01-17 23:23:49 -05:00 · 2026-01-17 23:18:38 -05:00
46 changed files with 295 additions and 593 deletions
@@ -250,7 +250,7 @@ log = "0.4.27"
 once_cell = "1.21.3"
 r2d2 = "0.8.10"
 rand = "0.9.1"
-redis = { version = "1.0.2", features = ["r2d2", "tokio-rustls-comp"] }
+redis = { version = "1.0.2", features = ["r2d2", "tokio-rustls-comp", "tokio-comp", "cluster-async", "aio"] }
 regex = "1.12.2"
 reqwest = { version = "0.12.23", default-features = false, features = [ "charset", "http2", "system-proxy", "json", "cookies", "rustls-tls" ] }
 reqwest-websocket = {  version = "0.5.0" }
@@ -216,7 +216,7 @@ pub async fn enqueue_face_animation_handler(
    Some(ref _user) => &server_state.redis_rate_limiters.logged_in
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(EnqueueFaceAnimationError::RateLimited);
  }

@@ -158,7 +158,7 @@ pub async fn enqueue_rerender_animation_handler(
        },
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueRerenderAnimationError::RateLimited);
    }

@@ -175,7 +175,7 @@ pub async fn enqueue_fbx_to_gltf_handler(
        Some(ref _user) => &server_state.redis_rate_limiters.logged_in,
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueFbxToGltfRequestError::RateLimited);
    }

@@ -208,7 +208,7 @@ pub async fn enqueue_render_engine_scene_to_video_handler(
        Some(ref _user) => &server_state.redis_rate_limiters.logged_in,
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueBvhToWorkflowRequestError::RateLimited);
    }

@@ -116,7 +116,7 @@ pub async fn create_scene_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(CreateSceneError::RateLimited);
  }

@@ -88,7 +88,7 @@ pub async fn process_upload_media_file(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -268,7 +268,7 @@ pub async fn enqueue_image_generation_request(
        Some(ref _user) => &server_state.redis_rate_limiters.logged_in,
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueImageGenRequestError::RateLimited);
    }

@@ -81,7 +81,7 @@ pub async fn handle_upload(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(UploadError::RateLimited);
  }

@@ -153,7 +153,7 @@ pub async fn enqueue_mocapnet_handler(
        },
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueMocapnetError::RateLimited);
    }

@@ -116,7 +116,7 @@ pub async fn enqueue_infer_w2l_with_uploads(
    Some(ref _user) => &server_state.redis_rate_limiters.logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(InferW2lWithUploadError::RateLimited);
  }

@@ -89,7 +89,7 @@ pub async fn upload_w2l_template_handler(
  request: web::Json<UploadW2lTemplateRequest>,
  server_state: web::Data<Arc<ServerState>>) -> Result<HttpResponse, UploadW2lTemplateError>
 {
-  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request) {
+  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request).await {
    return Err(UploadW2lTemplateError::RateLimited);
  }

@@ -178,7 +178,7 @@ pub async fn enqueue_face_fusion_workflow_handler(

  // ==================== RATE LIMIT ==================== //

-  if let Err(_err) = server_state.redis_rate_limiters.logged_in.rate_limit_request(&http_request) {
+  if let Err(_err) = server_state.redis_rate_limiters.logged_in.rate_limit_request(&http_request).await {
    return Err(EnqueueFaceFusionWorkflowError::RateLimited);
  }

@@ -186,7 +186,7 @@ pub async fn enqueue_live_portrait_workflow_handler(

  // ==================== RATE LIMIT ==================== //

-  if let Err(_err) = server_state.redis_rate_limiters.logged_in.rate_limit_request(&http_request) {
+  if let Err(_err) = server_state.redis_rate_limiters.logged_in.rate_limit_request(&http_request).await {
    return Err(EnqueueLivePortraitWorkflowError::RateLimited);
  }

@@ -154,7 +154,7 @@ pub async fn enqueue_studio_workflow_handler(
    },
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(VstError::RateLimited);
  }

@@ -155,7 +155,7 @@ pub async fn enqueue_video_style_transfer_workflow_handler(
    },
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(VstError::RateLimited);
  }

@@ -174,7 +174,7 @@ pub async fn enqueue_comfy_ui_handler(
        },
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueComfyError::RateLimited);
    }

@@ -301,7 +301,7 @@ pub async fn enqueue_video_style_transfer_handler(
        },
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueVideoStyleTransferError::RateLimited);
    }

@@ -185,7 +185,7 @@ pub async fn enqueue_workflow_upload_request(
        Some(ref _user) => &server_state.redis_rate_limiters.logged_in,
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueWorkFlowRequestError::RateLimited);
    }

@@ -104,7 +104,7 @@ pub async fn redeem_beta_key_handler(

  let rate_limiter = &server_state.redis_rate_limiters.logged_out;

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(RedeemBetaKeyError::RateLimited);
  }

@@ -72,7 +72,7 @@ pub async fn enqueue_generic_download_handler(
  request: web::Json<EnqueueGenericDownloadRequest>,
  server_state: web::Data<Arc<ServerState>>) -> Result<HttpResponse, EnqueueGenericDownloadError>
 {
-  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request) {
+  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request).await {
    return Err(EnqueueGenericDownloadError::RateLimited);
  }

@@ -149,7 +149,7 @@ pub async fn upload_audio_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -88,7 +88,7 @@ pub async fn upload_engine_asset_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -93,7 +93,7 @@ pub async fn process_upload_media_file(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -147,7 +147,7 @@ pub async fn upload_new_engine_asset_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -133,7 +133,7 @@ pub async fn upload_new_scene_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -130,7 +130,7 @@ pub async fn upload_saved_scene_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -132,7 +132,7 @@ pub async fn upload_scene_snapshot_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -183,7 +183,7 @@ pub async fn upload_new_video_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -100,7 +100,7 @@ pub async fn upload_video_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileUploadError::RateLimited);
  }

@@ -94,7 +94,7 @@ pub async fn write_engine_asset_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileWriteError::RateLimited);
  }

@@ -96,7 +96,7 @@ pub async fn write_scene_file_media_file_handler(
    Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(MediaFileWriteError::RateLimited);
  }

@@ -132,7 +132,7 @@ pub async fn enqueue_gptsovits_model_download_handler(

  let maybe_routing_tag= get_routing_tag_header(&http_request);

-  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request) {
+  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request).await {
    return Err(EnqueueGptSovitsModelDownloadError::RateLimited);
  }

@@ -211,7 +211,7 @@ pub async fn enqueue_studio_gen2_handler(
    },
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(EnqueueStudioGen2Error::RateLimited);
  }

@@ -261,7 +261,7 @@ pub async fn enqueue_infer_f5_tts_handler(
      },
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
      return Err(InferTtsError::RateLimited);
    }
  }
@@ -287,7 +287,7 @@ pub async fn enqueue_infer_tts_handler(
      },
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
      return Err(InferTtsError::RateLimited);
    }
  }
@@ -113,7 +113,7 @@ pub async fn upload_tts_model_handler(
  request: web::Json<UploadTtsModelRequest>,
  server_state: web::Data<Arc<ServerState>>) -> Result<HttpResponse, UploadTtsModelError>
 {
-  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request) {
+  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request).await {
    return Err(UploadTtsModelError::RateLimited);
  }

@@ -243,7 +243,7 @@ pub async fn enqueue_infer_seed_vc_handler(
      },
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
      return Err(InferSeedVcError::RateLimited);
    }
  }
@@ -251,7 +251,7 @@ pub async fn enqueue_voice_conversion_inference_handler(
      rate_limiter = &server_state.redis_rate_limiters.logged_in;
    }

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
      return Err(EnqueueVoiceConversionInferenceError::RateLimited);
    }
  }
@@ -172,7 +172,7 @@ pub async fn enqueue_tts_request(
        Some(ref _user) => &server_state.redis_rate_limiters.logged_in,
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueTTSRequestError::RateLimited);
    }

@@ -123,11 +123,11 @@ pub async fn upload_zs_sample_handler(
    Some(ref _user) => &server_state.redis_rate_limiters.file_upload_logged_in,
  };

-  if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+  if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
    return Err(UploadSampleError::RateLimited);
  }

-  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request) {
+  if let Err(_err) = server_state.redis_rate_limiters.model_upload.rate_limit_request(&http_request).await {
    return Err(UploadSampleError::RateLimited);
  }

@@ -155,7 +155,7 @@ pub async fn create_voice_handler(
        Some(ref _user) => &server_state.redis_rate_limiters.logged_in,
    };

-    if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
+    if let Err(_err) = rate_limiter.rate_limit_request(&http_request).await {
        return Err(EnqueueCreateVoiceRequestError::RateLimited);
    }

@@ -36,7 +36,7 @@ impl RedisRateLimiter {
    }
  }

-  pub fn rate_limit_request(&self, request: &HttpRequest) -> Result<(), RateLimiterError> {
+  pub async fn rate_limit_request(&self, request: &HttpRequest) -> Result<(), RateLimiterError> {
    if !self.limiter_enabled {
      return Ok(())
    }
@@ -50,19 +50,19 @@ impl RedisRateLimiter {
    let ip_address = get_request_ip(&request);

    let rate_limit_key = format!("rate_limit:{}:{}", &self.limit_key_prefix, ip_address);
-    self.rate_limit_key(&rate_limit_key)
+    self.rate_limit_key(&rate_limit_key).await
  }

-  pub fn rate_limit_key(&self, rate_limit_key: &str) -> Result<(), RateLimiterError> {
+  pub async fn rate_limit_key(&self, rate_limit_key: &str) -> Result<(), RateLimiterError> {
    if !self.limiter_enabled {
      return Ok(())
    }

    // NB: This library uses old-school futures (pre-async/await)
-    let permit = self.limiter.count(rate_limit_key);
+    let permit = self.limiter.count(rate_limit_key).await;

    // TODO/FIXME: This could block.
-    match permit.wait() {
+    match permit {
      Ok(_) => Ok(()),
      Err(err) => match err {
        Error::Client(_) => Ok(()), // NB: Fail open for failure to connect to Redis
@@ -23,7 +23,7 @@ normal = ["workspace-build-acceleration"]
 [dependencies]
 chrono = "0.4.9"
 futures = "0.1.29"
-redis = "0.13.0"
+redis.workspace = true
 time = "0.3.30"

 # Workspace hack (faster compile times)
@@ -194,10 +194,10 @@ impl Limiter {
    /// - A time computation failed
    ///
    /// [`Status`]: struct.Status.html
-    pub fn count<K: Into<String>>(&self, key: K) -> impl Future<Item = Status, Error = Error> {
+    pub async fn count<K: Into<String>>(&self, key: K) -> Result<Status, Error> {
        let limit = self.limit;

-        self.track(key).and_then(move |(count, reset_epoch_utc)| {
+        self.track(key).await.and_then(move |(count, reset_epoch_utc)| {
            let status = build_status(count, limit, reset_epoch_utc);

            if count > limit {
@@ -209,37 +209,47 @@ impl Limiter {
    }

    /// Tracks the given key in a period and returns the count and TTL for the key in seconds.
-    fn track<K: Into<String>>(&self, key: K) -> impl Future<Item = (usize, usize), Error = Error> {
+    async fn track<K: Into<String>>(&self, key: K) -> Result<(usize, usize), Error> {
        let key = key.into();
        let exipres = self.period.as_secs();

-        self.client
-            .get_async_connection()
-            .from_err()
-            .and_then(move |con| {
-                // The seed of this approach is outlined Atul R in a blog post about rate limiting
-                // using NodeJS and Redis. For more details, see
-                // https://blog.atulr.com/rate-limiter/
-                let mut pipe = redis::pipe();
-                pipe.atomic()
-                    .cmd("SET")
-                    .arg(&key)
-                    .arg(0)
-                    .arg("EX")
-                    .arg(exipres)
-                    .arg("NX")
-                    .ignore()
-                    .cmd("INCR")
-                    .arg(&key)
-                    .cmd("TTL")
-                    .arg(&key);
+        let mut conn = self.client
+            .get_multiplexed_async_connection()
+            .await
+            .map_err(|err| Error::from(err))?;
+        
+        // The seed of this approach is outlined Atul R in a blog post about rate limiting
+        // using NodeJS and Redis. For more details, see
+        // https://blog.atulr.com/rate-limiter/
+        let mut pipe = redis::pipe();
+        pipe.atomic()
+            .cmd("SET")
+            .arg(&key)
+            .arg(0)
+            .arg("EX")
+            .arg(exipres)
+            .arg("NX")
+            .ignore()
+            .cmd("INCR")
+            .arg(&key)
+            .cmd("TTL")
+            .arg(&key);

-                pipe.query_async(con)
-                    .from_err()
-                    .and_then(|(_, (count, ttl)): (_, (usize, u64))| {
-                        Ok((count, epoch_utc_plus(Duration::from_secs(ttl))?))
-                    })
-            })
+        //let result = pipe.query_async(&mut conn)
+        //    .await
+        //    .map_err(|err| Error::from(err))
+        //    .and_then(|(_, (count, ttl)): (_, (usize, u64))| {
+        //        Ok((count, epoch_utc_plus(Duration::from_secs(ttl))?))
+        //    })?;
+
+        let result = pipe.query_async(&mut conn)
+            .await
+            .map_err(|err| Error::from(err))
+            .and_then(|(ignore, count, ttl): ((), usize, u64)| {
+                Ok((count, epoch_utc_plus(Duration::from_secs(ttl))?))
+            })?;
+        
+        Ok(result)
    }
 }
Author	SHA1	Message	Date
Brandon Thomas	46d19b3b47	Merge branch 'main' into bt-redis-rate-limiter-to-async	2026-01-18 10:59:38 -05:00
Brandon Thomas	e2e81e4489	Port `limiter` to async from old futures This has been a long time coming...	2026-01-18 10:57:49 -05:00
Brandon Thomas	e75a135633	upgrade redis limitation crate	2026-01-18 00:56:02 -05:00
Brandon Thomas	336b887d63	workspace config	2026-01-17 23:23:49 -05:00
Brandon Thomas	991b573b49	finish upgrade	2026-01-17 23:18:38 -05:00
Brandon Thomas	9f3a6e4c61	r2d2_redis is long deprecated, start replacement	2026-01-17 23:14:00 -05:00