Channel API RC13 wrapper test (#212)

Adds the RC13 close/re-create test: the sender closes the channel and
creates it again under the same id, then sends. The receiver must get the
new message in causal order after the old one, with no replay of the
message that predates the re-create.

Backs it with a close_and_recreate command on the subprocess sender, and
runs the channel lifecycle tests in the wrapper CI job -- they are marked
smoke, but the smoke job ignores tests/wrappers_tests, so that job is the
only place they run.

.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

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

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.