//! 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::atlas::layer_proxy::{handle_atlas_request, AtlasLayerResponse, AtlasLayerStatus}; use crate::atlas::source_resolver::BodySourceResolverResource; use crate::bridge::{AtlasRequestBuffer, AtlasResponseBuffer}; use crate::simulation::rng::SimRng; use crate::simulation::time::SimulationTime; 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), ) .add_systems(Update, serve_atlas_requests.in_set(TickPhase::PreInput)); } } /// Drain inbound atlas layer requests and serve each through the proxy (#969, /// D-225): cache hit → Ready, miss → resolve + enqueue + Pending. Responses are /// buffered for the bridge to flush in `PostSnapshot`. fn serve_atlas_requests( mut requests: ResMut, mut responses: ResMut, mut cache: ResMut, queue: Res, resolver: Option>, rng: Option>, time: Option>, ) { if requests.0.is_empty() { return; } let world_seed = rng.as_ref().map(|r| r.seed()).unwrap_or(0); let tick = time.as_ref().map(|t| t.tick).unwrap_or(0); let pending: Vec<_> = requests.0.drain(..).collect(); for req in pending { let resp = match resolver.as_ref() { Some(r) => handle_atlas_request(&req, &mut cache, &queue, &r.0, world_seed, tick), None => AtlasLayerResponse { body_id: req.body_id.clone(), status: AtlasLayerStatus::Error("no body source resolver".to_string()), layer1: None, }, }; responses.0.push(resp); } } /// 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"); } // Insert district world state into the matching body's cache entry (D-230). GenCompletion::SkeletonGenerated { city_id, body_id, state, } => { if !body_id.is_empty() { if let Some(body_state) = cache.peek_mut(&body_id) { // Key by state.skeleton.district_id (D-194/D-230): a city has many // districts, each with its own DistrictId. `city_id` is only the // dispatch key used in the work item — the canonical insert key is // the district's own stable id. TODO(#957): the stub GenerateSkeleton // returns a default skeleton with district_id=0; real gen (#957) will // populate it from CityGenerationContext. let _ = city_id; // used as dispatch key only; district_id is the map key body_state .districts .insert(state.skeleton.district_id, *state); } else { tracing::warn!( city_id, body_id, "SkeletonGenerated: body not in cache — district state dropped" ); } } // body_id empty = stub result from GenerateSkeleton stub; silently ignore. } 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" ); } #[test] fn serve_drains_requests_into_responses() { use crate::atlas::cascade::CascadeLayer; use crate::atlas::layer_proxy::AtlasLayerRequest; let mut world = World::new(); world.insert_resource(AtlasRequestBuffer(vec![AtlasLayerRequest { body_id: "GJ1c".to_string(), up_to: CascadeLayer::Topography, }])); world.insert_resource(AtlasResponseBuffer::default()); world.insert_resource(BodyWorldStateCache::new(CACHE_CAPACITY)); world.insert_resource(GenerationQueue::with_threads(1)); // No resolver / SimRng / SimulationTime — all optional in the system. let mut sched = Schedule::default(); sched.add_systems(serve_atlas_requests); sched.run(&mut world); let responses = world.resource::(); assert_eq!(responses.0.len(), 1, "request should produce one response"); assert_eq!(responses.0[0].body_id, "GJ1c"); // No resolver wired → Error status (exercises the drain + push path). assert!(matches!(responses.0[0].status, AtlasLayerStatus::Error(_))); // The request buffer was drained. assert!(world.resource::().0.is_empty()); } }