1pub use solana_perf::report_target_features;
4use {
5 crate::{
6 admin_rpc_post_init::{AdminRpcRequestMetadataPostInit, KeyUpdaterType, KeyUpdaters},
7 banking_stage::{
8 BankingStage, transaction_scheduler::scheduler_controller::SchedulerConfig,
9 },
10 banking_trace::{self, BankingTracer, TraceError},
11 block_creation_loop::{BlockCreationLoop, BlockCreationLoopConfig, ReplayHighestFrozen},
12 cluster_info_vote_listener::VoteTracker,
13 completed_data_sets_service::CompletedDataSetsService,
14 consensus::{
15 ExternalRootSource, Tower, reconcile_blockstore_roots_with_external_source,
16 tower_storage::{NullTowerStorage, TowerStorage},
17 },
18 forwarding_stage::ForwardingClientConfig,
19 repair::{
20 self, repair_handler::RepairHandlerType, serve_repair_service::ServeRepairService,
21 },
22 resource_limits::{ResourceLimitError, adjust_nofile_limit},
23 sample_performance_service::SamplePerformanceService,
24 snapshot_packager_service::SnapshotPackagerService,
25 stats_reporter_service::StatsReporterService,
26 system_monitor_service::{
27 SystemMonitorService, SystemMonitorStatsReportConfig, XdpNetworkConfigReport,
28 verify_net_stats_access,
29 },
30 tpu::{Tpu, TpuSockets},
31 tvu::{AlpenglowInitializationState, Tvu, TvuConfig, TvuSockets},
32 },
33 agave_snapshots::{
34 SnapshotInterval, snapshot_archive_info::SnapshotArchiveInfoGetter as _,
35 snapshot_config::SnapshotConfig, snapshot_hash::StartingSnapshotHashes,
36 },
37 agave_votor::{
38 vote_history::{VoteHistory, VoteHistoryError},
39 vote_history_storage::{NullVoteHistoryStorage, VoteHistoryStorage},
40 voting_service::VotingServiceOverride,
41 },
42 agave_xdp::transmitter::{Transmitter, TransmitterBuilder},
43 anyhow::{Result, anyhow},
44 crossbeam_channel::{Receiver, bounded, unbounded},
45 serde::{Deserialize, Serialize},
46 solana_account::ReadableAccount,
47 solana_accounts_db::{
48 accounts_db::{ACCOUNTS_DB_CONFIG_FOR_TESTING, AccountsDbConfig},
49 accounts_update_notifier_interface::AccountsUpdateNotifier,
50 utils::validate_account_paths_for_direct_io,
51 },
52 solana_client::connection_cache::{ConnectionCache, Protocol},
53 solana_clock::Slot,
54 solana_entry::poh::compute_hash_time,
55 solana_epoch_schedule::MAX_LEADER_SCHEDULE_EPOCH_OFFSET,
56 solana_genesis_config::GenesisConfig,
57 solana_genesis_utils::{
58 MAX_GENESIS_ARCHIVE_UNPACKED_SIZE, OpenGenesisConfigError, open_genesis_config,
59 },
60 solana_geyser_plugin_manager::{
61 GeyserPluginManagerRequest,
62 contact_info_notifier::{
63 self as geyser_contact_info_notifier, ContactInfoNotifier as GeyserContactInfoNotifier,
64 },
65 geyser_plugin_service::GeyserPluginService,
66 },
67 solana_gossip::{
68 cluster_info::{
69 ClusterInfo, DEFAULT_CONTACT_DEBUG_INTERVAL_MILLIS,
70 DEFAULT_CONTACT_SAVE_INTERVAL_MILLIS,
71 },
72 contact_info::ContactInfo,
73 crds_gossip_pull::CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS,
74 gossip_service::GossipService,
75 node::{Node, NodeMultihoming},
76 },
77 solana_hard_forks::HardForks,
78 solana_hash::Hash,
79 solana_keypair::Keypair,
80 solana_leader_schedule::{FixedSchedule, SlotLeader},
81 solana_ledger::{
82 bank_forks_utils,
83 blockstore::{
84 Blockstore, BlockstoreError, MAX_COMPLETED_SLOTS_IN_CHANNEL,
85 MAX_REPLAY_WAKE_UP_SIGNALS, MAX_UPDATE_PARENT_SIGNALS, PurgeType, UpdateParentReceiver,
86 },
87 blockstore_metric_report_service::BlockstoreMetricReportService,
88 blockstore_options::{BLOCKSTORE_DIRECTORY_ROCKS_LEVEL, BlockstoreOptions},
89 blockstore_processor,
90 entry_notifier_interface::EntryNotifierArc,
91 entry_notifier_service::{EntryNotifierSender, EntryNotifierService},
92 leader_schedule_cache::LeaderScheduleCache,
93 shred::filter::TurbineMode,
94 use_snapshot_archives_at_startup::UseSnapshotArchivesAtStartup,
95 },
96 solana_measure::measure::Measure,
97 solana_metrics::{datapoint_info, metrics::metrics_config_sanity_check},
98 solana_net_utils::{PinnedXdpSender, SocketAddrSpace},
99 solana_poh::{
100 poh_controller::PohController,
101 poh_recorder::PohRecorder,
102 poh_service::{self, PohService},
103 record_channels::record_channels,
104 transaction_recorder::TransactionRecorder,
105 },
106 solana_pubkey::Pubkey,
107 solana_rpc::{
108 max_slots::MaxSlots,
109 optimistically_confirmed_bank_tracker::{
110 BankNotificationSenderConfig, OptimisticallyConfirmedBank,
111 OptimisticallyConfirmedBankTracker,
112 },
113 rpc::JsonRpcConfig,
114 rpc_completed_slots_service::RpcCompletedSlotsService,
115 rpc_pubsub_service::{PubSubConfig, PubSubService},
116 rpc_service::{JsonRpcService, JsonRpcServiceConfig, RpcTpuClientArgs},
117 rpc_subscriptions::RpcSubscriptions,
118 transaction_notifier_interface::TransactionNotifierArc,
119 transaction_status_service::TransactionStatusService,
120 },
121 solana_runtime::{
122 accounts_background_service::{
123 AbsRequestHandlers, AccountsBackgroundService, DroppedSlotsReceiver,
124 PendingSnapshotPackages, PrunedBanksRequestHandler, SnapshotRequestHandler,
125 },
126 bank::{Bank, MAX_ALPENGLOW_VOTE_ACCOUNTS},
127 bank_forks::BankForks,
128 bank_forks_controller::BankForksControllerHandle,
129 commitment::BlockCommitmentCache,
130 dependency_tracker::DependencyTracker,
131 prioritization_fee_cache::PrioritizationFeeCache,
132 runtime_config::RuntimeConfig,
133 snapshot_bank_utils,
134 snapshot_controller::SnapshotController,
135 snapshot_utils,
136 transaction_execution::TransactionStatusSender,
137 },
138 solana_send_transaction_service::send_transaction_service::Config as SendTransactionServiceConfig,
139 solana_shred_version::compute_shred_version,
140 solana_signer::Signer,
141 solana_streamer::{
142 nonblocking::{simple_qos::SimpleQosConfig, swqos::SwQosConfig},
143 quic::{QuicStreamerConfig, SimpleQosQuicStreamerConfig, SwQosQuicStreamerConfig},
144 streamer::StakedNodes,
145 },
146 solana_time_utils::timestamp,
147 solana_tpu_client::tpu_client::{DEFAULT_TPU_CONNECTION_POOL_SIZE, DEFAULT_VOTE_USE_QUIC},
148 solana_turbine::{self, broadcast_stage::BroadcastStageType},
149 solana_unified_scheduler_pool::DefaultSchedulerPool,
150 solana_validator_exit::Exit,
151 solana_vote_program::vote_state::{VoteStateV4, handler::VoteStateHandler},
152 std::{
153 borrow::Cow,
154 cmp,
155 collections::{HashMap, HashSet},
156 net::{Ipv4Addr, SocketAddr, SocketAddrV4},
157 num::{NonZeroU64, NonZeroUsize},
158 path::{Path, PathBuf},
159 str::FromStr,
160 sync::{
161 Arc, Mutex, RwLock,
162 atomic::{AtomicBool, AtomicU64, Ordering},
163 },
164 thread::{self, Builder, JoinHandle},
165 time::{Duration, Instant},
166 },
167 strum::VariantNames,
168 strum_macros::{Display, EnumCount, EnumIter, EnumString, IntoStaticStr},
169 thiserror::Error,
170 tokio::{runtime::Runtime as TokioRuntime, sync::mpsc},
171 tokio_util::sync::CancellationToken,
172};
173
174const MAX_COMPLETED_DATA_SETS_IN_CHANNEL: usize = 100_000;
175const WAIT_FOR_SUPERMAJORITY_THRESHOLD_PERCENT: u64 = 80;
176
177#[derive(Clone, EnumCount, EnumIter, EnumString, VariantNames, Default, IntoStaticStr, Display)]
178#[strum(serialize_all = "kebab-case")]
179pub enum BlockVerificationMethod {
180 #[default]
181 UnifiedScheduler,
182}
183
184impl BlockVerificationMethod {
185 pub const fn cli_names() -> &'static [&'static str] {
186 Self::VARIANTS
187 }
188
189 pub fn cli_message() -> &'static str {
190 "Switch transaction scheduling method for verifying ledger entries"
191 }
192}
193
194#[derive(
195 Clone,
196 Debug,
197 EnumCount,
198 EnumIter,
199 EnumString,
200 VariantNames,
201 Default,
202 IntoStaticStr,
203 Display,
204 Serialize,
205 Deserialize,
206 PartialEq,
207 Eq,
208)]
209#[strum(serialize_all = "kebab-case")]
210#[serde(rename_all = "kebab-case")]
211pub enum BlockProductionMethod {
212 CentralScheduler,
213 #[default]
214 CentralSchedulerGreedy,
215}
216
217impl BlockProductionMethod {
218 pub const fn cli_names() -> &'static [&'static str] {
219 Self::VARIANTS
220 }
221
222 pub fn cli_message() -> &'static str {
223 "Switch transaction scheduling method for producing ledger entries"
224 }
225
226 pub fn warn_if_deprecated_value(&self) {
227 if matches!(self, Self::CentralScheduler) {
228 warn!(
229 "`central-scheduler` is deprecated and will be removed in a future release; use \
230 `central-scheduler-greedy` instead"
231 );
232 }
233 }
234}
235
236#[derive(
237 Clone,
238 Debug,
239 EnumString,
240 VariantNames,
241 Default,
242 IntoStaticStr,
243 Display,
244 Serialize,
245 Deserialize,
246 PartialEq,
247 Eq,
248)]
249#[strum(serialize_all = "kebab-case")]
250#[serde(rename_all = "kebab-case")]
251pub enum TransactionStructure {
252 Sdk,
253 #[default]
254 View,
255}
256
257impl TransactionStructure {
258 pub const fn cli_names() -> &'static [&'static str] {
259 Self::VARIANTS
260 }
261
262 pub fn cli_message() -> &'static str {
263 "DEPRECATED: has no impact on banking stage; will be removed in a future version"
264 }
265}
266
267#[derive(
268 Clone, Debug, VariantNames, IntoStaticStr, Display, Serialize, Deserialize, PartialEq, Eq,
269)]
270#[strum(serialize_all = "kebab-case")]
271#[serde(rename_all = "kebab-case")]
272pub enum SchedulerPacing {
273 Disabled,
274 FillTimeMillis(NonZeroU64),
275}
276
277impl FromStr for SchedulerPacing {
278 type Err = String;
279
280 fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
281 if s.eq_ignore_ascii_case("disabled") {
282 Ok(SchedulerPacing::Disabled)
283 } else {
284 match s.parse::<u64>() {
285 Ok(v) if v > 0 => Ok(SchedulerPacing::FillTimeMillis(
286 NonZeroU64::new(v).ok_or_else(|| "value must be non-zero".to_string())?,
287 )),
288 _ => Err("value must be a positive integer or 'disabled'".to_string()),
289 }
290 }
291 }
292}
293
294impl SchedulerPacing {
295 pub fn fill_time(&self) -> Option<Duration> {
296 match self {
297 SchedulerPacing::Disabled => None,
298 SchedulerPacing::FillTimeMillis(millis) => Some(Duration::from_millis(millis.get())),
299 }
300 }
301}
302
303#[derive(Clone, Debug)]
305pub struct GeneratorConfig {
306 pub accounts_path: String,
307 pub starting_keypairs: Arc<Vec<Keypair>>,
308}
309
310#[derive(Clone, Debug)]
311pub struct ValidatorLogConfig {
312 pub logfile: PathBuf,
314 pub logrotate_flag: Arc<AtomicBool>,
318}
319
320pub struct ValidatorConfig {
321 pub log_config: Option<ValidatorLogConfig>,
323 pub expected_genesis_hash: Option<Hash>,
324 pub expected_bank_hash: Option<Hash>,
325 pub expected_shred_version: Option<u16>,
326 pub voting_disabled: bool,
327 pub account_paths: Vec<PathBuf>,
328 pub account_snapshot_paths: Vec<PathBuf>,
329 pub rpc_config: JsonRpcConfig,
330 pub on_start_geyser_plugin_config_files: Option<Vec<PathBuf>>,
332 pub geyser_plugin_always_enabled: bool,
333 pub rpc_addrs: Option<(SocketAddr, SocketAddr)>, pub pubsub_config: PubSubConfig,
335 pub snapshot_config: SnapshotConfig,
336 pub max_ledger_shreds: Option<u64>,
337 pub blockstore_options: BlockstoreOptions,
338 pub broadcast_stage_type: BroadcastStageType,
339 pub turbine_mode: TurbineMode,
340 pub fixed_leader_schedule: Option<FixedSchedule>,
341 pub wait_for_supermajority: Option<Slot>,
342 pub new_hard_forks: Option<Vec<Slot>>,
343 pub known_validators: Option<HashSet<Pubkey>>, pub repair_validators: Option<HashSet<Pubkey>>, pub repair_whitelist: Arc<RwLock<HashSet<Pubkey>>>, pub gossip_validators: Option<HashSet<Pubkey>>, pub should_check_duplicate_instance: bool,
348 pub max_genesis_archive_unpacked_size: u64,
349 pub run_verification: bool,
352 pub require_tower: bool,
353 pub require_vote_history: bool,
354 pub tower_storage: Arc<dyn TowerStorage>,
355 pub vote_history_storage: Arc<dyn VoteHistoryStorage>,
356 pub debug_keys: Option<Arc<HashSet<Pubkey>>>,
357 pub filter_keys: Arc<HashSet<Pubkey>>,
358 pub contact_debug_interval: u64,
359 pub contact_save_interval: u64,
360 pub send_transaction_service_config: SendTransactionServiceConfig,
361 pub no_poh_speed_test: bool,
362 pub no_os_memory_stats_reporting: bool,
363 pub no_os_network_stats_reporting: bool,
364 pub xdp_network_config_report: Option<XdpNetworkConfigReport>,
365 pub no_os_cpu_stats_reporting: bool,
366 pub no_os_disk_stats_reporting: bool,
367 pub enforce_ulimit_nofile: bool,
368 pub poh_pinned_cpu_core: Option<usize>,
369 pub poh_hashes_per_batch: u64,
370 pub process_ledger_before_services: bool,
371 pub accounts_db_config: AccountsDbConfig,
372 pub warp_slot: Option<Slot>,
373 pub accounts_db_skip_shrink: bool,
374 pub accounts_db_force_initial_clean: bool,
375 pub staked_nodes_overrides: Arc<RwLock<HashMap<Pubkey, u64>>>,
376 pub validator_exit: Arc<RwLock<Exit>>,
377 pub validator_exit_backpressure: HashMap<String, Arc<AtomicBool>>,
378 pub no_wait_for_vote_to_start_leader: bool,
379 pub wait_to_vote_slot: Option<Slot>,
380 pub runtime_config: RuntimeConfig,
381 pub banking_trace_dir_byte_limit: banking_trace::DirByteLimit,
382 pub block_verification_method: BlockVerificationMethod,
383 pub block_production_method: BlockProductionMethod,
384 pub block_production_num_workers: NonZeroUsize,
385 pub block_production_scheduler_config: SchedulerConfig,
386 pub enable_block_production_forwarding: bool,
387 pub enable_scheduler_bindings: bool,
388 pub generator_config: Option<GeneratorConfig>,
389 pub use_snapshot_archives_at_startup: UseSnapshotArchivesAtStartup,
390 pub unified_scheduler_handler_threads: Option<usize>,
391 pub ip_echo_server_threads: NonZeroUsize,
392 pub rayon_global_threads: NonZeroUsize,
393 pub replay_forks_threads: NonZeroUsize,
394 pub replay_transactions_threads: NonZeroUsize,
395 pub tvu_shred_sigverify_threads: NonZeroUsize,
396 pub tvu_bls_sigverify_threads: NonZeroUsize,
397 pub delay_leader_block_for_pending_fork: bool,
398 pub voting_service_test_override: Option<VotingServiceOverride>,
399 pub repair_handler_type: RepairHandlerType,
400 pub snapshot_packager_niceness_adj: i8,
402}
403
404impl ValidatorConfig {
405 pub fn default_for_test() -> Self {
406 Self {
407 log_config: None,
408 expected_genesis_hash: None,
409 expected_bank_hash: None,
410 expected_shred_version: None,
411 voting_disabled: false,
412 max_ledger_shreds: None,
413 blockstore_options: BlockstoreOptions::default_for_tests(),
414 account_paths: Vec::new(),
415 account_snapshot_paths: Vec::new(),
416 rpc_config: JsonRpcConfig::default_for_test(),
417 on_start_geyser_plugin_config_files: None,
418 geyser_plugin_always_enabled: false,
419 rpc_addrs: None,
420 pubsub_config: PubSubConfig::default_for_tests(),
421 snapshot_config: SnapshotConfig::new_load_only(),
422 broadcast_stage_type: BroadcastStageType::Standard,
423 turbine_mode: TurbineMode::default(),
424 fixed_leader_schedule: None,
425 wait_for_supermajority: None,
426 new_hard_forks: None,
427 known_validators: None,
428 repair_validators: None,
429 should_check_duplicate_instance: true,
430 repair_whitelist: Arc::new(RwLock::new(HashSet::default())),
431 gossip_validators: None,
432 max_genesis_archive_unpacked_size: MAX_GENESIS_ARCHIVE_UNPACKED_SIZE,
433 run_verification: true,
434 require_tower: false,
435 require_vote_history: false,
436 tower_storage: Arc::new(NullTowerStorage::default()),
437 vote_history_storage: Arc::new(NullVoteHistoryStorage::default()),
438 debug_keys: None,
439 filter_keys: Arc::default(),
440 contact_debug_interval: DEFAULT_CONTACT_DEBUG_INTERVAL_MILLIS,
441 contact_save_interval: DEFAULT_CONTACT_SAVE_INTERVAL_MILLIS,
442 send_transaction_service_config: SendTransactionServiceConfig::default(),
443 no_poh_speed_test: true,
444 no_os_memory_stats_reporting: true,
445 no_os_network_stats_reporting: true,
446 xdp_network_config_report: None,
447 no_os_cpu_stats_reporting: true,
448 no_os_disk_stats_reporting: true,
449 enforce_ulimit_nofile: false,
451 poh_pinned_cpu_core: poh_service::DEFAULT_PINNED_CPU_CORE,
452 poh_hashes_per_batch: poh_service::DEFAULT_HASHES_PER_BATCH,
453 process_ledger_before_services: false,
454 warp_slot: None,
455 accounts_db_skip_shrink: false,
456 accounts_db_force_initial_clean: false,
457 staked_nodes_overrides: Arc::new(RwLock::new(HashMap::new())),
458 validator_exit: Arc::new(RwLock::new(Exit::default())),
459 validator_exit_backpressure: HashMap::default(),
460 no_wait_for_vote_to_start_leader: true,
461 accounts_db_config: ACCOUNTS_DB_CONFIG_FOR_TESTING,
462 wait_to_vote_slot: None,
463 runtime_config: RuntimeConfig::default(),
464 banking_trace_dir_byte_limit: 0,
465 block_verification_method: BlockVerificationMethod::default(),
466 block_production_method: BlockProductionMethod::default(),
467 block_production_num_workers: BankingStage::default_num_workers(),
468 block_production_scheduler_config: SchedulerConfig::default(),
469 enable_block_production_forwarding: true,
471 enable_scheduler_bindings: false,
472 generator_config: None,
473 use_snapshot_archives_at_startup: UseSnapshotArchivesAtStartup::default(),
474 unified_scheduler_handler_threads: None,
475 ip_echo_server_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
478 rayon_global_threads: NonZeroUsize::new(2).expect("2 is non-zero"),
479 replay_forks_threads: NonZeroUsize::new(1).expect("1 is non-zero"),
480 replay_transactions_threads: NonZeroUsize::new(2).expect("2 is non-zero"),
481 tvu_shred_sigverify_threads: NonZeroUsize::new(2).expect("2 is non-zero"),
482 tvu_bls_sigverify_threads: NonZeroUsize::new(2).expect("2 is non-zero"),
483 delay_leader_block_for_pending_fork: true,
484 voting_service_test_override: None,
485 repair_handler_type: RepairHandlerType::default(),
486 snapshot_packager_niceness_adj: 0,
487 }
488 }
489
490 #[cfg(feature = "dev-context-only-utils")]
491 pub fn enable_default_rpc_block_subscribe(&mut self) {
492 self.pubsub_config = PubSubConfig {
493 enable_block_subscription: true,
494 ..PubSubConfig::default_for_tests()
495 };
496 self.rpc_config = JsonRpcConfig {
497 enable_rpc_transaction_history: true,
498 ..JsonRpcConfig::default_for_test()
499 };
500 }
501}
502
503#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
507pub enum ValidatorStartProgress {
508 #[default]
509 Initializing, SearchingForRpcService,
511 DownloadingSnapshot {
512 slot: Slot,
513 rpc_addr: SocketAddr,
514 },
515 CleaningBlockStore,
516 CleaningAccounts,
517 LoadingLedger,
518 ProcessingLedger {
519 slot: Slot,
520 max_slot: Slot,
521 },
522 StartingServices,
523 Halted,
528 WaitingForSupermajority {
529 slot: Slot,
530 gossip_stake_percent: u64,
531 },
532
533 Running,
536}
537
538pub struct XdpTransmitSetup {
539 pub transmitter_builder: TransmitterBuilder,
540 pub src_ip: Ipv4Addr,
541}
542
543struct BlockstoreRootScan {
544 thread: Option<JoinHandle<Result<usize, BlockstoreError>>>,
545}
546
547impl BlockstoreRootScan {
548 fn new(config: &ValidatorConfig, blockstore: Arc<Blockstore>, exit: Arc<AtomicBool>) -> Self {
549 let thread = if config.rpc_addrs.is_some()
550 && config.rpc_config.enable_rpc_transaction_history
551 && config.rpc_config.rpc_scan_and_fix_roots
552 {
553 Some(
554 Builder::new()
555 .name("solBStoreRtScan".to_string())
556 .spawn(move || blockstore.scan_and_fix_roots(None, None, &exit))
557 .unwrap(),
558 )
559 } else {
560 None
561 };
562 Self { thread }
563 }
564
565 fn join(self) {
566 if let Some(blockstore_root_scan) = self.thread
567 && let Err(err) = blockstore_root_scan.join()
568 {
569 warn!("blockstore_root_scan failed to join {err:?}");
570 }
571 }
572}
573
574#[derive(Default)]
575struct TransactionHistoryServices {
576 transaction_status_sender: Option<TransactionStatusSender>,
577 transaction_status_service: Option<TransactionStatusService>,
578 max_complete_transaction_status_slot: Arc<AtomicU64>,
579}
580
581pub struct ValidatorTpuConfig {
583 pub vote_use_quic: bool,
585 pub tpu_connection_pool_size: usize,
587 pub tpu_quic_server_config: SwQosQuicStreamerConfig,
589 pub tpu_fwd_quic_server_config: SwQosQuicStreamerConfig,
591 pub vote_quic_server_config: SimpleQosQuicStreamerConfig,
593 pub sigverify_threads: NonZeroUsize,
595}
596
597impl ValidatorTpuConfig {
598 pub fn new_for_tests() -> Self {
601 let tpu_quic_server_config = SwQosQuicStreamerConfig {
602 quic_streamer_config: QuicStreamerConfig {
603 max_connections_per_ipaddr_per_min: 32,
604 stream_receive_window_size: solana_message::v1::MAX_TRANSACTION_SIZE as u32,
605 max_stream_data_bytes: solana_message::v1::MAX_TRANSACTION_SIZE as u32,
606 ..Default::default()
607 },
608 qos_config: SwQosConfig::default(),
609 };
610
611 let tpu_fwd_quic_server_config = SwQosQuicStreamerConfig {
612 quic_streamer_config: QuicStreamerConfig {
613 max_connections_per_ipaddr_per_min: 32,
614 ..Default::default()
615 },
616 qos_config: SwQosConfig {
617 max_unstaked_connections: 0,
618 ..Default::default()
619 },
620 };
621
622 let vote_quic_server_config = SimpleQosQuicStreamerConfig {
624 quic_streamer_config: QuicStreamerConfig {
625 max_connections_per_ipaddr_per_min: 32,
626 ..Default::default()
627 },
628 qos_config: SimpleQosConfig::default(),
629 };
630
631 let sigverify_threads = NonZeroUsize::new(2).expect("2 is non-zero");
633
634 ValidatorTpuConfig {
635 vote_use_quic: DEFAULT_VOTE_USE_QUIC,
636 tpu_connection_pool_size: DEFAULT_TPU_CONNECTION_POOL_SIZE,
637 tpu_quic_server_config,
638 tpu_fwd_quic_server_config,
639 vote_quic_server_config,
640 sigverify_threads,
641 }
642 }
643}
644
645pub struct Validator {
646 exit: Arc<AtomicBool>,
648 validator_exit: Arc<RwLock<Exit>>,
649 #[cfg_attr(not(unix), allow(dead_code))]
650 log_config: Option<ValidatorLogConfig>,
651 json_rpc_service: Option<JsonRpcService>,
652 pubsub_service: Option<PubSubService>,
653 rpc_completed_slots_service: Option<JoinHandle<()>>,
654 optimistically_confirmed_bank_tracker: Option<OptimisticallyConfirmedBankTracker>,
655 transaction_status_service: Option<TransactionStatusService>,
656 entry_notifier_service: Option<EntryNotifierService>,
657 system_monitor_service: Option<SystemMonitorService>,
658 sample_performance_service: Option<SamplePerformanceService>,
659 stats_reporter_service: StatsReporterService,
660 gossip_service: GossipService,
661 serve_repair_service: ServeRepairService,
662 completed_data_sets_service: Option<CompletedDataSetsService>,
663 snapshot_packager_service: SnapshotPackagerService,
664 poh_recorder: Arc<RwLock<PohRecorder>>,
665 poh_service: PohService,
666 block_creation_loop: BlockCreationLoop,
667 tpu: Tpu,
668 tvu: Tvu,
669 ip_echo_server: Option<solana_net_utils::IpEchoServer>,
670 pub cluster_info: Arc<ClusterInfo>,
671 pub bank_forks: Arc<RwLock<BankForks>>,
672 pub blockstore: Arc<Blockstore>,
673 geyser_plugin_service: Option<GeyserPluginService>,
674 _contact_info_notifier: Option<GeyserContactInfoNotifier>,
678 blockstore_metric_report_service: BlockstoreMetricReportService,
679 accounts_background_service: AccountsBackgroundService,
680 xdp_transmitter: Option<Transmitter>,
681 _tpu_client_next_runtime: Option<TokioRuntime>,
685}
686
687impl Validator {
688 #[allow(clippy::too_many_arguments)]
689 pub fn new(
690 node: Node,
691 identity_keypair: Arc<Keypair>,
692 ledger_path: &Path,
693 vote_account: &Pubkey,
694 authorized_voter_keypairs: Arc<RwLock<Vec<Arc<Keypair>>>>,
695 cluster_entrypoints: Vec<ContactInfo>,
696 config: &ValidatorConfig,
697 rpc_to_plugin_manager_receiver: Option<Receiver<GeyserPluginManagerRequest>>,
698 start_progress: Arc<RwLock<ValidatorStartProgress>>,
699 socket_addr_space: SocketAddrSpace,
700 tpu_config: ValidatorTpuConfig,
701 admin_rpc_service_post_init: Arc<RwLock<Option<AdminRpcRequestMetadataPostInit>>>,
702 xdp_transmit_setup: Option<XdpTransmitSetup>,
703 ) -> Result<Self> {
704 let exit = Arc::new(AtomicBool::new(false));
705 Self::new_with_exit(
706 node,
707 identity_keypair,
708 ledger_path,
709 vote_account,
710 authorized_voter_keypairs,
711 cluster_entrypoints,
712 config,
713 rpc_to_plugin_manager_receiver,
714 start_progress,
715 socket_addr_space,
716 tpu_config,
717 admin_rpc_service_post_init,
718 xdp_transmit_setup,
719 exit,
720 )
721 }
722
723 #[allow(clippy::too_many_arguments)]
724 pub fn new_with_exit(
725 mut node: Node,
726 identity_keypair: Arc<Keypair>,
727 ledger_path: &Path,
728 vote_account: &Pubkey,
729 authorized_voter_keypairs: Arc<RwLock<Vec<Arc<Keypair>>>>,
730 cluster_entrypoints: Vec<ContactInfo>,
731 config: &ValidatorConfig,
732 rpc_to_plugin_manager_receiver: Option<Receiver<GeyserPluginManagerRequest>>,
733 start_progress: Arc<RwLock<ValidatorStartProgress>>,
734 socket_addr_space: SocketAddrSpace,
735 tpu_config: ValidatorTpuConfig,
736 admin_rpc_service_post_init: Arc<RwLock<Option<AdminRpcRequestMetadataPostInit>>>,
737 xdp_transmit_setup: Option<XdpTransmitSetup>,
738 exit: Arc<AtomicBool>,
739 ) -> Result<Self> {
740 #[cfg(debug_assertions)]
741 const DEBUG_ASSERTION_STATUS: &str = "enabled";
742 #[cfg(not(debug_assertions))]
743 const DEBUG_ASSERTION_STATUS: &str = "disabled";
744 info!("debug-assertion status: {DEBUG_ASSERTION_STATUS}");
745
746 let ValidatorTpuConfig {
747 vote_use_quic,
748 tpu_connection_pool_size,
749 tpu_quic_server_config,
750 tpu_fwd_quic_server_config,
751 vote_quic_server_config,
752 sigverify_threads: tpu_sigverify_threads,
753 } = tpu_config;
754
755 let start_time = Instant::now();
756
757 adjust_nofile_limit(config.enforce_ulimit_nofile)?;
758
759 if rayon::ThreadPoolBuilder::new()
763 .thread_name(|i| format!("solRayonGlob{i:02}"))
764 .num_threads(config.rayon_global_threads.get())
765 .build_global()
766 .is_err()
767 {
768 warn!("Rayon global thread pool already initialized");
769 }
770
771 let id = identity_keypair.pubkey();
772 assert_eq!(&id, node.info.pubkey());
773
774 info!("identity pubkey: {id}");
775 info!("vote account pubkey: {vote_account}");
776
777 if !config.no_os_network_stats_reporting {
778 verify_net_stats_access().map_err(|e| {
779 ValidatorError::Other(format!("Failed to access network stats: {e:?}"))
780 })?;
781 }
782
783 let mut bank_notification_senders = Vec::new();
784
785 let geyser_plugin_config_files = config
786 .on_start_geyser_plugin_config_files
787 .as_ref()
788 .map(Cow::Borrowed)
789 .or_else(|| {
790 config
791 .geyser_plugin_always_enabled
792 .then_some(Cow::Owned(vec![]))
793 });
794 let geyser_plugin_service =
795 if let Some(geyser_plugin_config_files) = geyser_plugin_config_files {
796 let (confirmed_bank_sender, confirmed_bank_receiver) = unbounded();
797 bank_notification_senders.push(confirmed_bank_sender);
798 let rpc_to_plugin_manager_receiver_and_exit =
799 rpc_to_plugin_manager_receiver.map(|receiver| (receiver, exit.clone()));
800 Some(
801 GeyserPluginService::new_with_receiver(
802 confirmed_bank_receiver,
803 config.geyser_plugin_always_enabled,
804 geyser_plugin_config_files.as_ref(),
805 rpc_to_plugin_manager_receiver_and_exit,
806 )
807 .map_err(|err| {
808 ValidatorError::Other(format!("Failed to load the Geyser plugin: {err:?}"))
809 })?,
810 )
811 } else {
812 None
813 };
814
815 if config.voting_disabled {
816 warn!("voting disabled");
817 authorized_voter_keypairs.write().unwrap().clear();
818 } else {
819 for authorized_voter_keypair in authorized_voter_keypairs.read().unwrap().iter() {
820 warn!("authorized voter: {}", authorized_voter_keypair.pubkey());
821 }
822 }
823
824 for cluster_entrypoint in &cluster_entrypoints {
825 info!("entrypoint: {cluster_entrypoint:?}");
826 }
827
828 if !ledger_path.is_dir() {
829 return Err(anyhow!(
830 "ledger directory does not exist or is not accessible: {ledger_path:?}"
831 ));
832 }
833 let genesis_config = load_genesis(config, ledger_path)?;
834 metrics_config_sanity_check(genesis_config.cluster_type)?;
835
836 info!("Validating accounts paths...");
837 *start_progress.write().unwrap() = ValidatorStartProgress::CleaningAccounts;
838 let mut timer = Measure::start("validate_account_paths");
839 validate_account_paths(config)?;
840 timer.stop();
841 info!("Validating accounts paths done. {timer}");
842
843 snapshot_utils::purge_incomplete_bank_snapshots(&config.snapshot_config.bank_snapshots_dir);
844 snapshot_utils::purge_old_bank_snapshots_at_startup(
845 &config.snapshot_config.bank_snapshots_dir,
846 );
847
848 let cancel = CancellationToken::new();
850 {
851 let exit = exit.clone();
852 config
853 .validator_exit
854 .write()
855 .unwrap()
856 .register_exit(Box::new(move || exit.store(true, Ordering::Relaxed)));
857 let cancel = cancel.clone();
858 config
859 .validator_exit
860 .write()
861 .unwrap()
862 .register_exit(Box::new(move || cancel.cancel()));
863 }
864
865 let (
866 accounts_update_notifier,
867 transaction_notifier,
868 deshred_transaction_notifier,
869 entry_notifier,
870 block_metadata_notifier,
871 slot_status_notifier,
872 ) = if let Some(service) = &geyser_plugin_service {
873 (
874 service.get_accounts_update_notifier(),
875 service.get_transaction_notifier(),
876 service.get_deshred_transaction_notifier(),
877 service.get_entry_notifier(),
878 service.get_block_metadata_notifier(),
879 service.get_slot_status_notifier(),
880 )
881 } else {
882 (None, None, None, None, None, None)
883 };
884
885 info!(
886 "Geyser plugin: accounts_update_notifier: {}, transaction_notifier: {}, \
887 deshred_transaction_notifier: {}, entry_notifier: {}",
888 accounts_update_notifier.is_some(),
889 transaction_notifier.is_some(),
890 deshred_transaction_notifier.is_some(),
891 entry_notifier.is_some()
892 );
893
894 let system_monitor_service = Some(SystemMonitorService::new(
895 exit.clone(),
896 SystemMonitorStatsReportConfig {
897 report_os_memory_stats: !config.no_os_memory_stats_reporting,
898 report_os_network_stats: !config.no_os_network_stats_reporting,
899 xdp_network_config_report: config.xdp_network_config_report.clone(),
900 report_os_cpu_stats: !config.no_os_cpu_stats_reporting,
901 report_os_disk_stats: !config.no_os_disk_stats_reporting,
902 },
903 ));
904
905 let dependency_tracker = Arc::new(DependencyTracker::default());
906
907 let (
908 bank_forks,
909 blockstore,
910 original_blockstore_root,
911 ledger_signal_receiver,
912 update_parent_receiver,
913 leader_schedule_cache,
914 starting_snapshot_hashes,
915 TransactionHistoryServices {
916 transaction_status_sender,
917 transaction_status_service,
918 max_complete_transaction_status_slot,
919 },
920 blockstore_process_options,
921 blockstore_root_scan,
922 pruned_banks_receiver,
923 entry_notifier_service,
924 ) = load_blockstore(
925 config,
926 ledger_path,
927 &genesis_config,
928 exit.clone(),
929 &start_progress,
930 accounts_update_notifier,
931 transaction_notifier,
932 entry_notifier,
933 config
934 .rpc_addrs
935 .is_some()
936 .then(|| dependency_tracker.clone()),
937 )
938 .map_err(ValidatorError::Other)?;
939
940 let migration_status = bank_forks.read().unwrap().migration_status();
941
942 if !config.no_poh_speed_test && !migration_status.is_alpenglow_enabled() {
943 check_poh_speed(&bank_forks.read().unwrap().root_bank(), None)?;
944 }
945
946 let (root_slot, hard_forks) = {
947 let root_bank = bank_forks.read().unwrap().root_bank();
948 (root_bank.slot(), root_bank.hard_forks())
949 };
950 let shred_version = compute_shred_version(&genesis_config.hash(), Some(&hard_forks));
951 info!("shred version: {shred_version}, hard forks: {hard_forks:?}");
952
953 if let Some(expected_shred_version) = config.expected_shred_version
954 && expected_shred_version != shred_version
955 {
956 return Err(ValidatorError::ShredVersionMismatch {
957 actual: shred_version,
958 expected: expected_shred_version,
959 }
960 .into());
961 }
962
963 if let Some(start_slot) = should_cleanup_blockstore_incorrect_shred_versions(
964 config,
965 &blockstore,
966 root_slot,
967 &hard_forks,
968 )? {
969 *start_progress.write().unwrap() = ValidatorStartProgress::CleaningBlockStore;
970 cleanup_blockstore_incorrect_shred_versions(
971 &blockstore,
972 config,
973 start_slot,
974 shred_version,
975 )?;
976 } else {
977 info!("Skipping the blockstore check for shreds with incorrect version");
978 }
979
980 node.info.set_shred_version(shred_version);
981 node.info.set_wallclock(timestamp());
982 Self::print_node_info(&node);
983
984 let mut cluster_info = ClusterInfo::new(
985 node.info.clone(),
986 identity_keypair.clone(),
987 socket_addr_space,
988 );
989 cluster_info.set_contact_debug_interval(config.contact_debug_interval);
990 if let Some(known_validators) = &config.known_validators {
991 cluster_info
992 .set_trim_keep_pubkeys(known_validators.iter().copied())
993 .expect("set_trim_keep_pubkeys should succeed as ClusterInfo was just created");
994 }
995 cluster_info.set_entrypoints(cluster_entrypoints);
996 cluster_info.restore_contact_info(ledger_path, config.contact_save_interval);
997 cluster_info.set_bind_ip_addrs(node.bind_ip_addrs.clone());
998 let cluster_info = Arc::new(cluster_info);
999 cluster_info.set_migration_status(migration_status.clone());
1000 let node_multihoming = Arc::new(NodeMultihoming::from(&node));
1001 migration_status.set_pubkey(cluster_info.id());
1002
1003 let contact_info_notifier = geyser_plugin_service.as_ref().and_then(|service| {
1007 geyser_contact_info_notifier::attach(
1008 service.plugin_manager_handle(),
1009 cluster_info.as_ref(),
1010 geyser_contact_info_notifier::DEFAULT_CHANNEL_CAPACITY,
1011 )
1012 });
1013
1014 assert!(is_snapshot_config_valid(&config.snapshot_config));
1015
1016 let (snapshot_request_sender, snapshot_request_receiver) = unbounded();
1017 let snapshot_controller = Arc::new(SnapshotController::new(
1018 snapshot_request_sender,
1019 config.snapshot_config.clone(),
1020 bank_forks.read().unwrap().root(),
1021 ));
1022
1023 let pending_snapshot_packages = Arc::new(Mutex::new(PendingSnapshotPackages::default()));
1024 let exit_backpressure = config
1025 .validator_exit_backpressure
1026 .get(SnapshotPackagerService::NAME)
1027 .cloned();
1028 let enable_gossip_push = true;
1029 let snapshot_packager_service = SnapshotPackagerService::new(
1030 pending_snapshot_packages.clone(),
1031 starting_snapshot_hashes,
1032 exit.clone(),
1033 exit_backpressure,
1034 cluster_info.clone(),
1035 snapshot_controller.clone(),
1036 enable_gossip_push,
1037 config.snapshot_packager_niceness_adj,
1038 );
1039 let snapshot_request_handler = SnapshotRequestHandler {
1040 snapshot_controller: snapshot_controller.clone(),
1041 snapshot_request_receiver,
1042 pending_snapshot_packages,
1043 };
1044 let pruned_banks_request_handler = PrunedBanksRequestHandler {
1045 pruned_banks_receiver,
1046 };
1047 let accounts_background_service = AccountsBackgroundService::new(
1048 bank_forks.clone(),
1049 exit.clone(),
1050 AbsRequestHandlers {
1051 snapshot_request_handler,
1052 pruned_banks_request_handler,
1053 },
1054 );
1055 info!(
1056 "Using: block-verification-method: {}, block-production-method: {}",
1057 config.block_verification_method, config.block_production_method,
1058 );
1059
1060 let (replay_vote_sender, replay_vote_receiver) = unbounded();
1061
1062 let prioritization_fee_cache = if config.rpc_config.full_api {
1063 Some(Arc::new(PrioritizationFeeCache::default()))
1064 } else {
1065 None
1066 };
1067
1068 let leader_schedule_cache = Arc::new(leader_schedule_cache);
1069 let (poh_recorder, entry_receiver) = {
1070 let bank = &bank_forks.read().unwrap().working_bank();
1071 PohRecorder::new_with_clear_signal(
1072 bank.tick_height(),
1073 bank.last_blockhash(),
1074 bank.clone(),
1075 None,
1076 bank.ticks_per_slot(),
1077 config.delay_leader_block_for_pending_fork,
1078 blockstore.clone(),
1079 blockstore.get_new_shred_signal(0),
1080 &leader_schedule_cache,
1081 &genesis_config.poh_config,
1082 exit.clone(),
1083 )
1084 };
1085 let (record_sender, record_receiver) = record_channels(transaction_status_sender.is_some());
1086 let transaction_recorder = TransactionRecorder::new(record_sender);
1087 let poh_recorder = Arc::new(RwLock::new(poh_recorder));
1088 let (poh_controller, poh_service_message_receiver) = PohController::new();
1089 let (bank_forks_controller, bank_forks_controller_receiver) =
1090 BankForksControllerHandle::new();
1091 let bank_forks_controller = Arc::new(bank_forks_controller);
1092
1093 let (banking_tracer, tracer_thread) =
1094 BankingTracer::new((config.banking_trace_dir_byte_limit > 0).then_some((
1095 &blockstore.banking_trace_path(),
1096 exit.clone(),
1097 config.banking_trace_dir_byte_limit,
1098 )))?;
1099 if banking_tracer.is_enabled() {
1100 info!(
1101 "Enabled banking trace (dir_byte_limit: {})",
1102 config.banking_trace_dir_byte_limit
1103 );
1104 } else {
1105 info!("Disabled banking trace");
1106 }
1107 let banking_tracer_channels = banking_tracer.create_channels();
1108
1109 let scheduler_pool = DefaultSchedulerPool::new(
1110 config.unified_scheduler_handler_threads,
1111 config.runtime_config.log_messages_bytes_limit,
1112 transaction_status_sender.clone(),
1113 Some(replay_vote_sender.clone()),
1114 prioritization_fee_cache.clone(),
1115 );
1116 bank_forks
1117 .write()
1118 .unwrap()
1119 .install_scheduler_pool(scheduler_pool);
1120
1121 let entry_notification_sender = entry_notifier_service
1122 .as_ref()
1123 .map(|service| service.sender());
1124 let mut process_blockstore = ProcessBlockStore::new(
1125 &id,
1126 vote_account,
1127 &start_progress,
1128 &blockstore,
1129 original_blockstore_root,
1130 &bank_forks,
1131 &leader_schedule_cache,
1132 &blockstore_process_options,
1133 transaction_status_sender.as_ref(),
1134 entry_notification_sender,
1135 blockstore_root_scan,
1136 &snapshot_controller,
1137 config,
1138 cluster_info.my_shred_version(),
1139 );
1140
1141 maybe_warp_slot(
1142 config,
1143 &mut process_blockstore,
1144 ledger_path,
1145 &bank_forks,
1146 &leader_schedule_cache,
1147 &snapshot_controller,
1148 )
1149 .map_err(ValidatorError::Other)?;
1150
1151 if config.process_ledger_before_services {
1152 process_blockstore
1153 .process()
1154 .map_err(ValidatorError::Other)?;
1155 }
1156 *start_progress.write().unwrap() = ValidatorStartProgress::StartingServices;
1157
1158 let mut block_commitment_cache = BlockCommitmentCache::default();
1159 let bank_forks_guard = bank_forks.read().unwrap();
1160 block_commitment_cache.initialize_slots(
1161 bank_forks_guard.working_bank().slot(),
1162 bank_forks_guard.root(),
1163 );
1164 drop(bank_forks_guard);
1165 let block_commitment_cache = Arc::new(RwLock::new(block_commitment_cache));
1166
1167 let optimistically_confirmed_bank =
1168 OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
1169
1170 let max_slots = Arc::new(MaxSlots::default());
1171
1172 let staked_nodes = Arc::new(RwLock::new(StakedNodes::default()));
1173
1174 let mut tpu_transactions_forwards_client_sockets =
1175 Some(node.sockets.tpu_transaction_forwarding_clients);
1176
1177 let vote_connection_cache = if vote_use_quic {
1178 let vote_connection_cache = ConnectionCache::new_with_client_options(
1179 "connection_cache_vote_quic",
1180 tpu_connection_pool_size,
1181 Some(node.sockets.quic_vote_client),
1182 Some((
1183 &identity_keypair,
1184 node.info
1185 .tpu_vote(Protocol::QUIC)
1186 .ok_or_else(|| {
1187 ValidatorError::Other(String::from("Invalid QUIC address for TPU Vote"))
1188 })?
1189 .ip(),
1190 )),
1191 Some((&staked_nodes, &identity_keypair.pubkey())),
1192 );
1193 Arc::new(vote_connection_cache)
1194 } else {
1195 Arc::new(ConnectionCache::with_udp(
1196 "connection_cache_vote_udp",
1197 tpu_connection_pool_size,
1198 ))
1199 };
1200
1201 let bls_connection_cache = Arc::new(ConnectionCache::new_with_max_connections(
1202 "connection_cache_bls_quic",
1203 1, MAX_ALPENGLOW_VOTE_ACCOUNTS * 2, Some(node.sockets.quic_alpenglow_client),
1209 Some((
1210 &identity_keypair,
1211 node.info
1212 .alpenglow()
1213 .ok_or_else(|| {
1214 ValidatorError::Other(String::from(
1215 "Invalid QUIC address for Alpenglow BLS",
1216 ))
1217 })?
1218 .ip(),
1219 )),
1220 Some((&staked_nodes, &identity_keypair.pubkey())),
1221 ));
1222 let key_notifiers = Arc::new(RwLock::new(KeyUpdaters::default()));
1223 key_notifiers.write().unwrap().add(
1224 KeyUpdaterType::BlsConnectionCache,
1225 bls_connection_cache.clone(),
1226 );
1227
1228 let current_runtime_handle = tokio::runtime::Handle::try_current();
1234 let tpu_client_next_runtime = current_runtime_handle.is_err().then(|| {
1235 tokio::runtime::Builder::new_multi_thread()
1236 .enable_all()
1237 .worker_threads(2)
1238 .thread_name("solTpuClientRt")
1239 .build()
1240 .unwrap()
1241 });
1242
1243 let rpc_override_health_check =
1244 Arc::new(AtomicBool::new(config.rpc_config.disable_health_check));
1245 let (
1246 json_rpc_service,
1247 rpc_subscriptions,
1248 pubsub_service,
1249 rpc_completed_slots_service,
1250 sample_performance_service,
1251 optimistically_confirmed_bank_tracker,
1252 bank_notification_sender,
1253 ) = if let Some((rpc_addr, rpc_pubsub_addr)) = config.rpc_addrs {
1254 assert_eq!(
1255 node.info.rpc().map(|addr| socket_addr_space.check(&addr)),
1256 node.info
1257 .rpc_pubsub()
1258 .map(|addr| socket_addr_space.check(&addr))
1259 );
1260 let (bank_notification_sender, bank_notification_receiver) = unbounded();
1261 let confirmed_bank_subscribers = if !bank_notification_senders.is_empty() {
1262 Some(Arc::new(RwLock::new(bank_notification_senders)))
1263 } else {
1264 None
1265 };
1266
1267 let rpc_tpu_client_args = {
1268 let runtime_handle = tpu_client_next_runtime
1269 .as_ref()
1270 .map(TokioRuntime::handle)
1271 .unwrap_or_else(|| current_runtime_handle.as_ref().unwrap());
1272
1273 RpcTpuClientArgs(
1274 Arc::as_ref(&identity_keypair),
1275 node.sockets.rpc_sts_client,
1276 runtime_handle.clone(),
1277 cancel.clone(),
1278 )
1279 };
1280 let rpc_svc_config = JsonRpcServiceConfig {
1281 rpc_addr,
1282 rpc_config: config.rpc_config.clone(),
1283 snapshot_config: Some(snapshot_controller.snapshot_config().clone()),
1284 bank_forks: bank_forks.clone(),
1285 block_commitment_cache: block_commitment_cache.clone(),
1286 blockstore: blockstore.clone(),
1287 cluster_info: cluster_info.clone(),
1288 poh_recorder: Some(poh_recorder.clone()),
1289 genesis_hash: genesis_config.hash(),
1290 ledger_path: ledger_path.to_path_buf(),
1291 validator_exit: config.validator_exit.clone(),
1292 exit: exit.clone(),
1293 override_health_check: rpc_override_health_check.clone(),
1294 optimistically_confirmed_bank: optimistically_confirmed_bank.clone(),
1295 send_transaction_service_config: config.send_transaction_service_config.clone(),
1296 max_slots: max_slots.clone(),
1297 leader_schedule_cache: leader_schedule_cache.clone(),
1298 max_complete_transaction_status_slot: max_complete_transaction_status_slot.clone(),
1299 prioritization_fee_cache: prioritization_fee_cache.clone(),
1300 rpc_tpu_client_args,
1301 };
1302 let json_rpc_service =
1303 JsonRpcService::new_with_config(rpc_svc_config).map_err(ValidatorError::Other)?;
1304 let rpc_subscriptions = Arc::new(RpcSubscriptions::new_with_config(
1305 exit.clone(),
1306 max_complete_transaction_status_slot,
1307 blockstore.clone(),
1308 bank_forks.clone(),
1309 block_commitment_cache.clone(),
1310 optimistically_confirmed_bank.clone(),
1311 &config.pubsub_config,
1312 None,
1313 ));
1314 let pubsub_service = if !config.rpc_config.full_api {
1315 None
1316 } else {
1317 let (trigger, pubsub_service) = PubSubService::new(
1318 config.pubsub_config.clone(),
1319 &rpc_subscriptions,
1320 rpc_pubsub_addr,
1321 );
1322 config
1323 .validator_exit
1324 .write()
1325 .unwrap()
1326 .register_exit(Box::new(move || trigger.cancel()));
1327
1328 Some(pubsub_service)
1329 };
1330
1331 let rpc_completed_slots_service =
1332 if config.rpc_config.full_api || geyser_plugin_service.is_some() {
1333 let (completed_slots_sender, completed_slots_receiver) =
1334 bounded(MAX_COMPLETED_SLOTS_IN_CHANNEL);
1335 blockstore.add_completed_slots_signal(completed_slots_sender);
1336
1337 Some(RpcCompletedSlotsService::spawn(
1338 completed_slots_receiver,
1339 rpc_subscriptions.clone(),
1340 slot_status_notifier.clone(),
1341 exit.clone(),
1342 ))
1343 } else {
1344 None
1345 };
1346
1347 let sample_performance_service = if config.rpc_config.enable_rpc_transaction_history {
1348 Some(SamplePerformanceService::new(
1349 bank_forks.clone(),
1350 blockstore.clone(),
1351 exit.clone(),
1352 ))
1353 } else {
1354 None
1355 };
1356
1357 let dependency_tracker = transaction_status_sender
1358 .is_some()
1359 .then_some(dependency_tracker);
1360 let optimistically_confirmed_bank_tracker =
1361 Some(OptimisticallyConfirmedBankTracker::new(
1362 bank_notification_receiver,
1363 exit.clone(),
1364 bank_forks.clone(),
1365 optimistically_confirmed_bank,
1366 rpc_subscriptions.clone(),
1367 confirmed_bank_subscribers,
1368 prioritization_fee_cache.clone(),
1369 dependency_tracker.clone(),
1370 ));
1371 let bank_notification_sender_config = Some(BankNotificationSenderConfig {
1372 sender: bank_notification_sender,
1373 should_send_parents: geyser_plugin_service.is_some(),
1374 dependency_tracker,
1375 });
1376 (
1377 Some(json_rpc_service),
1378 Some(rpc_subscriptions),
1379 pubsub_service,
1380 rpc_completed_slots_service,
1381 sample_performance_service,
1382 optimistically_confirmed_bank_tracker,
1383 bank_notification_sender_config,
1384 )
1385 } else {
1386 (None, None, None, None, None, None, None)
1387 };
1388
1389 let (completed_data_sets_sender, completed_data_sets_service) =
1397 if config.rpc_config.full_api || deshred_transaction_notifier.is_some() {
1398 let (completed_data_sets_sender, completed_data_sets_receiver) =
1399 bounded(MAX_COMPLETED_DATA_SETS_IN_CHANNEL);
1400 let completed_data_sets_service = CompletedDataSetsService::new(
1401 completed_data_sets_receiver,
1402 blockstore.clone(),
1403 rpc_subscriptions.clone(),
1404 deshred_transaction_notifier.clone(),
1405 exit.clone(),
1406 max_slots.clone(),
1407 bank_forks.clone(),
1408 );
1409 (
1410 Some(completed_data_sets_sender),
1411 Some(completed_data_sets_service),
1412 )
1413 } else {
1414 (None, None)
1415 };
1416
1417 let ip_echo_server = match node.sockets.ip_echo {
1418 None => None,
1419 Some(tcp_listener) => Some(solana_net_utils::ip_echo_server(
1420 tcp_listener,
1421 config.ip_echo_server_threads,
1422 Some(node.info.shred_version()),
1423 )),
1424 };
1425
1426 let (stats_reporter_sender, stats_reporter_receiver) = unbounded();
1427
1428 let stats_reporter_service =
1429 StatsReporterService::new(stats_reporter_receiver, exit.clone());
1430
1431 let epoch_specs: Box<dyn solana_gossip::epoch_specs::EpochSpecs> =
1432 Box::new(crate::epoch_specs::EpochSpecs::from(bank_forks.clone()));
1433
1434 let (
1435 xdp_transmitter,
1436 turbine_xdp_sender,
1437 quic_xdp_sender,
1438 repair_xdp_sender,
1439 gossip_xdp_sender,
1440 ) = if let Some(XdpTransmitSetup {
1441 transmitter_builder,
1442 src_ip,
1443 }) = xdp_transmit_setup
1444 {
1445 let turbine_src_port = node.sockets.retransmit_sockets[0]
1446 .local_addr()
1447 .expect("retransmit socket should have local address")
1448 .port();
1449
1450 let repair_src_port = node
1451 .sockets
1452 .repair
1453 .local_addr()
1454 .expect("repair socket should have local address")
1455 .port();
1456
1457 let gossip_src_port = node.sockets.gossip[0]
1458 .local_addr()
1459 .expect("gossip socket should have local address")
1460 .port();
1461
1462 let (transmitter, sender) = transmitter_builder.build();
1463 (
1464 Some(transmitter),
1465 Some(PinnedXdpSender::new(
1466 sender.clone(),
1467 SocketAddrV4::new(src_ip, turbine_src_port),
1468 )),
1469 Some((sender.clone(), src_ip)),
1470 Some(PinnedXdpSender::new(
1471 sender.clone(),
1472 SocketAddrV4::new(src_ip, repair_src_port),
1473 )),
1474 Some(PinnedXdpSender::new(
1475 sender,
1476 SocketAddrV4::new(src_ip, gossip_src_port),
1477 )),
1478 )
1479 } else {
1480 (None, None, None, None, None)
1481 };
1482
1483 let gossip_service = GossipService::new(
1484 &cluster_info,
1485 Some(epoch_specs),
1486 node.sockets.gossip.clone(),
1487 gossip_xdp_sender,
1488 config.gossip_validators.clone(),
1489 config.should_check_duplicate_instance,
1490 Some(stats_reporter_sender.clone()),
1491 exit.clone(),
1492 );
1493 let serve_repair = {
1494 let bank_forks_r = bank_forks.read().unwrap();
1495 let leader_state = poh_recorder.read().unwrap().shared_leader_state();
1496 config.repair_handler_type.create_serve_repair(
1497 blockstore.clone(),
1498 cluster_info.clone(),
1499 bank_forks_r.sharable_banks(),
1500 config.repair_whitelist.clone(),
1501 leader_state,
1502 leader_schedule_cache.clone(),
1503 bank_forks_r.migration_status(),
1504 )
1505 };
1506
1507 let waited_for_supermajority = wait_for_supermajority(
1508 config,
1509 Some(&mut process_blockstore),
1510 &bank_forks,
1511 &cluster_info,
1512 rpc_override_health_check,
1513 &start_progress,
1514 )?;
1515
1516 let blockstore_metric_report_service =
1517 BlockstoreMetricReportService::new(blockstore.clone(), exit.clone());
1518
1519 let wait_for_vote_to_start_leader =
1520 !waited_for_supermajority && !config.no_wait_for_vote_to_start_leader;
1521
1522 let (record_receiver_sender, record_receiver_receiver) = bounded(1);
1525 let (leader_window_info_sender, leader_window_info_receiver) = bounded(7);
1529
1530 let poh_service = PohService::new(
1531 poh_recorder.clone(),
1532 &genesis_config.poh_config,
1533 exit.clone(),
1534 bank_forks.read().unwrap().root_bank().ticks_per_slot(),
1535 config.poh_pinned_cpu_core,
1536 config.poh_hashes_per_batch,
1537 record_receiver,
1538 poh_service_message_receiver,
1539 migration_status.clone(),
1540 record_receiver_sender,
1541 );
1542
1543 let replay_highest_frozen = Arc::new(ReplayHighestFrozen::default());
1544 let highest_parent_ready = Arc::new(RwLock::default());
1545 let highest_finalized = Arc::new(RwLock::new(None));
1547 let (optimistic_parent_sender, optimistic_parent_receiver) = bounded(100);
1551
1552 let banking_stage_sender_for_bcl = banking_tracer_channels.non_vote_sender.clone();
1553
1554 let block_creation_loop_config = BlockCreationLoopConfig {
1555 exit: exit.clone(),
1556 bank_forks: bank_forks.clone(),
1557 bank_forks_controller: bank_forks_controller.clone(),
1558 blockstore: blockstore.clone(),
1559 cluster_info: cluster_info.clone(),
1560 poh_recorder: poh_recorder.clone(),
1561 leader_schedule_cache: leader_schedule_cache.clone(),
1562 rpc_subscriptions: rpc_subscriptions.clone(),
1563 banking_tracer: banking_tracer.clone(),
1564 slot_status_notifier: slot_status_notifier.clone(),
1565 leader_window_info_receiver,
1566 highest_parent_ready: highest_parent_ready.clone(),
1567 replay_highest_frozen: replay_highest_frozen.clone(),
1568 record_receiver_receiver,
1569 optimistic_parent_receiver: optimistic_parent_receiver.clone(),
1570 highest_finalized: highest_finalized.clone(),
1571 banking_stage_sender: banking_stage_sender_for_bcl,
1572 sharable_banks: bank_forks.read().unwrap().sharable_banks(),
1573 };
1574 let (block_creation_loop, reward_votes_sender) =
1575 BlockCreationLoop::new(block_creation_loop_config);
1576
1577 assert_eq!(
1578 blockstore.get_new_shred_signals_len(),
1579 1,
1580 "New shred signal for the TVU should be the same as the clear bank signal."
1581 );
1582
1583 let vote_tracker = Arc::<VoteTracker>::default();
1584
1585 let (retransmit_slots_sender, retransmit_slots_receiver) = unbounded();
1586 let (verified_vote_sender, verified_vote_receiver) = unbounded();
1587 let (gossip_verified_vote_hash_sender, gossip_verified_vote_hash_receiver) = unbounded();
1588 let (duplicate_confirmed_slot_sender, duplicate_confirmed_slots_receiver) = unbounded();
1589
1590 let entry_notification_sender = entry_notifier_service
1591 .as_ref()
1592 .map(|service| service.sender_cloned());
1593
1594 let serve_repair_service = ServeRepairService::new(
1595 serve_repair,
1596 node.sockets.serve_repair,
1597 socket_addr_space,
1598 stats_reporter_sender,
1599 exit.clone(),
1600 );
1601
1602 let (tower, vote_history) = process_blockstore.process().map_err(|e| {
1603 ValidatorError::Other(format!(
1604 "Unable to restore Tower or VoteHistory and either --require-tower was specified \
1605 or --do-not-require-vote-history was not specified. Aborting {e}"
1606 ))
1607 })?;
1608 info!("Tower state: {tower:?}, Vote History state: {vote_history:?}");
1609
1610 migration_status.log_phase();
1611
1612 let outstanding_repair_requests =
1613 Arc::<RwLock<repair::repair_service::OutstandingShredRepairs>>::default();
1614 let root_bank = bank_forks.read().unwrap().root_bank();
1615 let cluster_slots = Arc::new({
1616 crate::cluster_slots_service::cluster_slots::ClusterSlots::new(
1617 &root_bank,
1618 &cluster_info,
1619 )
1620 });
1621 let (votor_event_sender, votor_event_receiver) = bounded(1000);
1623
1624 let tvu = Tvu::new(
1625 vote_account,
1626 authorized_voter_keypairs,
1627 bank_forks.clone(),
1628 &cluster_info,
1629 TvuSockets {
1630 repair: node.sockets.repair.try_clone().unwrap(),
1631 retransmit: node.sockets.retransmit_sockets,
1632 fetch: node.sockets.tvu,
1633 ancestor_hashes_requests: node.sockets.ancestor_hashes_requests,
1634 alpenglow: node.sockets.alpenglow,
1635 block_id_repair: node.sockets.block_id_repair,
1636 },
1637 blockstore.clone(),
1638 ledger_signal_receiver,
1639 update_parent_receiver,
1640 rpc_subscriptions.clone(),
1641 &poh_recorder,
1642 poh_controller,
1643 tower,
1644 config.tower_storage.clone(),
1645 vote_history,
1646 config.vote_history_storage.clone(),
1647 &leader_schedule_cache,
1648 exit.clone(),
1649 block_commitment_cache,
1650 config.turbine_mode.clone(),
1651 transaction_status_sender.clone(),
1652 entry_notification_sender.clone(),
1653 vote_tracker.clone(),
1654 retransmit_slots_sender,
1655 gossip_verified_vote_hash_receiver,
1656 verified_vote_sender.clone(),
1657 verified_vote_receiver,
1658 replay_vote_sender.clone(),
1659 completed_data_sets_sender,
1660 bank_notification_sender.clone(),
1661 duplicate_confirmed_slots_receiver,
1662 TvuConfig {
1663 max_ledger_shreds: config.max_ledger_shreds,
1664 shred_version: node.info.shred_version(),
1665 repair_validators: config.repair_validators.clone(),
1666 repair_whitelist: config.repair_whitelist.clone(),
1667 wait_for_vote_to_start_leader,
1668 replay_forks_threads: config.replay_forks_threads,
1669 replay_transactions_threads: config.replay_transactions_threads,
1670 shred_sigverify_threads: config.tvu_shred_sigverify_threads,
1671 bls_sigverify_threads: config.tvu_bls_sigverify_threads,
1672 turbine_xdp_sender: turbine_xdp_sender.clone(),
1673 repair_xdp_sender,
1674 },
1675 &max_slots,
1676 block_metadata_notifier,
1677 config.wait_to_vote_slot,
1678 Some(snapshot_controller.clone()),
1679 banking_tracer,
1680 outstanding_repair_requests.clone(),
1681 cluster_slots.clone(),
1682 slot_status_notifier,
1683 vote_connection_cache,
1684 AlpenglowInitializationState {
1685 leader_window_info_sender,
1686 optimistic_parent_sender,
1687 optimistic_parent_receiver,
1688 replay_highest_frozen,
1689 highest_parent_ready,
1690 bank_forks_controller,
1691 bank_forks_controller_receiver,
1692 votor_event_sender: votor_event_sender.clone(),
1693 votor_event_receiver,
1694 cancel: cancel.clone(),
1695 staked_nodes: staked_nodes.clone(),
1696 key_notifiers: key_notifiers.clone(),
1697 bls_connection_cache,
1698 voting_service_test_override: config.voting_service_test_override.clone(),
1699 highest_finalized,
1700 },
1701 reward_votes_sender,
1702 )
1703 .map_err(ValidatorError::Other)?;
1704
1705 let tpu_forwarding_client_config = {
1706 let runtime_handle = tpu_client_next_runtime
1707 .as_ref()
1708 .map(TokioRuntime::handle)
1709 .unwrap_or_else(|| current_runtime_handle.as_ref().unwrap());
1710 ForwardingClientConfig {
1711 stake_identity: Arc::as_ref(&identity_keypair),
1712 tpu_client_sockets: tpu_transactions_forwards_client_sockets.take().unwrap(),
1713 runtime_handle: runtime_handle.clone(),
1714 cancel: cancel.clone(),
1715 node_multihoming: node_multihoming.clone(),
1716 }
1717 };
1718 let (banking_control_sender, banking_control_receiver) = mpsc::channel(1);
1719 let tpu = Tpu::new_with_client(
1720 &cluster_info,
1721 &poh_recorder,
1722 transaction_recorder,
1723 entry_receiver,
1724 retransmit_slots_receiver,
1725 TpuSockets {
1726 vote: node.sockets.tpu_vote,
1727 broadcast: node.sockets.broadcast,
1728 transactions_quic: node.sockets.tpu_quic,
1729 transactions_forwards_quic: node.sockets.tpu_forwards_quic,
1730 vote_quic: node.sockets.tpu_vote_quic,
1731 vote_forwarding_client: node.sockets.tpu_vote_forwarding_client,
1732 },
1733 rpc_subscriptions,
1734 transaction_status_sender,
1735 entry_notification_sender,
1736 blockstore.clone(),
1737 &config.broadcast_stage_type,
1738 leader_schedule_cache.clone(),
1739 turbine_xdp_sender,
1740 quic_xdp_sender,
1741 exit.clone(),
1742 node.info.shred_version(),
1743 vote_tracker,
1744 bank_forks.clone(),
1745 verified_vote_sender,
1746 gossip_verified_vote_hash_sender,
1747 replay_vote_receiver,
1748 replay_vote_sender,
1749 bank_notification_sender,
1750 duplicate_confirmed_slot_sender,
1751 tpu_forwarding_client_config,
1752 &identity_keypair,
1753 config.runtime_config.log_messages_bytes_limit,
1754 &staked_nodes,
1755 config.staked_nodes_overrides.clone(),
1756 banking_tracer_channels,
1757 tracer_thread,
1758 tpu_quic_server_config,
1759 tpu_fwd_quic_server_config,
1760 vote_quic_server_config,
1761 prioritization_fee_cache,
1762 tpu_sigverify_threads,
1763 config.block_production_method.clone(),
1764 config.block_production_num_workers,
1765 config.block_production_scheduler_config.clone(),
1766 config.filter_keys.clone(),
1767 config.enable_block_production_forwarding,
1768 config.generator_config.clone(),
1769 key_notifiers.clone(),
1770 banking_control_receiver,
1771 config.enable_scheduler_bindings.then(|| {
1772 (
1773 ledger_path.join("scheduler_bindings.ipc"),
1774 banking_control_sender.clone(),
1775 )
1776 }),
1777 cancel,
1778 votor_event_sender.clone(),
1779 );
1780
1781 datapoint_info!(
1782 "validator-new",
1783 ("id", id.to_string(), String),
1784 ("version", solana_version::version!(), String),
1785 ("cluster_type", genesis_config.cluster_type as u32, i64),
1786 ("elapsed_ms", start_time.elapsed().as_millis() as i64, i64),
1787 ("waited_for_supermajority", waited_for_supermajority, bool),
1788 ("shred_version", shred_version as i64, i64),
1789 );
1790
1791 *start_progress.write().unwrap() = ValidatorStartProgress::Running;
1792 if let Some(json_rpc_service) = &json_rpc_service {
1793 key_notifiers.write().unwrap().add(
1794 KeyUpdaterType::RpcService,
1795 json_rpc_service.get_client_key_updater(),
1796 );
1797 }
1798
1799 *admin_rpc_service_post_init.write().unwrap() = Some(AdminRpcRequestMetadataPostInit {
1800 bank_forks: bank_forks.clone(),
1801 cluster_info: cluster_info.clone(),
1802 vote_account: *vote_account,
1803 repair_whitelist: config.repair_whitelist.clone(),
1804 notifies: key_notifiers,
1805 repair_socket: Arc::new(node.sockets.repair),
1806 outstanding_repair_requests,
1807 cluster_slots,
1808 node: Some(node_multihoming),
1809 banking_control_sender,
1810 snapshot_controller,
1811 blockstore: blockstore.clone(),
1812 votor_event_sender,
1813 });
1814
1815 Ok(Self {
1816 log_config: config.log_config.clone(),
1817 exit,
1818 stats_reporter_service,
1819 gossip_service,
1820 serve_repair_service,
1821 json_rpc_service,
1822 pubsub_service,
1823 rpc_completed_slots_service,
1824 optimistically_confirmed_bank_tracker,
1825 transaction_status_service,
1826 entry_notifier_service,
1827 system_monitor_service,
1828 sample_performance_service,
1829 snapshot_packager_service,
1830 completed_data_sets_service,
1831 tpu,
1832 tvu,
1833 poh_service,
1834 block_creation_loop,
1835 poh_recorder,
1836 ip_echo_server,
1837 validator_exit: config.validator_exit.clone(),
1838 cluster_info,
1839 bank_forks,
1840 blockstore,
1841 geyser_plugin_service,
1842 _contact_info_notifier: contact_info_notifier,
1843 blockstore_metric_report_service,
1844 accounts_background_service,
1845 xdp_transmitter,
1846 _tpu_client_next_runtime: tpu_client_next_runtime,
1847 })
1848 }
1849
1850 pub fn register_logrotate_signal_handler() -> Result<Arc<AtomicBool>> {
1854 let flag = Arc::new(AtomicBool::new(false));
1855 #[cfg(unix)]
1856 {
1857 signal_hook::flag::register(libc::SIGUSR1, flag.clone())?;
1858 }
1859 Ok(flag)
1860 }
1861
1862 pub fn listen_for_signals(&self) -> Result<()> {
1864 info!("Validator::listen_for_signals() has started");
1865 loop {
1866 if self.exit.load(Ordering::Relaxed) {
1867 break;
1868 }
1869
1870 #[cfg(unix)]
1871 if let Some(ValidatorLogConfig {
1872 logfile,
1873 logrotate_flag,
1874 }) = self.log_config.as_ref()
1875 && logrotate_flag.load(Ordering::Relaxed)
1876 {
1877 info!("Received SIGUSR1, reopening {}", logfile.display());
1878 agave_logger::redirect_stderr(logfile);
1879 logrotate_flag.store(false, Ordering::Relaxed);
1883 }
1884
1885 thread::sleep(Duration::from_secs(1));
1888 }
1889 info!("Validator::listen_for_signals() has stopped");
1890
1891 Ok(())
1892 }
1893
1894 pub fn exit(&mut self) {
1896 self.validator_exit.write().unwrap().exit();
1897
1898 self.blockstore.drop_signal();
1900 }
1901
1902 pub fn close(mut self) {
1903 self.exit();
1904 self.join();
1905 }
1906
1907 fn print_node_info(node: &Node) {
1908 info!("{:?}", node.info);
1909 info!(
1910 "local gossip address: {}",
1911 node.sockets.gossip[0].local_addr().unwrap()
1912 );
1913 info!(
1914 "local broadcast address: {}",
1915 node.sockets
1916 .broadcast
1917 .first()
1918 .unwrap()
1919 .local_addr()
1920 .unwrap()
1921 );
1922 info!(
1923 "local repair address: {}",
1924 node.sockets.repair.local_addr().unwrap()
1925 );
1926 info!(
1927 "local retransmit address: {}",
1928 node.sockets.retransmit_sockets[0].local_addr().unwrap()
1929 );
1930 info!(
1931 "local alpenglow address: {}",
1932 node.sockets.alpenglow.local_addr().unwrap()
1933 );
1934 }
1935
1936 pub fn join(self) {
1937 drop(self.bank_forks);
1938 drop(self.cluster_info);
1939
1940 self.poh_service.join().expect("poh_service");
1941 self.block_creation_loop
1942 .join()
1943 .expect("block_creation_loop");
1944 drop(self.poh_recorder);
1945
1946 if let Some(json_rpc_service) = self.json_rpc_service {
1947 json_rpc_service.join().expect("rpc_service");
1948 }
1949
1950 if let Some(pubsub_service) = self.pubsub_service {
1951 pubsub_service.join().expect("pubsub_service");
1952 }
1953
1954 if let Some(rpc_completed_slots_service) = self.rpc_completed_slots_service {
1955 rpc_completed_slots_service
1956 .join()
1957 .expect("rpc_completed_slots_service");
1958 }
1959
1960 if let Some(optimistically_confirmed_bank_tracker) =
1961 self.optimistically_confirmed_bank_tracker
1962 {
1963 optimistically_confirmed_bank_tracker
1964 .join()
1965 .expect("optimistically_confirmed_bank_tracker");
1966 }
1967
1968 if let Some(transaction_status_service) = self.transaction_status_service {
1969 transaction_status_service
1970 .join()
1971 .expect("transaction_status_service");
1972 }
1973
1974 if let Some(system_monitor_service) = self.system_monitor_service {
1975 system_monitor_service
1976 .join()
1977 .expect("system_monitor_service");
1978 }
1979
1980 if let Some(sample_performance_service) = self.sample_performance_service {
1981 sample_performance_service
1982 .join()
1983 .expect("sample_performance_service");
1984 }
1985
1986 if let Some(entry_notifier_service) = self.entry_notifier_service {
1987 entry_notifier_service
1988 .join()
1989 .expect("entry_notifier_service");
1990 }
1991
1992 self.snapshot_packager_service
1993 .join()
1994 .expect("snapshot_packager_service");
1995
1996 self.gossip_service.join().expect("gossip_service");
1997 self.serve_repair_service
1998 .join()
1999 .expect("serve_repair_service");
2000 self.stats_reporter_service
2001 .join()
2002 .expect("stats_reporter_service");
2003 self.blockstore_metric_report_service
2004 .join()
2005 .expect("ledger_metric_report_service");
2006 self.accounts_background_service
2007 .join()
2008 .expect("accounts_background_service");
2009 if let Some(xdp_transmitter) = self.xdp_transmitter {
2010 xdp_transmitter.join().expect("xdp_transmitter");
2011 }
2012 self.tpu.join().expect("tpu");
2013 self.tvu.join().expect("tvu");
2014 if let Some(completed_data_sets_service) = self.completed_data_sets_service {
2015 completed_data_sets_service
2016 .join()
2017 .expect("completed_data_sets_service");
2018 }
2019 if let Some(ip_echo_server) = self.ip_echo_server {
2020 ip_echo_server.shutdown_background();
2021 }
2022
2023 if let Some(geyser_plugin_service) = self.geyser_plugin_service {
2024 geyser_plugin_service.join().expect("geyser_plugin_service");
2025 }
2026 }
2027}
2028
2029fn active_vote_account_exists_in_bank(bank: &Bank, vote_account: &Pubkey) -> bool {
2030 if let Some(account) = &bank.get_account(vote_account)
2031 && let Ok(vote_state) = VoteStateV4::deserialize(account.data(), vote_account)
2032 {
2033 return !vote_state.votes.is_empty();
2034 }
2035 false
2036}
2037
2038pub fn should_require_vote_history_file(
2040 bank: &Bank,
2041 vote_account: &Pubkey,
2042 identity: &Pubkey,
2043) -> bool {
2044 let Some(genesis_certificate) = bank.get_alpenglow_genesis_certificate() else {
2045 return false;
2047 };
2048
2049 let Some(Ok(vote_state)) = bank
2050 .get_account(vote_account)
2051 .map(|acct| acct.deserialize_data())
2052 else {
2053 return false;
2055 };
2056
2057 let Ok(vote_state_handler) = VoteStateHandler::try_new_from_vote_state_versions(vote_state)
2058 else {
2059 return false;
2060 };
2061
2062 if vote_state_handler.node_pubkey() != identity {
2063 return false;
2066 }
2067
2068 let Some(last_voted_slot) = vote_state_handler.last_voted_slot() else {
2069 return false;
2071 };
2072 let genesis_slot = genesis_certificate.block.slot;
2073
2074 last_voted_slot > genesis_slot
2076}
2077
2078fn restore_vote_history(
2079 config: &ValidatorConfig,
2080 bank_forks: &RwLock<BankForks>,
2081 identity: &Pubkey,
2082 vote_account: &Pubkey,
2083) -> Result<VoteHistory, String> {
2084 match VoteHistory::restore(config.vote_history_storage.as_ref(), identity) {
2085 Ok(vote_history) => Ok(vote_history),
2086 Err(err) => {
2087 let should_require_vote_history = {
2088 let bank_forks = bank_forks.read().unwrap();
2089 should_require_vote_history_file(&bank_forks.working_bank(), vote_account, identity)
2090 };
2091 if config.require_vote_history && should_require_vote_history {
2092 return Err(format!(
2093 "Unable to retrieve vote history for identity {identity}. The vote account \
2094 {vote_account} has prior Alpenglow votes. If this is intentional, use \
2095 --do-not-require-vote-history: {err:?}"
2096 )
2097 .to_string());
2098 }
2099 if err.is_file_missing() && !should_require_vote_history {
2100 info!(
2101 "Ignoring expected failed vote history restore because this vote account has \
2102 not voted before"
2103 );
2104 } else {
2105 warn!("Unable to retrieve vote history: {err:?} creating default vote history...");
2106 }
2107 Ok(VoteHistory::new(*identity, 0))
2108 }
2109 }
2110}
2111
2112fn check_poh_speed(bank: &Bank, maybe_hash_samples: Option<u64>) -> Result<(), ValidatorError> {
2113 let Some(hashes_per_tick) = bank.hashes_per_tick() else {
2114 warn!("Unable to read hashes per tick from Bank, skipping PoH speed check");
2115 return Ok(());
2116 };
2117
2118 let ticks_per_slot = bank.ticks_per_slot();
2119 let hashes_per_slot = hashes_per_tick * ticks_per_slot;
2120 let hash_samples = maybe_hash_samples.unwrap_or(hashes_per_slot);
2121
2122 let hash_time = compute_hash_time(hash_samples);
2123 let my_hashes_per_second = (hash_samples as f64 / hash_time.as_secs_f64()) as u64;
2124
2125 let target_slot_duration = Duration::from_nanos(bank.ns_per_slot as u64);
2126 let target_hashes_per_second =
2127 (hashes_per_slot as f64 / target_slot_duration.as_secs_f64()) as u64;
2128
2129 info!(
2130 "PoH speed check: computed hashes per second {my_hashes_per_second}, target hashes per \
2131 second {target_hashes_per_second}"
2132 );
2133 if my_hashes_per_second < target_hashes_per_second {
2134 return Err(ValidatorError::PohTooSlow {
2135 mine: my_hashes_per_second,
2136 target: target_hashes_per_second,
2137 });
2138 }
2139
2140 Ok(())
2141}
2142
2143fn maybe_cluster_restart_with_hard_fork(config: &ValidatorConfig, root_slot: Slot) -> Option<Slot> {
2144 if let Some(wait_slot_for_supermajority) = config.wait_for_supermajority
2146 && wait_slot_for_supermajority == root_slot
2147 {
2148 return Some(wait_slot_for_supermajority);
2149 }
2150
2151 None
2152}
2153
2154fn post_process_restored_tower(
2155 restored_tower: crate::consensus::Result<Tower>,
2156 validator_identity: &Pubkey,
2157 vote_account: &Pubkey,
2158 config: &ValidatorConfig,
2159 bank_forks: &BankForks,
2160) -> Result<Tower, String> {
2161 let mut should_require_tower = config.require_tower;
2162
2163 let restored_tower = restored_tower.and_then(|tower| {
2164 let root_bank = bank_forks.root_bank();
2165 let slot_history = root_bank
2166 .get_slot_history()
2167 .expect("slot history must exist");
2168 let tower = tower.adjust_lockouts_after_replay(root_bank.slot(), &slot_history);
2170
2171 if let Some(hard_fork_restart_slot) =
2172 maybe_cluster_restart_with_hard_fork(config, root_bank.slot())
2173 {
2174 let message =
2178 format!("Hard fork is detected; discarding tower restoration result: {tower:?}");
2179 datapoint_error!("tower_error", ("error", message, String),);
2180 error!("{message}");
2181
2182 should_require_tower = false;
2185 return Err(crate::consensus::TowerError::HardFork(
2186 hard_fork_restart_slot,
2187 ));
2188 }
2189
2190 if let Some(warp_slot) = config.warp_slot {
2191 should_require_tower = false;
2194 return Err(crate::consensus::TowerError::HardFork(warp_slot));
2195 }
2196
2197 tower
2198 });
2199
2200 let restored_tower = match restored_tower {
2201 Ok(tower) => tower,
2202 Err(err) => {
2203 let voting_has_been_active =
2204 active_vote_account_exists_in_bank(&bank_forks.working_bank(), vote_account);
2205 if !err.is_file_missing() {
2206 datapoint_error!(
2207 "tower_error",
2208 ("error", format!("Unable to restore tower: {err}"), String),
2209 );
2210 }
2211 if should_require_tower && voting_has_been_active {
2212 return Err(format!(
2213 "Requested mandatory tower restore failed: {err}. And there is an existing \
2214 vote_account containing actual votes. Aborting due to possible conflicting \
2215 duplicate votes"
2216 ));
2217 }
2218 if err.is_file_missing() && !voting_has_been_active {
2219 info!(
2221 "Ignoring expected failed tower restore because this is the initial validator \
2222 start with the vote account..."
2223 );
2224 } else {
2225 error!(
2226 "Rebuilding a new tower from the latest vote account due to failed tower \
2227 restore: {err}"
2228 );
2229 }
2230
2231 Tower::new_from_bankforks(bank_forks, validator_identity, vote_account)
2232 }
2233 };
2234
2235 Ok(restored_tower)
2236}
2237
2238fn post_process_restored_vote_history(
2239 mut vote_history: VoteHistory,
2240 validator_identity: &Pubkey,
2241 config: &ValidatorConfig,
2242 bank_forks: &BankForks,
2243) -> Result<VoteHistory, String> {
2244 let mut should_require_vote_history = config.require_vote_history;
2245
2246 let restored_vote_history = {
2247 let root_bank = bank_forks.root_bank();
2248
2249 if vote_history.root() < root_bank.slot() {
2250 vote_history.set_root(root_bank.slot());
2252 }
2253
2254 if let Some(hard_fork_restart_slot) =
2255 maybe_cluster_restart_with_hard_fork(config, root_bank.slot())
2256 {
2257 let message = format!(
2261 "Hard fork is detected; discarding vote_history restoration result: \
2262 {vote_history:?}"
2263 );
2264 datapoint_error!("vote_history_error", ("error", message, String),);
2265 error!("{message}");
2266
2267 should_require_vote_history = false;
2269 Err(VoteHistoryError::HardFork(hard_fork_restart_slot))
2270 } else if let Some(warp_slot) = config.warp_slot {
2271 should_require_vote_history = false;
2273 Err(VoteHistoryError::HardFork(warp_slot))
2274 } else {
2275 Ok(vote_history)
2276 }
2277 };
2278
2279 let restored_vote_history = match restored_vote_history {
2280 Ok(vote_history) => vote_history,
2281 Err(err) => {
2282 if !err.is_file_missing() {
2283 datapoint_error!(
2284 "vote_history_error",
2285 (
2286 "error",
2287 format!("Unable to restore vote_history: {err}"),
2288 String
2289 ),
2290 );
2291 }
2292 if should_require_vote_history {
2293 return Err(format!(
2294 "Requested mandatory vote_history restore failed: {err}. Ensure that the vote \
2295 history storage file has been copied to the correct directory. Aborting"
2296 ));
2297 }
2298 error!("Rebuilding an empty vote_history from root slot due to failed restore: {err}");
2299
2300 VoteHistory::new(*validator_identity, bank_forks.root())
2301 }
2302 };
2303
2304 Ok(restored_vote_history)
2305}
2306
2307fn load_genesis(
2308 config: &ValidatorConfig,
2309 ledger_path: &Path,
2310) -> Result<GenesisConfig, ValidatorError> {
2311 let genesis_config = open_genesis_config(ledger_path, config.max_genesis_archive_unpacked_size)
2312 .map_err(ValidatorError::OpenGenesisConfig)?;
2313
2314 let leader_schedule_slot_offset = genesis_config.epoch_schedule.leader_schedule_slot_offset;
2317 let slots_per_epoch = genesis_config.epoch_schedule.slots_per_epoch;
2318 let leader_epoch_offset = leader_schedule_slot_offset.div_ceil(slots_per_epoch);
2319 assert!(leader_epoch_offset <= MAX_LEADER_SCHEDULE_EPOCH_OFFSET);
2320
2321 let genesis_hash = genesis_config.hash();
2322 info!("genesis hash: {genesis_hash}");
2323
2324 if let Some(expected_genesis_hash) = config.expected_genesis_hash
2325 && genesis_hash != expected_genesis_hash
2326 {
2327 return Err(ValidatorError::GenesisHashMismatch(
2328 genesis_hash,
2329 expected_genesis_hash,
2330 ));
2331 }
2332
2333 Ok(genesis_config)
2334}
2335
2336#[allow(clippy::type_complexity)]
2337fn load_blockstore(
2338 config: &ValidatorConfig,
2339 ledger_path: &Path,
2340 genesis_config: &GenesisConfig,
2341 exit: Arc<AtomicBool>,
2342 start_progress: &Arc<RwLock<ValidatorStartProgress>>,
2343 accounts_update_notifier: Option<AccountsUpdateNotifier>,
2344 transaction_notifier: Option<TransactionNotifierArc>,
2345 entry_notifier: Option<EntryNotifierArc>,
2346 dependency_tracker: Option<Arc<DependencyTracker>>,
2347) -> Result<
2348 (
2349 Arc<RwLock<BankForks>>,
2350 Arc<Blockstore>,
2351 Slot,
2352 Receiver<bool>,
2353 UpdateParentReceiver,
2354 LeaderScheduleCache,
2355 Option<StartingSnapshotHashes>,
2356 TransactionHistoryServices,
2357 blockstore_processor::ProcessOptions,
2358 BlockstoreRootScan,
2359 DroppedSlotsReceiver,
2360 Option<EntryNotifierService>,
2361 ),
2362 String,
2363> {
2364 info!("loading ledger from {ledger_path:?}...");
2365 *start_progress.write().unwrap() = ValidatorStartProgress::LoadingLedger;
2366
2367 let mut process_options = blockstore_processor::ProcessOptions {
2368 run_verification: config.run_verification,
2369 halt_at_slot: None,
2370 new_hard_forks: config.new_hard_forks.clone(),
2371 debug_keys: config.debug_keys.clone(),
2372 accounts_db_config: config.accounts_db_config.clone(),
2373 accounts_db_skip_shrink: config.accounts_db_skip_shrink,
2374 accounts_db_force_initial_clean: config.accounts_db_force_initial_clean,
2375 runtime_config: config.runtime_config.clone(),
2376 use_snapshot_archives_at_startup: config.use_snapshot_archives_at_startup,
2377 ..blockstore_processor::ProcessOptions::default()
2378 };
2379
2380 let (blockstore, bank_from_snapshot_opt) = thread::scope(|scope| {
2381 let load_snapshot_handle = thread::Builder::new()
2382 .name("solBnkFrkSnap".into())
2383 .spawn_scoped(scope, || {
2384 bank_forks_utils::try_load_bank_forks_from_snapshot(
2385 genesis_config,
2386 &config.account_paths,
2387 &config.snapshot_config,
2388 &process_options,
2389 accounts_update_notifier.clone(),
2390 exit.clone(),
2391 )
2392 })
2393 .expect("should spawn thread");
2394 let blockstore =
2395 Blockstore::open_with_options(ledger_path, config.blockstore_options.clone())
2396 .map_err(|err| format!("Failed to open Blockstore: {err:?}"))?;
2397 let bank_from_snapshot_result = load_snapshot_handle.join().expect("join thread");
2398
2399 Ok::<_, String>((Arc::new(blockstore), bank_from_snapshot_result.transpose()))
2400 })?;
2401
2402 let original_blockstore_root = blockstore.max_root();
2405 process_options.halt_at_slot = blockstore.highest_slot().unwrap_or(None);
2406
2407 let enable_rpc_transaction_history =
2408 config.rpc_addrs.is_some() && config.rpc_config.enable_rpc_transaction_history;
2409 let is_plugin_transaction_history_required = transaction_notifier.as_ref().is_some();
2410 let transaction_history_services =
2411 if enable_rpc_transaction_history || is_plugin_transaction_history_required {
2412 initialize_rpc_transaction_history_services(
2413 blockstore.clone(),
2414 exit.clone(),
2415 enable_rpc_transaction_history,
2416 config.rpc_config.enable_extended_tx_metadata_storage,
2417 transaction_notifier,
2418 dependency_tracker,
2419 )
2420 } else {
2421 TransactionHistoryServices::default()
2422 };
2423
2424 let entry_notifier_service = entry_notifier
2425 .map(|entry_notifier| EntryNotifierService::new(entry_notifier, exit.clone()));
2426
2427 let (bank_forks, starting_snapshot_hashes) = bank_from_snapshot_opt
2428 .unwrap_or_else(|| {
2429 bank_forks_utils::discard_previous_run_state(
2431 &config.snapshot_config.bank_snapshots_dir,
2432 &config.account_paths,
2433 );
2434 bank_forks_utils::load_bank_forks_from_genesis(
2435 genesis_config,
2436 &blockstore,
2437 config.account_paths.clone(),
2438 &process_options,
2439 transaction_history_services
2440 .transaction_status_sender
2441 .as_ref(),
2442 entry_notifier_service
2443 .as_ref()
2444 .map(|service| service.sender()),
2445 accounts_update_notifier,
2446 exit.clone(),
2447 )
2448 })
2449 .map_err(|err| err.to_string())?;
2450
2451 let mut leader_schedule_cache =
2452 LeaderScheduleCache::new_from_bank(&bank_forks.read().unwrap().root_bank());
2453 leader_schedule_cache.set_fixed_leader_schedule(config.fixed_leader_schedule.clone());
2454
2455 let pruned_banks_receiver =
2461 AccountsBackgroundService::setup_bank_drop_callback(bank_forks.clone());
2462
2463 let blockstore_root_scan = BlockstoreRootScan::new(config, blockstore.clone(), exit);
2464 let (ledger_signal_sender, ledger_signal_receiver) = bounded(MAX_REPLAY_WAKE_UP_SIGNALS);
2465 blockstore.add_new_shred_signal(ledger_signal_sender);
2466 let (update_parent_sender, update_parent_receiver) = bounded(MAX_UPDATE_PARENT_SIGNALS);
2467 blockstore.add_update_parent_signal(update_parent_sender);
2468
2469 Ok((
2470 bank_forks,
2471 blockstore,
2472 original_blockstore_root,
2473 ledger_signal_receiver,
2474 update_parent_receiver,
2475 leader_schedule_cache,
2476 starting_snapshot_hashes,
2477 transaction_history_services,
2478 process_options,
2479 blockstore_root_scan,
2480 pruned_banks_receiver,
2481 entry_notifier_service,
2482 ))
2483}
2484
2485pub struct ProcessBlockStore<'a> {
2486 id: &'a Pubkey,
2487 vote_account: &'a Pubkey,
2488 start_progress: &'a Arc<RwLock<ValidatorStartProgress>>,
2489 blockstore: &'a Blockstore,
2490 original_blockstore_root: Slot,
2491 bank_forks: &'a Arc<RwLock<BankForks>>,
2492 leader_schedule_cache: &'a LeaderScheduleCache,
2493 process_options: &'a blockstore_processor::ProcessOptions,
2494 transaction_status_sender: Option<&'a TransactionStatusSender>,
2495 entry_notification_sender: Option<&'a EntryNotifierSender>,
2496 blockstore_root_scan: Option<BlockstoreRootScan>,
2497 snapshot_controller: &'a SnapshotController,
2498 config: &'a ValidatorConfig,
2499 tower: Option<Tower>,
2500 vote_history: Option<VoteHistory>,
2501 my_shred_version: u16,
2502}
2503
2504impl<'a> ProcessBlockStore<'a> {
2505 #[allow(clippy::too_many_arguments)]
2506 fn new(
2507 id: &'a Pubkey,
2508 vote_account: &'a Pubkey,
2509 start_progress: &'a Arc<RwLock<ValidatorStartProgress>>,
2510 blockstore: &'a Blockstore,
2511 original_blockstore_root: Slot,
2512 bank_forks: &'a Arc<RwLock<BankForks>>,
2513 leader_schedule_cache: &'a LeaderScheduleCache,
2514 process_options: &'a blockstore_processor::ProcessOptions,
2515 transaction_status_sender: Option<&'a TransactionStatusSender>,
2516 entry_notification_sender: Option<&'a EntryNotifierSender>,
2517 blockstore_root_scan: BlockstoreRootScan,
2518 snapshot_controller: &'a SnapshotController,
2519 config: &'a ValidatorConfig,
2520 my_shred_version: u16,
2521 ) -> Self {
2522 Self {
2523 id,
2524 vote_account,
2525 start_progress,
2526 blockstore,
2527 original_blockstore_root,
2528 bank_forks,
2529 leader_schedule_cache,
2530 process_options,
2531 transaction_status_sender,
2532 entry_notification_sender,
2533 blockstore_root_scan: Some(blockstore_root_scan),
2534 snapshot_controller,
2535 config,
2536 tower: None,
2537 vote_history: None,
2538 my_shred_version,
2539 }
2540 }
2541
2542 pub(crate) fn process(&mut self) -> Result<(Tower, VoteHistory), String> {
2543 if let (Some(tower), Some(vote_history)) = (self.tower.as_ref(), self.vote_history.as_ref())
2544 {
2545 return Ok((tower.clone(), vote_history.clone()));
2546 }
2547
2548 let previous_start_process = *self.start_progress.read().unwrap();
2550 *self.start_progress.write().unwrap() = ValidatorStartProgress::LoadingLedger;
2551
2552 let exit = Arc::new(AtomicBool::new(false));
2553 if let Ok(Some(max_slot)) = self.blockstore.highest_slot() {
2554 let bank_forks = self.bank_forks.clone();
2555 let exit = exit.clone();
2556 let start_progress = self.start_progress.clone();
2557
2558 let _ = Builder::new()
2559 .name("solRptLdgrStat".to_string())
2560 .spawn(move || {
2561 while !exit.load(Ordering::Relaxed) {
2562 let slot = bank_forks.read().unwrap().working_bank().slot();
2563 *start_progress.write().unwrap() =
2564 ValidatorStartProgress::ProcessingLedger { slot, max_slot };
2565 thread::sleep(Duration::from_secs(2));
2566 }
2567 })
2568 .unwrap();
2569 }
2570
2571 blockstore_processor::process_blockstore_from_root(
2572 self.blockstore,
2573 self.bank_forks,
2574 self.my_shred_version,
2575 self.leader_schedule_cache,
2576 self.process_options,
2577 self.transaction_status_sender,
2578 self.entry_notification_sender,
2579 Some(self.snapshot_controller),
2580 )
2581 .map_err(|err| {
2582 exit.store(true, Ordering::Relaxed);
2583 format!("Failed to load ledger: {err:?}")
2584 })?;
2585 exit.store(true, Ordering::Relaxed);
2586
2587 if let Some(blockstore_root_scan) = self.blockstore_root_scan.take() {
2588 blockstore_root_scan.join();
2589 }
2590
2591 let tower = {
2593 let restored_tower = Tower::restore(self.config.tower_storage.as_ref(), self.id);
2594 if let Ok(tower) = &restored_tower {
2595 reconcile_blockstore_roots_with_external_source(
2597 ExternalRootSource::Tower(tower.root()),
2598 self.blockstore,
2599 &mut self.original_blockstore_root,
2600 )
2601 .map_err(|err| format!("Failed to reconcile blockstore with tower: {err:?}"))?;
2602 }
2603
2604 post_process_restored_tower(
2605 restored_tower,
2606 self.id,
2607 self.vote_account,
2608 self.config,
2609 &self.bank_forks.read().unwrap(),
2610 )?
2611 };
2612
2613 let vote_history = {
2615 let vote_history =
2616 restore_vote_history(self.config, self.bank_forks, self.id, self.vote_account)?;
2617 reconcile_blockstore_roots_with_external_source(
2619 ExternalRootSource::VoteHistory(vote_history.root()),
2620 self.blockstore,
2621 &mut self.original_blockstore_root,
2622 )
2623 .map_err(|err| format!("Failed to reconcile blockstore with vote history: {err:?}"))?;
2624
2625 post_process_restored_vote_history(
2626 vote_history,
2627 self.id,
2628 self.config,
2629 &self.bank_forks.read().unwrap(),
2630 )?
2631 };
2632
2633 if let Some(hard_fork_restart_slot) = maybe_cluster_restart_with_hard_fork(
2634 self.config,
2635 self.bank_forks.read().unwrap().root(),
2636 ) {
2637 reconcile_blockstore_roots_with_external_source(
2641 ExternalRootSource::HardFork(hard_fork_restart_slot),
2642 self.blockstore,
2643 &mut self.original_blockstore_root,
2644 )
2645 .map_err(|err| format!("Failed to reconcile blockstore with hard fork: {err:?}"))?;
2646 }
2647
2648 *self.start_progress.write().unwrap() = previous_start_process;
2649 self.tower = Some(tower.clone());
2650 self.vote_history = Some(vote_history.clone());
2651 Ok((tower, vote_history))
2652 }
2653}
2654
2655fn maybe_warp_slot(
2657 config: &ValidatorConfig,
2658 process_blockstore: &mut ProcessBlockStore,
2659 ledger_path: &Path,
2660 bank_forks: &RwLock<BankForks>,
2661 leader_schedule_cache: &LeaderScheduleCache,
2662 snapshot_controller: &SnapshotController,
2663) -> Result<(), String> {
2664 if let Some(warp_slot) = config.warp_slot {
2665 let root_bank = {
2666 let bank_forks_r = bank_forks.read().unwrap();
2667 let working_bank = bank_forks_r.working_bank();
2668 if warp_slot <= working_bank.slot() {
2669 return Err(format!(
2670 "warp slot ({}) cannot be less than the working bank slot ({})",
2671 warp_slot,
2672 working_bank.slot()
2673 ));
2674 }
2675 bank_forks_r.root_bank()
2676 };
2677
2678 info!("warping to slot {warp_slot}");
2679
2680 root_bank.squash();
2684 root_bank.force_flush_accounts_cache();
2685
2686 let warp_bank = Bank::warp_from_parent(root_bank, SlotLeader::default(), warp_slot);
2690
2691 let mut bank_forks = bank_forks.write().unwrap();
2692 bank_forks.insert(warp_bank);
2693 let warp_bank = bank_forks.get(warp_slot).unwrap();
2697 Bank::calculate_and_set_block_id_for_dcou(&warp_bank);
2698 bank_forks.set_root(warp_slot, Some(snapshot_controller), Some(warp_slot));
2699 leader_schedule_cache.set_root(&warp_bank);
2700
2701 let snapshot_config = SnapshotConfig {
2702 bank_snapshots_dir: ledger_path.to_path_buf(),
2703 ..config.snapshot_config.clone()
2704 };
2705 let full_snapshot_archive_info = match snapshot_bank_utils::bank_to_full_snapshot_archive(
2706 &snapshot_config,
2707 &warp_bank,
2708 ) {
2709 Ok(archive_info) => archive_info,
2710 Err(e) => return Err(format!("Unable to create snapshot: {e}")),
2711 };
2712 info!(
2713 "created snapshot: {}",
2714 full_snapshot_archive_info.path().display()
2715 );
2716
2717 drop(bank_forks);
2718 process_blockstore.process()?;
2721 }
2722 Ok(())
2723}
2724
2725fn should_cleanup_blockstore_incorrect_shred_versions(
2728 config: &ValidatorConfig,
2729 blockstore: &Blockstore,
2730 root_slot: Slot,
2731 hard_forks: &HardForks,
2732) -> Result<Option<Slot>, BlockstoreError> {
2733 let maybe_cluster_restart_slot = maybe_cluster_restart_with_hard_fork(config, root_slot);
2735 if maybe_cluster_restart_slot.is_some() {
2736 return Ok(Some(root_slot + 1));
2737 }
2738
2739 let Some(latest_hard_fork) = hard_forks.iter().last().map(|(slot, _)| *slot) else {
2741 return Ok(None);
2742 };
2743
2744 let Some(blockstore_max_slot) = blockstore.highest_slot()? else {
2746 return Ok(None);
2747 };
2748 let blockstore_min_slot = blockstore.lowest_slot();
2749 info!(
2750 "Blockstore contains data from slot {blockstore_min_slot} to {blockstore_max_slot}, the \
2751 latest hard fork is {latest_hard_fork}"
2752 );
2753
2754 if latest_hard_fork < blockstore_min_slot {
2755 Ok(None)
2763 } else if latest_hard_fork < blockstore_max_slot {
2764 Ok(Some(latest_hard_fork + 1))
2775 } else {
2776 Ok(None)
2784 }
2785}
2786
2787fn scan_blockstore_for_incorrect_shred_version(
2790 blockstore: &Blockstore,
2791 start_slot: Slot,
2792 expected_shred_version: u16,
2793) -> Result<Option<u16>, BlockstoreError> {
2794 const TIMEOUT: Duration = Duration::from_secs(60);
2795 let timer = Instant::now();
2796 let slot_meta_iterator = blockstore.slot_meta_iterator(start_slot)?;
2798
2799 info!(
2800 "Blockstore search for shreds with incorrect version starting from slot {start_slot}; \
2801 searching for 60s"
2802 );
2803 for (slot, _meta) in slot_meta_iterator {
2804 let shreds = blockstore.get_data_shreds_for_slot(slot, 0)?;
2805 for shred in &shreds {
2806 if shred.version() != expected_shred_version {
2807 info!(
2808 "Blockstore search found shred with incorrect version {} in slot {slot}",
2809 shred.version()
2810 );
2811 return Ok(Some(shred.version()));
2812 }
2813 }
2814 if timer.elapsed() > TIMEOUT {
2815 info!("Blockstore search did not find any shreds with incorrect version");
2816 break;
2817 }
2818 }
2819 Ok(None)
2820}
2821
2822fn cleanup_blockstore_incorrect_shred_versions(
2825 blockstore: &Blockstore,
2826 config: &ValidatorConfig,
2827 start_slot: Slot,
2828 expected_shred_version: u16,
2829) -> Result<(), BlockstoreError> {
2830 let incorrect_shred_version = scan_blockstore_for_incorrect_shred_version(
2831 blockstore,
2832 start_slot,
2833 expected_shred_version,
2834 )?;
2835 let Some(incorrect_shred_version) = incorrect_shred_version else {
2836 info!("Only shreds with the correct version were found in the blockstore");
2837 return Ok(());
2838 };
2839
2840 let end_slot = blockstore.highest_slot()?.unwrap();
2842
2843 let backup_folder = format!(
2846 "{BLOCKSTORE_DIRECTORY_ROCKS_LEVEL}_backup_{incorrect_shred_version}_{start_slot}_{end_slot}"
2847 );
2848 match Blockstore::open_with_options(
2849 &blockstore.ledger_path().join(backup_folder),
2850 config.blockstore_options.clone(),
2851 ) {
2852 Ok(backup_blockstore) => {
2853 info!("Backing up slots from {start_slot} to {end_slot}");
2854 let mut timer = Measure::start("blockstore backup");
2855
2856 const PRINT_INTERVAL: Duration = Duration::from_secs(5);
2857 let mut print_timer = Instant::now();
2858 let mut num_slots_copied = 0;
2859 let slot_meta_iterator = blockstore.slot_meta_iterator(start_slot)?;
2860 for (slot, _meta) in slot_meta_iterator {
2861 let shreds = blockstore.get_data_shreds_for_slot(slot, 0)?;
2862 let shreds = shreds.into_iter().map(Cow::Owned);
2863 let _ = backup_blockstore.insert_cow_shreds(shreds, true);
2864 num_slots_copied += 1;
2865
2866 if print_timer.elapsed() > PRINT_INTERVAL {
2867 info!("Backed up {num_slots_copied} slots thus far");
2868 print_timer = Instant::now();
2869 }
2870 }
2871
2872 timer.stop();
2873 info!("Backing up slots done. {timer}");
2874 }
2875 Err(err) => {
2876 warn!("Unable to backup shreds with incorrect shred version: {err}");
2877 }
2878 }
2879
2880 info!("Purging slots {start_slot} to {end_slot} from blockstore");
2881 let mut timer = Measure::start("blockstore purge");
2882 blockstore.purge_from_next_slots(start_slot, end_slot);
2883 blockstore.purge_slots(start_slot, end_slot, PurgeType::Exact)?;
2884 timer.stop();
2885 info!("Purging slots done. {timer}");
2886
2887 Ok(())
2888}
2889
2890fn initialize_rpc_transaction_history_services(
2891 blockstore: Arc<Blockstore>,
2892 exit: Arc<AtomicBool>,
2893 enable_rpc_transaction_history: bool,
2894 enable_extended_tx_metadata_storage: bool,
2895 transaction_notifier: Option<TransactionNotifierArc>,
2896 dependency_tracker: Option<Arc<DependencyTracker>>,
2897) -> TransactionHistoryServices {
2898 let max_complete_transaction_status_slot = Arc::new(AtomicU64::new(blockstore.max_root()));
2899 let (transaction_status_sender, transaction_status_receiver) = unbounded();
2900 let transaction_status_sender = Some(TransactionStatusSender {
2901 sender: transaction_status_sender,
2902 dependency_tracker: dependency_tracker.clone(),
2903 });
2904 let transaction_status_service = Some(TransactionStatusService::new(
2905 transaction_status_receiver,
2906 max_complete_transaction_status_slot.clone(),
2907 enable_rpc_transaction_history,
2908 transaction_notifier,
2909 blockstore,
2910 enable_extended_tx_metadata_storage,
2911 dependency_tracker,
2912 exit,
2913 ));
2914
2915 TransactionHistoryServices {
2916 transaction_status_sender,
2917 transaction_status_service,
2918 max_complete_transaction_status_slot,
2919 }
2920}
2921
2922#[derive(Error, Debug)]
2923pub enum ValidatorError {
2924 #[error("bank hash mismatch: actual={0}, expected={1}")]
2925 BankHashMismatch(Hash, Hash),
2926
2927 #[error("blockstore error: {0}")]
2928 Blockstore(#[source] BlockstoreError),
2929
2930 #[error("genesis hash mismatch: actual={0}, expected={1}")]
2931 GenesisHashMismatch(Hash, Hash),
2932
2933 #[error(
2934 "ledger does not have enough data to wait for supermajority: current slot={0}, needed \
2935 slot={1}"
2936 )]
2937 NotEnoughLedgerData(Slot, Slot),
2938
2939 #[error("failed to open genesis: {0}")]
2940 OpenGenesisConfig(#[source] OpenGenesisConfigError),
2941
2942 #[error("{0}")]
2943 Other(String),
2944
2945 #[error(
2946 "PoH hashes/second rate is slower than the cluster target: mine {mine}, cluster {target}"
2947 )]
2948 PohTooSlow { mine: u64, target: u64 },
2949
2950 #[error(transparent)]
2951 ResourceLimitError(#[from] ResourceLimitError),
2952
2953 #[error("shred version mismatch: actual {actual}, expected {expected}")]
2954 ShredVersionMismatch { actual: u16, expected: u16 },
2955
2956 #[error(transparent)]
2957 TraceError(#[from] TraceError),
2958}
2959
2960fn wait_for_supermajority(
2967 config: &ValidatorConfig,
2968 process_blockstore: Option<&mut ProcessBlockStore>,
2969 bank_forks: &RwLock<BankForks>,
2970 cluster_info: &ClusterInfo,
2971 rpc_override_health_check: Arc<AtomicBool>,
2972 start_progress: &Arc<RwLock<ValidatorStartProgress>>,
2973) -> Result<bool, ValidatorError> {
2974 match config.wait_for_supermajority {
2975 None => Ok(false),
2976 Some(wait_for_supermajority_slot) => {
2977 if let Some(process_blockstore) = process_blockstore {
2978 process_blockstore
2979 .process()
2980 .map_err(ValidatorError::Other)?;
2981 }
2982
2983 let bank = bank_forks.read().unwrap().working_bank();
2984 match wait_for_supermajority_slot.cmp(&bank.slot()) {
2985 std::cmp::Ordering::Less => return Ok(false),
2986 std::cmp::Ordering::Greater => {
2987 return Err(ValidatorError::NotEnoughLedgerData(
2988 bank.slot(),
2989 wait_for_supermajority_slot,
2990 ));
2991 }
2992 _ => {}
2993 }
2994
2995 if let Some(expected_bank_hash) = config.expected_bank_hash
2996 && bank.hash() != expected_bank_hash
2997 {
2998 return Err(ValidatorError::BankHashMismatch(
2999 bank.hash(),
3000 expected_bank_hash,
3001 ));
3002 }
3003
3004 for i in 1.. {
3005 let logging = i % 10 == 1;
3006 if logging {
3007 info!(
3008 "Waiting for {}% of activated stake at slot {} to be in gossip...",
3009 WAIT_FOR_SUPERMAJORITY_THRESHOLD_PERCENT,
3010 bank.slot()
3011 );
3012 }
3013
3014 let gossip_stake_percent =
3015 get_stake_percent_in_gossip(&bank, cluster_info, logging);
3016
3017 *start_progress.write().unwrap() =
3018 ValidatorStartProgress::WaitingForSupermajority {
3019 slot: wait_for_supermajority_slot,
3020 gossip_stake_percent,
3021 };
3022
3023 if gossip_stake_percent >= WAIT_FOR_SUPERMAJORITY_THRESHOLD_PERCENT {
3024 info!(
3025 "Supermajority reached, {gossip_stake_percent}% active stake detected, \
3026 starting up now.",
3027 );
3028 break;
3029 }
3030 rpc_override_health_check.store(true, Ordering::Relaxed);
3034 thread::sleep(Duration::new(1, 0));
3035 }
3036 rpc_override_health_check.store(false, Ordering::Relaxed);
3037 Ok(true)
3038 }
3039 }
3040}
3041
3042fn get_stake_percent_in_gossip(bank: &Bank, cluster_info: &ClusterInfo, log: bool) -> u64 {
3044 let mut online_stake = 0;
3045 let mut offline_stake = 0;
3046 let mut offline_nodes = vec![];
3047
3048 let mut total_activated_stake = 0;
3049 let now = timestamp();
3050 let peers: HashMap<_, _> = cluster_info
3054 .tvu_peers(ContactInfo::clone)
3055 .into_iter()
3056 .filter(|node| {
3057 let age = now.saturating_sub(node.wallclock());
3058 age < CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS
3060 })
3061 .map(|node| (*node.pubkey(), node))
3062 .collect();
3063 let my_id = cluster_info.id();
3064
3065 for (activated_stake, vote_account) in bank.vote_accounts().values() {
3066 let activated_stake = *activated_stake;
3067 total_activated_stake += activated_stake;
3068
3069 if activated_stake == 0 {
3070 continue;
3071 }
3072 let vote_state_node_pubkey = *vote_account.node_pubkey();
3073
3074 if peers.contains_key(&vote_state_node_pubkey) {
3075 trace!(
3076 "observed {vote_state_node_pubkey} in gossip, (activated_stake={activated_stake})"
3077 );
3078 online_stake += activated_stake;
3079 } else if vote_state_node_pubkey == my_id {
3080 online_stake += activated_stake; } else {
3082 offline_stake += activated_stake;
3083 offline_nodes.push((activated_stake, vote_state_node_pubkey));
3084 }
3085 }
3086
3087 let online_stake_percentage = (online_stake as f64 / total_activated_stake as f64) * 100.;
3088 if log {
3089 info!("{online_stake_percentage:.3}% of active stake visible in gossip");
3090
3091 if !offline_nodes.is_empty() {
3092 info!(
3093 "{:.3}% of active stake is not visible in gossip",
3094 (offline_stake as f64 / total_activated_stake as f64) * 100.
3095 );
3096 offline_nodes.sort_by_key(|a| cmp::Reverse(a.0)); for (stake, identity) in offline_nodes {
3098 info!(
3099 " {:.3}% - {}",
3100 (stake as f64 / total_activated_stake as f64) * 100.,
3101 identity
3102 );
3103 }
3104 }
3105 datapoint_info!(
3106 "wfsm_gossip",
3107 ("online_stake", online_stake, i64),
3108 ("offline_stake", offline_stake, i64),
3109 ("total_activated_stake", total_activated_stake, i64),
3110 );
3111 }
3112
3113 online_stake_percentage as u64
3114}
3115
3116fn validate_account_paths(config: &ValidatorConfig) -> std::io::Result<()> {
3117 validate_account_paths_for_direct_io(
3118 config.snapshot_config.use_direct_io,
3119 config
3120 .account_paths
3121 .iter()
3122 .chain(&config.account_snapshot_paths)
3123 .chain([
3124 &config.snapshot_config.full_snapshot_archives_dir,
3125 &config.snapshot_config.incremental_snapshot_archives_dir,
3126 &config.snapshot_config.bank_snapshots_dir,
3127 ]),
3128 )
3129}
3130
3131pub fn is_snapshot_config_valid(snapshot_config: &SnapshotConfig) -> bool {
3132 if !snapshot_config.should_generate_snapshots() {
3134 return true;
3135 }
3136
3137 let SnapshotInterval::Slots(full_snapshot_interval_slots) =
3138 snapshot_config.full_snapshot_archive_interval
3139 else {
3140 return false;
3142 };
3143
3144 match snapshot_config.incremental_snapshot_archive_interval {
3145 SnapshotInterval::Disabled => true,
3146 SnapshotInterval::Slots(incremental_snapshot_interval_slots) => {
3147 full_snapshot_interval_slots > incremental_snapshot_interval_slots
3148 }
3149 }
3150}
3151
3152#[cfg(test)]
3153mod tests {
3154 use {
3155 super::*,
3156 agave_votor_messages::certificate::{CertSignature, GenesisCert},
3157 crossbeam_channel::{RecvTimeoutError, bounded},
3158 solana_entry::entry,
3159 solana_genesis_config::create_genesis_config,
3160 solana_gossip::contact_info::ContactInfo,
3161 solana_leader_schedule::SlotLeader,
3162 solana_ledger::{
3163 blockstore, create_new_tmp_ledger, genesis_utils::create_genesis_config_with_leader,
3164 get_tmp_ledger_path_auto_delete,
3165 },
3166 solana_poh_config::PohConfig,
3167 solana_sha256_hasher::hash,
3168 solana_vote_program::vote_state::{LandedVote, Lockout, VoteStateVersions},
3169 std::{fs::remove_dir_all, num::NonZeroU64, thread, time::Duration},
3170 };
3171
3172 #[test]
3173 fn test_should_require_vote_history_file() {
3174 use {
3175 agave_votor_messages::consensus_message::Block,
3176 solana_account::{AccountSharedData, state_traits::StateMut},
3177 solana_bls_signatures::{BLS_SIGNATURE_AFFINE_SIZE, Signature as BLSSignature},
3178 };
3179
3180 let genesis_config = create_genesis_config(1_000_000).0;
3181 let bank = Bank::new_for_tests(&genesis_config);
3182 let vote_account_pubkey = Pubkey::new_unique();
3183 let identity = Pubkey::new_unique();
3184
3185 assert!(!active_vote_account_exists_in_bank(
3186 &bank,
3187 &vote_account_pubkey
3188 ));
3189 assert!(!should_require_vote_history_file(
3190 &bank,
3191 &vote_account_pubkey,
3192 &identity,
3193 ));
3194
3195 let mut vote_state = VoteStateV4 {
3196 node_pubkey: identity,
3197 ..VoteStateV4::default()
3198 };
3199 let mut vote_account =
3200 AccountSharedData::new(1, VoteStateV4::size_of(), &solana_vote_program::id());
3201 vote_account
3202 .set_state(&VoteStateVersions::new_v4(vote_state.clone()))
3203 .unwrap();
3204 bank.store_account(&vote_account_pubkey, &vote_account);
3205 assert!(!active_vote_account_exists_in_bank(
3206 &bank,
3207 &vote_account_pubkey
3208 ));
3209 assert!(!should_require_vote_history_file(
3210 &bank,
3211 &vote_account_pubkey,
3212 &identity,
3213 ));
3214
3215 vote_state.votes.push_back(LandedVote {
3216 latency: 0,
3217 lockout: Lockout::new(7),
3218 });
3219 vote_account
3220 .set_state(&VoteStateVersions::new_v4(vote_state.clone()))
3221 .unwrap();
3222 bank.store_account(&vote_account_pubkey, &vote_account);
3223 assert!(active_vote_account_exists_in_bank(
3224 &bank,
3225 &vote_account_pubkey
3226 ));
3227 assert!(!should_require_vote_history_file(
3228 &bank,
3229 &vote_account_pubkey,
3230 &identity,
3231 ));
3232
3233 let cert = GenesisCert {
3234 block: Block {
3235 slot: 40,
3236 block_id: Hash::new_unique(),
3237 },
3238 signature: CertSignature {
3239 signature: BLSSignature([0; BLS_SIGNATURE_AFFINE_SIZE]),
3240 bitmap: vec![],
3241 },
3242 };
3243 bank.set_alpenglow_genesis_certificate(&cert);
3244 assert!(!should_require_vote_history_file(
3245 &bank,
3246 &vote_account_pubkey,
3247 &identity,
3248 ));
3249
3250 vote_state.votes.push_back(LandedVote {
3251 latency: 0,
3252 lockout: Lockout::new(43),
3253 });
3254 vote_state.root_slot = Some(42);
3255 vote_account
3256 .set_state(&VoteStateVersions::new_v4(vote_state))
3257 .unwrap();
3258 bank.store_account(&vote_account_pubkey, &vote_account);
3259 assert!(should_require_vote_history_file(
3260 &bank,
3261 &vote_account_pubkey,
3262 &identity,
3263 ));
3264
3265 assert!(!should_require_vote_history_file(
3267 &bank,
3268 &vote_account_pubkey,
3269 &Pubkey::new_unique(),
3270 ));
3271 }
3272
3273 #[test]
3274 fn validator_exit() {
3275 agave_logger::setup();
3276 let leader_keypair = Keypair::new();
3277 let leader_node = Node::new_localhost_with_pubkey(&leader_keypair.pubkey());
3278
3279 let validator_keypair = Keypair::new();
3280 let validator_node = Node::new_localhost_with_pubkey(&validator_keypair.pubkey());
3281 let genesis_config =
3282 create_genesis_config_with_leader(10_000, &leader_keypair.pubkey(), 1000)
3283 .genesis_config;
3284 let (validator_ledger_path, _blockhash) = create_new_tmp_ledger!(&genesis_config);
3285
3286 let voting_keypair = Arc::new(Keypair::new());
3287 let config = ValidatorConfig {
3288 rpc_addrs: Some((
3289 validator_node.info.rpc().unwrap(),
3290 validator_node.info.rpc_pubsub().unwrap(),
3291 )),
3292 ..ValidatorConfig::default_for_test()
3293 };
3294 let start_progress = Arc::new(RwLock::new(ValidatorStartProgress::default()));
3295 let validator = Validator::new(
3296 validator_node,
3297 Arc::new(validator_keypair),
3298 &validator_ledger_path,
3299 &voting_keypair.pubkey(),
3300 Arc::new(RwLock::new(vec![voting_keypair])),
3301 vec![leader_node.info],
3302 &config,
3303 None, start_progress.clone(),
3305 SocketAddrSpace::Unspecified,
3306 ValidatorTpuConfig::new_for_tests(),
3307 Arc::new(RwLock::new(None)),
3308 None,
3309 )
3310 .expect("assume successful validator start");
3311 assert_eq!(
3312 *start_progress.read().unwrap(),
3313 ValidatorStartProgress::Running
3314 );
3315 validator.close();
3316 remove_dir_all(validator_ledger_path).unwrap();
3317 }
3318
3319 #[test]
3320 fn test_should_cleanup_blockstore_incorrect_shred_versions() {
3321 agave_logger::setup();
3322
3323 let ledger_path = get_tmp_ledger_path_auto_delete!();
3324 let blockstore = Blockstore::open(ledger_path.path()).unwrap();
3325
3326 let mut validator_config = ValidatorConfig::default_for_test();
3327 let mut hard_forks = HardForks::default();
3328 let mut root_slot;
3329
3330 root_slot = 10;
3332 validator_config.wait_for_supermajority = Some(root_slot);
3333 assert_eq!(
3334 should_cleanup_blockstore_incorrect_shred_versions(
3335 &validator_config,
3336 &blockstore,
3337 root_slot,
3338 &hard_forks
3339 )
3340 .unwrap(),
3341 Some(root_slot + 1)
3342 );
3343
3344 root_slot = 15;
3347 assert_eq!(
3348 should_cleanup_blockstore_incorrect_shred_versions(
3349 &validator_config,
3350 &blockstore,
3351 root_slot,
3352 &hard_forks
3353 )
3354 .unwrap(),
3355 None,
3356 );
3357
3358 hard_forks.register(10);
3361 assert_eq!(
3362 should_cleanup_blockstore_incorrect_shred_versions(
3363 &validator_config,
3364 &blockstore,
3365 root_slot,
3366 &hard_forks
3367 )
3368 .unwrap(),
3369 None,
3370 );
3371
3372 let entries = entry::create_ticks(1, 0, Hash::default());
3374 for i in 20..35 {
3375 let shreds = blockstore::entries_to_test_shreds(
3376 &entries,
3377 i, i - 1, true, 1, );
3382 blockstore.insert_shreds(shreds, true).unwrap();
3383 }
3384
3385 assert_eq!(
3387 should_cleanup_blockstore_incorrect_shred_versions(
3388 &validator_config,
3389 &blockstore,
3390 root_slot,
3391 &hard_forks
3392 )
3393 .unwrap(),
3394 None,
3395 );
3396
3397 root_slot = 25;
3400 hard_forks.register(root_slot);
3401 validator_config.wait_for_supermajority = Some(root_slot);
3402 assert_eq!(
3403 should_cleanup_blockstore_incorrect_shred_versions(
3404 &validator_config,
3405 &blockstore,
3406 root_slot,
3407 &hard_forks
3408 )
3409 .unwrap(),
3410 Some(root_slot + 1),
3411 );
3412 validator_config.wait_for_supermajority = None;
3413 assert_eq!(
3414 should_cleanup_blockstore_incorrect_shred_versions(
3415 &validator_config,
3416 &blockstore,
3417 root_slot,
3418 &hard_forks
3419 )
3420 .unwrap(),
3421 Some(root_slot + 1),
3422 );
3423
3424 root_slot = 30;
3427 let latest_hard_fork = hard_forks.iter().last().unwrap().0;
3428 assert_eq!(
3429 should_cleanup_blockstore_incorrect_shred_versions(
3430 &validator_config,
3431 &blockstore,
3432 root_slot,
3433 &hard_forks
3434 )
3435 .unwrap(),
3436 Some(latest_hard_fork + 1),
3437 );
3438
3439 blockstore
3442 .purge_slots(0, latest_hard_fork, PurgeType::Exact)
3443 .unwrap();
3444 assert_eq!(
3445 should_cleanup_blockstore_incorrect_shred_versions(
3446 &validator_config,
3447 &blockstore,
3448 root_slot,
3449 &hard_forks
3450 )
3451 .unwrap(),
3452 None,
3453 );
3454 }
3455
3456 #[test]
3457 fn test_cleanup_blockstore_incorrect_shred_versions() {
3458 agave_logger::setup();
3459
3460 let validator_config = ValidatorConfig::default_for_test();
3461 let ledger_path = get_tmp_ledger_path_auto_delete!();
3462 let blockstore = Blockstore::open(ledger_path.path()).unwrap();
3463
3464 let entries = entry::create_ticks(1, 0, Hash::default());
3465 for i in 1..10 {
3466 let shreds = blockstore::entries_to_test_shreds(
3467 &entries,
3468 i, i - 1, true, 1, );
3473 blockstore.insert_shreds(shreds, true).unwrap();
3474 }
3475
3476 cleanup_blockstore_incorrect_shred_versions(&blockstore, &validator_config, 5, 2).unwrap();
3478 assert!(blockstore.meta(4).unwrap().unwrap().next_slots.is_empty());
3480 for i in 5..10 {
3481 assert!(
3482 blockstore
3483 .get_data_shreds_for_slot(i, 0)
3484 .unwrap()
3485 .is_empty()
3486 );
3487 }
3488 }
3489
3490 #[test]
3491 fn validator_parallel_exit() {
3492 let leader_keypair = Keypair::new();
3493 let leader_node = Node::new_localhost_with_pubkey(&leader_keypair.pubkey());
3494 let genesis_config =
3495 create_genesis_config_with_leader(10_000, &leader_keypair.pubkey(), 1000)
3496 .genesis_config;
3497
3498 let mut ledger_paths = vec![];
3499 let mut validators: Vec<Validator> = (0..2)
3500 .map(|_| {
3501 let validator_keypair = Keypair::new();
3502 let validator_node = Node::new_localhost_with_pubkey(&validator_keypair.pubkey());
3503 let (validator_ledger_path, _blockhash) = create_new_tmp_ledger!(&genesis_config);
3504 ledger_paths.push(validator_ledger_path.clone());
3505 let vote_account_keypair = Keypair::new();
3506 let config = ValidatorConfig {
3507 rpc_addrs: Some((
3508 validator_node.info.rpc().unwrap(),
3509 validator_node.info.rpc_pubsub().unwrap(),
3510 )),
3511 ..ValidatorConfig::default_for_test()
3512 };
3513 Validator::new(
3514 validator_node,
3515 Arc::new(validator_keypair),
3516 &validator_ledger_path,
3517 &vote_account_keypair.pubkey(),
3518 Arc::new(RwLock::new(vec![Arc::new(vote_account_keypair)])),
3519 vec![leader_node.info.clone()],
3520 &config,
3521 None, Arc::new(RwLock::new(ValidatorStartProgress::default())),
3523 SocketAddrSpace::Unspecified,
3524 ValidatorTpuConfig::new_for_tests(),
3525 Arc::new(RwLock::new(None)),
3526 None,
3527 )
3528 .expect("assume successful validator start")
3529 })
3530 .collect();
3531
3532 validators.iter_mut().for_each(|v| v.exit());
3534
3535 let (sender, receiver) = bounded(0);
3537 let _ = thread::spawn(move || {
3538 validators.into_iter().for_each(|validator| {
3539 validator.join();
3540 });
3541 sender.send(()).unwrap();
3542 });
3543
3544 let timeout = Duration::from_secs(60);
3545 if let Err(RecvTimeoutError::Timeout) = receiver.recv_timeout(timeout) {
3546 panic!("timeout for shutting down validators",);
3547 }
3548
3549 for path in ledger_paths {
3550 remove_dir_all(path).unwrap();
3551 }
3552 }
3553
3554 #[test]
3555 fn test_wait_for_supermajority() {
3556 agave_logger::setup();
3557 let node_keypair = Arc::new(Keypair::new());
3558 let cluster_info = ClusterInfo::new(
3559 ContactInfo::new_localhost(&node_keypair.pubkey(), timestamp()),
3560 node_keypair,
3561 SocketAddrSpace::Unspecified,
3562 );
3563
3564 let (genesis_config, _mint_keypair) = create_genesis_config(1);
3565 let bank_forks = BankForks::new_rw_arc(Bank::new_for_tests(&genesis_config));
3566 let mut config = ValidatorConfig::default_for_test();
3567 let rpc_override_health_check = Arc::new(AtomicBool::new(false));
3568 let start_progress = Arc::new(RwLock::new(ValidatorStartProgress::default()));
3569
3570 assert!(
3571 !wait_for_supermajority(
3572 &config,
3573 None,
3574 &bank_forks,
3575 &cluster_info,
3576 rpc_override_health_check.clone(),
3577 &start_progress,
3578 )
3579 .unwrap()
3580 );
3581
3582 config.wait_for_supermajority = Some(1);
3584 assert!(matches!(
3585 wait_for_supermajority(
3586 &config,
3587 None,
3588 &bank_forks,
3589 &cluster_info,
3590 rpc_override_health_check.clone(),
3591 &start_progress,
3592 ),
3593 Err(ValidatorError::NotEnoughLedgerData(_, _)),
3594 ));
3595
3596 let bank_forks = BankForks::new_rw_arc(Bank::new_from_parent(
3598 bank_forks.read().unwrap().root_bank(),
3599 SlotLeader::default(),
3600 1,
3601 ));
3602 config.wait_for_supermajority = Some(0);
3603 assert!(
3604 !wait_for_supermajority(
3605 &config,
3606 None,
3607 &bank_forks,
3608 &cluster_info,
3609 rpc_override_health_check.clone(),
3610 &start_progress,
3611 )
3612 .unwrap()
3613 );
3614
3615 config.wait_for_supermajority = Some(1);
3617 config.expected_bank_hash = Some(hash(&[1]));
3618 assert!(matches!(
3619 wait_for_supermajority(
3620 &config,
3621 None,
3622 &bank_forks,
3623 &cluster_info,
3624 rpc_override_health_check,
3625 &start_progress,
3626 ),
3627 Err(ValidatorError::BankHashMismatch(_, _)),
3628 ));
3629 }
3630
3631 #[test]
3632 fn test_is_snapshot_config_valid() {
3633 fn new_snapshot_config(
3634 full_snapshot_archive_interval_slots: Slot,
3635 incremental_snapshot_archive_interval_slots: Slot,
3636 ) -> SnapshotConfig {
3637 SnapshotConfig {
3638 full_snapshot_archive_interval: SnapshotInterval::Slots(
3639 NonZeroU64::new(full_snapshot_archive_interval_slots).unwrap(),
3640 ),
3641 incremental_snapshot_archive_interval: SnapshotInterval::Slots(
3642 NonZeroU64::new(incremental_snapshot_archive_interval_slots).unwrap(),
3643 ),
3644 ..SnapshotConfig::default()
3645 }
3646 }
3647
3648 assert!(is_snapshot_config_valid(&SnapshotConfig::default()));
3650
3651 assert!(is_snapshot_config_valid(&SnapshotConfig {
3653 incremental_snapshot_archive_interval: SnapshotInterval::Disabled,
3654 ..SnapshotConfig::default()
3655 }));
3656
3657 assert!(!is_snapshot_config_valid(&SnapshotConfig {
3659 full_snapshot_archive_interval: SnapshotInterval::Disabled,
3660 ..SnapshotConfig::default()
3661 }));
3662
3663 assert!(is_snapshot_config_valid(&new_snapshot_config(400, 200)));
3665 assert!(is_snapshot_config_valid(&new_snapshot_config(100, 42)));
3666 assert!(is_snapshot_config_valid(&new_snapshot_config(444, 200)));
3667 assert!(is_snapshot_config_valid(&new_snapshot_config(400, 222)));
3668
3669 assert!(!is_snapshot_config_valid(&new_snapshot_config(42, 100)));
3671 assert!(!is_snapshot_config_valid(&new_snapshot_config(100, 100)));
3672 assert!(!is_snapshot_config_valid(&new_snapshot_config(100, 200)));
3673
3674 assert!(is_snapshot_config_valid(&SnapshotConfig::new_disabled()));
3676 assert!(is_snapshot_config_valid(&SnapshotConfig::new_load_only()));
3677 assert!(is_snapshot_config_valid(&SnapshotConfig {
3678 full_snapshot_archive_interval: SnapshotInterval::Slots(NonZeroU64::new(37).unwrap()),
3679 incremental_snapshot_archive_interval: SnapshotInterval::Slots(
3680 NonZeroU64::new(41).unwrap()
3681 ),
3682 ..SnapshotConfig::new_load_only()
3683 }));
3684 assert!(is_snapshot_config_valid(&SnapshotConfig {
3685 full_snapshot_archive_interval: SnapshotInterval::Disabled,
3686 incremental_snapshot_archive_interval: SnapshotInterval::Disabled,
3687 ..SnapshotConfig::new_load_only()
3688 }));
3689 }
3690
3691 fn target_tick_duration() -> Duration {
3692 let target_tick_duration_us =
3693 solana_clock::DEFAULT_MS_PER_SLOT * 1000 / solana_clock::DEFAULT_TICKS_PER_SLOT;
3694 assert_eq!(target_tick_duration_us, 6250);
3695 Duration::from_micros(target_tick_duration_us)
3696 }
3697
3698 #[test]
3699 fn test_poh_speed() {
3700 agave_logger::setup();
3701 let poh_config = PohConfig {
3702 target_tick_duration: target_tick_duration(),
3703 hashes_per_tick: Some(100 * solana_clock::DEFAULT_HASHES_PER_TICK),
3705 ..PohConfig::default()
3706 };
3707 let genesis_config = GenesisConfig {
3708 poh_config,
3709 ..GenesisConfig::default()
3710 };
3711 let bank = Bank::new_for_tests(&genesis_config);
3712 assert!(check_poh_speed(&bank, Some(10_000)).is_err());
3713 }
3714
3715 #[test]
3716 fn test_poh_speed_no_hashes_per_tick() {
3717 agave_logger::setup();
3718 let poh_config = PohConfig {
3719 target_tick_duration: target_tick_duration(),
3720 hashes_per_tick: None,
3721 ..PohConfig::default()
3722 };
3723 let genesis_config = GenesisConfig {
3724 poh_config,
3725 ..GenesisConfig::default()
3726 };
3727 let bank = Bank::new_for_tests(&genesis_config);
3728 check_poh_speed(&bank, Some(10_000)).unwrap();
3729 }
3730}