diff --git a/.github/workflows/pr_tests.yml b/.github/workflows/pr_tests.yml index 2ddf4735b..a04f30dec 100644 --- a/.github/workflows/pr_tests.yml +++ b/.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 diff --git a/src/node/subprocess_node.py b/src/node/subprocess_node.py new file mode 100644 index 000000000..7722a1b65 --- /dev/null +++ b/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 diff --git a/tests/wrappers_tests/test_channel_delivery.py b/tests/wrappers_tests/test_channel_delivery.py new file mode 100644 index 000000000..8b1ad1563 --- /dev/null +++ b/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}" diff --git a/tests/wrappers_tests/test_channel_lifecycle.py b/tests/wrappers_tests/test_channel_lifecycle.py index fe1e3663a..937fd8000 100644 --- a/tests/wrappers_tests/test_channel_lifecycle.py +++ b/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: " 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: ". 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()