snix_castore/blobstore/combinators/
cache.rs

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

snix_castore/blobstore/combinators/
cache.rs
