Update Changelog

Author: pulpbot

CHANGES.md

@@ -8,6 +8,28 @@
 
 [//]: # (towncrier release notes start)
 
+## 3.92.1 (2025-10-29) {: #3.92.1 }
+
+### REST API {: #3.92.1-rest-api }
+
+#### Bugfixes {: #3.92.1-rest-api-bugfix }
+
+- Fixed an issue with dispatching repository.add_and_remove tasks for pull-through caching
+
+### Plugin API {: #3.92.1-plugin-api }
+
+No significant changes.
+
+### Pulp File {: #3.92.1-pulp-file }
+
+No significant changes.
+
+### Pulp Cert Guard {: #3.92.1-pulp-cert-guard }
+
+No significant changes.
+
+---
+
 ## 3.92.0 (2025-10-28) {: #3.92.0 }
 
 ### REST API {: #3.92.0-rest-api }

pulpcore/app/util.py

@@ -8,7 +8,7 @@
 from functools import lru_cache
 from gettext import gettext as _
 from urllib.parse import urlparse
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
 from contextvars import ContextVar
 from datetime import timedelta
 from uuid import UUID
@@ -619,6 +619,15 @@ def set_domain(new_domain):
     return new_domain
 
 
+@contextmanager
+def with_domain(domain):
+    try:
+        token = current_domain.set(domain)
+        yield
+    finally:
+        current_domain.reset(token)
+
+
 def cache_key(base_path):
     """Returns the base-key(s) used in the Cache for the passed base_path(s)."""
     if settings.DOMAIN_ENABLED:

pulpcore/tasking/worker.py

@@ -30,6 +30,7 @@
 from pulpcore.metrics import init_otel_meter
 from pulpcore.app.apps import pulp_plugin_configs
 from pulpcore.app.models import Task, AppStatus
+from pulpcore.app.util import with_domain
 
 from pulpcore.tasking.storage import WorkerDirectory
 from pulpcore.tasking._util import (
@@ -38,6 +39,7 @@
     perform_task,
     startup_hook,
 )
+from pulpcore.tasking.tasks import using_workdir, execute_task
 
 _logger = logging.getLogger(__name__)
 random.seed()
@@ -443,6 +445,18 @@ def sleep(self):
                 os.read(self.sentinel, 256)
         _logger.debug(_("Worker %s leaving sleep state."), self.name)
 
+    def supervise_immediate_task(self, task):
+        """Call and supervise the immediate async task process.
+
+        This function must only be called while holding the lock for that task."""
+        self.task = task
+        with with_domain(task.domain), using_workdir():
+            # TODO set user, cid, ...
+            execute_task(task)
+        if task.reserved_resources_record:
+            self.notify_workers(TASK_WAKEUP_UNBLOCK)
+        self.task = None
+
     def supervise_task(self, task):
         """Call and supervise the task process while heart beating.
 
@@ -596,7 +610,10 @@ def handle_unblocked_tasks(self):
                     # A running task without a lock must be abandoned.
                     self.cancel_abandoned_task(task, TASK_STATES.FAILED, "Worker has gone missing.")
                 elif task.state == TASK_STATES.WAITING and self.is_compatible(task):
-                    self.supervise_task(task)
+                    if task.immediate:
+                        self.supervise_immediate_task(task)
+                    else:
+                        self.supervise_task(task)
                 else:
                     # Probably incompatible, but for whatever reason we didn't pick it up this time,
                     # we don't need to look at it ever again.