add async pre-fetch logic

Author: frisitano

crates/database/db/src/operations.rs

@@ -1369,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?

crates/database/db/src/service/retry.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/derivation-pipeline/src/cache.rs

@@ -2,17 +2,18 @@ use moka::future::Cache;
 use scroll_alloy_rpc_types_engine::BlockDataHint;
 use scroll_db::{Database, DatabaseReadOperations};
 use std::{sync::Arc, time::Duration};
+use tokio::sync::Mutex;
 
 use crate::DerivationPipelineError;
 
 /// The default size of the block data hint pre-fetch cache (number of entries).
-pub(crate) const DEFAULT_CACHE_SIZE: usize = 80_000;
+pub(crate) const CACHE_SIZE: usize = 100_000;
 
 /// The default number of block data hints to pre-fetch.
-pub(crate) const DEFAULT_PREFETCH_COUNT: usize = 60_000;
+pub(crate) const PREFETCH_RANGE_SIZE: usize = 50_000;
 
 /// The default time-to-live (TTL) for cache entries.
-pub(crate) const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(120 * 60); // 120 minutes
+pub(crate) const CACHE_TTL: Duration = Duration::from_secs(120 * 60); // 120 minutes
 
 #[derive(Debug, Clone)]
 pub struct PreFetchCache {
@@ -21,6 +22,7 @@ pub struct PreFetchCache {
     // TODO: Add a cache for batches.
     max_block_data_hint_block_number: u64,
     pre_fetch_count: usize,
+    ranges_in_cache: Arc<Mutex<Vec<u64>>>,
 }
 
 impl PreFetchCache {
@@ -37,6 +39,7 @@ impl PreFetchCache {
             hint_cache: Cache::builder().max_capacity(size as u64).time_to_live(ttl).build(),
             max_block_data_hint_block_number,
             pre_fetch_count,
+            ranges_in_cache: Arc::new(Mutex::new(Vec::new())),
         })
     }
 
@@ -45,20 +48,52 @@ impl PreFetchCache {
         &self,
         block_number: u64,
     ) -> Result<Option<BlockDataHint>, DerivationPipelineError> {
+        tracing::trace!(
+            target: "scroll::derivation_pipeline::cache",
+            block_number = block_number,
+            "Fetching block data hint from cache",
+        );
+        // If the block number is beyond the maximum known block data hint, return None.
         if block_number > self.max_block_data_hint_block_number {
             return Ok(None);
         }
 
+        // If the hint is already cached, return it.
         if let Some(cached_hint) = self.hint_cache.get(&block_number).await {
             return Ok(Some(cached_hint));
         }
 
-        let hints = self.db.get_n_l2_block_data_hint(block_number, self.pre_fetch_count).await?;
-        let requested = hints.first().cloned();
-        for (idx, hint) in hints.into_iter().enumerate() {
-            self.hint_cache.insert(block_number + idx as u64, hint.clone()).await;
-        }
+        // The data hint is not in cache so we will pre-fetch a range of hints.
+        let range_start = self.pre_fetch_range_start(block_number);
+        let mut ranges_in_cache = self.ranges_in_cache.lock().await;
+
+        // The range has already been fetched by another task (this is a rare case caused by a race)
+        if !ranges_in_cache.contains(&range_start) {
+            tracing::info!(
+                target: "scroll::derivation_pipeline::cache",
+                range_start = range_start,
+                pre_fetch_count = self.pre_fetch_count,
+                "Pre-fetching block data hints for range",
+            );
+            let hints = self.db.get_n_l2_block_data_hint(range_start, self.pre_fetch_count).await?;
+            for (idx, hint) in hints.iter().enumerate() {
+                self.hint_cache.insert(range_start + idx as u64, hint.clone()).await;
+            }
+
+            if ranges_in_cache.len() == 2 {
+                ranges_in_cache.remove(0);
+            }
+
+            ranges_in_cache.push(range_start);
+        };
+
+        drop(ranges_in_cache);
+
+        // Now the requested hint should be in cache.
+        Ok(self.hint_cache.get(&block_number).await)
+    }
 
-        Ok(requested)
+    const fn pre_fetch_range_start(&self, block: u64) -> u64 {
+        (block / self.pre_fetch_count as u64) * self.pre_fetch_count as u64
     }
 }

crates/derivation-pipeline/src/lib.rs

@@ -15,7 +15,7 @@ use scroll_db::{Database, DatabaseReadOperations, L1MessageKey};
 use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
 
 mod cache;
-use cache::{PreFetchCache, DEFAULT_CACHE_SIZE, DEFAULT_CACHE_TTL, DEFAULT_PREFETCH_COUNT};
+use cache::{PreFetchCache, CACHE_SIZE, CACHE_TTL, PREFETCH_RANGE_SIZE};
 
 mod data_source;
 
@@ -130,13 +130,9 @@ impl<P> DerivationPipelineWorker<P> {
         batch_receiver: UnboundedReceiver<Arc<BatchDerivationRequest>>,
         result_sender: UnboundedSender<BatchDerivationResult>,
     ) -> Result<Self, DerivationPipelineError> {
-        let cache = PreFetchCache::new(
-            database.clone(),
-            DEFAULT_CACHE_SIZE,
-            DEFAULT_CACHE_TTL,
-            DEFAULT_PREFETCH_COUNT,
-        )
-        .await?;
+        let cache =
+            PreFetchCache::new(database.clone(), CACHE_SIZE, CACHE_TTL, PREFETCH_RANGE_SIZE)
+                .await?;
         Ok(Self {
             batch_receiver,
             result_sender,