Merge branch 'main' into py-multiaddr

Author: seetadev

examples/identify/identify.py

@@ -72,13 +72,46 @@ async def run(port: int, destination: str, use_varint_format: bool = True) -> No
             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(
                 f"First host listening (using {format_name} format). "
                 f"Run this from another console:\n\n"
-                f"identify-demo "
-                f"-d {client_addr}\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:
@@ -93,25 +126,99 @@ async def run(port: int, destination: str, use_varint_format: bool = True) -> No
             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...")
 
-                # Read the complete response (could be either format)
-                # Read a larger chunk to get all the data before stream closes
-                response = await stream.read(8192)  # Read enough data in one go
+                # 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()
 
                 # Parse the response using the robust protocol-level function
                 # This handles both old and new formats automatically
-                identify_msg = parse_identify_response(response)
+                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
 

libp2p/pubsub/pubsub.py

@@ -102,13 +102,17 @@ class TopicValidator(NamedTuple):
     is_async: bool
 
 
+MAX_CONCURRENT_VALIDATORS = 10
+
+
 class Pubsub(Service, IPubsub):
     host: IHost
 
     router: IPubsubRouter
 
     peer_receive_channel: trio.MemoryReceiveChannel[ID]
     dead_peer_receive_channel: trio.MemoryReceiveChannel[ID]
+    _validator_semaphore: trio.Semaphore
 
     seen_messages: LastSeenCache
 
@@ -143,6 +147,7 @@ def __init__(
         msg_id_constructor: Callable[
             [rpc_pb2.Message], bytes
         ] = get_peer_and_seqno_msg_id,
+        max_concurrent_validator_count: int = MAX_CONCURRENT_VALIDATORS,
     ) -> None:
         """
         Construct a new Pubsub object, which is responsible for handling all
@@ -168,6 +173,7 @@ def __init__(
         # Therefore, we can only close from the receive side.
         self.peer_receive_channel = peer_receive
         self.dead_peer_receive_channel = dead_peer_receive
+        self._validator_semaphore = trio.Semaphore(max_concurrent_validator_count)
         # Register a notifee
         self.host.get_network().register_notifee(
             PubsubNotifee(peer_send, dead_peer_send)
@@ -657,7 +663,11 @@ async def publish(self, topic_id: str | list[str], data: bytes) -> None:
 
         logger.debug("successfully published message %s", msg)
 
-    async def validate_msg(self, msg_forwarder: ID, msg: rpc_pb2.Message) -> None:
+    async def validate_msg(
+        self,
+        msg_forwarder: ID,
+        msg: rpc_pb2.Message,
+    ) -> None:
         """
         Validate the received message.
 
@@ -680,23 +690,34 @@ async def validate_msg(self, msg_forwarder: ID, msg: rpc_pb2.Message) -> None:
             if not validator(msg_forwarder, msg):
                 raise ValidationError(f"Validation failed for msg={msg}")
 
-        # TODO: Implement throttle on async validators
-
         if len(async_topic_validators) > 0:
             # Appends to lists are thread safe in CPython
-            results = []
-
-            async def run_async_validator(func: AsyncValidatorFn) -> None:
-                result = await func(msg_forwarder, msg)
-                results.append(result)
+            results: list[bool] = []
 
             async with trio.open_nursery() as nursery:
                 for async_validator in async_topic_validators:
-                    nursery.start_soon(run_async_validator, async_validator)
+                    nursery.start_soon(
+                        self._run_async_validator,
+                        async_validator,
+                        msg_forwarder,
+                        msg,
+                        results,
+                    )
 
             if not all(results):
                 raise ValidationError(f"Validation failed for msg={msg}")
 
+    async def _run_async_validator(
+        self,
+        func: AsyncValidatorFn,
+        msg_forwarder: ID,
+        msg: rpc_pb2.Message,
+        results: list[bool],
+    ) -> None:
+        async with self._validator_semaphore:
+            result = await func(msg_forwarder, msg)
+            results.append(result)
+
     async def push_msg(self, msg_forwarder: ID, msg: rpc_pb2.Message) -> None:
         """
         Push a pubsub message to others.