B5 — tolerate capped-response peer disconnects in the read-transport harness

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",

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"
 }

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<ResponseWriteOutcome> {
+    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,

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]