docs: def _save_state_and_schedule_next(self, shard_state, tstate, task_directive):
      """Save state and schedule task.
  
      Save shard state to datastore.
      Schedule next slice if needed.
      Set HTTP response code.
      No modification to any shard_state or tstate.
  
      Args:
        shard_state: model.ShardState for current shard.
        tstate: model.TransientShardState for current shard.
        task_directive: enum _TASK_DIRECTIVE.
  
      Returns:
        The task to retry if applicable.
      """
      spec = tstate.mapreduce_spec
  
      if task_directive == self._TASK_DIRECTIVE.DROP_TASK:
        return
      if task_directive in (self._TASK_DIRECTIVE.RETRY_SLICE,
                            self._TASK_DIRECTIVE.RETRY_TASK):
        # Set HTTP code to 500.
        return self.retry_task()
      elif task_directive == self._TASK_DIRECTIVE.ABORT_SHARD:
        logging.info("Aborting shard %d of job '%s'",
                     shard_state.shard_number, shard_state.mapreduce_id)
        task = None
      elif task_directive == self._TASK_DIRECTIVE.FAIL_TASK:
        logging.critical("Shard %s failed permanently.", shard_state.shard_id)
        task = None
      elif task_directive == self._TASK_DIRECTIVE.RETRY_SHARD:
        logging.warning("Shard %s is going to be attempted for the %s time.",
                        shard_state.shard_id,
                        shard_state.retries + 1)
        task = self._state_to_task(tstate, shard_state)
      elif task_directive == self._TASK_DIRECTIVE.RECOVER_SLICE:
        logging.warning("Shard %s slice %s is being recovered.",
                        shard_state.shard_id,
                        shard_state.slice_id)
        task = self._state_to_task(tstate, shard_state)
      else:
        assert task_directive == self._TASK_DIRECTIVE.PROCEED_TASK
        countdown = self._get_countdown_for_next_slice(spec)
        task = self._state_to_task(tstate, shard_state, countdown=countdown)
  
      # Prepare parameters for db transaction and taskqueue.
      queue_name = os.environ.get("HTTP_X_APPENGINE_QUEUENAME",
                                  # For test only.
                                  # TODO(user): Remove this.
                                  "default")
      config = util.create_datastore_write_config(spec)
  
      @db.transactional(retries=5)
      def _tx():
        """The Transaction helper."""
        fresh_shard_state = model.ShardState.get_by_shard_id(tstate.shard_id)
        if not fresh_shard_state:
          raise db.Rollback()
        if (not fresh_shard_state.active or
            "worker_active_state_collision" in _TEST_INJECTED_FAULTS):
          logging.warning("Shard %s is not active. Possible spurious task "
                          "execution. Dropping this task.", tstate.shard_id)
          logging.warning("Datastore's %s", str(fresh_shard_state))
          logging.warning("Slice's %s", str(shard_state))
          return
        fresh_shard_state.copy_from(shard_state)
        fresh_shard_state.put(config=config)
        # Add task in the same datastore transaction.
        # This way we guarantee taskqueue is never behind datastore states.
        # Old tasks will be dropped.
        # Future task won't run until datastore states catches up.
        if fresh_shard_state.active:
          # Not adding task transactionally.
          # transactional enqueue requires tasks with no name.
          self._add_task(task, spec, queue_name)
  
      try:
        _tx()
      except (datastore_errors.Error,
              taskqueue.Error,
              runtime.DeadlineExceededError,
              apiproxy_errors.Error), e:
        logging.warning(
            "Can't transactionally continue shard. "
            "Will retry slice %s %s for the %s time.",
            tstate.shard_id,
            tstate.slice_id,
            self.task_retry_count() + 1)
        self._try_free_lease(shard_state)
        raise e