fix: improve EOFError exception when remote nvim crashes

Problem: When the remote Nvim instance is aborted or terminates, the
pynvim client will face an exception `OSError: EOF`. It is not very
clear what is wrong and why EOF is received.

Solution: This happens when the remote the nvim process terminates
unexpectedly (or a tcp/file socket is broken or gets disconnected). We
can provide a bit more detailed information about why the asyncio
session is stopping, through asyncio Protocol. An `EOFError` will be
raised instead of OSError.

For example, during pynvim's unit tests we may see:
```
    EOFError: process_exited: pid = 40000, return_code = -6
```
which means that the Nvim subprocess (pid = 40000) exited unexpectedly
after getting SIGABRT (signal 6), i.e. segfault.

Author: wookayin

pynvim/msgpack_rpc/event_loop/asyncio.py

@@ -18,7 +18,7 @@
 from pynvim.msgpack_rpc.event_loop.base import BaseEventLoop, TTransportType
 
 logger = logging.getLogger(__name__)
-debug, info, warn = (logger.debug, logger.info, logger.warning,)
+debug, info, warn = (logger.debug, logger.info, logger.warning)
 
 loop_cls = asyncio.SelectorEventLoop
 
@@ -47,13 +47,14 @@ def __init__(self, on_data, on_error):
     @override
     def connection_made(self, transport):
         """Used to signal `asyncio.Protocol` of a successful connection."""
-        del transport  # no-op
+        self._transport = transport
 
     @override
     def connection_lost(self, exc: Optional[Exception]) -> None:
         """Used to signal `asyncio.Protocol` of a lost connection."""
-        debug(f"connection_lost: exc = {exc}")
-        self._on_error(exc if exc else EOFError())
+        warn(f"connection_lost: exc = {exc}")
+
+        self._on_error(exc if exc else EOFError("connection_lost"))
 
     @override
     def data_received(self, data: bytes) -> None:
@@ -63,11 +64,19 @@ def data_received(self, data: bytes) -> None:
     @override
     def pipe_connection_lost(self, fd: int, exc: Optional[Exception]) -> None:
         """Used to signal `asyncio.SubprocessProtocol` of a lost connection."""
-        debug("pipe_connection_lost: fd = %s, exc = %s", fd, exc)
+
+        assert isinstance(self._transport, asyncio.SubprocessTransport)
+        debug_info = {'fd': fd, 'exc': exc, 'pid': self._transport.get_pid()}
+        warn(f"pipe_connection_lost {debug_info}")
+
         if os.name == 'nt' and fd == 2:  # stderr
             # On windows, ignore piped stderr being closed immediately (#505)
             return
-        self._on_error(exc if exc else EOFError())
+
+        # pipe_connection_lost() *may* be called before process_exited() is
+        # called, when a Nvim subprocess crashes (SIGABRT). Do not handle
+        # errors here, as errors will be handled somewhere else
+        # self._on_error(exc if exc else EOFError("pipe_connection_lost"))
 
     @override
     def pipe_data_received(self, fd, data):
@@ -81,8 +90,13 @@ def pipe_data_received(self, fd, data):
     @override
     def process_exited(self) -> None:
         """Used to signal `asyncio.SubprocessProtocol` when the child exits."""
-        debug("process_exited")
-        self._on_error(EOFError())
+        assert isinstance(self._transport, asyncio.SubprocessTransport)
+        pid = self._transport.get_pid()
+        return_code = self._transport.get_returncode()
+
+        warn("process_exited, pid = %s, return_code = %s", pid, return_code)
+        err = EOFError(f"process_exited: pid = {pid}, return_code = {return_code}")
+        self._on_error(err)
 
 
 class AsyncioEventLoop(BaseEventLoop):
@@ -131,7 +145,8 @@ def _on_data(data: bytes) -> None:
     def _connect_tcp(self, address: str, port: int) -> None:
         async def connect_tcp():
             transport, protocol = await self._loop.create_connection(
-                self._protocol_factory, address, port)
+                self._protocol_factory, address, port
+            )
             debug(f"tcp connection successful: {address}:{port}")
             self._transport = transport
             self._protocol = protocol
@@ -146,8 +161,7 @@ async def connect_socket():
             else:
                 _create_connection = self._loop.create_unix_connection
 
-            transport, protocol = await _create_connection(
-                self._protocol_factory, path)
+            transport, protocol = await _create_connection(self._protocol_factory, path)
             debug("socket connection successful: %s", self._transport)
             self._transport = transport
             self._protocol = protocol
@@ -162,10 +176,12 @@ async def connect_stdin():
             else:
                 pipe = sys.stdin
             transport, protocol = await self._loop.connect_read_pipe(
-                self._protocol_factory, pipe)
+                self._protocol_factory, pipe
+            )
             debug("native stdin connection successful")
             self._to_close.append(transport)
             del protocol
+
         self._loop.run_until_complete(connect_stdin())
 
         # Make sure subprocesses don't clobber stdout,
@@ -177,26 +193,29 @@ async def connect_stdout():
             if os.name == 'nt':
                 pipe = PipeHandle(msvcrt.get_osfhandle(rename_stdout))
             else:
-                pipe = os.fdopen(rename_stdout, 'wb')
+                pipe = os.fdopen(rename_stdout, "wb")
 
             transport, protocol = await self._loop.connect_write_pipe(
-                self._protocol_factory, pipe)
+                self._protocol_factory, pipe
+            )
             debug("native stdout connection successful")
             self._transport = transport
             self._protocol = protocol
+
         self._loop.run_until_complete(connect_stdout())
 
     @override
     def _connect_child(self, argv: List[str]) -> None:
-        if os.name != 'nt':
+        if os.name != "nt":
             # see #238, #241
             self._child_watcher = asyncio.get_child_watcher()
             self._child_watcher.attach_loop(self._loop)
 
         async def create_subprocess():
             transport: asyncio.SubprocessTransport  # type: ignore
             transport, protocol = await self._loop.subprocess_exec(
-                self._protocol_factory, *argv)
+                self._protocol_factory, *argv
+            )
             pid = transport.get_pid()
             debug("child subprocess_exec successful, PID = %s", pid)
 
@@ -250,11 +269,13 @@ def _close_transport(transport):
             # Windows: for ProactorBasePipeTransport, close() doesn't take in
             # effect immediately (closing happens asynchronously inside the
             # event loop), need to wait a bit for completing graceful shutdown.
-            if os.name == 'nt' and hasattr(transport, '_sock'):
+            if os.name == 'nt' and hasattr(transport, "_sock"):
+
                 async def wait_until_closed():
                     # pylint: disable-next=protected-access
                     while transport._sock is not None:
                         await asyncio.sleep(0.01)
+
                 self._loop.run_until_complete(wait_until_closed())
 
         if self._transport:

pynvim/msgpack_rpc/event_loop/base.py

@@ -194,6 +194,13 @@ def run(self, data_cb: Callable[[bytes], None]) -> None:
             signal.signal(signal.SIGINT, default_int_handler)
         self._on_data = None
 
+        # eventloop was stopped due to an error, re-raise it
+        # (e.g. connection lost when subprocess nvim dies)
+        if self._error:
+            # Note: traceback is not preserved and attached for some reason,
+            # should be somewhere from msgpack_rpc.event_loop.asyncio.Protocol
+            raise self._error
+
     @abstractmethod
     def _run(self) -> None:
         raise NotImplementedError()
@@ -234,8 +241,11 @@ def _on_signal(self, signum: signal.Signals) -> None:
         self.stop()
 
     def _on_error(self, exc: Exception) -> None:
-        debug(str(exc))
-        self._error = exc
+        warn('on_error: %s', repr(exc))
+        if self._error is None:
+            # ignore subsequent exceptions, it's enough to raise only
+            # the first exception arrived
+            self._error = exc
         self.stop()
 
     def _on_interrupt(self) -> None: