Migrate BlockRc and BlockResyncManager to TypedTree

src/api/admin/repair.rs

@@ -360,14 +360,7 @@ impl Worker for BlockRcRepair {
 		_must_exit: &mut watch::Receiver<bool>,
 	) -> Result<WorkerState, GarageError> {
 		for _i in 0..RC_REPAIR_ITER_COUNT {
-			let next1 = self
-				.block_manager
-				.rc
-				.rc_table
-				.range(self.cursor.as_slice()..)?
-				.next()
-				.transpose()?
-				.map(|(k, _)| Hash::try_from(k.as_slice()).unwrap());
+			let next1 = self.block_manager.rc.get_first_hash_from(self.cursor)?;
 			let next2 = self
 				.block_ref_table
 				.data

src/block/manager.rs

@@ -144,7 +144,7 @@ impl BlockManager {
 
 		// Open metadata tables
 		let rc = db
-			.open_tree("block_local_rc")
+			.open_typed_tree("block_local_rc")
 			.expect("Unable to open block_local_rc tree");
 		let rc = BlockRc::new(rc);
 
@@ -158,9 +158,9 @@ impl BlockManager {
 
 		let metrics = BlockManagerMetrics::new(
 			config.compression_level,
-			rc.rc_table.clone(),
-			resync.queue.clone(),
-			resync.errors.clone(),
+			rc.rc_table.untyped().clone(),
+			resync.queue.untyped().clone(),
+			resync.errors.untyped().clone(),
 			buffer_kb_semaphore.clone(),
 		);
 
@@ -449,9 +449,8 @@ impl BlockManager {
 		let mut blocks = Vec::with_capacity(self.resync.errors.approximate_len()?);
 		for ent in self.resync.errors.iter()? {
 			let (hash, cnt) = ent?;
-			let cnt = ErrorCounter::decode(&cnt);
 			blocks.push(BlockResyncErrorInfo {
-				hash: Hash::try_from(&hash).unwrap(),
+				hash,
 				refcount: 0,
 				error_count: cnt.errors,
 				last_try: cnt.last_try,

src/block/rc.rs

@@ -14,12 +14,12 @@ pub type CalculateRefcount =
 	Box<dyn Fn(&db::Transaction, &Hash) -> db::TxResult<usize, Error> + Send + Sync>;
 
 pub struct BlockRc {
-	pub rc_table: db::Tree,
+	pub(crate) rc_table: db::TypedTree<Hash, RcEntry>,
 	pub(crate) recalc_rc: ArcSwapOption<Vec<CalculateRefcount>>,
 }
 
 impl BlockRc {
-	pub(crate) fn new(rc: db::Tree) -> Self {
+	pub(crate) fn new(rc: db::TypedTree<Hash, RcEntry>) -> Self {
 		Self {
 			rc_table: rc,
 			recalc_rc: ArcSwapOption::new(None),
@@ -33,11 +33,8 @@ impl BlockRc {
 		tx: &mut db::Transaction,
 		hash: &Hash,
 	) -> db::TxOpResult<bool> {
-		let old_rc = RcEntry::parse_opt(tx.get(&self.rc_table, hash)?);
-		match old_rc.increment().serialize() {
-			Some(x) => tx.insert(&self.rc_table, hash, x)?,
-			None => unreachable!(),
-		}
+		let old_rc = self.rc_table.tx_get(tx, hash)?.unwrap_or(RcEntry::Absent);
+		self.rc_table.tx_insert(tx, hash, &old_rc.increment())?;
 		Ok(old_rc.is_zero())
 	}
 
@@ -48,17 +45,27 @@ impl BlockRc {
 		tx: &mut db::Transaction,
 		hash: &Hash,
 	) -> db::TxOpResult<bool> {
-		let new_rc = RcEntry::parse_opt(tx.get(&self.rc_table, hash)?).decrement();
-		match new_rc.serialize() {
-			Some(x) => tx.insert(&self.rc_table, hash, x)?,
-			None => tx.remove(&self.rc_table, hash)?,
+		let new_rc = self.rc_table.tx_get(tx, hash)?.unwrap_or(RcEntry::Absent).decrement();
+		match new_rc {
+			RcEntry::Absent => self.rc_table.tx_remove(tx, hash)?,
+			_ => self.rc_table.tx_insert(tx, hash, &new_rc)?,
 		}
 		Ok(matches!(new_rc, RcEntry::Deletable { .. }))
 	}
 
 	/// Read a block's reference count
 	pub(crate) fn get_block_rc(&self, hash: &Hash) -> Result<RcEntry, Error> {
-		Ok(RcEntry::parse_opt(self.rc_table.get(hash.as_ref())?))
+		Ok(self.rc_table.get(hash)?.unwrap_or(RcEntry::Absent))
+	}
+
+	/// Return the first hash stored in the RC table at or after `cursor`
+	pub fn get_first_hash_from(&self, cursor: Hash) -> Result<Option<Hash>, Error> {
+		Ok(self
+			.rc_table
+			.range(cursor..)?
+			.next()
+			.transpose()?
+			.map(|(k, _)| k))
 	}
 
 	/// Delete an entry in the RC table if it is deletable and the
@@ -66,12 +73,11 @@ impl BlockRc {
 	pub(crate) fn clear_deleted_block_rc(&self, hash: &Hash) -> Result<(), Error> {
 		let now = now_msec();
 		self.rc_table.db().transaction(|tx| {
-			let rcval = RcEntry::parse_opt(tx.get(&self.rc_table, hash)?);
-			match rcval {
-				RcEntry::Deletable { at_time } if now > at_time => {
-					tx.remove(&self.rc_table, hash)?;
+			let rcval = self.rc_table.tx_get(tx, hash)?.unwrap_or(RcEntry::Absent);
+			if let RcEntry::Deletable { at_time } = rcval {
+				if now > at_time {
+					self.rc_table.tx_remove(tx, hash)?;
 				}
-				_ => (),
 			}
 			Ok(())
 		})?;
@@ -91,7 +97,7 @@ impl BlockRc {
 					for f in recalc_fns.iter() {
 						cnt += f(tx, hash)?;
 					}
-					let old_rc = RcEntry::parse_opt(tx.get(&self.rc_table, hash)?);
+					let old_rc = self.rc_table.tx_get(tx, hash)?.unwrap_or(RcEntry::Absent);
 					trace!(
 						"Block RC for {:?}: stored={}, calculated={}",
 						hash,
@@ -112,7 +118,7 @@ impl BlockRc {
 								at_time: now_msec() + BLOCK_GC_DELAY.as_millis() as u64,
 							}
 						};
-						tx.insert(&self.rc_table, hash, new_rc.serialize().unwrap())?;
+						self.rc_table.tx_insert(tx, hash, &new_rc)?;
 						Ok((cnt, true))
 					} else {
 						Ok((cnt, false))
@@ -131,6 +137,38 @@ impl BlockRc {
 	}
 }
 
+impl db::DbBytes for RcEntry {
+	fn encode(&self) -> Vec<u8> {
+		match self {
+			RcEntry::Present { count } => u64::to_be_bytes(*count).to_vec(),
+			RcEntry::Deletable { at_time } => {
+				[u64::to_be_bytes(0), u64::to_be_bytes(*at_time)].concat()
+			}
+			RcEntry::Absent => panic!("cannot encode RcEntry::Absent"),
+		}
+	}
+
+	fn decode(bytes: &[u8]) -> db::Result<Self> {
+		if bytes.len() == 8 {
+			Ok(RcEntry::Present {
+				count: u64::from_be_bytes(bytes.try_into().unwrap()),
+			})
+		} else if bytes.len() == 16 {
+			Ok(RcEntry::Deletable {
+				at_time: u64::from_be_bytes(bytes[8..16].try_into().unwrap()),
+			})
+		} else {
+			Err(db::Error::Decode(
+				format!(
+					"invalid RC entry: expected 8 or 16 bytes, got {}",
+					bytes.len()
+				)
+				.into(),
+			))
+		}
+	}
+}
+
 /// Describes the state of the reference counter for a block
 #[derive(Clone, Copy, Debug)]
 pub(crate) enum RcEntry {
@@ -154,38 +192,6 @@ pub(crate) enum RcEntry {
 }
 
 impl RcEntry {
-	fn parse(bytes: &[u8]) -> Self {
-		if bytes.len() == 8 {
-			RcEntry::Present {
-				count: u64::from_be_bytes(bytes.try_into().unwrap()),
-			}
-		} else if bytes.len() == 16 {
-			RcEntry::Deletable {
-				at_time: u64::from_be_bytes(bytes[8..16].try_into().unwrap()),
-			}
-		} else {
-			panic!("Invalid RC entry: {:?}, database is corrupted. This is an error Garage is currently unable to recover from. Sorry, and also please report a bug.",
-				bytes
-			)
-		}
-	}
-
-	fn parse_opt<V: AsRef<[u8]>>(bytes: Option<V>) -> Self {
-		bytes
-			.map(|b| Self::parse(b.as_ref()))
-			.unwrap_or(Self::Absent)
-	}
-
-	fn serialize(self) -> Option<Vec<u8>> {
-		match self {
-			RcEntry::Present { count } => Some(u64::to_be_bytes(count).to_vec()),
-			RcEntry::Deletable { at_time } => {
-				Some([u64::to_be_bytes(0), u64::to_be_bytes(at_time)].concat())
-			}
-			RcEntry::Absent => None,
-		}
-	}
-
 	fn increment(self) -> Self {
 		let old_count = match self {
 			RcEntry::Present { count } => count,

src/block/repair.rs

@@ -102,16 +102,15 @@ impl Worker for RepairWorker {
 				let mut batch_of_hashes = vec![];
 				let start_bound = match self.next_start.as_ref() {
 					None => Bound::Unbounded,
-					Some(x) => Bound::Excluded(x.as_slice()),
+					Some(x) => Bound::Excluded(*x),
 				};
 				for entry in self
 					.manager
 					.rc
 					.rc_table
-					.range::<&[u8], _>((start_bound, Bound::Unbounded))?
+					.range((start_bound, Bound::Unbounded))?
 				{
 					let (hash, _) = entry?;
-					let hash = Hash::try_from(&hash[..]).unwrap();
 					batch_of_hashes.push(hash);
 					if batch_of_hashes.len() >= 1000 {
 						break;

src/block/resync.rs

@@ -44,15 +44,58 @@ pub(crate) const MAX_RESYNC_WORKERS: usize = 8;
 const INITIAL_RESYNC_TRANQUILITY: u32 = 2;
 
 pub struct BlockResyncManager {
-	pub(crate) queue: db::Tree,
+	pub(crate) queue: db::TypedTree<ResyncQueueKey, Hash>,
 	pub(crate) notify: Arc<Notify>,
-	pub(crate) errors: db::Tree,
+	pub(crate) errors: db::TypedTree<Hash, ErrorCounter>,
 
 	busy_set: BusySet,
 
 	persister: PersisterShared<ResyncPersistedConfig>,
 }
 
+/// Key of the resync queue tree: blocks are resynced in order of increasing
+/// `when` (msec timestamp of the next try), with the block hash as tie-breaker.
+///
+// CAREFUL: this type implements `DbOrdKey`, so its byte encoding must be
+// order-preserving.
+// The derived `Ord` compares fields in declaration order, which must match
+// the order in which `encode()` writes them; and `when` must remain an
+// *unsigned* integer, as the big-endian encoding is only order-preserving
+// for unsigned types.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub(crate) struct ResyncQueueKey {
+	pub(crate) when: u64,
+	pub(crate) hash: Hash,
+}
+
+impl db::DbBytes for ResyncQueueKey {
+	fn encode(&self) -> Vec<u8> {
+		let mut v = Vec::with_capacity(40);
+		v.extend_from_slice(&u64::to_be_bytes(self.when));
+		v.extend_from_slice(self.hash.as_slice());
+		v
+	}
+
+	fn decode(bytes: &[u8]) -> std::result::Result<Self, db::DecodeError> {
+		if bytes.len() != 40 {
+			return Err(db::DecodeError(
+				format!(
+					"invalid resync queue key: expected 40 bytes, got {}",
+					bytes.len()
+				)
+				.into(),
+			));
+		}
+		Ok(ResyncQueueKey {
+			when: u64::from_be_bytes(bytes[..8].try_into().unwrap()),
+			hash: Hash::try_from(&bytes[8..])
+				.ok_or_else(|| db::DecodeError("invalid resync queue key: bad hash".into()))?,
+		})
+	}
+}
+
+impl db::DbOrdKey for ResyncQueueKey {}
+
 #[derive(Serialize, Deserialize, Clone, Copy)]
 struct ResyncPersistedConfig {
 	n_workers: usize,
@@ -74,22 +117,21 @@ enum ResyncIterResult {
 	IdleFor(Duration),
 }
 
-type BusySet = Arc<Mutex<HashSet<Vec<u8>>>>;
+type BusySet = Arc<Mutex<HashSet<ResyncQueueKey>>>;
 
 struct BusyBlock {
-	time_bytes: Vec<u8>,
-	hash_bytes: Vec<u8>,
+	key: ResyncQueueKey,
 	busy_set: BusySet,
 }
 
 impl BlockResyncManager {
 	pub(crate) fn new(db: &db::Db, system: &System) -> Self {
 		let queue = db
-			.open_tree("block_local_resync_queue")
+			.open_typed_tree("block_local_resync_queue")
 			.expect("Unable to open block_local_resync_queue tree");
 
 		let errors = db
-			.open_tree("block_local_resync_errors")
+			.open_typed_tree("block_local_resync_errors")
 			.expect("Unable to open block_local_resync_errors tree");
 
 		let persister = PersisterShared::new(&system.metadata_dir, "resync_cfg");
@@ -116,11 +158,10 @@ impl BlockResyncManager {
 	/// Clear the error counter for a block and put it in queue immediately
 	pub fn clear_backoff(&self, hash: &Hash) -> Result<(), Error> {
 		let now = now_msec();
-		if let Some(ec) = self.errors.get(hash)? {
-			let mut ec = ErrorCounter::decode(&ec);
+		if let Some(mut ec) = self.errors.get(hash)? {
 			if ec.errors > 0 {
 				ec.last_try = now - ec.delay_msec();
-				self.errors.insert(hash, ec.encode())?;
+				self.errors.insert(hash, &ec)?;
 				self.put_to_resync_at(hash, now)?;
 				return Ok(());
 			}
@@ -251,23 +292,21 @@ impl BlockResyncManager {
 
 	pub(crate) fn put_to_resync_at(&self, hash: &Hash, when: u64) -> db::Result<()> {
 		trace!("Put resync_queue: {} {:?}", when, hash);
-		let mut key = u64::to_be_bytes(when).to_vec();
-		key.extend(hash.as_ref());
-		self.queue.insert(key, hash.as_ref())?;
+		let qkey = ResyncQueueKey { when, hash: *hash };
+		self.queue.insert(&qkey, hash)?;
 		self.notify.notify_waiters();
 		Ok(())
 	}
 
 	async fn resync_iter(&self, manager: &BlockManager) -> Result<ResyncIterResult, db::Error> {
 		if let Some(block) = self.get_block_to_resync()? {
-			let time_msec = u64::from_be_bytes(block.time_bytes[0..8].try_into().unwrap());
+			let time_msec = block.key.when;
 			let now = now_msec();
 
 			if now >= time_msec {
-				let hash = Hash::try_from(&block.hash_bytes[..]).unwrap();
+				let hash = block.key.hash;
 
-				if let Some(ec) = self.errors.get(hash.as_slice())? {
-					let ec = ErrorCounter::decode(&ec);
+				if let Some(ec) = self.errors.get(&hash)? {
 					if now < ec.next_try() {
 						// if next retry after an error is not yet,
 						// don't do resync and return early, but still
@@ -277,7 +316,7 @@ impl BlockResyncManager {
 						// is not removing the one we added just above
 						// (we want to do the remove after the insert to ensure
 						// that the item is not lost if we crash in-between)
-						self.queue.remove(&block.time_bytes)?;
+						self.queue.remove(&block.key)?;
 						return Ok(ResyncIterResult::BusyDidNothing);
 					}
 				}
@@ -307,21 +346,21 @@ impl BlockResyncManager {
 					manager.metrics.resync_error_counter.add(1);
 					error!("Error when resyncing {:?}: {}", hash, e);
 
-					let err_counter = match self.errors.get(hash.as_slice())? {
-						Some(ec) => ErrorCounter::decode(&ec).add1(now + 1),
+					let err_counter = match self.errors.get(&hash)? {
+						Some(ec) => ec.add1(now + 1),
 						None => ErrorCounter::new(now + 1),
 					};
 
-					self.errors.insert(hash.as_slice(), err_counter.encode())?;
+					self.errors.insert(&hash, &err_counter)?;
 
 					self.put_to_resync_at(&hash, err_counter.next_try())?;
 					// err_counter.next_try() >= now + 1 > now,
 					// the entry we remove from the queue is not
 					// the entry we inserted with put_to_resync_at
-					self.queue.remove(&block.time_bytes)?;
+					self.queue.remove(&block.key)?;
 				} else {
-					self.errors.remove(hash.as_slice())?;
-					self.queue.remove(&block.time_bytes)?;
+					self.errors.remove(&hash)?;
+					self.queue.remove(&block.key)?;
 				}
 
 				Ok(ResyncIterResult::BusyDidSomething)
@@ -344,12 +383,11 @@ impl BlockResyncManager {
 	fn get_block_to_resync(&self) -> Result<Option<BusyBlock>, db::Error> {
 		let mut busy = self.busy_set.lock().unwrap();
 		for it in self.queue.iter()? {
-			let (time_bytes, hash_bytes) = it?;
-			if !busy.contains(&time_bytes) {
-				busy.insert(time_bytes.clone());
+			let (key, _) = it?;
+			if !busy.contains(&key) {
+				busy.insert(key);
 				return Ok(Some(BusyBlock {
-					time_bytes,
-					hash_bytes,
+					key,
 					busy_set: self.busy_set.clone(),
 				}));
 			}
@@ -506,7 +544,7 @@ impl BlockResyncManager {
 impl Drop for BusyBlock {
 	fn drop(&mut self) {
 		let mut busy = self.busy_set.lock().unwrap();
-		busy.remove(&self.time_bytes);
+		busy.remove(&self.key);
 	}
 }
 
@@ -602,6 +640,7 @@ impl Worker for ResyncWorker {
 
 /// Counts the number of errors when resyncing a block,
 /// and the time of the last try.
+///
 /// Used to implement exponential backoff.
 #[derive(Clone, Copy, Debug)]
 pub(crate) struct ErrorCounter {
@@ -609,6 +648,31 @@ pub(crate) struct ErrorCounter {
 	pub(crate) last_try: u64,
 }
 
+impl db::DbBytes for ErrorCounter {
+	fn encode(&self) -> Vec<u8> {
+		let mut v = Vec::with_capacity(16);
+		v.extend_from_slice(&u64::to_be_bytes(self.errors));
+		v.extend_from_slice(&u64::to_be_bytes(self.last_try));
+		v
+	}
+
+	fn decode(bytes: &[u8]) -> std::result::Result<Self, db::DecodeError> {
+		if bytes.len() != 16 {
+			return Err(db::DecodeError(
+				format!(
+					"invalid error counter: expected 16 bytes, got {}",
+					bytes.len()
+				)
+				.into(),
+			));
+		}
+		Ok(Self {
+			errors: u64::from_be_bytes(bytes[..8].try_into().unwrap()),
+			last_try: u64::from_be_bytes(bytes[8..].try_into().unwrap()),
+		})
+	}
+}
+
 impl ErrorCounter {
 	fn new(now: u64) -> Self {
 		Self {
@@ -617,21 +681,6 @@ impl ErrorCounter {
 		}
 	}
 
-	pub(crate) fn decode(data: &[u8]) -> 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()),
-		}
-	}
-
-	fn encode(&self) -> Vec<u8> {
-		[
-			u64::to_be_bytes(self.errors),
-			u64::to_be_bytes(self.last_try),
-		]
-		.concat()
-	}
-
 	fn add1(self, now: u64) -> Self {
 		Self {
 			errors: self.errors + 1,

src/util/data.rs

@@ -155,6 +155,21 @@ pub fn gen_uuid() -> Uuid {
 	rand::rng().random::<[u8; 32]>().into()
 }
 
+impl garage_db::DbBytes for FixedBytes32 {
+	fn encode(&self) -> Vec<u8> {
+		self.0.into()
+	}
+	fn decode(bytes: &[u8]) -> std::result::Result<Self, garage_db::DecodeError> {
+		Self::try_from(bytes).ok_or_else(|| {
+			garage_db::DecodeError(
+				format!("invalid hash: expected 32 bytes, got {}", bytes.len()).into(),
+			)
+		})
+	}
+}
+
+impl garage_db::DbOrdKey for FixedBytes32 {}
+
 #[cfg(test)]
 mod test {
 	use super::*;

src/util/error.rs

@@ -79,6 +79,18 @@ impl Error {
 	}
 }
 
+impl From<garage_db::DbError> for Error {
+	fn from(e: garage_db::DbError) -> Error {
+		Error::Db(e.into())
+	}
+}
+
+impl From<garage_db::DecodeError> for Error {
+	fn from(e: garage_db::DecodeError) -> Error {
+		Error::Db(e.into())
+	}
+}
+
 impl From<garage_db::TxError<Error>> for Error {
 	fn from(e: garage_db::TxError<Error>) -> Error {
 		match e {