fix(gateway): deprecate legacy batch policy (#515)

Author: alessandromazza98

services/gateway/src/batch_policy.rs

@@ -344,7 +344,6 @@ mod tests {
 
     fn cfg() -> BatchPolicyConfig {
         BatchPolicyConfig {
-            enabled: true,
             reeval_ms: 2_000,
             max_wait_secs: 30,
             cost_ema_alpha: 0.2,

services/gateway/src/config.rs

@@ -60,13 +60,9 @@ impl Default for OrphanSweeperConfig {
     }
 }
 
-/// Adaptive batching policy configuration.
+/// Policy-driven batching configuration.
 #[derive(Clone, Debug, clap::Args)]
 pub struct BatchPolicyConfig {
-    /// Enable adaptive policy-driven batch dispatch.
-    #[arg(long, env = "BATCH_POLICY_ENABLED", default_value = "false")]
-    pub enabled: bool,
-
     /// Re-evaluation cadence for policy decisions, in milliseconds.
     #[arg(long, env = "BATCH_REEVAL_MS", default_value = "1000")]
     pub reeval_ms: u64,
@@ -91,7 +87,6 @@ pub struct BatchPolicyConfig {
 impl Default for BatchPolicyConfig {
     fn default() -> Self {
         Self {
-            enabled: false,
             reeval_ms: 1_000,
             max_wait_secs: 30,
             cost_ema_alpha: 0.2,
@@ -229,12 +224,9 @@ impl GatewayConfig {
             ));
         }
 
-        if self.batch_policy.enabled
-            && self.sweeper().stale_queued_threshold_secs <= self.batch_policy.max_wait_secs
-        {
+        if self.sweeper().stale_queued_threshold_secs <= self.batch_policy.max_wait_secs {
             return Err(GatewayError::Config(
-                "STALE_QUEUED_THRESHOLD_SECS must be greater than BATCH_MAX_WAIT_SECS when adaptive batching is enabled"
-                    .to_string(),
+                "STALE_QUEUED_THRESHOLD_SECS must be greater than BATCH_MAX_WAIT_SECS".to_string(),
             ));
         }
 
@@ -376,4 +368,16 @@ mod tests {
         assert!(err.contains("BATCH_COST_HIGH_RATIO"));
         assert!(err.contains("finite"));
     }
+
+    #[test]
+    fn test_stale_queued_threshold_must_exceed_max_wait_secs() {
+        let mut config = parse_valid_config();
+        config.batch_policy.max_wait_secs = 30;
+        config.stale_queued_threshold_secs = 30;
+
+        let result = config.validate();
+        assert!(result.is_err());
+        let err = result.unwrap_err().to_string();
+        assert!(err.contains("STALE_QUEUED_THRESHOLD_SECS"));
+    }
 }

services/gateway/src/create_batcher.rs

@@ -65,11 +65,7 @@ impl CreateBatcherRunner {
     }
 
     pub async fn run(mut self) {
-        if self.batch_policy.enabled {
-            self.run_policy_loop().await;
-        } else {
-            self.run_legacy_loop().await;
-        }
+        self.run_policy_loop().await;
     }
 
     async fn drop_local_queue_and_inflight(

services/gateway/src/ops_batcher.rs

@@ -82,11 +82,7 @@ impl OpsBatcherRunner {
     }
 
     pub async fn run(mut self) {
-        if self.batch_policy.enabled {
-            self.run_policy_loop().await;
-        } else {
-            self.run_legacy_loop().await;
-        }
+        self.run_policy_loop().await;
     }
 
     async fn submit_ops_batch(&self, batch: Vec<OpsEnvelope>) {

services/gateway/src/policy_batcher.rs

@@ -25,7 +25,7 @@ enum PolicyLoopEvent<T> {
 ///
 /// Implementors provide batcher-specific primitives (queue source, submit path,
 /// backlog scope, and no-backlog reconciliation), while this trait provides the
-/// common legacy/policy event loops.
+/// common policy event loop.
 pub(crate) trait PolicyBatchLoopRunner {
     type Envelope: Send + 'static;
 
@@ -54,50 +54,6 @@ pub(crate) trait PolicyBatchLoopRunner {
     /// create clears queue and removes in-flight authenticators, while ops only clears queue.
     async fn handle_no_backlog(&self, queue: &mut VecDeque<TimedEnvelope<Self::Envelope>>);
 
-    /// Legacy batching loop: wait for first request, then fill until deadline or max batch size.
-    async fn run_legacy_loop(&mut self) {
-        let window = Duration::from_millis(self.batch_policy().reeval_ms);
-
-        loop {
-            let first = {
-                let rx = self.rx();
-                rx.recv().await
-            };
-            let Some(first) = first else {
-                tracing::info!("{} batcher channel closed", self.batch_type());
-                return;
-            };
-
-            let mut batch = vec![first];
-            let deadline = Instant::now() + window;
-
-            loop {
-                if batch.len() >= self.max_batch_size() {
-                    break;
-                }
-
-                let next_req = {
-                    let rx = self.rx();
-                    tokio::time::timeout_at(deadline, rx.recv()).await
-                };
-
-                match next_req {
-                    Ok(Some(req)) => batch.push(req),
-                    Ok(None) => {
-                        tracing::info!(
-                            "{} batcher channel closed while batching",
-                            self.batch_type()
-                        );
-                        break;
-                    }
-                    Err(_) => break, // Timeout expired
-                }
-            }
-
-            self.submit_batch(batch).await;
-        }
-    }
-
     /// Policy-driven batching loop with periodic re-evaluation and bounded local queueing.
     async fn run_policy_loop(&mut self) {
         let mut policy_engine = BatchPolicyEngine::new(self.batch_policy().clone());

services/gateway/src/routes.rs

@@ -71,13 +71,11 @@ pub(crate) async fn build_app(
     let tracker = RequestTracker::new(redis_url, rate_limit).await;
     let base_fee_cache = BaseFeeCache::default();
 
-    if batch_policy_config.enabled {
-        spawn_base_fee_sampler(
-            registry.provider().clone(),
-            Duration::from_millis(batch_policy_config.reeval_ms),
-            base_fee_cache.clone(),
-        );
-    }
+    spawn_base_fee_sampler(
+        registry.provider().clone(),
+        Duration::from_millis(batch_policy_config.reeval_ms),
+        base_fee_cache.clone(),
+    );
 
     let (tx, rx) = mpsc::channel(CREATE_BATCHER_CHANNEL_CAPACITY);
     let batcher = CreateBatcherHandle { tx };