From a9a8fd8af3bfdff33da61c992d9cf37f482d357b Mon Sep 17 00:00:00 2001 From: giria660 Date: Mon, 6 Apr 2026 20:40:12 -0400 Subject: [PATCH 1/3] Deduplicate function queue DLQ topics --- engine/src/workers/queue/queue.rs | 39 ++++++++++++++++++++++++++++--- 1 file changed, 36 insertions(+), 3 deletions(-) diff --git a/engine/src/workers/queue/queue.rs b/engine/src/workers/queue/queue.rs index 457929e047..fbd8c7738c 100644 --- a/engine/src/workers/queue/queue.rs +++ b/engine/src/workers/queue/queue.rs @@ -5,7 +5,7 @@ // See LICENSE and PATENTS files for details. use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, pin::Pin, sync::{Arc, RwLock}, }; @@ -434,16 +434,29 @@ impl QueueWorker { match self.adapter.list_topics().await { Ok(topics) => { let mut dlq_topics = Vec::new(); + let mut seen = HashSet::new(); for topic in &topics { + let (display_name, broker_type) = + if let Some(stripped) = topic.name.strip_prefix("__fn_queue::") { + (stripped.to_string(), "function_queue".to_string()) + } else { + (topic.name.clone(), topic.broker_type.clone()) + }; + if !seen.insert(display_name.clone()) { + continue; + } let dlq_count = self.adapter.dlq_count(&topic.name).await.unwrap_or(0); dlq_topics.push(json!({ - "topic": topic.name, - "broker_type": topic.broker_type, + "topic": display_name, + "broker_type": broker_type, "message_count": dlq_count, })); } // Also include function queue DLQs for name in self._config.queue_configs.keys() { + if !seen.insert(name.clone()) { + continue; + } let namespaced = format!("__fn_queue::{}", name); let dlq_count = self.adapter.dlq_count(&namespaced).await.unwrap_or(0); dlq_topics.push(json!({ @@ -2310,6 +2323,26 @@ mod tests { } } + #[tokio::test] + async fn console_dlq_topics_deduplicates_function_queues_reported_by_adapter() { + let (_engine, module, adapter) = setup_queue_module_with_configs(); + *adapter.list_topics_result.lock().await = vec![TopicInfo { + name: "__fn_queue::default".to_string(), + broker_type: "builtin".to_string(), + subscriber_count: 0, + }]; + + let result = module.console_dlq_topics(json!({})).await; + match result { + FunctionResult::Success(Some(val)) => { + let topics: Vec = serde_json::from_value(val).unwrap(); + let default_count = topics.iter().filter(|t| t["topic"] == "default").count(); + assert_eq!(default_count, 1, "function queue topics should not be duplicated"); + } + _ => panic!("Expected Success with DLQ topics"), + } + } + #[tokio::test] async fn console_dlq_topics_adapter_error() { let (_engine, module, adapter) = setup_queue_module(); From a0f63a6032f4309648bbf388c5acf172d549834e Mon Sep 17 00:00:00 2001 From: giria660 Date: Tue, 7 Apr 2026 21:28:43 -0400 Subject: [PATCH 2/3] Refine DLQ topic deduplication --- engine/src/workers/queue/queue.rs | 109 +++++++++++++++++++++++++----- 1 file changed, 92 insertions(+), 17 deletions(-) diff --git a/engine/src/workers/queue/queue.rs b/engine/src/workers/queue/queue.rs index 2443c51c69..8ac7f5c11d 100644 --- a/engine/src/workers/queue/queue.rs +++ b/engine/src/workers/queue/queue.rs @@ -434,27 +434,29 @@ impl QueueWorker { match self.adapter.list_topics().await { Ok(topics) => { let mut dlq_topics = Vec::new(); - let mut seen = HashSet::new(); + let mut seen_function_queues = HashSet::new(); for topic in &topics { - let (display_name, broker_type) = - if let Some(stripped) = topic.name.strip_prefix("__fn_queue::") { - (stripped.to_string(), "function_queue".to_string()) - } else { - (topic.name.clone(), topic.broker_type.clone()) - }; - if !seen.insert(display_name.clone()) { - continue; - } let dlq_count = self.adapter.dlq_count(&topic.name).await.unwrap_or(0); - dlq_topics.push(json!({ - "topic": display_name, - "broker_type": broker_type, - "message_count": dlq_count, - })); + if let Some(stripped) = topic.name.strip_prefix("__fn_queue::") { + if !seen_function_queues.insert(stripped.to_string()) { + continue; + } + dlq_topics.push(json!({ + "topic": stripped, + "broker_type": "function_queue", + "message_count": dlq_count, + })); + } else { + dlq_topics.push(json!({ + "topic": topic.name.clone(), + "broker_type": topic.broker_type.clone(), + "message_count": dlq_count, + })); + } } // Also include function queue DLQs for name in self._config.queue_configs.keys() { - if !seen.insert(name.clone()) { + if !seen_function_queues.insert(name.clone()) { continue; } let namespaced = format!("__fn_queue::{}", name); @@ -1752,6 +1754,41 @@ mod tests { ); } + #[tokio::test] + async fn integration_console_dlq_topics_deduplicates_builtin_function_queue_topics() { + let (_engine, module, adapter) = setup_integration_module().await; + let default_config = module + ._config + .queue_configs + .get("default") + .expect("default queue config should exist") + .clone(); + + adapter + .setup_function_queue("default", &default_config) + .await + .expect("builtin function queue setup should succeed"); + + let result = module.console_dlq_topics(json!({})).await; + match result { + FunctionResult::Success(Some(val)) => { + let topics: Vec = serde_json::from_value(val).unwrap(); + let default_count = topics.iter().filter(|t| t["topic"] == "default").count(); + assert_eq!( + default_count, 1, + "function queue topics should not be duplicated" + ); + assert!( + topics + .iter() + .any(|t| t["topic"] == "default" && t["broker_type"] == "function_queue"), + "builtin function queue should be normalized to function_queue" + ); + } + _ => panic!("Expected Success with DLQ topics"), + } + } + #[tokio::test] async fn integration_function_failure_nacks() { let (engine, module, adapter) = setup_integration_module().await; @@ -2346,7 +2383,45 @@ mod tests { FunctionResult::Success(Some(val)) => { let topics: Vec = serde_json::from_value(val).unwrap(); let default_count = topics.iter().filter(|t| t["topic"] == "default").count(); - assert_eq!(default_count, 1, "function queue topics should not be duplicated"); + assert_eq!( + default_count, 1, + "function queue topics should not be duplicated" + ); + } + _ => panic!("Expected Success with DLQ topics"), + } + } + + #[tokio::test] + async fn console_dlq_topics_keeps_regular_topics_with_function_queue_name() { + let (_engine, module, adapter) = setup_queue_module_with_configs(); + *adapter.list_topics_result.lock().await = vec![TopicInfo { + name: "default".to_string(), + broker_type: "builtin".to_string(), + subscriber_count: 0, + }]; + + let result = module.console_dlq_topics(json!({})).await; + match result { + FunctionResult::Success(Some(val)) => { + let topics: Vec = serde_json::from_value(val).unwrap(); + let default_topics: Vec<&Value> = + topics.iter().filter(|t| t["topic"] == "default").collect(); + assert_eq!( + default_topics.len(), + 2, + "regular topic and function queue should both be listed" + ); + assert!( + default_topics.iter().any(|t| t["broker_type"] == "builtin"), + "regular topic should keep its original broker type" + ); + assert!( + default_topics + .iter() + .any(|t| t["broker_type"] == "function_queue"), + "configured function queue should still be included" + ); } _ => panic!("Expected Success with DLQ topics"), } From 6c04a9a2f3edc16ee064f29c5ed1e5db3810195a Mon Sep 17 00:00:00 2001 From: giria660 Date: Tue, 7 Apr 2026 21:50:20 -0400 Subject: [PATCH 3/3] Tighten DLQ topic normalization assertions --- engine/src/workers/queue/queue.rs | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/engine/src/workers/queue/queue.rs b/engine/src/workers/queue/queue.rs index 8ac7f5c11d..3995e274de 100644 --- a/engine/src/workers/queue/queue.rs +++ b/engine/src/workers/queue/queue.rs @@ -1784,6 +1784,10 @@ mod tests { .any(|t| t["topic"] == "default" && t["broker_type"] == "function_queue"), "builtin function queue should be normalized to function_queue" ); + assert!( + topics.iter().all(|t| t["topic"] != "__fn_queue::default"), + "raw builtin function queue topic should not leak into console output" + ); } _ => panic!("Expected Success with DLQ topics"), } @@ -2422,6 +2426,10 @@ mod tests { .any(|t| t["broker_type"] == "function_queue"), "configured function queue should still be included" ); + assert!( + topics.iter().all(|t| t["topic"] != "__fn_queue::default"), + "raw builtin function queue topic should not be included alongside normalized output" + ); } _ => panic!("Expected Success with DLQ topics"), }