feat: store first block in epoch

chain/src/main.rs

@@ -136,6 +136,12 @@ async fn crawling_fn(
             .await
             .into_rpc_error()?;
 
+    tracing::info!("Query first block in epoch...");
+    let first_block_in_epoch =
+        namada_service::get_first_block_in_epoch(&client)
+            .await
+            .into_rpc_error()?;
+
     let block = Block::from(
         tm_block_response,
         &block_results,
@@ -198,6 +204,7 @@ async fn crawling_fn(
     let crawler_state = ChainCrawlerState {
         last_processed_block: block_height,
         last_processed_epoch: epoch,
+        first_block_in_epoch,
         timestamp: timestamp_in_sec,
     };
 
@@ -269,6 +276,9 @@ async fn initial_query(
         namada_service::get_epoch_at_block_height(client, block_height)
             .await
             .into_rpc_error()?;
+    let first_block_in_epoch = namada_service::get_first_block_in_epoch(client)
+        .await
+        .into_rpc_error()?;
 
     loop {
         let pos_crawler_state =
@@ -317,6 +327,7 @@ async fn initial_query(
     let crawler_state = ChainCrawlerState {
         last_processed_block: block_height,
         last_processed_epoch: epoch,
+        first_block_in_epoch,
         timestamp,
     };
 

chain/src/repository/crawler_state.rs

@@ -22,6 +22,8 @@ pub fn upsert_crawler_state(
                 .eq(excluded(crawler_state::last_processed_block)),
             crawler_state::last_processed_epoch
                 .eq(excluded(crawler_state::last_processed_epoch)),
+            crawler_state::first_block_in_epoch
+                .eq(excluded(crawler_state::first_block_in_epoch)),
         ))
         .execute(transaction_conn)
         .context("Failed to update crawler state in db")?;

chain/src/services/db.rs

@@ -19,6 +19,7 @@ pub async fn get_chain_crawler_state(
                 .select((
                     crawler_state::dsl::last_processed_block,
                     crawler_state::dsl::last_processed_epoch,
+                    crawler_state::dsl::first_block_in_epoch,
                     crawler_state::dsl::timestamp,
                 ))
                 .first(conn)
@@ -30,6 +31,7 @@ pub async fn get_chain_crawler_state(
     Ok(ChainCrawlerState {
         last_processed_block: crawler_state.last_processed_block as BlockHeight,
         last_processed_epoch: crawler_state.last_processed_epoch as Epoch,
+        first_block_in_epoch: crawler_state.first_block_in_epoch as BlockHeight,
         timestamp: crawler_state.timestamp.and_utc().timestamp(),
     })
 }

chain/src/services/namada.rs

@@ -38,6 +38,18 @@ pub async fn get_native_token(client: &HttpClient) -> anyhow::Result<Id> {
     Ok(Id::from(native_token))
 }
 
+pub async fn get_first_block_in_epoch(
+    client: &HttpClient,
+) -> anyhow::Result<BlockHeight> {
+    let block_height = RPC
+        .shell()
+        .first_block_height_of_current_epoch(client)
+        .await
+        .context("Failed to query native token")?;
+
+    Ok(block_height.0 as BlockHeight)
+}
+
 pub async fn get_epoch_at_block_height(
     client: &HttpClient,
     block_height: BlockHeight,

orm/migrations/2024-07-04-103941_crawler_state/up.sql

@@ -5,6 +5,7 @@ CREATE TYPE CRAWLER_NAME AS ENUM ('chain', 'governance', 'parameters', 'pos', 'r
 CREATE TABLE crawler_state (
     name CRAWLER_NAME PRIMARY KEY NOT NULL,
     last_processed_block INT,
+    first_block_in_epoch INT,
     last_processed_epoch INT,
     timestamp TIMESTAMP NOT NULL
 );

orm/src/crawler_state.rs

@@ -55,36 +55,41 @@ pub struct CrawlerStateDb {
     pub name: CrawlerNameDb,
     pub last_processed_block: Option<i32>,
     pub last_processed_epoch: Option<i32>,
+    pub first_block_in_epoch: Option<i32>,
     pub timestamp: chrono::NaiveDateTime,
 }
 
 #[derive(Serialize, Clone, Debug)]
 pub struct ChainCrawlerStateDb {
     pub last_processed_block: i32,
     pub last_processed_epoch: i32,
+    pub first_block_in_epoch: i32,
     pub timestamp: chrono::NaiveDateTime,
 }
 impl
     Queryable<
         (
+            Nullable<diesel::sql_types::Integer>,
             Nullable<diesel::sql_types::Integer>,
             Nullable<diesel::sql_types::Integer>,
             diesel::sql_types::Timestamp,
         ),
         Pg,
     > for ChainCrawlerStateDb
 {
-    type Row = (Option<i32>, Option<i32>, chrono::NaiveDateTime);
+    type Row = (Option<i32>, Option<i32>, Option<i32>, chrono::NaiveDateTime);
 
     fn build(row: Self::Row) -> diesel::deserialize::Result<Self> {
         match row {
             (
                 Some(last_processed_block),
                 Some(last_processed_epoch),
+                Some(first_block_in_epoch),
                 timestamp,
             ) => Ok(Self {
                 last_processed_block,
                 last_processed_epoch,
+                first_block_in_epoch,
                 timestamp,
             }),
             _ => Err("last_processed_block or last_processed_epoch missing \
@@ -167,6 +172,7 @@ pub struct ChainStateInsertDb {
     pub name: CrawlerNameDb,
     pub last_processed_block: i32,
     pub last_processed_epoch: i32,
+    pub first_block_in_epoch: i32,
     pub timestamp: chrono::NaiveDateTime,
 }
 
@@ -219,6 +225,7 @@ impl From<(CrawlerName, ChainCrawlerState)> for ChainStateInsertDb {
             name: crawler_name.into(),
             last_processed_block: state.last_processed_block as i32,
             last_processed_epoch: state.last_processed_epoch as i32,
+            first_block_in_epoch: state.first_block_in_epoch as i32,
             timestamp,
         }
     }

orm/src/schema.rs

@@ -109,6 +109,7 @@ diesel::table! {
     crawler_state (name) {
         name -> CrawlerName,
         last_processed_block -> Nullable<Int4>,
+        first_block_in_epoch -> Nullable<Int4>,
         last_processed_epoch -> Nullable<Int4>,
         timestamp -> Timestamp,
     }

shared/src/crawler_state.rs

@@ -13,6 +13,7 @@ pub enum CrawlerName {
 pub struct ChainCrawlerState {
     pub last_processed_block: BlockHeight,
     pub last_processed_epoch: Epoch,
+    pub first_block_in_epoch: Epoch,
     pub timestamp: i64,
 }
 

webserver/src/repository/chain.rs

@@ -66,6 +66,7 @@ impl ChainRepositoryTrait for ChainRepository {
                 .select((
                     crawler_state::dsl::last_processed_block,
                     crawler_state::dsl::last_processed_epoch,
+                    crawler_state::dsl::first_block_in_epoch,
                     crawler_state::dsl::timestamp,
                 ))
                 .first(conn)

webserver/src/response/governance.rs

@@ -122,8 +122,11 @@ impl Proposal {
         // in state and use that here, otherwise if any epoch had
         // different number of blocks than min_num_of_blocks
         // this will be off
-        let epoch_progress =
-            epoch_progress(chain_state.last_processed_block, min_num_of_blocks);
+        let epoch_progress = epoch_progress(
+            chain_state.last_processed_block,
+            chain_state.first_block_in_epoch,
+            min_num_of_blocks,
+        );
 
         let to_start = time_between_epochs(
             min_num_of_blocks,

webserver/src/response/pos.rs

@@ -178,8 +178,11 @@ impl Unbond {
         min_num_of_blocks: i32,
         min_duration: i32,
     ) -> Self {
-        let epoch_progress =
-            epoch_progress(chain_state.last_processed_block, min_num_of_blocks);
+        let epoch_progress = epoch_progress(
+            chain_state.last_processed_block,
+            chain_state.first_block_in_epoch,
+            min_num_of_blocks,
+        );
 
         let to_withdraw = time_between_epochs(
             min_num_of_blocks,

webserver/src/response/utils.rs

@@ -35,9 +35,11 @@ where
     }
 }
 
-pub fn epoch_progress(current_block: i32, min_num_of_blocks: i32) -> f64 {
-    // Not sure why but real min number of blocks is usually 2 more what is in
-    // store
+pub fn epoch_progress(
+    current_block: i32,
+    first_block_in_epoch: i32,
+    min_num_of_blocks: i32,
+) -> f64 {
     let min_num_of_blocks =
         min_num_of_blocks + (EPOCH_SWITCH_BLOCKS_DELAY as i32);
 
@@ -46,7 +48,7 @@ pub fn epoch_progress(current_block: i32, min_num_of_blocks: i32) -> f64 {
     let current_block = current_block - 1;
 
     // Calculate the block in the current epoch
-    let block_in_current_epoch = current_block % min_num_of_blocks;
+    let block_in_current_epoch = current_block - first_block_in_epoch;
 
     // Calculate how much into the epoch we are
     block_in_current_epoch as f64 / min_num_of_blocks as f64