laterite_core/cache/
mod.rs

1//! The cache: values kept for a time, invalidated by generation, shared by
2//! every process of a deployment.
3//!
4//! A value goes in under a key with a lifetime and the names of the
5//! namespaces it was built from; a read hands it back while every one of
6//! those namespaces still carries the generation it had when the value was
7//! stored. A write path bumps its namespace, and every value built from it
8//! misses from then on, in this process at once and in the others within the
9//! poll interval. Nothing is enumerated and nothing is listed by hand.
10//!
11//! ```ignore
12//! let menu: Vec<Item> = ctx
13//!     .cache()
14//!     .remember("acme.blog::menu", Duration::from_secs(3600), &["records::posts"], || async {
15//!         load_menu(ctx.db()).await
16//!     })
17//!     .await?;
18//! ```
19//!
20//! Values live in a [`ValueStore`] (`memory`, `null` or `redis`) and the
21//! generations and locks in a [`Signals`] store (the database, or Redis with
22//! the `redis` store). [`from_config`] assembles the pair a deployment
23//! configured; [`Cache::local`] is one process alone, for tests and tools.
24
25use std::collections::{HashMap, HashSet};
26use std::future::Future;
27use std::sync::{Arc, Mutex, RwLock};
28use std::time::Duration;
29
30use async_trait::async_trait;
31use serde::de::DeserializeOwned;
32use serde::Serialize;
33
34use crate::config::CacheConfig;
35use crate::error::{CoreError, CoreResult};
36use crate::{Db, Module, ModuleId};
37
38mod memory;
39pub mod migrations;
40mod null;
41#[cfg(feature = "redis")]
42pub mod redis;
43mod signals;
44
45pub use memory::{MemoryStore, DEFAULT_CAPACITY};
46pub use null::NullStore;
47pub use signals::{DbSignals, LocalSignals};
48
49/// The reserved namespace family of the record layer: `records::<entity>`,
50/// bumped after every save and delete of that entity. Any module may depend
51/// on one; none registers it.
52pub fn records(entity: &str) -> String {
53    format!("{RECORDS_PREFIX}{entity}")
54}
55
56const RECORDS_PREFIX: &str = "records::";
57
58/// A namespace a module registers, `registry.add(CacheNamespace::new("feed"))`;
59/// the framework qualifies it with the owning module's id, `acme.blog::feed`,
60/// so one module cannot claim another's.
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62pub struct CacheNamespace {
63    name: &'static str,
64}
65
66impl CacheNamespace {
67    pub const fn new(name: &'static str) -> Self {
68        Self { name }
69    }
70
71    /// The name as registered, without the owner.
72    pub fn name(&self) -> &'static str {
73        self.name
74    }
75
76    /// The full namespace, `owner::name`.
77    pub fn qualified(&self, owner: ModuleId) -> String {
78        format!("{}::{}", owner.as_str(), self.name)
79    }
80}
81
82/// The module that owns the generation and lock tables, listed among an
83/// application's built-ins.
84pub struct CacheModule;
85
86impl Module for CacheModule {
87    fn id(&self) -> ModuleId {
88        ModuleId::new(migrations::MODULE_ID)
89    }
90    fn migrations(&self) -> crate::MigrationSet {
91        migrations::migrations()
92    }
93}
94
95/// A stored value: its bytes and the generation of each namespace it was
96/// built from.
97#[derive(Debug, Clone, PartialEq, Eq)]
98pub struct Entry {
99    pub bytes: Vec<u8>,
100    pub deps: Vec<(String, u64)>,
101}
102
103/// Where values live. Every method answers for one key; a failure is the
104/// store's own (a connection lost), never a miss.
105#[async_trait]
106pub trait ValueStore: Send + Sync {
107    /// The driver's name, as `cache.store` names it.
108    fn name(&self) -> &'static str;
109    async fn get(&self, key: &str) -> CoreResult<Option<Entry>>;
110    async fn put(&self, key: &str, entry: Entry, ttl: Duration) -> CoreResult<()>;
111    /// Stores only when the key is absent; `true` when it did.
112    async fn add(&self, key: &str, entry: Entry, ttl: Duration) -> CoreResult<bool>;
113    /// `true` when a value was there.
114    async fn forget(&self, key: &str) -> CoreResult<bool>;
115    async fn flush(&self) -> CoreResult<()>;
116}
117
118/// Where generations and locks live, shared by every process.
119///
120/// A process keeps a replica of the generations: [`current`](Signals::current)
121/// answers from it without waiting, a bump through this process updates it at
122/// once, and [`refresh`](Signals::refresh) brings in the other processes'
123/// bumps.
124#[async_trait]
125pub trait Signals: Send + Sync {
126    fn name(&self) -> &'static str;
127    /// The generation this process last saw, or `None` before the namespace
128    /// was ever read here.
129    fn current(&self, namespace: &str) -> Option<u64>;
130    /// The generation, seeding the namespace when it is new.
131    async fn ensure(&self, namespace: &str) -> CoreResult<u64>;
132    /// Moves the namespace to a new generation and returns it.
133    async fn bump(&self, namespace: &str) -> CoreResult<u64>;
134    /// Reloads every generation from the shared store.
135    async fn refresh(&self) -> CoreResult<()>;
136    /// Every namespace the shared store knows, with its generation.
137    async fn known(&self) -> CoreResult<Vec<(String, u64)>>;
138    /// Takes the named lock for `owner` until `ttl` passes or it is released;
139    /// `false` while another owner holds it.
140    async fn lock(&self, name: &str, owner: &str, ttl: Duration) -> CoreResult<bool>;
141    /// Releases the lock when `owner` holds it; `true` when it did.
142    async fn unlock(&self, name: &str, owner: &str) -> CoreResult<bool>;
143}
144
145struct Inner {
146    values: Box<dyn ValueStore>,
147    signals: Box<dyn Signals>,
148    /// The namespaces a value may depend on and a write may bump: those
149    /// registered, plus the record family.
150    allowed: RwLock<HashSet<String>>,
151    /// One loader in flight per key: the second caller waits for the first
152    /// and reads what it stored.
153    flights: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
154}
155
156/// The cache handle. Cheap to clone: every clone is the same cache.
157#[derive(Clone)]
158pub struct Cache {
159    inner: Arc<Inner>,
160}
161
162impl std::fmt::Debug for Cache {
163    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
164        f.debug_struct("Cache")
165            .field("store", &self.inner.values.name())
166            .field("signals", &self.inner.signals.name())
167            .finish_non_exhaustive()
168    }
169}
170
171impl Cache {
172    /// A cache over the given stores.
173    pub fn new(values: impl ValueStore + 'static, signals: impl Signals + 'static) -> Self {
174        Self {
175            inner: Arc::new(Inner {
176                values: Box::new(values),
177                signals: Box::new(signals),
178                allowed: RwLock::new(HashSet::new()),
179                flights: Mutex::new(HashMap::new()),
180            }),
181        }
182    }
183
184    /// A cache for one process: values in memory, generations and locks
185    /// in memory too, nothing shared. For tests, tools and an application
186    /// assembled without the bootstrap.
187    pub fn local() -> Self {
188        Self::new(MemoryStore::new(), LocalSignals::new())
189    }
190
191    /// The configured value store's name.
192    pub fn store(&self) -> &'static str {
193        self.inner.values.name()
194    }
195
196    /// Registers a namespace, so values may depend on it and writes bump it.
197    /// Registering one twice is nothing.
198    pub fn allow(&self, namespace: impl Into<String>) {
199        self.inner
200            .allowed
201            .write()
202            .unwrap_or_else(|e| e.into_inner())
203            .insert(namespace.into());
204    }
205
206    /// Whether the namespace is registered or in the record family.
207    pub fn is_allowed(&self, namespace: &str) -> bool {
208        namespace.starts_with(RECORDS_PREFIX)
209            || self
210                .inner
211                .allowed
212                .read()
213                .unwrap_or_else(|e| e.into_inner())
214                .contains(namespace)
215    }
216
217    /// The registered namespaces, sorted.
218    pub fn namespaces(&self) -> Vec<String> {
219        let mut all: Vec<String> = self
220            .inner
221            .allowed
222            .read()
223            .unwrap_or_else(|e| e.into_inner())
224            .iter()
225            .cloned()
226            .collect();
227        all.sort();
228        all
229    }
230
231    fn check(&self, namespace: &str) -> CoreResult<()> {
232        if self.is_allowed(namespace) {
233            Ok(())
234        } else {
235            Err(CoreError::Validation(format!(
236                "cache namespace `{namespace}` is not registered"
237            )))
238        }
239    }
240
241    /// The value under `key` while every namespace it depends on is at the
242    /// generation it was stored at. A store failure reads as a miss and is
243    /// logged.
244    pub async fn get(&self, key: &str) -> Option<Vec<u8>> {
245        let entry = match self.inner.values.get(key).await {
246            Ok(entry) => entry?,
247            Err(e) => {
248                tracing::warn!(key, error = %e, "cache read failed");
249                return None;
250            }
251        };
252        if self.fresh(&entry).await {
253            Some(entry.bytes)
254        } else {
255            let _ = self.inner.values.forget(key).await;
256            None
257        }
258    }
259
260    async fn fresh(&self, entry: &Entry) -> bool {
261        for (namespace, generation) in &entry.deps {
262            let now = match self.inner.signals.current(namespace) {
263                Some(now) => now,
264                None => match self.inner.signals.ensure(namespace).await {
265                    Ok(now) => now,
266                    Err(e) => {
267                        tracing::warn!(namespace, error = %e, "cache generation read failed");
268                        return false;
269                    }
270                },
271            };
272            if now != *generation {
273                return false;
274            }
275        }
276        true
277    }
278
279    async fn deps_of(&self, deps: &[&str]) -> CoreResult<Vec<(String, u64)>> {
280        let mut out = Vec::with_capacity(deps.len());
281        for namespace in deps {
282            self.check(namespace)?;
283            out.push((namespace.to_string(), self.generation(namespace).await?));
284        }
285        Ok(out)
286    }
287
288    /// Stores `bytes` under `key` for `ttl`, built from `deps`. An
289    /// unregistered namespace in `deps` is refused; a store failure is
290    /// logged and the value is simply not kept.
291    pub async fn put(
292        &self,
293        key: &str,
294        bytes: Vec<u8>,
295        deps: &[&str],
296        ttl: Duration,
297    ) -> CoreResult<()> {
298        let deps = self.deps_of(deps).await?;
299        if let Err(e) = self.inner.values.put(key, Entry { bytes, deps }, ttl).await {
300            tracing::warn!(key, error = %e, "cache write failed");
301        }
302        Ok(())
303    }
304
305    /// Like [`Cache::put`], only when nothing is under `key`; `true` when
306    /// this call stored it.
307    pub async fn add(
308        &self,
309        key: &str,
310        bytes: Vec<u8>,
311        deps: &[&str],
312        ttl: Duration,
313    ) -> CoreResult<bool> {
314        let deps = self.deps_of(deps).await?;
315        match self.inner.values.add(key, Entry { bytes, deps }, ttl).await {
316            Ok(stored) => Ok(stored),
317            Err(e) => {
318                tracing::warn!(key, error = %e, "cache write failed");
319                Ok(false)
320            }
321        }
322    }
323
324    /// Drops the value under `key`; `true` when one was there.
325    pub async fn forget(&self, key: &str) -> bool {
326        match self.inner.values.forget(key).await {
327            Ok(had) => had,
328            Err(e) => {
329                tracing::warn!(key, error = %e, "cache forget failed");
330                false
331            }
332        }
333    }
334
335    /// Drops every value this process's store holds. Other processes keep
336    /// theirs; [`Cache::bump_all`] reaches them.
337    pub async fn flush(&self) {
338        if let Err(e) = self.inner.values.flush().await {
339            tracing::warn!(error = %e, "cache flush failed");
340        }
341    }
342
343    /// The JSON value under `key`, decoded as `T`.
344    pub async fn get_json<T: DeserializeOwned>(&self, key: &str) -> Option<T> {
345        let bytes = self.get(key).await?;
346        match serde_json::from_slice(&bytes) {
347            Ok(value) => Some(value),
348            Err(e) => {
349                tracing::warn!(key, error = %e, "cached value has another shape");
350                self.forget(key).await;
351                None
352            }
353        }
354    }
355
356    /// Stores `value` as JSON under `key`.
357    pub async fn put_json<T: Serialize + ?Sized>(
358        &self,
359        key: &str,
360        value: &T,
361        deps: &[&str],
362        ttl: Duration,
363    ) -> CoreResult<()> {
364        let bytes = serde_json::to_vec(value).map_err(|e| {
365            CoreError::Internal(format!("cache value `{key}` cannot serialize: {e}"))
366        })?;
367        self.put(key, bytes, deps, ttl).await
368    }
369
370    /// The value under `key`, or what `load` returns, stored for `ttl` and
371    /// built from `deps`. Concurrent callers for one key share one load.
372    pub async fn remember<T, F, Fut>(
373        &self,
374        key: &str,
375        ttl: Duration,
376        deps: &[&str],
377        load: F,
378    ) -> CoreResult<T>
379    where
380        T: Serialize + DeserializeOwned,
381        F: FnOnce() -> Fut,
382        Fut: Future<Output = CoreResult<T>>,
383    {
384        for namespace in deps {
385            self.check(namespace)?;
386        }
387        if let Some(value) = self.get_json(key).await {
388            return Ok(value);
389        }
390        let flight = {
391            let mut flights = self.inner.flights.lock().unwrap_or_else(|e| e.into_inner());
392            Arc::clone(flights.entry(key.to_string()).or_default())
393        };
394        let outcome = {
395            let _guard = flight.lock().await;
396            match self.get_json(key).await {
397                Some(value) => Ok(value),
398                None => match load().await {
399                    Ok(value) => {
400                        self.put_json(key, &value, deps, ttl).await?;
401                        Ok(value)
402                    }
403                    Err(e) => Err(e),
404                },
405            }
406        };
407        let mut flights = self.inner.flights.lock().unwrap_or_else(|e| e.into_inner());
408        if flights.get(key).is_some_and(|f| Arc::strong_count(f) <= 2) {
409            flights.remove(key);
410        }
411        outcome
412    }
413
414    /// The namespace's generation as this process last saw it, or `None`
415    /// before it was ever read here.
416    pub fn current(&self, namespace: &str) -> Option<u64> {
417        self.inner.signals.current(namespace)
418    }
419
420    /// The namespace's generation, seeded when new.
421    pub async fn generation(&self, namespace: &str) -> CoreResult<u64> {
422        self.check(namespace)?;
423        self.inner.signals.ensure(namespace).await
424    }
425
426    /// Moves the namespace on: every value built from it misses from now.
427    /// Returns the new generation. A failure is logged as well as returned,
428    /// since the write it follows already stands.
429    pub async fn bump(&self, namespace: &str) -> CoreResult<u64> {
430        self.check(namespace)?;
431        match self.inner.signals.bump(namespace).await {
432            Ok(generation) => {
433                tracing::debug!(namespace, generation, "cache namespace bumped");
434                Ok(generation)
435            }
436            Err(e) => {
437                tracing::error!(namespace, error = %e, "cache bump failed");
438                Err(e)
439            }
440        }
441    }
442
443    /// Bumps every namespace the shared store knows and drops this
444    /// process's values. Returns how many namespaces moved.
445    pub async fn bump_all(&self) -> CoreResult<usize> {
446        let known = self.inner.signals.known().await?;
447        for (namespace, _) in &known {
448            self.inner.signals.bump(namespace).await?;
449        }
450        self.flush().await;
451        Ok(known.len())
452    }
453
454    /// Every namespace the shared store knows, with its generation.
455    pub async fn known(&self) -> CoreResult<Vec<(String, u64)>> {
456        self.inner.signals.known().await
457    }
458
459    /// Brings in the other processes' bumps.
460    pub async fn refresh(&self) -> CoreResult<()> {
461        self.inner.signals.refresh().await
462    }
463
464    /// Refreshes every `every`, for the life of the process.
465    pub fn start_polling(&self, every: Duration) -> tokio::task::JoinHandle<()> {
466        let cache = self.clone();
467        tokio::spawn(async move {
468            loop {
469                tokio::time::sleep(every).await;
470                if let Err(e) = cache.refresh().await {
471                    tracing::warn!(error = %e, "cache generations refresh failed");
472                }
473            }
474        })
475    }
476
477    /// Takes the named lock for `ttl`, or `None` while another holder has
478    /// it. Release it with [`Lock::release`]; it lapses on its own after
479    /// `ttl`.
480    pub async fn lock(&self, name: &str, ttl: Duration) -> CoreResult<Option<Lock>> {
481        let owner = format!("{:032x}", rand::random::<u128>());
482        if self.inner.signals.lock(name, &owner, ttl).await? {
483            Ok(Some(Lock {
484                cache: self.clone(),
485                name: name.to_string(),
486                owner,
487            }))
488        } else {
489            Ok(None)
490        }
491    }
492}
493
494/// A held lock. Release it when the work is done; dropped unreleased, it
495/// lapses when its lifetime passes.
496#[must_use = "a lock does nothing unless held for the work's duration"]
497pub struct Lock {
498    cache: Cache,
499    name: String,
500    owner: String,
501}
502
503impl Lock {
504    pub fn name(&self) -> &str {
505        &self.name
506    }
507
508    /// The holder's token, unique to this acquisition.
509    pub fn owner(&self) -> &str {
510        &self.owner
511    }
512
513    /// Lets the lock go; `true` when it was still held.
514    pub async fn release(self) -> CoreResult<bool> {
515        self.cache
516            .inner
517            .signals
518            .unlock(&self.name, &self.owner)
519            .await
520    }
521}
522
523impl std::fmt::Debug for Lock {
524    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
525        f.debug_struct("Lock").field("name", &self.name).finish()
526    }
527}
528
529/// The cache a deployment configured: values in the store `cache.store`
530/// names (`auto` is `null` when `debug` and `memory` otherwise), generations
531/// and locks in the database, or in Redis for the `redis` store.
532pub async fn from_config(config: &CacheConfig, db: &Db, debug: bool) -> CoreResult<Cache> {
533    let store = match config.store.as_str() {
534        "auto" if debug => "null",
535        "auto" => "memory",
536        other => other,
537    };
538    match store {
539        "null" => Ok(Cache::new(NullStore, DbSignals::new(db.clone()))),
540        "memory" => Ok(Cache::new(
541            MemoryStore::with_capacity(config.memory_mib.saturating_mul(1024 * 1024)),
542            DbSignals::new(db.clone()),
543        )),
544        #[cfg(feature = "redis")]
545        "redis" => {
546            let url = config.url.as_deref().ok_or_else(|| {
547                CoreError::Config("cache.store = \"redis\" needs cache.url".to_string())
548            })?;
549            let (values, signals) = redis::connect(url, &config.prefix).await?;
550            Ok(Cache::new(values, signals))
551        }
552        #[cfg(not(feature = "redis"))]
553        "redis" => Err(CoreError::Config(
554            "cache.store = \"redis\" needs the `redis` feature of laterite-core".to_string(),
555        )),
556        other => Err(CoreError::Config(format!(
557            "cache.store `{other}` is not one of auto, null, memory, redis"
558        ))),
559    }
560}
561
562#[cfg(test)]
563mod tests {
564    use super::*;
565
566    #[test]
567    fn a_namespace_is_qualified_by_its_owner() {
568        let ns = CacheNamespace::new("feed");
569        assert_eq!(ns.qualified(ModuleId::new("acme.blog")), "acme.blog::feed");
570        assert_eq!(records("posts"), "records::posts");
571    }
572
573    #[tokio::test]
574    async fn an_unregistered_namespace_is_refused_and_records_are_not() {
575        let cache = Cache::local();
576        let err = cache
577            .put("k", vec![1], &["acme.blog::feed"], Duration::from_secs(1))
578            .await
579            .unwrap_err();
580        assert!(matches!(err, CoreError::Validation(_)));
581        cache
582            .put("k", vec![1], &[&records("posts")], Duration::from_secs(1))
583            .await
584            .unwrap();
585        cache.allow("acme.blog::feed");
586        cache
587            .put("k2", vec![1], &["acme.blog::feed"], Duration::from_secs(1))
588            .await
589            .unwrap();
590    }
591}
laterite_core/cache/mod.rs

laterite_core/cache/
mod.rs
