#[cfg(feature = "persistent_world")] use crate::TerrainPersistence; use crate::{client::Client, Settings}; use common::{ comp::{ Admin, AdminRole, CanBuild, ControlEvent, Controller, ForceUpdate, Health, Ori, Player, Pos, Presence, PresenceKind, SkillSet, Vel, }, event::{EventBus, ServerEvent}, link::Is, mounting::{Rider, VolumeRider}, resources::{DeltaTime, PlayerPhysicsSetting, PlayerPhysicsSettings}, slowjob::SlowJobPool, terrain::TerrainGrid, vol::ReadVol, }; use common_ecs::{Job, Origin, Phase, System}; use common_net::msg::{ClientGeneral, ServerGeneral}; use common_state::{AreasContainer, BlockChange, BuildArea}; use core::mem; use rayon::prelude::*; use specs::{Entities, Join, LendJoin, Read, ReadExpect, ReadStorage, Write, WriteStorage}; use std::{borrow::Cow, time::Instant}; use tracing::{debug, trace, warn}; use vek::*; #[cfg(feature = "persistent_world")] pub type TerrainPersistenceData<'a> = Option>; #[cfg(not(feature = "persistent_world"))] pub type TerrainPersistenceData<'a> = core::marker::PhantomData<&'a mut ()>; // NOTE: These writes are considered "rare", meaning (currently) that they are // admin-gated features that players shouldn't normally access, and which we're // not that concerned about the performance of when two players try to use them // at once. // // In such cases, we're okay putting them behind a mutex and penalizing the // system if they're actually used concurrently by lots of users. Please do not // put less rare writes here, unless you want to serialize the system! struct RareWrites<'a, 'b> { block_changes: &'b mut BlockChange, _terrain_persistence: &'b mut TerrainPersistenceData<'a>, } impl Sys { #[allow(clippy::too_many_arguments)] fn handle_client_in_game_msg( server_emitter: &mut common::event::Emitter<'_, ServerEvent>, entity: specs::Entity, client: &Client, maybe_presence: &mut Option<&mut Presence>, terrain: &ReadExpect<'_, TerrainGrid>, can_build: &ReadStorage<'_, CanBuild>, is_rider: &ReadStorage<'_, Is>, is_volume_rider: &ReadStorage<'_, Is>, force_update: Option<&&mut ForceUpdate>, skill_set: &mut Option>, healths: &ReadStorage<'_, Health>, rare_writes: &parking_lot::Mutex>, position: Option<&mut Pos>, controller: Option<&mut Controller>, settings: &Read<'_, Settings>, build_areas: &Read<'_, AreasContainer>, player_physics_setting: Option<&mut PlayerPhysicsSetting>, maybe_admin: &Option<&Admin>, time_for_vd_changes: Instant, msg: ClientGeneral, player_physics: &mut Option<(Pos, Vel, Ori)>, ) -> Result<(), crate::error::Error> { let presence = match maybe_presence.as_deref_mut() { Some(g) => g, None => { debug!(?entity, "client is not in_game, ignoring msg"); trace!(?msg, "ignored msg content"); return Ok(()); }, }; match msg { // Go back to registered state (char selection screen) ClientGeneral::ExitInGame => { server_emitter.emit(ServerEvent::ExitIngame { entity }); client.send(ServerGeneral::ExitInGameSuccess)?; *maybe_presence = None; }, ClientGeneral::SetViewDistance(view_distances) => { let clamped_vds = view_distances.clamp(settings.max_view_distance); presence.terrain_view_distance.set_target(clamped_vds.terrain, time_for_vd_changes); presence.entity_view_distance.set_target(clamped_vds.entity, time_for_vd_changes); // Correct client if its requested VD is too high. if view_distances.terrain != clamped_vds.terrain { client.send(ServerGeneral::SetViewDistance(clamped_vds.terrain))?; } }, ClientGeneral::ControllerInputs(inputs) => { if presence.kind.controlling_char() { if let Some(controller) = controller { controller.inputs.update_with_new(*inputs); } } }, ClientGeneral::ControlEvent(event) => { if presence.kind.controlling_char() && let Some(controller) = controller { // Skip respawn if client entity is alive let skip_respawn = matches!(event, ControlEvent::Respawn) && healths.get(entity).map_or(true, |h| !h.is_dead); if !skip_respawn { controller.push_event(event); } } }, ClientGeneral::ControlAction(event) => { if presence.kind.controlling_char() { if let Some(controller) = controller { controller.push_action(event); } } }, ClientGeneral::PlayerPhysics { pos, vel, ori, force_counter } => { if presence.kind.controlling_char() && force_update.map_or(true, |force_update| force_update.counter() == force_counter) && healths.get(entity).map_or(true, |h| !h.is_dead) && is_rider.get(entity).is_none() && is_volume_rider.get(entity).is_none() && player_physics_setting .as_ref() .map_or(true, |s| s.client_authoritative()) { *player_physics = Some((pos, vel, ori)); } }, ClientGeneral::BreakBlock(pos) => { if let Some(comp_can_build) = can_build.get(entity) { if comp_can_build.enabled { for area in comp_can_build.build_areas.iter() { if let Some(old_block) = build_areas .areas() .get(*area) // TODO: Make this an exclusive check on the upper bound of the AABB // Vek defaults to inclusive which is not optimal .filter(|aabb| aabb.contains_point(pos)) .and_then(|_| terrain.get(pos).ok()) { let new_block = old_block.into_vacant(); // Take the rare writes lock as briefly as possible. let mut guard = rare_writes.lock(); let _was_set = guard.block_changes.try_set(pos, new_block).is_some(); #[cfg(feature = "persistent_world")] if _was_set { if let Some(terrain_persistence) = guard._terrain_persistence.as_mut() { terrain_persistence.set_block(pos, new_block); } } } } } } }, ClientGeneral::PlaceBlock(pos, new_block) => { if let Some(comp_can_build) = can_build.get(entity) { if comp_can_build.enabled { for area in comp_can_build.build_areas.iter() { if build_areas .areas() .get(*area) // TODO: Make this an exclusive check on the upper bound of the AABB // Vek defaults to inclusive which is not optimal .filter(|aabb| aabb.contains_point(pos)) .is_some() { // Take the rare writes lock as briefly as possible. let mut guard = rare_writes.lock(); let _was_set = guard.block_changes.try_set(pos, new_block).is_some(); #[cfg(feature = "persistent_world")] if _was_set { if let Some(terrain_persistence) = guard._terrain_persistence.as_mut() { terrain_persistence.set_block(pos, new_block); } } } } } } }, ClientGeneral::UnlockSkill(skill) => { // FIXME: How do we want to handle the error? Probably not by swallowing it. let _ = skill_set.as_mut().map(|skill_set| { SkillSet::unlock_skill_cow(skill_set, skill, |skill_set| skill_set.to_mut()) }).transpose(); }, ClientGeneral::RequestSiteInfo(id) => { server_emitter.emit(ServerEvent::RequestSiteInfo { entity, id }); }, ClientGeneral::RequestPlayerPhysics { server_authoritative, } => { if let Some(setting) = player_physics_setting { setting.client_optin = server_authoritative; } }, ClientGeneral::RequestLossyTerrainCompression { lossy_terrain_compression, } => { presence.lossy_terrain_compression = lossy_terrain_compression; }, ClientGeneral::UpdateMapMarker(update) => { server_emitter.emit(ServerEvent::UpdateMapMarker { entity, update }); }, ClientGeneral::SpectatePosition(pos) => { if let Some(admin) = maybe_admin && admin.0 >= AdminRole::Moderator && presence.kind == PresenceKind::Spectator { if let Some(position) = position { position.0 = pos; } } }, ClientGeneral::RequestCharacterList | ClientGeneral::CreateCharacter { .. } | ClientGeneral::EditCharacter { .. } | ClientGeneral::DeleteCharacter(_) | ClientGeneral::Character(_, _) | ClientGeneral::Spectate(_) | ClientGeneral::TerrainChunkRequest { .. } | ClientGeneral::LodZoneRequest { .. } | ClientGeneral::ChatMsg(_) | ClientGeneral::Command(..) | ClientGeneral::Terminate => { debug!("Kicking possibly misbehaving client due to invalid client in game request"); server_emitter.emit(ServerEvent::ClientDisconnect( entity, common::comp::DisconnectReason::NetworkError, )); }, } Ok(()) } } /// This system will handle new messages from clients #[derive(Default)] pub struct Sys; impl<'a> System<'a> for Sys { #[allow(clippy::type_complexity)] type SystemData = ( Entities<'a>, Read<'a, EventBus>, ReadExpect<'a, TerrainGrid>, ReadExpect<'a, SlowJobPool>, ReadStorage<'a, CanBuild>, WriteStorage<'a, ForceUpdate>, ReadStorage<'a, Is>, ReadStorage<'a, Is>, WriteStorage<'a, SkillSet>, ReadStorage<'a, Health>, Write<'a, BlockChange>, WriteStorage<'a, Pos>, WriteStorage<'a, Vel>, WriteStorage<'a, Ori>, WriteStorage<'a, Presence>, WriteStorage<'a, Client>, WriteStorage<'a, Controller>, Read<'a, DeltaTime>, Read<'a, Settings>, Read<'a, AreasContainer>, Write<'a, PlayerPhysicsSettings>, TerrainPersistenceData<'a>, ReadStorage<'a, Player>, ReadStorage<'a, Admin>, ); const NAME: &'static str = "msg::in_game"; const ORIGIN: Origin = Origin::Server; const PHASE: Phase = Phase::Create; fn run( _job: &mut Job, ( entities, server_event_bus, terrain, slow_jobs, can_build, mut force_updates, is_rider, is_volume_rider, mut skill_sets, healths, mut block_changes, mut positions, mut velocities, mut orientations, mut presences, mut clients, mut controllers, dt, settings, build_areas, mut player_physics_settings_, mut terrain_persistence, players, admins, ): Self::SystemData, ) { let time_for_vd_changes = Instant::now(); // NOTE: stdlib mutex is more than good enough on Linux and (probably) Windows, // but not Mac. let rare_writes = parking_lot::Mutex::new(RareWrites { block_changes: &mut block_changes, _terrain_persistence: &mut terrain_persistence, }); let player_physics_settings = &*player_physics_settings_; let mut deferred_updates = ( &entities, &mut clients, (&mut presences).maybe(), players.maybe(), admins.maybe(), (&skill_sets).maybe(), (&mut positions).maybe(), (&mut velocities).maybe(), (&mut orientations).maybe(), (&mut controllers).maybe(), (&mut force_updates).maybe(), ) .join() // NOTE: Required because Specs has very poor work splitting for sparse joins. .par_bridge() .map_init( || server_event_bus.emitter(), |server_emitter, ( entity, client, mut maybe_presence, maybe_player, maybe_admin, skill_set, ref mut pos, ref mut vel, ref mut ori, ref mut controller, ref mut force_update, )| { let old_player_physics_setting = maybe_player.map(|p| { player_physics_settings .settings .get(&p.uuid()) .copied() .unwrap_or_default() }); let mut new_player_physics_setting = old_player_physics_setting; // If an `ExitInGame` message is received this is set to `None` allowing further // ingame messages to be ignored. let mut clearable_maybe_presence = maybe_presence.as_deref_mut(); let mut skill_set = skill_set.map(Cow::Borrowed); let mut player_physics = None; let _ = super::try_recv_all(client, 2, |client, msg| { Self::handle_client_in_game_msg( server_emitter, entity, client, &mut clearable_maybe_presence, &terrain, &can_build, &is_rider, &is_volume_rider, force_update.as_ref(), &mut skill_set, &healths, &rare_writes, pos.as_deref_mut(), controller.as_deref_mut(), &settings, &build_areas, new_player_physics_setting.as_mut(), &maybe_admin, time_for_vd_changes, msg, &mut player_physics, ) }); if let Some((new_pos, new_vel, new_ori)) = player_physics && let Some(old_pos) = pos.as_deref_mut() && let Some(old_vel) = vel.as_deref_mut() && let Some(old_ori) = ori.as_deref_mut() { enum Rejection { TooFar { old: Vec3, new: Vec3 }, TooFast { vel: Vec3 }, } let rejection = if maybe_admin.is_some() { None } else { // Reminder: review these frequently to ensure they're reasonable const MAX_H_VELOCITY: f32 = 75.0; const MAX_V_VELOCITY: std::ops::Range = -100.0..80.0; 'rejection: { let is_velocity_ok = new_vel.0.xy().magnitude_squared() < MAX_H_VELOCITY.powi(2) && MAX_V_VELOCITY.contains(&new_vel.0.z); if !is_velocity_ok { break 'rejection Some(Rejection::TooFast { vel: new_vel.0 }); } // How far the player is permitted to stray from the correct position (perhaps due to // latency problems). const POSITION_THRESHOLD: f32 = 16.0; // The position can either be sensible with respect to either the old or the new // velocity such that we don't punish for edge cases after a sudden change let is_position_ok = [old_vel.0, new_vel.0] .into_iter() .any(|ref_vel| { let rpos = new_pos.0 - old_pos.0; // Determine whether the change in position is broadly consistent with both // the magnitude and direction of the velocity, with appropriate thresholds. LineSegment3 { start: Vec3::zero(), end: ref_vel * dt.0, } .projected_point(rpos) // + 1.5 accounts for minor changes in position without corresponding // velocity like block hopping/snapping .distance_squared(rpos) < (rpos.magnitude() * 0.5 + 1.5 + POSITION_THRESHOLD).powi(2) }); if !is_position_ok { break 'rejection Some(Rejection::TooFar { old: old_pos.0, new: new_pos.0 }); } None } }; if let Some(rejection) = rejection { // TODO: Log when false positives aren't generated often let alias = maybe_player.map(|p| &p.alias); match rejection { Rejection::TooFar { old, new } => warn!("Rejected physics for player {alias:?} (new position {new:?} is too far from old position {old:?})"), Rejection::TooFast { vel } => warn!("Rejected physics for player {alias:?} (new velocity {vel:?} is too fast)"), } /* // Perhaps this is overzealous? if let Some(mut setting) = new_player_physics_setting.as_mut() { setting.server_force = true; warn!("Switching player {alias:?} to server-side physics"); } */ // Reject the change and force the server's view of the physics state force_update.as_mut().map(|fu| fu.update()); } else { *old_pos = new_pos; *old_vel = new_vel; *old_ori = new_ori; } } // Ensure deferred view distance changes are applied (if the // requsite time has elapsed). if let Some(presence) = maybe_presence { presence.terrain_view_distance.update(time_for_vd_changes); presence.entity_view_distance.update(time_for_vd_changes); } // Return the possibly modified skill set, and possibly modified server physics // settings. let skill_set_update = skill_set.and_then(|skill_set| match skill_set { Cow::Borrowed(_) => None, Cow::Owned(skill_set) => Some((entity, skill_set)), }); // NOTE: Since we pass Option<&mut _> rather than &mut Option<_> to // handle_client_in_game_msg, and the new player was initialized to the same // value as the old setting , we know that either both the new and old setting // are Some, or they are both None. let physics_update = maybe_player.map(|p| p.uuid()) .zip(new_player_physics_setting .filter(|_| old_player_physics_setting != new_player_physics_setting)); (skill_set_update, physics_update) }, ) // NOTE: Would be nice to combine this with the map_init somehow, but I'm not sure if // that's possible. .filter(|(x, y)| x.is_some() || y.is_some()) // NOTE: I feel like we shouldn't actually need to allocate here, but hopefully this // doesn't turn out to be important as there shouldn't be that many connected clients. // The reason we can't just use unzip is that the two sides might be different lengths. .collect::>(); let player_physics_settings = &mut *player_physics_settings_; // Deferred updates to skillsets and player physics. // // NOTE: It is an invariant that there is at most one client entry per player // uuid; since we joined on clients, it follows that there's just one update // per uuid, so the physics update is sound and doesn't depend on evaluation // order, even though we're not updating directly by entity or uid (note that // for a given entity, we process messages serially). deferred_updates .iter_mut() .for_each(|(skill_set_update, physics_update)| { if let Some((entity, new_skill_set)) = skill_set_update { // We know this exists, because we already iterated over it with the skillset // lock taken, so we can ignore the error. // // Note that we replace rather than just updating. This is in order to avoid // dropping here; we'll drop later on a background thread, in case skillsets are // slow to drop. skill_sets .get_mut(*entity) .map(|mut old_skill_set| mem::swap(&mut *old_skill_set, new_skill_set)); } if let &mut Some((uuid, player_physics_setting)) = physics_update { // We don't necessarily know this exists, but that's fine, because dropping // player physics is a no op. player_physics_settings .settings .insert(uuid, player_physics_setting); } }); // Finally, drop the deferred updates in another thread. slow_jobs.spawn("CHUNK_DROP", move || { drop(deferred_updates); }); } }