stalwartlabs/stalwart · error

Cluster::PublisherError

Error message

Cluster::PublisherError

What it means

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)`.

Source

Thrown at crates/coordinator/src/backend/nats/pubsub.rs:21

 *
 * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
 */

use super::NatsPubSub;
use crate::{Msg, PubSubStream};
use futures::StreamExt;
use trc::{ClusterEvent, Error, EventType};

pub struct NatsPubSubStream {
    subs: async_nats::Subscriber,
}

impl NatsPubSub {
    pub async fn publish(&self, topic: &'static str, message: Vec<u8>) -> trc::Result<()> {
        self.client
            .publish(topic, message.into())
            .await
            .map_err(|err| Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err))
    }

    pub async fn subscribe(&self, topic: &'static str) -> trc::Result<PubSubStream> {
        self.client
            .subscribe(topic)
            .await
            .map(|subs| PubSubStream::Nats(NatsPubSubStream { subs }))
            .map_err(|err| {
                Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err)
            })
    }
}

impl NatsPubSubStream {
    pub async fn next(&mut self) -> Option<Msg> {
        self.subs.next().await.map(Msg::Nats)
    }
}

View on GitHub (pinned to e962003857)

Solutions

  1. Read the `.reason` on the trc::Error for the exact async-nats failure (e.g. 'connection closed', 'server disconnected').
  2. Verify the NATS server URL and that the server is running and reachable from this node.
  3. Re-establish the connection / retry publishing; use async-nats reconnect options with retry buffers for transient drops.
  4. Validate the subject string (no spaces; wildcards '*' and '>' are only valid in subscriptions, not always in publishes).

Example fix

// before
let client = async_nats::connect("nats://127.0.0.1").await?;
// after: retry/reconnect handling
let client = async_nats::ConnectOptions::new()
    .retry_on_initial_connect()
    .connect("nats://127.0.0.1:4222").await?;
client.flush().await?; // detect dead connection early
Defensive patterns

Strategy: retry

Validate before calling

// probe connection before publishing
async fn nats_ready(client: &async_nats::Client) -> bool {
    client.publisher().is_none() == false && client.flush().await.is_ok()
}

Try / catch

for attempt in 1..=3 {
    match nats.publish(subject, payload.clone()).await {
        Ok(()) => break,
        Err(e) if attempt < 3 => {
            tracing::warn!(reason = ?e.reason(), "nats publish failed, retrying");
            tokio::time::sleep(Duration::from_millis(100 * 2u64.pow(attempt))).await;
        }
        Err(e) => return Err(e),
    }
}

Prevention

When it happens

Trigger: 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).

Common situations: 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).

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

AI-assisted analysis of stalwartlabs/stalwart@e962003857 (2026-09-06). Data as JSON: /api/errors/dd4dfbaf6cb3c633. Report an issue: GitHub.