feat: compress ledger execution details with dictionary

ledger/README.md

@@ -16,8 +16,13 @@ superblock-000000001/
 
 `blockstore.db` is a wincode stream. Blockstore decoding permits allocations up
 to the ledger's 25-bit encoded entry-size bound (33,554,431 bytes); larger
-entries are rejected. Execution headers and zstd-compressed bitcode details are
-stored separately in `executions.db`.
+entries are rejected. Wincode-encoded execution headers and bitcode details
+compressed at Zstd level 3 with the embedded dictionary are both stored in
+`executions.db`.
+Frames omit Zstd dictionary IDs; the version stored first in each
+`superblock.meta` selects the storage format, including its execution-details
+codec. The current format is version 1. Changing the dictionary or codec
+requires a ledger-version bump and explicit compatibility handling.
 
 ## Append and read paths
 

ledger/src/appender.rs

@@ -17,7 +17,7 @@ use wincode::Error;
 use zstd::bulk::Compressor;
 
 use crate::{
-    Ledger, Superblock,
+    Ledger, Superblock, codec,
     error::{LedgerError, Result},
     index::{IndexWriter, Span, TxSpan},
     metrics::{self, Operation},
@@ -120,7 +120,8 @@ impl LedgerAppender {
     fn rotate(&mut self, seal: SuperblockSeal) -> Result<()> {
         let _timer = metrics::time(Operation::Rotate);
         let head = seal.id + 1;
-        let superblock = Superblock::open(&self.ledger.directory, head, &self.ledger.index)?;
+        let meta = Superblock::open_meta(&self.ledger.directory, head)?;
+        let superblock = Superblock::open(meta, &self.ledger.index)?;
         // Seal N opens N+1, which stores N's snapshot archive and seal metadata.
         superblock.meta.checksum.store(seal.checksum, Release);
         superblock.meta.transactions.store(seal.transactions, Release);
@@ -302,7 +303,7 @@ impl SuperblockWriter {
                 &superblock.meta.cursors.executions,
             )?,
             superblock,
-            compressor: Compressor::new(0)?,
+            compressor: codec::compressor()?,
             buffer: Buffer::new(),
         })
     }

ledger/src/codec.rs

@@ -0,0 +1,26 @@
+//! Execution-details compression codec.
+
+use std::io;
+
+use zstd::{
+    bulk::{Compressor, Decompressor},
+    zstd_safe::CParameter,
+};
+
+const COMPRESSION_LEVEL: i32 = 3;
+const DICTIONARY: &[u8] = include_bytes!(concat!(
+    env!("CARGO_MANIFEST_DIR"),
+    "/../assets/execution-details.dict"
+));
+
+/// Creates a reusable compressor for execution-details frames.
+pub(crate) fn compressor() -> io::Result<Compressor<'static>> {
+    let mut compressor = Compressor::with_dictionary(COMPRESSION_LEVEL, DICTIONARY)?;
+    compressor.set_parameter(CParameter::DictIdFlag(false))?;
+    Ok(compressor)
+}
+
+/// Creates a reusable decompressor for execution-details frames.
+pub(crate) fn decompressor() -> io::Result<Decompressor<'static>> {
+    Decompressor::with_dictionary(DICTIONARY)
+}

ledger/src/error.rs

@@ -6,6 +6,8 @@ use agave_transaction_view::result::TransactionViewError;
 use oneshot::RecvError;
 use tokio::time::error::Elapsed;
 
+use crate::LedgerVersion;
+
 /// Errors returned by ledger storage, codecs, and indexes.
 #[derive(Debug, derive_more::From, thiserror::Error)]
 pub enum LedgerError {
@@ -31,6 +33,9 @@ pub enum LedgerError {
     #[error("ledger corruption: {0}")]
     #[from(skip)]
     Corruption(&'static str),
+    /// Opened superblock uses an unsupported on-disk format version.
+    #[error("unsupported ledger version: {0}")]
+    UnsupportedVersion(LedgerVersion),
     /// The background superblock cleanup worker panicked.
     #[error("ledger truncation worker panicked")]
     TruncationPanic,

ledger/src/lib.rs

@@ -21,6 +21,7 @@ use tokio::sync::broadcast;
 use tracing::info;
 
 mod appender;
+mod codec;
 mod error;
 mod index;
 mod metrics;
@@ -45,6 +46,10 @@ use crate::{
 const LEDGER_META: &str = "ledger.meta";
 const APPENDER_QUEUE_CAPACITY: usize = 2_048;
 const READER_QUEUE_CAPACITY: usize = 128;
+/// Current on-disk superblock format version.
+const VERSION: LedgerVersion = 1;
+/// Version tag stored at the start of every superblock metadata header.
+pub type LedgerVersion = u64;
 
 /// Top-level ledger handle.
 ///
@@ -147,17 +152,20 @@ impl Ledger {
     /// Opens ledger metadata and retained superblocks without starting services.
     fn new(directory: PathBuf, size_limit: u64) -> Result<Self> {
         fs::create_dir_all(&directory)?;
-        let index = Index::new(&directory)?;
         let meta = directory.join(LEDGER_META);
         // SAFETY: `LedgerMeta` and its nested headers have stable C layouts,
         // and all fields that can change while mapped are atomic. This process
         // exclusively creates and updates the metadata file at `meta`.
         let meta = unsafe { MetaMap::<LedgerMeta>::new(&meta) }?;
-        let retained = meta.superblocks();
+        let retained = meta
+            .superblocks()
+            .map(|id| Superblock::open_meta(&directory, id))
+            .collect::<Result<Vec<_>>>()?;
+        let index = Index::new(&directory)?;
         let mut superblocks = BTreeMap::new();
-        for id in retained {
-            let superblock = Superblock::open(&directory, id, &index)?;
-            superblocks.insert(id, superblock);
+        for meta in retained {
+            let id = meta.id;
+            superblocks.insert(id, Superblock::open(meta, &index)?);
         }
 
         info!(?directory, superblocks = superblocks.len(), "opened ledger");
@@ -268,6 +276,13 @@ pub struct Superblock {
     pub directory: PathBuf,
 }
 
+/// Mapped superblock metadata that passed format-version validation.
+struct ValidatedSuperblockMeta {
+    id: u64,
+    directory: PathBuf,
+    meta: MetaMap<SuperblockMeta>,
+}
+
 impl Superblock {
     /// Canonical directory and keyspace name for one superblock.
     fn name(id: u64) -> String {
@@ -290,11 +305,21 @@ impl Superblock {
         self.meta.transactions.load(Acquire)
     }
 
-    /// Opens a superblock directory, creating its data files when needed.
-    fn open(root: &Path, id: u64, index: &Index) -> Result<Arc<Self>> {
+    /// Maps and validates a superblock's metadata before opening shared storage.
+    fn open_meta(root: &Path, id: u64) -> Result<ValidatedSuperblockMeta> {
         let directory = Self::init_dir(root, id)?;
-        let index = index.keyspace(id)?;
         let meta = unsafe { MetaMap::<SuperblockMeta>::new(&directory.join(SUPERBLOCK_META)) }?;
+        if meta.version != VERSION {
+            return Err(LedgerError::UnsupportedVersion(meta.version));
+        }
+
+        Ok(ValidatedSuperblockMeta { id, directory, meta })
+    }
+
+    /// Opens a superblock's keyspace and data files from validated metadata.
+    fn open(validated: ValidatedSuperblockMeta, index: &Index) -> Result<Arc<Self>> {
+        let ValidatedSuperblockMeta { id, directory, meta } = validated;
+        let index = index.keyspace(id)?;
         let blockstore = Self::file(&directory.join(BLOCKSTORE_DB))?;
         let executions = Self::file(&directory.join(EXECUTIONS_DB))?;
 

ledger/src/reader.rs

@@ -22,7 +22,7 @@ use wincode::{Error, io::Cursor};
 use zstd::bulk::Decompressor;
 
 use crate::{
-    Ledger, LedgerError, Result, Superblock,
+    Ledger, LedgerError, Result, Superblock, codec,
     index::{IndexReader, Span},
     metrics::{self, Operation},
     request::{
@@ -59,7 +59,7 @@ impl LedgerReader {
         Ok(Self {
             rx,
             ledger,
-            decompressor: Decompressor::new()?,
+            decompressor: codec::decompressor()?,
             buffers,
         })
     }

ledger/src/storage.rs

@@ -22,7 +22,7 @@ use rustix::fs::{self, FallocateFlags};
 use tracing::debug;
 use zstd::bulk::Compressor;
 
-use crate::{Result, index::Span, schema::MAX_ENTRY_SIZE};
+use crate::{LedgerVersion, Result, VERSION, index::Span, schema::MAX_ENTRY_SIZE};
 
 /// Initial buffer size for append-heavy files.
 const FILE_BUFFER_SIZE: usize = 64 * MB;
@@ -293,9 +293,10 @@ impl LedgerMeta {
 }
 
 /// Metadata header for one superblock directory.
-#[derive(Default)]
 #[repr(C)]
 pub(crate) struct SuperblockMeta {
+    /// Immutable on-disk format version for this superblock.
+    pub(crate) version: LedgerVersion,
     /// Published append cursors for files in this superblock.
     pub(crate) cursors: FileCursors,
     /// Slot range stored in this segment.
@@ -306,6 +307,18 @@ pub(crate) struct SuperblockMeta {
     pub(crate) transactions: AtomicU64,
 }
 
+impl Default for SuperblockMeta {
+    fn default() -> Self {
+        Self {
+            version: VERSION,
+            cursors: Default::default(),
+            range: Default::default(),
+            checksum: 0.into(),
+            transactions: 0.into(),
+        }
+    }
+}
+
 /// Published append cursors for superblock data files.
 #[derive(Default)]
 #[repr(C)]

ledger/src/tests/codec.rs

@@ -0,0 +1,55 @@
+//! Execution-details codec tests.
+
+use std::sync::Arc;
+
+use bitcode::Buffer;
+use zstd::zstd_safe::get_dict_id_from_frame;
+
+use crate::{
+    codec::{compressor, decompressor},
+    schema::{Balances, CompiledInstruction, Cpis, ExecutionDetails, Instruction, ReturnData},
+};
+
+/// Proves frames omit dictionary IDs and matching Zstd/bitcode contexts round-trip.
+#[test]
+fn execution_details_dictionary_roundtrip() {
+    let details = Some(ExecutionDetails {
+        fee: 5_000,
+        balances: Balances {
+            pre: vec![10_000, 20_000],
+            post: vec![9_000, 21_000],
+        },
+        logs: Arc::new(vec![
+            "Program log: Instruction: Transfer".into(),
+            "Program consumed 150 of 200000 compute units".into(),
+        ]),
+        cpi: Some(vec![Cpis(vec![Instruction {
+            compiled: CompiledInstruction {
+                program_index: 2,
+                accounts: vec![0, 1],
+                data: vec![3, 4, 5],
+            },
+            stack_height: 2,
+        }])]),
+        compute_units: 150,
+        return_data: Some(ReturnData {
+            program: [7; 32],
+            data: Arc::new(vec![8, 9]),
+        }),
+    });
+
+    let mut encoder = Buffer::new();
+    let encoded = encoder.encode(&details).to_vec();
+    let compressed = compressor().unwrap().compress(&encoded).unwrap();
+    assert_eq!(get_dict_id_from_frame(&compressed), None);
+
+    let decoded = decompressor().unwrap().decompress(&compressed, encoded.len()).unwrap();
+    let mut decoder = Buffer::new();
+    let details: Option<ExecutionDetails> = decoder.decode(&decoded).unwrap();
+    let details = details.expect("execution details decoded");
+    assert_eq!(details.fee, 5_000);
+    assert_eq!(details.balances.post, [9_000, 21_000]);
+    assert_eq!(details.logs.len(), 2);
+    assert_eq!(details.cpi.unwrap()[0].0[0].stack_height, 2);
+    assert_eq!(details.return_data.unwrap().data.as_slice(), &[8, 9]);
+}

ledger/src/tests/mod.rs

@@ -1,7 +1,9 @@
 //! Ledger test modules.
 //!
-//! `index` covers the Fjall codec/index in isolation; `integration` drives the
-//! append→seal→read pipeline end to end through the appender and reader.
+//! `codec` covers serialization and compression, `index` covers Fjall storage,
+//! and `integration` drives the append→seal→read pipeline end to end through
+//! the appender and reader.
 
+mod codec;
 mod index;
 mod integration;