feat: database optimisation (#440)

* cleanup and optimise database

* add composite index

* update docker compose grafana

* add derivation pipeline cache

* add async pre-fetch logic

* cleanup code

Author: frisitano

Cargo.lock

@@ -223,6 +223,7 @@ futures = { version = "0.3", default-features = false }
 lru = "0.13.0"
 metrics = "0.24.0"
 metrics-derive = "0.1"
+moka = "0.12.11"
 parking_lot = "0.12"
 rand = { version = "0.9" }
 rayon = "1.7"

Cargo.toml

@@ -643,6 +643,14 @@ impl DatabaseReadOperations for Database {
         )
     }
 
+    async fn get_max_block_data_hint_block_number(&self) -> Result<u64, DatabaseError> {
+        metered!(
+            DatabaseOperation::GetMaxBlockDataHintBlockNumber,
+            self,
+            tx(|tx| async move { tx.get_max_block_data_hint_block_number().await })
+        )
+    }
+
     async fn get_l2_block_and_batch_info_by_hash(
         &self,
         block_hash: B256,
@@ -862,7 +870,7 @@ mod test {
         BatchCommitData, BatchInfo, BlockInfo, L1MessageEnvelope, L2BlockInfoWithL1Messages,
     };
     use scroll_alloy_consensus::TxL1Message;
-    use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
+    use sea_orm::EntityTrait;
 
     #[tokio::test]
     async fn test_database_round_trip_batch_commit() {
@@ -1227,11 +1235,13 @@ mod test {
 
     #[tokio::test]
     async fn test_delete_l1_messages_gt() {
+        reth_tracing::init_test_tracing();
+
         // Set up the test database.
         let db = setup_test_db().await;
 
         // Generate unstructured bytes.
-        let mut bytes = [0u8; 1024];
+        let mut bytes = [0u8; 10240];
         rand::rng().fill(bytes.as_mut_slice());
         let mut u = Unstructured::new(&bytes);
 
@@ -1276,6 +1286,8 @@ mod test {
 
     #[tokio::test]
     async fn test_get_l2_block_info_by_number() {
+        reth_tracing::init_test_tracing();
+
         // Set up the test database.
         let db = setup_test_db().await;
 
@@ -1475,67 +1487,6 @@ mod test {
         }
     }
 
-    #[tokio::test]
-    async fn test_insert_block_upsert_behavior() {
-        reth_tracing::init_test_tracing();
-
-        // Set up the test database.
-        let db = setup_test_db().await;
-
-        // Generate unstructured bytes.
-        let mut bytes = [0u8; 1024];
-        rand::rng().fill(bytes.as_mut_slice());
-        let mut u = Unstructured::new(&bytes);
-
-        // Generate batches
-        let batch_data_1 = BatchCommitData { index: 100, ..Arbitrary::arbitrary(&mut u).unwrap() };
-        let batch_info_1: BatchInfo = batch_data_1.clone().into();
-        let batch_data_2 = BatchCommitData { index: 200, ..Arbitrary::arbitrary(&mut u).unwrap() };
-        let batch_info_2: BatchInfo = batch_data_2.clone().into();
-
-        db.insert_batch(batch_data_1).await.unwrap();
-        db.insert_batch(batch_data_2).await.unwrap();
-
-        // Insert initial block
-        let block_info = BlockInfo { number: 600, hash: B256::arbitrary(&mut u).unwrap() };
-        db.insert_blocks(vec![block_info], batch_info_1).await.unwrap();
-
-        // Verify initial insertion
-        let retrieved_block = db.get_l2_block_info_by_number(600).await.unwrap();
-        assert_eq!(retrieved_block, Some(block_info));
-
-        // Verify initial batch association using model conversion
-        let initial_l2_block_model = models::l2_block::Entity::find()
-            .filter(models::l2_block::Column::BlockNumber.eq(600))
-            .one(db.inner().get_connection())
-            .await
-            .unwrap()
-            .unwrap();
-        let (initial_block_info, initial_batch_info): (BlockInfo, BatchInfo) =
-            initial_l2_block_model.into();
-        assert_eq!(initial_block_info, block_info);
-        assert_eq!(initial_batch_info, batch_info_1);
-
-        // Update the same block with different batch info (upsert)
-        db.insert_blocks(vec![block_info], batch_info_2).await.unwrap();
-
-        // Verify the block still exists and was updated
-        let retrieved_block = db.get_l2_block_info_by_number(600).await.unwrap().unwrap();
-        assert_eq!(retrieved_block, block_info);
-
-        // Verify batch association was updated using model conversion
-        let updated_l2_block_model = models::l2_block::Entity::find()
-            .filter(models::l2_block::Column::BlockNumber.eq(600))
-            .one(db.inner().get_connection())
-            .await
-            .unwrap()
-            .unwrap();
-        let (updated_block_info, updated_batch_info): (BlockInfo, BatchInfo) =
-            updated_l2_block_model.into();
-        assert_eq!(updated_block_info, block_info);
-        assert_eq!(updated_batch_info, batch_info_2);
-    }
-
     #[tokio::test]
     async fn test_prepare_on_startup() {
         let db = setup_test_db().await;

crates/database/db/src/db.rs

@@ -9,6 +9,9 @@ pub use db::Database;
 mod error;
 pub use error::DatabaseError;
 
+mod maintenance;
+pub use maintenance::DatabaseMaintenance;
+
 mod metrics;
 
 mod models;

crates/database/db/src/lib.rs

@@ -0,0 +1,68 @@
+use sea_orm::ConnectionTrait;
+
+use crate::DatabaseConnectionProvider;
+
+use super::Database;
+use std::sync::Arc;
+
+/// The interval in seconds between optimization runs in seconds.
+const PERIODIC_MAINTENANCE_INTERVAL_SECS: u64 = 600;
+
+/// Provides maintenance operations for the database.
+#[derive(Debug)]
+pub struct DatabaseMaintenance {
+    db: Arc<Database>,
+}
+
+impl DatabaseMaintenance {
+    /// Creates a new `DatabaseMaintenance` instance.
+    pub const fn new(db: Arc<Database>) -> Self {
+        Self { db }
+    }
+
+    /// Runs the maintenance tasks in a loop.
+    pub async fn run(self) {
+        self.startup_maintenance().await;
+
+        loop {
+            tokio::time::sleep(std::time::Duration::from_secs(PERIODIC_MAINTENANCE_INTERVAL_SECS))
+                .await;
+            self.periodic_maintenance().await;
+        }
+    }
+
+    /// Runs maintenance tasks at startup.
+    ///
+    /// This includes running `ANALYZE` and `PRAGMA optimize`.
+    async fn startup_maintenance(&self) {
+        let db = self.db.inner();
+        let conn = db.get_connection();
+
+        tracing::info!(target: "scroll::db::maintenance", "running startup ANALYZE...");
+        if let Err(err) = conn.execute_unprepared("ANALYZE;").await {
+            tracing::warn!(target: "scroll::db::maintenance", "ANALYZE failed: {:?}", err);
+        }
+
+        tracing::info!(target: "scroll::db::maintenance", "running PRAGMA optimize at startup...");
+        if let Err(err) = conn.execute_unprepared("PRAGMA optimize;").await {
+            tracing::warn!(target: "scroll::db::maintenance", "PRAGMA optimize failed: {:?}", err);
+        }
+
+        tracing::info!(target: "scroll::db::maintenance", "startup maintenance complete.");
+    }
+
+    /// Runs periodic maintenance tasks.
+    ///
+    /// This includes running `PRAGMA optimize`.
+    async fn periodic_maintenance(&self) {
+        let db = self.db.inner();
+        let conn = db.get_connection();
+
+        tracing::info!(target: "scroll::db::maintenance", "running periodic PRAGMA optimize...");
+        if let Err(err) = conn.execute_unprepared("PRAGMA optimize;").await {
+            tracing::warn!(target: "scroll::db::maintenance", "PRAGMA optimize failed: {:?}", err);
+        } else {
+            tracing::info!(target: "scroll::db::maintenance", "periodic PRAGMA optimize complete.");
+        }
+    }
+}

crates/database/db/src/maintenance.rs

@@ -64,6 +64,7 @@ pub(crate) enum DatabaseOperation {
     GetL2HeadBlockNumber,
     GetNL1Messages,
     GetNL2BlockDataHint,
+    GetMaxBlockDataHintBlockNumber,
     GetL2BlockAndBatchInfoByHash,
     GetL2BlockInfoByNumber,
     GetLatestSafeL2Info,
@@ -132,6 +133,7 @@ impl DatabaseOperation {
             Self::GetL2HeadBlockNumber => "get_l2_head_block_number",
             Self::GetNL1Messages => "get_n_l1_messages",
             Self::GetNL2BlockDataHint => "get_n_l2_block_data_hint",
+            Self::GetMaxBlockDataHintBlockNumber => "get_max_block_data_hint_block_number",
             Self::GetL2BlockAndBatchInfoByHash => "get_l2_block_and_batch_info_by_hash",
             Self::GetL2BlockInfoByNumber => "get_l2_block_info_by_number",
             Self::GetLatestSafeL2Info => "get_latest_safe_l2_info",

crates/database/db/src/metrics.rs

@@ -708,13 +708,15 @@ impl<T: WriteConnectionProvider + ?Sized + Sync> DatabaseWriteOperations for T {
             blocks.into_iter().map(|b| (b, batch_info).into()).collect();
         models::l2_block::Entity::insert_many(l2_blocks)
             .on_conflict(
-                OnConflict::column(models::l2_block::Column::BlockHash)
-                    .update_columns([
-                        models::l2_block::Column::BlockNumber,
-                        models::l2_block::Column::BatchHash,
-                        models::l2_block::Column::BatchIndex,
-                    ])
-                    .to_owned(),
+                OnConflict::columns([
+                    models::l2_block::Column::BlockHash,
+                    models::l2_block::Column::BatchHash,
+                ])
+                .update_columns([
+                    models::l2_block::Column::BlockNumber,
+                    models::l2_block::Column::BatchIndex,
+                ])
+                .to_owned(),
             )
             .on_empty_do_nothing()
             .exec_without_returning(self.get_connection())
@@ -994,6 +996,9 @@ pub trait DatabaseReadOperations {
         n: usize,
     ) -> Result<Vec<L1MessageEnvelope>, DatabaseError>;
 
+    /// Get the maximum block number for which we have stored extra data hints.
+    async fn get_max_block_data_hint_block_number(&self) -> Result<u64, DatabaseError>;
+
     /// Get the extra data for n block, starting at the provided block number.
     async fn get_n_l2_block_data_hint(
         &self,
@@ -1364,6 +1369,7 @@ impl<T: ReadConnectionProvider + Sync + ?Sized> DatabaseReadOperations for T {
     ) -> Result<Vec<BlockDataHint>, DatabaseError> {
         Ok(models::block_data::Entity::find()
             .filter(models::block_data::Column::Number.gte(block_number as i64))
+            .order_by_asc(models::block_data::Column::Number)
             .limit(Some(n as u64))
             .all(self.get_connection())
             .await?
@@ -1372,6 +1378,18 @@ impl<T: ReadConnectionProvider + Sync + ?Sized> DatabaseReadOperations for T {
             .collect())
     }
 
+    async fn get_max_block_data_hint_block_number(&self) -> Result<u64, DatabaseError> {
+        Ok(models::block_data::Entity::find()
+            .select_only()
+            .column_as(models::block_data::Column::Number.max(), "max_number")
+            .into_tuple::<Option<i64>>()
+            .one(self.get_connection())
+            .await?
+            .flatten()
+            .map(|n| n as u64)
+            .unwrap_or(0))
+    }
+
     async fn get_l2_block_and_batch_info_by_hash(
         &self,
         block_hash: B256,

crates/database/db/src/operations.rs

@@ -90,7 +90,7 @@ impl<S: DatabaseService> DatabaseService for Retry<S> {
                     };
 
                     attempt += 1;
-                    tracing::debug!(target: "scroll::chain_orchestrator", ?error, attempt, delay_ms, "Retrying database query");
+                    tracing::debug!(target: "scroll::db", ?error, attempt, delay_ms, "Retrying database query");
 
                     tokio::time::sleep(Duration::from_millis(delay_ms)).await;
                 }

crates/database/db/src/service/retry.rs

@@ -6,17 +6,7 @@ mod m20250408_132123_add_header_metadata;
 mod m20250408_150338_load_header_metadata;
 mod m20250411_072004_add_l2_block;
 mod m20250616_223947_add_metadata;
-mod m20250825_093350_remove_unsafe_l2_blocks;
-mod m20250829_042803_add_table_indexes;
 mod m20250904_175949_block_signature;
-mod m20250923_135359_add_index_block_hash;
-mod m20250929_161536_add_additional_indexes;
-mod m20251001_125444_add_index_status;
-mod m20251005_160938_add_initial_l1_block_numbers;
-mod m20251013_140946_add_initial_l1_processed_block_number;
-mod m20251021_070729_add_skipped_column;
-mod m20251021_144852_add_queue_index_index;
-mod m20251027_090416_add_table_statistics;
 mod m20251028_110719_add_l1_block_table;
 
 mod migration_info;
@@ -36,17 +26,7 @@ impl<MI: MigrationInfo + Send + Sync + 'static> MigratorTrait for Migrator<MI> {
             Box::new(m20250408_150338_load_header_metadata::Migration::<MI>(Default::default())),
             Box::new(m20250411_072004_add_l2_block::Migration::<MI>(Default::default())),
             Box::new(m20250616_223947_add_metadata::Migration),
-            Box::new(m20250825_093350_remove_unsafe_l2_blocks::Migration),
-            Box::new(m20250829_042803_add_table_indexes::Migration),
             Box::new(m20250904_175949_block_signature::Migration),
-            Box::new(m20250923_135359_add_index_block_hash::Migration),
-            Box::new(m20250929_161536_add_additional_indexes::Migration),
-            Box::new(m20251001_125444_add_index_status::Migration),
-            Box::new(m20251005_160938_add_initial_l1_block_numbers::Migration),
-            Box::new(m20251013_140946_add_initial_l1_processed_block_number::Migration),
-            Box::new(m20251021_070729_add_skipped_column::Migration),
-            Box::new(m20251021_144852_add_queue_index_index::Migration),
-            Box::new(m20251027_090416_add_table_statistics::Migration),
             Box::new(m20251028_110719_add_l1_block_table::Migration),
         ]
     }