From 1eaa95367976483840c3f356b4f5232c20cff94e Mon Sep 17 00:00:00 2001 From: Saurabh Singh <1623701+saurabh500@users.noreply.github.com> Date: Wed, 12 Aug 2026 23:28:26 -0700 Subject: [PATCH 1/4] Remove async_trait from SqlTypeDecode The trait has a generic method, so it was already dyn-incompatible and the per-call boxing bought nothing. Removal exposed two latent constraints: - SQL_VARIANT decoding is genuinely recursive (decode -> read_sql_variant -> decode_zero_propbyte_variant -> decode). async_trait's box was silently breaking that cycle; a native async fn yields E0733. Reintroduced deliberately via decode_boxed, which pays one allocation per nested variant column instead of one per column read. - Callers wrap decode in async_trait futures that require Send, and the bare AFIT desugaring does not promise it, so the trait declares -> impl Future + Send. Measured at roughly 0% on its own; it ships for API hygiene and as the prerequisite for making the decode chain generic over the row writer. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 41e46220-ad76-4a77-b477-e768951dcf92 --- mssql-tds/src/datatypes/decoder.rs | 34 ++++++++++++++++---- mssql-tds/src/token/parsers/nbcrow_parser.rs | 4 --- mssql-tds/src/token/parsers/row_parser.rs | 4 --- 3 files changed, 27 insertions(+), 15 deletions(-) diff --git a/mssql-tds/src/datatypes/decoder.rs b/mssql-tds/src/datatypes/decoder.rs index 91100aae..6b91facf 100644 --- a/mssql-tds/src/datatypes/decoder.rs +++ b/mssql-tds/src/datatypes/decoder.rs @@ -1,8 +1,9 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -use async_trait::async_trait; use core::fmt; +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::{fmt::Debug, io::Error, vec}; @@ -136,9 +137,12 @@ macro_rules! safe_vec { }}; } -#[async_trait] pub(crate) trait SqlTypeDecode { - async fn decode(&self, reader: &mut T, metadata: &ColumnMetadata) -> TdsResult + fn decode( + &self, + reader: &mut T, + metadata: &ColumnMetadata, + ) -> impl Future> + Send where T: TdsPacketReader + Send + Sync; } @@ -501,6 +505,24 @@ impl GenericDecoder { const MAX_PLP_CHUNK_SIZE: usize = 16 * 1024 * 1024; // Reads a SQL_VARIANT type from the TDS stream. + /// Boxed re-entry into [`SqlTypeDecode::decode`], used only by SQL_VARIANT. + /// + /// SQL_VARIANT embeds a base type, so decoding one re-enters the type switch. Native + /// `async fn` cannot express that cycle: the opaque return type would contain itself. + /// Naming a concrete `Pin>` in the signature severs the dependency, at + /// the cost of one allocation per nested variant column — a rare type, never on the hot + /// path of ordinary scalar columns. + fn decode_boxed<'a, T>( + &'a self, + reader: &'a mut T, + metadata: &'a ColumnMetadata, + ) -> Pin> + Send + 'a>> + where + T: TdsPacketReader + Send + Sync, + { + Box::pin(self.decode(reader, metadata)) + } + async fn read_sql_variant(&self, reader: &mut T) -> TdsResult where T: TdsPacketReader + Send + Sync, @@ -576,7 +598,7 @@ impl GenericDecoder { multi_part_name: None, crypto_metadata: None, }; - self.decode(reader, &variant_actual_type_md).await + self.decode_boxed(reader, &variant_actual_type_md).await } _ => { // If the type is not a fixed length type, we should not reach here. @@ -1370,7 +1392,6 @@ impl GenericDecoder { } } -#[async_trait] impl SqlTypeDecode for GenericDecoder { async fn decode(&self, reader: &mut T, metadata: &ColumnMetadata) -> TdsResult where @@ -1761,7 +1782,6 @@ impl StringDecoder { } } -#[async_trait] impl SqlTypeDecode for StringDecoder { async fn decode(&self, reader: &mut T, metadata: &ColumnMetadata) -> TdsResult where @@ -1833,7 +1853,7 @@ impl SqlTypeDecode for StringDecoder { } else { let length = reader.read_uint16().await? as usize; if length == 0xFFFF { - return Ok(ColumnValues::Null); + Ok(ColumnValues::Null) } else { let mut buffer = vec![0u8; length]; reader.read_bytes(&mut buffer).await?; diff --git a/mssql-tds/src/token/parsers/nbcrow_parser.rs b/mssql-tds/src/token/parsers/nbcrow_parser.rs index 5df89010..f80dc40b 100644 --- a/mssql-tds/src/token/parsers/nbcrow_parser.rs +++ b/mssql-tds/src/token/parsers/nbcrow_parser.rs @@ -104,8 +104,6 @@ impl TokenParser

mod tests { use std::sync::Arc; - use async_trait::async_trait; - use super::*; use crate::datatypes::sqldatatypes::{ FixedLengthTypes, TdsDataType, TypeInfo, TypeInfoVariant, @@ -118,7 +116,6 @@ mod tests { #[derive(Default)] struct MockDecoder; - #[async_trait] impl SqlTypeDecode for MockDecoder { async fn decode( &self, @@ -308,7 +305,6 @@ mod tests { #[derive(Default)] struct FailingDecoder; - #[async_trait] impl SqlTypeDecode for FailingDecoder { async fn decode( &self, diff --git a/mssql-tds/src/token/parsers/row_parser.rs b/mssql-tds/src/token/parsers/row_parser.rs index c4919aa8..e57587ae 100644 --- a/mssql-tds/src/token/parsers/row_parser.rs +++ b/mssql-tds/src/token/parsers/row_parser.rs @@ -182,8 +182,6 @@ impl mod tests { use std::sync::Arc; - use async_trait::async_trait; - use super::*; use crate::datatypes::sqldatatypes::{ FixedLengthTypes, TdsDataType, TypeInfo, TypeInfoVariant, @@ -196,7 +194,6 @@ mod tests { #[derive(Default)] struct MockDecoder; - #[async_trait] impl SqlTypeDecode for MockDecoder { async fn decode( &self, @@ -301,7 +298,6 @@ mod tests { #[derive(Default)] struct FailingDecoder; - #[async_trait] impl SqlTypeDecode for FailingDecoder { async fn decode( &self, From e977feb9c4c334eeb71ba0b65cfaa40a3d2540e9 Mon Sep 17 00:00:00 2001 From: Saurabh Singh <1623701+saurabh500@users.noreply.github.com> Date: Wed, 12 Aug 2026 23:31:50 -0700 Subject: [PATCH 2/4] Convert TdsPacketReader to RPITIT Replaces #[async_trait] on TdsPacketReader with explicit -> impl Future + Send. Every column read previously allocated a Pin>; a 39 INT + 9 VARCHAR row issues ~96 reads, so the boxing dominated the decode path. Two consequences the trait definition forced: - The bare AFIT desugaring does not promise Send, and callers wrap these futures in async_trait futures that do. Declaring + Send in the trait keeps every caller compiling unchanged. - RPITIT makes the trait dyn-incompatible. The seven &mut dyn parameters in PlpChunkStreamReader/PlpColumnStream were only ever parameters, so they became generic. The three stored Box live in fuzz-only code and admitted just two concrete types, replaced by the FuzzPacketReader enum. The blanket impl for Box is gone. fuzz_support.rs is #[cfg(fuzzing)]-gated and is not built by a default cargo check, so this was verified with RUSTFLAGS='--cfg fuzzing' against both the lib and the fuzz targets. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 41e46220-ad76-4a77-b477-e768951dcf92 --- .../fuzz_targets/fuzz_connection_provider.rs | 4 +- .../fuzz_connection_provider_context.rs | 4 +- .../fuzz_connection_provider_network.rs | 4 +- .../fuzz/fuzz_targets/fuzz_tds_client.rs | 4 +- mssql-tds/src/connection/tds_client.rs | 1 - .../connection/transport/network_transport.rs | 1 - mssql-tds/src/datatypes/decoder.rs | 61 ++--- mssql-tds/src/fuzz_support.rs | 195 +++++++++++++++- mssql-tds/src/io/packet_reader.rs | 210 ++++++------------ mssql-tds/src/io/token_stream.rs | 3 +- mssql-tds/src/message/prelogin.rs | 3 +- mssql-tds/src/token/parsers/common.rs | 1 - 12 files changed, 289 insertions(+), 202 deletions(-) diff --git a/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider.rs b/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider.rs index 79e1d18e..0b3adae9 100644 --- a/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider.rs +++ b/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider.rs @@ -20,7 +20,7 @@ use libfuzzer_sys::fuzz_target; use mssql_tds::connection::client_context::ClientContext; -use mssql_tds::fuzz_support::{FuzzReader, MockTransport, TdsConnectionProvider}; +use mssql_tds::fuzz_support::{FuzzPacketReader, MockTransport, TdsConnectionProvider}; fuzz_target!(|data: &[u8]| { // Need at least some data to work with @@ -37,7 +37,7 @@ fuzz_target!(|data: &[u8]| { async fn fuzz_connection_provider(data: &[u8]) { // Create a fuzz reader with the input data - let reader = Box::new(FuzzReader::new(data)); + let reader = FuzzPacketReader::from_data(data); let packet_size = 4096; // Create a mock transport with fuzzed data diff --git a/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_context.rs b/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_context.rs index 7e63f883..59300238 100644 --- a/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_context.rs +++ b/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_context.rs @@ -20,7 +20,7 @@ use arbitrary::Arbitrary; use libfuzzer_sys::fuzz_target; use mssql_tds::connection::client_context::{ClientContext, TdsAuthenticationMethod}; use mssql_tds::message::login_options::ApplicationIntent; -use mssql_tds::fuzz_support::{EmptyReader, MockTransport, TdsConnectionProvider}; +use mssql_tds::fuzz_support::{EmptyReader, FuzzPacketReader, MockTransport, TdsConnectionProvider}; #[derive(Debug, Arbitrary)] struct FuzzClientContext { @@ -101,7 +101,7 @@ fuzz_target!(|fuzz_context: FuzzClientContext| { }); async fn fuzz_client_context(fuzz_context: FuzzClientContext) { - let reader = Box::new(EmptyReader); + let reader = FuzzPacketReader::Empty(EmptyReader); let packet_size = 4096; let transport = MockTransport::new(reader, packet_size); diff --git a/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_network.rs b/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_network.rs index 0242281f..f76c4d5d 100644 --- a/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_network.rs +++ b/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_network.rs @@ -18,7 +18,7 @@ use libfuzzer_sys::fuzz_target; use mssql_tds::connection::client_context::ClientContext; -use mssql_tds::fuzz_support::{FuzzReader, MockTransport, TdsConnectionProvider}; +use mssql_tds::fuzz_support::{FuzzPacketReader, MockTransport, TdsConnectionProvider}; fuzz_target!(|data: &[u8]| { if data.is_empty() { @@ -32,7 +32,7 @@ fuzz_target!(|data: &[u8]| { }); async fn fuzz_network_response(data: &[u8]) { - let reader = Box::new(FuzzReader::new(data)); + let reader = FuzzPacketReader::from_data(data); let packet_size = 4096; let transport = MockTransport::new(reader, packet_size); diff --git a/mssql-tds/fuzz/fuzz_targets/fuzz_tds_client.rs b/mssql-tds/fuzz/fuzz_targets/fuzz_tds_client.rs index b7c25ad5..656dc83e 100644 --- a/mssql-tds/fuzz/fuzz_targets/fuzz_tds_client.rs +++ b/mssql-tds/fuzz/fuzz_targets/fuzz_tds_client.rs @@ -19,7 +19,7 @@ #![no_main] use libfuzzer_sys::fuzz_target; -use mssql_tds::fuzz_support::{FuzzReader, create_fuzz_tds_client}; +use mssql_tds::fuzz_support::{FuzzPacketReader, create_fuzz_tds_client}; fuzz_target!(|data: &[u8]| { // Need at least 2 bytes: 1 for scenario, 1+ for token data @@ -39,7 +39,7 @@ fuzz_target!(|data: &[u8]| { rt.block_on(async { // Create TdsClient with mock transport using fuzzer data - let packet_reader = Box::new(FuzzReader::new(token_data)); + let packet_reader = FuzzPacketReader::from_data(token_data); let mut client = create_fuzz_tds_client(packet_reader, 4096); // Execute scenario based on fuzzer input diff --git a/mssql-tds/src/connection/tds_client.rs b/mssql-tds/src/connection/tds_client.rs index cb23fc4d..93ae92f7 100644 --- a/mssql-tds/src/connection/tds_client.rs +++ b/mssql-tds/src/connection/tds_client.rs @@ -4316,7 +4316,6 @@ mod tests { } } - #[async_trait::async_trait] impl crate::io::packet_reader::TdsPacketReader for TestTransport { async fn read_byte(&mut self) -> TdsResult { Ok(self.take_packet_bytes(1)?[0]) diff --git a/mssql-tds/src/connection/transport/network_transport.rs b/mssql-tds/src/connection/transport/network_transport.rs index e0b14746..700aa19f 100644 --- a/mssql-tds/src/connection/transport/network_transport.rs +++ b/mssql-tds/src/connection/transport/network_transport.rs @@ -1099,7 +1099,6 @@ impl TransportSslHandler for NetworkTransport { } } -#[async_trait] impl TdsPacketReader for NetworkTransport { fn reset_reader(&mut self) { // Make sure that we have read all the data from the buffer. diff --git a/mssql-tds/src/datatypes/decoder.rs b/mssql-tds/src/datatypes/decoder.rs index 6b91facf..bc75f74f 100644 --- a/mssql-tds/src/datatypes/decoder.rs +++ b/mssql-tds/src/datatypes/decoder.rs @@ -197,9 +197,10 @@ impl PlpChunkStreamReader { } } - pub(crate) async fn begin( - reader: &mut (dyn TdsPacketReader + Send + Sync), - ) -> TdsResult> { + pub(crate) async fn begin(reader: &mut T) -> TdsResult> + where + T: TdsPacketReader + Send + Sync, + { let raw_len_i64 = reader.read_int64().await?; let raw_len = raw_len_i64 as u64; let raw_len_usize = raw_len as usize; @@ -243,10 +244,10 @@ impl PlpChunkStreamReader { self.reached_end } - async fn ensure_active_chunk( - &mut self, - reader: &mut (dyn TdsPacketReader + Send + Sync), - ) -> TdsResult { + async fn ensure_active_chunk(&mut self, reader: &mut T) -> TdsResult + where + T: TdsPacketReader + Send + Sync, + { if self.reached_end { return Ok(false); } @@ -310,11 +311,10 @@ impl PlpChunkStreamReader { Ok(true) } - pub(crate) async fn read_into( - &mut self, - reader: &mut (dyn TdsPacketReader + Send + Sync), - out: &mut [u8], - ) -> TdsResult { + pub(crate) async fn read_into(&mut self, reader: &mut T, out: &mut [u8]) -> TdsResult + where + T: TdsPacketReader + Send + Sync, + { // Supports the msodbcsql-style cbRequest==0 pattern to consume a // pending terminator after all data bytes were already read. if out.is_empty() { @@ -352,10 +352,10 @@ impl PlpChunkStreamReader { Ok(written) } - pub(crate) async fn skip_to_end( - &mut self, - reader: &mut (dyn TdsPacketReader + Send + Sync), - ) -> TdsResult<()> { + pub(crate) async fn skip_to_end(&mut self, reader: &mut T) -> TdsResult<()> + where + T: TdsPacketReader + Send + Sync, + { while self.ensure_active_chunk(reader).await? { if self.chunk_remaining > 0 { reader.skip_bytes(self.chunk_remaining).await?; @@ -413,10 +413,13 @@ impl PlpColumnStream { /// - `Ok(None)` for SQL NULL /// - `Ok(Some(stream))` ready for incremental reads /// - `Err` if the column is not PLP-typed or the header is malformed - pub(crate) async fn begin( + pub(crate) async fn begin( metadata: &ColumnMetadata, - reader: &mut (dyn TdsPacketReader + Send + Sync), - ) -> TdsResult> { + reader: &mut T, + ) -> TdsResult> + where + T: TdsPacketReader + Send + Sync, + { let (plp_type, collation) = Self::type_from_metadata(metadata)?; let inner = match PlpChunkStreamReader::begin(reader).await? { None => return Ok(None), @@ -456,19 +459,18 @@ impl PlpColumnStream { } /// Incrementally reads PLP payload bytes into `out`. - pub(crate) async fn read_into( - &mut self, - reader: &mut (dyn TdsPacketReader + Send + Sync), - out: &mut [u8], - ) -> TdsResult { + pub(crate) async fn read_into(&mut self, reader: &mut T, out: &mut [u8]) -> TdsResult + where + T: TdsPacketReader + Send + Sync, + { self.inner.read_into(reader, out).await } /// Discards all remaining PLP payload and terminator bytes. - pub(crate) async fn skip_to_end( - &mut self, - reader: &mut (dyn TdsPacketReader + Send + Sync), - ) -> TdsResult<()> { + pub(crate) async fn skip_to_end(&mut self, reader: &mut T) -> TdsResult<()> + where + T: TdsPacketReader + Send + Sync, + { self.inner.skip_to_end(reader).await } @@ -3158,7 +3160,7 @@ mod test { } mod decode_into_tests { - use async_trait::async_trait; + use byteorder::{ByteOrder, LittleEndian}; use crate::core::TdsResult; @@ -3199,7 +3201,6 @@ mod test { } } - #[async_trait] impl TdsPacketReader for ByteReader { async fn read_byte(&mut self) -> TdsResult { Ok(self.take(1)?[0]) diff --git a/mssql-tds/src/fuzz_support.rs b/mssql-tds/src/fuzz_support.rs index 4af3f3c8..6c2787fa 100644 --- a/mssql-tds/src/fuzz_support.rs +++ b/mssql-tds/src/fuzz_support.rs @@ -66,7 +66,6 @@ impl FuzzReader { } } -#[async_trait] impl TdsPacketReader for FuzzReader { async fn read_byte(&mut self) -> TdsResult { if self.position >= self.data.len() { @@ -271,7 +270,6 @@ impl TdsPacketReader for FuzzReader { /// Always-EOF reader for fuzz targets that only care about context variations. pub struct EmptyReader; -#[async_trait] impl TdsPacketReader for EmptyReader { async fn read_byte(&mut self) -> TdsResult { Err(mssql_tds_error_eof()) @@ -432,10 +430,191 @@ impl NetworkWriter for MockWriter { } } +/// Concrete packet-reader used by the fuzz harness. +/// +/// `TdsPacketReader` returns `impl Future` per method, which makes it dyn-incompatible. +/// The harness only ever supplies one of two readers, so an enum recovers the runtime +/// choice that `Box` previously provided, with static dispatch. +pub enum FuzzPacketReader { + /// Reads from the fuzzer-supplied byte slice. + Fuzz(FuzzReader), + /// Always reports end-of-stream. + Empty(EmptyReader), +} + +impl FuzzPacketReader { + /// Builds a reader over the fuzzer-supplied input. + pub fn from_data(data: &[u8]) -> Self { + Self::Fuzz(FuzzReader::new(data)) + } +} + +impl TdsPacketReader for FuzzPacketReader { + async fn read_byte(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_byte().await, + Self::Empty(r) => r.read_byte().await, + } + } + + async fn read_int16_big_endian(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_int16_big_endian().await, + Self::Empty(r) => r.read_int16_big_endian().await, + } + } + + async fn read_int32_big_endian(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_int32_big_endian().await, + Self::Empty(r) => r.read_int32_big_endian().await, + } + } + + async fn read_uint40(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_uint40().await, + Self::Empty(r) => r.read_uint40().await, + } + } + + async fn read_float32(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_float32().await, + Self::Empty(r) => r.read_float32().await, + } + } + + async fn read_float64(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_float64().await, + Self::Empty(r) => r.read_float64().await, + } + } + + async fn read_int16(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_int16().await, + Self::Empty(r) => r.read_int16().await, + } + } + + async fn read_uint16(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_uint16().await, + Self::Empty(r) => r.read_uint16().await, + } + } + + async fn read_uint24(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_uint24().await, + Self::Empty(r) => r.read_uint24().await, + } + } + + async fn read_int32(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_int32().await, + Self::Empty(r) => r.read_int32().await, + } + } + + async fn read_uint32(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_uint32().await, + Self::Empty(r) => r.read_uint32().await, + } + } + + async fn read_int64(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_int64().await, + Self::Empty(r) => r.read_int64().await, + } + } + + async fn read_uint64(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_uint64().await, + Self::Empty(r) => r.read_uint64().await, + } + } + + async fn read_bytes(&mut self, buffer: &mut [u8]) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_bytes(buffer).await, + Self::Empty(r) => r.read_bytes(buffer).await, + } + } + + async fn read_u8_varbyte(&mut self) -> TdsResult> { + match self { + Self::Fuzz(r) => r.read_u8_varbyte().await, + Self::Empty(r) => r.read_u8_varbyte().await, + } + } + + async fn read_u16_varbyte(&mut self) -> TdsResult> { + match self { + Self::Fuzz(r) => r.read_u16_varbyte().await, + Self::Empty(r) => r.read_u16_varbyte().await, + } + } + + async fn read_varchar_u16_length(&mut self) -> TdsResult> { + match self { + Self::Fuzz(r) => r.read_varchar_u16_length().await, + Self::Empty(r) => r.read_varchar_u16_length().await, + } + } + + async fn read_varchar_u8_length(&mut self) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_varchar_u8_length().await, + Self::Empty(r) => r.read_varchar_u8_length().await, + } + } + + async fn read_unicode(&mut self, string_length: usize) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_unicode(string_length).await, + Self::Empty(r) => r.read_unicode(string_length).await, + } + } + + async fn read_unicode_with_byte_length(&mut self, byte_length: usize) -> TdsResult { + match self { + Self::Fuzz(r) => r.read_unicode_with_byte_length(byte_length).await, + Self::Empty(r) => r.read_unicode_with_byte_length(byte_length).await, + } + } + + async fn skip_bytes(&mut self, skip_count: usize) -> TdsResult<()> { + match self { + Self::Fuzz(r) => r.skip_bytes(skip_count).await, + Self::Empty(r) => r.skip_bytes(skip_count).await, + } + } + + async fn cancel_read_stream(&mut self) -> TdsResult<()> { + match self { + Self::Fuzz(r) => r.cancel_read_stream().await, + Self::Empty(r) => r.cancel_read_stream().await, + } + } + + fn reset_reader(&mut self) { + match self { + Self::Fuzz(r) => r.reset_reader(), + Self::Empty(r) => r.reset_reader(), + } + } +} + /// MockTransport simulates a transport layer for fuzzing pub struct MockTransport { - token_stream_reader: - TokenStreamReader, GenericTokenParserRegistry>, + token_stream_reader: TokenStreamReader, mock_writer: MockWriter, packet_size: u32, encryption_setting: NegotiatedEncryptionSetting, @@ -451,7 +630,7 @@ impl std::fmt::Debug for MockTransport { } impl MockTransport { - pub fn new(packet_reader: Box, packet_size: u32) -> Self { + pub fn new(packet_reader: FuzzPacketReader, packet_size: u32) -> Self { let parser_registry = Box::new(GenericTokenParserRegistry::default()); let token_stream_reader = TokenStreamReader::new(packet_reader, parser_registry); @@ -615,7 +794,6 @@ impl TdsTransport for MockTransport { } } -#[async_trait] impl TdsPacketReader for MockTransport { async fn read_byte(&mut self) -> TdsResult { self.token_stream_reader.packet_reader.read_byte().await @@ -756,10 +934,7 @@ pub fn create_test_execution_context() -> crate::connection::execution_context:: } /// Helper function to create TdsClient for fuzzing -pub fn create_fuzz_tds_client( - packet_reader: Box, - packet_size: u32, -) -> TdsClient { +pub fn create_fuzz_tds_client(packet_reader: FuzzPacketReader, packet_size: u32) -> TdsClient { let mock_transport = MockTransport::new(packet_reader, packet_size); let negotiated_settings = create_test_negotiated_settings(); let execution_context = create_test_execution_context(); diff --git a/mssql-tds/src/io/packet_reader.rs b/mssql-tds/src/io/packet_reader.rs index da238125..cf9c9329 100644 --- a/mssql-tds/src/io/packet_reader.rs +++ b/mssql-tds/src/io/packet_reader.rs @@ -1,168 +1,84 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT License. -use async_trait::async_trait; +use std::future::Future; use crate::core::TdsResult; /// Sentinel `u16` length marking a length-prefixed varchar field as NULL. pub(crate) const LENGTH_NULL: u16 = 0xffff; -#[async_trait] #[cfg(not(fuzzing))] pub(crate) trait TdsPacketReader { - async fn read_byte(&mut self) -> TdsResult; - async fn read_int16_big_endian(&mut self) -> TdsResult; - async fn read_int32_big_endian(&mut self) -> TdsResult; - async fn read_uint40(&mut self) -> TdsResult; - - async fn read_float32(&mut self) -> TdsResult; - async fn read_float64(&mut self) -> TdsResult; - async fn read_int16(&mut self) -> TdsResult; - async fn read_uint16(&mut self) -> TdsResult; - async fn read_uint24(&mut self) -> TdsResult; - async fn read_int32(&mut self) -> TdsResult; - async fn read_uint32(&mut self) -> TdsResult; - async fn read_int64(&mut self) -> TdsResult; - async fn read_uint64(&mut self) -> TdsResult; - - async fn read_bytes(&mut self, buffer: &mut [u8]) -> TdsResult; - async fn read_u8_varbyte(&mut self) -> TdsResult>; + fn read_byte(&mut self) -> impl Future> + Send; + fn read_int16_big_endian(&mut self) -> impl Future> + Send; + fn read_int32_big_endian(&mut self) -> impl Future> + Send; + fn read_uint40(&mut self) -> impl Future> + Send; + + fn read_float32(&mut self) -> impl Future> + Send; + fn read_float64(&mut self) -> impl Future> + Send; + fn read_int16(&mut self) -> impl Future> + Send; + fn read_uint16(&mut self) -> impl Future> + Send; + fn read_uint24(&mut self) -> impl Future> + Send; + fn read_int32(&mut self) -> impl Future> + Send; + fn read_uint32(&mut self) -> impl Future> + Send; + fn read_int64(&mut self) -> impl Future> + Send; + fn read_uint64(&mut self) -> impl Future> + Send; + + fn read_bytes(&mut self, buffer: &mut [u8]) -> impl Future> + Send; + fn read_u8_varbyte(&mut self) -> impl Future>> + Send; #[allow(dead_code)] - async fn read_u16_varbyte(&mut self) -> TdsResult>; - async fn read_varchar_u16_length(&mut self) -> TdsResult>; - async fn read_varchar_u8_length(&mut self) -> TdsResult; + fn read_u16_varbyte(&mut self) -> impl Future>> + Send; + fn read_varchar_u16_length(&mut self) + -> impl Future>> + Send; + fn read_varchar_u8_length(&mut self) -> impl Future> + Send; #[allow(dead_code)] - async fn read_unicode(&mut self, string_length: usize) -> TdsResult; - async fn read_unicode_with_byte_length(&mut self, byte_length: usize) -> TdsResult; - async fn skip_bytes(&mut self, skip_count: usize) -> TdsResult<()>; - async fn cancel_read_stream(&mut self) -> TdsResult<()>; + fn read_unicode( + &mut self, + string_length: usize, + ) -> impl Future> + Send; + fn read_unicode_with_byte_length( + &mut self, + byte_length: usize, + ) -> impl Future> + Send; + fn skip_bytes(&mut self, skip_count: usize) -> impl Future> + Send; + fn cancel_read_stream(&mut self) -> impl Future> + Send; fn reset_reader(&mut self); } /// Low-level TDS packet reading operations (public under `fuzzing` cfg). -#[async_trait] #[cfg(fuzzing)] pub trait TdsPacketReader { - async fn read_byte(&mut self) -> TdsResult; - async fn read_int16_big_endian(&mut self) -> TdsResult; - async fn read_int32_big_endian(&mut self) -> TdsResult; - async fn read_uint40(&mut self) -> TdsResult; - - async fn read_float32(&mut self) -> TdsResult; - async fn read_float64(&mut self) -> TdsResult; - async fn read_int16(&mut self) -> TdsResult; - async fn read_uint16(&mut self) -> TdsResult; - async fn read_uint24(&mut self) -> TdsResult; - async fn read_int32(&mut self) -> TdsResult; - async fn read_uint32(&mut self) -> TdsResult; - async fn read_int64(&mut self) -> TdsResult; - async fn read_uint64(&mut self) -> TdsResult; - - async fn read_bytes(&mut self, buffer: &mut [u8]) -> TdsResult; - async fn read_u8_varbyte(&mut self) -> TdsResult>; - async fn read_u16_varbyte(&mut self) -> TdsResult>; - async fn read_varchar_u16_length(&mut self) -> TdsResult>; - async fn read_varchar_u8_length(&mut self) -> TdsResult; - async fn read_unicode(&mut self, string_length: usize) -> TdsResult; - async fn read_unicode_with_byte_length(&mut self, byte_length: usize) -> TdsResult; - async fn skip_bytes(&mut self, skip_count: usize) -> TdsResult<()>; - async fn cancel_read_stream(&mut self) -> TdsResult<()>; + fn read_byte(&mut self) -> impl Future> + Send; + fn read_int16_big_endian(&mut self) -> impl Future> + Send; + fn read_int32_big_endian(&mut self) -> impl Future> + Send; + fn read_uint40(&mut self) -> impl Future> + Send; + + fn read_float32(&mut self) -> impl Future> + Send; + fn read_float64(&mut self) -> impl Future> + Send; + fn read_int16(&mut self) -> impl Future> + Send; + fn read_uint16(&mut self) -> impl Future> + Send; + fn read_uint24(&mut self) -> impl Future> + Send; + fn read_int32(&mut self) -> impl Future> + Send; + fn read_uint32(&mut self) -> impl Future> + Send; + fn read_int64(&mut self) -> impl Future> + Send; + fn read_uint64(&mut self) -> impl Future> + Send; + + fn read_bytes(&mut self, buffer: &mut [u8]) -> impl Future> + Send; + fn read_u8_varbyte(&mut self) -> impl Future>> + Send; + fn read_u16_varbyte(&mut self) -> impl Future>> + Send; + fn read_varchar_u16_length(&mut self) + -> impl Future>> + Send; + fn read_varchar_u8_length(&mut self) -> impl Future> + Send; + fn read_unicode( + &mut self, + string_length: usize, + ) -> impl Future> + Send; + fn read_unicode_with_byte_length( + &mut self, + byte_length: usize, + ) -> impl Future> + Send; + fn skip_bytes(&mut self, skip_count: usize) -> impl Future> + Send; + fn cancel_read_stream(&mut self) -> impl Future> + Send; fn reset_reader(&mut self); } - -// Blanket implementation for Box to enable dynamic dispatch -#[async_trait] -impl TdsPacketReader for Box { - async fn read_byte(&mut self) -> TdsResult { - (**self).read_byte().await - } - - async fn read_int16_big_endian(&mut self) -> TdsResult { - (**self).read_int16_big_endian().await - } - - async fn read_int32_big_endian(&mut self) -> TdsResult { - (**self).read_int32_big_endian().await - } - - async fn read_uint40(&mut self) -> TdsResult { - (**self).read_uint40().await - } - - async fn read_float32(&mut self) -> TdsResult { - (**self).read_float32().await - } - - async fn read_float64(&mut self) -> TdsResult { - (**self).read_float64().await - } - - async fn read_int16(&mut self) -> TdsResult { - (**self).read_int16().await - } - - async fn read_uint16(&mut self) -> TdsResult { - (**self).read_uint16().await - } - - async fn read_uint24(&mut self) -> TdsResult { - (**self).read_uint24().await - } - - async fn read_int32(&mut self) -> TdsResult { - (**self).read_int32().await - } - - async fn read_uint32(&mut self) -> TdsResult { - (**self).read_uint32().await - } - - async fn read_int64(&mut self) -> TdsResult { - (**self).read_int64().await - } - - async fn read_uint64(&mut self) -> TdsResult { - (**self).read_uint64().await - } - - async fn read_bytes(&mut self, buffer: &mut [u8]) -> TdsResult { - (**self).read_bytes(buffer).await - } - - async fn read_u8_varbyte(&mut self) -> TdsResult> { - (**self).read_u8_varbyte().await - } - - async fn read_u16_varbyte(&mut self) -> TdsResult> { - (**self).read_u16_varbyte().await - } - - async fn read_varchar_u16_length(&mut self) -> TdsResult> { - (**self).read_varchar_u16_length().await - } - - async fn read_varchar_u8_length(&mut self) -> TdsResult { - (**self).read_varchar_u8_length().await - } - - async fn read_unicode(&mut self, string_length: usize) -> TdsResult { - (**self).read_unicode(string_length).await - } - - async fn read_unicode_with_byte_length(&mut self, byte_length: usize) -> TdsResult { - (**self).read_unicode_with_byte_length(byte_length).await - } - - async fn skip_bytes(&mut self, skip_count: usize) -> TdsResult<()> { - (**self).skip_bytes(skip_count).await - } - - async fn cancel_read_stream(&mut self) -> TdsResult<()> { - (**self).cancel_read_stream().await - } - - fn reset_reader(&mut self) { - (**self).reset_reader() - } -} diff --git a/mssql-tds/src/io/token_stream.rs b/mssql-tds/src/io/token_stream.rs index 38853b3b..6a3bf481 100644 --- a/mssql-tds/src/io/token_stream.rs +++ b/mssql-tds/src/io/token_stream.rs @@ -1056,7 +1056,7 @@ mod tests { use crate::datatypes::sqldatatypes::{TdsDataType, TypeInfo}; use crate::io::packet_reader::TdsPacketReader; use crate::token::tokens::{SqlCollation, TokenType}; - use async_trait::async_trait; + use std::collections::HashMap; use std::sync::Arc; @@ -1173,7 +1173,6 @@ mod tests { } } - #[async_trait] impl TdsPacketReader for TestByteReader { async fn read_byte(&mut self) -> TdsResult { Ok(self.take(1)?[0]) diff --git a/mssql-tds/src/message/prelogin.rs b/mssql-tds/src/message/prelogin.rs index cc80119a..e6a7f4e8 100644 --- a/mssql-tds/src/message/prelogin.rs +++ b/mssql-tds/src/message/prelogin.rs @@ -437,7 +437,7 @@ pub(crate) mod tests { use crate::io::packet_reader::TdsPacketReader; use crate::io::packet_writer::PacketWriter; use crate::io::packet_writer::tests::MockNetworkWriter; - use async_trait::async_trait; + use byteorder::{BigEndian, ReadBytesExt}; use futures::executor::block_on; @@ -446,7 +446,6 @@ pub(crate) mod tests { mockall::mock! { pub TestPacketReader {} - #[async_trait] impl TdsPacketReader for TestPacketReader { async fn read_byte(&mut self) -> TdsResult; async fn read_int16_big_endian(&mut self) -> TdsResult; diff --git a/mssql-tds/src/token/parsers/common.rs b/mssql-tds/src/token/parsers/common.rs index a442853e..4141f177 100644 --- a/mssql-tds/src/token/parsers/common.rs +++ b/mssql-tds/src/token/parsers/common.rs @@ -67,7 +67,6 @@ pub(crate) mod test_utils { } } - #[async_trait] impl TdsPacketReader for MockReader { async fn read_byte(&mut self) -> TdsResult { if self.position >= self.data.len() { From cfa0dfe7da003f20faf7f31775d7c3d592681850 Mon Sep 17 00:00:00 2001 From: Saurabh Singh <1623701+saurabh500@users.noreply.github.com> Date: Wed, 12 Aug 2026 23:33:00 -0700 Subject: [PATCH 3/4] Make the row-decode chain generic over the RowWriter Adds W: RowWriter + Send + ?Sized to drive_row_columns, decode_or_decrypt_column, receive_row_into_internal and resume_row_into_internal, so a caller holding a concrete writer gets static dispatch on the ~96 writer calls a wide row makes. GenericDecoder::decode_into was already generic over W; these four were the only reason it always instantiated as a trait object. Because the bound is ?Sized, dyn RowWriter + Send still satisfies W, so every existing caller compiles unchanged and TdsTransport stays dyn-compatible. Production delta today is 0%, and the reason is structural rather than incidental: TdsClient holds Box, and TdsTransport has TdsTokenStreamReader as a supertrait, so the writer's concrete type is erased at that vtable call before it ever reaches this chain. Neither a generic sibling method nor genericizing TdsTokenStreamReader can recover it -- a where Self: Sized method is uncallable on a trait object, and a bare generic method makes Box illegal. This commit makes everything below that boundary ready, so the win lands with no further decode-path work once the transport indirection is removed. The blocked headroom is measured on the PR. resume_row_into_internal is included for symmetry. Its only callers pass dyn, so it adds no extra monomorphization; drop it if reviewers prefer. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 41e46220-ad76-4a77-b477-e768951dcf92 --- mssql-tds/src/io/token_stream.rs | 25 +++++++++++++++++-------- 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/mssql-tds/src/io/token_stream.rs b/mssql-tds/src/io/token_stream.rs index 6a3bf481..bf2a5488 100644 --- a/mssql-tds/src/io/token_stream.rs +++ b/mssql-tds/src/io/token_stream.rs @@ -474,14 +474,14 @@ fn pause_after_column( /// bytes, or pause. This single loop replaces the former /// `decode_row_columns` / `decode_nbcrow_columns` pair and their /// `writer.pause_*` polling. -async fn drive_row_columns( +async fn drive_row_columns( reader: &mut R, metadata: &Arc, decryptor: Option<&Arc>, bitmap: Option<&[u8]>, start_col: usize, plan: ColumnPolicy, - writer: &mut (dyn RowWriter + Send), + writer: &mut W, ) -> TdsResult { let decoder = GenericDecoder::default(); let columns = &metadata.columns; @@ -559,13 +559,16 @@ async fn drive_row_columns( Ok(RowReadResult::RowWritten) } -async fn decode_or_decrypt_column( +async fn decode_or_decrypt_column< + R: TdsPacketReader + Send + Sync, + W: RowWriter + Send + ?Sized, +>( decoder: &GenericDecoder, reader: &mut R, meta: &ColumnMetadata, decryptor: Option<&Arc>, col: usize, - writer: &mut (dyn RowWriter + Send), + writer: &mut W, ) -> TdsResult<()> { match (meta.crypto_metadata.is_some(), decryptor) { (true, Some(dec)) => { @@ -588,12 +591,15 @@ async fn decode_or_decrypt_column( Ok(()) } -pub(crate) async fn receive_row_into_internal( +pub(crate) async fn receive_row_into_internal< + R: TdsPacketReader + Send + Sync, + W: RowWriter + Send + ?Sized, +>( reader: &mut R, registry: &impl TokenParserRegistry, context: &ParserContext, plan: ColumnPolicy, - writer: &mut (dyn RowWriter + Send), + writer: &mut W, ) -> TdsResult { let token_type_byte = reader.read_byte().await?; let token_type: TokenType = token_type_byte.try_into()?; @@ -664,11 +670,14 @@ pub(crate) async fn receive_row_header_internal( +pub(crate) async fn resume_row_into_internal< + R: TdsPacketReader + Send + Sync, + W: RowWriter + Send + ?Sized, +>( reader: &mut R, pause_state: RowPauseState, plan: ColumnPolicy, - writer: &mut (dyn RowWriter + Send), + writer: &mut W, ) -> TdsResult { let RowPauseState { next_column_index, From 4c64fd2f11c6908cb87471d8f31077f717a04a44 Mon Sep 17 00:00:00 2001 From: Saurabh Singh <1623701+saurabh500@users.noreply.github.com> Date: Thu, 13 Aug 2026 12:32:02 -0700 Subject: [PATCH 4/4] Address review feedback on the row-decode dispatch change Guard the decode-chain futures. row_fetch_futures_stay_small (#225) builds its futures on TdsClient, which re-boxes at Box, so it cannot observe this file. Measured: its four futures are byte-identical on main and on this branch (1128/376/1408/1160) while receive_row_into_internal went 752 -> 1392 B and drive_row_columns 592 -> 1232 B. Add row_decode_futures_stay_small below the boundary, covering both the dyn and monomorphic instantiations against the same 4096 B budget. Move the read_sql_variant comment back onto read_sql_variant; decode_boxed was inserted between them. Add FuzzPacketReader::empty() so both variants have a constructor, which drops the last EmptyReader import from the fuzz targets. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 41e46220-ad76-4a77-b477-e768951dcf92 --- .../fuzz_connection_provider_context.rs | 4 +- mssql-tds/src/datatypes/decoder.rs | 2 +- mssql-tds/src/fuzz_support.rs | 5 ++ mssql-tds/src/io/token_stream.rs | 82 +++++++++++++++++++ 4 files changed, 90 insertions(+), 3 deletions(-) diff --git a/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_context.rs b/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_context.rs index 59300238..6b959e07 100644 --- a/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_context.rs +++ b/mssql-tds/fuzz/fuzz_targets/fuzz_connection_provider_context.rs @@ -20,7 +20,7 @@ use arbitrary::Arbitrary; use libfuzzer_sys::fuzz_target; use mssql_tds::connection::client_context::{ClientContext, TdsAuthenticationMethod}; use mssql_tds::message::login_options::ApplicationIntent; -use mssql_tds::fuzz_support::{EmptyReader, FuzzPacketReader, MockTransport, TdsConnectionProvider}; +use mssql_tds::fuzz_support::{FuzzPacketReader, MockTransport, TdsConnectionProvider}; #[derive(Debug, Arbitrary)] struct FuzzClientContext { @@ -101,7 +101,7 @@ fuzz_target!(|fuzz_context: FuzzClientContext| { }); async fn fuzz_client_context(fuzz_context: FuzzClientContext) { - let reader = FuzzPacketReader::Empty(EmptyReader); + let reader = FuzzPacketReader::empty(); let packet_size = 4096; let transport = MockTransport::new(reader, packet_size); diff --git a/mssql-tds/src/datatypes/decoder.rs b/mssql-tds/src/datatypes/decoder.rs index bc75f74f..9a2f35d4 100644 --- a/mssql-tds/src/datatypes/decoder.rs +++ b/mssql-tds/src/datatypes/decoder.rs @@ -506,7 +506,6 @@ impl GenericDecoder { #[cfg(not(fuzzing))] const MAX_PLP_CHUNK_SIZE: usize = 16 * 1024 * 1024; - // Reads a SQL_VARIANT type from the TDS stream. /// Boxed re-entry into [`SqlTypeDecode::decode`], used only by SQL_VARIANT. /// /// SQL_VARIANT embeds a base type, so decoding one re-enters the type switch. Native @@ -525,6 +524,7 @@ impl GenericDecoder { Box::pin(self.decode(reader, metadata)) } + // Reads a SQL_VARIANT type from the TDS stream. async fn read_sql_variant(&self, reader: &mut T) -> TdsResult where T: TdsPacketReader + Send + Sync, diff --git a/mssql-tds/src/fuzz_support.rs b/mssql-tds/src/fuzz_support.rs index 6c2787fa..65df6f9b 100644 --- a/mssql-tds/src/fuzz_support.rs +++ b/mssql-tds/src/fuzz_support.rs @@ -447,6 +447,11 @@ impl FuzzPacketReader { pub fn from_data(data: &[u8]) -> Self { Self::Fuzz(FuzzReader::new(data)) } + + /// Builds a reader that always reports end-of-stream. + pub fn empty() -> Self { + Self::Empty(EmptyReader) + } } impl TdsPacketReader for FuzzPacketReader { diff --git a/mssql-tds/src/io/token_stream.rs b/mssql-tds/src/io/token_stream.rs index bf2a5488..2e483577 100644 --- a/mssql-tds/src/io/token_stream.rs +++ b/mssql-tds/src/io/token_stream.rs @@ -1069,6 +1069,88 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; + /// Companion to `row_fetch_futures_stay_small` (#225) for the decode chain below + /// `Box`. That guard measures futures built on `TdsClient`, which + /// re-boxes at the transport boundary, so it cannot observe anything in this file: + /// its four futures are byte-identical before and after this chain roughly doubled. + #[test] + fn row_decode_futures_stay_small() { + const MAX: usize = 4096; + + let metadata = Arc::new(ColMetadataToken { + column_count: 0, + columns: vec![], + cek_table: vec![], + }); + let context = ParserContext::ColumnMetadata(Arc::clone(&metadata), None); + let registry = GenericTokenParserRegistry::default(); + let mut reader = TestByteReader::new(vec![TokenType::Row as u8]); + let mut sink = DiscardRowWriter; + + // Constructing an async fn's future runs none of its body, so these are free to + // build and drop unpolled. Each borrow ends with its statement. + // + // Both instantiations are measured: `dyn` is what production reaches today, and + // the monomorphic one is what a concrete writer gets once the transport boundary + // stops erasing it (#265). + let receive_dyn = size_of_val(&receive_row_into_internal( + &mut reader, + ®istry, + &context, + ColumnPolicy::DecodeAll, + &mut sink as &mut (dyn RowWriter + Send), + )); + let receive_mono = size_of_val(&receive_row_into_internal( + &mut reader, + ®istry, + &context, + ColumnPolicy::DecodeAll, + &mut sink, + )); + let drive_dyn = size_of_val(&drive_row_columns( + &mut reader, + &metadata, + None, + None, + 0, + ColumnPolicy::DecodeAll, + &mut sink as &mut (dyn RowWriter + Send), + )); + let drive_mono = size_of_val(&drive_row_columns( + &mut reader, + &metadata, + None, + None, + 0, + ColumnPolicy::DecodeAll, + &mut sink, + )); + let resume_dyn = size_of_val(&resume_row_into_internal( + &mut reader, + RowPauseState { + next_column_index: 0, + metadata: Arc::clone(&metadata), + nbc_null_bitmap: None, + decryptor: None, + }, + ColumnPolicy::DecodeAll, + &mut sink as &mut (dyn RowWriter + Send), + )); + + for (name, size) in [ + ("receive_row_into_internal (dyn)", receive_dyn), + ("receive_row_into_internal (mono)", receive_mono), + ("drive_row_columns (dyn)", drive_dyn), + ("drive_row_columns (mono)", drive_mono), + ("resume_row_into_internal (dyn)", resume_dyn), + ] { + assert!( + size <= MAX, + "{name} future is {size} B, expected <= {MAX} B" + ); + } + } + #[test] fn test_parser_context_default() { let context = ParserContext::default();