feat: Track queue latency

.gitignore

@@ -17,6 +17,8 @@ target/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
 .DS_Store
+.serena
+.codegraph
 
 config.json
 

src/jobs/job.rs

@@ -24,6 +24,12 @@ pub struct Job<T> {
     pub data: T,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub request_id: Option<String>,
+    /// Unix epoch (seconds) when this job is intended to become available for
+    /// processing. Set by the producer when `scheduled_on` is provided; absent
+    /// for immediate jobs. Consumers use this to compute queue pickup latency
+    /// that excludes intentional scheduling delay.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub available_at: Option<String>,
 }
 
 impl<T> Job<T> {
@@ -35,12 +41,17 @@ impl<T> Job<T> {
             job_type,
             data,
             request_id: None,
+            available_at: None,
         }
     }
     pub fn with_request_id(mut self, id: Option<String>) -> Self {
         self.request_id = id;
         self
     }
+    pub fn with_scheduled_on(mut self, scheduled_on: Option<i64>) -> Self {
+        self.available_at = scheduled_on.map(|ts| ts.to_string());
+        self
+    }
 }
 
 // Enum to represent different message types
@@ -418,6 +429,35 @@ mod tests {
         assert_eq!(deserialized.data.relayer_id, "relayer-1");
     }
 
+    #[test]
+    fn test_job_with_scheduled_on_sets_available_at() {
+        let tx_request = TransactionRequest::new("tx123", "relayer-1");
+        let job = Job::new(JobType::TransactionRequest, tx_request).with_scheduled_on(Some(12345));
+
+        assert_eq!(job.available_at.as_deref(), Some("12345"));
+    }
+
+    #[test]
+    fn test_job_serialization_preserves_available_at() {
+        let tx_request = TransactionRequest::new("tx123", "relayer-1");
+        let job = Job::new(JobType::TransactionRequest, tx_request).with_scheduled_on(Some(12345));
+
+        let serialized = serde_json::to_string(&job).unwrap();
+        let deserialized: Job<TransactionRequest> = serde_json::from_str(&serialized).unwrap();
+
+        assert_eq!(deserialized.available_at.as_deref(), Some("12345"));
+    }
+
+    #[test]
+    fn test_job_serialization_omits_available_at_when_not_scheduled() {
+        let tx_request = TransactionRequest::new("tx123", "relayer-1");
+        let job = Job::new(JobType::TransactionRequest, tx_request);
+
+        let serialized = serde_json::to_string(&job).unwrap();
+
+        assert!(!serialized.contains("available_at"));
+    }
+
     #[test]
     fn test_notification_send_serialization() {
         let payload = WebhookPayload::Transaction(TransactionResponse::Evm(Box::new(

src/jobs/job_producer.rs

@@ -124,7 +124,8 @@ impl JobProducerTrait for JobProducer {
             transaction_process_job
         );
         let job = Job::new(JobType::TransactionRequest, transaction_process_job)
-            .with_request_id(get_request_id());
+            .with_request_id(get_request_id())
+            .with_scheduled_on(scheduled_on);
         let request_id = job.request_id.clone();
         let tx_id = job.data.transaction_id.clone();
         let relayer_id = job.data.relayer_id.clone();
@@ -155,7 +156,8 @@ impl JobProducerTrait for JobProducer {
         scheduled_on: Option<i64>,
     ) -> Result<(), JobProducerError> {
         let job = Job::new(JobType::TransactionSend, transaction_submit_job)
-            .with_request_id(get_request_id());
+            .with_request_id(get_request_id())
+            .with_scheduled_on(scheduled_on);
         let request_id = job.request_id.clone();
         let tx_id = job.data.transaction_id.clone();
         let relayer_id = job.data.relayer_id.clone();
@@ -191,7 +193,8 @@ impl JobProducerTrait for JobProducer {
             JobType::TransactionStatusCheck,
             transaction_status_check_job.clone(),
         )
-        .with_request_id(get_request_id());
+        .with_request_id(get_request_id())
+        .with_scheduled_on(scheduled_on);
         let request_id = job.request_id.clone();
         let tx_id = job.data.transaction_id.clone();
         let relayer_id = job.data.relayer_id.clone();
@@ -222,7 +225,8 @@ impl JobProducerTrait for JobProducer {
         scheduled_on: Option<i64>,
     ) -> Result<(), JobProducerError> {
         let job = Job::new(JobType::NotificationSend, notification_send_job)
-            .with_request_id(get_request_id());
+            .with_request_id(get_request_id())
+            .with_scheduled_on(scheduled_on);
         let request_id = job.request_id.clone();
         let notification_id = job.data.notification_id.clone();
 
@@ -249,8 +253,9 @@ impl JobProducerTrait for JobProducer {
         swap_request_job: TokenSwapRequest,
         scheduled_on: Option<i64>,
     ) -> Result<(), JobProducerError> {
-        let job =
-            Job::new(JobType::TokenSwapRequest, swap_request_job).with_request_id(get_request_id());
+        let job = Job::new(JobType::TokenSwapRequest, swap_request_job)
+            .with_request_id(get_request_id())
+            .with_scheduled_on(scheduled_on);
         let request_id = job.request_id.clone();
         let relayer_id = job.data.relayer_id.clone();
         let backend = self.queue_backend();
@@ -280,7 +285,8 @@ impl JobProducerTrait for JobProducer {
             JobType::RelayerHealthCheck,
             relayer_health_check_job.clone(),
         )
-        .with_request_id(get_request_id());
+        .with_request_id(get_request_id())
+        .with_scheduled_on(scheduled_on);
         let request_id = job.request_id.clone();
         let relayer_id = job.data.relayer_id.clone();
         let backend = self.queue_backend();
@@ -317,6 +323,8 @@ mod tests {
     struct TestRedisStorage<T> {
         pub push_called: bool,
         pub schedule_called: bool,
+        pub last_job: Option<T>,
+        pub last_scheduled_timestamp: Option<i64>,
         _phantom: std::marker::PhantomData<T>,
     }
 
@@ -325,17 +333,24 @@ mod tests {
             Self {
                 push_called: false,
                 schedule_called: false,
+                last_job: None,
+                last_scheduled_timestamp: None,
                 _phantom: std::marker::PhantomData,
             }
         }
+    }
 
-        async fn push(&mut self, _job: T) -> Result<(), JobProducerError> {
+    impl<T: Clone> TestRedisStorage<T> {
+        async fn push(&mut self, job: T) -> Result<(), JobProducerError> {
             self.push_called = true;
+            self.last_job = Some(job);
             Ok(())
         }
 
-        async fn schedule(&mut self, _job: T, _timestamp: i64) -> Result<(), JobProducerError> {
+        async fn schedule(&mut self, job: T, timestamp: i64) -> Result<(), JobProducerError> {
             self.schedule_called = true;
+            self.last_job = Some(job);
+            self.last_scheduled_timestamp = Some(timestamp);
             Ok(())
         }
     }
@@ -410,7 +425,8 @@ mod tests {
             scheduled_on: Option<i64>,
         ) -> Result<(), JobProducerError> {
             let mut queue = self.queue.lock().await;
-            let job = Job::new(JobType::TransactionRequest, transaction_process_job);
+            let job = Job::new(JobType::TransactionRequest, transaction_process_job)
+                .with_scheduled_on(scheduled_on);
 
             match scheduled_on {
                 Some(scheduled_on) => {
@@ -433,7 +449,8 @@ mod tests {
             scheduled_on: Option<i64>,
         ) -> Result<(), JobProducerError> {
             let mut queue = self.queue.lock().await;
-            let job = Job::new(JobType::TransactionSend, transaction_submit_job);
+            let job = Job::new(JobType::TransactionSend, transaction_submit_job)
+                .with_scheduled_on(scheduled_on);
 
             match scheduled_on {
                 Some(on) => {
@@ -456,7 +473,8 @@ mod tests {
             let job = Job::new(
                 JobType::TransactionStatusCheck,
                 transaction_status_check_job.clone(),
-            );
+            )
+            .with_scheduled_on(scheduled_on);
 
             // Route to the appropriate queue based on network type
             use crate::models::NetworkType;
@@ -485,7 +503,8 @@ mod tests {
             scheduled_on: Option<i64>,
         ) -> Result<(), JobProducerError> {
             let mut queue = self.queue.lock().await;
-            let job = Job::new(JobType::NotificationSend, notification_send_job);
+            let job = Job::new(JobType::NotificationSend, notification_send_job)
+                .with_scheduled_on(scheduled_on);
 
             match scheduled_on {
                 Some(on) => {
@@ -505,7 +524,8 @@ mod tests {
             scheduled_on: Option<i64>,
         ) -> Result<(), JobProducerError> {
             let mut queue = self.queue.lock().await;
-            let job = Job::new(JobType::TokenSwapRequest, swap_request_job);
+            let job = Job::new(JobType::TokenSwapRequest, swap_request_job)
+                .with_scheduled_on(scheduled_on);
 
             match scheduled_on {
                 Some(on) => {
@@ -525,7 +545,8 @@ mod tests {
             scheduled_on: Option<i64>,
         ) -> Result<(), JobProducerError> {
             let mut queue = self.queue.lock().await;
-            let job = Job::new(JobType::RelayerHealthCheck, relayer_health_check_job);
+            let job = Job::new(JobType::RelayerHealthCheck, relayer_health_check_job)
+                .with_scheduled_on(scheduled_on);
 
             match scheduled_on {
                 Some(scheduled_on) => {
@@ -841,6 +862,36 @@ mod tests {
         assert!(!queue.transaction_status_queue_evm.push_called);
     }
 
+    #[tokio::test]
+    async fn test_scheduled_status_check_persists_available_at() {
+        use crate::models::NetworkType;
+        let producer = TestJobProducer::new();
+
+        let status_job =
+            TransactionStatusCheck::new("tx-evm-scheduled", "relayer-1", NetworkType::Evm);
+        let scheduled_timestamp = calculate_scheduled_timestamp(30);
+        producer
+            .produce_check_transaction_status_job(status_job, Some(scheduled_timestamp))
+            .await
+            .unwrap();
+
+        let queue = producer.get_queue().await;
+        let stored_job = queue
+            .transaction_status_queue_evm
+            .last_job
+            .expect("scheduled status check should be stored");
+        let expected_available_at = scheduled_timestamp.to_string();
+
+        assert_eq!(
+            stored_job.available_at.as_deref(),
+            Some(expected_available_at.as_str())
+        );
+        assert_eq!(
+            queue.transaction_status_queue_evm.last_scheduled_timestamp,
+            Some(scheduled_timestamp)
+        );
+    }
+
     #[tokio::test]
     async fn test_submit_transaction_scheduled() {
         let producer = TestJobProducer::new();
@@ -857,6 +908,30 @@ mod tests {
         assert!(!queue.transaction_submission_queue.push_called);
     }
 
+    #[tokio::test]
+    async fn test_scheduled_submit_job_persists_available_at() {
+        let producer = TestJobProducer::new();
+
+        let submit_job = TransactionSend::submit("tx-scheduled", "relayer-1");
+        let scheduled_timestamp = calculate_scheduled_timestamp(15);
+        producer
+            .produce_submit_transaction_job(submit_job, Some(scheduled_timestamp))
+            .await
+            .unwrap();
+
+        let queue = producer.get_queue().await;
+        let stored_job = queue
+            .transaction_submission_queue
+            .last_job
+            .expect("scheduled submission should be stored");
+        let expected_available_at = scheduled_timestamp.to_string();
+
+        assert_eq!(
+            stored_job.available_at.as_deref(),
+            Some(expected_available_at.as_str())
+        );
+    }
+
     #[tokio::test]
     async fn test_notification_job_scheduled() {
         let producer = TestJobProducer::new();

src/metrics/mod.rs

@@ -21,6 +21,13 @@ pub fn observe_processing_time(relayer_id: &str, network_type: &str, stage: &str
         .with_label_values(&[relayer_id, network_type, stage])
         .observe(secs);
 }
+
+/// Observe queue pickup latency (time from send to consumer pickup).
+pub fn observe_queue_pickup_latency(queue_type: &str, backend: &str, secs: f64) {
+    QUEUE_PICKUP_LATENCY
+        .with_label_values(&[queue_type, backend])
+        .observe(secs);
+}
 use sysinfo::{Disks, System};
 
 lazy_static! {
@@ -206,6 +213,15 @@ lazy_static! {
         histogram_vec
     };
 
+    // Histogram for queue pickup latency (time from send to consumer pickup).
+    pub static ref QUEUE_PICKUP_LATENCY: HistogramVec = {
+        let histogram_opts = HistogramOpts::new("queue_pickup_latency_seconds", "Queue pickup latency in seconds (send to consumer pickup)")
+            .buckets(vec![0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 600.0, 1800.0, 3600.0, 14400.0, 86400.0]);
+        let histogram_vec = HistogramVec::new(histogram_opts, &["queue_type", "backend"]).unwrap();
+        REGISTRY.register(Box::new(histogram_vec.clone())).unwrap();
+        histogram_vec
+    };
+
     // Histogram for RPC call latency.
     pub static ref RPC_CALL_LATENCY: HistogramVec = {
         let histogram_opts = HistogramOpts::new("rpc_call_latency_seconds", "RPC call latency in seconds")
@@ -515,6 +531,9 @@ mod actix_tests {
             .with_label_values(&["test-relayer", "stellar"])
             .inc();
 
+        // Queue pickup latency
+        observe_queue_pickup_latency("transaction-request", "sqs", 0.5);
+
         let metrics = gather_metrics().expect("failed to gather metrics");
         let output = String::from_utf8(metrics).expect("metrics output is not valid UTF-8");
 
@@ -541,6 +560,9 @@ mod actix_tests {
         // TRY_AGAIN_LATER metrics
         assert!(output.contains("transactions_try_again_later_success_total"));
         assert!(output.contains("transactions_try_again_later_failed_total"));
+
+        // Queue pickup latency
+        assert!(output.contains("queue_pickup_latency_seconds"));
     }
 
     #[actix_rt::test]
@@ -786,4 +808,40 @@ mod processing_time_tests {
         let unique: std::collections::HashSet<&str> = stages.iter().copied().collect();
         assert_eq!(stages.len(), unique.len(), "stage constants must be unique");
     }
+
+    #[test]
+    fn test_observe_queue_pickup_latency_records_to_histogram() {
+        let before = QUEUE_PICKUP_LATENCY
+            .with_label_values(&["notification", "sqs"])
+            .get_sample_count();
+
+        observe_queue_pickup_latency("notification", "sqs", 1.5);
+
+        let after = QUEUE_PICKUP_LATENCY
+            .with_label_values(&["notification", "sqs"])
+            .get_sample_count();
+
+        assert_eq!(after, before + 1, "sample count should increase by 1");
+    }
+
+    #[test]
+    fn test_observe_queue_pickup_latency_both_backends() {
+        for backend in &["sqs", "redis"] {
+            let before = QUEUE_PICKUP_LATENCY
+                .with_label_values(&["relayer-health-check", backend])
+                .get_sample_count();
+
+            observe_queue_pickup_latency("relayer-health-check", backend, 0.25);
+
+            let after = QUEUE_PICKUP_LATENCY
+                .with_label_values(&["relayer-health-check", backend])
+                .get_sample_count();
+
+            assert_eq!(
+                after,
+                before + 1,
+                "sample count should increase by 1 for backend {backend}"
+            );
+        }
+    }
 }