perf(api): memoize ProviderManager.get_configurations within request/task scope (#27299)

Incremental step for #27299 — assembling `ProviderConfigurations` runs six
independent DB queries plus a provider-factory load, costing ~1s per call in
production. A single workflow step (e.g. a Retrieval node inside an Iteration)
invokes this path many times through independently-created `ProviderManager`
instances, so per-instance caching is not enough.

- `core/provider_manager.py`: add a module-level
  `RecyclableContextVar[dict[str, ProviderConfigurations]]` and memoize the
  result in `get_configurations` for the current request/task scope, keyed by
  `tenant_id`. The cache is shared across `ProviderManager` instances because
  callers routinely spawn fresh ones via `create_plugin_provider_manager`.
- `extensions/ext_celery.py`: call `RecyclableContextVar.increment_thread_recycles()`
  inside `FlaskTask.__call__`, matching the Flask `before_request` hook in
  `app_factory.py` so per-task caches reset cleanly between Celery tasks on
  recycled worker threads.
- `tests/unit_tests/core/test_provider_manager.py`: cover the memoization (hit
  within a scope, per-tenant isolation, sharing across `ProviderManager`
  instances, reset on new scope) behind an autouse fixture that bumps the
  recycle counter for each test.

No cross-request invalidation is introduced — the cache lives only for the
current request/task, so mutations performed through subsequent requests are
always observed. This deliberately scopes the change: it targets the reported
hot path (iterative retrieval/LLM nodes) with no staleness risk.

## Verification

- uv run --group dev basedpyright -> 0 errors, 0 warnings (project-wide).
- pytest tests/unit_tests/core/test_provider_manager.py and related -> 40 passed.
- ruff check / ruff format --check -> clean.

Author: YgorLeal

api/core/provider_manager.py

@@ -3,6 +3,7 @@
 import contextlib
 from collections import defaultdict
 from collections.abc import Sequence
+from contextvars import ContextVar
 from json import JSONDecodeError
 from typing import TYPE_CHECKING, Any
 
@@ -20,6 +21,7 @@
 from sqlalchemy.orm import Session
 
 from configs import dify_config
+from contexts.wrapper import RecyclableContextVar
 from core.entities.model_entities import DefaultModelEntity, DefaultModelProviderEntity
 from core.entities.provider_configuration import ProviderConfiguration, ProviderConfigurations, ProviderModelBundle
 from core.entities.provider_entities import (
@@ -60,6 +62,28 @@
 
 _credentials_adapter: TypeAdapter[dict[str, Any]] = TypeAdapter(dict[str, Any])
 
+# Request/task-scoped memoization of `get_configurations`.
+#
+# Assembling a `ProviderConfigurations` runs six separate DB queries plus a
+# provider-factory load, costing ~1s per call in production. A single workflow
+# step (e.g. a Retrieval node inside an Iteration) can invoke this path dozens
+# of times through independently-created `ProviderManager` instances, so
+# per-instance caching is not enough — the cache must live on the execution
+# context instead.
+#
+# `RecyclableContextVar` is Dify's existing pattern for per-request/per-task
+# state. It is reset by `RecyclableContextVar.increment_thread_recycles()`,
+# which is already called in Flask `before_request` and is called per Celery
+# task in `FlaskTask.__call__` (see `extensions/ext_celery.py`).
+#
+# Skipping `_init_trial_provider_records` on a cache hit is safe because it is
+# idempotent — it checks for existing records before inserting. Callers that
+# mutate a `ProviderConfiguration` in-place within the same scope will observe
+# their own writes on subsequent reads, which matches the pre-cache behavior.
+_provider_configurations_cache: RecyclableContextVar[dict[str, ProviderConfigurations]] = RecyclableContextVar(
+    ContextVar("provider_configurations_cache")
+)
+
 
 class ProviderManager:
     """
@@ -114,6 +138,16 @@ def get_configurations(self, tenant_id: str) -> ProviderConfigurations:
         :param tenant_id:
         :return:
         """
+        try:
+            scope_cache = _provider_configurations_cache.get()
+        except LookupError:
+            scope_cache = {}
+            _provider_configurations_cache.set(scope_cache)
+
+        cached = scope_cache.get(tenant_id)
+        if cached is not None:
+            return cached
+
         # Get all provider records of the workspace
         provider_name_to_provider_records_dict = self._get_all_providers(tenant_id)
 
@@ -273,6 +307,8 @@ def get_configurations(self, tenant_id: str) -> ProviderConfigurations:
 
             provider_configurations[str(provider_id_entity)] = provider_configuration
 
+        scope_cache[tenant_id] = provider_configurations
+
         # Return the encapsulated object
         return provider_configurations
 

api/extensions/ext_celery.py

@@ -75,11 +75,15 @@ def get_celery_broker_transport_options() -> CelerySentinelTransportDict | dict[
 def init_app(app: DifyApp) -> Celery:
     class FlaskTask(Task):
         def __call__(self, *args: object, **kwargs: object) -> object:
+            from contexts.wrapper import RecyclableContextVar
             from core.logging.context import init_request_context
 
             with app.app_context():
                 # Initialize logging context for this task (similar to before_request in Flask)
                 init_request_context()
+                # Reset RecyclableContextVar-backed per-task caches, matching the
+                # Flask before_request hook in app_factory.py.
+                RecyclableContextVar.increment_thread_recycles()
                 return self.run(*args, **kwargs)
 
     broker_transport_options = get_celery_broker_transport_options()

api/tests/unit_tests/core/test_provider_manager.py

@@ -6,12 +6,20 @@
 from graphon.model_runtime.entities.model_entities import ModelType
 from pytest_mock import MockerFixture
 
+from contexts.wrapper import RecyclableContextVar
 from core.entities.provider_entities import ModelSettings
 from core.provider_manager import ProviderManager
 from models.provider import LoadBalancingModelConfig, ProviderModelSetting, TenantDefaultModel
 from models.provider_ids import ModelProviderID
 
 
+@pytest.fixture(autouse=True)
+def _reset_provider_configurations_cache() -> None:
+    # Advance the recycle counter so the per-scope memoization inside
+    # ProviderManager.get_configurations starts fresh for every test.
+    RecyclableContextVar.increment_thread_recycles()
+
+
 def _build_provider_manager(mocker: MockerFixture) -> ProviderManager:
     return ProviderManager(model_runtime=mocker.Mock())
 
@@ -563,6 +571,70 @@ def test_get_all_provider_load_balancing_configs_returns_empty_when_cached_flag_
     mock_session_cls.assert_not_called()
 
 
+def _stub_get_configurations_helpers(manager: ProviderManager, mocker: MockerFixture) -> Mock:
+    """Replace every DB-touching helper on ``manager`` with no-op stubs and return the
+    ``_get_all_providers`` mock so tests can assert how many times the assembly ran."""
+    all_providers = mocker.patch.object(manager, "_get_all_providers", return_value={})
+    mocker.patch.object(manager, "_init_trial_provider_records", return_value={})
+    mocker.patch.object(manager, "_get_all_provider_models", return_value={})
+    mocker.patch.object(manager, "_get_all_preferred_model_providers", return_value={})
+    mocker.patch.object(manager, "_get_all_provider_model_settings", return_value={})
+    mocker.patch.object(manager, "_get_all_provider_load_balancing_configs", return_value={})
+    mocker.patch.object(manager, "_get_all_provider_model_credentials", return_value={})
+    mocker.patch("core.provider_manager.ModelProviderFactory")
+    return all_providers
+
+
+def test_get_configurations_memoizes_result_within_scope(mocker: MockerFixture) -> None:
+    manager = _build_provider_manager(mocker)
+    all_providers = _stub_get_configurations_helpers(manager, mocker)
+
+    first = manager.get_configurations("tenant-id")
+    second = manager.get_configurations("tenant-id")
+
+    assert first is second
+    all_providers.assert_called_once_with("tenant-id")
+
+
+def test_get_configurations_memoization_is_per_tenant(mocker: MockerFixture) -> None:
+    manager = _build_provider_manager(mocker)
+    all_providers = _stub_get_configurations_helpers(manager, mocker)
+
+    first = manager.get_configurations("tenant-a")
+    second = manager.get_configurations("tenant-b")
+    first_again = manager.get_configurations("tenant-a")
+
+    assert first is not second
+    assert first is first_again
+    assert all_providers.call_count == 2
+
+
+def test_get_configurations_memoization_shared_across_manager_instances(mocker: MockerFixture) -> None:
+    manager_one = _build_provider_manager(mocker)
+    manager_two = _build_provider_manager(mocker)
+    one_calls = _stub_get_configurations_helpers(manager_one, mocker)
+    two_calls = _stub_get_configurations_helpers(manager_two, mocker)
+
+    first = manager_one.get_configurations("tenant-id")
+    second = manager_two.get_configurations("tenant-id")
+
+    assert first is second
+    one_calls.assert_called_once_with("tenant-id")
+    two_calls.assert_not_called()
+
+
+def test_get_configurations_memoization_reset_on_new_scope(mocker: MockerFixture) -> None:
+    manager = _build_provider_manager(mocker)
+    all_providers = _stub_get_configurations_helpers(manager, mocker)
+
+    first = manager.get_configurations("tenant-id")
+    RecyclableContextVar.increment_thread_recycles()
+    second = manager.get_configurations("tenant-id")
+
+    assert first is not second
+    assert all_providers.call_count == 2
+
+
 def test_get_all_provider_load_balancing_configs_populates_cache_and_groups_configs() -> None:
     session = Mock()
     openai_config = SimpleNamespace(provider_name="openai")