docs: def transition_retry(self, pipeline_key, retry_message):
      """Marks the given pipeline as requiring another retry.
  
      Does nothing if all attempts have been exceeded.
  
      Args:
        pipeline_key: db.Key of the _PipelineRecord that needs to be retried.
        retry_message: User-supplied message indicating the reason for the retry.
      """
      def txn():
        pipeline_record = db.get(pipeline_key)
        if pipeline_record is None:
          logging.warning(
              'Tried to retry pipeline ID "%s" but it does not exist.',
              pipeline_key.name())
          raise db.Rollback()
        if pipeline_record.status not in (
            _PipelineRecord.WAITING, _PipelineRecord.RUN):
          logging.warning(
              'Tried to retry pipeline ID "%s", found bad state: %s',
              pipeline_key.name(), pipeline_record.status)
          raise db.Rollback()
  
        params = pipeline_record.params
        offset_seconds = (
            params['backoff_seconds'] *
            (params['backoff_factor'] ** pipeline_record.current_attempt))
        pipeline_record.next_retry_time = (
            self._gettime() + datetime.timedelta(seconds=offset_seconds))
        pipeline_record.current_attempt += 1
        pipeline_record.retry_message = retry_message
        pipeline_record.status = _PipelineRecord.WAITING
  
        if pipeline_record.current_attempt >= pipeline_record.max_attempts:
          root_pipeline_key = (
              _PipelineRecord.root_pipeline.get_value_for_datastore(
                  pipeline_record))
          logging.warning(
              'Giving up on pipeline ID "%s" after %d attempt(s); causing abort '
              'all the way to the root pipeline ID "%s"', pipeline_key.name(),
              pipeline_record.current_attempt, root_pipeline_key.name())
          # NOTE: We do *not* set the status to aborted here to ensure that
          # this pipeline will be finalized before it has been marked as aborted.
          pipeline_record.abort_message = (
              'Aborting after %d attempts' % pipeline_record.current_attempt)
          task = taskqueue.Task(
              url=self.fanout_abort_handler_path,
              params=dict(root_pipeline_key=root_pipeline_key))
          task.add(queue_name=self.queue_name, transactional=True)
        else:
          task = taskqueue.Task(
              url=self.pipeline_handler_path,
              eta=pipeline_record.next_retry_time,
              params=dict(pipeline_key=pipeline_key,
                          purpose=_BarrierRecord.START,
                          attempt=pipeline_record.current_attempt),
              headers={'X-Ae-Pipeline-Key': pipeline_key},
              target=pipeline_record.params['target'])
          task.add(queue_name=self.queue_name, transactional=True)
  
        pipeline_record.put()
  
      db.run_in_transaction(txn)