Formalize workflow invocation and invocation step outputs.

Workflow Invocations
--------------------

The workflow invocation outputs half of this is relatively straight forward. It is modelled somewhat on job outputs, output datasets and output dataset collections are now tracked for each workflow invocation and exposed via the workflow invocation API. This required adding new tables (linked to WorkflowInvocations and WorkflowOutputs) that track these output associations.

Previously one could imagine backtracking this information for simple tool steps via the WorkflowInvocationStep -> Job table, but for steps that have many jobs (i.e. mapping over a collection) or for non-tool steps such information was more difficult to recover (and simply couldn't be recovered from the API at all or even internally without significant knowledge of the underlying workflow).

Workflow Invocation Steps
-------------------------

Tracking the outputs of WorkflowInvocationSteps was not previously done at all, one would have to follow the Job table as well. A signficant downside to this is that one cannot map over empty collections in a workflow - since no such job would exist. Tracking job outputs for WorkflowInvocationSteps is not a simple matter of just attaching outputs to an existing table because we had no concept of a workflow step tracked - since there could be many WorklfowInvocationSteps corresponding to the same combination of WorkflowInvocation and WorkflowStep. That should feel wrong and that is because it is - when collections were added the possiblity of having many jobs for the same combination of WorkflowInvocation and WorkflowStep was added. I should have split WorkflowInvocationSteps into WorkflowInvocationSteps and WorkflowInvocationStepJobAssociations at that time but didn't. This commit now does it - effectively normalizing the ``workflow_invocation_step`` table by introducing the new ``workflow_invocation_step_job_association`` table.

Splitting up the WorkflowInvocationStep table this way allows recovering the mapped over output (e.g. the implicitly created collection from all the jobs) as well the outputs from the individual jobs (by walking WorkflowInvocationStep -> WorkflowInvocationStepJobAssociation -> Job -> JobToOutput*Association).

This split up involves failrly substantial changes to the workflow module interface. Any place a list of WorkflowInvocationSteps was assumed, I reworked it to just expect a single WorkflowInvocationStep. I vastly simplified recover_mapping to just use the persisted outputs (this was needed in order to also implment empty collection mapping in workflows). This also fixes a bug (or implements a missing feature) where Subworkflow moudles had no recover_mapping methods - so for instance if a tool that produces dynamic collections appeared anywhere in a workflow after a subworkflow step - that workflow would not complete scheduling properly.

Now that we have a way to reference the set of jobs corresponding to a workflow step within an invocation, we can start to track partial scheduling of such steps. This is outlined in https://github.com/galaxyproject/galaxy/issues/3883 and refactoring toward this goal is included here - including adding a state to WorkflowInvocationStep so Galaxy can determine if it has started scheduling this step and an index when scheduling jobs so it can tell how far into a scheduling things have gone as well as augmenting the tool executor to take a maximum number of jobs to execute and allow recovery of existing jobs for collection building purposes.

*Applications*

These changes will enable:

- A simple, consistent API for finding workflow outputs that can be consumed by Planemo for testing workflows.
- Mapping over empty collections in workflows.
- Re-scheduling workflow invocations that include subworkflow steps.
- Partial scheduling within steps requiring a large number of jobs when scheduling workflow invocations.

lib/galaxy/jobs/actions/post.py

@@ -269,10 +269,10 @@ class DeleteIntermediatesAction(DefaultJobAction):
         # concurrently, sometimes non-terminal steps won't be cleaned up
         # because of the lag in job state updates.
         sa_session.flush()
-        if not job.workflow_invocation_step:
+        if not job.workflow_invocation_step_assoc.workflow_invocation_step:
             log.debug("This job is not part of a workflow invocation, delete intermediates aborted.")
             return
-        wfi = job.workflow_invocation_step.workflow_invocation
+        wfi = job.workflow_invocation_step_assoc.workflow_invocation_step.workflow_invocation
         sa_session.refresh(wfi)
         if wfi.active:
             log.debug("Workflow still scheduling so new jobs may appear, skipping deletion of intermediate files.")
@@ -285,9 +285,9 @@ class DeleteIntermediatesAction(DefaultJobAction):
             jobs_to_check = []
             for wfi_step in wfi_steps:
                 sa_session.refresh(wfi_step)
-                wfi_step_job = wfi_step.job
-                if wfi_step_job:
-                    jobs_to_check.append(wfi_step_job)
+                wfi_step_job_assocs = wfi_step.jobs
+                if wfi_step_job_assocs:
+                    jobs_to_check.extend(map(lambda j: j.job, wfi_step_job_assocs))
                 else:
                     log.debug("No job found yet for wfi_step %s, (step %s)" % (wfi_step, wfi_step.workflow_step))
             for j2c in jobs_to_check:
@@ -302,7 +302,7 @@ class DeleteIntermediatesAction(DefaultJobAction):
                 for (input_dataset, creating_job) in creating_jobs:
                     sa_session.refresh(creating_job)
                     sa_session.refresh(input_dataset)
-                for input_dataset in [x.dataset for (x, creating_job) in creating_jobs if creating_job.workflow_invocation_step and creating_job.workflow_invocation_step.workflow_invocation == wfi]:
+                for input_dataset in [x.dataset for (x, creating_job) in creating_jobs if creating_job.workflow_invocation_step_assoc and creating_job.workflow_invocation_step_assoc.workflow_invocation_step.workflow_invocation == wfi]:
                     # note that the above input_dataset is a reference to a
                     # job.input_dataset.dataset at this point
                     safe_to_delete = True

lib/galaxy/model/__init__.py

@@ -803,8 +803,9 @@ class Job(object, JobLike, UsesCreateAndUpdateTime, Dictifiable):
 
     def set_final_state(self, final_state):
         self.set_state(final_state)
-        if self.workflow_invocation_step:
-            self.workflow_invocation_step.update()
+        workflow_invocation_step_assoc = self.workflow_invocation_step_assoc
+        if workflow_invocation_step_assoc:
+            workflow_invocation_step_assoc.workflow_invocation_step.update()
 
     def get_destination_configuration(self, config, key, default=None):
         """ Get a destination parameter that can be defaulted back
@@ -4036,17 +4037,16 @@ class WorkflowInvocation(object, UsesCreateAndUpdateTime, Dictifiable):
         step_invocations = {}
         for invocation_step in self.steps:
             step_id = invocation_step.workflow_step_id
-            if step_id not in step_invocations:
-                step_invocations[step_id] = []
-            step_invocations[step_id].append(invocation_step)
+            assert step_id not in step_invocations
+            step_invocations[step_id] = invocation_step
         return step_invocations
 
-    def step_invocations_for_step_id(self, step_id):
-        step_invocations = []
+    def step_invocation_for_step_id(self, step_id):
+        target_invocation_step = None
         for invocation_step in self.steps:
             if step_id == invocation_step.workflow_step_id:
-                step_invocations.append(invocation_step)
-        return step_invocations
+                target_invocation_step = invocation_step
+        return target_invocation_step
 
     @staticmethod
     def poll_active_workflow_ids(
@@ -4072,6 +4072,24 @@ class WorkflowInvocation(object, UsesCreateAndUpdateTime, Dictifiable):
         # is relatively intutitive.
         return [wid for wid in query.all()]
 
+    def add_output(self, workflow_output, step, output_object):
+        if output_object.history_content_type == "dataset":
+            output_assoc = WorkflowInvocationOutputDatasetAssociation()
+            output_assoc.workflow_invocation = self
+            output_assoc.workflow_output = workflow_output
+            output_assoc.workflow_step = step
+            output_assoc.dataset = output_object
+            self.output_datasets.append(output_assoc)
+        elif output_object.history_content_type == "dataset_collection":
+            output_assoc = WorkflowInvocationOutputDatasetCollectionAssociation()
+            output_assoc.workflow_invocation = self
+            output_assoc.workflow_output = workflow_output
+            output_assoc.workflow_step = step
+            output_assoc.dataset_collection = output_object
+            self.output_dataset_collections.append(output_assoc)
+        else:
+            raise Exception("Uknown output type encountered")
+
     def to_dict(self, view='collection', value_mapper=None, step_details=False):
         rval = super(WorkflowInvocation, self).to_dict(view=view, value_mapper=value_mapper)
         if view == 'element':
@@ -4087,17 +4105,43 @@ class WorkflowInvocation(object, UsesCreateAndUpdateTime, Dictifiable):
             inputs = {}
             for step in self.steps:
                 if step.workflow_step.type == 'tool':
-                    for step_input in step.workflow_step.input_connections:
-                        output_step_type = step_input.output_step.type
-                        if output_step_type in ['data_input', 'data_collection_input']:
-                            src = "hda" if output_step_type == 'data_input' else 'hdca'
-                            for job_input in step.job.input_datasets:
-                                if job_input.name == step_input.input_name:
-                                    inputs[str(step_input.output_step.order_index)] = {
-                                        "id": job_input.dataset_id, "src": src,
-                                        "uuid" : str(job_input.dataset.dataset.uuid) if job_input.dataset.dataset.uuid is not None else None
-                                    }
+                    for step_job_assoc in step.jobs:
+                        for step_input in step.workflow_step.input_connections:
+                            output_step_type = step_input.output_step.type
+                            if output_step_type in ['data_input', 'data_collection_input']:
+                                src = "hda" if output_step_type == 'data_input' else 'hdca'
+                                for job_input in step_job_assoc.job.input_datasets:
+                                    if job_input.name == step_input.input_name:
+                                        inputs[str(step_input.output_step.order_index)] = {
+                                            "id": job_input.dataset_id, "src": src,
+                                            "uuid" : str(job_input.dataset.dataset.uuid) if job_input.dataset.dataset.uuid is not None else None
+                                        }
             rval['inputs'] = inputs
+
+            outputs = {}
+            for output_assoc in self.output_datasets:
+                label = output_assoc.workflow_output.label
+                if not label:
+                    continue
+
+                outputs[label] = {
+                    'src': 'hda',
+                    'id': output_assoc.dataset_id,
+                }
+
+            output_collections = {}
+            for output_assoc in self.output_dataset_collections:
+                label = output_assoc.workflow_output.label
+                if not label:
+                    continue
+
+                output_collections[label] = {
+                    'src': 'hdca',
+                    'id': output_assoc.dataset_collection_id,
+                }
+
+            rval['outputs'] = outputs
+            rval['output_collections'] = output_collections
         return rval
 
     def update(self):
@@ -4137,36 +4181,70 @@ class WorkflowInvocationToSubworkflowInvocationAssociation(object, Dictifiable):
 
 
 class WorkflowInvocationStep(object, Dictifiable):
-    dict_collection_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'action']
-    dict_element_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'action']
+    dict_collection_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'state', 'action']
+    dict_element_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'state', 'action']
+    states = Bunch(
+        NEW='new',  # Brand new workflow invocation step
+        READY='ready',  # Workflow invocation step ready for another iteration of scheduling.
+        SCHEDULED='scheduled',  # Workflow invocation step has been scheduled.
+        # CANCELLED='cancelled',  TODO: implement and expose
+        # FAILED='failed',  TODO: implement and expose
+    )
 
     def update(self):
         self.workflow_invocation.update()
 
+    def add_output(self, output_name, output_object):
+        if output_object.history_content_type == "dataset":
+            output_assoc = WorkflowInvocationStepOutputDatasetAssociation()
+            output_assoc.workflow_invocation_step = self
+            output_assoc.dataset = output_object
+            output_assoc.output_name = output_name
+            self.output_datasets.append(output_assoc)
+        elif output_object.history_content_type == "dataset_collection":
+            output_assoc = WorkflowInvocationStepOutputDatasetCollectionAssociation()
+            output_assoc.workflow_invocation_step = self
+            output_assoc.dataset_collection = output_object
+            output_assoc.output_name = output_name
+            self.output_dataset_collections.append(output_assoc)
+        else:
+            raise Exception("Uknown output type encountered")
+
     def to_dict(self, view='collection', value_mapper=None):
         rval = super(WorkflowInvocationStep, self).to_dict(view=view, value_mapper=value_mapper)
         rval['order_index'] = self.workflow_step.order_index
         rval['workflow_step_label'] = self.workflow_step.label
         rval['workflow_step_uuid'] = str(self.workflow_step.uuid)
-        rval['state'] = self.job.state if self.job is not None else None
-        if self.job is not None and view == 'element':
-            output_dict = {}
-            for i in self.job.output_datasets:
-                if i.dataset is not None:
-                    output_dict[i.name] = {
-                        "id" : i.dataset.id, "src" : "hda",
-                        "uuid" : str(i.dataset.dataset.uuid) if i.dataset.dataset.uuid is not None else None
-                    }
-            for i in self.job.output_library_datasets:
-                if i.dataset is not None:
-                    output_dict[i.name] = {
-                        "id" : i.dataset.id, "src" : "ldda",
-                        "uuid" : str(i.dataset.dataset.uuid) if i.dataset.dataset.uuid is not None else None
-                    }
-            rval['outputs'] = output_dict
+        # Following no longer makes sense...
+        # rval['state'] = self.job.state if self.job is not None else None
+        if view == 'element':
+            outputs = {}
+            for output_assoc in self.output_datasets:
+                name = output_assoc.output_name
+                outputs[name] = {
+                    'src': 'hda',
+                    'id': output_assoc.dataset.id,
+                    'uuid': str(output_assoc.dataset.dataset.uuid) if output_assoc.dataset.dataset.uuid is not None else None
+                }
+
+            output_collections = {}
+            for output_assoc in self.output_dataset_collections:
+                name = output_assoc.output_name
+                output_collections[name] = {
+                    'src': 'hdca',
+                    'id': output_assoc.dataset_collection.id,
+                }
+
+            rval['outputs'] = outputs
+            rval['output_collections'] = output_collections
         return rval
 
 
+class WorkflowInvocationStepJobAssociation(object, Dictifiable):
+    dict_collection_visible_keys = ('id', 'job_id', 'workflow_invocation_step_id')
+    dict_element_visible_keys = ('id', 'job_id', 'workflow_invocation_step_id')
+
+
 class WorkflowRequest(object, Dictifiable):
     dict_collection_visible_keys = ['id', 'name', 'type', 'state', 'history_id', 'workflow_id']
     dict_element_visible_keys = ['id', 'name', 'type', 'state', 'history_id', 'workflow_id']
@@ -4221,6 +4299,26 @@ class WorkflowRequestInputStepParmeter(object, Dictifiable):
     dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'parameter_value']
 
 
+class WorkflowInvocationOutputDatasetAssociation(object, Dictifiable):
+    """Represents links to output datasets for the workflow."""
+    dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'dataset_id', 'name']
+
+
+class WorkflowInvocationOutputDatasetCollectionAssociation(object, Dictifiable):
+    """Represents links to output dataset collections for the workflow."""
+    dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'dataset_collection_id', 'name']
+
+
+class WorkflowInvocationStepOutputDatasetAssociation(object, Dictifiable):
+    """Represents links to output datasets for the workflow."""
+    dict_collection_visible_keys = ['id', 'workflow_invocation_step_id', 'dataset_id', 'output_name']
+
+
+class WorkflowInvocationStepOutputDatasetCollectionAssociation(object, Dictifiable):
+    """Represents links to output dataset collections for the workflow."""
+    dict_collection_visible_keys = ['id', 'workflow_invocation_step_id', 'dataset_collection_id', 'output_name']
+
+
 class MetadataFile(StorableObject):
 
     def __init__(self, dataset=None, name=None):

lib/galaxy/model/mapping.py

@@ -901,9 +901,54 @@ model.WorkflowInvocationStep.table = Table(
     Column("update_time", DateTime, default=now, onupdate=now),
     Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True, nullable=False),
     Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True, nullable=False),
-    Column("job_id", Integer, ForeignKey("job.id"), index=True, nullable=True),
+    Column("state", TrimmedString(64), index=True),
     Column("action", JSONType, nullable=True))
 
+
+model.WorkflowInvocationStepJobAssociation.table = Table(
+    "workflow_invocation_step_job_association", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True, nullable=False),
+    Column("order_index", Integer, nullable=True),  # recovering partially complete WorkflowInvocationSteps requires knowing which jobs have been scheduled
+    Column("job_id", Integer, ForeignKey("job.id"), index=True, nullable=False),
+)
+
+
+model.WorkflowInvocationOutputDatasetAssociation.table = Table(
+    "workflow_invocation_output_dataset_association", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True),
+    Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("dataset_id", Integer, ForeignKey("history_dataset_association.id"), index=True),
+    Column("workflow_output_id", Integer, ForeignKey("workflow_output.id"), index=True),
+)
+
+model.WorkflowInvocationOutputDatasetCollectionAssociation.table = Table(
+    "workflow_invocation_output_dataset_collection_association", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True),
+    Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("dataset_collection_id", Integer, ForeignKey("history_dataset_collection_association.id"), index=True),
+    Column("workflow_output_id", Integer, ForeignKey("workflow_output.id"), index=True),
+)
+
+model.WorkflowInvocationStepOutputDatasetAssociation.table = Table(
+    "workflow_invocation_step_output_dataset_association", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True),
+    Column("dataset_id", Integer, ForeignKey("history_dataset_association.id"), index=True),
+    Column("output_name", String(255), nullable=True),
+)
+
+model.WorkflowInvocationStepOutputDatasetCollectionAssociation.table = Table(
+    "workflow_invocation_step_output_dataset_collection_association", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True),
+    Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("dataset_collection_id", Integer, ForeignKey("history_dataset_collection_association.id"), index=True),
+    Column("output_name", String(255), nullable=True),
+)
+
 model.WorkflowInvocationToSubworkflowInvocationAssociation.table = Table(
     "workflow_invocation_to_subworkflow_invocation_association", metadata,
     Column("id", Integer, primary_key=True),
@@ -2310,7 +2355,7 @@ mapper(model.WorkflowInvocation, model.WorkflowInvocation.table, properties=dict
         uselist=True,
     ),
     steps=relation(model.WorkflowInvocationStep,
-        backref='workflow_invocation'),
+        backref="workflow_invocation"),
     workflow=relation(model.Workflow)
 ))
 
@@ -2323,12 +2368,17 @@ mapper(model.WorkflowInvocationToSubworkflowInvocationAssociation, model.Workflo
     workflow_step=relation(model.WorkflowStep),
 ))
 
-mapper(model.WorkflowInvocationStep, model.WorkflowInvocationStep.table, properties=dict(
-    workflow_step=relation(model.WorkflowStep),
+
+simple_mapping(model.WorkflowInvocationStepJobAssociation,
+    workflow_invocation_step=relation(model.WorkflowInvocationStep, backref="jobs"),
     job=relation(model.Job,
-        backref=backref('workflow_invocation_step',
-            uselist=False))
-))
+        backref=backref('workflow_invocation_step_assoc',
+            uselist=False)))
+
+
+simple_mapping(model.WorkflowInvocationStep,
+    workflow_step=relation(model.WorkflowStep))
+
 
 simple_mapping(model.WorkflowRequestInputParameter,
     workflow_invocation=relation(model.WorkflowInvocation))
@@ -2358,6 +2408,39 @@ mapper(model.MetadataFile, model.MetadataFile.table, properties=dict(
     library_dataset=relation(model.LibraryDatasetDatasetAssociation)
 ))
 
+
+simple_mapping(
+    model.WorkflowInvocationOutputDatasetAssociation,
+    workflow_invocation=relation(model.WorkflowInvocation, backref="output_datasets"),
+    workflow_step=relation(model.WorkflowStep),
+    dataset=relation(model.HistoryDatasetAssociation),
+    workflow_output=relation(model.WorkflowOutput),
+)
+
+
+simple_mapping(
+    model.WorkflowInvocationOutputDatasetCollectionAssociation,
+    workflow_invocation=relation(model.WorkflowInvocation, backref="output_dataset_collections"),
+    workflow_step=relation(model.WorkflowStep),
+    dataset_collection=relation(model.HistoryDatasetCollectionAssociation),
+    workflow_output=relation(model.WorkflowOutput),
+)
+
+
+simple_mapping(
+    model.WorkflowInvocationStepOutputDatasetAssociation,
+    workflow_invocation_step=relation(model.WorkflowInvocationStep, backref="output_datasets"),
+    dataset=relation(model.HistoryDatasetAssociation),
+)
+
+
+simple_mapping(
+    model.WorkflowInvocationStepOutputDatasetCollectionAssociation,
+    workflow_invocation_step=relation(model.WorkflowInvocationStep, backref="output_dataset_collections"),
+    dataset_collection=relation(model.HistoryDatasetCollectionAssociation),
+)
+
+
 mapper(model.PageRevision, model.PageRevision.table)
 
 mapper(model.Page, model.Page.table, properties=dict(

lib/galaxy/model/migrate/versions/0136_record_workflow_outputs.py

@@ -0,0 +1,185 @@
+"""
+Migration script for workflow request tables.
+"""
+from __future__ import print_function
+
+import datetime
+import logging
+
+from collections import OrderedDict
+
+from migrate.changeset.constraint import ForeignKeyConstraint
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, MetaData, String, Table
+
+from galaxy.model.custom_types import JSONType, TrimmedString
+
+
+now = datetime.datetime.utcnow
+
+log = logging.getLogger(__name__)
+metadata = MetaData()
+
+
+def get_new_tables():
+    # Normally we define this globally in the file, but we need to delay the
+    # reading of existing tables because an existing workflow_invocation_step
+    # table exists that we want to recreate.
+
+    workflow_invocation_output_dataset_association_table = Table(
+        "workflow_invocation_output_dataset_association", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True),
+        Column("workflow_step_id", Integer, ForeignKey("workflow_step.id")),
+        Column("dataset_id", Integer, ForeignKey("history_dataset_association.id"), index=True),
+        Column("workflow_output_id", Integer, ForeignKey("workflow_output.id")),
+    )
+
+    workflow_invocation_output_dataset_collection_association_table = Table(
+        "workflow_invocation_output_dataset_collection_association", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True),
+        Column("workflow_step_id", Integer, ForeignKey("workflow_step.id")),
+        Column("dataset_collection_id", Integer, ForeignKey("history_dataset_collection_association.id"), index=True),
+        Column("workflow_output_id", Integer, ForeignKey("workflow_output.id")),
+    )
+
+    workflow_invocation_step_output_dataset_association_table = Table(
+        "workflow_invocation_step_output_dataset_association", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True),
+        Column("dataset_id", Integer, ForeignKey("history_dataset_association.id"), index=True),
+        Column("output_name", String(255), nullable=True),
+    )
+
+    workflow_invocation_step_output_dataset_collection_association_table = Table(
+        "workflow_invocation_step_output_dataset_collection_association", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True),
+        Column("workflow_step_id", Integer, ForeignKey("workflow_step.id")),
+        Column("dataset_collection_id", Integer, ForeignKey("history_dataset_collection_association.id"), index=True),
+        Column("output_name", String(255), nullable=True),
+    )
+
+    workflow_invocation_step_table = Table(
+        "workflow_invocation_step", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("create_time", DateTime, default=now),
+        Column("update_time", DateTime, default=now, onupdate=now),
+        Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True, nullable=False),
+        Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True, nullable=False),
+        Column("action", JSONType, nullable=True),
+        Column("state", TrimmedString(64), default="new"),
+    )
+
+    workflow_invocation_step_job_association_table = Table(
+        "workflow_invocation_step_job_association", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True, nullable=False),
+        Column("order_index", Integer, nullable=True),
+        Column("job_id", Integer, ForeignKey("job.id"), index=True, nullable=False),
+    )
+
+    tables = OrderedDict()
+    tables["workflow_invocation_step"] = workflow_invocation_step_table
+    tables["workflow_invocation_output_dataset_association"] = workflow_invocation_output_dataset_association_table
+    tables["workflow_invocation_output_dataset_collection_association"] = workflow_invocation_output_dataset_collection_association_table
+    tables["workflow_invocation_step_output_dataset_association"] = workflow_invocation_step_output_dataset_association_table
+    tables["workflow_invocation_step_output_dataset_collection_association"] = workflow_invocation_step_output_dataset_collection_association_table
+    tables["workflow_invocation_step_job_association"] = workflow_invocation_step_job_association_table
+
+    return tables
+
+
+def upgrade(migrate_engine):
+    metadata.bind = migrate_engine
+    print(__doc__)
+
+    LegacyWorkflowInvocationStep_table = Table("workflow_invocation_step", metadata, autoload=True)
+    ExistingWorkflowInvocation_table = Table("workflow_invocation", metadata, autoload=True)
+
+    cons = ForeignKeyConstraint([LegacyWorkflowInvocationStep_table.c.workflow_invocation_id], [ExistingWorkflowInvocation_table.c.id])
+    cons.drop()
+
+    for index in LegacyWorkflowInvocationStep_table.indexes:
+        index.drop()
+
+    LegacyWorkflowInvocationStep_table.rename("workflow_invocation_step_premigrate135")
+    # Try to deregister that workflow_invocation_step to work around some caching problems
+    # it seems.
+    LegacyWorkflowInvocationStep_table.deregister()
+    metadata._remove_table("workflow_invocation_step", metadata.schema)
+
+    metadata.reflect()
+    tables = get_new_tables()
+    for table in tables.values():
+        __create(table)
+
+    def nextval(table, col='id'):
+        if migrate_engine.name in ['postgres', 'postgresql']:
+            return "nextval('%s_%s_seq')" % (table, col)
+        elif migrate_engine.name in ['mysql', 'sqlite']:
+            return "null"
+        else:
+            raise Exception("Unhandled database type")
+
+    # Skips action - since I can't aggregate (sql needs a RANDOM(col)) that and it is only used by optional
+    # beta extensions.
+    cmd = \
+        "INSERT INTO workflow_invocation_step " + \
+        "(id, create_time, update_time, workflow_invocation_id, workflow_step_id, action, state)" + \
+        "SELECT " + \
+        nextval('workflow_invocation_step') + " AS id, " \
+        "MIN(create_time) AS create_time, " + \
+        "MAX(update_time) AS update_time, " + \
+        "workflow_invocation_step_premigrate135.workflow_invocation_id AS workflow_invocation_id, " +\
+        "workflow_invocation_step_premigrate135.workflow_step_id AS workflow_step_id, " + \
+        "NULL AS action, " + \
+        "'scheduled' AS state " + \
+        "FROM workflow_invocation_step_premigrate135 " + \
+        "WHERE workflow_invocation_step_premigrate135.workflow_step_id IS NOT NULL AND workflow_invocation_step_premigrate135.workflow_invocation_id IS NOT NULL " +\
+        "GROUP BY workflow_invocation_step_premigrate135.workflow_invocation_id, workflow_invocation_step_premigrate135.workflow_step_id " + \
+        ""
+    migrate_engine.execute(cmd)
+
+    cmd = \
+        "INSERT INTO workflow_invocation_step_job_association " + \
+        "(id, workflow_invocation_step_id, order_index, job_id) " + \
+        "SELECT " + \
+        nextval('workflow_invocation_step_job_association') + " AS id, " \
+        "workflow_invocation_step.id AS workflow_invocation_step_id, " + \
+        "NULL AS order_index, " + \
+        "job_id AS job_id " + \
+        "FROM workflow_invocation_step_premigrate135 " + \
+        "LEFT JOIN workflow_invocation_step on (" + \
+        "  workflow_invocation_step.workflow_invocation_id = workflow_invocation_step_premigrate135.workflow_invocation_id " + \
+        "  AND workflow_invocation_step.workflow_step_id = workflow_invocation_step_premigrate135.workflow_step_id" + \
+        ") " + \
+        "WHERE job_id is not NULL "
+    migrate_engine.execute(cmd)
+
+
+def downgrade(migrate_engine):
+    metadata.bind = migrate_engine
+    metadata.reflect()
+
+    tables = get_new_tables()
+    for table in tables.values():
+        __drop(table)
+
+    # Drop new workflow invocation step and job association table and restore legacy data.
+    LegacyWorkflowInvocationStep_table = Table("workflow_invocation_step_premigrate135", metadata, autoload=True)
+    LegacyWorkflowInvocationStep_table.rename("workflow_invocation_step")
+
+
+def __create(table):
+    try:
+        table.create()
+    except Exception:
+        log.exception("Creating %s table failed.", table.name)
+
+
+def __drop(table):
+    try:
+        table.drop()
+    except Exception:
+        log.exception("Dropping %s table failed.", table.name)

lib/galaxy/tools/execute.py

@@ -18,7 +18,13 @@ log = logging.getLogger(__name__)
 EXECUTION_SUCCESS_MESSAGE = "Tool [%s] created job [%s] %s"
 
 
-def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, collection_info=None, workflow_invocation_uuid=None):
+class PartialJobExecution(Exception):
+
+    def __init__(self, jobs):
+        self.jobs = jobs
+
+
+def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, collection_info=None, workflow_invocation_uuid=None, invocation_step=None, max_num_jobs=None):
     """
     Execute a tool and return object containing summary (output data, number of
     failures, etc...).
@@ -28,6 +34,8 @@ def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, c
     app = trans.app
     execution_cache = ToolExecutionCache(trans)
 
+    new_jobs = []
+
     def execute_single_job(params):
         job_timer = ExecutionTimer()
         if workflow_invocation_uuid:
@@ -41,6 +49,7 @@ def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, c
             message = EXECUTION_SUCCESS_MESSAGE % (tool.id, job.id, job_timer)
             log.debug(message)
             execution_tracker.record_success(job, result)
+            new_jobs.append(job)
         else:
             execution_tracker.record_error(result)
 
@@ -59,11 +68,27 @@ def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, c
                 history
             )
 
+    if invocation_step:
+        execution_tracker.recover_successful_jobs(invocation_step)
+
+    previously_executed_jobs_count = len(execution_tracker.successful_jobs)
     job_count = len(execution_tracker.param_combinations)
-    if job_count < burst_at or burst_threads < 2:
-        for params in execution_tracker.param_combinations:
-            execute_single_job(params)
+
+    jobs_executed = 0
+    has_remaining_jobs = False
+
+    if (job_count < burst_at or burst_threads < 2):
+        for index, params in enumerate(execution_tracker.param_combinations):
+            if index < previously_executed_jobs_count:
+                continue
+            elif max_num_jobs and jobs_executed >= max_num_jobs:
+                has_remaining_jobs = True
+                break
+            else:
+                execute_single_job(params)
+                jobs_executed += 1
     else:
+        # TODO: re-record success...
         q = Queue()
 
         def worker():
@@ -77,11 +102,21 @@ def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, c
             t.daemon = True
             t.start()
 
-        for params in execution_tracker.param_combinations:
-            q.put(params)
+        for index, params in enumerate(execution_tracker.param_combinations):
+            if index < previously_executed_jobs_count:
+                continue
+            elif max_num_jobs and jobs_executed >= max_num_jobs:
+                has_remaining_jobs = True
+                break
+            else:
+                q.put(params)
+                jobs_executed += 1
 
         q.join()
 
+    if has_remaining_jobs:
+        raise PartialJobExecution(new_jobs)
+
     log.debug("Executed %d job(s) for tool %s request: %s" % (job_count, tool.id, all_jobs_timer))
     if collection_info:
         history = history or tool.get_default_history_by_trans(trans)
@@ -110,6 +145,17 @@ class ToolExecutionTracker(object):
         self.outputs_by_output_name = collections.defaultdict(list)
         self.implicit_collections = {}
 
+    def recover_successful_jobs(self, invocation_step):
+        # TODO: Optimize away the need to do this - we should just be dealing with IDs
+        # and such and we shouldn't fetch them until the very end when we need them to create
+        # collections.
+        for job_assoc in invocation_step.jobs:
+            job = job_assoc.job
+            for job_output in job.output_datasets:
+                self.outputs_by_output_name[job_output.name].append(job_output.dataset)
+            for job_output in job.output_dataset_collections:
+                self.outputs_by_output_name[job_output.name].append(job_output.dataset_collection)
+
     def record_success(self, job, outputs):
         self.successful_jobs.append(job)
         self.output_datasets.extend(outputs)

lib/galaxy/workflow/modules.py

@@ -21,7 +21,7 @@ from galaxy.tools import (
     DefaultToolState,
     ToolInputsNotReadyException
 )
-from galaxy.tools.execute import execute
+from galaxy.tools.execute import execute, PartialJobExecution
 from galaxy.tools.parameters import (
     check_param,
     params_to_incoming,
@@ -214,10 +214,14 @@ class WorkflowModule(object):
         state.decode(runtime_state, Bunch(inputs=self.get_runtime_inputs()), self.trans.app)
         return state
 
-    def execute(self, trans, progress, invocation, step):
-        """ Execute the given workflow step in the given workflow invocation.
+    def execute(self, trans, progress, invocation_step):
+        """ Execute the given workflow invocation step.
+
         Use the supplied workflow progress object to track outputs, find
-        inputs, etc...
+        inputs, etc....
+
+        Return jobs created and a boolean indicating if there are additional
+        jobs to create on subsequent workflow scheduling tests.
         """
         raise TypeError("Abstract method")
 
@@ -230,11 +234,19 @@ class WorkflowModule(object):
         """
         raise exceptions.RequestParameterInvalidException("Attempting to perform invocation step action on module that does not support actions.")
 
-    def recover_mapping(self, step, step_invocations, progress):
+    def recover_mapping(self, invocation_step, progress):
         """ Re-populate progress object with information about connections
-        from previously executed steps recorded via step_invocations.
+        from previously executed steps recorded via invocation_steps.
         """
-        raise TypeError("Abstract method")
+        outputs = {}
+
+        for output_dataset_assoc in invocation_step.output_datasets:
+            outputs[output_dataset_assoc.output_name] = output_dataset_assoc.dataset
+
+        for output_dataset_collection_assoc in invocation_step.output_dataset_collections:
+            outputs[output_dataset_collection_assoc.output_name] = output_dataset_collection_assoc.dataset_collection
+
+        progress.set_step_outputs(invocation_step, outputs, already_persisted=True)
 
 
 class SubWorkflowModule(WorkflowModule):
@@ -320,11 +332,12 @@ class SubWorkflowModule(WorkflowModule):
     def get_content_id(self):
         return self.trans.security.encode_id(self.subworkflow.id)
 
-    def execute(self, trans, progress, invocation, step):
+    def execute(self, trans, progress, invocation_step):
         """ Execute the given workflow step in the given workflow invocation.
         Use the supplied workflow progress object to track outputs, find
         inputs, etc...
         """
+        step = invocation_step.workflow_step
         subworkflow_invoker = progress.subworkflow_invoker(trans, step)
         subworkflow_invoker.invoke()
         subworkflow = subworkflow_invoker.workflow
@@ -334,7 +347,7 @@ class SubWorkflowModule(WorkflowModule):
             workflow_output_label = workflow_output.label or "%s:%s" % (step.order_index, workflow_output.output_name)
             replacement = subworkflow_progress.get_replacement_workflow_output(workflow_output)
             outputs[workflow_output_label] = replacement
-        progress.set_step_outputs(step, outputs)
+        progress.set_step_outputs(invocation_step, outputs)
         return None
 
     def get_runtime_state(self):
@@ -353,8 +366,10 @@ class InputModule(WorkflowModule):
     def get_data_inputs(self):
         return []
 
-    def execute(self, trans, progress, invocation, step):
-        job, step_outputs = None, dict(output=step.state.inputs['input'])
+    def execute(self, trans, progress, invocation_step):
+        invocation = invocation_step.workflow_invocation
+        step = invocation_step.workflow_step
+        step_outputs = dict(output=step.state.inputs['input'])
 
         # Web controller may set copy_inputs_to_history, API controller always sets
         # inputs.
@@ -378,11 +393,10 @@ class InputModule(WorkflowModule):
             content = next(iter(step_outputs.values()))
             if content:
                 invocation.add_input(content, step.id)
-        progress.set_outputs_for_input(step, step_outputs)
-        return job
+        progress.set_outputs_for_input(invocation_step, step_outputs)
 
-    def recover_mapping(self, step, step_invocations, progress):
-        progress.set_outputs_for_input(step)
+    def recover_mapping(self, invocation_step, progress):
+        progress.set_outputs_for_input(invocation_step)
 
 
 class InputDataModule(InputModule):
@@ -489,10 +503,10 @@ class InputParameterModule(WorkflowModule):
     def get_data_inputs(self):
         return []
 
-    def execute(self, trans, progress, invocation, step):
-        job, step_outputs = None, dict(output=step.state.inputs['input'])
-        progress.set_outputs_for_input(step, step_outputs)
-        return job
+    def execute(self, trans, progress, invocation_step):
+        step = invocation_step.workflow_step
+        step_outputs = dict(output=step.state.inputs['input'])
+        progress.set_outputs_for_input(invocation_step, step_outputs)
 
 
 class PauseModule(WorkflowModule):
@@ -520,18 +534,18 @@ class PauseModule(WorkflowModule):
         state.inputs = dict()
         return state
 
-    def execute(self, trans, progress, invocation, step):
+    def execute(self, trans, progress, invocation_step):
+        step = invocation_step.workflow_step
         progress.mark_step_outputs_delayed(step, why="executing pause step")
-        return None
 
-    def recover_mapping(self, step, step_invocations, progress):
-        if step_invocations:
-            step_invocation = step_invocations[0]
-            action = step_invocation.action
+    def recover_mapping(self, invocation_step, progress):
+        if invocation_step:
+            step = invocation_step.workflow_step
+            action = invocation_step.action
             if action:
                 connection = step.input_connections_by_name["input"][0]
                 replacement = progress.replacement_for_connection(connection)
-                progress.set_step_outputs(step, {'output': replacement})
+                progress.set_step_outputs(invocation_step, {'output': replacement})
                 return
             elif action is False:
                 raise CancelWorkflowEvaluation()
@@ -790,7 +804,9 @@ class ToolModule(WorkflowModule):
         else:
             raise ToolMissingException("Tool %s missing. Cannot recover runtime state." % self.tool_id)
 
-    def execute(self, trans, progress, invocation, step):
+    def execute(self, trans, progress, invocation_step):
+        invocation = invocation_step.workflow_invocation
+        step = invocation_step.workflow_step
         tool = trans.app.toolbox.get_tool(step.tool_id, tool_version=step.tool_version)
         tool_state = step.state
         # Not strictly needed - but keep Tool state clean by stripping runtime
@@ -855,6 +871,9 @@ class ToolModule(WorkflowModule):
 
             param_combinations.append(execution_state.inputs)
 
+        # Will be set if only a subset of required jobs have been scheduled and the
+        # workflow should be delayed.
+        partial_jobs = None
         try:
             execution_tracker = execute(
                 trans=self.trans,
@@ -862,50 +881,49 @@ class ToolModule(WorkflowModule):
                 param_combinations=param_combinations,
                 history=invocation.history,
                 collection_info=collection_info,
-                workflow_invocation_uuid=invocation.uuid.hex
+                workflow_invocation_uuid=invocation.uuid.hex,
+                invocation_step=invocation_step,
             )
+        except PartialJobExecution as p:
+            partial_jobs = p.jobs
         except ToolInputsNotReadyException:
             delayed_why = "tool [%s] inputs are not ready, this special tool requires inputs to be ready" % tool.id
             raise DelayedWorkflowEvaluation(why=delayed_why)
 
-        if collection_info:
-            step_outputs = dict(execution_tracker.implicit_collections)
+        if partial_jobs is None:
+            if collection_info:
+                step_outputs = dict(execution_tracker.implicit_collections)
+            else:
+                step_outputs = dict(execution_tracker.output_datasets)
+                step_outputs.update(execution_tracker.output_collections)
+            progress.set_step_outputs(invocation_step, step_outputs)
+            jobs = execution_tracker.successful_jobs
         else:
-            step_outputs = dict(execution_tracker.output_datasets)
-            step_outputs.update(execution_tracker.output_collections)
-        progress.set_step_outputs(step, step_outputs)
-        jobs = execution_tracker.successful_jobs
+            jobs = partial_jobs
+
         for job in jobs:
             self._handle_post_job_actions(step, job, invocation.replacement_dict)
+
         if execution_tracker.execution_errors:
             failed_count = len(execution_tracker.execution_errors)
             success_count = len(execution_tracker.successful_jobs)
             all_count = failed_count + success_count
             message = "Failed to create %d out of %s job(s) for workflow step." % (failed_count, all_count)
             raise Exception(message)
-        return jobs
 
-    def recover_mapping(self, step, step_invocations, progress):
-        # Grab a job representing this invocation - for normal workflows
-        # there will be just one job but if this step was mapped over there
-        # may be many.
-        job_0 = step_invocations[0].job
+        complete = partial_jobs is None
+        return jobs, complete
 
+    def recover_mapping(self, invocation_step, progress):
         outputs = {}
-        for job_output in job_0.output_datasets:
-            replacement_name = job_output.name
-            replacement_value = job_output.dataset
-            # If was a mapping step, grab the output mapped collection for
-            # replacement instead.
-            if replacement_value.hidden_beneath_collection_instance:
-                replacement_value = replacement_value.hidden_beneath_collection_instance
-            outputs[replacement_name] = replacement_value
-        for job_output_collection in job_0.output_dataset_collection_instances:
-            replacement_name = job_output_collection.name
-            replacement_value = job_output_collection.dataset_collection_instance
-            outputs[replacement_name] = replacement_value
 
-        progress.set_step_outputs(step, outputs)
+        for output_dataset_assoc in invocation_step.output_datasets:
+            outputs[output_dataset_assoc.output_name] = output_dataset_assoc.dataset
+
+        for output_dataset_collection_assoc in invocation_step.output_dataset_collections:
+            outputs[output_dataset_collection_assoc.output_name] = output_dataset_collection_assoc.dataset_collection
+
+        progress.set_step_outputs(invocation_step, outputs)
 
     def _find_collections_to_match(self, tool, progress, step):
         collections_to_match = matching.CollectionsToMatch()

lib/galaxy/workflow/run.py

@@ -162,24 +162,46 @@ class WorkflowInvoker(object):
 
         remaining_steps = self.progress.remaining_steps()
         delayed_steps = False
-        for step in remaining_steps:
+        for (step, workflow_invocation_step) in remaining_steps:
             step_delayed = False
             step_timer = ExecutionTimer()
             jobs = None
             try:
                 self.__check_implicitly_dependent_steps(step)
 
-                # TODO: step may fail to invoke, do something about that.
-                jobs = self._invoke_step(step)
-                for job in (util.listify(jobs) or [None]):
-                    # Record invocation
+                if not workflow_invocation_step:
                     workflow_invocation_step = model.WorkflowInvocationStep()
                     workflow_invocation_step.workflow_invocation = workflow_invocation
                     workflow_invocation_step.workflow_step = step
+                    workflow_invocation_step.state = 'new'
+                    previously_executed_jobs_count = 0
+
+                    workflow_invocation.steps.append(workflow_invocation_step)
+                else:
+                    previously_executed_jobs_count = len(workflow_invocation_step.jobs)
+
+                jobs_or_none = self._invoke_step(workflow_invocation_step)
+                if jobs_or_none:
+                    jobs, complete = jobs_or_none
+                else:
+                    jobs, complete = [], True
+
+                for job in (util.listify(jobs) or []):
+                    job_assoc = model.WorkflowInvocationStepJobAssociation()
+                    job_assoc.index = previously_executed_jobs_count
+                    job_assoc.workflow_invocation_step = workflow_invocation_step
                     # Job may not be generated in this thread if bursting is enabled
                     # https://github.com/galaxyproject/galaxy/issues/2259
-                    if job:
-                        workflow_invocation_step.job_id = job.id
+                    job_assoc.job_id = job.id
+
+                    previously_executed_jobs_count += 1
+
+                if not complete:
+                    step_delayed = delayed_steps = True
+                    workflow_invocation_step.state = 'ready'
+                    self.progress.mark_step_outputs_delayed(step, why="Not all jobs scheduled for state.")
+                else:
+                    workflow_invocation_step.state = 'scheduled'
             except modules.DelayedWorkflowEvaluation as de:
                 step_delayed = delayed_steps = True
                 self.progress.mark_step_outputs_delayed(step, why=de.why)
@@ -218,15 +240,19 @@ class WorkflowInvoker(object):
                 self.__check_implicitly_dependent_step(output_id)
 
     def __check_implicitly_dependent_step(self, output_id):
-        step_invocations = self.workflow_invocation.step_invocations_for_step_id(output_id)
+        step_invocation = self.workflow_invocation.step_invocation_for_step_id(output_id)
 
         # No steps created yet - have to delay evaluation.
-        if not step_invocations:
+        if not step_invocation:
             delayed_why = "depends on step [%s] but that step has not been invoked yet" % output_id
             raise modules.DelayedWorkflowEvaluation(why=delayed_why)
 
-        for step_invocation in step_invocations:
-            job = step_invocation.job
+        if step_invocation.state != 'scheduled':
+            delayed_why = "depends on step [%s] job has not finished scheduling yet" % output_id
+            raise modules.DelayedWorkflowEvaluation(delayed_why)
+
+        for job_assoc in step_invocation.jobs:
+            job = job_assoc.job
             if job:
                 # At least one job in incomplete.
                 if not job.finished:
@@ -241,9 +267,9 @@ class WorkflowInvoker(object):
                 # pause steps.
                 pass
 
-    def _invoke_step(self, step):
-        jobs = step.module.execute(self.trans, self.progress, self.workflow_invocation, step)
-        return jobs
+    def _invoke_step(self, invocation_step):
+        jobs_or_none = invocation_step.workflow_step.module.execute(self.trans, self.progress, invocation_step)
+        return jobs_or_none
 
 
 STEP_OUTPUT_DELAYED = object()
@@ -256,11 +282,19 @@ class WorkflowProgress(object):
         self.module_injector = module_injector
         self.workflow_invocation = workflow_invocation
         self.inputs_by_step_id = inputs_by_step_id
+        self.jobs_per_scheduling_iteration = 1
+
+    @property
+    def maximum_jobs_to_schedule(self):
+        return 1
 
     def remaining_steps(self):
         # Previously computed and persisted step states.
         step_states = self.workflow_invocation.step_states_by_step_id()
         steps = self.workflow_invocation.workflow.steps
+
+        # TODO: Wouldn't a generator be much better here so we don't have to reason about
+        # steps we are no where near ready to schedule?
         remaining_steps = []
         step_invocations_by_id = self.workflow_invocation.step_invocations_by_step_id()
         for step in steps:
@@ -274,11 +308,11 @@ class WorkflowProgress(object):
                 runtime_state = step_states[step_id].value
                 step.state = step.module.decode_runtime_state(runtime_state)
 
-            invocation_steps = step_invocations_by_id.get(step_id, None)
-            if invocation_steps:
-                self._recover_mapping(step, invocation_steps)
+            invocation_step = step_invocations_by_id.get(step_id, None)
+            if invocation_step and invocation_step.state == 'scheduled':
+                self._recover_mapping(invocation_step)
             else:
-                remaining_steps.append(step)
+                remaining_steps.append((step, invocation_step))
         return remaining_steps
 
     def replacement_for_tool_input(self, step, input, prefixed_name):
@@ -340,7 +374,9 @@ class WorkflowProgress(object):
         output_name = workflow_output.output_name
         return self.outputs[step.id][output_name]
 
-    def set_outputs_for_input(self, step, outputs=None):
+    def set_outputs_for_input(self, invocation_step, outputs=None):
+        step = invocation_step.workflow_step
+
         if outputs is None:
             outputs = {}
 
@@ -352,10 +388,34 @@ class WorkflowProgress(object):
                 raise ValueError(message)
             outputs['output'] = self.inputs_by_step_id[step_id]
 
-        self.set_step_outputs(step, outputs)
+        self.set_step_outputs(invocation_step, outputs)
 
-    def set_step_outputs(self, step, outputs):
+    def set_step_outputs(self, invocation_step, outputs, already_persisted=False):
+        step = invocation_step.workflow_step
         self.outputs[step.id] = outputs
+        if not already_persisted:
+            for output_name, output_object in outputs.items():
+                if hasattr(output_object, "history_content_type"):
+                    invocation_step.add_output(output_name, output_object)
+                else:
+                    # This is a problem, this non-data, non-collection output
+                    # won't be recovered on a subsequent workflow scheduling
+                    # iteration. This seems to have been a pre-existing problem
+                    # prior to #4584 though.
+                    pass
+            for workflow_output in step.workflow_outputs:
+                output_name = workflow_output.output_name
+                if output_name not in outputs:
+                    raise KeyError("Failed to find [%s] in step outputs [%s]" % (output_name, outputs))
+                output = outputs[output_name]
+                self._record_workflow_output(
+                    step,
+                    workflow_output,
+                    output=output,
+                )
+
+    def _record_workflow_output(self, step, workflow_output, output):
+        self.workflow_invocation.add_output(workflow_output, step, output)
 
     def mark_step_outputs_delayed(self, step, why=None):
         if why:
@@ -414,11 +474,11 @@ class WorkflowProgress(object):
             self.module_injector,
         )
 
-    def _recover_mapping(self, step, step_invocations):
+    def _recover_mapping(self, step_invocation):
         try:
-            step.module.recover_mapping(step, step_invocations, self)
+            step_invocation.workflow_step.module.recover_mapping(step_invocation, self)
         except modules.DelayedWorkflowEvaluation as de:
-            self.mark_step_outputs_delayed(step, de.why)
+            self.mark_step_outputs_delayed(step_invocation.workflow_step, de.why)
 
 
 __all__ = ('invoke', 'WorkflowRunConfig')

test/api/test_workflows.py

@@ -963,6 +963,77 @@ test_data:
         time.sleep(5)
         self.dataset_populator.wait_for_history(history_id, assert_ok=True)
 
+    def test_workflow_output_dataset(self):
+        history_id = self.dataset_populator.new_history()
+        summary = self._run_jobs("""
+class: GalaxyWorkflow
+inputs:
+  - id: input1
+outputs:
+  - id: wf_output_1
+    source: first_cat#out_file1
+steps:
+  - tool_id: cat1
+    label: first_cat
+    state:
+      input1:
+        $link: input1
+
+test_data:
+  input1: "hello world"
+""", history_id=history_id)
+        workflow_id = summary.workflow_id
+        invocation_id = summary.invocation_id
+        invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
+        self._assert_status_code_is(invocation_response, 200)
+        invocation = invocation_response.json()
+        self._assert_has_keys(invocation , "id", "outputs", "output_collections")
+        assert len(invocation["output_collections"]) == 0
+        assert len(invocation["outputs"]) == 1
+        output_content = self.dataset_populator.get_history_dataset_content(history_id, dataset_id=invocation["outputs"]["wf_output_1"]["id"])
+        assert "hello world" == output_content.strip()
+
+    def test_workflow_output_dataset_collection(self):
+        history_id = self.dataset_populator.new_history()
+        summary = self._run_jobs("""
+class: GalaxyWorkflow
+inputs:
+  - id: input1
+    type: data_collection_input
+    collection_type: list
+outputs:
+  - id: wf_output_1
+    source: first_cat#out_file1
+steps:
+  - tool_id: cat
+    label: first_cat
+    state:
+      input1:
+        $link: input1
+test_data:
+  input1:
+    type: list
+    name: the_dataset_list
+    elements:
+      - identifier: el1
+        value: 1.fastq
+        type: File
+""", history_id=history_id)
+        workflow_id = summary.workflow_id
+        invocation_id = summary.invocation_id
+        invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id))
+        self._assert_status_code_is(invocation_response, 200)
+        invocation = invocation_response.json()
+        self._assert_has_keys(invocation , "id", "outputs", "output_collections")
+        assert len(invocation["output_collections"]) == 1
+        assert len(invocation["outputs"]) == 0
+        output_content = self.dataset_populator.get_history_collection_details(history_id, content_id=invocation["output_collections"]["wf_output_1"]["id"])
+        self._assert_has_keys(output_content , "id", "elements")
+        elements = output_content["elements"]
+        assert len(elements) == 1
+        elements0 = elements[0]
+        assert elements0["element_identifier"] == "el1"
+
     @skip_without_tool("cat")
     def test_cancel_new_workflow_when_history_deleted(self):
         with self.dataset_populator.test_history() as history_id:

test/base/populators.py

@@ -282,6 +282,8 @@ class BaseDatasetPopulator(object):
         # the last dataset in the history will be fetched.
         if "dataset_id" in kwds:
             history_content_id = kwds["dataset_id"]
+        elif "content_id" in kwds:
+            history_content_id = kwds["content_id"]
         elif "dataset" in kwds:
             history_content_id = kwds["dataset"]["id"]
         else:

test/unit/workflows/test_workflow_progress.py

@@ -72,13 +72,17 @@ class WorkflowProgressTestCase(unittest.TestCase):
             self.invocation, self.inputs_by_step_id, MockModuleInjector(self.progress)
         )
 
-    def _set_previous_progress(self, outputs_dict):
-        for step_id, step_value in outputs_dict.items():
+    def _set_previous_progress(self, outputs):
+        for i, (step_id, step_value) in enumerate(outputs):
             if step_value is not UNSCHEDULED_STEP:
                 self.progress[step_id] = step_value
 
                 workflow_invocation_step = model.WorkflowInvocationStep()
                 workflow_invocation_step.workflow_step_id = step_id
+                workflow_invocation_step.state = 'scheduled'
+                workflow_invocation_step.workflow_step = self._step(i)
+                self.assertEqual(step_id, self._step(i).id)
+                # workflow_invocation_step.workflow_invocation = self.invocation
                 self.invocation.steps.append(workflow_invocation_step)
 
             workflow_invocation_step_state = model.WorkflowRequestStepState()
@@ -89,13 +93,21 @@ class WorkflowProgressTestCase(unittest.TestCase):
     def _step(self, index):
         return self.invocation.workflow.steps[index]
 
+    def _invocation_step(self, index):
+        if index < len(self.invocation.steps):
+            return self.invocation.steps[index]
+        else:
+            workflow_invocation_step = model.WorkflowInvocationStep()
+            workflow_invocation_step.workflow_step = self._step(index)
+            return workflow_invocation_step
+
     def test_connect_data_input(self):
         self._setup_workflow(TEST_WORKFLOW_YAML)
         hda = model.HistoryDatasetAssociation()
 
         self.inputs_by_step_id = {100: hda}
         progress = self._new_workflow_progress()
-        progress.set_outputs_for_input(self._step(0))
+        progress.set_outputs_for_input(self._invocation_step(0))
 
         conn = model.WorkflowStepConnection()
         conn.output_name = "output"
@@ -108,7 +120,7 @@ class WorkflowProgressTestCase(unittest.TestCase):
 
         self.inputs_by_step_id = {100: hda}
         progress = self._new_workflow_progress()
-        progress.set_outputs_for_input(self._step(0))
+        progress.set_outputs_for_input(self._invocation_step(0))
 
         replacement = progress.replacement_for_tool_input(self._step(2), MockInput(), "input1")
         assert replacement is hda
@@ -118,7 +130,7 @@ class WorkflowProgressTestCase(unittest.TestCase):
         hda = model.HistoryDatasetAssociation()
 
         progress = self._new_workflow_progress()
-        progress.set_step_outputs(self._step(2), {"out1": hda})
+        progress.set_step_outputs(self._invocation_step(2), {"out1": hda})
 
         conn = model.WorkflowStepConnection()
         conn.output_name = "out1"
@@ -128,17 +140,18 @@ class WorkflowProgressTestCase(unittest.TestCase):
     def test_remaining_steps_with_progress(self):
         self._setup_workflow(TEST_WORKFLOW_YAML)
         hda3 = model.HistoryDatasetAssociation()
-        self._set_previous_progress({
-            100: {"output": model.HistoryDatasetAssociation()},
-            101: {"output": model.HistoryDatasetAssociation()},
-            102: {"out_file1": hda3},
-            103: {"out_file1": model.HistoryDatasetAssociation()},
-            104: UNSCHEDULED_STEP,
-        })
+        self._set_previous_progress([
+            (100, {"output": model.HistoryDatasetAssociation()}),
+            (101, {"output": model.HistoryDatasetAssociation()}),
+            (102, {"out_file1": hda3}),
+            (103, {"out_file1": model.HistoryDatasetAssociation()}),
+            (104, UNSCHEDULED_STEP),
+        ])
         progress = self._new_workflow_progress()
         steps = progress.remaining_steps()
-        assert len(steps) == 1
-        assert steps[0] is self.invocation.workflow.steps[4]
+        assert len(steps) == 1, steps
+        step, invocation_step = steps[0]
+        assert step is self.invocation.workflow.steps[4]
 
         replacement = progress.replacement_for_tool_input(self._step(4), MockInput(), "input1")
         assert replacement is hda3
@@ -151,21 +164,27 @@ class WorkflowProgressTestCase(unittest.TestCase):
     def test_subworkflow_progress(self):
         self._setup_workflow(TEST_SUBWORKFLOW_YAML)
         hda = model.HistoryDatasetAssociation()
-        self._set_previous_progress({
-            100: {"output": hda},
-            101: UNSCHEDULED_STEP,
-        })
+        self._set_previous_progress([
+            (100, {"output": hda}),
+            (101, UNSCHEDULED_STEP),
+        ])
         self.invocation.create_subworkflow_invocation_for_step(
             self.invocation.workflow.step_by_index(1)
         )
         progress = self._new_workflow_progress()
         remaining_steps = progress.remaining_steps()
-        subworkflow_step = remaining_steps[0]
+        (subworkflow_step, subworkflow_invocation_step) = remaining_steps[0]
         subworkflow_progress = progress.subworkflow_progress(subworkflow_step)
         subworkflow = subworkflow_step.subworkflow
         assert subworkflow_progress.workflow_invocation.workflow == subworkflow
+
         subworkflow_input_step = subworkflow.step_by_index(0)
-        subworkflow_progress.set_outputs_for_input(subworkflow_input_step)
+        subworkflow_invocation_step = model.WorkflowInvocationStep()
+        subworkflow_invocation_step.workflow_step_id = subworkflow_input_step.id
+        subworkflow_invocation_step.state = 'new'
+        subworkflow_invocation_step.workflow_step = subworkflow_input_step
+
+        subworkflow_progress.set_outputs_for_input(subworkflow_invocation_step)
 
         subworkflow_cat_step = subworkflow.step_by_index(1)
 
@@ -200,7 +219,7 @@ class MockModule(object):
     def decode_runtime_state(self, runtime_state):
         return True
 
-    def recover_mapping(self, step, step_invocations, progress):
-        step_id = step.id
+    def recover_mapping(self, invocation_step, progress):
+        step_id = invocation_step.workflow_step.id
         if step_id in self.progress:
-            progress.set_step_outputs(step, self.progress[step_id])
+            progress.set_step_outputs(invocation_step, self.progress[step_id])