diff --git a/api/src/effect.rs b/api/src/effect.rs index e5078add..501a6a0d 100644 --- a/api/src/effect.rs +++ b/api/src/effect.rs @@ -3,6 +3,8 @@ use std::collections::HashMap; use borsh::{BorshDeserialize, BorshSerialize}; +#[cfg(feature = "serde")] +use serde::{Deserialize, Serialize}; use crate::{ engine::GameHandler, @@ -10,7 +12,10 @@ use crate::{ event::BridgeEvent, prelude::InitAccount, random::RandomSpec, - types::{Award, DecisionId, EntryLock, GameDeposit, GameId, GamePlayer, PlayerBalance, RandomId, Transfer}, + types::{ + Award, DecisionId, EntryLock, GameDeposit, GameId, GamePlayer, PlayerBalance, RandomId, + Transfer, + }, }; #[derive(BorshSerialize, BorshDeserialize, Debug, PartialEq, Eq)] @@ -55,6 +60,7 @@ pub struct ActionTimeout { } #[derive(BorshSerialize, BorshDeserialize, Debug, PartialEq, Eq, Clone)] +#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] pub struct SubGame { pub id: GameId, pub bundle_addr: String, @@ -92,6 +98,7 @@ impl SubGame { } #[derive(BorshSerialize, BorshDeserialize, Debug, PartialEq, Eq, Clone)] +#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] pub struct EmitBridgeEvent { pub dest: GameId, pub raw: Vec, @@ -104,6 +111,13 @@ impl EmitBridgeEvent { raw: borsh::to_vec(&bridge_event)?, }) } + + pub fn new_empty(dest: GameId) -> Self { + Self { + dest, + raw: vec![] + } + } } #[derive(BorshSerialize, BorshDeserialize, Debug, PartialEq, Eq, Clone)] diff --git a/api/src/engine.rs b/api/src/engine.rs index b13cb6e6..edb9dfe6 100644 --- a/api/src/engine.rs +++ b/api/src/engine.rs @@ -4,7 +4,7 @@ use crate::{effect::Effect, error::HandleResult, event::Event, init_account::Ini pub trait GameHandler: Sized + BorshSerialize + BorshDeserialize { /// Initialize handler state with on-chain game account data. - fn init_state(init_account: InitAccount) -> HandleResult; + fn init_state(effect: &mut Effect, init_account: InitAccount) -> HandleResult; /// Handle event. fn handle_event(&mut self, effect: &mut Effect, event: Event) -> HandleResult<()>; diff --git a/api/src/event.rs b/api/src/event.rs index 929ae495..bb069cfb 100644 --- a/api/src/event.rs +++ b/api/src/event.rs @@ -30,10 +30,6 @@ pub enum Event { raw: Vec, }, - /// A event sent by system, the first event sent by transactor - /// when game is loaded. - Ready, - /// Transactor shares its secert to specific player. /// The `secret_data` is encrypted with the receiver's public key. ShareSecrets { @@ -159,7 +155,6 @@ impl std::fmt::Display for Event { dest_game_id, from_game_id, raw[0] ) } - Event::Ready => write!(f, "Ready"), Event::ShareSecrets { sender, shares } => { let repr = shares .iter() diff --git a/api/src/init_account.rs b/api/src/init_account.rs index 14bfdd5c..9e887b01 100644 --- a/api/src/init_account.rs +++ b/api/src/init_account.rs @@ -1,9 +1,12 @@ use borsh::{BorshDeserialize, BorshSerialize}; +#[cfg(feature = "serde")] +use serde::{Deserialize, Serialize}; use crate::error::HandleError; /// A set of arguments for game handler initialization. #[derive(Debug, Clone, BorshSerialize, BorshDeserialize, PartialEq, Eq)] +#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] pub struct InitAccount { pub max_players: u16, pub data: Vec, diff --git a/api/src/types.rs b/api/src/types.rs index 159de474..5067ebc5 100644 --- a/api/src/types.rs +++ b/api/src/types.rs @@ -66,7 +66,7 @@ impl Settle { } } -#[derive(Clone, BorshSerialize, BorshDeserialize, Debug, PartialEq, Eq)] +#[derive(Clone, BorshSerialize, BorshDeserialize, Debug, PartialEq, Eq, Default)] #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] #[cfg_attr(feature = "serde", serde(rename_all = "camelCase"))] pub struct Award { diff --git a/core/src/checkpoint.rs b/core/src/checkpoint.rs index 3c9a7118..e2f767ba 100644 --- a/core/src/checkpoint.rs +++ b/core/src/checkpoint.rs @@ -1,17 +1,17 @@ use std::{collections::HashMap, fmt::Display}; -use borsh::{BorshDeserialize, BorshSerialize}; -use rs_merkle::{ - algorithms::Sha256, proof_serializers::ReverseHashesOrder, Hasher, - MerkleTree, +use crate::{ + context::{DispatchEvent, Versions}, + error::Error, }; -use crate::{context::Versions, error::Error}; +use borsh::{BorshDeserialize, BorshSerialize}; +use rs_merkle::{algorithms::Sha256, proof_serializers::ReverseHashesOrder, Hasher, MerkleTree}; #[cfg(feature = "serde")] use serde::{Deserialize, Serialize}; -use race_api::types::GameId; use crate::types::GameSpec; +use race_api::{effect::{EmitBridgeEvent, SubGame}, types::GameId}; /// Checkpoint represents the state snapshot of game. /// It is used as a submission to the blockchain. @@ -23,6 +23,7 @@ pub struct Checkpoint { pub access_version: u64, pub data: HashMap, pub proofs: HashMap>, + pub launch_subgames: Vec, } #[derive(Default, Debug, Clone, PartialEq, Eq, BorshSerialize, BorshDeserialize)] @@ -40,6 +41,7 @@ pub struct CheckpointOnChain { pub struct CheckpointOffChain { pub data: HashMap, pub proofs: HashMap>, + pub launch_subgames: Vec, } #[derive(Clone, Debug, Default, PartialEq, Eq, BorshSerialize, BorshDeserialize)] @@ -50,6 +52,15 @@ pub struct VersionedData { pub data: Vec, pub sha: Vec, pub game_spec: GameSpec, + pub dispatch: Option, + pub bridge_events: Vec, +} + +impl VersionedData { + fn clear_future_events(&mut self) { + self.dispatch = None; + self.bridge_events.clear(); + } } impl Display for Checkpoint { @@ -77,20 +88,27 @@ impl Checkpoint { versions, data: root_data, sha: sha.into(), + dispatch: None, + bridge_events: vec![], }, )]), proofs: HashMap::new(), + launch_subgames: vec![], }; ret.update_root_and_proofs(); ret } - pub fn new_from_parts(offchain_part: CheckpointOffChain, onchain_part: CheckpointOnChain) -> Self { + pub fn new_from_parts( + offchain_part: CheckpointOffChain, + onchain_part: CheckpointOnChain, + ) -> Self { Self { proofs: offchain_part.proofs, data: offchain_part.data, access_version: onchain_part.access_version, root: onchain_part.root, + launch_subgames: offchain_part.launch_subgames, } } @@ -103,7 +121,7 @@ impl Checkpoint { return; } let merkle_tree = self.to_merkle_tree(); - let Some(root) = merkle_tree.root() else { + let Some(root) = merkle_tree.root() else { // Skip root update as this is not a master checkpoint return; }; @@ -142,7 +160,13 @@ impl Checkpoint { } } - pub fn init_data(&mut self, id: GameId, game_spec: GameSpec, versions: Versions, data: Vec) -> Result<(), Error> { + pub fn init_data( + &mut self, + id: GameId, + game_spec: GameSpec, + versions: Versions, + data: Vec, + ) -> Result<(), Error> { match self.data.entry(id) { std::collections::hash_map::Entry::Occupied(_) => { return Err(Error::CheckpointAlreadyExists); @@ -156,6 +180,8 @@ impl Checkpoint { sha: sha.into(), game_spec, versions, + dispatch: None, + bridge_events: vec![], }; v.insert(versioned_data); self.update_root_and_proofs(); @@ -188,6 +214,48 @@ impl Checkpoint { } } + pub fn set_dispatch_in_versioned_data( + &mut self, + id: GameId, + dispatch: Option, + ) -> Result<(), Error> { + if let Some(versioned_data) = self.data.get_mut(&id) { + versioned_data.dispatch = dispatch; + Ok(()) + } else { + Err(Error::MissingCheckpoint) + } + } + + pub fn set_bridge_in_versioned_data( + &mut self, + id: GameId, + bridge_events: Vec, + ) -> Result<(), Error> { + if let Some(versioned_data) = self.data.get_mut(&id) { + versioned_data.bridge_events = bridge_events; + Ok(()) + } else { + Err(Error::MissingCheckpoint) + } + } + + pub fn append_launch_subgames(&mut self, subgame: SubGame) { + self.launch_subgames.push(subgame); + } + + pub fn delete_launch_subgames(&mut self, id: GameId) { + self.launch_subgames.retain(|subgame| subgame.id != id); + } + + pub fn get_launch_subgames(&self) -> Vec { + self.launch_subgames.clone() + } + + pub fn clear_future_events(&mut self) { + self.data.values_mut().for_each(VersionedData::clear_future_events); + } + pub fn list_versioned_data(&self) -> Vec<&VersionedData> { self.data.values().collect() } @@ -203,14 +271,14 @@ impl Checkpoint { pub fn get_sha(&self, id: GameId) -> Option<[u8; 32]> { self.data .get(&id) - .map(|d| d.sha.clone().try_into().unwrap()) + .map(|d| d.sha.clone().try_into().expect("Failed to get SHA")) } pub fn to_merkle_tree(&self) -> MerkleTree { let mut leaves: Vec<[u8; 32]> = vec![]; let mut i = 0; while let Some(vd) = self.data.get(&i) { - leaves.push(vd.sha.clone().try_into().unwrap()); + leaves.push(vd.sha.clone().try_into().expect("Failed to build merkle tree")); i += 1; } MerkleTree::from_leaves(&leaves) @@ -232,6 +300,7 @@ impl Checkpoint { CheckpointOffChain { proofs: self.proofs.clone(), data: self.data.clone(), + launch_subgames: self.launch_subgames.clone(), } } diff --git a/core/src/context.rs b/core/src/context.rs index 24334665..8ab26c90 100644 --- a/core/src/context.rs +++ b/core/src/context.rs @@ -1,16 +1,19 @@ use crate::checkpoint::{Checkpoint, CheckpointOffChain, VersionedData}; -use crate::random::{RandomState, RandomStatus}; -use crate::types::{ClientMode, EntryType, GameAccount, GameSpec}; use crate::decision::DecisionState; use crate::error::{Error, Result}; +use crate::random::{RandomState, RandomStatus}; +use crate::types::{ClientMode, EntryType, GameAccount, GameSpec}; use borsh::{BorshDeserialize, BorshSerialize}; -use race_api::effect::{Ask, Assign, Effect, EmitBridgeEvent, Log, Release, Reveal, SubGame, Withdraw}; +use race_api::effect::{ + Ask, Assign, Effect, EmitBridgeEvent, Log, Release, Reveal, SubGame, Withdraw, +}; use race_api::engine::GameHandler; use race_api::event::{CustomEvent, Event}; use race_api::prelude::InitAccount; use race_api::random::RandomSpec; use race_api::types::{ - Award, BalanceChange, Ciphertext, DecisionId, EntryLock, GameId, GamePlayer, GameStatus, PlayerBalance, RandomId, SecretDigest, SecretShare, Settle, Transfer + Award, BalanceChange, Ciphertext, DecisionId, EntryLock, GameId, GamePlayer, GameStatus, + PlayerBalance, RandomId, SecretDigest, SecretShare, Settle, Transfer, }; #[cfg(feature = "serde")] use serde::{Deserialize, Serialize}; @@ -29,7 +32,10 @@ pub struct Versions { impl Versions { pub fn new(access_version: u64, settle_version: u64) -> Self { - Self { access_version, settle_version } + Self { + access_version, + settle_version, + } } } @@ -91,6 +97,7 @@ impl Node { } #[derive(Clone, Debug, BorshSerialize, BorshDeserialize, PartialEq, Eq)] +#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] pub struct DispatchEvent { pub timeout: u64, pub event: Event, @@ -123,7 +130,6 @@ impl ContextPlayer { } } - #[derive(Debug, Clone)] pub enum SubGameInitSource { FromCheckpoint(VersionedData), @@ -137,7 +143,6 @@ pub struct SubGameInit { pub source: SubGameInitSource, } - /// The effects of an event, indicates what actions should be taken /// after the event handling. /// @@ -147,16 +152,54 @@ pub struct SubGameInit { /// - start_game: to start game. #[derive(Debug, Default, PartialEq, Eq)] pub struct EventEffects { - pub settles: Vec, - pub transfer: Option, - pub awards: Vec, - pub checkpoint: Option, pub launch_sub_games: Vec, pub bridge_events: Vec, pub start_game: bool, - pub entry_lock: Option, + pub stop_game: bool, pub logs: Vec, pub reject_deposits: Vec, + pub checkpoint: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SettleDetails { + pub settles: Vec, + pub transfer: Option, + pub awards: Vec, + pub checkpoint: Checkpoint, + pub access_version: u64, + pub settle_version: u64, + pub previous_settle_version: u64, + pub entry_lock: Option, + pub accept_deposits: Vec, + pub settle_locks: HashMap, +} + +impl SettleDetails { + pub fn handle_versioned_data(&mut self, versioned_data: VersionedData) -> Result<()> { + self.settle_locks.remove(&versioned_data.id); + if self.checkpoint.data.contains_key(&versioned_data.id) { + self.checkpoint.update_versioned_data(versioned_data)?; + } else { + self.checkpoint.init_versioned_data(versioned_data)?; + } + Ok(()) + } + + pub fn print(&self, prefix: String) { + println!("-------- {} --------", prefix); + println!("Version: {}", self.settle_version); + println!("Locks: {:?}", self.settle_locks); + println!("Settles: {:?}", self.settles); + println!("Transfer: {:?}", self.transfer); + println!("Awards: {:?}", self.awards); + println!("Checkpoint:"); + for data in self.checkpoint.data.iter() { + println!("- Id: {}", data.1.id); + println!(" Dispatch: {:?}", data.1.dispatch); + println!(" Bridge Events: {:?}", data.1.bridge_events); + } + } } /// The context for public data. @@ -217,11 +260,19 @@ pub struct GameContext { pub(crate) state_sha: String, /// Accepted deposits, saved for later use in settlement pub(crate) accept_deposits: Vec, + /// The pending settle details. They are here because the settle is blocked by the settle locks. + pub(crate) pending_settle_details: Vec, + /// The locks for the under construction settle + pub(crate) next_settle_locks: HashMap, } impl GameContext { pub fn try_new_with_sub_game_spec(init: SubGameInit) -> Result { - let SubGameInit { spec, nodes, source } = init; + let SubGameInit { + spec, + nodes, + source, + } = init; let (handler_state, versions, init_data, checkpoint) = match source { SubGameInitSource::FromCheckpoint(versioned_data) => { @@ -283,10 +334,7 @@ impl GameContext { game_account.players.iter().for_each(|p| { // Only include those players joined before last checkpoint if p.access_version <= access_version { - players.push(ContextPlayer::new( - p.access_version, - p.position, - )); + players.push(ContextPlayer::new(p.access_version, p.position)); } nodes.push(Node::new_pending( @@ -329,7 +377,7 @@ impl GameContext { init_account: InitAccount { max_players: checkpoint_data.game_spec.max_players, data: vec![], // Not necessary? - } + }, }); } } @@ -351,14 +399,16 @@ impl GameContext { entry_type: game_account.entry_type.clone(), state_sha, accept_deposits: vec![], + pending_settle_details: vec![], + next_settle_locks: HashMap::new(), }) } - pub fn init_account(&self) -> Result { - Ok(InitAccount { + pub fn init_account(&self) -> InitAccount { + InitAccount { max_players: self.spec.max_players, data: self.init_data.clone(), - }) + } } pub fn checkpoint_is_empty(&self) -> bool { @@ -413,6 +463,23 @@ impl GameContext { &mut self.checkpoint } + pub fn pending_settle_details_mut(&mut self) -> &mut Vec { + &mut self.pending_settle_details + } + + pub fn take_first_ready_settle_details(&mut self) -> Option { + if self + .pending_settle_details + .first() + .is_some_and(|cp| cp.settle_locks.is_empty()) + { + let cp = self.pending_settle_details.remove(0); + Some(cp) + } else { + None + } + } + pub fn set_handler_state(&mut self, handler: &H) where H: GameHandler, @@ -626,9 +693,9 @@ impl GameContext { } /// Dispatch an event if there's none - pub fn dispatch_safe(&mut self, event: Event, timeout: u64) { + pub fn set_dispatch(&mut self, event: Option) { if self.dispatch.is_none() { - self.dispatch = Some(DispatchEvent::new(event, timeout + self.timestamp)); + self.dispatch = event; } } @@ -642,7 +709,10 @@ impl GameContext { } pub fn has_sub_game(&self, game_id: GameId) -> bool { - self.sub_games.iter().find(|sub_game| sub_game.id == game_id).is_some() + self.sub_games + .iter() + .find(|sub_game| sub_game.id == game_id) + .is_some() } pub fn init_random_state(&mut self, spec: RandomSpec) -> Result { @@ -762,6 +832,52 @@ impl GameContext { Ok(()) } + pub fn handle_versioned_data( + &mut self, + game_id: GameId, + versioned_data: VersionedData, + is_init: bool, + ) -> Result<()> { + if let Some(&old_version) = self.next_settle_locks.get(&game_id) { + if old_version < versioned_data.versions.settle_version { + self.next_settle_locks.remove(&game_id); + } + } + + if is_init { + self.checkpoint_mut() + .init_versioned_data(versioned_data.clone())?; + self.checkpoint_mut().delete_launch_subgames(game_id); + } else { + self.checkpoint_mut() + .update_versioned_data(versioned_data.clone())?; + } + + let cur_id = self.game_id(); + for settle_detail in self.pending_settle_details.iter_mut() { + if let Some(&old_version) = settle_detail.settle_locks.get(&game_id) { + if old_version < versioned_data.versions.settle_version { + settle_detail.settle_locks.remove(&game_id); + } + } + + if settle_detail.settle_locks.is_empty() { + if let Some(master_cp) = settle_detail.checkpoint.get_versioned_data(0) { + for be in master_cp.bridge_events.clone() { + settle_detail + .checkpoint + .set_bridge_in_versioned_data(be.dest, vec![])?; + } + } + } + if cur_id == 0 { + settle_detail.print("handle_versioned_data".to_string()); + } + } + + Ok(()) + } + pub fn add_revealed_random( &mut self, random_id: RandomId, @@ -876,6 +992,7 @@ impl GameContext { reject_deposits, mut accept_deposits, balances, + entry_lock, .. } = effect; @@ -924,46 +1041,86 @@ impl GameContext { self.init_random_state(spec)?; } + let previous_settle_version = self.settle_version(); + if let Some(state) = handler_state { self.set_handler_state_raw(state.clone()); + let mut settles = vec![]; if is_checkpoint { + let settles_map = build_settles_map(&withdraws, &ejects, &self.balances, &balances); + self.balances = balances; + settles = settles_map.into_values().collect(); + settles.sort_by_key(|s| s.player_id); + self.random_states.clear(); self.decision_states.clear(); self.bump_settle_version()?; self.checkpoint.set_data(self.spec.game_id, state)?; - self.checkpoint.set_access_version(self.versions.access_version); + self.checkpoint + .set_access_version(self.versions.access_version); self.set_game_status(GameStatus::Idle); } else if is_init { self.bump_settle_version()?; - self.checkpoint.init_data(self.spec.game_id, self.spec.clone(), self.versions, state)?; - self.checkpoint.set_access_version(self.versions.access_version); + self.checkpoint.init_data( + self.spec.game_id, + self.spec.clone(), + self.versions, + state, + )?; + self.checkpoint + .set_access_version(self.versions.access_version); self.set_game_status(GameStatus::Idle); } + let game_id = self.game_id(); + let dispatch = self.dispatch.clone(); // Append SubGame to context for sub_game in launch_sub_games.iter().cloned() { - self.sub_games.push(sub_game); + self.sub_games.push(sub_game.clone()); + self.checkpoint_mut().append_launch_subgames(sub_game); + } + self.checkpoint_mut() + .set_dispatch_in_versioned_data(game_id, dispatch)?; + self.checkpoint_mut() + .set_bridge_in_versioned_data(game_id, bridge_events.clone())?; + + if is_checkpoint || is_init { + let settle_details = SettleDetails { + settles, + transfer, + awards, + checkpoint: self.checkpoint.clone(), + access_version: self.access_version(), + settle_version: self.settle_version(), + previous_settle_version, + entry_lock, + accept_deposits: self.accept_deposits.drain(..).collect(), + settle_locks: self.next_settle_locks.clone(), + }; + self.pending_settle_details.push(settle_details); } - let mut settles = vec![]; - if is_checkpoint { - let settles_map = build_settles_map(&withdraws, &ejects, &self.balances, &balances); - self.balances = balances; - settles = settles_map.into_values().collect(); + if self.spec.game_id == 0 { + // Add lock into next_settle_locks + for evt in bridge_events.clone() { + let settle_version = self + .checkpoint + .get_versioned_data(evt.dest) + .map(|d| d.versions.settle_version) + .unwrap_or(0); + self.next_settle_locks.insert(evt.dest, settle_version); + } } return Ok(EventEffects { - transfer, - awards, - settles, - checkpoint: (is_checkpoint || is_init).then(|| self.checkpoint.clone()), launch_sub_games, bridge_events, start_game, - entry_lock: effect.entry_lock, + stop_game, logs: effect.logs, reject_deposits, + checkpoint: (is_checkpoint || is_init).then(|| self.checkpoint.clone()), }); } else if let Some(e) = error { return Err(Error::HandleError(e)); @@ -1033,34 +1190,44 @@ impl Default for GameContext { entry_type: EntryType::Disabled, state_sha: "".into(), accept_deposits: vec![], + pending_settle_details: vec![], + next_settle_locks: HashMap::new(), } } } -fn build_settles_map(withdraws: &[Withdraw], ejects: &[u64], old_balances: &[PlayerBalance], new_balances: &[PlayerBalance]) -> HashMap { - +fn build_settles_map( + withdraws: &[Withdraw], + ejects: &[u64], + old_balances: &[PlayerBalance], + new_balances: &[PlayerBalance], +) -> HashMap { // Build settles // Settle is a combination of Withdraw, Balance Diff, and Eject let mut settles_map: HashMap = HashMap::new(); for withdraw in withdraws { - settles_map.entry(withdraw.player_id) + settles_map + .entry(withdraw.player_id) .and_modify(|e| e.withdraw += withdraw.amount) .or_insert_with(|| Settle::new(withdraw.player_id, withdraw.amount, None, false)); } for eject in ejects.iter() { - settles_map.entry(*eject) + settles_map + .entry(*eject) .and_modify(|e| e.eject = true) .or_insert_with(|| Settle::new(*eject, 0, None, true)); } - let mut balances_change: HashMap = old_balances.iter() + let mut balances_change: HashMap = old_balances + .iter() .map(|orig_balance| (orig_balance.player_id, -(orig_balance.balance as i128))) .collect(); for balance in new_balances.iter() { - balances_change.entry(balance.player_id) + balances_change + .entry(balance.player_id) .and_modify(|e| *e += balance.balance as i128) .or_insert(balance.balance as i128); } @@ -1072,7 +1239,8 @@ fn build_settles_map(withdraws: &[Withdraw], ejects: &[u64], old_balances: &[Pla _ => None, }; if change.is_some() { - settles_map.entry(player_id) + settles_map + .entry(player_id) .and_modify(|e| e.change = change) .or_insert_with(|| Settle::new(player_id, 0, change, false)); } @@ -1084,9 +1252,213 @@ fn build_settles_map(withdraws: &[Withdraw], ejects: &[u64], old_balances: &[Pla #[cfg(test)] mod tests { - use race_api::effect::Withdraw; - use super::*; + use race_api::effect::{EmitBridgeEvent, SubGame, Withdraw}; + + #[test] + fn given_effect_with_bridge_event_do_apply_effect() -> anyhow::Result<()> { + let mut ctx = GameContext::default(); + let mut ef = Effect::default(); + let mut cp = Checkpoint::new(0, GameSpec::default(), Versions::default(), vec![]); + cp.init_data(1, GameSpec::default(), Versions::default(), vec![])?; + ctx.checkpoint = cp; + ef.handler_state = Some(vec![1]); + ef.bridge_events = vec![EmitBridgeEvent::new_empty(1)]; + ef.is_checkpoint = true; + ctx.apply_effect(ef)?; + assert!(ctx.next_settle_locks.is_empty()); + assert_eq!(ctx.pending_settle_details.len(), 1); + assert_eq!( + ctx.checkpoint().data.get(&0).unwrap().bridge_events.len(), + 1 + ); + + // A future call on handle_versioned_data should remove the lock for SettleDetails + let mut vd = VersionedData::default(); + vd.id = 1; + vd.versions.settle_version = 1; + ctx.handle_versioned_data(1, vd, false)?; + assert!(ctx.pending_settle_details[0] + .checkpoint + .data + .get(&1) + .unwrap() + .bridge_events + .is_empty()); + assert_eq!( + ctx.pending_settle_details[0].settle_locks, + HashMap::default() + ); + Ok(()) + } + + #[test] + fn given_init_sub_checkpoint_do_handle_versioned_data() {} + + #[test] + fn given_sub_checkpoint_before_main_checkpoint_do_handle_versioned_data() -> anyhow::Result<()> + { + let mut ctx = GameContext::default(); + let mut vd = VersionedData::default(); + let mut cp = Checkpoint::default(); + cp.data.insert(0, VersionedData::default()); + cp.data.insert(1, VersionedData::default()); + vd.id = 1; + vd.bridge_events = vec![EmitBridgeEvent::new_empty(0)]; + ctx.next_settle_locks.insert(1, SettleLock::Locked(0)); + ctx.checkpoint = cp; + ctx.handle_versioned_data(1, vd.clone(), false)?; + vd.bridge_events = vec![]; + assert_eq!( + ctx.next_settle_locks.get(&1), + Some(&SettleLock::Unlocked(vd)) + ); + assert_eq!(ctx.checkpoint().data.get(&1).unwrap().bridge_events, vec![]); + match ctx.next_settle_locks.get(&1) { + Some(SettleLock::Unlocked(vd)) => { + assert_eq!(vd.bridge_events, vec![]); + } + _ => panic!("not a settle lock"), + } + Ok(()) + } + + #[test] + fn given_sub_checkpoint_after_main_checkpoint_do_handle_versioned_data() { + let mut ctx = GameContext::default(); + let mut vd = VersionedData::default(); + let mut cp = Checkpoint::default(); + let mut sd = SettleDetails::default(); + sd.settle_locks.insert(1, 0); + + cp.data.insert(0, VersionedData::default()); + cp.data.insert(1, VersionedData::default()); + sd.checkpoint = cp.clone(); + let e = EmitBridgeEvent::new_empty(0); + vd.id = 1; + vd.bridge_events = vec![e.clone()]; + ctx.pending_settle_details_mut().push(sd); + ctx.checkpoint = cp; + ctx.handle_versioned_data(1, vd.clone(), false).unwrap(); + assert_eq!(ctx.next_settle_locks.get(&1), None,); + assert_eq!(ctx.next_settle_locks.len(), 0); + assert_eq!(ctx.pending_settle_details[0].settle_locks.len(), 0); + } + + #[test] + fn test_apply_effect_with_all_fields() { + // Setting up a GameContext + let mut game_context = GameContext::default(); + game_context.versions = Versions::new(10, 10); + game_context.checkpoint = + Checkpoint::new(0, GameSpec::default(), game_context.versions(), vec![1]); + game_context.add_node("server".into(), 0, ClientMode::Transactor); + game_context.add_node("alice".into(), 1, ClientMode::Player); + game_context.add_node("bob".into(), 2, ClientMode::Player); + game_context.set_node_ready(10); + + // Defining handler state + let handler_state = vec![1, 2, 3, 4, 5]; + + // Creating Effect with non-empty fields + let effect = Effect { + start_game: true, + stop_game: false, + cancel_dispatch: true, + action_timeout: None, + wait_timeout: Some(1000), + timestamp: 1234567890, + curr_random_id: 1, + curr_decision_id: 1, + nodes_count: 2, + asks: vec![], + assigns: vec![], + reveals: vec![], + releases: vec![], + init_random_states: vec![RandomSpec::deck_of_cards()], + revealed: HashMap::new(), + answered: HashMap::new(), + is_checkpoint: true, + withdraws: vec![Withdraw { + player_id: 1, + amount: 1000, + }], + ejects: vec![1], + handler_state: Some(handler_state.clone()), + error: None, + transfer: Some(Transfer { amount: 200 }), + launch_sub_games: vec![SubGame { + id: 1, + bundle_addr: String::from("address"), + init_account: InitAccount { + max_players: 4, + data: vec![], + }, + }], + bridge_events: vec![EmitBridgeEvent { + dest: 1, + raw: vec![1], + }], + is_init: false, + entry_lock: Some(EntryLock::Closed), + logs: vec![], + awards: vec![Award::new(1, String::from("bonus"))], + reject_deposits: vec![1, 2, 3], + accept_deposits: vec![4, 5, 6], + curr_sub_game_id: 1, + balances: vec![PlayerBalance::new(1, 1000), PlayerBalance::new(2, 2000)], + }; + + let effect_bridge_events = effect.bridge_events.clone(); + + // Apply the effect + let event_effects = game_context.apply_effect(effect).unwrap(); + + assert_eq!(event_effects.launch_sub_games.len(), 1); + assert_eq!(event_effects.bridge_events.len(), 1); + assert!(event_effects.start_game); + assert!(event_effects.checkpoint.is_some()); + assert_eq!(event_effects.reject_deposits, vec![1, 2, 3]); + assert_eq!(game_context.settle_version(), 11); + assert_eq!(game_context.pending_settle_details.len(), 1); + let settle_details = game_context.pending_settle_details[0].clone(); + assert_eq!( + settle_details.settles, + [ + Settle { + player_id: 1, + withdraw: 1000, + change: Some(BalanceChange::Add(1000)), + eject: true + }, + Settle { + player_id: 2, + withdraw: 0, + change: Some(BalanceChange::Add(2000)), + eject: false + }, + ] + ); + assert_eq!(settle_details.transfer, Some(Transfer { amount: 200 })); + assert_eq!(settle_details.awards, vec![Award::new(1, "bonus".into())]); + assert!(!settle_details.checkpoint.root.is_empty()); + assert_eq!(settle_details.access_version, 10); + assert_eq!(settle_details.settle_version, 11); + assert_eq!(settle_details.previous_settle_version, 10); + assert_eq!(settle_details.entry_lock, Some(EntryLock::Closed)); + assert_eq!(settle_details.accept_deposits, vec![4, 5, 6]); + assert_eq!(settle_details.settle_locks, HashMap::from([(1, 0)])); + + let game_id = game_context.game_id(); + let bridge_events = game_context + .checkpoint() + .get_versioned_data(game_id) + .unwrap() + .bridge_events + .clone(); + assert_eq!(bridge_events, effect_bridge_events); + assert_eq!(game_context.dispatch, None); + } // #[test] // fn test_build_settle_map() { diff --git a/core/src/engine.rs b/core/src/engine.rs index 3c55773c..59b8ea73 100644 --- a/core/src/engine.rs +++ b/core/src/engine.rs @@ -11,8 +11,6 @@ pub fn general_handle_event( ) -> Result<(), Error> { // General event handling match event { - Event::Ready => Ok(()), - Event::ShareSecrets { sender, shares } => { let addr = context.id_to_addr(*sender)?; context.add_shared_secrets(addr, shares.clone())?; diff --git a/core/src/types/transactor_params.rs b/core/src/types/transactor_params.rs index 9565229c..c46b68a3 100644 --- a/core/src/types/transactor_params.rs +++ b/core/src/types/transactor_params.rs @@ -16,6 +16,7 @@ pub struct AttachGameParams { #[derive(Debug, Clone, PartialEq, Eq, BorshDeserialize, BorshSerialize)] #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))] +#[cfg_attr(feature = "serde", serde(rename_all = "camelCase"))] pub struct CheckpointParams { pub settle_version: u64, } diff --git a/dev/Justfile b/dev/Justfile index ecd81ec7..3196d89c 100644 --- a/dev/Justfile +++ b/dev/Justfile @@ -10,6 +10,9 @@ demo-app: mtt-simple: zellij --layout layouts/mtt-simple.kdl +mtt-multi: + zellij --layout layouts/mtt-multi.kdl + ltmtt-simple: zellij --layout layouts/ltmtt-simple.kdl diff --git a/dev/layouts/mtt-multi.kdl b/dev/layouts/mtt-multi.kdl new file mode 100644 index 00000000..364c2e54 --- /dev/null +++ b/dev/layouts/mtt-multi.kdl @@ -0,0 +1,30 @@ +// The layout to launch facade and one transactors in Zellij + +layout { + pane split_direction="vertical" { + pane { + name "Facade" + focus true + cwd "../" + command "bash" + args "dev/scripts/facade-mtt-game.sh" + start_suspended true + } + pane split_direction="horizontal" { + pane { + name "Server 1" + command "just" + cwd "../" + args "dev-transactor" "dev/server-confs/server1.toml" + start_suspended true + } + pane { + name "Server 2" + command "just" + cwd "../" + args "dev-transactor" "dev/server-confs/server2.toml" + start_suspended true + } + } + } +} diff --git a/dev/scripts/misc.sh b/dev/scripts/misc.sh index 4df10bb0..421f1d56 100644 --- a/dev/scripts/misc.sh +++ b/dev/scripts/misc.sh @@ -3,7 +3,7 @@ BORSH="$ROOT/../race-sdk/packages/borsh" # Make a JSON specification for tourney game # TITLE TOKEN START_TIME function make_tourney { - local ENTRY_CLOSE_TIME=$(expr $3 + 300000) + local ENTRY_CLOSE_TIME=$(expr $3 + 50000) local TICKET=100000000 local TABLE_SIZE=$4 local START_CHIPS=100000 diff --git a/local-db/src/lib.rs b/local-db/src/lib.rs index 7eb72e35..ec73fa8a 100644 --- a/local-db/src/lib.rs +++ b/local-db/src/lib.rs @@ -50,7 +50,6 @@ impl StorageT for LocalDbStorage { if let Some(checkpoint_bs) = checkpoint_bs { let checkpoint = CheckpointOffChain::try_from_slice(&checkpoint_bs) .map_err(|e| Error::StorageError(e.to_string()))?; - Ok(Some(checkpoint)) } else { Ok(None) diff --git a/proc-macro/src/lib.rs b/proc-macro/src/lib.rs index 17ee4443..fd2ddcc5 100644 --- a/proc-macro/src/lib.rs +++ b/proc-macro/src/lib.rs @@ -95,7 +95,7 @@ pub fn game_handler(_metadata: TokenStream, input: TokenStream) -> TokenStream { } else { return 2 }; - match #s_idt::init_state(init_account) { + match #s_idt::init_state(&mut effect, init_account) { Ok(handler) => effect.__set_handler_result(handler), Err(e) => effect.__set_error(e), } diff --git a/test/src/context_helpers.rs b/test/src/context_helpers.rs index 953cc901..ed46a7a0 100644 --- a/test/src/context_helpers.rs +++ b/test/src/context_helpers.rs @@ -71,7 +71,7 @@ impl TestContext { } pub fn init_account(&self) -> Result { - self.context.init_account() + Ok(self.context.init_account()) } pub fn state(&self) -> &H { @@ -181,6 +181,8 @@ impl TestContextBuilder { data: borsh::to_vec(checkpoint).expect("Failed to serialize checkpoint"), sha: vec![], game_spec: spec, + dispatch: None, + bridge_events: vec![], }, ); diff --git a/test/src/handler_helpers.rs b/test/src/handler_helpers.rs index fe5ce193..7e9af47f 100644 --- a/test/src/handler_helpers.rs +++ b/test/src/handler_helpers.rs @@ -29,7 +29,7 @@ where impl TestHandler { pub fn init_state(context: &mut GameContext) -> Result<(Self, EventEffects)> { let mut new_context = context.clone(); - let init_account = new_context.init_account()?; + let init_account = new_context.init_account(); let effect = new_context.derive_effect(true); let handler = H::init_state(init_account)?; let event_effects = new_context.apply_effect(effect)?; diff --git a/transactor/src/component.rs b/transactor/src/component.rs index 493342e7..9f9cdaa2 100644 --- a/transactor/src/component.rs +++ b/transactor/src/component.rs @@ -9,6 +9,7 @@ mod synchronizer; mod refunder; mod voter; mod wrapped_client; +mod handler; mod wrapped_handler; mod wrapped_transport; mod wrapped_storage; @@ -32,3 +33,4 @@ pub use wrapped_transport::WrappedTransport; pub use wrapped_storage::WrappedStorage; pub use event_bridge::{EventBridgeChild, EventBridgeParent, BridgeToParent}; pub use refunder::Refunder; +pub use handler::HandlerT; diff --git a/transactor/src/component/broadcaster.rs b/transactor/src/component/broadcaster.rs index aa8b80ec..37162664 100644 --- a/transactor/src/component/broadcaster.rs +++ b/transactor/src/component/broadcaster.rs @@ -5,8 +5,8 @@ use std::collections::LinkedList; use std::sync::Arc; use async_trait::async_trait; -use race_api::types::GameId; use race_api::event::Event; +use race_api::types::GameId; use race_core::checkpoint::CheckpointOffChain; use race_core::types::{BroadcastFrame, BroadcastSync, TxState}; use tokio::sync::{broadcast, RwLock}; @@ -142,8 +142,10 @@ impl Broadcaster { // By default, returns the histories with settle_version // greater than the given one + info!("Get backlog with settle_version: {}", settle_version); if settle_version > 0 { for group in event_backup_groups.iter() { + info!("group settle version: {:?}", group.settle_version); if group.settle_version == settle_version { checkpoint_off_chain = group.checkpoint_off_chain.clone(); state_sha = group.state_sha.clone(); @@ -197,7 +199,6 @@ impl Component for Broadcaster { settle_version, checkpoint, state_sha, - .. } => { info!( "{} Create new history group (via Checkpoint) with access_version = {}, settle_version = {}", @@ -213,6 +214,7 @@ impl Component for Broadcaster { checkpoint_off_chain: Some(checkpoint.derive_offchain_part()), }); } + EventFrame::InitState { access_version, settle_version, @@ -233,6 +235,7 @@ impl Component for Broadcaster { state_sha: "".into(), }); } + EventFrame::TxState { tx_state } => match tx_state { TxState::SettleSucceed { .. } => { let r = ctx.broadcast_tx.send(BroadcastFrame::TxState { tx_state }); @@ -256,6 +259,7 @@ impl Component for Broadcaster { } } }, + EventFrame::Broadcast { event, timestamp, @@ -274,7 +278,7 @@ impl Component for Broadcaster { } else { error!("{} Received event without checkpoint", env.log_prefix); } - // Keep 10 groups at most + // Keep 200 groups at most if event_backup_groups.len() > 200 { event_backup_groups.pop_front(); } @@ -291,6 +295,7 @@ impl Component for Broadcaster { debug!("{} Failed to broadcast event: {:?}", env.log_prefix, e); } } + EventFrame::Sync { new_servers, new_players, @@ -306,7 +311,11 @@ impl Component for Broadcaster { transactor_addr, }; - ctx.event_backup_groups.write().await.back_mut().map(|g| g.merge_sync(&sync)); + ctx.event_backup_groups + .write() + .await + .back_mut() + .map(|g| g.merge_sync(&sync)); let broadcast_frame = BroadcastFrame::Sync { sync }; let r = ctx.broadcast_tx.send(broadcast_frame); @@ -332,7 +341,11 @@ impl Component for Broadcaster { transactor_addr, }; - ctx.event_backup_groups.write().await.back_mut().map(|g| g.merge_sync(&sync)); + ctx.event_backup_groups + .write() + .await + .back_mut() + .map(|g| g.merge_sync(&sync)); let broadcast_frame = BroadcastFrame::Sync { sync }; let r = ctx.broadcast_tx.send(broadcast_frame); @@ -359,9 +372,87 @@ impl Component for Broadcaster { #[cfg(test)] mod tests { use super::*; - use race_core::types::PlayerJoin; + use race_core::{checkpoint::Checkpoint, types::PlayerJoin}; use race_test::prelude::*; + #[tokio::test(flavor = "multi_thread", worker_threads = 3)] + async fn test_event_reorder() { + // Test how the events are grouped when the checkpoint is delayed due to settle_lock. + // Assume we have the following case, where in event handler, things happen in this order: + // InitState, CustomEvent, Checkpoint, CustomEvent + // Due to the settle lock, the emitted frames are in this order: + // InitState, CustomEvent, CustomEvent, Checkpoint + + tracing_subscriber::fmt::init(); + + let mut alice = TestClient::player("alice"); + let mut bob = TestClient::player("bob"); + let game_account = TestGameAccountBuilder::default() + .add_player(&mut alice, 100) + .add_player(&mut bob, 100) + .build(); + + let (broadcaster, ctx) = Broadcaster::init(game_account.addr.clone(), 0); + let handle = broadcaster.start("", ctx); + + { + let event_frame = EventFrame::InitState { + access_version: 0, + settle_version: 1, + checkpoint: Checkpoint::default(), + }; + handle.send_unchecked(event_frame).await; + + let event_frame = EventFrame::Broadcast { + event: Event::Custom { + sender: 1, + raw: vec![], + }, + state_sha: "1".into(), + timestamp: 1, + }; + handle.send_unchecked(event_frame).await; + + let event_frame = EventFrame::Checkpoint { + checkpoint: Checkpoint::default(), + access_version: 1, + settle_version: 2, + state_sha: "2".into(), + }; + handle.send_unchecked(event_frame).await; + + let event_frame = EventFrame::Broadcast { + event: Event::Custom { + sender: 2, + raw: vec![], + }, + state_sha: "3".into(), + timestamp: 1, + }; + handle.send_unchecked(event_frame).await; + } + + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + + let broadcast_frame = broadcaster.get_backlogs(0).await; + assert_eq!( + broadcast_frame, + BroadcastFrame::Backlogs { + checkpoint_off_chain: Some(Checkpoint::default().derive_offchain_part()), + backlogs: Box::new(vec![BroadcastFrame::Sync { + sync: BroadcastSync { + new_players: vec![], + new_servers: vec![], + new_deposits: vec![], + transactor_addr: "".into(), + access_version: 1 + } + }]), + state_sha: "2".into(), + } + ); + } + #[tokio::test] async fn test_broadcast_event() { let mut alice = TestClient::player("alice"); diff --git a/transactor/src/component/connection.rs b/transactor/src/component/connection.rs index 9bb60169..e01468cd 100644 --- a/transactor/src/component/connection.rs +++ b/transactor/src/component/connection.rs @@ -20,17 +20,21 @@ use jsonrpsee::{ rpc_params, ws_client::{WsClient, WsClientBuilder}, }; -use race_core::error::{Error, Result}; use race_core::types::{BroadcastFrame, SubscribeEventParams}; +use race_core::{ + checkpoint::CheckpointOffChain, + error::{Error, Result}, + types::CheckpointParams, +}; use race_core::{ connection::ConnectionT, encryptor::EncryptorT, types::{AttachGameParams, ExitGameParams, SubmitEventParams}, }; -use crate::{frame::EventFrame, utils::current_timestamp}; use crate::utils::base64_decode; use crate::{component::common::Attachable, utils::base64_encode}; +use crate::{frame::EventFrame, utils::current_timestamp}; /// A connection to local event bus, for transactor loopback. #[allow(dead_code)] @@ -187,7 +191,7 @@ impl RemoteConnection { } Err(e) => { warn!("Error in request[{}]: {:?}", method, e); - }, + } } if retries < self.max_retries { @@ -199,6 +203,22 @@ impl RemoteConnection { } } + pub async fn get_checkpoint_off_chain( + &self, + game_addr: &str, + params: CheckpointParams, + ) -> Result> { + let req = self.make_request_no_sig(game_addr, ¶ms)?; + + match self.request::>>("get_checkpoint", req).await { + Ok(Some(res)) => { + let checkpoint_off_chain = CheckpointOffChain::try_from_slice(&res)?; + Ok(Some(checkpoint_off_chain)) + } + _ => Ok(None), + } + } + pub async fn subscribe_events( &self, game_addr: &str, diff --git a/transactor/src/component/event_bridge.rs b/transactor/src/component/event_bridge.rs index d3ccb9a4..e4e92a8f 100644 --- a/transactor/src/component/event_bridge.rs +++ b/transactor/src/component/event_bridge.rs @@ -1,8 +1,8 @@ use crate::frame::{EventFrame, SignalFrame}; use async_trait::async_trait; -use tokio::sync::{broadcast, mpsc}; -use tracing::{info, warn, log::error}; use race_api::types::GameId; +use tokio::sync::{broadcast, mpsc}; +use tracing::{info, log::error, warn}; use super::{common::PipelinePorts, CloseReason, Component, ComponentEnv}; @@ -45,9 +45,7 @@ impl EventBridgeParent { let (mpsc_tx, mpsc_rx) = mpsc::channel(10); let (bc_tx, _bc_rx) = broadcast::channel(10); ( - Self { - bc: bc_tx.clone(), - }, + Self { bc: bc_tx.clone() }, EventBridgeParentContext { tx: bc_tx, rx: mpsc_rx, @@ -96,7 +94,6 @@ impl Component for EventBridgeParent { mut ctx: EventBridgeParentContext, env: ComponentEnv, ) -> CloseReason { - // We save the pending events here. let mut pending_events: Vec<(GameId, EventFrame)> = Vec::with_capacity(10); @@ -105,15 +102,14 @@ impl Component for EventBridgeParent { while let Some((from_bridge, event_frame)) = Self::read_event(&mut ports, &mut ctx.rx).await { - if from_bridge { // Bridge parent receives event from bridge child + if from_bridge { + // Bridge parent receives event from bridge child match event_frame { EventFrame::SendBridgeEvent { from, dest, event, - access_version, - settle_version, - checkpoint_state, + versioned_data, } => { info!("{} Receives event: {}", env.log_prefix, event); ports @@ -121,9 +117,7 @@ impl Component for EventBridgeParent { from, dest, event, - access_version, - settle_version, - checkpoint_state, + versioned_data, }) .await; } @@ -148,9 +142,37 @@ impl Component for EventBridgeParent { ports.send(event_frame).await; } + + EventFrame::SubGameRecovered { game_id } => { + info!("{} Receives subgame recovered: {}", env.log_prefix, game_id); + // Remove the launching game's ID + launching_game_ids.retain(|id| *id != game_id); + // Send the pending events, for print log reason, not combine with SubGameReady + let mut i = 0; + while i < pending_events.len() { + if pending_events[i].0 == game_id { + let (_, event_frame) = pending_events.remove(i); + info!("{} Send pending event: {}", env.log_prefix, event_frame); + if let Err(e) = ctx.tx.send(event_frame) { + error!("{} Failed to send: {}", env.log_prefix, e); + } + } else { + i += 1; + } + } + + ports.send(event_frame).await; + } + + EventFrame::SubGameShutdown { game_id, .. } => { + info!("{} Receives subgame shutdown: {}", env.log_prefix, game_id); + ports.send(event_frame).await; + } + _ => (), } - } else { // Bridge parent receives event from event bus + } else { + // Bridge parent receives event from event bus match event_frame { EventFrame::LaunchSubGame { sub_game_init } => { let game_id = sub_game_init.spec.game_id; @@ -188,10 +210,19 @@ impl Component for EventBridgeParent { } } } - EventFrame::Sync { new_players, new_servers, transactor_addr, access_version, .. } => { + EventFrame::Sync { + new_players, + new_servers, + transactor_addr, + access_version, + .. + } => { if ctx.tx.receiver_count() > 0 { let sub_sync = EventFrame::SubSync { - new_players, new_servers, transactor_addr, access_version + new_players, + new_servers, + transactor_addr, + access_version, }; info!("{} Broadcast sync: {}", env.log_prefix, sub_sync); if let Err(e) = ctx.tx.send(sub_sync) { @@ -209,8 +240,10 @@ impl Component for EventBridgeParent { } impl EventBridgeChild { - - pub fn init(game_id: GameId, bridge_to_parent: BridgeToParent) -> (EventBridgeChild, EventBridgeChildContext) { + pub fn init( + game_id: GameId, + bridge_to_parent: BridgeToParent, + ) -> (EventBridgeChild, EventBridgeChildContext) { ( EventBridgeChild { game_id: game_id.clone(), @@ -262,7 +295,8 @@ impl Component for EventBridgeChild { ) -> CloseReason { while let Some((from_bridge, event_frame)) = Self::read_event(&mut ports, &mut ctx.rx).await { - if from_bridge { // Bridge child receives event from event parent + if from_bridge { + // Bridge child receives event from event parent match event_frame { EventFrame::Shutdown => { info!("{} Stopped", env.log_prefix); @@ -277,9 +311,9 @@ impl Component for EventBridgeChild { from, dest, event, - access_version, - settle_version, - checkpoint_state, + // access_version, + // settle_version, + versioned_data, } if dest == ctx.game_id => { info!("{} Receives {}", env.log_prefix, event); ports @@ -287,15 +321,16 @@ impl Component for EventBridgeChild { from, dest, event, - access_version, - settle_version, - checkpoint_state, + // access_version, + // settle_version, + versioned_data, }) .await; } _ => {} } - } else { // Bridge child receives event from event bus + } else { + // Bridge child receives event from event bus match event_frame { EventFrame::Shutdown => break, @@ -306,12 +341,27 @@ impl Component for EventBridgeChild { } } + EventFrame::SubGameRecovered { .. } => { + info!("{} Send SubGameRecovered to parent", env.log_prefix); + if let Err(e) = ctx.tx.send(event_frame).await { + error!("{} Failed to send: {}", env.log_prefix, e); + } + } + + EventFrame::SubGameShutdown { .. } => { + info!("{} Send SubGameShutdown to parent", env.log_prefix); + if let Err(e) = ctx.tx.send(event_frame).await { + error!("{} Failed to send: {}", env.log_prefix, e); + } + } + EventFrame::SendBridgeEvent { dest, .. } if dest != ctx.game_id => { info!("{} Sends event: {}", env.log_prefix, event_frame); if let Err(e) = ctx.tx.send(event_frame).await { error!("{} Failed to send: {}", env.log_prefix, e); } } + _ => continue, } } diff --git a/transactor/src/component/event_loop.rs b/transactor/src/component/event_loop.rs index d87bbc2d..e0e61884 100644 --- a/transactor/src/component/event_loop.rs +++ b/transactor/src/component/event_loop.rs @@ -7,18 +7,18 @@ use tracing::{error, info, warn}; use crate::component::common::{Component, PipelinePorts}; use crate::component::event_bus::CloseReason; -use crate::component::wrapped_handler::WrappedHandler; use crate::frame::EventFrame; use crate::utils::current_timestamp; use race_core::types::{ClientMode, GameMode, GamePlayer}; +use super::handler::HandlerT; use super::ComponentEnv; mod event_handler; mod misc; pub struct EventLoopContext { - handler: WrappedHandler, + handler: Box, game_context: GameContext, client_mode: ClientMode, game_mode: GameMode, @@ -56,13 +56,13 @@ impl Component for EventLoop { // An init account is created during the process and is passed to the init_state // function of the game handler. // - // Recover from chcekpoint. When the checkpoint is available, there's no need to call + // Recover from checkpoint. When the checkpoint is available, there's no need to call // init_state from the game handler. if !game_context.handler_is_initialized() { if let Some(close_reason) = event_handler::init_state( access_version, settle_version, - &mut handler, + &mut *handler, &mut game_context, &ports, ctx.client_mode, @@ -93,7 +93,7 @@ impl Component for EventLoop { if ctx.client_mode == ClientMode::Transactor { let event = Event::GameStart; if let Some(close_reason) = event_handler::handle_event( - &mut handler, + &mut *handler, &mut game_context, event, &ports, @@ -204,7 +204,7 @@ impl Component for EventLoop { if !players.is_empty() { let event = Event::Join { players }; if let Some(close_reason) = event_handler::handle_event( - &mut handler, + &mut *handler, &mut game_context, event, &ports, @@ -222,7 +222,7 @@ impl Component for EventLoop { if !deposits.is_empty() { let event = Event::Deposit { deposits }; if let Some(close_reason) = event_handler::handle_event( - &mut handler, + &mut *handler, &mut game_context, event, &ports, @@ -243,7 +243,7 @@ impl Component for EventLoop { if let Ok(player_id) = game_context.addr_to_id(&player_addr) { let event = Event::Leave { player_id }; if let Some(close_reason) = event_handler::handle_event( - &mut handler, + &mut *handler, &mut game_context, event, &ports, @@ -265,24 +265,32 @@ impl Component for EventLoop { } EventFrame::SubGameReady { - checkpoint_state, + versioned_data, game_id, init_data, max_players, } => { - if ctx.game_mode == GameMode::Main { + if ctx.game_mode == GameMode::Main && ctx.client_mode == ClientMode::Transactor + { info!("SubGameReady: Update checkpoint for sub game: {}", game_id); - if let Err(e) = game_context - .checkpoint_mut() - .init_versioned_data(checkpoint_state) + if let Err(e) = + game_context.handle_versioned_data(game_id, versioned_data, true) { - error!("{} Failed to init checkpoint data: {:?}", env.log_prefix, e); + error!( + "{} Failed in handling new sub game's versioned data: {:?}", + env.log_prefix, e + ); ports.send(EventFrame::Shutdown).await; + return CloseReason::Fault(e); } let timestamp = current_timestamp(); - let event = Event::SubGameReady { game_id, max_players, init_data }; + let event = Event::SubGameReady { + game_id, + max_players, + init_data, + }; if let Some(close_reason) = event_handler::handle_event( - &mut handler, + &mut *handler, &mut game_context, event, &ports, @@ -298,32 +306,58 @@ impl Component for EventLoop { } } + EventFrame::SubGameShutdown { + game_id, + versioned_data, + } => { + if ctx.game_mode == GameMode::Main + && ctx.client_mode == ClientMode::Transactor + && game_context.game_id() == 0 + { + info!( + "SubGameShutdown: Update checkpoint for sub game: {}", + game_id + ); + if let Err(e) = + game_context.handle_versioned_data(game_id, versioned_data, false) + { + error!( + "{} SubGameShutdown: Failed in handling new sub game's versioned data: {:?}", + env.log_prefix, e + ); + ports.send(EventFrame::Shutdown).await; + return CloseReason::Fault(e); + } + } + } + EventFrame::RecvBridgeEvent { event, dest, from, - checkpoint_state, - settle_version, + versioned_data, .. } => { - // In the case of parent, update the child game's - // checkpoint value. - + // In the case of parent, update the child game' checkpoint value. let timestamp = current_timestamp(); + let settle_version = versioned_data.versions.settle_version; if game_context.game_id() == 0 && dest == 0 && from != 0 && settle_version > 0 { info!("BridgeEvent: Update checkpoint for sub game: {}", from); - if let Err(e) = game_context - .checkpoint_mut() - .update_versioned_data(checkpoint_state) + if let Err(e) = + game_context.handle_versioned_data(from, versioned_data, false) { - error!("{} Failed to set checkpoint data: {:?}", env.log_prefix, e); + error!( + "{} Failed in handling new sub game's versioned data: {:?}", + env.log_prefix, e + ); ports.send(EventFrame::Shutdown).await; + return CloseReason::Fault(e); } } if let Some(close_reason) = event_handler::handle_event( - &mut handler, + &mut *handler, &mut game_context, event, &ports, @@ -339,7 +373,7 @@ impl Component for EventLoop { } EventFrame::SendEvent { event, timestamp } => { if let Some(close_reason) = event_handler::handle_event( - &mut handler, + &mut *handler, &mut game_context, event, &ports, @@ -360,7 +394,7 @@ impl Component for EventLoop { } if let Some(close_reason) = event_handler::handle_event( - &mut handler, + &mut *handler, &mut game_context, event, &ports, @@ -388,7 +422,7 @@ impl Component for EventLoop { impl EventLoop { pub fn init( - handler: WrappedHandler, + handler: Box, game_context: GameContext, client_mode: ClientMode, game_mode: GameMode, @@ -404,3 +438,75 @@ impl EventLoop { ) } } + +#[cfg(test)] +mod tests { + + use borsh::{BorshDeserialize, BorshSerialize}; + use race_api::{event::BridgeEvent, prelude::InitAccount}; + use race_core::error::Result; + use race_core::{checkpoint::VersionedData, context::EventEffects}; + + use super::*; + + #[derive(BorshSerialize, BorshDeserialize)] + struct EmptyBridgeEvent {} + + impl BridgeEvent for EmptyBridgeEvent {} + + struct TestHandlerForBridgeEvent {} + + impl HandlerT for TestHandlerForBridgeEvent { + fn init_state( + &mut self, + _context: &mut GameContext, + _init_account: &InitAccount, + ) -> Result { + Ok(EventEffects::default()) + } + + fn handle_event( + &mut self, + context: &mut GameContext, + _event: &Event, + ) -> Result { + let mut ef = context.derive_effect(false); + ef.checkpoint(); + ef.bridge_event(0, EmptyBridgeEvent {})?; + let ee = context.apply_effect(ef); + ee + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 10)] + async fn test_event_loop_receive_bridge_event() { + let handler = TestHandlerForBridgeEvent {}; + let game_context = GameContext::default(); + + let (event_loop, event_loop_ctx) = EventLoop::init( + Box::new(handler), + game_context, + ClientMode::Transactor, + GameMode::Main, + ); + let mut event_loop_handle = event_loop.start("fake addresses", event_loop_ctx); + let mut vd1 = VersionedData::default(); + vd1.id = 1; + event_loop_handle + .send_unchecked(EventFrame::RecvBridgeEvent { + from: 1, + dest: 0, + event: Event::Bridge { + dest_game_id: 0, + from_game_id: 1, + raw: vec![], + }, + versioned_data: vd1, + }) + .await; + println!("Sent!"); + let recv = event_loop_handle.recv_unchecked().await; + println!("{recv:?}"); + assert!(false); + } +} diff --git a/transactor/src/component/event_loop/event_handler.rs b/transactor/src/component/event_loop/event_handler.rs index ec917dbb..b725cf2c 100644 --- a/transactor/src/component/event_loop/event_handler.rs +++ b/transactor/src/component/event_loop/event_handler.rs @@ -2,16 +2,18 @@ use super::misc::log_execution_context; use race_api::{ effect::{EmitBridgeEvent, Log, SubGame}, event::Event, - types::{Award, EntryLock, Settle, Transfer}, + prelude::InitAccount, + types::GameId, }; use race_core::{ - context::{EventEffects, GameContext, SubGameInit, SubGameInitSource, Versions}, + checkpoint::{Checkpoint, VersionedData}, + context::{EventEffects, GameContext, SubGameInit, SubGameInitSource}, error::Error, types::{ClientMode, GameMode, GameSpec}, }; use crate::{ - component::{common::PipelinePorts, CloseReason, ComponentEnv, WrappedHandler}, + component::{common::PipelinePorts, handler::HandlerT, CloseReason, ComponentEnv}, frame::EventFrame, }; use tracing::{debug, error, info, warn}; @@ -29,11 +31,7 @@ fn print_logs(logs: &[Log], env: &ComponentEnv) { }) } -async fn broadcast_event( - event: Event, - game_context: &GameContext, - ports: &PipelinePorts, -) { +async fn broadcast_event(event: Event, game_context: &GameContext, ports: &PipelinePorts) { ports .send(EventFrame::Broadcast { event, @@ -43,6 +41,23 @@ async fn broadcast_event( .await; } +async fn send_checkpoint( + checkpoint: Option, + game_context: &GameContext, + ports: &PipelinePorts, +) { + if let Some(checkpoint) = checkpoint { + ports + .send(EventFrame::Checkpoint { + checkpoint, + access_version: game_context.access_version(), + settle_version: game_context.settle_version(), + state_sha: game_context.state_sha(), + }) + .await; + } +} + async fn update_local_client(game_context: &GameContext, ports: &PipelinePorts) { ports .send(EventFrame::ContextUpdated { @@ -59,6 +74,39 @@ async fn send_start_game(game_context: &GameContext, ports: &PipelinePorts) { .await; } +async fn send_subgame_ready( + versioned_data: VersionedData, + game_context: &GameContext, + init_account: InitAccount, + ports: &PipelinePorts, +) { + ports + .send(EventFrame::SubGameReady { + versioned_data, + game_id: game_context.game_id(), + max_players: game_context.max_players(), + init_data: init_account.data, + }) + .await; +} + +async fn send_subgame_recovered(game_id: GameId, ports: &PipelinePorts) { + ports.send(EventFrame::SubGameRecovered { game_id }).await; +} + +pub async fn send_subgame_shutdown( + game_id: GameId, + versioned_data: &VersionedData, + ports: &PipelinePorts, +) { + ports + .send(EventFrame::SubGameShutdown { + game_id, + versioned_data: versioned_data.clone(), + }) + .await; +} + async fn send_bridge_event( bridge_events: Vec, game_context: &GameContext, @@ -66,7 +114,12 @@ async fn send_bridge_event( env: &ComponentEnv, ) { for be in bridge_events { - info!("{} Send bridge event, dest: {}", env.log_prefix, be.dest); + info!( + "{} Send bridge event, dest: {}, from: {}", + env.log_prefix, + be.dest, + game_context.game_id() + ); let checkpoint_state = game_context .checkpoint() .get_versioned_data(game_context.game_id()); @@ -85,9 +138,7 @@ async fn send_bridge_event( from_game_id: game_context.game_id(), raw: be.raw, }, - access_version: game_context.access_version(), - settle_version: game_context.settle_version(), - checkpoint_state: checkpoint_state.clone(), + versioned_data: checkpoint_state.clone(), }; ports.send(ef).await; } @@ -98,49 +149,38 @@ async fn send_reject_deposits( ports: &PipelinePorts, env: &ComponentEnv, ) { - - info!("{} Send reject deposits, {:?}", env.log_prefix, reject_deposits); + info!( + "{} Send reject deposits, {:?}", + env.log_prefix, reject_deposits + ); let ef = EventFrame::RejectDeposits { reject_deposits }; ports.send(ef).await; } -async fn send_settlement( - transfer: Option, - settles: Vec, - awards: Vec, - entry_lock: Option, - original_versions: Versions, +async fn do_send_settlements( game_context: &mut GameContext, ports: &PipelinePorts, env: &ComponentEnv, ) { - let checkpoint = game_context.checkpoint().clone(); - let checkpoint_size = checkpoint.get_data(game_context.game_id()).map(|d| d.len()); - info!( - "{} Create checkpoint, settle_version: {}, size: {:?}", - env.log_prefix, - game_context.settle_version(), - checkpoint_size, - ); + while let Some(settle_details) = game_context.take_first_ready_settle_details() { + info!( + "{} Send settlement, settle_version: {}", + env.log_prefix, + game_context.settle_version(), + ); - let accept_deposits = game_context.take_accept_deposits(); + if game_context.game_id() == 0 { + settle_details.print("do_send_settlements".to_string()); + } - ports - .send(EventFrame::Checkpoint { - access_version: game_context.access_version(), - settle_version: game_context.settle_version(), - previous_settle_version: original_versions.settle_version, - checkpoint: checkpoint.clone(), - settles, - transfer, - awards, - state_sha: game_context.state_sha(), - entry_lock, - accept_deposits, - }) - .await; + ports + .send(EventFrame::Settle { + settle_details: Box::new(settle_details), + }) + .await; + } } async fn launch_sub_game( @@ -178,19 +218,14 @@ async fn launch_sub_game( pub async fn init_state( access_version: u64, settle_version: u64, - handler: &mut WrappedHandler, + handler: &mut dyn HandlerT, mut game_context: &mut GameContext, ports: &PipelinePorts, - client_mode: ClientMode, + _client_mode: ClientMode, game_mode: GameMode, env: &ComponentEnv, ) -> Option { - let init_account = match game_context.init_account() { - Ok(init_account) => init_account, - Err(e) => return Some(CloseReason::Fault(e)), - }; - - let original_versions = game_context.versions(); + let init_account = game_context.init_account(); let effects = match handler.init_state(&mut game_context, &init_account) { Ok(effects) => { @@ -205,10 +240,6 @@ pub async fn init_state( } }; - let EventEffects { - checkpoint, .. - } = effects; - info!( "{} Initialize game state, access_version: {}, settle_version: {}, SHA: {}", env.log_prefix, @@ -217,41 +248,21 @@ pub async fn init_state( game_context.state_sha() ); - let Some(checkpoint) = checkpoint else { + let Some(checkpoint) = effects.checkpoint else { ports.send(EventFrame::Shutdown).await; return Some(CloseReason::Fault(Error::CheckpointNotFoundAfterInit)); }; - send_settlement( - None, - vec![], - vec![], - None, - original_versions, - &mut game_context, - ports, - env, - ) - .await; - - // Dispatch the initial Ready event if running in Transactor mode. - if client_mode == ClientMode::Transactor { - game_context.dispatch_safe(Event::Ready, 0); - } + send_checkpoint(Some(checkpoint.clone()), game_context, ports).await; + + do_send_settlements(game_context, ports, env).await; // Tell master game the subgame is successfully created. if game_mode == GameMode::Sub { let game_id = game_context.game_id(); - let checkpoint_state = checkpoint.get_versioned_data(game_id); - if let Some(checkpoint_state) = checkpoint_state { - ports - .send(EventFrame::SubGameReady { - checkpoint_state: checkpoint_state.clone(), - game_id: game_context.game_id(), - max_players: game_context.max_players(), - init_data: init_account.data, - }) - .await; + if let Some(versioned_data) = checkpoint.get_versioned_data(game_id) { + println!("init sub game state"); + send_subgame_ready(versioned_data.clone(), game_context, init_account, ports).await; } else { ports.send(EventFrame::Shutdown).await; return Some(CloseReason::Fault(Error::CheckpointNotFoundAfterInit)); @@ -297,14 +308,45 @@ pub async fn recover_from_checkpoint( } if client_mode == ClientMode::Transactor { - game_context.dispatch_safe(Event::Ready, 0); + if game_mode == GameMode::Sub { + send_subgame_recovered(game_context.game_id(), ports).await; + } + + if game_mode == GameMode::Main { + launch_sub_game( + game_context.checkpoint().get_launch_subgames(), + game_context, + ports, + env, + ) + .await; + } + + if let Some(versioned_data) = game_context + .checkpoint() + .list_versioned_data() + .iter() + .find(|vd| vd.id == game_context.game_id()) + { + if !versioned_data.bridge_events.is_empty() { + send_bridge_event( + versioned_data.bridge_events.clone(), + game_context, + ports, + env, + ) + .await; + } + + game_context.set_dispatch(versioned_data.dispatch.clone()); + } } None } pub async fn handle_event( - handler: &mut WrappedHandler, + handler: &mut dyn HandlerT, game_context: &mut GameContext, event: Event, ports: &PipelinePorts, @@ -319,21 +361,17 @@ pub async fn handle_event( ); game_context.set_timestamp(timestamp); - let original_versions = game_context.versions(); match handler.handle_event(game_context, &event) { Ok(effects) => { let EventEffects { - settles, - transfer, - awards, - checkpoint, launch_sub_games, bridge_events, start_game, - entry_lock, + stop_game, logs, reject_deposits, + checkpoint, } = effects; print_logs(&logs, env); @@ -343,6 +381,8 @@ pub async fn handle_event( broadcast_event(event, &game_context, ports).await; } + send_checkpoint(checkpoint, game_context, ports).await; + // Update the local client update_local_client(&game_context, ports).await; @@ -351,6 +391,13 @@ pub async fn handle_event( send_start_game(&game_context, ports).await; } + if client_mode == ClientMode::Transactor && stop_game { + let game_id = game_context.game_id(); + if let Some(vd) = game_context.checkpoint().get_versioned_data(game_id) { + send_subgame_shutdown(game_id, vd, ports).await; + } + } + // Launch sub games if game_mode == GameMode::Main { launch_sub_game(launch_sub_games, game_context, ports, env).await; @@ -360,19 +407,7 @@ pub async fn handle_event( send_reject_deposits(reject_deposits, ports, env).await; } - if checkpoint.is_some() { - send_settlement( - transfer, - settles, - awards, - entry_lock, - original_versions, - game_context, - ports, - env, - ) - .await; - } + do_send_settlements(game_context, ports, env).await; // Emit bridge events if client_mode == ClientMode::Transactor { diff --git a/transactor/src/component/handler.rs b/transactor/src/component/handler.rs new file mode 100644 index 00000000..b06dd5a2 --- /dev/null +++ b/transactor/src/component/handler.rs @@ -0,0 +1,10 @@ +use race_api::prelude::InitAccount; +use race_core::context::{EventEffects, GameContext}; +use race_core::error::Result; +use race_api::event::Event; + +pub trait HandlerT: Send + Sync { + fn handle_event(&mut self, context: &mut GameContext, event: &Event) -> Result; + + fn init_state(&mut self, context: &mut GameContext, init_account: &InitAccount) -> Result; +} diff --git a/transactor/src/component/submitter.rs b/transactor/src/component/submitter.rs index ce24a5d3..e65f1e33 100644 --- a/transactor/src/component/submitter.rs +++ b/transactor/src/component/submitter.rs @@ -3,6 +3,7 @@ use std::time::Duration; use async_trait::async_trait; use race_api::types::{Settle, Transfer}; +use race_core::context::SettleDetails; use race_core::error::Error; use race_core::storage::StorageT; use race_core::types::{GameAccount, SaveCheckpointParams, SettleParams, SettleResult, TxState}; @@ -23,7 +24,7 @@ use super::ComponentEnv; const DEFAULT_SUBMITTER_SQUASH_TIME_WINDOW: u64 = 30; // The default for maximum number of transactions for one squash. -const DEFAULT_SUBMITTER_SQUASH_COMPLEXITY_LIMIT: usize = 50; +const DEFAULT_SUBMITTER_SQUASH_COMPLEXITY_LIMIT: usize = 1; // The default for size of transcation queue. const DEFAULT_SUBMITTER_TX_QUEUE_SIZE: usize = 100; @@ -50,7 +51,6 @@ fn merge_settle_with_same_player_id(a: &mut Settle, b: Settle) { } fn merge_settles(a: &mut Vec, mut b: Vec) { - for settle_from_a in a.iter_mut() { let (drained, v) = b .into_iter() @@ -242,18 +242,20 @@ impl Component for Submitter { while let Some(event) = ports.recv().await { match event { - EventFrame::Checkpoint { - settles, - transfer, - checkpoint, - awards, - access_version, - settle_version, - previous_settle_version, - entry_lock, - accept_deposits, - .. - } => { + EventFrame::Settle { settle_details } => { + let SettleDetails { + settles, + transfer, + awards, + checkpoint, + access_version, + settle_version, + previous_settle_version, + entry_lock, + accept_deposits, + .. + } = *settle_details; + let checkpoint_onchain = checkpoint.derive_onchain_part(); let checkpoint_offchain = checkpoint.derive_offchain_part(); @@ -350,18 +352,20 @@ mod tests { merge_settles(&mut settles1, settles2); assert_eq!( settles1, - vec![Settle { - player_id: 0, - withdraw: 400, - change: Some(race_api::types::BalanceChange::Add(100)), - eject: false - }, - Settle { - player_id: 1, - withdraw: 0, - change: Some(race_api::types::BalanceChange::Add(500)), - eject: false - }] + vec![ + Settle { + player_id: 0, + withdraw: 400, + change: Some(race_api::types::BalanceChange::Add(100)), + eject: false + }, + Settle { + player_id: 1, + withdraw: 0, + change: Some(race_api::types::BalanceChange::Add(500)), + eject: false + } + ] ) } } diff --git a/transactor/src/component/wrapped_client.rs b/transactor/src/component/wrapped_client.rs index d19a4e6e..4d6e0306 100644 --- a/transactor/src/component/wrapped_client.rs +++ b/transactor/src/component/wrapped_client.rs @@ -101,6 +101,7 @@ impl Component for WrappedClient { } } EventFrame::Checkpoint { .. } => { + info!("{} Flush secrets states", env.log_prefix); client.flush_secret_states(); } EventFrame::Shutdown => { diff --git a/transactor/src/component/wrapped_handler.rs b/transactor/src/component/wrapped_handler.rs index faa75142..82d25e3c 100644 --- a/transactor/src/component/wrapped_handler.rs +++ b/transactor/src/component/wrapped_handler.rs @@ -16,6 +16,8 @@ use tracing::info; use tracing::log::error; use wasmer::{imports, Instance, Module, Store, TypedFunction}; +use super::HandlerT; + fn log_execution_context(effect_bs: &Vec, event_bs: &Vec) { info!("Execution context"); info!("===== Effect Bytes ====="); @@ -31,6 +33,33 @@ pub struct WrappedHandler { encryptor: Arc, } +impl HandlerT for WrappedHandler { + fn handle_event( + &mut self, + context: &mut GameContext, + event: &Event, + ) -> Result { + let mut new_context = context.clone(); + general_handle_event(&mut new_context, event, self.encryptor.as_ref())?; + let event_effects = self.custom_handle_event(&mut new_context, event)?; + swap(context, &mut new_context); + Ok(event_effects) + } + + // Initialize game state + fn init_state( + &mut self, + context: &mut GameContext, + init_account: &InitAccount, + ) -> Result { + let mut new_context = context.clone(); + new_context.set_timestamp(0); + let event_effects = self.custom_init_state(&mut new_context, init_account)?; + swap(context, &mut new_context); + Ok(event_effects) + } +} + impl WrappedHandler { /// Load WASM bundle by game address pub async fn load_by_bundle( @@ -216,31 +245,6 @@ impl WrappedHandler { context.apply_effect(effect) } } - - pub fn handle_event( - &mut self, - context: &mut GameContext, - event: &Event, - ) -> Result { - let mut new_context = context.clone(); - general_handle_event(&mut new_context, event, self.encryptor.as_ref())?; - let event_effects = self.custom_handle_event(&mut new_context, event)?; - swap(context, &mut new_context); - Ok(event_effects) - } - - // Initialize game state - pub fn init_state( - &mut self, - context: &mut GameContext, - init_account: &InitAccount, - ) -> Result { - let mut new_context = context.clone(); - new_context.set_timestamp(0); - let event_effects = self.custom_init_state(&mut new_context, init_account)?; - swap(context, &mut new_context); - Ok(event_effects) - } } #[cfg(test)] @@ -286,12 +290,13 @@ mod tests { WrappedHandler::load_by_path(bundle_path).unwrap() } + #[ignore] #[test] fn test_init_state() { let mut hdlr = make_wrapped_handler(); let (game_account, _tx) = setup_game(); let mut ctx = GameContext::try_new(&game_account, None).unwrap(); - let init_account = ctx.init_account().unwrap(); + let init_account = ctx.init_account(); hdlr.init_state(&mut ctx, &init_account).unwrap(); assert_eq!( &vec![42u8, 0, 0, 0, 0, 0, 0, 0], @@ -299,13 +304,14 @@ mod tests { ); } + #[ignore] #[test] fn test_handle_event() { let mut hdlr = make_wrapped_handler(); let (game_account, _tx) = setup_game(); let mut ctx = GameContext::try_new(&game_account, None).unwrap(); let event = Event::GameStart; - let init_account = ctx.init_account().unwrap(); + let init_account = ctx.init_account(); hdlr.init_state(&mut ctx, &init_account).unwrap(); println!("ctx: {:?}", ctx); hdlr.handle_event(&mut ctx, &event).unwrap(); @@ -316,13 +322,14 @@ mod tests { assert_eq!(ctx.get_status(), GameStatus::Running); } + #[ignore] #[test] fn test_handle_custom_event() { let mut hdlr = make_wrapped_handler(); let (game_account, _tx) = setup_game(); let mut ctx = GameContext::try_new(&game_account, None).unwrap(); let event = Event::custom(0, &MinimalEvent::Increment(1)); - let init_account = ctx.init_account().unwrap(); + let init_account = ctx.init_account(); hdlr.init_state(&mut ctx, &init_account).unwrap(); println!("ctx: {:?}", ctx); hdlr.handle_event(&mut ctx, &event).unwrap(); diff --git a/transactor/src/frame.rs b/transactor/src/frame.rs index bececa69..63bd7ace 100644 --- a/transactor/src/frame.rs +++ b/transactor/src/frame.rs @@ -1,22 +1,30 @@ use race_api::{ - event::{Event, Message}, types::{Award, EntryLock, Settle, GameId} + event::{Event, Message}, + types::GameId, }; use race_core::{ - checkpoint::{Checkpoint, VersionedData}, context::{GameContext, SubGameInit}, types::{ClientMode, PlayerDeposit, PlayerJoin, ServerJoin, Transfer, TxState, VoteType} + checkpoint::{Checkpoint, VersionedData}, + context::{GameContext, SettleDetails, SubGameInit}, + types::{ClientMode, PlayerDeposit, PlayerJoin, ServerJoin, TxState, VoteType}, }; use crate::component::BridgeToParent; #[derive(Debug)] pub enum SignalFrame { - StartGame { game_addr: String, mode: ClientMode }, + StartGame { + game_addr: String, + mode: ClientMode, + }, LaunchSubGame { sub_game_init: SubGameInit, bridge_to_parent: BridgeToParent, }, #[allow(unused)] Shutdown, - RemoveGame { game_addr: String }, + RemoveGame { + game_addr: String, + }, } #[derive(Debug, Clone)] @@ -56,16 +64,13 @@ pub enum EventFrame { timestamp: u64, }, Checkpoint { - settles: Vec, - transfer: Option, - awards: Vec, checkpoint: Checkpoint, access_version: u64, settle_version: u64, - previous_settle_version: u64, state_sha: String, - entry_lock: Option, - accept_deposits: Vec, + }, + Settle { + settle_details: Box, }, Broadcast { event: Event, @@ -89,19 +94,14 @@ pub enum EventFrame { from: GameId, dest: GameId, event: Event, - access_version: u64, - settle_version: u64, - checkpoint_state: VersionedData, + versioned_data: VersionedData, }, /// Similar to `SendBridgeEvent`, but for receiver's event bus. RecvBridgeEvent { from: GameId, dest: GameId, event: Event, - #[allow(unused)] - access_version: u64, - settle_version: u64, - checkpoint_state: VersionedData, + versioned_data: VersionedData, }, /// Launch a subgame. @@ -120,15 +120,26 @@ pub enum EventFrame { /// Subgames send this frame after they made their first checkpoint. SubGameReady { game_id: GameId, - checkpoint_state: VersionedData, + versioned_data: VersionedData, max_players: u16, init_data: Vec, }, + /// Subgames send this frame when start via recovering from checkpoint. + SubGameRecovered { + game_id: GameId, + }, + + /// Subgame send this frame when it will be shutdown. + SubGameShutdown { + game_id: GameId, + versioned_data: VersionedData, + }, + /// Reject a deposit RejectDeposits { reject_deposits: Vec, - } + }, } impl std::fmt::Display for EventFrame { @@ -138,7 +149,11 @@ impl std::fmt::Display for EventFrame { EventFrame::GameStart { access_version } => { write!(f, "GameStart, access_version = {}", access_version) } - EventFrame::InitState { access_version, settle_version, .. } => write!( + EventFrame::InitState { + access_version, + settle_version, + .. + } => write!( f, "InitState, access_version = {}, settle_version = {}", access_version, settle_version @@ -165,6 +180,7 @@ impl std::fmt::Display for EventFrame { EventFrame::PlayerLeaving { .. } => write!(f, "PlayerLeaving"), EventFrame::SendEvent { event, .. } => write!(f, "SendEvent: {}", event), EventFrame::SendServerEvent { event, .. } => write!(f, "SendServerEvent: {}", event), + EventFrame::Settle { .. } => write!(f, "Settle"), EventFrame::Checkpoint { .. } => write!(f, "Checkpoint"), EventFrame::Broadcast { event, .. } => write!(f, "Broadcast: {}", event), EventFrame::SendMessage { message } => write!(f, "SendMessage: {}", message.sender), @@ -173,21 +189,40 @@ impl std::fmt::Display for EventFrame { EventFrame::Vote { votee, vote_type } => { write!(f, "Vote: to {} for {:?}", votee, vote_type) } - EventFrame::SendBridgeEvent { dest, event, settle_version, .. } => { - write!(f, "SendBridgeEvent: dest {}, settle_version: {}, event: {}", dest, settle_version, event) + EventFrame::SendBridgeEvent { dest, event, .. } => { + write!(f, "SendBridgeEvent: dest {}, event: {}", dest, event) } - EventFrame::RecvBridgeEvent { dest, event, settle_version, .. } => { - write!(f, "RecvBridgeEvent: dest {}, settle_version: {}, event: {}", dest, settle_version, event) + EventFrame::RecvBridgeEvent { dest, event, .. } => { + write!(f, "RecvBridgeEvent: dest {}, event: {}", dest, event) } EventFrame::LaunchSubGame { sub_game_init } => { - write!(f, "LaunchSubGame: {}#{}", sub_game_init.spec.game_addr, sub_game_init.spec.game_id) + write!( + f, + "LaunchSubGame: {}#{}", + sub_game_init.spec.game_addr, sub_game_init.spec.game_id + ) } - EventFrame::SubSync { new_players, new_servers, .. } => { - write!(f, "SyncNodes: new_players: {}, new_servers: {}", new_players.len(), new_servers.len()) + EventFrame::SubSync { + new_players, + new_servers, + .. + } => { + write!( + f, + "SyncNodes: new_players: {}, new_servers: {}", + new_players.len(), + new_servers.len() + ) } EventFrame::SubGameReady { game_id, .. } => { write!(f, "SubGameReady, game_id: {}", game_id) } + EventFrame::SubGameRecovered { game_id } => { + write!(f, "SubGameRecovered, game_id: {}", game_id) + } + EventFrame::SubGameShutdown { game_id, .. } => { + write!(f, "SubGameShutdown, game_id: {}", game_id) + } EventFrame::RejectDeposits { reject_deposits } => { write!(f, "Reject deposits, {:?}", reject_deposits) } diff --git a/transactor/src/handle/subgame.rs b/transactor/src/handle/subgame.rs index c574b6df..e156c46c 100644 --- a/transactor/src/handle/subgame.rs +++ b/transactor/src/handle/subgame.rs @@ -47,7 +47,7 @@ impl SubGameHandle { let settle_version = game_context.settle_version(); let checkpoint = game_context.checkpoint().clone(); - let handler = WrappedHandler::load_by_bundle(&bundle_account, encryptor.clone()).await?; + let handler = Box::new(WrappedHandler::load_by_bundle(&bundle_account, encryptor.clone()).await?); let (broadcaster, broadcaster_ctx) = Broadcaster::init(addr.clone(), game_id); let mut broadcaster_handle = broadcaster.start(&addr, broadcaster_ctx); diff --git a/transactor/src/handle/transactor.rs b/transactor/src/handle/transactor.rs index 14f68377..7ee50d79 100644 --- a/transactor/src/handle/transactor.rs +++ b/transactor/src/handle/transactor.rs @@ -104,7 +104,7 @@ impl TransactorHandle { return Err(Error::GameBundleNotFound); }; - let handler = WrappedHandler::load_by_bundle(&bundle_account, encryptor.clone()).await?; + let handler = Box::new(WrappedHandler::load_by_bundle(&bundle_account, encryptor.clone()).await?); let event_bus = EventBus::new(game_account.addr.clone()); diff --git a/transactor/src/handle/validator.rs b/transactor/src/handle/validator.rs index 267a0c90..aaf7a9bf 100644 --- a/transactor/src/handle/validator.rs +++ b/transactor/src/handle/validator.rs @@ -5,11 +5,11 @@ use crate::component::{ Voter, WrappedClient, WrappedHandler, }; use crate::frame::{EventFrame, SignalFrame}; -use race_core::error::{Error, Result}; use race_core::context::GameContext; +use race_core::error::{Error, Result}; use race_core::storage::StorageT; use race_core::transport::TransportT; -use race_core::types::{ClientMode, GameMode, GetCheckpointParams, ServerAccount}; +use race_core::types::{CheckpointParams, ClientMode, GameMode, ServerAccount}; use race_encryptor::Encryptor; use race_env::TransactorConfig; use tokio::sync::mpsc; @@ -29,34 +29,15 @@ impl ValidatorHandle { server_account: &ServerAccount, encryptor: Arc, transport: Arc, - storage: Arc, + _storage: Arc, signal_tx: mpsc::Sender, _config: &TransactorConfig, ) -> Result { - info!( - "Start game handle for {} with Validator mode", - game_addr, - ); + info!("Start game handle for {} with Validator mode", game_addr,); let Some(game_account) = transport.get_game_account(game_addr).await? else { return Err(Error::GameAccountNotFound); }; - let checkpoint_off_chain = storage - .get_checkpoint(GetCheckpointParams { - game_addr: game_addr.to_owned(), - settle_version: game_account.settle_version, - }) - .await?; - - let game_context = GameContext::try_new(&game_account, checkpoint_off_chain)?; - let checkpoint = game_context.checkpoint().clone(); - - let Some(bundle_account) = transport.get_game_bundle(&game_account.bundle_addr).await? else { - return Err(Error::GameBundleNotFound); - }; - - let handler = WrappedHandler::load_by_bundle(&bundle_account, encryptor.clone()).await?; - let transactor_addr = game_account .transactor_addr .as_ref() @@ -66,6 +47,39 @@ impl ValidatorHandle { .await? .ok_or(Error::CantFindTransactor)?; + let connection = Arc::new( + RemoteConnection::try_new( + &server_account.addr, + &transactor_account.endpoint, + encryptor.clone(), + ) + .await?, + ); + + let checkpoint_off_chain = connection.get_checkpoint_off_chain( + game_addr, + CheckpointParams { + settle_version: game_account.settle_version, + }, + ).await?; + + // let checkpoint_off_chain = storage + // .get_checkpoint(GetCheckpointParams { + // game_addr: game_addr.to_owned(), + // settle_version: game_account.settle_version, + // }) + // .await?; + + let game_context = GameContext::try_new(&game_account, checkpoint_off_chain)?; + let checkpoint = game_context.checkpoint().clone(); + + let Some(bundle_account) = transport.get_game_bundle(&game_account.bundle_addr).await? + else { + return Err(Error::GameBundleNotFound); + }; + + let handler = Box::new(WrappedHandler::load_by_bundle(&bundle_account, encryptor.clone()).await?); + info!("Creating components"); let event_bus = EventBus::new(game_account.addr.clone()); @@ -76,14 +90,6 @@ impl ValidatorHandle { EventLoop::init(handler, game_context, ClientMode::Validator, GameMode::Main); let mut event_loop_handle = event_loop.start(&game_account.addr, event_loop_ctx); - let connection = Arc::new( - RemoteConnection::try_new( - &server_account.addr, - &transactor_account.endpoint, - encryptor.clone(), - ) - .await?, - ); let (subscriber, subscriber_context) = Subscriber::init(&game_account, server_account, connection.clone()); let mut subscriber_handle = subscriber.start(&game_account.addr, subscriber_context);