reffactoring and improvement. Add configs.

Author: kobayurii

configuration/src/configs/general.rs

@@ -21,6 +21,8 @@ pub struct GeneralRpcServerConfig {
     pub block_cache_size: f64,
     pub shadow_data_consistency_rate: f64,
     pub prefetch_state_size_limit: u64,
+    pub available_data_ranges: u64,
+    pub archival_mode: bool,
 }
 
 #[derive(Debug, Clone)]
@@ -120,6 +122,14 @@ pub struct CommonGeneralRpcServerConfig {
     pub shadow_data_consistency_rate: Option<f64>,
     #[serde(deserialize_with = "deserialize_optional_data_or_env", default)]
     pub prefetch_state_size_limit: Option<u64>,
+    #[validate(range(
+        min = 1,
+        message = "Available data ranges must be greater than or equal to 1"
+    ))]
+    #[serde(deserialize_with = "deserialize_optional_data_or_env", default)]
+    pub available_data_ranges: Option<u64>,
+    #[serde(deserialize_with = "deserialize_optional_data_or_env", default)]
+    pub archival_mode: Option<bool>,
 }
 
 impl CommonGeneralRpcServerConfig {
@@ -146,6 +156,14 @@ impl CommonGeneralRpcServerConfig {
     pub fn default_prefetch_state_size_limit() -> u64 {
         100_000
     }
+
+    pub fn default_available_data_ranges() -> u64 {
+        1
+    }
+
+    pub fn default_archival_mode() -> bool {
+        false
+    }
 }
 
 impl Default for CommonGeneralRpcServerConfig {
@@ -157,6 +175,8 @@ impl Default for CommonGeneralRpcServerConfig {
             block_cache_size: Some(Self::default_block_cache_size()),
             shadow_data_consistency_rate: Some(Self::default_shadow_data_consistency_rate()),
             prefetch_state_size_limit: Some(Self::default_prefetch_state_size_limit()),
+            available_data_ranges: Some(Self::default_available_data_ranges()),
+            archival_mode: Some(Self::default_archival_mode()),
         }
     }
 }
@@ -262,6 +282,14 @@ impl From<CommonGeneralConfig> for GeneralRpcServerConfig {
                 .rpc_server
                 .prefetch_state_size_limit
                 .unwrap_or_else(CommonGeneralRpcServerConfig::default_prefetch_state_size_limit),
+            available_data_ranges: common_config
+                .rpc_server
+                .available_data_ranges
+                .unwrap_or_else(CommonGeneralRpcServerConfig::default_available_data_ranges),
+            archival_mode: common_config
+                .rpc_server
+                .archival_mode
+                .unwrap_or_else(CommonGeneralRpcServerConfig::default_archival_mode),
         }
     }
 }

configuration/src/default_env_configs.rs

@@ -30,6 +30,17 @@ rpc_auth_token = "${RPC_AUTH_TOKEN}"
 ## Default value is redis://127.0.0.1/
 redis_url = "${REDIS_URL}"
 
+## Represents the number of available data ranges.
+## 1 means it will be available last 500_000 blocks (~11 epochs).
+## 2 means it will be available last 1_000_000 blocks (~22 epochs) etc.
+## Default value is 1
+available_data_ranges = "${AVAILABLE_DATA_RANGES}"
+
+## If true, means we store whole state changes in the database 
+## and available_data_ranges will be ignored 
+## and genesis block_height will be used as earliest_available_block_height
+archival_mode = "${ARCHIVAL_MODE}"
+
 ### Rpc server general configuration
 [general.rpc_server]
 

configuration/src/lib.rs

@@ -7,6 +7,7 @@ use validator::Validate;
 
 mod configs;
 mod default_env_configs;
+pub mod utils;
 
 pub use crate::configs::database::DatabaseConfig;
 pub use crate::configs::general::ChainId;

configuration/src/utils.rs

@@ -0,0 +1,70 @@
+use near_lake_framework::near_indexer_primitives::near_primitives;
+
+/// This constant represents the block height range.
+/// It is set to 500_000, which means that the data splits avery 500_000 blocks.
+pub const BLOCK_HEIGHT_RANGE: u64 = 500_000;
+
+/// # Explanation of the logic:
+///
+/// 1. `block_height / 500000`
+///
+///     * This integer division truncates any remainder, effectively grouping numbers into bins of 500,000.
+///
+///     * Example: 73908345 / 500000 = 147 (as integer division discards the remainder).
+///
+/// 2. `* 5`
+///
+///     * Since each bin represents a multiple of 500,000, multiplying by 5 scales the result to match the pattern you provided.
+///
+/// # Example Calculations:
+/// Example 1: 73908345
+///
+///     1. 73908345 / 500000 = 147
+///
+///     2. 147 * 5 = 735
+///
+///     Output: 735
+///
+/// Example 2: 130501000
+///     
+///     1. 130501000 / 500000 = 261
+///
+///     2. 261 * 5 = 1305
+///
+///     Output: 1305
+pub async fn get_data_range_id(
+    block_height: &near_primitives::types::BlockHeight,
+) -> anyhow::Result<u64> {
+    Ok((block_height / BLOCK_HEIGHT_RANGE) * 5)
+}
+
+/// This function calculates the earliest available block height based on the final block height
+pub async fn get_earliest_available_block_height(
+    final_block_height: u64,
+    available_data_ranges: u64,
+) -> anyhow::Result<u64> {
+    let final_range_id = get_data_range_id(&final_block_height).await?;
+    Ok(final_range_id * 100_000 + available_data_ranges * BLOCK_HEIGHT_RANGE)
+}
+
+/// This function checks if the block height is available in the database
+/// It returns an error if the block height is less than the earliest available block
+pub async fn check_block_height(
+    block_height: &near_primitives::types::BlockHeight,
+    final_block_height: &near_primitives::types::BlockHeight,
+    available_data_ranges: u64,
+    archival_mode: bool,
+) -> anyhow::Result<()> {
+    if archival_mode {
+        return Ok(());
+    }
+    let earliest_available_block =
+        get_earliest_available_block_height(*final_block_height, available_data_ranges).await?;
+    if *block_height < earliest_available_block {
+        anyhow::bail!(
+            "The data for block #{} is garbage collected on this node, use an archival node to fetch historical data",
+            block_height
+        )
+    }
+    Ok(())
+}

database/src/postgres/mod.rs

@@ -28,9 +28,27 @@ impl PageState {
     }
 }
 
+/// This struct is used to manage the connection to a specific shard database.
+/// It contains the shard_id, data_range_id, and a reference to the database pool.
+/// `shard_id` is used to identify the shard database.
+/// `data_range_id` is used to identify the range of data in the shard database.
+/// `pool` is a reference to the database pool that is used to connect to the shard database.
+///
+/// `data_range_id` is calculated based on the block height.
+/// Since we split the database into ranges of 500,000 blocks,
+/// we can use the block height to determine the range.
+/// For database, we store ranges in different tables and use `data_range_id` as the suffix.
+/// `state_changes_data_{data_range_id}`
+///
+/// Example:
+/// block_height 73_908_345, the `data_range_id` will be 735 -> `state_changes_data_735`
+/// block_height 130_501_000, the `data_range_id` will be 1305 -> `state_changes_data_1305`
+///
+/// How to get `data_range_id` from block_height
+/// see more details in the `get_data_range_id` function below.
 pub struct ShardIdPool<'a> {
     shard_id: near_primitives::types::ShardId,
-    table_number: u64,
+    data_range_id: u64,
     pool: &'a sqlx::Pool<sqlx::Postgres>,
 }
 
@@ -39,7 +57,6 @@ pub struct PostgresDBManager {
     shards_pool:
         std::collections::HashMap<near_primitives::types::ShardId, sqlx::Pool<sqlx::Postgres>>,
     meta_db_pool: sqlx::Pool<sqlx::Postgres>,
-    earliest_available_block: Option<near_primitives::types::BlockHeight>,
 }
 
 impl PostgresDBManager {
@@ -81,61 +98,14 @@ impl PostgresDBManager {
         let shard_id = self.shard_layout.account_id_to_shard_id(account_id);
         Ok(ShardIdPool {
             shard_id,
-            table_number: self.get_table_number(block_height).await?,
+            data_range_id: configuration::utils::get_data_range_id(block_height).await?,
             pool: self.shards_pool.get(&shard_id).ok_or(anyhow::anyhow!(
                 "Database connection for Shard_{} not found",
                 shard_id
             ))?,
         })
     }
 
-    /// # Explanation of the logic:
-    ///
-    /// 1. `block_height / 500000`
-    ///
-    ///     * This integer division truncates any remainder, effectively grouping numbers into bins of 500,000.
-    ///
-    ///     * Example: 73908345 / 500000 = 147 (as integer division discards the remainder).
-    ///
-    /// 2. `* 5`
-    ///
-    ///     * Since each bin represents a multiple of 500,000, multiplying by 5 scales the result to match the pattern you provided.
-    ///
-    /// # Example Calculations:
-    /// Example 1: 73908345
-    ///
-    ///     1. 73908345 / 500000 = 147
-    ///
-    ///     2. 147 * 5 = 735
-    ///
-    ///     Output: 735
-    ///
-    /// Example 2: 130501000
-    ///     
-    ///     1. 130501000 / 500000 = 261
-    ///
-    ///     2. 261 * 5 = 1305
-    ///
-    ///     Output: 1305
-    async fn get_table_number(
-        &self,
-        block_height: &near_primitives::types::BlockHeight,
-    ) -> anyhow::Result<u64> {
-        if let Some(earliest_available_block) = self.earliest_available_block {
-            if *block_height < earliest_available_block {
-                anyhow::bail!(
-                    "Block {} has been garbage collected. The earliest available block is {}",
-                    block_height,
-                    earliest_available_block
-                )
-            } else {
-                Ok((block_height / 500000) * 5)
-            }
-        } else {
-            Ok((block_height / 500000) * 5)
-        }
-    }
-
     async fn run_migrations(
         migrator: &sqlx::migrate::Migrator,
         pool: &sqlx::Pool<sqlx::Postgres>,
@@ -172,10 +142,6 @@ impl crate::BaseDbManager for PostgresDBManager {
             shard_layout,
             shards_pool,
             meta_db_pool,
-            // TODO: This should be set from the config and not hardcoded
-            // Should be updated when garbage collection is run
-            // or should be None for archive node_mode
-            earliest_available_block: None,
         }))
     }
 }