From 0400612cb294a08a00d3d627d48e51df733bacf7 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Mon, 3 Aug 2026 09:51:22 -0500 Subject: [PATCH 1/6] feat(connector)!: pass frame arrival time into Sequence::step Connect-time bandwidth measurement needs to know when bytes arrived, and nothing in the sans-I/O layer could tell it. The connector answered a Bandwidth Measure Stop with a fabricated interval because the real one was not observable from where the response is built. Introduce MonotonicInstant, a millisecond counter with no epoch and no platform dependency, and make it a required parameter of Sequence::step. The I/O drivers already know when a read completed, so Framed records the arrival time of each read and hands it to the state machine. Callers that genuinely have no frame to time (the no-input path) use step_no_input. With arrival times available, the connector measures for real: a Bandwidth Measure Start opens a window, Payload messages accumulate their byte counts, and Stop reports the elapsed time between its own arrival and the Start's. A Stop with no preceding Start still gets its response, because the server blocks without one, but reports a zero interval rather than inventing a figure. BREAKING CHANGE: Sequence::step takes a MonotonicInstant. Implementors must add the parameter; callers driving a sequence by hand must supply an arrival time or switch to step_no_input. --- .../src/channel_connection.rs | 9 +- crates/ironrdp-acceptor/src/connection.rs | 17 +- crates/ironrdp-acceptor/src/finalization.rs | 11 +- crates/ironrdp-async/src/framed.rs | 38 +++- crates/ironrdp-blocking/src/connector.rs | 2 +- crates/ironrdp-blocking/src/framed.rs | 23 ++- crates/ironrdp-client/src/rdp.rs | 6 +- .../src/channel_connection.rs | 10 +- crates/ironrdp-connector/src/connection.rs | 180 ++++++++++++------ .../src/connection_activation.rs | 11 +- .../src/connection_finalization.rs | 10 +- crates/ironrdp-connector/src/lib.rs | 42 +++- .../ironrdp-connector/src/license_exchange.rs | 11 +- .../tests/connector/autodetect.rs | 140 +++++++++++++- .../tests/server/acceptor.rs | 33 +++- .../tests/session/connection_activation.rs | 33 ++-- crates/ironrdp-web/src/session.rs | 6 +- ffi/src/connector/activation.rs | 6 +- ffi/src/connector/mod.rs | 2 +- 19 files changed, 472 insertions(+), 118 deletions(-) diff --git a/crates/ironrdp-acceptor/src/channel_connection.rs b/crates/ironrdp-acceptor/src/channel_connection.rs index 66a066823a..0d3b4a58ef 100644 --- a/crates/ironrdp-acceptor/src/channel_connection.rs +++ b/crates/ironrdp-acceptor/src/channel_connection.rs @@ -1,7 +1,7 @@ use std::collections::HashSet; use ironrdp_connector::{ - ConnectorError, ConnectorErrorExt as _, ConnectorResult, Sequence, State, Written, reason_err, + ConnectorError, ConnectorErrorExt as _, ConnectorResult, MonotonicInstant, Sequence, State, Written, reason_err, }; use ironrdp_core::WriteBuf; use ironrdp_pdu::mcs; @@ -72,7 +72,12 @@ impl Sequence for ChannelConnectionSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: MonotonicInstant, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match core::mem::take(&mut self.state) { ChannelConnectionState::WaitErectDomainRequest => { let erect_domain_request = ironrdp_core::decode::>(input) diff --git a/crates/ironrdp-acceptor/src/connection.rs b/crates/ironrdp-acceptor/src/connection.rs index 7217bfab03..a5c2cc635e 100644 --- a/crates/ironrdp-acceptor/src/connection.rs +++ b/crates/ironrdp-acceptor/src/connection.rs @@ -2,8 +2,8 @@ use core::any::TypeId; use core::mem; use ironrdp_connector::{ - ConnectorError, ConnectorErrorExt as _, ConnectorResult, DesktopSize, Sequence, State, Written, encode_x224_packet, - general_err, reason_err, + ConnectorError, ConnectorErrorExt as _, ConnectorResult, DesktopSize, MonotonicInstant, Sequence, State, Written, + encode_x224_packet, general_err, reason_err, }; use ironrdp_core::{WriteBuf, decode}; use ironrdp_pdu as pdu; @@ -287,7 +287,8 @@ impl Acceptor { /// Panics if state is not [AcceptorState::SecurityUpgrade]. pub fn mark_security_upgrade_as_done(&mut self) { assert!(self.reached_security_upgrade().is_some()); - self.step(&[], &mut WriteBuf::new()).expect("transition to next state"); + self.step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + .expect("transition to next state"); debug_assert!(self.reached_security_upgrade().is_none()); } @@ -300,7 +301,9 @@ impl Acceptor { /// Panics if state is not [AcceptorState::Credssp]. pub fn mark_credssp_as_done(&mut self) { assert!(self.should_perform_credssp()); - let res = self.step(&[], &mut WriteBuf::new()).expect("transition to next state"); + let res = self + .step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + .expect("transition to next state"); debug_assert!(!self.should_perform_credssp()); assert_eq!(res, Written::Nothing); } @@ -458,7 +461,7 @@ impl Sequence for Acceptor { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step(&mut self, input: &[u8], received_at: MonotonicInstant, output: &mut WriteBuf) -> ConnectorResult { let prev_state = mem::take(&mut self.state); let (written, next_state) = match prev_state { @@ -724,7 +727,7 @@ impl Sequence for Acceptor { channels, mut connection, } => { - let written = connection.step(input, output)?; + let written = connection.step(input, received_at, output)?; let state = if connection.is_done() { AcceptorState::RdpSecurityCommencement { protocol, @@ -964,7 +967,7 @@ impl Sequence for Acceptor { channels, client_capabilities, } => { - let written = finalization.step(input, output)?; + let written = finalization.step(input, received_at, output)?; let state = if finalization.is_done() { AcceptorState::Accepted { diff --git a/crates/ironrdp-acceptor/src/finalization.rs b/crates/ironrdp-acceptor/src/finalization.rs index 9961e87ce7..6f8663a068 100644 --- a/crates/ironrdp-acceptor/src/finalization.rs +++ b/crates/ironrdp-acceptor/src/finalization.rs @@ -1,4 +1,6 @@ -use ironrdp_connector::{ConnectorError, ConnectorErrorExt as _, ConnectorResult, Sequence, State, Written}; +use ironrdp_connector::{ + ConnectorError, ConnectorErrorExt as _, ConnectorResult, MonotonicInstant, Sequence, State, Written, +}; use ironrdp_core::WriteBuf; use ironrdp_pdu::rdp; use ironrdp_pdu::x224::X224; @@ -78,7 +80,12 @@ impl Sequence for FinalizationSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: MonotonicInstant, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match core::mem::take(&mut self.state) { FinalizationState::WaitSynchronize => { let synchronize = decode_share_control(input); diff --git a/crates/ironrdp-async/src/framed.rs b/crates/ironrdp-async/src/framed.rs index eff516456c..9b6ad21f05 100644 --- a/crates/ironrdp-async/src/framed.rs +++ b/crates/ironrdp-async/src/framed.rs @@ -1,6 +1,7 @@ use std::io; use bytes::{Bytes, BytesMut}; +use ironrdp_connector::MonotonicInstant; use ironrdp_connector::{ConnectorResult, Sequence, Written}; use ironrdp_core::WriteBuf; use ironrdp_pdu::PduHint; @@ -56,12 +57,23 @@ pub trait StreamWrapper: Sized { pub struct Framed { stream: S, buf: BytesMut, + /// When the most recent socket read completed. + /// + /// A PDU served entirely from `buf` arrived at the socket read that filled it, + /// not at the moment the caller happened to drain it, so this is the honest + /// arrival time for anything `read_by_hint` returns. + last_read_at: MonotonicInstant, } impl Framed { pub fn peek(&self) -> &[u8] { &self.buf } + + /// When the bytes currently buffered last arrived from the socket. + pub fn last_read_at(&self) -> MonotonicInstant { + self.last_read_at + } } impl Framed @@ -76,6 +88,7 @@ where Self { stream: S::from_inner(stream), buf: leftover, + last_read_at: MonotonicInstant::ZERO, } } @@ -197,7 +210,9 @@ where /// `tokio::select!` statement and some other branch /// completes first, then it is guaranteed that no data was read. async fn read(&mut self) -> io::Result { - self.stream.read(&mut self.buf).await + let len = self.stream.read(&mut self.buf).await?; + self.last_read_at = monotonic_now(); + Ok(len) } } @@ -261,7 +276,7 @@ where trace!(length = pdu.len(), "PDU received"); - sequence.step(&pdu, buf) + sequence.step(&pdu, framed.last_read_at(), buf) } else { sequence.step_no_input(buf) } @@ -287,3 +302,22 @@ where Ok(()) } + +/// Reads the driver-owned monotonic clock. +/// +/// The epoch is the first call; only differences are meaningful. +/// +/// `std::time::Instant::now` panics on `wasm32-unknown-unknown`, and this crate is +/// reached from `ironrdp-web` through `ironrdp-futures`, so the browser build gets +/// [`MonotonicInstant::ZERO`] instead. Intervals computed from it are zero, which +/// the connector already treats as "not measured" rather than as a measurement. +#[cfg(not(target_arch = "wasm32"))] +fn monotonic_now() -> MonotonicInstant { + static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(std::time::Instant::now); + MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)) +} + +#[cfg(target_arch = "wasm32")] +fn monotonic_now() -> MonotonicInstant { + MonotonicInstant::ZERO +} diff --git a/crates/ironrdp-blocking/src/connector.rs b/crates/ironrdp-blocking/src/connector.rs index 8b24701f89..0e4e3ad585 100644 --- a/crates/ironrdp-blocking/src/connector.rs +++ b/crates/ironrdp-blocking/src/connector.rs @@ -230,7 +230,7 @@ where trace!(length = pdu.len(), "PDU received"); - connector.step(&pdu, buf)? + connector.step(&pdu, framed.last_read_at(), buf)? } else { connector.step_no_input(buf)? }; diff --git a/crates/ironrdp-blocking/src/framed.rs b/crates/ironrdp-blocking/src/framed.rs index a2964d4481..21b16ae845 100644 --- a/crates/ironrdp-blocking/src/framed.rs +++ b/crates/ironrdp-blocking/src/framed.rs @@ -1,12 +1,16 @@ use std::io::{self, Read, Write}; use bytes::{Bytes, BytesMut}; +use ironrdp_connector::MonotonicInstant; use ironrdp_pdu::PduHint; use tracing::debug; pub struct Framed { stream: S, buf: BytesMut, + /// When the most recent socket read completed. A PDU served from `buf` + /// arrived at the read that filled it, not when the caller drained it. + last_read_at: MonotonicInstant, } impl Framed { @@ -15,7 +19,11 @@ impl Framed { } pub fn new_with_leftover(stream: S, leftover: BytesMut) -> Self { - Self { stream, buf: leftover } + Self { + stream, + buf: leftover, + last_read_at: MonotonicInstant::ZERO, + } } pub fn into_inner(self) -> (S, BytesMut) { @@ -36,6 +44,11 @@ impl Framed { (&mut self.stream, &mut self.buf) } + /// When the bytes currently buffered last arrived from the socket. + pub fn last_read_at(&self) -> MonotonicInstant { + self.last_read_at + } + pub fn peek(&self) -> &[u8] { &self.buf } @@ -118,6 +131,7 @@ where let mut read_bytes = [0u8; 1024]; let len = self.stream.read(&mut read_bytes)?; + self.last_read_at = monotonic_now(); self.buf.extend_from_slice(&read_bytes[..len]); Ok(len) @@ -133,3 +147,10 @@ where self.stream.write_all(buf) } } + +/// Reads the driver-owned monotonic clock. Epoch is the first call; only +/// differences are meaningful. +fn monotonic_now() -> MonotonicInstant { + static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(std::time::Instant::now); + MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)) +} diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs index bd441342da..9818f942bf 100644 --- a/crates/ironrdp-client/src/rdp.rs +++ b/crates/ironrdp-client/src/rdp.rs @@ -1896,7 +1896,11 @@ where debug_assert!(connector.next_pdu_hint().is_some()); buf.clear(); - let written = connector.step(x224_connection_response.as_bytes(), &mut buf)?; + let written = connector.step( + x224_connection_response.as_bytes(), + ironrdp_connector::MonotonicInstant::ZERO, + &mut buf, + )?; debug_assert!(written.is_nothing()); let should_upgrade = ironrdp_tokio::skip_connect_begin(connector); diff --git a/crates/ironrdp-connector/src/channel_connection.rs b/crates/ironrdp-connector/src/channel_connection.rs index 199946aed4..cf0a1e2668 100644 --- a/crates/ironrdp-connector/src/channel_connection.rs +++ b/crates/ironrdp-connector/src/channel_connection.rs @@ -7,7 +7,8 @@ use ironrdp_pdu::{PduHint, mcs}; use tracing::{debug, warn}; use crate::{ - ConnectorError, ConnectorErrorExt as _, ConnectorResult, Sequence, State, Written, general_err, reason_err, + ConnectorError, ConnectorErrorExt as _, ConnectorResult, MonotonicInstant, Sequence, State, Written, general_err, + reason_err, }; #[derive(Default, Debug)] @@ -94,7 +95,12 @@ impl Sequence for ChannelConnectionSequence { } } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: MonotonicInstant, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { ChannelConnectionState::Consumed => { return Err(general_err!( diff --git a/crates/ironrdp-connector/src/connection.rs b/crates/ironrdp-connector/src/connection.rs index 82d5e2b0a0..962e43f466 100644 --- a/crates/ironrdp-connector/src/connection.rs +++ b/crates/ironrdp-connector/src/connection.rs @@ -18,7 +18,7 @@ use crate::connection_activation::{ }; use crate::license_exchange::{LicenseExchangeSequence, NoopLicenseCache}; use crate::{ - Config, ConnectorError, ConnectorErrorExt as _, ConnectorErrorKind, ConnectorResult, DesktopSize, + Config, ConnectorError, ConnectorErrorExt as _, ConnectorErrorKind, ConnectorResult, DesktopSize, MonotonicInstant, NegotiationFailure, Sequence, State, Written, encode_x224_packet, general_err, reason_err, }; @@ -27,6 +27,14 @@ use crate::{ /// transport protocol: reliable + lossy UDP). const MAX_MULTITRANSPORT_REQUESTS: usize = 2; +/// Reported as `timeDelta` when a connect-time bandwidth window cannot be timed. +/// +/// One millisecond rather than zero, because a server computing +/// `byteCount * 8 / timeDelta` divides by it. The result understates a fast link, +/// which is the safe direction: the figure is an informational QoS hint, and the +/// server proceeds on receipt either way. +const UNMEASURABLE_INTERVAL_MS: u32 = 1; + /// Outcome of a single multitransport bootstrapping request, passed to /// [`ClientConnector::complete_multitransport()`]. /// @@ -237,7 +245,7 @@ impl State for ClientConnectorState { #[expect( clippy::partial_pub_fields, - reason = "server response flags are negotiated internally and must not expand the public connector construction API" + reason = "server response flags are negotiated internally and must not expand the public connector construction API; the connect-time bandwidth accumulators are likewise internal to the measurement, and exposing them would let a caller break the Start/Payload/Stop invariant" )] #[derive(Debug)] pub struct ClientConnector { @@ -260,6 +268,14 @@ pub struct ClientConnector { /// /// Set via [`ClientConnector::with_auto_reconnect_cookie`]. pub auto_reconnect_cookie: Option, + /// Start of the in-flight connect-time bandwidth measurement window. + /// + /// Set when the server's Bandwidth Measure Start arrives and cleared when the + /// matching Stop is answered. `None` means no window is open, in which case the + /// Stop is answered with [`UNMEASURABLE_INTERVAL_MS`] rather than a measurement. + connect_time_bw_started_at: Option, + /// Bytes seen in the open window, accumulated across Payload messages. + connect_time_bw_bytes: u32, } impl ClientConnector { @@ -273,6 +289,8 @@ impl ClientConnector { response_flags: nego::ResponseFlags::empty(), server_multitransport_flags: None, auto_reconnect_cookie: None, + connect_time_bw_started_at: None, + connect_time_bw_bytes: 0, } } @@ -476,7 +494,8 @@ impl ClientConnector { /// Panics if state is not [ClientConnectorState::EnhancedSecurityUpgrade]. pub fn mark_security_upgrade_as_done(&mut self) { assert!(self.should_perform_security_upgrade()); - self.step(&[], &mut WriteBuf::new()).expect("transition to next state"); + self.step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + .expect("transition to next state"); debug_assert!(!self.should_perform_security_upgrade()); } @@ -491,7 +510,9 @@ impl ClientConnector { /// Panics if state is not [ClientConnectorState::Credssp]. pub fn mark_credssp_as_done(&mut self) { assert!(self.should_perform_credssp()); - let res = self.step(&[], &mut WriteBuf::new()).expect("transition to next state"); + let res = self + .step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + .expect("transition to next state"); debug_assert!(!self.should_perform_credssp()); assert_eq!(res, Written::Nothing); } @@ -685,6 +706,91 @@ impl ClientConnector { Written::from_size(total_written) } } + + fn respond_to_connect_time_autodetect( + &mut self, + request: rdp::autodetect::AutoDetectRequest, + received_at: MonotonicInstant, + message_channel_id: u16, + user_channel_id: u16, + output: &mut WriteBuf, + ) -> ConnectorResult { + use ironrdp_pdu::rdp::autodetect::{ + AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu, BW_RESULTS_CONNECT_TIME, + }; + + match request { + AutoDetectRequest::RttRequest { sequence_number, .. } => { + let response = AutoDetectRspPdu::new(AutoDetectResponse::RttResponse { sequence_number }); + let written = encode_send_data_request(user_channel_id, message_channel_id, &response, output)?; + Written::from_size(written) + } + // Start opens the measurement window ([MS-RDPBCGR] 2.2.14.1.2). No reply is + // due; we only note when it arrived. + AutoDetectRequest::BandwidthMeasureStart { .. } => { + self.connect_time_bw_started_at = Some(received_at); + self.connect_time_bw_bytes = 0; + Ok(Written::Nothing) + } + // Payload carries the bytes whose transfer is being timed ([MS-RDPBCGR] + // 2.2.14.1.3). No reply is due; accumulate so Stop can report the total. + AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => { + let len = u32::try_from(payload.len()).unwrap_or(u32::MAX); + self.connect_time_bw_bytes = self.connect_time_bw_bytes.saturating_add(len); + Ok(Written::Nothing) + } + // A connect-time Bandwidth Measure Stop ([MS-RDPBCGR] 2.2.14.1.4) warrants a + // Bandwidth Measure Results reply ([MS-RDPBCGR] 2.2.14.2.2). This reply must + // be sent: FreeRDP-based servers (for example GNOME Remote Desktop) block in + // their AWAIT_BW_RESULT state until they receive it and never proceed to + // licensing without it, so omitting it stalls the whole connection. + // + // The interval is measured from the Start that opened the window to the + // arrival of this Stop, using the times the I/O driver observed rather than + // any clock this sequence could read for itself. + // + // Two cases yield no interval to report. Either no Start was seen, or the + // whole Start/Payload/Stop sequence arrived in a single socket read, which + // is the common case on a fast link because the payload is small enough to + // coalesce. Both are unmeasurable at read granularity rather than + // instantaneous, and the distinction matters on the wire: `timeDelta` of 0 + // divides out to an unbounded bandwidth for a server that computes + // `byteCount * 8 / timeDelta`. Report the floor instead, which is the + // slowest rate consistent with what was observed. + AutoDetectRequest::BandwidthMeasureStop { + sequence_number, + payload, + .. + } => { + let stop_bytes = payload + .as_ref() + .map_or(0, |p| u32::try_from(p.len()).unwrap_or(u32::MAX)); + let byte_count = self.connect_time_bw_bytes.saturating_add(stop_bytes); + + let measured_ms = self + .connect_time_bw_started_at + .map(|started_at| { + u32::try_from(received_at.duration_since(started_at).as_millis()).unwrap_or(u32::MAX) + }) + .unwrap_or(0); + let time_delta_ms = measured_ms.max(UNMEASURABLE_INTERVAL_MS); + + self.connect_time_bw_started_at = None; + self.connect_time_bw_bytes = 0; + + let response = AutoDetectRspPdu::new(AutoDetectResponse::BandwidthMeasureResults { + sequence_number, + response_type: BW_RESULTS_CONNECT_TIME, + time_delta_ms, + byte_count, + }); + let written = encode_send_data_request(user_channel_id, message_channel_id, &response, output)?; + Written::from_size(written) + } + // The Network Characteristics Result is informational; nothing to send. + _ => Ok(Written::Nothing), + } + } } /// Build an Initiate Multitransport Response carrying `hr_response`. @@ -708,9 +814,10 @@ fn advance_licensing_exchange( user_channel_id: u16, message_channel_id: Option, input: &[u8], + received_at: MonotonicInstant, output: &mut WriteBuf, ) -> ConnectorResult<(Written, ClientConnectorState)> { - let written = license_exchange.step(input, output)?; + let written = license_exchange.step(input, received_at, output)?; let next_state = if license_exchange.state.is_terminal() { ClientConnectorState::MultitransportBootstrapping { @@ -773,7 +880,7 @@ impl Sequence for ClientConnector { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step(&mut self, input: &[u8], received_at: MonotonicInstant, output: &mut WriteBuf) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { // Invalid state ClientConnectorState::Consumed => { @@ -954,7 +1061,7 @@ impl Sequence for ClientConnector { mut channel_connection, } => { debug!("Channel Connection"); - let written = channel_connection.step(input, output)?; + let written = channel_connection.step(input, received_at, output)?; let next_state = if let ChannelConnectionState::AllJoined { user_channel_id } = channel_connection.state { @@ -1035,8 +1142,9 @@ impl Sequence for ClientConnector { if let Some((message_channel_id, data)) = message_channel_pdu { if let Ok(autodetect) = decode::(&data.user_data) { - let written = respond_to_connect_time_autodetect( + let written = self.respond_to_connect_time_autodetect( autodetect.request, + received_at, message_channel_id, user_channel_id, output, @@ -1085,6 +1193,7 @@ impl Sequence for ClientConnector { user_channel_id, self.message_channel_id, input, + received_at, output, )? } else { @@ -1116,6 +1225,7 @@ impl Sequence for ClientConnector { user_channel_id, self.message_channel_id, input, + received_at, output, )? } @@ -1184,7 +1294,7 @@ impl Sequence for ClientConnector { // exchange with the PDU intact. let mut connection_activation = ConnectionActivationSequence::new(self.config.clone(), io_channel_id, user_channel_id); - let written = connection_activation.step(input, output)?; + let written = connection_activation.step(input, received_at, output)?; ( written, @@ -1221,7 +1331,7 @@ impl Sequence for ClientConnector { ClientConnectorState::CapabilitiesExchange { mut connection_activation, } => { - let written = connection_activation.step(input, output)?; + let written = connection_activation.step(input, received_at, output)?; match connection_activation.connection_activation_state() { ConnectionActivationState::ConnectionFinalization { .. } => ( written, @@ -1245,7 +1355,7 @@ impl Sequence for ClientConnector { ClientConnectorState::ConnectionFinalization { mut connection_activation, } => { - let written = connection_activation.step(input, output)?; + let written = connection_activation.step(input, received_at, output)?; let next_state = if !connection_activation.connection_activation_state().is_terminal() { ClientConnectorState::ConnectionFinalization { connection_activation } @@ -1328,54 +1438,6 @@ pub fn encode_send_data_request( Ok(written) } -fn respond_to_connect_time_autodetect( - request: rdp::autodetect::AutoDetectRequest, - message_channel_id: u16, - user_channel_id: u16, - output: &mut WriteBuf, -) -> ConnectorResult { - use ironrdp_pdu::rdp::autodetect::{ - AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu, BW_RESULTS_CONNECT_TIME, - }; - - match request { - AutoDetectRequest::RttRequest { sequence_number, .. } => { - let response = AutoDetectRspPdu::new(AutoDetectResponse::RttResponse { sequence_number }); - let written = encode_send_data_request(user_channel_id, message_channel_id, &response, output)?; - Written::from_size(written) - } - // A connect-time Bandwidth Measure Stop ([MS-RDPBCGR] 2.2.14.1.4) warrants a - // Bandwidth Measure Results reply ([MS-RDPBCGR] 2.2.14.2.2). This reply must - // be sent: FreeRDP-based servers (for example GNOME Remote Desktop) block in - // their AWAIT_BW_RESULT state until they receive it and never proceed to - // licensing without it, so omitting it stalls the whole connection. We do not - // run a stateful connect-time measurement, so we report the payload the - // server handed us over a nominal interval; the figure is an informational - // QoS hint and the server proceeds on receipt. A precise measurement (timing - // the Start/Payload/Stop window) can refine the reported bandwidth later. - AutoDetectRequest::BandwidthMeasureStop { - sequence_number, - payload, - .. - } => { - let byte_count = payload - .as_ref() - .map_or(0, |p| u32::try_from(p.len()).unwrap_or(u32::MAX)); - let response = AutoDetectRspPdu::new(AutoDetectResponse::BandwidthMeasureResults { - sequence_number, - response_type: BW_RESULTS_CONNECT_TIME, - time_delta_ms: 1, - byte_count, - }); - let written = encode_send_data_request(user_channel_id, message_channel_id, &response, output)?; - Written::from_size(written) - } - // Bandwidth Measure Start and Payload carry no client reply, and the Network - // Characteristics Result is informational; nothing to send for those. - _ => Ok(Written::Nothing), - } -} - #[expect(single_use_lifetimes)] // anonymous lifetimes in `impl Trait` are unstable fn create_gcc_blocks<'a>( config: &Config, diff --git a/crates/ironrdp-connector/src/connection_activation.rs b/crates/ironrdp-connector/src/connection_activation.rs index 9e606fff87..0cb8c33a98 100644 --- a/crates/ironrdp-connector/src/connection_activation.rs +++ b/crates/ironrdp-connector/src/connection_activation.rs @@ -8,7 +8,7 @@ use tracing::{debug, warn}; use crate::{ Config, ConnectionFinalizationSequence, ConnectorError, ConnectorErrorExt as _, ConnectorResult, DesktopSize, - Sequence, State, Written, general_err, reason_err, + MonotonicInstant, Sequence, State, Written, general_err, reason_err, }; /// Represents the Capability Exchange and Connection Finalization phases @@ -125,7 +125,12 @@ impl Sequence for ConnectionActivationSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut ironrdp_core::WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + received_at: MonotonicInstant, + output: &mut ironrdp_core::WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { ConnectionActivationState::Consumed | ConnectionActivationState::Finalized { .. } => { return Err(general_err!( @@ -323,7 +328,7 @@ impl Sequence for ConnectionActivationSequence { } => { debug!("Connection Finalization"); - let written = connection_finalization.step(input, output)?; + let written = connection_finalization.step(input, received_at, output)?; let next_state = if !connection_finalization.state.is_terminal() { ConnectionActivationState::ConnectionFinalization { diff --git a/crates/ironrdp-connector/src/connection_finalization.rs b/crates/ironrdp-connector/src/connection_finalization.rs index 06c53ec9da..e88f6915d0 100644 --- a/crates/ironrdp-connector/src/connection_finalization.rs +++ b/crates/ironrdp-connector/src/connection_finalization.rs @@ -8,7 +8,8 @@ use ironrdp_pdu::rdp::{finalization_messages, server_error_info}; use tracing::{debug, warn}; use crate::{ - ConnectorError, ConnectorErrorExt as _, ConnectorResult, Sequence, State, Written, general_err, reason_err, + ConnectorError, ConnectorErrorExt as _, ConnectorResult, MonotonicInstant, Sequence, State, Written, general_err, + reason_err, }; #[derive(Default, Debug, Copy, Clone)] @@ -87,7 +88,12 @@ impl Sequence for ConnectionFinalizationSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: MonotonicInstant, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { ConnectionFinalizationState::Consumed => { return Err(general_err!( diff --git a/crates/ironrdp-connector/src/lib.rs b/crates/ironrdp-connector/src/lib.rs index 585dafb787..b86e1ad2c6 100644 --- a/crates/ironrdp-connector/src/lib.rs +++ b/crates/ironrdp-connector/src/lib.rs @@ -355,15 +355,53 @@ impl Written { } } +/// A point on a monotonic millisecond clock owned by the I/O driver. +/// +/// The epoch is arbitrary and carries no meaning; only differences between two +/// instants do. The clock deliberately lives outside the sans-I/O sequences: +/// `std::time::Instant::now` panics on `wasm32-unknown-unknown`, which +/// `ironrdp-connector` compiles for, and a sequence reading a clock itself would +/// measure how quickly it drained an already-filled buffer rather than how long +/// the bytes took to arrive. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct MonotonicInstant(u64); + +impl MonotonicInstant { + /// The clock's origin. + /// + /// Drivers that do not measure intervals pass this. Every duration computed + /// from it is then zero, which is a visibly inert value rather than a + /// plausible-looking measurement. + pub const ZERO: Self = Self(0); + + /// Builds an instant from a monotonic millisecond reading. + #[must_use] + pub fn from_millis(milliseconds: u64) -> Self { + Self(milliseconds) + } + + /// The time elapsed since `earlier`, saturating at zero if the clock went + /// backwards or the arguments were transposed. + #[must_use] + pub fn duration_since(self, earlier: Self) -> core::time::Duration { + core::time::Duration::from_millis(self.0.saturating_sub(earlier.0)) + } +} + pub trait Sequence: Send { fn next_pdu_hint(&self) -> Option<&dyn PduHint>; fn state(&self) -> &dyn State; - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult; + /// Advances the sequence. + /// + /// `received_at` is when `input` arrived on the wire, as observed by the I/O + /// driver. Sequences that do not measure intervals ignore it; drivers that do + /// not measure pass [`MonotonicInstant::ZERO`]. + fn step(&mut self, input: &[u8], received_at: MonotonicInstant, output: &mut WriteBuf) -> ConnectorResult; fn step_no_input(&mut self, output: &mut WriteBuf) -> ConnectorResult { - self.step(&[], output) + self.step(&[], MonotonicInstant::ZERO, output) } } diff --git a/crates/ironrdp-connector/src/license_exchange.rs b/crates/ironrdp-connector/src/license_exchange.rs index 8b77ec76a0..981f7534e2 100644 --- a/crates/ironrdp-connector/src/license_exchange.rs +++ b/crates/ironrdp-connector/src/license_exchange.rs @@ -11,7 +11,9 @@ use rand::RngCore as _; use tracing::{debug, error, info, trace}; use super::{ConnectorError, ConnectorErrorExt as _, custom_err, general_err}; -use crate::{ConnectorResult, ConnectorResultExt as _, Sequence, State, Written, encode_send_data_request}; +use crate::{ + ConnectorResult, ConnectorResultExt as _, MonotonicInstant, Sequence, State, Written, encode_send_data_request, +}; #[derive(Default, Debug)] #[non_exhaustive] @@ -117,7 +119,12 @@ impl Sequence for LicenseExchangeSequence { &self.state } - fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + _received_at: MonotonicInstant, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { LicenseExchangeState::Consumed => { return Err(general_err!( diff --git a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs index ed4e62a618..846e3f8cdb 100644 --- a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs @@ -9,11 +9,12 @@ use std::borrow::Cow; +use ironrdp_connector::MonotonicInstant; use ironrdp_connector::{ClientConnector, ClientConnectorState, Credentials, DesktopSize, Sequence as _, Written}; use ironrdp_core::{WriteBuf, encode_vec}; use ironrdp_pdu::gcc; use ironrdp_pdu::mcs::{McsMessage, SendDataIndication}; -use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest}; +use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu}; use ironrdp_pdu::rdp::capability_sets::MajorPlatformType; use ironrdp_pdu::rdp::headers::{BasicSecurityHeader, BasicSecurityHeaderFlags}; use ironrdp_pdu::rdp::server_license::{ @@ -104,7 +105,7 @@ fn connect_time_autodetect_request_is_answered_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, &mut output).unwrap(); + let written = connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert!(written.size().is_some(), "an RTT request must produce a response frame"); assert!( @@ -127,7 +128,7 @@ fn unrelated_message_channel_pdu_is_ignored_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, &mut output).unwrap(); + let written = connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert_eq!( written, @@ -166,7 +167,7 @@ fn first_licensing_pdu_leaves_autodetect_for_the_licensing_path() { let frame = server_send_data_indication(IO_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert!( matches!( @@ -193,7 +194,7 @@ fn connect_time_bandwidth_measure_stop_is_answered_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, &mut output).unwrap(); + let written = connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert!( written.size().is_some(), @@ -204,3 +205,132 @@ fn connect_time_bandwidth_measure_stop_is_answered_and_phase_continues() { "the connector keeps listening after answering the bandwidth measurement" ); } + +/// Unwrap a Bandwidth Measure Results response and return `(time_delta_ms, byte_count)`. +/// +/// The response frame is X224 > MCS SendDataRequest > Auto-Detect Response PDU. +fn decode_bandwidth_results(output: &WriteBuf) -> (u32, u32) { + let X224(McsMessage::SendDataRequest(send_data)) = ironrdp_core::decode(output.filled()).unwrap() else { + panic!("expected a SendDataRequest in the response frame"); + }; + + let response = ironrdp_core::decode::(&send_data.user_data).unwrap(); + match response.response { + AutoDetectResponse::BandwidthMeasureResults { + time_delta_ms, + byte_count, + .. + } => (time_delta_ms, byte_count), + other => panic!("expected BandwidthMeasureResults, got {other:?}"), + } +} + +/// The reported interval is the time between the Start that opened the window and +/// the Stop that closed it, taken from the arrival times the driver observed. +#[test] +fn connect_time_bandwidth_reports_the_measured_interval_and_total_bytes() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + let start = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_start_connect_time(0x1111))).unwrap(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, start), + MonotonicInstant::from_millis(1_000), + &mut output, + ) + .unwrap(); + + output.clear(); + let stop = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_stop_connect_time( + 0x1111, + vec![0u8; 4096], + ))) + .unwrap(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), + MonotonicInstant::from_millis(1_250), + &mut output, + ) + .unwrap(); + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 250, "interval is Stop arrival minus Start arrival"); + assert_eq!(results.1, 4096, "byte count is what the server sent in the window"); +} + +/// A Stop with no preceding Start has nothing to have measured. It is still +/// answered, because the server blocks without a reply, but the interval reported +/// is the unmeasurable floor rather than an invented figure. +#[test] +fn connect_time_bandwidth_stop_without_start_reports_the_floor() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + let stop = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_stop_connect_time( + 0x2222, + vec![0u8; 512], + ))) + .unwrap(); + let written = connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), + MonotonicInstant::from_millis(9_999), + &mut output, + ) + .unwrap(); + + assert!(written.size().is_some(), "the server still needs its reply"); + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 1, "no window was open, so no interval was measured"); +} + +/// Start, Payload and Stop delivered by one socket read carry the same arrival +/// time, so there is no interval to measure. TCP coalescing makes this the common +/// case on a fast link, since the connect-time payload is small. +/// +/// Reporting `timeDelta` of 0 here would divide out to an unbounded bandwidth for +/// a server computing `byteCount * 8 / timeDelta`, so the floor is reported and the +/// bytes are still counted in full. +#[test] +fn connect_time_bandwidth_coalesced_into_one_read_reports_the_floor() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + // One instant for all three, which is what `Framed` hands down when a single + // read filled the buffer that all three PDUs were then extracted from. + let arrival = MonotonicInstant::from_millis(7_000); + + for request in [ + AutoDetectRequest::bw_start_connect_time(0x3333), + AutoDetectRequest::bw_payload(0x3333, vec![0u8; 1024]), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + arrival, + &mut output, + ) + .unwrap(); + } + + output.clear(); + let stop = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_stop_connect_time( + 0x3333, + vec![0u8; 512], + ))) + .unwrap(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), + arrival, + &mut output, + ) + .unwrap(); + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 1, "a window that arrived in one read cannot be timed"); + assert_eq!(results.1, 1536, "every byte in the window is still counted"); +} diff --git a/crates/ironrdp-testsuite-core/tests/server/acceptor.rs b/crates/ironrdp-testsuite-core/tests/server/acceptor.rs index 0d5d76f79b..9dc7a4cbcd 100644 --- a/crates/ironrdp-testsuite-core/tests/server/acceptor.rs +++ b/crates/ironrdp-testsuite-core/tests/server/acceptor.rs @@ -1,4 +1,5 @@ use ironrdp_acceptor::Acceptor; +use ironrdp_connector::MonotonicInstant; use ironrdp_connector::{DesktopSize, Sequence as _, Written, encode_x224_packet}; use ironrdp_core::{WriteBuf, decode}; use ironrdp_pdu::gcc::ClientMessageChannelData; @@ -37,12 +38,14 @@ fn neg_failure_on_protocol_mismatch() { // Step 1: feed the connection request (HYBRID | HYBRID_EX, no SSL) let request_bytes = encode_connection_request(SecurityProtocol::HYBRID | SecurityProtocol::HYBRID_EX); let mut output = WriteBuf::new(); - let written = acceptor.step(&request_bytes, &mut output).unwrap(); + let written = acceptor + .step(&request_bytes, MonotonicInstant::ZERO, &mut output) + .unwrap(); assert!(matches!(written, Written::Nothing)); // Step 2: acceptor tries to send confirm, finds no common protocol let mut output = WriteBuf::new(); - let result = acceptor.step(&[], &mut output); + let result = acceptor.step(&[], MonotonicInstant::ZERO, &mut output); // Must be an error assert!(result.is_err(), "expected error on protocol mismatch"); @@ -78,10 +81,12 @@ fn neg_success_when_protocols_match() { let request_bytes = encode_connection_request(SecurityProtocol::SSL | SecurityProtocol::HYBRID); let mut output = WriteBuf::new(); - acceptor.step(&request_bytes, &mut output).unwrap(); + acceptor + .step(&request_bytes, MonotonicInstant::ZERO, &mut output) + .unwrap(); let mut output = WriteBuf::new(); - let written = acceptor.step(&[], &mut output).unwrap(); + let written = acceptor.step(&[], MonotonicInstant::ZERO, &mut output).unwrap(); assert!(!matches!(written, Written::Nothing)); let response_bytes = output.filled(); @@ -119,8 +124,12 @@ fn message_channel_advertised_when_client_requests_it() { // Connection request -> confirm -> (TLS upgrade) -> ready for ConnectInitial. let request_bytes = encode_connection_request(SecurityProtocol::SSL); - acceptor.step(&request_bytes, &mut WriteBuf::new()).unwrap(); - acceptor.step(&[], &mut WriteBuf::new()).unwrap(); + acceptor + .step(&request_bytes, MonotonicInstant::ZERO, &mut WriteBuf::new()) + .unwrap(); + acceptor + .step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + .unwrap(); acceptor.mark_security_upgrade_as_done(); // Client GCC with the message channel block and no network channels, so the @@ -132,10 +141,12 @@ fn message_channel_advertised_when_client_requests_it() { let mut initial_buf = WriteBuf::new(); encode_x224_packet(&connect_initial, &mut initial_buf).unwrap(); - acceptor.step(initial_buf.filled(), &mut WriteBuf::new()).unwrap(); + acceptor + .step(initial_buf.filled(), MonotonicInstant::ZERO, &mut WriteBuf::new()) + .unwrap(); let mut output = WriteBuf::new(); - acceptor.step(&[], &mut output).unwrap(); + acceptor.step(&[], MonotonicInstant::ZERO, &mut output).unwrap(); let payload = decode::>>(output.filled()).unwrap().0; let response = decode::(payload.data.as_ref()).unwrap(); @@ -171,10 +182,12 @@ fn neg_failure_hybrid_required() { let request_bytes = encode_connection_request(SecurityProtocol::SSL); let mut output = WriteBuf::new(); - acceptor.step(&request_bytes, &mut output).unwrap(); + acceptor + .step(&request_bytes, MonotonicInstant::ZERO, &mut output) + .unwrap(); let mut output = WriteBuf::new(); - let result = acceptor.step(&[], &mut output); + let result = acceptor.step(&[], MonotonicInstant::ZERO, &mut output); assert!(result.is_err()); let response_bytes = output.filled(); diff --git a/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs b/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs index 33c0da3cb9..737286dc6d 100644 --- a/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs +++ b/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs @@ -1,5 +1,6 @@ use std::borrow::Cow; +use ironrdp_connector::MonotonicInstant; use ironrdp_connector::connection_activation::{ConnectionActivationSequence, ConnectionActivationState}; use ironrdp_connector::{ ClientConnector, ClientConnectorState, Credentials, DesktopSize, MultitransportResult, Sequence as _, Written, @@ -107,7 +108,7 @@ fn demand_active_static_channel_chunk_size(chunk_size: Option) -> usize { let mut output = WriteBuf::new(); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(demand_active)); sequence - .step(&frame, &mut output) + .step(&frame, MonotonicInstant::ZERO, &mut output) .expect("demand active should be accepted"); match sequence.connection_activation_state() { @@ -154,7 +155,7 @@ fn deactivate_all_during_capabilities_exchange_stays_in_same_state() { let frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); let mut output = WriteBuf::new(); - let written = seq.step(&frame, &mut output).unwrap(); + let written = seq.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -177,7 +178,7 @@ fn client_connector_stays_in_capabilities_exchange_on_deactivate_all() { let frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); let mut output = WriteBuf::new(); - let written = connector.step(&frame, &mut output).unwrap(); + let written = connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -207,7 +208,7 @@ fn set_error_info_during_capabilities_exchange_surfaces_the_disconnect_reason() let mut output = WriteBuf::new(); let err = seq - .step(&frame, &mut output) + .step(&frame, MonotonicInstant::ZERO, &mut output) .expect_err("a Set Error Info PDU during capabilities exchange must end the sequence with an error"); let message = err.to_string(); @@ -240,7 +241,7 @@ fn none_error_info_during_capabilities_exchange_is_skipped() { let frame = encode_server_share_control(none_error_info); let mut output = WriteBuf::new(); - let written = seq.step(&frame, &mut output).unwrap(); + let written = seq.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -260,13 +261,17 @@ fn demand_active_after_deactivate_all_transitions_to_connection_finalization() { // First: feed DeactivateAll let deactivate_frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); - let written = seq.step(&deactivate_frame, &mut output).unwrap(); + let written = seq + .step(&deactivate_frame, MonotonicInstant::ZERO, &mut output) + .unwrap(); assert_eq!(written, Written::Nothing); // Then: feed ServerDemandActive let demand_active_frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); - let written = seq.step(&demand_active_frame, &mut output).unwrap(); + let written = seq + .step(&demand_active_frame, MonotonicInstant::ZERO, &mut output) + .unwrap(); assert!(written != Written::Nothing, "should have written ClientConfirmActive"); assert!( @@ -287,7 +292,7 @@ fn demand_active_captures_server_input_flags() { let mut output = WriteBuf::new(); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); - seq.step(&frame, &mut output).unwrap(); + seq.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); match seq.connection_activation_state() { ConnectionActivationState::ConnectionFinalization { input_flags, .. } => { @@ -316,7 +321,7 @@ fn demand_active_without_input_capability_yields_empty_input_flags() { .retain(|c| !matches!(c, CapabilitySet::Input(_))); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(demand_active)); - seq.step(&frame, &mut output).unwrap(); + seq.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); match seq.connection_activation_state() { ConnectionActivationState::ConnectionFinalization { input_flags, .. } => { @@ -394,7 +399,7 @@ fn multitransport_request_is_surfaced_without_waiting_for_another_pdu() { ); let mut output = WriteBuf::new(); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert!( connector.should_perform_multitransport(), @@ -416,7 +421,7 @@ fn responding_returns_to_bootstrapping_for_the_next_request() { &multitransport_request(request_id, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert!(connector.should_perform_multitransport()); assert_eq!(connector.multitransport_request().unwrap().request_id, request_id); @@ -445,7 +450,7 @@ fn third_multitransport_request_is_rejected() { &multitransport_request(request_id, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); connector .complete_multitransport(MultitransportResult::Success, &mut output) .unwrap(); @@ -455,7 +460,7 @@ fn third_multitransport_request_is_rejected() { &multitransport_request(3, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - assert!(connector.step(&frame, &mut output).is_err()); + assert!(connector.step(&frame, MonotonicInstant::ZERO, &mut output).is_err()); } #[test] @@ -466,7 +471,7 @@ fn demand_active_on_the_io_channel_ends_bootstrapping() { let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); let mut output = WriteBuf::new(); - connector.step(&frame, &mut output).unwrap(); + connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); assert!( matches!(connector.state, ClientConnectorState::ConnectionFinalization { .. }), diff --git a/crates/ironrdp-web/src/session.rs b/crates/ironrdp-web/src/session.rs index 1093fe545a..8d4d9da922 100644 --- a/crates/ironrdp-web/src/session.rs +++ b/crates/ironrdp-web/src/session.rs @@ -1866,7 +1866,11 @@ where debug_assert!(connector.next_pdu_hint().is_some()); buf.clear(); - let written = connector.step(x224_connection_response.as_bytes(), &mut buf)?; + let written = connector.step( + x224_connection_response.as_bytes(), + connector::MonotonicInstant::ZERO, + &mut buf, + )?; debug_assert!(written.is_nothing()); let should_upgrade = ironrdp_futures::skip_connect_begin(connector); diff --git a/ffi/src/connector/activation.rs b/ffi/src/connector/activation.rs index bd1db7473d..d84eb9ccf9 100644 --- a/ffi/src/connector/activation.rs +++ b/ffi/src/connector/activation.rs @@ -27,7 +27,11 @@ pub mod ffi { } pub fn step(&mut self, pdu_hint: &[u8], buf: &mut WriteBuf) -> Result, Box> { - let res = self.0.step(pdu_hint, &mut buf.0).map(Written).map(Box::new)?; + let res = self + .0 + .step(pdu_hint, ironrdp::connector::MonotonicInstant::ZERO, &mut buf.0) + .map(Written) + .map(Box::new)?; Ok(res) } diff --git a/ffi/src/connector/mod.rs b/ffi/src/connector/mod.rs index 055ced5737..3384ce300c 100644 --- a/ffi/src/connector/mod.rs +++ b/ffi/src/connector/mod.rs @@ -193,7 +193,7 @@ pub mod ffi { let Some(connector) = self.0.as_mut() else { return Err(ValueConsumedError::for_item("connector").into()); }; - let written = connector.step(input, &mut write_buf.0)?; + let written = connector.step(input, ironrdp::connector::MonotonicInstant::ZERO, &mut write_buf.0)?; Ok(Box::new(Written(written))) } From 6c70ff9238303949691abcc0da7221853a189ae1 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Tue, 4 Aug 2026 12:48:16 -0500 Subject: [PATCH 2/6] fix(connector): distinguish a window that was not timed from one that was The connect-time Bandwidth Measure handler opened a counting window on every Start and reported the accumulated Payload bytes on the Stop, whatever the elapsed time turned out to be. On a driver that reports no arrival times the elapsed time is always zero and falls back to the 1 ms floor, so the reply paired a full byte count with an interval the client never measured. A server computing byteCount * 8 / timeDelta then scales its estimate with however much payload it chose to send, and the wasm32 and FFI drivers are in that position for the whole connection. Model the absence of a reading as the absence of a reading: Sequence::step takes Option, and MonotonicInstant::ZERO is gone. It had served as both the clock's origin and the "did not measure" sentinel, which is what made the two cases indistinguishable at the point where it mattered. A window now opens only when the Start carried a reading, so the two outcomes separate. A driver with no clock opens nothing, accumulates nothing, and answers the Stop with its own payload alone. A window that was timed reports every byte per MS-RDPBCGR 3.2.5.14, including when the elapsed time rounds down to the floor because one socket read delivered the whole exchange: the bytes did arrive inside that millisecond, so the floor bounds a real measurement. Framed::last_read_at becomes Option and the wasm arm of monotonic_now returns None rather than a fixed instant. --- .../src/channel_connection.rs | 2 +- crates/ironrdp-acceptor/src/connection.rs | 11 +- crates/ironrdp-acceptor/src/finalization.rs | 2 +- crates/ironrdp-async/src/framed.rs | 33 +++--- crates/ironrdp-blocking/src/framed.rs | 13 ++- crates/ironrdp-client/src/rdp.rs | 6 +- .../src/channel_connection.rs | 2 +- crates/ironrdp-connector/src/connection.rs | 101 +++++++++++------ .../src/connection_activation.rs | 2 +- .../src/connection_finalization.rs | 2 +- crates/ironrdp-connector/src/lib.rs | 23 ++-- .../ironrdp-connector/src/license_exchange.rs | 2 +- .../tests/connector/autodetect.rs | 103 +++++++++++++++--- .../tests/server/acceptor.rs | 33 ++---- .../tests/session/connection_activation.rs | 33 +++--- crates/ironrdp-web/src/session.rs | 6 +- ffi/src/connector/activation.rs | 6 +- ffi/src/connector/mod.rs | 2 +- 18 files changed, 236 insertions(+), 146 deletions(-) diff --git a/crates/ironrdp-acceptor/src/channel_connection.rs b/crates/ironrdp-acceptor/src/channel_connection.rs index 0d3b4a58ef..e531deb5e4 100644 --- a/crates/ironrdp-acceptor/src/channel_connection.rs +++ b/crates/ironrdp-acceptor/src/channel_connection.rs @@ -75,7 +75,7 @@ impl Sequence for ChannelConnectionSequence { fn step( &mut self, input: &[u8], - _received_at: MonotonicInstant, + _received_at: Option, output: &mut WriteBuf, ) -> ConnectorResult { let (written, next_state) = match core::mem::take(&mut self.state) { diff --git a/crates/ironrdp-acceptor/src/connection.rs b/crates/ironrdp-acceptor/src/connection.rs index a5c2cc635e..286dea151a 100644 --- a/crates/ironrdp-acceptor/src/connection.rs +++ b/crates/ironrdp-acceptor/src/connection.rs @@ -287,7 +287,7 @@ impl Acceptor { /// Panics if state is not [AcceptorState::SecurityUpgrade]. pub fn mark_security_upgrade_as_done(&mut self) { assert!(self.reached_security_upgrade().is_some()); - self.step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + self.step(&[], None, &mut WriteBuf::new()) .expect("transition to next state"); debug_assert!(self.reached_security_upgrade().is_none()); } @@ -302,7 +302,7 @@ impl Acceptor { pub fn mark_credssp_as_done(&mut self) { assert!(self.should_perform_credssp()); let res = self - .step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + .step(&[], None, &mut WriteBuf::new()) .expect("transition to next state"); debug_assert!(!self.should_perform_credssp()); assert_eq!(res, Written::Nothing); @@ -461,7 +461,12 @@ impl Sequence for Acceptor { &self.state } - fn step(&mut self, input: &[u8], received_at: MonotonicInstant, output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let prev_state = mem::take(&mut self.state); let (written, next_state) = match prev_state { diff --git a/crates/ironrdp-acceptor/src/finalization.rs b/crates/ironrdp-acceptor/src/finalization.rs index 6f8663a068..3ad262a9d8 100644 --- a/crates/ironrdp-acceptor/src/finalization.rs +++ b/crates/ironrdp-acceptor/src/finalization.rs @@ -83,7 +83,7 @@ impl Sequence for FinalizationSequence { fn step( &mut self, input: &[u8], - _received_at: MonotonicInstant, + _received_at: Option, output: &mut WriteBuf, ) -> ConnectorResult { let (written, next_state) = match core::mem::take(&mut self.state) { diff --git a/crates/ironrdp-async/src/framed.rs b/crates/ironrdp-async/src/framed.rs index 9b6ad21f05..528ff8a823 100644 --- a/crates/ironrdp-async/src/framed.rs +++ b/crates/ironrdp-async/src/framed.rs @@ -57,12 +57,13 @@ pub trait StreamWrapper: Sized { pub struct Framed { stream: S, buf: BytesMut, - /// When the most recent socket read completed. + /// When the most recent socket read completed, if this build observes time. /// /// A PDU served entirely from `buf` arrived at the socket read that filled it, /// not at the moment the caller happened to drain it, so this is the honest - /// arrival time for anything `read_by_hint` returns. - last_read_at: MonotonicInstant, + /// arrival time for anything `read_by_hint` returns. `None` until the first + /// read, and always `None` on a build with no clock. + last_read_at: Option, } impl Framed { @@ -70,8 +71,9 @@ impl Framed { &self.buf } - /// When the bytes currently buffered last arrived from the socket. - pub fn last_read_at(&self) -> MonotonicInstant { + /// When the bytes currently buffered last arrived from the socket, or `None` + /// if nothing has been read yet or this build cannot observe time. + pub fn last_read_at(&self) -> Option { self.last_read_at } } @@ -88,7 +90,7 @@ where Self { stream: S::from_inner(stream), buf: leftover, - last_read_at: MonotonicInstant::ZERO, + last_read_at: None, } } @@ -303,21 +305,24 @@ where Ok(()) } -/// Reads the driver-owned monotonic clock. +/// Reads the driver-owned monotonic clock, if this build has one. /// /// The epoch is the first call; only differences are meaningful. /// /// `std::time::Instant::now` panics on `wasm32-unknown-unknown`, and this crate is -/// reached from `ironrdp-web` through `ironrdp-futures`, so the browser build gets -/// [`MonotonicInstant::ZERO`] instead. Intervals computed from it are zero, which -/// the connector already treats as "not measured" rather than as a measurement. +/// reached from `ironrdp-web` through `ironrdp-futures`, so the browser build has no +/// clock to read and reports `None` rather than a fabricated reading. Giving it one +/// means plumbing a clock in from the embedder, which has `Performance.now()` +/// available to it. #[cfg(not(target_arch = "wasm32"))] -fn monotonic_now() -> MonotonicInstant { +fn monotonic_now() -> Option { static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(std::time::Instant::now); - MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)) + Some(MonotonicInstant::from_millis( + u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX), + )) } #[cfg(target_arch = "wasm32")] -fn monotonic_now() -> MonotonicInstant { - MonotonicInstant::ZERO +fn monotonic_now() -> Option { + None } diff --git a/crates/ironrdp-blocking/src/framed.rs b/crates/ironrdp-blocking/src/framed.rs index 21b16ae845..04bf1d1729 100644 --- a/crates/ironrdp-blocking/src/framed.rs +++ b/crates/ironrdp-blocking/src/framed.rs @@ -10,7 +10,8 @@ pub struct Framed { buf: BytesMut, /// When the most recent socket read completed. A PDU served from `buf` /// arrived at the read that filled it, not when the caller drained it. - last_read_at: MonotonicInstant, + /// `None` until the first read. + last_read_at: Option, } impl Framed { @@ -22,7 +23,7 @@ impl Framed { Self { stream, buf: leftover, - last_read_at: MonotonicInstant::ZERO, + last_read_at: None, } } @@ -45,7 +46,7 @@ impl Framed { } /// When the bytes currently buffered last arrived from the socket. - pub fn last_read_at(&self) -> MonotonicInstant { + pub fn last_read_at(&self) -> Option { self.last_read_at } @@ -150,7 +151,9 @@ where /// Reads the driver-owned monotonic clock. Epoch is the first call; only /// differences are meaningful. -fn monotonic_now() -> MonotonicInstant { +fn monotonic_now() -> Option { static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(std::time::Instant::now); - MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)) + Some(MonotonicInstant::from_millis( + u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX), + )) } diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs index 9818f942bf..396bfffb80 100644 --- a/crates/ironrdp-client/src/rdp.rs +++ b/crates/ironrdp-client/src/rdp.rs @@ -1896,11 +1896,7 @@ where debug_assert!(connector.next_pdu_hint().is_some()); buf.clear(); - let written = connector.step( - x224_connection_response.as_bytes(), - ironrdp_connector::MonotonicInstant::ZERO, - &mut buf, - )?; + let written = connector.step(x224_connection_response.as_bytes(), None, &mut buf)?; debug_assert!(written.is_nothing()); let should_upgrade = ironrdp_tokio::skip_connect_begin(connector); diff --git a/crates/ironrdp-connector/src/channel_connection.rs b/crates/ironrdp-connector/src/channel_connection.rs index cf0a1e2668..abf55a1d06 100644 --- a/crates/ironrdp-connector/src/channel_connection.rs +++ b/crates/ironrdp-connector/src/channel_connection.rs @@ -98,7 +98,7 @@ impl Sequence for ChannelConnectionSequence { fn step( &mut self, input: &[u8], - _received_at: MonotonicInstant, + _received_at: Option, output: &mut WriteBuf, ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { diff --git a/crates/ironrdp-connector/src/connection.rs b/crates/ironrdp-connector/src/connection.rs index 962e43f466..07f00bdfb3 100644 --- a/crates/ironrdp-connector/src/connection.rs +++ b/crates/ironrdp-connector/src/connection.rs @@ -27,12 +27,12 @@ use crate::{ /// transport protocol: reliable + lossy UDP). const MAX_MULTITRANSPORT_REQUESTS: usize = 2; -/// Reported as `timeDelta` when a connect-time bandwidth window cannot be timed. +/// Reported as `timeDelta` when a connect-time bandwidth window was not timed. /// /// One millisecond rather than zero, because a server computing -/// `byteCount * 8 / timeDelta` divides by it. The result understates a fast link, -/// which is the safe direction: the figure is an informational QoS hint, and the -/// server proceeds on receipt either way. +/// `byteCount * 8 / timeDelta` divides by it ([MS-RDPBCGR] 3.3.5.14). It also +/// floors a window that was timed and elapsed in under a millisecond, where it is +/// a real bound rather than a stand-in. const UNMEASURABLE_INTERVAL_MS: u32 = 1; /// Outcome of a single multitransport bootstrapping request, passed to @@ -270,11 +270,15 @@ pub struct ClientConnector { pub auto_reconnect_cookie: Option, /// Start of the in-flight connect-time bandwidth measurement window. /// - /// Set when the server's Bandwidth Measure Start arrives and cleared when the - /// matching Stop is answered. `None` means no window is open, in which case the - /// Stop is answered with [`UNMEASURABLE_INTERVAL_MS`] rather than a measurement. + /// Set when the server's Bandwidth Measure Start arrives, and only when the + /// driver reported an arrival time for it. Cleared when the matching Stop is + /// answered. `None` therefore means no window is open, whether because no Start + /// was seen or because this driver does not observe time at all. connect_time_bw_started_at: Option, /// Bytes seen in the open window, accumulated across Payload messages. + /// + /// Only accumulated while a window is open, since a total with no interval to + /// divide it by is not a measurement of anything. connect_time_bw_bytes: u32, } @@ -494,7 +498,7 @@ impl ClientConnector { /// Panics if state is not [ClientConnectorState::EnhancedSecurityUpgrade]. pub fn mark_security_upgrade_as_done(&mut self) { assert!(self.should_perform_security_upgrade()); - self.step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + self.step(&[], None, &mut WriteBuf::new()) .expect("transition to next state"); debug_assert!(!self.should_perform_security_upgrade()); } @@ -511,7 +515,7 @@ impl ClientConnector { pub fn mark_credssp_as_done(&mut self) { assert!(self.should_perform_credssp()); let res = self - .step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) + .step(&[], None, &mut WriteBuf::new()) .expect("transition to next state"); debug_assert!(!self.should_perform_credssp()); assert_eq!(res, Written::Nothing); @@ -710,7 +714,7 @@ impl ClientConnector { fn respond_to_connect_time_autodetect( &mut self, request: rdp::autodetect::AutoDetectRequest, - received_at: MonotonicInstant, + received_at: Option, message_channel_id: u16, user_channel_id: u16, output: &mut WriteBuf, @@ -727,16 +731,25 @@ impl ClientConnector { } // Start opens the measurement window ([MS-RDPBCGR] 2.2.14.1.2). No reply is // due; we only note when it arrived. + // + // A driver that reports no arrival time cannot time this window, so it does + // not open one. That keeps the two unmeasurable situations distinct: a + // window that was timed and turned out to be short is still a measurement, + // while a driver with no clock never took one. AutoDetectRequest::BandwidthMeasureStart { .. } => { - self.connect_time_bw_started_at = Some(received_at); + self.connect_time_bw_started_at = received_at; self.connect_time_bw_bytes = 0; Ok(Written::Nothing) } // Payload carries the bytes whose transfer is being timed ([MS-RDPBCGR] // 2.2.14.1.3). No reply is due; accumulate so Stop can report the total. + // With no window open there is nothing for the total to be divided by, so + // there is nothing worth accumulating. AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => { - let len = u32::try_from(payload.len()).unwrap_or(u32::MAX); - self.connect_time_bw_bytes = self.connect_time_bw_bytes.saturating_add(len); + if self.connect_time_bw_started_at.is_some() { + let len = u32::try_from(payload.len()).unwrap_or(u32::MAX); + self.connect_time_bw_bytes = self.connect_time_bw_bytes.saturating_add(len); + } Ok(Written::Nothing) } // A connect-time Bandwidth Measure Stop ([MS-RDPBCGR] 2.2.14.1.4) warrants a @@ -745,18 +758,28 @@ impl ClientConnector { // their AWAIT_BW_RESULT state until they receive it and never proceed to // licensing without it, so omitting it stalls the whole connection. // - // The interval is measured from the Start that opened the window to the - // arrival of this Stop, using the times the I/O driver observed rather than - // any clock this sequence could read for itself. + // [MS-RDPBCGR] 3.2.5.14 has the client increment its Network Characteristics + // Byte Count store on each Payload and on this Stop, then send the store + // together with the elapsed timer. That is what a timed window reports here, + // measured from the Start that opened it using the times the I/O driver + // observed rather than any clock this sequence could read for itself. // - // Two cases yield no interval to report. Either no Start was seen, or the - // whole Start/Payload/Stop sequence arrived in a single socket read, which - // is the common case on a fast link because the payload is small enough to - // coalesce. Both are unmeasurable at read granularity rather than - // instantaneous, and the distinction matters on the wire: `timeDelta` of 0 - // divides out to an unbounded bandwidth for a server that computes - // `byteCount * 8 / timeDelta`. Report the floor instead, which is the - // slowest rate consistent with what was observed. + // The spec does not contemplate a window it could not time, and two arise in + // practice. No Start was seen, so no window exists. Or the driver reports no + // arrival times at all, as the wasm32 and FFI drivers do for the whole + // connection, so no window was opened to accumulate into. + // + // Both still owe the server a reply, and `timeDelta` of 0 divides out to an + // unbounded bandwidth for a server computing `byteCount * 8 / timeDelta` + // (3.2.5.14 again, and [MS-RDPBCGR] 3.3.5.14 for the server side). They + // report the floor against this Stop's payload alone, which is the smallest + // claim that answers the question asked. + // + // A window that was timed reports its full count even when the elapsed time + // rounds down to the floor, which happens when one socket read delivered the + // whole exchange. The bytes did arrive within that millisecond, so the floor + // is a real bound on a real measurement rather than a stand-in for a missing + // one, and the quotient it yields is honest. AutoDetectRequest::BandwidthMeasureStop { sequence_number, payload, @@ -765,15 +788,20 @@ impl ClientConnector { let stop_bytes = payload .as_ref() .map_or(0, |p| u32::try_from(p.len()).unwrap_or(u32::MAX)); - let byte_count = self.connect_time_bw_bytes.saturating_add(stop_bytes); - let measured_ms = self - .connect_time_bw_started_at - .map(|started_at| { - u32::try_from(received_at.duration_since(started_at).as_millis()).unwrap_or(u32::MAX) - }) - .unwrap_or(0); - let time_delta_ms = measured_ms.max(UNMEASURABLE_INTERVAL_MS); + // A window only opens when the Start carried a reading, so the same + // driver stamps this Stop. The `None` arm covers the unopened window. + let (time_delta_ms, byte_count) = match (self.connect_time_bw_started_at, received_at) { + (Some(started_at), Some(stopped_at)) => { + let measured_ms = + u32::try_from(stopped_at.duration_since(started_at).as_millis()).unwrap_or(u32::MAX); + ( + measured_ms.max(UNMEASURABLE_INTERVAL_MS), + self.connect_time_bw_bytes.saturating_add(stop_bytes), + ) + } + _ => (UNMEASURABLE_INTERVAL_MS, stop_bytes), + }; self.connect_time_bw_started_at = None; self.connect_time_bw_bytes = 0; @@ -814,7 +842,7 @@ fn advance_licensing_exchange( user_channel_id: u16, message_channel_id: Option, input: &[u8], - received_at: MonotonicInstant, + received_at: Option, output: &mut WriteBuf, ) -> ConnectorResult<(Written, ClientConnectorState)> { let written = license_exchange.step(input, received_at, output)?; @@ -880,7 +908,12 @@ impl Sequence for ClientConnector { &self.state } - fn step(&mut self, input: &[u8], received_at: MonotonicInstant, output: &mut WriteBuf) -> ConnectorResult { + fn step( + &mut self, + input: &[u8], + received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { // Invalid state ClientConnectorState::Consumed => { diff --git a/crates/ironrdp-connector/src/connection_activation.rs b/crates/ironrdp-connector/src/connection_activation.rs index 0cb8c33a98..672c310e4b 100644 --- a/crates/ironrdp-connector/src/connection_activation.rs +++ b/crates/ironrdp-connector/src/connection_activation.rs @@ -128,7 +128,7 @@ impl Sequence for ConnectionActivationSequence { fn step( &mut self, input: &[u8], - received_at: MonotonicInstant, + received_at: Option, output: &mut ironrdp_core::WriteBuf, ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { diff --git a/crates/ironrdp-connector/src/connection_finalization.rs b/crates/ironrdp-connector/src/connection_finalization.rs index e88f6915d0..672a1494ac 100644 --- a/crates/ironrdp-connector/src/connection_finalization.rs +++ b/crates/ironrdp-connector/src/connection_finalization.rs @@ -91,7 +91,7 @@ impl Sequence for ConnectionFinalizationSequence { fn step( &mut self, input: &[u8], - _received_at: MonotonicInstant, + _received_at: Option, output: &mut WriteBuf, ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { diff --git a/crates/ironrdp-connector/src/lib.rs b/crates/ironrdp-connector/src/lib.rs index b86e1ad2c6..ff06324eb1 100644 --- a/crates/ironrdp-connector/src/lib.rs +++ b/crates/ironrdp-connector/src/lib.rs @@ -367,13 +367,6 @@ impl Written { pub struct MonotonicInstant(u64); impl MonotonicInstant { - /// The clock's origin. - /// - /// Drivers that do not measure intervals pass this. Every duration computed - /// from it is then zero, which is a visibly inert value rather than a - /// plausible-looking measurement. - pub const ZERO: Self = Self(0); - /// Builds an instant from a monotonic millisecond reading. #[must_use] pub fn from_millis(milliseconds: u64) -> Self { @@ -396,12 +389,20 @@ pub trait Sequence: Send { /// Advances the sequence. /// /// `received_at` is when `input` arrived on the wire, as observed by the I/O - /// driver. Sequences that do not measure intervals ignore it; drivers that do - /// not measure pass [`MonotonicInstant::ZERO`]. - fn step(&mut self, input: &[u8], received_at: MonotonicInstant, output: &mut WriteBuf) -> ConnectorResult; + /// driver, or `None` from a driver that does not observe arrival times. The + /// absence of a reading is deliberately not expressible as an instant: a + /// driver that cannot measure has taken no measurement, which is a different + /// thing from one that measured no elapsed time, and only the sequence + /// knows which of the two its reply may be derived from. + fn step( + &mut self, + input: &[u8], + received_at: Option, + output: &mut WriteBuf, + ) -> ConnectorResult; fn step_no_input(&mut self, output: &mut WriteBuf) -> ConnectorResult { - self.step(&[], MonotonicInstant::ZERO, output) + self.step(&[], None, output) } } diff --git a/crates/ironrdp-connector/src/license_exchange.rs b/crates/ironrdp-connector/src/license_exchange.rs index 981f7534e2..4ced01a52f 100644 --- a/crates/ironrdp-connector/src/license_exchange.rs +++ b/crates/ironrdp-connector/src/license_exchange.rs @@ -122,7 +122,7 @@ impl Sequence for LicenseExchangeSequence { fn step( &mut self, input: &[u8], - _received_at: MonotonicInstant, + _received_at: Option, output: &mut WriteBuf, ) -> ConnectorResult { let (written, next_state) = match mem::take(&mut self.state) { diff --git a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs index 846e3f8cdb..9e13ba2436 100644 --- a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs @@ -105,7 +105,7 @@ fn connect_time_autodetect_request_is_answered_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + let written = connector.step(&frame, None, &mut output).unwrap(); assert!(written.size().is_some(), "an RTT request must produce a response frame"); assert!( @@ -128,7 +128,7 @@ fn unrelated_message_channel_pdu_is_ignored_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + let written = connector.step(&frame, None, &mut output).unwrap(); assert_eq!( written, @@ -167,7 +167,7 @@ fn first_licensing_pdu_leaves_autodetect_for_the_licensing_path() { let frame = server_send_data_indication(IO_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); assert!( matches!( @@ -194,7 +194,7 @@ fn connect_time_bandwidth_measure_stop_is_answered_and_phase_continues() { let frame = server_send_data_indication(MESSAGE_CHANNEL_ID, user_data); let mut output = WriteBuf::new(); - let written = connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + let written = connector.step(&frame, None, &mut output).unwrap(); assert!( written.size().is_some(), @@ -236,7 +236,7 @@ fn connect_time_bandwidth_reports_the_measured_interval_and_total_bytes() { connector .step( &server_send_data_indication(MESSAGE_CHANNEL_ID, start), - MonotonicInstant::from_millis(1_000), + Some(MonotonicInstant::from_millis(1_000)), &mut output, ) .unwrap(); @@ -250,7 +250,7 @@ fn connect_time_bandwidth_reports_the_measured_interval_and_total_bytes() { connector .step( &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), - MonotonicInstant::from_millis(1_250), + Some(MonotonicInstant::from_millis(1_250)), &mut output, ) .unwrap(); @@ -276,7 +276,7 @@ fn connect_time_bandwidth_stop_without_start_reports_the_floor() { let written = connector .step( &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), - MonotonicInstant::from_millis(9_999), + Some(MonotonicInstant::from_millis(9_999)), &mut output, ) .unwrap(); @@ -287,12 +287,14 @@ fn connect_time_bandwidth_stop_without_start_reports_the_floor() { } /// Start, Payload and Stop delivered by one socket read carry the same arrival -/// time, so there is no interval to measure. TCP coalescing makes this the common -/// case on a fast link, since the connect-time payload is small. +/// time, so the elapsed time rounds down to nothing. TCP coalescing makes this the +/// common case on a fast link, since the connect-time payload is small. /// -/// Reporting `timeDelta` of 0 here would divide out to an unbounded bandwidth for -/// a server computing `byteCount * 8 / timeDelta`, so the floor is reported and the -/// bytes are still counted in full. +/// Reporting `timeDelta` of 0 would divide out to an unbounded bandwidth for a +/// server computing `byteCount * 8 / timeDelta`, so the floor is reported. Every +/// byte is still counted, per [MS-RDPBCGR] 3.2.5.14: the window was timed, and the +/// bytes really did arrive inside that millisecond, so the floor bounds a real +/// measurement instead of standing in for a missing one. #[test] fn connect_time_bandwidth_coalesced_into_one_read_reports_the_floor() { let mut connector = connect_time_autodetect_connector(); @@ -300,7 +302,7 @@ fn connect_time_bandwidth_coalesced_into_one_read_reports_the_floor() { // One instant for all three, which is what `Framed` hands down when a single // read filled the buffer that all three PDUs were then extracted from. - let arrival = MonotonicInstant::from_millis(7_000); + let arrival = Some(MonotonicInstant::from_millis(7_000)); for request in [ AutoDetectRequest::bw_start_connect_time(0x3333), @@ -331,6 +333,77 @@ fn connect_time_bandwidth_coalesced_into_one_read_reports_the_floor() { .unwrap(); let results = decode_bandwidth_results(&output); - assert_eq!(results.0, 1, "a window that arrived in one read cannot be timed"); - assert_eq!(results.1, 1536, "every byte in the window is still counted"); + assert_eq!(results.0, 1, "a window that arrived in one read floors to 1 ms"); + assert_eq!(results.1, 1536, "every byte in the timed window is still counted"); +} + +/// A driver with no clock reports no arrival times, so no window is ever opened and +/// there is nothing to accumulate into. The wasm32 and FFI drivers are in this +/// position for the whole connection. +/// +/// The server still blocks without a reply, so one is sent, but it claims no more +/// than this Stop's own payload. Counting the Payload messages here would pair a +/// full byte count with a `timeDelta` the client never measured, which yields a +/// bandwidth figure that grows with however much the server chose to send. +#[test] +fn connect_time_bandwidth_without_a_clock_reports_the_stop_payload_alone() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + for request in [ + AutoDetectRequest::bw_start_connect_time(0x5555), + AutoDetectRequest::bw_payload(0x5555, vec![0u8; 1024]), + AutoDetectRequest::bw_stop_connect_time(0x5555, vec![0u8; 512]), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + None, + &mut output, + ) + .unwrap(); + } + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 1, "a driver with no clock measured no interval"); + assert_eq!( + results.1, 512, + "the 1024-byte Payload is not counted, since no window was open to count it" + ); +} + +/// Payload messages accumulate across a timed window, and the total they reach is +/// what the Stop reports. This is the path the floor case deliberately skips, so it +/// needs its own coverage: without it, nothing would catch an accumulator that had +/// stopped adding. +#[test] +fn connect_time_bandwidth_measured_window_reports_every_payload() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + // Each message lands in its own read, a millisecond apart, so the window is + // timed and the accumulator is what decides the reported total. + for (request, arrival) in [ + (AutoDetectRequest::bw_start_connect_time(0x4444), 2_000), + (AutoDetectRequest::bw_payload(0x4444, vec![0u8; 2048]), 2_100), + (AutoDetectRequest::bw_payload(0x4444, vec![0u8; 1024]), 2_200), + (AutoDetectRequest::bw_stop_connect_time(0x4444, vec![0u8; 512]), 2_250), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + Some(MonotonicInstant::from_millis(arrival)), + &mut output, + ) + .unwrap(); + } + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 250, "interval is Stop arrival minus Start arrival"); + assert_eq!( + results.1, 3584, + "both Payload messages and the Stop payload are counted" + ); } diff --git a/crates/ironrdp-testsuite-core/tests/server/acceptor.rs b/crates/ironrdp-testsuite-core/tests/server/acceptor.rs index 9dc7a4cbcd..ed95dd60f7 100644 --- a/crates/ironrdp-testsuite-core/tests/server/acceptor.rs +++ b/crates/ironrdp-testsuite-core/tests/server/acceptor.rs @@ -1,5 +1,4 @@ use ironrdp_acceptor::Acceptor; -use ironrdp_connector::MonotonicInstant; use ironrdp_connector::{DesktopSize, Sequence as _, Written, encode_x224_packet}; use ironrdp_core::{WriteBuf, decode}; use ironrdp_pdu::gcc::ClientMessageChannelData; @@ -38,14 +37,12 @@ fn neg_failure_on_protocol_mismatch() { // Step 1: feed the connection request (HYBRID | HYBRID_EX, no SSL) let request_bytes = encode_connection_request(SecurityProtocol::HYBRID | SecurityProtocol::HYBRID_EX); let mut output = WriteBuf::new(); - let written = acceptor - .step(&request_bytes, MonotonicInstant::ZERO, &mut output) - .unwrap(); + let written = acceptor.step(&request_bytes, None, &mut output).unwrap(); assert!(matches!(written, Written::Nothing)); // Step 2: acceptor tries to send confirm, finds no common protocol let mut output = WriteBuf::new(); - let result = acceptor.step(&[], MonotonicInstant::ZERO, &mut output); + let result = acceptor.step(&[], None, &mut output); // Must be an error assert!(result.is_err(), "expected error on protocol mismatch"); @@ -81,12 +78,10 @@ fn neg_success_when_protocols_match() { let request_bytes = encode_connection_request(SecurityProtocol::SSL | SecurityProtocol::HYBRID); let mut output = WriteBuf::new(); - acceptor - .step(&request_bytes, MonotonicInstant::ZERO, &mut output) - .unwrap(); + acceptor.step(&request_bytes, None, &mut output).unwrap(); let mut output = WriteBuf::new(); - let written = acceptor.step(&[], MonotonicInstant::ZERO, &mut output).unwrap(); + let written = acceptor.step(&[], None, &mut output).unwrap(); assert!(!matches!(written, Written::Nothing)); let response_bytes = output.filled(); @@ -124,12 +119,8 @@ fn message_channel_advertised_when_client_requests_it() { // Connection request -> confirm -> (TLS upgrade) -> ready for ConnectInitial. let request_bytes = encode_connection_request(SecurityProtocol::SSL); - acceptor - .step(&request_bytes, MonotonicInstant::ZERO, &mut WriteBuf::new()) - .unwrap(); - acceptor - .step(&[], MonotonicInstant::ZERO, &mut WriteBuf::new()) - .unwrap(); + acceptor.step(&request_bytes, None, &mut WriteBuf::new()).unwrap(); + acceptor.step(&[], None, &mut WriteBuf::new()).unwrap(); acceptor.mark_security_upgrade_as_done(); // Client GCC with the message channel block and no network channels, so the @@ -141,12 +132,10 @@ fn message_channel_advertised_when_client_requests_it() { let mut initial_buf = WriteBuf::new(); encode_x224_packet(&connect_initial, &mut initial_buf).unwrap(); - acceptor - .step(initial_buf.filled(), MonotonicInstant::ZERO, &mut WriteBuf::new()) - .unwrap(); + acceptor.step(initial_buf.filled(), None, &mut WriteBuf::new()).unwrap(); let mut output = WriteBuf::new(); - acceptor.step(&[], MonotonicInstant::ZERO, &mut output).unwrap(); + acceptor.step(&[], None, &mut output).unwrap(); let payload = decode::>>(output.filled()).unwrap().0; let response = decode::(payload.data.as_ref()).unwrap(); @@ -182,12 +171,10 @@ fn neg_failure_hybrid_required() { let request_bytes = encode_connection_request(SecurityProtocol::SSL); let mut output = WriteBuf::new(); - acceptor - .step(&request_bytes, MonotonicInstant::ZERO, &mut output) - .unwrap(); + acceptor.step(&request_bytes, None, &mut output).unwrap(); let mut output = WriteBuf::new(); - let result = acceptor.step(&[], MonotonicInstant::ZERO, &mut output); + let result = acceptor.step(&[], None, &mut output); assert!(result.is_err()); let response_bytes = output.filled(); diff --git a/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs b/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs index 737286dc6d..70b05f57e6 100644 --- a/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs +++ b/crates/ironrdp-testsuite-core/tests/session/connection_activation.rs @@ -1,6 +1,5 @@ use std::borrow::Cow; -use ironrdp_connector::MonotonicInstant; use ironrdp_connector::connection_activation::{ConnectionActivationSequence, ConnectionActivationState}; use ironrdp_connector::{ ClientConnector, ClientConnectorState, Credentials, DesktopSize, MultitransportResult, Sequence as _, Written, @@ -108,7 +107,7 @@ fn demand_active_static_channel_chunk_size(chunk_size: Option) -> usize { let mut output = WriteBuf::new(); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(demand_active)); sequence - .step(&frame, MonotonicInstant::ZERO, &mut output) + .step(&frame, None, &mut output) .expect("demand active should be accepted"); match sequence.connection_activation_state() { @@ -155,7 +154,7 @@ fn deactivate_all_during_capabilities_exchange_stays_in_same_state() { let frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); let mut output = WriteBuf::new(); - let written = seq.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + let written = seq.step(&frame, None, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -178,7 +177,7 @@ fn client_connector_stays_in_capabilities_exchange_on_deactivate_all() { let frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); let mut output = WriteBuf::new(); - let written = connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + let written = connector.step(&frame, None, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -208,7 +207,7 @@ fn set_error_info_during_capabilities_exchange_surfaces_the_disconnect_reason() let mut output = WriteBuf::new(); let err = seq - .step(&frame, MonotonicInstant::ZERO, &mut output) + .step(&frame, None, &mut output) .expect_err("a Set Error Info PDU during capabilities exchange must end the sequence with an error"); let message = err.to_string(); @@ -241,7 +240,7 @@ fn none_error_info_during_capabilities_exchange_is_skipped() { let frame = encode_server_share_control(none_error_info); let mut output = WriteBuf::new(); - let written = seq.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + let written = seq.step(&frame, None, &mut output).unwrap(); assert_eq!(written, Written::Nothing); assert!( @@ -261,17 +260,13 @@ fn demand_active_after_deactivate_all_transitions_to_connection_finalization() { // First: feed DeactivateAll let deactivate_frame = encode_server_share_control(ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll)); - let written = seq - .step(&deactivate_frame, MonotonicInstant::ZERO, &mut output) - .unwrap(); + let written = seq.step(&deactivate_frame, None, &mut output).unwrap(); assert_eq!(written, Written::Nothing); // Then: feed ServerDemandActive let demand_active_frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); - let written = seq - .step(&demand_active_frame, MonotonicInstant::ZERO, &mut output) - .unwrap(); + let written = seq.step(&demand_active_frame, None, &mut output).unwrap(); assert!(written != Written::Nothing, "should have written ClientConfirmActive"); assert!( @@ -292,7 +287,7 @@ fn demand_active_captures_server_input_flags() { let mut output = WriteBuf::new(); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); - seq.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + seq.step(&frame, None, &mut output).unwrap(); match seq.connection_activation_state() { ConnectionActivationState::ConnectionFinalization { input_flags, .. } => { @@ -321,7 +316,7 @@ fn demand_active_without_input_capability_yields_empty_input_flags() { .retain(|c| !matches!(c, CapabilitySet::Input(_))); let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(demand_active)); - seq.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + seq.step(&frame, None, &mut output).unwrap(); match seq.connection_activation_state() { ConnectionActivationState::ConnectionFinalization { input_flags, .. } => { @@ -399,7 +394,7 @@ fn multitransport_request_is_surfaced_without_waiting_for_another_pdu() { ); let mut output = WriteBuf::new(); - connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); assert!( connector.should_perform_multitransport(), @@ -421,7 +416,7 @@ fn responding_returns_to_bootstrapping_for_the_next_request() { &multitransport_request(request_id, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); assert!(connector.should_perform_multitransport()); assert_eq!(connector.multitransport_request().unwrap().request_id, request_id); @@ -450,7 +445,7 @@ fn third_multitransport_request_is_rejected() { &multitransport_request(request_id, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); connector .complete_multitransport(MultitransportResult::Success, &mut output) .unwrap(); @@ -460,7 +455,7 @@ fn third_multitransport_request_is_rejected() { &multitransport_request(3, RequestedProtocol::UdpFecR), MESSAGE_CHANNEL_ID, ); - assert!(connector.step(&frame, MonotonicInstant::ZERO, &mut output).is_err()); + assert!(connector.step(&frame, None, &mut output).is_err()); } #[test] @@ -471,7 +466,7 @@ fn demand_active_on_the_io_channel_ends_bootstrapping() { let frame = encode_server_share_control(ShareControlPdu::ServerDemandActive(SERVER_DEMAND_ACTIVE.clone())); let mut output = WriteBuf::new(); - connector.step(&frame, MonotonicInstant::ZERO, &mut output).unwrap(); + connector.step(&frame, None, &mut output).unwrap(); assert!( matches!(connector.state, ClientConnectorState::ConnectionFinalization { .. }), diff --git a/crates/ironrdp-web/src/session.rs b/crates/ironrdp-web/src/session.rs index 8d4d9da922..c77670f9d5 100644 --- a/crates/ironrdp-web/src/session.rs +++ b/crates/ironrdp-web/src/session.rs @@ -1866,11 +1866,7 @@ where debug_assert!(connector.next_pdu_hint().is_some()); buf.clear(); - let written = connector.step( - x224_connection_response.as_bytes(), - connector::MonotonicInstant::ZERO, - &mut buf, - )?; + let written = connector.step(x224_connection_response.as_bytes(), None, &mut buf)?; debug_assert!(written.is_nothing()); let should_upgrade = ironrdp_futures::skip_connect_begin(connector); diff --git a/ffi/src/connector/activation.rs b/ffi/src/connector/activation.rs index d84eb9ccf9..d6f43aa9c0 100644 --- a/ffi/src/connector/activation.rs +++ b/ffi/src/connector/activation.rs @@ -27,11 +27,7 @@ pub mod ffi { } pub fn step(&mut self, pdu_hint: &[u8], buf: &mut WriteBuf) -> Result, Box> { - let res = self - .0 - .step(pdu_hint, ironrdp::connector::MonotonicInstant::ZERO, &mut buf.0) - .map(Written) - .map(Box::new)?; + let res = self.0.step(pdu_hint, None, &mut buf.0).map(Written).map(Box::new)?; Ok(res) } diff --git a/ffi/src/connector/mod.rs b/ffi/src/connector/mod.rs index 3384ce300c..e13df8dc3a 100644 --- a/ffi/src/connector/mod.rs +++ b/ffi/src/connector/mod.rs @@ -193,7 +193,7 @@ pub mod ffi { let Some(connector) = self.0.as_mut() else { return Err(ValueConsumedError::for_item("connector").into()); }; - let written = connector.step(input, ironrdp::connector::MonotonicInstant::ZERO, &mut write_buf.0)?; + let written = connector.step(input, None, &mut write_buf.0)?; Ok(Box::new(Written(written))) } From ec1a5147f3e62b40323f8ba6e0e869709c8a947e Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Thu, 6 Aug 2026 18:07:58 -0500 Subject: [PATCH 3/6] test(connector): cover the bandwidth window reset and arrival stamping Two gaps the deep reviewer named on #1530 that were left open. A second Bandwidth Measure Start has to clear both stores and restart the timer per MS-RDPBCGR 3.2.5.14. That was the one branch the spec explicitly requires and nothing exercised it; the new test fails independently if either half of the reset is dropped. Nothing pinned Framed::last_read_at either. The connector tests feed Sequence::step hand-picked instants, so a regression that stopped stamping reads would leave them green while every measured window collapsed to the unmeasurable floor. These cover both directions: a stamp that stops advancing, and a stamp taken on drain rather than on the read, which would break the rule that a PDU served from the buffer keeps the arrival time of the read that filled it. They live in ironrdp-testsuite-core rather than inline because ironrdp-async sets [lib] test = false, so an inline module would compile and never run. --- Cargo.lock | 1 + crates/ironrdp-testsuite-core/Cargo.toml | 1 + .../tests/async_framed.rs | 171 ++++++++++++++++++ .../tests/connector/autodetect.rs | 34 ++++ crates/ironrdp-testsuite-core/tests/main.rs | 1 + 5 files changed, 208 insertions(+) create mode 100644 crates/ironrdp-testsuite-core/tests/async_framed.rs diff --git a/Cargo.lock b/Cargo.lock index 74cfbeaf08..50f5688d92 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3241,6 +3241,7 @@ dependencies = [ "expect-test", "hex", "ironrdp-acceptor", + "ironrdp-async", "ironrdp-bulk", "ironrdp-cfg", "ironrdp-cliprdr", diff --git a/crates/ironrdp-testsuite-core/Cargo.toml b/crates/ironrdp-testsuite-core/Cargo.toml index 06e21ac0a2..50983c596a 100644 --- a/crates/ironrdp-testsuite-core/Cargo.toml +++ b/crates/ironrdp-testsuite-core/Cargo.toml @@ -40,6 +40,7 @@ hex = "0.4" ironrdp-cliprdr-format.path = "../ironrdp-cliprdr-format" ironrdp-cliprdr = { path = "../ironrdp-cliprdr", features = ["__test"] } ironrdp-acceptor.path = "../ironrdp-acceptor" +ironrdp-async.path = "../ironrdp-async" ironrdp-bulk.path = "../ironrdp-bulk" ironrdp-connector.path = "../ironrdp-connector" ironrdp-displaycontrol.path = "../ironrdp-displaycontrol" diff --git a/crates/ironrdp-testsuite-core/tests/async_framed.rs b/crates/ironrdp-testsuite-core/tests/async_framed.rs new file mode 100644 index 0000000000..a9ad9910a1 --- /dev/null +++ b/crates/ironrdp-testsuite-core/tests/async_framed.rs @@ -0,0 +1,171 @@ +//! Arrival-time stamping on [`Framed`]. +//! +//! The connect-time bandwidth measurement in `ironrdp-connector` is only as +//! honest as `Framed::last_read_at`, and nothing else pins that down: the +//! connector tests drive `Sequence::step` with hand-picked instants, so a +//! regression that stopped stamping reads would leave them green while every +//! measured window silently collapsed to the unmeasurable floor. +//! +//! These live here rather than inline in `ironrdp-async` because that crate +//! sets `[lib] test = false`, so an inline `#[cfg(test)]` module would compile +//! and never run. + +use core::pin::Pin; +use core::task::{Context, Poll, Waker}; +use core::time::Duration; +use std::collections::VecDeque; +use std::io; + +use ironrdp_async::bytes::BytesMut; +use ironrdp_async::{Framed, FramedRead, StreamWrapper}; + +/// Drives a future to completion on the current thread. The mock below never +/// yields, so polling in a loop is enough and saves pulling in a runtime. +fn block_on(fut: F) -> F::Output { + let mut fut = core::pin::pin!(fut); + let mut cx = Context::from_waker(Waker::noop()); + loop { + if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) { + return value; + } + } +} + +/// A stream that hands over pre-arranged chunks, one per read, so a test can +/// decide exactly which PDUs share a socket read and which get their own. +struct ChunkedStream { + chunks: VecDeque>, + /// Delay applied before each read completes, so consecutive reads land on + /// distinguishable instants rather than relying on clock resolution. + delay: Duration, +} + +impl ChunkedStream { + fn new(chunks: impl IntoIterator>) -> Self { + Self { + chunks: chunks.into_iter().collect(), + delay: Duration::from_millis(5), + } + } +} + +impl StreamWrapper for ChunkedStream { + type InnerStream = Self; + + fn from_inner(stream: Self::InnerStream) -> Self { + stream + } + + fn into_inner(self) -> Self::InnerStream { + self + } + + fn get_inner(&self) -> &Self::InnerStream { + self + } + + fn get_inner_mut(&mut self) -> &mut Self::InnerStream { + self + } +} + +impl FramedRead for ChunkedStream { + type ReadFut<'read> + = Pin> + 'read>> + where + Self: 'read; + + fn read<'a>(&'a mut self, buf: &'a mut BytesMut) -> Self::ReadFut<'a> { + Box::pin(async move { + std::thread::sleep(self.delay); + match self.chunks.pop_front() { + Some(chunk) => { + buf.extend_from_slice(&chunk); + Ok(chunk.len()) + } + None => Ok(0), + } + }) + } +} + +/// Smallest frame `ironrdp_pdu::find_size` will accept: a TPKT header whose +/// length field covers itself plus `payload`. +fn tpkt(payload: &[u8]) -> Vec { + let length = u16::try_from(4 + payload.len()).expect("frame fits"); + let mut frame = vec![0x03, 0x00]; + frame.extend_from_slice(&length.to_be_bytes()); + frame.extend_from_slice(payload); + frame +} + +#[test] +fn nothing_read_yet_means_no_arrival_time() { + let framed = Framed::::new(ChunkedStream::new([])); + + assert!( + framed.last_read_at().is_none(), + "an unread Framed has observed no arrival" + ); +} + +#[test] +fn each_socket_read_advances_the_arrival_time() { + // One frame per read, so each PDU is stamped by the read that carried it. + let mut framed = Framed::::new(ChunkedStream::new([tpkt(&[0xAA; 8]), tpkt(&[0xBB; 8])])); + + block_on(framed.read_pdu()).expect("first frame"); + let first = framed.last_read_at().expect("host build observes time"); + + block_on(framed.read_pdu()).expect("second frame"); + let second = framed.last_read_at().expect("host build observes time"); + + assert!( + second.duration_since(first) >= Duration::from_millis(2), + "the second read must stamp a later arrival than the first, got {:?}", + second.duration_since(first) + ); +} + +#[test] +fn pdus_sharing_a_socket_read_share_its_arrival_time() { + // Both frames arrive in one read. The second is served from the buffer, so + // it arrived when that read completed, not when the caller drained it. + let mut chunk = tpkt(&[0xAA; 8]); + chunk.extend_from_slice(&tpkt(&[0xBB; 8])); + let mut framed = Framed::::new(ChunkedStream::new([chunk])); + + block_on(framed.read_pdu()).expect("first frame"); + let first = framed.last_read_at().expect("host build observes time"); + + // Wait before draining the second one. `MonotonicInstant` counts whole + // milliseconds, so without this the two drains would be indistinguishable + // and the assertion below would hold even if the stamp were taken on drain + // rather than on the read. + std::thread::sleep(Duration::from_millis(10)); + + block_on(framed.read_pdu()).expect("second frame"); + let second = framed.last_read_at().expect("host build observes time"); + + assert_eq!( + first, second, + "a PDU served from the buffer keeps the arrival time of the read that filled it" + ); +} + +#[test] +fn leftover_carried_into_a_new_framed_has_no_arrival_time() { + // `ironrdp-tokio`'s split/unsplit helpers rebuild a `Framed` around bytes + // that were read by the previous one. Those bytes did arrive, but not on + // this `Framed`, and it has no way to know when: reporting an arrival here + // would be inventing one. + let leftover = BytesMut::from(&*tpkt(&[0xDD; 8])); + let mut framed = Framed::::new_with_leftover(ChunkedStream::new([]), leftover); + + block_on(framed.read_pdu()).expect("frame served entirely from leftover"); + + assert!( + framed.last_read_at().is_none(), + "a frame served from leftover was never read by this Framed, so it has no arrival time" + ); +} diff --git a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs index 9e13ba2436..66f8bcdf51 100644 --- a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs @@ -407,3 +407,37 @@ fn connect_time_bandwidth_measured_window_reports_every_payload() { "both Payload messages and the Stop payload are counted" ); } + +#[test] +fn connect_time_bandwidth_second_start_discards_the_first_window() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + // [MS-RDPBCGR] 3.2.5.14 has the client clear both stores and restart the + // timer on each Bandwidth Measure Start, so the 4096 bytes counted into the + // abandoned window must not survive into the reported total, and the + // interval must be measured from the second Start rather than the first. + for (request, arrival) in [ + (AutoDetectRequest::bw_start_connect_time(0x5555), 1_000), + (AutoDetectRequest::bw_payload(0x5555, vec![0u8; 4096]), 1_100), + (AutoDetectRequest::bw_start_connect_time(0x5555), 2_000), + (AutoDetectRequest::bw_payload(0x5555, vec![0u8; 1024]), 2_100), + (AutoDetectRequest::bw_stop_connect_time(0x5555, vec![0u8; 512]), 2_500), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + Some(MonotonicInstant::from_millis(arrival)), + &mut output, + ) + .unwrap(); + } + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 500, "interval runs from the second Start, not the first"); + assert_eq!( + results.1, 1536, + "the 4096 bytes counted before the second Start are discarded" + ); +} diff --git a/crates/ironrdp-testsuite-core/tests/main.rs b/crates/ironrdp-testsuite-core/tests/main.rs index f40b8fdb7c..318198a37a 100644 --- a/crates/ironrdp-testsuite-core/tests/main.rs +++ b/crates/ironrdp-testsuite-core/tests/main.rs @@ -12,6 +12,7 @@ //! Cargo will run all tests from a single binary in parallel, but //! binaries themselves are run sequentially. +mod async_framed; mod cfg; mod clipboard; mod connector; From 2cdd0c2f015981c60986b11ebf86336d37a6f705 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Fri, 7 Aug 2026 11:04:10 -0500 Subject: [PATCH 4/6] fix(connector): read a wasm clock and trim the arrival-time tests `web-time` is `std::time` on every target except `wasm32-unknown-unknown`, where it reads `Performance.now()`, so the driver clock no longer has a hole in the browser build and `monotonic_now` is one function again rather than two behind a `cfg`. The crate is already in `Cargo.lock` through quinn and rustls-pki-types, so this is an edge and not a new dependency. The `Option` stays. `ffi/src/connector/` does not use `Framed`: the embedder owns the read loop and hands `step` bytes it has already read, so a clock read there would stamp when the caller called rather than when the bytes arrived, which is the error this series exists to remove. `MonotonicInstant`'s doc said the absence was about wasm; it now says what it is actually about. Test pass: ten to eight, no assertion lost. `connect_time_bandwidth_reports_the_measured_interval_and_total_bytes` was subsumed by `..._measured_window_reports_every_payload`, which asserts the same interval and also covers accumulation. `nothing_read_yet_means_no_arrival_time` survives as the opening assertion of `each_socket_read_advances_the_arrival_time`. --- Cargo.lock | 1 + crates/ironrdp-async/Cargo.toml | 1 + crates/ironrdp-async/src/framed.rs | 28 ++++++--------- crates/ironrdp-blocking/src/framed.rs | 8 ++--- crates/ironrdp-connector/src/lib.rs | 11 +++--- .../tests/async_framed.rs | 11 ++---- .../tests/connector/autodetect.rs | 35 ------------------- 7 files changed, 24 insertions(+), 71 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 50f5688d92..6a079a8636 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2588,6 +2588,7 @@ dependencies = [ "ironrdp-core 0.2.1", "ironrdp-pdu", "tracing", + "web-time", ] [[package]] diff --git a/crates/ironrdp-async/Cargo.toml b/crates/ironrdp-async/Cargo.toml index 749f3257c8..5423515e3e 100644 --- a/crates/ironrdp-async/Cargo.toml +++ b/crates/ironrdp-async/Cargo.toml @@ -22,6 +22,7 @@ ironrdp-core = { path = "../ironrdp-core", version = "0.2", features = ["alloc"] ironrdp-pdu = { path = "../ironrdp-pdu", version = "0.9" } # public tracing = { version = "0.1", features = ["log"] } bytes = "1" # public +web-time = "1.1" [lints] workspace = true diff --git a/crates/ironrdp-async/src/framed.rs b/crates/ironrdp-async/src/framed.rs index 528ff8a823..774858d988 100644 --- a/crates/ironrdp-async/src/framed.rs +++ b/crates/ironrdp-async/src/framed.rs @@ -213,7 +213,7 @@ where /// completes first, then it is guaranteed that no data was read. async fn read(&mut self) -> io::Result { let len = self.stream.read(&mut self.buf).await?; - self.last_read_at = monotonic_now(); + self.last_read_at = Some(monotonic_now()); Ok(len) } } @@ -305,24 +305,16 @@ where Ok(()) } -/// Reads the driver-owned monotonic clock, if this build has one. +/// Reads the driver-owned monotonic clock. /// /// The epoch is the first call; only differences are meaningful. /// -/// `std::time::Instant::now` panics on `wasm32-unknown-unknown`, and this crate is -/// reached from `ironrdp-web` through `ironrdp-futures`, so the browser build has no -/// clock to read and reports `None` rather than a fabricated reading. Giving it one -/// means plumbing a clock in from the embedder, which has `Performance.now()` -/// available to it. -#[cfg(not(target_arch = "wasm32"))] -fn monotonic_now() -> Option { - static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(std::time::Instant::now); - Some(MonotonicInstant::from_millis( - u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX), - )) -} - -#[cfg(target_arch = "wasm32")] -fn monotonic_now() -> Option { - None +/// `web_time::Instant` is `std::time::Instant` everywhere except +/// `wasm32-unknown-unknown`, where `std`'s panics and this one reads +/// `Performance.now()` instead. This crate is reached from `ironrdp-web` through +/// `ironrdp-futures`, so without it the browser build has no clock and every +/// measurement there is lost. +fn monotonic_now() -> MonotonicInstant { + static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(web_time::Instant::now); + MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)) } diff --git a/crates/ironrdp-blocking/src/framed.rs b/crates/ironrdp-blocking/src/framed.rs index 04bf1d1729..50bc5c8c5f 100644 --- a/crates/ironrdp-blocking/src/framed.rs +++ b/crates/ironrdp-blocking/src/framed.rs @@ -132,7 +132,7 @@ where let mut read_bytes = [0u8; 1024]; let len = self.stream.read(&mut read_bytes)?; - self.last_read_at = monotonic_now(); + self.last_read_at = Some(monotonic_now()); self.buf.extend_from_slice(&read_bytes[..len]); Ok(len) @@ -151,9 +151,7 @@ where /// Reads the driver-owned monotonic clock. Epoch is the first call; only /// differences are meaningful. -fn monotonic_now() -> Option { +fn monotonic_now() -> MonotonicInstant { static EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(std::time::Instant::now); - Some(MonotonicInstant::from_millis( - u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX), - )) + MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX)) } diff --git a/crates/ironrdp-connector/src/lib.rs b/crates/ironrdp-connector/src/lib.rs index ff06324eb1..cd69e8244e 100644 --- a/crates/ironrdp-connector/src/lib.rs +++ b/crates/ironrdp-connector/src/lib.rs @@ -358,11 +358,12 @@ impl Written { /// A point on a monotonic millisecond clock owned by the I/O driver. /// /// The epoch is arbitrary and carries no meaning; only differences between two -/// instants do. The clock deliberately lives outside the sans-I/O sequences: -/// `std::time::Instant::now` panics on `wasm32-unknown-unknown`, which -/// `ironrdp-connector` compiles for, and a sequence reading a clock itself would -/// measure how quickly it drained an already-filled buffer rather than how long -/// the bytes took to arrive. +/// instants do. The clock deliberately lives outside the sans-I/O sequences, +/// because a sequence reading a clock itself would measure how quickly it +/// drained an already-filled buffer rather than how long the bytes took to +/// arrive. Only the driver that performed the read knows the latter, and a +/// driver whose caller owns the read loop does not know it either: that is what +/// the `None` in [`Sequence::step`] is for, and why it stays. #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] pub struct MonotonicInstant(u64); diff --git a/crates/ironrdp-testsuite-core/tests/async_framed.rs b/crates/ironrdp-testsuite-core/tests/async_framed.rs index a9ad9910a1..bcde3ac1e1 100644 --- a/crates/ironrdp-testsuite-core/tests/async_framed.rs +++ b/crates/ironrdp-testsuite-core/tests/async_framed.rs @@ -100,19 +100,14 @@ fn tpkt(payload: &[u8]) -> Vec { } #[test] -fn nothing_read_yet_means_no_arrival_time() { - let framed = Framed::::new(ChunkedStream::new([])); +fn each_socket_read_advances_the_arrival_time() { + // One frame per read, so each PDU is stamped by the read that carried it. + let mut framed = Framed::::new(ChunkedStream::new([tpkt(&[0xAA; 8]), tpkt(&[0xBB; 8])])); assert!( framed.last_read_at().is_none(), "an unread Framed has observed no arrival" ); -} - -#[test] -fn each_socket_read_advances_the_arrival_time() { - // One frame per read, so each PDU is stamped by the read that carried it. - let mut framed = Framed::::new(ChunkedStream::new([tpkt(&[0xAA; 8]), tpkt(&[0xBB; 8])])); block_on(framed.read_pdu()).expect("first frame"); let first = framed.last_read_at().expect("host build observes time"); diff --git a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs index 66f8bcdf51..0e3985cec2 100644 --- a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs @@ -225,41 +225,6 @@ fn decode_bandwidth_results(output: &WriteBuf) -> (u32, u32) { } } -/// The reported interval is the time between the Start that opened the window and -/// the Stop that closed it, taken from the arrival times the driver observed. -#[test] -fn connect_time_bandwidth_reports_the_measured_interval_and_total_bytes() { - let mut connector = connect_time_autodetect_connector(); - let mut output = WriteBuf::new(); - - let start = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_start_connect_time(0x1111))).unwrap(); - connector - .step( - &server_send_data_indication(MESSAGE_CHANNEL_ID, start), - Some(MonotonicInstant::from_millis(1_000)), - &mut output, - ) - .unwrap(); - - output.clear(); - let stop = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_stop_connect_time( - 0x1111, - vec![0u8; 4096], - ))) - .unwrap(); - connector - .step( - &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), - Some(MonotonicInstant::from_millis(1_250)), - &mut output, - ) - .unwrap(); - - let results = decode_bandwidth_results(&output); - assert_eq!(results.0, 250, "interval is Stop arrival minus Start arrival"); - assert_eq!(results.1, 4096, "byte count is what the server sent in the window"); -} - /// A Stop with no preceding Start has nothing to have measured. It is still /// answered, because the server blocks without a reply, but the interval reported /// is the unmeasurable floor rather than an invented figure. From 5b3afc00f556e4ca12dac86bf502513446486609 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Sat, 15 Aug 2026 13:50:19 -0500 Subject: [PATCH 5/6] fix(connector): address the review round on connect-time bandwidth measurement Fixes the byteCount undercount: MS-RDPBCGR 3.2.5.14 has the client add the 8-byte PDU header to payloadLength on both the Bandwidth Measure Payload accumulator and the Stop's own contribution, not payloadLength alone. Both sites were missing it. Makes the accumulated-byte discard visible when a Stop arrives with no arrival time although its window was open, instead of dropping the count with no trace. Corrects a comment that claimed wasm32 has no clock for the whole connection; it only lacks one for the single x224_connection_response step. Moves the Framed arrival-time test out of ironrdp-testsuite-core into ironrdp-testsuite-extra, since pulling in ironrdp-async as a direct dev-dependency violated the core tier's no-extra-tier-dependency invariant. --- Cargo.lock | 1 - crates/ironrdp-connector/src/connection.rs | 40 ++++++++-- crates/ironrdp-testsuite-core/Cargo.toml | 1 - .../tests/connector/autodetect.rs | 78 ++++++++++++++++--- crates/ironrdp-testsuite-core/tests/main.rs | 1 - .../tests/async_framed.rs | 0 crates/ironrdp-testsuite-extra/tests/main.rs | 1 + 7 files changed, 101 insertions(+), 21 deletions(-) rename crates/{ironrdp-testsuite-core => ironrdp-testsuite-extra}/tests/async_framed.rs (100%) diff --git a/Cargo.lock b/Cargo.lock index 6a079a8636..ee396f48c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3242,7 +3242,6 @@ dependencies = [ "expect-test", "hex", "ironrdp-acceptor", - "ironrdp-async", "ironrdp-bulk", "ironrdp-cfg", "ironrdp-cliprdr", diff --git a/crates/ironrdp-connector/src/connection.rs b/crates/ironrdp-connector/src/connection.rs index 07f00bdfb3..7efee90317 100644 --- a/crates/ironrdp-connector/src/connection.rs +++ b/crates/ironrdp-connector/src/connection.rs @@ -745,9 +745,15 @@ impl ClientConnector { // 2.2.14.1.3). No reply is due; accumulate so Stop can report the total. // With no window open there is nothing for the total to be divided by, so // there is nothing worth accumulating. + // + // [MS-RDPBCGR] 3.2.5.14 increments the Byte Count store by payloadLength + // plus the size of the header fields (8 bytes: headerLength, headerTypeId, + // sequenceNumber, requestType, and payloadLength itself), not by + // payloadLength alone. `payload.len()` is exactly payloadLength, since + // decode reads that many bytes into it after consuming the header fields. AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => { if self.connect_time_bw_started_at.is_some() { - let len = u32::try_from(payload.len()).unwrap_or(u32::MAX); + let len = u32::try_from(payload.len()).unwrap_or(u32::MAX).saturating_add(8); self.connect_time_bw_bytes = self.connect_time_bw_bytes.saturating_add(len); } Ok(Written::Nothing) @@ -766,8 +772,9 @@ impl ClientConnector { // // The spec does not contemplate a window it could not time, and two arise in // practice. No Start was seen, so no window exists. Or the driver reports no - // arrival times at all, as the wasm32 and FFI drivers do for the whole - // connection, so no window was opened to accumulate into. + // arrival time for this particular step, as the FFI driver does for the whole + // connection and the wasm32 driver does for the single x224_connection_response + // step, so no window was opened to accumulate into. // // Both still owe the server a reply, and `timeDelta` of 0 divides out to an // unbounded bandwidth for a server computing `byteCount * 8 / timeDelta` @@ -785,12 +792,21 @@ impl ClientConnector { payload, .. } => { + // Same 8-byte header addition as the Payload arm above, and for the + // same spec reason ([MS-RDPBCGR] 3.2.5.14): only applies when this Stop + // actually carries a payloadLength/payload pair. let stop_bytes = payload .as_ref() - .map_or(0, |p| u32::try_from(p.len()).unwrap_or(u32::MAX)); - - // A window only opens when the Start carried a reading, so the same - // driver stamps this Stop. The `None` arm covers the unopened window. + .map_or(0, |p| u32::try_from(p.len()).unwrap_or(u32::MAX).saturating_add(8)); + + // A window normally opens and closes on the same driver, so the same + // driver stamps both Start and this Stop. Nothing enforces that: a + // `Framed` rebuilt between the two (leftover bytes handed to a fresh + // `Framed`, which starts with no arrival time of its own) would open a + // window on one driver and close it on another with no reading, landing + // in the `(Some, None)` arm below. That arm silently drops whatever this + // window had accumulated; the debug log makes the drop visible instead of + // leaving it indistinguishable from the ordinary no-window case. let (time_delta_ms, byte_count) = match (self.connect_time_bw_started_at, received_at) { (Some(started_at), Some(stopped_at)) => { let measured_ms = @@ -800,7 +816,15 @@ impl ClientConnector { self.connect_time_bw_bytes.saturating_add(stop_bytes), ) } - _ => (UNMEASURABLE_INTERVAL_MS, stop_bytes), + (Some(_), None) => { + debug!( + dropped_bytes = self.connect_time_bw_bytes, + "Bandwidth Measure Stop arrived with no arrival time although its window was open; \ + dropping the accumulated count" + ); + (UNMEASURABLE_INTERVAL_MS, stop_bytes) + } + (None, _) => (UNMEASURABLE_INTERVAL_MS, stop_bytes), }; self.connect_time_bw_started_at = None; diff --git a/crates/ironrdp-testsuite-core/Cargo.toml b/crates/ironrdp-testsuite-core/Cargo.toml index 50983c596a..06e21ac0a2 100644 --- a/crates/ironrdp-testsuite-core/Cargo.toml +++ b/crates/ironrdp-testsuite-core/Cargo.toml @@ -40,7 +40,6 @@ hex = "0.4" ironrdp-cliprdr-format.path = "../ironrdp-cliprdr-format" ironrdp-cliprdr = { path = "../ironrdp-cliprdr", features = ["__test"] } ironrdp-acceptor.path = "../ironrdp-acceptor" -ironrdp-async.path = "../ironrdp-async" ironrdp-bulk.path = "../ironrdp-bulk" ironrdp-connector.path = "../ironrdp-connector" ironrdp-displaycontrol.path = "../ironrdp-displaycontrol" diff --git a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs index 0e3985cec2..804d1ad1d7 100644 --- a/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs +++ b/crates/ironrdp-testsuite-core/tests/connector/autodetect.rs @@ -299,12 +299,70 @@ fn connect_time_bandwidth_coalesced_into_one_read_reports_the_floor() { let results = decode_bandwidth_results(&output); assert_eq!(results.0, 1, "a window that arrived in one read floors to 1 ms"); - assert_eq!(results.1, 1536, "every byte in the timed window is still counted"); + assert_eq!( + results.1, 1552, + "every byte in the timed window is still counted, plus the 8-byte header on each of the Payload and Stop" + ); +} + +/// A window can open on one driver and close on another, for example when a +/// `Framed` is rebuilt with leftover bytes between the Start and this Stop: the +/// rebuilt `Framed` starts with no arrival time of its own, so the Stop reports +/// `None` even though the Start that opened the window reported `Some`. Nothing +/// upstream of this function stops that from happening, so it is reachable even +/// though it did not use to be exercised by a test. +/// +/// The window's accumulated bytes are dropped in that case: reporting them +/// against `timeDelta = UNMEASURABLE_INTERVAL_MS` would pair a byte count that +/// arrived over the window's real duration with a floor timer that understates +/// it, inflating the reported bandwidth the same way an uncounted header would. +#[test] +fn connect_time_bandwidth_stop_with_no_arrival_time_drops_the_open_window() { + let mut connector = connect_time_autodetect_connector(); + let mut output = WriteBuf::new(); + + let arrival = Some(MonotonicInstant::from_millis(9_000)); + + for request in [ + AutoDetectRequest::bw_start_connect_time(0x7777), + AutoDetectRequest::bw_payload(0x7777, vec![0u8; 2048]), + ] { + output.clear(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, encode_vec(&AutoDetectReqPdu::new(request)).unwrap()), + arrival, + &mut output, + ) + .unwrap(); + } + + output.clear(); + let stop = encode_vec(&AutoDetectReqPdu::new(AutoDetectRequest::bw_stop_connect_time( + 0x7777, + vec![0u8; 512], + ))) + .unwrap(); + connector + .step( + &server_send_data_indication(MESSAGE_CHANNEL_ID, stop), + None, + &mut output, + ) + .unwrap(); + + let results = decode_bandwidth_results(&output); + assert_eq!(results.0, 1, "an unmeasured window still floors to 1 ms"); + assert_eq!( + results.1, 520, + "the 2048-byte Payload is dropped along with the window; only the Stop's own 512 bytes plus its 8-byte header remain" + ); } -/// A driver with no clock reports no arrival times, so no window is ever opened and -/// there is nothing to accumulate into. The wasm32 and FFI drivers are in this -/// position for the whole connection. +/// A driver with no clock reports no arrival time for this step, so no window is +/// ever opened and there is nothing to accumulate into. The FFI driver is in this +/// position for the whole connection; the wasm32 driver only for the single +/// x224_connection_response step. /// /// The server still blocks without a reply, so one is sent, but it claims no more /// than this Stop's own payload. Counting the Payload messages here would pair a @@ -333,8 +391,8 @@ fn connect_time_bandwidth_without_a_clock_reports_the_stop_payload_alone() { let results = decode_bandwidth_results(&output); assert_eq!(results.0, 1, "a driver with no clock measured no interval"); assert_eq!( - results.1, 512, - "the 1024-byte Payload is not counted, since no window was open to count it" + results.1, 520, + "the 1024-byte Payload is not counted, since no window was open; the Stop's own 512 bytes plus its 8-byte header are" ); } @@ -368,8 +426,8 @@ fn connect_time_bandwidth_measured_window_reports_every_payload() { let results = decode_bandwidth_results(&output); assert_eq!(results.0, 250, "interval is Stop arrival minus Start arrival"); assert_eq!( - results.1, 3584, - "both Payload messages and the Stop payload are counted" + results.1, 3608, + "both Payload messages and the Stop payload are counted, each with its 8-byte header" ); } @@ -402,7 +460,7 @@ fn connect_time_bandwidth_second_start_discards_the_first_window() { let results = decode_bandwidth_results(&output); assert_eq!(results.0, 500, "interval runs from the second Start, not the first"); assert_eq!( - results.1, 1536, - "the 4096 bytes counted before the second Start are discarded" + results.1, 1552, + "the 4096 bytes counted before the second Start are discarded; only the second window's Payload and Stop, each with an 8-byte header, remain" ); } diff --git a/crates/ironrdp-testsuite-core/tests/main.rs b/crates/ironrdp-testsuite-core/tests/main.rs index 318198a37a..f40b8fdb7c 100644 --- a/crates/ironrdp-testsuite-core/tests/main.rs +++ b/crates/ironrdp-testsuite-core/tests/main.rs @@ -12,7 +12,6 @@ //! Cargo will run all tests from a single binary in parallel, but //! binaries themselves are run sequentially. -mod async_framed; mod cfg; mod clipboard; mod connector; diff --git a/crates/ironrdp-testsuite-core/tests/async_framed.rs b/crates/ironrdp-testsuite-extra/tests/async_framed.rs similarity index 100% rename from crates/ironrdp-testsuite-core/tests/async_framed.rs rename to crates/ironrdp-testsuite-extra/tests/async_framed.rs diff --git a/crates/ironrdp-testsuite-extra/tests/main.rs b/crates/ironrdp-testsuite-extra/tests/main.rs index f24df9ccd4..7535c6bfc6 100644 --- a/crates/ironrdp-testsuite-extra/tests/main.rs +++ b/crates/ironrdp-testsuite-extra/tests/main.rs @@ -2,6 +2,7 @@ #![allow(clippy::unwrap_used, reason = "unwrap is fine in tests")] mod agent; +mod async_framed; mod capture_helpers; mod client_config; mod dvc_pipe_proxy; From 4cbef7ad69c964dec70410fd3dd3903815719d80 Mon Sep 17 00:00:00 2001 From: Greg Lamberson Date: Wed, 19 Aug 2026 17:12:55 -0500 Subject: [PATCH 6/6] refactor(core): move MonotonicInstant into ironrdp-core, shared with rdpeudp ironrdp-connector and ironrdp-rdpeudp each defined their own MonotonicInstant, identical in spirit but not in API (rdpeudp's carries as_millis/saturating_add, connector's doesn't), so a value from one could never be handed to the other. Move the fuller rdpeudp version into ironrdp-core, which both crates already depend on, and re-export it from both existing paths. No call site outside the two definitions changes: everything already imports MonotonicInstant through ironrdp_connector:: or ironrdp_rdpeudp::, and both keep working unchanged. --- crates/ironrdp-connector/src/lib.rs | 32 ++++---------- crates/ironrdp-core/src/lib.rs | 2 + crates/ironrdp-core/src/time.rs | 65 +++++++++++++++++++++++++++++ crates/ironrdp-rdpeudp/src/time.rs | 59 ++------------------------ 4 files changed, 79 insertions(+), 79 deletions(-) create mode 100644 crates/ironrdp-core/src/time.rs diff --git a/crates/ironrdp-connector/src/lib.rs b/crates/ironrdp-connector/src/lib.rs index cd69e8244e..5173485e9a 100644 --- a/crates/ironrdp-connector/src/lib.rs +++ b/crates/ironrdp-connector/src/lib.rs @@ -357,30 +357,14 @@ impl Written { /// A point on a monotonic millisecond clock owned by the I/O driver. /// -/// The epoch is arbitrary and carries no meaning; only differences between two -/// instants do. The clock deliberately lives outside the sans-I/O sequences, -/// because a sequence reading a clock itself would measure how quickly it -/// drained an already-filled buffer rather than how long the bytes took to -/// arrive. Only the driver that performed the read knows the latter, and a -/// driver whose caller owns the read loop does not know it either: that is what -/// the `None` in [`Sequence::step`] is for, and why it stays. -#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] -pub struct MonotonicInstant(u64); - -impl MonotonicInstant { - /// Builds an instant from a monotonic millisecond reading. - #[must_use] - pub fn from_millis(milliseconds: u64) -> Self { - Self(milliseconds) - } - - /// The time elapsed since `earlier`, saturating at zero if the clock went - /// backwards or the arguments were transposed. - #[must_use] - pub fn duration_since(self, earlier: Self) -> core::time::Duration { - core::time::Duration::from_millis(self.0.saturating_sub(earlier.0)) - } -} +/// Lives in `ironrdp-core`, shared with `ironrdp-rdpeudp`, and re-exported +/// here. The clock deliberately lives outside the sans-I/O sequences, because +/// a sequence reading a clock itself would measure how quickly it drained an +/// already-filled buffer rather than how long the bytes took to arrive. Only +/// the driver that performed the read knows the latter, and a driver whose +/// caller owns the read loop does not know it either: that is what the `None` +/// in [`Sequence::step`] is for, and why it stays. +pub use ironrdp_core::MonotonicInstant; pub trait Sequence: Send { fn next_pdu_hint(&self) -> Option<&dyn PduHint>; diff --git a/crates/ironrdp-core/src/lib.rs b/crates/ironrdp-core/src/lib.rs index bbef57bdda..35a36c37a1 100644 --- a/crates/ironrdp-core/src/lib.rs +++ b/crates/ironrdp-core/src/lib.rs @@ -19,6 +19,7 @@ mod into_owned; #[cfg(feature = "alloc")] mod non_empty; mod padding; +mod time; #[cfg(feature = "alloc")] mod write_buf; @@ -48,5 +49,6 @@ pub use self::into_owned::IntoOwned; #[cfg(feature = "alloc")] pub use self::non_empty::NonEmpty; pub use self::padding::{read_padding, write_padding}; +pub use self::time::MonotonicInstant; #[cfg(feature = "alloc")] pub use self::write_buf::WriteBuf; diff --git a/crates/ironrdp-core/src/time.rs b/crates/ironrdp-core/src/time.rs new file mode 100644 index 0000000000..1383d46fe4 --- /dev/null +++ b/crates/ironrdp-core/src/time.rs @@ -0,0 +1,65 @@ +//! Monotonic time readings supplied by the caller. +//! +//! Sans-I/O state machines never read a clock themselves. They receive the +//! current instant on every call that can advance time, and report when they +//! next want to be woken. The clock deliberately lives outside the sequence +//! or connection type, because a state machine reading a clock itself would +//! measure how quickly it drained an already-filled buffer rather than when +//! the bytes actually arrived on the wire; only the I/O driver that performed +//! the read knows that. +//! +//! `std::time::Instant` is deliberately not used here. `Instant::now` panics +//! on `wasm32-unknown-unknown`, which some drivers using this type compile +//! for, and this crate is `no_std` besides. + +use core::ops::Add; +use core::time::Duration; + +/// A monotonic instant, in milliseconds, from an epoch chosen by the caller. +/// +/// The epoch is arbitrary and carries no meaning; only differences between two +/// instants do. Millisecond resolution is well below the shortest interval any +/// current caller needs to measure. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct MonotonicInstant(u64); + +impl MonotonicInstant { + /// Builds an instant from a monotonic millisecond reading. + #[must_use] + pub fn from_millis(milliseconds: u64) -> Self { + Self(milliseconds) + } + + /// The reading this instant was built from. + #[must_use] + pub fn as_millis(self) -> u64 { + self.0 + } + + /// The time elapsed since `earlier`, saturating at zero if the clock went + /// backwards or the arguments were transposed. + #[must_use] + pub fn duration_since(self, earlier: Self) -> Duration { + Duration::from_millis(self.0.saturating_sub(earlier.0)) + } + + /// This instant advanced by `duration`, saturating at the end of the + /// representable range. + /// + /// Saturating is the right behavior for the only caller: a deadline that + /// cannot be represented is one that never fires, which is what a timer set + /// beyond the end of time should do. + #[must_use] + pub fn saturating_add(self, duration: Duration) -> Self { + let milliseconds = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX); + Self(self.0.saturating_add(milliseconds)) + } +} + +impl Add for MonotonicInstant { + type Output = Self; + + fn add(self, rhs: Duration) -> Self { + self.saturating_add(rhs) + } +} diff --git a/crates/ironrdp-rdpeudp/src/time.rs b/crates/ironrdp-rdpeudp/src/time.rs index 20b5ac5b2c..5be4464bd1 100644 --- a/crates/ironrdp-rdpeudp/src/time.rs +++ b/crates/ironrdp-rdpeudp/src/time.rs @@ -4,61 +4,10 @@ //! instant on every call that can advance time, and reports when it next wants //! to be woken through [`RdpeudpConnection::poll_timeout`]. //! -//! `std::time::Instant` is deliberately not used here. `Instant::now` panics -//! on `wasm32-unknown-unknown`, which this crate compiles for, and a state -//! machine that read a clock itself would measure how quickly it drained an -//! already-filled buffer rather than when the datagram actually arrived. +//! [`MonotonicInstant`] itself lives in `ironrdp-core`, shared with +//! `ironrdp-connector`'s `Sequence::step`, and is re-exported here so +//! existing callers of this crate are unaffected. //! //! [`RdpeudpConnection::poll_timeout`]: crate::RdpeudpConnection::poll_timeout -use core::ops::Add; -use core::time::Duration; - -/// A monotonic instant, in milliseconds, from an epoch chosen by the caller. -/// -/// The epoch is arbitrary and carries no meaning; only differences between two -/// instants do. Millisecond resolution is well below the shortest interval the -/// protocol asks us to measure, the delayed-ACK timer of [MS-RDPEUDP2] 3.1.1. -#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] -pub struct MonotonicInstant(u64); - -impl MonotonicInstant { - /// Builds an instant from a monotonic millisecond reading. - #[must_use] - pub fn from_millis(milliseconds: u64) -> Self { - Self(milliseconds) - } - - /// The reading this instant was built from. - #[must_use] - pub fn as_millis(self) -> u64 { - self.0 - } - - /// The time elapsed since `earlier`, saturating at zero if the clock went - /// backwards or the arguments were transposed. - #[must_use] - pub fn duration_since(self, earlier: Self) -> Duration { - Duration::from_millis(self.0.saturating_sub(earlier.0)) - } - - /// This instant advanced by `duration`, saturating at the end of the - /// representable range. - /// - /// Saturating is the right behavior for the only caller: a deadline that - /// cannot be represented is one that never fires, which is what a timer set - /// beyond the end of time should do. - #[must_use] - pub fn saturating_add(self, duration: Duration) -> Self { - let milliseconds = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX); - Self(self.0.saturating_add(milliseconds)) - } -} - -impl Add for MonotonicInstant { - type Output = Self; - - fn add(self, rhs: Duration) -> Self { - self.saturating_add(rhs) - } -} +pub use ironrdp_core::MonotonicInstant;