diff --git a/.gitignore b/.gitignore index 446a966cc..7de123256 100644 --- a/.gitignore +++ b/.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 diff --git a/src/jobs/job.rs b/src/jobs/job.rs index c1d845db6..d89d6e721 100644 --- a/src/jobs/job.rs +++ b/src/jobs/job.rs @@ -24,6 +24,12 @@ pub struct Job { pub data: T, #[serde(skip_serializing_if = "Option::is_none")] pub request_id: Option, + /// 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, } impl Job { @@ -35,12 +41,17 @@ impl Job { job_type, data, request_id: None, + available_at: None, } } pub fn with_request_id(mut self, id: Option) -> Self { self.request_id = id; self } + pub fn with_scheduled_on(mut self, scheduled_on: Option) -> 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 = 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( diff --git a/src/jobs/job_producer.rs b/src/jobs/job_producer.rs index a2415309b..aad6cb029 100644 --- a/src/jobs/job_producer.rs +++ b/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, ) -> 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, ) -> 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, ) -> 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 { pub push_called: bool, pub schedule_called: bool, + pub last_job: Option, + pub last_scheduled_timestamp: Option, _phantom: std::marker::PhantomData, } @@ -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 TestRedisStorage { + 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, ) -> 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, ) -> 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, ) -> 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, ) -> 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, ) -> 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(); diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index 6cea71c4f..3e7afd340 100644 --- a/src/metrics/mod.rs +++ b/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}" + ); + } + } } diff --git a/src/queues/redis/worker.rs b/src/queues/redis/worker.rs index 8a216a628..7085020b4 100644 --- a/src/queues/redis/worker.rs +++ b/src/queues/redis/worker.rs @@ -45,6 +45,8 @@ use std::{str::FromStr, time::Duration}; use tokio::signal::unix::SignalKind; use tracing::{debug, error, info}; +use crate::metrics::observe_queue_pickup_latency; + use super::{filter_relayers_for_swap, QueueType, WorkerContext}; use crate::queues::retry_config::{ RetryBackoffConfig, NOTIFICATION_BACKOFF, RELAYER_HEALTH_BACKOFF, STATUS_EVM_BACKOFF, @@ -62,12 +64,71 @@ use crate::queues::retry_config::{ // keeping all handler business logic backend-neutral. // --------------------------------------------------------------------------- +/// Sanity threshold (ms) for non-scheduled latency observations. +/// +/// Latencies above this for a job whose only baseline was `Job.timestamp` +/// almost certainly indicate clock skew between producer and consumer rather +/// than a real backlog of that duration. We still observe the value (so the +/// `+Inf` bucket reflects reality), but log a warning so operators can detect +/// bad data instead of alerting on it. +const PICKUP_LATENCY_CLOCK_SKEW_THRESHOLD_MS: i64 = 60 * 60 * 1000; + +/// Observe queue pickup latency for Redis/Apalis workers. +/// +/// Uses `available_at` (the intended availability time) when present to exclude +/// intentional scheduling delay. Falls back to `timestamp` (job creation time) +/// for immediate jobs. Only records on the initial attempt — apalis +/// `Attempt::current()` is 1-indexed, so `attempt == 1` is the first delivery; +/// subsequent attempts would inflate the metric with retry backoff time. +/// +/// If the chosen baseline fails to parse, the alternative is tried so a single +/// corrupted field cannot silently drop the observation. +fn observe_redis_pickup_latency( + attempt: usize, + available_at: Option<&String>, + job_timestamp: &str, + queue_type: &str, +) { + if attempt != 1 { + return; + } + let baseline_epoch_secs = available_at + .and_then(|s| s.parse::().ok()) + .or_else(|| job_timestamp.parse::().ok()); + let Some(baseline_epoch_secs) = baseline_epoch_secs else { + tracing::warn!( + queue_type = queue_type, + available_at = ?available_at, + job_timestamp = %job_timestamp, + "skipping queue_pickup_latency: failed to parse both available_at and job_timestamp" + ); + return; + }; + let now_ms = chrono::Utc::now().timestamp_millis(); + let delta_ms = now_ms - baseline_epoch_secs * 1000; + if available_at.is_none() && delta_ms > PICKUP_LATENCY_CLOCK_SKEW_THRESHOLD_MS { + tracing::warn!( + queue_type = queue_type, + latency_ms = delta_ms, + "queue_pickup_latency above sanity threshold for non-scheduled job; check producer/consumer clock skew" + ); + } + let latency_secs = delta_ms.max(0) as f64 / 1000.0; + observe_queue_pickup_latency(queue_type, "redis", latency_secs); +} + async fn apalis_transaction_request_handler( job: Job, state: Data>, attempt: Attempt, task_id: TaskId, ) -> Result<(), apalis::prelude::Error> { + observe_redis_pickup_latency( + attempt.current(), + job.available_at.as_ref(), + &job.timestamp, + QueueType::TransactionRequest.queue_name(), + ); let ctx = WorkerContext::new(attempt.current(), task_id.to_string()); transaction_request_handler(job, (*state).clone(), ctx) .await @@ -80,6 +141,12 @@ async fn apalis_transaction_submission_handler( attempt: Attempt, task_id: TaskId, ) -> Result<(), apalis::prelude::Error> { + observe_redis_pickup_latency( + attempt.current(), + job.available_at.as_ref(), + &job.timestamp, + QueueType::TransactionSubmission.queue_name(), + ); let ctx = WorkerContext::new(attempt.current(), task_id.to_string()); transaction_submission_handler(job, (*state).clone(), ctx) .await @@ -92,6 +159,48 @@ async fn apalis_transaction_status_handler( attempt: Attempt, task_id: TaskId, ) -> Result<(), apalis::prelude::Error> { + observe_redis_pickup_latency( + attempt.current(), + job.available_at.as_ref(), + &job.timestamp, + QueueType::StatusCheck.queue_name(), + ); + let ctx = WorkerContext::new(attempt.current(), task_id.to_string()); + transaction_status_handler(job, (*state).clone(), ctx) + .await + .map_err(Into::into) +} + +async fn apalis_transaction_status_evm_handler( + job: Job, + state: Data>, + attempt: Attempt, + task_id: TaskId, +) -> Result<(), apalis::prelude::Error> { + observe_redis_pickup_latency( + attempt.current(), + job.available_at.as_ref(), + &job.timestamp, + QueueType::StatusCheckEvm.queue_name(), + ); + let ctx = WorkerContext::new(attempt.current(), task_id.to_string()); + transaction_status_handler(job, (*state).clone(), ctx) + .await + .map_err(Into::into) +} + +async fn apalis_transaction_status_stellar_handler( + job: Job, + state: Data>, + attempt: Attempt, + task_id: TaskId, +) -> Result<(), apalis::prelude::Error> { + observe_redis_pickup_latency( + attempt.current(), + job.available_at.as_ref(), + &job.timestamp, + QueueType::StatusCheckStellar.queue_name(), + ); let ctx = WorkerContext::new(attempt.current(), task_id.to_string()); transaction_status_handler(job, (*state).clone(), ctx) .await @@ -104,6 +213,12 @@ async fn apalis_notification_handler( attempt: Attempt, task_id: TaskId, ) -> Result<(), apalis::prelude::Error> { + observe_redis_pickup_latency( + attempt.current(), + job.available_at.as_ref(), + &job.timestamp, + QueueType::Notification.queue_name(), + ); let ctx = WorkerContext::new(attempt.current(), task_id.to_string()); notification_handler(job, (*state).clone(), ctx) .await @@ -116,6 +231,12 @@ async fn apalis_token_swap_request_handler( attempt: Attempt, task_id: TaskId, ) -> Result<(), apalis::prelude::Error> { + observe_redis_pickup_latency( + attempt.current(), + job.available_at.as_ref(), + &job.timestamp, + QueueType::TokenSwapRequest.queue_name(), + ); let ctx = WorkerContext::new(attempt.current(), task_id.to_string()); token_swap_request_handler(job, (*state).clone(), ctx) .await @@ -128,6 +249,12 @@ async fn apalis_relayer_health_check_handler( attempt: Attempt, task_id: TaskId, ) -> Result<(), apalis::prelude::Error> { + observe_redis_pickup_latency( + attempt.current(), + job.available_at.as_ref(), + &job.timestamp, + QueueType::RelayerHealthCheck.queue_name(), + ); let ctx = WorkerContext::new(attempt.current(), task_id.to_string()); relayer_health_check_handler(job, (*state).clone(), ctx) .await @@ -306,7 +433,7 @@ where )) .data(app_state.clone()) .backend(queue.transaction_status_queue_evm.clone()) - .build_fn(apalis_transaction_status_handler); + .build_fn(apalis_transaction_status_evm_handler); // Stellar status checker - fast retries for fast finality // Stellar has sub-second finality, needs more frequent status checks @@ -326,7 +453,7 @@ where )) .data(app_state.clone()) .backend(queue.transaction_status_queue_stellar.clone()) - .build_fn(apalis_transaction_status_handler); + .build_fn(apalis_transaction_status_stellar_handler); let notification_queue_worker = WorkerBuilder::new(NOTIFICATION_SENDER) .layer(ErrorHandlingLayer::new()) @@ -937,4 +1064,98 @@ mod tests { fn test_monitor_handle_event_custom_does_not_panic() { monitor_handle_event(make_worker_event(Event::Custom("test-custom".to_string()))); } + + // ── observe_redis_pickup_latency tests ───────────────────────────── + + fn pickup_sample_count(queue_type: &str) -> u64 { + crate::metrics::QUEUE_PICKUP_LATENCY + .with_label_values(&[queue_type, "redis"]) + .get_sample_count() + } + + #[test] + fn test_observe_redis_pickup_latency_records_on_first_attempt() { + let queue = "test-pickup-first-attempt"; + let before = pickup_sample_count(queue); + + let ts = chrono::Utc::now().timestamp().to_string(); + observe_redis_pickup_latency(1, None, &ts, queue); + + assert_eq!(pickup_sample_count(queue), before + 1); + } + + #[test] + fn test_observe_redis_pickup_latency_skips_retry_attempts() { + let queue = "test-pickup-skip-retry"; + let before = pickup_sample_count(queue); + + let ts = chrono::Utc::now().timestamp().to_string(); + observe_redis_pickup_latency(2, None, &ts, queue); + observe_redis_pickup_latency(99, None, &ts, queue); + + assert_eq!(pickup_sample_count(queue), before); + } + + #[test] + fn test_observe_redis_pickup_latency_prefers_available_at_over_timestamp() { + // available_at is "now" so latency should be ~0; job_timestamp is far in the + // past, so if the fallback were used we'd see a large value. We verify the + // preference indirectly via the histogram sum delta. + let queue = "test-pickup-prefers-available-at"; + let histogram = crate::metrics::QUEUE_PICKUP_LATENCY.with_label_values(&[queue, "redis"]); + let sum_before = histogram.get_sample_sum(); + + let now = chrono::Utc::now().timestamp(); + let available_at = now.to_string(); + let stale_timestamp = (now - 3600).to_string(); + + observe_redis_pickup_latency(1, Some(&available_at), &stale_timestamp, queue); + + let delta = histogram.get_sample_sum() - sum_before; + assert!( + delta < 5.0, + "expected near-zero latency when available_at is now, got {delta}" + ); + } + + #[test] + fn test_observe_redis_pickup_latency_falls_back_to_timestamp_when_available_at_absent() { + let queue = "test-pickup-fallback-timestamp"; + let before = pickup_sample_count(queue); + + let ts = chrono::Utc::now().timestamp().to_string(); + observe_redis_pickup_latency(1, None, &ts, queue); + + assert_eq!(pickup_sample_count(queue), before + 1); + } + + #[test] + fn test_observe_redis_pickup_latency_clamps_negative_skew_to_zero() { + // baseline in the future (producer clock ahead of consumer) → delta negative + // → clamped to 0, but the observation still records. + let queue = "test-pickup-clamps-negative"; + let histogram = crate::metrics::QUEUE_PICKUP_LATENCY.with_label_values(&[queue, "redis"]); + let sum_before = histogram.get_sample_sum(); + let count_before = histogram.get_sample_count(); + + let future_ts = (chrono::Utc::now().timestamp() + 3600).to_string(); + observe_redis_pickup_latency(1, None, &future_ts, queue); + + assert_eq!(histogram.get_sample_count(), count_before + 1); + let delta_sum = histogram.get_sample_sum() - sum_before; + assert!( + delta_sum.abs() < f64::EPSILON, + "expected 0 latency for future baseline, got {delta_sum}" + ); + } + + #[test] + fn test_observe_redis_pickup_latency_skips_when_baseline_unparsable() { + let queue = "test-pickup-unparsable"; + let before = pickup_sample_count(queue); + + observe_redis_pickup_latency(1, None, "not-a-number", queue); + + assert_eq!(pickup_sample_count(queue), before); + } } diff --git a/src/queues/sqs/worker.rs b/src/queues/sqs/worker.rs index 5bf272b90..68db4c976 100644 --- a/src/queues/sqs/worker.rs +++ b/src/queues/sqs/worker.rs @@ -19,6 +19,7 @@ use tokio::sync::watch; use tokio::task::{JoinHandle, JoinSet}; use tracing::{debug, error, info, warn}; +use crate::metrics::observe_queue_pickup_latency; use crate::queues::{backoff_config_for_queue, retry_delay_secs}; use crate::{ config::ServerConfig, @@ -231,6 +232,7 @@ async fn run_poll_loop( .visibility_timeout(visibility_timeout as i32) .message_system_attribute_names(MessageSystemAttributeName::ApproximateReceiveCount) .message_system_attribute_names(MessageSystemAttributeName::MessageGroupId) + .message_system_attribute_names(MessageSystemAttributeName::SentTimestamp) .message_attribute_names("target_scheduled_on") .message_attribute_names("retry_attempt") .send() => result, @@ -440,6 +442,42 @@ async fn process_message( .receipt_handle() .ok_or_else(|| QueueBackendError::QueueError("Missing receipt handle".to_string()))?; + // Observe queue pickup latency on the FIRST physical delivery, before the + // defer block consumes it. Placement here is deliberate: + // - Standard queues hold scheduled messages invisible via DelaySeconds + // and deliver at ~target_scheduled_on, so latency reflects actual + // sub-second pickup delay. + // - FIFO queues deliver scheduled messages immediately (no native + // DelaySeconds) and the consumer then defers via visibility timeout. + // The negative `now - target_scheduled_on` clamps to 0, which honestly + // says "consumer is keeping up with the schedule". + // Either way: receive_count==1 in `queue_pickup_baseline_ms` ensures we + // observe exactly once per logical message lifecycle. FIFO defer/retry + // re-deliveries (which bump receive_count) are skipped; standard-queue + // retries are skipped via the `retry_attempt` attribute. + if let Some(baseline) = queue_pickup_baseline_ms(&message) { + let now_ms = chrono::Utc::now().timestamp_millis(); + // SentTimestamp is set by the AWS broker; if the consumer clock runs + // ahead of the broker by more than this threshold for a non-scheduled + // message, the latency is almost certainly clock skew, not a real + // backlog. Log so operators can detect bad data rather than alert on it. + let delta_ms = now_ms - baseline; + if parse_target_scheduled_on(&message).is_none() + && delta_ms > PICKUP_LATENCY_CLOCK_SKEW_THRESHOLD_MS + { + warn!( + queue_type = ?queue_type, + latency_ms = delta_ms, + "queue_pickup_latency above sanity threshold for non-scheduled SQS message; check broker/consumer clock skew" + ); + } + observe_queue_pickup_latency( + queue_type.queue_name(), + "sqs", + pickup_latency_secs(baseline, now_ms), + ); + } + // For jobs with scheduling beyond SQS 15-minute max delay, keep deferring in hops. if let Some(target_scheduled_on) = parse_target_scheduled_on(&message) { let now = std::time::SystemTime::now() @@ -744,6 +782,71 @@ fn parse_retry_attempt(message: &Message) -> Option { .and_then(|value| value.parse::().ok()) } +/// Compute pickup latency in seconds, clamping negative deltas to 0 so a +/// consumer clock running ahead of the AWS broker (or a future-dated +/// `target_scheduled_on`) cannot produce a negative-then-huge cast value. +fn pickup_latency_secs(baseline_ms: i64, now_ms: i64) -> f64 { + (now_ms - baseline_ms).max(0) as f64 / 1000.0 +} + +/// Sanity threshold (ms) for non-scheduled latency observations. Above this, +/// the consumer clock is almost certainly skewed relative to the AWS broker +/// rather than the queue genuinely being backed up by an hour+. Used only to +/// emit a warning — the value is still observed in the histogram. +const PICKUP_LATENCY_CLOCK_SKEW_THRESHOLD_MS: i64 = 60 * 60 * 1000; + +fn queue_pickup_baseline_ms(message: &Message) -> Option { + // Observe pickup latency only on the very first physical delivery of + // a message. The gate is intentionally narrow because the relayer + // supports both standard and FIFO SQS queues — and FIFO defer-hops and + // error retries both reuse the same physical message via + // `change_message_visibility`, which cannot mutate the `retry_attempt` + // attribute. Without the receive-count gate, every FIFO redelivery + // would re-observe the latency, conflating the metric with retry + // backoff time. + // + // The trade-offs: + // - Standard queues: behaves correctly. Initial delivery has + // receive_count=1; standard-queue retries re-send a new message + // (also receive_count=1) but carry `retry_attempt > 0`, so the + // second check below skips them. + // - Standard queues with defer-hop (only triggered when + // scheduled_on - now > 900s): each defer-hop creates a new message + // with receive_count=1 and no `retry_attempt`, so the metric + // observes each hop. This is an accepted limitation; long-delay + // scheduling is rare in this codebase (status checks use seconds- + // scale backoff). + // - FIFO queues: only the very first delivery observes. For + // scheduled jobs that arrive before `target_scheduled_on`, the + // computed latency is clamped to 0 by the caller. This is narrower + // than the standard-queue semantic but consistent and free of + // retry inflation. + let receive_count = message + .attributes() + .and_then(|attrs| attrs.get(&MessageSystemAttributeName::ApproximateReceiveCount)) + .and_then(|count| count.parse::().ok()) + .unwrap_or(1); + if receive_count != 1 { + return None; + } + + // Standard-queue retries re-enqueue as new messages with receive_count=1 + // and an explicit `retry_attempt` attribute. Skip those so the metric + // doesn't include retry backoff time. + if parse_retry_attempt(message).is_some_and(|n| n > 0) { + return None; + } + + parse_target_scheduled_on(message) + .map(|ts_secs| ts_secs * 1000) + .or_else(|| { + message + .attributes() + .and_then(|a| a.get(&MessageSystemAttributeName::SentTimestamp)) + .and_then(|v| v.parse::().ok()) + }) +} + fn is_fifo_queue_url(queue_url: &str) -> bool { queue_url.ends_with(".fifo") } @@ -1419,6 +1522,159 @@ mod tests { assert_eq!(parse_retry_attempt(&message), Some(999999)); } + #[test] + fn test_queue_pickup_baseline_ms_uses_scheduled_time_on_first_delivery() { + let message = Message::builder() + .message_attributes( + "target_scheduled_on", + MessageAttributeValue::builder() + .data_type("Number") + .string_value("123") + .build() + .unwrap(), + ) + .set_attributes(Some(std::collections::HashMap::from([( + MessageSystemAttributeName::SentTimestamp, + "999999".to_string(), + )]))) + .build(); + + // No retry_attempt attribute → first attempt + assert_eq!(queue_pickup_baseline_ms(&message), Some(123_000)); + } + + #[test] + fn test_queue_pickup_baseline_ms_falls_back_to_sent_timestamp() { + let message = Message::builder() + .set_attributes(Some(std::collections::HashMap::from([( + MessageSystemAttributeName::SentTimestamp, + "123456".to_string(), + )]))) + .build(); + + assert_eq!(queue_pickup_baseline_ms(&message), Some(123456)); + } + + #[test] + fn test_pickup_latency_secs_clamps_negative_skew() { + // Consumer clock running 5s behind the broker (baseline is "in the future") + // must not produce a negative-then-huge cast value — clamp to 0. + let now_ms = 1_000_000_i64; + let baseline_ms = now_ms + 5_000; + assert_eq!(pickup_latency_secs(baseline_ms, now_ms), 0.0); + } + + #[test] + fn test_pickup_latency_secs_positive_delta() { + // 2.5s positive delta should be reported in seconds with ms precision. + assert_eq!(pickup_latency_secs(1_000_000, 1_002_500), 2.5); + } + + #[test] + fn test_queue_pickup_baseline_ms_skips_when_retry_attempt_positive() { + let message = Message::builder() + .message_attributes( + "target_scheduled_on", + MessageAttributeValue::builder() + .data_type("Number") + .string_value("123") + .build() + .unwrap(), + ) + .message_attributes( + "retry_attempt", + MessageAttributeValue::builder() + .data_type("Number") + .string_value("1") + .build() + .unwrap(), + ) + .set_attributes(Some(std::collections::HashMap::from([( + MessageSystemAttributeName::SentTimestamp, + "123456".to_string(), + )]))) + .build(); + + // retry_attempt > 0 → genuine retry, skip + assert_eq!(queue_pickup_baseline_ms(&message), None); + } + + #[test] + fn test_queue_pickup_baseline_ms_accepts_retry_attempt_zero() { + // Explicit retry_attempt=0 should be treated as first attempt + // (consistent with absent attribute). + let message = Message::builder() + .message_attributes( + "target_scheduled_on", + MessageAttributeValue::builder() + .data_type("Number") + .string_value("777") + .build() + .unwrap(), + ) + .message_attributes( + "retry_attempt", + MessageAttributeValue::builder() + .data_type("Number") + .string_value("0") + .build() + .unwrap(), + ) + .build(); + + assert_eq!(queue_pickup_baseline_ms(&message), Some(777_000)); + } + + #[test] + fn test_queue_pickup_baseline_ms_skips_when_receive_count_gt_one() { + // Receive count > 1 means the message has been delivered before, + // which on FIFO queues happens for both scheduling defer-hops and + // error retries — neither of which we want to record as a fresh + // pickup. Gating on receive_count == 1 prevents this conflation. + let message = Message::builder() + .message_attributes( + "target_scheduled_on", + MessageAttributeValue::builder() + .data_type("Number") + .string_value("500") + .build() + .unwrap(), + ) + .set_attributes(Some(std::collections::HashMap::from([ + ( + MessageSystemAttributeName::ApproximateReceiveCount, + "2".to_string(), + ), + (MessageSystemAttributeName::SentTimestamp, "999".to_string()), + ]))) + .build(); + + assert_eq!(queue_pickup_baseline_ms(&message), None); + } + + #[test] + fn test_queue_pickup_baseline_ms_observes_when_receive_count_explicitly_one() { + // Explicit receive_count=1 (first physical delivery) should be + // observed, mirroring the implicit default when the attribute is + // missing. + let message = Message::builder() + .message_attributes( + "target_scheduled_on", + MessageAttributeValue::builder() + .data_type("Number") + .string_value("250") + .build() + .unwrap(), + ) + .set_attributes(Some(std::collections::HashMap::from([( + MessageSystemAttributeName::ApproximateReceiveCount, + "1".to_string(), + )]))) + .build(); + + assert_eq!(queue_pickup_baseline_ms(&message), Some(250_000)); + } + // ── is_fifo_queue_url: comprehensive cases ──────────────────────── #[test]