Merge branch 'main' into feat/docker-compose-tests-issue-transactions

Author: jonastheis

Cargo.lock

@@ -220,6 +220,7 @@ parking_lot = "0.12"
 rand = { version = "0.9" }
 reqwest = "0.12"
 serde = { version = "1.0" }
+serde_json = { version = "1.0" }
 sea-orm = { version = "1.1.0" }
 thiserror = "2.0"
 tokio = { version = "1.39", default-features = false }

Cargo.toml

@@ -71,5 +71,5 @@ futures.workspace = true
 parking_lot.workspace = true
 rand.workspace = true
 reqwest.workspace = true
-serde_json = { version = "1.0" }
+serde_json.workspace = true
 tokio.workspace = true

crates/chain-orchestrator/Cargo.toml

@@ -1,7 +1,7 @@
 use alloy_json_rpc::RpcError;
 use alloy_primitives::B256;
 use alloy_transport::TransportErrorKind;
-use scroll_db::{DatabaseError, L1MessageStart};
+use scroll_db::{DatabaseError, L1MessageKey};
 
 /// A type that represents an error that occurred in the chain orchestrator.
 #[derive(Debug, thiserror::Error)]
@@ -28,7 +28,7 @@ pub enum ChainOrchestratorError {
     },
     /// An L1 message was not found in the database.
     #[error("L1 message not found at {0}")]
-    L1MessageNotFound(L1MessageStart),
+    L1MessageNotFound(L1MessageKey),
     /// A gap was detected in the L1 message queue: the previous message before index {0} is
     /// missing.
     #[error("L1 message queue gap detected at index {0}, previous L1 message not found")]

crates/chain-orchestrator/src/error.rs

@@ -19,7 +19,7 @@ use scroll_alloy_hardforks::ScrollHardforks;
 use scroll_alloy_network::Scroll;
 use scroll_db::{
     Database, DatabaseError, DatabaseReadOperations, DatabaseTransactionProvider,
-    DatabaseWriteOperations, L1MessageStart, UnwindResult,
+    DatabaseWriteOperations, L1MessageKey, UnwindResult,
 };
 use scroll_network::NewBlockWithPeer;
 use std::{
@@ -215,9 +215,10 @@ impl<
         let block_info: L2BlockInfoWithL1Messages = (&block_with_peer.block).into();
         Self::do_handle_block_from_peer(ctx, block_with_peer).await?;
         Retry::default()
-            .retry("update_l1_messages_with_l2_block", || async {
+            .retry("handle_sequenced_block", || async {
                 let tx = database.tx_mut().await?;
                 tx.update_l1_messages_with_l2_block(block_info.clone()).await?;
+                tx.set_l2_head_block_info(block_info.block_info).await?;
                 tx.commit().await?;
                 Ok::<_, ChainOrchestratorError>(())
             })
@@ -483,9 +484,7 @@ impl<
                 Retry::default()
                     .retry("insert_block", || async {
                         let tx = database.tx_mut().await?;
-                        for block in block_infos.clone() {
-                            tx.insert_block(block, batch_info).await?;
-                        }
+                        tx.insert_blocks(block_infos.clone(), batch_info).await?;
                         tx.commit().await?;
                         Ok::<_, ChainOrchestratorError>(())
                     })
@@ -537,6 +536,7 @@ impl<
                     .retry("update_l1_messages_from_l2_blocks", || async {
                         let tx = database.tx_mut().await?;
                         tx.update_l1_messages_from_l2_blocks(block_info.clone()).await?;
+                        tx.set_l2_head_block_info(head.block_info).await?;
                         tx.commit().await?;
                         Ok::<_, ChainOrchestratorError>(())
                     })
@@ -838,7 +838,7 @@ async fn compute_l1_message_queue_hash(
             })
             .await?
             .map(|m| m.queue_hash)
-            .ok_or(DatabaseError::L1MessageNotFound(L1MessageStart::Index(index)))?
+            .ok_or(DatabaseError::L1MessageNotFound(L1MessageKey::QueueIndex(index)))?
             .unwrap_or_default()
             .to_vec();
         input.append(&mut l1_message.tx_hash().to_vec());
@@ -1078,7 +1078,9 @@ async fn validate_l1_messages(
     let l1_message_stream = Retry::default()
         .retry("get_l1_messages", || async {
             let messages = tx
-                .get_l1_messages(l1_message_hashes.first().map(|tx| L1MessageStart::Hash(*tx)))
+                .get_l1_messages(
+                    l1_message_hashes.first().map(|tx| L1MessageKey::TransactionHash(*tx)),
+                )
                 .await?;
             Ok::<_, ChainOrchestratorError>(messages)
         })
@@ -1093,7 +1095,9 @@ async fn validate_l1_messages(
             .await
             .map(|m| m.map(|msg| msg.transaction.tx_hash()))
             .transpose()?
-            .ok_or(ChainOrchestratorError::L1MessageNotFound(L1MessageStart::Hash(message_hash)))?;
+            .ok_or(ChainOrchestratorError::L1MessageNotFound(L1MessageKey::TransactionHash(
+                message_hash,
+            )))?;
 
         // If the received and expected L1 messages do not match return an error.
         if message_hash != expected_hash {

crates/chain-orchestrator/src/lib.rs

@@ -29,6 +29,7 @@ futures.workspace = true
 metrics.workspace = true
 metrics-derive.workspace = true
 sea-orm = { workspace = true, features = ["sqlx-sqlite", "runtime-tokio-native-tls", "macros"] }
+serde_json.workspace = true
 tempfile = { version = "3.20.0", optional = true }
 thiserror.workspace = true
 tokio = { workspace = true, features = ["macros", "sync"] }

crates/database/db/Cargo.toml

@@ -98,6 +98,12 @@ impl Database {
         db.tmp_dir = Some(dir);
         Ok(db)
     }
+
+    /// Returns a reference to the database tmp dir.
+    #[cfg(feature = "test-utils")]
+    pub const fn tmp_dir(&self) -> Option<&tempfile::TempDir> {
+        self.tmp_dir.as_ref()
+    }
 }
 
 #[async_trait::async_trait]
@@ -961,4 +967,27 @@ mod test {
         assert!(retried_block_3.is_none());
         assert!(retried_block_4.is_none());
     }
+
+    #[tokio::test]
+    async fn test_l2_block_head_roundtrip() {
+        // Set up the test database.
+        let db = setup_test_db().await;
+        let tx = db.tx_mut().await.unwrap();
+
+        // Generate unstructured bytes.
+        let mut bytes = [0u8; 40];
+        rand::rng().fill(bytes.as_mut_slice());
+        let mut u = Unstructured::new(&bytes);
+
+        // Generate and insert a block info as the head.
+        let block_info = BlockInfo::arbitrary(&mut u).unwrap();
+        tx.set_l2_head_block_info(block_info).await.unwrap();
+        tx.commit().await.unwrap();
+
+        // Retrieve and verify the head block info.
+        let tx = db.tx().await.unwrap();
+        let head_block_info = tx.get_l2_head_block_info().await.unwrap().unwrap();
+
+        assert_eq!(head_block_info, block_info);
+    }
 }

crates/database/db/src/db.rs

@@ -1,4 +1,4 @@
-use super::L1MessageStart;
+use super::L1MessageKey;
 use sea_orm::sqlx::Error as SqlxError;
 
 /// The error type for database operations.
@@ -13,7 +13,10 @@ pub enum DatabaseError {
     /// A generic error occurred.
     #[error("parse signature error: {0}")]
     ParseSignatureError(String),
+    /// Failed to serde the metadata value.
+    #[error("failed to serde metadata value: {0}")]
+    MetadataSerdeError(#[from] serde_json::Error),
     /// The L1 message was not found in database.
-    #[error("L1 message at index [{0}] not found in database")]
-    L1MessageNotFound(L1MessageStart),
+    #[error("L1 message at key [{0}] not found in database")]
+    L1MessageNotFound(L1MessageKey),
 }

crates/database/db/src/error.rs

@@ -15,9 +15,7 @@ mod models;
 pub use models::*;
 
 mod operations;
-pub use operations::{
-    DatabaseReadOperations, DatabaseWriteOperations, L1MessageStart, UnwindResult,
-};
+pub use operations::{DatabaseReadOperations, DatabaseWriteOperations, L1MessageKey, UnwindResult};
 
 mod transaction;
 pub use transaction::{DatabaseTransactionProvider, TXMut, TX};

crates/database/db/src/lib.rs

@@ -21,16 +21,6 @@ impl ActiveModelBehavior for ActiveModel {}
 
 impl From<Metadata> for ActiveModel {
     fn from(metadata: Metadata) -> Self {
-        Self {
-            key: ActiveValue::Set("l1_finalized_block".to_owned()),
-            value: ActiveValue::Set(metadata.l1_finalized_block.to_string()),
-        }
-    }
-}
-
-impl From<Model> for Metadata {
-    fn from(value: Model) -> Self {
-        debug_assert!(value.key == "l1_finalized_block");
-        Self { l1_finalized_block: value.value.parse().expect("invalid value") }
+        Self { key: ActiveValue::Set(metadata.key), value: ActiveValue::Set(metadata.value) }
     }
 }