feat(connector)!: pass frame arrival time into Sequence::step

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<Written> {
+    fn step(
+        &mut self,
+        input: &[u8],
+        _received_at: Option<MonotonicInstant>,
+        output: &mut WriteBuf,
+    ) -> ConnectorResult<Written> {
         let (written, next_state) = match core::mem::take(&mut self.state) {
             ChannelConnectionState::WaitErectDomainRequest => {
                 let erect_domain_request = ironrdp_core::decode::<X224<mcs::ErectDomainPdu>>(input)

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(&[], None, &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(&[], None, &mut WriteBuf::new())
+            .expect("transition to next state");
         debug_assert!(!self.should_perform_credssp());
         assert_eq!(res, Written::Nothing);
     }
@@ -458,7 +461,12 @@ impl Sequence for Acceptor {
         &self.state
     }
 
-    fn step(&mut self, input: &[u8], output: &mut WriteBuf) -> ConnectorResult<Written> {
+    fn step(
+        &mut self,
+        input: &[u8],
+        received_at: Option<MonotonicInstant>,
+        output: &mut WriteBuf,
+    ) -> ConnectorResult<Written> {
         let prev_state = mem::take(&mut self.state);
 
         let (written, next_state) = match prev_state {
@@ -724,7 +732,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 +972,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 {

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<Written> {
+    fn step(
+        &mut self,
+        input: &[u8],
+        _received_at: Option<MonotonicInstant>,
+        output: &mut WriteBuf,
+    ) -> ConnectorResult<Written> {
         let (written, next_state) = match core::mem::take(&mut self.state) {
             FinalizationState::WaitSynchronize => {
                 let synchronize = decode_share_control(input);

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

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,25 @@ pub trait StreamWrapper: Sized {
 pub struct Framed<S> {
     stream: S,
     buf: BytesMut,
+    /// 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. `None` until the first
+    /// read, and always `None` on a build with no clock.
+    last_read_at: Option<MonotonicInstant>,
 }
 
 impl<S> Framed<S> {
     pub fn peek(&self) -> &[u8] {
         &self.buf
     }
+
+    /// 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<MonotonicInstant> {
+        self.last_read_at
+    }
 }
 
 impl<S> Framed<S>
@@ -76,6 +90,7 @@ where
         Self {
             stream: S::from_inner(stream),
             buf: leftover,
+            last_read_at: None,
         }
     }
 
@@ -197,7 +212,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<usize> {
-        self.stream.read(&mut self.buf).await
+        let len = self.stream.read(&mut self.buf).await?;
+        self.last_read_at = Some(monotonic_now());
+        Ok(len)
     }
 }
 
@@ -261,7 +278,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 +304,17 @@ where
 
     Ok(())
 }
+
+/// Reads the driver-owned monotonic clock.
+///
+/// The epoch is the first call; only differences are meaningful.
+///
+/// `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<web_time::Instant> = std::sync::LazyLock::new(web_time::Instant::now);
+    MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX))
+}

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)?
     };

crates/ironrdp-blocking/src/framed.rs

@@ -1,12 +1,17 @@
 use std::io::{self, Read, Write};
 
 use bytes::{Bytes, BytesMut};
+use ironrdp_connector::MonotonicInstant;
 use ironrdp_pdu::PduHint;
 use tracing::debug;
 
 pub struct Framed<S> {
     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.
+    /// `None` until the first read.
+    last_read_at: Option<MonotonicInstant>,
 }
 
 impl<S> Framed<S> {
@@ -15,7 +20,11 @@ impl<S> Framed<S> {
     }
 
     pub fn new_with_leftover(stream: S, leftover: BytesMut) -> Self {
-        Self { stream, buf: leftover }
+        Self {
+            stream,
+            buf: leftover,
+            last_read_at: None,
+        }
     }
 
     pub fn into_inner(self) -> (S, BytesMut) {
@@ -36,6 +45,11 @@ impl<S> Framed<S> {
         (&mut self.stream, &mut self.buf)
     }
 
+    /// When the bytes currently buffered last arrived from the socket.
+    pub fn last_read_at(&self) -> Option<MonotonicInstant> {
+        self.last_read_at
+    }
+
     pub fn peek(&self) -> &[u8] {
         &self.buf
     }
@@ -118,6 +132,7 @@ where
 
         let mut read_bytes = [0u8; 1024];
         let len = self.stream.read(&mut read_bytes)?;
+        self.last_read_at = Some(monotonic_now());
         self.buf.extend_from_slice(&read_bytes[..len]);
 
         Ok(len)
@@ -133,3 +148,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::time::Instant> = std::sync::LazyLock::new(std::time::Instant::now);
+    MonotonicInstant::from_millis(u64::try_from(EPOCH.elapsed().as_millis()).unwrap_or(u64::MAX))
+}

crates/ironrdp-client/src/rdp.rs

@@ -1896,7 +1896,7 @@ 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(), None, &mut buf)?;
                 debug_assert!(written.is_nothing());
 
                 let should_upgrade = ironrdp_tokio::skip_connect_begin(connector);

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<Written> {
+    fn step(
+        &mut self,
+        input: &[u8],
+        _received_at: Option<MonotonicInstant>,
+        output: &mut WriteBuf,
+    ) -> ConnectorResult<Written> {
         let (written, next_state) = match mem::take(&mut self.state) {
             ChannelConnectionState::Consumed => {
                 return Err(general_err!(