Release shared resources for async tasks also. (#869)

Author: dkliban

images/assets/patches/0033-Adds-redis-WORKER_TYPE.patch

@@ -1,4 +1,4 @@
-From 30d49b1b5adf0acf72d331b2572f5e54af1c35fb Mon Sep 17 00:00:00 2001
+From 52efb38152f8a4386a7e289cd716741ff6f59f66 Mon Sep 17 00:00:00 2001
 From: Dennis Kliban <dkliban@redhat.com>
 Date: Tue, 25 Nov 2025 11:06:48 -0500
 Subject: [PATCH] Adds 'redis' WORKER_TYPE
@@ -13,12 +13,12 @@ PostgreSQL database.
  pulpcore/app/viewsets/task.py                 |  22 +-
  pulpcore/tasking/_util.py                     |   9 +-
  pulpcore/tasking/entrypoint.py                |  13 +-
- pulpcore/tasking/redis_locks.py               | 245 ++++++
- pulpcore/tasking/redis_tasks.py               | 607 ++++++++++++++
+ pulpcore/tasking/redis_locks.py               | 263 ++++++
+ pulpcore/tasking/redis_tasks.py               | 631 ++++++++++++++
  pulpcore/tasking/redis_worker.py              | 778 ++++++++++++++++++
  pulpcore/tasking/tasks.py                     |  17 +
  pulpcore/tests/functional/api/test_tasking.py | 288 +++++++
- 11 files changed, 2032 insertions(+), 12 deletions(-)
+ 11 files changed, 2074 insertions(+), 12 deletions(-)
  create mode 100644 pulpcore/tasking/redis_locks.py
  create mode 100644 pulpcore/tasking/redis_tasks.py
  create mode 100644 pulpcore/tasking/redis_worker.py
@@ -244,10 +244,10 @@ index fdc9b0de8..a3eb7e01c 100644
 +        raise ValueError(f"Invalid WORKER_TYPE: {worker_type}. Must be 'pulpcore' or 'redis'.")
 diff --git a/pulpcore/tasking/redis_locks.py b/pulpcore/tasking/redis_locks.py
 new file mode 100644
-index 000000000..84e8c6a4b
+index 000000000..0ddc6dfff
 --- /dev/null
 +++ b/pulpcore/tasking/redis_locks.py
-@@ -0,0 +1,245 @@
+@@ -0,0 +1,263 @@
 +"""
 +Redis distributed lock utilities for task resource coordination.
 +
@@ -465,40 +465,58 @@ index 000000000..84e8c6a4b
 +        release_shared_resource_locks(redis_conn, lock_owner, shared_resources)
 +
 +
-+async def async_release_resource_locks(redis_conn, lock_owner, resources):
++async def async_release_resource_locks(redis_conn, lock_owner, resources, shared_resources=None):
 +    """
-+    Async version: Release Redis distributed locks for the given resources.
++    Async version: Release Redis distributed locks for exclusive and shared resources.
 +
-+    Uses a Lua script to ensure we only release locks that we own.
++    Uses a Lua script to ensure we only release exclusive locks that we own.
++    Removes task from shared resource sets.
 +
 +    Args:
 +        redis_conn: Redis connection
 +        lock_owner (str): The identifier of the lock owner
-+        resources (list): List of resource names to release locks for
++        resources (list): List of exclusive resource names to release locks for
++        shared_resources (list): Optional list of shared resource names
 +    """
 +    if not redis_conn:
 +        return
 +
-+    # Register the unlock script
-+    unlock_script = await sync_to_async(redis_conn.register_script)(REDIS_UNLOCK_SCRIPT)
++    # Release exclusive locks
++    if resources:
++        # Register the unlock script
++        unlock_script = await sync_to_async(redis_conn.register_script)(REDIS_UNLOCK_SCRIPT)
 +
-+    for resource in resources:
-+        try:
-+            lock_key = resource_to_lock_key(resource)
-+            # Use Lua script to atomically check and delete only if we own the lock
-+            released = await sync_to_async(unlock_script)(keys=[lock_key], args=[lock_owner])
-+            if released:
-+                _logger.debug("Released lock for resource: %s", resource)
-+            else:
-+                _logger.warning("Lock for resource %s was not owned by %s", resource, lock_owner)
-+        except Exception as e:
-+            _logger.error("Error releasing lock for resource %s: %s", resource, e)
++        for resource in resources:
++            try:
++                lock_key = resource_to_lock_key(resource)
++                # Use Lua script to atomically check and delete only if we own the lock
++                released = await sync_to_async(unlock_script)(keys=[lock_key], args=[lock_owner])
++                if released:
++                    _logger.debug("Released exclusive lock for resource: %s", resource)
++                else:
++                    _logger.warning("Lock for resource %s was not owned by %s", resource, lock_owner)
++            except Exception as e:
++                _logger.error("Error releasing lock for resource %s: %s", resource, e)
++
++    # Release shared resources
++    if shared_resources:
++        for resource in shared_resources:
++            try:
++                lock_key = resource_to_lock_key(resource)
++                # Remove this task from the shared resource set
++                removed = await sync_to_async(redis_conn.srem)(lock_key, lock_owner)
++                if removed:
++                    _logger.debug("Released shared resource: %s", resource)
++                else:
++                    _logger.warning("Shared resource %s did not contain %s", resource, lock_owner)
++            except Exception as e:
++                _logger.error("Error releasing shared resource %s: %s", resource, e)
 diff --git a/pulpcore/tasking/redis_tasks.py b/pulpcore/tasking/redis_tasks.py
 new file mode 100644
-index 000000000..8c759c5db
+index 000000000..cc950f1db
 --- /dev/null
 +++ b/pulpcore/tasking/redis_tasks.py
-@@ -0,0 +1,607 @@
+@@ -0,0 +1,631 @@
 +"""
 +Task dispatch functions for Redis-based worker implementation.
 +
@@ -726,20 +744,32 @@ index 000000000..8c759c5db
 +            return None
 +    finally:
 +        # Release Redis locks if this was an immediate task
-+        if hasattr(task, '_locked_resources') and task._locked_resources:
++        exclusive_resources = getattr(task, '_locked_resources', None)
++        shared_resources = getattr(task, '_locked_shared_resources', None)
++
++        if exclusive_resources or shared_resources:
 +            current_app = AppStatus.objects.current()
 +            redis_conn = get_redis_connection()
 +            lock_owner = current_app.name if current_app else f"immediate-{task.pk}"
 +            _logger.info(
-+                "RESOURCE LOCK RELEASE: Task %s releasing resource locks with owner=%s (AppStatus.current=%s) for resources: %s",
++                "RESOURCE LOCK RELEASE: Task %s releasing %d exclusive and %d shared resource locks with owner=%s (AppStatus.current=%s)",
 +                task.pk,
++                len(exclusive_resources) if exclusive_resources else 0,
++                len(shared_resources) if shared_resources else 0,
++                lock_owner,
++                current_app.name if current_app else "None"
++            )
++            release_resource_locks(
++                redis_conn,
 +                lock_owner,
-+                current_app.name if current_app else "None",
-+                task._locked_resources
++                exclusive_resources or [],
++                shared_resources or []
 +            )
-+            release_resource_locks(redis_conn, lock_owner, task._locked_resources)
-+            # Clear the attribute so worker knows locks were released
-+            del task._locked_resources
++            # Clear the attributes so worker knows locks were released
++            if hasattr(task, '_locked_resources'):
++                del task._locked_resources
++            if hasattr(task, '_locked_shared_resources'):
++                del task._locked_shared_resources
 +
 +
 +async def aexecute_task(task):
@@ -784,20 +814,32 @@ index 000000000..8c759c5db
 +            return None
 +    finally:
 +        # Release Redis locks if this was an immediate task
-+        if hasattr(task, '_locked_resources') and task._locked_resources:
++        exclusive_resources = getattr(task, '_locked_resources', None)
++        shared_resources = getattr(task, '_locked_shared_resources', None)
++
++        if exclusive_resources or shared_resources:
 +            current_app = await sync_to_async(AppStatus.objects.current)()
 +            redis_conn = get_redis_connection()
 +            lock_owner = current_app.name if current_app else f"immediate-{task.pk}"
 +            _logger.info(
-+                "RESOURCE LOCK RELEASE (async): Task %s releasing resource locks with owner=%s (AppStatus.current=%s) for resources: %s",
++                "RESOURCE LOCK RELEASE (async): Task %s releasing %d exclusive and %d shared resource locks with owner=%s (AppStatus.current=%s)",
 +                task.pk,
++                len(exclusive_resources) if exclusive_resources else 0,
++                len(shared_resources) if shared_resources else 0,
++                lock_owner,
++                current_app.name if current_app else "None"
++            )
++            await async_release_resource_locks(
++                redis_conn,
 +                lock_owner,
-+                current_app.name if current_app else "None",
-+                task._locked_resources
++                exclusive_resources or [],
++                shared_resources or []
 +            )
-+            await async_release_resource_locks(redis_conn, lock_owner, task._locked_resources)
-+            # Clear the attribute so worker knows locks were released
-+            del task._locked_resources
++            # Clear the attributes so worker knows locks were released
++            if hasattr(task, '_locked_resources'):
++                del task._locked_resources
++            if hasattr(task, '_locked_shared_resources'):
++                del task._locked_shared_resources
 +
 +
 +def are_resources_available(colliding_resources, task: Task) -> bool: