diff --git a/lib/galaxy/config/sample/object_store_conf.xml.sample b/lib/galaxy/config/sample/object_store_conf.xml.sample
index b99028bc4f7a..2d726077a164 100644
--- a/lib/galaxy/config/sample/object_store_conf.xml.sample
+++ b/lib/galaxy/config/sample/object_store_conf.xml.sample
@@ -7,6 +7,19 @@
     backends to the distributed and hierarchical object stores (including
     distributed and hierarchical themselves).
 -->
+<!--
+    Most of the Object Stores have <cache> option like:
+    <cache path="database/object_store_cache" size="1000" cache_updated_data="True" />
+
+    Here
+    "path" - local path to store cached data,
+    "size" - size of the cache in gigabytes.
+    "cache_updated_data" - optional parameter that allows to control data
+       is being sent directly
+       to an object store without storing it in the cache.
+       By default data is also copied to the cache (cache_updated_data="True").
+-->
+
 
 <!--
     Sample Disk Object Store
@@ -135,7 +148,7 @@
 <object_store type="aws_s3">
      <auth access_key="...." secret_key="....." />
      <bucket name="unique_bucket_name_all_lowercase" use_reduced_redundancy="False" />
-     <cache path="database/object_store_cache" size="1000" />
+     <cache path="database/object_store_cache" size="1000" cache_updated_data="True" />
      <extra_dir type="job_work" path="database/job_working_directory_s3"/>
      <extra_dir type="temp" path="database/tmp_s3"/>
 </object_store>
@@ -150,7 +163,7 @@
     <resource name="demoResc" />
     <zone name="tempZone" />
     <connection host="localhost" port="1247" timeout="30" refresh_time="300" connection_pool_monitor_interval="3600"/>
-    <cache path="database/object_store_cache_irods" size="1000" />
+    <cache path="database/object_store_cache_irods" size="1000" cache_updated_data="True" />
     <extra_dir type="job_work" path="database/job_working_directory_irods"/>
     <extra_dir type="temp" path="database/tmp_irods"/>
 </object_store>
@@ -166,7 +179,7 @@
     <auth access_key="...." secret_key="....." />
     <bucket name="unique_bucket_name" use_reduced_redundancy="False" max_chunk_size="250"/>
     <connection host="" port="" is_secure="" conn_path="" multipart="True"/>
-    <cache path="database/object_store_cache" size="1000" />
+    <cache path="database/object_store_cache" size="1000" cache_updated_data="True" />
     <extra_dir type="job_work" path="database/job_working_directory_swift"/>
     <extra_dir type="temp" path="database/tmp_swift"/>
 </object_store>
@@ -181,7 +194,7 @@
 <object_store type="azure_blob">
     <auth account_name="..." account_key="...." />
     <container name="unique_container_name" max_chunk_size="250"/>
-    <cache path="database/object_store_cache" size="100" />
+    <cache path="database/object_store_cache" size="100" cache_updated_data="True" />
     <extra_dir type="job_work" path="database/job_working_directory_azure"/>
     <extra_dir type="temp" path="database/tmp_azure"/>
 </object_store>
@@ -196,7 +209,7 @@
 <object_store type="cloud" provider="aws" order="0">
     <auth access_key="..." secret_key="..." />
     <bucket name="..." use_reduced_redundancy="False" />
-    <cache path="database/object_store_cache" size="100" />
+    <cache path="database/object_store_cache" size="100" cache_updated_data="True" />
     <extra_dir type="job_work" path="database/job_working_directory_s3"/>
     <extra_dir type="temp" path="database/tmp_s3"/>
 </object_store>
@@ -211,7 +224,7 @@
 <object_store type="cloud" provider="azure" order="0">
     <auth subscription_id="..." client_id="..." secret="..." tenant="..." />
     <bucket name="..." use_reduced_redundancy="False" />
-    <cache path="database/object_store_cache" size="100" />
+    <cache path="database/object_store_cache" size="100" cache_updated_data="True" />
     <extra_dir type="job_work" path="database/job_working_directory_azure"/>
     <extra_dir type="temp" path="database/tmp_azure"/>
 </object_store>
@@ -226,7 +239,7 @@
 <object_store type="cloud" provider="google" order="0">
     <auth credentials_file="..." />
     <bucket name="..." use_reduced_redundancy="False" />
-    <cache path="database/object_store_cache" size="1000" />
+    <cache path="database/object_store_cache" size="1000" cache_updated_data="True" />
     <extra_dir type="job_work" path="database/job_working_directory_gcp"/>
     <extra_dir type="temp" path="database/tmp_gcp"/>
 </object_store>
@@ -319,7 +332,7 @@
             <resource name="demoResc" />
             <zone name="tempZone" />
             <connection host="localhost" port="1247" timeout="30" refresh_time="300" connection_pool_monitor_interval="3600"/>
-            <cache path="database/object_store_cache_irods" size="1000" />
+            <cache path="database/object_store_cache_irods" size="1000" cache_updated_data="True" />
             <badges>
                 <less_stable />
                 <backed_up>This data is backed up using iRODs native hierarchal storage management mechanisms. The rules describing how data is stored and backed up in iRODS can be found in our institutional [iRODS documentation](https://irods.org/uploads/2018/Saum-SURFsara-Data_Archiving_in_iRODS-slides.pdf)</backed_up>
diff --git a/lib/galaxy/datatypes/protocols.py b/lib/galaxy/datatypes/protocols.py
index 93b2875a9795..150866a8adae 100644
--- a/lib/galaxy/datatypes/protocols.py
+++ b/lib/galaxy/datatypes/protocols.py
@@ -30,7 +30,7 @@ def extra_files_path(self):
 
 
 class HasFileName(Protocol):
-    def get_file_name(self) -> str:
+    def get_file_name(self, sync_cache=True) -> str:
         ...
 
 
diff --git a/lib/galaxy/job_execution/setup.py b/lib/galaxy/job_execution/setup.py
index de321944418e..1cff9c37c532 100644
--- a/lib/galaxy/job_execution/setup.py
+++ b/lib/galaxy/job_execution/setup.py
@@ -276,7 +276,10 @@ def compute_outputs(self) -> None:
             da_false_path = dataset_path_rewriter.rewrite_dataset_path(da.dataset, "output")
             mutable = da.dataset.dataset.external_filename is None
             dataset_path = DatasetPath(
-                da.dataset.dataset.id, da.dataset.get_file_name(), false_path=da_false_path, mutable=mutable
+                da.dataset.dataset.id,
+                da.dataset.get_file_name(sync_cache=False),
+                false_path=da_false_path,
+                mutable=mutable,
             )
             job_outputs.append(JobOutput(da.name, da.dataset, dataset_path))
 
diff --git a/lib/galaxy/managers/datasets.py b/lib/galaxy/managers/datasets.py
index 6734b0d2fd10..c639316e9f5e 100644
--- a/lib/galaxy/managers/datasets.py
+++ b/lib/galaxy/managers/datasets.py
@@ -273,7 +273,7 @@ def serialize_file_name(self, item, key, user=None, **context):
         # expensive: allow config option due to cost of operation
         if is_admin or self.app.config.expose_dataset_path:
             if not dataset.purged:
-                return dataset.get_file_name()
+                return dataset.get_file_name(sync_cache=False)
         self.skip()
 
     def serialize_extra_files_path(self, item, key, user=None, **context):
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 909b1d57f332..4cdf328771a3 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -2536,8 +2536,8 @@ def __init__(self, dataset=None):
         self.dataset = dataset
         self.metadata = dict()
 
-    def get_file_name(self):
-        return self.dataset.get_file_name()
+    def get_file_name(self, sync_cache=True):
+        return self.dataset.get_file_name(sync_cache)
 
     def __eq__(self, other):
         return isinstance(other, FakeDatasetAssociation) and self.dataset == other.dataset
@@ -3960,14 +3960,14 @@ def ensure_shareable(self):
         if not self.shareable:
             raise Exception(CANNOT_SHARE_PRIVATE_DATASET_MESSAGE)
 
-    def get_file_name(self):
+    def get_file_name(self, sync_cache=True):
         if self.purged:
             log.warning(f"Attempt to get file name of purged dataset {self.id}")
             return ""
         if not self.external_filename:
             object_store = self._assert_object_store_set()
             if object_store.exists(self):
-                file_name = object_store.get_filename(self)
+                file_name = object_store.get_filename(self, sync_cache=sync_cache)
             else:
                 file_name = ""
             if not file_name and self.state not in (self.states.NEW, self.states.QUEUED):
@@ -4421,10 +4421,10 @@ def set_dataset_state(self, state):
 
     state = property(get_dataset_state, set_dataset_state)
 
-    def get_file_name(self) -> str:
+    def get_file_name(self, sync_cache=True) -> str:
         if self.dataset.purged:
             return ""
-        return self.dataset.get_file_name()
+        return self.dataset.get_file_name(sync_cache=sync_cache)
 
     def set_file_name(self, filename: str):
         return self.dataset.set_file_name(filename)
@@ -9265,7 +9265,7 @@ def update_from_file(self, file_name):
             alt_name=os.path.basename(self.get_file_name()),
         )
 
-    def get_file_name(self):
+    def get_file_name(self, sync_cache=True):
         # Ensure the directory structure and the metadata file object exist
         try:
             da = self.history_dataset or self.library_dataset
@@ -9280,7 +9280,7 @@ def get_file_name(self):
             if not object_store.exists(self, extra_dir="_metadata_files", extra_dir_at_root=True, alt_name=alt_name):
                 object_store.create(self, extra_dir="_metadata_files", extra_dir_at_root=True, alt_name=alt_name)
             path = object_store.get_filename(
-                self, extra_dir="_metadata_files", extra_dir_at_root=True, alt_name=alt_name
+                self, extra_dir="_metadata_files", extra_dir_at_root=True, alt_name=alt_name, sync_cache=sync_cache
             )
             return path
         except AttributeError:
diff --git a/lib/galaxy/model/none_like.py b/lib/galaxy/model/none_like.py
index d24a70329cda..80e96c16b7ea 100644
--- a/lib/galaxy/model/none_like.py
+++ b/lib/galaxy/model/none_like.py
@@ -38,5 +38,5 @@ def __getattr__(self, name):
     def missing_meta(self):
         return False
 
-    def get_file_name(self):
+    def get_file_name(self, sync_cache=True):
         return "None"
diff --git a/lib/galaxy/objectstore/azure_blob.py b/lib/galaxy/objectstore/azure_blob.py
index b72e6b9dbfd6..83c1d700e195 100644
--- a/lib/galaxy/objectstore/azure_blob.py
+++ b/lib/galaxy/objectstore/azure_blob.py
@@ -109,6 +109,7 @@ def __init__(self, config, config_dict):
 
         self.cache_size = cache_dict.get("size") or self.config.object_store_cache_size
         self.staging_path = cache_dict.get("path") or self.config.object_store_cache_path
+        self.cache_updated_data = cache_dict.get("cache_updated_data", True)
 
         self._initialize()
 
@@ -136,6 +137,7 @@ def to_dict(self):
                 "cache": {
                     "size": self.cache_size,
                     "path": self.staging_path,
+                    "cache_updated_data": self.cache_updated_data,
                 },
             }
         )
@@ -501,12 +503,15 @@ def _get_filename(self, obj, **kwargs):
         base_dir = kwargs.get("base_dir", None)
         dir_only = kwargs.get("dir_only", False)
         obj_dir = kwargs.get("obj_dir", False)
+        sync_cache = kwargs.get("sync_cache", True)
 
         # for JOB_WORK directory
         if base_dir and dir_only and obj_dir:
             return os.path.abspath(rel_path)
 
         cache_path = self._get_cache_path(rel_path)
+        if not sync_cache:
+            return cache_path
         # S3 does not recognize directories as files so cannot check if those exist.
         # So, if checking dir only, ensure given dir exists in cache and return
         # the expected cache path.
@@ -515,8 +520,8 @@ def _get_filename(self, obj, **kwargs):
         #     if not os.path.exists(cache_path):
         #         os.makedirs(cache_path)
         #     return cache_path
-        # Check if the file exists in the cache first
-        if self._in_cache(rel_path):
+        # Check if the file exists in the cache first, always pull if file size in cache is zero
+        if self._in_cache(rel_path) and (dir_only or os.path.getsize(self._get_cache_path(rel_path)) > 0):
             return cache_path
         # Check if the file exists in persistent storage and, if it does, pull it into cache
         elif self._exists(obj, **kwargs):
@@ -542,7 +547,7 @@ def _update_from_file(self, obj, file_name=None, create=False, **kwargs):
                 # Copy into cache
                 cache_file = self._get_cache_path(rel_path)
                 try:
-                    if source_file != cache_file:
+                    if source_file != cache_file and self.cache_updated_data:
                         # FIXME? Should this be a `move`?
                         shutil.copy2(source_file, cache_file)
                     self._fix_permissions(cache_file)
diff --git a/lib/galaxy/objectstore/caching.py b/lib/galaxy/objectstore/caching.py
index 2097da2c680e..2df468681498 100644
--- a/lib/galaxy/objectstore/caching.py
+++ b/lib/galaxy/objectstore/caching.py
@@ -12,7 +12,10 @@
 
 from typing_extensions import NamedTuple
 
-from galaxy.util import nice_size
+from galaxy.util import (
+    nice_size,
+    string_as_bool,
+)
 from galaxy.util.sleeper import Sleeper
 
 log = logging.getLogger(__name__)
@@ -124,11 +127,13 @@ def parse_caching_config_dict_from_xml(config_xml):
         cache_size = float(c_xml.get("size", -1))
         staging_path = c_xml.get("path", None)
         monitor = c_xml.get("monitor", "auto")
+        cache_updated_data = string_as_bool(c_xml.get("cache_updated_data", "True"))
 
         cache_dict = {
             "size": cache_size,
             "path": staging_path,
             "monitor": monitor,
+            "cache_updated_data": cache_updated_data,
         }
     else:
         cache_dict = {}
diff --git a/lib/galaxy/objectstore/cloud.py b/lib/galaxy/objectstore/cloud.py
index 4a995d833f66..625b1f9547e7 100644
--- a/lib/galaxy/objectstore/cloud.py
+++ b/lib/galaxy/objectstore/cloud.py
@@ -66,6 +66,7 @@ def _config_to_dict(self):
             "cache": {
                 "size": self.cache_size,
                 "path": self.staging_path,
+                "cache_updated_data": self.cache_updated_data,
             },
         }
 
@@ -103,6 +104,7 @@ def __init__(self, config, config_dict):
 
         self.cache_size = cache_dict.get("size") or self.config.object_store_cache_size
         self.staging_path = cache_dict.get("path") or self.config.object_store_cache_path
+        self.cache_updated_data = cache_dict.get("cache_updated_data", True)
 
         self._initialize()
 
@@ -621,12 +623,15 @@ def _get_filename(self, obj, **kwargs):
         dir_only = kwargs.get("dir_only", False)
         obj_dir = kwargs.get("obj_dir", False)
         rel_path = self._construct_path(obj, **kwargs)
+        sync_cache = kwargs.get("sync_cache", True)
 
         # for JOB_WORK directory
         if base_dir and dir_only and obj_dir:
             return os.path.abspath(rel_path)
 
         cache_path = self._get_cache_path(rel_path)
+        if not sync_cache:
+            return cache_path
         # S3 does not recognize directories as files so cannot check if those exist.
         # So, if checking dir only, ensure given dir exists in cache and return
         # the expected cache path.
@@ -635,8 +640,8 @@ def _get_filename(self, obj, **kwargs):
         #     if not os.path.exists(cache_path):
         #         os.makedirs(cache_path)
         #     return cache_path
-        # Check if the file exists in the cache first
-        if self._in_cache(rel_path):
+        # Check if the file exists in the cache first, always pull if file size in cache is zero
+        if self._in_cache(rel_path) and (dir_only or os.path.getsize(self._get_cache_path(rel_path)) > 0):
             return cache_path
         # Check if the file exists in persistent storage and, if it does, pull it into cache
         elif self._exists(obj, **kwargs):
@@ -663,7 +668,7 @@ def _update_from_file(self, obj, file_name=None, create=False, **kwargs):
                 # Copy into cache
                 cache_file = self._get_cache_path(rel_path)
                 try:
-                    if source_file != cache_file:
+                    if source_file != cache_file and self.cache_updated_data:
                         # FIXME? Should this be a `move`?
                         shutil.copy2(source_file, cache_file)
                     self._fix_permissions(cache_file)
diff --git a/lib/galaxy/objectstore/irods.py b/lib/galaxy/objectstore/irods.py
index 63b81e7a1a9d..d2ac18fdca94 100644
--- a/lib/galaxy/objectstore/irods.py
+++ b/lib/galaxy/objectstore/irods.py
@@ -26,6 +26,7 @@
 from galaxy.util import (
     directory_hash_id,
     ExecutionTimer,
+    string_as_bool,
     umask_fix_perms,
     unlink,
 )
@@ -81,6 +82,7 @@ def parse_config_xml(config_xml):
             _config_xml_error("cache")
         cache_size = float(c_xml[0].get("size", -1))
         staging_path = c_xml[0].get("path", None)
+        cache_updated_data = string_as_bool(c_xml[0].get("cache_updated_data", "True"))
 
         attrs = ("type", "path")
         e_xml = config_xml.findall("extra_dir")
@@ -109,6 +111,7 @@ def parse_config_xml(config_xml):
             "cache": {
                 "size": cache_size,
                 "path": staging_path,
+                "cache_updated_data": cache_updated_data,
             },
             "extra_dirs": extra_dirs,
             "private": DiskObjectStore.parse_private_from_config_xml(config_xml),
@@ -142,6 +145,7 @@ def _config_to_dict(self):
             "cache": {
                 "size": self.cache_size,
                 "path": self.staging_path,
+                "cache_updated_data": self.cache_updated_data,
             },
         }
 
@@ -209,6 +213,7 @@ def __init__(self, config, config_dict):
         self.staging_path = cache_dict.get("path") or self.config.object_store_cache_path
         if self.staging_path is None:
             _config_dict_error("cache->path")
+        self.cache_updated_data = cache_dict.get("cache_updated_data", True)
 
         extra_dirs = {e["type"]: e["path"] for e in config_dict.get("extra_dirs", [])}
         if not extra_dirs:
@@ -440,7 +445,10 @@ def _download(self, rel_path):
 
         collection_path = f"{self.home}/{subcollection_name}"
         data_object_path = f"{collection_path}/{data_object_name}"
-        options = {kw.DEST_RESC_NAME_KW: self.resource}
+        # we need to allow irods to override already existing zero-size output files created
+        # in object store cache during job setup (see also https://github.com/galaxyproject/galaxy/pull/17025#discussion_r1394517033)
+        # TODO: get rid of this flag when Galaxy stops pre-creating those output files in cache
+        options = {kw.FORCE_FLAG_KW: "", kw.DEST_RESC_NAME_KW: self.resource}
 
         try:
             cache_path = self._get_cache_path(rel_path)
@@ -716,6 +724,7 @@ def _get_filename(self, obj, **kwargs):
         dir_only = kwargs.get("dir_only", False)
         obj_dir = kwargs.get("obj_dir", False)
         rel_path = self._construct_path(obj, **kwargs)
+        sync_cache = kwargs.get("sync_cache", True)
 
         # for JOB_WORK directory
         if base_dir and dir_only and obj_dir:
@@ -723,6 +732,8 @@ def _get_filename(self, obj, **kwargs):
             return os.path.abspath(rel_path)
 
         cache_path = self._get_cache_path(rel_path)
+        if not sync_cache:
+            return cache_path
         # iRODS does not recognize directories as files so cannot check if those exist.
         # So, if checking dir only, ensure given dir exists in cache and return
         # the expected cache path.
@@ -731,8 +742,8 @@ def _get_filename(self, obj, **kwargs):
         #     if not os.path.exists(cache_path):
         #         os.makedirs(cache_path)
         #     return cache_path
-        # Check if the file exists in the cache first
-        if self._in_cache(rel_path):
+        # Check if the file exists in the cache first, always pull if file size in cache is zero
+        if self._in_cache(rel_path) and (dir_only or os.path.getsize(self._get_cache_path(rel_path)) > 0):
             log.debug("irods_pt _get_filename: %s", ipt_timer)
             return cache_path
         # Check if the file exists in persistent storage and, if it does, pull it into cache
@@ -764,7 +775,7 @@ def _update_from_file(self, obj, file_name=None, create=False, **kwargs):
                 # Copy into cache
                 cache_file = self._get_cache_path(rel_path)
                 try:
-                    if source_file != cache_file:
+                    if source_file != cache_file and self.cache_updated_data:
                         # FIXME? Should this be a `move`?
                         shutil.copy2(source_file, cache_file)
                     self._fix_permissions(cache_file)
diff --git a/lib/galaxy/objectstore/s3.py b/lib/galaxy/objectstore/s3.py
index e0656864a939..41b4d58e96a9 100644
--- a/lib/galaxy/objectstore/s3.py
+++ b/lib/galaxy/objectstore/s3.py
@@ -145,6 +145,7 @@ def _config_to_dict(self):
             "cache": {
                 "size": self.cache_size,
                 "path": self.staging_path,
+                "cache_updated_data": self.cache_updated_data,
             },
         }
 
@@ -186,6 +187,7 @@ def __init__(self, config, config_dict):
 
         self.cache_size = cache_dict.get("size") or self.config.object_store_cache_size
         self.staging_path = cache_dict.get("path") or self.config.object_store_cache_path
+        self.cache_updated_data = cache_dict.get("cache_updated_data", True)
 
         extra_dirs = {e["type"]: e["path"] for e in config_dict.get("extra_dirs", [])}
         self.extra_dirs.update(extra_dirs)
@@ -634,7 +636,7 @@ def _delete(self, obj, entire_dir=False, **kwargs):
     def _get_data(self, obj, start=0, count=-1, **kwargs):
         rel_path = self._construct_path(obj, **kwargs)
         # Check cache first and get file if not there
-        if not self._in_cache(rel_path):
+        if not self._in_cache(rel_path) or os.path.getsize(self._get_cache_path(rel_path)) == 0:
             self._pull_into_cache(rel_path)
         # Read the file content from cache
         data_file = open(self._get_cache_path(rel_path))
@@ -647,6 +649,8 @@ def _get_filename(self, obj, **kwargs):
         base_dir = kwargs.get("base_dir", None)
         dir_only = kwargs.get("dir_only", False)
         obj_dir = kwargs.get("obj_dir", False)
+        sync_cache = kwargs.get("sync_cache", True)
+
         rel_path = self._construct_path(obj, **kwargs)
 
         # for JOB_WORK directory
@@ -654,6 +658,8 @@ def _get_filename(self, obj, **kwargs):
             return os.path.abspath(rel_path)
 
         cache_path = self._get_cache_path(rel_path)
+        if not sync_cache:
+            return cache_path
         # S3 does not recognize directories as files so cannot check if those exist.
         # So, if checking dir only, ensure given dir exists in cache and return
         # the expected cache path.
@@ -662,8 +668,8 @@ def _get_filename(self, obj, **kwargs):
         #     if not os.path.exists(cache_path):
         #         os.makedirs(cache_path)
         #     return cache_path
-        # Check if the file exists in the cache first
-        if self._in_cache(rel_path):
+        # Check if the file exists in the cache first, always pull if file size in cache is zero
+        if self._in_cache(rel_path) and (dir_only or os.path.getsize(self._get_cache_path(rel_path)) > 0):
             return cache_path
         # Check if the file exists in persistent storage and, if it does, pull it into cache
         elif self._exists(obj, **kwargs):
@@ -691,7 +697,7 @@ def _update_from_file(self, obj, file_name=None, create=False, **kwargs):
                 # Copy into cache
                 cache_file = self._get_cache_path(rel_path)
                 try:
-                    if source_file != cache_file:
+                    if source_file != cache_file and self.cache_updated_data:
                         # FIXME? Should this be a `move`?
                         shutil.copy2(source_file, cache_file)
                     self._fix_permissions(cache_file)
diff --git a/lib/galaxy/tool_util/data/__init__.py b/lib/galaxy/tool_util/data/__init__.py
index 7846188320d1..849be7bb90e6 100644
--- a/lib/galaxy/tool_util/data/__init__.py
+++ b/lib/galaxy/tool_util/data/__init__.py
@@ -891,7 +891,7 @@ def extra_files_path_exists(self) -> bool:
 class OutputDataset(HasExtraFiles, Protocol):
     ext: str
 
-    def get_file_name(self) -> str:
+    def get_file_name(self, sync_cache=True) -> str:
         ...
 
 
diff --git a/test/integration/objectstore/_base.py b/test/integration/objectstore/_base.py
index 272aa0fa1348..95796bd6f0b1 100644
--- a/test/integration/objectstore/_base.py
+++ b/test/integration/objectstore/_base.py
@@ -17,7 +17,7 @@
             <auth access_key="${access_key}" secret_key="${secret_key}" />
             <bucket name="galaxy" use_reduced_redundancy="False" max_chunk_size="250"/>
             <connection host="${host}" port="${port}" is_secure="False" conn_path="" multipart="True"/>
-            <cache path="${temp_directory}/object_store_cache" size="1000" />
+            <cache path="${temp_directory}/object_store_cache" size="1000" cache_updated_data="${cache_updated_data}" />
             <extra_dir type="job_work" path="${temp_directory}/job_working_directory_swift"/>
             <extra_dir type="temp" path="${temp_directory}/tmp_swift"/>
         </object_store>
@@ -102,6 +102,7 @@ def handle_galaxy_config_kwds(cls, config):
                         "port": OBJECT_STORE_PORT,
                         "access_key": OBJECT_STORE_ACCESS_KEY,
                         "secret_key": OBJECT_STORE_SECRET_KEY,
+                        "cache_updated_data": cls.updateCacheData(),
                     }
                 )
             )
@@ -110,3 +111,7 @@ def handle_galaxy_config_kwds(cls, config):
     def setUp(self):
         super().setUp()
         self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def updateCacheData(cls):
+        return True
diff --git a/test/integration/objectstore/test_remote_objectstore_cache_operations.py b/test/integration/objectstore/test_remote_objectstore_cache_operations.py
index 5ac29c6d9ff5..cf3c9391a1ac 100644
--- a/test/integration/objectstore/test_remote_objectstore_cache_operations.py
+++ b/test/integration/objectstore/test_remote_objectstore_cache_operations.py
@@ -3,6 +3,7 @@
 
 import pytest
 
+from galaxy_test.base.populators import skip_without_tool
 from ._base import (
     BaseSwiftObjectStoreIntegrationTestCase,
     files_count,
@@ -78,3 +79,42 @@ def test_upload_updates_cache(self):
         assert files_count(self.object_store_cache_path) == 1
         only_file = next(iter(get_files(self.object_store_cache_path)))
         assert os.path.getsize(only_file) == 4
+
+
+class TestCacheOperationWithNoCacheUpdate(BaseSwiftObjectStoreIntegrationTestCase):
+    @classmethod
+    def updateCacheData(cls):
+        return False
+
+    def tearDown(self):
+        shutil.rmtree(self.object_store_cache_path)
+        os.mkdir(self.object_store_cache_path)
+        return super().tearDown()
+
+    @skip_without_tool("create_2")
+    def test_cache_populated_after_tool_run(self):
+        history_id = self.dataset_populator.new_history()
+        running_response = self.dataset_populator.run_tool_raw(
+            "create_2",
+            {"sleep_time": 0},
+            history_id,
+        )
+        result = self.dataset_populator.wait_for_tool_run(
+            run_response=running_response, history_id=history_id, assert_ok=False
+        ).json()
+        details = self.dataset_populator.get_job_details(result["jobs"][0]["id"], full=True).json()
+        assert details["state"] == "ok"
+        # check files in the cache are empty after job finishes (they are created by Galaxy before the tool executes)
+        assert files_count(self.object_store_cache_path) == 2
+        hda1_details = self.dataset_populator.get_history_dataset_details(history_id, assert_ok=True, hid=1)
+        hda2_details = self.dataset_populator.get_history_dataset_details(history_id, assert_ok=True, hid=2)
+        assert os.path.getsize(hda1_details["file_name"]) == 0
+        assert os.path.getsize(hda2_details["file_name"]) == 0
+        # get dataset content - this should trigger pulling to the cache
+        content1 = self.dataset_populator.get_history_dataset_content(history_id, dataset_id=hda1_details["dataset_id"])
+        content2 = self.dataset_populator.get_history_dataset_content(history_id, dataset_id=hda2_details["dataset_id"])
+        assert content1 == "1\n"
+        assert content2 == "2\n"
+        # check files in the cache are not empty now
+        assert os.path.getsize(hda1_details["file_name"]) != 0
+        assert os.path.getsize(hda2_details["file_name"]) != 0
diff --git a/test/unit/app/jobs/dynamic_tool_destination/mockGalaxy.py b/test/unit/app/jobs/dynamic_tool_destination/mockGalaxy.py
index 4a618a34b032..17f4ce9cf726 100644
--- a/test/unit/app/jobs/dynamic_tool_destination/mockGalaxy.py
+++ b/test/unit/app/jobs/dynamic_tool_destination/mockGalaxy.py
@@ -43,7 +43,7 @@ def __init__(self, file_name, file_ext, value):
     def get_metadata(self):
         return self.metadata
 
-    def get_file_name(self, **_kwargs):
+    def get_file_name(self, sync_cache=True):
         return self.file_name_
 
 
diff --git a/test/unit/app/tools/test_wrappers.py b/test/unit/app/tools/test_wrappers.py
index e9185de69ef9..69cca55f204f 100644
--- a/test/unit/app/tools/test_wrappers.py
+++ b/test/unit/app/tools/test_wrappers.py
@@ -312,7 +312,7 @@ def __init__(self):
         self.ext = MOCK_DATASET_EXT
         self.tags = []
 
-    def get_file_name(self):
+    def get_file_name(self, sync_cache=True):
         return MOCK_DATASET_PATH
 
 
diff --git a/test/unit/data/datatypes/test_connectivity_table.py b/test/unit/data/datatypes/test_connectivity_table.py
index 3a3037c27c7a..b73ad1fdb923 100644
--- a/test/unit/data/datatypes/test_connectivity_table.py
+++ b/test/unit/data/datatypes/test_connectivity_table.py
@@ -9,7 +9,7 @@
 @pytest.fixture
 def dataset():
     class MockDataset:
-        def get_file_name(self):
+        def get_file_name(self, sync_cache=True):
             return os.path.join(galaxy_directory(), "test-data/1.ct")
 
     return MockDataset()
diff --git a/test/unit/data/datatypes/test_validation.py b/test/unit/data/datatypes/test_validation.py
index dc6ed5ac6623..5ce7e2149539 100644
--- a/test/unit/data/datatypes/test_validation.py
+++ b/test/unit/data/datatypes/test_validation.py
@@ -60,5 +60,5 @@ def __init__(self, file_name, datatype):
         self.file_name_ = file_name
         self.datatype = datatype
 
-    def get_file_name(self):
+    def get_file_name(self, sync_cache=True):
         return self.file_name_
diff --git a/test/unit/data/datatypes/util.py b/test/unit/data/datatypes/util.py
index 881f847ab7aa..ff0de2538994 100644
--- a/test/unit/data/datatypes/util.py
+++ b/test/unit/data/datatypes/util.py
@@ -13,7 +13,7 @@ def __init__(self, file_name):
         self.purged = False
         self.file_name_ = file_name
 
-    def get_file_name(self):
+    def get_file_name(self, sync_cache=True):
         return self.file_name_
 
     def set_file_name(self, file_name):
@@ -23,7 +23,7 @@ def set_file_name(self, file_name):
 class MockMetadata:
     file_name_: Optional[str] = None
 
-    def get_file_name(self):
+    def get_file_name(self, sync_cache=True):
         return self.file_name_
 
     def set_file_name(self, file_name):
@@ -37,7 +37,7 @@ def __init__(self, id):
         self.dataset = None
         self.file_name_: Optional[str] = None
 
-    def get_file_name(self):
+    def get_file_name(self, sync_cache=True):
         return self.file_name_
 
     def set_file_name(self, file_name):
diff --git a/test/unit/tool_util/data/test_tool_data_bundles.py b/test/unit/tool_util/data/test_tool_data_bundles.py
index a884bf2486d1..2e1a1f271740 100644
--- a/test/unit/tool_util/data/test_tool_data_bundles.py
+++ b/test/unit/tool_util/data/test_tool_data_bundles.py
@@ -76,7 +76,7 @@ class OutputDataset:
     extra_files_path: str
     ext: str = "data_manager_json"
 
-    def get_file_name(self) -> str:
+    def get_file_name(self, sync_cache=True) -> str:
         return self.file_name_
 
     def extra_files_path_exists(self) -> bool: