Allow speciyfing latest as start slot (#209)

helius-labs · Sep 24, 2024 · 59a6b02 · 59a6b02
1 parent a22eace
commit 59a6b02
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 12 deletions.
diff --git a/src/ingester/indexer/mod.rs b/src/ingester/indexer/mod.rs
@@ -58,7 +58,11 @@ pub async fn index_block_stream(
 ) {
     pin_mut!(block_stream);
     let current_slot = end_slot.unwrap_or(fetch_current_slot_with_infinite_retry(rpc_client).await);
-    let number_of_blocks_to_backfill = current_slot - last_indexed_slot_at_start;
+    let number_of_blocks_to_backfill = if current_slot > last_indexed_slot_at_start {
+        current_slot - last_indexed_slot_at_start
+    } else {
+        0
+    };
     info!(
         "Backfilling historical blocks. Current number of blocks to backfill: {}",
         number_of_blocks_to_backfill

diff --git a/src/main.rs b/src/main.rs
@@ -54,7 +54,7 @@ struct Args {
     /// The start slot to begin indexing from. Defaults to the last indexed slot in the database plus
     /// one.  
     #[arg(short, long)]
-    start_slot: Option<u64>,
+    start_slot: Option<String>,
 
     /// Max database connections to use in database pool
     #[arg(long, default_value_t = 10)]
@@ -256,15 +256,27 @@ async fn main() {
                     }
                 }
             };
-
             let last_indexed_slot = match args.start_slot {
-                Some(start_slot) => fetch_block_parent_slot(&rpc_client.client, start_slot).await,
-                None => {
-                    (fetch_last_indexed_slot_with_infinite_retry(db_conn.as_ref())
+                Some(start_slot) => match start_slot.as_str() {
+                    "latest" => fetch_current_slot_with_infinite_retry(&rpc_client.client).await,
+                    _ => {
+                        fetch_block_parent_slot(
+                            &rpc_client.client,
+                            start_slot.parse::<u64>().unwrap(),
+                        )
                         .await
-                        .unwrap_or(get_network_start_slot(&rpc_client.client).await as i64))
-                        as u64
-                }
+                    }
+                },
+                None => fetch_last_indexed_slot_with_infinite_retry(db_conn.as_ref())
+                    .await
+                    .unwrap_or(
+                        get_network_start_slot(&rpc_client.client)
+                            .await
+                            .try_into()
+                            .unwrap(),
+                    )
+                    .try_into()
+                    .unwrap(),
             };
 
             let block_stream_config = BlockStreamConfig {

diff --git a/src/snapshot/snapshotter/main.rs b/src/snapshot/snapshotter/main.rs
@@ -5,6 +5,7 @@ use photon_indexer::common::typedefs::rpc_client_with_uri::RpcClientWithUri;
 use photon_indexer::common::{
     fetch_block_parent_slot, get_network_start_slot, setup_logging, setup_metrics, LoggingFormat,
 };
+use photon_indexer::ingester::fetchers::poller::fetch_current_slot_with_infinite_retry;
 use photon_indexer::ingester::fetchers::BlockStreamConfig;
 use photon_indexer::snapshot::{
     get_snapshot_files_with_metadata, load_byte_stream_from_directory_adapter, DirectoryAdapter,
@@ -31,9 +32,9 @@ struct Args {
     #[arg(short, long, default_value = "http://127.0.0.1:8899")]
     rpc_url: String,
 
-    /// The start slot to begin indexing from
+    /// The start slot to begin indexing from. If "latest", the latest slot is used.
     #[arg(short, long)]
-    start_slot: Option<u64>,
+    start_slot: Option<String>,
 
     /// Logging format
     #[arg(short, long, default_value_t = LoggingFormat::Standard)]
@@ -224,12 +225,23 @@ async fn main() {
         let snapshot_files = get_snapshot_files_with_metadata(directory_adapter.as_ref())
             .await
             .unwrap();
+
         let last_indexed_slot = match args.start_slot {
             Some(start_slot) => {
                 if !snapshot_files.is_empty() {
                     panic!("Cannot specify start_slot when snapshot files are present");
                 }
-                fetch_block_parent_slot(&rpc_client.client, start_slot).await
+                let start_slot = match start_slot.as_str() {
+                    "latest" => fetch_current_slot_with_infinite_retry(&rpc_client.client).await,
+                    _ => {
+                        fetch_block_parent_slot(
+                            &rpc_client.client,
+                            start_slot.parse::<u64>().unwrap(),
+                        )
+                        .await
+                    }
+                };
+                start_slot
             }
             None => {
                 if snapshot_files.is_empty() {