// SPDX-License-Identifier: AGPL-3.0-only //! Authoritative chunk storage and generation logic for the server. use bevy_ecs::prelude::Resource; use crossbeam_channel::{Receiver, Sender}; use shared::{ generator::VoxelGenerator, save::SaveError, world::{Chunk, ChunkData, ChunkPos}, }; use std::{ collections::{HashMap, HashSet}, num::NonZeroUsize, path::PathBuf, sync::Arc, thread::JoinHandle, }; use tracing::warn; use crate::chunk_cache::ChunkCache; use crate::save::{SaveActor, SaveRequest}; /// Outcome of a single streaming reconcile pass, surfaced for logging and tests. #[derive(Copy, Clone, Debug, PartialEq, Eq)] pub struct StreamStats { /// Number of chunks generated and inserted this pass. pub loaded: usize, /// Number of chunks evicted this pass. pub unloaded: usize, /// Number of chunks resident after the pass. pub resident: usize, /// Number of chunks in flight. pub in_flight: usize, } /// The server's authoritative representation of the world. #[derive(Resource)] pub struct ServerWorld { /// Currently resident chunks, keyed by chunk-space position. chunks: HashMap, /// Sending end of the job channel; the main thread pushes positions to load. job_tx: Sender, /// Receiving end of the result channel; the main thread drains finished chunks returned by workers. result_rx: Receiver<(ChunkPos, Chunk)>, /// Positions dispatched to a worker but not yet returned, preventing the same chunk being re-dispatched on subsequent passes. in_flight: HashSet, /// The dedicated thread owning all region files, kept alive for the world's lifetime. #[expect( dead_code, reason = "retained to keep the save request channel open for the workers" )] save_actor: SaveActor, /// Handles to the generation worker threads, retained so they can be joined on shutdown. // TODO: Remove once the server has a graceful-stop sequence #[expect( dead_code, reason = "retained for a future graceful-shutdown join path" )] workers: Vec>, /// The same read-only generator handle the workers share, held so eviction can regenerate a chunk's baseline to diff against. generator: Arc, /// A handle onto the shared baseline cache, used to resolve the baseline during the unload diff. cache: ChunkCache, /// The save actor's request sender, used to issue `Write` and `Remove` on unload. save_tx: Sender, } impl ServerWorld { /// Initializes a new authoritative server world with the provided generator, resolving chunk loads against region files under `region_dir` and caching up to `cache_capacity` regenerated baselines. #[must_use] pub fn new( generator: VoxelGenerator, region_dir: PathBuf, cache_capacity: NonZeroUsize, ) -> Self { let generator = Arc::new(generator); let (job_tx, job_rx) = crossbeam_channel::unbounded::(); let (result_tx, result_rx) = crossbeam_channel::unbounded::<(ChunkPos, Chunk)>(); // The actor owns every region file; workers reach it only through cloned request senders. let save_actor = SaveActor::spawn(region_dir); // Baselines are shared across the pool through cloned handles onto one bounded store. let cache = ChunkCache::new(cache_capacity); let worker_count = std::thread::available_parallelism().map_or(4, std::num::NonZero::get); let workers = (0..worker_count) .map(|_| { // Each worker shares a handle to the read-only generator, its own view of the shared job queue, its own sender back into the result channel, its own request sender to the save actor, and a handle onto the shared baseline cache. let generator = Arc::clone(&generator); let job_rx = job_rx.clone(); let result_tx = result_tx.clone(); let save_tx = save_actor.sender(); let cache = cache.clone(); std::thread::spawn(move || { // Block until a job arrives. while let Ok(pos) = job_rx.recv() { let chunk = load_chunk(&generator, &save_tx, &cache, pos); // A send error means the main thread has gone away; nothing is left to do but let the worker wind down. if result_tx.send((pos, chunk)).is_err() { break; } } }) }) .collect(); // Drop the template ends left over after cloning so the channels close once the real holders are gone. drop(job_rx); drop(result_tx); let save_tx = save_actor.sender(); Self { chunks: HashMap::new(), job_tx, result_rx, in_flight: HashSet::new(), save_actor, workers, generator, cache, save_tx, } } /// Number of chunks currently resident in memory. #[must_use] pub fn loaded_count(&self) -> usize { self.chunks.len() } /// Returns the resident chunk at `pos`, or `None` if it is not currently loaded. #[must_use] pub fn chunk(&self, pos: ChunkPos) -> Option<&Chunk> { self.chunks.get(&pos) } /// Number of chunks dispatched to the worker pool but not yet returned. #[must_use] pub fn in_flight_count(&self) -> usize { self.in_flight.len() } /// Returns `true` when the worker pool has no outstanding work, i.e. every dispatched chunk has been returned. The loading gate polls this to decide when the initial region has finished streaming. #[must_use] pub fn streaming_idle(&self) -> bool { self.in_flight.is_empty() } /// Reconciles resident chunks against the desired set without blocking the caller: finished chunks are drained from the worker pool, resident chunks absent from `desired` are evicted, and still-missing chunks are dispatched to the pool. pub fn reconcile(&mut self, desired: &HashSet) -> StreamStats { // Drain: absorb every chunk the workers finished since the last pass. let mut loaded = 0; while let Ok((pos, chunk)) = self.result_rx.try_recv() { self.in_flight.remove(&pos); // A finished chunk is only kept if it is still wanted. if desired.contains(&pos) { self.chunks.insert(pos, chunk); loaded += 1; } } // Evict: drop resident chunks no anchor wants any more. let stale: Vec = self .chunks .keys() .filter(|pos| !desired.contains(pos)) .copied() .collect(); for pos in &stale { // The chunk is taken by value so it can be diffed against its baseline before being dropped. let Some(chunk) = self.chunks.remove(pos) else { continue; }; let baseline = self.cache.get_or_generate(*pos, &self.generator); // * NOTE: The diff is stamped with worldgen version 0: a single version exists today. This must become the chunk's stored version once worldgen versioning lands. let data = ChunkData::from_diff(*pos, 0, &baseline, &chunk); // A clean chunk drops any prior record into the region free list; a dirty chunk writes its diff back. Both only mutate the actor's in-memory image until a flush. A send error means the actor is gone, which the tick thread cannot act on. let request = if data.is_unmodified() { SaveRequest::Remove { pos: *pos } } else { SaveRequest::Write { pos: *pos, data } }; let _ = self.save_tx.send(request); } // Dispatch: request a load for every wanted position that is neither resident nor already in flight. for &pos in desired { if !self.chunks.contains_key(&pos) && !self.in_flight.contains(&pos) { self.in_flight.insert(pos); // A send error means the workers shut down; nothing useful can be done with the position, so the failure is ignored. let _ = self.job_tx.send(pos); } } StreamStats { loaded, unloaded: stale.len(), resident: self.chunks.len(), in_flight: self.in_flight.len(), } } } /// Resolves a chunk position to dense voxel data: a saved modification is applied over its baseline, otherwise the deterministic baseline is regenerated directly. fn load_chunk( generator: &VoxelGenerator, save_tx: &Sender, cache: &ChunkCache, pos: ChunkPos, ) -> Chunk { match request_saved_chunk(save_tx, pos) { Ok(Some(data)) => { // A saved modification stores only edits, so the baseline is regenerated and the edits are layered on top. // TODO: once worldgen versioning exists, the baseline must be regenerated at `data.worldgen_version()` rather than the current version; today there is a single version, so the current baseline matches. data.materialize(&cache.get_or_generate(pos, generator)) } // The chunk was never modified, so its content is exactly the deterministic baseline. Ok(None) => cache.get_or_generate(pos, generator), Err(error) => { // A save-layer failure must not wedge streaming; the chunk falls back to a fresh baseline and the error is logged. warn!(?error, ?pos, "chunk load failed; regenerating baseline"); cache.get_or_generate(pos, generator) } } } /// Sends a read request to the save actor and blocks for its reply, mapping a departed actor to an absent record so generation can still proceed. /// /// # Errors /// /// Returns the [`SaveError`] reported by the save actor if reading the stored chunk fails. A departed actor yields `Ok(None)` rather than an error. fn request_saved_chunk( save_tx: &Sender, pos: ChunkPos, ) -> Result, SaveError> { let (reply_tx, reply_rx) = crossbeam_channel::bounded(1); if save_tx .send(SaveRequest::Read { pos, reply: reply_tx, }) .is_err() { return Ok(None); } // A receive error means the actor dropped the reply end, treated the same as no saved data. reply_rx.recv().unwrap_or(Ok(None)) } /// Inserts every chunk position inside the streaming cylinder around `center` into `out`. pub fn cylinder_chunks( center: ChunkPos, radius: i32, out: &mut HashSet, ) { for x in center.x - radius..=center.x + radius { for z in center.z - radius..=center.z + radius { let dx = x - center.x; let dz = z - center.z; // Keep only the columns whose XZ distance falls within the disc. if dx * dx + dz * dz <= radius * radius { for y in center.y - radius / 2..=center.y + radius / 2 { out.insert(ChunkPos::new(x, y, z)); } } } } } #[cfg(test)] #[path = "tests/world_server.rs"] mod tests;