Recover stalled database connections and skip unreachable chat recipients

tests/test_chat_delivery.py

@@ -2,6 +2,7 @@ import asyncio
 from datetime import datetime, timezone
 from types import SimpleNamespace
 from unittest.mock import AsyncMock
+import pytest
 
 from aiogram import Bot
 from aiogram.exceptions import TelegramRetryAfter
@@ -95,3 +96,27 @@ async def test_inactivity_job_preserves_existing_delivery_block_and_continues():
         assert await ActivityService.mark_inactive_users(session) == 1
         assert await ActivityService.mark_inactive_users(session) == 0
         assert await session.scalar(select(BlockedUser.error_type).where(BlockedUser.telegram_id == 201)) == "blocked_bot"
+
+
+@pytest.mark.parametrize("error_kind", ["forbidden", "chat_not_found", "invalid_content"])
+async def test_chat_skips_unreachable_recipients_and_restores_them_on_incoming_activity(monkeypatch, error_kind):
+    from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
+    from aiogram.methods import SendMessage
+    from src.core.activity_service import ActivityService
+    from src.handlers import chat_handlers
+
+    async with async_session_maker() as session:
+        await UserService.get_or_create_user(session, 301)
+        await UserService.get_or_create_user(session, 302)
+    error = (TelegramForbiddenError if error_kind == "forbidden" else TelegramBadRequest)(
+        method=SendMessage(chat_id=301, text="test"),
+        message={"forbidden": "bot was blocked by the user", "chat_not_found": "chat not found",
+                 "invalid_content": "CUSTOM_EMOJI_INVALID"}[error_kind],
+    )
+    monkeypatch.setattr(chat_handlers, "copy_preserving_entities", AsyncMock(side_effect=error))
+    assert await chat_handlers._copy_with_sender(SimpleNamespace(), 301, "Sender") is None
+    async with async_session_maker() as session:
+        recipients = {user.telegram_id for user in await chat_handlers.get_all_active_users(session)}
+        assert recipients == ({301, 302} if error_kind == "invalid_content" else {302})
+        await ActivityService.reactivate_user(session, 301)
+        assert {user.telegram_id for user in await chat_handlers.get_all_active_users(session)} == {301, 302}

tests/test_health.py

@@ -0,0 +1,31 @@
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+
+def test_watchdog_exits_a_process_with_stalled_event_loop(tmp_path):
+    # A blocked event loop cannot run coroutine-based timeout handlers.
+    code = """import time
+from src.core.health import start_watchdog
+start_watchdog(grace_seconds=0.1, interval=0.02)
+time.sleep(10)
+"""
+    result = subprocess.run([sys.executable, "-c", code], cwd=Path(__file__).resolve().parents[1],
+                            env=dict(os.environ, HEARTBEAT_FILE=str(tmp_path / "absent")),
+                            capture_output=True, timeout=8)
+    assert result.returncode == 1
+    assert b"Heartbeat expired" in result.stderr
+
+
+def test_stopped_watchdog_allows_normal_shutdown(tmp_path):
+    code = """import time
+from src.core.health import start_watchdog
+stopped = start_watchdog(grace_seconds=0.1, interval=0.02)
+stopped.set()
+time.sleep(0.2)
+"""
+    result = subprocess.run([sys.executable, "-c", code], cwd=Path(__file__).resolve().parents[1],
+                            env=dict(os.environ, HEARTBEAT_FILE=str(tmp_path / "absent")),
+                            capture_output=True, timeout=8)
+    assert result.returncode == 0