Merge pull request #657 from ably/stop-retrying-failures

ci: some improvements to CI stability and warnings

Author: owenpearson

.github/workflows/check.yml

@@ -46,4 +46,4 @@ jobs:
       - name: Generate rest sync code and tests
         run: uv run unasync
       - name: Test with pytest
-        run: uv run pytest --verbose --tb=short --reruns 3
+        run: uv run pytest --verbose --tb=short --capture=no

ably/realtime/realtimepresence.py

@@ -468,10 +468,6 @@ async def subscribe(self, *args) -> None:
         Raises:
             AblyException: If channel state prevents subscription
         """
-        # RTP6d: Implicitly attach
-        if self.channel.state in [ChannelState.INITIALIZED, ChannelState.DETACHED, ChannelState.DETACHING]:
-            asyncio.create_task(self.channel.attach())
-
         # Parse arguments: similar to channel subscribe
         if len(args) == 1:
             # subscribe(listener)
@@ -485,6 +481,10 @@ async def subscribe(self, *args) -> None:
         else:
             raise ValueError('Invalid subscribe arguments')
 
+        # RTP6d: Implicitly attach
+        if self.channel.state in [ChannelState.INITIALIZED, ChannelState.DETACHED, ChannelState.DETACHING]:
+            await self.channel.attach()
+
     def unsubscribe(self, *args) -> None:
         """
         Unsubscribe from presence events on this channel (RTP7).

ably/transport/websockettransport.py

@@ -203,7 +203,9 @@ async def ws_read_loop(self):
                     log.exception(
                         f"WebSocketTransport.decode(): Unexpected exception handling channel message: {e}"
                     )
-        except ConnectionClosedOK:
+        except (ConnectionClosedOK, GeneratorExit):
+            # ConnectionClosedOK: normal websocket closure
+            # GeneratorExit: coroutine being closed (e.g., during event loop shutdown)
             return
 
     def decode_raw_websocket_frame(self, raw: str | bytes) -> dict:
@@ -229,18 +231,38 @@ def on_read_loop_done(self, task: asyncio.Task):
 
     async def dispose(self):
         self.is_disposed = True
+
+        # Cancel tasks but don't await them yet to avoid deadlock
+        tasks_to_await = []
+
         if self.read_loop:
             self.read_loop.cancel()
+            tasks_to_await.append(self.read_loop)
         if self.ws_connect_task:
             self.ws_connect_task.cancel()
+            tasks_to_await.append(self.ws_connect_task)
         if self.idle_timer:
             self.idle_timer.cancel()
+
+        # Schedule cleanup of cancelled tasks in the background to avoid blocking dispose()
+        # This prevents deadlock when dispose() is called from within these tasks
+        if tasks_to_await:
+            asyncio.create_task(self._cleanup_tasks(tasks_to_await))
+
         if self.websocket:
             try:
                 await self.websocket.close()
             except asyncio.CancelledError:
                 return
 
+    async def _cleanup_tasks(self, tasks):
+        """Wait for cancelled tasks to complete their cleanup."""
+        for task in tasks:
+            try:
+                await task
+            except Exception:
+                pass  # Ignore all exceptions from cancelled/failed tasks
+
     async def close(self):
         await self.send({'action': ProtocolMessageAction.CLOSE})
 

pyproject.toml

@@ -52,8 +52,6 @@ dev = [
   "respx>=0.22.0,<0.23.0; python_version>='3.8'",
   "importlib-metadata>=4.12,<5.0",
   "pytest-timeout>=2.1.0,<3.0.0",
-  "pytest-rerunfailures>=13.0,<14.0; python_version=='3.7'",
-  "pytest-rerunfailures>=14.0,<15.0; python_version>='3.8'",
   "async-case>=10.1.0,<11.0.0; python_version=='3.7'",
   "tokenize_rt",
   "vcdiff-decoder>=0.1.0a1",

test/ably/realtime/realtimepresence_test.py

@@ -342,7 +342,8 @@ async def test_presence_message_action(self):
         received = asyncio.Future()
 
         def on_presence(msg):
-            received.set_result(msg)
+            if msg.action == PresenceAction.ENTER:
+                received.set_result(msg)
 
         await channel1.presence.subscribe(on_presence)
         await channel1.presence.enter()