diff --git a/docs/tally/compatibility/compatibility-matrix.json b/docs/tally/compatibility/compatibility-matrix.json index 2c76339..22853eb 100644 --- a/docs/tally/compatibility/compatibility-matrix.json +++ b/docs/tally/compatibility/compatibility-matrix.json @@ -1,7 +1,7 @@ { "schema_version": 1, "bridge_commit_sha": "be1c20cc3fd66fa1ece196505c69f26e555e4b8e", - "compatibility_surface_sha256": "af8d0ba7d679001bd44a954ce90f9b164188a146af44efe55d74c472d29d18e5", + "compatibility_surface_sha256": "e1e2d20a7e13f22c6f1369d333a71f47c83f3b877483daa86218e7ea723b9e45", "claims": [ { "claim_id": "erp9-6-6-3-windows-education-xml-one-company", diff --git a/docs/tally/compatibility/compatibility-surface.json b/docs/tally/compatibility/compatibility-surface.json index 3bee61a..47ffef9 100644 --- a/docs/tally/compatibility/compatibility-surface.json +++ b/docs/tally/compatibility/compatibility-surface.json @@ -211,7 +211,7 @@ }, { "path": "src-tauri/crates/tally-protocol-simulator/src/server.rs", - "sha256": "7c6830ae96c030e2ad499d1f9daf6078cc5777a78c411213f4a659b8b4c81147" + "sha256": "69b826947cc6393a104f32cedd306f05fb7d871b7d696a6ee8e9cc0f078a84d8" }, { "path": "src-tauri/crates/tally-protocol-simulator/tests/protocol_simulator.rs", @@ -427,8 +427,8 @@ }, { "path": "tools/bridge-tally-read-transport/src/lib.rs", - "sha256": "ef1d33e90da527faa9735469ea5040c9258fbab9e4c761f2473d79e7b3dbd0c4" + "sha256": "a27f294ee15e407b69fdfc73609e8708ac0509b6e6a8872daef5451fde61a8db" } ], - "manifest_sha256": "af8d0ba7d679001bd44a954ce90f9b164188a146af44efe55d74c472d29d18e5" + "manifest_sha256": "e1e2d20a7e13f22c6f1369d333a71f47c83f3b877483daa86218e7ea723b9e45" } diff --git a/src-tauri/crates/tally-protocol-simulator/src/server.rs b/src-tauri/crates/tally-protocol-simulator/src/server.rs index c06ef7e..f9cdc49 100644 --- a/src-tauri/crates/tally-protocol-simulator/src/server.rs +++ b/src-tauri/crates/tally-protocol-simulator/src/server.rs @@ -34,6 +34,17 @@ pub struct ObservedRequest { pub request_body_sha256: String, pub request_processed: bool, pub cancelled: bool, + /// The client stopped consuming a response after the request was processed. + /// This is observable in bounded-response transport tests and is not a + /// simulator failure. + pub client_stopped_reading_response: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ResponseWriteOutcome { + Complete, + Cancelled, + ClientStoppedReading, } pub struct Simulator { @@ -250,6 +261,7 @@ fn serve_request( request_body_sha256: hex::encode(Sha256::digest(request_body)), request_processed: false, cancelled: cancelled.load(Ordering::Acquire), + client_stopped_reading_response: false, }; if observed.cancelled { return Ok(observed); @@ -262,9 +274,10 @@ fn serve_request( observed.request_processed = true; stream.write_all(headers.as_bytes())?; stream.flush()?; - observed.cancelled = - write_framed_body(&mut stream, &body, plan.framing, None, cancelled)?; - finish_complete_response(&mut stream, observed.cancelled)?; + record_response_write_outcome( + &mut observed, + write_complete_response(&mut stream, &body, plan.framing, None, cancelled)?, + ); } Delivery::SlowHeaders(delay) => { if sleep_cancellable(delay, cancelled) { @@ -274,9 +287,10 @@ fn serve_request( observed.request_processed = true; stream.write_all(headers.as_bytes())?; stream.flush()?; - observed.cancelled = - write_framed_body(&mut stream, &body, plan.framing, None, cancelled)?; - finish_complete_response(&mut stream, observed.cancelled)?; + record_response_write_outcome( + &mut observed, + write_complete_response(&mut stream, &body, plan.framing, None, cancelled)?, + ); } Delivery::SlowBody { chunk_bytes, delay } => { if chunk_bytes == 0 { @@ -288,14 +302,16 @@ fn serve_request( observed.request_processed = true; stream.write_all(headers.as_bytes())?; stream.flush()?; - observed.cancelled = write_framed_body( - &mut stream, - &body, - plan.framing, - Some((chunk_bytes, delay)), - cancelled, - )?; - finish_complete_response(&mut stream, observed.cancelled)?; + record_response_write_outcome( + &mut observed, + write_complete_response( + &mut stream, + &body, + plan.framing, + Some((chunk_bytes, delay)), + cancelled, + )?, + ); } Delivery::ResetBeforeBody => { stream.write_all(headers.as_bytes())?; @@ -313,17 +329,59 @@ fn serve_request( Ok(observed) } -fn finish_complete_response(stream: &mut TcpStream, cancelled: bool) -> io::Result<()> { - if !cancelled { - stream.flush()?; - stream.shutdown(Shutdown::Write)?; - // Give Windows' loopback stack time to deliver the FIN and buffered body before the - // server thread drops the socket. Immediate drop is observably flaky under parallel CI. - thread::sleep(Duration::from_millis(5)); +fn record_response_write_outcome(observed: &mut ObservedRequest, outcome: ResponseWriteOutcome) { + match outcome { + ResponseWriteOutcome::Complete => {} + ResponseWriteOutcome::Cancelled => observed.cancelled = true, + ResponseWriteOutcome::ClientStoppedReading => { + observed.client_stopped_reading_response = true; + } } +} + +fn write_complete_response( + stream: &mut TcpStream, + body: &[u8], + framing: ResponseFraming, + slow_delivery: Option<(usize, Duration)>, + cancelled: &AtomicBool, +) -> io::Result { + match write_framed_body(stream, body, framing, slow_delivery, cancelled) { + Ok(true) => Ok(ResponseWriteOutcome::Cancelled), + Ok(false) => match finish_complete_response(stream) { + Ok(()) => Ok(ResponseWriteOutcome::Complete), + Err(error) if client_stopped_reading(&error) => { + Ok(ResponseWriteOutcome::ClientStoppedReading) + } + Err(error) => Err(error), + }, + Err(error) if client_stopped_reading(&error) => { + Ok(ResponseWriteOutcome::ClientStoppedReading) + } + Err(error) => Err(error), + } +} + +fn finish_complete_response(stream: &mut TcpStream) -> io::Result<()> { + stream.flush()?; + stream.shutdown(Shutdown::Write)?; + // Give Windows' loopback stack time to deliver the FIN and buffered body before the + // server thread drops the socket. Immediate drop is observably flaky under parallel CI. + thread::sleep(Duration::from_millis(5)); Ok(()) } +fn client_stopped_reading(error: &io::Error) -> bool { + matches!( + error.kind(), + io::ErrorKind::BrokenPipe + | io::ErrorKind::ConnectionAborted + | io::ErrorKind::ConnectionReset + | io::ErrorKind::TimedOut + | io::ErrorKind::WouldBlock + ) +} + fn read_request( stream: &mut TcpStream, cancelled: &AtomicBool, diff --git a/tools/bridge-tally-read-transport/src/lib.rs b/tools/bridge-tally-read-transport/src/lib.rs index 418438d..bf32581 100644 --- a/tools/bridge-tally-read-transport/src/lib.rs +++ b/tools/bridge-tally-read-transport/src/lib.rs @@ -363,6 +363,8 @@ mod native_outstandings_tests { assert_eq!(error.safe_code(), "response_size_limit_exceeded"); let observed = simulator.finish().unwrap(); assert_eq!(observed.request_body_sha256, candidate().request_sha256()); + assert!(observed.request_processed); + assert!(!observed.cancelled); } #[tokio::test]