mdebray/tvl-depot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

chore(tvix/[ca]store): Drop sled support completely in favor of redb

Browse source 
Over the past couple of months we've been using redb instead of sled as
the default filesystem-based database in PS and DS. I am confident that
we can get rid of sled completely now, and just keep redb.

Change-Id: I11fa1e4453e280253855f8eade990b37eb6965ae
Reviewed-on: https://cl.tvl.fyi/c/depot/+/12567
Reviewed-by: yuka <yuka@yuka.dev>
Reviewed-by: flokli <flokli@flokli.de>
Tested-by: BuildkiteCI
Autosubmit: Ilan Joselevich <personal@ilanjoselevich.com>
This commit is contained in:
Ilan Joselevich 2024-10-03 19:50:30 +02:00
parent 56b306f884
commit ab6e8d28aa
14 changed files with 18 additions and 812 deletions
Download patch file Download diff file Expand all files Collapse all files

1
tvix/castore/Cargo.toml
View file

@ -19,7 +19,6 @@ object_store = { workspace = true, features = ["http"] }
parking_lot = { workspace = true }
pin-project-lite = { workspace = true }
prost = { workspace = true }
sled = { workspace = true }
thiserror = { workspace = true }
tokio-stream = { workspace = true, features = ["fs", "net"] }
tokio-util = { workspace = true, features = ["io", "io-util", "codec"] }

16
tvix/castore/src/directoryservice/from_addr.rs
View file

@ -13,10 +13,6 @@ use super::DirectoryService;
/// The following URIs are supported:
/// - `memory:`
/// Uses a in-memory implementation.
/// - `sled:`
/// Uses a in-memory sled implementation.
/// - `sled:///absolute/path/to/somewhere`
/// Uses sled, using a path on the disk for persistency. Can be only opened
/// from one process at the same time.
/// - `redb:`
/// Uses a in-memory redb implementation.
@ -55,8 +51,6 @@ mod tests {
use tempfile::TempDir;
lazy_static! {
static ref TMPDIR_SLED_1: TempDir = TempDir::new().unwrap();
static ref TMPDIR_SLED_2: TempDir = TempDir::new().unwrap();
static ref TMPDIR_REDB_1: TempDir = TempDir::new().unwrap();
static ref TMPDIR_REDB_2: TempDir = TempDir::new().unwrap();
}
@ -64,16 +58,6 @@ mod tests {
#[rstest]
/// This uses an unsupported scheme.
#[case::unsupported_scheme("http://foo.example/test", false)]
/// This configures sled in temporary mode.
#[case::sled_valid_temporary("sled://", true)]
/// This configures sled with /, which should fail.
#[case::sled_invalid_root("sled:///", false)]
/// This configures sled with a host, not path, which should fail.
#[case::sled_invalid_host("sled://foo.example", false)]
/// This configures sled with a valid path path, which should succeed.
#[case::sled_valid_path(&format!("sled://{}", &TMPDIR_SLED_1.path().to_str().unwrap()), true)]
/// This configures sled with a host, and a valid path path, which should fail.
#[case::sled_invalid_host_with_valid_path(&format!("sled://foo.example{}", &TMPDIR_SLED_2.path().to_str().unwrap()), false)]
/// This correctly sets the scheme, and doesn't set a path.
#[case::memory_valid("memory://", true)]
/// This sets a memory url host to `foo`

3
tvix/castore/src/directoryservice/mod.rs
View file

@ -12,7 +12,6 @@ mod object_store;
mod order_validator;
mod redb;
mod simple_putter;
mod sled;
#[cfg(test)]
pub mod tests;
mod traverse;
@ -27,7 +26,6 @@ pub use self::object_store::{ObjectStoreDirectoryService, ObjectStoreDirectorySe
pub use self::order_validator::{LeavesToRootValidator, OrderValidator, RootToLeavesValidator};
pub use self::redb::{RedbDirectoryService, RedbDirectoryServiceConfig};
pub use self::simple_putter::SimplePutter;
pub use self::sled::{SledDirectoryService, SledDirectoryServiceConfig};
pub use self::traverse::descend_to;
pub use self::utils::traverse_directory;
@ -138,7 +136,6 @@ pub(crate) fn register_directory_services(reg: &mut Registry) {
reg.register::<Box<dyn ServiceBuilder<Output = dyn DirectoryService>>, super::directoryservice::MemoryDirectoryServiceConfig>("memory");
reg.register::<Box<dyn ServiceBuilder<Output = dyn DirectoryService>>, super::directoryservice::CacheConfig>("cache");
reg.register::<Box<dyn ServiceBuilder<Output = dyn DirectoryService>>, super::directoryservice::GRPCDirectoryServiceConfig>("grpc");
reg.register::<Box<dyn ServiceBuilder<Output = dyn DirectoryService>>, super::directoryservice::SledDirectoryServiceConfig>("sled");
reg.register::<Box<dyn ServiceBuilder<Output = dyn DirectoryService>>, super::directoryservice::RedbDirectoryServiceConfig>("redb");
#[cfg(feature = "cloud")]
{

263
tvix/castore/src/directoryservice/sled.rs
View file

@ -1,263 +0,0 @@
use futures::stream::BoxStream;
use prost::Message;
use std::ops::Deref;
use std::path::Path;
use std::sync::Arc;
use tonic::async_trait;
use tracing::{instrument, warn};
use super::utils::traverse_directory;
use super::{Directory, DirectoryGraph, DirectoryPutter, DirectoryService, LeavesToRootValidator};
use crate::composition::{CompositionContext, ServiceBuilder};
use crate::{proto, B3Digest, Error};
#[derive(Clone)]
pub struct SledDirectoryService {
db: sled::Db,
}
impl SledDirectoryService {
pub fn new<P: AsRef<Path>>(p: P) -> Result<Self, sled::Error> {
if p.as_ref() == Path::new("/") {
return Err(sled::Error::Unsupported(
"cowardly refusing to open / with sled".to_string(),
));
}
let config = sled::Config::default()
.use_compression(false) // is a required parameter
.path(p);
let db = config.open()?;
Ok(Self { db })
}
pub fn new_temporary() -> Result<Self, sled::Error> {
let config = sled::Config::default().temporary(true);
let db = config.open()?;
Ok(Self { db })
}
}
#[async_trait]
impl DirectoryService for SledDirectoryService {
#[instrument(skip(self, digest), fields(directory.digest = %digest))]
async fn get(&self, digest: &B3Digest) -> Result<Option<Directory>, Error> {
let resp = tokio::task::spawn_blocking({
let db = self.db.clone();
let digest = digest.clone();
move || db.get(digest.as_slice())
})
.await?
.map_err(|e| {
warn!("failed to retrieve directory: {}", e);
Error::StorageError(format!("failed to retrieve directory: {}", e))
})?;
match resp {
// The directory was not found, return
None => Ok(None),
// The directory was found, try to parse the data as Directory message
Some(data) => match proto::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
)));
}
let directory = directory.try_into().map_err(|e| {
warn!("failed to retrieve directory: {}", e);
Error::StorageError(format!("failed to retrieve directory: {}", e))
})?;
Ok(Some(directory))
}
Err(e) => {
warn!("unable to parse directory {}: {}", digest, e);
Err(Error::StorageError(e.to_string()))
}
},
}
}
#[instrument(skip(self, directory), fields(directory.digest = %directory.digest()))]
async fn put(&self, directory: Directory) -> Result<B3Digest, Error> {
tokio::task::spawn_blocking({
let db = self.db.clone();
move || {
let digest = directory.digest();
// store it
db.insert(
digest.as_slice(),
proto::Directory::from(directory).encode_to_vec(),
)
.map_err(|e| Error::StorageError(e.to_string()))?;
Ok(digest)
}
})
.await?
}
#[instrument(skip_all, fields(directory.digest = %root_directory_digest))]
fn get_recursive(
&self,
root_directory_digest: &B3Digest,
) -> BoxStream<'static, Result<Directory, Error>> {
traverse_directory(self.clone(), root_directory_digest)
}
#[instrument(skip_all)]
fn put_multiple_start(&self) -> Box<(dyn DirectoryPutter + 'static)>
where
Self: Clone,
{
Box::new(SledDirectoryPutter {
tree: self.db.deref().clone(),
directory_validator: Some(Default::default()),
})
}
}
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SledDirectoryServiceConfig {
is_temporary: bool,
#[serde(default)]
/// required when is_temporary = false
path: Option<String>,
}
impl TryFrom<url::Url> for SledDirectoryServiceConfig {
type Error = Box<dyn std::error::Error + Send + Sync>;
fn try_from(url: url::Url) -> Result<Self, Self::Error> {
// sled doesn't support host, and a path can be provided (otherwise
// it'll live in memory only).
if url.has_host() {
return Err(Error::StorageError("no host allowed".to_string()).into());
}
// TODO: expose compression and other parameters as URL parameters?
Ok(if url.path().is_empty() {
SledDirectoryServiceConfig {
is_temporary: true,
path: None,
}
} else {
SledDirectoryServiceConfig {
is_temporary: false,
path: Some(url.path().to_string()),
}
})
}
}
#[async_trait]
impl ServiceBuilder for SledDirectoryServiceConfig {
type Output = dyn DirectoryService;
async fn build<'a>(
&'a self,
_instance_name: &str,
_context: &CompositionContext,
) -> Result<Arc<dyn DirectoryService>, Box<dyn std::error::Error + Send + Sync + 'static>> {
match self {
SledDirectoryServiceConfig {
is_temporary: true,
path: None,
} => Ok(Arc::new(SledDirectoryService::new_temporary()?)),
SledDirectoryServiceConfig {
is_temporary: true,
path: Some(_),
} => Err(Error::StorageError(
"Temporary SledDirectoryService can not have path".into(),
)
.into()),
SledDirectoryServiceConfig {
is_temporary: false,
path: None,
} => Err(Error::StorageError("SledDirectoryService is missing path".into()).into()),
SledDirectoryServiceConfig {
is_temporary: false,
path: Some(path),
} => Ok(Arc::new(SledDirectoryService::new(path)?)),
}
}
}
/// Buffers Directory messages to be uploaded and inserts them in a batch
/// transaction on close.
pub struct SledDirectoryPutter {
tree: sled::Tree,
/// The directories (inside the directory validator) that we insert later,
/// or None, if they were already inserted.
directory_validator: Option<DirectoryGraph<LeavesToRootValidator>>,
}
#[async_trait]
impl DirectoryPutter for SledDirectoryPutter {
#[instrument(level = "trace", skip_all, fields(directory.digest=%directory.digest()), err)]
async fn put(&mut self, directory: Directory) -> Result<(), Error> {
match self.directory_validator {
None => return Err(Error::StorageError("already closed".to_string())),
Some(ref mut validator) => {
validator
.add(directory)
.map_err(|e| Error::StorageError(e.to_string()))?;
}
}
Ok(())
}
#[instrument(level = "trace", skip_all, ret, err)]
async fn close(&mut self) -> Result<B3Digest, Error> {
match self.directory_validator.take() {
None => Err(Error::InvalidRequest("already closed".to_string())),
Some(validator) => {
// Insert all directories as a batch.
tokio::task::spawn_blocking({
let tree = self.tree.clone();
move || {
// retrieve the validated directories.
let directories = validator
.validate()
.map_err(|e| Error::StorageError(e.to_string()))?
.drain_leaves_to_root()
.collect::<Vec<_>>();
// Get the root digest, which is at the end (cf. insertion order)
let root_digest = directories
.last()
.ok_or_else(|| Error::InvalidRequest("got no directories".to_string()))?
.digest();
let mut batch = sled::Batch::default();
for directory in directories {
batch.insert(
directory.digest().as_slice(),
proto::Directory::from(directory).encode_to_vec(),
);
}
tree.apply_batch(batch).map_err(|e| {
Error::StorageError(format!("unable to apply batch: {}", e))
})?;
Ok(root_digest)
}
})
.await?
}
}
}
}

1
tvix/castore/src/directoryservice/tests/mod.rs
View file

@ -23,7 +23,6 @@ use self::utils::make_grpc_directory_service_client;
#[rstest]
#[case::grpc(make_grpc_directory_service_client().await)]
#[case::memory(directoryservice::from_addr("memory://").await.unwrap())]
#[case::sled(directoryservice::from_addr("sled://").await.unwrap())]
#[case::redb(directoryservice::from_addr("redb://").await.unwrap())]
#[case::objectstore(directoryservice::from_addr("objectstore+memory://").await.unwrap())]
#[cfg_attr(all(feature = "cloud", feature = "integration"), case::bigtable(directoryservice::from_addr("bigtable://instance-1?project_id=project-1&table_name=table-1&family_name=cf1").await.unwrap()))]