From 9dd9754b4e891b1d7be98141e99529f8bd1f1542 Mon Sep 17 00:00:00 2001 From: Alexander Medvedev Date: Sat, 18 Apr 2026 08:35:48 +0200 Subject: [PATCH] chore: update linear format from v1 to v2 --- Cargo.lock | 7 + pumpkin-config/src/chunk.rs | 2 +- pumpkin-world/Cargo.toml | 1 + pumpkin-world/src/chunk/format/linear.rs | 1003 +++++++++++++--------- pumpkin-world/src/level.rs | 11 +- 5 files changed, 612 insertions(+), 412 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c90c79bc3..dd95f4b97 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3051,6 +3051,7 @@ dependencies = [ "tokio-util", "tracing", "uuid", + "xxhash-rust", ] [[package]] @@ -5516,6 +5517,12 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xxhash-rust" +version = "0.8.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fdd20c5420375476fbd4394763288da7eb0cc0b8c11deed431a91562af7335d3" + [[package]] name = "yoke" version = "0.8.2" diff --git a/pumpkin-config/src/chunk.rs b/pumpkin-config/src/chunk.rs index 881935887..6d6c697b0 100644 --- a/pumpkin-config/src/chunk.rs +++ b/pumpkin-config/src/chunk.rs @@ -18,7 +18,7 @@ pub enum ChunkConfig { impl Default for ChunkConfig { fn default() -> Self { - Self::Anvil(AnvilChunkConfig::default()) + Self::Linear(LinearChunkConfig::default()) } } diff --git a/pumpkin-world/Cargo.toml b/pumpkin-world/Cargo.toml index 462488d83..ea9e30d3b 100644 --- a/pumpkin-world/Cargo.toml +++ b/pumpkin-world/Cargo.toml @@ -41,6 +41,7 @@ dashmap.workspace = true flate2.workspace = true lz4-java-wrc.workspace = true ruzstd.workspace = true +xxhash-rust = { version = "0.8", features = ["xxh64"] } itertools.workspace = true enum_dispatch.workspace = true diff --git a/pumpkin-world/src/chunk/format/linear.rs b/pumpkin-world/src/chunk/format/linear.rs index 6c541000a..2166218a8 100644 --- a/pumpkin-world/src/chunk/format/linear.rs +++ b/pumpkin-world/src/chunk/format/linear.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::io::{ErrorKind, Read}; use std::marker::PhantomData; use std::path::PathBuf; @@ -12,155 +13,361 @@ use pumpkin_util::math::vector2::Vector2; use ruzstd::decoding::StreamingDecoder; use ruzstd::encoding::{CompressionLevel, compress_to_vec}; use tokio::io::{AsyncWriteExt, BufWriter}; -use tracing::{error, trace, warn}; +use tracing::{error, warn}; +use xxhash_rust::xxh64::xxh64; use super::anvil::CHUNK_COUNT; -/// The signature of the linear file format -/// used as a header and footer described in +/// The signature used as both header and footer. +/// const SIGNATURE: [u8; 8] = u64::to_be_bytes(0xc3ff13183cca9d9a); -#[derive(Default, Copy, Clone)] -struct LinearChunkHeader { - size: u32, - timestamp: u32, +/// Allowed grid sizes for v2 bucket subdivision. +const VALID_GRID_SIZES: &[u8] = &[1, 2, 4, 8, 16, 32]; + +/// Default grid size. 2×2 = 4 buckets gives a reasonable write amplification +/// tradeoff for most worlds. +const DEFAULT_GRID_SIZE: u8 = 2; + +// --------------------------------------------------------------------------- +// Superblock (26 bytes) +// --------------------------------------------------------------------------- +// +// 0.. 8 uint64 Signature (0xc3ff13183cca9d9a) +// 8.. 9 uint8 Version (2) +// 9..17 uint64 Newest timestamp +// 17..18 int8 Grid size +// 18..22 int32 Region X +// 22..26 int32 Region Z + +struct LinearV2Superblock { + newest_timestamp: u64, + /// One of {1, 2, 4, 8, 16, 32}. + grid_size: u8, + region_x: i32, + region_z: i32, } -#[derive(Default, PartialEq, Eq, Clone, Copy)] -pub enum LinearVersion { - #[default] - /// Represents an invalid or uninitialized version. - None = 0x00, - /// Version 1 of the Linear Region File Format. (Default) - /// - /// Described in: - V1 = 0x01, - /// Version 2 of the Linear Region File Format (currently unsupported). - /// - /// Described in: - V2 = 0x02, +impl LinearV2Superblock { + const SIZE: usize = 26; + + fn from_bytes(buf: &[u8]) -> Result { + if buf.len() < Self::SIZE { + return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); + } + let mut b = buf; + let mut sig = [0u8; 8]; + sig.copy_from_slice(&b[..8]); + b = &b[8..]; + + if sig != SIGNATURE { + error!("Linear v2: invalid superblock signature"); + return Err(ChunkReadingError::InvalidHeader); + } + + let version = b.get_u8(); + if version != 0x02 { + error!("Linear v2: unexpected version byte {version:#x} in superblock"); + return Err(ChunkReadingError::InvalidHeader); + } + + let newest_timestamp = b.get_u64(); + let grid_size = b.get_u8(); + let region_x = b.get_i32(); + let region_z = b.get_i32(); + + if !VALID_GRID_SIZES.contains(&grid_size) { + error!("Linear v2: invalid grid size {grid_size}"); + return Err(ChunkReadingError::InvalidHeader); + } + + Ok(Self { + newest_timestamp, + grid_size, + region_x, + region_z, + }) + } + + fn to_bytes(&self) -> [u8; Self::SIZE] { + let mut out = [0u8; Self::SIZE]; + let mut b = out.as_mut_slice(); + b.put_slice(&SIGNATURE); + b.put_u8(0x02); + b.put_u64(self.newest_timestamp); + b.put_u8(self.grid_size); + b.put_i32(self.region_x); + b.put_i32(self.region_z); + out + } } -struct LinearFileHeader { - /// ( 0.. 1 Bytes) The version of the Linear Region File format. - version: LinearVersion, - /// ( 1.. 9 Bytes) The timestamp of the newest chunk in the region file. - newest_timestamp: u64, - /// ( 9..10 Bytes) The zstd compression level used for chunk data. - compression_level: u8, - /// (10..12 Bytes) The number of non-zero-size chunks in the region file. - chunks_count: u16, - /// (12..16 Bytes) The total size in bytes of the compressed chunk headers and chunk data. - chunks_bytes: usize, - /// (16..24 Bytes) A hash of the region file (unused). - region_hash: u64, + +// --------------------------------------------------------------------------- +// Chunk existence bitmap (128 bytes = 1024 bits) +// --------------------------------------------------------------------------- + +struct ChunkBitmap([u8; 128]); + +impl ChunkBitmap { + const SIZE: usize = 128; + + fn new() -> Self { + Self([0u8; 128]) + } + + fn set(&mut self, index: usize, exists: bool) { + let byte = index / 8; + let bit = index % 8; + if exists { + self.0[byte] |= 1 << bit; + } else { + self.0[byte] &= !(1 << bit); + } + } + + #[allow(dead_code)] + fn get(&self, index: usize) -> bool { + let byte = index / 8; + let bit = index % 8; + (self.0[byte] >> bit) & 1 == 1 + } } -pub struct LinearFile { - chunks_headers: [LinearChunkHeader; CHUNK_COUNT], + +// --------------------------------------------------------------------------- +// NBT features dictionary +// --------------------------------------------------------------------------- +// +// Repeated: u8 key_len, key bytes, u32 value +// Terminated by a single 0x00 byte. + +struct NbtFeatures(HashMap); + +impl NbtFeatures { + fn empty() -> Self { + Self(HashMap::new()) + } + + fn from_bytes(buf: &mut impl Buf) -> Result { + let mut map = HashMap::new(); + loop { + if !buf.has_remaining() { + return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); + } + let key_len = buf.get_u8(); + if key_len == 0 { + break; + } + if buf.remaining() < key_len as usize + 4 { + return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); + } + let key_bytes: Vec = (0..key_len).map(|_| buf.get_u8()).collect(); + let key = String::from_utf8(key_bytes).map_err(|_| ChunkReadingError::InvalidHeader)?; + let value = buf.get_u32(); + map.insert(key, value); + } + Ok(Self(map)) + } + + fn to_bytes(&self) -> Vec { + let mut out = Vec::new(); + for (key, value) in &self.0 { + let key_bytes = key.as_bytes(); + out.push(key_bytes.len() as u8); + out.extend_from_slice(key_bytes); + out.extend_from_slice(&value.to_be_bytes()); + } + out.push(0); // end marker + out + } +} + +// --------------------------------------------------------------------------- +// Bucket size entry (13 bytes per bucket) +// --------------------------------------------------------------------------- +// +// 0..4 uint32 Bucket size (compressed) +// 4..5 int8 Compression level +// 5..13 uint64 xxhash64 of compressed bucket data + +struct BucketSizeEntry { + size: u32, + compression_level: i8, + xxhash: u64, +} + +impl BucketSizeEntry { + const SIZE: usize = 13; + + fn from_bytes(buf: &mut impl Buf) -> Result { + if buf.remaining() < Self::SIZE { + return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); + } + Ok(Self { + size: buf.get_u32(), + compression_level: buf.get_i8(), + xxhash: buf.get_u64(), + }) + } + + fn to_bytes(&self) -> [u8; Self::SIZE] { + let mut out = [0u8; Self::SIZE]; + let mut b = out.as_mut_slice(); + b.put_u32(self.size); + b.put_i8(self.compression_level); + b.put_u64(self.xxhash); + out + } +} + +// --------------------------------------------------------------------------- +// Per-chunk record inside a decompressed bucket +// --------------------------------------------------------------------------- +// +// 0.. 4 uint32 Chunk size (0 = absent) +// 4..12 uint64 Chunk timestamp +// 12.. bytes Chunk NBT data (only when size > 0) + +struct BucketChunkEntry { + timestamp: u64, + data: Option, +} + +impl BucketChunkEntry { + /// Number of bytes for the fixed part (size + timestamp). + const FIXED_SIZE: usize = 12; + + fn write_into(&self, out: &mut Vec) { + match &self.data { + None => { + out.put_u32(0); + out.put_u64(self.timestamp); + } + Some(bytes) => { + out.put_u32(bytes.len() as u32); + out.put_u64(self.timestamp); + out.extend_from_slice(bytes); + } + } + } + + fn read_from(buf: &mut Bytes) -> Result { + if buf.remaining() < Self::FIXED_SIZE { + return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); + } + let size = buf.get_u32() as usize; + let timestamp = buf.get_u64(); + if size == 0 { + return Ok(Self { + timestamp, + data: None, + }); + } + if buf.remaining() < size { + warn!( + "Linear v2: not enough bytes for chunk (need {size}, have {})", + buf.remaining() + ); + return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); + } + let data = buf.split_to(size); + Ok(Self { + timestamp, + data: Some(data), + }) + } +} + +// --------------------------------------------------------------------------- +// Main LinearV2File struct +// --------------------------------------------------------------------------- + +pub struct LinearV2File { + /// Region coordinates, needed to rebuild the superblock on write. + region_x: i32, + region_z: i32, + + /// Grid size (1, 2, 4, 8, 16, or 32). + grid_size: u8, + + /// Per-chunk timestamps. + timestamps: [u64; CHUNK_COUNT], + + /// Per-chunk compressed NBT data (None = chunk absent). chunks_data: [Option; CHUNK_COUNT], _dummy: PhantomData, } -impl LinearChunkHeader { - const CHUNK_HEADER_SIZE: usize = 8; - fn from_bytes(bytes: &[u8]) -> Self { - let mut bytes = bytes; +impl Default for LinearV2File { + fn default() -> Self { Self { - size: bytes.get_u32(), - timestamp: bytes.get_u32(), - } - } - - pub fn to_bytes(self) -> [u8; Self::CHUNK_HEADER_SIZE] { - let mut bytes = [0u8; Self::CHUNK_HEADER_SIZE]; - bytes[0..4].copy_from_slice(&self.size.to_be_bytes()); - bytes[4..8].copy_from_slice(&self.timestamp.to_be_bytes()); - - bytes - } -} - -impl From for LinearVersion { - fn from(value: u8) -> Self { - match value { - 0x01 => Self::V1, - 0x02 => Self::V2, - _ => Self::None, + region_x: 0, + region_z: 0, + grid_size: DEFAULT_GRID_SIZE, + timestamps: [0u64; CHUNK_COUNT], + chunks_data: [const { None }; CHUNK_COUNT], + _dummy: PhantomData, } } } -impl LinearFileHeader { - const FILE_HEADER_SIZE: usize = 24; - - fn check_version(&self) -> Result<(), ChunkReadingError> { - match self.version { - LinearVersion::None => { - error!("Invalid version in the file header"); - Err(ChunkReadingError::InvalidHeader) - } - LinearVersion::V2 => { - error!("LinearFormat Version 2 for Chunks is not supported yet"); - Err(ChunkReadingError::InvalidHeader) - } - _ => Ok(()), - } - } - fn from_bytes(bytes: &[u8]) -> Self { - let mut buf = bytes; - - Self { - version: buf.get_u8().into(), - newest_timestamp: buf.get_u64(), - compression_level: buf.get_u8(), - chunks_count: buf.get_u16(), - chunks_bytes: buf.get_u32() as usize, - region_hash: buf.get_u64(), - } - } - - fn to_bytes(&self) -> [u8; Self::FILE_HEADER_SIZE] { - let mut bytes = [0u8; Self::FILE_HEADER_SIZE]; - let mut buf = bytes.as_mut_slice(); - buf.put_u8(self.version as u8); - buf.put_u64(self.newest_timestamp); - buf.put_u8(self.compression_level); - buf.put_u16(self.chunks_count); - buf.put_u32(self.chunks_bytes as u32); - buf.put_u64(self.region_hash); - bytes - } -} - -impl LinearFile { +impl LinearV2File { const fn get_chunk_index(x: i32, z: i32) -> usize { AnvilChunkFile::::get_chunk_index(x, z) } - fn check_signature(bytes: &[u8]) -> Result<(), ChunkReadingError> { - if bytes == SIGNATURE { - Ok(()) - } else { - error!("Linear signature is invalid!"); - Err(ChunkReadingError::InvalidHeader) + /// Number of buckets in the grid (grid_size²). + fn bucket_count(grid_size: u8) -> usize { + (grid_size as usize) * (grid_size as usize) + } + + /// Number of chunks in each bucket. + fn chunks_per_bucket(grid_size: u8) -> usize { + CHUNK_COUNT / Self::bucket_count(grid_size) + } + + /// Build the decompressed byte stream for bucket `bucket_idx`. + fn serialise_bucket(&self, bucket_idx: usize) -> Vec { + let grid_size = self.grid_size; + let cpb = Self::chunks_per_bucket(grid_size); + let mut buf = Vec::new(); + + for local in 0..cpb { + // Recover the global chunk index from bucket + local. + let chunk_index = Self::global_chunk_index(bucket_idx, local, grid_size); + let entry = BucketChunkEntry { + timestamp: self.timestamps[chunk_index], + data: self.chunks_data[chunk_index].clone(), + }; + entry.write_into(&mut buf); } + buf + } + + /// Inverse of `chunk_bucket_index` + `chunk_local_index`. + fn global_chunk_index(bucket_idx: usize, local: usize, grid_size: u8) -> usize { + let stride = 32 / grid_size as usize; + let bucket_row = bucket_idx / grid_size as usize; + let bucket_col = bucket_idx % grid_size as usize; + let local_row = local / stride; + let local_col = local % stride; + let cz = bucket_row * stride + local_row; + let cx = bucket_col * stride + local_col; + cz * 32 + cx + } + + fn build_bitmap(&self) -> ChunkBitmap { + let mut bitmap = ChunkBitmap::new(); + for (i, chunk) in self.chunks_data.iter().enumerate() { + bitmap.set(i, chunk.is_some()); + } + bitmap } } -impl Default for LinearFile { - fn default() -> Self { - Self { - chunks_headers: [LinearChunkHeader::default(); CHUNK_COUNT], - chunks_data: [const { None }; CHUNK_COUNT], - _dummy: Default::default(), - } - } -} - -impl ChunkSerializer for LinearFile { +impl ChunkSerializer for LinearV2File { type Data = S; type WriteBackend = PathBuf; - type ChunkConfig = LinearChunkConfig; fn should_write(&self, is_watched: bool) -> bool { @@ -174,138 +381,158 @@ impl ChunkSerializer for LinearFile { async fn write(&self, path: &PathBuf) -> Result<(), std::io::Error> { let temp_path = path.with_extension("tmp"); - trace!("Writing tmp file to disk: {}", temp_path.display()); - let file = tokio::fs::File::create(&temp_path).await?; - let mut write = BufWriter::new(file); + let mut writer = BufWriter::new(file); - // Parse the headers to a buffer - let mut data_buffer: Vec = self - .chunks_headers - .iter() - .flat_map(|header| header.to_bytes()) - .collect(); + let grid_size = self.grid_size; + let bucket_count = Self::bucket_count(grid_size); - for chunk in self.chunks_data.iter().flatten() { - data_buffer.extend_from_slice(chunk); + let mut compressed_buckets: Vec> = Vec::with_capacity(bucket_count); + let mut bucket_entries: Vec = Vec::with_capacity(bucket_count); + + for bucket_idx in 0..bucket_count { + let raw = self.serialise_bucket(bucket_idx); + // TODO: ruzstd currently only supports Fastest level. + let compressed = + compress_to_vec(raw.as_slice(), CompressionLevel::Fastest).into_boxed_slice(); + let hash = xxh64(&compressed, 0); + bucket_entries.push(BucketSizeEntry { + size: compressed.len() as u32, + compression_level: 1, + xxhash: hash, + }); + compressed_buckets.push(compressed); } - // TODO: Currently ruzstd only supports fastest - let compressed_buffer = - compress_to_vec(data_buffer.as_slice(), CompressionLevel::Fastest).into_boxed_slice(); + let newest_timestamp = self.timestamps.iter().copied().max().unwrap_or(0); - let file_header = LinearFileHeader { - chunks_bytes: compressed_buffer.len(), - compression_level: 1, // TODO: Currently ruzstd only supports fastest - chunks_count: self - .chunks_headers - .iter() - .filter(|&header| header.size != 0) - .count() as u16, - newest_timestamp: self - .chunks_headers - .iter() - .map(|header| header.timestamp) - .max() - .unwrap_or(0) as u64, - version: LinearVersion::V1, - region_hash: 0, + let superblock = LinearV2Superblock { + newest_timestamp, + grid_size, + region_x: self.region_x, + region_z: self.region_z, + }; + + let bitmap = self.build_bitmap(); + + let features = NbtFeatures::empty(); + + writer.write_all(&superblock.to_bytes()).await?; + writer.write_all(&bitmap.0).await?; + writer.write_all(&features.to_bytes()).await?; + + for entry in &bucket_entries { + writer.write_all(&entry.to_bytes()).await?; } - .to_bytes(); - write.write_all(&SIGNATURE).await?; - write.write_all(&file_header).await?; - write.write_all(&compressed_buffer).await?; - write.write_all(&SIGNATURE).await?; + for compressed in &compressed_buckets { + writer.write_all(compressed).await?; + } - write.flush().await?; + writer.write_all(&SIGNATURE).await?; + writer.flush().await?; - // The rename of the file works like an atomic operation ensuring - // that the data is not corrupted before the rename is completed - tokio::fs::rename(temp_path, &path).await?; - - trace!("Wrote file to Disk: {}", path.display()); + // Atomic rename so a crash during write cannot produce a torn file. + tokio::fs::rename(temp_path, path).await?; Ok(()) } fn read(raw_file: Bytes) -> Result { - let Some((signature, raw_file_bytes)) = raw_file.split_at_checked(SIGNATURE.len()) else { + let mut buf = raw_file; + + if buf.remaining() < LinearV2Superblock::SIZE { return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); - }; + } + let superblock_bytes = buf.split_to(LinearV2Superblock::SIZE); + let superblock = LinearV2Superblock::from_bytes(&superblock_bytes)?; + let grid_size = superblock.grid_size; + let bucket_count = Self::bucket_count(grid_size); - Self::check_signature(signature)?; - - let Some((header_bytes, raw_file_bytes)) = - raw_file_bytes.split_at_checked(LinearFileHeader::FILE_HEADER_SIZE) - else { + if buf.remaining() < ChunkBitmap::SIZE { return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); - }; + } + // We read the bitmap but per the spec it is not yet reliable, so we + // do not use it to short-circuit — actual presence is determined by + // the per-chunk size field inside each bucket. + let mut bitmap_raw = [0u8; ChunkBitmap::SIZE]; + bitmap_raw.copy_from_slice(&buf.split_to(ChunkBitmap::SIZE)); + let _bitmap = ChunkBitmap(bitmap_raw); - // Parse the header - let file_header = LinearFileHeader::from_bytes(header_bytes); - file_header.check_version()?; + let _features = NbtFeatures::from_bytes(&mut buf)?; - let Some((mut raw_file_bytes, signature)) = - raw_file_bytes.split_at_checked(file_header.chunks_bytes) - else { + let total_bucket_meta = bucket_count * BucketSizeEntry::SIZE; + if buf.remaining() < total_bucket_meta { return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); - }; - - Self::check_signature(signature)?; - - let mut decoder = StreamingDecoder::new(&mut raw_file_bytes) - .map_err(|_| ChunkReadingError::RegionIsInvalid)?; - - let mut buffer = Vec::new(); - decoder - .read_to_end(&mut buffer) - .map_err(|err| ChunkReadingError::IoError(err.kind()))?; - let mut buffer: Bytes = buffer.into(); - - let headers_buffer = buffer.split_to(LinearChunkHeader::CHUNK_HEADER_SIZE * CHUNK_COUNT); - - // Parse the chunk headers - let chunk_headers: [LinearChunkHeader; CHUNK_COUNT] = headers_buffer - .chunks_exact(8) - .map(LinearChunkHeader::from_bytes) - .collect::>() - .try_into() - .map_err(|_| ChunkReadingError::InvalidHeader)?; - - // Check if the total bytes of the chunks match the header - let total_bytes = chunk_headers.iter().map(|header| header.size).sum::() as usize; - if buffer.len() != total_bytes { - error!( - "Invalid total bytes of the chunks {} != {}", - total_bytes, - buffer.len(), - ); - return Err(ChunkReadingError::InvalidHeader); + } + let mut bucket_entries: Vec = Vec::with_capacity(bucket_count); + for _ in 0..bucket_count { + bucket_entries.push(BucketSizeEntry::from_bytes(&mut buf)?); } - let mut chunks = [const { None }; CHUNK_COUNT]; - let mut bytes_offset = 0; - for (i, header) in chunk_headers.iter().enumerate() { - if header.size != 0 { - let last_index = bytes_offset; - bytes_offset += header.size as usize; - if bytes_offset > buffer.len() { - warn!( - "Not enough bytes are available for chunk {} ({} vs {})", - i, - header.size, - buffer.len() - last_index - ); - } else { - chunks[i] = Some(buffer.slice(last_index..bytes_offset)); - } + // Work out where the footer starts so we can validate it without + // consuming the bucket data bytes yet. + let total_compressed: usize = bucket_entries.iter().map(|e| e.size as usize).sum(); + if buf.remaining() < total_compressed + SIGNATURE.len() { + return Err(ChunkReadingError::IoError(ErrorKind::UnexpectedEof)); + } + let footer_offset = total_compressed; + { + let footer = &buf[footer_offset..footer_offset + SIGNATURE.len()]; + if footer != SIGNATURE { + error!("Linear v2: invalid footer signature"); + return Err(ChunkReadingError::InvalidHeader); + } + } + + let mut timestamps = [0u64; CHUNK_COUNT]; + let mut chunks_data: [Option; CHUNK_COUNT] = [const { None }; CHUNK_COUNT]; + + for (bucket_idx, entry) in bucket_entries.iter().enumerate() { + let compressed_size = entry.size as usize; + + // xxhash64 integrity check. + let actual_hash = xxh64(&buf[..compressed_size], 0); + if actual_hash != entry.xxhash { + error!( + "Linear v2: xxhash mismatch for bucket {bucket_idx} \ + (expected {:#x}, got {:#x})", + entry.xxhash, actual_hash + ); + // Skip the corrupted bucket rather than aborting the whole + // file — the remaining buckets may be intact. + buf.advance(compressed_size); + continue; + } + + let compressed_slice = &buf[..compressed_size]; + let mut decompressed = Vec::new(); + { + let mut decoder = StreamingDecoder::new(compressed_slice) + .map_err(|_| ChunkReadingError::RegionIsInvalid)?; + decoder + .read_to_end(&mut decompressed) + .map_err(|e| ChunkReadingError::IoError(e.kind()))?; + } + buf.advance(compressed_size); + + let mut bucket_buf: Bytes = decompressed.into(); + let cpb = Self::chunks_per_bucket(grid_size); + + for local in 0..cpb { + let chunk_index = Self::global_chunk_index(bucket_idx, local, grid_size); + let record = BucketChunkEntry::read_from(&mut bucket_buf)?; + timestamps[chunk_index] = record.timestamp; + chunks_data[chunk_index] = record.data; } } Ok(Self { - chunks_headers: chunk_headers, - chunks_data: chunks, - _dummy: Default::default(), + region_x: superblock.region_x, + region_z: superblock.region_z, + grid_size, + timestamps, + chunks_data, + _dummy: PhantomData, }) } @@ -320,16 +547,11 @@ impl ChunkSerializer for LinearFile { .await .map_err(|err| ChunkWritingError::ChunkSerializingError(err.to_string()))?; - let header = &mut self.chunks_headers[index]; - header.size = chunk_raw.len() as u32; - header.timestamp = SystemTime::now() + self.timestamps[index] = SystemTime::now() .duration_since(UNIX_EPOCH) - .map(|d| d.as_secs() as u32) + .map(|d| d.as_secs()) .unwrap_or(0); - - // We update the data buffer self.chunks_data[index] = Some(chunk_raw); - Ok(()) } @@ -338,208 +560,177 @@ impl ChunkSerializer for LinearFile { chunks: Vec>, stream: tokio::sync::mpsc::Sender>, ) { - // Don't par iter here so we can prevent backpressure with the await in the async - // runtime for chunk in chunks { let index = Self::get_chunk_index(chunk.x, chunk.y); - let linear_chunk_data = &self.chunks_data[index]; - let result = if let Some(data) = linear_chunk_data { - match S::from_bytes(data, chunk) { - Ok(chunk) => LoadedData::Loaded(chunk), + let result = match &self.chunks_data[index] { + Some(data) => match S::from_bytes(data, chunk) { + Ok(c) => LoadedData::Loaded(c), Err(err) => LoadedData::Error((chunk, err)), - } - } else { - LoadedData::Missing(chunk) + }, + None => LoadedData::Missing(chunk), }; if stream.send(result).await.is_err() { - // The stream is closed. Return early to prevent unneeded work and IO + // Receiver dropped — stop early to avoid unnecessary work. return; } } } } -/* #[cfg(test)] mod tests { + use super::*; - use core::panic; - use pumpkin_data::BlockDirection; - use pumpkin_util::math::position::BlockPos; - use pumpkin_util::math::vector2::Vector2; - use std::fs; - use std::path::PathBuf; - use std::sync::Arc; - use temp_dir::TempDir; - use tokio::sync::RwLock; + // Helper: build a tiny chunk payload. + fn fake_nbt(seed: u8) -> Bytes { + Bytes::from(vec![seed; 64]) + } - use crate::chunk::ChunkData; - use crate::chunk::format::linear::LinearFile; - use crate::chunk::io::file_manager::ChunkFileManager; - use crate::chunk::io::{FileIO, LoadedData}; - use crate::dimension::Dimension; - use crate::generation::{Seed, get_world_gen}; - use crate::level::{Level, LevelFolder}; - use crate::world::{BlockAccessor, BlockRegistryExt}; + #[test] + fn bitmap_round_trip() { + let mut bm = ChunkBitmap::new(); + bm.set(0, true); + bm.set(7, true); + bm.set(1023, true); + assert!(bm.get(0)); + assert!(!bm.get(1)); + assert!(bm.get(7)); + assert!(!bm.get(8)); + assert!(bm.get(1023)); + assert!(!bm.get(1022)); + } - struct BlockRegistry; + #[test] + fn nbt_features_empty_round_trip() { + let f = NbtFeatures::empty(); + let encoded = f.to_bytes(); + assert_eq!(encoded, vec![0u8]); // just the end marker - impl BlockRegistryExt for BlockRegistry { - fn can_place_at( - &self, - _block: &pumpkin_data::Block, - _block_accessor: &dyn BlockAccessor, - _block_pos: &BlockPos, - _face: BlockDirection, - ) -> bool { - true + let mut buf: Bytes = encoded.into(); + let decoded = NbtFeatures::from_bytes(&mut buf).unwrap(); + assert!(decoded.0.is_empty()); + } + + #[test] + fn nbt_features_round_trip() { + let mut f = NbtFeatures::empty(); + f.0.insert("version".into(), 42); + let encoded = f.to_bytes(); + let mut buf: Bytes = encoded.into(); + let decoded = NbtFeatures::from_bytes(&mut buf).unwrap(); + assert_eq!(decoded.0.get("version"), Some(&42u32)); + } + + fn chunk_bucket_index(i: usize, gs: u8) -> usize { + let cx = i % 32; + let cz = i / 32; + let stride = 32 / gs as usize; + (cz / stride) * gs as usize + (cx / stride) + } + + fn chunk_local_index(i: usize, gs: u8) -> usize { + let cx = i % 32; + let cz = i / 32; + let stride = 32 / gs as usize; + (cz % stride) * stride + (cx % stride) + } + + fn global_chunk_index(bucket: usize, local: usize, gs: u8) -> usize { + let stride = 32 / gs as usize; + let brow = bucket / gs as usize; + let bcol = bucket % gs as usize; + let lrow = local / stride; + let lcol = local % stride; + let cz = brow * stride + lrow; + let cx = bcol * stride + lcol; + cz * 32 + cx + } + + #[test] + fn index_round_trip_grid2() { + for i in 0..CHUNK_COUNT { + let b = chunk_bucket_index(i, 2); + let l = chunk_local_index(i, 2); + let recovered = global_chunk_index(b, l, 2); + assert_eq!(i, recovered, "chunk index {i} round-trip failed"); } } - #[tokio::test(flavor = "multi_thread")] - async fn not_existing() { - let region_path = PathBuf::from("not_existing"); - let chunk_saver = ChunkFileManager::>::default(); - - let mut chunks = Vec::new(); - let (send, mut recv) = tokio::sync::mpsc::channel(1); - - chunk_saver - .fetch_chunks( - &LevelFolder { - root_folder: PathBuf::from(""), - region_folder: region_path, - entities_folder: PathBuf::from(""), - }, - &[Vector2::new(0, 0)], - send, - ) - .await; - - while let Some(data) = recv.recv().await { - chunks.push(data); + #[test] + fn index_round_trip_grid4() { + for i in 0..CHUNK_COUNT { + let b = chunk_bucket_index(i, 4); + let l = chunk_local_index(i, 4); + let recovered = global_chunk_index(b, l, 4); + assert_eq!(i, recovered, "chunk index {i} round-trip failed"); } - - assert!(chunks.len() == 1 && matches!(chunks[0], LoadedData::Missing(_))); } - #[tokio::test(flavor = "multi_thread")] - async fn writing() { - let _ = env_logger::try_init(); + #[test] + fn index_round_trip_grid1() { + for i in 0..CHUNK_COUNT { + let b = chunk_bucket_index(i, 1); + let l = chunk_local_index(i, 1); + let recovered = global_chunk_index(b, l, 1); + assert_eq!(i, recovered, "chunk index {i} round-trip failed"); + } + } - let generator = get_world_gen(Seed(0), Dimension::Overworld); - - let temp_dir = TempDir::new().unwrap(); - let level_folder = LevelFolder { - root_folder: temp_dir.path().to_path_buf(), - region_folder: temp_dir.path().join("region"), - entities_folder: PathBuf::from("entities"), + #[test] + fn superblock_round_trip() { + let sb = LinearV2Superblock { + newest_timestamp: 0xDEADBEEFCAFEBABE, + grid_size: 4, + region_x: -7, + region_z: 13, }; - fs::create_dir(&level_folder.region_folder).expect("couldn't create region folder"); - let chunk_saver = ChunkFileManager::>::default(); - let block_registry = Arc::new(BlockRegistry); - // Generate chunks - let mut chunks = vec![]; - let level = Arc::new(Level::from_root_folder( - temp_dir.path().to_path_buf(), - block_registry.clone(), - 0, - Dimension::Overworld, - )); - for x in -5..5 { - for y in -5..5 { - let position = Vector2::new(x, y); - let chunk = generator.generate_chunk(&level, block_registry.as_ref(), &position); - chunks.push((position, Arc::new(RwLock::new(chunk)))); - } + let encoded = sb.to_bytes(); + let decoded = LinearV2Superblock::from_bytes(&encoded).unwrap(); + assert_eq!(decoded.newest_timestamp, sb.newest_timestamp); + assert_eq!(decoded.grid_size, 4); + assert_eq!(decoded.region_x, -7); + assert_eq!(decoded.region_z, 13); + } + + #[test] + fn superblock_rejects_bad_signature() { + let mut bytes = LinearV2Superblock { + newest_timestamp: 0, + grid_size: 2, + region_x: 0, + region_z: 0, } + .to_bytes(); + bytes[0] ^= 0xFF; // corrupt the signature + assert!(LinearV2Superblock::from_bytes(&bytes).is_err()); + } - let section_count = chunks[0].1.read().await.section.sections.len(); + #[test] + fn bucket_entry_present_round_trip() { + let entry = BucketChunkEntry { + timestamp: 12345, + data: Some(fake_nbt(0xAB)), + }; + let mut buf = Vec::new(); + entry.write_into(&mut buf); + let mut b: Bytes = buf.into(); + let decoded = BucketChunkEntry::read_from(&mut b).unwrap(); + assert_eq!(decoded.timestamp, 12345); + assert_eq!(decoded.data.unwrap().as_ref(), &[0xABu8; 64][..]); + } - for i in 0..5 { - println!("Iteration {}", i + 1); - // Mark the chunks as dirty so we save them again - for (_, chunk) in &chunks { - let mut chunk = chunk.write().await; - chunk.dirty = true; - } - - chunk_saver - .save_chunks( - &level_folder, - chunks.clone().into_iter().collect::>(), - ) - .await - .expect("Failed to write chunk"); - - let mut read_chunks = Vec::new(); - let (send, mut recv) = tokio::sync::mpsc::channel(1); - - let chunk_pos = chunks.iter().map(|(at, _)| *at).collect::>(); - let spawn = chunk_saver.fetch_chunks(&level_folder, &chunk_pos, send); - - let collect = async { - while let Some(data) = recv.recv().await { - read_chunks.push(data); - } - }; - - tokio::join!(spawn, collect); - - let read_chunks = read_chunks - .into_iter() - .map(|chunk| match chunk { - LoadedData::Loaded(chunk) => chunk, - LoadedData::Missing(_) => panic!("Missing chunk"), - LoadedData::Error((position, error)) => { - panic!("Error reading chunk at {:?} | Error: {:?}", position, error) - } - }) - .collect::>(); - - for (_, chunk) in &chunks { - let chunk = chunk.read().await; - for read_chunk in read_chunks.iter() { - let read_chunk = read_chunk.read().await; - - // Before this commit the chunks got an extra section after saving and reading - // so we prevent that from happening in the future :) - assert_eq!(read_chunk.section.sections.len(), section_count); - if read_chunk.position == chunk.position { - let original = chunk.section.dump_blocks(); - let read = read_chunk.section.dump_blocks(); - - original - .into_iter() - .zip(read) - .enumerate() - .for_each(|(i, (o, r))| { - if o != r { - panic!("Data miss-match expected {}, got {} ({})", o, r, i); - } - }); - - let original = chunk.section.dump_biomes(); - let read = read_chunk.section.dump_biomes(); - - original - .into_iter() - .zip(read) - .enumerate() - .for_each(|(i, (o, r))| { - if o != r { - panic!("Data miss-match expected {}, got {} ({})", o, r, i); - } - }); - break; - } - } - } - } - - println!("Checked chunks successfully"); + #[test] + fn bucket_entry_absent_round_trip() { + let entry = BucketChunkEntry { + timestamp: 0, + data: None, + }; + let mut buf = Vec::new(); + entry.write_into(&mut buf); + let mut b: Bytes = buf.into(); + let decoded = BucketChunkEntry::read_from(&mut b).unwrap(); + assert!(decoded.data.is_none()); } } -*/ diff --git a/pumpkin-world/src/level.rs b/pumpkin-world/src/level.rs index 764c8bd05..78b1d1a89 100644 --- a/pumpkin-world/src/level.rs +++ b/pumpkin-world/src/level.rs @@ -1,3 +1,4 @@ +use crate::chunk::format::linear::LinearV2File; use crate::chunk_system::{ChunkListener, ChunkLoading, GenerationSchedule, LevelChannel}; use crate::generation::generator::VanillaGenerator; use crate::lighting::DynamicLightEngine; @@ -6,7 +7,7 @@ use crate::{ block::{RawBlockState, entities::BlockEntity}, chunk::{ ChunkData, ChunkEntityData, ChunkReadingError, - format::{anvil::AnvilChunkFile, linear::LinearFile}, + format::anvil::AnvilChunkFile, io::{Dirtiable, FileIO, LoadedData, file_manager::ChunkFileManager}, palette::has_random_ticking_fluid, }, @@ -163,16 +164,16 @@ impl Level { let chunk_saver: Arc> = match &level_config.chunk { ChunkConfig::Linear(config) => Arc::new( - ChunkFileManager::>::new(config.clone()), + ChunkFileManager::>::new(config.clone()), ), ChunkConfig::Anvil(config) => Arc::new( ChunkFileManager::>::new(config.clone()), ), }; let entity_saver: Arc> = match &level_config.chunk { - ChunkConfig::Linear(config) => Arc::new( - ChunkFileManager::>::new(config.clone()), - ), + ChunkConfig::Linear(config) => Arc::new(ChunkFileManager::< + LinearV2File, + >::new(config.clone())), ChunkConfig::Anvil(config) => Arc::new(ChunkFileManager::< AnvilChunkFile, >::new(config.clone())),