Merge pull request #10 from step-finance/reload-config-with-plugin

SF-5148: Reload Config With Plugin

core/src/validator.rs

@@ -512,6 +512,7 @@ impl Validator {
         let mut bank_notification_senders = Vec::new();
 
         let exit = Arc::new(AtomicBool::new(false));
+        let inclusions_copy = config.runtime_config.program_datum_inclusions.clone();
 
         let geyser_plugin_service =
             if let Some(geyser_plugin_config_files) = &config.on_start_geyser_plugin_config_files {
@@ -524,6 +525,7 @@ impl Validator {
                         confirmed_bank_receiver,
                         geyser_plugin_config_files,
                         rpc_to_plugin_manager_receiver_and_exit,
+                        inclusions_copy,
                     )
                     .map_err(|err| format!("Failed to load the Geyser plugin: {err:?}"))?,
                 )

geyser-plugin-manager/src/geyser_plugin_manager.rs

@@ -4,9 +4,11 @@ use {
     jsonrpc_server_utils::tokio::sync::oneshot::Sender as OneShotSender,
     libloading::Library,
     log::*,
+    solana_runtime::program_inclusions::{load_datum_program_inclusions, ProgramDatumInclusions},
     std::{
         ops::{Deref, DerefMut},
         path::Path,
+        sync::{Arc, RwLock},
     },
 };
 
@@ -47,13 +49,15 @@ impl DerefMut for LoadedGeyserPlugin {
 pub struct GeyserPluginManager {
     pub plugins: Vec<LoadedGeyserPlugin>,
     libs: Vec<Library>,
+    inclusions: Arc<RwLock<ProgramDatumInclusions>>,
 }
 
 impl GeyserPluginManager {
-    pub fn new() -> Self {
+    pub fn new(inclusions: Arc<RwLock<ProgramDatumInclusions>>) -> Self {
         GeyserPluginManager {
             plugins: Vec::default(),
             libs: Vec::default(),
+            inclusions,
         }
     }
 
@@ -239,6 +243,13 @@ impl GeyserPluginManager {
             Ok(()) => {
                 self.plugins.push(new_plugin);
                 self.libs.push(new_lib);
+
+                // Reload datum inclusions
+                log::info!("Reloading datum inclusions");
+                let mut inclusions_write_lock = self.inclusions.write().unwrap();
+                *inclusions_write_lock =
+                    load_datum_program_inclusions(&Some(vec![config_file.into()]));
+                log::info!("Reloaded datum inclusions");
             }
 
             // On failure, return error
@@ -483,7 +494,9 @@ mod tests {
     #[test]
     fn test_geyser_reload() {
         // Initialize empty manager
-        let plugin_manager = Arc::new(RwLock::new(GeyserPluginManager::new()));
+        let plugin_manager = Arc::new(RwLock::new(GeyserPluginManager::new(Arc::new(
+            RwLock::new(Default::default()),
+        ))));
 
         // No plugins are loaded, this should fail
         let mut plugin_manager_lock = plugin_manager.write().unwrap();
@@ -524,7 +537,9 @@ mod tests {
     #[test]
     fn test_plugin_list() {
         // Initialize empty manager
-        let plugin_manager = Arc::new(RwLock::new(GeyserPluginManager::new()));
+        let plugin_manager = Arc::new(RwLock::new(GeyserPluginManager::new(Arc::new(
+            RwLock::new(Default::default()),
+        ))));
         let mut plugin_manager_lock = plugin_manager.write().unwrap();
 
         // Load two plugins
@@ -548,7 +563,9 @@ mod tests {
     #[test]
     fn test_plugin_load_unload() {
         // Initialize empty manager
-        let plugin_manager = Arc::new(RwLock::new(GeyserPluginManager::new()));
+        let plugin_manager = Arc::new(RwLock::new(GeyserPluginManager::new(Arc::new(
+            RwLock::new(Default::default()),
+        ))));
         let mut plugin_manager_lock = plugin_manager.write().unwrap();
 
         // Load rpc call

geyser-plugin-manager/src/geyser_plugin_service.rs

@@ -17,6 +17,7 @@ use {
         optimistically_confirmed_bank_tracker::SlotNotification,
         transaction_notifier_interface::TransactionNotifierArc,
     },
+    solana_runtime::program_inclusions::ProgramDatumInclusions,
     std::{
         path::{Path, PathBuf},
         sync::{
@@ -55,8 +56,14 @@ impl GeyserPluginService {
     pub fn new(
         confirmed_bank_receiver: Receiver<SlotNotification>,
         geyser_plugin_config_files: &[PathBuf],
+        inclusions: Arc<RwLock<ProgramDatumInclusions>>,
     ) -> Result<Self, GeyserPluginServiceError> {
-        Self::new_with_receiver(confirmed_bank_receiver, geyser_plugin_config_files, None)
+        Self::new_with_receiver(
+            confirmed_bank_receiver,
+            geyser_plugin_config_files,
+            None,
+            inclusions,
+        )
     }
 
     pub fn new_with_receiver(
@@ -66,12 +73,13 @@ impl GeyserPluginService {
             Receiver<GeyserPluginManagerRequest>,
             Arc<AtomicBool>,
         )>,
+        inclusions: Arc<RwLock<ProgramDatumInclusions>>,
     ) -> Result<Self, GeyserPluginServiceError> {
         info!(
             "Starting GeyserPluginService from config files: {:?}",
             geyser_plugin_config_files
         );
-        let mut plugin_manager = GeyserPluginManager::new();
+        let mut plugin_manager = GeyserPluginManager::new(inclusions);
 
         for geyser_plugin_config_file in geyser_plugin_config_files {
             Self::load_plugin(&mut plugin_manager, geyser_plugin_config_file)?;

ledger-tool/src/ledger_utils.rs

@@ -259,9 +259,16 @@ pub fn load_and_process_ledger(
 
         let (confirmed_bank_sender, confirmed_bank_receiver) = unbounded();
         drop(confirmed_bank_sender);
-        let geyser_service =
-            GeyserPluginService::new(confirmed_bank_receiver, &geyser_config_files)
-                .map_err(LoadAndProcessLedgerError::GeyserServiceSetup)?;
+        let inclusions_copy = process_options
+            .runtime_config
+            .program_datum_inclusions
+            .clone();
+        let geyser_service = GeyserPluginService::new(
+            confirmed_bank_receiver,
+            &geyser_config_files,
+            inclusions_copy,
+        )
+        .map_err(LoadAndProcessLedgerError::GeyserServiceSetup)?;
         (
             geyser_service.get_accounts_update_notifier(),
             geyser_service.get_transaction_notifier(),

runtime/src/bank.rs

@@ -857,7 +857,7 @@ pub struct Bank {
 
     epoch_reward_status: EpochRewardStatus,
     /// programs that will have their account data sent to geyser
-    pub program_datum_inclusions: Arc<ProgramDatumInclusions>,
+    pub program_datum_inclusions: Arc<RwLock<ProgramDatumInclusions>>,
 }
 
 struct VoteWithStakeDelegations {
@@ -1045,7 +1045,7 @@ impl Bank {
             ))),
             check_program_modification_slot: false,
             epoch_reward_status: EpochRewardStatus::default(),
-            program_datum_inclusions: Arc::<ProgramDatumInclusions>::default(),
+            program_datum_inclusions: Arc::new(RwLock::new(ProgramDatumInclusions::default())),
         };
 
         let accounts_data_size_initial = bank.get_total_accounts_stats().unwrap().data_len as u64;
@@ -6511,7 +6511,8 @@ impl Bank {
     ) -> Option<Vec<u8>> {
         let data = account.data();
         let owner = account.owner();
-        let inclusion = self.program_datum_inclusions.get(owner)?;
+        let inclusions_lock = self.program_datum_inclusions.read().unwrap();
+        let inclusion = inclusions_lock.get(owner)?;
         let include_data = inclusion.can_include_datum(pre_or_post, &data);
 
         if !include_data {

runtime/src/runtime_config.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 
 use solana_program_runtime::compute_budget::ComputeBudget;
 
@@ -10,5 +10,5 @@ pub struct RuntimeConfig {
     pub compute_budget: Option<ComputeBudget>,
     pub log_messages_bytes_limit: Option<usize>,
     pub transaction_account_lock_limit: Option<usize>,
-    pub program_datum_inclusions: Arc<ProgramDatumInclusions>,
+    pub program_datum_inclusions: Arc<RwLock<ProgramDatumInclusions>>,
 }

test-validator/src/lib.rs

@@ -180,7 +180,9 @@ impl Default for TestValidatorGenesis {
             log_messages_bytes_limit: Option::<usize>::default(),
             transaction_account_lock_limit: Option::<usize>::default(),
             tpu_enable_udp: DEFAULT_TPU_ENABLE_UDP,
-            geyser_plugin_manager: Arc::new(RwLock::new(GeyserPluginManager::new())),
+            geyser_plugin_manager: Arc::new(RwLock::new(GeyserPluginManager::new(Arc::new(
+                RwLock::new(Default::default()),
+            )))),
             admin_rpc_service_post_init:
                 Arc::<RwLock<Option<AdminRpcRequestMetadataPostInit>>>::default(),
         }
@@ -985,9 +987,9 @@ impl TestValidator {
             ..AccountsDbConfig::default()
         });
 
-        let program_datum_inclusions = Arc::new(load_datum_program_inclusions(
+        let program_datum_inclusions = Arc::new(RwLock::new(load_datum_program_inclusions(
             &config.geyser_plugin_config_files,
-        ));
+        )));
 
         let runtime_config = RuntimeConfig {
             compute_budget: config

validator/src/main.rs

@@ -1239,9 +1239,9 @@ pub fn main() {
     };
     let starting_with_geyser_plugins: bool = on_start_geyser_plugin_config_files.is_some();
 
-    let program_datum_inclusions = Arc::new(load_datum_program_inclusions(
+    let program_datum_inclusions = Arc::new(RwLock::new(load_datum_program_inclusions(
         &on_start_geyser_plugin_config_files,
-    ));
+    )));
 
     let rpc_bigtable_config = if matches.is_present("enable_rpc_bigtable_ledger_storage")
         || matches.is_present("enable_bigtable_ledger_upload")