Merge pull request #23 from holaplex/dev

v0.6 Release

Author: ray-kast

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "holaplex-indexer-rabbitmq-geyser"
-version = "0.5.0"
+version = "0.6.0"
 authors = [
   "ryans <[email protected]>",
 ]
@@ -28,14 +28,14 @@ parking_lot = "0.12.0"
 reqwest = "0.11.6"
 serde = { version = "1.0.133", features = ["derive"] }
 serde_json = "1.0.75"
-serde_with = "1.11.0"
+serde_with = "2.0.0"
 tokio-executor-trait = "2.1.0"
 tokio-reactor-trait = "1.1.0"
 
-solana-geyser-plugin-interface = "1.10.25"
-solana-logger = "1.10.25"
-solana-program = "1.10.25"
-solana-metrics = "1.10.25"
+solana-geyser-plugin-interface = "~1.10.25"
+solana-logger = "~1.10.25"
+solana-program = "~1.10.25"
+solana-metrics = "~1.10.25"
 spl-token = "3.3.0"
 
 [dependencies.tokio]
@@ -51,7 +51,7 @@ features = [
 
 [dependencies.indexer-rabbitmq]
 package = "holaplex-indexer-rabbitmq"
-version = "=0.2.0"
+version = "=0.3.0"
 path = "../rabbitmq"
 default-features = false
 features = ["producer", "geyser"]

crates/plugin/Cargo.toml

@@ -32,6 +32,7 @@
       "HAbiTatJVqoCJd9asyr6RxMEdwtfrQugwp7VAFyKWb1g",
       "GovER5Lthms3bLBqWub97yVrMmEogzX7xNjdXpPPCVZw"
     ],
+    "pubkeys": [],
     "startup": false
   },
   "instructions": {

crates/plugin/sample_config.json

@@ -7,7 +7,7 @@ use crate::{
 };
 
 #[derive(Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct Config {
     amqp: Amqp,
     jobs: Jobs,
@@ -17,11 +17,15 @@ pub struct Config {
 
     accounts: Accounts,
     instructions: Instructions,
+
+    /// Unused but required by the validator to load the plugin
+    #[allow(dead_code)]
+    libpath: String,
 }
 
 #[serde_with::serde_as]
 #[derive(Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct Amqp {
     pub address: String,
 
@@ -30,7 +34,7 @@ pub struct Amqp {
 }
 
 #[derive(Debug, Deserialize)]
-#[serde(rename_all = "camelCase")]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct Jobs {
     pub limit: usize,
 
@@ -39,22 +43,27 @@ pub struct Jobs {
 }
 
 #[derive(Debug, Default, Deserialize)]
-#[serde(rename_all = "camelCase")]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct Metrics {
     pub config: Option<String>,
 }
 
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct Accounts {
+    #[serde(default)]
     pub owners: HashSet<String>,
 
+    #[serde(default)]
+    pub pubkeys: HashSet<String>,
+
     /// Filter for changing how to interpret the `is_startup` flag.
     ///
     /// This option has three states:
     ///  - `None`: Ignore the `is_startup` flag and send all updates.
     ///  - `Some(true)`: Only send updates when `is_startup` is `true`.
     ///  - `Some(false)`: Only send updates when `is_startup` is `false`.
+    #[serde(default)]
     pub startup: Option<bool>,
 
     /// Set to true to disable heuristics to reduce the number of incoming
@@ -67,6 +76,7 @@ pub struct Accounts {
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub struct Instructions {
+    #[serde(default)]
     pub programs: HashSet<String>,
 
     /// Set to true to disable heuristics to reduce the number of incoming
@@ -93,6 +103,7 @@ impl Config {
             metrics,
             accounts,
             instructions,
+            libpath: _,
         } = self;
 
         let acct =

crates/plugin/src/config.rs

@@ -13,6 +13,7 @@ use crate::{
 #[derive(Debug)]
 pub struct AccountSelector {
     owners: HashSet<[u8; 32]>,
+    pubkeys: HashSet<[u8; 32]>,
     startup: Option<bool>,
     token_addresses: Option<HashSet<Pubkey>>,
 }
@@ -22,6 +23,7 @@ impl AccountSelector {
         let Accounts {
             owners,
             all_tokens,
+            pubkeys,
             startup,
         } = config;
 
@@ -31,8 +33,15 @@ impl AccountSelector {
             .collect::<Result<_, _>>()
             .context("Failed to parse account owner keys")?;
 
+        let pubkeys = pubkeys
+            .into_iter()
+            .map(|s| s.parse().map(Pubkey::to_bytes))
+            .collect::<Result<_, _>>()
+            .context("Failed to parse account pubkeys")?;
+
         Ok(Self {
             owners,
+            pubkeys,
             startup,
             token_addresses: if all_tokens {
                 None
@@ -63,7 +72,9 @@ impl AccountSelector {
     pub fn is_selected(&self, acct: &ReplicaAccountInfo, is_startup: bool) -> bool {
         let ReplicaAccountInfo { owner, data, .. } = *acct;
 
-        if self.startup.map_or(false, |s| is_startup != s) || !self.owners.contains(owner) {
+        if self.startup.map_or(false, |s| is_startup != s)
+            || !(self.owners.contains(owner) || self.pubkeys.contains(acct.pubkey))
+        {
             return false;
         }
 

crates/plugin/src/selectors.rs

@@ -1,6 +1,6 @@
 [package]
 name = "holaplex-indexer-rabbitmq"
-version = "0.2.0"
+version = "0.3.0"
 authors = [
   "ryans <[email protected]>",
 ]
@@ -18,6 +18,7 @@ consumer = ["suffix"]
 default = ["consumer"]
 geyser = ["solana-program", "suffix"]
 http-indexer = ["solana-program", "suffix"]
+job-runner = []
 producer = ["suffix"]
 search-indexer = ["serde_json", "solana-program", "suffix"]
 suffix = ["clap"]
@@ -33,5 +34,5 @@ serde = { version = "1.0.133", features = ["derive"] }
 serde_json = { version = "1.0.79", optional = true }
 # TODO: tighten this bound once more things support 1.10
 solana-program = { version = ">=1.9", optional = true }
-strum = { version = "0.23.0", features = ["derive"] }
+strum = { version = "0.24.1", features = ["derive"] }
 thiserror = "1.0.30"

crates/rabbitmq/Cargo.toml

@@ -125,9 +125,9 @@ impl QueueType {
                 binding: Binding::Fanout,
                 prefetch: 4096,
                 max_len_bytes: if suffix.is_debug() || matches!(startup_type, StartupType::Normal) {
-                    100 * 1024 * 1024 // 100 MiB
+                    512 * 1024 * 1024 // 512 MiB
                 } else {
-                    8 * 1024 * 1024 * 1024 // 8 GiB
+                    12 * 1024 * 1024 * 1024 // 12 GiB
                 },
                 auto_delete: suffix.is_debug(),
                 retry: Some(RetryProps {

crates/rabbitmq/src/geyser.rs

@@ -47,6 +47,9 @@ pub trait Entity: std::fmt::Debug + Serialize + for<'a> Deserialize<'a> {
 
     /// A name to use when declaring queues and exchanges
     const ID: Self::Id;
+
+    /// The maximum queue length to use for this entity
+    const MAX_LEN_BYTES: i64;
 }
 
 /// Fetch the off-chain JSON for a metadata account
@@ -67,6 +70,8 @@ impl Entity for MetadataJson {
     type Id = EntityId;
 
     const ID: EntityId = EntityId::MetadataJson;
+
+    const MAX_LEN_BYTES: i64 = 512 * 1024 * 1024; // 512 MiB
 }
 
 /// Fetch the off-chain JSON config for a storefront
@@ -82,6 +87,8 @@ impl Entity for StoreConfig {
     type Id = EntityId;
 
     const ID: EntityId = EntityId::StoreConfig;
+
+    const MAX_LEN_BYTES: i64 = 100 * 1024 * 1024; // 100 MiB
 }
 
 impl<E: Entity> QueueType<E> {
@@ -100,7 +107,7 @@ impl<E: Entity> QueueType<E> {
                 queue,
                 binding: Binding::Fanout,
                 prefetch: 1024,
-                max_len_bytes: 100 * 1024 * 1024, // 100 MiB
+                max_len_bytes: E::MAX_LEN_BYTES,
                 auto_delete: suffix.is_debug(),
                 retry: Some(RetryProps {
                     max_tries: 8,

crates/rabbitmq/src/http_indexer.rs

@@ -0,0 +1,68 @@
+//! Queue configuration for dispatching background jobs.
+
+use std::time::Duration;
+
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    queue_type::{Binding, QueueProps, RetryProps},
+    suffix::Suffix,
+    Result,
+};
+
+/// Message data for a job dispatch request
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum Message {
+    /// Refresh a table of cached data
+    RefreshTable(String),
+}
+
+/// AMQP configuration for job runners
+#[derive(Debug, Clone)]
+pub struct QueueType {
+    props: QueueProps,
+}
+
+impl QueueType {
+    /// Construct a new queue configuration given the expected sender and queue
+    /// suffix configuration
+    ///
+    /// # Errors
+    /// This function fails if the given queue suffix is invalid.
+    pub fn new(sender: &str, suffix: &Suffix) -> Result<Self> {
+        let exchange = format!("{}.jobs", sender);
+        let queue = suffix.format(format!("{}.runner", exchange))?;
+
+        Ok(Self {
+            props: QueueProps {
+                exchange,
+                queue,
+                binding: Binding::Fanout,
+                prefetch: 1,
+                max_len_bytes: 100 * 1024 * 1024, // 100 MiB
+                auto_delete: suffix.is_debug(),
+                retry: Some(RetryProps {
+                    max_tries: 5,
+                    delay_hint: Duration::from_secs(5),
+                    max_delay: Duration::from_secs(10 * 60),
+                }),
+            },
+        })
+    }
+}
+
+impl crate::QueueType for QueueType {
+    type Message = Message;
+
+    #[inline]
+    fn info(&self) -> crate::queue_type::QueueInfo {
+        (&self.props).into()
+    }
+}
+
+/// The type of an search indexer producer
+#[cfg(feature = "producer")]
+pub type Producer = crate::producer::Producer<QueueType>;
+/// The type of an search indexer consumer
+#[cfg(feature = "consumer")]
+pub type Consumer = crate::consumer::Consumer<QueueType>;

crates/rabbitmq/src/job_runner.rs

@@ -46,6 +46,8 @@ pub mod dl_consumer;
 pub mod geyser;
 #[cfg(feature = "http-indexer")]
 pub mod http_indexer;
+#[cfg(feature = "job-runner")]
+pub mod job_runner;
 #[cfg(feature = "producer")]
 pub mod producer;
 mod queue_type;