Merge branch 'main' into feat/mcp-06-18

api/core/app/apps/base_app_queue_manager.py

@@ -1,10 +1,12 @@
 import logging
 import queue
+import threading
 import time
 from abc import abstractmethod
 from enum import IntEnum, auto
 from typing import Any
 
+from cachetools import TTLCache, cachedmethod
 from redis.exceptions import RedisError
 from sqlalchemy.orm import DeclarativeMeta
 
@@ -45,6 +47,8 @@ class AppQueueManager:
         q: queue.Queue[WorkflowQueueMessage | MessageQueueMessage | None] = queue.Queue()
 
         self._q = q
+        self._stopped_cache: TTLCache[tuple, bool] = TTLCache(maxsize=1, ttl=1)
+        self._cache_lock = threading.Lock()
 
     def listen(self):
         """
@@ -157,6 +161,7 @@ class AppQueueManager:
         stopped_cache_key = cls._generate_stopped_cache_key(task_id)
         redis_client.setex(stopped_cache_key, 600, 1)
 
+    @cachedmethod(lambda self: self._stopped_cache, lock=lambda self: self._cache_lock)
     def _is_stopped(self) -> bool:
         """
         Check if task is stopped

api/core/plugin/impl/base.py

@@ -2,7 +2,7 @@ import inspect
 import json
 import logging
 from collections.abc import Callable, Generator
-from typing import Any, TypeVar
+from typing import Any, TypeVar, cast
 
 import httpx
 from pydantic import BaseModel
@@ -31,6 +31,17 @@ from core.plugin.impl.exc import (
 )
 
 plugin_daemon_inner_api_baseurl = URL(str(dify_config.PLUGIN_DAEMON_URL))
+_plugin_daemon_timeout_config = cast(
+    float | httpx.Timeout | None,
+    getattr(dify_config, "PLUGIN_DAEMON_TIMEOUT", 300.0),
+)
+plugin_daemon_request_timeout: httpx.Timeout | None
+if _plugin_daemon_timeout_config is None:
+    plugin_daemon_request_timeout = None
+elif isinstance(_plugin_daemon_timeout_config, httpx.Timeout):
+    plugin_daemon_request_timeout = _plugin_daemon_timeout_config
+else:
+    plugin_daemon_request_timeout = httpx.Timeout(_plugin_daemon_timeout_config)
 
 T = TypeVar("T", bound=(BaseModel | dict | list | bool | str))
 
@@ -58,6 +69,7 @@ class BasePluginClient:
             "headers": headers,
             "params": params,
             "files": files,
+            "timeout": plugin_daemon_request_timeout,
         }
         if isinstance(prepared_data, dict):
             request_kwargs["data"] = prepared_data
@@ -116,6 +128,7 @@ class BasePluginClient:
             "headers": headers,
             "params": params,
             "files": files,
+            "timeout": plugin_daemon_request_timeout,
         }
         if isinstance(prepared_data, dict):
             stream_kwargs["data"] = prepared_data

api/core/rag/embedding/cached_embedding.py

@@ -43,8 +43,7 @@ class CacheEmbedding(Embeddings):
             else:
                 embedding_queue_indices.append(i)
 
-        # release database connection, because embedding may take a long time
-        db.session.close()
+        # NOTE: avoid closing the shared scoped session here; downstream code may still have pending work
 
         if embedding_queue_indices:
             embedding_queue_texts = [texts[i] for i in embedding_queue_indices]