Failed streamed TVP request poison other requests in the application

Start with TdsParser.FailureCleanup in src/Microsoft.Data.SqlClient/src/Microsoft.Data.SqlClient/Parser/TdsParser.cs, then trace how WritePacket updates _outputPacketNumber and how attention cleanup is handled. Use the supplied TVP reproduction with a row source that throws around row 256,000, in both sync and async modes. Done means the failure sends attention or dooms the connection so a subsequent command on the pooled connection succeeds.

Describe the bug

Pooled connection left with an unfinished RPC when a streamed TVP throws in a packet whose number wrapped to 1

When the IEnumerable<SqlDataRecord> value of a table-valued parameter throws while SqlClient is writing the RPC,
TdsParser.FailureCleanup sends an attention only if stateObj._outputPacketNumber != 1, i.e. if it thinks a packet
has already been sent:

// src/Microsoft.Data.SqlClient/src/Microsoft/Data/SqlClient/Parser/TdsParser.cs, FailureCleanup
int old_outputPacketNumber = stateObj._outputPacketNumber;
...
stateObj.ResetBuffer();
stateObj.ResetPacketCounters();

if (old_outputPacketNumber != 1 && _state == TdsParserState.OpenLoggedIn)
{
    ...
    // If _outputPacketNumber prior to ResetBuffer was not equal to 1, a packet was already
    // sent to the server and so we need to send an attention and process the attention ack.
    stateObj.SendAttention();
    ProcessAttention(stateObj);
}

_outputPacketNumber is a byte incremented for every packet (TdsParserStateObject.WritePacket), so it wraps
around: while the 257th (513th, ...) packet of the request is being filled it is 1 again, although 256 packets were
already sent. In that case no attention is sent, the server keeps waiting for the rest of the RPC, and the connection
is returned to the pool as healthy. The next command on that pooled connection (any caller) fails, because the server
parses its bytes as the continuation of the unfinished TVP:

Microsoft.Data.SqlClient.SqlException (0x80131904): The incoming tabular data stream (TDS) remote procedure call (RPC)
protocol stream is incorrect. Parameter 4 (""): Data type 0x12 is unknown.

The exact error depends on the bytes; in production we get 8037:

The incoming tabular data stream (TDS) remote procedure call (RPC) protocol stream is incorrect. Table-valued parameter 3 (""),
row 213890, column 4: Data type 0x68 has an invalid data length or metadata length.
The data for table-valued parameter "@ImportTable" doesn't conform to the table type of the parameter. SQL Server error is: 8037, state: 44

When the exception is thrown in any other packet, SqlClient sends the attention and the connection is fine.

To reproduce

TvpPoolRepro.csproj

<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net9.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
    <SqlClientVersion Condition="'$(SqlClientVersion)' == ''">6.1.2</SqlClientVersion>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="Microsoft.Data.SqlClient" Version="$(SqlClientVersion)" />
  </ItemGroup>

</Project>

Program.cs

using System.Data;
using System.Reflection;
using Microsoft.Data.SqlClient;
using Microsoft.Data.SqlClient.Server;

var connectionString = args.FirstOrDefault(arg => !arg.StartsWith("--"))
                       ?? @"Server=(localdb)\MSSQLLocalDB;Database=tempdb;Integrated Security=true";
var useSync = args.Contains("--sync");

Console.WriteLine(
    $"Microsoft.Data.SqlClient {typeof(SqlConnection).Assembly.GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion}, "
    + (useSync ? "sync" : "async"));

await using (var connection = new SqlConnection(connectionString))
{
    await connection.OpenAsync();
    await using var command = new SqlCommand(
        "IF TYPE_ID('dbo.ReproTvp') IS NULL CREATE TYPE dbo.ReproTvp AS TABLE (Id int NOT NULL, Flag bit NOT NULL)",
        connection);
    await command.ExecuteNonQueryAsync();
}

// 1. TVP streamed from an enumerable that throws mid-request.
// A row is 8 bytes, so an 8000 byte packet holds 999 rows: rows 255717..256715 are written into the 257th packet,
// whose TDS packet number wrapped around to 1 (any row in this window reproduces, rows outside it don't)
await using (var connection = new SqlConnection(connectionString))
{
    await connection.OpenAsync();
    Console.WriteLine($"TVP command on connection {connection.ClientConnectionId}");

    await using var command = new SqlCommand("SELECT COUNT(*) FROM @Rows", connection);
    command.Parameters.Add(
        new SqlParameter("@Rows", SqlDbType.Structured) { TypeName = "dbo.ReproTvp", Value = GetRows(failAtRow: 256_000) });

    try
    {
        if (useSync)
            command.ExecuteNonQuery();
        else
            await command.ExecuteNonQueryAsync();
    }
    catch (Exception exception)
    {
        Console.WriteLine($"  expected: {exception.GetType().Name}: {exception.Message}");
    }
}

// 2. unrelated parameterized command; the pool hands out the same physical connection
await using (var connection = new SqlConnection(connectionString))
{
    await connection.OpenAsync();
    Console.WriteLine($"Next command on connection {connection.ClientConnectionId}");

    await using var command = new SqlCommand("SELECT @Value", connection);
    command.Parameters.Add(new SqlParameter("@Value", SqlDbType.Int) { Value = 1 });

    try
    {
        var result = useSync ? command.ExecuteScalar() : await command.ExecuteScalarAsync();
        Console.WriteLine($"  OK, result {result}");
    }
    catch (SqlException exception)
    {
        Console.WriteLine($"  BUG, error {exception.Number}: {exception.Message}");
    }
}

static IEnumerable<SqlDataRecord> GetRows(int failAtRow)
{
    SqlMetaData[] metaData = [new("Id", SqlDbType.Int), new("Flag", SqlDbType.Bit)];

    for (var row = 1; ; row++)
    {
        // e.g. the rows come from an HTTP request body and the client disconnects
        if (row == failAtRow)
            throw new InvalidOperationException($"Row source failed at row {row}");

        var record = new SqlDataRecord(metaData);
        record.SetInt32(0, row);
        record.SetBoolean(1, row % 2 == 0);

        yield return record;
    }
}

Output (dotnet run, same with -- --sync, -p:SqlClientVersion=7.1.1, -p:SqlClientVersion=5.1.6):

Microsoft.Data.SqlClient 6.1.2+607b8c975a0866b674a63aa5ee389c07f85f88e8, async
TVP command on connection ad96acab-3318-47a5-864c-3ee0a3935a0a
  expected: InvalidOperationException: Row source failed at row 256000
Next command on connection ad96acab-3318-47a5-864c-3ee0a3935a0a
  BUG, error 8009: The incoming tabular data stream (TDS) remote procedure call (RPC) protocol stream is incorrect. Parameter 4 (""): Data type 0x12 is unknown.

Sweeping failAtRow from 1 to 400 000 (step 997) fails only inside one window; a step-1 sweep of its edges gives exactly
rows 255 717..256 715, i.e. the 999 rows of the 257th packet. The next window is around 511 500..512 500 (the next wrap). Reading TdsParserStateObject._outputPacketNumber via reflection at the moment of the throw shows 1 for every
failing row. With failAtRow: 10_000 the next command succeeds.

Expected behavior

SqlClient sends an attention (as it does when the row source throws in any other packet), or dooms the connection,
so the next command on the pooled connection succeeds.

Further technical details

• Microsoft.Data.SqlClient version: 6.1.2; also reproduced with 7.1.1 and 5.1.6
• .NET target: net9.0
• SQL Server version: LocalDB 15.0.4382 (repro); Amazon RDS SQL Server 16.00.4255 (production)
• Operating system: Windows 11 (repro); Windows Server on EC2 (production)
• Both ExecuteNonQueryAsync and ExecuteNonQuery are affected

Additional context
In production the TVP rows are streamed from an ASP.NET Core request body, so the row source throws when
the server rejects the body ("Request body too large") or the client disconnects. The failure then surfaces seconds later
in unrelated requests that get the pooled connection. We confirmed via ClientConnectionId that the victims run on the same physical connection.