SqlTransaction.Dispose() throws NullReferenceException from SqlInternalTransaction.Rollback() when it races the background close of a broken connection

<Project Sdk="Microsoft.NET.Sdk">
  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net10.0</TargetFramework>
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
  </PropertyGroup>
  <ItemGroup>
    <PackageReference Include="Microsoft.Data.SqlClient" Version="7.1.1" />
  </ItemGroup>
</Project>

using System.Collections.Concurrent;
using System.Diagnostics.Tracing;
using System.Net;
using System.Net.Sockets;
using Microsoft.Data.SqlClient;

// Repro: SqlTransaction.Dispose() throws NullReferenceException from SqlInternalTransaction.Rollback() (or
// InvalidOperationException "connection is closed") when it races the background close of a connection that
// broke under a pending async command.
//
// The EventListener below sleeps on the disposing thread when SqlClient writes the
// "SqlInternalTransaction.Rollback | API" trace event, which is emitted after Dispose's null check and before
// Rollback's dereference. That widens the race window; no SqlClient code is changed. holdMs=0 shows the race
// without the hold.
//
// Usage: dotnet run -- "<connection string>" [holdMs=300] [iterations=10] [mode=reset|freeze] [pauseMs=0] [compat=false]
//   mode reset:  TCP reset while the command is pending
//   mode freeze: the server stops responding; CommandTimeout (1 s) expires, the attention is never
//                acknowledged, and after the 5 s attention timeout SqlClient faults the task, then closes the connection
//   pauseMs:     random 0..pauseMs delay between the failure and Dispose
//   compat:      value of both UseCompatibilityAsyncBehaviour and UseCompatibilityProcessSni
string target = args[0];
int holdMs = args.Length > 1 ? int.Parse(args[1]) : 300;
int iterations = args.Length > 2 ? int.Parse(args[2]) : 10;
string mode = args.Length > 3 ? args[3] : "reset";
int pauseMs = args.Length > 4 ? int.Parse(args[4]) : 0;
bool compat = args.Length > 5 && bool.Parse(args[5]);
AppContext.SetSwitch("Switch.Microsoft.Data.SqlClient.UseCompatibilityAsyncBehaviour", compat);
AppContext.SetSwitch("Switch.Microsoft.Data.SqlClient.UseCompatibilityProcessSni", compat);
var random = new Random(1);

var targetBuilder = new SqlConnectionStringBuilder(target);
string[] hostPort = targetBuilder.DataSource.Split(',');

using var proxy = new ResettingProxy(hostPort[0], hostPort.Length > 1 ? int.Parse(hostPort[1]) : 1433);
proxy.Start();
var viaProxy = new SqlConnectionStringBuilder(target) { DataSource = $"127.0.0.1,{proxy.Port}", Pooling = false };

using var hold = new RollbackHold();
var outcomes = new ConcurrentDictionary<string, int>();
string? firstNre = null;

for (int i = 0; i < iterations; i++)
{
    var connection = new SqlConnection(viaProxy.ConnectionString);
    await connection.OpenAsync();
    var transaction = connection.BeginTransaction();

    var pending = Task.Run(async () =>
    {
        using var command = new SqlCommand("WAITFOR DELAY '00:00:30'; SELECT 1", connection, transaction) { CommandTimeout = 1 };
        await command.ExecuteScalarAsync();
    });
    await Task.Delay(200);
    if (mode == "freeze")
    {
        proxy.Frozen = true; // the server stops responding, the socket stays open
    }
    else
    {
        proxy.ResetAll(); // the network drops while the async command is pending
    }

    string operation;
    try { await pending; operation = "no error"; }
    catch (Exception ex) { operation = ex.GetType().Name + (ex is SqlException se ? $" {se.Number}" : ""); }

    if (pauseMs > 0)
    {
        await Task.Delay(random.Next(0, pauseMs));
    }

    // Dispose the transaction straight after the failure, as a using block or an ORM does.
    RollbackHold.HoldMs = holdMs;
    string dispose;
    try { transaction.Dispose(); dispose = "ok"; }
    catch (Exception ex)
    {
        dispose = $"{ex.GetType().Name}: {ex.Message}";
        if (ex is NullReferenceException) firstNre ??= ex.ToString();
    }
    finally { RollbackHold.HoldMs = 0; }

    connection.Dispose();
    proxy.ResetAll();
    proxy.Frozen = false;
    outcomes.AddOrUpdate($"command: {operation}; transaction.Dispose(): {dispose}", 1, (_, n) => n + 1);
}

var version = typeof(SqlConnection).Assembly
    .GetCustomAttributes(typeof(System.Reflection.AssemblyInformationalVersionAttribute), false)
    .OfType<System.Reflection.AssemblyInformationalVersionAttribute>().First().InformationalVersion;
Console.WriteLine($"Microsoft.Data.SqlClient {version}; holdMs={holdMs}");
foreach (var (outcome, count) in outcomes) Console.WriteLine($"{count,4} x {outcome}");
if (firstNre != null) Console.WriteLine(firstNre);

internal sealed class RollbackHold : EventListener
{
    public static volatile int HoldMs;

    protected override void OnEventSourceCreated(EventSource source)
    {
        if (source.Name == "Microsoft.Data.SqlClient.EventSource")
        {
            EnableEvents(source, EventLevel.Verbose, EventKeywords.All);
        }
    }

    protected override void OnEventWritten(EventWrittenEventArgs e)
    {
        if (HoldMs > 0 && e.Payload is { Count: > 0 } p && p[0] is string s
            && s.StartsWith("SqlInternalTransaction.Rollback | API", StringComparison.Ordinal))
        {
            Thread.Sleep(HoldMs);
        }
    }
}

// Forwards TCP to the server; ResetAll() aborts every forwarded connection with RST.
internal sealed class ResettingProxy(string host, int port) : IDisposable
{
    private readonly TcpListener _listener = new(IPAddress.Loopback, 0);
    private readonly ConcurrentDictionary<int, (TcpClient Client, TcpClient Server)> _live = new();
    private int _ids;
    public volatile bool Frozen;

    public int Port => ((IPEndPoint)_listener.LocalEndpoint).Port;

    public void Start()
    {
        _listener.Start();
        _ = Task.Run(async () =>
        {
            while (true)
            {
                TcpClient client;
                try { client = await _listener.AcceptTcpClientAsync(); }
                catch (Exception) { return; }
                _ = Task.Run(() => Forward(client));
            }
        });
    }

    private async Task Forward(TcpClient client)
    {
        int id = Interlocked.Increment(ref _ids);
        var server = new TcpClient();
        try
        {
            await server.ConnectAsync(host, port);
            _live[id] = (client, server);
            await Task.WhenAny(Pump(client.GetStream(), server.GetStream()), Pump(server.GetStream(), client.GetStream()));
        }
        catch (Exception) { }
        finally
        {
            _live.TryRemove(id, out _);
            client.Close();
            server.Close();
        }
    }

    private async Task Pump(NetworkStream from, NetworkStream to)
    {
        var buffer = new byte[16384];
        try
        {
            int n;
            while ((n = await from.ReadAsync(buffer)) > 0)
            {
                while (Frozen) await Task.Delay(10);
                await to.WriteAsync(buffer.AsMemory(0, n));
            }
        }
        catch (Exception) { }
    }

    public void ResetAll()
    {
        foreach (var id in _live.Keys)
        {
            if (_live.TryRemove(id, out var pair))
            {
                try { pair.Client.Client.LingerState = new LingerOption(true, 0); pair.Client.Close(); } catch (Exception) { }
                try { pair.Server.Close(); } catch (Exception) { }
            }
        }
    }

    public void Dispose() => _listener.Stop();
}