Add length to set_big and validate it in get_big to detect corrupt bi…

…g items

NOTE: this is a breaking change, but I assume big items are not used in the field

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "candystore"
-version = "0.5.0"
+version = "0.5.1"
 edition = "2021"
 license = "Apache-2.0"
 keywords = ["key-value", "database", "persistent", "store", "rocksdb"]

src/store.rs

@@ -1,4 +1,5 @@
 use anyhow::{anyhow, bail, ensure};
+use bytemuck::{bytes_of, from_bytes};
 use fslock::LockFile;
 use parking_lot::Mutex;
 use std::{
@@ -531,24 +532,28 @@ impl CandyStore {
     ) -> Result<bool> {
         let existed = self.discard_queue(key)?;
         self.extend_queue(key, val.as_ref().chunks(MAX_VALUE_SIZE))?;
+        self.push_to_queue_tail(key, bytes_of(&val.as_ref().len()))?;
         Ok(existed)
     }
 
     /// Returns a big item, collecting all the underlying chunks into a single value that's returned to the
     /// caller.
     pub fn get_big(&self, key: &[u8]) -> Result<Option<Vec<u8>>> {
         let mut val = vec![];
-        let mut exists = false;
+        let range = self.queue_range(key)?;
         for res in self.iter_queue(key) {
-            let (_, chunk) = res?;
-            exists = true;
-            val.extend_from_slice(&chunk);
-        }
-        if exists {
-            Ok(Some(val))
-        } else {
-            Ok(None)
+            let (idx, chunk) = res?;
+            // last element should encode the byte length of the item - if it's missing or encodes a different length,
+            // consider it corrupt and ignore this element
+            if idx + 1 == range.end {
+                if chunk.len() == size_of::<usize>() && *from_bytes::<usize>(&chunk) == val.len() {
+                    return Ok(Some(val));
+                }
+            } else {
+                val.extend_from_slice(&chunk);
+            }
         }
+        Ok(None)
     }
 
     /// Removes a big item by key. Returns true if the key had existed, false otherwise.