{"record":{"id":"9b7f5ebdffb6e397","repo":"stalwartlabs/stalwart","slug":"cluster-publishererror-9b7f5e","errorCode":null,"errorMessage":"Cluster::PublisherError","messagePattern":"Cluster::PublisherError","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/coordinator/src/backend/zenoh/pubsub.rs","lineNumber":21,"sourceCode":" *\n * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL\n */\n\nuse super::ZenohPubSub;\nuse crate::{Msg, PubSubStream};\nuse trc::{ClusterEvent, Error, EventType};\n\npub struct ZenohPubSubStream {\n    subs: zenoh::pubsub::Subscriber<zenoh::handlers::FifoChannelHandler<zenoh::sample::Sample>>,\n}\n\nimpl ZenohPubSub {\n    pub async fn publish(&self, topic: &'static str, message: Vec<u8>) -> trc::Result<()> {\n        self.session\n            .declare_publisher(topic)\n            .await\n            .map_err(|err| {\n                Error::new(EventType::Cluster(ClusterEvent::PublisherError)).reason(err)\n            })?\n            .put(message)\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.session\n            .declare_subscriber(topic)\n            .await\n            .map(|subs| PubSubStream::Zenoh(ZenohPubSubStream { subs }))\n            .map_err(|err| {\n                Error::new(EventType::Cluster(ClusterEvent::SubscriberError)).reason(err)\n            })\n    }\n}\n\nimpl ZenohPubSubStream {","sourceCodeStart":3,"sourceCodeEnd":39,"githubUrl":"https://github.com/stalwartlabs/stalwart/blob/e96200385781a6a9995a8b839ac27d6c75a983ee/crates/coordinator/src/backend/zenoh/pubsub.rs#L3-L39","documentation":"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.","triggerScenarios":"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.","commonSituations":"Malformed Zenoh key expression (empty string, spaces, invalid characters); using the publisher after the Zenoh session was closed; Zenoh router unreachable causing session issues.","solutions":["Read the `.reason` payload for the exact zenoh declaration error.","Validate the topic as a Zenoh key expression: non-empty, slash-separated, no spaces or control characters.","Ensure the Zenoh session is open and healthy (check connect success; reconnect if the session was closed).","Confirm Zenoh configuration (locator/mode) so the session can communicate with a peer/router."],"exampleFix":"// before\nlet session = zenoh::open(Config::default()).await?;\nsession.declare_publisher(\"my key/x\").await?; // space invalid\n// after\nsession.declare_publisher(\"cluster/events/x\").await?;","handlingStrategy":"validation","validationCode":"fn is_valid_zenoh_key(k: &str) -> bool {\n    !k.is_empty()\n        && !k.contains(char::is_whitespace)\n        && k.split('/').all(|seg| !seg.contains(['?', '#', '[', ']']))\n}","typeGuard":null,"tryCatchPattern":"match zenoh.publish(topic, message).await {\n    Ok(()) => (),\n    Err(e) => {\n        tracing::error!(topic, reason = ?e.reason(), \"zenoh declare_publisher failed\");\n        return Err(e);\n    }\n}","preventionTips":["Validate key expressions before calling publish; reject empty or whitespace-containing topics.","Keep the Zenoh session alive for the publisher's lifetime; close them together.","Test key expressions with the z_keyexpr validator utilities when constructing them dynamically."],"tags":["zenoh","pubsub","publisher","key-expression"],"backgroundTag":"invalid-argument-value","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"}