Ayush701-code-zm · Nov 10, 2026
diff --git a/tests/test_base.py b/tests/test_base.py
@@ -773,6 +773,98 @@ def scheduler():
         thread.join()
         self.assertEqual(counter[0], ITERATIONS)
 
+    def _poll_error_pipe(self):
+        # A pipe whose read end is closed: on Linux, epoll reports
+        # EPOLLERR for the write end.
+        r, w = os.pipe()
+        os.set_blocking(w, False)
+        self.addCleanup(os.close, w)
+        return r, w
+
+    def _run_briefly_for(self, delay):
+        self.loop.call_later(delay, self.loop.stop)
+        self.loop.run_forever()
+
+    @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux only')
+    def test_add_writer_keeps_watching_after_error(self):
+        # https://github.com/MagicStack/uvloop/issues/787
+        errors = []
+        self.loop.set_exception_handler(lambda loop, ctx: errors.append(ctx))
+        r, w = self._poll_error_pipe()
+        calls = 0
+
+        def writer():
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                os.close(r)
+            elif calls == 100:
+                self.loop.remove_writer(w)
+
+        self.loop.add_writer(w, writer)
+        self._run_briefly_for(0.2)
+
+        self.assertEqual(calls, 100)
+        self.assertEqual(errors, [])
+
+    def _test_add_writer_after_error(self, remove_first):
+        errors = []
+        self.loop.set_exception_handler(lambda loop, ctx: errors.append(ctx))
+        r, w = self._poll_error_pipe()
+        calls = []
+
+        def writer():
+            calls.append('call')
+            if len(calls) == 1:
+                os.close(r)
+
+        def rearm():
+            calls.append('rearm')
+            if remove_first:
+                self.loop.remove_writer(w)
+            self.loop.add_writer(w, writer)
+
+        self.loop.add_writer(w, writer)
+        self.loop.call_later(0.1, rearm)
+        self._run_briefly_for(0.2)
+        self.loop.remove_writer(w)
+
+        i = calls.index('rearm')
+        self.assertGreater(i, 3)
+        self.assertGreater(len(calls) - i - 1, 3)
+        self.assertEqual(errors, [])
+
+    @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux only')
+    def test_add_writer_after_error(self):
+        self._test_add_writer_after_error(remove_first=False)
+
+    @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux only')
+    def test_remove_and_add_writer_after_error(self):
+        self._test_add_writer_after_error(remove_first=True)
+
+    @unittest.skipUnless(sys.platform.startswith('linux'), 'Linux only')
+    def test_add_writer_rearm_in_callback_after_error(self):
+        errors = []
+        self.loop.set_exception_handler(lambda loop, ctx: errors.append(ctx))
+        r, w = self._poll_error_pipe()
+        calls = 0
+
+        def writer():
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                os.close(r)
+            elif calls == 2:
+                self.loop.add_writer(w, writer)
+            elif calls == 100:
+                self.loop.remove_writer(w)
+
+        self.loop.add_writer(w, writer)
+        self._run_briefly_for(0.2)
+
+        self.assertEqual(calls, 100)
+        self.assertEqual(errors, [])
+
     def test_freethreading(self):
         if not hasattr(sys, "_is_gil_enabled"):
             raise unittest.SkipTest("No sys._is_gil_enabled()")

diff --git a/uvloop/handles/poll.pxd b/uvloop/handles/poll.pxd
@@ -21,5 +21,8 @@ cdef class UVPoll(UVHandle):
     cdef stop_writing(self)
     cdef stop(self)
 
+    cdef _on_poll_error(self, exc)
+    cdef _close_on_error(self, exc)
+
     @staticmethod
     cdef UVPoll new(Loop loop, int fd)
diff --git a/uvloop/handles/poll.pyx b/uvloop/handles/poll.pyx
@@ -189,6 +189,67 @@ cdef class UVPoll(UVHandle):
         finally:
             self._close()
 
+    cdef _on_poll_error(self, exc):
+        # On Linux, libuv reports POLLERR as UV_EBADF after it has
+        # stopped the watcher.  To match asyncio (level-triggered
+        # epoll), run the callbacks as for a ready fd and keep
+        # watching the fd until the callbacks are removed.
+        cdef:
+            int mask = 0
+            int err
+
+        if not self.is_active():
+            self._close_on_error(None)
+            return
+
+        if self.reading_handle is not None:
+            try:
+                self.reading_handle._run()
+            except BaseException as ex:
+                self._error(ex, False)
+
+        if self.writing_handle is not None:
+            try:
+                self.writing_handle._run()
+            except BaseException as ex:
+                self._error(ex, False)
+
+        if not self._is_alive():
+            # The callbacks removed the reader and the writer,
+            # and the loop closed the handle.
+            return
+
+        if self.reading_handle is not None:
+            mask |= uv.UV_READABLE
+        if self.writing_handle is not None:
+            mask |= uv.UV_WRITABLE
+        if mask == 0:
+            return
+
+        try:
+            # Restarting the watcher on a closed fd would make libuv
+            # abort() in epoll_ctl(), so check the fd first.
+            os_fstat(self.fd)
+        except OSError as ex:
+            self._close_on_error(ex)
+            return
+
+        err = uv.uv_poll_start(
+            <uv.uv_poll_t*>self._handle,
+            mask,
+            __on_uvpoll_event)
+        if err < 0:
+            self._close_on_error(convert_error(err))
+
+    cdef _close_on_error(self, exc):
+        # Remove the closed handle from the loop, so that a later
+        # add_reader() or add_writer() creates a new one.
+        if self._loop._polls.get(self.fd) is self:
+            del self._loop._polls[self.fd]
+        self._close()
+        if exc is not None:
+            self._loop._handle_exception(exc)
+
 
 cdef void __on_uvpoll_event(
     uv.uv_poll_t* handle,
@@ -204,7 +265,7 @@ cdef void __on_uvpoll_event(
 
     if status < 0:
         exc = convert_error(status)
-        poll._fatal_error(exc, False)
+        poll._on_poll_error(exc)
         return
 
     if ((events & (uv.UV_READABLE | uv.UV_DISCONNECT)) and

diff --git a/uvloop/includes/stdlib.pxi b/uvloop/includes/stdlib.pxi
@@ -112,6 +112,7 @@ cdef os_pipe = os.pipe
 cdef os_read = os.read
 cdef os_remove = os.remove
 cdef os_stat = os.stat
+cdef os_fstat = os.fstat
 cdef os_unlink = os.unlink
 cdef os_fspath = os.fspath