60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
"""Process-local session lease pool backed by TenantRouter authority."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from threading import RLock
|
|
|
|
from .router import Route, TenantRouter
|
|
from .tenant_context import ScopeError, TenantContext
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SessionLease:
|
|
tenant_id: str
|
|
cell_id: str
|
|
principal_id: str
|
|
session_id: str
|
|
|
|
|
|
class SessionPool:
|
|
"""Tracks active leases without creating a second session authority store."""
|
|
|
|
def __init__(self, router: TenantRouter) -> None:
|
|
self._router = router
|
|
self._leases: dict[tuple[str, str], SessionLease] = {}
|
|
self._lock = RLock()
|
|
|
|
@staticmethod
|
|
def _assert_route(context: TenantContext, route: Route) -> None:
|
|
if (
|
|
route.tenant_id != context.tenant_id
|
|
or route.cell_id != context.cell_id
|
|
or route.principal_id != context.principal_id
|
|
or route.session_id != context.session_id
|
|
):
|
|
raise ScopeError("session lease does not match authoritative route")
|
|
|
|
def assert_available(self, context: TenantContext, route: Route) -> None:
|
|
"""Re-resolve the context from the router before a ledger mutation."""
|
|
self._assert_route(context, route)
|
|
self._assert_route(context, self._router.assert_context_route(context))
|
|
|
|
def acquire(self, context: TenantContext, route: Route) -> SessionLease:
|
|
self._assert_route(context, route)
|
|
lease = SessionLease(
|
|
tenant_id=context.tenant_id,
|
|
cell_id=context.cell_id,
|
|
principal_id=context.principal_id,
|
|
session_id=context.session_id,
|
|
)
|
|
with self._lock:
|
|
return self._leases.setdefault((lease.tenant_id, lease.session_id), lease)
|
|
|
|
def release(self, lease: SessionLease) -> None:
|
|
with self._lock:
|
|
self._leases.pop((lease.tenant_id, lease.session_id), None)
|
|
|
|
def active_count(self) -> int:
|
|
with self._lock:
|
|
return len(self._leases)
|