Fixed asyncpg/redis start() while cancelled resulting in an inconsistent cancel scope stack

2 files changed, 9 insertions, 5 deletions

diff --git a/src/apscheduler/eventbrokers/async_redis.py b/src/apscheduler/eventbrokers/async_redis.py
index c2a5711..5e71621 100644
--- a/src/apscheduler/eventbrokers/async_redis.py
+++ b/src/apscheduler/eventbrokers/async_redis.py
@@ -80,7 +80,7 @@ class AsyncRedisEventBroker(LocalAsyncEventBroker, DistributedEventBrokerMixin):
         pubsub = self.client.pubsub()
         try:
             await pubsub.subscribe(self.channel)
-        except Exception:
+        except BaseException:
             await self.stop(force=True)
             raise
 
diff --git a/src/apscheduler/eventbrokers/asyncpg.py b/src/apscheduler/eventbrokers/asyncpg.py
index 4ee2339..27b1bed 100644
--- a/src/apscheduler/eventbrokers/asyncpg.py
+++ b/src/apscheduler/eventbrokers/asyncpg.py
@@ -130,10 +130,14 @@ class AsyncpgEventBroker(LocalAsyncEventBroker, DistributedEventBrokerMixin):
 
     async def start(self) -> None:
         await super().start()
-        self._send = cast(
-            MemoryObjectSendStream[str],
-            await self._task_group.start(self._listen_notifications),
-        )
+        try:
+            self._send = cast(
+                MemoryObjectSendStream[str],
+                await self._task_group.start(self._listen_notifications),
+            )
+        except BaseException:
+            await super().stop(force=True)
+            raise
 
     async def stop(self, *, force: bool = False) -> None:
         self._send.close()