Feat/issue 148 distributed lock cleanup

app/services/cleanup_service.py

@@ -9,6 +9,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.config import get_settings
+from app.core.infrastructure.redis import get_redis_client
 from app.db.session import get_session_factory
 from app.repositories.email_verification_repository import (
     EmailVerificationTokenRepository,
@@ -62,9 +63,38 @@ async def _run_cleanup_loop(interval_minutes: int) -> None:
     try:
         while True:
             try:
-                session_factory = get_session_factory()
-                async with session_factory() as session:
-                    await cleanup_expired_tokens(session)
+                redis_client = get_redis_client()
+                if redis_client is not None:
+                    lock = redis_client.lock('lock:token_cleanup', timeout=300)
+                    acquired = False
+                    try:
+                        acquired = await lock.acquire(blocking=False)
+                    except Exception as e:  # noqa: BLE001
+                        logger.warning(
+                            'Failed to acquire token cleanup lock: %s', e
+                        )
+
+                    if not acquired:
+                        logger.debug(
+                            'Token cleanup lock already held by another instance; skipping iteration'
+                        )
+                    else:
+                        try:
+                            session_factory = get_session_factory()
+                            async with session_factory() as session:
+                                await cleanup_expired_tokens(session)
+                        finally:
+                            try:
+                                await lock.release()
+                            except Exception as lock_err:  # noqa: BLE001
+                                logger.warning(
+                                    'Failed to release token cleanup lock: %s',
+                                    lock_err,
+                                )
+                else:
+                    session_factory = get_session_factory()
+                    async with session_factory() as session:
+                        await cleanup_expired_tokens(session)
             except Exception as e:  # noqa: BLE001
                 logger.warning('Token cleanup loop execution failed: %s', e)
             await asyncio.sleep(interval_seconds)

tests/test_cleanup_service.py

@@ -119,34 +119,41 @@ async def _run_cleanup(factory):
 
 @pytest.mark.asyncio
 async def test_start_and_stop_token_cleanup_loop() -> None:
-    """Test start_token_cleanup_loop and stop_token_cleanup_loop functions."""
-    from unittest.mock import AsyncMock, patch
+    """Test start_token_cleanup_loop and stop_token_cleanup_loop when Redis is not available."""
+    from unittest.mock import AsyncMock, MagicMock, patch
 
     from app.services.cleanup_service import (
         start_token_cleanup_loop,
         stop_token_cleanup_loop,
     )
 
     mock_session = AsyncMock()
-    mock_factory = AsyncMock()
-    mock_factory.return_value.__aenter__.return_value = mock_session
+    mock_factory = MagicMock()
+    mock_factory.return_value.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_factory.return_value.__aexit__ = AsyncMock(return_value=None)
 
     with (
         patch('app.services.cleanup_service.get_settings') as mock_settings,
+        patch(
+            'app.services.cleanup_service.get_redis_client',
+            return_value=None,
+        ),
         patch(
             'app.services.cleanup_service.get_session_factory',
             return_value=mock_factory,
         ),
         patch(
             'app.services.cleanup_service.cleanup_expired_tokens',
             new_callable=AsyncMock,
-        ),
+        ) as mock_cleanup,
     ):
         mock_settings.return_value.TOKEN_CLEANUP_INTERVAL_MINUTES = 60
         await start_token_cleanup_loop()
         await asyncio.sleep(0.05)
         await stop_token_cleanup_loop()
 
+    assert mock_cleanup.called
+
 
 @pytest.mark.asyncio
 async def test_start_token_cleanup_loop_disabled() -> None:
@@ -158,3 +165,95 @@ async def test_start_token_cleanup_loop_disabled() -> None:
     with patch('app.services.cleanup_service.get_settings') as mock_settings:
         mock_settings.return_value.TOKEN_CLEANUP_INTERVAL_MINUTES = 0
         await start_token_cleanup_loop()
+
+
+@pytest.mark.asyncio
+async def test_token_cleanup_loop_with_redis_lock_acquired() -> None:
+    """Test token cleanup loop when Redis lock is successfully acquired."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from app.services.cleanup_service import (
+        start_token_cleanup_loop,
+        stop_token_cleanup_loop,
+    )
+
+    mock_redis = MagicMock()
+    mock_lock = MagicMock()
+    mock_lock.acquire = AsyncMock(return_value=True)
+    mock_lock.release = AsyncMock()
+    mock_redis.lock.return_value = mock_lock
+
+    mock_session = AsyncMock()
+    mock_factory = MagicMock()
+    mock_factory.return_value.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_factory.return_value.__aexit__ = AsyncMock(return_value=None)
+
+    with (
+        patch('app.services.cleanup_service.get_settings') as mock_settings,
+        patch(
+            'app.services.cleanup_service.get_redis_client',
+            return_value=mock_redis,
+        ),
+        patch(
+            'app.services.cleanup_service.get_session_factory',
+            return_value=mock_factory,
+        ),
+        patch(
+            'app.services.cleanup_service.cleanup_expired_tokens',
+            new_callable=AsyncMock,
+        ) as mock_cleanup,
+    ):
+        mock_settings.return_value.TOKEN_CLEANUP_INTERVAL_MINUTES = 60
+        await start_token_cleanup_loop()
+        await asyncio.sleep(0.05)
+        await stop_token_cleanup_loop()
+
+    mock_redis.lock.assert_called_once_with('lock:token_cleanup', timeout=300)
+    mock_lock.acquire.assert_called_once_with(blocking=False)
+    mock_cleanup.assert_called_once()
+    mock_lock.release.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_token_cleanup_loop_with_redis_lock_not_acquired() -> None:
+    """Test token cleanup loop when Redis lock cannot be acquired by replica."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from app.services.cleanup_service import (
+        start_token_cleanup_loop,
+        stop_token_cleanup_loop,
+    )
+
+    mock_redis = MagicMock()
+    mock_lock = MagicMock()
+    mock_lock.acquire = AsyncMock(return_value=False)
+    mock_lock.release = AsyncMock()
+    mock_redis.lock.return_value = mock_lock
+
+    mock_factory = MagicMock()
+
+    with (
+        patch('app.services.cleanup_service.get_settings') as mock_settings,
+        patch(
+            'app.services.cleanup_service.get_redis_client',
+            return_value=mock_redis,
+        ),
+        patch(
+            'app.services.cleanup_service.get_session_factory',
+            return_value=mock_factory,
+        ),
+        patch(
+            'app.services.cleanup_service.cleanup_expired_tokens',
+            new_callable=AsyncMock,
+        ) as mock_cleanup,
+    ):
+        mock_settings.return_value.TOKEN_CLEANUP_INTERVAL_MINUTES = 60
+        await start_token_cleanup_loop()
+        await asyncio.sleep(0.05)
+        await stop_token_cleanup_loop()
+
+    mock_redis.lock.assert_called_once_with('lock:token_cleanup', timeout=300)
+    mock_lock.acquire.assert_called_once_with(blocking=False)
+    mock_cleanup.assert_not_called()
+    mock_factory.assert_not_called()
+    mock_lock.release.assert_not_called()