feat: poison

In case the database encounters an IO error during the commit, there
will be some intermediate partial changes to the database.

It is possible to recover from them, however, in the interest of
reducing complexity, we mark the database as poisoned and reject any
further modification attempts. That allows NOMT to avoid spreading the
recovery logic all around but concentrate it in the startup logic.

Author: pepyakin

nomt/src/lib.rs

@@ -311,6 +311,19 @@ impl<T: HashAlgorithm> Nomt<T> {
         self.store.sync_seqn()
     }
 
+    /// Whether the database is poisoned.
+    ///
+    /// A database becomes poisoned when an error occurred during a commit operation.
+    ///
+    /// From this point on, the database is in an inconsistent state and should be considered
+    /// read-only. Any further modifying operations will return an error.
+    ///
+    /// In order to recover from a poisoned database, the application should discard this instance
+    /// and create a new one.
+    pub fn is_poisoned(&self) -> bool {
+        self.store.is_poisoned()
+    }
+
     /// Create a new [`Session`] object with the given parameters.
     ///
     /// The [`Session`] is a read-only handle on the database and is used to create a changeset to

nomt/src/store/mod.rs

@@ -17,7 +17,7 @@ use nomt_core::{page_id::PageId, trie::KeyPath};
 use parking_lot::Mutex;
 use std::{
     fs::{File, OpenOptions},
-    sync::Arc,
+    sync::{atomic::AtomicBool, Arc},
 };
 
 #[cfg(target_os = "linux")]
@@ -46,6 +46,7 @@ struct Shared {
     meta_fd: File,
     #[allow(unused)]
     flock: flock::Flock,
+    poisoned: AtomicBool,
 
     // Retained for the lifetime of the store.
     _db_dir_fd: Arc<File>,
@@ -199,10 +200,17 @@ impl Store {
                 _db_dir_fd: db_dir_fd,
                 meta_fd,
                 flock,
+                poisoned: false.into(),
             }),
         })
     }
 
+    pub fn is_poisoned(&self) -> bool {
+        self.shared
+            .poisoned
+            .load(std::sync::atomic::Ordering::Relaxed)
+    }
+
     pub fn sync_seqn(&self) -> u32 {
         self.sync.lock().sync_seqn
     }
@@ -278,16 +286,28 @@ impl Store {
     ) -> anyhow::Result<()> {
         let mut sync = self.sync.lock();
 
-        sync.sync(
+        if self
+            .shared
+            .poisoned
+            .load(std::sync::atomic::Ordering::Relaxed)
+        {
+            anyhow::bail!("Store is poisoned due to prior error");
+        }
+
+        if let Err(e) = sync.sync(
             &self.shared,
             value_tx,
             self.shared.pages.clone(),
             self.shared.values.clone(),
             self.shared.rollback.clone(),
             page_cache,
             updated_pages,
-        )
-        .unwrap();
+        ) {
+            self.shared
+                .poisoned
+                .store(true, std::sync::atomic::Ordering::Relaxed);
+            return Err(e);
+        }
         Ok(())
     }
 }

nomt/src/store/sync.rs

@@ -51,9 +51,8 @@ impl Sync {
             rollback.begin_sync();
         }
 
-        // TODO: comprehensive error handling is coming later.
-        bitbox_sync.wait_pre_meta().unwrap();
-        let beatree_meta_wd = beatree_sync.wait_pre_meta().unwrap();
+        bitbox_sync.wait_pre_meta()?;
+        let beatree_meta_wd = beatree_sync.wait_pre_meta()?;
         let (rollback_start_live, rollback_end_live) = match rollback_sync {
             Some(ref rollback) => rollback.wait_pre_meta(),
             None => (0, 0),
@@ -90,8 +89,7 @@ impl Sync {
         beatree_sync.post_meta();
 
         if let Some(ref rollback) = rollback_sync {
-            // TODO: comprehensive error handling is coming later.
-            rollback.wait_post_meta().unwrap();
+            rollback.wait_post_meta()?;
         }
 
         Ok(())