From 39ea7555b0ca741a2301b0144c3dcd34778999cc Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 6 Nov 2026 17:06:16 +0400 Subject: [PATCH 1/4] fix!: replicate rent and features at reset boundaries --- Cargo.lock | 1 - accountsdb/README.md | 7 +- accountsdb/src/lib.rs | 17 +- accountsdb/src/tests.rs | 54 +++- accountsdb/src/volatile.rs | 13 +- engine/src/lib.rs | 8 +- engine/src/pacemaker.rs | 11 +- engine/tests/recovery.rs | 6 - engine/tests/transactions.rs | 16 + keeper/README.md | 10 +- keeper/src/accessor.rs | 17 +- keeper/src/builder.rs | 273 ++++++++---------- keeper/src/cache.rs | 9 +- keeper/src/error.rs | 4 +- keeper/src/lib.rs | 139 ++++++--- keeper/src/tests/caches.rs | 3 +- keeper/src/tests/recovery.rs | 70 ++++- ledger/src/appender.rs | 3 +- nucleus/Cargo.toml | 3 - nucleus/src/ledger.rs | 56 ++-- processor/src/executor.rs | 4 +- processor/src/sequencer/mod.rs | 34 ++- processor/src/sequencer/tests.rs | 2 + processor/src/simulator.rs | 4 +- processor/src/svm.rs | 59 ++-- processor/src/tests.rs | 83 +++++- replicator/src/tests.rs | 14 +- replicator/tests/integration.rs | 60 +++- solana/account/README.md | 12 +- solana/account/src/mode.rs | 93 +++--- solana/account/src/tests/account.rs | 26 +- solana/program-runtime/src/loaded_programs.rs | 8 + solana/program-runtime/src/sysvar_cache.rs | 8 + .../src/instruction_accounts.rs | 1 + 34 files changed, 723 insertions(+), 405 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e5f1176a..7d9e2cd4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2368,7 +2368,6 @@ name = "magicblock-engine-nucleus" version = "0.22.0" dependencies = [ "agave-transaction-view", - "bytemuck", "derive_more", "futures", "oneshot", diff --git a/accountsdb/README.md b/accountsdb/README.md index a9807913..902752c6 100644 --- a/accountsdb/README.md +++ b/accountsdb/README.md @@ -2,13 +2,14 @@ Accountsdb stores current account state; the [ledger](../ledger/README.md) retains execution history. The store separates account state that requires persistence -from rebuildable base mirrors and runtime state, which live in memory. +from locally supplied programs and rebuildable base mirrors, which live in memory. Routing follows [lifecycle mode](../solana/account/README.md#account-lifecycle), -not memory representation. Delegated, Magic, and undelegating accounts are persisted, +not memory representation. Delegated, Magic, Runtime, and undelegating accounts are persisted, including undelegating state that no longer permits user writes. +`BuiltIn` program images are volatile, read-only, and have no base-chain sync. Closed accounts are removed. Reset discards mirrored state while preserving -persisted and runtime accounts. +ER-owned state, including volatile local programs. ## Storage access and writeback diff --git a/accountsdb/src/lib.rs b/accountsdb/src/lib.rs index 2ca1db51..b3db8546 100644 --- a/accountsdb/src/lib.rs +++ b/accountsdb/src/lib.rs @@ -38,7 +38,7 @@ const ACTIVE_DIR: &str = "CURRENT"; pub struct AccountsDB { /// On-disk store for account modes that require persistence. persisted: PersistedStore, - /// Rebuildable in-memory store for mirrors and runtime accounts. + /// In-memory store for programs and base-chain mirrors. volatile: VolatileStore, /// Database root directory. root: PathBuf, @@ -65,6 +65,12 @@ impl AccountsDB { Ok(db) } + /// Whether persisted storage has no allocated account images. + /// Genesis seeding makes this false; reopening alone does not. + pub fn is_new(&self) -> bool { + self.persisted.storage.cursor() == 0 + } + /// Returns the active database directory under `root`. pub fn directory(root: &Path) -> PathBuf { root.join(ACTIVE_DIR) @@ -230,12 +236,8 @@ impl AccountsDB { self.persisted.checksum().map_err(Into::into) } - /// Drops chain-mirrored volatile state while retaining system accounts; - /// persisted state is left untouched. - /// - /// Chain-owned accounts can be fetched again when synchronization resumes. - /// System accounts hold internal runtime state and survive the reset; their - /// volatile owner indexes are rebuilt. Persisted state is never reset. + /// Drops rebuildable chain mirrors and their owner indexes. + /// ER-owned state, including local programs and runtime accounts, is retained. pub fn reset(&self) { self.volatile.reset(); } @@ -421,6 +423,7 @@ fn persisted((_, account): &&AccountEntry) -> bool { // Mode changes still need persisted cleanup after borrowed data grows into owned data. matches!(account.cow(), CoWAccount::Borrowed(_)) || account.mode().requires_persistence() + || account.is(AccountMode::Closed) || account.markers().contains(DirtyMarkers::MODE) } diff --git a/accountsdb/src/tests.rs b/accountsdb/src/tests.rs index 900e747b..276e0bff 100644 --- a/accountsdb/src/tests.rs +++ b/accountsdb/src/tests.rs @@ -129,7 +129,7 @@ fn mutable_at_least(lamports: u64, units: u64, owner: &Pubkey) -> AccountSharedD } /// Proves lifecycle changes route accounts and owner indexes between backends, -/// while closing removes state and reset retains persisted and runtime accounts. +/// while closing removes state and reset retains ER-owned state, including volatile programs. #[test] fn test_routing_and_persistence_flips() { let (_dir, db) = db(); @@ -137,7 +137,11 @@ fn test_routing_and_persistence_flips() { let (a, b) = (Pubkey::new_unique(), Pubkey::new_unique()); let aacc = delegated_account(10, vec![], p); - let bacc = AccountBuilder::default().lamports(20).owner(p); + let bacc = AccountBuilder::default() + .lamports(20) + .owner(p) + .mode(AccountMode::Mirrored) + .executable(true); // `a` is authoritative, `b` is non-authoritative; both are owned by `p`. db.store(&[(a, aacc.build()), (b, bacc.build())]).unwrap(); assert!(in_persisted(&db, &a) && !in_volatile(&db, &a)); @@ -203,12 +207,40 @@ fn test_routing_and_persistence_flips() { .mode(AccountMode::Runtime) .build(), ); - assert!(in_volatile(&db, &runtime) && !in_persisted(&db, &runtime)); + assert!(!in_volatile(&db, &runtime) && in_persisted(&db, &runtime)); + let allocations = || { + let stats = db.persisted.storage.stats(); + (stats.allocs.load(Relaxed), stats.reallocs.load(Relaxed)) + }; + let before = allocations(); + for lamports in [41, 42] { + let mut account = reload(&db, &runtime); + account.set_lamports(lamports); + store(&db, runtime, account); + } + assert_eq!( + allocations(), + before, + "fixed-size runtime updates commit without reallocating" + ); + let builtin = Pubkey::new_unique(); + store( + &db, + builtin, + AccountBuilder::default() + .owner(p) + .mode(AccountMode::BuiltIn) + .executable(true) + .build(), + ); + assert!(in_volatile(&db, &builtin) && !in_persisted(&db, &builtin)); db.reset(); assert!(!in_volatile(&db, &b)); assert!(in_persisted(&db, &c)); - assert!(in_volatile(&db, &runtime)); + assert!(in_persisted(&db, &runtime)); assert!(program(&db, &p).contains(&runtime)); + assert!(in_volatile(&db, &builtin)); + assert!(program(&db, &p).contains(&builtin)); } // Both migration directions remove the source image and owner mapping, retain @@ -346,16 +378,30 @@ fn test_dump_restores_volatile_on_reopen() { { let db = AccountsDB::new(dir.path()).unwrap(); + assert!(db.is_new(), "persisted storage starts unseeded"); let account = AccountBuilder::default().lamports(42).mode(AccountMode::Mirrored).build(); store(&db, key, account); + assert!( + db.is_new(), + "volatile mirrors do not initialize persisted storage" + ); db.dump(None).unwrap(); assert!(dump.exists(), "dump is written into the active tree"); } let db = AccountsDB::new(dir.path()).unwrap(); + assert!(db.is_new(), "reopening does not seed persisted storage"); assert_eq!(lamports(&db, &key), 42); assert!(in_volatile(&db, &key)); assert!(!dump.exists(), "restored dump is consumed on open"); + store(&db, Pubkey::new_unique(), delegated(1)); + assert!( + !db.is_new(), + "persisted allocation marks the store initialized" + ); + db.flush(true).unwrap(); + drop(db); + assert!(!AccountsDB::new(dir.path()).unwrap().is_new()); } // A freed span is reused for a same-sized insert instead of growing the file; diff --git a/accountsdb/src/volatile.rs b/accountsdb/src/volatile.rs index 8fffe4d5..23087917 100644 --- a/accountsdb/src/volatile.rs +++ b/accountsdb/src/volatile.rs @@ -108,15 +108,18 @@ impl VolatileStore { self.programs.read_sync(owner, |_, s| s.clone()).unwrap_or_default() } - /// Drops chain-mirrored accounts while retaining internal system accounts. + /// Drops base-chain state while retaining ER-owned programs and their indexes. pub(crate) fn reset(&self) { self.programs.clear_sync(); - self.accounts.retain_sync(|k, a| { - if !a.is(AccountMode::Runtime) { + self.accounts.retain_sync(|key, account| { + if !account.mode().is_er_owned() { return false; } - let mut set = self.programs.entry_sync(a.owner()).or_default(); - BTreeSet::insert(&mut set, *k) + BTreeSet::insert( + &mut self.programs.entry_sync(account.owner()).or_default(), + *key, + ); + true }); } diff --git a/engine/src/lib.rs b/engine/src/lib.rs index 72145b9e..05eb6201 100644 --- a/engine/src/lib.rs +++ b/engine/src/lib.rs @@ -92,8 +92,12 @@ impl Engine { cache.assign_program(id, entry.into()); } let blockstore = builder.blockstore; + let rent = builder.rent.clone(); let state = Arc::new(builder.build(shutdown).await?); Self::try_replay(&state, &cache, cpus).await?; + if pacer.is_none() { + state.reset(rent)?; + } let (service, sequencer) = Sequencer::new(cpus / 2, state.clone(), cache, shutdown, false)?; service.spawn()?; let terminating = Arc::new(AtomicBool::new(false)); @@ -190,7 +194,7 @@ impl Engine { TransactionVerifier::new(self.authority()) } - /// Drains in-flight execution and keeps the sequencer paused until the handle is dropped. + /// Drains execution and simulation, keeping both paused until the handle is dropped. pub async fn barrier(&self) -> Result { self.pause(SequencerMessage::Barrier).await } @@ -237,7 +241,7 @@ impl Engine { } OwnedBlockstoreEntry::Reset(reset) => { let _guard = self.barrier().await?; - self.apply_reset(*reset)?; + self.apply_reset(&reset.payload)?; } OwnedBlockstoreEntry::Checkpoint(expected) => { let _guard = self.barrier().await?; diff --git a/engine/src/pacemaker.rs b/engine/src/pacemaker.rs index 2b574673..89984618 100644 --- a/engine/src/pacemaker.rs +++ b/engine/src/pacemaker.rs @@ -93,9 +93,8 @@ impl ExternalBlock { impl PaceMaker { /// Registers and starts the pacemaker task. /// - /// Uses an external block source when supplied. Otherwise it records one - /// reset at the keeper's current slot, clears chain-mirrored volatile state, - /// and starts emitting slots on the configured block interval. + /// Uses an external block source when supplied; otherwise emits slots on + /// the configured block interval. pub(crate) fn spawn( engine: Engine, pacer: Option, @@ -104,11 +103,7 @@ impl PaceMaker { ) -> Result<()> { let pacer = match pacer { Some(rx) => Pacer::External(rx), - None => { - let ticker = BlockTicker::new(&engine, blockstore.blocktime); - engine.reset(ticker.slot)?; - Pacer::Internal(ticker) - } + None => Pacer::Internal(BlockTicker::new(&engine, blockstore.blocktime)), }; let shutdown = shutdown.handle(Service::PaceMaker); let superblock = blockstore.superblock; diff --git a/engine/tests/recovery.rs b/engine/tests/recovery.rs index 7ba7c116..0f4d624c 100644 --- a/engine/tests/recovery.rs +++ b/engine/tests/recovery.rs @@ -205,12 +205,6 @@ async fn clean_restart_reopens_persisted_and_volatile_state() { let (dirs, authority) = te.close().await; let te2 = TestEngine::with(dirs, authority).await; - // Leader startup reconstructs the block timestamp as zero; epoch semantics - // depend only on the recovered slot, not on that existing timestamp policy. - let expected_clock = solana_sysvar::clock::Clock { - unix_timestamp: 0, - ..expected_clock - }; assert_eq!(te2.clock(te2.blocks().latest()), expected_clock); let stored: solana_sysvar::clock::Clock = te2 .get_account(solana_sdk_ids::sysvar::clock::ID) diff --git a/engine/tests/transactions.rs b/engine/tests/transactions.rs index 25d882f2..084653e0 100644 --- a/engine/tests/transactions.rs +++ b/engine/tests/transactions.rs @@ -3,6 +3,8 @@ //! receipts, signature observers, duplicate admission, and client wire formats. #![cfg(test)] +use std::time::Duration; + use agave_transaction_view::MAX_STANDARD_TRANSACTION_SIZE; use engine::testkit::TestEngine; use keeper::testkit::{ @@ -14,6 +16,7 @@ use solana_account::{AccountMode, ReadableAccount}; use solana_instruction_error::InstructionError; use solana_packet::PACKET_DATA_SIZE; use solana_transaction::TransactionError; +use tokio::time; use v42_calculator_interface::builder::{Expr as E, transfer}; // The Engine accepts the same standard wire formats produced by Solana clients @@ -231,6 +234,7 @@ async fn failed_execution_surfaces_error_result() { /// Proves scheduling acknowledges queueing before admission and repeated bytes /// execute once without consuming signature observers registered around resends. +/// The same barrier also holds simulation until execution is allowed to resume. #[tokio::test(flavor = "multi_thread")] async fn schedule_is_fire_and_forget() { let te = TestEngine::new().await; @@ -240,6 +244,15 @@ async fn schedule_is_fire_and_forget() { signed_view(&te, None, (E::acc(0) + E::lit(7)).compose(output, &[])); let first = te.transactions().subscribe_signature(signature).await.unwrap(); let barrier = te.barrier().await.unwrap(); + { + let simulation = te + .transaction(transaction.inner_data().as_ref().clone()) + .unwrap() + .simulate(true); + tokio::pin!(simulation); + // Poll the request into the simulator queue; it must not run across this boundary. + assert!(time::timeout(Duration::from_millis(50), &mut simulation).await.is_err()); + } for _ in 0..3 { te.transaction(transaction.inner_data().as_ref().clone()) .unwrap() @@ -259,6 +272,9 @@ async fn schedule_is_fire_and_forget() { Err(oneshot::TryRecvError::Empty) )); drop(barrier); + let txn = transaction.inner_data().as_ref().clone(); + let simulation = te.transaction(txn).unwrap().simulate(true); + assert!(simulation.await.unwrap().is_ok()); assert_eq!(first.await.unwrap().result, Ok(())); assert_eq!(second.await.unwrap().result, Ok(())); assert_eq!( diff --git a/keeper/README.md b/keeper/README.md index 92d622c3..91e7f80f 100644 --- a/keeper/README.md +++ b/keeper/README.md @@ -9,7 +9,10 @@ to interpret their state. ## Startup and recovery Startup validates account storage against ledger progress, restores a retained -snapshot when necessary, and seeds programs, features, and sysvars. +snapshot when necessary, and seeds features and sysvars only on genesis. +Later startups preserve those account images and restore runtime configuration from them. +Locally supplied native and SBF programs are volatile `BuiltIn` accounts, reloaded +from builder inputs on every startup and retained across mirror resets. [Engine](../engine/README.md#startup-and-recovery) replays the remaining history. Recovery can fail rather than open inconsistent state and depends on valid retained snapshots and history. @@ -67,8 +70,9 @@ irreversibly closes storage services. Keeper supplies Solana sysvars without validator epoch policy. Its informational epoch schedule follows local `blockstore.superblock` configuration, without warmup; -zero disables periodic sealing and uses 432,000-slot epochs. Schedules can differ -between peers or change on restart. Sealing doesn't itself advance the schedule. +zero disables periodic sealing and uses 432,000-slot epochs. The schedule is seeded +at genesis and restored from account state; changing the local sealing interval +doesn't replace it. Sealing doesn't itself advance the schedule. Execution and simulation see Clock at the slot after the latest completed block, with epochs derived from that slot and Unix timestamp from the block. The unsupported diff --git a/keeper/src/accessor.rs b/keeper/src/accessor.rs index db91d9c5..27cffa97 100644 --- a/keeper/src/accessor.rs +++ b/keeper/src/accessor.rs @@ -37,7 +37,7 @@ use tokio::sync::mpsc::{Receiver, UnboundedReceiver}; use crate::{ AccountLease, ExecutionRecord, FullTransaction, Keeper, ResolvedTransaction, - error::Result, + error::{KeeperError, Result}, util::{execution_commit, request, transaction_status}, }; @@ -93,22 +93,21 @@ impl<'a> AccountsAccessor<'a> { /// Updates stored `SlotHashes` and `Clock` sysvar accounts from `block`. /// - /// If either sysvar account is absent, no account updates are stored. + /// Both sysvars must exist; fixed-capacity borrowed commits preserve their index offsets. pub fn update_sysvars(&self, block: Block) -> Result<()> { let loader = self.loader(); - let Some(mut hacc) = loader.read(&SlotHashes::id(), Clone::clone)? else { - return Ok(()); - }; + // SAFETY: the sequencer is the exclusive sysvar writer and excludes compaction + // through this block transition. The guarded loader outlives both borrowed images. + let mut hacc = unsafe { loader.load(&SlotHashes::id()) }? + .ok_or(KeeperError::MissingSysvar(SlotHashes::id()))?; + let mut cacc = + unsafe { loader.load(&Clock::id()) }?.ok_or(KeeperError::MissingSysvar(Clock::id()))?; let mut hashes: SlotHashes = hacc.deserialize_data().map_err(AccountsDBError::from)?; hashes.add(block.slot, block.hash); hacc.serialize_data(&hashes).map_err(AccountsDBError::from)?; - let Some(mut cacc) = loader.read(&Clock::id(), Clone::clone)? else { - return Ok(()); - }; let clock = self.keeper.clock(block); cacc.serialize_data(&clock).map_err(AccountsDBError::from)?; - drop(loader); self.store(&[(SlotHashes::id(), hacc), (Clock::id(), cacc)]).map_err(Into::into) } } diff --git a/keeper/src/builder.rs b/keeper/src/builder.rs index 905b0ba6..eec81372 100644 --- a/keeper/src/builder.rs +++ b/keeper/src/builder.rs @@ -6,11 +6,14 @@ use std::{ time::Duration, }; -use accountsdb::{AccountEntry, AccountsDB, AccountsDBError, BackupOp, SnapshotError}; +use accountsdb::{ + AccountEntry, AccountLoader, AccountsDB, AccountsDBError, BackupOp, SnapshotError, +}; use agave_feature_set::FeatureSet; +use arc_swap::ArcSwap; use ledger::{ Ledger, LedgerHandle, - request::{BlockDetails, BlockParams, BlockRangeParams, ReadRequest, RequestPayload}, + request::{BlockRangeParams, ReadRequest, RequestPayload}, }; use nucleus::{ Slot, @@ -18,28 +21,32 @@ use nucleus::{ ledger::{ACCOUNTSDB_SNAPSHOT_FILE, Block}, shutdown::ShutdownManager, }; -use serde::Serialize; -use solana_account::{AccountBuilder, AccountMode, AccountSharedData, ReadableAccount}; -use solana_feature_gate_interface::Feature; +use serde::de::DeserializeOwned; +use solana_account::{AccountBuilder, AccountMode, AccountSharedData}; use solana_program_runtime::invoke_context::BuiltinFunctionWithContext; use solana_pubkey::Pubkey; use solana_sdk_ids::sysvar; -#[allow(deprecated)] -use solana_sysvar::fees::Fees; use solana_sysvar::{ clock::Clock, epoch_rewards::EpochRewards, epoch_schedule::EpochSchedule, + fees, last_restart_slot::LastRestartSlot, rent::Rent, slot_hashes::{SlotHashes, SysvarId}, }; +use tar::Archive; use tokio::sync::mpsc; use tracing::{error, info, warn}; +use zstd::Decoder; use crate::{ - Keeper, LEDGER_STREAM_CAPACITY, cache::Caches, error::Result, metrics, + Keeper, LEDGER_STREAM_CAPACITY, + cache::Caches, + error::{KeeperError, Result}, + feature_account, metrics, runtime_account, subscriptions::Subscriptions, + sysvar_account, }; /// Initial balance assigned to the authority account that sponsors account creation. @@ -62,13 +69,13 @@ pub struct KeeperBuilder { pub ledger: LedgerParams, /// Block production timing and superblock sealing parameters. pub blockstore: BlockstoreParams, - /// Native builtin program ids to seed as executable accounts. + /// Native builtin program ids to install as volatile executable accounts. pub builtins: HashMap, - /// SBF programs to seed as executable loader-v4 accounts: `(program id, ELF bytes)`. + /// SBF programs to reload as volatile loader-v4 accounts: `(program id, ELF bytes)`. pub programs: HashMap>, /// Accounts to seed at startup if missing from the database; existing accounts are kept. pub possibly_missing_accounts: HashMap, - /// Rent parameters used to size seeded accounts and the Rent sysvar. + /// Startup rent for a new leader; recovery uses the persisted Rent sysvar. pub rent: Rent, } @@ -78,13 +85,27 @@ impl KeeperBuilder { let ledger = Ledger::init(&self.ledger.directory, self.ledger.size_limit, shutdown)?; let accountsdb = self.accountsdb(&ledger)?; accountsdb.cleanup_snapshots()?; - let (caches, featureset) = self.prepopulate(&accountsdb, &ledger).await?; + self.prepopulate(&accountsdb)?; + let featureset = Self::restore_features(&accountsdb)?; + let (rent, epoch_schedule, slothashes, clock) = { + let loader = accountsdb.loader(); + ( + load_sysvar::(&loader)?, + load_sysvar::(&loader)?, + load_sysvar::(&loader)?, + load_sysvar::(&loader)?, + ) + }; + let caches = if self.authority.remote.is_some() { + self.follower_caches(&ledger, accountsdb.slot(), &slothashes, &clock).await? + } else { + self.leader_caches(&slothashes, &clock) + }; metrics::init(); - let epoch_schedule = self.epoch_schedule(); Ok(Keeper { authority: self.authority, - featureset, - rent: self.rent, + featureset: ArcSwap::from_pointee(featureset), + rent: rent.lamports_per_byte.into(), epoch_schedule, accountsdb, ledger, @@ -93,16 +114,16 @@ impl KeeperBuilder { }) } - /// Seeds accounts needed before the engine starts serving reads. - async fn prepopulate( - &mut self, - accountsdb: &AccountsDB, - ledger: &LedgerHandle, - ) -> Result<(Caches, FeatureSet)> { + /// Reloads local programs, initializes genesis runtime state, and seeds missing accounts. + fn prepopulate(&mut self, accountsdb: &AccountsDB) -> Result<()> { let mut accounts = Vec::new(); - let featureset = self.seed_featureset(&mut accounts)?; - self.seed_programs(&mut accounts)?; - let caches = self.seed_sysvars(accountsdb, ledger, &mut accounts).await?; + if accountsdb.is_new() { + for id in DEFAULT_FEATURES { + accounts.push((id, feature_account())); + } + self.seed_sysvars(&mut accounts); + } + self.seed_programs(&mut accounts); let loader = accountsdb.loader(); let authority = self.authority.pubkey(); if !loader.contains(&authority)? { @@ -117,8 +138,7 @@ impl KeeperBuilder { } } drop(loader); - accountsdb.store(&accounts)?; - Ok((caches, featureset)) + accountsdb.store(&accounts).map_err(Into::into) } /// Epochs follow the local sealing interval; zero uses Solana's no-warmup default. @@ -147,136 +167,64 @@ impl KeeperBuilder { 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 { - let mut featureset = FeatureSet::default(); - [ - agave_feature_set::curve25519_syscall_enabled::ID, - agave_feature_set::curve25519_restrict_msm_length::ID, - agave_feature_set::enable_poseidon_syscall::ID, - agave_feature_set::enable_sbpf_v3_deployment_and_execution::ID, - agave_feature_set::virtual_address_space_adjustments::ID, - agave_feature_set::syscall_parameter_address_restrictions::ID, - agave_feature_set::get_sysvar_syscall_enabled::ID, - agave_feature_set::ed25519_program_enabled::ID, - agave_feature_set::secp256k1_program_enabled::ID, - agave_feature_set::enable_secp256r1_precompile::ID, - ] - .iter() - .for_each(|f| featureset.activate(f, 0)); - for (&id, &slot) in featureset.active() { - let feature = &Feature { activated_at: Some(slot) }; - let account = self.account(feature, &solana_feature_gate_interface::ID)?; - accounts.push((id, account.build())); + /// Reconstructs feature membership from authoritative runtime accounts. + fn restore_features(accountsdb: &AccountsDB) -> Result { + let mut features = FeatureSet::default(); + let accounts = accountsdb.program(&solana_feature_gate_interface::ID, |_, account| { + account.mode() + })?; + for (id, _) in accounts.filter(|(_, mode)| *mode == AccountMode::Runtime) { + features.activate(&id, 0); } - Ok(featureset) + Ok(features) } - /// Seeds native-loader accounts for builtins and precompiles, and raw ELF - /// accounts owned by loader-v4 for configured SBF programs. - fn seed_programs(&self, accounts: &mut Vec) -> Result<()> { + /// Reloads native-loader builtins and precompiles, and loader-v4 SBF programs. + fn seed_programs(&mut self, accounts: &mut Vec) { let precompiles = [ solana_sdk_ids::ed25519_program::ID, solana_sdk_ids::secp256k1_program::ID, solana_sdk_ids::secp256r1_program::ID, ]; - for program in self.builtins.keys().copied().chain(precompiles) { - let account = self.account(&(), &solana_sdk_ids::native_loader::ID)?; - let account = account.executable(true).build(); - accounts.push((program, account)); - } - - for (&program, elf) in &self.programs { - let lamports = self.rent.minimum_balance(elf.len()); + let builtins = self + .builtins + .keys() + .copied() + .chain(precompiles) + .map(|id| (id, solana_sdk_ids::native_loader::ID, Vec::new())); + let programs = + self.programs.drain().map(|(id, elf)| (id, solana_sdk_ids::loader_v4::ID, elf)); + for (program, owner, elf) in builtins.chain(programs) { + let lamports = Rent::default().minimum_balance(elf.len()); let account = AccountBuilder::default() .lamports(lamports) - .mode(AccountMode::Runtime) - .owner(solana_sdk_ids::loader_v4::ID) + .mode(AccountMode::BuiltIn) + .owner(owner) .executable(true) - .data(elf.clone()); + .data(elf); accounts.push((program, account.build())); } - Ok(()) } - /// Seeds block-derived sysvars and builds the caches needed for this node's role. - async fn seed_sysvars( - &self, - accountsdb: &AccountsDB, - ledger: &LedgerHandle, - accounts: &mut Vec, - ) -> Result { - let slot = accountsdb.slot(); - let id = SlotHashes::id(); - // AccountsDB starts at slot 1 and the ledger starts at head 1. - let genesis = slot == 1 && ledger.head() == 1; - let slothashes = accountsdb - .loader() - .read(&id, |account| account.deserialize_data::())?; - let slothashes = match slothashes { - Some(hashes) => hashes.map_err(AccountsDBError::from)?, - 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]); - if !genesis { - // Leader shutdown does not dump volatile sysvars. Recover only - // through AccountsDB's boundary, never an unapplied ledger tail. - let start = slot.saturating_sub(SLOTHASH_ENTRIES as Slot - 1).max(1); - for slot in start..=slot { - let params = BlockParams { - slot, - details: BlockDetails::None, - }; - let (payload, response) = RequestPayload::new(params); - ledger.reader.send(ReadRequest::Block(payload))?; - let Some(response) = response.recv().await?? else { continue }; - let block = response.block(); - hashes.add(block.slot, block.hash); - } - } - let account = self.account(&hashes, &sysvar::ID)?; - accounts.push((id, account.build())); - hashes - } - }; - - // A remote authority makes this engine a replication client. Followers - // need retained dedup history; leaders deliberately retain only the tip - // without reconstructing signature history. - let caches = if self.authority.remote.is_some() { - self.follower_caches(ledger, slot, &slothashes).await? - } else { - self.leader_caches(&slothashes) - }; - - let epoch_schedule = self.epoch_schedule(); - let clock = crate::clock(&epoch_schedule, caches.latest()); - accounts.push((Clock::id(), self.account(&clock, &sysvar::ID)?.build())); - accounts.push((Rent::id(), self.account(&self.rent, &sysvar::ID)?.build())); - #[allow(deprecated)] - accounts.push(( - Fees::id(), - self.account(&Fees::default(), &sysvar::ID)?.build(), - )); - accounts.push(( - sysvar::last_restart_slot::id(), - self.account(&LastRestartSlot::default(), &sysvar::ID)?.build(), - )); - accounts.push(( - sysvar::instructions::id(), - self.account(&(), &sysvar::ID)?.build(), - )); - accounts.push(( - EpochSchedule::id(), - self.account(&epoch_schedule, &sysvar::ID)?.build(), - )); - accounts.push(( - EpochRewards::id(), - self.account(&EpochRewards::default(), &sysvar::ID)?.build(), - )); - Ok(caches) + /// Creates the genesis sysvars; subsequent startup always reads persisted state. + fn seed_sysvars(&self, accounts: &mut Vec) { + let hashes = SlotHashes::new(&[Default::default(); SLOTHASH_ENTRIES]); + let schedule = self.epoch_schedule(); + let clock = crate::clock(&schedule, Block::default()); + accounts.extend([ + sysvar_account(&hashes), + sysvar_account(&clock), + sysvar_account(&self.rent), + #[allow(deprecated)] + sysvar_account(&fees::Fees::default()), + sysvar_account(&LastRestartSlot::default()), + ( + sysvar::instructions::ID, + runtime_account(&(), &sysvar::ID).build(), + ), + sysvar_account(&schedule), + sysvar_account(&EpochRewards::default()), + ]); } /// Streams retained history into follower caches without materializing the range. @@ -285,6 +233,7 @@ impl KeeperBuilder { ledger: &LedgerHandle, slot: Slot, slothashes: &SlotHashes, + clock: &Clock, ) -> Result { let retained = self .ttl(BLOCK_CACHE_WINDOW) @@ -298,7 +247,7 @@ impl KeeperBuilder { let (payload, response) = RequestPayload::new(params); ledger.reader.send(ReadRequest::BlockRange(payload))?; - let caches = self.caches(persisted_latest(slothashes)); + let caches = self.caches(persisted_latest(slothashes, clock)); // Merge the two ascending streams so ExpiringCache's eviction queue // stays ordered, preferring blockstore data for overlapping slots. let mut persisted = slothashes.slot_hashes().iter().rev().copied().peekable(); @@ -317,21 +266,13 @@ impl KeeperBuilder { } /// Restores only AccountsDB's latest boundary so replay can advance it from there. - fn leader_caches(&self, slothashes: &SlotHashes) -> Caches { - let latest = persisted_latest(slothashes); + fn leader_caches(&self, slothashes: &SlotHashes, clock: &Clock) -> Caches { + let latest = persisted_latest(slothashes, clock); let caches = self.caches(latest); caches.restore_hash(latest.slot, latest.hash); caches } - /// Builds a rent-exempt system account containing a serialized sysvar-like state. - fn account(&self, state: &S, owner: &Pubkey) -> Result { - let account = - AccountSharedData::new_data(0, state, owner).map_err(AccountsDBError::from)?; - let lamports = self.rent.minimum_balance(account.data().len()); - Ok(AccountBuilder::from(account).lamports(lamports).mode(AccountMode::Runtime)) - } - /// Opens accountsdb, restoring the newest archived snapshot after corruption. /// /// A restored store trails the ledger tip — snapshots are archived at sealed @@ -388,7 +329,7 @@ impl KeeperBuilder { } let dst = AccountsDB::directory(&self.accountsdb.directory); let file = File::open(src)?; - let mut tar = tar::Archive::new(zstd::Decoder::new(file)?); + let mut tar = Archive::new(Decoder::new(file)?); tar.unpack(dst)?; info!(directory = ?superblock.directory, "restored accountsdb snapshot"); return Ok(()); @@ -397,12 +338,34 @@ impl KeeperBuilder { } } -fn persisted_latest(slothashes: &SlotHashes) -> Block { +fn persisted_latest(slothashes: &SlotHashes, clock: &Clock) -> Block { let hashes = slothashes.slot_hashes(); hashes.first().map_or(Block::default(), |&(slot, hash)| Block { slot, hash, - time: 0, + time: clock.unix_timestamp, parent: hashes.get(1).map(|(_, hash)| *hash).unwrap_or_default(), }) } + +/// Engine-selected features published by a leader at startup. +pub(crate) const DEFAULT_FEATURES: [Pubkey; 10] = [ + agave_feature_set::curve25519_syscall_enabled::ID, + agave_feature_set::curve25519_restrict_msm_length::ID, + agave_feature_set::enable_poseidon_syscall::ID, + agave_feature_set::enable_sbpf_v3_deployment_and_execution::ID, + agave_feature_set::virtual_address_space_adjustments::ID, + agave_feature_set::syscall_parameter_address_restrictions::ID, + agave_feature_set::get_sysvar_syscall_enabled::ID, + agave_feature_set::ed25519_program_enabled::ID, + agave_feature_set::secp256k1_program_enabled::ID, + agave_feature_set::enable_secp256r1_precompile::ID, +]; + +fn load_sysvar(loader: &AccountLoader<'_>) -> Result { + loader + .read(&S::id(), |account| account.deserialize_data::())? + .ok_or(KeeperError::MissingSysvar(S::id()))? + .map_err(AccountsDBError::from) + .map_err(Into::into) +} diff --git a/keeper/src/cache.rs b/keeper/src/cache.rs index e2f19147..11938e41 100644 --- a/keeper/src/cache.rs +++ b/keeper/src/cache.rs @@ -64,11 +64,6 @@ impl Caches { self.signatures.push(signature, None, block.slot); } } - - /// Returns the latest restored block boundary. - pub(crate) fn latest(&self) -> Block { - **self.blocks.latest.load() - } } /// Account access cache along with per-account mutation reservations. @@ -129,7 +124,7 @@ impl AccountLease { /// Returns whether the observed account may be evicted from the cache. pub fn evictable(&self) -> bool { - self.observed.is_some_and(|(mode, _)| !mode.requires_persistence()) + self.observed.is_some_and(|(mode, _)| !mode.is_er_owned()) } /// Returns the local mode when the incoming image is older or duplicates @@ -170,7 +165,7 @@ impl Drop for AccountLease { impl AccountCache { /// Tracks a materialized account and returns one displaced from recency. fn track(&self, pubkey: Pubkey, mode: AccountMode) -> Option { - if !mode.requires_persistence() { + if !mode.is_er_owned() { return self.admit_recency(pubkey); } self.remove_recency(&pubkey); diff --git a/keeper/src/error.rs b/keeper/src/error.rs index 87a6ff84..da7c07c1 100644 --- a/keeper/src/error.rs +++ b/keeper/src/error.rs @@ -1,5 +1,7 @@ //! Keeper error types. +use std::io; + use accountsdb::{AccountsDBError, SnapshotError}; use derive_more::From; use flume::SendError; @@ -12,7 +14,7 @@ use solana_pubkey::Pubkey; pub enum KeeperError { /// Filesystem or archive IO failed. #[error("io: {0}")] - IO(#[source] std::io::Error), + IO(#[source] io::Error), /// Accounts database operation failed. #[error("accountsdb: {0}")] AccountsDB(#[source] AccountsDBError), diff --git a/keeper/src/lib.rs b/keeper/src/lib.rs index 738a6d3d..20470f86 100644 --- a/keeper/src/lib.rs +++ b/keeper/src/lib.rs @@ -3,36 +3,46 @@ use std::{ fs::{self, File}, path::PathBuf, - sync::Arc, + sync::{ + Arc, + atomic::{AtomicU64, Ordering::*}, + }, thread, }; use agave_feature_set::FeatureSet; -use solana_account::AccountBuilder; +use arc_swap::ArcSwap; +use serde::Serialize; +use solana_account::{AccountBuilder, AccountMode, AccountSharedData, ReadableAccount}; +use solana_feature_gate_interface::Feature; use solana_hash::Hash; use solana_keypair::Keypair; use solana_pubkey::Pubkey; +use solana_sdk_ids::sysvar; +use tar::Builder; use tokio::sync::mpsc; use tracing::{error, info, warn}; +use zstd::Encoder; -use accountsdb::{AccountsDB, SnapshotError}; +use accountsdb::{AccountEntry, AccountsDB, SnapshotError, SnapshotResult}; use ledger::{ LedgerHandle, Superblock, request::{ReadRequest, ReplayHandle, ReplayParams, RequestPayload}, schema::Event, }; use nucleus::{ - Slot, config::Authority, - ledger::{ACCOUNTSDB_SNAPSHOT_FILE, Checkpoint, Reset, Signed, SuperblockSeal}, + ledger::{ACCOUNTSDB_SNAPSHOT_FILE, Block, Checkpoint, Reset, Signed, SuperblockSeal}, +}; +use solana_sysvar::{ + clock::Clock, epoch_schedule::EpochSchedule, rent::Rent, slot_hashes::SysvarId, }; -use solana_sysvar::{clock::Clock, epoch_schedule::EpochSchedule, rent::Rent}; use crate::{ accessor::{AccountsAccessor, BlocksAccessor, SuperblockAccessor, TransactionsAccessor}, - builder::SPONSOR_INIT_BALANCE, + builder::{DEFAULT_FEATURES, SPONSOR_INIT_BALANCE}, cache::Caches, - error::Result, + error::{KeeperError, Result}, metrics::Operation, subscriptions::Subscriptions, }; @@ -66,9 +76,9 @@ pub struct Keeper { /// Local signer and optional remote authority represented by this engine. authority: Authority, /// Active feature set governing runtime behavior. - featureset: FeatureSet, + featureset: ArcSwap, /// Rent parameters applied during execution. - rent: Rent, + rent: AtomicU64, /// Local slot-based schedule shared by sysvars and execution. epoch_schedule: EpochSchedule, /// Account state store. @@ -172,7 +182,7 @@ impl Keeper { Some(seal) if seal.payload == payload => seal, Some(expected) => { error!(?expected, observed = ?payload, "superblock state mismatch"); - return Err(error::KeeperError::SealMismatch(expected.payload.id)); + return Err(KeeperError::SealMismatch(expected.payload.id)); } }; let completion = self.superblocks().append(seal)?; @@ -197,23 +207,24 @@ impl Keeper { Some(checkpoint) if checkpoint.payload == observed => checkpoint, Some(expected) => { error!(?expected, ?observed, "checkpoint state mismatch"); - return Err(error::KeeperError::CheckpointMismatch); + return Err(KeeperError::CheckpointMismatch); } }; self.ledger.appender.send(Event::Checkpoint(checkpoint))?; Ok(()) } - /// Resolved once from the seeded feature accounts at startup and fixed for - /// the engine's lifetime — features never activate mid-run. - pub fn features(&self) -> &FeatureSet { - &self.featureset + /// Immutable feature selection effective at the current execution boundary. + pub fn features(&self) -> Arc { + self.featureset.load_full() } - /// Supplied by the builder at startup and fixed for the engine's lifetime; - /// the same parameters that sized the seeded accounts. - pub fn rent(&self) -> &Rent { - &self.rent + /// Rent effective at the current ordered execution boundary. + pub fn rent(&self) -> Rent { + Rent { + lamports_per_byte: self.rent.load(Acquire), + ..Default::default() + } } /// Returns the informational epoch schedule derived from local superblock configuration. @@ -222,7 +233,7 @@ impl Keeper { } /// Clock for the slot executing after `block`. Epoch-start timestamps are unsupported. - pub fn clock(&self, block: nucleus::ledger::Block) -> Clock { + pub fn clock(&self, block: Block) -> Clock { clock(&self.epoch_schedule, block) } @@ -241,22 +252,52 @@ impl Keeper { self.accountsdb.flush(true).map_err(Into::into) } - /// Appends a reset marker before discarding chain-synchronized volatile - /// accounts and restoring the authority sponsor's initial balance. + /// Appends a startup reset establishing rent and Engine-default features, + /// discarding chain mirrors, and restoring the sponsor's initial balance. /// /// Internal runtime accounts and persisted state remain available. - pub fn reset(&self, slot: Slot) -> Result<()> { - self.append_reset(Signed::new(Reset(slot), self.signer())) + /// Execution and simulation must be quiesced. + pub fn reset(&self, rent: Rent) -> Result<()> { + let reset = Reset { + slot: self.blocks().current_slot(), + rent: rent.lamports_per_byte, + features: DEFAULT_FEATURES.to_vec(), + }; + self.append_reset(Signed::new(reset, self.signer())) } - /// Appends an authenticated reset before applying it. Execution must be quiesced. + /// Appends an authenticated reset before applying it. + /// Execution and simulation must be quiesced. pub fn append_reset(&self, reset: Signed) -> Result<()> { - self.ledger.appender.send(Event::Reset(reset))?; - self.apply_reset(reset.payload) + self.ledger.appender.send(Event::Reset(reset.clone()))?; + self.apply_reset(&reset.payload) } - /// Discards volatile state without appending; execution must be quiesced. - pub fn apply_reset(&self, reset: Reset) -> Result<()> { + /// Applies rent and discards volatile state without appending; execution and simulation must be quiesced. + pub fn apply_reset(&self, reset: &Reset) -> Result<()> { + let mut features = FeatureSet::default(); + for id in &reset.features { + features.activate(id, 0); + } + let previous = self.features(); + let mut accounts = Vec::new(); + for id in previous.active().keys().filter(|id| !features.is_active(id)) { + let acc = AccountBuilder::default().mode(AccountMode::Closed).build(); + accounts.push((*id, acc)); + } + for id in features.active().keys().filter(|id| !previous.is_active(id)) { + accounts.push((*id, feature_account())); + } + let rent = Rent { + lamports_per_byte: reset.rent, + ..Default::default() + }; + accounts.push(sysvar_account(&rent)); + self.accountsdb.store(&accounts)?; + if *previous != features { + self.featureset.store(Arc::new(features)); + } + self.rent.store(rent.lamports_per_byte, Release); self.accountsdb.reset(); let authority = self.authority(); let account = self.accounts().loader().read(&authority, Clone::clone)?; @@ -264,17 +305,13 @@ impl Keeper { let acc = AccountBuilder::from(account).lamports(SPONSOR_INIT_BALANCE); self.accounts().store(&[(authority, acc.build())])?; } - info!(slot = *reset, "reset volatile state"); + info!(slot = reset.slot, "reset volatile state"); Ok(()) } /// Spawns a background thread that tars and zstd-compresses the accountsdb /// snapshot at `snapshot` into `target`, removing the snapshot afterward. - fn archive( - &self, - snapshot: PathBuf, - target: PathBuf, - ) -> std::result::Result<(), SnapshotError> { + fn archive(&self, snapshot: PathBuf, target: PathBuf) -> SnapshotResult<()> { let path = target.join(ACCOUNTSDB_SNAPSHOT_FILE); let tmp = target.join(format!("{ACCOUNTSDB_SNAPSHOT_FILE}.tmp")); let dst = File::options().write(true).create(true).truncate(true).open(&tmp)?; @@ -282,7 +319,7 @@ impl Keeper { thread::Builder::new().name("snapshot-archiver".into()).spawn(move || { { let _timer = metrics::time(Operation::ArchiveSnapshot); - let mut tar = tar::Builder::new(zstd::Encoder::new(dst, 0)?); + let mut tar = Builder::new(Encoder::new(dst, 0)?); tar.append_dir_all(".", &snapshot)?; { let archive = tar.into_inner()?.finish()?; @@ -302,7 +339,7 @@ impl Keeper { } /// Shared by startup seeding and live transitions; seals never enter this calculation. -fn clock(schedule: &EpochSchedule, block: nucleus::ledger::Block) -> Clock { +fn clock(schedule: &EpochSchedule, block: Block) -> Clock { let slot = block.slot + 1; Clock { slot, @@ -313,3 +350,29 @@ fn clock(schedule: &EpochSchedule, block: nucleus::ledger::Block) -> Clock { epoch_start_timestamp: 0, } } + +/// Synthetic account funding is fixed so restarts and rent changes cannot alter +/// program-visible balances independently on leaders and followers. +#[expect( + clippy::expect_used, + reason = "All callers serialize known Engine-generated account state" +)] +fn runtime_account(state: &S, owner: &Pubkey) -> AccountBuilder { + let account = AccountSharedData::new_data(0, state, owner) + .expect("Engine-generated runtime accounts must be encodable"); + let lamports = Rent::default().minimum_balance(account.data().len()); + AccountBuilder::from(account).lamports(lamports).mode(AccountMode::Runtime) +} + +/// Active feature membership is persisted independently of startup defaults. +fn feature_account() -> AccountSharedData { + runtime_account( + &Feature { activated_at: Some(0) }, + &solana_feature_gate_interface::ID, + ) + .build() +} + +fn sysvar_account(state: &S) -> AccountEntry { + (S::id(), runtime_account(state, &sysvar::ID).build()) +} diff --git a/keeper/src/tests/caches.rs b/keeper/src/tests/caches.rs index a32aaa64..8bf73e20 100644 --- a/keeper/src/tests/caches.rs +++ b/keeper/src/tests/caches.rs @@ -115,7 +115,8 @@ async fn account_lease_coordinates_recency_and_waiters() { (AccountMode::Delegated, false), (AccountMode::Magic, false), (AccountMode::Undelegating, false), - (AccountMode::Runtime, true), + (AccountMode::Runtime, false), + (AccountMode::BuiltIn, false), ]; for (mode, tracked) in modes { let cache = Arc::new(AccountCache::new(256)); diff --git a/keeper/src/tests/recovery.rs b/keeper/src/tests/recovery.rs index 19bf5169..3f656712 100644 --- a/keeper/src/tests/recovery.rs +++ b/keeper/src/tests/recovery.rs @@ -4,8 +4,11 @@ use std::fs; use accountsdb::AccountEntry; use ledger::schema::Signed; -use nucleus::testkit::{V42_ID, block, signed_view}; -use solana_account::{ReadableAccount, testkit::delegated_account}; +use nucleus::{ + ledger::Reset, + testkit::{V42_ID, block, signed_view}, +}; +use solana_account::{AccountMode, ReadableAccount, testkit::delegated_account}; use solana_instruction::Instruction; use solana_keypair::Keypair; use solana_pubkey::Pubkey; @@ -20,13 +23,15 @@ use crate::ResolvedTransaction; use crate::testkit::{Dirs, archived_snapshot, corrupt, keeper_builder, seal_and_archive}; /// Proves startup seeds features, programs, and a no-warmup epoch schedule, -/// including sub-32-slot intervals and the zero-sealing fallback. +/// including fixed runtime funding under nondefault rent, sub-32-slot intervals, +/// and the zero-sealing fallback. #[tokio::test] async fn seeds_features_programs_and_sysvars() { for interval in [1, 7, 0] { let dirs = Dirs::default(); let mut builder = keeper_builder(&dirs); builder.blockstore.superblock = interval; + builder.rent.lamports_per_byte = 5080; let program = Pubkey::new_unique(); let elf = vec![1u8, 2, 3, 4, 5, 6, 7, 8]; builder.programs.insert(program, elf.clone()); @@ -65,6 +70,7 @@ async fn seeds_features_programs_and_sysvars() { // Builtins are seeded through the same path with an executable native-loader // account, so they share this shape. let acc = loader.read(&program, Clone::clone).unwrap().expect("program seeded"); + assert!(acc.is(AccountMode::BuiltIn)); assert!(acc.executable()); assert_eq!(acc.owner(), &loader_v4::ID); assert_eq!(acc.data(), elf.as_slice()); @@ -112,7 +118,7 @@ async fn seeds_features_programs_and_sysvars() { } } -/// Startup seeds missing accounts without replacing account state saved by a prior run. +/// Startup reloads local programs but preserves persisted state and seeds missing accounts. #[tokio::test] async fn preserves_seeded_accounts_on_reopen() { let existing = Pubkey::new_unique(); @@ -125,20 +131,33 @@ async fn preserves_seeded_accounts_on_reopen() { ); let keeper = TestKeeper::from_builder(dirs, builder.clone()).await; + let schedule = keeper.epoch_schedule().clone(); let marker_owner = Pubkey::new_unique(); let marker_data = vec![7, 7, 7]; let marker = delegated_account(2, marker_data.clone(), marker_owner); keeper.accounts().store(&[(existing, marker.build())]).unwrap(); let dirs = keeper.close().await; + builder.programs.insert(V42_ID, vec![9]); + builder.blockstore.superblock = 3; builder.possibly_missing_accounts.insert( added, delegated_account(3, vec![], Pubkey::default()).build(), ); let keeper = TestKeeper::from_builder(dirs, builder).await; + assert_eq!( + keeper.epoch_schedule(), + &schedule, + "restored schedule overrides local configuration" + ); { let accounts = keeper.accounts(); let loader = accounts.loader(); + assert_eq!( + loader.read(&V42_ID, |account| account.data().to_vec()).unwrap(), + Some(vec![9]), + "startup reloads the supplied program bytes" + ); assert_eq!( loader.read(&existing, |account| account.lamports()).unwrap(), Some(2) @@ -173,7 +192,7 @@ async fn preserves_seeded_accounts_on_reopen() { async fn recovers_the_newest_snapshot() { let marker = Pubkey::new_unique(); let dirs = Dirs::default(); - let builder = keeper_builder(&dirs); + let mut builder = keeper_builder(&dirs); let keeper = TestKeeper::from_builder(dirs, builder.clone()).await; let schedule = keeper.epoch_schedule().clone(); @@ -186,6 +205,21 @@ async fn recovers_the_newest_snapshot() { "snapshot archived under superblock" ); // Second snapshot, in a later superblock, captures marker == 2. + let mut rent = keeper.rent(); + rent.lamports_per_byte = 5080; + keeper + .apply_reset(&Reset { + slot: 0, + rent: rent.lamports_per_byte, + features: keeper + .features() + .active() + .keys() + .copied() + .filter(|id| *id != agave_feature_set::enable_poseidon_syscall::ID) + .collect(), + }) + .unwrap(); store_marker(&keeper, marker, 2); seal_and_archive(&keeper).await; assert_eq!( @@ -193,7 +227,13 @@ async fn recovers_the_newest_snapshot() { clock, "seals do not advance epochs" ); - keeper.apply_reset(nucleus::ledger::Reset(0)).unwrap(); + keeper + .apply_reset(&Reset { + slot: 0, + rent: rent.lamports_per_byte, + features: keeper.features().active().keys().copied().collect(), + }) + .unwrap(); assert_eq!( keeper.epoch_schedule(), &schedule, @@ -207,7 +247,25 @@ async fn recovers_the_newest_snapshot() { // valid checksum over the poisoned word. corrupt(dirs.accounts.path(), 8, 0xABAB_ABAB_ABAB_ABAB); + builder.rent = Rent::default(); let keeper = TestKeeper::from_builder(dirs, builder).await; + assert_eq!( + keeper.rent(), + rent, + "snapshot rent wins over local defaults" + ); + assert!( + !keeper.features().active().is_empty(), + "snapshot restores active features" + ); + assert!( + !keeper.features().is_active(&agave_feature_set::enable_poseidon_syscall::ID), + "startup defaults must not override restored feature membership" + ); + assert!( + !keeper.accounts().is_new(), + "snapshot restoration does not create a fresh store" + ); keeper.accounts().validate().expect("restored store validates"); assert_eq!(keeper.epoch_schedule(), &schedule); assert_eq!(keeper.clock(keeper.blocks().latest()), clock); diff --git a/ledger/src/appender.rs b/ledger/src/appender.rs index d654167f..d3d195d3 100644 --- a/ledger/src/appender.rs +++ b/ledger/src/appender.rs @@ -214,9 +214,10 @@ impl LedgerAppender { /// Writes and publishes a volatile-state reset marker. fn write_reset(&mut self, reset: Signed) -> Result<()> { + let slot = reset.slot; self.writer.write_blockstore(&BlockstoreEntry::Reset(reset))?; self.sync()?; - info!(slot = **reset, "appended volatile state reset"); + info!(slot, "appended volatile state reset"); Ok(()) } diff --git a/nucleus/Cargo.toml b/nucleus/Cargo.toml index 7455cdab..2e7384f8 100644 --- a/nucleus/Cargo.toml +++ b/nucleus/Cargo.toml @@ -22,11 +22,9 @@ config = [ ] default = [] ledger = [ - "dep:bytemuck", "dep:derive_more", "dep:solana-keypair", "dep:solana-pubkey", - "solana-hash/bytemuck", "solana-hash/copy", "solana-hash/wincode", "solana-signature/verify", @@ -64,7 +62,6 @@ testkit = [ tls = ["dep:solana-instruction-error", "dep:solana-pubkey", "dep:wincode"] [dependencies] -bytemuck = { workspace = true, optional = true, features = ["derive"] } derive_more = { workspace = true, optional = true, features = ["deref", "from"] } futures = { workspace = true, features = ["alloc"], optional = true } oneshot = { workspace = true, features = ["async"], optional = true } diff --git a/nucleus/src/ledger.rs b/nucleus/src/ledger.rs index 9a68b65f..d11d8f1c 100644 --- a/nucleus/src/ledger.rs +++ b/nucleus/src/ledger.rs @@ -1,18 +1,14 @@ //! Ledger block-boundary schema shared by storage-adjacent crates. -use bytemuck::NoUninit; use derive_more::Deref; use solana_hash::Hash; use solana_keypair::{Keypair, Signer}; use solana_pubkey::Pubkey; use solana_signature::Signature; -use wincode::{SchemaRead, SchemaWrite}; +use wincode::{Result, SchemaRead, SchemaWrite, config::DefaultConfig}; use crate::Slot; -#[cfg(not(target_endian = "little"))] -compile_error!("ledger signatures require little-endian targets"); - /// File name of the archived accountsdb snapshot tarball inside a superblock directory. pub const ACCOUNTSDB_SNAPSHOT_FILE: &str = "accountsdb.tar.zst"; @@ -28,7 +24,7 @@ pub struct BlockstorePosition { } /// Block boundary entry stored after all transactions in the block. -#[derive(SchemaRead, SchemaWrite, Clone, Copy, Default, PartialEq, Eq, Debug, NoUninit)] +#[derive(SchemaRead, SchemaWrite, Clone, Copy, Default, PartialEq, Eq, Debug)] #[repr(C)] pub struct Block { /// Slot that produced the block. @@ -49,7 +45,7 @@ impl Block { } /// Superblock boundary entry stored at the end of the blockstore stream. -#[derive(SchemaRead, SchemaWrite, Clone, Copy, Debug, PartialEq, Eq, NoUninit)] +#[derive(SchemaRead, SchemaWrite, Clone, Copy, Debug, PartialEq, Eq)] #[repr(C)] pub struct SuperblockSeal { /// Id of the superblock this seal closes. @@ -61,12 +57,18 @@ pub struct SuperblockSeal { } /// Volatile-state reset stored at its ordered position in the ledger. -#[derive(SchemaRead, SchemaWrite, Clone, Copy, Debug, PartialEq, Eq, Deref, NoUninit)] -#[repr(transparent)] -pub struct Reset(pub Slot); +#[derive(SchemaRead, SchemaWrite, Clone, Debug, PartialEq, Eq)] +pub struct Reset { + /// Slot at which volatile state is discarded. + pub slot: Slot, + /// Rent rate effective before subsequent transactions. + pub rent: u64, + /// Complete active feature set. + pub features: Vec, +} /// Fresh persisted-state checksum at an ordered ledger boundary. -#[derive(SchemaRead, SchemaWrite, Clone, Copy, Debug, PartialEq, Eq, Deref, NoUninit)] +#[derive(SchemaRead, SchemaWrite, Clone, Copy, Debug, PartialEq, Eq, Deref)] #[repr(transparent)] pub struct Checkpoint(pub u64); @@ -82,10 +84,8 @@ pub struct Signed { const DOMAIN_SIZE: usize = 8; -/// Padding-free ledger payload with a stable layout and distinct signing domain. -/// Its little-endian memory image must match its wincode payload and fit within -/// `size_of::()` bytes. Violating this contract is a programming error. -pub trait Record: NoUninit { +/// Canonically serialized ledger payload with a distinct signing domain. +pub trait Record: SchemaWrite { /// Domain prefix separating this payload from other signed records. const DOMAIN: &'static [u8; DOMAIN_SIZE]; } @@ -104,26 +104,26 @@ impl Record for Checkpoint { } impl Signed { - /// Signs the canonical payload without allocating an intermediate message. + /// Signs the domain-prefixed canonical payload. Encoding failure is a programming error. + #[expect( + clippy::expect_used, + reason = "Engine-owned record encoding is an invariant; verification remains fallible" + )] pub fn new(payload: T, signer: &Keypair) -> Self { - let signature = Self::message(&payload, |bytes| signer.sign_message(bytes)); + let bytes = Self::message(&payload).expect("Engine-owned ledger records must be encodable"); + let signature = signer.sign_message(&bytes); Self { payload, signature } } /// Authenticates the canonical payload against the configured producer. pub fn verify(&self, authority: &Pubkey) -> bool { - Self::message(&self.payload, |bytes| { - self.signature.verify(authority.as_ref(), bytes) - }) + Self::message(&self.payload) + .is_ok_and(|bytes| self.signature.verify(authority.as_ref(), &bytes)) } - fn message(payload: &T, consume: impl FnOnce(&[u8]) -> R) -> R { - let payload = bytemuck::bytes_of(payload); - let len = DOMAIN_SIZE + payload.len(); - // Only the domain prefix is copied separately; NoUninit excludes padding. - let mut bytes = [0; DOMAIN_SIZE + size_of::()]; - bytes[..DOMAIN_SIZE].copy_from_slice(T::DOMAIN); - bytes[DOMAIN_SIZE..len].copy_from_slice(payload); - consume(&bytes[..len]) + fn message(payload: &T) -> Result> { + let mut bytes = T::DOMAIN.to_vec(); + wincode::serialize_into(&mut bytes, payload)?; + Ok(bytes) } } diff --git a/processor/src/executor.rs b/processor/src/executor.rs index 760d547b..2566373f 100644 --- a/processor/src/executor.rs +++ b/processor/src/executor.rs @@ -113,7 +113,7 @@ impl TransactionExecutor { events: Sender, replay: bool, ) -> Result { - let svm = SvmContext::new(&state, cache)?; + let svm = SvmContext::new(&state, cache); let shutdown = shutdown.handle(Service::TransactionExecutor(id)); let (tx, rx) = mpsc::sync_channel(2); let executor = Self { @@ -178,7 +178,7 @@ impl TransactionExecutor { // executor completes processing, including commit and owned fanout. // Replay follows the same drain protocol; no borrowed result escapes. let loader = unsafe { AccountLoader::unguarded(&accounts) }; - let output = self.svm.execute::(loader, &txn, self.state.features()); + let output = self.svm.execute::(loader, &txn, &self.state); if !output.processing_result.was_processed_with_successful_result() { metrics::failed_transaction(FailureKind::Execution); } diff --git a/processor/src/sequencer/mod.rs b/processor/src/sequencer/mod.rs index d03036b6..2cad4eff 100644 --- a/processor/src/sequencer/mod.rs +++ b/processor/src/sequencer/mod.rs @@ -8,7 +8,9 @@ use keeper::Keeper; use nucleus::{ Slot, ledger::Signed, - runtime::{BarrierGuard, BlockInput, ExecutionRequest, SequencerHandle}, + runtime::{ + self, BarrierGuard, BlockInput, ExecutionRequest, SequencerHandle, SimulatorMessage, + }, shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, }; use solana_hash::Hash; @@ -54,10 +56,14 @@ pub(crate) struct ReadyTransaction { pub struct Sequencer { /// Durable engine state used for appends and account/block lookups. state: Arc, + /// Shared compiled programs, invalidated at feature boundaries. + cache: Arc, /// Block-local dependency graph preserving input order for account conflicts. ordering: OrderingTable, /// Inbound stream of transactions and block boundaries. rx: mpsc::Receiver, + /// Simulator participates in execution-boundary quiescence. + simulation: mpsc::Sender, /// Hash-chain state for the block currently being sequenced. hasher: BlockHasher, /// The executor pool and its availability bookkeeping. @@ -80,6 +86,7 @@ impl Sequencer { shutdown: &mut ShutdownManager, replay: bool, ) -> Result<(Self, SequencerHandle)> { + cache.invalidate(); let count = executors.clamp(1, MAX_EXECUTORS as usize); metrics::init(); let (event_tx, event_rx) = mpsc::channel(count); @@ -99,12 +106,14 @@ impl Sequencer { let hasher = BlockHasher::new(state.blockhash()); let (execution, rx) = mpsc::channel(1024); - let simulation = TransactionSimulator::spawn(state.clone(), cache, shutdown)?; + let simulation = TransactionSimulator::spawn(state.clone(), cache.clone(), shutdown)?; let shutdown = shutdown.handle(Service::Sequencer); let sequencer = Self { slot: state.blocks().current_slot(), state, + cache, + simulation: simulation.clone(), ordering: Default::default(), executors: Executors::new(executors, event_rx), rx, @@ -166,7 +175,9 @@ impl Sequencer { SequencerMessage::Barrier(guard) => self.barrier(guard).await, SequencerMessage::Shutdown(guard) => { self.rx.close(); - self.barrier(guard).await?; + // The simulator may already have stopped during coordinated shutdown. + self.drain().await?; + self.pause(guard).await; // Do not resume ingress after terminal storage synchronization. self.shutdown.signalled().await; return Ok(()); @@ -231,13 +242,22 @@ impl Sequencer { Ok(()) } - /// Raises a quiescence barrier: drains all in-flight work so the sequencer - /// and its executors are idle, acknowledges the controller, then waits to be - /// released before resuming. Used to take a consistent state snapshot at - /// superblock boundaries. + /// Drains execution and simulation before acknowledging the boundary, then + /// holds both services until the controller releases it. async fn barrier(&mut self, guard: BarrierGuard) -> Result<()> { self.drain().await?; + let (simulation, barrier) = runtime::barrier(); + self.simulation + .send(SimulatorMessage::Barrier(barrier)) + .await + .map_err(|_| Service::TransactionSimulator)?; + simulation.acknowledged.await.map_err(|_| Service::TransactionSimulator)?; + let features = self.state.features(); self.pause(guard).await; + if !Arc::ptr_eq(&features, &self.state.features()) { + self.cache.invalidate(); + } + drop(simulation.released); Ok(()) } diff --git a/processor/src/sequencer/tests.rs b/processor/src/sequencer/tests.rs index 2b25916a..baac9e43 100644 --- a/processor/src/sequencer/tests.rs +++ b/processor/src/sequencer/tests.rs @@ -30,6 +30,8 @@ fn sequencer(tk: &mut TestKeeper) -> Sequencer { slot: state.blocks().current_slot(), hasher: BlockHasher::new(state.blockhash()), state, + cache: Default::default(), + simulation: tokio_mpsc::channel(1).0, ordering: Default::default(), rx, executors: Executors::new(Vec::new(), event_rx), diff --git a/processor/src/simulator.rs b/processor/src/simulator.rs index 12edc801..2f46718f 100644 --- a/processor/src/simulator.rs +++ b/processor/src/simulator.rs @@ -37,7 +37,7 @@ impl TransactionSimulator { cache: Arc, shutdown: &mut ShutdownManager, ) -> Result> { - let svm = SvmContext::new(&state, cache)?; + let svm = SvmContext::new(&state, cache); let shutdown = shutdown.handle(Service::TransactionSimulator); let (tx, rx) = mpsc::channel(16); let executor = Self { rx, svm, state, shutdown }; @@ -100,7 +100,7 @@ impl TransactionSimulator { return; }; let accounts = self.state.accounts(); - let output = self.svm.execute::(accounts.loader(), &txn, self.state.features()); + let output = self.svm.execute::(accounts.loader(), &txn, &self.state); let execution = ExecutionRecord { result: output.processing_result, balances: output.balance_collector, diff --git a/processor/src/svm.rs b/processor/src/svm.rs index 3cf87c16..4133212b 100644 --- a/processor/src/svm.rs +++ b/processor/src/svm.rs @@ -28,16 +28,14 @@ use solana_svm::{ use solana_sysvar::clock::Clock; use solana_transaction_error::TransactionResult; -use crate::{ - Result, - callback::{InvokeCallback, LoadCallback}, -}; +use crate::callback::{InvokeCallback, LoadCallback}; /// SVM batch processor together with its per-block environment and recording /// config, shared verbatim by the executor and simulator workers. pub(crate) struct SvmContext { /// SVM batch processor that loads and executes transactions. processor: TransactionBatchProcessor, + features: Arc, /// Per-block processing environment (blockhash, features, rent, ...). env: TransactionProcessingEnvironment, /// Recording and logging configuration for execution. @@ -46,22 +44,15 @@ pub(crate) struct SvmContext { impl SvmContext { /// Builds the SVM runtime environment and batch processor from current state. - pub(crate) fn new(state: &Keeper, cache: Arc) -> Result { - let budget = Default::default(); - let runtime_env = agave_syscalls::create_program_runtime_environment( - &state.features().runtime_features(), - &budget, - false, // Accept legacy ELFs already deployed on the base chain. - false, - ) - .map_err(|e| e.to_string())?; - let envs = ProgramRuntimeEnvironments::new(runtime_env); + pub(crate) fn new(state: &Keeper, cache: Arc) -> Self { + let features = state.features(); + let envs = Self::environments(&features); let block = state.blocks().latest(); let env = TransactionProcessingEnvironment { blockhash: block.hash, - feature_set: state.features().runtime_features(), + feature_set: features.runtime_features(), program_runtime_environments_for_execution: envs, - rent: state.rent().clone(), + rent: state.rent(), blockhash_lamports_per_signature: 0, epoch_total_stake: 0, }; @@ -73,16 +64,45 @@ impl SvmContext { log_messages_bytes_limit: None, recording_config: ExecutionRecordingConfig::new_single_setting(true), }; - Ok(Self { processor, env, config }) + Self { processor, env, config, features } + } + + #[expect( + clippy::expect_used, + reason = "Static syscall registry collisions are a programming error" + )] + fn environments(features: &FeatureSet) -> ProgramRuntimeEnvironments { + let budget = Default::default(); + let runtime_env = agave_syscalls::create_program_runtime_environment( + &features.runtime_features(), + &budget, + false, // Accept legacy ELFs already deployed on the base chain. + false, + ) + .expect("static syscall registry must not contain symbol hash collisions"); + ProgramRuntimeEnvironments::new(runtime_env) } /// Loads and executes a single transaction through the SVM. pub(crate) fn execute( - &self, + &mut self, loader: AccountLoader<'_>, txn: &ResolvedTransaction, - features: &FeatureSet, + state: &Keeper, ) -> LoadAndExecuteSanitizedTransactionOutput { + let load = LoadCallback:: { loader }; + let rent = state.rent(); + if self.env.rent != rent { + self.processor.sysvar_cache_mut().set_rent(&rent); + self.env.rent = rent; + } + let features = state.features(); + if !Arc::ptr_eq(&self.features, &features) { + self.env.program_runtime_environments_for_execution = Self::environments(&features); + self.env.feature_set = features.runtime_features(); + self.features = features; + } + let features = &self.features; let details = match Self::parse_details(txn, features) { Ok(d) => d, Err(e) => { @@ -92,7 +112,6 @@ impl SvmContext { }; } }; - let load = LoadCallback:: { loader }; let invoke = InvokeCallback { featureset: features }; self.processor.load_and_execute_sanitized_transaction( load, diff --git a/processor/src/tests.rs b/processor/src/tests.rs index 6ffcac65..6cbc7e14 100644 --- a/processor/src/tests.rs +++ b/processor/src/tests.rs @@ -9,7 +9,7 @@ use keeper::{ testkit::{TestKeeper, V42_ID, load_v42_data, load_v42_lamports, signed_view, store_v42}, }; use nucleus::{ - ledger::Block, + ledger::{Block, Reset, Signed}, runtime::{ExecutionRequest, SequencerHandle, Simulation, barrier}, }; use solana_account::{AccountMode, ReadableAccount}; @@ -45,6 +45,7 @@ struct Harness { keeper: TestKeeper, /// Channels used by tests to drive execution and simulation paths. handle: SequencerHandle, + cache: Arc, } impl Harness { @@ -64,15 +65,16 @@ impl Harness { /// consume from it, forcing contention resolution to happen from a backlog. async fn unspawned(replay: bool) -> (Self, Sequencer) { let mut keeper = TestKeeper::new().await; + let cache = Arc::new(ProgramCache::default()); let (sequencer, handle) = Sequencer::new( 2, keeper.clone(), - Arc::new(ProgramCache::default()), + cache.clone(), &mut keeper.shutdown, replay, ) .unwrap(); - (Self { keeper, handle }, sequencer) + (Self { keeper, handle, cache }, sequencer) } /// A signed transaction view paid for by a fresh payer, not the shared one. @@ -140,7 +142,8 @@ impl Harness { /// /// [`TestKeeper::close`] flushes, so nothing needs flushing here. async fn close(self) { - let Self { keeper, handle } = self; + let Self { keeper, handle, cache } = self; + drop(cache); drop(handle); keeper.close().await; } @@ -329,11 +332,43 @@ async fn seeded_program_and_recursive_cpi_return_data_work() { harness.barrier().await; assert_eq!(load_v42_data(&harness, output), Some(110)); + let compiled = harness.cache.get(&V42_ID).expect("program compiled"); + let (controller, guard) = barrier(); + harness.handle.execution.send(SequencerMessage::Barrier(guard)).await.unwrap(); + controller.acknowledged.await.unwrap(); + let selected = harness + .features() + .active() + .keys() + .copied() + .filter(|id| *id != agave_feature_set::enable_poseidon_syscall::ID) + .collect(); + harness + .append_reset(Signed::new( + Reset { + slot: 1, + rent: harness.rent().lamports_per_byte, + features: selected, + }, + harness.signer(), + )) + .unwrap(); + drop(controller.released); + let (_, tx) = signed_view(&harness, None, E::lit(111).compose(output, &[])); + harness.execute(tx).await; + harness.barrier().await; + assert_eq!(load_v42_data(&harness, output), Some(111)); + let recompiled = harness.cache.get(&V42_ID).expect("program recompiled"); + assert!( + !Arc::ptr_eq(&compiled, &recompiled), + "old compilation must not survive a feature change" + ); + harness.close().await; } /// Proves stored and runtime-cached sysvars agree in execution and simulation at -/// startup, around epoch boundaries, and after skipped epochs, including zero fallback. +/// startup, across rent resets and epoch boundaries, including skipped epochs and zero fallback. #[tokio::test(flavor = "current_thread")] async fn block_transition_updates_execution_and_simulation_sysvars() { use keeper::testkit::{Dirs, keeper_builder}; @@ -354,6 +389,9 @@ async fn block_transition_updates_execution_and_simulation_sysvars() { .into_iter() .flat_map(u64::to_le_bytes) .collect(); + observed + .extend_from_slice(&ctx.get_sysvar_cache().get_rent()?.lamports_per_byte.to_le_bytes()); + observed.push(u8::from(ctx.get_feature_set().enable_poseidon_syscall)); // This fork exposes EpochSchedule through the generic serialized cache. observed.extend_from_slice( ctx.get_sysvar_cache() @@ -378,10 +416,16 @@ async fn block_transition_updates_execution_and_simulation_sysvars() { program, ProgramCacheEntry::new_builtin((ObserveEpoch::vm, ObserveEpoch::codegen)).into(), ); - let (sequencer, handle) = - Sequencer::new(2, keeper.clone(), cache, &mut keeper.shutdown, false).unwrap(); + let (sequencer, handle) = Sequencer::new( + 2, + keeper.clone(), + cache.clone(), + &mut keeper.shutdown, + false, + ) + .unwrap(); sequencer.spawn().unwrap(); - let harness = Harness { keeper, handle }; + let harness = Harness { keeper, handle, cache }; let schedule = harness.epoch_schedule(); let n = schedule.slots_per_epoch; let mut slots = vec![1, n.saturating_sub(1).max(1), n, n + 1, 4 * n + 1]; @@ -392,6 +436,27 @@ async fn block_transition_updates_execution_and_simulation_sysvars() { harness.set_block(block).await; harness.barrier().await; } + let (controller, guard) = barrier(); + harness.handle.execution.send(SequencerMessage::Barrier(guard)).await.unwrap(); + controller.acknowledged.await.unwrap(); + let mut rent = harness.rent(); + rent.lamports_per_byte = 5080 + slot % 2; + let mut selected: Vec<_> = harness.features().active().keys().copied().collect(); + selected.retain(|id| *id != agave_feature_set::enable_poseidon_syscall::ID); + if slot % 2 == 0 { + selected.push(agave_feature_set::enable_poseidon_syscall::ID); + } + harness + .append_reset(Signed::new( + Reset { + slot, + rent: rent.lamports_per_byte, + features: selected, + }, + harness.signer(), + )) + .unwrap(); + drop(controller.released); let expected = harness.clock(harness.blocks().latest()); assert_eq!(expected.slot, slot); assert_eq!(expected.epoch, slot / n); @@ -420,6 +485,8 @@ async fn block_transition_updates_execution_and_simulation_sysvars() { .into_iter() .flat_map(u64::to_le_bytes) .collect(); + data.extend_from_slice(&rent.lamports_per_byte.to_le_bytes()); + data.push(u8::from(slot % 2 == 0)); data.extend_from_slice(schedule_account.data()); let ix = Instruction::new_with_bytes(program, &data, vec![]); let (_, sim_tx) = signed_view(&harness, None, ix.clone()); diff --git a/replicator/src/tests.rs b/replicator/src/tests.rs index d0464e0c..375abf14 100644 --- a/replicator/src/tests.rs +++ b/replicator/src/tests.rs @@ -9,6 +9,7 @@ use std::{ use engine::{Engine, testkit::TestEngine}; use ledger::schema::{Block, Checkpoint, OwnedBlockstoreEntry, Reset, Signed, SuperblockSeal}; use solana_keypair::Keypair; +use solana_pubkey::Pubkey; use crate::{ ReplicationError, Result, @@ -67,12 +68,21 @@ async fn authenticates_boundary_records() { }, signer, ); - let mut reset = Signed::new(Reset(1), signer); + let mut reset = Signed::new( + Reset { + slot: 1, + rent: engine.rent().lamports_per_byte, + features: engine.features().active().keys().copied().collect(), + }, + signer, + ); let mut checkpoint = Signed::new(Checkpoint(2), signer); // Change payloads after signing, independently for each scenario. block.payload.slot += tamper; seal.payload.checksum += tamper; - reset.payload.0 += tamper; + if tamper != 0 { + reset.payload.features.push(Pubkey::new_unique()); + } checkpoint.payload.0 += tamper; for (kind, record) in [ ("block", OwnedBlockstoreEntry::Block(block)), diff --git a/replicator/tests/integration.rs b/replicator/tests/integration.rs index 37f772ee..e1f6f129 100644 --- a/replicator/tests/integration.rs +++ b/replicator/tests/integration.rs @@ -17,7 +17,12 @@ use keeper::testkit::{ v42_builder, v42_padded_value, }; use ledger::request::{BlockDetails, BlockParams, BlockResponse}; -use nucleus::{KB, Slot, config::Authority, ledger::BlockstorePosition, shutdown::ShutdownManager}; +use nucleus::{ + KB, Slot, + config::Authority, + ledger::{BlockstorePosition, Reset, Signed}, + shutdown::ShutdownManager, +}; use replicator::{ReplicationClient, ReplicationDispatcher}; use solana_account::{AccountBuilder, AccountMode, ReadableAccount}; use solana_keypair::{Keypair, Signer}; @@ -506,6 +511,31 @@ async fn streams_and_resumes_without_duplicate_application() { let mut second_dispatcher = dispatcher(upstream, &leader, &[follower_identity]).await; await_replication(&mut positions, &follower, expected, state, 5).await; + // The live follower must switch rent before executing the minimally funded + // materialization, even though its SVM workers were started at the old rate. + let extra_feature = Pubkey::new_unique(); + let mut rent = leader.rent(); + rent.lamports_per_byte = 5080; + { + let _guard = leader.barrier().await.unwrap(); + leader + .append_reset(Signed::new( + Reset { + slot: leader.blocks().current_slot(), + rent: rent.lamports_per_byte, + features: leader + .features() + .active() + .keys() + .copied() + .chain([extra_feature]) + .collect(), + }, + leader.signer(), + )) + .unwrap(); + } + // Account creation debits the sponsor on both nodes before reset replenishes it. let authority = leader.authority(); let authority_before = leader.get_account(authority).expect("leader sponsor exists").lamports(); @@ -514,7 +544,10 @@ async fn streams_and_resumes_without_duplicate_application() { .account(sponsored) .await .unwrap() - .materialize(v42_builder(0, AccountMode::Delegated), None) + .materialize( + v42_builder(0, AccountMode::Delegated).lamports(rent.minimum_balance(8)), + None, + ) .await .expect("sponsored account creation commits"); let authority_after = leader.get_account(authority).expect("leader sponsor remains").lamports(); @@ -525,6 +558,19 @@ async fn streams_and_resumes_without_duplicate_application() { leader.advance(1).await; let expected = leader.sync().await; await_replication(&mut positions, &follower, expected, state, 5).await; + assert_eq!(follower.rent(), rent); + assert!( + follower.features().is_active(&extra_feature), + "reset communicates active feature membership" + ); + assert_eq!( + follower + .get_account(sponsored) + .expect("rent-sensitive account replicated") + .lamports(), + rent.minimum_balance(8), + "the follower must execute the materialization, not merely adopt the reset rent" + ); assert_eq!( follower.get_account(authority).expect("follower sponsor remains").lamports(), authority_after, @@ -532,7 +578,11 @@ async fn streams_and_resumes_without_duplicate_application() { ); // Reset discards volatile accounts, retains delegated state, and replenishes the sponsor. - leader.reset(99).expect("leader reset records"); + let rent = leader.rent(); + { + let _guard = leader.barrier().await.unwrap(); + leader.reset(rent).expect("leader reset records"); + } assert_eq!( leader .get_account(authority) @@ -544,6 +594,10 @@ async fn streams_and_resumes_without_duplicate_application() { let expected = leader.sync().await; await_replication(&mut positions, &follower, expected, state, 5).await; assert!(follower.get_account(volatile).is_none()); + assert!( + !follower.features().is_active(&extra_feature), + "startup reset restores Engine defaults" + ); assert_eq!( follower .get_account(authority) diff --git a/solana/account/README.md b/solana/account/README.md index 6dcc7eb3..cbe892bd 100644 --- a/solana/account/README.md +++ b/solana/account/README.md @@ -19,10 +19,13 @@ resolution but reject user writes. `Closed` marks state for storage removal. Transitions to undelegating or closed revoke mutation immediately, including in CPI. The transaction-final writeback guard can accept a dirty transition into those modes without permitting further program writes. Other modes represent cached -external absence (`Missing`), a read-only mirror (`Mirrored`), and internal sysvar, -feature, or precompile state (`Runtime`). +external absence (`Missing`), read-only base mirrors (`Mirrored`), persisted sysvar +and feature state (`Runtime`), and locally supplied programs (`BuiltIn`). -`Runtime` has no base sync, is read-only, and survives mirror reset. +`Runtime` has no base sync, is read-only, survives local recovery and mirror +reset, and is included in persisted-state checksums. +`BuiltIn` has the same access and sync policy but stays in volatile storage; +the host reloads program images at startup. Base-chain programs remain `Mirrored`. Privileged replacement validates mode and source slot together. Delegated, Magic, and Undelegating state can't be refreshed in the same mode, even at a newer slot; @@ -31,9 +34,10 @@ transaction mutations aren't replacements and don't follow that restriction. | From | Allowed at the same or a newer slot | Requires a strictly newer slot | | :-- | :-- | :-- | -| Missing | Mirrored, Runtime, Delegated, Magic, Closed | Missing | +| Missing | Mirrored, Runtime, BuiltIn, Delegated, Magic, Closed | Missing | | Mirrored | Delegated, Magic, Closed | Mirrored, Missing | | Runtime | — | Runtime | +| BuiltIn | — | BuiltIn | | Delegated | Undelegating | — | | Magic | Delegated, Closed | — | | Undelegating | Mirrored, Missing, Closed | Delegated | diff --git a/solana/account/src/mode.rs b/solana/account/src/mode.rs index f268476e..edbc382d 100644 --- a/solana/account/src/mode.rs +++ b/solana/account/src/mode.rs @@ -20,12 +20,12 @@ const FLOW_MASK: u16 = 0x0003 << FLOW_OFFSET; pub(crate) enum AccountKind { /// Cached absence of an account on the base chain. Missing, - /// Ordinary account state, including mirrors, delegation, and independent ER state. + /// Account state whose access and synchronization policy determine its behavior. Regular, - /// Internal runtime state, such as sysvars, features, and programs. - Runtime, + /// Locally supplied program image rebuilt at startup instead of persisted. + BuiltIn, /// Deletion tombstone retained until storage removes the account image. - Closed, + Closed = 3, } /// Encodes kind, access, and data flow into a 16-bit mode discriminant. @@ -101,8 +101,11 @@ pub enum AccountMode { /// A counterpart at the same address may independently exist on base. Magic = encode(Regular, Writable, DataFlow::NoSync), - /// Runtime-maintained state, such as locally generated sysvars. - Runtime = encode(Runtime, ReadOnly, DataFlow::NoSync), + /// Read-only, locally maintained state without base-chain synchronization. + Runtime = encode(Regular, ReadOnly, DataFlow::NoSync), + + /// Read-only local program image, kept in memory without base-chain synchronization. + BuiltIn = encode(BuiltIn, ReadOnly, DataFlow::NoSync), /// Deletion tombstone awaiting physical removal. Closed = encode(Closed, ReadOnly, DataFlow::NoSync), @@ -113,15 +116,14 @@ pub enum AccountMode { #[error("invalid account mode bytes: {0:?}")] pub struct InvalidAccountMode([u8; 2]); -#[allow(unused_qualifications)] impl AccountMode { /// Returns the remaining runtime/lifecycle classification. pub(crate) const fn kind(self) -> AccountKind { match ((self as u16) & KIND_MASK) >> KIND_OFFSET { - 0 => AccountKind::Missing, - 1 => AccountKind::Regular, - 2 => AccountKind::Runtime, - 3 => AccountKind::Closed, + 0 => Missing, + 1 => Regular, + 2 => BuiltIn, + 3 => Closed, _ => unreachable!(), } } @@ -153,16 +155,20 @@ impl AccountMode { /// Returns who maintains the account image, or `None` for missing and closed accounts. pub(crate) const fn authority(self) -> Option { match (self.kind(), self.data_flow()) { - (AccountKind::Missing | AccountKind::Closed, _) => None, + (Missing | Closed, _) => None, (_, DataFlow::BaseToEr) => Some(StateAuthority::Base), (_, DataFlow::NoSync | DataFlow::ErToBase) => Some(StateAuthority::Er), } } + /// Returns whether the ER is authoritative for this account's state. + pub const fn is_er_owned(self) -> bool { + matches!(self.authority(), Some(StateAuthority::Er)) + } + /// Returns whether local state requires persistence. pub const fn requires_persistence(self) -> bool { - matches!(self.kind(), AccountKind::Regular) - && matches!(self.authority(), Some(StateAuthority::Er)) + matches!(self.kind(), Regular) && self.is_er_owned() } /// Validates lifecycle eligibility before slot freshness. @@ -174,50 +180,18 @@ impl AccountMode { ) -> Result<(), AccountPatchError> { use AccountMode::*; - enum TransitionRule { - Forbidden, - /// The lifecycle change may use the current image's slot. - SameOrNewer, - /// A fresh observation is required, not another update at the same slot. - Newer, - } - use TransitionRule::*; - - let rule = match self { - Missing => match to { - Missing => Newer, - Mirrored | Runtime | Delegated | Magic | Closed => SameOrNewer, - Undelegating => Forbidden, - }, - Mirrored => match to { - Missing | Mirrored => Newer, - Delegated | Magic | Closed => SameOrNewer, - Runtime | Undelegating => Forbidden, - }, - Runtime => match to { - Runtime => Newer, - Missing | Mirrored | Delegated | Magic | Undelegating | Closed => Forbidden, - }, - Delegated => match to { - Undelegating => SameOrNewer, - Missing | Mirrored | Runtime | Delegated | Magic | Closed => Forbidden, - }, - Magic => match to { - Delegated | Closed => SameOrNewer, - Missing | Mirrored | Runtime | Magic | Undelegating => Forbidden, - }, - Undelegating => match to { - Missing | Mirrored | Closed => SameOrNewer, - Delegated => Newer, - Runtime | Magic | Undelegating => Forbidden, - }, - Closed => Forbidden, - }; - - let valid = match rule { - Forbidden => return Err(AccountPatchError::InvalidModeTransition { from: self, to }), - SameOrNewer => to_slot >= from_slot, - Newer => to_slot > from_slot, + let valid = match (self, to) { + (Missing, Missing) + | (Mirrored, Missing | Mirrored) + | (Runtime, Runtime) + | (BuiltIn, BuiltIn) + | (Undelegating, Delegated) => to_slot > from_slot, + (Missing, Mirrored | Runtime | BuiltIn | Delegated | Magic | Closed) + | (Mirrored, Delegated | Magic | Closed) + | (Delegated, Undelegating) + | (Magic, Delegated | Closed) + | (Undelegating, Missing | Mirrored | Closed) => to_slot >= from_slot, + _ => return Err(AccountPatchError::InvalidModeTransition { from: self, to }), }; if !valid { return Err(AccountPatchError::InvalidSlotTransition { from: from_slot, to: to_slot }); @@ -239,7 +213,8 @@ impl TryFrom<[u8; 2]> for AccountMode { match bytes { [0, 4] => Ok(Self::Missing), [1, 4] => Ok(Self::Mirrored), - [2, 0] => Ok(Self::Runtime), + [1, 0] => Ok(Self::Runtime), + [2, 0] => Ok(Self::BuiltIn), [1, 9] => Ok(Self::Delegated), [1, 1] => Ok(Self::Magic), [1, 8] => Ok(Self::Undelegating), diff --git a/solana/account/src/tests/account.rs b/solana/account/src/tests/account.rs index 2d84bb7c..9436e2aa 100644 --- a/solana/account/src/tests/account.rs +++ b/solana/account/src/tests/account.rs @@ -198,13 +198,14 @@ fn test_lifecycle_matrix() { // Columns are destinations in `modes` order. // 0 forbids the pair, 1 permits equal/newer slots, 2 requires a newer slot. let cases = [ - (Mode::Missing, [2, 1, 1, 1, 1, 0, 1]), - (Mode::Mirrored, [2, 2, 0, 1, 1, 0, 1]), - (Mode::Runtime, [0, 0, 2, 0, 0, 0, 0]), - (Mode::Delegated, [0, 0, 0, 0, 0, 1, 0]), - (Mode::Magic, [0, 0, 0, 1, 0, 0, 1]), - (Mode::Undelegating, [1, 1, 0, 2, 0, 0, 1]), - (Mode::Closed, [0, 0, 0, 0, 0, 0, 0]), + (Mode::Missing, [2, 1, 1, 1, 0, 1, 1, 1]), + (Mode::Mirrored, [2, 2, 1, 1, 0, 1, 0, 0]), + (Mode::Delegated, [0, 0, 0, 0, 1, 0, 0, 0]), + (Mode::Magic, [0, 0, 1, 0, 0, 1, 0, 0]), + (Mode::Undelegating, [1, 1, 2, 0, 0, 1, 0, 0]), + (Mode::Closed, [0, 0, 0, 0, 0, 0, 0, 0]), + (Mode::Runtime, [0, 0, 0, 0, 0, 0, 2, 0]), + (Mode::BuiltIn, [0, 0, 0, 0, 0, 0, 0, 2]), ]; let modes = cases.map(|(mode, _)| mode); let slot_pairs = @@ -302,7 +303,8 @@ fn test_mode_layout_and_decoding() { let cases = [ (Mode::Missing, [0, 4], None), (Mode::Mirrored, [1, 4], Some(Base)), - (Mode::Runtime, [2, 0], Some(Er)), + (Mode::Runtime, [1, 0], Some(Er)), + (Mode::BuiltIn, [2, 0], Some(Er)), (Mode::Delegated, [1, 9], Some(Er)), (Mode::Magic, [1, 1], Some(Er)), (Mode::Undelegating, [1, 8], Some(Er)), @@ -332,12 +334,16 @@ fn test_mode_layout_and_decoding() { ); assert_eq!( mode.requires_persistence(), - mode == Mode::Delegated || mode == Mode::Magic || mode == Mode::Undelegating + mode == Mode::Delegated + || mode == Mode::Magic + || mode == Mode::Undelegating + || mode == Mode::Runtime ); + assert_eq!(mode.is_er_owned(), authority == Some(Er)); } assert_eq!(Mode::default(), Mode::Missing); - for bytes in [[4, 0], [1, 0x10], [1, 2], [1, 3], [1, 12], [2, 1], [1, 0]] { + for bytes in [[2, 4], [4, 0], [1, 0x10], [1, 2], [1, 3], [1, 12], [2, 1]] { assert!(Mode::try_from(bytes).is_err()); assert!(bincode::deserialize::(&bytes).is_err()); #[cfg(feature = "wincode")] diff --git a/solana/program-runtime/src/loaded_programs.rs b/solana/program-runtime/src/loaded_programs.rs index fd8b62ff..90d8e3bf 100644 --- a/solana/program-runtime/src/loaded_programs.rs +++ b/solana/program-runtime/src/loaded_programs.rs @@ -274,6 +274,11 @@ impl ProgramCacheForTxBatch { } impl ProgramCache { + /// Drops programs compiled or verified under a previous feature environment. + pub fn invalidate(&self) { + self.index.retain_sync(|_, entry| entry.program.get_environment().is_none()); + } + pub fn assign_program(&self, key: Pubkey, entry: Arc) { self.index.upsert_sync(key, entry); } @@ -326,6 +331,7 @@ mod tests { }) } + /// Proves environment-bound entries are discarded rather than reused after a feature change. #[test] fn assign_program_makes_entry_retrievable() { let cache = ProgramCache::default(); @@ -336,5 +342,7 @@ mod tests { let fetched = cache.get(&key); assert!(fetched.is_some()); assert!(Arc::ptr_eq(&fetched.unwrap(), &entry)); + cache.invalidate(); + assert!(cache.get(&key).is_none()); } } diff --git a/solana/program-runtime/src/sysvar_cache.rs b/solana/program-runtime/src/sysvar_cache.rs index 087100b7..6cf9a3b9 100644 --- a/solana/program-runtime/src/sysvar_cache.rs +++ b/solana/program-runtime/src/sysvar_cache.rs @@ -84,6 +84,14 @@ impl SysvarCache { self.get_sysvar_obj(&Clock::id()) } + /// Stores a serialized rent sysvar. + pub fn set_rent(&mut self, rent: &Rent) { + let buffer = self.rent.get_or_insert_default(); + buffer.clear(); + // Rent serialization into a Vec cannot fail. + let _ = bincode::serialize_into(buffer, rent); + } + /// Returns the cached rent sysvar. pub fn get_rent(&self) -> Result, InstructionError> { self.get_sysvar_obj(&Rent::id()) diff --git a/solana/transaction-context/src/instruction_accounts.rs b/solana/transaction-context/src/instruction_accounts.rs index 527a1ac3..8b705f9d 100644 --- a/solana/transaction-context/src/instruction_accounts.rs +++ b/solana/transaction-context/src/instruction_accounts.rs @@ -414,6 +414,7 @@ mod tests { AccountMode::Missing, AccountMode::Mirrored, AccountMode::Runtime, + AccountMode::BuiltIn, AccountMode::Undelegating, AccountMode::Closed, ] { From 4901f783351222ee9353e222a3b9a5b6d649572d Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 6 Nov 2026 18:32:23 +0400 Subject: [PATCH 2/4] refactor: restore persisted runtime program accounts --- accountsdb/README.md | 5 ++--- accountsdb/src/lib.rs | 4 ++-- accountsdb/src/tests.rs | 14 +++++++------- accountsdb/src/volatile.rs | 13 ++----------- keeper/README.md | 4 +--- keeper/src/builder.rs | 12 ++++++------ keeper/src/cache.rs | 4 ++-- keeper/src/tests/caches.rs | 1 - keeper/src/tests/recovery.rs | 9 +++++---- solana/account/README.md | 10 ++++------ solana/account/src/mode.rs | 17 ++--------------- solana/account/src/tests/account.rs | 19 ++++++++----------- .../src/instruction_accounts.rs | 1 - 13 files changed, 41 insertions(+), 72 deletions(-) diff --git a/accountsdb/README.md b/accountsdb/README.md index 902752c6..c89aa4a4 100644 --- a/accountsdb/README.md +++ b/accountsdb/README.md @@ -2,14 +2,13 @@ Accountsdb stores current account state; the [ledger](../ledger/README.md) retains execution history. The store separates account state that requires persistence -from locally supplied programs and rebuildable base mirrors, which live in memory. +from rebuildable base mirrors, which live in memory. Routing follows [lifecycle mode](../solana/account/README.md#account-lifecycle), not memory representation. Delegated, Magic, Runtime, and undelegating accounts are persisted, including undelegating state that no longer permits user writes. -`BuiltIn` program images are volatile, read-only, and have no base-chain sync. Closed accounts are removed. Reset discards mirrored state while preserving -ER-owned state, including volatile local programs. +persisted accounts, including locally supplied `Runtime` program images. ## Storage access and writeback diff --git a/accountsdb/src/lib.rs b/accountsdb/src/lib.rs index b3db8546..ffd2b81b 100644 --- a/accountsdb/src/lib.rs +++ b/accountsdb/src/lib.rs @@ -38,7 +38,7 @@ const ACTIVE_DIR: &str = "CURRENT"; pub struct AccountsDB { /// On-disk store for account modes that require persistence. persisted: PersistedStore, - /// In-memory store for programs and base-chain mirrors. + /// Rebuildable in-memory store for base-chain mirrors. volatile: VolatileStore, /// Database root directory. root: PathBuf, @@ -237,7 +237,7 @@ impl AccountsDB { } /// Drops rebuildable chain mirrors and their owner indexes. - /// ER-owned state, including local programs and runtime accounts, is retained. + /// Persisted state, including local programs and runtime accounts, is retained. pub fn reset(&self) { self.volatile.reset(); } diff --git a/accountsdb/src/tests.rs b/accountsdb/src/tests.rs index 276e0bff..75241e07 100644 --- a/accountsdb/src/tests.rs +++ b/accountsdb/src/tests.rs @@ -129,7 +129,7 @@ fn mutable_at_least(lamports: u64, units: u64, owner: &Pubkey) -> AccountSharedD } /// Proves lifecycle changes route accounts and owner indexes between backends, -/// while closing removes state and reset retains ER-owned state, including volatile programs. +/// while closing removes state and reset retains persisted runtime accounts and programs. #[test] fn test_routing_and_persistence_flips() { let (_dir, db) = db(); @@ -223,24 +223,24 @@ fn test_routing_and_persistence_flips() { before, "fixed-size runtime updates commit without reallocating" ); - let builtin = Pubkey::new_unique(); + let executable = Pubkey::new_unique(); store( &db, - builtin, + executable, AccountBuilder::default() .owner(p) - .mode(AccountMode::BuiltIn) + .mode(AccountMode::Runtime) .executable(true) .build(), ); - assert!(in_volatile(&db, &builtin) && !in_persisted(&db, &builtin)); + assert!(in_persisted(&db, &executable) && !in_volatile(&db, &executable)); db.reset(); assert!(!in_volatile(&db, &b)); assert!(in_persisted(&db, &c)); assert!(in_persisted(&db, &runtime)); assert!(program(&db, &p).contains(&runtime)); - assert!(in_volatile(&db, &builtin)); - assert!(program(&db, &p).contains(&builtin)); + assert!(in_persisted(&db, &executable)); + assert!(program(&db, &p).contains(&executable)); } // Both migration directions remove the source image and owner mapping, retain diff --git a/accountsdb/src/volatile.rs b/accountsdb/src/volatile.rs index 23087917..14c5bc97 100644 --- a/accountsdb/src/volatile.rs +++ b/accountsdb/src/volatile.rs @@ -108,19 +108,10 @@ impl VolatileStore { self.programs.read_sync(owner, |_, s| s.clone()).unwrap_or_default() } - /// Drops base-chain state while retaining ER-owned programs and their indexes. + /// Drops rebuildable base-chain state and its ownership indexes. pub(crate) fn reset(&self) { + self.accounts.clear_sync(); self.programs.clear_sync(); - self.accounts.retain_sync(|key, account| { - if !account.mode().is_er_owned() { - return false; - } - BTreeSet::insert( - &mut self.programs.entry_sync(account.owner()).or_default(), - *key, - ); - true - }); } /// Removes the cached account and drops its owner mapping. diff --git a/keeper/README.md b/keeper/README.md index 91e7f80f..5b688bdc 100644 --- a/keeper/README.md +++ b/keeper/README.md @@ -9,10 +9,8 @@ to interpret their state. ## Startup and recovery Startup validates account storage against ledger progress, restores a retained -snapshot when necessary, and seeds features and sysvars only on genesis. +snapshot when necessary, and seeds programs, features, and sysvars only on genesis. Later startups preserve those account images and restore runtime configuration from them. -Locally supplied native and SBF programs are volatile `BuiltIn` accounts, reloaded -from builder inputs on every startup and retained across mirror resets. [Engine](../engine/README.md#startup-and-recovery) replays the remaining history. Recovery can fail rather than open inconsistent state and depends on valid retained snapshots and history. diff --git a/keeper/src/builder.rs b/keeper/src/builder.rs index eec81372..6bf77827 100644 --- a/keeper/src/builder.rs +++ b/keeper/src/builder.rs @@ -69,9 +69,9 @@ pub struct KeeperBuilder { pub ledger: LedgerParams, /// Block production timing and superblock sealing parameters. pub blockstore: BlockstoreParams, - /// Native builtin program ids to install as volatile executable accounts. + /// Native builtin program ids to seed as executable runtime accounts at genesis. pub builtins: HashMap, - /// SBF programs to reload as volatile loader-v4 accounts: `(program id, ELF bytes)`. + /// SBF programs to seed at genesis as loader-v4 accounts: `(program id, ELF bytes)`. pub programs: HashMap>, /// Accounts to seed at startup if missing from the database; existing accounts are kept. pub possibly_missing_accounts: HashMap, @@ -114,16 +114,16 @@ impl KeeperBuilder { }) } - /// Reloads local programs, initializes genesis runtime state, and seeds missing accounts. + /// Initializes runtime state once and seeds caller-supplied missing accounts. fn prepopulate(&mut self, accountsdb: &AccountsDB) -> Result<()> { let mut accounts = Vec::new(); if accountsdb.is_new() { for id in DEFAULT_FEATURES { accounts.push((id, feature_account())); } + self.seed_programs(&mut accounts); self.seed_sysvars(&mut accounts); } - self.seed_programs(&mut accounts); let loader = accountsdb.loader(); let authority = self.authority.pubkey(); if !loader.contains(&authority)? { @@ -179,7 +179,7 @@ impl KeeperBuilder { Ok(features) } - /// Reloads native-loader builtins and precompiles, and loader-v4 SBF programs. + /// Seeds native-loader builtins and precompiles, and loader-v4 SBF programs. fn seed_programs(&mut self, accounts: &mut Vec) { let precompiles = [ solana_sdk_ids::ed25519_program::ID, @@ -198,7 +198,7 @@ impl KeeperBuilder { let lamports = Rent::default().minimum_balance(elf.len()); let account = AccountBuilder::default() .lamports(lamports) - .mode(AccountMode::BuiltIn) + .mode(AccountMode::Runtime) .owner(owner) .executable(true) .data(elf); diff --git a/keeper/src/cache.rs b/keeper/src/cache.rs index 11938e41..4d062578 100644 --- a/keeper/src/cache.rs +++ b/keeper/src/cache.rs @@ -124,7 +124,7 @@ impl AccountLease { /// Returns whether the observed account may be evicted from the cache. pub fn evictable(&self) -> bool { - self.observed.is_some_and(|(mode, _)| !mode.is_er_owned()) + self.observed.is_some_and(|(mode, _)| !mode.requires_persistence()) } /// Returns the local mode when the incoming image is older or duplicates @@ -165,7 +165,7 @@ impl Drop for AccountLease { impl AccountCache { /// Tracks a materialized account and returns one displaced from recency. fn track(&self, pubkey: Pubkey, mode: AccountMode) -> Option { - if !mode.is_er_owned() { + if !mode.requires_persistence() { return self.admit_recency(pubkey); } self.remove_recency(&pubkey); diff --git a/keeper/src/tests/caches.rs b/keeper/src/tests/caches.rs index 8bf73e20..e922f3f9 100644 --- a/keeper/src/tests/caches.rs +++ b/keeper/src/tests/caches.rs @@ -116,7 +116,6 @@ async fn account_lease_coordinates_recency_and_waiters() { (AccountMode::Magic, false), (AccountMode::Undelegating, false), (AccountMode::Runtime, false), - (AccountMode::BuiltIn, false), ]; for (mode, tracked) in modes { let cache = Arc::new(AccountCache::new(256)); diff --git a/keeper/src/tests/recovery.rs b/keeper/src/tests/recovery.rs index 3f656712..3a767540 100644 --- a/keeper/src/tests/recovery.rs +++ b/keeper/src/tests/recovery.rs @@ -70,7 +70,7 @@ async fn seeds_features_programs_and_sysvars() { // Builtins are seeded through the same path with an executable native-loader // account, so they share this shape. let acc = loader.read(&program, Clone::clone).unwrap().expect("program seeded"); - assert!(acc.is(AccountMode::BuiltIn)); + assert!(acc.is(AccountMode::Runtime)); assert!(acc.executable()); assert_eq!(acc.owner(), &loader_v4::ID); assert_eq!(acc.data(), elf.as_slice()); @@ -118,7 +118,7 @@ async fn seeds_features_programs_and_sysvars() { } } -/// Startup reloads local programs but preserves persisted state and seeds missing accounts. +/// Startup preserves persisted programs and accounts while seeding missing accounts. #[tokio::test] async fn preserves_seeded_accounts_on_reopen() { let existing = Pubkey::new_unique(); @@ -131,6 +131,7 @@ async fn preserves_seeded_accounts_on_reopen() { ); let keeper = TestKeeper::from_builder(dirs, builder.clone()).await; + let program = builder.programs[&V42_ID].clone(); let schedule = keeper.epoch_schedule().clone(); let marker_owner = Pubkey::new_unique(); let marker_data = vec![7, 7, 7]; @@ -155,8 +156,8 @@ async fn preserves_seeded_accounts_on_reopen() { let loader = accounts.loader(); assert_eq!( loader.read(&V42_ID, |account| account.data().to_vec()).unwrap(), - Some(vec![9]), - "startup reloads the supplied program bytes" + Some(program), + "startup must not replace persisted program bytes" ); assert_eq!( loader.read(&existing, |account| account.lamports()).unwrap(), diff --git a/solana/account/README.md b/solana/account/README.md index cbe892bd..a2b9f7ec 100644 --- a/solana/account/README.md +++ b/solana/account/README.md @@ -19,13 +19,12 @@ resolution but reject user writes. `Closed` marks state for storage removal. Transitions to undelegating or closed revoke mutation immediately, including in CPI. The transaction-final writeback guard can accept a dirty transition into those modes without permitting further program writes. Other modes represent cached -external absence (`Missing`), read-only base mirrors (`Mirrored`), persisted sysvar -and feature state (`Runtime`), and locally supplied programs (`BuiltIn`). +external absence (`Missing`), read-only base mirrors (`Mirrored`), and persisted +sysvar, feature, and locally supplied program state (`Runtime`). `Runtime` has no base sync, is read-only, survives local recovery and mirror reset, and is included in persisted-state checksums. -`BuiltIn` has the same access and sync policy but stays in volatile storage; -the host reloads program images at startup. Base-chain programs remain `Mirrored`. +Base-chain programs remain `Mirrored`. Privileged replacement validates mode and source slot together. Delegated, Magic, and Undelegating state can't be refreshed in the same mode, even at a newer slot; @@ -34,10 +33,9 @@ transaction mutations aren't replacements and don't follow that restriction. | From | Allowed at the same or a newer slot | Requires a strictly newer slot | | :-- | :-- | :-- | -| Missing | Mirrored, Runtime, BuiltIn, Delegated, Magic, Closed | Missing | +| Missing | Mirrored, Runtime, Delegated, Magic, Closed | Missing | | Mirrored | Delegated, Magic, Closed | Mirrored, Missing | | Runtime | — | Runtime | -| BuiltIn | — | BuiltIn | | Delegated | Undelegating | — | | Magic | Delegated, Closed | — | | Undelegating | Mirrored, Missing, Closed | Delegated | diff --git a/solana/account/src/mode.rs b/solana/account/src/mode.rs index edbc382d..2f579cd5 100644 --- a/solana/account/src/mode.rs +++ b/solana/account/src/mode.rs @@ -22,8 +22,6 @@ pub(crate) enum AccountKind { Missing, /// Account state whose access and synchronization policy determine its behavior. Regular, - /// Locally supplied program image rebuilt at startup instead of persisted. - BuiltIn, /// Deletion tombstone retained until storage removes the account image. Closed = 3, } @@ -104,9 +102,6 @@ pub enum AccountMode { /// Read-only, locally maintained state without base-chain synchronization. Runtime = encode(Regular, ReadOnly, DataFlow::NoSync), - /// Read-only local program image, kept in memory without base-chain synchronization. - BuiltIn = encode(BuiltIn, ReadOnly, DataFlow::NoSync), - /// Deletion tombstone awaiting physical removal. Closed = encode(Closed, ReadOnly, DataFlow::NoSync), } @@ -122,7 +117,6 @@ impl AccountMode { match ((self as u16) & KIND_MASK) >> KIND_OFFSET { 0 => Missing, 1 => Regular, - 2 => BuiltIn, 3 => Closed, _ => unreachable!(), } @@ -161,14 +155,9 @@ impl AccountMode { } } - /// Returns whether the ER is authoritative for this account's state. - pub const fn is_er_owned(self) -> bool { - matches!(self.authority(), Some(StateAuthority::Er)) - } - /// Returns whether local state requires persistence. pub const fn requires_persistence(self) -> bool { - matches!(self.kind(), Regular) && self.is_er_owned() + matches!(self.authority(), Some(StateAuthority::Er)) } /// Validates lifecycle eligibility before slot freshness. @@ -184,9 +173,8 @@ impl AccountMode { (Missing, Missing) | (Mirrored, Missing | Mirrored) | (Runtime, Runtime) - | (BuiltIn, BuiltIn) | (Undelegating, Delegated) => to_slot > from_slot, - (Missing, Mirrored | Runtime | BuiltIn | Delegated | Magic | Closed) + (Missing, Mirrored | Runtime | Delegated | Magic | Closed) | (Mirrored, Delegated | Magic | Closed) | (Delegated, Undelegating) | (Magic, Delegated | Closed) @@ -214,7 +202,6 @@ impl TryFrom<[u8; 2]> for AccountMode { [0, 4] => Ok(Self::Missing), [1, 4] => Ok(Self::Mirrored), [1, 0] => Ok(Self::Runtime), - [2, 0] => Ok(Self::BuiltIn), [1, 9] => Ok(Self::Delegated), [1, 1] => Ok(Self::Magic), [1, 8] => Ok(Self::Undelegating), diff --git a/solana/account/src/tests/account.rs b/solana/account/src/tests/account.rs index 9436e2aa..4f868fba 100644 --- a/solana/account/src/tests/account.rs +++ b/solana/account/src/tests/account.rs @@ -198,14 +198,13 @@ fn test_lifecycle_matrix() { // Columns are destinations in `modes` order. // 0 forbids the pair, 1 permits equal/newer slots, 2 requires a newer slot. let cases = [ - (Mode::Missing, [2, 1, 1, 1, 0, 1, 1, 1]), - (Mode::Mirrored, [2, 2, 1, 1, 0, 1, 0, 0]), - (Mode::Delegated, [0, 0, 0, 0, 1, 0, 0, 0]), - (Mode::Magic, [0, 0, 1, 0, 0, 1, 0, 0]), - (Mode::Undelegating, [1, 1, 2, 0, 0, 1, 0, 0]), - (Mode::Closed, [0, 0, 0, 0, 0, 0, 0, 0]), - (Mode::Runtime, [0, 0, 0, 0, 0, 0, 2, 0]), - (Mode::BuiltIn, [0, 0, 0, 0, 0, 0, 0, 2]), + (Mode::Missing, [2, 1, 1, 1, 0, 1, 1]), + (Mode::Mirrored, [2, 2, 1, 1, 0, 1, 0]), + (Mode::Delegated, [0, 0, 0, 0, 1, 0, 0]), + (Mode::Magic, [0, 0, 1, 0, 0, 1, 0]), + (Mode::Undelegating, [1, 1, 2, 0, 0, 1, 0]), + (Mode::Closed, [0, 0, 0, 0, 0, 0, 0]), + (Mode::Runtime, [0, 0, 0, 0, 0, 0, 2]), ]; let modes = cases.map(|(mode, _)| mode); let slot_pairs = @@ -304,7 +303,6 @@ fn test_mode_layout_and_decoding() { (Mode::Missing, [0, 4], None), (Mode::Mirrored, [1, 4], Some(Base)), (Mode::Runtime, [1, 0], Some(Er)), - (Mode::BuiltIn, [2, 0], Some(Er)), (Mode::Delegated, [1, 9], Some(Er)), (Mode::Magic, [1, 1], Some(Er)), (Mode::Undelegating, [1, 8], Some(Er)), @@ -339,11 +337,10 @@ fn test_mode_layout_and_decoding() { || mode == Mode::Undelegating || mode == Mode::Runtime ); - assert_eq!(mode.is_er_owned(), authority == Some(Er)); } assert_eq!(Mode::default(), Mode::Missing); - for bytes in [[2, 4], [4, 0], [1, 0x10], [1, 2], [1, 3], [1, 12], [2, 1]] { + for bytes in [[2, 0], [4, 0], [1, 0x10], [1, 2], [1, 3], [1, 12], [2, 1]] { assert!(Mode::try_from(bytes).is_err()); assert!(bincode::deserialize::(&bytes).is_err()); #[cfg(feature = "wincode")] diff --git a/solana/transaction-context/src/instruction_accounts.rs b/solana/transaction-context/src/instruction_accounts.rs index 8b705f9d..527a1ac3 100644 --- a/solana/transaction-context/src/instruction_accounts.rs +++ b/solana/transaction-context/src/instruction_accounts.rs @@ -414,7 +414,6 @@ mod tests { AccountMode::Missing, AccountMode::Mirrored, AccountMode::Runtime, - AccountMode::BuiltIn, AccountMode::Undelegating, AccountMode::Closed, ] { From e23a67bd3b39ee4d70b7d9ed8b960af771b4bf85 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 6 Nov 2026 18:58:01 +0400 Subject: [PATCH 3/4] fix: seed missing programs on every startup --- keeper/README.md | 5 +++-- keeper/src/builder.rs | 30 ++++++++++++++++-------------- keeper/src/tests/recovery.rs | 27 ++++++++++++++++++++++++--- 3 files changed, 43 insertions(+), 19 deletions(-) diff --git a/keeper/README.md b/keeper/README.md index 5b688bdc..45cca824 100644 --- a/keeper/README.md +++ b/keeper/README.md @@ -9,8 +9,9 @@ to interpret their state. ## Startup and recovery Startup validates account storage against ledger progress, restores a retained -snapshot when necessary, and seeds programs, features, and sysvars only on genesis. -Later startups preserve those account images and restore runtime configuration from them. +snapshot when necessary, and seeds features and sysvars only on genesis. +Programs and caller-supplied account seeds are inserted whenever absent; existing +images are preserved. Runtime configuration is restored from persisted account state. [Engine](../engine/README.md#startup-and-recovery) replays the remaining history. Recovery can fail rather than open inconsistent state and depends on valid retained snapshots and history. diff --git a/keeper/src/builder.rs b/keeper/src/builder.rs index 6bf77827..fcd791d1 100644 --- a/keeper/src/builder.rs +++ b/keeper/src/builder.rs @@ -69,9 +69,9 @@ pub struct KeeperBuilder { pub ledger: LedgerParams, /// Block production timing and superblock sealing parameters. pub blockstore: BlockstoreParams, - /// Native builtin program ids to seed as executable runtime accounts at genesis. + /// Native builtin program ids to seed as executable runtime accounts if absent. pub builtins: HashMap, - /// SBF programs to seed at genesis as loader-v4 accounts: `(program id, ELF bytes)`. + /// SBF programs to seed if absent as loader-v4 accounts: `(program id, ELF bytes)`. pub programs: HashMap>, /// Accounts to seed at startup if missing from the database; existing accounts are kept. pub possibly_missing_accounts: HashMap, @@ -114,16 +114,16 @@ impl KeeperBuilder { }) } - /// Initializes runtime state once and seeds caller-supplied missing accounts. + /// Initializes runtime configuration once and seeds missing programs and accounts. fn prepopulate(&mut self, accountsdb: &AccountsDB) -> Result<()> { let mut accounts = Vec::new(); if accountsdb.is_new() { for id in DEFAULT_FEATURES { accounts.push((id, feature_account())); } - self.seed_programs(&mut accounts); self.seed_sysvars(&mut accounts); } + self.seed_programs(); let loader = accountsdb.loader(); let authority = self.authority.pubkey(); if !loader.contains(&authority)? { @@ -179,8 +179,8 @@ impl KeeperBuilder { Ok(features) } - /// Seeds native-loader builtins and precompiles, and loader-v4 SBF programs. - fn seed_programs(&mut self, accounts: &mut Vec) { + /// Adds program images to the common insert-if-missing startup seeds. + fn seed_programs(&mut self) { let precompiles = [ solana_sdk_ids::ed25519_program::ID, solana_sdk_ids::secp256k1_program::ID, @@ -195,14 +195,16 @@ impl KeeperBuilder { let programs = self.programs.drain().map(|(id, elf)| (id, solana_sdk_ids::loader_v4::ID, elf)); for (program, owner, elf) in builtins.chain(programs) { - let lamports = Rent::default().minimum_balance(elf.len()); - let account = AccountBuilder::default() - .lamports(lamports) - .mode(AccountMode::Runtime) - .owner(owner) - .executable(true) - .data(elf); - accounts.push((program, account.build())); + self.possibly_missing_accounts.entry(program).or_insert_with(|| { + let lamports = Rent::default().minimum_balance(elf.len()); + AccountBuilder::default() + .lamports(lamports) + .mode(AccountMode::Runtime) + .owner(owner) + .executable(true) + .data(elf) + .build() + }); } } diff --git a/keeper/src/tests/recovery.rs b/keeper/src/tests/recovery.rs index 3a767540..5e7c5250 100644 --- a/keeper/src/tests/recovery.rs +++ b/keeper/src/tests/recovery.rs @@ -8,7 +8,7 @@ use nucleus::{ ledger::Reset, testkit::{V42_ID, block, signed_view}, }; -use solana_account::{AccountMode, ReadableAccount, testkit::delegated_account}; +use solana_account::{AccountBuilder, AccountMode, ReadableAccount, testkit::delegated_account}; use solana_instruction::Instruction; use solana_keypair::Keypair; use solana_pubkey::Pubkey; @@ -118,11 +118,14 @@ async fn seeds_features_programs_and_sysvars() { } } -/// Startup preserves persisted programs and accounts while seeding missing accounts. +/// Startup preserves existing programs and accounts while persisting missing SBF, +/// native program, and caller-supplied account seeds on an initialized database. #[tokio::test] async fn preserves_seeded_accounts_on_reopen() { let existing = Pubkey::new_unique(); let added = Pubkey::new_unique(); + let added_program = Pubkey::new_unique(); + let native = solana_sdk_ids::secp256k1_program::ID; let dirs = Dirs::default(); let mut builder = keeper_builder(&dirs); builder.possibly_missing_accounts.insert( @@ -136,10 +139,16 @@ async fn preserves_seeded_accounts_on_reopen() { let marker_owner = Pubkey::new_unique(); let marker_data = vec![7, 7, 7]; let marker = delegated_account(2, marker_data.clone(), marker_owner); - keeper.accounts().store(&[(existing, marker.build())]).unwrap(); + // Reopening must restore a missing native program, not just retain an existing image. + let closed = AccountBuilder::default().mode(AccountMode::Closed).build(); + keeper + .accounts() + .store(&[(existing, marker.build()), (native, closed)]) + .unwrap(); let dirs = keeper.close().await; builder.programs.insert(V42_ID, vec![9]); + builder.programs.insert(added_program, vec![1, 2, 3]); builder.blockstore.superblock = 3; builder.possibly_missing_accounts.insert( added, @@ -159,6 +168,18 @@ async fn preserves_seeded_accounts_on_reopen() { Some(program), "startup must not replace persisted program bytes" ); + for (id, owner, data) in [ + (added_program, loader_v4::ID, vec![1, 2, 3]), + (native, solana_sdk_ids::native_loader::ID, vec![]), + ] { + let account = loader + .read(&id, Clone::clone) + .unwrap() + .expect("missing programs are seeded on an initialized database"); + assert!(account.is(AccountMode::Runtime) && account.executable()); + assert_eq!(account.owner(), &owner); + assert_eq!(account.data(), data); + } assert_eq!( loader.read(&existing, |account| account.lamports()).unwrap(), Some(2) From d92ee49a32a2294f4ec5d938278a45b2267bf2f1 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 6 Nov 2026 20:28:34 +0400 Subject: [PATCH 4/4] fix(replication): start fresh followers from a snapshot --- replicator/README.md | 4 +++ replicator/src/server.rs | 7 +++- replicator/tests/integration.rs | 62 +++++++++++++++++++++++---------- 3 files changed, 53 insertions(+), 20 deletions(-) diff --git a/replicator/README.md b/replicator/README.md index afb434fc..e7228732 100644 --- a/replicator/README.md +++ b/replicator/README.md @@ -32,6 +32,10 @@ boundary. Peers must agree on record formats and execution semantics. ## Follower recovery +Fresh followers must connect after the source has completed its first snapshot. +They install the newest available snapshot before streaming: replay from the first +segment alone cannot reconstruct migrated or startup-seeded initial accounts. + Reconnect resumes after already applied work while preserving stream order. A synchronized ledger position is sufficient while the source retains the required history. Otherwise the follower stages an available snapshot and requests a host diff --git a/replicator/src/server.rs b/replicator/src/server.rs index 8a6bfa15..58517224 100644 --- a/replicator/src/server.rs +++ b/replicator/src/server.rs @@ -236,7 +236,7 @@ impl ReplicationServer { protocol::write(&mut self.connection, &handshake) } - /// Selects retained streaming when possible, otherwise the newest ready snapshot. + /// Starts fresh followers from a snapshot; otherwise resumes retained history. fn handshake(&mut self) -> Result<(ReplicationAction, Arc<()>)> { let _timer = metrics::time(Operation::ServerHandshake); let handshake: Handshake = protocol::read(&mut self.connection)?; @@ -250,6 +250,11 @@ impl ReplicationServer { if requested > self.position.current { return Err(ReplicationError::PositionNotFound(requested)); } + // The first segment has no snapshot baseline: replay alone cannot restore + // accounts imported before recording began (for example, migrated state). + if requested.superblock == 1 { + return Ok((self.snapshot()?, lease)); + } let action = match self.ledger().cursor(requested.superblock) { Some(end) if requested.offset <= end => ReplicationAction::Stream { from: requested, diff --git a/replicator/tests/integration.rs b/replicator/tests/integration.rs index e1f6f129..05af246d 100644 --- a/replicator/tests/integration.rs +++ b/replicator/tests/integration.rs @@ -21,7 +21,7 @@ use nucleus::{ KB, Slot, config::Authority, ledger::{BlockstorePosition, Reset, Signed}, - shutdown::ShutdownManager, + shutdown::{ShutdownManager, ShutdownReason}, }; use replicator::{ReplicationClient, ReplicationDispatcher}; use solana_account::{AccountBuilder, AccountMode, ReadableAccount}; @@ -74,6 +74,14 @@ async fn engines( }; let leader = engine(leader_authority, leader_accounts, Pacing::External).await; let follower = engine(follower_authority, follower_accounts, Pacing::External).await; + { + let _guard = leader.barrier().await.unwrap(); + keeper::testkit::seal_and_archive(&leader).await; + } + let addr = loopback_addr(); + let mut server = dispatcher(addr, &leader, &[follower.signer().pubkey()]).await; + let follower = restart_from_snapshot(addr, follower).await; + server.terminate().await; (leader, follower) } @@ -104,7 +112,11 @@ fn stream(addr: SocketAddr, follower: &mut TestEngine) -> broadcast::Receiver TestEngine { replicate(addr, &mut follower); - follower.shutdown().wait().await; + let reason = time::timeout(TIMEOUT, follower.shutdown().wait()).await.unwrap(); + assert!( + matches!(reason, ShutdownReason::RestartRequired), + "{reason:?}" + ); let (dirs, authority) = follower.close().await; TestEngine::with(dirs, authority).await } @@ -157,7 +169,7 @@ async fn checkpoint_mismatch_stops_replication() { leader.advance(1).await; leader.sync().await; let reason = time::timeout(TIMEOUT, follower.shutdown().wait()).await.unwrap(); - let nucleus::shutdown::ShutdownReason::Error(error) = reason else { + let ShutdownReason::Error(error) = reason else { panic!("checkpoint mismatch must stop replication: {reason:?}"); }; assert!(matches!( @@ -428,10 +440,20 @@ async fn internally_paced_replication_persists_across_restart() { let leader = engine(leader_authority, &seed, Pacing::Internal).await; let mut follower = engine(follower_authority, &seed, Pacing::External).await; + let mut snapshots = leader.accounts().subscribe_snapshots(); + if !leader + .ledger() + .iter() + .any(|segment| segment.directory.join(nucleus::ledger::ACCOUNTSDB_SNAPSHOT_FILE).is_file()) + { + time::timeout(TIMEOUT, snapshots.recv()).await.unwrap().unwrap(); + } + leader.superblocks().sync(false).unwrap(); let addr = loopback_addr(); - let initial = follower.superblocks().position(); let follower_identity = follower.signer().pubkey(); let mut dispatcher = dispatcher(addr, &leader, &[follower_identity]).await; + follower = restart_from_snapshot(addr, follower).await; + let initial = follower.superblocks().position(); let mut positions = stream(addr, &mut follower); let observed = time::timeout(TIMEOUT, async { loop { @@ -465,7 +487,7 @@ async fn streams_and_resumes_without_duplicate_application() { let seed = [(state, 0, AccountMode::Delegated), (volatile, 7, AccountMode::Mirrored)]; let (mut leader, mut follower) = engines(&seed, &seed).await; - // Catch-up crosses two sealed superblocks and a live tail from byte zero. + // Catch-up streams across two sealed superblocks and the live tail. for _ in 0..2 { increment(&leader, state).await; leader.seal_and_archive().await; @@ -713,14 +735,16 @@ async fn graceful_shutdown_drains_to_next_block_and_reopens() { leader.close().await; } -/// Installs the newest retained snapshot on restart, then streams every durable -/// ledger entry committed after that snapshot. +/// A cold follower prefers the newest snapshot even while its cursor is retained, +/// then streams every durable ledger entry committed after that snapshot. #[tokio::test(flavor = "multi_thread")] async fn restores_the_newest_snapshot_then_streams_its_tail() { let state = Pubkey::new_unique(); let seed = [(state, 0, AccountMode::Delegated)]; // The empty follower can acquire `state` only through snapshot restoration. - let (mut leader, follower) = engines(&seed, &[]).await; + let (mut leader, installed) = engines(&seed, &[]).await; + let (_, authority) = installed.close().await; + let follower = engine(authority, &[], Pacing::External).await; let set = |value| E::lit(value).compose(state, &[]); // Distinct values distinguish the newest snapshot from its streamed tail. @@ -731,16 +755,9 @@ async fn restores_the_newest_snapshot_then_streams_its_tail() { leader.execute(&[set(30)]).await.unwrap(); leader.advance(1).await; let expected = leader.sync().await; - // Make the newest snapshot the only possible handshake response. - truncate(leader.ledger()); // superblock 0 - truncate(leader.ledger()); // superblock 1 - assert!( - leader.ledger().cursor(0).is_none(), - "follower cursor was retained away" - ); assert!( - leader.ledger().cursor(1).is_none(), - "older snapshot history was retained away" + leader.ledger().cursor(follower.superblocks().position().superblock).is_some(), + "the cold follower could otherwise stream retained history" ); let addr = loopback_addr(); @@ -786,9 +803,15 @@ async fn cascades_replication_through_a_follower() { let middle_addr = loopback_addr(); let middle_identity = middle.signer().pubkey(); let tail_identity = tail.signer().pubkey(); + { + let _guard = leader.barrier().await.unwrap(); + keeper::testkit::seal_and_archive(&leader).await; + } let mut leader_dispatcher = dispatcher(leader_addr, &leader, &[middle_identity]).await; + middle = restart_from_snapshot(leader_addr, middle).await; let mut middle_positions = stream(leader_addr, &mut middle); let mut middle_dispatcher = dispatcher(middle_addr, &middle, &[tail_identity]).await; + let tail = restart_from_snapshot(middle_addr, tail).await; // Subscribe before the boundary because this seal is driven by replication, // outside the testkit's race-free seal-and-archive helper. @@ -814,9 +837,10 @@ async fn cascades_replication_through_a_follower() { await_replication(&mut middle_positions, &middle, expected, state, 2).await; // The successor archive must become the only answer to the tail's cursor. - truncate(middle.ledger()); // superblock 0 + truncate(middle.ledger()); // initial segment + truncate(middle.ledger()); // tail's installed snapshot segment assert!( - middle.ledger().cursor(0).is_none(), + middle.ledger().cursor(tail.superblocks().position().superblock).is_none(), "tail cursor was retained away" );