cratefield_runtime_native/ports/
kv.rs

1//! `KeyValue` over Redis, plus the shared-connection convenience both
2//! Redis ports are wired through (`redis_from_env`). The honest
3//! differences from a Workers KV namespace are listed on [`RedisKv`].
4
5use std::sync::Arc;
6use std::time::Duration;
7
8use async_trait::async_trait;
9use cratefield_core::{Config, KeyValue, KvError};
10use redis::aio::ConnectionManager;
11
12use crate::ports::rate_limit::RedisRateLimiter;
13
14/// `KeyValue` over Redis (one shared `ConnectionManager`, cloned per
15/// call — a cheap channel copy, not a new connection).
16///
17/// # Differences from the Workers KV binding (documented, not papered
18/// over)
19///
20/// | | Workers KV | `RedisKv` |
21/// |---|---|---|
22/// | Consistency | eventually consistent; a write can take up to ~60 s to propagate to other colos, and reads may hit a stale per-colo cache | strongly consistent single Redis: a read after a write sees the write |
23/// | Failure mode | KV outages degrade per colo; values survive Redis-down | Redis is a single point of failure and of latency for the port |
24/// | TTL | minimum 60 s | any duration, millisecond granularity (`PX`) |
25/// | Value size | 25 MiB per value | 512 MiB per string, but the port is meant for small values |
26/// | Key space | the binding's own namespace | one Redis database shared with the rate limiter — keys are prefixed `kv:` here to keep the two apart |
27///
28/// The port has no list/scan method, so KV's list operations have no
29/// counterpart to mismatch.
30pub struct RedisKv(ConnectionManager);
31
32impl RedisKv {
33    pub fn new(conn: ConnectionManager) -> Self {
34        Self(conn)
35    }
36}
37
38fn op(err: &redis::RedisError) -> KvError {
39    KvError::Operation(err.to_string())
40}
41
42#[async_trait]
43impl KeyValue for RedisKv {
44    async fn get(&self, key: &str) -> Result<Option<String>, KvError> {
45        let mut conn = self.0.clone();
46        redis::cmd("GET")
47            .arg(format!("kv:{key}"))
48            .query_async(&mut conn)
49            .await
50            .map_err(|err| op(&err))
51    }
52
53    async fn put(&self, key: &str, value: &str, ttl: Option<Duration>) -> Result<(), KvError> {
54        let mut conn = self.0.clone();
55        let mut cmd = redis::cmd("SET");
56        cmd.arg(format!("kv:{key}")).arg(value);
57        if let Some(ttl) = ttl {
58            // PX (milliseconds) rather than EX (seconds): the port takes
59            // a Duration and the port's granularity survives intact.
60            let millis = i64::try_from(ttl.as_millis()).unwrap_or(i64::MAX).max(1);
61            cmd.arg("PX").arg(millis);
62        }
63        cmd.query_async::<()>(&mut conn)
64            .await
65            .map_err(|err| op(&err))
66    }
67
68    async fn delete(&self, key: &str) -> Result<(), KvError> {
69        let mut conn = self.0.clone();
70        redis::cmd("DEL")
71            .arg(format!("kv:{key}"))
72            .query_async::<i64>(&mut conn)
73            .await
74            .map_err(|err| op(&err))?;
75        Ok(())
76    }
77}
78
79/// Both Redis ports over one shared connection, what a venture gets from
80/// `REDIS_URL` when it is set.
81pub struct RedisBundle {
82    pub rate_limiter: Arc<RedisRateLimiter>,
83    pub kv: Arc<RedisKv>,
84}
85
86#[derive(Debug, thiserror::Error)]
87pub enum RedisPortError {
88    /// `REDIS_URL` is not a valid Redis URL, or the manager could not
89    /// establish its first connection.
90    #[error("redis port setup failed: {0}")]
91    Connection(String),
92}
93
94/// Wires both Redis ports from config: `REDIS_URL` opens a client and
95/// one reconnecting [`ConnectionManager`] shared (cloned) by the two
96/// adapters; `RATE_LIMIT_MAX` (default `100`) and
97/// `RATE_LIMIT_PERIOD_SECS` (default `60`) configure the limiter's
98/// window. `Ok(None)` when `REDIS_URL` is unset — the ports stay absent
99/// and a module that requires one fails `Harness::build` by name.
100///
101/// # Errors
102///
103/// [`RedisPortError::Connection`] when the URL cannot be parsed or the
104/// first connection fails (fail fast at startup, not on the first
105/// request).
106pub async fn redis_from_env(config: &dyn Config) -> Result<Option<RedisBundle>, RedisPortError> {
107    let Some(url) = config.get("REDIS_URL") else {
108        return Ok(None);
109    };
110    let client = redis::Client::open(url.as_str())
111        .map_err(|err| RedisPortError::Connection(err.to_string()))?;
112    let manager = client
113        .get_connection_manager()
114        .await
115        .map_err(|err| RedisPortError::Connection(err.to_string()))?;
116    let rate_limiter = Arc::new(RedisRateLimiter::from_env(manager.clone(), config));
117    let kv = Arc::new(RedisKv::new(manager));
118    Ok(Some(RedisBundle { rate_limiter, kv }))
119}
cratefield_runtime_native/ports/kv.rs

cratefield_runtime_native/ports/
kv.rs
