test: validate the channels API against a real peer

The channel test only checked that create/send/close returned values, so
it passed while every send failed downstream: channel_send returns a
request id immediately and reports the outcome asynchronously, and with
no listener on onChannelMessageError there was nothing to notice. Asking
the question properly -- two nodes, one channel, does the payload arrive
-- surfaced three things, none of them caused by the nim-ffi migration.

Encryption was never installed, so no segment could ever be sent; that is
fixed in the vendor by naming the mechanism at channel creation.

Autosharding could not be switched on from here at all. A channel's shard
is derived from its content topic, so without it the send cannot even
auto-subscribe. num-shards-in-network exists on the kernel conf but
WakuNodeConfig never exposed it, so it is added here.

Ingress is still dropped below this layer, so channel_message_reaches_peer
is #[ignore]d with the evidence in its doc comment rather than deleted or
weakened into a test that cannot fail. The single-node test keeps its
narrow scope and now says why it asserts nothing about arrival.

The error listeners stay: a send that fails silently is exactly how this
went unnoticed, and they turn a timeout into a diagnosis.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

waku-bindings/src/node/config.rs

@@ -28,6 +28,11 @@ pub struct WakuNodeConfig {
     /// Cluster id that the node is running in
     #[default(Some(0))]
     pub cluster_id: Option<usize>,
+    /// Number of shards in the cluster. Enables autosharding, which content
+    /// topics need to resolve to a shard; `0` (the default) means static
+    /// sharding, and an explicit shard on every send.
+    #[serde(skip_serializing_if = "Option::is_none", rename = "num-shards-in-network")]
+    pub num_shards_in_network: Option<u16>,
 
     /// Relay protocol
     #[default(Some(true))]

waku-bindings/tests/channels.rs

@@ -1,14 +1,21 @@
 use base64::Engine;
+use regex::Regex;
 use serde::Serialize;
 use serial_test::serial;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
-use waku_bindings::{ChannelMessageSentPayload, LogosDeliveryCtx, WakuNodeConfig};
+use waku_bindings::{
+    ChannelMessageReceivedPayload, ChannelMessageSentPayload, LogosDeliveryCtx, WakuNodeConfig,
+};
 
 const TEST_CHANNEL_ID: &str = "test-channel";
 const TEST_CONTENT_TOPIC: &str = "/test/1/channels/proto";
 const TEST_SENDER_ID: &str = "test-sender";
+const NOOP_ENCRYPTION: &str = "noop";
+const OTHER_SENDER_ID: &str = "other-sender";
+const CHANNEL_PAYLOAD: &[u8] = b"Hi from a reliable channel!";
 const TIMEOUT: Duration = Duration::from_secs(30);
+const SHARDS_IN_NETWORK: usize = 8;
 
 /// Body of `channel_send`, whose payload travels base64-encoded.
 #[derive(Serialize)]
@@ -17,18 +24,47 @@ struct ChannelMessage {
     ephemeral: bool,
 }
 
-#[tokio::test]
-#[serial]
-async fn channel_create_send_close() {
+fn new_node(tcp_port: usize) -> LogosDeliveryCtx {
+    // A channel's shard is derived from its content topic, so the node needs
+    // autosharding on, and has to sit on every shard a topic might hash to.
     let config = serde_json::to_string(&WakuNodeConfig {
-        tcp_port: Some(60070),
+        tcp_port: Some(tcp_port),
+        num_shards_in_network: Some(SHARDS_IN_NETWORK as u16),
+        shards: (0..SHARDS_IN_NETWORK).collect(),
         ..Default::default()
     })
     .expect("config should serialise");
 
-    let node = LogosDeliveryCtx::new_async(config, TIMEOUT)
-        .await
-        .expect("node should instantiate");
+    LogosDeliveryCtx::create(config, TIMEOUT).expect("node should instantiate")
+}
+
+fn channel_message(payload: &[u8]) -> String {
+    serde_json::to_string(&ChannelMessage {
+        payload: base64::engine::general_purpose::STANDARD.encode(payload),
+        ephemeral: false,
+    })
+    .expect("message should serialise")
+}
+
+/// Dials `to` from `from`, rewriting the advertised address to loopback so NAT
+/// or a firewall cannot interfere.
+async fn connect(from: &LogosDeliveryCtx, to: &LogosDeliveryCtx) -> Result<(), String> {
+    let addresses = to.waku_listen_addresses_async().await?;
+    let address = addresses
+        .split(',')
+        .next()
+        .ok_or("peer should report a listen address")?;
+
+    let re = Regex::new(r"\b(?:\d{1,3}\.){3}\d{1,3}\b").unwrap();
+    let address = re.replace_all(address, "127.0.0.1").to_string();
+
+    from.waku_connect_async(address, 10_000).await.map(|_| ())
+}
+
+#[tokio::test]
+#[serial]
+async fn channel_create_send_close() {
+    let node = new_node(60070);
 
     // Channel traffic is reported through events, so capture them to prove the
     // typed listener registers and delivers ChannelMessageSentPayload.
@@ -45,19 +81,14 @@ async fn channel_create_send_close() {
             TEST_CHANNEL_ID.to_string(),
             TEST_CONTENT_TOPIC.to_string(),
             TEST_SENDER_ID.to_string(),
+            NOOP_ENCRYPTION.to_string(),
         )
         .await
         .expect("channel should be created");
     assert_eq!(channel_id, TEST_CHANNEL_ID);
 
-    let message = serde_json::to_string(&ChannelMessage {
-        payload: base64::engine::general_purpose::STANDARD.encode(b"Hi from a reliable channel!"),
-        ephemeral: false,
-    })
-    .expect("message should serialise");
-
     let request_id = node
-        .channel_send_async(TEST_CHANNEL_ID.to_string(), message)
+        .channel_send_async(TEST_CHANNEL_ID.to_string(), channel_message(CHANNEL_PAYLOAD))
         .await
         .expect("channel send should succeed");
     assert!(!request_id.is_empty(), "send should return a request id");
@@ -69,7 +100,10 @@ async fn channel_create_send_close() {
         .await
         .expect("channel should close");
 
-    // Scoped so the guard is dropped before the await below.
+    // Scoped so the guard is dropped before the await below. Nothing is asserted
+    // about arrival: a lone node has no peer to confirm delivery with, so
+    // onChannelMessageSent legitimately never fires. Delivery is covered by
+    // channel_message_reaches_peer.
     {
         let sent = sent.lock().unwrap();
         println!("channel sent events observed: {sent:?}");
@@ -78,3 +112,126 @@ async fn channel_create_send_close() {
     node.stop_node_async().await.expect("node should stop");
     // The context is torn down by LogosDeliveryCtx's Drop impl.
 }
+
+/// Two nodes sharing a channel: a channel message should cross the wire and
+/// surface through the typed onChannelMessageReceived listener.
+///
+/// Ignored: it does not, and the cause is below this binding layer. The message
+/// reaches the receiver's messaging layer with the right content topic and the
+/// `RELIABLE-CHANNEL-API/1` marker -- both of the channel ingress filters pass --
+/// and is then consumed silently by SDS `handleIncoming` (empty result means
+/// parked or duplicate), so no event and no error is ever raised. logos-delivery
+/// has no two-node channel test of its own; its suite fabricates the wire by
+/// emitting a MessageReceivedEvent locally with a stand-in peer's SDS envelope,
+/// so this path looks unexercised rather than regressed.
+///
+/// Un-ignore once the SDS ingress question is settled upstream: everything on
+/// this side (autosharding, per-channel encryption, send, connectivity) works.
+#[tokio::test]
+#[serial]
+#[ignore = "channel ingress is dropped inside SDS; see the doc comment"]
+async fn channel_message_reaches_peer() -> Result<(), String> {
+    let sender = new_node(60080);
+    let receiver = new_node(60090);
+
+    let received: Arc<Mutex<Vec<ChannelMessageReceivedPayload>>> = Arc::new(Mutex::new(Vec::new()));
+    let received_cloned = received.clone();
+    receiver.add_on_channel_message_received_listener(move |event| {
+        received_cloned.lock().unwrap().push(event.clone());
+    });
+
+    // The raw message is observed too: it proves delivery reaches the messaging
+    // layer, which is what localises the failure to channel ingress.
+    let raw: Arc<Mutex<usize>> = Arc::new(Mutex::new(0));
+    let raw_cloned = raw.clone();
+    receiver.add_on_message_received_listener(move |_| {
+        *raw_cloned.lock().unwrap() += 1;
+    });
+
+    // A failed send reports itself here; without this the test could only time
+    // out and say nothing about why.
+    let errors: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
+    let errors_cloned = errors.clone();
+    sender.add_on_channel_message_error_listener(move |event| {
+        errors_cloned.lock().unwrap().push(event.error.clone());
+    });
+
+    let msg_errors: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
+    let msg_errors_cloned = msg_errors.clone();
+    sender.add_on_message_error_listener(move |event| {
+        msg_errors_cloned.lock().unwrap().push(event.error.clone());
+    });
+
+    sender.start_node_async().await?;
+    receiver.start_node_async().await?;
+
+    // Both ends open the same channel on the same content topic, under distinct
+    // participant ids.
+    sender
+        .channel_create_async(
+            TEST_CHANNEL_ID.to_string(),
+            TEST_CONTENT_TOPIC.to_string(),
+            TEST_SENDER_ID.to_string(),
+            NOOP_ENCRYPTION.to_string(),
+        )
+        .await?;
+    receiver
+        .channel_create_async(
+            TEST_CHANNEL_ID.to_string(),
+            TEST_CONTENT_TOPIC.to_string(),
+            OTHER_SENDER_ID.to_string(),
+            NOOP_ENCRYPTION.to_string(),
+        )
+        .await?;
+
+    // Creating a channel does not subscribe: ingress arrives through the
+    // messaging layer, so both ends have to subscribe to the content topic.
+    sender.subscribe_async(TEST_CONTENT_TOPIC.to_string()).await?;
+    receiver
+        .subscribe_async(TEST_CONTENT_TOPIC.to_string())
+        .await?;
+
+    connect(&receiver, &sender).await?;
+
+    // Wait for the mesh to form before publishing, or the message goes nowhere.
+    tokio::time::sleep(Duration::from_secs(5)).await;
+
+    sender
+        .channel_send_async(TEST_CHANNEL_ID.to_string(), channel_message(CHANNEL_PAYLOAD))
+        .await?;
+
+    let mut delivered = None;
+    for _ in 0..100 {
+        if let Some(event) = received.lock().unwrap().first().cloned() {
+            delivered = Some(event);
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+
+    let event = delivered.ok_or_else(|| {
+        let errors = errors.lock().unwrap();
+        let msg_errors = msg_errors.lock().unwrap();
+        let raw = raw.lock().unwrap();
+        format!(
+            "never reached the peer; raw messages seen by receiver: {raw}; \
+             channel errors: {errors:?}; messaging errors: {msg_errors:?}"
+        )
+    })?;
+    assert_eq!(event.channel_id, TEST_CHANNEL_ID);
+    assert_eq!(event.sender_id, TEST_SENDER_ID);
+    let payload = base64::engine::general_purpose::STANDARD
+        .decode(&event.payload)
+        .map_err(|e| e.to_string())?;
+    assert_eq!(payload, CHANNEL_PAYLOAD);
+
+    sender.channel_close_async(TEST_CHANNEL_ID.to_string()).await?;
+    receiver
+        .channel_close_async(TEST_CHANNEL_ID.to_string())
+        .await?;
+
+    sender.stop_node_async().await?;
+    receiver.stop_node_async().await?;
+
+    Ok(())
+}

waku-sys/vendor

@@ -1 +1 @@
-Subproject commit 7080a629da867f37b0e2e5f6012c5cffc150671d
+Subproject commit 1ef53c9506c6662328345404ea58e138278451e1