Fix: Let delete dataset/document to a dedicated thread to avoid blocking othe APIs (#17800)

api/apps/restful_apis/document_api.py

@@ -71,7 +71,7 @@ from api.utils.validation_utils import (
 from common import settings
 from common.constants import ParserType, RetCode, TaskStatus, SANDBOX_ARTIFACT_BUCKET
 from common.metadata_utils import convert_conditions, meta_filter, turn2jsonschema
-from common.misc_utils import get_uuid, thread_pool_exec
+from common.misc_utils import get_uuid, thread_pool_exec, thread_pool_exec_long_time
 from api.utils.file_utils import filename_type, thumbnail
 from api.utils.file_response import apply_preview_file_response_headers
 from api.utils.web_utils import CONTENT_TYPE_MAP, html2pdf, is_valid_url, apply_safe_file_response_headers
@@ -1203,7 +1203,7 @@ async def delete_documents(tenant_id, dataset_id):
             doc_ids = unique_doc_ids
 
         # Delete documents using existing FileService.delete_docs
-        errors = await thread_pool_exec(FileService.delete_docs, doc_ids, tenant_id)
+        errors = await thread_pool_exec_long_time(FileService.delete_docs, doc_ids, tenant_id)
 
         if errors:
             return get_error_data_result(message=str(errors))

api/apps/services/dataset_api_service.py

@@ -13,7 +13,6 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
-import asyncio
 import logging
 import json
 import os
@@ -33,7 +32,7 @@ from api.db.services.tenant_model_service import TenantModelService
 from api.db.services.user_service import TenantService, UserService, UserTenantService
 from common.constants import FileSource, StatusEnum
 from api.utils.api_utils import deep_merge, get_parser_config, remap_dictionary_keys, verify_embedding_availability
-from common.misc_utils import thread_pool_exec
+from common.misc_utils import thread_pool_exec, thread_pool_exec_long_time
 from rag.advanced_rag.knowlege_compile.wiki import WIKI_PAGE_COMPILE_KWD
 
 # KB-wide structure-graph merge index types. Each (re)builds the ``dataset_graph``
@@ -222,7 +221,7 @@ def _delete_datasets_sync(tenant_id: str, ids: list = None, delete_all: bool = F
 
 
 async def delete_datasets(tenant_id: str, ids: list = None, delete_all: bool = False):
-    return await asyncio.to_thread(_delete_datasets_sync, tenant_id, ids, delete_all)
+    return await thread_pool_exec_long_time(_delete_datasets_sync, tenant_id, ids, delete_all)
 
 
 def get_dataset(dataset_id: str, tenant_id: str):

common/misc_utils.py

@@ -30,6 +30,7 @@ from urllib.parse import urljoin
 from concurrent.futures import ThreadPoolExecutor
 
 logger = logging.getLogger(__name__)
+_LONG_TIME_THREAD_POOL_EXECUTOR = ThreadPoolExecutor(max_workers=int(os.getenv("LONG_TIME_THREAD_POOL_WORKERS", "1")), thread_name_prefix="long-time")
 
 
 def get_uuid():
@@ -257,3 +258,33 @@ async def thread_pool_exec(func, *args, **kwargs):
             inner = functools.partial(func, *args, **kwargs)
             return await loop.run_in_executor(executor, ctx.run, inner)
         return await loop.run_in_executor(executor, ctx.run, func, *args)
+
+
+async def thread_pool_exec_long_time(func, *args, **kwargs):
+    """Run long blocking work in a shared bounded executor.
+
+    Use this for synchronous work that can outlive the HTTP request, such as
+    large document or dataset cleanup. Do not use ``thread_pool_exec`` for
+    those paths: it creates a temporary executor with a ``with`` block, and
+    leaving that block calls ``shutdown(wait=True)``. If the client disconnects
+    or the HTTP request times out while the worker is still running, request
+    cancellation can unwind the coroutine into that shutdown path and wait for
+    the long worker to finish anyway.
+
+    This helper uses a process-level executor instead, so there is no per-call
+    executor shutdown during request cancellation. The running sync callable is
+    still not force-cancelled by Python; it continues in the long-task pool. The
+    important behavior is that the Quart event loop/request task can be released
+    and continue serving other API calls. The pool is bounded by
+    ``LONG_TIME_THREAD_POOL_WORKERS`` (default 1), so multiple expensive jobs
+    queue instead of spawning unbounded cleanup threads or competing with the
+    event loop's default executor.
+
+    ContextVars are copied into the worker thread, matching ``thread_pool_exec``.
+    """
+    loop = asyncio.get_running_loop()
+    ctx = contextvars.copy_context()
+    if kwargs:
+        inner = functools.partial(func, *args, **kwargs)
+        return await loop.run_in_executor(_LONG_TIME_THREAD_POOL_EXECUTOR, ctx.run, inner)
+    return await loop.run_in_executor(_LONG_TIME_THREAD_POOL_EXECUTOR, ctx.run, func, *args)

test/unit_test/api/apps/services/test_dataset_api_service_list_datasets.py

@@ -169,6 +169,7 @@ def _load_list_datasets_module(monkeypatch, *, kbs, parsing_status_by_kb):
         monkeypatch,
         "common.misc_utils",
         thread_pool_exec=MagicMock(),
+        thread_pool_exec_long_time=MagicMock(),
     )
     _stub(
         monkeypatch,