snix_store/pathinfoservice/combinators/
priority.rs

1use std::sync::Arc;
2
3use futures::{StreamExt, stream::BoxStream};
4use nix_compat::nixbase32;
5use snix_castore::composition::{CompositionContext, ServiceBuilder};
6use tonic::async_trait;
7use tracing::instrument;
8
9use crate::{
10    nar::NarCalculationService,
11    path_info::PathInfo,
12    pathinfoservice::{self, PathInfoService},
13};
14
15/// Holds references to many different PathInfoService.
16/// Read requests try services sequentially.
17/// Any error in a service bubbles up.
18/// Write requests are not implemented.
19pub struct Priority<PS> {
20    instance_name: String,
21    // NOTE: Arc<dyn PS> implements PS too, so you can put different service types in here.
22    services: Vec<PS>,
23}
24
25impl<DS> Priority<DS> {
26    /// Construct from an iterator of services.
27    pub fn new<I: IntoIterator<Item = DS>>(instance_name: String, iter: I) -> Priority<DS> {
28        Self {
29            instance_name,
30            services: Vec::from_iter(iter),
31        }
32    }
33}
34
35#[async_trait]
36impl<PS> PathInfoService for Priority<PS>
37where
38    PS: PathInfoService,
39{
40    #[instrument(skip_all, err, fields(path_info.digest = nixbase32::encode(&digest), instance_name = %self.instance_name))]
41    async fn get(&self, digest: [u8; 20]) -> Result<Option<PathInfo>, pathinfoservice::Error> {
42        // traverse the list of services. If any service has it, return from there.
43        // Errors cause the combinator to bail out early.
44        for (idx, service) in self.services.iter().enumerate() {
45            if let Some(directory) = service
46                .get(digest)
47                .await
48                .map_err(|err| Error::Backend(idx, err))?
49            {
50                return Ok(Some(directory));
51            }
52        }
53
54        Ok(None)
55    }
56
57    #[instrument(skip_all, err, fields(path_info.digest = nixbase32::encode(&digest), instance_name = %self.instance_name))]
58    async fn has(&self, digest: [u8; 20]) -> Result<bool, pathinfoservice::Error> {
59        // traverse the list of services. If any service has it, return true.
60        // Errors cause the combinator to bail out early.
61        for (idx, service) in self.services.iter().enumerate() {
62            if service
63                .has(digest)
64                .await
65                .map_err(|err| Error::Backend(idx, err))?
66            {
67                return Ok(true);
68            }
69        }
70
71        Ok(false)
72    }
73
74    async fn put(&self, _path_info: PathInfo) -> Result<PathInfo, pathinfoservice::Error> {
75        return Err(Error::Unimplemented.into());
76    }
77
78    fn list(&self) -> BoxStream<'static, Result<PathInfo, pathinfoservice::Error>> {
79        futures::stream::once(async { Err(Box::new(Error::Unimplemented))? }).boxed()
80    }
81
82    fn nar_calculation_service(&self) -> Option<Arc<dyn NarCalculationService>> {
83        // We can't possibly know which one has all contents to calculate.
84        None
85    }
86}
87
88#[derive(thiserror::Error, Debug)]
89pub enum Error {
90    #[error("wrong arguments: {0}")]
91    WrongConfig(&'static str),
92
93    #[error("error from service with index {0}")]
94    Backend(usize, #[source] pathinfoservice::Error),
95
96    #[error("unimplemented")]
97    Unimplemented,
98}
99
100#[derive(serde::Deserialize, Debug)]
101#[serde(deny_unknown_fields)]
102pub struct PriorityConfig {
103    services: Vec<String>,
104}
105
106impl TryFrom<url::Url> for PriorityConfig {
107    type Error = Box<dyn std::error::Error + Send + Sync>;
108    fn try_from(url: url::Url) -> Result<Self, Self::Error> {
109        if url.has_authority() || !url.path().is_empty() {
110            return Err(Error::WrongConfig("no authority or path allowed").into());
111        }
112        Ok(serde_qs::from_str(url.query().unwrap_or_default())?)
113    }
114}
115
116#[async_trait]
117impl ServiceBuilder for PriorityConfig {
118    type Output = dyn PathInfoService;
119    async fn build<'a>(
120        &'a self,
121        instance_name: &str,
122        context: &CompositionContext,
123    ) -> Result<Arc<Self::Output>, Box<dyn std::error::Error + Send + Sync>> {
124        let services =
125            futures::future::try_join_all(self.services.iter().map(|instance_ref| async move {
126                context.resolve::<Self::Output>(instance_ref).await
127            }))
128            .await?;
129
130        Ok(Arc::new(Priority::new(instance_name.to_string(), services)))
131    }
132}
snix_store/pathinfoservice/combinators/priority.rs

snix_store/pathinfoservice/combinators/
priority.rs
