add Error::Unrecoverable

Author: marvin-j97

src/error.rs

@@ -28,6 +28,9 @@ pub enum Error {
 
     /// Decompression failed
     Decompress,
+
+    /// Some required segments could not be recovered from disk
+    Unrecoverable,
     // TODO:
     // /// Checksum check failed
     // ChecksumMismatch,
@@ -45,7 +48,9 @@ impl std::error::Error for Error {
             Self::Io(e) => Some(e),
             Self::Encode(e) => Some(e),
             Self::Decode(e) => Some(e),
-            Self::Decompress | Self::InvalidVersion(_) | Self::Compress => None,
+            Self::Decompress | Self::InvalidVersion(_) | Self::Compress | Self::Unrecoverable => {
+                None
+            }
         }
     }
 }

src/manifest.rs

@@ -159,6 +159,10 @@ impl<C: Compressor + Clone> SegmentManifest<C> {
             map
         };
 
+        if segments.len() < ids.len() {
+            return Err(crate::Error::Unrecoverable);
+        }
+
         Ok(Self(Arc::new(SegmentManifestInner {
             path: manifest_path,
             segments: RwLock::new(segments),

tests/recovery_fail.rs

@@ -0,0 +1,64 @@
+use test_log::test;
+use value_log::{Compressor, Config, IndexWriter, MockIndex, MockIndexWriter, ValueLog};
+
+#[derive(Clone, Default)]
+struct NoCompressor;
+
+impl Compressor for NoCompressor {
+    fn compress(&self, bytes: &[u8]) -> value_log::Result<Vec<u8>> {
+        Ok(bytes.into())
+    }
+
+    fn decompress(&self, bytes: &[u8]) -> value_log::Result<Vec<u8>> {
+        Ok(bytes.into())
+    }
+}
+
+#[test]
+fn recovery_fail() -> value_log::Result<()> {
+    let folder = tempfile::tempdir()?;
+    let vl_path = folder.path();
+
+    let index = MockIndex::default();
+
+    let items = ["a", "b", "c", "d", "e"];
+
+    {
+        let value_log = ValueLog::open(vl_path, Config::<NoCompressor>::default())?;
+
+        for _ in 0..2 {
+            let mut index_writer = MockIndexWriter(index.clone());
+            let mut writer = value_log.get_writer()?;
+
+            for key in &items {
+                let value = key.repeat(10_000);
+                let value = value.as_bytes();
+
+                let key = key.as_bytes();
+
+                let vhandle = writer.get_next_value_handle();
+                index_writer.insert_indirect(key, vhandle, value.len() as u32)?;
+
+                writer.write(key, value)?;
+            }
+
+            value_log.register_writer(writer)?;
+        }
+
+        {
+            assert_eq!(2, value_log.segment_count());
+        }
+    }
+
+    // NOTE: Delete blob file to trigger Unrecoverable (it is referenced in manifest, but not on disk anymore)
+    std::fs::remove_file(vl_path.join("segments").join("1"))?;
+
+    {
+        matches!(
+            ValueLog::open(vl_path, Config::<NoCompressor>::default()),
+            Err(value_log::Error::Unrecoverable),
+        );
+    }
+
+    Ok(())
+}