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>
houseme