Background Tasks

The backend offloads long-running work to Celery: test execution, trace enrichment, metric evaluation, and Architect chat turns. Tasks carry tenant context (organization and user) so they run with the same isolation as an API request.

Celery configuration

The Celery app is created in celery/core.py, which applies CELERY_CONFIG from celery/config.py and auto-discovers tasks under rhesis.backend.tasks:

celery/core.py
app = Celery("rhesis")
app.conf.update(CELERY_CONFIG)
app.autodiscover_tasks(["rhesis.backend.tasks"], force=True)

Redis serves as both broker and result backend. Use rediss:// for TLS; the ssl_cert_reqs=CERT_NONE parameter connects to managed Redis services that use self-signed certificates.

.env
# Local
BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/1

# Production (Redis with TLS)
BROKER_URL=rediss://:password@redis-host:6378/0?ssl_cert_reqs=CERT_NONE
CELERY_RESULT_BACKEND=rediss://:password@redis-host:6378/1?ssl_cert_reqs=CERT_NONE

CELERY_CONFIG sets Redis-oriented defaults: result_expires=3600, result_compression="gzip", broker connection retries, and transport options with 30-second socket timeouts for TLS connections.

Base task class

Tasks inherit from BaseTask, which adds retry settings and tenant-context management. It reads organization_id and user_id from task kwargs when the task is queued, stores them in the task headers, and restores them onto the request before the task starts:

tasks/base.py
class BaseTask(Task):
    autoretry_for = (Exception,)
    max_retries = 3
    retry_backoff = True
    retry_backoff_max = 600  # 10 minutes max delay
    track_started = True

    @contextmanager
    def get_db_session(self):
        request = getattr(self, 'request', None)
        org_id = getattr(request, 'organization_id', None) or ''
        user_id = getattr(request, 'user_id', None) or ''
        project_id = getattr(request, 'project_id', None) or ''
        with get_db_with_tenant_variables(org_id, user_id, project_id) as db:
            yield db

Tenant context decorator

with_tenant_context opens a database session with the task’s tenant context and passes it to the function as db:

tasks/decorators.py
def with_tenant_context(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        with self.get_db_session() as db:
            kwargs['db'] = db
            return func(self, *args, **kwargs)
    return wrapper

Task launcher

task_launcher launches a task from a FastAPI route, pulling organization_id and user_id off current_user so callers don’t pass them explicitly:

tasks/launcher.py
def task_launcher(task: Callable, *args: Any, current_user=None, **kwargs: Any):
    if current_user is not None:
        if getattr(current_user, 'id', None) is not None:
            kwargs.setdefault('user_id', str(current_user.id))
        if getattr(current_user, 'organization_id', None) is not None:
            kwargs.setdefault('organization_id', str(current_user.organization_id))
    return task.delay(*args, **kwargs)

Writing tasks

You do not pass organization_id and user_id as explicit parameters — the context system propagates them. Use @with_tenant_context when the task needs a database session:

tasks/test_configuration.py
@app.task(base=BaseTask, name="rhesis.backend.tasks.get_test_configuration")
@with_tenant_context
def get_test_configuration(test_configuration_id: str, db=None):
    config_id = UUID(test_configuration_id)
    test_config = crud.get_test_configuration(db, test_configuration_id=config_id)
    return {"found": test_config is not None,
            "id": str(test_config.id) if test_config else None}

Without the decorator, open a session manually via self.get_db_session(), which already carries tenant context:

tasks/manual_example.py
@app.task(base=BaseTask, name="rhesis.backend.tasks.manual_db_example")
def manual_db_example():
    with manual_db_example.get_db_session() as db:
        test_sets = crud.get_test_sets(db)
        return {"test_set_count": len(test_sets)}

Launch tasks from a route with task_launcher:

routers/test_configuration.py
from rhesis.backend.tasks import task_launcher, execute_test_configuration

@router.post("/{test_configuration_id}/execute")
def execute_test_configuration_endpoint(
    test_configuration_id: UUID,
    current_user: schemas.User = Depends(require_current_user_or_token),
):
    result = task_launcher(
        execute_test_configuration,
        str(test_configuration_id),
        current_user=current_user,
    )
    return {"task_id": result.id}

Running workers

apps/worker/start.sh runs two Celery workers against the same broker:

a main worker on the celery, execution, and telemetry queues
an architect worker on the architect queue

Both use the thread pool. Concurrency and prefetch are set through environment variables (CELERY_WORKER_CONCURRENCY, CELERY_WORKER_PREFETCH_MULTIPLIER, and the CELERY_ARCHITECT_* equivalents):

start.sh
celery -A rhesis.backend.worker.app worker \
    --pool threads -n main@%h \
    --queues=celery,execution,telemetry \
    --loglevel="$CELERY_WORKER_LOGLEVEL" \
    --concurrency="$CELERY_WORKER_CONCURRENCY" \
    --prefetch-multiplier="$CELERY_WORKER_PREFETCH_MULTIPLIER" \
    --optimization=fair -E

LOG_LEVEL controls application logs; CELERY_WORKER_LOGLEVEL controls Celery’s own task-lifecycle messages and defaults to LOG_LEVEL. Set ENABLE_FLOWER=yes to run the Flower monitoring UI on port 5555.

Monitoring task status

Query a task’s status through the result backend:

routers/tasks.py
@router.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    return {
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.ready() else None,
        "error": str(result.error) if result.failed() else None,
    }

Error handling

BaseTask logs exceptions with tenant context, retries failed tasks with exponential backoff up to max_retries, and records the final error in the result backend once retries are exhausted.

For stuck tasks, broker connectivity, and tenant-context errors, see Troubleshooting.