diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 7b3cb75ea03e..f9a520e8511e 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -2324,17 +2324,27 @@ def update_hdca_update_time_for_job(self, update_time, sa_session): def set_final_state(self, final_state): self.set_state(final_state) - # TODO: migrate to where-in subqueries? + sa_session = required_object_session(self) + update_time = now() + self.update_hdca_update_time_for_job(update_time=update_time, sa_session=sa_session) + params = {"job_id": self.id, "update_time": update_time} + # Update workflow_invocation_step for direct job_id statement = text(""" UPDATE workflow_invocation_step SET update_time = :update_time WHERE job_id = :job_id; """) - sa_session = required_object_session(self) - update_time = now() - self.update_hdca_update_time_for_job(update_time=update_time, sa_session=sa_session) - params = {"job_id": self.id, "update_time": update_time} sa_session.execute(statement, params) + # Also update via implicit_collection_jobs link + statement_icj = text(""" + UPDATE workflow_invocation_step + SET update_time = :update_time + WHERE implicit_collection_jobs_id IN ( + SELECT implicit_collection_jobs_id FROM implicit_collection_jobs_job_association + WHERE job_id = :job_id + ); + """) + sa_session.execute(statement_icj, params) def get_destination_configuration(self, dest_params, config, key, default=None): """Get a destination parameter that can be defaulted back diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index 997b3957da3d..feb4f9e8e84d 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -8,6 +8,8 @@ import re from collections import defaultdict from collections.abc import Iterable +from dataclasses import dataclass +from enum import Enum from typing import ( Any, cast, @@ -176,7 +178,7 @@ def to_cwl( if step: if not value.dataset.in_ready_state(): why = f"dataset [{value.id}] is needed for valueFrom expression and is non-ready" - raise DelayedWorkflowEvaluation(why=why) + raise DelayedWorkflowEvaluation(why=why, dependency=SchedulingDependency(DependencyType.HDA, value.id)) if not value.is_ok: raise FailWorkflowEvaluation( why=InvocationFailureDatasetFailed( @@ -905,6 +907,9 @@ def execute( subworkflow_invoker.invoke() subworkflow = subworkflow_invoker.workflow subworkflow_progress = subworkflow_invoker.progress + # Propagate scheduling dependencies from subworkflow to parent + if subworkflow_progress.scheduling_dependencies: + progress.scheduling_dependencies.update(subworkflow_progress.scheduling_dependencies) outputs = {} for workflow_output in subworkflow.workflow_outputs: workflow_output_label = ( @@ -1935,7 +1940,10 @@ def recover_mapping(self, invocation_step, progress): ) ) delayed_why = "workflow paused at this step waiting for review" - raise DelayedWorkflowEvaluation(why=delayed_why) + dependency = None + if invocation_step: + dependency = SchedulingDependency(DependencyType.WORKFLOW_INVOCATION_STEP, invocation_step.id) + raise DelayedWorkflowEvaluation(why=delayed_why, dependency=dependency) def do_invocation_step_action(self, step, action): """Update or set the workflow invocation state action - generic @@ -2782,9 +2790,23 @@ def from_workflow_step(self, trans, step: WorkflowStep, **kwargs) -> WorkflowMod module_factory = WorkflowModuleFactory(module_types) +class DependencyType(str, Enum): + JOB = "job" + HDA = "hda" + HDCA = "hdca" + WORKFLOW_INVOCATION_STEP = "workflow_invocation_step" + + +@dataclass(frozen=True) +class SchedulingDependency: + dependency_type: DependencyType + id: int + + class DelayedWorkflowEvaluation(Exception): - def __init__(self, why=None): + def __init__(self, why=None, dependency: Optional[SchedulingDependency] = None): self.why = why + self.dependency = dependency class CancelWorkflowEvaluation(Exception): diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index db0442ec4e09..ac7724eb3a2b 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -68,8 +68,11 @@ def schedule( workflow: "Workflow", workflow_run_config: WorkflowRunConfig, workflow_invocation: WorkflowInvocation, -) -> tuple[WorkflowOutputsType, WorkflowInvocation]: - return __invoke(trans, workflow, workflow_run_config, workflow_invocation) +) -> set[modules.SchedulingDependency]: + _outputs, _workflow_invocation, scheduling_dependencies = __invoke( + trans, workflow, workflow_run_config, workflow_invocation + ) + return scheduling_dependencies def __invoke( @@ -78,7 +81,7 @@ def __invoke( workflow_run_config: WorkflowRunConfig, workflow_invocation: Optional[WorkflowInvocation] = None, populate_state: bool = False, -) -> tuple[WorkflowOutputsType, WorkflowInvocation]: +) -> tuple[WorkflowOutputsType, WorkflowInvocation, set[modules.SchedulingDependency]]: """Run the supplied workflow in the supplied target_history.""" if populate_state: modules.populate_module_and_state( @@ -118,11 +121,13 @@ def __invoke( workflow_invocation.fail() workflow_invocation.add_message(failure) + scheduling_dependencies = invoker.progress.scheduling_dependencies + # Be sure to update state of workflow_invocation. trans.sa_session.add(workflow_invocation) trans.sa_session.commit() - return outputs, workflow_invocation + return outputs, workflow_invocation, scheduling_dependencies def queue_invoke( @@ -259,6 +264,8 @@ def invoke(self) -> dict[int, Any]: except modules.DelayedWorkflowEvaluation as de: step_delayed = delayed_steps = True self.progress.mark_step_outputs_delayed(step, why=de.why) + if de.dependency: + self.progress.scheduling_dependencies.add(de.dependency) except Exception as e: log_function = log.error failure_details = [] @@ -349,7 +356,10 @@ def __check_implicitly_dependent_step(self, output_id: int, step_id: int): delayed_why = ( f"depends on step [{output_id}] but one or more jobs created from that step have not finished yet" ) - raise modules.DelayedWorkflowEvaluation(why=delayed_why) + raise modules.DelayedWorkflowEvaluation( + why=delayed_why, + dependency=modules.SchedulingDependency(modules.DependencyType.JOB, job.id), + ) if job.state != job.states.OK: raise modules.FailWorkflowEvaluation( @@ -403,6 +413,7 @@ def __init__( when_values=None, ) -> None: self.outputs: dict[int, Any] = {} + self.scheduling_dependencies: set[modules.SchedulingDependency] = set() self.module_injector = module_injector self.workflow_invocation = workflow_invocation self.inputs_by_step_id = inputs_by_step_id @@ -550,7 +561,10 @@ def replacement_for_connection(self, connection: "WorkflowStepConnection", is_da ) delayed_why = f"dependent collection [{replacement.id}] not yet populated with datasets" - raise modules.DelayedWorkflowEvaluation(why=delayed_why) + raise modules.DelayedWorkflowEvaluation( + why=delayed_why, + dependency=modules.SchedulingDependency(modules.DependencyType.HDCA, replacement.id), + ) if isinstance(replacement, model.DatasetCollection): raise NotImplementedError @@ -559,7 +573,9 @@ def replacement_for_connection(self, connection: "WorkflowStepConnection", is_da ): if isinstance(replacement, model.HistoryDatasetAssociation): if replacement.is_pending: - raise modules.DelayedWorkflowEvaluation() + raise modules.DelayedWorkflowEvaluation( + dependency=modules.SchedulingDependency(modules.DependencyType.HDA, replacement.id) + ) if not replacement.is_ok: raise modules.FailWorkflowEvaluation( why=InvocationFailureDatasetFailed( @@ -571,11 +587,15 @@ def replacement_for_connection(self, connection: "WorkflowStepConnection", is_da ) else: if not replacement.collection.populated: - raise modules.DelayedWorkflowEvaluation() + raise modules.DelayedWorkflowEvaluation( + dependency=modules.SchedulingDependency(modules.DependencyType.HDCA, replacement.id) + ) pending = False + pending_dataset_instance = None for dataset_instance in replacement.dataset_instances: if dataset_instance.is_pending: pending = True + pending_dataset_instance = dataset_instance elif not dataset_instance.is_ok: raise modules.FailWorkflowEvaluation( why=InvocationFailureDatasetFailed( @@ -586,7 +606,10 @@ def replacement_for_connection(self, connection: "WorkflowStepConnection", is_da ) ) if pending: - raise modules.DelayedWorkflowEvaluation() + assert pending_dataset_instance is not None + raise modules.DelayedWorkflowEvaluation( + dependency=modules.SchedulingDependency(modules.DependencyType.HDA, pending_dataset_instance.id) + ) return replacement @@ -831,6 +854,8 @@ def _recover_mapping(self, step_invocation: WorkflowInvocationStep) -> None: step_invocation.workflow_step.module.recover_mapping(step_invocation, self) except modules.DelayedWorkflowEvaluation as de: self.mark_step_outputs_delayed(step_invocation.workflow_step, de.why) + if de.dependency: + self.scheduling_dependencies.add(de.dependency) __all__ = ("queue_invoke", "WorkflowRunConfig") diff --git a/lib/galaxy/workflow/schedulers/__init__.py b/lib/galaxy/workflow/schedulers/__init__.py index 70683cc0d6c5..eaf76aa8e07f 100644 --- a/lib/galaxy/workflow/schedulers/__init__.py +++ b/lib/galaxy/workflow/schedulers/__init__.py @@ -7,6 +7,10 @@ ABCMeta, abstractmethod, ) +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from galaxy.workflow.modules import SchedulingDependency class WorkflowSchedulingPlugin(metaclass=ABCMeta): @@ -35,8 +39,7 @@ def shutdown(self): class ActiveWorkflowSchedulingPlugin(WorkflowSchedulingPlugin, metaclass=ABCMeta): @abstractmethod - def schedule(self, workflow_invocation): - """Optionally return one or more commands to instrument job. These - commands will be executed on the compute server prior to the job - running. + def schedule(self, workflow_invocation) -> "set[SchedulingDependency]": + """Schedule the workflow invocation and return any scheduling + dependencies that should be tracked for the next iteration. """ diff --git a/lib/galaxy/workflow/schedulers/core.py b/lib/galaxy/workflow/schedulers/core.py index 35409c7ce9f7..1cd946ecda45 100644 --- a/lib/galaxy/workflow/schedulers/core.py +++ b/lib/galaxy/workflow/schedulers/core.py @@ -14,6 +14,7 @@ if TYPE_CHECKING: from galaxy.model import WorkflowInvocation + from galaxy.workflow.modules import SchedulingDependency log = logging.getLogger(__name__) @@ -31,14 +32,14 @@ def startup(self, app): def shutdown(self): pass - def schedule(self, workflow_invocation: "WorkflowInvocation") -> None: + def schedule(self, workflow_invocation: "WorkflowInvocation") -> "set[SchedulingDependency]": workflow = workflow_invocation.workflow history = workflow_invocation.history request_context = context.WorkRequestContext( app=self.app, history=history, user=history.user ) # trans-like object not tied to a web-thread. workflow_run_config = run_request.workflow_request_to_run_config(workflow_invocation) - run.schedule( + return run.schedule( trans=request_context, workflow=workflow, workflow_run_config=workflow_run_config, diff --git a/lib/galaxy/workflow/scheduling_manager.py b/lib/galaxy/workflow/scheduling_manager.py index 233decaa89b7..b7299a2ca475 100644 --- a/lib/galaxy/workflow/scheduling_manager.py +++ b/lib/galaxy/workflow/scheduling_manager.py @@ -1,8 +1,5 @@ import os -from datetime import ( - datetime, - timedelta, -) +from datetime import datetime from functools import partial from typing import ( Optional, @@ -10,6 +7,11 @@ Union, ) +from sqlalchemy import ( + case, + func, + select, +) from sqlalchemy.orm import Session import galaxy.workflow.schedulers @@ -36,6 +38,10 @@ from galaxy.util.xml_macros import load from galaxy.web_stack.handlers import ConfiguresHandlers from galaxy.web_stack.message import WorkflowSchedulingMessage +from galaxy.workflow.modules import ( + DependencyType, + SchedulingDependency, +) if TYPE_CHECKING: from galaxy.structured_app import MinimalManagerApp @@ -49,7 +55,6 @@ DEFAULT_SCHEDULER_ID = "default" # well actually this should be called DEFAULT_DEFAULT_SCHEDULER_ID... DEFAULT_SCHEDULER_PLUGIN_TYPE = "core" -DEFAULT_SCHEDULER_BACKFILL_SECONDS = int(os.getenv("GALAXY_SCHEDULER_BACKFILL_SECONDS", 300)) EXCEPTION_MESSAGE_SHUTDOWN = "Exception raised while attempting to shutdown workflow scheduler." EXCEPTION_MESSAGE_NO_SCHEDULERS = "Failed to defined workflow schedulers - no workflow schedulers defined." @@ -317,12 +322,7 @@ def __init__(self, app: "MinimalManagerApp", workflow_scheduling_manager: Workfl ) self.invocation_grabber = None self.update_time_tracking_dict: dict[int, datetime] = {} - backfill_seconds = ( - min(app.config.maximum_workflow_invocation_duration, DEFAULT_SCHEDULER_BACKFILL_SECONDS) - if app.config.maximum_workflow_invocation_duration > 0 - else DEFAULT_SCHEDULER_BACKFILL_SECONDS - ) - self.timedelta = timedelta(seconds=backfill_seconds) + self.dependency_tracking_dict: dict[int, set[SchedulingDependency]] = {} self_handler_tags = set(self.app.job_config.self_handler_tags) self_handler_tags.add(self.workflow_scheduling_manager.default_handler_id) handler_assignment_method = InvocationGrabber.get_grabbable_handler_assignment_method( @@ -338,27 +338,109 @@ def __init__(self, app: "MinimalManagerApp", workflow_scheduling_manager: Workfl ) def ready_to_schedule_more(self, invocation: model.WorkflowInvocation): - # Improve reactivity of scheduling using the history update_time as a heuristic. - # If there wasn't a change in the history we're unlikely to be able to make more progress. + # After process restart, in-memory dicts are empty — always schedule + # the first iteration so dependencies get (re-)captured. if invocation.id not in self.update_time_tracking_dict: return True - else: - last_schedule_time = self.update_time_tracking_dict[invocation.id] - last_history_update_time = invocation.history.update_time - do_schedule = last_history_update_time > last_schedule_time - if not do_schedule and ( - invocation_step_update_time := invocation.get_last_workflow_invocation_step_update_time() - ): - do_schedule = invocation_step_update_time > last_schedule_time - if not do_schedule and (datetime.now() - last_schedule_time) > self.timedelta: - # If we haven't scheduled in a while, schedule anyway. - log.debug( - "Scheduling workflow invocation [%s] after %s seconds without scheduling.", - invocation.id, - (datetime.now() - last_schedule_time).total_seconds(), + + # Always allow scheduling if maximum duration has been exceeded, + # so invoke() can fail the invocation with the appropriate state. + maximum_duration = getattr(self.app.config, "maximum_workflow_invocation_duration", -1) + if maximum_duration > 0 and invocation.seconds_since_created > maximum_duration: + return True + + last_schedule_time = self.update_time_tracking_dict[invocation.id] + + # Check tracked dependencies — most precise signal + dependencies = self.dependency_tracking_dict.get(invocation.id) + if dependencies and self._any_dependency_satisfied(dependencies, invocation): + return True + + # Fallback: check history update_time (covers HDA/HDCA inserts from DB triggers, + # also catches PartialJobExecution which creates new HDAs without recording a dependency) + if invocation.history.update_time > last_schedule_time: + return True + + # Fallback: check workflow_invocation_step.update_time (catches pause step + # actions set via API, and job completions via Job.set_final_state) + if invocation_step_update_time := invocation.get_last_workflow_invocation_step_update_time(): + if invocation_step_update_time > last_schedule_time: + return True + + return False + + def _any_dependency_satisfied( + self, dependencies: set[SchedulingDependency], invocation: model.WorkflowInvocation + ) -> bool: + session = Session.object_session(invocation) + if session is None: + return True + + # Group dependencies by type for batch queries + job_ids = {d.id for d in dependencies if d.dependency_type == DependencyType.JOB} + hda_ids = {d.id for d in dependencies if d.dependency_type == DependencyType.HDA} + hdca_ids = {d.id for d in dependencies if d.dependency_type == DependencyType.HDCA} + step_ids = {d.id for d in dependencies if d.dependency_type == DependencyType.WORKFLOW_INVOCATION_STEP} + + if job_ids: + terminal_count = session.execute( + select(func.count()) + .select_from(model.Job) + .where( + model.Job.id.in_(job_ids), + model.Job.state.in_(model.Job.terminal_states), ) - do_schedule = True - return do_schedule + ).scalar() + if terminal_count: + return True + + if hda_ids: + # HDA.state = HDA._state if set, else Dataset.state + ready_count = session.execute( + select(func.count()) + .select_from(model.HistoryDatasetAssociation) + .join(model.Dataset) + .where( + model.HistoryDatasetAssociation.id.in_(hda_ids), + case( + ( + model.HistoryDatasetAssociation._state.isnot(None), + model.HistoryDatasetAssociation._state, + ), + else_=model.Dataset.state, + ).notin_(model.Dataset.non_ready_states), + ) + ).scalar() + if ready_count: + return True + + if hdca_ids: + populated_count = session.execute( + select(func.count()) + .select_from(model.HistoryDatasetCollectionAssociation) + .join(model.DatasetCollection) + .where( + model.HistoryDatasetCollectionAssociation.id.in_(hdca_ids), + model.DatasetCollection.populated_state == model.DatasetCollection.populated_states.OK, + ) + ).scalar() + if populated_count: + return True + + if step_ids: + # Check if any tracked step has had its action set (e.g. pause step reviewed) + action_count = session.execute( + select(func.count()) + .select_from(model.WorkflowInvocationStep) + .where( + model.WorkflowInvocationStep.id.in_(step_ids), + model.WorkflowInvocationStep.action.isnot(None), + ) + ).scalar() + if action_count: + return True + + return False def __monitor(self): to_monitor = self.workflow_scheduling_manager.active_workflow_schedulers @@ -445,10 +527,12 @@ def __attempt_schedule(self, invocation_id, workflow_scheduler): workflow_invocation.mark_cancelled() session.commit() self.update_time_tracking_dict.pop(invocation_id, None) + self.dependency_tracking_dict.pop(invocation_id, None) return False if not workflow_invocation or not workflow_invocation.active: self.update_time_tracking_dict.pop(invocation_id, None) + self.dependency_tracking_dict.pop(invocation_id, None) return False # This ensures we're only ever working on the 'first' active @@ -460,10 +544,15 @@ def __attempt_schedule(self, invocation_id, workflow_scheduler): return False if self.ready_to_schedule_more(workflow_invocation): self.update_time_tracking_dict[invocation_id] = datetime.now() - workflow_scheduler.schedule(workflow_invocation) + scheduling_deps = workflow_scheduler.schedule(workflow_invocation) + if scheduling_deps: + self.dependency_tracking_dict[invocation_id] = scheduling_deps + else: + self.dependency_tracking_dict.pop(invocation_id, None) log.debug("Workflow invocation [%s] scheduled", invocation_id) except Exception: self.update_time_tracking_dict.pop(invocation_id, None) + self.dependency_tracking_dict.pop(invocation_id, None) # TODO: eventually fail this - or fail it right away? log.exception("Exception raised while attempting to schedule workflow request.") return False