docs: def _ProcessCompletedRequests(self, notification):
      """Does the actual processing of the completed requests."""
      # First ensure that client messages are all removed. NOTE: We make a new
      # queue manager here because we want only the client messages to be removed
      # ASAP. This must happen before we actually run the flow to ensure the
      # client requests are removed from the client queues.
      with queue_manager.QueueManager(token=self.token) as manager:
        for request, _ in manager.FetchCompletedRequests(
            self.session_id, timestamp=(0, notification.timestamp)):
          # Requests which are not destined to clients have no embedded request
          # message.
          if request.HasField("request"):
            manager.DeQueueClientRequest(request.request)
  
      # The flow is dead - remove all outstanding requests and responses.
      if not self.IsRunning():
        self.queue_manager.DestroyFlowStates(self.session_id)
        return
  
      processing = []
      while True:
        try:
          # Here we only care about completed requests - i.e. those requests with
          # responses followed by a status message.
          for request, responses in self.queue_manager.FetchCompletedResponses(
              self.session_id, timestamp=(0, notification.timestamp)):
  
            if request.id == 0:
              continue
  
            if not responses:
              break
  
            # We are missing a needed request - maybe its not completed yet.
            if request.id > self.context.next_processed_request:
              stats_collector_instance.Get().IncrementCounter(
                  "grr_response_out_of_order")
              break
  
            # Not the request we are looking for - we have seen it before
            # already.
            if request.id < self.context.next_processed_request:
              self.queue_manager.DeleteRequest(request)
              continue
  
            if not responses:
              continue
  
            # Do we have all the responses here? This can happen if some of the
            # responses were lost.
            if len(responses) != responses[-1].response_id:
              # If we can retransmit do so. Note, this is different from the
              # automatic retransmission facilitated by the task scheduler (the
              # Task.task_ttl field) which would happen regardless of these.
              if request.transmission_count < 5:
                stats_collector_instance.Get().IncrementCounter(
                    "grr_request_retransmission_count")
                request.transmission_count += 1
                self.ReQueueRequest(request)
              break
  
            # If we get here its all good - run the flow.
            if self.IsRunning():
              self.flow_obj.HeartBeat()
              self.RunStateMethod(request.next_state, request, responses)
  
            # Quit early if we are no longer alive.
            else:
              break
  
            # At this point we have processed this request - we can remove it and
            # its responses from the queue.
            self.queue_manager.DeleteRequest(request)
            self.context.next_processed_request += 1
            self.DecrementOutstandingRequests()
  
          # Are there any more outstanding requests?
          if not self.OutstandingRequests():
            # Allow the flow to cleanup
            if self.IsRunning() and self.context.current_state != "End":
              self.RunStateMethod("End")
  
          # Rechecking the OutstandingRequests allows the End state (which was
          # called above) to issue further client requests - hence postpone
          # termination.
          if not self.OutstandingRequests():
            # TODO(user): Deprecate in favor of 'flow_completions' metric.
            stats_collector_instance.Get().IncrementCounter(
                "grr_flow_completed_count")
  
            stats_collector_instance.Get().IncrementCounter(
                "flow_completions", fields=[self.flow_obj.Name()])
            logging.debug(
                "Destroying session %s(%s) for client %s", self.session_id,
                self.flow_obj.Name(), self.runner_args.client_id)
  
            self.flow_obj.Terminate()
  
          # We are done here.
          return
  
        except queue_manager.MoreDataException:
          # Join any threads.
          for event in processing:
            event.wait()
  
          # We did not read all the requests/responses in this run in order to
          # keep a low memory footprint and have to make another pass.
          self.FlushMessages()
          self.flow_obj.Flush()
          continue
  
        finally:
          # Join any threads.
          for event in processing:
            event.wait()