Fix race condition during ssh session probe

docs/source/changelog.rst

@@ -1,12 +1,12 @@
-.. Created by changelog.py at 2025-04-08, command
+.. Created by changelog.py at 2025-04-15, command
    '/Users/giffler/.cache/pre-commit/repoecmh3ah8/py_env-python3.12/bin/changelog docs/source/changes compile --categories Added Changed Fixed Security Deprecated --output=docs/source/changelog.rst'
    based on the format of 'https://keepachangelog.com/'
 
 #########
 CHANGELOG
 #########
 
-[Unreleased] - 2025-04-08
+[Unreleased] - 2025-04-15
 =========================
 
 Added

tardis/utilities/executors/sshexecutor.py

@@ -142,7 +142,7 @@ async def bounded_connection(self):
         :py:class:`~asyncssh.SSHClientConnection`
         so that only `MaxSessions` commands run at once.
         """
-        if self._ssh_connection is None:
+        if self._ssh_connection is None or self._session_bound is None:
             async with self.lock:
                 # check that connection has not been initialized in a different task
                 while self._ssh_connection is None:

tests/utilities_t/executors_t/test_sshexecutor.py

@@ -205,6 +205,36 @@ async def force_connection():
         # make sure the connection is not needlessly replaced
         self.assertEqual(self.executor._ssh_connection, current_ssh_connection)
 
+    def test_connection_race(self):
+        # see https://github.com/MatterMiners/tardis/issues/369
+        waiter = asyncio.Event()
+
+        async def mocked_probe_max_session(connection):
+            await waiter.wait()
+            return 10
+
+        async def run_bounded_connection():
+            async with self.executor.bounded_connection as connection:
+                return connection
+
+        async def run_race_condition():
+            first_connection = asyncio.ensure_future(run_bounded_connection())
+            await asyncio.sleep(0.1)  # give some time to hit the waiter
+            self.assertIsNotNone(self.executor._ssh_connection)
+            self.assertIsNone(self.executor._session_bound)
+            second_connection = asyncio.ensure_future(run_bounded_connection())
+            await asyncio.sleep(0.1)  # give some time to schedule the second tasks
+            waiter.set()
+            # check that no new connection is established
+            self.assertEqual(await first_connection, await second_connection)
+
+        # monkey patch prob session
+        with patch(
+            "tardis.utilities.executors.sshexecutor.probe_max_session",
+            mocked_probe_max_session,
+        ):
+            run_async(run_race_condition)
+
     def test_lock(self):
         self.assertIsInstance(self.executor.lock, asyncio.Lock)