Fix Cloud object store setup for tee streaming

Cloud tee-streaming integration failed in two setup paths before a download could be exercised.

First, fetch metadata preparation synchronously resolved a new job output. At that point the caching store contained only the empty file the job would write and no remote object. Cache synchronization rejected the empty file, failed its remote pull, removed the file, and raised ObjectNotFound, leaving the __DATA_FETCH__ job queued. Resolve the association with sync_cache=False, matching the output path that metadata setup already receives. This only selects the job's destination path; normal read-time validation of zero-byte datasets is unchanged.

Once fetch jobs could finish, the tests exposed a second configuration error: the Cloud object store discarded the MinIO endpoint and CloudBridge sent requests to AWS, where the test credentials produced InvalidAccessKeyId and dataset display returned 404. Preserve endpoint_url through Cloud configuration serialization and map it to CloudBridge's s3_endpoint_url setting.

Cover metadata setup with a real caching-base fake containing an empty local output and no remote object, scoped to one dataset instance. Also verify XML and YAML endpoint parsing, serialization, and the configuration passed across the optional CloudBridge factory boundary.

lib/galaxy/metadata/__init__.py

@@ -197,7 +197,7 @@ class PortableDirectoryMetadataGenerator(MetadataCollectionStrategy):
             )
 
             outputs[name] = {
-                "filename_override": _get_filename_override(output_fnames, dataset.get_file_name()),
+                "filename_override": _get_filename_override(output_fnames, dataset.get_file_name(sync_cache=False)),
                 "validate": validate_outputs,
                 "object_store_store_by": dataset.dataset.store_by,
                 "id": dataset.id,

lib/galaxy/objectstore/cloud.py

@@ -55,6 +55,7 @@ class Cloud(CachingConcreteObjectStore, UsesAxel):
 
         self.provider = config_dict["provider"]
         self.credentials = config_dict["auth"]
+        self.endpoint_url = (config_dict.get("connection") or {}).get("endpoint_url")
         self.bucket_name = bucket_dict.get("name")
         self.use_rr = bucket_dict.get("use_reduced_redundancy", False)
         self.max_chunk_size = bucket_dict.get("max_chunk_size", 250)
@@ -68,19 +69,21 @@ class Cloud(CachingConcreteObjectStore, UsesAxel):
         if CloudProviderFactory is None:
             raise Exception(NO_CLOUDBRIDGE_ERROR_MESSAGE)
 
-        self.conn = self._get_connection(self.provider, self.credentials)
+        self.conn = self._get_connection(self.provider, self.credentials, self.endpoint_url)
         self.bucket = self._get_bucket(self.bucket_name)
         self._ensure_staging_path_writable()
         self._start_cache_monitor_if_needed()
         self._init_axel()
 
     @staticmethod
-    def _get_connection(provider, credentials):
+    def _get_connection(provider, credentials, endpoint_url=None):
         log.debug(f"Configuring `{provider}` Connection")
         if provider == "aws":
             config = {"aws_access_key": credentials["access_key"], "aws_secret_key": credentials["secret_key"]}
             if "region" in credentials:
                 config["aws_region_name"] = credentials["region"]
+            if endpoint_url:
+                config["s3_endpoint_url"] = endpoint_url
             connection = CloudProviderFactory().create_provider(ProviderList.AWS, config)
         elif provider == "azure":
             config = {
@@ -141,6 +144,9 @@ class Cloud(CachingConcreteObjectStore, UsesAxel):
                 raise Exception(msg)
             provider = provider.lower()
             config["provider"] = provider
+            connection_element = config_xml.find("connection")
+            if connection_element is not None:
+                config["connection"]["endpoint_url"] = connection_element.get("endpoint_url")
 
             # Read any provider-specific configuration.
             auth_element = config_xml.findall("auth")[0]
@@ -195,7 +201,7 @@ class Cloud(CachingConcreteObjectStore, UsesAxel):
         return as_dict
 
     def _config_to_dict(self):
-        return {
+        config = {
             "provider": self.provider,
             "auth": self.credentials,
             "bucket": {
@@ -204,6 +210,9 @@ class Cloud(CachingConcreteObjectStore, UsesAxel):
             },
             "cache": self._cache_config_to_dict(),
         }
+        if self.endpoint_url:
+            config["connection"] = {"endpoint_url": self.endpoint_url}
+        return config
 
     def _get_bucket(self, bucket_name):
         try:

test/unit/app/tools/test_metadata.py

@@ -1,11 +1,18 @@
+import json
 import os
 import subprocess
+from types import SimpleNamespace
 
 from galaxy import model
 from galaxy.app_unittest_utils import tools_support
 from galaxy.job_execution.datasets import DatasetPath
 from galaxy.metadata import get_metadata_compute_strategy
 from galaxy.objectstore import ObjectStorePopulator
+from galaxy.objectstore._caching_base import CachingConcreteObjectStore
+from galaxy.objectstore.caching import (
+    CacheShard,
+    CacheShardManager,
+)
 from galaxy.util import (
     galaxy_directory,
     safe_makedirs,
@@ -13,6 +20,36 @@ from galaxy.util import (
 from galaxy.util.unittest import TestCase
 
 
+class MissingRemoteCachingObjectStore(CachingConcreteObjectStore):
+    def __init__(self, cache_path):
+        self._cache_shards = CacheShardManager([CacheShard(path=cache_path, weight=1, size=-1)])
+        self.config = SimpleNamespace(umask=0o002, gid=-1)
+        self.cache_updated_data = True
+        self.store_by = "id"
+        self.extra_dirs = {}
+
+    def _get_remote_size(self, rel_path):
+        return 0
+
+    def _exists_remotely(self, rel_path):
+        return False
+
+    def _download(self, rel_path, *, cache_path, cache_target):
+        return False
+
+    def _push_string_to_path(self, rel_path, from_string):
+        return True
+
+    def _push_file_to_path(self, rel_path, source_file):
+        return True
+
+    def _delete_existing_remote(self, rel_path):
+        return True
+
+    def _delete_remote_all(self, rel_path):
+        return True
+
+
 class TestMetadata(TestCase, tools_support.UsesTools):
     def setUp(self):
         super().setUp()
@@ -43,6 +80,24 @@ class TestMetadata(TestCase, tools_support.UsesTools):
         self.app.config.metadata_strategy = "extended"
         self._test_simple_output()
 
+    def test_setup_does_not_sync_empty_job_output(self):
+        self.app.config.metadata_strategy = "directory"
+        source_file_name = os.path.join(galaxy_directory(), "test/functional/tools/for_workflows/cat.xml")
+        self._init_tool_for_path(source_file_name)
+        output_dataset = self._create_output_dataset(extension="txt")
+        object_store = MissingRemoteCachingObjectStore(os.path.join(self.test_directory, "remote_cache"))
+        object_store.create(output_dataset.dataset)
+        output_path = object_store.get_filename(output_dataset.dataset, sync_cache=False)
+
+        output_dataset.dataset.object_store = object_store
+        self.metadata_command({"out_file1": output_dataset})
+
+        assert os.path.exists(output_path)
+        assert os.path.getsize(output_path) == 0
+        with open(os.path.join(self.job_working_directory, "metadata", "params.json")) as f:
+            metadata_params = json.load(f)
+        assert metadata_params["outputs"]["out_file1"]["filename_override"] == output_path
+
     def _test_simple_output(self):
         source_file_name = os.path.join(galaxy_directory(), "test/functional/tools/for_workflows/cat.xml")
         self._init_tool_for_path(source_file_name)
@@ -211,7 +266,9 @@ class TestMetadata(TestCase, tools_support.UsesTools):
         safe_makedirs(os.path.join(self.job_working_directory, "metadata"))
         self.app.datatypes_registry.to_xml_file(path=datatypes_config)
         job_metadata = os.path.join(self.tool_working_directory, self.tool.provided_metadata_file)
-        output_fnames = [DatasetPath(o.dataset.id, o.dataset.get_file_name(), None) for o in output_datasets.values()]
+        output_fnames = [
+            DatasetPath(o.dataset.id, o.dataset.get_file_name(sync_cache=False), None) for o in output_datasets.values()
+        ]
         command = metadata_compute_strategy.setup_external_metadata(
             output_datasets,
             output_collections,

test/unit/objectstore/test_objectstore.py

@@ -1286,6 +1286,13 @@ def test_config_parse_boto3_separated_transfer_options():
 
 CLOUD_AWS_TEST_CONFIG = get_example("cloud_aws_simple.xml")
 CLOUD_AWS_TEST_CONFIG_YAML = get_example("cloud_aws_simple.yml")
+CLOUD_AWS_CUSTOM_ENDPOINT = "http://127.0.0.1:9000"
+CLOUD_AWS_CUSTOM_ENDPOINT_TEST_CONFIG = CLOUD_AWS_TEST_CONFIG.replace(
+    "<bucket ", f'<connection endpoint_url="{CLOUD_AWS_CUSTOM_ENDPOINT}" />\n     <bucket ', 1
+)
+CLOUD_AWS_CUSTOM_ENDPOINT_TEST_CONFIG_YAML = CLOUD_AWS_TEST_CONFIG_YAML.replace(
+    "\nbucket:\n", f"\nconnection:\n  endpoint_url: {CLOUD_AWS_CUSTOM_ENDPOINT}\n\nbucket:\n", 1
+)
 
 CLOUD_AZURE_TEST_CONFIG = get_example("cloud_azure_simple.xml")
 CLOUD_AZURE_TEST_CONFIG_YAML = get_example("cloud_azure_simple.yml")
@@ -1352,6 +1359,32 @@ def test_config_parse_cloud():
             assert len(extra_dirs) == 2
 
 
+@patch_object_stores_to_skip_initialize
+def test_config_parse_cloud_aws_custom_endpoint():
+    for config_str in [CLOUD_AWS_CUSTOM_ENDPOINT_TEST_CONFIG, CLOUD_AWS_CUSTOM_ENDPOINT_TEST_CONFIG_YAML]:
+        with TestConfig(config_str) as (_, object_store):
+            assert object_store.endpoint_url == CLOUD_AWS_CUSTOM_ENDPOINT
+            assert object_store.to_dict()["connection"]["endpoint_url"] == CLOUD_AWS_CUSTOM_ENDPOINT
+
+            with (
+                patch("galaxy.objectstore.cloud.CloudProviderFactory") as provider_factory,
+                patch("galaxy.objectstore.cloud.ProviderList") as providers,
+            ):
+                connection = object_store._get_connection(
+                    object_store.provider, object_store.credentials, object_store.endpoint_url
+                )
+
+            assert connection is provider_factory.return_value.create_provider.return_value
+            provider_factory.return_value.create_provider.assert_called_once_with(
+                providers.AWS,
+                {
+                    "aws_access_key": "access_moo",
+                    "aws_secret_key": "secret_cow",
+                    "s3_endpoint_url": CLOUD_AWS_CUSTOM_ENDPOINT,
+                },
+            )
+
+
 CLOUD_AWS_NO_AUTH_TEST_CONFIG = get_example("cloud_aws_no_auth.xml")