From 41269d8305cc90dd47b421bdf1e6c156fc566bd3 Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Wed, 12 Aug 2026 15:38:12 +0200 Subject: [PATCH 1/3] fix: clean up stale and orphaned Redis locks (#7919, #7920) RedisWorker could leave distributed locks stranded in Redis, causing tasks to stay WAITING and resources to stay blocked until a manual release-task-locks: - #7919: a same-name pod restart reused a worker name while Redis still held locks from the dead incarnation, so the new worker was blocked by its own name's stale locks. - #7920: periodic cleanup only handled owners found via AppStatus.objects.missing(); locks whose AppStatus row was already deleted (graceful shutdown, prior cleanup, kill -9) persisted with no DB record pointing at them. Introduce a per-owner lock registry (Redis SET pulp:owner_locks:{owner}) maintained atomically inside the acquire/release Lua scripts, making cleanup O(locks-held-by-owner) instead of scanning the whole keyspace. A throttled legacy SCAN fallback (pulp:last_legacy_owner_scan) covers locks predating the registry during rolling upgrades. Add release_stale_locks_for_self() at worker startup (with successor detection and a brand-new-worker fast path) and reconcile_orphan_redis_locks() to the periodic cleanup (releasing locks only for owners with zero AppStatus row, never for stale heartbeats). Refactor cleanup_redis_locks_for_worker() to release locks without failing WAITING tasks, isolate per-task failures, and retain the AppStatus row for retry on failure. Immediate-task locks are given a grace period while their task is still incomplete. Add unit tests covering the S1-S12 scenario matrix. Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> --- CHANGES/7919.bugfix | 1 + CHANGES/7920.bugfix | 1 + pulpcore/tasking/redis_locks.py | 183 ++++++++ pulpcore/tasking/redis_worker.py | 207 ++++++-- pulpcore/tests/unit/tasking/__init__.py | 0 .../unit/tasking/test_orphan_redis_locks.py | 440 ++++++++++++++++++ 6 files changed, 783 insertions(+), 49 deletions(-) create mode 100644 CHANGES/7919.bugfix create mode 100644 CHANGES/7920.bugfix create mode 100644 pulpcore/tests/unit/tasking/__init__.py create mode 100644 pulpcore/tests/unit/tasking/test_orphan_redis_locks.py diff --git a/CHANGES/7919.bugfix b/CHANGES/7919.bugfix new file mode 100644 index 00000000000..05bf29ac801 --- /dev/null +++ b/CHANGES/7919.bugfix @@ -0,0 +1 @@ +Fixed RedisWorker being blocked by its own stale locks after a same-name pod restart by releasing them at startup. diff --git a/CHANGES/7920.bugfix b/CHANGES/7920.bugfix new file mode 100644 index 00000000000..2cfbffa4625 --- /dev/null +++ b/CHANGES/7920.bugfix @@ -0,0 +1 @@ +Fixed orphaned Redis task/resource locks left behind when a worker's AppStatus was already deleted, via periodic reconciliation. diff --git a/pulpcore/tasking/redis_locks.py b/pulpcore/tasking/redis_locks.py index e38fdbc9334..dbc17bc6e93 100644 --- a/pulpcore/tasking/redis_locks.py +++ b/pulpcore/tasking/redis_locks.py @@ -17,6 +17,19 @@ # Redis key prefix for resource locks REDIS_LOCK_PREFIX = "pulp:resource_lock:" +# Redis key prefix for the per-owner lock registry. Each owner has a SET listing the +# lock keys it currently holds so cleanup is O(locks held by owner), not O(all locks). +REDIS_OWNER_REGISTRY_PREFIX = "pulp:owner_locks:" + +# Throttle key + interval (seconds) for the legacy full-keyspace SCAN fallback used +# during rolling upgrades (locks acquired before the registry existed). +LEGACY_OWNER_SCAN_KEY = "pulp:last_legacy_owner_scan" +LEGACY_OWNER_SCAN_INTERVAL = 900 # ~15 min, fleet-wide + +# Owner name prefix used by safe_release_task_locks for immediate tasks that run in an +# API process without an AppStatus. These owners never have an AppStatus row. +IMMEDIATE_OWNER_PREFIX = "immediate-" + REDIS_ACQUIRE_LOCKS_SCRIPT = """ -- KEYS[1]: task_lock_key -- KEYS[2...]: exclusive_lock_keys, then shared_lock_keys @@ -29,6 +42,7 @@ local lock_owner = ARGV[1] local num_exclusive = tonumber(ARGV[2]) local blocked_resources = {} +local owner_registry_key = "pulp:owner_locks:" .. lock_owner -- Check task lock first (fail fast) if redis.call("exists", task_lock_key) == 1 then @@ -89,6 +103,13 @@ redis.call("sadd", key, lock_owner) end +-- Register every held lock key under the owner registry (atomic with acquisition). +-- This lets cleanup enumerate an owner's locks without scanning the whole keyspace. +redis.call("sadd", owner_registry_key, task_lock_key) +for i = 1, #KEYS - 1 do + redis.call("sadd", owner_registry_key, KEYS[1 + i]) +end + -- Return empty table to indicate success return {} """ @@ -107,6 +128,7 @@ local not_owned_exclusive = {} local not_in_shared = {} local task_lock_not_owned = false +local owner_registry_key = "pulp:owner_locks:" .. lock_owner -- Release exclusive locks -- Resource keys start at KEYS[2] @@ -118,6 +140,7 @@ local current_owner = redis.call("get", key) if current_owner == lock_owner then redis.call("del", key) + redis.call("srem", owner_registry_key, key) elseif current_owner ~= false then -- Lock exists but we don't own it table.insert(not_owned_exclusive, resource_name) @@ -133,6 +156,8 @@ -- Remove from set local removed = redis.call("srem", key, lock_owner) + -- No longer a member, so drop the registry entry for this shared key. + redis.call("srem", owner_registry_key, key) if removed == 0 then -- We weren't in the set table.insert(not_in_shared, resource_name) @@ -143,6 +168,7 @@ local task_lock_owner = redis.call("get", task_lock_key) if task_lock_owner == lock_owner then redis.call("del", task_lock_key) + redis.call("srem", owner_registry_key, task_lock_key) elseif task_lock_owner ~= false then -- Task lock exists but we don't own it task_lock_not_owned = true @@ -152,6 +178,60 @@ """ +REDIS_CLEANUP_OWNER_LOCKS_SCRIPT = """ +-- Release every lock held by an owner, using the per-owner registry set. +-- ARGV[1]: lock_owner +-- Returns: number of locks released (best effort) +local lock_owner = ARGV[1] +local owner_registry_key = "pulp:owner_locks:" .. lock_owner +local keys = redis.call("smembers", owner_registry_key) +local released = 0 + +for _, key in ipairs(keys) do + local key_type = redis.call("type", key)["ok"] + if key_type == "string" then + -- Only delete if we still own it (a successor may have re-taken the name). + if redis.call("get", key) == lock_owner then + redis.call("del", key) + released = released + 1 + end + elseif key_type == "set" then + -- srem; the set auto-deletes once its last member leaves. + released = released + redis.call("srem", key, lock_owner) + end + -- key_type == "none": stale registry entry, nothing to release. +end + +redis.call("del", owner_registry_key) +return released +""" + + +REDIS_DELETE_STRING_IF_OWNER_SCRIPT = """ +-- Atomically delete a string lock only if it is owned by lock_owner. +-- KEYS[1]: lock key +-- ARGV[1]: lock_owner +-- ARGV[2]: owner_registry_key +if redis.call("get", KEYS[1]) == ARGV[1] then + redis.call("del", KEYS[1]) + redis.call("srem", ARGV[2], KEYS[1]) + return 1 +end +return 0 +""" + + +REDIS_SREM_OWNER_SCRIPT = """ +-- Atomically remove lock_owner from a shared set (auto-deletes when empty). +-- KEYS[1]: shared set key +-- ARGV[1]: lock_owner +-- ARGV[2]: owner_registry_key +local removed = redis.call("srem", KEYS[1], ARGV[1]) +redis.call("srem", ARGV[2], KEYS[1]) +return removed +""" + + def resource_to_lock_key(resource_name): """ Convert a resource name to a Redis lock key. @@ -178,6 +258,109 @@ def get_task_lock_key(task_id): return f"task:{task_id}" +def get_owner_registry_key(owner): + """Return the Redis key for an owner's lock registry SET.""" + return f"{REDIS_OWNER_REGISTRY_PREFIX}{owner}" + + +def _decode(value): + """Decode a redis bytes value to str (redis-py returns bytes by default).""" + return value.decode() if isinstance(value, bytes) else value + + +def _legacy_scan_cleanup_for_owner(redis_conn, owner): + """ + Release an owner's locks by scanning the keyspace (no registry available). + + Used only for locks acquired before the per-owner registry existed (rolling + upgrade). Uses SCAN (never KEYS) and atomic per-key Lua so a concurrent worker + that re-took a key by the same name is not clobbered. + """ + registry_key = get_owner_registry_key(owner) + delete_if_owner = redis_conn.register_script(REDIS_DELETE_STRING_IF_OWNER_SCRIPT) + srem_owner = redis_conn.register_script(REDIS_SREM_OWNER_SCRIPT) + + for key in redis_conn.scan_iter(match="task:*", count=500): + if _decode(redis_conn.get(key)) == owner: + delete_if_owner(keys=[key], args=[owner, registry_key]) + + for key in redis_conn.scan_iter(match=f"{REDIS_LOCK_PREFIX}*", count=500): + if _decode(redis_conn.type(key)) == "string": + if _decode(redis_conn.get(key)) == owner: + delete_if_owner(keys=[key], args=[owner, registry_key]) + else: + srem_owner(keys=[key], args=[owner, registry_key]) + + redis_conn.delete(registry_key) + + +def cleanup_locks_for_owner(redis_conn, owner, allow_legacy_scan=False): + """ + Release all Redis locks held by ``owner``. + + Prefers the per-owner registry (O(locks held by owner)). Falls back to a legacy + keyspace SCAN only when the registry is missing and ``allow_legacy_scan`` is set. + + Args: + redis_conn: Redis connection + owner (str): The lock owner (worker name or ``immediate-{task_pk}``) + allow_legacy_scan (bool): Permit the legacy SCAN fallback for pre-registry locks + + Returns: + bool: True if cleanup completed (including a no-op), False on error so the + caller can retain state and retry on a later pass. + """ + registry_key = get_owner_registry_key(owner) + try: + if redis_conn.exists(registry_key): + cleanup_script = redis_conn.register_script(REDIS_CLEANUP_OWNER_LOCKS_SCRIPT) + cleanup_script(keys=[], args=[owner]) + elif allow_legacy_scan: + _legacy_scan_cleanup_for_owner(redis_conn, owner) + # else: no registry and legacy scan not permitted -> nothing to do. + return True + except Exception as e: + _logger.error("Error cleaning up Redis locks for owner %s: %s", owner, e) + return False + + +def collect_lock_owners(redis_conn, allow_legacy_scan=False): + """ + Return the set of owner names that currently hold Redis locks. + + The registry SCAN is cheap (one key per active owner). The legacy SCAN of the + full ``task:*`` / ``pulp:resource_lock:*`` keyspace is expensive and is only run + when ``allow_legacy_scan`` is set (throttled by the caller). + + Args: + redis_conn: Redis connection + allow_legacy_scan (bool): Also discover owners of pre-registry (legacy) locks + + Returns: + set: Owner names holding at least one lock. + """ + owners = set() + prefix_len = len(REDIS_OWNER_REGISTRY_PREFIX) + for key in redis_conn.scan_iter(match=f"{REDIS_OWNER_REGISTRY_PREFIX}*", count=500): + owners.add(_decode(key)[prefix_len:]) + + if allow_legacy_scan: + for key in redis_conn.scan_iter(match="task:*", count=500): + value = redis_conn.get(key) + if value: + owners.add(_decode(value)) + for key in redis_conn.scan_iter(match=f"{REDIS_LOCK_PREFIX}*", count=500): + if _decode(redis_conn.type(key)) == "string": + value = redis_conn.get(key) + if value: + owners.add(_decode(value)) + else: + for member in redis_conn.smembers(key): + owners.add(_decode(member)) + + return owners + + def extract_task_resources(task): """ Extract exclusive and shared resources from a task. diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index 59ffdca9f95..a0a8736d491 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -42,8 +42,14 @@ startup_hook, ) from pulpcore.tasking.redis_locks import ( + IMMEDIATE_OWNER_PREFIX, + LEGACY_OWNER_SCAN_INTERVAL, + LEGACY_OWNER_SCAN_KEY, acquire_locks, + cleanup_locks_for_owner, + collect_lock_owners, extract_task_resources, + get_owner_registry_key, get_task_lock_key, release_resource_locks, safe_release_task_locks, @@ -200,6 +206,11 @@ def __init__(self): except redis.RedisError: raise RuntimeError(f"Redis is not reachable. RedisWorker {self.name} cannot start.") + # A restarted pod can reuse this worker's name while Redis still holds locks + # from the dead incarnation. Release them before we start fetching tasks so we + # are not blocked by our own name's stale locks. + self.release_stale_locks_for_self() + # Add a file descriptor to trigger select on signals self.sentinel, sentinel_w = os.pipe() os.set_blocking(self.sentinel, False) @@ -297,80 +308,178 @@ def cleanup_ignored_tasks(self): def cleanup_redis_locks_for_worker(self, app_worker): """ - Clean up Redis locks held by a specific worker and fail its tasks. + Clean up Redis locks held by a specific missing worker and fail its tasks. + + For each non-final task held by the missing worker (via app_lock FK): + 1. Release the task's Redis resource locks (unless a live successor reuses + the name -- then the locks are legitimately held by the successor). + 2. WAITING tasks: release locks but do NOT fail them (crash window between + acquire_locks() and app_lock assignment). + 3. RUNNING/CANCELING tasks: reassign app_lock to us and mark FAILED. + Finally, sweep any remaining locks registered under the worker's name. - This is called when a worker is detected as missing to: - 1. Query the database for tasks held by the worker (via app_lock FK) - 2. Release the task's Redis resource locks - 3. Set app_lock to the current (cleaning) worker - 4. Mark those tasks as FAILED + Each task is handled in isolation so one failure does not abort the rest. Args: app_worker (AppStatus): The AppStatus object of the missing worker + + Returns: + bool: True if cleanup fully succeeded, False if any step failed (so the + caller can retain the AppStatus row for a later retry pass). """ worker_name = app_worker.name + success = True - try: - # Primary path: query database for the task held by the missing worker. - # A worker runs at most one task at a time, so we expect at most one. - task = ( - Task.objects.filter(app_lock=app_worker) - .exclude(state__in=TASK_FINAL_STATES) - .select_related("pulp_domain") - .first() - ) + # If a live successor already reuses this name, its Redis locks are legitimate + # -- do not release by name. The successor cleared/owns them. + successor_online = ( + AppStatus.objects.online().filter(name=worker_name).exclude(pk=app_worker.pk).exists() + ) - if task: - # Extract resources from the task's reserved_resources_record - exclusive_resources, shared_resources = extract_task_resources(task) + tasks = ( + Task.objects.filter(app_lock=app_worker) + .exclude(state__in=TASK_FINAL_STATES) + .select_related("pulp_domain") + ) + for task in tasks: + try: + if not successor_online: + exclusive_resources, shared_resources = extract_task_resources(task) + release_resource_locks( + self.redis_conn, + worker_name, + get_task_lock_key(task.pk), + exclusive_resources, + shared_resources, + ) - # Release Redis locks using the missing worker's name as the lock owner - task_lock_key = get_task_lock_key(task.pk) - release_resource_locks( - self.redis_conn, - worker_name, - task_lock_key, - exclusive_resources, - shared_resources, - ) - _logger.info( - "Released task lock + %d exclusive + %d shared resource locks " - "for task %s from missing worker %s", - len(exclusive_resources), - len(shared_resources), - task.pk, - worker_name, - ) + if task.state == TASK_STATES.WAITING: + # Crash window: locks were acquired but the task never ran. + # Release the locks (done above) but leave it WAITING to be + # re-fetched; just detach the stale app_lock. + Task.objects.filter(pk=task.pk).update(app_lock=None) + continue - # Set app_lock to the current (cleaning) worker so set_failed() - # ownership check passes + # Running/canceling task -> reassign app_lock to us and fail it. Task.objects.filter(pk=task.pk).update(app_lock=self.app_status) task.app_lock = self.app_status - - # Set to canceling first task.set_canceling() - error_msg = "Worker has gone missing." - task.set_canceled(final_state=TASK_STATES.FAILED, reason=error_msg) + task.set_canceled(final_state=TASK_STATES.FAILED, reason="Worker has gone missing.") _logger.warning( "Marked task %s as FAILED (was being executed by missing worker %s)", task.pk, worker_name, ) - except Exception as e: - _logger.error("Error cleaning up locks for worker %s: %s", worker_name, e) + except Exception as e: + _logger.error( + "Error cleaning up task %s of missing worker %s: %s", + task.pk, + worker_name, + e, + ) + success = False + + # Registry-driven sweep of any other locks still held under this name. + if not successor_online: + if not cleanup_locks_for_owner(self.redis_conn, worker_name, allow_legacy_scan=False): + success = False + + return success + + def release_stale_locks_for_self(self): + """ + Release Redis locks lingering under our own worker name at startup. + + A restarted pod can reuse the same name while Redis still holds locks from + the dead process. Skips if a live successor already owns the name, and skips + entirely for a brand-new worker (no registry, no prior AppStatus row). + + Returns: + bool: True if handled (including no-op), False on cleanup error. + """ + # Another live worker already owns this name -> it manages its own locks. + if ( + AppStatus.objects.online() + .filter(name=self.name) + .exclude(pk=self.app_status.pk) + .exists() + ): + return True + + registry_exists = bool(self.redis_conn.exists(get_owner_registry_key(self.name))) + prior_status = ( + AppStatus.objects.filter(name=self.name).exclude(pk=self.app_status.pk).exists() + ) + + # Brand-new worker -> nothing could exist under our name; skip any SCAN. + if not registry_exists and not prior_status: + return True + + # Rolling upgrade: a prior incarnation left locks but no registry -> allow the + # legacy SCAN for our own name only. + allow_legacy = prior_status and not registry_exists + return cleanup_locks_for_owner(self.redis_conn, self.name, allow_legacy_scan=allow_legacy) + + def _immediate_owner_is_finished(self, owner): + """Return True if an ``immediate-{pk}`` owner's task is gone or finished.""" + task_pk = owner[len(IMMEDIATE_OWNER_PREFIX) :] + return not Task.objects.filter(pk=task_pk, state__in=TASK_INCOMPLETE_STATES).exists() + + def reconcile_orphan_redis_locks(self): + """ + Release Redis locks whose owner has no AppStatus row at all. + + The missing-worker path only handles owners with a (stale) AppStatus row. + This reconciles owners whose AppStatus was already deleted (graceful + shutdown, prior cleanup, kill -9), leaving locks with no DB record. + + Only owners with ZERO AppStatus rows are cleaned; a stale heartbeat still + counts as "has a row" and is left to the missing-worker path. Each owner is + handled in isolation. The expensive legacy keyspace SCAN is throttled + fleet-wide via a Redis key. + """ + # Winner of the throttle key runs the expensive legacy scan this pass. + allow_legacy = bool( + self.redis_conn.set( + LEGACY_OWNER_SCAN_KEY, self.name, nx=True, ex=LEGACY_OWNER_SCAN_INTERVAL + ) + ) + + owners = collect_lock_owners(self.redis_conn, allow_legacy_scan=allow_legacy) + if not owners: + return + + # An owner is orphaned only if it has NO AppStatus row (stale != orphaned). + existing = set(AppStatus.objects.filter(name__in=owners).values_list("name", flat=True)) + for owner in owners - existing: + try: + if owner.startswith( + IMMEDIATE_OWNER_PREFIX + ) and not self._immediate_owner_is_finished(owner): + # Immediate task still running in another process; keep its locks. + continue + cleanup_locks_for_owner(self.redis_conn, owner, allow_legacy_scan=allow_legacy) + except Exception: + _logger.exception("Failed reconciling orphan lock owner %s", owner) @exclusive(WORKER_CLEANUP_LOCK) def app_worker_cleanup(self): """Cleanup records of missing app processes and their Redis locks.""" - qs = AppStatus.objects.missing() - for app_worker in qs: + for app_worker in list(AppStatus.objects.missing()): _logger.warning( "Cleanup record of missing %s process %s.", app_worker.app_type, app_worker.name ) - # Clean up any Redis locks held by this missing process - # This includes workers and API processes (which can hold locks for immediate tasks) - self.cleanup_redis_locks_for_worker(app_worker) - qs.delete() + # Clean up any Redis locks held by this missing process. This includes + # workers and API processes (which can hold locks for immediate tasks). + # Only delete the record if cleanup fully succeeded, otherwise retain it + # for a later retry pass. + if self.cleanup_redis_locks_for_worker(app_worker): + app_worker.delete() + else: + _logger.warning("Retaining AppStatus %s for a later cleanup pass.", app_worker.name) + + # Reconcile locks whose owner has no AppStatus row at all. + self.reconcile_orphan_redis_locks() @exclusive(TASK_SCHEDULING_LOCK) def dispatch_scheduled_tasks(self): diff --git a/pulpcore/tests/unit/tasking/__init__.py b/pulpcore/tests/unit/tasking/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/pulpcore/tests/unit/tasking/test_orphan_redis_locks.py b/pulpcore/tests/unit/tasking/test_orphan_redis_locks.py new file mode 100644 index 00000000000..58b7bf72eff --- /dev/null +++ b/pulpcore/tests/unit/tasking/test_orphan_redis_locks.py @@ -0,0 +1,440 @@ +""" +Unit tests for orphan Redis lock cleanup (same-name restart and orphan owners). + +Tests use a real Redis provided by the ``redisdb`` pytest fixture (mirroring +``pulp_redisdb`` in test_cache.py) and real DB rows via ``@pytest.mark.django_db``. +""" + +from datetime import timedelta + +import pytest +import redis +from django.utils import timezone + +import pulpcore.app.redis_connection +from pulpcore.app.models import AppStatus, Task +from pulpcore.constants import TASK_STATES +from pulpcore.tasking import redis_locks, redis_worker +from pulpcore.tasking.redis_locks import ( + acquire_locks, + cleanup_locks_for_owner, + collect_lock_owners, + get_owner_registry_key, + get_task_lock_key, + resource_to_lock_key, +) +from pulpcore.tasking.redis_worker import RedisWorker + + +# --------------------------------------------------------------------------- # +# Fixtures / helpers +# --------------------------------------------------------------------------- # +@pytest.fixture +def pulp_redisdb(settings, redisdb, monkeypatch): + """Point pulpcore's redis connection at the ephemeral ``redisdb`` instance.""" + monkeypatch.setattr(pulpcore.app.redis_connection, "_conn", None) + monkeypatch.setattr(pulpcore.app.redis_connection, "_a_conn", None) + settings.CACHE_ENABLED = True + settings.REDIS_URL = "unix://" + redisdb.get_connection_kwargs()["path"] + return pulpcore.app.redis_connection.get_redis_connection() + + +@pytest.fixture +def reset_singleton(monkeypatch): + """Reset the in-process AppStatus singleton so ``create()`` works per-test.""" + monkeypatch.setattr(AppStatus.objects, "_current_app_status", None) + + +def _smembers(conn, key): + """SMEMBERS as a set of str (redisdb returns bytes).""" + return {m.decode() if isinstance(m, bytes) else m for m in conn.smembers(key)} + + +def make_app_status(name, online=True, app_type="worker"): + """Create an AppStatus row directly (bypasses the singleton-enforcing manager).""" + st = AppStatus(app_type=app_type, name=name, ttl=timedelta(seconds=30)) + st.save() + if not online: + AppStatus.objects.filter(pk=st.pk).update( + last_heartbeat=timezone.now() - timedelta(seconds=3600) + ) + st.refresh_from_db() + return st + + +def make_worker(conn, name, app_status): + """Build a bare RedisWorker with only the attributes the cleanup methods use.""" + w = RedisWorker.__new__(RedisWorker) + w.redis_conn = conn + w.name = name + w.app_status = app_status + return w + + +def seed_locks_via_acquire(conn, owner, task_id, exclusive=None, shared=None): + """Acquire locks the normal way so the owner registry is populated.""" + exclusive = exclusive or [] + shared = shared or [] + task_lock_key = get_task_lock_key(task_id) + blocked = acquire_locks(conn, owner, task_lock_key, exclusive, shared) + assert blocked == [] + return task_lock_key + + +# --------------------------------------------------------------------------- # +# Registry maintenance in the Lua scripts +# --------------------------------------------------------------------------- # +def test_acquire_registers_owner_locks(pulp_redisdb): + """Given a worker acquires locks, When acquisition succeeds, Then the owner + registry set lists every held lock key.""" + conn = pulp_redisdb + task_lock_key = seed_locks_via_acquire( + conn, "owner-a", "t1", exclusive=["res-excl"], shared=["res-shared"] + ) + + registry = _smembers(conn, get_owner_registry_key("owner-a")) + assert task_lock_key in registry + assert resource_to_lock_key("res-excl") in registry + assert resource_to_lock_key("res-shared") in registry + + +def test_release_unregisters_owner_locks(pulp_redisdb): + """Given locks acquired, When they are released, Then the owner registry set is + emptied/auto-deleted.""" + conn = pulp_redisdb + task_lock_key = seed_locks_via_acquire( + conn, "owner-a", "t1", exclusive=["res-excl"], shared=["res-shared"] + ) + redis_locks.release_resource_locks(conn, "owner-a", task_lock_key, ["res-excl"], ["res-shared"]) + assert conn.exists(get_owner_registry_key("owner-a")) == 0 + + +# --------------------------------------------------------------------------- # +# cleanup_locks_for_owner / collect_lock_owners +# --------------------------------------------------------------------------- # +def test_cleanup_locks_for_owner_registry_path_only_touches_owner(pulp_redisdb): + """Given two owners each with registered locks, When one is cleaned, + Then only that owner's keys are removed and the other is untouched.""" + conn = pulp_redisdb + seed_locks_via_acquire(conn, "owner-a", "ta", exclusive=["a-excl"], shared=["a-shared"]) + seed_locks_via_acquire(conn, "owner-b", "tb", exclusive=["b-excl"], shared=["b-shared"]) + + assert cleanup_locks_for_owner(conn, "owner-a") is True + + assert conn.exists(get_task_lock_key("ta")) == 0 + assert conn.exists(resource_to_lock_key("a-excl")) == 0 + assert conn.exists(resource_to_lock_key("a-shared")) == 0 + assert conn.exists(get_owner_registry_key("owner-a")) == 0 + # owner-b intact + assert conn.exists(get_task_lock_key("tb")) == 1 + assert conn.exists(resource_to_lock_key("b-excl")) == 1 + assert conn.exists(get_owner_registry_key("owner-b")) == 1 + + +def test_cleanup_locks_for_owner_legacy_scan(pulp_redisdb): + """Given legacy locks with no registry, When cleaned with the legacy scan + allowed, Then they are removed (and left alone when the scan is not allowed).""" + conn = pulp_redisdb + conn.set(get_task_lock_key("leg"), "legacy-1") + conn.set(resource_to_lock_key("leg-excl"), "legacy-1") + conn.sadd(resource_to_lock_key("leg-shared"), "legacy-1") + + # Without legacy scan permission and no registry: no-op. + assert cleanup_locks_for_owner(conn, "legacy-1", allow_legacy_scan=False) is True + assert conn.exists(get_task_lock_key("leg")) == 1 + + # With legacy scan permission: cleaned. + assert cleanup_locks_for_owner(conn, "legacy-1", allow_legacy_scan=True) is True + assert conn.exists(get_task_lock_key("leg")) == 0 + assert conn.exists(resource_to_lock_key("leg-excl")) == 0 + assert conn.exists(resource_to_lock_key("leg-shared")) == 0 + + +def test_cleanup_locks_for_owner_returns_false_on_error(): + """Given Redis raises during cleanup, When cleanup is attempted, Then it + returns False so the caller can retry later.""" + + class BoomConn: + def exists(self, *args, **kwargs): + raise redis.RedisError("boom") + + assert cleanup_locks_for_owner(BoomConn(), "owner-x") is False + + +def test_collect_lock_owners_registry_and_legacy(pulp_redisdb): + """Registry owners are always collected; legacy owners only when the legacy scan + is allowed.""" + conn = pulp_redisdb + seed_locks_via_acquire(conn, "reg-1", "tr", exclusive=["r-excl"]) + conn.set(get_task_lock_key("legt"), "leg-1") + + assert collect_lock_owners(conn, allow_legacy_scan=False) == {"reg-1"} + assert collect_lock_owners(conn, allow_legacy_scan=True) == {"reg-1", "leg-1"} + + +# --------------------------------------------------------------------------- # +# release_stale_locks_for_self +# --------------------------------------------------------------------------- # +@pytest.mark.django_db +def test_release_stale_locks_for_self_releases_prior_locks(pulp_redisdb, reset_singleton): + """Given prior locks under our worker name, When a new incarnation runs + startup self-cleanup, Then those locks are released.""" + conn = pulp_redisdb + seed_locks_via_acquire(conn, "1@pod-abc", "old", exclusive=["res"]) + + app_status = AppStatus.objects.create(app_type="worker", name="1@pod-abc") + w = make_worker(conn, "1@pod-abc", app_status) + + assert w.release_stale_locks_for_self() is True + assert conn.exists(get_task_lock_key("old")) == 0 + assert conn.exists(resource_to_lock_key("res")) == 0 + + +@pytest.mark.django_db +def test_release_stale_locks_for_self_brand_new_skips_scan( + pulp_redisdb, reset_singleton, monkeypatch +): + """Given a brand-new worker (no registry, no prior AppStatus), When startup + self-cleanup runs, Then it does no keyspace SCAN.""" + conn = pulp_redisdb + app_status = AppStatus.objects.create(app_type="worker", name="brand-new") + w = make_worker(conn, "brand-new", app_status) + + called = [] + monkeypatch.setattr(conn, "scan_iter", lambda *a, **k: called.append((a, k)) or iter(())) + + assert w.release_stale_locks_for_self() is True + assert called == [] + + +@pytest.mark.django_db +def test_release_stale_locks_for_self_skips_when_successor_online(pulp_redisdb, reset_singleton): + """Given another live AppStatus already owns our name, When startup + self-cleanup runs, Then we skip and leave its locks alone.""" + conn = pulp_redisdb + seed_locks_via_acquire(conn, "1@pod-abc", "held", exclusive=["res"]) + + app_status = AppStatus.objects.create(app_type="worker", name="1@pod-abc") + make_app_status("1@pod-abc", online=True) # live successor with same name + w = make_worker(conn, "1@pod-abc", app_status) + + assert w.release_stale_locks_for_self() is True + # Locks untouched -- the live peer manages them. + assert conn.exists(get_task_lock_key("held")) == 1 + assert conn.exists(resource_to_lock_key("res")) == 1 + + +# --------------------------------------------------------------------------- # +# reconcile_orphan_redis_locks +# --------------------------------------------------------------------------- # +@pytest.mark.django_db +def test_reconcile_releases_orphan_owner(pulp_redisdb, reset_singleton): + """Given locks in Redis with no AppStatus row for the owner, When reconcile + runs, Then the orphan locks are released.""" + conn = pulp_redisdb + seed_locks_via_acquire(conn, "orphan-1", "torphan", exclusive=["ores"]) + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + w.reconcile_orphan_redis_locks() + assert conn.exists(get_task_lock_key("torphan")) == 0 + assert conn.exists(resource_to_lock_key("ores")) == 0 + + +@pytest.mark.django_db +def test_reconcile_skips_owner_with_stale_appstatus(pulp_redisdb, reset_singleton): + """Given an owner with a stale heartbeat but an existing AppStatus row, When + reconcile runs, Then its locks are NOT released.""" + conn = pulp_redisdb + seed_locks_via_acquire(conn, "stale-1", "tstale", exclusive=["sres"]) + make_app_status("stale-1", online=False) # exists but missing heartbeat + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + w.reconcile_orphan_redis_locks() + assert conn.exists(get_task_lock_key("tstale")) == 1 + assert conn.exists(resource_to_lock_key("sres")) == 1 + + +@pytest.mark.django_db +def test_reconcile_isolates_per_owner_errors(pulp_redisdb, reset_singleton, monkeypatch): + """Given cleanup of one orphan owner raises, When reconcile runs, Then the + remaining orphan owners are still processed.""" + conn = pulp_redisdb + seed_locks_via_acquire(conn, "bad", "tbad", exclusive=["bres"]) + seed_locks_via_acquire(conn, "good", "tgood", exclusive=["gres"]) + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + processed = [] + + def fake_cleanup(rconn, owner, allow_legacy_scan=False): + processed.append(owner) + if owner == "bad": + raise redis.RedisError("boom") + return True + + monkeypatch.setattr(redis_worker, "cleanup_locks_for_owner", fake_cleanup) + + w.reconcile_orphan_redis_locks() + assert "good" in processed # not aborted by "bad" failing + + +@pytest.mark.django_db +def test_reconcile_immediate_owner_grace(pulp_redisdb, reset_singleton): + """Immediate grace: Given an ``immediate-{pk}`` owner whose task is still + incomplete, When reconcile runs, Then its locks are kept; once the task is + finished they are reclaimed.""" + conn = pulp_redisdb + task = Task.objects.create(name="imm", state=TASK_STATES.RUNNING) + owner = f"immediate-{task.pk}" + seed_locks_via_acquire(conn, owner, str(task.pk), exclusive=["ires"]) + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + # Task still RUNNING -> locks kept. + w.reconcile_orphan_redis_locks() + assert conn.exists(resource_to_lock_key("ires")) == 1 + + # Task finished -> locks reclaimed. + Task.objects.filter(pk=task.pk).update(state=TASK_STATES.FAILED) + w.reconcile_orphan_redis_locks() + assert conn.exists(resource_to_lock_key("ires")) == 0 + + +# --------------------------------------------------------------------------- # +# cleanup_redis_locks_for_worker +# --------------------------------------------------------------------------- # +@pytest.mark.django_db +def test_cleanup_worker_waiting_task_not_failed(pulp_redisdb, reset_singleton): + """Given a WAITING task held by a missing worker, When cleanup runs, Then + its locks are released but the task stays WAITING.""" + conn = pulp_redisdb + missing = make_app_status("missing-w", online=False) + task = Task.objects.create( + name="w", state=TASK_STATES.WAITING, app_lock=missing, reserved_resources_record=["wres"] + ) + seed_locks_via_acquire(conn, "missing-w", str(task.pk), exclusive=["wres"]) + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + assert w.cleanup_redis_locks_for_worker(missing) is True + task.refresh_from_db() + assert task.state == TASK_STATES.WAITING + assert task.app_lock_id is None + assert conn.exists(resource_to_lock_key("wres")) == 0 + + +@pytest.mark.django_db +def test_cleanup_worker_running_task_failed(pulp_redisdb, reset_singleton): + """Given a RUNNING task held by a missing worker, When cleanup runs, Then + its locks are released and the task is FAILED.""" + conn = pulp_redisdb + missing = make_app_status("missing-w", online=False) + task = Task.objects.create( + name="r", state=TASK_STATES.RUNNING, app_lock=missing, reserved_resources_record=["rres"] + ) + seed_locks_via_acquire(conn, "missing-w", str(task.pk), exclusive=["rres"]) + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + assert w.cleanup_redis_locks_for_worker(missing) is True + task.refresh_from_db() + assert task.state == TASK_STATES.FAILED + assert conn.exists(resource_to_lock_key("rres")) == 0 + + +@pytest.mark.django_db +def test_cleanup_worker_returns_false_on_error(pulp_redisdb, reset_singleton, monkeypatch): + """Given releasing a task's locks raises, When cleanup runs, Then it returns + False so the AppStatus can be retained for retry.""" + conn = pulp_redisdb + missing = make_app_status("missing-w", online=False) + Task.objects.create( + name="r", state=TASK_STATES.RUNNING, app_lock=missing, reserved_resources_record=["rres"] + ) + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + def boom(*args, **kwargs): + raise redis.RedisError("boom") + + monkeypatch.setattr(redis_worker, "release_resource_locks", boom) + + assert w.cleanup_redis_locks_for_worker(missing) is False + + +@pytest.mark.django_db +def test_cleanup_worker_isolates_per_task_errors(pulp_redisdb, reset_singleton, monkeypatch): + """Given a missing worker with two tasks where releasing the first task's locks + raises, When cleanup runs, Then the second task is still processed (FAILED, locks + released) and cleanup returns False. + + The per-task try/except must isolate one Redis failure from the rest. + """ + conn = pulp_redisdb + missing = make_app_status("missing-w", online=False) + bad_task = Task.objects.create( + name="bad", + state=TASK_STATES.RUNNING, + app_lock=missing, + reserved_resources_record=["bad-res"], + ) + good_task = Task.objects.create( + name="good", + state=TASK_STATES.RUNNING, + app_lock=missing, + reserved_resources_record=["good-res"], + ) + seed_locks_via_acquire(conn, "missing-w", str(bad_task.pk), exclusive=["bad-res"]) + seed_locks_via_acquire(conn, "missing-w", str(good_task.pk), exclusive=["good-res"]) + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + real_release = redis_locks.release_resource_locks + bad_lock_key = get_task_lock_key(bad_task.pk) + + def flaky_release(redis_conn, lock_owner, task_lock_key, *args, **kwargs): + if task_lock_key == bad_lock_key: + raise redis.RedisError("boom") + return real_release(redis_conn, lock_owner, task_lock_key, *args, **kwargs) + + monkeypatch.setattr(redis_worker, "release_resource_locks", flaky_release) + + # The bad task raises, but the good task must still be handled -> overall False. + assert w.cleanup_redis_locks_for_worker(missing) is False + + good_task.refresh_from_db() + assert good_task.state == TASK_STATES.FAILED + assert conn.exists(resource_to_lock_key("good-res")) == 0 + # The bad task never reached its DB update (its release raised first). + bad_task.refresh_from_db() + assert bad_task.state == TASK_STATES.RUNNING + + +@pytest.mark.django_db +def test_cleanup_worker_skips_release_when_successor_online(pulp_redisdb, reset_singleton): + """Given a missing worker whose name is reused by a live successor, When + cleanup runs, Then the successor's Redis locks are not released.""" + conn = pulp_redisdb + missing = make_app_status("reused-name", online=False) + make_app_status("reused-name", online=True) # live successor same name + task = Task.objects.create( + name="r", state=TASK_STATES.RUNNING, app_lock=missing, reserved_resources_record=["kres"] + ) + seed_locks_via_acquire(conn, "reused-name", str(task.pk), exclusive=["kres"]) + + cleaner = AppStatus.objects.create(app_type="worker", name="cleaner") + w = make_worker(conn, "cleaner", cleaner) + + w.cleanup_redis_locks_for_worker(missing) + # Successor's locks left intact. + assert conn.exists(resource_to_lock_key("kres")) == 1 From cee82a905e9ba949d88d38227b52e9c2e26de5c5 Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Wed, 12 Aug 2026 21:52:49 +0200 Subject: [PATCH 2/3] fix: clean up stale and orphaned Redis locks (#7919, #7920) When a Kubernetes pod restarts, the new worker reuses the dead predecessor's name while Redis still holds its locks. These orphan locks blocked resources indefinitely: the periodic cleanup either saw the old AppStatus as still online, or the AppStatus was already deleted without its locks being released. Cleanup now scales with the work to be done, not with the size of the Redis keyspace. Previously each cleanup SCANned every key, which during an incident of many crash-looping workers compounded Redis pressure. - Per-owner lock registry ( SET), maintained atomically inside the acquire/release Lua scripts, so an owner's locks are cleaned in O(locks held) via SMEMBERS+Lua, not a keyspace SCAN. - Global active-owners SET () so orphan-owner enumeration is O(#owners) via SMEMBERS; no steady-state path SCANs the keyspace (a SCAN with MATCH still walks every key server-side). under this worker's name; brand-new workers skip it, and the legacy keyspace SCAN (for pre-registry locks during a rolling upgrade) is throttled fleet-wide. - reconcile_orphan_redis_locks() in the periodic cleanup releases locks whose owner has no AppStatus row at all. - cleanup_redis_locks_for_worker() sweeps remaining locks via the registry with per-task exception isolation, preserves WAITING tasks, and skips release when a live successor reuses the name. - Atomic delete-if-owner Lua for the legacy fallback; reclaimed-lock counts and startup-cleanup failures are logged for diagnosis. Adds unit tests for every cleanup path, including deterministic keys-touched proofs that each path's Redis cost is independent of the keyspace size, plus a control test that exercises the legacy SCAN. Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> --- CHANGES/7919.bugfix | 2 +- pulpcore/tasking/redis_locks.py | 64 ++++-- pulpcore/tasking/redis_worker.py | 20 +- .../unit/tasking/test_orphan_redis_locks.py | 189 +++++++++++++++++- 4 files changed, 244 insertions(+), 31 deletions(-) diff --git a/CHANGES/7919.bugfix b/CHANGES/7919.bugfix index 05bf29ac801..008380a8b51 100644 --- a/CHANGES/7919.bugfix +++ b/CHANGES/7919.bugfix @@ -1 +1 @@ -Fixed RedisWorker being blocked by its own stale locks after a same-name pod restart by releasing them at startup. +Fixed RedisWorker being blocked by its own stale locks after a worker restarts under the same name, by releasing them at startup. diff --git a/pulpcore/tasking/redis_locks.py b/pulpcore/tasking/redis_locks.py index dbc17bc6e93..a44db73789b 100644 --- a/pulpcore/tasking/redis_locks.py +++ b/pulpcore/tasking/redis_locks.py @@ -21,6 +21,12 @@ # lock keys it currently holds so cleanup is O(locks held by owner), not O(all locks). REDIS_OWNER_REGISTRY_PREFIX = "pulp:owner_locks:" +# Redis SET of owner names that currently hold at least one lock. Enumerating owners +# via SMEMBERS on this key is O(#owners); it avoids a full-keyspace SCAN (a SCAN with +# MATCH still walks every key). Kept in sync inside the acquire/release/cleanup Lua +# scripts, which hardcode this literal -- keep them matching this constant. +ACTIVE_OWNERS_KEY = "pulp:active_owners" + # Throttle key + interval (seconds) for the legacy full-keyspace SCAN fallback used # during rolling upgrades (locks acquired before the registry existed). LEGACY_OWNER_SCAN_KEY = "pulp:last_legacy_owner_scan" @@ -105,10 +111,12 @@ -- Register every held lock key under the owner registry (atomic with acquisition). -- This lets cleanup enumerate an owner's locks without scanning the whole keyspace. -redis.call("sadd", owner_registry_key, task_lock_key) -for i = 1, #KEYS - 1 do - redis.call("sadd", owner_registry_key, KEYS[1 + i]) -end +-- One variadic SADD: all of KEYS are held lock keys (task lock + resource locks). +redis.call("sadd", owner_registry_key, unpack(KEYS)) + +-- Track this owner in the global active-owners set so reconcile can enumerate +-- lock owners with SMEMBERS instead of a full-keyspace SCAN. +redis.call("sadd", "pulp:active_owners", lock_owner) -- Return empty table to indicate success return {} @@ -150,6 +158,10 @@ -- Release shared locks -- Shared keys start at KEYS[2 + num_exclusive] +-- INVARIANT: an owner runs one task at a time (see RedisWorker.handle_tasks), so it +-- never holds the same shared resource for two concurrent tasks. That lets us drop +-- the registry entry on release unconditionally. If workers ever become concurrent, +-- this must become reference-counted or the shared lock could be released early. for i = num_exclusive + 1, #KEYS - 1 do local key = KEYS[1 + i] local resource_name = ARGV[2 + i] @@ -174,6 +186,12 @@ task_lock_not_owned = true end +-- If this owner no longer holds any locks, drop it from the active-owners set +-- (the registry key auto-deletes once empty, so scard == 0 means "no locks left"). +if redis.call("scard", owner_registry_key) == 0 then + redis.call("srem", "pulp:active_owners", lock_owner) +end + return {not_owned_exclusive, not_in_shared, task_lock_not_owned} """ @@ -203,6 +221,7 @@ end redis.call("del", owner_registry_key) +redis.call("srem", "pulp:active_owners", lock_owner) return released """ @@ -275,23 +294,29 @@ def _legacy_scan_cleanup_for_owner(redis_conn, owner): Used only for locks acquired before the per-owner registry existed (rolling upgrade). Uses SCAN (never KEYS) and atomic per-key Lua so a concurrent worker that re-took a key by the same name is not clobbered. + + Returns: + int: Number of locks released (best effort). """ registry_key = get_owner_registry_key(owner) delete_if_owner = redis_conn.register_script(REDIS_DELETE_STRING_IF_OWNER_SCRIPT) srem_owner = redis_conn.register_script(REDIS_SREM_OWNER_SCRIPT) + released = 0 for key in redis_conn.scan_iter(match="task:*", count=500): if _decode(redis_conn.get(key)) == owner: - delete_if_owner(keys=[key], args=[owner, registry_key]) + released += delete_if_owner(keys=[key], args=[owner, registry_key]) for key in redis_conn.scan_iter(match=f"{REDIS_LOCK_PREFIX}*", count=500): if _decode(redis_conn.type(key)) == "string": if _decode(redis_conn.get(key)) == owner: - delete_if_owner(keys=[key], args=[owner, registry_key]) + released += delete_if_owner(keys=[key], args=[owner, registry_key]) else: - srem_owner(keys=[key], args=[owner, registry_key]) + released += srem_owner(keys=[key], args=[owner, registry_key]) redis_conn.delete(registry_key) + redis_conn.srem(ACTIVE_OWNERS_KEY, owner) + return released def cleanup_locks_for_owner(redis_conn, owner, allow_legacy_scan=False): @@ -312,12 +337,18 @@ def cleanup_locks_for_owner(redis_conn, owner, allow_legacy_scan=False): """ registry_key = get_owner_registry_key(owner) try: + released = 0 if redis_conn.exists(registry_key): cleanup_script = redis_conn.register_script(REDIS_CLEANUP_OWNER_LOCKS_SCRIPT) - cleanup_script(keys=[], args=[owner]) + released = cleanup_script(keys=[], args=[owner]) elif allow_legacy_scan: - _legacy_scan_cleanup_for_owner(redis_conn, owner) - # else: no registry and legacy scan not permitted -> nothing to do. + released = _legacy_scan_cleanup_for_owner(redis_conn, owner) + else: + # No registry and no scan: nothing to release, but drop any stale + # active-owners marker so reconcile stops re-visiting this owner. + redis_conn.srem(ACTIVE_OWNERS_KEY, owner) + if released: + _logger.info("Reclaimed %d Redis lock(s) held by owner %s", released, owner) return True except Exception as e: _logger.error("Error cleaning up Redis locks for owner %s: %s", owner, e) @@ -328,9 +359,11 @@ def collect_lock_owners(redis_conn, allow_legacy_scan=False): """ Return the set of owner names that currently hold Redis locks. - The registry SCAN is cheap (one key per active owner). The legacy SCAN of the - full ``task:*`` / ``pulp:resource_lock:*`` keyspace is expensive and is only run - when ``allow_legacy_scan`` is set (throttled by the caller). + Owners are read from the global active-owners SET via SMEMBERS -- O(#owners) and + scan-free (a SCAN with MATCH still walks the whole keyspace, so scanning for + registry keys would be O(all locks)). The legacy SCAN of the full ``task:*`` / + ``pulp:resource_lock:*`` keyspace is expensive and is only run when + ``allow_legacy_scan`` is set (throttled by the caller) to catch pre-registry locks. Args: redis_conn: Redis connection @@ -339,10 +372,7 @@ def collect_lock_owners(redis_conn, allow_legacy_scan=False): Returns: set: Owner names holding at least one lock. """ - owners = set() - prefix_len = len(REDIS_OWNER_REGISTRY_PREFIX) - for key in redis_conn.scan_iter(match=f"{REDIS_OWNER_REGISTRY_PREFIX}*", count=500): - owners.add(_decode(key)[prefix_len:]) + owners = {_decode(member) for member in redis_conn.smembers(ACTIVE_OWNERS_KEY)} if allow_legacy_scan: for key in redis_conn.scan_iter(match="task:*", count=500): diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index a0a8736d491..99ce297413f 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -206,10 +206,15 @@ def __init__(self): except redis.RedisError: raise RuntimeError(f"Redis is not reachable. RedisWorker {self.name} cannot start.") - # A restarted pod can reuse this worker's name while Redis still holds locks - # from the dead incarnation. Release them before we start fetching tasks so we - # are not blocked by our own name's stale locks. - self.release_stale_locks_for_self() + # A restarted worker can reuse its name while Redis still holds the dead + # incarnation's locks. Release them before fetching tasks so we are not + # blocked by our own stale locks. + if not self.release_stale_locks_for_self(): + _logger.warning( + "Startup lock cleanup for %s failed; the worker may be blocked by its " + "own stale locks until the periodic reconcile runs.", + self.name, + ) # Add a file descriptor to trigger select on signals self.sentinel, sentinel_w = os.pipe() @@ -390,7 +395,7 @@ def release_stale_locks_for_self(self): """ Release Redis locks lingering under our own worker name at startup. - A restarted pod can reuse the same name while Redis still holds locks from + A restarted worker can reuse the same name while Redis still holds locks from the dead process. Skips if a live successor already owns the name, and skips entirely for a brand-new worker (no registry, no prior AppStatus row). @@ -451,7 +456,10 @@ def reconcile_orphan_redis_locks(self): # An owner is orphaned only if it has NO AppStatus row (stale != orphaned). existing = set(AppStatus.objects.filter(name__in=owners).values_list("name", flat=True)) - for owner in owners - existing: + orphans = owners - existing + if orphans: + _logger.info("Reconciling Redis locks for %d orphan owner(s).", len(orphans)) + for owner in orphans: try: if owner.startswith( IMMEDIATE_OWNER_PREFIX diff --git a/pulpcore/tests/unit/tasking/test_orphan_redis_locks.py b/pulpcore/tests/unit/tasking/test_orphan_redis_locks.py index 58b7bf72eff..fdb6ce3d430 100644 --- a/pulpcore/tests/unit/tasking/test_orphan_redis_locks.py +++ b/pulpcore/tests/unit/tasking/test_orphan_redis_locks.py @@ -16,6 +16,8 @@ from pulpcore.constants import TASK_STATES from pulpcore.tasking import redis_locks, redis_worker from pulpcore.tasking.redis_locks import ( + LEGACY_OWNER_SCAN_INTERVAL, + LEGACY_OWNER_SCAN_KEY, acquire_locks, cleanup_locks_for_owner, collect_lock_owners, @@ -109,6 +111,34 @@ def test_release_unregisters_owner_locks(pulp_redisdb): assert conn.exists(get_owner_registry_key("owner-a")) == 0 +def test_shared_lock_release_preserves_other_owners(pulp_redisdb): + """Given two owners share a resource, When one releases, Then the other keeps the + shared lock and only the releasing owner's registry entry is cleared. + + Guards the single-task-per-owner invariant that REDIS_RELEASE_LOCKS_SCRIPT relies + on: a release removes only the releasing owner from the shared set/registry. + """ + conn = pulp_redisdb + shared_key = resource_to_lock_key("res-shared") + + lock_a = seed_locks_via_acquire(conn, "owner-a", "t1", shared=["res-shared"]) + lock_b = seed_locks_via_acquire(conn, "owner-b", "t2", shared=["res-shared"]) + + # Both owners are members of the shared set. + assert _smembers(conn, shared_key) == {"owner-a", "owner-b"} + + # owner-a releases its task; only owner-a leaves the shared set. + redis_locks.release_resource_locks(conn, "owner-a", lock_a, [], ["res-shared"]) + assert _smembers(conn, shared_key) == {"owner-b"} + assert conn.exists(get_owner_registry_key("owner-a")) == 0 + assert shared_key in _smembers(conn, get_owner_registry_key("owner-b")) + + # owner-b releases; the shared set auto-deletes once its last member leaves. + redis_locks.release_resource_locks(conn, "owner-b", lock_b, [], ["res-shared"]) + assert conn.exists(shared_key) == 0 + assert conn.exists(get_owner_registry_key("owner-b")) == 0 + + # --------------------------------------------------------------------------- # # cleanup_locks_for_owner / collect_lock_owners # --------------------------------------------------------------------------- # @@ -180,10 +210,10 @@ def test_release_stale_locks_for_self_releases_prior_locks(pulp_redisdb, reset_s """Given prior locks under our worker name, When a new incarnation runs startup self-cleanup, Then those locks are released.""" conn = pulp_redisdb - seed_locks_via_acquire(conn, "1@pod-abc", "old", exclusive=["res"]) + seed_locks_via_acquire(conn, "1@host-abc", "old", exclusive=["res"]) - app_status = AppStatus.objects.create(app_type="worker", name="1@pod-abc") - w = make_worker(conn, "1@pod-abc", app_status) + app_status = AppStatus.objects.create(app_type="worker", name="1@host-abc") + w = make_worker(conn, "1@host-abc", app_status) assert w.release_stale_locks_for_self() is True assert conn.exists(get_task_lock_key("old")) == 0 @@ -212,11 +242,11 @@ def test_release_stale_locks_for_self_skips_when_successor_online(pulp_redisdb, """Given another live AppStatus already owns our name, When startup self-cleanup runs, Then we skip and leave its locks alone.""" conn = pulp_redisdb - seed_locks_via_acquire(conn, "1@pod-abc", "held", exclusive=["res"]) + seed_locks_via_acquire(conn, "1@host-abc", "held", exclusive=["res"]) - app_status = AppStatus.objects.create(app_type="worker", name="1@pod-abc") - make_app_status("1@pod-abc", online=True) # live successor with same name - w = make_worker(conn, "1@pod-abc", app_status) + app_status = AppStatus.objects.create(app_type="worker", name="1@host-abc") + make_app_status("1@host-abc", online=True) # live successor with same name + w = make_worker(conn, "1@host-abc", app_status) assert w.release_stale_locks_for_self() is True # Locks untouched -- the live peer manages them. @@ -438,3 +468,148 @@ def test_cleanup_worker_skips_release_when_successor_online(pulp_redisdb, reset_ w.cleanup_redis_locks_for_worker(missing) # Successor's locks left intact. assert conn.exists(resource_to_lock_key("kres")) == 1 + + +# --------------------------------------------------------------------------- # +# Scale / Redis-pressure proof: cleanup cost must not grow with the keyspace. +# +# The reviewer's concern was "each SCAN touches every key" x "150 workers" = +# O(total_keys) x concurrency of Redis pressure. These tests prove the O(total_keys) +# factor is gone from all three hot paths (measured by counting the Redis commands +# the client issues), with a control proving the legacy SCAN path *did* grow. +# --------------------------------------------------------------------------- # +def _command_log(conn, monkeypatch): + """Record every Redis command issued via this connection as a tuple of str args.""" + log = [] + orig = conn.execute_command + + def spy(*args, **kwargs): + if args: + log.append(tuple(str(a) for a in args)) + return orig(*args, **kwargs) + + monkeypatch.setattr(conn, "execute_command", spy) + return log + + +def _scans(log): + """Return the recorded SCAN commands.""" + return [cmd for cmd in log if cmd and cmd[0].upper() == "SCAN"] + + +def _seed_noise_locks(conn, n): + """Grow the keyspace with n unrelated task locks (no owner registries).""" + pipe = conn.pipeline() + for i in range(n): + pipe.set(get_task_lock_key(f"noise-{i}"), f"other-{i}") + pipe.execute() + + +@pytest.mark.django_db +def test_startup_cleanup_no_keyspace_scan_and_flat(pulp_redisdb, reset_singleton, monkeypatch): + """release_stale_locks_for_self must never SCAN the keyspace and its Redis cost + must not grow with total locks -- so 150 concurrent restarts stay cheap.""" + conn = pulp_redisdb + worker = make_worker(conn, "restarter", make_app_status("restarter", online=True)) + log = _command_log(conn, monkeypatch) + + # Warm the Lua script so EVALSHA caching does not skew the measured counts. + seed_locks_via_acquire(conn, "restarter", "warm", exclusive=["w"]) + worker.release_stale_locks_for_self() + + counts = {} + for total in (100, 50_000): + conn.flushdb() + _seed_noise_locks(conn, total) + seed_locks_via_acquire(conn, "restarter", "rt", exclusive=["e1", "e2"], shared=["s1"]) + log.clear() + assert worker.release_stale_locks_for_self() is True + measured = list(log) + assert _scans(measured) == [], f"startup cleanup must not SCAN (total={total})" + counts[total] = len(measured) + assert conn.exists(get_owner_registry_key("restarter")) == 0 # locks reclaimed + assert counts[100] == counts[50_000], counts + + +@pytest.mark.django_db +def test_reconcile_no_keyspace_scan_and_flat(pulp_redisdb, reset_singleton, monkeypatch): + """reconcile_orphan_redis_locks enumerates owners via SMEMBERS -- no keyspace + SCAN -- so its cost is independent of the number of locks in Redis, and the + orphan owner's locks are reclaimed.""" + conn = pulp_redisdb + worker = make_worker(conn, "reconciler", make_app_status("reconciler", online=True)) + log = _command_log(conn, monkeypatch) + + # Warm the cleanup script (steady state: legacy-scan throttle already taken). + seed_locks_via_acquire(conn, "dead-warm", "dw", exclusive=["dw"]) + conn.set(LEGACY_OWNER_SCAN_KEY, "other", nx=True, ex=LEGACY_OWNER_SCAN_INTERVAL) + worker.reconcile_orphan_redis_locks() + + counts = {} + for total in (100, 50_000): + conn.flushdb() + _seed_noise_locks(conn, total) + # Orphan owner: holds a contended resource, has NO AppStatus row. + seed_locks_via_acquire(conn, "dead", "dt", exclusive=["contended"]) + conn.set(LEGACY_OWNER_SCAN_KEY, "other", nx=True, ex=LEGACY_OWNER_SCAN_INTERVAL) + # Precondition: the resource is currently blocked by the dead owner. + assert acquire_locks(conn, "probe", get_task_lock_key("pt"), ["contended"], []) != [] + log.clear() + worker.reconcile_orphan_redis_locks() + measured = list(log) + + assert _scans(measured) == [], f"reconcile must not SCAN the keyspace (total={total})" + counts[total] = len(measured) + + # Correctness: the orphan's lock is gone and the resource is re-acquirable. + assert conn.exists(get_owner_registry_key("dead")) == 0 + assert acquire_locks(conn, "probe", get_task_lock_key("pt"), ["contended"], []) == [] + assert counts[100] == counts[50_000], counts + + +@pytest.mark.django_db +def test_missing_worker_cleanup_no_keyspace_scan_and_flat( + pulp_redisdb, reset_singleton, monkeypatch +): + """cleanup_redis_locks_for_worker sweeps a missing worker's locks via the + registry -- never a keyspace SCAN -- at constant Redis cost.""" + conn = pulp_redisdb + cleaner = make_worker(conn, "cleaner", make_app_status("cleaner", online=True)) + log = _command_log(conn, monkeypatch) + + # Warm the cleanup script. + warm = make_app_status("gone-warm", online=False) + seed_locks_via_acquire(conn, "gone-warm", "gw", exclusive=["gw"]) + cleaner.cleanup_redis_locks_for_worker(warm) + + counts = {} + for total in (100, 50_000): + conn.flushdb() + _seed_noise_locks(conn, total) + gone = make_app_status(f"gone-{total}", online=False) + seed_locks_via_acquire(conn, f"gone-{total}", f"gt-{total}", exclusive=["contended"]) + log.clear() + assert cleaner.cleanup_redis_locks_for_worker(gone) is True + measured = list(log) + assert _scans(measured) == [], f"missing-worker cleanup must not SCAN (total={total})" + counts[total] = len(measured) + assert conn.exists(get_owner_registry_key(f"gone-{total}")) == 0 # locks reclaimed + assert counts[100] == counts[50_000], counts + + +def test_legacy_scan_control_scales_with_keyspace(pulp_redisdb, monkeypatch): + """Control: the legacy SCAN fallback DOES walk the keyspace, so its SCAN count + grows with total locks. Proves the harness detects the behavior the registry + path removes -- otherwise the "no SCAN" assertions above would be vacuous.""" + conn = pulp_redisdb + log = _command_log(conn, monkeypatch) + scans = {} + for total in (100, 10_000): + conn.flushdb() + _seed_noise_locks(conn, total) + conn.set(get_task_lock_key("legacy"), "legacy-owner") # legacy lock, no registry + log.clear() + assert cleanup_locks_for_owner(conn, "legacy-owner", allow_legacy_scan=True) is True + scans[total] = len(_scans(log)) + assert scans[100] > 0, scans + assert scans[10_000] > scans[100], scans From 73d431a2f8634f4deef120bb4783143abe513f45 Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Wed, 12 Aug 2026 22:25:21 +0200 Subject: [PATCH 3/3] Retrigger CI Previous run failed on a transient network error downloading the GPG fixture key in the pulp-cli tests (unrelated to this change). Co-Authored-By: Claude Opus 4.8 (1M context)