From 03914fba09c4262a76e23f637411e80d4bd37b57 Mon Sep 17 00:00:00 2001 From: Andrey Date: Tue, 8 Sep 2026 13:58:39 +0300 Subject: [PATCH] =?UTF-8?q?:zap:=20perf(conversations):=20=D1=80=D0=B0?= =?UTF-8?q?=D1=81=D1=88=D0=B8=D1=84=D1=80=D0=BE=D0=B2=D0=BA=D0=B0=20=D0=B6?= =?UTF-8?q?=D0=B4=D1=91=D1=82=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B0=D0=B9=D0=B4?= =?UTF-8?q?=D0=B5=D1=80=D0=B0=20=D0=B1=D0=B5=D0=B7=20=D1=82=D1=80=D0=B0?= =?UTF-8?q?=D0=BD=D0=B7=D0=B0=D0=BA=D1=86=D0=B8=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TenantContextMiddleware заворачивает весь вызов view в транзакцию: RLS-контекст ставится через SET LOCAL и живёт ровно столько же. Удобно и даёт запросу атомарность, но пока идёт обращение наружу, запрос держит соединение из пула, а пул на процесс небольшой. Расшифровка голосового ждёт провайдера до 90 секунд (30 с таймаут на три попытки) — несколько операторов, нажавших «расшифровать», вставали поперёк всех остальных запросов процесса. Появился отказ от общей транзакции: вьюха выставляет `tenant_manages_own_transaction` и открывает tenant_atomic сама. Забытый блок не опасен — без транзакции RLS-настройка пуста и строки просто не видны, то есть ошибка проявится сразу, а не утечкой в чужую организацию. `transcribe_voice_message` разложена на prepare → run → store, и вьюха держит транзакцию только вокруг чтения и вокруг записи. Прежний вызов целиком остался для ingest, где входящее сообщение обрабатывается одной транзакцией. Тест на TransactionTestCase (обычный TestCase сам держит транзакцию на весь тест) проверяет напрямую: во время вызова провайдера in_atomic_block — False, RLS-контекст снят, результат сохранён. Остальные держатели транзакции не тронуты намеренно. В ingest_inbound обращения наружу чередуются с записью по всей длине, и общая транзакция даёт там атомарность приёма — разносить её по фазам значит менять семантику, а не рефакторить. Отправка файла и голосового оператором опирается на синхронный 502 от канала. Проверка интеграции для WEB-провайдера сама пишет в БД. Co-Authored-By: Claude Opus 5 --- .../backend/chatballs/conversations/ingest.py | 100 +++++++++++++----- .../conversations/test_voice_messages.py | 83 +++++++++++---- .../chatballs/conversations/voice_views.py | 50 +++++++-- apps/backend/chatballs/tenancy/middleware.py | 29 ++++- 4 files changed, 202 insertions(+), 60 deletions(-) diff --git a/apps/backend/chatballs/conversations/ingest.py b/apps/backend/chatballs/conversations/ingest.py index 6d7f036..b46100c 100644 --- a/apps/backend/chatballs/conversations/ingest.py +++ b/apps/backend/chatballs/conversations/ingest.py @@ -8,6 +8,7 @@ from __future__ import annotations import hashlib import logging +from dataclasses import dataclass from django.db import IntegrityError, transaction from django.utils import timezone @@ -68,38 +69,89 @@ def _history(conversation: Conversation) -> list[dict]: return [{"role": _ROLE.get(m.author_type, "user"), "content": m.text or m.transcript} for m in prior if m.text or m.transcript] -def transcribe_voice_message(channel, message: Message, *, raise_errors: bool = False) -> str: - """Стенограмма голосового через BYOK-провайдера организации; пустая строка, - если провайдер не умеет или недоступен (статус FAILED — оператор повторит кнопкой).""" +@dataclass(frozen=True, slots=True) +class TranscriptionJob: + """Всё, что нужно провайдеру, — уже прочитанное из базы и хранилища. + + Разложено на три шага (``prepare`` → ``run`` → ``store``), чтобы вызывающий + мог держать транзакцию только вокруг первого и третьего: обращение к + провайдеру ждёт ответа десятки секунд, и всё это время транзакция занимала + бы соединение из пула (chatballs.tenancy.middleware). + """ + + provider: object + model: str + audio: bytes + filename: str + content_type: str + + +def prepare_transcription(channel, message: Message) -> TranscriptionJob | None: + """Шаг в транзакции: провайдер организации, модель и байты аудио.""" from chatballs.ai.provider.factory import get_provider from chatballs.ai.provider.routing import DEFAULT_TRANSCRIPTION_MODEL, resolve_transcription_model if not message.audio: - return "" + return None + provider = get_provider(channel=channel) try: - provider = get_provider(channel=channel) - try: - model = resolve_transcription_model(channel) - except ProviderError: - model = DEFAULT_TRANSCRIPTION_MODEL # тестовый провайдер без интеграции - with message.audio.open("rb") as handle: - audio = handle.read() - transcript = provider.transcribe( - audio=audio, - filename=message.audio.name.rsplit("/", 1)[-1], - content_type=message.audio_content_type or "audio/ogg", - model=model, - ).strip() - except ProviderError as error: - logger.info("Voice transcription unavailable for message %s: %s", message.id, error) - message.transcript_status = TranscriptStatus.FAILED - message.save(update_fields=["transcript_status"]) - if raise_errors: - raise - return "" + model = resolve_transcription_model(channel) + except ProviderError: + model = DEFAULT_TRANSCRIPTION_MODEL # тестовый провайдер без интеграции + with message.audio.open("rb") as handle: + audio = handle.read() + return TranscriptionJob( + provider=provider, + model=model, + audio=audio, + filename=message.audio.name.rsplit("/", 1)[-1], + content_type=message.audio_content_type or "audio/ogg", + ) + + +def run_transcription(job: TranscriptionJob) -> str: + """Шаг без транзакции: обращение к провайдеру.""" + return job.provider.transcribe( + audio=job.audio, + filename=job.filename, + content_type=job.content_type, + model=job.model, + ).strip() + + +def store_transcription(message: Message, transcript: str) -> None: + """Шаг в транзакции: сохранить стенограмму и статус.""" message.transcript = transcript message.transcript_status = TranscriptStatus.READY if transcript else TranscriptStatus.FAILED message.save(update_fields=["transcript", "transcript_status"]) + + +def mark_transcription_failed(message: Message) -> None: + """Статус FAILED — оператор повторит кнопкой.""" + message.transcript_status = TranscriptStatus.FAILED + message.save(update_fields=["transcript_status"]) + + +def transcribe_voice_message(channel, message: Message, *, raise_errors: bool = False) -> str: + """Стенограмма голосового через BYOK-провайдера организации; пустая строка, + если провайдер не умеет или недоступен (статус FAILED — оператор повторит кнопкой). + + Три шага подряд, в транзакции вызывающего: так входящее сообщение + обрабатывается целиком (ingest_inbound). Оператору, нажавшему «расшифровать», + ждать под транзакцией незачем — там шаги разнесены (voice_views). + """ + try: + job = prepare_transcription(channel, message) + if job is None: + return "" + transcript = run_transcription(job) + except ProviderError as error: + logger.info("Voice transcription unavailable for message %s: %s", message.id, error) + mark_transcription_failed(message) + if raise_errors: + raise + return "" + store_transcription(message, transcript) return transcript diff --git a/apps/backend/chatballs/conversations/test_voice_messages.py b/apps/backend/chatballs/conversations/test_voice_messages.py index 5778ff4..37173af 100644 --- a/apps/backend/chatballs/conversations/test_voice_messages.py +++ b/apps/backend/chatballs/conversations/test_voice_messages.py @@ -1,10 +1,12 @@ from unittest import mock from django.core.files.base import ContentFile +from django.db import connections from django.core.files.uploadedfile import SimpleUploadedFile -from django.test import TestCase +from django.test import TestCase, TransactionTestCase from chatballs.testing import TenantAPIClient as APIClient +from chatballs.ai.provider.local import LocalProvider from chatballs.channels.models import Channel from chatballs.conversations.ingest import ingest_inbound from chatballs.conversations.models import ( @@ -18,7 +20,7 @@ from chatballs.conversations.models import ( ) from chatballs.conversations.transports.base import InboundMessage from chatballs.identity.bootstrap import bootstrap_owner -from chatballs.tenancy.database import tenant_atomic +from chatballs.tenancy.database import current_tenant_id, tenant_atomic from chatballs.identity.models import ( EmployeeRole, HumanUser, @@ -32,7 +34,7 @@ from chatballs.integrations.models import ( ) -class VoiceTestCase(TestCase): +class VoiceFixtureMixin: """Голосовые сообщения (дизайн-базлайн v2, кадр H): приём, отдача, расшифровка через BYOK, отправка оператором в Telegram и MAX.""" @@ -65,6 +67,30 @@ class VoiceTestCase(TestCase): voice_mime="audio/ogg", ) + def _voice_message(self) -> Message: + contact = Contact.objects.create(organization=self.organization, name="Ольга") + conversation = Conversation.objects.create( + organization=self.organization, + channel=self.channel, + connection=self.integration, + contact=contact, + ) + message = Message.objects.create( + conversation=conversation, + author_type=MessageAuthor.CONTACT, + kind=MessageKind.VOICE, + audio_content_type="audio/ogg", + duration_seconds=14, + ) + with tenant_atomic(self.organization.id): + message.audio.save("voice.ogg", ContentFile(b"OGGDATA"), save=False) + message.save(update_fields=["audio"]) + return message + + +class VoiceTestCase(VoiceFixtureMixin, TestCase): + pass + class VoiceIngestTests(VoiceTestCase): def test_inbound_voice_is_stored_and_queued_for_operator(self) -> None: @@ -98,26 +124,6 @@ class VoiceIngestTests(VoiceTestCase): class VoiceApiTests(VoiceTestCase): - def _voice_message(self) -> Message: - contact = Contact.objects.create(organization=self.organization, name="Ольга") - conversation = Conversation.objects.create( - organization=self.organization, - channel=self.channel, - connection=self.integration, - contact=contact, - ) - message = Message.objects.create( - conversation=conversation, - author_type=MessageAuthor.CONTACT, - kind=MessageKind.VOICE, - audio_content_type="audio/ogg", - duration_seconds=14, - ) - with tenant_atomic(self.organization.id): - message.audio.save("voice.ogg", ContentFile(b"OGGDATA"), save=False) - message.save(update_fields=["audio"]) - return message - def test_audio_is_served_to_visible_viewer_only(self) -> None: message = self._voice_message() response = self.client.get( @@ -308,3 +314,34 @@ class VoiceApiTests(VoiceTestCase): voice = _normalize({"update_id": 3, "message": {**base, "voice": {"file_id": "vc", "duration": 4}}}) self.assertEqual(voice.voice_file_id, "vc") self.assertEqual(voice.files, ()) + + +class TranscriptionTransactionTests(VoiceFixtureMixin, TransactionTestCase): + """TransactionTestCase намеренно: обычный TestCase сам держит транзакцию на + весь тест, и проверять под ним in_atomic_block бессмысленно.""" + + def test_provider_is_awaited_without_holding_a_transaction(self) -> None: + # Ответа провайдера ждут десятки секунд. Транзакция всё это время + # занимала бы соединение из пула, а пул на процесс небольшой: несколько + # операторов, нажавших «расшифровать», встали бы поперёк всех остальных + # запросов процесса (chatballs.tenancy.middleware). + message = self._voice_message() + seen = {} + original = LocalProvider.transcribe + + def spy(provider_self, **kwargs): + seen["in_atomic_block"] = connections["default"].in_atomic_block + seen["tenant"] = current_tenant_id() + return original(provider_self, **kwargs) + + with mock.patch.object(LocalProvider, "transcribe", spy): + response = self.client.post( + f"/api/v1/conversations/messages/{message.id}/transcribe/" + ) + + self.assertEqual(response.status_code, 200, response.content) + self.assertIs(seen["in_atomic_block"], False) + self.assertIsNone(seen["tenant"]) + # Результат при этом сохранён: вокруг записи транзакция своя. + message.refresh_from_db() + self.assertEqual(message.transcript_status, TranscriptStatus.READY) diff --git a/apps/backend/chatballs/conversations/voice_views.py b/apps/backend/chatballs/conversations/voice_views.py index 00d4899..97f3da2 100644 --- a/apps/backend/chatballs/conversations/voice_views.py +++ b/apps/backend/chatballs/conversations/voice_views.py @@ -30,6 +30,7 @@ from chatballs.conversations.selectors import conversation_for_context from chatballs.conversations.serializers import message_payload from chatballs.conversations.view_base import ConversationViewBase from chatballs.integrations.features import voice_messages_allowed +from chatballs.tenancy.database import tenant_atomic MAX_VOICE_BYTES = 10 * 1024 * 1024 ALLOWED_AUDIO_TYPES = ("audio/ogg", "audio/webm", "audio/mpeg", "audio/mp4", "audio/wav") @@ -64,26 +65,53 @@ class MessageAudioView(ConversationViewBase): class MessageTranscribeView(ConversationViewBase): required_capability = "conversations.view" + # Расшифровка идёт к провайдеру организации и ждёт ответа десятки секунд. + # Держать на это время транзакцию нельзя: вместе с ней занято соединение из + # пула, а пул на процесс небольшой — несколько операторов, нажавших + # «расшифровать», встали бы поперёк всех остальных запросов. Поэтому здесь + # транзакции открываются вручную: вокруг чтения и вокруг записи, а вызов + # провайдера остаётся между ними (chatballs.tenancy.middleware). + tenant_manages_own_transaction = True def post(self, request: Request, message_id: int) -> Response: - try: - message = _visible_message(request, message_id) - except (Message.DoesNotExist, Conversation.DoesNotExist): - return Response({"detail": "Сообщение не найдено"}, status=404) - if message.kind != MessageKind.VOICE or not message.audio: - return Response({"detail": "Это не голосовое сообщение"}, status=400) - if message.transcript_status == TranscriptStatus.READY: - return Response({"message": message_payload(message)}) + from chatballs.conversations.ingest import ( + mark_transcription_failed, + prepare_transcription, + run_transcription, + store_transcription, + ) - from chatballs.conversations.ingest import transcribe_voice_message + context = request.tenant_context + with tenant_atomic(context): + try: + message = _visible_message(request, message_id) + except (Message.DoesNotExist, Conversation.DoesNotExist): + return Response({"detail": "Сообщение не найдено"}, status=404) + if message.kind != MessageKind.VOICE or not message.audio: + return Response({"detail": "Это не голосовое сообщение"}, status=400) + if message.transcript_status == TranscriptStatus.READY: + return Response({"message": message_payload(message)}) + try: + job = prepare_transcription(message.conversation.channel, message) + except ProviderError as error: + mark_transcription_failed(message) + return Response({"detail": str(error)}, status=502) + if job is None: + return Response({"detail": "Аудио недоступно"}, status=404) try: - transcript = transcribe_voice_message(message.conversation.channel, message, raise_errors=True) + transcript = run_transcription(job) except ProviderError as error: + with tenant_atomic(context): + mark_transcription_failed(message) return Response({"detail": str(error)}, status=502) + + with tenant_atomic(context): + store_transcription(message, transcript) + payload = message_payload(message) if not transcript: return Response({"detail": "Провайдер вернул пустую расшифровку"}, status=502) - return Response({"message": message_payload(message)}) + return Response({"message": payload}) class ConversationVoiceView(ConversationViewBase): diff --git a/apps/backend/chatballs/tenancy/middleware.py b/apps/backend/chatballs/tenancy/middleware.py index eec6f33..7b04248 100644 --- a/apps/backend/chatballs/tenancy/middleware.py +++ b/apps/backend/chatballs/tenancy/middleware.py @@ -5,6 +5,7 @@ import re import uuid from django.http import Http404, HttpRequest, HttpResponse +from django.urls import Resolver404, resolve from chatballs.events.context import get_correlation_id from chatballs.identity.models import Organization, OrganizationMembership @@ -13,7 +14,20 @@ from chatballs.tenancy.database import tenant_atomic class TenantContextMiddleware: - """Resolve an authenticated membership from the organization URL UUID.""" + """Resolve an authenticated membership from the organization URL UUID. + + По умолчанию весь вызов view проходит в одной транзакции: RLS-контекст + ставится через ``SET LOCAL`` и живёт ровно столько же, сколько она. Это + удобно и даёт запросу атомарность, но у этого есть цена — пока идёт + обращение наружу (провайдер AI, мессенджер), запрос держит соединение из + пула, а пул на процесс небольшой. + + Поэтому вьюха, которая ходит наружу и умеет разложить работу на «прочитать + — сходить — записать», может выставить ``tenant_manages_own_transaction`` и + открывать ``tenant_atomic`` сама, вокруг обращений к базе. Забытый блок не + опасен: без транзакции RLS-настройка пуста и строки просто не видны — + ошибка проявится сразу, а не утечкой в чужую организацию. + """ route_kwarg = "organization_public_id" route_pattern = re.compile( @@ -57,7 +71,18 @@ class TenantContextMiddleware: membership, correlation_id=get_correlation_id(), ) - return self.get_response(request) + if not self._view_manages_own_transaction(request): + return self.get_response(request) + return self.get_response(request) + + @staticmethod + def _view_manages_own_transaction(request: HttpRequest) -> bool: + try: + view = resolve(request.path_info).func + except Resolver404: + return False + # DRF кладёт класс вьюхи в атрибут cls у результата as_view(). + return bool(getattr(getattr(view, "cls", None), "tenant_manages_own_transaction", False)) def process_view(self, request: HttpRequest, view_func, view_args, view_kwargs): public_id = view_kwargs.get(self.route_kwarg)