FluxIP/app/rotation/schedule_service.py

from __future__ import annotations

import asyncio
import logging
import time
from typing import Any

from app.core.errors import ConflictError
from app.core.time import to_iso
from app.rotation.repository import RotationRepository
from app.rotation.service import RotationService

logger = logging.getLogger(__name__)


class SchedulerLoop:
    def __init__(
        self,
        repository: RotationRepository,
        rotation_service: RotationService,
    ) -> None:
        self.repository = repository
        self.rotation_service = rotation_service
        self._task: asyncio.Task[None] | None = None
        self._stopping = asyncio.Event()
        self._tick_seconds: float | None = None
        self._started_at: str | None = None
        self._last_tick_at: str | None = None
        self._last_success_at: str | None = None
        self._last_dispatch_at: str | None = None
        self._last_error_at: str | None = None
        self._consecutive_errors = 0
        self._started_monotonic: float | None = None
        self._last_tick_monotonic: float | None = None

    def start(self, tick_seconds: float) -> None:
        if self._task and not self._task.done():
            return
        self._stopping = asyncio.Event()
        self._tick_seconds = tick_seconds
        self._started_at = to_iso()
        self._started_monotonic = time.monotonic()
        self._last_tick_monotonic = None
        self._task = asyncio.create_task(self._run(tick_seconds), name="fleet-schedule-loop")
        self._task.add_done_callback(self._task_done)
        logger.info("scheduler_started", extra={"tick_seconds": tick_seconds})

    async def stop(self) -> None:
        self._stopping.set()
        if self._task:
            self._task.cancel()
            await asyncio.gather(self._task, return_exceptions=True)

    def status(self, *, enabled: bool) -> dict[str, Any]:
        running = bool(self._task and not self._task.done())
        state = "disabled" if not enabled else "running" if running else "stopped"
        healthy = not enabled or (
            running and self._consecutive_errors < 3 and self._tick_is_fresh()
        )
        return {
            "enabled": enabled,
            "running": running,
            "healthy": healthy,
            "state": state,
            "tick_seconds": self._tick_seconds,
            "started_at": self._started_at,
            "last_tick_at": self._last_tick_at,
            "last_success_at": self._last_success_at,
            "last_dispatch_at": self._last_dispatch_at,
            "last_error_at": self._last_error_at,
            "consecutive_errors": self._consecutive_errors,
        }

    async def _run(self, tick_seconds: float) -> None:
        while not self._stopping.is_set():
            self._last_tick_at = to_iso()
            self._last_tick_monotonic = time.monotonic()
            try:
                tick_succeeded = self._dispatch_due_group()
            except asyncio.CancelledError:
                raise
            except Exception:
                self._record_error()
                self._consecutive_errors += 1
                logger.exception("scheduler_tick_failed")
            else:
                if tick_succeeded:
                    self._last_success_at = to_iso()
                    self._consecutive_errors = 0
                else:
                    self._consecutive_errors += 1
            try:
                await asyncio.wait_for(self._stopping.wait(), timeout=tick_seconds)
            except TimeoutError:
                continue

    def _dispatch_due_group(self) -> bool:
        succeeded = True
        due_group_ids = self.repository.due_group_ids()
        if due_group_ids:
            logger.info(
                "scheduler_due_groups_found",
                extra={"due_group_count": len(due_group_ids)},
            )
        for group_id in due_group_ids:
            try:
                run = self.rotation_service.start_group(group_id, trigger="scheduled")
            except ConflictError as exc:
                logger.warning(
                    "scheduler_dispatch_conflict",
                    extra={"group_id": group_id, "code": exc.code},
                )
                if exc.code == "ROTATION_IN_PROGRESS":
                    break
            except Exception:
                self._record_error()
                succeeded = False
                logger.exception("scheduler_dispatch_failed", extra={"group_id": group_id})
                try:
                    self.repository.defer_group(group_id)
                except Exception:
                    self._record_error()
                    logger.exception("scheduler_defer_failed", extra={"group_id": group_id})
            else:
                self._last_dispatch_at = to_iso()
                logger.info(
                    "scheduler_group_dispatched",
                    extra={"group_id": group_id, "run_id": run.id},
                )
                break
        return succeeded

    def _record_error(self) -> None:
        self._last_error_at = to_iso()

    def _tick_is_fresh(self) -> bool:
        reference = self._last_tick_monotonic or self._started_monotonic
        if reference is None:
            return False
        allowed_silence = max((self._tick_seconds or 10) * 3, 10)
        return time.monotonic() - reference <= allowed_silence

    def _task_done(self, task: asyncio.Task[None]) -> None:
        if task.cancelled():
            if self._stopping.is_set():
                logger.info("scheduler_stopped")
            else:
                self._record_error()
                self._consecutive_errors += 1
                logger.error("scheduler_cancelled_unexpectedly")
            return
        try:
            task.result()
        except Exception:
            self._record_error()
            self._consecutive_errors += 1
            logger.exception("scheduler_exited_unexpectedly")
            return
        if not self._stopping.is_set():
            self._record_error()
            self._consecutive_errors += 1
            logger.error("scheduler_exited_unexpectedly")
        else:
            logger.info("scheduler_stopped")