delete and recreate the consumer (#162)

rpcpool · Feb 10, 2025 · db7ee2d · db7ee2d
1 parent 6efad62
commit db7ee2d
Show file tree

Hide file tree

Showing 3 changed files with 95 additions and 161 deletions.
diff --git a/grpc-ingest/src/config.rs b/grpc-ingest/src/config.rs
@@ -56,6 +56,11 @@ pub struct ConfigIngestStream {
         deserialize_with = "deserialize_usize_str"
     )]
     pub max_concurrency: usize,
+    #[serde(
+        default = "ConfigIngestStream::default_ack_concurrency",
+        deserialize_with = "deserialize_usize_str"
+    )]
+    pub ack_concurrency: usize,
     #[serde(
         default = "ConfigIngestStream::default_xack_buffer_size",
         deserialize_with = "deserialize_usize_str"
@@ -71,6 +76,10 @@ impl ConfigIngestStream {
         2
     }
 
+    pub const fn default_ack_concurrency() -> usize {
+        5
+    }
+
     pub const fn default_xack_batch_max_idle() -> Duration {
         Duration::from_millis(10_000)
     }

diff --git a/grpc-ingest/src/ingester.rs b/grpc-ingest/src/ingester.rs
@@ -108,7 +108,8 @@ pub async fn run(config: ConfigIngester) -> anyhow::Result<()> {
                     .map_err(Into::into)
             })
         })
-        .start()?;
+        .start()
+        .await?;
     let account_stream = IngestStream::build()
         .config(config.accounts.clone())
         .connection(connection.clone())
@@ -124,7 +125,8 @@ pub async fn run(config: ConfigIngester) -> anyhow::Result<()> {
                     .map_err(Into::into)
             })
         })
-        .start()?;
+        .start()
+        .await?;
     let transactions_stream = IngestStream::build()
         .config(config.transactions.clone())
         .connection(connection.clone())
@@ -140,7 +142,8 @@ pub async fn run(config: ConfigIngester) -> anyhow::Result<()> {
                     .map_err(Into::into)
             })
         })
-        .start()?;
+        .start()
+        .await?;
     let snapshot_stream = IngestStream::build()
         .config(config.snapshots.clone())
         .connection(connection.clone())
@@ -156,7 +159,8 @@ pub async fn run(config: ConfigIngester) -> anyhow::Result<()> {
                     .map_err(Into::into)
             })
         })
-        .start()?;
+        .start()
+        .await?;
 
     let mut shutdown = create_shutdown()?;