From e0a45c5a46c437a763709ea109359115f6a68bd2 Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Sat, 18 Jul 2026 19:53:31 +0200 Subject: [PATCH] sync-v2 step 1: worker roles + interactive/bulk queue split - config: WORKER_ROLE (sync|jobs|all), INTERACTIVE_CONCURRENCY, BULK_CONCURRENCY. - queue: named queues (interactive/bulk) + enqueue_job() router; sync_bootstrap goes to bulk, all other jobs to interactive. - workers.run: single entrypoint; role=jobs runs both arq consumers (per-queue timeout/concurrency), role=sync runs the supervisor, all=both (dev default). - docker-compose: worker-sync + worker-jobs services (shared image, nofile ulimit); backend/Dockerfile; Makefile worker/worker-sync/worker-jobs targets. - tests repointed to app.queue.get_arq_pool; supervisor behavior unchanged. Co-Authored-By: Claude Opus 4.8 --- Makefile | 8 +- backend/Dockerfile | 16 +++ backend/app/api/accounts.py | 11 +- backend/app/api/actions.py | 14 +-- backend/app/api/compose.py | 15 +-- backend/app/config.py | 7 ++ backend/app/outbox/draft_sync.py | 8 +- backend/app/outbox/jobs.py | 12 +- backend/app/queue.py | 28 ++++- backend/app/sync/actions.py | 4 +- .../migrations/versions/0006_user_quota.py | 4 +- backend/tests/accounts/test_account_api.py | 21 ++-- backend/tests/mail/test_actions.py | 6 +- backend/tests/outbox/test_draft_sync.py | 4 +- backend/tests/outbox/test_outbox.py | 4 +- backend/workers/run.py | 111 ++++++++++++++++++ backend/workers/worker.py | 60 ++++------ docker/docker-compose.yml | 31 +++++ 18 files changed, 262 insertions(+), 102 deletions(-) create mode 100644 backend/Dockerfile create mode 100644 backend/workers/run.py diff --git a/Makefile b/Makefile index 90652bc..3815c37 100644 --- a/Makefile +++ b/Makefile @@ -9,8 +9,12 @@ reset-db: ## DESTRUCTIVE: wipe db + search + blobs, re-migrate cd backend && alembic upgrade head api: uvicorn app.main:create_app --factory --reload --port 8000 --app-dir backend -worker: - cd backend && arq workers.worker.WorkerSettings +worker: ## run both sync + jobs in one process (dev) + cd backend && WORKER_ROLE=all python -m workers.run +worker-sync: ## run only the account-sync supervisor + cd backend && WORKER_ROLE=sync python -m workers.run +worker-jobs: ## run only the arq job consumers (interactive + bulk) + cd backend && WORKER_ROLE=jobs python -m workers.run web: cd frontend && npm run dev test: diff --git a/backend/Dockerfile b/backend/Dockerfile new file mode 100644 index 0000000..f0b36eb --- /dev/null +++ b/backend/Dockerfile @@ -0,0 +1,16 @@ +# Shared image for the worker roles (refactor §1). Built from backend/. +FROM python:3.11-slim + +WORKDIR /app + +# Install deps first for layer caching. +COPY pyproject.toml ./ +COPY app ./app +COPY workers ./workers +COPY migrations ./migrations +COPY alembic.ini ./alembic.ini +RUN pip install --no-cache-dir . + +# Default to the combined role; docker-compose overrides WORKER_ROLE per service. +ENV WORKER_ROLE=all +CMD ["python", "-m", "workers.run"] diff --git a/backend/app/api/accounts.py b/backend/app/api/accounts.py index e9206f3..972ca73 100644 --- a/backend/app/api/accounts.py +++ b/backend/app/api/accounts.py @@ -30,7 +30,7 @@ from app.api.schemas import AccountOut from app.db import get_session from app.identity.deps import CurrentUser, require_csrf from app.models import Account -from app.queue import get_arq_pool +from app.queue import enqueue_job from app.redis_client import get_redis from app.security.crypto import encrypt_secret @@ -91,9 +91,8 @@ async def list_accounts(user: CurrentUser, db: DbSession) -> list[AccountOut]: async def sync_all_accounts(user: CurrentUser, db: DbSession) -> dict[str, str]: """Poll every folder of each of the user's accounts (manual Refresh).""" account_ids = list(await db.scalars(select(Account.id).where(Account.user_id == user.id))) - pool = await get_arq_pool() for account_id in account_ids: - await pool.enqueue_job("refresh_account_job", str(account_id)) + await enqueue_job("refresh_account_job", str(account_id)) return {"status": "queued"} @@ -145,8 +144,7 @@ async def create_new_account( raise_api_error(409, "account_exists", "an account for this email already exists") await db.commit() - pool = await get_arq_pool() - await pool.enqueue_job("sync_bootstrap_account", str(account.id)) + await enqueue_job("sync_bootstrap_account", str(account.id)) return _detail(account) @@ -227,8 +225,7 @@ async def update_account( await db.commit() if conn_change: - pool = await get_arq_pool() - await pool.enqueue_job("sync_bootstrap_account", str(account.id)) + await enqueue_job("sync_bootstrap_account", str(account.id)) return _detail(account) diff --git a/backend/app/api/actions.py b/backend/app/api/actions.py index 277c0fb..e0cc20c 100644 --- a/backend/app/api/actions.py +++ b/backend/app/api/actions.py @@ -15,7 +15,7 @@ from app.db import get_session from app.events import get_event_bus from app.identity.deps import CurrentUser, require_csrf from app.models import Mailbox, Message, Thread -from app.queue import get_arq_pool +from app.queue import enqueue_job from app.sync.actions import apply_flags_to_messages, sanitize_flags router = APIRouter(tags=["mail"], dependencies=[Depends(require_csrf)]) @@ -26,10 +26,9 @@ DbSession = Annotated[AsyncSession, Depends(get_session)] async def _push_and_publish( user_id: uuid.UUID, changed: list[Message], add: frozenset[str], remove: frozenset[str] ) -> None: - pool = await get_arq_pool() bus = get_event_bus() for message in changed: - await pool.enqueue_job("push_flags_job", str(message.id), sorted(add), sorted(remove)) + await enqueue_job("push_flags_job", str(message.id), sorted(add), sorted(remove)) await bus.publish( "flags.changed", { @@ -95,8 +94,7 @@ async def _move_thread( ) if thread is None: raise_api_error(404, "not_found", "thread not found") - pool = await get_arq_pool() - await pool.enqueue_job("move_thread_to_folder_job", str(thread_id), special) + await enqueue_job("move_thread_to_folder_job", str(thread_id), special) @router.post("/threads/{thread_id}/archive") @@ -126,8 +124,7 @@ async def delete_thread(thread_id: uuid.UUID, user: CurrentUser, db: DbSession) ) if thread is None: raise_api_error(404, "not_found", "thread not found") - pool = await get_arq_pool() - await pool.enqueue_job("delete_thread_job", str(thread_id)) + await enqueue_job("delete_thread_job", str(thread_id)) return {"status": "queued"} @@ -154,6 +151,5 @@ async def move_message( message.mailbox_id = dst.id # optimistic await db.commit() - pool = await get_arq_pool() - await pool.enqueue_job("move_job", str(message.id), str(dst.id), str(src_mailbox_id)) + await enqueue_job("move_job", str(message.id), str(dst.id), str(src_mailbox_id)) return {"id": str(message.id), "mailbox_id": str(dst.id)} diff --git a/backend/app/api/compose.py b/backend/app/api/compose.py index 7145129..622a0c3 100644 --- a/backend/app/api/compose.py +++ b/backend/app/api/compose.py @@ -25,7 +25,7 @@ from app.outbox.schemas import ( SendResponse, ) from app.outbox.service import DraftError, cancel_outbox, create_outbox, upsert_draft -from app.queue import get_arq_pool +from app.queue import enqueue_job router = APIRouter(tags=["compose"], dependencies=[Depends(require_csrf)]) @@ -155,9 +155,7 @@ async def list_remote_drafts(user: CurrentUser, db: DbSession) -> list[RemoteDra @router.post("/drafts/adopt/{message_id}", response_model=DraftOut) -async def adopt_remote_draft( - message_id: uuid.UUID, user: CurrentUser, db: DbSession -) -> DraftOut: +async def adopt_remote_draft(message_id: uuid.UUID, user: CurrentUser, db: DbSession) -> DraftOut: """Turn a Drafts-folder message into an editable local draft (linked by UID).""" msg = await db.scalar( select(Message) @@ -235,8 +233,7 @@ async def publish_draft(draft_id: uuid.UUID, user: CurrentUser, db: DbSession) - draft = await db.scalar(select(Draft).where(Draft.id == draft_id, Draft.user_id == user.id)) if draft is None: raise_api_error(404, "not_found", "draft not found") - pool = await get_arq_pool() - await pool.enqueue_job("push_draft_to_imap_job", str(draft_id)) + await enqueue_job("push_draft_to_imap_job", str(draft_id)) return {"status": "queued"} @@ -250,8 +247,7 @@ async def delete_draft(draft_id: uuid.UUID, user: CurrentUser, db: DbSession) -> await db.commit() # Remove the mirrored copy from the server Drafts folder too. if imap_uid: - pool = await get_arq_pool() - await pool.enqueue_job("remove_imap_draft_job", str(account_id), imap_uid) + await enqueue_job("remove_imap_draft_job", str(account_id), imap_uid) return {"status": "deleted"} @@ -265,8 +261,7 @@ async def send(body: SendRequest, user: CurrentUser, db: DbSession) -> SendRespo outbox = await create_outbox(db, user_id=user.id, draft=draft) await db.commit() - pool = await get_arq_pool() - await pool.enqueue_job( + await enqueue_job( "send_job", str(outbox.id), _defer_by=timedelta(seconds=get_settings().undo_seconds), diff --git a/backend/app/config.py b/backend/app/config.py index 49f5574..483cafa 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -34,6 +34,13 @@ class Settings(BaseSettings): max_attachment_mb: int = Field(default=25, alias="MAX_ATTACHMENT_MB") log_level: str = Field(default="INFO", alias="LOG_LEVEL") + # --- Sync engine v2 (workers layer) -------------------------------------- + # Worker role: "sync" (lease-owned account runners) | "jobs" (arq consumer) + # | "all" (both, single-process dev default). + worker_role: str = Field(default="all", alias="WORKER_ROLE") + interactive_concurrency: int = Field(default=20, alias="INTERACTIVE_CONCURRENCY") + bulk_concurrency: int = Field(default=2, alias="BULK_CONCURRENCY") + oauth_google_client_id: str = Field(default="", alias="OAUTH_GOOGLE_CLIENT_ID") oauth_google_client_secret: str = Field(default="", alias="OAUTH_GOOGLE_CLIENT_SECRET") oauth_ms_client_id: str = Field(default="", alias="OAUTH_MS_CLIENT_ID") diff --git a/backend/app/outbox/draft_sync.py b/backend/app/outbox/draft_sync.py index 7e7763a..8400891 100644 --- a/backend/app/outbox/draft_sync.py +++ b/backend/app/outbox/draft_sync.py @@ -34,9 +34,7 @@ async def _ensure_drafts_folder(provider: Any, account: Account, user_id: uuid.U uidvalidity = next((r.uidvalidity for r in remote if r.path == path), 0) async with session_scope() as session: existing = await session.scalar( - select(Mailbox).where( - Mailbox.account_id == account.id, Mailbox.special_use == "drafts" - ) + select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "drafts") ) if existing is not None: return existing.path @@ -72,9 +70,7 @@ async def push_draft_to_imap_job(ctx: dict[str, Any], draft_id: str) -> None: attachments = await load_attachments(session, user_id, draft) _, sent_raw, _, _ = build_mime(account, draft, reply_to, attachments) drafts_mb = await session.scalar( - select(Mailbox).where( - Mailbox.account_id == account.id, Mailbox.special_use == "drafts" - ) + select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "drafts") ) drafts_path = drafts_mb.path if drafts_mb else None account_obj = account diff --git a/backend/app/outbox/jobs.py b/backend/app/outbox/jobs.py index 48cd58c..6ac1655 100644 --- a/backend/app/outbox/jobs.py +++ b/backend/app/outbox/jobs.py @@ -52,9 +52,7 @@ async def _harvest_recipients(session: AsyncSession, user_id: uuid.UUID, emails: await session.execute(stmt) -async def _persist_sent_copy( - user_id: uuid.UUID, account: Account, raw: bytes, uid: int -) -> None: +async def _persist_sent_copy(user_id: uuid.UUID, account: Account, raw: bytes, uid: int) -> None: """Store the just-sent message in the DB (Sent mailbox) and notify the UI. No-op without a UIDPLUS append UID (uid <= 0); in that case a future Sent sync @@ -64,9 +62,7 @@ async def _persist_sent_copy( return async with session_scope() as session: sent_mb = await session.scalar( - select(Mailbox).where( - Mailbox.account_id == account.id, Mailbox.special_use == "sent" - ) + select(Mailbox).where(Mailbox.account_id == account.id, Mailbox.special_use == "sent") ) if sent_mb is None: return @@ -99,9 +95,7 @@ async def _persist_sent_copy( }, ) if thread_id is not None: - await bus.publish( - "thread.updated", {"user_id": str(user_id), "thread_id": str(thread_id)} - ) + await bus.publish("thread.updated", {"user_id": str(user_id), "thread_id": str(thread_id)}) async def send_job(ctx: dict[str, Any], outbox_id: str) -> None: diff --git a/backend/app/queue.py b/backend/app/queue.py index ca99526..1e638ac 100644 --- a/backend/app/queue.py +++ b/backend/app/queue.py @@ -1,14 +1,33 @@ -"""arq queue helpers (enqueue side, used by the API).""" +"""arq queue helpers (enqueue side, used by the API). + +Two named queues (refactor §6): ``interactive`` (fast user actions) and ``bulk`` +(backfills). Enqueue via :func:`enqueue_job`, which routes each job to its queue — +the API never needs to know queue names. +""" from __future__ import annotations +from typing import Any + from arq import create_pool from arq.connections import ArqRedis, RedisSettings +from arq.jobs import Job from app.config import get_settings _pool: ArqRedis | None = None +QUEUE_INTERACTIVE = "interactive" +QUEUE_BULK = "bulk" + +# Jobs default to the interactive queue; only bulk backfills go to the bulk queue. +_BULK_JOBS = frozenset({"sync_bootstrap_account"}) + + +def queue_for(job_name: str) -> str: + """Return the queue a job should run on.""" + return QUEUE_BULK if job_name in _BULK_JOBS else QUEUE_INTERACTIVE + def redis_settings() -> RedisSettings: """Return arq RedisSettings derived from ``REDIS_URL``.""" @@ -23,6 +42,13 @@ async def get_arq_pool() -> ArqRedis: return _pool +async def enqueue_job(name: str, *args: Any, **kwargs: Any) -> Job | None: + """Enqueue ``name`` onto its routed queue (interactive/bulk).""" + pool = await get_arq_pool() + kwargs.setdefault("_queue_name", queue_for(name)) + return await pool.enqueue_job(name, *args, **kwargs) + + async def close_arq_pool() -> None: """Close the arq pool.""" global _pool diff --git a/backend/app/sync/actions.py b/backend/app/sync/actions.py index f7d2028..0d327b1 100644 --- a/backend/app/sync/actions.py +++ b/backend/app/sync/actions.py @@ -273,9 +273,7 @@ async def delete_thread_job(ctx: dict[str, Any], thread_id: str) -> None: .where(Draft.reply_to_message_id.in_(msg_ids)) .values(reply_to_message_id=None) ) - await session.execute( - update(Draft).where(Draft.thread_id == tid).values(thread_id=None) - ) + await session.execute(update(Draft).where(Draft.thread_id == tid).values(thread_id=None)) for m in msgs: await session.delete(m) thread = await session.get(Thread, tid) diff --git a/backend/migrations/versions/0006_user_quota.py b/backend/migrations/versions/0006_user_quota.py index ec75e72..a74ac29 100644 --- a/backend/migrations/versions/0006_user_quota.py +++ b/backend/migrations/versions/0006_user_quota.py @@ -21,9 +21,7 @@ depends_on: str | Sequence[str] | None = None def upgrade() -> None: - op.execute( - "ALTER TABLE users ADD COLUMN quota_bytes bigint NOT NULL DEFAULT 5368709120" - ) + op.execute("ALTER TABLE users ADD COLUMN quota_bytes bigint NOT NULL DEFAULT 5368709120") def downgrade() -> None: diff --git a/backend/tests/accounts/test_account_api.py b/backend/tests/accounts/test_account_api.py index 94e32b9..0219f59 100644 --- a/backend/tests/accounts/test_account_api.py +++ b/backend/tests/accounts/test_account_api.py @@ -13,6 +13,7 @@ from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool +from app import queue as app_queue from app.api import accounts as accounts_api from app.config import get_settings from app.identity.passwords import hash_password @@ -114,14 +115,14 @@ async def test_create_account_enqueues_sync(ctx: Ctx, monkeypatch) -> None: # t enqueued: list[tuple[str, str]] = [] class FakePool: - async def enqueue_job(self, name: str, arg: str) -> None: + async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None: enqueued.append((name, arg)) async def fake_pool() -> FakePool: return FakePool() monkeypatch.setattr(accounts_api, "test_account", fake_test) - monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool) + monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool) resp = await ctx.client.post( "/api/accounts", @@ -151,14 +152,14 @@ async def _make_account(ctx: Ctx, monkeypatch) -> str: # type: ignore[no-untype return None class FakePool: - async def enqueue_job(self, name: str, arg: str) -> None: + async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None: return None async def fake_pool() -> FakePool: return FakePool() monkeypatch.setattr(accounts_api, "test_account", fake_test) - monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool) + monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool) resp = await ctx.client.post( "/api/accounts", json={**GOOD_CFG, "email": "me@example.com", "display_name": "Me", "password": "secret"}, @@ -174,13 +175,13 @@ async def test_sync_all_enqueues_refresh_per_account(ctx: Ctx, monkeypatch) -> N enqueued: list[tuple[str, str]] = [] class FakePool: - async def enqueue_job(self, name: str, arg: str) -> None: + async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None: enqueued.append((name, arg)) async def fake_pool() -> FakePool: return FakePool() - monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool) + monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool) resp = await ctx.client.post("/api/accounts/sync", headers=_hdr()) assert resp.status_code == 200 assert enqueued == [("refresh_account_job", aid)] @@ -221,7 +222,7 @@ async def test_update_password_reencrypts_and_resyncs(ctx: Ctx, monkeypatch) -> enqueued: list[tuple[str, str]] = [] class FakePool: - async def enqueue_job(self, name: str, arg: str) -> None: + async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None: enqueued.append((name, arg)) async def fake_pool() -> FakePool: @@ -231,7 +232,7 @@ async def test_update_password_reencrypts_and_resyncs(ctx: Ctx, monkeypatch) -> return None monkeypatch.setattr(accounts_api, "test_account", fake_test) - monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool) + monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool) resp = await ctx.client.patch( f"/api/accounts/{aid}", json={"password": "newsecret"}, headers=_hdr() @@ -259,14 +260,14 @@ async def test_delete_account(ctx: Ctx, monkeypatch) -> None: # type: ignore[no return None class FakePool: - async def enqueue_job(self, name: str, arg: str) -> None: + async def enqueue_job(self, name: str, arg: str, **kwargs: object) -> None: return None async def fake_pool() -> FakePool: return FakePool() monkeypatch.setattr(accounts_api, "test_account", fake_test) - monkeypatch.setattr(accounts_api, "get_arq_pool", fake_pool) + monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool) created = await ctx.client.post( "/api/accounts", diff --git a/backend/tests/mail/test_actions.py b/backend/tests/mail/test_actions.py index 18923ba..7bda4fb 100644 --- a/backend/tests/mail/test_actions.py +++ b/backend/tests/mail/test_actions.py @@ -15,7 +15,7 @@ from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool -from app.api import actions as actions_api +from app import queue as app_queue from app.config import get_settings from app.db import session_scope from app.identity.passwords import hash_password @@ -112,13 +112,13 @@ async def ctx(monkeypatch) -> AsyncIterator[Ctx]: # type: ignore[no-untyped-def # capture enqueues instead of hitting a real arq pool class FakePool: - async def enqueue_job(self, *args: object) -> None: + async def enqueue_job(self, *args: object, **kwargs: object) -> None: c.enqueued.append(args) async def fake_pool() -> FakePool: return FakePool() - monkeypatch.setattr(actions_api, "get_arq_pool", fake_pool) + monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool) token = await create_session(user.id, "pytest") c.client = httpx.AsyncClient( diff --git a/backend/tests/outbox/test_draft_sync.py b/backend/tests/outbox/test_draft_sync.py index 9fd38bf..869b8a9 100644 --- a/backend/tests/outbox/test_draft_sync.py +++ b/backend/tests/outbox/test_draft_sync.py @@ -127,9 +127,7 @@ async def test_draft_messages_are_not_threaded(ctx: Ctx) -> None: m.set_content("body") drafts_mb = await ctx.session.scalar( - select(Mailbox).where( - Mailbox.account_id == ctx.account_id, Mailbox.special_use == "drafts" - ) + select(Mailbox).where(Mailbox.account_id == ctx.account_id, Mailbox.special_use == "drafts") ) assert drafts_mb is not None remote = RemoteMessage( diff --git a/backend/tests/outbox/test_outbox.py b/backend/tests/outbox/test_outbox.py index c90444e..0d6f153 100644 --- a/backend/tests/outbox/test_outbox.py +++ b/backend/tests/outbox/test_outbox.py @@ -16,7 +16,7 @@ from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool -from app.api import compose as compose_api +from app import queue as app_queue from app.config import get_settings from app.db import session_scope from app.identity.passwords import hash_password @@ -181,7 +181,7 @@ async def ctx(monkeypatch) -> AsyncIterator[Ctx]: # type: ignore[no-untyped-def async def fake_pool() -> FakePool: return FakePool() - monkeypatch.setattr(compose_api, "get_arq_pool", fake_pool) + monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool) token = await create_session(user.id, "pytest") c.client = httpx.AsyncClient( diff --git a/backend/workers/run.py b/backend/workers/run.py new file mode 100644 index 0000000..2eac061 --- /dev/null +++ b/backend/workers/run.py @@ -0,0 +1,111 @@ +"""Worker entrypoint (refactor §1). + +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 account-sync supervisor (long-lived per-account loops). +- ``all`` — both, for single-process development (the default). + +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 workers.worker import ( + BULK_FUNCTIONS, + INTERACTIVE_FUNCTIONS, + sync_on_shutdown, + sync_on_startup, +) + +logger = logging.getLogger(__name__) + +# arq requires a non-empty function set; the sync worker consumes a queue nothing +# is ever enqueued to, so it just keeps the event loop alive for the supervisor. +_SYNC_CONTROL_QUEUE = "sync:control" + + +async def _noop(ctx: dict[str, Any]) -> None: + return None + + +def _worker( + functions: list[Any], + queue_name: str, + max_jobs: int, + job_timeout: int, + on_startup: Any = None, + on_shutdown: Any = None, +) -> Worker: + return Worker( + functions=functions, + queue_name=queue_name, + redis_settings=redis_settings(), + max_jobs=max_jobs, + job_timeout=job_timeout, + on_startup=on_startup, + on_shutdown=on_shutdown, + handle_signals=False, # the runner owns signal handling (graceful shutdown) + ) + + +def build_workers(role: str) -> list[Worker]: + """Construct the arq Worker instances for the given role.""" + s = get_settings() + workers: list[Worker] = [] + if role in ("sync", "all"): + workers.append( + _worker([_noop], _SYNC_CONTROL_QUEUE, 1, 3600, sync_on_startup, sync_on_shutdown) + ) + if role in ("jobs", "all"): + workers.append( + _worker(INTERACTIVE_FUNCTIONS, QUEUE_INTERACTIVE, s.interactive_concurrency, 120) + ) + workers.append(_worker(BULK_FUNCTIONS, QUEUE_BULK, s.bulk_concurrency, 3600)) + return workers + + +async def _run() -> None: + logging.basicConfig(level=get_settings().log_level) + role = get_settings().worker_role + workers = build_workers(role) + logger.info("worker starting: role=%s (%d arq worker(s))", role, len(workers)) + + tasks = [asyncio.create_task(w.async_run()) for w in workers] + 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: cancelling %d worker(s)", len(tasks)) + for task in tasks: + task.cancel() # arq runs each worker's on_shutdown in its finally + await asyncio.gather(*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() diff --git a/backend/workers/worker.py b/backend/workers/worker.py index 01199e1..59d2ec9 100644 --- a/backend/workers/worker.py +++ b/backend/workers/worker.py @@ -1,6 +1,8 @@ -"""arq worker entrypoint (SPEC §2 ``workers/worker.py``). +"""Worker job functions and the sync-worker lifecycle (refactor §1, §6). -Run with: ``arq workers.worker.WorkerSettings`` (from ``backend/``). +The process is launched via ``workers.run`` (``WORKER_ROLE`` decides which arq +workers run). This module only defines the job callables and the two queue +function-sets; the runner wires them onto the ``interactive`` / ``bulk`` queues. """ from __future__ import annotations @@ -9,10 +11,8 @@ import logging import uuid from typing import Any -from app.config import get_settings from app.outbox.draft_sync import push_draft_to_imap_job, remove_imap_draft_job from app.outbox.jobs import send_job -from app.queue import redis_settings as arq_redis_settings from app.sync.actions import ( delete_thread_job, move_job, @@ -32,41 +32,33 @@ async def sync_bootstrap_account(ctx: dict[str, Any], account_id: str) -> int: return count -async def on_startup(ctx: dict[str, Any]) -> None: - logging.basicConfig(level=get_settings().log_level) +# Fast, interactive user actions (§6): short timeout, high concurrency. +INTERACTIVE_FUNCTIONS = [ + push_flags_job, + move_job, + move_thread_to_folder_job, + delete_thread_job, + send_job, + push_draft_to_imap_job, + remove_imap_draft_job, + refresh_account_job, +] + +# Bulk work (§6): long timeout, low concurrency, isolated from interactive jobs. +BULK_FUNCTIONS = [sync_bootstrap_account] + + +async def sync_on_startup(ctx: dict[str, Any]) -> None: + """Sync-worker startup: launch the account-sync supervisor.""" from app.sync.supervisor import start_supervisor await start_supervisor(ctx) - logger.info("webmail worker started") + logger.info("sync worker: supervisor started") -async def on_shutdown(ctx: dict[str, Any]) -> None: - from app.db import dispose_engine - from app.redis_client import close_redis +async def sync_on_shutdown(ctx: dict[str, Any]) -> None: + """Sync-worker shutdown: stop the supervisor (process cleanup is in the runner).""" from app.sync.supervisor import stop_supervisor await stop_supervisor(ctx) - await dispose_engine() - await close_redis() - logger.info("webmail worker stopped") - - -class WorkerSettings: - """arq worker configuration.""" - - functions = [ - sync_bootstrap_account, - push_flags_job, - move_job, - move_thread_to_folder_job, - delete_thread_job, - send_job, - push_draft_to_imap_job, - remove_imap_draft_job, - refresh_account_job, - ] - on_startup = on_startup - on_shutdown = on_shutdown - max_jobs = 10 # global worker cap (SPEC §6.3) - job_timeout = 3600 # backfill of a large account can be long - redis_settings = arq_redis_settings() + logger.info("sync worker: supervisor stopped") diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index a142ad9..5f925c5 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -36,6 +36,36 @@ services: timeout: 3s retries: 10 + # Sync-engine v2 workers (refactor §1): two roles sharing one image. Scale the + # sync role with `--scale worker-sync=N` (lease protocol prevents double-owning). + worker-sync: + build: { context: ../backend } + command: ["python", "-m", "workers.run"] + environment: &worker-env + WORKER_ROLE: sync + APP_DOMAIN: dev.webmail.local + DATABASE_URL: postgresql+asyncpg://webmail:webmail_dev@postgres:5432/webmail_dev + REDIS_URL: redis://redis:6379/0 + MASTER_KEY: CHANGE_ME_base64_32_bytes + BLOB_ROOT: /var/blobs + volumes: [blobs:/var/blobs] + ulimits: { nofile: 65536 } + depends_on: + postgres: { condition: service_healthy } + redis: { condition: service_healthy } + + worker-jobs: + build: { context: ../backend } + command: ["python", "-m", "workers.run"] + environment: + <<: *worker-env + WORKER_ROLE: jobs + volumes: [blobs:/var/blobs] + ulimits: { nofile: 65536 } + depends_on: + postgres: { condition: service_healthy } + redis: { condition: service_healthy } + # Test mail servers (integration tests + manual dev against a throwaway mailbox) greenmail: image: greenmail/standalone:2.0.1 @@ -51,3 +81,4 @@ services: volumes: pgdata: meilidata: + blobs: