feat: page checksum

Author: peeeuzin

Cargo.toml

@@ -25,6 +25,7 @@ serde = { version = "1.0", features = ["derive"] }
 bloomfilter = "1.0"
 fs2 = "0.4.3"
 bincode = "1.3.3"
+crc32fast = "1.4.2"
 
 [build-dependencies]
 hooky-rs = "1.0.0"

src/btree/mod.rs

@@ -449,12 +449,22 @@ impl<
     }
 }
 
-// #[cfg(test)]
-// mod btree_tests {
-//     use super::*;
-
-//     #[test]
-//     fn create_btree() {
-//         let mut btree = BTree::<u32, String>::new("test_data/btree").unwrap();
-//     }
-// }
+#[cfg(test)]
+mod btree_tests {
+    use super::*;
+
+    #[test]
+    fn create_btree() {
+        let mut btree = BTree::<u32, u32>::new("test_data/btree").unwrap();
+
+        btree.insert(12, 12 * 2).unwrap();
+        btree.insert(6, 6 * 2).unwrap();
+        btree.insert(2, 2 * 2).unwrap();
+        btree.insert(16, 16 * 2).unwrap();
+
+        assert_eq!(btree.values().unwrap(), vec![24, 12, 4, 32]);
+
+        let value = btree.get(&2).unwrap().unwrap();
+        assert_eq!(value, 4)
+    }
+}

src/collection/storage.rs

@@ -154,13 +154,7 @@ impl Storage {
         let mut file = fs::OpenOptions::new()
             .read(true)
             .open(self.storage_path.join(filename.clone()))
-            .map_err(|r| match r.kind() {
-                std::io::ErrorKind::NotFound => Error::CorruptedData(format!(
-                    "Data chunk {} not found, but index contains it",
-                    filename
-                )),
-                _ => Error::IoError(r),
-            })?;
+            .map_err(Error::IoError)?;
 
         Ok(Some(Self::deserialize_value(&mut file, offset, &filename)?))
     }
@@ -200,12 +194,7 @@ impl Storage {
         let mut value = vec![0; length];
         file.read_exact(&mut value).map_err(Error::IoError)?;
 
-        let value = bincode::deserialize(&value).map_err(|e| {
-            Error::CorruptedData(format!(
-                "Corrupted data chunk {} and offset {}. Error: {}",
-                filename, offset, e
-            ))
-        })?;
+        let value = bincode::deserialize(&value).map_err(Error::SerializeError)?;
 
         Ok(value)
     }

src/collection/wal.rs

@@ -253,13 +253,7 @@ impl Wal {
         let mut file = fs::OpenOptions::new()
             .read(true)
             .open(self.config.data_path.join("log").join(&filename))
-            .map_err(|r| match r.kind() {
-                std::io::ErrorKind::NotFound => Error::CorruptedData(format!(
-                    "WAL Log {} not found, but wal index contains it",
-                    filename
-                )),
-                _ => Error::IoError(r),
-            })?;
+            .map_err(Error::IoError)?;
 
         Self::deserialize_value(&mut file, offset, &filename)
     }
@@ -296,12 +290,7 @@ impl Wal {
         let mut value = vec![0; length];
         file.read_exact(&mut value).unwrap();
 
-        let value = bincode::deserialize(&value).map_err(|e| {
-            Error::CorruptedData(format!(
-                "Corrupted wal log {} and offset {}. Error: {}",
-                filename, offset, e
-            ))
-        })?;
+        let value = bincode::deserialize(&value).map_err(Error::SerializeError)?;
 
         Ok(Some(value))
     }

src/error.rs

@@ -7,7 +7,7 @@ pub enum Error {
     DatabaseLocked,
     AlreadyExists(String),
     NotFound(String),
-    CorruptedData(String),
+    CorruptedData(CorruptedDataError),
     Other(String),
     Cannot(String),
 }
@@ -21,7 +21,7 @@ impl Debug for Error {
                 write!(f, "Database is locked, maybe another instance is running?")
             }
             Error::Other(err) => write!(f, "Other Error: {}", err),
-            Error::CorruptedData(err) => write!(f, "Corrupted data: {}", err),
+            Error::CorruptedData(err) => write!(f, "{:?}", err),
             Error::AlreadyExists(message) => write!(f, "{} already exists", message),
             Error::NotFound(message) => write!(f, "{} not found", message),
             Error::Cannot(message) => write!(f, "cannot {}", message),
@@ -30,4 +30,21 @@ impl Debug for Error {
     }
 }
 
+pub struct CorruptedDataError {
+    pub kind: CorruptedDataKind,
+    pub message: String,
+}
+
+#[derive(Debug)]
+pub enum CorruptedDataKind {
+    ChecksumNotMatch,
+    UnsyncWithWAL,
+}
+
+impl Debug for CorruptedDataError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Corrupted data: {}. {:?}", self.message, self.kind)
+    }
+}
+
 pub type Result<T> = std::result::Result<T, Error>;

src/page.rs

@@ -5,9 +5,10 @@ use std::{
 };
 
 use crate::{
-    error::{Error, Result},
+    error::{CorruptedDataError, CorruptedDataKind, Error, Result},
     Either,
 };
+use crc32fast::Hasher;
 use layout::{CellPointerFlags, CellPointerMetadata, PageHeader};
 use serde::{de::DeserializeOwned, Serialize};
 use spec::{LocationOffset, CELL_POINTER_SIZE, PAGE_FREE_SPACE_BYTE, PAGE_HEADER_SIZE, PAGE_SIZE};
@@ -35,28 +36,45 @@ impl<T: Serialize + DeserializeOwned + PartialOrd + Ord + Clone> Page<T> {
             upper,
             lower,
             special,
+            checksum: 0,
         };
 
         let header_bytes = bincode::serialize(&header).map_err(Error::SerializeError)?;
         io.seek(SeekFrom::Start(0)).map_err(Error::IoError)?;
         io.write(&header_bytes).map_err(Error::IoError)?;
 
-        Ok(Page {
+        let mut page: Page<T> = Page {
             header,
             io,
             _t: PhantomData,
-        })
+        };
+
+        // update checksum
+        page.write_header()?;
+
+        Ok(page)
     }
 
     pub fn open(data: [u8; PAGE_SIZE as usize]) -> Result<Self> {
         let mut io = Cursor::new(data);
         let header = Self::read_header(&mut io)?;
 
-        Ok(Self {
+        let page = Self {
             io,
-            header,
+            header: header.clone(),
             _t: PhantomData,
-        })
+        };
+
+        let checksum = page.checksum();
+
+        if header.checksum != checksum {
+            return Err(Error::CorruptedData(CorruptedDataError {
+                kind: CorruptedDataKind::ChecksumNotMatch,
+                message: "checksum does not match".to_string(),
+            }));
+        }
+
+        Ok(page)
     }
 
     pub fn write(&mut self, data: T) -> Result<(LocationOffset, LocationOffset)> {
@@ -442,6 +460,9 @@ impl<T: Serialize + DeserializeOwned + PartialOrd + Ord + Clone> Page<T> {
     }
 
     fn write_header(&mut self) -> Result<()> {
+        let checksum = self.checksum();
+        self.header.checksum = checksum;
+
         let buffer = bincode::serialize(&self.header).map_err(Error::SerializeError)?;
 
         self.io.seek(SeekFrom::Start(0)).map_err(Error::IoError)?;
@@ -458,6 +479,12 @@ impl<T: Serialize + DeserializeOwned + PartialOrd + Ord + Clone> Page<T> {
 
         bincode::deserialize(&buffer).map_err(Error::SerializeError)
     }
+
+    fn checksum(&self) -> u32 {
+        let mut hasher = Hasher::new();
+        hasher.update(&self.io.get_ref()[PAGE_HEADER_SIZE..]);
+        hasher.finalize()
+    }
 }
 
 #[cfg(test)]
@@ -589,4 +616,17 @@ mod page_tests {
 
         assert_eq!(value, 90);
     }
+
+    #[test]
+    fn page_checksum() {
+        let mut page = Page::<u32>::create(0).unwrap();
+
+        page.write(99).unwrap();
+
+        let mut page_bytes = page.to_bytes().unwrap();
+        //change a random byte
+        page_bytes[26] = 2u8;
+
+        Page::<u32>::open(page_bytes).err().unwrap();
+    }
 }

src/page/layout.rs

@@ -3,6 +3,7 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct PageHeader {
+    pub checksum: u32,
     pub lower: LocationOffset,
     pub upper: LocationOffset,
     pub special: LocationOffset,