Store shared resources in Redis.

[dkliban]

This moves all the resource locking logic into a Lua script executed atomically in Redis.

Summary by Sourcery

Centralize resource locking for workers by delegating both exclusive and shared lock management to a Redis Lua script and wiring it into task acquisition and release.

Enhancements:

• Extend worker resource lock release to handle both exclusive and shared resources.
• Simplify task fetching by removing Python-side checks for shared vs exclusive resource conflicts in favor of atomic Redis-based lock acquisition.
• Track locked shared resources on tasks so they can be released alongside exclusive locks when tasks are not executed or on cleanup.

[sourcery-ai]

Reviewer's Guide

Centralizes task resource locking in a single atomic Redis Lua script that handles both exclusive and shared resources, updates worker lock acquisition/release flows to use it, and removes previous Python-side shared-resource coordination logic.

Sequence diagram for atomic Redis Lua-based task lock acquisition

sequenceDiagram
    actor Worker
    participant NewWorker
    participant Redis
    participant LuaScript as AcquireLocksLua

    Worker->>NewWorker: fetch_task()
    loop Iterate_candidate_tasks
        NewWorker->>Redis: SET task:task_id worker_name NX EX 86400
        alt task_lock_acquired
            NewWorker->>LuaScript: acquire_locks(redis_conn, worker_name, exclusive_resources, shared_resources)
            LuaScript->>Redis: Atomic lock operations
            Redis-->>LuaScript: blocked_resource_list
            LuaScript-->>NewWorker: blocked_resource_list
            alt All_locks_acquired (blocked_resource_list empty)
                NewWorker->>NewWorker: task._locked_resources = exclusive_resources
                NewWorker->>NewWorker: task._locked_shared_resources = shared_resources
                NewWorker-->>Worker: return task
            else Some_resources_blocked
                NewWorker->>Redis: DEL task:task_id
                NewWorker->>NewWorker: blocked_resources.update(blocked_resource_list)
                NewWorker->>NewWorker: continue to next task
            end
        else Task_lock_not_acquired
            NewWorker->>NewWorker: skip task and continue
        end
    end

Loading

Sequence diagram for releasing exclusive and shared resource locks

sequenceDiagram
    actor Worker
    participant NewWorker
    participant Redis
    participant LuaRelease as ReleaseLocksLua

    Worker->>NewWorker: handle_tasks()
    alt Incompatible_task
        NewWorker->>NewWorker: exclusive_resources = getattr(task, _locked_resources, [])
        NewWorker->>NewWorker: shared_resources = getattr(task, _locked_shared_resources, [])
        alt Has_any_locked_resources
            NewWorker->>LuaRelease: release_resource_locks(redis_conn, worker_name, exclusive_resources, shared_resources)
            LuaRelease->>Redis: Atomic unlock operations
            Redis-->>LuaRelease: ok
            LuaRelease-->>NewWorker: ok
        end
        NewWorker->>Redis: DEL task:task_id
    else Task_executed
        NewWorker->>NewWorker: _execute_task(task)
        alt _execute_task did_not_cleanup_locks
            NewWorker->>NewWorker: exclusive_resources = getattr(task, _locked_resources, [])
            NewWorker->>NewWorker: shared_resources = getattr(task, _locked_shared_resources, [])
            alt Has_any_locked_resources
                NewWorker->>LuaRelease: release_resource_locks(redis_conn, worker_name, exclusive_resources, shared_resources)
                LuaRelease->>Redis: Atomic unlock operations
                Redis-->>LuaRelease: ok
                LuaRelease-->>NewWorker: ok
            end
        end
    end

Loading

Updated class diagram for NewWorker and Redis lock helpers

classDiagram
    class NewWorker {
        +redis_conn
        +name
        +fetch_task()
        +handle_tasks()
        -_try_acquire_resource_locks(resources)
        -_release_resource_locks(resources, shared_resources)
    }

    class AcquireLocksLua {
        +acquire_locks(redis_conn, worker_name, exclusive_resources, shared_resources) blocked_resources
    }

    class ReleaseLocksLua {
        +release_resource_locks(redis_conn, worker_name, resources, shared_resources)
    }

    NewWorker --> AcquireLocksLua : uses
    NewWorker --> ReleaseLocksLua : uses

Loading

File-Level Changes

Change	Details	Files
Refactor resource lock acquisition to a single atomic Redis Lua script that manages both exclusive and shared locks.	Replace Python-side checks for shared and exclusive resource conflicts with a call to acquire_locks, which returns a list of blocked resources if acquisition fails. Rename local variable for task lock acquisition to task_lock_acquired for clarity and use it to gate subsequent lock operations. On successful lock acquisition, store both exclusive and shared resource lists on the task object (_locked_resources and _locked_shared_resources) for later cleanup. On lock acquisition failure, release the task lock and add all blocked resources returned from acquire_locks into the blocked_resources set.	pulp_service/pulp_service/tasking/new_worker.py
Extend resource lock release handling to support both exclusive and shared resources via the Lua-based release helper.	Update _release_resource_locks to accept an optional shared_resources list and pass both exclusive and shared resources to release_resource_locks. Adjust handle_tasks to release both exclusive and shared locks when tasks are incompatible or after task handling, using the task's _locked_resources and _locked_shared_resources attributes if present.	pulp_service/pulp_service/tasking/new_worker.py
Remove legacy Python logic that tracked shared-resource usage and exclusive lock conflicts in the worker loop.	Delete the query that precomputes resources_in_shared_use from RUNNING tasks’ reserved_resources_record for enforcing exclusive-vs-shared constraints. Remove per-task Redis EXISTS checks for exclusive locks on shared resources and the corresponding blocked_by_exclusive flow. Remove checks that prevented acquiring exclusive resources when they were in shared use, deferring that logic to the Lua script.	pulp_service/pulp_service/tasking/new_worker.py

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - I've left some high level feedback:

• The release path is now split across multiple branches (incompatible-task handling and the finally block) with slightly different ways of deriving exclusive_resources and shared_resources; consider centralizing this into a single helper to avoid divergence in future changes.
• When calling acquire_locks and _release_resource_locks, it may be safer to normalize exclusive_resources and shared_resources to empty lists at the call site, so the Lua/Redis helpers never have to deal with None and you avoid subtle type-handling edge cases.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The release path is now split across multiple branches (incompatible-task handling and the `finally` block) with slightly different ways of deriving `exclusive_resources` and `shared_resources`; consider centralizing this into a single helper to avoid divergence in future changes.
- When calling `acquire_locks` and `_release_resource_locks`, it may be safer to normalize `exclusive_resources` and `shared_resources` to empty lists at the call site, so the Lua/Redis helpers never have to deal with `None` and you avoid subtle type-handling edge cases.

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[bmbouter]

/retest