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.

retry_document() 1: 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() 2: 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() 3: deletes the paused flag, sets is_paused = False, then dispatches recover_document_indexing_task.delay().

sync_website_document() 4: 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:

1. Pass 1 — Validate: For each document ID, check document_{id}_indexing Redis cache; if set, the document is actively being indexed and raises DocumentIndexingError 5. 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

2. Pass 2 — Commit then dispatch: All DB field mutations (e.g., enabled, archived, disabled_at) are applied and committed in a single transaction 6. Only after a successful commit are Celery tasks dispatched and the document_{id}_indexing cache key set (SETEX with 600 s TTL) 7. 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: disable requires the document to already be in COMPLETED status 8. All other actions are idempotent — they return None if 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.

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 10 — 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 11. 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 12: After the document row is committed, save_document_with_dataset_id() calls:

• DocumentIndexingTaskProxy(tenant_id, dataset_id, document_ids).delay() — for new documents
• DuplicateDocumentIndexingTaskProxy(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:

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#