Nexus/server/background/retry_runner.py

"""Nexus — Retry Runner Background Task
Every 5 minutes: check pending PushRetryJobs, retry failed pushes with exponential backoff.
"""

import asyncio
import logging
from datetime import datetime, timezone, timedelta

from server.infrastructure.database.session import AsyncSessionLocal
from server.domain.models import PushRetryJob

logger = logging.getLogger("nexus.retry_runner")

RETRY_CHECK_INTERVAL = 300  # 5 minutes
RETRY_BACKOFF_BASE = 60  # seconds — doubles each retry


def _backoff_seconds(retry_count: int) -> int:
    return RETRY_BACKOFF_BASE * (2 ** min(max(retry_count - 1, 0), 6))


def apply_push_retry_outcome(
    job: PushRetryJob,
    sync_result: dict,
    *,
    now: datetime,
) -> None:
    """Update retry job fields from a sync_files() result (retry_count already incremented)."""
    sync_error = sync_result.get("error")
    sync_total = int(sync_result.get("total") or 0)
    sync_completed = int(sync_result.get("completed") or 0)
    sync_failed = int(sync_result.get("failed") or 0)
    sync_cancelled = int(sync_result.get("cancelled") or 0)

    if sync_error or sync_total == 0:
        job.last_error = (sync_error or "推送源不可用或无目标服务器")[:500]
        if job.retry_count >= job.max_retries:
            job.status = "failed"
        else:
            job.status = "pending"
            job.next_retry_at = now + timedelta(seconds=_backoff_seconds(job.retry_count))
        return

    if sync_failed == 0 and sync_completed > 0:
        job.status = "completed"
        return

    if sync_cancelled > 0 and sync_failed == 0 and sync_completed == 0:
        job.status = "failed"
        job.last_error = "推送已取消"
        return

    if job.retry_count >= job.max_retries:
        job.status = "failed"
        job.last_error = f"Retry exhausted after {job.retry_count} attempts"
        return

    job.status = "pending"
    job.next_retry_at = now + timedelta(seconds=_backoff_seconds(job.retry_count))
    job.last_error = f"Retry {job.retry_count} failed"


async def _claim_next_retry_job(session, now: datetime) -> PushRetryJob | None:
    """Claim one due retry job with row lock (SKIP LOCKED)."""
    from sqlalchemy import select

    result = await session.execute(
        select(PushRetryJob)
        .where(
            PushRetryJob.status == "pending",
            PushRetryJob.retry_count < PushRetryJob.max_retries,
            PushRetryJob.next_retry_at <= now,
        )
        .limit(1)
        .with_for_update(skip_locked=True)
    )
    job = result.scalar_one_or_none()
    if not job:
        return None
    job.status = "running"
    await session.commit()
    return job


async def retry_runner_loop():
    """Background task: every 5min, retry pending retry jobs that are due."""
    logger.info("Retry runner loop started (interval: 5min)")
    while True:
        await asyncio.sleep(RETRY_CHECK_INTERVAL)

        try:
            async with AsyncSessionLocal() as session:
                now = datetime.now(timezone.utc)
                retried = 0

                while True:
                    job = await _claim_next_retry_job(session, now)
                    if not job:
                        break

                    job_id = job.id
                    try:
                        from server.application.services.sync_engine_v2 import SyncEngineV2
                        from server.infrastructure.database.server_repo import ServerRepositoryImpl
                        from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
                        from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
                        from server.infrastructure.database.push_schedule_repo import PushRetryJobRepositoryImpl

                        engine = SyncEngineV2(
                            server_repo=ServerRepositoryImpl(session),
                            sync_log_repo=SyncLogRepositoryImpl(session),
                            audit_repo=AuditLogRepositoryImpl(session),
                            retry_repo=PushRetryJobRepositoryImpl(session),
                        )

                        sync_result = await engine.sync_files(
                            server_ids=[job.server_id],
                            source_path=job.source_path,
                            target_path=job.target_path,
                            trigger_type="retry",
                            operator=job.operator or "retry_runner",
                        )

                        job.retry_count += 1
                        apply_push_retry_outcome(job, sync_result, now=now)

                        if job.status == "completed":
                            logger.info(
                                f"Retry job {job_id}: success after {job.retry_count} attempts"
                            )
                        elif job.status == "failed":
                            logger.warning(
                                f"Retry job {job_id}: failed — {job.last_error}"
                            )
                        else:
                            backoff = int((job.next_retry_at - now).total_seconds())
                            logger.warning(
                                f"Retry job {job_id}: attempt {job.retry_count} pending, "
                                f"next retry in {backoff}s — {job.last_error}"
                            )

                        await session.commit()
                        retried += 1

                    except Exception as e:
                        logger.error(f"Retry job {job_id} execution error: {e}")
                        await session.rollback()
                        job = await session.get(PushRetryJob, job_id)
                        if not job:
                            continue
                        job.retry_count += 1
                        job.last_error = str(e)[:500]
                        if job.retry_count >= job.max_retries:
                            job.status = "failed"
                            logger.warning(
                                f"Retry job {job_id}: max retries reached (exception), marking failed"
                            )
                        else:
                            job.status = "pending"
                            job.next_retry_at = now + timedelta(
                                seconds=_backoff_seconds(job.retry_count)
                            )
                        await session.commit()

                if retried:
                    logger.info(f"Retry runner: {retried} jobs processed")

        except Exception as e:
            logger.error(f"Retry runner check failed: {e}")