perf: keep hot-path DynamoDB calls off the event loop

app/api/messages.py

@@ -536,7 +536,7 @@ def _end_trace_spans(error=None):
                     )
 
                 # Record usage
-                usage_tracker.record_usage(
+                usage_tracker.record_usage_nowait(
                     api_key=api_key_info.get("api_key"),
                     request_id=request_id,
                     model=request_data.model,
@@ -632,7 +632,7 @@ def _end_trace_spans(error=None):
                 )
 
                 # Record usage
-                usage_tracker.record_usage(
+                usage_tracker.record_usage_nowait(
                     api_key=api_key_info.get("api_key"),
                     request_id=request_id,
                     model=request_data.model,
@@ -716,7 +716,7 @@ async def _web_search_stream_with_usage():
                         ws_error = str(e)
                         raise
                     finally:
-                        usage_tracker.record_usage(
+                        usage_tracker.record_usage_nowait(
                             api_key=api_key_info.get("api_key"),
                             request_id=request_id,
                             model=request_data.model,
@@ -746,7 +746,7 @@ async def _web_search_stream_with_usage():
                 )
 
                 # Record usage
-                usage_tracker.record_usage(
+                usage_tracker.record_usage_nowait(
                     api_key=api_key_info.get("api_key"),
                     request_id=request_id,
                     model=request_data.model,
@@ -824,7 +824,7 @@ async def _web_fetch_stream_with_usage():
                         wf_error = str(e)
                         raise
                     finally:
-                        usage_tracker.record_usage(
+                        usage_tracker.record_usage_nowait(
                             api_key=api_key_info.get("api_key"),
                             request_id=request_id,
                             model=request_data.model,
@@ -854,7 +854,7 @@ async def _web_fetch_stream_with_usage():
                 )
 
                 # Record usage
-                usage_tracker.record_usage(
+                usage_tracker.record_usage_nowait(
                     api_key=api_key_info.get("api_key"),
                     request_id=request_id,
                     model=request_data.model,
@@ -959,7 +959,7 @@ async def _web_fetch_stream_with_usage():
                 else:
                     result = await provider.invoke(request_data, target_model, api_key_info)
                     # Record usage
-                    usage_tracker.record_usage(
+                    usage_tracker.record_usage_nowait(
                         api_key=api_key_info.get("api_key"),
                         request_id=request_id,
                         model=target_model,
@@ -1060,7 +1060,7 @@ async def _web_fetch_stream_with_usage():
                 _turn_span.end()
 
             # Record usage
-            usage_tracker.record_usage(
+            usage_tracker.record_usage_nowait(
                 api_key=api_key_info.get("api_key"),
                 request_id=request_id,
                 model=request_data.model,
@@ -1090,7 +1090,7 @@ async def _web_fetch_stream_with_usage():
         print(f"[ERROR] HTTP Status: {e.http_status}")
         print(f"[ERROR] Error Type: {e.error_type}\n")
 
-        usage_tracker.record_usage(
+        usage_tracker.record_usage_nowait(
             api_key=api_key_info.get("api_key"),
             request_id=request_id,
             model=request_data.model,
@@ -1117,7 +1117,7 @@ async def _web_fetch_stream_with_usage():
             exc_info=True,
         )
 
-        usage_tracker.record_usage(
+        usage_tracker.record_usage_nowait(
             api_key=api_key_info.get("api_key"),
             request_id=request_id,
             model=request_data.model,
@@ -1379,7 +1379,7 @@ async def _handle_streaming_request(
 
     finally:
         # Record usage after stream completes
-        usage_tracker.record_usage(
+        usage_tracker.record_usage_nowait(
             api_key=api_key_info.get("api_key"),
             request_id=request_id,
             model=request_data.model,

app/api/openai_passthrough/router.py

@@ -204,7 +204,7 @@ def _record_usage(
     _, usage, _ = _managers()
     norm = normalize_usage(raw_usage, api_surface)
     try:
-        usage.record_usage(
+        usage.record_usage_nowait(
             api_key=api_key_info.get("api_key", ""),
             request_id=str(uuid4()),
             model=model,

app/core/config.py

@@ -124,6 +124,15 @@ class Settings(BaseSettings):
     api_key_header: str = Field(default="x-api-key", alias="API_KEY_HEADER")
     require_api_key: bool = Field(default=True, alias="REQUIRE_API_KEY")
     master_api_key: Optional[str] = Field(default=None, alias="MASTER_API_KEY")
+    api_key_cache_ttl_seconds: int = Field(
+        default=60, alias="API_KEY_CACHE_TTL_SECONDS",
+        description=(
+            "TTL in seconds for the in-process API key validation cache "
+            "(0 to disable). Avoids a DynamoDB read per request; key "
+            "changes (create/disable) made in another process take up to "
+            "this long to apply on running workers."
+        )
+    )
 
     # Rate Limiting Settings
     rate_limit_enabled: bool = Field(default=True, alias="RATE_LIMIT_ENABLED")
@@ -290,6 +299,15 @@ class Settings(BaseSettings):
                     "inference profile ARNs to their underlying foundation model ID.",
     )
 
+    # Model Mapping Cache
+    model_mapping_cache_ttl_seconds: int = Field(
+        default=300,
+        alias="MODEL_MAPPING_CACHE_TTL_SECONDS",
+        description="TTL (seconds) for the in-process model mapping cache "
+                    "(0 to disable). Mapping changes made in another process "
+                    "(e.g. the admin portal) take up to this long to apply.",
+    )
+
     # Beta features that require InvokeModel API instead of Converse API
     # These features are only available via InvokeModel/InvokeModelWithResponseStream
     beta_headers_requiring_invoke_model: List[str] = Field(

app/core/metrics.py

@@ -3,11 +3,10 @@
 
 Provides Prometheus-compatible metrics for monitoring.
 """
-from prometheus_client import Counter, Histogram, Gauge, Info
-from typing import Optional
 
-from app.core.config import settings
+from prometheus_client import Counter, Gauge, Histogram, Info
 
+from app.core.config import settings
 
 # Request metrics
 request_counter = Counter(
@@ -69,6 +68,12 @@
     ["api_key"],
 )
 
+# Usage tracking metrics
+usage_writes_dropped_counter = Counter(
+    "usage_writes_dropped_total",
+    "Usage/billing rows dropped because the background write backlog was full",
+)
+
 # Authentication metrics
 auth_failures_counter = Counter(
     "auth_failures_total",

app/core/ttl_cache.py

@@ -0,0 +1,71 @@
+"""Minimal thread-safe TTL cache shared by hot-path lookups.
+
+Used to keep per-request DynamoDB reads (API key validation, model
+mapping) off the request path. Deliberately small instead of pulling in
+cachetools: get/set/invalidate with per-entry expiry and a hard entry
+bound so client-controlled keys cannot grow memory without limit.
+"""
+import threading
+import time
+from typing import Any
+
+
+class TTLCache:
+    """Thread-safe key/value cache with per-entry TTL and a size bound.
+
+    ``get`` returns ``(hit, value)`` so a cached ``None`` (negative
+    caching) is distinguishable from a miss.
+
+    Eviction: when full, expired entries are purged first; if the cache
+    is still full, the soonest-expiring entries are dropped. Cache keys
+    are client-controlled input, and negative entries carry the shortest
+    TTLs — so spam evicts itself before it can evict hot positive
+    entries. Entries are cheap to recompute (single DynamoDB reads), so
+    bounded memory matters more than hit-rate precision.
+    """
+
+    def __init__(self, max_entries: int = 10_000):
+        self._max_entries = max_entries
+        self._entries: dict[Any, tuple[Any, float]] = {}
+        self._lock = threading.Lock()
+
+    def get(self, key: Any) -> tuple[bool, Any]:
+        with self._lock:
+            entry = self._entries.get(key)
+            if entry is None:
+                return False, None
+            value, expires_at = entry
+            if time.monotonic() >= expires_at:
+                del self._entries[key]
+                return False, None
+            return True, value
+
+    def set(self, key: Any, value: Any, ttl_seconds: float) -> None:
+        with self._lock:
+            if key not in self._entries and len(self._entries) >= self._max_entries:
+                self._evict_locked()
+            self._entries[key] = (value, time.monotonic() + ttl_seconds)
+
+    def invalidate(self, key: Any) -> None:
+        with self._lock:
+            self._entries.pop(key, None)
+
+    def clear(self) -> None:
+        with self._lock:
+            self._entries.clear()
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._entries)
+
+    def _evict_locked(self) -> None:
+        """Purge expired entries; drop the soonest-expiring if still full."""
+        now = time.monotonic()
+        expired = [k for k, (_, exp) in self._entries.items() if now >= exp]
+        for k in expired:
+            del self._entries[k]
+        if len(self._entries) >= self._max_entries:
+            n_drop = max(1, len(self._entries) // 10)
+            by_expiry = sorted(self._entries.items(), key=lambda kv: kv[1][1])
+            for k, _ in by_expiry[:n_drop]:
+                del self._entries[k]