//! Background generation queue — prioritized Rayon thread pool (D-206). //! //! All runtime-background generation work runs through this queue. The main //! tick thread submits work items (non-blocking) and drains completion events //! once per tick via a `crossbeam` channel. //! //! **Priority levels (D-206):** //! - `Immediate`: player arrives within 1 game-minute. Runs first. //! - `High`: player arrives within 5 game-minutes. //! - `Medium`: player is in the same system. //! - `Low`: player has heard of this location via NPC/news. //! //! **Work item types (D-206):** //! - `AnalyzeBody`: D8 drainage + attractor extraction for a body. //! - `GenerateSkeleton`: Phase 1 QuarterSkeleton for a city. //! - `FillChunk`: Phase 2 chunk fill for a pre-loaded quarter. //! - `DeriveWindow`: District-resolution window derive (D-226 T-1124, T-1137). //! //! Completion events are delivered to the main thread via //! `GenerationQueue::drain_completions()`, called once per tick from a Bevy //! system in `TickPhase::PreInput`. //! //! **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::attractor_matching::CityRecord; use crate::atlas::body_world_state::BodyWorldState; use crate::atlas::cascade::{run_cascade_from_heightmap, CascadeLayer}; use crate::atlas::district_profile::{BodyParams, ClimateConstants, DistrictPos}; use crate::atlas::features::TerrainAnalysis; use crate::atlas::heightmap::{load_heightmap_png, GRID_H, GRID_W}; use crate::atlas::layer1::Layer1Output; use crate::atlas::layer_proxy::{ build_district_window_layer, DistrictWindowLayer, WindowGranularity, }; use crate::atlas::shell::{fill_chunk, FilledChunk}; use crate::atlas::skeleton_gen::{assign_all_block_tags, generate_quarter_skeleton}; use crate::atlas::trait_catalog_reader::ExteriorCatalog; use crate::bridge::ConnectionId; use crate::seed::SeedChain; use crate::simulation::generator::{BuildingPropertyTag, CityGenerationContext, QuarterWorldState}; // --------------------------------------------------------------------------- // Priority // --------------------------------------------------------------------------- /// Work priority levels — lower discriminant = higher priority. #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub enum GenPriority { /// Player arrives within ~1 game-minute. Runs before all other levels. Immediate = 0, /// Player arrives within ~5 game-minutes. High = 1, /// Player is in the same system. Medium = 2, /// Player has seen or heard of this location (NPC dialogue, news ticker). Low = 3, } // --------------------------------------------------------------------------- // Work item types // --------------------------------------------------------------------------- /// A unit of background generation work (D-206). #[derive(Debug, Clone)] pub enum GenWorkItem { /// 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, /// The body's settlements (from `atlas_city_names`), pre-resolved at /// dispatch time so the cascade stays DB-free. Fed to Layer-3 placement /// (#955); empty if the body has no settlements (cascade stops at Layer 1). cities: Vec, /// The body's system `dominant_faction` (D-237), pre-resolved at dispatch /// time. Drives Layer-3 TerritorialStatus + spatial character (#956, /// D-212/214/215). `None` → `FrontierUnclaimed`. dominant_faction: Option, /// Body physical parameters for the DistrictProfile layer (T-1023, D-239 §1). /// Pre-resolved at dispatch time. `None` → district layer skipped for this body. /// Boxed: `BodyParams` is large relative to other variants (clippy /// large_enum_variant) — boxing keeps `GenWorkItem` compact. body_params: Option>, }, /// Generate a Phase 1 QuarterSkeleton for this city. /// /// `context` is the D-199 economic read-set pre-resolved at dispatch time. /// All 6 required fields must be populated before this item is submitted /// (D-199: "Missing fields abort the task … generation does not proceed with /// partial context") — EXCEPT `morphology_zone`, which is resolved during /// execution (see below, D-256(d)). /// /// `body_id` routes the resulting `SkeletonGenerated` completion into the /// correct `BodyWorldState` cache entry (D-230). /// /// `quarter_id` is the stable content-addressable id for the generated /// quarter (keyed by city position + world seed). /// /// `economic_role`, `population`, and `founding_age_years` are D-199 fields /// carried alongside the context because `generate_quarter_skeleton` accepts them as /// separate parameters (its signature is not changed by this ticket). GenerateSkeleton { city_id: u64, body_id: String, /// D-199 economic read-set + all other context fields. /// /// `context.morphology_zone` still starts life as /// `context_from_read_set`'s `AlluvialPlain` stub at dispatch time — it /// is overwritten during execution (D-256(d), see `settlement_world_m` /// below), not by `build_skeleton_work_item` as it was pre-D-256. context: Box, /// Stable content-addressable quarter id (D-194/D-230). quarter_id: u64, /// Quarter-level seed chain (D-224). chain: SeedChain, // D-199 raw fields passed to generate_quarter_skeleton separately. economic_role: String, population: i64, founding_age_years: u32, /// D-235 exterior-grammar content (T-988): the trait-template catalog's /// `visual_bundle`s plus the two sibling content tables /// (`architecture_zone_bias`, `color_register_bands`), pre-resolved at /// dispatch time — mirrors `context`'s own pre-resolution rationale, so /// `assign_block_tags` never touches `systems.db` (T-987/D-230 purity). /// A `Vec`/`BTreeMap`-backed struct is a handful of words on the stack /// regardless of its contents' size (same reasoning already documented /// for `FillChunk.block_tags` below), so this needs no `Box`. exterior_catalog: ExteriorCatalog, /// D-256(d) exact-position `morphology_zone` judgment inputs — the /// settlement's own world metres (NOT the survey cell's centre, which /// can be hundreds of km off for a settlement near a survey-cell edge) /// and body params, resolved during `run_work_item` execution (where /// the terrain cache is reachable, unlike `BodyWorldState`'s /// D-203/T-1048 dropped `TerrainAnalysis`) via `derive_at_metres`. /// `None` body_params (no DB row for this body) skips the resolution /// and leaves `context.morphology_zone` at its `AlluvialPlain` stub — /// the same fallback the pre-D-256 dispatch-time lookup used for an /// empty district grid. settlement_world_m: (f64, f64), body_params: Option>, /// `SeedChain::for_body(world_seed, body_id)` — the SAME body-scoped /// chain `derive_all_districts`/`derive_district_profile` use, distinct /// from `chain` (the quarter-level chain derived further for skeleton /// RNG). Needed because `derive_at_metres`'s domain-separated warp /// fields must key on the same seed the rest of the cascade uses for /// this body, not a chain re-derived from the quarter seed. body_seed: SeedChain, /// Shared per-body working-grid heightmap (D-256(d)) — an `Arc` so /// every settlement dispatched for the same `BodyAnalyzed` completion /// clones a pointer, not the multi-hundred-KB `Vec`. Reconstructed /// once at dispatch time from `BodyWorldState.heightmap`/dims/`sea_level` /// (already in memory — no disk re-read) so `run_work_item` can feed /// [`TerrainAnalysisCache::get_or_derive`] without touching the /// filesystem on the Rayon worker thread. heightmap: std::sync::Arc, }, /// Derive the building shell for one 64 m chunk of an existing quarter /// (D-230 derive phase, T-987). /// /// The covering block's `block_tags` are **pre-resolved into the item** at enqueue /// time because `run_work_item` is cache-free (mirrors `GenerateSkeleton`). A 64 m /// chunk lies wholly within one 128 m block and footprints are block-confined, so /// the covering block's tags are exactly the relevant set. Build items with /// [`build_fill_chunk_item`] — the D-230 "skeleton not yet processed → re-enqueue /// at `High`" precondition is the caller's cache lookup, which only reaches this /// constructor once the `QuarterWorldState` exists. FillChunk { /// Stable id of the quarter being filled (D-194/D-230). quarter_id: u64, /// Block grid position within the quarter (0..4, 0..4). block_pos: (u8, u8), /// Sub-chunk quadrant within the block (0..2, 0..2) — a block is 2×2 chunks. sub_chunk: (u8, u8), /// Covering block's building tags, pre-resolved from the cached /// `QuarterWorldState`. Empty for an open/un-built block (→ empty shell). /// A `Vec` is three words on the stack regardless of element size (the tags /// live behind the pointer), so — unlike `AnalyzeBody`'s boxed `BodyParams` — /// this variant needs no `Box` to stay clippy `large_enum_variant`-clean. block_tags: Vec, }, /// Derive a district-resolution window (D-226 T-1124 amendment, T-1137). /// /// **Binding serving model:** window derives ride this SAME Rayon queue as /// every other expensive atlas path — never inline on the `PreInput` drain /// (the amendment's §1 is explicit: a window derive at n=32/64 is /// ~7–29 ms, which would blow the "cheap channel drain" contract /// `drain_generation_completions` documents). /// /// **TerrainAnalysis availability (T-1137 binding decision, with numbers; /// corrected 2026-07-21 per PR #187 review — Tyre C1; RE-CORRECTED /// 2026-07-25 per PR #200 review — Hoshe finding 1, T-1184's hydrology /// field addition):** `BodyWorldState` does NOT retain `TerrainAnalysis` /// after cascade completion (T-1044 scoped its transient-carry fix to /// *within-cascade* reuse only — `cascade.rs` drops it once /// `DistrictProfile`+`RoadGraph` finish; see the doc on /// `CascadeSnapshot::terrain_analysis`). Caching it alongside every /// `BodyWorldStateCache` entry would cost **~2.62 MB × 50-body capacity ≈ /// 131 MB** of PERMANENT resident cost (real per-entry figure below — /// this multiplier is unchanged, only the per-entry base moved), paid by /// every cached body whether or not a window is ever requested for it — /// the exact D-203 budget concern T-1044's own ticket text guarded /// against. So `run_work_item` re-derives via `run_layer1` (matching /// `aliveness_probe`'s existing `--render` workaround) rather than /// persisting a field on `BodyWorldState` — but NOT unconditionally on /// every work item: the actual model is a small **per-body LRU** /// (`TerrainAnalysisCache`, capacity 8), consulted before every /// re-derive. /// /// **Real per-entry size (T-1184, corrected from the stale "~2 MB" /// figure the PR #187 pass computed before hydrology existed):** at the /// real `GRID_W × GRID_H = 512 × 256` working grid (131,072 cells, the /// SAME grid every `TerrainAnalysisCache` entry is built at — NOT the /// 1024×512 stored-heightmap-PNG resolution, D-202, which is downsampled /// away before `run_layer1` ever runs) — pre-T-1184 dense fields /// (`ocean_mask`/`lake_mask`: `Vec`, 1 B/cell each; `water_dist`: /// `Vec`, 2 B/cell; `slope_deg`/`elev_pct`: `Vec`, 4 B/cell /// each) sum to **~1.57 MB**. T-1184's `HydrologySample` (`elevation` + /// `filled`, both `Vec`, 4 B/cell) adds **~1.05 MB** — **not quite a /// doubling** (×1.67, not ×2), but a real, non-trivial per-entry growth: /// **post-T-1184 total ~2.62 MB/entry.** At capacity 8 that is **~21.0 MB /// worst case** (was ~12.6 MB pre-T-1184; the old "~16 MB" comment was /// itself a round-up of the pre-T-1184 figure, not a post-hydrology one). /// Both this cache and `BodyWorldStateCache`'s counterfactual above use /// the corrected ~2.62 MB base. /// /// **Clone-on-HIT, not just insert (Hoshe finding 1, binding for the next /// sizing decision):** `get_or_derive`'s cache-hit branch /// (`return (l1.clone(), ta.clone())`) deep-copies the FULL /// `TerrainAnalysis` — including both new `HydrologySample` `Vec` /// fields — on EVERY hit, not merely on the first miss/insert. A /// pan-heavy session hitting the same body's cache entry repeatedly pays /// the ~2.62 MB clone cost per `DeriveWindow` work item, not once per /// body. This was already true pre-T-1184 for the smaller ~1.57 MB /// struct; T-1184 makes the per-hit clone cost ~1.67× larger, not a new /// category of cost. **Not fixed this round** (an `Arc` /// refactor — sharing one heap allocation across hits instead of /// deep-copying — is the obvious next step if hit-heavy clone cost ever /// shows up in a profile, but is out of scope for T-1184; flagged for a /// follow-up ticket rather than done speculatively here). /// /// The FIRST `DeriveWindow` on a /// body pays the full ~45 ms `run_layer1` cost and populates that body's /// cache entry; EVERY SUBSEQUENT window on the SAME body (any `center`/`n`, /// not just an exact repeat — that narrower case is what /// `DistrictWindowCache` in `layer_proxy.rs` already catches) hits the LRU /// and skips straight to the ~7–29 ms per-window pack in /// `build_district_window_layer`. An LRU eviction re-pays the ~45 ms on the /// next window for that body. Because every `DeriveWindow` work item still /// runs off the tick thread on the Rayon queue regardless of hit or miss, /// both costs are invisible to the main thread either way — the LRU's /// value is throughput/worker-occupancy (bounding how many ~45 ms re-derives /// a pan-burst across one body can force), not tick-thread latency. DeriveWindow { body_id: String, /// Coalescing/routing key (D-226 T-1124 amendment §1 "recommended" /// per-connection coalescing) — NOT used by `run_work_item` itself /// (the derive is connection-agnostic), only by /// `GenerationQueue::submit_window` to decide which still-pending item /// a new one for the same connection+body supersedes. conn_id: ConnectionId, /// Mod-resolved source heightmap PNG (mirrors `AnalyzeBody`). heightmap_path: PathBuf, sea_level: f32, /// This body's `SeedChain` position — `derive_district`'s seed input. body_seed: SeedChain, /// Body physical parameters. Boxed for the same `large_enum_variant` /// reason `AnalyzeBody.body_params` is boxed. body_params: Box, /// Window centre + side length in districts. `n` is ALREADY clamped to /// `[1, DISTRICT_WINDOW_MAX_N]` AND the granularity-aware /// `WIRE_CAP_CELLS` ceiling by the caller (`handle_atlas_request`) /// before this item is built — never trusted from the wire again here. center: DistrictPos, n: u32, /// Derivation granularity (T-1150, widened T-1152 to the full /// [`WindowGranularity`] vocabulary — `District`/`Quarter` (finer) /// plus `Region` (coarser, T-1152)). Already resolved via /// `resolve_window_granularity_v2` by the caller — this is a /// concrete rung, never a raw wire value. granularity: WindowGranularity, /// Octave cutoff in whole metres (T-1149/T-1150), `0` = no cutoff. /// Meaningless for `granularity: Region` (`derive_orbital_at_metres` /// never calls the octave-scatter path this cuts off) but still /// carried and echoed uniformly — see `derive_orbital_at_metres`'s /// doc for why the field is harmless-but-unused there, not /// special-cased away. min_wl_m: u32, }, /// Derive a D-255(a) step canvas (T-1181, the tagged-envelope migration). /// /// **Binding serving model — same as `DeriveWindow`:** step canvases ride /// this SAME Rayon queue, never inline on the `PreInput` drain (D-255(d): /// "never inline here"). Reuses the SAME `TerrainAnalysisCache` LRU /// `DeriveWindow` already populates — a step-canvas request and a legacy /// `district_window` request for the same body share one /// `run_layer1`-derived `(Layer1Output, TerrainAnalysis)` cache entry /// (keyed on `body_id` alone), so a player already panning the legacy /// window viewer and then opening the new stepped map (or vice versa) /// pays the ~45 ms `run_layer1` cost at most once per body, not once per /// carrier. DeriveStepCanvas { body_id: String, /// Coalescing/routing key — NOT used by `run_work_item` itself (the /// derive is connection-agnostic), only by /// `GenerationQueue::submit_step_canvas` to decide which still-pending /// item a new one for the same connection+body+rung supersedes. conn_id: ConnectionId, heightmap_path: PathBuf, sea_level: f32, body_seed: SeedChain, body_params: Box, /// The body's placed settlements, pre-resolved at dispatch time from /// `BodyWorldState.placements` if already cached (empty otherwise — /// `settlement_id` coverage is simply all-zero until the body's own /// `AnalyzeBody` cascade has placed settlements; see /// `step_canvas::serve_step_canvas_request`'s doc). placements: Vec, /// The D-255(a) rung this canvas targets. rung: crate::atlas::step_canvas::StepCanvasRung, /// World-metre centre (ignored for `StepCanvasRung::Global`, whose /// canvas is whole-body/origin-anchored). center: (i64, i64), /// Canvas pixel budget for every FIXED rung (ignored for `Global`, /// whose extent is the body's own region grid). extent: (u32, u32), /// Octave cutoff in whole metres, already quantized by the caller /// (`step_canvas::quantize_min_wl_m_for_rung`) — never a raw wire /// value. min_wl_m: u32, }, } impl GenWorkItem { pub fn body_id(&self) -> Option<&str> { match self { GenWorkItem::AnalyzeBody { body_id, .. } => Some(body_id), _ => None, } } /// Coalescing key for `DeriveWindow` items only — `(connection, body, /// granularity)` (T-1150, design doc §3 [SOFT] recommendation, extending /// T-1137's `(connection, body)`; widened T-1152 to carry the full /// [`WindowGranularity`] enum rather than the legacy `u32`, so `Region` /// occupies its own coalescing slot exactly like `District`/`Quarter` do /// — this is one of the five T-1150 touch points the R5 redesign must /// carry the new representation through). `granularity` is part of the /// key so an in-flight district-spacing pan-burst is never superseded by /// an unrelated quarter- or region-spacing request for the same /// connection+body, and vice versa — every rung is a separate in-flight /// derive, not a competing update to the same one. /// `None` for every other variant (they don't coalesce this way). pub fn window_supersede_key(&self) -> Option<(ConnectionId, &str, WindowGranularity)> { if let GenWorkItem::DeriveWindow { body_id, conn_id, granularity, .. } = self { Some((*conn_id, body_id, *granularity)) } else { None } } /// Coalescing key for `DeriveStepCanvas` items only — `(connection, body, /// rung)` (T-1181, mirroring `window_supersede_key`'s exact reasoning): /// a pan-burst that queues several step-canvas requests for the same /// connection+body+rung before the first is dispatched collapses to one /// derive. A different rung for the same connection+body does NOT /// coalesce — every rung is a separate in-flight derive, not a competing /// update to the same one. `None` for every other variant. pub fn step_canvas_supersede_key( &self, ) -> Option<( ConnectionId, &str, crate::atlas::step_canvas::StepCanvasRung, )> { if let GenWorkItem::DeriveStepCanvas { body_id, conn_id, rung, .. } = self { Some((*conn_id, body_id, *rung)) } else { None } } } // --------------------------------------------------------------------------- // Completion event // --------------------------------------------------------------------------- /// Sent back to the main thread when a work item finishes (D-206). #[derive(Debug)] pub enum GenCompletion { BodyAnalyzed { body_id: String, /// The computed world state, ready for `BodyWorldStateCache::insert`. /// Boxed to keep `GenCompletion` variant sizes balanced — the same /// discipline as `SkeletonGenerated`/`ChunkFilled`/`WindowDerived` /// below, all boxed for the same reason (`GenCompletion`'s overall /// size is bounded by its largest unboxed variant; a `BodyWorldState` /// field carried by value here would force every other variant to /// pay for its full stack size on every match/move). /// /// **Not this variant's own recent growth** (T-1170 PR #197 review, /// Tyre issue 2 — the prior comment here overattributed the boxing /// rationale): `BodyWorldState` itself only grew by /// `RiverNetwork`'s two new `Vec` fields (`river_downstream`, /// `river_seaward` — a few bytes/river-cell, Ruling 2a / Hoshe #1). /// The much larger `Layer1Output` retention (T-1170 Ruling 4b) lives /// on `TerrainAnalysisCache` — a queue-scoped struct further down /// this file, never part of `BodyWorldState`/`BodyWorldStateCache` at /// all. The box here predates T-1170 and stays for the pre-existing /// variant-size-balancing reason, unrelated to this ticket's growth. state: Box, }, SkeletonGenerated { city_id: u64, /// The body this skeleton belongs to — used to route state into /// `BodyWorldState.quarters` (D-230). body_id: String, /// District-level world state (skeleton + block tags) produced by the plan phase (D-230). /// Boxed to keep `GenCompletion` variant sizes balanced (D-230 skeleton is ~2.7 KB). state: Box, }, ChunkFilled { /// The derived shell for this chunk (D-230, T-987). Sparse — only the /// non-`Void` shell voxels. Carries its own quarter/block/sub-chunk address. /// Boxed to keep `GenCompletion` variant sizes balanced. filled: Box, }, /// A district window finished deriving (D-226 T-1124 amendment, T-1137). /// The main thread inserts `layer` into the `DistrictWindowCache` keyed by /// `(body_id, layer.center, layer.n)` — NOT pushed directly into any /// in-flight response (the window's requester re-polls per the existing /// D-225 loop and hits the now-populated cache on its next request; see /// `handle_atlas_request`'s window branch). WindowDerived { body_id: String, /// Boxed to keep `GenCompletion` variant sizes balanced (six /// `Vec`s — comparable to `SkeletonGenerated`/`ChunkFilled`'s own /// boxing rationale). layer: Box, }, /// A D-255(a) step canvas finished deriving (T-1181). The main thread /// inserts `canvas` into `GlobalTierCache` (rung `Global`) or /// `StepCanvasCache` (every other rung), keyed as /// `serve_step_canvas_request` documents — NOT pushed into any in-flight /// response, same re-poll-and-hit-cache model as `WindowDerived`. StepCanvasDerived { body_id: String, rung: crate::atlas::step_canvas::StepCanvasRung, center: (i64, i64), extent: (u32, u32), min_wl_m: u32, /// Boxed to keep `GenCompletion` variant sizes balanced — an /// `EncodedStepCanvas` carries eight PNG/msgpack-encoded fields, /// comparable to `WindowDerived`'s own boxing rationale. canvas: Box, }, /// Work item failed — body_id or city_id for logging. Failed { item: GenWorkItem, reason: String }, } // --------------------------------------------------------------------------- // Internal queued work // --------------------------------------------------------------------------- struct QueuedWork { priority: GenPriority, item: GenWorkItem, } // --------------------------------------------------------------------------- // GenerationQueue — Bevy Resource // --------------------------------------------------------------------------- /// Bevy `Resource` managing the background generation queue (D-206). /// /// Submit work with `submit()`. Drain completions with `drain_completions()` /// once per tick. The Rayon thread pool runs tasks in priority order. /// /// Also owns the queue-scoped `TerrainAnalysisCache` (T-1137, PR #187 review /// C1) — a small per-body LRU consulted by every `DeriveWindow` work item /// before paying the `run_layer1` re-derive cost. Lives here (not as a /// separate `Resource`) because it must be reachable from `run_work_item` /// while it executes on a Rayon worker thread, the same reason /// `in_flight`/`in_flight_count` are `Arc>` fields on this struct /// rather than plain fields. /// /// Priority is respected because `dispatch_next()` is gated on pool saturation /// via `in_flight_count`: it only dispatches when fewer than `n_threads` tasks /// are running. This applies to all work item types — `in_flight` (body-id set) /// is only for AnalyzeBody dedup; `in_flight_count` is the general saturation gate. #[derive(Resource)] pub struct GenerationQueue { /// Pending work items, sorted by priority (index 0 = highest priority). pending: Arc>>, /// Completions channel — background tasks send here; main thread reads. completion_tx: Sender, completion_rx: Receiver, /// Rayon thread pool dedicated to generation work. pool: rayon::ThreadPool, /// Set of body_ids currently in-flight — used only for AnalyzeBody dedup. in_flight: Arc>>, /// Count of all work items currently executing in the Rayon pool. /// This is the saturation gate — all work item types increment/decrement it. in_flight_count: Arc>, /// Thread count — caps concurrent dispatches so pending items accumulate /// and priority ordering is consulted before the pool has free threads. n_threads: usize, /// Per-body `TerrainAnalysis` LRU shared by every `DeriveWindow` work item /// on this queue (T-1137, PR #187 review C1) — see [`TerrainAnalysisCache`]. terrain_cache: Arc>, } impl std::fmt::Debug for GenerationQueue { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let pending_len = self.pending.lock().map(|p| p.len()).unwrap_or(0); f.debug_struct("GenerationQueue") .field("pending_count", &pending_len) .finish() } } impl GenerationQueue { /// Create a new queue with the D-206 thread count: /// `available_parallelism - 2`, minimum 1. pub fn new() -> Self { let n_threads = std::thread::available_parallelism() .map(|p| p.get().saturating_sub(2).max(1)) .unwrap_or(1); Self::with_threads(n_threads) } /// Create a queue with a specific thread count (for testing). pub fn with_threads(n_threads: usize) -> Self { let pool = rayon::ThreadPoolBuilder::new() .num_threads(n_threads) .thread_name(|i| format!("gen-worker-{i}")) .build() .expect("failed to build generation rayon pool"); let (tx, rx) = crossbeam_channel::unbounded(); Self { pending: Arc::new(Mutex::new(Vec::new())), completion_tx: tx, completion_rx: rx, pool, in_flight: Arc::new(Mutex::new(std::collections::BTreeSet::new())), in_flight_count: Arc::new(Mutex::new(0)), n_threads, terrain_cache: Arc::new(Mutex::new(TerrainAnalysisCache::new( TERRAIN_ANALYSIS_CACHE_CAPACITY, ))), } } /// Submit a work item at the given priority. /// /// If an `AnalyzeBody` item for the same body_id is already in-flight or /// pending, the submission is silently ignored (idempotent). pub fn submit(&self, item: GenWorkItem, priority: GenPriority) { // Dedup AnalyzeBody submissions. if let Some(body_id) = item.body_id() { let in_flight = self.in_flight.lock().unwrap(); if in_flight.contains(body_id) { return; } drop(in_flight); // Check pending list. let pending = self.pending.lock().unwrap(); if pending.iter().any(|q| q.item.body_id() == Some(body_id)) { return; } drop(pending); } let mut pending = self.pending.lock().unwrap(); let pos = pending .iter() .position(|q| q.priority > priority) .unwrap_or(pending.len()); pending.insert(pos, QueuedWork { priority, item }); drop(pending); self.dispatch_next(); } /// Submit a `DeriveWindow` item with per-connection coalescing (D-226 /// T-1124 amendment §1, "recommended"; extended T-1150 to key on /// granularity too): if a `DeriveWindow` item for the SAME `(connection, /// body, granularity)` is still sitting in the pending queue (not yet /// dispatched to a Rayon worker), it is replaced in place by the new one /// — a pan-burst that queues several window requests for the same /// connection+body+granularity before the first is dispatched collapses /// to one derive. A district-spacing and quarter-spacing request for the /// same connection+body do NOT coalesce with each other — they're /// separate in-flight derives, not competing updates to the same rung. /// /// Deliberately does **not** attempt to cancel an item already dispatched /// to a Rayon worker (no cancellation channel exists, and the amendment /// does not require it — bounding queue buildup is the goal, not /// interrupting in-flight compute). `item` MUST be a `DeriveWindow` /// variant; any other variant is submitted via the plain `submit` path /// with no coalescing (this method still accepts it for caller /// convenience, but the supersede check is a no-op when /// `window_supersede_key()` returns `None`). pub fn submit_window(&self, item: GenWorkItem, priority: GenPriority) { if let Some(key) = item.window_supersede_key() { let key = (key.0, key.1.to_string(), key.2); let mut pending = self.pending.lock().unwrap(); pending.retain(|q| { q.item .window_supersede_key() .map(|k| (k.0, k.1.to_string(), k.2) != key) .unwrap_or(true) }); let pos = pending .iter() .position(|q| q.priority > priority) .unwrap_or(pending.len()); pending.insert(pos, QueuedWork { priority, item }); drop(pending); self.dispatch_next(); } else { self.submit(item, priority); } } /// Submit a `DeriveStepCanvas` item with per-connection-per-rung /// coalescing (T-1181, mirroring `submit_window`'s exact discipline): if /// an item for the SAME `(connection, body, rung)` is still pending /// (not yet dispatched), it is replaced in place — a step-cross burst /// that queues several requests for the same connection+body+rung before /// the first is dispatched collapses to one derive. `item` MUST be a /// `DeriveStepCanvas` variant; any other variant falls through to plain /// `submit` with no coalescing. pub fn submit_step_canvas(&self, item: GenWorkItem, priority: GenPriority) { if let Some(key) = item.step_canvas_supersede_key() { let key = (key.0, key.1.to_string(), key.2); let mut pending = self.pending.lock().unwrap(); pending.retain(|q| { q.item .step_canvas_supersede_key() .map(|k| (k.0, k.1.to_string(), k.2) != key) .unwrap_or(true) }); let pos = pending .iter() .position(|q| q.priority > priority) .unwrap_or(pending.len()); pending.insert(pos, QueuedWork { priority, item }); drop(pending); self.dispatch_next(); } else { self.submit(item, priority); } } /// Drain all completed items from the channel and dispatch pending work. /// /// Call once per tick from the main thread. Returns all completions /// available without blocking. After draining, dispatches as many pending /// items as there are free thread slots — this is the point where priority /// ordering matters, since the pool was saturated when items were submitted. pub fn drain_completions(&self) -> Vec { let mut out = Vec::new(); while let Ok(c) = self.completion_rx.try_recv() { out.push(c); } // Fill any newly-freed slots. for _ in 0..out.len() { self.dispatch_next(); } out } /// Number of items waiting in the pending queue. pub fn pending_count(&self) -> usize { self.pending.lock().unwrap().len() } // Dispatch the highest-priority pending item to the Rayon pool. // // Gated on in_flight_count < n_threads — applies to all work item types, // not just AnalyzeBody. When the pool is full, items stay in the sorted // pending Vec so priority ordering is consulted on the next free slot. fn dispatch_next(&self) { let item = { let count = self.in_flight_count.lock().unwrap(); if *count >= self.n_threads { return; } drop(count); let mut pending = self.pending.lock().unwrap(); if pending.is_empty() { return; } pending.remove(0).item }; // Mark body as in-flight (AnalyzeBody dedup). if let Some(body_id) = item.body_id() { self.in_flight.lock().unwrap().insert(body_id.to_string()); } // Increment general in-flight counter for all item types. *self.in_flight_count.lock().unwrap() += 1; let tx = self.completion_tx.clone(); let in_flight = Arc::clone(&self.in_flight); let in_flight_count = Arc::clone(&self.in_flight_count); let terrain_cache = Arc::clone(&self.terrain_cache); self.pool.spawn(move || { let completion = run_work_item(&item, &terrain_cache); // Un-mark body dedup set (AnalyzeBody only). if let Some(body_id) = item.body_id() { in_flight.lock().unwrap().remove(body_id); } // Decrement general counter for all item types. *in_flight_count.lock().unwrap() -= 1; let _ = tx.send(completion); }); } } impl Default for GenerationQueue { fn default() -> Self { Self::new() } } // --------------------------------------------------------------------------- // TerrainAnalysisCache (T-1137, PR #187 review — Tyre C1) // --------------------------------------------------------------------------- /// Small per-body LRU of re-derived [`TerrainAnalysis`] (~1.5–2 MB/entry), /// consulted by the `DeriveWindow` execution path before paying the ~45 ms /// `run_layer1` re-derive cost (T-1137 binding decision — see the /// `DeriveWindow` variant doc on why `TerrainAnalysis` is re-derived rather /// than cached on `BodyWorldState` at all). /// /// **Why this exists (PR #187 review finding, binding):** the original /// T-1137 landing called `run_layer1` unconditionally on every `DeriveWindow` /// work item — nothing memoized it within a body, so the *dominant* usage /// pattern (panning around ONE body, many windows) paid the ~45 ms re-derive /// on every single window instead of just the first. This cache closes that /// gap: first window on a body pays the full re-derive and populates the /// entry; every subsequent window on the SAME body (until eviction) hits the /// cache and skips straight to the ~7–29 ms `build_district_window_layer` /// pack (§4's actual per-window number). /// /// **Shape:** `Arc>` — lives on [`GenerationQueue`] alongside /// `in_flight`/`in_flight_count` (the same "shared state cloned into every /// Rayon closure" pattern), because `run_work_item` executes ON a Rayon /// worker thread, potentially concurrently with other workers when /// `n_threads > 1`; a queue-owned, plain (non-`Resource`) cache is the /// correct home — `handle_atlas_request`'s `DistrictWindowCache` is main- /// thread-only (D-225 poll loop) and cannot be reused here. /// /// **Capacity (8, ~16 MB worst case):** deliberately small relative to /// `BodyWorldStateCache::CACHE_CAPACITY` (50) — this caches a re-derive /// shortcut for bodies actually being window-browsed RIGHT NOW, not a /// body-indexed store meant to grow with session length. True LRU /// (access-recency, mirroring `BodyWorldStateCache`'s own eviction policy) /// rather than `DistrictWindowCache`'s FIFO-by-insertion: unlike a /// D-227-pure derived window (valid forever, no recency signal to track), /// which body a player keeps panning around IS a recency signal, so /// access-order eviction is the right fit here. /// /// **T-1170 (Ruling 4b) — also retains the `Layer1Output` produced alongside /// `TerrainAnalysis`, not just the latter.** The original entry only kept /// `TerrainAnalysis` and discarded `run_layer1`'s `Layer1Output` half (the /// `let (_, ta) = run_layer1(...)` at the old call site) — fine for the /// six-array district/quarter/vegetation classification the window path /// used before this ticket, but it meant the window derive path had no way /// to know which river edges exist near a window without re-running the /// whole ~45 ms drainage pass a second time. Since this cache already pays /// that cost once per body and holds the result for the session, keeping /// BOTH halves of `run_layer1`'s return value is free — `Layer1Output` /// itself is small (a `RiverNetwork` + basin list + attractor list, not the /// full grid) relative to `TerrainAnalysis`'s ~1.5–2 MB of dense per-cell /// Vecs. /// `pub(crate)` (D-256(d), T-1174): `plugin.rs`'s tests construct one /// directly to unit-test `resolve_settlement_morphology_zone` without /// spinning up a full `GenerationQueue`. Otherwise entirely internal to this /// module's Rayon-thread execution path. #[derive(Debug)] pub(crate) struct TerrainAnalysisCache { entries: std::collections::BTreeMap, /// Monotonic access counter (substitutes for `BodyWorldStateCache`'s /// `SimTick` — there is no tick concept on a background Rayon thread). clock: u64, capacity: usize, } /// Default capacity for [`TerrainAnalysisCache`] (PR #187 review — Tyre C1 /// binding numbers: "capacity ~8, ~2MB/entry = ~16MB worst case"). const TERRAIN_ANALYSIS_CACHE_CAPACITY: usize = 8; impl TerrainAnalysisCache { fn new(capacity: usize) -> Self { Self { entries: std::collections::BTreeMap::new(), clock: 0, capacity, } } /// Test-only constructor alias (D-256(d)) — `new` stays private-module /// idiomatic; this is the `pub(crate)` door for `plugin.rs`'s tests. #[cfg(test)] pub(crate) fn new_for_test(capacity: usize) -> Self { Self::new(capacity) } /// Look up a cached `(Layer1Output, TerrainAnalysis)` pair for `body_id`, /// re-deriving via `run_layer1_with_moisture` on a miss and inserting the /// result (evicting the LRU entry first if at capacity). Bumps the access /// clock on both a hit and a fresh insert (both are "this body was just /// used"). /// /// Returns both halves of `run_layer1`'s output (T-1170 Ruling 4b) — the /// window derive path (`GenWorkItem::DeriveWindow`) needs `Layer1Output`'s /// `RiverNetwork` to know which river edges exist near the requested /// window, in addition to the `TerrainAnalysis` it always needed. /// /// `body_params` (T-1184) is `Option` — `None` when the caller has no DB /// row for this body, matching the same "params absent → fall back" /// posture every other `body_params: Option<&BodyParams>` consumer in this /// module already has (`resolve_settlement_morphology_zone`). Falling /// through to `run_layer1`'s body-agnostic moisture default in that case /// is a body-classification-quality concern (which basins read Endorheic /// vs. Overflow), never a correctness one — lake EXTENT never depends on /// moisture (only the elevation-geometry-gated filled-surface comparison /// does; see `district_profile::derive_morphology_zone`'s lake tier). fn get_or_derive( &mut self, body_id: &str, heightmap: &crate::atlas::heightmap::BodyHeightmap, body_params: Option<&BodyParams>, ) -> (Layer1Output, TerrainAnalysis) { self.clock += 1; let now = self.clock; if let Some((l1, ta, last_used)) = self.entries.get_mut(body_id) { *last_used = now; return (l1.clone(), ta.clone()); } let (l1, ta) = match body_params { Some(params) => crate::atlas::layer1::run_layer1_with_moisture( heightmap, crate::atlas::district_profile::derive_moisture_ceiling_q(params), ), None => crate::atlas::layer1::run_layer1(heightmap), }; if self.entries.len() >= self.capacity && !self.entries.contains_key(body_id) { if let Some(victim) = self .entries .iter() .min_by_key(|(_, (_, _, last_used))| *last_used) .map(|(id, _)| id.clone()) { self.entries.remove(&victim); } } self.entries .insert(body_id.to_string(), (l1.clone(), ta.clone(), now)); (l1, ta) } #[cfg(test)] fn len(&self) -> usize { self.entries.len() } #[cfg(test)] fn contains(&self, body_id: &str) -> bool { self.entries.contains_key(body_id) } } /// D-256(d): resolve a settlement's `MorphologyZone` at its EXACT world /// position via `derive_at_metres`, using the terrain cache to reach a /// `TerrainAnalysis` without a disk re-read (`heightmap` is already in /// memory, reconstructed once at dispatch time from `BodyWorldState`'s /// cached working-grid data). /// /// Returns `None` when `body_params` is absent (no DB row for this body) — /// the caller then leaves `context.morphology_zone` at its dispatch-time /// `AlluvialPlain` stub, the same fallback the pre-D-256 dispatch-time /// district-grid lookup used for an empty grid. /// /// Extracted from `run_work_item`'s `GenerateSkeleton` arm so the resolution /// itself is unit-testable without going through the full skeleton-generation /// pipeline (`QuarterSkeleton` does not expose `morphology_zone` directly — /// it only affects derived fields like `layout_mode`/corridors). pub(crate) fn resolve_settlement_morphology_zone( terrain_cache: &Arc>, body_id: &str, body_params: Option<&BodyParams>, settlement_world_m: (f64, f64), body_seed: SeedChain, heightmap: &crate::atlas::heightmap::BodyHeightmap, ) -> Option { let params = body_params?; let (_l1, ta) = terrain_cache .lock() .unwrap() .get_or_derive(body_id, heightmap, Some(params)); let climate = ClimateConstants::default(); let profile = crate::atlas::district_profile::derive_at_metres( body_seed, body_id, params, &ta, settlement_world_m.0, settlement_world_m.1, &climate, 0.0, &[], ); Some(profile.morphology_zone) } // --------------------------------------------------------------------------- // Work execution stub // --------------------------------------------------------------------------- /// Execute one work item. This is the Rayon task body (off the tick thread). /// /// `AnalyzeBody` runs the real Layer-1 cascade (#968, D-225); `GenerateSkeleton` /// runs the real plan phase (#957, D-229) producing the skeleton + block tags; /// `FillChunk` runs the real derive phase (T-987, D-230) producing the building /// shell from the pre-resolved tags. /// /// `terrain_cache` serves `DeriveWindow`'s `TerrainAnalysis` re-derive /// shortcut (T-1137, PR #187 review C1) — unused by every other variant /// (they don't touch `TerrainAnalysis` at all, or — `AnalyzeBody` — derive it /// once already as part of the normal in-cascade path, T-1044). fn run_work_item( item: &GenWorkItem, terrain_cache: &Arc>, ) -> GenCompletion { match item { GenWorkItem::AnalyzeBody { body_id, heightmap_path, sea_level, body_seed, cities, dominant_faction, body_params, } => 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 }; // Run the full cascade through Region (T-1113), the terminal // layer. It subsumes RoadGraph (T-1038), Settlement, // DistrictProfile (T-1023), and all prior layers. // DistrictProfile and Region derivation both gate on // body_params internally (skipped when absent — e.g. a body // with no params row), but the road graph needs no body // params, so it runs for every analyzed body. let up_to = CascadeLayer::Region; let snapshot = run_cascade_from_heightmap( *body_seed, working, cities, dominant_faction.as_deref(), body_params.as_deref(), up_to, ); GenCompletion::BodyAnalyzed { body_id: body_id.clone(), state: Box::new(snapshot.into_body_world_state()), } } Err(e) => GenCompletion::Failed { item: item.clone(), reason: format!("heightmap load failed: {e}"), }, }, GenWorkItem::GenerateSkeleton { city_id, body_id, context, quarter_id, chain, economic_role, population, founding_age_years, exterior_catalog, settlement_world_m, body_params, body_seed, heightmap, } => { // D-256(d): resolve morphology_zone at the settlement's EXACT world // position, here at execution time — this is where TerrainAnalysis // is reachable (BodyWorldState drops it, D-203/T-1048). A survey // cell's centre can be hundreds of km from a settlement near its // edge; the exact-position derive closes that annotation-vs-canvas // disagreement class before T-1181/T-1182 draw batch-judged // annotations onto window-derived canvases. let mut resolved_context = (**context).clone(); if let Some(zone) = resolve_settlement_morphology_zone( terrain_cache, body_id, body_params.as_deref(), *settlement_world_m, *body_seed, heightmap, ) { resolved_context.morphology_zone = zone; } // Build the Phase 1 skeleton from the pre-resolved D-199 context. // `economic_role`, `population`, and `founding_age_years` are the // D-199 raw fields carried alongside the context because // `generate_quarter_skeleton` accepts them as separate parameters. let skeleton = generate_quarter_skeleton( &resolved_context, *population, economic_role, *quarter_id, *founding_age_years, *chain, ); // Step-3 building-property tags per footprint (D-229, #957): subdivide // each block into building plots and tag them. `exterior_catalog` // (D-235, T-988) resolves each tag's BuildingExteriorTag in the // same pass. let block_tags = assign_all_block_tags( &skeleton, &resolved_context, economic_role, *founding_age_years, *chain, exterior_catalog, ); GenCompletion::SkeletonGenerated { city_id: *city_id, body_id: body_id.clone(), state: Box::new(QuarterWorldState { skeleton, block_tags, }), } } GenWorkItem::FillChunk { quarter_id, block_pos, sub_chunk, block_tags, } => { // D-230 derive phase: pure rectangle-containment + z-range shell fill over // the pre-resolved tags. No cache read here — that is what keeps FillChunk // trivially fast and re-derivable (D-227). let filled = fill_chunk(*quarter_id, *block_pos, *sub_chunk, block_tags); GenCompletion::ChunkFilled { filled: Box::new(filled), } } GenWorkItem::DeriveWindow { body_id, conn_id: _, // routing-only (queue-level coalescing); the derive itself is connection-agnostic heightmap_path, sea_level, body_seed, body_params, center, n, granularity, min_wl_m, } => match load_heightmap_png(heightmap_path, body_id, *sea_level) { Ok(hm) => { // Same GRID_W×GRID_H downsample AnalyzeBody applies (D-202) — the // window derive must run on the SAME working-grid resolution the // whole-body cascade uses, or the TRUE DistrictPos → world-metres // mapping between the two views would disagree (derive_district // maps DistrictPos through ta.w/ta.h, T-1137 decision note). This // is resolution consistency (ta.w/ta.h must match), a DIFFERENT // concern from D-256's survey-raster/true-district namespace // collision — both views here already address the true D-243 // grid, so D-256 doesn't touch this comment's claim. let working = if hm.width > GRID_W || hm.height > GRID_H { hm.downsample(GRID_W, GRID_H) } else { hm }; // (Layer1Output, TerrainAnalysis) via the per-body LRU (T-1137 // binding decision + PR #187 review C1, extended T-1170 Ruling // 4b to retain Layer1Output too): first window on a body pays // the ~45 ms run_layer1 re-derive and populates the cache // entry; every subsequent window on the SAME body (until // eviction) hits the cache and skips straight to the ~7–29 ms // per-window pack below. This is the memoized form of the SAME // workaround aliveness_probe --render uses when // CascadeSnapshot.terrain_analysis is None (it has no cache — // a one-shot CLI run doesn't need one). // // `l1.river_network` is what lets the window derive know which // river edges exist near this window (T-1170 A2) without a // second drainage pass — the fix for the former // `let (_, ta) = run_layer1(...)` discard (Ruling 4b). let (l1, ta) = terrain_cache.lock().unwrap().get_or_derive( body_id, &working, Some(body_params), ); let climate = ClimateConstants::default(); let layer = build_district_window_layer( *body_seed, body_id, body_params, &ta, &l1.river_network, *center, *n, &climate, *granularity, *min_wl_m, ); GenCompletion::WindowDerived { body_id: body_id.clone(), layer: Box::new(layer), } } Err(e) => GenCompletion::Failed { item: item.clone(), reason: format!("heightmap load failed: {e}"), }, }, GenWorkItem::DeriveStepCanvas { body_id, conn_id: _, // routing-only (queue-level coalescing); the derive itself is connection-agnostic heightmap_path, sea_level, body_seed, body_params, placements, rung, center, extent, min_wl_m, } => match load_heightmap_png(heightmap_path, body_id, *sea_level) { Ok(hm) => { // Same GRID_W×GRID_H downsample AnalyzeBody/DeriveWindow apply // (D-202) — the true DistrictPos<->world-metres mapping must // stay consistent across every derive path. let working = if hm.width > GRID_W || hm.height > GRID_H { hm.downsample(GRID_W, GRID_H) } else { hm }; // Same per-body TerrainAnalysisCache LRU DeriveWindow already // populates (T-1137 binding decision, extended here) — keyed // on body_id alone, so a step-canvas request and a legacy // district_window request for the same body share one // ~45 ms run_layer1 re-derive, never pay it twice. // body_params threads the real moisture ceiling into the // T-1184 hydrology solve (merge seam resolution). let (l1, ta) = terrain_cache.lock().unwrap().get_or_derive( body_id, &working, Some(body_params.as_ref()), ); let climate = ClimateConstants::default(); let raw = crate::atlas::step_canvas::build_step_canvas( *body_seed, body_id, body_params, &ta, &l1.river_network, placements, *rung, *center, *extent, &climate, *min_wl_m, ); let canvas = crate::atlas::step_canvas::encode_step_canvas(&raw); GenCompletion::StepCanvasDerived { body_id: body_id.clone(), rung: *rung, center: *center, extent: *extent, min_wl_m: *min_wl_m, canvas: Box::new(canvas), } } Err(e) => GenCompletion::Failed { item: item.clone(), reason: format!("heightmap load failed: {e}"), }, }, } } /// Build a [`GenWorkItem::FillChunk`] for one 64 m sub-chunk of a quarter, pulling the /// covering block's tags out of the cached `QuarterWorldState` (D-230 derive phase, T-987). /// /// Pure (no queue/cache handle), so it unit-tests without a running app. The D-230 /// precondition — "`FillChunk` is only dispatched after `SkeletonGenerated` for that /// district has been processed; if absent, re-enqueue at `High`" — is the caller's /// cache lookup: this constructor only runs once the `QuarterWorldState` exists. A /// block with no buildings yields empty `block_tags` (→ an empty, terrain-only shell), /// which is a valid ready state, not a not-yet-generated one. pub fn build_fill_chunk_item( quarter: &QuarterWorldState, block_pos: (u8, u8), sub_chunk: (u8, u8), ) -> GenWorkItem { let block_tags = quarter .block_tags .get(&block_pos) .cloned() .unwrap_or_default(); GenWorkItem::FillChunk { quarter_id: quarter.skeleton.quarter_id, block_pos, sub_chunk, block_tags, } } // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- #[cfg(test)] mod tests { use super::*; use std::time::Duration; fn make_queue() -> GenerationQueue { 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), cities: vec![], dominant_faction: None, body_params: None, // T-1023: no body params in queue-mechanic unit tests } } /// Build a minimal `GenerateSkeleton` work item with a stub context. /// /// The stub context uses Commission/Regional/Urban defaults — the same /// values the existing skeleton_gen tests use. These tests exercise queue /// mechanics (ordering, saturation, drain), not economic read-set content. fn gen_skeleton(city_id: u64) -> GenWorkItem { use crate::simulation::generator::{ BulkClass, CityGenerationContext, FoundingOrientation, MorphologyZone, PoliticalArchetype, ProductionUbiquity, SettingType, WorldTier, }; GenWorkItem::GenerateSkeleton { city_id, body_id: format!("TestBody{city_id}"), context: Box::new(CityGenerationContext { city_id, political_archetype: PoliticalArchetype::Commission, prosperity_baseline_bps: 6_000, surrounding_biome: SettingType::Urban, road_entry_directions: vec![], footprint_radius_km: 5.0, founding_orientation: FoundingOrientation::Cardinal, world_tier: WorldTier::Regional, morphology_zone: MorphologyZone::AlluvialPlain, trait_selection: vec![], dominant_bulk_class: BulkClass::NonPhysical, dominant_production_ubiquity: ProductionUbiquity::Common, geographic_sector: None, body_district_type_mix: vec![], settlement_district_pos: (0, 0), district_dominant_by_type: Default::default(), swerve_rates_bps: (0, 0), swerve_foreign_pool: vec![], swerve_heritage_pool: vec![], }), quarter_id: city_id * 10, chain: SeedChain::root(42 + city_id), economic_role: "service_mixed".to_string(), population: 500_000, founding_age_years: 200, exterior_catalog: ExteriorCatalog::default(), // These tests exercise queue mechanics, not the D-256(d) // exact-position resolution — `body_params: None` skips it // entirely (the same fallback path an empty district grid used // pre-D-256), so the position/heightmap values below are inert. settlement_world_m: (0.0, 0.0), body_params: None, body_seed: SeedChain::for_body(42 + city_id, &format!("TestBody{city_id}")), heightmap: std::sync::Arc::new(crate::atlas::heightmap::BodyHeightmap { body_id: format!("TestBody{city_id}"), width: 1, height: 1, data: vec![0.5], sea_level: 0.3, }), } } #[test] fn submit_and_drain() { let q = make_queue(); 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, state } if body_id == "TestBody" && state.heightmap_width == 32 && state.heightmap_height == 16 )); } #[test] fn dedup_analyze_body() { let q = make_queue(); // Submit the same body twice before it can complete. 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. assert_eq!(completions.len(), 1); } #[test] fn priority_ordering() { // Submit three items rapidly; Immediate should be dispatched first. // Uses 3 threads so all items can dispatch without hitting saturation. let q = GenerationQueue::with_threads(3); // Using GenerateSkeleton (no dedup logic) to test ordering directly. q.submit(gen_skeleton(1), GenPriority::Low); q.submit(gen_skeleton(2), GenPriority::Immediate); q.submit(gen_skeleton(3), GenPriority::Medium); std::thread::sleep(Duration::from_millis(100)); let completions = q.drain_completions(); assert_eq!(completions.len(), 3); } #[test] fn priority_ordering_respected_under_saturation() { // Single-thread queue: in_flight_count saturates at 1, so the second // item stays in the pending Vec and is dispatched in priority order. // Uses AnalyzeBody (distinct body_ids) so all paths — dedup set AND // in_flight_count — are exercised. let q = GenerationQueue::with_threads(1); // Submit Low first, then Immediate. With 1 thread: // - "BodyA" (Low) dispatches immediately (pool empty). // - "BodyB" (Immediate) is inserted at index 0 of the sorted pending // 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(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). let first = q.drain_completions(); // Wait for BodyB to complete. std::thread::sleep(Duration::from_millis(50)); let second = q.drain_completions(); 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") ); } #[test] fn drain_empty_returns_empty() { let q = make_queue(); let result = q.drain_completions(); assert!(result.is_empty()); } #[test] fn pending_count_decreases_after_completion() { let q = make_queue(); q.submit( GenWorkItem::FillChunk { quarter_id: 99, block_pos: (0, 0), sub_chunk: (0, 0), block_tags: vec![], }, GenPriority::High, ); std::thread::sleep(Duration::from_millis(50)); let completions = q.drain_completions(); assert!(!completions.is_empty() || q.pending_count() == 0); } /// A `QuarterWorldState` with one building in block (0,0), used to exercise the /// full FillChunk path (build item from cached state → Rayon → completion). fn quarter_with_one_building() -> QuarterWorldState { use crate::atlas::tile_condition::TileCondition; use crate::simulation::generator::{ ArchitectureFlavorRef, BuildingEntryClass, BuildingExteriorTag, BuildingPropertyTag, ConstructionEra, EraCause, FacadeRhythm, FloorExtent, FloorHeightProfile, HsvColor, QuarterSkeleton, RoofForm, SetbackTier, StreetSurface, TileRect, WallMaterial, ZoneTypeId, }; use std::collections::BTreeMap; let mut block_tags: BTreeMap<(u8, u8), Vec> = BTreeMap::new(); block_tags.insert( (0, 0), vec![BuildingPropertyTag { zone_type_id: ZoneTypeId::new("residential_low"), footprint: TileRect::new(4, 4, 6, 6), extent: FloorExtent { base_floor: 0, floor_count: 2, heights: FloorHeightProfile::Uniform(3), }, entry_class: BuildingEntryClass::Public, flavor_ref: ArchitectureFlavorRef::InVocabulary(0), era: ConstructionEra::Founding, era_cause: EraCause::Original, initial_condition: TileCondition::Intact, exterior: BuildingExteriorTag { wall_material: WallMaterial::Generic, roof_form: RoofForm::Generic, facade_rhythm: FacadeRhythm::Generic, setback_tier: SetbackTier::Standard, color: HsvColor { hue: 0, sat: 0, val: 5_000, }, street_surface: StreetSurface::Generic, }, doors: Vec::new(), }], ); QuarterWorldState { skeleton: QuarterSkeleton { quarter_id: 4242, ..Default::default() }, block_tags, } } #[test] fn build_fill_chunk_item_pulls_block_tags() { let quarter = quarter_with_one_building(); let item = build_fill_chunk_item(&quarter, (0, 0), (0, 0)); let GenWorkItem::FillChunk { quarter_id, block_pos, sub_chunk, block_tags, } = item else { panic!("expected FillChunk"); }; assert_eq!(quarter_id, 4242); assert_eq!(block_pos, (0, 0)); assert_eq!(sub_chunk, (0, 0)); assert_eq!( block_tags.len(), 1, "covering block's tags must be resolved" ); // A block with no buildings is a valid empty fill, not a missing-skeleton error. let empty = build_fill_chunk_item(&quarter, (3, 3), (0, 0)); let GenWorkItem::FillChunk { block_tags, .. } = empty else { panic!("expected FillChunk"); }; assert!(block_tags.is_empty(), "empty block → empty tags"); } #[test] fn fill_chunk_round_trip_produces_populated_shell() { let q = make_queue(); let quarter = quarter_with_one_building(); q.submit( build_fill_chunk_item(&quarter, (0, 0), (0, 0)), GenPriority::High, ); std::thread::sleep(Duration::from_millis(50)); let completions = q.drain_completions(); assert_eq!(completions.len(), 1); let GenCompletion::ChunkFilled { filled } = &completions[0] else { panic!("expected ChunkFilled, got {:?}", completions[0]); }; assert_eq!(filled.quarter_id, 4242); assert_eq!(filled.chunk_in_quarter(), (0, 0)); assert!( filled.voxel_count() > 0, "a chunk containing a building must derive shell voxels" ); } // ------------------------------------------------------------------- // DeriveWindow / submit_window coalescing (D-226 T-1124 amendment, T-1137) // ------------------------------------------------------------------- /// Build a `DeriveWindow` work item pointing at a tiny test heightmap, /// mirroring `analyze()`'s fixture shape. `granularity` defaults to /// district (matching every pre-T-1150 call site) via /// `derive_window_at()` below — extended (PR #191 review, Hoshe 2) so /// coalescing tests can exercise the granularity axis of /// `window_supersede_key()` without a second near-duplicate helper. fn derive_window(body_id: &str, conn_id: ConnectionId, center: DistrictPos) -> GenWorkItem { derive_window_at(body_id, conn_id, center, WindowGranularity::District) } fn derive_window_at( body_id: &str, conn_id: ConnectionId, center: DistrictPos, granularity: WindowGranularity, ) -> GenWorkItem { GenWorkItem::DeriveWindow { body_id: body_id.to_string(), conn_id, heightmap_path: test_heightmap_path(), sea_level: 0.3, body_seed: SeedChain::for_body(42, body_id), body_params: Box::new(BodyParams { hydrosphere: Some("ocean".into()), atmosphere: Some("breathable".into()), planet_class: Some("temperate".into()), body_radius_km: Some(6371.0), ..Default::default() }), center, n: 4, granularity, min_wl_m: 0, } } /// The full `DeriveWindow` → Rayon → `WindowDerived` round trip: the /// real re-derive-via-`run_layer1` path (T-1137 binding decision) runs /// end to end and produces a populated `DistrictWindowLayer`. #[test] fn derive_window_round_trip_produces_populated_layer() { let q = make_queue(); q.submit( derive_window("TestBody", ConnectionId(0), (2, -1)), GenPriority::Immediate, ); std::thread::sleep(Duration::from_millis(150)); let completions = q.drain_completions(); assert_eq!(completions.len(), 1); let GenCompletion::WindowDerived { body_id, layer } = &completions[0] else { panic!("expected WindowDerived, got {:?}", completions[0]); }; assert_eq!(body_id, "TestBody"); assert_eq!(layer.center, (2, -1)); assert_eq!(layer.n, 4); assert_eq!(layer.morphology.len(), 16); assert_eq!(layer.elev_q.len(), 16); assert_eq!(layer.temp_dc.len(), 16); assert_eq!(layer.moisture_q.len(), 16); assert_eq!(layer.vegetation.len(), 16); assert_eq!(layer.glaciation.len(), 16); } /// `submit_window` coalescing (D-226 T-1124 amendment §1 "recommended"): /// two `DeriveWindow` items for the SAME `(connection, body)` queued /// while the pool is saturated collapse to ONE pending entry — the /// second submission replaces the first rather than queuing alongside it. #[test] fn submit_window_coalesces_same_connection_and_body() { // Single-thread pool: the first item occupies the only worker, so // subsequent DeriveWindow submissions stay in `pending` long enough // to inspect (mirrors `priority_ordering_respected_under_saturation`'s // saturation trick). Uses `analyze()` (real cascade work — measurable // latency), NOT `FillChunk` (documented "trivially fast" — it can // complete before the next `submit_window` call even runs, which // would make `pending_count()` observe 0 instead of 1, a real race // this test hit before switching occupiers). let q = GenerationQueue::with_threads(1); q.submit(analyze("Occupier"), GenPriority::Low); let conn = ConnectionId(7); q.submit_window(derive_window("Coal", conn, (0, 0)), GenPriority::Immediate); assert_eq!( q.pending_count(), 1, "one DeriveWindow queued behind the saturating item" ); // A second DeriveWindow for the SAME (connection, body) supersedes // the first — pending count stays at 1, not 2. q.submit_window(derive_window("Coal", conn, (5, 5)), GenPriority::Immediate); assert_eq!( q.pending_count(), 1, "same (connection, body) DeriveWindow must supersede, not queue alongside" ); // Drain everything and confirm exactly one WindowDerived for "Coal", // carrying the SECOND (superseding) center — not the first. std::thread::sleep(Duration::from_millis(150)); let mut completions = q.drain_completions(); std::thread::sleep(Duration::from_millis(150)); completions.extend(q.drain_completions()); let window_completions: Vec<_> = completions .iter() .filter_map(|c| { if let GenCompletion::WindowDerived { body_id, layer } = c { if body_id == "Coal" { return Some(layer); } } None }) .collect(); assert_eq!( window_completions.len(), 1, "exactly one WindowDerived for the coalesced body, not two" ); assert_eq!( window_completions[0].center, (5, 5), "the surviving item must be the SECOND (superseding) submission" ); } /// `submit_window` does NOT coalesce across different connections or /// different bodies — only an exact `(connection, body)` match supersedes. #[test] fn submit_window_does_not_coalesce_different_keys() { let q = GenerationQueue::with_threads(1); // See `submit_window_coalesces_same_connection_and_body`'s comment on // why the occupier must be `analyze()`, not `FillChunk`. q.submit(analyze("Occupier2"), GenPriority::Low); // Different connections, same body — must NOT coalesce. q.submit_window( derive_window("Shared", ConnectionId(1), (0, 0)), GenPriority::Immediate, ); q.submit_window( derive_window("Shared", ConnectionId(2), (1, 1)), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 2, "different connections requesting the same body must NOT coalesce" ); } /// **PR #191 review, Hoshe 2 — zero coverage before this test.** /// `window_supersede_key()`'s doc claims district and quarter requests /// for the SAME `(connection, body)` are separate in-flight slots (the /// key is `(conn_id, body_id, granularity)`, not `(conn_id, body_id)`). /// Two submissions for the same connection+body but DIFFERENT /// granularity must NOT coalesce — both survive as independent pending /// items. #[test] fn submit_window_does_not_coalesce_different_granularity() { let q = GenerationQueue::with_threads(1); // See `submit_window_coalesces_same_connection_and_body`'s comment on // why the occupier must be `analyze()`, not `FillChunk`. q.submit(analyze("Occupier3"), GenPriority::Low); let conn = ConnectionId(9); q.submit_window( derive_window_at("GranBody", conn, (0, 0), WindowGranularity::District), GenPriority::Immediate, ); q.submit_window( derive_window_at("GranBody", conn, (0, 0), WindowGranularity::Quarter), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 2, "same (connection, body) but DIFFERENT granularity must NOT coalesce — \ district and quarter are separate in-flight slots" ); } /// The coalescing-DOES-happen counterpart to the test above: two /// submissions for the SAME `(connection, body, granularity)` still /// collapse to one pending item — confirms the granularity axis didn't /// accidentally loosen the existing same-key coalescing behavior. #[test] fn submit_window_coalesces_same_connection_body_and_granularity() { let q = GenerationQueue::with_threads(1); q.submit(analyze("Occupier4"), GenPriority::Low); let conn = ConnectionId(11); q.submit_window( derive_window_at("SameGranBody", conn, (0, 0), WindowGranularity::Quarter), GenPriority::Immediate, ); q.submit_window( derive_window_at("SameGranBody", conn, (5, 5), WindowGranularity::Quarter), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 1, "same (connection, body, granularity) must still coalesce to one pending item" ); } /// **PR #192 review — Hoshe 1: zero coalescing coverage for /// `WindowGranularity::Region` before this test**, despite Region being /// the highest-fan-out path (progressive capped-density tiling fires /// multiple concurrent Region `DeriveWindow` items per pan/zoom). Mirrors /// `submit_window_does_not_coalesce_different_granularity`'s pattern /// exactly, substituting Region for Quarter: a Region request and a /// District request for the SAME `(connection, body)` are separate /// in-flight slots (the coalescing key is `(conn_id, body_id, /// granularity)`) and must NOT coalesce — both survive as independent /// pending items. #[test] fn submit_window_does_not_coalesce_region_and_district() { let q = GenerationQueue::with_threads(1); // See `submit_window_coalesces_same_connection_and_body`'s comment on // why the occupier must be `analyze()`, not `FillChunk`. q.submit(analyze("Occupier5"), GenPriority::Low); let conn = ConnectionId(13); q.submit_window( derive_window_at("OrbitalGranBody", conn, (0, 0), WindowGranularity::Region), GenPriority::Immediate, ); q.submit_window( derive_window_at("OrbitalGranBody", conn, (0, 0), WindowGranularity::District), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 2, "same (connection, body) but Region vs. District must NOT coalesce — \ separate in-flight slots, same as the existing District/Quarter pair" ); } /// The coalescing-DOES-happen counterpart to the test above, for Region /// specifically: two submissions for the SAME `(connection, body, /// Region)` still collapse to one pending item — confirms Region's /// coalescing key behaves identically to District/Quarter's, not just /// that it avoids cross-granularity aliasing. #[test] fn submit_window_coalesces_same_connection_body_and_region_granularity() { let q = GenerationQueue::with_threads(1); q.submit(analyze("Occupier6"), GenPriority::Low); let conn = ConnectionId(15); q.submit_window( derive_window_at( "SameOrbitalGranBody", conn, (0, 0), WindowGranularity::Region, ), GenPriority::Immediate, ); q.submit_window( derive_window_at( "SameOrbitalGranBody", conn, (5, 5), WindowGranularity::Region, ), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 1, "same (connection, body, Region) must still coalesce to one pending item" ); } // ------------------------------------------------------------------- // DeriveStepCanvas / submit_step_canvas coalescing (T-1181, D-255(c)(d)) // // PR #201 review, Hoshe finding 2 — zero coverage before these tests. // `submit_step_canvas` is a near-verbatim sibling of `submit_window` // (same coalesce-in-place-on-supersede shape, same defensive // `dispatch_next` call), keyed on `step_canvas_supersede_key()` // (`(conn_id, body_id, rung)`) instead of `window_supersede_key()`'s // `(conn_id, body_id, granularity)`. Mirrors // `submit_window_coalesces_same_connection_and_body` + // `submit_window_does_not_coalesce_different_keys` exactly, substituting // `DeriveStepCanvas`/`rung` for `DeriveWindow`/`granularity`. // ------------------------------------------------------------------- /// Build a `DeriveStepCanvas` work item pointing at a tiny test /// heightmap, mirroring `derive_window_at`'s fixture shape. fn derive_step_canvas_at( body_id: &str, conn_id: ConnectionId, center: (i64, i64), rung: crate::atlas::step_canvas::StepCanvasRung, ) -> GenWorkItem { GenWorkItem::DeriveStepCanvas { body_id: body_id.to_string(), conn_id, heightmap_path: test_heightmap_path(), sea_level: 0.3, body_seed: SeedChain::for_body(42, body_id), body_params: Box::new(BodyParams { hydrosphere: Some("ocean".into()), atmosphere: Some("breathable".into()), planet_class: Some("temperate".into()), body_radius_km: Some(6371.0), ..Default::default() }), placements: Vec::new(), rung, center, extent: (4, 4), min_wl_m: 0, } } /// `submit_step_canvas` coalescing (T-1181, mirroring `submit_window`'s /// D-226 T-1124 amendment §1 discipline): two `DeriveStepCanvas` items /// for the SAME `(connection, body, rung)` queued while the pool is /// saturated collapse to ONE pending entry — the second submission /// replaces the first rather than queuing alongside it. #[test] fn submit_step_canvas_coalesces_same_connection_body_and_rung() { // Single-thread pool: the first item occupies the only worker, so // subsequent DeriveStepCanvas submissions stay in `pending` long // enough to inspect (same saturation trick // `submit_window_coalesces_same_connection_and_body` uses, and for // the same reason: `analyze()` is real measurable-latency cascade // work, unlike `FillChunk`, which could complete before the next // `submit_step_canvas` call even runs). let q = GenerationQueue::with_threads(1); q.submit(analyze("StepOccupier"), GenPriority::Low); let conn = ConnectionId(21); q.submit_step_canvas( derive_step_canvas_at( "Canvas", conn, (0, 0), crate::atlas::step_canvas::StepCanvasRung::Chunk, ), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 1, "one DeriveStepCanvas queued behind the saturating item" ); // A second DeriveStepCanvas for the SAME (connection, body, rung) // supersedes the first — pending count stays at 1, not 2. q.submit_step_canvas( derive_step_canvas_at( "Canvas", conn, (5_000, 5_000), crate::atlas::step_canvas::StepCanvasRung::Chunk, ), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 1, "same (connection, body, rung) DeriveStepCanvas must supersede, not queue alongside" ); // Drain everything and confirm exactly one StepCanvasDerived for // "Canvas", carrying the SECOND (superseding) center — not the // first. std::thread::sleep(Duration::from_millis(150)); let mut completions = q.drain_completions(); std::thread::sleep(Duration::from_millis(150)); completions.extend(q.drain_completions()); let step_canvas_completions: Vec<_> = completions .iter() .filter_map(|c| { if let GenCompletion::StepCanvasDerived { body_id, center, .. } = c { if body_id == "Canvas" { return Some(*center); } } None }) .collect(); assert_eq!( step_canvas_completions.len(), 1, "exactly one StepCanvasDerived for the coalesced body, not two" ); assert_eq!( step_canvas_completions[0], (5_000, 5_000), "the surviving item must be the SECOND (superseding) submission" ); } /// `submit_step_canvas` does NOT coalesce across different connections, /// different bodies, or different rungs — only an exact `(connection, /// body, rung)` match supersedes. Combines `submit_window_does_not_ /// coalesce_different_keys` (connection axis) and `submit_window_does_ /// not_coalesce_different_granularity` (rung axis) into one test, since /// `step_canvas_supersede_key` is a flat 3-tuple with no separate /// legacy-vs-v2 field split to test independently the way `WindowGranularity` /// needed. #[test] fn submit_step_canvas_does_not_coalesce_different_keys() { let q = GenerationQueue::with_threads(1); // See `submit_step_canvas_coalesces_same_connection_body_and_rung`'s // comment on why the occupier must be `analyze()`, not `FillChunk`. q.submit(analyze("StepOccupier2"), GenPriority::Low); // Different connections, same body, same rung — must NOT coalesce. q.submit_step_canvas( derive_step_canvas_at( "Shared", ConnectionId(1), (0, 0), crate::atlas::step_canvas::StepCanvasRung::District, ), GenPriority::Immediate, ); q.submit_step_canvas( derive_step_canvas_at( "Shared", ConnectionId(2), (1, 1), crate::atlas::step_canvas::StepCanvasRung::District, ), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 2, "different connections requesting the same body+rung must NOT coalesce" ); // Same connection, same body, but DIFFERENT rung — must NOT // coalesce (District vs. Chunk are separate in-flight slots). let conn = ConnectionId(23); q.submit_step_canvas( derive_step_canvas_at( "RungBody", conn, (0, 0), crate::atlas::step_canvas::StepCanvasRung::District, ), GenPriority::Immediate, ); q.submit_step_canvas( derive_step_canvas_at( "RungBody", conn, (0, 0), crate::atlas::step_canvas::StepCanvasRung::Chunk, ), GenPriority::Immediate, ); assert_eq!( q.pending_count(), 4, "same (connection, body) but DIFFERENT rung must NOT coalesce — \ District and Chunk are separate in-flight slots (2 from the \ connection-axis case above + 2 more here)" ); } // ------------------------------------------------------------------- // TerrainAnalysisCache (T-1137, PR #187 review — Tyre C1) // ------------------------------------------------------------------- fn window_test_hm() -> crate::atlas::heightmap::BodyHeightmap { use crate::atlas::heightmap::BodyHeightmap; let (w, h) = (32u32, 16u32); let n = (w * h) as usize; let data = (0..n) .map(|i| { let r = (i / w as usize) as f32 / h as f32; let c = (i % w as usize) as f32 / w as f32; (r * 0.6 + c * 0.4).min(1.0) }) .collect(); BodyHeightmap { body_id: "test".into(), width: w, height: h, data, sea_level: 0.3, } } /// A miss re-derives and populates the entry; a subsequent hit for the /// SAME body returns an equal `(Layer1Output, TerrainAnalysis)` pair /// (D-227: the same heightmap always derives to the same analysis) /// WITHOUT growing the cache — `len()` stays at 1, proving the second /// call short-circuited past `run_layer1` rather than /// deriving-then-overwriting. #[test] fn terrain_analysis_cache_hit_reuses_entry() { let mut cache = TerrainAnalysisCache::new(8); let hm = window_test_hm(); assert!(!cache.contains("BodyA")); let (l1_first, ta_first) = cache.get_or_derive("BodyA", &hm, None); assert_eq!(cache.len(), 1); assert!(cache.contains("BodyA")); let (l1_second, ta_second) = cache.get_or_derive("BodyA", &hm, None); assert_eq!( cache.len(), 1, "a hit must not insert a second entry for the same body" ); assert_eq!( ta_first.ocean_mask, ta_second.ocean_mask, "same heightmap → identical re-derived analysis (D-227)" ); assert_eq!(ta_first.slope_deg, ta_second.slope_deg); assert_eq!(ta_first.elev_pct, ta_second.elev_pct); // T-1170 Ruling 4b: Layer1Output (river_network in particular) is // ALSO retained and identically re-derived, not just TerrainAnalysis. assert_eq!( l1_first.river_network.river_cells, l1_second.river_network.river_cells, "Layer1Output.river_network must also be cached/reused, not just TerrainAnalysis" ); } /// Different bodies get independent entries, and a capacity-2 cache /// evicts the LEAST-RECENTLY-USED entry — not insertion order — when a /// third body is derived. Mirrors `BodyWorldStateCache`'s own /// `update_last_accessed_on_get` precedent: touching "BodyA" again before /// the third insert must save it from eviction. #[test] fn terrain_analysis_cache_evicts_lru_not_fifo() { let mut cache = TerrainAnalysisCache::new(2); let hm = window_test_hm(); cache.get_or_derive("BodyA", &hm, None); cache.get_or_derive("BodyB", &hm, None); assert_eq!(cache.len(), 2); // Touch BodyA again — it is now the MOST recently used, so BodyB // (untouched since its own insert) is the true LRU victim. cache.get_or_derive("BodyA", &hm, None); // Insert a third body — capacity 2 forces an eviction. cache.get_or_derive("BodyC", &hm, None); assert_eq!(cache.len(), 2); assert!( cache.contains("BodyA"), "recently re-touched BodyA must survive eviction" ); assert!( !cache.contains("BodyB"), "BodyB (true LRU — untouched since its own insert) must be evicted" ); assert!(cache.contains("BodyC")); } /// End-to-end: two `DeriveWindow` work items for the SAME body, submitted /// through the real `GenerationQueue` (not the bare `TerrainAnalysisCache` /// unit above), share ONE `TerrainAnalysisCache` entry — the fix for the /// PR #187 review C1 finding (the original landing called `run_layer1` /// unconditionally on every `DeriveWindow`, so panning around one body /// paid the ~45 ms re-derive on every window instead of just the first). /// Both windows must still complete correctly (content assertions, not /// timing — a wall-clock assertion would be flaky); the cache-population /// count is the load-bearing proof of reuse. #[test] fn two_windows_on_same_body_share_one_terrain_analysis_entry() { let q = GenerationQueue::with_threads(2); let conn_a = ConnectionId(1); let conn_b = ConnectionId(2); // Two DIFFERENT connections so submit_window's coalescing (which // supersedes same-connection/same-body pending items) doesn't collapse // these into one work item — the point here is two DISTINCT completed // derives sharing the cache, not coalescing (already covered above). q.submit_window( derive_window("SharedBody", conn_a, (0, 0)), GenPriority::Immediate, ); q.submit_window( derive_window("SharedBody", conn_b, (10, 10)), GenPriority::Immediate, ); std::thread::sleep(Duration::from_millis(200)); let completions = q.drain_completions(); let windows: Vec<_> = completions .iter() .filter_map(|c| { if let GenCompletion::WindowDerived { body_id, layer } = c { if body_id == "SharedBody" { return Some(layer); } } None }) .collect(); assert_eq!(windows.len(), 2, "both windows must complete"); let centers: std::collections::BTreeSet<_> = windows.iter().map(|l| l.center).collect(); assert_eq!( centers, std::collections::BTreeSet::from([(0, 0), (10, 10)]), "both distinct windows survived, not coalesced" ); // The queue's own TerrainAnalysisCache must hold exactly ONE entry // for "SharedBody" — both derives shared it rather than each // re-deriving independently. let cache = q.terrain_cache.lock().unwrap(); assert_eq!( cache.len(), 1, "two DeriveWindow items for the same body must share one TerrainAnalysis entry" ); assert!(cache.contains("SharedBody")); } }