languard-servers-manager/backend/core/threads/base_thread.py

"""
BaseServerThread — base class for all per-server background threads.

Rules every subclass MUST follow:
- Call super().__init__(server_id, name) in __init__
- Implement _run_loop() — called repeatedly until _stop_event is set
- Do NOT override run() directly
- Use self._db for all database operations — it is a thread-local connection
- Call self._close_db() in your finally block if you open additional connections
- Exceptions raised from _run_loop() are caught, logged, and the loop continues
  unless the exception is a fatal error — set self._fatal_error = True to stop
"""
from __future__ import annotations

import logging
import threading
from abc import ABC, abstractmethod

from database import get_thread_db

logger = logging.getLogger(__name__)

_EXCEPTION_BACKOFF_BASE = 2.0
_EXCEPTION_BACKOFF_MAX = 60.0
_EXCEPTION_BACKOFF_MULTIPLIER = 2.0


class BaseServerThread(ABC, threading.Thread):
    """
    Abstract base for all per-server background threads.

    Subclasses implement _run_loop(). This base class handles:
    - Stop event signaling
    - Thread-local DB connection lifecycle
    - Exception backoff to prevent tight crash loops
    - Structured logging with server_id context
    """

    def __init__(self, server_id: int, name: str) -> None:
        super().__init__(name=f"{name}-server-{server_id}", daemon=True)
        self.server_id = server_id
        self._stop_event = threading.Event()
        self._fatal_error = False
        self._db = None
        self._exception_count = 0

    # ── Public API ──

    def stop(self) -> None:
        """Signal the thread to stop. Does not block."""
        self._stop_event.set()

    def stop_and_join(self, timeout: float = 5.0) -> None:
        """Signal stop and wait for the thread to exit."""
        self._stop_event.set()
        self.join(timeout=timeout)

    @property
    def is_stopping(self) -> bool:
        return self._stop_event.is_set()

    # ── Thread entry point ──

    def run(self) -> None:
        logger.info("[%s] Starting", self.name)
        backoff = _EXCEPTION_BACKOFF_BASE

        try:
            self._db = get_thread_db()
            self._on_start()

            while not self._stop_event.is_set() and not self._fatal_error:
                try:
                    self._run_loop()
                    backoff = _EXCEPTION_BACKOFF_BASE
                    self._exception_count = 0
                except Exception as exc:
                    self._exception_count += 1
                    logger.error(
                        "[%s] Unhandled exception in _run_loop (count=%d): %s",
                        self.name, self._exception_count, exc, exc_info=True,
                    )
                    if self._fatal_error:
                        break
                    self._stop_event.wait(timeout=backoff)
                    backoff = min(backoff * _EXCEPTION_BACKOFF_MULTIPLIER, _EXCEPTION_BACKOFF_MAX)

        except Exception as exc:
            logger.critical("[%s] Fatal error in thread setup: %s", self.name, exc, exc_info=True)
        finally:
            self._on_stop()
            self._close_db()
            logger.info("[%s] Stopped", self.name)

    # ── Hooks for subclasses ──

    def _on_start(self) -> None:
        """Called once before the loop starts. Override for setup."""

    def _on_stop(self) -> None:
        """Called once after the loop ends. Override for cleanup."""

    @abstractmethod
    def _run_loop(self) -> None:
        """
        Implement the thread's work here.
        Called repeatedly until stop() is called or _fatal_error is set.
        Should block for a short period (sleep or wait) to avoid busy-looping.
        """

    # ── Internal helpers ──

    def _close_db(self) -> None:
        if self._db is not None:
            try:
                self._db.close()
            except Exception as exc:
                logger.debug("[%s] Error closing DB connection: %s", self.name, exc)
            self._db = None

    def _sleep(self, seconds: float) -> None:
        """Interruptible sleep — wakes up early if stop() is called."""
        self._stop_event.wait(timeout=seconds)