Add credential consolidator

DogLooksGood · DogLooksGood · commit ce2f83946d1d · 2026-01-05T11:47:21.000+08:00
diff --git a/encryptor/src/lib.rs b/encryptor/src/lib.rs
@@ -366,36 +366,9 @@ impl EncryptorT for Encryptor {
         Ok(())
     }
 
-    // fn add_public_key(&self, addr: String, raw: &NodePublicKeyRaw) -> EncryptorResult<()> {
-    //     let ec = import_ec_public(&raw.ec)?;
-    //     let rsa = import_rsa_public(&raw.rsa)?;
-
-    //     let mut public_keys = self
-    //         .publics
-    //         .lock()
-    //         .map_err(|_| EncryptorError::AddPublicKeyError)?;
-
-    //     public_keys.insert(addr, NodePublicKey { rsa, ec });
-    //     Ok(())
-    // }
-
     fn digest(&self, text: &[u8]) -> SecretDigest {
         Sha256::digest(text).to_vec()
     }
-
-    // fn export_public_key(&self, addr: Option<&str>) -> EncryptorResult<NodePublicKeyRaw> {
-    //     let publics = self
-    //         .publics
-    //         .lock()
-    //         .map_err(|_| EncryptorError::ReadPublicKeyError)?;
-    //     Ok(match addr {
-    //         Some(addr) => publics
-    //             .get(addr)
-    //             .ok_or(EncryptorError::PublicKeyNotfound)?
-    //             .try_into()?,
-    //         None => (&self.private).try_into()?,
-    //     })
-    // }
 }
 
 #[cfg(test)]
diff --git a/transactor-components/src/credential_consolidator.rs b/transactor-components/src/credential_consolidator.rs
@@ -0,0 +1,145 @@
+//! This component read the raw Sync event and update the encryptor to include their credentials.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use race_transactor_frames::EventFrame;
+use race_core::credentials::Credentials;
+use race_core::transport::TransportT;
+use race_core::encryptor::EncryptorT;
+use race_core::types::GameAccount;
+use tracing::{info, error};
+use borsh::BorshDeserialize;
+use super::{common::PipelinePorts, ComponentEnv};
+use crate::Component;
+use crate::CloseReason;
+
+pub async fn maybe_fetch_server_credentials(
+    server_addr: &str,
+    cached_server_credentials: &mut HashMap<String, Credentials>,
+    transport: Arc<dyn TransportT>,
+    env: &ComponentEnv,
+) -> Credentials {
+    if let Some(credentials) = cached_server_credentials.get(server_addr) {
+        return credentials.clone();
+    } else {
+        loop {
+            if let Ok(Some(profile)) = transport.get_server_account(&server_addr).await {
+                info!("{} Load server credentials: {}", env.log_prefix, server_addr);
+                let credentials = Credentials::try_from_slice(&profile.credentials).expect("Failed to deserialize Credentials");
+                cached_server_credentials.insert(server_addr.to_string(), credentials.clone());
+                return credentials;
+            } else {
+                error!("Failed to fetch server profile for {}, will retry.", server_addr);
+            }
+        }
+    }
+}
+
+pub async fn maybe_fetch_player_credentials(
+    player_addr: &str,
+    cached_player_credentials: &mut HashMap<String, Credentials>,
+    transport: Arc<dyn TransportT>,
+    env: &ComponentEnv,
+) -> Credentials {
+    if let Some(credentials) = cached_player_credentials.get(player_addr) {
+        return credentials.clone();
+    } else {
+        loop {
+            if let Ok(Some(profile)) = transport.get_player_profile(player_addr).await {
+                info!("{} Load server credentials: {}", env.log_prefix, player_addr);
+                let credentials = Credentials::try_from_slice(&profile.credentials).expect("Failed to deserialize Credentials");
+                cached_player_credentials.insert(player_addr.to_string(), credentials.clone());
+                return credentials;
+            } else {
+                error!("Failed to fetch player profile for {}, will retry.", player_addr);
+            }
+        }
+    }
+}
+
+pub struct CredentialConsolidatorContext {
+    transport: Arc<dyn TransportT>,
+    encryptor: Arc<dyn EncryptorT>,
+    game_addr: String,
+}
+
+pub struct CredentialConsolidator {}
+
+impl CredentialConsolidator {
+    pub fn init(
+        transport: Arc<dyn TransportT>,
+        encryptor: Arc<dyn EncryptorT>,
+        game_account: &GameAccount
+    ) -> (Self, CredentialConsolidatorContext) {
+        (
+            Self {},
+            CredentialConsolidatorContext {
+                transport,
+                encryptor,
+                game_addr: game_account.addr.clone(),
+            },
+        )
+    }
+}
+
+#[async_trait]
+impl Component<PipelinePorts, CredentialConsolidatorContext> for CredentialConsolidator {
+    fn name() -> &'static str {
+        "Credential Consolidator"
+    }
+
+    async fn run(
+        mut ports: PipelinePorts,
+        ctx: CredentialConsolidatorContext,
+        env: ComponentEnv,
+    ) -> CloseReason {
+        let CredentialConsolidatorContext {
+            transport, encryptor, game_addr: _
+        } = ctx;
+
+        let mut cached_player_credentials = HashMap::<String, Credentials>::default();
+        let mut cached_server_credentials = HashMap::<String, Credentials>::default();
+
+        while let Some(event_frame) = ports.recv().await {
+            match event_frame {
+                EventFrame::Shutdown => {
+                    return CloseReason::Complete;
+                }
+                EventFrame::Sync {
+                    new_players,
+                    new_servers,
+                    new_deposits,
+                    access_version,
+                    transactor_addr,
+                } => {
+                    for p in new_players.iter() {
+                        let credentials = maybe_fetch_player_credentials(&p.addr, &mut cached_player_credentials, transport.clone(), &env).await;
+                        if let Err(e) = encryptor.import_credentials(&p.addr, credentials) {
+                            return CloseReason::Fault(e.into());
+                        }
+                    }
+
+                    for s in new_servers.iter() {
+                        let credentials = maybe_fetch_server_credentials(&s.addr, &mut cached_server_credentials, transport.clone(), &env).await;
+                        if let Err(e) = encryptor.import_credentials(&s.addr, credentials) {
+                            return CloseReason::Fault(e.into());
+                        }
+                    }
+
+                    ports.send(EventFrame::SyncWithCredentials {
+                        new_players,
+                        new_servers,
+                        new_deposits,
+                        access_version,
+                        transactor_addr,
+                    }).await;
+                }
+                _ => {},
+            }
+        }
+
+        CloseReason::Complete
+    }
+}
diff --git a/transactor-components/src/event_loop.rs b/transactor-components/src/event_loop.rs
@@ -147,7 +147,7 @@ impl Component<PipelinePorts, EventLoopContext> for EventLoop {
                     }
                 }
 
-                EventFrame::Sync {
+                EventFrame::SyncWithCredentials {
                     new_players,
                     new_servers,
                     new_deposits,
diff --git a/transactor-components/src/lib.rs b/transactor-components/src/lib.rs
@@ -8,6 +8,7 @@ mod subscriber;
 mod synchronizer;
 mod refunder;
 mod voter;
+mod credential_consolidator;
 mod wrapped_client;
 mod wrapped_transport;
 mod wrapped_storage;
@@ -28,6 +29,7 @@ pub use subscriber::Subscriber;
 pub use synchronizer::GameSynchronizer;
 #[allow(unused)]
 pub use recorder::Recorder;
+pub use credential_consolidator::CredentialConsolidator;
 pub use voter::Voter;
 pub use wrapped_client::WrappedClient;
 pub use wrapped_transport::WrappedTransport;
diff --git a/transactor-components/src/subscriber.rs b/transactor-components/src/subscriber.rs
@@ -89,25 +89,21 @@ async fn handle_frame(frame: BroadcastFrame, ports: &mut PipelinePorts, env: &Co
 
             new_deposits.retain(|d| d.status == DepositStatus::Pending);
 
-            // XXX parse to the version with credentials
-            // if let Err(e) = ports
-            //     .try_send(EventFrame::Sync {
-            //         new_players,
-            //         new_servers,
-            //         new_deposits,
-            //         transactor_addr,
-            //         access_version,
-            //     })
-            //     .await
-            // {
-            //     error!("{} Send update node error: {}", env.log_prefix, e);
-            //     Some(CloseReason::Complete)
-            // } else {
-            //     None
-            // }
-
-            // REMOVE THIS
-            None
+            if let Err(e) = ports
+                .try_send(EventFrame::Sync {
+                    new_players,
+                    new_servers,
+                    new_deposits,
+                    transactor_addr,
+                    access_version,
+                })
+                .await
+            {
+                error!("{} Send update node error: {}", env.log_prefix, e);
+                Some(CloseReason::Complete)
+            } else {
+                None
+            }
         }
 
         BroadcastFrame::Message { .. } => {
diff --git a/transactor-components/src/synchronizer.rs b/transactor-components/src/synchronizer.rs
@@ -4,11 +4,8 @@ use async_trait::async_trait;
 use race_core::error::Error;
 use tokio::select;
 use tokio_stream::StreamExt;
-use std::collections::HashMap;
-use borsh::BorshDeserialize;
-use race_core::credentials::Credentials;
 
-use race_transactor_frames::{EventFrame, PlayerJoinSync, ServerJoinSync};
+use race_transactor_frames::EventFrame;
 use race_core::{
     transport::TransportT,
     types::{GameAccount, PlayerDeposit, PlayerJoin, ServerJoin},
@@ -25,70 +22,7 @@ pub struct GameSynchronizerContext {
     game_addr: String,
 }
 
-async fn fetch_server_with_credentials(
-    server_join: ServerJoin,
-    cached_server_credentials: &mut HashMap<String, Credentials>,
-    transport: Arc<dyn TransportT>,
-) -> ServerJoinSync {
-    if let Some(credentials) = cached_server_credentials.get(&server_join.addr) {
-        return ServerJoinSync {
-            addr: server_join.addr,
-            endpoint: server_join.endpoint,
-            access_version: server_join.access_version,
-            credentials: credentials.clone(),
-        }
-    } else {
-        loop {
-            if let Ok(Some(profile)) = transport.get_player_profile(&server_join.addr).await {
-                let credentials = Credentials::try_from_slice(&profile.credentials).expect("Failed to deserialize Credentials");
-                cached_server_credentials.insert(server_join.addr.clone(), credentials.clone());
-                return ServerJoinSync {
-                    addr: server_join.addr,
-                    endpoint: server_join.endpoint,
-                    access_version: server_join.access_version,
-                    credentials,
-                }
-            } else {
-                error!("Failed to fetch server profile for {}, will retry.", server_join.addr);
-            }
-        }
-    }
-}
-
-async fn fetch_player_with_credentials(
-    player_join: PlayerJoin,
-    cached_player_credentials: &mut HashMap<String, Credentials>,
-    transport: Arc<dyn TransportT>,
-) -> PlayerJoinSync {
-    if let Some(credentials) = cached_player_credentials.get(&player_join.addr) {
-        return PlayerJoinSync {
-            addr: player_join.addr,
-            position: player_join.position,
-            access_version: player_join.access_version,
-            credentials: credentials.clone(),
-        }
-    } else {
-        loop {
-            if let Ok(Some(profile)) = transport.get_player_profile(&player_join.addr).await {
-                let credentials = Credentials::try_from_slice(&profile.credentials).expect("Failed to deserialize Credentials");
-                cached_player_credentials.insert(player_join.addr.clone(), credentials.clone());
-                return PlayerJoinSync {
-                    addr: player_join.addr,
-                    position: player_join.position,
-                    access_version: player_join.access_version,
-                    credentials,
-                }
-            } else {
-                error!("Failed to fetch player profile for {}, will retry.", player_join.addr);
-            }
-        }
-    }
-}
-
 async fn maybe_send_sync(
-    transport: Arc<dyn TransportT>,
-    cached_player_credentials: &mut HashMap<String, Credentials>,
-    cached_server_credentials: &mut HashMap<String, Credentials>,
     prev_access_version: u64,
     game_account: GameAccount,
     ports: &mut PipelinePorts,
@@ -113,22 +47,15 @@ async fn maybe_send_sync(
         env.log_prefix, game_account.access_version, game_account.settle_version,
     );
 
-    let mut new_players: Vec<PlayerJoinSync> = Vec::with_capacity(players.len());
-    let mut new_servers: Vec<ServerJoinSync> = Vec::with_capacity(servers.len());
-
-    for p in players {
-        if p.access_version > prev_access_version { // We care only new players
-            let player_sync = fetch_player_with_credentials(p, cached_player_credentials, transport.clone()).await;
-            new_players.push(player_sync);
-        }
-    }
+    let new_players: Vec<PlayerJoin> = players
+        .into_iter()
+        .filter(|p| p.access_version > prev_access_version)
+        .collect();
 
-    for s in servers {
-        if s.access_version > prev_access_version { // We care only new servers
-            let server_sync = fetch_server_with_credentials(s, cached_server_credentials, transport.clone()).await;
-            new_servers.push(server_sync);
-        }
-    }
+    let new_servers: Vec<ServerJoin> = servers
+        .into_iter()
+        .filter(|s| s.access_version > prev_access_version)
+        .collect();
 
     let new_deposits: Vec<PlayerDeposit> = deposits
         .into_iter()
@@ -216,9 +143,6 @@ impl Component<PipelinePorts, GameSynchronizerContext> for GameSynchronizer {
         ctx: GameSynchronizerContext,
         env: ComponentEnv,
     ) -> CloseReason {
-        let mut cached_player_credentials = HashMap::<String, Credentials>::default();
-        let mut cached_server_credentials = HashMap::<String, Credentials>::default();
-
         let mut prev_access_version = ctx.access_version;
 
         let mut sub = match ctx.transport.subscribe_game_account(&ctx.game_addr).await {
@@ -253,9 +177,6 @@ impl Component<PipelinePorts, GameSynchronizerContext> for GameSynchronizer {
                     match sub_item {
                         Some(Ok(game_account)) => {
                             let (new_access_version, close_reason) = maybe_send_sync(
-                                ctx.transport.clone(),
-                                &mut cached_player_credentials,
-                                &mut cached_server_credentials,
                                 prev_access_version,
                                 game_account,
                                 &mut ports,
diff --git a/transactor-frames/src/lib.rs b/transactor-frames/src/lib.rs

Original file line number	Diff line number	Diff line change
`@@ -147,7 +147,7 @@ impl Component<PipelinePorts, EventLoopContext> for EventLoop {`
`147`	`147`	`}`
`148`	`148`	`}`
`149`	`149`
`150`		`- EventFrame::Sync {`
	`150`	`+ EventFrame::SyncWithCredentials {`
`151`	`151`	`new_players,`
`152`	`152`	`new_servers,`
`153`	`153`	`new_deposits,`