From d9312141f6e8d46cf5fb64458fb87bb435292aee Mon Sep 17 00:00:00 2001 From: Alexander Medvedev Date: Sun, 24 May 2026 11:45:40 +0200 Subject: [PATCH] chore(net): don't run keep alive on player tick --- .github/workflows/docker.yml | 1 + Cargo.toml | 2 +- .../src/java/server/config/keep_alive.rs | 9 ++ .../src/java/server/config/mod.rs | 2 + pumpkin/src/entity/player.rs | 49 ++-------- pumpkin/src/net/java/config.rs | 23 ++++- pumpkin/src/net/java/mod.rs | 91 +++++++++++++++---- pumpkin/src/net/java/play.rs | 13 +-- 8 files changed, 121 insertions(+), 69 deletions(-) create mode 100644 pumpkin-protocol/src/java/server/config/keep_alive.rs diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index d2f14d216..6f31695e1 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -65,6 +65,7 @@ jobs: id: build uses: docker/build-push-action@v7 with: + context: . platforms: ${{ matrix.platform }} labels: ${{ steps.meta.outputs.labels }} tags: ${{ env.REGISTRY }}/${{ env.REGISTRY_IMAGE }} diff --git a/Cargo.toml b/Cargo.toml index 0b0432bac..f62fc603f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -190,7 +190,7 @@ rustc_version_runtime = { version = "0.3", default-features = false } ordered-float = { version = "5.3", default-features = false, features = ["std"] } xxhash-rust = { version = "0.8", default-features = false, features = ["xxh64"] } -wasmtime = { version = "45.0", default-features = false, features = ["runtime", "component-model", "async", "cache", "cranelift", "gc", "threads", "std"] } +wasmtime = { version = "45.0", default-features = false, features = ["runtime", "component-model", "async", "cache", "cranelift", "gc", "gc-drc", "threads", "std"] } wasmtime-wasi = { version = "45.0", default-features = false, features = ["p2"] } wasmtime-wasi-http = { version = "45.0", default-features = false, features = ["p2", "default-send-request"] } # needed for interacting with wasmtime-wasi-http - keep in sync diff --git a/pumpkin-protocol/src/java/server/config/keep_alive.rs b/pumpkin-protocol/src/java/server/config/keep_alive.rs new file mode 100644 index 000000000..d5472b957 --- /dev/null +++ b/pumpkin-protocol/src/java/server/config/keep_alive.rs @@ -0,0 +1,9 @@ +use pumpkin_data::packet::serverbound::CONFIG_KEEP_ALIVE; +use pumpkin_macros::java_packet; +use serde::{Deserialize, Serialize}; + +#[derive(Deserialize, Serialize)] +#[java_packet(CONFIG_KEEP_ALIVE)] +pub struct SKeepAlive { + pub keep_alive_id: i64, +} diff --git a/pumpkin-protocol/src/java/server/config/mod.rs b/pumpkin-protocol/src/java/server/config/mod.rs index bcd543a7f..552a7a436 100644 --- a/pumpkin-protocol/src/java/server/config/mod.rs +++ b/pumpkin-protocol/src/java/server/config/mod.rs @@ -2,6 +2,7 @@ mod acknowledge_finish_config; mod client_information; mod cookie_response; mod custom_click_action; +mod keep_alive; mod known_packs; mod plugin_message; mod resource_pack_response; @@ -10,6 +11,7 @@ pub use acknowledge_finish_config::*; pub use client_information::*; pub use cookie_response::*; pub use custom_click_action::*; +pub use keep_alive::*; pub use known_packs::*; pub use plugin_message::*; pub use resource_pack_response::*; diff --git a/pumpkin/src/entity/player.rs b/pumpkin/src/entity/player.rs index 816967298..c4623bf65 100644 --- a/pumpkin/src/entity/player.rs +++ b/pumpkin/src/entity/player.rs @@ -4,7 +4,7 @@ use std::f64::consts::TAU; use std::mem; use std::num::NonZeroU8; use std::str::FromStr; -use std::sync::atomic::{AtomicBool, AtomicI32, AtomicI64, AtomicU8, AtomicU32, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU8, AtomicU32, Ordering}; use std::sync::{Arc, Weak}; use std::time::{Duration, Instant}; @@ -62,12 +62,12 @@ use pumpkin_protocol::java::client::play::{ Animation, CAcknowledgeBlockChange, CActionBar, CChangeDifficulty, CChunkBatchEnd, CChunkBatchStart, CChunkData, CCloseContainer, CCombatDeath, CCustomPayload, CDisguisedChatMessage, CEntityAnimation, CEntityPositionSync, CGameEvent, CItemCooldown, - CKeepAlive, CMapItemData, COpenScreen, CParticle, CPlayerAbilities, CPlayerInfoUpdate, - CPlayerPosition, CPlayerSpawnPosition, CRespawn, CSetContainerContent, CSetContainerProperty, - CSetContainerSlot, CSetCursorItem, CSetEquipment, CSetExperience, CSetHealth, - CSetPlayerInventory, CSetSelectedSlot, CSoundEffect, CStopSound, CSubtitle, CSystemChatMessage, - CTabList, CTitleAnimation, CTitleText, CUnloadChunk, CUpdateMobEffect, CUpdateTime, GameEvent, - MapIcon, MapPatch, Metadata, PlayerAction, PlayerInfoFlags, PreviousMessage, + CMapItemData, COpenScreen, CParticle, CPlayerAbilities, CPlayerInfoUpdate, CPlayerPosition, + CPlayerSpawnPosition, CRespawn, CSetContainerContent, CSetContainerProperty, CSetContainerSlot, + CSetCursorItem, CSetEquipment, CSetExperience, CSetHealth, CSetPlayerInventory, + CSetSelectedSlot, CSoundEffect, CStopSound, CSubtitle, CSystemChatMessage, CTabList, + CTitleAnimation, CTitleText, CUnloadChunk, CUpdateMobEffect, CUpdateTime, GameEvent, MapIcon, + MapPatch, Metadata, PlayerAction, PlayerInfoFlags, PreviousMessage, }; use pumpkin_protocol::java::server::play::{ SClickSlot, SContainerButtonClick, SRenameItem, SlotActionType, @@ -445,12 +445,6 @@ pub struct Player { pub awaiting_teleport: Mutex)>>, /// The coordinates of the chunk section the player is currently watching. pub watched_section: AtomicCell, - /// Whether we are waiting for a response after sending a keep alive packet. - pub wait_for_keep_alive: AtomicBool, - /// The keep alive packet payload we send. The client should respond with the same id. - pub keep_alive_id: AtomicI64, - /// The last time we sent a keep alive packet. - pub last_keep_alive_time: AtomicCell, /// The last time the player performed an action (for idle timeout). pub last_action_time: AtomicCell, /// The ping in millis. @@ -585,9 +579,6 @@ impl Player { // Since 1 is not possible in vanilla it is used as uninit NonZeroU8::new(1).unwrap(), )), - wait_for_keep_alive: AtomicBool::new(false), - keep_alive_id: AtomicI64::new(0), - last_keep_alive_time: AtomicCell::new(std::time::Instant::now()), last_action_time: AtomicCell::new(std::time::Instant::now()), ping: AtomicU32::new(0), last_attacked_ticks: AtomicU32::new(0), @@ -1868,34 +1859,8 @@ impl Player { ), ) .await; - return; } } - - // TODO This should only be handled by the ClientPlatform - if now.duration_since(self.last_keep_alive_time.load()) >= Duration::from_secs(15) { - if matches!(self.client, ClientPlatform::Bedrock(_)) { - return; - } - // We never got a response from the last keep alive we sent. - if self.wait_for_keep_alive.load(Ordering::Relaxed) { - self.kick( - DisconnectReason::Timeout, - TextComponent::translate_cross( - translation::java::DISCONNECT_TIMEOUT, - translation::bedrock::DISCONNECT_TIMEOUT, - [], - ), - ) - .await; - return; - } - self.wait_for_keep_alive.store(true, Ordering::Relaxed); - self.last_keep_alive_time.store(now); - let id = now.elapsed().as_millis() as i64; - self.keep_alive_id.store(id, Ordering::Relaxed); - self.client.enqueue_packet(&CKeepAlive::new(id)).await; - } } async fn continue_mining( diff --git a/pumpkin/src/net/java/config.rs b/pumpkin/src/net/java/config.rs index ab8b70b20..1460006a8 100644 --- a/pumpkin/src/net/java/config.rs +++ b/pumpkin/src/net/java/config.rs @@ -1,4 +1,7 @@ -use std::{num::NonZeroU8, sync::Arc}; +use std::{ + num::NonZeroU8, + sync::{Arc, atomic::Ordering}, +}; use crate::{ entity::player::ChatMode, @@ -9,14 +12,14 @@ use crate::{ server::Server, }; use core::str; -use pumpkin_data::registry::Registry; +use pumpkin_data::{registry::Registry, translation}; use pumpkin_protocol::{ ConnectionState, java::{ client::config::{CFinishConfig, CRegistryData, CUpdateTags, RegistryEntry}, server::config::{ ResourcePackResponseResult, SClientInformationConfig, SConfigCookieResponse, - SConfigResourcePack, SKnownPacks, SPluginMessage, + SConfigResourcePack, SKeepAlive, SKnownPacks, SPluginMessage, }, }, }; @@ -229,6 +232,20 @@ impl JavaClient { None } + pub async fn handle_config_keep_alive(&self, keep_alive: SKeepAlive) { + if self.wait_for_keep_alive.load(Ordering::Relaxed) + && keep_alive.keep_alive_id == self.keep_alive_id.load() + { + self.wait_for_keep_alive.store(false, Ordering::Relaxed); + } else { + self.kick(TextComponent::translate( + translation::java::DISCONNECT_TIMEOUT, + [], + )) + .await; + } + } + pub async fn handle_config_acknowledged(&self, server: &Arc) -> PacketHandlerResult { debug!("Handling config acknowledgement"); self.connection_state.store(ConnectionState::Play); diff --git a/pumpkin/src/net/java/mod.rs b/pumpkin/src/net/java/mod.rs index 9988a40f7..aa80408c4 100644 --- a/pumpkin/src/net/java/mod.rs +++ b/pumpkin/src/net/java/mod.rs @@ -1,15 +1,18 @@ use std::net::SocketAddr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; use std::{io::Write, sync::Arc}; use bytes::Bytes; use crossbeam::atomic::AtomicCell; use pumpkin_config::networking::compression::CompressionInfo; use pumpkin_data::packet::CURRENT_MC_VERSION; +use pumpkin_data::translation; use pumpkin_protocol::java::server::play::{ SAttack, SChangeGameMode, SChatCommand, SChatMessage, SChunkBatch, SClickSlot, SClientCommand, SClientInformationPlay, SClientTickEnd, SCloseContainer, SCommandSuggestion, SConfirmTeleport, SContainerButtonClick, SCookieResponse as SPCookieResponse, SCustomPayload, SInteract, - SKeepAlive, SMoveVehicle, SPaddleBoat, SPickItemFromBlock, SPlaceRecipe, SPlayPingRequest, + SMoveVehicle, SPaddleBoat, SPickItemFromBlock, SPlaceRecipe, SPlayPingRequest, SPlayerAbilities, SPlayerAction, SPlayerCommand, SPlayerInput, SPlayerLoaded, SPlayerPosition, SPlayerPositionRotation, SPlayerRotation, SPlayerSession, SRecipeBookChangeSettings, SRecipeBookSeenRecipe, SRenameItem, SSelectTrade, SSetCommandBlock, SSetCreativeSlot, @@ -99,6 +102,14 @@ pub struct JavaClient { network_writer: Arc>>>, /// The packet decoder for incoming packets. network_reader: Mutex>>, + /// Keep Alive: + /// + /// Whether we are waiting for a response after sending a keep alive packet. + pub wait_for_keep_alive: AtomicBool, + /// The keep alive packet payload we send. The client should respond with the same id. + pub keep_alive_id: AtomicCell, + /// The last time we sent a keep alive packet. + pub last_keep_alive_time: AtomicCell, } pub enum PacketHandlerResult { @@ -151,6 +162,9 @@ impl JavaClient { network_writer: Arc::new(Mutex::new(TCPNetworkEncoder::new(BufWriter::new(write)))), network_reader: Mutex::new(TCPNetworkDecoder::new(BufReader::new(read))), brand: Mutex::new(None), + wait_for_keep_alive: AtomicBool::new(false), + keep_alive_id: AtomicCell::new(0), + last_keep_alive_time: AtomicCell::new(std::time::Instant::now()), } } pub async fn set_encryption( @@ -224,21 +238,55 @@ impl JavaClient { } pub async fn progress_player_packets(&self, player: &Arc, server: &Arc) { - while let Some(packet) = self.get_packet().await { - match self.handle_play_packet(player, server, &packet).await { - Ok(()) => {} - Err(e) => { - if e.is_kick() { - if let Some(kick_reason) = e.client_kick_reason() { - self.kick(TextComponent::text(kick_reason)).await; - } else { - self.kick(TextComponent::text(format!( - "Error while handling incoming packet {e}" - ))) - .await; + let mut keep_alive_interval = tokio::time::interval(std::time::Duration::from_secs(15)); + + // Skip the immediate first tick so we don't send a keep-alive the exact millisecond they join + keep_alive_interval.tick().await; + + loop { + tokio::select! { + // KEEP-ALIVE TIMER + _ = keep_alive_interval.tick() => { + // If the client never responded to the LAST keep-alive, they timed out. + if self.wait_for_keep_alive.load(Ordering::Relaxed) { + self.kick(TextComponent::translate(translation::java::DISCONNECT_TIMEOUT, [])).await; + break; + } + + // Generate a unique ID (current timestamp in ms) + let keep_alive_id = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as i64; + + self.keep_alive_id.store(keep_alive_id); + self.wait_for_keep_alive.store(true, Ordering::Relaxed); + self.last_keep_alive_time.store(Instant::now()); + self.enqueue_packet(&pumpkin_protocol::java::client::play::CKeepAlive::new(keep_alive_id)).await; + } + + // INCOMING PACKETS + packet_opt = self.get_packet() => { + let Some(packet) = packet_opt else { + break; + }; + + match self.handle_play_packet(player, server, &packet).await { + Ok(()) => {} + Err(e) => { + if e.is_kick() { + if let Some(kick_reason) = e.client_kick_reason() { + self.kick(TextComponent::text(kick_reason)).await; + } else { + self.kick(TextComponent::text(format!( + "Error while handling incoming packet {e}" + ))) + .await; + } + } + e.log(); } } - e.log(); } } } @@ -708,6 +756,12 @@ impl JavaClient { return Ok(Some(i)); } } + id if id == pumpkin_protocol::java::server::config::SKeepAlive::to_id(version) => { + self.handle_config_keep_alive( + pumpkin_protocol::java::server::config::SKeepAlive::read(payload, &version)?, + ) + .await; + } id if id == SConfigCookieResponse::to_id(version) => { self.handle_config_cookie_response(&SConfigCookieResponse::read( payload, &version, @@ -788,9 +842,12 @@ impl JavaClient { self.handle_attack(player, SAttack::read(payload, &version)?, server) .await; } - id if id == SKeepAlive::to_id(version) => { - self.handle_keep_alive(player, SKeepAlive::read(payload, &version)?) - .await; + id if id == pumpkin_protocol::java::server::play::SKeepAlive::to_id(version) => { + self.handle_keep_alive( + player, + pumpkin_protocol::java::server::play::SKeepAlive::read(payload, &version)?, + ) + .await; } id if id == SClientTickEnd::to_id(version) => { // TODO diff --git a/pumpkin/src/net/java/play.rs b/pumpkin/src/net/java/play.rs index 0ed86830c..5b27a1d78 100644 --- a/pumpkin/src/net/java/play.rs +++ b/pumpkin/src/net/java/play.rs @@ -1865,19 +1865,20 @@ impl JavaClient { } pub async fn handle_keep_alive(&self, player: &Player, keep_alive: SKeepAlive) { - if player.wait_for_keep_alive.load(Ordering::Relaxed) - && keep_alive.keep_alive_id == player.keep_alive_id.load(Ordering::Relaxed) + if self.wait_for_keep_alive.load(Ordering::Relaxed) + && keep_alive.keep_alive_id == self.keep_alive_id.load() { - let ping = player.last_keep_alive_time.load().elapsed(); + let ping = self.last_keep_alive_time.load().elapsed(); // Vanilla logic player.ping.store( (player.ping.load(Ordering::Relaxed) * 3 + ping.as_millis() as u32) / 4, Ordering::Relaxed, ); - player.wait_for_keep_alive.store(false, Ordering::Relaxed); + self.wait_for_keep_alive.store(false, Ordering::Relaxed); } else { - self.kick(TextComponent::text( - "Timeout, The server probably has a deadlock!", + self.kick(TextComponent::translate( + translation::java::DISCONNECT_TIMEOUT, + [], )) .await; }