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