docs: def transition_run(self,
                       pipeline_key,
                       blocking_slot_keys=None,
                       fanned_out_pipelines=None,
                       pipelines_to_run=None):
      """Marks an asynchronous or generator pipeline as running.
  
      Does nothing if the pipeline is no longer in a runnable state.
  
      Args:
        pipeline_key: The db.Key of the _PipelineRecord to update.
        blocking_slot_keys: List of db.Key instances that this pipeline's
          finalization barrier should wait on in addition to the existing one.
          This is used to update the barrier to include all child outputs. When
          None, the barrier will not be updated.
        fanned_out_pipelines: List of db.Key instances of _PipelineRecords that
          were fanned out by this generator pipeline. This is distinct from the
          'pipelines_to_run' list because not all of the pipelines listed here
          will be immediately ready to execute. When None, then this generator
          yielded no children.
        pipelines_to_run: List of db.Key instances of _PipelineRecords that should
          be kicked off (fan-out) transactionally as part of this transition.
          When None, no child pipelines will run. All db.Keys in this list must
          also be present in the fanned_out_pipelines list.
  
      Raises:
        UnexpectedPipelineError if blocking_slot_keys was not empty and the
        _BarrierRecord has gone missing.
      """
      def txn():
        pipeline_record = db.get(pipeline_key)
        if pipeline_record is None:
          logging.warning('Pipeline ID "%s" cannot be marked as run. '
                          'Does not exist.', pipeline_key.name())
          raise db.Rollback()
        if pipeline_record.status != _PipelineRecord.WAITING:
          logging.warning('Pipeline ID "%s" in bad state to be marked as run: %s',
                          pipeline_key.name(), pipeline_record.status)
          raise db.Rollback()
  
        pipeline_record.status = _PipelineRecord.RUN
  
        if fanned_out_pipelines:
          # NOTE: We must model the pipeline relationship in a top-down manner,
          # meaning each pipeline must point forward to the pipelines that it
          # fanned out to. The reason is race conditions. If evaluate()
          # dies early, it may create many unused _PipelineRecord and _SlotRecord
          # instances that never progress. The only way we know which of these
          # are valid is by traversing the graph from the root, where the
          # fanned_out property refers to those pipelines that were run using a
          # transactional task.
          child_pipeline_list = list(fanned_out_pipelines)
          pipeline_record.fanned_out = child_pipeline_list
  
          if pipelines_to_run:
            child_indexes = [
                child_pipeline_list.index(p) for p in pipelines_to_run]
            child_indexes.sort()
            task = taskqueue.Task(
                url=self.fanout_handler_path,
                params=dict(parent_key=str(pipeline_key),
                            child_indexes=child_indexes))
            task.add(queue_name=self.queue_name, transactional=True)
  
        pipeline_record.put()
  
        if blocking_slot_keys:
          # NOTE: Always update a generator pipeline's finalization barrier to
          # include all of the outputs of any pipelines that it runs, to ensure
          # that finalized calls will not happen until all child pipelines have
          # completed. This must happen transactionally with the enqueue of
          # the fan-out kickoff task above to ensure the child output slots and
          # the barrier blocking slots are the same.
          barrier_key = db.Key.from_path(
              _BarrierRecord.kind(), _BarrierRecord.FINALIZE,
              parent=pipeline_key)
          finalize_barrier = db.get(barrier_key)
          if finalize_barrier is None:
            raise UnexpectedPipelineError(
                'Pipeline ID "%s" cannot update finalize barrier. '
                'Does not exist.' % pipeline_key.name())
          else:
            finalize_barrier.blocking_slots = list(
                blocking_slot_keys.union(set(finalize_barrier.blocking_slots)))
            finalize_barrier.put()
  
      db.run_in_transaction(txn)