feat(etl-uvicorn): settings-scoped cache for per-invoke derived state

A plugin consuming current_invocation_settings() builds its handler per
distinct settings payload instead of once at boot, and construction
typically does network work (model resolution, prechecks). This gives that
pattern one home next to the accessor that creates the need:
settings_cache_key digests the canonical settings JSON so secret-bearing
payloads are never raw keys, and SettingsScopedCache memoizes derived
state bounded by both size and age — age matters because state built from
since-rotated credentials must not outlive them on a quiet pod.
Stdlib-only, so the package's dependency set is unchanged.

Author: CyMule

test/api/test_settings_scoped_cache.py

@@ -0,0 +1,94 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+from unstructured_platform_plugins.invocation_settings import (
+    SettingsScopedCache,
+    settings_cache_key,
+)
+
+
+class TestSettingsCacheKey:
+    def test_key_is_insensitive_to_key_order(self):
+        assert settings_cache_key({"a": 1, "b": 2}) == settings_cache_key({"b": 2, "a": 1})
+
+    def test_key_is_sensitive_to_values(self):
+        assert settings_cache_key({"a": 1}) != settings_cache_key({"a": 2})
+
+    def test_secret_values_do_not_appear_in_the_key(self):
+        secret = "sk-super-secret-credential"
+        key = settings_cache_key({"api_key": secret})
+        assert secret not in key
+
+
+class TestSettingsScopedCache:
+    def test_second_lookup_with_same_settings_does_not_rebuild(self):
+        cache = SettingsScopedCache()
+        build = MagicMock(return_value="handler")
+
+        first = cache.get_or_build({"model": "a"}, build)
+        second = cache.get_or_build({"model": "a"}, build)
+
+        assert first == second == "handler"
+        build.assert_called_once()
+
+    def test_distinct_settings_build_distinct_values(self):
+        cache = SettingsScopedCache()
+
+        first = cache.get_or_build({"model": "a"}, lambda: object())
+        second = cache.get_or_build({"model": "b"}, lambda: object())
+
+        assert first is not second
+
+    def test_entry_expires_after_ttl(self):
+        now = [0.0]
+        cache = SettingsScopedCache(ttl_seconds=10, clock=lambda: now[0])
+        build = MagicMock(return_value="handler")
+
+        cache.get_or_build({"model": "a"}, build)
+        now[0] = 11.0
+        cache.get_or_build({"model": "a"}, build)
+
+        assert build.call_count == 2
+
+    def test_entry_survives_within_ttl(self):
+        now = [0.0]
+        cache = SettingsScopedCache(ttl_seconds=10, clock=lambda: now[0])
+        build = MagicMock(return_value="handler")
+
+        cache.get_or_build({"model": "a"}, build)
+        now[0] = 9.0
+        cache.get_or_build({"model": "a"}, build)
+
+        build.assert_called_once()
+
+    def test_size_bound_evicts_least_recently_used(self):
+        cache = SettingsScopedCache(maxsize=2)
+        builds = {name: MagicMock(return_value=name) for name in ("a", "b", "c")}
+
+        cache.get_or_build({"model": "a"}, builds["a"])
+        cache.get_or_build({"model": "b"}, builds["b"])
+        # Refresh "a" so "b" is the eviction candidate when "c" lands.
+        cache.get_or_build({"model": "a"}, builds["a"])
+        cache.get_or_build({"model": "c"}, builds["c"])
+
+        cache.get_or_build({"model": "a"}, builds["a"])
+        cache.get_or_build({"model": "b"}, builds["b"])
+
+        builds["a"].assert_called_once()
+        assert builds["b"].call_count == 2
+
+    def test_clear_forces_rebuild(self):
+        cache = SettingsScopedCache()
+        build = MagicMock(return_value="handler")
+
+        cache.get_or_build({"model": "a"}, build)
+        cache.clear()
+        cache.get_or_build({"model": "a"}, build)
+
+        assert build.call_count == 2
+
+    @pytest.mark.parametrize("kwargs", [{"ttl_seconds": 0}, {"ttl_seconds": -1}, {"maxsize": 0}])
+    def test_degenerate_bounds_are_rejected(self, kwargs):
+        with pytest.raises(ValueError):
+            SettingsScopedCache(**kwargs)

unstructured_platform_plugins/invocation_settings.py

@@ -14,12 +14,16 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import json
 import logging
-from collections.abc import Iterator
+import threading
+import time
+from collections import OrderedDict
+from collections.abc import Callable, Iterator, Mapping
 from contextlib import contextmanager
 from contextvars import ContextVar
-from typing import Any, Optional
+from typing import Any, Optional, TypeVar
 
 from fastapi import FastAPI
 from starlette.types import ASGIApp, Receive, Scope, Send
@@ -36,6 +40,8 @@
 
 logger = logging.getLogger(__name__)
 
+T = TypeVar("T")
+
 _METADATA_PATH = "/metadata"
 _INVOKE_PATH = "/invoke"
 
@@ -241,3 +247,64 @@ def install_invocation_envelope(app: FastAPI) -> None:
         return
     app.state.invocation_envelope_installed = True
     app.add_middleware(InvocationEnvelopeMiddleware)
+
+
+def settings_cache_key(invocation_settings: Mapping[str, Any]) -> str:
+    """Digest of the canonical settings JSON, safe as a cache key for secret-bearing payloads."""
+    return hashlib.sha256(json.dumps(invocation_settings, sort_keys=True).encode()).hexdigest()
+
+
+class SettingsScopedCache:
+    """Bind expensive derived state (clients, models, handlers) to the settings that built it.
+
+    A plugin consuming ``current_invocation_settings()`` builds its handler per distinct settings
+    payload instead of once at boot, and construction typically does network work (model
+    resolution, prechecks), so results are memoized keyed by ``settings_cache_key``. Both bounds
+    matter under shared tenancy: size caps how many distinct payloads stay live, and age evicts
+    state built from credentials that may since have been rotated — eviction driven only by the
+    count of distinct payloads can take arbitrarily long on a quiet pod.
+
+    Thread-safe for lookups and inserts. Concurrent misses for the same settings may build twice;
+    the extra build is wasted work, never wrong state.
+    """
+
+    def __init__(
+        self,
+        *,
+        ttl_seconds: float = 15 * 60,
+        maxsize: int = 32,
+        clock: Callable[[], float] = time.monotonic,
+    ) -> None:
+        if ttl_seconds <= 0:
+            raise ValueError("ttl_seconds must be positive")
+        if maxsize < 1:
+            raise ValueError("maxsize must be at least 1")
+        self._ttl_seconds = float(ttl_seconds)
+        self._maxsize = maxsize
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._entries: OrderedDict[str, tuple[float, Any]] = OrderedDict()
+
+    def get_or_build(self, invocation_settings: Mapping[str, Any], build: Callable[[], T]) -> T:
+        """Return the cached value for these settings, building it on a miss."""
+        key = settings_cache_key(invocation_settings)
+        now = self._clock()
+        with self._lock:
+            entry = self._entries.get(key)
+            if entry is not None:
+                expires_at, value = entry
+                if now < expires_at:
+                    self._entries.move_to_end(key)
+                    return value
+                del self._entries[key]
+        value = build()
+        with self._lock:
+            self._entries[key] = (now + self._ttl_seconds, value)
+            self._entries.move_to_end(key)
+            while len(self._entries) > self._maxsize:
+                self._entries.popitem(last=False)
+        return value
+
+    def clear(self) -> None:
+        with self._lock:
+            self._entries.clear()