experiment: Add compression to archive

Cargo.toml

@@ -15,6 +15,7 @@ base64.workspace = true
 bech32.workspace = true
 bincode.workspace = true
 chrono.workspace = true
+flate2.workspace = true
 futures-core.workspace = true
 futures-util.workspace = true
 hex.workspace = true
@@ -61,7 +62,6 @@ comfy-table = { version = "7.1.1", optional = true }
 inquire = { version = "0.7.5", optional = true }
 toml = { version = "0.8.13", optional = true }
 console-subscriber = { version = "0.3.0", optional = true }
-flate2 = "1.0.34"
 tar = "0.4.41"
 paste = "1.0.15"
 
@@ -147,6 +147,7 @@ hex = "0.4.3"
 tracing = "0.1.37"
 itertools = "0.13.0"
 bincode = "1.3.3"
+flate2 = { version = "1.1.4", features = ["zlib-rs"]}
 futures-core = "0.3.31"
 futures-util = "0.3.28"
 trait-variant = "0.1.2"

crates/redb3/Cargo.toml

@@ -15,6 +15,7 @@ bincode.workspace = true
 hex.workspace = true
 thiserror.workspace = true
 xxhash-rust.workspace = true
+flate2.workspace = true
 
 dolos-core = { path = "../core" }
 

crates/redb3/src/archive/mod.rs

@@ -81,6 +81,12 @@ impl From<::redb::TransactionError> for RedbArchiveError {
     }
 }
 
+impl From<std::io::Error> for RedbArchiveError {
+    fn from(value: std::io::Error) -> Self {
+        Self(ArchiveError::InternalError(value.to_string()))
+    }
+}
+
 const DEFAULT_CACHE_SIZE_MB: usize = 500;
 
 #[derive(Clone)]

crates/redb3/src/archive/tables.rs

@@ -1,33 +1,56 @@
+use std::io::{Read, Write};
+
+use pallas::ledger::primitives::byron::Block;
 use redb::{Range, ReadTransaction, ReadableTable as _, TableDefinition, WriteTransaction};
+use flate2::{read::ZlibDecoder, write::ZlibEncoder};
+use flate2::Compression;
 use tracing::trace;
 
 use dolos_core::{BlockBody, BlockSlot, ChainPoint, RawBlock};
 
 type Error = super::RedbArchiveError;
 
+type CompressedBlockBody = Vec<u8>;
+
 pub struct BlocksTable;
 
 impl BlocksTable {
-    pub const DEF: TableDefinition<'static, BlockSlot, BlockBody> = TableDefinition::new("blocks");
+    pub const DEF: TableDefinition<'static, BlockSlot, CompressedBlockBody> = TableDefinition::new("blocks");
 
     pub fn initialize(wx: &WriteTransaction) -> Result<(), Error> {
         wx.open_table(Self::DEF)?;
 
         Ok(())
     }
 
+    fn compress(block: BlockBody) -> std::io::Result<CompressedBlockBody> {
+        let mut e = ZlibEncoder::new(Vec::new(), Compression::best());
+        e.write_all(block.as_slice())?;
+        e.finish()
+    }
+
+    fn decompress(compressed: CompressedBlockBody) -> std::io::Result<BlockBody> {
+        let mut z = ZlibDecoder::new(compressed.as_slice());
+        let mut result = Vec::new();
+        z.read_to_end(&mut result)?;
+        Ok(result)
+    }
+
     pub fn get_tip(rx: &ReadTransaction) -> Result<Option<(BlockSlot, BlockBody)>, Error> {
         let table = rx.open_table(Self::DEF)?;
         let result = table
             .last()?
-            .map(|(slot, raw)| (slot.value(), raw.value().clone()));
-        Ok(result)
+            .map(|(slot, raw)| 
+                Ok((slot.value(), BlocksTable::decompress(raw.value().clone())?)));
+        result.transpose()
     }
 
     pub fn get_by_slot(rx: &ReadTransaction, slot: BlockSlot) -> Result<Option<BlockBody>, Error> {
         let table = rx.open_table(Self::DEF)?;
         match table.get(slot)? {
-            Some(value) => Ok(Some(value.value().clone())),
+            Some(value) => {
+                Ok(Some(BlocksTable::decompress(value.value().clone())?))
+            },
             None => Ok(None),
         }
     }
@@ -36,7 +59,7 @@
         let mut table = wx.open_table(Self::DEF)?;
 
         let slot = point.slot();
-        table.insert(slot, block.clone())?;
+        table.insert(slot, BlocksTable::compress(block.to_vec())?)?;
 
         Ok(())
     }
@@ -66,8 +89,8 @@
         let table = rx.open_table(Self::DEF)?;
         let result = table
             .first()?
-            .map(|(slot, raw)| (slot.value(), raw.value().clone()));
-        Ok(result)
+            .map(|(slot, raw)| Ok((slot.value(), BlocksTable::decompress(raw.value().clone())?)));
+        result.transpose()
     }
 
     pub fn last(rx: &ReadTransaction) -> Result<Option<(BlockSlot, BlockBody)>, Error> {