stalwartlabs/stalwart · error

Cluster::PublisherError

Error message

Cluster::PublisherError

What it means

The Zenoh pub-sub backend raises Cluster::PublisherError when `session.declare_publisher(topic)` fails. Declaring the publisher is the first step of publish(); a failure here means Zenoh refused the key expression / publisher declaration, wrapped in a trc::Error with the zenoh error as reason.

Source

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

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

use super::ZenohPubSub;
use crate::{Msg, PubSubStream};
use trc::{ClusterEvent, Error, EventType};

pub struct ZenohPubSubStream {
    subs: zenoh::pubsub::Subscriber<zenoh::handlers::FifoChannelHandler<zenoh::sample::Sample>>,
}

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

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

impl ZenohPubSubStream {

View on GitHub (pinned to e962003857)

Solutions

  1. Read the `.reason` payload for the exact zenoh declaration error.
  2. Validate the topic as a Zenoh key expression: non-empty, slash-separated, no spaces or control characters.
  3. Ensure the Zenoh session is open and healthy (check connect success; reconnect if the session was closed).
  4. Confirm Zenoh configuration (locator/mode) so the session can communicate with a peer/router.

Example fix

// before
let session = zenoh::open(Config::default()).await?;
session.declare_publisher("my key/x").await?; // space invalid
// after
session.declare_publisher("cluster/events/x").await?;
Defensive patterns

Strategy: validation

Validate before calling

fn is_valid_zenoh_key(k: &str) -> bool {
    !k.is_empty()
        && !k.contains(char::is_whitespace)
        && k.split('/').all(|seg| !seg.contains(['?', '#', '[', ']']))
}

Try / catch

match zenoh.publish(topic, message).await {
    Ok(()) => (),
    Err(e) => {
        tracing::error!(topic, reason = ?e.reason(), "zenoh declare_publisher failed");
        return Err(e);
    }
}

Prevention

When it happens

Trigger: Calling `ZenohPubSub::publish(topic, message)` where `declare_publisher(topic).await` returns Err — typically an invalid key expression (topic string) or a session that is closed/invalid.

Common situations: Malformed Zenoh key expression (empty string, spaces, invalid characters); using the publisher after the Zenoh session was closed; Zenoh router unreachable causing session issues.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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