Merge branch 'master' into CI_Allure_reports_fix

.github/workflows/pr_tests.yml

@@ -209,6 +209,16 @@ jobs:
             --reruns 2 \
             --junit-xml=wrapper-results-corner-cases.xml
 
+      - name: Run wrapper tests - channel delivery
+        continue-on-error: true
+        env:
+          PYTHONPATH: ${{ github.workspace }}/vendor/logos-delivery-python-bindings/waku
+        run: |
+          pytest tests/wrappers_tests/test_channel_delivery.py \
+            -m "not docker_required" \
+            --reruns 2 \
+            --junit-xml=wrapper-results-channel-delivery.xml
+
       - name: Test Report
         if: always()
         uses: dorny/test-reporter@95058abb17504553158e70e2c058fe1fda4392c2

src/node/subprocess_node.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import multiprocessing as mp
+import os
+import queue
+import tempfile
+
+from src.libs.common import delay
+from src.node.wrapper_helpers import EventCollector, create_message_bindings, wait_for_connected
+from src.node.wrappers_manager import WrapperManager
+
+# `spawn`, not `fork`: the child loads its own liblogosdelivery, so it gets an
+# independent process-wide SDS Persistency singleton. Co-located nodes share
+# that singleton and the receiver drops the sender's own message as a duplicate.
+_SPAWN = mp.get_context("spawn")
+
+SENDER_READY_TIMEOUT_S = 120.0
+SENDER_STOP_TIMEOUT_S = 30.0
+
+
+def _sender_worker(config, content_topic, channel_id, sender_id, payload_b64, settle_s, result_q, stop_evt):
+    # chdir before the node starts so the library's default "./data" store
+    # resolves to a private path — separate globals still share ./data/sds.db.
+    os.chdir(tempfile.mkdtemp(prefix="rc_sender_"))
+
+    collector = EventCollector()
+    started = WrapperManager.create_and_start(config=config, event_cb=collector.event_callback)
+    if started.is_err():
+        result_q.put(f"sender start failed: {started.err()}")
+        return
+
+    with started.ok_value as sender:
+        if wait_for_connected(collector) is None:
+            result_q.put("sender did not reach Connected/PartiallyConnected state")
+            return
+
+        subscribe_result = sender.subscribe_content_topic(content_topic)
+        if subscribe_result.is_err():
+            result_q.put(f"sender subscribe_content_topic failed: {subscribe_result.err()}")
+            return
+
+        create_result = sender.channel_create(channel_id, content_topic, sender_id)
+        if create_result.is_err():
+            result_q.put(f"sender channel_create failed: {create_result.err()}")
+            return
+
+        delay(settle_s)
+
+        send_result = sender.channel_send(channel_id, create_message_bindings(payload=payload_b64))
+        if send_result.is_err():
+            result_q.put(f"sender channel_send failed: {send_result.err()}")
+            return
+        if not send_result.ok_value:
+            result_q.put(f"sender channel_send returned an empty handle: {send_result.ok_value!r}")
+            return
+
+        # Signal the message is on the wire, then stay alive so the relay
+        # connection persists while it propagates to the receiver.
+        result_q.put(None)
+        stop_evt.wait()
+
+
+class ChannelSenderProcess:
+    """Run a channel sender node in a separate, storage-isolated process.
+
+    `__enter__` starts the sender and blocks until the message is on the wire
+    (raising on failure/timeout); `__exit__` tears it down. Needed because
+    co-located nodes share the library's process-wide SDS Persistency singleton.
+    """
+
+    def __init__(self, config, *, content_topic, channel_id, sender_id, payload_b64, settle_s):
+        self._args = (config, content_topic, channel_id, sender_id, payload_b64, settle_s)
+        self._stop_evt = _SPAWN.Event()
+        self._result_q = _SPAWN.Queue()
+        self._proc = None
+
+    def __enter__(self) -> "ChannelSenderProcess":
+        self._proc = _SPAWN.Process(
+            target=_sender_worker,
+            args=(*self._args, self._result_q, self._stop_evt),
+            daemon=True,
+        )
+        self._proc.start()
+        try:
+            outcome = self._result_q.get(timeout=SENDER_READY_TIMEOUT_S)
+        except queue.Empty:
+            self.__exit__(None, None, None)
+            raise AssertionError(f"sender subprocess did not report readiness within {SENDER_READY_TIMEOUT_S}s")
+        if outcome is not None:
+            self.__exit__(None, None, None)
+            raise AssertionError(outcome)
+        return self
+
+    def __exit__(self, *_) -> None:
+        self._stop_evt.set()
+        if self._proc is not None:
+            self._proc.join(timeout=SENDER_STOP_TIMEOUT_S)
+            if self._proc.is_alive():
+                self._proc.terminate()
+                self._proc.join()
+            self._proc = None

tests/wrappers_tests/test_channel_delivery.py

@@ -0,0 +1,241 @@
+import base64
+import time
+
+from src.libs.common import delay, to_base64
+from src.node.subprocess_node import ChannelSenderProcess
+from src.node.wrappers_manager import WrapperManager
+from src.node.wrapper_helpers import (
+    EventCollector,
+    create_message_bindings,
+    get_node_multiaddr,
+    wait_for_connected,
+)
+
+CHANNEL_ID = "rc05-channel"
+CONTENT_TOPIC = "/test/1/rc05-channel/proto"
+SENDER_A = "rc05-sender-a"
+SENDER_B = "rc05-sender-b"
+
+CHANNEL_RECEIVED_EVENT = "channel_message_received"
+MESSAGE_RECEIVED_EVENT = "message_received"
+
+RC06_CHANNEL_ID = "rc06-channel"
+RC06_CONTENT_TOPIC = "/test/1/rc06-channel/proto"
+
+CLOSED_CHANNEL_ID = "rc-closed-channel"
+CLOSED_CONTENT_TOPIC = "/test/1/rc-closed-channel/proto"
+
+MESH_SETTLE_S = 10
+DELIVERY_TIMEOUT_S = 50.0
+# Once the unmarked message has provably reached B's messaging layer, the
+# channel ingress decision has already been made on that same event, so this is
+# just a short grace window to catch any late channel_message_received.
+NO_CHANNEL_DELIVERY_WINDOW_S = 10.0
+
+
+def wait_for_channel_received(collector, channel_id, timeout_s, poll_interval_s=0.5):
+    deadline = time.monotonic() + timeout_s
+    while True:
+        for event in collector.snapshot():
+            if event.get("eventType") == CHANNEL_RECEIVED_EVENT and event.get("channelId") == channel_id:
+                return event
+        if time.monotonic() >= deadline:
+            return None
+        time.sleep(poll_interval_s)
+
+
+def wait_for_message_received(collector, content_topic, timeout_s, poll_interval_s=0.5):
+    """Wait for a messaging-layer message_received event on `content_topic`.
+
+    Used to prove a raw relay message actually reached the receiver, independent
+    of the channel layer (the content topic lives under the nested `message`).
+    """
+    deadline = time.monotonic() + timeout_s
+    while True:
+        for event in collector.snapshot():
+            if event.get("eventType") == MESSAGE_RECEIVED_EVENT and (event.get("message") or {}).get("contentTopic") == content_topic:
+                return event
+        if time.monotonic() >= deadline:
+            return None
+        time.sleep(poll_interval_s)
+
+
+class TestChannelDelivery:
+    def test_rc05_basic_a_to_b_delivery(self, node_config):
+        """RC05: A's channel send is delivered to B on the matching channel + topic.
+
+        Relay-only (store + reliability off). B must fire a channel_message_received
+        event carrying the payload intact and tagged with A's senderId.
+
+        A runs in a separate, storage-isolated process: co-located nodes share the
+        library's SDS Persistency singleton and B drops A's own message as a
+        duplicate. See src/node/subprocess_node.py.
+        """
+        payload_b64 = to_base64("rc05 hello from A")
+
+        node_config.update(
+            {
+                "relay": True,
+                "store": False,
+                "reliabilityEnabled": False,
+                "numShardsInNetwork": 1,
+            }
+        )
+
+        receiver_collector = EventCollector()
+        receiver_result = WrapperManager.create_and_start(config=node_config, event_cb=receiver_collector.event_callback)
+        assert receiver_result.is_ok(), f"Failed to start receiver: {receiver_result.err()}"
+
+        with receiver_result.ok_value as receiver:
+            sender_config = {
+                **node_config,
+                "staticnodes": [get_node_multiaddr(receiver)],
+                "portsShift": 1,
+            }
+
+            subscribe_result = receiver.subscribe_content_topic(CONTENT_TOPIC)
+            assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}"
+
+            receiver_create = receiver.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_B)
+            assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}"
+
+            with ChannelSenderProcess(
+                sender_config,
+                content_topic=CONTENT_TOPIC,
+                channel_id=CHANNEL_ID,
+                sender_id=SENDER_A,
+                payload_b64=payload_b64,
+                settle_s=MESH_SETTLE_S,
+            ):
+                received = wait_for_channel_received(receiver_collector, CHANNEL_ID, DELIVERY_TIMEOUT_S)
+                assert received is not None, (
+                    f"No {CHANNEL_RECEIVED_EVENT} on B for {CHANNEL_ID} within {DELIVERY_TIMEOUT_S}s. "
+                    f"Collected events: {receiver_collector.snapshot()}"
+                )
+                assert base64.b64decode(received["payload"]) == base64.b64decode(
+                    payload_b64
+                ), f"delivered payload mismatch: got {received['payload']!r}, sent {payload_b64!r}"
+                assert (
+                    received["senderId"] == SENDER_A
+                ), f"received event must carry the originator's senderId {SENDER_A!r}, got {received.get('senderId')!r}"
+
+    def test_rc06_missing_marker_dropped(self, node_config):
+        """RC06: a plain message on the channel's content topic but without the
+        Reliable-Channel spec marker is dropped at channel ingress.
+
+        A publishes a raw relay message (no RELIABLE-CHANNEL-API/1 meta) on B's
+        content topic. B's messaging layer still surfaces it (message_received),
+        proving the message arrived, but the channel ingress filter drops it: no
+        channel_message_received fires for the channel.
+        """
+        payload_b64 = to_base64("rc06 unmarked message")
+
+        node_config.update(
+            {
+                "relay": True,
+                "store": False,
+                "reliabilityEnabled": False,
+                "numShardsInNetwork": 1,
+            }
+        )
+
+        receiver_collector = EventCollector()
+        receiver_result = WrapperManager.create_and_start(config=node_config, event_cb=receiver_collector.event_callback)
+        assert receiver_result.is_ok(), f"Failed to start receiver: {receiver_result.err()}"
+
+        with receiver_result.ok_value as receiver:
+            sender_config = {
+                **node_config,
+                "staticnodes": [get_node_multiaddr(receiver)],
+                "portsShift": 1,
+            }
+            sender_collector = EventCollector()
+            sender_result = WrapperManager.create_and_start(config=sender_config, event_cb=sender_collector.event_callback)
+            assert sender_result.is_ok(), f"Failed to start sender: {sender_result.err()}"
+
+            with sender_result.ok_value as sender:
+                assert wait_for_connected(sender_collector) is not None, "Sender did not reach Connected/PartiallyConnected state"
+
+                subscribe_result = receiver.subscribe_content_topic(RC06_CONTENT_TOPIC)
+                assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}"
+                subscribe_result = sender.subscribe_content_topic(RC06_CONTENT_TOPIC)
+                assert subscribe_result.is_ok(), f"sender subscribe_content_topic failed: {subscribe_result.err()}"
+
+                # Only B runs a channel; A never marks its traffic.
+                receiver_create = receiver.channel_create(RC06_CHANNEL_ID, RC06_CONTENT_TOPIC, SENDER_B)
+                assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}"
+
+                delay(MESH_SETTLE_S)
+
+                # Plain relay publish: right content topic, but no Reliable-Channel
+                # marker, so B's channel ingress filter must drop it.
+                send_result = sender.send_message(create_message_bindings(contentTopic=RC06_CONTENT_TOPIC, payload=payload_b64))
+                assert send_result.is_ok(), f"send_message failed: {send_result.err()}"
+
+                # It reaches B's messaging layer, so the drop below is the marker
+                # filter at work, not a message that simply never arrived.
+                arrived = wait_for_message_received(receiver_collector, RC06_CONTENT_TOPIC, DELIVERY_TIMEOUT_S)
+                assert arrived is not None, (
+                    f"receiver never saw a {MESSAGE_RECEIVED_EVENT} for {RC06_CONTENT_TOPIC} within {DELIVERY_TIMEOUT_S}s; "
+                    f"cannot conclude the channel dropped it. Collected events: {receiver_collector.snapshot()}"
+                )
+
+                # The channel ingress filter drops the unmarked message: no channel event.
+                leaked = wait_for_channel_received(receiver_collector, RC06_CHANNEL_ID, NO_CHANNEL_DELIVERY_WINDOW_S)
+                assert leaked is None, f"an unmarked message must not surface as a {CHANNEL_RECEIVED_EVENT}; got: {leaked!r}"
+
+    def test_receive_after_close_emits_no_channel_event(self, node_config):
+        """A closed channel must not deliver: B closes its channel, then A sends a
+        valid channel message on the same channel + content topic.
+
+        B stays subscribed to the content topic, so B's messaging layer still
+        surfaces the message (message_received), proving it arrived, but the
+        closed channel must not fire channel_message_received.
+        """
+        payload_b64 = to_base64("message for a closed channel")
+
+        node_config.update(
+            {
+                "relay": True,
+                "store": False,
+                "reliabilityEnabled": False,
+                "numShardsInNetwork": 1,
+            }
+        )
+
+        receiver_collector = EventCollector()
+        receiver_result = WrapperManager.create_and_start(config=node_config, event_cb=receiver_collector.event_callback)
+        assert receiver_result.is_ok(), f"Failed to start receiver: {receiver_result.err()}"
+
+        with receiver_result.ok_value as receiver:
+            sender_config = {
+                **node_config,
+                "staticnodes": [get_node_multiaddr(receiver)],
+                "portsShift": 1,
+            }
+
+            subscribe_result = receiver.subscribe_content_topic(CLOSED_CONTENT_TOPIC)
+            assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}"
+
+            receiver_create = receiver.channel_create(CLOSED_CHANNEL_ID, CLOSED_CONTENT_TOPIC, SENDER_B)
+            assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}"
+
+            receiver_close = receiver.channel_close(CLOSED_CHANNEL_ID)
+            assert receiver_close.is_ok(), f"receiver channel_close failed: {receiver_close.err()}"
+
+            with ChannelSenderProcess(
+                sender_config,
+                content_topic=CLOSED_CONTENT_TOPIC,
+                channel_id=CLOSED_CHANNEL_ID,
+                sender_id=SENDER_A,
+                payload_b64=payload_b64,
+                settle_s=MESH_SETTLE_S,
+            ):
+                arrived = wait_for_message_received(receiver_collector, CLOSED_CONTENT_TOPIC, DELIVERY_TIMEOUT_S)
+                assert arrived is not None, (
+                    f"receiver never saw a {MESSAGE_RECEIVED_EVENT} for {CLOSED_CONTENT_TOPIC} within {DELIVERY_TIMEOUT_S}s; "
+                    f"cannot conclude the closed channel stayed silent. Collected events: {receiver_collector.snapshot()}"
+                )
+
+                leaked = wait_for_channel_received(receiver_collector, CLOSED_CHANNEL_ID, NO_CHANNEL_DELIVERY_WINDOW_S)
+                assert leaked is None, f"a closed channel must not emit {CHANNEL_RECEIVED_EVENT}; got: {leaked!r}"

tests/wrappers_tests/test_channel_lifecycle.py

@@ -1,5 +1,3 @@
-import uuid
-
 import pytest
 from src.node.wrappers_manager import WrapperManager
 from src.node.wrapper_helpers import EventCollector, create_message_bindings
@@ -14,8 +12,7 @@ UNKNOWN_CHANNEL_ID = "rc02-unknown-channel"
 RC03_CHANNEL_ID = "rc03-channel"
 
 RC04_CHANNEL_ID = "rc04-channel"
-RC04_EPHEMERAL_CHANNEL_ID = "rc04-ephemeral-channel"
-RC04_DISTINCT_CHANNEL_ID = "rc04-distinct-channel"
+RC04_CLOSED_CHANNEL_ID = "rc04-closed-channel"
 
 # Give the reliable channel manager a moment to settle a freshly-created
 # channel before we act on it.
@@ -106,41 +103,6 @@ class TestChannelLifecycle:
         finally:
             node.stop_and_destroy()
 
-    def test_rc03_close_random_id_leaves_open_channel_intact(self, node_config):
-        """RC03 variant: closing a random unknown id while a real channel is open.
-
-        A genuine channel is created and left open; closing a random id Errs with
-        "unknown channel: <id>" and must not disturb the open channel, which then
-        still closes cleanly. No events are expected.
-        """
-        node_config.update({"mode": "Core"})
-
-        channel_id = f"rc03-open-{uuid.uuid4()}"
-        random_id = f"rc03-random-{uuid.uuid4()}"
-
-        collector = EventCollector()
-        create_node_result = WrapperManager.create_and_start(config=node_config, event_cb=collector.event_callback)
-        assert create_node_result.is_ok(), f"Failed to create and start node: {create_node_result.err()}"
-        node = create_node_result.ok_value
-
-        try:
-            create_result = node.channel_create(channel_id, CONTENT_TOPIC, SENDER_ID)
-            assert create_result.is_ok(), f"channel_create failed: {create_result.err()}"
-
-            delay(CHANNEL_SETTLE_S)
-
-            random_close_result = node.channel_close(random_id)
-            assert random_close_result.is_err(), f"channel_close on a random id must fail, got Ok({random_close_result.ok_value!r})"
-            assert f"unknown channel: {random_id}" in random_close_result.err(), f"unexpected error message: {random_close_result.err()!r}"
-
-            # The real channel was left untouched and still closes cleanly.
-            close_result = node.channel_close(channel_id)
-            assert close_result.is_ok(), f"open channel_close failed after a bad-id close: {close_result.err()}"
-
-            assert collector.events == [], f"expected no events, got: {collector.events}"
-        finally:
-            node.stop_and_destroy()
-
     def test_rc04_empty_payload_rejected_and_send_returns_handle(self, node_config):
         """RC04: empty payload is rejected; a non-empty send returns a handle synchronously.
 
@@ -170,58 +132,33 @@ class TestChannelLifecycle:
         finally:
             node.stop_and_destroy()
 
-    def test_rc04_ephemeral_send_returns_handle(self, node_config):
-        """RC04 variant: an ephemeral channel_send still returns a handle synchronously.
+    def test_rc04_send_after_close_rejected(self, node_config):
+        """RC04 variant: sending after the channel is closed is rejected.
 
-        The ephemeral flag only affects downstream persistence / terminal events,
-        not the send contract, so channel_send returns Ok(channelReqId) immediately.
+        Exercises the teardown -> send ordering (distinct from RC02's
+        never-created id): once channel_close removes the id, channel_send Errs
+        with "unknown channel: <id>". No events are expected.
         """
         node_config.update({"mode": "Core"})
 
-        create_node_result = WrapperManager.create_and_start(config=node_config)
+        collector = EventCollector()
+        create_node_result = WrapperManager.create_and_start(config=node_config, event_cb=collector.event_callback)
         assert create_node_result.is_ok(), f"Failed to create and start node: {create_node_result.err()}"
         node = create_node_result.ok_value
 
         try:
-            create_result = node.channel_create(RC04_EPHEMERAL_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)
+            create_result = node.channel_create(RC04_CLOSED_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)
             assert create_result.is_ok(), f"channel_create failed: {create_result.err()}"
 
             delay(CHANNEL_SETTLE_S)
 
-            send_result = node.channel_send(RC04_EPHEMERAL_CHANNEL_ID, create_message_bindings(ephemeral=True))
-            assert send_result.is_ok(), f"ephemeral channel_send failed: {send_result.err()}"
-            assert send_result.ok_value, f"channel_send must return a non-empty channelReqId handle, got: {send_result.ok_value!r}"
-        finally:
-            node.stop_and_destroy()
-
-    def test_rc04_sequential_sends_return_distinct_handles(self, node_config):
-        """RC04 variant: back-to-back sends each return a distinct handle.
-
-        Every channel_send is its own request, so two sends on the same channel
-        return two different (non-empty) channelReqId handles.
-        """
-        node_config.update({"mode": "Core"})
-
-        create_node_result = WrapperManager.create_and_start(config=node_config)
-        assert create_node_result.is_ok(), f"Failed to create and start node: {create_node_result.err()}"
-        node = create_node_result.ok_value
-
-        try:
-            create_result = node.channel_create(RC04_DISTINCT_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID)
-            assert create_result.is_ok(), f"channel_create failed: {create_result.err()}"
-
-            delay(CHANNEL_SETTLE_S)
-
-            first_result = node.channel_send(RC04_DISTINCT_CHANNEL_ID, create_message_bindings())
-            assert first_result.is_ok(), f"first channel_send failed: {first_result.err()}"
-            assert first_result.ok_value, f"first channel_send returned an empty handle: {first_result.ok_value!r}"
-
-            second_result = node.channel_send(RC04_DISTINCT_CHANNEL_ID, create_message_bindings())
-            assert second_result.is_ok(), f"second channel_send failed: {second_result.err()}"
-            assert second_result.ok_value, f"second channel_send returned an empty handle: {second_result.ok_value!r}"
-
-            assert (
-                first_result.ok_value != second_result.ok_value
-            ), f"channel_send handles must be distinct, got the same id twice: {first_result.ok_value!r}"
+            close_result = node.channel_close(RC04_CLOSED_CHANNEL_ID)
+            assert close_result.is_ok(), f"channel_close failed: {close_result.err()}"
+
+            send_result = node.channel_send(RC04_CLOSED_CHANNEL_ID, create_message_bindings())
+            assert send_result.is_err(), f"channel_send after close must fail, got Ok({send_result.ok_value!r})"
+            assert f"unknown channel: {RC04_CLOSED_CHANNEL_ID}" in send_result.err(), f"unexpected error message: {send_result.err()!r}"
+
+            assert collector.events == [], f"expected no events, got: {collector.events}"
         finally:
             node.stop_and_destroy()