92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
"""Read-only alias registry with bounded cache and explicit invalidation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from collections.abc import Iterable
|
|
from dataclasses import dataclass
|
|
from typing import Protocol
|
|
from uuid import UUID
|
|
|
|
from .models import AliasRecord, EnvironmentName, RouterSettings
|
|
|
|
|
|
class AliasStateSource(Protocol):
|
|
def aliases(self) -> Iterable[AliasRecord]: ...
|
|
|
|
|
|
class StaticAliasSource:
|
|
"""Immutable source used by tests and bootstrap wiring; the router has no mutation API."""
|
|
|
|
def __init__(self, records: Iterable[AliasRecord]) -> None:
|
|
values = tuple(records)
|
|
identities = {(record.project_slug, record.environment) for record in values}
|
|
if len(identities) != len(values):
|
|
raise ValueError("duplicate stable endpoint alias")
|
|
self._records = values
|
|
|
|
def aliases(self) -> tuple[AliasRecord, ...]:
|
|
return self._records
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _CacheEntry:
|
|
record: AliasRecord
|
|
expires_at: float
|
|
|
|
|
|
class AliasRegistry:
|
|
def __init__(self, source: AliasStateSource, *, settings: RouterSettings | None = None) -> None:
|
|
self.source = source
|
|
self.settings = settings or RouterSettings()
|
|
self._entries: dict[tuple[str, EnvironmentName], _CacheEntry] = {}
|
|
self._lock = threading.RLock()
|
|
|
|
def resolve(self, project_slug: str, environment: EnvironmentName) -> AliasRecord | None:
|
|
key = (project_slug, environment)
|
|
now = time.monotonic()
|
|
with self._lock:
|
|
cached = self._entries.get(key)
|
|
if cached is not None and cached.expires_at > now:
|
|
return cached.record
|
|
found = next(
|
|
(
|
|
record
|
|
for record in self.source.aliases()
|
|
if record.alias == f"/mcp/{project_slug}/{environment}"
|
|
),
|
|
None,
|
|
)
|
|
if found is not None:
|
|
with self._lock:
|
|
self._entries[key] = _CacheEntry(found, now + self.settings.alias_cache_ttl_seconds)
|
|
return found
|
|
|
|
def invalidate(self, *, project_slug: str | None = None, client_id: UUID | None = None) -> None:
|
|
with self._lock:
|
|
if project_slug is None and client_id is None:
|
|
self._entries.clear()
|
|
return
|
|
keys = list(self._entries)
|
|
for key in keys:
|
|
entry = self._entries[key]
|
|
project_match = project_slug is None or entry.record.project_slug == project_slug
|
|
client_match = client_id is None or any(
|
|
binding.client_id == client_id for binding in entry.record.bindings
|
|
)
|
|
if project_match and client_match:
|
|
self._entries.pop(key, None)
|
|
|
|
def health(self) -> dict[str, int | str]:
|
|
try:
|
|
registered = len(tuple(self.source.aliases()))
|
|
except Exception:
|
|
registered = 0
|
|
with self._lock:
|
|
return {
|
|
"status": "ready" if registered else "not-ready",
|
|
"registered_aliases": registered,
|
|
"cached_aliases": len(self._entries),
|
|
}
|