Fix EOF for buffered app protocol

tests/test_unix.py

@@ -471,6 +471,57 @@ def test_create_unix_server_path_stream_bittype(self):
         finally:
             os.unlink(fn)
 
+    @unittest.skipUnless(sys.platform.startswith('linux'), 'requires epoll')
+    def test_epollhup(self):
+        SIZE = 50
+        eof = False
+        done = False
+        recvd = b''
+
+        class Proto(asyncio.BaseProtocol):
+            def connection_made(self, tr):
+                tr.write(b'hello')
+                self.data = bytearray(SIZE)
+                self.buf = memoryview(self.data)
+
+            def get_buffer(self, sizehint):
+                return self.buf
+
+            def buffer_updated(self, nbytes):
+                nonlocal recvd
+                recvd += self.buf[:nbytes]
+
+            def eof_received(self):
+                nonlocal eof
+                eof = True
+
+            def connection_lost(self, exc):
+                nonlocal done
+                done = exc
+
+        async def test():
+            with tempfile.TemporaryDirectory() as td:
+                sock_name = os.path.join(td, 'sock')
+                srv = await self.loop.create_unix_server(Proto, sock_name)
+
+                s = socket.socket(socket.AF_UNIX)
+                with s:
+                    s.setblocking(False)
+                    await self.loop.sock_connect(s, sock_name)
+                    d = await self.loop.sock_recv(s, 100)
+                    self.assertEqual(d, b'hello')
+
+                    # IMPORTANT: overflow recv buffer and close immediately
+                    await self.loop.sock_sendall(s, b'a' * (SIZE + 1))
+
+                srv.close()
+                await srv.wait_closed()
+
+        self.loop.run_until_complete(test())
+        self.assertTrue(eof)
+        self.assertIsNone(done)
+        self.assertEqual(recvd, b'a' * (SIZE + 1))
+
 
 class Test_AIO_Unix(_TestUnix, tb.AIOTestCase):
     pass

uvloop/handles/stream.pyx

@@ -953,7 +953,12 @@ cdef void __uv_stream_buffered_on_read(uv.uv_stream_t* stream,
         return
 
     try:
-        if not sc._read_pybuf_acquired:
+        if nread > 0 and not sc._read_pybuf_acquired:
+            # From libuv docs:
+            #     nread is > 0 if there is data available or < 0 on error. When
+            #     we’ve reached EOF, nread will be set to UV_EOF. When
+            #     nread < 0, the buf parameter might not point to a valid
+            #     buffer; in that case buf.len and buf.base are both set to 0.
             raise RuntimeError(
                 f'no python buffer is allocated in on_read; nread={nread}')