chore(net): don't run keep alive on player tick

.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 }}

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

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,
+}

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::*;

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,
+    CMapItemData, COpenScreen, CParticle, CPlayerAbilities, CPlayerInfoUpdate, CPlayerPosition,
-    CPlayerPosition, CPlayerSpawnPosition, CRespawn, CSetContainerContent, CSetContainerProperty,
+    CPlayerSpawnPosition, CRespawn, CSetContainerContent, CSetContainerProperty, CSetContainerSlot,
-    CSetContainerSlot, CSetCursorItem, CSetEquipment, CSetExperience, CSetHealth,
+    CSetCursorItem, CSetEquipment, CSetExperience, CSetHealth, CSetPlayerInventory,
-    CSetPlayerInventory, CSetSelectedSlot, CSoundEffect, CStopSound, CSubtitle, CSystemChatMessage,
+    CSetSelectedSlot, CSoundEffect, CStopSound, CSubtitle, CSystemChatMessage, CTabList,
-    CTabList, CTitleAnimation, CTitleText, CUnloadChunk, CUpdateMobEffect, CUpdateTime, GameEvent,
+    CTitleAnimation, CTitleText, CUnloadChunk, CUpdateMobEffect, CUpdateTime, GameEvent, MapIcon,
-    MapIcon, MapPatch, Metadata, PlayerAction, PlayerInfoFlags, PreviousMessage,
+    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<Option<(VarInt, Vector3<f64>)>>,
     /// The coordinates of the chunk section the player is currently watching.
     pub watched_section: AtomicCell<Cylindrical>,
-    /// 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<Instant>,
     /// The last time the player performed an action (for idle timeout).
     pub last_action_time: AtomicCell<Instant>,
     /// 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(

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<Server>) -> PacketHandlerResult {
         debug!("Handling config acknowledgement");
         self.connection_state.store(ConnectionState::Play);

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<Mutex<TCPNetworkEncoder<BufWriter<OwnedWriteHalf>>>>,
     /// The packet decoder for incoming packets.
     network_reader: Mutex<TCPNetworkDecoder<BufReader<OwnedReadHalf>>>,
+    /// 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<i64>,
+    /// The last time we sent a keep alive packet.
+    pub last_keep_alive_time: AtomicCell<Instant>,
 }
 
 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<Player>, server: &Arc<Server>) {
-        while let Some(packet) = self.get_packet().await {
+        let mut keep_alive_interval = tokio::time::interval(std::time::Duration::from_secs(15));
-            match self.handle_play_packet(player, server, &packet).await {
+
-                Ok(()) => {}
+        // Skip the immediate first tick so we don't send a keep-alive the exact millisecond they join
-                Err(e) => {
+        keep_alive_interval.tick().await;
-                    if e.is_kick() {
+
-                        if let Some(kick_reason) = e.client_kick_reason() {
+        loop {
-                            self.kick(TextComponent::text(kick_reason)).await;
+            tokio::select! {
-                        } else {
+                // KEEP-ALIVE TIMER
-                            self.kick(TextComponent::text(format!(
+                _ = keep_alive_interval.tick() => {
-                                "Error while handling incoming packet {e}"
+                    // If the client never responded to the LAST keep-alive, they timed out.
-                            )))
+                    if self.wait_for_keep_alive.load(Ordering::Relaxed) {
-                            .await;
+                        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) => {
+            id if id == pumpkin_protocol::java::server::play::SKeepAlive::to_id(version) => {
-                self.handle_keep_alive(player, SKeepAlive::read(payload, &version)?)
+                self.handle_keep_alive(
-                    .await;
+                    player,
+                    pumpkin_protocol::java::server::play::SKeepAlive::read(payload, &version)?,
+                )
+                .await;
             }
             id if id == SClientTickEnd::to_id(version) => {
                 // TODO

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)
+        if self.wait_for_keep_alive.load(Ordering::Relaxed)
-            && keep_alive.keep_alive_id == player.keep_alive_id.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(
+            self.kick(TextComponent::translate(
-                "Timeout, The server probably has a deadlock!",
+                translation::java::DISCONNECT_TIMEOUT,
+                [],
             ))
             .await;
         }