python-zeroconf
diff --git a/src/zeroconf/_core.py b/src/zeroconf/_core.py
@@ -676,13 +676,22 @@ def _close(self) -> None:
 
     def _shutdown_threads(self) -> None:
         """Shutdown any threads."""
+        assert self.loop is not None
+        if self.loop.is_closed():
+            # close() is documented as idempotent — a second call after the
+            # loop has been torn down must be a no-op rather than raising.
+            return
         self.notify_all()
         if not self._loop_thread:
             return
-        assert self.loop is not None
         shutdown_loop(self.loop)
         self._loop_thread.join()
         self._loop_thread = None
+        # The loop's selector (epoll FD on Linux) and self-pipe sockets stay
+        # open until loop.close() is called. We own this loop because
+        # _start_thread() created it, so close it here to avoid leaking
+        # those file descriptors across Zeroconf() construct/close cycles.
+        self.loop.close()
 
     def close(self) -> None:
         """Ends the background threads, and prevent this instance from

diff --git a/src/zeroconf/_services/browser.py b/src/zeroconf/_services/browser.py
@@ -23,6 +23,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import heapq
 import queue
 import random
@@ -793,7 +794,16 @@ def cancel(self) -> None:
         """Cancel the browser."""
         assert self.zc.loop is not None
         self.queue.put(None)
-        self.zc.loop.call_soon_threadsafe(self._async_cancel)
+        # While the loop is running, _async_cancel stops the query scheduler
+        # and cancels the query-sender task — that is the normal cleanup
+        # path. Skip scheduling solely because the loop is closed: a closed
+        # loop rejects call_soon_threadsafe with RuntimeError. The
+        # is_closed() check narrows the common case (loop already closed by
+        # Zeroconf.close()) without paying for raise/catch; suppress covers
+        # the residual is_closed() -> call_soon_threadsafe race window.
+        with contextlib.suppress(RuntimeError):
+            if not self.zc.loop.is_closed():
+                self.zc.loop.call_soon_threadsafe(self._async_cancel)
         self.join()
 
     def run(self) -> None:

diff --git a/tests/test_core.py b/tests/test_core.py
@@ -12,6 +12,7 @@
 import unittest
 import unittest.mock
 import warnings
+from pathlib import Path
 from typing import cast
 from unittest.mock import AsyncMock, Mock, patch
 
@@ -83,6 +84,40 @@ def test_close_multiple_times(self):
         rv.close()
         rv.close()
 
+    def test_close_releases_owned_event_loop(self):
+        """Closing a Zeroconf that started its own loop thread closes that loop.
+
+        Regression test for issue #1589 — without loop.close(), the selector
+        (epoll on Linux) and its self-pipe sockets stay open across each
+        Zeroconf construct/close cycle and the process eventually exhausts
+        its FD limit.
+        """
+        rv = r.Zeroconf(interfaces=["127.0.0.1"])
+        loop = rv.loop
+        assert loop is not None
+        assert loop.is_running()
+        rv.close()
+        assert loop.is_closed()
+
+    @unittest.skipUnless(sys.platform.startswith("linux"), "Requires /proc/<pid>/fd")
+    @unittest.skipUnless(Path(f"/proc/{os.getpid()}/fd").is_dir(), "/proc/<pid>/fd not available")
+    def test_close_does_not_leak_file_descriptors(self):
+        """Tight loops of Zeroconf()/close() do not leak FDs (issue #1589)."""
+        fd_dir = Path(f"/proc/{os.getpid()}/fd")
+
+        def _fd_count() -> int:
+            return sum(1 for _ in fd_dir.iterdir())
+
+        # Warm-up cycle so any one-shot import-time FDs land before measuring.
+        r.Zeroconf(interfaces=["127.0.0.1"]).close()
+        baseline = _fd_count()
+        for _ in range(10):
+            r.Zeroconf(interfaces=["127.0.0.1"]).close()
+        # Allow tiny slack for unrelated FDs the test harness may open
+        # (e.g. coverage), but reject the per-cycle linear growth pattern
+        # the bug produced (~3 FDs per cycle, so >=30 over 10 cycles).
+        assert _fd_count() - baseline < 10
+
     @unittest.skipIf(not has_working_ipv6(), "Requires IPv6")
     @unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled")
     def test_launch_and_close_v4_v6(self):