snix_castore/directoryservice/
memory.rs

1use crate::{B3Digest, Error};
2use futures::stream::BoxStream;
3use std::collections::HashMap;
4use std::sync::Arc;
5use tokio::sync::RwLock;
6use tonic::async_trait;
7use tracing::{instrument, warn};
8
9use super::utils::traverse_directory;
10use super::{Directory, DirectoryPutter, DirectoryService, SimplePutter};
11use crate::composition::{CompositionContext, ServiceBuilder};
12use crate::proto;
13
14#[derive(Clone, Default)]
15pub struct MemoryDirectoryService {
16    instance_name: String,
17    db: Arc<RwLock<HashMap<B3Digest, proto::Directory>>>,
18}
19
20#[async_trait]
21impl DirectoryService for MemoryDirectoryService {
22    #[instrument(skip(self, digest), err, fields(directory.digest = %digest, instance_name=%self.instance_name))]
23    async fn get(&self, digest: &B3Digest) -> Result<Option<Directory>, Error> {
24        let db = self.db.read().await;
25
26        match db.get(digest) {
27            // The directory was not found, return
28            None => Ok(None),
29
30            // The directory was found, try to parse the data as Directory message
31            Some(directory) => {
32                // Validate the retrieved Directory indeed has the
33                // digest we expect it to have, to detect corruptions.
34                let actual_digest = directory.digest();
35                if actual_digest != *digest {
36                    return Err(Error::StorageError(format!(
37                        "requested directory with digest {digest}, but got {actual_digest}"
38                    )));
39                }
40
41                Ok(Some(directory.clone().try_into().map_err(|e| {
42                    crate::Error::StorageError(format!("corrupted directory: {e}"))
43                })?))
44            }
45        }
46    }
47
48    #[instrument(skip(self, directory), err, fields(directory.digest = %directory.digest(), instance_name=%self.instance_name))]
49    async fn put(&self, directory: Directory) -> Result<B3Digest, Error> {
50        let digest = directory.digest();
51
52        // store it
53        let mut db = self.db.write().await;
54        db.insert(digest.clone(), directory.into());
55
56        Ok(digest)
57    }
58
59    #[instrument(skip_all, fields(directory.digest = %root_directory_digest, instance_name=%self.instance_name))]
60    fn get_recursive(
61        &self,
62        root_directory_digest: &B3Digest,
63    ) -> BoxStream<'static, Result<Directory, Error>> {
64        traverse_directory(self.clone(), root_directory_digest)
65    }
66
67    #[instrument(skip_all, fields(instance_name=%self.instance_name))]
68    fn put_multiple_start(&self) -> Box<(dyn DirectoryPutter + '_)> {
69        Box::new(SimplePutter::new(self))
70    }
71}
72
73#[derive(serde::Deserialize, Debug)]
74#[serde(deny_unknown_fields)]
75pub struct MemoryDirectoryServiceConfig {}
76
77impl TryFrom<url::Url> for MemoryDirectoryServiceConfig {
78    type Error = Box<dyn std::error::Error + Send + Sync>;
79    fn try_from(url: url::Url) -> Result<Self, Self::Error> {
80        // memory doesn't support host or path in the URL.
81        if url.has_host() || !url.path().is_empty() {
82            return Err(Error::StorageError("invalid url".to_string()).into());
83        }
84        Ok(MemoryDirectoryServiceConfig {})
85    }
86}
87
88#[async_trait]
89impl ServiceBuilder for MemoryDirectoryServiceConfig {
90    type Output = dyn DirectoryService;
91    async fn build<'a>(
92        &'a self,
93        instance_name: &str,
94        _context: &CompositionContext,
95    ) -> Result<Arc<dyn DirectoryService>, Box<dyn std::error::Error + Send + Sync + 'static>> {
96        Ok(Arc::new(MemoryDirectoryService {
97            instance_name: instance_name.to_string(),
98            db: Default::default(),
99        }))
100    }
101}
snix_castore/directoryservice/memory.rs

snix_castore/directoryservice/
memory.rs
