Fix RC05

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

@@ -2,6 +2,7 @@ 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,
@@ -60,8 +61,12 @@ 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); only B subscribes. B must fire a
-        channel_message_received event carrying the payload intact and tagged with A's senderId.
+        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")
 
@@ -84,30 +89,21 @@ class TestChannelDelivery:
                 "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(CONTENT_TOPIC)
+            assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}"
 
-                subscribe_result = receiver.subscribe_content_topic(CONTENT_TOPIC)
-                assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}"
-                subscribe_result = sender.subscribe_content_topic(CONTENT_TOPIC)
-                assert subscribe_result.is_ok(), f"sender 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()}"
-
-                sender_create = sender.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_A)
-                assert sender_create.is_ok(), f"sender channel_create failed: {sender_create.err()}"
-
-                delay(MESH_SETTLE_S)
-
-                send_result = sender.channel_send(CHANNEL_ID, create_message_bindings(payload=payload_b64))
-                assert send_result.is_ok(), f"channel_send failed: {send_result.err()}"
-                assert send_result.ok_value, f"channel_send returned an empty handle: {send_result.ok_value!r}"
+            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. "