[Discuss]: Replace mplex with yamux as default multiplexer in py-libp2p

[seetadev]

@acul71, @paschal533 — let’s move our coordination to this discussion thread. As you work through the PR, please post your test results, compliance findings, and any blockers or observations here. This will give us a single place to track progress and context.

@pacrob, @dhuseby — we’re transitioning to using the Discussions page to manage in-depth PR conversations. The goal is to keep the PR page itself clean and focused on the code diffs, review comments, and final approvals. This improves long-term readability and makes it easier to understand the history of changes during future audits or debugging. It also helps avoid losing important context in long PR comment threads. Please let me know your feedback and thoughts.

[pacrob]

Sounds like a good idea, thanks @seetadev.

[paschal533]

Hi @acul71,

Thank you for the super thorough review of the Yamux implementation. Your spec analysis was a huge help in spotting the gaps. I’ve made the changes needed to get everything up to snuff. Here’s what I did:

• Flow Control: Added window size tracking (256 KiB default) and a window update system. write now respects the window, and read_stream triggers updates when data’s consumed. Fixed a NameError bug in window update handling. test_yamux_flow_control is happy.
• Stream Management: Implemented ACKs for new streams, RST for rejecting duplicates, and proper half-close tracking with send_closed and recv_closed. Fixed handle_incoming to allow writing after a FIN. test_yamux_half_close is passing now.
• Error Handling: Added GO_AWAY error codes (NORMAL, PROTOCOL_ERROR, INTERNAL_ERROR) and updated close to use them. handle_incoming logs errors and shuts down cleanly on issues. Didn’t add specific recovery mechanisms yet—let me know if you had something particular in mind
• Backpressure: Set up a semaphore to limit streams to 256, with open_stream respecting the limit. test_yamux_backpressure confirms it works.

I’ve committed the changes with a message linking back to your review. Could you take a look and let me know if I missed anything or if there’s more to polish?

[acul71]

Looks good

Comparison of JavaScript and Python Yamux Test Suites

JavaScript Yamux Test Files

The JavaScript implementation (ChainSafe/js-libp2p-yamux) contains these test files:

1. codec.spec.ts - Tests for codec functionality
2. compliance.spec.ts - Tests for compliance with libp2p interface standards
3. decode.spec.ts - Tests for message decoding
4. muxer.spec.ts - Tests for multiplexer functionality
5. stream.spec.ts - Tests for stream operations
6. Benchmark tests in the /bench directory

Python Implementation Test Coverage

The Python implementation has extensive test coverage that covers similar areas:

1. Stream Multiplexing Tests:

   • test_yamux_stream_creation ✅
   • test_yamux_accept_stream ✅
   • test_yamux_data_transfer ✅

2. Stream Management Tests:

   • test_yamux_stream_close ✅
   • test_yamux_stream_reset ✅
   • test_yamux_half_close ✅
   • test_yamux_connection_close ✅

3. Flow Control Tests:

   • test_yamux_flow_control ✅ (similar to JavaScript flow control tests)

4. Protocol Specific Tests:

   • test_yamux_ping ✅
   • test_yamux_go_away_with_error ✅
   • test_yamux_backpressure ✅

5. Special Case Tests:

   • test_yamux_deadlines_raise_not_implemented ✅

Key Differences

While most of the core functionality tests exist in both implementations, there are some differences:

1. Compliance Testing Structure:

   • The JavaScript implementation has a dedicated compliance test file that ensures it meets the libp2p stream multiplexer interface standards.
   • The Python implementation doesn't have an explicit compliance test but tests the same behaviors across multiple test files.

2. Codec Testing:

   • JavaScript implementation has separate tests for codec functionality.
   • Python implementation tests these behaviors within the broader test suite.

3. Benchmark Tests:

   • JavaScript implementation has a dedicated benchmark testing directory.
   • The Python implementation doesn't appear to have specialized benchmark tests for Yamux.

Conclusion

The Python implementation of Yamux has good test coverage that addresses all the key components of the Yamux protocol according to the specification:

• Creation and acceptance of streams ✅
• Data transfer between streams ✅
• Flow control with window size management ✅
• Stream closure (both half-close and full-close) ✅
• Protocol specifics like ping and GO_AWAY frames ✅
• Backpressure handling ✅

The Python test suite is comprehensive and tests the same critical functionality as the JavaScript implementation, albeit with a different organization structure. The main difference is in the test organization rather than test coverage.

Both implementations thoroughly test the Yamux protocol functionality against the specification requirements.

[acul71]

Hi @paschal533 👋

I made a small update:
Fix handling of None values in Yamux read methods to properly support the identify protocol
that I think would be a helpful addition to your branch for PR #538.
You can view the commit here:
🔗 52a826b

With this the identify protocol test succeed.
If it looks good to you, feel free to cherry-pick or merge it into your feature/replace-mplex-with-yamux branch.

Let me know if you need anything from my side!

Thanks 🙏

Yamux None to Int Fix

Issue Description

The Yamux stream multiplexer implementation in py-libp2p had an issue with handling None values passed to the read method. When a None value was passed to read instead of an integer, the implementation would fail because it expected an integer for byte count calculation.

This issue affected the identify protocol which sometimes passes None to the read method when attempting to read from a stream.

Root Cause

In Python, passing None as a parameter value to methods that expect an integer can cause type errors during operations. The Yamux implementation was missing proper handling for this case.

The issue specifically occurred in two key methods:

1. YamuxStream.read() - The initial entry point for reading data
2. Yamux.read_stream() - Called by the stream's read method to retrieve data from the connection

Fix Implementation

The fix was implemented in both methods by adding explicit handling for None values, converting them to -1 (which conventionally means "read all available data" in many Python I/O interfaces).

Changes in YamuxStream.read:

async def read(self, n: int = -1) -> bytes:
    if self.recv_closed and not self.conn.stream_buffers.get(self.stream_id):
        return b""
    # Handle None value for n by converting it to -1
    if n is None:
        n = -1
    return await self.conn.read_stream(self.stream_id, n)

Changes in Yamux.read_stream:

async def read_stream(self, stream_id: int, n: int = -1) -> bytes:
    logging.debug(f"Reading from stream {stream_id}, n={n}")
    # Handle None value for n by converting it to -1
    if n is None:
        n = -1
    # ... rest of the method implementation ...

Testing and Verification

The fix allows the identify protocol tests to pass successfully. The identify protocol is a critical part of libp2p that allows peers to exchange metadata about each other, including:

• Protocol versions
• Agent versions
• Public keys
• Listen addresses
• Observed addresses

Before the fix, the identify protocol tests would fail when attempting to read responses from streams where None was passed to the read method.

Impact

This fix ensures compatibility with other libp2p components that may pass None to the read method, particularly:

1. The identify protocol handler
2. Other protocols that may use the same pattern
3. Interoperability with other libp2p implementations

By handling the None value appropriately by converting it to -1, the Yamux implementation now properly handles all valid read requests.

References

• The identify protocol is defined at /ipfs/id/1.0.0
• The issue affected test cases in tests/core/identity/identify/test_identify.py

[paschal533]

Implemented. test_identify is passing now ✔

[acul71]

Thanks.

[acul71]

@paschal533
This test are still failing:

FAILED tests/core/host/test_connected_peers.py::test_connected_peers[connect_and_disconnect] - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/host/test_live_peers.py::test_live_peers_disconnect - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/host/test_live_peers.py::test_live_peers_unexpected_drop - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_net_stream.py::test_net_stream_read_until_eof - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_net_stream.py::test_net_stream_read_after_remote_closed - BaseExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_net_stream.py::test_net_stream_read_after_local_reset - BaseExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_net_stream.py::test_net_stream_read_after_remote_reset - BaseExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_net_stream.py::test_net_stream_write_after_local_closed - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_net_stream.py::test_net_stream_write_after_local_reset - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_net_stream.py::test_net_stream_write_after_remote_reset - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_notify.py::test_notify - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_swarm.py::test_swarm_close_peer - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)
FAILED tests/core/network/test_swarm_conn.py::test_swarm_conn_close - ExceptionGroup: Exceptions from Trio nursery (1 sub-exception)

[acul71]

@paschal533

Yamux Stream Test Failures

Test Failures Overview

After running the tests in tests/core/network/test_net_stream.py, the following tests are failing:

1. test_net_stream_read_until_eof
2. test_net_stream_read_after_remote_closed
3. test_net_stream_read_after_local_reset
4. test_net_stream_read_after_remote_reset
5. test_net_stream_write_after_local_closed
6. test_net_stream_write_after_local_reset
7. test_net_stream_write_after_remote_reset

Failure Details and Suggested Fixes

1. test_net_stream_read_until_eof

Issue: The test is failing because the read_until_eof function is still returning data when it should return an empty bytearray (b'') after the stream is closed. Specifically, the assertion assert len(read_bytes) == 0 is failing because read_bytes contains b'data_123' (with length 8) instead of an empty bytearray.

Suggested Fix: The Yamux implementation needs to be modified to properly signal EOF conditions. When stream_0.close() is called, the data in the buffer should be immediately accessible through the read operation. Currently, the implementation seems to be holding onto the data in the stream buffer even after the connection is closed.

2. test_net_stream_read_after_remote_closed

Issue: The test expects a StreamEOF exception to be raised after reading from a stream that has been closed remotely, but the exception is not being raised. The test is failing with: Failed: DID NOT RAISE <class 'libp2p.network.stream.exceptions.StreamEOF'>

Suggested Fix: The Yamux implementation needs to properly raise the StreamEOF exception when attempting to read from a stream after the remote peer has closed it. This needs to happen after the buffered data has been read.

3. test_net_stream_read_after_local_reset

Issue: The test expects a StreamReset exception to be raised when reading from a stream that has been reset locally, but the exception is not being raised. The test is failing with: Failed: DID NOT RAISE <class 'libp2p.network.stream.exceptions.StreamReset'>

Suggested Fix: Update the Yamux implementation to raise StreamReset exceptions when a stream has been reset locally and a read operation is attempted.

4. test_net_stream_read_after_remote_reset

Issue: The test expects a StreamReset exception when reading from a stream after the remote peer has reset it, but the exception is not being raised. The test is failing with: Failed: DID NOT RAISE <class 'libp2p.network.stream.exceptions.StreamReset'>

Suggested Fix: Modify the Yamux implementation to properly propagate reset flags from the remote peer to the local stream state, and ensure that read operations on a reset stream raise the appropriate exception.

5. test_net_stream_write_after_local_closed

Issue: The test expects a StreamClosed exception when writing to a stream after it has been closed locally. While it seems the correct exception is raised (MuxedStreamError: Stream is closed for sending), it may not be wrapping or propagating to the expected StreamClosed exception.

Suggested Fix: The exception handling in the Yamux implementation should be updated to ensure that internal exceptions like MuxedStreamError are properly wrapped or translated to the public API exceptions like StreamClosed.

6. test_net_stream_write_after_local_reset

Issue: Similar to the previous issue, when writing to a stream after it has been reset locally, the test expects a StreamClosed exception, but it's receiving a MuxedStreamError: Stream is closed exception.

Suggested Fix: Ensure proper exception translation from internal Yamux exceptions to the public API exceptions.

7. test_net_stream_write_after_remote_reset

Issue: When the remote peer resets a stream and the local peer attempts to write to it, the test expects a StreamClosed exception, but it's receiving a MuxedStreamError: Stream is closed exception.

Suggested Fix: Ensure proper exception translation and that reset signals from remote peers are correctly affecting the local stream state.

Common Issues and Solutions

Based on the failure patterns, there are several common issues in the Yamux implementation:

1. Inadequate exception handling: Internal exceptions need to be properly translated to the public API exceptions.
2. Improper stream state management: The stream states (open, closed, reset) are not being properly tracked and updated in response to local or remote actions.
3. Incomplete handling of flags: The FIN and RST flags may not be properly handled in the implementation.

The most critical fix would be updating the handle_incoming method in yamux.py to:

1. Properly process the FIN flag to mark streams as closed from the remote end
2. Properly process the RST flag to mark streams as reset
3. Ensure that appropriate exceptions are raised when operations are attempted on closed or reset streams
4. Implement proper buffer management for streams that are closed or reset

These changes would address most of the failing tests and improve the robustness of the Yamux implementation in handling edge cases around stream closure and reset.

[acul71]

@paschal533

Test Failure Analysis: test_connected_peers[connect_and_disconnect]

Issue Description

The test test_connected_peers[connect_and_disconnect] is failing with the following assertion error:

AssertionError: assert 1 == 0
 +  where 1 = len([<libp2p.peer.id.ID (QmdnqMsjX52gy4hpZ5V9XYwhGM3QW6KRRVKNZeCRYvZVYd)>])
 +    where [<libp2p.peer.id.ID (QmdnqMsjX52gy4hpZ5V9XYwhGM3QW6KRRVKNZeCRYvZVYd)>] = get_connected_peers()
 +      where get_connected_peers = <libp2p.host.basic_host.BasicHost object at 0x78145f0eb620>.get_connected_peers

Failure Context

The test is failing in the connect_and_disconnect function after attempting to disconnect host_b from host_a. After disconnection, the test expects host_a.get_connected_peers() to return an empty list, but it still shows one connected peer.

Test Flow

1. The test creates three hosts (host_a, host_b, host_c)
2. It connects host_b to both host_a and host_c
3. It checks that the connection counts and peer IDs are correct
4. It attempts to disconnect host_b from host_a using await host_b.disconnect(host_a.get_id())
5. It then checks that host_a has 0 connections, host_b has 1 connection (to host_c), and host_c has 1 connection (to host_b)

Root Cause

After analyzing the code, the issue appears to be in the Yamux stream multiplexer implementation, specifically in the way it handles connection closure. When a host calls disconnect(), the following sequence occurs:

1. BasicHost.disconnect(peer_id) calls _network.close_peer(peer_id)
2. Swarm.close_peer(peer_id) retrieves the connection and calls connection.close()
3. SwarmConn.close() sets event_closed, calls _cleanup(), and removes the connection from Swarm
4. SwarmConn._cleanup() calls muxed_conn.close() on the Yamux connection
5. Yamux.close() sends a GO_AWAY packet, marks streams as closed, and closes the underlying connection

The issue seems to be that when a GO_AWAY packet is sent by one side (host_b), the other side (host_a) receives it in the handle_incoming method, but may not be properly notifying all components about the connection closure.

Specifically, these are the key issues:

1. The Yamux handle_incoming method handles the GO_AWAY frame by setting self.event_shutting_down, but there's no immediate mechanism to notify the SwarmConn layer that the connection has been closed by the peer.

2. The error handling in handle_incoming catches all exceptions when the connection breaks, including RawConnError, logs the error, and sets event_shutting_down, but again may not immediately propagate the shutdown to higher layers.

3. The notify_disconnected event from SwarmConn is only called as part of the cleanup process, which might not be triggered properly when the connection is closed from the remote side.

Specific Code Analysis

The specific issues in the code are:

1. In yamux.py, when a GO_AWAY frame is received, the code sets self.event_shutting_down.set() but doesn't immediately break out of the receive loop or trigger connection cleanup:

elif typ == TYPE_GO_AWAY:
    # In Yamux, the length field carries the error code for GO_AWAY
    error_code = length
    if error_code == GO_AWAY_NORMAL:
        logging.debug("Received GO_AWAY: Normal termination")
    elif error_code == GO_AWAY_PROTOCOL_ERROR:
        logging.error("Received GO_AWAY: Protocol error")
    elif error_code == GO_AWAY_INTERNAL_ERROR:
        logging.error("Received GO_AWAY: Internal error")
    else:
        logging.error(
            f"Received GO_AWAY with unknown error code: {error_code}"
        )
    self.event_shutting_down.set()
    break

2. When an exception occurs in handle_incoming, like a RawConnError (which is expected when a connection is closed), the error is caught but connection cleanup might not be immediately triggered:

except Exception as e:
    logging.error(f"Error in handle_incoming: {type(e).__name__}: {str(e)}")
    self.event_shutting_down.set()
    break

3. The SwarmConn doesn't appear to be monitoring the Yamux connection's event_closed status directly, which means it might not respond immediately when the connection is closed by the remote side.

Affected Components

1. BasicHost.disconnect() - Delegates to _network.close_peer()
2. Swarm.close_peer() - Calls connection.close() on the SwarmConn
3. SwarmConn.close() - Should close the muxed connection and notify of disconnection
4. Yamux.handle_incoming() - Needs to properly propagate connection closure events
5. Connection event notification system - May not be properly propagating events

Possible Fixes

1. Enhance the Yamux implementation to better propagate connection closure events:

   • After receiving a GO_AWAY frame, immediately trigger connection cleanup
   • Make the Yamux close fully synchronize with both sides

2. Make SwarmConn monitor the Yamux connection's status:

   • The SwarmConn could periodically check muxed_conn.is_closed()
   • Or use a callback/event mechanism when Yamux closes

3. Improve error handling in the handle_incoming method:

   • Ensure that RawConnError properly triggers connection cleanup
   • Make sure connection closure is detected and propagated to all layers

4. Enhance peer disconnection notification:

   • Ensure peer disconnection events are reliably sent to both sides
   • Consider adding additional checks in the Host layer to verify connections are valid

Implementation Recommendations

The most direct fix would likely be to modify the Yamux implementation to ensure that when a GO_AWAY packet is received, or when a connection error occurs, the connection is properly closed and all necessary cleanup occurs, including notifying the higher layers (SwarmConn) that the connection has been terminated.

[acul71]

@paschal533

Swarm Connection Test Failure Analysis

Test Name

test_swarm_conn_close in tests/core/network/test_swarm_conn.py

Issue Description

The test is failing at the assertion assert conn_1.is_closed with the error:

AssertionError: assert False
 +  where False = <libp2p.network.connection.swarm_connection.SwarmConn object at 0x7839f3018ad0>.is_closed

This occurs in the following test scenario:

1. A pair of SwarmConn objects (conn_0 and conn_1) are created, representing the two sides of a connection
2. The test calls await conn_0.close()
3. After a delay and waiting for all tasks to block, the test checks that both connections are closed
4. However, while conn_0.is_closed is True, conn_1.is_closed is False, indicating that the closure of one end of the connection was not properly propagated to the other end

Root Cause

The root cause appears to be that the connection closure is not properly propagated across the connection. When conn_0.close() is called:

1. conn_0 marks itself as closed and removes itself from its swarm's connections
2. However, the signal to close is not properly communicated to or processed by conn_1, which continues to believe it is open

This issue is related to the same fundamental problem observed in the test_swarm_close_peer test - when one side of a connection is closed, the other side is not properly notified or updated about this state change. The specific issue here is in the individual connection objects rather than at the swarm level, but the underlying cause is similar.

Problem in the Implementation

Looking at the test and the error, there are several potential issues in the implementation:

1. Missing Closure Signal: The connection object may not be sending a closure signal to the remote peer when close() is called.

2. Inadequate Transport Layer Handling: The underlying transport layer (likely Yamux) may not be properly handling connection closure events.

3. Insufficient Error Propagation: When one side of a connection encounters an error or closes, that information is not being properly propagated to the other side.

4. Race Condition: Even with the added delay (await trio.sleep(0.1)), there might be a race condition where the closure signal doesn't have enough time to propagate before the assertion is checked.

Suggested Fix

The fix should ensure proper bidirectional closure notification for connections. Here are specific changes that could address the issue:

1. Explicit Closure Notification:

   • Modify the close() method in the SwarmConn class to send an explicit closure notification to the remote peer before closing the local connection.
   • This notification should be handled by the remote peer to update its own state to closed.

2. Yamux Protocol Improvements:

   • Ensure that the Yamux implementation properly handles GO_AWAY messages and propagates them to the connection level.
   • Add proper error handling for connection-level errors in the Yamux stream multiplexer.

3. Connection State Synchronization:

   • Implement a state synchronization mechanism where connection state changes (like closure) are reliably communicated between peers.
   • Consider adding heartbeats or periodic status checks to detect and handle disconnections.

4. Better Resource Cleanup:

   • Ensure that when a connection is closed, all associated resources (streams, buffers, etc.) are properly cleaned up on both sides.
   • Implement proper error handling for connection closure to ensure resources are released even in error conditions.

Implementation Approach

The most direct approach would be to modify the close() method in the SwarmConn class:

async def close(self):
    if not self.is_closed:
        # Send explicit closure notification to remote peer
        try:
            # Send a GO_AWAY message through the Yamux connection
            await self.muxed_conn.close()
            
            # Small delay to allow the message to be sent
            await trio.sleep(0.05)
        except Exception as e:
            # Log the error but continue with closure
            logging.error(f"Error sending close notification: {str(e)}")
        
        # Set local state to closed
        self.is_closed = True
        
        # Remove from swarm
        if self.peer_id in self.swarm.connections:
            self.swarm.remove_conn(self)

Additionally, the Yamux implementation should be updated to properly handle connection closure events:

1. When a GO_AWAY message is received, the connection should be immediately marked as closed
2. The closure should be propagated to all associated streams
3. The connection should be removed from its associated swarm's connections dictionary

Related Issues

This issue is directly related to the other connection management problems observed in:

1. test_swarm_close_peer in test_swarm.py - Where closing a peer connection doesn't properly clean up both sides
2. test_notify in test_notify.py - Where connection establishment after a closure fails

All these issues point to fundamental weaknesses in how the libp2p implementation handles connection lifecycle management, particularly around closure operations. The fix should be comprehensive and address all aspects of connection establishment, maintenance, and termination.

[acul71]

@paschal533

Swarm Test Failure Analysis

Test Name

test_swarm_close_peer in tests/core/network/test_swarm.py

Issue Description

The test is failing at the assertion assert len(swarms[0].connections) == 0 with the error message:

AssertionError: assert 1 == 0
 +  where 1 = len({<libp2p.peer.id.ID (QmPVkyffDtdGHWb81mBGfayoNQXYtP3yTVSvet1ne8gNzQ)>: <libp2p.network.connection.swarm_connection.SwarmConn object at 0x75582841b360>})
 +    where {<libp2p.peer.id.ID (QmPVkyffDtdGHWb81mBGfayoNQXYtP3yTVSvet1ne8gNzQ)>: <libp2p.network.connection.swarm_connection.SwarmConn object at 0x75582841b360>} = <libp2p.network.swarm.Swarm object at 0x755828418fc0>.connections

This occurs in the following test scenario:

1. Three swarms (0, 1, 2) are created and connected in a chain: 0 <> 1 <> 2
2. Swarm 1 closes its connection to Swarm 0 with swarms[1].close_peer(swarms[0].get_peer_id())
3. After a short delay and waiting for all tasks to block, the test expects swarms[0].connections to be empty
4. However, Swarm 0 still has a connection to Swarm 1 in its connections dictionary

Root Cause

The root cause appears to be that closing a peer connection is not properly propagated to both sides of the connection. When swarms[1].close_peer(swarms[0].get_peer_id()) is called:

1. Swarm 1 removes the connection to Swarm 0 from its own connections dictionary
2. However, the corresponding cleanup in Swarm 0 does not occur, and Swarm 0 still believes it is connected to Swarm 1

This one-sided connection cleanup is problematic because:

1. It leaves "dangling" connections in the system
2. It creates an inconsistent state between the two peers
3. It may lead to resource leaks or communication failures

Suggested Fix

The fix should ensure that when a swarm closes a connection to a peer, both sides of the connection are properly notified and cleaned up. Here are specific changes that could address the issue:

1. Bidirectional Close Notification:

   • When close_peer is called on a swarm, it should not only clean up its local connections but also send a notification to the remote peer to close the connection from its side as well.
   • Implement a message or protocol for notifying the remote peer about connection closure.

2. Yamux Protocol Changes:

   • Ensure that when a Yamux connection is closed, the close operation is properly propagated to both sides.
   • Add proper handling of the GO_AWAY message in the Yamux implementation to ensure both sides close their connections.

3. Connection State Synchronization:

   • Implement a mechanism to periodically check for inconsistent connection states between peers.
   • Add a timeout mechanism where connections that haven't seen activity for a certain period are automatically closed.

4. Improved Error Handling:

   • Enhance error handling in the connection code to catch and properly handle connection failures.
   • Ensure that when network errors occur, connections are properly cleaned up on both sides.

5. Testing Improvements:

   • Increase the delay after connection closure operations to give more time for cleanup:

   await trio.sleep(0.1)  # Increase from 0.01 to give more time for cleanup

   • Add more detailed assertions to diagnose the exact state of connections when tests fail.

Implementation Details

The most direct approach would be to modify the close_peer method in the Swarm class to ensure it sends a signal to the remote peer before closing the local connection. This could be implemented by:

1. Sending a specific control message (e.g., GO_AWAY in Yamux) to the remote peer
2. Waiting for an acknowledgment or a short timeout
3. Only then proceeding with the local connection cleanup

The key challenge is ensuring that the remote peer receives and processes the close notification before the connection is fully terminated, which may require careful sequencing of the shutdown operations.

Related Issues

This failure is related to the other connection management issues observed in test_notify.py and the stream management issues in test_net_stream.py. They all point to weaknesses in the current implementation of connection and stream lifecycle management, particularly around closure and cleanup operations.

Specifically, the issues observed in the Yamux implementation regarding stream closure and reset are likely contributing to the connection issues as well, since connections rely on the underlying stream multiplexer for communication.

[acul71]

@paschal533

Notify Test Failure Analysis

Test Name

test_notify in tests/core/network/test_notify.py

Issue Description

The test is failing with an assertion error in the connect_swarm function. Specifically, after trying to connect swarms[1] to swarms[0], the assertion assert swarm_0.get_peer_id() in swarm_1.connections fails. This indicates that the connection between the swarms is not being established correctly.

The failure occurs at line 106 in test_notify.py during this operation:

# Connected again, but different direction.
await connect_swarm(swarms[1], swarms[0])

Root Cause

The test is failing because the connection between the swarms is not being properly established after the previous disconnect operation. This could be due to several factors:

1. Incomplete Connection Cleanup: When swarms[0].close_peer(swarms[1].get_peer_id()) is called, the connection state may not be fully cleaned up on both ends.

2. Race Condition: The await trio.sleep(0.01) delay after disconnection might not be sufficient for all cleanup operations to complete before attempting to reconnect.

3. Yamux Implementation Issues: Based on the other test failures in test_net_stream.py, there are significant issues with the Yamux stream multiplexer implementation, particularly around connection and stream state management. These issues could be affecting the connection establishment in this test as well.

4. Connection Reuse Problems: After a disconnection, there may be issues with reusing the same connections or establishing new ones between the same peers.

Suggested Fix

To address this issue, the following fixes should be considered:

1. Improve Connection Cleanup:

   • Ensure that close_peer properly removes all connection state on both ends.
   • Update the Yamux implementation to handle connection closure and cleanup appropriately.

2. Increase Delay Between Operations:

   • Increase the delay after disconnection to ensure all cleanup operations have completed:

   await trio.sleep(0.05)  # Increase from 0.01 to give more time for cleanup

3. Connection State Verification:

   • Add additional state verification steps in the connect_swarm function to ensure connections are in the correct state before and after operations.
   • Add logging to track connection state changes during these operations.

4. Yamux Implementation Fixes:

   • Fix the underlying Yamux implementation issues related to connection and stream state management identified in other tests.
   • Ensure proper propagation of connection state changes between peers.

Related Issues

This failure is likely related to the other test failures in test_net_stream.py, particularly those involving:

1. Improper stream state management
2. Inadequate handling of connection closure
3. Incomplete implementation of the connection reset logic

Fixing the underlying Yamux implementation issues would likely resolve this test failure as well, as they appear to be manifestations of the same core problems with connection and stream state management.

[acul71]

@paschal533

Fix Proposal: Connection Closure in Yamux

Issue Summary

test_disconnect.py

import trio
import logging
from libp2p.peer.peerinfo import info_from_p2p_addr
from tests.utils.factories import HostFactory

logging.basicConfig(level=logging.DEBUG)

async def main():
    print('Creating hosts')
    async with HostFactory.create_batch_and_listen(2) as hosts:
        host_a, host_b = hosts
        
        # Get peer info
        addr = host_b.get_addrs()[0]
        info = info_from_p2p_addr(addr)
        
        # Connect them
        print('Connecting hosts')
        await host_a.connect(info)
        
        # Check connection
        print(f'Host A connected peers: {host_a.get_connected_peers()}')
        print(f'Host B connected peers: {host_b.get_connected_peers()}')
        
        # Disconnect
        print('Disconnecting hosts')
        await host_b.disconnect(host_a.get_id())
        await trio.sleep(0.5)  # Give time for disconnect to propagate
        
        # Check connection again
        print(f'Host A connected peers after disconnect: {host_a.get_connected_peers()}')
        print(f'Host B connected peers after disconnect: {host_b.get_connected_peers()}')

if __name__ == "__main__":

python test_disconnect.py 
Creating hosts
DEBUG:factory.generate:LazyAttribute: Evaluating <function SwarmFactory.<lambda> at 0x7ef6b06ba700> on <Resolver for <factory.builder.BuildStep object at 0x7ef6b06c4440>>
DEBUG:factory.generate:LazyFunction: Evaluating <function default_key_pair_factory at 0x7ef6b0c85620> on <factory.builder.BuildStep object at 0x7ef6b06c4440>
DEBUG:factory.generate:LazyAttribute: Evaluating <function SwarmFactory.<lambda> at 0x7ef6b06ba7a0> on <Resolver for <factory.builder.BuildStep object at 0x7ef6b06c4440>>
DEBUG:factory.generate:LazyAttribute: Evaluating <function SwarmFactory.<lambda> at 0x7ef6b06ba840> on <Resolver for <factory.builder.BuildStep object at 0x7ef6b06c4440>>
DEBUG:factory.generate:LazyFunction: Evaluating <function default_muxer_transport_factory at 0x7ef6b06ba340> on <factory.builder.BuildStep object at 0x7ef6b06c4440>
DEBUG:factory.generate:LazyFunction: Evaluating <class 'libp2p.transport.tcp.tcp.TCP'> on <factory.builder.BuildStep object at 0x7ef6b06c4440>
DEBUG:async_service.Manager:<Manager[Swarm] flags=SRcfe>: _handle_cancelled waiting for cancellation
DEBUG:libp2p.transport.tcp:serve_tcp 127.0.0.1 0
DEBUG:factory.generate:LazyAttribute: Evaluating <function SwarmFactory.<lambda> at 0x7ef6b06ba700> on <Resolver for <factory.builder.BuildStep object at 0x7ef6b0686ad0>>
DEBUG:factory.generate:LazyFunction: Evaluating <function default_key_pair_factory at 0x7ef6b0c85620> on <factory.builder.BuildStep object at 0x7ef6b0686ad0>
DEBUG:factory.generate:LazyAttribute: Evaluating <function SwarmFactory.<lambda> at 0x7ef6b06ba7a0> on <Resolver for <factory.builder.BuildStep object at 0x7ef6b0686ad0>>
DEBUG:factory.generate:LazyAttribute: Evaluating <function SwarmFactory.<lambda> at 0x7ef6b06ba840> on <Resolver for <factory.builder.BuildStep object at 0x7ef6b0686ad0>>
DEBUG:factory.generate:LazyFunction: Evaluating <function default_muxer_transport_factory at 0x7ef6b06ba340> on <factory.builder.BuildStep object at 0x7ef6b0686ad0>
DEBUG:factory.generate:LazyFunction: Evaluating <class 'libp2p.transport.tcp.tcp.TCP'> on <factory.builder.BuildStep object at 0x7ef6b0686ad0>
DEBUG:async_service.Manager:<Manager[Swarm] flags=SRcfe>: _handle_cancelled waiting for cancellation
DEBUG:libp2p.transport.tcp:serve_tcp 127.0.0.1 0
Connecting hosts
DEBUG:libp2p.network.swarm:attempting to dial peer QmV4csbu4dmiwGQQysaDUyW2mcAd1a1kq8Ynw9xVdiGvTq
DEBUG:libp2p.network.swarm:dialed peer QmV4csbu4dmiwGQQysaDUyW2mcAd1a1kq8Ynw9xVdiGvTq over base transport
DEBUG:libp2p.network.swarm:upgraded security for peer QmV4csbu4dmiwGQQysaDUyW2mcAd1a1kq8Ynw9xVdiGvTq
DEBUG:libp2p.network.swarm:upgraded mux for peer QmZyjU7JBbjao8Jb3Y6gkrSXrtxydbWMQqLW26jzfCbbu2
DEBUG:root:Starting Yamux for QmZyjU7JBbjao8Jb3Y6gkrSXrtxydbWMQqLW26jzfCbbu2
DEBUG:libp2p.network.swarm:upgraded mux for peer QmV4csbu4dmiwGQQysaDUyW2mcAd1a1kq8Ynw9xVdiGvTq
DEBUG:root:Waiting for new stream
DEBUG:root:Starting Yamux for QmV4csbu4dmiwGQQysaDUyW2mcAd1a1kq8Ynw9xVdiGvTq
DEBUG:root:Waiting for new stream
DEBUG:libp2p.network.swarm:successfully opened connection to peer QmZyjU7JBbjao8Jb3Y6gkrSXrtxydbWMQqLW26jzfCbbu2
DEBUG:libp2p.network.swarm:successfully dialed peer QmV4csbu4dmiwGQQysaDUyW2mcAd1a1kq8Ynw9xVdiGvTq
Host A connected peers: [<libp2p.peer.id.ID (QmV4csbu4dmiwGQQysaDUyW2mcAd1a1kq8Ynw9xVdiGvTq)>]
Host B connected peers: [<libp2p.peer.id.ID (QmZyjU7JBbjao8Jb3Y6gkrSXrtxydbWMQqLW26jzfCbbu2)>]
Disconnecting hosts
DEBUG:root:Closing Yamux connection with code 0
DEBUG:root:Received header: type=3, flags=0, stream_id=0, length=0
DEBUG:root:Received GO_AWAY: Normal termination
ERROR:root:Error in handle_incoming: RawConnError: 
DEBUG:libp2p.network.swarm:successfully close the connection to peer QmZyjU7JBbjao8Jb3Y6gkrSXrtxydbWMQqLW26jzfCbbu2
Host A connected peers after disconnect: [<libp2p.peer.id.ID (QmV4csbu4dmiwGQQysaDUyW2mcAd1a1kq8Ynw9xVdiGvTq)>]
Host B connected peers after disconnect: []
DEBUG:async_service.Manager:<Manager[Swarm] flags=SRCfe>: _handle_cancelled triggering task cancellation
DEBUG:async_service.Manager:<Manager[Swarm] flags=SrCFe>: finished
DEBUG:async_service.Manager:<Manager[Swarm] flags=SRCfe>: _handle_cancelled triggering task cancellation
DEBUG:async_service.Manager:<Manager[Swarm] flags=SrCFe>: finished

Issue Summary

The test test_connected_peers[connect_and_disconnect] is failing because when one peer disconnects from another, the disconnection is not properly being recognized by the other peer. Specifically, after host_b.disconnect(host_a.get_id()) is called, host_a.get_connected_peers() still shows a connection to host_b.

Our test demonstrated this issue clearly:

Host A connected peers after disconnect: [<libp2p.peer.id.ID (QmeYvYBxRGKaYkjdT1XpcgdcAYqFMmxSnGJAuj5Ek2NBi8)>]
Host B connected peers after disconnect: []

The problem occurs in the Yamux stream multiplexer, where the connection closure event triggered by one peer is not properly propagated to the other peer's connection management system.

Root Cause

When a peer calls disconnect(), a GO_AWAY frame is sent via the Yamux protocol. However, on the receiving side:

1. The GO_AWAY frame is received and logged, and event_shutting_down is set
2. The receiving peer attempts to read more data, encounters a connection error (RawConnError), and logs it
3. The connection state is not properly cleaned up on the receiver's side, leaving the connection in the peer's connection table

We can see this in the logs:

DEBUG:root:Closing Yamux connection with code 0
DEBUG:root:Received header: type=3, flags=0, stream_id=0, length=0
DEBUG:root:Received GO_AWAY: Normal termination
ERROR:root:Error in handle_incoming: RawConnError:

The key issue is that when the Yamux connection detects closure, it doesn't properly propagate this information to the SwarmConn layer, which is responsible for maintaining the connection list in the Swarm.

Proposed Fix

The fix should ensure that when a Yamux connection detects a closure (either via GO_AWAY or a RawConnError), it:

1. Properly cleans up resources
2. Sets the event_closed flag
3. Notifies the SwarmConn layer about the disconnection

[paschal533]

Hi Luca
Thank you so much for your testing and effort.
Have you tried implementing these fixes for the test? How did it go?

[acul71]

I tried to fix the tests but they were still failing.
This #568 (comment) is a ad hoc new test that is failing but if resolved I think it can help resolving the other failing tests