feat(node/sequencer) Break apart Engine operations to prevent blocking

crates/node/engine/Cargo.toml

@@ -48,7 +48,7 @@ thiserror.workspace = true
 url.workspace = true
 tower.workspace = true
 http-body-util.workspace = true
-derive_more = { workspace = true, features = ["display", "deref", "from_str"] }
+derive_more = { workspace = true, features = ["display", "deref", "from_str", "constructor"] }
 serde_json.workspace = true
 
 # metrics

crates/node/engine/src/lib.rs

@@ -42,8 +42,8 @@ mod task_queue;
 pub use task_queue::{
     BuildTask, BuildTaskError, ConsolidateTask, ConsolidateTaskError, Engine, EngineBuildError,
     EngineResetError, EngineTask, EngineTaskError, EngineTaskErrorSeverity, EngineTaskErrors,
-    EngineTaskExt, FinalizeTask, FinalizeTaskError, InsertTask, InsertTaskError, SynchronizeTask,
-    SynchronizeTaskError,
+    EngineTaskExt, FinalizeTask, FinalizeTaskError, InsertTask, InsertTaskError, SealError,
+    SealTask, SealTaskError, SynchronizeTask, SynchronizeTaskError,
 };
 
 mod attributes;

crates/node/engine/src/metrics/mod.rs

@@ -57,6 +57,8 @@ impl Metrics {
     pub const FORKCHOICE_TASK_LABEL: &str = "forkchoice-update";
     /// Build task label.
     pub const BUILD_TASK_LABEL: &str = "build";
+    /// Seal task label.
+    pub const SEAL_TASK_LABEL: &str = "seal";
     /// Finalize task label.
     pub const FINALIZE_TASK_LABEL: &str = "finalize";
 

crates/node/engine/src/task_queue/tasks/build/error.rs

@@ -1,13 +1,10 @@
 //! Contains error types for the [crate::SynchronizeTask].
 
 use crate::{
-    EngineTaskError, InsertTaskError, SynchronizeTaskError,
-    task_queue::tasks::task::EngineTaskErrorSeverity,
+    EngineTaskError, SynchronizeTaskError, task_queue::tasks::task::EngineTaskErrorSeverity,
 };
-use alloy_rpc_types_engine::PayloadStatusEnum;
+use alloy_rpc_types_engine::{PayloadId, PayloadStatusEnum};
 use alloy_transport::{RpcError, TransportErrorKind};
-use kona_protocol::FromBlockError;
-use op_alloy_rpc_types_engine::OpExecutionPayloadEnvelope;
 use thiserror::Error;
 use tokio::sync::mpsc;
 
@@ -47,7 +44,7 @@ pub enum EngineBuildError {
     EngineSyncing,
 }
 
-/// An error that occurs when running the [crate::SynchronizeTask].
+/// An error that occurs when running the [crate::BuildTask].
 #[derive(Debug, Error)]
 pub enum BuildTaskError {
     /// An error occurred when building the payload attributes in the engine.
@@ -56,41 +53,15 @@ pub enum BuildTaskError {
     /// The initial forkchoice update call to the engine api failed.
     #[error(transparent)]
     ForkchoiceUpdateFailed(#[from] SynchronizeTaskError),
-    /// Impossible to insert the payload into the engine.
-    #[error(transparent)]
-    PayloadInsertionFailed(#[from] Box<InsertTaskError>),
-    /// The get payload call to the engine api failed.
-    #[error(transparent)]
-    GetPayloadFailed(RpcError<TransportErrorKind>),
-    /// A deposit-only payload failed to import.
-    #[error("Deposit-only payload failed to import")]
-    DepositOnlyPayloadFailed,
-    /// Failed to re-attempt payload import with deposit-only payload.
-    #[error("Failed to re-attempt payload import with deposit-only payload")]
-    DepositOnlyPayloadReattemptFailed,
-    /// The payload is invalid, and the derivation pipeline must
-    /// be flushed post-holocene.
-    #[error("Invalid payload, must flush post-holocene")]
-    HoloceneInvalidFlush,
-    /// Failed to convert a [`OpExecutionPayload`] to a [`L2BlockInfo`].
-    ///
-    /// [`OpExecutionPayload`]: op_alloy_rpc_types_engine::OpExecutionPayload
-    /// [`L2BlockInfo`]: kona_protocol::L2BlockInfo
-    #[error(transparent)]
-    FromBlock(#[from] FromBlockError),
     /// Error sending the built payload envelope.
     #[error(transparent)]
-    MpscSend(#[from] Box<mpsc::error::SendError<OpExecutionPayloadEnvelope>>),
-    /// The clock went backwards.
-    #[error("The clock went backwards")]
-    ClockWentBackwards,
+    MpscSend(#[from] Box<mpsc::error::SendError<PayloadId>>),
 }
 
 impl EngineTaskError for BuildTaskError {
     fn severity(&self) -> EngineTaskErrorSeverity {
         match self {
             Self::ForkchoiceUpdateFailed(inner) => inner.severity(),
-            Self::PayloadInsertionFailed(inner) => inner.severity(),
             Self::EngineBuildError(EngineBuildError::FinalizedAheadOfUnsafe(_, _)) => {
                 EngineTaskErrorSeverity::Critical
             }
@@ -109,13 +80,7 @@ impl EngineTaskError for BuildTaskError {
             Self::EngineBuildError(EngineBuildError::EngineSyncing) => {
                 EngineTaskErrorSeverity::Temporary
             }
-            Self::GetPayloadFailed(_) => EngineTaskErrorSeverity::Temporary,
-            Self::HoloceneInvalidFlush => EngineTaskErrorSeverity::Flush,
-            Self::DepositOnlyPayloadReattemptFailed => EngineTaskErrorSeverity::Critical,
-            Self::DepositOnlyPayloadFailed => EngineTaskErrorSeverity::Critical,
-            Self::FromBlock(_) => EngineTaskErrorSeverity::Critical,
             Self::MpscSend(_) => EngineTaskErrorSeverity::Critical,
-            Self::ClockWentBackwards => EngineTaskErrorSeverity::Critical,
         }
     }
 }