✨(worker) add worker.py command and improve task routing on queues (#504)

This avoids a case where a very large import would overflow the
"celery" queue with reindex tasks that would block the inbound
processing tasks. Now we have good priorization.

Makefile

@@ -132,7 +132,7 @@ logs: ## display all services logs (follow mode)
 .PHONY: logs
 
 start: ## start all development services
-	@$(COMPOSE) up --force-recreate --build -d frontend-dev backend-dev celery-dev mta-in --wait
+	@$(COMPOSE) up --force-recreate --build -d frontend-dev backend-dev worker-dev mta-in --wait
 .PHONY: start
 
 start-minimal: ## start minimal services (backend, frontend, keycloak and DB)

Procfile

@@ -1,3 +1,3 @@
 web: bin/scalingo_run_web
-worker: celery -A messages.celery_app worker --task-events --beat -l INFO -c $CELERY_CONCURRENCY -Q celery,default
+worker: python worker.py
 postdeploy: python manage.py migrate

compose.yaml

@@ -147,14 +147,14 @@ services:
       target: poetry
     pull_policy: build
 
-  celery-dev:
+  worker-dev:
     build:
       context: src/backend
       target: runtime-dev
       args:
         DOCKER_USER: ${DOCKER_USER:-1000}
     user: ${DOCKER_USER:-1000}
-    command: ["celery", "-A", "messages.celery_app", "worker", "-l", "DEBUG", "-Q", "celery,default"]
+    command: ["python", "worker.py", "--loglevel=DEBUG"]
     environment:
       - DJANGO_CONFIGURATION=Development
     env_file:
@@ -166,7 +166,7 @@ services:
     depends_on:
       - backend-dev
 
-  celery-ui:
+  worker-ui:
     build:
       context: src/backend
       target: runtime-dev

docs/worker.md

@@ -0,0 +1,108 @@
+# Background Task Worker
+
+The application uses a background task worker to process asynchronous jobs like email processing, file imports, and search indexing.
+
+## Quick Start
+
+```bash
+# Start with all queues and scheduler (default)
+python worker.py
+
+# Start with specific queues only
+python worker.py --queues=inbound,outbound
+
+# Exclude low-priority queues
+python worker.py --exclude=reindex,imports
+
+# Disable the scheduler (for secondary workers)
+python worker.py --disable-scheduler
+```
+
+## Queues
+
+Tasks are routed to specific queues based on their type. Queues are listed in priority order.
+
+| Priority | Queue | Description |
+|----------|-------|-------------|
+| 1 (highest) | `management` | Admin/management tasks (migrations, cleanup) |
+| 2 | `inbound` | Inbound email processing (time-sensitive) |
+| 3 | `outbound` | Outbound email sending and retries |
+| 4 | `default` | General tasks (fallback for unrouted tasks) |
+| 5 | `imports` | File import processing (MBOX, EML, IMAP) |
+| 6 (lowest) | `reindex` | Search indexing |
+
+### Queue Routing
+
+Tasks are automatically routed to queues based on their module:
+
+| Task Module | Queue |
+|-------------|-------|
+| `core.mda.inbound_tasks.*` | `inbound` |
+| `core.mda.outbound_tasks.*` | `outbound` |
+| `core.services.importer.tasks.*` | `imports` |
+| `core.services.search.tasks.*` | `reindex` |
+| Everything else | `default` |
+
+## CLI Options
+
+| Option | Description |
+|--------|-------------|
+| `--queues`, `-Q` | Comma-separated list of queues to process |
+| `--exclude`, `-X` | Comma-separated list of queues to exclude |
+| `--concurrency`, `-c` | Number of worker processes (default: CPU count) |
+| `--disable-scheduler` | Disable the task scheduler (enabled by default) |
+| `--loglevel`, `-l` | Logging level (default: INFO) |
+
+## Environment Variables
+
+| Variable | Description |
+|----------|-------------|
+| `WORKER_CONCURRENCY` | Default concurrency if `--concurrency` not specified |
+| `CELERY_CONCURRENCY` | Fallback for `WORKER_CONCURRENCY` (legacy) |
+
+## Deployment
+
+### Scalingo (Procfile)
+
+```text
+worker: python worker.py
+```
+
+### Docker Compose
+
+The `worker-dev` service in `compose.yaml` runs the worker for local development:
+
+```yaml
+worker-dev:
+  command: ["python", "worker.py", "--loglevel=DEBUG"]
+```
+
+### Running Multiple Workers
+
+For high-throughput deployments or strict queue isolation, run specialized workers for different queue groups:
+
+```bash
+# Worker 1: High-priority email processing (with scheduler)
+python worker.py --queues=management,inbound,outbound
+
+# Worker 2: Background tasks only (no scheduler)
+python worker.py --queues=default,imports,reindex --disable-scheduler
+```
+
+This ensures that low-priority tasks (imports, reindex) never compete with email processing.
+
+## Monitoring
+
+The `worker-ui` service provides a web UI for monitoring tasks (Flower). Access it at `http://localhost:8903` in development.
+
+Task events are enabled by default for monitoring tools.
+
+## Scheduled Tasks
+
+The scheduler is enabled by default. The following tasks are scheduled:
+
+| Task | Schedule | Queue |
+|------|----------|-------|
+| Retry pending messages | Every 5 minutes | `outbound` |
+| System selfcheck | Configurable interval | `outbound` |
+| Process inbound queue | Every 5 minutes | `inbound` |

src/backend/core/tests/test_worker.py

@@ -0,0 +1,343 @@
+"""Tests for the background task worker configuration."""
+
+# pylint: disable=import-outside-toplevel
+
+from django.conf import settings
+
+import pytest
+
+
+class TestWorkerQueueConfiguration:
+    """Test worker queue definitions and routing."""
+
+    def test_all_queues_defined_in_priority_order(self):
+        """Verify all queues are defined in the expected priority order."""
+        # Import here to avoid import issues during test collection
+        import worker
+
+        expected_queues = [
+            "management",
+            "inbound",
+            "outbound",
+            "default",
+            "imports",
+            "reindex",
+        ]
+        assert worker.ALL_QUEUES == expected_queues
+
+    def test_default_queues_includes_all(self):
+        """Verify default queues includes all defined queues."""
+        import worker
+
+        assert worker.DEFAULT_QUEUES == worker.ALL_QUEUES
+
+    def test_celery_default_queue_is_default(self):
+        """Verify the celery default queue is set to 'default'."""
+        assert settings.CELERY_TASK_DEFAULT_QUEUE == "default"
+
+    def test_task_routes_configured(self):
+        """Verify task routes are configured for all expected modules."""
+        routes = settings.CELERY_TASK_ROUTES
+
+        assert "core.mda.inbound_tasks.*" in routes
+        assert routes["core.mda.inbound_tasks.*"]["queue"] == "inbound"
+
+        assert "core.mda.outbound_tasks.*" in routes
+        assert routes["core.mda.outbound_tasks.*"]["queue"] == "outbound"
+
+        assert "core.services.importer.tasks.*" in routes
+        assert routes["core.services.importer.tasks.*"]["queue"] == "imports"
+
+        assert "core.services.search.tasks.*" in routes
+        assert routes["core.services.search.tasks.*"]["queue"] == "reindex"
+
+
+class TestWorkerCLIParsing:
+    """Test worker CLI argument parsing."""
+
+    def test_parse_args_defaults(self):
+        """Test default argument values."""
+        import sys
+
+        import worker
+
+        # Save original argv
+        original_argv = sys.argv
+        try:
+            sys.argv = ["worker.py"]
+            args = worker.parse_args()
+
+            assert args.queues is None
+            assert args.exclude is None
+            assert args.disable_scheduler is False
+            assert args.loglevel == "INFO"
+        finally:
+            sys.argv = original_argv
+
+    def test_parse_args_with_queues(self):
+        """Test parsing --queues argument."""
+        import sys
+
+        import worker
+
+        original_argv = sys.argv
+        try:
+            sys.argv = ["worker.py", "--queues=inbound,outbound"]
+            args = worker.parse_args()
+
+            assert args.queues == "inbound,outbound"
+        finally:
+            sys.argv = original_argv
+
+    def test_parse_args_with_exclude(self):
+        """Test parsing --exclude argument."""
+        import sys
+
+        import worker
+
+        original_argv = sys.argv
+        try:
+            sys.argv = ["worker.py", "--exclude=reindex,imports"]
+            args = worker.parse_args()
+
+            assert args.exclude == "reindex,imports"
+        finally:
+            sys.argv = original_argv
+
+    def test_parse_args_with_disable_scheduler(self):
+        """Test parsing --disable-scheduler flag."""
+        import sys
+
+        import worker
+
+        original_argv = sys.argv
+        try:
+            sys.argv = ["worker.py", "--disable-scheduler"]
+            args = worker.parse_args()
+
+            assert args.disable_scheduler is True
+        finally:
+            sys.argv = original_argv
+
+    def test_parse_args_with_concurrency(self):
+        """Test parsing --concurrency argument."""
+        import sys
+
+        import worker
+
+        original_argv = sys.argv
+        try:
+            sys.argv = ["worker.py", "--concurrency=4"]
+            args = worker.parse_args()
+
+            assert args.concurrency == 4
+        finally:
+            sys.argv = original_argv
+
+    def test_parse_args_with_loglevel(self):
+        """Test parsing --loglevel argument."""
+        import sys
+
+        import worker
+
+        original_argv = sys.argv
+        try:
+            sys.argv = ["worker.py", "--loglevel=DEBUG"]
+            args = worker.parse_args()
+
+            assert args.loglevel == "DEBUG"
+        finally:
+            sys.argv = original_argv
+
+    def test_parse_args_short_flags(self):
+        """Test parsing short flag versions."""
+        import sys
+
+        import worker
+
+        original_argv = sys.argv
+        try:
+            sys.argv = [
+                "worker.py",
+                "-Q",
+                "inbound",
+                "-X",
+                "reindex",
+                "-c",
+                "2",
+                "-l",
+                "WARNING",
+            ]
+            args = worker.parse_args()
+
+            assert args.queues == "inbound"
+            assert args.exclude == "reindex"
+            assert args.concurrency == 2
+            assert args.loglevel == "WARNING"
+        finally:
+            sys.argv = original_argv
+
+
+class TestWorkerQueueValidation:
+    """Test queue validation logic."""
+
+    def test_valid_queues_accepted(self):
+        """Test that valid queue names are accepted."""
+        import worker
+
+        valid_queues = [
+            "management",
+            "inbound",
+            "outbound",
+            "default",
+            "imports",
+            "reindex",
+        ]
+        for queue in valid_queues:
+            assert queue in worker.ALL_QUEUES
+
+    def test_queue_exclusion_logic(self):
+        """Test that queue exclusion works correctly."""
+        import worker
+
+        queues = worker.ALL_QUEUES.copy()
+        exclude = ["reindex", "imports"]
+        result = [q for q in queues if q not in exclude]
+
+        assert "reindex" not in result
+        assert "imports" not in result
+        assert "inbound" in result
+        assert "outbound" in result
+        assert "default" in result
+        assert "management" in result
+
+    def test_queue_order_preserved_after_exclusion(self):
+        """Test that queue priority order is preserved after exclusion."""
+        import worker
+
+        queues = worker.ALL_QUEUES.copy()
+        exclude = ["outbound", "imports"]
+        result = [q for q in queues if q not in exclude]
+
+        expected = ["management", "inbound", "default", "reindex"]
+        assert result == expected
+
+
+class TestBeatScheduleQueues:
+    """Test that beat schedule tasks use correct queues."""
+
+    def test_beat_schedule_uses_correct_queues(self):
+        """Verify scheduled tasks are routed to appropriate queues."""
+        from messages.celery_app import app
+
+        if not hasattr(app.conf, "beat_schedule") or not app.conf.beat_schedule:
+            pytest.skip("Beat schedule is disabled")
+
+        schedule = app.conf.beat_schedule
+
+        # Check retry-pending-messages uses outbound queue
+        if "retry-pending-messages" in schedule:
+            assert schedule["retry-pending-messages"]["options"]["queue"] == "outbound"
+
+        # Check selfcheck uses outbound queue
+        if "selfcheck" in schedule:
+            assert schedule["selfcheck"]["options"]["queue"] == "outbound"
+
+        # Check process-inbound-messages-queue uses inbound queue
+        if "process-inbound-messages-queue" in schedule:
+            assert (
+                schedule["process-inbound-messages-queue"]["options"]["queue"]
+                == "inbound"
+            )
+
+
+class TestWorkerE2E:
+    """End-to-end tests for the worker process."""
+
+    def test_worker_starts_successfully(self):
+        """Test that the worker process starts without immediate errors."""
+        import subprocess
+
+        # Start worker with minimal config, disable scheduler to avoid side effects
+        # pylint: disable=consider-using-with
+        process = subprocess.Popen(
+            [
+                "python",
+                "worker.py",
+                "--queues=default",
+                "--disable-scheduler",
+                "--loglevel=INFO",
+                "--concurrency=1",
+            ],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+        )
+
+        try:
+            # Wait briefly for startup - if it crashes immediately, we'll know
+            # Use communicate with timeout to capture output
+            try:
+                stdout, _ = process.communicate(timeout=3)
+            except subprocess.TimeoutExpired:
+                # Worker is still running after 3 seconds - this is expected
+                stdout = ""
+
+            # Check if process exited with an error
+            exit_code = process.poll()
+            if exit_code is not None and exit_code != 0:
+                pytest.fail(
+                    f"Worker process exited with code {exit_code}. Output: {stdout}"
+                )
+
+            # If still running or exited cleanly, the test passes
+            # Worker starting without crashing is the success criterion
+        finally:
+            # Cleanup: terminate the worker if still running
+            if process.poll() is None:
+                process.terminate()
+                try:
+                    process.wait(timeout=5)
+                except subprocess.TimeoutExpired:
+                    process.kill()
+                    process.wait()
+
+    def test_worker_rejects_invalid_queues(self):
+        """Test that the worker rejects invalid queue names."""
+        import subprocess
+
+        result = subprocess.run(
+            [
+                "python",
+                "worker.py",
+                "--queues=invalid_queue_name",
+            ],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            check=False,
+        )
+
+        assert result.returncode != 0
+        assert (
+            "Unknown queues" in result.stderr or "invalid_queue_name" in result.stderr
+        )
+
+    def test_worker_rejects_invalid_exclude_queues(self):
+        """Test that the worker rejects invalid queue names in --exclude."""
+        import subprocess
+
+        result = subprocess.run(
+            [
+                "python",
+                "worker.py",
+                "--exclude=invalid_queue_name",
+            ],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            check=False,
+        )
+
+        assert result.returncode != 0
+        assert "Unknown queues to exclude" in result.stderr

src/backend/messages/celery_app.py

@@ -33,16 +33,16 @@ if not settings.DISABLE_CELERY_BEAT_SCHEDULE:
         "retry-pending-messages": {
             "task": "core.mda.outbound_tasks.retry_messages_task",
             "schedule": 300.0,  # Every 5 minutes (300 seconds)
-            "options": {"queue": "default"},
+            "options": {"queue": "outbound"},
         },
         "selfcheck": {
             "task": "core.mda.outbound_tasks.selfcheck_task",
             "schedule": settings.MESSAGES_SELFCHECK_INTERVAL,
-            "options": {"queue": "default"},
+            "options": {"queue": "outbound"},
         },
         "process-inbound-messages-queue": {
             "task": "core.mda.inbound_tasks.process_inbound_messages_queue_task",
             "schedule": 300.0,  # Every 5 minutes
-            "options": {"queue": "default"},
+            "options": {"queue": "inbound"},
         },
     }

src/backend/messages/settings.py

@@ -594,6 +594,22 @@ class Base(Configuration):
     CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 * 30  # 30 days
     CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"
 
+    # Default queue for tasks without explicit routing
+    CELERY_TASK_DEFAULT_QUEUE = "default"
+
+    # Queue routing - tasks are routed to specific queues based on their type
+    # Priority order: management > inbound > outbound > default > imports > reindex
+    CELERY_TASK_ROUTES = {
+        # Inbound email processing - highest priority, time-sensitive
+        "core.mda.inbound_tasks.*": {"queue": "inbound"},
+        # Outbound email sending - high priority
+        "core.mda.outbound_tasks.*": {"queue": "outbound"},
+        # Import tasks - lower priority than regular tasks
+        "core.services.importer.tasks.*": {"queue": "imports"},
+        # Search indexing - lowest priority, can be delayed
+        "core.services.search.tasks.*": {"queue": "reindex"},
+    }
+
     DISABLE_CELERY_BEAT_SCHEDULE = values.BooleanValue(
         default=False, environ_name="DISABLE_CELERY_BEAT_SCHEDULE", environ_prefix=None
     )

src/backend/worker.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python
+"""
+Background task worker with sensible queue defaults.
+
+Usage:
+    python worker.py                    # Process all queues with default priority
+    python worker.py --queues=inbound,default  # Process only specific queues
+    python worker.py --exclude=reindex  # Process all queues except reindex
+    python worker.py --concurrency=4    # Set worker concurrency
+    python worker.py --disable-scheduler  # Disable the scheduler
+
+Queue priority order (highest to lowest):
+    1. management - Admin/management tasks (migrations, cleanup)
+    2. inbound    - Inbound email processing (time-sensitive)
+    3. outbound   - Outbound email sending
+    4. default    - General tasks
+    5. imports    - File import processing (can be delayed)
+    6. reindex    - Search indexing (lowest priority)
+"""
+
+import argparse
+import logging
+import os
+import sys
+
+# Setup Django before importing the task runner
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "messages.settings")
+os.environ.setdefault("DJANGO_CONFIGURATION", "Development")
+
+from configurations.importer import install  # pylint: disable=wrong-import-position
+
+install(check_options=True)
+
+from messages.celery_app import app  # pylint: disable=wrong-import-position
+
+# Queue definitions in priority order
+ALL_QUEUES = ["management", "inbound", "outbound", "default", "imports", "reindex"]
+DEFAULT_QUEUES = ALL_QUEUES  # By default, process all queues
+
+
+def get_default_concurrency():
+    """Get default concurrency from environment variables."""
+    env_value = os.environ.get("WORKER_CONCURRENCY") or os.environ.get(
+        "CELERY_CONCURRENCY"
+    )
+    if env_value:
+        try:
+            return int(env_value)
+        except ValueError:
+            return None
+    return None
+
+
+def parse_args():
+    """Parse command-line arguments."""
+    parser = argparse.ArgumentParser(
+        description="Start a background task worker with sensible queue defaults.",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=__doc__,
+    )
+    parser.add_argument(
+        "--queues",
+        "-Q",
+        type=str,
+        default=None,
+        help=f"Comma-separated list of queues to process. Default: {','.join(DEFAULT_QUEUES)}",
+    )
+    parser.add_argument(
+        "--exclude",
+        "-X",
+        type=str,
+        default=None,
+        help="Comma-separated list of queues to exclude from processing.",
+    )
+    parser.add_argument(
+        "--concurrency",
+        "-c",
+        type=int,
+        default=get_default_concurrency(),
+        help="Number of worker processes. Default: WORKER_CONCURRENCY env var or number of CPUs.",
+    )
+    parser.add_argument(
+        "--disable-scheduler",
+        action="store_true",
+        help="Disable the task scheduler (enabled by default).",
+    )
+    parser.add_argument(
+        "--loglevel",
+        "-l",
+        type=str,
+        default="INFO",
+        help="Logging level. Default: INFO",
+    )
+    return parser.parse_args()
+
+
+def main():
+    """Start the background task worker."""
+    logger = logging.getLogger(__name__)
+    args = parse_args()
+
+    # Determine which queues to process
+    if args.queues:
+        queues = [q.strip() for q in args.queues.split(",")]
+        # Validate queues
+        invalid = set(queues) - set(ALL_QUEUES)
+        if invalid:
+            sys.stderr.write(f"Error: Unknown queues: {', '.join(invalid)}\n")
+            sys.stderr.write(f"Valid queues are: {', '.join(ALL_QUEUES)}\n")
+            sys.exit(1)
+    else:
+        queues = DEFAULT_QUEUES.copy()
+
+    # Apply exclusions
+    if args.exclude:
+        exclude = [q.strip() for q in args.exclude.split(",")]
+        # Validate excluded queue names
+        invalid_exclude = set(exclude) - set(ALL_QUEUES)
+        if invalid_exclude:
+            sys.stderr.write(
+                f"Error: Unknown queues to exclude: {', '.join(invalid_exclude)}\n"
+            )
+            sys.stderr.write(f"Valid queues are: {', '.join(ALL_QUEUES)}\n")
+            sys.exit(1)
+        queues = [q for q in queues if q not in exclude]
+
+    if not queues:
+        sys.stderr.write("Error: No queues to process after exclusions.\n")
+        sys.exit(1)
+
+    # Build worker arguments
+    worker_args = [
+        "worker",
+        f"--queues={','.join(queues)}",
+        f"--loglevel={args.loglevel}",
+    ]
+
+    if args.concurrency:
+        worker_args.append(f"--concurrency={args.concurrency}")
+
+    if not args.disable_scheduler:
+        worker_args.append("--beat")
+
+    # Always enable task events for monitoring
+    worker_args.append("--task-events")
+
+    logger.info("Starting worker with queues: %s", ", ".join(queues))
+    app.worker_main(argv=worker_args)
+
+
+if __name__ == "__main__":
+    main()

src/e2e/compose.yaml

@@ -13,10 +13,10 @@ services:
       service: redis
     ports: !reset []
 
-  celery:
+  worker:
     extends:
       file: ../../compose.yaml
-      service: celery-dev
+      service: worker-dev
     environment:
       - DJANGO_CONFIGURATION=E2E
     env_file:
@@ -171,7 +171,7 @@ services:
         condition: service_started
       mta-in:
         condition: service_started
-      celery:
+      worker:
         condition: service_started
     command: npm run test
     ports: