From 7247b05a564eef32d92dc5d796884ee591af5f04 Mon Sep 17 00:00:00 2001 From: leo <260626284+cameledev@users.noreply.github.com> Date: Tue, 8 Sep 2026 17:12:17 +0200 Subject: [PATCH] wip --- CHANGELOG.md | 1 + src/backend/core/models.py | 9 ++--- .../recording/services/recording_events.py | 33 +++++++++++++++++-- src/backend/core/services/livekit_events.py | 22 ++++++++++++- 4 files changed, 56 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 38637535..733a5bfb 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/backend/core/models.py b/src/backend/core/models.py index 895deb1b..27926a50 100644 --- a/src/backend/core/models.py +++ b/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.""" diff --git a/src/backend/core/recording/services/recording_events.py b/src/backend/core/recording/services/recording_events.py index 0151c052..bea00bef 100644 --- a/src/backend/core/recording/services/recording_events.py +++ b/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.""" diff --git a/src/backend/core/services/livekit_events.py b/src/backend/core/services/livekit_events.py index 8534647e..53317725 100644 --- a/src/backend/core/services/livekit_events.py +++ b/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,