wip

CHANGELOG.md

@@ -33,6 +33,7 @@ and this project adheres to
 - 🩹(backend) identify externally provisioned users to PostHog
 - 🐛(backend) fix info panel crash for unregistered rooms
 - ♿️(frontend) focus side panel container on open #1452
+- 🩹(backend) handle failed and aborted egresses
 
 ## [1.23.0] - 2026-07-08
 

src/backend/core/models.py

@@ -57,6 +57,7 @@ class RecordingStatusChoices(models.TextChoices):
     STOPPED = "stopped", _("Stopped")
     SAVED = "saved", _("Saved")
     ABORTED = "aborted", _("Aborted")
+    FAILED = "failed", _("Failed")
     FAILED_TO_START = "failed_to_start", _("Failed to Start")
     FAILED_TO_STOP = "failed_to_stop", _("Failed to Stop")
     NOTIFICATION_SUCCEEDED = "notification_succeeded", _("Notification succeeded")
@@ -80,6 +81,7 @@ class RecordingStatusChoices(models.TextChoices):
             cls.ABORTED,
             cls.EXTERNAL_PROCESS_SUCCESSFUL,
             cls.EXTERNAL_PROCESS_FAILED,
+            cls.FAILED,
             cls.FAILED_TO_START,
             cls.FAILED_TO_STOP,
         }
@@ -87,7 +89,12 @@ class RecordingStatusChoices(models.TextChoices):
     @classmethod
     def is_unsuccessful(cls, status):
         """Determine if the recording status represents an unsuccessful state."""
-        return status in {cls.ABORTED, cls.FAILED_TO_START, cls.FAILED_TO_STOP}
+        return status in {
+            cls.ABORTED,
+            cls.FAILED,
+            cls.FAILED_TO_START,
+            cls.FAILED_TO_STOP,
+        }
 
 
 class RecordingModeChoices(models.TextChoices):

src/backend/core/recording/services/recording_events.py

@@ -26,14 +26,16 @@ class RecordingEventsService:
 
     @staticmethod
     def handle_update(recording: Recording, egress_status):
-        """Handle egress status updates and sync recording state to room metadata."""
+        """Handle egress update and sync recording state to room metadata.
+
+        Egress updates are sent for statuses EGRESS_ACTIVE and EGRESS_ENDING.
+        """
 
         room_name = str(recording.room.id)
 
         status_mapping = {
             api.EgressStatus.EGRESS_ACTIVE: "started",
             api.EgressStatus.EGRESS_ENDING: "saving",
-            api.EgressStatus.EGRESS_ABORTED: "aborted",
         }
 
         recording_status = status_mapping.get(egress_status)
@@ -59,6 +61,13 @@ class RecordingEventsService:
 
         notification_type = notification_mapping.get(recording.mode)
         if not notification_type:
+            logger.warning(
+                "Could not find notification type for: "
+                "room=%s, recording_id=%s, mode=%s",
+                recording.room.id,
+                recording.id,
+                recording.mode,
+            )
             return
 
         try:
@@ -79,6 +88,18 @@ class RecordingEventsService:
                 f"recording limit reached (recording_id={recording.id})"
             ) from e
 
+    @staticmethod
+    def handle_failed(recording: Recording):
+        """Set recording status to failed, matching egress status."""
+        recording.status = models.RecordingStatusChoices.FAILED
+        recording.save()
+
+    @staticmethod
+    def handle_aborted(recording: Recording):
+        """Set recording status to aborted, matching egress status."""
+        recording.status = models.RecordingStatusChoices.ABORTED
+        recording.save()
+
     @staticmethod
     def handle_complete(recording: Recording):
         """Notify external services and save recording."""

src/backend/core/services/livekit_events.py

@@ -200,7 +200,21 @@ class LiveKitEventsService:
                     f"Failed to process limit reached event for recording {recording}"
                 ) from e
 
-        # Fallback for completion when no MinIO/S3 webhooks are configured
+        if (
+            data.egress_info.status == api.EgressStatus.EGRESS_ABORTED
+            and recording.status == models.RecordingStatusChoices.ACTIVE
+        ):
+            return self.recording_events.handle_aborted(recording)
+
+        if (
+            data.egress_info.status == api.EgressStatus.EGRESS_FAILED
+            and recording.status == models.RecordingStatusChoices.ACTIVE
+        ):
+            return self.recording_events.handle_failed(recording)
+
+        # Fallback for completion when no MinIO/S3 webhooks are configured.
+        # If RECORDING_STORAGE_EVENT_ENABLE is True, completion is handled
+        # by ``on_storage_event_received``.
         if (
             not settings.RECORDING_STORAGE_EVENT_ENABLE
         ) and data.egress_info.status in [
@@ -216,8 +230,6 @@ class LiveKitEventsService:
                     recording.id,
                 )
 
-        # Silently ignoring EGRESS_ABORTED, EGRESS_FAILED
-
     def _handle_room_started(self, data):
         """Handle 'room_started' event."""
 

src/backend/core/tests/recording/service/test_recording_events.py

@@ -9,6 +9,7 @@ from unittest import mock
 import pytest
 
 from core.factories import RecordingFactory
+from core.models import RecordingStatusChoices
 from core.recording.services.recording_events import (
     RecordingEventsError,
     RecordingEventsService,
@@ -70,3 +71,22 @@ def test_handle_limit_reached_error(mock_notify, mode, notification_type, servic
     mock_notify.assert_called_once_with(
         room_name=str(recording.room.id), notification_data={"type": notification_type}
     )
+
+
+@pytest.mark.parametrize(
+    ("handler_name", "expected_status"),
+    (
+        ("handle_aborted", RecordingStatusChoices.ABORTED),
+        ("handle_failed", RecordingStatusChoices.FAILED),
+    ),
+)
+def test_handle_unsuccessful_egress(handler_name, expected_status, service):
+    """Test unsuccessful egress handlers persist a final, unsuccessful status."""
+
+    recording = RecordingFactory(status="active")
+    getattr(service, handler_name)(recording)
+
+    recording.refresh_from_db()
+    assert recording.status == expected_status
+    assert RecordingStatusChoices.is_final(recording.status)
+    assert RecordingStatusChoices.is_unsuccessful(recording.status)

src/backend/core/tests/services/test_livekit_events.py

@@ -101,7 +101,6 @@ def test_handle_egress_ended_success(
     (
         (EgressStatus.EGRESS_ACTIVE, "started"),
         (EgressStatus.EGRESS_ENDING, "saving"),
-        (EgressStatus.EGRESS_ABORTED, "aborted"),
     ),
 )
 @mock.patch("core.utils.update_room_metadata")
@@ -126,6 +125,7 @@ def test_handle_egress_updated_success(
     "egress_status",
     (
         EgressStatus.EGRESS_FAILED,
+        EgressStatus.EGRESS_ABORTED,
         EgressStatus.EGRESS_LIMIT_REACHED,
     ),
 )
@@ -420,8 +420,6 @@ def test_handle_egress_ended_does_not_finalize_when_webhooks_enabled(  # noqa: P
         EgressStatus.EGRESS_STARTING,
         EgressStatus.EGRESS_ACTIVE,
         EgressStatus.EGRESS_ENDING,
-        EgressStatus.EGRESS_FAILED,
-        EgressStatus.EGRESS_ABORTED,
     ],
 )
 @mock.patch("core.utils.update_room_metadata")
@@ -442,6 +440,47 @@ def test_handle_egress_ended_does_not_save_on_wrong_status(
     assert recording.status == "active"
 
 
+@pytest.mark.parametrize(
+    ("egress_status", "expected_status"),
+    (
+        (EgressStatus.EGRESS_ABORTED, "aborted"),
+        (EgressStatus.EGRESS_FAILED, "failed"),
+    ),
+)
+@mock.patch(
+    "core.recording.services.recording_events.notification_service."
+    "notify_external_services"
+)
+@mock.patch("core.utils.update_room_metadata")
+def test_handle_egress_ended_marks_unsuccessful_egress(  # noqa: PLR0913
+    mock_update_room_metadata,
+    mock_notify_external_services,
+    egress_status,
+    expected_status,
+    service,
+    settings,
+):  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    """An aborted/failed egress moves an active recording to its terminal
+    unsuccessful status, and never notifies external services."""
+
+    settings.RECORDING_STORAGE_EVENT_ENABLE = False
+
+    recording = RecordingFactory(worker_id="worker-1", status="active")
+    mock_data = mock.MagicMock()
+    mock_data.egress_info.egress_id = recording.worker_id
+    mock_data.egress_info.status = egress_status
+
+    service._handle_egress_ended(mock_data)
+
+    mock_notify_external_services.assert_not_called()
+    mock_update_room_metadata.assert_called_once_with(
+        str(recording.room.id), {}, ["recording_mode", "recording_status"]
+    )
+
+    recording.refresh_from_db()
+    assert recording.status == expected_status
+
+
 @pytest.mark.parametrize(
     "status", ["failed_to_start", "aborted", "failed_to_stop", "saved", "initiated"]
 )

src/backend/core/tests/test_models_recording.py

@@ -160,6 +160,7 @@ def test_models_recording_is_savable_normal():
         RecordingStatusChoices.FAILED_TO_STOP,
         RecordingStatusChoices.FAILED_TO_START,
         RecordingStatusChoices.ABORTED,
+        RecordingStatusChoices.FAILED,
     ],
 )
 def test_models_recording_is_savable_error(status):
@@ -282,6 +283,7 @@ def test_models_recording_is_saved_false_initiated():
         RecordingStatusChoices.FAILED_TO_STOP,
         RecordingStatusChoices.FAILED_TO_START,
         RecordingStatusChoices.ABORTED,
+        RecordingStatusChoices.FAILED,
     ],
 )
 def test_models_recording_is_saved_false_error_states(status):