Merge branch 'main' of https://github.com/Nkovaturient/py-libp2p into feat/add-webrtc-transport

Author: Nkovaturient

docs/release_notes.rst

@@ -3,6 +3,65 @@ Release Notes
 
 .. towncrier release notes start
 
+py-libp2p v0.2.9 (2025-07-09)
+-----------------------------
+
+Breaking Changes
+~~~~~~~~~~~~~~~~
+
+- Reordered the arguments to ``upgrade_security`` to place ``is_initiator`` before ``peer_id``, and made ``peer_id`` optional.
+  This allows the method to reflect the fact that peer identity is not required for inbound connections. (`#681 <https://github.com/libp2p/py-libp2p/issues/681>`__)
+
+
+Bugfixes
+~~~~~~~~
+
+- Add timeout wrappers in:
+  1. ``multiselect.py``: ``negotiate`` function
+  2. ``multiselect_client.py``: ``select_one_of`` , ``query_multistream_command`` functions
+  to prevent indefinite hangs when a remote peer does not respond. (`#696 <https://github.com/libp2p/py-libp2p/issues/696>`__)
+- Align stream creation logic with yamux specification (`#701 <https://github.com/libp2p/py-libp2p/issues/701>`__)
+- Fixed an issue in ``Pubsub`` where async validators were not handled reliably under concurrency. Now uses a safe aggregator list for consistent behavior. (`#702 <https://github.com/libp2p/py-libp2p/issues/702>`__)
+
+
+Features
+~~~~~~~~
+
+- Added support for ``Kademlia DHT`` in py-libp2p. (`#579 <https://github.com/libp2p/py-libp2p/issues/579>`__)
+- Limit concurrency in ``push_identify_to_peers`` to prevent resource congestion under high peer counts. (`#621 <https://github.com/libp2p/py-libp2p/issues/621>`__)
+- Store public key and peer ID in peerstore during handshake
+
+  Modified the InsecureTransport class to accept an optional peerstore parameter and updated the handshake process to store the received public key and peer ID in the peerstore when available.
+
+  Added test cases to verify:
+  1. The peerstore remains unchanged when handshake fails due to peer ID mismatch
+  2. The handshake correctly adds a public key to a peer ID that already exists in the peerstore but doesn't have a public key yet (`#631 <https://github.com/libp2p/py-libp2p/issues/631>`__)
+- Fixed several flow-control and concurrency issues in the ``YamuxStream`` class. Previously, stress-testing revealed that transferring data over ``DEFAULT_WINDOW_SIZE`` would break the stream due to inconsistent window update handling and lock management. The fixes include:
+
+  - Removed sending of window updates during writes to maintain correct flow-control.
+  - Added proper timeout handling when releasing and acquiring locks to prevent concurrency errors.
+  - Corrected the ``read`` function to properly handle window updates for both ``read_until_EOF`` and ``read_n_bytes``.
+  - Added event logging at ``send_window_updates`` and ``waiting_for_window_updates`` for better observability. (`#639 <https://github.com/libp2p/py-libp2p/issues/639>`__)
+- Added support for ``Multicast DNS`` in py-libp2p (`#649 <https://github.com/libp2p/py-libp2p/issues/649>`__)
+- Optimized pubsub publishing to send multiple topics in a single message instead of separate messages per topic. (`#685 <https://github.com/libp2p/py-libp2p/issues/685>`__)
+- Optimized pubsub message writing by implementing a write_msg() method that uses pre-allocated buffers and single write operations, improving performance by eliminating separate varint prefix encoding and write operations in FloodSub and GossipSub. (`#687 <https://github.com/libp2p/py-libp2p/issues/687>`__)
+- Added peer exchange and backoff logic as part of Gossipsub v1.1 upgrade (`#690 <https://github.com/libp2p/py-libp2p/issues/690>`__)
+
+
+Internal Changes - for py-libp2p Contributors
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+- Added sparse connect utility function to pubsub test utilities for creating test networks with configurable connectivity. (`#679 <https://github.com/libp2p/py-libp2p/issues/679>`__)
+- Added comprehensive tests for pubsub connection utility functions to verify degree limits are enforced, excess peers are handled correctly, and edge cases (degree=0, negative values, empty lists) are managed gracefully. (`#707 <https://github.com/libp2p/py-libp2p/issues/707>`__)
+- Added extra tests for identify push concurrency cap under high peer load (`#708 <https://github.com/libp2p/py-libp2p/issues/708>`__)
+
+
+Miscellaneous Changes
+~~~~~~~~~~~~~~~~~~~~~
+
+- `#678 <https://github.com/libp2p/py-libp2p/issues/678>`__, `#684 <https://github.com/libp2p/py-libp2p/issues/684>`__
+
+
 py-libp2p v0.2.8 (2025-06-10)
 -----------------------------
 

examples/identify/identify.py

@@ -8,9 +8,10 @@
 from libp2p import (
     new_host,
 )
-from libp2p.identity.identify.identify import ID as IDENTIFY_PROTOCOL_ID
-from libp2p.identity.identify.pb.identify_pb2 import (
-    Identify,
+from libp2p.identity.identify.identify import (
+    ID as IDENTIFY_PROTOCOL_ID,
+    identify_handler_for,
+    parse_identify_response,
 )
 from libp2p.peer.peerinfo import (
     info_from_p2p_addr,
@@ -50,21 +51,67 @@ def print_identify_response(identify_response):
     )
 
 
-async def run(port: int, destination: str) -> None:
+async def run(port: int, destination: str, use_varint_format: bool = True) -> None:
     localhost_ip = "0.0.0.0"
 
     if not destination:
         # Create first host (listener)
         listen_addr = multiaddr.Multiaddr(f"/ip4/{localhost_ip}/tcp/{port}")
         host_a = new_host()
 
+        # Set up identify handler with specified format
+        identify_handler = identify_handler_for(
+            host_a, use_varint_format=use_varint_format
+        )
+        host_a.set_stream_handler(IDENTIFY_PROTOCOL_ID, identify_handler)
+
         async with host_a.run(listen_addrs=[listen_addr]):
+            # Get the actual address and replace 0.0.0.0 with 127.0.0.1 for client
+            # connections
+            server_addr = str(host_a.get_addrs()[0])
+            client_addr = server_addr.replace("/ip4/0.0.0.0/", "/ip4/127.0.0.1/")
+
+            format_name = "length-prefixed" if use_varint_format else "raw protobuf"
+            format_flag = "--raw-format" if not use_varint_format else ""
             print(
-                "First host listening. Run this from another console:\n\n"
-                f"identify-demo "
-                f"-d {host_a.get_addrs()[0]}\n"
+                f"First host listening (using {format_name} format). "
+                f"Run this from another console:\n\n"
+                f"identify-demo {format_flag} -d {client_addr}\n"
             )
             print("Waiting for incoming identify request...")
+
+            # Add a custom handler to show connection events
+            async def custom_identify_handler(stream):
+                peer_id = stream.muxed_conn.peer_id
+                print(f"\n🔗 Received identify request from peer: {peer_id}")
+
+                # Show remote address in multiaddr format
+                try:
+                    from libp2p.identity.identify.identify import (
+                        _remote_address_to_multiaddr,
+                    )
+
+                    remote_address = stream.get_remote_address()
+                    if remote_address:
+                        observed_multiaddr = _remote_address_to_multiaddr(
+                            remote_address
+                        )
+                        # Add the peer ID to create a complete multiaddr
+                        complete_multiaddr = f"{observed_multiaddr}/p2p/{peer_id}"
+                        print(f"   Remote address: {complete_multiaddr}")
+                    else:
+                        print(f"   Remote address: {remote_address}")
+                except Exception:
+                    print(f"   Remote address: {stream.get_remote_address()}")
+
+                # Call the original handler
+                await identify_handler(stream)
+
+                print(f"✅ Successfully processed identify request from {peer_id}")
+
+            # Replace the handler with our custom one
+            host_a.set_stream_handler(IDENTIFY_PROTOCOL_ID, custom_identify_handler)
+
             await trio.sleep_forever()
 
     else:
@@ -79,28 +126,112 @@ async def run(port: int, destination: str) -> None:
             info = info_from_p2p_addr(maddr)
             print(f"Second host connecting to peer: {info.peer_id}")
 
-            await host_b.connect(info)
+            try:
+                await host_b.connect(info)
+            except Exception as e:
+                error_msg = str(e)
+                if "unable to connect" in error_msg or "SwarmException" in error_msg:
+                    print(f"\n❌ Cannot connect to peer: {info.peer_id}")
+                    print(f"   Address: {destination}")
+                    print(f"   Error: {error_msg}")
+                    print(
+                        "\n💡 Make sure the peer is running and the address is correct."
+                    )
+                    return
+                else:
+                    # Re-raise other exceptions
+                    raise
+
             stream = await host_b.new_stream(info.peer_id, (IDENTIFY_PROTOCOL_ID,))
 
             try:
                 print("Starting identify protocol...")
-                response = await stream.read()
+
+                # Read the response properly based on the format
+                if use_varint_format:
+                    # For length-prefixed format, read varint length first
+                    from libp2p.utils.varint import decode_varint_from_bytes
+
+                    # Read varint length prefix
+                    length_bytes = b""
+                    while True:
+                        b = await stream.read(1)
+                        if not b:
+                            raise Exception("Stream closed while reading varint length")
+                        length_bytes += b
+                        if b[0] & 0x80 == 0:
+                            break
+
+                    msg_length = decode_varint_from_bytes(length_bytes)
+                    print(f"Expected message length: {msg_length} bytes")
+
+                    # Read the protobuf message
+                    response = await stream.read(msg_length)
+                    if len(response) != msg_length:
+                        raise Exception(
+                            f"Incomplete message: expected {msg_length} bytes, "
+                            f"got {len(response)}"
+                        )
+
+                    # Combine length prefix and message
+                    full_response = length_bytes + response
+                else:
+                    # For raw format, read all available data
+                    response = await stream.read(8192)
+                    full_response = response
+
                 await stream.close()
-                identify_msg = Identify()
-                identify_msg.ParseFromString(response)
+
+                # Parse the response using the robust protocol-level function
+                # This handles both old and new formats automatically
+                identify_msg = parse_identify_response(full_response)
                 print_identify_response(identify_msg)
+
             except Exception as e:
-                print(f"Identify protocol error: {e}")
+                error_msg = str(e)
+                print(f"Identify protocol error: {error_msg}")
+
+                # Check for specific format mismatch errors
+                if "Error parsing message" in error_msg or "DecodeError" in error_msg:
+                    print("\n" + "=" * 60)
+                    print("FORMAT MISMATCH DETECTED!")
+                    print("=" * 60)
+                    if use_varint_format:
+                        print(
+                            "You are using length-prefixed format (default) but the "
+                            "listener"
+                        )
+                        print("is using raw protobuf format.")
+                        print(
+                            "\nTo fix this, run the dialer with the --raw-format flag:"
+                        )
+                        print(f"identify-demo --raw-format -d {destination}")
+                    else:
+                        print("You are using raw protobuf format but the listener")
+                        print("is using length-prefixed format (default).")
+                        print(
+                            "\nTo fix this, run the dialer without the --raw-format "
+                            "flag:"
+                        )
+                        print(f"identify-demo -d {destination}")
+                    print("=" * 60)
+                else:
+                    import traceback
+
+                    traceback.print_exc()
 
             return
 
 
 def main() -> None:
     description = """
     This program demonstrates the libp2p identify protocol.
-    First run identify-demo -p <PORT>' to start a listener.
+    First run 'identify-demo -p <PORT> [--raw-format]' to start a listener.
     Then run 'identify-demo <ANOTHER_PORT> -d <DESTINATION>'
     where <DESTINATION> is the multiaddress shown by the listener.
+
+    Use --raw-format to send raw protobuf messages (old format) instead of
+    length-prefixed protobuf messages (new format, default).
     """
 
     example_maddr = (
@@ -115,10 +246,22 @@ def main() -> None:
         type=str,
         help=f"destination multiaddr string, e.g. {example_maddr}",
     )
+    parser.add_argument(
+        "--raw-format",
+        action="store_true",
+        help=(
+            "use raw protobuf format (old format) instead of "
+            "length-prefixed (new format)"
+        ),
+    )
     args = parser.parse_args()
 
+    # Determine format: raw format if --raw-format is specified, otherwise
+    # length-prefixed
+    use_varint_format = not args.raw_format
+
     try:
-        trio.run(run, *(args.port, args.destination))
+        trio.run(run, *(args.port, args.destination, use_varint_format))
     except KeyboardInterrupt:
         pass