From 11ebc3bb60a8cc63bd63804ff3547c1acb17a3d5 Mon Sep 17 00:00:00 2001 From: Yasen Date: Wed, 4 Nov 2026 13:10:52 +0100 Subject: [PATCH] Fix task cancellation during graceful worker shutdown A graceful shutdown (SIGTERM/SIGHUP) sets task_grace_timeout to None so the worker waits indefinitely for the running task. The cancellation path reused task_grace_timeout as its kill deadline, so canceling that task only logged "Wait for canceled task to abort." forever and never sent SIGUSR1. The task stayed in "canceling", the worker could never exit, and it kept heartbeating while holding the task's exclusive resource locks, blocking every task queued on them. Track the kill deadline for a canceled (or shutdown-aborted) task in a separate cancel_kill_deadline, reset per task, so an explicit cancellation takes effect regardless of the shutdown grace period. This also stops the cancel path from mutating task_grace_timeout. Once a task is being aborted, the shutdown branch no longer overrides its state, so a user cancellation stays CANCELED instead of becoming FAILED. Applied to both PulpcoreWorker and RedisWorker. Co-Authored-By: Claude Opus 5.5 --- .../+cancel-during-graceful-shutdown.bugfix | 2 + pulpcore/tasking/redis_worker.py | 18 +- pulpcore/tasking/worker.py | 19 +- .../test_cancel_during_graceful_shutdown.py | 189 ++++++++++++++++++ 4 files changed, 222 insertions(+), 6 deletions(-) create mode 100644 CHANGES/+cancel-during-graceful-shutdown.bugfix create mode 100644 pulpcore/tests/unit/tasking/test_cancel_during_graceful_shutdown.py diff --git a/CHANGES/+cancel-during-graceful-shutdown.bugfix b/CHANGES/+cancel-during-graceful-shutdown.bugfix new file mode 100644 index 0000000000..cb74afd2dc --- /dev/null +++ b/CHANGES/+cancel-during-graceful-shutdown.bugfix @@ -0,0 +1,2 @@ +Fixed canceling a task while its worker is shutting down gracefully. The cancellation never +aborted the task, so the worker could not exit and kept holding the task's resource locks. diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index 52d319f4d9..bc2604e792 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -185,6 +185,13 @@ def __init__(self): # and set to None for fully graceful shutdown. self.task_grace_timeout = timezone.now() + # Deadline for force-killing an explicitly canceled (or shutdown-aborted) task + # process. Kept separate from task_grace_timeout: that field can be set to None + # to make the worker wait forever for a task to finish naturally on a graceful + # shutdown, but an explicit cancellation must still take effect on its own + # schedule regardless of that wait, or it can never actually abort the task. + self.cancel_kill_deadline = None + self.worker_cleanup_countdown = random.randint( int(WORKER_CLEANUP_INTERVAL / 10), WORKER_CLEANUP_INTERVAL ) @@ -702,6 +709,7 @@ def supervise_task(self, task): self.task = task cancel_state = None cancel_reason = None + self.cancel_kill_deadline = None domain = task.pulp_domain with TemporaryDirectory(dir=".") as task_working_dir_rel_path: task_process = Process(target=perform_task, args=(task.pk, task_working_dir_rel_path)) @@ -710,10 +718,10 @@ def supervise_task(self, task): # Heartbeat while waiting for task to complete while task_process.is_alive(): if cancel_state: - if self.task_grace_timeout is None or self.task_grace_timeout > timezone.now(): + if self.cancel_kill_deadline > timezone.now(): _logger.info("Wait for canceled task to abort.") else: - self.task_grace_timeout = timezone.now() + timezone.timedelta( + self.cancel_kill_deadline = timezone.now() + timezone.timedelta( seconds=TASK_KILL_INTERVAL ) _logger.info( @@ -741,6 +749,7 @@ def supervise_task(self, task): domain.name, ) cancel_state = TASK_STATES.CANCELED + self.cancel_kill_deadline = timezone.now() clear_cancel_signal(task.pk) if self.sentinel in r: @@ -751,7 +760,9 @@ def supervise_task(self, task): break # If shutdown was requested, handle gracefully or abort - if self.shutdown_requested: + # Once the task is being aborted, keep its state (e.g. CANCELED) and let the + # cancellation branch above drive the kill. + if self.shutdown_requested and not cancel_state: if self.task_grace_timeout is None or self.task_grace_timeout > timezone.now(): msg = ( "Worker shutdown requested, waiting for task {pk} in domain: {name} " @@ -766,6 +777,7 @@ def supervise_task(self, task): ) cancel_state = TASK_STATES.FAILED cancel_reason = "Aborted during worker shutdown." + self.cancel_kill_deadline = timezone.now() task_process.join() if not cancel_state and task_process.exitcode != 0: diff --git a/pulpcore/tasking/worker.py b/pulpcore/tasking/worker.py index 236afaf4c9..2cecae3866 100644 --- a/pulpcore/tasking/worker.py +++ b/pulpcore/tasking/worker.py @@ -107,6 +107,14 @@ def __init__(self, auxiliary=False): # It will be set into the future on moderately graceful worker shutdown, # and set to None for fully graceful shutdown. self.task_grace_timeout = timezone.now() + + # Deadline for force-killing an explicitly canceled (or shutdown-aborted) task + # process. Kept separate from task_grace_timeout: that field can be set to None + # to make the worker wait forever for a task to finish naturally on a graceful + # shutdown, but an explicit cancellation must still take effect on its own + # schedule regardless of that wait, or it can never actually abort the task. + self.cancel_kill_deadline = None + self.worker_cleanup_countdown = random.randint( int(WORKER_CLEANUP_INTERVAL / 10), WORKER_CLEANUP_INTERVAL ) @@ -460,16 +468,17 @@ def supervise_task(self, task): self.task = task cancel_state = None cancel_reason = None + self.cancel_kill_deadline = None domain = task.pulp_domain with TemporaryDirectory(dir=".") as task_working_dir_rel_path: task_process = Process(target=perform_task, args=(task.pk, task_working_dir_rel_path)) task_process.start() while True: if cancel_state: - if self.task_grace_timeout is None or self.task_grace_timeout > timezone.now(): + if self.cancel_kill_deadline > timezone.now(): _logger.info("Wait for canceled task to abort.") else: - self.task_grace_timeout = timezone.now() + timezone.timedelta( + self.cancel_kill_deadline = timezone.now() + timezone.timedelta( seconds=TASK_KILL_INTERVAL ) _logger.info( @@ -496,6 +505,7 @@ def supervise_task(self, task): ) cancel_state = TASK_STATES.CANCELED self.cancel_task = False + self.cancel_kill_deadline = timezone.now() if self.wakeup_unblock: self.unblock_tasks() if task_process.sentinel in r: @@ -503,7 +513,9 @@ def supervise_task(self, task): break if self.sentinel in r: os.read(self.sentinel, 256) - if self.shutdown_requested: + # Once the task is being aborted, keep its state (e.g. CANCELED) and let the + # cancellation branch above drive the kill. + if self.shutdown_requested and not cancel_state: if self.task_grace_timeout is None or self.task_grace_timeout > timezone.now(): msg = ( "Worker shutdown requested, waiting for task {pk} in domain: {name} " @@ -519,6 +531,7 @@ def supervise_task(self, task): ) cancel_state = TASK_STATES.FAILED cancel_reason = "Aborted during worker shutdown." + self.cancel_kill_deadline = timezone.now() task_process.join() if not cancel_state and task_process.exitcode != 0: diff --git a/pulpcore/tests/unit/tasking/test_cancel_during_graceful_shutdown.py b/pulpcore/tests/unit/tasking/test_cancel_during_graceful_shutdown.py new file mode 100644 index 0000000000..958fda1cff --- /dev/null +++ b/pulpcore/tests/unit/tasking/test_cancel_during_graceful_shutdown.py @@ -0,0 +1,189 @@ +""" +Unit tests for task cancellation while a worker is shutting down gracefully. + +A graceful shutdown (SIGTERM/SIGHUP) sets ``task_grace_timeout`` to ``None`` so the worker waits +indefinitely for the in-flight task to finish. An explicit cancellation of that task must still +abort it (SIGUSR1 to the task process); previously the cancellation reused +``task_grace_timeout`` as its kill deadline, so it waited forever too and the worker deadlocked, +heartbeating and holding the task's resource locks. +""" + +import signal +from datetime import timedelta +from unittest import mock + +import pytest + +from pulpcore.constants import TASK_STATES +from pulpcore.tasking import redis_worker as redis_worker_module +from pulpcore.tasking import worker as worker_module + +MAX_LOOP_ITERATIONS = 50 + + +class FakeProcess: + """Stand-in for the task ``multiprocessing.Process``; it only exits once it gets SIGUSR1.""" + + def __init__(self, *args, **kwargs): + self.pid = 4242 + self.sentinel = object() + self.alive = True + self.exitcode = None + + def start(self): + pass + + def is_alive(self): + return self.alive + + def join(self): + pass + + def finish(self, exitcode=0): + self.alive = False + self.exitcode = exitcode + + +class Harness: + """Patches the process/select/kill machinery used by ``supervise_task``.""" + + def __init__(self, monkeypatch, module, on_first_select=None, finish_after=None): + self.process = None + self.kills = [] + self.select_calls = 0 + self.on_first_select = on_first_select + self.finish_after = finish_after + + def make_process(*args, **kwargs): + self.process = FakeProcess() + return self.process + + def fake_kill(pid, sig): + assert pid == self.process.pid + self.kills.append(sig) + if sig == signal.SIGUSR1: + self.process.finish(exitcode=0) + + def fake_select(rlist, wlist, xlist, timeout=None): + self.select_calls += 1 + if self.select_calls > MAX_LOOP_ITERATIONS: + raise AssertionError("supervise_task never let the task process exit (deadlock)") + if self.select_calls == 1 and self.on_first_select: + self.on_first_select() + if self.finish_after and self.select_calls >= self.finish_after: + self.process.finish(exitcode=0) + return ([self.process.sentinel] if not self.process.alive else [], [], []) + + monkeypatch.setattr(module, "Process", make_process) + monkeypatch.setattr(module.os, "kill", fake_kill) + monkeypatch.setattr(module.select, "select", fake_select) + monkeypatch.setattr(module, "connection", mock.MagicMock()) + + +def _make_task(): + task = mock.MagicMock() + task.pk = "01a1841c-b360-7f5d-adb6-4ee443942c8e" + task.pulp_domain.name = "default" + task.reserved_resources_record = None + return task + + +def _pulpcore_worker(graceful_shutdown): + worker = object.__new__(worker_module.PulpcoreWorker) + worker.sentinel = -1 + worker.heartbeat_period = timedelta(seconds=1) + worker.wakeup_unblock = False + worker.shutdown_requested = graceful_shutdown + worker.task_grace_timeout = None if graceful_shutdown else worker_module.timezone.now() + worker.cancel_kill_deadline = None + worker.beat = mock.Mock() + worker.cancel_abandoned_task = mock.Mock() + worker.notify_workers = mock.Mock() + return worker + + +def _redis_worker(graceful_shutdown): + worker = object.__new__(redis_worker_module.RedisWorker) + worker.sentinel = -1 + worker.heartbeat_period = timedelta(seconds=1) + worker.shutdown_requested = graceful_shutdown + worker.task_grace_timeout = None if graceful_shutdown else redis_worker_module.timezone.now() + worker.cancel_kill_deadline = None + worker.beat = mock.Mock() + worker._maybe_release_locks = mock.Mock() + return worker + + +@pytest.fixture +def in_tmp_dir(tmp_path, monkeypatch): + # supervise_task creates its TemporaryDirectory in the current directory. + monkeypatch.chdir(tmp_path) + + +def _patch_redis_cancel_signal(monkeypatch, pending): + from pulpcore.tasking import _util, redis_tasks + + monkeypatch.setattr(redis_tasks, "check_cancel_signal", lambda pk: pending["value"]) + monkeypatch.setattr(redis_tasks, "clear_cancel_signal", lambda pk: pending.update(value=False)) + monkeypatch.setattr(_util, "delete_incomplete_resources", mock.Mock()) + + +# --- PulpcoreWorker ------------------------------------------------------------------------ + + +@pytest.mark.parametrize("graceful_shutdown", [True, False], ids=["graceful-shutdown", "running"]) +def test_pulpcore_worker_cancel_aborts_task(monkeypatch, in_tmp_dir, graceful_shutdown): + worker = _pulpcore_worker(graceful_shutdown) + harness = Harness( + monkeypatch, worker_module, on_first_select=lambda: setattr(worker, "cancel_task", True) + ) + task = _make_task() + + worker.supervise_task(task) + + assert harness.kills == [signal.SIGUSR1] + worker.cancel_abandoned_task.assert_called_once_with(task, TASK_STATES.CANCELED, None) + + +def test_pulpcore_worker_graceful_shutdown_waits_for_task(monkeypatch, in_tmp_dir): + worker = _pulpcore_worker(graceful_shutdown=True) + harness = Harness(monkeypatch, worker_module, finish_after=5) + + worker.supervise_task(_make_task()) + + assert harness.kills == [] + assert harness.select_calls == 5 + worker.cancel_abandoned_task.assert_not_called() + + +# --- RedisWorker --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("graceful_shutdown", [True, False], ids=["graceful-shutdown", "running"]) +def test_redis_worker_cancel_aborts_task(monkeypatch, in_tmp_dir, graceful_shutdown): + pending = {"value": False} + _patch_redis_cancel_signal(monkeypatch, pending) + worker = _redis_worker(graceful_shutdown) + harness = Harness( + monkeypatch, redis_worker_module, on_first_select=lambda: pending.update(value=True) + ) + task = _make_task() + + worker.supervise_task(task) + + assert harness.kills == [signal.SIGUSR1] + task.set_canceled.assert_called_once_with(final_state=TASK_STATES.CANCELED, reason=None) + + +def test_redis_worker_graceful_shutdown_waits_for_task(monkeypatch, in_tmp_dir): + pending = {"value": False} + _patch_redis_cancel_signal(monkeypatch, pending) + worker = _redis_worker(graceful_shutdown=True) + harness = Harness(monkeypatch, redis_worker_module, finish_after=5) + task = _make_task() + + worker.supervise_task(task) + + assert harness.kills == [] + assert harness.select_calls == 5 + task.set_canceled.assert_not_called()