stalwartlabs/stalwart · error

Cluster::SubscriberError

Error message

Cluster::SubscriberError

What it means

The Zenoh backend raises Cluster::SubscriberError when `session.declare_subscriber(topic)` fails while creating a subscription for the given key expression. The zenoh error is wrapped into a trc::Error tagged Cluster::SubscriberError.

Source

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

    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 {
    pub async fn next(&mut self) -> Option<Msg> {
        self.subs
            .recv_async()
            .await
            .map(|sample| Msg::Zenoh(sample.payload().to_bytes().into_owned()))
            .ok()
    }
}

View on GitHub (pinned to e962003857)

Solutions

  1. Check the attached reason for the exact zenoh declare_subscriber error.
  2. Validate the topic as a Zenoh key expression (non-empty, valid characters, correct selector syntax if using wildcards).
  3. Ensure the session is open before subscribing and handle reconnects.
  4. Review the Zenoh session config (mode, locators, scouting) if the session itself is failing.

Example fix

// before
let subs = session.declare_subscriber("").await?; // empty key
// after
let subs = session.declare_subscriber("cluster/events").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.subscribe(topic).await {
    Ok(stream) => stream,
    Err(e) => {
        tracing::error!(topic, reason = ?e.reason(), "zenoh declare_subscriber failed");
        return Err(e);
    }
}

Prevention

When it happens

Trigger: Calling `ZenohPubSub::subscribe(topic)` where `declare_subscriber(topic).await` returns Err — invalid key expression or a dead/closed session.

Common situations: Malformed key expression (empty, spaces, bad selectors); subscribing after the Zenoh session closed; Zenoh config pointing at an unreachable router so the session is unusable.

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/6f01a206c08522f1. Report an issue: GitHub.