Celery Workers¶

Carbon Connect uses Celery for background task processing with Valkey (Redis-compatible) as the message broker and result backend.

Sources:

• backend/app/worker/celery_app.py -- Application configuration
• backend/app/worker/celery_beat.py -- Periodic task schedules
• backend/app/worker/async_utils.py -- Async bridge utility
• backend/app/tasks/grant_pipeline_tasks.py -- Pipeline tasks

Queue Architecture¶

The system uses four dedicated queues with direct exchange routing:

Task Routing¶

task_routes = {
    # Matching tasks
    "backend.app.worker.tasks.matching_tasks.*": {"queue": "matching"},
    # Email tasks
    "backend.app.worker.tasks.email_tasks.*": {"queue": "email"},
    # Sync tasks
    "backend.app.worker.tasks.sync_tasks.*": {"queue": "sync"},
    "backend.app.tasks.grant_pipeline_tasks.*": {"queue": "sync"},
}

Worker Configuration¶

Key Settings¶

Retry Policies¶

Each queue has a specific retry policy:

{
    "max_retries": 3,
    "retry_backoff": True,
    "retry_backoff_max": 300,  # 5 minutes
}

{
    "max_retries": 5,
    "retry_backoff": True,
    "retry_backoff_max": 1800,  # 30 minutes
}

{
    "max_retries": 3,
    "retry_backoff": True,
    "retry_backoff_max": 600,  # 10 minutes
}

Running Workers¶

Start All Queues¶

poetry run celery -A backend.app.worker.celery_app worker -l info

Start Specific Queue¶

# Matching only
poetry run celery -A backend.app.worker.celery_app worker -Q matching -l info

# Email only
poetry run celery -A backend.app.worker.celery_app worker -Q email -l info

# Sync only
poetry run celery -A backend.app.worker.celery_app worker -Q sync -l info

Start Beat Scheduler¶

poetry run celery -A backend.app.worker.celery_app beat -l info

Start Flower Monitoring¶

poetry run celery -A backend.app.worker.celery_app flower --port=5555

Celery Beat Schedule¶

All scheduled tasks are defined in backend/app/worker/celery_beat.py.

Grant Synchronization¶

Grant Maintenance¶

Matching¶

Notifications¶

Maintenance¶

Async Bridge¶

Source: backend/app/worker/async_utils.py

Celery tasks are synchronous, but the application code uses async (SQLAlchemy AsyncSession, httpx). The run_async() utility bridges this gap:

from backend.app.worker.async_utils import run_async

def run_async(coro_factory: Callable[[], Coroutine]) -> T:
    """
    Run an async coroutine from synchronous Celery task code.

    Handles two environments:
    - No running loop: calls asyncio.run() directly
    - Running loop exists: runs in a dedicated thread with its own loop
    """

This is used in all pipeline tasks:

@shared_task(bind=True)
def sync_cordis_grants(self, ...):
    async def _fetch():
        async with CordisScraper(config) as scraper:
            return await scraper.fetch_all_projects(...)

    projects = run_async(_fetch)

Task Modules¶

Grant Pipeline Tasks¶

backend/app/tasks/grant_pipeline_tasks.py

Worker Tasks¶

Configuration¶