feat: replace DelayQueue with sleep , timeouts CLI-configurable, add rotation guards

Signed-off-by: rohan2794 <rohan2794@gmial.com>

Author: rohan2794

Cargo.lock

@@ -23,6 +23,7 @@ tokio = { version = "1.41.0", features = ["full"] }
 tokio-retry = "0.3"
 futures = "0.3"
 async-trait = "0.1.83"
+async-nats = "0.37.0"
 
 # tracing
 tracing = "0.1.40"

Cargo.toml

@@ -4,24 +4,16 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-# Async runtime
 tokio = { workspace = true }
-# NATS client
-async-nats = { version = "0.37.0" }
-# JSON serialization
+async-nats = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
-# Logging and Observability
 tracing = { workspace = true }
-tracing-subscriber = { workspace = true, features = ["json"]}
-tracing-appender = "0.2"
-# Error handling
 anyhow = { workspace = true }
 parse-size = { workspace = true }
 futures = { workspace = true }
 rand = { workspace = true }
-tokio-util = { version = "0.7.13", features = ["time"] }
-tokio-stream = { version = "0.1.16", features = ["sync"] }
 clap = { workspace = true }
 url = { workspace = true }
+humantime = { workspace = true }
 utils = { path = "../dependencies/control-plane/utils/utils-lib" }

events-aggregator/Cargo.toml

@@ -1,6 +1,5 @@
 use crate::constant::{
-    CONNECTION_TIMEOUT, JETSTREAM_CONSUMER_NAME, JETSTREAM_STREAM_NAME, MAX_ATTEMPTS,
-    MAX_BACKOFF_ATTEMPTS, REQUEST_TIMEOUT,
+    JETSTREAM_CONSUMER_NAME, JETSTREAM_STREAM_NAME, MAX_ATTEMPTS, MAX_BACKOFF_ATTEMPTS,
 };
 use anyhow::bail;
 use async_nats::jetstream::consumer::pull;
@@ -47,13 +46,17 @@ impl NatsManager {
     }
 
     /// Establish a connection to the NATS server with strict timeouts and a robust retry strategy.
-    pub async fn new(url: &str) -> Result<Self, async_nats::Error> {
+    pub async fn new(
+        url: &str,
+        connection_timeout: Duration,
+        request_timeout: Duration,
+    ) -> Result<Self, async_nats::Error> {
         info!("Attempting to connect to NATS at {url}...");
 
         // Configure native NATS backoff and retries
         let options = async_nats::ConnectOptions::new()
-            .connection_timeout(CONNECTION_TIMEOUT)
-            .request_timeout(Some(REQUEST_TIMEOUT))
+            .connection_timeout(connection_timeout)
+            .request_timeout(Some(request_timeout))
             // Enable retries on initial connection failure
             .retry_on_initial_connect()
             .reconnect_delay_callback(|attempts| {
@@ -78,6 +81,7 @@ impl NatsManager {
         subject: String,
         jetstream_enabled: bool,
         tx: mpsc::Sender<UnifiedMessage>,
+        request_timeout: Duration,
     ) -> anyhow::Result<()> {
         // Check the environment-derived flag (e.g., JETSTREAM_ENABLED=true)
         if !jetstream_enabled {
@@ -89,15 +93,16 @@ impl NatsManager {
         }
 
         info!("JetStream is enabled. Booting JetStream consumer with retry support...");
-        self.setup_jetstream(subject, tx).await?;
+        self.setup_jetstream(subject, tx, request_timeout).await?;
         Ok(())
     }
 
-    // JetStream Consumer Setup with Robust Retry Logic and Backoff
+    // JetStream Consumer Setup with Robust Retry Logic and Backoff.
     async fn setup_jetstream(
         &self,
         subject: String,
         tx: mpsc::Sender<UnifiedMessage>,
+        request_timeout: Duration,
     ) -> anyhow::Result<()> {
         let js = async_nats::jetstream::new(self.client.clone());
         let mut attempt = 0;
@@ -130,7 +135,7 @@ impl NatsManager {
                 continue;
             }
 
-            match tokio::time::timeout(REQUEST_TIMEOUT, js.get_stream(JETSTREAM_STREAM_NAME)).await
+            match tokio::time::timeout(request_timeout, js.get_stream(JETSTREAM_STREAM_NAME)).await
             {
                 Ok(Ok(stream)) => match stream
                     .get_or_create_consumer(

events-aggregator/src/client/nats_client.rs

@@ -1,11 +1,6 @@
-use std::time::Duration;
-
 /// Channel capacity for internal message passing and exporter buffering.
 pub const CHANNEL_CAPACITY: usize = 512;
 
-/// Maximum payload bytes to log when a message payload is not valid JSON.
-pub const MAX_NON_JSON_PAYLOAD_LOG_BYTES: usize = 4096;
-
 /// Maximum number of events to batch in a single export flush.
 pub const BATCH_MAX_SIZE: usize = 100;
 
@@ -24,12 +19,6 @@ pub const JETSTREAM_CONSUMER_NAME: &str = "events-aggregator-consumer";
 /// Default JetStream stream name used for publishing and consuming events.
 pub const JETSTREAM_STREAM_NAME: &str = "events-stream";
 
-/// Timeout for establishing a NATS connection.
-pub const CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);
-
-/// Timeout for NATS request/response operations.
-pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
-
 /// Maximum number of attempts for retryable operations.
 pub const MAX_ATTEMPTS: u32 = 10;
 
@@ -38,3 +27,6 @@ pub const MAX_BACKOFF_ATTEMPTS: u32 = 6;
 
 /// Base filename for the local event export file.
 pub const EVENTS_JSON_FILE: &str = "events.json";
+
+/// Filename for the rotated (previous) local event export file.
+pub const EVENTS_JSON_ROTATED_FILE: &str = "events.1.json";