Merge branch 'release_25.0' into dev

Author: mvdbeek

lib/galaxy/model/__init__.py

@@ -7093,6 +7093,9 @@ def populated_optimized(self):
 
         return self._populated_optimized
 
+    def expire_populated_state(self):
+        required_object_session(self).expire(self, ("populated_state",))
+
     @property
     def allow_implicit_mapping(self):
         return self.collection_type != "record"

lib/galaxy/model/store/__init__.py

@@ -224,6 +224,15 @@ class SessionlessContext:
     def __init__(self) -> None:
         self.objects: dict[type, dict] = defaultdict(dict)
 
+    def execute(self, query: Any, *args, **kwargs) -> Any:
+        pass
+
+    def delete(self, obj: model.RepresentById) -> None:
+        self.objects[obj.__class__].pop(obj.id, None)
+
+    def scalars(self, query: Any, *args, **kwargs) -> Any:
+        pass
+
     def commit(self) -> None:
         pass
 

lib/galaxy/model/tags.py

@@ -3,6 +3,7 @@
 from typing import (
     Optional,
     TYPE_CHECKING,
+    Union,
 )
 
 from sqlalchemy.exc import IntegrityError
@@ -26,6 +27,7 @@
 
 if TYPE_CHECKING:
     from galaxy.model import User
+    from galaxy.model.store import SessionlessContext
 
 log = logging.getLogger(__name__)
 
@@ -43,7 +45,9 @@ class TagHandler:
     Manages CRUD operations related to tagging objects.
     """
 
-    def __init__(self, sa_session: scoped_session, galaxy_session: Optional[GalaxySession] = None) -> None:
+    def __init__(
+        self, sa_session: Union[scoped_session, "SessionlessContext"], galaxy_session: Optional[GalaxySession] = None
+    ) -> None:
         self.sa_session = sa_session
         # Minimum tag length.
         self.min_tag_len = 1
@@ -61,7 +65,10 @@ def __init__(self, sa_session: scoped_session, galaxy_session: Optional[GalaxySe
 
     def create_tag_handler_session(self, galaxy_session: Optional[GalaxySession]):
         # Creates a transient tag handler that avoids repeated flushes
-        return GalaxyTagHandlerSession(self.sa_session, galaxy_session=galaxy_session)
+        if isinstance(self.sa_session, scoped_session):
+            return GalaxyTagHandlerSession(self.sa_session, galaxy_session=galaxy_session)
+        else:
+            return self
 
     def add_tags_from_list(self, user, item, new_tags_list, flush=True):
         new_tags_set = set(new_tags_list)
@@ -324,7 +331,7 @@ def _get_tag(self, tag_name):
     def _create_tag_instance(self, tag_name):
         # For good performance caller should first check if there's already an appropriate tag
         tag = Tag(type=0, name=tag_name)
-        if not self.sa_session:
+        if not isinstance(self.sa_session, scoped_session):
             return tag
         Session = sessionmaker(self.sa_session.bind)
         with Session() as separate_session:

lib/galaxy/workflow/run.py

@@ -503,14 +503,18 @@ def replacement_for_connection(self, connection: "WorkflowStepConnection", is_da
                     # If we are not waiting for elements, there was some
                     # problem creating the collection. Collection will never
                     # be populated.
-                    raise modules.FailWorkflowEvaluation(
-                        why=InvocationFailureCollectionFailed(
-                            reason=FailureReason.collection_failed,
-                            hdca_id=replacement.id,
-                            workflow_step_id=connection.input_step_id,
-                            dependent_workflow_step_id=output_step_id,
+                    # We want to be certain of this however, so refresh attribute ...
+                    replacement.collection.expire_populated_state()
+                    # ... and repeat check to avoid race condition
+                    if not replacement.collection.populated:
+                        raise modules.FailWorkflowEvaluation(
+                            why=InvocationFailureCollectionFailed(
+                                reason=FailureReason.collection_failed,
+                                hdca_id=replacement.id,
+                                workflow_step_id=connection.input_step_id,
+                                dependent_workflow_step_id=output_step_id,
+                            )
                         )
-                    )
 
                 delayed_why = f"dependent collection [{replacement.id}] not yet populated with datasets"
                 raise modules.DelayedWorkflowEvaluation(why=delayed_why)

test/integration/test_pulsar_embedded_extended_metadata.py

@@ -31,5 +31,6 @@ def handle_galaxy_config_kwds(cls, config):
         "metadata_bam",
         "job_properties",
         "from_work_dir_glob",
+        "gx_group_tag",
     ]
 )