snix_store/pathinfoservice/combinators/
race.rs

1use std::sync::Arc;
2
3use futures::{StreamExt, stream::BoxStream};
4use snix_castore::{
5    combinators,
6    composition::{CompositionContext, ServiceBuilder},
7};
8use tonic::async_trait;
9
10use crate::{
11    nar::NarCalculationService,
12    pathinfoservice::{self, PathInfo, PathInfoService},
13};
14
15/// Fans out requests to multiple stores in parallel, returning the first positive or erroneous answer.
16///
17/// A negative response is returned if all backends return them.
18/// Write requests are not implemented.
19pub struct Race<PS> {
20    #[allow(unused)]
21    instance_name: String,
22    services: Vec<PS>,
23}
24
25impl<PS> Race<PS> {
26    /// Construct from an iterator of services.
27    pub fn new<I: IntoIterator<Item = PS>>(instance_name: String, iter: I) -> Race<PS> {
28        Self {
29            instance_name,
30            services: Vec::from_iter(iter),
31        }
32    }
33
34    /// Add another sevice to the list.
35    pub fn add(&mut self, svc: PS) {
36        self.services.push(svc);
37    }
38}
39
40#[async_trait]
41impl<PS> PathInfoService for Race<PS>
42where
43    PS: PathInfoService,
44{
45    async fn get(&self, digest: [u8; 20]) -> Result<Option<PathInfo>, pathinfoservice::Error> {
46        Ok(combinators::race::race_unary(&self.services, |svc| async {
47            // Skip over `Ok(None)` by returning None,
48            // but keep the Option<PathInfo> in the returned Ok() value.
49            Some(svc.get(digest).await.transpose()?.map(Some))
50        })
51        .await
52        .map_err(Error::Racing)?)
53    }
54
55    async fn has(&self, digest: [u8; 20]) -> Result<bool, pathinfoservice::Error> {
56        Ok(combinators::race::race_unary(&self.services, |svc| async {
57            // Skip over Ok(false) by returning None.
58            match svc.has(digest).await {
59                Ok(false) => None,
60                resp => Some(resp),
61            }
62        })
63        .await
64        .map_err(Error::Racing)?)
65    }
66
67    async fn put(&self, _path_info: PathInfo) -> Result<PathInfo, pathinfoservice::Error> {
68        return Err(Error::Unimplemented.into());
69    }
70
71    fn list(&self) -> BoxStream<'static, Result<PathInfo, pathinfoservice::Error>> {
72        futures::stream::once(async { Err(Box::new(Error::Unimplemented))? }).boxed()
73    }
74
75    fn nar_calculation_service(&self) -> Option<Arc<dyn NarCalculationService>> {
76        // We can't possibly know which one has all contents to calculate.
77        None
78    }
79}
80
81#[derive(thiserror::Error, Debug)]
82pub enum Error {
83    #[error("wrong arguments: {0}")]
84    WrongConfig(&'static str),
85
86    #[error("error from racing")]
87    Racing(combinators::race::Error<pathinfoservice::Error>),
88
89    #[error("unimplemented")]
90    Unimplemented,
91}
92
93#[derive(serde::Deserialize, Debug)]
94#[serde(deny_unknown_fields)]
95pub struct RaceConfig {
96    services: Vec<String>,
97}
98
99impl TryFrom<url::Url> for RaceConfig {
100    type Error = Box<dyn std::error::Error + Send + Sync>;
101    fn try_from(url: url::Url) -> Result<Self, Self::Error> {
102        if url.has_authority() || !url.path().is_empty() {
103            return Err(Error::WrongConfig("no authority or path allowed").into());
104        }
105        Ok(serde_qs::from_str(url.query().unwrap_or_default())?)
106    }
107}
108
109#[async_trait]
110impl ServiceBuilder for RaceConfig {
111    type Output = dyn PathInfoService;
112    async fn build<'a>(
113        &'a self,
114        instance_name: &str,
115        context: &CompositionContext,
116    ) -> Result<Arc<Self::Output>, Box<dyn std::error::Error + Send + Sync>> {
117        let services =
118            futures::future::try_join_all(self.services.iter().map(|instance_ref| async move {
119                context.resolve::<Self::Output>(instance_ref).await
120            }))
121            .await?;
122
123        Ok(Arc::new(Race::new(instance_name.to_string(), services)))
124    }
125}
126
127#[cfg(test)]
128mod test {
129    use std::sync::LazyLock;
130
131    use mockall::predicate;
132    use pretty_assertions::assert_matches;
133    use snix_castore::combinators;
134
135    use crate::{
136        fixtures::PATH_INFO,
137        pathinfoservice::{MockPathInfoService, PathInfoService},
138    };
139
140    use super::{Error, Race};
141
142    static PATH_INFO_DIGEST: LazyLock<[u8; 20]> = LazyLock::new(|| *PATH_INFO.store_path.digest());
143
144    /// backends are tried exhaustively if all report None.
145    #[tokio::test]
146    async fn get_tries_exhaustively_on_none() {
147        let first = {
148            let mut svc = MockPathInfoService::new();
149            svc.expect_get()
150                .with(predicate::eq(*PATH_INFO_DIGEST))
151                .once()
152                .returning(|_| Ok(None));
153            svc
154        };
155        let second = {
156            let mut svc = MockPathInfoService::new();
157            svc.expect_get()
158                .with(predicate::eq(*PATH_INFO_DIGEST))
159                .once()
160                .returning(|_| Ok(None));
161            svc
162        };
163
164        let uut = Race::new("uut".to_string(), [first, second]);
165
166        assert!(
167            uut.get(*PATH_INFO_DIGEST)
168                .await
169                .expect("to succeed")
170                .is_none()
171        )
172    }
173
174    /// backends are tried exhaustively if all report None.
175    #[tokio::test]
176    async fn has_tries_exhaustively_on_none() {
177        let first = {
178            let mut svc = MockPathInfoService::new();
179            svc.expect_has()
180                .with(predicate::eq(*PATH_INFO_DIGEST))
181                .once()
182                .returning(|_| Ok(false));
183            svc
184        };
185        let second = {
186            let mut svc = MockPathInfoService::new();
187            svc.expect_has()
188                .with(predicate::eq(*PATH_INFO_DIGEST))
189                .once()
190                .returning(|_| Ok(false));
191            svc
192        };
193
194        let uut = Race::new("uut".to_string(), [first, second]);
195
196        assert!(!uut.has(*PATH_INFO_DIGEST).await.expect("to succeed"),);
197    }
198
199    // if one has it and one does not, we return the positive result.
200    #[tokio::test]
201    async fn get_returns_positive() {
202        let first = {
203            let mut svc = MockPathInfoService::new();
204            svc.expect_get()
205                .with(predicate::eq(*PATH_INFO_DIGEST))
206                .once()
207                .returning(|_| Ok(Some(PATH_INFO.clone())));
208            svc
209        };
210
211        let second = {
212            let mut svc = MockPathInfoService::new();
213            svc.expect_get()
214                .with(predicate::eq(*PATH_INFO_DIGEST))
215                // We cannot be certain this is called at all, so no `once()` here.
216                .returning(|_| Ok(None));
217            svc
218        };
219
220        let uut = Race::new("uut".to_string(), [first, second]);
221
222        assert_eq!(
223            Some(PATH_INFO.clone()),
224            uut.get(*PATH_INFO_DIGEST).await.expect("to succeed")
225        )
226    }
227
228    /// Errors are bubbled up, and the error contains the correct service index.
229    #[tokio::test]
230    async fn get_return_error() {
231        let first = {
232            let mut svc = MockPathInfoService::new();
233            svc.expect_get()
234                .with(predicate::eq(*PATH_INFO_DIGEST))
235                .once()
236                .returning(|_| Err("".into()));
237            svc
238        };
239
240        // Ideally this one would be just slower than `first`.
241        let second = {
242            let mut svc = MockPathInfoService::new();
243            svc.expect_get()
244                .with(predicate::eq(*PATH_INFO_DIGEST))
245                // We cannot be certain this is called at all, so no `once()` here.
246                .returning(|_| Ok(None));
247            svc
248        };
249
250        let uut = Race::new("uut".to_string(), [first, second]);
251
252        let err = uut.get(*PATH_INFO_DIGEST).await.expect_err("to fail");
253        let err = err.downcast_ref::<Error>().unwrap();
254        assert_matches!(err, Error::Racing(combinators::race::Error::Backend(0, _)))
255    }
256
257    // FUTUREWORK: ideally we'd be constructing mocks that take longer than others / never return,
258    // but that's not supported in automock: https://github.com/asomers/mockall/issues/189
259    // So it's a bit tough to create test cases reliably.
260}
snix_store/pathinfoservice/combinators/race.rs

snix_store/pathinfoservice/combinators/
race.rs
