diff --git a/assets/execution-details.dict b/assets/execution-details.dict new file mode 100644 index 0000000..031b86a Binary files /dev/null and b/assets/execution-details.dict differ diff --git a/ledger/README.md b/ledger/README.md index 04b6565..e4fef21 100644 --- a/ledger/README.md +++ b/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 diff --git a/ledger/src/appender.rs b/ledger/src/appender.rs index 74e0a1b..ff4b8ee 100644 --- a/ledger/src/appender.rs +++ b/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(), }) } diff --git a/ledger/src/codec.rs b/ledger/src/codec.rs new file mode 100644 index 0000000..39f34d6 --- /dev/null +++ b/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> { + 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::with_dictionary(DICTIONARY) +} diff --git a/ledger/src/error.rs b/ledger/src/error.rs index 28ac3be..82f07d3 100644 --- a/ledger/src/error.rs +++ b/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, diff --git a/ledger/src/lib.rs b/ledger/src/lib.rs index 36971b1..d5c0453 100644 --- a/ledger/src/lib.rs +++ b/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 { 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::::new(&meta) }?; - let retained = meta.superblocks(); + let retained = meta + .superblocks() + .map(|id| Superblock::open_meta(&directory, id)) + .collect::>>()?; + 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, +} + 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> { + /// Maps and validates a superblock's metadata before opening shared storage. + fn open_meta(root: &Path, id: u64) -> Result { let directory = Self::init_dir(root, id)?; - let index = index.keyspace(id)?; let meta = unsafe { MetaMap::::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> { + 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))?; diff --git a/ledger/src/reader.rs b/ledger/src/reader.rs index d324535..c22e0b2 100644 --- a/ledger/src/reader.rs +++ b/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, }) } diff --git a/ledger/src/storage.rs b/ledger/src/storage.rs index d5ed950..8efc7a9 100644 --- a/ledger/src/storage.rs +++ b/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)] diff --git a/ledger/src/tests/codec.rs b/ledger/src/tests/codec.rs new file mode 100644 index 0000000..5159f65 --- /dev/null +++ b/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 = 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]); +} diff --git a/ledger/src/tests/mod.rs b/ledger/src/tests/mod.rs index 95c6033..9f8bd4f 100644 --- a/ledger/src/tests/mod.rs +++ b/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;