diff --git a/src/api/admin/repair.rs b/src/api/admin/repair.rs index 1d5665d1..de62c75a 100644 --- a/src/api/admin/repair.rs +++ b/src/api/admin/repair.rs @@ -360,14 +360,7 @@ impl Worker for BlockRcRepair { _must_exit: &mut watch::Receiver, ) -> Result { 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 diff --git a/src/block/manager.rs b/src/block/manager.rs index ce14f625..c7a4c317 100644 --- a/src/block/manager.rs +++ b/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, diff --git a/src/block/rc.rs b/src/block/rc.rs index 881df68b..d0dc49d4 100644 --- a/src/block/rc.rs +++ b/src/block/rc.rs @@ -1,4 +1,5 @@ use std::convert::TryInto; +use std::num::NonZeroU64; use arc_swap::ArcSwapOption; @@ -14,12 +15,12 @@ pub type CalculateRefcount = Box db::TxResult + Send + Sync>; pub struct BlockRc { - pub rc_table: db::Tree, + pub(crate) rc_table: db::TypedTree, pub(crate) recalc_rc: ArcSwapOption>, } impl BlockRc { - pub(crate) fn new(rc: db::Tree) -> Self { + pub(crate) fn new(rc: db::TypedTree) -> Self { Self { rc_table: rc, recalc_rc: ArcSwapOption::new(None), @@ -33,11 +34,8 @@ impl BlockRc { tx: &mut db::Transaction, hash: &Hash, ) -> db::TxOpResult { - 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 = RcState(self.rc_table.tx_get(tx, hash)?); + self.rc_table.tx_insert(tx, hash, &old_rc.increment())?; Ok(old_rc.is_zero()) } @@ -48,17 +46,27 @@ impl BlockRc { tx: &mut db::Transaction, hash: &Hash, ) -> db::TxOpResult { - 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 = RcState(self.rc_table.tx_get(tx, hash)?).decrement(); + match &new_rc.0 { + None => self.rc_table.tx_remove(tx, hash)?, + Some(rc) => self.rc_table.tx_insert(tx, hash, rc)?, } - Ok(matches!(new_rc, RcEntry::Deletable { .. })) + Ok(matches!(new_rc.0, Some(RcEntry::Deletable { .. }))) } - /// Read a block's reference count - pub(crate) fn get_block_rc(&self, hash: &Hash) -> Result { - Ok(RcEntry::parse_opt(self.rc_table.get(hash.as_ref())?)) + /// Read a block's reference counting state + pub(crate) fn get_block_rc(&self, hash: &Hash) -> Result { + Ok(RcState(self.rc_table.get(hash)?)) + } + + /// Return the first hash stored in the RC table at or after `cursor` + pub fn get_first_hash_from(&self, cursor: Hash) -> Result, 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 +74,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)?; + if let Some(RcEntry::Deletable { at_time }) = rcval { + if now > at_time { + self.rc_table.tx_remove(tx, hash)?; } - _ => (), } Ok(()) })?; @@ -91,28 +98,25 @@ 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_count = RcState(self.rc_table.tx_get(tx, hash)?).as_u64(); trace!( "Block RC for {:?}: stored={}, calculated={}", hash, - old_rc.as_u64(), + old_count, cnt ); - if cnt as u64 != old_rc.as_u64() { + if cnt as u64 != old_count { warn!( "Fixing inconsistent block RC for {:?}: was {}, should be {}", - hash, - old_rc.as_u64(), - cnt + hash, old_count, cnt ); - let new_rc = if cnt > 0 { - RcEntry::Present { count: cnt as u64 } - } else { - RcEntry::Deletable { + let new_rc = match NonZeroU64::new(cnt as u64) { + Some(count) => RcEntry::Present { count }, + None => RcEntry::Deletable { 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,13 +135,47 @@ impl BlockRc { } } -/// Describes the state of the reference counter for a block +impl db::DbBytes for RcEntry { + fn encode(&self) -> Vec { + match self { + RcEntry::Present { count } => u64::to_be_bytes(count.get()).to_vec(), + RcEntry::Deletable { at_time } => { + [u64::to_be_bytes(0), u64::to_be_bytes(*at_time)].concat() + } + } + } + + fn decode(bytes: &[u8]) -> std::result::Result { + if bytes.len() == 8 { + let count = NonZeroU64::new(u64::from_be_bytes(bytes.try_into().unwrap())) + .ok_or(db::DecodeError("invalid RC entry: zero count".into()))?; + Ok(RcEntry::Present { count }) + } else if bytes.len() == 16 { + Ok(RcEntry::Deletable { + at_time: u64::from_be_bytes(bytes[8..16].try_into().unwrap()), + }) + } else { + Err(db::DecodeError( + format!( + "invalid RC entry: expected 8 or 16 bytes, got {}", + bytes.len() + ) + .into(), + )) + } + } +} + +/// A block's entry in the RC table. +/// +/// A block with zero references and no pending deletion has no entry +/// in the RC table at all: see [`RcState`]. #[derive(Clone, Copy, Debug)] pub(crate) enum RcEntry { - /// Present: the block has `count` references, with `count` > 0. + /// Present: the block has `count` references. /// /// This is stored as `u64::to_be_bytes(count)` - Present { count: u64 }, + Present { count: NonZeroU64 }, /// Deletable: the block has zero references, and can be deleted /// once time (returned by `now_msec`) is larger than `at_time` @@ -147,72 +185,39 @@ pub(crate) enum RcEntry { /// (this allows for the data format to be backwards compatible with /// previous Garage versions that didn't have this intermediate state) Deletable { at_time: u64 }, - - /// Absent: the block has zero references, and can be deleted - /// immediately - Absent, } -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 - ) - } - } +/// Describes the state of the reference counter for a block: the block's +/// entry in the RC table, or `None` if it has none, meaning the block has +/// zero references and can be deleted immediately. +#[derive(Clone, Copy, Debug)] +pub(crate) struct RcState(Option); - fn parse_opt>(bytes: Option) -> Self { - bytes - .map(|b| Self::parse(b.as_ref())) - .unwrap_or(Self::Absent) - } - - fn serialize(self) -> Option> { - 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, - _ => 0, +impl RcState { + /// The new RC table entry after a reference is taken on the block + fn increment(&self) -> RcEntry { + let count = match self.0 { + Some(RcEntry::Present { count }) => count.saturating_add(1), + _ => NonZeroU64::new(1).unwrap(), }; - RcEntry::Present { - count: old_count + 1, - } + RcEntry::Present { count } } - fn decrement(self) -> Self { - match self { - RcEntry::Present { count } => { - if count > 1 { - RcEntry::Present { count: count - 1 } - } else { - RcEntry::Deletable { - at_time: now_msec() + BLOCK_GC_DELAY.as_millis() as u64, - } - } - } - del => del, - } + /// The new state after a reference to the block is dropped + fn decrement(&self) -> Self { + RcState(match self.0 { + Some(RcEntry::Present { count }) => Some(match NonZeroU64::new(count.get() - 1) { + Some(count) => RcEntry::Present { count }, + None => RcEntry::Deletable { + at_time: now_msec() + BLOCK_GC_DELAY.as_millis() as u64, + }, + }), + unchanged => unchanged, + }) } pub(crate) fn is_zero(&self) -> bool { - matches!(self, RcEntry::Deletable { .. } | RcEntry::Absent) + matches!(self.0, None | Some(RcEntry::Deletable { .. })) } pub(crate) fn is_nonzero(&self) -> bool { @@ -220,10 +225,10 @@ impl RcEntry { } pub(crate) fn is_deletable(&self) -> bool { - match self { - RcEntry::Present { .. } => false, - RcEntry::Deletable { at_time } => now_msec() > *at_time, - RcEntry::Absent => true, + match self.0 { + Some(RcEntry::Present { .. }) => false, + Some(RcEntry::Deletable { at_time }) => now_msec() > at_time, + None => true, } } @@ -232,8 +237,8 @@ impl RcEntry { } pub(crate) fn as_u64(&self) -> u64 { - match self { - RcEntry::Present { count } => *count, + match self.0 { + Some(RcEntry::Present { count }) => count.get(), _ => 0, } } diff --git a/src/block/repair.rs b/src/block/repair.rs index 9499d4ab..6c443a1c 100644 --- a/src/block/repair.rs +++ b/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; diff --git a/src/block/resync.rs b/src/block/resync.rs index 003c1172..f89716ad 100644 --- a/src/block/resync.rs +++ b/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, pub(crate) notify: Arc, - pub(crate) errors: db::Tree, + pub(crate) errors: db::TypedTree, busy_set: BusySet, persister: PersisterShared, } +/// 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 { + 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 { + 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>>>; +type BusySet = Arc>>; struct BusyBlock { - time_bytes: Vec, - hash_bytes: Vec, + 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 { 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, 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); } } @@ -516,6 +554,7 @@ pub(crate) struct ResyncWorker { tranquilizer: Tranquilizer, next_delay: Duration, persister: PersisterShared, + had_decode_error: bool, } impl ResyncWorker { @@ -527,6 +566,7 @@ impl ResyncWorker { tranquilizer: Tranquilizer::new(30), next_delay: Duration::from_secs(10), persister, + had_decode_error: false, } } } @@ -574,6 +614,15 @@ impl Worker for ResyncWorker { self.next_delay = delay; Ok(WorkerState::Idle) } + Err(db::Error::Decode(e)) => { + // We give it one second chance in the very unlikely case that the bytes would somehow + // have been corrupted during read and that a new read would lead to a correct decoding. + if self.had_decode_error { + panic!("An error has happened when decoding something stored in the local k/v store: {}.", e); + } + self.had_decode_error = true; + Ok(WorkerState::Busy) + } Err(e) => { // The errors that we have here are only db errors // We don't really know how to handle them so just ¯\_(ツ)_/¯ @@ -581,6 +630,7 @@ impl Worker for ResyncWorker { // if it does there is not much we can do -- TODO should we just panic?) // Here we just give the error to the worker manager, // it will print it to the logs and increment a counter + self.had_decode_error = false; Err(e.into()) } } @@ -602,6 +652,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 +660,31 @@ pub(crate) struct ErrorCounter { pub(crate) last_try: u64, } +impl db::DbBytes for ErrorCounter { + fn encode(&self) -> Vec { + 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 { + 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 +693,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 { - [ - 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, diff --git a/src/db/fjall_adapter.rs b/src/db/fjall_adapter.rs index 9e9efe9f..aa8bfeab 100644 --- a/src/db/fjall_adapter.rs +++ b/src/db/fjall_adapter.rs @@ -12,7 +12,7 @@ use fjall::{ use crate::{ open::{Engine, OpenOpt}, - Db, Error, IDb, ITx, ITxFn, OnCommit, Result, TxError, TxFnResult, TxOpError, TxOpResult, + Db, DbError, DbResult, Error, IDb, ITx, ITxFn, OnCommit, TxError, TxFnResult, TxOpError, TxResult, TxValueIter, Value, ValueIter, }; @@ -20,10 +20,10 @@ pub use fjall; // -- -pub(crate) fn open_db(path: &Path, opt: &OpenOpt) -> Result { +pub(crate) fn open_db(path: &Path, opt: &OpenOpt) -> DbResult { info!("Opening Fjall database at: {}", path.display()); if opt.fsync { - return Err(Error( + return Err(DbError( "metadata_fsync is not supported with the Fjall database engine".into(), )); } @@ -37,21 +37,27 @@ pub(crate) fn open_db(path: &Path, opt: &OpenOpt) -> Result { // -- err -impl From for Error { - fn from(e: fjall::Error) -> Error { - Error(format!("fjall: {}", e).into()) +impl From for DbError { + fn from(e: fjall::Error) -> DbError { + DbError(format!("fjall: {}", e).into()) } } -impl From for Error { - fn from(e: fjall::LsmError) -> Error { - Error(format!("fjall lsm_tree: {}", e).into()) +impl From for DbError { + fn from(e: fjall::LsmError) -> DbError { + DbError(format!("fjall lsm_tree: {}", e).into()) + } +} + +impl From for Error { + fn from(e: fjall::Error) -> Error { + Error::Db(DbError::from(e)) } } impl From for TxOpError { fn from(e: fjall::Error) -> TxOpError { - TxOpError(e.into()) + DbError::from(e).into() } } @@ -76,11 +82,11 @@ impl FjallDb { fn get_tree( &self, i: usize, - ) -> Result> { + ) -> DbResult> { RwLockReadGuard::try_map(self.trees.read(), |trees: &Vec<_>| { trees.get(i).map(|tup| &tup.1) }) - .map_err(|_| Error("invalid tree id".into())) + .map_err(|_| DbError("invalid tree id".into())) } } @@ -89,7 +95,7 @@ impl IDb for FjallDb { "Fjall (EXPERIMENTAL!)".into() } - fn open_tree(&self, name: &str) -> Result { + fn open_tree(&self, name: &str) -> DbResult { let mut trees = self.trees.write(); let safe_name = encode_name(name)?; if let Some(i) = trees.iter().position(|(name, _)| *name == safe_name) { @@ -104,15 +110,15 @@ impl IDb for FjallDb { } } - fn list_trees(&self) -> Result> { + fn list_trees(&self) -> DbResult> { self.keyspace .list_partitions() .iter() .map(|n| decode_name(n)) - .collect::>>() + .collect::>>() } - fn snapshot(&self, base_path: &Path) -> Result<()> { + fn snapshot(&self, base_path: &Path) -> DbResult<()> { std::fs::create_dir_all(base_path)?; let path = Engine::Fjall.db_path(base_path); @@ -138,7 +144,7 @@ impl IDb for FjallDb { // ---- - fn get(&self, tree_idx: usize, key: &[u8]) -> Result> { + fn get(&self, tree_idx: usize, key: &[u8]) -> DbResult> { let tree = self.get_tree(tree_idx)?; let tx = self.keyspace.read_tx(); let val = tx.get(&tree, key)?; @@ -148,17 +154,17 @@ impl IDb for FjallDb { } } - fn approximate_len(&self, tree_idx: usize) -> Result { + fn approximate_len(&self, tree_idx: usize) -> DbResult { let tree = self.get_tree(tree_idx)?; Ok(tree.approximate_len()) } - fn is_empty(&self, tree_idx: usize) -> Result { + fn is_empty(&self, tree_idx: usize) -> DbResult { let tree = self.get_tree(tree_idx)?; let tx = self.keyspace.read_tx(); Ok(tx.is_empty(&tree)?) } - fn insert(&self, tree_idx: usize, key: &[u8], value: &[u8]) -> Result<()> { + fn insert(&self, tree_idx: usize, key: &[u8], value: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree_idx)?; let mut tx = self.keyspace.write_tx(); tx.insert(&tree, key, value); @@ -166,7 +172,7 @@ impl IDb for FjallDb { Ok(()) } - fn remove(&self, tree_idx: usize, key: &[u8]) -> Result<()> { + fn remove(&self, tree_idx: usize, key: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree_idx)?; let mut tx = self.keyspace.write_tx(); tx.remove(&tree, key); @@ -174,11 +180,11 @@ impl IDb for FjallDb { Ok(()) } - fn clear(&self, tree_idx: usize) -> Result<()> { + fn clear(&self, tree_idx: usize) -> DbResult<()> { let mut trees = self.trees.write(); if tree_idx >= trees.len() { - return Err(Error("invalid tree id".into())); + return Err(DbError("invalid tree id".into())); } let (name, tree) = trees.remove(tree_idx); @@ -191,13 +197,13 @@ impl IDb for FjallDb { Ok(()) } - fn iter(&self, tree_idx: usize) -> Result> { + fn iter(&self, tree_idx: usize) -> DbResult> { 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> { + fn iter_rev(&self, tree_idx: usize) -> DbResult> { let tree = self.get_tree(tree_idx)?; let tx = self.keyspace.read_tx(); Ok(Box::new(tx.iter(&tree).rev().map(iterator_remap))) @@ -208,7 +214,7 @@ impl IDb for FjallDb { tree_idx: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> Result> { + ) -> DbResult> { let tree = self.get_tree(tree_idx)?; let tx = self.keyspace.read_tx(); Ok(Box::new( @@ -221,7 +227,7 @@ impl IDb for FjallDb { tree_idx: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> Result> { + ) -> DbResult> { let tree = self.get_tree(tree_idx)?; let tx = self.keyspace.read_tx(); Ok(Box::new( @@ -252,9 +258,9 @@ impl IDb for FjallDb { } TxFnResult::DbErr => { tx.tx.rollback(); - Err(TxError::Db(Error( - "(this message will be discarded)".into(), - ))) + Err(TxError::Db( + DbError("(this message will be discarded)".into()).into(), + )) } } } @@ -268,47 +274,47 @@ struct FjallTx<'a> { } impl<'a> FjallTx<'a> { - fn get_tree(&self, i: usize) -> TxOpResult<&TransactionalPartitionHandle> { + fn get_tree(&self, i: usize) -> DbResult<&TransactionalPartitionHandle> { self.trees.get(i).map(|tup| &tup.1).ok_or_else(|| { - TxOpError(Error( + DbError( "invalid tree id (it might have been opened after the transaction started)".into(), - )) + ) }) } } impl<'a> ITx for FjallTx<'a> { - fn get(&self, tree_idx: usize, key: &[u8]) -> TxOpResult> { + fn get(&self, tree_idx: usize, key: &[u8]) -> DbResult> { 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 { + fn len(&self, tree_idx: usize) -> DbResult { let tree = self.get_tree(tree_idx)?; Ok(self.tx.len(tree)?) } - fn insert(&mut self, tree_idx: usize, key: &[u8], value: &[u8]) -> TxOpResult<()> { + fn insert(&mut self, tree_idx: usize, key: &[u8], value: &[u8]) -> DbResult<()> { 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<()> { + fn remove(&mut self, tree_idx: usize, key: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree_idx)?.clone(); self.tx.remove(&tree, key); Ok(()) } - fn clear(&mut self, _tree_idx: usize) -> TxOpResult<()> { + fn clear(&mut self, _tree_idx: usize) -> DbResult<()> { unimplemented!("LSM tree clearing in cross-partition transaction is not supported") } - fn iter(&self, tree_idx: usize) -> TxOpResult> { + fn iter(&self, tree_idx: usize) -> DbResult> { 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> { + fn iter_rev(&self, tree_idx: usize) -> DbResult> { let tree = self.get_tree(tree_idx)?.clone(); Ok(Box::new(self.tx.iter(&tree).rev().map(iterator_remap_tx))) } @@ -318,7 +324,7 @@ impl<'a> ITx for FjallTx<'a> { tree_idx: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> TxOpResult> { + ) -> DbResult> { let tree = self.get_tree(tree_idx)?; let low = clone_bound(low); let high = clone_bound(high); @@ -333,7 +339,7 @@ impl<'a> ITx for FjallTx<'a> { tree_idx: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> TxOpResult> { + ) -> DbResult> { let tree = self.get_tree(tree_idx)?; let low = clone_bound(low); let high = clone_bound(high); @@ -348,14 +354,14 @@ impl<'a> ITx for FjallTx<'a> { // -- maps fjall's (k, v) to ours -fn iterator_remap(r: fjall::Result<(fjall::Slice, fjall::Slice)>) -> Result<(Value, Value)> { +fn iterator_remap(r: fjall::Result<(fjall::Slice, fjall::Slice)>) -> DbResult<(Value, Value)> { r.map(|(k, v)| (k.to_vec(), v.to_vec())) - .map_err(|e| e.into()) + .map_err(DbError::from) } -fn iterator_remap_tx(r: fjall::Result<(fjall::Slice, fjall::Slice)>) -> TxOpResult<(Value, Value)> { +fn iterator_remap_tx(r: fjall::Result<(fjall::Slice, fjall::Slice)>) -> DbResult<(Value, Value)> { r.map(|(k, v)| (k.to_vec(), v.to_vec())) - .map_err(|e| e.into()) + .map_err(DbError::from) } // -- utils to deal with Garage's tightness on Bound lifetimes @@ -378,7 +384,7 @@ fn clone_bound(bound: Bound<&[u8]>) -> ByteVecBound { // -- utils to encode table names -- -fn encode_name(s: &str) -> Result { +fn encode_name(s: &str) -> DbResult { let base = 'A' as u32; let mut ret = String::with_capacity(s.len() + 10); @@ -392,7 +398,7 @@ fn encode_name(s: &str) -> Result { ret.push(char::from_u32(base + c_hi).unwrap()); ret.push(char::from_u32(base + c_lo).unwrap()); } else { - return Err(Error( + return Err(DbError( format!("table name {} could not be safely encoded", s).into(), )); } @@ -400,10 +406,10 @@ fn encode_name(s: &str) -> Result { Ok(ret) } -fn decode_name(s: &str) -> Result { +fn decode_name(s: &str) -> DbResult { use std::convert::TryFrom; - let errfn = || Error(format!("encoded table name {} is invalid", s).into()); + let errfn = || DbError(format!("encoded table name {} is invalid", s).into()); let c_map = |c: char| { let c = c as u32; let base = 'A' as u32; diff --git a/src/db/lib.rs b/src/db/lib.rs index bb04d6cc..936d829b 100644 --- a/src/db/lib.rs +++ b/src/db/lib.rs @@ -9,6 +9,7 @@ pub mod lmdb_adapter; pub mod sqlite_adapter; pub mod open; +pub mod typed; #[cfg(test)] pub mod test; @@ -23,6 +24,7 @@ use std::sync::Arc; use thiserror::Error; pub use open::*; +pub use typed::{DbBytes, DbOrdKey, TypedIter, TypedTree, TypedTxIter}; pub(crate) type OnCommit = Vec>; @@ -38,21 +40,34 @@ pub struct Transaction<'a> { pub struct Tree(Arc, usize); pub type Value = Vec; -pub type ValueIter<'a> = Box> + 'a>; -pub type TxValueIter<'a> = Box> + 'a>; +pub type ValueIter<'a> = Box> + 'a>; +pub type TxValueIter<'a> = Box> + 'a>; // ---- #[derive(Debug, Error)] -#[error("{0}")] -pub struct Error(pub Cow<'static, str>); +#[error("database error: {0}")] +pub struct DbError(pub Cow<'static, str>); -impl From for Error { - fn from(e: std::io::Error) -> Error { - Error(format!("IO: {}", e).into()) +#[derive(Debug, Error)] +#[error("decode error: {0}")] +pub struct DecodeError(pub Cow<'static, str>); + +#[derive(Debug, Error)] +pub enum Error { + #[error(transparent)] + Db(#[from] DbError), + #[error(transparent)] + Decode(#[from] DecodeError), +} + +impl From for DbError { + fn from(e: std::io::Error) -> DbError { + DbError(format!("IO: {}", e).into()) } } +pub type DbResult = std::result::Result; pub type Result = std::result::Result; #[derive(Debug, Error)] @@ -60,6 +75,18 @@ pub type Result = std::result::Result; pub struct TxOpError(pub(crate) Error); pub type TxOpResult = std::result::Result; +impl From for TxOpError { + fn from(e: DbError) -> TxOpError { + TxOpError(e.into()) + } +} + +impl From for TxOpError { + fn from(e: DecodeError) -> TxOpError { + TxOpError(e.into()) + } +} + #[derive(Debug)] pub enum TxError { Abort(E), @@ -73,6 +100,12 @@ impl From for TxError { } } +impl From for TxError { + fn from(e: DbError) -> TxError { + TxError::Db(e.into()) + } +} + pub fn unabort(res: TxResult) -> TxOpResult> { match res { Ok(v) => Ok(Ok(v)), @@ -88,12 +121,12 @@ impl Db { self.0.engine() } - pub fn open_tree>(&self, name: S) -> Result { + pub fn open_tree>(&self, name: S) -> DbResult { let tree_id = self.0.open_tree(name.as_ref())?; Ok(Tree(self.0.clone(), tree_id)) } - pub fn list_trees(&self) -> Result> { + pub fn list_trees(&self) -> DbResult> { self.0.list_trees() } @@ -147,27 +180,28 @@ impl Db { } } - pub fn snapshot(&self, path: &Path) -> Result<()> { + pub fn snapshot(&self, path: &Path) -> DbResult<()> { self.0.snapshot(path) } pub fn import(&self, other: &Db) -> Result<()> { let existing_trees = self.list_trees()?; if !existing_trees.is_empty() { - return Err(Error( + return Err(DbError( format!( "destination database already contains data: {:?}", existing_trees ) .into(), - )); + ) + .into()); } let tree_names = other.list_trees()?; for name in tree_names { let tree = self.open_tree(&name)?; if !tree.is_empty()? { - return Err(Error(format!("tree {} already contains data", name).into())); + return Err(DbError(format!("tree {} already contains data", name).into()).into()); } let ex_tree = other.open_tree(&name)?; @@ -186,7 +220,7 @@ impl Db { }); let total = match tx_res { Err(TxError::Db(e)) => return Err(e), - Err(TxError::Abort(e)) => return Err(e), + Err(TxError::Abort(e)) => return Err(e.into()), Ok(x) => x, }; @@ -204,24 +238,24 @@ impl Tree { } #[inline] - pub fn get>(&self, key: T) -> Result> { + pub fn get>(&self, key: T) -> DbResult> { self.0.get(self.1, key.as_ref()) } #[inline] - pub fn approximate_len(&self) -> Result { + pub fn approximate_len(&self) -> DbResult { self.0.approximate_len(self.1) } #[inline] - pub fn is_empty(&self) -> Result { + pub fn is_empty(&self) -> DbResult { self.0.is_empty(self.1) } #[inline] - pub fn first(&self) -> Result> { + pub fn first(&self) -> DbResult> { self.iter()?.next().transpose() } #[inline] - pub fn get_gt>(&self, from: T) -> Result> { + pub fn get_gt>(&self, from: T) -> DbResult> { if from.as_ref().is_empty() { self.iter()?.next().transpose() } else { @@ -233,31 +267,31 @@ impl Tree { /// Returns the old value if there was one #[inline] - pub fn insert, U: AsRef<[u8]>>(&self, key: T, value: U) -> Result<()> { + pub fn insert, U: AsRef<[u8]>>(&self, key: T, value: U) -> DbResult<()> { self.0.insert(self.1, key.as_ref(), value.as_ref()) } /// Returns the old value if there was one #[inline] - pub fn remove>(&self, key: T) -> Result<()> { + pub fn remove>(&self, key: T) -> DbResult<()> { self.0.remove(self.1, key.as_ref()) } /// Clears all values from the tree #[inline] - pub fn clear(&self) -> Result<()> { + pub fn clear(&self) -> DbResult<()> { self.0.clear(self.1) } #[inline] - pub fn iter(&self) -> Result> { + pub fn iter(&self) -> DbResult> { self.0.iter(self.1) } #[inline] - pub fn iter_rev(&self) -> Result> { + pub fn iter_rev(&self) -> DbResult> { self.0.iter_rev(self.1) } #[inline] - pub fn range(&self, range: R) -> Result> + pub fn range(&self, range: R) -> DbResult> where K: AsRef<[u8]>, R: RangeBounds, @@ -267,7 +301,7 @@ impl Tree { self.0.range(self.1, get_bound(sb), get_bound(eb)) } #[inline] - pub fn range_rev(&self, range: R) -> Result> + pub fn range_rev(&self, range: R) -> DbResult> where K: AsRef<[u8]>, R: RangeBounds, @@ -282,11 +316,11 @@ impl Tree { impl<'a> Transaction<'a> { #[inline] pub fn get>(&self, tree: &Tree, key: T) -> TxOpResult> { - self.tx.get(tree.1, key.as_ref()) + self.tx.get(tree.1, key.as_ref()).map_err(Into::into) } #[inline] pub fn len(&self, tree: &Tree) -> TxOpResult { - self.tx.len(tree.1) + self.tx.len(tree.1).map_err(Into::into) } /// Returns the old value if there was one @@ -297,26 +331,28 @@ impl<'a> Transaction<'a> { key: T, value: U, ) -> TxOpResult<()> { - self.tx.insert(tree.1, key.as_ref(), value.as_ref()) + self.tx + .insert(tree.1, key.as_ref(), value.as_ref()) + .map_err(Into::into) } /// Returns the old value if there was one #[inline] pub fn remove>(&mut self, tree: &Tree, key: T) -> TxOpResult<()> { - self.tx.remove(tree.1, key.as_ref()) + self.tx.remove(tree.1, key.as_ref()).map_err(Into::into) } /// Clears all values in a tree #[inline] pub fn clear(&mut self, tree: &Tree) -> TxOpResult<()> { - self.tx.clear(tree.1) + self.tx.clear(tree.1).map_err(Into::into) } #[inline] pub fn iter(&self, tree: &Tree) -> TxOpResult> { - self.tx.iter(tree.1) + self.tx.iter(tree.1).map_err(Into::into) } #[inline] pub fn iter_rev(&self, tree: &Tree) -> TxOpResult> { - self.tx.iter_rev(tree.1) + self.tx.iter_rev(tree.1).map_err(Into::into) } #[inline] @@ -327,7 +363,9 @@ impl<'a> Transaction<'a> { { let sb = range.start_bound(); let eb = range.end_bound(); - self.tx.range(tree.1, get_bound(sb), get_bound(eb)) + self.tx + .range(tree.1, get_bound(sb), get_bound(eb)) + .map_err(Into::into) } #[inline] pub fn range_rev(&self, tree: &Tree, range: R) -> TxOpResult> @@ -337,7 +375,9 @@ impl<'a> Transaction<'a> { { let sb = range.start_bound(); let eb = range.end_bound(); - self.tx.range_rev(tree.1, get_bound(sb), get_bound(eb)) + self.tx + .range_rev(tree.1, get_bound(sb), get_bound(eb)) + .map_err(Into::into) } #[inline] @@ -350,60 +390,60 @@ impl<'a> Transaction<'a> { pub(crate) trait IDb: Send + Sync { fn engine(&self) -> String; - fn open_tree(&self, name: &str) -> Result; - fn list_trees(&self) -> Result>; - fn snapshot(&self, path: &Path) -> Result<()>; + fn open_tree(&self, name: &str) -> DbResult; + fn list_trees(&self) -> DbResult>; + fn snapshot(&self, path: &Path) -> DbResult<()>; - fn get(&self, tree: usize, key: &[u8]) -> Result>; - fn approximate_len(&self, tree: usize) -> Result; - fn is_empty(&self, tree: usize) -> Result; + fn get(&self, tree: usize, key: &[u8]) -> DbResult>; + fn approximate_len(&self, tree: usize) -> DbResult; + fn is_empty(&self, tree: usize) -> DbResult; - fn insert(&self, tree: usize, key: &[u8], value: &[u8]) -> Result<()>; - fn remove(&self, tree: usize, key: &[u8]) -> Result<()>; - fn clear(&self, tree: usize) -> Result<()>; + fn insert(&self, tree: usize, key: &[u8], value: &[u8]) -> DbResult<()>; + fn remove(&self, tree: usize, key: &[u8]) -> DbResult<()>; + fn clear(&self, tree: usize) -> DbResult<()>; - fn iter(&self, tree: usize) -> Result>; - fn iter_rev(&self, tree: usize) -> Result>; + fn iter(&self, tree: usize) -> DbResult>; + fn iter_rev(&self, tree: usize) -> DbResult>; fn range<'r>( &self, tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> Result>; + ) -> DbResult>; fn range_rev<'r>( &self, tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> Result>; + ) -> DbResult>; fn transaction(&self, f: &dyn ITxFn) -> TxResult; } pub(crate) trait ITx { - fn get(&self, tree: usize, key: &[u8]) -> TxOpResult>; - fn len(&self, tree: usize) -> TxOpResult; + fn get(&self, tree: usize, key: &[u8]) -> DbResult>; + fn len(&self, tree: usize) -> DbResult; - fn insert(&mut self, tree: usize, key: &[u8], value: &[u8]) -> TxOpResult<()>; - fn remove(&mut self, tree: usize, key: &[u8]) -> TxOpResult<()>; - fn clear(&mut self, tree: usize) -> TxOpResult<()>; + fn insert(&mut self, tree: usize, key: &[u8], value: &[u8]) -> DbResult<()>; + fn remove(&mut self, tree: usize, key: &[u8]) -> DbResult<()>; + fn clear(&mut self, tree: usize) -> DbResult<()>; - fn iter(&self, tree: usize) -> TxOpResult>; - fn iter_rev(&self, tree: usize) -> TxOpResult>; + fn iter(&self, tree: usize) -> DbResult>; + fn iter_rev(&self, tree: usize) -> DbResult>; fn range<'r>( &self, tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> TxOpResult>; + ) -> DbResult>; fn range_rev<'r>( &self, tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> TxOpResult>; + ) -> DbResult>; } pub(crate) trait ITxFn { diff --git a/src/db/lmdb_adapter.rs b/src/db/lmdb_adapter.rs index c5360d04..4c93f5c1 100644 --- a/src/db/lmdb_adapter.rs +++ b/src/db/lmdb_adapter.rs @@ -14,7 +14,7 @@ type Database = heed::Database; use crate::{ open::{Engine, OpenOpt}, - Db, Error, IDb, ITx, ITxFn, OnCommit, Result, TxError, TxFnResult, TxOpError, TxOpResult, + Db, DbError, DbResult, Error, IDb, ITx, ITxFn, OnCommit, TxError, TxFnResult, TxOpError, TxResult, TxValueIter, Value, ValueIter, }; @@ -22,10 +22,10 @@ pub use heed; // ---- top-level open function -pub(crate) fn open_db(path: &PathBuf, opt: &OpenOpt) -> Result { +pub(crate) fn open_db(path: &PathBuf, opt: &OpenOpt) -> DbResult { info!("Opening LMDB database at: {}", path.display()); if let Err(e) = std::fs::create_dir_all(path) { - return Err(Error( + return Err(DbError( format!("Unable to create LMDB data directory: {}", e).into(), )); } @@ -48,7 +48,7 @@ pub(crate) fn open_db(path: &PathBuf, opt: &OpenOpt) -> Result { env_builder.open(path) }; match open_res { - Err(heed::Error::Io(e)) if e.kind() == std::io::ErrorKind::OutOfMemory => Err(Error( + Err(heed::Error::Io(e)) if e.kind() == std::io::ErrorKind::OutOfMemory => Err(DbError( "OutOfMemory error while trying to open LMDB database. This can happen \ if your operating system is not allowing you to use sufficient virtual \ memory address space. Please check that no limit is set (ulimit -v). \ @@ -56,22 +56,28 @@ pub(crate) fn open_db(path: &PathBuf, opt: &OpenOpt) -> Result { On 32-bit machines, you should probably switch to another database engine." .into(), )), - Err(e) => Err(Error(format!("Cannot open LMDB database: {}", e).into())), + Err(e) => Err(DbError(format!("Cannot open LMDB database: {}", e).into())), Ok(db) => Ok(LmdbDb::init(db)), } } // -- err +impl From for DbError { + fn from(e: heed::Error) -> DbError { + DbError(format!("LMDB: {}", e).into()) + } +} + impl From for Error { fn from(e: heed::Error) -> Error { - Error(format!("LMDB: {}", e).into()) + DbError::from(e).into() } } impl From for TxOpError { fn from(e: heed::Error) -> TxOpError { - TxOpError(e.into()) + DbError::from(e).into() } } @@ -91,14 +97,14 @@ impl LmdbDb { Db(Arc::new(s)) } - fn get_tree(&self, i: usize) -> Result { + fn get_tree(&self, i: usize) -> DbResult { self.trees .read() .unwrap() .0 .get(i) .cloned() - .ok_or_else(|| Error("invalid tree id".into())) + .ok_or_else(|| DbError("invalid tree id".into())) } } @@ -107,7 +113,7 @@ impl IDb for LmdbDb { "LMDB (using Heed crate)".into() } - fn open_tree(&self, name: &str) -> Result { + fn open_tree(&self, name: &str) -> DbResult { let mut trees = self.trees.write().unwrap(); if let Some(i) = trees.1.get(name) { Ok(*i) @@ -122,7 +128,7 @@ impl IDb for LmdbDb { } } - fn list_trees(&self) -> Result> { + fn list_trees(&self) -> DbResult> { let rtxn = self.db.read_txn()?; let tree0 = match self .db @@ -153,7 +159,7 @@ impl IDb for LmdbDb { Ok(ret2) } - fn snapshot(&self, base_path: &Path) -> Result<()> { + fn snapshot(&self, base_path: &Path) -> DbResult<()> { std::fs::create_dir_all(base_path)?; let path = Engine::Lmdb.db_path(base_path); self.db @@ -163,7 +169,7 @@ impl IDb for LmdbDb { // ---- - fn get(&self, tree: usize, key: &[u8]) -> Result> { + fn get(&self, tree: usize, key: &[u8]) -> DbResult> { let tree = self.get_tree(tree)?; let tx = self.db.read_txn()?; @@ -174,18 +180,18 @@ impl IDb for LmdbDb { } } - fn approximate_len(&self, tree: usize) -> Result { + fn approximate_len(&self, tree: usize) -> DbResult { let tree = self.get_tree(tree)?; let tx = self.db.read_txn()?; Ok(tree.len(&tx)?.try_into().unwrap()) } - fn is_empty(&self, tree: usize) -> Result { + fn is_empty(&self, tree: usize) -> DbResult { let tree = self.get_tree(tree)?; let tx = self.db.read_txn()?; Ok(tree.is_empty(&tx)?) } - fn insert(&self, tree: usize, key: &[u8], value: &[u8]) -> Result<()> { + fn insert(&self, tree: usize, key: &[u8], value: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree)?; let mut tx = self.db.write_txn()?; tree.put(&mut tx, key, value)?; @@ -193,7 +199,7 @@ impl IDb for LmdbDb { Ok(()) } - fn remove(&self, tree: usize, key: &[u8]) -> Result<()> { + fn remove(&self, tree: usize, key: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree)?; let mut tx = self.db.write_txn()?; tree.delete(&mut tx, key)?; @@ -201,7 +207,7 @@ impl IDb for LmdbDb { Ok(()) } - fn clear(&self, tree: usize) -> Result<()> { + fn clear(&self, tree: usize) -> DbResult<()> { let tree = self.get_tree(tree)?; let mut tx = self.db.write_txn()?; tree.clear(&mut tx)?; @@ -209,14 +215,14 @@ impl IDb for LmdbDb { Ok(()) } - fn iter(&self, tree: usize) -> Result> { + fn iter(&self, tree: usize) -> DbResult> { let tree = self.get_tree(tree)?; let tx = self.db.read_txn()?; // Safety: the cloture does not store its argument anywhere, unsafe { TxAndIterator::make(tx, |tx| Ok(tree.iter(tx)?)) } } - fn iter_rev(&self, tree: usize) -> Result> { + fn iter_rev(&self, tree: usize) -> DbResult> { let tree = self.get_tree(tree)?; let tx = self.db.read_txn()?; // Safety: the cloture does not store its argument anywhere, @@ -228,7 +234,7 @@ impl IDb for LmdbDb { tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> Result> { + ) -> DbResult> { let tree = self.get_tree(tree)?; let tx = self.db.read_txn()?; // Safety: the cloture does not store its argument anywhere, @@ -239,7 +245,7 @@ impl IDb for LmdbDb { tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> Result> { + ) -> DbResult> { let tree = self.get_tree(tree)?; let tx = self.db.read_txn()?; // Safety: the cloture does not store its argument anywhere, @@ -271,9 +277,9 @@ impl IDb for LmdbDb { } TxFnResult::DbErr => { tx.tx.abort(); - Err(TxError::Db(Error( - "(this message will be discarded)".into(), - ))) + Err(TxError::Db( + DbError("(this message will be discarded)".into()).into(), + )) } } } @@ -287,49 +293,49 @@ struct LmdbTx<'a> { } impl<'a> LmdbTx<'a> { - fn get_tree(&self, i: usize) -> TxOpResult<&Database> { + fn get_tree(&self, i: usize) -> DbResult<&Database> { self.trees.get(i).ok_or_else(|| { - TxOpError(Error( + DbError( "invalid tree id (it might have been opened after the transaction started)".into(), - )) + ) }) } } impl<'a> ITx for LmdbTx<'a> { - fn get(&self, tree: usize, key: &[u8]) -> TxOpResult> { + fn get(&self, tree: usize, key: &[u8]) -> DbResult> { let tree = self.get_tree(tree)?; match tree.get(&self.tx, key)? { Some(v) => Ok(Some(v.to_vec())), None => Ok(None), } } - fn len(&self, tree: usize) -> TxOpResult { + fn len(&self, tree: usize) -> DbResult { let tree = self.get_tree(tree)?; Ok(tree.len(&self.tx)? as usize) } - fn insert(&mut self, tree: usize, key: &[u8], value: &[u8]) -> TxOpResult<()> { + fn insert(&mut self, tree: usize, key: &[u8], value: &[u8]) -> DbResult<()> { let tree = *self.get_tree(tree)?; tree.put(&mut self.tx, key, value)?; Ok(()) } - fn remove(&mut self, tree: usize, key: &[u8]) -> TxOpResult<()> { + fn remove(&mut self, tree: usize, key: &[u8]) -> DbResult<()> { let tree = *self.get_tree(tree)?; tree.delete(&mut self.tx, key)?; Ok(()) } - fn clear(&mut self, tree: usize) -> TxOpResult<()> { + fn clear(&mut self, tree: usize) -> DbResult<()> { let tree = *self.get_tree(tree)?; tree.clear(&mut self.tx)?; Ok(()) } - fn iter(&self, tree: usize) -> TxOpResult> { + fn iter(&self, tree: usize) -> DbResult> { let tree = *self.get_tree(tree)?; Ok(Box::new(tree.iter(&self.tx)?.map(tx_iter_item))) } - fn iter_rev(&self, tree: usize) -> TxOpResult> { + fn iter_rev(&self, tree: usize) -> DbResult> { let tree = *self.get_tree(tree)?; Ok(Box::new(tree.rev_iter(&self.tx)?.map(tx_iter_item))) } @@ -339,7 +345,7 @@ impl<'a> ITx for LmdbTx<'a> { tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> TxOpResult> { + ) -> DbResult> { let tree = *self.get_tree(tree)?; Ok(Box::new( tree.range(&self.tx, &(low, high))?.map(tx_iter_item), @@ -350,7 +356,7 @@ impl<'a> ITx for LmdbTx<'a> { tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> TxOpResult> { + ) -> DbResult> { let tree = *self.get_tree(tree)?; Ok(Box::new( tree.rev_range(&self.tx, &(low, high))?.map(tx_iter_item), @@ -386,9 +392,9 @@ where } /// Safety: iterfun must not store its argument anywhere but in its result. - unsafe fn make(tx: RoTxn<'a, WithTls>, iterfun: F) -> Result> + unsafe fn make(tx: RoTxn<'a, WithTls>, iterfun: F) -> DbResult> where - F: FnOnce(&'a RoTxn<'a>) -> Result, + F: FnOnce(&'a RoTxn<'a>) -> DbResult, { let res = TxAndIterator { tx, @@ -436,13 +442,13 @@ impl<'a, I> Iterator for TxAndIteratorPin<'a, I> where I: Iterator> + 'a, { - type Item = Result<(Value, Value)>; + type Item = DbResult<(Value, Value)>; fn next(&mut self) -> Option { let mut_ref = Pin::as_mut(&mut self.0); let next = mut_ref.iter().as_mut()?.next()?; let res = match next { - Err(e) => Err(e.into()), + Err(e) => Err(DbError::from(e)), Ok((k, v)) => Ok((k.to_vec(), v.to_vec())), }; Some(res) @@ -453,9 +459,9 @@ where fn tx_iter_item<'a>( item: std::result::Result<(&'a [u8], &'a [u8]), heed::Error>, -) -> TxOpResult<(Vec, Vec)> { +) -> DbResult<(Vec, Vec)> { item.map(|(k, v)| (k.to_vec(), v.to_vec())) - .map_err(|e| TxOpError(Error::from(e))) + .map_err(DbError::from) } // ---- utility ---- diff --git a/src/db/open.rs b/src/db/open.rs index dad0492a..199380dc 100644 --- a/src/db/open.rs +++ b/src/db/open.rs @@ -1,6 +1,6 @@ use std::path::{Path, PathBuf}; -use crate::{Db, Error, Result}; +use crate::{Db, DbError, Error, Result}; /// List of supported database engine types /// @@ -49,14 +49,14 @@ impl std::str::FromStr for Engine { "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( + "sled" => Err(DbError("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()).into()), + kind => Err(DbError( format!( "Invalid DB engine: {} (options are: lmdb, sqlite, fjall)", kind ) .into(), - )), + ).into()), } } } @@ -72,22 +72,23 @@ pub fn open_db(path: &PathBuf, engine: Engine, opt: &OpenOpt) -> Result { match engine { // ---- Sqlite DB ---- #[cfg(feature = "sqlite")] - Engine::Sqlite => crate::sqlite_adapter::open_db(path, opt), + Engine::Sqlite => Ok(crate::sqlite_adapter::open_db(path, opt)?), // ---- LMDB DB ---- #[cfg(feature = "lmdb")] - Engine::Lmdb => crate::lmdb_adapter::open_db(path, opt), + Engine::Lmdb => Ok(crate::lmdb_adapter::open_db(path, opt)?), // ---- Fjall DB ---- #[cfg(feature = "fjall")] - Engine::Fjall => crate::fjall_adapter::open_db(path, opt), + Engine::Fjall => Ok(crate::fjall_adapter::open_db(path, opt)?), // 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. #[allow(unreachable_patterns)] - engine => Err(Error( + engine => Err(DbError( format!("DB engine support not available in this build: {}", engine).into(), - )), + ) + .into()), } } diff --git a/src/db/sqlite_adapter.rs b/src/db/sqlite_adapter.rs index cb778eb5..c5297cb6 100644 --- a/src/db/sqlite_adapter.rs +++ b/src/db/sqlite_adapter.rs @@ -12,7 +12,7 @@ use rusqlite::{params, Rows, Statement, Transaction}; use crate::{ open::{Engine, OpenOpt}, - Db, Error, IDb, ITx, ITxFn, OnCommit, Result, TxError, TxFnResult, TxOpError, TxOpResult, + Db, DbError, DbResult, Error, IDb, ITx, ITxFn, OnCommit, TxError, TxFnResult, TxOpError, TxResult, TxValueIter, Value, ValueIter, }; @@ -20,7 +20,7 @@ pub use rusqlite; // ---- top-level open function -pub(crate) fn open_db(path: &PathBuf, opt: &OpenOpt) -> Result { +pub(crate) fn open_db(path: &PathBuf, opt: &OpenOpt) -> DbResult { info!("Opening Sqlite database at: {}", path.display()); let manager = r2d2_sqlite::SqliteConnectionManager::file(path); SqliteDb::open(manager, opt.fsync) @@ -32,21 +32,33 @@ type Connection = r2d2::PooledConnection; // --- err +impl From for DbError { + fn from(e: rusqlite::Error) -> DbError { + DbError(format!("Sqlite: {}", e).into()) + } +} + impl From for Error { fn from(e: rusqlite::Error) -> Error { - Error(format!("Sqlite: {}", e).into()) + DbError::from(e).into() + } +} + +impl From for DbError { + fn from(e: r2d2::Error) -> DbError { + DbError(format!("Sqlite: {}", e).into()) } } impl From for Error { fn from(e: r2d2::Error) -> Error { - Error(format!("Sqlite: {}", e).into()) + DbError::from(e).into() } } impl From for TxOpError { fn from(e: rusqlite::Error) -> TxOpError { - TxOpError(e.into()) + DbError::from(e).into() } } @@ -62,7 +74,7 @@ pub struct SqliteDb { } impl SqliteDb { - pub fn open(manager: SqliteConnectionManager, sync_mode: bool) -> Result { + pub fn open(manager: SqliteConnectionManager, sync_mode: bool) -> DbResult { let manager = manager.with_init(move |db| { db.pragma_update(None, "journal_mode", "WAL")?; if sync_mode { @@ -82,16 +94,16 @@ impl SqliteDb { } impl SqliteDb { - fn get_tree(&self, i: usize) -> Result> { + fn get_tree(&self, i: usize) -> DbResult> { self.trees .read() .unwrap() .get(i) .cloned() - .ok_or_else(|| Error("invalid tree id".into())) + .ok_or_else(|| DbError("invalid tree id".into())) } - fn internal_get(&self, db: &Connection, tree: &str, key: &[u8]) -> Result> { + fn internal_get(&self, db: &Connection, tree: &str, key: &[u8]) -> DbResult> { let mut stmt = db.prepare(&format!("SELECT v FROM {} WHERE k = ?1", tree))?; let mut res_iter = stmt.query([key])?; match res_iter.next()? { @@ -106,7 +118,7 @@ impl IDb for SqliteDb { format!("sqlite3 v{} (using rusqlite crate)", rusqlite::version()) } - fn open_tree(&self, name: &str) -> Result { + fn open_tree(&self, name: &str) -> DbResult { let name = format!("tree_{}", name.replace(':', "_COLON_")); let mut trees = self.trees.write().unwrap(); @@ -133,7 +145,7 @@ impl IDb for SqliteDb { } } - fn list_trees(&self) -> Result> { + fn list_trees(&self) -> DbResult> { let mut trees = vec![]; let db = self.db.get()?; @@ -150,13 +162,13 @@ impl IDb for SqliteDb { Ok(trees) } - fn snapshot(&self, base_path: &Path) -> Result<()> { + fn snapshot(&self, base_path: &Path) -> DbResult<()> { std::fs::create_dir_all(base_path)?; let path = Engine::Sqlite .db_path(base_path) .into_os_string() .into_string() - .map_err(|_| Error("invalid sqlite path string".into()))?; + .map_err(|_| DbError("invalid sqlite path string".into()))?; info!("Start sqlite VACUUM INTO `{}`", path); self.db.get()?.execute("VACUUM INTO ?1", params![path])?; @@ -167,12 +179,12 @@ impl IDb for SqliteDb { // ---- - fn get(&self, tree: usize, key: &[u8]) -> Result> { + fn get(&self, tree: usize, key: &[u8]) -> DbResult> { let tree = self.get_tree(tree)?; self.internal_get(&self.db.get()?, &tree, key) } - fn approximate_len(&self, tree: usize) -> Result { + fn approximate_len(&self, tree: usize) -> DbResult { let tree = self.get_tree(tree)?; let db = self.db.get()?; @@ -184,11 +196,11 @@ impl IDb for SqliteDb { } } - fn is_empty(&self, tree: usize) -> Result { + fn is_empty(&self, tree: usize) -> DbResult { Ok(self.approximate_len(tree)? == 0) } - fn insert(&self, tree: usize, key: &[u8], value: &[u8]) -> Result<()> { + fn insert(&self, tree: usize, key: &[u8], value: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree)?; let db = self.db.get()?; let lock = self.write_lock.lock(); @@ -206,7 +218,7 @@ impl IDb for SqliteDb { Ok(()) } - fn remove(&self, tree: usize, key: &[u8]) -> Result<()> { + fn remove(&self, tree: usize, key: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree)?; let db = self.db.get()?; let lock = self.write_lock.lock(); @@ -217,7 +229,7 @@ impl IDb for SqliteDb { Ok(()) } - fn clear(&self, tree: usize) -> Result<()> { + fn clear(&self, tree: usize) -> DbResult<()> { let tree = self.get_tree(tree)?; let db = self.db.get()?; let lock = self.write_lock.lock(); @@ -228,13 +240,13 @@ impl IDb for SqliteDb { Ok(()) } - fn iter(&self, tree: usize) -> Result> { + fn iter(&self, tree: usize) -> DbResult> { let tree = self.get_tree(tree)?; let sql = format!("SELECT k, v FROM {} ORDER BY k ASC", tree); DbValueIterator::make(self.db.get()?, &sql, []) } - fn iter_rev(&self, tree: usize) -> Result> { + fn iter_rev(&self, tree: usize) -> DbResult> { let tree = self.get_tree(tree)?; let sql = format!("SELECT k, v FROM {} ORDER BY k DESC", tree); DbValueIterator::make(self.db.get()?, &sql, []) @@ -245,7 +257,7 @@ impl IDb for SqliteDb { tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> Result> { + ) -> DbResult> { let tree = self.get_tree(tree)?; let (bounds_sql, params) = bounds_sql(low, high); @@ -263,7 +275,7 @@ impl IDb for SqliteDb { tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> Result> { + ) -> DbResult> { let tree = self.get_tree(tree)?; let (bounds_sql, params) = bounds_sql(low, high); @@ -300,9 +312,9 @@ impl IDb for SqliteDb { } TxFnResult::DbErr => { tx.tx.rollback().map_err(Error::from).map_err(TxError::Db)?; - Err(TxError::Db(Error( - "(this message will be discarded)".into(), - ))) + Err(TxError::Db( + DbError("(this message will be discarded)".into()).into(), + )) } }; @@ -320,15 +332,15 @@ struct SqliteTx<'a> { } impl<'a> SqliteTx<'a> { - fn get_tree(&self, i: usize) -> TxOpResult<&'_ str> { + fn get_tree(&self, i: usize) -> DbResult<&'_ str> { self.trees.get(i).map(Arc::as_ref).ok_or_else(|| { - TxOpError(Error( + DbError( "invalid tree id (it might have been opened after the transaction started)".into(), - )) + ) }) } - fn internal_get(&self, tree: &str, key: &[u8]) -> TxOpResult> { + fn internal_get(&self, tree: &str, key: &[u8]) -> DbResult> { let mut stmt = self .tx .prepare(&format!("SELECT v FROM {} WHERE k = ?1", tree))?; @@ -341,11 +353,11 @@ impl<'a> SqliteTx<'a> { } impl<'a> ITx for SqliteTx<'a> { - fn get(&self, tree: usize, key: &[u8]) -> TxOpResult> { + fn get(&self, tree: usize, key: &[u8]) -> DbResult> { let tree = self.get_tree(tree)?; self.internal_get(tree, key) } - fn len(&self, tree: usize) -> TxOpResult { + fn len(&self, tree: usize) -> DbResult { let tree = self.get_tree(tree)?; let mut stmt = self.tx.prepare(&format!("SELECT COUNT(*) FROM {}", tree))?; let mut res_iter = stmt.query([])?; @@ -355,30 +367,30 @@ impl<'a> ITx for SqliteTx<'a> { } } - fn insert(&mut self, tree: usize, key: &[u8], value: &[u8]) -> TxOpResult<()> { + fn insert(&mut self, tree: usize, key: &[u8], value: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree)?; let sql = format!("INSERT OR REPLACE INTO {} (k, v) VALUES (?1, ?2)", tree); self.tx.execute(&sql, params![key, value])?; Ok(()) } - fn remove(&mut self, tree: usize, key: &[u8]) -> TxOpResult<()> { + fn remove(&mut self, tree: usize, key: &[u8]) -> DbResult<()> { let tree = self.get_tree(tree)?; self.tx .execute(&format!("DELETE FROM {} WHERE k = ?1", tree), params![key])?; Ok(()) } - fn clear(&mut self, tree: usize) -> TxOpResult<()> { + fn clear(&mut self, tree: usize) -> DbResult<()> { let tree = self.get_tree(tree)?; self.tx.execute(&format!("DELETE FROM {}", tree), [])?; Ok(()) } - fn iter(&self, tree: usize) -> TxOpResult> { + fn iter(&self, tree: usize) -> DbResult> { let tree = self.get_tree(tree)?; let sql = format!("SELECT k, v FROM {} ORDER BY k ASC", tree); TxValueIterator::make(self, &sql, []) } - fn iter_rev(&self, tree: usize) -> TxOpResult> { + fn iter_rev(&self, tree: usize) -> DbResult> { let tree = self.get_tree(tree)?; let sql = format!("SELECT k, v FROM {} ORDER BY k DESC", tree); TxValueIterator::make(self, &sql, []) @@ -389,7 +401,7 @@ impl<'a> ITx for SqliteTx<'a> { tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> TxOpResult> { + ) -> DbResult> { let tree = self.get_tree(tree)?; let (bounds_sql, params) = bounds_sql(low, high); @@ -407,7 +419,7 @@ impl<'a> ITx for SqliteTx<'a> { tree: usize, low: Bound<&'r [u8]>, high: Bound<&'r [u8]>, - ) -> TxOpResult> { + ) -> DbResult> { let tree = self.get_tree(tree)?; let (bounds_sql, params) = bounds_sql(low, high); @@ -439,7 +451,7 @@ impl DbValueIterator { db: Connection, sql: &str, args: P, - ) -> Result> { + ) -> DbResult> { let res = DbValueIterator { db, stmt: None, @@ -484,7 +496,7 @@ impl Drop for DbValueIterator { struct DbValueIteratorPin(Pin>); impl Iterator for DbValueIteratorPin { - type Item = Result<(Value, Value)>; + type Item = DbResult<(Value, Value)>; fn next(&mut self) -> Option { let mut_ref = Pin::as_mut(&mut self.0); @@ -509,7 +521,7 @@ impl<'a> TxValueIterator<'a> { tx: &'a SqliteTx<'a>, sql: &str, args: P, - ) -> TxOpResult> { + ) -> DbResult> { let stmt = tx.tx.prepare(sql)?; let res = TxValueIterator { stmt, @@ -543,7 +555,7 @@ impl<'a> Drop for TxValueIterator<'a> { struct TxValueIteratorPin<'a>(Pin>>); impl<'a> Iterator for TxValueIteratorPin<'a> { - type Item = TxOpResult<(Value, Value)>; + type Item = DbResult<(Value, Value)>; fn next(&mut self) -> Option { let mut_ref = Pin::as_mut(&mut self.0); diff --git a/src/db/typed.rs b/src/db/typed.rs new file mode 100644 index 00000000..b97d2e27 --- /dev/null +++ b/src/db/typed.rs @@ -0,0 +1,241 @@ +use std::marker::PhantomData; +use std::ops::{Bound, RangeBounds}; + +// Todo: some parts of this code, notably around ranges are never used but are here to prepare +// the migration of the parts of the codebase that still use untyped trees. At one point, this +// migration should be done or these functions deleted. + +use super::{ + DbResult, DecodeError, Error, Result, Transaction, Tree, TxOpError, TxOpResult, TxValueIter, + ValueIter, +}; + +pub use super::Db; + +pub trait DbBytes: Sized { + fn encode(&self) -> Vec; + fn decode(bytes: &[u8]) -> std::result::Result; +} + +/// Subtrait of [`DbBytes`] for types used as tree keys with operations where order matters +/// (`get_gt`, range, etc...). +/// +/// Implementors must guarantee that the byte encoding is order-preserving: +/// for any `a, b: Self`, `a.cmp(&b) == a.encode().cmp(&b.encode())`. +pub trait DbOrdKey: DbBytes + Ord {} + +#[derive(Clone)] +pub struct TypedTree { + inner: Tree, + _phantom: PhantomData<[(K, V)]>, +} + +impl TypedTree { + pub fn new(tree: Tree) -> Self { + Self { + inner: tree, + _phantom: PhantomData, + } + } + + pub fn db(&self) -> Db { + self.inner.db() + } + + pub fn untyped(&self) -> &Tree { + &self.inner + } + + pub fn get(&self, key: &K) -> Result> { + self.inner + .get(key.encode())? + .map(|v| V::decode(&v).map_err(Error::from)) + .transpose() + } + + pub fn approximate_len(&self) -> DbResult { + self.inner.approximate_len() + } + + pub fn is_empty(&self) -> DbResult { + self.inner.is_empty() + } + + pub fn insert(&self, key: &K, value: &V) -> DbResult<()> { + self.inner.insert(key.encode(), value.encode()) + } + + pub fn remove(&self, key: &K) -> DbResult<()> { + self.inner.remove(key.encode()) + } + + pub fn clear(&self) -> DbResult<()> { + self.inner.clear() + } + + pub fn tx_get(&self, tx: &Transaction<'_>, key: &K) -> TxOpResult> { + tx.get(&self.inner, key.encode())? + .map(|v| V::decode(&v).map_err(TxOpError::from)) + .transpose() + } + + pub fn tx_insert(&self, tx: &mut Transaction<'_>, key: &K, value: &V) -> TxOpResult<()> { + tx.insert(&self.inner, key.encode(), value.encode()) + } + + pub fn tx_remove(&self, tx: &mut Transaction<'_>, key: &K) -> TxOpResult<()> { + tx.remove(&self.inner, key.encode()) + } + + pub fn tx_clear(&self, tx: &mut Transaction<'_>) -> TxOpResult<()> { + tx.clear(&self.inner) + } +} + +impl TypedTree { + pub fn first(&self) -> Result> { + self.iter()?.next().transpose() + } + + pub fn get_gt(&self, from: &K) -> Result> { + self.inner + .get_gt(from.encode())? + .map(|(k, v)| { + Ok(( + K::decode(&k).map_err(Error::from)?, + V::decode(&v).map_err(Error::from)?, + )) + }) + .transpose() + } + + pub fn iter(&self) -> Result> { + Ok(TypedIter::new(self.inner.iter()?)) + } + + pub fn iter_rev(&self) -> Result> { + Ok(TypedIter::new(self.inner.iter_rev()?)) + } + + pub fn range>(&self, range: R) -> Result> { + Ok(TypedIter::new(self.inner.range(encode_range(range))?)) + } + + pub fn range_rev>(&self, range: R) -> Result> { + Ok(TypedIter::new(self.inner.range_rev(encode_range(range))?)) + } + + pub fn tx_iter<'t>(&self, tx: &'t Transaction<'_>) -> TxOpResult> { + Ok(TypedTxIter::new(tx.iter(&self.inner)?)) + } + + pub fn tx_iter_rev<'t>(&self, tx: &'t Transaction<'_>) -> TxOpResult> { + Ok(TypedTxIter::new(tx.iter_rev(&self.inner)?)) + } + + pub fn tx_range<'t, R: RangeBounds>( + &self, + tx: &'t Transaction<'_>, + range: R, + ) -> TxOpResult> { + Ok(TypedTxIter::new( + tx.range(&self.inner, encode_range(range))?, + )) + } + + pub fn tx_range_rev<'t, R: RangeBounds>( + &self, + tx: &'t Transaction<'_>, + range: R, + ) -> TxOpResult> { + Ok(TypedTxIter::new( + tx.range_rev(&self.inner, encode_range(range))?, + )) + } +} + +impl From for TypedTree { + fn from(tree: Tree) -> Self { + Self::new(tree) + } +} + +impl Db { + pub fn open_typed_tree>( + &self, + name: S, + ) -> DbResult> { + Ok(TypedTree::new(self.open_tree(name)?)) + } +} + +pub struct TypedIter<'a, K, V> { + inner: ValueIter<'a>, + _phantom: PhantomData<(K, V)>, +} + +impl<'a, K, V> TypedIter<'a, K, V> { + fn new(inner: ValueIter<'a>) -> Self { + Self { + inner, + _phantom: PhantomData, + } + } +} + +impl Iterator for TypedIter<'_, K, V> { + type Item = Result<(K, V)>; + + fn next(&mut self) -> Option { + self.inner.next().map(|res| { + let (k, v) = res?; + Ok(( + K::decode(&k).map_err(Error::from)?, + V::decode(&v).map_err(Error::from)?, + )) + }) + } +} + +pub struct TypedTxIter<'a, K, V> { + inner: TxValueIter<'a>, + _phantom: PhantomData<(K, V)>, +} + +impl<'a, K, V> TypedTxIter<'a, K, V> { + fn new(inner: TxValueIter<'a>) -> Self { + Self { + inner, + _phantom: PhantomData, + } + } +} + +impl Iterator for TypedTxIter<'_, K, V> { + type Item = TxOpResult<(K, V)>; + + fn next(&mut self) -> Option { + self.inner.next().map(|res| { + let (k, v) = res?; + Ok(( + K::decode(&k).map_err(TxOpError::from)?, + V::decode(&v).map_err(TxOpError::from)?, + )) + }) + } +} + +fn encode_range>(range: R) -> (Bound>, Bound>) { + ( + encode_bound(range.start_bound()), + encode_bound(range.end_bound()), + ) +} + +fn encode_bound(bound: Bound<&K>) -> Bound> { + match bound { + Bound::Included(k) => Bound::Included(k.encode()), + Bound::Excluded(k) => Bound::Excluded(k.encode()), + Bound::Unbounded => Bound::Unbounded, + } +} diff --git a/src/garage/cli/local/convert_db.rs b/src/garage/cli/local/convert_db.rs index e1a67d02..01a5a52d 100644 --- a/src/garage/cli/local/convert_db.rs +++ b/src/garage/cli/local/convert_db.rs @@ -49,7 +49,7 @@ pub struct OpenLmdbOpt { pub(crate) fn do_conversion(args: ConvertDbOpt) -> Result<()> { if args.input_engine == args.output_engine { - return Err(Error("input and output database engine must differ".into())); + return Err(DbError("input and output database engine must differ".into()).into()); } let opt = OpenOpt { diff --git a/src/model/index_counter.rs b/src/model/index_counter.rs index 50abdec3..885b8b3a 100644 --- a/src/model/index_counter.rs +++ b/src/model/index_counter.rs @@ -163,6 +163,7 @@ impl TableSchema for CounterTable { pub struct IndexCounter { this_node: Uuid, + // TODO: migrate to TypedTree local_counter: db::Tree, pub table: Arc, TableShardedReplication>>, } diff --git a/src/model/k2v/rpc.rs b/src/model/k2v/rpc.rs index 7d2bede3..d14308df 100644 --- a/src/model/k2v/rpc.rs +++ b/src/model/k2v/rpc.rs @@ -84,6 +84,7 @@ pub struct K2VRpcHandler { // Using a mutex on the local_timestamp_tree is not strictly necessary, // but it helps to not try to do several inserts at the same time, // which would create transaction conflicts and force many useless retries. + // TODO: migrate to TypedTree local_timestamp_tree: Mutex, endpoint: Arc>, diff --git a/src/table/data.rs b/src/table/data.rs index 86bf7b93..591e9eea 100644 --- a/src/table/data.rs +++ b/src/table/data.rs @@ -26,15 +26,20 @@ pub struct TableData { pub instance: F, pub replication: R, + // TODO: migrate to TypedTree pub store: db::Tree, + // TODO: migrate to TypedTree pub(crate) merkle_tree: db::Tree, + // TODO: migrate to TypedTree pub(crate) merkle_todo: db::Tree, pub(crate) merkle_todo_notify: Notify, + // TODO: migrate to TypedTree pub(crate) insert_queue: db::Tree, pub(crate) insert_queue_notify: Arc, + // TODO: migrate to TypedTree pub(crate) gc_todo: db::Tree, pub(crate) metrics: TableMetrics, diff --git a/src/util/data.rs b/src/util/data.rs index 63c4379f..4b500b14 100644 --- a/src/util/data.rs +++ b/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 { + self.0.into() + } + fn decode(bytes: &[u8]) -> std::result::Result { + 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::*; diff --git a/src/util/error.rs b/src/util/error.rs index c3d4ab58..7cbe0a4a 100644 --- a/src/util/error.rs +++ b/src/util/error.rs @@ -79,6 +79,18 @@ impl Error { } } +impl From for Error { + fn from(e: garage_db::DbError) -> Error { + Error::Db(e.into()) + } +} + +impl From for Error { + fn from(e: garage_db::DecodeError) -> Error { + Error::Db(e.into()) + } +} + impl From> for Error { fn from(e: garage_db::TxError) -> Error { match e {