Fail job gracefully when output name exceeds 255 character DB limit

When discovered dataset designations are very long (e.g., taxonomic names),
the job output association name (__new_primary_file_{name}|{designation}__)
can exceed the VARCHAR(255) column limit, causing StringDataRightTruncation
which corrupts the SQLAlchemy session and triggers PendingRollbackError.

Add name length validation in Job.add_output_dataset and related methods
that raises ValueError before the DB INSERT. Catch the error in both the
normal and extended metadata code paths to fail the job with a clear error
message instead of crashing the transaction session.

Fixes https://github.com/galaxyproject/galaxy/issues/22045

lib/galaxy/job_execution/output_collect.py

@@ -337,6 +337,13 @@ class SessionlessJobContext(SessionlessModelPersistenceContext, BaseJobContext):
             self.export_store.collection_datasets.add(collection_dataset.id)
 
     def add_output_dataset_association(self, name, dataset_instance):
+        from galaxy.model import JOB_IO_NAME_MAX_LENGTH
+
+        if name and len(name) > JOB_IO_NAME_MAX_LENGTH:
+            raise ValueError(
+                f"Tool produced an output name that exceeds the {JOB_IO_NAME_MAX_LENGTH} character name length limit "
+                f"(got {len(name)} characters), tool is likely broken"
+            )
         assert self.export_store
         self.export_store.add_job_output_dataset_associations(self.get_job_id(), name, dataset_instance)
 

lib/galaxy/jobs/__init__.py

@@ -2143,10 +2143,10 @@ class MinimalJobWrapper(HasResourceParameters):
             except store.FileTracebackException as e:
                 job.traceback = e.traceback
                 log.exception(f"Problem generating command line for Job {job.id}.\n{job.traceback}")
-                raise
-            except Exception:
+                return fail(str(e), exception=e)
+            except Exception as e:
                 log.exception(f"problem importing job outputs. stdout [{job.stdout}] stderr [{job.stderr}]")
-                raise
+                return fail(str(e), exception=e)
         else:
             if self.tool.version_string_cmd:
                 version_filename = self.get_version_string_path()
@@ -2159,7 +2159,7 @@ class MinimalJobWrapper(HasResourceParameters):
             # importing metadata will discover outputs if extended metadata
             try:
                 self.discover_outputs(job, inp_data, out_data, out_collections, final_job_state=final_job_state)
-            except MaxDiscoveredFilesExceededError as e:
+            except (MaxDiscoveredFilesExceededError, ValueError) as e:
                 final_job_state = job.states.ERROR
                 job.job_messages = [
                     {

lib/galaxy/metadata/set_metadata.py

@@ -346,7 +346,7 @@ def set_metadata_portable(
                 input_ext=input_ext,
             )
             collect_dynamic_outputs(job_context, output_collections)
-        except MaxDiscoveredFilesExceededError as e:
+        except (MaxDiscoveredFilesExceededError, ValueError) as e:
             final_job_state = Job.states.ERROR
             job_messages.append(
                 {

lib/galaxy/model/__init__.py

@@ -1579,6 +1579,9 @@ class IoDicts(NamedTuple):
     out_collections: OutCollectionsDictT
 
 
+JOB_IO_NAME_MAX_LENGTH = 255
+
+
 class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
     """
     A job represents a request to run a tool given input datasets, tool
@@ -2017,7 +2020,16 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
     def add_parameter(self, name, value):
         self.parameters.append(JobParameter(name, value))
 
+    @staticmethod
+    def _check_name_length(name: str) -> None:
+        if name and len(name) > JOB_IO_NAME_MAX_LENGTH:
+            raise ValueError(
+                f"Tool produced an output name that exceeds the {JOB_IO_NAME_MAX_LENGTH} character name length limit "
+                f"(got {len(name)} characters), tool is likely broken"
+            )
+
     def add_input_dataset(self, name, dataset=None, dataset_id=None, adapter_json=None):
+        self._check_name_length(name)
         assoc = JobToInputDatasetAssociation(name, dataset, adapter_json)
         if dataset is None and dataset_id is not None:
             assoc.dataset_id = dataset_id
@@ -2025,6 +2037,7 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
         self.input_datasets.append(assoc)
 
     def add_output_dataset(self, name: str, dataset: "HistoryDatasetAssociation"):
+        self._check_name_length(name)
         joda = JobToOutputDatasetAssociation(name, dataset)
         if dataset.dataset.job is None:
             # Only set job if dataset doesn't already have associated job.
@@ -2034,29 +2047,35 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
         self.output_datasets.append(joda)
 
     def add_input_dataset_collection(self, name, dataset_collection, adapter_json=None):
+        self._check_name_length(name)
         self.input_dataset_collections.append(
             JobToInputDatasetCollectionAssociation(name, dataset_collection, adapter_json)
         )
 
     def add_input_dataset_collection_element(self, name, dataset_collection_element, adapter_json=None):
+        self._check_name_length(name)
         self.input_dataset_collection_elements.append(
             JobToInputDatasetCollectionElementAssociation(name, dataset_collection_element, adapter_json)
         )
 
     def add_output_dataset_collection(self, name, dataset_collection_instance):
+        self._check_name_length(name)
         self.output_dataset_collection_instances.append(
             JobToOutputDatasetCollectionAssociation(name, dataset_collection_instance)
         )
 
     def add_implicit_output_dataset_collection(self, name, dataset_collection):
+        self._check_name_length(name)
         self.output_dataset_collections.append(
             JobToImplicitOutputDatasetCollectionAssociation(name, dataset_collection)
         )
 
     def add_input_library_dataset(self, name, dataset):
+        self._check_name_length(name)
         self.input_library_datasets.append(JobToInputLibraryDatasetAssociation(name, dataset))
 
     def add_output_library_dataset(self, name, dataset):
+        self._check_name_length(name)
         self.output_library_datasets.append(JobToOutputLibraryDatasetAssociation(name, dataset))
 
     def add_post_job_action(self, pja):

lib/galaxy/tools/__init__.py

@@ -944,6 +944,7 @@ class JobContext(BaseJobContext):
         return hdca
 
     def add_output_dataset_association(self, name, dataset):
+        Job._check_name_length(name)
         assoc = JobToOutputDatasetAssociation(name, dataset)
         assoc.job = self.job
         self.sa_session.add(assoc)

test/functional/tools/discover_long_name.xml

@@ -0,0 +1,13 @@
+<tool id="discover_long_name" name="discover_long_name" version="0.1.0" profile="20.09">
+    <command><![CDATA[
+echo "data" > '${output_name}.txt'
+    ]]></command>
+    <inputs>
+        <param name="output_name" type="text" label="Output name" />
+    </inputs>
+    <outputs>
+        <data format="txt" name="output">
+            <discover_datasets pattern="__name_and_ext__" visible="true" />
+        </data>
+    </outputs>
+</tool>

test/functional/tools/sample_tool_conf.xml

@@ -219,6 +219,7 @@
   <tool file="collection_paired_or_unpaired.xml" />
   <tool file="collection_list_paired_or_unpaired.xml" />
   <tool file="discover_sort_by.xml" />
+  <tool file="discover_long_name.xml" />
   <tool file="discover_default_ext.xml" />
   <tool file="expression_forty_two.xml" />
   <tool file="expression_pick_larger_file.xml" />

test/integration/test_long_output_name.py

@@ -0,0 +1,60 @@
+"""Integration tests for job output name length validation."""
+
+from galaxy_test.base.populators import DatasetPopulator
+from galaxy_test.driver import integration_util
+
+
+class TestLongOutputName(integration_util.IntegrationTestCase):
+    """Test that jobs with output names exceeding 255 chars fail gracefully."""
+
+    dataset_populator: DatasetPopulator
+    framework_tool_and_types = True
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    def test_short_output_name_succeeds(self):
+        with self.dataset_populator.test_history() as history_id:
+            response = self.dataset_populator.run_tool(
+                "discover_long_name",
+                {"output_name": "normal_name"},
+                history_id,
+            )
+            job_id = response["jobs"][0]["id"]
+            self.dataset_populator.wait_for_job(job_id, assert_ok=True)
+
+    def _assert_long_name_error(self, job_details):
+        assert (
+            job_details["state"] == "error"
+        ), f"Expected error state, got: {job_details['state']}, details: {job_details}"
+        error_message = job_details.get("info", "")
+        job_messages = job_details.get("job_messages", [])
+        if job_messages:
+            error_message += " " + job_messages[0].get("desc", "")
+        assert (
+            "255 character" in error_message
+        ), f"Expected '255 character' in error details, got full details: {job_details}"
+
+    def test_long_output_name_fails_gracefully(self):
+        with self.dataset_populator.test_history() as history_id:
+            # Use 240 chars: fits within ext4's 255-byte filename limit (240 + 4 for ".txt" = 244),
+            # but the association name __new_primary_file_output|<240 chars>__ = 269 chars, exceeding 255.
+            long_name = "a" * 240
+            response = self.dataset_populator.run_tool(
+                "discover_long_name",
+                {"output_name": long_name},
+                history_id,
+            )
+            job_id = response["jobs"][0]["id"]
+            self.dataset_populator.wait_for_job(job_id, assert_ok=False)
+            job_details = self.dataset_populator.get_job_details(job_id, full=True).json()
+            self._assert_long_name_error(job_details)
+
+
+class TestExtendedMetadataLongOutputName(TestLongOutputName):
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        config["metadata_strategy"] = "extended"
+        config["object_store_store_by"] = "uuid"
+        config["retry_metadata_internally"] = False