diff --git a/CHANGES/7919.bugfix b/CHANGES/7919.bugfix new file mode 100644 index 0000000000..008380a8b5 --- /dev/null +++ b/CHANGES/7919.bugfix @@ -0,0 +1 @@ +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/CHANGES/7920.bugfix b/CHANGES/7920.bugfix new file mode 100644 index 0000000000..2cfbffa462 --- /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 e38fdbc933..a44db73789 100644 --- a/pulpcore/tasking/redis_locks.py +++ b/pulpcore/tasking/redis_locks.py @@ -17,6 +17,25 @@ # 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:" + +# 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" +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 +48,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 +109,15 @@ 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. +-- 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 {} """ @@ -107,6 +136,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 +148,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) @@ -127,12 +158,18 @@ -- 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] -- 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,15 +180,77 @@ 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 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} """ +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) +redis.call("srem", "pulp:active_owners", lock_owner) +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 +277,120 @@ 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. + + 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: + 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: + released += delete_if_owner(keys=[key], args=[owner, registry_key]) + else: + 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): + """ + 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: + released = 0 + if redis_conn.exists(registry_key): + cleanup_script = redis_conn.register_script(REDIS_CLEANUP_OWNER_LOCKS_SCRIPT) + released = cleanup_script(keys=[], args=[owner]) + elif allow_legacy_scan: + 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) + return False + + +def collect_lock_owners(redis_conn, allow_legacy_scan=False): + """ + Return the set of owner names that currently hold Redis locks. + + 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 + allow_legacy_scan (bool): Also discover owners of pre-registry (legacy) locks + + Returns: + set: Owner names holding at least one lock. + """ + 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): + 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 59ffdca9f9..99ce297413 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,16 @@ def __init__(self): except redis.RedisError: raise RuntimeError(f"Redis is not reachable. RedisWorker {self.name} cannot start.") + # 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() os.set_blocking(self.sentinel, False) @@ -297,80 +313,181 @@ 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. - 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 + 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. + + 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 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). + + 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)) + 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 + ) 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 0000000000..e69de29bb2 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 0000000000..fdb6ce3d43 --- /dev/null +++ b/pulpcore/tests/unit/tasking/test_orphan_redis_locks.py @@ -0,0 +1,615 @@ +""" +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 ( + LEGACY_OWNER_SCAN_INTERVAL, + LEGACY_OWNER_SCAN_KEY, + 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 + + +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 +# --------------------------------------------------------------------------- # +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@host-abc", "old", exclusive=["res"]) + + 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 + 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@host-abc", "held", exclusive=["res"]) + + 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. + 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 + + +# --------------------------------------------------------------------------- # +# 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