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 <noreply@anthropic.com>

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:

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"]

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)
 

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)}

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),

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")

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

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:

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

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)

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:

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",

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(

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(

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(

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()

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")

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: