From 044cb4de7e35b39096af333aaeb704c4e912fc9f Mon Sep 17 00:00:00 2001 From: 3252a8 <3252a8@proton.me> Date: Thu, 21 May 2026 06:26:48 +0300 Subject: [PATCH] refactor(redis): strengthen shared cache invalidation --- backend/bot/app/web/webapp/account.py | 6 ++- backend/bot/app/web/webapp/assets.py | 43 ++++++++++--------- backend/bot/app/web/webapp/auth.py | 5 +++ backend/bot/app/web/webapp/common.py | 24 +++++++++++ backend/bot/infra/redis.py | 35 +++++++++++++-- backend/bot/infra/webhook_queue.py | 44 ++++++++++++------- backend/bot/services/panel_api_service.py | 52 +++++++++++++---------- backend/bot/utils/ttl_cache.py | 17 ++++++++ tests/test_performance_caches.py | 34 +++++++++++++++ tests/test_webapp_redis_cache.py | 38 +++++++++++++++++ 10 files changed, 237 insertions(+), 61 deletions(-) create mode 100644 tests/test_webapp_redis_cache.py diff --git a/backend/bot/app/web/webapp/account.py b/backend/bot/app/web/webapp/account.py index 5a05e2c..826829e 100644 --- a/backend/bot/app/web/webapp/account.py +++ b/backend/bot/app/web/webapp/account.py @@ -1,6 +1,7 @@ # ruff: noqa: F401,F403,F405,I001 from ._runtime import * # noqa: F403,F405 from .auth import _hash_email_password +from .common import _invalidate_webapp_user_caches async def account_email_request_route(request: web.Request) -> web.Response: @@ -174,6 +175,7 @@ async def account_email_verify_route(request: web.Request) -> web.Response: logger.exception("Email account link failed") return _json_error(500, "link_failed", "Link failed") + await _invalidate_webapp_user_caches(settings, user_id, final_user_id, include_devices=True) if should_notify_email_linked: try: from bot.services.notification_service import NotificationService @@ -276,7 +278,7 @@ async def account_password_confirm_route(request: web.Request) -> web.Response: logger.exception("Email password setup failed") return _json_error(500, "password_setup_failed", "Password setup failed") - await cache_delete(settings, redis_key(settings, "cache", "webapp", "me", user_id)) + await _invalidate_webapp_user_caches(settings, user_id) return web.json_response({"ok": True, "password_auth_enabled": True}) @@ -396,6 +398,7 @@ async def account_telegram_link_route(request: web.Request) -> web.Response: logger.exception("Telegram account link failed") return _json_error(500, "link_failed", "Link failed") + await _invalidate_webapp_user_caches(settings, user_id, final_user_id, include_devices=True) if should_notify_telegram_linked and final_telegram_id: try: from bot.services.notification_service import NotificationService @@ -488,6 +491,7 @@ async def account_language_route(request: web.Request) -> web.Response: await session.flush() await session.commit() + await _invalidate_webapp_user_caches(settings, user_id) return web.json_response({"ok": True, "language": language}) diff --git a/backend/bot/app/web/webapp/assets.py b/backend/bot/app/web/webapp/assets.py index 75bfab0..8cd84f0 100644 --- a/backend/bot/app/web/webapp/assets.py +++ b/backend/bot/app/web/webapp/assets.py @@ -820,25 +820,30 @@ async def _enforce_webapp_rate_limit( or "unknown" ) key = f"{action}:{ip_address}:{int(user_id)}" - redis = await get_redis(settings) - if redis is not None: - redis_rate_key = redis_key(settings, "rate-limit", "webapp", key) - current = await redis.incr(redis_rate_key) - if current == 1: - await redis.expire(redis_rate_key, settings.WEBAPP_RATE_LIMIT_TTL_SECONDS) - if current > settings.WEBAPP_RATE_LIMIT_MAX_REQUESTS: - ttl = await redis.ttl(redis_rate_key) - retry_after = max(1, int(ttl if ttl and ttl > 0 else WEBAPP_RATE_LIMIT_WINDOW_SECONDS)) - return web.json_response( - { - "ok": False, - "error": "rate_limited", - "retry_after": retry_after, - }, - status=429, - headers={"Retry-After": str(retry_after)}, - ) - return None + try: + redis = await get_redis(settings) + if redis is not None: + redis_rate_key = redis_key(settings, "rate-limit", "webapp", key) + current = await redis.incr(redis_rate_key) + if current == 1: + await redis.expire(redis_rate_key, settings.WEBAPP_RATE_LIMIT_TTL_SECONDS) + if current > settings.WEBAPP_RATE_LIMIT_MAX_REQUESTS: + ttl = await redis.ttl(redis_rate_key) + retry_after = max( + 1, int(ttl if ttl and ttl > 0 else WEBAPP_RATE_LIMIT_WINDOW_SECONDS) + ) + return web.json_response( + { + "ok": False, + "error": "rate_limited", + "retry_after": retry_after, + }, + status=429, + headers={"Retry-After": str(retry_after)}, + ) + return None + except Exception as exc: + logger.warning("Redis webapp rate limiter unavailable; using local fallback: %s", exc) buckets: Dict[str, deque[float]] = request.app["webapp_rate_limit_buckets"] lock: asyncio.Lock = request.app["webapp_rate_limit_lock"] diff --git a/backend/bot/app/web/webapp/auth.py b/backend/bot/app/web/webapp/auth.py index 5c1195f..bceffc6 100644 --- a/backend/bot/app/web/webapp/auth.py +++ b/backend/bot/app/web/webapp/auth.py @@ -1,5 +1,6 @@ # ruff: noqa: F401,F403,F405,I001 from ._runtime import * # noqa: F403,F405 +from .common import _invalidate_webapp_user_caches def _resolve_telegram_bot_id(bot_token: str) -> Optional[int]: @@ -386,6 +387,7 @@ async def telegram_oauth_callback_route(request: web.Request) -> web.Response: logger.exception("Telegram OAuth callback failed") raise redirect(redirect_path, "failed") + await _invalidate_webapp_user_caches(settings, final_user_id, include_devices=True) token = create_webapp_session_token(settings, int(final_user_id)) response = web.HTTPFound(_telegram_oauth_redirect_url(redirect_path, status="success")) _clear_telegram_oauth_state_cookie(response) @@ -480,6 +482,7 @@ async def auth_token_route(request: web.Request) -> web.Response: logger.exception("WebApp auth failed") return _json_error(500, "auth_failed", "Auth failed") + await _invalidate_webapp_user_caches(settings, authenticated_user_id, include_devices=True) token = create_webapp_session_token(settings, int(authenticated_user_id)) return _build_webapp_auth_response(settings, {"ok": True}, token=token) @@ -694,6 +697,7 @@ async def email_auth_verify_route(request: web.Request) -> web.Response: logger.exception("Email WebApp auth failed") return _json_error(500, "auth_failed", "Auth failed") + await _invalidate_webapp_user_caches(settings, int(db_user.user_id), include_devices=True) if created_user: try: from bot.services.notification_service import NotificationService @@ -801,6 +805,7 @@ async def email_auth_magic_route(request: web.Request) -> web.Response: logger.exception("Email magic-link auth failed") return _json_error(500, "auth_failed", "Auth failed") + await _invalidate_webapp_user_caches(settings, int(db_user.user_id), include_devices=True) if created_user and verified_email: try: from bot.services.notification_service import NotificationService diff --git a/backend/bot/app/web/webapp/common.py b/backend/bot/app/web/webapp/common.py index d6c32e3..459724a 100644 --- a/backend/bot/app/web/webapp/common.py +++ b/backend/bot/app/web/webapp/common.py @@ -17,6 +17,30 @@ def _json_error(status: int, code: str, message: str) -> web.Response: ) +async def _invalidate_webapp_user_caches( + settings: Settings, + *user_ids: Optional[int], + include_devices: bool = False, +) -> None: + keys: List[str] = [] + seen: set[int] = set() + for raw_user_id in user_ids: + if raw_user_id is None: + continue + try: + user_id = int(raw_user_id) + except (TypeError, ValueError): + continue + if user_id in seen: + continue + seen.add(user_id) + keys.append(redis_key(settings, "cache", "webapp", "me", user_id)) + if include_devices: + keys.append(redis_key(settings, "cache", "webapp", "devices", user_id)) + if keys: + await cache_delete(settings, *keys) + + def _validation_error_response(exc: ValidationError) -> web.Response: for error in exc.errors(): loc = error.get("loc") or () diff --git a/backend/bot/infra/redis.py b/backend/bot/infra/redis.py index 7703bfc..5d093de 100644 --- a/backend/bot/infra/redis.py +++ b/backend/bot/infra/redis.py @@ -51,7 +51,11 @@ async def cache_get_json(settings: Settings, key: str) -> Any: redis = await get_redis(settings) if redis is None: return None - raw = await redis.get(key) + try: + raw = await redis.get(key) + except Exception as exc: + logger.warning("Redis cache get failed for key %s: %s", key, exc) + return None if raw is None: return None try: @@ -65,14 +69,39 @@ async def cache_set_json(settings: Settings, key: str, value: Any, ttl_seconds: redis = await get_redis(settings) if redis is None: return - await redis.set(key, json.dumps(value, ensure_ascii=False, default=str), ex=ttl_seconds) + try: + await redis.set(key, json.dumps(value, ensure_ascii=False, default=str), ex=ttl_seconds) + except Exception as exc: + logger.warning("Redis cache set failed for key %s: %s", key, exc) async def cache_delete(settings: Settings, *keys: str) -> None: redis = await get_redis(settings) if redis is None or not keys: return - await redis.delete(*keys) + try: + await redis.delete(*keys) + except Exception as exc: + logger.warning("Redis cache delete failed for %s key(s): %s", len(keys), exc) + + +async def cache_delete_pattern(settings: Settings, pattern: str) -> int: + redis = await get_redis(settings) + if redis is None or not pattern: + return 0 + deleted = 0 + batch = [] + try: + async for key in redis.scan_iter(match=pattern, count=100): + batch.append(key) + if len(batch) >= 100: + deleted += int(await redis.delete(*batch)) + batch.clear() + if batch: + deleted += int(await redis.delete(*batch)) + except Exception as exc: + logger.warning("Redis cache pattern delete failed for %s: %s", pattern, exc) + return deleted @asynccontextmanager diff --git a/backend/bot/infra/webhook_queue.py b/backend/bot/infra/webhook_queue.py index 5d794be..646eb24 100644 --- a/backend/bot/infra/webhook_queue.py +++ b/backend/bot/infra/webhook_queue.py @@ -24,28 +24,36 @@ async def enqueue_webhook_event( if redis is None: return False - dedupe_id = event_id or payload.get("id") or payload.get("event_id") - if dedupe_id: - dedupe_key = redis_key(settings, "webhook", "seen", provider, dedupe_id) - if not await redis.set(dedupe_key, "1", nx=True, ex=24 * 60 * 60): - logger.info("Skipping duplicate %s webhook event %s", provider, dedupe_id) - return True + try: + dedupe_id = event_id or payload.get("id") or payload.get("event_id") + if dedupe_id: + dedupe_key = redis_key(settings, "webhook", "seen", provider, dedupe_id) + if not await redis.set(dedupe_key, "1", nx=True, ex=24 * 60 * 60): + logger.info("Skipping duplicate %s webhook event %s", provider, dedupe_id) + return True - message = { - "provider": provider, - "event_id": dedupe_id, - "payload": payload, - "enqueued_at": time.time(), - } - await redis.lpush(webhook_queue_key(settings), json.dumps(message, ensure_ascii=False)) - return True + message = { + "provider": provider, + "event_id": dedupe_id, + "payload": payload, + "enqueued_at": time.time(), + } + await redis.lpush(webhook_queue_key(settings), json.dumps(message, ensure_ascii=False)) + return True + except Exception as exc: + logger.warning("Redis webhook enqueue failed for %s: %s", provider, exc) + return False async def pop_webhook_event(settings: Settings, timeout_seconds: int = 5) -> Optional[dict]: redis = await get_redis(settings) if redis is None: return None - item = await redis.brpop(webhook_queue_key(settings), timeout=timeout_seconds) + try: + item = await redis.brpop(webhook_queue_key(settings), timeout=timeout_seconds) + except Exception as exc: + logger.warning("Redis webhook pop failed: %s", exc) + return None if not item: return None _, raw = item @@ -60,4 +68,8 @@ async def webhook_queue_depth(settings: Settings) -> int: redis = await get_redis(settings) if redis is None: return 0 - return int(await redis.llen(webhook_queue_key(settings))) + try: + return int(await redis.llen(webhook_queue_key(settings))) + except Exception as exc: + logger.warning("Redis webhook queue depth failed: %s", exc) + return 0 diff --git a/backend/bot/services/panel_api_service.py b/backend/bot/services/panel_api_service.py index 0f12cc9..0b1a89a 100644 --- a/backend/bot/services/panel_api_service.py +++ b/backend/bot/services/panel_api_service.py @@ -496,7 +496,7 @@ class PanelApiService: "POST", "/users", json=payload, log_full_response=log_response ) if response and not response.get("error") and "response" in response: - self._invalidate_all_users_cache() + await self._invalidate_all_users_cache() logging.info( f"Panel user '{username_on_panel}' created successfully (UUID: {response.get('response', {}).get('uuid')})." # noqa: E501 ) @@ -518,8 +518,8 @@ class PanelApiService: ) if full_response and not full_response.get("error") and "response" in full_response: logging.debug("User %s details updated on panel.", user_uuid) - self._invalidate_user_cache(user_uuid) - self._invalidate_all_users_cache() + await self._invalidate_user_cache(user_uuid) + await self._invalidate_all_users_cache() return full_response.get("response") logging.error( @@ -535,8 +535,8 @@ class PanelApiService: response_data = await self._request("POST", endpoint, log_full_response=log_response) if response_data and not response_data.get("error") and "response" in response_data: - self._invalidate_user_cache(user_uuid) - self._invalidate_all_users_cache() + await self._invalidate_user_cache(user_uuid) + await self._invalidate_all_users_cache() actual_status = response_data.get("response", {}).get("status") expected_status = "ACTIVE" if enable else "DISABLED" if actual_status == expected_status: @@ -573,17 +573,17 @@ class PanelApiService: logging.info( f"Panel user {user_uuid} already absent (errorCode {error_code}). Treating as deleted." # noqa: E501 ) - self._invalidate_user_cache(user_uuid) - self._invalidate_devices_cache(user_uuid) - self._invalidate_all_users_cache() + await self._invalidate_user_cache(user_uuid) + await self._invalidate_devices_cache(user_uuid) + await self._invalidate_all_users_cache() return True logging.error(f"Failed to delete user {user_uuid} on panel. Response: {response_data}") return False logging.info(f"Panel user {user_uuid} deleted successfully.") - self._invalidate_user_cache(user_uuid) - self._invalidate_devices_cache(user_uuid) - self._invalidate_all_users_cache() + await self._invalidate_user_cache(user_uuid) + await self._invalidate_devices_cache(user_uuid) + await self._invalidate_all_users_cache() return True async def get_subscription_link( @@ -618,7 +618,7 @@ class PanelApiService: payload = {"userUuid": user_uuid, "hwid": hwid} response_data = await self._request("POST", endpoint, json=payload, log_full_response=False) if response_data and not response_data.get("error") and "response" in response_data: - self._invalidate_devices_cache(user_uuid) + await self._invalidate_devices_cache(user_uuid) return True logging.error( f"Failed to disconnect device {hwid} for user {user_uuid}. Payload: {payload}, Response: {response_data}" # noqa: E501 @@ -720,21 +720,21 @@ class PanelApiService: ) return None - def _invalidate_squad_caches(self) -> None: - self._squads_cache.invalidate() + async def _invalidate_squad_caches(self) -> None: + await self._squads_cache.invalidate_remote() - def _invalidate_user_cache(self, user_uuid: Optional[str]) -> None: + async def _invalidate_user_cache(self, user_uuid: Optional[str]) -> None: if not user_uuid: return - self._users_cache.invalidate(f"uuid:{user_uuid}") + await self._users_cache.invalidate_remote(f"uuid:{user_uuid}") - def _invalidate_all_users_cache(self) -> None: - self._all_users_cache.invalidate() + async def _invalidate_all_users_cache(self) -> None: + await self._all_users_cache.invalidate_remote() - def _invalidate_devices_cache(self, user_uuid: Optional[str]) -> None: + async def _invalidate_devices_cache(self, user_uuid: Optional[str]) -> None: if not user_uuid: return - self._devices_cache.invalidate(f"user:{user_uuid}") + await self._devices_cache.invalidate_remote(f"user:{user_uuid}") async def get_internal_squads(self) -> Optional[List[Dict[str, Any]]]: return await self._squads_cache.get_or_load("list", self._get_internal_squads_uncached) @@ -835,6 +835,8 @@ class PanelApiService: endpoint = f"/users/{user_uuid}/actions/reset-traffic" response_data = await self._request("POST", endpoint, log_full_response=False) if response_data and not response_data.get("error"): + await self._invalidate_user_cache(user_uuid) + await self._invalidate_all_users_cache() return True logging.error("Failed to reset traffic for user %s. Response: %s", user_uuid, response_data) return False @@ -848,7 +850,10 @@ class PanelApiService: log_full_response=False, ) if response_data and not response_data.get("error"): - self._invalidate_squad_caches() + await self._invalidate_squad_caches() + for user_uuid in user_uuids: + await self._invalidate_user_cache(user_uuid) + await self._invalidate_all_users_cache() return True logging.error("Failed to add users to squad %s. Response: %s", squad_uuid, response_data) return False @@ -864,7 +869,10 @@ class PanelApiService: log_full_response=False, ) if response_data and not response_data.get("error"): - self._invalidate_squad_caches() + await self._invalidate_squad_caches() + for user_uuid in user_uuids: + await self._invalidate_user_cache(user_uuid) + await self._invalidate_all_users_cache() return True logging.error( "Failed to remove users from squad %s. Response: %s", squad_uuid, response_data diff --git a/backend/bot/utils/ttl_cache.py b/backend/bot/utils/ttl_cache.py index 6ba79db..278e492 100644 --- a/backend/bot/utils/ttl_cache.py +++ b/backend/bot/utils/ttl_cache.py @@ -83,3 +83,20 @@ class AsyncTTLCache: self._data.clear() return self._data.pop(key, None) + + async def invalidate_remote(self, key: Optional[str] = None) -> None: + self.invalidate(key) + if self.settings is None or not self.namespace: + return + try: + from bot.infra.redis import cache_delete, cache_delete_pattern, redis_key + + if key is None: + pattern = redis_key(self.settings, "cache", self.namespace, "*") + await cache_delete_pattern(self.settings, pattern) + return + await cache_delete( + self.settings, redis_key(self.settings, "cache", self.namespace, key) + ) + except Exception: + return diff --git a/tests/test_performance_caches.py b/tests/test_performance_caches.py index 203780b..6093678 100644 --- a/tests/test_performance_caches.py +++ b/tests/test_performance_caches.py @@ -1,4 +1,5 @@ import asyncio +import time import unittest from types import SimpleNamespace from unittest.mock import patch @@ -40,6 +41,39 @@ class AsyncTTLCacheSingleflightTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(set_calls, 1) +class AsyncTTLCacheInvalidationTests(unittest.IsolatedAsyncioTestCase): + async def test_invalidate_remote_deletes_single_redis_key(self): + settings = SimpleNamespace(REDIS_URL="redis://example", REDIS_KEY_PREFIX="test") + cache = AsyncTTLCache(ttl_seconds=60, settings=settings, namespace="bench") + cache._data["same"] = (time.monotonic() + 60, {"ok": True}) + deleted = [] + + async def fake_delete(_settings, *keys): + deleted.extend(keys) + + with patch("bot.infra.redis.cache_delete", new=fake_delete): + await cache.invalidate_remote("same") + + self.assertIsNone(cache.get_fresh("same")) + self.assertEqual(deleted, ["test:cache:bench:same"]) + + async def test_invalidate_remote_deletes_namespace_pattern(self): + settings = SimpleNamespace(REDIS_URL="redis://example", REDIS_KEY_PREFIX="test") + cache = AsyncTTLCache(ttl_seconds=60, settings=settings, namespace="bench") + cache._data["same"] = (time.monotonic() + 60, {"ok": True}) + patterns = [] + + async def fake_delete_pattern(_settings, pattern): + patterns.append(pattern) + return 1 + + with patch("bot.infra.redis.cache_delete_pattern", new=fake_delete_pattern): + await cache.invalidate_remote() + + self.assertIsNone(cache.get_fresh("same")) + self.assertEqual(patterns, ["test:cache:bench:*"]) + + class Crypt4LinkCacheTests(unittest.IsolatedAsyncioTestCase): async def asyncTearDown(self): config_link._CRYPT4_LINK_CACHES.clear() diff --git a/tests/test_webapp_redis_cache.py b/tests/test_webapp_redis_cache.py new file mode 100644 index 0000000..711dc4b --- /dev/null +++ b/tests/test_webapp_redis_cache.py @@ -0,0 +1,38 @@ +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +import bot.app.web.subscription_webapp # noqa: F401 +from bot.app.web.webapp import common as common_module + + +class WebappRedisCacheInvalidationTests(unittest.IsolatedAsyncioTestCase): + async def test_invalidate_webapp_user_caches_deletes_me_and_devices_keys(self): + settings = SimpleNamespace(REDIS_URL="redis://redis:6379/0", REDIS_KEY_PREFIX="shop") + deleted = [] + + async def fake_delete(_settings, *keys): + deleted.extend(keys) + + with patch.object(common_module, "cache_delete", fake_delete): + await common_module._invalidate_webapp_user_caches( + settings, + 42, + "42", + 99, + include_devices=True, + ) + + self.assertEqual( + deleted, + [ + "shop:cache:webapp:me:42", + "shop:cache:webapp:devices:42", + "shop:cache:webapp:me:99", + "shop:cache:webapp:devices:99", + ], + ) + + +if __name__ == "__main__": + unittest.main()