snix_store/pathinfoservice/combinators/
cache.rs

1use std::sync::Arc;
2
3use futures::stream::BoxStream;
4use nix_compat::nixbase32;
5use snix_castore::composition::{CompositionContext, ServiceBuilder};
6use tonic::async_trait;
7use tracing::{debug, instrument, warn};
8
9use crate::pathinfoservice::{self, PathInfo, PathInfoService};
10
11/// Asks near first, if not found, asks far.
12/// If found in there, returns it, and *inserts* it into
13/// near.
14/// There is no negative cache.
15/// Inserts and listings are not implemented for now.
16pub struct Cache<PS1, PS2> {
17    instance_name: String,
18    near: PS1,
19    far: PS2,
20}
21
22impl<PS1, PS2> Cache<PS1, PS2> {
23    pub fn new(instance_name: String, near: PS1, far: PS2) -> Self {
24        Self {
25            instance_name,
26            near,
27            far,
28        }
29    }
30}
31
32#[async_trait]
33impl<PS1, PS2> PathInfoService for Cache<PS1, PS2>
34where
35    PS1: PathInfoService,
36    PS2: PathInfoService,
37{
38    #[instrument(level = "trace", skip_all, err, fields(path_info.digest = nixbase32::encode(&digest), instance_name = %self.instance_name))]
39    async fn get(&self, digest: [u8; 20]) -> Result<Option<PathInfo>, pathinfoservice::Error> {
40        match self.near.get(digest).await.map_err(Error::NearGet)? {
41            Some(path_info) => {
42                debug!("serving from cache");
43                Ok(Some(path_info))
44            }
45            None => {
46                debug!("not found in near, asking remote…");
47                match self.far.get(digest).await.map_err(Error::FarGet)? {
48                    None => Ok(None),
49                    Some(path_info) => {
50                        debug!("found in remote, adding to cache");
51                        self.near
52                            .put(path_info.clone())
53                            .await
54                            .map_err(Error::NearPut)?;
55                        Ok(Some(path_info))
56                    }
57                }
58            }
59        }
60    }
61
62    #[instrument(level = "trace", skip_all, err, fields(path_info.digest = nixbase32::encode(&digest), instance_name = %self.instance_name))]
63    async fn has(&self, digest: [u8; 20]) -> Result<bool, pathinfoservice::Error> {
64        // FUTUREWORK: queue background tasks if ! self.near.has && self.far.has ? (configurable)
65        Ok(self.near.has(digest).await.map_err(Error::NearGet)?
66            || self.far.has(digest).await.map_err(Error::FarGet)?)
67    }
68
69    #[instrument(level = "trace", skip_all, err, fields(path_info.digest = nixbase32::encode(_path_info.store_path.digest()), instance_name = %self.instance_name))]
70    async fn put(&self, _path_info: PathInfo) -> Result<PathInfo, pathinfoservice::Error> {
71        Err(Error::Unsupported)?
72    }
73
74    fn list(&self) -> BoxStream<'static, Result<PathInfo, pathinfoservice::Error>> {
75        warn!("listing unsupported");
76        Box::pin(tokio_stream::once(Err(Error::Unsupported.into())))
77    }
78}
79
80#[derive(serde::Deserialize)]
81#[serde(deny_unknown_fields)]
82pub struct CacheConfig {
83    pub near: String,
84    pub far: String,
85}
86
87#[derive(thiserror::Error, Debug)]
88pub enum Error {
89    #[error("instantiating from a url is not supported")]
90    URLNotSupported,
91
92    #[error("getting from near: {0}")]
93    NearGet(#[source] pathinfoservice::Error),
94    #[error("putting into near: {0}")]
95    NearPut(#[source] pathinfoservice::Error),
96    #[error("getting from far: {0}")]
97    FarGet(#[source] pathinfoservice::Error),
98
99    #[error("unsupported")]
100    Unsupported,
101}
102
103impl TryFrom<url::Url> for CacheConfig {
104    type Error = Box<dyn std::error::Error + Send + Sync>;
105    fn try_from(_url: url::Url) -> Result<Self, Self::Error> {
106        Err(Error::URLNotSupported)?
107    }
108}
109
110#[async_trait]
111impl ServiceBuilder for CacheConfig {
112    type Output = dyn PathInfoService;
113    async fn build<'a>(
114        &'a self,
115        instance_name: &str,
116        context: &CompositionContext,
117    ) -> Result<Arc<Self::Output>, Box<dyn std::error::Error + Send + Sync>> {
118        let (near, far) = futures::join!(
119            context.resolve::<Self::Output>(&self.near),
120            context.resolve::<Self::Output>(&self.far)
121        );
122        Ok(Arc::new(Cache {
123            instance_name: instance_name.to_string(),
124            near: near?,
125            far: far?,
126        }))
127    }
128}
129
130#[cfg(test)]
131mod test {
132    use std::num::NonZeroUsize;
133
134    use crate::{
135        fixtures::PATH_INFO,
136        pathinfoservice::{LruPathInfoService, PathInfoService},
137        utils::gen_test_pathinfo_service,
138    };
139
140    /// Helper function setting up an instance of a Cache PathInfoService.
141    async fn create_pathinfoservice() -> super::Cache<LruPathInfoService, impl PathInfoService> {
142        // Create an instance of a "far" PathInfoService.
143        let far = gen_test_pathinfo_service();
144
145        // … and an instance of a "near" PathInfoService.
146        let near = LruPathInfoService::with_capacity("near".into(), NonZeroUsize::new(1).unwrap());
147
148        // create a Pathinfoservice combining the two and return it.
149        super::Cache::new("root".into(), near, far)
150    }
151
152    /// Getting from the far backend is gonna insert it into the near one.
153    #[tokio::test]
154    async fn test_populate_cache() {
155        let svc = create_pathinfoservice().await;
156
157        // query the PathInfo, things should not be there.
158        assert!(
159            svc.get(*PATH_INFO.store_path.digest())
160                .await
161                .unwrap()
162                .is_none()
163        );
164
165        // insert it into the far one.
166        svc.far.put(PATH_INFO.clone()).await.unwrap();
167
168        // now try getting it again, it should succeed.
169        assert_eq!(
170            Some(PATH_INFO.clone()),
171            svc.get(*PATH_INFO.store_path.digest()).await.unwrap()
172        );
173
174        // peek near, it should now be there.
175        assert_eq!(
176            Some(PATH_INFO.clone()),
177            svc.near.get(*PATH_INFO.store_path.digest()).await.unwrap()
178        );
179    }
180}
snix_store/pathinfoservice/combinators/cache.rs

snix_store/pathinfoservice/combinators/
cache.rs
