From 77121957831d123f3055501d0b6e6edfa2077780 Mon Sep 17 00:00:00 2001 From: kennyromanov Date: Thu, 5 Nov 2026 11:59:08 +0500 Subject: [PATCH 1/2] fix: The issue when the voip becomes unhealthy after the cleanup error has been fixed --- tests/calls/test_call_cleanup.py | 193 +++++++++++++++---- tests/calls/test_call_recording.py | 50 +++++ tests/calls/test_greenapi.py | 21 +- whatsapp_chatbot_python/calls/call_runner.py | 150 +++++++------- whatsapp_chatbot_python/calls/greenapi.py | 23 ++- whatsapp_chatbot_python/calls/models.py | 4 +- whatsapp_chatbot_python/calls/recorder.py | 134 +++++++++---- whatsapp_chatbot_python/calls/service.py | 15 +- 8 files changed, 415 insertions(+), 175 deletions(-) diff --git a/tests/calls/test_call_cleanup.py b/tests/calls/test_call_cleanup.py index d8192c0..72b6229 100644 --- a/tests/calls/test_call_cleanup.py +++ b/tests/calls/test_call_cleanup.py @@ -2,6 +2,8 @@ import asyncio import unittest +from types import SimpleNamespace +from threading import Event from time import monotonic from unittest.mock import patch @@ -39,14 +41,16 @@ def voice_factory(*args, **kwargs): patcher.start() self.addCleanup(patcher.stop) - def service(self, *, setup=0.2): + def service(self, *, setup=0.2, talk=0.2): return WhatsAppCallService( api_url="https://example.invalid", id_instance="1", api_token_instance="token", openai_api_key="key", realtime_model="test", realtime_voice="marin", options=CallOptions( - setup_seconds=setup, command_seconds=setup, - ring_seconds=0.2, talk_seconds=0.2, + setup_seconds=setup, + command_seconds=setup, + ring_seconds=0.2, + talk_seconds=talk, shutdown_seconds=0.2, recording=RecordingOptions(enabled=False), ), @@ -128,62 +132,184 @@ async def broken_close(rtc): raise RuntimeError("RTC close failed") with patch.object(FakeRtc, "closeAsync", broken_close): - with self.assertRaisesRegex(RuntimeError, "RTC close failed"): + with self.assertLogs("whatsapp_chatbot_python.calls.call_runner", level="WARNING") as logs: await asyncio.wait_for(self.service().execute(session, self.fsm.apply), 0.5) self.assertTrue(self.apis[0].session_closed) self.assertTrue(self.voices[0].closed) # The remote outcome is already terminal; cleanup errors surface separately. self.assertEqual(session.state, CallState.REMOTE_ENDED) + self.assertTrue(any("WARNING!!!" in message for message in logs.output)) - async def test_rtc_close_timeout_disables_service_and_continues_later(self): + async def test_rtc_close_timeout_warns_and_allows_next_call(self): self.scenario.gates["rtc_close"] = asyncio.Event() self.scenario.remote_hangup = True session = self.session() service = self.service() - try: - with self.assertRaises(TimeoutError): + with self.assertLogs("whatsapp_chatbot_python.calls.call_runner", level="WARNING") as logs: + await asyncio.wait_for(service.execute(session, self.fsm.apply), 1) + + self.assertTrue(service.healthy) + self.assertTrue(self.apis[0].session_closed) + self.assertTrue(self.voices[0].closed) + self.assertFalse(self.apis[0].rtc.closed) + self.assertTrue(any("WARNING!!!" in message for message in logs.output)) + + self.scenario.gates["rtc_close"].set() + await asyncio.wait_for(service.execute(self.session(), self.fsm.apply), 1) + self.assertEqual(len(self.apis), 2) + self.assertTrue(self.apis[1].rtc.closed) + + async def test_remote_finish_failure_recovers_on_next_call(self): + self.scenario.remote_hangup = False + service = self.service(talk=0.02) + original_get_state = FakeApi.getStateAsync + + async def fail_first_finish(api): + if api is self.apis[0] and api.state != "idle": + raise ConnectionError("Green API temporarily unavailable") + + return await original_get_state(api) + + with patch.object(FakeApi, "getStateAsync", fail_first_finish): + with self.assertLogs("whatsapp_chatbot_python.calls.call_runner", level="WARNING") as logs: + with self.assertRaisesRegex(RuntimeError, "greenapi"): + await asyncio.wait_for(service.execute(self.session(), self.fsm.apply), 1) + + self.assertTrue(service.healthy) + self.assertTrue(self.apis[0].session_closed) + self.assertTrue(self.apis[0].rtc.closed) + self.assertTrue(any("is_remote_finish_unknown=True" in line for line in logs.output)) + + self.scenario.remote_hangup = True + await asyncio.wait_for(service.execute(self.session(), self.fsm.apply), 1) + self.assertEqual(len(self.apis), 2) + self.assertIn(("dial", "sender@c.us"), self.scenario.actions) + self.assertTrue(self.apis[1].session_closed) + + async def test_remote_finish_failure_recovers_across_call_event_loops(self): + self.scenario.remote_hangup = False + service = self.service(talk=0.02) + original_get_state = FakeApi.getStateAsync + + async def fail_first_finish(api): + if api is self.apis[0] and api.state != "idle": + raise ConnectionError("Green API temporarily unavailable") + + return await original_get_state(api) + + def execute_in_new_loop(): + return asyncio.run(service.execute(self.session(), self.fsm.apply)) + + with patch.object(FakeApi, "getStateAsync", fail_first_finish): + with self.assertRaisesRegex(RuntimeError, "greenapi"): + await asyncio.to_thread(execute_in_new_loop) + + self.scenario.remote_hangup = True + await asyncio.to_thread(execute_in_new_loop) + self.assertEqual(len(self.apis), 2) + self.assertTrue(service.healthy) + + async def test_unknown_remote_finish_marks_this_call_failed_but_keeps_service_ready(self): + self.scenario.remote_hangup = False + service = self.service(talk=0.02) + session = self.session() + original_get_state = FakeApi.getStateAsync + + async def unknown_at_finish(api): + if api.state != "idle": + return SimpleNamespace(state="on-call", info=None, reason=None) + + return await original_get_state(api) + + with patch.object(FakeApi, "getStateAsync", unknown_at_finish): + with self.assertLogs("whatsapp_chatbot_python.calls.call_runner", level="ERROR") as logs: await asyncio.wait_for(service.execute(session, self.fsm.apply), 1) - self.assertFalse(service.healthy) - self.assertTrue(self.apis[0].session_closed) - self.assertTrue(self.voices[0].closed) - self.assertFalse(self.apis[0].rtc.closed) + self.assertEqual(session.state, CallState.FAILED) + self.assertEqual(session.error_code, "finish_unknown") + self.assertTrue(service.healthy) + self.assertTrue(any("is_remote_finish_unknown=True" in line for line in logs.output)) + + async def test_next_call_starts_while_previous_recording_is_still_exporting(self): + from whatsapp_chatbot_python.calls.recorder import CallRecorder + + started, release = Event(), Event() + original_finish = CallRecorder._finish_blocking + exporting_recorders = [] + service = self.service() + + def slow_first_export(recorder): + exporting_recorders.append(recorder) + started.set() + release.wait(2) + + return original_finish(recorder) + + try: + with patch.object(CallRecorder, "_finish_blocking", slow_first_export): + with self.assertLogs("whatsapp_chatbot_python.calls.call_runner", level="ERROR") as logs: + first = await asyncio.wait_for(service.execute(self.session(), self.fsm.apply), 1) + + self.assertTrue(started.is_set()) + self.assertIsNone(first.recording_path) + self.assertTrue(any("is_recording_export_running=True" in line for line in logs.output)) + + second = await asyncio.wait_for(service.execute(self.session(), self.fsm.apply), 1) + self.assertIsNone(second.recording_path) + self.assertEqual(len(self.apis), 2) + self.assertTrue(service.healthy) finally: - self.scenario.gates["rtc_close"].set() + release.set() - await asyncio.wait_for(service._runner_with_incomplete_cleanup._resources.api._rtc_close_task, 0.5) - self.assertTrue(self.apis[0].rtc.closed) + for recorder in exporting_recorders: + await asyncio.to_thread(recorder._export.result, 1) - async def test_slow_operations_cleanup_does_not_skip_other_owners(self): + async def test_unfinished_local_close_warns_without_skipping_other_resources(self): from whatsapp_chatbot_python.calls.call_runner import Operations release = asyncio.Event() self.scenario.remote_hangup = True service = self.service() - async def delayed_close(operations): - await release.wait() + async def ignore_first_cancel(operations): + try: + await release.wait() + except asyncio.CancelledError: + await release.wait() try: - with patch.object(Operations, "close", delayed_close): - with self.assertRaises(TimeoutError): - await asyncio.wait_for( - service.execute(self.session(), self.fsm.apply), 1, - ) - - self.assertFalse(service.healthy) - self.assertFalse(self.apis[0].session_closed) - self.assertFalse(self.voices[0].closed) + with patch.object(Operations, "close", ignore_first_cancel): + with self.assertLogs("whatsapp_chatbot_python.calls.call_runner", level="WARNING") as logs: + await asyncio.wait_for(service.execute(self.session(), self.fsm.apply), 1) + + self.assertTrue(self.apis[0].session_closed) + self.assertTrue(self.voices[0].closed) + self.assertTrue(service.healthy) + self.assertTrue(any("unfinished_tasks=1" in line for line in logs.output)) finally: release.set() + await asyncio.sleep(0) + + async def test_slow_operations_cleanup_closes_other_owners_before_returning(self): + from whatsapp_chatbot_python.calls.call_runner import Operations + + release = asyncio.Event() + self.scenario.remote_hangup = True + service = self.service() + + async def delayed_close(operations): + await release.wait() - cleanup = service._runner_with_incomplete_cleanup._resources._cleanup_still_running - await asyncio.wait_for(cleanup, 0.5) + with patch.object(Operations, "close", delayed_close): + await asyncio.wait_for(service.execute(self.session(), self.fsm.apply), 1) + + self.assertTrue(service.healthy) self.assertTrue(self.apis[0].session_closed) self.assertTrue(self.apis[0].rtc.closed) self.assertTrue(self.voices[0].closed) + release.set() async def test_late_prepare_cannot_open_rtc_after_cleanup_started(self): release = asyncio.Event() @@ -202,18 +328,13 @@ async def late_voice_start(voice): try: await self.wait_action("voice_start") service.request_stop() - with self.assertRaises(TimeoutError): - await asyncio.wait_for(execution, 1) + await asyncio.wait_for(execution, 1) - self.assertFalse(service.healthy) - self.assertFalse(self.apis[0].session_closed) + self.assertFalse(service.healthy) # request_stop() is permanent. + self.assertTrue(self.apis[0].session_closed) self.assertIsNone(self.apis[0].rtc) finally: release.set() - cleanup = service._runner_with_incomplete_cleanup._resources._cleanup_still_running - await asyncio.wait_for(cleanup, 0.5) - - self.assertTrue(self.apis[0].session_closed) self.assertTrue(self.voices[0].closed) self.assertIsNone(self.apis[0].rtc) diff --git a/tests/calls/test_call_recording.py b/tests/calls/test_call_recording.py index 848b045..3777971 100644 --- a/tests/calls/test_call_recording.py +++ b/tests/calls/test_call_recording.py @@ -239,6 +239,56 @@ def slow_encode(tracks, path): exported.unlink() await recorder.close() + async def test_discarded_export_is_removed_after_call_loop_closes(self): + started, release = Event(), Event() + + with tempfile.TemporaryDirectory() as directory: + recorder = CallRecorder(RecordingOptions(directory=Path(directory))) + + def slow_encode(tracks, path): + started.set() + release.wait(2) + path.write_bytes(b"mp3 placeholder") + + def finish_in_own_loop(): + async def finish_call(): + recorder.write(Track.CALLER, PcmChunk(0, pcm(1000))) + + with self.assertRaises(TimeoutError): + await asyncio.wait_for(recorder.finish(), 0.02) + + self.assertIsNone(recorder.take_ready_or_discard()) + + asyncio.run(finish_call()) + + try: + with patch.object(recorder._encoder, "encode", side_effect=slow_encode): + await asyncio.wait_for(asyncio.to_thread(finish_in_own_loop), 0.5) + + self.assertTrue(started.is_set()) + self.assertTrue(recorder.export_running) + finally: + release.set() + + await asyncio.to_thread(recorder._export.result, 1) + self.assertFalse(recorder.export_running) + self.assertEqual(list(Path(directory).glob("*.mp3")), []) + + async def test_discard_after_worker_finishes_removes_file(self): + with tempfile.TemporaryDirectory() as directory: + recorder = CallRecorder(RecordingOptions(directory=Path(directory))) + recorder.write(Track.CALLER, PcmChunk(0, pcm(1000))) + + def encode(tracks, path): + path.write_bytes(b"mp3 placeholder") + + with patch.object(recorder._encoder, "encode", side_effect=encode): + await recorder.finish() + + recorder.discard() + self.assertEqual(list(Path(directory).glob("*.mp3")), []) + self.assertIsNone(recorder.take_ready_or_discard()) + async def test_callback_delay_does_not_insert_false_silence(self): clock = FakeClock() capture = FakeRecordingCapture() diff --git a/tests/calls/test_greenapi.py b/tests/calls/test_greenapi.py index 8f73ff4..a765bd5 100644 --- a/tests/calls/test_greenapi.py +++ b/tests/calls/test_greenapi.py @@ -110,7 +110,7 @@ async def test_provider_error_retains_details_for_caller_without_exposing_token( self.assertIn("accept unavailable", raised.exception.detail) self.assertNotIn("token", str(raised.exception)) - async def test_rtc_timeout_does_not_cancel_sdk_close(self): + async def test_rtc_timeout_cancels_sdk_close_and_closes_http_session(self): call = await self.start_outgoing() release = asyncio.Event() @@ -118,19 +118,14 @@ async def delayed_close(rtc): await release.wait() rtc.closed = True - try: - with patch.object(FakeRtc, "closeAsync", delayed_close): - with self.assertRaises(CallCleanupError) as raised: - await call.__aexit__(None, None, None) - - self.assertEqual(raised.exception.stage, "rtc_close") - self.assertTrue(self.apis[0].session_closed) - self.assertFalse(self.apis[0].rtc.closed) - finally: - release.set() - await asyncio.wait_for(call._rtc_close_task, 0.5) + with patch.object(FakeRtc, "closeAsync", delayed_close): + with self.assertRaises(CallCleanupError) as raised: + await call.__aexit__(None, None, None) - self.assertTrue(self.apis[0].rtc.closed) + self.assertEqual(raised.exception.stage, "rtc_close") + self.assertTrue(self.apis[0].session_closed) + self.assertTrue(call._rtc_close_task.done()) + self.assertFalse(self.apis[0].rtc.closed) async def test_late_rtc_open_cannot_restart_closed_adapter(self): started, release = asyncio.Event(), asyncio.Event() diff --git a/whatsapp_chatbot_python/calls/call_runner.py b/whatsapp_chatbot_python/calls/call_runner.py index 3cf1f9a..aeb02cf 100644 --- a/whatsapp_chatbot_python/calls/call_runner.py +++ b/whatsapp_chatbot_python/calls/call_runner.py @@ -6,6 +6,7 @@ import sys from collections.abc import Awaitable, Callable from contextlib import AbstractAsyncContextManager, AsyncExitStack +from dataclasses import dataclass from time import monotonic from .models import ( @@ -41,6 +42,14 @@ logger = logging.getLogger(__name__) +@dataclass(frozen=True) +class CleanupResult: + is_remote_finish_unknown: bool = False + is_local_close_unknown: bool = False + unfinished_task_count: int = 0 + is_recording_export_running: bool = False + + class Events: """Thread-safe ingress; one consumer orders observations against deadlines.""" def __init__(self) -> None: @@ -156,6 +165,10 @@ async def close(self) -> None: def begin_stop(self) -> None: self._stopping = True + @property + def pending_task_count(self) -> int: + return sum(not task.done() for task in self._tasks.values()) + async def finish_cleanup(work: Awaitable[None]) -> None: """Drain cleanup despite repeated cancellation, then propagate cancellation.""" @@ -227,8 +240,6 @@ def __init__( self.cleanup_errors: list[CallCleanupError] = [] self._pending_cleanup: set[asyncio.Task] = set() self._closing = False - self._cleanup_still_running: asyncio.Task | None = None - self._discard_recording = False async def __aenter__(self) -> CallResources: try: @@ -297,56 +308,10 @@ async def _finish_remote(self) -> None: async def _close(self) -> None: self._closing = True self.operations.begin_stop() - operations = asyncio.create_task(self.operations.close(), name="call:stop-operations") - - try: - await asyncio.wait_for(asyncio.shield(operations), self.options.shutdown_seconds) - except Exception as error: - self.cleanup_errors.append(CallCleanupError("operations_close", error)) - - self._cleanup_still_running = asyncio.create_task( - self._close_after_operations(operations), name="call:finish-late-cleanup", - ) - - self._cleanup_still_running.add_done_callback(self._observe_late_cleanup) - - raise - + await self._close_step("operations_close", self.operations.close) await self._close_owned() self._raise_cleanup_errors() - async def _close_after_operations(self, operations: asyncio.Task) -> None: - previous_errors = len(self.cleanup_errors) - try: - try: - await operations - except Exception as error: - self.cleanup_errors.append(CallCleanupError("operations_close", error)) - - await self._close_owned() - finally: - for error in self.cleanup_errors[previous_errors:]: - logger.error( - "Late call cleanup failed: session=%s stage=%s code=%s", - self._session.session_id, error.stage, error.code, - ) - - if self._discard_recording: - self.recorder.discard() - - def _observe_late_cleanup(self, task: asyncio.Task) -> None: - if task.cancelled(): - logger.error("Late call cleanup was cancelled: session=%s", self._session.session_id) - return - - error = task.exception() - - if error is not None: - logger.error( - "Late call cleanup failed: session=%s type=%s", - self._session.session_id, type(error).__name__, - ) - async def _close_owned(self) -> None: try: await self._stack.aclose() @@ -354,38 +319,66 @@ async def _close_owned(self) -> None: await self._close_step("recording_close", self._finish_recording) def _raise_cleanup_errors(self) -> None: - if self.cleanup_errors: + if self.cleanup_errors and self._is_cleanup_error_changes_call_result(): error = self.cleanup_errors[0] raise error.cause from error - def discard_recording(self) -> None: - if self._cleanup_still_running is not None and not self._cleanup_still_running.done(): - self._discard_recording = True - return + def _is_cleanup_error_changes_call_result(self) -> bool: + return not self._session.finished and self.finish_result is None + def discard_recording(self) -> None: self.recorder.discard() + @property + def cleanup_result(self) -> CleanupResult: + stages = {error.stage for error in self.cleanup_errors} + pending = sum(not task.done() for task in self._pending_cleanup) + pending += self.operations.pending_task_count + pending += int(self.api.is_rtc_close_pending) + + return CleanupResult( + is_remote_finish_unknown=( + "remote_finish" in stages or self.finish_result == FinishResult.UNKNOWN + ), + is_local_close_unknown=bool(stages - {"remote_finish"}) or pending > 0, + unfinished_task_count=pending, + is_recording_export_running=self.recorder.export_running, + ) + async def _close_step( self, stage: str, operation: Callable[..., Awaitable[None]], *args, ) -> None: task = asyncio.create_task(operation(*args), name=f"call:close:{stage}") timeout = self.options.shutdown_seconds * (2 if stage == "greenapi_close" else 1) + done, _ = await asyncio.wait({task}, timeout=timeout) - try: - await asyncio.wait_for(asyncio.shield(task), timeout) - except Exception as error: - if not task.done(): + if not done: + task.cancel() + done, _ = await asyncio.wait({task}, timeout=min(timeout, 1.0)) + + if not done: self._pending_cleanup.add(task) - task.add_done_callback(lambda done: self._observe_late_close(stage, done)) + task.add_done_callback(lambda finished: self._observe_late_close(stage, finished)) + elif not task.cancelled(): + task.exception() - if stage == "greenapi_close" and self.api.close_errors: - self.cleanup_errors.extend(self.api.close_errors) - return + self._record_close_error(stage, TimeoutError()) - if isinstance(error, CallCleanupError): - self.cleanup_errors.append(error) - return + return + try: + task.result() + except asyncio.CancelledError: + self._record_close_error(stage, RuntimeError("Close task was cancelled")) + except Exception as error: + self._record_close_error(stage, error) + + def _record_close_error(self, stage: str, error: Exception) -> None: + if stage == "greenapi_close" and self.api.close_errors: + self.cleanup_errors.extend(self.api.close_errors) + elif isinstance(error, CallCleanupError): + self.cleanup_errors.append(error) + else: self.cleanup_errors.append(CallCleanupError(stage, error)) def _observe_late_close(self, stage: str, task: asyncio.Task) -> None: @@ -419,7 +412,7 @@ def take_result(self, keep_recording: bool) -> CallExecutionResult: self.recorder.discard() return CallExecutionResult() - return CallExecutionResult(self.recorder.take()) + return CallExecutionResult(self.recorder.take_ready_or_discard()) _DEADLINE_EVENTS = { @@ -457,8 +450,29 @@ def request_stop(self) -> None: self._events.emit(StopRequested()) @property - def cleanup_failed(self) -> bool: - return self._resources is not None and bool(self._resources.cleanup_errors) + def cleanup_result(self) -> CleanupResult: + return self._resources.cleanup_result if self._resources is not None else CleanupResult() + + def log_cleanup_error(self) -> None: + outcome = self.cleanup_result + + if not ( + outcome.is_remote_finish_unknown or outcome.is_local_close_unknown + or outcome.is_recording_export_running + ): + return + + logger.error( + "WARNING!!! Call cleanup was not fully finished. You may need to perform a reboot: session=%s " + "is_remote_finish_unknown=%s is_local_close_unknown=%s " + "unfinished_tasks=%s is_recording_export_running=%s; " + "cleanup failure does not disable later call attempts", + self.session.session_id, + outcome.is_remote_finish_unknown, + outcome.is_local_close_unknown, + outcome.unfinished_task_count, + outcome.is_recording_export_running, + ) def notify_incoming_ended(self, call_id: str) -> bool: if ( diff --git a/whatsapp_chatbot_python/calls/greenapi.py b/whatsapp_chatbot_python/calls/greenapi.py index d76cd8f..6f88b9b 100644 --- a/whatsapp_chatbot_python/calls/greenapi.py +++ b/whatsapp_chatbot_python/calls/greenapi.py @@ -88,6 +88,10 @@ def _ensure_open(self) -> None: if self._closing: raise RuntimeError("Green API call is closing") + @property + def is_rtc_close_pending(self) -> bool: + return self._rtc_close_task is not None and not self._rtc_close_task.done() + async def _close_rtc(self) -> None: if self._rtc_close_task is None: self._rtc_close_task = asyncio.create_task( @@ -97,10 +101,23 @@ async def _close_rtc(self) -> None: self._rtc_close_task.add_done_callback(self._observe_rtc_close) - try: - await asyncio.wait_for( - asyncio.shield(self._rtc_close_task), self._shutdown_timeout, + done, _ = await asyncio.wait({self._rtc_close_task}, timeout=self._shutdown_timeout) + + if not done: + self._rtc_close_task.cancel() + + done, _ = await asyncio.wait( + {self._rtc_close_task}, timeout=min(self._shutdown_timeout, 1.0), ) + + self.close_errors.append(CallCleanupError("rtc_close", TimeoutError())) + + return + + try: + self._rtc_close_task.result() + except asyncio.CancelledError: + self.close_errors.append(CallCleanupError("rtc_close", RuntimeError("RTC close was cancelled"))) except Exception as error: self.close_errors.append(CallCleanupError("rtc_close", error)) diff --git a/whatsapp_chatbot_python/calls/models.py b/whatsapp_chatbot_python/calls/models.py index ad4cb9e..29b3683 100644 --- a/whatsapp_chatbot_python/calls/models.py +++ b/whatsapp_chatbot_python/calls/models.py @@ -456,7 +456,7 @@ def __init__(self, provider: str, code: str, detail: str) -> None: class CallCleanupError(RuntimeError): - """A resource could not be confirmed closed; the instance needs recovery.""" + """A call resource could not be confirmed closed during cleanup.""" def __init__(self, stage: str, cause: Exception) -> None: self.stage = stage self.cause = cause @@ -465,7 +465,7 @@ def __init__(self, stage: str, cause: Exception) -> None: class CallServiceUnavailable(RuntimeError): - """A previous call left resource ownership uncertain for this instance.""" + """A call service cannot accept a new call.""" @dataclass(frozen=True) diff --git a/whatsapp_chatbot_python/calls/recorder.py b/whatsapp_chatbot_python/calls/recorder.py index 59b742a..e61664b 100644 --- a/whatsapp_chatbot_python/calls/recorder.py +++ b/whatsapp_chatbot_python/calls/recorder.py @@ -5,13 +5,14 @@ import logging import sys from array import array +from concurrent.futures import Future from contextlib import ExitStack from enum import Enum from pathlib import Path from queue import Full, Queue from tempfile import NamedTemporaryFile, TemporaryFile -from threading import Thread -from typing import BinaryIO +from threading import Event, Lock, Thread +from typing import BinaryIO, Callable, TypeVar from .models import ( PcmChunk, @@ -23,10 +24,13 @@ ) MP3_ENCODER = "libmp3lame" + MP3_CONTAINER = "mp3" logger = logging.getLogger(__name__) +T = TypeVar("T") + class Track(Enum): CALLER = "caller" @@ -200,9 +204,11 @@ def __init__(self, options: RecordingOptions) -> None: self._options = options self._capture = PcmCapture(options) self._encoder = Mp3Encoder(options) - self._export: asyncio.Task | None = None + self._export: Future[Path | None] | None = None self._cleanup: asyncio.Task | None = None self._path: Path | None = None + self._path_lock = Lock() + self._export_worker_finished = Event() self._closed = False self._failed = False self._discard_requested = False @@ -224,37 +230,63 @@ def disable(self, error: Exception) -> None: async def finish(self) -> Path | None: if self._export is None: - self._export = asyncio.create_task(asyncio.to_thread(self._finish_blocking)) - self._export.add_done_callback(self._after_export) + self._export = self._start_worker(self._finish_blocking, "call-mp3-export") - return await asyncio.shield(self._export) + return await asyncio.shield(asyncio.wrap_future(self._export)) def _finish_blocking(self) -> Path | None: - try: - tracks = self._capture.seal() - - if self._failed or not self._capture.has_samples: - return None - - with NamedTemporaryFile( - suffix=".mp3", - dir=self._options.directory, - delete=False, - ) as output: - self._path = Path(output.name) + path = None - self._encoder.encode(tracks, self._path) - - return self._path + try: + path = self._export_recording() except Exception as error: self.discard() logger.warning("VoIP recording export failed: error=%s", type(error).__name__) - return None finally: try: self._capture.close() except Exception as error: logger.warning("VoIP recording cleanup failed: error=%s", type(error).__name__) + finally: + self._export_worker_finished.set() + self._delete_discarded_file() + + with self._path_lock: + return None if self._discard_requested else path + + def _export_recording(self) -> Path | None: + tracks = self._capture.seal() + + if self._failed or not self._capture.has_samples: + return None + + with NamedTemporaryFile( + suffix=".mp3", dir=self._options.directory, delete=False, + ) as output: + path = Path(output.name) + + with self._path_lock: + self._path = path + + self._encoder.encode(tracks, path) + + return path + + @staticmethod + def _start_worker(work: Callable[[], T], name: str) -> Future[T]: + future: Future[T] = Future() + + future.set_running_or_notify_cancel() + + def run() -> None: + try: + future.set_result(work()) + except BaseException as error: + future.set_exception(error) + + Thread(target=run, name=name, daemon=True).start() + + return future def take(self) -> Path | None: if self._export is None or not self._export.done(): @@ -262,36 +294,52 @@ def take(self) -> Path | None: self._export.result() - path, self._path = self._path, None + with self._path_lock: + if self._discard_requested: + return None + + path, self._path = self._path, None - return path + return path + + def take_ready_or_discard(self) -> Path | None: + """Transfer a finished MP3, or discard it after its worker finishes.""" + if self._export is None or not self._export.done(): + self.discard() + return None + + try: + return self.take() + except Exception as error: + self.discard() + logger.error("VoIP recording result unavailable: error=%s", type(error).__name__) + return None + + @property + def export_running(self) -> bool: + return self._export is not None and not self._export_worker_finished.is_set() def discard(self) -> None: - if self._export is not None and not self._export.done(): + with self._path_lock: self._discard_requested = True - return - self._discard_now() + if self._export is None or self._export_worker_finished.is_set(): + self._delete_discarded_file() - def _after_export(self, task: asyncio.Task) -> None: - if not task.cancelled(): - task.exception() - - if self._discard_requested: - self._discard_now() + def _delete_discarded_file(self) -> None: + with self._path_lock: + if not self._discard_requested or self._path is None: + return - def _discard_now(self) -> None: - if self._path is None: - return + path, self._path = self._path, None try: - self._path.unlink(missing_ok=True) + path.unlink(missing_ok=True) except OSError as error: logger.warning("VoIP recording removal failed: error=%s", type(error).__name__) - return - self._path = None - self._discard_requested = False + with self._path_lock: + self._path = path async def close(self) -> None: """Own cleanup even if this particular waiter is cancelled.""" @@ -305,9 +353,11 @@ async def close(self) -> None: async def _close_owned(self) -> None: try: if self._export is not None: - await self._export + await asyncio.wrap_future(self._export) else: - await asyncio.to_thread(self._capture.close) + await asyncio.wrap_future( + self._start_worker(self._capture.close, "call-pcm-close") + ) except Exception as error: logger.warning("VoIP recording cleanup failed: error=%s", type(error).__name__) finally: diff --git a/whatsapp_chatbot_python/calls/service.py b/whatsapp_chatbot_python/calls/service.py index d3cd105..391beb4 100644 --- a/whatsapp_chatbot_python/calls/service.py +++ b/whatsapp_chatbot_python/calls/service.py @@ -11,7 +11,6 @@ CallExecutionResult, CallOptions, CallSession, - CallServiceUnavailable, PlaybackOptions, ProviderConfig, RecordingOptions, @@ -64,13 +63,12 @@ def __init__( self._gate = asyncio.Lock() self._runner = None - self._runner_with_incomplete_cleanup = None self._stopped = False - self._healthy = True @property def healthy(self) -> bool: - return self._healthy + """Whether another call may be attempted.""" + return not self._stopped async def execute( self, @@ -120,9 +118,8 @@ async def _execute_admitted( raise finally: - if self._runner is not None and self._runner.cleanup_failed: - self._healthy = False - self._runner_with_incomplete_cleanup = self._runner + if self._runner is not None: + self._runner.log_cleanup_error() self._runner = None @@ -130,10 +127,6 @@ def _admit(self, session: CallSession, transition: TransitionCallback) -> bool: if session.finished: return False - if not self._healthy: - transition(session, CallEvent.INTERNAL_ERROR, error_code="service_unhealthy") - raise CallServiceUnavailable("Call service requires recovery after cleanup failure") - if self._stopped: transition(session, CallEvent.SHUTDOWN_REQUESTED) return False From 3e4987e58f9c09d10b6e82c3403bfc51bd0a1328 Mon Sep 17 00:00:00 2001 From: kennyromanov Date: Thu, 5 Nov 2026 15:12:50 +0500 Subject: [PATCH 2/2] chore: The unused class has been removed and the package version has been changed --- setup.py | 2 +- tests/calls/test_call_cleanup.py | 41 ++++++++++++++++++++ whatsapp_chatbot_python/calls/__init__.py | 3 +- whatsapp_chatbot_python/calls/call_runner.py | 14 +++++-- whatsapp_chatbot_python/calls/models.py | 4 -- 5 files changed, 54 insertions(+), 10 deletions(-) diff --git a/setup.py b/setup.py index 65afef5..023dd10 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setup( name="whatsapp-chatbot-python", - version="1.2.0", + version="1.3.0", description=( "This library helps you easily create" " a Python chatbot with WhatsApp API." diff --git a/tests/calls/test_call_cleanup.py b/tests/calls/test_call_cleanup.py index 72b6229..01df69d 100644 --- a/tests/calls/test_call_cleanup.py +++ b/tests/calls/test_call_cleanup.py @@ -311,6 +311,47 @@ async def delayed_close(operations): self.assertTrue(self.voices[0].closed) release.set() + async def test_close_timeout_records_only_the_task_outcome_after_cancel(self): + from whatsapp_chatbot_python.calls.call_runner import CallResources + + async def make_resources(): + resources = object.__new__(CallResources) + resources.options = SimpleNamespace(shutdown_seconds=0.5) + resources.api = SimpleNamespace(close_errors=[]) + resources.cleanup_errors = [] + resources._pending_cleanup = set() + return resources + + for close_result in ("success", ValueError("close failed")): + with self.subTest(close_result=close_result): + resources = await make_resources() + cancelled = asyncio.Event() + release = asyncio.Event() + + async def close_operation(): + try: + await asyncio.sleep(1) + except asyncio.CancelledError: + cancelled.set() + await release.wait() + + if isinstance(close_result, Exception): + raise close_result + + close_task = asyncio.create_task( + resources._close_step("test_close", lambda: close_operation()) + ) + + await asyncio.wait_for(cancelled.wait(), 1) + release.set() + await asyncio.wait_for(close_task, 1) + + if close_result == "success": + self.assertEqual(resources.cleanup_errors, []) + else: + self.assertEqual(len(resources.cleanup_errors), 1) + self.assertIs(resources.cleanup_errors[0].cause, close_result) + async def test_late_prepare_cannot_open_rtc_after_cleanup_started(self): release = asyncio.Event() service = self.service(setup=2) diff --git a/whatsapp_chatbot_python/calls/__init__.py b/whatsapp_chatbot_python/calls/__init__.py index df12a37..16bc130 100644 --- a/whatsapp_chatbot_python/calls/__init__.py +++ b/whatsapp_chatbot_python/calls/__init__.py @@ -13,7 +13,6 @@ CallExecutor, CallOptions, CallSession, - CallServiceUnavailable, CallState, IncomingCallExecutor, InputAudioOptions, @@ -35,7 +34,7 @@ __all__ = [ "ACTIVE_STATES", "TERMINAL_STATES", "WAITING_STATES", "AudioLoss", "CallDirection", "CallEndReason", "CallEvent", - "CallExecutionResult", "CallExecutor", "CallOptions", "CallSession", "CallServiceUnavailable", "CallState", + "CallExecutionResult", "CallExecutor", "CallOptions", "CallSession", "CallState", "CallStateMachine", "IncomingCallExecutor", "InputAudioOptions", "InputOverflowPolicy", "PlaybackOptions", "PlaybackOverflowPolicy", "ProviderError", "RecordingOptions", "StateTransition", "TransitionCallback", "VadOptions", "WhatsAppCallService", diff --git a/whatsapp_chatbot_python/calls/call_runner.py b/whatsapp_chatbot_python/calls/call_runner.py index aeb02cf..7afa572 100644 --- a/whatsapp_chatbot_python/calls/call_runner.py +++ b/whatsapp_chatbot_python/calls/call_runner.py @@ -359,10 +359,18 @@ async def _close_step( if not done: self._pending_cleanup.add(task) task.add_done_callback(lambda finished: self._observe_late_close(stage, finished)) - elif not task.cancelled(): - task.exception() + self._record_close_error(stage, TimeoutError()) - self._record_close_error(stage, TimeoutError()) + return + + if task.cancelled(): + self._record_close_error(stage, TimeoutError()) + return + + error = task.exception() + + if error is not None: + self._record_close_error(stage, error) return diff --git a/whatsapp_chatbot_python/calls/models.py b/whatsapp_chatbot_python/calls/models.py index 29b3683..7f5801d 100644 --- a/whatsapp_chatbot_python/calls/models.py +++ b/whatsapp_chatbot_python/calls/models.py @@ -464,10 +464,6 @@ def __init__(self, stage: str, cause: Exception) -> None: super().__init__(f"Call cleanup failed at {stage}: {type(cause).__name__}") -class CallServiceUnavailable(RuntimeError): - """A call service cannot accept a new call.""" - - @dataclass(frozen=True) class DeadlineExpired: kind: DeadlineKind