Adds 'redis' WORKER_TYPE

This adds WORKER_TYPE setting. The default value is 'pulpcore'. When 'redis' is selected,
the tasking system uses Redis to lock resources. Redis workers produce less load on the
PostgreSQL database.

closes: #7210

Generated By: Claude Code.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: dkliban

.github/workflows/scripts/before_install.sh

@@ -105,7 +105,7 @@ if [ "$TEST" = "azure" ]; then
     image: "mcr.microsoft.com/azure-storage/azurite"
     command: "azurite-blob --skipApiVersionCheck --blobHost 0.0.0.0"
 azure_test: true
-pulp_scenario_settings: {"MEDIA_ROOT": "", "STORAGES": {"default": {"BACKEND": "storages.backends.azure_storage.AzureStorage", "OPTIONS": {"account_key": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "account_name": "devstoreaccount1", "azure_container": "pulp-test", "connection_string": "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://ci-azurite:10000/devstoreaccount1;", "expiration_secs": 120, "location": "pulp3", "overwrite_files": true}}, "staticfiles": {"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage"}}, "api_root_rewrite_header": "X-API-Root", "content_origin": null, "domain_enabled": true, "rest_framework__default_authentication_classes": "@merge pulpcore.app.authentication.PulpRemoteUserAuthentication", "rest_framework__default_permission_classes": ["pulpcore.plugin.access_policy.DefaultAccessPolicy"]}
+pulp_scenario_settings: {"MEDIA_ROOT": "", "STORAGES": {"default": {"BACKEND": "storages.backends.azure_storage.AzureStorage", "OPTIONS": {"account_key": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "account_name": "devstoreaccount1", "azure_container": "pulp-test", "connection_string": "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://ci-azurite:10000/devstoreaccount1;", "expiration_secs": 120, "location": "pulp3", "overwrite_files": true}}, "staticfiles": {"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage"}}, "api_root_rewrite_header": "X-API-Root", "content_origin": null, "domain_enabled": true, "rest_framework__default_authentication_classes": "@merge pulpcore.app.authentication.PulpRemoteUserAuthentication", "rest_framework__default_permission_classes": ["pulpcore.plugin.access_policy.DefaultAccessPolicy"], "task_diagnostics": ["memory"], "worker_type": "redis"}
 pulp_scenario_env: {}
 VARSYAML
 fi

CHANGES/7210.feature

@@ -0,0 +1 @@
+Added WORKER_TYPE setting. Defaults to 'pulpcore'. 'redis' is also available.

docs/admin/reference/settings.md

@@ -413,6 +413,20 @@ The number of seconds before a worker should be considered lost.
 
 Defaults to `30` seconds.
 
+### WORKER\_TYPE
+
+The worker implementation to use for task execution.
+
+Available options:
+
+- `"pulpcore"` (default): Uses PostgreSQL advisory locks for task coordination. This is the traditional worker implementation.
+- `"redis"`: Uses Redis distributed locks for task coordination. This implementation produces less load on the DB.
+
+!!! note
+    The Redis worker requires a Redis server to be configured and accessible.
+
+Defaults to `"pulpcore"`.
+
 ### WORKING\_DIRECTORY
 
 The directory used by workers to stage files temporarily.

pulpcore/app/redis_connection.py

@@ -7,8 +7,12 @@
 _a_conn = None
 
 
+def _redis_is_needed():
+    return settings.get("CACHE_ENABLED") or settings.get("WORKER_TYPE") == "redis"
+
+
 def _get_connection_from_class(redis_class):
-    if not settings.get("CACHE_ENABLED"):
+    if not _redis_is_needed():
         return None
     redis_url = settings.get("REDIS_URL")
     if redis_url is not None:

pulpcore/app/settings.py

@@ -292,6 +292,10 @@
 CONTENT_APP_TTL = 30
 WORKER_TTL = 30
 
+# Worker implementation type
+# Options: "pulpcore" (default, PostgreSQL advisory locks) or "redis" (Redis distributed locks)
+WORKER_TYPE = "pulpcore"
+
 # Seconds for a task to finish on semi graceful worker shutdown (approx)
 # On SIGHUP, SIGTERM the currently running task will be awaited forever.
 # On SIGINT, this value represents the time before the worker will attempt to kill the subprocess.
@@ -457,6 +461,31 @@
     "for more information."
 )
 
+worker_type_allowed_values_validator = Validator(
+    "WORKER_TYPE",
+    is_in=["pulpcore", "redis"],
+    messages={"operations": "WORKER_TYPE must be either 'pulpcore' or 'redis', got '{value}'."},
+)
+
+worker_type_redis_validator_condition = Validator("WORKER_TYPE", eq="redis")
+worker_type_redis_url_validator = Validator(
+    "REDIS_URL", must_exist=True, when=worker_type_redis_validator_condition
+)
+worker_type_redis_host_validator = Validator(
+    "REDIS_HOST", must_exist=True, when=worker_type_redis_validator_condition
+)
+worker_type_redis_port_validator = Validator(
+    "REDIS_PORT", must_exist=True, when=worker_type_redis_validator_condition
+)
+worker_type_redis_validator = worker_type_redis_url_validator | (
+    worker_type_redis_host_validator & worker_type_redis_port_validator
+)
+worker_type_redis_validator.messages["combined"] = (
+    "WORKER_TYPE is set to 'redis' but it requires REDIS to be configured. Please check "
+    "https://pulpproject.org/pulpcore/docs/admin/reference/settings/?h=settings#redis-settings "
+    "for more information."
+)
+
 sha256_validator = Validator(
     "ALLOWED_CONTENT_CHECKSUMS",
     cont="sha256",
@@ -551,6 +580,8 @@ def otel_middleware_hook(settings):
     validators=[
         api_root_validator,
         cache_validator,
+        worker_type_allowed_values_validator,
+        worker_type_redis_validator,
         sha256_validator,
         storage_validator,
         unknown_algs_validator,

pulpcore/app/tasks/test.py

@@ -1,5 +1,7 @@
 import asyncio
 import backoff
+import os
+import signal
 import time
 from pulpcore.app.models import TaskGroup
 from pulpcore.tasking.tasks import dispatch
@@ -32,3 +34,44 @@ def dummy_group_task(inbetween=3, intervals=None):
     for interval in intervals:
         dispatch(sleep, args=(interval,), task_group=task_group)
         time.sleep(inbetween)
+
+
+def missing_worker():
+    """
+    Simulates a worker crash by sending SIGKILL to parent process and itself.
+
+    This task is used for testing worker cleanup behavior when a worker
+    unexpectedly dies while executing a task.
+    """
+    parent_pid = os.getppid()
+    current_pid = os.getpid()
+
+    # Kill parent process (the worker)
+    os.kill(parent_pid, signal.SIGKILL)
+
+    # Kill current process (the task subprocess)
+    os.kill(current_pid, signal.SIGKILL)
+
+
+def failing_task(error_message="Task intentionally failed"):
+    """
+    A task that always raises a RuntimeError.
+
+    This task is used for testing error handling in worker task execution.
+
+    Args:
+        error_message (str): The error message to include in the RuntimeError
+    """
+    raise RuntimeError(error_message)
+
+
+async def afailing_task(error_message="Task intentionally failed"):
+    """
+    An async task that always raises a RuntimeError.
+
+    This task is used for testing error handling in immediate task execution.
+
+    Args:
+        error_message (str): The error message to include in the RuntimeError
+    """
+    raise RuntimeError(error_message)

pulpcore/app/viewsets/task.py

@@ -42,7 +42,7 @@
     CreatedResourcesFilter,
 )
 from pulpcore.constants import TASK_INCOMPLETE_STATES, TASK_STATES
-from pulpcore.tasking.tasks import dispatch, cancel_task, cancel_task_group
+from pulpcore.tasking.tasks import cancel_task, cancel_task_group, dispatch
 from pulpcore.app.role_util import get_objects_for_user
 
 
@@ -228,6 +228,7 @@ def partial_update(self, request, pk=None, partial=True):
 
         task = self.get_object()
         task = cancel_task(task.pk)
+
         # Check whether task is actually canceled
         http_status = (
             None
@@ -346,6 +347,7 @@ def partial_update(self, request, pk=None, partial=True):
                 ).count()
             ):
                 raise PermissionDenied()
+
         task_group = cancel_task_group(task_group.pk)
         # Check whether task group is actually canceled
         serializer = TaskGroupSerializer(task_group, context={"request": request})

pulpcore/pytest_plugin.py

@@ -55,6 +55,14 @@ def pytest_collection_modifyitems(config, items):
             if "nightly" in item.keywords:
                 item.add_marker(skip_nightly)
 
+        # Skip long_running tests if --timeout is below 600
+        timeout = config.getoption("--timeout", default=None)
+        if timeout is None or float(timeout) < 600:
+            skip_long = pytest.mark.skip(reason="needs --timeout >= 600 to run")
+            for item in items:
+                if "long_running" in item.keywords:
+                    item.add_marker(skip_long)
+
 
 class PulpTaskTimeoutError(Exception):
     """Exception to describe task and taskgroup timeout errors."""
@@ -81,6 +89,10 @@ def pytest_configure(config):
         "markers",
         "nightly: marks tests as intended to run during the nightly CI run",
     )
+    config.addinivalue_line(
+        "markers",
+        "long_running: marks tests that need a long pytest timeout (--timeout >= 600)",
+    )
 
 
 @pytest.fixture(scope="session")

pulpcore/tasking/_util.py

@@ -23,7 +23,14 @@
     configure_periodic_telemetry,
 )
 from pulpcore.constants import TASK_FINAL_STATES, TASK_STATES
-from pulpcore.tasking.tasks import dispatch, execute_task
+from pulpcore.tasking.tasks import dispatch
+
+# Conditionally import execute_task based on WORKER_TYPE
+if settings.WORKER_TYPE == "redis":
+    from pulpcore.tasking.redis_tasks import execute_task
+else:
+    from pulpcore.tasking.tasks import execute_task
+
 
 _logger = logging.getLogger(__name__)
 

pulpcore/tasking/entrypoint.py

@@ -10,6 +10,7 @@
 
 from django.conf import settings  # noqa: E402: module level not at top
 from pulpcore.tasking.worker import PulpcoreWorker  # noqa: E402: module level not at top
+from pulpcore.tasking.redis_worker import RedisWorker  # noqa: E402: module level not at top
 
 _logger = logging.getLogger(__name__)
 
@@ -58,6 +59,16 @@ def worker(
     if name_template:
         settings.set("WORKER_NAME_TEMPLATE", name_template)
 
-    _logger.info("Starting distributed type worker")
+    worker_type = settings.WORKER_TYPE
+    _logger.info("Starting %s worker", worker_type)
 
-    PulpcoreWorker(auxiliary=auxiliary).run(burst=burst)
+    if worker_type == "redis":
+        if auxiliary:
+            _logger.warning(
+                "RedisWorker does not support auxiliary mode, ignoring --auxiliary flag"
+            )
+        RedisWorker().run(burst=burst)
+    elif worker_type == "pulpcore":
+        PulpcoreWorker(auxiliary=auxiliary).run(burst=burst)
+    else:
+        raise ValueError(f"Invalid WORKER_TYPE: {worker_type}. Must be 'pulpcore' or 'redis'.")