'Does the real work of self.QueryAndOwn().'

def _QueryAndOwn(self, subject, lease_seconds=100, limit=1, user=''):

    tasks = []
    lease = long((lease_seconds * 1000000.0))
    delete_attrs = set()
    serialized_tasks_dict = {}
    for (predicate, task, timestamp) in self.data_store.ResolvePrefix(subject, self.TASK_PREDICATE_PREFIX, timestamp=(0, (self.frozen_timestamp or rdfvalue.RDFDatetime.Now())), token=self.token):
        task = rdf_flows.GrrMessage.FromSerializedString(task)
        task.eta = timestamp
        task.last_lease = ('%s@%s:%d' % (user, socket.gethostname(), os.getpid()))
        task.task_ttl -= 1
        if (task.task_ttl <= 0):
            delete_attrs.add(predicate)
            stats.STATS.IncrementCounter('grr_task_ttl_expired_count')
        else:
            if (task.task_ttl != (rdf_flows.GrrMessage.max_ttl - 1)):
                stats.STATS.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):
        self.data_store.MultiSet(subject, serialized_tasks_dict, replace=True, timestamp=(long((time.time() * 1000000.0)) + lease), sync=True, to_delete=delete_attrs, token=self.token)
    if delete_attrs:
        logging.info('TTL    exceeded    for    %d    messages    on    queue    %s', len(delete_attrs), subject)
    return tasks
