Use specialized DBError type instead of io::Error

5 files changed, 72 insertions, 50 deletions

diff --git a/log_db/benches/utils.rs b/log_db/benches/utils.rs
index afa899a..9aed90e 100644
--- a/log_db/benches/utils.rs
+++ b/log_db/benches/utils.rs
@@ -2,7 +2,6 @@ use log_db::*;
 use rand::distributions::Alphanumeric;
 use rand::Rng;
 use std::fmt::Debug;
-use std::io;
 
 // Function to generate a random integer
 pub fn random_int(from: i64, to: i64) -> i64 {
@@ -35,7 +34,7 @@ pub fn prefill_db<T: Eq + Clone + Debug>(
     db: &mut DB<T>,
     n_records: usize,
     compact: bool,
-) -> Result<(), io::Error> {
+) -> Result<(), DBError> {
     for _ in 0..n_records {
         let record = random_record(0, n_records as i64);
         db.upsert(&record)?;
diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index e379364..bf949c2 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -7,6 +7,7 @@ use std::fs::{self, metadata, File};
 use std::io::{self, Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 use std::thread;
+use thiserror::Error;
 use uuid::Uuid;
 
 // For Unix-like systems
@@ -29,6 +30,18 @@ pub fn metadata_filename(num: u16) -> String {
     format!("metadata.{}", num)
 }
 
+#[derive(Debug, Error)]
+pub enum DBError {
+    #[error("lock request failed: {0}")]
+    LockRequestError(#[from] LockRequestError),
+    #[error("validation failed: {0}")]
+    ValidationError(String),
+    #[error("consistency check failed: {0}")]
+    ConsistencyError(String),
+    #[error("unexpected IO error: {0}")]
+    IOError(#[from] io::Error),
+}
+
 /// LogKey is a packed struct that contains:
 /// - a log segment number (16 bits)
 /// - a log index within the segment (48 bits)
@@ -667,11 +680,10 @@ pub fn read_metadata_header(metadata_file: &mut fs::File) -> Result<MetadataHead
     Ok(header)
 }
 
-pub fn validate_metadata_header(header: &MetadataHeader) -> Result<(), io::Error> {
+pub fn validate_metadata_header(header: &MetadataHeader) -> Result<(), DBError> {
     if header.version != 1 {
-        return Err(io::Error::new(
-            io::ErrorKind::InvalidData,
-            "Unsupported metadata file version",
+        return Err(DBError::ValidationError(
+            "Unsupported metadata file version".to_owned(),
         ));
     }
 
@@ -773,7 +785,19 @@ pub fn ensure_active_metadata_is_valid(
     }
 }
 
-pub fn is_exclusive_lock_requested(data_dir: &Path) -> Result<bool, io::Error> {
+const LOCK_WAIT_MAX_MS: u64 = 100;
+
+#[derive(Error, Debug)]
+pub enum LockRequestError {
+    #[error("lock request file was removed unexpectedly")]
+    LockRequestFileRemoved,
+    #[error("timed out while waiting for a lock, max wait time: {0} ms")]
+    TimedOut(u64),
+    #[error("unexpected IO error: {0}")]
+    IOError(#[from] io::Error),
+}
+
+pub fn is_exclusive_lock_requested(data_dir: &Path) -> Result<bool, LockRequestError> {
     let lock_request_path = data_dir.join(EXCL_LOCK_REQUEST_FILENAME);
     let lock_request_file = fs::OpenOptions::new()
         .create(true)
@@ -787,16 +811,14 @@ pub fn is_exclusive_lock_requested(data_dir: &Path) -> Result<bool, io::Error> {
             if e.kind() == lock_contended_error().kind() {
                 return Ok(true);
             }
-            return Err(e);
+            return Err(LockRequestError::IOError(e));
         }
+
         Ok(_) => {
             // Check that the exclusive lock request file is still the same as the one we opened
             if !is_file_same_as_path(&lock_request_file, &lock_request_path)? {
                 // The lock request file has been removed
-                return Err(io::Error::new(
-                    io::ErrorKind::Other,
-                    "Lock request file was removed unexpectedly",
-                ));
+                return Err(LockRequestError::LockRequestFileRemoved);
             }
 
             lock_request_file.unlock()?;
@@ -805,8 +827,7 @@ pub fn is_exclusive_lock_requested(data_dir: &Path) -> Result<bool, io::Error> {
     }
 }
 
-pub fn request_shared_lock(data_dir: &Path, file: &mut fs::File) -> Result<(), io::Error> {
-    const SHARED_LOCK_WAIT_MAX_MS: u64 = 100;
+pub fn request_shared_lock(data_dir: &Path, file: &mut fs::File) -> Result<(), LockRequestError> {
     let mut timeout = 5;
     loop {
         if is_exclusive_lock_requested(data_dir)? {
@@ -815,7 +836,11 @@ pub fn request_shared_lock(data_dir: &Path, file: &mut fs::File) -> Result<(), i
                 timeout
             );
             thread::sleep(std::time::Duration::from_millis(timeout));
-            timeout = std::cmp::min(timeout * 2, SHARED_LOCK_WAIT_MAX_MS);
+            timeout *= 2;
+
+            if timeout > LOCK_WAIT_MAX_MS {
+                return Err(LockRequestError::TimedOut(LOCK_WAIT_MAX_MS));
+            }
         } else {
             file.lock_shared()?;
             return Ok(());
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index bce9d13..400d580 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -19,6 +19,7 @@ use std::fmt::Debug;
 use std::fs::{self};
 use std::io::{self, Read, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
+use thiserror::Error;
 
 pub struct ConfigBuilder<Field: Eq + Clone + Debug> {
     data_dir: Option<String>,
@@ -96,7 +97,7 @@ impl<'a, Field: Eq + Clone + Debug> ConfigBuilder<Field> {
         self
     }
 
-    pub fn initialize(&self) -> Result<DB<Field>, io::Error> {
+    pub fn initialize(&self) -> Result<DB<Field>, DBError> {
         let config = Config::<Field> {
             data_dir: self.data_dir.clone().unwrap_or("db_data".to_string()),
             segment_size: self.segment_size.unwrap_or(4 * 1024 * 1024), // 4MB
@@ -155,7 +156,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         ConfigBuilder::new()
     }
 
-    fn initialize(config: &Config<Field>) -> Result<DB<Field>, io::Error> {
+    fn initialize(config: &Config<Field>) -> Result<DB<Field>, DBError> {
         info!("Initializing DB...");
         // If data_dir does not exist or is empty, create it and any necessary files
         // After creation, the directory should always be in a complete state
@@ -168,7 +169,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             Ok(_) => {}
             Err(e) => {
                 if e.kind() != io::ErrorKind::AlreadyExists {
-                    return Err(e);
+                    return Err(DBError::IOError(e));
                 }
             }
         }
@@ -231,9 +232,8 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             match prim_value_type {
                 PrimValueType::Int | PrimValueType::String => {}
                 _ => {
-                    return Err(io::Error::new(
-                        io::ErrorKind::InvalidInput,
-                        "Secondary key must be an IndexableValue",
+                    return Err(DBError::ValidationError(
+                        "Secondary key must be an IndexableValue".to_owned(),
                     ))
                 }
             }
@@ -277,7 +277,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
         Ok(db)
     }
 
-    fn refresh_indexes(&mut self) -> Result<(), io::Error> {
+    fn refresh_indexes(&mut self) -> Result<(), DBError> {
         let active_symlink_path = self.data_dir.join(ACTIVE_SYMLINK_FILENAME);
         let active_target = fs::read_link(active_symlink_path)?;
         let active_metadata_path = self.data_dir.join(active_target);
@@ -292,17 +292,15 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
 
             let metadata_len = metadata_file.seek(SeekFrom::End(0))?;
             if (metadata_len - METADATA_FILE_HEADER_SIZE as u64) % METADATA_ROW_LENGTH as u64 != 0 {
-                return Err(io::Error::new(
-                    io::ErrorKind::InvalidData,
-                    format!(
-                        "Metadata file {} has invalid size: {}",
-                        metadata_path.display(),
-                        metadata_len
-                    ),
-                ));
+                return Err(DBError::ConsistencyError(format!(
+                    "Metadata file {} has invalid size: {}",
+                    metadata_path.display(),
+                    metadata_len
+                )));
             }
 
-            request_shared_lock(&self.data_dir, &mut metadata_file)?;
+            request_shared_lock(&self.data_dir, &mut metadata_file)
+                .map_err(|lre| DBError::LockRequestError(lre))?;
 
             let metadata_header = read_metadata_header(&mut metadata_file)?;
             validate_metadata_header(&metadata_header)?;
@@ -353,7 +351,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
 
     /// Insert a record into the database. If the primary key value already exists,
     /// the existing record will be replaced by the supplied one.
-    pub fn upsert(&mut self, record: &Record) -> Result<(), io::Error> {
+    pub fn upsert(&mut self, record: &Record) -> Result<(), DBError> {
         debug!("Upserting record: {:?}", record);
 
         record.validate(&self.config.fields)?;
@@ -439,16 +437,13 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
 
     /// Get a record by its primary index value.
     /// E.g. `db.get(Value::Int(10))`.
-    pub fn get(&mut self, query_key: &Value) -> Result<Option<Record>, io::Error> {
+    pub fn get(&mut self, query_key: &Value) -> Result<Option<Record>, DBError> {
         let pk_type = &self.config.fields[self.primary_key_index].1;
         if !type_check(&query_key, &pk_type) {
-            return Err(io::Error::new(
-                io::ErrorKind::InvalidInput,
-                format!(
-                    "Queried value does not match primary key type: {:?}",
-                    pk_type
-                ),
-            ));
+            return Err(DBError::ValidationError(format!(
+                "Queried value does not match primary key type: {:?}",
+                pk_type
+            )));
         }
 
         debug!(
@@ -515,7 +510,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
 
     /// Get a collection of records based on a field value.
     /// Indexes will be used if they contain the requested key.
-    pub fn find_all(&mut self, field: &Field, query_key: &Value) -> Result<Vec<Record>, io::Error> {
+    pub fn find_all(&mut self, field: &Field, query_key: &Value) -> Result<Vec<Record>, DBError> {
         // If querying by primary key, return the result of `get` wrapped in a vec.
         if field == &self.config.primary_key {
             return match self.get(query_key)? {
@@ -539,9 +534,8 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
             match get_secondary_memtable_index_by_field(&self.config.secondary_keys, field) {
                 Some(index) => index,
                 None => {
-                    return Err(io::Error::new(
-                        io::ErrorKind::NotFound,
-                        "Cannot find_all by non-secondary key",
+                    return Err(DBError::ValidationError(
+                        "Cannot find_all by non-secondary key".to_owned(),
                     ))
                 }
             };
@@ -607,7 +601,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
     /// Ensures that the `self.metadata_file` and `self.data_file` handles are still pointing to the correct files.
     /// If the segment has been rotated, the handle will be closed and reopened.
     /// Returns `false` if the file has been rotated and the handle has been reopened, `true` otherwise.
-    fn ensure_metadata_file_is_active(&mut self) -> Result<bool, io::Error> {
+    fn ensure_metadata_file_is_active(&mut self) -> Result<bool, DBError> {
         let active_target = fs::read_link(&self.data_dir.join(ACTIVE_SYMLINK_FILENAME))?;
         let active_metadata_path = &self.data_dir.join(active_target);
 
@@ -640,7 +634,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> {
     /// Note that this function is synchronous and may block for a relatively long time.
     /// You may call this function in a separate thread or process to avoid blocking the main thread.
     /// However, the database will be exclusively locked, so all writes and reads will be blocked during the tasks.
-    pub fn do_maintenance_tasks(&mut self) -> Result<(), io::Error> {
+    pub fn do_maintenance_tasks(&mut self) -> Result<(), DBError> {
         request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?;
 
         ensure_active_metadata_is_valid(&self.data_dir, &mut self.active_metadata_file)?;
diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs
index 154423b..2d1615b 100644
--- a/log_db/tests/integration.rs
+++ b/log_db/tests/integration.rs
@@ -309,6 +309,7 @@ fn test_one_writer_and_multiple_reading_threads() {
         threads.push(thread::spawn(move || {
             let mut db = DB::configure()
                 .data_dir(&data_dir)
+                .segment_size(1000) // should cause rotations
                 .fields(&[(Field::Id, ValueType::int())])
                 .primary_key(Field::Id)
                 .initialize()
@@ -316,8 +317,6 @@ fn test_one_writer_and_multiple_reading_threads() {
 
             let mut timeout = 5;
             loop {
-                db.do_maintenance_tasks() // Run maintenance tasks on every read, just to test it
-                    .expect("Failed to do maintenance tasks");
                 let result = db.get(&Value::Int(i)).expect("Failed to get record");
                 match result {
                     None => {
@@ -349,6 +348,9 @@ fn test_one_writer_and_multiple_reading_threads() {
         for i in 0..threads_n {
             let record = Record::from(&[Value::Int(i)]);
             db.upsert(&record).expect("Failed to upsert record");
+
+            db.do_maintenance_tasks() // Run maintenance tasks after every write, just to test it
+                .expect("Failed to do maintenance tasks");
         }
     }));
 
diff --git a/py_bindings/src/lib.rs b/py_bindings/src/lib.rs
index ed63029..aa79680 100644
--- a/py_bindings/src/lib.rs
+++ b/py_bindings/src/lib.rs
@@ -98,7 +98,9 @@ impl Config {
             config.write_durability(tmp.write_durability.clone());
         }
 
-        let db = config.initialize().map_err(|e| PyException::new_err(e))?;
+        let db = config
+            .initialize()
+            .map_err(|e| PyException::new_err(e.to_string()))?;
         Ok(DB { db })
     }
 }
@@ -177,7 +179,7 @@ impl DB {
 
         self.db
             .upsert(&log_db::Record::from(&values))
-            .map_err(|e| PyException::new_err(e))?;
+            .map_err(|e| PyException::new_err(e.to_string()))?;
         Ok(())
     }