Allow ignoring programs

Author: Richard Patel

Cargo.toml

@@ -13,8 +13,9 @@ crate-type = ["cdylib", "rlib"]
 [dependencies]
 prost = "0.9"
 rdkafka = { version = "0.28", features = ["ssl-vendored", "sasl"] }
-solana-logger = { version = "=1.9.9" }
 solana-accountsdb-plugin-interface = { version = "=1.9.9" }
+solana-logger = { version = "=1.9.9" }
+solana-program = { version = "=1.9.9" }
 log = "0.4"
 serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }

README.md

@@ -39,7 +39,12 @@ Config is specified via the plugin's JSON config file.
     "partitioner": "murmur2_random"
   },
   "shutdown_timeout_ms": 30000,
-  "update_account_topic": "solana.testnet.account_updates"
+  "update_account_topic": "solana.testnet.account_updates",
+  "slot_status_topic": "solana.testnet.slot_status",
+  "program_ignores": [
+    "Sysvar1111111111111111111111111111111111111",
+    "Vote111111111111111111111111111111111111111"
+  ],
 }
 ```
 
@@ -50,6 +55,8 @@ Config is specified via the plugin's JSON config file.
   This plugin overrides the defaults as seen in the example config.
 - `shutdown_timeout_ms`: Time the plugin is given to flush out all messages to Kafka upon exit request.
 - `update_account_topic`: Topic name of account updates. Omit to disable.
+- `slot_status_topic`: Topic name of slot status update. Omit to disable.
+- `program_ignores`: Solana program IDs for which to ignore updates for owned accounts.
 
 ## Buffering
 

src/config.rs

@@ -40,6 +40,9 @@ pub struct Config {
     /// Kafka topic to send slot status updates to.
     #[serde(default)]
     pub slot_status_topic: String,
+    /// List of programs to ignore.
+    #[serde(default)]
+    pub program_ignores: Vec<String>,
 }
 
 impl Default for Config {
@@ -49,6 +52,7 @@ impl Default for Config {
             shutdown_timeout_ms: 30_000,
             update_account_topic: "".to_owned(),
             slot_status_topic: "".to_owned(),
+            program_ignores: Vec::new(),
         }
     }
 }

src/filter.rs

@@ -0,0 +1,71 @@
+// Copyright 2022 Blockdaemon Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use {
+    crate::*,
+    solana_program::pubkey::Pubkey,
+    std::{collections::HashSet, str::FromStr},
+};
+
+pub struct Filter {
+    program_ignores: HashSet<[u8; 32]>,
+}
+
+impl Filter {
+    pub fn new(config: &Config) -> Self {
+        Self {
+            program_ignores: config
+                .program_ignores
+                .iter()
+                .flat_map(|p| Pubkey::from_str(p).ok().map(|p| p.to_bytes()))
+                .collect(),
+        }
+    }
+
+    pub fn wants_program(&self, program: &[u8]) -> bool {
+        let key = match <&[u8; 32]>::try_from(program) {
+            Ok(key) => key,
+            _ => return true,
+        };
+        !self.program_ignores.contains(key)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_filter() {
+        let mut config = Config::default();
+        config.program_ignores = vec![
+            "Sysvar1111111111111111111111111111111111111".to_owned(),
+            "Vote111111111111111111111111111111111111111".to_owned(),
+        ];
+
+        let filter = Filter::new(&config);
+        assert_eq!(filter.program_ignores.len(), 2);
+
+        assert!(filter.wants_program(
+            &Pubkey::from_str("9xQeWvG816bUx9EPjHmaT23yvVM2ZWbrrpZb9PusVFin")
+                .unwrap()
+                .to_bytes()
+        ));
+        assert!(!filter.wants_program(
+            &Pubkey::from_str("Vote111111111111111111111111111111111111111")
+                .unwrap()
+                .to_bytes()
+        ));
+    }
+}

src/lib.rs

@@ -16,12 +16,14 @@ use solana_accountsdb_plugin_interface::accountsdb_plugin_interface::AccountsDbP
 
 mod config;
 mod event;
+mod filter;
 mod plugin;
 mod publisher;
 
 pub use {
     config::{Config, Producer},
     event::*,
+    filter::Filter,
     plugin::KafkaPlugin,
     publisher::Publisher,
 };

src/plugin.rs

@@ -27,6 +27,7 @@ use {
 #[derive(Default)]
 pub struct KafkaPlugin {
     publisher: Option<Publisher>,
+    filter: Option<Filter>,
 }
 
 impl Debug for KafkaPlugin {
@@ -64,15 +65,15 @@ impl AccountsDbPlugin for KafkaPlugin {
 
         let publisher = Publisher::new(producer, &config);
         self.publisher = Some(publisher);
+        self.filter = Some(Filter::new(&config));
         info!("Spawned producer");
 
         Ok(())
     }
 
     fn on_unload(&mut self) {
-        if let Some(publisher) = self.publisher.take() {
-            drop(publisher);
-        }
+        self.publisher = None;
+        self.filter = None;
     }
 
     fn update_account(
@@ -86,6 +87,10 @@ impl AccountsDbPlugin for KafkaPlugin {
         }
 
         let info = Self::unwrap_update_account(account);
+        if !self.unwrap_filter().wants_program(info.pubkey) {
+            return Ok(());
+        }
+
         let event = UpdateAccountEvent {
             slot,
             pubkey: info.pubkey.to_vec(),
@@ -143,6 +148,10 @@ impl KafkaPlugin {
         self.publisher.as_ref().expect("publisher is unavailable")
     }
 
+    fn unwrap_filter(&self) -> &Filter {
+        self.filter.as_ref().expect("filter is unavailable")
+    }
+
     fn unwrap_update_account(account: ReplicaAccountInfoVersions) -> &ReplicaAccountInfo {
         match account {
             ReplicaAccountInfoVersions::V0_0_1(info) => info,