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

snix_store/pathinfoservice/combinators/
race.rs
