Add retry in game account fetching after subscription

Author: DogLooksGood

transactor-components/src/synchronizer.rs

@@ -39,7 +39,7 @@ async fn maybe_send_sync(
 
     // Drop duplicated updates
     if access_version <= prev_access_version {
-        info!("{} Drop update, this version {} < previous version {}", env.log_prefix, access_version, prev_access_version);
+        info!("{} Drop update, this version({}) is not newer than previous version({})", env.log_prefix, access_version, prev_access_version);
         return (prev_access_version, None);
     }
 
@@ -161,9 +161,8 @@ impl Component<PipelinePorts, GameSynchronizerContext> for GameSynchronizer {
             }
         };
 
-
+        // Do a first query, to handle those transactions made when our transactor is offline.
         let account = ctx.transport.get_game_account(&ctx.game_addr).await;
-
         if let Ok(Some(game_account)) = account {
             let (new_access_version, close_reason) = maybe_send_sync(
                 prev_access_version,
@@ -191,13 +190,16 @@ impl Component<PipelinePorts, GameSynchronizerContext> for GameSynchronizer {
                 }
 
                 sub_item = sub.next() => {
-
                     // The retry mechanism is implemented in `WrappedTransport`.  An Ok(Err(..))
                     // means the transport has gave up on reading game state.  The Ok(None) stands
                     // for the end of the stream, which is supposed to be sent after an error.  In
                     // both cases, we shutdown the game by sending a Shutdown frame.
                     match sub_item {
                         Some(Ok(game_account)) => {
+
+                            info!("{} Get account from subscription, access_version = {}, previous access version = {}",
+                                  env.log_prefix, game_account.access_version, prev_access_version);
+
                             let (new_access_version, close_reason) = maybe_send_sync(
                                 prev_access_version,
                                 game_account,
@@ -227,6 +229,7 @@ impl Component<PipelinePorts, GameSynchronizerContext> for GameSynchronizer {
             }
         }
 
+        info!("{} Synchronizer exits.", env.log_prefix);
         return CloseReason::Complete;
     }
 }

transactor/src/reg.rs

@@ -152,7 +152,7 @@ pub async fn start_reg_task(context: &ApplicationContext) -> JoinHandle<()> {
                         }
                     }
                 } else {
-                    warn!("Failed to load registration at {}", addr);
+                    warn!("Failed to load registration at {}, skip it for this time.", addr);
                 }
             }
 

transport/src/solana.rs

@@ -7,7 +7,6 @@ use constants::*;
 use futures::{Stream, StreamExt};
 use solana_transaction_status::UiTransactionEncoding;
 use tokio::sync::Mutex;
-use tokio::time::timeout;
 use std::time::Duration;
 use tracing::{error, info, warn};
 use types::*;
@@ -65,7 +64,6 @@ mod nft;
 const METAPLEX_PROGRAM_ID: &str = "metaqbxxUerdq28cj1RbAWkYQm3ybzjb6a8bt518x1s";
 
 const FEE_CAP: u64 = 20000;
-const RESTART_SUBSCRIPTION_TIMEOUT_SECS: u64 = 180;
 
 const PLAYERS_REG_HEAD_LEN: usize = 152;
 const PLAYERS_REG_INIT_LEN: usize = PLAYERS_REG_HEAD_LEN + 4;
@@ -1233,9 +1231,8 @@ impl TransportT for SolanaTransport {
                 }
             };
 
-            while let Ok(Some(rpc_response)) = timeout(
-                Duration::from_secs(RESTART_SUBSCRIPTION_TIMEOUT_SECS), stream.next()
-            ).await {
+            info!("Start the subscription for {}", addr);
+            while let Some(rpc_response) = stream.next().await {
                 let ui_account: UiAccount = rpc_response.value;
 
                 let Some(data) = ui_account.data.decode() else {
@@ -1259,16 +1256,31 @@ impl TransportT for SolanaTransport {
                     }
                 };
 
-                let (game_state, players_reg) = match self.internal_get_game_state_and_players_reg_state(
-                    &game_account_pubkey,
-                    &state.players_reg_account,
-                ).await {
-                    Ok((game_state, players_reg)) => (game_state, players_reg),
-                    Err(e) => {
-                        error!("Get players reg error: {}", e.to_string());
-                        unsub().await;
-                        return;
-                    }
+                info!("GameState From Sub, versions #A{}#S{}", state.access_version, state.settle_version);
+
+
+                // We have to make sure the we get the data that is as new as the one from subscription
+                // So we keep checking their access_version and settle_version.
+                let (game_state, players_reg) = loop {
+                    match self.internal_get_game_state_and_players_reg_state(
+                        &game_account_pubkey,
+                        &state.players_reg_account,
+                    ).await {
+                        Ok((game_state, players_reg)) => {
+                            info!("GameState From Query, versions #A{}#S{}", game_state.access_version, game_state.settle_version);
+                            if game_state.access_version != state.access_version
+                                || game_state.settle_version != state.settle_version {
+                                    info!("Retry query");
+                                    continue;
+                                }
+                            break (game_state, players_reg)
+                        }
+                        Err(e) => {
+                            error!("Get players reg error: {}", e.to_string());
+                            unsub().await;
+                            return;
+                        }
+                    };
                 };
 
                 let players = players_reg.players;
@@ -1288,6 +1300,7 @@ impl TransportT for SolanaTransport {
             // We restart subscription regularly to avoid broken connection.
             // A broken connection is the case where the connection is established,
             // no error raises, but no further update is delivered.
+            info!("Stop the subscription for {}", addr);
             unsub().await;
             return;
         }))