Restart Failed Workflows

beeflow/client/bee_client.py

@@ -38,8 +38,8 @@
 from beeflow.common.object_models import generate_workflow_id
 from beeflow.client import remote_client
 from beeflow.wf_manager.models import (
-    CopyWorkflowRequest,
-    CopyWorkflowResponse,
+    ResubmitWorkflowRequest,
+    ResubmitWorkflowResponse,
     ListWorkflowsResponse,
     SubmitWorkflowRequest,
     ModifyWorkflowRequest,
@@ -456,15 +456,15 @@ def submit(  # pylint:disable=R0915
         ..., help="the workflow name"
     ),
     wf_path: pathlib.Path = typer.Argument(
-        ..., help="path to the workflow .tgz or dir"
+        None, help="path to the workflow .tgz or dir"
     ),
     main_cwl: str = typer.Argument(
-        ...,
+        None,
         help="filename of main CWL (if using CWL tarball), "
         + "path of main CWL (if using CWL directory)",
     ),
     yaml_file: str = typer.Argument(
-        ...,
+        None,
         help="filename of yaml file (if using CWL tarball), "
         + "path of yaml file (if using CWL directory)",
     ),
@@ -476,7 +476,7 @@ def submit(  # pylint:disable=R0915
         False, "--no-start", "-n", help="do not start the workflow"
     ),
 ):
-    """Submit a new workflow."""
+    """Submit a new workflow or resubmit a failed workflow."""
 
     def is_parent(parent, path):
         """Return true if the path is a child of the other path."""
@@ -860,23 +860,19 @@ def cancel(
 
 
 @app.command()
-def copy(wf_id: str = typer.Argument(..., callback=match_short_id)):
-    """Copy an archived workflow."""
+def resubmit(wf_id: str = typer.Argument(..., callback=match_short_id)):
+    """Resubmit a failed archived workflow."""
     long_wf_id = wf_id
     try:
         conn = _wfm_conn()
         resp = conn.patch(
-            _url(), json=CopyWorkflowRequest(wf_id=long_wf_id).model_dump(), timeout=60
+            _url(), json=ResubmitWorkflowRequest(wf_id=long_wf_id).model_dump(), timeout=60
         )
     except requests.exceptions.ConnectionError:
         error_exit("Could not reach WF Manager.")
     if resp.status_code != requests.codes.okay:  # pylint: disable=no-member
-        error_exit("WF Manager could not copy workflow.")
-    archive_info = CopyWorkflowResponse.model_validate(resp.json())
-    archive_file = jsonpickle.decode(archive_info.archive_file_pickle)
-    archive_filename = archive_info.archive_filename
-    logging.info(f"Copy workflow: {resp.text}")
-    return archive_file, archive_filename
+        error_exit("WF Manager could not resubmit workflow.")
+    logging.info(f"Resubmit workflow: {resp.text}")
 
 
 @app.command()

beeflow/common/db/gdb_db.py

@@ -261,6 +261,16 @@ def set_task_state(self, task_id: str, state: str):
             WHERE id = :task_id;"""
         bdb.run(self.db_file, set_task_state_query, {'task_id': task_id, 'state': state})
 
+    def reset_failed_tasks(self, workflow_id: str):
+        """Reset failed tasks."""
+        placeholders = ", ".join("?" for _ in failed_task_states)
+        query = f"""
+            UPDATE task
+            SET state = 'WAITING'
+            WHERE workflow_id = :workflow_id
+              AND state IN ({placeholders});
+        """
+        bdb.run(self.db_file, query, [workflow_id, *failed_task_states])
 
     def add_dependencies(self, task: Task, old_task: Task=None, restarted_task=False):
         """Add dependencies for a task based on its inputs and outputs."""

beeflow/common/gdb/gdb_driver.py

@@ -83,6 +83,17 @@ def restart_task(self, old_task, new_task):
         :type new_task: Task
         """
 
+    @abstractmethod
+    def reset_failed_tasks(self, workflow_id):
+        """Restart a failed task.
+
+        Create a Task node for new_task with state 'RESTARTED' and an edge
+        to indicate that it is the child of the Task node of old_task.
+
+        :rtype: Workflow
+        """
+
+
     @abstractmethod
     def finalize_task(self, task):
         """Set task state to 'COMPLETED' and set inputs from source.

beeflow/common/gdb/sqlite3_driver.py

@@ -103,6 +103,12 @@ def restart_task(self, old_task, new_task):
         self.db.set_task_state(new_task.id, 'WAITING')
         self.db.add_dependencies(new_task, old_task=old_task, restarted_task=True)
 
+    def reset_failed_tasks(self, workflow_id):
+        """Set failed tasks to 'WAITING'. 
+
+        Used after resubmiting a wokrflow. 
+        """
+        self.db.reset_failed_tasks(workflow_id)
 
     def finalize_task(self, task):
         """Set task state to 'COMPLETED' and set inputs from source.
@@ -139,6 +145,12 @@ def get_workflow_description(self, workflow_id):
         """
         return self.db.get_workflow(workflow_id)
 
+    def get_workflow_workdir(self, workflow_id):
+        """Return the workdir for the specified workflow. 
+
+        :rtype: str 
+        """
+
 
     def get_workflow_state(self, workflow_id):
         """Return the current state of the workflow.
@@ -281,7 +293,7 @@ def set_task_input(self, task_id, input_id, value):
         """Set the value of a task input.
 
         :param task_id: the ID of the task whose input to set
-        :type task_id: str
+        :type task_id: stsr
         :param input_id: the ID of the input
         :type input_id: str
         :param value: str or int or float

beeflow/common/wf_interface.py

@@ -73,6 +73,12 @@ def reset_workflow(self, workflow_id):
         self._workflow_id = workflow_id
         self._gdb_driver.set_workflow_state(self._workflow_id, 'SUBMITTED')
 
+    def reset_failed_workflow(self, wf_id):
+        """Reset the execution state and ID of a BEE workflow."""
+        self._gdb_driver.reset_failed_tasks(self._workflow_id)
+        #self._workflow_id = workflow_id
+        self._gdb_driver.set_workflow_state(self._workflow_id, 'RESTARTED')
+
     def add_task(self, task):
         """Add a new task to a BEE workflow.
 

beeflow/wf_manager/models.py

@@ -30,14 +30,14 @@ class SubmitWorkflowResponse(BaseModel):
     status: str
     wf_id: Optional[str] = None
 
-class CopyWorkflowRequest(BaseModel):
+class ResubmitWorkflowRequest(BaseModel):
     """Request model for copying a workflow."""
     wf_id: str
 
-class CopyWorkflowResponse(BaseModel):
+class ResubmitWorkflowResponse(BaseModel):
     """Response model for workflow copy."""
-    archive_file_pickle: str
-    archive_filename: str
+    msg: str
+    status: str
 
 class TaskStateUpdate(BaseModel):
     """Information about a task state update."""

beeflow/wf_manager/resources/wf_list.py

@@ -1,6 +1,6 @@
 """The workflow list module.
 
-This contains endpoints forsubmitting, starting, and reexecuting workflows.
+This module contains endpoints for submitting, starting, and reexecuting workflows.
 """
 
 import base64
@@ -20,11 +20,11 @@
 # from beeflow.common.wf_profiler import WorkflowProfiler
 
 from beeflow.wf_manager.models import (
-    CopyWorkflowRequest,
-    CopyWorkflowResponse,
     ListWorkflowsResponse,
     SubmitWorkflowRequest,
     SubmitWorkflowResponse,
+    ResubmitWorkflowRequest,
+    ResubmitWorkflowResponse,
 )
 from beeflow.wf_manager.resources import wf_utils
 
@@ -82,7 +82,7 @@ def get(self):
         return ListWorkflowsResponse(workflow_info_list=info).model_dump(), 200
 
     def post(self):
-        """Upload a workflown and start."""
+        """Upload a workflow and start."""
         try:
             data = SubmitWorkflowRequest.model_validate(request.json)
         except ValidationError as e:
@@ -115,16 +115,16 @@ def post(self):
         )
 
     def patch(self):
-        """Copy workflow archive."""
-        wf_id = CopyWorkflowRequest.model_validate(request.json).wf_id
-        archive_dir = bc.get("DEFAULT", "bee_archive_dir")
-        archive_path = os.path.join(archive_dir, wf_id + ".tgz")
-        with open(archive_path, "rb") as archive:
-            archive_file = jsonpickle.encode(archive.read())
-        archive_filename = os.path.basename(archive_path)
-        return (
-            CopyWorkflowResponse(
-                archive_file_pickle=archive_file, archive_filename=archive_filename
-            ).model_dump(),
-            200,
-        )
+        """Resubmit failed workflow."""
+        try:
+            data = ResubmitWorkflowRequest.model_validate(request.json)
+        except ValidationError as e:
+            log.error(f"Error parsing request data: {e}")
+            return (
+                ResubmitWorkflowResponse(
+                    msg="Invalid request data", status="error", wf_id=None
+                ).model_dump(),
+                400,
+            )
+        wf_id = data.wf_id
+        wf_utils.restart_workflow(wf_id)

beeflow/wf_manager/resources/wf_utils.py

@@ -222,6 +222,13 @@ def setup_workflow(wf_id, wf_name, wf_dir, wf_workdir, no_start, workflow=None,
         log.info("Starting workflow")
         start_workflow.delay(wf_id)
 
+def restart_workflow(wf_id):
+    """Restart a failed workflow."""
+    wfi = get_workflow_interface(wf_id)
+    wfi.reset_failed_workflow(wf_id)
+    update_wf_status(wf_id, "Starting")
+    log.info("Reset failed worflow tasks.")
+
 
 def export_dag(wf_id, output_dir, graphmls_dir, no_dag_dir, workflow_dir=None):
     """Export the DAG of the workflow."""