65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
"""The only V2R1 ingress adapter: verified context to routed ledger record."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from threading import RLock
|
|
from typing import Callable, Mapping, Any
|
|
|
|
from .message_ledger import LedgerRecord, MessageLedger
|
|
from .router import Route, TenantRouter
|
|
from .session_pool import SessionLease, SessionPool
|
|
from .tenant_context import CanonicalMessage, TenantContext, utc_now
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IngressResult:
|
|
record: LedgerRecord
|
|
route: Route
|
|
session: SessionLease
|
|
|
|
|
|
class ChannelAdapter:
|
|
"""Fail-closed runtime ingress composition.
|
|
|
|
The adapter intentionally accepts no tenant or Cell authority outside the
|
|
signed context and TenantRouter. A process lock keeps rollback handling
|
|
deterministic for a single runtime while SQLite protects cross-process
|
|
ledger writes.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
router: TenantRouter,
|
|
ledger: MessageLedger,
|
|
session_pool: SessionPool,
|
|
signing_key: bytes,
|
|
clock: Callable[[], datetime] = utc_now,
|
|
) -> None:
|
|
self._router = router
|
|
self._ledger = ledger
|
|
self._session_pool = session_pool
|
|
self._signing_key = signing_key
|
|
self._clock = clock
|
|
self._lock = RLock()
|
|
|
|
def ingest(self, *, token: str, message: Mapping[str, Any]) -> IngressResult:
|
|
with self._lock:
|
|
context = TenantContext.verify(token, self._signing_key, now=self._clock())
|
|
canonical = CanonicalMessage.from_mapping(message)
|
|
canonical.assert_context(context)
|
|
route = self._router.assert_context_route(context)
|
|
self._router.assert_message_route(canonical)
|
|
self._session_pool.assert_available(context, route)
|
|
|
|
existed = self._ledger.matches(canonical)
|
|
record = self._ledger.append(canonical)
|
|
try:
|
|
lease = self._session_pool.acquire(context, route)
|
|
except Exception:
|
|
if not existed:
|
|
self._ledger.remove(record)
|
|
raise
|
|
return IngressResult(record=record, route=route, session=lease)
|