fix(forward): make tracker connection_lost fire exactly once

The local-forward _forward() coroutine calls connection_lost(exc) manually
on channel-open failure, and the local transport's later close fires a
second connection_lost(None) on the protocol via asyncio. Clearing
self._tracker on the first call makes the hook fire exactly once.

Also tightens the patch:
- Drop redundant self._tracker = None in _create_tracker's except (the
  __init__ default already covers a raising factory).
- New regression test asserting the tracker's connection_lost fires
  exactly once on a denied-by-accept_handler forward.
- Replace asyncio.sleep(0.1) await-for-lost in existing tests with a
  per-test asyncio.Event resolved by the hook itself.

Author: AlexMKX

asyncssh/forward.py

@@ -316,8 +316,9 @@ def _create_tracker(self) -> None:
         try:
             self._tracker = self._tracker_factory()
         except Exception: # pylint: disable=broad-exception-caught
-            # A buggy factory must not break forwarding.
-            self._tracker = None
+            # A buggy factory must not break forwarding;
+            # self._tracker remains the __init__ default of None.
+            pass
 
     def data_received(self, data: bytes,
                       datatype: Optional[int] = None) -> None:
@@ -343,11 +344,20 @@ def write(self, data: bytes) -> None:
         super().write(data)
 
     def connection_lost(self, exc: Optional[Exception]) -> None:
-        """Handle a closed local connection"""
+        """Handle a closed local connection
 
-        if self._tracker is not None:
+           This is also called manually from `_forward()` on a channel
+           open failure, so the local transport's eventual close fires
+           a second `connection_lost(None)` on the protocol. The tracker
+           reference is cleared on the first call so the hook fires
+           exactly once per connection.
+        """
+
+        tracker, self._tracker = self._tracker, None
+
+        if tracker is not None:
             try:
-                self._tracker.connection_lost(exc)
+                tracker.connection_lost(exc)
             except Exception: # pylint: disable=broad-exception-caught
                 pass
 

tests/test_forward.py

@@ -702,20 +702,22 @@ async def test_forward_local_port_tracker_factory_fires_made_and_lost(self):
         """A port tracker sees connection_made and connection_lost"""
 
         events = []
+        lost = asyncio.Event()
 
         class _RecordingTracker(asyncssh.SSHPortForwardTracker):
             def connection_made(self, forwarder, orig_host, orig_port):
                 events.append(('made', forwarder, orig_host, orig_port))
 
             def connection_lost(self, exc):
                 events.append(('lost', exc))
+                lost.set()
 
         async with self.connect() as conn:
             async with conn.forward_local_port(
                     '', 0, '', 7,
                     tracker_factory=_RecordingTracker) as listener:
                 await self._check_local_connection(listener.get_port())
-                await asyncio.sleep(0.1)
+                await asyncio.wait_for(lost.wait(), timeout=1.0)
 
         kinds = [event[0] for event in events]
         self.assertIn('made', kinds)
@@ -731,10 +733,15 @@ async def test_forward_local_port_tracker_factory_per_connection(self):
         """The factory is called once per accepted connection"""
 
         trackers = []
+        lost_events = []
 
         class _CountingTracker(asyncssh.SSHPortForwardTracker):
             def __init__(self):
                 trackers.append(self)
+                lost_events.append(asyncio.Event())
+
+            def connection_lost(self, exc):
+                lost_events[trackers.index(self)].set()
 
         def factory():
             return _CountingTracker()
@@ -745,7 +752,9 @@ def factory():
                 listen_port = listener.get_port()
                 await self._check_local_connection(listen_port)
                 await self._check_local_connection(listen_port)
-                await asyncio.sleep(0.1)
+                await asyncio.wait_for(
+                    asyncio.gather(*(e.wait() for e in lost_events)),
+                    timeout=1.0)
 
         self.assertEqual(len(trackers), 2)
 
@@ -755,6 +764,7 @@ async def test_forward_local_port_tracker_byte_hooks(self):
 
         local_bytes = bytearray()
         remote_bytes = bytearray()
+        lost = asyncio.Event()
 
         class _ByteTracker(asyncssh.SSHPortForwardTracker):
             def forward_local_bytes(self, data):
@@ -763,12 +773,15 @@ def forward_local_bytes(self, data):
             def forward_remote_bytes(self, data):
                 remote_bytes.extend(data)
 
+            def connection_lost(self, exc):
+                lost.set()
+
         async with self.connect() as conn:
             async with conn.forward_local_port(
                     '', 0, '', 7,
                     tracker_factory=_ByteTracker) as listener:
                 await self._check_local_connection(listener.get_port())
-                await asyncio.sleep(0.1)
+                await asyncio.wait_for(lost.wait(), timeout=1.0)
 
         line = (str(id(self)) + '\n').encode('utf-8')
         self.assertEqual(bytes(local_bytes), line)
@@ -809,7 +822,36 @@ def forward_remote_bytes(self, data):
                     '', 0, '', 7,
                     tracker_factory=_BuggyTracker) as listener:
                 await self._check_local_connection(listener.get_port(),
-                                                   delay=0.1)
+                                                    delay=0.1)
+
+    @asynctest
+    async def test_forward_local_port_tracker_lost_fires_exactly_once(self):
+        """connection_lost fires once even when ChannelOpenError triggers
+           a manual notify in _forward() followed by the asyncio close path."""
+
+        lost_count = 0
+
+        class _Counting(asyncssh.SSHPortForwardTracker):
+            def connection_lost(self, exc):
+                nonlocal lost_count
+                lost_count += 1
+
+        async def deny(_h, _p):
+            return False
+
+        async with self.connect() as conn:
+            async with conn.forward_local_port(
+                    '', 0, '', 7,
+                    accept_handler=deny,
+                    tracker_factory=_Counting) as listener:
+                reader, writer = await asyncio.open_connection(
+                    '127.0.0.1', listener.get_port())
+                self.assertEqual((await reader.read()), b'')
+                writer.close()
+                await maybe_wait_closed(writer)
+                await asyncio.sleep(0.1)  # bounded: upper-bound for any spurious duplicate
+
+        self.assertEqual(lost_count, 1)
 
     @unittest.skipIf(sys.platform == 'win32',
                      'skip UNIX domain socket tests on Windows')
@@ -1268,20 +1310,22 @@ async def test_forward_local_path_tracker_factory(self):
         """A path tracker sees connection_made (no addr) and connection_lost"""
 
         events = []
+        lost = asyncio.Event()
 
         class _RecordingTracker(asyncssh.SSHPathForwardTracker):
             def connection_made(self, forwarder):
                 events.append(('made', forwarder))
 
             def connection_lost(self, exc):
                 events.append(('lost', exc))
+                lost.set()
 
         async with self.connect() as conn:
             async with conn.forward_local_path(
                     'local', '/echo',
                     tracker_factory=_RecordingTracker):
                 await self._check_local_unix_connection('local')
-                await asyncio.sleep(0.1)
+                await asyncio.wait_for(lost.wait(), timeout=1.0)
 
         try_remove('local')