Added filtering of transactions (#81)

Signed-off-by: Anthony Milbourne <[email protected]>

Author: amilbourne

src/filter.rs

@@ -33,8 +33,8 @@ impl Filter {
         }
     }
 
-    pub fn wants_program(&self, program: &[u8]) -> bool {
-        let key = match <&[u8; 32]>::try_from(program) {
+    pub fn wants_account_key(&self, account_key: &[u8]) -> bool {
+        let key = match <&[u8; 32]>::try_from(account_key) {
             Ok(key) => key,
             _ => return true,
         };
@@ -59,12 +59,12 @@ mod tests {
         let filter = Filter::new(&config);
         assert_eq!(filter.program_ignores.len(), 2);
 
-        assert!(filter.wants_program(
+        assert!(filter.wants_account_key(
             &Pubkey::from_str("9xQeWvG816bUx9EPjHmaT23yvVM2ZWbrrpZb9PusVFin")
                 .unwrap()
                 .to_bytes()
         ));
-        assert!(!filter.wants_program(
+        assert!(!filter.wants_account_key(
             &Pubkey::from_str("Vote111111111111111111111111111111111111111")
                 .unwrap()
                 .to_bytes()

src/plugin.rs

@@ -14,14 +14,15 @@
 
 use {
     crate::*,
-    log::info,
+    log::{debug, info, log_enabled},
     rdkafka::util::get_rdkafka_version,
     simple_error::simple_error,
     solana_geyser_plugin_interface::geyser_plugin_interface::{
         GeyserPlugin, GeyserPluginError as PluginError, ReplicaAccountInfo,
-        ReplicaAccountInfoVersions, ReplicaTransactionInfoVersions, Result as PluginResult,
-        SlotStatus as PluginSlotStatus,
+        ReplicaAccountInfoVersions, ReplicaTransactionInfo, ReplicaTransactionInfoVersions,
+        Result as PluginResult, SlotStatus as PluginSlotStatus,
     },
+    solana_program::pubkey::Pubkey,
     std::fmt::{Debug, Formatter},
 };
 
@@ -90,7 +91,8 @@ impl GeyserPlugin for KafkaPlugin {
         }
 
         let info = Self::unwrap_update_account(account);
-        if !self.unwrap_filter().wants_program(info.owner) {
+        if !self.unwrap_filter().wants_account_key(info.owner) {
+            Self::log_ignore_account_update(info);
             return Ok(());
         }
 
@@ -143,7 +145,23 @@ impl GeyserPlugin for KafkaPlugin {
             return Ok(());
         }
 
-        let event = Self::build_transaction_event(slot, transaction);
+        let info = Self::unwrap_transaction(transaction);
+        let maybe_ignored = info
+            .transaction
+            .message()
+            .account_keys()
+            .iter()
+            .find(|key| !self.unwrap_filter().wants_account_key(&key.to_bytes()));
+        if maybe_ignored.is_some() {
+            debug!(
+                "Ignoring transaction {:?} due to account key: {:?}",
+                info.signature,
+                &maybe_ignored.unwrap()
+            );
+            return Ok(());
+        }
+
+        let event = Self::build_transaction_event(slot, info);
 
         publisher
             .update_transaction(event)
@@ -178,6 +196,12 @@ impl KafkaPlugin {
         }
     }
 
+    fn unwrap_transaction(transaction: ReplicaTransactionInfoVersions) -> &ReplicaTransactionInfo {
+        match transaction {
+            ReplicaTransactionInfoVersions::V0_0_1(info) => info,
+        }
+    }
+
     fn build_compiled_instruction(
         ix: &solana_program::instruction::CompiledInstruction,
     ) -> CompiledInstruction {
@@ -216,9 +240,8 @@ impl KafkaPlugin {
 
     fn build_transaction_event(
         slot: u64,
-        transaction: ReplicaTransactionInfoVersions,
+        transaction: &ReplicaTransactionInfo,
     ) -> TransactionEvent {
-        let ReplicaTransactionInfoVersions::V0_0_1(transaction) = transaction;
         let transaction_status_meta = transaction.transaction_status_meta;
         let signature = transaction.signature;
         let is_vote = transaction.is_vote;
@@ -380,4 +403,17 @@ impl KafkaPlugin {
             }),
         }
     }
+
+    fn log_ignore_account_update(info: &ReplicaAccountInfo) {
+        if log_enabled!(::log::Level::Debug) {
+            match <&[u8; 32]>::try_from(info.owner) {
+                Ok(key) => debug!(
+                    "Ignoring update for account key: {:?}",
+                    Pubkey::new_from_array(*key)
+                ),
+                // Err should never happen because wants_account_key only returns false if the input is &[u8; 32]
+                Err(_err) => debug!("Ignoring update for account key: {:?}", info.owner),
+            };
+        }
+    }
 }