TransactionLeak.cs

using System;
using System.Data.SqlClient;
using System.Threading.Tasks;
using System.Transactions;
using NServiceBus;

namespace ConnectionLeakTest
{
    class Program
    {
        public const string ConnectionString = "data source=(local); initial catalog=nservicebus; integrated security=true; max pool size=20";

        static void Main(string[] args)
        {
            Start().GetAwaiter().GetResult();
        }

        static async Task Start()
        {
            var config = new EndpointConfiguration("LeakyEndpoint");
            config.EnableInstallers();
            config.Recoverability().Immediate(i => i.NumberOfRetries(0));
            config.Recoverability().Delayed(d => d.NumberOfRetries(0));
            config.SendFailedMessagesTo("error");

            config.UsePersistence<InMemoryPersistence>();

            var transport = config.UseTransport<SqlServerTransport>();
            transport.ConnectionString(ConnectionString);

            var endpoint = await Endpoint.Start(config);

            Console.WriteLine("Press m to send a message or l to leak a connection");
            while (true)
            {
                var ch = char.ToLower(Console.ReadKey().KeyChar);
                if (ch == 'm')
                {
                    try
                    {
                        await endpoint.SendLocal(new MyMessage());
                        Console.Write(".");
                    }
                    catch (Exception e)
                    {
                        Console.WriteLine(e);
                    }
                }
                else if (ch == 'l')
                {
                    try
                    {
                        //Force completing transaction scope on a different thread.
                        //Will throw exception and poison the connection pool
                        using (var scope = new TransactionScope(TransactionScopeAsyncFlowOption.Suppress))
                        {
                            Console.WriteLine($"Thread ID {System.Threading.Thread.CurrentThread.ManagedThreadId}");
                            using (var connection = new SqlConnection(Program.ConnectionString))
                            {
                                await connection.OpenAsync();
                                await Task.Delay(100);
                            }
                            Console.WriteLine($"Thread ID {System.Threading.Thread.CurrentThread.ManagedThreadId}");
                            scope.Complete();
                        }
                    }
                    catch (Exception e)
                    {
                        Console.WriteLine(e);
                    }
                }
            }
        }
    }

    class MyMessageHandler : IHandleMessages<MyMessage>
    {
        public async Task Handle(MyMessage message, IMessageHandlerContext context)
        {
            using (var connection = new SqlConnection(Program.ConnectionString))
            {
                await connection.OpenAsync();
                //NOOP
            }
        }
    }

    class MyMessage : IMessage
    {
    }
}