feat: add interop btw rust and pylibp2p (#2)

Co-authored-by: lla-dane <[email protected]>
Co-authored-by: Manu Sheel Gupta <[email protected]>

Author: 3 people

docs/conf.py

@@ -290,7 +290,7 @@
 ]
 
 # Prevent autodoc from trying to import module from tests.factories
-autodoc_mock_imports = ["tests.factories"]
+autodoc_mock_imports = ["tests.factories", "redis"]
 
 # Documents to append as an appendix to all manuals.
 # texinfo_appendices = []

docs/index.rst

@@ -18,6 +18,7 @@ The Python implementation of the libp2p networking stack
 
     Examples <examples>
     API <libp2p>
+    Interop <interop>
 
 .. toctree::
     :maxdepth: 1

docs/interop.rst

@@ -0,0 +1,29 @@
+interop package
+===============
+
+Submodules
+----------
+
+interop.arch module
+-------------------
+
+.. automodule:: interop.arch
+   :members:
+   :show-inheritance:
+   :undoc-members:
+
+interop.lib module
+------------------
+
+.. automodule:: interop.lib
+   :members:
+   :show-inheritance:
+   :undoc-members:
+
+Module contents
+---------------
+
+.. automodule:: interop
+   :members:
+   :show-inheritance:
+   :undoc-members:

examples/ping/ping.py

@@ -1,4 +1,5 @@
 import argparse
+import logging
 
 import multiaddr
 import trio
@@ -55,8 +56,8 @@ async def send_ping(stream: INetStream) -> None:
 
 
 async def run(port: int, destination: str) -> None:
-    listen_addr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")
-    host = new_host(listen_addrs=[listen_addr])
+    listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}")
+    host = new_host()
 
     async with host.run(listen_addrs=[listen_addr]), trio.open_nursery() as nursery:
         if not destination:
@@ -103,8 +104,20 @@ def main() -> None:
         type=str,
         help=f"destination multiaddr string, e.g. {example_maddr}",
     )
+    parser.add_argument(
+        "-v", "--verbose", action="store_true", help="enable verbose logging"
+    )
+
     args = parser.parse_args()
 
+    if args.verbose:
+        # Enable even more detailed logging
+        logging.getLogger("libp2p").setLevel(logging.DEBUG)
+        logging.getLogger("libp2p.network").setLevel(logging.DEBUG)
+        logging.getLogger("libp2p.transport").setLevel(logging.DEBUG)
+        logging.getLogger("libp2p.security").setLevel(logging.DEBUG)
+        logging.getLogger("libp2p.stream_muxer").setLevel(logging.DEBUG)
+
     try:
         trio.run(run, *(args.port, args.destination))
     except KeyboardInterrupt:

libp2p/stream_muxer/yamux/yamux.py

@@ -33,6 +33,9 @@
 from libp2p.io.exceptions import (
     IncompleteReadError,
 )
+from libp2p.io.utils import (
+    read_exactly,
+)
 from libp2p.network.connection.exceptions import (
     RawConnError,
 )
@@ -127,20 +130,22 @@ async def write(self, data: bytes) -> None:
                         "Timed out waiting for window update after 5 seconds."
                     )
 
-                if self.closed:
-                    raise MuxedStreamError("Stream is closed")
+            if self.closed:
+                raise MuxedStreamError("Stream is closed")
+
 
-                # Calculate how much we can send now
+            # Calculate how much we can send now
+            async with self.window_lock:
                 to_send = min(self.send_window, total_len - sent)
                 chunk = data[sent : sent + to_send]
                 self.send_window -= to_send
 
-                # Send the data
-                header = struct.pack(
-                    YAMUX_HEADER_FORMAT, 0, TYPE_DATA, 0, self.stream_id, len(chunk)
-                )
-                await self.conn.secured_conn.write(header + chunk)
-                sent += to_send
+            # Send the data
+            header = struct.pack(
+                YAMUX_HEADER_FORMAT, 0, TYPE_DATA, 0, self.stream_id, len(chunk)
+            )
+            await self.conn.secured_conn.write(header + chunk)
+            sent += to_send
 
     async def send_window_update(self, increment: int, skip_lock: bool = False) -> None:
         """
@@ -397,7 +402,6 @@ async def close(self, error_code: int = GO_AWAY_NORMAL) -> None:
             else:
                 if self.on_close is not None:
                     await self.on_close()
-        await trio.sleep(0.1)
 
     @property
     def is_closed(self) -> bool:
@@ -537,13 +541,17 @@ async def handle_incoming(self) -> None:
                     self.event_shutting_down.set()
                     await self._cleanup_on_error()
                     break
+
+                # Debug: log raw header bytes
+                logging.debug(f"Raw header bytes: {header.hex()}")
+
                 version, typ, flags, stream_id, length = struct.unpack(
                     YAMUX_HEADER_FORMAT, header
                 )
                 logger.debug(
                     f"Received header for peer {self.peer_id}:"
-                    f"type={typ}, flags={flags}, stream_id={stream_id},"
-                    f"length={length}"
+                    f"version={version}, type={typ}, flags={flags}, "
+                    f"stream_id={stream_id}, length={length}"
                 )
                 if (typ == TYPE_DATA or typ == TYPE_WINDOW_UPDATE) and flags & FLAG_SYN:
                     async with self.streams_lock:
@@ -668,6 +676,85 @@ async def handle_incoming(self) -> None:
                                     f" increment: {increment}"
                                 )
                                 stream.send_window += increment
+                elif typ == TYPE_DATA:
+                    async with self.streams_lock:
+                        if stream_id in self.streams:
+                            # Store data - ensure data is not None before extending
+                            if data is not None and len(data) > 0:
+                                self.stream_buffers[stream_id].extend(data)
+                                if stream_id in self.stream_events:
+                                    self.stream_events[stream_id].set()
+                            # Handle flags
+                            if flags & FLAG_SYN:
+                                logging.debug(
+                                    f"Received late SYN for stream {stream_id} "
+                                    f"for peer {self.peer_id}"
+                                )
+                            if flags & FLAG_ACK:
+                                logging.debug(
+                                    f"Received ACK for stream {stream_id} "
+                                    f"for peer {self.peer_id}"
+                                )
+                            if flags & FLAG_FIN:
+                                logging.debug(
+                                    f"Received FIN for stream {self.peer_id}:"
+                                    f"{stream_id}, marking recv_closed"
+                                )
+                                self.streams[stream_id].recv_closed = True
+                                if self.streams[stream_id].send_closed:
+                                    self.streams[stream_id].closed = True
+                                if stream_id in self.stream_events:
+                                    self.stream_events[stream_id].set()
+                            if flags & FLAG_RST:
+                                logging.debug(
+                                    f"Resetting stream {stream_id} "
+                                    f"for peer {self.peer_id}"
+                                )
+                                self.streams[stream_id].closed = True
+                                self.streams[stream_id].reset_received = True
+                                if stream_id in self.stream_events:
+                                    self.stream_events[stream_id].set()
+                        else:
+                            if flags & FLAG_SYN:
+                                if stream_id not in self.streams:
+                                    stream = YamuxStream(stream_id, self, False)
+                                    self.streams[stream_id] = stream
+                                    # Initialize stream buffer
+                                    buffer = bytearray()
+                                    if data is not None and len(data) > 0:
+                                        buffer.extend(data)
+                                    self.stream_buffers[stream_id] = buffer
+                                    self.stream_events[stream_id] = trio.Event()
+                                    self.stream_events[stream_id].set()
+                                    ack_header = struct.pack(
+                                        YAMUX_HEADER_FORMAT,
+                                        0,
+                                        TYPE_DATA,
+                                        FLAG_ACK,
+                                        stream_id,
+                                        0,
+                                    )
+                                    await self.secured_conn.write(ack_header)
+                                    logging.debug(
+                                        f"Sending stream {stream_id}"
+                                        f"to channel for peer {self.peer_id}"
+                                    )
+                                    await self.new_stream_send_channel.send(stream)
+                                else:
+                                    rst_header = struct.pack(
+                                        YAMUX_HEADER_FORMAT,
+                                        0,
+                                        TYPE_DATA,
+                                        FLAG_RST,
+                                        stream_id,
+                                        0,
+                                    )
+                                    await self.secured_conn.write(rst_header)
+                            else:
+                                logging.warning(
+                                    f"Received data for unknown stream {stream_id} "
+                                    f"from peer {self.peer_id} (length={length})"
+                                )
             except Exception as e:
                 # Special handling for expected IncompleteReadError on stream close
                 if isinstance(e, IncompleteReadError):
@@ -687,7 +774,7 @@ async def handle_incoming(self) -> None:
                     else:
                         logger.error(
                             f"Error in handle_incoming for peer {self.peer_id}: "
-                            + f"{type(e).__name__}: {str(e)}"
+                            f"{type(e).__name__}: {str(e)}"
                         )
                 else:
                     # Handle RawConnError with more nuance
@@ -717,8 +804,6 @@ async def handle_incoming(self) -> None:
                 ):
                     await self._cleanup_on_error()
                     break
-                # For other errors, log and continue
-                await trio.sleep(0.01)
 
     async def _cleanup_on_error(self) -> None:
         # Set shutdown flag first to prevent other operations
@@ -760,6 +845,6 @@ async def _cleanup_on_error(self) -> None:
             except Exception as callback_error:
                 logger.error(f"Error in on_close callback: {callback_error}")
 
-        # Cancel nursery tasks
+        # Cancel nursery tasks if available
         if self._nursery:
             self._nursery.cancel_scope.cancel()

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "trio-typing>=0.0.4",
     "trio>=0.26.0",
     "fastecdsa==2.3.2; sys_platform != 'win32'",
+    "cryptography>=42.0.0; sys_platform == 'win32'",  # Alternative for Windows
     "zeroconf (>=0.147.0,<0.148.0)",
 ]
 classifiers = [
@@ -79,6 +80,7 @@ dev = [
     "factory-boy>=2.12.0,<3.0.0",
     "ruff>=0.11.10",
     "pyrefly (>=0.17.1,<0.18.0)",
+    "pytest-timeout",
 ]
 docs = [
     "sphinx>=6.0.0",
@@ -93,6 +95,7 @@ test = [
     "pytest-xdist>=2.4.0",
     "pytest-trio>=0.5.2",
     "factory-boy>=2.12.0,<3.0.0",
+    "pytest-timeout",
 ]
 
 [tool.setuptools]

tests/core/stream_muxer/test_yamux.py

@@ -51,6 +51,11 @@ async def read(self, n: int | None = None) -> bytes:
             data = await self.receive_stream.receive_some(n)
             logging.debug(f"Read {len(data)} bytes")
             return data
+        # Raise IncompleteReadError on timeout to simulate connection closed
+        logging.debug("Read timed out after 2 seconds, raising IncompleteReadError")
+        from libp2p.io.exceptions import IncompleteReadError
+
+        raise IncompleteReadError({"requested_count": n, "received_count": 0})
 
     async def close(self) -> None:
         logging.debug("Closing stream")

tox.ini

@@ -62,14 +62,16 @@ skip_install=true
 deps=
     wheel
     build[virtualenv]
+platform = win32
 allowlist_externals=
-    bash.exe
+    cmd.exe
 commands=
     python --version
     python -m pip install --upgrade pip
-    bash.exe -c "rm -rf build dist"
+    cmd.exe /c "if exist build rd /s /q build"
+    cmd.exe /c "if exist dist rd /s /q dist"
     python -m build
-    bash.exe -c 'python -m pip install --upgrade "$(ls dist/libp2p-*-py3-none-any.whl)" --progress-bar off'
+    cmd.exe /c "for %i in (dist\libp2p-*-py3-none-any.whl) do python -m pip install --upgrade "%i" --progress-bar off"
     python -c "import libp2p"
 skip_install=true