From ddc448346213f1302a96ac562c33a4b2962e8b02 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Wed, 22 Jul 2026 20:32:09 +0200 Subject: [PATCH] test(simulation): connection-zero district_window delivery regression (bridge_tcp) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dudley's audit of the suspected ConnectionId(0) starvation: exhaustive trace of the delivery path (bridge/mod.rs insert/lookup/send paths, atlas plugin serve/drain, gen_queue coalescing) found NO sentinel or default-value collision — ConnectionId doesn't even derive Default, and every id path is a plain monotonic allocation with linear lookup. The real gap was coverage: every existing window test used ConnectionId(1). New connection_zero_window_delivery module replicates main.rs's actual bootstrap (blocking first-accept -> BridgeResource -> accept loop) and pins: the first-ever connection receives a district window; the exact six-tile entry shape delivers all six to connection 0; connection 1 keeps working. 5/5 repeated runs, single- and parallel-threaded. The live 'starvation' itself was client-side (pending responses never re-requested on the tile path — separate fix in flight); these tests stay as the server half's permanent pin. --- server/tests/bridge_tcp.rs | 497 +++++++++++++++++++++++++++++++++++++ 1 file changed, 497 insertions(+) diff --git a/server/tests/bridge_tcp.rs b/server/tests/bridge_tcp.rs index 957802a26..6e2dbb7a9 100644 --- a/server/tests/bridge_tcp.rs +++ b/server/tests/bridge_tcp.rs @@ -1396,6 +1396,503 @@ fn browse_index_empty_table_is_ready_with_empty_list_over_tcp() { let _ = std::fs::remove_file(&db_path); } +// ----------------------------------------------------------------------- +// ConnectionId(0) district_window delivery regression (live-verified bug, +// coordinator repro 2026-07-22): a fresh server's FIRST connection (whatever +// role) never received a district_window response over the wire, while a +// second connection to the same (or a fresh) server was served correctly. +// Every existing window test in layer_proxy.rs's unit-test module calls +// `test_conn_id() -> ConnectionId(1)` — NEVER ConnectionId(0) — so this class +// of bug had zero unit-test coverage. This test drives the REAL +// id-assignment path (`BridgeResource::default()` + `insert_reader`, the +// exact path `main.rs`'s first-connection handling and +// `accept_new_connections`'s reader promotion both use) rather than a +// hand-picked id, over the real TCP wire, through the real +// receive->serve->drain->send tick-phase pipeline. +// ----------------------------------------------------------------------- + +mod connection_zero_window_delivery { + use super::*; + use bevy_app::App; + use rusqlite::Connection; + use settled_reach_server::atlas::body_params_reader::{ + BodyParamsReader, BodyParamsReaderResource, + }; + use settled_reach_server::atlas::cascade::CascadeLayer; + use settled_reach_server::atlas::layer_proxy::{ + AtlasLayerRequest, AtlasLayerStatus, WindowGranularity, + }; + use settled_reach_server::atlas::source_resolver::{ + BodySourceResolver, BodySourceResolverResource, + }; + use settled_reach_server::atlas::GenerationPlugin; + use settled_reach_server::bridge::{ + BridgePlugin, ConnectionId, ConnectionListener, HandshakeState, PendingConnections, + }; + use settled_reach_server::simulation::SimulationPlugin; + use std::path::{Path, PathBuf}; + use std::sync::atomic::{AtomicU32, Ordering}; + + static SEQ: AtomicU32 = AtomicU32::new(0); + + /// Same fixture shape as `layer_proxy.rs`'s own `write_tiny_heightmap` + /// (not exported for cross-crate reuse — duplicated intentionally rather + /// than widening that module's visibility for one integration test). + fn write_tiny_heightmap(path: &Path) { + use std::io::BufWriter; + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + let file = std::fs::File::create(path).unwrap(); + 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().unwrap(); + let data: Vec = (0..32u32 * 16) + .flat_map(|i| (((i * 600) % 65536) as u16).to_be_bytes()) + .collect(); + w.write_image_data(&data).unwrap(); + } + + /// Same fixture shape as `layer_proxy.rs`'s `resolver_and_params_reader` + /// (duplicated for the same cross-crate-visibility reason as above). + fn resolver_and_params_reader( + body_id: &str, + ) -> (BodySourceResolver, BodyParamsReader, PathBuf) { + const REL: &str = "wiki/star-systems/GJ-1/bodies/GJ1c/heightmap.png"; + let n = SEQ.fetch_add(1, Ordering::Relaxed); + let db = std::env::temp_dir().join(format!("sr_connzero_{}_{n}.db", std::process::id())); + let _ = std::fs::remove_file(&db); + let conn = Connection::open(&db).unwrap(); + conn.execute_batch( + "CREATE TABLE star_systems ( + system_id TEXT PRIMARY KEY, + spectral_class TEXT, + star_type TEXT + ); + CREATE TABLE bodies ( + body_id TEXT PRIMARY KEY, + system_id TEXT, + terrain_reference TEXT, + hydrosphere TEXT, + atmosphere TEXT, + planet_class TEXT, + body_radius_km REAL, + orbital_period_days REAL, + axial_tilt_deg REAL + );", + ) + .unwrap(); + conn.execute( + "INSERT INTO star_systems (system_id, spectral_class, star_type) VALUES ('GJ-1', 'G', 'main_sequence')", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO bodies (body_id, system_id, terrain_reference, hydrosphere, atmosphere, planet_class, body_radius_km, orbital_period_days, axial_tilt_deg) + VALUES (?1, 'GJ-1', ?2, 'ocean', 'breathable', 'temperate', 6371.0, 365.25, 23.5)", + rusqlite::params![body_id, REL], + ) + .unwrap(); + drop(conn); + + let root = std::env::temp_dir().join(format!("sr_connzeroroot_{}_{n}", std::process::id())); + write_tiny_heightmap(&root.join(REL)); + + let resolver = BodySourceResolver::open(&db, vec![root.clone()]).unwrap(); + let params_reader = BodyParamsReader::open(&db).unwrap(); + (resolver, params_reader, root) + } + + /// A district-window `AtlasLayerRequest` at Region granularity — the + /// exact rung the coordinator's live repro used (six Region tile + /// requests on body entry). + fn region_window_request(body_id: &str, center: (i32, i32)) -> AtlasLayerRequest { + AtlasLayerRequest { + body_id: body_id.to_string(), + up_to: CascadeLayer::Topography, + window_center: Some(center), + window_n: 4, + window_granularity: 0, + window_granularity_v2: Some(WindowGranularity::Region), + window_min_wl_m: 0, + } + } + + /// Build a real `App`, replicating `main.rs`'s ACTUAL bootstrap sequence + /// for the first connection — not an approximation. `BridgePlugin` never + /// `init_resource`s `BridgeResource` itself (confirmed: no + /// `init_resource`/`insert_resource::` anywhere in + /// `bridge/mod.rs`'s `Plugin::build`) — `main.rs` constructs it AFTER a + /// single BLOCKING `TcpListener::accept()` + `TcpBridge::accept_on()` + + /// handshake/startup exchange, pre-populated with that first connection + /// already installed via `insert_player`/`insert_reader`, and only + /// THEN starts the tick loop with a SEPARATE non-blocking listener clone + /// for `accept_new_connections` to handle connections 2+. This function + /// reproduces exactly that two-listener-handle, blocking-then-async + /// split (`main.rs` lines ~104-145, ~368-377) rather than the + /// all-non-blocking shortcut the first draft of this test used (which + /// masked the real bootstrap entirely and hit "BridgeResource does not + /// exist" — the WRONG failure, not the bug under investigation). + /// + /// `role` selects Reader (the standalone Atlas companion's real role, + /// matching the coordinator's live repro) or Player. + fn accept_first_connection_and_build_app( + resolver: BodySourceResolver, + params_reader: settled_reach_server::atlas::body_params_reader::BodyParamsReader, + role: ConnectionRole, + ) -> (App, std::net::SocketAddr, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); + let addr = listener.local_addr().expect("failed to get local address"); + // try_clone BEFORE any accept, mirroring main.rs exactly: one handle + // for the blocking first accept, a second (later set non-blocking) + // for the async per-tick accept-loop — the same underlying socket. + let listener_for_loop = listener.try_clone().expect("failed to clone listener"); + + let client_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect"); + client_handshake(stream, role) + }); + + // Blocking accept — genuinely blocks this thread until the client + // above connects, exactly like main.rs's listener.accept() call. + let bridge = TcpBridge::accept_on(listener).expect("failed to accept first connection"); + // main.rs's real sequence (lines ~145-161): send the protocol + // handshake, THEN read the client's StartupMessage — BEFORE the + // client thread's client_handshake() (which reads the handshake + // first) can complete. Omitting these two calls was the first + // draft's bug: both sides deadlocked waiting on each other with + // nothing ever written first. + bridge.send_handshake().expect("failed to send handshake"); + let _startup = bridge + .receive_startup() + .expect("failed to receive startup message"); + let client_stream = client_handle.join().expect("client thread panicked"); + + let mut bridge_resource = BridgeResource::default(); + let first_id = match role { + ConnectionRole::Player => bridge_resource.insert_player(bridge), + ConnectionRole::Reader => bridge_resource.insert_reader(bridge), + }; + assert_eq!( + first_id, + ConnectionId(0), + "sanity: the first-ever insert on a fresh BridgeResource must be id 0" + ); + + // Full main.rs plugin composition (not a hand-picked subset) — a + // missing resource from any of these panicked the first draft + // (KnowledgePlugin/NpcPlugin/StorytellerPlugin/SettingsPlugin/ + // BookmarkPlugin's systems are NOT gated behind Option> the + // way GenerationPlugin's atlas-reader resources are), so matching + // main.rs exactly is the correct fix, not narrowing the plugin set. + let mut app = App::new(); + app.add_plugins(SimulationPlugin { seed: 42 }); + app.add_plugins(BridgePlugin); + app.add_plugins(settled_reach_server::knowledge::KnowledgePlugin); + app.add_plugins(settled_reach_server::npc::NpcPlugin); + 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(GenerationPlugin); + // main.rs unconditionally calls setup_proof_room()/setup_gauntlet() + // AFTER building every plugin above — even for a Reader-only spawn + // (D-254 §1's own comment on that call site: "this call is NOT + // gated on the first connection's role ... A Reader-only spawned + // server therefore has an inert, unpiloted PlayerCharacter entity"). + // setup_proof_room isn't `pub` (private to the main.rs binary crate) + // so it can't be called directly from an integration test; inserting + // WalkabilityMap alone (the specific resource several PreInput/ + // Movement-phase systems require unconditionally, confirmed by this + // test's first draft panicking with it absent) is the minimal + // equivalent for a Reader-only session that spawns no character — + // matching game_loop.rs's own established minimal pattern. + app.insert_resource( + settled_reach_server::simulation::movement::WalkabilityMap::new(32, 32, 1), + ); + app.insert_resource(bridge_resource); + app.insert_resource(HandshakeState::Complete); + // Non-blocking for the tick loop's accept_new_connections, exactly + // as main.rs's listener_for_loop.set_nonblocking(true) does. + listener_for_loop + .set_nonblocking(true) + .expect("failed to set accept-loop listener non-blocking"); + app.insert_resource(ConnectionListener(Some(listener_for_loop))); + app.world_mut() + .resource_mut::() + .0 + .clear(); + app.insert_resource(BodySourceResolverResource(resolver)); + app.insert_resource(BodyParamsReaderResource(params_reader)); + (app, addr, client_stream) + } + + /// Drive `app.update()` (one full real tick — every registered system, + /// in the real `TickPhase` order) repeatedly, checking the accept-loop + /// condition after each — mirrors `drive_accept_loop_until`'s pattern but + /// against a full `App` rather than a hand-picked system. + fn drive_app_accept_loop_until(app: &mut App, condition: impl Fn(&App) -> bool) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !condition(app) { + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for accept-loop condition" + ); + app.update(); + thread::sleep(std::time::Duration::from_millis(1)); + } + } + + /// Drive `app.update()` ticks until a `Ready` `AtlasLayerResponse` + /// carrying a populated `district_window` arrives on `stream`, or a + /// wall-clock deadline expires — mirrors + /// `drive_browse_tick_until_response`'s established pattern (bounded + /// retry loop over a genuinely blocking client-side socket, not a fixed + /// sleep/tick count) but must also tolerate intermediate `Pending` + /// frames (the D-225 poll-and-recheck-cache contract: the FIRST response + /// for a cold window is `Ready` with `district_window: None`, not the + /// final answer) by reading and discarding them until the populated one + /// lands. + fn drive_app_until_window_response( + app: &mut App, + stream: &mut TcpStream, + body_id: &str, + center: (i32, i32), + ) -> settled_reach_server::atlas::layer_proxy::AtlasLayerResponse { + stream + .set_read_timeout(Some(std::time::Duration::from_millis(50))) + .expect("failed to set read timeout"); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + loop { + app.update(); + match read_framed(stream) { + Ok(Some(payload)) => { + let resp: settled_reach_server::atlas::layer_proxy::AtlasLayerResponse = + rmp_serde::from_slice(&payload) + .expect("failed to decode AtlasLayerResponse"); + if resp.district_window.is_some() { + return resp; + } + // Pending (cold cache) or Ready-with-None (derive still + // in flight) — re-request and keep polling, exactly the + // client's real re-poll behavior (atlas_window_request.gd's + // on_response()/_schedule_retry(): the server never pushes + // a second response on its own — D-225's poll-and-recheck- + // cache contract requires the CLIENT to re-send). Omitting + // this re-send was this test's own first-draft bug: the + // request was consumed by serve_atlas_requests within the + // SAME app.update() it arrived in (receive -> serve -> + // send all run inside one PreInput/PostSnapshot pass), so + // every later tick correctly had nothing left to serve — + // not a production hang, a missing re-poll in the harness. + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for a populated district_window \ + (last status: {:?})", + resp.status + ); + send_region_window_request(stream, body_id, center); + } + Ok(None) => panic!("unexpected EOF reading atlas response"), + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut => + { + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for an atlas response frame at all" + ); + } + Err(e) => panic!("failed to read atlas response frame: {e}"), + } + } + } + + fn send_region_window_request(stream: &mut TcpStream, body_id: &str, center: (i32, i32)) { + let req = region_window_request(body_id, center); + let payload = rmp_serde::to_vec_named(&req).expect("failed to encode AtlasLayerRequest"); + write_framed(stream, &payload).expect("failed to write AtlasLayerRequest"); + } + + /// **THE regression test.** A fresh server's FIRST connection ever + /// accepted (Reader role, matching the standalone Atlas companion) must + /// receive its district_window response — proven via the REAL + /// `insert_reader`-assigned id, not `ConnectionId(1)` the way every + /// existing window unit test does. If this test alone is added without + /// re-requesting after the initial Pending (i.e. treats the first + /// response as final), it would have passed even on a broken server — + /// the assertion on `resp.status == Ready` AND `district_window.is_some()` + /// together, reached only via `drive_window_tick_until_response`'s + /// re-poll loop, is what actually exercises the async derive-then-cache + /// path a live client depends on. + #[test] + fn first_connection_ever_accepted_receives_district_window() { + let (resolver, params_reader, _root) = resolver_and_params_reader("GJ1c"); + let (mut app, _addr, mut stream) = + accept_first_connection_and_build_app(resolver, params_reader, ConnectionRole::Reader); + + // Confirm this really is the id-zero path before asserting delivery + // — a false pass here (e.g. BridgeResource pre-seeded elsewhere) + // would silently defeat the whole point of the test. (Also asserted + // inside accept_first_connection_and_build_app; re-checked here at + // the point of use for a self-contained failure message.) + let conn_id = app + .world() + .resource::() + .reader_ids() + .first() + .copied() + .expect("reader must be installed"); + assert_eq!( + conn_id, + ConnectionId(0), + "this test only proves what it claims to prove if the reader under \ + test genuinely got the FIRST id a fresh BridgeResource ever assigns" + ); + + send_region_window_request(&mut stream, "GJ1c", (0, 0)); + let resp = drive_app_until_window_response(&mut app, &mut stream, "GJ1c", (0, 0)); + + assert_eq!(resp.status, AtlasLayerStatus::Ready); + assert_eq!(resp.body_id, "GJ1c"); + let window = resp + .district_window + .expect("district_window must be populated (checked above; re-asserted for clarity)"); + assert_eq!(window.granularity_v2, WindowGranularity::Region); + assert!( + !window.morphology.is_empty(), + "the delivered window must carry real derived cell data, not an empty payload" + ); + } + + /// The EXACT shape of the coordinator's live repro: six Region tile + /// requests fired on body entry (the standalone Atlas companion's real + /// window-tile-set fan-out, not a single request) on connection 0, all + /// six must be delivered — not just the first, in case a many-in-flight + /// scenario surfaces an ordering issue a single-request test can't see + /// (e.g. coalescing/supersede logic dropping later requests, or the + /// per-tick MAX_READER_INBOUND_FRAMES_PER_TICK cap interacting badly + /// with six queued frames). + #[test] + fn first_connection_receives_all_six_region_tiles() { + let (resolver, params_reader, _root) = resolver_and_params_reader("GJ1c"); + let (mut app, _addr, mut stream) = + accept_first_connection_and_build_app(resolver, params_reader, ConnectionRole::Reader); + + let conn_id = app + .world() + .resource::() + .reader_ids() + .first() + .copied() + .expect("reader must be installed"); + assert_eq!(conn_id, ConnectionId(0)); + + // Six distinct region-tile centers, matching the design doc's tiled + // orbital-view worked example shape (a 3x2 or similar tile grid + // around the entry point) — distinct centers so each is a genuinely + // separate cache entry, not accidental coalescing onto one. + let centers: [(i32, i32); 6] = [ + (0, 0), + (12739, -3200), + (12739, 3200), + (0, -3200), + (0, 3200), + (6400, -3200), + ]; + + for ¢er in ¢ers { + send_region_window_request(&mut stream, "GJ1c", center); + } + + let mut received: std::collections::BTreeSet<(i32, i32)> = + std::collections::BTreeSet::new(); + stream + .set_read_timeout(Some(std::time::Duration::from_millis(50))) + .expect("failed to set read timeout"); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15); + while received.len() < centers.len() { + assert!( + std::time::Instant::now() < deadline, + "timed out with only {}/{} tiles delivered: {:?}", + received.len(), + centers.len(), + received + ); + app.update(); + match read_framed(&mut stream) { + Ok(Some(payload)) => { + let resp: settled_reach_server::atlas::layer_proxy::AtlasLayerResponse = + rmp_serde::from_slice(&payload) + .expect("failed to decode AtlasLayerResponse"); + if let Some(window) = resp.district_window { + received.insert(window.center); + } else { + // Pending/Ready-with-None for one of the six — re-send + // ALL not-yet-received centers (mirrors the real + // client's per-tile independent re-poll). + for ¢er in centers.iter().filter(|c| !received.contains(c)) { + send_region_window_request(&mut stream, "GJ1c", center); + } + } + } + Ok(None) => panic!("unexpected EOF reading atlas response"), + Err(e) + if e.kind() == std::io::ErrorKind::WouldBlock + || e.kind() == std::io::ErrorKind::TimedOut => {} + Err(e) => panic!("failed to read atlas response frame: {e}"), + } + } + + assert_eq!( + received.len(), + centers.len(), + "all six region tiles must be delivered to connection 0" + ); + } + + /// The delivery-order counterpart: a SECOND connection's request must + /// ALSO be served correctly (this already passed in the coordinator's + /// live repro — included here as a same-file regression guard so a + /// future fix to the id-0 case can't accidentally break id-1 while + /// fixing id-0, and so this file documents the FULL observed shape of + /// the bug, not just half of it). + #[test] + fn second_connection_also_receives_district_window() { + let (resolver, params_reader, _root) = resolver_and_params_reader("GJ1c"); + // First connection occupies ConnectionId(0) via the SAME real + // blocking-accept bootstrap main.rs uses (see + // accept_first_connection_and_build_app's doc) but sends no + // requests — isolates "is it specifically the SECOND id that works" + // from "does a second connection existing change anything for the + // first". + let (mut app, addr, _first_stream) = + accept_first_connection_and_build_app(resolver, params_reader, ConnectionRole::Reader); + + let second_handle = thread::spawn(move || { + let stream = TcpStream::connect(addr).expect("failed to connect"); + client_handshake(stream, ConnectionRole::Reader) + }); + drive_app_accept_loop_until(&mut app, |a| { + a.world().resource::().reader_count() == 2 + }); + let mut second_stream = second_handle.join().expect("client thread panicked"); + + let second_id = app + .world() + .resource::() + .reader_ids() + .get(1) + .copied() + .expect("second reader must be installed"); + assert_eq!(second_id, ConnectionId(1)); + + send_region_window_request(&mut second_stream, "GJ1c", (0, 0)); + let resp = drive_app_until_window_response(&mut app, &mut second_stream, "GJ1c", (0, 0)); + + assert_eq!(resp.status, AtlasLayerStatus::Ready); + assert!(resp.district_window.is_some()); + } +} + /// Minimal `ObserverSnapshot` for the tests above — same shape as /// `snapshot_roundtrip_over_tcp`'s at the top of this file, parameterized /// only by `tick` (the one field these tests assert on).