60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.config import settings
|
|
from app.models.base import async_session
|
|
from app.services.module_async_recovery_service import recover_module_async_tasks_once
|
|
from app.services.redis_registry_service import get_registry_redis, redis_acquire_lock, redis_release_lock
|
|
from app.tasks.async_runner import run_async
|
|
from app.tasks.celery_app import celery_app
|
|
|
|
|
|
async def _run_recover_module_async_tasks_once() -> dict[str, Any]:
|
|
redis = await get_registry_redis()
|
|
token: str | None = None
|
|
if redis is not None:
|
|
token = await redis_acquire_lock(
|
|
lock_key=settings.MODULE_ASYNC_RECOVERY_LOCK_KEY,
|
|
ttl_seconds=int(settings.CELERY_RECOVERY_TASK_LOCK_TTL_SECONDS or 600),
|
|
log_context="module_async_recovery",
|
|
)
|
|
if not token:
|
|
return {"skipped": "lock_held", "lock_key": settings.MODULE_ASYNC_RECOVERY_LOCK_KEY}
|
|
|
|
try:
|
|
async with async_session() as db:
|
|
result = await recover_module_async_tasks_once(db)
|
|
result["execution_lock"] = "lock_acquired" if token else "redis_unavailable_run_db_fallback"
|
|
return result
|
|
finally:
|
|
if token:
|
|
await redis_release_lock(
|
|
lock_key=settings.MODULE_ASYNC_RECOVERY_LOCK_KEY,
|
|
token=token,
|
|
log_context="module_async_recovery",
|
|
)
|
|
|
|
|
|
if celery_app:
|
|
|
|
@celery_app.task(
|
|
name="module_async.recover_module_async_tasks_once",
|
|
bind=True,
|
|
soft_time_limit=settings.CELERY_RECOVERY_SOFT_TIME_LIMIT_SECONDS,
|
|
time_limit=settings.CELERY_RECOVERY_TIME_LIMIT_SECONDS,
|
|
)
|
|
def recover_module_async_tasks_once_task(self) -> dict[str, Any]:
|
|
return run_async(_run_recover_module_async_tasks_once())
|
|
|
|
else:
|
|
|
|
class _DisabledTask:
|
|
def delay(self, *args: Any, **kwargs: Any) -> None:
|
|
raise RuntimeError("Celery is disabled")
|
|
|
|
def apply_async(self, *args: Any, **kwargs: Any) -> None:
|
|
raise RuntimeError("Celery is disabled")
|
|
|
|
recover_module_async_tasks_once_task = _DisabledTask()
|