amalgam/
distributed.rs

1//! L2 distributed cache: the wire envelope, a serializer abstraction, the
2//! [`DistributedCache`] backend trait, and an in-memory reference backend.
3//!
4//! FusionCache's L2 is any `IDistributedCache`; here it is any implementor of
5//! [`DistributedCache`] (a byte-oriented key/value store with TTL). Values cross
6//! the wire as a [`DistributedEntry`] — value plus the metadata needed to
7//! reconstruct freshness on another node — encoded by a [`DistributedSerializer`].
8//!
9//! A Redis-backed backend is intentionally left to a feature-gated adapter; the
10//! [`InMemoryDistributedCache`] here is a faithful reference used by tests and
11//! single-process multi-instance scenarios.
12
13use std::sync::Arc;
14use std::time::Duration;
15
16use async_trait::async_trait;
17use dashmap::DashMap;
18use serde::de::DeserializeOwned;
19use serde::{Deserialize, Serialize};
20
21use crate::entry::Entry;
22use crate::error::{Error, Result};
23use crate::tags::{Tag, collect_tags};
24use crate::time::{Clock, Timestamp};
25
26/// The serializable L2 envelope: a value together with the metadata required to
27/// rebuild its freshness/fail-safe state on any node.
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub struct DistributedEntry<V> {
30    /// The cached value.
31    pub value: V,
32    /// Creation tick (for tag-marker comparison).
33    pub created_ticks: i64,
34    /// Logical-expiration tick.
35    pub logical_expiration_ticks: i64,
36    /// Physical-expiration tick (fail-safe boundary).
37    pub physical_expiration_ticks: i64,
38    /// Whether the value came from a fail-safe activation.
39    pub is_from_fail_safe: bool,
40    /// Optional `ETag` for conditional refresh.
41    pub etag: Option<String>,
42    /// Optional `LastModified` tick for conditional refresh.
43    pub last_modified_ticks: Option<i64>,
44    /// The tags attached to the entry.
45    pub tags: Vec<String>,
46}
47
48impl<V: Clone> DistributedEntry<V> {
49    /// Captures an in-memory [`Entry`] as a wire envelope.
50    #[must_use]
51    pub fn from_entry(entry: &Entry<V>) -> Self {
52        let meta = entry.meta();
53        Self {
54            value: entry.value_cloned(),
55            created_ticks: meta.created().ticks(),
56            logical_expiration_ticks: meta.logical_expiration().ticks(),
57            physical_expiration_ticks: meta.physical_expiration().ticks(),
58            is_from_fail_safe: meta.is_from_fail_safe(),
59            etag: meta.etag().map(str::to_owned),
60            last_modified_ticks: meta.last_modified().map(Timestamp::ticks),
61            tags: meta.tags().iter().map(|t| t.as_str().to_owned()).collect(),
62        }
63    }
64
65    /// Rebuilds an in-memory [`Entry`] from the wire envelope, relative to `now`.
66    #[must_use]
67    pub fn into_entry(self, now: Timestamp) -> Entry<V> {
68        let tags: Box<[Tag]> = collect_tags(self.tags);
69        Entry::rehydrate(
70            self.value,
71            Timestamp::from_ticks(self.created_ticks),
72            Timestamp::from_ticks(self.logical_expiration_ticks),
73            Timestamp::from_ticks(self.physical_expiration_ticks),
74            self.is_from_fail_safe,
75            self.etag,
76            self.last_modified_ticks.map(Timestamp::from_ticks),
77            tags,
78            now,
79        )
80    }
81}
82
83/// Encodes and decodes [`DistributedEntry`] values for the wire.
84///
85/// This is object-safe (no generic methods), so a cache holds it as
86/// `Arc<dyn DistributedSerializer<V>>` and the serialization format is fully
87/// pluggable.
88pub trait DistributedSerializer<V>: Send + Sync {
89    /// Serializes an envelope to bytes.
90    ///
91    /// # Errors
92    /// Returns [`Error::Serialization`] if encoding fails.
93    fn serialize(&self, entry: &DistributedEntry<V>) -> Result<Vec<u8>>;
94
95    /// Deserializes an envelope from bytes.
96    ///
97    /// # Errors
98    /// Returns [`Error::Deserialization`] if decoding fails.
99    fn deserialize(&self, bytes: &[u8]) -> Result<DistributedEntry<V>>;
100}
101
102/// A JSON serializer built on `serde_json`.
103#[derive(Debug, Clone, Copy, Default)]
104pub struct JsonSerializer;
105
106impl<V> DistributedSerializer<V> for JsonSerializer
107where
108    V: Serialize + DeserializeOwned,
109{
110    fn serialize(&self, entry: &DistributedEntry<V>) -> Result<Vec<u8>> {
111        serde_json::to_vec(entry).map_err(|e| Error::Serialization(e.to_string()))
112    }
113
114    fn deserialize(&self, bytes: &[u8]) -> Result<DistributedEntry<V>> {
115        serde_json::from_slice(bytes).map_err(|e| Error::Deserialization(e.to_string()))
116    }
117}
118
119/// A byte-oriented L2 distributed cache backend.
120///
121/// Implement this over Redis, Memcached, a database, etc. The cache layer adds
122/// serialization, fail-safe and stampede protection on top.
123#[async_trait]
124pub trait DistributedCache: Send + Sync {
125    /// Reads the bytes stored at `key`, if present and unexpired.
126    ///
127    /// # Errors
128    /// Returns [`Error::Distributed`] on backend failure.
129    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>>;
130
131    /// Writes `value` at `key` with an optional TTL.
132    ///
133    /// # Errors
134    /// Returns [`Error::Distributed`] on backend failure.
135    async fn set(&self, key: &str, value: Vec<u8>, ttl: Option<Duration>) -> Result<()>;
136
137    /// Removes `key`.
138    ///
139    /// # Errors
140    /// Returns [`Error::Distributed`] on backend failure.
141    async fn remove(&self, key: &str) -> Result<()>;
142}
143
144/// An in-memory reference [`DistributedCache`] — a concurrent map with TTL.
145///
146/// Share one instance between multiple [`Cache`](crate::Cache) instances (via
147/// `Arc`) to simulate several nodes pointing at the same L2 within one process.
148#[derive(Clone)]
149pub struct InMemoryDistributedCache {
150    map: Arc<DashMap<String, StoredBytes>>,
151    clock: Arc<dyn Clock>,
152}
153
154#[derive(Clone)]
155struct StoredBytes {
156    bytes: Vec<u8>,
157    expires_at: Option<Timestamp>,
158}
159
160impl InMemoryDistributedCache {
161    /// Creates an empty backend using the given clock for TTL accounting.
162    #[must_use]
163    pub fn new(clock: Arc<dyn Clock>) -> Self {
164        Self {
165            map: Arc::new(DashMap::new()),
166            clock,
167        }
168    }
169}
170
171#[async_trait]
172impl DistributedCache for InMemoryDistributedCache {
173    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
174        let now = self.clock.now();
175        // Resolve to an owned value so no DashMap guard is held across `remove`.
176        let hit = self.map.get(key).and_then(|stored| {
177            if stored.expires_at.is_none_or(|exp| now.is_before(exp)) {
178                Some(stored.bytes.clone())
179            } else {
180                None
181            }
182        });
183        if hit.is_none() {
184            // Absent or expired: drop any expired entry lazily.
185            self.map.remove(key);
186        }
187        Ok(hit)
188    }
189
190    async fn set(&self, key: &str, value: Vec<u8>, ttl: Option<Duration>) -> Result<()> {
191        let expires_at = ttl.map(|d| self.clock.now().saturating_add(d));
192        self.map.insert(
193            key.to_owned(),
194            StoredBytes {
195                bytes: value,
196                expires_at,
197            },
198        );
199        Ok(())
200    }
201
202    async fn remove(&self, key: &str) -> Result<()> {
203        self.map.remove(key);
204        Ok(())
205    }
206}
207
208#[cfg(test)]
209mod tests {
210    use super::*;
211    use crate::time::ManualClock;
212
213    #[tokio::test]
214    async fn in_memory_l2_round_trips_and_expires() {
215        let clock = Arc::new(ManualClock::default());
216        let dyn_clock: Arc<dyn Clock> = clock.clone();
217        let l2 = InMemoryDistributedCache::new(dyn_clock);
218
219        l2.set("k", b"hello".to_vec(), Some(Duration::from_secs(10)))
220            .await
221            .unwrap();
222        assert_eq!(l2.get("k").await.unwrap(), Some(b"hello".to_vec()));
223
224        clock.advance(Duration::from_secs(11));
225        assert_eq!(l2.get("k").await.unwrap(), None);
226    }
227
228    #[test]
229    fn json_serializer_round_trips_envelope() {
230        let entry = DistributedEntry {
231            value: "v".to_owned(),
232            created_ticks: 1,
233            logical_expiration_ticks: 2,
234            physical_expiration_ticks: 3,
235            is_from_fail_safe: false,
236            etag: Some("e".to_owned()),
237            last_modified_ticks: None,
238            tags: vec!["t".to_owned()],
239        };
240        let ser = JsonSerializer;
241        let bytes = DistributedSerializer::<String>::serialize(&ser, &entry).unwrap();
242        let back = DistributedSerializer::<String>::deserialize(&ser, &bytes).unwrap();
243        assert_eq!(back.value, "v");
244        assert_eq!(back.tags, vec!["t".to_owned()]);
245    }
246}
amalgam/distributed.rs

amalgam/
distributed.rs
