Fixed how a dropped redis connection is handled.

Improved how task cancellation is called from the viewset.

Author: dkliban

pulpcore/app/viewsets/task.py

@@ -1,6 +1,5 @@
 from gettext import gettext as _
 
-from django.conf import settings
 from django.db import transaction
 from django.db.models import Prefetch
 from django_filters.rest_framework import filters
@@ -43,7 +42,7 @@
     CreatedResourcesFilter,
 )
 from pulpcore.constants import TASK_INCOMPLETE_STATES, TASK_STATES
-from pulpcore.tasking.tasks import dispatch
+from pulpcore.tasking.tasks import cancel_task, cancel_task_group, dispatch
 from pulpcore.app.role_util import get_objects_for_user
 
 
@@ -228,16 +227,7 @@ def partial_update(self, request, pk=None, partial=True):
         serializer.is_valid(raise_exception=True)
 
         task = self.get_object()
-
-        # Call the appropriate cancel_task function based on worker type
-        if settings.WORKER_TYPE == "redis":
-            from pulpcore.tasking.redis_tasks import cancel_task
-
-            task = cancel_task(task.pk)
-        else:
-            from pulpcore.tasking.tasks import cancel_task
-
-            task = cancel_task(task.pk)
+        task = cancel_task(task.pk)
 
         # Check whether task is actually canceled
         http_status = (
@@ -358,15 +348,7 @@ def partial_update(self, request, pk=None, partial=True):
             ):
                 raise PermissionDenied()
 
-        # Call the appropriate cancel_task_group function based on worker type
-        if settings.WORKER_TYPE == "redis":
-            from pulpcore.tasking.redis_tasks import cancel_task_group
-
-            task_group = cancel_task_group(task_group.pk)
-        else:
-            from pulpcore.tasking.tasks import cancel_task_group
-
-            task_group = cancel_task_group(task_group.pk)
+        task_group = cancel_task_group(task_group.pk)
         # Check whether task group is actually canceled
         serializer = TaskGroupSerializer(task_group, context={"request": request})
         task_statuses = (

pulpcore/tasking/redis_locks.py

@@ -323,7 +323,7 @@ def acquire_locks(redis_conn, lock_owner, task_lock_key, exclusive_resources, sh
               (includes "__task_lock__" if task lock is held by another worker)
     """
     if not redis_conn:
-        return []
+        return ["error:no_redis_connection"]
 
     # Sort resources deterministically to prevent deadlocks
     exclusive_resources = sorted(exclusive_resources) if exclusive_resources else []

pulpcore/tasking/redis_worker.py

@@ -206,6 +206,19 @@ def handle_worker_heartbeat(self):
             _logger.error(f"Updating the heartbeat of worker {self.name} failed.")
             self.shutdown_requested = True
 
+    def handle_redis_heartbeat(self):
+        """
+        Check Redis connection health.
+
+        If the check fails, the worker is shut down, similar to how a PostgreSQL
+        heartbeat failure triggers shutdown in handle_worker_heartbeat().
+        """
+        try:
+            self.redis_conn.ping()
+        except Exception:
+            _logger.error("Redis connection check failed for worker %s. Shutting down.", self.name)
+            self.shutdown_requested = True
+
     def cleanup_ignored_tasks(self):
         """Remove tasks from ignored list that are no longer incomplete."""
         for pk in (
@@ -382,6 +395,7 @@ def beat(self):
         now = timezone.now()
         if self.app_status.last_heartbeat < now - self.heartbeat_period:
             self.handle_worker_heartbeat()
+            self.handle_redis_heartbeat()
             if self.ignored_task_ids:
                 self.ignored_task_countdown -= 1
                 if self.ignored_task_countdown <= 0:

pulpcore/tasking/tasks.py

@@ -478,6 +478,11 @@ def cancel_task(task_id):
     Raises:
         rest_framework.exceptions.NotFound: If a task with given task_id does not exist
     """
+    if settings.WORKER_TYPE == "redis":
+        from pulpcore.tasking.redis_tasks import cancel_task as redis_cancel_task
+
+        return redis_cancel_task(task_id)
+
     task = Task.objects.select_related("pulp_domain").get(pk=task_id)
 
     if task.state in TASK_FINAL_STATES:
@@ -515,6 +520,11 @@ def cancel_task_group(task_group_id):
     Raises:
         TaskGroup.DoesNotExist: If a task group with given task_group_id does not exist
     """
+    if settings.WORKER_TYPE == "redis":
+        from pulpcore.tasking.redis_tasks import cancel_task_group as redis_cancel_task_group
+
+        return redis_cancel_task_group(task_group_id)
+
     task_group = TaskGroup.objects.get(pk=task_group_id)
     task_group.all_tasks_dispatched = True
     task_group.save(update_fields=["all_tasks_dispatched"])