use crate::directoryservice::DirectoryPutter; use crate::proto::Directory; use crate::{proto, B3Digest, Error}; use futures::Stream; use prost::Message; use std::path::PathBuf; use std::pin::Pin; use tonic::async_trait; use tracing::{instrument, warn}; use super::utils::{traverse_directory, SimplePutter}; use super::DirectoryService; #[derive(Clone)] pub struct SledDirectoryService { db: sled::Db, } impl SledDirectoryService { pub fn new(p: PathBuf) -> Result { let config = sled::Config::default().use_compression(true).path(p); let db = config.open()?; Ok(Self { db }) } pub fn new_temporary() -> Result { let config = sled::Config::default().temporary(true); let db = config.open()?; Ok(Self { db }) } } #[async_trait] impl DirectoryService for SledDirectoryService { /// Constructs a [SledDirectoryService] from the passed [url::Url]: /// - scheme has to be `sled://` /// - there may not be a host. /// - a path to the sled needs to be provided (which may not be `/`). fn from_url(url: &url::Url) -> Result { if url.scheme() != "sled" { return Err(crate::Error::StorageError("invalid scheme".to_string())); } if url.has_host() { return Err(crate::Error::StorageError(format!( "invalid host: {}", url.host().unwrap() ))); } // TODO: expose compression and other parameters as URL parameters, drop new and new_temporary? if url.path().is_empty() { Self::new_temporary().map_err(|e| Error::StorageError(e.to_string())) } else if url.path() == "/" { Err(crate::Error::StorageError( "cowardly refusing to open / with sled".to_string(), )) } else { Self::new(url.path().into()).map_err(|e| Error::StorageError(e.to_string())) } } #[instrument(skip(self, digest), fields(directory.digest = %digest))] async fn get(&self, digest: &B3Digest) -> Result, Error> { match self.db.get(digest.to_vec()) { // The directory was not found, return Ok(None) => Ok(None), // The directory was found, try to parse the data as Directory message Ok(Some(data)) => match Directory::decode(&*data) { Ok(directory) => { // Validate the retrieved Directory indeed has the // digest we expect it to have, to detect corruptions. let actual_digest = directory.digest(); if actual_digest != *digest { return Err(Error::StorageError(format!( "requested directory with digest {}, but got {}", digest, actual_digest ))); } // Validate the Directory itself is valid. if let Err(e) = directory.validate() { warn!("directory failed validation: {}", e.to_string()); return Err(Error::StorageError(format!( "directory {} failed validation: {}", actual_digest, e, ))); } Ok(Some(directory)) } Err(e) => { warn!("unable to parse directory {}: {}", digest, e); Err(Error::StorageError(e.to_string())) } }, // some storage error? Err(e) => Err(Error::StorageError(e.to_string())), } } #[instrument(skip(self, directory), fields(directory.digest = %directory.digest()))] async fn put(&self, directory: proto::Directory) -> Result { let digest = directory.digest(); // validate the directory itself. if let Err(e) = directory.validate() { return Err(Error::InvalidRequest(format!( "directory {} failed validation: {}", digest, e, ))); } // store it let result = self.db.insert(digest.to_vec(), directory.encode_to_vec()); if let Err(e) = result { return Err(Error::StorageError(e.to_string())); } Ok(digest) } #[instrument(skip_all, fields(directory.digest = %root_directory_digest))] fn get_recursive( &self, root_directory_digest: &B3Digest, ) -> Pin> + Send + 'static)>> { traverse_directory(self.clone(), root_directory_digest) } #[instrument(skip_all)] fn put_multiple_start(&self) -> Box<(dyn DirectoryPutter + 'static)> where Self: Clone, { Box::new(SimplePutter::new(self.clone())) } } #[cfg(test)] mod tests { use tempfile::TempDir; use super::DirectoryService; use super::SledDirectoryService; /// This uses a wrong scheme. #[test] fn test_invalid_scheme() { let url = url::Url::parse("http://foo.example/test").expect("must parse"); assert!(SledDirectoryService::from_url(&url).is_err()); } /// This uses the correct scheme, and doesn't specify a path (temporary sled). #[test] fn test_valid_scheme_temporary() { let url = url::Url::parse("sled://").expect("must parse"); assert!(SledDirectoryService::from_url(&url).is_ok()); } /// This sets the path to a location that doesn't exist, which should fail (as sled doesn't mkdir -p) #[test] fn test_nonexistent_path() { let tmpdir = TempDir::new().unwrap(); let mut url = url::Url::parse("sled://foo.example").expect("must parse"); url.set_path(tmpdir.path().join("foo").join("bar").to_str().unwrap()); assert!(SledDirectoryService::from_url(&url).is_err()); } /// This uses the correct scheme, and specifies / as path (which should fail // for obvious reasons) #[test] fn test_invalid_path_root() { let url = url::Url::parse("sled:///").expect("must parse"); assert!(SledDirectoryService::from_url(&url).is_err()); } /// This uses the correct scheme, and sets a tempdir as location. #[test] fn test_valid_scheme_path() { let tmpdir = TempDir::new().unwrap(); let mut url = url::Url::parse("sled://").expect("must parse"); url.set_path(tmpdir.path().to_str().unwrap()); assert!(SledDirectoryService::from_url(&url).is_ok()); } /// This sets a host, rather than a path, which should fail. #[test] fn test_invalid_host() { let url = url::Url::parse("sled://foo.example").expect("must parse"); assert!(SledDirectoryService::from_url(&url).is_err()); } /// This sets a host AND a valid path, which should fail #[test] fn test_invalid_host_and_path() { let tmpdir = TempDir::new().unwrap(); let mut url = url::Url::parse("sled://foo.example").expect("must parse"); url.set_path(tmpdir.path().to_str().unwrap()); assert!(SledDirectoryService::from_url(&url).is_err()); } }