Add blacklisting to plugin for failish thread that should not be exec…

…uted.
AdrenaFoundation · Oct 20, 2024 · 2b1bdaf · 2b1bdaf
1 parent 6a0c081
commit 2b1bdaf
Show file tree

Hide file tree

Showing 9 changed files with 123 additions and 8 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/admin.json b/admin.json
@@ -0,0 +1,66 @@
+[
+    88,
+    7,
+    211,
+    24,
+    0,
+    182,
+    164,
+    24,
+    144,
+    100,
+    136,
+    47,
+    112,
+    99,
+    119,
+    67,
+    23,
+    78,
+    12,
+    142,
+    32,
+    219,
+    70,
+    41,
+    47,
+    14,
+    120,
+    213,
+    253,
+    213,
+    13,
+    74,
+    61,
+    211,
+    162,
+    203,
+    106,
+    68,
+    49,
+    126,
+    252,
+    196,
+    151,
+    32,
+    19,
+    245,
+    55,
+    157,
+    142,
+    197,
+    224,
+    209,
+    3,
+    52,
+    251,
+    74,
+    145,
+    156,
+    102,
+    104,
+    2,
+    152,
+    169,
+    116
+]
diff --git a/plugin/Cargo.toml b/plugin/Cargo.toml
@@ -37,6 +37,7 @@ solana-sdk.workspace = true
 tokio.workspace = true
 futures.workspace = true
 thiserror.workspace = true
+lazy_static = "1.4.0"
 
 [build-dependencies]
 cargo_metadata.workspace = true

diff --git a/plugin/src/builders/thread_exec.rs b/plugin/src/builders/thread_exec.rs
@@ -1,4 +1,5 @@
-use std::sync::Arc;
+use std::collections::HashSet;
+use std::sync::{Arc, Mutex, RwLock};
 
 use anchor_lang::{InstructionData, ToAccountMetas};
 use log::info;
@@ -33,14 +34,17 @@ static TRANSACTION_COMPUTE_UNIT_LIMIT: u32 = 1_400_000;
 /// The buffer amount to add to transactions' compute units in case on-chain PDA derivations take more CUs than used in simulation.
 static TRANSACTION_COMPUTE_UNIT_BUFFER: u32 = 1000;
 
+pub const INSTRUCTION_ERROR_REENTRANCY_NOT_ALLOWED: i64 = 4615034;
+pub const INSTRUCTION_ERROR_ANCHOR_ACCOUNT_OWNED_BY_WRONG_PROGRAM: i64 = 3007;
+
 pub async fn build_thread_exec_tx(
     client: Arc<RpcClient>,
     payer: &Keypair,
     slot: u64,
     thread: VersionedThread,
     thread_pubkey: Pubkey,
     worker_id: u64,
-) -> Result<Option<Transaction>, PluginError> {
+) -> Result<(Option<Transaction>, /* Blacklisted */ Option<Pubkey>), PluginError> {
     // Grab the thread and relevant data.
     let now = std::time::Instant::now();
     let blockhash = client.get_latest_blockhash().await?;
@@ -102,10 +106,25 @@ pub async fn build_thread_exec_tx(
         {
             // If there was a simulation error, stop packing and exit now.
             Err(err) => {
+                // Check for specific error and blacklist the thread
                 if let solana_client::client_error::ClientErrorKind::RpcError(
                     solana_client::rpc_request::RpcError::RpcResponseError { code, .. },
                 ) = err.kind
                 {
+                    if code == INSTRUCTION_ERROR_REENTRANCY_NOT_ALLOWED {
+                        info!(
+                            "INSTRUCTION_ERROR_REENTRANCY_NOT_ALLOWED, blacklisting thread: {}",
+                            thread_pubkey
+                        );
+                        return Ok((None, Some(thread_pubkey)));
+                    }
+                    if code == INSTRUCTION_ERROR_ANCHOR_ACCOUNT_OWNED_BY_WRONG_PROGRAM {
+                        info!(
+                            "INSTRUCTION_ERROR_ANCHOR_ACCOUNT_OWNED_BY_WRONG_PROGRAM, blacklisting thread: {}",
+                            thread_pubkey
+                        );
+                        return Ok((None, Some(thread_pubkey)));
+                    }
                     if code == JSON_RPC_SERVER_ERROR_MIN_CONTEXT_SLOT_NOT_REACHED {
                         return Err(PluginError::MinContextSlotNotReached);
                     }
@@ -172,7 +191,7 @@ pub async fn build_thread_exec_tx(
     // If there were no successful instructions, then exit early. There is nothing to do.
     // Alternatively, exit early if only the kickoff instruction (and no execs) succeeded.
     if successful_ixs.is_empty() {
-        return Ok(None);
+        return Ok((None, None));
     }
 
     // Set the transaction's compute unit limit to be exactly the amount that was used in simulation.
@@ -199,7 +218,7 @@ pub async fn build_thread_exec_tx(
         units_consumed,
         tx.signatures[0]
     );
-    Ok(Some(tx))
+    Ok((Some(tx), None))
 }
 
 fn build_kickoff_ix(

diff --git a/plugin/src/executors/state/executable_threads.rs b/plugin/src/executors/state/executable_threads.rs
@@ -95,7 +95,7 @@ impl ExecutableThreads {
         info!(
             "dropped_threads: {:?} executable_threads: {:?}",
             self.1.load(Ordering::Relaxed),
-            *w_state
+            *w_state,
         );
     }
 

diff --git a/plugin/src/executors/tx.rs b/plugin/src/executors/tx.rs
@@ -1,7 +1,8 @@
 use std::{
     collections::{HashMap, HashSet},
     fmt::Debug,
-    sync::Arc,
+    io::Write,
+    sync::{Arc, RwLock},
 };
 
 use bincode::serialize;
@@ -37,6 +38,8 @@ use super::{
 pub struct TxExecutor {
     pub config: PluginConfig,
     pub executable_threads: ExecutableThreads,
+    // Temporary state for blacklisted threads (from old positions that were deleted and SLTP cleanups call not working)
+    pub blacklisted_threads: RwLock<HashSet<Pubkey>>,
     pub transaction_history: TransactionHistory,
     pub rotation_history: RotationHistory,
     pub keypair: Keypair,
@@ -60,6 +63,7 @@ impl TxExecutor {
         Self {
             config: config.clone(),
             executable_threads: ExecutableThreads::default(),
+            blacklisted_threads: RwLock::new(HashSet::new()),
             transaction_history: TransactionHistory::default(),
             rotation_history: RotationHistory::default(),
             keypair: read_or_new_keypair(config.keypath),
@@ -73,6 +77,10 @@ impl TxExecutor {
         slot: u64,
         runtime: Arc<Runtime>,
     ) -> PluginResult<()> {
+        info!(
+            "blacklisted_threads: {:?}",
+            self.blacklisted_threads.read().unwrap().len()
+        );
         self.executable_threads
             .rebase_threads(slot, &thread_pubkeys)
             .await;
@@ -301,7 +309,15 @@ impl TxExecutor {
             }
         }
 
-        if let Ok(tx) = crate::builders::build_thread_exec_tx(
+        // check if the thread is blacklisted
+        if self
+            .blacklisted_threads
+            .read()
+            .unwrap()
+            .contains(&thread_pubkey)
+        {
+            return None;
+        } else if let Ok((tx, blacklisted)) = crate::builders::build_thread_exec_tx(
             client.clone(),
             &self.keypair,
             due_slot,
@@ -311,7 +327,13 @@ impl TxExecutor {
         )
         .await
         {
-            if let Some(tx) = tx {
+            if let Some(blacklisted) = blacklisted {
+                self.blacklisted_threads
+                    .write()
+                    .unwrap()
+                    .insert(blacklisted);
+                None
+            } else if let Some(tx) = tx {
                 if !self
                     .transaction_history
                     .is_duplicate_tx(observed_slot, thread_pubkey, &tx)

diff --git a/sablier-geyser-plugin-release-aarch64-apple-darwin.tar.bz2 b/sablier-geyser-plugin-release-aarch64-apple-darwin.tar.bz2
diff --git a/sablier-geyser-plugin-release-aarch64-apple-darwin.yml b/sablier-geyser-plugin-release-aarch64-apple-darwin.yml
@@ -0,0 +1,3 @@
+channel: 
+commit: 42475e29fd1afa983a9f003b1cb0ca106a840485
+target: aarch64-apple-darwin
diff --git a/sablier-geyser-plugin-release/version.yml b/sablier-geyser-plugin-release/version.yml
@@ -0,0 +1,3 @@
+channel: 
+commit: 42475e29fd1afa983a9f003b1cb0ca106a840485
+target: aarch64-apple-darwin