use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::mpsc;
use chia_protocol::Bytes32;
use dig_block::{AttestedBlock, BlockStatus, L2Block, L2BlockHeader};
use parking_lot::{Mutex, RwLock};
use rocksdb::{Direction, IteratorMode, Options, WriteBatch, DB};
use crate::cache::sharded::{ShardedBlockCache, ShardedHeaderCache, ShardedLruCache};
use crate::canonical::mmap::CanonicalBin;
use crate::cf_options;
use crate::compression::resolve_zstd_dictionary;
use crate::constants::{
CF_ATTESTED, CF_BLOCKS, CF_CANONICAL, CF_CHECKPOINTS, CF_HEADERS, CF_METADATA,
META_GENESIS_HASH, META_MIN_HEIGHT, META_TIP,
};
use crate::encoding::{hash_key, height_key};
use crate::error::{
BlockStoreError, ERR_ASYNC_JOIN_PREFIX, ERR_INIT_GENESIS_ALREADY_INITIALIZED,
ERR_INIT_GENESIS_READ_ONLY, ERR_MUTATION_READ_ONLY, ERR_OPEN_READONLY_PATH_MISSING_PREFIX,
ERR_UPDATE_STATUS_RECORD_NOT_CACHED_PREFIX,
};
use crate::pipeline::PipelineJob;
use crate::types::{BlockRecord, ChainTip, ReorgResult, StorageStats};
use crate::BlockStoreConfig;
pub use crate::pipeline::StreamBlocksInRange;
#[doc(hidden)]
pub struct BlockStoreInner {
pub(crate) db: Arc<DB>,
pub(crate) read_only: bool,
pub(crate) tip: RwLock<Option<ChainTip>>,
pub(crate) warm_blocks_loaded: AtomicUsize,
pub(crate) compression_level: i32,
pub(crate) use_compression_dict: bool,
pub(crate) max_decompressed_block_bytes: usize,
pub(crate) zstd_dict: RwLock<Option<Arc<Vec<u8>>>>,
pub(crate) record_cache: Mutex<HashMap<Bytes32, BlockRecord>>,
pub(crate) block_cache: Arc<ShardedBlockCache>,
pub(crate) cf_blocks_physical_gets: AtomicUsize,
pub(crate) cf_blocks_multi_get_batches: AtomicUsize,
pub(crate) cf_blocks_stream_physical_gets: AtomicUsize,
pub(crate) readahead_size: usize,
pub(crate) header_cache: Arc<ShardedHeaderCache>,
pub(crate) cf_headers_physical_gets: AtomicUsize,
pub(crate) pipeline_batch_size: usize,
pub(crate) pipeline_flush_ms: u64,
pub(crate) pipeline_channel_capacity: usize,
pub(crate) pipeline_write_batches: AtomicUsize,
pub(crate) canonical_bin: RwLock<CanonicalBin>,
pub(crate) min_retained_height_cached: Arc<AtomicU64>,
pub(crate) canonical_height_cache: RwLock<std::collections::BTreeMap<u64, Bytes32>>,
pub(crate) canonical_height_cache_capacity: usize,
pub(crate) hash_to_height_cache: Arc<ShardedLruCache<u64>>,
}
pub struct BlockStore {
pub(crate) inner: Arc<BlockStoreInner>,
pub(crate) pipeline_tx: Arc<tokio::sync::Mutex<Option<mpsc::Sender<PipelineJob>>>>,
}
impl Clone for BlockStore {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
pipeline_tx: self.pipeline_tx.clone(),
}
}
}
impl std::ops::Deref for BlockStore {
type Target = BlockStoreInner;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl BlockStore {
pub fn open(config: BlockStoreConfig) -> Result<Self, BlockStoreError> {
let compression_level = config.compression_level;
let use_compression_dict = config.use_compression_dict;
let max_decompressed_block_bytes = config.max_decompressed_block_bytes;
let zstd_dictionary_override = config.zstd_dictionary_override.clone();
std::fs::create_dir_all(&config.path).map_err(|e| {
BlockStoreError::Serialization(format!(
"filesystem error creating database directory {}: {e}",
config.path.display()
))
})?;
let mut opts = Options::default();
opts.create_if_missing(true);
opts.create_missing_column_families(true);
let prune_threshold = if config.enable_compaction_pruning {
Some(Arc::new(AtomicU64::new(0)))
} else {
None
};
let cfs = cf_options::column_family_descriptors(&config, prune_threshold.clone());
let db = DB::open_cf_descriptors(&opts, &config.path, cfs)?;
let db = Arc::new(db);
let canonical_bin =
RwLock::new(CanonicalBin::open_synced(&db, config.path.as_path(), true)?);
let zstd_dict =
resolve_zstd_dictionary(&db, use_compression_dict, zstd_dictionary_override)?;
let tip = load_tip(&db)?;
let warm_cache_on_open = config.warm_cache_on_open;
let warm_cache_depth = config.warm_cache_depth;
let readahead_size = config.readahead_size;
let shards = config.cache_shards.max(1);
let block_cache = Arc::new(ShardedBlockCache::new(config.block_cache_capacity, shards));
let header_cache = Arc::new(ShardedHeaderCache::new(
config.header_cache_capacity,
shards,
));
let store = Self {
inner: Arc::new(BlockStoreInner {
db,
read_only: false,
tip: RwLock::new(tip),
warm_blocks_loaded: AtomicUsize::new(0),
compression_level,
use_compression_dict,
max_decompressed_block_bytes,
zstd_dict: RwLock::new(zstd_dict),
record_cache: Mutex::new(HashMap::new()),
block_cache,
cf_blocks_physical_gets: AtomicUsize::new(0),
cf_blocks_multi_get_batches: AtomicUsize::new(0),
cf_blocks_stream_physical_gets: AtomicUsize::new(0),
readahead_size,
header_cache,
cf_headers_physical_gets: AtomicUsize::new(0),
pipeline_batch_size: config.write_pipeline_batch_size.max(1),
pipeline_flush_ms: config.write_pipeline_flush_ms.max(1),
pipeline_channel_capacity: config.write_pipeline_channel_capacity.max(1),
pipeline_write_batches: AtomicUsize::new(0),
canonical_bin,
min_retained_height_cached: prune_threshold
.unwrap_or_else(|| Arc::new(AtomicU64::new(0))),
canonical_height_cache: RwLock::new(std::collections::BTreeMap::new()),
canonical_height_cache_capacity: config.canonical_height_cache_capacity,
hash_to_height_cache: Arc::new(ShardedLruCache::new(
config.hash_to_height_cache_capacity,
shards,
)),
}),
pipeline_tx: Arc::new(tokio::sync::Mutex::new(None)),
};
if let Ok(Some(h)) = store.read_min_retained_height() {
store.min_retained_height_cached.store(h, Ordering::Release);
}
if warm_cache_on_open {
let warmed = store.warm_caches(warm_cache_depth);
store.warm_blocks_loaded.store(warmed, Ordering::Relaxed);
}
Ok(store)
}
pub fn open_readonly(path: impl AsRef<Path>) -> Result<Self, BlockStoreError> {
let path = path.as_ref();
if !path.exists() {
return Err(BlockStoreError::Serialization(format!(
"{ERR_OPEN_READONLY_PATH_MISSING_PREFIX}{}",
path.display()
)));
}
let opts = Options::default();
let readonly_cfg = BlockStoreConfig {
path: path.to_path_buf(),
..BlockStoreConfig::default()
};
let compression_level = readonly_cfg.compression_level;
let use_compression_dict = readonly_cfg.use_compression_dict;
let max_decompressed_block_bytes = readonly_cfg.max_decompressed_block_bytes;
let zstd_dictionary_override = readonly_cfg.zstd_dictionary_override.clone();
let cfs = cf_options::column_family_descriptors(&readonly_cfg, None);
let db = DB::open_cf_descriptors_read_only(&opts, path, cfs, false)?;
let db = Arc::new(db);
let canonical_bin = RwLock::new(CanonicalBin::open_synced(&db, path, false)?);
let zstd_dict =
resolve_zstd_dictionary(&db, use_compression_dict, zstd_dictionary_override)?;
let tip = load_tip(&db)?;
let readahead_size = readonly_cfg.readahead_size;
let shards = readonly_cfg.cache_shards.max(1);
let block_cache = Arc::new(ShardedBlockCache::new(
readonly_cfg.block_cache_capacity,
shards,
));
let header_cache = Arc::new(ShardedHeaderCache::new(
readonly_cfg.header_cache_capacity,
shards,
));
let store = Self {
inner: Arc::new(BlockStoreInner {
db,
read_only: true,
tip: RwLock::new(tip),
warm_blocks_loaded: AtomicUsize::new(0),
compression_level,
use_compression_dict,
max_decompressed_block_bytes,
zstd_dict: RwLock::new(zstd_dict),
record_cache: Mutex::new(HashMap::new()),
block_cache,
cf_blocks_physical_gets: AtomicUsize::new(0),
cf_blocks_multi_get_batches: AtomicUsize::new(0),
cf_blocks_stream_physical_gets: AtomicUsize::new(0),
readahead_size,
header_cache,
cf_headers_physical_gets: AtomicUsize::new(0),
pipeline_batch_size: readonly_cfg.write_pipeline_batch_size.max(1),
pipeline_flush_ms: readonly_cfg.write_pipeline_flush_ms.max(1),
pipeline_channel_capacity: readonly_cfg.write_pipeline_channel_capacity.max(1),
pipeline_write_batches: AtomicUsize::new(0),
canonical_bin,
min_retained_height_cached: Arc::new(AtomicU64::new(0)),
canonical_height_cache: RwLock::new(std::collections::BTreeMap::new()),
canonical_height_cache_capacity: readonly_cfg.canonical_height_cache_capacity,
hash_to_height_cache: Arc::new(ShardedLruCache::new(
readonly_cfg.hash_to_height_cache_capacity,
shards,
)),
}),
pipeline_tx: Arc::new(tokio::sync::Mutex::new(None)),
};
if let Ok(Some(h)) = store.read_min_retained_height() {
store.min_retained_height_cached.store(h, Ordering::Release);
}
Ok(store)
}
pub fn init_genesis(&self, block: &L2Block) -> Result<(), BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_INIT_GENESIS_READ_ONLY.into(),
));
}
let meta = self.cf(CF_METADATA)?;
if self.db.get_cf(meta, META_TIP.as_bytes())?.is_some()
|| self
.db
.get_cf(meta, META_GENESIS_HASH.as_bytes())?
.is_some()
{
return Err(BlockStoreError::Serialization(
ERR_INIT_GENESIS_ALREADY_INITIALIZED.into(),
));
}
let hash = block.hash();
if block.height() != 0 {
return Err(BlockStoreError::Serialization(format!(
"init_genesis: genesis block height must be 0, got {}",
block.height()
)));
}
let compressed = self.serialize_block(block)?;
let header_bytes = Self::serialize_header(&block.header)?;
let tip = ChainTip { hash, height: 0 };
let mut batch = WriteBatch::default();
let cf_b = self.cf(CF_BLOCKS)?;
let cf_h = self.cf(CF_HEADERS)?;
let cf_c = self.cf(CF_CANONICAL)?;
batch.put_cf(cf_b, hash_key(&hash).as_slice(), &compressed);
batch.put_cf(cf_h, hash_key(&hash).as_slice(), &header_bytes);
batch.put_cf(cf_c, height_key(0), hash_key(&hash).as_slice());
batch.put_cf(meta, META_TIP.as_bytes(), tip.to_bytes().as_slice());
batch.put_cf(meta, META_GENESIS_HASH.as_bytes(), hash.as_ref());
self.db.write(batch)?;
self.canonical_bin.write().extend_write(0, &hash)?;
*self.tip.write() = Some(tip);
let record = BlockRecord::from_header(&block.header, BlockStatus::Validated);
self.record_cache.lock().insert(hash, record);
self.block_cache.insert(hash, block.clone());
self.header_cache.insert(hash, block.header.clone());
self.maybe_train_dictionary()?;
Ok(())
}
pub fn disable_canonical_bin_acceleration(&self) {
self.canonical_bin.write().disable();
}
pub fn tip(&self) -> Option<ChainTip> {
*self.tip.read()
}
#[must_use]
pub fn height(&self) -> Option<u64> {
self.tip().map(|t| t.height)
}
pub fn set_tip(&self, tip: ChainTip) -> Result<(), BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_MUTATION_READ_ONLY.into(),
));
}
let cf = self.cf(CF_METADATA)?;
self.db
.put_cf(cf, META_TIP.as_bytes(), tip.to_bytes().as_slice())?;
*self.tip.write() = Some(tip);
Ok(())
}
pub fn warm_blocks_loaded_count(&self) -> usize {
self.warm_blocks_loaded.load(Ordering::Relaxed)
}
pub fn get_block(&self, hash: &Bytes32) -> Result<Option<L2Block>, BlockStoreError> {
if let Some(block) = self.block_cache.get_clone(hash) {
return Ok(Some(block));
}
let cf = self.cf(CF_BLOCKS)?;
self.cf_blocks_physical_gets.fetch_add(1, Ordering::Relaxed);
let raw_opt = self.db.get_cf(cf, hash_key(hash).as_slice())?;
let Some(raw) = raw_opt else {
return Ok(None);
};
let block = self.deserialize_block(&raw)?;
self.block_cache.insert(*hash, block.clone());
self.header_cache.insert(*hash, block.header.clone());
self.hash_to_height_cache.insert(*hash, block.height());
Ok(Some(block))
}
pub fn has_block(&self, hash: &Bytes32) -> Result<bool, BlockStoreError> {
if self.block_cache.contains(hash) || self.header_cache.contains(hash) {
return Ok(true);
}
let key = hash_key(hash);
let cf_h = self.cf(CF_HEADERS)?;
if self.db.get_cf(cf_h, key.as_slice())?.is_some() {
return Ok(true);
}
let cf_b = self.cf(CF_BLOCKS)?;
Ok(self.db.get_cf(cf_b, key.as_slice())?.is_some())
}
pub fn stats(&self) -> Result<StorageStats, BlockStoreError> {
Ok(StorageStats {
block_count: self.count_cf_entries(CF_BLOCKS)?,
canonical_block_count: self.count_cf_entries(CF_CANONICAL)?,
header_count: self.count_cf_entries(CF_HEADERS)?,
checkpoint_count: self.count_cf_entries(CF_CHECKPOINTS)?,
attested_count: self.count_cf_entries(CF_ATTESTED)?,
tip_height: self.tip().map(|t| t.height),
min_height: self.read_min_retained_height()?,
total_size_bytes: self.sum_cf_live_data_size_estimates()?,
})
}
fn count_cf_entries(&self, cf_name: &'static str) -> Result<u64, BlockStoreError> {
let cf = self.cf(cf_name)?;
let mut n = 0u64;
for entry in self.db.iterator_cf(cf, IteratorMode::Start) {
let (_k, _v) = entry?;
n += 1;
}
Ok(n)
}
fn sum_cf_live_data_size_estimates(&self) -> Result<u64, BlockStoreError> {
const PROP: &str = "rocksdb.estimate-live-data-size";
let mut sum = 0u64;
for name in [
CF_BLOCKS,
CF_HEADERS,
CF_CANONICAL,
CF_METADATA,
CF_ATTESTED,
CF_CHECKPOINTS,
] {
let cf = self.cf(name)?;
if let Some(v) = self.db.property_int_value_cf(cf, PROP)? {
sum = sum.saturating_add(v);
}
}
Ok(sum)
}
fn read_min_retained_height(&self) -> Result<Option<u64>, BlockStoreError> {
let meta = self.cf(CF_METADATA)?;
let Some(bytes) = self.db.get_cf(meta, META_MIN_HEIGHT.as_bytes())? else {
return Ok(None);
};
let arr: [u8; 8] = bytes.as_slice().try_into().map_err(|_| {
BlockStoreError::Serialization(format!(
"stats: META_MIN_HEIGHT value must be exactly 8 bytes (little-endian u64), got {} bytes",
bytes.len()
))
})?;
Ok(Some(u64::from_le_bytes(arr)))
}
pub fn flush(&self) -> Result<(), BlockStoreError> {
self.db.flush_wal(true)?;
self.db.flush()?;
Ok(())
}
pub fn compact(&self) -> Result<(), BlockStoreError> {
for &name in crate::constants::ALL_COLUMN_FAMILIES {
let cf = self.cf(name)?;
self.db.compact_range_cf(cf, None::<&[u8]>, None::<&[u8]>);
}
Ok(())
}
pub fn get_blocks_by_hash(
&self,
hashes: &[Bytes32],
) -> Result<Vec<Option<L2Block>>, BlockStoreError> {
let mut results: Vec<Option<L2Block>> = vec![None; hashes.len()];
let mut miss_indices: Vec<usize> = Vec::new();
for (i, hash) in hashes.iter().enumerate() {
if let Some(block) = self.block_cache.get_clone(hash) {
results[i] = Some(block);
} else {
miss_indices.push(i);
}
}
if miss_indices.is_empty() {
return Ok(results);
}
let cf = self.cf(CF_BLOCKS)?;
self.cf_blocks_multi_get_batches
.fetch_add(1, Ordering::Relaxed);
let keys: Vec<[u8; 32]> = miss_indices
.iter()
.map(|&idx| *hash_key(&hashes[idx]))
.collect();
let db_results = self
.db
.multi_get_cf(keys.iter().map(|k| (cf, k.as_slice())));
for (j, db_result) in db_results.into_iter().enumerate() {
let idx = miss_indices[j];
let maybe_raw = db_result?;
let Some(raw) = maybe_raw else {
continue;
};
let block = self.deserialize_block(&raw)?;
self.block_cache.insert(hashes[idx], block.clone());
self.header_cache.insert(hashes[idx], block.header.clone());
results[idx] = Some(block);
}
Ok(results)
}
pub fn invalidate_block_cache_entry(&self, hash: &Bytes32) {
self.block_cache.remove(hash);
}
pub fn get_block_by_height(&self, height: u64) -> Result<Option<L2Block>, BlockStoreError> {
let Some(hash) = self.get_hash_by_height(height)? else {
return Ok(None);
};
self.get_block(&hash)
}
pub fn get_blocks_in_range(
&self,
start_height: u64,
end_height: u64,
) -> Result<Vec<L2Block>, BlockStoreError> {
if start_height > end_height {
return Ok(Vec::new());
}
let mut blocks = Vec::with_capacity((end_height - start_height + 1) as usize);
for height in start_height..=end_height {
if let Some(block) = self.get_block_by_height(height)? {
blocks.push(block);
}
}
Ok(blocks)
}
pub fn get_record_by_height(
&self,
height: u64,
) -> Result<Option<BlockRecord>, BlockStoreError> {
let Some(hash) = self.get_hash_by_height(height)? else {
return Ok(None);
};
self.get_record(&hash)
}
pub fn get_header_by_height(
&self,
height: u64,
) -> Result<Option<L2BlockHeader>, BlockStoreError> {
let Some(hash) = self.get_hash_by_height(height)? else {
return Ok(None);
};
self.get_header(&hash)
}
pub fn get_epoch_block_hashes(&self, epoch: u64) -> Result<Vec<Bytes32>, BlockStoreError> {
let start = dig_epoch::first_height_in_epoch(epoch);
let end = dig_epoch::epoch_checkpoint_height(epoch);
let mut hashes = Vec::new();
for height in start..=end {
if let Some(hash) = self.get_hash_by_height(height)? {
hashes.push(hash);
} else {
break; }
}
Ok(hashes)
}
pub fn get_records_in_range(
&self,
start_height: u64,
end_height: u64,
) -> Result<Vec<BlockRecord>, BlockStoreError> {
if start_height > end_height {
return Ok(Vec::new());
}
let mut records = Vec::with_capacity((end_height - start_height + 1) as usize);
for height in start_height..=end_height {
if let Some(record) = self.get_record_by_height(height)? {
records.push(record);
}
}
Ok(records)
}
pub fn cf_blocks_physical_get_count(&self) -> u64 {
self.cf_blocks_physical_gets.load(Ordering::Relaxed) as u64
}
#[inline]
pub fn cf_blocks_multi_get_batch_count(&self) -> u64 {
self.cf_blocks_multi_get_batches.load(Ordering::Relaxed) as u64
}
#[must_use]
pub fn readahead_size(&self) -> usize {
self.readahead_size
}
#[must_use]
pub fn cf_blocks_stream_physical_get_count(&self) -> u64 {
self.cf_blocks_stream_physical_gets.load(Ordering::Relaxed) as u64
}
pub fn get_header(&self, hash: &Bytes32) -> Result<Option<L2BlockHeader>, BlockStoreError> {
if let Some(header) = self.header_cache.get_clone(hash) {
return Ok(Some(header));
}
let cf = self.cf(CF_HEADERS)?;
self.cf_headers_physical_gets
.fetch_add(1, Ordering::Relaxed);
let raw_opt = self.db.get_cf(cf, hash_key(hash).as_slice())?;
let Some(raw) = raw_opt else {
return Ok(None);
};
let header = Self::deserialize_header(&raw)?;
self.header_cache.insert(*hash, header.clone());
self.hash_to_height_cache.insert(*hash, header.height);
Ok(Some(header))
}
pub fn invalidate_header_cache_entry(&self, hash: &Bytes32) {
self.header_cache.remove(hash);
}
pub fn cf_headers_physical_get_count(&self) -> u64 {
self.cf_headers_physical_gets.load(Ordering::Relaxed) as u64
}
pub fn put_block(&self, block: &L2Block, canonical: bool) -> Result<bool, BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_MUTATION_READ_ONLY.into(),
));
}
let hash = block.hash();
let cf_b = self.cf(CF_BLOCKS)?;
if self.db.get_cf(cf_b, hash_key(&hash).as_slice())?.is_some() {
return Ok(false);
}
let compressed = self.serialize_block(block)?;
let header_bytes = Self::serialize_header(&block.header)?;
let mut batch = WriteBatch::default();
let cf_h = self.cf(CF_HEADERS)?;
batch.put_cf(cf_b, hash_key(&hash).as_slice(), &compressed);
batch.put_cf(cf_h, hash_key(&hash).as_slice(), &header_bytes);
if canonical {
let cf_c = self.cf(CF_CANONICAL)?;
batch.put_cf(cf_c, height_key(block.height()), hash_key(&hash).as_slice());
}
self.db.write(batch)?;
if canonical {
self.canonical_bin
.write()
.extend_write(block.height(), &hash)?;
self.insert_canonical_height_cache(block.height(), hash);
}
let record = BlockRecord::from_header(&block.header, BlockStatus::Validated);
self.record_cache.lock().insert(hash, record);
self.block_cache.insert(hash, block.clone());
self.header_cache.insert(hash, block.header.clone());
self.hash_to_height_cache.insert(hash, block.height());
self.maybe_train_dictionary()?;
Ok(true)
}
#[inline]
pub fn put(&self, block: &L2Block, canonical: bool) -> Result<bool, BlockStoreError> {
self.put_block(block, canonical)
}
pub fn min_retained_height(&self) -> Result<u64, BlockStoreError> {
Ok(self.min_retained_height_cached.load(Ordering::Acquire))
}
pub fn blocks_to_revert(&self, target_height: u64) -> Result<Vec<Bytes32>, BlockStoreError> {
let Some(current_tip) = self.tip() else {
return Ok(Vec::new());
};
if target_height >= current_tip.height {
return Ok(Vec::new());
}
let mut reverted = Vec::new();
for h in (target_height + 1..=current_tip.height).rev() {
if let Some(hash) = self.get_hash_by_height(h)? {
reverted.push(hash);
}
}
Ok(reverted)
}
pub fn rollback_to_height(&self, target_height: u64) -> Result<Vec<Bytes32>, BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_MUTATION_READ_ONLY.into(),
));
}
let current_tip = self.tip().ok_or(BlockStoreError::NoTip)?;
if target_height > current_tip.height {
return Err(BlockStoreError::RollbackAboveTip {
target: target_height,
tip: current_tip.height,
});
}
let min_height = self.min_retained_height()?;
if target_height < min_height {
return Err(BlockStoreError::RollbackBelowMin {
target: target_height,
min: min_height,
});
}
if target_height == current_tip.height {
return Ok(Vec::new());
}
let mut reverted = Vec::new();
for h in (target_height + 1..=current_tip.height).rev() {
if let Some(hash) = self.get_hash_by_height(h)? {
reverted.push(hash);
}
}
let cf_c = self.cf(CF_CANONICAL)?;
let mut batch = WriteBatch::default();
for h in target_height + 1..=current_tip.height {
batch.delete_cf(cf_c, height_key(h));
}
self.db.write(batch)?;
self.canonical_bin
.write()
.truncate_to_height(target_height)?;
if let Some(target_hash) = self.get_hash_by_height(target_height)? {
self.set_tip(ChainTip {
hash: target_hash,
height: target_height,
})?;
}
{
let mut cache = self.record_cache.lock();
for hash in &reverted {
if let Some(r) = cache.get_mut(hash) {
r.in_canonical_chain = false;
}
}
}
{
let mut hcache = self.canonical_height_cache.write();
for h in target_height + 1..=current_tip.height {
hcache.remove(&h);
}
}
Ok(reverted)
}
pub fn apply_reorg(
&self,
ancestor_height: u64,
new_chain_hashes: &[Bytes32],
) -> Result<ReorgResult, BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_MUTATION_READ_ONLY.into(),
));
}
let current_tip = self.tip().ok_or(BlockStoreError::NoTip)?;
if new_chain_hashes.is_empty() {
return Err(BlockStoreError::EmptyReorgChain);
}
let mut new_records: Vec<(Bytes32, BlockRecord)> =
Vec::with_capacity(new_chain_hashes.len());
for hash in new_chain_hashes {
let record = self
.get_record(hash)?
.ok_or(BlockStoreError::BlockNotInStore(*hash))?;
new_records.push((*hash, record));
}
let cf_c = self.cf(CF_CANONICAL)?;
let cf_meta = self.cf(CF_METADATA)?;
let mut batch = WriteBatch::default();
let mut reverted = Vec::new();
for h in (ancestor_height + 1..=current_tip.height).rev() {
if let Some(hash) = self.get_hash_by_height(h)? {
reverted.push(hash);
}
batch.delete_cf(cf_c, height_key(h));
}
for (hash, record) in &new_records {
batch.put_cf(cf_c, height_key(record.height), hash_key(hash).as_slice());
}
let new_tip_hash = new_chain_hashes
.last()
.copied()
.expect("non-empty checked above");
let new_tip_height = new_records.last().expect("non-empty").1.height;
let new_tip = ChainTip {
hash: new_tip_hash,
height: new_tip_height,
};
batch.put_cf(cf_meta, META_TIP.as_bytes(), new_tip.to_bytes().as_slice());
self.db.write(batch)?;
self.canonical_bin
.write()
.truncate_to_height(ancestor_height)?;
for (hash, record) in &new_records {
self.canonical_bin
.write()
.extend_write(record.height, hash)?;
}
{
let mut cache = self.record_cache.lock();
for hash in &reverted {
if let Some(r) = cache.get_mut(hash) {
r.in_canonical_chain = false;
}
}
for (hash, _) in &new_records {
if let Some(r) = cache.get_mut(hash) {
r.in_canonical_chain = true;
}
}
}
{
let mut hcache = self.canonical_height_cache.write();
for h in ancestor_height + 1..=current_tip.height {
hcache.remove(&h);
}
for (hash, record) in &new_records {
hcache.insert(record.height, *hash);
}
}
*self.tip.write() = Some(new_tip);
Ok(ReorgResult {
reverted,
applied: new_chain_hashes.to_vec(),
new_tip,
})
}
pub fn find_common_ancestor(
&self,
hash: &Bytes32,
max_depth: u64,
) -> Result<Option<(Bytes32, u64)>, BlockStoreError> {
let mut current_hash = *hash;
for _ in 0..max_depth {
let record = match self.get_record(¤t_hash)? {
Some(r) => r,
None => return Ok(None), };
let height = record.height;
self.hash_to_height_cache.insert(current_hash, height);
if let Some(canonical_hash) = self.get_hash_by_height(height)? {
if canonical_hash == current_hash {
return Ok(Some((current_hash, height)));
}
}
current_hash = record.parent_hash;
}
Ok(None) }
pub fn put_attestation(
&self,
hash: &Bytes32,
attested: &AttestedBlock,
) -> Result<(), BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_MUTATION_READ_ONLY.into(),
));
}
let bytes = bincode::serialize(attested)
.map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
let cf = self.cf(CF_ATTESTED)?;
self.db.put_cf(cf, hash_key(hash).as_slice(), &bytes)?;
Ok(())
}
pub fn get_attestation(
&self,
hash: &Bytes32,
) -> Result<Option<AttestedBlock>, BlockStoreError> {
let cf = self.cf(CF_ATTESTED)?;
let raw = match self.db.get_cf(cf, hash_key(hash).as_slice())? {
Some(b) => b,
None => return Ok(None),
};
let attested: AttestedBlock = bincode::deserialize(&raw).map_err(|e| {
BlockStoreError::Serialization(format!(
"get_attestation: bincode deserialize failed: {e}"
))
})?;
Ok(Some(attested))
}
pub fn put_checkpoint(
&self,
checkpoint: &crate::StoredCheckpoint,
) -> Result<(), BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_MUTATION_READ_ONLY.into(),
));
}
let epoch = checkpoint.checkpoint.epoch;
let key = crate::encoding::epoch_key(epoch);
let value = checkpoint
.encode_bincode()
.map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
let cf = self.cf(CF_CHECKPOINTS)?;
self.db.put_cf(cf, key.as_slice(), &value)?;
Ok(())
}
pub fn get_checkpoint(
&self,
epoch: u64,
) -> Result<Option<crate::StoredCheckpoint>, BlockStoreError> {
let cf = self.cf(CF_CHECKPOINTS)?;
let key = crate::encoding::epoch_key(epoch);
let Some(bytes) = self.db.get_cf(cf, key.as_slice())? else {
return Ok(None);
};
let checkpoint = crate::StoredCheckpoint::decode_bincode(&bytes)
.map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
Ok(Some(checkpoint))
}
pub fn get_latest_checkpoint(
&self,
) -> Result<Option<crate::StoredCheckpoint>, BlockStoreError> {
let cf = self.cf(CF_CHECKPOINTS)?;
let mut iter = self.db.iterator_cf(cf, IteratorMode::End);
let Some(item) = iter.next() else {
return Ok(None);
};
let (_key, value) = item?;
let checkpoint = crate::StoredCheckpoint::decode_bincode(&value)
.map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
Ok(Some(checkpoint))
}
pub fn get_checkpoints_in_range(
&self,
start_epoch: u64,
end_epoch: u64,
) -> Result<Vec<crate::StoredCheckpoint>, BlockStoreError> {
if start_epoch > end_epoch {
return Ok(Vec::new());
}
let cf = self.cf(CF_CHECKPOINTS)?;
let start_key = crate::encoding::epoch_key(start_epoch);
let mode = IteratorMode::From(&start_key, Direction::Forward);
let iter = self.db.iterator_cf(cf, mode);
let mut result = Vec::new();
for item in iter {
let (key_bytes, value) = item?;
if key_bytes.len() != 8 {
continue;
}
let key_arr: [u8; 8] = key_bytes.as_ref().try_into().unwrap_or([0; 8]);
let epoch = crate::encoding::decode_epoch_key(&key_arr);
if epoch > end_epoch {
break;
}
let checkpoint = crate::StoredCheckpoint::decode_bincode(&value)
.map_err(|e| BlockStoreError::Serialization(e.to_string()))?;
result.push(checkpoint);
}
Ok(result)
}
pub fn prune_before_height(&self, height: u64) -> Result<usize, BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_MUTATION_READ_ONLY.into(),
));
}
let current_min = self.min_retained_height()?;
if height <= current_min {
return Ok(0);
}
let cf_b = self.cf(CF_BLOCKS)?;
let cf_h = self.cf(CF_HEADERS)?;
let cf_a = self.cf(CF_ATTESTED)?;
let cf_c = self.cf(CF_CANONICAL)?;
let cf_meta = self.cf(CF_METADATA)?;
let mut batch = WriteBatch::default();
let mut pruned_hashes: Vec<Bytes32> = Vec::new();
for h in current_min..height {
if let Some(hash) = self.get_hash_by_height(h)? {
batch.delete_cf(cf_b, hash_key(&hash).as_slice());
batch.delete_cf(cf_h, hash_key(&hash).as_slice());
batch.delete_cf(cf_a, hash_key(&hash).as_slice());
batch.delete_cf(cf_c, height_key(h));
pruned_hashes.push(hash);
}
}
let canonical_set: std::collections::HashSet<Bytes32> =
pruned_hashes.iter().copied().collect();
let header_iter = self.db.iterator_cf(cf_h, IteratorMode::Start);
for item in header_iter {
let (key_bytes, value_bytes) = item?;
if key_bytes.len() != 32 {
continue;
}
let arr: [u8; 32] = key_bytes.as_ref().try_into().unwrap_or([0; 32]);
let hash = Bytes32::new(arr);
if canonical_set.contains(&hash) {
continue; }
if let Ok(header) = Self::deserialize_header(&value_bytes) {
if header.height < height {
batch.delete_cf(cf_b, hash_key(&hash).as_slice());
batch.delete_cf(cf_h, hash_key(&hash).as_slice());
batch.delete_cf(cf_a, hash_key(&hash).as_slice());
pruned_hashes.push(hash);
}
}
}
batch.put_cf(cf_meta, META_MIN_HEIGHT.as_bytes(), height.to_le_bytes());
let count = pruned_hashes.len();
self.db.write(batch)?;
self.min_retained_height_cached
.store(height, Ordering::Release);
for hash in &pruned_hashes {
self.block_cache.remove(hash);
self.header_cache.remove(hash);
self.record_cache.lock().remove(hash);
self.hash_to_height_cache.remove(hash);
}
{
let mut hcache = self.canonical_height_cache.write();
let to_remove: Vec<u64> = hcache.range(..height).map(|(&h, _)| h).collect();
for h in to_remove {
hcache.remove(&h);
}
}
Ok(count)
}
pub fn prune_checkpoints_before_epoch(&self, epoch: u64) -> Result<usize, BlockStoreError> {
if self.read_only {
return Err(BlockStoreError::Serialization(
ERR_MUTATION_READ_ONLY.into(),
));
}
if epoch == 0 {
return Ok(0);
}
let cf = self.cf(CF_CHECKPOINTS)?;
let mut batch = WriteBatch::default();
let mut count = 0usize;
let iter = self.db.iterator_cf(cf, IteratorMode::Start);
for item in iter {
let (key_bytes, _value) = item?;
if key_bytes.len() != 8 {
continue;
}
let key_arr: [u8; 8] = key_bytes.as_ref().try_into().unwrap_or([0; 8]);
let e = crate::encoding::decode_epoch_key(&key_arr);
if e >= epoch {
break;
}
batch.delete_cf(cf, key_bytes.as_ref());
count += 1;
}
if count > 0 {
self.db.write(batch)?;
}
Ok(count)
}
pub fn get_record(&self, hash: &Bytes32) -> Result<Option<BlockRecord>, BlockStoreError> {
{
let guard = self.record_cache.lock();
if let Some(r) = guard.get(hash) {
return Ok(Some(r.clone()));
}
}
if let Some(header) = self.header_cache.get_clone(hash) {
let record = BlockRecord::from_header(&header, BlockStatus::Validated);
self.record_cache.lock().insert(*hash, record.clone());
return Ok(Some(record));
}
let cf = self.cf(CF_HEADERS)?;
self.cf_headers_physical_gets
.fetch_add(1, Ordering::Relaxed);
let Some(bytes) = self.db.get_cf(cf, hash_key(hash).as_slice())? else {
return Ok(None);
};
let header = Self::deserialize_header(&bytes)?;
self.header_cache.insert(*hash, header.clone());
let record = BlockRecord::from_header(&header, BlockStatus::Validated);
self.record_cache.lock().insert(*hash, record.clone());
Ok(Some(record))
}
pub fn invalidate_record_cache_entry(&self, hash: &Bytes32) {
let mut guard = self.record_cache.lock();
let _ = guard.remove(hash);
}
pub fn update_status(
&self,
hash: &Bytes32,
status: BlockStatus,
) -> Result<(), BlockStoreError> {
let mut guard = self.record_cache.lock();
let record = guard.get_mut(hash).ok_or_else(|| {
BlockStoreError::Serialization(format!(
"{ERR_UPDATE_STATUS_RECORD_NOT_CACHED_PREFIX}{hash}"
))
})?;
record.status = status;
record.in_canonical_chain = status.is_canonical();
Ok(())
}
pub async fn get_block_async(
&self,
hash: &Bytes32,
) -> Result<Option<L2Block>, BlockStoreError> {
if let Some(block) = self.block_cache.get_clone(hash) {
return Ok(Some(block));
}
let store = self.clone();
let hash = *hash;
tokio::task::spawn_blocking(move || store.get_block(&hash))
.await
.map_err(Self::map_spawn_join)?
}
pub async fn get_header_async(
&self,
hash: &Bytes32,
) -> Result<Option<L2BlockHeader>, BlockStoreError> {
if let Some(header) = self.header_cache.get_clone(hash) {
return Ok(Some(header));
}
let store = self.clone();
let hash = *hash;
tokio::task::spawn_blocking(move || store.get_header(&hash))
.await
.map_err(Self::map_spawn_join)?
}
pub async fn get_block_by_height_async(
&self,
height: u64,
) -> Result<Option<L2Block>, BlockStoreError> {
let store = self.clone();
tokio::task::spawn_blocking(move || store.get_block_by_height(height))
.await
.map_err(Self::map_spawn_join)?
}
#[inline]
fn map_spawn_join(err: tokio::task::JoinError) -> BlockStoreError {
BlockStoreError::Serialization(format!("{ERR_ASYNC_JOIN_PREFIX}{err}"))
}
pub(crate) fn cf(&self, name: &'static str) -> Result<&rocksdb::ColumnFamily, BlockStoreError> {
self.db
.cf_handle(name)
.ok_or_else(|| BlockStoreError::Serialization(format!("missing column family {name}")))
}
}
fn load_tip(db: &DB) -> Result<Option<ChainTip>, BlockStoreError> {
let meta = db
.cf_handle(CF_METADATA)
.ok_or_else(|| BlockStoreError::Serialization("missing CF_METADATA".into()))?;
let Some(raw) = db.get_cf(meta, META_TIP.as_bytes())? else {
return Ok(None);
};
ChainTip::from_bytes(&raw).map(Some)
}