runtime: add execution concurrency limits

crates/crank-runtime/Cargo.toml

@@ -21,6 +21,7 @@ serde.workspace = true
 serde_json.workspace = true
 sha2.workspace = true
 thiserror.workspace = true
+tokio = { workspace = true, features = ["sync"] }
 tracing.workspace = true
 
 [dev-dependencies]
@@ -28,6 +29,5 @@ axum.workspace = true
 crank-adapter-grpc = { path = "../crank-adapter-grpc", features = ["test-support"] }
 futures-util = "0.3"
 time.workspace = true
-tokio.workspace = true
 tokio-tungstenite.workspace = true
 tracing-subscriber.workspace = true

crates/crank-runtime/src/error.rs

@@ -33,6 +33,8 @@ pub enum RuntimeError {
         operation_id: String,
         mode: ExecutionMode,
     },
+    #[error("runtime concurrency limit exceeded for {kind} executions (limit {limit})")]
+    ConcurrencyLimitExceeded { kind: &'static str, limit: usize },
     #[error("invalid prepared request at {field}: {reason}")]
     InvalidPreparedRequest { field: String, reason: String },
     #[error("invalid streaming payload at {field}: {reason}")]

crates/crank-runtime/src/executor.rs

@@ -1,4 +1,5 @@
 use std::collections::BTreeMap;
+use std::sync::Arc;
 
 use crank_adapter_graphql::{GraphqlAdapter, GraphqlRequest};
 use crank_adapter_grpc::{GrpcAdapter, GrpcRequest, GrpcWindowRequest};
@@ -7,10 +8,11 @@ use crank_adapter_soap::{SoapAdapter, SoapRequest};
 use crank_adapter_websocket::{WebsocketAdapter, WebsocketWindowRequest};
 use crank_core::{ExecutionMode, Target, TransportBehavior};
 use serde_json::{Map, Value, json};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tracing::debug;
 
 use crate::{
-    AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeOperation,
+    AdapterResponse, PreparedRequest, ResolvedAuth, RuntimeError, RuntimeLimits, RuntimeOperation,
     RuntimeRequestContext, WindowExecutionResult,
 };
 
@@ -21,6 +23,11 @@ pub struct RuntimeExecutor {
     rest_adapter: RestAdapter,
     soap_adapter: SoapAdapter,
     websocket_adapter: WebsocketAdapter,
+    limits: RuntimeLimits,
+    unary_limiter: Arc<Semaphore>,
+    window_limiter: Arc<Semaphore>,
+    session_limiter: Arc<Semaphore>,
+    job_limiter: Arc<Semaphore>,
 }
 
 impl Default for RuntimeExecutor {
@@ -31,12 +38,21 @@ impl Default for RuntimeExecutor {
 
 impl RuntimeExecutor {
     pub fn new() -> Self {
+        Self::with_limits(RuntimeLimits::default())
+    }
+
+    pub fn with_limits(limits: RuntimeLimits) -> Self {
         Self {
             graphql_adapter: GraphqlAdapter::new(),
             grpc_adapter: GrpcAdapter::new(),
             rest_adapter: RestAdapter::new(),
             soap_adapter: SoapAdapter::new(),
             websocket_adapter: WebsocketAdapter::new(),
+            unary_limiter: Arc::new(Semaphore::new(limits.max_concurrent_unary)),
+            window_limiter: Arc::new(Semaphore::new(limits.max_concurrent_window)),
+            session_limiter: Arc::new(Semaphore::new(limits.max_concurrent_sessions)),
+            job_limiter: Arc::new(Semaphore::new(limits.max_concurrent_jobs)),
+            limits,
         }
     }
 
@@ -77,6 +93,7 @@ impl RuntimeExecutor {
         request_context: Option<&RuntimeRequestContext>,
     ) -> Result<Value, RuntimeError> {
         log_runtime_event("unary.execute", operation, request_context);
+        let _permit = self.acquire_unary_permit(operation)?;
         let prepared_request = self.prepare_request(operation, input)?;
         let prepared_request = apply_resolved_auth(prepared_request, resolved_auth);
         self.execute_prepared(operation, prepared_request, request_context)
@@ -120,6 +137,7 @@ impl RuntimeExecutor {
         request_context: Option<&RuntimeRequestContext>,
     ) -> Result<WindowExecutionResult, RuntimeError> {
         log_runtime_event("window.execute", operation, request_context);
+        let _permit = self.acquire_window_permit()?;
         let Some(streaming) = operation.execution_config.streaming.as_ref() else {
             return Err(RuntimeError::MissingStreamingConfig {
                 operation_id: operation.operation_id.as_str().to_owned(),
@@ -189,6 +207,7 @@ impl RuntimeExecutor {
         request_context: Option<&RuntimeRequestContext>,
     ) -> Result<WindowExecutionResult, RuntimeError> {
         log_runtime_event("session.seed", operation, request_context);
+        let _permit = self.acquire_session_permit()?;
         let Some(streaming) = operation.execution_config.streaming.as_ref() else {
             return Err(RuntimeError::MissingStreamingConfig {
                 operation_id: operation.operation_id.as_str().to_owned(),
@@ -482,6 +501,48 @@ impl RuntimeExecutor {
             }
         }
     }
+
+    fn acquire_unary_permit(
+        &self,
+        operation: &RuntimeOperation,
+    ) -> Result<OwnedSemaphorePermit, RuntimeError> {
+        let (kind, limit, limiter) = if operation
+            .execution_config
+            .streaming
+            .as_ref()
+            .is_some_and(|streaming| streaming.mode == ExecutionMode::AsyncJob)
+        {
+            (
+                "async_job",
+                self.limits.max_concurrent_jobs,
+                Arc::clone(&self.job_limiter),
+            )
+        } else {
+            (
+                "unary",
+                self.limits.max_concurrent_unary,
+                Arc::clone(&self.unary_limiter),
+            )
+        };
+
+        try_acquire_limit(limiter, kind, limit)
+    }
+
+    fn acquire_window_permit(&self) -> Result<OwnedSemaphorePermit, RuntimeError> {
+        try_acquire_limit(
+            Arc::clone(&self.window_limiter),
+            "window",
+            self.limits.max_concurrent_window,
+        )
+    }
+
+    fn acquire_session_permit(&self) -> Result<OwnedSemaphorePermit, RuntimeError> {
+        try_acquire_limit(
+            Arc::clone(&self.session_limiter),
+            "session",
+            self.limits.max_concurrent_sessions,
+        )
+    }
 }
 
 impl PreparedRequest {
@@ -556,6 +617,16 @@ fn runtime_context_headers(
         .unwrap_or_default()
 }
 
+fn try_acquire_limit(
+    limiter: Arc<Semaphore>,
+    kind: &'static str,
+    limit: usize,
+) -> Result<OwnedSemaphorePermit, RuntimeError> {
+    limiter
+        .try_acquire_owned()
+        .map_err(|_| RuntimeError::ConcurrencyLimitExceeded { kind, limit })
+}
+
 fn log_runtime_event(
     stage: &'static str,
     operation: &RuntimeOperation,
@@ -665,7 +736,8 @@ mod tests {
     use tracing_subscriber::{filter::LevelFilter, fmt::MakeWriter, prelude::*};
 
     use crate::{
-        PreparedRequest, RuntimeError, RuntimeExecutor, RuntimeOperation, RuntimeRequestContext,
+        PreparedRequest, RuntimeError, RuntimeExecutor, RuntimeLimits, RuntimeOperation,
+        RuntimeRequestContext,
     };
 
     fn timestamp(value: &str) -> OffsetDateTime {
@@ -788,6 +860,96 @@ mod tests {
         assert!(logs.contains("op_rest_runtime"));
     }
 
+    #[tokio::test]
+    async fn rejects_unary_execution_when_unary_limit_is_exhausted() {
+        let executor = RuntimeExecutor::with_limits(RuntimeLimits {
+            max_concurrent_unary: 0,
+            ..RuntimeLimits::default()
+        });
+        let operation = test_rest_operation("http://example.invalid", false, false);
+
+        let error = executor
+            .execute(&operation, &json!({ "email": "user@example.com" }))
+            .await
+            .unwrap_err();
+
+        assert!(matches!(
+            error,
+            RuntimeError::ConcurrencyLimitExceeded {
+                kind: "unary",
+                limit: 0
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn rejects_window_execution_when_window_limit_is_exhausted() {
+        let executor = RuntimeExecutor::with_limits(RuntimeLimits {
+            max_concurrent_window: 0,
+            ..RuntimeLimits::default()
+        });
+        let operation = test_window_snapshot_operation(
+            "http://example.invalid",
+            AggregationMode::RawItems,
+            None,
+            None,
+        );
+
+        let error = executor
+            .execute_window(&operation, &json!({}))
+            .await
+            .unwrap_err();
+
+        assert!(matches!(
+            error,
+            RuntimeError::ConcurrencyLimitExceeded {
+                kind: "window",
+                limit: 0
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn rejects_session_seed_when_session_limit_is_exhausted() {
+        let executor = RuntimeExecutor::with_limits(RuntimeLimits {
+            max_concurrent_sessions: 0,
+            ..RuntimeLimits::default()
+        });
+        let operation = test_session_seed_operation("http://example.invalid");
+
+        let error = executor
+            .execute_session_seed(&operation, &json!({}))
+            .await
+            .unwrap_err();
+
+        assert!(matches!(
+            error,
+            RuntimeError::ConcurrencyLimitExceeded {
+                kind: "session",
+                limit: 0
+            }
+        ));
+    }
+
+    #[tokio::test]
+    async fn rejects_async_job_execution_when_job_limit_is_exhausted() {
+        let executor = RuntimeExecutor::with_limits(RuntimeLimits {
+            max_concurrent_jobs: 0,
+            ..RuntimeLimits::default()
+        });
+        let operation = test_async_job_operation("http://example.invalid");
+
+        let error = executor.execute(&operation, &json!({})).await.unwrap_err();
+
+        assert!(matches!(
+            error,
+            RuntimeError::ConcurrencyLimitExceeded {
+                kind: "async_job",
+                limit: 0
+            }
+        ));
+    }
+
     #[tokio::test]
     async fn executes_graphql_operation_end_to_end() {
         let endpoint = spawn_graphql_server().await;
@@ -2106,6 +2268,49 @@ mod tests {
         operation
     }
 
+    fn test_session_seed_operation(base_url: &str) -> RuntimeOperation {
+        let mut operation =
+            test_window_snapshot_operation(base_url, AggregationMode::RawItems, Some(10), None);
+
+        operation
+            .execution_config
+            .streaming
+            .as_mut()
+            .expect("streaming config")
+            .mode = ExecutionMode::Session;
+
+        operation
+    }
+
+    fn test_async_job_operation(base_url: &str) -> RuntimeOperation {
+        let mut operation = test_rest_operation(base_url, false, false);
+        operation.execution_config.streaming = Some(StreamingConfig {
+            mode: ExecutionMode::AsyncJob,
+            transport_behavior: TransportBehavior::RequestResponse,
+            window_duration_ms: None,
+            poll_interval_ms: Some(5_000),
+            upstream_timeout_ms: Some(operation.execution_config.timeout_ms),
+            idle_timeout_ms: None,
+            max_session_lifetime_ms: Some(300_000),
+            max_items: None,
+            max_bytes: None,
+            aggregation_mode: AggregationMode::RawItems,
+            summary_path: None,
+            items_path: None,
+            cursor_path: None,
+            status_path: None,
+            done_path: None,
+            redacted_paths: Vec::new(),
+            truncate_item_fields: false,
+            max_field_length: None,
+            drop_duplicates: false,
+            sampling_rate: None,
+            tool_family: ToolFamilyConfig::default(),
+        });
+
+        operation
+    }
+
     fn object_schema(field_name: &str, kind: SchemaKind) -> Schema {
         Schema {
             kind: SchemaKind::Object,

crates/crank-runtime/src/lib.rs

@@ -2,6 +2,7 @@ mod aggregation;
 mod auth;
 mod error;
 mod executor;
+mod limits;
 mod model;
 mod redaction;
 mod request_context;
@@ -11,6 +12,7 @@ mod streaming;
 pub use auth::ResolvedAuth;
 pub use error::RuntimeError;
 pub use executor::RuntimeExecutor;
+pub use limits::{RuntimeLimits, RuntimeLimitsConfigError};
 pub use model::{AdapterResponse, PreparedRequest, RuntimeOperation};
 pub use request_context::RuntimeRequestContext;
 pub use secret_crypto::SecretCrypto;

crates/crank-runtime/src/limits.rs

@@ -0,0 +1,119 @@
+use std::{env, num::ParseIntError};
+
+use thiserror::Error;
+
+const DEFAULT_MAX_CONCURRENT_UNARY: usize = 64;
+const DEFAULT_MAX_CONCURRENT_WINDOW: usize = 16;
+const DEFAULT_MAX_CONCURRENT_SESSIONS: usize = 16;
+const DEFAULT_MAX_CONCURRENT_JOBS: usize = 16;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct RuntimeLimits {
+    pub max_concurrent_unary: usize,
+    pub max_concurrent_window: usize,
+    pub max_concurrent_sessions: usize,
+    pub max_concurrent_jobs: usize,
+}
+
+impl Default for RuntimeLimits {
+    fn default() -> Self {
+        Self {
+            max_concurrent_unary: DEFAULT_MAX_CONCURRENT_UNARY,
+            max_concurrent_window: DEFAULT_MAX_CONCURRENT_WINDOW,
+            max_concurrent_sessions: DEFAULT_MAX_CONCURRENT_SESSIONS,
+            max_concurrent_jobs: DEFAULT_MAX_CONCURRENT_JOBS,
+        }
+    }
+}
+
+impl RuntimeLimits {
+    pub fn from_env() -> Result<Self, RuntimeLimitsConfigError> {
+        Ok(Self {
+            max_concurrent_unary: parse_limit(
+                "CRANK_RUNTIME_MAX_CONCURRENT_UNARY",
+                DEFAULT_MAX_CONCURRENT_UNARY,
+            )?,
+            max_concurrent_window: parse_limit(
+                "CRANK_RUNTIME_MAX_CONCURRENT_WINDOW",
+                DEFAULT_MAX_CONCURRENT_WINDOW,
+            )?,
+            max_concurrent_sessions: parse_limit(
+                "CRANK_RUNTIME_MAX_CONCURRENT_SESSIONS",
+                DEFAULT_MAX_CONCURRENT_SESSIONS,
+            )?,
+            max_concurrent_jobs: parse_limit(
+                "CRANK_RUNTIME_MAX_CONCURRENT_JOBS",
+                DEFAULT_MAX_CONCURRENT_JOBS,
+            )?,
+        })
+    }
+}
+
+fn parse_limit(name: &'static str, default: usize) -> Result<usize, RuntimeLimitsConfigError> {
+    match env::var(name) {
+        Ok(raw) => {
+            let value =
+                raw.parse::<usize>()
+                    .map_err(|source| RuntimeLimitsConfigError::InvalidValue {
+                        name,
+                        value: raw,
+                        source,
+                    })?;
+            if value == 0 {
+                return Err(RuntimeLimitsConfigError::ZeroValue { name });
+            }
+            Ok(value)
+        }
+        Err(env::VarError::NotPresent) => Ok(default),
+        Err(env::VarError::NotUnicode(_)) => Err(RuntimeLimitsConfigError::InvalidUnicode { name }),
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum RuntimeLimitsConfigError {
+    #[error("{name} must contain valid UTF-8")]
+    InvalidUnicode { name: &'static str },
+    #[error("{name} must be a positive integer, got {value}")]
+    InvalidValue {
+        name: &'static str,
+        value: String,
+        source: ParseIntError,
+    },
+    #[error("{name} must be greater than zero")]
+    ZeroValue { name: &'static str },
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{RuntimeLimits, RuntimeLimitsConfigError};
+
+    #[test]
+    fn defaults_are_positive() {
+        let limits = RuntimeLimits::default();
+
+        assert!(limits.max_concurrent_unary > 0);
+        assert!(limits.max_concurrent_window > 0);
+        assert!(limits.max_concurrent_sessions > 0);
+        assert!(limits.max_concurrent_jobs > 0);
+    }
+
+    #[test]
+    fn rejects_zero_limit_values() {
+        unsafe {
+            std::env::set_var("CRANK_RUNTIME_MAX_CONCURRENT_UNARY", "0");
+        }
+
+        let error = RuntimeLimits::from_env().unwrap_err();
+
+        assert!(matches!(
+            error,
+            RuntimeLimitsConfigError::ZeroValue {
+                name: "CRANK_RUNTIME_MAX_CONCURRENT_UNARY"
+            }
+        ));
+
+        unsafe {
+            std::env::remove_var("CRANK_RUNTIME_MAX_CONCURRENT_UNARY");
+        }
+    }
+}