fix(transport): remove nursery param from IListener.listen (adopts #1055) (#1308)

* refactor(abc): remove nursery parameter from IListener.listen

The caller-supplied nursery was a leaky abstraction that forced Swarm
to manage a listener_nursery on behalf of every transport. Listeners
should own their own background task lifetimes internally.

This commit updates the interface only; concrete implementations follow
in subsequent commits. Part of #726.

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* refactor(transport/tcp): move nursery ownership inside TCPListener

TCPListener now spawns its own internal nursery as a trio system task
in listen(). The nursery hosts trio.serve_tcp and stays open until
close() cancels it.

listen() drops the nursery parameter and returns once the socket is
bound (signalled via an internal trio.Event). Startup errors are
captured and re-raised as OpenConnectionError to preserve the existing
error contract.

close() cancels the internal nursery, closes all SocketListeners, and
waits for the background system task to finish so callers observe a
fully-quiesced listener on return. Safe to call multiple times.

Part of #726.

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* refactor(transport/websocket): move nursery ownership inside WebsocketListener

Same pattern as TCPListener: listen() drops the nursery parameter and
spawns a trio system task that hosts an internal nursery. serve_websocket
runs under that nursery until close() cancels it. Startup errors are
captured and re-raised as OpenConnectionError.

close() now also awaits the background task's completion after cancelling
the nursery, so callers observe a fully-quiesced listener on return.

Part of #726.

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* refactor(transport/quic): remove nursery param from QUICListener.listen

QUIC is a special case: the transport already carries a background
nursery via set_background_nursery(). Prefer that nursery when
available (the common path when used through a Swarm), and fall back
to a self-spawned trio system task with its own internal nursery when
it isn't. Either way, the listen() API no longer requires a caller-
supplied nursery.

close() cancels the self-spawned nursery when it was used and waits
for the background task to finish.

Part of #726.

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* refactor(relay/circuit_v2): drop unused nursery param from CircuitV2Listener.listen

The CircuitV2 listener never used the nursery parameter — it just
records the multiaddr to advertise. Drop the parameter to align with
the updated IListener interface.

Part of #726.

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* refactor(swarm): drop listener_nursery in favor of listener-owned nurseries

Listeners now own their internal nursery, so Swarm no longer manages
listener_nursery / event_listener_nursery_created.  The outer nursery
in run() is still needed for transport-level background tasks (QUIC /
WebSocket set_background_nursery) and for the auto-connector, so it
is renamed to background_nursery / event_background_nursery_created to
reflect its actual purpose.

run() now also calls listener.close() on every listener during
shutdown, so their internal system-task nurseries are cancelled and
fully awaited.

listen() drops the nursery argument from its listener.listen() call.

Part of #726.

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* test(transport): drop nursery arg from all listener.listen() call sites

Updates the full test suite to match the new IListener.listen(maddr)
signature: TCP, WebSocket, QUIC listener/integration tests, shared
trio I/O fixtures, and the tests/utils/factories factory.

Part of #726.

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* docs(news): add breaking changelog entry for #726

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* fix: resolve CI lint failures (ruff F841, mypy no-redef/union-attr, format)

- test_listener.py, test_websocket.py: remove now-unused `nursery`
  variable from `async with trio.open_nursery() as nursery:` blocks
  (ruff F841) — listen() no longer takes a nursery parameter
- quic/listener.py: avoid re-declaring _owned_start_error with a type
  annotation in listen() (mypy no-redef); capture Event objects in
  local vars before entering the system-task closure so the `set()`
  calls resolve against non-Optional types (mypy union-attr)
- tcp/tcp.py: ruff format reformatting

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

* fix(tcp): own a single long-lived nursery across repeated listen() calls

TCPListener used to reset per-call lifecycle state and open a fresh
nursery on every listen(), orphaning earlier server tasks so close()
could only cancel the last bind. Now the background system task and
nursery are created once on first listen() and reused for subsequent
binds. close() cancels them together, is idempotent, and listen()
after close() raises OpenConnectionError.

A trio.Lock serializes concurrent listen()/close() so the lazy spawn
can't race. Adds test_tcp_listener_close_cancels_all_binds covering
multi-bind teardown, idempotent close, and post-close rejection.

Fixes the multi-bind regression acul71 flagged in the PR #1308 review.
Renames the newsfragment to match issue #1314.

* test: close listener explicitly in direct-listener tests and factory

Tests that create TCPListener directly (test_trio_real_connection and
raw_conn_factory) now call await listener.close() before letting the
enclosing nursery cancel. This exercises the new listener-owns-its-
lifetime contract and stops leaking listener tasks between tests.

Addresses acul71's review on PR #1308.

---------

Co-authored-by: Michael Eze <code.maestro64@gmail.com>

Author: yashksaini-coder

libp2p/abc.py

@@ -1421,16 +1421,18 @@ class IListener(ABC):
     """
 
     @abstractmethod
-    async def listen(self, maddr: Multiaddr, nursery: trio.Nursery) -> None:
+    async def listen(self, maddr: Multiaddr) -> None:
         """
         Start listening on the specified multiaddress.
 
+        The listener manages its own background tasks internally and keeps
+        them alive until :meth:`close` is called.  Callers do not need to
+        supply a nursery.
+
         Parameters
         ----------
         maddr : Multiaddr
             The multiaddress on which to listen.
-        nursery : trio.Nursery
-            The nursery for spawning listening tasks.
 
         Raises
         ------

libp2p/network/swarm.py

@@ -98,8 +98,11 @@ class Swarm(Service, INetworkService):
     connections: dict[ID, list[INetConn]]
     listeners: dict[str, IListener]
     common_stream_handler: StreamHandlerFn
-    listener_nursery: trio.Nursery | None
-    event_listener_nursery_created: trio.Event
+    # Background nursery used for transport-level background tasks (QUIC /
+    # WebSocket set_background_nursery, auto-connector).  Listeners no longer
+    # need a caller-supplied nursery — they manage their own internally.
+    background_nursery: trio.Nursery | None
+    event_background_nursery_created: trio.Event
 
     notifees: list[INotifee]
 
@@ -145,8 +148,8 @@ def __init__(
 
         self.common_stream_handler = create_default_stream_handler(self)
 
-        self.listener_nursery = None
-        self.event_listener_nursery_created = trio.Event()
+        self.background_nursery = None
+        self.event_background_nursery_created = trio.Event()
 
         # Load balancing state
         self._round_robin_index = {}
@@ -205,8 +208,10 @@ def set_resource_manager(
 
     async def run(self) -> None:
         async with trio.open_nursery() as nursery:
-            # Create a nursery for listener tasks.
-            self.listener_nursery = nursery
+            # This nursery hosts transport-level background tasks (QUIC /
+            # WebSocket) and the auto-connector.  Listeners own their own
+            # internal nurseries and no longer use this one.
+            self.background_nursery = nursery
 
             # Set background nursery BEFORE setting the event
             # This ensures transports have the nursery when they check
@@ -218,9 +223,8 @@ async def run(self) -> None:
                 # for connection management
                 self.transport.set_background_nursery(nursery)  # type: ignore[attr-defined]
 
-            # Set event after background nursery is configured
-            # This ensures transports have the nursery when they check the event
-            self.event_listener_nursery_created.set()
+            # Signal that the background nursery is available.
+            self.event_background_nursery_created.set()
 
             # Start connection management components (go-libp2p style)
             try:
@@ -244,10 +248,17 @@ async def run(self) -> None:
                         f"Error stopping connection management components: {e}"
                     )
 
-                # The service ended. Cancel listener tasks.
+                # Close all listeners so their internal nurseries are
+                # cancelled and system tasks finish cleanly.
+                for listener in list(self.listeners.values()):
+                    try:
+                        await listener.close()
+                    except Exception as e:
+                        logger.debug("Error closing listener during shutdown: %s", e)
+
+                # Cancel the background nursery (transport / auto-connector).
                 nursery.cancel_scope.cancel()
-                # Indicate that the nursery has been cancelled.
-                self.listener_nursery = None
+                self.background_nursery = None
 
     def get_peer_id(self) -> ID:
         return self.self_id
@@ -1119,9 +1130,11 @@ async def listen(self, *multiaddrs: Multiaddr) -> bool:
               - Map multiaddr to listener
         """
         logger.debug(f"Swarm.listen called with multiaddrs: {multiaddrs}")
-        # We need to wait until `self.listener_nursery` is created.
+        # Wait until the background nursery is available so that transports
+        # which need it (QUIC, WebSocket) can reach it via their transport
+        # reference.  Listeners themselves no longer require a nursery.
         logger.debug("Starting to listen")
-        await self.event_listener_nursery_created.wait()
+        await self.event_background_nursery_created.wait()
 
         success_count = 0
         for maddr in multiaddrs:
@@ -1199,13 +1212,10 @@ async def conn_handler(
                 listener = self.transport.create_listener(conn_handler)
                 logger.debug(f"Swarm.listen: listener created for {maddr}")
                 self.listeners[str(maddr)] = listener
-                # TODO: `listener.listen` is not bounded with nursery. If we want to be
-                #   I/O agnostic, we should change the API.
-                if self.listener_nursery is None:
+                if self.background_nursery is None:
                     raise SwarmException("swarm instance hasn't been run")
-                assert self.listener_nursery is not None  # For type checker
                 logger.debug(f"Swarm.listen: calling listener.listen for {maddr}")
-                await listener.listen(maddr, self.listener_nursery)
+                await listener.listen(maddr)
                 logger.debug(f"Swarm.listen: listener.listen completed for {maddr}")
 
                 # Call notifiers since event occurred

libp2p/relay/circuit_v2/transport.py

@@ -1077,16 +1077,14 @@ async def stream_handler(stream: INetStream) -> None:
         finally:
             logger.debug("CircuitV2Listener stopped")
 
-    async def listen(self, maddr: multiaddr.Multiaddr, nursery: trio.Nursery) -> None:
+    async def listen(self, maddr: multiaddr.Multiaddr) -> None:
         """
         Start listening on the given multiaddr.
 
         Parameters
         ----------
         maddr : multiaddr.Multiaddr
             The multiaddr to listen on
-        nursery : trio.Nursery
-            The nursery to run tasks in
 
         """
         # Convert string to Multiaddr if needed

libp2p/transport/quic/listener.py

@@ -107,6 +107,12 @@ def __init__(
         self._listening = False
         self._nursery: trio.Nursery | None = None
 
+        # State for the "owned nursery" fallback path in listen().  Populated
+        # only when no transport background nursery is available.
+        self._owned_started: trio.Event | None = None
+        self._owned_stopped: trio.Event | None = None
+        self._owned_start_error: BaseException | None = None
+
         # Serialize promotion of pending connections to avoid duplicate promotions
         # and repeated connect() calls under concurrent packet processing.
         self._promotion_lock = trio.Lock()
@@ -1197,49 +1203,90 @@ async def _transmit_for_connection(
         except Exception as e:
             logger.error(f"Transmission error: {e}", exc_info=True)
 
-    async def listen(self, maddr: Multiaddr, nursery: trio.Nursery) -> None:
-        """Start listening on the given multiaddr with enhanced connection handling."""
+    async def listen(self, maddr: Multiaddr) -> None:
+        """
+        Start listening on the given multiaddr with enhanced connection handling.
+
+        The listener uses the transport's background nursery when available
+        (set via :meth:`QUICTransport.set_background_nursery`), and otherwise
+        spawns a private internal nursery as a trio system task.  In either
+        case the listener no longer requires the caller to supply a nursery.
+        """
         if self._listening:
             raise QUICListenError("Already listening")
 
         if not is_quic_multiaddr(maddr):
             raise QUICListenError(f"Invalid QUIC multiaddr: {maddr}")
 
-        if self._transport._background_nursery:
-            active_nursery = self._transport._background_nursery
-            logger.debug("Using transport background nursery for listener")
-        elif nursery:
-            active_nursery = nursery
-            self._transport._background_nursery = nursery
-            logger.debug("Using provided nursery for listener")
-        else:
-            raise QUICListenError("No nursery available")
+        host, port = quic_multiaddr_to_endpoint(maddr)
 
-        try:
-            host, port = quic_multiaddr_to_endpoint(maddr)
+        if self._transport._background_nursery is not None:
+            # Preferred path: reuse the transport's already-running nursery.
+            try:
+                self._socket = await self._create_socket(host, port)
+                self._nursery = self._transport._background_nursery
 
-            # Create and configure socket
-            self._socket = await self._create_socket(host, port)
-            self._nursery = active_nursery
+                bound_host, bound_port = self._socket.getsockname()[:2]
+                quic_version = multiaddr_to_quic_version(maddr)
+                bound_maddr = create_quic_multiaddr(
+                    bound_host, bound_port, quic_version
+                )
+                self._bound_addresses = [bound_maddr]
+                self._listening = True
 
-            # Get the actual bound address (IPv4: 2-tuple, IPv6: 4-tuple)
-            bound_host, bound_port = self._socket.getsockname()[:2]
-            quic_version = multiaddr_to_quic_version(maddr)
-            bound_maddr = create_quic_multiaddr(bound_host, bound_port, quic_version)
-            self._bound_addresses = [bound_maddr]
+                self._nursery.start_soon(self._handle_incoming_packets)
+                logger.info(
+                    f"QUIC listener started on {bound_maddr} with connection ID support"
+                )
+                return
+            except Exception as e:
+                await self.close()
+                raise QUICListenError(f"Failed to start listening: {e}") from e
+
+        # Fallback: spawn our own internal nursery as a trio system task
+        # so the listener is fully self-contained.
+        started = trio.Event()
+        stopped = trio.Event()
+        self._owned_started = started
+        self._owned_stopped = stopped
+        self._owned_start_error = None
+
+        async def _run_server() -> None:
+            try:
+                async with trio.open_nursery() as inner_nursery:
+                    try:
+                        self._socket = await self._create_socket(host, port)
+                        self._nursery = inner_nursery
 
-            self._listening = True
+                        bound_host, bound_port = self._socket.getsockname()[:2]
+                        quic_version = multiaddr_to_quic_version(maddr)
+                        bound_maddr = create_quic_multiaddr(
+                            bound_host, bound_port, quic_version
+                        )
+                        self._bound_addresses = [bound_maddr]
+                        self._listening = True
 
-            # Start packet handling loop
-            active_nursery.start_soon(self._handle_incoming_packets)
+                        inner_nursery.start_soon(self._handle_incoming_packets)
+                        logger.info(
+                            f"QUIC listener started on {bound_maddr} "
+                            "with connection ID support"
+                        )
+                    except BaseException as error:
+                        self._owned_start_error = error
+                    finally:
+                        started.set()
+            finally:
+                stopped.set()
+                self._nursery = None
 
-            logger.info(
-                f"QUIC listener started on {bound_maddr} with connection ID support"
-            )
+        trio.lowlevel.spawn_system_task(_run_server)
+        await self._owned_started.wait()
 
-        except Exception as e:
+        if self._owned_start_error is not None:
             await self.close()
-            raise QUICListenError(f"Failed to start listening: {e}") from e
+            raise QUICListenError(
+                f"Failed to start listening: {self._owned_start_error}"
+            ) from self._owned_start_error
 
     async def _create_socket(self, host: str, port: int) -> trio.socket.SocketType:
         """Create and configure UDP socket."""
@@ -1333,6 +1380,14 @@ async def close(self) -> None:
 
             self._bound_addresses.clear()
 
+            # If we spawned our own internal nursery (fallback path),
+            # cancel it and wait for the background system task to finish.
+            if self._owned_started is not None and self._owned_started.is_set():
+                if self._nursery is not None:
+                    self._nursery.cancel_scope.cancel()
+                if self._owned_stopped is not None:
+                    await self._owned_stopped.wait()
+
             logger.info("QUIC listener closed")
 
         except Exception as e: