snix_castore/chunkstore/combinators/
cache.rs

1use std::sync::Arc;
2
3use tonic::async_trait;
4use tracing::{instrument, trace};
5
6use crate::{
7    B3Digest,
8    chunkstore::{self, Chunk, ChunkStore},
9    composition::{CompositionContext, ServiceBuilder},
10};
11
12/// Asks near first, if not found, asks far.
13/// If found in there, returns it, and *inserts* it into
14/// near.
15/// There is no negative cache.
16pub struct Cache<CN, CF> {
17    instance_name: String,
18    near: CN,
19    far: CF,
20}
21
22impl<CN, CF> Cache<CN, CF> {
23    /// Constructs a new [Cache] from the passed `instance_name`, `near` and `far` [ChunkStore]s.
24    pub fn new(instance_name: String, near: CN, far: CF) -> Self {
25        Self {
26            instance_name,
27            near,
28            far,
29        }
30    }
31}
32
33#[async_trait]
34impl<CN, CF> ChunkStore for Cache<CN, CF>
35where
36    CN: ChunkStore,
37    CF: ChunkStore,
38{
39    #[instrument(skip(self, digest), fields(chunk.digest=%digest, instance_name=%self.instance_name))]
40    async fn get(&self, digest: &B3Digest) -> Result<Option<Chunk>, chunkstore::Error> {
41        if let Some(chunk) = self.near.get(digest).await.map_err(Error::NearGet)? {
42            trace!("serving from cache");
43            return Ok(Some(chunk));
44        }
45
46        trace!("not found in near, asking remote…");
47
48        match self.far.get(digest).await.map_err(Error::FarGet)? {
49            // chunk doesn't exist on the far side either, nothing we can do.
50            None => Ok(None),
51            Some(chunk) => {
52                // chunk is present on the far side, insert it into near before returning.
53                let digest_near = self.near.put(chunk.clone()).await.map_err(Error::NearPut)?;
54                if digest_near != *digest {
55                    Err(Error::InsertingMismatch {
56                        expected: *digest,
57                        actual: digest_near,
58                    })?;
59                }
60
61                Ok(Some(chunk))
62            }
63        }
64    }
65
66    #[instrument(skip_all, fields(chunk.digest=%digest, instance_name=%self.instance_name))]
67    async fn has(&self, digest: &B3Digest) -> Result<bool, chunkstore::Error> {
68        Ok(self.near.has(digest).await.map_err(Error::NearGet)?
69            || self.far.has(digest).await.map_err(Error::FarGet)?)
70    }
71
72    #[instrument(skip_all, fields(instance_name=%self.instance_name))]
73    async fn put(&self, chunk: Chunk) -> Result<B3Digest, chunkstore::Error> {
74        Ok(self.near.put(chunk).await.map_err(Error::NearPut)?)
75    }
76}
77
78/// Error returned from [Cache].
79#[derive(thiserror::Error, Debug)]
80pub enum Error {
81    #[error("wrong arguments: {0}")]
82    WrongConfig(&'static str),
83
84    #[error("getting from near: {0}")]
85    NearGet(#[source] chunkstore::Error),
86    #[error("putting into near: {0}")]
87    NearPut(#[source] chunkstore::Error),
88    #[error("getting from far: {0}")]
89    FarGet(#[source] chunkstore::Error),
90    #[error(
91        "inserting chunk with digest {expected} into near returned different digest ({actual})"
92    )]
93    InsertingMismatch {
94        expected: B3Digest,
95        actual: B3Digest,
96    },
97}
98
99impl From<Error> for chunkstore::Error {
100    fn from(value: Error) -> Self {
101        Self(Box::new(value))
102    }
103}
104
105/// Configuration for a [Cache] [ChunkStore].
106#[derive(serde::Deserialize, Debug)]
107#[serde(deny_unknown_fields)]
108pub struct CacheConfig {
109    near: String,
110    far: String,
111}
112
113impl TryFrom<url::Url> for CacheConfig {
114    type Error = Box<dyn std::error::Error + Send + Sync>;
115    fn try_from(url: url::Url) -> Result<Self, Self::Error> {
116        // cache doesn't support host or path in the URL.
117        if url.has_authority() || !url.path().is_empty() {
118            return Err(Error::WrongConfig("no authority or path allowed").into());
119        }
120        Ok(serde_qs::from_str(url.query().unwrap_or_default())?)
121    }
122}
123
124#[async_trait]
125impl ServiceBuilder for CacheConfig {
126    type Output = dyn ChunkStore;
127    async fn build<'a>(
128        &'a self,
129        instance_name: &str,
130        context: &CompositionContext,
131    ) -> Result<Arc<Self::Output>, Box<dyn std::error::Error + Send + Sync>> {
132        let (near, far) = futures::join!(
133            context.resolve::<Self::Output>(&self.near),
134            context.resolve::<Self::Output>(&self.far)
135        );
136        Ok(Arc::new(Cache {
137            instance_name: instance_name.to_string(),
138            near: near?,
139            far: far?,
140        }))
141    }
142}
143
144#[cfg(test)]
145mod test {
146    use crate::chunkstore::{Chunk, ChunkStore, MockChunkStore, memory::MemoryChunkStore};
147
148    use super::Cache;
149
150    const CHUNK: Chunk = Chunk::from_static(b"foo");
151
152    /// A chunk present in near is served from there, far is not queried.
153    #[tokio::test]
154    async fn get_near_only() {
155        let near = MemoryChunkStore::default();
156        near.put(CHUNK.clone()).await.expect("to succeed");
157        let mut far = MockChunkStore::new();
158        far.expect_get().never();
159
160        let uut = Cache::new("uut".to_string(), near, far);
161
162        assert_eq!(
163            Some(CHUNK.clone()),
164            uut.get(&CHUNK.digest()).await.expect("to succeed")
165        );
166    }
167
168    /// A chunk only present in far is returned, and inserted into near.
169    #[tokio::test]
170    async fn get_populates_near() {
171        let near = MemoryChunkStore::default();
172        let far_with_chunk = MemoryChunkStore::default();
173        far_with_chunk.put(CHUNK.clone()).await.expect("to succeed");
174
175        let uut = Cache::new("uut".to_string(), &near, far_with_chunk);
176
177        // we know near is empty.
178        assert!(
179            !near.has(&CHUNK.digest()).await.expect("to succeed"),
180            "near should be empty"
181        );
182
183        // query uut, which will populate near
184        uut.get(&CHUNK.digest()).await.expect("to succeed");
185
186        // now near should have the chunk.
187        assert!(
188            near.has(&CHUNK.digest()).await.expect("to succeed"),
189            "near should be populated now"
190        );
191    }
192
193    /// If neither has it, None is returned.
194    #[tokio::test]
195    async fn get_not_found() {
196        let near = MemoryChunkStore::default();
197        let far = MemoryChunkStore::default();
198
199        let uut = Cache::new("uut".to_string(), near, far);
200
201        assert_eq!(None, uut.get(&CHUNK.digest()).await.expect("to succeed"));
202    }
203
204    /// has() returns true if either near or far has the chunk.
205    #[tokio::test]
206    async fn has_near_or_far() {
207        let cs_without = MemoryChunkStore::default();
208        let cs_with = MemoryChunkStore::default();
209        cs_with.put(CHUNK.clone()).await.expect("to succeed");
210
211        let uut = Cache::new("uut".to_string(), &cs_without, &cs_with);
212        assert!(uut.has(&CHUNK.digest()).await.expect("to succeed"));
213        let uut = Cache::new("uut".to_string(), &cs_with, &cs_without);
214        assert!(uut.has(&CHUNK.digest()).await.expect("to succeed"));
215    }
216
217    /// put() writes to near only.
218    #[tokio::test]
219    async fn put_writes_to_near() {
220        let near = MemoryChunkStore::default();
221        let far = MemoryChunkStore::default();
222
223        let uut = Cache::new("uut".to_string(), near.clone(), far.clone());
224
225        assert_eq!(
226            CHUNK.digest(),
227            uut.put(CHUNK.clone()).await.expect("to succeed")
228        );
229
230        assert!(near.has(&CHUNK.digest()).await.expect("to succeed"));
231        assert!(!far.has(&CHUNK.digest()).await.expect("to succeed"));
232    }
233}
snix_castore/chunkstore/combinators/cache.rs

snix_castore/chunkstore/combinators/
cache.rs
