feat(server): add chunk loading pipeline

crates/server/src/main.rs

@@ -73,8 +73,11 @@ fn main() -> anyhow::Result<()> {
 
     let generator = VoxelGenerator::new(worldgen_config, seed);
 
+    // The region directory holds the `.region` save files for this world; a later slice will resolve it per named world under the platform user-data directory.
+    let region_dir = std::path::PathBuf::from("saves/default/region");
+
     let mut world = World::new();
-    world.insert_resource(ServerWorld::new(generator));
+    world.insert_resource(ServerWorld::new(generator, region_dir));
 
     // Spawn a single dummy player anchor at the world origin.
     world.spawn((

crates/server/src/save.rs

@@ -8,6 +8,8 @@
 //! (`.tmp` + fsync + rename); the on-disk format is unchanged, so a later slice can switch to an
 //! append-in-place strategy without a format change.
 
+mod region_actor;
 mod region_file;
 
+pub use region_actor::{SaveActor, SaveRequest};
 pub use region_file::{REGION_SIZE, RegionFile, region_coords, region_path};

crates/server/src/save/region_actor.rs

@@ -0,0 +1,83 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//! A dedicated thread that owns every open region file and answers load requests over a channel.
+
+use std::collections::HashMap;
+use std::collections::hash_map::Entry;
+use std::path::{Path, PathBuf};
+use std::thread::{self, JoinHandle};
+
+use crossbeam_channel::{Receiver, Sender};
+use shared::save::SaveError;
+use shared::world::{ChunkData, ChunkPos};
+
+use super::region_file::{RegionFile, region_coords, region_path};
+
+/// A request sent to the save actor. Each variant carries its own one-shot reply channel.
+pub enum SaveRequest {
+    /// Reads the stored chunk at a position, replying with the saved modification if one exists.
+    Read {
+        /// The chunk position to look up.
+        pos: ChunkPos,
+        /// The one-shot channel the actor replies on: `Ok(Some(data))` for a saved modification, `Ok(None)` when the chunk was never modified, or `Err` on a save-layer failure.
+        reply: Sender<Result<Option<ChunkData>, SaveError>>,
+    },
+}
+
+/// A handle to the running save actor: the request sender plus the owning thread's join handle.
+pub struct SaveActor {
+    /// The sending end of the request channel; cloned into every worker so it can issue reads.
+    request_tx: Sender<SaveRequest>,
+    /// The actor thread handle, retained so it can be joined on shutdown.
+    // * Retained ahead of a dedicated shutdown path; not yet read because the server has no graceful-stop sequence.
+    #[expect(dead_code)]
+    handle: JoinHandle<()>,
+}
+
+impl SaveActor {
+    /// Spawns the actor thread, which owns the region files beneath `region_dir` for its lifetime.
+    #[must_use]
+    pub fn spawn(region_dir: PathBuf) -> Self {
+        let (request_tx, request_rx) = crossbeam_channel::unbounded::<SaveRequest>();
+        let handle = thread::spawn(move || actor_loop(&region_dir, &request_rx));
+        Self { request_tx, handle }
+    }
+
+    /// Returns a fresh sender for a worker to issue requests through.
+    #[must_use]
+    pub fn sender(&self) -> Sender<SaveRequest> {
+        self.request_tx.clone()
+    }
+}
+
+/// The actor's run loop: it owns the region-file map and answers requests until every sender is dropped.
+fn actor_loop(region_dir: &Path, request_rx: &Receiver<SaveRequest>) {
+    // The actor is the sole owner of this map, so region files need no lock of their own.
+    let mut regions: HashMap<(i32, i32), RegionFile> = HashMap::new();
+    while let Ok(request) = request_rx.recv() {
+        match request {
+            SaveRequest::Read { pos, reply } => {
+                let result = read_chunk(&mut regions, region_dir, pos);
+                // A send error means the requesting worker has gone away; the reply is simply dropped.
+                let _ = reply.send(result);
+            }
+        }
+    }
+}
+
+/// Reads the stored chunk at `pos`, opening and caching its region file on first access.
+fn read_chunk(
+    regions: &mut HashMap<(i32, i32), RegionFile>,
+    region_dir: &Path,
+    pos: ChunkPos,
+) -> Result<Option<ChunkData>, SaveError> {
+    let key = region_coords(pos.x, pos.z);
+    // The region file is opened once on first touch; every later read of a chunk in it hits the in-memory copy.
+    let region = match regions.entry(key) {
+        Entry::Occupied(entry) => entry.into_mut(),
+        Entry::Vacant(entry) => {
+            entry.insert(RegionFile::open(region_path(region_dir, pos.x, pos.z))?)
+        }
+    };
+    region.read_chunk(pos)
+}

crates/server/src/world_server.rs

@@ -6,15 +6,20 @@ use bevy_ecs::prelude::Resource;
 use crossbeam_channel::{Receiver, Sender};
 use shared::{
     generator::VoxelGenerator,
-    world::{Chunk, ChunkPos},
+    save::SaveError,
+    world::{Chunk, ChunkData, ChunkPos},
 };
 use std::{
     collections::{HashMap, HashSet},
+    path::PathBuf,
     sync::Arc,
     thread::JoinHandle,
 };
+use tracing::warn;
 
-/// Outcome of a single streaming reconcile pass, for logging and tests.
+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.
@@ -32,12 +37,16 @@ pub struct StreamStats {
 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 that require generation.
+    /// 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 from being re-dispatched on subsequent passes.
+    /// 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.
+    // Retained so its request channel stays open for the workers; not read again after construction.
+    #[expect(dead_code)]
+    save_actor: SaveActor,
     /// Handles to the generation worker threads, retained so they can be joined on shutdown.
     // Retained ahead of a dedicated shutdown path; not yet read because the server has no graceful-stop sequence.
     #[expect(dead_code)]
@@ -45,28 +54,31 @@ pub struct ServerWorld {
 }
 
 impl ServerWorld {
-    /// Initializes a new authoritative server world with the provided generator.
+    /// Initializes a new authoritative server world with the provided generator, resolving chunk loads against region files under `region_dir`.
     #[must_use]
-    pub fn new(generator: VoxelGenerator) -> Self {
+    pub fn new(generator: VoxelGenerator, region_dir: PathBuf) -> 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)>();
 
-        // One worker per logical core, falling back to a small fixed pool if the platform cannot report its parallelism.
+        // The actor owns every region file; workers reach it only through cloned request senders.
+        let save_actor = SaveActor::spawn(region_dir);
+
         let worker_count = std::thread::available_parallelism().map_or(4, std::num::NonZero::get);
 
         let workers = (0..worker_count)
             .map(|_| {
-                // Each worker owns its own clones: a shared handle to the read-only generator, its own view of the shared job queue, and its own sender back into the result channel.
+                // 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, and its own request sender to the save actor.
                 let generator = Arc::clone(&generator);
                 let job_rx = job_rx.clone();
                 let result_tx = result_tx.clone();
+                let save_tx = save_actor.sender();
 
                 std::thread::spawn(move || {
                     // Block until a job arrives.
                     while let Ok(pos) = job_rx.recv() {
-                        let chunk = generator.generate_chunk(pos);
+                        let chunk = load_chunk(&generator, &save_tx, pos);
-                        // A send error means the main thread has gone away; nothing left to do but let the worker wind down.
+                        // 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;
                         }
@@ -75,7 +87,7 @@ impl ServerWorld {
             })
             .collect();
 
-        // Drop the template ends left over after cloning.
+        // Drop the template ends left over after cloning so the channels close once the real holders are gone.
         drop(job_rx);
         drop(result_tx);
 
@@ -84,6 +96,7 @@ impl ServerWorld {
             job_tx,
             result_rx,
             in_flight: HashSet::new(),
+            save_actor,
             workers,
         }
     }
@@ -94,33 +107,38 @@ impl ServerWorld {
         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 generation work, i.e. every dispatched chunk has been returned. A loading gate can poll this to decide when an initial region has finished streaming.
+    /// 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 a 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.
+    /// 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 have finished since the last pass.
+        // 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() {
-            // The position is no longer dispatched now that its chunk has returned.
             self.in_flight.remove(&pos);
-            // Guard against the eviction race: the anchor may have moved away while this chunk was generating, so a returned chunk is only kept if it is still wanted.
+            // 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 that no anchor wants any more.
+        // Evict: drop resident chunks no anchor wants any more.
         let stale: Vec<ChunkPos> = self
             .chunks
             .keys()
@@ -131,11 +149,11 @@ impl ServerWorld {
             self.chunks.remove(pos);
         }
 
-        // Dispatch: request generation for every wanted position that is neither resident nor already in flight.
+        // 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 every worker has shut down; there is nothing useful to do with the position, so the failure is ignored.
+                // 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);
             }
         }
@@ -149,6 +167,43 @@ impl ServerWorld {
     }
 }
 
+/// 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>, 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(&generator.generate_chunk(pos))
+        }
+        // The chunk was never modified, so its content is exactly the deterministic baseline.
+        Ok(None) => generator.generate_chunk(pos),
+        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");
+            generator.generate_chunk(pos)
+        }
+    }
+}
+
+/// 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.
+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,
@@ -172,14 +227,17 @@ pub fn cylinder_chunks<S: std::hash::BuildHasher>(
 
 #[cfg(test)]
 mod tests {
-    use super::{ChunkPos, ServerWorld, StreamStats, cylinder_chunks};
+    use super::{ServerWorld, StreamStats, cylinder_chunks};
     use shared::generator::{VoxelGenerator, WorldGenConfig};
-    use shared::world::BlockId;
+    use shared::save::SaveError;
+    use shared::world::{BlockId, ChunkData, ChunkPos};
     use std::collections::HashSet;
     use std::time::{Duration, Instant};
 
-    /// Builds a server world backed by a real worker pool for streaming tests.
+    use crate::save::{RegionFile, region_path};
-    fn test_world() -> ServerWorld {
+
+    /// Builds a generator with a small, cheap terrain configuration for streaming tests.
+    fn test_generator() -> VoxelGenerator {
         let config = WorldGenConfig {
             base_height: 8,
             noise_scale: 0.05,
@@ -187,10 +245,15 @@ mod tests {
             subsurface_block: BlockId(2),
             stone_block: BlockId(3),
         };
-        ServerWorld::new(VoxelGenerator::new(config, 42))
+        VoxelGenerator::new(config, 42)
     }
 
-    /// Repeatedly reconciles against `desired` until the worker pool reports no outstanding work, returning the final pass's stats. Fails the test if the pool does not drain within a fixed timeout.
+    /// Builds a server world whose saves resolve against `region_dir`.
+    fn test_world(region_dir: std::path::PathBuf) -> ServerWorld {
+        ServerWorld::new(test_generator(), region_dir)
+    }
+
+    /// Repeatedly reconciles `desired` until the worker pool reports no outstanding work, returning the final pass's stats. Fails the test if the pool does not drain within a fixed timeout.
     fn drain_to_idle(world: &mut ServerWorld, desired: &HashSet<ChunkPos>) -> StreamStats {
         let deadline = Instant::now() + Duration::from_secs(5);
         loop {
@@ -207,12 +270,14 @@ mod tests {
     }
 
     #[test]
-    fn reconcile_converges_over_multiple_passes() {
+    fn reconcile_converges_over_multiple_passes() -> Result<(), SaveError> {
-        let mut world = test_world();
+        // A fresh empty directory means every load is a miss and resolves to the baseline.
+        let dir = tempfile::tempdir()?;
+        let mut world = test_world(dir.path().to_path_buf());
         let mut desired = HashSet::new();
         cylinder_chunks(ChunkPos::new(0, 0, 0), 2, &mut desired);
 
-        // The first pass only dispatches work; because generation is off-thread, nothing is resident yet and every position is in flight.
+        // The first pass only dispatches work; because loading is off-thread, nothing is resident yet and every position is in flight.
         let first = world.reconcile(&desired);
         assert_eq!(first.loaded, 0);
         assert_eq!(first.resident, 0);
@@ -222,11 +287,13 @@ mod tests {
         let final_stats = drain_to_idle(&mut world, &desired);
         assert_eq!(final_stats.in_flight, 0);
         assert_eq!(final_stats.resident, desired.len());
+        Ok(())
     }
 
     #[test]
-    fn evicted_chunk_is_not_repopulated_on_arrival() {
+    fn evicted_chunk_is_not_repopulated_on_arrival() -> Result<(), SaveError> {
-        let mut world = test_world();
+        let dir = tempfile::tempdir()?;
+        let mut world = test_world(dir.path().to_path_buf());
         let target = ChunkPos::new(0, 0, 0);
         let mut desired = HashSet::new();
         desired.insert(target);
@@ -234,28 +301,54 @@ mod tests {
         // Dispatch the chunk, then immediately stop wanting it.
         world.reconcile(&desired);
 
-        // Every subsequent pass reconciles against an empty desired set, so whenever the worker returns the chunk the drain guard discards it rather than resurrecting an unwanted chunk.
+        // Every subsequent pass reconciles against an empty desired set, so the finished chunk is discarded on arrival rather than inserted.
         let empty = HashSet::new();
         let final_stats = drain_to_idle(&mut world, &empty);
         assert_eq!(final_stats.in_flight, 0);
-        assert_eq!(
+        assert_eq!(final_stats.resident, 0);
-            final_stats.resident, 0,
+        Ok(())
-            "a chunk that is no longer wanted must not become resident when it arrives"
-        );
     }
 
     #[test]
-    fn cylinder_is_symmetric_and_bounded() {
+    fn saved_modification_is_applied_over_baseline() -> Result<(), SaveError> {
+        // A modified chunk is written to disk, then streamed back; the resident chunk must show the edit rather than the bare baseline.
+        let dir = tempfile::tempdir()?;
+        let pos = ChunkPos::new(0, 0, 0);
+        let edited_index = 100u32;
+        let edited_block = BlockId(999);
+
+        let mut data = ChunkData::new(pos, 0);
+        data.set(edited_index, edited_block);
+
+        let mut region = RegionFile::open(region_path(dir.path(), pos.x, pos.z))?;
+        region.write_chunk(pos, &data, 0)?;
+        region.save()?;
+
+        let mut world = test_world(dir.path().to_path_buf());
+        let mut desired = HashSet::new();
+        desired.insert(pos);
+        drain_to_idle(&mut world, &desired);
+
+        // The resident chunk must carry the stored edit layered over its regenerated baseline.
+        assert!(
+            world
+                .chunk(pos)
+                .is_some_and(|chunk| chunk.blocks[edited_index as usize] == edited_block)
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn cylinder_contains_expected_columns() {
         let mut set = HashSet::new();
         cylinder_chunks(ChunkPos::new(0, 0, 0), 2, &mut set);
 
-        // Center column is always included.
         assert!(set.contains(&ChunkPos::new(0, 0, 0)));
-        // A corner outside the disc (dx=2, dz=2 -> 8 > 4) is excluded.
+        // A corner cell is outside the disc (dx=2, dz=2 -> 8 > 4).
         assert!(!set.contains(&ChunkPos::new(2, 0, 2)));
         // An axis cell at exactly the radius is included (dx=2, dz=0 -> 4 == 4).
         assert!(set.contains(&ChunkPos::new(2, 0, 0)));
-        // Vertical extent is radius/2 = 1, so y=2 is out of range.
+        // The vertical extent is radius/2 = 1, so y=2 is out of range.
         assert!(!set.contains(&ChunkPos::new(0, 2, 0)));
         assert!(set.contains(&ChunkPos::new(0, 1, 0)));
     }
@@ -268,7 +361,7 @@ mod tests {
         let mut shifted = HashSet::new();
         cylinder_chunks(ChunkPos::new(10, 0, -5), 3, &mut shifted);
 
-        // Shape is translation-invariant: same count regardless of center.
+        // The shape is translation-invariant: the same count regardless of center.
         assert_eq!(origin.len(), shifted.len());
     }
 }