Background Workers (Celery)
Relevant source files
The following files were used as context for generating this wiki page:
Onyx utilizes Celery as its primary asynchronous task processing engine to handle resource-intensive and long-running operations outside of the main API request/response cycle. These background workers are responsible for critical system functions, including document fetching from external connectors, processing and indexing data into the Vespa vector database, system monitoring, and periodic maintenance tasks.
The architecture is modular, employing multiple specialized worker types that subscribe to specific queues to ensure stability, scalability, and task isolation. All workers utilize thread-based pools rather than processes to maintain stability within the environment.
Sources: AGENTS.md:23-25, backend/onyx/background/README.md:1-7
Architecture and Worker Types
Onyx distributes background work across several specialized worker instances. Each worker type is optimized for specific workloads, ranging from lightweight metadata updates to heavy document pruning.
Specialized Worker Overview
| Worker Type | Primary Responsibility | Concurrency / Configuration |
|---|---|---|
| Primary | Coordinates core tasks, manages connectors, and dispatches indexing jobs. Acts as a singleton. | 4 threads |
| Docfetching | Pulls documents from external APIs (Google Drive, Confluence, etc.) and batches them. | Configurable; 2 threads (Helm default) |
| Docprocessing | Chunks, embeds, and indexes fetched documents into Vespa. | Configurable; 6 threads (Helm default) |
| Light | Fast, short-lived tasks like permission syncs and document deletions. | High (up to 24 workers/192 tasks) |
| Heavy | Resource-intensive operations like pruning and CSV generation. | Low (4 threads max) |
| Beat | The scheduler that generates per-tenant periodic task schedules. | N/A |
Sources: AGENTS.md:30-85, backend/onyx/background/README.md:13-30, deployment/helm/charts/onyx/templates/celery-worker-docfetching.yaml:85, deployment/helm/charts/onyx/templates/celery-worker-docprocessing.yaml:86
System Data Flow
The following diagram illustrates the typical flow of data from an external connector through the background processing pipeline.
This diagram shows the asynchronous transition from scheduling a check to fetching data and finally indexing it for search.
Sources: backend/onyx/background/README.md:58-65, AGENTS.md:43-55
The Primary Worker (Coordinator)
The Primary Worker serves as the central dispatcher for the system. It is a singleton worker, a state ensured by a PRIMARY_WORKER Redis lock. If a worker cannot acquire this lock, it will not process the main coordination tasks.
Core Responsibilities
- Initialization: On startup, it verifies the health of Redis, Postgres, and the Document Index. It also cleans up orphaned indexing attempts.
- Task Dispatching: It scans for connectors that need work and dispatches them to specialized queues.
- Maintenance: Periodically cleans up old indexing checkpoints and handles user file synchronization.
Sources: backend/onyx/background/README.md:35-48
Periodic Schedule (Primary/Beat)
| Task | Frequency | Target Queue |
|---|---|---|
check_for_indexing | 15s | connector_doc_fetching |
check_for_vespa_sync_task | 20s | vespa_metadata_sync |
check_for_pruning | 20s | connector_pruning |
check_for_connector_deletion | 20s | connector_deletion |
check_for_checkpoint_cleanup | 1h | checkpoint_cleanup |
Sources: backend/onyx/background/README.md:50-58, AGENTS.md:92-105
Document Indexing Pipeline
The indexing pipeline is split between Docfetching and Docprocessing to allow for independent scaling of I/O-bound (fetching) and CPU/Memory-bound (embedding/indexing) tasks.
Docfetching Worker
- Runs connector-specific logic to pull documents from external sources.
- Implements watchdog monitoring to identify and recover stuck connectors.
- Stores document batches to intermediate file storage to prevent memory overflow.
- Dispatches tasks to the Docprocessing queue once a batch is ready.
Docprocessing Worker
- PostgreSQL Sync: Upserts document metadata into the relational database.
- Transformation: Chunks documents and adds contextual information.
- Embedding: Sends chunks to the model server for vector representation.
- Vespa Sync: Writes the resulting chunks and metadata to the Vespa vector database.
Sources: AGENTS.md:43-55, backend/onyx/background/README.md:66-70
Implementation Standards
Developers contributing new background tasks or connectors must adhere to specific implementation patterns to ensure system health and multi-tenant support.
Task Definition Guidelines
- Shared Tasks: Always use the
@shared_taskdecorator rather than tying tasks to a specificcelery_appinstance. - Expiration: All tasks must be enqueued with an
expires=argument to prevent unbounded queue growth in the event of system backlogs. - Timeouts: Since workers use thread pools, standard Celery time limits are ineffective. Timeout logic must be implemented explicitly within the task code.
- Tenant Awareness: Tasks use
TenantAwareTaskto automatically handle tenant context in multi-tenant (Enterprise) environments.
Sources: AGENTS.md:118-135, backend/onyx/background/README.md:32-34
Connector Interfaces
New connectors are integrated into the background worker system by subclassing specific interfaces:
- LoadConnector: For bulk indexing at a point in time.
- PollConnector: For incremental updates based on time ranges.
- Slim Connector: For pruning jobs; it fetches only document IDs to verify existence.
Sources: backend/onyx/connectors/README.md:14-30
Summary
The background worker system in Onyx is a multi-tiered architecture powered by Celery that separates orchestration from execution. By utilizing specialized workers for fetching, processing, and maintenance, the system ensures that high-volume document ingestion does not block critical system updates or lightweight metadata synchronization. The use of Redis-based locking for the Primary worker ensures coordinated task dispatching, while thread-based concurrency provides a stable execution environment for all asynchronous operations.