diff --git a/crates/targets/src/target/pulsar.rs b/crates/targets/src/target/pulsar.rs index 15c8122f3..f94cee68e 100644 --- a/crates/targets/src/target/pulsar.rs +++ b/crates/targets/src/target/pulsar.rs @@ -25,7 +25,7 @@ use crate::{ target::{ ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot, TargetTlsState, TargetType, build_queued_payload_with_records, build_target_tls_fingerprint, open_target_queue_store, - persist_queued_payload_to_store, redacted_secret, + persist_queued_payload_to_store, redacted_secret, sanitize_queue_dir_component, }, }; use async_trait::async_trait; @@ -42,6 +42,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use tokio::sync::Mutex as AsyncMutex; use tracing::{info, instrument}; use url::Url; +use uuid::Uuid; #[derive(Clone)] pub struct PulsarArgs { @@ -156,6 +157,15 @@ pub fn validate_pulsar_broker(broker: &str) -> Result { Ok(url) } +fn pulsar_producer_name(target_id: &TargetID, target_type: TargetType, nonce: Uuid) -> String { + let target_type = match target_type { + TargetType::NotifyEvent => "notify", + TargetType::AuditLog => "audit", + }; + + format!("rustfs-{target_type}-pulsar-{}-{nonce}", sanitize_queue_dir_component(&target_id.id)) +} + pub async fn connect_pulsar(args: &PulsarArgs) -> Result, TargetError> { args.validate()?; @@ -307,7 +317,7 @@ where let producer = client .producer() .with_topic(self.args.topic.clone()) - .with_name(self.id.id.clone()) + .with_name(pulsar_producer_name(&self.id, self.args.target_type, Uuid::new_v4())) .build() .await .map_err(|e| TargetError::Network(format!("Failed to create Pulsar producer: {e}")))?; @@ -573,4 +583,76 @@ mod tests { }; assert!(args.validate().is_err()); } + + #[test] + fn pulsar_producer_name_keeps_target_context_and_unique_suffix() { + let id = TargetID::new("test2".to_string(), ChannelTargetType::Pulsar.as_str().to_string()); + let nonce = Uuid::from_u128(0x12345678_90ab_cdef_1234_567890abcdef); + + let name = pulsar_producer_name(&id, TargetType::NotifyEvent, nonce); + + assert_eq!(name, "rustfs-notify-pulsar-test2-12345678-90ab-cdef-1234-567890abcdef"); + } + + #[test] + fn pulsar_producer_name_includes_target_type() { + let id = TargetID::new("audit-target".to_string(), ChannelTargetType::Pulsar.as_str().to_string()); + let nonce = Uuid::from_u128(0x12345678_90ab_cdef_1234_567890abcdef); + + let name = pulsar_producer_name(&id, TargetType::AuditLog, nonce); + + assert_eq!(name, "rustfs-audit-pulsar-audit-target-12345678-90ab-cdef-1234-567890abcdef"); + } + + #[test] + fn pulsar_producer_name_sanitizes_target_id() { + let raw_target_id = "tenant:alpha/beta"; + let id = TargetID::new(raw_target_id.to_string(), ChannelTargetType::Pulsar.as_str().to_string()); + let nonce = Uuid::from_u128(0x12345678_90ab_cdef_1234_567890abcdef); + + let name = pulsar_producer_name(&id, TargetType::NotifyEvent, nonce); + + assert!(!name.contains(':')); + assert!(!name.contains('/')); + assert_eq!( + name, + format!( + "rustfs-notify-pulsar-{}-12345678-90ab-cdef-1234-567890abcdef", + sanitize_queue_dir_component(raw_target_id) + ) + ); + } + + #[test] + fn pulsar_producer_name_changes_for_each_producer_generation() { + let id = TargetID::new("test".to_string(), ChannelTargetType::Pulsar.as_str().to_string()); + let first = Uuid::from_u128(0x12345678_90ab_cdef_1234_567890abcdef); + let second = Uuid::from_u128(0xfedcba09_8765_4321_fedc_ba0987654321); + + assert_ne!( + pulsar_producer_name(&id, TargetType::NotifyEvent, first), + pulsar_producer_name(&id, TargetType::NotifyEvent, second) + ); + } + + #[test] + fn init_producer_uses_generated_unique_producer_name_contract() { + fn without_ascii_whitespace(value: &str) -> String { + let mut compacted = String::with_capacity(value.len()); + compacted.extend(value.bytes().filter(|byte| !byte.is_ascii_whitespace()).map(char::from)); + compacted + } + + let source = without_ascii_whitespace(include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/src/target/pulsar.rs"))); + let expected_name_call = [ + ".with_name(", + "pulsar_producer_name(&self.id, self.args.target_type, Uuid::new_v4())", + ")", + ] + .concat(); + let forbidden_target_id_call = [".with_name(", "self.id.id.clone()", ")"].concat(); + + assert!(source.contains(&without_ascii_whitespace(&expected_name_call))); + assert!(!source.contains(&without_ascii_whitespace(&forbidden_target_id_call))); + } }