diff --git a/queue/src/adapters/rabbitmq/adapter.rs b/queue/src/adapters/rabbitmq/adapter.rs index f9c0bcf14..e32c06aeb 100644 --- a/queue/src/adapters/rabbitmq/adapter.rs +++ b/queue/src/adapters/rabbitmq/adapter.rs @@ -117,14 +117,48 @@ struct SubscriptionInfo { impl RabbitMQAdapter { /// Resolve the DLQ queue name for a given topic. Handles both function /// queues (`__fn_queue::` prefix) and topic-based queues. - fn resolve_dlq_name(topic: &str) -> (String, bool) { + fn resolve_dlq_name(topic: &str, is_function_queue: bool) -> (String, bool) { if let Some(queue_name) = topic.strip_prefix("__fn_queue::") { (FnQueueNames::new(queue_name).dlq(), true) + } else if is_function_queue { + (FnQueueNames::new(topic).dlq(), true) } else { (RabbitNames::new(topic).dlq(), false) } } + async fn resolve_dlq_name_for_topic(&self, topic: &str) -> (String, bool) { + let is_function_queue = self.function_queue_configs.read().await.contains_key(topic); + Self::resolve_dlq_name(topic, is_function_queue) + } + + async fn operation_channel(&self, operation: &str) -> anyhow::Result { + self.connection + .create_channel() + .await + .map_err(|e| anyhow::anyhow!("Failed to create channel for {}: {}", operation, e)) + } + + async fn passive_queue_count(&self, queue_name: &str) -> u64 { + let Ok(channel) = self.operation_channel("queue inspection").await else { + return 0; + }; + match channel + .queue_declare( + queue_name, + lapin::options::QueueDeclareOptions { + passive: true, + ..Default::default() + }, + lapin::types::FieldTable::default(), + ) + .await + { + Ok(info) => info.message_count() as u64, + Err(_) => 0, + } + } + /// Scan a DLQ for a specific message by ID, applying `on_found` when the /// target is located. Non-target messages are nacked back to the queue. /// @@ -143,10 +177,10 @@ impl RabbitMQAdapter { F: FnOnce(&Delivery, bool, &str) -> Fut, Fut: std::future::Future>, { - let (dlq_name, is_fn_queue) = Self::resolve_dlq_name(topic); + let (dlq_name, is_fn_queue) = self.resolve_dlq_name_for_topic(topic).await; + let channel = self.operation_channel("DLQ message lookup").await?; - let queue_info = self - .channel + let queue_info = channel .queue_declare( &dlq_name, lapin::options::QueueDeclareOptions { @@ -161,8 +195,7 @@ impl RabbitMQAdapter { let count = queue_info.message_count(); for _ in 0..count { - let get_result = self - .channel + let get_result = channel .basic_get(&dlq_name, BasicGetOptions { no_ack: false }) .await .map_err(|e| anyhow::anyhow!("Failed to get message from DLQ: {}", e))?; @@ -489,12 +522,12 @@ impl QueueAdapter for RabbitMQAdapter { } async fn redrive_dlq(&self, topic: &str) -> anyhow::Result { - let (dlq_name, is_fn_queue) = Self::resolve_dlq_name(topic); + let (dlq_name, is_fn_queue) = self.resolve_dlq_name_for_topic(topic).await; + let channel = self.operation_channel("DLQ redrive").await?; let mut count: u64 = 0; loop { - let get_result = self - .channel + let get_result = channel .basic_get(&dlq_name, BasicGetOptions { no_ack: false }) .await .map_err(|e| anyhow::anyhow!("Failed to get message from DLQ: {}", e))?; @@ -503,7 +536,7 @@ impl QueueAdapter for RabbitMQAdapter { Some(delivery) => { let republish_result: anyhow::Result<()> = if is_fn_queue { // Function queue DLQ: raw data payload, republish directly - let queue_name = topic.strip_prefix("__fn_queue::").unwrap(); + let queue_name = topic.strip_prefix("__fn_queue::").unwrap_or(topic); let names = FnQueueNames::new(queue_name); let mut headers = delivery @@ -526,7 +559,7 @@ impl QueueAdapter for RabbitMQAdapter { properties }; - self.channel + channel .basic_publish( &names.exchange(), queue_name, @@ -674,10 +707,10 @@ impl QueueAdapter for RabbitMQAdapter { async fn dlq_count(&self, topic: &str) -> anyhow::Result { // Function queues use FnQueueNames (e.g., __fn_queue::orders -> ::dlq.queue), // while topic-based queues use RabbitNames (e.g., user.created -> .dlq). - let (dlq_name, _is_fn_queue) = Self::resolve_dlq_name(topic); + let (dlq_name, _is_fn_queue) = self.resolve_dlq_name_for_topic(topic).await; + let channel = self.operation_channel("DLQ count").await?; - let queue = self - .channel + let queue = channel .queue_declare( &dlq_name, lapin::options::QueueDeclareOptions { @@ -697,10 +730,10 @@ impl QueueAdapter for RabbitMQAdapter { /// the same fields the engine's `DlqMessage` struct carries as a JSON /// object per entry instead of a dedicated type. async fn dlq_peek(&self, topic: &str, offset: u64, limit: u64) -> anyhow::Result> { - let (dlq_name, is_fn_queue) = Self::resolve_dlq_name(topic); + let (dlq_name, is_fn_queue) = self.resolve_dlq_name_for_topic(topic).await; + let channel = self.operation_channel("DLQ browse").await?; - let queue_depth = match self - .channel + let queue_depth = match channel .queue_declare( &dlq_name, lapin::options::QueueDeclareOptions { @@ -720,8 +753,7 @@ impl QueueAdapter for RabbitMQAdapter { let mut deliveries_to_nack = Vec::new(); for i in 0..fetch_count { - let get_result = self - .channel + let get_result = channel .basic_get(&dlq_name, BasicGetOptions { no_ack: false }) .await .map_err(|e| anyhow::anyhow!("Failed to get message from DLQ: {}", e))?; @@ -1322,53 +1354,22 @@ impl QueueAdapter for RabbitMQAdapter { /// dropped (no field for it in this worker's `TopicStats`); /// `delivered`/`failed` are always `0` (not tracked by this adapter). /// - /// Inherits one engine quirk verbatim: the "main queue depth" is looked - /// up via `RabbitNames::new(topic).queue()` (`iii.{topic}.queue`), a - /// queue name that topic-fanout subscriptions never actually declare - /// (`subscribe` declares per-function queues via - /// `RabbitNames::function_queue`, never the bare `.queue()` name) -- so - /// for topic-based (non-`__fn_queue::`) topics this always resolves to - /// depth `0` in both the engine and this port. async fn topic_stats(&self, topic: &str) -> anyhow::Result { + let is_configured_function_queue = + self.function_queue_configs.read().await.contains_key(topic); let (queue_name, dlq_name) = if let Some(name) = topic.strip_prefix("__fn_queue::") { let names = FnQueueNames::new(name); (names.queue(), names.dlq()) + } else if is_configured_function_queue { + let names = FnQueueNames::new(topic); + (names.queue(), names.dlq()) } else { let names = RabbitNames::new(topic); (names.queue(), names.dlq()) }; - let depth = match self - .channel - .queue_declare( - &queue_name, - lapin::options::QueueDeclareOptions { - passive: true, - ..Default::default() - }, - lapin::types::FieldTable::default(), - ) - .await - { - Ok(info) => info.message_count() as u64, - Err(_) => 0, - }; - - let dlq_depth = match self - .channel - .queue_declare( - &dlq_name, - lapin::options::QueueDeclareOptions { - passive: true, - ..Default::default() - }, - lapin::types::FieldTable::default(), - ) - .await - { - Ok(info) => info.message_count() as u64, - Err(_) => 0, - }; + let depth = self.passive_queue_count(&queue_name).await; + let dlq_depth = self.passive_queue_count(&dlq_name).await; Ok(TopicStats { depth, @@ -1498,3 +1499,16 @@ async fn requeue_rabbitmq_deliveries( anyhow::bail!(errors.join("; ")) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn bare_function_queue_name_resolves_to_function_dlq() { + assert_eq!( + RabbitMQAdapter::resolve_dlq_name("tax-returns", true), + ("iii.__fn_queue::tax-returns::dlq.queue".to_string(), true,) + ); + } +} diff --git a/queue/src/functions.rs b/queue/src/functions.rs index 876904990..3dd733c1f 100644 --- a/queue/src/functions.rs +++ b/queue/src/functions.rs @@ -120,7 +120,7 @@ fn default_dlq_limit() -> u64 { 50 } -#[derive(Debug, Clone, Serialize, JsonSchema, PartialEq)] +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq)] pub struct DlqMessage { pub id: String, pub payload: Value, @@ -130,6 +130,13 @@ pub struct DlqMessage { pub size_bytes: u64, } +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum AdapterDlqMessage { + Normalized(DlqMessage), + Job(Job), +} + pub fn register_all( iii: &Arc, adapter: Arc, @@ -378,8 +385,12 @@ pub async fn dlq_messages( .into_iter() .skip(input.offset as usize) .take(input.limit as usize) - .filter_map(|value| serde_json::from_value::(value).ok()) - .map(dlq_message_from_job) + .filter_map( + |value| match serde_json::from_value::(value).ok()? { + AdapterDlqMessage::Normalized(message) => Some(message), + AdapterDlqMessage::Job(job) => Some(dlq_message_from_job(job)), + }, + ) .collect(); Ok(messages) } @@ -783,6 +794,42 @@ mod tests { ); } + #[tokio::test] + async fn dlq_browse_preserves_normalized_messages() { + let (adapter, mock) = adapter(); + *mock.dlq_messages_result.lock().unwrap() = Some(Ok(vec![json!({ + "id": "m1", + "payload": {"dead": true}, + "error": "Function tax::process exhausted retries", + "failed_at": 123, + "retries": 1, + "size_bytes": 13 + })])); + + let messages = dlq_messages( + adapter, + DlqMessagesInput { + topic: "__fn_queue::tax-returns".to_string(), + offset: 0, + limit: 50, + }, + ) + .await + .unwrap(); + + assert_eq!( + messages, + vec![DlqMessage { + id: "m1".to_string(), + payload: json!({"dead": true}), + error: "Function tax::process exhausted retries".to_string(), + failed_at: 123, + retries: 1, + size_bytes: 13, + }] + ); + } + #[tokio::test] async fn dlq_messages_rejects_empty_topic() { let (adapter, _mock) = adapter(); diff --git a/queue/tests/e2e_rabbitmq.rs b/queue/tests/e2e_rabbitmq.rs index 258c9571c..34ca8f57f 100644 --- a/queue/tests/e2e_rabbitmq.rs +++ b/queue/tests/e2e_rabbitmq.rs @@ -202,24 +202,9 @@ async fn basic_delivery_connect_or_skip() { /// Exercised through the function-queue trait methods /// (`setup_function_queue`/`publish_to_function_queue`/ /// `consume_function_queue`/`ack_function_queue`/`nack_function_queue`) -/// rather than through `subscribe`'s topic-fanout path. Discovered while -/// writing this test: `RabbitMQAdapter::resolve_dlq_name` (and therefore -/// `dlq_count`/`redrive_dlq`/`redrive_dlq_message`/`discard_dlq_message`) -/// resolves a bare (non-`__fn_queue::`) topic to `RabbitNames::dlq()` -/// (`iii.{topic}.dlq`) -- a queue name that `subscribe`'s -/// `setup_subscriber_queue` never actually declares (it declares -/// `RabbitNames::function_dlq(function_id)`, i.e. -/// `iii.{topic}.{function_id}.dlq`, a *different* name). That mismatch is -/// inherited verbatim from the engine -/// (`engine/src/workers/queue/adapters/rabbitmq/adapter.rs`) -- the engine's -/// own `rabbitmq_queue_integration.rs` test suite never calls -/// `dlq_count`/`redrive_dlq` against a bare subscribed topic either (only -/// against `__fn_queue::`-prefixed function-queue names, which resolve -/// correctly). A passive `queue_declare` against a queue that doesn't exist -/// is a channel-level AMQP error, which would poison this adapter's shared -/// channel for every other operation -- so this test deliberately exercises -/// the function-queue path, which resolves correctly end-to-end, instead of -/// tripping that pre-existing engine gap. +/// rather than through `subscribe`'s topic-fanout path. DLQ operations use +/// the same bare function-queue name returned by `list_topics`; the adapter +/// must resolve it to the internal `__fn_queue::` RabbitMQ topology. #[tokio::test] #[serial] async fn function_queue_retry_then_dlq_then_redrive_connect_or_skip() { @@ -234,7 +219,6 @@ async fn function_queue_retry_then_dlq_then_redrive_connect_or_skip() { .expect("rabbitmq adapter should connect"); let queue_name = format!("e2e-rmq-fnq-{}", Uuid::new_v4()); - let dlq_topic = format!("__fn_queue::{queue_name}"); adapter .setup_function_queue( @@ -292,20 +276,24 @@ async fn function_queue_retry_then_dlq_then_redrive_connect_or_skip() { wait_until( || { let adapter = &adapter; - let dlq_topic = dlq_topic.clone(); - async move { adapter.dlq_count(&dlq_topic).await.unwrap_or(0) >= 1 } + let queue_name = queue_name.clone(); + async move { adapter.dlq_count(&queue_name).await.unwrap_or(0) >= 1 } }, Duration::from_secs(5), ) .await; - assert_eq!(adapter.dlq_count(&dlq_topic).await.unwrap(), 1); + assert_eq!(adapter.dlq_count(&queue_name).await.unwrap(), 1); + + let messages = adapter.dlq_messages(&queue_name, 10).await.unwrap(); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0]["payload"], json!({"n": 1})); let redriven = adapter - .redrive_dlq(&dlq_topic) + .redrive_dlq(&queue_name) .await .expect("redrive_dlq should succeed"); assert_eq!(redriven, 1); - assert_eq!(adapter.dlq_count(&dlq_topic).await.unwrap(), 0); + assert_eq!(adapter.dlq_count(&queue_name).await.unwrap(), 0); let msg3 = tokio::time::timeout(Duration::from_secs(5), rx.recv()) .await @@ -320,6 +308,41 @@ async fn function_queue_retry_then_dlq_then_redrive_connect_or_skip() { adapter.shutdown().await; } +/// A missing passive declaration is an AMQP channel-level error. DLQ +/// inspection must isolate that failure from the channel used by consumers. +#[tokio::test] +#[serial] +async fn missing_dlq_inspection_does_not_close_consumer_channel() { + let container = docker::start_rabbitmq() + .await + .expect("Docker and a healthy RabbitMQ container are required for this E2E test"); + + let invoker: Arc = Arc::new(NoopInvoker); + let adapter = + RabbitMQAdapter::from_config(Some(&json!({"amqp_url": container.amqp_url()})), invoker) + .await + .expect("rabbitmq adapter should connect"); + + let missing_topic = format!("e2e-rmq-missing-{}", Uuid::new_v4()); + assert!( + adapter.dlq_count(&missing_topic).await.is_err(), + "missing DLQ inspection should report the broker error" + ); + + let queue_name = format!("e2e-rmq-after-missing-{}", Uuid::new_v4()); + adapter + .setup_function_queue(&queue_name, &FunctionQueueConfig::default()) + .await + .expect("function queue topology should still be configurable"); + let receiver = adapter + .consume_function_queue(&queue_name, 1) + .await + .expect("consumer channel should remain open after failed DLQ inspection"); + + drop(receiver); + adapter.shutdown().await; +} + /// (c) Priority ordering: 3 messages, published with `priority_field` /// values 1, 9, 5 -- in that order -- onto a subscriber queue declared with /// `max_priority: 10`, all BEFORE any consumer attaches (the topology is