fix(targets): harden messaging backend delivery and error classification (#4506)

Address a batch of correctness/security audit findings in the messaging
notification backends (mqtt/nats/kafka/amqp/redis/pulsar/webhook).

MQTT (backlog#971): wait for broker acknowledgement via publish_tracked +
wait_completion_async (PUBACK/PUBCOMP for QoS>=1, flush for QoS0) with a
30s timeout before reporting success, instead of treating the enqueue as
delivered. Replace substring error matching with typed ClientError /
PublishNoticeError classification.

NATS (backlog#971, #973, #983): flush() after publish to confirm the broker
received the message before the durable copy is deleted; classify publish
and flush failures by typed error kind; warn when credentials are sent
without TLS.

Kafka (backlog#973, #983): use Error::is_retriable() so transient broker
states (NotLeaderForPartition, LeaderNotAvailable, RequestTimedOut, ...) are
retried instead of dropped as permanent; add a bucket/object message key for
per-object partition ordering; build the producer without holding the cache
lock across the connect await.

AMQP (backlog#973, #980): classify permanent broker protocol errors
(404/403/406, NOT_ALLOWED, ...) as request-level errors rather than
connectivity errors to avoid reconnect storms; bound publish and
publisher-confirm waits with a timeout; check is_enabled in is_active; run
full close() cleanup even when the broker close fails; warn when
mandatory=false may silently drop unroutable messages.

Redis (backlog#982): warn when PUBLISH reaches 0 subscribers; reuse the
cached ConnectionManager for health probes instead of a fresh handshake;
warn when tls_allow_insecure disables certificate verification.

Pulsar (backlog#983): replace std::sync::Mutex + unwrap with parking_lot
Mutex so a panic while holding the guard cannot poison later accesses.

Webhook (backlog#983): drain the response body so the connection can be
reused (keep-alive). The #974 redirect-follow SSRF fix already landed.

Relates to rustfs/backlog#971
Relates to rustfs/backlog#973
Relates to rustfs/backlog#974
Relates to rustfs/backlog#980
Relates to rustfs/backlog#982
Relates to rustfs/backlog#983

Co-authored-by: heihutu <heihutu@gmail.com>

crates/targets/src/target/amqp.rs

@@ -51,6 +51,10 @@ use tokio::sync::Mutex as AsyncMutex;
 use tracing::{info, instrument, warn};
 use url::Url;
 
+/// Upper bound on how long a single publish and its publisher-confirm may block
+/// before being treated as a timeout (backlog#980).
+const AMQP_PUBLISH_TIMEOUT: Duration = Duration::from_secs(30);
+
 #[derive(Clone)]
 pub struct AMQPArgs {
     pub enable: bool,
@@ -238,8 +242,40 @@ fn build_publish_properties(args: &AMQPArgs) -> BasicProperties {
     properties
 }
 
+/// Returns true for AMQP broker protocol errors that indicate a permanent,
+/// non-connectivity condition (e.g. the exchange/queue does not exist, access is
+/// refused, or a precondition failed). These must not be treated as transient
+/// connectivity errors, otherwise a misconfigured target triggers an endless
+/// reconnect storm instead of surfacing a delivery failure (backlog#973).
+fn is_permanent_amqp_protocol_error(err: &lapin::Error) -> bool {
+    use lapin::protocol::{AMQPErrorKind, AMQPHardError, AMQPSoftError};
+    if let LapinErrorKind::ProtocolError(amqp_err) = err.kind() {
+        return match amqp_err.kind() {
+            // 404 NOT_FOUND (missing exchange/queue), 403 ACCESS_REFUSED,
+            // 406 PRECONDITION_FAILED.
+            AMQPErrorKind::Soft(soft) => {
+                matches!(
+                    soft,
+                    AMQPSoftError::NOTFOUND | AMQPSoftError::ACCESSREFUSED | AMQPSoftError::PRECONDITIONFAILED
+                )
+            }
+            // 530 NOT_ALLOWED, 540 NOT_IMPLEMENTED, 402 INVALID_PATH.
+            AMQPErrorKind::Hard(hard) => {
+                matches!(
+                    hard,
+                    AMQPHardError::NOTALLOWED | AMQPHardError::NOTIMPLEMENTED | AMQPHardError::INVALIDPATH
+                )
+            }
+        };
+    }
+    false
+}
+
 fn map_lapin_error(err: lapin::Error, context: &str) -> TargetError {
     let message = format!("{context}: {err}");
+    if is_permanent_amqp_protocol_error(&err) {
+        return TargetError::Request(message);
+    }
     match err.kind() {
         LapinErrorKind::IOError(io_err) if io_err.kind() == std::io::ErrorKind::TimedOut => TargetError::Timeout(message),
         LapinErrorKind::IOError(_)
@@ -332,6 +368,18 @@ where
     #[instrument(skip(args), fields(target_id_as_string = %id))]
     pub fn new(id: String, args: AMQPArgs) -> Result<Self, TargetError> {
         args.validate()?;
+        if args.enable && !args.mandatory {
+            // With mandatory=false the broker silently discards messages that
+            // route to no queue, and the publish is still confirmed as success.
+            // Warn so operators expecting reliable delivery know unroutable
+            // events are dropped without a trace (backlog#980).
+            warn!(
+                target_id = %id,
+                exchange = %args.exchange,
+                routing_key = %args.routing_key,
+                "AMQP target has mandatory=false: messages that route to no queue are silently dropped. Set mandatory=true for reliable delivery."
+            );
+        }
         let target_id = TargetID::new(id, ChannelTargetType::Amqp.as_str().to_string());
         let queue_store = open_target_queue_store(
             &args.queue_dir,
@@ -417,9 +465,11 @@ where
 
     async fn send_body(&self, body: &[u8]) -> Result<(), TargetError> {
         let connection = self.get_or_connect().await?;
-        let publish = connection
-            .channel
-            .basic_publish(
+        // Bound the publish and publisher-confirm waits so a stuck broker cannot
+        // block the send path indefinitely (backlog#980).
+        let publish = match tokio::time::timeout(
+            AMQP_PUBLISH_TIMEOUT,
+            connection.channel.basic_publish(
                 self.args.exchange.clone().into(),
                 self.args.routing_key.clone().into(),
                 BasicPublishOptions {
@@ -428,17 +478,33 @@ where
                 },
                 body,
                 build_publish_properties(&self.args),
-            )
-            .await;
+            ),
+        )
+        .await
+        {
+            Ok(publish) => publish,
+            Err(_) => {
+                self.clear_connection();
+                return Err(TargetError::Timeout("AMQP publish timed out".to_string()));
+            }
+        };
 
-        let confirm = match publish {
-            Ok(confirm) => confirm.await,
+        let confirm_future = match publish {
+            Ok(confirm) => confirm,
             Err(err) => {
                 self.clear_connection();
                 return Err(map_lapin_error(err, "Failed to publish AMQP message"));
             }
         };
 
+        let confirm = match tokio::time::timeout(AMQP_PUBLISH_TIMEOUT, confirm_future).await {
+            Ok(confirm) => confirm,
+            Err(_) => {
+                self.clear_connection();
+                return Err(TargetError::Timeout("AMQP publisher confirm timed out".to_string()));
+            }
+        };
+
         match confirm {
             Ok(Confirmation::Ack(None) | Confirmation::NotRequested) => {
                 self.delivery_counters.record_success();
@@ -506,6 +572,11 @@ where
     }
 
     async fn is_active(&self) -> Result<bool, TargetError> {
+        // A disabled target is never active; avoid opening a connection for it
+        // (backlog#980).
+        if !self.args.enable {
+            return Ok(false);
+        }
         let connection = self.get_or_connect().await?;
         Ok(connection.connection.status().connected() && connection.channel.status().connected())
     }
@@ -540,12 +611,14 @@ where
 
     async fn close(&self) -> Result<(), TargetError> {
         let connection = self.connection.lock().take();
-        if let Some(connection) = connection {
-            connection
-                .connection
-                .close(200, "OK".into())
-                .await
-                .map_err(|e| map_lapin_error(e, "Failed to close AMQP connection"))?;
+        // Capture any close failure but still run the remaining cleanup so a
+        // failed broker close does not leave stale TLS/adapter state behind
+        // (backlog#980).
+        let mut close_result = Ok(());
+        if let Some(connection) = connection
+            && let Err(e) = connection.connection.close(200, "OK".into()).await
+        {
+            close_result = Err(map_lapin_error(e, "Failed to close AMQP connection"));
         }
         self.tls_state.lock().reset();
         // If a TLS reload adapter is attached, reset its error tracking
@@ -554,7 +627,7 @@ where
             *adapter.runtime_state().last_error.write() = None;
         }
         info!(target_id = %self.id, "AMQP target closed");
-        Ok(())
+        close_result
     }
 
     fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
@@ -752,6 +825,27 @@ mod tests {
         assert_connect_failure(&err);
     }
 
+    #[test]
+    fn permanent_amqp_protocol_errors_are_not_connectivity_errors() {
+        use lapin::protocol::{AMQPError, AMQPErrorKind, AMQPHardError, AMQPSoftError};
+
+        let make = |kind: AMQPErrorKind| lapin::Error::from(LapinErrorKind::ProtocolError(AMQPError::new(kind, "boom".into())));
+
+        // 404 (missing exchange/queue) is permanent: it must be a request-level
+        // error so a misconfigured target does not reconnect-storm (backlog#973).
+        let not_found = make(AMQPErrorKind::Soft(AMQPSoftError::NOTFOUND));
+        assert!(is_permanent_amqp_protocol_error(&not_found));
+        assert!(matches!(map_lapin_error(not_found, "publish"), TargetError::Request(_)));
+
+        assert!(is_permanent_amqp_protocol_error(&make(AMQPErrorKind::Soft(AMQPSoftError::ACCESSREFUSED))));
+        assert!(is_permanent_amqp_protocol_error(&make(AMQPErrorKind::Hard(AMQPHardError::NOTALLOWED))));
+
+        // A transient soft error (broker resource locked) is not treated as permanent.
+        assert!(!is_permanent_amqp_protocol_error(&make(AMQPErrorKind::Soft(
+            AMQPSoftError::RESOURCELOCKED
+        ))));
+    }
+
     #[test]
     fn debug_masks_secret_values() {
         let args = AMQPArgs {

crates/targets/src/target/kafka.rs

@@ -29,7 +29,7 @@ use crate::{
     },
 };
 use async_trait::async_trait;
-use rustfs_kafka_async::error::{ConnectionError, Error as KafkaError};
+use rustfs_kafka_async::error::{ConnectionError, Error as KafkaError, KafkaCode};
 use rustfs_kafka_async::{AsyncProducer, AsyncProducerConfig, Record, RequiredAcks, SaslConfig, SecurityConfig};
 use rustfs_tls_runtime::{load_cert_bundle_der_bytes, load_private_key};
 use std::{fmt, marker::PhantomData, sync::Arc, time::Duration};
@@ -247,10 +247,23 @@ where
     E: PluginEvent,
 {
     fn map_kafka_error(err: KafkaError, context: &str) -> TargetError {
-        match err {
-            KafkaError::Connection(ConnectionError::NoHostReachable) => TargetError::NotConnected,
-            KafkaError::Connection(ConnectionError::Timeout(_)) => TargetError::Timeout(format!("{context}: {err}")),
-            KafkaError::Connection(_) => TargetError::Network(format!("{context}: {err}")),
+        // Prefer the client's own retriable classification so transient broker
+        // states (leader election / NotLeaderForPartition, coordinator load,
+        // network blips, RequestTimedOut) are retried via store replay instead of
+        // being dropped as permanent failures (backlog#973).
+        if err.is_retriable() {
+            return match &err {
+                KafkaError::Connection(ConnectionError::Timeout(_)) | KafkaError::Kafka(KafkaCode::RequestTimedOut) => {
+                    TargetError::Timeout(format!("{context}: {err}"))
+                }
+                _ => TargetError::NotConnected,
+            };
+        }
+
+        // Non-retriable errors: configuration problems are permanent config
+        // errors; everything else (e.g. UnknownTopicOrPartition, authorization
+        // failures, oversize messages) is a permanent request-level failure.
+        match &err {
             KafkaError::Config(_) => TargetError::Configuration(format!("{context}: {err}")),
             _ => TargetError::Request(format!("{context}: {err}")),
         }
@@ -332,12 +345,22 @@ where
             self.tls_state.lock().await.refresh(next_fingerprint);
         }
 
-        let mut cached = self.producer.lock().await;
-        if let Some(producer) = cached.as_ref() {
-            return Ok(Arc::clone(producer));
+        {
+            let cached = self.producer.lock().await;
+            if let Some(producer) = cached.as_ref() {
+                return Ok(Arc::clone(producer));
+            }
         }
 
+        // Build the producer without holding the cache lock so a slow connect
+        // does not block other senders (which only need to read the cache).
+        // Re-check the cache after building in case another task raced us
+        // (backlog#983).
         let producer = Arc::new(self.build_producer().await?);
+        let mut cached = self.producer.lock().await;
+        if let Some(existing) = cached.as_ref() {
+            return Ok(Arc::clone(existing));
+        }
         *cached = Some(Arc::clone(&producer));
         Ok(producer)
     }
@@ -367,7 +390,14 @@ where
 
         let producer = self.get_or_build_producer().await?;
 
-        if let Err(err) = producer.send(&Record::from_value(&self.args.topic, body.as_slice())).await {
+        // Use "<bucket>/<object>" as the message key so all events for the same
+        // object hash to the same partition and preserve per-object ordering
+        // across multiple partitions (backlog#983).
+        let partition_key = format!("{}/{}", meta.bucket_name, meta.object_name);
+        if let Err(err) = producer
+            .send(&Record::from_key_value(&self.args.topic, partition_key, body.as_slice()))
+            .await
+        {
             let mapped = Self::map_kafka_error(err, "Failed to send message to Kafka");
             invalidate_cache_on_connectivity_error(&mapped, || self.invalidate_cached_producer()).await;
             return Err(mapped);
@@ -663,6 +693,39 @@ mod tests {
         assert_eq!(sasl.password(), "secret");
     }
 
+    #[test]
+    fn map_kafka_error_treats_transient_broker_states_as_retriable() {
+        // Leader election / metadata staleness must be retried, not dropped
+        // as a permanent failure (backlog#973).
+        assert!(matches!(
+            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Kafka(KafkaCode::NotLeaderForPartition), "send"),
+            TargetError::NotConnected
+        ));
+        assert!(matches!(
+            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Kafka(KafkaCode::LeaderNotAvailable), "send"),
+            TargetError::NotConnected
+        ));
+        // RequestTimedOut is retriable and surfaced as a timeout.
+        assert!(matches!(
+            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Kafka(KafkaCode::RequestTimedOut), "send"),
+            TargetError::Timeout(_)
+        ));
+    }
+
+    #[test]
+    fn map_kafka_error_treats_permanent_broker_states_as_request_error() {
+        // A missing topic/partition is a permanent condition; retrying would
+        // storm the broker.
+        assert!(matches!(
+            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Kafka(KafkaCode::UnknownTopicOrPartition), "send"),
+            TargetError::Request(_)
+        ));
+        assert!(matches!(
+            KafkaTarget::<serde_json::Value>::map_kafka_error(KafkaError::Config("bad".to_string()), "send"),
+            TargetError::Configuration(_)
+        ));
+    }
+
     #[test]
     fn test_debug_redacts_sasl_password_and_tls_key() {
         let rendered = format!(

crates/targets/src/target/mqtt.rs

@@ -32,8 +32,8 @@ use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use hyper_rustls::ConfigBuilderExt;
 use rumqttc::{
-    AsyncClient, Broker, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, QoS, Transport,
-    mqttbytes::Error as MqttBytesError,
+    AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, PublishNoticeError, QoS,
+    Transport, mqttbytes::Error as MqttBytesError,
 };
 use rustfs_config::{
     EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST,
@@ -58,6 +58,10 @@ const DEFAULT_MQTT_TCP_PORT: u16 = 1883;
 const DEFAULT_MQTT_TLS_PORT: u16 = 8883;
 const DEFAULT_MQTT_WSS_PORT: u16 = 443;
 const MAX_MQTT_PACKET_SIZE_BYTES: u32 = 100 * 1024 * 1024;
+/// Upper bound on how long a single publish may wait for broker acknowledgement
+/// (PUBACK/PUBCOMP for QoS>=1, or network flush for QoS0) before it is treated as
+/// a timeout so the durable copy is retained and replayed (backlog#971).
+const MQTT_PUBLISH_CONFIRM_TIMEOUT: Duration = Duration::from_secs(30);
 /// Minimum delay before the supervisor rebuilds the client and event loop
 /// after a session exits. Also the delay used right after a session that had
 /// successfully connected, so a transient drop reconnects promptly.
@@ -779,40 +783,78 @@ where
             "mqtt delivery state"
         );
 
-        client
-            .publish(&self.args.topic, self.args.qos, false, body)
-            .await
-            .map_err(|e| {
-                if e.to_string().contains("Connection") || e.to_string().contains("Timeout") {
-                    warn!(
-                        event = EVENT_MQTT_DELIVERY_STATE,
-                        component = LOG_COMPONENT_TARGETS,
-                        subsystem = LOG_SUBSYSTEM_MQTT,
-                        target_id = %self.id,
-                        state = "publish_failed",
-                        reason = "connectivity_error",
-                        error = %e,
-                        "mqtt delivery state"
-                    );
-                    let err = TargetError::NotConnected;
-                    mark_target_disconnected_on_connectivity_error(&self.connected, &err);
-                    err
-                } else {
-                    TargetError::Request(format!("Failed to publish message: {e}"))
-                }
-            })?;
+        // Enqueue a tracked publish so we can wait for broker acknowledgement
+        // (PUBACK for QoS1, PUBCOMP for QoS2, or network flush for QoS0) before
+        // reporting success. Previously the publish was treated as delivered as
+        // soon as it was queued on the event loop, so a disconnect after queueing
+        // silently dropped the event while its durable copy was already deleted
+        // (backlog#971). Error classification now matches on the typed error
+        // instead of substring matching on the display string.
+        let notice = match client.publish_tracked(&self.args.topic, self.args.qos, false, body).await {
+            Ok(notice) => notice,
+            Err(e) => {
+                let err = classify_mqtt_client_error(&e);
+                warn!(
+                    event = EVENT_MQTT_DELIVERY_STATE,
+                    component = LOG_COMPONENT_TARGETS,
+                    subsystem = LOG_SUBSYSTEM_MQTT,
+                    target_id = %self.id,
+                    state = "publish_failed",
+                    reason = "enqueue_error",
+                    error = %e,
+                    "mqtt delivery state"
+                );
+                mark_target_disconnected_on_connectivity_error(&self.connected, &err);
+                return Err(err);
+            }
+        };
 
-        debug!(
-            event = EVENT_MQTT_DELIVERY_STATE,
-            component = LOG_COMPONENT_TARGETS,
-            subsystem = LOG_SUBSYSTEM_MQTT,
-            target_id = %self.id,
-            topic = %self.args.topic,
-            state = "published",
-            "mqtt delivery state"
-        );
-        self.delivery_counters.record_success();
-        Ok(())
+        // Release the client lock before awaiting the broker acknowledgement so a
+        // slow/hung broker never blocks other senders from queueing publishes.
+        drop(client_guard);
+
+        match tokio::time::timeout(MQTT_PUBLISH_CONFIRM_TIMEOUT, notice.wait_completion_async()).await {
+            Ok(Ok(())) => {
+                debug!(
+                    event = EVENT_MQTT_DELIVERY_STATE,
+                    component = LOG_COMPONENT_TARGETS,
+                    subsystem = LOG_SUBSYSTEM_MQTT,
+                    target_id = %self.id,
+                    topic = %self.args.topic,
+                    state = "published",
+                    "mqtt delivery state"
+                );
+                self.delivery_counters.record_success();
+                Ok(())
+            }
+            Ok(Err(e)) => {
+                let err = classify_mqtt_notice_error(&e);
+                warn!(
+                    event = EVENT_MQTT_DELIVERY_STATE,
+                    component = LOG_COMPONENT_TARGETS,
+                    subsystem = LOG_SUBSYSTEM_MQTT,
+                    target_id = %self.id,
+                    state = "publish_unconfirmed",
+                    error = %e,
+                    "mqtt delivery state"
+                );
+                mark_target_disconnected_on_connectivity_error(&self.connected, &err);
+                Err(err)
+            }
+            Err(_) => {
+                let err = TargetError::Timeout("Timed out waiting for MQTT publish acknowledgement".to_string());
+                warn!(
+                    event = EVENT_MQTT_DELIVERY_STATE,
+                    component = LOG_COMPONENT_TARGETS,
+                    subsystem = LOG_SUBSYSTEM_MQTT,
+                    target_id = %self.id,
+                    state = "publish_confirm_timeout",
+                    "mqtt delivery state"
+                );
+                mark_target_disconnected_on_connectivity_error(&self.connected, &err);
+                Err(err)
+            }
+        }
     }
 
     pub fn clone_target(&self) -> Box<dyn Target<E> + Send + Sync> {
@@ -1190,6 +1232,33 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc<Ato
     initial_connection_established
 }
 
+/// Classifies a publish-enqueue failure. Every [`ClientError`] variant means the
+/// publish could not be handed to the event loop (channel closed/full, or the
+/// tracked-publish API is unavailable), i.e. the client is not currently able to
+/// deliver. These are treated as retriable connectivity errors so the durable
+/// copy is preserved and replayed rather than dropped (backlog#971).
+fn classify_mqtt_client_error(err: &ClientError) -> TargetError {
+    match err {
+        ClientError::Request(_) | ClientError::TryRequest(_) | ClientError::TrackingUnavailable => TargetError::NotConnected,
+    }
+}
+
+/// Classifies a publish acknowledgement failure returned while waiting for the
+/// broker to confirm delivery. Connectivity/session problems keep the event for
+/// replay; a broker rejection with a failing reason code is surfaced as a
+/// request-level error (backlog#971).
+fn classify_mqtt_notice_error(err: &PublishNoticeError) -> TargetError {
+    match err {
+        PublishNoticeError::Recv
+        | PublishNoticeError::SessionReset
+        | PublishNoticeError::Qos0NotFlushed
+        | PublishNoticeError::TopicAliasReplayUnavailable(_) => TargetError::NotConnected,
+        PublishNoticeError::V5PubAck(_) | PublishNoticeError::V5PubRec(_) | PublishNoticeError::V5PubComp(_) => {
+            TargetError::Request(format!("MQTT broker rejected publish: {err}"))
+        }
+    }
+}
+
 /// Check whether the given MQTT connection error should be considered a fatal error,
 /// For fatal errors, the event loop should terminate.
 fn is_fatal_mqtt_error(err: &ConnectionError) -> bool {
@@ -1636,9 +1705,11 @@ where
 #[cfg(test)]
 mod tests {
     use super::{
-        MQTT_RECONNECT_BACKOFF_MAX, MQTT_RECONNECT_BACKOFF_MIN, MQTTArgs, MQTTTlsConfig, QoS, next_reconnect_backoff,
-        reconnect_supervisor, validate_mqtt_broker_url,
+        ClientError, MQTT_RECONNECT_BACKOFF_MAX, MQTT_RECONNECT_BACKOFF_MIN, MQTTArgs, MQTTTlsConfig, PublishNoticeError, QoS,
+        classify_mqtt_client_error, classify_mqtt_notice_error, next_reconnect_backoff, reconnect_supervisor,
+        validate_mqtt_broker_url,
     };
+    use crate::error::TargetError;
     use crate::target::{REDACTED_SECRET, TargetType};
     use std::sync::Arc;
     use std::sync::atomic::{AtomicUsize, Ordering};
@@ -1646,6 +1717,38 @@ mod tests {
     use tokio::sync::mpsc;
     use url::Url;
 
+    #[test]
+    fn mqtt_client_error_classified_as_not_connected() {
+        // A publish that cannot be handed to the event loop means the client is
+        // not connected; the durable copy must be kept for replay (backlog#971).
+        assert!(matches!(
+            classify_mqtt_client_error(&ClientError::TrackingUnavailable),
+            TargetError::NotConnected
+        ));
+    }
+
+    #[test]
+    fn mqtt_notice_connectivity_errors_kept_for_replay() {
+        for err in [
+            PublishNoticeError::SessionReset,
+            PublishNoticeError::Qos0NotFlushed,
+            PublishNoticeError::Recv,
+        ] {
+            assert!(
+                matches!(classify_mqtt_notice_error(&err), TargetError::NotConnected),
+                "unconfirmed publish {err:?} should be retriable"
+            );
+        }
+    }
+
+    #[test]
+    fn mqtt_notice_broker_rejection_is_request_error() {
+        // The broker acknowledged the publish but rejected it: this is a
+        // request-level failure, not a transient disconnect.
+        let err = PublishNoticeError::V5PubAck(rumqttc::PubAckReason::NotAuthorized);
+        assert!(matches!(classify_mqtt_notice_error(&err), TargetError::Request(_)));
+    }
+
     #[test]
     fn next_reconnect_backoff_doubles_until_capped() {
         let mut backoff = MQTT_RECONNECT_BACKOFF_MIN;

crates/targets/src/target/nats.rs

@@ -24,8 +24,8 @@ use crate::{
     store::{Key, Store},
     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,
+        TargetTlsState, TargetType, build_queued_payload_with_records, build_target_tls_fingerprint, is_connectivity_error,
+        open_target_queue_store, persist_queued_payload_to_store, redacted_secret,
     },
 };
 use async_trait::async_trait;
@@ -160,6 +160,21 @@ fn validate_nats_auth(args: &NATSArgs) -> Result<(), TargetError> {
     Ok(())
 }
 
+/// Returns true when the target is configured to send credentials (token,
+/// username/password, or a credentials file) over a connection that does not
+/// require TLS, which would transmit the secrets in cleartext (backlog#983).
+///
+/// TLS is considered active when `tls_required` is set or the address uses the
+/// `tls://` scheme.
+fn nats_sends_credentials_without_tls(args: &NATSArgs) -> bool {
+    let has_auth = !args.token.is_empty() || !args.credentials_file.is_empty() || !args.username.is_empty();
+    if !has_auth {
+        return false;
+    }
+    let scheme_is_tls = args.address.trim_start().to_ascii_lowercase().starts_with("tls://");
+    !(args.tls_required || scheme_is_tls)
+}
+
 pub async fn connect_nats(args: &NATSArgs) -> Result<async_nats::Client, TargetError> {
     args.validate()?;
 
@@ -228,6 +243,13 @@ where
     #[instrument(skip(args), fields(target_id_as_string = %id))]
     pub fn new(id: String, args: NATSArgs) -> Result<Self, TargetError> {
         args.validate()?;
+        if args.enable && nats_sends_credentials_without_tls(&args) {
+            warn!(
+                target_id = %id,
+                address = %args.address,
+                "NATS target sends authentication credentials without TLS; secrets are transmitted in cleartext. Enable tls_required or use a tls:// address."
+            );
+        }
         let target_id = TargetID::new(id, ChannelTargetType::Nats.as_str().to_string());
         let queue_store = open_target_queue_store(
             &args.queue_dir,
@@ -305,15 +327,55 @@ where
 
     async fn send_body(&self, body: Vec<u8>) -> Result<(), TargetError> {
         let client = self.get_or_connect().await?;
-        client
-            .publish(self.args.subject.clone(), body.into())
-            .await
-            .map_err(|e| TargetError::Request(format!("Failed to publish NATS message: {e}")))?;
+        if let Err(e) = client.publish(self.args.subject.clone(), body.into()).await {
+            let err = classify_nats_publish_error(&e);
+            if is_connectivity_error(&err) {
+                self.invalidate_cached_client_connection().await;
+                self.connected.store(false, Ordering::SeqCst);
+            }
+            return Err(err);
+        }
+
+        // `publish` only enqueues the message on the client's outbound channel;
+        // it does not guarantee the broker received it. Flush to confirm the
+        // message reached the server before the caller treats delivery as
+        // successful and deletes the durable copy (backlog#971).
+        if let Err(e) = client.flush().await {
+            let err = classify_nats_flush_error(&e);
+            self.invalidate_cached_client_connection().await;
+            self.connected.store(false, Ordering::SeqCst);
+            return Err(err);
+        }
+
         self.delivery_counters.record_success();
         Ok(())
     }
 }
 
+/// Classifies a NATS publish failure using the typed error kind rather than a
+/// substring match. `Send` means the outbound channel is closed (the connection
+/// is gone) and is retriable; payload/subject violations are permanent
+/// request-level errors (backlog#971, backlog#973).
+fn classify_nats_publish_error(err: &async_nats::PublishError) -> TargetError {
+    use async_nats::PublishErrorKind;
+    match err.kind() {
+        PublishErrorKind::Send => TargetError::NotConnected,
+        PublishErrorKind::MaxPayloadExceeded | PublishErrorKind::InvalidSubject => {
+            TargetError::Request(format!("Failed to publish NATS message: {err}"))
+        }
+    }
+}
+
+/// Classifies a NATS flush failure. Both flush error kinds indicate the message
+/// was not confirmed on the wire (a connectivity problem), so the event is kept
+/// for replay (backlog#971, backlog#973).
+fn classify_nats_flush_error(err: &async_nats::client::FlushError) -> TargetError {
+    use async_nats::client::FlushErrorKind;
+    match err.kind() {
+        FlushErrorKind::SendError | FlushErrorKind::FlushError => TargetError::NotConnected,
+    }
+}
+
 #[async_trait]
 impl<E> Target<E> for NATSTarget<E>
 where
@@ -510,4 +572,56 @@ mod tests {
         };
         assert!(args.validate().is_err());
     }
+
+    #[test]
+    fn nats_credentials_without_tls_is_detected() {
+        // Token auth over a plaintext nats:// address without tls_required leaks
+        // the credential in cleartext (backlog#983).
+        let insecure = NATSArgs {
+            token: "secret-token".to_string(),
+            tls_required: false,
+            ..base_args()
+        };
+        assert!(nats_sends_credentials_without_tls(&insecure));
+
+        // tls_required protects the credentials.
+        let with_tls = NATSArgs {
+            tls_required: true,
+            ..insecure.clone()
+        };
+        assert!(!nats_sends_credentials_without_tls(&with_tls));
+
+        // A tls:// address also counts as protected.
+        let tls_scheme = NATSArgs {
+            address: "tls://127.0.0.1:4222".to_string(),
+            ..insecure
+        };
+        assert!(!nats_sends_credentials_without_tls(&tls_scheme));
+
+        // No credentials configured: nothing to leak.
+        let no_auth = base_args();
+        assert!(!nats_sends_credentials_without_tls(&no_auth));
+    }
+
+    #[test]
+    fn nats_publish_error_classification() {
+        use async_nats::PublishErrorKind;
+        // `Send` means the outbound channel is gone: retriable connectivity error.
+        let send_err = async_nats::PublishError::new(PublishErrorKind::Send);
+        assert!(matches!(classify_nats_publish_error(&send_err), TargetError::NotConnected));
+
+        // Payload/subject violations are permanent request-level errors.
+        let payload_err = async_nats::PublishError::new(PublishErrorKind::MaxPayloadExceeded);
+        assert!(matches!(classify_nats_publish_error(&payload_err), TargetError::Request(_)));
+        let subject_err = async_nats::PublishError::new(PublishErrorKind::InvalidSubject);
+        assert!(matches!(classify_nats_publish_error(&subject_err), TargetError::Request(_)));
+    }
+
+    #[test]
+    fn nats_flush_error_classification() {
+        use async_nats::client::{FlushError, FlushErrorKind};
+        for kind in [FlushErrorKind::SendError, FlushErrorKind::FlushError] {
+            assert!(matches!(classify_nats_flush_error(&FlushError::new(kind)), TargetError::NotConnected));
+        }
+    }
 }

crates/targets/src/target/pulsar.rs

@@ -29,12 +29,16 @@ use crate::{
     },
 };
 use async_trait::async_trait;
+// Use parking_lot's Mutex for the synchronous client/TLS state guards: it does
+// not poison on panic, so a panic while a guard is held cannot cascade into
+// `.unwrap()` panics on every later access (backlog#983).
+use parking_lot::Mutex;
 use pulsar::{Authentication, Producer, Pulsar, TokioExecutor};
 use rustfs_tls_runtime::load_cert_bundle_der_bytes;
 use std::fmt;
 use std::path::Path;
+use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Arc, Mutex};
 use tokio::sync::Mutex as AsyncMutex;
 use tracing::{info, instrument};
 use url::Url;
@@ -208,8 +212,8 @@ where
         Box::new(PulsarTarget::<E> {
             id: self.id.clone(),
             args: self.args.clone(),
-            client: Mutex::new(self.client.lock().unwrap().clone()),
-            tls_state: Mutex::new(self.tls_state.lock().unwrap().clone()),
+            client: Mutex::new(self.client.lock().clone()),
+            tls_state: Mutex::new(self.tls_state.lock().clone()),
             tls_adapter: self.tls_adapter.clone(),
             producer: AsyncMutex::new(None),
             store: self.store.as_ref().map(|s| s.boxed_clone()),
@@ -247,12 +251,12 @@ where
     }
 
     fn clear_cached_client_connection(&self) {
-        self.client.lock().unwrap().take();
+        self.client.lock().take();
     }
 
     fn clear_cached_client(&self) {
         self.clear_cached_client_connection();
-        self.tls_state.lock().unwrap().reset();
+        self.tls_state.lock().reset();
     }
 
     async fn get_or_connect_client(&self) -> Result<Pulsar<TokioExecutor>, TargetError> {
@@ -261,28 +265,28 @@ where
         if let Some(adapter) = &self.tls_adapter {
             let material = adapter.current_material();
             {
-                let mut guard = self.client.lock().unwrap();
+                let mut guard = self.client.lock();
                 *guard = Some((*material).clone());
             }
         } else {
             let next_fingerprint = build_target_tls_fingerprint(&self.args.tls_ca, "", "").await?;
             let tls_changed = {
-                let tls_state_guard = self.tls_state.lock().unwrap();
+                let tls_state_guard = self.tls_state.lock();
                 tls_state_guard.needs_update(&next_fingerprint)
             };
             if tls_changed {
                 self.clear_cached_client_connection();
-                self.tls_state.lock().unwrap().refresh(next_fingerprint);
+                self.tls_state.lock().refresh(next_fingerprint);
             }
         }
 
-        if let Some(client) = self.client.lock().unwrap().clone() {
+        if let Some(client) = self.client.lock().clone() {
             return Ok(client);
         }
 
         let client = connect_pulsar(&self.args).await?;
         self.connected.store(true, Ordering::SeqCst);
-        let mut guard = self.client.lock().unwrap();
+        let mut guard = self.client.lock();
         let shared = guard.get_or_insert_with(|| client.clone()).clone();
         Ok(shared)
     }
@@ -363,7 +367,7 @@ where
     ) -> Result<(), TargetError> {
         // Pulsar client is Clone, so we clone from the Arc and store it.
         {
-            let mut guard = self.client.lock().unwrap();
+            let mut guard = self.client.lock();
             *guard = Some((*material).clone());
         }
         // Producer is bound to the old client; clear it so next send rebuilds.

crates/targets/src/target/redis.rs

@@ -474,8 +474,26 @@ where
                 .publish::<_, _, i64>(self.args.channel.as_str(), body.as_slice())
                 .await
             {
-                Ok(_) => {
-                    debug!(target_id = %self.id, channel = %self.args.channel, attempt, "Event published to Redis channel");
+                Ok(receiver_count) => {
+                    // PUBLISH returns the number of subscribers that received the
+                    // message. Redis pub/sub is best-effort: with zero subscribers
+                    // the event is delivered to no one, yet the durable copy is
+                    // deleted. Warn so operators relying on reliable delivery are
+                    // not silently losing events (backlog#982).
+                    if receiver_count == 0 {
+                        warn!(
+                            target_id = %self.id,
+                            channel = %self.args.channel,
+                            "Redis PUBLISH reached 0 subscribers; the event was not received by any consumer (pub/sub is best-effort)"
+                        );
+                    }
+                    debug!(
+                        target_id = %self.id,
+                        channel = %self.args.channel,
+                        attempt,
+                        receiver_count,
+                        "Event published to Redis channel"
+                    );
                     self.delivery_counters.record_success();
                     return Ok(());
                 }
@@ -528,14 +546,17 @@ where
             return Ok(false);
         }
 
-        let client = self.publisher_client.lock().clone();
-        match tokio::time::timeout(Duration::from_secs(5), ping_redis_server(&client, &self.args)).await {
+        // Reuse the cached ConnectionManager for the health probe (via
+        // ensure_publisher_ready) instead of building a brand-new manager — and
+        // thus a fresh TCP+TLS handshake — on every health check (backlog#982).
+        // ensure_publisher_ready already invalidates the cached manager on a
+        // connectivity error so the next attempt rebuilds it.
+        match tokio::time::timeout(Duration::from_secs(5), self.ensure_publisher_ready()).await {
             Ok(Ok(())) => {
                 self.connected.store(true, Ordering::SeqCst);
                 Ok(true)
             }
             Ok(Err(err)) => {
-                invalidate_cache_on_connectivity_error(&err, || self.invalidate_cached_publisher()).await;
                 mark_target_disconnected_on_connectivity_error(&self.connected, &err);
                 Err(err)
             }
@@ -650,6 +671,13 @@ pub(crate) fn build_redis_client(args: &RedisArgs) -> Result<Client, TargetError
     let mut url = args.url.clone();
     if args.tls.allow_insecure {
         url.set_fragment(Some("insecure"));
+        // Mirror the webhook `skip_tls_verify` warning: this disables Redis
+        // server certificate verification and must only be used for testing
+        // (backlog#982).
+        warn!(
+            target = %redact_redis_url(&args.url),
+            "Redis tls_allow_insecure is enabled: server certificate verification is DISABLED (insecure). Use for testing only."
+        );
     }
 
     let mut connection_info: ConnectionInfo = url.into_connection_info().map_err(map_redis_error)?;

crates/targets/src/target/webhook.rs

@@ -452,6 +452,10 @@ where
         })?;
 
         let status = resp.status();
+        // Drain the response body so the underlying connection is returned to the
+        // pool and can be reused (keep-alive) instead of being closed mid-stream
+        // (backlog#983). The body content is not needed for delivery accounting.
+        let _ = resp.bytes().await;
         if status.is_success() {
             debug!(
                 event = EVENT_WEBHOOK_DELIVERY_STATE,