4 files changed, 55 insertions, 39 deletions

diff --git a/src/block/Cargo.toml b/src/block/Cargo.toml
index 9cba69ee..80346aca 100644
--- a/src/block/Cargo.toml
+++ b/src/block/Cargo.toml
@@ -14,6 +14,7 @@ path = "lib.rs"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+garage_db = { version = "0.8.0", path = "../db" }
 garage_rpc = { version = "0.7.0", path = "../rpc" }
 garage_util = { version = "0.7.0", path = "../util" }
 garage_table = { version = "0.7.0", path = "../table" }
@@ -27,8 +28,6 @@ tracing = "0.1.30"
 rand = "0.8"
 zstd = { version = "0.9", default-features = false }
 
-sled = "0.34"
-
 rmp-serde = "0.15"
 serde = { version = "1.0", default-features = false, features = ["derive", "rc"] }
 serde_bytes = "0.11"
diff --git a/src/block/manager.rs b/src/block/manager.rs
index 9b2d9cad..50039d2b 100644
--- a/src/block/manager.rs
+++ b/src/block/manager.rs
@@ -17,10 +17,11 @@ use opentelemetry::{
 	Context, KeyValue,
 };
 
+use garage_db as db;
+
 use garage_util::data::*;
 use garage_util::error::*;
 use garage_util::metrics::RecordDuration;
-use garage_util::sled_counter::SledCountedTree;
 use garage_util::time::*;
 use garage_util::tranquilizer::Tranquilizer;
 
@@ -91,9 +92,9 @@ pub struct BlockManager {
 
 	rc: BlockRc,
 
-	resync_queue: SledCountedTree,
+	resync_queue: db::Tree,
 	resync_notify: Notify,
-	resync_errors: SledCountedTree,
+	resync_errors: db::Tree,
 
 	system: Arc<System>,
 	endpoint: Arc<Endpoint<BlockRpc, Self>>,
@@ -108,7 +109,7 @@ struct BlockManagerLocked();
 
 impl BlockManager {
 	pub fn new(
-		db: &sled::Db,
+		db: &db::Db,
 		data_dir: PathBuf,
 		compression_level: Option<i32>,
 		background_tranquility: u32,
@@ -123,12 +124,10 @@ impl BlockManager {
 		let resync_queue = db
 			.open_tree("block_local_resync_queue")
 			.expect("Unable to open block_local_resync_queue tree");
-		let resync_queue = SledCountedTree::new(resync_queue);
 
 		let resync_errors = db
 			.open_tree("block_local_resync_errors")
 			.expect("Unable to open block_local_resync_errors tree");
-		let resync_errors = SledCountedTree::new(resync_errors);
 
 		let endpoint = system
 			.netapp
@@ -219,7 +218,7 @@ impl BlockManager {
 	/// to fix any mismatch between the two.
 	pub async fn repair_data_store(&self, must_exit: &watch::Receiver<bool>) -> Result<(), Error> {
 		// 1. Repair blocks from RC table.
-		for (i, entry) in self.rc.rc.iter().enumerate() {
+		for (i, entry) in self.rc.rc.iter()?.enumerate() {
 			let (hash, _) = entry?;
 			let hash = Hash::try_from(&hash[..]).unwrap();
 			self.put_to_resync(&hash, Duration::from_secs(0))?;
@@ -265,17 +264,17 @@ impl BlockManager {
 
 	/// Get lenght of resync queue
 	pub fn resync_queue_len(&self) -> usize {
-		self.resync_queue.len()
+		self.resync_queue.len().unwrap() // TODO fix unwrap
 	}
 
 	/// Get number of blocks that have an error
 	pub fn resync_errors_len(&self) -> usize {
-		self.resync_errors.len()
+		self.resync_errors.len().unwrap() // TODO fix unwrap
 	}
 
 	/// Get number of items in the refcount table
 	pub fn rc_len(&self) -> usize {
-		self.rc.rc.len()
+		self.rc.rc.len().unwrap() // TODO fix unwrap
 	}
 
 	//// ----- Managing the reference counter ----
@@ -503,12 +502,12 @@ impl BlockManager {
 		});
 	}
 
-	fn put_to_resync(&self, hash: &Hash, delay: Duration) -> Result<(), sled::Error> {
+	fn put_to_resync(&self, hash: &Hash, delay: Duration) -> Result<(), db::Error> {
 		let when = now_msec() + delay.as_millis() as u64;
 		self.put_to_resync_at(hash, when)
 	}
 
-	fn put_to_resync_at(&self, hash: &Hash, when: u64) -> Result<(), sled::Error> {
+	fn put_to_resync_at(&self, hash: &Hash, when: u64) -> Result<(), db::Error> {
 		trace!("Put resync_queue: {} {:?}", when, hash);
 		let mut key = u64::to_be_bytes(when).to_vec();
 		key.extend(hash.as_ref());
@@ -547,11 +546,8 @@ impl BlockManager {
 	// - Ok(true) -> a block was processed (successfully or not)
 	// - Ok(false) -> no block was processed, but we are ready for the next iteration
 	// - Err(_) -> a Sled error occurred when reading/writing from resync_queue/resync_errors
-	async fn resync_iter(
-		&self,
-		must_exit: &mut watch::Receiver<bool>,
-	) -> Result<bool, sled::Error> {
-		if let Some(first_pair_res) = self.resync_queue.iter().next() {
+	async fn resync_iter(&self, must_exit: &mut watch::Receiver<bool>) -> Result<bool, db::Error> {
+		if let Some(first_pair_res) = self.resync_queue.iter()?.next() {
 			let (time_bytes, hash_bytes) = first_pair_res?;
 
 			let time_msec = u64::from_be_bytes(time_bytes[0..8].try_into().unwrap());
@@ -966,7 +962,7 @@ impl ErrorCounter {
 		}
 	}
 
-	fn decode(data: sled::IVec) -> Self {
+	fn decode<'a>(data: db::Value<'a>) -> Self {
 		Self {
 			errors: u64::from_be_bytes(data[0..8].try_into().unwrap()),
 			last_try: u64::from_be_bytes(data[8..16].try_into().unwrap()),
diff --git a/src/block/metrics.rs b/src/block/metrics.rs
index f0f541a3..1d4d0028 100644
--- a/src/block/metrics.rs
+++ b/src/block/metrics.rs
@@ -1,6 +1,6 @@
 use opentelemetry::{global, metrics::*};
 
-use garage_util::sled_counter::SledCountedTree;
+use garage_db as db;
 
 /// TableMetrics reference all counter used for metrics
 pub struct BlockManagerMetrics {
@@ -23,12 +23,12 @@ pub struct BlockManagerMetrics {
 }
 
 impl BlockManagerMetrics {
-	pub fn new(resync_queue: SledCountedTree, resync_errors: SledCountedTree) -> Self {
+	pub fn new(resync_queue: db::Tree, resync_errors: db::Tree) -> Self {
 		let meter = global::meter("garage_model/block");
 		Self {
 			_resync_queue_len: meter
 				.u64_value_observer("block.resync_queue_length", move |observer| {
-					observer.observe(resync_queue.len() as u64, &[])
+					observer.observe(resync_queue.len().unwrap() as u64, &[]) // TODO fix unwrap
 				})
 				.with_description(
 					"Number of block hashes queued for local check and possible resync",
@@ -36,7 +36,7 @@ impl BlockManagerMetrics {
 				.init(),
 			_resync_errored_blocks: meter
 				.u64_value_observer("block.resync_errored_blocks", move |observer| {
-					observer.observe(resync_errors.len() as u64, &[])
+					observer.observe(resync_errors.len().unwrap() as u64, &[]) // TODO fix unwrap
 				})
 				.with_description("Number of block hashes whose last resync resulted in an error")
 				.init(),
diff --git a/src/block/rc.rs b/src/block/rc.rs
index ec3ea44e..f6d8c2aa 100644
--- a/src/block/rc.rs
+++ b/src/block/rc.rs
@@ -1,5 +1,7 @@
 use std::convert::TryInto;
 
+use garage_db as db;
+
 use garage_util::data::*;
 use garage_util::error::*;
 use garage_util::time::*;
@@ -7,31 +9,47 @@ use garage_util::time::*;
 use crate::manager::BLOCK_GC_DELAY;
 
 pub struct BlockRc {
-	pub(crate) rc: sled::Tree,
+	pub(crate) rc: db::Tree,
 }
 
 impl BlockRc {
-	pub(crate) fn new(rc: sled::Tree) -> Self {
+	pub(crate) fn new(rc: db::Tree) -> Self {
 		Self { rc }
 	}
 
 	/// Increment the reference counter associated to a hash.
 	/// Returns true if the RC goes from zero to nonzero.
 	pub(crate) fn block_incref(&self, hash: &Hash) -> Result<bool, Error> {
-		let old_rc = self
-			.rc
-			.fetch_and_update(&hash, |old| RcEntry::parse_opt(old).increment().serialize())?;
-		let old_rc = RcEntry::parse_opt(old_rc);
+		let old_rc = self.rc.db().transaction(|tx| {
+			let old_rc = RcEntry::parse_opt(tx.get(&self.rc, &hash)?);
+			match old_rc.increment().serialize() {
+				Some(x) => {
+					tx.insert(&self.rc, &hash, x)?;
+				}
+				None => {
+					tx.remove(&self.rc, &hash)?;
+				}
+			};
+			tx.commit(old_rc)
+		})?;
 		Ok(old_rc.is_zero())
 	}
 
 	/// Decrement the reference counter associated to a hash.
 	/// Returns true if the RC is now zero.
 	pub(crate) fn block_decref(&self, hash: &Hash) -> Result<bool, Error> {
-		let new_rc = self
-			.rc
-			.update_and_fetch(&hash, |old| RcEntry::parse_opt(old).decrement().serialize())?;
-		let new_rc = RcEntry::parse_opt(new_rc);
+		let new_rc = self.rc.db().transaction(|tx| {
+			let new_rc = RcEntry::parse_opt(tx.get(&self.rc, &hash)?).decrement();
+			match new_rc.serialize() {
+				Some(x) => {
+					tx.insert(&self.rc, &hash, x)?;
+				}
+				None => {
+					tx.remove(&self.rc, &hash)?;
+				}
+			};
+			tx.commit(new_rc)
+		})?;
 		Ok(matches!(new_rc, RcEntry::Deletable { .. }))
 	}
 
@@ -44,12 +62,15 @@ impl BlockRc {
 	/// deletion time has passed
 	pub(crate) fn clear_deleted_block_rc(&self, hash: &Hash) -> Result<(), Error> {
 		let now = now_msec();
-		self.rc.update_and_fetch(&hash, |rcval| {
-			let updated = match RcEntry::parse_opt(rcval) {
-				RcEntry::Deletable { at_time } if now > at_time => RcEntry::Absent,
-				v => v,
+		self.rc.db().transaction(|tx| {
+			let rcval = RcEntry::parse_opt(tx.get(&self.rc, &hash)?);
+			match rcval {
+				RcEntry::Deletable { at_time } if now > at_time => {
+					tx.remove(&self.rc, &hash)?;
+				}
+				_ => (),
 			};
-			updated.serialize()
+			tx.commit(())
 		})?;
 		Ok(())
 	}