feat(engine): T-1131 browse data proxy — six entity kinds, index+detail, one wire envelope (D-254 SS4)
browse_reader.rs: BrowseReader on the CityContextReader::open() pattern
— six index + six detail reads against systems.db (bodies filterable by
containing system; system/corporation/commodity details fold their join
partners). browse_proxy.rs: BrowseRequest{browse, kind, query} /
BrowseResponse{kind, status, index, detail} wire types + dispatcher;
BrowseIndexRow{id, primary, secondary} generic across kinds;
BrowseDetail a per-kind enum of field-exhaustive structs.
Demux: Inbound::BrowseRequest is the FIFTH map shape — deliberately the
last; the doc's four-shape ceiling is re-pinned at five with rationale
(six kinds x two forms folded into ONE envelope whose internal enums
pick sub-behavior, the AtlasLayerRequest.up_to precedent, instead of
twelve top-level shapes) and a hard rule that a sixth shape must
migrate to the D-225 tagged-envelope framing. Served for BOTH roles,
connection-tagged 1:1 in-order like atlas/starmap/citynames.
serve_browse_requests pub so integration tests drive the true
end-to-end pipeline. Wire-only per D-254 (T-949 precedent); v1
exclusions (cascade geometry, event logs) respected.
30 unit tests + 7 bridge_tcp integration tests (six-kind round-trip
over real TCP, reader-can-browse, no crossed responses between two
readers, unknown-id NotFound, empty-table Ready); 2 pre-existing tests
updated for the new receive_bridge_inputs parameter.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -9,6 +9,8 @@ pub mod believability;
|
|||||||
pub mod block_irregularity;
|
pub mod block_irregularity;
|
||||||
pub mod body_params_reader;
|
pub mod body_params_reader;
|
||||||
pub mod body_world_state;
|
pub mod body_world_state;
|
||||||
|
pub mod browse_proxy;
|
||||||
|
pub mod browse_reader;
|
||||||
pub mod cascade;
|
pub mod cascade;
|
||||||
pub mod chunk_context;
|
pub mod chunk_context;
|
||||||
pub mod city_context_reader;
|
pub mod city_context_reader;
|
||||||
|
|||||||
@@ -19,6 +19,8 @@ use crate::atlas::atlas_data_proxy::{
|
|||||||
use crate::atlas::attractor_matching::CityPlacement;
|
use crate::atlas::attractor_matching::CityPlacement;
|
||||||
use crate::atlas::body_params_reader::BodyParamsReaderResource;
|
use crate::atlas::body_params_reader::BodyParamsReaderResource;
|
||||||
use crate::atlas::body_world_state::{BodyWorldStateCache, CACHE_CAPACITY};
|
use crate::atlas::body_world_state::{BodyWorldStateCache, CACHE_CAPACITY};
|
||||||
|
use crate::atlas::browse_proxy::handle_browse_request;
|
||||||
|
use crate::atlas::browse_reader::BrowseReaderResource;
|
||||||
use crate::atlas::city_context_reader::{
|
use crate::atlas::city_context_reader::{
|
||||||
context_from_read_set, CityContextReaderResource, CityEconomicReadSet,
|
context_from_read_set, CityContextReaderResource, CityEconomicReadSet,
|
||||||
};
|
};
|
||||||
@@ -41,8 +43,8 @@ use crate::atlas::trait_swerve::{
|
|||||||
build_swerve_pools, compute_swerve_rates, SwerveDrivers, SwervePools,
|
build_swerve_pools, compute_swerve_rates, SwerveDrivers, SwervePools,
|
||||||
};
|
};
|
||||||
use crate::bridge::{
|
use crate::bridge::{
|
||||||
AtlasRequestBuffer, AtlasResponseBuffer, CityNamesRequestBuffer, CityNamesResponseBuffer,
|
AtlasRequestBuffer, AtlasResponseBuffer, BrowseRequestBuffer, BrowseResponseBuffer,
|
||||||
StarMapRequestBuffer, StarMapResponseBuffer,
|
CityNamesRequestBuffer, CityNamesResponseBuffer, StarMapRequestBuffer, StarMapResponseBuffer,
|
||||||
};
|
};
|
||||||
use crate::seed::{SeedChain, SeedDomain};
|
use crate::seed::{SeedChain, SeedDomain};
|
||||||
use crate::simulation::generator::{
|
use crate::simulation::generator::{
|
||||||
@@ -68,7 +70,8 @@ impl Plugin for GenerationPlugin {
|
|||||||
.add_systems(
|
.add_systems(
|
||||||
Update,
|
Update,
|
||||||
serve_city_names_requests.in_set(TickPhase::PreInput),
|
serve_city_names_requests.in_set(TickPhase::PreInput),
|
||||||
);
|
)
|
||||||
|
.add_systems(Update, serve_browse_requests.in_set(TickPhase::PreInput));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -170,6 +173,34 @@ fn serve_city_names_requests(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Drain inbound data-browser requests and serve each through the proxy
|
||||||
|
/// (D-254 §4, T-1131): one of the six v1 registry-tier entity kinds, dispatched
|
||||||
|
/// to `BrowseReader` by `(kind, query)`.
|
||||||
|
///
|
||||||
|
/// `pub` (unlike its atlas/star-map/city-names siblings, which stay private)
|
||||||
|
/// so `server/tests/bridge_tcp.rs`'s browse integration tests can drive the
|
||||||
|
/// TRUE full pipeline (demux -> receive_bridge_inputs -> BrowseRequestBuffer
|
||||||
|
/// -> serve_browse_requests -> BrowseResponseBuffer -> send_browse_responses)
|
||||||
|
/// end-to-end via `RunSystemOnce`, rather than bypassing this system the way
|
||||||
|
/// `reader_receives_tagged_star_map_response` bypasses `serve_star_map_requests`
|
||||||
|
/// (see that test's own doc comment) because it has no way to call it.
|
||||||
|
pub fn serve_browse_requests(
|
||||||
|
mut requests: ResMut<BrowseRequestBuffer>,
|
||||||
|
mut responses: ResMut<BrowseResponseBuffer>,
|
||||||
|
browse_reader: Option<Res<BrowseReaderResource>>,
|
||||||
|
) {
|
||||||
|
if requests.0.is_empty() {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let reader = browse_reader.as_ref().map(|r| &r.0);
|
||||||
|
let pending: Vec<_> = requests.0.drain(..).collect();
|
||||||
|
for (conn_id, req) in pending {
|
||||||
|
responses
|
||||||
|
.0
|
||||||
|
.push((conn_id, handle_browse_request(&req, reader)));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Drain finished background work each tick and apply it to the cache (D-206).
|
/// Drain finished background work each tick and apply it to the cache (D-206).
|
||||||
///
|
///
|
||||||
/// Runs in `PreInput` (off the Rayon workers, on the main thread): a cheap
|
/// Runs in `PreInput` (off the Rayon workers, on the main thread): a cheap
|
||||||
|
|||||||
@@ -4,6 +4,7 @@
|
|||||||
|
|
||||||
use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge};
|
use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge};
|
||||||
use crate::atlas::atlas_data_proxy::{CityNamesResponse, StarMapResponse};
|
use crate::atlas::atlas_data_proxy::{CityNamesResponse, StarMapResponse};
|
||||||
|
use crate::atlas::browse_proxy::BrowseResponse;
|
||||||
use crate::atlas::layer_proxy::AtlasLayerResponse;
|
use crate::atlas::layer_proxy::AtlasLayerResponse;
|
||||||
use crate::bridge::framing::{read_framed, write_framed};
|
use crate::bridge::framing::{read_framed, write_framed};
|
||||||
use std::fs;
|
use std::fs;
|
||||||
@@ -166,6 +167,16 @@ impl SimBridge for LocalBridge {
|
|||||||
write_framed(writer.get_mut(), &payload)?;
|
write_framed(writer.get_mut(), &payload)?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn send_browse_response(&self, resp: &BrowseResponse) -> Result<(), BridgeError> {
|
||||||
|
let payload = rmp_serde::to_vec_named(resp)?;
|
||||||
|
let mut writer = self
|
||||||
|
.writer
|
||||||
|
.lock()
|
||||||
|
.map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?;
|
||||||
|
write_framed(writer.get_mut(), &payload)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Drop for LocalBridge {
|
impl Drop for LocalBridge {
|
||||||
|
|||||||
+98
-11
@@ -9,6 +9,7 @@ use bevy_ecs::schedule::IntoScheduleConfigs;
|
|||||||
use crate::atlas::atlas_data_proxy::{
|
use crate::atlas::atlas_data_proxy::{
|
||||||
CityNamesRequest, CityNamesResponse, StarMapRequest, StarMapResponse,
|
CityNamesRequest, CityNamesResponse, StarMapRequest, StarMapResponse,
|
||||||
};
|
};
|
||||||
|
use crate::atlas::browse_proxy::{BrowseRequest, BrowseResponse};
|
||||||
use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse};
|
use crate::atlas::layer_proxy::{AtlasLayerRequest, AtlasLayerResponse};
|
||||||
use crate::bridge::tcp::TcpBridge;
|
use crate::bridge::tcp::TcpBridge;
|
||||||
|
|
||||||
@@ -63,9 +64,15 @@ pub enum BridgeError {
|
|||||||
/// frames that carry more than one shape's discriminators outright (PR #176
|
/// frames that carry more than one shape's discriminators outright (PR #176
|
||||||
/// review H1). `AtlasLayerRequest` itself is untouched byte-for-byte.
|
/// review H1). `AtlasLayerRequest` itself is untouched byte-for-byte.
|
||||||
///
|
///
|
||||||
/// **Ceiling (D-225 trajectory):** four shapes is the practical limit of this
|
/// **Ceiling (D-225 trajectory):** [`BrowseRequest`] (T-1131) is the FIFTH
|
||||||
/// hand-rolled sniffing. The next new inbound shape must migrate the channel
|
/// map shape and, per the ceiling this doc already called at four, the last
|
||||||
/// to the tagged-envelope framing D-225 deferred — do not add a fifth probe.
|
/// one this hand-rolled scheme should ever carry — it stays at five only
|
||||||
|
/// because six entity kinds x two forms were folded into ONE new shape
|
||||||
|
/// (`browse`'s own internal `kind`/`query` enums pick the sub-behavior,
|
||||||
|
/// exactly as `AtlasLayerRequest.up_to: CascadeLayer` already does) rather
|
||||||
|
/// than added as twelve more top-level shapes. The next genuinely NEW
|
||||||
|
/// inbound shape (a sixth) must migrate the channel to the tagged-envelope
|
||||||
|
/// framing D-225 deferred — do not add a sixth probe.
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub enum Inbound {
|
pub enum Inbound {
|
||||||
/// A batch of player inputs (the gameplay path).
|
/// A batch of player inputs (the gameplay path).
|
||||||
@@ -76,6 +83,9 @@ pub enum Inbound {
|
|||||||
StarMapRequest(StarMapRequest),
|
StarMapRequest(StarMapRequest),
|
||||||
/// A per-body city-names request (T-949b).
|
/// A per-body city-names request (T-949b).
|
||||||
CityNamesRequest(CityNamesRequest),
|
CityNamesRequest(CityNamesRequest),
|
||||||
|
/// A data-browser request — one of the six D-254 §4 v1 entity kinds
|
||||||
|
/// (T-1131).
|
||||||
|
BrowseRequest(BrowseRequest),
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Key-presence probe for the defensive multi-shape check in
|
/// Key-presence probe for the defensive multi-shape check in
|
||||||
@@ -88,12 +98,14 @@ struct ShapeProbe {
|
|||||||
up_to: Option<serde::de::IgnoredAny>,
|
up_to: Option<serde::de::IgnoredAny>,
|
||||||
star_map: Option<serde::de::IgnoredAny>,
|
star_map: Option<serde::de::IgnoredAny>,
|
||||||
city_names: Option<serde::de::IgnoredAny>,
|
city_names: Option<serde::de::IgnoredAny>,
|
||||||
|
browse: Option<serde::de::IgnoredAny>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Demux a received frame payload into an [`Inbound`] (D-225, T-949). Tries,
|
/// Demux a received frame payload into an [`Inbound`] (D-225, T-949, T-1131).
|
||||||
/// in order: `Vec<PlayerInput>` (array) → `AtlasLayerRequest` (map,
|
/// Tries, in order: `Vec<PlayerInput>` (array) → `AtlasLayerRequest` (map,
|
||||||
/// `body_id`+`up_to`) → `StarMapRequest` (map, `star_map` discriminator) →
|
/// `body_id`+`up_to`) → `StarMapRequest` (map, `star_map` discriminator) →
|
||||||
/// `CityNamesRequest` (map, `city_names` discriminator + `body_id`).
|
/// `CityNamesRequest` (map, `city_names` discriminator + `body_id`) →
|
||||||
|
/// `BrowseRequest` (map, `browse` discriminator).
|
||||||
///
|
///
|
||||||
/// Mutual exclusivity is enforced, not assumed: no minimal well-formed
|
/// Mutual exclusivity is enforced, not assumed: no minimal well-formed
|
||||||
/// instance of one shape satisfies another (see the [`Inbound`] doc), and a
|
/// instance of one shape satisfies another (see the [`Inbound`] doc), and a
|
||||||
@@ -101,7 +113,7 @@ struct ShapeProbe {
|
|||||||
/// more than one shape — e.g. a buggy encoder emitting
|
/// more than one shape — e.g. a buggy encoder emitting
|
||||||
/// `{"star_map": true, "city_names": true, ...}` — instead of silently
|
/// `{"star_map": true, "city_names": true, ...}` — instead of silently
|
||||||
/// routing it to whichever shape is tried first (PR #176 review H1). A frame
|
/// routing it to whichever shape is tried first (PR #176 review H1). A frame
|
||||||
/// satisfying none of the four shapes is a genuinely malformed input frame.
|
/// satisfying none of the five shapes is a genuinely malformed input frame.
|
||||||
pub fn decode_inbound(payload: &[u8]) -> Result<Inbound, BridgeError> {
|
pub fn decode_inbound(payload: &[u8]) -> Result<Inbound, BridgeError> {
|
||||||
if let Ok(inputs) = rmp_serde::from_slice::<Vec<PlayerInput>>(payload) {
|
if let Ok(inputs) = rmp_serde::from_slice::<Vec<PlayerInput>>(payload) {
|
||||||
return Ok(Inbound::Inputs(inputs));
|
return Ok(Inbound::Inputs(inputs));
|
||||||
@@ -114,15 +126,20 @@ pub fn decode_inbound(payload: &[u8]) -> Result<Inbound, BridgeError> {
|
|||||||
let atlas = probe.body_id.is_some() && probe.up_to.is_some();
|
let atlas = probe.body_id.is_some() && probe.up_to.is_some();
|
||||||
let star_map = probe.star_map.is_some();
|
let star_map = probe.star_map.is_some();
|
||||||
let city_names = probe.city_names.is_some();
|
let city_names = probe.city_names.is_some();
|
||||||
let shapes = usize::from(atlas) + usize::from(star_map) + usize::from(city_names);
|
let browse = probe.browse.is_some();
|
||||||
|
let shapes = usize::from(atlas)
|
||||||
|
+ usize::from(star_map)
|
||||||
|
+ usize::from(city_names)
|
||||||
|
+ usize::from(browse);
|
||||||
if shapes > 1 {
|
if shapes > 1 {
|
||||||
let dump_len = payload.len().min(256);
|
let dump_len = payload.len().min(256);
|
||||||
tracing::error!(
|
tracing::error!(
|
||||||
"inbound frame matches {} request shapes at once (atlas={}, star_map={}, city_names={}) — rejecting ambiguous frame. Raw ({} of {} bytes): {:02x?}",
|
"inbound frame matches {} request shapes at once (atlas={}, star_map={}, city_names={}, browse={}) — rejecting ambiguous frame. Raw ({} of {} bytes): {:02x?}",
|
||||||
shapes,
|
shapes,
|
||||||
atlas,
|
atlas,
|
||||||
star_map,
|
star_map,
|
||||||
city_names,
|
city_names,
|
||||||
|
browse,
|
||||||
dump_len,
|
dump_len,
|
||||||
payload.len(),
|
payload.len(),
|
||||||
&payload[..dump_len]
|
&payload[..dump_len]
|
||||||
@@ -139,8 +156,11 @@ pub fn decode_inbound(payload: &[u8]) -> Result<Inbound, BridgeError> {
|
|||||||
if let Ok(req) = rmp_serde::from_slice::<StarMapRequest>(payload) {
|
if let Ok(req) = rmp_serde::from_slice::<StarMapRequest>(payload) {
|
||||||
return Ok(Inbound::StarMapRequest(req));
|
return Ok(Inbound::StarMapRequest(req));
|
||||||
}
|
}
|
||||||
match rmp_serde::from_slice::<CityNamesRequest>(payload) {
|
if let Ok(req) = rmp_serde::from_slice::<CityNamesRequest>(payload) {
|
||||||
Ok(req) => Ok(Inbound::CityNamesRequest(req)),
|
return Ok(Inbound::CityNamesRequest(req));
|
||||||
|
}
|
||||||
|
match rmp_serde::from_slice::<BrowseRequest>(payload) {
|
||||||
|
Ok(req) => Ok(Inbound::BrowseRequest(req)),
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
let dump_len = payload.len().min(256);
|
let dump_len = payload.len().min(256);
|
||||||
tracing::error!(
|
tracing::error!(
|
||||||
@@ -189,6 +209,9 @@ pub trait SimBridge: Send + Sync {
|
|||||||
|
|
||||||
/// Send a city-names response to the client (T-949b).
|
/// Send a city-names response to the client (T-949b).
|
||||||
fn send_city_names_response(&self, resp: &CityNamesResponse) -> Result<(), BridgeError>;
|
fn send_city_names_response(&self, resp: &CityNamesResponse) -> Result<(), BridgeError>;
|
||||||
|
|
||||||
|
/// Send a browse response to the client (T-1131).
|
||||||
|
fn send_browse_response(&self, resp: &BrowseResponse) -> Result<(), BridgeError>;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Identifies one connection for response-tagging and role-lookup purposes
|
/// Identifies one connection for response-tagging and role-lookup purposes
|
||||||
@@ -433,6 +456,27 @@ impl BridgeResource {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Send a browse response to exactly the connection that requested it
|
||||||
|
/// (T-1131 — same per-connection routing D-254 §2 established for
|
||||||
|
/// atlas/star-map/city-names).
|
||||||
|
pub fn send_browse_response_to(
|
||||||
|
&self,
|
||||||
|
id: ConnectionId,
|
||||||
|
resp: &BrowseResponse,
|
||||||
|
) -> Result<(), BridgeError> {
|
||||||
|
match self.connection(id) {
|
||||||
|
Some(c) => c.bridge.send_browse_response(resp),
|
||||||
|
None => {
|
||||||
|
tracing::debug!(
|
||||||
|
"browse response for {:?} dropped — connection {:?} no longer present",
|
||||||
|
resp.kind,
|
||||||
|
id
|
||||||
|
);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Tracks whether the protocol handshake has been sent (#555).
|
/// Tracks whether the protocol handshake has been sent (#555).
|
||||||
@@ -484,6 +528,7 @@ pub fn receive_bridge_inputs(
|
|||||||
mut atlas_requests: ResMut<AtlasRequestBuffer>,
|
mut atlas_requests: ResMut<AtlasRequestBuffer>,
|
||||||
mut star_map_requests: ResMut<StarMapRequestBuffer>,
|
mut star_map_requests: ResMut<StarMapRequestBuffer>,
|
||||||
mut city_names_requests: ResMut<CityNamesRequestBuffer>,
|
mut city_names_requests: ResMut<CityNamesRequestBuffer>,
|
||||||
|
mut browse_requests: ResMut<BrowseRequestBuffer>,
|
||||||
time: Option<Res<crate::simulation::time::SimulationTime>>,
|
time: Option<Res<crate::simulation::time::SimulationTime>>,
|
||||||
) {
|
) {
|
||||||
let Some(mut bridge) = bridge else { return };
|
let Some(mut bridge) = bridge else { return };
|
||||||
@@ -526,6 +571,9 @@ pub fn receive_bridge_inputs(
|
|||||||
Ok(Some(Inbound::CityNamesRequest(req))) => {
|
Ok(Some(Inbound::CityNamesRequest(req))) => {
|
||||||
city_names_requests.0.push((player_id, req));
|
city_names_requests.0.push((player_id, req));
|
||||||
}
|
}
|
||||||
|
Ok(Some(Inbound::BrowseRequest(req))) => {
|
||||||
|
browse_requests.0.push((player_id, req));
|
||||||
|
}
|
||||||
// No complete frame ready — the backlog is drained.
|
// No complete frame ready — the backlog is drained.
|
||||||
Ok(None) => break,
|
Ok(None) => break,
|
||||||
Err(BridgeError::Disconnected) => {
|
Err(BridgeError::Disconnected) => {
|
||||||
@@ -637,6 +685,9 @@ pub fn receive_bridge_inputs(
|
|||||||
Ok(Some(Inbound::CityNamesRequest(req))) => {
|
Ok(Some(Inbound::CityNamesRequest(req))) => {
|
||||||
city_names_requests.0.push((reader_id, req));
|
city_names_requests.0.push((reader_id, req));
|
||||||
}
|
}
|
||||||
|
Ok(Some(Inbound::BrowseRequest(req))) => {
|
||||||
|
browse_requests.0.push((reader_id, req));
|
||||||
|
}
|
||||||
Ok(None) => break,
|
Ok(None) => break,
|
||||||
Err(BridgeError::Disconnected) => {
|
Err(BridgeError::Disconnected) => {
|
||||||
tracing::info!("Reader connection {:?} disconnected", reader_id);
|
tracing::info!("Reader connection {:?} disconnected", reader_id);
|
||||||
@@ -851,6 +902,36 @@ pub fn send_city_names_responses(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Inbound data-browser requests routed off the bridge (T-1131), drained by
|
||||||
|
/// the proxy serve system in `PreInput`. Connection-tagged (D-254 §2).
|
||||||
|
#[derive(Resource, Default)]
|
||||||
|
pub struct BrowseRequestBuffer(pub Vec<(ConnectionId, BrowseRequest)>);
|
||||||
|
|
||||||
|
/// Outbound browse responses, filled by the proxy serve system and flushed
|
||||||
|
/// to the client in `PostSnapshot` (T-1131). Connection-tagged (D-254 §2).
|
||||||
|
#[derive(Resource, Default)]
|
||||||
|
pub struct BrowseResponseBuffer(pub Vec<(ConnectionId, BrowseResponse)>);
|
||||||
|
|
||||||
|
/// Flush buffered browse responses to their requesting connections (T-1131 —
|
||||||
|
/// same per-connection routing D-254 §2 established for
|
||||||
|
/// atlas/star-map/city-names). A failed send is logged but not fatal.
|
||||||
|
pub fn send_browse_responses(
|
||||||
|
bridge: Option<Res<BridgeResource>>,
|
||||||
|
mut buffer: ResMut<BrowseResponseBuffer>,
|
||||||
|
) {
|
||||||
|
let Some(bridge) = bridge else { return };
|
||||||
|
for (id, resp) in buffer.0.drain(..) {
|
||||||
|
if let Err(e) = bridge.send_browse_response_to(id, &resp) {
|
||||||
|
tracing::warn!(
|
||||||
|
"failed to send browse response for {:?} to {:?}: {}",
|
||||||
|
resp.kind,
|
||||||
|
id,
|
||||||
|
e
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Holds the server's TCP listener for accepting connections AFTER the
|
/// Holds the server's TCP listener for accepting connections AFTER the
|
||||||
/// first Player connection (D-254 §2, T-1130).
|
/// first Player connection (D-254 §2, T-1130).
|
||||||
///
|
///
|
||||||
@@ -1008,6 +1089,8 @@ impl Plugin for BridgePlugin {
|
|||||||
.init_resource::<StarMapResponseBuffer>()
|
.init_resource::<StarMapResponseBuffer>()
|
||||||
.init_resource::<CityNamesRequestBuffer>()
|
.init_resource::<CityNamesRequestBuffer>()
|
||||||
.init_resource::<CityNamesResponseBuffer>()
|
.init_resource::<CityNamesResponseBuffer>()
|
||||||
|
.init_resource::<BrowseRequestBuffer>()
|
||||||
|
.init_resource::<BrowseResponseBuffer>()
|
||||||
.init_resource::<ConnectionListener>()
|
.init_resource::<ConnectionListener>()
|
||||||
.init_resource::<PendingConnections>()
|
.init_resource::<PendingConnections>()
|
||||||
// Multi-connection accept-loop (D-254 §2, T-1130) — must run
|
// Multi-connection accept-loop (D-254 §2, T-1130) — must run
|
||||||
@@ -1032,6 +1115,10 @@ impl Plugin for BridgePlugin {
|
|||||||
Update,
|
Update,
|
||||||
send_city_names_responses.in_set(TickPhase::PostSnapshot),
|
send_city_names_responses.in_set(TickPhase::PostSnapshot),
|
||||||
)
|
)
|
||||||
|
.add_systems(
|
||||||
|
Update,
|
||||||
|
send_browse_responses.in_set(TickPhase::PostSnapshot),
|
||||||
|
)
|
||||||
// Debug commands — Snapshot phase
|
// Debug commands — Snapshot phase
|
||||||
.add_systems(
|
.add_systems(
|
||||||
Update,
|
Update,
|
||||||
|
|||||||
@@ -5,6 +5,7 @@
|
|||||||
|
|
||||||
use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge};
|
use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge};
|
||||||
use crate::atlas::atlas_data_proxy::{CityNamesResponse, StarMapResponse};
|
use crate::atlas::atlas_data_proxy::{CityNamesResponse, StarMapResponse};
|
||||||
|
use crate::atlas::browse_proxy::BrowseResponse;
|
||||||
use crate::atlas::layer_proxy::AtlasLayerResponse;
|
use crate::atlas::layer_proxy::AtlasLayerResponse;
|
||||||
use crate::bridge::framing::{read_framed, write_framed, FrameAccumulator};
|
use crate::bridge::framing::{read_framed, write_framed, FrameAccumulator};
|
||||||
use std::io::BufWriter;
|
use std::io::BufWriter;
|
||||||
@@ -313,6 +314,20 @@ impl SimBridge for TcpBridge {
|
|||||||
result?;
|
result?;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn send_browse_response(&self, resp: &BrowseResponse) -> Result<(), BridgeError> {
|
||||||
|
let payload = rmp_serde::to_vec_named(resp)?;
|
||||||
|
let mut writer = self
|
||||||
|
.writer
|
||||||
|
.lock()
|
||||||
|
.map_err(|e| BridgeError::MutexPoisoned(format!("writer: {}", e)))?;
|
||||||
|
let stream = writer.get_mut();
|
||||||
|
stream.set_nonblocking(false).map_err(BridgeError::Io)?;
|
||||||
|
let result = write_framed(stream, &payload);
|
||||||
|
stream.set_nonblocking(true).map_err(BridgeError::Io)?;
|
||||||
|
result?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A connection that has been TCP-accepted but has not yet completed the
|
/// A connection that has been TCP-accepted but has not yet completed the
|
||||||
|
|||||||
@@ -253,6 +253,25 @@ fn main() {
|
|||||||
),
|
),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Data browser reader (D-254 §4, T-1131): read-only access to the six v1
|
||||||
|
// registry-tier entity kinds (star systems, bodies, stations,
|
||||||
|
// corporations, commodities, trait templates) for the companion app's
|
||||||
|
// browse UI. Absent -> BrowseRequests are answered with an Error status
|
||||||
|
// per request rather than a hard failure (matches every other reader's
|
||||||
|
// "unavailable, log + degrade" convention below).
|
||||||
|
match settled_reach_server::atlas::browse_reader::BrowseReader::open(&systems_db_path) {
|
||||||
|
Ok(reader) => {
|
||||||
|
tracing::info!("Browse reader opened: {:?}", systems_db_path);
|
||||||
|
app.insert_resource(
|
||||||
|
settled_reach_server::atlas::browse_reader::BrowseReaderResource(reader),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Err(e) => tracing::warn!(
|
||||||
|
"Browse reader unavailable ({}). Browse requests will error.",
|
||||||
|
e
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
// Body physical params reader for DistrictProfile carrier layer (T-1032, D-239 §1, D-240):
|
// Body physical params reader for DistrictProfile carrier layer (T-1032, D-239 §1, D-240):
|
||||||
// reads hydrosphere / atmosphere / planet_class on a cache miss so the Rayon
|
// reads hydrosphere / atmosphere / planet_class on a cache miss so the Rayon
|
||||||
// cascade work item stays DB-free (D-225 pattern).
|
// cascade work item stays DB-free (D-225 pattern).
|
||||||
|
|||||||
+499
-6
@@ -1,9 +1,12 @@
|
|||||||
//! Integration tests for TcpBridge over TCP localhost (D-030 Layer 2: IPC roundtrip).
|
//! Integration tests for TcpBridge over TCP localhost (D-030 Layer 2: IPC roundtrip).
|
||||||
|
|
||||||
|
use settled_reach_server::atlas::browse_proxy::{
|
||||||
|
BrowseEntityKind, BrowseQuery, BrowseRequest, BrowseResponse, BrowseStatus,
|
||||||
|
};
|
||||||
use settled_reach_server::bridge::framing::{read_framed, write_framed};
|
use settled_reach_server::bridge::framing::{read_framed, write_framed};
|
||||||
use settled_reach_server::bridge::tcp::TcpBridge;
|
use settled_reach_server::bridge::tcp::TcpBridge;
|
||||||
use settled_reach_server::bridge::types::*;
|
use settled_reach_server::bridge::types::*;
|
||||||
use settled_reach_server::bridge::{Inbound, SimBridge};
|
use settled_reach_server::bridge::{BridgeResource, Inbound, SimBridge};
|
||||||
use settled_reach_server::simulation::time::{DayPhase, TickRate};
|
use settled_reach_server::simulation::time::{DayPhase, TickRate};
|
||||||
use std::net::{TcpListener, TcpStream};
|
use std::net::{TcpListener, TcpStream};
|
||||||
use std::thread;
|
use std::thread;
|
||||||
@@ -338,8 +341,8 @@ fn single_tick_drains_all_ready_inbound_frames() {
|
|||||||
use settled_reach_server::atlas::cascade::CascadeLayer;
|
use settled_reach_server::atlas::cascade::CascadeLayer;
|
||||||
use settled_reach_server::atlas::layer_proxy::AtlasLayerRequest;
|
use settled_reach_server::atlas::layer_proxy::AtlasLayerRequest;
|
||||||
use settled_reach_server::bridge::{
|
use settled_reach_server::bridge::{
|
||||||
receive_bridge_inputs, AtlasRequestBuffer, BridgeResource, CityNamesRequestBuffer,
|
receive_bridge_inputs, AtlasRequestBuffer, BridgeResource, BrowseRequestBuffer,
|
||||||
HandshakeState, ServerRunning, StarMapRequestBuffer,
|
CityNamesRequestBuffer, HandshakeState, ServerRunning, StarMapRequestBuffer,
|
||||||
};
|
};
|
||||||
use settled_reach_server::simulation::input::InputQueue;
|
use settled_reach_server::simulation::input::InputQueue;
|
||||||
|
|
||||||
@@ -395,6 +398,7 @@ fn single_tick_drains_all_ready_inbound_frames() {
|
|||||||
world.init_resource::<AtlasRequestBuffer>();
|
world.init_resource::<AtlasRequestBuffer>();
|
||||||
world.init_resource::<StarMapRequestBuffer>();
|
world.init_resource::<StarMapRequestBuffer>();
|
||||||
world.init_resource::<CityNamesRequestBuffer>();
|
world.init_resource::<CityNamesRequestBuffer>();
|
||||||
|
world.init_resource::<BrowseRequestBuffer>();
|
||||||
|
|
||||||
world
|
world
|
||||||
.run_system_once(receive_bridge_inputs)
|
.run_system_once(receive_bridge_inputs)
|
||||||
@@ -489,9 +493,10 @@ fn drive_accept_loop_until(
|
|||||||
/// buffers), bound to a fresh OS-assigned port.
|
/// buffers), bound to a fresh OS-assigned port.
|
||||||
fn new_multi_connection_world() -> (bevy_ecs::world::World, std::net::SocketAddr) {
|
fn new_multi_connection_world() -> (bevy_ecs::world::World, std::net::SocketAddr) {
|
||||||
use settled_reach_server::bridge::{
|
use settled_reach_server::bridge::{
|
||||||
AtlasRequestBuffer, AtlasResponseBuffer, BridgeResource, CityNamesRequestBuffer,
|
AtlasRequestBuffer, AtlasResponseBuffer, BridgeResource, BrowseRequestBuffer,
|
||||||
CityNamesResponseBuffer, ConnectionListener, HandshakeState, PendingConnections,
|
BrowseResponseBuffer, CityNamesRequestBuffer, CityNamesResponseBuffer, ConnectionListener,
|
||||||
ServerRunning, SnapshotBuffer, StarMapRequestBuffer, StarMapResponseBuffer,
|
HandshakeState, PendingConnections, ServerRunning, SnapshotBuffer, StarMapRequestBuffer,
|
||||||
|
StarMapResponseBuffer,
|
||||||
};
|
};
|
||||||
use settled_reach_server::simulation::input::InputQueue;
|
use settled_reach_server::simulation::input::InputQueue;
|
||||||
|
|
||||||
@@ -515,6 +520,8 @@ fn new_multi_connection_world() -> (bevy_ecs::world::World, std::net::SocketAddr
|
|||||||
world.init_resource::<StarMapResponseBuffer>();
|
world.init_resource::<StarMapResponseBuffer>();
|
||||||
world.init_resource::<CityNamesRequestBuffer>();
|
world.init_resource::<CityNamesRequestBuffer>();
|
||||||
world.init_resource::<CityNamesResponseBuffer>();
|
world.init_resource::<CityNamesResponseBuffer>();
|
||||||
|
world.init_resource::<BrowseRequestBuffer>();
|
||||||
|
world.init_resource::<BrowseResponseBuffer>();
|
||||||
world.init_resource::<SnapshotBuffer>();
|
world.init_resource::<SnapshotBuffer>();
|
||||||
(world, addr)
|
(world, addr)
|
||||||
}
|
}
|
||||||
@@ -898,6 +905,492 @@ fn second_player_attempt_is_cleanly_rejected() {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────
|
||||||
|
// T-1131: data browser (D-254 §4) end-to-end over the real TCP wire.
|
||||||
|
//
|
||||||
|
// These tests exercise the FULL pipeline — demux (decode_inbound) ->
|
||||||
|
// receive_bridge_inputs (drain + connection-tag) -> BrowseRequestBuffer ->
|
||||||
|
// serve_browse_requests (the atlas-plugin proxy system) -> BrowseResponseBuffer
|
||||||
|
// -> send_browse_responses -> the socket — not just the buffer-push shortcut
|
||||||
|
// `reader_receives_tagged_star_map_response` uses above, since T-1131's own
|
||||||
|
// ticket text calls for the connection-tagging proof specifically over
|
||||||
|
// requests that actually round-trip through the demux.
|
||||||
|
// ─────────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// Build a fixture `systems.db`-shaped file with one row in each of the six
|
||||||
|
/// v1 browse tables, wired as a `BrowseReaderResource` into `world`. A
|
||||||
|
/// self-contained minimal fixture local to THIS file — `browse_reader`'s own
|
||||||
|
/// (larger, column-exhaustive) fixture lives behind `#[cfg(test)]` in the
|
||||||
|
/// library crate, which is only compiled for `cargo test --lib`, not for a
|
||||||
|
/// separate integration-test binary linking against the built library (a
|
||||||
|
/// cross-crate `#[cfg(test)]` visibility boundary — confirmed by attempting
|
||||||
|
/// the reuse first). This file only needs to prove the WIRE plumbing (demux
|
||||||
|
/// -> serve -> send), not the SQL correctness `browse_reader`'s own unit
|
||||||
|
/// tests already verify column-by-column, so a minimal one-row-per-table
|
||||||
|
/// fixture is the right scope here, not a duplicate of the exhaustive one.
|
||||||
|
fn wire_browse_reader(world: &mut bevy_ecs::world::World) {
|
||||||
|
use settled_reach_server::atlas::browse_reader::{BrowseReader, BrowseReaderResource};
|
||||||
|
|
||||||
|
let db_path = std::env::temp_dir().join(format!(
|
||||||
|
"sr_browse_tcp_fixture_{}_{}.db",
|
||||||
|
std::process::id(),
|
||||||
|
std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.unwrap()
|
||||||
|
.as_nanos()
|
||||||
|
));
|
||||||
|
let _ = std::fs::remove_file(&db_path);
|
||||||
|
{
|
||||||
|
let conn = rusqlite::Connection::open(&db_path).expect("create fixture db");
|
||||||
|
conn.execute_batch(
|
||||||
|
"CREATE TABLE star_systems (
|
||||||
|
system_id TEXT PRIMARY KEY, proper_name TEXT, system_name TEXT,
|
||||||
|
star_type TEXT, spectral_class TEXT, dist_ly REAL,
|
||||||
|
geographic_sector TEXT, geographic_band TEXT, political_zone TEXT,
|
||||||
|
habitable_planet_count INTEGER, inhabited_planet_count INTEGER,
|
||||||
|
asteroid_belt INTEGER, gas_giant INTEGER, habitability_profile TEXT,
|
||||||
|
earth_alignment TEXT, earth_proximity TEXT, earth_tension TEXT,
|
||||||
|
stability_index INTEGER, system_volatility TEXT, cultural_corridor TEXT,
|
||||||
|
currency_zone TEXT
|
||||||
|
);
|
||||||
|
CREATE TABLE system_economy (
|
||||||
|
system_id TEXT PRIMARY KEY, economic_tier INTEGER, population INTEGER,
|
||||||
|
economic_base_primary TEXT, economic_base_secondary TEXT
|
||||||
|
);
|
||||||
|
CREATE TABLE system_factions (
|
||||||
|
system_id TEXT PRIMARY KEY, governance_type TEXT, dominant_faction TEXT
|
||||||
|
);
|
||||||
|
CREATE TABLE system_culture (
|
||||||
|
system_id TEXT PRIMARY KEY, cultural_register TEXT,
|
||||||
|
atmospheric_tone TEXT, primary_archetype TEXT
|
||||||
|
);
|
||||||
|
CREATE TABLE bodies (
|
||||||
|
body_id TEXT PRIMARY KEY, system_id TEXT NOT NULL, parent_body_id TEXT,
|
||||||
|
body_type TEXT NOT NULL, orbit_index INTEGER, proper_name TEXT,
|
||||||
|
mass_class TEXT, atmosphere TEXT, surface_gravity REAL,
|
||||||
|
orbital_period_days REAL, rotation_period_hours REAL, planet_class TEXT,
|
||||||
|
hydrosphere TEXT, biosphere_class TEXT, inhabited INTEGER NOT NULL DEFAULT 0,
|
||||||
|
population INTEGER, economic_role TEXT, founding_age_years INTEGER,
|
||||||
|
settlement_pattern TEXT, cultural_corridor TEXT, industrial_corridor TEXT,
|
||||||
|
body_radius_km REAL, axial_tilt_deg REAL
|
||||||
|
);
|
||||||
|
CREATE TABLE stations (
|
||||||
|
station_id TEXT PRIMARY KEY, system_id TEXT NOT NULL, orbits_body_id TEXT,
|
||||||
|
station_type TEXT NOT NULL, proper_name TEXT, population INTEGER,
|
||||||
|
economic_role TEXT, governance_type TEXT, docking_class TEXT,
|
||||||
|
has_gate_infrastructure INTEGER DEFAULT 0, district_count INTEGER
|
||||||
|
);
|
||||||
|
CREATE TABLE corporations (
|
||||||
|
corp_id TEXT PRIMARY KEY, proper_name TEXT NOT NULL, corp_type TEXT NOT NULL,
|
||||||
|
scope TEXT, headquarters_system TEXT, headquarters_body TEXT,
|
||||||
|
specialization TEXT, parent_corp TEXT, notes TEXT, behavioral_archetype TEXT,
|
||||||
|
supply_chain_role TEXT, shadow_economy_access INTEGER DEFAULT 0,
|
||||||
|
corp_specialization TEXT, hq_placement TEXT
|
||||||
|
);
|
||||||
|
CREATE TABLE corp_presence (
|
||||||
|
corp_id TEXT NOT NULL, location_id TEXT NOT NULL, location_type TEXT NOT NULL,
|
||||||
|
primary_operation TEXT, PRIMARY KEY (corp_id, location_id)
|
||||||
|
);
|
||||||
|
CREATE TABLE corp_financial_state (corp_id TEXT PRIMARY KEY, health_metric REAL NOT NULL DEFAULT 1.0);
|
||||||
|
CREATE TABLE commodities (
|
||||||
|
commodity_id TEXT PRIMARY KEY, name TEXT NOT NULL, tier TEXT NOT NULL,
|
||||||
|
elasticity TEXT NOT NULL, base_price REAL NOT NULL, bulk_class TEXT,
|
||||||
|
unit TEXT, production_ubiquity TEXT, demand_model TEXT,
|
||||||
|
commission_certifiable INTEGER DEFAULT 0, compact_contested INTEGER DEFAULT 0,
|
||||||
|
shadow_viable INTEGER DEFAULT 0, panic_threshold_weeks INTEGER DEFAULT 0, description TEXT
|
||||||
|
);
|
||||||
|
CREATE TABLE production_chains (
|
||||||
|
chain_id TEXT PRIMARY KEY, output_commodity_id TEXT NOT NULL,
|
||||||
|
output_quantity REAL NOT NULL DEFAULT 1.0, location_bound INTEGER DEFAULT 0, description TEXT
|
||||||
|
);
|
||||||
|
CREATE TABLE chain_inputs (
|
||||||
|
chain_id TEXT NOT NULL, input_commodity_id TEXT NOT NULL, quantity REAL NOT NULL,
|
||||||
|
PRIMARY KEY (chain_id, input_commodity_id)
|
||||||
|
);
|
||||||
|
CREATE TABLE trait_templates (
|
||||||
|
tag TEXT PRIMARY KEY, label TEXT NOT NULL, cultural_description TEXT,
|
||||||
|
corridor_pool TEXT NOT NULL DEFAULT 'baseline', geographic_sector TEXT,
|
||||||
|
bulk_class_gate TEXT, production_ubiquity_gate TEXT,
|
||||||
|
min_prosperity_bps INTEGER NOT NULL DEFAULT 0, base_weight INTEGER NOT NULL DEFAULT 10000,
|
||||||
|
weight_mods TEXT, zone_affinity TEXT, allow_tags TEXT, block_tags TEXT,
|
||||||
|
era_scope TEXT, visual_bundle TEXT
|
||||||
|
);",
|
||||||
|
)
|
||||||
|
.expect("create fixture tables");
|
||||||
|
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO star_systems (system_id, proper_name, star_type) VALUES ('GJ-1', 'Aldren', 'M')",
|
||||||
|
[],
|
||||||
|
)
|
||||||
|
.expect("insert star system");
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO bodies (body_id, system_id, body_type, proper_name, inhabited)
|
||||||
|
VALUES ('GJ1c', 'GJ-1', 'planet', 'Aldren Prime', 1)",
|
||||||
|
[],
|
||||||
|
)
|
||||||
|
.expect("insert body");
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO stations (station_id, system_id, station_type, proper_name)
|
||||||
|
VALUES ('GJ1c-S1', 'GJ-1', 'commercial', 'Aldren Orbital')",
|
||||||
|
[],
|
||||||
|
)
|
||||||
|
.expect("insert station");
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO corporations (corp_id, proper_name, corp_type, headquarters_system)
|
||||||
|
VALUES ('gate-corporation', 'Gate Corporation', 'corporation', 'GJ-1')",
|
||||||
|
[],
|
||||||
|
)
|
||||||
|
.expect("insert corp");
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO commodities (commodity_id, name, tier, elasticity, base_price)
|
||||||
|
VALUES ('fusion_fuel', 'Fusion Fuel', 'intermediate', 'inelastic', 42.5)",
|
||||||
|
[],
|
||||||
|
)
|
||||||
|
.expect("insert commodity");
|
||||||
|
conn.execute(
|
||||||
|
"INSERT INTO trait_templates (tag, label) VALUES ('frontier_utilitarian', 'Frontier Utilitarian')",
|
||||||
|
[],
|
||||||
|
)
|
||||||
|
.expect("insert trait template");
|
||||||
|
}
|
||||||
|
|
||||||
|
let reader = BrowseReader::open(&db_path).expect("open fixture browse db");
|
||||||
|
world.insert_resource(BrowseReaderResource(reader));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drive one full server tick's worth of browse plumbing: `receive_bridge_inputs`
|
||||||
|
/// (drains the socket into `BrowseRequestBuffer`, connection-tagged),
|
||||||
|
/// `serve_browse_requests` (the atlas-plugin proxy — reads `BrowseReaderResource`,
|
||||||
|
/// fills `BrowseResponseBuffer`), and `send_browse_responses` (flushes back
|
||||||
|
/// out to the originating connection). Matches how `BridgePlugin` +
|
||||||
|
/// `GenerationPlugin` actually schedule these three systems in `PreInput`/
|
||||||
|
/// `PostSnapshot`, just run directly rather than through a full `App`.
|
||||||
|
fn drive_browse_tick(world: &mut bevy_ecs::world::World) {
|
||||||
|
use bevy_ecs::system::RunSystemOnce;
|
||||||
|
use settled_reach_server::atlas::plugin::serve_browse_requests;
|
||||||
|
use settled_reach_server::bridge::{receive_bridge_inputs, send_browse_responses};
|
||||||
|
|
||||||
|
world
|
||||||
|
.run_system_once(receive_bridge_inputs)
|
||||||
|
.expect("receive_bridge_inputs failed to run");
|
||||||
|
world
|
||||||
|
.run_system_once(serve_browse_requests)
|
||||||
|
.expect("serve_browse_requests failed to run");
|
||||||
|
world
|
||||||
|
.run_system_once(send_browse_responses)
|
||||||
|
.expect("send_browse_responses failed to run");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drive `drive_browse_tick` repeatedly until a framed `BrowseResponse`
|
||||||
|
/// arrives on `stream`, or a wall-clock deadline expires (this file's
|
||||||
|
/// established hardening pattern — see `drive_accept_loop_until`/
|
||||||
|
/// `collect_input_batches` — rather than a single tick or a fixed sleep).
|
||||||
|
///
|
||||||
|
/// A single `drive_browse_tick` call races the client thread's write
|
||||||
|
/// actually landing in the server's non-blocking socket buffer before
|
||||||
|
/// `receive_bridge_inputs` polls it — `stream` here is a genuinely blocking
|
||||||
|
/// client-side socket (only the SERVER's accepted connections are toggled
|
||||||
|
/// non-blocking, by `TcpBridge::from_connected_stream`/D-254 §2's
|
||||||
|
/// accept-loop design), so `set_read_timeout` + a bounded retry loop is the
|
||||||
|
/// correct fix, not a longer single wait.
|
||||||
|
fn drive_browse_tick_until_response(
|
||||||
|
world: &mut bevy_ecs::world::World,
|
||||||
|
stream: &mut TcpStream,
|
||||||
|
) -> BrowseResponse {
|
||||||
|
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(5);
|
||||||
|
loop {
|
||||||
|
drive_browse_tick(world);
|
||||||
|
match read_framed(stream) {
|
||||||
|
Ok(Some(payload)) => {
|
||||||
|
return rmp_serde::from_slice(&payload).expect("failed to decode BrowseResponse");
|
||||||
|
}
|
||||||
|
Ok(None) => panic!("unexpected EOF reading browse 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 a browse response"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Err(e) => panic!("failed to read browse response frame: {e}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn send_browse_index_request(stream: &mut TcpStream, kind: BrowseEntityKind) {
|
||||||
|
let req = BrowseRequest {
|
||||||
|
browse: true,
|
||||||
|
kind,
|
||||||
|
query: BrowseQuery::Index {
|
||||||
|
filter_system_id: None,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
let payload = rmp_serde::to_vec_named(&req).expect("failed to encode BrowseRequest");
|
||||||
|
write_framed(stream, &payload).expect("failed to write BrowseRequest");
|
||||||
|
}
|
||||||
|
|
||||||
|
fn send_browse_detail_request(stream: &mut TcpStream, kind: BrowseEntityKind, id: &str) {
|
||||||
|
let req = BrowseRequest {
|
||||||
|
browse: true,
|
||||||
|
kind,
|
||||||
|
query: BrowseQuery::Detail { id: id.to_string() },
|
||||||
|
};
|
||||||
|
let payload = rmp_serde::to_vec_named(&req).expect("failed to encode BrowseRequest");
|
||||||
|
write_framed(stream, &payload).expect("failed to write BrowseRequest");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// T-1131: index + detail round-trip for all six D-254 §4 v1 entity kinds,
|
||||||
|
/// over one Reader connection, through the full demux->serve->send pipeline.
|
||||||
|
/// Parameterized-style — one test iterating all six kinds, per the ticket's
|
||||||
|
/// own suggested test shape.
|
||||||
|
#[test]
|
||||||
|
fn browse_index_and_detail_round_trip_for_all_six_kinds_over_tcp() {
|
||||||
|
let (mut world, addr) = new_multi_connection_world();
|
||||||
|
wire_browse_reader(&mut world);
|
||||||
|
|
||||||
|
let client_handle = thread::spawn(move || {
|
||||||
|
let stream = TcpStream::connect(addr).expect("failed to connect");
|
||||||
|
client_handshake(stream, ConnectionRole::Reader)
|
||||||
|
});
|
||||||
|
drive_accept_loop_until(&mut world, |w| {
|
||||||
|
w.resource::<BridgeResource>().reader_count() == 1
|
||||||
|
});
|
||||||
|
let mut stream = client_handle.join().expect("client thread panicked");
|
||||||
|
|
||||||
|
let cases: &[(BrowseEntityKind, &str)] = &[
|
||||||
|
(BrowseEntityKind::StarSystem, "GJ-1"),
|
||||||
|
(BrowseEntityKind::Body, "GJ1c"),
|
||||||
|
(BrowseEntityKind::Station, "GJ1c-S1"),
|
||||||
|
(BrowseEntityKind::Corporation, "gate-corporation"),
|
||||||
|
(BrowseEntityKind::Commodity, "fusion_fuel"),
|
||||||
|
(BrowseEntityKind::TraitTemplate, "frontier_utilitarian"),
|
||||||
|
];
|
||||||
|
|
||||||
|
for &(kind, id) in cases {
|
||||||
|
send_browse_index_request(&mut stream, kind);
|
||||||
|
let index_resp = drive_browse_tick_until_response(&mut world, &mut stream);
|
||||||
|
assert_eq!(index_resp.kind, kind, "index response kind echo");
|
||||||
|
assert_eq!(index_resp.status, BrowseStatus::Ready, "index({kind:?})");
|
||||||
|
let rows = index_resp.index.expect("index populated");
|
||||||
|
assert!(
|
||||||
|
rows.iter().any(|r| r.id == id),
|
||||||
|
"index({kind:?}) should list {id}"
|
||||||
|
);
|
||||||
|
|
||||||
|
send_browse_detail_request(&mut stream, kind, id);
|
||||||
|
let detail_resp = drive_browse_tick_until_response(&mut world, &mut stream);
|
||||||
|
assert_eq!(detail_resp.kind, kind, "detail response kind echo");
|
||||||
|
assert_eq!(detail_resp.status, BrowseStatus::Ready, "detail({kind:?})");
|
||||||
|
assert!(
|
||||||
|
detail_resp.detail.is_some(),
|
||||||
|
"detail({kind:?}) should be populated"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// T-1131: a Reader (not just a Player) can send `BrowseRequest` and get a
|
||||||
|
/// `Ready` response — proves the permitted-message-matrix row (D-254 §2:
|
||||||
|
/// atlas/star-map/city-names/browse all say "yes" for Reader) actually holds
|
||||||
|
/// for the new request type specifically, not just by code inspection.
|
||||||
|
#[test]
|
||||||
|
fn reader_can_browse() {
|
||||||
|
let (mut world, addr) = new_multi_connection_world();
|
||||||
|
wire_browse_reader(&mut world);
|
||||||
|
|
||||||
|
let client_handle = thread::spawn(move || {
|
||||||
|
let stream = TcpStream::connect(addr).expect("failed to connect");
|
||||||
|
client_handshake(stream, ConnectionRole::Reader)
|
||||||
|
});
|
||||||
|
drive_accept_loop_until(&mut world, |w| {
|
||||||
|
w.resource::<BridgeResource>().reader_count() == 1
|
||||||
|
});
|
||||||
|
let mut stream = client_handle.join().expect("client thread panicked");
|
||||||
|
|
||||||
|
send_browse_index_request(&mut stream, BrowseEntityKind::Commodity);
|
||||||
|
let resp = drive_browse_tick_until_response(&mut world, &mut stream);
|
||||||
|
assert_eq!(resp.status, BrowseStatus::Ready);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// T-1131: a Player (not just a Reader) can also browse — the permitted-
|
||||||
|
/// message-matrix row says "yes" for both roles, and this is the other half
|
||||||
|
/// of that proof.
|
||||||
|
#[test]
|
||||||
|
fn player_can_browse() {
|
||||||
|
let (mut world, addr) = new_multi_connection_world();
|
||||||
|
wire_browse_reader(&mut world);
|
||||||
|
|
||||||
|
let client_handle = thread::spawn(move || {
|
||||||
|
let stream = TcpStream::connect(addr).expect("failed to connect");
|
||||||
|
client_handshake(stream, ConnectionRole::Player)
|
||||||
|
});
|
||||||
|
drive_accept_loop_until(&mut world, |w| w.resource::<BridgeResource>().has_player());
|
||||||
|
let mut stream = client_handle.join().expect("client thread panicked");
|
||||||
|
|
||||||
|
send_browse_index_request(&mut stream, BrowseEntityKind::TraitTemplate);
|
||||||
|
let resp = drive_browse_tick_until_response(&mut world, &mut stream);
|
||||||
|
assert_eq!(resp.status, BrowseStatus::Ready);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// T-1131: two concurrently-connected Readers each get back only their OWN
|
||||||
|
/// browse response — the connection-tagging plumbing D-254 §2 established
|
||||||
|
/// for atlas/star-map/city-names must hold for browse too, proven with two
|
||||||
|
/// simultaneously-live sockets asking for DIFFERENT things so a crossed wire
|
||||||
|
/// would be immediately visible (not just "a response arrived", but "the
|
||||||
|
/// WRONG response arrived").
|
||||||
|
#[test]
|
||||||
|
fn two_readers_do_not_cross_browse_responses() {
|
||||||
|
let (mut world, addr) = new_multi_connection_world();
|
||||||
|
wire_browse_reader(&mut world);
|
||||||
|
|
||||||
|
let reader_a_handle = thread::spawn(move || {
|
||||||
|
let stream = TcpStream::connect(addr).expect("failed to connect (reader A)");
|
||||||
|
client_handshake(stream, ConnectionRole::Reader)
|
||||||
|
});
|
||||||
|
drive_accept_loop_until(&mut world, |w| {
|
||||||
|
w.resource::<BridgeResource>().reader_count() == 1
|
||||||
|
});
|
||||||
|
let mut stream_a = reader_a_handle.join().expect("reader A thread panicked");
|
||||||
|
|
||||||
|
let reader_b_handle = thread::spawn(move || {
|
||||||
|
let stream = TcpStream::connect(addr).expect("failed to connect (reader B)");
|
||||||
|
client_handshake(stream, ConnectionRole::Reader)
|
||||||
|
});
|
||||||
|
drive_accept_loop_until(&mut world, |w| {
|
||||||
|
w.resource::<BridgeResource>().reader_count() == 2
|
||||||
|
});
|
||||||
|
let mut stream_b = reader_b_handle.join().expect("reader B thread panicked");
|
||||||
|
|
||||||
|
// A asks about star systems, B asks about commodities — deliberately
|
||||||
|
// different kinds so a crossed response is unmistakable (not just "wrong
|
||||||
|
// data", but "wrong KIND"). Both requests are in flight before any
|
||||||
|
// response is expected, so `drive_browse_tick` is retried until BOTH
|
||||||
|
// streams have something readable (rather than draining/reading one
|
||||||
|
// stream to completion before the other's request has even arrived).
|
||||||
|
send_browse_index_request(&mut stream_a, BrowseEntityKind::StarSystem);
|
||||||
|
send_browse_index_request(&mut stream_b, BrowseEntityKind::Commodity);
|
||||||
|
|
||||||
|
stream_a
|
||||||
|
.set_read_timeout(Some(std::time::Duration::from_millis(50)))
|
||||||
|
.expect("failed to set read timeout (A)");
|
||||||
|
stream_b
|
||||||
|
.set_read_timeout(Some(std::time::Duration::from_millis(50)))
|
||||||
|
.expect("failed to set read timeout (B)");
|
||||||
|
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||||
|
let mut resp_a: Option<BrowseResponse> = None;
|
||||||
|
let mut resp_b: Option<BrowseResponse> = None;
|
||||||
|
while resp_a.is_none() || resp_b.is_none() {
|
||||||
|
drive_browse_tick(&mut world);
|
||||||
|
if resp_a.is_none() {
|
||||||
|
if let Ok(Some(payload)) = read_framed(&mut stream_a) {
|
||||||
|
resp_a = Some(rmp_serde::from_slice(&payload).expect("decode A"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if resp_b.is_none() {
|
||||||
|
if let Ok(Some(payload)) = read_framed(&mut stream_b) {
|
||||||
|
resp_b = Some(rmp_serde::from_slice(&payload).expect("decode B"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert!(
|
||||||
|
std::time::Instant::now() < deadline,
|
||||||
|
"timed out waiting for both readers' browse responses (A={}, B={})",
|
||||||
|
resp_a.is_some(),
|
||||||
|
resp_b.is_some()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let resp_a = resp_a.expect("resp_a set by loop exit condition");
|
||||||
|
let resp_b = resp_b.expect("resp_b set by loop exit condition");
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
resp_a.kind,
|
||||||
|
BrowseEntityKind::StarSystem,
|
||||||
|
"reader A must get back ITS OWN request's kind, not reader B's"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
resp_b.kind,
|
||||||
|
BrowseEntityKind::Commodity,
|
||||||
|
"reader B must get back ITS OWN request's kind, not reader A's"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// T-1131: a `Detail` request for an id that doesn't exist in that kind's
|
||||||
|
/// table comes back `NotFound` over the wire (not an error, not a hang, not
|
||||||
|
/// silently dropped).
|
||||||
|
#[test]
|
||||||
|
fn browse_detail_unknown_id_is_not_found_over_tcp() {
|
||||||
|
let (mut world, addr) = new_multi_connection_world();
|
||||||
|
wire_browse_reader(&mut world);
|
||||||
|
|
||||||
|
let client_handle = thread::spawn(move || {
|
||||||
|
let stream = TcpStream::connect(addr).expect("failed to connect");
|
||||||
|
client_handshake(stream, ConnectionRole::Reader)
|
||||||
|
});
|
||||||
|
drive_accept_loop_until(&mut world, |w| {
|
||||||
|
w.resource::<BridgeResource>().reader_count() == 1
|
||||||
|
});
|
||||||
|
let mut stream = client_handle.join().expect("client thread panicked");
|
||||||
|
|
||||||
|
send_browse_detail_request(&mut stream, BrowseEntityKind::Corporation, "no-such-corp");
|
||||||
|
let resp = drive_browse_tick_until_response(&mut world, &mut stream);
|
||||||
|
assert_eq!(resp.status, BrowseStatus::NotFound);
|
||||||
|
assert!(resp.detail.is_none());
|
||||||
|
}
|
||||||
|
|
||||||
|
/// T-1131: an `Index` request against a kind with zero rows still comes back
|
||||||
|
/// `Ready` with an empty list, matching the existing city-names convention
|
||||||
|
/// (`CityNamesStatus`'s "unknown body -> Ready, empty" pattern) — over the
|
||||||
|
/// real wire, not just at the reader layer (see
|
||||||
|
/// `browse_reader::tests::index_stations_empty_table_is_empty_vec` for that
|
||||||
|
/// half of the proof).
|
||||||
|
#[test]
|
||||||
|
fn browse_index_empty_table_is_ready_with_empty_list_over_tcp() {
|
||||||
|
use settled_reach_server::atlas::browse_reader::{BrowseReader, BrowseReaderResource};
|
||||||
|
|
||||||
|
let (mut world, addr) = new_multi_connection_world();
|
||||||
|
|
||||||
|
// A fixture db with the stations table present but genuinely empty — a
|
||||||
|
// distinct db from wire_browse_reader's shared fixture (which always has
|
||||||
|
// one station), built directly here so this test controls the "empty"
|
||||||
|
// precondition explicitly rather than relying on incidental fixture state.
|
||||||
|
let db_path =
|
||||||
|
std::env::temp_dir().join(format!("sr_browse_tcp_empty_{}.db", std::process::id()));
|
||||||
|
let _ = std::fs::remove_file(&db_path);
|
||||||
|
{
|
||||||
|
let conn = rusqlite::Connection::open(&db_path).expect("create empty fixture db");
|
||||||
|
conn.execute_batch(
|
||||||
|
"CREATE TABLE stations (
|
||||||
|
station_id TEXT PRIMARY KEY, system_id TEXT NOT NULL, orbits_body_id TEXT,
|
||||||
|
station_type TEXT NOT NULL, proper_name TEXT, population INTEGER,
|
||||||
|
economic_role TEXT, governance_type TEXT, docking_class TEXT,
|
||||||
|
has_gate_infrastructure INTEGER DEFAULT 0, district_count INTEGER
|
||||||
|
);",
|
||||||
|
)
|
||||||
|
.expect("create empty stations table");
|
||||||
|
}
|
||||||
|
let reader = BrowseReader::open(&db_path).expect("open empty fixture db");
|
||||||
|
world.insert_resource(BrowseReaderResource(reader));
|
||||||
|
|
||||||
|
let client_handle = thread::spawn(move || {
|
||||||
|
let stream = TcpStream::connect(addr).expect("failed to connect");
|
||||||
|
client_handshake(stream, ConnectionRole::Reader)
|
||||||
|
});
|
||||||
|
drive_accept_loop_until(&mut world, |w| {
|
||||||
|
w.resource::<BridgeResource>().reader_count() == 1
|
||||||
|
});
|
||||||
|
let mut stream = client_handle.join().expect("client thread panicked");
|
||||||
|
|
||||||
|
send_browse_index_request(&mut stream, BrowseEntityKind::Station);
|
||||||
|
let resp = drive_browse_tick_until_response(&mut world, &mut stream);
|
||||||
|
assert_eq!(resp.status, BrowseStatus::Ready);
|
||||||
|
assert_eq!(resp.index.unwrap().len(), 0);
|
||||||
|
|
||||||
|
let _ = std::fs::remove_file(&db_path);
|
||||||
|
}
|
||||||
|
|
||||||
/// Minimal `ObserverSnapshot` for the tests above — same shape as
|
/// Minimal `ObserverSnapshot` for the tests above — same shape as
|
||||||
/// `snapshot_roundtrip_over_tcp`'s at the top of this file, parameterized
|
/// `snapshot_roundtrip_over_tcp`'s at the top of this file, parameterized
|
||||||
/// only by `tick` (the one field these tests assert on).
|
/// only by `tick` (the one field these tests assert on).
|
||||||
|
|||||||
Reference in New Issue
Block a user