From 87124a396a3d8277ea53b62712c70cb7356b94d9 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Fri, 7 Jun 2024 09:34:34 +0200
Subject: [PATCH 01/16] Don't copy from_work_dir outputs to purged path in
 shell fragment

---
 lib/galaxy/jobs/command_factory.py         | 6 +++++-
 test/unit/app/jobs/test_command_factory.py | 4 ++--
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/lib/galaxy/jobs/command_factory.py b/lib/galaxy/jobs/command_factory.py
index 1c2a67da86ad..bda5c81ce845 100644
--- a/lib/galaxy/jobs/command_factory.py
+++ b/lib/galaxy/jobs/command_factory.py
@@ -293,7 +293,11 @@ def __copy_if_exists_command(work_dir_output):
     source_file, destination = work_dir_output
     if "?" in source_file or "*" in source_file:
         source_file = source_file.replace("*", '"*"').replace("?", '"?"')
-    return f'\nif [ -f "{source_file}" ] ; then cp "{source_file}" "{destination}" ; fi'
+    # Check if source and destination exist.
+    # Users can purge outputs before the job completes,
+    # in that case we don't want to copy the output to a purged path.
+    # Static, non work_dir_output files are handled in job_finish code.
+    return f'\nif [ -f "{source_file}" -a -f "{destination}" ] ; then cp "{source_file}" "{destination}" ; fi'
 
 
 class CommandsBuilder:
diff --git a/test/unit/app/jobs/test_command_factory.py b/test/unit/app/jobs/test_command_factory.py
index 48c097a37139..e185904e94e6 100644
--- a/test/unit/app/jobs/test_command_factory.py
+++ b/test/unit/app/jobs/test_command_factory.py
@@ -21,7 +21,7 @@
 TEST_FILES_PATH = "file_path"
 TEE_REDIRECT = '> "$__out" 2> "$__err"'
 RETURN_CODE_CAPTURE = "; return_code=$?; echo $return_code > galaxy_1.ec"
-CP_WORK_DIR_OUTPUTS = '; \nif [ -f "foo" ] ; then cp "foo" "bar" ; fi'
+CP_WORK_DIR_OUTPUTS = '; \nif [ -f "foo" -a -f "bar" ] ; then cp "foo" "bar" ; fi'
 
 
 class TestCommandFactory(TestCase):
@@ -105,7 +105,7 @@ def test_workdir_outputs_with_glob(self):
         self.workdir_outputs = [("foo*bar", "foo_x_bar")]
         self._assert_command_is(
             self._surround_command(
-                MOCK_COMMAND_LINE, '; \nif [ -f "foo"*"bar" ] ; then cp "foo"*"bar" "foo_x_bar" ; fi'
+                MOCK_COMMAND_LINE, '; \nif [ -f "foo"*"bar" -a -f "foo_x_bar" ] ; then cp "foo"*"bar" "foo_x_bar" ; fi'
             )
         )
 

From c3a5d61d61861dc479a805a6c603ae5ff3d89c2f Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Fri, 7 Jun 2024 09:35:19 +0200
Subject: [PATCH 02/16] Don't include purged outputs in from_work_dir outputs

---
 lib/galaxy/jobs/runners/__init__.py | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/lib/galaxy/jobs/runners/__init__.py b/lib/galaxy/jobs/runners/__init__.py
index 58485fc955ec..d15aa558648d 100644
--- a/lib/galaxy/jobs/runners/__init__.py
+++ b/lib/galaxy/jobs/runners/__init__.py
@@ -380,6 +380,13 @@ def get_work_dir_outputs(
         job_tool = job_wrapper.tool
         for joda, dataset in self._walk_dataset_outputs(job):
             if joda and job_tool:
+                if dataset.dataset.purged:
+                    log.info(
+                        "Output dataset %s for job %s purged before job completed, skipping output collection.",
+                        joda.name,
+                        job.id,
+                    )
+                    continue
                 hda_tool_output = job_tool.find_output_def(joda.name)
                 if hda_tool_output and hda_tool_output.from_work_dir:
                     # Copy from working dir to HDA.

From a1d41ab0d6e71d217f832d27bb1fe86d89012708 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Fri, 7 Jun 2024 09:35:58 +0200
Subject: [PATCH 03/16] Raise appropriate exception if require path parameter
 is missing

---
 lib/galaxy/webapps/galaxy/api/job_files.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/lib/galaxy/webapps/galaxy/api/job_files.py b/lib/galaxy/webapps/galaxy/api/job_files.py
index 74c25e07ea35..3bae6d62d137 100644
--- a/lib/galaxy/webapps/galaxy/api/job_files.py
+++ b/lib/galaxy/webapps/galaxy/api/job_files.py
@@ -99,6 +99,8 @@ def create(self, trans, job_id, payload, **kwargs):
         """
         job = self.__authorize_job_access(trans, job_id, **payload)
         path = payload.get("path")
+        if not path:
+            raise exceptions.RequestParameterInvalidException("'path' parameter not provided or empty.")
         self.__check_job_can_write_to_path(trans, job, path)
 
         # Is this writing an unneeded file? Should this just copy in Python?

From 8e6e25d7fbc436315b83450a21dccaba6b46c4f6 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Fri, 7 Jun 2024 09:36:43 +0200
Subject: [PATCH 04/16] Don't push purged dataset contents to object store

This in particular needs a lot of new tests.
We will also need to actively purge datasets in the model store import
code, since users might have purged datasets while the job ran.
Again, more tests needed.
---
 lib/galaxy/metadata/set_metadata.py |  4 ++--
 lib/galaxy/model/__init__.py        | 15 ++++++++-------
 lib/galaxy/model/store/__init__.py  | 21 +++++++++++----------
 lib/galaxy/model/store/discover.py  |  3 +++
 lib/galaxy/objectstore/__init__.py  |  2 +-
 5 files changed, 25 insertions(+), 20 deletions(-)

diff --git a/lib/galaxy/metadata/set_metadata.py b/lib/galaxy/metadata/set_metadata.py
index 8f1f9ea86be9..d287a9a84c94 100644
--- a/lib/galaxy/metadata/set_metadata.py
+++ b/lib/galaxy/metadata/set_metadata.py
@@ -96,7 +96,7 @@ def push_if_necessary(object_store: ObjectStore, dataset: DatasetInstance, exter
     # or a remote object store from its cache path.
     # empty files could happen when outputs are discovered from working dir,
     # empty file check needed for e.g. test/integration/test_extended_metadata_outputs_to_working_directory.py::test_tools[multi_output_assign_primary]
-    if os.path.getsize(external_filename):
+    if not dataset.dataset.purged and os.path.getsize(external_filename):
         object_store.update_from_file(dataset.dataset, file_name=external_filename, create=True)
 
 
@@ -477,7 +477,7 @@ def set_meta(new_dataset_instance, file_dict):
                     object_store_update_actions.append(partial(reset_external_filename, dataset))
                 object_store_update_actions.append(partial(dataset.set_total_size))
                 object_store_update_actions.append(partial(export_store.add_dataset, dataset))
-                if dataset_instance_id not in unnamed_id_to_path:
+                if dataset_instance_id not in unnamed_id_to_path and not dataset.dataset.purged:
                     object_store_update_actions.append(partial(collect_extra_files, object_store, dataset, "."))
                     dataset_state = "deferred" if (is_deferred and final_job_state == "ok") else final_job_state
                     if not dataset.state == dataset.states.ERROR:
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 97ff0f6b0fbf..582da171d315 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -9485,13 +9485,14 @@ def dataset(self) -> Optional[Dataset]:
     def update_from_file(self, file_name):
         if not self.dataset:
             raise Exception("Attempted to write MetadataFile, but no DatasetAssociation set")
-        self.dataset.object_store.update_from_file(
-            self,
-            file_name=file_name,
-            extra_dir="_metadata_files",
-            extra_dir_at_root=True,
-            alt_name=os.path.basename(self.get_file_name()),
-        )
+        if not self.dataset.purged:
+            self.dataset.object_store.update_from_file(
+                self,
+                file_name=file_name,
+                extra_dir="_metadata_files",
+                extra_dir_at_root=True,
+                alt_name=os.path.basename(self.get_file_name()),
+            )
 
     def get_file_name(self, sync_cache=True):
         # Ensure the directory structure and the metadata file object exist
diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py
index 1b78f52ebcc8..42c6e927e2c1 100644
--- a/lib/galaxy/model/store/__init__.py
+++ b/lib/galaxy/model/store/__init__.py
@@ -654,17 +654,18 @@ def handle_dataset_object_edit(dataset_instance, dataset_attrs):
                         dataset_instance.state = dataset_state
                         if not self.object_store:
                             raise Exception(f"self.object_store is missing from {self}.")
-                        self.object_store.update_from_file(
-                            dataset_instance.dataset, file_name=temp_dataset_file_name, create=True
-                        )
+                        if not dataset_instance.dataset.purged:
+                            self.object_store.update_from_file(
+                                dataset_instance.dataset, file_name=temp_dataset_file_name, create=True
+                            )
 
-                        # Import additional files if present. Histories exported previously might not have this attribute set.
-                        dataset_extra_files_path = dataset_attrs.get("extra_files_path", None)
-                        if dataset_extra_files_path:
-                            assert file_source_root
-                            dataset_extra_files_path = os.path.join(file_source_root, dataset_extra_files_path)
-                            persist_extra_files(self.object_store, dataset_extra_files_path, dataset_instance)
-                        # Don't trust serialized file size
+                            # Import additional files if present. Histories exported previously might not have this attribute set.
+                            dataset_extra_files_path = dataset_attrs.get("extra_files_path", None)
+                            if dataset_extra_files_path:
+                                assert file_source_root
+                                dataset_extra_files_path = os.path.join(file_source_root, dataset_extra_files_path)
+                                persist_extra_files(self.object_store, dataset_extra_files_path, dataset_instance)
+                            # Don't trust serialized file size
                         dataset_instance.dataset.file_size = None
                         dataset_instance.dataset.set_total_size()  # update the filesize record in the database
 
diff --git a/lib/galaxy/model/store/discover.py b/lib/galaxy/model/store/discover.py
index 9329bda9ee40..2db3749feac9 100644
--- a/lib/galaxy/model/store/discover.py
+++ b/lib/galaxy/model/store/discover.py
@@ -214,6 +214,9 @@ def create_dataset(
         return primary_data
 
     def finalize_storage(self, primary_data, dataset_attributes, extra_files, filename, link_data, output_name):
+        if primary_data.dataset.purged:
+            # metadata won't be set, maybe we should do that, then purge ?
+            return
         # Move data from temp location to dataset location
         if not link_data:
             dataset = primary_data.dataset
diff --git a/lib/galaxy/objectstore/__init__.py b/lib/galaxy/objectstore/__init__.py
index f6bb0f4b07ae..49b559525ea9 100644
--- a/lib/galaxy/objectstore/__init__.py
+++ b/lib/galaxy/objectstore/__init__.py
@@ -1670,7 +1670,7 @@ def persist_extra_files(
     primary_data: "DatasetInstance",
     extra_files_path_name: Optional[str] = None,
 ) -> None:
-    if os.path.exists(src_extra_files_path):
+    if not primary_data.dataset.purged and os.path.exists(src_extra_files_path):
         assert primary_data.dataset
         if not extra_files_path_name:
             extra_files_path_name = primary_data.dataset.extra_files_path_name_from(object_store)

From 8bc3f2151af1a5d684062b705b36675bfa8e5a02 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Sat, 8 Jun 2024 16:27:49 +0200
Subject: [PATCH 05/16] Add tool that produces all types of outputs

---
 test/functional/tools/all_output_types.xml | 84 ++++++++++++++++++++++
 test/functional/tools/sample_tool_conf.xml | 24 +++----
 2 files changed, 96 insertions(+), 12 deletions(-)
 create mode 100644 test/functional/tools/all_output_types.xml

diff --git a/test/functional/tools/all_output_types.xml b/test/functional/tools/all_output_types.xml
new file mode 100644
index 000000000000..1914c3b6009a
--- /dev/null
+++ b/test/functional/tools/all_output_types.xml
@@ -0,0 +1,84 @@
+<tool id="all_output_types" name="all_output_types" version="1.0.0" profile="24.0">
+    <command><![CDATA[
+        echo hi > output.txt &&
+        cp output.txt '$static_output' &&
+        cp '$c1' galaxy.json
+        ]]>
+    </command>
+    <configfiles>
+        <configfile name="c1">{"output_tool_supplied_metadata": {
+    "name": "my dynamic name",
+    "ext": "txt",
+    "info": "my dynamic info"
+  }}
+        </configfile>
+      </configfiles>
+    <outputs>
+        <data name="static_output" format="txt" />
+        <data name="output_workdir" from_work_dir="output.txt" format="txt" />
+        <data name="output_tool_supplied_metadata" from_work_dir="output.txt" format="auto" />
+        <data format="txt" name="discovered_output">
+            <discover_datasets pattern="(?P&lt;designation&gt;.+)\.txt" ext="txt" visible="true"/>
+        </data>
+        <data format="txt" name="discovered_output_replaced">
+            <discover_datasets pattern="(?P&lt;designation&gt;.+)\.txt" ext="txt" visible="true" assign_primary_output="true" />
+        </data>
+        <collection type="paired" name="static_pair" format="txt">
+            <data name="forward" from_work_dir="output.txt"></data>
+            <data name="reverse" from_work_dir="output.txt"></data>
+        </collection>
+        <collection type="list" name="discovered_list" format="txt">
+            <discover_datasets pattern="(?P&lt;designation&gt;.+)\.txt" ext="txt" visible="true" />
+        </collection>
+    </outputs>
+    <tests>
+        <test>
+            <output name="static_output">
+                <assert_contents>
+                    <has_text text="hi"/>
+                </assert_contents>
+            </output>
+            <output name="output_workdir">
+                <assert_contents>
+                    <has_text text="hi"/>
+                </assert_contents>
+            </output>
+            <output name="output_tool_supplied_metadata">
+                <assert_contents>
+                    <has_text text="hi"/>
+                </assert_contents>
+            </output>
+            <output name="discovered_output">
+                <discovered_dataset designation="output" ftype="txt">
+                    <assert_contents>
+                        <has_text text="hi"/>
+                    </assert_contents>
+                  </discovered_dataset>
+            </output>
+            <output name="discovered_output_replaced" count="1">
+                <assert_contents>
+                    <has_text text="hi"/>
+                </assert_contents>
+            </output>
+            <output_collection name="static_pair" type="paired">
+                <element name="forward" ftype="txt">
+                    <assert_contents>
+                        <has_text text="hi"></has_text>
+                    </assert_contents>
+                </element>
+                <element name="reverse" ftype="txt">
+                    <assert_contents>
+                        <has_text text="hi"></has_text>
+                    </assert_contents>
+                </element>
+            </output_collection>
+            <output_collection name="discovered_list">
+                <element name="output" ftype="txt">
+                    <assert_contents>
+                        <has_text text="hi"></has_text>
+                    </assert_contents>
+                </element>
+            </output_collection>
+        </test>
+    </tests>
+</tool>
diff --git a/test/functional/tools/sample_tool_conf.xml b/test/functional/tools/sample_tool_conf.xml
index 8775d130e520..2c074381db5e 100644
--- a/test/functional/tools/sample_tool_conf.xml
+++ b/test/functional/tools/sample_tool_conf.xml
@@ -8,10 +8,10 @@
     <tool file="param_text_option.xml" />
     <tool file="column_param.xml" />
   </section>
-  <tool file="image_diff.xml"/>
-  <tool file="output_format_input.xml"/>
-  <tool file="ucsc_tablebrowser.xml"/>
-  <tool file="test_data_source.xml"/>
+  <tool file="image_diff.xml" />
+  <tool file="output_format_input.xml" />
+  <tool file="ucsc_tablebrowser.xml" />
+  <tool file="test_data_source.xml" />
   <tool file="simple_constructs.xml" />
   <tool file="color_param.xml" />
   <tool file="inheritance_simple.xml" />
@@ -89,7 +89,7 @@
   <tool file="sim_size_delta.xml" />
   <tool file="composite_shapefile.xml" />
   <tool file="is_valid_xml.xml" />
-  <tool file="expression_tools/parse_values_from_file.xml"/>
+  <tool file="expression_tools/parse_values_from_file.xml" />
   <tool file="expression_tools/pick_value.xml" />
   <!--
   TODO: Figure out why this transiently fails on Jenkins.
@@ -137,7 +137,7 @@
   <tool file="column_param_configfile.xml" />
   <tool file="column_param_list.xml" />
   <tool file="column_multi_param.xml" />
-  <tool file="select_optional.xml"/>
+  <tool file="select_optional.xml" />
   <tool file="hidden_param.xml" />
   <tool file="special_params.xml" />
   <tool file="section.xml" />
@@ -150,10 +150,10 @@
   <tool file="validation_default.xml" />
   <tool file="validation_sanitizer.xml" />
   <tool file="validation_repeat.xml" />
-  <tool file="validation_metadata_in_range.xml"/>
-  <tool file="validation_metadata_in_datatable.xml"/>
-  <tool file="validation_dataset_metadata_in_file.xml"/>
-  <tool file="validation_value_in_datatable.xml"/>
+  <tool file="validation_metadata_in_range.xml" />
+  <tool file="validation_metadata_in_datatable.xml" />
+  <tool file="validation_dataset_metadata_in_file.xml" />
+  <tool file="validation_value_in_datatable.xml" />
   <tool file="empty_output.xml" />
   <tool file="validation_empty_dataset.xml" />
   <tool file="implicit_conversion.xml" />
@@ -170,7 +170,7 @@
   <tool file="identifier_multiple_in_conditional.xml" />
   <tool file="identifier_multiple_in_repeat.xml" />
   <tool file="identifier_collection.xml" />
-  <tool file="identifier_all_collection_types.xml"/>
+  <tool file="identifier_all_collection_types.xml" />
   <tool file="identifier_in_actions.xml" />
   <tool file="fail_identifier.xml" />
   <tool file="fail_writing_work_dir_file.xml" />
@@ -224,7 +224,7 @@
   <tool file="expect_num_outputs.xml" />
   <tool file="text_repeat.xml" />
   <tool file="integer_default.xml" />
-
+  <tool file="all_output_types.xml" />
   <tool file="multiple_versions_v01.xml" />
   <tool file="multiple_versions_v01galaxy6.xml" />
   <tool file="multiple_versions_v02.xml" />

From 65e9605f4a4c77c5af2ab6d473a3a894eabb6bcb Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Sat, 8 Jun 2024 20:02:32 +0200
Subject: [PATCH 06/16] Don't fail _finish_or_resubmit if tool_stdout /
 tool_stderr not written

These are only written if the command actually ran, so we'd fail here
for instance if the outputs are deleted before the job had a chance to
run.
---
 lib/galaxy/jobs/runners/__init__.py | 21 +++++++++++++++++----
 1 file changed, 17 insertions(+), 4 deletions(-)

diff --git a/lib/galaxy/jobs/runners/__init__.py b/lib/galaxy/jobs/runners/__init__.py
index d15aa558648d..db1e1975f5e3 100644
--- a/lib/galaxy/jobs/runners/__init__.py
+++ b/lib/galaxy/jobs/runners/__init__.py
@@ -625,10 +625,23 @@ def _finish_or_resubmit_job(self, job_state: "JobState", job_stdout, job_stderr,
 
             tool_stdout_path = os.path.join(outputs_directory, "tool_stdout")
             tool_stderr_path = os.path.join(outputs_directory, "tool_stderr")
-            with open(tool_stdout_path, "rb") as stdout_file:
-                tool_stdout = self._job_io_for_db(stdout_file)
-            with open(tool_stderr_path, "rb") as stderr_file:
-                tool_stderr = self._job_io_for_db(stderr_file)
+            try:
+                with open(tool_stdout_path, "rb") as stdout_file:
+                    tool_stdout = self._job_io_for_db(stdout_file)
+                with open(tool_stderr_path, "rb") as stderr_file:
+                    tool_stderr = self._job_io_for_db(stderr_file)
+            except FileNotFoundError:
+                if job.state in (model.Job.states.DELETING, model.Job.states.DELETED):
+                    # We killed the job, so we may not even have the tool stdout / tool stderr
+                    tool_stdout = ""
+                    tool_stderr = "Job cancelled"
+                else:
+                    # Should we instead just move on ?
+                    # In the end the only consequence here is that we won't be able to determine
+                    # if the job failed for known tool reasons (check_tool_output).
+                    # OTOH I don't know if this can even be reached
+                    # Deal with it if we ever get reports about this.
+                    raise
 
             check_output_detected_state = job_wrapper.check_tool_output(
                 tool_stdout,

From 435a8cd37fefda96d2e7eeaa170b03529a446cf2 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Sun, 9 Jun 2024 08:48:20 +0200
Subject: [PATCH 07/16] Fix up test cases

Add a sleep so we can delay output purging until command line is
templated. If we don't sleep we generate a command line where the
output path is just `''`. That needs another fix!
---
 test/functional/tools/all_output_types.xml    | 10 +++-
 .../objectstore/_purged_handling.py           | 45 ++++++++++++++
 test/integration/test_extended_metadata.py    |  5 ++
 test/integration/test_pulsar_embedded_mq.py   | 60 ++++++++++++++-----
 4 files changed, 102 insertions(+), 18 deletions(-)
 create mode 100644 test/integration/objectstore/_purged_handling.py

diff --git a/test/functional/tools/all_output_types.xml b/test/functional/tools/all_output_types.xml
index 1914c3b6009a..58cd10e8a4a9 100644
--- a/test/functional/tools/all_output_types.xml
+++ b/test/functional/tools/all_output_types.xml
@@ -1,7 +1,9 @@
 <tool id="all_output_types" name="all_output_types" version="1.0.0" profile="24.0">
     <command><![CDATA[
+        sleep $sleep_param &&
         echo hi > output.txt &&
-        cp output.txt '$static_output' &&
+        echo hi > '$static_output' &&
+        echo hi > '$static_output_2' &&
         cp '$c1' galaxy.json
         ]]>
     </command>
@@ -12,9 +14,13 @@
     "info": "my dynamic info"
   }}
         </configfile>
-      </configfiles>
+    </configfiles>
+    <inputs>
+        <param name="sleep_param" type="integer" value="0" />
+    </inputs>
     <outputs>
         <data name="static_output" format="txt" />
+        <data name="static_output_2" format="txt" />
         <data name="output_workdir" from_work_dir="output.txt" format="txt" />
         <data name="output_tool_supplied_metadata" from_work_dir="output.txt" format="auto" />
         <data format="txt" name="discovered_output">
diff --git a/test/integration/objectstore/_purged_handling.py b/test/integration/objectstore/_purged_handling.py
new file mode 100644
index 000000000000..9f118b77f3cf
--- /dev/null
+++ b/test/integration/objectstore/_purged_handling.py
@@ -0,0 +1,45 @@
+import time
+
+from galaxy_test.base.populators import DatasetPopulator
+
+
+def purge_while_job_running(dataset_populator: DatasetPopulator, sleep_before_purge=4):
+    with dataset_populator.test_history() as history_id:
+        response = dataset_populator.run_tool(
+            "all_output_types",
+            inputs={
+                "sleep_param": 5,
+            },
+            history_id=history_id,
+        )
+        job = dataset_populator.get_job_details(response["jobs"][0]["id"], full=True).json()
+        # Sleep a second to make sure we template command before purging output
+        time.sleep(sleep_before_purge)
+        hda_ids = []
+        for output_name, output in job["outputs"].items():
+            if output_name == "static_output_2":
+                # We need to keep one output so the job doesn't get cancelled
+                continue
+            dataset_populator.delete_dataset(
+                history_id=history_id, content_id=output["id"], purge=True, wait_for_purge=True
+            )
+            hda_ids.append(output["id"])
+        for output_collection in job["output_collections"].values():
+            hdca = dataset_populator.get_history_collection_details(
+                history_id=history_id, content_id=output_collection["id"]
+            )
+            for element in hdca["elements"]:
+                hda_id = element["object"]["id"]
+                dataset_populator.delete_dataset(
+                    history_id=history_id, content_id=hda_id, purge=True, wait_for_purge=True
+                )
+                hda_ids.append(hda_id)
+        dataset_populator.wait_for_job(job["id"], assert_ok=True)
+        # Now make sure we can't find the datasets
+        for hda_id in hda_ids:
+            exception = None
+            try:
+                dataset_populator.get_history_dataset_content(history_id=history_id, dataset_id=hda_id)
+            except AssertionError as e:
+                exception = e
+            assert exception and "The dataset you are attempting to view has been purged" in str(exception)
diff --git a/test/integration/test_extended_metadata.py b/test/integration/test_extended_metadata.py
index ea433f5c2fab..cdb3ea1c6e2e 100644
--- a/test/integration/test_extended_metadata.py
+++ b/test/integration/test_extended_metadata.py
@@ -5,6 +5,7 @@
     LibraryPopulator,
 )
 from galaxy_test.driver import integration_util
+from .objectstore._purged_handling import purge_while_job_running
 
 TEST_TOOL_IDS = [
     "from_work_dir_glob",
@@ -47,6 +48,7 @@
 
 class TestExtendedMetadataIntegration(integration_util.IntegrationTestCase):
     dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
 
     def setUp(self):
         super().setUp()
@@ -95,6 +97,9 @@ def test_fetch_data_library(self):
         assert dataset["file_ext"] == "bed", dataset
         assert dataset["created_from_basename"] == "4.bed"
 
+    def test_purge_while_job_running(self):
+        purge_while_job_running(self.dataset_populator)
+
 
 class TestExtendedMetadataDeferredIntegration(integration_util.IntegrationTestCase):
     dataset_populator: DatasetPopulator
diff --git a/test/integration/test_pulsar_embedded_mq.py b/test/integration/test_pulsar_embedded_mq.py
index e0b9e156992b..ef1979866409 100644
--- a/test/integration/test_pulsar_embedded_mq.py
+++ b/test/integration/test_pulsar_embedded_mq.py
@@ -7,7 +7,9 @@
 import pytest
 
 from galaxy.util import safe_makedirs
+from galaxy_test.base.populators import DatasetPopulator
 from galaxy_test.driver import integration_util
+from .objectstore._purged_handling import purge_while_job_running
 
 SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
 EMBEDDED_PULSAR_JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "embedded_pulsar_mq_job_conf.yml")
@@ -46,6 +48,40 @@
 """
 
 
+def _handle_galaxy_config_kwds(cls, config):
+    amqp_url = os.environ.get("GALAXY_TEST_AMQP_URL", None)
+    if amqp_url is None:
+        pytest.skip("External AMQP URL not configured for test")
+
+    jobs_directory = os.path.join(cls._test_driver.mkdtemp(), "pulsar_staging")
+    safe_makedirs(jobs_directory)
+    job_conf_template = string.Template(JOB_CONF_TEMPLATE)
+    job_conf_str = job_conf_template.substitute(
+        amqp_url=AMQP_URL, jobs_directory=jobs_directory, galaxy_home=os.path.join(SCRIPT_DIRECTORY, os.pardir)
+    )
+    with tempfile.NamedTemporaryFile(suffix="_mq_job_conf.yml", mode="w", delete=False) as job_conf:
+        job_conf.write(job_conf_str)
+    config["job_config_file"] = job_conf.name
+    infrastructure_url = "http://localhost:$GALAXY_WEB_PORT"
+    config["galaxy_infrastructure_url"] = infrastructure_url
+
+
+class TestEmbeddedMessageQueuePulsarPurge(integration_util.IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
+
+    def setUp(self) -> None:
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        _handle_galaxy_config_kwds(cls, config)
+
+    def test_purge_while_job_running(self):
+        purge_while_job_running(self.dataset_populator)
+
+
 class EmbeddedMessageQueuePulsarIntegrationInstance(integration_util.IntegrationInstance):
     """Describe a Galaxy test instance with embedded pulsar configured.
 
@@ -53,27 +89,19 @@ class EmbeddedMessageQueuePulsarIntegrationInstance(integration_util.Integration
     $ GALAXY_TEST_AMQP_URL='amqp://guest:guest@localhost:5672//' pytest -s test/integration/test_pulsar_embedded_mq.py
     """
 
+    dataset_populator: DatasetPopulator
     framework_tool_and_types = True
 
     @classmethod
     def handle_galaxy_config_kwds(cls, config):
-        amqp_url = os.environ.get("GALAXY_TEST_AMQP_URL", None)
-        if amqp_url is None:
-            pytest.skip("External AMQP URL not configured for test")
-
-        jobs_directory = os.path.join(cls._test_driver.mkdtemp(), "pulsar_staging")
-        safe_makedirs(jobs_directory)
-        job_conf_template = string.Template(JOB_CONF_TEMPLATE)
-        job_conf_str = job_conf_template.substitute(
-            amqp_url=AMQP_URL, jobs_directory=jobs_directory, galaxy_home=os.path.join(SCRIPT_DIRECTORY, os.pardir)
-        )
-        with tempfile.NamedTemporaryFile(suffix="_mq_job_conf.yml", mode="w", delete=False) as job_conf:
-            job_conf.write(job_conf_str)
-        config["job_config_file"] = job_conf.name
-        infrastructure_url = "http://localhost:$GALAXY_WEB_PORT"
-        config["galaxy_infrastructure_url"] = infrastructure_url
+        _handle_galaxy_config_kwds(cls, config)
+
+    def test_purge_while_job_running(self):
+        purge_while_job_running(self.dataset_populator)
 
 
 instance = integration_util.integration_module_instance(EmbeddedMessageQueuePulsarIntegrationInstance)
 
-test_tools = integration_util.integration_tool_runner(["simple_constructs", "composite_output_tests"])
+test_tools = integration_util.integration_tool_runner(
+    ["simple_constructs", "composite_output_tests", "all_output_types"]
+)

From 03eca7cc521cf695f5ffc24ae10f18a4f8081eae Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Sun, 9 Jun 2024 08:50:19 +0200
Subject: [PATCH 08/16] Fix handling of collecting discovered but purged
 outputs

---
 lib/galaxy/job_execution/output_collect.py | 12 ++++++++++--
 lib/galaxy/model/store/discover.py         |  7 ++++++-
 2 files changed, 16 insertions(+), 3 deletions(-)

diff --git a/lib/galaxy/job_execution/output_collect.py b/lib/galaxy/job_execution/output_collect.py
index aae1336f6484..8e37864767d0 100644
--- a/lib/galaxy/job_execution/output_collect.py
+++ b/lib/galaxy/job_execution/output_collect.py
@@ -521,7 +521,8 @@ def collect_primary_datasets(job_context: Union[JobContext, SessionlessJobContex
                 outdata.designation = designation
                 outdata.dataset.external_filename = None  # resets filename_override
                 # Move data from temp location to dataset location
-                job_context.object_store.update_from_file(outdata.dataset, file_name=filename, create=True)
+                if not outdata.dataset.purged:
+                    job_context.object_store.update_from_file(outdata.dataset, file_name=filename, create=True)
                 primary_output_assigned = True
                 continue
             if name not in primary_datasets:
@@ -554,6 +555,7 @@ def collect_primary_datasets(job_context: Union[JobContext, SessionlessJobContex
                 dataset_attributes=new_primary_datasets_attributes,
                 creating_job_id=job_context.get_job_id() if job_context else None,
                 storage_callbacks=storage_callbacks,
+                purged=outdata.dataset.purged,
             )
             # Associate new dataset with job
             job_context.add_output_dataset_association(f"__new_primary_file_{name}|{designation}__", primary_data)
@@ -563,7 +565,13 @@ def collect_primary_datasets(job_context: Union[JobContext, SessionlessJobContex
         if primary_output_assigned:
             outdata.name = new_outdata_name
             outdata.init_meta()
-            outdata.set_meta()
+            if not outdata.dataset.purged:
+                try:
+                    outdata.set_meta()
+                except Exception:
+                    # We don't want to fail here on a single "bad" discovered dataset
+                    log.debug("set meta failed for %s", outdata, exc_info=True)
+                    outdata.state = HistoryDatasetAssociation.states.FAILED_METADATA
             outdata.set_peek()
             outdata.discovered = True
             sa_session = job_context.sa_session
diff --git a/lib/galaxy/model/store/discover.py b/lib/galaxy/model/store/discover.py
index 2db3749feac9..b3a902ff8487 100644
--- a/lib/galaxy/model/store/discover.py
+++ b/lib/galaxy/model/store/discover.py
@@ -91,6 +91,7 @@ def create_dataset(
         creating_job_id=None,
         output_name=None,
         storage_callbacks=None,
+        purged=False,
     ):
         tag_list = tag_list or []
         sources = sources or []
@@ -190,7 +191,11 @@ def create_dataset(
 
         if info is not None:
             primary_data.info = info
-        if filename:
+
+        if purged:
+            primary_data.dataset.purged = True
+            primary_data.purged = True
+        if filename and not purged:
             if storage_callbacks is None:
                 self.finalize_storage(
                     primary_data=primary_data,

From eb2e318b3de99505a4540f6c9a86b6bcaa51353c Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 Jun 2024 11:52:08 +0200
Subject: [PATCH 09/16] Elevate external metadata setting to warning

This is a tricky one. There could be systematic issues here that an
admin would want to fix, but it could also just be the case that a user
forced a wrong datatype. Ideally we'd probably tag this as job execution
issue ? It's not very different from a job error IMO.
---
 lib/galaxy/metadata/__init__.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/galaxy/metadata/__init__.py b/lib/galaxy/metadata/__init__.py
index 975d7ccf79f4..1f44e272ab8a 100644
--- a/lib/galaxy/metadata/__init__.py
+++ b/lib/galaxy/metadata/__init__.py
@@ -118,7 +118,7 @@ def _metadata_results_from_file(self, dataset, filename_results_code):
             rstring = f"Metadata results could not be read from '{filename_results_code}'"
 
         if not rval:
-            log.debug(f"setting metadata externally failed for {dataset.__class__.__name__} {dataset.id}: {rstring}")
+            log.warning(f"setting metadata externally failed for {dataset.__class__.__name__} {dataset.id}: {rstring}")
         return rval
 
 

From fc42b61d300234d8a0fc9fe9c01b3d6cc5d3a6f4 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 Jun 2024 11:55:36 +0200
Subject: [PATCH 10/16] Fix metadata setting on purged outputs

---
 lib/galaxy/metadata/set_metadata.py | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/lib/galaxy/metadata/set_metadata.py b/lib/galaxy/metadata/set_metadata.py
index d287a9a84c94..38f3f7917729 100644
--- a/lib/galaxy/metadata/set_metadata.py
+++ b/lib/galaxy/metadata/set_metadata.py
@@ -426,6 +426,10 @@ def set_meta(new_dataset_instance, file_dict):
                         # as opposed to perhaps a storage issue.
                         with open(external_filename, "wb"):
                             pass
+                    elif not os.path.exists(dataset_filename_override):
+                        # purged output ?
+                        dataset.purged = True
+                        dataset.dataset.purged = True
                     else:
                         raise Exception(f"Output file '{external_filename}' not found")
 
@@ -485,7 +489,8 @@ def set_meta(new_dataset_instance, file_dict):
                         dataset.state = dataset.dataset.state = dataset_state
                     # We're going to run through set_metadata in collect_dynamic_outputs with more contextual metadata,
                     # so only run set_meta for fixed outputs
-                    set_meta(dataset, file_dict)
+                    if not dataset.dataset.purged:
+                        set_meta(dataset, file_dict)
                 # TODO: merge expression_context into tool_provided_metadata so we don't have to special case this (here and in _finish_dataset)
                 meta = tool_provided_metadata.get_dataset_meta(output_name, dataset.dataset.id, dataset.dataset.uuid)
                 if meta:
@@ -512,7 +517,7 @@ def set_meta(new_dataset_instance, file_dict):
                         context_value = context[context_key]
                         setattr(dataset, context_key, context_value)
             else:
-                if dataset_instance_id not in unnamed_id_to_path:
+                if dataset_instance_id not in unnamed_id_to_path and not dataset.dataset.purged:
                     # We're going to run through set_metadata in collect_dynamic_outputs with more contextual metadata,
                     # so only run set_meta for fixed outputs
                     set_meta(dataset, file_dict)

From a35033d821feb7a42d78e2a87a99e1ca3be710cf Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 Jun 2024 11:56:03 +0200
Subject: [PATCH 11/16] Extend test to make sure disk path remains purged

---
 test/integration/objectstore/_purged_handling.py | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/test/integration/objectstore/_purged_handling.py b/test/integration/objectstore/_purged_handling.py
index 9f118b77f3cf..9156b4964809 100644
--- a/test/integration/objectstore/_purged_handling.py
+++ b/test/integration/objectstore/_purged_handling.py
@@ -1,3 +1,4 @@
+import os
 import time
 
 from galaxy_test.base.populators import DatasetPopulator
@@ -16,10 +17,13 @@ def purge_while_job_running(dataset_populator: DatasetPopulator, sleep_before_pu
         # Sleep a second to make sure we template command before purging output
         time.sleep(sleep_before_purge)
         hda_ids = []
+        hda_filenames = []
         for output_name, output in job["outputs"].items():
             if output_name == "static_output_2":
                 # We need to keep one output so the job doesn't get cancelled
                 continue
+            details = dataset_populator.get_history_dataset_details(history_id=history_id, dataset_id=output["id"])
+            hda_filenames.append(details["file_name"])
             dataset_populator.delete_dataset(
                 history_id=history_id, content_id=output["id"], purge=True, wait_for_purge=True
             )
@@ -43,3 +47,9 @@ def purge_while_job_running(dataset_populator: DatasetPopulator, sleep_before_pu
             except AssertionError as e:
                 exception = e
             assert exception and "The dataset you are attempting to view has been purged" in str(exception)
+            output_details = dataset_populator.get_history_dataset_details(history_id=history_id, dataset_id=hda_id)
+            # Make sure that we don't revert state while finishing job
+            assert output_details["purged"]
+        for file_name in hda_filenames:
+            # Make sure job didn't push to object store
+            assert not os.path.exists(file_name), f"Expected {file_name} to be purged."

From c489eb20af0eb9779bd84ffa10df915f1662253e Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 Jun 2024 12:10:53 +0200
Subject: [PATCH 12/16] Set file_size and total_size to 0 for discovered but
 purged datasets

---
 lib/galaxy/model/store/discover.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/lib/galaxy/model/store/discover.py b/lib/galaxy/model/store/discover.py
index b3a902ff8487..679bd2aa38f7 100644
--- a/lib/galaxy/model/store/discover.py
+++ b/lib/galaxy/model/store/discover.py
@@ -221,6 +221,8 @@ def create_dataset(
     def finalize_storage(self, primary_data, dataset_attributes, extra_files, filename, link_data, output_name):
         if primary_data.dataset.purged:
             # metadata won't be set, maybe we should do that, then purge ?
+            primary_data.dataset.file_size = 0
+            primary_data.dataset.total_size = 0
             return
         # Move data from temp location to dataset location
         if not link_data:

From 5ab312027041394d3350ca8af10ee534c7c5c3db Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 Jun 2024 15:42:20 +0200
Subject: [PATCH 13/16] Purge dataset if finishing job for purged output

---
 lib/galaxy/jobs/__init__.py | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index fe3d4995fb4e..e6fd99f2340e 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -2000,10 +2000,14 @@ def fail(message=job.info, exception=None):
         quota_source_info = None
         # Once datasets are collected, set the total dataset size (includes extra files)
         for dataset_assoc in job.output_datasets:
-            if not dataset_assoc.dataset.dataset.purged:
+            dataset = dataset_assoc.dataset.dataset
+            if not dataset.purged:
                 # assume all datasets in a job get written to the same objectstore
-                quota_source_info = dataset_assoc.dataset.dataset.quota_source_info
-                collected_bytes += dataset_assoc.dataset.set_total_size()
+                quota_source_info = dataset.quota_source_info
+                collected_bytes += dataset.set_total_size()
+            else:
+                # Purge, in case job wrote directly to object store
+                dataset.full_delete()
 
         user = job.user
         if user and collected_bytes > 0 and quota_source_info is not None and quota_source_info.use:

From 50b7f8f8daa98d6a22e042744914f4fd13fad19e Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 Jun 2024 15:44:04 +0200
Subject: [PATCH 14/16] Do not overwrite purged or deleted states when
 importing datasets from job

---
 lib/galaxy/model/__init__.py       | 2 ++
 lib/galaxy/model/store/__init__.py | 2 --
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 582da171d315..84a28aaa396a 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -4211,6 +4211,8 @@ def full_delete(self):
         # TODO: purge metadata files
         self.deleted = True
         self.purged = True
+        self.file_size = 0
+        self.total_size = 0
 
     def get_access_roles(self, security_agent):
         roles = []
diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py
index 42c6e927e2c1..4fd9855ef751 100644
--- a/lib/galaxy/model/store/__init__.py
+++ b/lib/galaxy/model/store/__init__.py
@@ -183,8 +183,6 @@ class ImportDiscardedDataType(Enum):
 
 class DatasetAttributeImportModel(BaseModel):
     state: Optional[DatasetStateField] = None
-    deleted: Optional[bool] = None
-    purged: Optional[bool] = None
     external_filename: Optional[str] = None
     _extra_files_path: Optional[str] = None
     file_size: Optional[int] = None

From 3c7730c1994543b33e68816c318407d1a3a82809 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 Jun 2024 15:45:49 +0200
Subject: [PATCH 15/16] Delete purged outputs from object store when importing
 dataset for purged output

---
 lib/galaxy/model/store/__init__.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py
index 4fd9855ef751..4880b83e7b96 100644
--- a/lib/galaxy/model/store/__init__.py
+++ b/lib/galaxy/model/store/__init__.py
@@ -468,6 +468,8 @@ def handle_dataset_object_edit(dataset_instance, dataset_attrs):
                 )
                 for attribute, value in dataset_attributes.items():
                     setattr(dataset_instance.dataset, attribute, value)
+                if dataset_instance.dataset.purged:
+                    dataset_instance.dataset.full_delete()
                 self._attach_dataset_hashes(dataset_attrs["dataset"], dataset_instance)
                 self._attach_dataset_sources(dataset_attrs["dataset"], dataset_instance)
                 if "id" in dataset_attrs["dataset"] and self.import_options.allow_edit:

From f0e09cc8bdf19d4d06731ab6bfc8ea6ebfc26b28 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Mon, 10 Jun 2024 15:46:32 +0200
Subject: [PATCH 16/16] Fix up tests

---
 .../objectstore/_purged_handling.py           | 34 ++++++++++++++-----
 test/integration/test_extended_metadata.py    |  4 ++-
 test/integration/test_pulsar_embedded_mq.py   | 13 +++++++
 3 files changed, 41 insertions(+), 10 deletions(-)

diff --git a/test/integration/objectstore/_purged_handling.py b/test/integration/objectstore/_purged_handling.py
index 9156b4964809..1437f6742ae5 100644
--- a/test/integration/objectstore/_purged_handling.py
+++ b/test/integration/objectstore/_purged_handling.py
@@ -4,7 +4,7 @@
 from galaxy_test.base.populators import DatasetPopulator
 
 
-def purge_while_job_running(dataset_populator: DatasetPopulator, sleep_before_purge=4):
+def purge_while_job_running(dataset_populator: DatasetPopulator, extra_sleep=0):
     with dataset_populator.test_history() as history_id:
         response = dataset_populator.run_tool(
             "all_output_types",
@@ -14,16 +14,21 @@ def purge_while_job_running(dataset_populator: DatasetPopulator, sleep_before_pu
             history_id=history_id,
         )
         job = dataset_populator.get_job_details(response["jobs"][0]["id"], full=True).json()
-        # Sleep a second to make sure we template command before purging output
-        time.sleep(sleep_before_purge)
+        # Make sure job runs (and thus command is templated before purging output)
+        dataset_populator.wait_for_job(job["id"], ok_states=["running"])
+        time.sleep(extra_sleep)
         hda_ids = []
         hda_filenames = []
         for output_name, output in job["outputs"].items():
             if output_name == "static_output_2":
                 # We need to keep one output so the job doesn't get cancelled
                 continue
-            details = dataset_populator.get_history_dataset_details(history_id=history_id, dataset_id=output["id"])
-            hda_filenames.append(details["file_name"])
+            details = dataset_populator.get_history_dataset_details(
+                history_id=history_id, content_id=output["id"], wait=False
+            )
+            if not output_name.startswith("discovered_output"):
+                # We're not precreating discovered outputs on disk
+                hda_filenames.append(details["file_name"])
             dataset_populator.delete_dataset(
                 history_id=history_id, content_id=output["id"], purge=True, wait_for_purge=True
             )
@@ -33,23 +38,34 @@ def purge_while_job_running(dataset_populator: DatasetPopulator, sleep_before_pu
                 history_id=history_id, content_id=output_collection["id"]
             )
             for element in hdca["elements"]:
+                # Technically the static pair elements are already included in job["outputs"],
+                # but no harm purging them again here, in case we ever change that logic.
                 hda_id = element["object"]["id"]
                 dataset_populator.delete_dataset(
                     history_id=history_id, content_id=hda_id, purge=True, wait_for_purge=True
                 )
                 hda_ids.append(hda_id)
         dataset_populator.wait_for_job(job["id"], assert_ok=True)
-        # Now make sure we can't find the datasets
+        # Now make sure we can't find the datasets on disk
+        # We're not covering discovered datasets here, those can't be purged while the job is running.
         for hda_id in hda_ids:
             exception = None
             try:
                 dataset_populator.get_history_dataset_content(history_id=history_id, dataset_id=hda_id)
             except AssertionError as e:
                 exception = e
-            assert exception and "The dataset you are attempting to view has been purged" in str(exception)
-            output_details = dataset_populator.get_history_dataset_details(history_id=history_id, dataset_id=hda_id)
+            assert exception and "The dataset you are attempting to view has been purged" in str(exception), str(
+                exception
+            )
+            output_details = dataset_populator.get_history_dataset_details(
+                history_id=history_id, content_id=hda_id, wait=False
+            )
             # Make sure that we don't revert state while finishing job
-            assert output_details["purged"]
+            assert output_details["purged"], f"expected output '{output_name}' to be purged, but it is not purged."
+            assert not output_details.get("file_name")
+            assert (
+                output_details["file_size"] == 0
+            ), f"expected file_size for '{output_name}' to be 0, but it is {output_details['file_size']}."
         for file_name in hda_filenames:
             # Make sure job didn't push to object store
             assert not os.path.exists(file_name), f"Expected {file_name} to be purged."
diff --git a/test/integration/test_extended_metadata.py b/test/integration/test_extended_metadata.py
index cdb3ea1c6e2e..d267ed01c418 100644
--- a/test/integration/test_extended_metadata.py
+++ b/test/integration/test_extended_metadata.py
@@ -98,7 +98,9 @@ def test_fetch_data_library(self):
         assert dataset["created_from_basename"] == "4.bed"
 
     def test_purge_while_job_running(self):
-        purge_while_job_running(self.dataset_populator)
+        # pass extra_sleep, since templating the command line will fail if the output
+        # is deleted before remote_tool_eval runs.
+        purge_while_job_running(self.dataset_populator, extra_sleep=4)
 
 
 class TestExtendedMetadataDeferredIntegration(integration_util.IntegrationTestCase):
diff --git a/test/integration/test_pulsar_embedded_mq.py b/test/integration/test_pulsar_embedded_mq.py
index ef1979866409..19022a9b4b2f 100644
--- a/test/integration/test_pulsar_embedded_mq.py
+++ b/test/integration/test_pulsar_embedded_mq.py
@@ -82,6 +82,19 @@ def test_purge_while_job_running(self):
         purge_while_job_running(self.dataset_populator)
 
 
+class TestEmbeddedMessageQueuePulsarExtendedMetadataPurge(TestEmbeddedMessageQueuePulsarPurge):
+    """Describe a Galaxy test instance with embedded pulsar and extended metadata configured.
+
+    $ Setup RabbitMQ (e.g. https://www.rabbitmq.com/install-homebrew.html)
+    $ GALAXY_TEST_AMQP_URL='amqp://guest:guest@localhost:5672//' pytest -s test/integration/test_pulsar_embedded_mq.py
+    """
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["metadata_strategy"] = "extended"
+        _handle_galaxy_config_kwds(cls, config)
+
+
 class EmbeddedMessageQueuePulsarIntegrationInstance(integration_util.IntegrationInstance):
     """Describe a Galaxy test instance with embedded pulsar configured.