fix: packet spam and mob leaks (#1865)

* feat: add mob ai throttling

* chore: formating

* fix: prevent entity leak and server timeout during death animation

* fix: vanilla parity + optimize entity position and rotation updates to reduce unnecessary broadcasts

* chore: formatting

pumpkin/src/entity/falling.rs

@@ -81,9 +81,10 @@ impl EntityBase for FallingEntity {
 
             entity.velocity.store(velo.multiply(0.98, 0.98, 0.98));
 
-            entity.send_pos_rot().await;
-
-            entity.send_velocity().await;
+            if entity.velocity_dirty.swap(false, Ordering::SeqCst) {
+                entity.send_pos_rot().await;
+                entity.send_velocity().await;
+            }
         })
     }
 

pumpkin/src/entity/living.rs

@@ -2123,8 +2123,9 @@ impl EntityBase for LivingEntity {
             }
             if self.health.load() <= 0.0 {
                 let time = self.death_time.fetch_add(1, Relaxed);
-                if time >= 20 && self.entity.is_alive() {
-                    // Spawn Death particles
+                // Only send death particles once (on the exact tick death_time reaches 20)
+                // and then remove the entity, preventing entity_event spam.
+                if time == 20 && !self.entity.removed.swap(true, Ordering::Relaxed) {
                     self.entity
                         .world
                         .load()

pumpkin/src/entity/mod.rs

@@ -55,7 +55,7 @@ use std::pin::Pin;
 use std::sync::{
     Arc,
     atomic::{
-        AtomicBool, AtomicI32, AtomicU32,
+        AtomicBool, AtomicI32, AtomicU8, AtomicU32,
         Ordering::{self, Relaxed},
     },
 };
@@ -476,6 +476,14 @@ pub struct Entity {
     pub velocity_dirty: AtomicBool,
     /// Set when an Entity is to be removed but could still be referenced
     pub removed: AtomicBool,
+    /// The last sent yaw value (encoded as u8) for change detection
+    pub last_sent_yaw: AtomicU8,
+    /// The last sent pitch value (encoded as u8) for change detection
+    pub last_sent_pitch: AtomicU8,
+    /// Cache for the last sent position to optimize Entity Pos update packets
+    pub last_sent_pos: AtomicCell<Vector3<f64>>,
+    /// Cache for the last sent head yaw byte
+    pub last_sent_head_yaw: AtomicU8,
 }
 
 impl Entity {
@@ -565,6 +573,10 @@ impl Entity {
             movement_multiplier: AtomicCell::new(Vector3::default()),
             velocity_dirty: AtomicBool::new(true),
             removed: AtomicBool::new(false),
+            last_sent_yaw: AtomicU8::new(0),
+            last_sent_pitch: AtomicU8::new(0),
+            last_sent_head_yaw: AtomicU8::new(0),
+            last_sent_pos: AtomicCell::new(position),
         }
     }
 
@@ -694,20 +706,22 @@ impl Entity {
 
         // Broadcast the update packet.
 
-        // TODO: Do caching to only send the packet when needed.
-
-        let yaw = (yaw * 256.0 / 360.0).rem_euclid(256.0);
-
         let yaw = (yaw * 256.0 / 360.0).rem_euclid(256.0) as u8;
+        let pitch = (pitch * 256.0 / 360.0).rem_euclid(256.0) as u8;
 
-        let pitch = (pitch * 256.0 / 360.0).rem_euclid(256.0);
+        if yaw == self.last_sent_yaw.load(Relaxed) && pitch == self.last_sent_pitch.load(Relaxed) {
+            return;
+        }
+
+        self.last_sent_yaw.store(yaw, Relaxed);
+        self.last_sent_pitch.store(pitch, Relaxed);
 
         self.world
             .load()
             .broadcast_packet_all(&CUpdateEntityRot::new(
                 self.entity_id.into(),
                 yaw,
-                pitch as u8,
+                pitch,
                 self.on_ground.load(Relaxed),
             ))
             .await;
@@ -716,6 +730,11 @@ impl Entity {
     }
 
     pub async fn send_head_rot(&self, head_yaw: u8) {
+        if head_yaw == self.last_sent_head_yaw.load(Relaxed) {
+            return;
+        }
+        self.last_sent_head_yaw.store(head_yaw, Relaxed);
+
         self.world
             .load()
             .broadcast_packet_all(&CHeadRot::new(self.entity_id.into(), head_yaw))
@@ -1031,8 +1050,7 @@ impl Entity {
     }
 
     pub async fn send_pos_rot(&self) {
-        let old = self.update_last_pos();
-
+        let old = self.last_sent_pos.load();
         let new = self.pos.load();
 
         let converted = Vector3::new(
@@ -1044,25 +1062,54 @@ impl Entity {
         let yaw = self.yaw.load();
 
         let pitch = self.pitch.load();
-
-        // Broadcast the update packet.
-
-        // TODO: Do caching to only send the packet when needed.
-
         let yaw = (yaw * 256.0 / 360.0).rem_euclid(256.0) as u8;
+        let pitch = (pitch * 256.0 / 360.0).rem_euclid(256.0) as u8;
 
-        let pitch = (pitch * 256.0 / 360.0).rem_euclid(256.0);
+        // Only broadcast when position or rotation has actually changed.
+        let pos_changed = converted.x != 0 || converted.y != 0 || converted.z != 0;
+        let rot_changed =
+            yaw != self.last_sent_yaw.load(Relaxed) || pitch != self.last_sent_pitch.load(Relaxed);
 
-        self.world
-            .load()
-            .broadcast_packet_all(&CUpdateEntityPosRot::new(
-                self.entity_id.into(),
-                Vector3::new(converted.x, converted.y, converted.z),
-                yaw,
-                pitch as u8,
-                self.on_ground.load(Relaxed),
-            ))
-            .await;
+        if !pos_changed && !rot_changed {
+            return;
+        }
+
+        self.last_sent_pos.store(new);
+        self.last_sent_yaw.store(yaw, Relaxed);
+        self.last_sent_pitch.store(pitch, Relaxed);
+
+        // Dynamically pick the most efficient packet
+        if pos_changed && rot_changed {
+            self.world
+                .load()
+                .broadcast_packet_all(&CUpdateEntityPosRot::new(
+                    self.entity_id.into(),
+                    Vector3::new(converted.x, converted.y, converted.z),
+                    yaw,
+                    pitch,
+                    self.on_ground.load(Relaxed),
+                ))
+                .await;
+        } else if pos_changed {
+            self.world
+                .load()
+                .broadcast_packet_all(&CUpdateEntityPos::new(
+                    self.entity_id.into(),
+                    Vector3::new(converted.x, converted.y, converted.z),
+                    self.on_ground.load(Relaxed),
+                ))
+                .await;
+        } else if rot_changed {
+            self.world
+                .load()
+                .broadcast_packet_all(&CUpdateEntityRot::new(
+                    self.entity_id.into(),
+                    yaw,
+                    pitch,
+                    self.on_ground.load(Relaxed),
+                ))
+                .await;
+        }
         self.send_head_rot(yaw).await;
     }
 
@@ -1075,7 +1122,7 @@ impl Entity {
     }
 
     pub async fn send_pos(&self) {
-        let old = self.update_last_pos();
+        let old = self.last_sent_pos.load();
         let new = self.pos.load();
 
         let converted = Vector3::new(
@@ -1084,6 +1131,13 @@ impl Entity {
             new.z.mul_add(4096.0, -(old.z * 4096.0)) as i16,
         );
 
+        // Only broadcast when position has actually changed.
+        if converted.x == 0 && converted.y == 0 && converted.z == 0 {
+            return;
+        }
+
+        self.last_sent_pos.store(new);
+
         self.world
             .load()
             .broadcast_packet_all(&CUpdateEntityPos::new(
@@ -2187,6 +2241,18 @@ impl Entity {
         if let Some(pitch) = pitch {
             self.set_pitch(pitch);
         }
+        // Update cache so we don't send rubberbanding deltas
+        self.last_sent_pos.store(position);
+        if let Some(yaw) = yaw {
+            self.last_sent_yaw
+                .store((yaw * 256.0 / 360.0).rem_euclid(256.0) as u8, Relaxed);
+            self.last_sent_head_yaw
+                .store((yaw * 256.0 / 360.0).rem_euclid(256.0) as u8, Relaxed);
+        }
+        if let Some(pitch) = pitch {
+            self.last_sent_pitch
+                .store((pitch * 256.0 / 360.0).rem_euclid(256.0) as u8, Relaxed);
+        }
         self.world
             .load()
             .broadcast_packet_all(&CEntityPositionSync::new(