Add support for an LSM-tree-based backend with Fjall

src/db/Cargo.toml

@@ -19,6 +19,7 @@ heed = { workspace = true, optional = true }
 rusqlite = { workspace = true, optional = true, features = ["backup"] }
 r2d2 = { workspace = true, optional = true }
 r2d2_sqlite = { workspace = true, optional = true }
+fjall = { workspace = true, optional = true }
 
 [dev-dependencies]
 mktemp.workspace = true
@@ -27,4 +28,5 @@ mktemp.workspace = true
 default = [ "lmdb", "sqlite" ]
 bundled-libs = [ "rusqlite?/bundled" ]
 lmdb = [ "heed" ]
+fjall = [ "dep:fjall" ]
 sqlite = [ "rusqlite", "r2d2", "r2d2_sqlite" ]

src/db/fjall_adapter.rs

@@ -0,0 +1,366 @@
+use core::ops::Bound;
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::sync::{Arc, RwLock};
+
+use fjall::{
+	PartitionCreateOptions, PersistMode, TransactionalKeyspace, TransactionalPartitionHandle,
+	WriteTransaction,
+};
+
+use crate::{
+	Db, Error, IDb, ITx, ITxFn, OnCommit, Result, TxError, TxFnResult, TxOpError, TxOpResult,
+	TxResult, TxValueIter, Value, ValueIter,
+};
+
+pub use fjall;
+
+// -- err
+
+impl From<fjall::Error> for Error {
+	fn from(e: fjall::Error) -> Error {
+		Error(format!("fjall: {}", e).into())
+	}
+}
+
+impl From<fjall::LsmError> for Error {
+	fn from(e: fjall::LsmError) -> Error {
+		Error(format!("fjall lsm_tree: {}", e).into())
+	}
+}
+
+impl From<fjall::Error> for TxOpError {
+	fn from(e: fjall::Error) -> TxOpError {
+		TxOpError(e.into())
+	}
+}
+
+// -- db
+
+pub struct FjallDb {
+	path: PathBuf,
+	keyspace: TransactionalKeyspace,
+	trees: RwLock<(Vec<TransactionalPartitionHandle>, HashMap<String, usize>)>,
+}
+
+type ByteRefRangeBound<'r> = (Bound<&'r [u8]>, Bound<&'r [u8]>);
+
+impl FjallDb {
+	pub fn init(path: &PathBuf, keyspace: TransactionalKeyspace) -> Db {
+		let s = Self {
+			path: path.clone(),
+			keyspace,
+			trees: RwLock::new((Vec::new(), HashMap::new())),
+		};
+		Db(Arc::new(s))
+	}
+
+	fn get_tree(&self, i: usize) -> Result<TransactionalPartitionHandle> {
+		self.trees
+			.read()
+			.unwrap()
+			.0
+			.get(i)
+			.cloned()
+			.ok_or_else(|| Error("invalid tree id".into()))
+	}
+
+	fn canonicalize(name: &str) -> String {
+		name.chars()
+			.map(|c| {
+				if c.is_alphanumeric() || c == '-' || c == '_' {
+					c
+				} else {
+					'_'
+				}
+			})
+			.collect::<String>()
+	}
+}
+
+impl IDb for FjallDb {
+	fn engine(&self) -> String {
+		"LSM trees (using Fjall crate)".into()
+	}
+
+	fn open_tree(&self, name: &str) -> Result<usize> {
+		let mut trees = self.trees.write().unwrap();
+		let canonical_name = FjallDb::canonicalize(name);
+		if let Some(i) = trees.1.get(&canonical_name) {
+			Ok(*i)
+		} else {
+			let tree = self
+				.keyspace
+				.open_partition(&canonical_name, PartitionCreateOptions::default())?;
+			let i = trees.0.len();
+			trees.0.push(tree);
+			trees.1.insert(canonical_name, i);
+			Ok(i)
+		}
+	}
+
+	fn list_trees(&self) -> Result<Vec<String>> {
+		Ok(self
+			.keyspace
+			.list_partitions()
+			.iter()
+			.map(|n| n.to_string())
+			.collect())
+	}
+
+	fn snapshot(&self, to: &PathBuf) -> Result<()> {
+		std::fs::create_dir_all(to)?;
+		let mut path = to.clone();
+		path.push("data.fjall");
+
+		let source_keyspace = fjall::Config::new(&self.path).open()?;
+		let copy_keyspace = fjall::Config::new(path).open()?;
+
+		for partition_name in source_keyspace.list_partitions() {
+			let source_partition = source_keyspace
+				.open_partition(&partition_name, PartitionCreateOptions::default())?;
+			let snapshot = source_partition.snapshot();
+			let copy_partition =
+				copy_keyspace.open_partition(&partition_name, PartitionCreateOptions::default())?;
+
+			for entry in snapshot.iter() {
+				let (key, value) = entry?;
+				copy_partition.insert(key, value)?;
+			}
+		}
+
+		copy_keyspace.persist(PersistMode::SyncAll)?;
+		Ok(())
+	}
+
+	// ----
+
+	fn get(&self, tree_idx: usize, key: &[u8]) -> Result<Option<Value>> {
+		let tree = self.get_tree(tree_idx)?;
+		let tx = self.keyspace.read_tx();
+		let val = tx.get(&tree, key)?;
+		match val {
+			None => Ok(None),
+			Some(v) => Ok(Some(v.to_vec())),
+		}
+	}
+
+	fn len(&self, tree_idx: usize) -> Result<usize> {
+		let tree = self.get_tree(tree_idx)?;
+		let tx = self.keyspace.read_tx();
+		Ok(tx.len(&tree)?)
+	}
+
+	fn insert(&self, tree_idx: usize, key: &[u8], value: &[u8]) -> Result<()> {
+		let tree = self.get_tree(tree_idx)?;
+		let mut tx = self.keyspace.write_tx();
+		tx.insert(&tree, key, value);
+		tx.commit()?;
+		Ok(())
+	}
+
+	fn remove(&self, tree_idx: usize, key: &[u8]) -> Result<()> {
+		let tree = self.get_tree(tree_idx)?;
+		let mut tx = self.keyspace.write_tx();
+		tx.remove(&tree, key);
+		tx.commit()?;
+		Ok(())
+	}
+
+	fn clear(&self, tree_idx: usize) -> Result<()> {
+		let tree = self.get_tree(tree_idx)?;
+		let tree_name = tree.inner().name.clone();
+		self.keyspace.delete_partition(tree)?;
+		let tree = self
+			.keyspace
+			.open_partition(&tree_name, PartitionCreateOptions::default())?;
+		let mut trees = self.trees.write().unwrap();
+		trees.0[tree_idx] = tree;
+		Ok(())
+	}
+
+	fn iter(&self, tree_idx: usize) -> Result<ValueIter<'_>> {
+		let tree = self.get_tree(tree_idx)?;
+		let tx = self.keyspace.read_tx();
+		Ok(Box::new(tx.iter(&tree).map(iterator_remap)))
+	}
+
+	fn iter_rev(&self, tree_idx: usize) -> Result<ValueIter<'_>> {
+		let tree = self.get_tree(tree_idx)?;
+		let tx = self.keyspace.read_tx();
+		Ok(Box::new(tx.iter(&tree).rev().map(iterator_remap)))
+	}
+
+	fn range<'r>(
+		&self,
+		tree_idx: usize,
+		low: Bound<&'r [u8]>,
+		high: Bound<&'r [u8]>,
+	) -> Result<ValueIter<'_>> {
+		let tree = self.get_tree(tree_idx)?;
+		let tx = self.keyspace.read_tx();
+		Ok(Box::new(
+			tx.range::<&'r [u8], ByteRefRangeBound>(&tree, (low, high))
+				.map(iterator_remap),
+		))
+	}
+	fn range_rev<'r>(
+		&self,
+		tree_idx: usize,
+		low: Bound<&'r [u8]>,
+		high: Bound<&'r [u8]>,
+	) -> Result<ValueIter<'_>> {
+		let tree = self.get_tree(tree_idx)?;
+		let tx = self.keyspace.read_tx();
+		Ok(Box::new(
+			tx.range::<&'r [u8], ByteRefRangeBound>(&tree, (low, high))
+				.rev()
+				.map(iterator_remap),
+		))
+	}
+
+	// ----
+
+	fn transaction(&self, f: &dyn ITxFn) -> TxResult<OnCommit, ()> {
+		let trees = self.trees.read().unwrap();
+		let mut tx = FjallTx {
+			trees: &trees.0[..],
+			tx: self.keyspace.write_tx(),
+		};
+
+		let res = f.try_on(&mut tx);
+		match res {
+			TxFnResult::Ok(on_commit) => {
+				tx.tx.commit().map_err(Error::from).map_err(TxError::Db)?;
+				Ok(on_commit)
+			}
+			TxFnResult::Abort => {
+				tx.tx.rollback();
+				Err(TxError::Abort(()))
+			}
+			TxFnResult::DbErr => {
+				tx.tx.rollback();
+				Err(TxError::Db(Error(
+					"(this message will be discarded)".into(),
+				)))
+			}
+		}
+	}
+}
+
+// ----
+
+struct FjallTx<'a> {
+	trees: &'a [TransactionalPartitionHandle],
+	tx: WriteTransaction<'a>,
+}
+
+impl<'a> FjallTx<'a> {
+	fn get_tree(&self, i: usize) -> TxOpResult<&TransactionalPartitionHandle> {
+		self.trees.get(i).ok_or_else(|| {
+			TxOpError(Error(
+				"invalid tree id (it might have been openned after the transaction started)".into(),
+			))
+		})
+	}
+}
+
+impl<'a> ITx for FjallTx<'a> {
+	fn get(&self, tree_idx: usize, key: &[u8]) -> TxOpResult<Option<Value>> {
+		let tree = self.get_tree(tree_idx)?;
+		match self.tx.get(tree, key)? {
+			Some(v) => Ok(Some(v.to_vec())),
+			None => Ok(None),
+		}
+	}
+	fn len(&self, tree_idx: usize) -> TxOpResult<usize> {
+		let tree = self.get_tree(tree_idx)?;
+		Ok(self.tx.len(tree)? as usize)
+	}
+
+	fn insert(&mut self, tree_idx: usize, key: &[u8], value: &[u8]) -> TxOpResult<()> {
+		let tree = self.get_tree(tree_idx)?.clone();
+		self.tx.insert(&tree, key, value);
+		Ok(())
+	}
+	fn remove(&mut self, tree_idx: usize, key: &[u8]) -> TxOpResult<()> {
+		let tree = self.get_tree(tree_idx)?.clone();
+		self.tx.remove(&tree, key);
+		Ok(())
+	}
+	fn clear(&mut self, _tree_idx: usize) -> TxOpResult<()> {
+		unimplemented!("LSM tree clearing in cross-partition transaction is not supported")
+	}
+
+	fn iter(&self, tree_idx: usize) -> TxOpResult<TxValueIter<'_>> {
+		let tree = self.get_tree(tree_idx)?.clone();
+		Ok(Box::new(self.tx.iter(&tree).map(iterator_remap_tx)))
+	}
+	fn iter_rev(&self, tree_idx: usize) -> TxOpResult<TxValueIter<'_>> {
+		let tree = self.get_tree(tree_idx)?.clone();
+		Ok(Box::new(self.tx.iter(&tree).rev().map(iterator_remap_tx)))
+	}
+
+	fn range<'r>(
+		&self,
+		tree_idx: usize,
+		low: Bound<&'r [u8]>,
+		high: Bound<&'r [u8]>,
+	) -> TxOpResult<TxValueIter<'_>> {
+		let tree = self.get_tree(tree_idx)?;
+		let low = clone_bound(low);
+		let high = clone_bound(high);
+		Ok(Box::new(
+			self.tx
+				.range::<Vec<u8>, ByteVecRangeBounds>(&tree, (low, high))
+				.map(iterator_remap_tx),
+		))
+	}
+	fn range_rev<'r>(
+		&self,
+		tree_idx: usize,
+		low: Bound<&'r [u8]>,
+		high: Bound<&'r [u8]>,
+	) -> TxOpResult<TxValueIter<'_>> {
+		let tree = self.get_tree(tree_idx)?;
+		let low = clone_bound(low);
+		let high = clone_bound(high);
+		Ok(Box::new(
+			self.tx
+				.range::<Vec<u8>, ByteVecRangeBounds>(&tree, (low, high))
+				.rev()
+				.map(iterator_remap_tx),
+		))
+	}
+}
+
+// -- maps fjall's (k, v) to ours
+
+fn iterator_remap(r: fjall::Result<(fjall::Slice, fjall::Slice)>) -> Result<(Value, Value)> {
+	r.map(|(k, v)| (k.to_vec(), v.to_vec()))
+		.map_err(|e| e.into())
+}
+
+fn iterator_remap_tx(r: fjall::Result<(fjall::Slice, fjall::Slice)>) -> TxOpResult<(Value, Value)> {
+	r.map(|(k, v)| (k.to_vec(), v.to_vec()))
+		.map_err(|e| e.into())
+}
+
+// -- utils to deal with Garage's tightness on Bound lifetimes
+
+type ByteVecBound = Bound<Vec<u8>>;
+type ByteVecRangeBounds = (ByteVecBound, ByteVecBound);
+
+fn clone_bound(bound: Bound<&[u8]>) -> ByteVecBound {
+	let value = match bound {
+		Bound::Excluded(v) | Bound::Included(v) => v.to_vec(),
+		Bound::Unbounded => vec![],
+	};
+
+	match bound {
+		Bound::Included(_) => Bound::Included(value),
+		Bound::Excluded(_) => Bound::Excluded(value),
+		Bound::Unbounded => Bound::Unbounded,
+	}
+}

src/db/lib.rs

@@ -1,6 +1,8 @@
 #[macro_use]
 extern crate tracing;
 
+#[cfg(feature = "fjall")]
+pub mod fjall_adapter;
 #[cfg(feature = "lmdb")]
 pub mod lmdb_adapter;
 #[cfg(feature = "sqlite")]

src/db/open.rs

@@ -1,4 +1,6 @@
+use std::convert::TryInto;
 use std::path::PathBuf;
+use std::sync::Arc;
 
 use crate::{Db, Error, Result};
 
@@ -11,6 +13,7 @@ use crate::{Db, Error, Result};
 pub enum Engine {
 	Lmdb,
 	Sqlite,
+	Fjall,
 }
 
 impl Engine {
@@ -19,6 +22,7 @@ impl Engine {
 		match self {
 			Self::Lmdb => "lmdb",
 			Self::Sqlite => "sqlite",
+			Self::Fjall => "fjall",
 		}
 	}
 }
@@ -36,6 +40,7 @@ impl std::str::FromStr for Engine {
 		match text {
 			"lmdb" | "heed" => Ok(Self::Lmdb),
 			"sqlite" | "sqlite3" | "rusqlite" => Ok(Self::Sqlite),
+            "fjall" => Ok(Self::Fjall),
 			"sled" => Err(Error("Sled is no longer supported as a database engine. Converting your old metadata db can be done using an older Garage binary (e.g. v0.9.4).".into())),
 			kind => Err(Error(
 				format!(
@@ -51,6 +56,7 @@ impl std::str::FromStr for Engine {
 pub struct OpenOpt {
 	pub fsync: bool,
 	pub lmdb_map_size: Option<usize>,
+	pub fjall_block_cache_size: Option<usize>,
 }
 
 impl Default for OpenOpt {
@@ -58,6 +64,7 @@ impl Default for OpenOpt {
 		Self {
 			fsync: false,
 			lmdb_map_size: None,
+			fjall_block_cache_size: None,
 		}
 	}
 }
@@ -114,6 +121,22 @@ pub fn open_db(path: &PathBuf, engine: Engine, opt: &OpenOpt) -> Result<Db> {
 			}
 		}
 
+		// ---- Fjall DB ----
+		#[cfg(feature = "fjall")]
+		Engine::Fjall => {
+			info!("Opening Fjall database at: {}", path.display());
+			let fsync_ms = opt.fsync.then(|| 1000 as u16);
+			let mut config = fjall::Config::new(path).fsync_ms(fsync_ms);
+			if let Some(block_cache_size) = opt.fjall_block_cache_size {
+				let block_cache = Arc::new(fjall::BlockCache::with_capacity_bytes(
+					block_cache_size.try_into().unwrap(),
+				));
+				config = config.block_cache(block_cache);
+			}
+			let keyspace = config.open_transactional()?;
+			Ok(crate::fjall_adapter::FjallDb::init(path, keyspace))
+		}
+
 		// Pattern is unreachable when all supported DB engines are compiled into binary. The allow
 		// attribute is added so that we won't have to change this match in case stop building
 		// support for one or more engines by default.

src/garage/Cargo.toml

@@ -91,6 +91,7 @@ k2v = [ "garage_util/k2v", "garage_api_k2v" ]
 # Database engines
 lmdb = [ "garage_model/lmdb" ]
 sqlite = [ "garage_model/sqlite" ]
+fjall = [ "garage_model/fjall" ]
 
 # Automatic registration and discovery via Consul API
 consul-discovery = [ "garage_rpc/consul-discovery" ]

src/model/Cargo.toml

@@ -44,3 +44,4 @@ default = [ "lmdb", "sqlite" ]
 k2v = [ "garage_util/k2v" ]
 lmdb = [ "garage_db/lmdb" ]
 sqlite = [ "garage_db/sqlite" ]
+fjall = [ "garage_db/fjall" ]

src/model/garage.rs

@@ -124,6 +124,9 @@ impl Garage {
 			db::Engine::Lmdb => {
 				db_path.push("db.lmdb");
 			}
+			db::Engine::Fjall => {
+				db_path.push("db.fjall");
+			}
 		}
 		let db_opt = db::OpenOpt {
 			fsync: config.metadata_fsync,
@@ -131,6 +134,10 @@ impl Garage {
 				v if v == usize::default() => None,
 				v => Some(v),
 			},
+			fjall_block_cache_size: match config.fjall_block_cache_size {
+				v if v == usize::default() => None,
+				v => Some(v),
+			},
 		};
 		let db = db::open_db(&db_path, db_engine, &db_opt)
 			.ok_or_message("Unable to open metadata db")?;

src/util/config.rs

@@ -122,6 +122,10 @@ pub struct Config {
 	#[serde(deserialize_with = "deserialize_capacity", default)]
 	pub lmdb_map_size: usize,
 
+	/// Fjall block cache size
+	#[serde(deserialize_with = "deserialize_capacity", default)]
+	pub fjall_block_cache_size: usize,
+
 	// -- APIs
 	/// Configuration for S3 api
 	pub s3_api: S3ApiConfig,