From bd1e060bd213822055f3ed1758954bc8ce332a01 Mon Sep 17 00:00:00 2001 From: None Date: Thu, 26 Feb 2026 00:11:15 -0600 Subject: [PATCH 01/15] fix(dvr): fix duplicate recording execution and add manual stop Fixes #940: Duplicate DVR task executions were caused by ClockedSchedule objects being shared across PeriodicTasks, triggering run_recording multiple times for the same recording. Added an idempotency guard in run_recording that exits early if status is already 'recording', 'completed', or 'stopped'. revoke_task() now deletes the PeriodicTask and orphaned ClockedSchedule on execution rather than relying on Celery revocation alone. Fixes #454: Added POST /api/channels/recordings/{id}/stop/ action that marks the recording as stopped and terminates only the specific DVR proxy client (identified via User-Agent: Dispatcharr-DVR/recording-{id}) without disrupting simultaneous recordings on the same or other channels. The API returns immediately; DVR client teardown and task revocation happen in a background thread to avoid 504 timeouts on the recordings list endpoint. Additional changes: - destroy() now only calls _stop_dvr_clients() for in-progress recordings, preventing accidental stream termination when deleting completed recordings - WebSocket events differentiated: recording_stopped, recording_cancelled (in-progress cancel), and recording_cancelled (delete) with was_in_progress flag so the frontend shows distinct "Recording deleted" vs "Recording cancelled" notifications - Frontend: Stop and Cancel split into separate buttons with confirmation modals; stopped recordings move immediately to Previously Recorded via optimistic status bucketing; removed dead 'Stopped' badge state - 27 new unit tests covering scheduling, idempotency, revocation, and DVR client isolation --- apps/channels/api_views.py | 176 ++++-- apps/channels/signals.py | 98 +++- apps/channels/tasks.py | 241 ++++++-- .../tests/test_recording_scheduling.py | 547 ++++++++++++++++++ frontend/src/WebSocket.jsx | 28 + frontend/src/api.js | 11 + .../src/components/cards/RecordingCard.jsx | 125 +++- .../src/utils/cards/RecordingCardUtils.js | 4 + frontend/src/utils/pages/DVRUtils.js | 2 +- 9 files changed, 1112 insertions(+), 120 deletions(-) create mode 100644 apps/channels/tests/test_recording_scheduling.py diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 9147532b..bd94b552 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -2248,6 +2248,54 @@ class RecurringRecordingRuleViewSet(viewsets.ModelViewSet): logger.warning(f"Failed to purge recordings for rule {rule_id}: {err}") +def _stop_dvr_clients(channel_uuid, recording_id=None): + """Stop DVR recording clients for a channel. + + If recording_id is provided, only the client whose User-Agent contains that + recording ID is stopped (safe for simultaneous recordings on the same channel). + If recording_id is None, all Dispatcharr-DVR clients for the channel are stopped + (used by destroy() when deleting a recording whose task_id is unknown). + + Returns the number of DVR clients stopped. + """ + from core.utils import RedisClient + from apps.proxy.ts_proxy.redis_keys import RedisKeys + from apps.proxy.ts_proxy.services.channel_service import ChannelService + + r = RedisClient.get_client() + if not r: + return 0 + client_set_key = RedisKeys.clients(channel_uuid) + client_ids = r.smembers(client_set_key) or [] + stopped = 0 + for raw_id in client_ids: + try: + cid = raw_id.decode("utf-8") if isinstance(raw_id, (bytes, bytearray)) else str(raw_id) + meta_key = RedisKeys.client_metadata(channel_uuid, cid) + ua = r.hget(meta_key, "user_agent") + ua_s = ua.decode("utf-8") if isinstance(ua, (bytes, bytearray)) else (ua or "") + if not (ua_s and "Dispatcharr-DVR" in ua_s): + continue + # When a recording_id is specified, only stop the client for that recording. + # Each run_recording task connects with User-Agent "Dispatcharr-DVR/recording-{id}", + # so we can safely target just this recording without affecting others on the channel. + if recording_id is not None and f"recording-{recording_id}" not in ua_s: + continue + try: + ChannelService.stop_client(channel_uuid, cid) + stopped += 1 + except Exception as inner_e: + logger.debug(f"Failed to stop DVR client {cid} for channel {channel_uuid}: {inner_e}") + except Exception as inner: + logger.debug(f"Error while checking client metadata: {inner}") + # Do not call ChannelService.stop_channel() here. + # Stopping the channel proxy would terminate the source connection which may + # be shared with other recordings on the same channel. The TS proxy server + # already detects when client count reaches zero and tears down the channel + # cleanly on its own (with the configured shutdown delay). + return stopped + + class RecordingViewSet(viewsets.ModelViewSet): queryset = Recording.objects.all() serializer_class = RecordingSerializer @@ -2342,69 +2390,105 @@ class RecordingViewSet(viewsets.ModelViewSet): response["Content-Disposition"] = f"inline; filename=\"{file_name}\"" return response + @action(detail=True, methods=["post"], url_path="stop") + def stop(self, request, pk=None): + """Stop a recording early while retaining the partial content for playback.""" + instance = self.get_object() + + # Mark as stopped in the DB first so run_recording detects it. + # This is the only operation that MUST be synchronous — run_recording reads + # the status field to decide whether the stream disconnection was deliberate. + cp = instance.custom_properties or {} + cp["status"] = "stopped" + cp["stopped_at"] = str(timezone.now()) + instance.custom_properties = cp + instance.save(update_fields=["custom_properties"]) + + # Everything else (DVR client teardown, task revocation, WebSocket notification) + # is moved to a daemon thread so the API returns immediately. Callers were + # seeing 5-15 s delays because _stop_dvr_clients / revoke_task / async_to_sync + # have occasional slow paths (Redis timeouts, Celery control broadcasts, etc.). + channel_uuid = str(instance.channel.uuid) + recording_id = instance.id + task_id = instance.task_id + channel_name = instance.channel.name + + def _background_stop(): + try: + stopped = _stop_dvr_clients(channel_uuid, recording_id=recording_id) + if stopped: + logger.info( + f"Stopped {stopped} DVR client(s) for channel {channel_uuid} (recording stopped early)" + ) + except Exception as e: + logger.debug(f"Unable to stop DVR clients for stopped recording: {e}") + + try: + from apps.channels.signals import revoke_task + revoke_task(task_id) + except Exception as e: + logger.debug(f"Unable to revoke task for stopped recording: {e}") + + try: + from core.utils import send_websocket_update + send_websocket_update('updates', 'update', { + "success": True, + "type": "recording_stopped", + "channel": channel_name, + }) + except Exception: + pass + + # Close the DB connection opened by this thread so it isn't leaked + try: + from django.db import connection as _conn + _conn.close() + except Exception: + pass + + import threading + threading.Thread(target=_background_stop, daemon=True).start() + + return Response({"success": True, "status": "stopped"}) + def destroy(self, request, *args, **kwargs): """Delete the Recording and ensure any active DVR client connection is closed. Also removes the associated file(s) from disk if present. """ instance = self.get_object() - - # Attempt to close the DVR client connection for this channel if active - try: - channel_uuid = str(instance.channel.uuid) - # Lazy imports to avoid module overhead if proxy isn't used - from core.utils import RedisClient - from apps.proxy.ts_proxy.redis_keys import RedisKeys - from apps.proxy.ts_proxy.services.channel_service import ChannelService - - r = RedisClient.get_client() - if r: - client_set_key = RedisKeys.clients(channel_uuid) - client_ids = r.smembers(client_set_key) or [] - stopped = 0 - for raw_id in client_ids: - try: - cid = raw_id.decode("utf-8") if isinstance(raw_id, (bytes, bytearray)) else str(raw_id) - meta_key = RedisKeys.client_metadata(channel_uuid, cid) - ua = r.hget(meta_key, "user_agent") - ua_s = ua.decode("utf-8") if isinstance(ua, (bytes, bytearray)) else (ua or "") - # Identify DVR recording client by its user agent - if ua_s and "Dispatcharr-DVR" in ua_s: - try: - ChannelService.stop_client(channel_uuid, cid) - stopped += 1 - except Exception as inner_e: - logger.debug(f"Failed to stop DVR client {cid} for channel {channel_uuid}: {inner_e}") - except Exception as inner: - logger.debug(f"Error while checking client metadata: {inner}") - if stopped: - logger.info(f"Stopped {stopped} DVR client(s) for channel {channel_uuid} due to recording cancellation") - # If no clients remain after stopping DVR clients, proactively stop the channel - try: - remaining = r.scard(client_set_key) or 0 - except Exception: - remaining = 0 - if remaining == 0: - try: - ChannelService.stop_channel(channel_uuid) - logger.info(f"Stopped channel {channel_uuid} (no clients remain)") - except Exception as sc_e: - logger.debug(f"Unable to stop channel {channel_uuid}: {sc_e}") - except Exception as e: - logger.debug(f"Unable to stop DVR clients for cancelled recording: {e}") + channel_name = instance.channel.name # Capture paths before deletion cp = instance.custom_properties or {} + rec_status = cp.get("status", "") + + # Only stop the DVR client if this recording is actively streaming. + # Stopping for completed/upcoming recordings would kill an unrelated + # in-progress recording on the same channel. + if rec_status == "recording": + try: + channel_uuid = str(instance.channel.uuid) + stopped = _stop_dvr_clients(channel_uuid) + if stopped: + logger.info(f"Stopped {stopped} DVR client(s) for channel {channel_uuid} due to recording cancellation") + except Exception as e: + logger.debug(f"Unable to stop DVR clients for cancelled recording: {e}") file_path = cp.get("file_path") temp_ts_path = cp.get("_temp_file_path") # Perform DB delete first, then try to remove files response = super().destroy(request, *args, **kwargs) - # Notify frontends to refresh recordings + # Notify frontends try: from core.utils import send_websocket_update - send_websocket_update('updates', 'update', {"success": True, "type": "recordings_refreshed"}) + send_websocket_update('updates', 'update', { + "success": True, + "type": "recording_cancelled", + "channel": channel_name, + "was_in_progress": rec_status == "recording", + }) except Exception: pass diff --git a/apps/channels/signals.py b/apps/channels/signals.py index 75ef16dd..00b8fb7c 100644 --- a/apps/channels/signals.py +++ b/apps/channels/signals.py @@ -2,14 +2,15 @@ from django.db.models.signals import m2m_changed, pre_save, post_save, post_delete from django.dispatch import receiver -from django.utils.timezone import now +from django.utils.timezone import now, is_aware, make_aware from celery.result import AsyncResult +from django_celery_beat.models import ClockedSchedule, PeriodicTask from .models import Channel, Stream, ChannelProfile, ChannelProfileMembership, Recording from apps.m3u.models import M3UAccount from apps.epg.tasks import parse_programs_for_tvg_id -import logging, requests, time +import json +import logging from .tasks import run_recording, prefetch_recording_artwork -from django.utils.timezone import now, is_aware, make_aware from datetime import timedelta logger = logging.getLogger(__name__) @@ -85,29 +86,76 @@ def create_profile_memberships(sender, instance, created, **kwargs): for channel in channels ]) +def _dvr_task_name(recording_id): + """Predictable PeriodicTask name for a DVR recording.""" + return f"dvr-recording-{recording_id}" + + def schedule_recording_task(instance, eta=None): - # Use the explicitly-passed (and timezone-aware) eta if provided; - # fall back to instance.start_time only as a last resort. + """Schedule a recording task via ClockedSchedule + one-off PeriodicTask. + + The task is stored in the database and dispatched by Celery Beat at the + scheduled time with no countdown. This avoids the Redis visibility_timeout + redelivery bug that caused duplicate recordings when using apply_async + with long countdowns. See https://github.com/Dispatcharr/Dispatcharr/issues/940 + """ if eta is None: eta = instance.start_time - # Ensure eta is timezone-aware before comparing against now() if eta is not None and not is_aware(eta): eta = make_aware(eta) - # countdown=0 fires immediately (in-progress programs whose start_time was - # clamped to now by the serializer), countdown>0 delays until start_time - # (future programs). Using an integer countdown avoids any timezone - # serialization ambiguity that can occur with an absolute eta datetime. - countdown = max(0, int((eta - now()).total_seconds())) - # Pass recording_id first so task can persist metadata to the correct row - task = run_recording.apply_async( - args=[instance.id, instance.channel_id, str(instance.start_time), str(instance.end_time)], - countdown=countdown, + # Clamp to now so Beat dispatches immediately for past/current start times + if eta <= now(): + eta = now() + + task_args = [ + instance.id, + instance.channel_id, + str(instance.start_time), + str(instance.end_time), + ] + + clocked, _ = ClockedSchedule.objects.get_or_create(clocked_time=eta) + task_name = _dvr_task_name(instance.id) + PeriodicTask.objects.update_or_create( + name=task_name, + defaults={ + "task": "apps.channels.tasks.run_recording", + "clocked": clocked, + "args": json.dumps(task_args), + "one_off": True, + "enabled": True, + "interval": None, + "crontab": None, + "solar": None, + }, ) - return task.id + return task_name + def revoke_task(task_id): - if task_id: + """Cancel a pending recording task. + + task_id is normally a PeriodicTask name (e.g. "dvr-recording-42"). + For backwards compatibility with task_ids stored before this fix + (Celery async-result UUIDs), falls back to AsyncResult.revoke(). + """ + if not task_id: + return + # Primary path: delete the PeriodicTask and clean up its ClockedSchedule + try: + pt = PeriodicTask.objects.get(name=task_id) + old_clocked = pt.clocked + pt.delete() + if old_clocked and not PeriodicTask.objects.filter(clocked=old_clocked).exists(): + old_clocked.delete() + return + except PeriodicTask.DoesNotExist: + pass + # Fallback for legacy Celery task UUIDs + try: AsyncResult(task_id).revoke() + except Exception: + pass @receiver(pre_save, sender=Recording) def revoke_old_task_on_update(sender, instance, **kwargs): @@ -152,11 +200,17 @@ def schedule_task_on_save(sender, instance, created, **kwargs): instance.save(update_fields=['task_id']) else: print("Start time is in the past. Not scheduling.") - # Kick off poster/artwork prefetch to enrich Upcoming cards - try: - prefetch_recording_artwork.apply_async(args=[instance.id], countdown=1) - except Exception as e: - print("Error scheduling artwork prefetch:", e) + # Kick off poster/artwork prefetch to enrich Upcoming cards. + # Skip when the recording is already active or finished — run_recording + # handles its own poster resolution, and scheduling artwork prefetch + # while the task is running causes a race that can overwrite status. + cp = instance.custom_properties or {} + rec_status = cp.get("status", "") + if rec_status not in ("recording", "completed", "stopped", "interrupted"): + try: + prefetch_recording_artwork.apply_async(args=[instance.id], countdown=1) + except Exception as e: + print("Error scheduling artwork prefetch:", e) except Exception as e: import traceback print("Error in post_save signal:", e) diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 8d49287b..87abb90d 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -1497,6 +1497,44 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): - Attempts to capture stream stats from TS proxy (codec, resolution, fps, etc.) - Attempts to capture a poster (via program.custom_properties) and store a Logo reference """ + from .models import Recording, Logo + + # --- Idempotency guard (prevents duplicate recordings from task redelivery) --- + try: + rec_check = Recording.objects.filter(id=recording_id).only("custom_properties").first() + if not rec_check: + # Recording was deleted before this task got a chance to run (e.g. user + # cancelled from the Upcoming list, or destroy() ran concurrently). + logger.info( + f"run_recording called for recording {recording_id} but it no longer exists — " + f"skipping (recording was cancelled before task started)." + ) + return + status = (rec_check.custom_properties or {}).get("status", "") + if status in ("recording", "completed", "stopped"): + logger.warning( + f"run_recording called for recording {recording_id} but status " + f"is already '{status}' - skipping duplicate execution." + ) + return + except Exception as e: + logger.debug(f"Idempotency guard check failed (proceeding anyway): {e}") + + # --- Clean up the one-off PeriodicTask that dispatched task --- + try: + from django_celery_beat.models import ClockedSchedule as _CS, PeriodicTask as _PT + task_name = f"dvr-recording-{recording_id}" + try: + pt = _PT.objects.get(name=task_name) + old_clocked = pt.clocked + pt.delete() + if old_clocked and not _PT.objects.filter(clocked=old_clocked).exists(): + old_clocked.delete() + except _PT.DoesNotExist: + pass + except Exception as e: + logger.debug(f"PeriodicTask cleanup failed (non-fatal): {e}") + channel = Channel.objects.get(id=channel_id) start_time = datetime.fromisoformat(start_time_str) @@ -1536,7 +1574,6 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # Try to resolve the Recording row up front recording_obj = None try: - from .models import Recording, Logo recording_obj = Recording.objects.get(id=recording_id) # Prime custom_properties with file info/status cp = recording_obj.custom_properties or {} @@ -1822,7 +1859,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): with requests.get( test_url, headers={ - 'User-Agent': 'Dispatcharr-DVR', + 'User-Agent': f'Dispatcharr-DVR/recording-{recording_id}', }, stream=True, timeout=(10, 15), @@ -1878,6 +1915,38 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): except Exception as e: last_error = str(e) logger.warning(f"DVR: attempt failed for base {base}: {e}") + # If there was a working stream that was interrupted (e.g. recording + # cancelled via stop_client), do not retry on another base — that would + # re-establish the DVR connection and make the recording re-appear. + if bytes_written > 0: + # Check if this was a deliberate "stop" vs an unexpected interruption + try: + rec_check = Recording.objects.filter(id=recording_id).only("custom_properties").first() + if rec_check and (rec_check.custom_properties or {}).get("status") == "stopped": + # Graceful stop — not an error, just an early end + interrupted = False + logger.info(f"Recording {recording_id} was stopped by user — ending stream.") + else: + interrupted = True + interrupted_reason = f"stream_interrupted: {e}" + except Exception: + interrupted = True + interrupted_reason = f"stream_interrupted: {e}" + break + # Also bail if the Recording was deleted (user cancelled before data arrived) + rec_exists = Recording.objects.filter(id=recording_id).exists() + if not rec_exists: + interrupted = True + interrupted_reason = "recording_deleted_during_connect" + break + # Check for "stopped" status even before data arrived + try: + rec_check = Recording.objects.filter(id=recording_id).only("custom_properties").first() + if rec_check and (rec_check.custom_properties or {}).get("status") == "stopped": + interrupted = False + break + except Exception: + pass if chosen_base is None and bytes_written == 0: interrupted = True @@ -1892,39 +1961,48 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # We cannot compute accurate elapsed here; fine to leave as is pass - # If no bytes were written at all, mark detail + # If no bytes were written at all, check whether this was a deliberate stop or a + # genuine failure. The exception handler above already sets interrupted=False when + # it detects "stopped" status, but do not override that decision here. if bytes_written == 0 and not interrupted: - interrupted = True - interrupted_reason = f"no_stream_data: {last_error or 'unknown'}" - - # Update DB status immediately so the UI reflects the change on the event below + _deliberately_stopped = False try: - if recording_obj is None: - from .models import Recording - recording_obj = Recording.objects.get(id=recording_id) - cp_now = recording_obj.custom_properties or {} - cp_now.update({ - "status": "interrupted" if interrupted else "completed", - "ended_at": str(datetime.now()), - "file_name": filename or cp_now.get("file_name"), - "file_path": final_path or cp_now.get("file_path"), - }) - if interrupted and interrupted_reason: - cp_now["interrupted_reason"] = interrupted_reason - recording_obj.custom_properties = cp_now - recording_obj.save(update_fields=["custom_properties"]) - except Exception as e: - logger.debug(f"Failed to update immediate recording status: {e}") + _rc = Recording.objects.filter(id=recording_id).only("custom_properties").first() + if _rc and (_rc.custom_properties or {}).get("status") == "stopped": + _deliberately_stopped = True + except Exception: + pass - async_to_sync(channel_layer.group_send)( - "updates", - { - "type": "update", - "data": {"success": True, "type": "recording_ended", "channel": channel.name} - }, - ) - # After the loop, the file and response are closed automatically. - logger.info(f"Finished recording for channel {channel.name}") + if not _deliberately_stopped: + interrupted = True + interrupted_reason = f"no_stream_data: {last_error or 'unknown'}" + + # Update DB status immediately so the UI reflects the change on the event below + try: + if recording_obj is None: + recording_obj = Recording.objects.get(id=recording_id) + cp_now = recording_obj.custom_properties or {} + cp_now.update({ + "status": "interrupted", + "ended_at": str(datetime.now()), + "file_name": filename or cp_now.get("file_name"), + "file_path": final_path or cp_now.get("file_path"), + "interrupted_reason": interrupted_reason, + }) + recording_obj.custom_properties = cp_now + recording_obj.save(update_fields=["custom_properties"]) + except Exception as e: + logger.debug(f"Failed to update immediate recording status: {e}") + + async_to_sync(channel_layer.group_send)( + "updates", + { + "type": "update", + "data": {"success": True, "type": "recording_ended", "channel": channel.name} + }, + ) + # After the loop, the file and response are closed automatically. + logger.info(f"Finished recording for channel {channel.name}") # Log system event for recording end try: @@ -1940,6 +2018,20 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): except Exception as e: logger.error(f"Could not log recording end event: {e}") + # If the Recording was deleted (cancelled by user), skip post-processing + recording_cancelled = not Recording.objects.filter(id=recording_id).exists() + if recording_cancelled: + logger.info(f"Recording {recording_id} was cancelled — skipping remux and metadata.") + # Clean up any remaining temp or stub output files + for _cleanup_path in [temp_ts_path, final_path]: + try: + if _cleanup_path and os.path.exists(_cleanup_path): + os.remove(_cleanup_path) + logger.info(f"Cleaned up cancelled recording artifact: {_cleanup_path}") + except Exception: + pass + return + # Remux TS to MKV container remux_success = False try: @@ -2039,19 +2131,25 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # Persist final metadata to Recording (status, ended_at, and stream stats if available) try: if recording_obj is None: - from .models import Recording recording_obj = Recording.objects.get(id=recording_id) + # Re-read from DB to get the latest status (stop endpoint may have set it) + recording_obj.refresh_from_db() cp = recording_obj.custom_properties or {} - cp.update({ - "ended_at": str(datetime.now()), - }) - if interrupted: + cp["ended_at"] = str(datetime.now()) + + # Determine final status. + # A deliberate user stop ("stopped" in DB) always wins over any interrupted flag — + # this handles races where the stream connection broke milliseconds before or + # after the stop endpoint committed its DB save. + db_status_now = cp.get("status", "") + if db_status_now == "stopped" or not interrupted: + cp["status"] = "completed" + cp.pop("interrupted_reason", None) + else: cp["status"] = "interrupted" if interrupted_reason: cp["interrupted_reason"] = interrupted_reason - else: - cp["status"] = "completed" cp["bytes_written"] = bytes_written cp["remux_success"] = remux_success @@ -2112,8 +2210,29 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): # Removed: local thumbnail generation. We rely on EPG/VOD/TMDB/OMDb/keyless providers only. + # Final cancellation guard: destroy() may have deleted the record while + # remuxing. If it's gone now, skip saving "interrupted" status and + # skip the notification — destroy() already sent recording_cancelled. + if not Recording.objects.filter(id=recording_id).exists(): + logger.info( + f"Recording {recording_id} was deleted during post-processing — skipping final save." + ) + return + recording_obj.custom_properties = cp recording_obj.save(update_fields=["custom_properties"]) + + # Notify frontends so the UI refreshes immediately (e.g. "Stopped" → "Completed") + try: + async_to_sync(channel_layer.group_send)( + "updates", + { + "type": "update", + "data": {"success": True, "type": "recording_ended", "channel": channel.name}, + }, + ) + except Exception: + pass except Exception as e: logger.debug(f"Unable to finalize Recording metadata: {e}") @@ -2154,6 +2273,20 @@ def recover_recordings_on_startup(): for rec in active: try: cp = rec.custom_properties or {} + current_status = cp.get("status", "") + + # Skip recordings that are already in a terminal or active state. + # "completed" / "stopped" — user stopped or it finished normally; do NOT + # overwrite the status and re-schedule (that would cause the + # Interrupted → In-Progress → Previously-Recorded ghost cycle). + # "recording" — a worker is already streaming; leave it alone. + if current_status in ("completed", "stopped", "recording"): + logger.info( + f"recover_recordings_on_startup: skipping recording {rec.id} " + f"(status={current_status!r}, already in terminal/active state)." + ) + continue + # Mark interrupted due to restart; will flip to 'recording' when task starts cp["status"] = "interrupted" cp["interrupted_reason"] = "server_restarted" @@ -2167,11 +2300,20 @@ def recover_recordings_on_startup(): except Exception as e: logger.warning(f"Failed to resume recording {rec.id}: {e}") - # Ensure future recordings are scheduled + # Ensure future recordings are scheduled. + # With ClockedSchedule, PeriodicTasks survive restarts in the DB. + # Only recreate if the PeriodicTask is missing (safety net). upcoming = Recording.objects.filter(start_time__gt=now, end_time__gt=now) for rec in upcoming: try: - # Schedule task at start_time + from django_celery_beat.models import PeriodicTask as _PT + task_name = f"dvr-recording-{rec.id}" + if _PT.objects.filter(name=task_name).exists(): + if rec.task_id != task_name: + rec.task_id = task_name + rec.save(update_fields=["task_id"]) + continue + # PeriodicTask missing - recreate it task_id = schedule_recording_task(rec) if task_id: rec.task_id = task_id @@ -2553,6 +2695,13 @@ def prefetch_recording_artwork(recording_id): from .models import Recording rec = Recording.objects.get(id=recording_id) cp = rec.custom_properties or {} + + # Bail out if the recording is already active or finished — run_recording + # handles poster resolution itself, and saving here can race with status updates. + current_status = cp.get("status", "") + if current_status in ("recording", "completed", "stopped", "interrupted"): + return "skipped: status is " + current_status + program = cp.get("program") or {} poster_logo_id, poster_url = _resolve_poster_for_program(rec.channel.name, program) updated = False @@ -2593,7 +2742,15 @@ def prefetch_recording_artwork(recording_id): pass if updated: - rec.custom_properties = cp + # Re-read from DB to avoid overwriting status changes made by + # the stop endpoint or run_recording's final metadata save. + rec.refresh_from_db() + fresh_cp = rec.custom_properties or {} + for key in ("poster_logo_id", "poster_url", "rating", "rating_system", + "season", "episode", "onscreen_episode"): + if key in cp: + fresh_cp[key] = cp[key] + rec.custom_properties = fresh_cp rec.save(update_fields=["custom_properties"]) try: from core.utils import send_websocket_update diff --git a/apps/channels/tests/test_recording_scheduling.py b/apps/channels/tests/test_recording_scheduling.py new file mode 100644 index 00000000..56bdbe87 --- /dev/null +++ b/apps/channels/tests/test_recording_scheduling.py @@ -0,0 +1,547 @@ +"""Tests for DVR recording scheduling with ClockedSchedule. + + Prefers Celery ClockedScheduled over Redis time due to 3600s limitation on scheduled tasks; + scheduled tasks beyond 3600s cause Redis time to hand process between workers and duplicate jobs +""" +from datetime import timedelta +from unittest.mock import patch, MagicMock + +from django.test import TestCase +from django.utils import timezone +from django_celery_beat.models import ClockedSchedule, PeriodicTask + +from apps.channels.models import Channel, Recording +from apps.channels.signals import ( + schedule_recording_task, + revoke_task, + _dvr_task_name, +) + + +class ScheduleRecordingTaskTests(TestCase): + """Tests for schedule_recording_task().""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=1, name="Test Channel") + + def tearDown(self): + PeriodicTask.objects.filter(name__startswith="dvr-recording-").delete() + ClockedSchedule.objects.all().delete() + + @patch("apps.channels.signals.run_recording") + def test_future_recording_creates_periodic_task(self, mock_run_recording): + """Recordings in the future create a ClockedSchedule + PeriodicTask.""" + future_time = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future_time, + end_time=future_time + timedelta(hours=1), + ) + + task_id = schedule_recording_task(rec, eta=future_time) + + expected_name = f"dvr-recording-{rec.id}" + self.assertEqual(task_id, expected_name) + + pt = PeriodicTask.objects.get(name=expected_name) + self.assertTrue(pt.one_off) + self.assertTrue(pt.enabled) + self.assertEqual(pt.task, "apps.channels.tasks.run_recording") + self.assertIsNotNone(pt.clocked) + + # apply_async should not have been called + mock_run_recording.apply_async.assert_not_called() + + @patch("apps.channels.signals.run_recording") + def test_immediate_recording_creates_periodic_task(self, mock_run_recording): + """Recordings starting now also use ClockedSchedule for consistency.""" + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now, + end_time=now + timedelta(hours=1), + ) + + task_id = schedule_recording_task(rec, eta=now) + + expected_name = f"dvr-recording-{rec.id}" + self.assertEqual(task_id, expected_name) + self.assertTrue(PeriodicTask.objects.filter(name=expected_name).exists()) + + @patch("apps.channels.signals.run_recording") + def test_past_start_time_clamps_to_now(self, mock_run_recording): + """Recordings with past start_time get clamped to now.""" + past_time = timezone.now() - timedelta(minutes=5) + rec = Recording.objects.create( + channel=self.channel, + start_time=past_time, + end_time=timezone.now() + timedelta(hours=1), + ) + + task_id = schedule_recording_task(rec, eta=past_time) + + expected_name = f"dvr-recording-{rec.id}" + self.assertEqual(task_id, expected_name) + pt = PeriodicTask.objects.get(name=expected_name) + # Clocked time should be >= now + self.assertGreaterEqual(pt.clocked.clocked_time, past_time) + + @patch("apps.channels.signals.run_recording") + def test_reschedule_updates_existing_periodic_task(self, mock_run_recording): + """Calling schedule_recording_task twice updates the existing PeriodicTask.""" + future_time = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future_time, + end_time=future_time + timedelta(hours=1), + ) + + schedule_recording_task(rec, eta=future_time) + + # Reschedule with a different time + new_eta = future_time + timedelta(hours=1) + schedule_recording_task(rec, eta=new_eta) + + # Should still be exactly one PeriodicTask + task_name = f"dvr-recording-{rec.id}" + self.assertEqual(PeriodicTask.objects.filter(name=task_name).count(), 1) + + @patch("apps.channels.signals.run_recording") + def test_naive_eta_is_made_aware(self, mock_run_recording): + """A naive (timezone-unaware) eta is made timezone-aware.""" + from datetime import datetime + naive_eta = datetime(2030, 6, 15, 14, 0, 0) + rec = Recording.objects.create( + channel=self.channel, + start_time=timezone.now() + timedelta(hours=1), + end_time=timezone.now() + timedelta(hours=2), + ) + + task_id = schedule_recording_task(rec, eta=naive_eta) + + expected_name = f"dvr-recording-{rec.id}" + self.assertEqual(task_id, expected_name) + pt = PeriodicTask.objects.get(name=expected_name) + self.assertTrue(timezone.is_aware(pt.clocked.clocked_time)) + + +class RevokeTaskTests(TestCase): + """Tests for revoke_task().""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=1, name="Test Channel") + + def tearDown(self): + PeriodicTask.objects.filter(name__startswith="dvr-recording-").delete() + ClockedSchedule.objects.all().delete() + + def test_revoke_deletes_periodic_task_and_clocked_schedule(self): + """revoke_task deletes the PeriodicTask and orphaned ClockedSchedule.""" + eta = timezone.now() + timedelta(hours=5) + clocked = ClockedSchedule.objects.create(clocked_time=eta) + PeriodicTask.objects.create( + name="dvr-recording-10", + task="apps.channels.tasks.run_recording", + clocked=clocked, + one_off=True, + enabled=True, + ) + + revoke_task("dvr-recording-10") + + self.assertFalse(PeriodicTask.objects.filter(name="dvr-recording-10").exists()) + self.assertFalse(ClockedSchedule.objects.filter(id=clocked.id).exists()) + + def test_revoke_keeps_shared_clocked_schedule(self): + """ClockedSchedule is kept if another PeriodicTask still references it.""" + eta = timezone.now() + timedelta(hours=5) + clocked = ClockedSchedule.objects.create(clocked_time=eta) + PeriodicTask.objects.create( + name="dvr-recording-10", + task="apps.channels.tasks.run_recording", + clocked=clocked, + one_off=True, + ) + PeriodicTask.objects.create( + name="dvr-recording-11", + task="apps.channels.tasks.run_recording", + clocked=clocked, + one_off=True, + ) + + revoke_task("dvr-recording-10") + + self.assertFalse(PeriodicTask.objects.filter(name="dvr-recording-10").exists()) + self.assertTrue(ClockedSchedule.objects.filter(id=clocked.id).exists()) + + @patch("apps.channels.signals.AsyncResult") + def test_revoke_falls_back_to_async_result_for_legacy_ids(self, mock_async_result): + """revoke_task falls back to AsyncResult.revoke() for old-style UUIDs.""" + revoke_task("550e8400-e29b-41d4-a716-446655440000") + + mock_async_result.assert_called_once_with("550e8400-e29b-41d4-a716-446655440000") + mock_async_result.return_value.revoke.assert_called_once() + + def test_revoke_none_is_noop(self): + """revoke_task(None) does nothing.""" + revoke_task(None) # Should not raise + + def test_revoke_empty_string_is_noop(self): + """revoke_task('') does nothing.""" + revoke_task("") # Should not raise + + +class DvrTaskNameTests(TestCase): + """Tests for the naming convention helper.""" + + def test_task_name_format(self): + self.assertEqual(_dvr_task_name(42), "dvr-recording-42") + + def test_task_name_fits_in_charfield(self): + name = _dvr_task_name(999999999) + self.assertLessEqual(len(name), 255) + + +class SignalIntegrationTests(TestCase): + """Integration tests for the post_save / post_delete signal handlers.""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=1, name="Test Channel") + + def tearDown(self): + PeriodicTask.objects.filter(name__startswith="dvr-recording-").delete() + ClockedSchedule.objects.all().delete() + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_post_save_creates_periodic_task_for_future_recording(self, mock_artwork): + """Saving a future Recording creates a PeriodicTask via post_save signal.""" + mock_artwork.apply_async.return_value = MagicMock() + + future = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + ) + + rec.refresh_from_db() + task_name = f"dvr-recording-{rec.id}" + self.assertEqual(rec.task_id, task_name) + self.assertTrue(PeriodicTask.objects.filter(name=task_name).exists()) + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_post_delete_removes_periodic_task(self, mock_artwork): + """Deleting a Recording removes its PeriodicTask.""" + mock_artwork.apply_async.return_value = MagicMock() + + future = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + ) + + rec.refresh_from_db() + task_name = rec.task_id + self.assertTrue(PeriodicTask.objects.filter(name=task_name).exists()) + + rec.delete() + self.assertFalse(PeriodicTask.objects.filter(name=task_name).exists()) + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_bulk_delete_cleans_up_all_periodic_tasks(self, mock_artwork): + """Bulk deleting recordings cleans up all their PeriodicTasks.""" + mock_artwork.apply_async.return_value = MagicMock() + + future = timezone.now() + timedelta(hours=2) + rec_ids = [] + for i in range(5): + rec = Recording.objects.create( + channel=self.channel, + start_time=future + timedelta(hours=i), + end_time=future + timedelta(hours=i + 1), + ) + rec_ids.append(rec.id) + + for rid in rec_ids: + self.assertTrue( + PeriodicTask.objects.filter(name=f"dvr-recording-{rid}").exists() + ) + + Recording.objects.filter(channel=self.channel).delete() + + self.assertEqual( + PeriodicTask.objects.filter(name__startswith="dvr-recording-").count(), 0 + ) + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_pre_save_revokes_on_time_change(self, mock_artwork): + """Changing a recording's start_time revokes the old task and creates a new one.""" + mock_artwork.apply_async.return_value = MagicMock() + + future = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + ) + + rec.refresh_from_db() + old_task_name = rec.task_id + self.assertTrue(PeriodicTask.objects.filter(name=old_task_name).exists()) + + # Change the start time — pre_save clears task_id, post_save reschedules + new_future = future + timedelta(hours=3) + rec.start_time = new_future + rec.end_time = new_future + timedelta(hours=1) + rec.save() + + rec.refresh_from_db() + # Old PeriodicTask should be deleted; new one should exist + self.assertIsNotNone(rec.task_id) + self.assertTrue( + PeriodicTask.objects.filter(name=f"dvr-recording-{rec.id}").exists() + ) + + +class IdempotencyGuardTests(TestCase): + """Tests for the idempotency guard in run_recording().""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=1, name="Test Channel") + + @patch("apps.channels.tasks.get_channel_layer") + def test_skips_if_already_recording(self, mock_layer): + """run_recording returns early if status is already 'recording'.""" + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now, + end_time=now + timedelta(hours=1), + custom_properties={"status": "recording", "started_at": str(now)}, + ) + + from apps.channels.tasks import run_recording as run_rec_task + result = run_rec_task(rec.id, self.channel.id, str(now), str(now + timedelta(hours=1))) + + self.assertIsNone(result) + # get_channel_layer should not have been called (returned before) + mock_layer.assert_not_called() + + @patch("apps.channels.tasks.get_channel_layer") + def test_skips_if_already_completed(self, mock_layer): + """run_recording returns early if status is already 'completed'.""" + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(hours=2), + end_time=now - timedelta(hours=1), + custom_properties={"status": "completed"}, + ) + + from apps.channels.tasks import run_recording as run_rec_task + result = run_rec_task(rec.id, self.channel.id, str(rec.start_time), str(rec.end_time)) + + self.assertIsNone(result) + mock_layer.assert_not_called() + + @patch("apps.channels.tasks.get_channel_layer") + def test_skips_if_already_stopped(self, mock_layer): + """run_recording returns early if status is already 'stopped' (user stopped it early).""" + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(hours=1), + end_time=now + timedelta(hours=1), + custom_properties={"status": "stopped", "stopped_at": str(now)}, + ) + + from apps.channels.tasks import run_recording as run_rec_task + result = run_rec_task(rec.id, self.channel.id, str(rec.start_time), str(rec.end_time)) + + self.assertIsNone(result) + mock_layer.assert_not_called() + + +class ArtworkPrefetchSignalGuardTests(TestCase): + """Tests that the post_save signal does not schedule artwork prefetch when + the recording is in an active or terminal state.""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=1, name="Test Channel") + + def tearDown(self): + PeriodicTask.objects.filter(name__startswith="dvr-recording-").delete() + ClockedSchedule.objects.all().delete() + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_artwork_prefetch_not_scheduled_when_status_recording(self, mock_artwork): + """post_save must NOT schedule artwork prefetch when status='recording' + to prevent a race that overwrites the running task's status updates.""" + future = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + custom_properties={"status": "recording"}, + ) + + # Simulate a save that run_recording itself might do mid-recording + rec.custom_properties = {"status": "recording", "file_path": "/data/recordings/test.mkv"} + rec.save(update_fields=["custom_properties"]) + + # apply_async was not called for the "recording" save + mock_artwork.apply_async.assert_not_called() + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_artwork_prefetch_not_scheduled_when_status_completed(self, mock_artwork): + """post_save must NOT schedule artwork prefetch when status='completed'.""" + future = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + custom_properties={"status": "completed"}, + ) + + rec.custom_properties = {"status": "completed"} + rec.save(update_fields=["custom_properties"]) + + mock_artwork.apply_async.assert_not_called() + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_artwork_prefetch_not_scheduled_when_status_stopped(self, mock_artwork): + """post_save must NOT schedule artwork prefetch when status='stopped'.""" + future = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + custom_properties={"status": "stopped"}, + ) + + rec.custom_properties = {"status": "stopped"} + rec.save(update_fields=["custom_properties"]) + + mock_artwork.apply_async.assert_not_called() + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_artwork_prefetch_scheduled_for_new_upcoming_recording(self, mock_artwork): + """post_save SHOULD schedule artwork prefetch for a newly created upcoming recording.""" + mock_artwork.apply_async.return_value = MagicMock() + future = timezone.now() + timedelta(hours=2) + Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + custom_properties={}, # no status yet — should trigger prefetch + ) + + self.assertTrue(mock_artwork.apply_async.called) + + +class DestroyDvrClientIsolationTests(TestCase): + """Tests that deleting a recording only stops DVR clients when the + recording is actively streaming — never for completed/upcoming recordings + that could share a channel with an unrelated in-progress recording.""" + + def setUp(self): + from django.contrib.auth import get_user_model + from rest_framework.test import APIRequestFactory, force_authenticate + self.channel = Channel.objects.create(channel_number=1, name="Test Channel") + User = get_user_model() + self.user = User.objects.create_user( + username="dvr_test_admin", password="pass", + user_level=User.UserLevel.ADMIN, + ) + self.factory = APIRequestFactory() + self.force_authenticate = force_authenticate + + def _delete_recording(self, rec): + from apps.channels.api_views import RecordingViewSet + request = self.factory.delete(f"/api/channels/recordings/{rec.id}/") + self.force_authenticate(request, user=self.user) + view = RecordingViewSet.as_view({"delete": "destroy"}) + return view(request, pk=rec.id) + + @patch("apps.channels.api_views._stop_dvr_clients") + def test_destroy_completed_recording_does_not_stop_dvr_clients(self, mock_stop): + """Deleting a completed recording must NOT call _stop_dvr_clients.""" + rec = Recording.objects.create( + channel=self.channel, + start_time=timezone.now() - timedelta(hours=2), + end_time=timezone.now() - timedelta(hours=1), + custom_properties={"status": "completed", "file_path": "/data/recordings/test.mkv"}, + ) + self._delete_recording(rec) + mock_stop.assert_not_called() + + @patch("apps.channels.api_views._stop_dvr_clients") + def test_destroy_upcoming_recording_does_not_stop_dvr_clients(self, mock_stop): + """Deleting an upcoming (scheduled) recording must NOT call _stop_dvr_clients.""" + future = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + custom_properties={}, + ) + self._delete_recording(rec) + mock_stop.assert_not_called() + + @patch("apps.channels.api_views._stop_dvr_clients") + def test_destroy_active_recording_does_stop_dvr_clients(self, mock_stop): + """Deleting an in-progress recording MUST call _stop_dvr_clients.""" + mock_stop.return_value = 1 + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(minutes=5), + end_time=now + timedelta(hours=1), + custom_properties={"status": "recording"}, + ) + self._delete_recording(rec) + mock_stop.assert_called_once_with(str(self.channel.uuid)) + + +class PeriodicTaskCleanupOnExecutionTests(TestCase): + """Tests for PeriodicTask cleanup when run_recording starts.""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=1, name="Test Channel") + + def tearDown(self): + PeriodicTask.objects.filter(name__startswith="dvr-recording-").delete() + ClockedSchedule.objects.all().delete() + + @patch("apps.channels.signals.prefetch_recording_artwork") + @patch("apps.channels.tasks.get_channel_layer") + def test_periodic_task_cleaned_up_on_execution(self, mock_layer, mock_artwork): + """When run_recording executes, it deletes its own PeriodicTask.""" + mock_layer.return_value = MagicMock() + mock_artwork.apply_async.return_value = MagicMock() + + future = timezone.now() + timedelta(hours=2) + rec = Recording.objects.create( + channel=self.channel, + start_time=future, + end_time=future + timedelta(hours=1), + custom_properties={}, + ) + + # post_save signal should have created the PeriodicTask + task_name = f"dvr-recording-{rec.id}" + self.assertTrue(PeriodicTask.objects.filter(name=task_name).exists()) + pt = PeriodicTask.objects.get(name=task_name) + clocked_id = pt.clocked_id + + from apps.channels.tasks import run_recording as run_rec_task + # This will proceed past guards, clean up the PeriodicTask, then + # eventually fail on the actual stream connection (expected) + try: + run_rec_task(rec.id, self.channel.id, str(future), str(future + timedelta(hours=1))) + except Exception: + pass + + self.assertFalse(PeriodicTask.objects.filter(name=task_name).exists()) + self.assertFalse(ClockedSchedule.objects.filter(id=clocked_id).exists()) diff --git a/frontend/src/WebSocket.jsx b/frontend/src/WebSocket.jsx index c18a7635..f595b747 100644 --- a/frontend/src/WebSocket.jsx +++ b/frontend/src/WebSocket.jsx @@ -547,6 +547,34 @@ export const WebsocketProvider = ({ children }) => { } break; + case 'recording_stopped': + notifications.show({ + title: 'Recording stopped', + message: `Recording stopped early for ${parsedEvent.data.channel || 'channel'}. Partial content has been saved.`, + color: 'yellow', + }); + try { + await useChannelsStore.getState().fetchRecordings(); + } catch (e) { + console.warn('Failed to refresh recordings on stop:', e); + } + break; + + case 'recording_cancelled': + notifications.show({ + title: parsedEvent.data.was_in_progress ? 'Recording cancelled' : 'Recording deleted', + message: parsedEvent.data.was_in_progress + ? 'Recording cancelled and content removed.' + : 'Recording deleted.', + color: 'red', + }); + try { + await useChannelsStore.getState().fetchRecordings(); + } catch (e) { + console.warn('Failed to refresh recordings on cancel:', e); + } + break; + case 'epg_fetch_error': notifications.show({ title: 'EPG Source Error', diff --git a/frontend/src/api.js b/frontend/src/api.js index 10a836ae..4fbf1018 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -2662,6 +2662,17 @@ export default class API { } } + static async stopRecording(id) { + try { + await request(`${host}/api/channels/recordings/${id}/stop/`, { + method: 'POST', + }); + } catch (e) { + errorNotification(`Failed to stop recording ${id}`, e); + throw e; + } + } + static async runComskip(recordingId) { try { const resp = await request( diff --git a/frontend/src/components/cards/RecordingCard.jsx b/frontend/src/components/cards/RecordingCard.jsx index e1dbf020..2f428768 100644 --- a/frontend/src/components/cards/RecordingCard.jsx +++ b/frontend/src/components/cards/RecordingCard.jsx @@ -22,7 +22,7 @@ import { Text, Tooltip, } from '@mantine/core'; -import { AlertTriangle, SquareX } from 'lucide-react'; +import { AlertTriangle, Square, SquareX } from 'lucide-react'; import RecordingSynopsis from '../RecordingSynopsis'; import { deleteRecordingById, @@ -34,6 +34,7 @@ import { getShowVideoUrl, removeRecording, runComSkip, + stopRecordingById, } from './../../utils/cards/RecordingCardUtils.js'; const RecordingCard = ({ @@ -73,7 +74,7 @@ const RecordingCard = ({ const status = customProps.status; const isTimeActive = now.isAfter(start) && now.isBefore(end); const isInterrupted = status === 'interrupted'; - const isInProgress = isTimeActive; // Show as recording by time, regardless of status glitches + const isInProgress = isTimeActive && !isInterrupted && status !== 'completed' && status !== 'stopped'; const isUpcoming = now.isBefore(start); const isSeriesGroup = Boolean( recording._group_count && recording._group_count > 1 @@ -117,10 +118,18 @@ const RecordingCard = ({ } }; - // Cancel handling for series groups + // Stop / Cancel / Delete state and handlers const [cancelOpen, setCancelOpen] = React.useState(false); + const [stopConfirmOpen, setStopConfirmOpen] = React.useState(false); + const [deleteConfirmOpen, setDeleteConfirmOpen] = React.useState(false); const [busy, setBusy] = React.useState(false); - const handleCancelClick = (e) => { + + const handleStopClick = (e) => { + e.stopPropagation(); + setStopConfirmOpen(true); + }; + + const handleDeleteClick = (e) => { e.stopPropagation(); if (isRecurringRule) { onOpenRecurring?.(recording, true); @@ -129,7 +138,30 @@ const RecordingCard = ({ if (isSeriesGroup) { setCancelOpen(true); } else { + setDeleteConfirmOpen(true); + } + }; + + const confirmStop = async () => { + try { + setBusy(true); + await stopRecordingById(recording.id); + } catch (error) { + console.error('Failed to stop recording', error); + } finally { + setBusy(false); + setStopConfirmOpen(false); + try { await fetchRecordings(); } catch {} + } + }; + + const confirmDelete = async () => { + try { + setBusy(true); removeRecording(recording.id); + } finally { + setBusy(false); + setDeleteConfirmOpen(false); } }; @@ -279,18 +311,30 @@ const RecordingCard = ({ -
- + + {isInProgress && ( + + e.stopPropagation()} + onClick={handleStopClick} + > + + + + )} + e.stopPropagation()} - onClick={handleCancelClick} + onClick={handleDeleteClick} > -
+ @@ -378,11 +422,74 @@ const RecordingCard = ({ )} ); - if (!isSeriesGroup) return MainCard; + + // Confirmation modals for stop and cancel/delete + const ConfirmModals = ( + <> + setStopConfirmOpen(false)} + title="Stop Recording" + centered + size="md" + zIndex={9999} + > + + + The recording will be stopped early. The portion already recorded + will be saved and available for playback. + + + + + + + + + setDeleteConfirmOpen(false)} + title={isInProgress ? 'Cancel Recording' : isUpcoming ? 'Cancel Recording' : 'Delete Recording'} + centered + size="md" + zIndex={9999} + > + + + {isInProgress + ? 'The recording will be cancelled and all recorded content will be permanently deleted.' + : isUpcoming + ? 'This scheduled recording will be cancelled.' + : 'This recording and all associated files will be permanently deleted.'} + + + + + + + + + ); + + if (!isSeriesGroup) return ( + <> + {ConfirmModals} + {MainCard} + + ); // Stacked look for series groups: render two shadow layers behind the main card return ( + {ConfirmModals} setCancelOpen(false)} diff --git a/frontend/src/utils/cards/RecordingCardUtils.js b/frontend/src/utils/cards/RecordingCardUtils.js index 6232064b..60dd2cd5 100644 --- a/frontend/src/utils/cards/RecordingCardUtils.js +++ b/frontend/src/utils/cards/RecordingCardUtils.js @@ -51,6 +51,10 @@ export const deleteRecordingById = async (recordingId) => { await API.deleteRecording(recordingId); }; +export const stopRecordingById = async (recordingId) => { + await API.stopRecording(recordingId); +}; + export const deleteSeriesAndRule = async (seriesInfo) => { const { tvg_id, title } = seriesInfo; if (tvg_id) { diff --git a/frontend/src/utils/pages/DVRUtils.js b/frontend/src/utils/pages/DVRUtils.js index bcb200b2..fb80dd15 100644 --- a/frontend/src/utils/pages/DVRUtils.js +++ b/frontend/src/utils/pages/DVRUtils.js @@ -32,7 +32,7 @@ const dedupeById = (list, toUserTime, completed, now, inProgress, upcoming) => { const e = toUserTime(rec.end_time); const status = rec.custom_properties?.status; - if (status === 'interrupted' || status === 'completed') { + if (status === 'interrupted' || status === 'completed' || status === 'stopped') { completed.push(rec); } else { if (now.isAfter(s) && now.isBefore(e)) inProgress.push(rec); From f28530a47b23ab2f9a6babbf3949f999cca997c4 Mon Sep 17 00:00:00 2001 From: None Date: Sun, 1 Mar 2026 17:01:16 -0600 Subject: [PATCH 02/15] fix(dvr): add recording extend, fix cross-recording interference, artwork race, and keepalive gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit File changes: Backend apps/channels/api_views.py — Added extend() action that uses queryset .update() to bypass signals, letting the running task pick up the new end_time via polling. Moved WS recording_stopped notification to synchronous (before HTTP response). Refactored destroy() to delete DB row first, send WS event immediately, defer cleanup to background thread. apps/channels/signals.py — Added pre_save guard to skip task revocation when status is "recording". Added post_save reentrancy guard to skip artwork prefetch for internal field-only saves (custom_properties, task_id, end_time). apps/channels/tasks.py — Added 2-second polling loop in streaming that checks for stop, deletion, and extended end_time. Added refresh_from_db() + merge before metadata save to prevent overwriting concurrent artwork prefetch. Replaced inline PeriodicTask cleanup with revoke_task()/_dvr_task_name(). Consolidated redundant DB queries in error path. Fixed TOCTOU os.path.exists + os.remove. Preserved "stopped" status in final-status logic instead of overwriting with "completed". core/utils.py — Made send_websocket_update gevent-safe by detecting monkey-patching and offloading async_to_sync to gevent's native threadpool, fixing the event loop deadlock that caused cross-recording interference. apps/proxy/ts_proxy/client_manager.py — Moved _trigger_stats_update to a background thread so remove_client() returns immediately. apps/proxy/ts_proxy/stream_generator.py — Added Redis-based health check for non-owner workers in _should_send_keepalive(), so keepalives fire correctly even when stream_manager is None. Frontend frontend/src/WebSocket.jsx — Added 400ms debounced scheduleRecordingFetch() replacing all inline fetchRecordings() calls. Added recording_extended and recording_updated event handlers. recording_cancelled now does surgical removeRecording() by ID when available. frontend/src/api.js — Added extendRecording(id, extraMinutes). frontend/src/components/cards/RecordingCard.jsx — Added Extend menu (+15m/+30m/+1h) for in-progress recordings. Widened Comskip button to stopped/interrupted statuses. frontend/src/components/forms/ProgramRecordingModal.jsx — Removed manual fetchRecordings() calls, relying on WS-driven debounced refresh. frontend/src/components/forms/RecordingDetailsModal.jsx — Added stopped to playable statuses. Widened Comskip button. Removed manual fetchRecordings(). frontend/src/components/forms/RecurringRuleModal.jsx — Removed all manual fetchRecordings() calls, relying on WS events. frontend/src/pages/Guide.jsx — Removed isLoading subscription that caused loading flash on debounced refresh. Removed manual fetchRecordings() after saving a series rule. frontend/src/store/channels.jsx — Stripped isLoading/error state from fetchRecordings to prevent full-page loading indicators. Added no-op guards in removeRecording. frontend/src/utils/cards/RecordingCardUtils.js — Added extendRecordingById wrapper. Tests apps/channels/tests/test_recording_extend.py (new) — 14 tests: extend endpoint validation, stacked extensions, signal bypass, pre_save guard behavior. apps/channels/tests/test_recording_stop_cancel.py (new) — 18 tests: stop endpoint, DVR client teardown, cancel was_in_progress flag, run_recording race guards, signal reentrancy guards. apps/channels/tests/test_ts_proxy_keepalive.py (new) — 21 tests: keepalive owner/non-owner logic, stats update error handling, client removal non-blocking, timing invariants. apps/channels/tests/test_recording_scheduling.py — Updated mock assertion to match new _stop_dvr_clients call signature. Summary: Adds an Extend action for in-progress recordings that bypasses Django signals and lets the running task dynamically pick up the new end_time via its 2-second DB polling loop. Fixes cross-recording interference caused by send_websocket_update deadlocking the gevent event loop. Fixes an artwork race condition where run_recording overwrote concurrent prefetch data. Adds Redis-based keepalive health checks for non-owner workers to prevent DVR read timeouts during source transitions. Replaces manual fetchRecordings() calls throughout the frontend with a debounced WS-driven refresh. 53 new tests. --- apps/channels/api_views.py | 148 ++++++-- apps/channels/signals.py | 19 +- apps/channels/tasks.py | 162 ++++++--- apps/channels/tests/test_recording_extend.py | 235 ++++++++++++ .../tests/test_recording_scheduling.py | 7 +- .../tests/test_recording_stop_cancel.py | 341 ++++++++++++++++++ .../channels/tests/test_ts_proxy_keepalive.py | 324 +++++++++++++++++ apps/proxy/ts_proxy/client_manager.py | 14 +- apps/proxy/ts_proxy/stream_generator.py | 25 +- core/utils.py | 46 +-- frontend/src/WebSocket.jsx | 71 ++-- frontend/src/api.js | 14 + .../src/components/cards/RecordingCard.jsx | 55 ++- .../forms/ProgramRecordingModal.jsx | 13 +- .../forms/RecordingDetailsModal.jsx | 11 +- .../components/forms/RecurringRuleModal.jsx | 9 +- frontend/src/pages/Guide.jsx | 12 +- frontend/src/store/channels.jsx | 12 +- .../src/utils/cards/RecordingCardUtils.js | 4 + 19 files changed, 1328 insertions(+), 194 deletions(-) create mode 100644 apps/channels/tests/test_recording_extend.py create mode 100644 apps/channels/tests/test_recording_stop_cancel.py create mode 100644 apps/channels/tests/test_ts_proxy_keepalive.py diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index bd94b552..b2bc0e14 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -11,7 +11,8 @@ from django.shortcuts import get_object_or_404, get_list_or_404 from django.db import transaction from django.db.models import Count, F from django.db.models import Q -import os, json, requests, logging, mimetypes +import os, json, requests, logging, mimetypes, threading +from datetime import timedelta from django.utils.http import http_date from urllib.parse import unquote from apps.accounts.permissions import ( @@ -2404,15 +2405,27 @@ class RecordingViewSet(viewsets.ModelViewSet): instance.custom_properties = cp instance.save(update_fields=["custom_properties"]) - # Everything else (DVR client teardown, task revocation, WebSocket notification) - # is moved to a daemon thread so the API returns immediately. Callers were - # seeing 5-15 s delays because _stop_dvr_clients / revoke_task / async_to_sync - # have occasional slow paths (Redis timeouts, Celery control broadcasts, etc.). + # Send the WebSocket notification before returning the response. + # send_websocket_update is gevent-safe (offloads async_to_sync to a + # real OS thread when monkey-patching is active). channel_uuid = str(instance.channel.uuid) recording_id = instance.id task_id = instance.task_id channel_name = instance.channel.name + try: + from core.utils import send_websocket_update + send_websocket_update('updates', 'update', { + "success": True, + "type": "recording_stopped", + "channel": channel_name, + }) + except Exception: + pass + + # DVR client teardown and task revocation are deferred to a daemon thread + # because they have occasional slow paths (Redis timeouts, Celery control + # broadcasts) that would otherwise add 5-15 s to the HTTP response time. def _background_stop(): try: stopped = _stop_dvr_clients(channel_uuid, recording_id=recording_id) @@ -2429,16 +2442,6 @@ class RecordingViewSet(viewsets.ModelViewSet): except Exception as e: logger.debug(f"Unable to revoke task for stopped recording: {e}") - try: - from core.utils import send_websocket_update - send_websocket_update('updates', 'update', { - "success": True, - "type": "recording_stopped", - "channel": channel_name, - }) - except Exception: - pass - # Close the DB connection opened by this thread so it isn't leaked try: from django.db import connection as _conn @@ -2446,52 +2449,106 @@ class RecordingViewSet(viewsets.ModelViewSet): except Exception: pass - import threading threading.Thread(target=_background_stop, daemon=True).start() return Response({"success": True, "status": "stopped"}) + @action(detail=True, methods=["post"], url_path="extend") + def extend(self, request, pk=None): + """Extend an in-progress recording's end_time without interrupting the stream. + + The running task re-reads end_time every ~2 s and adjusts its deadline + dynamically. The pre_save signal skips task revocation while the + recording status is 'recording'. + """ + instance = self.get_object() + cp = instance.custom_properties or {} + + if cp.get("status") in ("completed", "stopped", "interrupted"): + return Response( + {"success": False, "error": "Recording has already finished"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + try: + extra_minutes = int(request.data.get("extra_minutes", 0)) + except (TypeError, ValueError): + extra_minutes = 0 + + if extra_minutes <= 0: + return Response( + {"success": False, "error": "extra_minutes must be a positive integer"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + new_end_time = instance.end_time + timedelta(minutes=extra_minutes) + # Use queryset .update() to bypass pre_save/post_save signals. + # This avoids the pre_save signal revoking the scheduled/running + # Celery task. The running task's 2-second polling loop re-reads + # end_time from the DB and extends its deadline dynamically. + # If the task hasn't started yet (still in Beat's queue), it will + # read the updated end_time from the DB on its first poll cycle. + Recording.objects.filter(pk=instance.pk).update(end_time=new_end_time) + + try: + from core.utils import send_websocket_update + send_websocket_update('updates', 'update', { + "success": True, + "type": "recording_extended", + "recording_id": instance.id, + "new_end_time": new_end_time.isoformat(), + "extra_minutes": extra_minutes, + "channel": instance.channel.name, + }) + except Exception: + pass + + return Response({"success": True, "new_end_time": new_end_time.isoformat()}) + def destroy(self, request, *args, **kwargs): """Delete the Recording and ensure any active DVR client connection is closed. Also removes the associated file(s) from disk if present. + + Operation order matters for correctness: + 1. Delete the DB record first — run_recording's cancellation guard + (Recording.objects.filter(id=...).exists()) will now return False, + preventing it from saving 'interrupted' status or sending + recording_ended after the stream is torn down. + 2. Send recording_cancelled WebSocket immediately so the frontend + removes the card without waiting for the slow DVR client teardown. + 3. Spawn a background thread to stop the DVR client and delete files. + This mirrors the stop() endpoint's approach and avoids the 5-15 s + delay that _stop_dvr_clients() can introduce. """ instance = self.get_object() + recording_id = instance.pk channel_name = instance.channel.name - # Capture paths before deletion + # Capture state before the DB row is deleted cp = instance.custom_properties or {} rec_status = cp.get("status", "") - - # Only stop the DVR client if this recording is actively streaming. - # Stopping for completed/upcoming recordings would kill an unrelated - # in-progress recording on the same channel. - if rec_status == "recording": - try: - channel_uuid = str(instance.channel.uuid) - stopped = _stop_dvr_clients(channel_uuid) - if stopped: - logger.info(f"Stopped {stopped} DVR client(s) for channel {channel_uuid} due to recording cancellation") - except Exception as e: - logger.debug(f"Unable to stop DVR clients for cancelled recording: {e}") file_path = cp.get("file_path") temp_ts_path = cp.get("_temp_file_path") + channel_uuid = str(instance.channel.uuid) - # Perform DB delete first, then try to remove files + # 1. Delete the DB record (also fires post_delete → revoke_task_on_delete) response = super().destroy(request, *args, **kwargs) - # Notify frontends + # 2. Notify frontends immediately try: from core.utils import send_websocket_update send_websocket_update('updates', 'update', { "success": True, "type": "recording_cancelled", + "recording_id": recording_id, "channel": channel_name, "was_in_progress": rec_status == "recording", }) except Exception: pass + # 3. Defer slow teardown to a background thread library_dir = '/data' allowed_roots = ['/data/', library_dir.rstrip('/') + '/'] @@ -2505,8 +2562,33 @@ class RecordingViewSet(viewsets.ModelViewSet): except Exception as ex: logger.warning(f"Failed to delete recording artifact {path}: {ex}") - _safe_remove(file_path) - _safe_remove(temp_ts_path) + def _background_cancel(): + # Only stop the DVR client if the recording was actively streaming. + # Stopping for completed/upcoming recordings would kill an unrelated + # in-progress recording on the same channel. + if rec_status == "recording": + try: + stopped = _stop_dvr_clients(channel_uuid, recording_id=recording_id) + if stopped: + logger.info( + f"Stopped {stopped} DVR client(s) for channel {channel_uuid} due to recording cancellation" + ) + except Exception as e: + logger.debug(f"Unable to stop DVR clients for cancelled recording: {e}") + + # Best-effort file cleanup in case run_recording already exited + # before the DB delete. + _safe_remove(file_path) + _safe_remove(temp_ts_path) + + # Close the DB connection opened by this thread so it isn't leaked + try: + from django.db import connection as _conn + _conn.close() + except Exception: + pass + + threading.Thread(target=_background_cancel, daemon=True).start() return response diff --git a/apps/channels/signals.py b/apps/channels/signals.py index 00b8fb7c..a42b8c04 100644 --- a/apps/channels/signals.py +++ b/apps/channels/signals.py @@ -97,7 +97,7 @@ def schedule_recording_task(instance, eta=None): The task is stored in the database and dispatched by Celery Beat at the scheduled time with no countdown. This avoids the Redis visibility_timeout redelivery bug that caused duplicate recordings when using apply_async - with long countdowns. See https://github.com/Dispatcharr/Dispatcharr/issues/940 + with long countdowns. """ if eta is None: eta = instance.start_time @@ -136,8 +136,8 @@ def revoke_task(task_id): """Cancel a pending recording task. task_id is normally a PeriodicTask name (e.g. "dvr-recording-42"). - For backwards compatibility with task_ids stored before this fix - (Celery async-result UUIDs), falls back to AsyncResult.revoke(). + For backwards compatibility with legacy Celery async-result UUIDs, + falls back to AsyncResult.revoke(). """ if not task_id: return @@ -168,6 +168,12 @@ def revoke_old_task_on_update(sender, instance, **kwargs): old.end_time != instance.end_time or old.channel_id != instance.channel_id ): + # Do NOT revoke while the recording is actively streaming. + # run_recording re-reads end_time from the DB every ~2 s and extends + # its internal deadline dynamically — revoking here would kill the task. + old_status = (old.custom_properties or {}).get("status", "") + if old_status == "recording": + return revoke_task(old.task_id) instance.task_id = None except Recording.DoesNotExist: @@ -176,6 +182,13 @@ def revoke_old_task_on_update(sender, instance, **kwargs): @receiver(post_save, sender=Recording) def schedule_task_on_save(sender, instance, created, **kwargs): try: + # Skip processing for internal field-only saves (metadata updates, + # task_id assignment, end_time extensions) to prevent re-entrant + # artwork dispatch and redundant recording_updated WS events. + update_fields = kwargs.get('update_fields') + if not created and update_fields is not None and set(update_fields) <= {'custom_properties', 'task_id', 'end_time'}: + return + if not instance.task_id: start_time = instance.start_time diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 87abb90d..26df8c78 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -1503,11 +1503,8 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): try: rec_check = Recording.objects.filter(id=recording_id).only("custom_properties").first() if not rec_check: - # Recording was deleted before this task got a chance to run (e.g. user - # cancelled from the Upcoming list, or destroy() ran concurrently). logger.info( - f"run_recording called for recording {recording_id} but it no longer exists — " - f"skipping (recording was cancelled before task started)." + f"run_recording called for recording {recording_id} but it no longer exists — skipping." ) return status = (rec_check.custom_properties or {}).get("status", "") @@ -1520,18 +1517,10 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): except Exception as e: logger.debug(f"Idempotency guard check failed (proceeding anyway): {e}") - # --- Clean up the one-off PeriodicTask that dispatched task --- + # --- Clean up the one-off PeriodicTask that dispatched this task --- try: - from django_celery_beat.models import ClockedSchedule as _CS, PeriodicTask as _PT - task_name = f"dvr-recording-{recording_id}" - try: - pt = _PT.objects.get(name=task_name) - old_clocked = pt.clocked - pt.delete() - if old_clocked and not _PT.objects.filter(clocked=old_clocked).exists(): - old_clocked.delete() - except _PT.DoesNotExist: - pass + from apps.channels.signals import revoke_task, _dvr_task_name + revoke_task(_dvr_task_name(recording_id)) except Exception as e: logger.debug(f"PeriodicTask cleanup failed (non-fatal): {e}") @@ -1541,8 +1530,7 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): end_time = datetime.fromisoformat(end_time_str) duration_seconds = int((end_time - start_time).total_seconds()) - # Build output paths from templates - # We need program info; will refine after we load Recording cp below + # Build output paths from templates (refined after loading Recording cp below) filename = None final_path = None temp_ts_path = None @@ -1575,6 +1563,16 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): recording_obj = None try: recording_obj = Recording.objects.get(id=recording_id) + # If the stop endpoint already wrote "stopped" before the task started, + # honor it instead of overwriting with "recording". + _pre_cp = recording_obj.custom_properties or {} + if _pre_cp.get("status") == "stopped": + logger.info( + f"run_recording {recording_id}: 'stopped' found in DB before stream started " + f"— task exits without connecting." + ) + return + # Prime custom_properties with file info/status cp = recording_obj.custom_properties or {} cp.update({ @@ -1817,8 +1815,38 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): except Exception: pass - recording_obj.custom_properties = cp + # Re-read from DB to preserve concurrent changes (e.g., artwork + # prefetch may have saved poster/rating info while resolving). + recording_obj.refresh_from_db() + fresh_cp = recording_obj.custom_properties or {} + + # If the stop endpoint set "stopped" while resolving, honor it. + if fresh_cp.get("status") == "stopped": + logger.info( + f"run_recording {recording_id}: 'stopped' found after metadata " + f"prep — task exits without streaming." + ) + return + + # Merge only the keys explicitly set into the fresh copy + for key in ("status", "started_at", "file_url", "output_file_url", + "file_name", "file_path", "_temp_file_path", + "poster_logo_id", "poster_url"): + if key in cp: + fresh_cp[key] = cp[key] + recording_obj.custom_properties = fresh_cp recording_obj.save(update_fields=["custom_properties"]) + + # Notify frontends so the tile picks up poster/metadata immediately + try: + from core.utils import send_websocket_update + send_websocket_update('updates', 'update', { + "success": True, + "type": "recording_updated", + "recording_id": recording_id, + }) + except Exception: + pass except Exception as e: logger.debug(f"Unable to prime Recording metadata: {e}") interrupted = False @@ -1888,7 +1916,12 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): elapsed = time.time() - started_at if elapsed > duration_seconds: break - # Continue draining the stream + # Continue draining the stream. + # Periodically poll the DB for a user-initiated stop so + # the task exits promptly if stop_client() was missed + # (e.g., fired before the DVR client was registered in Redis). + _stop_poll_interval = 2.0 # seconds between DB checks + _last_stop_poll = time.time() for chunk2 in response.iter_content(chunk_size=8192): if not chunk2: continue @@ -1897,6 +1930,50 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): elapsed = time.time() - started_at if elapsed > duration_seconds: break + # Periodic stopped-status + end_time extension check + _now = time.time() + if _now - _last_stop_poll >= _stop_poll_interval: + _last_stop_poll = _now + try: + _sc = Recording.objects.filter( + id=recording_id + ).only("custom_properties", "end_time").first() + if _sc is None: + # Recording was deleted — exit gracefully. + logger.info( + f"DVR recording {recording_id}: " + f"deleted — exiting stream loop" + ) + interrupted = False + break + if (_sc.custom_properties or {}).get("status") == "stopped": + logger.info( + f"DVR recording {recording_id}: stop requested " + f"— exiting stream loop early" + ) + break + # Check for extended end_time + try: + new_end = _sc.end_time + if new_end is not None: + from django.utils import timezone as _tz + if _tz.is_naive(new_end): + new_end = _tz.make_aware(new_end) + _ref = start_time + if _tz.is_naive(_ref): + _ref = _tz.make_aware(_ref) + new_duration = int((new_end - _ref).total_seconds()) + if new_duration > duration_seconds: + logger.info( + f"run_recording {recording_id}: " + f"end_time extended to {new_end}, " + f"new duration {new_duration}s" + ) + duration_seconds = new_duration + except Exception: + pass + except Exception: + pass break # exit outer for-loop once we switched to full drain # If we wrote any bytes, treat as success and stop trying candidates @@ -1933,16 +2010,14 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): interrupted = True interrupted_reason = f"stream_interrupted: {e}" break - # Also bail if the Recording was deleted (user cancelled before data arrived) - rec_exists = Recording.objects.filter(id=recording_id).exists() - if not rec_exists: - interrupted = True - interrupted_reason = "recording_deleted_during_connect" - break - # Check for "stopped" status even before data arrived + # Check if recording was deleted or stopped before data arrived try: rec_check = Recording.objects.filter(id=recording_id).only("custom_properties").first() - if rec_check and (rec_check.custom_properties or {}).get("status") == "stopped": + if rec_check is None: + interrupted = True + interrupted_reason = "recording_deleted_during_connect" + break + if (rec_check.custom_properties or {}).get("status") == "stopped": interrupted = False break except Exception: @@ -1951,15 +2026,6 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): if chosen_base is None and bytes_written == 0: interrupted = True interrupted_reason = f"no_stream_data: {last_error or 'all_bases_failed'}" - else: - # If we ended before reaching planned duration, record reason - actual_elapsed = 0 - try: - actual_elapsed = os.path.getsize(temp_ts_path) and (duration_seconds) # Best effort; we streamed until duration or disconnect above - except Exception: - pass - # We cannot compute accurate elapsed here; fine to leave as is - pass # If no bytes were written at all, check whether this was a deliberate stop or a # genuine failure. The exception handler above already sets interrupted=False when @@ -2024,10 +2090,13 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): logger.info(f"Recording {recording_id} was cancelled — skipping remux and metadata.") # Clean up any remaining temp or stub output files for _cleanup_path in [temp_ts_path, final_path]: + if not _cleanup_path: + continue try: - if _cleanup_path and os.path.exists(_cleanup_path): - os.remove(_cleanup_path) - logger.info(f"Cleaned up cancelled recording artifact: {_cleanup_path}") + os.remove(_cleanup_path) + logger.info(f"Cleaned up cancelled recording artifact: {_cleanup_path}") + except FileNotFoundError: + pass except Exception: pass return @@ -2138,12 +2207,14 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): cp = recording_obj.custom_properties or {} cp["ended_at"] = str(datetime.now()) - # Determine final status. - # A deliberate user stop ("stopped" in DB) always wins over any interrupted flag — - # this handles races where the stream connection broke milliseconds before or - # after the stop endpoint committed its DB save. + # Final status priority: stopped > completed > interrupted. + # "stopped" is set by the stop endpoint before stream teardown, so + # refresh_from_db() above guarantees it is visible here. db_status_now = cp.get("status", "") - if db_status_now == "stopped" or not interrupted: + if db_status_now == "stopped": + # Deliberate user stop — preserve; do not overwrite with "completed". + cp.pop("interrupted_reason", None) + elif not interrupted: cp["status"] = "completed" cp.pop("interrupted_reason", None) else: @@ -2307,7 +2378,8 @@ def recover_recordings_on_startup(): for rec in upcoming: try: from django_celery_beat.models import PeriodicTask as _PT - task_name = f"dvr-recording-{rec.id}" + from apps.channels.signals import _dvr_task_name + task_name = _dvr_task_name(rec.id) if _PT.objects.filter(name=task_name).exists(): if rec.task_id != task_name: rec.task_id = task_name diff --git a/apps/channels/tests/test_recording_extend.py b/apps/channels/tests/test_recording_extend.py new file mode 100644 index 00000000..a083b6d3 --- /dev/null +++ b/apps/channels/tests/test_recording_extend.py @@ -0,0 +1,235 @@ +"""Tests for the Extend In-Progress Recording feature. + +Covers: + - extend() API endpoint (happy path and validation) + - pre_save signal guard: end_time change must NOT revoke a live recording + - pre_save signal guard: end_time change MUST still revoke an upcoming recording + - TOCTOU edge cases (extend on a completed/stopped/nonexistent recording) +""" +from datetime import timedelta +from unittest.mock import MagicMock, patch + +from django.test import TestCase +from django.utils import timezone +from rest_framework.test import APIRequestFactory, force_authenticate + +from apps.channels.models import Channel, Recording +from apps.channels.api_views import RecordingViewSet + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_admin(): + from django.contrib.auth import get_user_model + User = get_user_model() + u, _ = User.objects.get_or_create( + username="extend_test_admin", + defaults={"user_level": User.UserLevel.ADMIN}, + ) + u.set_password("pass") + u.save() + return u + + +# --------------------------------------------------------------------------- +# Extend endpoint tests +# --------------------------------------------------------------------------- + +class ExtendEndpointTests(TestCase): + """Tests for POST /api/channels/recordings/{id}/extend/""" + + def setUp(self): + self.channel = Channel.objects.create( + channel_number=88, name="Extend Test Channel" + ) + self.user = _make_admin() + self.factory = APIRequestFactory() + + def _extend(self, rec, extra_minutes): + request = self.factory.post( + f"/api/channels/recordings/{rec.id}/extend/", + {"extra_minutes": extra_minutes}, + format="json", + ) + force_authenticate(request, user=self.user) + view = RecordingViewSet.as_view({"post": "extend"}) + return view(request, pk=rec.id) + + def _make_rec(self, status="recording"): + now = timezone.now() + return Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(hours=1), + end_time=now + timedelta(hours=1), + custom_properties={"status": status}, + ) + + @patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None) + def test_extend_updates_end_time_in_db(self, _ws): + rec = self._make_rec() + original_end = rec.end_time + response = self._extend(rec, 30) + self.assertEqual(response.status_code, 200) + self.assertTrue(response.data.get("success")) + rec.refresh_from_db() + expected = original_end + timedelta(minutes=30) + delta = abs((rec.end_time - expected).total_seconds()) + self.assertLess(delta, 1, "end_time was not extended by the correct amount") + + @patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None) + def test_extend_stacks_multiple_extensions(self, _ws): + """Calling extend() twice adds both increments.""" + rec = self._make_rec() + original_end = rec.end_time + self._extend(rec, 15) + self._extend(rec, 30) + rec.refresh_from_db() + expected = original_end + timedelta(minutes=45) + delta = abs((rec.end_time - expected).total_seconds()) + self.assertLess(delta, 1) + + @patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None) + def test_extend_does_not_clear_task_id(self, _ws): + """The running Celery task must survive the DB save.""" + rec = self._make_rec() + rec.task_id = "dvr-recording-999" + rec.save(update_fields=["task_id"]) + self._extend(rec, 30) + rec.refresh_from_db() + self.assertEqual(rec.task_id, "dvr-recording-999") + + def test_extend_returns_400_if_finished(self): + """Cannot extend a completed, stopped, or interrupted recording.""" + for bad_status in ("completed", "stopped", "interrupted"): + with self.subTest(status=bad_status): + rec = self._make_rec(status=bad_status) + response = self._extend(rec, 30) + self.assertEqual(response.status_code, 400) + self.assertFalse(response.data.get("success")) + + @patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None) + def test_extend_succeeds_before_task_sets_status(self, _ws): + """Extend must work when status is empty (task hasn't started yet).""" + rec = self._make_rec(status="") + response = self._extend(rec, 15) + self.assertEqual(response.status_code, 200) + rec.refresh_from_db() + expected = rec.end_time # already extended + self.assertTrue(response.data.get("success")) + + @patch("apps.channels.signals.revoke_task") + @patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None) + def test_extend_bypasses_signals_no_revoke(self, _ws, mock_revoke): + """Extend uses .update() to bypass pre_save — revoke_task must never fire.""" + rec = self._make_rec(status="") + rec.task_id = "dvr-recording-500" + rec.save(update_fields=["task_id"]) + self._extend(rec, 15) + self._extend(rec, 30) + mock_revoke.assert_not_called() + rec.refresh_from_db() + self.assertEqual(rec.task_id, "dvr-recording-500") + + def test_extend_returns_400_for_zero_minutes(self): + response = self._extend(self._make_rec(), 0) + self.assertEqual(response.status_code, 400) + + def test_extend_returns_400_for_negative_minutes(self): + response = self._extend(self._make_rec(), -15) + self.assertEqual(response.status_code, 400) + + def test_extend_returns_400_for_non_numeric_minutes(self): + rec = self._make_rec() + request = self.factory.post( + f"/api/channels/recordings/{rec.id}/extend/", + {"extra_minutes": "lots"}, + format="json", + ) + force_authenticate(request, user=self.user) + view = RecordingViewSet.as_view({"post": "extend"}) + response = view(request, pk=rec.id) + self.assertEqual(response.status_code, 400) + + def test_extend_returns_404_for_nonexistent_recording(self): + request = self.factory.post( + "/api/channels/recordings/999999/extend/", + {"extra_minutes": 30}, + format="json", + ) + force_authenticate(request, user=self.user) + view = RecordingViewSet.as_view({"post": "extend"}) + response = view(request, pk=999999) + self.assertEqual(response.status_code, 404) + + +# --------------------------------------------------------------------------- +# pre_save signal guard tests +# --------------------------------------------------------------------------- + +class PreSaveExtendGuardTests(TestCase): + """The pre_save signal must NOT revoke a live recording when end_time changes, + but MUST still revoke a scheduled (upcoming) recording as before.""" + + def setUp(self): + self.channel = Channel.objects.create( + channel_number=77, name="Signal Guard Channel" + ) + + def _make_rec(self, status="", task_id="dvr-recording-42"): + now = timezone.now() + return Recording.objects.create( + channel=self.channel, + start_time=now + timedelta(hours=1), + end_time=now + timedelta(hours=2), + task_id=task_id, + custom_properties={"status": status} if status else {}, + ) + + @patch("apps.channels.signals.revoke_task") + def test_end_time_change_does_not_revoke_live_recording(self, mock_revoke): + """When status='recording', extending end_time must not call revoke_task.""" + rec = self._make_rec(status="recording", task_id="dvr-recording-42") + rec.end_time = rec.end_time + timedelta(minutes=30) + rec.save(update_fields=["end_time"]) + mock_revoke.assert_not_called() + + @patch("apps.channels.signals.revoke_task") + def test_task_id_preserved_after_extend_on_live_recording(self, mock_revoke): + """task_id must not be cleared for a live recording's end_time change.""" + rec = self._make_rec(status="recording", task_id="dvr-recording-42") + original_task_id = rec.task_id + rec.end_time = rec.end_time + timedelta(minutes=30) + rec.save(update_fields=["end_time"]) + rec.refresh_from_db() + self.assertEqual(rec.task_id, original_task_id) + + @patch("apps.channels.signals.revoke_task") + def test_end_time_change_still_revokes_upcoming_recording(self, mock_revoke): + """The guard must NOT apply to upcoming recordings — existing behavior preserved.""" + rec = self._make_rec(status="", task_id="dvr-recording-77") + rec.end_time = rec.end_time + timedelta(minutes=30) + rec.save(update_fields=["end_time"]) + mock_revoke.assert_called_once_with("dvr-recording-77") + + @patch("apps.channels.signals.revoke_task") + @patch("core.utils.send_websocket_update", side_effect=lambda *a, **kw: None) + def test_pre_save_guard_reads_db_status_not_memory_status(self, _ws, mock_revoke): + """pre_save reads status from DB (old object), not from the instance being saved.""" + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(hours=1), + end_time=now + timedelta(hours=1), + task_id="dvr-recording-66", + custom_properties={"status": "recording"}, + ) + # Simulate: DB status changes to 'completed' behind the instance's back + Recording.objects.filter(pk=rec.pk).update( + custom_properties={"status": "completed"} + ) + rec.end_time = rec.end_time + timedelta(minutes=30) + rec.save(update_fields=["end_time"]) + # revoke_task should be called because DB status is "completed", not "recording" + mock_revoke.assert_called_once_with("dvr-recording-66") diff --git a/apps/channels/tests/test_recording_scheduling.py b/apps/channels/tests/test_recording_scheduling.py index 56bdbe87..e9daecd7 100644 --- a/apps/channels/tests/test_recording_scheduling.py +++ b/apps/channels/tests/test_recording_scheduling.py @@ -1,7 +1,8 @@ """Tests for DVR recording scheduling with ClockedSchedule. - Prefers Celery ClockedScheduled over Redis time due to 3600s limitation on scheduled tasks; - scheduled tasks beyond 3600s cause Redis time to hand process between workers and duplicate jobs +Uses ClockedSchedule instead of apply_async with countdown because Redis +visibility_timeout (default 3600s) causes task redelivery for long countdowns, +leading to duplicate recordings. """ from datetime import timedelta from unittest.mock import patch, MagicMock @@ -501,7 +502,7 @@ class DestroyDvrClientIsolationTests(TestCase): custom_properties={"status": "recording"}, ) self._delete_recording(rec) - mock_stop.assert_called_once_with(str(self.channel.uuid)) + mock_stop.assert_called_once_with(str(self.channel.uuid), recording_id=rec.id) class PeriodicTaskCleanupOnExecutionTests(TestCase): diff --git a/apps/channels/tests/test_recording_stop_cancel.py b/apps/channels/tests/test_recording_stop_cancel.py new file mode 100644 index 00000000..962160ee --- /dev/null +++ b/apps/channels/tests/test_recording_stop_cancel.py @@ -0,0 +1,341 @@ +"""Tests for the DVR Stop/Cancel feature set. + +Covers: + - stop() endpoint + - destroy() was_in_progress field in recording_cancelled WebSocket event + - signals.py update_fields re-entrancy guard + - run_recording race guard before status write + - _stop_dvr_clients() DVR client isolation +""" +import time +from datetime import timedelta +from unittest.mock import MagicMock, AsyncMock, patch + +from django.test import TestCase +from django.utils import timezone +from rest_framework.test import APIRequestFactory, force_authenticate + +from apps.channels.models import Channel, Recording +from apps.channels.api_views import RecordingViewSet, _stop_dvr_clients + + +def _make_admin(): + from django.contrib.auth import get_user_model + User = get_user_model() + u, _ = User.objects.get_or_create( + username="stop_test_admin", + defaults={"user_level": User.UserLevel.ADMIN}, + ) + u.set_password("pass") + u.save() + return u + + +def _async_channel_layer_mock(): + layer = MagicMock() + layer.group_send = AsyncMock() + return layer + + +class StopEndpointTests(TestCase): + """Tests for POST /api/channels/recordings/{id}/stop/""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=99, name="Stop Test Channel") + self.user = _make_admin() + self.factory = APIRequestFactory() + + def _stop(self, rec): + request = self.factory.post(f"/api/channels/recordings/{rec.id}/stop/") + force_authenticate(request, user=self.user) + view = RecordingViewSet.as_view({"post": "stop"}) + return view(request, pk=rec.id) + + def _make_rec(self, status="recording"): + now = timezone.now() + return Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(hours=1), + end_time=now + timedelta(hours=1), + custom_properties={"status": status}, + ) + + @patch("core.utils.send_websocket_update") + @patch("threading.Thread") + def test_stop_writes_status_to_db_before_returning(self, mock_thread, mock_ws): + """DB write is synchronous — run_recording polls for this.""" + mock_thread.return_value.start = MagicMock() + rec = self._make_rec() + response = self._stop(rec) + self.assertEqual(response.status_code, 200) + self.assertTrue(response.data.get("success")) + rec.refresh_from_db() + self.assertEqual(rec.custom_properties.get("status"), "stopped") + + @patch("core.utils.send_websocket_update") + @patch("threading.Thread") + def test_stop_writes_stopped_at_timestamp(self, mock_thread, mock_ws): + mock_thread.return_value.start = MagicMock() + rec = self._make_rec() + self._stop(rec) + rec.refresh_from_db() + self.assertIn("stopped_at", rec.custom_properties) + + def test_stop_calls_stop_dvr_clients_in_background(self): + rec = self._make_rec() + with patch("apps.channels.api_views._stop_dvr_clients", return_value=1) as mock_stop, \ + patch("core.utils.send_websocket_update"): + self._stop(rec) + time.sleep(0.5) + self.assertTrue(mock_stop.called) + args, kwargs = mock_stop.call_args + actual_rec_id = kwargs.get("recording_id") or (args[1] if len(args) > 1 else None) + self.assertEqual(actual_rec_id, rec.id) + + def test_stop_returns_404_for_nonexistent(self): + request = self.factory.post("/api/channels/recordings/99999/stop/") + force_authenticate(request, user=self.user) + view = RecordingViewSet.as_view({"post": "stop"}) + self.assertEqual(view(request, pk=99999).status_code, 404) + + @patch("core.utils.send_websocket_update") + @patch("threading.Thread") + def test_stop_idempotent_on_already_stopped(self, mock_thread, mock_ws): + mock_thread.return_value.start = MagicMock() + rec = self._make_rec(status="stopped") + self.assertEqual(self._stop(rec).status_code, 200) + + +class CancelDestroyWasInProgressTests(TestCase): + """was_in_progress field in the recording_cancelled WebSocket event.""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=98, name="Cancel Test Channel") + self.user = _make_admin() + self.factory = APIRequestFactory() + + def _delete(self, rec): + request = self.factory.delete(f"/api/channels/recordings/{rec.id}/") + force_authenticate(request, user=self.user) + return RecordingViewSet.as_view({"delete": "destroy"})(request, pk=rec.id) + + @patch("apps.channels.api_views._stop_dvr_clients", return_value=1) + @patch("core.utils.send_websocket_update") + def test_in_progress_sends_was_in_progress_true(self, mock_ws, _): + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(minutes=10), + end_time=now + timedelta(hours=1), + custom_properties={"status": "recording"}, + ) + self._delete(rec) + payload = mock_ws.call_args[0][2] + self.assertEqual(payload["type"], "recording_cancelled") + self.assertTrue(payload["was_in_progress"]) + + @patch("core.utils.send_websocket_update") + def test_completed_sends_was_in_progress_false(self, mock_ws): + rec = Recording.objects.create( + channel=self.channel, + start_time=timezone.now() - timedelta(hours=2), + end_time=timezone.now() - timedelta(hours=1), + custom_properties={"status": "completed"}, + ) + self._delete(rec) + self.assertFalse(mock_ws.call_args[0][2]["was_in_progress"]) + + +class SignalUpdateFieldsReentrancyGuardTests(TestCase): + """update_fields guard in schedule_task_on_save prevents redundant WS events.""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=97, name="Signal Guard Channel") + + def _create_upcoming(self): + future = timezone.now() + timedelta(hours=2) + return Recording.objects.create( + channel=self.channel, start_time=future, + end_time=future + timedelta(hours=1), custom_properties={}, + ) + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_custom_properties_save_skips_artwork(self, mock_artwork): + rec = self._create_upcoming() + mock_artwork.reset_mock() + rec.custom_properties = {"poster_url": "https://example.com/p.jpg"} + rec.save(update_fields=["custom_properties"]) + mock_artwork.apply_async.assert_not_called() + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_task_id_save_skips_artwork(self, mock_artwork): + rec = self._create_upcoming() + mock_artwork.reset_mock() + rec.task_id = "dvr-recording-999" + rec.save(update_fields=["task_id"]) + mock_artwork.apply_async.assert_not_called() + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_combined_metadata_save_skips_artwork(self, mock_artwork): + rec = self._create_upcoming() + mock_artwork.reset_mock() + rec.task_id = "dvr-recording-1000" + rec.custom_properties = {"poster_url": "x"} + rec.save(update_fields=["custom_properties", "task_id"]) + mock_artwork.apply_async.assert_not_called() + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_creation_dispatches_artwork(self, mock_artwork): + mock_artwork.apply_async.return_value = MagicMock() + self._create_upcoming() + self.assertTrue(mock_artwork.apply_async.called) + + @patch("apps.channels.signals.prefetch_recording_artwork") + def test_scheduling_field_update_dispatches_artwork(self, mock_artwork): + """save(update_fields=['start_time']) is not a metadata save — dispatch runs.""" + mock_artwork.apply_async.return_value = MagicMock() + rec = self._create_upcoming() + mock_artwork.reset_mock() + future = timezone.now() + timedelta(hours=3) + rec.start_time = future + rec.end_time = future + timedelta(hours=1) + rec.save(update_fields=["start_time", "end_time"]) + mock_artwork.apply_async.assert_called() + + +class RunRecordingRaceGuardTests(TestCase): + """Race guard: stop() fires between idempotency check and status write.""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=96, name="Race Guard Channel") + + def test_race_guard_exits_when_stopped_at_db_read(self): + """If Recording.objects.get() shows 'stopped', the task must exit + without writing 'recording' to the DB.""" + from apps.channels.tasks import run_recording as run_rec + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(minutes=1), + end_time=now + timedelta(hours=1), + custom_properties={}, + ) + mock_layer = _async_channel_layer_mock() + original_get = Recording.objects.get + + def patched_get(*args, **kwargs): + obj = original_get(*args, **kwargs) + if kwargs.get("id") == rec.id or (args and args[0] == rec.id): + obj.custom_properties = {"status": "stopped"} + return obj + + with patch("apps.channels.tasks.get_channel_layer", return_value=mock_layer), \ + patch("core.utils.log_system_event", side_effect=Exception("skip")), \ + patch.object(Recording.objects, "get", side_effect=patched_get): + result = run_rec( + rec.id, self.channel.id, str(rec.start_time), str(rec.end_time), + ) + + self.assertIsNone(result) + rec.refresh_from_db() + self.assertNotEqual( + rec.custom_properties.get("status"), "recording", + "Race guard failed: task overwrote 'stopped' with 'recording'", + ) + + def test_idempotency_guard_catches_stopped_before_channel_layer(self): + """When status='stopped' at the idempotency check, get_channel_layer is never called.""" + from apps.channels.tasks import run_recording as run_rec + now = timezone.now() + rec = Recording.objects.create( + channel=self.channel, + start_time=now - timedelta(minutes=5), + end_time=now + timedelta(hours=1), + custom_properties={"status": "stopped"}, + ) + with patch("apps.channels.tasks.get_channel_layer") as mock_get_layer: + result = run_rec( + rec.id, self.channel.id, str(rec.start_time), str(rec.end_time), + ) + self.assertIsNone(result) + mock_get_layer.assert_not_called() + + +class StopDvrClientsTests(TestCase): + """_stop_dvr_clients() DVR client isolation.""" + + def setUp(self): + self.channel = Channel.objects.create(channel_number=95, name="DVR Clients Channel") + self._redis = "core.utils.RedisClient" + self._sc = "apps.proxy.ts_proxy.services.channel_service.ChannelService.stop_client" + self._sch = "apps.proxy.ts_proxy.services.channel_service.ChannelService.stop_channel" + + def _mock_redis(self, client_ids, ua_map): + r = MagicMock() + r.smembers.return_value = {c.encode() for c in client_ids} + def hget_side(key, field): + ks = key if isinstance(key, str) else key.decode("utf-8", errors="replace") + for cid, ua in ua_map.items(): + if cid in ks: + return ua.encode() if isinstance(ua, str) else ua + return b"" + r.hget.side_effect = hget_side + return r + + def test_returns_zero_when_redis_none(self): + with patch(self._redis) as rc: + rc.get_client.return_value = None + self.assertEqual(_stop_dvr_clients(str(self.channel.uuid)), 0) + + def test_stops_only_matching_client_when_recording_id_given(self): + r = self._mock_redis( + ["client-a", "client-b"], + {"client-a": "Dispatcharr-DVR/recording-42", + "client-b": "Dispatcharr-DVR/recording-99"}, + ) + with patch(self._redis) as rc, patch(self._sc) as sc: + rc.get_client.return_value = r + result = _stop_dvr_clients(str(self.channel.uuid), recording_id=42) + self.assertEqual(result, 1) + stopped = [c[0][1] for c in sc.call_args_list] + self.assertIn("client-a", stopped) + self.assertNotIn("client-b", stopped) + + def test_stops_all_dvr_clients_without_recording_id(self): + r = self._mock_redis( + ["client-a", "client-b"], + {"client-a": "Dispatcharr-DVR/recording-42", + "client-b": "Dispatcharr-DVR/recording-99"}, + ) + with patch(self._redis) as rc, patch(self._sc) as sc: + rc.get_client.return_value = r + result = _stop_dvr_clients(str(self.channel.uuid)) + self.assertEqual(result, 2) + + def test_skips_non_dvr_clients(self): + r = self._mock_redis( + ["viewer", "dvr-client"], + {"viewer": "Mozilla/5.0", "dvr-client": "Dispatcharr-DVR/recording-1"}, + ) + with patch(self._redis) as rc, patch(self._sc) as sc: + rc.get_client.return_value = r + result = _stop_dvr_clients(str(self.channel.uuid)) + self.assertEqual(result, 1) + stopped = [c[0][1] for c in sc.call_args_list] + self.assertNotIn("viewer", stopped) + + def test_returns_zero_for_empty_channel(self): + r = MagicMock() + r.smembers.return_value = set() + with patch(self._redis) as rc, patch(self._sc) as sc: + rc.get_client.return_value = r + self.assertEqual(_stop_dvr_clients(str(self.channel.uuid)), 0) + sc.assert_not_called() + + def test_never_calls_stop_channel(self): + """Must not stop the whole channel proxy — only individual clients.""" + r = self._mock_redis(["dvr-1"], {"dvr-1": "Dispatcharr-DVR/recording-1"}) + with patch(self._redis) as rc, patch(self._sc), patch(self._sch) as sch: + rc.get_client.return_value = r + _stop_dvr_clients(str(self.channel.uuid)) + sch.assert_not_called() diff --git a/apps/channels/tests/test_ts_proxy_keepalive.py b/apps/channels/tests/test_ts_proxy_keepalive.py new file mode 100644 index 00000000..db388b3d --- /dev/null +++ b/apps/channels/tests/test_ts_proxy_keepalive.py @@ -0,0 +1,324 @@ +"""Tests for ts_proxy keepalive and stats-update behavior. + +Covers: + - stream_generator._should_send_keepalive() owner vs non-owner worker paths + - stream_generator._should_send_keepalive() Redis last_data health check + - client_manager._do_stats_update() error handling and WebSocket dispatch + - client_manager.remove_client() non-blocking stats update + - Keepalive/DVR-timeout timing invariants +""" +import threading +import time +from unittest.mock import MagicMock, patch + +from django.test import TestCase + + +# --------------------------------------------------------------------------- +# _should_send_keepalive: owner worker path +# --------------------------------------------------------------------------- + +class OwnerWorkerKeepaliveTests(TestCase): + """Owner worker has a stream_manager; keepalive logic uses it directly.""" + + def _make_generator(self, healthy, at_buffer_head, consecutive_empty): + from apps.proxy.ts_proxy.stream_generator import StreamGenerator + gen = StreamGenerator.__new__(StreamGenerator) + gen.channel_id = "00000000-0000-0000-0000-000000000001" + gen.client_id = "test-client" + + buffer = MagicMock() + buffer.index = 10 if at_buffer_head else 100 + gen.local_index = 10 + gen.buffer = buffer + + stream_manager = MagicMock() + stream_manager.healthy = healthy + gen.stream_manager = stream_manager + + gen.consecutive_empty = consecutive_empty + return gen + + def test_owner_healthy_returns_false(self): + """Owner worker, healthy stream -> no keepalive.""" + gen = self._make_generator(healthy=True, at_buffer_head=True, consecutive_empty=10) + self.assertFalse(gen._should_send_keepalive(gen.local_index)) + + def test_owner_unhealthy_at_head_returns_true(self): + """Owner worker, unhealthy stream, at buffer head -> send keepalive.""" + gen = self._make_generator(healthy=False, at_buffer_head=True, consecutive_empty=10) + self.assertTrue(gen._should_send_keepalive(gen.local_index)) + + def test_owner_unhealthy_not_at_head_returns_false(self): + """Owner worker, unhealthy stream, but NOT at buffer head -> no keepalive.""" + gen = self._make_generator(healthy=False, at_buffer_head=False, consecutive_empty=10) + self.assertFalse(gen._should_send_keepalive(gen.local_index)) + + def test_owner_insufficient_consecutive_empty_returns_false(self): + """Owner worker, unhealthy, at head but consecutive_empty < 5 -> no keepalive.""" + gen = self._make_generator(healthy=False, at_buffer_head=True, consecutive_empty=3) + self.assertFalse(gen._should_send_keepalive(gen.local_index)) + + def test_owner_exactly_5_consecutive_empty_returns_true(self): + """consecutive_empty == 5 is the minimum threshold.""" + gen = self._make_generator(healthy=False, at_buffer_head=True, consecutive_empty=5) + self.assertTrue(gen._should_send_keepalive(gen.local_index)) + + +# --------------------------------------------------------------------------- +# _should_send_keepalive: non-owner worker path +# --------------------------------------------------------------------------- + +class NonOwnerWorkerKeepaliveTests(TestCase): + """Non-owner worker has stream_manager=None; health determined from Redis.""" + + def _make_generator(self, consecutive_empty=10): + from apps.proxy.ts_proxy.stream_generator import StreamGenerator + gen = StreamGenerator.__new__(StreamGenerator) + gen.channel_id = "00000000-0000-0000-0000-000000000002" + gen.client_id = "test-client-nonowner" + + buffer = MagicMock() + buffer.index = 10 + gen.local_index = 10 + gen.buffer = buffer + + gen.stream_manager = None # non-owner worker + gen.consecutive_empty = consecutive_empty + return gen + + def _mock_proxy_server(self, last_data_value): + """Return a mock ProxyServer with a redis_client pre-configured.""" + server = MagicMock() + redis_client = MagicMock() + server.redis_client = redis_client + redis_client.get.return_value = last_data_value + return server + + def test_non_owner_fresh_data_returns_false(self): + """Non-owner, last_data < 10s ago -> stream healthy -> no keepalive.""" + gen = self._make_generator() + fresh_ts = str(time.time() - 2.0).encode() + server = self._mock_proxy_server(fresh_ts) + + with patch("apps.proxy.ts_proxy.stream_generator.ProxyServer") as MockPS: + MockPS.get_instance.return_value = server + result = gen._should_send_keepalive(gen.local_index) + + self.assertFalse(result, "Fresh data should NOT trigger keepalive") + + def test_non_owner_stale_data_returns_true(self): + """Non-owner, last_data >= 10s ago -> stream unhealthy -> send keepalive.""" + gen = self._make_generator() + stale_ts = str(time.time() - 12.0).encode() + server = self._mock_proxy_server(stale_ts) + + with patch("apps.proxy.ts_proxy.stream_generator.ProxyServer") as MockPS: + MockPS.get_instance.return_value = server + result = gen._should_send_keepalive(gen.local_index) + + self.assertTrue(result, "Stale data (12s) should trigger keepalive") + + def test_non_owner_exactly_at_timeout_returns_true(self): + """Data age exactly equal to CONNECTION_TIMEOUT (10s) -> send keepalive.""" + gen = self._make_generator() + ts = str(time.time() - 10.0).encode() + server = self._mock_proxy_server(ts) + + with patch("apps.proxy.ts_proxy.stream_generator.ProxyServer") as MockPS: + MockPS.get_instance.return_value = server + result = gen._should_send_keepalive(gen.local_index) + + self.assertTrue(result, "Data at exactly timeout threshold should trigger keepalive") + + def test_non_owner_no_redis_key_returns_true(self): + """Non-owner, last_data key missing from Redis -> assume unhealthy.""" + gen = self._make_generator() + server = self._mock_proxy_server(None) + + with patch("apps.proxy.ts_proxy.stream_generator.ProxyServer") as MockPS: + MockPS.get_instance.return_value = server + result = gen._should_send_keepalive(gen.local_index) + + self.assertTrue(result, "Missing last_data key should trigger keepalive") + + def test_non_owner_redis_client_none_returns_false(self): + """Non-owner, redis_client is None (disconnected) -> conservative, no keepalive.""" + gen = self._make_generator() + server = MagicMock() + server.redis_client = None + + with patch("apps.proxy.ts_proxy.stream_generator.ProxyServer") as MockPS: + MockPS.get_instance.return_value = server + result = gen._should_send_keepalive(gen.local_index) + + self.assertFalse(result, "No redis_client -> conservative, no keepalive") + + def test_non_owner_redis_exception_returns_false(self): + """Non-owner, Redis raises an exception -> conservative, no keepalive.""" + gen = self._make_generator() + + with patch("apps.proxy.ts_proxy.stream_generator.ProxyServer") as MockPS: + MockPS.get_instance.side_effect = Exception("Redis error") + result = gen._should_send_keepalive(gen.local_index) + + self.assertFalse(result, "Redis error -> conservative, no keepalive") + + def test_non_owner_not_at_buffer_head_returns_false(self): + """Non-owner, NOT at buffer head -> no keepalive regardless of Redis.""" + gen = self._make_generator() + gen.buffer.index = 100 # far ahead of local_index=10 + server = self._mock_proxy_server(None) + + with patch("apps.proxy.ts_proxy.stream_generator.ProxyServer") as MockPS: + MockPS.get_instance.return_value = server + result = gen._should_send_keepalive(gen.local_index) + + self.assertFalse(result) + + def test_non_owner_insufficient_consecutive_empty_returns_false(self): + """Non-owner, at head, but consecutive_empty < 5 -> no keepalive.""" + gen = self._make_generator(consecutive_empty=2) + stale_ts = str(time.time() - 30.0).encode() + server = self._mock_proxy_server(stale_ts) + + with patch("apps.proxy.ts_proxy.stream_generator.ProxyServer") as MockPS: + MockPS.get_instance.return_value = server + result = gen._should_send_keepalive(gen.local_index) + + self.assertFalse(result) + + +# --------------------------------------------------------------------------- +# _do_stats_update: error handling and WebSocket dispatch +# --------------------------------------------------------------------------- + +class DoStatsUpdateTests(TestCase): + """_do_stats_update runs the actual Redis scan + WebSocket call.""" + + def _make_client_manager(self): + from apps.proxy.ts_proxy.client_manager import ClientManager + cm = ClientManager.__new__(ClientManager) + cm.channel_id = "00000000-0000-0000-0000-000000000004" + cm._heartbeat_running = False + return cm + + def test_do_stats_update_calls_send_websocket_update(self): + """_do_stats_update must call send_websocket_update with channel_stats.""" + cm = self._make_client_manager() + + mock_redis = MagicMock() + mock_redis.scan.return_value = (0, []) + + with patch("apps.proxy.ts_proxy.client_manager.send_websocket_update") as mock_ws, \ + patch("redis.Redis.from_url", return_value=mock_redis): + cm._do_stats_update() + + mock_ws.assert_called_once() + event_type = mock_ws.call_args[0][1] + self.assertEqual(event_type, "update") + payload = mock_ws.call_args[0][2] + self.assertEqual(payload["type"], "channel_stats") + + def test_do_stats_update_does_not_raise_on_redis_error(self): + """Redis failure must be swallowed (logged), not propagated.""" + cm = self._make_client_manager() + + with patch("redis.Redis.from_url", side_effect=Exception("Redis down")): + try: + cm._do_stats_update() + except Exception as e: + self.fail(f"_do_stats_update raised an exception: {e}") + + def test_do_stats_update_scans_channel_client_keys(self): + """Must scan for ts_proxy:channel:*:clients pattern.""" + cm = self._make_client_manager() + + mock_redis = MagicMock() + mock_redis.scan.return_value = (0, []) + + with patch("apps.proxy.ts_proxy.client_manager.send_websocket_update"), \ + patch("redis.Redis.from_url", return_value=mock_redis): + cm._do_stats_update() + + scan_call = mock_redis.scan.call_args + self.assertIn("ts_proxy:channel:*:clients", str(scan_call)) + + +# --------------------------------------------------------------------------- +# Integration: remove_client must not block on WebSocket +# --------------------------------------------------------------------------- + +class ClientRemoveIntegrationTests(TestCase): + """When remove_client() fires, _trigger_stats_update must not block.""" + + def test_remove_client_does_not_block_on_websocket(self): + """remove_client() must return quickly even if WebSocket is slow.""" + from apps.proxy.ts_proxy.client_manager import ClientManager + + cm = ClientManager.__new__(ClientManager) + cm.channel_id = "00000000-0000-0000-0000-000000000005" + cm._heartbeat_running = False + cm.clients = {"test-client-1"} + cm.last_heartbeat_time = {"test-client-1": time.time()} + cm.last_active_time = time.time() + cm.client_set_key = f"ts_proxy:channel:{cm.channel_id}:clients" + cm.client_ttl = 60 + cm.worker_id = "worker-1" + cm.proxy_server = MagicMock() + cm.proxy_server.am_i_owner.return_value = False + cm.lock = threading.Lock() + + mock_redis = MagicMock() + mock_redis.hgetall.return_value = {b"ip_address": b"127.0.0.1"} + mock_redis.scard.return_value = 1 + cm.redis_client = mock_redis + + slow_ws_called = threading.Event() + + def slow_websocket(*args, **kwargs): + time.sleep(2.0) + slow_ws_called.set() + + start = time.time() + with patch("apps.proxy.ts_proxy.client_manager.send_websocket_update", side_effect=slow_websocket): + cm.remove_client("test-client-1") + elapsed = time.time() - start + + self.assertLess(elapsed, 1.0, + f"remove_client() blocked for {elapsed:.2f}s waiting for WebSocket " + f"(should dispatch to background thread and return immediately)") + + +# --------------------------------------------------------------------------- +# DVR timeout threshold vs keepalive timing +# --------------------------------------------------------------------------- + +class KeepaliveTimingTests(TestCase): + """Verify that keepalive threshold gives sufficient margin before DVR timeout.""" + + def test_keepalive_threshold_less_than_dvr_timeout(self): + """CONNECTION_TIMEOUT (keepalive trigger) must be < DVR read timeout (15s).""" + from apps.proxy.config import TSConfig as Config + connection_timeout = getattr(Config, "CONNECTION_TIMEOUT", 10) + dvr_read_timeout = 15 # hard-coded in run_recording: timeout=(10, 15) + self.assertLess( + connection_timeout, + dvr_read_timeout, + f"CONNECTION_TIMEOUT ({connection_timeout}s) must be < DVR timeout ({dvr_read_timeout}s) " + f"so keepalives fire before DVR times out", + ) + + def test_keepalive_interval_is_short(self): + """KEEPALIVE_INTERVAL must be short enough to send multiple keepalives in the gap.""" + from apps.proxy.config import TSConfig as Config + interval = getattr(Config, "KEEPALIVE_INTERVAL", 0.5) + connection_timeout = getattr(Config, "CONNECTION_TIMEOUT", 10) + remaining_window = 15 - connection_timeout + self.assertGreater( + remaining_window / interval, + 3, + f"KEEPALIVE_INTERVAL ({interval}s) is too long: only " + f"{remaining_window/interval:.1f} keepalives would fit in the " + f"{remaining_window}s window before DVR timeout", + ) diff --git a/apps/proxy/ts_proxy/client_manager.py b/apps/proxy/ts_proxy/client_manager.py index a361bfa1..3351cf56 100644 --- a/apps/proxy/ts_proxy/client_manager.py +++ b/apps/proxy/ts_proxy/client_manager.py @@ -43,14 +43,20 @@ class ClientManager: self._registered_clients = set() # Track already registered client IDs def _trigger_stats_update(self): - """Trigger a channel stats update via WebSocket""" + """Trigger a channel stats update via WebSocket in a background thread. + + Offloaded so the caller is not blocked. send_websocket_update is + gevent-safe (offloads async_to_sync to a native OS thread). + """ + threading.Thread(target=self._do_stats_update, daemon=True).start() + + def _do_stats_update(self): + """Perform the stats update in the background.""" try: - # Import here to avoid potential import issues from apps.proxy.ts_proxy.channel_status import ChannelStatus import redis from django.conf import settings - # Get all channels from Redis using settings redis_url = getattr(settings, 'REDIS_URL', 'redis://localhost:6379/0') redis_client = redis.Redis.from_url(redis_url, decode_responses=True) all_channels = [] @@ -59,7 +65,6 @@ class ClientManager: while True: cursor, keys = redis_client.scan(cursor, match="ts_proxy:channel:*:clients", count=100) for key in keys: - # Extract channel ID from key parts = key.split(':') if len(parts) >= 4: ch_id = parts[2] @@ -70,7 +75,6 @@ class ClientManager: if cursor == 0: break - # Send WebSocket update using existing infrastructure send_websocket_update( "updates", "update", diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 50404f1d..8ba38fc0 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -381,10 +381,29 @@ class StreamGenerator: """Determine if a keepalive packet should be sent.""" # Check if we're caught up to buffer head at_buffer_head = local_index >= self.buffer.index + if not at_buffer_head or self.consecutive_empty < 5: + return False - # If we're at buffer head and no data is coming, send keepalive - stream_healthy = self.stream_manager.healthy if self.stream_manager else True - return at_buffer_head and not stream_healthy and self.consecutive_empty >= 5 + if self.stream_manager is not None: + # Owner worker: use the in-memory health flag directly. + return not self.stream_manager.healthy + else: + # Non-owner worker: stream_manager only exists in the owner process. + # Approximate health from the Redis last_data timestamp; if stale + # beyond CONNECTION_TIMEOUT, send keepalives to prevent DVR timeout. + try: + proxy_server = ProxyServer.get_instance() + if proxy_server.redis_client: + raw = proxy_server.redis_client.get(RedisKeys.last_data(self.channel_id)) + if raw: + age = time.time() - float(raw) + timeout_threshold = getattr(Config, 'CONNECTION_TIMEOUT', 10) + return age >= timeout_threshold + # No timestamp in Redis → key missing or expired → unhealthy + return True + except Exception: + pass + return False def _is_ghost_client(self, local_index): """Check if this appears to be a ghost client (stuck but buffer advancing).""" diff --git a/core/utils.py b/core/utils.py index f0e02bb4..ef5aa702 100644 --- a/core/utils.py +++ b/core/utils.py @@ -294,30 +294,34 @@ def send_websocket_update(group_name, event_type, data, collect_garbage=False): """ Standardized function to send WebSocket updates with proper memory management. - Args: - group_name: The WebSocket group to send to (e.g. 'updates') - event_type: The type of message (e.g. 'update') - data: The data to send - collect_garbage: Whether to force garbage collection after sending + In uWSGI + gevent deployments, async_to_sync creates an asyncio event loop + whose native EpollSelector blocks the entire OS thread, freezing all gevent + greenlets in the worker. To avoid this, the actual send is offloaded to a + real OS thread from gevent's native threadpool when monkey-patching is active. """ channel_layer = get_channel_layer() - try: - async_to_sync(channel_layer.group_send)( - group_name, - { - 'type': event_type, - 'data': data - } - ) - except Exception as e: - logger.warning(f"Failed to send WebSocket update: {e}") - finally: - # Explicitly release references to help garbage collection - channel_layer = None + message = {'type': event_type, 'data': data} - # Force garbage collection if requested - if collect_garbage: - gc.collect() + def _do_send(): + try: + async_to_sync(channel_layer.group_send)(group_name, message) + except Exception as e: + logger.warning(f"Failed to send WebSocket update: {e}") + + try: + import gevent.monkey + if gevent.monkey.is_module_patched('threading'): + from gevent import get_hub + get_hub().threadpool.spawn(_do_send) + return + except Exception: + pass + + # Not in a gevent-patched environment — call directly + _do_send() + + if collect_garbage: + gc.collect() def send_websocket_event(event, success, data): """Acquire a lock to prevent concurrent task execution.""" diff --git a/frontend/src/WebSocket.jsx b/frontend/src/WebSocket.jsx index f595b747..e687198c 100644 --- a/frontend/src/WebSocket.jsx +++ b/frontend/src/WebSocket.jsx @@ -19,6 +19,21 @@ import useAuthStore from './store/auth'; export const WebsocketContext = createContext([false, () => {}, null]); +// Debounce: coalesces rapid recording WS events into a single fetchRecordings() +// call (400 ms window) to prevent redundant re-renders in the TV Guide. +let _recordingFetchTimer = null; +function scheduleRecordingFetch() { + if (_recordingFetchTimer) clearTimeout(_recordingFetchTimer); + _recordingFetchTimer = setTimeout(async () => { + _recordingFetchTimer = null; + try { + await useChannelsStore.getState().fetchRecordings(); + } catch (e) { + console.warn('Failed to refresh recordings:', e); + } + }, 400); +} + export const WebsocketProvider = ({ children }) => { const [isReady, setIsReady] = useState(false); const [val, setVal] = useState(null); @@ -193,9 +208,7 @@ export const WebsocketProvider = ({ children }) => { loading: false, autoClose: 4000, }); - try { - await useChannelsStore.getState().fetchRecordings(); - } catch {} + scheduleRecordingFetch(); } else if (status === 'skipped') { notifications.update({ id, @@ -205,9 +218,7 @@ export const WebsocketProvider = ({ children }) => { loading: false, autoClose: 3000, }); - try { - await useChannelsStore.getState().fetchRecordings(); - } catch {} + scheduleRecordingFetch(); } else if (status === 'error') { notifications.update({ id, @@ -217,9 +228,7 @@ export const WebsocketProvider = ({ children }) => { loading: false, autoClose: 6000, }); - try { - await useChannelsStore.getState().fetchRecordings(); - } catch {} + scheduleRecordingFetch(); } break; } @@ -508,19 +517,11 @@ export const WebsocketProvider = ({ children }) => { break; case 'recording_updated': - try { - await useChannelsStore.getState().fetchRecordings(); - } catch (e) { - console.warn('Failed to refresh recordings on update:', e); - } + scheduleRecordingFetch(); break; case 'recordings_refreshed': - try { - await useChannelsStore.getState().fetchRecordings(); - } catch (e) { - console.warn('Failed to refresh recordings on refreshed:', e); - } + scheduleRecordingFetch(); break; case 'recording_started': @@ -528,11 +529,7 @@ export const WebsocketProvider = ({ children }) => { title: 'Recording started!', message: `Started recording channel ${parsedEvent.data.channel}`, }); - try { - await useChannelsStore.getState().fetchRecordings(); - } catch (e) { - console.warn('Failed to refresh recordings on start:', e); - } + scheduleRecordingFetch(); break; case 'recording_ended': @@ -540,11 +537,7 @@ export const WebsocketProvider = ({ children }) => { title: 'Recording finished!', message: `Stopped recording channel ${parsedEvent.data.channel}`, }); - try { - await useChannelsStore.getState().fetchRecordings(); - } catch (e) { - console.warn('Failed to refresh recordings on end:', e); - } + scheduleRecordingFetch(); break; case 'recording_stopped': @@ -553,11 +546,11 @@ export const WebsocketProvider = ({ children }) => { message: `Recording stopped early for ${parsedEvent.data.channel || 'channel'}. Partial content has been saved.`, color: 'yellow', }); - try { - await useChannelsStore.getState().fetchRecordings(); - } catch (e) { - console.warn('Failed to refresh recordings on stop:', e); - } + scheduleRecordingFetch(); + break; + + case 'recording_extended': + scheduleRecordingFetch(); break; case 'recording_cancelled': @@ -568,10 +561,12 @@ export const WebsocketProvider = ({ children }) => { : 'Recording deleted.', color: 'red', }); - try { - await useChannelsStore.getState().fetchRecordings(); - } catch (e) { - console.warn('Failed to refresh recordings on cancel:', e); + // Surgical removal by ID avoids a full fetchRecordings() re-render. + // Fall back to a full refresh if the ID is missing (e.g. older server). + if (parsedEvent.data.recording_id != null) { + useChannelsStore.getState().removeRecording(parsedEvent.data.recording_id); + } else { + scheduleRecordingFetch(); } break; diff --git a/frontend/src/api.js b/frontend/src/api.js index 4fbf1018..43aca702 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -2673,6 +2673,20 @@ export default class API { } } + static async extendRecording(id, extraMinutes) { + try { + const resp = await request(`${host}/api/channels/recordings/${id}/extend/`, { + method: 'POST', + body: JSON.stringify({ extra_minutes: extraMinutes }), + headers: { 'Content-Type': 'application/json' }, + }); + return resp; + } catch (e) { + errorNotification(`Failed to extend recording ${id}`, e); + throw e; + } + } + static async runComskip(recordingId) { try { const resp = await request( diff --git a/frontend/src/components/cards/RecordingCard.jsx b/frontend/src/components/cards/RecordingCard.jsx index 2f428768..2386748e 100644 --- a/frontend/src/components/cards/RecordingCard.jsx +++ b/frontend/src/components/cards/RecordingCard.jsx @@ -13,20 +13,21 @@ import { Box, Button, Card, - Center, Flex, Group, Image, + Menu, Modal, Stack, Text, Tooltip, } from '@mantine/core'; -import { AlertTriangle, Square, SquareX } from 'lucide-react'; +import { AlertTriangle, Plus, Square, SquareX } from 'lucide-react'; import RecordingSynopsis from '../RecordingSynopsis'; import { deleteRecordingById, deleteSeriesAndRule, + extendRecordingById, getPosterUrl, getRecordingUrl, getSeasonLabel, @@ -43,7 +44,6 @@ const RecordingCard = ({ onOpenRecurring, channel: channelProp = null, }) => { - const channels = useChannelsStore((s) => s.channels); const env_mode = useSettingsStore((s) => s.environment.env_mode); const showVideo = useVideoStore((s) => s.showVideo); const fetchRecordings = useChannelsStore((s) => s.fetchRecordings); @@ -118,6 +118,27 @@ const RecordingCard = ({ } }; + const handleExtend = async (minutes, e) => { + e?.stopPropagation?.(); + try { + await extendRecordingById(recording.id, minutes); + notifications.show({ + title: 'Recording extended', + message: `Added ${minutes} minutes to this recording`, + color: 'teal', + autoClose: 2000, + }); + } catch (error) { + console.error('Failed to extend recording', error); + notifications.show({ + title: 'Extension failed', + message: 'Could not extend the recording', + color: 'red', + autoClose: 3000, + }); + } + }; + // Stop / Cancel / Delete state and handlers const [cancelOpen, setCancelOpen] = React.useState(false); const [stopConfirmOpen, setStopConfirmOpen] = React.useState(false); @@ -312,6 +333,30 @@ const RecordingCard = ({ + {isInProgress && ( + + + + + e.stopPropagation()} + onClick={(e) => e.stopPropagation()} + > + + + + e.stopPropagation()}> + Extend recording by + handleExtend(15, e)}>+15 minutes + handleExtend(30, e)}>+30 minutes + handleExtend(60, e)}>+1 hour + + + + + )} {isInProgress && ( } {!isUpcoming && - customProps?.status === 'completed' && + (customProps?.status === 'completed' || customProps?.status === 'stopped' || customProps?.status === 'interrupted') && (!customProps?.comskip || customProps?.comskip?.status !== 'completed') && ( + @@ -385,11 +484,20 @@ const RecordingDetailsModal = ({ )} - {description && ( + {editing ? ( +