Business logic helper for QueueQueryAndOwn().

def _QueueQueryAndOwn(self,
                        subject,
                        lease_seconds=100,
                        limit=1,
                        timestamp=None):
    """Business logic helper for QueueQueryAndOwn()."""
    tasks = []

    lease = int(lease_seconds * 1e6)

    # Only grab attributes with timestamps in the past.
    delete_attrs = set()
    serialized_tasks_dict = {}
    for predicate, task, timestamp in DB.ResolvePrefix(
        subject,
        DataStore.QUEUE_TASK_PREDICATE_PREFIX,
        timestamp=(0, timestamp or rdfvalue.RDFDatetime.Now())):
      task = rdf_flows.GrrMessage.FromSerializedString(task)
      task.leased_until = timestamp
      task.leased_by = utils.ProcessIdString()
      # Decrement the ttl
      task.task_ttl -= 1
      if task.task_ttl <= 0:
        # Remove the task if ttl is exhausted.
        delete_attrs.add(predicate)
        stats_collector_instance.Get().IncrementCounter(
            "grr_task_ttl_expired_count")
      else:
        if task.task_ttl != rdf_flows.GrrMessage.max_ttl - 1:
          stats_collector_instance.Get().IncrementCounter(
              "grr_task_retransmission_count")

        serialized_tasks_dict.setdefault(predicate,
                                         []).append(task.SerializeToString())
        tasks.append(task)
        if len(tasks) >= limit:
          break

    if delete_attrs or serialized_tasks_dict:
      # Update the timestamp on claimed tasks to be in the future and decrement
      # their TTLs, delete tasks with expired ttls.
      self.MultiSet(
          subject,
          serialized_tasks_dict,
          replace=True,
          timestamp=int(time.time() * 1e6) + lease,
          to_delete=delete_attrs)

    if delete_attrs:
      logging.info("TTL exceeded for %d messages on queue %s",
                   len(delete_attrs), subject)
    return tasks