[RFC] Unify object store and file sources on a shared transport

doc/source/dev/index.rst

@@ -27,3 +27,4 @@ A multi-hour long video playlist covering these slides can be found at
   debugging_galaxy_slurm
   translating
   create_release
+  objectstore_filesource_unification

lib/galaxy/files/sources/__init__.py

@@ -611,6 +611,84 @@ def _realize_to(
     ):
         pass
 
+    def exists(
+        self,
+        path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: FilesSourceOptions | None = None,
+    ) -> bool:
+        """Return whether an object exists at ``path`` (relative to the URI root)."""
+        self._check_user_access(user_context)
+        self._check_credentials_fresh()
+        context = self._get_runtime_context(opts, user_context)
+        return self._exists(path, context)
+
+    def _exists(self, path: str, context: FilesSourceRuntimeContext[TResolvedConfig]) -> bool:
+        raise NotImplementedError()
+
+    def size(
+        self,
+        path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: FilesSourceOptions | None = None,
+    ) -> int:
+        """Return the size in bytes of the object at ``path`` (relative to the URI root)."""
+        self._check_user_access(user_context)
+        self._check_credentials_fresh()
+        context = self._get_runtime_context(opts, user_context)
+        return self._size(path, context)
+
+    def _size(self, path: str, context: FilesSourceRuntimeContext[TResolvedConfig]) -> int:
+        raise NotImplementedError()
+
+    def remove(
+        self,
+        path: str,
+        recursive: bool = False,
+        user_context: "OptionalUserContext" = None,
+        opts: FilesSourceOptions | None = None,
+    ) -> bool:
+        """Delete the object at ``path`` (relative to the URI root).
+
+        With ``recursive=True``, delete every object beneath ``path``.
+        """
+        self._ensure_writeable()
+        self._check_user_access(user_context)
+        self._check_credentials_fresh()
+        context = self._get_runtime_context(opts, user_context)
+        return self._remove(path, context, recursive=recursive)
+
+    def _remove(self, path: str, context: FilesSourceRuntimeContext[TResolvedConfig], recursive: bool = False) -> bool:
+        raise NotImplementedError()
+
+    def get_local_path(
+        self,
+        path: str,
+        user_context: "OptionalUserContext" = None,
+        opts: FilesSourceOptions | None = None,
+    ) -> str:
+        """Resolve ``path`` to a local filesystem path the caller may read IN PLACE.
+
+        Unlike :meth:`realize_to` (which copies to a caller-chosen destination), the returned
+        path is owned by the file source -- callers must not move or delete it. Local sources
+        return their real path (zero copy); cache-backed sources return a cache path. Sources
+        without a stable local representation do not implement this.
+        """
+        self._check_user_access(user_context)
+        self._check_credentials_fresh()
+        context = self._get_runtime_context(opts, user_context)
+        return self._get_local_path(path, context)
+
+    def _get_local_path(self, path: str, context: FilesSourceRuntimeContext[TResolvedConfig]) -> str:
+        raise NotImplementedError()
+
+    def usage_percent(self, user_context: "OptionalUserContext" = None) -> float | None:
+        """Return the percent of backing storage used, or None if unknown / not applicable.
+
+        Local sources can report this (e.g. via ``statvfs``); remote sources typically cannot.
+        """
+        return None
+
     def _ensure_writeable(self):
         if not self.get_writable():
             raise Exception("Cannot write to a non-writable file source.")

lib/galaxy/files/sources/_fsspec.py

@@ -198,6 +198,35 @@ def _write_from(
         target_path = self._to_filesystem_path(target_path, context.config)
         fs.put_file(native_path, target_path)
 
+    def _exists(
+        self,
+        path: str,
+        context: FilesSourceRuntimeContext[FsspecResolvedConfigurationType],
+    ) -> bool:
+        cache_options = self._get_cache_options(context.config)
+        fs = self._open_fs(context, cache_options)
+        return fs.exists(self._to_filesystem_path(path, context.config))
+
+    def _size(
+        self,
+        path: str,
+        context: FilesSourceRuntimeContext[FsspecResolvedConfigurationType],
+    ) -> int:
+        cache_options = self._get_cache_options(context.config)
+        fs = self._open_fs(context, cache_options)
+        return int(fs.size(self._to_filesystem_path(path, context.config)) or 0)
+
+    def _remove(
+        self,
+        path: str,
+        context: FilesSourceRuntimeContext[FsspecResolvedConfigurationType],
+        recursive: bool = False,
+    ) -> bool:
+        cache_options = self._get_cache_options(context.config)
+        fs = self._open_fs(context, cache_options)
+        fs.rm(self._to_filesystem_path(path, context.config), recursive=recursive)
+        return True
+
     def _adapt_entry_path(self, filesystem_path: str, config: FsspecResolvedConfigurationType) -> str:
         """Adapt the filesystem path to the desired entry path.
 

lib/galaxy/files/sources/posix.py

@@ -142,6 +142,39 @@ def _write_from(self, target_path: str, native_path: str, context: FilesSourceRu
         shutil.copyfile(native_path, target_native_path_part)
         os.rename(target_native_path_part, target_native_path)
 
+    def usage_percent(self, user_context=None) -> float | None:
+        context = self._get_runtime_context(None, user_context)
+        st = os.statvfs(self._effective_root(context.config))
+        return (float(st.f_blocks - st.f_bavail) / st.f_blocks) * 100
+
+    def _get_local_path(self, path: str, context: FilesSourceRuntimeContext[PosixConfiguration]) -> str:
+        native_path = self._to_native_path(path, context.config)
+        if context.config.enforce_symlink_security:
+            if not safe_contains(self._effective_root(context.config), native_path, allowlist=self._allowlist):
+                raise Exception("Operation not allowed.")
+        if not os.path.exists(native_path):
+            raise FileNotFoundError(native_path)
+        return native_path
+
+    def _exists(self, path: str, context: FilesSourceRuntimeContext[PosixConfiguration]) -> bool:
+        return os.path.exists(self._to_native_path(path, context.config))
+
+    def _size(self, path: str, context: FilesSourceRuntimeContext[PosixConfiguration]) -> int:
+        return os.path.getsize(self._to_native_path(path, context.config))
+
+    def _remove(
+        self, path: str, context: FilesSourceRuntimeContext[PosixConfiguration], recursive: bool = False
+    ) -> bool:
+        native_path = self._to_native_path(path, context.config)
+        if context.config.enforce_symlink_security:
+            if not safe_contains(self._effective_root(context.config), native_path, allowlist=self._allowlist):
+                raise Exception("Operation not allowed.")
+        if recursive:
+            shutil.rmtree(native_path, ignore_errors=True)
+        else:
+            os.remove(native_path)
+        return True
+
     def _to_native_path(self, source_path: str, config: PosixConfiguration):
         source_path = os.path.normpath(source_path)
         if source_path.startswith("/"):

lib/galaxy/objectstore/__init__.py

@@ -1847,6 +1847,10 @@ def type_to_object_store_class(
     objectstore_constructor_kwds: dict[str, Any] = {}
     if store == "disk":
         objectstore_class = DiskObjectStore
+    elif store == "source":
+        from .source_store import SourceObjectStore
+
+        objectstore_class = SourceObjectStore
     elif store == "boto3":
         from .s3_boto3 import S3ObjectStore as Boto3ObjectStore
 

lib/galaxy/objectstore/_caching_base.py

@@ -1,7 +1,6 @@
 import logging
 import os
 import shutil
-from contextlib import contextmanager
 from datetime import datetime
 from typing import (
     Any,
@@ -19,6 +18,7 @@
 from galaxy.util.path import safe_relpath
 from ._util import fix_permissions
 from .caching import (
+    CacheArea,
     CacheTarget,
     InProcessCacheMonitor,
 )
@@ -35,18 +35,16 @@ class CachingConcreteObjectStore(ConcreteObjectStore):
     cache_size: int
     cache_monitor: InProcessCacheMonitor | None = None
     cache_monitor_interval: int
+    _cache_area: CacheArea | None = None
 
-    def _ensure_staging_path_writable(self):
-        staging_path = self.staging_path
-        if not os.path.exists(staging_path):
-            os.makedirs(staging_path, exist_ok=True)
-            if not os.path.exists(staging_path):
-                raise Exception(f"Caching object store created with path '{staging_path}' that does not exist")
+    @property
+    def _cache(self) -> CacheArea:
+        if self._cache_area is None:
+            self._cache_area = CacheArea(self.staging_path, self.cache_size)
+        return self._cache_area
 
-        if not os.access(staging_path, os.R_OK):
-            raise Exception(f"Caching object store created with path '{staging_path}' that does not readable")
-        if not os.access(staging_path, os.W_OK):
-            raise Exception(f"Caching object store created with path '{staging_path}' that does not writable")
+    def _ensure_staging_path_writable(self):
+        self._cache.ensure_writable()
 
     def _construct_path(
         self,
@@ -103,12 +101,11 @@ def _construct_path(
         return rel_path
 
     def _get_cache_path(self, rel_path: str) -> str:
-        return os.path.abspath(os.path.join(self.staging_path, rel_path))
+        return self._cache.path(rel_path)
 
     def _in_cache(self, rel_path: str) -> bool:
         """Check if the given dataset is in the local cache and return True if so."""
-        cache_path = self._get_cache_path(rel_path)
-        return os.path.exists(cache_path)
+        return self._cache.contains(rel_path)
 
     def _pull_into_cache(self, rel_path, **kwargs) -> bool:
         # Ensure the cache directory structure exists (e.g., dataset_#_files/)
@@ -253,7 +250,7 @@ def _empty(self, obj, **kwargs) -> bool:
             raise ObjectNotFound(f"objectstore.empty, object does not exist: {obj}, kwargs: {kwargs}")
 
     def _get_size_in_cache(self, rel_path):
-        return os.path.getsize(self._get_cache_path(rel_path))
+        return self._cache.size(rel_path)
 
     def _size(self, obj, **kwargs) -> int:
         rel_path = self._construct_path(obj, **kwargs)
@@ -375,11 +372,7 @@ def _update_from_file(
 
     @property
     def cache_target(self) -> CacheTarget:
-        return CacheTarget(
-            self.staging_path,
-            self.cache_size,
-            0.9,
-        )
+        return self._cache.target
 
     def _shutdown_cache_monitor(self) -> None:
         self.cache_monitor and self.cache_monitor.shutdown()
@@ -394,7 +387,6 @@ def _get_remote_size(self, rel_path: str) -> int:
     def _exists_remotely(self, rel_path: str) -> bool:
         raise NotImplementedError()
 
-    @contextmanager
     def _atomic_download(self, cache_path):
         """Download to a temp file then atomically rename to prevent serving partial files.
 
@@ -403,18 +395,7 @@ def _atomic_download(self, cache_path):
             with self._atomic_download(local_destination) as tmp_path:
                 do_download(tmp_path)
         """
-        tmp_path = cache_path + ".tmp"
-        try:
-            yield tmp_path
-            os.rename(tmp_path, cache_path)
-        except BaseException:
-            # Catch BaseException (not just Exception) so that KeyboardInterrupt
-            # and SystemExit also trigger cleanup — we re-raise immediately, so
-            # propagation is not blocked. Without this, interrupted downloads
-            # leave .tmp files that poison the cache on next startup.
-            if os.path.exists(tmp_path):
-                os.remove(tmp_path)
-            raise
+        return self._cache.atomic_write(cache_path)
 
     def _download(self, rel_path: str) -> bool:
         raise NotImplementedError()

lib/galaxy/objectstore/_caching_source.py

@@ -0,0 +1,97 @@
+"""A cache wrapper that gives any whole-file FilesSource a stable local path.
+
+Stage 2 of the ObjectStore/FilesSource unification (see
+``doc/source/dev/objectstore_filesource_unification.md``). This wraps a raw FilesSource with
+a :class:`~galaxy.objectstore.caching.CacheArea` -- the *same* cache implementation the
+caching object stores use -- and adds the one primitive a raw remote source lacks:
+``get_local_path``, which resolves a key to a real, seekable local path by pulling it into
+the cache on demand. A local source (posix) needs no wrapper; a remote source (s3, ...) is
+wrapped so an object store built on it can honor ``get_filename``.
+
+Placed in the object store package for now because it reuses ``galaxy.objectstore.caching``;
+a full Stage 2 would relocate the cache primitives to a neutral module so this can live in
+``galaxy.files``. It keeps no runtime dependency on ``galaxy.files`` (the wrapped source is
+imported only under ``TYPE_CHECKING``).
+"""
+
+import logging
+import os
+import shutil
+from typing import TYPE_CHECKING
+
+from .caching import CacheArea
+
+if TYPE_CHECKING:
+    from galaxy.files import OptionalUserContext
+    from galaxy.files.sources import BaseFilesSource
+
+log = logging.getLogger(__name__)
+
+
+class CachingFilesSource:
+    """Wraps a raw FilesSource with a local cache so it can resolve stable local paths."""
+
+    def __init__(
+        self,
+        inner: "BaseFilesSource",
+        staging_path: str,
+        cache_size: int = -1,
+        user_context: "OptionalUserContext" = None,
+    ):
+        self._inner = inner
+        self._cache = CacheArea(staging_path, cache_size)
+        self._cache.ensure_writable()
+        self._user_context = user_context
+
+    def _caching_allowed(self, path: str) -> bool:
+        remote_size = self._inner.size(path, user_context=self._user_context)
+        return self._cache.fits(remote_size or 0)
+
+    # --- FilesSource-shaped interface consumed by the object store ---
+
+    def exists(self, path: str, user_context: "OptionalUserContext" = None) -> bool:
+        return self._cache.contains_nonempty(path) or self._inner.exists(
+            path, user_context=user_context or self._user_context
+        )
+
+    def size(self, path: str, user_context: "OptionalUserContext" = None) -> int:
+        if self._cache.contains_nonempty(path):
+            return self._cache.size(path)
+        return self._inner.size(path, user_context=user_context or self._user_context)
+
+    def get_local_path(self, path: str, user_context: "OptionalUserContext" = None) -> str:
+        cache_path = self._cache.path(path)
+        if self._cache.contains_nonempty(path):
+            return cache_path
+        if not self._caching_allowed(path):
+            raise Exception(f"File {path} is larger than the configured cache allows.")
+        self._cache.makedirs_for(path)
+        with self._cache.atomic_write(cache_path) as tmp:
+            self._inner.realize_to(path, tmp, user_context=user_context or self._user_context)
+        return cache_path
+
+    def realize_to(self, source_path: str, native_path: str, user_context: "OptionalUserContext" = None, opts=None):
+        self._inner.realize_to(source_path, native_path, user_context=user_context or self._user_context, opts=opts)
+
+    def write_from(
+        self, target_path: str, native_path: str, user_context: "OptionalUserContext" = None, opts=None
+    ) -> str:
+        # Push to the backing store (the source of truth), then warm the cache.
+        result = self._inner.write_from(
+            target_path, native_path, user_context=user_context or self._user_context, opts=opts
+        )
+        cache_path = self._cache.path(target_path)
+        self._cache.makedirs_for(target_path)
+        if os.path.abspath(native_path) != cache_path:
+            shutil.copyfile(native_path, cache_path)
+        return result or target_path
+
+    def remove(self, path: str, recursive: bool = False, user_context: "OptionalUserContext" = None) -> bool:
+        if recursive:
+            self._cache.evict_dir(path)
+        else:
+            self._cache.evict(path)
+        return self._inner.remove(path, recursive=recursive, user_context=user_context or self._user_context)
+
+    def usage_percent(self, user_context: "OptionalUserContext" = None) -> float | None:
+        return self._inner.usage_percent(user_context=user_context or self._user_context)