Merge pull request #15 from holaplex/dev

ray-kast · web-flow · commit c33467ee79a6 · 2022-08-10T13:34:09.000-07:00
Release v0.5
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/plugin/Cargo.toml b/crates/plugin/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "holaplex-indexer-rabbitmq-geyser"
-version = "0.4.0"
+version = "0.5.0"
 authors = [
   "ryans <ryans@holaplex.com>",
 ]
@@ -51,7 +51,7 @@ features = [
 
 [dependencies.indexer-rabbitmq]
 package = "holaplex-indexer-rabbitmq"
-version = "=0.1.0"
+version = "=0.2.0"
 path = "../rabbitmq"
 default-features = false
 features = ["producer", "geyser"]
diff --git a/crates/plugin/sample_config.json b/crates/plugin/sample_config.json
@@ -26,9 +26,20 @@
       "useZ65tbyvWpdYCLDJaegGK34Lnsi8S3jZdwx8122qp",
       "LocktDzaV1W2Bm9DeZeiyz4J9zs4fRqNiYqQyracRXw",
       "Govz1VyoyLD5BL6CSCxUJLVLsQHRwjfFj1prNsdNg5Jw",
-      "GokivDYuQXPZCWRkwMhdH2h91KpDQXBEmpgBgs55bnpH"
+      "GokivDYuQXPZCWRkwMhdH2h91KpDQXBEmpgBgs55bnpH",
+      "TBondmkCYxaPCKG4CHYfVTcwQ8on31xnJrPzk8F8WsS",
+      "nameXpT2PwZ2iA6DTNYTotTmiMYusBCYqwBLN2QgF4w",
+      "HAbiTatJVqoCJd9asyr6RxMEdwtfrQugwp7VAFyKWb1g",
+      "GovER5Lthms3bLBqWub97yVrMmEogzX7xNjdXpPPCVZw"
     ],
     "startup": false
   },
-  "instructionPrograms": []
+  "instructions": {
+    "programs": [
+      "M2mx93ekt1fmXSVkTrUL9xVFHkmME8HTUi5Cyc5aF7K",
+      "MEisE1HzehtrDpAAT8PnLHjpSSkRYakotTuJRPjTpo8",
+      "hausS13jsjafwWwGqZTUQRmWyvyxn9EQpqMwV1PBBmk",
+      "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"
+    ]
+  }
 }
diff --git a/crates/plugin/src/config.rs b/crates/plugin/src/config.rs
@@ -16,8 +16,7 @@ pub struct Config {
     metrics: Metrics,
 
     accounts: Accounts,
-
-    instruction_programs: HashSet<String>,
+    instructions: Instructions,
 }
 
 #[serde_with::serde_as]
@@ -46,7 +45,7 @@ pub struct Metrics {
 }
 
 #[derive(Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct Accounts {
     pub owners: HashSet<String>,
 
@@ -57,6 +56,26 @@ pub struct Accounts {
     ///  - `Some(true)`: Only send updates when `is_startup` is `true`.
     ///  - `Some(false)`: Only send updates when `is_startup` is `false`.
     pub startup: Option<bool>,
+
+    /// Set to true to disable heuristics to reduce the number of incoming
+    /// token account updates.  Has no effect if the spl-token pubkey is not in
+    /// the owners list.
+    #[serde(default)]
+    pub all_tokens: bool,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct Instructions {
+    pub programs: HashSet<String>,
+
+    /// Set to true to disable heuristics to reduce the number of incoming
+    /// token instructions.  Has no effect if the spl-token pubkey is not in the
+    /// programs list.  Currently the heuristics are tailored towards NFT burns,
+    /// only passing through instructions whose data indicates a burn of amount
+    /// 1.
+    #[serde(default)]
+    pub all_token_calls: bool,
 }
 
 impl Config {
@@ -73,12 +92,12 @@ impl Config {
             jobs,
             metrics,
             accounts,
-            instruction_programs,
+            instructions,
         } = self;
 
         let acct =
             AccountSelector::from_config(accounts).context("Failed to create account selector")?;
-        let ins = InstructionSelector::from_config(instruction_programs)
+        let ins = InstructionSelector::from_config(instructions)
             .context("Failed to create instruction selector")?;
 
         Ok((amqp, jobs, metrics, acct, ins))
diff --git a/crates/plugin/src/plugin.rs b/crates/plugin/src/plugin.rs
@@ -3,10 +3,10 @@ use std::{env, sync::Arc};
 use anyhow::Context;
 use hashbrown::HashSet;
 use indexer_rabbitmq::geyser::{AccountUpdate, InstructionNotify, Message};
-use solana_program::{instruction::CompiledInstruction, message::AccountKeys, program_pack::Pack};
-use spl_token::state::Account as TokenAccount;
+use solana_program::{instruction::CompiledInstruction, message::AccountKeys};
 
-static TOKEN_KEY: Pubkey = solana_program::pubkey!("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA");
+pub(crate) static TOKEN_KEY: Pubkey =
+    solana_program::pubkey!("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA");
 
 use serde::Deserialize;
 
@@ -41,7 +41,6 @@ pub(crate) struct Inner {
     acct_sel: AccountSelector,
     ins_sel: InstructionSelector,
     metrics: Arc<Metrics>,
-    token_addresses: HashSet<Pubkey>,
 }
 
 impl Inner {
@@ -148,7 +147,7 @@ impl GeyserPlugin for GeyserPluginRabbitMq {
                 .map_err(custom_err(&metrics.errs))?;
         }
 
-        let (amqp, jobs, metrics_conf, acct_sel, ins_sel) = Config::read(cfg)
+        let (amqp, jobs, metrics_conf, mut acct_sel, ins_sel) = Config::read(cfg)
             .and_then(Config::into_parts)
             .map_err(custom_err(&metrics.errs))?;
 
@@ -172,7 +171,7 @@ impl GeyserPlugin for GeyserPluginRabbitMq {
             .build()
             .map_err(custom_err(&metrics.errs))?;
 
-        let (producer, token_addresses) = rt.block_on(async {
+        let producer = rt.block_on(async {
             let producer = Sender::new(
                 amqp,
                 format!("geyser-rabbitmq-{}@{}", version, host),
@@ -182,11 +181,15 @@ impl GeyserPlugin for GeyserPluginRabbitMq {
             .await
             .map_err(custom_err(&metrics.errs))?;
 
-            let tokens = Self::load_token_reg()
-                .await
-                .map_err(custom_err(&metrics.errs))?;
+            if acct_sel.screen_tokens() {
+                acct_sel.init_tokens(
+                    Self::load_token_reg()
+                        .await
+                        .map_err(custom_err(&metrics.errs))?,
+                );
+            }
 
-            Result::<_>::Ok((producer, tokens))
+            Result::<_>::Ok(producer)
         })?;
 
         self.0 = Some(Arc::new(Inner {
@@ -195,7 +198,6 @@ impl GeyserPlugin for GeyserPluginRabbitMq {
             acct_sel,
             ins_sel,
             metrics,
-            token_addresses,
         }));
 
         Ok(())
@@ -228,20 +230,6 @@ impl GeyserPlugin for GeyserPluginRabbitMq {
                             write_version,
                         } = *acct;
 
-                        if owner == TOKEN_KEY.as_ref()
-                            && data.len() == TokenAccount::get_packed_len()
-                        {
-                            let token_account = TokenAccount::unpack_from_slice(data);
-
-                            if let Ok(token_account) = token_account {
-                                if token_account.amount > 1
-                                    || this.token_addresses.contains(&token_account.mint)
-                                {
-                                    return Ok(());
-                                }
-                            }
-                        }
-
                         let key = Pubkey::new_from_array(pubkey.try_into()?);
                         let owner = Pubkey::new_from_array(owner.try_into()?);
                         let data = data.to_owned();
@@ -288,7 +276,7 @@ impl GeyserPlugin for GeyserPluginRabbitMq {
                 .get(ins.program_id_index as usize)
                 .ok_or_else(|| anyhow!("Couldn't get program ID for instruction"))?;
 
-            if !sel.is_selected(&program) {
+            if !sel.is_selected(&program, ins) {
                 return Ok(None);
             }
 
diff --git a/crates/plugin/src/selectors.rs b/crates/plugin/src/selectors.rs
@@ -1,53 +1,111 @@
 use hashbrown::HashSet;
 use indexer_rabbitmq::geyser::StartupType;
+use solana_program::{instruction::CompiledInstruction, program_pack::Pack};
+use spl_token::state::Account as TokenAccount;
 
-use super::config::Accounts;
-use crate::{interface::ReplicaAccountInfo, prelude::*};
+use crate::{
+    config::{Accounts, Instructions},
+    interface::ReplicaAccountInfo,
+    plugin::TOKEN_KEY,
+    prelude::*,
+};
 
 #[derive(Debug)]
 pub struct AccountSelector {
     owners: HashSet<[u8; 32]>,
     startup: Option<bool>,
+    token_addresses: Option<HashSet<Pubkey>>,
 }
 
 impl AccountSelector {
     pub fn from_config(config: Accounts) -> Result<Self> {
-        let Accounts { owners, startup } = config;
+        let Accounts {
+            owners,
+            all_tokens,
+            startup,
+        } = config;
 
         let owners = owners
             .into_iter()
             .map(|s| s.parse().map(Pubkey::to_bytes))
             .collect::<Result<_, _>>()
             .context("Failed to parse account owner keys")?;
 
-        Ok(Self { owners, startup })
+        Ok(Self {
+            owners,
+            startup,
+            token_addresses: if all_tokens {
+                None
+            } else {
+                Some(HashSet::new())
+            },
+        })
+    }
+
+    /// Lazy-load the token addresses.  Fails if token addresses are not wanted
+    /// or if they have already been loaded.
+    pub fn init_tokens(&mut self, addrs: HashSet<Pubkey>) {
+        assert!(self.token_addresses.as_ref().unwrap().is_empty());
+        self.token_addresses = Some(addrs);
     }
 
     #[inline]
     pub fn startup(&self) -> StartupType {
         StartupType::new(self.startup)
     }
 
+    #[inline]
+    pub fn screen_tokens(&self) -> bool {
+        self.token_addresses.is_some()
+    }
+
     #[inline]
     pub fn is_selected(&self, acct: &ReplicaAccountInfo, is_startup: bool) -> bool {
-        self.startup.map_or(true, |s| is_startup == s) && self.owners.contains(acct.owner)
+        let ReplicaAccountInfo { owner, data, .. } = *acct;
+
+        if self.startup.map_or(false, |s| is_startup != s) || !self.owners.contains(owner) {
+            return false;
+        }
+
+        if owner == TOKEN_KEY.as_ref() && data.len() == TokenAccount::get_packed_len() {
+            if let Some(ref addrs) = self.token_addresses {
+                let token_account = TokenAccount::unpack_from_slice(data);
+
+                if let Ok(token_account) = token_account {
+                    if token_account.amount > 1 || addrs.contains(&token_account.mint) {
+                        return false;
+                    }
+                }
+            }
+        }
+
+        true
     }
 }
 
 #[derive(Debug)]
 pub struct InstructionSelector {
     programs: HashSet<Pubkey>,
+    screen_tokens: bool,
 }
 
 impl InstructionSelector {
-    pub fn from_config(programs: HashSet<String>) -> Result<Self> {
+    pub fn from_config(config: Instructions) -> Result<Self> {
+        let Instructions {
+            programs,
+            all_token_calls,
+        } = config;
+
         let programs = programs
             .into_iter()
             .map(|s| s.parse())
             .collect::<Result<_, _>>()
             .context("Failed to parse instruction program keys")?;
 
-        Ok(Self { programs })
+        Ok(Self {
+            programs,
+            screen_tokens: !all_token_calls,
+        })
     }
 
     #[inline]
@@ -56,7 +114,28 @@ impl InstructionSelector {
     }
 
     #[inline]
-    pub fn is_selected(&self, pgm: &Pubkey) -> bool {
-        self.programs.contains(pgm)
+    pub fn is_selected(&self, pgm: &Pubkey, ins: &CompiledInstruction) -> bool {
+        if !self.programs.contains(pgm) {
+            return false;
+        }
+
+        if self.screen_tokens && *pgm == TOKEN_KEY {
+            if let [8, rest @ ..] = ins.data.as_slice() {
+                let amt = rest.try_into().map(u64::from_le_bytes);
+
+                if !matches!(amt, Ok(1)) {
+                    return false;
+                }
+
+                debug_assert_eq!(
+                    ins.data,
+                    spl_token::instruction::TokenInstruction::Burn { amount: 1_u64 }.pack(),
+                );
+            } else {
+                return false;
+            }
+        }
+
+        true
     }
 }
diff --git a/crates/plugin/src/sender.rs b/crates/plugin/src/sender.rs
@@ -55,7 +55,7 @@ impl Sender {
 
         Producer::new(
             &conn,
-            QueueType::new(amqp.network, startup_type, &Suffix::Production)?,
+            QueueType::new(amqp.network, startup_type, &Suffix::ProductionUnchecked)?,
         )
         .await
     }
diff --git a/crates/rabbitmq/Cargo.toml b/crates/rabbitmq/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "holaplex-indexer-rabbitmq"
-version = "0.1.0"
+version = "0.2.0"
 authors = [
   "ryans <ryans@holaplex.com>",
 ]
@@ -19,7 +19,7 @@ default = ["consumer"]
 geyser = ["solana-program", "suffix"]
 http-indexer = ["solana-program", "suffix"]
 producer = ["suffix"]
-search-indexer = ["serde_json", "suffix"]
+search-indexer = ["serde_json", "solana-program", "suffix"]
 suffix = ["clap"]
 
 [dependencies]
diff --git a/crates/rabbitmq/src/search_indexer.rs b/crates/rabbitmq/src/search_indexer.rs
diff --git a/crates/rabbitmq/src/suffix.rs b/crates/rabbitmq/src/suffix.rs

Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,7 @@ impl Sender {`
`55`	`55`
`56`	`56`	`Producer::new(`
`57`	`57`	`&conn,`
`58`		`- QueueType::new(amqp.network, startup_type, &Suffix::Production)?,`
	`58`	`+ QueueType::new(amqp.network, startup_type, &Suffix::ProductionUnchecked)?,`
`59`	`59`	`)`
`60`	`60`	`.await`
`61`	`61`	`}`