diff --git a/app/services/nalogo_service.py b/app/services/nalogo_service.py index 8a182ba6..705913a4 100644 --- a/app/services/nalogo_service.py +++ b/app/services/nalogo_service.py @@ -99,10 +99,11 @@ class NaloGoService: ) return False - # Проверяем не в очереди ли уже + # Атомарная проверка и установка флага "в очереди" (защита от race condition) queued_key = f"nalogo:queued:{payment_id}" - already_queued = await cache.get(queued_key) - if already_queued: + lock_acquired = await cache.setnx(queued_key, "queued", expire=7 * 24 * 3600) + if not lock_acquired: + # Ключ уже существует — чек уже в очереди logger.info( f"Чек для payment_id={payment_id} уже в очереди, пропускаем дубликат" ) @@ -121,16 +122,16 @@ class NaloGoService: } success = await cache.lpush(NALOGO_QUEUE_KEY, receipt_data) if success: - # Помечаем что чек в очереди (TTL 7 дней) - if payment_id: - queued_key = f"nalogo:queued:{payment_id}" - await cache.set(queued_key, "queued", expire=7 * 24 * 3600) - queue_len = await cache.llen(NALOGO_QUEUE_KEY) logger.info( f"Чек добавлен в очередь (payment_id={payment_id}, " f"сумма={amount}₽, в очереди: {queue_len})" ) + else: + # Если не удалось добавить в очередь — удаляем флаг + if payment_id: + queued_key = f"nalogo:queued:{payment_id}" + await cache.delete(queued_key) return success async def authenticate(self) -> bool: diff --git a/app/utils/cache.py b/app/utils/cache.py index 5bbc916f..1f3c4209 100644 --- a/app/utils/cache.py +++ b/app/utils/cache.py @@ -64,6 +64,33 @@ class CacheService: logger.error(f"Ошибка записи в кеш {key}: {e}") return False + async def setnx( + self, + key: str, + value: Any, + expire: Union[int, timedelta] = None + ) -> bool: + """Атомарная операция SET IF NOT EXISTS. + + Устанавливает значение только если ключ не существует. + Возвращает True если значение было установлено, False если ключ уже существовал. + """ + if not self._connected: + return False + + try: + serialized_value = json.dumps(value, default=str) + + if isinstance(expire, timedelta): + expire = int(expire.total_seconds()) + + # SET с NX возвращает True если установлено, None если ключ существует + result = await self.redis_client.set(key, serialized_value, ex=expire, nx=True) + return result is True + except Exception as e: + logger.error(f"Ошибка setnx в кеш {key}: {e}") + return False + async def delete(self, key: str) -> bool: if not self._connected: return False