[fix/block-offset-overflow] Merge to develop

src/core/engine.rs

@@ -1,9 +1,9 @@
 use crate::core::log_record::LogRecord;
 use crate::core::memtable::MemTable;
-use crate::infra::codec::decode;
 use crate::infra::config::LsmConfig;
 use crate::infra::error::{LsmError, Result};
-use crate::storage::sstable::SStable;
+use crate::storage::builder::SstableBuilder;
+use crate::storage::reader::SstableReader;
 use crate::storage::wal::WriteAheadLog;
 
 use std::collections::HashMap;
@@ -29,7 +29,7 @@ pub struct LsmStats {
 pub struct LsmEngine {
     pub(crate) memtable: Mutex<MemTable>,
     pub(crate) wal: WriteAheadLog,
-    pub(crate) sstables: Mutex<Vec<SStable>>,
+    pub(crate) sstables: Mutex<Vec<SstableReader>>,
     pub(crate) dir_path: PathBuf,
     pub(crate) config: LsmConfig,
 }
@@ -46,14 +46,15 @@ impl LsmEngine {
             let entry = entry?;
             let path = entry.path();
             if path.extension().is_some_and(|ext| ext == "sst") {
-                match SStable::load(&path) {
+                match SstableReader::open(path.clone(), config.storage.clone()) {
                     Ok(sst) => sstables.push(sst),
                     Err(e) => warn!("Failed to load SSTable {}: {}", path.display(), e),
                 }
             }
         }
 
-        sstables.sort_by(|a, b| b.metadata.timestamp.cmp(&a.metadata.timestamp));
+        // Sort by timestamp descending (newest first)
+        sstables.sort_by(|a, b| b.metadata().timestamp.cmp(&a.metadata().timestamp));
 
         let mut memtable = MemTable::new(config.core.memtable_max_size);
         for record in wal_records {
@@ -81,7 +82,7 @@ impl LsmEngine {
             .map_err(|_| LsmError::LockPoisoned("memtable"))
     }
 
-    fn sstables_lock(&self) -> Result<MutexGuard<'_, Vec<SStable>>> {
+    fn sstables_lock(&self) -> Result<MutexGuard<'_, Vec<SstableReader>>> {
         self.sstables
             .lock()
             .map_err(|_| LsmError::LockPoisoned("sstables"))
@@ -128,7 +129,7 @@ impl LsmEngine {
         }
         drop(memtable);
 
-        // 2. Verificar SSTables (da mais recente para a mais antiga)
+        // 2. Check SSTables (newest to oldest)
         let mut sstables = self.sstables_lock()?;
         for sst in sstables.iter_mut() {
             if let Some(record) = sst.get(key)? {
@@ -189,11 +190,21 @@ impl LsmEngine {
         }
 
         let timestamp = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
+        let filename = format!("{}.sst", timestamp);
+        let path = self.dir_path.join(filename);
 
-        let sst = SStable::create(&self.dir_path, timestamp, &self.config.storage, &records)?;
+        // Create new SSTable using Builder (V2)
+        let mut builder = SstableBuilder::new(path, self.config.storage.clone(), timestamp)?;
+        for (key, record) in records {
+            builder.add(key.as_bytes(), &record)?;
+        }
+        let sst_path = builder.finish()?;
+
+        // Open the new SSTable as Reader (V2)
+        let reader = SstableReader::open(sst_path, self.config.storage.clone())?;
 
         let mut sstables = self.sstables_lock()?;
-        sstables.insert(0, sst);
+        sstables.insert(0, reader);
         let cleared = memtable.clear();
 
         info!(
@@ -222,11 +233,12 @@ impl LsmEngine {
         }
         drop(memtable);
 
-        let sstables = self.sstables_lock()?;
-        for sst in sstables.iter() {
-            let records = self.read_all_from_sstable(sst)?;
-            for record in records {
-                result_map.entry(record.key.clone()).or_insert((
+        let mut sstables = self.sstables_lock()?;
+        for sst in sstables.iter_mut() {
+            let records = sst.scan()?;
+            for (key_bytes, record) in records {
+                let key = String::from_utf8(key_bytes).map_err(|e| LsmError::CorruptedData(e.to_string()))?;
+                result_map.entry(key).or_insert((
                     record.value,
                     record.timestamp,
                     record.is_deleted,
@@ -250,35 +262,6 @@ impl LsmEngine {
         Ok(results)
     }
 
-    fn read_all_from_sstable(&self, sst: &SStable) -> Result<Vec<LogRecord>> {
-        use std::fs::File;
-        use std::io::{BufReader, Read, Seek, SeekFrom};
-
-        let mut file = BufReader::new(File::open(&sst.path)?);
-        file.seek(SeekFrom::Current(8))?;
-
-        let mut len_buf = [0u8; 4];
-
-        file.read_exact(&mut len_buf)?;
-        let bloom_len = u32::from_le_bytes(len_buf) as usize;
-        file.seek(SeekFrom::Current(bloom_len as i64))?;
-
-        file.read_exact(&mut len_buf)?;
-        let meta_len = u32::from_le_bytes(len_buf) as usize;
-        file.seek(SeekFrom::Current(meta_len as i64))?;
-
-        let mut records = Vec::new();
-        for _ in 0..sst.metadata.record_count {
-            file.read_exact(&mut len_buf)?;
-            let record_len = u32::from_le_bytes(len_buf) as usize;
-            let mut record_data = vec![0u8; record_len];
-            file.read_exact(&mut record_data)?;
-            let record: LogRecord = decode(&record_data)?;
-            records.push(record);
-        }
-        Ok(records)
-    }
-
     pub fn keys(&self) -> Result<Vec<String>> {
         let all_data = self.scan()?;
         Ok(all_data.into_iter().map(|(k, _)| k).collect())
@@ -313,12 +296,12 @@ impl LsmEngine {
         let mem_records = memtable.data.len();
         let sst_records_total: u64 = sstables
             .iter()
-            .map(|s| s.metadata.record_count as u64)
+            .map(|s| s.metadata().record_count)
             .sum();
 
         let sst_bytes_total: u64 = sstables
             .iter()
-            .map(|s| std::fs::metadata(&s.path).map(|m| m.len()).unwrap_or(0))
+            .map(|s| std::fs::metadata(s.path()).map(|m| m.len()).unwrap_or(0))
             .sum();
 
         let wal_bytes: u64 = std::fs::metadata(&self.wal.path)

src/storage/block.rs

@@ -2,12 +2,12 @@ use crate::infra::config::StorageConfig;
 use std::mem::size_of;
 
 pub const BLOCK_SIZE: usize = 4096;
-const U16_SIZE: usize = size_of::<u16>();
+const U32_SIZE: usize = size_of::<u32>();
 
 #[derive(Debug, Clone)]
 pub struct Block {
     pub(crate) data: Vec<u8>,
-    pub(crate) offsets: Vec<u16>,
+    pub(crate) offsets: Vec<u32>,
     block_size: usize,
 }
 
@@ -25,11 +25,15 @@ impl Block {
     }
 
     fn entry_size(key: &[u8], value: &[u8]) -> usize {
-        U16_SIZE + key.len() + U16_SIZE + value.len()
+        // KeyLen(2) + Key + ValLen(2) + Value
+        // Note: Using u16 for key/value length storage within the block data
+        // to maintain compactness for individual entries, while allowing
+        // the overall block to be larger than 64KB via u32 offsets.
+        2 + key.len() + 2 + value.len()
     }
 
     fn metadata_size(num_entries: usize) -> usize {
-        (num_entries * U16_SIZE) + U16_SIZE
+        (num_entries * U32_SIZE) + U32_SIZE
     }
 
     fn current_size(&self) -> usize {
@@ -38,16 +42,18 @@ impl Block {
 
     pub fn add(&mut self, key: &[u8], value: &[u8]) -> bool {
         let entry_size = Self::entry_size(key, value);
-        let new_offset_size = U16_SIZE;
+        let new_offset_size = U32_SIZE;
         let total_needed = self.current_size() + entry_size + new_offset_size;
 
         if total_needed > self.block_size {
             return false;
         }
 
-        let offset = self.data.len() as u16;
+        let offset = self.data.len() as u32;
         self.offsets.push(offset);
 
+        // Cast to u16 is safe for key/value lengths as we assume
+        // individual entries don't exceed 64KB, even if the block does.
         let key_len = key.len() as u16;
         let val_len = value.len() as u16;
 
@@ -67,35 +73,35 @@ impl Block {
             encoded.extend_from_slice(&offset.to_le_bytes());
         }
 
-        let num_elements = self.offsets.len() as u16;
+        let num_elements = self.offsets.len() as u32;
         encoded.extend_from_slice(&num_elements.to_le_bytes());
 
         encoded
     }
 
     pub fn decode(data: &[u8]) -> Self {
-        if data.len() < U16_SIZE {
+        if data.len() < U32_SIZE {
             return Self {
                 data: Vec::new(),
                 offsets: Vec::new(),
                 block_size: BLOCK_SIZE,
             };
         }
 
-        let num_elements_start = data.len() - U16_SIZE;
+        let num_elements_start = data.len() - U32_SIZE;
         let num_elements =
-            u16::from_le_bytes([data[num_elements_start], data[num_elements_start + 1]]) as usize;
+            u32::from_le_bytes([data[num_elements_start], data[num_elements_start + 1], data[num_elements_start + 2], data[num_elements_start + 3]]) as usize;
 
-        let offsets_start = data.len() - U16_SIZE - (num_elements * U16_SIZE);
+        let offsets_start = data.len() - U32_SIZE - (num_elements * U32_SIZE);
         let records_data = data[..offsets_start].to_vec();
 
         let mut offsets = Vec::with_capacity(num_elements);
         let mut offset_pos = offsets_start;
 
         for _ in 0..num_elements {
-            let offset = u16::from_le_bytes([data[offset_pos], data[offset_pos + 1]]);
+            let offset = u32::from_le_bytes([data[offset_pos], data[offset_pos + 1], data[offset_pos + 2], data[offset_pos + 3]]);
             offsets.push(offset);
-            offset_pos += U16_SIZE;
+            offset_pos += U32_SIZE;
         }
 
         Self {
@@ -178,6 +184,41 @@ mod tests {
         assert!(!result, "Should reject entry when block is full");
     }
 
+    #[test]
+    fn test_block_overflow_u16() {
+        // Create a block larger than 64KB (u16::MAX is 65535)
+        let block_size = 70_000;
+        let mut block = Block::new(block_size);
+
+        // Fill with enough data to exceed 64KB
+        // Each entry approx 1024 bytes
+        let val_size = 1000;
+        let large_value = vec![b'x'; val_size];
+        let key_base = "key";
+
+        let mut count = 0;
+        while block.data_size() < 66000 {
+             let key = format!("{}{}", key_base, count);
+             if !block.add(key.as_bytes(), &large_value) {
+                 break;
+             }
+             count += 1;
+        }
+
+        assert!(block.data_size() > 65535, "Block data size should be > 64KB");
+
+        // Verify integrity
+        let encoded = block.encode();
+        let decoded = Block::decode(&encoded);
+
+        assert_eq!(decoded.len(), block.len());
+        assert_eq!(decoded.offsets.len(), block.offsets.len());
+
+        // Verify last entry is correct
+        let last_offset = *decoded.offsets.last().unwrap();
+        assert!(last_offset > 65535, "Last offset should exceed u16 limit");
+    }
+
     #[test]
     fn test_overflow_large_entry() {
         let mut block = Block::new(128);

src/storage/builder.rs

@@ -10,7 +10,7 @@ use std::fs::File;
 use std::io::{BufWriter, Write};
 use std::path::PathBuf;
 
-const SST_MAGIC_V2: &[u8; 8] = b"LSMSST02";
+const SST_MAGIC_V2: &[u8; 8] = b"LSMSST03";
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BlockMeta {

src/storage/config.rs

@@ -1,8 +1,18 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum CompactionStrategy {
+    SizeTiered,
+    Leveled,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct StorageConfig {
     pub block_size: usize,
     pub block_cache_size_mb: usize,
     pub sparse_index_interval: usize,
     pub compaction_strategy: CompactionStrategy,
+    pub bloom_false_positive_rate: f64,
 }
 
 impl Default for StorageConfig {
@@ -12,13 +22,7 @@ impl Default for StorageConfig {
             block_cache_size_mb: 64,
             sparse_index_interval: 16,
             compaction_strategy: CompactionStrategy::SizeTiered,
+            bloom_false_positive_rate: 0.01,
         }
     }
 }
-
-// src/core/engine.rs
-pub struct LsmConfig {
-    pub dir_path: PathBuf,
-    pub memtable_max_size: usize,
-    pub storage: StorageConfig, // ✅ Composição
-}

src/storage/mod.rs

@@ -1,5 +1,5 @@
 pub mod block;
 pub mod builder;
+pub mod config;
 pub mod reader;
-pub mod sstable;
 pub mod wal;

src/storage/reader.rs

@@ -12,7 +12,7 @@ use std::io::{Read, Seek, SeekFrom};
 use std::num::NonZeroUsize;
 use std::path::PathBuf;
 
-const SST_MAGIC_V2: &[u8; 8] = b"LSMSST02";
+const SST_MAGIC_V2: &[u8; 8] = b"LSMSST03";
 const FOOTER_SIZE: u64 = 8;
 
 /// SSTable V2 Reader with sparse index, Bloom filter, and block caching