From 5bfb864f0f4d3dcc45629bc200843925c1817967 Mon Sep 17 00:00:00 2001 From: zuub-don Date: Thu, 13 Aug 2026 05:40:31 -0700 Subject: [PATCH 1/4] fix(websocket): confirm a timeout against the clock before reporting one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A socket reports `SO_RCVTIMEO` expiry and a spurious `EAGAIN` as the same error, and platforms disagree about when each happens. The deadline stream trusted the errno: outside a cancellable session, one `WouldBlock` ended the session as a timeout without asking whether the deadline had passed. The observation then said the peer ran out of time when it had seconds left. The retry that already existed for cancellable sessions now applies to every session, guarded by the clock rather than by the errno, with a millisecond pause so a socket reporting readiness it does not have waits for the deadline instead of spinning at it. A non-blocking socket reports exactly what a spurious wakeup reports, so the regression test reproduces the condition deterministically instead of waiting for a platform to produce it. Without the fix it fails in 0.00s — the same instant-timeout signature as the macOS CI failure that prompted this. The seeded oracle test now asserts the terminal cause before the exit code, so a future failure names the deadline that fired instead of only reporting that one did. --- crates/kahea-exec/src/websocket.rs | 81 ++++++++++++++++++++++++------ 1 file changed, 67 insertions(+), 14 deletions(-) diff --git a/crates/kahea-exec/src/websocket.rs b/crates/kahea-exec/src/websocket.rs index a0f395f..e182098 100644 --- a/crates/kahea-exec/src/websocket.rs +++ b/crates/kahea-exec/src/websocket.rs @@ -2440,6 +2440,25 @@ impl DeadlineTcpStream { }) } + /// Confirm a reported timeout against the clock before treating it as one. + /// + /// `SO_RCVTIMEO` expiry and a spurious `EAGAIN` arrive as the same error, and the platforms do + /// not agree on when each happens. Ending a session on the errno alone reports a deadline that + /// has not passed: the observation says the peer ran out of time when it had seconds left. Ask + /// the clock instead, and only stop when the deadline is genuinely gone. + /// + /// The pause bounds the retry: a socket that reports readiness it does not have would otherwise + /// spin until the deadline rather than wait for it. + fn await_deadline(&self) -> io::Result<()> { + let before = Instant::now(); + self.remaining()?; + if before.elapsed() < Duration::from_millis(1) { + std::thread::sleep(Duration::from_millis(1)); + self.remaining()?; + } + Ok(()) + } + fn note_activity(&self) -> io::Result<()> { self.deadline .lock() @@ -2460,10 +2479,8 @@ impl Read for DeadlineTcpStream { } return Ok(read); } - Err(error) - if error.kind() == io::ErrorKind::TimedOut && self.cancellation.is_some() => - { - self.remaining()?; + Err(error) if error.kind() == io::ErrorKind::TimedOut => { + self.await_deadline()?; } Err(error) => return Err(error), } @@ -2482,10 +2499,8 @@ impl Write for DeadlineTcpStream { } return Ok(written); } - Err(error) - if error.kind() == io::ErrorKind::TimedOut && self.cancellation.is_some() => - { - self.remaining()?; + Err(error) if error.kind() == io::ErrorKind::TimedOut => { + self.await_deadline()?; } Err(error) => return Err(error), } @@ -2497,10 +2512,8 @@ impl Write for DeadlineTcpStream { self.stream.set_write_timeout(Some(self.remaining()?))?; match self.stream.flush().map_err(normalize_timeout) { Ok(()) => return Ok(()), - Err(error) - if error.kind() == io::ErrorKind::TimedOut && self.cancellation.is_some() => - { - self.remaining()?; + Err(error) if error.kind() == io::ErrorKind::TimedOut => { + self.await_deadline()?; } Err(error) => return Err(error), } @@ -3145,6 +3158,40 @@ mod tests { remove_temporary_store(&root); } + /// A socket that reports "would block" before the deadline must not end the session. + /// + /// A non-blocking socket reports exactly what a spurious wakeup reports, so this reproduces the + /// condition deterministically rather than waiting for a platform to produce it: the read is + /// answered late, and the stream has to wait for it instead of calling the deadline elapsed. + #[test] + fn a_would_block_before_the_deadline_is_not_a_timeout() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let server = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + thread::sleep(Duration::from_millis(150)); + stream.write_all(b"late").unwrap(); + stream.flush().unwrap(); + }); + let stream = TcpStream::connect(address).unwrap(); + // Every read now answers WouldBlock immediately until the data lands. + stream.set_nonblocking(true).unwrap(); + let deadline = Arc::new(Mutex::new(DeadlineState::fixed( + Instant::now() + Duration::from_secs(5), + WebSocketTerminalCause::ActionTimeout, + ))); + let mut deadline_stream = DeadlineTcpStream::new(stream, deadline, None); + let started = Instant::now(); + let mut buffer = [0_u8; 4]; + let read = deadline_stream.read(&mut buffer).unwrap(); + assert_eq!(&buffer[..read], b"late"); + assert!( + started.elapsed() >= Duration::from_millis(100), + "the read returned before the data could have arrived" + ); + server.join().unwrap(); + } + #[test] fn controlled_oracle_is_seeded_scripted_bounded_and_tls_capable() { let mut scenario = generate_websocket_scenario(0x51_0c_e7); @@ -3164,11 +3211,17 @@ mod tests { else { panic!("seeded oracle must produce a terminal observation") }; - assert_eq!(observation.exit, 0); + // Assert the cause before the exit code: the cause names which deadline or failure ended + // the session, and the exit code only says that one did. A bare `exit 0 != 3` tells an + // investigator nothing about where to look. assert_eq!( observation.terminal_cause, - WebSocketTerminalCause::Completed + WebSocketTerminalCause::Completed, + "seeded oracle ended early: exit={} counters={:?}", + observation.exit, + observation.counters ); + assert_eq!(observation.exit, 0); assert_eq!(observation.negotiated_subprotocol, scenario.subprotocol); let oracle_observation = oracle.wait().unwrap(); assert_eq!(oracle_observation.seed, scenario.seed); From 1d4133fea24204918e71e8cbddd4ed3870b038fb Mon Sep 17 00:00:00 2001 From: zuub-don Date: Thu, 13 Aug 2026 06:02:24 -0700 Subject: [PATCH 2/4] fix(websocket): a signal is not a cancelled session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `EINTR` says a signal arrived while a thread was blocked. It says nothing about the session: no deadline passed, no peer acted, nothing was decided. The executor mapped it straight to `cancelled`, so a stray signal ended a healthy session with exit 3 and a terminal cause claiming the caller had asked for it. This is the third reading of the same failure and the first that fits all of it. Exit 3 is `SessionTerminal::error`, which covers every error cause, not the timeouts alone — reading it as a deadline is what sent the previous two fixes after budgets and clocks. A signal explains what those could not: failures that land in milliseconds with the budget untouched, on whichever test happens to be reading, only on the noisiest runner. The deadline stream now retries a bare interruption, and the terminal cause is `cancelled` only when cancellation was actually requested. The two meanings were already conflated because `remaining` reports its own cancellation as `Interrupted`, so the flag decides, not the errno. --- crates/kahea-exec/src/websocket.rs | 66 +++++++++++++++++++++++++++++- 1 file changed, 65 insertions(+), 1 deletion(-) diff --git a/crates/kahea-exec/src/websocket.rs b/crates/kahea-exec/src/websocket.rs index e182098..3270f16 100644 --- a/crates/kahea-exec/src/websocket.rs +++ b/crates/kahea-exec/src/websocket.rs @@ -220,6 +220,7 @@ impl Write for AccountedStream { pub struct WebSocketConnection { pub metadata: WebSocketHandshakeMetadata, socket: Socket, + cancellation: Option>, deadline: Arc>, started: Instant, total_deadline: Instant, @@ -243,6 +244,12 @@ impl WebSocketCancellation { } impl WebSocketConnection { + fn cancellation_requested(&self) -> bool { + self.cancellation + .as_ref() + .is_some_and(|cancelled| cancelled.load(Ordering::Acquire)) + } + pub fn is_open(&self) -> bool { self.socket.can_read() || self.socket.can_write() } @@ -504,6 +511,7 @@ fn connect_websocket_resolving_cancellable( connect_deadline, WebSocketTerminalCause::ConnectTimeout, ))); + let cancellation_handle = cancellation.clone(); let stream = DeadlineTcpStream::new(stream, Arc::clone(&deadline_handle), cancellation); let config = websocket_config(plan)?; let stream = match websocket_stream(stream, &target, tls) { @@ -589,6 +597,7 @@ fn connect_websocket_resolving_cancellable( WebSocketConnection { metadata, socket, + cancellation: cancellation_handle, deadline: deadline_handle, started, total_deadline, @@ -1391,7 +1400,14 @@ fn socket_error( } match error { WebSocketError::Io(error) if error.kind() == io::ErrorKind::Interrupted => { - SessionTerminal::error(WebSocketTerminalCause::Cancelled) + // The deadline stream retries a bare signal, so an interruption reaching here came from + // the cancellation path. Report it as cancelled only if cancellation was in fact + // requested; otherwise a stray signal would be reported as a decision the caller made. + if interruption_is_cancellation(connection.cancellation_requested()) { + SessionTerminal::error(WebSocketTerminalCause::Cancelled) + } else { + SessionTerminal::error(WebSocketTerminalCause::IoFailure) + } } WebSocketError::Io(error) if matches!( @@ -2459,6 +2475,12 @@ impl DeadlineTcpStream { Ok(()) } + fn cancellation_requested(&self) -> bool { + self.cancellation + .as_ref() + .is_some_and(|cancelled| cancelled.load(Ordering::Acquire)) + } + fn note_activity(&self) -> io::Result<()> { self.deadline .lock() @@ -2482,6 +2504,12 @@ impl Read for DeadlineTcpStream { Err(error) if error.kind() == io::ErrorKind::TimedOut => { self.await_deadline()?; } + Err(error) + if error.kind() == io::ErrorKind::Interrupted + && !interruption_is_cancellation(self.cancellation_requested()) => + { + self.remaining()?; + } Err(error) => return Err(error), } } @@ -2502,6 +2530,12 @@ impl Write for DeadlineTcpStream { Err(error) if error.kind() == io::ErrorKind::TimedOut => { self.await_deadline()?; } + Err(error) + if error.kind() == io::ErrorKind::Interrupted + && !interruption_is_cancellation(self.cancellation_requested()) => + { + self.remaining()?; + } Err(error) => return Err(error), } } @@ -2515,12 +2549,28 @@ impl Write for DeadlineTcpStream { Err(error) if error.kind() == io::ErrorKind::TimedOut => { self.await_deadline()?; } + Err(error) + if error.kind() == io::ErrorKind::Interrupted + && !interruption_is_cancellation(self.cancellation_requested()) => + { + self.remaining()?; + } Err(error) => return Err(error), } } } } +/// Whether an interrupted syscall means this session was cancelled. +/// +/// `EINTR` says a signal arrived while the thread was blocked. It says nothing about the session: +/// no deadline passed, no peer acted, nothing was decided. Only the caller's cancellation flag can +/// make an interruption meaningful, and the flag is the thing to ask — the errno is the same either +/// way, which is why this is a function and not a match arm. +const fn interruption_is_cancellation(cancellation_requested: bool) -> bool { + cancellation_requested +} + fn normalize_timeout(error: io::Error) -> io::Error { if error.kind() == io::ErrorKind::WouldBlock { io::Error::new(io::ErrorKind::TimedOut, "WebSocket deadline elapsed") @@ -3158,6 +3208,20 @@ mod tests { remove_temporary_store(&root); } + /// A signal is not a decision, and only the caller's flag can make an interruption one. + /// + /// `EINTR` and a cancelled session arrive as the same `ErrorKind::Interrupted`, and the code + /// used to report both as `cancelled` — so a stray signal on a macOS runner ended a healthy + /// session with a terminal cause claiming the caller had asked for it. + #[test] + fn an_interrupted_syscall_is_only_cancellation_when_cancellation_was_requested() { + assert!(interruption_is_cancellation(true)); + assert!( + !interruption_is_cancellation(false), + "a bare signal must not be reported as a cancelled session" + ); + } + /// A socket that reports "would block" before the deadline must not end the session. /// /// A non-blocking socket reports exactly what a spurious wakeup reports, so this reproduces the From 7472fa126f110ff2f501fcf0478d9eada1fc3dbc Mon Sep 17 00:00:00 2001 From: zuub-don Date: Thu, 13 Aug 2026 06:48:06 -0700 Subject: [PATCH 3/4] test(websocket): read the whole payload rather than one read's worth A single read may legally return fewer bytes than the peer sent, so the deadline test could fail on a short read for a reason it is not about. Review finding from #41. --- crates/kahea-exec/src/websocket.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/kahea-exec/src/websocket.rs b/crates/kahea-exec/src/websocket.rs index 3270f16..42da025 100644 --- a/crates/kahea-exec/src/websocket.rs +++ b/crates/kahea-exec/src/websocket.rs @@ -3247,8 +3247,10 @@ mod tests { let mut deadline_stream = DeadlineTcpStream::new(stream, deadline, None); let started = Instant::now(); let mut buffer = [0_u8; 4]; - let read = deadline_stream.read(&mut buffer).unwrap(); - assert_eq!(&buffer[..read], b"late"); + // read_exact, not read: a single read may legally return fewer bytes than the peer sent, and + // a short read would fail this test for a reason it is not about. + deadline_stream.read_exact(&mut buffer).unwrap(); + assert_eq!(&buffer, b"late"); assert!( started.elapsed() >= Duration::from_millis(100), "the read returned before the data could have arrived" From 9bd7614ff3b98739955f76266b606adeec755a30 Mon Sep 17 00:00:00 2001 From: zuub-don Date: Thu, 13 Aug 2026 06:55:28 -0700 Subject: [PATCH 4/4] fix(websocket): an accepted close survives a peer that vanishes The diagnostic added earlier named the failure the moment it recurred: IoFailure after a session whose counters show the whole script ran. A peer that sends an acceptable close and then drops the socket resets the connection, and that reset lands on our acknowledging write. #35 made a rejected close outrank that reset and left an accepted one reporting IoFailure, calling it unchanged behaviour. It was the same mistake with the sign flipped: a session that met every expectation and received a close its plan accepts was reported as broken, on exactly the runs where the reset won the race. Whether our reply landed belongs in the transcript, not in the verdict. `close_precedence` now takes the acknowledgement and ignores it, so the signature still says out loud that it was considered. --- crates/kahea-exec/src/websocket.rs | 77 +++++++++++++++++++++++------- 1 file changed, 59 insertions(+), 18 deletions(-) diff --git a/crates/kahea-exec/src/websocket.rs b/crates/kahea-exec/src/websocket.rs index 42da025..2384da7 100644 --- a/crates/kahea-exec/src/websocket.rs +++ b/crates/kahea-exec/src/websocket.rs @@ -1045,11 +1045,6 @@ fn read_expectation( exit: 1, close: Some(close), }), - ClosePrecedence::NotAcknowledged => Err(socket_error( - connection, - acknowledged.expect_err("only reached when the flush failed"), - timeout_cause, - )), }; } Message::Frame(frame) => { @@ -1360,22 +1355,24 @@ fn sync_wire_counters( enum ClosePrecedence { Accepted, Rejected, - NotAcknowledged, } /// What to report when the peer's close frame has been read and acknowledging it may have failed. /// -/// The verdict is settled the moment the frame is read; the acknowledgement is courtesy. A peer that +/// The close frame decides, and only the close frame. Acknowledging it is courtesy: a peer that /// closes with our bytes still unread resets the connection, and that reset surfaces on the -/// acknowledging write rather than on the read that already told us what happened. Reporting the I/O -/// error would replace the diagnosis the operator needs — an unacceptable close code — with the -/// failure to reply to it, and would do so only on the runs where the reset won the race. An I/O -/// failure is reported only when there is no verdict of its own to report. -const fn close_precedence(matched: bool, acknowledged: bool) -> ClosePrecedence { - match (matched, acknowledged) { - (false, _) => ClosePrecedence::Rejected, - (true, true) => ClosePrecedence::Accepted, - (true, false) => ClosePrecedence::NotAcknowledged, +/// acknowledging write rather than on the read that already told us what happened. +/// +/// `acknowledged` is a parameter and then ignored on purpose. An earlier version let a failed +/// acknowledgement turn an accepted close into an I/O failure, which reported a session that met +/// every expectation and received a close its plan accepted as though it had broken — on exactly the +/// runs where the peer's reset won the race. Whether our reply landed is recorded in the transcript, +/// where it belongs; it is not a verdict about the session. +const fn close_precedence(matched: bool, _acknowledged: bool) -> ClosePrecedence { + if matched { + ClosePrecedence::Accepted + } else { + ClosePrecedence::Rejected } } @@ -3208,6 +3205,50 @@ mod tests { remove_temporary_store(&root); } + /// A peer that resets right after an acceptable close still ran an acceptable session. + /// + /// The server sends its close and drops the socket without draining, which is what macOS turns + /// into a reset on the acknowledging write. Every expectation was met and the close code is one + /// the plan accepts, so the session completed. + #[test] + fn an_accepted_close_survives_a_peer_that_vanishes_before_the_reply() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let mut plan = ws_plan(&listener); + plan.actions = vec![WebSocketAction::ExpectClose { + codes: vec![1000], + reason: None, + timeout_ms: None, + }]; + let plan = plan.seal().unwrap(); + let server = thread::spawn(move || { + let mut socket = tungstenite::accept(accept_test_connection(&listener)).unwrap(); + socket + .write(Message::Frame(Frame::close(Some(CloseFrame { + code: CloseCode::Normal, + reason: "".into(), + })))) + .unwrap(); + socket.flush().unwrap(); + drop(socket); + }); + let (root, store) = store(); + let WebSocketConnectResult::Observation(observation) = + execute_websocket(&plan, &options(&plan), &store).unwrap() + else { + panic!("an accepted close must produce an observation") + }; + assert_eq!( + observation.terminal_cause, + WebSocketTerminalCause::Completed, + "exit={}", + observation.exit + ); + assert_eq!(observation.exit, 0); + server.join().unwrap(); + drop(store); + remove_temporary_store(&root); + } + /// A signal is not a decision, and only the caller's flag can make an interruption one. /// /// `EINTR` and a cancelled session arrive as the same `ErrorKind::Interrupted`, and the code @@ -4081,8 +4122,8 @@ mod tests { assert_eq!(close_precedence(true, true), ClosePrecedence::Accepted); assert_eq!( close_precedence(true, false), - ClosePrecedence::NotAcknowledged, - "with nothing to report about the close itself, the I/O failure is the result" + ClosePrecedence::Accepted, + "a peer that vanishes after an acceptable close still ran an acceptable session" ); for acknowledged in [true, false] { assert_eq!(