add pending update detection during stake extract

stake_aggregate/src/main.rs

@@ -11,17 +11,6 @@
   }
 '
 
-
-curl http://localhost:3000 -X POST -H "Content-Type: application/json" -d '
-  {
-    "jsonrpc": "2.0",
-    "id" : 1,
-    "method": "bootstrap_accounts",
-    "params": []
-  }
-' -o extract_stake_532_agg.json
-
-
 curl http://localhost:3000 -X POST -H "Content-Type: application/json" -d '
   {
     "jsonrpc": "2.0",
@@ -299,9 +288,21 @@ async fn run_loop<F: Interceptor>(mut client: GeyserGrpcClient<F>) -> anyhow::Re
                     }
                     crate::rpc::Requests::GetStakestore(tx) => {
                         let current_stakes = stakestore.get_cloned_stake_map();
-                        if let Err(err) = tx.send((current_stakes, current_epoch_state.current_slot.confirmed_slot)){
+                        let extract_slot = if stakestore.extracted {
+                            log::error!("Stakestore extracted:{}", stakestore.extracted);
+                            0
+                        }
+                        else if stakestore.updates.len() > 0 {
+                            log::error!("Extract with pending stake:{}", stakestore.updates.len());
+                            0
+                        } else {
+                            current_epoch_state.current_slot.confirmed_slot
+                        };
+
+                        if let Err(err) = tx.send((current_stakes, extract_slot)){
                             println!("Channel error during sending back request status error:{err:?}");
                         }
+
                         log::info!("RPC GetStakestore account send");
                     },
                     crate::rpc::Requests::GetVotestore(tx) => {

stake_aggregate/src/rpc.rs

@@ -253,17 +253,21 @@ pub(crate) async fn run_server(request_tx: Sender<Requests>) -> Result<ServerHan
             .map_err(|err| format!("error during bootstrap query processing:{err}"))
             .and_then(|(accounts, slot)| {
                 println!("RPC end request status");
-                //replace pubkey with String. Json only allow distionary key with string.
-                let ret: HashMap<String, StoredVote> = accounts
-                    .into_iter()
-                    .map(|(pk, acc)| (pk.to_string(), acc))
-                    .collect();
-                serde_json::to_value((slot, ret)).map_err(|err| {
-                    format!(
-                        "error during json serialisation:{}",
-                        JsonRpcError::ParseError(err)
-                    )
-                })
+                if slot == 0 {
+                    Err("Error stake store extracted".to_string())
+                } else {
+                    //replace pubkey with String. Json only allow distionary key with string.
+                    let ret: HashMap<String, StoredVote> = accounts
+                        .into_iter()
+                        .map(|(pk, acc)| (pk.to_string(), acc))
+                        .collect();
+                    serde_json::to_value((slot, ret)).map_err(|err| {
+                        format!(
+                            "error during json serialisation:{}",
+                            JsonRpcError::ParseError(err)
+                        )
+                    })
+                }
             })
             .unwrap_or_else(|err| serde_json::Value::String(err.to_string()))
     })?;

stake_aggregate/src/stakestore.rs

@@ -133,8 +133,8 @@ impl StoredStake {
 pub struct StakeStore {
     stakes: StakeMap,
     stake_history: Option<StakeHistory>,
-    updates: Vec<ExtractedAction>,
-    extracted: bool,
+    pub updates: Vec<ExtractedAction>,
+    pub extracted: bool,
 }
 
 impl StakeStore {