From 15f3bf02740e8987536377aac6556d880dcf4d13 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Sun, 15 Mar 2026 13:06:45 +0100 Subject: [PATCH] Track scheduling dependencies for workflow invocations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace fragile proxy signals in ready_to_schedule_more() with precise dependency tracking. When scheduling is delayed, record the concrete dependency (job ID, HDA ID, HDCA ID, or invocation step ID) and check whether it's been satisfied on the next iteration. - Add SchedulingDependency and DependencyType to modules.py; attach dependency info to DelayedWorkflowEvaluation where possible. - Collect dependencies on WorkflowProgress.scheduling_dependencies and expose them on the WorkflowInvocation via _scheduling_dependencies. - Propagate scheduling dependencies from subworkflow invokers to the parent progress so the scheduling manager can check them. - In scheduling_manager.py, maintain dependency_tracking_dict alongside update_time_tracking_dict; use batch queries for each dependency type (Job terminal states, HDA ready states, HDCA populated, step action set). - Keep history.update_time and step.update_time as fallbacks for signals not captured by explicit dependencies (e.g. PartialJobExecution). - Fix Job.set_final_state() to update workflow_invocation_step via implicit_collection_jobs link, not just direct job_id. - Fix maximum_workflow_invocation_duration enforcement for paused workflows: always allow scheduling when the duration is exceeded so invoke() can fail the invocation. | Delay scenario | Dependency tracked | Fallback signal | |---|---|---| | Job not finished | `JOB ` | `step.update_time` (via `Job.set_final_state`) | | HDA not ready (pending) | `HDA ` | `history.update_time` | | HDCA not populated | `HDCA ` | `history.update_time` | | Pause step waiting for review | `WORKFLOW_INVOCATION_STEP ` | `step.update_time` (action set via API) | | Dependent step not yet invoked | *(none — transient)* | next iteration re-evaluates | | Dependent step not yet scheduled | *(none — transient)* | next iteration re-evaluates | | Dependent step outputs delayed | *(inherits from root cause)* | inherits from root cause | | Tool inputs not ready | *(none — rare special tools)* | `history.update_time` | | PartialJobExecution | *(none — creates new HDAs)* | `history.update_time` | | Workflow output not yet created | *(none — transient)* | next iteration re-evaluates | | `maximum_workflow_invocation_duration` exceeded | *(duration check in `ready_to_schedule_more`)* | — | | Process restart | *(in-memory dicts empty)* | first iteration always schedules | --- lib/galaxy/model/__init__.py | 20 ++- lib/galaxy/workflow/modules.py | 28 +++- lib/galaxy/workflow/run.py | 43 ++++-- lib/galaxy/workflow/schedulers/__init__.py | 11 +- lib/galaxy/workflow/schedulers/core.py | 5 +- lib/galaxy/workflow/scheduling_manager.py | 149 ++++++++++++++++----- 6 files changed, 203 insertions(+), 53 deletions(-) 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