Document Indexing Operations#
Primary file: api/services/dataset_service.py β DocumentService class
This article covers the lifecycle management layer for knowledge base documents: how documents transition between indexing states, how concurrent operations are guarded, and how async work is dispatched. For the multi-stage ingestion pipeline (Extract β Clean β Segment β Index), see the Knowledge Base Document Processing article. For the per-tenant Redis FIFO queue, see Tenant-Isolated Document Indexing Queue.
Retry / Pause / Recover / Sync State Management#
These four operations in DocumentService share a common pattern: check Redis flag β mutate DB β set flag β dispatch Celery task. The flag acts as an in-flight guard; a second call while the flag is live raises immediately.
| Operation | Redis key | TTL | Redis op | Celery task |
|---|---|---|---|---|
retry_document() | document_{id}_is_retried | 600 s | Redis Lock | retry_document_indexing_task |
pause_document() | document_{id}_is_paused | None | SETNX | β (no task; pauses the running runner) |
recover_document() | document_{id}_is_paused | β | DELETE | recover_document_indexing_task |
sync_website_document() | document_{id}_is_sync | 600 s | SETEX | sync_website_document_indexing_task |
retry_document() : atomically reserves locks for all documents in the batch before changing any state. Acquires a 600-second Redis lock (redis_client.lock(retry_indexing_cache_key, timeout=600, thread_local=False).acquire(blocking=False)) for each document. If any lock acquisition fails, all previously acquired locks are released and the operation raises ValueError. Only after all locks are held does it reset each document's indexing_status to WAITING and commit the batch in a single transaction. On DB commit failure, locks are released and the session is rolled back. Finally dispatches retry_document_indexing_task.delay(dataset_id, document_ids, current_user.id). Locks are released only when owned by the current request.
pause_document() : only valid on documents in WAITING/PARSING/CLEANING/SPLITTING/INDEXING states. Sets is_paused = True in DB, then calls redis_client.setnx (no TTL β the flag persists until explicitly deleted).
recover_document() : deletes the paused flag, sets is_paused = False, then dispatches recover_document_indexing_task.delay().
sync_website_document() : sets mode: "scrape" on data_source_info, resets status to WAITING, sets a 600-second flag, dispatches sync_website_document_indexing_task.delay().
Batch Enable / Disable / Archive Operations#
batch_update_document_status() accepts action β {"enable", "disable", "archive", "un_archive"} and applies a two-pass transaction-then-dispatch pattern:
-
Pass 1 β Validate: For each document ID, check
document_{id}_indexingRedis cache; if set, the document is actively being indexed and raisesDocumentIndexingError. Delegates to_prepare_document_status_update()which routes to one of four helpers:Action Helper Celery task Sets Redis cache? enable_prepare_enable_update()add_document_to_index_taskβ Yes disable_prepare_disable_update()remove_document_from_index_taskβ Yes archive_prepare_archive_update()remove_document_from_index_task(only if enabled)Conditional un_archive_prepare_unarchive_update()add_document_to_index_task(only if enabled)Conditional -
Pass 2 β Commit then dispatch: All DB field mutations (e.g.,
enabled,archived,disabled_at) are applied and committed in a single transaction . Only after a successful commit are Celery tasks dispatched and thedocument_{id}_indexingcache key set (SETEXwith 600 s TTL) . Task dispatch errors are logged but do not trigger a rollback β the DB state is preserved and the error is re-raised after all documents are processed.
Key invariant:
disablerequires the document to already be inCOMPLETEDstatus . All other actions are idempotent β they returnNoneif the document is already in the target state, silently skipping it.
Redis-Based Distributed Locking#
redis_client.lock() (a RedisLock context manager) protects critical sections that must not run concurrently across multiple API worker processes.
| Call site | Lock key | Timeout |
|---|---|---|
save_document_with_dataset_id() | add_document_lock_dataset_id_{dataset_id} | 600 s |
create_segment() | add_segment_lock_document_id_{document_id} | 600 s |
multi_create_segment() | multi_add_segment_lock_document_id_{document_id} | 600 s |
create_child_chunk() | add_child_lock_{segment_id} | 20 s |
save_document_with_dataset_id() is the entry point for creating new documents in a dataset. The lock prevents position collisions and duplicate-detection races when multiple requests add files to the same dataset simultaneously. LockNotOwnedError (raised when the lock expires before the block finishes) is caught and silently swallowed β the caller receives an empty document list in that case.
Segment locks (create_segment, multi_create_segment) prevent position field collisions when multiple concurrent API requests append segments to the same document . The shorter 20-second child chunk lock is appropriate because child chunk creation is faster and does not require embedding computation.
Async Celery Task Dispatch#
All dispatch happens after a successful session.commit() to prevent tasks from running on uncommitted data. The main dispatch sites:
New document creation : After the document row is committed, save_document_with_dataset_id() calls:
DocumentIndexingTaskProxy(tenant_id, dataset_id, document_ids).delay()β for new documentsDuplicateDocumentIndexingTaskProxy(tenant_id, dataset_id, duplicate_document_ids).delay()β for detected duplicates
These proxy classes route to the correct Celery queue based on deployment edition and subscription plan, handling per-tenant queuing. DocumentTaskProxyBase._dispatch() checks DEPLOYMENT_EDITION == DeploymentEdition.CLOUD to determine whether to use Cloud-specific tenant-isolated queues (with priority routing for paid plans) or the simpler priority queue for self-hosted deployments. See Tenant-Isolated Document Indexing Queue for routing details.
Additional tasks dispatched from DocumentService:
| Task | Trigger |
|---|---|
add_document_to_index_task | enable / un_archive (if enabled) |
remove_document_from_index_task | disable / archive (if enabled) |
retry_document_indexing_task | explicit retry request |
recover_document_indexing_task | unpause |
sync_website_document_indexing_task | website re-sync |
batch_clean_document_task | document deletion |
document_indexing_update_task | document reprocessing |
enable_segments_to_index_task / disable_segments_from_index_task | segment enable/disable |
delete_segment_from_index_task | segment deletion |
The document_{id}_indexing Redis key (600 s TTL) used by batch_update_document_status() is a post-commit signal, not a pre-commit lock. Its presence tells the batch-update path that async index work is already in flight, preventing duplicate enable/disable tasks on the same document.
Key Files#
| File | Role |
|---|---|
api/services/dataset_service.py | DocumentService β all lifecycle operations described above |
api/tasks/document_indexing_task.py | retry_document_indexing_task, recover_document_indexing_task, normal/priority Celery tasks |
api/tasks/add_document_to_index_task.py | Re-adds a single enabled document to the vector/keyword index |
api/tasks/remove_document_from_index_task.py | Removes a single document from the index |
api/services/document_indexing_proxy/base.py | DocumentTaskProxyBase β deployment-edition-aware dispatch routing |
api/core/rag/pipeline/queue.py | TenantIsolatedTaskQueue β per-tenant Redis queue primitives |