docs: def RunOnce(self):
      """Processes one set of messages from Task Scheduler.
  
      The worker processes new jobs from the task master. For each job
      we retrieve the session from the Task Scheduler.
  
      Returns:
          Total number of messages processed by this call.
      """
      start_time = time.time()
      processed = 0
  
      queue_manager = queue_manager_lib.QueueManager(token=self.token)
      for queue in self.queues:
        # Freezeing the timestamp used by queue manager to query/delete
        # notifications to avoid possible race conditions.
        queue_manager.FreezeTimestamp()
  
        fetch_messages_start = time.time()
        notifications = queue_manager.GetNotifications(queue)
        stats_collector_instance.Get().RecordEvent(
            "worker_time_to_retrieve_notifications",
            time.time() - fetch_messages_start)
  
        stuck_flows = []
        for n in notifications:
          if n.in_progress:
            stuck_flows.append(n)
  
        # Process stuck flows first
        if stuck_flows:
          self.ProcessStuckFlows(stuck_flows, queue_manager)
  
        notifications_available = []
        for notification in notifications:
          # Filter out session ids we already tried to lock but failed.
          if notification.session_id not in self.queued_flows:
            notifications_available.append(notification)
  
        try:
          # If we spent too much time processing what we have so far, the
          # active_sessions list might not be current. We therefore break here
          # so we can re-fetch a more up to date version of the list, and try
          # again later. The risk with running with an old active_sessions list
          # is that another worker could have already processed this message,
          # and when we try to process it, there is nothing to do - costing us a
          # lot of processing time. This is a tradeoff between checking the data
          # store for current information and processing out of date
          # information.
          processed += self.ProcessMessages(
              notifications_available, queue_manager,
              self.RUN_ONCE_MAX_SECONDS - (time.time() - start_time))
  
        # We need to keep going no matter what.
        except Exception as e:  # pylint: disable=broad-except
          logging.error("Error processing message %s. %s.", e,
                        traceback.format_exc())
          stats_collector_instance.Get().IncrementCounter("grr_worker_exceptions")
          if flags.FLAGS.pdb_post_mortem:
            pdb.post_mortem()
  
        queue_manager.UnfreezeTimestamp()
        # If we have spent too much time, stop.
        if (time.time() - start_time) > self.RUN_ONCE_MAX_SECONDS:
          return processed
      return processed