fix(sv-consumer): Ensure block is process before act on it and get next block

Author: luizstacio

services/consumer/src/executor/block_executor.rs

@@ -53,8 +53,8 @@ use super::{
 };
 use crate::{errors::ConsumerError, metrics::Metrics};
 
-const MAX_CONCURRENT_TASKS: usize = 32;
-const BATCH_SIZE: usize = 100;
+const MAX_CONCURRENT_TASKS: usize = 30;
+const BATCH_SIZE: usize = 30;
 
 #[derive(Debug)]
 enum ProcessResult {
@@ -91,7 +91,6 @@ impl BlockExecutor {
         &self,
         token: &CancellationToken,
     ) -> Result<(), ConsumerError> {
-        let mut join_set = JoinSet::new();
         tracing::info!(
             "Starting consumer with max concurrent tasks: {}",
             MAX_CONCURRENT_TASKS
@@ -101,26 +100,18 @@ impl BlockExecutor {
         let queue = NatsQueue::BlockImporter(self.message_broker.clone());
 
         while !token.is_cancelled() {
-            tokio::select! {
-                msg_result = queue.subscribe(BATCH_SIZE) => {
-                    let mut messages = msg_result?;
-                    while let Some(msg) = messages.next().await {
-                        let msg = msg?;
-                        self.spawn_processing_tasks(msg, &mut join_set,)
-                        .await?;
-                    }
-                }
-                Some(result) = join_set.join_next() => {
+            let mut messages = queue.subscribe(BATCH_SIZE).await?;
+            while let Some(msg) = messages.next().await {
+                let mut join_set = JoinSet::new();
+                let msg = msg?;
+                self.spawn_processing_tasks(msg, &mut join_set).await?;
+                // Wait for all spawned tasks to complete before processing next message
+                while let Some(result) = join_set.join_next().await {
                     Self::handle_task_result(result, &telemetry).await?;
                 }
             }
         }
 
-        // Wait for all tasks to finish
-        while let Some(result) = join_set.join_next().await {
-            Self::handle_task_result(result, &telemetry).await?;
-        }
-
         tracing::info!("Stopping broker ...");
         shutdown_broker_with_timeout(&self.message_broker).await;
         tracing::info!("Broker stopped successfully!");
@@ -138,13 +129,20 @@ impl BlockExecutor {
         let payload = msg.payload();
         let msg_payload = MsgPayload::decode_json(&payload)?.arc();
         let packets = Self::build_packets(&msg_payload);
+
         join_set.spawn({
             let semaphore = semaphore.clone();
             let packets = packets.clone();
             let msg_payload = msg_payload.clone();
             async move {
                 let _permit = semaphore.acquire().await?;
                 let result = handle_stores(&db, &packets, &msg_payload).await;
+                if result.is_ok() {
+                    msg.ack().await.map_err(|e| {
+                        tracing::error!("Failed to ack message: {:?}", e);
+                        ConsumerError::MessageBrokerClient(e)
+                    })?;
+                }
                 Ok::<_, ConsumerError>(ProcessResult::Store(result))
             }
         });
@@ -162,11 +160,6 @@ impl BlockExecutor {
             }
         });
 
-        msg.ack().await.map_err(|e| {
-            tracing::error!("Failed to ack message: {:?}", e);
-            ConsumerError::MessageBrokerClient(e)
-        })?;
-
         Ok(())
     }