Merge pull request #22 from holaplex/ryans/increased-limits

Job Runner + Increased Limits

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]>",
 ]
@@ -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

@@ -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"]

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;

crates/rabbitmq/src/lib.rs

@@ -279,8 +279,7 @@ impl<'a> QueueInfo<'a> {
             let mut queue_fields = FieldTable::default();
             queue_fields.insert(
                 "x-max-length-bytes".into(),
-                // Top out length at 100 MiB
-                AMQPValue::LongLongInt(self.0.max_len_bytes.min(100 * 1024 * 1024)),
+                AMQPValue::LongLongInt(self.0.max_len_bytes),
             );
 
             // TODO: add a true DL queue