diff --git a/src/node/subprocess_node.py b/src/node/subprocess_node.py index 8d41d51ee..0ebaff5c0 100644 --- a/src/node/subprocess_node.py +++ b/src/node/subprocess_node.py @@ -43,8 +43,7 @@ def _sender_worker(config, content_topic, channel_id, sender_id, payload_b64, se return with started.ok_value as sender: - # Without staticnodes this peer is the first node up and has nobody to - # dial: the node under test joins later and dials `multiaddr` instead. + # No staticnodes: this peer is first up, the node under test dials it later. if config.get("staticnodes") and wait_for_connected(collector) is None: result_q.put("sender did not reach Connected/PartiallyConnected state") return @@ -67,9 +66,8 @@ def _sender_worker(config, content_topic, channel_id, sender_id, payload_b64, se result_q.put(outcome) return - # Signal readiness (and how to dial us), then stay alive so the relay - # connection persists while messages propagate — and so SDS keeps - # answering repair requests for what this peer already sent. + # Signal readiness, then stay alive so the relay connection persists + # while messages propagate and SDS answers repair requests. result_q.put({"multiaddr": get_node_multiaddr(sender)}) forwarded = 0 @@ -89,14 +87,12 @@ def _sender_worker(config, content_topic, channel_id, sender_id, payload_b64, se class ChannelSenderProcess: """Run a channel peer node in a separate, storage-isolated process. - `__enter__` starts the peer and blocks until it is ready, including its - initial `payload_b64` send when one is given (raising on failure/timeout); - `__exit__` tears it down. Needed because co-located nodes share the - library's process-wide SDS Persistency singleton. + `__enter__` blocks until the peer is ready, including its initial + `payload_b64` send when given; `__exit__` tears it down. Needed because + co-located nodes share the library's SDS Persistency singleton. - `multiaddr` lets the node under test dial this peer when it joins later, - `send()` drives further sends, and `wait_for_received()` reports the channel - messages this peer itself received. + `multiaddr` lets the node under test dial this peer, `send()` drives further + sends, `wait_for_received()` reports what this peer received. """ def __init__(self, config, *, content_topic, channel_id, sender_id, payload_b64=None, settle_s): diff --git a/src/node/wrapper_helpers.py b/src/node/wrapper_helpers.py index 2afe6b250..6f776af1d 100644 --- a/src/node/wrapper_helpers.py +++ b/src/node/wrapper_helpers.py @@ -5,6 +5,7 @@ import json import re import threading import time +import uuid from typing import Optional from src.libs.common import to_base64 @@ -245,6 +246,15 @@ def enr_udp_port(enr_uri: str) -> int: return int.from_bytes(payload[key + 5 : key + 5 + size], "big") +def unique_channel_id(prefix: str) -> str: + """A per-run channel id. + + SDS state is persisted per channelId and restored on create, so a fixed id + leaks one run's causal history into the next. + """ + return f"{prefix}-{uuid.uuid4().hex[:8]}" + + def create_message_bindings(**overrides) -> dict: envelope = { "contentTopic": DEFAULT_CONTENT_TOPIC, diff --git a/tests/wrappers_tests/test_channel_delivery.py b/tests/wrappers_tests/test_channel_delivery.py index d48726167..d98777956 100644 --- a/tests/wrappers_tests/test_channel_delivery.py +++ b/tests/wrappers_tests/test_channel_delivery.py @@ -8,10 +8,11 @@ from src.node.wrapper_helpers import ( EventCollector, create_message_bindings, get_node_multiaddr, + unique_channel_id, wait_for_connected, ) -CHANNEL_ID = "rc05-channel" +RC05_CHANNEL_PREFIX = "rc05-channel" CONTENT_TOPIC = "/test/1/rc05-channel/proto" SENDER_A = "rc05-sender-a" SENDER_B = "rc05-sender-b" @@ -19,24 +20,23 @@ SENDER_B = "rc05-sender-b" CHANNEL_RECEIVED_EVENT = "channel_message_received" MESSAGE_RECEIVED_EVENT = "message_received" -RC06_CHANNEL_ID = "rc06-channel" +RC06_CHANNEL_PREFIX = "rc06-channel" RC06_CONTENT_TOPIC = "/test/1/rc06-channel/proto" -RC07_CHANNEL_ID = "rc07-channel" +RC07_CHANNEL_PREFIX = "rc07-channel" RC07_CONTENT_TOPIC = "/test/1/rc07-channel/proto" RC07_OTHER_CONTENT_TOPIC = "/test/1/rc07-other/proto" -RC08_CHANNEL_ID = "rc08-channel" +RC08_CHANNEL_PREFIX = "rc08-channel" RC08_CONTENT_TOPIC = "/test/1/rc08-channel/proto" -# A reply only references the message it answers once that message has landed in -# the replier's SDS history. Replying immediately races that write and produces a -# reply with an empty causal history, which asserts nothing about ordering. +# A reply references what it answers only once that landed in the sender's SDS +# history; replying immediately races that write and asserts nothing. RC08_CAUSAL_SETTLE_S = 5 -RC09_CHANNEL_ID = "rc09-channel" +RC09_CHANNEL_PREFIX = "rc09-channel" RC09_CONTENT_TOPIC = "/test/1/rc09-channel/proto" -CLOSED_CHANNEL_ID = "rc-closed-channel" +CLOSED_CHANNEL_PREFIX = "rc-closed-channel" CLOSED_CONTENT_TOPIC = "/test/1/rc-closed-channel/proto" MESH_SETTLE_S = 10 @@ -46,10 +46,10 @@ DELIVERY_TIMEOUT_S = 50.0 # just a short grace window to catch any late channel_message_received. NO_CHANNEL_DELIVERY_WINDOW_S = 10.0 -# A never gets a peer to dial before it sends, so there is no mesh to settle. +# A has no peer to dial before it sends, so no mesh to settle. RC09_SENDER_SETTLE_S = 2 -# DefaultRepairTMin in nim-sds: a repair request only becomes eligible to ride -# on an outgoing message once T_min has passed since the dependency went missing. +# nim-sds DefaultRepairTMin: a repair request rides on an outgoing message only +# after T_min has passed. RC09_REPAIR_REQUEST_DELAY_S = 35 RC09_RECOVERY_TIMEOUT_S = 90.0 @@ -111,6 +111,7 @@ class TestChannelDelivery: library's SDS Persistency singleton and B drops A's own message as a duplicate. See src/node/subprocess_node.py. """ + channel_id = unique_channel_id(RC05_CHANNEL_PREFIX) payload_b64 = to_base64("rc05 hello from A") node_config.update( @@ -136,20 +137,20 @@ class TestChannelDelivery: 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) + 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, + 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) + 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"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( @@ -168,6 +169,7 @@ class TestChannelDelivery: proving the message arrived, but the channel ingress filter drops it: no channel_message_received fires for the channel. """ + channel_id = unique_channel_id(RC06_CHANNEL_PREFIX) payload_b64 = to_base64("rc06 unmarked message") node_config.update( @@ -202,7 +204,7 @@ class TestChannelDelivery: 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) + receiver_create = receiver.channel_create(channel_id, RC06_CONTENT_TOPIC, SENDER_B) assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}" delay(MESH_SETTLE_S) @@ -221,7 +223,7 @@ class TestChannelDelivery: ) # 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) + leaked = wait_for_channel_received(receiver_collector, 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_rc07_wrong_content_topic_dropped(self, node_config): @@ -233,6 +235,7 @@ class TestChannelDelivery: message (message_received), proving it arrived, but B's channel is bound to RC07_CONTENT_TOPIC and must not fire channel_message_received. """ + channel_id = unique_channel_id(RC07_CHANNEL_PREFIX) payload_b64 = to_base64("rc07 message on the wrong content topic") node_config.update( @@ -258,13 +261,13 @@ class TestChannelDelivery: subscribe_result = receiver.subscribe_content_topic(content_topic) assert subscribe_result.is_ok(), f"receiver subscribe_content_topic {content_topic} failed: {subscribe_result.err()}" - receiver_create = receiver.channel_create(RC07_CHANNEL_ID, RC07_CONTENT_TOPIC, SENDER_B) + receiver_create = receiver.channel_create(channel_id, RC07_CONTENT_TOPIC, SENDER_B) assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}" with ChannelSenderProcess( sender_config, content_topic=RC07_OTHER_CONTENT_TOPIC, - channel_id=RC07_CHANNEL_ID, + channel_id=channel_id, sender_id=SENDER_A, payload_b64=payload_b64, settle_s=MESH_SETTLE_S, @@ -275,23 +278,23 @@ class TestChannelDelivery: f"cannot conclude the channel dropped it. Collected events: {receiver_collector.snapshot()}" ) - leaked = wait_for_channel_received(receiver_collector, RC07_CHANNEL_ID, NO_CHANNEL_DELIVERY_WINDOW_S) + leaked = wait_for_channel_received(receiver_collector, channel_id, NO_CHANNEL_DELIVERY_WINDOW_S) assert leaked is None, f"a message on a foreign content topic must not surface as a {CHANNEL_RECEIVED_EVENT}; got: {leaked!r}" def test_rc08_bidirectional_exchange_preserves_causal_order(self, node_config): """RC08: A and B exchange interleaved messages on one channel, and each - side delivers the other's via channel_message_received in causal order. + side delivers the other's in causal order. - A sends m1; B replies m2 only once m1 has landed, so m2's SDS causal - history references m1; A then sends m3, which in turn references m2. B - must deliver m1 then m3, and A must deliver m2. + Each reply is sent only after the message it answers has landed, so m2 + references m1 and m3 references m2. B must deliver m1 then m3, A must + deliver m2. """ + channel_id = unique_channel_id(RC08_CHANNEL_PREFIX) m1, m2, m3 = "rc08 from A", "rc08 reply from B", "rc08 follow-up from A" node_config.update( { "relay": True, - "store": False, "reliabilityEnabled": False, "numShardsInNetwork": 1, } @@ -311,18 +314,18 @@ class TestChannelDelivery: subscribe_result = receiver.subscribe_content_topic(RC08_CONTENT_TOPIC) assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}" - receiver_create = receiver.channel_create(RC08_CHANNEL_ID, RC08_CONTENT_TOPIC, SENDER_B) + receiver_create = receiver.channel_create(channel_id, RC08_CONTENT_TOPIC, SENDER_B) assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}" with ChannelSenderProcess( sender_config, content_topic=RC08_CONTENT_TOPIC, - channel_id=RC08_CHANNEL_ID, + 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, RC08_CHANNEL_ID, DELIVERY_TIMEOUT_S) + first = wait_for_channel_received(receiver_collector, channel_id, DELIVERY_TIMEOUT_S) assert first is not None, ( f"B never saw A's first message within {DELIVERY_TIMEOUT_S}s; " f"nothing to reply to. Collected events: {receiver_collector.snapshot()}" @@ -330,31 +333,33 @@ class TestChannelDelivery: # B replies only now, so m2 carries m1 in its causal history. delay(RC08_CAUSAL_SETTLE_S) - reply_result = receiver.channel_send(RC08_CHANNEL_ID, create_message_bindings(payload=to_base64(m2))) + reply_result = receiver.channel_send(channel_id, create_message_bindings(payload=to_base64(m2))) assert reply_result.is_ok(), f"receiver channel_send failed: {reply_result.err()}" delivered_to_a = sender.wait_for_received(1, DELIVERY_TIMEOUT_S) assert channel_payloads(delivered_to_a) == [m2.encode()], f"A must deliver B's reply; got {channel_payloads(delivered_to_a)!r}" + # Same race on A's side. + delay(RC08_CAUSAL_SETTLE_S) sender.send(to_base64(m3)) - delivered_to_b = wait_for_channel_received_count(receiver_collector, RC08_CHANNEL_ID, 2, DELIVERY_TIMEOUT_S) + delivered_to_b = wait_for_channel_received_count(receiver_collector, channel_id, 2, DELIVERY_TIMEOUT_S) assert channel_payloads(delivered_to_b) == [m1.encode(), m3.encode()], ( f"B must deliver A's messages in causal order; got {channel_payloads(delivered_to_b)!r}. " f"Collected events: {receiver_collector.snapshot()}" ) def test_rc09_offline_receiver_recovers_via_sds_repair(self, node_config): - """RC09: A sends while B is not up yet; once B joins it still ends up - delivering the missed message, via SDS-R repair. + """RC09: A sends while B is not up yet; B joins and still delivers the + missed message, via SDS-R repair. - Nothing re-publishes a message on its own — the unacked-buffer sweep only - counts attempts and never re-emits, and the periodic-sync callback is - unwired — so recovery runs through SDS-R: B learns m1 exists from m2's - causal history and parks m2, attaches a repair request to its own next - send once T_min has passed, and A answers by rebroadcasting m1, which - releases the parked m2. B must end up with m1 then m2. + Nothing re-publishes on its own (the unacked sweep only counts attempts, + the periodic-sync callback is unwired), so recovery goes: B learns m1 + exists from m2's causal history and parks m2, rides a repair request on + its own next send, A rebroadcasts m1, the parked m2 is released. B must + end up with m1 then m2. """ + channel_id = unique_channel_id(RC09_CHANNEL_PREFIX) m1, m2, m3 = "rc09 sent while B is away", "rc09 sent after B joins", "rc09 from B" node_config.update( @@ -371,7 +376,7 @@ class TestChannelDelivery: with ChannelSenderProcess( sender_config, content_topic=RC09_CONTENT_TOPIC, - channel_id=RC09_CHANNEL_ID, + channel_id=channel_id, sender_id=SENDER_A, payload_b64=to_base64(m1), settle_s=RC09_SENDER_SETTLE_S, @@ -387,22 +392,21 @@ class TestChannelDelivery: subscribe_result = receiver.subscribe_content_topic(RC09_CONTENT_TOPIC) assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}" - receiver_create = receiver.channel_create(RC09_CHANNEL_ID, RC09_CONTENT_TOPIC, SENDER_B) + receiver_create = receiver.channel_create(channel_id, RC09_CONTENT_TOPIC, SENDER_B) assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}" delay(MESH_SETTLE_S) - # m2 is B's only clue that it missed m1: SDS parks m2 and books a - # repair request for the dependency it cannot resolve. + # m2 is B's only clue it missed m1: SDS parks it and books a repair request. sender.send(to_base64(m2)) delay(RC09_REPAIR_REQUEST_DELAY_S) - # The request only travels attached to B's own outgoing traffic. - repair_carrier = receiver.channel_send(RC09_CHANNEL_ID, create_message_bindings(payload=to_base64(m3))) + # The request only travels on B's own outgoing traffic. + repair_carrier = receiver.channel_send(channel_id, create_message_bindings(payload=to_base64(m3))) assert repair_carrier.is_ok(), f"receiver channel_send failed: {repair_carrier.err()}" - recovered = wait_for_channel_received_count(receiver_collector, RC09_CHANNEL_ID, 2, RC09_RECOVERY_TIMEOUT_S) + recovered = wait_for_channel_received_count(receiver_collector, channel_id, 2, RC09_RECOVERY_TIMEOUT_S) assert channel_payloads(recovered) == [m1.encode(), m2.encode()], ( f"B must recover the message it missed and release the parked one, in causal order; " f"got {channel_payloads(recovered)!r}. Collected events: {receiver_collector.snapshot()}" @@ -416,6 +420,7 @@ class TestChannelDelivery: surfaces the message (message_received), proving it arrived, but the closed channel must not fire channel_message_received. """ + channel_id = unique_channel_id(CLOSED_CHANNEL_PREFIX) payload_b64 = to_base64("message for a closed channel") node_config.update( @@ -441,16 +446,16 @@ class TestChannelDelivery: 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) + receiver_create = receiver.channel_create(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) + receiver_close = receiver.channel_close(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, + channel_id=channel_id, sender_id=SENDER_A, payload_b64=payload_b64, settle_s=MESH_SETTLE_S, @@ -461,5 +466,5 @@ class TestChannelDelivery: 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) + leaked = wait_for_channel_received(receiver_collector, 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 937fd8000..04d84394a 100644 --- a/tests/wrappers_tests/test_channel_lifecycle.py +++ b/tests/wrappers_tests/test_channel_lifecycle.py @@ -1,18 +1,18 @@ import pytest from src.node.wrappers_manager import WrapperManager -from src.node.wrapper_helpers import EventCollector, create_message_bindings +from src.node.wrapper_helpers import EventCollector, create_message_bindings, unique_channel_id from src.libs.common import delay -CHANNEL_ID = "rc01-channel" +RC01_CHANNEL_PREFIX = "rc01-channel" CONTENT_TOPIC = "/test/1/channel/proto" SENDER_ID = "rc01-sender" -UNKNOWN_CHANNEL_ID = "rc02-unknown-channel" +UNKNOWN_CHANNEL_PREFIX = "rc02-unknown-channel" -RC03_CHANNEL_ID = "rc03-channel" +RC03_CHANNEL_PREFIX = "rc03-channel" -RC04_CHANNEL_ID = "rc04-channel" -RC04_CLOSED_CHANNEL_ID = "rc04-closed-channel" +RC04_CHANNEL_PREFIX = "rc04-channel" +RC04_CLOSED_CHANNEL_PREFIX = "rc04-closed-channel" # Give the reliable channel manager a moment to settle a freshly-created # channel before we act on it. @@ -28,6 +28,7 @@ class TestChannelLifecycle: node, so no store / --reliability is needed. No events are expected. """ node_config.update({"mode": "Core"}) + channel_id = unique_channel_id(RC01_CHANNEL_PREFIX) collector = EventCollector() create_node_result = WrapperManager.create_and_start(config=node_config, event_cb=collector.event_callback) @@ -35,13 +36,13 @@ class TestChannelLifecycle: node = create_node_result.ok_value try: - create_result = node.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_ID) + create_result = node.channel_create(channel_id, CONTENT_TOPIC, SENDER_ID) assert create_result.is_ok(), f"channel_create failed: {create_result.err()}" - assert create_result.ok_value == CHANNEL_ID, f"channel_create returned unexpected id: {create_result.ok_value!r}" + assert create_result.ok_value == channel_id, f"channel_create returned unexpected id: {create_result.ok_value!r}" - duplicate_result = node.channel_create(CHANNEL_ID, CONTENT_TOPIC, SENDER_ID) + duplicate_result = node.channel_create(channel_id, CONTENT_TOPIC, SENDER_ID) assert duplicate_result.is_err(), f"duplicate channel_create must fail, got Ok({duplicate_result.ok_value!r})" - assert f"channel already exists: {CHANNEL_ID}" in duplicate_result.err(), f"unexpected error message: {duplicate_result.err()!r}" + assert f"channel already exists: {channel_id}" in duplicate_result.err(), f"unexpected error message: {duplicate_result.err()!r}" assert collector.events == [], f"expected no events, got: {collector.events}" finally: @@ -53,6 +54,7 @@ class TestChannelLifecycle: The send must Err with "unknown channel: " and emit no events. """ node_config.update({"mode": "Core"}) + unknown_channel_id = unique_channel_id(UNKNOWN_CHANNEL_PREFIX) collector = EventCollector() create_node_result = WrapperManager.create_and_start(config=node_config, event_cb=collector.event_callback) @@ -60,9 +62,9 @@ class TestChannelLifecycle: node = create_node_result.ok_value try: - send_result = node.channel_send(UNKNOWN_CHANNEL_ID, create_message_bindings()) + send_result = node.channel_send(unknown_channel_id, create_message_bindings()) assert send_result.is_err(), f"channel_send on unknown channel must fail, got Ok({send_result.ok_value!r})" - assert f"unknown channel: {UNKNOWN_CHANNEL_ID}" in send_result.err(), f"unexpected error message: {send_result.err()!r}" + assert f"unknown channel: {unknown_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: @@ -76,6 +78,8 @@ class TestChannelLifecycle: Lifecycle teardown + idempotency guard. No events are expected. """ node_config.update({"mode": "Core"}) + channel_id = unique_channel_id(RC03_CHANNEL_PREFIX) + unknown_channel_id = unique_channel_id(UNKNOWN_CHANNEL_PREFIX) collector = EventCollector() create_node_result = WrapperManager.create_and_start(config=node_config, event_cb=collector.event_callback) @@ -83,21 +87,21 @@ class TestChannelLifecycle: node = create_node_result.ok_value try: - create_result = node.channel_create(RC03_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID) + 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) - close_result = node.channel_close(RC03_CHANNEL_ID) + close_result = node.channel_close(channel_id) assert close_result.is_ok(), f"channel_close on an existing channel failed: {close_result.err()}" - reclose_result = node.channel_close(RC03_CHANNEL_ID) + reclose_result = node.channel_close(channel_id) assert reclose_result.is_err(), f"second channel_close must fail, got Ok({reclose_result.ok_value!r})" - assert f"unknown channel: {RC03_CHANNEL_ID}" in reclose_result.err(), f"unexpected error message: {reclose_result.err()!r}" + assert f"unknown channel: {channel_id}" in reclose_result.err(), f"unexpected error message: {reclose_result.err()!r}" - unknown_close_result = node.channel_close(UNKNOWN_CHANNEL_ID) + unknown_close_result = node.channel_close(unknown_channel_id) assert unknown_close_result.is_err(), f"channel_close on unknown channel must fail, got Ok({unknown_close_result.ok_value!r})" - assert f"unknown channel: {UNKNOWN_CHANNEL_ID}" in unknown_close_result.err(), f"unexpected error message: {unknown_close_result.err()!r}" + assert f"unknown channel: {unknown_channel_id}" in unknown_close_result.err(), f"unexpected error message: {unknown_close_result.err()!r}" assert collector.events == [], f"expected no events, got: {collector.events}" finally: @@ -111,22 +115,23 @@ class TestChannelLifecycle: Input validation + the immediate-handle contract of send. """ node_config.update({"mode": "Core"}) + channel_id = unique_channel_id(RC04_CHANNEL_PREFIX) 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_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID) + 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) - empty_result = node.channel_send(RC04_CHANNEL_ID, create_message_bindings(payload="")) + empty_result = node.channel_send(channel_id, create_message_bindings(payload="")) assert empty_result.is_err(), f"channel_send with an empty payload must fail, got Ok({empty_result.ok_value!r})" assert "empty payload" in empty_result.err(), f"unexpected error message: {empty_result.err()!r}" - send_result = node.channel_send(RC04_CHANNEL_ID, create_message_bindings()) + send_result = node.channel_send(channel_id, create_message_bindings()) assert send_result.is_ok(), f"channel_send with a non-empty payload 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: @@ -140,6 +145,7 @@ class TestChannelLifecycle: with "unknown channel: ". No events are expected. """ node_config.update({"mode": "Core"}) + channel_id = unique_channel_id(RC04_CLOSED_CHANNEL_PREFIX) collector = EventCollector() create_node_result = WrapperManager.create_and_start(config=node_config, event_cb=collector.event_callback) @@ -147,17 +153,17 @@ class TestChannelLifecycle: node = create_node_result.ok_value try: - create_result = node.channel_create(RC04_CLOSED_CHANNEL_ID, CONTENT_TOPIC, SENDER_ID) + 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) - close_result = node.channel_close(RC04_CLOSED_CHANNEL_ID) + close_result = node.channel_close(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()) + send_result = node.channel_send(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 f"unknown channel: {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: diff --git a/tests/wrappers_tests/test_channel_persistence_repro.py b/tests/wrappers_tests/test_channel_persistence_repro.py new file mode 100644 index 000000000..5b89a67be --- /dev/null +++ b/tests/wrappers_tests/test_channel_persistence_repro.py @@ -0,0 +1,93 @@ +"""Repro harness for the SDS persistence crash. Skipped — out of scope. + +Issue #191 lists "SDS state survives node restart" as Nim-side, not E2E. Kept +only to reproduce the teardown segfault: drop the skip marker, then drive it +with log/sds_crash_report/sds_wal_repro.sh (seed files are there too). + +The FIXED channelId is deliberate — it accumulates persisted state instead of +avoiding it like unique_channel_id(). Enough state segfaults the node on stop +in Persistency.reset, killing the pytest process, so never run it unattended. +""" + +import pytest + +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 +from tests.wrappers_tests.test_channel_delivery import channel_payloads, wait_for_channel_received, wait_for_channel_received_count + +REPRO_CHANNEL_ID = "repro-fixed-channel" +REPRO_CONTENT_TOPIC = "/test/1/repro-fixed-channel/proto" +SENDER_A = "repro-sender-a" +SENDER_B = "repro-sender-b" + +MESH_SETTLE_S = 10 +CAUSAL_SETTLE_S = 5 +DELIVERY_TIMEOUT_S = 50.0 + + +@pytest.mark.skip( + reason="out of scope for the channel-layer E2E round (issue #191: SDS state across restarts is Nim-side); crashes the runner on purpose" +) +class TestChannelPersistenceRepro: + def test_fixed_channel_id_exchange(self, node_config): + """The RC08 exchange on a fixed channelId. + + Exit code is the driver's signal: 0 delivered, 1 stale causal history, + 139 crash in the library. + """ + m1, m2, m3 = "repro from A", "repro reply from B", "repro follow-up 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(REPRO_CONTENT_TOPIC) + assert subscribe_result.is_ok(), f"receiver subscribe_content_topic failed: {subscribe_result.err()}" + + receiver_create = receiver.channel_create(REPRO_CHANNEL_ID, REPRO_CONTENT_TOPIC, SENDER_B) + assert receiver_create.is_ok(), f"receiver channel_create failed: {receiver_create.err()}" + + with ChannelSenderProcess( + sender_config, + content_topic=REPRO_CONTENT_TOPIC, + channel_id=REPRO_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, REPRO_CHANNEL_ID, DELIVERY_TIMEOUT_S) + assert first is not None, f"B never saw A's first message. Collected events: {receiver_collector.snapshot()}" + + delay(CAUSAL_SETTLE_S) + reply_result = receiver.channel_send(REPRO_CHANNEL_ID, create_message_bindings(payload=to_base64(m2))) + assert reply_result.is_ok(), f"receiver channel_send failed: {reply_result.err()}" + + delivered_to_a = sender.wait_for_received(1, DELIVERY_TIMEOUT_S) + assert channel_payloads(delivered_to_a) == [m2.encode()], f"A must deliver B's reply; got {channel_payloads(delivered_to_a)!r}" + + delay(CAUSAL_SETTLE_S) + sender.send(to_base64(m3)) + + delivered_to_b = wait_for_channel_received_count(receiver_collector, REPRO_CHANNEL_ID, 2, DELIVERY_TIMEOUT_S) + assert channel_payloads(delivered_to_b) == [ + m1.encode(), + m3.encode(), + ], f"B must deliver A's messages in causal order; got {channel_payloads(delivered_to_b)!r}"