diff --git a/.pql/changelog/ticket_idmap/2026-06.sql b/.pql/changelog/ticket_idmap/2026-06.sql index cdbb6793b..06eeccd1c 100644 --- a/.pql/changelog/ticket_idmap/2026-06.sql +++ b/.pql/changelog/ticket_idmap/2026-06.sql @@ -79,3 +79,4 @@ INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_ INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBPTXS5YZ04F4CPPYNZS4Z2W', 'T-1069', '2026-06-12 10:40:59', '2026-06-12 10:40:59', NULL, 'b91a55eefa15f117a0222e7fe8656d6e', 2) ON CONFLICT(record_id) DO UPDATE SET ticket_id=excluded.ticket_id, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > ticket_idmap.updated_at OR (excluded.updated_at = ticket_idmap.updated_at AND excluded.hash > ticket_idmap.hash); INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBPTXSA4WFXPD21YK7N8SF60', 'T-1070', '2026-06-12 10:40:59', '2026-06-12 10:40:59', NULL, '7a1bce4e8c3705623cf50834005ba19a', 2) ON CONFLICT(record_id) DO UPDATE SET ticket_id=excluded.ticket_id, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > ticket_idmap.updated_at OR (excluded.updated_at = ticket_idmap.updated_at AND excluded.hash > ticket_idmap.hash); INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBPTXSEC1F0RKJVH7JVPS8NW', 'T-1071', '2026-06-12 10:40:59', '2026-06-12 10:40:59', NULL, 'b3f82a325e1724e3d218fe54ba344f64', 2) ON CONFLICT(record_id) DO UPDATE SET ticket_id=excluded.ticket_id, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > ticket_idmap.updated_at OR (excluded.updated_at = ticket_idmap.updated_at AND excluded.hash > ticket_idmap.hash); +INSERT INTO ticket_idmap (record_id, ticket_id, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBQ8WGMA3552756BQAAFPMER', 'T-1072', '2026-06-12 11:41:59', '2026-06-12 11:41:59', NULL, 'fefd6fa8565a07b57ff8bbd5cadfeb7e', 2) ON CONFLICT(record_id) DO UPDATE SET ticket_id=excluded.ticket_id, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > ticket_idmap.updated_at OR (excluded.updated_at = ticket_idmap.updated_at AND excluded.hash > ticket_idmap.hash); diff --git a/.pql/changelog/tickets/2026-06.sql b/.pql/changelog/tickets/2026-06.sql index 673efe217..416655eb9 100644 --- a/.pql/changelog/tickets/2026-06.sql +++ b/.pql/changelog/tickets/2026-06.sql @@ -1607,3 +1607,34 @@ INSERT INTO tickets (record_id, type, parent_record_id, title, description, stat Round-3 batch refinement (2026-05-26): also define the MorphologyZone enum here (closed set: Fjord, Canyon, Delta, AlluvialPlain, Island, MountainPass, Coastal, Lake, Sea, OpenOcean, MeanderReach, Other) -- it had no home and 954/976/1005 all need it. bulk_class is the 8->5 PROJECTION (size variants bulk/standard/compact/oversized -> BulkSolid; precision -> PrecisionDense; the 8 stay on commodities). Note: the bare enum/struct type DEFS now live in the foundational types task; 982 owns the projection + dominant-commodity DERIVATION logic (derived from highest-output location-bound production chain, D-178, not an authored role->commodity table). Status check 2026-06-12 (fable-ous audit S-62): flipped in_progress → backlog. Verified the enum/type half landed via T-1006 (BulkClass/ProductionUbiquity + context fields, generator.rs:1016/1039/1097-1100), but this ticket''s remaining scope — the 8→5 bulk_class projection, D-184 backfill, and the dominant-commodity DERIVATION (highest-output location-bound chain per D-178) — has zero hits in city_context_reader.rs. Nobody is actively on it; it was stale-WIP for 18 days while being T-1014''s sole live blocker. Sequencing input: Q-111 (live-econ coupling) decides whether the dominant-commodity lookup stays static build-time.', 'backlog', 'medium', NULL, 'server', 'D-233', '2026-05-25 19:11:05', '2026-06-12 10:41:25', NULL, 'fe1326ed22e42707c3b33575638db3a8', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBPQ02QTVSZ1PXTGNCN6Y8KM', 'bug', '06FB0TNSRZXCHGS16BFHSSGSV4', 'BUG: river channel centrelines anchored to the world-origin axis — channels absent in almost every chunk', 'Workshopped-but-unticketed cascade work captured 2026-06-12 (cascade-refocus grounding pass). Full grounded scope follows. + +BUG: channel centrelines are anchored to the world-origin axis. compute_channel_state (AlluvialPlain, server/src/atlas/voxel.rs:1648-1707) and compute_meander_channel_state (MeanderReach, :1556-1595) compute perp_distance = (cross - meander_displacement).abs() (:1694, :1583) where cross is the raw world-space voxel coordinate (derive_voxel_column takes tile_x/tile_y in world metres, :328-338; cross = vx or vy at :1653-1661). The channel centreline is therefore the world x=0/y=0 axis plus-or-minus meander amplitude: only chunks within ~(channel_width/2 + amplitude + jitter) metres of the world origin axis ever contain channel water, regardless of ChunkContext.has_active_channel / channel_width_m (chunk_context.rs:103-110). in_levee_band (:1602-1633) inherits the same defect, so levees are origin-locked too. Every has_active_channel chunk elsewhere on the body silently renders dry floodplain. + +Intended model (workshopped): channel positions derive from region/chunk-scale data, not world axes — D-239 §10 (one basin-direction + global meander-curve params; wavelength > 64 m seeds Region-or-higher) and tile-derivation-contract/gestalt-round1.md:152-162 (channel branch positions from a multi-seed Voronoi partition of the chunk area with D8 flow direction as attractor; meander_intensity controls deviation from the straight downslope path). + +Fix, two stages: +1. Minimal correctness: derive a deterministic cross-offset channel anchor at region scale (alongside meander_phase in derive_chunk_context, chunk_context.rs:124-141 — same Region-or-higher seed path) and measure perp_distance from that anchor, so an active channel actually crosses the chunks the region says it crosses, continuously across chunk boundaries. +2. Full model: gestalt''s multi-seed Voronoi + D8-attractor branch placement when BraidedDelta sub-classification (D-239 implementation note: deferred to ChunkContext) lands. + +Add a T-1031-harness believability check: a has_active_channel chunk at an arbitrary large world offset (e.g. chunk_pos (1000, -750)) contains in-channel voxels; channel position is continuous across adjacent chunk pairs. Sibling bug: landform-64m-repeat (same family, opposite direction — channels appear nowhere, landforms appear everywhere). Parent: T-750. Refs: D-239 §10, D-208 (D8), D-010; T-1029 (origin), T-1031 (harness).', 'in_progress', 'high', NULL, 'server', 'D-239', '2026-06-12 10:23:49', '2026-06-12 10:56:40', NULL, '6744373fe8042356f58e89d8a889071b', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBPTXNWWGS3S0RFMFYQC9BWR', 'bug', '06FB0TNSRZXCHGS16BFHSSGSV4', 'Bridge: fix TCP partial-read frame desync; drain all ready inbound frames per tick', '(description follows in first append) + +Filed 2026-06-12 from the fable-ous.md audit (S-01, S-02; citations adversarially verified). +DESYNC (S-01, high): the stream is non-blocking (tcp.rs:48-52); read_framed uses read_exact for the 4-byte prefix and payload with no partial-frame state (framing.rs:36-61); read_exact hitting WouldBlock mid-frame irrecoverably consumes bytes, and receive() maps WouldBlock to Ok(None) (tcp.rs:211) — a frame split across TCP segments permanently desyncs the stream (next read treats mid-frame bytes as a length prefix). Works today only because localhost frames arrive in single segments; BufReader is bypassed via get_mut(). Fix: persistent partial-frame accumulation buffer on TcpBridge (or blocking reader thread feeding a channel). Regression test: deliver one frame in two writes with flush+delay. +DRAIN (S-02, med): receive_bridge_inputs calls receive() exactly once per 50 ms tick (bridge/mod.rs:162, no loop; tick at main.rs:306) — an input batch + AtlasLayerRequest in one window backlogs and drains one frame per tick. Fix: loop receive() until Ok(None) with a per-tick cap. The D-225 atlas channel shares this path, so both fixes gate Atlas streaming reliability. +Extension constraint when touching this code: D-225 amendment 2026-06-12 (tagged envelope for the next inbound type).', 'in_progress', 'high', NULL, 'server', 'D-225', '2026-06-12 10:40:58', '2026-06-12 10:56:40', NULL, '94f6ede90ac014280daee68a1216466a', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBPQ02VWJGXWDYT2SMXM9DTC', 'bug', '06FB0TNSRZXCHGS16BFHSSGSV4', 'BUG: fjord/cliff/gorge/delta landforms repeat on the 64 m chunk grid (D-239 §10 violation)', 'Workshopped-but-unticketed cascade work captured 2026-06-12 (cascade-refocus grounding pass). Full grounded scope follows. + +BUG: four morphology generators fold world coordinates into the 64 m chunk frame, so region-scale landforms repeat every chunk — violating D-239 §10 (''features with wavelength > 64 m seed from Region-or-higher, not the chunk seed''; deliberately strengthened from Block-or-higher in the adversarial verification pass, tile-derivation-contract/workshop-outcomes.md:83). + +Sites (server/src/atlas/voxel.rs): +- generate_fjord_wall :594-597 — cross.rem_euclid(64)-32: a complete fjord (floor|moraine|wall cross-section) centred in EVERY chunk of a FjordWall region; a real fjord valley is one feature spanning the region. +- generate_cliff_coast :721 — seaward_coord.rem_euclid(64): the cliff face sits at intra-chunk offset 48-55 in every chunk, sawtoothing the coastline at 64 m pitch. +- generate_braided_delta :825 — cross.rem_euclid(64): the same three braid threads restart per chunk instead of anastomosing across the fan. +- generate_incised_gorge :1054-1057 — same fold: a gorge per chunk. + +Fix: compute each landform''s axis/centreline in REGION-anchored coordinates — derive the feature anchor (centreline position, coast-face line) once per region from the region-scale seed path that meander_phase/basin_direction already use correctly (chunk_context.rs:124-141), then measure each voxel''s distance to that anchor in continuous world coordinates (no rem_euclid against the chunk size). This matches the workshop''s seam-freedom principle: the algorithm must see positions with no knowledge of the chunk grid (tyre-round1.md:174-176). Preserve D-239 §9 chokepoint widths (fjord floor 4-8 m, gorge floor 2-8 m — currently honoured at :607-609, :1059-1062). + +Harness gap to close (T-1031 follow-up): current tests check per-chunk boundary smoothness, not feature uniqueness — add a multi-chunk assertion that a FjordWall/IncisedGorge region contains ONE valley spanning its chunks, and a CliffCoast region one continuous (warp-displaced) coast line. Sibling bug: river-channel-anchor (channels are origin-anchored and appear nowhere; landforms are chunk-folded and appear everywhere — fix both against the same region-anchor mechanism). Depends on scale-anchoring Q for the region''s physical extent, but the region-index anchor works regardless. Parent: T-750. Refs: D-239 §5/§8/§9/§10, D-010; T-1029, T-1031.', 'in_progress', 'high', NULL, 'server', 'D-239', '2026-06-12 10:23:49', '2026-06-12 10:56:40', NULL, 'c1189a915611efac3027617e7915776a', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FBQ8WGMA3552756BQAAFPMER', 'task', '06FB0TNSRZXCHGS16BFHSSGSV4', 'Bridge: escalate persistent corrupt-stream errors to shutdown (oversized-prefix poison)', 'Follow-up from the T-1045 review (2026-06-12). The receive_bridge_inputs catch-all arm logs unknown bridge errors once per tick and retries; a protocol-violating stream is unrecoverable by design (the FrameAccumulator oversized-prefix guard deliberately re-reports InvalidData on every poll), so a corrupt stream logs at 20 Hz forever with no escalation and no client feedback. Roughly parity with pre-T-1045 behavior (garbage cascade), so not a regression — but decide and implement escalation: map InvalidData to a fatal arm in receive_bridge_inputs, or count consecutive identical-error ticks and shut down after N. See the catch-all comment in server/src/bridge/mod.rs (receive_bridge_inputs) and the poison-state comment in framing.rs. Optional rider from the same review: harden assert_split_frame_does_not_desync (server/tests/bridge_tcp.rs) with channel synchronization so the mid-frame poll is guaranteed rather than timing-likely.', 'backlog', 'low', NULL, 'server', NULL, '2026-06-12 11:41:59', '2026-06-12 11:41:59', NULL, '77ded3c2d4cf7c6c623868bdb7ff5dd5', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); diff --git a/server/src/bridge/framing.rs b/server/src/bridge/framing.rs index e9bbf51b7..2a24162e8 100644 --- a/server/src/bridge/framing.rs +++ b/server/src/bridge/framing.rs @@ -60,6 +60,110 @@ pub fn read_framed(reader: &mut impl Read) -> io::Result>> { Ok(Some(payload)) } +/// Incremental frame reader for non-blocking streams (T-1045). +/// +/// `read_framed` uses `read_exact`, which consumes bytes before erroring — a +/// `WouldBlock` mid-frame on a non-blocking socket loses the bytes already +/// read and permanently desyncs the stream (the next read treats mid-frame +/// bytes as a length prefix). The accumulator instead retains partial +/// prefix/payload bytes across calls, so a frame split across TCP segments +/// is reassembled over multiple polls. +pub struct FrameAccumulator { + state: AccumState, +} + +enum AccumState { + /// Reading the 4-byte big-endian length prefix. + Prefix { buf: [u8; 4], filled: usize }, + /// Prefix complete; reading `buf.len()` payload bytes. + Payload { buf: Vec, filled: usize }, +} + +impl Default for FrameAccumulator { + fn default() -> Self { + Self::new() + } +} + +impl FrameAccumulator { + pub fn new() -> Self { + Self { + state: AccumState::Prefix { + buf: [0u8; 4], + filled: 0, + }, + } + } + + /// Pump bytes from `reader` into the current frame. + /// + /// - `Ok(Some(payload))` — one complete frame. + /// - `Ok(None)` — clean EOF at a frame boundary (same as `read_framed`). + /// - `Err(WouldBlock)` — no complete frame yet; accumulated bytes are + /// retained and the next call resumes mid-frame. + /// - `Err(UnexpectedEof)` — EOF mid-frame (truncated stream). + pub fn poll_frame(&mut self, reader: &mut impl Read) -> io::Result>> { + loop { + match &mut self.state { + AccumState::Prefix { buf, filled } => { + while *filled < buf.len() { + match reader.read(&mut buf[*filled..]) { + Ok(0) if *filled == 0 => return Ok(None), // clean EOF + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "EOF inside length prefix", + )); + } + Ok(n) => *filled += n, + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, + Err(e) => return Err(e), + } + } + let len = u32::from_be_bytes(*buf); + if len > MAX_MESSAGE_SIZE { + // State is left with the prefix filled: the stream is + // corrupt, so every subsequent poll re-reports the + // error instead of misreading payload bytes as a prefix. + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "message too large: {} bytes (max {})", + len, MAX_MESSAGE_SIZE + ), + )); + } + self.state = AccumState::Payload { + buf: vec![0u8; len as usize], + filled: 0, + }; + } + AccumState::Payload { buf, filled } => { + while *filled < buf.len() { + match reader.read(&mut buf[*filled..]) { + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "EOF inside payload", + )); + } + Ok(n) => *filled += n, + Err(e) if e.kind() == io::ErrorKind::Interrupted => continue, + Err(e) => return Err(e), + } + } + let payload = std::mem::take(buf); + self.state = AccumState::Prefix { + buf: [0u8; 4], + filled: 0, + }; + return Ok(Some(payload)); + } + } + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -109,4 +213,125 @@ mod tests { let result = read_framed(&mut cursor).expect("read failed"); assert_eq!(result, None); } + + /// Scripted reader for FrameAccumulator tests: each `Some(bytes)` step is + /// delivered (possibly partially) by `read`; each `None` step yields one + /// WouldBlock; an exhausted script yields EOF. + struct ScriptedReader { + steps: std::collections::VecDeque>>, + } + + impl ScriptedReader { + fn new(steps: Vec>>) -> Self { + Self { + steps: steps.into(), + } + } + } + + impl Read for ScriptedReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + match self.steps.pop_front() { + Some(Some(mut bytes)) => { + let n = bytes.len().min(buf.len()); + buf[..n].copy_from_slice(&bytes[..n]); + if n < bytes.len() { + self.steps.push_front(Some(bytes.split_off(n))); + } + Ok(n) + } + Some(None) => Err(io::Error::new(io::ErrorKind::WouldBlock, "no data")), + None => Ok(0), // EOF + } + } + } + + fn frame_bytes(payload: &[u8]) -> Vec { + let mut frame = Vec::new(); + write_framed(&mut frame, payload).expect("write failed"); + frame + } + + #[test] + fn accumulator_survives_split_inside_prefix() { + let frame = frame_bytes(b"split prefix"); + let mut reader = ScriptedReader::new(vec![ + Some(frame[..2].to_vec()), // half the length prefix + None, // WouldBlock mid-prefix + Some(frame[2..].to_vec()), + ]); + + let mut accum = FrameAccumulator::new(); + let err = accum.poll_frame(&mut reader).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::WouldBlock); + + let payload = accum + .poll_frame(&mut reader) + .expect("read failed") + .expect("expected frame"); + assert_eq!(payload, b"split prefix"); + } + + #[test] + fn accumulator_survives_split_inside_payload() { + let frame = frame_bytes(b"split payload"); + let mut reader = ScriptedReader::new(vec![ + Some(frame[..7].to_vec()), // prefix + 3 payload bytes + None, // WouldBlock mid-payload + Some(frame[7..].to_vec()), + ]); + + let mut accum = FrameAccumulator::new(); + let err = accum.poll_frame(&mut reader).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::WouldBlock); + + let payload = accum + .poll_frame(&mut reader) + .expect("read failed") + .expect("expected frame"); + assert_eq!(payload, b"split payload"); + } + + #[test] + fn accumulator_reads_consecutive_frames_from_one_chunk() { + let mut bytes = frame_bytes(b"first"); + bytes.extend_from_slice(&frame_bytes(b"second")); + let mut reader = ScriptedReader::new(vec![Some(bytes)]); + + let mut accum = FrameAccumulator::new(); + let first = accum.poll_frame(&mut reader).unwrap().unwrap(); + let second = accum.poll_frame(&mut reader).unwrap().unwrap(); + assert_eq!(first, b"first"); + assert_eq!(second, b"second"); + } + + #[test] + fn accumulator_clean_eof_at_frame_boundary_returns_none() { + let frame = frame_bytes(b"only"); + let mut reader = ScriptedReader::new(vec![Some(frame)]); + + let mut accum = FrameAccumulator::new(); + assert_eq!(accum.poll_frame(&mut reader).unwrap().unwrap(), b"only"); + assert_eq!(accum.poll_frame(&mut reader).unwrap(), None); + } + + #[test] + fn accumulator_eof_mid_frame_is_error() { + let frame = frame_bytes(b"truncated"); + let mut reader = ScriptedReader::new(vec![Some(frame[..6].to_vec())]); + + let mut accum = FrameAccumulator::new(); + let err = accum.poll_frame(&mut reader).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::UnexpectedEof); + } + + #[test] + fn accumulator_rejects_oversized_message() { + let prefix = (MAX_MESSAGE_SIZE + 1).to_be_bytes().to_vec(); + let mut reader = ScriptedReader::new(vec![Some(prefix)]); + + let mut accum = FrameAccumulator::new(); + let err = accum.poll_frame(&mut reader).unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::InvalidData); + } } diff --git a/server/src/bridge/mod.rs b/server/src/bridge/mod.rs index 8157f0343..da8efd0c6 100644 --- a/server/src/bridge/mod.rs +++ b/server/src/bridge/mod.rs @@ -89,8 +89,11 @@ pub trait SimBridge: Send + Sync { /// Send an observer snapshot to the client fn send_snapshot(&self, snapshot: &ObserverSnapshot) -> Result<(), BridgeError>; - /// Receive one inbound message, or `None` if no frame is ready this tick. + /// Receive one inbound message, or `None` if no complete frame is ready. /// The single client→server stream is demuxed by frame shape (D-225). + /// `receive_bridge_inputs` loops this until `None` (T-1045), so the + /// transport behind `BridgeResource` must not block when no frame is + /// buffered (TcpBridge is non-blocking; LocalBridge blocks — test-only). fn receive(&self) -> Result, BridgeError>; /// Send an atlas layer-stream response to the client (#969, D-225). @@ -144,7 +147,15 @@ pub enum HandshakeState { Complete, } +/// Per-tick cap on drained inbound frames (T-1045) — a safety valve so a +/// client flooding the stream cannot starve the simulation tick. Generous: +/// normal traffic is one input batch plus the occasional atlas request. +const MAX_INBOUND_FRAMES_PER_TICK: usize = 64; + /// Receive inputs from bridge and push to InputQueue. +/// Drains every complete frame buffered this tick (T-1045) — a single +/// receive() per tick would backlog mixed input/atlas traffic at one frame +/// per 50 ms. Relies on receive() being non-blocking (Ok(None) = no frame). /// Protocol errors (malformed input) are recoverable: the frame is skipped /// and a SimError is pushed to the SimErrorBuffer for client reporting (#85). pub fn receive_bridge_inputs( @@ -159,64 +170,77 @@ pub fn receive_bridge_inputs( let Some(bridge) = bridge else { return }; let current_tick = time.as_ref().map(|t| t.tick).unwrap_or(0); - match bridge.receive() { - Ok(Some(Inbound::Inputs(inputs))) => { - if !inputs.is_empty() && *handshake == HandshakeState::Pending { - tracing::warn!( - "Received {} input(s) before handshake completed — processing anyway (forward-compatible)", - inputs.len() - ); + for _ in 0..MAX_INBOUND_FRAMES_PER_TICK { + match bridge.receive() { + Ok(Some(Inbound::Inputs(inputs))) => { + if !inputs.is_empty() && *handshake == HandshakeState::Pending { + tracing::warn!( + "Received {} input(s) before handshake completed — processing anyway (forward-compatible)", + inputs.len() + ); + } + for input in &inputs { + tracing::trace!( + "Received input: tick={} action={:?}", + input.tick, + input.action + ); + } + for input in inputs { + input_queue.push(input); + } } - for input in &inputs { - tracing::trace!( - "Received input: tick={} action={:?}", - input.tick, - input.action - ); + Ok(Some(Inbound::AtlasRequest(req))) => { + atlas_requests.0.push(req); } - for input in inputs { - input_queue.push(input); + // No complete frame ready — the backlog is drained. + Ok(None) => break, + Err(BridgeError::Disconnected) => { + tracing::info!("Client disconnected, shutting down"); + running.0 = false; + break; + } + Err(BridgeError::Io(ref e)) + if e.kind() == std::io::ErrorKind::BrokenPipe + || e.kind() == std::io::ErrorKind::ConnectionReset => + { + tracing::info!("Pipe broken, shutting down cleanly"); + running.0 = false; + break; + } + Err(BridgeError::MutexPoisoned(ref msg)) => { + tracing::error!("Bridge mutex poisoned: {}. Shutting down.", msg); + running.0 = false; + break; + } + Err(BridgeError::DeserializationWithDump(ref msg)) => { + // Recoverable: skip this frame's input, report to client (#85), + // keep draining — the frame was consumed, later ones may be fine. + tracing::error!("Skipping malformed input frame: {}", msg); + error_buffer.push(SimError { + kind: SimErrorKind::ProtocolError, + message: format!("Malformed input frame: {}", msg), + tick: current_tick, + }); + } + Err(ref e @ BridgeError::Deserialization(_)) => { + // Recoverable deserialization error without dump + tracing::error!("Skipping malformed input: {}", e); + error_buffer.push(SimError { + kind: SimErrorKind::ProtocolError, + message: format!("Deserialization error: {}", e), + tick: current_tick, + }); + } + Err(e) => { + // Unknown error: log once per tick instead of hammering a + // persistently failing stream within one tick. A permanently + // corrupt stream (e.g. the oversized-prefix poison state) + // therefore logs every tick without escalation — follow-up + // ticket covers shutdown-after-N-consecutive-errors. + tracing::error!("Bridge receive error: {}", e); + break; } - } - Ok(Some(Inbound::AtlasRequest(req))) => { - atlas_requests.0.push(req); - } - Ok(None) => {} - Err(BridgeError::Disconnected) => { - tracing::info!("Client disconnected, shutting down"); - running.0 = false; - } - Err(BridgeError::Io(ref e)) - if e.kind() == std::io::ErrorKind::BrokenPipe - || e.kind() == std::io::ErrorKind::ConnectionReset => - { - tracing::info!("Pipe broken, shutting down cleanly"); - running.0 = false; - } - Err(BridgeError::MutexPoisoned(ref msg)) => { - tracing::error!("Bridge mutex poisoned: {}. Shutting down.", msg); - running.0 = false; - } - Err(BridgeError::DeserializationWithDump(ref msg)) => { - // Recoverable: skip this frame's input, report to client (#85) - tracing::error!("Skipping malformed input frame: {}", msg); - error_buffer.push(SimError { - kind: SimErrorKind::ProtocolError, - message: format!("Malformed input frame: {}", msg), - tick: current_tick, - }); - } - Err(ref e @ BridgeError::Deserialization(_)) => { - // Recoverable deserialization error without dump - tracing::error!("Skipping malformed input: {}", e); - error_buffer.push(SimError { - kind: SimErrorKind::ProtocolError, - message: format!("Deserialization error: {}", e), - tick: current_tick, - }); - } - Err(e) => { - tracing::error!("Bridge receive error: {}", e); } } } diff --git a/server/src/bridge/tcp.rs b/server/src/bridge/tcp.rs index 64f850cd8..65f53fcbe 100644 --- a/server/src/bridge/tcp.rs +++ b/server/src/bridge/tcp.rs @@ -5,15 +5,33 @@ use super::{decode_inbound, BridgeError, Inbound, ObserverSnapshot, SimBridge}; use crate::atlas::layer_proxy::AtlasLayerResponse; -use crate::bridge::framing::{read_framed, write_framed}; -use std::io::{BufReader, BufWriter}; +use crate::bridge::framing::{read_framed, write_framed, FrameAccumulator}; +use std::io::BufWriter; use std::net::{SocketAddr, TcpListener, TcpStream}; use std::sync::Mutex; +/// Read half of the bridge: the raw stream plus the partial-frame accumulator. +/// The accumulator must persist across `receive()` calls (T-1045) — the socket +/// is non-blocking, so a frame split across TCP segments is reassembled over +/// multiple ticks instead of desyncing the stream. +struct ReadHalf { + stream: TcpStream, + accum: FrameAccumulator, +} + +impl ReadHalf { + fn new(stream: TcpStream) -> Self { + Self { + stream, + accum: FrameAccumulator::new(), + } + } +} + /// TcpBridge: TCP transport for client-server IPC /// Same semantics as LocalBridge but over TCP localhost pub struct TcpBridge { - reader: Mutex>, + reader: Mutex, writer: Mutex>, local_addr: SocketAddr, } @@ -45,8 +63,9 @@ impl TcpBridge { local_addr ); - // Set non-blocking so receive_inputs doesn't stall the game loop. - // receive_inputs catches WouldBlock from read_framed and returns Ok(vec![]). + // Set non-blocking so receive() doesn't stall the game loop. A + // WouldBlock mid-frame is handled by the ReadHalf accumulator; + // receive() maps "no complete frame" to Ok(None). stream .set_nonblocking(true) .map_err(|e| BridgeError::Transport(format!("failed to set non-blocking: {}", e)))?; @@ -57,7 +76,7 @@ impl TcpBridge { })?; Ok(Self { - reader: Mutex::new(BufReader::new(reader_stream)), + reader: Mutex::new(ReadHalf::new(reader_stream)), writer: Mutex::new(BufWriter::new(stream)), local_addr, }) @@ -91,7 +110,7 @@ impl TcpBridge { })?; Ok(Self { - reader: Mutex::new(BufReader::new(reader_stream)), + reader: Mutex::new(ReadHalf::new(reader_stream)), writer: Mutex::new(BufWriter::new(stream)), local_addr, }) @@ -117,7 +136,7 @@ impl TcpBridge { })?; Ok(Self { - reader: Mutex::new(BufReader::new(reader_stream)), + reader: Mutex::new(ReadHalf::new(reader_stream)), writer: Mutex::new(BufWriter::new(stream)), local_addr, }) @@ -137,15 +156,16 @@ impl SimBridge for TcpBridge { .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?; // Toggle to blocking for reliable startup message read. // The client sends StartupMessage immediately after handshake validation, - // so this read should complete quickly. + // so this read should complete quickly. Runs before the first receive(), + // so the frame accumulator is empty and read_framed sees a fresh stream. reader - .get_mut() + .stream .set_nonblocking(false) .map_err(BridgeError::Io)?; - let result = read_framed(reader.get_mut()); + let result = read_framed(&mut reader.stream); // Restore non-blocking for the tick loop reader - .get_mut() + .stream .set_nonblocking(true) .map_err(BridgeError::Io)?; match result? { @@ -204,11 +224,19 @@ impl SimBridge for TcpBridge { .lock() .map_err(|e| BridgeError::MutexPoisoned(format!("reader: {}", e)))?; - match read_framed(reader.get_mut()) { + let ReadHalf { stream, accum } = &mut *reader; + match accum.poll_frame(stream) { Ok(Some(payload)) => decode_inbound(&payload).map(Some), Ok(None) => Err(BridgeError::Disconnected), - // Non-blocking socket: no data available this tick — not an error. + // Non-blocking socket: no complete frame yet — not an error. Any + // partial frame stays in the accumulator for the next tick. Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => Ok(None), + // EOF mid-frame: the peer died leaving a truncated stream. No data + // can ever complete this frame — escalate like the clean-EOF case + // above instead of surfacing an Io error every tick forever. + Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => { + Err(BridgeError::Disconnected) + } Err(e) => Err(BridgeError::Io(e)), } } diff --git a/server/tests/bridge_tcp.rs b/server/tests/bridge_tcp.rs index d278925ad..eafa07e01 100644 --- a/server/tests/bridge_tcp.rs +++ b/server/tests/bridge_tcp.rs @@ -190,3 +190,210 @@ fn tcp_bridge_eof_returns_error() { server_handle.join().expect("server thread panicked"); } + +/// T-1045 regression: EOF *mid-frame* (peer dies after a partial write) must +/// escalate to `Disconnected` like clean EOF — not surface an Io error every +/// tick forever against a frame that can never complete. +#[test] +fn tcp_bridge_eof_mid_frame_escalates_to_disconnected() { + let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); + let server_addr = listener.local_addr().expect("failed to get local address"); + + let server_handle = thread::spawn(move || { + let bridge = TcpBridge::accept_on(listener).expect("failed to accept"); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + match bridge.receive() { + Ok(None) => { + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for mid-frame EOF" + ); + thread::sleep(std::time::Duration::from_millis(1)); + } + Ok(Some(_)) => panic!("expected Disconnected, got a message"), + Err(settled_reach_server::bridge::BridgeError::Disconnected) => break, + Err(e) => panic!("expected Disconnected, got: {}", e), + } + } + }); + + // Client: write 2 of the 4 length-prefix bytes, then die. + use std::io::Write; + let mut stream = TcpStream::connect(server_addr).expect("failed to connect"); + stream.write_all(&[0x00, 0x00]).expect("partial write failed"); + stream.flush().expect("flush failed"); + thread::sleep(std::time::Duration::from_millis(50)); + drop(stream); + + server_handle.join().expect("server thread panicked"); +} + +/// Build a raw frame (4-byte BE length prefix + payload) for split-write tests. +fn raw_frame(payload: &[u8]) -> Vec { + let mut frame = (payload.len() as u32).to_be_bytes().to_vec(); + frame.extend_from_slice(payload); + frame +} + +/// Poll the non-blocking bridge until `count` input batches arrive. +fn collect_input_batches(bridge: &TcpBridge, count: usize) -> Vec> { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let mut batches = Vec::new(); + while batches.len() < count { + match bridge.receive() { + Ok(Some(Inbound::Inputs(inputs))) if !inputs.is_empty() => batches.push(inputs), + Ok(_) => { + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for {} input batches (got {})", + count, + batches.len() + ); + thread::sleep(std::time::Duration::from_millis(1)); + } + Err(e) => panic!("failed to receive inputs: {}", e), + } + } + batches +} + +/// T-1045 regression: a frame delivered in two TCP writes (split at +/// `split_point(frame_len)`) must decode, and the NEXT frame must also decode +/// — i.e. a partial read mid-frame must not desync the stream. The server +/// polls receive() during the gap, so it observes WouldBlock mid-frame. +fn assert_split_frame_does_not_desync(split_point: impl Fn(usize) -> usize) { + use std::io::Write; + + let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); + let server_addr = listener.local_addr().expect("failed to get local address"); + + let server_handle = thread::spawn(move || { + let bridge = TcpBridge::accept_on(listener).expect("failed to accept"); + collect_input_batches(&bridge, 2) + }); + + let mut stream = TcpStream::connect(server_addr).expect("failed to connect"); + + let inputs1 = vec![PlayerInput { + tick: 10, + action: PlayerAction::MoveNorth, + }]; + let payload1 = rmp_serde::to_vec_named(&inputs1).expect("failed to serialize"); + let frame1 = raw_frame(&payload1); + let split = split_point(frame1.len()); + assert!(split > 0 && split < frame1.len(), "split must be mid-frame"); + + // First half, then a gap long enough for the server to poll mid-frame. + stream + .write_all(&frame1[..split]) + .expect("write first half"); + stream.flush().expect("flush first half"); + thread::sleep(std::time::Duration::from_millis(50)); + stream + .write_all(&frame1[split..]) + .expect("write second half"); + stream.flush().expect("flush second half"); + + // A second frame in a single write — decodes only if the stream is in sync. + let inputs2 = vec![PlayerInput { + tick: 11, + action: PlayerAction::MoveSouth, + }]; + let payload2 = rmp_serde::to_vec_named(&inputs2).expect("failed to serialize"); + stream + .write_all(&raw_frame(&payload2)) + .expect("write second frame"); + stream.flush().expect("flush second frame"); + + let batches = server_handle.join().expect("server thread panicked"); + assert_eq!(batches[0].len(), 1); + assert_eq!(batches[0][0].tick, 10); + assert!(matches!(batches[0][0].action, PlayerAction::MoveNorth)); + assert_eq!(batches[1].len(), 1); + assert_eq!(batches[1][0].tick, 11); + assert!(matches!(batches[1][0].action, PlayerAction::MoveSouth)); +} + +#[test] +fn frame_split_inside_prefix_does_not_desync() { + // Split inside the 4-byte length prefix. + assert_split_frame_does_not_desync(|_| 2); +} + +#[test] +fn frame_split_inside_payload_does_not_desync() { + // Split midway through the payload (prefix is 4 bytes). + assert_split_frame_does_not_desync(|frame_len| 4 + (frame_len - 4) / 2); +} + +/// T-1045 drain: one receive_bridge_inputs run (= one 50 ms tick) must drain +/// every complete frame buffered on the socket, not one frame per tick. +#[test] +fn single_tick_drains_all_ready_inbound_frames() { + use bevy_ecs::system::RunSystemOnce; + use settled_reach_server::atlas::cascade::CascadeLayer; + use settled_reach_server::atlas::layer_proxy::AtlasLayerRequest; + use settled_reach_server::bridge::{ + receive_bridge_inputs, AtlasRequestBuffer, BridgeResource, HandshakeState, ServerRunning, + }; + use settled_reach_server::simulation::input::InputQueue; + + let listener = TcpListener::bind("127.0.0.1:0").expect("failed to bind"); + let server_addr = listener.local_addr().expect("failed to get local address"); + + // Client: three frames back-to-back in one tick window — two input + // batches plus one atlas request (the D-225 demux path). Returns the + // stream so it stays open until assertions complete (no EOF race). + let client_handle = thread::spawn(move || { + let mut stream = TcpStream::connect(server_addr).expect("failed to connect"); + for tick in [20u64, 21] { + let inputs = vec![PlayerInput { + tick, + action: PlayerAction::MoveNorth, + }]; + let payload = rmp_serde::to_vec_named(&inputs).expect("failed to serialize"); + write_framed(&mut stream, &payload).expect("write input frame"); + } + let req = AtlasLayerRequest { + body_id: "GJ1c".into(), + up_to: CascadeLayer::Topography, + }; + let payload = rmp_serde::to_vec_named(&req).expect("failed to serialize"); + write_framed(&mut stream, &payload).expect("write atlas frame"); + stream + }); + + let bridge = TcpBridge::accept_on(listener).expect("failed to accept"); + let _stream = client_handle.join().expect("client thread panicked"); + // Writes are flushed and joined; small grace period for loopback delivery. + thread::sleep(std::time::Duration::from_millis(100)); + + let mut world = bevy_ecs::world::World::new(); + world.insert_resource(BridgeResource::new(bridge)); + world.init_resource::(); + world.init_resource::(); + world.insert_resource(HandshakeState::Complete); + world.init_resource::(); + world.init_resource::(); + + world + .run_system_once(receive_bridge_inputs) + .expect("receive_bridge_inputs failed to run"); + + assert_eq!( + world.resource::().len(), + 2, + "both input batches must drain in a single tick" + ); + assert_eq!( + world.resource::().0.len(), + 1, + "the atlas request must drain in the same tick" + ); + assert!( + world.resource::().0, + "draining must not shut the server down" + ); +}