341 lines
15 KiB
Python
341 lines
15 KiB
Python
"""Реестр живых сессий и подписки наблюдателей.
|
||
|
||
Дельты курсанту, полное состояние наблюдателям — сознательно: у курсанта одно
|
||
соединение и важна задержка, наблюдателей несколько и подключаются они
|
||
в произвольный момент (docs/arch/CONTRACT.md).
|
||
"""
|
||
|
||
import asyncio
|
||
import contextlib
|
||
import logging
|
||
from contextvars import ContextVar, Token
|
||
from collections.abc import AsyncIterator, Iterator
|
||
from datetime import UTC, datetime
|
||
from typing import Protocol
|
||
from uuid import UUID
|
||
|
||
from pydantic import BaseModel
|
||
|
||
from app.domain.events import (
|
||
CardReceived,
|
||
ErrorEvent,
|
||
ErrorKind,
|
||
Exercise,
|
||
StationState,
|
||
TimerTick,
|
||
)
|
||
from app.session.state import SessionState
|
||
|
||
#: Очередь одного подписчика. Медленный наблюдатель не тормозит занятие:
|
||
#: очередь ограничена, переполнение роняет соединение, а не сессию.
|
||
QUEUE_SIZE = 256
|
||
|
||
TICK_SECONDS = 1.0
|
||
LEASE_FENCED_MESSAGE = "Занятие передано другому backend-узлу; переподключитесь."
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
def _current_task():
|
||
try:
|
||
return asyncio.current_task()
|
||
except RuntimeError: # synchronous tests and tooling have no running loop
|
||
return None
|
||
|
||
|
||
class Journal(Protocol):
|
||
"""Запись в БД. Вынесена за хаб: без базы занятие должно идти,
|
||
но молчать об ошибке записи нельзя."""
|
||
|
||
async def start_lesson(
|
||
self, session_id: UUID, scenario_id: str, mode: str, trainee_name: str | None,
|
||
trainee_id: UUID | None = None, owner_login: str | None = None,
|
||
backend_node_id: str | None = None,
|
||
) -> tuple[int, UUID | None, str | None, int] | None: ...
|
||
async def utterance(self, session_id: UUID, entry) -> None: ...
|
||
async def hint(self, session_id: UUID, checklist_id: str, question: str, at) -> None: ...
|
||
async def note(self, session_id: UUID, ref: str, text: str, author: str) -> None: ...
|
||
async def self_assessment(
|
||
self, session_id: UUID, missed: list[str], comment: str, at
|
||
) -> bool: ...
|
||
async def score(self, session_id: UUID, score_auto: float, report: dict) -> bool: ...
|
||
async def score_snapshot(self, session_id: UUID, report: dict) -> None: ...
|
||
async def score_override(
|
||
self, session_id: UUID, score_final: float, author: str, comment: str,
|
||
) -> bool: ...
|
||
async def session_started(self, session_id: UUID, at) -> None: ...
|
||
async def session_ended(self, session_id: UUID, at, reason: str) -> None: ...
|
||
async def checkpoint(self, state: SessionState) -> None: ...
|
||
async def restore_active(self) -> list[SessionState]: ...
|
||
async def renew(self, session_id: UUID) -> None: ...
|
||
async def claim_expired(self, session_id: UUID | None = None) -> list[SessionState]: ...
|
||
|
||
|
||
class SessionHub:
|
||
def __init__(self, journal: Journal | None = None) -> None:
|
||
self.journal = journal
|
||
self._sessions: dict[UUID, SessionState] = {}
|
||
self._observers: dict[UUID, set[asyncio.Queue]] = {}
|
||
self._trainees: dict[UUID, set[asyncio.Queue]] = {}
|
||
self._stations: dict[UUID, set[asyncio.Queue]] = {}
|
||
self._tickers: dict[UUID, asyncio.Task] = {}
|
||
self._event_batch: ContextVar[dict | None] = ContextVar(
|
||
f"session-event-batch-{id(self)}", default=None
|
||
)
|
||
|
||
# ── реестр ──
|
||
|
||
def register(self, state: SessionState) -> SessionState:
|
||
self._sessions[state.session_id] = state
|
||
return state
|
||
|
||
def get(self, session_id: UUID) -> SessionState | None:
|
||
state = self._sessions.get(session_id)
|
||
return None if state is not None and state.lease_fenced else state
|
||
|
||
def is_lease_fenced(self, session_id: UUID) -> bool:
|
||
state = self._sessions.get(session_id)
|
||
return state is not None and state.lease_fenced
|
||
|
||
def active_sessions(self, owner_login: str) -> list[SessionState]:
|
||
"""Живые занятия только преподавателя-владельца для группового обзора."""
|
||
return [
|
||
state for state in self._sessions.values()
|
||
if not state.ended and not state.lease_fenced and state.owner_login == owner_login
|
||
]
|
||
|
||
def history(
|
||
self, *, owner_login: str | None = None, trainee_id: UUID | None = None,
|
||
mode: str | None = None, since: datetime | None = None, limit: int = 100,
|
||
) -> list[SessionState]:
|
||
"""Volatile session history for the explicit no-database demo mode."""
|
||
if since is not None and since.tzinfo is None:
|
||
since = since.replace(tzinfo=UTC)
|
||
states = [
|
||
state for state in self._sessions.values()
|
||
if not state.lease_fenced
|
||
and (owner_login is None or state.owner_login == owner_login)
|
||
and (trainee_id is None or state.trainee_id == trainee_id)
|
||
and (mode is None or state.mode.value == mode)
|
||
and (since is None or (state.started_at is not None and state.started_at >= since))
|
||
]
|
||
# Hub insertion order is creation order; completed lessons sort by
|
||
# their finish time, while unanswered calls retain their start time.
|
||
states.sort(
|
||
key=lambda state: state.ended_at or state.started_at or datetime.min.replace(tzinfo=UTC),
|
||
reverse=True,
|
||
)
|
||
return states[:max(0, limit)]
|
||
|
||
def has_active_scenario(self, scenario_id: str) -> bool:
|
||
"""Архивирование контента не должно менять уже идущее занятие."""
|
||
return any(
|
||
not state.ended and not state.lease_fenced and (
|
||
state.scenario_id == scenario_id
|
||
or any(item.id == scenario_id for item in state.dds_scenarios)
|
||
)
|
||
for state in self._sessions.values()
|
||
)
|
||
|
||
def drop(self, session_id: UUID) -> None:
|
||
self._sessions.pop(session_id, None)
|
||
self.stop_ticker(session_id)
|
||
|
||
async def checkpoint(self, session_id: UUID) -> None:
|
||
"""Зафиксировать подтверждённое состояние, если журнал доступен."""
|
||
state = self._sessions.get(session_id)
|
||
if state is not None and state.lease_fenced:
|
||
raise RuntimeError(LEASE_FENCED_MESSAGE)
|
||
if state is not None and self.journal is not None:
|
||
try:
|
||
await self.journal.checkpoint(state)
|
||
except Exception:
|
||
self._discard_event_batch(session_id)
|
||
await self.fence(state)
|
||
raise
|
||
self._flush_event_batch(session_id)
|
||
|
||
def begin_event_stream(self, session_id: UUID) -> Token:
|
||
"""Stage controller output until each explicit checkpoint in its loop."""
|
||
return self._event_batch.set({
|
||
"session_id": session_id, "events": [], "committed": False,
|
||
"persistent": True, "owner_task": _current_task(),
|
||
})
|
||
|
||
async def end_event_stream(self, token: Token) -> None:
|
||
batch = self._event_batch.get()
|
||
try:
|
||
if batch is not None and batch["events"]:
|
||
session_id = batch["session_id"]
|
||
batch["events"].clear()
|
||
state = self._sessions.get(session_id)
|
||
if self.journal is not None and state is not None and not state.lease_fenced:
|
||
await self.fence(state)
|
||
finally:
|
||
self._event_batch.reset(token)
|
||
|
||
@contextlib.asynccontextmanager
|
||
async def durable_transition(self, session_id: UUID):
|
||
"""Do not publish state-changing events until its checkpoint commits."""
|
||
batch = {
|
||
"session_id": session_id, "events": [], "committed": False,
|
||
"persistent": False, "owner_task": _current_task(),
|
||
}
|
||
token: Token = self._event_batch.set(batch)
|
||
try:
|
||
yield
|
||
if not batch["committed"]:
|
||
await self.checkpoint(session_id)
|
||
else:
|
||
self._flush_event_batch(session_id)
|
||
except Exception:
|
||
self._discard_event_batch(session_id)
|
||
state = self._sessions.get(session_id)
|
||
if self.journal is not None and state is not None and not state.lease_fenced:
|
||
await self.fence(state)
|
||
raise
|
||
finally:
|
||
self._event_batch.reset(token)
|
||
|
||
def _discard_event_batch(self, session_id: UUID) -> None:
|
||
batch = self._event_batch.get()
|
||
if batch is not None and batch["session_id"] == session_id:
|
||
batch["events"].clear()
|
||
|
||
def _flush_event_batch(self, session_id: UUID) -> None:
|
||
batch = self._event_batch.get()
|
||
if batch is None or batch["session_id"] != session_id:
|
||
return
|
||
pending, batch["events"] = batch["events"], []
|
||
batch["committed"] = not batch.get("persistent", False)
|
||
for registry, target_session_id, event in pending:
|
||
self._put(registry.get(target_session_id, set()), event)
|
||
|
||
def _send(self, registry: dict[UUID, set[asyncio.Queue]], session_id: UUID,
|
||
event: BaseModel) -> None:
|
||
batch = self._event_batch.get()
|
||
if isinstance(event, ErrorEvent):
|
||
self._put(registry.get(session_id, set()), event)
|
||
elif (batch is not None and batch["session_id"] == session_id
|
||
and batch["owner_task"] is _current_task()):
|
||
batch["events"].append((registry, session_id, event))
|
||
else:
|
||
self._put(registry.get(session_id, set()), event)
|
||
|
||
async def fence(self, state: SessionState) -> None:
|
||
"""Fail closed when durable ownership is lost or cannot be confirmed."""
|
||
if state.lease_fenced:
|
||
return
|
||
state.lease_fenced = True
|
||
self.stop_ticker(state.session_id)
|
||
if state.voice is not None:
|
||
try:
|
||
await state.voice.close()
|
||
except Exception as exc: # noqa: BLE001 — fencing must still close data channels
|
||
log.error("не удалось закрыть голос при fencing занятия %s (%s)",
|
||
state.session_id, type(exc).__name__)
|
||
event = ErrorEvent(code=ErrorKind.INTERNAL, message=LEASE_FENCED_MESSAGE)
|
||
self.broadcast(state.session_id, event)
|
||
self.to_station(state.session_id, event)
|
||
|
||
# ── подписки ──
|
||
|
||
@contextlib.contextmanager
|
||
def _subscribe(self, registry: dict[UUID, set[asyncio.Queue]], session_id: UUID) -> Iterator[asyncio.Queue]:
|
||
queue: asyncio.Queue = asyncio.Queue(maxsize=QUEUE_SIZE)
|
||
registry.setdefault(session_id, set()).add(queue)
|
||
try:
|
||
yield queue
|
||
finally:
|
||
registry.get(session_id, set()).discard(queue)
|
||
|
||
def observer(self, session_id: UUID):
|
||
return self._subscribe(self._observers, session_id)
|
||
|
||
def trainee(self, session_id: UUID):
|
||
return self._subscribe(self._trainees, session_id)
|
||
|
||
def station(self, session_id: UUID):
|
||
return self._subscribe(self._stations, session_id)
|
||
|
||
# ── вещание ──
|
||
|
||
@staticmethod
|
||
def _put(queues: set[asyncio.Queue], event: BaseModel | bytes) -> None:
|
||
for queue in list(queues):
|
||
try:
|
||
queue.put_nowait(event)
|
||
except asyncio.QueueFull:
|
||
queues.discard(queue)
|
||
|
||
def to_observers(self, session_id: UUID, event: BaseModel) -> None:
|
||
self._send(self._observers, session_id, event)
|
||
|
||
def to_trainee(self, session_id: UUID, event: BaseModel | bytes) -> None:
|
||
if isinstance(event, BaseModel):
|
||
self._send(self._trainees, session_id, event)
|
||
else:
|
||
self._put(self._trainees.get(session_id, set()), event)
|
||
|
||
def to_station(self, session_id: UUID, event: BaseModel) -> None:
|
||
self._send(self._stations, session_id, event)
|
||
|
||
def broadcast(self, session_id: UUID, event: BaseModel) -> None:
|
||
self.to_trainee(session_id, event)
|
||
self.to_observers(session_id, event)
|
||
|
||
def observer_count(self, session_id: UUID) -> int:
|
||
return len(self._observers.get(session_id, set()))
|
||
|
||
# ── такт таймеров ──
|
||
|
||
def start_ticker(self, session_id: UUID) -> None:
|
||
"""`timer.tick` раз в секунду, а не на каждое изменение: таймеров дюжина,
|
||
а UI всё равно рисует секунды."""
|
||
if session_id in self._tickers:
|
||
return
|
||
self._tickers[session_id] = asyncio.create_task(self._tick(session_id))
|
||
|
||
def stop_ticker(self, session_id: UUID) -> None:
|
||
task = self._tickers.pop(session_id, None)
|
||
if task is not None:
|
||
task.cancel()
|
||
|
||
async def shutdown(self) -> None:
|
||
"""Погасить все такты при остановке приложения.
|
||
|
||
Без этого задачи тикеров переживают выключение и держат событийный цикл:
|
||
первым это ловит не продакшен, а тест, который не может закрыть клиент.
|
||
"""
|
||
for session_id in list(self._tickers):
|
||
self.stop_ticker(session_id)
|
||
|
||
async def _tick(self, session_id: UUID) -> None:
|
||
try:
|
||
while True:
|
||
await asyncio.sleep(TICK_SECONDS)
|
||
state = self.get(session_id)
|
||
if state is None or state.ended:
|
||
return
|
||
if state.exercise is Exercise.DDS or (state.handoff_to_dds and state.dds_scenarios):
|
||
from app.session.dds import deliver_due_cards
|
||
|
||
active_before = state.dds_active_card_id
|
||
delivered = deliver_due_cards(state)
|
||
if state.dds_active_card_id != active_before and state.dds_active_card_id:
|
||
self.to_station(session_id, state.card_received_event())
|
||
if delivered:
|
||
await self.checkpoint(session_id)
|
||
# Keep the pending count and countdown live even while
|
||
# the active dispatcher card is being handled.
|
||
self.to_station(session_id, StationState(snapshot=state.station_snapshot()))
|
||
self.broadcast(session_id, TimerTick(timers=state.timers.snapshot()))
|
||
except asyncio.CancelledError:
|
||
raise
|
||
|
||
|
||
hub = SessionHub()
|
||
|
||
|
||
async def drain(queue: asyncio.Queue) -> AsyncIterator[BaseModel]:
|
||
while True:
|
||
yield await queue.get()
|