feat: add ARQ background worker to container
- Create app/core/worker.py with WorkerSettings for ARQ - Register all job functions: reindex, index_*, embedding_batch, deep_analysis - Update prestart.sh to start ARQ worker in background before uvicorn - Worker runs with max_jobs=10, job_timeout=300s
This commit is contained in:
@@ -0,0 +1,88 @@
|
|||||||
|
"""ARQ worker configuration and entrypoint."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from arq.connections import RedisSettings
|
||||||
|
|
||||||
|
from app.config import get_settings
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def _get_redis_settings() -> RedisSettings:
|
||||||
|
"""Get Redis settings from app config."""
|
||||||
|
settings = get_settings()
|
||||||
|
return RedisSettings.from_dsn(settings.redis_url)
|
||||||
|
|
||||||
|
|
||||||
|
async def on_startup(ctx: dict[str, Any]) -> None:
|
||||||
|
"""Called when worker starts."""
|
||||||
|
logger.info("ARQ worker starting...")
|
||||||
|
|
||||||
|
|
||||||
|
async def on_shutdown(ctx: dict[str, Any]) -> None:
|
||||||
|
"""Called when worker shuts down."""
|
||||||
|
logger.info("ARQ worker shutting down...")
|
||||||
|
|
||||||
|
|
||||||
|
def _collect_job_functions() -> dict[str, Any]:
|
||||||
|
"""Collect all job functions from plugins and core.
|
||||||
|
|
||||||
|
Each plugin's jobs module may define async functions with the
|
||||||
|
signature (ctx: dict, *args) -> None. We import them all and
|
||||||
|
register by name.
|
||||||
|
"""
|
||||||
|
functions: dict[str, Any] = {}
|
||||||
|
|
||||||
|
# Unified search jobs
|
||||||
|
try:
|
||||||
|
from app.plugins.builtins.unified_search.jobs import (
|
||||||
|
index_mails,
|
||||||
|
index_file,
|
||||||
|
index_contact,
|
||||||
|
index_company,
|
||||||
|
index_event,
|
||||||
|
reindex,
|
||||||
|
embedding_batch,
|
||||||
|
)
|
||||||
|
functions.update({
|
||||||
|
"index_mails": index_mails,
|
||||||
|
"index_file": index_file,
|
||||||
|
"index_contact": index_contact,
|
||||||
|
"index_company": index_company,
|
||||||
|
"index_event": index_event,
|
||||||
|
"reindex": reindex,
|
||||||
|
"embedding_batch": embedding_batch,
|
||||||
|
})
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Failed to import unified_search jobs", exc_info=True)
|
||||||
|
|
||||||
|
# AI proactive jobs
|
||||||
|
try:
|
||||||
|
from app.plugins.builtins.ai_proactive.jobs import deep_analysis
|
||||||
|
functions["deep_analysis"] = deep_analysis
|
||||||
|
except Exception:
|
||||||
|
logger.warning("Failed to import ai_proactive jobs", exc_info=True)
|
||||||
|
|
||||||
|
# Calendar jobs (calendar_reminder is enqueued but no handler exists yet)
|
||||||
|
# try:
|
||||||
|
# from app.plugins.builtins.calendar.jobs import calendar_reminder
|
||||||
|
# functions["calendar_reminder"] = calendar_reminder
|
||||||
|
# except Exception:
|
||||||
|
# logger.warning("Failed to import calendar jobs", exc_info=True)
|
||||||
|
|
||||||
|
return functions
|
||||||
|
|
||||||
|
|
||||||
|
class WorkerSettings:
|
||||||
|
"""ARQ worker settings."""
|
||||||
|
functions = _collect_job_functions()
|
||||||
|
redis_settings = _get_redis_settings()
|
||||||
|
on_startup = on_startup
|
||||||
|
on_shutdown = on_shutdown
|
||||||
|
max_jobs = 10
|
||||||
|
job_timeout = 300
|
||||||
|
queue_name = "arq:queue"
|
||||||
Regular → Executable
+8
-7
@@ -4,15 +4,13 @@
|
|||||||
#
|
#
|
||||||
# Responsibilities:
|
# Responsibilities:
|
||||||
# 1. Run Alembic DB migrations (alembic upgrade head).
|
# 1. Run Alembic DB migrations (alembic upgrade head).
|
||||||
# 2. Start uvicorn as PID 1 (so signals like SIGTERM are forwarded correctly).
|
# 2. Start ARQ background worker in background.
|
||||||
|
# 3. Start uvicorn as PID 1 (so signals like SIGTERM are forwarded correctly).
|
||||||
#
|
#
|
||||||
# Notes:
|
# Notes:
|
||||||
# - `set -e` ensures the container crashes loudly if migrations fail,
|
# - `set -e` ensures the container crashes loudly if migrations fail.
|
||||||
# rather than starting a broken app on an inconsistent schema.
|
# - ARQ worker runs in background, uvicorn becomes PID 1.
|
||||||
# - `exec uvicorn ...` replaces the shell process with uvicorn, so uvicorn
|
# - `--workers 1` is intentional for v1.
|
||||||
# becomes PID 1 and receives Docker's SIGTERM directly for graceful shutdown.
|
|
||||||
# - `--workers 1` is intentional for v1: SQLite (dev) is single-threaded,
|
|
||||||
# and asyncpg (prod) scales fine with one worker + an async connection pool.
|
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
|
|
||||||
set -e
|
set -e
|
||||||
@@ -21,6 +19,9 @@ echo "[prestart] $(date -u +%Y-%m-%dT%H:%M:%SZ) - Running alembic upgrade head..
|
|||||||
alembic upgrade head
|
alembic upgrade head
|
||||||
echo "[prestart] DB migrations completed successfully."
|
echo "[prestart] DB migrations completed successfully."
|
||||||
|
|
||||||
|
echo "[prestart] Starting ARQ background worker..."
|
||||||
|
arq app.core.worker.WorkerSettings &
|
||||||
|
|
||||||
echo "[prestart] Starting uvicorn on 0.0.0.0:8000 (workers=1)..."
|
echo "[prestart] Starting uvicorn on 0.0.0.0:8000 (workers=1)..."
|
||||||
exec uvicorn app.main:app \
|
exec uvicorn app.main:app \
|
||||||
--host 0.0.0.0 \
|
--host 0.0.0.0 \
|
||||||
|
|||||||
Reference in New Issue
Block a user