lokahi: serve the experimental opstack block-building namespace on chain routes

op-devstack/sysgo/l2_cl_lokahi.go

@@ -513,9 +513,12 @@ func lokahiConfigFile(t devtest.T, dir string, cfg lokahiSupernodeConfig, entrie
 		fmt.Fprintf(&b, "[interop]\nactivation-timestamp = %d\n\n", *cfg.interopActivationTimestamp)
 	}
 	// Acceptance tests drive a node through its admin API, which kona only registers when
-	// admin is enabled; op-node's devstack node enables it too.
+	// admin is enabled; op-node's devstack node enables it too. The experimental opstack
+	// namespace mirrors op-supernode's virtual nodes, which always run with
+	// ExperimentalOPStackAPI (multichain_supernode_runtime.go): the test sequencer drives
+	// block building through it on each chain's route.
 	fmt.Fprintf(&b, "[defaults]\ndatadir = %q\nmode = \"validator\"\n"+
-		"rpc-enable-admin = true\np2p-listen-ip = \"127.0.0.1\"\n\n",
+		"rpc-enable-admin = true\nexperimental-opstack-api = true\np2p-listen-ip = \"127.0.0.1\"\n\n",
 		lokahiDataDir(dir))
 	b.WriteString(strings.Join(entries, "\n"))
 	return b.String()

op-devstack/sysgo/l2_cl_lokahi_test.go

@@ -71,6 +71,21 @@ func TestLokahiConfigCarriesTheRequestedInteropActivation(t *testing.T) {
 		"the requested activation must reach lokahi verbatim")
 }
 
+// Every hosted chain serves the experimental opstack block-building namespace, because
+// op-supernode's virtual op-nodes do (makeNodeCfg sets ExperimentalOPStackAPI on all of them):
+// the test sequencer drives block building through opstack_* on each chain's route.
+func TestLokahiConfigEnablesTheOpstackNamespace(t *testing.T) {
+	cfg := lokahiSupernodeConfig{
+		l1Net:        &L1Network{genesis: &core.Genesis{Config: params.MainnetChainConfig}},
+		l1ELRPC:      "http://127.0.0.1:8545",
+		l1BeaconAddr: "http://127.0.0.1:5052",
+	}
+	rendered := lokahiConfigFile(newGateT(), t.TempDir(), cfg, nil)
+
+	require.Contains(t, rendered, "experimental-opstack-api = true",
+		"the devstack must turn the opstack namespace on, as it does on op-supernode")
+}
+
 // A preset that requests no activation must not write the table at all, so a node that was not
 // told one keeps reading its activation from the rollup configs -- the default path, unchanged.
 func TestLokahiConfigOmitsInteropWhenNoActivationIsRequested(t *testing.T) {

rust/kona/bin/node/src/flags/rpc.rs

@@ -42,6 +42,15 @@ pub struct RpcArgs {
     /// Enables development RPC endpoints for engine state introspection
     #[arg(long = "rpc.dev-enabled", default_value = "false", env = "KONA_NODE_RPC_DEV_ENABLED")]
     pub dev_enabled: bool,
+    /// Enables the experimental `opstack` block-building namespace, op-node's
+    /// `--experimental.sequencer-api`: the RPC surface the op-test-sequencer drives block
+    /// building through.
+    #[arg(
+        long = "rpc.experimental-opstack-api",
+        default_value = "false",
+        env = "KONA_NODE_RPC_EXPERIMENTAL_OPSTACK_API"
+    )]
+    pub experimental_opstack: bool,
 }
 
 impl Default for RpcArgs {
@@ -64,6 +73,7 @@ impl From<RpcArgs> for Option<RpcBuilder> {
             admin_persistence: args.admin_persistence,
             ws_enabled: args.ws_enabled,
             dev_enabled: args.dev_enabled,
+            experimental_opstack: args.experimental_opstack,
         })
     }
 }
@@ -82,6 +92,7 @@ mod tests {
     #[case::disable_rpc_alias(&["--rpc.port", "8743"], |args: &mut RpcArgs| { args.listen_port = 8743; })]
     #[case::disable_rpc(&["--rpc.enable-admin"], |args: &mut RpcArgs| { args.enable_admin = true; })]
     #[case::disable_rpc(&["--rpc.admin-state", "/"], |args: &mut RpcArgs| { args.admin_persistence = Some(PathBuf::from("/")); })]
+    #[case::experimental_opstack(&["--rpc.experimental-opstack-api"], |args: &mut RpcArgs| { args.experimental_opstack = true; })]
     fn test_parse_rpc_args(#[case] args: &[&str], #[case] mutate: impl Fn(&mut RpcArgs)) {
         let args = [&["kona-node"], args].concat();
         let cli = RpcArgs::parse_from(args);

rust/kona/crates/node/engine/src/lib.rs

@@ -44,11 +44,12 @@ extern crate tracing;
 
 mod task_queue;
 pub use task_queue::{
-    BuildSealCoupling, BuildTask, BuildTaskError, ConsolidateInput, ConsolidateTask,
-    ConsolidateTaskError, Engine, EngineBuildError, EngineResetError, EngineTask, EngineTaskError,
-    EngineTaskErrorSeverity, EngineTaskErrors, EngineTaskExt, FinalizeBlockId, FinalizeTask,
-    FinalizeTaskError, InsertTask, InsertTaskError, PromoteCrossSafeTask,
-    PromoteCrossSafeTaskError, SealTask, SealTaskError, SynchronizeTask, SynchronizeTaskError,
+    BuildSealCoupling, BuildTask, BuildTaskError, CommitBlockError, CommitTask, CommitTaskError,
+    ConsolidateInput, ConsolidateTask, ConsolidateTaskError, Engine, EngineBuildError,
+    EngineResetError, EngineTask, EngineTaskError, EngineTaskErrorSeverity, EngineTaskErrors,
+    EngineTaskExt, FinalizeBlockId, FinalizeTask, FinalizeTaskError, InsertTask, InsertTaskError,
+    PromoteCrossSafeTask, PromoteCrossSafeTaskError, SealTask, SealTaskError, SynchronizeTask,
+    SynchronizeTaskError,
 };
 
 mod attributes;

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

@@ -56,6 +56,8 @@ impl Metrics {
 
     /// Insert task label.
     pub const INSERT_TASK_LABEL: &str = "insert";
+    /// Commit task label.
+    pub const COMMIT_TASK_LABEL: &str = "commit";
     /// Consolidate task label.
     pub const CONSOLIDATE_TASK_LABEL: &str = "consolidate";
     /// Forkchoice task label.
@@ -128,6 +130,7 @@ impl Metrics {
         // emit exactly.
         for task in [
             Self::INSERT_TASK_LABEL,
+            Self::COMMIT_TASK_LABEL,
             Self::CONSOLIDATE_TASK_LABEL,
             Self::BUILD_TASK_LABEL,
             Self::SEAL_TASK_LABEL,

rust/kona/crates/node/engine/src/task_queue/tasks/commit/error.rs

@@ -0,0 +1,38 @@
+//! Contains the error types for the [`CommitTask`](crate::CommitTask).
+
+use crate::{EngineTaskError, InsertTaskError, task_queue::tasks::task::EngineTaskErrorSeverity};
+
+/// Why a commit was refused, as answered to the caller that requested it.
+///
+/// This is the channel payload, not the task's own error: a refused commit is a normal answer to
+/// the requester, and the task that delivered it has done its job.
+#[derive(Debug, thiserror::Error)]
+pub enum CommitBlockError {
+    /// The insert failed: the execution layer rejected the payload, could not be reached, or the
+    /// canonicalizing forkchoice update failed.
+    #[error(transparent)]
+    Insert(#[from] InsertTaskError),
+    /// The payload does not descend from the local-safe head, so making it the unsafe head would
+    /// rewind the chain under a head derived from L1.
+    ///
+    /// op-node's `CommitBlock` has no counterpart to this check; kona refuses the write rather
+    /// than corrupting its head ordering, and tells the caller so instead of dropping the payload
+    /// silently the way the gossip path does.
+    #[error("the payload does not descend from the local-safe head")]
+    DoesNotDescendFromLocalSafe,
+}
+
+/// An error that occurs when running the [`CommitTask`](crate::CommitTask).
+///
+/// Uninhabited: the commit's outcome — success or [`CommitBlockError`] — travels to the requester
+/// over the task's channel, and a requester that went away before hearing it (an RPC caller that
+/// disconnected) is logged rather than escalated, because failing the task would either halt the
+/// node over a dead client or retry a send that can never succeed.
+#[derive(Debug, thiserror::Error)]
+pub enum CommitTaskError {}
+
+impl EngineTaskError for CommitTaskError {
+    fn severity(&self) -> EngineTaskErrorSeverity {
+        unreachable!("CommitTaskError is uninhabited: no value of it can exist to be asked")
+    }
+}

rust/kona/crates/node/engine/src/task_queue/tasks/commit/mod.rs

@@ -0,0 +1,7 @@
+//! Task to commit an externally built payload and answer the caller that asked.
+
+mod task;
+pub use task::CommitTask;
+
+mod error;
+pub use error::{CommitBlockError, CommitTaskError};

rust/kona/crates/node/engine/src/task_queue/tasks/commit/task.rs

@@ -0,0 +1,188 @@
+//! A task to commit an externally built payload, answering the caller that requested it.
+
+use super::{CommitBlockError, CommitTaskError};
+use crate::{EngineClient, EngineState, EngineTaskExt, ImportedBlockSink, InsertTask};
+use derive_more::Constructor;
+use kona_genesis::RollupConfig;
+use kona_protocol::L2BlockInfo;
+use op_alloy_rpc_types_engine::OpExecutionPayloadEnvelope;
+use std::sync::Arc;
+use tokio::sync::mpsc;
+
+/// Task to commit an externally built payload: `opstack_commitBlockV1`'s write.
+///
+/// This is an [`InsertTask`] with an answer. The gossip path enqueues its inserts fire-and-forget
+/// — a peer that sent a bad payload is not waiting to hear about it — but a caller of the opstack
+/// API is: op-node's `CommitBlock` (`op-node/rollup/engine/api.go`) returns the `newPayload`
+/// verdict synchronously. So the insert's outcome, success or failure, is delivered over
+/// `result_tx`, and — like [`SealTask`]'s channel — delivering it *is* the task succeeding: a
+/// refused commit must reach the caller once, not be retried by the queue behind their back.
+///
+/// [`SealTask`]: crate::SealTask
+#[derive(Debug, Clone, Constructor)]
+pub struct CommitTask<EngineClient_: EngineClient> {
+    /// The engine API client.
+    pub engine: Arc<EngineClient_>,
+    /// The [`RollupConfig`].
+    pub cfg: Arc<RollupConfig>,
+    /// The payload to commit.
+    pub payload: OpExecutionPayloadEnvelope,
+    /// Where the commit's outcome is delivered.
+    pub result_tx: mpsc::Sender<Result<L2BlockInfo, CommitBlockError>>,
+    /// Where the decoded block goes once the engine has canonicalized it, same as any other
+    /// insert: a committed block is an imported block, and consumers reading imported blocks
+    /// (e.g. the system-config lookup) must see the commit path's blocks too.
+    pub block_sink: Arc<dyn ImportedBlockSink>,
+}
+
+impl<EngineClient_: EngineClient> CommitTask<EngineClient_> {
+    /// Runs the insert and reports what happened.
+    async fn commit(&self, state: &mut EngineState) -> Result<L2BlockInfo, CommitBlockError> {
+        // The same admission rule the gossip path applies before its inserts
+        // (`EngineTask::execute_inner`): a payload at or below the local-safe head must not
+        // become the unsafe head. The gossip path drops such a payload silently; here the caller
+        // hears the refusal.
+        let insert = InsertTask::new(
+            Arc::clone(&self.engine),
+            self.cfg.clone(),
+            self.payload.clone(),
+            None,
+            Arc::clone(&self.block_sink),
+        );
+        if !insert.descends_from_local_safe(state) {
+            return Err(CommitBlockError::DoesNotDescendFromLocalSafe);
+        }
+
+        insert.execute(state).await.map_err(CommitBlockError::from)
+    }
+}
+
+#[async_trait::async_trait]
+impl<EngineClient_: EngineClient> EngineTaskExt for CommitTask<EngineClient_> {
+    type Output = ();
+
+    type Error = CommitTaskError;
+
+    async fn execute(&self, state: &mut EngineState) -> Result<(), CommitTaskError> {
+        let result = self.commit(state).await;
+        // A requester that went away before hearing the result — an RPC caller that disconnected —
+        // is not a task failure: the commit itself already happened (or was refused), and there is
+        // no severity that fits a dead client. Failing Critical would halt the node over it, and
+        // Temporary would retry a send that can never succeed.
+        if self.result_tx.send(result).await.is_err() {
+            warn!(
+                target: "engine",
+                "The commit requester went away before hearing the result"
+            );
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        EngineSyncStateUpdate, LocalSafeHead, NoopBlockSink, task_queue::tasks::task::EngineTask,
+        test_utils::MockEngineClient,
+    };
+    use alloy_consensus::Block;
+    use alloy_primitives::B256;
+    use alloy_rpc_types_engine::{ExecutionPayloadV1, PayloadStatus, PayloadStatusEnum};
+    use kona_protocol::BlockInfo;
+    use op_alloy_consensus::OpTxEnvelope;
+    use std::time::Duration;
+
+    fn payload_at(number: u64, parent_hash: B256) -> OpExecutionPayloadEnvelope {
+        let mut payload = ExecutionPayloadV1::from_block_slow(&Block::<OpTxEnvelope>::default());
+        payload.block_number = number;
+        payload.parent_hash = parent_hash;
+        OpExecutionPayloadEnvelope::V1(payload)
+    }
+
+    /// A state whose unsafe and local-safe heads both sit at `head`.
+    fn state_at(head: L2BlockInfo) -> EngineState {
+        let mut state = EngineState::default();
+        state.sync_state = state.apply_sync_update(EngineSyncStateUpdate {
+            unsafe_head: Some(head),
+            local_safe_head: Some(LocalSafeHead::unpaired(head)),
+            ..Default::default()
+        });
+        state
+    }
+
+    /// A rejected commit answers the caller instead of being dropped like a gossiped payload, and
+    /// the task completes: the queue must not retry a write whose requester was already refused.
+    #[tokio::test]
+    async fn a_refused_commit_answers_the_caller_and_completes() {
+        let head = L2BlockInfo {
+            block_info: BlockInfo {
+                number: 7,
+                hash: B256::repeat_byte(7),
+                parent_hash: B256::repeat_byte(6),
+                timestamp: 14,
+            },
+            ..Default::default()
+        };
+        let mut state = state_at(head);
+
+        let config = Arc::new(RollupConfig::default());
+        let client = Arc::new(MockEngineClient::builder().with_config(config.clone()).build());
+        let (result_tx, mut result_rx) = mpsc::channel(1);
+
+        // Number 3 is behind local-safe head 7: the two decidable rejection cases both refuse it.
+        let task = EngineTask::Commit(Box::new(CommitTask::new(
+            client,
+            config,
+            payload_at(3, B256::repeat_byte(2)),
+            result_tx,
+            Arc::new(NoopBlockSink),
+        )));
+
+        tokio::time::timeout(Duration::from_secs(1), task.execute(&mut state))
+            .await
+            .expect("a refused commit must not retry")
+            .expect("delivering the refusal is the task succeeding");
+
+        let answer = result_rx.recv().await.expect("the caller hears the refusal");
+        assert!(matches!(answer, Err(CommitBlockError::DoesNotDescendFromLocalSafe)));
+        assert_eq!(state.sync_state.unsafe_head(), head, "a refused commit moves no head");
+    }
+
+    /// An insert the execution layer rejects reaches the caller as the insert's error, once,
+    /// rather than riding the queue's temporary-error retry loop forever.
+    #[tokio::test]
+    async fn a_rejected_payload_reaches_the_caller_once() {
+        let config = Arc::new(RollupConfig::default());
+        let client = Arc::new(
+            MockEngineClient::builder()
+                .with_config(config.clone())
+                .with_new_payload_v1_response(PayloadStatus::from_status(
+                    PayloadStatusEnum::Invalid { validation_error: "bad".into() },
+                ))
+                .build(),
+        );
+        let (result_tx, mut result_rx) = mpsc::channel(1);
+
+        let task = EngineTask::Commit(Box::new(CommitTask::new(
+            client,
+            config,
+            payload_at(1, B256::ZERO),
+            result_tx,
+            Arc::new(NoopBlockSink),
+        )));
+
+        tokio::time::timeout(Duration::from_secs(1), task.execute(&mut EngineState::default()))
+            .await
+            .expect("a rejected commit must not retry")
+            .expect("delivering the rejection is the task succeeding");
+
+        let answer = result_rx.recv().await.expect("the caller hears the rejection");
+        assert!(matches!(
+            answer,
+            Err(CommitBlockError::Insert(crate::InsertTaskError::UnexpectedPayloadStatus(
+                PayloadStatusEnum::Invalid { .. }
+            )))
+        ));
+    }
+}

rust/kona/crates/node/engine/src/task_queue/tasks/mod.rs

@@ -11,6 +11,9 @@ pub use synchronize::{SynchronizeTask, SynchronizeTaskError};
 mod insert;
 pub use insert::{InsertTask, InsertTaskError};
 
+mod commit;
+pub use commit::{CommitBlockError, CommitTask, CommitTaskError};
+
 mod build;
 pub use build::{BuildTask, BuildTaskError, EngineBuildError};