Fix dead locks, vanilla chunk loading and packet issues

pumpkin-nbt/src/deserializer.rs

@@ -1,8 +1,5 @@
-use std::vec::IntoIter;
-
 use crate::*;
 use io::Read;
-use serde::de::value::SeqDeserializer;
 use serde::de::{self, DeserializeSeed, IntoDeserializer, MapAccess, SeqAccess, Visitor};
 use serde::{Deserialize, forward_to_deserialize_any};
 
@@ -116,6 +113,8 @@ pub struct Deserializer<R: Read> {
     // Yes, this breaks with recursion. Just an attempt at a sanity check
     in_list: bool,
     is_named: bool,
+    // For debugging
+    key_stack: Vec<String>,
 }
 
 impl<R: Read> Deserializer<R> {
@@ -125,6 +124,7 @@ impl<R: Read> Deserializer<R> {
             tag_to_deserialize_stack: Vec::new(),
             in_list: false,
             is_named,
+            key_stack: Vec::new(),
         }
     }
 }
@@ -181,21 +181,28 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer<R> {
             END_ID => Err(Error::SerdeError(
                 "Trying to deserialize an END tag!".to_string(),
             )),
-            LIST_ID => {
+            LIST_ID | INT_ARRAY_ID | LONG_ARRAY_ID | BYTE_ARRAY_ID => {
-                let list_type = self.input.get_u8_be()?;
+                let list_type = match tag_to_deserialize {
+                    LIST_ID => self.input.get_u8_be()?,
+                    INT_ARRAY_ID => INT_ID,
+                    LONG_ARRAY_ID => LONG_ID,
+                    BYTE_ARRAY_ID => BYTE_ID,
+                    _ => unreachable!(),
+                };
 
                 let remaining_values = self.input.get_i32_be()?;
                 if remaining_values < 0 {
                     return Err(Error::NegativeLength(remaining_values));
                 }
 
-                visitor.visit_seq(ListAccess {
+                let result = visitor.visit_seq(ListAccess {
                     de: self,
                     list_type,
                     remaining_values: remaining_values as usize,
-                })
+                })?;
+                Ok(result)
             }
-            COMPOUND_ID => self.deserialize_map(visitor),
+            COMPOUND_ID => visitor.visit_map(CompoundAccess { de: self }),
             _ => {
                 let result = match NbtTag::deserialize_data(&mut self.input, tag_to_deserialize)? {
                     NbtTag::Byte(value) => visitor.visit_i8::<Error>(value)?,
@@ -205,22 +212,6 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer<R> {
                     NbtTag::Float(value) => visitor.visit_f32::<Error>(value)?,
                     NbtTag::Double(value) => visitor.visit_f64::<Error>(value)?,
                     NbtTag::String(value) => visitor.visit_string::<Error>(value)?,
-                    NbtTag::LongArray(value) => visitor
-                        .visit_seq::<SeqDeserializer<IntoIter<i64>, Error>>(
-                            value.into_deserializer(),
-                        )?,
-                    NbtTag::IntArray(value) => visitor
-                        .visit_seq::<SeqDeserializer<IntoIter<i32>, Error>>(
-                            value.into_deserializer(),
-                        )?,
-                    NbtTag::ByteArray(value) => {
-                        // For compatibility, we serialize byte arrays as Vec<i8>
-                        // It could be probably changed in the future
-                        let array: Vec<_> = value.iter().map(|&byte| byte as i8).collect();
-                        visitor.visit_seq::<SeqDeserializer<IntoIter<i8>, Error>>(
-                            array.into_deserializer(),
-                        )?
-                    }
                     _ => unreachable!(),
                 };
                 Ok(result)
@@ -277,11 +268,21 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer<R> {
             if *tag_id == BYTE_ID {
                 let value = self.input.get_u8_be()?;
                 if value != 0 {
-                    return visitor.visit_bool(true);
+                    visitor.visit_bool(true)
+                } else {
+                    visitor.visit_bool(false)
                 }
+            } else {
+                Err(Error::UnsupportedType(format!(
+                    "Non-byte bool (found type {tag_id})"
+                )))
             }
+        } else {
+            Err(Error::SerdeError(
+                "Wanted to deserialize a bool, but there was no type hint on the stack!"
+                    .to_string(),
+            ))
         }
-        visitor.visit_bool(false)
     }
 
     fn deserialize_enum<V>(
@@ -312,7 +313,12 @@ impl<'de, R: Read> de::Deserializer<'de> for &mut Deserializer<R> {
         if let Some(tag_id) = self.tag_to_deserialize_stack.pop() {
             if tag_id != COMPOUND_ID {
                 return Err(Error::SerdeError(format!(
-                    "Trying to deserialize a map without a compound ID (with id {tag_id})"
+                    "Trying to deserialize a map without a compound ID ({} with id {})",
+                    self.key_stack
+                        .last()
+                        .cloned()
+                        .unwrap_or_else(|| "compound root".to_string()),
+                    tag_id
                 )));
             }
         } else {
@@ -381,7 +387,9 @@ impl<'de, R: Read> MapAccess<'de> for CompoundAccess<'_, R> {
     where
         V: DeserializeSeed<'de>,
     {
-        seed.deserialize(&mut *self.de)
+        let result = seed.deserialize(&mut *self.de);
+        self.de.key_stack.pop();
+        result
     }
 }
 
@@ -397,6 +405,7 @@ impl<'de, R: Read> de::Deserializer<'de> for MapKey<'_, R> {
         V: de::Visitor<'de>,
     {
         let key = get_nbt_string(&mut self.de.input)?;
+        self.de.key_stack.push(key.clone());
         visitor.visit_string(key)
     }
 

pumpkin-world/src/chunk/io/file_manager.rs

@@ -287,8 +287,7 @@ where
                         error!("Error reading the data before write: {err}");
                         Err(ChunkWritingError::IoError(err))
                     }
-                    Err(err) => {
+                    Err(_) => {
-                        error!("Error reading the data before write: {err:?}");
                         Err(ChunkWritingError::IoError(std::io::ErrorKind::Other))
                     }
                 }?;

pumpkin-world/src/level.rs

@@ -881,13 +881,14 @@ impl Level {
                         match error {
                             // this is expected, and is not an error
                             ChunkReadingError::ChunkNotExist
+                            | ChunkReadingError::InvalidHeader
                             | ChunkReadingError::ParsingError(
                                 ChunkParsingError::ChunkNotGenerated,
                             ) => {}
                             // this is an error, and we should log it
                             error => {
                                 log::error!(
-                                    "Failed to load chunk at {pos:?}: {error} (regenerating)"
+                                    "Failed to load a Entity chunk at {pos:?}: {error} (regenerating)"
                                 );
                             }
                         };

pumpkin/src/entity/decoration/mod.rs

@@ -0,0 +1 @@
+pub mod painting;

pumpkin/src/entity/decoration/painting.rs

@@ -0,0 +1,34 @@
+use std::sync::atomic::Ordering;
+
+use async_trait::async_trait;
+
+use crate::entity::{Entity, EntityBase, living::LivingEntity};
+
+pub struct PaintingEntity {
+    entity: Entity,
+}
+
+impl PaintingEntity {
+    pub fn new(entity: Entity) -> Self {
+        Self { entity }
+    }
+}
+
+#[async_trait]
+impl EntityBase for PaintingEntity {
+    fn get_entity(&self) -> &Entity {
+        &self.entity
+    }
+
+    fn get_living_entity(&self) -> Option<&LivingEntity> {
+        None
+    }
+    async fn write_nbt(&self, nbt: &mut pumpkin_nbt::compound::NbtCompound) {
+        nbt.put_byte("facing", self.entity.data.load(Ordering::Relaxed) as i8);
+    }
+
+    async fn read_nbt(&self, _nbt: &pumpkin_nbt::compound::NbtCompound) {
+        // TODO
+        self.entity.data.store(3, Ordering::Relaxed);
+    }
+}

pumpkin/src/entity/mod.rs

@@ -43,6 +43,7 @@ use tokio::sync::{Mutex, RwLock};
 use crate::world::World;
 
 pub mod ai;
+pub mod decoration;
 pub mod effect;
 pub mod experience_orb;
 pub mod hunger;
@@ -174,6 +175,9 @@ pub struct Entity {
     pub portal_cooldown: AtomicU32,
 
     pub portal_manager: Mutex<Option<Mutex<PortalManager>>>,
+
+    /// The data send in the Entity Spawn packet
+    pub data: AtomicI32,
 }
 
 impl Entity {
@@ -221,6 +225,7 @@ impl Entity {
             bounding_box_size: AtomicCell::new(bounding_box_size),
             invulnerable: AtomicBool::new(invulnerable),
             damage_immunities: Vec::new(),
+            data: AtomicI32::new(0),
             fire_ticks: AtomicI32::new(-1),
             has_visual_fire: AtomicBool::new(false),
             portal_cooldown: AtomicU32::new(0),
@@ -439,7 +444,7 @@ impl Entity {
             self.pitch.load(),
             self.yaw.load(),
             self.head_yaw.load(), // todo: head_yaw and yaw are swapped, find out why
-            0.into(),
+            self.data.load(Relaxed).into(),
             entity_vel,
         )
     }

pumpkin/src/entity/type.rs

@@ -9,6 +9,7 @@ use crate::{
     entity::{
         Entity, EntityBase,
         ai::path::Navigator,
+        decoration::painting::PaintingEntity,
         living::LivingEntity,
         mob::{MobEntity, zombie::Zombie},
     },
@@ -23,15 +24,15 @@ pub fn from_type(
 ) -> Arc<dyn EntityBase> {
     let entity = Entity::new(uuid, world.clone(), position, entity_type, false);
 
-    #[allow(clippy::single_match)]
+    let base: Arc<dyn EntityBase> = match entity_type {
-    let mob = match entity_type {
+        EntityType::ZOMBIE => Arc::new(Zombie::make(entity)),
-        EntityType::ZOMBIE => Zombie::make(entity),
+        EntityType::PAINTING => Arc::new(PaintingEntity::new(entity)),
         // TODO
-        _ => MobEntity {
+        _ => Arc::new(MobEntity {
             living_entity: LivingEntity::new(entity),
             goals: Mutex::new(vec![]),
             navigator: Mutex::new(Navigator::default()),
-        },
+        }),
     };
-    Arc::new(mob)
+    base
 }

pumpkin/src/lib.rs

@@ -325,7 +325,7 @@ impl PumpkinServer {
     pub async fn unified_listener_task(
         &self,
         mut master_client_id_counter: u64,
-        _tasks: &Arc<TaskTracker>,
+        tasks: &Arc<TaskTracker>,
         bedrock_clients: &Arc<tokio::sync::Mutex<HashMap<SocketAddr, Arc<BedrockClientPlatform>>>>,
     ) -> bool {
         let mut udp_buf = vec![0; 4096]; // Buffer for UDP receive
@@ -355,25 +355,25 @@ impl PumpkinServer {
 
                         let server_clone = self.server.clone();
 
-                        tokio::spawn(async move {
+                        tasks.spawn(async move {
-                                    java_client.process_packets(&server_clone).await;
+                                java_client.process_packets(&server_clone).await;
-                                    java_client.close();
+                                java_client.close();
-                                    java_client.await_tasks().await;
+                                java_client.await_tasks().await;
 
-                                    let player = java_client.player.lock().await;
+                                let player = java_client.player.lock().await;
-                                    if let Some(player) = player.as_ref() {
+                                if let Some(player) = player.as_ref() {
-                                        log::debug!("Cleaning up player for id {client_id}");
+                                    log::debug!("Cleaning up player for id {client_id}");
 
-                                        if let Err(e) = server_clone.player_data_storage
+                                    if let Err(e) = server_clone.player_data_storage
                                             .handle_player_leave(player)
                                             .await
-                                        {
+                                    {
-                                            log::error!("Failed to save player data on disconnect: {e}");
+                                        log::error!("Failed to save player data on disconnect: {e}");
-                                        }
-
-                                        player.remove().await;
-                                        server_clone.remove_player(player).await;
                                     }
+
+                                    player.remove().await;
+                                    server_clone.remove_player(player).await;
+                                }
                         });
                     }
                     Err(e) => {
@@ -411,7 +411,7 @@ impl PumpkinServer {
 
                         let reader = Cursor::new(received_data.to_vec());
                         let client = client.clone();
-                        tokio::spawn(async move {
+                        tasks.spawn(async move {
                             client.process_packet(&server_clone, reader).await;
                         });
                     }

pumpkin/src/server/mod.rs

@@ -14,11 +14,9 @@ use connection_cache::{CachedBranding, CachedStatus};
 use key_store::KeyStore;
 use pumpkin_config::{BASIC_CONFIG, advanced_config};
 
-use pumpkin_inventory::screen_handler::InventoryPlayer;
 use pumpkin_macros::send_cancellable;
 use pumpkin_protocol::java::client::login::CEncryptionRequest;
 use pumpkin_protocol::java::client::play::CChangeDifficulty;
-use pumpkin_protocol::java::client::play::CSetSelectedSlot;
 use pumpkin_protocol::{ClientPacket, java::client::config::CPluginMessage};
 use pumpkin_registry::{Registry, VanillaDimensionType};
 use pumpkin_util::Difficulty;
@@ -353,10 +351,6 @@ impl Server {
                         }
                     }
 
-                    player.enqueue_set_held_item_packet(&CSetSelectedSlot::new(
-                        player.get_inventory().get_selected_slot() as i8,
-                    )).await;
-
                     // Send tick rate information to the new player
                     if let ClientPlatform::Java(_) = &player.client {
                         self.tick_rate_manager.update_joining_player(&player).await;

pumpkin/src/world/mod.rs

@@ -49,7 +49,7 @@ use pumpkin_data::{
     sound::{Sound, SoundCategory},
     world::{RAW, WorldEvent},
 };
-use pumpkin_inventory::equipment_slot::EquipmentSlot;
+use pumpkin_inventory::{equipment_slot::EquipmentSlot, screen_handler::InventoryPlayer};
 use pumpkin_macros::send_cancellable;
 use pumpkin_nbt::{compound::NbtCompound, to_bytes_unnamed};
 use pumpkin_protocol::{
@@ -68,8 +68,8 @@ use pumpkin_protocol::{
         client::play::{
             CBlockEntityData, CEntityStatus, CGameEvent, CLogin, CMultiBlockUpdate,
             CPlayerChatMessage, CPlayerInfoUpdate, CRemoveEntities, CRemovePlayerInfo,
-            CSoundEffect, CSpawnEntity, FilterType, GameEvent, InitChat, PlayerAction,
+            CSetSelectedSlot, CSoundEffect, CSpawnEntity, FilterType, GameEvent, InitChat,
-            PlayerInfoFlags,
+            PlayerAction, PlayerInfoFlags,
         },
         server::play::SChatMessage,
     },
@@ -1040,6 +1040,13 @@ impl World {
         }
         player.send_client_information().await;
 
+        // Sync selected slot
+        player
+            .enqueue_set_held_item_packet(&CSetSelectedSlot::new(
+                player.get_inventory().get_selected_slot() as i8,
+            ))
+            .await;
+
         // Start waiting for level chunks. Sets the "Loading Terrain" screen
         log::debug!("Sending waiting chunks to {}", player.gameprofile.name);
         player