⚡ perf(conversations): расшифровка ждёт провайдера без транзакции

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 <noreply@anthropic.com>

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
 
 

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)

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):

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)