{"record":{"id":"dd4dfbaf6cb3c633","repo":"stalwartlabs/stalwart","slug":"cluster-publishererror-dd4dfb","errorCode":null,"errorMessage":"Cluster::PublisherError","messagePattern":"Cluster::PublisherError","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/coordinator/src/backend/nats/pubsub.rs","lineNumber":21,"sourceCode":" *\n * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL\n */\n\nuse super::NatsPubSub;\nuse crate::{Msg, PubSubStream};\nuse futures::StreamExt;\nuse trc::{ClusterEvent, Error, EventType};\n\npub struct NatsPubSubStream {\n    subs: async_nats::Subscriber,\n}\n\nimpl NatsPubSub {\n    pub async fn publish(&self, topic: &'static str, message: Vec<u8>) -> trc::Result<()> {\n        self.client\n            .publish(topic, message.into())\n            .await\n            .map_err(|err| Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err))\n    }\n\n    pub async fn subscribe(&self, topic: &'static str) -> trc::Result<PubSubStream> {\n        self.client\n            .subscribe(topic)\n            .await\n            .map(|subs| PubSubStream::Nats(NatsPubSubStream { subs }))\n            .map_err(|err| {\n                Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err)\n            })\n    }\n}\n\nimpl NatsPubSubStream {\n    pub async fn next(&mut self) -> Option<Msg> {\n        self.subs.next().await.map(Msg::Nats)\n    }\n}","sourceCodeStart":3,"sourceCodeEnd":39,"githubUrl":"https://github.com/stalwartlabs/stalwart/blob/e96200385781a6a9995a8b839ac27d6c75a983ee/crates/coordinator/src/backend/nats/pubsub.rs#L3-L39","documentation":"The NATS pub-sub backend wraps a failed `client.publish(topic, message)` in a trc::Error tagged Cluster::PublisherError. It means the NATS client could not enqueue or send the message onto the given subject. The underlying async-nats error is attached via `.reason(err)`.","triggerScenarios":"Calling `NatsPubSub::publish(topic, message)` when the NATS connection is closed/disconnected, the client's publish future fails (connection dropped, buffer flush failure, or an invalid subject string).","commonSituations":"NATS server restarted or unreachable between connect and publish; publish called on a client whose connection was closed after a disconnect; subject names with illegal characters (spaces, wildcards in publish subjects).","solutions":["Read the `.reason` on the trc::Error for the exact async-nats failure (e.g. 'connection closed', 'server disconnected').","Verify the NATS server URL and that the server is running and reachable from this node.","Re-establish the connection / retry publishing; use async-nats reconnect options with retry buffers for transient drops.","Validate the subject string (no spaces; wildcards '*' and '>' are only valid in subscriptions, not always in publishes)."],"exampleFix":"// before\nlet client = async_nats::connect(\"nats://127.0.0.1\").await?;\n// after: retry/reconnect handling\nlet client = async_nats::ConnectOptions::new()\n    .retry_on_initial_connect()\n    .connect(\"nats://127.0.0.1:4222\").await?;\nclient.flush().await?; // detect dead connection early","handlingStrategy":"retry","validationCode":"// probe connection before publishing\nasync fn nats_ready(client: &async_nats::Client) -> bool {\n    client.publisher().is_none() == false && client.flush().await.is_ok()\n}","typeGuard":null,"tryCatchPattern":"for attempt in 1..=3 {\n    match nats.publish(subject, payload.clone()).await {\n        Ok(()) => break,\n        Err(e) if attempt < 3 => {\n            tracing::warn!(reason = ?e.reason(), \"nats publish failed, retrying\");\n            tokio::time::sleep(Duration::from_millis(100 * 2u64.pow(attempt))).await;\n        }\n        Err(e) => return Err(e),\n    }\n}","preventionTips":["Build the client with ConnectOptions::retry_on_initial_connect and generous reconnect buffers.","Call flush() periodically to detect dead connections early.","Monitor NATS server health and alert on disconnect events."],"tags":["nats","pubsub","publisher","network"],"backgroundTag":"connection-refused","analyzedSha":"e96200385781a6a9995a8b839ac27d6c75a983ee","analyzedAt":"2026-09-06T22:07:17.982Z","contentChangedAt":"2026-09-06T22:07:17.982Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}