agave_validator/
admin_rpc_service.rs

1use {
2    crossbeam_channel::Sender,
3    jsonrpc_core::{BoxFuture, ErrorCode, MetaIoHandler, Metadata, Result},
4    jsonrpc_core_client::{transports::ipc, RpcError},
5    jsonrpc_derive::rpc,
6    jsonrpc_ipc_server::{
7        tokio::sync::oneshot::channel as oneshot_channel, RequestContext, ServerBuilder,
8    },
9    log::*,
10    serde::{de::Deserializer, Deserialize, Serialize},
11    solana_accounts_db::accounts_index::AccountIndex,
12    solana_core::{
13        admin_rpc_post_init::AdminRpcRequestMetadataPostInit,
14        consensus::{tower_storage::TowerStorage, Tower},
15        repair::repair_service,
16        validator::ValidatorStartProgress,
17    },
18    solana_geyser_plugin_manager::GeyserPluginManagerRequest,
19    solana_gossip::contact_info::{ContactInfo, Protocol, SOCKET_ADDR_UNSPECIFIED},
20    solana_keypair::{read_keypair_file, Keypair},
21    solana_pubkey::Pubkey,
22    solana_rpc::rpc::verify_pubkey,
23    solana_rpc_client_api::{config::RpcAccountIndex, custom_error::RpcCustomError},
24    solana_signer::Signer,
25    solana_validator_exit::Exit,
26    std::{
27        collections::{HashMap, HashSet},
28        env, error,
29        fmt::{self, Display},
30        net::SocketAddr,
31        path::{Path, PathBuf},
32        sync::{
33            atomic::{AtomicBool, Ordering},
34            Arc, RwLock,
35        },
36        thread::{self, Builder},
37        time::{Duration, SystemTime},
38    },
39    tokio::runtime::Runtime,
40};
41
42#[derive(Clone)]
43pub struct AdminRpcRequestMetadata {
44    pub rpc_addr: Option<SocketAddr>,
45    pub start_time: SystemTime,
46    pub start_progress: Arc<RwLock<ValidatorStartProgress>>,
47    pub validator_exit: Arc<RwLock<Exit>>,
48    pub validator_exit_backpressure: HashMap<String, Arc<AtomicBool>>,
49    pub authorized_voter_keypairs: Arc<RwLock<Vec<Arc<Keypair>>>>,
50    pub tower_storage: Arc<dyn TowerStorage>,
51    pub staked_nodes_overrides: Arc<RwLock<HashMap<Pubkey, u64>>>,
52    pub post_init: Arc<RwLock<Option<AdminRpcRequestMetadataPostInit>>>,
53    pub rpc_to_plugin_manager_sender: Option<Sender<GeyserPluginManagerRequest>>,
54}
55
56impl Metadata for AdminRpcRequestMetadata {}
57
58impl AdminRpcRequestMetadata {
59    fn with_post_init<F, R>(&self, func: F) -> Result<R>
60    where
61        F: FnOnce(&AdminRpcRequestMetadataPostInit) -> Result<R>,
62    {
63        if let Some(post_init) = self.post_init.read().unwrap().as_ref() {
64            func(post_init)
65        } else {
66            Err(jsonrpc_core::error::Error::invalid_params(
67                "Retry once validator start up is complete",
68            ))
69        }
70    }
71}
72
73#[derive(Debug, Deserialize, Serialize)]
74pub struct AdminRpcContactInfo {
75    pub id: String,
76    pub gossip: SocketAddr,
77    pub tvu: SocketAddr,
78    pub tvu_quic: SocketAddr,
79    pub serve_repair_quic: SocketAddr,
80    pub tpu: SocketAddr,
81    pub tpu_forwards: SocketAddr,
82    pub tpu_vote: SocketAddr,
83    pub rpc: SocketAddr,
84    pub rpc_pubsub: SocketAddr,
85    pub serve_repair: SocketAddr,
86    pub last_updated_timestamp: u64,
87    pub shred_version: u16,
88}
89
90#[derive(Debug, Deserialize, Serialize)]
91pub struct AdminRpcRepairWhitelist {
92    pub whitelist: Vec<Pubkey>,
93}
94
95impl From<ContactInfo> for AdminRpcContactInfo {
96    fn from(node: ContactInfo) -> Self {
97        macro_rules! unwrap_socket {
98            ($name:ident) => {
99                node.$name().unwrap_or(SOCKET_ADDR_UNSPECIFIED)
100            };
101            ($name:ident, $protocol:expr) => {
102                node.$name($protocol).unwrap_or(SOCKET_ADDR_UNSPECIFIED)
103            };
104        }
105        Self {
106            id: node.pubkey().to_string(),
107            last_updated_timestamp: node.wallclock(),
108            gossip: unwrap_socket!(gossip),
109            tvu: unwrap_socket!(tvu, Protocol::UDP),
110            tvu_quic: unwrap_socket!(tvu, Protocol::QUIC),
111            serve_repair_quic: unwrap_socket!(serve_repair, Protocol::QUIC),
112            tpu: unwrap_socket!(tpu, Protocol::UDP),
113            tpu_forwards: unwrap_socket!(tpu_forwards, Protocol::UDP),
114            tpu_vote: unwrap_socket!(tpu_vote, Protocol::UDP),
115            rpc: unwrap_socket!(rpc),
116            rpc_pubsub: unwrap_socket!(rpc_pubsub),
117            serve_repair: unwrap_socket!(serve_repair, Protocol::UDP),
118            shred_version: node.shred_version(),
119        }
120    }
121}
122
123impl Display for AdminRpcContactInfo {
124    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
125        writeln!(f, "Identity: {}", self.id)?;
126        writeln!(f, "Gossip: {}", self.gossip)?;
127        writeln!(f, "TVU: {}", self.tvu)?;
128        writeln!(f, "TVU QUIC: {}", self.tvu_quic)?;
129        writeln!(f, "TPU: {}", self.tpu)?;
130        writeln!(f, "TPU Forwards: {}", self.tpu_forwards)?;
131        writeln!(f, "TPU Votes: {}", self.tpu_vote)?;
132        writeln!(f, "RPC: {}", self.rpc)?;
133        writeln!(f, "RPC Pubsub: {}", self.rpc_pubsub)?;
134        writeln!(f, "Serve Repair: {}", self.serve_repair)?;
135        writeln!(f, "Last Updated Timestamp: {}", self.last_updated_timestamp)?;
136        writeln!(f, "Shred Version: {}", self.shred_version)
137    }
138}
139impl solana_cli_output::VerboseDisplay for AdminRpcContactInfo {}
140impl solana_cli_output::QuietDisplay for AdminRpcContactInfo {}
141
142impl Display for AdminRpcRepairWhitelist {
143    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
144        writeln!(f, "Repair whitelist: {:?}", &self.whitelist)
145    }
146}
147impl solana_cli_output::VerboseDisplay for AdminRpcRepairWhitelist {}
148impl solana_cli_output::QuietDisplay for AdminRpcRepairWhitelist {}
149
150#[rpc]
151pub trait AdminRpc {
152    type Metadata;
153
154    /// Initiates validator exit; exit is asynchronous so the validator
155    /// will almost certainly still be running when this method returns
156    #[rpc(meta, name = "exit")]
157    fn exit(&self, meta: Self::Metadata) -> Result<()>;
158
159    /// Return the process id (pid)
160    #[rpc(meta, name = "pid")]
161    fn pid(&self, meta: Self::Metadata) -> Result<u32>;
162
163    #[rpc(meta, name = "reloadPlugin")]
164    fn reload_plugin(
165        &self,
166        meta: Self::Metadata,
167        name: String,
168        config_file: String,
169    ) -> BoxFuture<Result<()>>;
170
171    #[rpc(meta, name = "unloadPlugin")]
172    fn unload_plugin(&self, meta: Self::Metadata, name: String) -> BoxFuture<Result<()>>;
173
174    #[rpc(meta, name = "loadPlugin")]
175    fn load_plugin(&self, meta: Self::Metadata, config_file: String) -> BoxFuture<Result<String>>;
176
177    #[rpc(meta, name = "listPlugins")]
178    fn list_plugins(&self, meta: Self::Metadata) -> BoxFuture<Result<Vec<String>>>;
179
180    #[rpc(meta, name = "rpcAddress")]
181    fn rpc_addr(&self, meta: Self::Metadata) -> Result<Option<SocketAddr>>;
182
183    #[rpc(name = "setLogFilter")]
184    fn set_log_filter(&self, filter: String) -> Result<()>;
185
186    #[rpc(meta, name = "startTime")]
187    fn start_time(&self, meta: Self::Metadata) -> Result<SystemTime>;
188
189    #[rpc(meta, name = "startProgress")]
190    fn start_progress(&self, meta: Self::Metadata) -> Result<ValidatorStartProgress>;
191
192    #[rpc(meta, name = "addAuthorizedVoter")]
193    fn add_authorized_voter(&self, meta: Self::Metadata, keypair_file: String) -> Result<()>;
194
195    #[rpc(meta, name = "addAuthorizedVoterFromBytes")]
196    fn add_authorized_voter_from_bytes(&self, meta: Self::Metadata, keypair: Vec<u8>)
197        -> Result<()>;
198
199    #[rpc(meta, name = "removeAllAuthorizedVoters")]
200    fn remove_all_authorized_voters(&self, meta: Self::Metadata) -> Result<()>;
201
202    #[rpc(meta, name = "setIdentity")]
203    fn set_identity(
204        &self,
205        meta: Self::Metadata,
206        keypair_file: String,
207        require_tower: bool,
208    ) -> Result<()>;
209
210    #[rpc(meta, name = "setIdentityFromBytes")]
211    fn set_identity_from_bytes(
212        &self,
213        meta: Self::Metadata,
214        identity_keypair: Vec<u8>,
215        require_tower: bool,
216    ) -> Result<()>;
217
218    #[rpc(meta, name = "setStakedNodesOverrides")]
219    fn set_staked_nodes_overrides(&self, meta: Self::Metadata, path: String) -> Result<()>;
220
221    #[rpc(meta, name = "contactInfo")]
222    fn contact_info(&self, meta: Self::Metadata) -> Result<AdminRpcContactInfo>;
223
224    #[rpc(meta, name = "repairShredFromPeer")]
225    fn repair_shred_from_peer(
226        &self,
227        meta: Self::Metadata,
228        pubkey: Option<Pubkey>,
229        slot: u64,
230        shred_index: u64,
231    ) -> Result<()>;
232
233    #[rpc(meta, name = "repairWhitelist")]
234    fn repair_whitelist(&self, meta: Self::Metadata) -> Result<AdminRpcRepairWhitelist>;
235
236    #[rpc(meta, name = "setRepairWhitelist")]
237    fn set_repair_whitelist(&self, meta: Self::Metadata, whitelist: Vec<Pubkey>) -> Result<()>;
238
239    #[rpc(meta, name = "getSecondaryIndexKeySize")]
240    fn get_secondary_index_key_size(
241        &self,
242        meta: Self::Metadata,
243        pubkey_str: String,
244    ) -> Result<HashMap<RpcAccountIndex, usize>>;
245
246    #[rpc(meta, name = "setPublicTpuAddress")]
247    fn set_public_tpu_address(
248        &self,
249        meta: Self::Metadata,
250        public_tpu_addr: SocketAddr,
251    ) -> Result<()>;
252
253    #[rpc(meta, name = "setPublicTpuForwardsAddress")]
254    fn set_public_tpu_forwards_address(
255        &self,
256        meta: Self::Metadata,
257        public_tpu_forwards_addr: SocketAddr,
258    ) -> Result<()>;
259}
260
261pub struct AdminRpcImpl;
262impl AdminRpc for AdminRpcImpl {
263    type Metadata = AdminRpcRequestMetadata;
264
265    fn exit(&self, meta: Self::Metadata) -> Result<()> {
266        debug!("exit admin rpc request received");
267
268        thread::Builder::new()
269            .name("solProcessExit".into())
270            .spawn(move || {
271                // Delay exit signal until this RPC request completes, otherwise the caller of `exit` might
272                // receive a confusing error as the validator shuts down before a response is sent back.
273                thread::sleep(Duration::from_millis(100));
274
275                info!("validator exit requested");
276                meta.validator_exit.write().unwrap().exit();
277
278                if !meta.validator_exit_backpressure.is_empty() {
279                    let service_names = meta.validator_exit_backpressure.keys();
280                    info!("Wait for these services to complete: {service_names:?}");
281                    loop {
282                        // The initial sleep is a grace period to allow the services to raise their
283                        // backpressure flags.
284                        // Subsequent sleeps are to throttle how often we check and log.
285                        thread::sleep(Duration::from_secs(1));
286
287                        let mut any_flags_raised = false;
288                        for (name, flag) in meta.validator_exit_backpressure.iter() {
289                            let is_flag_raised = flag.load(Ordering::Relaxed);
290                            if is_flag_raised {
291                                info!("{name}'s exit backpressure flag is raised");
292                                any_flags_raised = true;
293                            }
294                        }
295                        if !any_flags_raised {
296                            break;
297                        }
298                    }
299                    info!("All services have completed");
300                }
301
302                // TODO: Debug why Exit doesn't always cause the validator to fully exit
303                // (rocksdb background processing or some other stuck thread perhaps?).
304                //
305                // If the process is still alive after five seconds, exit harder
306                thread::sleep(Duration::from_secs(
307                    env::var("SOLANA_VALIDATOR_EXIT_TIMEOUT")
308                        .ok()
309                        .and_then(|x| x.parse().ok())
310                        .unwrap_or(5),
311                ));
312                warn!("validator exit timeout");
313                std::process::exit(0);
314            })
315            .unwrap();
316
317        Ok(())
318    }
319
320    fn pid(&self, _meta: Self::Metadata) -> Result<u32> {
321        Ok(std::process::id())
322    }
323
324    fn reload_plugin(
325        &self,
326        meta: Self::Metadata,
327        name: String,
328        config_file: String,
329    ) -> BoxFuture<Result<()>> {
330        Box::pin(async move {
331            // Construct channel for plugin to respond to this particular rpc request instance
332            let (response_sender, response_receiver) = oneshot_channel();
333
334            // Send request to plugin manager if there is a geyser service
335            if let Some(ref rpc_to_manager_sender) = meta.rpc_to_plugin_manager_sender {
336                rpc_to_manager_sender
337                    .send(GeyserPluginManagerRequest::ReloadPlugin {
338                        name,
339                        config_file,
340                        response_sender,
341                    })
342                    .expect("GeyerPluginService should never drop request receiver");
343            } else {
344                return Err(jsonrpc_core::Error {
345                    code: ErrorCode::InvalidRequest,
346                    message: "No geyser plugin service".to_string(),
347                    data: None,
348                });
349            }
350
351            // Await response from plugin manager
352            response_receiver
353                .await
354                .expect("GeyerPluginService's oneshot sender shouldn't drop early")
355        })
356    }
357
358    fn load_plugin(&self, meta: Self::Metadata, config_file: String) -> BoxFuture<Result<String>> {
359        Box::pin(async move {
360            // Construct channel for plugin to respond to this particular rpc request instance
361            let (response_sender, response_receiver) = oneshot_channel();
362
363            // Send request to plugin manager if there is a geyser service
364            if let Some(ref rpc_to_manager_sender) = meta.rpc_to_plugin_manager_sender {
365                rpc_to_manager_sender
366                    .send(GeyserPluginManagerRequest::LoadPlugin {
367                        config_file,
368                        response_sender,
369                    })
370                    .expect("GeyerPluginService should never drop request receiver");
371            } else {
372                return Err(jsonrpc_core::Error {
373                    code: ErrorCode::InvalidRequest,
374                    message: "No geyser plugin service".to_string(),
375                    data: None,
376                });
377            }
378
379            // Await response from plugin manager
380            response_receiver
381                .await
382                .expect("GeyerPluginService's oneshot sender shouldn't drop early")
383        })
384    }
385
386    fn unload_plugin(&self, meta: Self::Metadata, name: String) -> BoxFuture<Result<()>> {
387        Box::pin(async move {
388            // Construct channel for plugin to respond to this particular rpc request instance
389            let (response_sender, response_receiver) = oneshot_channel();
390
391            // Send request to plugin manager if there is a geyser service
392            if let Some(ref rpc_to_manager_sender) = meta.rpc_to_plugin_manager_sender {
393                rpc_to_manager_sender
394                    .send(GeyserPluginManagerRequest::UnloadPlugin {
395                        name,
396                        response_sender,
397                    })
398                    .expect("GeyerPluginService should never drop request receiver");
399            } else {
400                return Err(jsonrpc_core::Error {
401                    code: ErrorCode::InvalidRequest,
402                    message: "No geyser plugin service".to_string(),
403                    data: None,
404                });
405            }
406
407            // Await response from plugin manager
408            response_receiver
409                .await
410                .expect("GeyerPluginService's oneshot sender shouldn't drop early")
411        })
412    }
413
414    fn list_plugins(&self, meta: Self::Metadata) -> BoxFuture<Result<Vec<String>>> {
415        Box::pin(async move {
416            // Construct channel for plugin to respond to this particular rpc request instance
417            let (response_sender, response_receiver) = oneshot_channel();
418
419            // Send request to plugin manager
420            if let Some(ref rpc_to_manager_sender) = meta.rpc_to_plugin_manager_sender {
421                rpc_to_manager_sender
422                    .send(GeyserPluginManagerRequest::ListPlugins { response_sender })
423                    .expect("GeyerPluginService should never drop request receiver");
424            } else {
425                return Err(jsonrpc_core::Error {
426                    code: ErrorCode::InvalidRequest,
427                    message: "No geyser plugin service".to_string(),
428                    data: None,
429                });
430            }
431
432            // Await response from plugin manager
433            response_receiver
434                .await
435                .expect("GeyerPluginService's oneshot sender shouldn't drop early")
436        })
437    }
438
439    fn rpc_addr(&self, meta: Self::Metadata) -> Result<Option<SocketAddr>> {
440        debug!("rpc_addr admin rpc request received");
441        Ok(meta.rpc_addr)
442    }
443
444    fn set_log_filter(&self, filter: String) -> Result<()> {
445        debug!("set_log_filter admin rpc request received");
446        solana_logger::setup_with(&filter);
447        Ok(())
448    }
449
450    fn start_time(&self, meta: Self::Metadata) -> Result<SystemTime> {
451        debug!("start_time admin rpc request received");
452        Ok(meta.start_time)
453    }
454
455    fn start_progress(&self, meta: Self::Metadata) -> Result<ValidatorStartProgress> {
456        debug!("start_progress admin rpc request received");
457        Ok(*meta.start_progress.read().unwrap())
458    }
459
460    fn add_authorized_voter(&self, meta: Self::Metadata, keypair_file: String) -> Result<()> {
461        debug!("add_authorized_voter request received");
462
463        let authorized_voter = read_keypair_file(keypair_file)
464            .map_err(|err| jsonrpc_core::error::Error::invalid_params(format!("{err}")))?;
465
466        AdminRpcImpl::add_authorized_voter_keypair(meta, authorized_voter)
467    }
468
469    fn add_authorized_voter_from_bytes(
470        &self,
471        meta: Self::Metadata,
472        keypair: Vec<u8>,
473    ) -> Result<()> {
474        debug!("add_authorized_voter_from_bytes request received");
475
476        let authorized_voter = Keypair::from_bytes(&keypair).map_err(|err| {
477            jsonrpc_core::error::Error::invalid_params(format!(
478                "Failed to read authorized voter keypair from provided byte array: {err}"
479            ))
480        })?;
481
482        AdminRpcImpl::add_authorized_voter_keypair(meta, authorized_voter)
483    }
484
485    fn remove_all_authorized_voters(&self, meta: Self::Metadata) -> Result<()> {
486        debug!("remove_all_authorized_voters received");
487        meta.authorized_voter_keypairs.write().unwrap().clear();
488        Ok(())
489    }
490
491    fn set_identity(
492        &self,
493        meta: Self::Metadata,
494        keypair_file: String,
495        require_tower: bool,
496    ) -> Result<()> {
497        debug!("set_identity request received");
498
499        let identity_keypair = read_keypair_file(&keypair_file).map_err(|err| {
500            jsonrpc_core::error::Error::invalid_params(format!(
501                "Failed to read identity keypair from {keypair_file}: {err}"
502            ))
503        })?;
504
505        AdminRpcImpl::set_identity_keypair(meta, identity_keypair, require_tower)
506    }
507
508    fn set_identity_from_bytes(
509        &self,
510        meta: Self::Metadata,
511        identity_keypair: Vec<u8>,
512        require_tower: bool,
513    ) -> Result<()> {
514        debug!("set_identity_from_bytes request received");
515
516        let identity_keypair = Keypair::from_bytes(&identity_keypair).map_err(|err| {
517            jsonrpc_core::error::Error::invalid_params(format!(
518                "Failed to read identity keypair from provided byte array: {err}"
519            ))
520        })?;
521
522        AdminRpcImpl::set_identity_keypair(meta, identity_keypair, require_tower)
523    }
524
525    fn set_staked_nodes_overrides(&self, meta: Self::Metadata, path: String) -> Result<()> {
526        let loaded_config = load_staked_nodes_overrides(&path)
527            .map_err(|err| {
528                error!(
529                    "Failed to load staked nodes overrides from {}: {}",
530                    &path, err
531                );
532                jsonrpc_core::error::Error::internal_error()
533            })?
534            .staked_map_id;
535        let mut write_staked_nodes = meta.staked_nodes_overrides.write().unwrap();
536        write_staked_nodes.clear();
537        write_staked_nodes.extend(loaded_config);
538        info!("Staked nodes overrides loaded from {}", path);
539        debug!("overrides map: {:?}", write_staked_nodes);
540        Ok(())
541    }
542
543    fn contact_info(&self, meta: Self::Metadata) -> Result<AdminRpcContactInfo> {
544        meta.with_post_init(|post_init| Ok(post_init.cluster_info.my_contact_info().into()))
545    }
546
547    fn repair_shred_from_peer(
548        &self,
549        meta: Self::Metadata,
550        pubkey: Option<Pubkey>,
551        slot: u64,
552        shred_index: u64,
553    ) -> Result<()> {
554        debug!("repair_shred_from_peer request received");
555
556        meta.with_post_init(|post_init| {
557            repair_service::RepairService::request_repair_for_shred_from_peer(
558                post_init.cluster_info.clone(),
559                post_init.cluster_slots.clone(),
560                pubkey,
561                slot,
562                shred_index,
563                &post_init.repair_socket.clone(),
564                post_init.outstanding_repair_requests.clone(),
565            );
566            Ok(())
567        })
568    }
569
570    fn repair_whitelist(&self, meta: Self::Metadata) -> Result<AdminRpcRepairWhitelist> {
571        debug!("repair_whitelist request received");
572
573        meta.with_post_init(|post_init| {
574            let whitelist: Vec<_> = post_init
575                .repair_whitelist
576                .read()
577                .unwrap()
578                .iter()
579                .copied()
580                .collect();
581            Ok(AdminRpcRepairWhitelist { whitelist })
582        })
583    }
584
585    fn set_repair_whitelist(&self, meta: Self::Metadata, whitelist: Vec<Pubkey>) -> Result<()> {
586        debug!("set_repair_whitelist request received");
587
588        let whitelist: HashSet<Pubkey> = whitelist.into_iter().collect();
589        meta.with_post_init(|post_init| {
590            *post_init.repair_whitelist.write().unwrap() = whitelist;
591            warn!(
592                "Repair whitelist set to {:?}",
593                &post_init.repair_whitelist.read().unwrap()
594            );
595            Ok(())
596        })
597    }
598
599    fn get_secondary_index_key_size(
600        &self,
601        meta: Self::Metadata,
602        pubkey_str: String,
603    ) -> Result<HashMap<RpcAccountIndex, usize>> {
604        debug!(
605            "get_secondary_index_key_size rpc request received: {:?}",
606            pubkey_str
607        );
608        let index_key = verify_pubkey(&pubkey_str)?;
609        meta.with_post_init(|post_init| {
610            let bank = post_init.bank_forks.read().unwrap().root_bank();
611
612            // Take ref to enabled AccountSecondaryIndexes
613            let enabled_account_indexes = &bank.accounts().accounts_db.account_indexes;
614
615            // Exit if secondary indexes are not enabled
616            if enabled_account_indexes.is_empty() {
617                debug!("get_secondary_index_key_size: secondary index not enabled.");
618                return Ok(HashMap::new());
619            };
620
621            // Make sure the requested key is not explicitly excluded
622            if !enabled_account_indexes.include_key(&index_key) {
623                return Err(RpcCustomError::KeyExcludedFromSecondaryIndex {
624                    index_key: index_key.to_string(),
625                }
626                .into());
627            }
628
629            // Grab a ref to the AccountsDbfor this Bank
630            let accounts_index = &bank.accounts().accounts_db.accounts_index;
631
632            // Find the size of the key in every index where it exists
633            let found_sizes = enabled_account_indexes
634                .indexes
635                .iter()
636                .filter_map(|index| {
637                    accounts_index
638                        .get_index_key_size(index, &index_key)
639                        .map(|size| (rpc_account_index_from_account_index(index), size))
640                })
641                .collect::<HashMap<_, _>>();
642
643            // Note: Will return an empty HashMap if no keys are found.
644            if found_sizes.is_empty() {
645                debug!("get_secondary_index_key_size: key not found in the secondary index.");
646            }
647            Ok(found_sizes)
648        })
649    }
650
651    fn set_public_tpu_address(
652        &self,
653        meta: Self::Metadata,
654        public_tpu_addr: SocketAddr,
655    ) -> Result<()> {
656        debug!("set_public_tpu_address rpc request received: {public_tpu_addr}");
657
658        meta.with_post_init(|post_init| {
659            post_init
660                .cluster_info
661                .my_contact_info()
662                .tpu(Protocol::UDP)
663                .ok_or_else(|| {
664                    error!(
665                        "The public TPU address isn't being published. The node is likely in \
666                         repair mode. See help for --restricted-repair-only-mode for more \
667                         information."
668                    );
669                    jsonrpc_core::error::Error::internal_error()
670                })?;
671            post_init
672                .cluster_info
673                .set_tpu(public_tpu_addr)
674                .map_err(|err| {
675                    error!("Failed to set public TPU address to {public_tpu_addr}: {err}");
676                    jsonrpc_core::error::Error::internal_error()
677                })?;
678            let my_contact_info = post_init.cluster_info.my_contact_info();
679            warn!(
680                "Public TPU addresses set to {:?} (udp) and {:?} (quic)",
681                my_contact_info.tpu(Protocol::UDP),
682                my_contact_info.tpu(Protocol::QUIC),
683            );
684            Ok(())
685        })
686    }
687
688    fn set_public_tpu_forwards_address(
689        &self,
690        meta: Self::Metadata,
691        public_tpu_forwards_addr: SocketAddr,
692    ) -> Result<()> {
693        debug!("set_public_tpu_forwards_address rpc request received: {public_tpu_forwards_addr}");
694
695        meta.with_post_init(|post_init| {
696            post_init
697                .cluster_info
698                .my_contact_info()
699                .tpu_forwards(Protocol::UDP)
700                .ok_or_else(|| {
701                    error!(
702                        "The public TPU Forwards address isn't being published. The node is \
703                         likely in repair mode. See help for --restricted-repair-only-mode for \
704                         more information."
705                    );
706                    jsonrpc_core::error::Error::internal_error()
707                })?;
708            post_init
709                .cluster_info
710                .set_tpu_forwards(public_tpu_forwards_addr)
711                .map_err(|err| {
712                    error!("Failed to set public TPU address to {public_tpu_forwards_addr}: {err}");
713                    jsonrpc_core::error::Error::internal_error()
714                })?;
715            let my_contact_info = post_init.cluster_info.my_contact_info();
716            warn!(
717                "Public TPU Forwards addresses set to {:?} (udp) and {:?} (quic)",
718                my_contact_info.tpu_forwards(Protocol::UDP),
719                my_contact_info.tpu_forwards(Protocol::QUIC),
720            );
721            Ok(())
722        })
723    }
724}
725
726impl AdminRpcImpl {
727    fn add_authorized_voter_keypair(
728        meta: AdminRpcRequestMetadata,
729        authorized_voter: Keypair,
730    ) -> Result<()> {
731        let mut authorized_voter_keypairs = meta.authorized_voter_keypairs.write().unwrap();
732
733        if authorized_voter_keypairs
734            .iter()
735            .any(|x| x.pubkey() == authorized_voter.pubkey())
736        {
737            Err(jsonrpc_core::error::Error::invalid_params(
738                "Authorized voter already present",
739            ))
740        } else {
741            authorized_voter_keypairs.push(Arc::new(authorized_voter));
742            Ok(())
743        }
744    }
745
746    fn set_identity_keypair(
747        meta: AdminRpcRequestMetadata,
748        identity_keypair: Keypair,
749        require_tower: bool,
750    ) -> Result<()> {
751        meta.with_post_init(|post_init| {
752            if require_tower {
753                let _ = Tower::restore(meta.tower_storage.as_ref(), &identity_keypair.pubkey())
754                    .map_err(|err| {
755                        jsonrpc_core::error::Error::invalid_params(format!(
756                            "Unable to load tower file for identity {}: {}",
757                            identity_keypair.pubkey(),
758                            err
759                        ))
760                    })?;
761            }
762
763            for (key, notifier) in &*post_init.notifies.read().unwrap() {
764                if let Err(err) = notifier.update_key(&identity_keypair) {
765                    error!("Error updating network layer keypair: {err} on {key:?}");
766                }
767            }
768
769            solana_metrics::set_host_id(identity_keypair.pubkey().to_string());
770            post_init
771                .cluster_info
772                .set_keypair(Arc::new(identity_keypair));
773            warn!("Identity set to {}", post_init.cluster_info.id());
774            Ok(())
775        })
776    }
777}
778
779fn rpc_account_index_from_account_index(account_index: &AccountIndex) -> RpcAccountIndex {
780    match account_index {
781        AccountIndex::ProgramId => RpcAccountIndex::ProgramId,
782        AccountIndex::SplTokenOwner => RpcAccountIndex::SplTokenOwner,
783        AccountIndex::SplTokenMint => RpcAccountIndex::SplTokenMint,
784    }
785}
786
787// Start the Admin RPC interface
788pub fn run(ledger_path: &Path, metadata: AdminRpcRequestMetadata) {
789    let admin_rpc_path = admin_rpc_path(ledger_path);
790
791    let event_loop = tokio::runtime::Builder::new_multi_thread()
792        .thread_name("solAdminRpcEl")
793        .worker_threads(3) // Three still seems like a lot, and better than the default of available core count
794        .enable_all()
795        .build()
796        .unwrap();
797
798    Builder::new()
799        .name("solAdminRpc".to_string())
800        .spawn(move || {
801            let mut io = MetaIoHandler::default();
802            io.extend_with(AdminRpcImpl.to_delegate());
803
804            let validator_exit = metadata.validator_exit.clone();
805            let server = ServerBuilder::with_meta_extractor(io, move |_req: &RequestContext| {
806                metadata.clone()
807            })
808            .event_loop_executor(event_loop.handle().clone())
809            .start(&format!("{}", admin_rpc_path.display()));
810
811            match server {
812                Err(err) => {
813                    warn!("Unable to start admin rpc service: {:?}", err);
814                }
815                Ok(server) => {
816                    info!("started admin rpc service!");
817                    let close_handle = server.close_handle();
818                    validator_exit
819                        .write()
820                        .unwrap()
821                        .register_exit(Box::new(move || {
822                            close_handle.close();
823                        }));
824
825                    server.wait();
826                }
827            }
828        })
829        .unwrap();
830}
831
832fn admin_rpc_path(ledger_path: &Path) -> PathBuf {
833    #[cfg(target_family = "windows")]
834    {
835        // More information about the wackiness of pipe names over at
836        // https://docs.microsoft.com/en-us/windows/win32/ipc/pipe-names
837        if let Some(ledger_filename) = ledger_path.file_name() {
838            PathBuf::from(format!(
839                "\\\\.\\pipe\\{}-admin.rpc",
840                ledger_filename.to_string_lossy()
841            ))
842        } else {
843            PathBuf::from("\\\\.\\pipe\\admin.rpc")
844        }
845    }
846    #[cfg(not(target_family = "windows"))]
847    {
848        ledger_path.join("admin.rpc")
849    }
850}
851
852// Connect to the Admin RPC interface
853pub async fn connect(ledger_path: &Path) -> std::result::Result<gen_client::Client, RpcError> {
854    let admin_rpc_path = admin_rpc_path(ledger_path);
855    if !admin_rpc_path.exists() {
856        Err(RpcError::Client(format!(
857            "{} does not exist",
858            admin_rpc_path.display()
859        )))
860    } else {
861        ipc::connect::<_, gen_client::Client>(&format!("{}", admin_rpc_path.display())).await
862    }
863}
864
865pub fn runtime() -> Runtime {
866    tokio::runtime::Builder::new_multi_thread()
867        .thread_name("solAdminRpcRt")
868        .enable_all()
869        .build()
870        .expect("new tokio runtime")
871}
872
873#[derive(Default, Deserialize, Clone)]
874pub struct StakedNodesOverrides {
875    #[serde(deserialize_with = "deserialize_pubkey_map")]
876    pub staked_map_id: HashMap<Pubkey, u64>,
877}
878
879pub fn deserialize_pubkey_map<'de, D>(des: D) -> std::result::Result<HashMap<Pubkey, u64>, D::Error>
880where
881    D: Deserializer<'de>,
882{
883    let container: HashMap<String, u64> = serde::Deserialize::deserialize(des)?;
884    let mut container_typed: HashMap<Pubkey, u64> = HashMap::new();
885    for (key, value) in container.iter() {
886        let typed_key = Pubkey::try_from(key.as_str())
887            .map_err(|_| serde::de::Error::invalid_type(serde::de::Unexpected::Map, &"PubKey"))?;
888        container_typed.insert(typed_key, *value);
889    }
890    Ok(container_typed)
891}
892
893pub fn load_staked_nodes_overrides(
894    path: &String,
895) -> std::result::Result<StakedNodesOverrides, Box<dyn error::Error>> {
896    debug!("Loading staked nodes overrides configuration from {}", path);
897    if Path::new(&path).exists() {
898        let file = std::fs::File::open(path)?;
899        Ok(serde_yaml::from_reader(file)?)
900    } else {
901        Err(format!("Staked nodes overrides provided '{path}' a non-existing file path.").into())
902    }
903}
904
905#[cfg(test)]
906mod tests {
907    use {
908        super::*,
909        serde_json::Value,
910        solana_account::{Account, AccountSharedData},
911        solana_accounts_db::{
912            accounts_db::{AccountsDbConfig, ACCOUNTS_DB_CONFIG_FOR_TESTING},
913            accounts_index::AccountSecondaryIndexes,
914        },
915        solana_core::{
916            admin_rpc_post_init::{KeyUpdaterType, KeyUpdaters},
917            consensus::tower_storage::NullTowerStorage,
918            validator::{Validator, ValidatorConfig, ValidatorTpuConfig},
919        },
920        solana_gossip::cluster_info::{ClusterInfo, Node},
921        solana_ledger::{
922            create_new_tmp_ledger,
923            genesis_utils::{
924                create_genesis_config, create_genesis_config_with_leader, GenesisConfigInfo,
925            },
926        },
927        solana_net_utils::bind_to_unspecified,
928        solana_program_option::COption,
929        solana_program_pack::Pack,
930        solana_pubkey::Pubkey,
931        solana_rpc::rpc::create_validator_exit,
932        solana_runtime::{
933            bank::{Bank, BankTestConfig},
934            bank_forks::BankForks,
935        },
936        solana_streamer::socket::SocketAddrSpace,
937        solana_system_interface::program as system_program,
938        solana_tpu_client::tpu_client::DEFAULT_TPU_ENABLE_UDP,
939        spl_generic_token::token,
940        spl_token_2022::state::{Account as TokenAccount, AccountState as TokenAccountState, Mint},
941        std::{collections::HashSet, fs::remove_dir_all, sync::atomic::AtomicBool},
942    };
943
944    #[derive(Default)]
945    struct TestConfig {
946        account_indexes: AccountSecondaryIndexes,
947    }
948
949    struct RpcHandler {
950        io: MetaIoHandler<AdminRpcRequestMetadata>,
951        meta: AdminRpcRequestMetadata,
952        bank_forks: Arc<RwLock<BankForks>>,
953    }
954
955    impl RpcHandler {
956        fn _start() -> Self {
957            Self::start_with_config(TestConfig::default())
958        }
959
960        fn start_with_config(config: TestConfig) -> Self {
961            let keypair = Arc::new(Keypair::new());
962            let cluster_info = Arc::new(ClusterInfo::new(
963                ContactInfo::new(
964                    keypair.pubkey(),
965                    solana_time_utils::timestamp(), // wallclock
966                    0u16,                           // shred_version
967                ),
968                keypair,
969                SocketAddrSpace::Unspecified,
970            ));
971            let exit = Arc::new(AtomicBool::new(false));
972            let validator_exit = create_validator_exit(exit);
973            let (bank_forks, vote_keypair) = new_bank_forks_with_config(BankTestConfig {
974                accounts_db_config: AccountsDbConfig {
975                    account_indexes: Some(config.account_indexes),
976                    ..ACCOUNTS_DB_CONFIG_FOR_TESTING
977                },
978            });
979            let vote_account = vote_keypair.pubkey();
980            let start_progress = Arc::new(RwLock::new(ValidatorStartProgress::default()));
981            let repair_whitelist = Arc::new(RwLock::new(HashSet::new()));
982            let meta = AdminRpcRequestMetadata {
983                rpc_addr: None,
984                start_time: SystemTime::now(),
985                start_progress,
986                validator_exit,
987                validator_exit_backpressure: HashMap::default(),
988                authorized_voter_keypairs: Arc::new(RwLock::new(vec![vote_keypair])),
989                tower_storage: Arc::new(NullTowerStorage {}),
990                post_init: Arc::new(RwLock::new(Some(AdminRpcRequestMetadataPostInit {
991                    cluster_info,
992                    bank_forks: bank_forks.clone(),
993                    vote_account,
994                    repair_whitelist,
995                    notifies: Arc::new(RwLock::new(KeyUpdaters::default())),
996                    repair_socket: Arc::new(bind_to_unspecified().unwrap()),
997                    outstanding_repair_requests: Arc::<
998                        RwLock<repair_service::OutstandingShredRepairs>,
999                    >::default(),
1000                    cluster_slots: Arc::new(
1001                        solana_core::cluster_slots_service::cluster_slots::ClusterSlots::default(),
1002                    ),
1003                }))),
1004                staked_nodes_overrides: Arc::new(RwLock::new(HashMap::new())),
1005                rpc_to_plugin_manager_sender: None,
1006            };
1007            let mut io = MetaIoHandler::default();
1008            io.extend_with(AdminRpcImpl.to_delegate());
1009
1010            Self {
1011                io,
1012                meta,
1013                bank_forks,
1014            }
1015        }
1016
1017        fn root_bank(&self) -> Arc<Bank> {
1018            self.bank_forks.read().unwrap().root_bank()
1019        }
1020    }
1021
1022    fn new_bank_forks_with_config(
1023        config: BankTestConfig,
1024    ) -> (Arc<RwLock<BankForks>>, Arc<Keypair>) {
1025        let GenesisConfigInfo {
1026            genesis_config,
1027            voting_keypair,
1028            ..
1029        } = create_genesis_config(1_000_000_000);
1030
1031        let bank = Bank::new_with_config_for_tests(&genesis_config, config);
1032        (BankForks::new_rw_arc(bank), Arc::new(voting_keypair))
1033    }
1034
1035    #[test]
1036    fn test_secondary_index_key_sizes() {
1037        for secondary_index_enabled in [true, false] {
1038            let account_indexes = if secondary_index_enabled {
1039                AccountSecondaryIndexes {
1040                    keys: None,
1041                    indexes: HashSet::from([
1042                        AccountIndex::ProgramId,
1043                        AccountIndex::SplTokenMint,
1044                        AccountIndex::SplTokenOwner,
1045                    ]),
1046                }
1047            } else {
1048                AccountSecondaryIndexes::default()
1049            };
1050
1051            // RPC & Bank Setup
1052            let rpc = RpcHandler::start_with_config(TestConfig { account_indexes });
1053
1054            let bank = rpc.root_bank();
1055            let RpcHandler { io, meta, .. } = rpc;
1056
1057            // Pubkeys
1058            let token_account1_pubkey = Pubkey::new_unique();
1059            let token_account2_pubkey = Pubkey::new_unique();
1060            let token_account3_pubkey = Pubkey::new_unique();
1061            let mint1_pubkey = Pubkey::new_unique();
1062            let mint2_pubkey = Pubkey::new_unique();
1063            let wallet1_pubkey = Pubkey::new_unique();
1064            let wallet2_pubkey = Pubkey::new_unique();
1065            let non_existent_pubkey = Pubkey::new_unique();
1066            let delegate = Pubkey::new_unique();
1067
1068            let mut num_default_spl_token_program_accounts = 0;
1069            let mut num_default_system_program_accounts = 0;
1070
1071            if !secondary_index_enabled {
1072                // Test first with no accounts added & no secondary indexes enabled:
1073                let req = format!(
1074                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{token_account1_pubkey}"]}}"#,
1075                );
1076                let res = io.handle_request_sync(&req, meta.clone());
1077                let result: Value = serde_json::from_str(&res.expect("actual response"))
1078                    .expect("actual response deserialization");
1079                let sizes: HashMap<RpcAccountIndex, usize> =
1080                    serde_json::from_value(result["result"].clone()).unwrap();
1081                assert!(sizes.is_empty());
1082            } else {
1083                // Count SPL Token Program Default Accounts
1084                let req = format!(
1085                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{}"]}}"#,
1086                    token::id(),
1087                );
1088                let res = io.handle_request_sync(&req, meta.clone());
1089                let result: Value = serde_json::from_str(&res.expect("actual response"))
1090                    .expect("actual response deserialization");
1091                let sizes: HashMap<RpcAccountIndex, usize> =
1092                    serde_json::from_value(result["result"].clone()).unwrap();
1093                assert_eq!(sizes.len(), 1);
1094                num_default_spl_token_program_accounts =
1095                    *sizes.get(&RpcAccountIndex::ProgramId).unwrap();
1096                // Count System Program Default Accounts
1097                let req = format!(
1098                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{}"]}}"#,
1099                    system_program::id(),
1100                );
1101                let res = io.handle_request_sync(&req, meta.clone());
1102                let result: Value = serde_json::from_str(&res.expect("actual response"))
1103                    .expect("actual response deserialization");
1104                let sizes: HashMap<RpcAccountIndex, usize> =
1105                    serde_json::from_value(result["result"].clone()).unwrap();
1106                assert_eq!(sizes.len(), 1);
1107                num_default_system_program_accounts =
1108                    *sizes.get(&RpcAccountIndex::ProgramId).unwrap();
1109            }
1110
1111            // Add 2 basic wallet accounts
1112            let wallet1_account = AccountSharedData::from(Account {
1113                lamports: 11111111,
1114                owner: system_program::id(),
1115                ..Account::default()
1116            });
1117            bank.store_account(&wallet1_pubkey, &wallet1_account);
1118            let wallet2_account = AccountSharedData::from(Account {
1119                lamports: 11111111,
1120                owner: system_program::id(),
1121                ..Account::default()
1122            });
1123            bank.store_account(&wallet2_pubkey, &wallet2_account);
1124
1125            // Add a token account
1126            let mut account1_data = vec![0; TokenAccount::get_packed_len()];
1127            let token_account1 = TokenAccount {
1128                mint: mint1_pubkey,
1129                owner: wallet1_pubkey,
1130                delegate: COption::Some(delegate),
1131                amount: 420,
1132                state: TokenAccountState::Initialized,
1133                is_native: COption::None,
1134                delegated_amount: 30,
1135                close_authority: COption::Some(wallet1_pubkey),
1136            };
1137            TokenAccount::pack(token_account1, &mut account1_data).unwrap();
1138            let token_account1 = AccountSharedData::from(Account {
1139                lamports: 111,
1140                data: account1_data.to_vec(),
1141                owner: token::id(),
1142                ..Account::default()
1143            });
1144            bank.store_account(&token_account1_pubkey, &token_account1);
1145
1146            // Add the mint
1147            let mut mint1_data = vec![0; Mint::get_packed_len()];
1148            let mint1_state = Mint {
1149                mint_authority: COption::Some(wallet1_pubkey),
1150                supply: 500,
1151                decimals: 2,
1152                is_initialized: true,
1153                freeze_authority: COption::Some(wallet1_pubkey),
1154            };
1155            Mint::pack(mint1_state, &mut mint1_data).unwrap();
1156            let mint_account1 = AccountSharedData::from(Account {
1157                lamports: 222,
1158                data: mint1_data.to_vec(),
1159                owner: token::id(),
1160                ..Account::default()
1161            });
1162            bank.store_account(&mint1_pubkey, &mint_account1);
1163
1164            // Add another token account with the different owner, but same delegate, and mint
1165            let mut account2_data = vec![0; TokenAccount::get_packed_len()];
1166            let token_account2 = TokenAccount {
1167                mint: mint1_pubkey,
1168                owner: wallet2_pubkey,
1169                delegate: COption::Some(delegate),
1170                amount: 420,
1171                state: TokenAccountState::Initialized,
1172                is_native: COption::None,
1173                delegated_amount: 30,
1174                close_authority: COption::Some(wallet2_pubkey),
1175            };
1176            TokenAccount::pack(token_account2, &mut account2_data).unwrap();
1177            let token_account2 = AccountSharedData::from(Account {
1178                lamports: 333,
1179                data: account2_data.to_vec(),
1180                owner: token::id(),
1181                ..Account::default()
1182            });
1183            bank.store_account(&token_account2_pubkey, &token_account2);
1184
1185            // Add another token account with the same owner and delegate but different mint
1186            let mut account3_data = vec![0; TokenAccount::get_packed_len()];
1187            let token_account3 = TokenAccount {
1188                mint: mint2_pubkey,
1189                owner: wallet2_pubkey,
1190                delegate: COption::Some(delegate),
1191                amount: 42,
1192                state: TokenAccountState::Initialized,
1193                is_native: COption::None,
1194                delegated_amount: 30,
1195                close_authority: COption::Some(wallet2_pubkey),
1196            };
1197            TokenAccount::pack(token_account3, &mut account3_data).unwrap();
1198            let token_account3 = AccountSharedData::from(Account {
1199                lamports: 444,
1200                data: account3_data.to_vec(),
1201                owner: token::id(),
1202                ..Account::default()
1203            });
1204            bank.store_account(&token_account3_pubkey, &token_account3);
1205
1206            // Add the new mint
1207            let mut mint2_data = vec![0; Mint::get_packed_len()];
1208            let mint2_state = Mint {
1209                mint_authority: COption::Some(wallet2_pubkey),
1210                supply: 200,
1211                decimals: 3,
1212                is_initialized: true,
1213                freeze_authority: COption::Some(wallet2_pubkey),
1214            };
1215            Mint::pack(mint2_state, &mut mint2_data).unwrap();
1216            let mint_account2 = AccountSharedData::from(Account {
1217                lamports: 555,
1218                data: mint2_data.to_vec(),
1219                owner: token::id(),
1220                ..Account::default()
1221            });
1222            bank.store_account(&mint2_pubkey, &mint_account2);
1223
1224            // Accounts should now look like the following:
1225            //
1226            //                   -----system_program------
1227            //                  /                         \
1228            //                 /-(owns)                    \-(owns)
1229            //                /                             \
1230            //             wallet1                   ---wallet2---
1231            //               /                      /             \
1232            //              /-(SPL::owns)          /-(SPL::owns)   \-(SPL::owns)
1233            //             /                      /                 \
1234            //      token_account1         token_account2       token_account3
1235            //            \                     /                   /
1236            //             \-(SPL::mint)       /-(SPL::mint)       /-(SPL::mint)
1237            //              \                 /                   /
1238            //               --mint_account1--               mint_account2
1239
1240            if secondary_index_enabled {
1241                // ----------- Test for a non-existent key -----------
1242                let req = format!(
1243                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{non_existent_pubkey}"]}}"#,
1244                );
1245                let res = io.handle_request_sync(&req, meta.clone());
1246                let result: Value = serde_json::from_str(&res.expect("actual response"))
1247                    .expect("actual response deserialization");
1248                let sizes: HashMap<RpcAccountIndex, usize> =
1249                    serde_json::from_value(result["result"].clone()).unwrap();
1250                assert!(sizes.is_empty());
1251                // --------------- Test Queries ---------------
1252                // 1) Wallet1 - Owns 1 SPL Token
1253                let req = format!(
1254                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{wallet1_pubkey}"]}}"#,
1255                );
1256                let res = io.handle_request_sync(&req, meta.clone());
1257                let result: Value = serde_json::from_str(&res.expect("actual response"))
1258                    .expect("actual response deserialization");
1259                let sizes: HashMap<RpcAccountIndex, usize> =
1260                    serde_json::from_value(result["result"].clone()).unwrap();
1261                assert_eq!(sizes.len(), 1);
1262                assert_eq!(*sizes.get(&RpcAccountIndex::SplTokenOwner).unwrap(), 1);
1263                // 2) Wallet2 - Owns 2 SPL Tokens
1264                let req = format!(
1265                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{wallet2_pubkey}"]}}"#,
1266                );
1267                let res = io.handle_request_sync(&req, meta.clone());
1268                let result: Value = serde_json::from_str(&res.expect("actual response"))
1269                    .expect("actual response deserialization");
1270                let sizes: HashMap<RpcAccountIndex, usize> =
1271                    serde_json::from_value(result["result"].clone()).unwrap();
1272                assert_eq!(sizes.len(), 1);
1273                assert_eq!(*sizes.get(&RpcAccountIndex::SplTokenOwner).unwrap(), 2);
1274                // 3) Mint1 - Is in 2 SPL Accounts
1275                let req = format!(
1276                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{mint1_pubkey}"]}}"#,
1277                );
1278                let res = io.handle_request_sync(&req, meta.clone());
1279                let result: Value = serde_json::from_str(&res.expect("actual response"))
1280                    .expect("actual response deserialization");
1281                let sizes: HashMap<RpcAccountIndex, usize> =
1282                    serde_json::from_value(result["result"].clone()).unwrap();
1283                assert_eq!(sizes.len(), 1);
1284                assert_eq!(*sizes.get(&RpcAccountIndex::SplTokenMint).unwrap(), 2);
1285                // 4) Mint2 - Is in 1 SPL Account
1286                let req = format!(
1287                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{mint2_pubkey}"]}}"#,
1288                );
1289                let res = io.handle_request_sync(&req, meta.clone());
1290                let result: Value = serde_json::from_str(&res.expect("actual response"))
1291                    .expect("actual response deserialization");
1292                let sizes: HashMap<RpcAccountIndex, usize> =
1293                    serde_json::from_value(result["result"].clone()).unwrap();
1294                assert_eq!(sizes.len(), 1);
1295                assert_eq!(*sizes.get(&RpcAccountIndex::SplTokenMint).unwrap(), 1);
1296                // 5) SPL Token Program Owns 6 Accounts - 1 Default, 5 created above.
1297                let req = format!(
1298                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{}"]}}"#,
1299                    token::id(),
1300                );
1301                let res = io.handle_request_sync(&req, meta.clone());
1302                let result: Value = serde_json::from_str(&res.expect("actual response"))
1303                    .expect("actual response deserialization");
1304                let sizes: HashMap<RpcAccountIndex, usize> =
1305                    serde_json::from_value(result["result"].clone()).unwrap();
1306                assert_eq!(sizes.len(), 1);
1307                assert_eq!(
1308                    *sizes.get(&RpcAccountIndex::ProgramId).unwrap(),
1309                    (num_default_spl_token_program_accounts + 5)
1310                );
1311                // 5) System Program Owns 4 Accounts + 2 Default, 2 created above.
1312                let req = format!(
1313                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{}"]}}"#,
1314                    system_program::id(),
1315                );
1316                let res = io.handle_request_sync(&req, meta.clone());
1317                let result: Value = serde_json::from_str(&res.expect("actual response"))
1318                    .expect("actual response deserialization");
1319                let sizes: HashMap<RpcAccountIndex, usize> =
1320                    serde_json::from_value(result["result"].clone()).unwrap();
1321                assert_eq!(sizes.len(), 1);
1322                assert_eq!(
1323                    *sizes.get(&RpcAccountIndex::ProgramId).unwrap(),
1324                    (num_default_system_program_accounts + 2)
1325                );
1326            } else {
1327                // ------------ Secondary Indexes Disabled ------------
1328                let req = format!(
1329                    r#"{{"jsonrpc":"2.0","id":1,"method":"getSecondaryIndexKeySize","params":["{token_account2_pubkey}"]}}"#,
1330                );
1331                let res = io.handle_request_sync(&req, meta.clone());
1332                let result: Value = serde_json::from_str(&res.expect("actual response"))
1333                    .expect("actual response deserialization");
1334                let sizes: HashMap<RpcAccountIndex, usize> =
1335                    serde_json::from_value(result["result"].clone()).unwrap();
1336                assert!(sizes.is_empty());
1337            }
1338        }
1339    }
1340
1341    // This test checks that the rpc call to `set_identity` works a expected with
1342    // Bank but without validator.
1343    #[test]
1344    fn test_set_identity() {
1345        let rpc = RpcHandler::start_with_config(TestConfig::default());
1346
1347        let RpcHandler { io, meta, .. } = rpc;
1348
1349        let expected_validator_id = Keypair::new();
1350        let validator_id_bytes = format!("{:?}", expected_validator_id.to_bytes());
1351
1352        let set_id_request = format!(
1353            r#"{{"jsonrpc":"2.0","id":1,"method":"setIdentityFromBytes","params":[{validator_id_bytes}, false]}}"#,
1354        );
1355        let response = io.handle_request_sync(&set_id_request, meta.clone());
1356        let actual_parsed_response: Value =
1357            serde_json::from_str(&response.expect("actual response"))
1358                .expect("actual response deserialization");
1359
1360        let expected_parsed_response: Value = serde_json::from_str(
1361            r#"{
1362                "id": 1,
1363                "jsonrpc": "2.0",
1364                "result": null
1365            }"#,
1366        )
1367        .expect("Failed to parse expected response");
1368        assert_eq!(actual_parsed_response, expected_parsed_response);
1369
1370        let contact_info_request =
1371            r#"{"jsonrpc":"2.0","id":1,"method":"contactInfo","params":[]}"#.to_string();
1372        let response = io.handle_request_sync(&contact_info_request, meta.clone());
1373        let parsed_response: Value = serde_json::from_str(&response.expect("actual response"))
1374            .expect("actual response deserialization");
1375        let actual_validator_id = parsed_response["result"]["id"]
1376            .as_str()
1377            .expect("Expected a string");
1378        assert_eq!(
1379            actual_validator_id,
1380            expected_validator_id.pubkey().to_string()
1381        );
1382    }
1383
1384    struct TestValidatorWithAdminRpc {
1385        meta: AdminRpcRequestMetadata,
1386        io: MetaIoHandler<AdminRpcRequestMetadata>,
1387        validator_ledger_path: PathBuf,
1388    }
1389
1390    impl TestValidatorWithAdminRpc {
1391        fn new() -> Self {
1392            let leader_keypair = Keypair::new();
1393            let leader_node = Node::new_localhost_with_pubkey(&leader_keypair.pubkey());
1394
1395            let validator_keypair = Keypair::new();
1396            let validator_node = Node::new_localhost_with_pubkey(&validator_keypair.pubkey());
1397            let genesis_config =
1398                create_genesis_config_with_leader(10_000, &leader_keypair.pubkey(), 1000)
1399                    .genesis_config;
1400            let (validator_ledger_path, _blockhash) = create_new_tmp_ledger!(&genesis_config);
1401
1402            let voting_keypair = Arc::new(Keypair::new());
1403            let voting_pubkey = voting_keypair.pubkey();
1404            let authorized_voter_keypairs = Arc::new(RwLock::new(vec![voting_keypair]));
1405            let validator_config = ValidatorConfig {
1406                rpc_addrs: Some((
1407                    validator_node.info.rpc().unwrap(),
1408                    validator_node.info.rpc_pubsub().unwrap(),
1409                )),
1410                ..ValidatorConfig::default_for_test()
1411            };
1412            let start_progress = Arc::new(RwLock::new(ValidatorStartProgress::default()));
1413
1414            let post_init = Arc::new(RwLock::new(None));
1415            let meta = AdminRpcRequestMetadata {
1416                rpc_addr: validator_config.rpc_addrs.map(|(rpc_addr, _)| rpc_addr),
1417                start_time: SystemTime::now(),
1418                start_progress: start_progress.clone(),
1419                validator_exit: validator_config.validator_exit.clone(),
1420                validator_exit_backpressure: HashMap::default(),
1421                authorized_voter_keypairs: authorized_voter_keypairs.clone(),
1422                tower_storage: Arc::new(NullTowerStorage {}),
1423                post_init: post_init.clone(),
1424                staked_nodes_overrides: Arc::new(RwLock::new(HashMap::new())),
1425                rpc_to_plugin_manager_sender: None,
1426            };
1427
1428            let _validator = Validator::new(
1429                validator_node,
1430                Arc::new(validator_keypair),
1431                &validator_ledger_path,
1432                &voting_pubkey,
1433                authorized_voter_keypairs,
1434                vec![leader_node.info],
1435                &validator_config,
1436                true, // should_check_duplicate_instance
1437                None, // rpc_to_plugin_manager_receiver
1438                start_progress.clone(),
1439                SocketAddrSpace::Unspecified,
1440                ValidatorTpuConfig::new_for_tests(DEFAULT_TPU_ENABLE_UDP),
1441                post_init.clone(),
1442            )
1443            .expect("assume successful validator start");
1444            assert_eq!(
1445                *start_progress.read().unwrap(),
1446                ValidatorStartProgress::Running
1447            );
1448            let post_init = post_init.read().unwrap();
1449
1450            assert!(post_init.is_some());
1451            let post_init = post_init.as_ref().unwrap();
1452            let notifies = post_init.notifies.read().unwrap();
1453            let updater_keys: HashSet<KeyUpdaterType> =
1454                notifies.into_iter().map(|(key, _)| key.clone()).collect();
1455            assert_eq!(
1456                updater_keys,
1457                HashSet::from_iter(vec![
1458                    KeyUpdaterType::Tpu,
1459                    KeyUpdaterType::TpuForwards,
1460                    KeyUpdaterType::TpuVote,
1461                    KeyUpdaterType::Forward,
1462                    KeyUpdaterType::RpcService
1463                ])
1464            );
1465            let mut io = MetaIoHandler::default();
1466            io.extend_with(AdminRpcImpl.to_delegate());
1467            Self {
1468                meta,
1469                io,
1470                validator_ledger_path,
1471            }
1472        }
1473
1474        fn handle_request(&self, request: &str) -> Option<String> {
1475            self.io.handle_request_sync(request, self.meta.clone())
1476        }
1477    }
1478
1479    impl Drop for TestValidatorWithAdminRpc {
1480        fn drop(&mut self) {
1481            remove_dir_all(self.validator_ledger_path.clone()).unwrap();
1482        }
1483    }
1484
1485    // This test checks that `set_identity` call works with working validator and client.
1486    #[test]
1487    fn test_set_identity_with_validator() {
1488        let test_validator = TestValidatorWithAdminRpc::new();
1489        let expected_validator_id = Keypair::new();
1490        let validator_id_bytes = format!("{:?}", expected_validator_id.to_bytes());
1491
1492        let set_id_request = format!(
1493            r#"{{"jsonrpc":"2.0","id":1,"method":"setIdentityFromBytes","params":[{validator_id_bytes}, false]}}"#,
1494        );
1495        let response = test_validator.handle_request(&set_id_request);
1496        let actual_parsed_response: Value =
1497            serde_json::from_str(&response.expect("actual response"))
1498                .expect("actual response deserialization");
1499
1500        let expected_parsed_response: Value = serde_json::from_str(
1501            r#"{
1502                "id": 1,
1503                "jsonrpc": "2.0",
1504                "result": null
1505            }"#,
1506        )
1507        .expect("Failed to parse expected response");
1508        assert_eq!(actual_parsed_response, expected_parsed_response);
1509
1510        let contact_info_request =
1511            r#"{"jsonrpc":"2.0","id":1,"method":"contactInfo","params":[]}"#.to_string();
1512        let response = test_validator.handle_request(&contact_info_request);
1513        let parsed_response: Value = serde_json::from_str(&response.expect("actual response"))
1514            .expect("actual response deserialization");
1515        let actual_validator_id = parsed_response["result"]["id"]
1516            .as_str()
1517            .expect("Expected a string");
1518        assert_eq!(
1519            actual_validator_id,
1520            expected_validator_id.pubkey().to_string()
1521        );
1522
1523        let contact_info_request =
1524            r#"{"jsonrpc":"2.0","id":1,"method":"exit","params":[]}"#.to_string();
1525        let exit_response = test_validator.handle_request(&contact_info_request);
1526        let actual_parsed_response: Value =
1527            serde_json::from_str(&exit_response.expect("actual response"))
1528                .expect("actual response deserialization");
1529        assert_eq!(actual_parsed_response, expected_parsed_response);
1530    }
1531}
agave_validator/admin_rpc_service.rs

agave_validator/
admin_rpc_service.rs
