feat: set initial fcs from db (#335)

* feat: set initial fcs from db

* feat: answer comments

* feat: save and restart with latest sequenced block for fcs

* test: test correct restart at last sequenced block

* test: fix graceful shutfown test

* fix: lint

* feat: store head for followers and sequencer

* fix: lint

* feat: answer comments

Author: greged93

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

@@ -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>(())
                     })

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

@@ -13,6 +13,9 @@ 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),

crates/database/db/src/error.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) }
     }
 }

crates/database/db/src/models/metadata.rs

@@ -71,7 +71,27 @@ pub trait DatabaseWriteOperations: WriteConnectionProvider + DatabaseReadOperati
     ) -> Result<(), DatabaseError> {
         tracing::trace!(target: "scroll::db", block_number, "Updating the latest finalized L1 block number in the database.");
         let metadata: models::metadata::ActiveModel =
-            Metadata { l1_finalized_block: block_number }.into();
+            Metadata { key: "l1_finalized_block".to_string(), value: block_number.to_string() }
+                .into();
+        Ok(models::metadata::Entity::insert(metadata)
+            .on_conflict(
+                OnConflict::column(models::metadata::Column::Key)
+                    .update_column(models::metadata::Column::Value)
+                    .to_owned(),
+            )
+            .exec(self.get_connection())
+            .await
+            .map(|_| ())?)
+    }
+
+    /// Set the L2 head block info.
+    async fn set_l2_head_block_info(&self, block_info: BlockInfo) -> Result<(), DatabaseError> {
+        tracing::trace!(target: "scroll::db", ?block_info, "Updating the L2 head block info in the database.");
+        let metadata: models::metadata::ActiveModel = Metadata {
+            key: "l2_head_block".to_string(),
+            value: serde_json::to_string(&block_info)?,
+        }
+        .into();
         Ok(models::metadata::Entity::insert(metadata)
             .on_conflict(
                 OnConflict::column(models::metadata::Column::Key)
@@ -444,6 +464,18 @@ pub trait DatabaseReadOperations: ReadConnectionProvider + Sync {
             .map(|x| x.and_then(|x| x.parse::<u64>().ok()))?)
     }
 
+    /// Get the latest L2 head block info.
+    async fn get_l2_head_block_info(&self) -> Result<Option<BlockInfo>, DatabaseError> {
+        Ok(models::metadata::Entity::find()
+            .filter(models::metadata::Column::Key.eq("l2_head_block"))
+            .select_only()
+            .column(models::metadata::Column::Value)
+            .into_tuple::<String>()
+            .one(self.get_connection())
+            .await
+            .map(|x| x.and_then(|x| serde_json::from_str(&x).ok()))?)
+    }
+
     /// Get an iterator over all [`BatchCommitData`]s in the database.
     async fn get_batches<'a>(
         &'a self,
@@ -571,7 +603,9 @@ pub trait DatabaseReadOperations: ReadConnectionProvider + Sync {
             })?)
     }
 
-    /// Get the latest safe L2 ([`BlockInfo`], [`BatchInfo`]) from the database.
+    /// Get the latest safe/finalized L2 ([`BlockInfo`], [`BatchInfo`]) from the database. Until we
+    /// update the batch handling logic with issue #273, we don't differentiate between safe and
+    /// finalized l2 blocks.
     async fn get_latest_safe_l2_info(
         &self,
     ) -> Result<Option<(BlockInfo, BatchInfo)>, DatabaseError> {

crates/database/db/src/operations.rs

@@ -40,6 +40,7 @@ rollup-node-providers.workspace = true
 rollup-node-signer.workspace = true
 
 # scroll
+scroll-db.workspace = true
 scroll-network.workspace = true
 
 # misc
@@ -70,6 +71,7 @@ test-utils = [
     "rollup-node-providers/test-utils",
     "reth-chainspec/test-utils",
     "reth-primitives-traits/test-utils",
+    "scroll-db/test-utils",
 ]
 serde = [
     "alloy-eips/serde",