Route incoming attachment reads through the connection factory (#795)

* Route incoming attachment reads through the connection factory

Reads (GetStream, CopyTo, GetBytes, ProcessStream, …) now always open a
  fresh connection from the configured factory instead of sharing the
  receive transaction's connection. This lets a handler hold an
  OpenOutgoingAttachment sink open while reading incoming attachments
  without colliding on a non-MARS connection. Writes are unchanged — they
  still use the ambient connection/transaction so saves remain atomic
  with the receive under SendsAtomicWithReceive.

  Tradeoff: incoming reads are no longer enlisted in the receive
  transaction. The attachment row was committed by the sender before the
  message arrived, so reads see committed data either way; the change
  only affects callers that deliberately relied on rolling back the
  receive to "undo" a read.

  Also fixes a lifetime bug in Persister.GetStream where the command and
  reader were disposed at method exit (via await using), killing the
  returned SqlSequentialStream before the caller could read it. Disposal
  is now deferred to the AttachmentStream's cleanups on the success path.

  The unused MessageAttachmentsFromTransaction,
  MessageAttachmentsFromSqlTransaction, and
  MessageAttachmentsFromSqlConnection wrapper classes are removed.

* Update Directory.Build.props

Author: SimonCropp

docs/mdsource/sql.source.md

@@ -32,6 +32,8 @@ Attachments can leverage the ambient SQL connectivity from either the [transport
 
 If both `UseSynchronizedStorageSessionConnectivity` and `UseTransportConnectivity` are defined, the `SynchronizedStorageSession` will be used first, followed by the `TransportTransaction`.
 
+Ambient connectivity applies to attachment writes only — attachment saves run on the ambient connection/transaction so the save is atomic with the receive (under `SendsAtomicWithReceive`) or the persister's storage session. Attachment reads always run on a fresh connection from the `connectionFactory` and are not enlisted in the receive transaction. This lets a handler hold an `OpenOutgoingAttachment` sink open while reading incoming attachments without colliding with the write on a non-MARS connection. Each read call (`GetStream`, `CopyTo`, `GetBytes`, `ProcessStream`, etc.) opens its own short-lived connection; SQL connection pooling makes this cheap.
+
 
 #### Use SynchronizedStorageSession connectivity
 
@@ -42,8 +44,8 @@ snippet: UseSynchronizedStorageSessionConnectivity
 This approach attempts to use the SynchronizedStorageSession using the following steps:
 
  * For the current context attempt to retrieve an instance of `SynchronizedStorageSession`. If no `SynchronizedStorageSession` exists, don't continue and fall back to the [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) retrieved by the `connectionFactory`.
- * Attempt to retrieve a property named 'Transaction' that is a [SqlTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqltransaction) from the `SynchronizedStorageSession`. If it exists, use it for all SQL operations in the current pipeline.
- * Attempt to retrieve a property named 'Connection' that is a [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) from the `SynchronizedStorageSession`. If it exists, use it for all SQL operations in the current pipeline.
+ * Attempt to retrieve a property named 'Transaction' that is a [SqlTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqltransaction) from the `SynchronizedStorageSession`. If it exists, use it for outgoing attachment operations in the current pipeline.
+ * Attempt to retrieve a property named 'Connection' that is a [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) from the `SynchronizedStorageSession`. If it exists, use it for outgoing attachment operations in the current pipeline.
 
 The properties are retrieved using [reflection](https://docs.microsoft.com/en-us/dotnet/framework/reflection-and-codedom/reflection) since there is no API in NServiceBus to access SynchronizedStorageSession data via type.
 
@@ -57,9 +59,9 @@ snippet: UseTransportConnectivity
 This approach attempts to use the transport transaction using the following steps:
 
  * For the current context, attempt to retrieve an instance of `TransportTransaction`. If no `TransportTransaction` exists, don't continue and fall back to using the [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) retrieved by the `connectionFactory`.
- * Attempt to retrieve an instance of [Transaction](https://docs.microsoft.com/en-us/dotnet/api/system.transactions.transaction) from the `TransportTransaction`. If it exists, use it in [SqlConnection.EnlistTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection.enlisttransaction) with an instance of [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) retrieved by the `connectionFactory`. Then use that [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) for all SQL operations in the current pipeline.
- * Attempt to retrieve an instance of [SqlTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqltransaction) from the `TransportTransaction`. If it exists, use it for all SQL operations in the current pipeline.
- * Attempt to retrieve an instance of [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) from the `TransportTransaction`. If it exists, use it for all SQL operations in the current pipeline.
+ * Attempt to retrieve an instance of [Transaction](https://docs.microsoft.com/en-us/dotnet/api/system.transactions.transaction) from the `TransportTransaction`. If it exists, use it in [SqlConnection.EnlistTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection.enlisttransaction) with an instance of [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) retrieved by the `connectionFactory`. Then use that [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) for outgoing attachment operations in the current pipeline.
+ * Attempt to retrieve an instance of [SqlTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqltransaction) from the `TransportTransaction`. If it exists, use it for outgoing attachment operations in the current pipeline.
+ * Attempt to retrieve an instance of [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) from the `TransportTransaction`. If it exists, use it for outgoing attachment operations in the current pipeline.
  * Any attachments associated with a message send will be deleted after message processing.
 
 

docs/sql.md

@@ -87,6 +87,8 @@ Attachments can leverage the ambient SQL connectivity from either the [transport
 
 If both `UseSynchronizedStorageSessionConnectivity` and `UseTransportConnectivity` are defined, the `SynchronizedStorageSession` will be used first, followed by the `TransportTransaction`.
 
+Ambient connectivity applies to attachment writes only — attachment saves run on the ambient connection/transaction so the save is atomic with the receive (under `SendsAtomicWithReceive`) or the persister's storage session. Attachment reads always run on a fresh connection from the `connectionFactory` and are not enlisted in the receive transaction. This lets a handler hold an `OpenOutgoingAttachment` sink open while reading incoming attachments without colliding with the write on a non-MARS connection. Each read call (`GetStream`, `CopyTo`, `GetBytes`, `ProcessStream`, etc.) opens its own short-lived connection; SQL connection pooling makes this cheap.
+
 
 #### Use SynchronizedStorageSession connectivity
 
@@ -106,8 +108,8 @@ attachments.UseSynchronizedStorageSessionConnectivity();
 This approach attempts to use the SynchronizedStorageSession using the following steps:
 
  * For the current context attempt to retrieve an instance of `SynchronizedStorageSession`. If no `SynchronizedStorageSession` exists, don't continue and fall back to the [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) retrieved by the `connectionFactory`.
- * Attempt to retrieve a property named 'Transaction' that is a [SqlTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqltransaction) from the `SynchronizedStorageSession`. If it exists, use it for all SQL operations in the current pipeline.
- * Attempt to retrieve a property named 'Connection' that is a [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) from the `SynchronizedStorageSession`. If it exists, use it for all SQL operations in the current pipeline.
+ * Attempt to retrieve a property named 'Transaction' that is a [SqlTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqltransaction) from the `SynchronizedStorageSession`. If it exists, use it for outgoing attachment operations in the current pipeline.
+ * Attempt to retrieve a property named 'Connection' that is a [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) from the `SynchronizedStorageSession`. If it exists, use it for outgoing attachment operations in the current pipeline.
 
 The properties are retrieved using [reflection](https://docs.microsoft.com/en-us/dotnet/framework/reflection-and-codedom/reflection) since there is no API in NServiceBus to access SynchronizedStorageSession data via type.
 
@@ -130,9 +132,9 @@ attachments.UseTransportConnectivity();
 This approach attempts to use the transport transaction using the following steps:
 
  * For the current context, attempt to retrieve an instance of `TransportTransaction`. If no `TransportTransaction` exists, don't continue and fall back to using the [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) retrieved by the `connectionFactory`.
- * Attempt to retrieve an instance of [Transaction](https://docs.microsoft.com/en-us/dotnet/api/system.transactions.transaction) from the `TransportTransaction`. If it exists, use it in [SqlConnection.EnlistTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection.enlisttransaction) with an instance of [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) retrieved by the `connectionFactory`. Then use that [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) for all SQL operations in the current pipeline.
- * Attempt to retrieve an instance of [SqlTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqltransaction) from the `TransportTransaction`. If it exists, use it for all SQL operations in the current pipeline.
- * Attempt to retrieve an instance of [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) from the `TransportTransaction`. If it exists, use it for all SQL operations in the current pipeline.
+ * Attempt to retrieve an instance of [Transaction](https://docs.microsoft.com/en-us/dotnet/api/system.transactions.transaction) from the `TransportTransaction`. If it exists, use it in [SqlConnection.EnlistTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection.enlisttransaction) with an instance of [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) retrieved by the `connectionFactory`. Then use that [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) for outgoing attachment operations in the current pipeline.
+ * Attempt to retrieve an instance of [SqlTransaction](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqltransaction) from the `TransportTransaction`. If it exists, use it for outgoing attachment operations in the current pipeline.
+ * Attempt to retrieve an instance of [SqlConnection](https://docs.microsoft.com/en-us/dotnet/api/system.data.sqlclient.sqlconnection) from the `TransportTransaction`. If it exists, use it for outgoing attachment operations in the current pipeline.
  * Any attachments associated with a message send will be deleted after message processing.
 
 

src/Attachments.Sql.Tests/OpenOutgoingAttachmentConcurrentReadTests.cs

@@ -0,0 +1,112 @@
+using Microsoft.Extensions.DependencyInjection;
+
+[NotInParallel]
+public class OpenOutgoingAttachmentConcurrentReadTests
+{
+    static readonly TestState state = new();
+
+    // Reads run on a fresh connection (MessageContextExtensions.Attachments goes through the
+    // connection factory). This test holds an OpenOutgoingAttachment sink open on the receive
+    // transport's connection while reading the incoming attachment — the previous design would
+    // have failed with "MultipleActiveResultSets".
+    [Test]
+    [Arguments(TransportTransactionMode.SendsAtomicWithReceive)]
+    [Arguments(TransportTransactionMode.ReceiveOnly)]
+    public async Task ReadsIncomingWhileSinkIsOpen(TransportTransactionMode transactionMode)
+    {
+        state.Bytes = null;
+        state.Reply.Reset();
+
+        var dbName = $"OpenConcurrent_{transactionMode}";
+        await using var database = await Connection.SqlInstance.Build(dbName);
+        var connectionString = database.ConnectionString;
+        var databaseName = new SqlConnectionStringBuilder(connectionString).InitialCatalog;
+
+        var configuration = new EndpointConfiguration($"SqlOpenConcurrent_{transactionMode}");
+        SqlConnection NewConnection() => new(connectionString);
+        var attachments = configuration.EnableAttachments(NewConnection, TimeToKeep.Default, database: databaseName, table: "Attachments");
+        attachments.UseTransportConnectivity();
+        configuration.UseSerialization<SystemJsonSerializer>();
+        configuration.UsePersistence<LearningPersistence>();
+        configuration.DisableRetries();
+        configuration.EnableInstallers();
+        configuration.PurgeOnStartup(true);
+        attachments.DisableCleanupTask();
+        configuration.RegisterComponents(_ => _.AddSingleton(state));
+
+        var transport = configuration.UseTransport<SqlServerTransport>();
+        transport.ConnectionString(connectionString);
+        transport.Transactions(transactionMode);
+
+        var endpoint = await Endpoint.Start(configuration);
+
+        var sendOptions = new SendOptions();
+        sendOptions.RouteToThisEndpoint();
+        var outgoing = sendOptions.Attachments();
+        outgoing.AddStream("input", async stream =>
+        {
+            await using var writer = new StreamWriter(stream, leaveOpen: true);
+            await writer.WriteAsync("hello");
+        });
+        await endpoint.Send(new InMessage(), sendOptions);
+
+        if (!state.Reply.WaitOne(TimeSpan.FromSeconds(30)))
+        {
+            await endpoint.Stop();
+            throw new($"TimedOut for mode {transactionMode}");
+        }
+
+        await endpoint.Stop();
+
+        await Assert.That(Encoding.UTF8.GetString(state.Bytes!)).IsEqualTo("HELLO");
+    }
+
+    class TestState
+    {
+        public byte[]? Bytes;
+        public ManualResetEvent Reply = new(false);
+    }
+
+    class InMessage :
+        IMessage;
+
+    class OutMessage :
+        IMessage;
+
+    class InHandler :
+        IHandleMessages<InMessage>
+    {
+        public async Task Handle(InMessage message, HandlerContext context)
+        {
+            var incoming = context.Attachments();
+            var replyOptions = new ReplyOptions();
+
+            await using (var sink = await context.OpenOutgoingAttachment(replyOptions, "output"))
+            {
+                // Read the incoming attachment while the outgoing sink is held open.
+                using var sourceBuffer = new MemoryStream();
+                await incoming.CopyTo("input", sourceBuffer, context.CancellationToken);
+                sourceBuffer.Position = 0;
+                using var reader = new StreamReader(sourceBuffer, leaveOpen: true);
+                var content = await reader.ReadToEndAsync(context.CancellationToken);
+                await using var writer = new StreamWriter(sink, leaveOpen: true);
+                await writer.WriteAsync(content.ToUpperInvariant());
+            }
+
+            await context.Reply(new OutMessage(), replyOptions);
+        }
+    }
+
+    class OutHandler :
+        IHandleMessages<OutMessage>
+    {
+        public async Task Handle(OutMessage message, HandlerContext context)
+        {
+            await using var memoryStream = new MemoryStream();
+            var incoming = context.Attachments();
+            await incoming.CopyTo("output", memoryStream, context.CancellationToken);
+            state.Bytes = memoryStream.ToArray();
+            state.Reply.Set();
+        }
+    }
+}