webmail-client/backend/workers/run.py

"""Worker entrypoint (refactor §1, §7).

One image, two roles selected by ``WORKER_ROLE``:

- ``jobs`` — arq consumer for the ``interactive`` and ``bulk`` queues (both in one
  process, per-queue concurrency/timeout).
- ``sync`` — runs the lease-owned account-sync coordinator (§2) directly.
- ``all``  — both, for single-process development (the default).

Graceful shutdown (SIGINT/SIGTERM): stop the coordinator first — it stops the
claimer, cancels the runners, and releases every lease so a peer adopts the
accounts immediately — then drain the arq consumers. Target < 10 s.

Run with: ``python -m workers.run``.
"""

from __future__ import annotations

import asyncio
import logging
import signal
from typing import Any

from arq.worker import Worker

from app.config import get_settings
from app.queue import QUEUE_BULK, QUEUE_INTERACTIVE, redis_settings
from app.sync.coordinator import SyncCoordinator
from workers.worker import BULK_FUNCTIONS, INTERACTIVE_FUNCTIONS

logger = logging.getLogger(__name__)

_SHUTDOWN_DRAIN_S = 10.0  # §7: release leases + drain within this budget


def _worker(
    functions: list[Any],
    queue_name: str,
    max_jobs: int,
    job_timeout: int,
    cron_jobs: list[Any] | None = None,
) -> Worker:
    return Worker(
        functions=functions,
        queue_name=queue_name,
        redis_settings=redis_settings(),
        max_jobs=max_jobs,
        job_timeout=job_timeout,
        handle_signals=False,  # the runner owns signal handling (graceful shutdown)
        cron_jobs=cron_jobs or [],
    )


def build_arq_workers(role: str) -> list[Worker]:
    """Construct the arq Worker instances for the given role (jobs only)."""
    s = get_settings()
    if role not in ("jobs", "all"):
        return []
    # Plugin jobs (SPEC §11.2) join the interactive queue; cron-scheduled ones run
    # on that worker too. With no plugins installed these are empty (inert).
    from app.plugins.jobs import plugin_arq_functions, plugin_cron_jobs

    interactive: list[Any] = [*INTERACTIVE_FUNCTIONS, *plugin_arq_functions()]
    return [
        _worker(
            interactive,
            QUEUE_INTERACTIVE,
            s.interactive_concurrency,
            120,
            cron_jobs=plugin_cron_jobs(),
        ),
        _worker(BULK_FUNCTIONS, QUEUE_BULK, s.bulk_concurrency, 3600),
    ]


async def _run() -> None:
    logging.basicConfig(level=get_settings().log_level)
    role = get_settings().worker_role

    # Subscribe plugin hooks so events published during sync (message.new, …) reach
    # them in this process too (SPEC §11; the API process registers separately).
    from app.events import get_event_bus
    from app.plugins import loader

    loader.register_hooks(get_event_bus())

    coordinator = SyncCoordinator() if role in ("sync", "all") else None
    if coordinator is not None:
        await coordinator.start()
    arq_tasks = [asyncio.create_task(w.async_run()) for w in build_arq_workers(role)]
    logger.info("worker starting: role=%s (%d arq worker(s))", role, len(arq_tasks))

    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop.set)
    await stop.wait()

    logger.info("worker stopping: releasing leases + draining")
    if coordinator is not None:
        try:
            await asyncio.wait_for(coordinator.stop(), timeout=_SHUTDOWN_DRAIN_S)
        except TimeoutError:
            logger.warning("coordinator stop exceeded %.0fs drain budget", _SHUTDOWN_DRAIN_S)
    for task in arq_tasks:
        task.cancel()
    await asyncio.gather(*arq_tasks, return_exceptions=True)

    from app.db import dispose_engine
    from app.queue import close_arq_pool
    from app.redis_client import close_redis

    await dispose_engine()
    await close_redis()
    await close_arq_pool()
    logger.info("worker stopped")


def main() -> None:
    asyncio.run(_run())


if __name__ == "__main__":
    main()