mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-29 17:42:12 +08:00
Merge pull request #21982 from mvdbeek/improve_workflow_failure_message_when_input_deleted
[26.0] Fix dict leaking to process_dataset() during workflow execution
This commit is contained in:
@@ -16375,6 +16375,39 @@ export interface components {
|
||||
*/
|
||||
workflow_step_index_path?: number[] | null;
|
||||
};
|
||||
/** InvocationFailureStepInputDeletedResponse */
|
||||
InvocationFailureStepInputDeletedResponse: {
|
||||
/**
|
||||
* Details
|
||||
* @description Details about which input referenced a deleted dataset.
|
||||
*/
|
||||
details: string;
|
||||
/**
|
||||
* HistoryDatasetAssociation ID
|
||||
* @description HistoryDatasetAssociation ID of the deleted dataset, if applicable.
|
||||
*/
|
||||
hda_id?: string | null;
|
||||
/**
|
||||
* HistoryDatasetCollectionAssociation ID
|
||||
* @description HistoryDatasetCollectionAssociation ID of the deleted collection, if applicable.
|
||||
*/
|
||||
hdca_id?: string | null;
|
||||
/**
|
||||
* @description discriminator enum property added by openapi-typescript
|
||||
* @enum {string}
|
||||
*/
|
||||
reason: "step_input_deleted";
|
||||
/**
|
||||
* Workflow Step Id
|
||||
* @description Workflow step id of step that failed.
|
||||
*/
|
||||
workflow_step_id: number;
|
||||
/**
|
||||
* Workflow Step Index Path
|
||||
* @description Path of workflow step IDs from parent workflow through subworkflows (excludes the failing step itself).
|
||||
*/
|
||||
workflow_step_index_path?: number[] | null;
|
||||
};
|
||||
/** InvocationFailureWhenNotBooleanResponse */
|
||||
InvocationFailureWhenNotBooleanResponse: {
|
||||
/**
|
||||
@@ -16499,7 +16532,8 @@ export interface components {
|
||||
| components["schemas"]["InvocationFailureWhenNotBooleanResponse"]
|
||||
| components["schemas"]["InvocationUnexpectedFailureResponse"]
|
||||
| components["schemas"]["InvocationEvaluationWarningWorkflowOutputNotFoundResponse"]
|
||||
| components["schemas"]["InvocationFailureWorkflowParameterInvalidResponse"];
|
||||
| components["schemas"]["InvocationFailureWorkflowParameterInvalidResponse"]
|
||||
| components["schemas"]["InvocationFailureStepInputDeletedResponse"];
|
||||
/** InvocationOutput */
|
||||
InvocationOutput: {
|
||||
/**
|
||||
|
||||
@@ -212,6 +212,12 @@ const messageInfo = computed(() => {
|
||||
suffix: `failed validation: ${invocationMessage.details}`,
|
||||
hasStepPath: true,
|
||||
};
|
||||
} else if (reason === "step_input_deleted") {
|
||||
return {
|
||||
prefix: `${FAIL_FRAGMENT}a dataset input on`,
|
||||
suffix: "has been deleted.",
|
||||
hasStepPath: true,
|
||||
};
|
||||
} else {
|
||||
return { text: reason, hasStepPath: false };
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@ export const INVOCATION_MSG_LEVEL = {
|
||||
unexpected_failure: "error",
|
||||
workflow_output_not_found: "warning",
|
||||
workflow_parameter_invalid: "error",
|
||||
step_input_deleted: "error",
|
||||
} as const satisfies Readonly<Record<string, "cancel" | "error" | "warning">>;
|
||||
|
||||
function countStates(jobSummary: InvocationJobsSummary | StepJobSummary | null, queryStates: string[]): number {
|
||||
|
||||
@@ -89,6 +89,7 @@ class FailureReason(str, Enum):
|
||||
when_not_boolean = "when_not_boolean"
|
||||
unexpected_failure = "unexpected_failure"
|
||||
workflow_parameter_invalid = "workflow_parameter_invalid"
|
||||
step_input_deleted = "step_input_deleted"
|
||||
|
||||
|
||||
# The reasons below are attached to the invocation and user-actionable.
|
||||
@@ -103,6 +104,7 @@ FAILURE_REASONS_EXPECTED = (
|
||||
FailureReason.job_failed,
|
||||
FailureReason.output_not_found,
|
||||
FailureReason.when_not_boolean,
|
||||
FailureReason.step_input_deleted,
|
||||
)
|
||||
|
||||
|
||||
@@ -236,6 +238,21 @@ class GenericInvocationFailureWorkflowParameterInvalid(InvocationFailureMessageB
|
||||
details: str = Field(..., description="Message raised by validator")
|
||||
|
||||
|
||||
class GenericInvocationFailureStepInputDeleted(InvocationFailureMessageBase[DatabaseIdT], Generic[DatabaseIdT]):
|
||||
reason: Literal[FailureReason.step_input_deleted]
|
||||
hda_id: Optional[DatabaseIdT] = Field(
|
||||
None,
|
||||
title="HistoryDatasetAssociation ID",
|
||||
description="HistoryDatasetAssociation ID of the deleted dataset, if applicable.",
|
||||
)
|
||||
hdca_id: Optional[DatabaseIdT] = Field(
|
||||
None,
|
||||
title="HistoryDatasetCollectionAssociation ID",
|
||||
description="HistoryDatasetCollectionAssociation ID of the deleted collection, if applicable.",
|
||||
)
|
||||
details: str = Field(..., description="Details about which input referenced a deleted dataset.")
|
||||
|
||||
|
||||
InvocationCancellationReviewFailed = GenericInvocationCancellationReviewFailed[int]
|
||||
InvocationCancellationHistoryDeleted = GenericInvocationCancellationHistoryDeleted[int]
|
||||
InvocationCancellationUserRequest = GenericInvocationCancellationUserRequest[int]
|
||||
@@ -248,6 +265,7 @@ InvocationFailureWhenNotBoolean = GenericInvocationFailureWhenNotBoolean[int]
|
||||
InvocationUnexpectedFailure = GenericInvocationUnexpectedFailure[int]
|
||||
InvocationWarningWorkflowOutputNotFound = GenericInvocationEvaluationWarningWorkflowOutputNotFound[int]
|
||||
InvocationFailureWorkflowParameterInvalid = GenericInvocationFailureWorkflowParameterInvalid[int]
|
||||
InvocationFailureStepInputDeleted = GenericInvocationFailureStepInputDeleted[int]
|
||||
|
||||
InvocationMessageUnion = Union[
|
||||
InvocationCancellationReviewFailed,
|
||||
@@ -262,6 +280,7 @@ InvocationMessageUnion = Union[
|
||||
InvocationUnexpectedFailure,
|
||||
InvocationWarningWorkflowOutputNotFound,
|
||||
InvocationFailureWorkflowParameterInvalid,
|
||||
InvocationFailureStepInputDeleted,
|
||||
]
|
||||
|
||||
InvocationCancellationReviewFailedResponseModel = GenericInvocationCancellationReviewFailed[EncodedDatabaseIdField]
|
||||
@@ -282,6 +301,7 @@ InvocationWarningWorkflowOutputNotFoundResponseModel = GenericInvocationEvaluati
|
||||
InvocationFailureWorkflowParameterInvalidResponseModel = GenericInvocationFailureWorkflowParameterInvalid[
|
||||
EncodedDatabaseIdField
|
||||
]
|
||||
InvocationFailureStepInputDeletedResponseModel = GenericInvocationFailureStepInputDeleted[EncodedDatabaseIdField]
|
||||
|
||||
_InvocationMessageResponseUnion = Annotated[
|
||||
Union[
|
||||
@@ -297,6 +317,7 @@ _InvocationMessageResponseUnion = Annotated[
|
||||
InvocationUnexpectedFailureResponseModel,
|
||||
InvocationWarningWorkflowOutputNotFoundResponseModel,
|
||||
InvocationFailureWorkflowParameterInvalidResponseModel,
|
||||
InvocationFailureStepInputDeletedResponseModel,
|
||||
],
|
||||
Field(discriminator="reason"),
|
||||
]
|
||||
|
||||
@@ -2247,7 +2247,9 @@ class DataToolParameter(BaseDataToolParameter):
|
||||
raise ParameterValueError("Collection element in unexpected state", self.name)
|
||||
if isinstance(value_to_check, DatasetInstance):
|
||||
if value_to_check.deleted:
|
||||
raise ParameterValueError("the previously selected dataset has been deleted.", self.name)
|
||||
raise ParameterValueError(
|
||||
"the previously selected dataset has been deleted.", self.name, value_to_check
|
||||
)
|
||||
elif value_to_check.dataset and value_to_check.dataset.state in [
|
||||
Dataset.states.ERROR,
|
||||
Dataset.states.DISCARDED,
|
||||
@@ -2260,7 +2262,11 @@ class DataToolParameter(BaseDataToolParameter):
|
||||
value_to_check.implicit_conversion = True # type: ignore[attr-defined]
|
||||
elif isinstance(value_to_check, HistoryDatasetCollectionAssociation):
|
||||
if value_to_check.deleted:
|
||||
raise ParameterValueError("the previously selected dataset collection has been deleted.", self.name)
|
||||
raise ParameterValueError(
|
||||
"the previously selected dataset collection has been deleted.",
|
||||
self.name,
|
||||
value_to_check,
|
||||
)
|
||||
value_to_check = value_to_check.collection
|
||||
if isinstance(value_to_check, DatasetCollection):
|
||||
if value_to_check.elements_deleted:
|
||||
@@ -2595,7 +2601,9 @@ class DataCollectionToolParameter(BaseDataToolParameter):
|
||||
if rval:
|
||||
if isinstance(rval, HistoryDatasetCollectionAssociation):
|
||||
if rval.deleted:
|
||||
raise ParameterValueError("the previously selected dataset collection has been deleted", self.name)
|
||||
raise ParameterValueError(
|
||||
"the previously selected dataset collection has been deleted", self.name, rval
|
||||
)
|
||||
if rval.collection.elements_deleted:
|
||||
raise ParameterValueError(
|
||||
"the previously selected dataset collection has elements that are deleted.", self.name
|
||||
|
||||
@@ -31,6 +31,8 @@ from galaxy.job_execution.actions.post import ActionBox
|
||||
from galaxy.job_execution.compute_environment import ComputeEnvironment
|
||||
from galaxy.managers.credentials import _build_user_credentials_query
|
||||
from galaxy.model import (
|
||||
DatasetInstance,
|
||||
HistoryDatasetCollectionAssociation,
|
||||
Job,
|
||||
PostJobAction,
|
||||
Workflow,
|
||||
@@ -56,6 +58,7 @@ from galaxy.schema.invocation import (
|
||||
InvocationFailureDatasetFailed,
|
||||
InvocationFailureExpressionEvaluationFailed,
|
||||
InvocationFailureOutputNotFound,
|
||||
InvocationFailureStepInputDeleted,
|
||||
InvocationFailureWhenNotBoolean,
|
||||
InvocationFailureWorkflowParameterInvalid,
|
||||
)
|
||||
@@ -94,6 +97,7 @@ from galaxy.tools.parameters.basic import (
|
||||
HiddenToolParameter,
|
||||
IntegerToolParameter,
|
||||
parameter_types,
|
||||
ParameterValueError,
|
||||
raw_to_galaxy,
|
||||
SelectToolParameter,
|
||||
TextToolParameter,
|
||||
@@ -502,7 +506,9 @@ class WorkflowModule:
|
||||
|
||||
def update_value(input, context, prefixed_name, **kwargs):
|
||||
if prefixed_name in step_updates:
|
||||
value, error = check_param(trans, input, step_updates.get(prefixed_name), context)
|
||||
value, error = check_param(
|
||||
trans, input, step_updates.get(prefixed_name), context, simple_errors=False
|
||||
)
|
||||
if error is not None:
|
||||
step_errors[prefixed_name] = error
|
||||
return value
|
||||
@@ -2389,9 +2395,12 @@ class ToolModule(WorkflowModule):
|
||||
# TODO: why do we even create an invocation, seems like something we could check on submit?
|
||||
message = f"Specified tool [{tool.id}] in step {step.order_index + 1} is not workflow-compatible."
|
||||
raise exceptions.MessageException(message)
|
||||
self.state, _ = self.compute_runtime_state(
|
||||
self.state, step_errors = self.compute_runtime_state(
|
||||
trans, step, step_updates=progress.param_map.get(step.id), replace_default_values=True
|
||||
)
|
||||
if step_errors:
|
||||
failure = self._build_step_error_failure(trans, step, step_errors, progress)
|
||||
raise FailWorkflowEvaluation(why=failure)
|
||||
step.state = self.state
|
||||
tool_state = step.state
|
||||
assert tool_state is not None
|
||||
@@ -2605,6 +2614,36 @@ class ToolModule(WorkflowModule):
|
||||
|
||||
return complete
|
||||
|
||||
@staticmethod
|
||||
def _build_step_error_failure(trans, step, step_errors, progress):
|
||||
"""Build the appropriate invocation failure message for step parameter errors.
|
||||
|
||||
Inspects the ParameterValueError objects to determine whether the error
|
||||
is due to a deleted dataset/collection input or a generic validation failure.
|
||||
"""
|
||||
for error in step_errors.values():
|
||||
if isinstance(error, ParameterValueError):
|
||||
pv = error.parameter_value
|
||||
if isinstance(pv, DatasetInstance) and pv.deleted:
|
||||
return InvocationFailureStepInputDeleted(
|
||||
reason=FailureReason.step_input_deleted,
|
||||
workflow_step_id=step.id,
|
||||
hda_id=pv.id,
|
||||
details=str(error),
|
||||
)
|
||||
if isinstance(pv, HistoryDatasetCollectionAssociation) and pv.deleted:
|
||||
return InvocationFailureStepInputDeleted(
|
||||
reason=FailureReason.step_input_deleted,
|
||||
workflow_step_id=step.id,
|
||||
hdca_id=pv.id,
|
||||
details=str(error),
|
||||
)
|
||||
return InvocationFailureWorkflowParameterInvalid(
|
||||
reason=FailureReason.workflow_parameter_invalid,
|
||||
workflow_step_id=step.id,
|
||||
details=str({k: str(v) for k, v in step_errors.items()}),
|
||||
)
|
||||
|
||||
def _effective_post_job_actions(self, step):
|
||||
effective_post_job_actions = step.post_job_actions[:]
|
||||
for key, value in self.runtime_post_job_actions.items():
|
||||
@@ -2839,7 +2878,8 @@ def populate_module_and_state(
|
||||
step_errors = module_injector.compute_runtime_state(step, step_args=step_args)
|
||||
if step_errors:
|
||||
raise exceptions.MessageException(
|
||||
"Error computing workflow step runtime state", err_data={step.order_index: step_errors}
|
||||
"Error computing workflow step runtime state",
|
||||
err_data={step.order_index: {k: str(v) for k, v in step_errors.items()}},
|
||||
)
|
||||
if step.upgrade_messages:
|
||||
if allow_tool_state_corrections:
|
||||
|
||||
@@ -5765,6 +5765,64 @@ steps:
|
||||
assert datasets[PAUSED_1]["state"] == state
|
||||
assert datasets[PAUSED_2]["state"] == "paused"
|
||||
|
||||
@skip_without_tool("cat")
|
||||
def test_workflow_with_deleted_dataset_step_parameter(self):
|
||||
"""Verify workflow fails gracefully when a step parameter references a deleted dataset.
|
||||
|
||||
Uses a pause step so we can delete the dataset after the invocation is
|
||||
queued but before the cat step executes, avoiding a race condition.
|
||||
"""
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
workflow_id = self._upload_yaml_workflow("""
|
||||
class: GalaxyWorkflow
|
||||
inputs:
|
||||
input1:
|
||||
type: data
|
||||
steps:
|
||||
the_pause:
|
||||
type: pause
|
||||
in:
|
||||
input: input1
|
||||
first_cat:
|
||||
tool_id: cat
|
||||
in:
|
||||
input1: the_pause
|
||||
""")
|
||||
valid_hda = self.dataset_populator.new_dataset(history_id, wait=True)
|
||||
to_delete_hda = self.dataset_populator.new_dataset(history_id, wait=True)
|
||||
to_delete_id = to_delete_hda["id"]
|
||||
# Pass the to-be-deleted dataset as a step parameter override for the cat step
|
||||
# (step order_index 2). This passes initial validation because the dataset
|
||||
# is still valid at invocation time.
|
||||
invocation_id = self.workflow_populator.invoke_workflow_and_assert_ok(
|
||||
workflow_id,
|
||||
history_id=history_id,
|
||||
inputs={"input1": {"src": "hda", "id": valid_hda["id"]}},
|
||||
request={
|
||||
"parameters": dumps({"2": {"input1": {"src": "hda", "id": to_delete_id}}}),
|
||||
"parameters_normalized": True,
|
||||
},
|
||||
inputs_by="name",
|
||||
)
|
||||
# Invocation is paused — delete the dataset before resuming.
|
||||
self.dataset_populator.delete_dataset(history_id=history_id, content_id=to_delete_id, purge=False)
|
||||
# Resume the pause step. The cat step will now run and
|
||||
# ToolModule.execute() → compute_runtime_state will find the deleted dataset.
|
||||
self.__review_paused_steps(workflow_id, invocation_id, order_index=1, action=True)
|
||||
self.workflow_populator.wait_for_invocation_and_jobs(
|
||||
history_id=history_id,
|
||||
workflow_id=workflow_id,
|
||||
invocation_id=invocation_id,
|
||||
assert_ok=False,
|
||||
)
|
||||
invocation = self.workflow_populator.get_invocation(invocation_id, step_details=True)
|
||||
assert invocation["state"] == "failed", invocation
|
||||
assert len(invocation["messages"]) == 1
|
||||
message = invocation["messages"][0]
|
||||
assert message["reason"] == "step_input_deleted"
|
||||
assert message["hda_id"] == to_delete_id
|
||||
assert "the previously selected dataset has been deleted" in message["details"]
|
||||
|
||||
def test_run_with_implicit_connection(self):
|
||||
with self.dataset_populator.test_history() as history_id:
|
||||
run_summary = self._run_workflow(
|
||||
|
||||
Reference in New Issue
Block a user