gproxy_cache/
lib.rs

1//! Portable short-lived state and atomic coordination, without business entities.
2//! Notifications are lossy invalidation hints. Subscribe first, reload durable
3//! state on ResyncRequired/reconnect, and periodically check its revision.
4//! A transport error or cancellation after dispatch does not prove no mutation
5//! occurred. Non-idempotent operations are never automatically replayed here.
6
7#![forbid(unsafe_code)]
8
9use std::{fmt, time::Duration};
10
11#[cfg(feature = "memory")]
12mod memory;
13#[cfg(feature = "memory")]
14pub use memory::{MemoryCache, MemoryOptions};
15#[cfg(all(feature = "redis", not(target_arch = "wasm32")))]
16mod redis_backend;
17#[cfg(all(feature = "redis", not(target_arch = "wasm32")))]
18pub use redis_backend::{RedisCache, RedisOptions};
19
20pub type Result<T> = std::result::Result<T, CacheError>;
21
22#[derive(Debug, thiserror::Error)]
23pub enum CacheError {
24    #[error("invalid cache operation: {0}")]
25    Invalid(&'static str),
26    #[error("cache capacity exhausted")]
27    Capacity,
28    #[error("counter would underflow")]
29    Underflow,
30    #[error("cache state is corrupt or belongs to another implementation")]
31    Corrupt,
32    #[error("cache mutex poisoned")]
33    Poisoned,
34    #[error("notification subscription closed; resubscribe and reload durable state")]
35    Closed,
36    #[error("cannot generate cache token: {0}")]
37    Entropy(String),
38    #[cfg(all(feature = "redis", not(target_arch = "wasm32")))]
39    #[error("Redis operation failed: {0}")]
40    Redis(#[from] redis::RedisError),
41    #[error("cache connection timed out")]
42    Timeout,
43}
44
45/// Random identity for a value generation or a permit holder. Never reused by
46/// the library, even after expiry/deletion. This is not a monotonic fencing ID.
47#[derive(Clone, Copy, PartialEq, Eq, Hash)]
48pub struct Version([u8; 32]);
49impl Version {
50    pub fn from_bytes(bytes: [u8; 32]) -> Self {
51        Self(bytes)
52    }
53    pub fn as_bytes(&self) -> &[u8; 32] {
54        &self.0
55    }
56    #[cfg(any(
57        feature = "memory",
58        all(feature = "redis", not(target_arch = "wasm32"))
59    ))]
60    fn fresh() -> Result<Self> {
61        let mut bytes = [0; 32];
62        getrandom::fill(&mut bytes).map_err(|e| CacheError::Entropy(e.to_string()))?;
63        Ok(Self(bytes))
64    }
65}
66impl fmt::Debug for Version {
67    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
68        f.write_str("Version(..)")
69    }
70}
71
72#[derive(Clone, PartialEq, Eq)]
73pub struct Entry {
74    pub value: Vec<u8>,
75    pub version: Version,
76}
77impl fmt::Debug for Entry {
78    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
79        f.debug_struct("Entry")
80            .field("bytes", &self.value.len())
81            .field("version", &self.version)
82            .finish()
83    }
84}
85#[derive(Clone)]
86pub struct Replacement {
87    pub value: Vec<u8>,
88    pub ttl: Duration,
89}
90#[derive(Clone, Copy, Debug, PartialEq, Eq)]
91pub enum CasOutcome {
92    Applied(Option<Version>),
93    Conflict,
94}
95
96/// Exact nonnegative counter, bounded by signed i64 storage. Expiry is fixed
97/// when the generation is first created; later increments do not extend it.
98#[derive(Clone, Copy, Debug, PartialEq, Eq)]
99pub struct Counter {
100    pub value: u64,
101    pub generation: Version,
102}
103#[derive(Clone, Copy, Debug, PartialEq, Eq)]
104pub enum IncrementOutcome {
105    Applied(Counter),
106    Limited { current: u64 },
107}
108
109#[derive(Clone, Debug, PartialEq, Eq)]
110pub enum Notification {
111    /// Application-defined invalidation payload, not a durable event log.
112    Message(Vec<u8>),
113    /// Initial subscription, lag or invalid message. Reload authoritative state.
114    ResyncRequired,
115}
116
117#[async_trait::async_trait]
118pub trait Subscription: Send {
119    /// Cancel-safe for both supplied backends. Closed subscriptions must be
120    /// replaced; each replacement starts with ResyncRequired.
121    async fn recv(&mut self) -> Result<Notification>;
122}
123
124/// Each method is atomic for its one key. There are no cross-key transactions.
125/// KV values, counters, permits and notification topics have separate domains.
126#[async_trait::async_trait]
127pub trait Cache: Send + Sync {
128    async fn get(&self, key: &str) -> Result<Option<Entry>>;
129    /// `get` for each key, answered in the order asked. Each key is read
130    /// atomically on its own, exactly as `get` reads it; the batch is not a
131    /// snapshot across keys. It exists so a caller that needs many keys at
132    /// once pays one lock, round trip or query instead of one per key.
133    async fn get_many(&self, keys: &[String]) -> Result<Vec<Option<Entry>>> {
134        let mut entries = Vec::with_capacity(keys.len());
135        for key in keys {
136            entries.push(self.get(key).await?);
137        }
138        Ok(entries)
139    }
140    async fn put(&self, key: &str, value: Vec<u8>, ttl: Duration) -> Result<Version>;
141    async fn delete(&self, key: &str) -> Result<bool>;
142    /// None expects absence, including expiry. None replacement deletes.
143    async fn compare_exchange(
144        &self,
145        key: &str,
146        expected: Option<Version>,
147        replacement: Option<Replacement>,
148    ) -> Result<CasOutcome>;
149    async fn counter(&self, key: &str) -> Result<Option<Counter>>;
150    async fn increment(
151        &self,
152        key: &str,
153        amount: u64,
154        limit: u64,
155        ttl: Duration,
156    ) -> Result<IncrementOutcome>;
157    /// Generation check prevents a late decrement touching a new window.
158    /// This is not an idempotent receipt: the caller must release each charge
159    /// once and must not blindly retry after an ambiguous transport failure.
160    async fn decrement(
161        &self,
162        key: &str,
163        generation: Version,
164        amount: u64,
165    ) -> Result<Option<Counter>>;
166    /// Expiring concurrency permit. Use limit=1 for a refresh lease. All callers
167    /// for a resource must use the same limit. No automatic renewal is performed.
168    async fn acquire_permit(&self, key: &str, limit: u32, ttl: Duration)
169    -> Result<Option<Version>>;
170    async fn renew_permit(&self, key: &str, owner: Version, ttl: Duration) -> Result<bool>;
171    async fn release_permit(&self, key: &str, owner: Version) -> Result<bool>;
172    async fn acquire_lease(&self, key: &str, ttl: Duration) -> Result<Option<Version>> {
173        self.acquire_permit(key, 1, ttl).await
174    }
175    async fn publish(&self, topic: &str, payload: Vec<u8>) -> Result<()>;
176    /// Returns after registration. Initial ResyncRequired ensures changes
177    /// before registration are recovered by an authoritative reload.
178    async fn subscribe(&self, topic: &str) -> Result<Box<dyn Subscription>>;
179}
180
181#[derive(Clone, Debug)]
182pub struct Limits {
183    pub max_key_bytes: usize,
184    pub max_value_bytes: usize,
185    pub max_permits_per_key: u32,
186    pub notification_capacity: usize,
187}
188impl Default for Limits {
189    fn default() -> Self {
190        Self {
191            max_key_bytes: 1024,
192            max_value_bytes: 1024 * 1024,
193            max_permits_per_key: 10_000,
194            notification_capacity: 256,
195        }
196    }
197}
198#[cfg(any(
199    feature = "memory",
200    all(feature = "redis", not(target_arch = "wasm32"))
201))]
202impl Limits {
203    fn validate(&self) -> Result<()> {
204        if self.max_key_bytes == 0
205            || self.max_value_bytes == 0
206            || self.max_permits_per_key == 0
207            || self.notification_capacity == 0
208            || self.notification_capacity > 1_048_576
209        {
210            return Err(CacheError::Invalid(
211                "limits must be positive; notification capacity must be <= 1048576",
212            ));
213        }
214        Ok(())
215    }
216    fn key(&self, key: &str) -> Result<()> {
217        if key.is_empty() || key.len() > self.max_key_bytes {
218            return Err(CacheError::Invalid("empty or oversized key/topic"));
219        }
220        Ok(())
221    }
222    fn value(&self, value: &[u8]) -> Result<()> {
223        if value.len() > self.max_value_bytes {
224            return Err(CacheError::Invalid("oversized value/notification"));
225        }
226        Ok(())
227    }
228    fn permits(&self, limit: u32) -> Result<()> {
229        if limit == 0 || limit > self.max_permits_per_key {
230            return Err(CacheError::Invalid("invalid permit limit"));
231        }
232        Ok(())
233    }
234}
235
236// Lua sorted-set timestamps use exact IEEE-754 integers. This bound leaves
237// room for current Unix milliseconds and is far beyond practical cache TTLs.
238#[cfg(any(
239    feature = "memory",
240    all(feature = "redis", not(target_arch = "wasm32"))
241))]
242const MAX_TTL_MS: u64 = (1_u64 << 52) - 1;
243#[cfg(any(
244    feature = "memory",
245    all(feature = "redis", not(target_arch = "wasm32"))
246))]
247fn ttl_ms(ttl: Duration) -> Result<u64> {
248    let ms = ttl.as_nanos().div_ceil(1_000_000);
249    if ms == 0 || ms > u128::from(MAX_TTL_MS) {
250        return Err(CacheError::Invalid(
251            "TTL must be positive and representable in exact milliseconds",
252        ));
253    }
254    Ok(ms as u64)
255}
256#[cfg(any(
257    feature = "memory",
258    all(feature = "redis", not(target_arch = "wasm32"))
259))]
260fn counter_args(amount: u64, limit: u64) -> Result<()> {
261    if amount == 0 || amount > i64::MAX as u64 || limit > i64::MAX as u64 {
262        return Err(CacheError::Invalid(
263            "counter amounts must be positive and values <= i64::MAX",
264        ));
265    }
266    Ok(())
267}
gproxy_cache/lib.rs

gproxy_cache/
lib.rs
