docs: 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