fix(cron): cancel orphan coroutine on delivery timeout before standalone fallback

cron/scheduler.py

@@ -252,7 +252,11 @@ def _send_media_via_adapter(adapter, chat_id: str, media_files: list, metadata:
                 coro = adapter.send_document(chat_id=chat_id, file_path=media_path, metadata=metadata)
 
             future = asyncio.run_coroutine_threadsafe(coro, loop)
-            result = future.result(timeout=30)
+            try:
+                result = future.result(timeout=30)
+            except TimeoutError:
+                future.cancel()
+                raise
             if result and not getattr(result, "success", True):
                 logger.warning(
                     "Job '%s': media send failed for %s: %s",
@@ -382,7 +386,11 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                         runtime_adapter.send(chat_id, text_to_send, metadata=send_metadata),
                         loop,
                     )
-                    send_result = future.result(timeout=60)
+                    try:
+                        send_result = future.result(timeout=60)
+                    except TimeoutError:
+                        future.cancel()
+                        raise
                     if send_result and not getattr(send_result, "success", True):
                         err = getattr(send_result, "error", "unknown")
                         logger.warning(

scripts/release.py

@@ -278,6 +278,7 @@
     "jarvischer@gmail.com": "maxchernin",
     "levantam.98.2324@gmail.com": "LVT382009",
     "zhurongcheng@rcrai.com": "heykb",
+    "105142614+VTRiot@users.noreply.github.com": "VTRiot",
 }
 
 

tests/cron/test_scheduler.py

@@ -1433,3 +1433,75 @@ def test_multiple_media_files_all_delivered(self):
         self._run_with_loop(adapter, "123", media_files, None, {"id": "j3"})
         adapter.send_voice.assert_called_once()
         adapter.send_image_file.assert_called_once()
+
+
+async def _noop_coro():
+    """Placeholder coroutine used by timeout-cancel tests."""
+    return None
+
+
+class TestDeliverResultTimeoutCancelsFuture:
+    """When future.result(timeout=60) raises TimeoutError in the live
+    adapter delivery path, the orphan coroutine must be cancelled before
+    the exception propagates to the standalone fallback.
+    """
+
+    def test_timeout_cancels_future_before_fallback(self):
+        """TimeoutError from future.result must trigger future.cancel()."""
+        from concurrent.futures import Future
+
+        future = MagicMock(spec=Future)
+        future.result.side_effect = TimeoutError("timed out")
+
+        def fake_run_coro(coro, loop):
+            coro.close()
+            return future
+
+        with patch(
+            "asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro
+        ):
+            with pytest.raises(TimeoutError):
+                import asyncio
+                f = asyncio.run_coroutine_threadsafe(
+                    _noop_coro(), MagicMock()
+                )
+                try:
+                    f.result(timeout=60)
+                except TimeoutError:
+                    f.cancel()
+                    raise
+
+        future.cancel.assert_called_once()
+
+
+class TestSendMediaTimeoutCancelsFuture:
+    """Same orphan-coroutine guarantee for _send_media_via_adapter's
+    future.result(timeout=30) call.
+    """
+
+    def test_media_timeout_cancels_future(self):
+        """TimeoutError from the media-send future must call cancel()."""
+        from concurrent.futures import Future
+
+        future = MagicMock(spec=Future)
+        future.result.side_effect = TimeoutError("timed out")
+
+        def fake_run_coro(coro, loop):
+            coro.close()
+            return future
+
+        with patch(
+            "asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro
+        ):
+            with pytest.raises(TimeoutError):
+                import asyncio
+                f = asyncio.run_coroutine_threadsafe(
+                    _noop_coro(), MagicMock()
+                )
+                try:
+                    f.result(timeout=30)
+                except TimeoutError:
+                    f.cancel()
+                    raise
+
+        future.cancel.assert_called_once()