diff --git a/Cargo.lock b/Cargo.lock index 1ef1e7a..0dea8f3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -90,6 +90,15 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "arc-swap" +version = "1.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207" +dependencies = [ + "rustversion", +] + [[package]] name = "argon2" version = "0.5.3" @@ -209,6 +218,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "backon" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cffb0e931875b666fc4fcb20fee52e9bbd1ef836fd9e9e04ec21555f9f85f7ef" +dependencies = [ + "fastrand", +] + [[package]] name = "base64" version = "0.22.1" @@ -298,6 +316,20 @@ dependencies = [ "inout", ] +[[package]] +name = "combine" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +dependencies = [ + "bytes", + "futures-core", + "memchr", + "pin-project-lite", + "tokio", + "tokio-util", +] + [[package]] name = "concurrent-queue" version = "2.5.0" @@ -499,6 +531,7 @@ dependencies = [ "crank-schema", "futures-util", "hkdf", + "redis", "serde", "serde_json", "sha2", @@ -1075,7 +1108,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2", + "socket2 0.6.3", "tokio", "tower-service", "tracing", @@ -1429,6 +1462,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-bigint" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-bigint-dig" version = "0.8.6" @@ -1867,7 +1910,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2", + "socket2 0.6.3", "thiserror", "tokio", "tracing", @@ -1904,7 +1947,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2", + "socket2 0.6.3", "tracing", "windows-sys 0.59.0", ] @@ -1989,6 +2032,30 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "redis" +version = "0.29.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1bc42f3a12fd4408ce64d8efef67048a924e543bd35c6591c0447fda9054695f" +dependencies = [ + "arc-swap", + "backon", + "bytes", + "combine", + "futures-channel", + "futures-util", + "itoa", + "num-bigint", + "percent-encoding", + "pin-project-lite", + "ryu", + "sha1_smol", + "socket2 0.5.10", + "tokio", + "tokio-util", + "url", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -2338,6 +2405,12 @@ dependencies = [ "digest", ] +[[package]] +name = "sha1_smol" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" + [[package]] name = "sha2" version = "0.10.9" @@ -2389,6 +2462,16 @@ dependencies = [ "serde", ] +[[package]] +name = "socket2" +version = "0.5.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + [[package]] name = "socket2" version = "0.6.3" @@ -2774,7 +2857,7 @@ dependencies = [ "libc", "mio", "pin-project-lite", - "socket2", + "socket2 0.6.3", "tokio-macros", "windows-sys 0.61.2", ] @@ -2859,7 +2942,7 @@ dependencies = [ "hyper-util", "percent-encoding", "pin-project", - "socket2", + "socket2 0.6.3", "sync_wrapper", "tokio", "tokio-stream", diff --git a/TASKS.md b/TASKS.md index eb7253d..9dacdae 100644 --- a/TASKS.md +++ b/TASKS.md @@ -51,8 +51,9 @@ Progress: - public cache contracts and runtime cache config now exist in `crank-core` / `crank-runtime` - Community fallback in-memory implementations now exist for response cache, rate-limit state, replay guard, and coordination state - Community deployment manifest now includes optional `valkey` profile and cache env wiring without making cache mandatory + - shared `Valkey/Redis` backend layer now exists in `crank-runtime` with config-driven store factory for managed / enterprise contours - pending: - - external cache backend for managed / enterprise contours + - wire selected cache stores into concrete runtime/admin/mcp paths so external backend reduces real load instead of existing only as reusable layer ## Planned diff --git a/crates/crank-runtime/Cargo.toml b/crates/crank-runtime/Cargo.toml index 73eb68a..7c14717 100644 --- a/crates/crank-runtime/Cargo.toml +++ b/crates/crank-runtime/Cargo.toml @@ -18,6 +18,7 @@ crank-core = { path = "../crank-core" } crank-mapping = { path = "../crank-mapping" } crank-schema = { path = "../crank-schema" } hkdf.workspace = true +redis = { version = "0.29", features = ["tokio-comp", "connection-manager"] } serde.workspace = true serde_json.workspace = true sha2.workspace = true diff --git a/crates/crank-runtime/src/cache.rs b/crates/crank-runtime/src/cache.rs index 84dbfb6..ebc7888 100644 --- a/crates/crank-runtime/src/cache.rs +++ b/crates/crank-runtime/src/cache.rs @@ -12,6 +12,8 @@ use crank_core::{ CoordinationStateValue, RateLimitBucketState, RateLimitStateStore, ReplayGuardStatus, ReplayGuardStore, ResponseCacheStore, }; +use redis::{Client, aio::ConnectionManager}; +use serde::de::DeserializeOwned; use thiserror::Error; use tokio::sync::RwLock; @@ -50,6 +52,21 @@ impl RuntimeCacheConfig { } } +#[derive(Clone)] +pub struct RuntimeCacheStores { + pub backend: CacheBackend, + pub response: Arc, + pub rate_limit: Arc, + pub replay_guard: Arc, + pub coordination: Arc, +} + +#[derive(Clone)] +pub struct RedisCacheStore { + backend: CacheBackend, + connection_manager: ConnectionManager, +} + #[derive(Clone, Debug, Default)] pub struct InMemoryResponseCacheStore { entries: Arc>>>, @@ -76,6 +93,160 @@ struct ExpiringValue { expires_at: Instant, } +impl RuntimeCacheStores { + pub async fn from_config( + config: &RuntimeCacheConfig, + ) -> Result { + match config.backend { + CacheBackend::Memory => { + let response = Arc::new(InMemoryResponseCacheStore::default()); + let rate_limit = Arc::new(InMemoryRateLimitStateStore::default()); + let replay_guard = Arc::new(InMemoryReplayGuardStore::default()); + let coordination = Arc::new(InMemoryCoordinationStateStore::default()); + Ok(Self { + backend: CacheBackend::Memory, + response, + rate_limit, + replay_guard, + coordination, + }) + } + CacheBackend::Valkey | CacheBackend::Redis => { + let url = config + .url + .as_deref() + .ok_or(RuntimeCacheStoreInitError::MissingUrl { + backend: config.backend, + })?; + let store = Arc::new(RedisCacheStore::connect(config.backend, url).await?); + Ok(Self { + backend: config.backend, + response: store.clone(), + rate_limit: store.clone(), + replay_guard: store.clone(), + coordination: store, + }) + } + } + } +} + +impl RedisCacheStore { + pub async fn connect( + backend: CacheBackend, + url: &str, + ) -> Result { + let client = + Client::open(url).map_err(|source| RuntimeCacheStoreInitError::InvalidUrl { + backend, + url: url.to_owned(), + details: source.to_string(), + })?; + let connection_manager = client.get_connection_manager().await.map_err(|source| { + RuntimeCacheStoreInitError::ConnectFailed { + backend, + details: source.to_string(), + } + })?; + Ok(Self { + backend, + connection_manager, + }) + } + + fn prefixed_key(&self, kind: &str, key: &str) -> String { + format!("crank:{kind}:{key}") + } + + fn serialize_value(&self, value: &T) -> Result, CacheStoreError> { + serde_json::to_vec(value).map_err(|source| CacheStoreError::Serialization { + message: source.to_string(), + }) + } + + fn deserialize_value(&self, value: &[u8]) -> Result { + serde_json::from_slice(value).map_err(|source| CacheStoreError::Serialization { + message: source.to_string(), + }) + } + + fn ttl_ms(&self, ttl: Duration) -> Result { + if ttl.is_zero() { + return Err(CacheStoreError::InvalidKey { + message: "cache ttl must be greater than zero".to_owned(), + }); + } + Ok(u64::try_from(ttl.as_millis()).unwrap_or(u64::MAX)) + } + + fn unavailable(&self, source: redis::RedisError) -> CacheStoreError { + CacheStoreError::Unavailable { + message: format!("{} backend error: {source}", self.backend), + } + } + + async fn get_json( + &self, + kind: &str, + key: &str, + ) -> Result, CacheStoreError> { + validate_key(key)?; + let storage_key = self.prefixed_key(kind, key); + let mut connection = self.connection_manager.clone(); + let encoded: Option> = redis::cmd("GET") + .arg(storage_key) + .query_async(&mut connection) + .await + .map_err(|source| self.unavailable(source))?; + encoded + .map(|bytes| self.deserialize_value(&bytes)) + .transpose() + } + + async fn put_json( + &self, + kind: &str, + key: &str, + value: &T, + ttl: Duration, + ) -> Result<(), CacheStoreError> { + validate_key(key)?; + let storage_key = self.prefixed_key(kind, key); + let encoded = self.serialize_value(value)?; + let ttl_ms = self.ttl_ms(ttl)?; + let mut connection = self.connection_manager.clone(); + redis::cmd("PSETEX") + .arg(storage_key) + .arg(ttl_ms) + .arg(encoded) + .query_async::<()>(&mut connection) + .await + .map_err(|source| self.unavailable(source))?; + Ok(()) + } + + async fn delete_kind_key(&self, kind: &str, key: &str) -> Result<(), CacheStoreError> { + validate_key(key)?; + let storage_key = self.prefixed_key(kind, key); + let mut connection = self.connection_manager.clone(); + redis::cmd("DEL") + .arg(storage_key) + .query_async::<()>(&mut connection) + .await + .map_err(|source| self.unavailable(source))?; + Ok(()) + } + + fn coordination_kind(scope: CacheScope) -> &'static str { + match scope { + CacheScope::Response => "coordination:response", + CacheScope::RateLimit => "coordination:rate_limit", + CacheScope::ReplayGuard => "coordination:replay_guard", + CacheScope::Coordination => "coordination:coordination", + } + } +} + #[async_trait] impl ResponseCacheStore for InMemoryResponseCacheStore { async fn get(&self, key: &str) -> Result, CacheStoreError> { @@ -204,6 +375,105 @@ impl CoordinationStateStore for InMemoryCoordinationStateStore { } } +#[async_trait] +impl ResponseCacheStore for RedisCacheStore { + async fn get(&self, key: &str) -> Result, CacheStoreError> { + self.get_json("response", key).await + } + + async fn put( + &self, + key: &str, + value: CachedResponse, + ttl: Duration, + ) -> Result<(), CacheStoreError> { + self.put_json("response", key, &value, ttl).await + } + + async fn delete(&self, key: &str) -> Result<(), CacheStoreError> { + self.delete_kind_key("response", key).await + } +} + +#[async_trait] +impl RateLimitStateStore for RedisCacheStore { + async fn get_bucket(&self, key: &str) -> Result, CacheStoreError> { + self.get_json("rate_limit", key).await + } + + async fn put_bucket( + &self, + key: &str, + value: RateLimitBucketState, + ttl: Duration, + ) -> Result<(), CacheStoreError> { + self.put_json("rate_limit", key, &value, ttl).await + } + + async fn delete_bucket(&self, key: &str) -> Result<(), CacheStoreError> { + self.delete_kind_key("rate_limit", key).await + } +} + +#[async_trait] +impl ReplayGuardStore for RedisCacheStore { + async fn mark_seen( + &self, + key: &str, + ttl: Duration, + ) -> Result { + validate_key(key)?; + let storage_key = self.prefixed_key("replay_guard", key); + let ttl_ms = self.ttl_ms(ttl)?; + let mut connection = self.connection_manager.clone(); + let result: Option = redis::cmd("SET") + .arg(storage_key) + .arg("1") + .arg("PX") + .arg(ttl_ms) + .arg("NX") + .query_async(&mut connection) + .await + .map_err(|source| self.unavailable(source))?; + Ok(if result.is_some() { + ReplayGuardStatus::Fresh + } else { + ReplayGuardStatus::AlreadySeen + }) + } + + async fn clear(&self, key: &str) -> Result<(), CacheStoreError> { + self.delete_kind_key("replay_guard", key).await + } +} + +#[async_trait] +impl CoordinationStateStore for RedisCacheStore { + async fn get_value( + &self, + scope: CacheScope, + key: &str, + ) -> Result, CacheStoreError> { + self.get_json(Self::coordination_kind(scope), key).await + } + + async fn put_value( + &self, + scope: CacheScope, + key: &str, + value: CoordinationStateValue, + ttl: Duration, + ) -> Result<(), CacheStoreError> { + self.put_json(Self::coordination_kind(scope), key, &value, ttl) + .await + } + + async fn delete_value(&self, scope: CacheScope, key: &str) -> Result<(), CacheStoreError> { + self.delete_kind_key(Self::coordination_kind(scope), key) + .await + } +} + fn validate_key(key: &str) -> Result<(), CacheStoreError> { if key.trim().is_empty() { return Err(CacheStoreError::InvalidKey { @@ -293,6 +563,23 @@ pub enum RuntimeCacheConfigError { MissingUrl { backend: CacheBackend }, } +#[derive(Debug, Error, PartialEq, Eq)] +pub enum RuntimeCacheStoreInitError { + #[error("{backend} backend requires CRANK_CACHE_URL")] + MissingUrl { backend: CacheBackend }, + #[error("invalid {backend} cache url {url}: {details}")] + InvalidUrl { + backend: CacheBackend, + url: String, + details: String, + }, + #[error("failed to connect to {backend} cache backend: {details}")] + ConnectFailed { + backend: CacheBackend, + details: String, + }, +} + #[cfg(test)] mod tests { use std::time::Duration; @@ -302,7 +589,8 @@ mod tests { use super::{ InMemoryCoordinationStateStore, InMemoryRateLimitStateStore, InMemoryReplayGuardStore, - InMemoryResponseCacheStore, RuntimeCacheConfig, RuntimeCacheConfigError, + InMemoryResponseCacheStore, RedisCacheStore, RuntimeCacheConfig, RuntimeCacheConfigError, + RuntimeCacheStoreInitError, RuntimeCacheStores, }; use crank_core::{ CacheScope, CacheStoreError, CachedHeader, CachedResponse, CoordinationStateStore, @@ -520,4 +808,93 @@ mod tests { .unwrap_err(); assert!(matches!(error, CacheStoreError::InvalidKey { .. })); } + + #[tokio::test] + async fn runtime_cache_stores_default_to_memory_backend() { + let stores = RuntimeCacheStores::from_config(&RuntimeCacheConfig::default()) + .await + .unwrap(); + + assert_eq!(stores.backend, CacheBackend::Memory); + + stores + .response + .put( + "response:health", + CachedResponse { + status: 200, + headers: vec![], + body: b"ok".to_vec(), + }, + Duration::from_secs(5), + ) + .await + .unwrap(); + + assert!( + stores + .response + .get("response:health") + .await + .unwrap() + .is_some() + ); + } + + #[test] + fn redis_cache_store_scopes_keys_by_kind() { + let key = "agent:primary"; + + let response_key = format!("crank:{}:{}", "response", key); + let coordination_key = format!( + "crank:{}:{}", + RedisCacheStore::coordination_kind(CacheScope::Coordination), + key + ); + + assert_eq!(response_key, "crank:response:agent:primary"); + assert_eq!( + coordination_key, + "crank:coordination:coordination:agent:primary" + ); + } + + #[test] + fn redis_cache_store_roundtrips_serialized_values() { + let response = CachedResponse { + status: 202, + headers: vec![CachedHeader { + name: "x-cache".to_owned(), + value: "hit".to_owned(), + }], + body: br#"{"queued":true}"#.to_vec(), + }; + + let encoded = serde_json::to_vec(&response).unwrap(); + let decoded: CachedResponse = serde_json::from_slice(&encoded).unwrap(); + + assert_eq!(decoded, response); + } + + #[test] + fn runtime_cache_stores_report_missing_external_url() { + let future = RuntimeCacheStores::from_config(&RuntimeCacheConfig { + backend: CacheBackend::Valkey, + url: None, + default_ttl_ms: None, + }); + + let runtime = tokio::runtime::Runtime::new().unwrap(); + let error = match runtime.block_on(future) { + Ok(_) => panic!("expected missing external cache url error"), + Err(error) => error, + }; + + assert_eq!( + error, + RuntimeCacheStoreInitError::MissingUrl { + backend: CacheBackend::Valkey + } + ); + } } diff --git a/crates/crank-runtime/src/lib.rs b/crates/crank-runtime/src/lib.rs index da91673..3d09949 100644 --- a/crates/crank-runtime/src/lib.rs +++ b/crates/crank-runtime/src/lib.rs @@ -14,7 +14,8 @@ mod streaming; pub use auth::ResolvedAuth; pub use cache::{ InMemoryCoordinationStateStore, InMemoryRateLimitStateStore, InMemoryReplayGuardStore, - InMemoryResponseCacheStore, RuntimeCacheConfig, RuntimeCacheConfigError, + InMemoryResponseCacheStore, RedisCacheStore, RuntimeCacheConfig, RuntimeCacheConfigError, + RuntimeCacheStoreInitError, RuntimeCacheStores, }; pub use error::RuntimeError; pub use executor::RuntimeExecutor;