diff --git a/server/src/atlas/cascade.rs b/server/src/atlas/cascade.rs index b6f84c639..236826cb1 100644 --- a/server/src/atlas/cascade.rs +++ b/server/src/atlas/cascade.rs @@ -17,6 +17,7 @@ use std::path::Path; +use crate::atlas::body_world_state::{BodyWorldState, RiverNetwork}; use crate::atlas::heightmap::{self, BodyHeightmap, HeightmapLoadError}; use crate::atlas::layer1::{self, Layer1Output}; use crate::seed::SeedChain; @@ -48,6 +49,29 @@ pub struct CascadeSnapshot { pub layer1: Option, } +impl CascadeSnapshot { + /// Convert into a [`BodyWorldState`] for the D-203 cache (#968). Moves the + /// heightmap raster and the Layer-1 outputs in; `last_accessed` starts at 0 + /// (the cache stamps it on read). A snapshot that stopped at Layer 0 yields + /// empty river/basin/attractor data. + pub fn into_body_world_state(self) -> BodyWorldState { + let (river_network, drainage_basins, attractors) = match self.layer1 { + Some(l1) => (l1.river_network, l1.drainage_basins, l1.attractors), + None => (RiverNetwork::default(), Vec::new(), Vec::new()), + }; + BodyWorldState { + body_id: self.body_id, + heightmap: self.heightmap.data, + heightmap_width: self.heightmap.width, + heightmap_height: self.heightmap.height, + river_network, + drainage_basins, + attractors, + last_accessed: 0, + } + } +} + /// Run the cascade from a heightmap already in memory, up to `up_to`. /// /// Pure (no I/O); this is the testable core. `body_seed` is this body's diff --git a/server/src/atlas/gen_queue.rs b/server/src/atlas/gen_queue.rs index 95e99c563..b66ca86f4 100644 --- a/server/src/atlas/gen_queue.rs +++ b/server/src/atlas/gen_queue.rs @@ -21,11 +21,17 @@ //! //! **Thread count (D-206):** `available_parallelism - 2`, minimum 1. +use std::path::PathBuf; use std::sync::{Arc, Mutex}; use bevy_ecs::prelude::Resource; use crossbeam_channel::{Receiver, Sender}; +use crate::atlas::body_world_state::BodyWorldState; +use crate::atlas::cascade::{run_cascade_from_heightmap, CascadeLayer}; +use crate::atlas::heightmap::{load_heightmap_png, GRID_H, GRID_W}; +use crate::seed::SeedChain; + // --------------------------------------------------------------------------- // Priority // --------------------------------------------------------------------------- @@ -50,8 +56,16 @@ pub enum GenPriority { /// A unit of background generation work (D-206). #[derive(Debug, Clone)] pub enum GenWorkItem { - /// Run D8 drainage analysis + attractor extraction for this body. - AnalyzeBody { body_id: String }, + /// Run the Layer-1 cascade (drainage → features → sub-biome) for this body. + /// The enqueuer resolves the inputs (D-225): `heightmap_path` is the + /// mod-resolved source PNG, `body_seed` is this body's SeedChain position. + /// `run_work_item` is pure compute — it does no path/DB resolution. + AnalyzeBody { + body_id: String, + heightmap_path: PathBuf, + sea_level: f32, + body_seed: SeedChain, + }, /// Generate a Phase 1 DistrictSkeleton for this city. GenerateSkeleton { city_id: u64 }, /// Pre-fill a chunk in an existing district. @@ -63,7 +77,7 @@ pub enum GenWorkItem { impl GenWorkItem { pub fn body_id(&self) -> Option<&str> { - if let GenWorkItem::AnalyzeBody { body_id } = self { + if let GenWorkItem::AnalyzeBody { body_id, .. } = self { Some(body_id) } else { None @@ -80,6 +94,8 @@ impl GenWorkItem { pub enum GenCompletion { BodyAnalyzed { body_id: String, + /// The computed world state, ready for `BodyWorldStateCache::insert`. + state: BodyWorldState, }, SkeletonGenerated { city_id: u64, @@ -286,15 +302,38 @@ impl Default for GenerationQueue { // Work execution stub // --------------------------------------------------------------------------- -/// Execute one work item. This is the Rayon task body. +/// Execute one work item. This is the Rayon task body (off the tick thread). /// -/// Currently a stub — real implementations will call `drainage::analyze()`, -/// the attractor pipeline, and the district skeleton generator. Stubs return -/// immediate success to allow the queue infrastructure to be tested independently. +/// `AnalyzeBody` runs the real Layer-1 cascade (#968, D-225). `GenerateSkeleton` +/// and `FillChunk` remain stubs — their layers (#957 / #959) are not built yet — +/// returning immediate success so the queue infrastructure stays testable. fn run_work_item(item: &GenWorkItem) -> GenCompletion { match item { - GenWorkItem::AnalyzeBody { body_id } => GenCompletion::BodyAnalyzed { - body_id: body_id.clone(), + GenWorkItem::AnalyzeBody { + body_id, + heightmap_path, + sea_level, + body_seed, + } => match load_heightmap_png(heightmap_path, body_id, *sea_level) { + Ok(hm) => { + // Layer 1 runs at the GRID_W×GRID_H working resolution (D-202): + // downsample the higher-res stored heightmap first. + let working = if hm.width > GRID_W || hm.height > GRID_H { + hm.downsample(GRID_W, GRID_H) + } else { + hm + }; + let snapshot = + run_cascade_from_heightmap(*body_seed, working, CascadeLayer::Topography); + GenCompletion::BodyAnalyzed { + body_id: body_id.clone(), + state: snapshot.into_body_world_state(), + } + } + Err(e) => GenCompletion::Failed { + item: item.clone(), + reason: format!("heightmap load failed: {e}"), + }, }, GenWorkItem::GenerateSkeleton { city_id } => { GenCompletion::SkeletonGenerated { city_id: *city_id } @@ -322,22 +361,51 @@ mod tests { GenerationQueue::with_threads(2) } + /// Write a tiny 16-bit grayscale heightmap PNG to a unique temp path so the + /// real cascade can run in `run_work_item` without a committed fixture. + fn test_heightmap_path() -> std::path::PathBuf { + use std::io::BufWriter; + use std::sync::atomic::{AtomicU32, Ordering}; + static SEQ: AtomicU32 = AtomicU32::new(0); + let n = SEQ.fetch_add(1, Ordering::Relaxed); + let path = std::env::temp_dir().join(format!("sr_genq_{}_{n}.png", std::process::id())); + let file = std::fs::File::create(&path).expect("create test heightmap"); + let mut enc = png::Encoder::new(BufWriter::new(file), 32, 16); + enc.set_color(png::ColorType::Grayscale); + enc.set_depth(png::BitDepth::Sixteen); + let mut w = enc.write_header().expect("png header"); + let data: Vec = (0..32u32 * 16) + .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes()) + .collect(); + w.write_image_data(&data).expect("png data"); + path + } + + /// Build an `AnalyzeBody` work item pointing at a tiny test heightmap. + fn analyze(body_id: &str) -> GenWorkItem { + GenWorkItem::AnalyzeBody { + body_id: body_id.to_string(), + heightmap_path: test_heightmap_path(), + sea_level: 0.3, + body_seed: SeedChain::for_body(42, body_id), + } + } + #[test] fn submit_and_drain() { let q = make_queue(); - q.submit( - GenWorkItem::AnalyzeBody { - body_id: "TestBody".to_string(), - }, - GenPriority::Medium, - ); - // Give Rayon time to complete the (stub) task. - std::thread::sleep(Duration::from_millis(50)); + q.submit(analyze("TestBody"), GenPriority::Medium); + // Give Rayon time to load the heightmap and run the cascade. + std::thread::sleep(Duration::from_millis(100)); let completions = q.drain_completions(); assert_eq!(completions.len(), 1); + // The work item ran the real cascade and produced a populated state. assert!(matches!( &completions[0], - GenCompletion::BodyAnalyzed { body_id } if body_id == "TestBody" + GenCompletion::BodyAnalyzed { body_id, state } + if body_id == "TestBody" + && state.heightmap_width == 32 + && state.heightmap_height == 16 )); } @@ -345,18 +413,8 @@ mod tests { fn dedup_analyze_body() { let q = make_queue(); // Submit the same body twice before it can complete. - q.submit( - GenWorkItem::AnalyzeBody { - body_id: "Dup".to_string(), - }, - GenPriority::Low, - ); - q.submit( - GenWorkItem::AnalyzeBody { - body_id: "Dup".to_string(), - }, - GenPriority::Low, - ); + q.submit(analyze("Dup"), GenPriority::Low); + q.submit(analyze("Dup"), GenPriority::Low); std::thread::sleep(Duration::from_millis(50)); let completions = q.drain_completions(); // Should have completed exactly once. @@ -399,18 +457,8 @@ mod tests { // Vec while "BodyA" is in-flight (in_flight_count = 1 = n_threads). // - When "BodyA" completes, drain_completions() calls dispatch_next() // which picks index 0 = "BodyB" (Immediate). - q.submit( - GenWorkItem::AnalyzeBody { - body_id: "BodyA".to_string(), - }, - GenPriority::Low, - ); - q.submit( - GenWorkItem::AnalyzeBody { - body_id: "BodyB".to_string(), - }, - GenPriority::Immediate, - ); + q.submit(analyze("BodyA"), GenPriority::Low); + q.submit(analyze("BodyB"), GenPriority::Immediate); // Wait for BodyA to complete. std::thread::sleep(Duration::from_millis(50)); // drain_completions dispatches BodyB (Immediate, index 0 of pending). @@ -421,9 +469,11 @@ mod tests { assert_eq!(first.len(), 1); assert_eq!(second.len(), 1); - assert!(matches!(&first[0], GenCompletion::BodyAnalyzed { body_id } if body_id == "BodyA")); assert!( - matches!(&second[0], GenCompletion::BodyAnalyzed { body_id } if body_id == "BodyB") + matches!(&first[0], GenCompletion::BodyAnalyzed { body_id, .. } if body_id == "BodyA") + ); + assert!( + matches!(&second[0], GenCompletion::BodyAnalyzed { body_id, .. } if body_id == "BodyB") ); } diff --git a/server/src/atlas/mod.rs b/server/src/atlas/mod.rs index d73ca3a7e..f971695dd 100644 --- a/server/src/atlas/mod.rs +++ b/server/src/atlas/mod.rs @@ -13,6 +13,9 @@ pub mod features; pub mod gen_queue; pub mod heightmap; pub mod layer1; +pub mod plugin; pub mod skeleton_gen; pub mod subbiome; pub mod tile_condition; + +pub use plugin::GenerationPlugin; diff --git a/server/src/atlas/plugin.rs b/server/src/atlas/plugin.rs new file mode 100644 index 000000000..d5eaf96b9 --- /dev/null +++ b/server/src/atlas/plugin.rs @@ -0,0 +1,116 @@ +//! Generation tier plugin (#968, D-206) — wires the background generation queue +//! and the per-body world-state cache into the running app. +//! +//! Registers [`GenerationQueue`] and [`BodyWorldStateCache`] as resources and +//! adds a `PreInput` system that drains completed work each tick and inserts the +//! computed [`BodyWorldState`](crate::atlas::body_world_state::BodyWorldState) +//! into the cache. The queue's *submitter* is the atlas layer-stream proxy +//! (#969, D-225); this plugin closes the submit→Rayon→cascade→drain→cache loop. + +use bevy_app::prelude::*; +use bevy_ecs::prelude::*; +use bevy_ecs::schedule::IntoScheduleConfigs; + +use crate::atlas::body_world_state::{BodyWorldStateCache, CACHE_CAPACITY}; +use crate::atlas::gen_queue::{GenCompletion, GenerationQueue}; +use crate::tick_phases::TickPhase; + +/// Wires the D-206 background generation tier into the app (#968). +pub struct GenerationPlugin; + +impl Plugin for GenerationPlugin { + fn build(&self, app: &mut App) { + app.insert_resource(GenerationQueue::new()) + .insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY)) + .add_systems( + Update, + drain_generation_completions.in_set(TickPhase::PreInput), + ); + } +} + +/// Drain finished background work each tick and apply it to the cache (D-206). +/// +/// Runs in `PreInput` (off the Rayon workers, on the main thread): a cheap +/// channel drain + cache insert, never the ~45 ms cascade itself. +fn drain_generation_completions( + queue: Res, + mut cache: ResMut, +) { + for completion in queue.drain_completions() { + match completion { + GenCompletion::BodyAnalyzed { state, .. } => cache.insert(state), + GenCompletion::Failed { item, reason } => { + tracing::warn!(?item, %reason, "background generation work item failed"); + } + // Produced once the later layers land (#957 / #959); no consumer yet. + GenCompletion::SkeletonGenerated { .. } | GenCompletion::ChunkFilled { .. } => {} + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::atlas::gen_queue::{GenPriority, GenWorkItem}; + use crate::seed::SeedChain; + use bevy_ecs::schedule::Schedule; + use std::time::Duration; + + /// Tiny 16-bit grayscale heightmap PNG at a unique temp path, so the real + /// cascade can run without a committed fixture. + fn test_heightmap_path() -> std::path::PathBuf { + use std::io::BufWriter; + use std::sync::atomic::{AtomicU32, Ordering}; + static SEQ: AtomicU32 = AtomicU32::new(0); + let n = SEQ.fetch_add(1, Ordering::Relaxed); + let path = + std::env::temp_dir().join(format!("sr_genplugin_{}_{n}.png", std::process::id())); + let file = std::fs::File::create(&path).expect("create test heightmap"); + let mut enc = png::Encoder::new(BufWriter::new(file), 32, 16); + enc.set_color(png::ColorType::Grayscale); + enc.set_depth(png::BitDepth::Sixteen); + let mut w = enc.write_header().expect("png header"); + let data: Vec = (0..32u32 * 16) + .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes()) + .collect(); + w.write_image_data(&data).expect("png data"); + path + } + + #[test] + fn drain_system_populates_cache() { + // The full loop: submit → Rayon cascade → completion → drain → cache. + let mut world = World::new(); + world.insert_resource(GenerationQueue::new()); + world.insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY)); + + world.resource::().submit( + GenWorkItem::AnalyzeBody { + body_id: "PlanetX".to_string(), + heightmap_path: test_heightmap_path(), + sea_level: 0.3, + body_seed: SeedChain::for_body(42, "PlanetX"), + }, + GenPriority::Immediate, + ); + + let mut sched = Schedule::default(); + sched.add_systems(drain_generation_completions); + + // Rayon runs the cascade asynchronously; the drain runs each schedule pass. + let mut found = false; + for _ in 0..100 { + sched.run(&mut world); + if world.resource::().contains("PlanetX") { + found = true; + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + found, + "drain system should insert the analyzed body into the cache" + ); + } +} diff --git a/server/src/main.rs b/server/src/main.rs index 133c613f1..fe26b5c18 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -154,6 +154,7 @@ fn main() { app.add_plugins(settled_reach_server::storyteller::StorytellerPlugin); app.add_plugins(settled_reach_server::settings::SettingsPlugin); app.add_plugins(settled_reach_server::bookmark::BookmarkPlugin::default()); + app.add_plugins(settled_reach_server::atlas::GenerationPlugin); // Initialize culture resolver (#679, D-128). // systems.db is shipped read-only alongside the binary. @@ -391,6 +392,7 @@ fn dump_schedule_graph() { app.add_plugins(settled_reach_server::storyteller::StorytellerPlugin); app.add_plugins(settled_reach_server::settings::SettingsPlugin); app.add_plugins(settled_reach_server::bookmark::BookmarkPlugin::default()); + app.add_plugins(settled_reach_server::atlas::GenerationPlugin); app.insert_resource(settled_reach_server::simulation::rng::SimRng::new(0)); // Access Schedules resource directly — schedules are populated by plugins