diff --git a/.github/workflows/pr_tests.yml b/.github/workflows/pr_tests.yml index 8321f573f..8baa106a2 100644 --- a/.github/workflows/pr_tests.yml +++ b/.github/workflows/pr_tests.yml @@ -219,6 +219,18 @@ jobs: --reruns 2 \ --junit-xml=wrapper-results-channel-delivery.xml + # Marked smoke, but the smoke job ignores tests/wrappers_tests, so this + # job is the only place the channel lifecycle tests run. + - name: Run wrapper tests - channel lifecycle + continue-on-error: true + env: + PYTHONPATH: ${{ github.workspace }}/vendor/logos-delivery-python-bindings/waku + run: | + pytest tests/wrappers_tests/test_channel_lifecycle.py \ + -m "not docker_required" \ + --reruns 2 \ + --junit-xml=wrapper-results-channel-lifecycle.xml + - name: Test Report if: always() uses: dorny/test-reporter@a43b3a5f7366b97d083190328d2c652e1a8b6aa2 # v3.0.0 diff --git a/src/node/subprocess_node.py b/src/node/subprocess_node.py index 0ebaff5c0..94ba1a709 100644 --- a/src/node/subprocess_node.py +++ b/src/node/subprocess_node.py @@ -20,6 +20,9 @@ SENDER_STOP_TIMEOUT_S = 30.0 SEND_ACK_TIMEOUT_S = 60.0 _SERVE_POLL_S = 0.2 +CMD_SEND = "send" +CMD_RECREATE = "recreate" + def _send(sender, channel_id, payload_b64): """Sends on the channel; returns None on success, an error string otherwise.""" @@ -31,6 +34,18 @@ def _send(sender, channel_id, payload_b64): return None +def _recreate(sender, channel_id, content_topic, sender_id): + """Closes the channel and creates it again under the same id.""" + close_result = sender.channel_close(channel_id) + if close_result.is_err(): + return f"sender channel_close failed: {close_result.err()}" + + create_result = sender.channel_create(channel_id, content_topic, sender_id) + if create_result.is_err(): + return f"sender channel_create failed: {create_result.err()}" + return None + + def _sender_worker(config, content_topic, channel_id, sender_id, payload_b64, settle_s, result_q, cmd_q, evt_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. @@ -78,10 +93,21 @@ def _sender_worker(config, content_topic, channel_id, sender_id, payload_b64, se forwarded = len(received) try: - next_payload = cmd_q.get(timeout=_SERVE_POLL_S) + op, arg = cmd_q.get(timeout=_SERVE_POLL_S) except queue.Empty: continue - result_q.put(_send(sender, channel_id, next_payload)) + + if op == CMD_SEND: + result_q.put(_send(sender, channel_id, arg)) + continue + + outcome = _recreate(sender, channel_id, content_topic, sender_id) + if outcome is None: + # channel_close drops the content topic subscription and + # channel_create re-adds it; let the mesh catch up before the + # next send, or it goes out to nobody. + delay(settle_s) + result_q.put(outcome) class ChannelSenderProcess: @@ -92,7 +118,8 @@ class ChannelSenderProcess: co-located nodes share the library's SDS Persistency singleton. `multiaddr` lets the node under test dial this peer, `send()` drives further - sends, `wait_for_received()` reports what this peer received. + sends, `close_and_recreate()` cycles the channel under the same id, and + `wait_for_received()` reports what this peer received. """ def __init__(self, config, *, content_topic, channel_id, sender_id, payload_b64=None, settle_s): @@ -123,16 +150,24 @@ class ChannelSenderProcess: self.multiaddr = outcome["multiaddr"] return self - def send(self, payload_b64) -> None: - """Sends another message on the channel, blocking until the peer acks it.""" - self._cmd_q.put(payload_b64) + def _run(self, op, arg, what) -> None: + self._cmd_q.put((op, arg)) try: outcome = self._result_q.get(timeout=SEND_ACK_TIMEOUT_S) except queue.Empty: - raise AssertionError(f"sender subprocess did not acknowledge a send within {SEND_ACK_TIMEOUT_S}s") + raise AssertionError(f"sender subprocess did not acknowledge {what} within {SEND_ACK_TIMEOUT_S}s") if outcome is not None: raise AssertionError(outcome) + def send(self, payload_b64) -> None: + """Sends another message on the channel, blocking until the peer acks it.""" + self._run(CMD_SEND, payload_b64, "a send") + + def close_and_recreate(self) -> None: + """Closes and re-creates the channel under the same id, blocking until + the peer has re-subscribed and the mesh has settled.""" + self._run(CMD_RECREATE, None, "a close/re-create") + def wait_for_received(self, count, timeout_s) -> list: """Channel messages this peer received, oldest first; waits for `count`.""" deadline = time.monotonic() + timeout_s diff --git a/tests/wrappers_tests/test_channel_delivery.py b/tests/wrappers_tests/test_channel_delivery.py index dbb2fac22..be18290aa 100644 --- a/tests/wrappers_tests/test_channel_delivery.py +++ b/tests/wrappers_tests/test_channel_delivery.py @@ -45,6 +45,9 @@ RC12_CHANNEL_PREFIX = "rc12-channel" RC12_CONTENT_TOPIC = "/test/1/rc12-channel/proto" SENDER_C = "rc12-sender-c" +RC13_CHANNEL_PREFIX = "rc13-channel" +RC13_CONTENT_TOPIC = "/test/1/rc13-channel/proto" + CLOSED_CHANNEL_PREFIX = "rc-closed-channel" CLOSED_CONTENT_TOPIC = "/test/1/rc-closed-channel/proto" @@ -574,6 +577,78 @@ class TestChannelDelivery: SENDER_A, ], f"both events must carry {SENDER_A!r}, got {[e.get('senderId') for e in on_b + on_c]!r}" + def test_rc13_close_recreate_then_send_delivers_new_message(self, node_config): + """RC13: A closes and re-creates its channel, then sends again. + + A sends m1, cycles the channel under the same id, then sends m2. B must + deliver m2 exactly once and ordered after m1, and must not replay m1 — + the re-created channel picks up the restored SDS history rather than + starting a fresh one. + + Distinct from the nim in-process test, which cycles the *receiver's* + channel and asserts a replayed m1 is suppressed on ingress; here the + cycle is on the send path. + """ + channel_id = unique_channel_id(RC13_CHANNEL_PREFIX) + m1, m2 = "rc13 before close", "rc13 after re-create" + + 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(RC13_CONTENT_TOPIC) + assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}" + + receiver_create = receiver.channel_create(channel_id, RC13_CONTENT_TOPIC, SENDER_B) + assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}" + + with ChannelSenderProcess( + sender_config, + content_topic=RC13_CONTENT_TOPIC, + channel_id=channel_id, + sender_id=SENDER_A, + payload_b64=to_base64(m1), + settle_s=MESH_SETTLE_S, + ) as sender: + first = wait_for_channel_received(receiver_collector, channel_id, DELIVERY_TIMEOUT_S) + assert first is not None, ( + f"No {CHANNEL_RECEIVED_EVENT} for m1 on {channel_id} within {DELIVERY_TIMEOUT_S}s; " + f"the close/re-create is only meaningful once m1 landed. Collected events: {receiver_collector.snapshot()}" + ) + + sender.close_and_recreate() + sender.send(to_base64(m2)) + + received = wait_for_channel_received_count(receiver_collector, channel_id, 2, DELIVERY_TIMEOUT_S) + assert len(received) == 2, ( + f"expected m1 then m2 on {channel_id}, got {len(received)}: {channel_payloads(received)!r}. " + f"Collected events: {receiver_collector.snapshot()}" + ) + assert channel_payloads(received) == [ + m1.encode(), + m2.encode(), + ], f"expected [m1, m2] in causal order, got: {channel_payloads(received)!r}" + + # A third event could only be a replayed m1 from the re-created channel. + settled = wait_for_channel_received_count(receiver_collector, channel_id, 3, NO_CHANNEL_DELIVERY_WINDOW_S) + assert len(settled) == 2, f"re-create must not re-deliver m1; got: {channel_payloads(settled)!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.