Tenant-Isolated Document Indexing Queue#

Dify's document indexing pipeline uses a Redis-backed, per-tenant FIFO queue to prevent one tenant's heavy indexing workload from starving others. The core abstraction is TenantIsolatedTaskQueue, which manages two Redis keys per tenant:

The presence flag acts as a soft mutex: if it exists, new submissions are queued; if absent, the submission sets the flag and dispatches immediately 1.

TENANT_ISOLATED_TASK_CONCURRENCY (default 1) controls how many tasks are dequeued per handoff 2.

Dispatch Routing#

The entry point for callers is DocumentTaskProxyBase.delay(), which routes based on deployment edition 3:

Subclass BatchDocumentIndexingProxy implements the actual queue check-and-dispatch for batch submissions. Tasks are serialized as DocumentTask dataclasses (fields: tenant_id, dataset_id, document_ids) and wrapped with TaskWrapper for JSON serialization 4.

The two concrete Celery tasks that run under the queue system are:

• normal_document_indexing_task — dataset Celery queue
• priority_document_indexing_task — priority_dataset Celery queue

Duplicate-document (re-index) flows mirror the same structure via normal_duplicate_document_indexing_task and priority_duplicate_document_indexing_task, using a separate queue key "duplicate_document_indexing" 5.

Queue Handoff Mechanism#

Tenant isolation is maintained through a finally-block chain: regardless of whether indexing succeeds or raises, the completing task always checks the queue for its tenant and hands off to the next waiting task.

The handoff logic in _document_indexing_with_tenant_queue():

1. Run _document_indexing() inside a try block.
2. In finally: call pull_tasks(count=TENANT_ISOLATED_TASK_CONCURRENCY) — a Redis RPOP from the tenant's list 6.
3. If tasks exist: refresh the TTL via set_task_waiting_time() and dispatch each via task_func.apply_async() 7.
4. If queue is empty: call delete_task_key() to clear the presence flag 8.

The duplicate-document variant (_duplicate_document_indexing_task_with_tenant_queue()) follows the same pattern but uses task_func.delay() instead of apply_async with a shared producer.

TTL-Based Cleanup and Error Recovery#

The presence flag key is set via Redis SETEX with a 1-hour TTL . This is a safety net: if a Celery worker crashes mid-task before the finally block fires, the flag auto-expires and the next submission for that tenant will correctly re-enter the dispatch path rather than being silently queued forever.

Error handling within tasks:

• Billing limit violations — documents are marked IndexingStatus.ERROR immediately before any indexing begins 10.
• DocumentIsPausedError — logged at INFO level; the queue handoff still runs 11.
• All other exceptions — caught, logged, then the queue handoff still runs in finally to avoid blocking the tenant's queue 12.
• Sync task errors — document_indexing_sync_task writes IndexingStatus.ERROR with the exception message directly to the document record 13.

Bypass Paths (No Tenant Isolation)#

Four paths bypass the TenantIsolatedTaskQueue entirely and run indexing directly:

1. Event Handler (create_document_index.py)#

handle() listens on the document_index_created Blinker signal and runs IndexingRunner.run() synchronously in-process — no Redis, no Celery, no per-tenant queuing. This is typically triggered in contexts where the calling thread already manages concurrency.

2. Legacy Tasks (to be deprecated)#

document_indexing_task and duplicate_document_indexing_task accept only (dataset_id, document_ids) — no tenant_id parameter — and call the shared _document_indexing / _duplicate_document_indexing_task functions directly without any queue coordination. Both are marked TO BE DEPRECATED in their docstrings.

3. Notion Sync Task#

document_indexing_sync_task is a standalone Celery task (queue: dataset) that re-indexes a single Notion-sourced document. It accepts only (dataset_id, document_id), has no tenant_id routing argument, and performs no queue check. It includes its own early-exit if the document is already in PARSING status 14 and if Notion content is unchanged since last sync 15.

4. Self-Hosted / Enterprise Direct Dispatch#

As noted in Dispatch Routing, non-Cloud deployments call _send_to_priority_direct_queue() , which skips TenantIsolatedTaskQueue and sends directly to Celery.

Key Files Reference#