def _try_acquire_lease(self, shard_state, tstate):
    """Validate datastore and the task payload are consistent.

    If so, attempt to get a lease on this slice's execution.
    See model.ShardState doc on slice_start_time.

    Args:
      shard_state: model.ShardState from datastore.
      tstate: model.TransientShardState from taskqueue paylod.

    Returns:
      A _TASK_DIRECTIVE enum. PROCEED_TASK if lock is acquired.
    RETRY_TASK if task should be retried, DROP_TASK if task should
    be dropped. Only old tasks (comparing to datastore state)
    will be dropped. Future tasks are retried until they naturally
    become old so that we don't ever stuck MR.
    """
    # Controller will tally shard_states and properly handle the situation.
    if not shard_state:
      logging.warning("State not found for shard %s; Possible spurious task "
                      "execution. Dropping this task.",
                      tstate.shard_id)
      return self._TASK_DIRECTIVE.DROP_TASK

    if not shard_state.active:
      logging.warning("Shard %s is not active. Possible spurious task "
                      "execution. Dropping this task.", tstate.shard_id)
      logging.warning(str(shard_state))
      return self._TASK_DIRECTIVE.DROP_TASK

    # Validate shard retry count.
    if shard_state.retries > tstate.retries:
      logging.warning(
          "Got shard %s from previous shard retry %s. Possible spurious "
          "task execution. Dropping this task.",
          tstate.shard_id,
          tstate.retries)
      logging.warning(str(shard_state))
      return self._TASK_DIRECTIVE.DROP_TASK
    elif shard_state.retries < tstate.retries:
      # By the end of last slice, task enqueue succeeded but datastore commit
      # failed. That transaction will be retried and adding the same task
      # will pass.
      logging.warning(
          "ShardState for %s is behind slice. Waiting for it to catch up",
          shard_state.shard_id)
      return self._TASK_DIRECTIVE.RETRY_TASK

    # Validate slice id.
    # Taskqueue executes old successful tasks.
    if shard_state.slice_id > tstate.slice_id:
      logging.warning(
          "Task %s-%s is behind ShardState %s. Dropping task.""",
          tstate.shard_id, tstate.slice_id, shard_state.slice_id)
      return self._TASK_DIRECTIVE.DROP_TASK
    # By the end of last slice, task enqueue succeeded but datastore commit
    # failed. That transaction will be retried and adding the same task
    # will pass. User data is duplicated in this case.
    elif shard_state.slice_id < tstate.slice_id:
      logging.warning(
          "Task %s-%s is ahead of ShardState %s. Waiting for it to catch up.",
          tstate.shard_id, tstate.slice_id, shard_state.slice_id)
      return self._TASK_DIRECTIVE.RETRY_TASK

    # Check potential duplicated tasks for the same slice.
    # See model.ShardState doc.
    if shard_state.slice_start_time:
      countdown = self._wait_time(shard_state,
                                  parameters._LEASE_DURATION_SEC)
      if countdown > 0:
        logging.warning(
            "Last retry of slice %s-%s may be still running."
            "Will try again in %s seconds", tstate.shard_id, tstate.slice_id,
            countdown)
        # TODO(user): There might be a better way. Taskqueue's countdown
        # only applies to add new tasks, not retry of tasks.
        # Reduce contention.
        time.sleep(countdown)
        return self._TASK_DIRECTIVE.RETRY_TASK
      # lease could have expired. Verify with logs API.
      else:
        if self._wait_time(shard_state,
                           parameters._MAX_LEASE_DURATION_SEC):
          if not self._has_old_request_ended(shard_state):
            logging.warning(
                "Last retry of slice %s-%s is still in flight with request_id "
                "%s. Will try again later.", tstate.shard_id, tstate.slice_id,
                shard_state.slice_request_id)
            return self._TASK_DIRECTIVE.RETRY_TASK
        else:
          logging.warning(
              "Last retry of slice %s-%s has no log entry and has"
              "timed out after %s seconds",
              tstate.shard_id, tstate.slice_id,
              parameters._MAX_LEASE_DURATION_SEC)

    # Lease expired or slice_start_time not set.
    config = util.create_datastore_write_config(tstate.mapreduce_spec)
    @db.transactional(retries=5)
    def _tx():
      """Use datastore to set slice_start_time to now.

      If failed for any reason, raise error to retry the task (hence all
      the previous validation code). The task would die naturally eventually.

      Raises:
        Rollback: If the shard state is missing.

      Returns:
        A _TASK_DIRECTIVE enum.
      """
      fresh_state = model.ShardState.get_by_shard_id(tstate.shard_id)
      if not fresh_state:
        logging.warning("ShardState missing.")
        raise db.Rollback()
      if (fresh_state.active and
          fresh_state.slice_id == shard_state.slice_id and
          fresh_state.slice_start_time == shard_state.slice_start_time):
        shard_state.slice_start_time = datetime.datetime.now()
        shard_state.slice_request_id = os.environ.get("REQUEST_LOG_ID")
        shard_state.acquired_once = True
        shard_state.put(config=config)
        return self._TASK_DIRECTIVE.PROCEED_TASK
      else:
        logging.warning(
            "Contention on slice %s-%s execution. Will retry again.",
            tstate.shard_id, tstate.slice_id)
        # One proposer should win. In case all lost, back off arbitrarily.
        time.sleep(random.randrange(1, 5))
        return self._TASK_DIRECTIVE.RETRY_TASK

    return _tx()
Validate datastore and the task payload are consistent.

    If so, attempt to get a lease on this slice's execution.
    See model.ShardState doc on slice_start_time.

    Args:
      shard_state: model.ShardState from datastore.
      tstate: model.TransientShardState from taskqueue paylod.

    Returns:
      A _TASK_DIRECTIVE enum. PROCEED_TASK if lock is acquired.
    RETRY_TASK if task should be retried, DROP_TASK if task should
    be dropped. Only old tasks (comparing to datastore state)
    will be dropped. Future tasks are retried until they naturally
    become old so that we don't ever stuck MR.