Make JWD a job concern

lib/galaxy/celery/tasks.py

@@ -1,6 +1,5 @@
 import datetime
 import json
-import shutil
 from collections.abc import Callable
 from concurrent.futures import TimeoutError
 from functools import lru_cache
@@ -34,6 +33,7 @@
 from galaxy.datatypes import sniff
 from galaxy.datatypes.registry import Registry as DatatypesRegistry
 from galaxy.exceptions import ObjectNotFound
+from galaxy.job_execution.setup import JobWorkingDirectory
 from galaxy.jobs import MinimalJobWrapper
 from galaxy.managers.collections import DatasetCollectionManager
 from galaxy.managers.dataset_storage_operations import DatasetStorageOperationManager
@@ -810,13 +810,12 @@ def _cleanup_jwds(
 
     def _delete_jwd(job: model.Job) -> bool:
         try:
-            path = object_store.get_filename(job, base_dir="job_work", dir_only=True, obj_dir=True)
-            shutil.rmtree(path)
+            JobWorkingDirectory(job, object_store).delete()
             return True
         except ObjectNotFound:
             return False
         except OSError as e:
-            log.error(f"Error deleting job working directory: {path} : {e.strerror}")
+            log.error(f"Error deleting job working directory for job {job.id}: {e.strerror}")
             return False
 
     deleted_count = 0

lib/galaxy/job_execution/setup.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+import shutil
 import threading
 from typing import (
     Any,
@@ -27,7 +28,11 @@
     JobExportHistoryArchive,
     MetadataFile,
 )
-from galaxy.util import safe_makedirs
+from galaxy.objectstore import ObjectStore
+from galaxy.util import (
+    directory_hash_id,
+    safe_makedirs,
+)
 from galaxy.util.dictifiable import UsesDictVisibleKeys
 from galaxy.util.path import StrPath
 
@@ -328,7 +333,166 @@ def ensure_configs_directory(work_dir: str) -> str:
     return configs_dir
 
 
-def create_working_directory_for_job(object_store, job) -> str:
-    object_store.create(job, base_dir="job_work", dir_only=True, obj_dir=True)
-    working_directory = object_store.get_filename(job, base_dir="job_work", dir_only=True, obj_dir=True)
-    return working_directory
+# Sentinel object-store keys for the job working directory. Centralized so that
+# every JWD operation goes through the same code path instead of open-coding
+# ``if job.working_directory:`` forks at each call site.
+_JOB_WORK_BASE_DIR = "job_work"
+_CLEARED_CONTENTS_EXTRA_DIR = "_cleared_contents"
+
+
+class JobWorkingDirectory:
+    """Unified handle for a job's working directory.
+
+    Hides the two backing strategies — a custom path on ``job.working_directory``
+    (from the ``job_working_directory`` destination param) and the legacy
+    object-store-derived path — behind a single object.
+    """
+
+    __slots__ = ("_job", "_object_store")
+
+    def __init__(self, job: Job, object_store: ObjectStore) -> None:
+        self._job = job
+        self._object_store = object_store
+
+    @property
+    def _custom_path(self) -> str | None:
+        """Read ``job.working_directory`` fresh on each access (not cached)."""
+        return self._job.working_directory
+
+    def _per_job_subpath(self) -> str:
+        """Return the per-job relative subpath ``<directory_hash_id(job.id)>/<job.id>/``."""
+        obj_id = self._job.id
+        return os.path.join(*directory_hash_id(obj_id), str(obj_id))
+
+    def _per_job_path(self, custom_path: str) -> str:
+        """Return ``<custom_base>/<directory_hash_id(job.id)>/<job.id>/``."""
+        return os.path.join(custom_path, self._per_job_subpath())
+
+    def resolve(self) -> str:
+        """Return the working directory path, creating nothing on disk."""
+        if custom_path := self._custom_path:
+            return self._per_job_path(custom_path)
+        return self._object_store.get_filename(
+            self._job,
+            base_dir=_JOB_WORK_BASE_DIR,
+            dir_only=True,
+            obj_dir=True,
+        )
+
+    def exists(self) -> bool:
+        """Check whether the working directory exists on disk."""
+        if custom_path := self._custom_path:
+            return os.path.exists(self._per_job_path(custom_path))
+        return self._object_store.exists(
+            self._job,
+            base_dir=_JOB_WORK_BASE_DIR,
+            dir_only=True,
+            obj_dir=True,
+        )
+
+    def create(self) -> str:
+        """Create the working directory and return its path.
+
+        Raises ``FileExistsError`` if the per-job directory already exists
+        (job id collision or leftover from a crashed run).
+        """
+        if custom_path := self._custom_path:
+            validate_working_directory_path(custom_path)
+            path = self._per_job_path(custom_path)
+            try:
+                os.makedirs(path, exist_ok=False)
+            except FileExistsError as exc:
+                raise FileExistsError(
+                    f"Job working directory already exists for job {self._job.id} "
+                    f"at {path!r}; this indicates a job id collision or a leftover "
+                    f"directory from a previous run."
+                ) from exc
+            return path
+        self._object_store.create(
+            self._job,
+            base_dir=_JOB_WORK_BASE_DIR,
+            dir_only=True,
+            obj_dir=True,
+        )
+        return self._object_store.get_filename(
+            self._job,
+            base_dir=_JOB_WORK_BASE_DIR,
+            dir_only=True,
+            obj_dir=True,
+        )
+
+    def delete(self) -> None:
+        """Recursively delete the working directory.
+
+        For custom paths, only the per-job subdirectory is removed; the
+        admin-supplied base is preserved for other jobs.
+        """
+        if custom_path := self._custom_path:
+            validate_working_directory_path(custom_path)
+            resolved = self._per_job_path(custom_path)
+            if os.path.exists(resolved):
+                shutil.rmtree(resolved)
+            return
+        self._object_store.delete(
+            self._job,
+            base_dir=_JOB_WORK_BASE_DIR,
+            entire_dir=True,
+            dir_only=True,
+            obj_dir=True,
+        )
+
+    def cleared_contents_base(self) -> str:
+        """Return the directory under which cleared JWDs are archived.
+
+        Creates the archive directory if it does not exist. The archive is a
+        sibling tree of the JWD (keyed by ``<hash>/<job.id>``) so resubmits
+        don't collide on the archive name.
+        """
+        if custom_path := self._custom_path:
+            validate_working_directory_path(custom_path)
+            path = os.path.join(
+                custom_path,
+                _CLEARED_CONTENTS_EXTRA_DIR,
+                self._per_job_subpath(),
+            )
+            os.makedirs(path, exist_ok=True)
+            return path
+        self._object_store.create(
+            self._job,
+            base_dir=_JOB_WORK_BASE_DIR,
+            dir_only=True,
+            obj_dir=True,
+            extra_dir=_CLEARED_CONTENTS_EXTRA_DIR,
+            extra_dir_at_root=True,
+        )
+        return self._object_store.get_filename(
+            self._job,
+            base_dir=_JOB_WORK_BASE_DIR,
+            dir_only=True,
+            obj_dir=True,
+            extra_dir=_CLEARED_CONTENTS_EXTRA_DIR,
+            extra_dir_at_root=True,
+        )
+
+
+def validate_working_directory_path(path: str) -> None:
+    """Validate a custom ``job.working_directory`` path before disk operations.
+
+    This is the single canonical validator for ``job.working_directory``. It is
+    called both at set-time (before persisting the column) and at use-time
+    (before each disk operation on a custom path), so callers can rely on the
+    column value being validated without assuming set-time validation is the
+    only guard.
+
+    ``job.working_directory`` is a ``String(1024)`` populated from destination
+    params (admin/TPV-controlled) and is treated as a **parent/base** path:
+    Galaxy appends ``<directory_hash_id(job.id)>/<job.id>/`` for per-job
+    isolation. Refuse to operate on anything that is not an absolute,
+    non-root, non-empty path.
+    """
+    if not path:
+        raise ValueError("Refusing to operate on empty job working_directory")
+    if not os.path.isabs(path):
+        raise ValueError(f"Refusing to operate on relative job working_directory: {path!r}")
+    if os.path.normpath(path) == os.path.normpath(os.sep):
+        raise ValueError("Refusing to operate on filesystem root as job working_directory")

lib/galaxy/jobs/__init__.py

@@ -61,11 +61,12 @@
     collect_shrinked_content_from_path,
 )
 from galaxy.job_execution.setup import (
-    create_working_directory_for_job,
     ensure_configs_directory,
     JobIO,
+    JobWorkingDirectory,
     TOOL_PROVIDED_JOB_METADATA_FILE,
     TOOL_PROVIDED_JOB_METADATA_KEYS,
+    validate_working_directory_path,
 )
 from galaxy.jobs.job_destination import JobDestination
 from galaxy.jobs.mapper import (
@@ -1017,6 +1018,11 @@ def __init__(
         # Tool versioning variables
         self.version_string = ""
         self.__galaxy_lib_dir = None
+        # params is unused but is referenced by the job_destination property
+        # (via job_runner_mapper.get_job_destination(self.params)), which
+        # _setup_working_directory -> _set_working_directory -> get_destination_configuration
+        # now triggers during __init__. Initialize it before that call.
+        self.params = None  # unused
         # If the job has an object_store_id ensure working directory is setup, otherwise
         # wait for that to be assigned before configuring it. Either way the working
         # directory does not to be configured on this object before prepare() is called.
@@ -1026,7 +1032,6 @@ def __init__(
         # resolved
         self._job_io: JobIO | None = None
         self.tool_provided_job_metadata = None
-        self.params = None  # unused
         self.runner_command_line = None
 
         # Wrapper holding the info required to restore and clean up from files used for setting metadata externally
@@ -1321,6 +1326,9 @@ def get_special():
     def _setup_working_directory(self, job=None):
         if job is None:
             job = self.get_job()
+        # Resolve and set job.working_directory from destination params before
+        # creating anything on disk.
+        self._set_working_directory(job)
         try:
             working_directory = self._create_working_directory(job)
             self.__working_directory = working_directory
@@ -1347,29 +1355,19 @@ def guest_ports(self):
     def working_directory(self):
         if self.__working_directory is None:
             job = self.get_job()
-
-            # object_store_id needs to be set before get_filename can be called, this
-            # will also create the directory on the worker.
-            # It is possible these next two lines are not needed - if a job a cannot be recovered
-            # before enqueue is called (seems likely) - this shouldn't be needed.
-            if job.object_store_id:
-                self._set_object_store_ids(job)
-
-            self.__working_directory = self.app.object_store.get_filename(
-                job, base_dir="job_work", dir_only=True, obj_dir=True
-            )
+            self.__working_directory = JobWorkingDirectory(job, self.app.object_store).resolve()
         return self.__working_directory
 
     def working_directory_exists(self) -> bool:
         job = self.get_job()
-        return self.app.object_store.exists(job, base_dir="job_work", dir_only=True, obj_dir=True)
+        return JobWorkingDirectory(job, self.app.object_store).exists()
 
     @property
     def tool_working_directory(self):
         return os.path.join(self.working_directory, "working")
 
     def _create_working_directory(self, job):
-        return create_working_directory_for_job(self.object_store, job)
+        return JobWorkingDirectory(job, self.object_store).create()
 
     def clear_working_directory(self):
         job = self.get_job()
@@ -1379,16 +1377,15 @@ def clear_working_directory(self):
             )
             return
 
-        self.object_store.create(
-            job, base_dir="job_work", dir_only=True, obj_dir=True, extra_dir="_cleared_contents", extra_dir_at_root=True
-        )
-        base = self.object_store.get_filename(
-            job, base_dir="job_work", dir_only=True, obj_dir=True, extra_dir="_cleared_contents", extra_dir_at_root=True
-        )
+        jwd = JobWorkingDirectory(job, self.object_store)
+        base = jwd.cleared_contents_base()
         date_str = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
         arc_dir = os.path.join(base, date_str)
         shutil.move(self.working_directory, arc_dir)
         self._setup_working_directory(job=job)
+        # Flush so the working_directory column survives the sa_session.refresh()
+        # that mark_as_resubmitted() performs at the end of the resubmit flow.
+        self.sa_session.flush()
         log.debug("(%s) Previous working directory moved to %s", self.job_id, arc_dir)
 
     def default_compute_environment(self, job=None):
@@ -1852,6 +1849,46 @@ def _set_object_store_ids_basic(self, job: Job):
         job.object_store_id = object_store_populator.object_store_id
         self._setup_working_directory(job=job)
 
+    def _set_working_directory(self, job: Job):
+        """Resolve and persist ``job_working_directory`` from destination params.
+
+        Validates the resolved path before persisting it via the canonical
+        ``validate_working_directory_path`` (also enforced at disk-operation
+        time by ``JobWorkingDirectory``).
+
+        Does not flush the session. Callers that need the column to survive a
+        subsequent ``sa_session.refresh()`` (notably the resubmit path via
+        ``clear_working_directory``) must flush/commit themselves. The enqueue
+        path is covered by the ``commit()`` in ``enqueue()``.
+        """
+        # Reads from ``job.destination_params`` (persisted) directly, not via
+        # ``get_destination_configuration``. The latter would fall back to
+        # ``app.config.job_working_directory`` (mapped to ``jobs_directory``) when
+        # no destination param specifies the key, causing the column to be set to
+        # the config default even for destinations that don't specify one. We only
+        # want to persist the column when a destination param explicitly sets it.
+
+        # Reading the persisted params is also what makes the resubmit reorder
+        # load-bearing: ``set_job_destination`` persists the new params but does
+        # not update the mapper's cached destination, so reading
+        # ``self.job_destination.params`` directly would return stale values.
+        working_directory = (job.destination_params or {}).get("job_working_directory", None)
+        if isinstance(working_directory, str):
+            validate_working_directory_path(working_directory)
+            job.working_directory = working_directory
+        else:
+            # Reset to None so a resubmitted job whose new destination no longer
+            # specifies a custom path falls back to the object-store strategy
+            # instead of retaining the stale value from the prior attempt.
+            job.working_directory = None
+            if working_directory is not None:
+                log.warning(
+                    "(%s) job_working_directory destination param is not a string (%r), "
+                    "falling back to object-store path.",
+                    self.job_id,
+                    working_directory,
+                )
+
     def _set_object_store_ids_full(self, job: Job):
         user = job.user
         object_store_id = self.get_destination_configuration("object_store_id", None)
@@ -2388,9 +2425,8 @@ def cleanup(self, delete_files: bool = True) -> None:
                         if e.errno != errno.ENOENT:
                             raise
             if delete_files:
-                self.object_store.delete(
-                    self.get_job(), base_dir="job_work", entire_dir=True, dir_only=True, obj_dir=True
-                )
+                job = self.get_job()
+                JobWorkingDirectory(job, self.object_store).delete()
         except Exception:
             log.exception("Unable to cleanup job %d", self.job_id)