Synvael/crates/server/src/world_server.rs

// 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<ChunkPos, Chunk>,
    /// Sending end of the job channel; the main thread pushes positions to load.
    job_tx: Sender<ChunkPos>,
    /// 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<ChunkPos>,
    /// 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<JoinHandle<()>>,
    /// The same read-only generator handle the workers share, held so eviction can regenerate a chunk's baseline to diff against.
    generator: Arc<VoxelGenerator>,
    /// 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<SaveRequest>,
}

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::<ChunkPos>();
        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<ChunkPos>) -> 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<ChunkPos> = 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<SaveRequest>,
    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<SaveRequest>,
    pos: ChunkPos,
) -> Result<Option<ChunkData>, 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<S: std::hash::BuildHasher>(
    center: ChunkPos,
    radius: i32,
    out: &mut HashSet<ChunkPos, S>,
) {
    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;