feat: gap detection & rewind

.gitignore

@@ -13,3 +13,4 @@ test-ledger/
 minio
 test.db
 docker-compose.yml
+*.txt

Cargo.toml

@@ -33,6 +33,10 @@ path = "src/snapshot/loader/main.rs"
 name = "photon-tree-validator"
 path = "src/tools/tree_validator/main.rs"
 
+[[bin]]
+name = "photon-analyze-snapshot"
+path = "src/tools/analyze_snapshot.rs"
+
 [dependencies]
 ark-serialize = "0.5"
 ark-bn254 = "0.5"
@@ -82,9 +86,11 @@ solana-pubkey = "2.3.0"
 solana-transaction-status = "1.18.0"
 
 light-concurrent-merkle-tree = "2.1.0"
-light-batched-merkle-tree = "0.3.0"
-light-merkle-tree-metadata = "0.3.0"
-light-compressed-account = { version = "0.3.0", features = ["anchor"] }
+light-batched-merkle-tree = { version = "0.3.0", git = "https://github.com/lightprotocol/light-protocol", rev = "341aae4dfc89a27913e6ff1af65572d626b0cc19" }
+light-merkle-tree-metadata = { version = "0.3.0", git = "https://github.com/lightprotocol/light-protocol", rev = "341aae4dfc89a27913e6ff1af65572d626b0cc19" }
+light-compressed-account = { version = "0.3.0", features = [
+    "anchor",
+], git = "https://github.com/lightprotocol/light-protocol", rev = "341aae4dfc89a27913e6ff1af65572d626b0cc19" }
 light-hasher = { version = "3.1.0" }
 
 light-poseidon = "0.3.0"

src/api/method/get_multiple_compressed_accounts.rs

@@ -148,7 +148,11 @@ pub async fn get_multiple_compressed_accounts(
             }
             fetch_account_from_addresses(conn, addresses).await?
         }
-        _ => panic!("Either hashes or addresses must be provided"),
+        _ => {
+            return Err(PhotonApiError::ValidationError(
+                "Either hashes or addresses must be provided".to_string(),
+            ));
+        }
     };
 
     Ok(GetMultipleCompressedAccountsResponse {

src/api/method/get_transaction_with_compression_info.rs

@@ -210,6 +210,7 @@ pub async fn get_transaction_helper(
             PhotonApiError::UnexpectedError(format!("Failed to parse transaction {}", signature.0))
         })?,
         slot,
+        None,
     )
     .map_err(|_e| {
         PhotonApiError::UnexpectedError(format!("Failed to parse transaction {}", signature.0))
@@ -360,6 +361,7 @@ pub async fn get_transaction_helper_v2(
             PhotonApiError::UnexpectedError(format!("Failed to parse transaction {}", signature.0))
         })?,
         slot,
+        None,
     )
     .map_err(|_e| {
         PhotonApiError::UnexpectedError(format!("Failed to parse transaction {}", signature.0))

src/ingester/error.rs

@@ -14,6 +14,10 @@ pub enum IngesterError {
     EmptyBatchEvent,
     #[error("Invalid event.")]
     InvalidEvent,
+    #[error("Custom error: {0}")]
+    CustomError(String),
+    #[error("Gap detected, triggering rewind")]
+    GapDetectedRewind,
 }
 
 impl From<sea_orm::error::DbErr> for IngesterError {

src/ingester/fetchers/grpc.rs

@@ -15,6 +15,7 @@ use solana_client::nonblocking::rpc_client::RpcClient;
 use solana_pubkey::Pubkey;
 use solana_sdk::pubkey::Pubkey as SdkPubkey;
 use solana_sdk::signature::Signature;
+use tokio::sync::mpsc;
 use tokio::time::sleep;
 use tracing::error;
 use yellowstone_grpc_client::{GeyserGrpcBuilderResult, GeyserGrpcClient, Interceptor};
@@ -30,6 +31,7 @@ use yellowstone_grpc_proto::solana::storage::confirmed_block::InnerInstructions;
 use crate::api::method::get_indexer_health::HEALTH_CHECK_SLOT_DISTANCE;
 use crate::common::typedefs::hash::Hash;
 use crate::ingester::fetchers::poller::get_block_poller_stream;
+use crate::ingester::gap::RewindCommand;
 use crate::ingester::typedefs::block_info::{
     BlockInfo, BlockMetadata, Instruction, InstructionGroup, TransactionInfo,
 };
@@ -43,6 +45,7 @@ pub fn get_grpc_stream_with_rpc_fallback(
     rpc_client: Arc<RpcClient>,
     mut last_indexed_slot: u64,
     max_concurrent_block_fetches: usize,
+    rewind_receiver: Option<mpsc::UnboundedReceiver<RewindCommand>>,
 ) -> impl Stream<Item = Vec<BlockInfo>> {
     stream! {
         start_latest_slot_updater(rpc_client.clone()).await;
@@ -53,6 +56,7 @@ pub fn get_grpc_stream_with_rpc_fallback(
                 rpc_client.clone(),
                 last_indexed_slot,
                 max_concurrent_block_fetches,
+                rewind_receiver,
             ))
         );
 
@@ -115,6 +119,7 @@ pub fn get_grpc_stream_with_rpc_fallback(
                                 rpc_client.clone(),
                                 last_indexed_slot,
                                 max_concurrent_block_fetches,
+                                None, // No rewind receiver for timeout fallback
                             )));
                             continue;
                         }
@@ -132,6 +137,7 @@ pub fn get_grpc_stream_with_rpc_fallback(
                             rpc_client.clone(),
                             last_indexed_slot,
                             max_concurrent_block_fetches,
+                            None, // No rewind receiver for out-of-order fallback
                         )));
                         continue;
                     }
@@ -144,6 +150,7 @@ pub fn get_grpc_stream_with_rpc_fallback(
                             rpc_client.clone(),
                             last_indexed_slot,
                             max_concurrent_block_fetches,
+                            None, // No rewind receiver for unhealthy fallback
                         )));
                     }
                 }

src/ingester/fetchers/mod.rs

@@ -3,12 +3,14 @@ use std::sync::Arc;
 use async_stream::stream;
 use futures::{pin_mut, Stream, StreamExt};
 use solana_client::nonblocking::rpc_client::RpcClient;
+use tokio::sync::mpsc;
 
 use super::typedefs::block_info::BlockInfo;
 
 pub mod grpc;
 pub mod poller;
 
+use crate::ingester::gap::RewindCommand;
 use grpc::get_grpc_stream_with_rpc_fallback;
 use poller::get_block_poller_stream;
 
@@ -17,10 +19,11 @@ pub struct BlockStreamConfig {
     pub geyser_url: Option<String>,
     pub max_concurrent_block_fetches: usize,
     pub last_indexed_slot: u64,
+    pub rewind_receiver: Option<mpsc::UnboundedReceiver<RewindCommand>>,
 }
 
 impl BlockStreamConfig {
-    pub fn load_block_stream(&self) -> impl Stream<Item = Vec<BlockInfo>> {
+    pub fn load_block_stream(mut self) -> impl Stream<Item = Vec<BlockInfo>> {
         let grpc_stream = self.geyser_url.as_ref().map(|geyser_url| {
             let auth_header = std::env::var("GRPC_X_TOKEN").unwrap();
             get_grpc_stream_with_rpc_fallback(
@@ -29,6 +32,7 @@ impl BlockStreamConfig {
                 self.rpc_client.clone(),
                 self.last_indexed_slot,
                 self.max_concurrent_block_fetches,
+                self.rewind_receiver.take(),
             )
         });
 
@@ -37,6 +41,7 @@ impl BlockStreamConfig {
                 self.rpc_client.clone(),
                 self.last_indexed_slot,
                 self.max_concurrent_block_fetches,
+                self.rewind_receiver.take(),
             ))
         } else {
             None