wip

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to
 ### Fixed
 
 - 🔒️(backend) enforce display name setting on rename API
+- 🩹(backend) handle failed and aborted egresses
 
 ## [1.31.0] - 2026-09-08
 

src/backend/core/models.py

@@ -57,7 +57,8 @@ class RecordingStatusChoices(models.TextChoices):
     ACTIVE = "active", _("Active")
     STOPPED = "stopped", _("Stopped")
     SAVED = "saved", _("Saved")
-    ABORTED = "aborted", _("Aborted")
+    ABORTED = "aborted", _("Aborted")  # from livekit egress
+    FAILED = "failed", _("Failed")  # from livekit egress
     FAILED_TO_START = "failed_to_start", _("Failed to Start")
     FAILED_TO_STOP = "failed_to_stop", _("Failed to Stop")
     NOTIFICATION_SUCCEEDED = "notification_succeeded", _("Notification succeeded")
@@ -79,17 +80,13 @@ class RecordingStatusChoices(models.TextChoices):
             cls.STOPPED,
             cls.SAVED,
             cls.ABORTED,
+            cls.FAILED,
             cls.EXTERNAL_PROCESS_SUCCESSFUL,
             cls.EXTERNAL_PROCESS_FAILED,
             cls.FAILED_TO_START,
             cls.FAILED_TO_STOP,
         }
 
-    @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}
-
 
 class RecordingModeChoices(models.TextChoices):
     """Recording mode choices."""

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

@@ -31,14 +31,16 @@ class RecordingEventsService:
 
     @staticmethod
     def handle_update(recording: Recording, egress_status):
-        """Handle egress status updates and sync recording state to room metadata."""
+        """Handle egress status updates 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)
@@ -69,6 +71,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:
@@ -89,6 +98,26 @@ 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.
+
+        EGRESS_FAILED: used when an actual runtime/pipeline error occurs after the
+        egress has started
+        """
+        recording.status = models.RecordingStatusChoices.FAILED
+        recording.save()
+
+    @staticmethod
+    def handle_aborted(recording: Recording):
+        """Set recording status to aborted, matching egress status.
+
+        EGRESS_ABORTED: used when the egress stops before it ever became
+        active/recording
+        """
+        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

@@ -181,6 +181,7 @@ class LiveKitEventsService:
     def _handle_egress_ended(self, data):
         """Handle 'egress_ended' event."""
 
+        # Fetch recording
         try:
             recording = models.Recording.objects.select_related("room").get(
                 worker_id=data.egress_info.egress_id
@@ -190,6 +191,7 @@ class LiveKitEventsService:
                 f"Recording with worker ID {data.egress_info.egress_id} does not exist"
             ) from err
 
+        # Update room
         try:
             room_name = str(recording.room.id)
             RoomManagement.update_metadata(
@@ -203,15 +205,18 @@ class LiveKitEventsService:
         except RoomManagementException as e:
             logger.exception("Failed to update room's metadata: %s", e)
 
+        # Stop metadata collector
         if recording.options.get("metadata_collector_dispatch_id", None) is not None:
             try:
                 MetadataCollectorService().stop(recording)
             except MetadataCollectorException:
                 logger.warning("Failed to stop the MetadataCollectorService")
 
+        # Handle case: EGRESS_LIMIT_REACHED
         if (
             data.egress_info.status == api.EgressStatus.EGRESS_LIMIT_REACHED
-            and recording.status == models.RecordingStatusChoices.ACTIVE
+            and recording.status
+            == models.RecordingStatusChoices.ACTIVE  # question: can we remove or factorize condition on ACTIVE ?
         ):
             try:
                 self.recording_events.handle_limit_reached(recording)
@@ -220,6 +225,21 @@ class LiveKitEventsService:
                     f"Failed to process limit reached event for recording {recording}"
                 ) from e
 
+        # Handle case: EGRESS_ABORTED
+        if (
+            data.egress_info.status == api.EgressStatus.EGRESS_ABORTED
+            and recording.status == models.RecordingStatusChoices.ACTIVE
+        ):
+            return self.recording_events.handle_aborted(recording)
+
+        # Handle case: EGRESS_FAILED
+        if (
+            data.egress_info.status == api.EgressStatus.EGRESS_FAILED
+            and recording.status == models.RecordingStatusChoices.ACTIVE
+        ):
+            return self.recording_events.handle_failed(recording)
+
+        # Handle cases: EGRESS_COMPLETE & EGRESS_LIMIT_REACHED
         # Finalize the recording, the egress has uploaded the file to the storage
         if data.egress_info.status in [
             api.EgressStatus.EGRESS_COMPLETE,