cache: add in-memory fallback stores

Cargo.lock

@@ -486,6 +486,7 @@ name = "crank-runtime"
 version = "0.1.0"
 dependencies = [
  "aes-gcm",
+ "async-trait",
  "axum",
  "base64",
  "crank-adapter-graphql",

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
 

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" }

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<RwLock<HashMap<String, ExpiringValue<CachedResponse>>>>,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct InMemoryRateLimitStateStore {
+    entries: Arc<RwLock<HashMap<String, ExpiringValue<RateLimitBucketState>>>>,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct InMemoryReplayGuardStore {
+    entries: Arc<RwLock<HashMap<String, Instant>>>,
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct InMemoryCoordinationStateStore {
+    entries: Arc<RwLock<HashMap<String, ExpiringValue<CoordinationStateValue>>>>,
+}
+
+#[derive(Clone, Debug)]
+struct ExpiringValue<T> {
+    value: T,
+    expires_at: Instant,
+}
+
+#[async_trait]
+impl ResponseCacheStore for InMemoryResponseCacheStore {
+    async fn get(&self, key: &str) -> Result<Option<CachedResponse>, 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<Option<RateLimitBucketState>, 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<ReplayGuardStatus, CacheStoreError> {
+        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<Option<CoordinationStateValue>, 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<Instant, CacheStoreError> {
+    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<T>(entries: &mut HashMap<String, ExpiringValue<T>>, now: Instant) {
+    entries.retain(|_, entry| entry.expires_at > now);
+}
+
+fn scoped_key(scope: CacheScope, key: &str) -> String {
+    format!("{scope:?}:{key}")
+}
+
 fn parse_backend() -> Result<CacheBackend, RuntimeCacheConfigError> {
     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 { .. }));
+    }
 }

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};