Handle MaxDiscoveredFilesExceededError gracefully for dynamic collections

Tools whose only discoverable outputs are dynamic collections (no primary
assign_primary_output dataset) were raising MaxDiscoveredFilesExceededError
from inside collect_dynamic_outputs, where a blanket `except Exception`
swallowed it via `log.exception`, shipping a traceback to Sentry and
leaving the outer handlers in set_metadata_portable / MinimalJobWrapper
without the error to record in job_messages.

Catch MaxDiscoveredFilesExceededError in collect_dynamic_outputs, mark the
collection as population-failed, register it with the job context so the
FAILED state is serialized through the extended-metadata export store, and
re-raise so the outer handlers can record a max_discovered_files entry in
job_messages. Log at warning level instead of error in the outer handlers.

In set_metadata_portable the pre-_finalize export_job was redundant
(immediately overwritten by _finalize) so drop it; add a single
post-_finalize export_job in the error path because _finalize cannot find
the job through included_datasets / included_collections when discovery
failed before anything was added to them.

Fixes #22394

lib/galaxy/job_execution/output_collect.py

@@ -29,6 +29,7 @@ from galaxy.model.store.discover import (
     discover_target_directory,
     DiscoveredFile,
     JsonCollectedDatasetMatch,
+    MaxDiscoveredFilesExceededError,
     MetadataSourceProvider as AbstractMetadataSourceProvider,
     ModelPersistenceContext,
     PermissionProvider as AbstractPermissionProvider,
@@ -213,6 +214,16 @@ def collect_dynamic_outputs(
                 change_datatype_actions=job_context.change_datatype_actions,
             )
             collection_builder.populate()
+        except MaxDiscoveredFilesExceededError:
+            # Mark the collection as population-failed so it is not left in NEW,
+            # then let the outer metadata/job handler record this in job_messages.
+            collection.handle_population_failed("Job generated more than the maximum number of output datasets.")
+            # Register the (failed) collection with the job context so that in
+            # the extended-metadata path the updated populated_state is
+            # serialized to the export store, and the host side imports the
+            # FAILED collection state rather than leaving it stuck in NEW.
+            job_context.add_dataset_collection(has_collection)
+            raise
         except Exception:
             log.exception("Problem gathering output collection.")
             collection.handle_population_failed("Problem building datasets for collection.")

lib/galaxy/jobs/__init__.py

@@ -2161,6 +2161,7 @@ class MinimalJobWrapper(HasResourceParameters):
             try:
                 self.discover_outputs(job, inp_data, out_data, out_collections, final_job_state=final_job_state)
             except (MaxDiscoveredFilesExceededError, JobOutputNameTooLongError) as e:
+                log.warning("Job %s failed during output discovery: %s", job.id, e)
                 final_job_state = job.states.ERROR
                 job.job_messages = [
                     {

lib/galaxy/metadata/set_metadata.py

@@ -223,6 +223,7 @@ def set_metadata_portable(
 
     export_store = None
     final_job_state = Job.states.OK
+    discovery_failed = False
     job_messages: list[AnyJobMessage] = []
     if extended_metadata_collection:
         tool_dict = metadata_params["tool"]
@@ -348,6 +349,8 @@ def set_metadata_portable(
             )
             collect_dynamic_outputs(job_context, output_collections)
         except (MaxDiscoveredFilesExceededError, JobOutputNameTooLongError) as e:
+            log.warning("Job failed during extended metadata output discovery: %s", e)
+            discovery_failed = True
             final_job_state = Job.states.ERROR
             job_messages.append(
                 {
@@ -537,6 +540,16 @@ def set_metadata_portable(
     if export_store:
         export_store.push_metadata_files()
         export_store._finalize()
+        if discovery_failed and job:
+            # _finalize() builds the jobs attrs file from included_datasets /
+            # included_collections via `creating_job_associations`. For tools
+            # whose only discoverable outputs are dynamic collections, nothing
+            # reaches either of those before discovery fails, so _finalize()
+            # writes an empty jobs list - and the ERROR state (plus
+            # job_messages) we set on the job is not persisted. Export the job
+            # once here so perform_import on the host side picks it up from
+            # the jobs attrs file.
+            export_store.export_job(job, include_job_data=False)
     write_job_metadata(tool_job_working_directory, job_metadata, set_meta, tool_provided_metadata)
 
 

test/integration/test_max_discovered_files.py

@@ -9,7 +9,7 @@ class TestMaxDiscoveredFiles(integration_util.IntegrationTestCase):
 
     dataset_populator: DatasetPopulator
     framework_tool_and_types = True
-    max_discovered_files = 9
+    max_discovered_files = 5
 
     @classmethod
     def handle_galaxy_config_kwds(cls, config):
@@ -34,6 +34,35 @@ class TestMaxDiscoveredFiles(integration_util.IntegrationTestCase):
                 in job_details["job_messages"][0]["desc"]
             )
 
+    def test_discover_dynamic_collection_only(self):
+        # Regression test for https://github.com/galaxyproject/galaxy/issues/22394.
+        with self.dataset_populator.test_history() as history_id:
+            response = self.dataset_populator.run_tool(
+                "collection_creates_dynamic_list_of_pairs",
+                inputs={"foo": "bar"},
+                history_id=history_id,
+            )
+            job_id = response["jobs"][0]["id"]
+            self.dataset_populator.wait_for_job(job_id, assert_ok=False)
+            job_details_response = self.dataset_populator.get_job_details(job_id, full=True)
+            job_details_response.raise_for_status()
+            job_details = job_details_response.json()
+            assert job_details["state"] == "error"
+            assert job_details["job_messages"], "expected a job_messages entry for max_discovered_files"
+            job_message = job_details["job_messages"][0]
+            assert job_message["type"] == "max_discovered_files"
+            assert (
+                f"Job generated more than maximum number ({self.max_discovered_files}) of output datasets"
+                in job_message["desc"]
+            )
+            # The dynamic output collection must be marked as failed, not left
+            # stuck in 'new'.
+            hdca_id = response["output_collections"][0]["id"]
+            collection_details = self.dataset_populator.get_history_collection_details(
+                history_id, content_id=hdca_id, assert_ok=False
+            )
+            assert collection_details["populated_state"] == "failed", collection_details
+
 
 class TestExtendedMetadataMaxDiscoveredFiles(TestMaxDiscoveredFiles):
     @classmethod