Method drain

crates/consensus/engine/src/task_queue/core.rs:532–555

Attempts to drain the queue by executing all [`EngineTask`]s in-order. If any task returns an error along the way, it is not popped from the queue (in case it must be retried) and the error is returned. Exception: tasks that fail with [`EngineTaskErrorSeverity::Flush`] are popped from the queue before the error is returned. The poisoned task must not be retried in-place — the engine processor wil

(&mut self)

    /// to popping, mirroring the success path so watch consumers (e.g. the RPC state view) stay
    /// in sync.
    pub async fn drain(&mut self) -> Result<(), EngineTaskErrors> {
        // Drain tasks in order of priority, halting on errors for a retry to be attempted.
        while let Some((task, _)) = self.tasks.peek() {
            // Execute the task.
            let outcome = match task.execute(&mut self.state).await {
                Ok(()) => Ok(()),
                Err(err) if err.severity() == EngineTaskErrorSeverity::Flush => Err(err),
                Err(err) => return Err(err),
            };

            // Update the state and notify the engine actor.
            self.state_sender.send_replace(self.state);

            // Pop the task from the queue now that it's been executed.
            self.tasks.pop();

            self.task_queue_length.send_replace(self.tasks.len());
            Metrics::engine_task_queue_depth().set(self.tasks.len() as f64);

            outcome?;
        }

        Ok(())
    }
}

/// An error occurred while attempting to reset the [`Engine`].