From c5cd6c1526156cb199ddb02c528077da8c840429 Mon Sep 17 00:00:00 2001 From: "a.tolmachev" Date: Sun, 3 May 2026 21:19:29 +0000 Subject: [PATCH] cache: add in-memory fallback stores --- Cargo.lock | 1 + TASKS.md | 2 +- crates/crank-runtime/Cargo.toml | 1 + crates/crank-runtime/src/cache.rs | 351 +++++++++++++++++++++++++++++- crates/crank-runtime/src/lib.rs | 5 +- 5 files changed, 354 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 142d978..1ef1e7a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -486,6 +486,7 @@ name = "crank-runtime" version = "0.1.0" dependencies = [ "aes-gcm", + "async-trait", "axum", "base64", "crank-adapter-graphql", diff --git a/TASKS.md b/TASKS.md index 778ef95..6195403 100644 --- a/TASKS.md +++ b/TASKS.md @@ -49,8 +49,8 @@ Progress: - done: - product/docs/runtime plan already define cache layer as optional and edition-aware - 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 - pending: - - Community fallback implementations - optional `Valkey` in Community manifests - external cache backend for managed / enterprise contours diff --git a/crates/crank-runtime/Cargo.toml b/crates/crank-runtime/Cargo.toml index 8336a43..73eb68a 100644 --- a/crates/crank-runtime/Cargo.toml +++ b/crates/crank-runtime/Cargo.toml @@ -7,6 +7,7 @@ version.workspace = true [dependencies] aes-gcm.workspace = true +async-trait = "0.1" base64.workspace = true crank-adapter-graphql = { path = "../crank-adapter-graphql" } crank-adapter-grpc = { path = "../crank-adapter-grpc" } diff --git a/crates/crank-runtime/src/cache.rs b/crates/crank-runtime/src/cache.rs index 0c00152..84dbfb6 100644 --- a/crates/crank-runtime/src/cache.rs +++ b/crates/crank-runtime/src/cache.rs @@ -1,7 +1,19 @@ -use std::{env, num::ParseIntError}; +use std::{ + collections::HashMap, + env, + num::ParseIntError, + sync::Arc, + time::{Duration, Instant}, +}; -use crank_core::CacheBackend; +use async_trait::async_trait; +use crank_core::{ + CacheBackend, CacheScope, CacheStoreError, CachedResponse, CoordinationStateStore, + CoordinationStateValue, RateLimitBucketState, RateLimitStateStore, ReplayGuardStatus, + ReplayGuardStore, ResponseCacheStore, +}; use thiserror::Error; +use tokio::sync::RwLock; #[derive(Clone, Debug, PartialEq, Eq)] pub struct RuntimeCacheConfig { @@ -38,6 +50,186 @@ impl RuntimeCacheConfig { } } +#[derive(Clone, Debug, Default)] +pub struct InMemoryResponseCacheStore { + entries: Arc>>>, +} + +#[derive(Clone, Debug, Default)] +pub struct InMemoryRateLimitStateStore { + entries: Arc>>>, +} + +#[derive(Clone, Debug, Default)] +pub struct InMemoryReplayGuardStore { + entries: Arc>>, +} + +#[derive(Clone, Debug, Default)] +pub struct InMemoryCoordinationStateStore { + entries: Arc>>>, +} + +#[derive(Clone, Debug)] +struct ExpiringValue { + value: T, + expires_at: Instant, +} + +#[async_trait] +impl ResponseCacheStore for InMemoryResponseCacheStore { + async fn get(&self, key: &str) -> Result, CacheStoreError> { + validate_key(key)?; + let now = Instant::now(); + let mut entries = self.entries.write().await; + retain_unexpired(&mut entries, now); + Ok(entries.get(key).map(|entry| entry.value.clone())) + } + + async fn put( + &self, + key: &str, + value: CachedResponse, + ttl: Duration, + ) -> Result<(), CacheStoreError> { + validate_key(key)?; + let expires_at = expiry_from_ttl(ttl)?; + let mut entries = self.entries.write().await; + entries.insert(key.to_owned(), ExpiringValue { value, expires_at }); + Ok(()) + } + + async fn delete(&self, key: &str) -> Result<(), CacheStoreError> { + validate_key(key)?; + let mut entries = self.entries.write().await; + entries.remove(key); + Ok(()) + } +} + +#[async_trait] +impl RateLimitStateStore for InMemoryRateLimitStateStore { + async fn get_bucket(&self, key: &str) -> Result, CacheStoreError> { + validate_key(key)?; + let now = Instant::now(); + let mut entries = self.entries.write().await; + retain_unexpired(&mut entries, now); + Ok(entries.get(key).map(|entry| entry.value)) + } + + async fn put_bucket( + &self, + key: &str, + value: RateLimitBucketState, + ttl: Duration, + ) -> Result<(), CacheStoreError> { + validate_key(key)?; + let expires_at = expiry_from_ttl(ttl)?; + let mut entries = self.entries.write().await; + entries.insert(key.to_owned(), ExpiringValue { value, expires_at }); + Ok(()) + } + + async fn delete_bucket(&self, key: &str) -> Result<(), CacheStoreError> { + validate_key(key)?; + let mut entries = self.entries.write().await; + entries.remove(key); + Ok(()) + } +} + +#[async_trait] +impl ReplayGuardStore for InMemoryReplayGuardStore { + async fn mark_seen( + &self, + key: &str, + ttl: Duration, + ) -> Result { + validate_key(key)?; + let expires_at = expiry_from_ttl(ttl)?; + let now = Instant::now(); + let mut entries = self.entries.write().await; + entries.retain(|_, entry_expires_at| *entry_expires_at > now); + if entries.get(key).is_some() { + return Ok(ReplayGuardStatus::AlreadySeen); + } + entries.insert(key.to_owned(), expires_at); + Ok(ReplayGuardStatus::Fresh) + } + + async fn clear(&self, key: &str) -> Result<(), CacheStoreError> { + validate_key(key)?; + let mut entries = self.entries.write().await; + entries.remove(key); + Ok(()) + } +} + +#[async_trait] +impl CoordinationStateStore for InMemoryCoordinationStateStore { + async fn get_value( + &self, + scope: CacheScope, + key: &str, + ) -> Result, CacheStoreError> { + validate_key(key)?; + let storage_key = scoped_key(scope, key); + let now = Instant::now(); + let mut entries = self.entries.write().await; + retain_unexpired(&mut entries, now); + Ok(entries.get(&storage_key).map(|entry| entry.value.clone())) + } + + async fn put_value( + &self, + scope: CacheScope, + key: &str, + value: CoordinationStateValue, + ttl: Duration, + ) -> Result<(), CacheStoreError> { + validate_key(key)?; + let expires_at = expiry_from_ttl(ttl)?; + let storage_key = scoped_key(scope, key); + let mut entries = self.entries.write().await; + entries.insert(storage_key, ExpiringValue { value, expires_at }); + Ok(()) + } + + async fn delete_value(&self, scope: CacheScope, key: &str) -> Result<(), CacheStoreError> { + validate_key(key)?; + let storage_key = scoped_key(scope, key); + let mut entries = self.entries.write().await; + entries.remove(&storage_key); + Ok(()) + } +} + +fn validate_key(key: &str) -> Result<(), CacheStoreError> { + if key.trim().is_empty() { + return Err(CacheStoreError::InvalidKey { + message: "cache key must not be empty".to_owned(), + }); + } + Ok(()) +} + +fn expiry_from_ttl(ttl: Duration) -> Result { + if ttl.is_zero() { + return Err(CacheStoreError::InvalidKey { + message: "cache ttl must be greater than zero".to_owned(), + }); + } + Ok(Instant::now() + ttl) +} + +fn retain_unexpired(entries: &mut HashMap>, now: Instant) { + entries.retain(|_, entry| entry.expires_at > now); +} + +fn scoped_key(scope: CacheScope, key: &str) -> String { + format!("{scope:?}:{key}") +} + fn parse_backend() -> Result { match env::var("CRANK_CACHE_BACKEND") { Ok(raw) => raw @@ -103,9 +295,20 @@ pub enum RuntimeCacheConfigError { #[cfg(test)] mod tests { - use crank_core::CacheBackend; + use std::time::Duration; - use super::{RuntimeCacheConfig, RuntimeCacheConfigError}; + use crank_core::CacheBackend; + use serde_json::json; + + use super::{ + InMemoryCoordinationStateStore, InMemoryRateLimitStateStore, InMemoryReplayGuardStore, + InMemoryResponseCacheStore, RuntimeCacheConfig, RuntimeCacheConfigError, + }; + use crank_core::{ + CacheScope, CacheStoreError, CachedHeader, CachedResponse, CoordinationStateStore, + CoordinationStateValue, RateLimitBucketState, RateLimitStateStore, ReplayGuardStatus, + ReplayGuardStore, ResponseCacheStore, + }; #[test] fn defaults_to_in_memory_cache_without_url() { @@ -177,4 +380,144 @@ mod tests { std::env::remove_var("CRANK_CACHE_DEFAULT_TTL_MS"); } } + + #[tokio::test] + async fn in_memory_response_cache_roundtrips_values() { + let store = InMemoryResponseCacheStore::default(); + let value = CachedResponse { + status: 200, + headers: vec![CachedHeader { + name: "content-type".to_owned(), + value: "application/json".to_owned(), + }], + body: br#"{"ok":true}"#.to_vec(), + }; + + store + .put("response:crm:list", value.clone(), Duration::from_secs(60)) + .await + .unwrap(); + + assert_eq!( + store.get("response:crm:list").await.unwrap(), + Some(value.clone()) + ); + + store.delete("response:crm:list").await.unwrap(); + + assert_eq!(store.get("response:crm:list").await.unwrap(), None); + } + + #[tokio::test] + async fn in_memory_rate_limit_store_roundtrips_bucket_state() { + let store = InMemoryRateLimitStateStore::default(); + let bucket = RateLimitBucketState { + tokens_micros: 1_500_000, + last_refill_unix_ms: 1_735_689_000_000, + }; + + store + .put_bucket("tenant:alpha", bucket, Duration::from_secs(30)) + .await + .unwrap(); + + assert_eq!( + store.get_bucket("tenant:alpha").await.unwrap(), + Some(bucket) + ); + + store.delete_bucket("tenant:alpha").await.unwrap(); + + assert_eq!(store.get_bucket("tenant:alpha").await.unwrap(), None); + } + + #[tokio::test] + async fn in_memory_replay_guard_marks_key_only_once_until_cleared() { + let store = InMemoryReplayGuardStore::default(); + + assert_eq!( + store + .mark_seen("token:nonce:1", Duration::from_secs(10)) + .await + .unwrap(), + ReplayGuardStatus::Fresh + ); + assert_eq!( + store + .mark_seen("token:nonce:1", Duration::from_secs(10)) + .await + .unwrap(), + ReplayGuardStatus::AlreadySeen + ); + + store.clear("token:nonce:1").await.unwrap(); + + assert_eq!( + store + .mark_seen("token:nonce:1", Duration::from_secs(10)) + .await + .unwrap(), + ReplayGuardStatus::Fresh + ); + } + + #[tokio::test] + async fn in_memory_coordination_store_scopes_keys() { + let store = InMemoryCoordinationStateStore::default(); + let response_value = CoordinationStateValue { + payload: json!({ "cursor": "abc" }), + }; + let session_value = CoordinationStateValue { + payload: json!({ "cursor": "xyz" }), + }; + + store + .put_value( + CacheScope::Response, + "job-1", + response_value.clone(), + Duration::from_secs(20), + ) + .await + .unwrap(); + store + .put_value( + CacheScope::Coordination, + "job-1", + session_value.clone(), + Duration::from_secs(20), + ) + .await + .unwrap(); + + assert_eq!( + store + .get_value(CacheScope::Response, "job-1") + .await + .unwrap(), + Some(response_value) + ); + assert_eq!( + store + .get_value(CacheScope::Coordination, "job-1") + .await + .unwrap(), + Some(session_value) + ); + } + + #[tokio::test] + async fn in_memory_stores_reject_empty_keys() { + let response_store = InMemoryResponseCacheStore::default(); + let replay_store = InMemoryReplayGuardStore::default(); + + let error = response_store.get("").await.unwrap_err(); + assert!(matches!(error, CacheStoreError::InvalidKey { .. })); + + let error = replay_store + .mark_seen("", Duration::from_secs(1)) + .await + .unwrap_err(); + assert!(matches!(error, CacheStoreError::InvalidKey { .. })); + } } diff --git a/crates/crank-runtime/src/lib.rs b/crates/crank-runtime/src/lib.rs index 7060992..da91673 100644 --- a/crates/crank-runtime/src/lib.rs +++ b/crates/crank-runtime/src/lib.rs @@ -12,7 +12,10 @@ mod secret_crypto; mod streaming; pub use auth::ResolvedAuth; -pub use cache::{RuntimeCacheConfig, RuntimeCacheConfigError}; +pub use cache::{ + InMemoryCoordinationStateStore, InMemoryRateLimitStateStore, InMemoryReplayGuardStore, + InMemoryResponseCacheStore, RuntimeCacheConfig, RuntimeCacheConfigError, +}; pub use error::RuntimeError; pub use executor::RuntimeExecutor; pub use limits::{RuntimeLimits, RuntimeLimitsConfigError};