Document Indexing Operations#

Primary file: api/services/dataset_service.py — DocumentService class 1

This article covers the lifecycle management layer for knowledge base documents: state transitions (retry/pause/recover), batch enable/disable/archive, Redis-based distributed locking, and async Celery task dispatch. For the multi-stage ingestion pipeline (Extract → Clean → Segment → Index), see Knowledge Base Document Processing. For per-tenant Redis FIFO queuing, see Tenant-Isolated Document Indexing Queue.

Retry / Pause / Recover State Management#

Four operations in DocumentService share a common pattern: check Redis flag → mutate DB → set flag → dispatch Celery task. The Redis flag acts as an in-flight guard; a second concurrent call while the flag is live raises immediately. 1

retry_document() 1 atomically acquires a 600-second Redis lock for every document in the batch before mutating any state. If any lock acquisition fails (blocking=False), all already-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 in a single transaction. On DB commit failure, locks are released and the session is rolled back.

pause_document() 1 is only valid on documents in WAITING/PARSING/CLEANING/SPLITTING/INDEXING states. It sets is_paused = True and calls redis_client.setnx with no TTL — the flag persists until recover_document() explicitly deletes it.

sync_website_document() 1 sets mode: "scrape" on data_source_info, resets status to WAITING, and sets a 600-second SETEX flag before dispatching.

Batch Enable / Disable / Archive Operations#

batch_update_document_status() accepts action ∈ {"enable", "disable", "archive", "un_archive"} and follows a two-pass validate-then-commit pattern 1:

1. Validate: For each document, check if the document_{id}_indexing Redis key is set. If present, the document is actively being indexed and the call raises DocumentIndexingError. Then route to an action-specific helper:

2. Commit then dispatch: All DB mutations (enabled, archived, disabled_at) are committed in a single transaction. Only after a successful commit are Celery tasks dispatched and the document_{id}_indexing key set via SETEX with 600 s TTL. 1 Task dispatch errors are logged but do not trigger a rollback.

Key invariant: disable requires the document to already be in COMPLETED status. All other actions are idempotent — already-in-target-state documents are silently skipped. The document_{id}_indexing key is a post-commit signal, not a pre-commit lock: its presence tells a subsequent batch call that async index work is already in flight. 1

Redis-Based Distributed Locking#

redis_client.lock() (a RedisLock context manager) guards critical sections that must not run concurrently across multiple API worker processes. 1

LockNotOwnedError (raised when the lock expires before the block finishes) is caught and silently swallowed in save_document_with_dataset_id() — the caller receives an empty document list in that case. 1

Async Celery Task Dispatch#

All task dispatch happens after session.commit() to prevent tasks from running on uncommitted data. 1 2

New document creation: After the document row is committed, save_document_with_dataset_id() dispatches via proxy classes that route to the correct Celery queue by deployment edition and subscription plan: 1

• DocumentIndexingTaskProxy(tenant_id, dataset_id, document_ids).delay() — for new documents
• DuplicateDocumentIndexingTaskProxy(tenant_id, dataset_id, duplicate_document_ids).delay() — for detected duplicates

For routing details (Cloud vs. self-hosted, dataset vs. priority_dataset queues), see Tenant-Isolated Document Indexing Queue.

Full task inventory dispatched from DocumentService:

document_indexing_update_task 3 handles document reprocessing: it transitions the document to PARSING, calls IndexProcessorFactory to clean existing vector/keyword index entries (including child chunks), then performs cleanup:

• For multimodal datasets, deletes all SegmentAttachmentBinding rows for the document, then performs orphan checking: removes attachment vectors (via index_processor.clean() with with_keywords=False) and UploadFile rows only for attachments that are no longer referenced by any remaining bindings, preserving shared attachments
• Deletes all DocumentSegment rows
• After the database commit, deletes orphaned attachment storage objects with per-key error handling

The task then re-runs IndexingRunner.run() from scratch. If the clean step fails, the task recovers by re-setting PARSING state on the document and continuing. After successful completion, it optionally dispatches generate_summary_index_task for HIGH_QUALITY datasets with summary indexing enabled. 4

Document Deletion Cleanup#

clean_document_task runs on the dataset Celery queue and performs a multi-step teardown when a document is deleted. Steps execute sequentially in separate sessions, so a failure in one phase does not abort subsequent cleanup:

1. Collect metadata — fetch DocumentSegment rows, SegmentAttachmentBinding + UploadFile attachment records in a single JOIN query. 5
2. Vector/keyword index cleanup — call index_processor.clean() with with_keywords=True, delete_child_chunks=True, delete_summaries=True. Wrapped in try/except; a transient vector-backend failure logs the error but lets PG/storage cleanup continue. 6
3. PG row cleanup — delete image UploadFile rows extracted from segment content, then all DocumentSegment rows, then segment attachment UploadFile and SegmentAttachmentBinding rows, then DatasetMetadataBinding rows. 7
4. Object storage cleanup — delete image files, the source UploadFile, and multimodal attachment files from the storage backend. Each deletion is individually guarded with try/except. 8
5. Billing refresh — schedule_billing_vector_space_refresh() is called only if vector cleanup succeeded. 9

Note: There is no DB cascade from Document to DocumentSegment or SegmentAttachmentBinding. Manual deletion through the task is the only cleanup path. Deleting a knowledge base while documents are actively indexing can leave orphaned PG rows if this task runs before indexing completes.

Key Files#