fix: clean up stale and orphaned Redis locks

[carlosthe19916]

Problem

When a K8s pod restarts, the new worker gets the same name but Redis still holds locks from the dead predecessor. These orphan locks block resources indefinitely because the periodic cleanup either sees the old AppStatus as still online or it was already deleted without releasing its locks.

Solution

• Introduce a per-owner lock registry (pulp:owner_locks:{owner} Redis SET), maintained atomically inside the acquire/release Lua scripts, so an owner's locks can be cleaned in O(locks held) without scanning the whole keyspace.
• Add release_stale_locks_for_self() at worker startup to release Redis locks left under this worker's name by a dead same-name predecessor. (Stale AppStatus rows are cleaned separately by the periodic missing-worker pass — startup only handles the locks.)
• Add reconcile_orphan_redis_locks() to the periodic cleanup to release locks whose owner has no AppStatus row at all.
• Make cleanup_redis_locks_for_worker() sweep any remaining locks via the registry, with per-task exception isolation so one Redis failure doesn't abort cleanup for the worker's other tasks. A throttled keyspace SCAN is kept only as a rolling-upgrade fallback for locks acquired before the registry existed.

Fixes #7919, #7920.

Test plan

• oci-env test -p pulpcore unit -k "orphan_redis_locks"
• oci-env test -p pulpcore functional -k "test_tasking"

[dkliban]

Review

The design is solid — startup cleanup, successor detection, retry-safe cleanup, WAITING task preservation, and the test coverage are all well done. The main concern is the SCAN-based lock discovery which creates scaling issues.

SCAN cost at scale

cleanup_locks_for_owner() does a full SCAN of all task:* keys and all pulp:resource_lock:* keys, checking each key's value against the owner. collect_lock_owners() does the same to enumerate all owners. These are O(all_locks_in_redis) operations.

Three code paths trigger these scans:

1. release_stale_locks_for_self() — runs at every worker startup unconditionally (even with no prior incarnation). At scale-up from 10 to 150 workers, that's 150 concurrent full SCANs.

2. reconcile_orphan_redis_locks() — runs every ~1000 seconds. Does collect_lock_owners() (full SCAN) plus cleanup_locks_for_owner() for each orphan (another full SCAN each).

3. cleanup_redis_locks_for_worker() — runs per missing worker during periodic cleanup. One full SCAN per missing worker.

With 200+ locks in production (observed on 2026-07-24), each SCAN touches every key. During the incident with 150 workers crash-looping, this would compound the Redis pressure.

Recommendation: per-owner lock registry

Instead of scanning all keys to find locks for a specific owner, maintain a per-owner set in Redis:

Key: pulp:owner_locks:{worker_name}
Type: Redis SET
Members: all lock keys held by this owner (task locks + resource locks)

Add SADD pulp:owner_locks:{owner} <key> to the acquire_locks Lua script and SREM pulp:owner_locks:{owner} <key> to the release_resource_locks Lua script. This is atomic with the lock operations — no sync issues.

Then cleanup_locks_for_owner() becomes:

keys = redis_conn.smembers(f"pulp:owner_locks:{owner}")
for key in keys:
    # delete-if-owner-matches (Lua for atomicity)
redis_conn.delete(f"pulp:owner_locks:{owner}")

This is O(locks_held_by_owner) instead of O(all_locks_in_redis). For a worker that held 5 locks, it reads 5 keys instead of scanning 200+.

For backward compatibility during rolling upgrades: old workers don't write to the registry, so their locks won't appear in it. Keep the SCAN as a fallback only when the registry set doesn't exist for an owner. After a full rollout, all owners will have registries and SCAN is never used.

Per-task exception handling in cleanup_redis_locks_for_worker

The DB-linked cleanup loop (the for task in tasks block) has a single try/except around the entire loop. After #7945 merged, release_resource_locks raises RedisError. A single Redis failure on one task aborts cleanup for all remaining tasks of that worker.

Recommendation: catch per-task exceptions inside the for loop:

for task in tasks:
    try:
        safe_release_task_locks(task, lock_owner=self.name)
        self._fail_incomplete_task(task, worker_name, "Worker has gone missing.")
    except Exception as e:
        _logger.error("Error cleaning up task %s for worker %s: %s", task.pk, worker_name, e)

Non-atomic delete in cleanup_locks_for_owner

The function uses GET key then DEL key as separate commands. Between them, another worker could acquire the same lock key. Use a Lua script for atomic delete-if-owner-matches:

if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
end
return 0

What's good

• Startup cleanup before accepting work — correct placement
• Successor detection avoids releasing locks belonging to new incarnation
• Retry-safe design with bool return from cleanup
• WAITING tasks left reclaimable
• Comprehensive test coverage (12 unit tests)