Task Cancellation#

RAGFlow's sync task cancellation spans two layers: database-level status updates that mark tasks as canceled when triggered by API events, and executor-side polling that detects those flags and halts processing mid-flight. The two layers are loosely coupled — the API side writes a Redis key and DB status; the executor checks both periodically without being directly interrupted.

The cancellation flag primitive is a Redis key with the pattern {task_id}-cancel. The has_canceled() function in api/db/services/task_service.py checks for that key's presence and returns True if it exists. The custom exception TaskCanceledException (in common/exceptions.py) is raised to unwind the call stack when cancellation is detected mid-operation.

Database-Level Status Updates (API Triggers)#

Three API events cancel sync tasks by writing TaskStatus.CANCEL ("2") to the database before the executor reads the next flag:

PATCH /connectors/<id>#

update_connector() calls ConnectorService.cancel_tasks() when req.status == "CANCEL" or when req.reschedule is truthy (cancel + reschedule). The same path runs when configuration fields are changed and reschedule is set.

DELETE /connectors/<id>#

rm_connector() calls cancel_tasks() before deleting the connector row, ensuring in-flight tasks are signaled before the connector record disappears.

KB Unlink (via PATCH on KB)#

When a connector is removed from a knowledge base's linked connector list, Connector2KbService.link_connectors() issues a SyncLogsService.filter_update(...) to flip any SCHEDULE or RUNNING sync log rows for that connector+KB pair to TaskStatus.CANCEL 1.

What cancel_tasks() does#

ConnectorService.cancel_tasks() iterates over all Connector2Kb rows for the connector and calls SyncLogsService.filter_update to set status = TaskStatus.CANCEL for any row currently in SCHEDULE or RUNNING state. It also updates the connector row itself to CANCEL.

Executor-Side Cancellation Checks#

The task executor (rag/svr/task_executor.py) polls has_canceled(task_id) at multiple points during processing. The check always reads the Redis key — not the DB row — for low-latency detection.

Check 1 — Before Pickup (collect())#

After dequeuing a Redis message, collect() fetches the task from the database and immediately calls has_canceled(task["id"]). If the task is already canceled (or not found), it is ACKed and dropped without ever entering do_handle_task. This prevents wasted work on tasks that were canceled between enqueue and pickup.

Check 2 — After Model Binding (do_handle_task())#

Inside do_handle_task(), before any document processing begins, a second has_canceled check runs immediately after the embedding model is successfully bound. This catches tasks canceled during the (potentially slow) model initialization phase.

Check 3 — Pre-Insert Gate (_maybe_insert_chunks())#

A wrapper function _maybe_insert_chunks() gates every chunk insertion with a has_canceled call. If the task is canceled at this point, no chunks are written and the function returns False.

Check 4 — Per-Batch in insert_chunks()#

Inside insert_chunks(), after every bulk insert of "mother" chunks and after every DOC_BULK_SIZE batch of regular chunks, has_canceled is polled. On cancellation, it:

1. Rolls back any partially written RAPTOR summary chunks (deletes them from the doc store) to prevent false checkpoint hits on the next run.
2. Returns False to abort further inserts.

Check 5 — During LLM-Driven Sub-operations (keywords, questions, tags, metadata)#

Inside build_chunks(), each per-chunk async task for keyword extraction, question proposal, metadata generation, and content tagging checks has_canceled before calling the LLM. This prevents issuing new LLM calls once cancellation is signaled.

Check 6 — Post-Insert in do_handle_task()#

After successful chunk insertion and TOC processing, do_handle_task() performs a final has_canceled check before recording progress = 1.0.

Cancellation in the set_progress() path#

set_progress() calls has_canceled on every progress update. If canceled, it appends "[Canceled]" to the message, sets prog = -1, and then raises TaskCanceledException — propagating cancellation out of any deep call stack that uses progress_callback.

Cleanup on cancellation#

In the finally block of both do_handle_task() and TaskHandler.handle_task(), if has_canceled is True for a standard chunking task, all doc-store chunks for task_doc_id are deleted via docStoreConn.delete({"doc_id": task_doc_id}), ensuring no partial index is left behind.

Refactored Executor Path (TaskHandler)#

The refactored executor (rag/svr/task_executor_refactor/task_handler.py) mirrors the same cancellation contract through TaskHandler:

• handle() checks ctx.has_canceled_func(task_id) immediately after the task-type check, before model binding.
• Inside _run_standard_chunking_impl(), the pre-insert check and post-insert check mirror the original executor's pattern, calling abort_doc_chunking_counter in addition.
• The finally block in handle_task() deletes all doc-store entries for task_doc_id on cancellation, identical to the original path.

Which path runs is controlled by the TE_RUN_MODE environment variable 2: 0 = refactored (default), 1 = dry-run comparison, 2 = original.

Note: The refactored path (TE_RUN_MODE=0) also calls abort_doc_chunking_counter(task_doc_id) on cancellation 3, which decrements an in-memory counter used to track active chunking jobs.

Key Files and Entry Points#