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 {}, but got {}",
38                        digest, actual_digest
39                    )));
40                }
41
42                Ok(Some(directory.clone().try_into().map_err(|e| {
43                    crate::Error::StorageError(format!("corrupted directory: {}", e))
44                })?))
45            }
46        }
47    }
48
49    #[instrument(skip(self, directory), err, fields(directory.digest = %directory.digest(), instance_name=%self.instance_name))]
50    async fn put(&self, directory: Directory) -> Result<B3Digest, Error> {
51        let digest = directory.digest();
52
53        // store it
54        let mut db = self.db.write().await;
55        db.insert(digest.clone(), directory.into());
56
57        Ok(digest)
58    }
59
60    #[instrument(skip_all, fields(directory.digest = %root_directory_digest, instance_name=%self.instance_name))]
61    fn get_recursive(
62        &self,
63        root_directory_digest: &B3Digest,
64    ) -> BoxStream<'static, Result<Directory, Error>> {
65        traverse_directory(self.clone(), root_directory_digest)
66    }
67
68    #[instrument(skip_all, fields(instance_name=%self.instance_name))]
69    fn put_multiple_start(&self) -> Box<(dyn DirectoryPutter + '_)> {
70        Box::new(SimplePutter::new(self))
71    }
72}
73
74#[derive(serde::Deserialize, Debug)]
75#[serde(deny_unknown_fields)]
76pub struct MemoryDirectoryServiceConfig {}
77
78impl TryFrom<url::Url> for MemoryDirectoryServiceConfig {
79    type Error = Box<dyn std::error::Error + Send + Sync>;
80    fn try_from(url: url::Url) -> Result<Self, Self::Error> {
81        // memory doesn't support host or path in the URL.
82        if url.has_host() || !url.path().is_empty() {
83            return Err(Error::StorageError("invalid url".to_string()).into());
84        }
85        Ok(MemoryDirectoryServiceConfig {})
86    }
87}
88
89#[async_trait]
90impl ServiceBuilder for MemoryDirectoryServiceConfig {
91    type Output = dyn DirectoryService;
92    async fn build<'a>(
93        &'a self,
94        instance_name: &str,
95        _context: &CompositionContext,
96    ) -> Result<Arc<dyn DirectoryService>, Box<dyn std::error::Error + Send + Sync + 'static>> {
97        Ok(Arc::new(MemoryDirectoryService {
98            instance_name: instance_name.to_string(),
99            db: Default::default(),
100        }))
101    }
102}
snix_castore/directoryservice/memory.rs

snix_castore/directoryservice/
memory.rs
