From 8ac55ebcc27d9485a6b4da930ad5773133354e8f Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Fri, 1 May 2026 12:11:53 +0000 Subject: [PATCH] runtime: add execution concurrency limits --- .env.example | 4 + TASKS.md | 11 +- apps/admin-api/src/error.rs | 22 +++ apps/admin-api/src/main.rs | 16 +- apps/admin-api/src/service.rs | 20 ++- apps/mcp-server/src/app.rs | 25 +++- apps/mcp-server/src/main.rs | 13 +- crates/crank-runtime/Cargo.toml | 2 +- crates/crank-runtime/src/error.rs | 2 + crates/crank-runtime/src/executor.rs | 209 ++++++++++++++++++++++++++- crates/crank-runtime/src/lib.rs | 2 + crates/crank-runtime/src/limits.rs | 119 +++++++++++++++ docs/runtime-config.md | 18 +++ 13 files changed, 448 insertions(+), 15 deletions(-) create mode 100644 crates/crank-runtime/src/limits.rs diff --git a/.env.example b/.env.example index c498136..ded7fa7 100644 --- a/.env.example +++ b/.env.example @@ -16,6 +16,10 @@ CRANK_PUBLISH_BIND=127.0.0.1 CRANK_ADMIN_BIND=0.0.0.0:3001 CRANK_MCP_BIND=0.0.0.0:3002 CRANK_MCP_REFRESH_MS=5000 +CRANK_RUNTIME_MAX_CONCURRENT_UNARY=64 +CRANK_RUNTIME_MAX_CONCURRENT_WINDOW=16 +CRANK_RUNTIME_MAX_CONCURRENT_SESSIONS=16 +CRANK_RUNTIME_MAX_CONCURRENT_JOBS=16 CRANK_LOG_LEVEL=info CRANK_MASTER_KEY=change-me-master-key CRANK_SESSION_SECRET=change-me-session-secret diff --git a/TASKS.md b/TASKS.md index 00e7890..dbd22c5 100644 --- a/TASKS.md +++ b/TASKS.md @@ -2,18 +2,19 @@ ## Current -### `feat/correlation-id-propagation` +### `feat/runtime-rate-limiting-and-backpressure` Status: in_progress DoD: -- admin-api, mcp-server, runtime, and adapters share a propagated correlation id -- request-scoped logs can be joined across layers without guessing from timestamps -- outbound adapter calls include correlation metadata where transport allows it +- runtime rejects overload deterministically with stable error codes +- unary, window, session, and async-job execution paths are concurrency-bounded +- configuration is explicit and documented +- admin-api and mcp-server use the configured runtime limits ## Next -- `feat/runtime-rate-limiting-and-backpressure` +- `feat/distributed-mcp-session-store` ## Backlog diff --git a/apps/admin-api/src/error.rs b/apps/admin-api/src/error.rs index aa33ef9..3c19c14 100644 --- a/apps/admin-api/src/error.rs +++ b/apps/admin-api/src/error.rs @@ -403,6 +403,7 @@ fn runtime_test_failure_code(error: &RuntimeError) -> &'static str { RuntimeError::SoapAdapter(_) => "runtime_soap_error", RuntimeError::WebsocketAdapter(_) => "runtime_websocket_error", RuntimeError::UnsupportedProtocol { .. } => "runtime_protocol_error", + RuntimeError::ConcurrencyLimitExceeded { .. } => "runtime_overloaded", RuntimeError::InvalidPreparedRequest { .. } => "runtime_request_error", RuntimeError::MissingStreamingConfig { .. } => "runtime_streaming_config_error", RuntimeError::UnsupportedExecutionMode { .. } => "runtime_streaming_mode_error", @@ -454,6 +455,10 @@ pub fn runtime_error_context(error: &RuntimeError) -> Option { RuntimeError::UnsupportedProtocol { protocol } => Some(json!({ "protocol": protocol, })), + RuntimeError::ConcurrencyLimitExceeded { kind, limit } => Some(json!({ + "kind": kind, + "limit": limit, + })), _ => None, } } @@ -500,6 +505,23 @@ mod tests { ); } + #[test] + fn runtime_test_failure_includes_runtime_overload_context() { + let payload = runtime_test_failure(&RuntimeError::ConcurrencyLimitExceeded { + kind: "window", + limit: 16, + }); + + assert_eq!(payload["code"], "runtime_overloaded"); + assert_eq!( + payload["context"], + json!({ + "kind": "window", + "limit": 16 + }) + ); + } + #[test] fn registry_errors_preserve_structured_context_in_api_error() { let error = ApiError::from(RegistryError::InvalidAsyncJobTransition { diff --git a/apps/admin-api/src/main.rs b/apps/admin-api/src/main.rs index ce23057..ff1b17b 100644 --- a/apps/admin-api/src/main.rs +++ b/apps/admin-api/src/main.rs @@ -20,7 +20,7 @@ use crate::{ service::AdminService, state::AppState, }; -use crank_runtime::SecretCrypto; +use crank_runtime::{RuntimeExecutor, RuntimeLimits, SecretCrypto}; #[tokio::main] async fn main() -> Result<(), Box> { @@ -57,8 +57,16 @@ async fn main() -> Result<(), Box> { .unwrap_or_else(|_| "Crank Owner".into()), }, }; + let runtime_limits = RuntimeLimits::from_env()?; let secret_crypto = SecretCrypto::new(&env::var("CRANK_MASTER_KEY")?)?; - let service = AdminService::new(registry, storage_root, auth_settings, secret_crypto); + let runtime = RuntimeExecutor::with_limits(runtime_limits); + let service = AdminService::new_with_runtime( + registry, + storage_root, + auth_settings, + secret_crypto, + runtime, + ); service.bootstrap_admin_user().await?; if env_flag("CRANK_DEMO_SEED") { service.seed_demo_assets().await?; @@ -68,6 +76,10 @@ async fn main() -> Result<(), Box> { let listener = TcpListener::bind(socket_addr).await?; info!( + runtime_max_concurrent_unary = runtime_limits.max_concurrent_unary, + runtime_max_concurrent_window = runtime_limits.max_concurrent_window, + runtime_max_concurrent_sessions = runtime_limits.max_concurrent_sessions, + runtime_max_concurrent_jobs = runtime_limits.max_concurrent_jobs, max_connections = pool_config.max_connections, min_connections = pool_config.min_connections, acquire_timeout_ms = pool_config.acquire_timeout_ms, diff --git a/apps/admin-api/src/service.rs b/apps/admin-api/src/service.rs index 402e7d8..2534436 100644 --- a/apps/admin-api/src/service.rs +++ b/apps/admin-api/src/service.rs @@ -698,15 +698,32 @@ fn default_operation_category() -> String { } impl AdminService { + #[cfg(test)] pub fn new( registry: PostgresRegistry, storage_root: PathBuf, auth_settings: AuthSettings, secret_crypto: SecretCrypto, + ) -> Self { + Self::new_with_runtime( + registry, + storage_root, + auth_settings, + secret_crypto, + RuntimeExecutor::new(), + ) + } + + pub fn new_with_runtime( + registry: PostgresRegistry, + storage_root: PathBuf, + auth_settings: AuthSettings, + secret_crypto: SecretCrypto, + runtime: RuntimeExecutor, ) -> Self { Self { registry, - runtime: RuntimeExecutor::new(), + runtime, storage: LocalArtifactStorage::new(storage_root), auth_settings, secret_crypto, @@ -4881,6 +4898,7 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str { RuntimeError::SoapAdapter(_) => "soap_error", RuntimeError::WebsocketAdapter(_) => "websocket_error", RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", + RuntimeError::ConcurrencyLimitExceeded { .. } => "runtime_overloaded", RuntimeError::MissingStreamingConfig { .. } => "streaming_config_error", RuntimeError::UnsupportedExecutionMode { .. } => "streaming_mode_error", RuntimeError::InvalidStreamingPayload { .. } => "streaming_payload_error", diff --git a/apps/mcp-server/src/app.rs b/apps/mcp-server/src/app.rs index 0c53833..8380a15 100644 --- a/apps/mcp-server/src/app.rs +++ b/apps/mcp-server/src/app.rs @@ -128,11 +128,12 @@ pub fn build_app( refresh_interval: Duration, public_base_url: Option, secret_crypto: SecretCrypto, + runtime: RuntimeExecutor, ) -> Router { let state = Arc::new(AppState { registry: registry.clone(), catalog: PublishedToolCatalog::new(registry, refresh_interval), - runtime: RuntimeExecutor::new(), + runtime, secret_crypto, sessions: SessionStore::new(), allowed_origins: AllowedOrigins::new(public_base_url), @@ -2047,6 +2048,7 @@ fn runtime_error_code(error: &RuntimeError) -> &'static str { RuntimeError::SoapAdapter(_) => "adapter_execution_error", RuntimeError::WebsocketAdapter(_) => "adapter_execution_error", RuntimeError::UnsupportedProtocol { .. } => "unsupported_protocol", + RuntimeError::ConcurrencyLimitExceeded { .. } => "runtime_overloaded", RuntimeError::MissingStreamingConfig { .. } => "streaming_config_error", RuntimeError::UnsupportedExecutionMode { .. } => "streaming_mode_error", RuntimeError::InvalidPreparedRequest { .. } => "runtime_error", @@ -2098,6 +2100,10 @@ fn runtime_error_context(error: &RuntimeError) -> Option { RuntimeError::UnsupportedProtocol { protocol } => Some(json!({ "protocol": protocol, })), + RuntimeError::ConcurrencyLimitExceeded { kind, limit } => Some(json!({ + "kind": kind, + "limit": limit, + })), _ => None, } } @@ -2506,6 +2512,23 @@ mod tests { ); } + #[test] + fn runtime_error_context_includes_runtime_overload_details() { + let context = runtime_error_context(&RuntimeError::ConcurrencyLimitExceeded { + kind: "async_job", + limit: 8, + }) + .unwrap(); + + assert_eq!( + context, + json!({ + "kind": "async_job", + "limit": 8 + }) + ); + } + #[tokio::test] async fn tool_error_response_includes_structured_context() { let response = tool_error_response( diff --git a/apps/mcp-server/src/main.rs b/apps/mcp-server/src/main.rs index 2a5514e..42ddb6a 100644 --- a/apps/mcp-server/src/main.rs +++ b/apps/mcp-server/src/main.rs @@ -6,7 +6,7 @@ mod session; use std::{env, net::SocketAddr, time::Duration}; use crank_registry::{PostgresPoolConfig, PostgresRegistry}; -use crank_runtime::SecretCrypto; +use crank_runtime::{RuntimeExecutor, RuntimeLimits, SecretCrypto}; use sqlx::postgres::PgConnectOptions; use tokio::net::TcpListener; use tracing::info; @@ -31,16 +31,22 @@ async fn main() -> Result<(), Box> { .unwrap_or_else(|| Duration::from_secs(5)); let socket_addr: SocketAddr = bind_addr.parse()?; let pool_config = PostgresPoolConfig::from_env()?; + let runtime_limits = RuntimeLimits::from_env()?; let registry = PostgresRegistry::connect_with_options_and_pool_config( database_options_from_env()?, pool_config, ) .await?; let secret_crypto = SecretCrypto::new(&env::var("CRANK_MASTER_KEY")?)?; - let app = build_app(registry, refresh_interval, base_url, secret_crypto); + let runtime = RuntimeExecutor::with_limits(runtime_limits); + let app = build_app(registry, refresh_interval, base_url, secret_crypto, runtime); let listener = TcpListener::bind(socket_addr).await?; info!( + runtime_max_concurrent_unary = runtime_limits.max_concurrent_unary, + runtime_max_concurrent_window = runtime_limits.max_concurrent_window, + runtime_max_concurrent_sessions = runtime_limits.max_concurrent_sessions, + runtime_max_concurrent_jobs = runtime_limits.max_concurrent_jobs, max_connections = pool_config.max_connections, min_connections = pool_config.min_connections, acquire_timeout_ms = pool_config.acquire_timeout_ms, @@ -106,7 +112,7 @@ mod tests { CreateAgentRequest, CreatePlatformApiKeyRequest, ListInvocationLogsQuery, PostgresRegistry, PublishAgentRequest, PublishRequest, }; - use crank_runtime::SecretCrypto; + use crank_runtime::{RuntimeExecutor, SecretCrypto}; use crank_schema::{Schema, SchemaKind}; use futures_util::stream; use serde_json::{Value, json}; @@ -173,6 +179,7 @@ mod tests { refresh_interval, public_base_url, SecretCrypto::new("test-master-key").unwrap(), + RuntimeExecutor::new(), ) } diff --git a/crates/crank-runtime/Cargo.toml b/crates/crank-runtime/Cargo.toml index a616160..8336a43 100644 --- a/crates/crank-runtime/Cargo.toml +++ b/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 diff --git a/crates/crank-runtime/src/error.rs b/crates/crank-runtime/src/error.rs index 8e602c8..25fcfbc 100644 --- a/crates/crank-runtime/src/error.rs +++ b/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}")] diff --git a/crates/crank-runtime/src/executor.rs b/crates/crank-runtime/src/executor.rs index be9311a..72d996c 100644 --- a/crates/crank-runtime/src/executor.rs +++ b/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, + window_limiter: Arc, + session_limiter: Arc, + job_limiter: Arc, } 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 { 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 { 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 { 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 { + 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 { + try_acquire_limit( + Arc::clone(&self.window_limiter), + "window", + self.limits.max_concurrent_window, + ) + } + + fn acquire_session_permit(&self) -> Result { + 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, + kind: &'static str, + limit: usize, +) -> Result { + 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, diff --git a/crates/crank-runtime/src/lib.rs b/crates/crank-runtime/src/lib.rs index 9e9c35c..383752c 100644 --- a/crates/crank-runtime/src/lib.rs +++ b/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; diff --git a/crates/crank-runtime/src/limits.rs b/crates/crank-runtime/src/limits.rs new file mode 100644 index 0000000..d284fed --- /dev/null +++ b/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 { + 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 { + match env::var(name) { + Ok(raw) => { + let value = + raw.parse::() + .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"); + } + } +} diff --git a/docs/runtime-config.md b/docs/runtime-config.md index 9db13ec..111aa45 100644 --- a/docs/runtime-config.md +++ b/docs/runtime-config.md @@ -73,6 +73,10 @@ var/crank/ - `CRANK_ADMIN_BIND` - `CRANK_MCP_BIND` - `CRANK_MCP_REFRESH_MS` +- `CRANK_RUNTIME_MAX_CONCURRENT_UNARY` +- `CRANK_RUNTIME_MAX_CONCURRENT_WINDOW` +- `CRANK_RUNTIME_MAX_CONCURRENT_SESSIONS` +- `CRANK_RUNTIME_MAX_CONCURRENT_JOBS` - `CRANK_LOG_LEVEL` - `CRANK_MASTER_KEY` - `CRANK_BASE_URL` @@ -86,6 +90,13 @@ var/crank/ - `CRANK_MCP_REFRESH_MS=5000` +Стартовые значения для runtime concurrency limits: + +- `CRANK_RUNTIME_MAX_CONCURRENT_UNARY=64` +- `CRANK_RUNTIME_MAX_CONCURRENT_WINDOW=16` +- `CRANK_RUNTIME_MAX_CONCURRENT_SESSIONS=16` +- `CRANK_RUNTIME_MAX_CONCURRENT_JOBS=16` + ## 6. Логирование и трассировка Для MVP нужно использовать: @@ -180,6 +191,13 @@ runtime env-переменной. Это значит, что: - `POSTGRES_IDLE_TIMEOUT_MS=600000` - `POSTGRES_MAX_LIFETIME_MS=1800000` +Для runtime concurrency используются явные defaults: + +- `CRANK_RUNTIME_MAX_CONCURRENT_UNARY=64` +- `CRANK_RUNTIME_MAX_CONCURRENT_WINDOW=16` +- `CRANK_RUNTIME_MAX_CONCURRENT_SESSIONS=16` +- `CRANK_RUNTIME_MAX_CONCURRENT_JOBS=16` + `CRANK_DATABASE_URL` допускается только как backward-compatible fallback для локальных тестов и переходного периода, но не как основная deployment-модель.