diff --git a/keeper/README.md b/keeper/README.md index 6ace65a..b296090 100644 --- a/keeper/README.md +++ b/keeper/README.md @@ -24,6 +24,12 @@ non-zero superblock interval used by pacing and cache TTL calculation. The shared accountsdb, blockstore, and ledger parameters are defined by nucleus; keeper consumes them when opening its durable stores and caches. +Startup reconstructs the recent-blockhash cache from retained ledger blocks for +the configured 60-second validity window, bounded by the accountsdb slot so an +unreplayed ledger tail cannot advance startup state. Persisted `SlotHashes` +supplies the newest entries, extended with older ledger hashes when available; +snapshot bootstrap without ledger history uses `SlotHashes` alone. + ## Authority `nucleus::config::Authority::local` is the keypair used for locally signed diff --git a/keeper/src/builder.rs b/keeper/src/builder.rs index d50f48c..857c91d 100644 --- a/keeper/src/builder.rs +++ b/keeper/src/builder.rs @@ -11,7 +11,7 @@ use accountsdb::{AccountEntry, AccountsDB, AccountsDBError, BackupOp, SnapshotEr use agave_feature_set::FeatureSet; use ledger::{ Ledger, LedgerHandle, - request::{BlockDetails, BlockParams, ReadRequest, RequestPayload}, + request::{ReadRequest, RequestPayload}, }; use nucleus::{ Slot, @@ -39,7 +39,7 @@ use tracing::{error, info, warn}; use crate::{ Keeper, - cache::{AccountCache, BlocksCache, Caches, ExpiringCache}, + cache::{AccountCache, BlockSeed, BlocksCache, Caches, ExpiringCache}, error::Result, metrics, subscriptions::Subscriptions, @@ -80,8 +80,8 @@ impl KeeperBuilder { pub async fn build(mut self, shutdown: &mut ShutdownManager) -> Result { let ledger = Ledger::init(&self.ledger.directory, self.ledger.size_limit, shutdown)?; let accountsdb = self.accountsdb(&ledger)?; - let (block, featureset) = self.prepopulate(&accountsdb, &ledger).await?; - let caches = self.caches(block); + let (blocks, featureset) = self.prepopulate(&accountsdb, &ledger).await?; + let caches = self.caches(blocks); metrics::init(); Ok(Keeper { authority: self.authority, @@ -99,11 +99,11 @@ impl KeeperBuilder { &mut self, accountsdb: &AccountsDB, ledger: &LedgerHandle, - ) -> Result<(Block, FeatureSet)> { + ) -> Result<(BlockSeed, FeatureSet)> { let mut accounts = Vec::new(); let featureset = self.seed_featureset(&mut accounts)?; self.seed_programs(&mut accounts)?; - let block = self.seed_sysvars(accountsdb, ledger, &mut accounts).await?; + let blocks = self.seed_sysvars(accountsdb, ledger, &mut accounts).await?; let authority = self.authority.pubkey(); if accountsdb.loader().load(&authority)?.is_none() { let sponsor = AccountBuilder::default() @@ -113,20 +113,23 @@ impl KeeperBuilder { } accounts.extend(self.accounts.drain()); accountsdb.store(&accounts)?; - Ok((block, featureset)) + Ok((blocks, featureset)) } /// Builds read-side caches using blocktime-derived slot TTLs. - fn caches(&self, latest: Block) -> Caches { - let blocktime = self.blockstore.blocktime; - let ttl = |window: Duration| window.div_duration_f64(blocktime).ceil() as Slot; - let blocks = BlocksCache::new(latest, ttl(BLOCK_CACHE_WINDOW)); - let signatures = ExpiringCache::new(ttl(SIGNATURE_CACHE_WINDOW)); + fn caches(&self, blocks: BlockSeed) -> Caches { + let blocks = BlocksCache::new(blocks, self.ttl(BLOCK_CACHE_WINDOW)); + let signatures = ExpiringCache::new(self.ttl(SIGNATURE_CACHE_WINDOW)); let accounts = Arc::new(AccountCache::new(self.accountsdb.lru_capacity)); Caches { signatures, blocks, accounts } } + /// Converts a wall-clock cache window into whole configured block slots. + fn ttl(&self, window: Duration) -> Slot { + window.div_duration_f64(self.blockstore.blocktime).ceil() as Slot + } + /// Activates the engine's required feature gates at slot 0, seeds a feature /// account for each, and returns the resulting [`FeatureSet`]. fn seed_featureset(&self, accounts: &mut Vec) -> Result { @@ -176,44 +179,43 @@ impl KeeperBuilder { /// Seeds sysvars derived from retained ledger state and keeper config. /// - /// Returns the latest available block, resolved from accountsdb or ledger + /// Returns the latest available block and its anchored retained history. async fn seed_sysvars( &self, accountsdb: &AccountsDB, ledger: &LedgerHandle, accounts: &mut Vec, - ) -> Result { + ) -> Result { let slot = accountsdb.slot(); let loader = accountsdb.loader(); - let mut last_block = None; - if let Some(hashes) = loader.load(&SlotHashes::id())? { - let hashes = hashes.deserialize_data::().map_err(AccountsDBError::from)?; - // `SlotHashes` is ordered newest-first, so the latest block is `first` - if let Some(&(slot, hash)) = hashes.first() { - let parent = &slot.saturating_sub(1); - let parent = hashes.get(parent).copied().unwrap_or_default(); - let time = self.blocktime(ledger, slot).await?; - last_block.replace(Block { slot, hash, time, parent }); - } - } else { - let range = slot.saturating_sub(SLOTHASH_ENTRIES as u64)..slot + 1; - let (payload, handle) = RequestPayload::new(range); - ledger.reader.send(ReadRequest::BlockRange(payload))?; + let slothashes = loader + .load(&SlotHashes::id())? + .map(|account| account.deserialize_data::().map_err(AccountsDBError::from)) + .transpose()?; + + let retained = self.ttl(BLOCK_CACHE_WINDOW).max(SLOTHASH_ENTRIES as Slot); + let start = slot.saturating_sub(retained - 1); + let (payload, handle) = RequestPayload::new(start..slot.saturating_add(1)); + ledger.reader.send(ReadRequest::BlockRange(payload))?; + let blocks = handle.recv_timeout().await??; + if slothashes.is_none() { + // Keep the sysvar account at its fixed serialized capacity so live + // updates can replace entries without resizing the account. let mut hashes = SlotHashes::new(&[Default::default(); SLOTHASH_ENTRIES]); - for block in handle.recv_timeout().await?? { + for block in blocks.iter().take(SLOTHASH_ENTRIES) { hashes.add(block.slot, block.hash); - last_block.replace(block); } let acc = self.account(&hashes, &sysvar::ID)?; accounts.push((SlotHashes::id(), acc.build())); } - let block = last_block.unwrap_or_default(); + let blocks = Self::block_seed(blocks, slothashes.as_ref()); + // Set the clock slot one ahead from the last let clock = Clock { - slot: block.slot + 1, - unix_timestamp: block.time, + slot: blocks.latest.slot + 1, + unix_timestamp: blocks.latest.time, ..Default::default() }; accounts.push((Clock::id(), self.account(&clock, &sysvar::ID)?.build())); @@ -239,7 +241,26 @@ impl KeeperBuilder { EpochRewards::id(), self.account(&EpochRewards::default(), &sysvar::ID)?.build(), )); - Ok(block) + Ok(blocks) + } + + /// Extends persisted SlotHashes with older retained ledger history. + fn block_seed(blocks: Vec, slothashes: Option<&SlotHashes>) -> BlockSeed { + let Some(hashes) = slothashes.map(SlotHashes::slot_hashes) else { + let latest = blocks.first().copied().unwrap_or_default(); + let history = blocks.iter().rev().map(|b| (b.slot, b.hash)).collect(); + return BlockSeed { latest, history }; + }; + let mut history = hashes.to_vec(); + history.extend(blocks.iter().skip(history.len()).map(|b| (b.slot, b.hash))); + let latest = history.first().map_or(Block::default(), |&(slot, hash)| Block { + slot, + hash, + time: blocks.first().map_or(0, |b| b.time), + parent: history.get(1).map(|(_, hash)| *hash).unwrap_or_default(), + }); + history.reverse(); + BlockSeed { latest, history } } /// Builds a rent-exempt system account containing a serialized sysvar-like state. @@ -250,15 +271,6 @@ impl KeeperBuilder { Ok(AccountBuilder::from(account).lamports(lamports).mode(AccountMode::System)) } - /// Returns the retained block time for the given slot. - async fn blocktime(&self, ledger: &LedgerHandle, slot: Slot) -> Result { - let (payload, handle) = RequestPayload::new(BlockParams { - slot, - details: BlockDetails::None, - }); - ledger.reader.send(ReadRequest::Block(payload))?; - Ok(handle.recv_timeout().await??.map(|r| r.block().time).unwrap_or_default()) - } /// Opens accountsdb, restoring the newest archived snapshot after corruption. /// /// A restored store trails the ledger tip — snapshots are archived at sealed diff --git a/keeper/src/cache.rs b/keeper/src/cache.rs index 13526ca..3b73fb3 100644 --- a/keeper/src/cache.rs +++ b/keeper/src/cache.rs @@ -146,13 +146,20 @@ pub(crate) struct BlocksCache { } impl BlocksCache { - /// Creates a block cache seeded with the current latest block. - pub(crate) fn new(block: Block, ttl: Slot) -> Self { + /// Creates a block cache seeded with retained history in ascending slot order. + pub(crate) fn new(blocks: BlockSeed, ttl: Slot) -> Self { + let latest = blocks.latest; let cache = Self { - latest: ArcSwap::new(block.into()), + latest: ArcSwap::new(latest.into()), history: ExpiringCache::new(ttl), }; - cache.history.push(block.hash, block.slot, block.slot); + if blocks.history.is_empty() { + cache.history.push(latest.hash, latest.slot, latest.slot); + } else { + for (slot, hash) in blocks.history { + cache.history.push(hash, slot, slot); + } + } cache } @@ -248,3 +255,10 @@ impl ExpiringRecord { instant >= self.expires } } + +/// Latest committed boundary and retained hash history used to seed block caches. +pub(crate) struct BlockSeed { + pub(crate) latest: Block, + /// Entries are ordered oldest-to-newest and retain their original slots. + pub(crate) history: Vec<(Slot, SolanaHash)>, +} diff --git a/keeper/src/tests/recovery.rs b/keeper/src/tests/recovery.rs index e521a67..b12d0e6 100644 --- a/keeper/src/tests/recovery.rs +++ b/keeper/src/tests/recovery.rs @@ -1,6 +1,11 @@ //! Startup seeding, corruption recovery +use std::fs; + +use nucleus::testkit::{V42_ID, block, signed_view}; use solana_account::{AccountBuilder, AccountMode, ReadableAccount}; +use solana_instruction::Instruction; +use solana_keypair::Keypair; use solana_pubkey::Pubkey; use solana_sdk_ids::{loader_v4, sysvar}; use solana_sysvar::{ @@ -8,6 +13,7 @@ use solana_sysvar::{ }; use super::TestKeeper; +use crate::ResolvedTransaction; use crate::testkit::{Dirs, archived_snapshot, corrupt, keeper_builder, seal_and_archive}; // Startup seeds the engine's required feature gates, the configured upgradeable @@ -122,6 +128,50 @@ async fn recovers_the_newest_snapshot() { keeper.close().await; } +/// Proves startup merges ledger history beyond SlotHashes and falls back to SlotHashes alone. +#[tokio::test] +async fn restores_blockhash_history_from_ledger_and_snapshot() { + let dirs = Dirs::default(); + let builder = keeper_builder(&dirs); + let keeper = TestKeeper::from_builder(dirs, builder.clone()).await; + for slot in 1..=600 { + keeper.blocks().append(block(slot), false).unwrap(); + } + let ledger_hash = block(50).hash; + let snapshot_hash = block(100).hash; + let dirs = keeper.close().await; + + let keeper = TestKeeper::from_builder(dirs, builder.clone()).await; + assert!( + keeper.blocks().is_valid(&ledger_hash), + "slot 50 remains inside the 600-slot TTL but outside SlotHashes" + ); + keeper.accounts().dump(None).unwrap(); + let dirs = keeper.close().await; + fs::remove_dir_all(dirs.ledger.path()).unwrap(); + fs::create_dir(dirs.ledger.path()).unwrap(); + + let keeper = TestKeeper::from_builder(dirs, builder).await; + assert!( + !keeper.blocks().is_valid(&ledger_hash), + "clean ledger cannot restore history older than SlotHashes" + ); + assert!(keeper.blocks().is_valid(&snapshot_hash)); + let payer = Keypair::new(); + let (_, view) = signed_view( + &payer, + [Instruction::new_with_bytes(V42_ID, &[], vec![])], + snapshot_hash, + ); + let transaction = + ResolvedTransaction::try_new(view, Some(Default::default()), &Default::default()).unwrap(); + assert!( + keeper.transactions().append(&transaction).await.unwrap(), + "snapshot-retained non-latest hash remains valid without ledger history" + ); + keeper.close().await; +} + /// Stores the recovery marker account at `lamports`, the value each snapshot /// captures and recovery must bring back. fn store_marker(keeper: &TestKeeper, marker: Pubkey, lamports: u64) { diff --git a/ledger/src/reader.rs b/ledger/src/reader.rs index c22e0b2..af7d0cf 100644 --- a/ledger/src/reader.rs +++ b/ledger/src/reader.rs @@ -230,7 +230,7 @@ impl LedgerReader { self.populate_block_response(&superblock, request, span).map(Some) } - /// Reads block boundaries for a slot range in ascending slot order. + /// Reads block boundaries for a slot range in descending slot order. fn blocks(&mut self, range: Range) -> Result> { let mut blocks = Vec::with_capacity(range.clone().count()); let Some(last) = range.end.checked_sub(1) else { return Ok(blocks) }; @@ -251,7 +251,6 @@ impl LedgerReader { } } } - blocks.reverse(); Ok(blocks) } diff --git a/ledger/src/tests/integration.rs b/ledger/src/tests/integration.rs index 358d1e7..6da03c5 100644 --- a/ledger/src/tests/integration.rs +++ b/ledger/src/tests/integration.rs @@ -496,7 +496,7 @@ async fn test_reopen_resumes_state() { assert_eq!(ledger.meta.blocks.load(Acquire), 2); } -// A block-range read returns every block in the range, in ascending slot order, +// A block-range read returns every block in the range, in descending slot order, // even when the range straddles a superblock boundary. The reader walks slots // descending across superblocks newest-first, so a boundary slot must be handed // to the older segment instead of being consumed against the newer one. @@ -519,13 +519,13 @@ async fn test_block_range_spans_superblocks() { .map(|b| b.slot) .collect::>() }; - // The full range comes back once each, in order, across the boundary. - assert_eq!(slots(1..4).await, vec![1, 2, 3]); + // The full range comes back once each, newest-first, across the boundary. + assert_eq!(slots(1..4).await, vec![3, 2, 1]); // A sub-range inside one segment returns only its blocks. assert_eq!(slots(2..3).await, vec![2]); // A tail past the retained tip yields the retained blocks without dropping // the boundary slot. - assert_eq!(slots(1..9).await, vec![1, 2, 3]); + assert_eq!(slots(1..9).await, vec![3, 2, 1]); } // The account index keeps one entry per touching transaction, excludes unrelated